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

"""
Tests for DataSanitizer security module.
"""

from local_deep_research.security.data_sanitizer import (
    DataSanitizer,
    filter_research_metadata,
    redact_data,
    sanitize_data,
    strip_settings_snapshot,
)


class TestDataSanitizerSanitize:
    """Tests for DataSanitizer.sanitize()."""

    def test_sanitize_removes_api_key(self):
        """Removes api_key from data."""
        data = {"username": "user", "api_key": "sk-secret-12345"}
        result = DataSanitizer.sanitize(data)
        assert "username" in result
        assert "api_key" not in result

    def test_sanitize_removes_password(self):
        """Removes password from data."""
        data = {"user": "test", "password": "supersecret"}
        result = DataSanitizer.sanitize(data)
        assert "user" in result
        assert "password" not in result

    def test_sanitize_removes_access_token(self):
        """Removes access_token from data."""
        data = {"id": 1, "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"}
        result = DataSanitizer.sanitize(data)
        assert "id" in result
        assert "access_token" not in result

    def test_sanitize_case_insensitive(self):
        """Sanitization is case-insensitive."""
        data = {"API_KEY": "secret1", "ApiKey": "secret2", "apikey": "secret3"}
        result = DataSanitizer.sanitize(data)
        assert len(result) == 0

    def test_sanitize_nested_dict(self):
        """Sanitizes nested dictionaries."""
        data = {
            "user": "test",
            "settings": {
                "theme": "dark",
                "api_key": "nested_secret",
            },
        }
        result = DataSanitizer.sanitize(data)
        assert result["user"] == "test"
        assert "theme" in result["settings"]
        assert "api_key" not in result["settings"]

    def test_sanitize_deeply_nested(self):
        """Sanitizes deeply nested structures."""
        data = {
            "level1": {
                "level2": {
                    "level3": {
                        "password": "deep_secret",
                        "value": "keep_this",
                    }
                }
            }
        }
        result = DataSanitizer.sanitize(data)
        assert result["level1"]["level2"]["level3"]["value"] == "keep_this"
        assert "password" not in result["level1"]["level2"]["level3"]

    def test_sanitize_list_of_dicts(self):
        """Sanitizes list of dictionaries."""
        data = [
            {"name": "item1", "secret": "value1"},
            {"name": "item2", "secret": "value2"},
        ]
        result = DataSanitizer.sanitize(data)
        assert len(result) == 2
        assert result[0]["name"] == "item1"
        assert "secret" not in result[0]
        assert result[1]["name"] == "item2"
        assert "secret" not in result[1]

    def test_sanitize_mixed_structure(self):
        """Sanitizes mixed nested structures."""
        data = {
            "users": [
                {"username": "user1", "auth_token": "token1"},
                {"username": "user2", "auth_token": "token2"},
            ],
            "config": {
                "public": True,
                "private_key": "key123",
            },
        }
        result = DataSanitizer.sanitize(data)
        assert result["users"][0]["username"] == "user1"
        assert "auth_token" not in result["users"][0]
        assert result["config"]["public"] is True
        assert "private_key" not in result["config"]

    def test_sanitize_preserves_primitives(self):
        """Preserves primitive values unchanged."""
        assert DataSanitizer.sanitize("string") == "string"
        assert DataSanitizer.sanitize(123) == 123
        assert DataSanitizer.sanitize(12.5) == 12.5
        assert DataSanitizer.sanitize(True) is True
        assert DataSanitizer.sanitize(None) is None

    def test_sanitize_empty_dict(self):
        """Handles empty dictionary."""
        assert DataSanitizer.sanitize({}) == {}

    def test_sanitize_empty_list(self):
        """Handles empty list."""
        assert DataSanitizer.sanitize([]) == []

    def test_sanitize_custom_keys(self):
        """Uses custom sensitive keys."""
        data = {
            "custom_secret": "value",
            "api_key": "standard_secret",
            "name": "keep",
        }
        result = DataSanitizer.sanitize(data, sensitive_keys={"custom_secret"})
        assert "custom_secret" not in result
        assert "api_key" in result  # Not in custom keys
        assert "name" in result

    def test_sanitize_all_default_keys(self):
        """Tests all default sensitive keys."""
        data = {
            "api_key": "1",
            "apikey": "2",
            "password": "3",
            "secret": "4",
            "access_token": "5",
            "refresh_token": "6",
            "private_key": "7",
            "auth_token": "8",
            "session_token": "9",
            "csrf_token": "10",
            "safe_key": "keep",
        }
        result = DataSanitizer.sanitize(data)
        assert result == {"safe_key": "keep"}


