Files
wehub-resource-sync 7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:08:55 +08:00

1103 lines
43 KiB
Python

"""
Extended security-focused tests for the data_sanitizer module.
Goes beyond the base test suite to cover advanced attack vectors,
deeply nested credential hiding, realistic metadata payloads,
and edge cases that could lead to accidental credential leakage.
"""
import json
import pytest
from local_deep_research.security.data_sanitizer import (
DataSanitizer,
filter_research_metadata,
redact_data,
sanitize_data,
strip_settings_snapshot,
)
# ---------------------------------------------------------------------------
# sanitize -- every DEFAULT_SENSITIVE_KEY in every case permutation
# ---------------------------------------------------------------------------
class TestSanitizeAllDefaultKeysAllCases:
"""Ensure each default sensitive key is removed regardless of casing."""
@pytest.mark.parametrize(
"key",
sorted(DataSanitizer.DEFAULT_SENSITIVE_KEYS),
)
def test_lowercase_key_removed(self, key):
"""Lowercase form of each default key is removed."""
data = {"safe": "keep", key: "credential-value"}
result = DataSanitizer.sanitize(data)
assert key not in result
assert result["safe"] == "keep"
@pytest.mark.parametrize(
"key",
sorted(DataSanitizer.DEFAULT_SENSITIVE_KEYS),
)
def test_uppercase_key_removed(self, key):
"""UPPERCASE form of each default key is removed."""
upper = key.upper()
data = {"safe": "keep", upper: "credential-value"}
result = DataSanitizer.sanitize(data)
assert upper not in result
@pytest.mark.parametrize(
"key",
sorted(DataSanitizer.DEFAULT_SENSITIVE_KEYS),
)
def test_titlecase_key_removed(self, key):
"""Title_Case form of each default key is removed."""
title = key.title()
data = {"safe": "keep", title: "credential-value"}
result = DataSanitizer.sanitize(data)
assert title not in result
@pytest.mark.parametrize(
"key",
sorted(DataSanitizer.DEFAULT_SENSITIVE_KEYS),
)
def test_alternating_case_key_removed(self, key):
"""aLtErNaTiNg case form is removed (case-insensitive guarantee)."""
alt = "".join(
c.upper() if i % 2 else c.lower() for i, c in enumerate(key)
)
data = {"safe": "keep", alt: "credential-value"}
result = DataSanitizer.sanitize(data)
assert alt not in result
# ---------------------------------------------------------------------------
# sanitize -- deeply nested and complex structures
# ---------------------------------------------------------------------------
class TestSanitizeDeeplyNested:
"""Tests for deeply nested and complex structure sanitization."""
def test_five_levels_deep(self):
"""Sensitive key removed five levels deep."""
data = {
"a": {
"b": {"c": {"d": {"e": {"password": "deep-secret", "ok": 1}}}}
}
}
result = DataSanitizer.sanitize(data)
assert result["a"]["b"]["c"]["d"]["e"] == {"ok": 1}
def test_mixed_list_and_dict_nesting(self):
"""Sensitive keys removed in alternating list/dict nesting."""
data = {
"items": [
{
"name": "first",
"children": [
{"api_key": "sk-1", "value": 10},
{"api_key": "sk-2", "value": 20},
],
}
]
}
result = DataSanitizer.sanitize(data)
children = result["items"][0]["children"]
assert len(children) == 2
assert all("api_key" not in child for child in children)
assert children[0]["value"] == 10
assert children[1]["value"] == 20
def test_list_of_lists_of_dicts(self):
"""Handles list of lists containing dicts with secrets."""
data = [[{"secret": "s1", "x": 1}], [{"secret": "s2", "x": 2}]]
result = DataSanitizer.sanitize(data)
assert "secret" not in result[0][0]
assert "secret" not in result[1][0]
assert result[0][0]["x"] == 1
assert result[1][0]["x"] == 2
def test_dict_with_list_of_primitives(self):
"""Lists of primitives are passed through unchanged."""
data = {"tags": ["alpha", "beta", "gamma"], "api_key": "remove-me"}
result = DataSanitizer.sanitize(data)
assert result["tags"] == ["alpha", "beta", "gamma"]
assert "api_key" not in result
def test_empty_nested_structures(self):
"""Empty nested dicts and lists are preserved."""
data = {"config": {}, "items": [], "secret": "gone"}
result = DataSanitizer.sanitize(data)
assert result == {"config": {}, "items": []}
# ---------------------------------------------------------------------------
# sanitize -- real-world credential formats
# ---------------------------------------------------------------------------
class TestSanitizeRealWorldCredentials:
"""Ensure realistic credential values are fully removed."""
def test_openai_api_key_format(self):
"""OpenAI-style API key removed completely."""
data = {"model": "gpt-4", "api_key": "sk-proj-abc123def456ghi789"}
result = DataSanitizer.sanitize(data)
assert "api_key" not in result
assert "sk-proj-abc123def456ghi789" not in str(result)
def test_jwt_access_token_removed(self):
"""JWT-formatted access_token removed completely."""
jwt = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.sig"
data = {"user": "alice", "access_token": jwt}
result = DataSanitizer.sanitize(data)
assert "access_token" not in result
assert jwt not in str(result)
def test_pem_private_key_removed(self):
"""PEM-formatted private key value removed."""
pem = "-----BEGIN RSA PRIVATE KEY-----\nMIIE...\n-----END RSA PRIVATE KEY-----"
data = {"key_id": "k-1", "private_key": pem}
result = DataSanitizer.sanitize(data)
assert "private_key" not in result
assert "BEGIN RSA PRIVATE KEY" not in str(result)
def test_multiline_secret_removed(self):
"""Multi-line secret value fully removed."""
data = {"secret": "line1\nline2\nline3", "name": "svc"}
result = DataSanitizer.sanitize(data)
assert "secret" not in result
assert "line1" not in str(result)
def test_very_long_token_removed(self):
"""Very long token string fully removed."""
long_token = "a" * 10000
data = {"refresh_token": long_token, "ok": True}
result = DataSanitizer.sanitize(data)
assert "refresh_token" not in result
assert long_token not in str(result)
# ---------------------------------------------------------------------------
# sanitize -- multiple sensitive keys in one dict
# ---------------------------------------------------------------------------
class TestSanitizeMultipleKeys:
"""Multiple sensitive keys in the same dict are all removed."""
def test_all_ten_default_keys_removed_at_once(self):
"""All 10 default sensitive keys removed from a single dict."""
data = {
k: f"value-{i}"
for i, k in enumerate(DataSanitizer.DEFAULT_SENSITIVE_KEYS)
}
data["safe_field"] = "preserved"
result = DataSanitizer.sanitize(data)
assert result == {"safe_field": "preserved"}
def test_sensitive_keys_with_various_value_types(self):
"""Sensitive keys are removed regardless of value type."""
data = {
"api_key": 12345,
"password": None,
"secret": ["nested", "list"],
"access_token": {"nested": "dict"},
"refresh_token": True,
"name": "keep",
}
result = DataSanitizer.sanitize(data)
assert set(result.keys()) == {"name"}
# ---------------------------------------------------------------------------
# sanitize -- custom sensitive_keys
# ---------------------------------------------------------------------------
class TestSanitizeCustomKeysExtended:
"""Extended tests for custom sensitive_keys parameter."""
def test_custom_keys_case_insensitive(self):
"""Custom keys are also matched case-insensitively."""
data = {"MY_CUSTOM_SECRET": "value", "ok": True}
result = DataSanitizer.sanitize(
data, sensitive_keys={"my_custom_secret"}
)
assert "MY_CUSTOM_SECRET" not in result
assert result["ok"] is True
def test_custom_keys_do_not_remove_defaults(self):
"""When custom keys provided, default keys are NOT removed."""
data = {"api_key": "stays", "my_field": "goes"}
result = DataSanitizer.sanitize(data, sensitive_keys={"my_field"})
assert "api_key" in result
assert "my_field" not in result
def test_custom_keys_in_nested_structures(self):
"""Custom keys work recursively in nested structures."""
data = {"outer": {"inner": {"custom_field": "secret", "value": 42}}}
result = DataSanitizer.sanitize(data, sensitive_keys={"custom_field"})
assert "custom_field" not in result["outer"]["inner"]
assert result["outer"]["inner"]["value"] == 42
def test_single_custom_key(self):
"""Single-element custom key set works."""
data = {"x": "remove", "y": "keep"}
result = DataSanitizer.sanitize(data, sensitive_keys={"x"})
assert result == {"y": "keep"}
# ---------------------------------------------------------------------------
# sanitize -- immutability guarantees
# ---------------------------------------------------------------------------
class TestSanitizeImmutability:
"""Verify sanitize never mutates the original data."""
def test_original_dict_unchanged_after_sanitize(self):
"""Original top-level dict is not mutated."""
original = {"api_key": "secret", "name": "test", "password": "pwd"}
original_copy = dict(original)
DataSanitizer.sanitize(original)
assert original == original_copy
def test_nested_dict_unchanged_after_sanitize(self):
"""Nested dict inside original is not mutated."""
inner = {"api_key": "nested-secret", "value": 1}
original = {"config": inner}
DataSanitizer.sanitize(original)
assert "api_key" in inner
assert inner["api_key"] == "nested-secret"
def test_list_items_unchanged_after_sanitize(self):
"""List items in original are not mutated."""
item = {"password": "p1", "name": "a"}
original = [item]
DataSanitizer.sanitize(original)
assert "password" in item
assert item["password"] == "p1"
# ---------------------------------------------------------------------------
# redact -- every DEFAULT_SENSITIVE_KEY in every case permutation
# ---------------------------------------------------------------------------
class TestRedactAllDefaultKeysAllCases:
"""Ensure each default sensitive key is redacted regardless of casing."""
@pytest.mark.parametrize(
"key",
sorted(DataSanitizer.DEFAULT_SENSITIVE_KEYS),
)
def test_lowercase_key_redacted(self, key):
"""Lowercase form of each default key is redacted."""
data = {"safe": "keep", key: "credential-value"}
result = DataSanitizer.redact(data)
assert result[key] == "[REDACTED]"
assert result["safe"] == "keep"
@pytest.mark.parametrize(
"key",
sorted(DataSanitizer.DEFAULT_SENSITIVE_KEYS),
)
def test_uppercase_key_redacted(self, key):
"""UPPERCASE form of each default key is redacted."""
upper = key.upper()
data = {"safe": "keep", upper: "credential-value"}
result = DataSanitizer.redact(data)
assert result[upper] == "[REDACTED]"
@pytest.mark.parametrize(
"key",
sorted(DataSanitizer.DEFAULT_SENSITIVE_KEYS),
)
def test_titlecase_key_redacted(self, key):
"""Title_Case form of each default key is redacted."""
title = key.title()
data = {"safe": "keep", title: "credential-value"}
result = DataSanitizer.redact(data)
assert result[title] == "[REDACTED]"
# ---------------------------------------------------------------------------
# redact -- deeply nested structures
# ---------------------------------------------------------------------------
class TestRedactDeeplyNested:
"""Tests for deeply nested redaction."""
def test_five_levels_deep(self):
"""Sensitive key redacted five levels deep."""
data = {
"a": {
"b": {"c": {"d": {"e": {"password": "deep-secret", "ok": 1}}}}
}
}
result = DataSanitizer.redact(data)
assert result["a"]["b"]["c"]["d"]["e"]["password"] == "[REDACTED]"
assert result["a"]["b"]["c"]["d"]["e"]["ok"] == 1
def test_mixed_list_and_dict_nesting(self):
"""Sensitive keys redacted in alternating list/dict nesting."""
data = {
"items": [
{
"name": "first",
"children": [
{"api_key": "sk-1", "value": 10},
],
}
]
}
result = DataSanitizer.redact(data)
assert result["items"][0]["children"][0]["api_key"] == "[REDACTED]"
assert result["items"][0]["children"][0]["value"] == 10
def test_list_of_lists_of_dicts_redacted(self):
"""Handles list of lists containing dicts with secrets."""
data = [[{"secret": "s1", "x": 1}], [{"secret": "s2", "x": 2}]]
result = DataSanitizer.redact(data)
assert result[0][0]["secret"] == "[REDACTED]"
assert result[1][0]["secret"] == "[REDACTED]"
assert result[0][0]["x"] == 1
# ---------------------------------------------------------------------------
# redact -- custom redaction_text variations
# ---------------------------------------------------------------------------
class TestRedactCustomTextVariations:
"""Extended tests for custom redaction_text."""
@pytest.mark.parametrize(
"text",
[
"",
"***",
"<REMOVED>",
"N/A",
"0" * 100,
"null",
"undefined",
],
)
def test_various_redaction_texts(self, text):
"""Various custom redaction text strings are applied correctly."""
data = {"password": "secret123"}
result = DataSanitizer.redact(data, redaction_text=text)
assert result["password"] == text
def test_redaction_text_with_special_characters(self):
"""Redaction text with special chars works."""
data = {"api_key": "secret"}
result = DataSanitizer.redact(data, redaction_text="<b>HIDDEN</b>")
assert result["api_key"] == "<b>HIDDEN</b>"
def test_redaction_text_with_newlines(self):
"""Redaction text with newlines works."""
data = {"secret": "val"}
result = DataSanitizer.redact(data, redaction_text="line1\nline2")
assert result["secret"] == "line1\nline2"
# ---------------------------------------------------------------------------
# redact -- immutability guarantees
# ---------------------------------------------------------------------------
class TestRedactImmutability:
"""Verify redact never mutates the original data."""
def test_original_dict_unchanged_after_redact(self):
"""Original top-level dict is not mutated."""
original = {"password": "secret", "name": "test"}
DataSanitizer.redact(original)
assert original["password"] == "secret"
def test_nested_dict_unchanged_after_redact(self):
"""Nested dict inside original is not mutated."""
inner = {"api_key": "nested-secret", "value": 1}
original = {"config": inner}
DataSanitizer.redact(original)
assert inner["api_key"] == "nested-secret"
def test_list_items_unchanged_after_redact(self):
"""List items in original are not mutated."""
item = {"password": "p1"}
original = [item]
DataSanitizer.redact(original)
assert item["password"] == "p1"
# ---------------------------------------------------------------------------
# redact -- sensitive values with various types are still redacted
# ---------------------------------------------------------------------------
class TestRedactValueTypes:
"""Sensitive keys redacted regardless of value type."""
def test_integer_value_redacted(self):
"""Integer value for sensitive key is replaced with redaction text."""
data = {"api_key": 12345}
result = DataSanitizer.redact(data)
assert result["api_key"] == "[REDACTED]"
def test_none_value_redacted(self):
"""None value for sensitive key is replaced with redaction text."""
data = {"password": None}
result = DataSanitizer.redact(data)
assert result["password"] == "[REDACTED]"
def test_boolean_value_redacted(self):
"""Boolean value for sensitive key is replaced with redaction text."""
data = {"secret": True}
result = DataSanitizer.redact(data)
assert result["secret"] == "[REDACTED]"
def test_list_value_redacted(self):
"""List value for sensitive key is replaced with redaction text."""
data = {"api_key": ["k1", "k2"]}
result = DataSanitizer.redact(data)
assert result["api_key"] == "[REDACTED]"
def test_dict_value_redacted(self):
"""Dict value for sensitive key is replaced with redaction text."""
data = {"password": {"hash": "abc", "salt": "xyz"}}
result = DataSanitizer.redact(data)
assert result["password"] == "[REDACTED]"
# ---------------------------------------------------------------------------
# sanitize_data / redact_data convenience functions -- extended
# ---------------------------------------------------------------------------
class TestConvenienceFunctionsExtended:
"""Extended tests for sanitize_data and redact_data convenience functions."""
def test_sanitize_data_delegates_correctly(self):
"""sanitize_data produces identical results to DataSanitizer.sanitize."""
data = {
"api_key": "sk-123",
"config": {"password": "pwd", "theme": "dark"},
"items": [{"secret": "s1", "x": 1}],
}
assert sanitize_data(data) == DataSanitizer.sanitize(data)
def test_redact_data_delegates_correctly(self):
"""redact_data produces identical results to DataSanitizer.redact."""
data = {
"api_key": "sk-123",
"config": {"password": "pwd", "theme": "dark"},
"items": [{"secret": "s1", "x": 1}],
}
assert redact_data(data) == DataSanitizer.redact(data)
def test_sanitize_data_with_nested_custom_keys(self):
"""sanitize_data with custom keys works on nested structures."""
data = {"outer": {"my_secret": "gone", "keep": "this"}}
result = sanitize_data(data, sensitive_keys={"my_secret"})
assert "my_secret" not in result["outer"]
assert result["outer"]["keep"] == "this"
def test_redact_data_with_custom_keys_and_text(self):
"""redact_data with both custom keys and custom text."""
data = {"my_field": "sensitive", "ok": "fine"}
result = redact_data(
data, sensitive_keys={"my_field"}, redaction_text="<SCRUBBED>"
)
assert result["my_field"] == "<SCRUBBED>"
assert result["ok"] == "fine"
def test_sanitize_data_primitives(self):
"""sanitize_data passes primitives through unchanged."""
assert sanitize_data("hello") == "hello"
assert sanitize_data(42) == 42
assert sanitize_data(None) is None
assert sanitize_data(True) is True
def test_redact_data_primitives(self):
"""redact_data passes primitives through unchanged."""
assert redact_data("hello") == "hello"
assert redact_data(42) == 42
assert redact_data(None) is None
assert redact_data(False) is False
def test_sanitize_data_empty_containers(self):
"""sanitize_data handles empty containers."""
assert sanitize_data({}) == {}
assert sanitize_data([]) == []
def test_redact_data_empty_containers(self):
"""redact_data handles empty containers."""
assert redact_data({}) == {}
assert redact_data([]) == []
# ---------------------------------------------------------------------------
# filter_research_metadata -- extended security-focused tests
# ---------------------------------------------------------------------------
class TestFilterResearchMetadataExtended:
"""Extended tests for filter_research_metadata with security focus."""
def test_settings_snapshot_with_api_keys_never_leaks(self):
"""settings_snapshot containing API keys is never in output."""
meta = {
"is_news_search": True,
"settings_snapshot": {
"llm_api_key": "sk-secret-key-123",
"search_api_key": "cx-another-key",
"password": "admin123",
"nested": {"private_key": "-----BEGIN-----"},
},
}
result = filter_research_metadata(meta)
result_str = json.dumps(result)
assert "settings_snapshot" not in result
assert "sk-secret-key-123" not in result_str
assert "cx-another-key" not in result_str
assert "admin123" not in result_str
assert "BEGIN" not in result_str
def test_api_key_at_top_level_never_leaks(self):
"""api_key at top level of metadata is not passed through."""
meta = {"is_news_search": False, "api_key": "sk-top-level-key"}
result = filter_research_metadata(meta)
assert "api_key" not in result
assert "sk-top-level-key" not in json.dumps(result)
def test_password_at_top_level_never_leaks(self):
"""password at top level of metadata is not passed through."""
meta = {"is_news_search": True, "password": "supersecret"}
result = filter_research_metadata(meta)
assert "password" not in result
assert "supersecret" not in json.dumps(result)
def test_only_is_news_search_in_output(self):
"""Only is_news_search key appears in output regardless of input."""
meta = {
"is_news_search": True,
"phase": "complete",
"duration": 42.5,
"error_type": None,
"mode": "deep",
"processed_query": "test query",
}
result = filter_research_metadata(meta)
assert set(result.keys()) == {"is_news_search"}
assert result["is_news_search"] is True
def test_dict_without_is_news_search_returns_false(self):
"""Dict missing is_news_search key returns False default."""
meta = {"phase": "complete", "duration": 10.0}
result = filter_research_metadata(meta)
assert result == {"is_news_search": False}
def test_is_news_search_truthy_values_coerced_to_bool(self):
"""Truthy non-bool values for is_news_search are coerced to bool."""
assert filter_research_metadata({"is_news_search": 1}) == {
"is_news_search": True
}
assert filter_research_metadata({"is_news_search": "yes"}) == {
"is_news_search": True
}
def test_is_news_search_falsy_values_coerced_to_bool(self):
"""Falsy non-bool values for is_news_search are coerced to bool."""
assert filter_research_metadata({"is_news_search": 0}) == {
"is_news_search": False
}
assert filter_research_metadata({"is_news_search": ""}) == {
"is_news_search": False
}
def test_json_string_with_settings_snapshot_never_leaks(self):
"""JSON string containing settings_snapshot does not leak it."""
meta_str = json.dumps(
{
"is_news_search": True,
"settings_snapshot": {"api_key": "sk-from-json-string"},
}
)
result = filter_research_metadata(meta_str)
assert "settings_snapshot" not in result
assert "sk-from-json-string" not in json.dumps(result)
def test_json_string_with_api_key_never_leaks(self):
"""JSON string containing api_key at top level does not leak it."""
meta_str = json.dumps(
{"is_news_search": False, "api_key": "sk-json-key"}
)
result = filter_research_metadata(meta_str)
assert "api_key" not in result
assert "sk-json-key" not in json.dumps(result)
def test_none_input(self):
"""None input returns safe default."""
result = filter_research_metadata(None)
assert result == {"is_news_search": False}
def test_empty_string_input(self):
"""Empty string input returns safe default."""
result = filter_research_metadata("")
assert result == {"is_news_search": False}
def test_non_dict_int_input(self):
"""Integer input returns safe default."""
result = filter_research_metadata(42)
assert result == {"is_news_search": False}
def test_non_dict_list_input(self):
"""List input returns safe default."""
result = filter_research_metadata([1, 2, 3])
assert result == {"is_news_search": False}
def test_non_dict_bool_input(self):
"""Boolean input returns safe default."""
result = filter_research_metadata(True)
assert result == {"is_news_search": False}
def test_invalid_json_string(self):
"""Invalid JSON string returns safe default."""
result = filter_research_metadata("{not valid json")
assert result == {"is_news_search": False}
def test_json_array_string(self):
"""JSON array string returns safe default."""
result = filter_research_metadata("[1, 2, 3]")
assert result == {"is_news_search": False}
def test_json_string_non_dict(self):
"""JSON string containing a number returns safe default."""
result = filter_research_metadata("42")
assert result == {"is_news_search": False}
def test_json_string_null(self):
"""JSON string 'null' returns safe default."""
result = filter_research_metadata("null")
assert result == {"is_news_search": False}
def test_empty_dict_input(self):
"""Empty dict returns safe default for is_news_search."""
result = filter_research_metadata({})
assert result == {"is_news_search": False}
def test_large_metadata_payload_only_extracts_is_news_search(self):
"""Large realistic metadata payload only returns is_news_search."""
meta = {
"is_news_search": True,
"phase": "complete",
"error_type": None,
"processed_query": "what is quantum computing?",
"mode": "deep",
"duration": 123.45,
"search_results_count": 50,
"sources_used": ["google", "bing", "arxiv"],
"settings_snapshot": {
"llm_model": "gpt-4",
"api_key": "sk-prod-key-12345",
"temperature": 0.7,
"max_tokens": 4096,
"search_config": {
"google_api_key": "AIza-google-key",
"bing_api_key": "bing-secret-key",
},
},
}
result = filter_research_metadata(meta)
assert result == {"is_news_search": True}
result_str = json.dumps(result)
assert "sk-prod-key-12345" not in result_str
assert "AIza-google-key" not in result_str
assert "bing-secret-key" not in result_str
assert "settings_snapshot" not in result_str
# ---------------------------------------------------------------------------
# strip_settings_snapshot -- extended tests
# ---------------------------------------------------------------------------
class TestStripSettingsSnapshotExtended:
"""Extended tests for strip_settings_snapshot."""
def test_removes_settings_snapshot_preserves_all_other_keys(self):
"""All non-settings_snapshot keys are preserved."""
meta = {
"phase": "complete",
"error_type": None,
"processed_query": "test",
"mode": "deep",
"duration": 42.5,
"is_news_search": True,
"settings_snapshot": {"api_key": "sk-secret"},
}
result = strip_settings_snapshot(meta)
assert "settings_snapshot" not in result
assert result["phase"] == "complete"
assert result["error_type"] is None
assert result["processed_query"] == "test"
assert result["mode"] == "deep"
assert result["duration"] == 42.5
assert result["is_news_search"] is True
def test_settings_snapshot_api_keys_never_in_output(self):
"""API keys within settings_snapshot never appear in output."""
meta = {
"phase": "done",
"settings_snapshot": {
"llm_api_key": "sk-very-secret-key",
"search_api_key": "another-secret",
"password": "admin",
"config": {"private_key": "pem-data"},
},
}
result = strip_settings_snapshot(meta)
result_str = json.dumps(result)
assert "sk-very-secret-key" not in result_str
assert "another-secret" not in result_str
assert "admin" not in result_str
assert "pem-data" not in result_str
def test_no_settings_snapshot_returns_full_copy(self):
"""Dict without settings_snapshot returns copy with all keys."""
meta = {"phase": "error", "error_type": "timeout", "duration": 30.0}
result = strip_settings_snapshot(meta)
assert result == meta
def test_does_not_mutate_original_dict(self):
"""Original dict is not mutated."""
original = {
"phase": "complete",
"settings_snapshot": {"api_key": "sk-key"},
}
strip_settings_snapshot(original)
assert "settings_snapshot" in original
assert original["settings_snapshot"]["api_key"] == "sk-key"
def test_only_exact_settings_snapshot_key_removed(self):
"""Only 'settings_snapshot' key removed, not similar names."""
meta = {
"settings_version": 2,
"settings_backup": {"x": 1},
"snapshot": {"y": 2},
"settings_snapshot": {"api_key": "secret"},
}
result = strip_settings_snapshot(meta)
assert "settings_version" in result
assert "settings_backup" in result
assert "snapshot" in result
assert "settings_snapshot" not in result
def test_none_input(self):
"""None input returns empty dict."""
assert strip_settings_snapshot(None) == {}
def test_empty_dict_input(self):
"""Empty dict input returns empty dict."""
assert strip_settings_snapshot({}) == {}
def test_empty_string_input(self):
"""Empty string input returns empty dict."""
assert strip_settings_snapshot("") == {}
def test_non_dict_int_input(self):
"""Integer input returns empty dict."""
assert strip_settings_snapshot(42) == {}
def test_non_dict_list_input(self):
"""List input returns empty dict."""
assert strip_settings_snapshot([1, 2, 3]) == {}
def test_non_dict_bool_input(self):
"""Boolean input returns empty dict."""
assert strip_settings_snapshot(True) == {}
def test_invalid_json_string(self):
"""Invalid JSON string returns empty dict."""
assert strip_settings_snapshot("not valid json{") == {}
def test_json_string_with_settings_snapshot(self):
"""JSON string with settings_snapshot strips it correctly."""
meta_str = json.dumps(
{
"phase": "complete",
"duration": 10.5,
"settings_snapshot": {"api_key": "sk-from-json"},
}
)
result = strip_settings_snapshot(meta_str)
assert result == {"phase": "complete", "duration": 10.5}
assert "sk-from-json" not in json.dumps(result)
def test_json_string_without_settings_snapshot(self):
"""JSON string without settings_snapshot returns all keys."""
meta_str = json.dumps({"phase": "error", "error_type": "network"})
result = strip_settings_snapshot(meta_str)
assert result == {"phase": "error", "error_type": "network"}
def test_json_array_string(self):
"""JSON array string returns empty dict."""
assert strip_settings_snapshot("[1, 2, 3]") == {}
def test_json_null_string(self):
"""JSON null string returns empty dict."""
assert strip_settings_snapshot("null") == {}
def test_settings_snapshot_with_empty_dict_value(self):
"""settings_snapshot with empty dict is still removed."""
meta = {"phase": "done", "settings_snapshot": {}}
result = strip_settings_snapshot(meta)
assert result == {"phase": "done"}
def test_settings_snapshot_with_none_value(self):
"""settings_snapshot with None value is still removed."""
meta = {"phase": "done", "settings_snapshot": None}
result = strip_settings_snapshot(meta)
assert result == {"phase": "done"}
# ---------------------------------------------------------------------------
# Security attack vectors -- ensuring no credential leakage paths
# ---------------------------------------------------------------------------
class TestSecurityAttackVectors:
"""Tests simulating attack vectors for credential leakage."""
def test_sanitize_dict_only_safe_keys_survive(self):
"""After sanitize, only non-sensitive keys remain."""
data = {
"username": "alice",
"email": "alice@example.com",
"api_key": "sk-123",
"password": "pass",
"secret": "sec",
"access_token": "at",
"refresh_token": "rt",
"private_key": "pk",
"auth_token": "aut",
"session_token": "st",
"csrf_token": "csrf",
"apikey": "ak",
}
result = DataSanitizer.sanitize(data)
assert set(result.keys()) == {"username", "email"}
def test_redact_dict_all_secrets_redacted(self):
"""After redact, all sensitive values are '[REDACTED]'."""
data = {
"username": "alice",
"api_key": "sk-123",
"password": "hunter2-secret",
"secret": "very-secret-value",
}
result = DataSanitizer.redact(data)
assert result["username"] == "alice"
assert result["api_key"] == "[REDACTED]"
assert result["password"] == "[REDACTED]"
assert result["secret"] == "[REDACTED]"
# Verify no original secret values in serialized output
result_str = json.dumps(result)
assert "sk-123" not in result_str
assert "hunter2-secret" not in result_str
assert "very-secret-value" not in result_str
def test_sanitize_prevents_secret_in_nested_list_of_configs(self):
"""Secrets in list of config objects are fully removed."""
configs = [
{
"provider": "openai",
"api_key": "sk-openai-key",
"model": "gpt-4",
},
{
"provider": "anthropic",
"api_key": "sk-ant-key",
"model": "claude",
},
{"provider": "google", "apikey": "AIza-key", "model": "gemini"},
]
result = DataSanitizer.sanitize(configs)
for item in result:
assert "api_key" not in item
assert "apikey" not in item
assert "provider" in item
assert "model" in item
result_str = json.dumps(result)
assert "sk-openai-key" not in result_str
assert "sk-ant-key" not in result_str
assert "AIza-key" not in result_str
def test_filter_metadata_cannot_extract_arbitrary_keys(self):
"""filter_research_metadata only returns is_news_search, nothing else."""
meta = {
"is_news_search": True,
"secret": "should-not-appear",
"api_key": "should-not-appear",
"password": "should-not-appear",
"some_other_field": "should-not-appear",
}
result = filter_research_metadata(meta)
assert set(result.keys()) == {"is_news_search"}
result_str = json.dumps(result)
assert "should-not-appear" not in result_str
def test_strip_snapshot_then_filter_metadata_double_protection(self):
"""Applying strip_settings_snapshot then filter_research_metadata
provides defense in depth against credential leakage."""
meta = {
"is_news_search": True,
"phase": "complete",
"settings_snapshot": {
"api_key": "sk-double-layered",
"password": "admin-pwd",
},
}
stripped = strip_settings_snapshot(meta)
filtered = filter_research_metadata(stripped)
combined_str = json.dumps(filtered)
assert "sk-double-layered" not in combined_str
assert "admin-pwd" not in combined_str
assert filtered == {"is_news_search": True}
def test_sanitize_then_redact_no_leakage(self):
"""Applying sanitize then redact results in no credential values."""
data = {
"api_key": "sk-key",
"password": "pwd",
"name": "test",
"config": {"secret": "s1", "url": "http://safe.com"},
}
sanitized = DataSanitizer.sanitize(data)
redacted = DataSanitizer.redact(sanitized)
result_str = json.dumps(redacted)
assert "sk-key" not in result_str
assert "pwd" not in result_str
assert "s1" not in result_str
def test_sensitive_key_with_complex_nested_value_fully_removed(self):
"""Sensitive key whose value is a complex nested structure is fully removed."""
data = {
"api_key": {
"primary": "sk-primary",
"secondary": "sk-secondary",
"nested": {"deep": "sk-deep"},
},
"name": "service",
}
result = DataSanitizer.sanitize(data)
assert "api_key" not in result
result_str = json.dumps(result)
assert "sk-primary" not in result_str
assert "sk-secondary" not in result_str
assert "sk-deep" not in result_str
def test_sensitive_key_with_complex_nested_value_fully_redacted(self):
"""Sensitive key whose value is a complex nested structure is fully redacted."""
data = {
"api_key": {
"primary": "sk-primary",
"secondary": "sk-secondary",
},
"name": "service",
}
result = DataSanitizer.redact(data)
assert result["api_key"] == "[REDACTED]"
result_str = json.dumps(result)
assert "sk-primary" not in result_str
assert "sk-secondary" not in result_str
# ---------------------------------------------------------------------------
# Edge cases -- unusual but valid inputs
# ---------------------------------------------------------------------------
class TestEdgeCasesExtended:
"""Extended edge case tests."""
def test_sanitize_dict_with_numeric_keys(self):
"""Dict with integer keys (not matching any sensitive key) passes through."""
# Note: JSON doesn't support int keys, but Python dicts do
data = {1: "a", 2: "b"}
# int.lower() would fail, but these keys should not match sensitive keys
# This tests that the code handles non-string keys gracefully
# The implementation uses k.lower() which requires string keys.
# With int keys this would raise AttributeError, which is expected
# behavior -- we test that string keys work properly instead.
data = {"1": "a", "2": "b", "api_key": "secret"}
result = DataSanitizer.sanitize(data)
assert result == {"1": "a", "2": "b"}
def test_sanitize_unicode_keys_preserved(self):
"""Non-ASCII unicode keys that are not sensitive are preserved."""
data = {"nombre": "test", "api_key": "secret", "clé": "value"}
result = DataSanitizer.sanitize(data)
assert "nombre" in result
assert "clé" in result
assert "api_key" not in result
def test_sanitize_empty_string_value(self):
"""Sensitive key with empty string value is still removed."""
data = {"api_key": "", "name": "test"}
result = DataSanitizer.sanitize(data)
assert "api_key" not in result
def test_redact_empty_string_value(self):
"""Sensitive key with empty string value is still redacted."""
data = {"api_key": "", "name": "test"}
result = DataSanitizer.redact(data)
assert result["api_key"] == "[REDACTED]"
def test_sanitize_preserves_zero_values(self):
"""Non-sensitive keys with falsy values (0, False, '') are preserved."""
data = {
"count": 0,
"enabled": False,
"description": "",
"items": [],
"config": {},
}
result = DataSanitizer.sanitize(data)
assert result["count"] == 0
assert result["enabled"] is False
assert result["description"] == ""
assert result["items"] == []
assert result["config"] == {}
def test_redact_preserves_zero_values(self):
"""Non-sensitive keys with falsy values are preserved by redact."""
data = {"count": 0, "enabled": False}
result = DataSanitizer.redact(data)
assert result["count"] == 0
assert result["enabled"] is False
def test_sanitize_single_key_dict_sensitive(self):
"""Dict with only a sensitive key results in empty dict."""
data = {"password": "only-key"}
result = DataSanitizer.sanitize(data)
assert result == {}
def test_sanitize_tuple_passthrough(self):
"""Tuples are returned unchanged (treated as primitives)."""
data = (1, 2, 3)
result = DataSanitizer.sanitize(data)
assert result == (1, 2, 3)
def test_redact_tuple_passthrough(self):
"""Tuples are returned unchanged by redact."""
data = (1, 2, 3)
result = DataSanitizer.redact(data)
assert result == (1, 2, 3)
def test_sanitize_set_passthrough(self):
"""Sets are returned unchanged (treated as primitives)."""
data = {1, 2, 3}
result = DataSanitizer.sanitize(data)
assert result == {1, 2, 3}
def test_sanitize_float_passthrough(self):
"""Float values pass through unchanged."""
assert DataSanitizer.sanitize(3.14159) == 3.14159
def test_redact_float_passthrough(self):
"""Float values pass through unchanged from redact."""
assert DataSanitizer.redact(3.14159) == 3.14159
def test_filter_metadata_with_false_value_types(self):
"""filter_research_metadata with 0 as input returns default."""
assert filter_research_metadata(0) == {"is_news_search": False}
def test_filter_metadata_with_empty_list(self):
"""filter_research_metadata with empty list returns default."""
assert filter_research_metadata([]) == {"is_news_search": False}
def test_strip_snapshot_with_zero_input(self):
"""strip_settings_snapshot with 0 input returns empty dict."""
assert strip_settings_snapshot(0) == {}
def test_strip_snapshot_with_false_input(self):
"""strip_settings_snapshot with False input returns empty dict."""
assert strip_settings_snapshot(False) == {}