learningcircuit--local-deep-research/tests/security/test_sanitize_error_message.py

"""Tests for the standalone ``sanitize_error_message`` function.

Verifies that the regex-based sanitizer catches common credential formats
in exception messages. This function was extracted from
``BaseSearchEngine._sanitize_error_message`` so it can be used outside the
search-engine inheritance tree (e.g. in LLM config and error handling).
"""

from local_deep_research.security.log_sanitizer import sanitize_error_message


class TestSanitizeErrorMessageStandalone:
    """Unit tests for the pattern-based credential sanitizer."""

    def test_bearer_token_redacted(self):
        msg = 'Error: Connection refused, token="Bearer sk-proj-abc123xyz456def789ghi012"'
        result = sanitize_error_message(msg)
        assert "sk-proj-abc123xyz456def789ghi012" not in result
        assert "Bearer [REDACTED]" in result

    def test_url_query_api_key_redacted(self):
        msg = "HTTPSConnectionPool: Max retries exceeded with url: /v1/models?api_key=sk-secret-key-value-12345&q=test"
        result = sanitize_error_message(msg)
        assert "sk-secret-key-value-12345" not in result
        assert "api_key=[REDACTED]" in result

    def test_url_query_apikey_no_underscore_redacted(self):
        msg = "?apikey=my-secret-api-key-value-here&format=json"
        result = sanitize_error_message(msg)
        assert "my-secret-api-key-value-here" not in result
        assert "apikey=[REDACTED]" in result

    def test_url_query_api_key_hyphenated_redacted(self):
        """Guardian uses ``api-key`` (with hyphen) as the param name."""
        msg = "?api-key=guardian-secret-key-value-here&q=test"
        result = sanitize_error_message(msg)
        assert "guardian-secret-key-value-here" not in result
        assert "api-key=[REDACTED]" in result

    def test_url_query_key_redacted(self):
        """Google PSE uses ``key`` as the param name."""
        msg = "?key=google-pse-secret-key-value&q=test"
        result = sanitize_error_message(msg)
        assert "google-pse-secret-key-value" not in result
        assert "key=[REDACTED]" in result

    def test_url_query_named_secrets_redacted(self):
        """Secret-named query params beyond the classic tokens must also be
        redacted, matching DataSanitizer.DEFAULT_SENSITIVE_KEYS."""
        for name in (
            "client_secret",
            "secret_key",
            "bearer_token",
            "api_secret",
            "app_secret",
        ):
            value = f"{name}-oauth-value-here-1234"
            result = sanitize_error_message(f"POST /cb?{name}={value}&x=1")
            assert value not in result, name
            assert f"{name}=[REDACTED]" in result, name

    def test_url_query_client_id_not_redacted(self):
        """client_id is a public identifier, not a secret — stays readable."""
        msg = "?client_id=public-app-id-value-12345&scope=read"
        result = sanitize_error_message(msg)
        assert "public-app-id-value-12345" in result

    def test_url_query_token_redacted(self):
        msg = "?token=secret-token-value-1234567890&page=1"
        result = sanitize_error_message(msg)
        assert "secret-token-value-1234567890" not in result
        assert "token=[REDACTED]" in result

    def test_url_query_secret_redacted(self):
        msg = "&secret=super-secret-value-abcde12345&format=json"
        result = sanitize_error_message(msg)
        assert "super-secret-value-abcde12345" not in result
        assert "secret=[REDACTED]" in result

    def test_url_credentials_redacted(self):
        msg = "Connection to https://admin:supersecretpassword@api.example.com failed"
        result = sanitize_error_message(msg)
        assert "admin" not in result
        assert "supersecretpassword" not in result
        assert "[REDACTED]:[REDACTED]@" in result

    def test_url_credentials_inside_api_key_param_value(self):
        """URL-embedded credentials must survive being an api-key param value.

        Regression test for pattern ordering: when the param value is itself
        a URL with embedded credentials, the param pattern used to consume
        the ``https`` scheme first (``api-key=[REDACTED]://user:pass@...``),
        leaving the credentials unredacted. The URL-credentials pattern now
        runs before the URL-param pattern.
        """
        msg = "Request failed for ?api-key=https://admin:supersecret@internal.example.com/v1"
        result = sanitize_error_message(msg)
        assert "supersecret" not in result
        assert "admin" not in result

    def test_url_credentials_inside_key_param_value(self):
        """Same ordering interaction via the ``key=`` spelling."""
        msg = "?key=https://user:hunter2@host.example.com/path"
        result = sanitize_error_message(msg)
        assert "hunter2" not in result
        assert "user:" not in result

    def test_url_credentials_inside_token_param_value(self):
        """Same ordering interaction via the ``token=`` spelling."""
        msg = "retry failed: &token=http://svc:p4ssw0rd@10.0.0.5:8080/auth"
        result = sanitize_error_message(msg)
        assert "p4ssw0rd" not in result
        assert "svc:" not in result


