fix(cli): Mask database URL credentials in agents info CLI output (#8395)
This commit is contained in:
@@ -0,0 +1,53 @@
|
||||
Feature: Database URL sanitisation — credentials must never be exposed
|
||||
|
||||
As a CleverAgents operator
|
||||
I want database URLs in CLI output to have their credentials masked
|
||||
So that running ``agents info`` never leaks passwords or API tokens
|
||||
|
||||
@tdd_bug_8395
|
||||
Scenario Outline: PostgreSQL URL with user and password — all are masked
|
||||
Given the database url is "<url>"
|
||||
When I run sanitise_db_url
|
||||
Then the sanitised database url should be "<expected>"
|
||||
|
||||
Examples:
|
||||
| url | expected |
|
||||
| postgresql://user:secret@localhost/mydb | postgresql://***:***@localhost/mydb |
|
||||
| postgresql://admin:p%40ssw0rd@db.example.com:5432/agents | postgresql://***:***@db.example.com:5432/agents |
|
||||
| postgresql://readonly:r0ad_only@pg.cluster.internal:5433/production | postgresql://***:***@pg.cluster.internal:5433/production |
|
||||
|
||||
Scenario Outline: MySQL URL with credentials — all are masked
|
||||
Given the database url is "<url>"
|
||||
When I run sanitise_db_url
|
||||
Then the sanitised database url should be "<expected>"
|
||||
|
||||
Examples:
|
||||
| url | expected |
|
||||
| mysql://app:s3cret@mysql.internal:3306/agents | mysql://***:***@mysql.internal:3306/agents |
|
||||
| mysql+pymysql://root:toor@localhost/testdb | mysql+pymysql://***:***@localhost/testdb |
|
||||
|
||||
Scenario Outline: SQLite URLs — remain unchanged (no credentials)
|
||||
Given the database url is "<url>"
|
||||
When I run sanitise_db_url
|
||||
Then the sanitised database url should be "<expected>"
|
||||
|
||||
Examples:
|
||||
| url | expected |
|
||||
| sqlite:///data/cleveragents.db | sqlite:///data/cleveragents.db |
|
||||
| sqlite:////absolute/path/to/db.sqlite | sqlite:////absolute/path/to/db.sqlite |
|
||||
| memory | memory |
|
||||
|
||||
Scenario: Username-only URL — password is still masked
|
||||
Given the database url is "postgres://deploy@db.example.com/prod"
|
||||
When I run sanitise_db_url
|
||||
Then the sanitised database url should be "postgres://***:***@db.example.com/prod"
|
||||
|
||||
Scenario: SQLite URL without credentials remains unchanged
|
||||
Given the database url is "sqlite:///test.db"
|
||||
When I run sanitise_db_url
|
||||
Then the sanitised database url should be "sqlite:///test.db"
|
||||
|
||||
Scenario: build_info_data returns sanitised database URL
|
||||
Given a mock settings object with database url "postgresql://admin:supersecret@host.example.com:5432/mydb"
|
||||
When I call build_info_data
|
||||
Then the database field in info data should be "postgresql://***:***@host.example.com:5432/mydb"
|
||||
@@ -0,0 +1,93 @@
|
||||
"""Step definitions for db_url_sanitisation.feature.
|
||||
|
||||
Tests the ``_sanitise_db_url`` helper and its use in ``build_info_data``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from tempfile import mkdtemp
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Given
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given('the database url is "{url}"')
|
||||
def step_db_url_given(context: Context, url: str) -> None:
|
||||
"""Store the raw database URL for processing."""
|
||||
context.raw_db_url = url
|
||||
|
||||
|
||||
@given(
|
||||
"a mock settings object with database url \"{db_url}\""
|
||||
)
|
||||
def step_mock_settings_with_db_url(context: Context, db_url: str) -> None:
|
||||
"""Build a full mock Settings that ``build_info_data`` expects."""
|
||||
tmpdir = mkdtemp()
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.database_url = db_url
|
||||
mock_settings.data_dir = Path(tmpdir)
|
||||
mock_settings.storage_path = Path(tmpdir)
|
||||
mock_settings.config_path = Path(tmpdir) / "config.toml"
|
||||
mock_settings.log_dir = Path(tmpdir) / "logs"
|
||||
mock_settings.default_automation_profile = "auto"
|
||||
mock_settings.has_provider_configured = MagicMock(return_value=False)
|
||||
mock_settings.configured_provider_names = MagicMock(return_value=[])
|
||||
mock_settings.debug_enabled = False
|
||||
context.mock_settings = mock_settings
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# When
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I run sanitise_db_url")
|
||||
def step_run_sanitise(context: Context) -> None:
|
||||
"""Call _sanitise_db_url with the stored raw URL."""
|
||||
from cleveragents.cli.commands.system import _sanitise_db_url
|
||||
|
||||
context.sanitised_url = _sanitise_db_url(context.raw_db_url)
|
||||
|
||||
|
||||
@when("I call build_info_data")
|
||||
def step_call_build_info(context: Context) -> None:
|
||||
"""Call ``build_info_data`` with the mock settings."""
|
||||
from cleveragents.cli.commands.system import build_info_data
|
||||
|
||||
with patch(
|
||||
"cleveragents.config.settings.get_settings",
|
||||
return_value=context.mock_settings,
|
||||
):
|
||||
context.info_data = build_info_data()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Then
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then('the sanitised database url should be "{expected}"')
|
||||
def step_assert_sanitised_url(context: Context, expected: str) -> None:
|
||||
"""Verify the sanitised URL matches expectation."""
|
||||
actual = context.sanitised_url
|
||||
assert actual == expected, (
|
||||
f"Expected '{expected}', got '{actual}'"
|
||||
)
|
||||
|
||||
|
||||
@then(
|
||||
"the database field in info data should be \"{expected}\""
|
||||
)
|
||||
def step_assert_info_db_field(context: Context, expected: str) -> None:
|
||||
"""Verify the sanitised URL appears correctly in build_info_data output."""
|
||||
actual = context.info_data["database"]
|
||||
assert actual == expected, (
|
||||
f"Expected database field '{expected}', got '{actual}'"
|
||||
)
|
||||
@@ -17,6 +17,7 @@ from datetime import UTC, datetime
|
||||
from enum import StrEnum
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse, urlunparse
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy import inspect as sa_inspect
|
||||
@@ -82,6 +83,41 @@ def _dep_version(package: str) -> str:
|
||||
return "not installed"
|
||||
|
||||
|
||||
def _sanitise_db_url(url: str) -> str:
|
||||
"""Sanitise a database URL by masking credentials.
|
||||
|
||||
Parses the URL and replaces username/password components with masked
|
||||
placeholders so that CLI output never exposes real credentials.
|
||||
|
||||
Examples::
|
||||
|
||||
>>> _sanitise_db_url("postgresql://user:secret@localhost/mydb")
|
||||
'postgresql://***:***@localhost/mydb'
|
||||
>>> _sanitise_db_url("sqlite:///path/to/db.sqlite")
|
||||
'sqlite:///path/to/db.sqlite'
|
||||
|
||||
Args:
|
||||
url: The raw database URL.
|
||||
|
||||
Returns:
|
||||
The sanitised URL with credentials masked (or unchanged if no
|
||||
username/password is present).
|
||||
"""
|
||||
parsed = urlparse(url)
|
||||
|
||||
# If there is no userinfo segment, return as-is (e.g. sqlite URLs)
|
||||
if not parsed.username and not parsed.password:
|
||||
return url
|
||||
|
||||
# Rebuild the netloc with masked credentials
|
||||
userinfo = "***:***"
|
||||
port_part = f":{parsed.port}" if parsed.port else ""
|
||||
new_netloc = f"{userinfo}@{parsed.hostname}{port_part}"
|
||||
|
||||
sanitized = parsed._replace(netloc=new_netloc)
|
||||
return urlunparse(sanitized)
|
||||
|
||||
|
||||
def build_version_data() -> dict[str, Any]:
|
||||
"""Assemble structured data for the ``version`` command."""
|
||||
return {
|
||||
@@ -145,7 +181,7 @@ def build_info_data() -> dict[str, Any]:
|
||||
"version": __version__,
|
||||
"data_dir": str(data_dir),
|
||||
"config_path": str(config_path),
|
||||
"database": db_url,
|
||||
"database": _sanitise_db_url(db_url),
|
||||
"server_mode": server_mode,
|
||||
"platform": f"{platform.system()} {platform.release()} ({platform.machine()})",
|
||||
"automation": settings.default_automation_profile,
|
||||
|
||||
Reference in New Issue
Block a user