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

"""Tests for log string sanitization."""

import pytest

from local_deep_research.security.log_sanitizer import (
    redact_secrets,
    sanitize_error_for_client,
    sanitize_for_log,
    scrub_error,
    strip_control_chars,
)


class TestLogSanitizer:
    """Unit tests for sanitize_for_log."""

    def test_normal_string_passes_through(self):
        assert sanitize_for_log("hello") == "hello"

    def test_non_printable_chars_stripped(self):
        assert sanitize_for_log("hello\x00world\x07") == "helloworld"

    def test_truncation_respects_max_length(self):
        result = sanitize_for_log("a" * 100, max_length=10)
        assert result == "a" * 7 + "..."
        assert len(result) == 10

    def test_no_truncation_at_exact_max_length(self):
        result = sanitize_for_log("a" * 50, max_length=50)
        assert result == "a" * 50

    def test_empty_string(self):
        assert sanitize_for_log("") == ""

    def test_newlines_stripped(self):
        assert sanitize_for_log("line1\nline2") == "line1line2"

    def test_tabs_stripped(self):
        assert sanitize_for_log("col1\tcol2") == "col1col2"

    def test_unicode_preserved(self):
        assert sanitize_for_log("café") == "café"

    def test_cjk_preserved(self):
        assert sanitize_for_log("你好") == "你好"

    def test_control_chars_stripped_unicode_preserved(self):
        assert sanitize_for_log("café\x00\x07") == "café"


class TestStripControlChars:
    """Unit tests for strip_control_chars."""

    def test_strips_c0_control_chars(self):
        assert strip_control_chars("a\x00b\x1fc") == "abc"

    def test_strips_c1_control_chars(self):
        assert strip_control_chars("a\x7fb\x9fc") == "abc"

    def test_preserves_normal_text(self):
        assert strip_control_chars("hello world") == "hello world"

    def test_preserves_unicode(self):
        assert strip_control_chars("café 你好 émoji") == "café 你好 émoji"

    def test_strips_rlo_override(self):
        assert strip_control_chars("hello\u202eworld") == "helloworld"

    def test_strips_arabic_letter_mark(self):
        assert strip_control_chars("hello\u061cworld") == "helloworld"

    def test_strips_zero_width_space(self):
        assert strip_control_chars("hello\u200bworld") == "helloworld"

    def test_strips_bom(self):
        assert strip_control_chars("\ufeffhello") == "hello"

    def test_strips_word_joiner(self):
        assert strip_control_chars("hello\u2060world") == "helloworld"

    def test_strips_digit_shape_controls(self):
        assert strip_control_chars("hello\u206aworld") == "helloworld"

    def test_strips_mixed_format_chars(self):
        assert (
            strip_control_chars("café\u202e\u200b\ufeff 你好\u2060")
            == "café 你好"
        )

    def test_empty_string(self):
        assert strip_control_chars("") == ""