class TestDatabaseDsnCredentials:
    """URL-credential redaction covers database DSNs, not just http(s).

    A raw SQLAlchemy / driver connection error is the most common way a
    credential-bearing string reaches an exception message. The userinfo in
    any URL scheme is a credential, so it is redacted regardless of scheme.
    """

    def test_postgres_dsn_password_redacted(self):
        msg = (
            "OperationalError: could not connect to "
            "postgresql://newsuser:s3cr3tPw0rd@db.internal:5432/news"
        )
        result = sanitize_error_message(msg)
        assert "s3cr3tPw0rd" not in result
        assert "newsuser" not in result
        assert "[REDACTED]:[REDACTED]@db.internal" in result

    def test_sqlalchemy_dialect_driver_scheme_redacted(self):
        # SQLAlchemy DSNs carry a `dialect+driver` scheme. Each DSN is built
        # by f-string concatenation so no literal "user:password@" pair
        # appears in source — keeps regex-based secret scanners from flagging
        # classic default-cred shapes like root:toor alongside the MongoDB
        # fixture that originally triggered alert #2.
        pg_user, pg_pw = "user", "pa55"
        my_user, my_pw = "root", "toor"
        mongo_user, mongo_pw = "appuser", "letmein"
        for msg in (
            f"postgresql+psycopg2://{pg_user}:{pg_pw}@dbhost/appdb",
            f"mysql+pymysql://{my_user}:{my_pw}@localhost:3306/db",
            f"mongodb+srv://{mongo_user}:{mongo_pw}@cluster0.mongodb.net/test",
        ):
            result = sanitize_error_message(msg)
            assert "[REDACTED]:[REDACTED]@" in result, msg
            assert my_pw not in sanitize_error_message(
                f"mysql+pymysql://{my_user}:{my_pw}@localhost:3306/db"
            )

    def test_password_only_redis_dsn_redacted(self):
        # redis uses a password-only DSN (empty username).
        msg = "redis://:mypassword@127.0.0.1:6379/0 connection reset"
        result = sanitize_error_message(msg)
        assert "mypassword" not in result
        assert "[REDACTED]:[REDACTED]@" in result

    def test_credential_less_dsn_with_port_unchanged(self):
        # No userinfo (just host:port) must NOT be touched — the trailing `@`
        # is required and `/` is excluded from the userinfo.
        for msg in (
            "connect failed to postgresql://dbhost:5432/news",
            "redis://cache.local:6379/0 timed out",
        ):
            assert sanitize_error_message(msg) == msg

    def test_http_credentials_still_redacted(self):
        # Regression guard: generalizing the scheme must not lose http(s).
        msg = "Connection to https://admin:supersecretpassword@api.example.com failed"
        result = sanitize_error_message(msg)
        assert "supersecretpassword" not in result
        assert "[REDACTED]:[REDACTED]@" in result

    def test_uppercase_scheme_redacted(self):
        # URL schemes are case-insensitive (RFC 3986); an uppercase/mixed-case
        # DSN scheme must not defeat redaction.
        for msg in (
            "POSTGRESQL://newsuser:s3cr3tPw0rd@db.internal:5432/news",
            "Postgresql+Psycopg2://user:pa55@dbhost/appdb",
        ):
            result = sanitize_error_message(msg)
            assert "s3cr3tPw0rd" not in result
            assert "pa55" not in result
            assert "[REDACTED]:[REDACTED]@" in result

    def test_url_glued_to_preceding_word_char_still_redacted(self):
        # Regression guard against a leading `\b` anchor: a credential URL
        # glued to a preceding word char must still have its userinfo redacted
        # (the scheme's first char is a word char, so `\b` would fail to anchor
        # here and leak the password).
        msg = "prefixXhttps://admin:supersecretpassword@host/path"
        result = sanitize_error_message(msg)
        assert "supersecretpassword" not in result
        assert "[REDACTED]:[REDACTED]@" in result

    def test_sk_prefix_redacted(self):
        """OpenAI-style key with modern hyphenated format (sk-proj-...)."""
        msg = "Invalid API key: sk-proj-abc123xyz456def789ghi012jkl"
        result = sanitize_error_message(msg)
        assert "sk-proj-abc123xyz456def789ghi012jkl" not in result
        assert "[REDACTED_KEY]" in result

    def test_sk_prefix_pure_alphanumeric(self):
        """Old-style OpenAI key (pure alphanumeric after sk-)."""
        msg = "key: sk-ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
        result = sanitize_error_message(msg)
        assert "sk-ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" not in result
        assert "[REDACTED_KEY]" in result

    def test_pk_prefix_redacted(self):
        msg = "public key pk-test-abc123xyz456def789ghi012jkl345"
        result = sanitize_error_message(msg)
        assert "pk-test-abc123xyz456def789ghi012jkl345" not in result
        assert "[REDACTED_KEY]" in result

    def test_empty_string_passthrough(self):
        assert sanitize_error_message("") == ""

    def test_no_secrets_unchanged(self):
        msg = (
            "Connection timeout after 10s for https://api.example.com/v1/models"
        )
        assert sanitize_error_message(msg) == msg

    def test_multiple_secrets_in_one_message(self):
        msg = (
            "Bearer sk-proj-abc123xyz456def789ghi012 "
            "with ?api_key=my-secret-key-12345&token=another-secret-value-67890"
        )
        result = sanitize_error_message(msg)
        assert "sk-proj-abc123xyz456def789ghi012" not in result
        assert "my-secret-key-12345" not in result
        assert "another-secret-value-67890" not in result
        assert "Bearer [REDACTED]" in result
        assert "api_key=[REDACTED]" in result
        assert "token=[REDACTED]" in result

    def test_sk_prefix_too_short_not_redacted(self):
        """Short strings like ``sk-abc`` should NOT be redacted (false positive risk)."""
        msg = "prefix: sk-abc"
        assert sanitize_error_message(msg) == msg

    def test_delegate_preserves_backward_compat(self):
        """The BaseSearchEngine method delegates to the standalone function,
        so this test verifies the output matches the old inline behavior
        for a representative message."""
        from local_deep_research.web_search_engines.search_engine_base import (
            BaseSearchEngine,
        )

        msg = "Error: ?api_key=secret12345678&key=mykey1234567890123456"
        expected = sanitize_error_message(msg)

        # Create a minimal subclass instance to call the instance method.
        class _MinimalEngine(BaseSearchEngine):
            def _get_previews(self, query, *args, **kwargs):
                return []

        engine = _MinimalEngine.__new__(_MinimalEngine)
        actual = engine._sanitize_error_message(msg)
        assert actual == expected