class TestDataSanitizerRedact:
    """Tests for DataSanitizer.redact()."""

    def test_redact_replaces_api_key(self):
        """Replaces api_key value with redaction text."""
        data = {"username": "user", "api_key": "sk-secret-12345"}
        result = DataSanitizer.redact(data)
        assert result["username"] == "user"
        assert result["api_key"] == "[REDACTED]"

    def test_redact_preserves_structure(self):
        """Preserves data structure while redacting values."""
        data = {"id": 1, "password": "secret123"}
        result = DataSanitizer.redact(data)
        assert "id" in result
        assert "password" in result
        assert result["password"] == "[REDACTED]"

    def test_redact_custom_text(self):
        """Uses custom redaction text."""
        data = {"password": "secret"}
        result = DataSanitizer.redact(data, redaction_text="***")
        assert result["password"] == "***"

    def test_redact_case_insensitive(self):
        """Redaction is case-insensitive."""
        data = {"API_KEY": "secret1", "ApiKey": "secret2", "apikey": "secret3"}
        result = DataSanitizer.redact(data)
        assert all(v == "[REDACTED]" for v in result.values())

    def test_redact_nested_dict(self):
        """Redacts nested dictionaries."""
        data = {
            "user": "test",
            "settings": {
                "theme": "dark",
                "api_key": "nested_secret",
            },
        }
        result = DataSanitizer.redact(data)
        assert result["user"] == "test"
        assert result["settings"]["theme"] == "dark"
        assert result["settings"]["api_key"] == "[REDACTED]"

    def test_redact_list_of_dicts(self):
        """Redacts list of dictionaries."""
        data = [
            {"name": "item1", "secret": "value1"},
            {"name": "item2", "secret": "value2"},
        ]
        result = DataSanitizer.redact(data)
        assert result[0]["name"] == "item1"
        assert result[0]["secret"] == "[REDACTED]"
        assert result[1]["name"] == "item2"
        assert result[1]["secret"] == "[REDACTED]"

    def test_redact_preserves_primitives(self):
        """Preserves primitive values unchanged."""
        assert DataSanitizer.redact("string") == "string"
        assert DataSanitizer.redact(123) == 123
        assert DataSanitizer.redact(True) is True

    def test_redact_custom_keys(self):
        """Uses custom sensitive keys."""
        data = {"custom_secret": "value", "api_key": "keep", "name": "keep"}
        result = DataSanitizer.redact(data, sensitive_keys={"custom_secret"})
        assert result["custom_secret"] == "[REDACTED]"
        assert result["api_key"] == "keep"  # Not in custom keys
        assert result["name"] == "keep"


class TestConvenienceFunctions:
    """Tests for convenience functions."""

    def test_sanitize_data_function(self):
        """sanitize_data() works correctly."""
        data = {"username": "user", "password": "secret"}
        result = sanitize_data(data)
        assert "username" in result
        assert "password" not in result

    def test_sanitize_data_with_custom_keys(self):
        """sanitize_data() with custom keys."""
        data = {"custom": "secret", "keep": "value"}
        result = sanitize_data(data, sensitive_keys={"custom"})
        assert "custom" not in result
        assert "keep" in result

    def test_redact_data_function(self):
        """redact_data() works correctly."""
        data = {"username": "user", "password": "secret"}
        result = redact_data(data)
        assert result["username"] == "user"
        assert result["password"] == "[REDACTED]"

    def test_redact_data_with_custom_text(self):
        """redact_data() with custom redaction text."""
        data = {"password": "secret"}
        result = redact_data(data, redaction_text="<hidden>")
        assert result["password"] == "<hidden>"


class TestDefaultSensitiveKeys:
    """Tests for DEFAULT_SENSITIVE_KEYS constant."""

    def test_default_keys_exist(self):
        """Default sensitive keys are defined."""
        assert isinstance(DataSanitizer.DEFAULT_SENSITIVE_KEYS, set)
        assert len(DataSanitizer.DEFAULT_SENSITIVE_KEYS) > 0

    def test_default_keys_include_common_secrets(self):
        """Default keys include common secret patterns."""
        expected = {
            "api_key",
            "password",
            "secret",
            "access_token",
            "refresh_token",
            "private_key",
        }
        assert expected.issubset(DataSanitizer.DEFAULT_SENSITIVE_KEYS)