class TestRedactSecrets:
    """Unit tests for redact_secrets."""

    def test_redacts_single_secret(self):
        result = redact_secrets(
            "call to ?key=sk-abc1234567 failed", "sk-abc1234567"
        )
        assert result == "call to ?key=***REDACTED*** failed"

    def test_redacts_multiple_secrets(self):
        result = redact_secrets(
            "user=alice12345 token=tok-xyz98765",
            "alice12345",
            "tok-xyz98765",
        )
        assert "alice12345" not in result
        assert "tok-xyz98765" not in result
        assert result.count("***REDACTED***") == 2

    def test_redacts_all_occurrences(self):
        result = redact_secrets("X sk-12345678 Y sk-12345678 Z", "sk-12345678")
        assert result == "X ***REDACTED*** Y ***REDACTED*** Z"

    def test_none_secret_ignored(self):
        assert redact_secrets("message stays put", None) == "message stays put"

    def test_empty_secret_ignored(self):
        # Replacing the empty string would insert the token between every
        # character of the message; this is the load-bearing guard.
        assert redact_secrets("message stays put", "") == "message stays put"

    def test_short_secret_below_min_length_ignored(self):
        # 7 characters is below the default min_length of 8.
        result = redact_secrets("password is hunter", "hunter")
        assert result == "password is hunter"

    def test_min_length_parameter_lowers_threshold(self):
        result = redact_secrets("password is hunter", "hunter", min_length=6)
        assert result == "password is ***REDACTED***"

    def test_no_secrets_returns_message_unchanged(self):
        assert redact_secrets("hello world") == "hello world"

    def test_empty_message_returned_unchanged(self):
        assert redact_secrets("", "sk-abc1234567") == ""

    def test_message_without_secret_returned_unchanged(self):
        assert redact_secrets("hello world", "sk-abc1234567") == "hello world"

    def test_custom_replacement(self):
        result = redact_secrets(
            "key=sk-abc1234567", "sk-abc1234567", replacement="[KEY]"
        )
        assert result == "key=[KEY]"

    def test_empty_replacement_strips_secret(self):
        # Replacement may be empty — strips the secret entirely. This
        # is the right answer when the secret's presence itself is
        # sensitive (not just its value).
        result = redact_secrets(
            "before sk-abc1234567 after", "sk-abc1234567", replacement=""
        )
        assert result == "before  after"

    def test_overlapping_secrets_redacted_longest_first(self):
        # If two secrets overlap (one is a substring of the other), the
        # function must apply the longer one first so a shorter
        # secret cannot consume part of the longer match. Without
        # length-sorting, the test fails: redacting "abc12345" first
        # would leave "sk-***REDACTED***" in the message, then the
        # longer "sk-abc12345" no longer matches.
        result = redact_secrets(
            "found sk-abc12345 here", "abc12345", "sk-abc12345"
        )
        assert result == "found ***REDACTED*** here"
        assert "abc12345" not in result

    def test_redaction_is_not_recursive(self):
        # If a secret happens to equal the replacement token, the
        # function does not loop forever — it does a single pass per
        # secret and ``str.replace`` is not recursive.
        result = redact_secrets("X ***REDACTED*** Y", "***REDACTED***")
        # The pre-existing token gets replaced with the same token —
        # net no-op, but importantly: no recursion, no exception.
        assert result == "X ***REDACTED*** Y"

    def test_importable_from_security_package(self):
        # ``redact_secrets`` is exported from
        # ``local_deep_research.security`` so future callers don't need
        # to know the submodule path.
        from local_deep_research.security import (
            redact_secrets as exported,
        )

        assert exported is redact_secrets

    @pytest.mark.parametrize(
        "secret",
        [
            "sk-abc1234567890",
            "AIzaSy-mock-google-key-12345",
            "sk-ant-api03-very-long-anthropic-key-12345",
        ],
    )
    def test_realistic_provider_key_shapes_redacted(self, secret):
        message = f"upstream failed: ?key={secret}&model=x"
        assert secret not in redact_secrets(message, secret)

    def test_literal_substring_match_only(self):
        # Document the contract: URL-encoded or otherwise transformed
        # forms are NOT redacted. Callers must pass each form they need
        # to scrub.
        result = redact_secrets("encoded=%2Bsk-abc12345", "+sk-abc12345")
        assert "%2Bsk-abc12345" in result