class TestSanitizeErrorDetails:
    """Unit tests for the shared ``sanitize_error_details`` helper used by the
    exception ``to_dict()`` backstops (NewsAPIException / WebAPIException)."""

    def test_str_leaf_credential_redacted(self):
        from local_deep_research.security.log_sanitizer import (
            sanitize_error_details,
        )

        out = sanitize_error_details(
            {"note": "Authorization: Bearer sk-abc123DEF456ghi789"}
        )
        assert "sk-abc123DEF456ghi789" not in out["note"]
        assert "REDACTED" in out["note"]

    def test_benign_and_non_string_leaves_pass_through(self):
        from local_deep_research.security.log_sanitizer import (
            sanitize_error_details,
        )

        value = {
            "query": "reset my password",
            "count": 3,
            "ok": True,
            "x": None,
        }
        assert sanitize_error_details(value) == value

    def test_nested_dict_list_recursion(self):
        from local_deep_research.security.log_sanitizer import (
            sanitize_error_details,
        )

        out = sanitize_error_details(
            {"outer": {"inner": ["?api_key=AIzaSyD-EXAMPLE_1234567890", 7]}}
        )
        leaf = out["outer"]["inner"][0]
        assert "AIzaSyD-EXAMPLE_1234567890" not in leaf
        assert "REDACTED" in leaf
        assert out["outer"]["inner"][1] == 7

    def test_tuple_normalizes_to_list(self):
        from local_deep_research.security.log_sanitizer import (
            sanitize_error_details,
        )

        assert sanitize_error_details({"pair": ("a", "b")})["pair"] == [
            "a",
            "b",
        ]

    def test_namedtuple_does_not_crash(self):
        from collections import namedtuple

        from local_deep_research.security.log_sanitizer import (
            sanitize_error_details,
        )

        Pair = namedtuple("Pair", ["a", "b"])
        out = sanitize_error_details(
            {"pair": Pair("Bearer sk-abc123DEF456ghi789", "x")}
        )["pair"]
        assert isinstance(out, list)
        assert "sk-abc123DEF456ghi789" not in out[0]
        assert "REDACTED" in out[0]
        assert out[1] == "x"

    def test_input_is_not_mutated(self):
        from local_deep_research.security.log_sanitizer import (
            sanitize_error_details,
        )

        original = {"note": "Bearer sk-secret999value000", "keep": [1, 2]}
        out = sanitize_error_details(original)
        assert original["note"] == "Bearer sk-secret999value000"
        assert out["note"] != original["note"]
        assert out["keep"] is not original["keep"]

    def test_non_container_scalar_passthrough(self):
        from local_deep_research.security.log_sanitizer import (
            sanitize_error_details,
        )

        assert sanitize_error_details(42) == 42
        assert sanitize_error_details(None) is None

    def test_dataclass_leaf_is_recursed_and_redacted(self):
        # Flask's default JSON provider serializes a dataclass via asdict(), so
        # a credential in a dataclass field would ship to the client unless the
        # helper recurses into it.
        import dataclasses

        from local_deep_research.security.log_sanitizer import (
            sanitize_error_details,
        )

        @dataclasses.dataclass
        class Ctx:
            note: str
            count: int

        out = sanitize_error_details(
            {"ctx": Ctx("Authorization: Bearer sk-abc123DEF456ghi789", 5)}
        )
        assert "sk-abc123DEF456ghi789" not in out["ctx"]["note"]
        assert "REDACTED" in out["ctx"]["note"]
        assert out["ctx"]["count"] == 5

    def test_credential_shaped_dict_key_redacted(self):
        from local_deep_research.security.log_sanitizer import (
            sanitize_error_details,
        )

        out = sanitize_error_details(
            {"Authorization: Bearer sk-abc123DEF456ghi789": "v"}
        )
        assert not any("sk-abc123DEF456ghi789" in k for k in out)
        # benign field-name keys are untouched
        assert sanitize_error_details({"username": "bob"}) == {
            "username": "bob"
        }

    def test_set_leaf_passes_through(self):
        # A set is not JSON-serializable, so it is left as-is (fails closed at
        # jsonify rather than leaking) — pin the passthrough behavior.
        from local_deep_research.security.log_sanitizer import (
            sanitize_error_details,
        )

        s = {"a", "b"}
        assert sanitize_error_details({"tags": s})["tags"] == s