diff --git a/features/db_url_sanitisation.feature b/features/db_url_sanitisation.feature new file mode 100644 index 000000000..a5f12ae1f --- /dev/null +++ b/features/db_url_sanitisation.feature @@ -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 "" + When I run sanitise_db_url + Then the sanitised database url should be "" + + 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 "" + When I run sanitise_db_url + Then the sanitised database url should be "" + + 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 "" + When I run sanitise_db_url + Then the sanitised database url should be "" + + 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" diff --git a/features/steps/db_url_sanitisation_steps.py b/features/steps/db_url_sanitisation_steps.py new file mode 100644 index 000000000..9ff628028 --- /dev/null +++ b/features/steps/db_url_sanitisation_steps.py @@ -0,0 +1,85 @@ +"""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}'" diff --git a/src/cleveragents/cli/commands/system.py b/src/cleveragents/cli/commands/system.py index 42bba1fe5..31a3dccc8 100644 --- a/src/cleveragents/cli/commands/system.py +++ b/src/cleveragents/cli/commands/system.py @@ -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,