class TestSanitizeErrorForClient:
    """The client-facing composition used for exception text returned over
    HTTP/SSE (e.g. the library download stream)."""

    def test_redacts_api_key_in_url(self):
        msg = (
            "HTTPError for https://api.example.com/doc?api_key=secret1234567890"
        )
        result = sanitize_error_for_client(msg)
        assert "secret1234567890" not in result

    def test_redacts_url_embedded_credentials(self):
        msg = "ConnectionError: https://alice:hunter2pass@host/file.pdf failed"
        result = sanitize_error_for_client(msg)
        assert "hunter2pass" not in result

    def test_truncates_to_max_length(self):
        result = sanitize_error_for_client("e" * 500, max_length=200)
        assert len(result) <= 200

    def test_credential_scrubbed_before_truncation(self):
        # The whole point of scrub-before-truncate: a key sitting past the
        # max_length boundary must still be redacted, not merely cut off
        # (a truncate-first order would leave the leading chars exposed if
        # the cut landed mid-token).
        key = "sk-abcdefghijklmnopqrstuvwxyz0123456789"
        msg = "x" * 190 + " " + key
        result = sanitize_error_for_client(msg, max_length=400)
        assert key not in result
        assert "sk-abcdefghij" not in result

    @pytest.mark.parametrize(
        "msg, secret",
        [
            (
                "err https://api.x.com?access_token=SECRET_AT_123 fail",
                "SECRET_AT_123",
            ),
            ("https://api.x.com?refresh_token=SECRET_RT_123", "SECRET_RT_123"),
            (
                "https://x.cognitive.microsoft.com?subscription-key=SECRET_SUB1",
                "SECRET_SUB1",
            ),
            ("oauth ?client_secret=SECRET_CS_123&grant=x", "SECRET_CS_123"),
            ("?secret_key=SECRET_SK_123&page=2", "SECRET_SK_123"),
            (
                "401 Authorization: Basic SECRETBASICVALUE123 denied",
                "SECRETBASICVALUE123",
            ),
            (
                "Authorization: token ghp_TOKENVALUE1234567890 x",
                "ghp_TOKENVALUE1234567890",
            ),
            (
                "req x-api-key: SECRETXAPIKEYVALUE12345 blocked",
                "SECRETXAPIKEYVALUE12345",
            ),
            # Schemeless Authorization carrying a token-shaped raw value.
            (
                "Authorization: ghp_RAWTOKEN1234567890 rejected",
                "ghp_RAWTOKEN1234567890",
            ),
            # Anchored (scheme / x-api-key) values redact regardless of shape,
            # incl. purely-alphabetic ones — the anchor rules out prose.
            (
                "Authorization: Basic ABCDEFGHIJKLMNOP denied",
                "ABCDEFGHIJKLMNOP",
            ),
            (
                "x-api-key: ALPHAONLYKEYVALUEABCDEFGH blocked",
                "ALPHAONLYKEYVALUEABCDEFGH",
            ),
            (
                "google AIzaSyD-SECRET_GOOGLE_KEY_1234567890 abc",
                "AIzaSyD-SECRET_GOOGLE_KEY_1234567890",
            ),
            # Case-insensitive query params (capitalized names must not bypass).
            ("https://x.com?API_KEY=SECRET_UP1", "SECRET_UP1"),
            ("https://x.com?Access_Token=SECRET_MIX1", "SECRET_MIX1"),
        ],
    )
    def test_redacts_additional_credential_formats(self, msg, secret):
        """OAuth tokens, Azure subscription keys, Authorization/x-api-key
        headers, Google API keys, and case-variant query params must all be
        scrubbed (#4633 follow-up)."""
        assert secret not in sanitize_error_for_client(msg)

    @pytest.mark.parametrize(
        "msg, preserved",
        [
            ("page request ?page=2&limit=10", "page=2"),
            ("Basic understanding of the topic", "understanding"),
            ("key concepts explained here", "concepts"),
            ("fetched https://example.com/path?id=42 ok", "id=42"),
            # Schemeless "Authorization:" followed by an all-alphabetic word
            # is prose, not a token — the token-shaped requirement on the
            # schemeless branch keeps it intact.
            ("401: Authorization: required for this endpoint", "required"),
            ("Authorization: denied because user lacks role", "denied"),
            # Short values below the floors are left intact (x-api-key needs
            # >=16, the scheme branch needs >=8).
            ("Response 403: x-api-key: invalid or expired", "invalid"),
            ("Header x-api-key: missing from request", "missing"),
            ("Authorization: bearer ab12 failed", "ab12"),
            ("x-api-key: abc12 next", "abc12"),
        ],
    )
    def test_does_not_over_redact_innocuous_text(self, msg, preserved):
        """Non-credential query params, schemeless-Authorization prose, and
        sub-floor values must survive untouched (no false-positive
        redaction)."""
        result = sanitize_error_for_client(msg)
        assert preserved in result
        assert "REDACTED" not in result

    @pytest.mark.parametrize(
        "msg, secret",
        [
            (
                "failed ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789 here",
                "ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789",
            ),
            (
                "github_pat_11ABCDE0aBcDeFgHiJ_kLmNoPqRsTuVwXyZ0123456789AbCdEf x",
                "github_pat_11ABCDE0aBcDeFgHiJ_kLmNoPqRsTuVwXyZ0123456789AbCdEf",
            ),
            ("AWS AKIAIOSFODNN7EXAMPLE rejected", "AKIAIOSFODNN7EXAMPLE"),
            ("creds ASIAY34FZKBOKMUTVV7A expired", "ASIAY34FZKBOKMUTVV7A"),
            (
                # Prefix split ("xox" + ...) so the literal isn't a scannable
                # Slack-token shape that trips GitHub push protection.
                "slack " + "xox" + "b-2345678901-abcdefABCDEF123456 failed",
                "xox" + "b-2345678901-abcdefABCDEF123456",
            ),
            (
                "cfg " + "xox" + "e-1-AB-2345678901-abcdef123456 expired",
                "xox" + "e-1-AB-2345678901-abcdef123456",
            ),
            (
                "app " + "xa" + "pp-1-A012345-678901234567-abcdef0123 bad",
                "xa" + "pp-1-A012345-678901234567-abcdef0123",
            ),
            (
                "token ya29.a0AfH6SMBsecretOAUTH_token-12345 expired",
                "ya29.a0AfH6SMBsecretOAUTH_token-12345",
            ),
            (
                "auth eyJhbGciOiJIUzI1Ni2.eyJzdWIiOiIxMjM0NTY3OD2.dozjgNryP4 bad",
                # assert the WHOLE JWT is gone, not just one segment
                "eyJhbGciOiJIUzI1Ni2.eyJzdWIiOiIxMjM0NTY3OD2.dozjgNryP4",
            ),
        ],
    )
    def test_redacts_provider_token_prefixes(self, msg, secret):
        """GitHub/AWS/Slack/Google-OAuth tokens and JWTs (canonical gitleaks
        prefixes) must be scrubbed from error text."""
        assert secret not in sanitize_error_for_client(msg)

    @pytest.mark.parametrize(
        "msg, preserved",
        [
            ("the AKIApattern doc explains this", "AKIApattern"),
            ("ghp build pipeline failed", "ghp"),
            ("see ya29 release notes for details", "ya29"),
            ("eyJot just an ordinary word here", "eyJot"),
            # The generic Slack ``xapp``/``xox`` prefixes require a long
            # numeric workspace ID, so hyphenated prose that merely starts
            # with them (and carries no 9+ digit run) survives untouched.
            ("xapp-release-notes-2026 published", "xapp-release-notes-2026"),
            ("rolling out xapp-config-reload now", "xapp-config-reload"),
            ("xoxe-release-bundle-2026 attached", "xoxe-release-bundle-2026"),
            ("module xoxb-internal-doc loaded", "xoxb-internal-doc"),
        ],
    )
    def test_token_prefixes_do_not_over_redact_prose(self, msg, preserved):
        """The prefix patterns are distinctive/length-floored — ordinary
        prose that merely starts with a prefix is not redacted."""
        result = sanitize_error_for_client(msg)
        assert preserved in result
        assert "REDACTED" not in result

    def test_importable_from_security_package(self):
        from local_deep_research.security import (
            sanitize_error_for_client as exported,
        )

        assert exported is sanitize_error_for_client