class TestExpandedSecretLeafNames:
    """Secret leaf-names beyond the classic set must be caught by the
    name-only predicate the bulk settings GET relies on (ui_element=None).
    """

    NEW_NAMES = [
        "client_secret",
        "secret_key",
        "bearer_token",
        "api_secret",
        "app_secret",
    ]

    def test_present_in_default_set(self):
        assert set(self.NEW_NAMES).issubset(
            DataSanitizer.DEFAULT_SENSITIVE_KEYS
        )

    def test_recognized_by_leaf(self):
        for name in self.NEW_NAMES:
            key = f"integrations.oauth.{name}"
            assert DataSanitizer.is_sensitive_setting(key)
            # mirrors the bulk GET call: redact_value(key, None, value)
            assert (
                DataSanitizer.redact_value(key, None, "topsecret-value")
                == DataSanitizer.REDACTION_TEXT
            )

    def test_no_over_redaction_of_similar_names(self):
        # exact-leaf match must NOT redact these legit non-secret keys.
        # The last two leaves CONTAIN a sensitive name as a proper substring
        # ("password" in "passwordless", "secret" in "secretary") — they pin
        # the predicate to exact-match, so a substring-broadening regression
        # (e.g. `any(s in leaf ...)`) would be caught here.
        for key in (
            "llm.max_tokens",
            "cache.key",
            "search.public_key",
            "app.token_count",
            "oauth.client_id",
            "llm.passwordless",
            "integrations.secretary",
        ):
            assert DataSanitizer.redact_value(key, None, "value") == "value"


class TestEdgeCases:
    """Edge case tests for DataSanitizer."""

    def test_sanitize_with_none_sensitive_keys(self):
        """Uses default keys when sensitive_keys is None."""
        data = {"api_key": "secret", "name": "keep"}
        result = DataSanitizer.sanitize(data, sensitive_keys=None)
        assert "api_key" not in result
        assert "name" in result

    def test_sanitize_with_empty_sensitive_keys(self):
        """Empty sensitive keys keeps all data."""
        data = {"api_key": "secret", "password": "pass"}
        result = DataSanitizer.sanitize(data, sensitive_keys=set())
        assert "api_key" in result
        assert "password" in result

    def test_sanitize_list_with_primitives(self):
        """Handles list with primitive values."""
        data = [1, 2, "string", None, True]
        result = DataSanitizer.sanitize(data)
        assert result == [1, 2, "string", None, True]

    def test_sanitize_does_not_mutate_original(self):
        """Sanitization does not mutate original data."""
        original = {"api_key": "secret", "name": "keep"}
        _ = DataSanitizer.sanitize(original)
        assert "api_key" in original
        assert original["api_key"] == "secret"

    def test_redact_does_not_mutate_original(self):
        """Redaction does not mutate original data."""
        original = {"password": "secret", "name": "keep"}
        _ = DataSanitizer.redact(original)
        assert original["password"] == "secret"


class TestFilterResearchMetadata:
    """Tests for filter_research_metadata() helper."""

    def test_none_input(self):
        """None returns default."""
        assert filter_research_metadata(None) == {"is_news_search": False}

    def test_empty_dict(self):
        """Empty dict returns default."""
        assert filter_research_metadata({}) == {"is_news_search": False}

    def test_strips_settings_snapshot(self):
        """Extracts only is_news_search, strips settings_snapshot."""
        meta = {
            "is_news_search": True,
            "settings_snapshot": {"api_key": "sk-secret"},
        }
        result = filter_research_metadata(meta)
        assert result == {"is_news_search": True}
        assert "settings_snapshot" not in result

    def test_explicit_false_preserved(self):
        """Explicit is_news_search=False is preserved."""
        assert filter_research_metadata({"is_news_search": False}) == {
            "is_news_search": False
        }

    def test_json_string_input(self):
        """JSON string is parsed correctly."""
        result = filter_research_metadata('{"is_news_search": true}')
        assert result == {"is_news_search": True}

    def test_invalid_json_string(self):
        """Invalid JSON string returns default."""
        assert filter_research_metadata("invalid json") == {
            "is_news_search": False
        }

    def test_non_dict_json(self):
        """Non-dict JSON (e.g. array) returns default."""
        assert filter_research_metadata("[]") == {"is_news_search": False}

    def test_non_string_non_dict(self):
        """Non-string, non-dict input (e.g. int) returns default."""
        assert filter_research_metadata(42) == {"is_news_search": False}


