"""The redactor is the only thing standing between an exception payload and a password in the database. These tests try to defeat it the way real code does: secrets nested in dicts, buried in DSNs, wearing an unexpected key name. """ from __future__ import annotations import pytest from app.core.redact import ( REDACTED, is_sensitive_key, redact, redact_headers, redact_mapping, redact_text, ) # --------------------------------------------------------------------------- # Structural layer — sensitive keys # --------------------------------------------------------------------------- @pytest.mark.parametrize( "key", [ "password", "PASSWORD", "Passwd", "api_key", "apiKey", "X-API-Key", "authorization", "Authorization", "access_token", "refreshToken", "client_secret", "private_key", "gitea_admin_token", "stripe_secret_key", "db_password", "encryption_key", ], ) def test_sensitive_keys_are_recognised(key: str) -> None: assert is_sensitive_key(key) is True @pytest.mark.parametrize( "key", [ "token_count", "input_tokens", "output_tokens", "max_tokens", "total_tokens", "name", "email", "organization_id", "command", "status_code", ], ) def test_usage_counters_and_ordinary_keys_are_not_sensitive(key: str) -> None: """``input_tokens`` is billing data, not a credential. Over-eager matching here would quietly destroy the usage numbers the error log exists to help debug. """ assert is_sensitive_key(key) is False def test_sensitive_key_value_is_replaced_whatever_it_looks_like() -> None: # The value has no recognisable shape at all — only the key gives it away. payload = {"password": "correct horse battery staple", "username": "demo"} result = redact_mapping(payload) assert result["password"] == REDACTED assert result["username"] == "demo" def test_redaction_reaches_into_nested_structures() -> None: payload = { "request": { "headers": {"Authorization": "Bearer abcdefghijklmnop"}, "body": {"user": {"password": "hunter2", "email": "demo@example.com"}}, }, "providers": [{"api_key": "sk-abcdefghijklmnopqrstuvwxyz"}, {"name": "openai"}], } result = redact(payload) assert result["request"]["headers"]["Authorization"] == REDACTED assert result["request"]["body"]["user"]["password"] == REDACTED assert result["request"]["body"]["user"]["email"] == "demo@example.com" assert result["providers"][0]["api_key"] == REDACTED assert result["providers"][1]["name"] == "openai" # --------------------------------------------------------------------------- # Lexical layer — recognisable values under harmless keys # --------------------------------------------------------------------------- def test_bearer_token_in_a_free_text_message_is_scrubbed() -> None: text = "upstream rejected the credential Bearer eyJhbGciOiJIUzI1NiJ9.abcdefgh.ijklmnop" scrubbed = redact_text(text) assert "eyJhbGciOiJIUzI1NiJ9" not in scrubbed assert "Bearer" in scrubbed # the scheme stays, it is not the secret def test_authorization_header_form_is_scrubbed_once_not_twice() -> None: """``Authorization: Bearer `` is a *sensitive key*, so the whole value goes — and it must not leave a doubled ``[REDACTED] [REDACTED]`` behind.""" text = "request failed: Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.abcdefgh.ijklmnop" scrubbed = redact_text(text) assert "eyJhbGciOiJIUzI1NiJ9" not in scrubbed assert scrubbed.count(REDACTED) == 1 def test_jwt_anywhere_in_a_message_is_scrubbed() -> None: token = ( "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9" ".eyJzdWIiOiIxMjM0NTY3ODkwIn0" ".dBjftJeZ4CVPmB92K27uhbUJU1p1r_wW1gFWFOEjXk" ) scrubbed = redact_text(f"token decode failed for {token} at step 3") assert token not in scrubbed assert "at step 3" in scrubbed def test_url_credentials_are_stripped_but_host_is_kept() -> None: scrubbed = redact_text("could not connect to postgresql://agy:s3cr3t@db.internal:5432/agy") assert "s3cr3t" not in scrubbed assert "db.internal" in scrubbed # the useful part survives def test_provider_key_shapes_are_scrubbed() -> None: for secret in ( "sk-abcdefghijklmnopqrstuvwxyz012345", "ghp_abcdefghijklmnopqrstuvwxyz0123456789", "AKIAIOSFODNN7EXAMPLE", "xoxb-1234567890-abcdefghijkl", ): assert secret not in redact_text(f"provider call failed: {secret}") def test_pem_private_key_block_is_scrubbed() -> None: pem = ( "-----BEGIN RSA PRIVATE KEY-----\n" "MIIEowIBAAKCAQEA1234567890abcdefghijklmnop\n" "-----END RSA PRIVATE KEY-----" ) scrubbed = redact_text(f"failed to load key:\n{pem}") assert "MIIEowIBAAKCAQEA" not in scrubbed assert "REDACTED_PRIVATE_KEY" in scrubbed def test_inline_key_value_pair_in_a_larger_string() -> None: scrubbed = redact_text("webhook rejected: password=hunter2&retries=3") assert "hunter2" not in scrubbed assert "retries=3" in scrubbed # --------------------------------------------------------------------------- # Third net — configured secrets matched by value # --------------------------------------------------------------------------- def test_the_configured_jwt_secret_is_scrubbed_by_value() -> None: """The strongest guarantee: a real configured secret never reaches storage. The test suite sets a 32+ byte ``JWT_SECRET``; whatever it is, it must not survive redaction even when it appears with no key and no recognisable shape. """ from app.core.config import settings secret = settings.jwt_secret scrubbed = redact_text(f"signature check failed comparing against {secret} directly") assert secret not in scrubbed # --------------------------------------------------------------------------- # Robustness — the redactor must never break a handler # --------------------------------------------------------------------------- def test_headers_are_scrubbed_and_lowercased() -> None: result = redact_headers( {"Authorization": "Bearer abcdefghijkl", "User-Agent": "curl/8.0", "Cookie": "sid=xyz"} ) assert result["authorization"] == REDACTED assert result["cookie"] == REDACTED assert result["user-agent"] == "curl/8.0" def test_bytes_become_a_length_not_a_body() -> None: assert redact(b"\x00\x01\x02\x03") == "<4 bytes>" def test_scalars_pass_through_untouched() -> None: assert redact(42) == 42 assert redact(1.5) == 1.5 assert redact(True) is True assert redact(None) is None def test_empty_inputs_are_handled() -> None: assert redact_mapping(None) == {} assert redact_headers(None) == {} assert redact_text("") == "" def test_very_long_input_is_bounded() -> None: """A 5 MB blob must not be handed to the regex engine whole.""" scrubbed = redact_text("x" * 5_000_000) assert len(scrubbed) < 100_000 assert "truncated" in scrubbed def test_non_string_keys_do_not_raise() -> None: result = redact_mapping({1: "a", (2, 3): "b", None: "c"}) assert result["1"] == "a" assert result["None"] == "c"