class TestScrubError:
    """Unit tests for scrub_error (the shared dual-scrub helper)."""

    def test_scrubs_credential_shapes_from_exception(self):
        err = ValueError("401 for Bearer sk-abc1234567890abcdef")
        result = scrub_error(err)
        assert "sk-abc1234567890abcdef" not in result

    def test_redacts_literal_secrets(self):
        # A bare key with no recognizable shape only falls to the
        # literal pass — the reason the dual-scrub exists.
        err = RuntimeError("auth failed for key plainkey12345678")
        result = scrub_error(err, "plainkey12345678")
        assert "plainkey12345678" not in result
        assert "***REDACTED***" in result

    def test_accepts_prebuilt_message_string(self):
        result = scrub_error("token tok-xyz9876543 rejected", "tok-xyz9876543")
        assert "tok-xyz9876543" not in result

    def test_non_string_secret_coerced(self):
        # A misconfigured numeric secret must not crash the except
        # handler that calls this (redact_secrets calls len()).
        err = RuntimeError("key 123456789012 invalid")
        result = scrub_error(err, 123456789012)
        assert "123456789012" not in result

    def test_none_and_falsy_secrets_skipped(self):
        msg = "plain message"
        assert scrub_error(RuntimeError(msg), None, "", 0) == msg

    def test_multiple_secrets_all_redacted(self):
        # Kills the "forward only the first secret" mutant: every
        # truthy secret must survive the coercion. (Longest-first
        # ordering itself is pinned by
        # test_overlapping_secrets_redacted_longest_first.)
        err = RuntimeError("user token1234567 key sk-abc12345 done")
        result = scrub_error(err, "token1234567", "abc12345", "sk-abc12345")
        assert "token1234567" not in result
        assert "abc12345" not in result

    def test_pathological_secret_does_not_crash(self):
        # The never-raise contract covers secrets too: a value whose
        # __bool__/__str__ raises is skipped, not propagated out of the
        # except handler; the other secrets are still redacted.
        class BadBool:
            def __bool__(self):
                raise ValueError("nobool")

        err = RuntimeError("key goodsecret123456 invalid")
        result = scrub_error(err, BadBool(), "goodsecret123456")
        assert "goodsecret123456" not in result

    def test_unprintable_exception_guarded(self):
        class Unprintable(Exception):
            def __str__(self):
                raise RuntimeError("boom")

        result = scrub_error(Unprintable())
        assert result == "<unprintable Unprintable>"

    def test_matches_engine_scrub_error(self):
        # BaseSearchEngine._scrub_error delegates here; the two must
        # produce identical output for the same inputs.
        from local_deep_research.web_search_engines.search_engine_base import (
            BaseSearchEngine,
        )

        class _Engine(BaseSearchEngine):
            def _get_previews(self, query):
                return []

            def _get_full_content(self, relevant_items):
                return relevant_items

        # __new__ skips BaseSearchEngine.__init__ — only _scrub_error's
        # _secret_attrs lookup matters here.
        engine = _Engine.__new__(_Engine)
        engine.api_key = "literalsecret123"
        err = RuntimeError("denied for literalsecret123")
        assert engine._scrub_error(err) == scrub_error(err, "literalsecret123")

    def test_exported_from_security_package(self):
        from local_deep_research.security import scrub_error as exported

        assert exported is scrub_error