class TestStripSettingsSnapshot:
    """Tests for strip_settings_snapshot() helper."""

    def test_none_input(self):
        """None returns empty dict."""
        assert strip_settings_snapshot(None) == {}

    def test_strips_settings_snapshot(self):
        """Removes settings_snapshot, keeps other keys."""
        meta = {
            "phase": "complete",
            "settings_snapshot": {"api_key": "sk-secret"},
        }
        assert strip_settings_snapshot(meta) == {"phase": "complete"}

    def test_no_op_when_absent(self):
        """Returns same dict when settings_snapshot absent."""
        meta = {"phase": "error", "error_type": "timeout"}
        assert strip_settings_snapshot(meta) == meta

    def test_only_exact_key_stripped(self):
        """Only strips exact 'settings_snapshot' key, not similar names."""
        meta = {"settings_version": 2, "settings_snapshot": {}}
        assert strip_settings_snapshot(meta) == {"settings_version": 2}

    def test_json_string_input(self):
        """JSON string is parsed and stripped."""
        result = strip_settings_snapshot(
            '{"phase": "done", "settings_snapshot": {}}'
        )
        assert result == {"phase": "done"}

    def test_invalid_json_string(self):
        """Invalid JSON string returns empty dict."""
        assert strip_settings_snapshot("invalid json") == {}

    def test_does_not_mutate_original(self):
        """Original input dict is NOT modified."""
        original = {
            "phase": "complete",
            "settings_snapshot": {"api_key": "sk-secret"},
        }
        _ = strip_settings_snapshot(original)
        assert "settings_snapshot" in original
        assert original["settings_snapshot"]["api_key"] == "sk-secret"


class TestRedactValueNestedDict:
    """redact_value must recurse into subtree dicts.

    A namespace request (e.g. get_setting("llm")) returns a nested dict whose
    outer key is not sensitive but whose inner keys are — without recursion
    those nested secrets would ship in the clear via GET /settings/api/bulk.
    """

    def test_nested_sensitive_leaf_redacted(self):
        value = {"provider": "openai", "openai.api_key": "sk-secret"}
        out = DataSanitizer.redact_value("llm", None, value)
        assert out["openai.api_key"] == DataSanitizer.REDACTION_TEXT
        assert out["provider"] == "openai"

    def test_deeply_nested_secret_redacted(self):
        value = {"openai": {"api_key": "sk-secret"}}
        out = DataSanitizer.redact_value("llm", None, value)
        assert out["openai"]["api_key"] == DataSanitizer.REDACTION_TEXT

    def test_sensitive_outer_key_with_dict_redacted_wholesale(self):
        # A sensitive-named key holding a dict is masked entirely (not recursed)
        out = DataSanitizer.redact_value("x.api_key", None, {"nested": "v"})
        assert out == DataSanitizer.REDACTION_TEXT

    def test_primitive_value_unchanged(self):
        assert (
            DataSanitizer.redact_value("llm.provider", None, "openai")
            == "openai"
        )

    def test_empty_dict_unchanged(self):
        assert DataSanitizer.redact_value("llm", None, {}) == {}

    def test_list_of_dicts_secret_redacted(self):
        # a JSON list-of-dicts leaf (e.g. mcp.servers) must not ship secrets
        value = [{"name": "prod", "api_key": "sk-secret"}]
        out = DataSanitizer.redact_value("mcp.servers", None, value)
        assert out[0]["api_key"] == DataSanitizer.REDACTION_TEXT
        assert out[0]["name"] == "prod"

    def test_plain_list_unchanged(self):
        # a list of scalars under a non-sensitive key is not mangled
        value = ["host1", "host2"]
        assert DataSanitizer.redact_value("search.hosts", None, value) == [
            "host1",
            "host2",
        ]

    def test_sensitive_key_with_list_redacted_wholesale(self):
        out = DataSanitizer.redact_value("x.secret", None, ["a", "b"])
        assert out == DataSanitizer.REDACTION_TEXT