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 "<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"

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}'"

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,