mjalmoudy94
717e53a7ce
مخزن اختصاصی برنامهٔ وب؛ محتوای پوشهٔ web در ریشهٔ مخزن قرار گرفت. تاریخچهٔ پروژهٔ ربات (agy-bot) در این مخزن وجود ندارد و هیچ وابستگی runtime به آن نیست.
222 lines
7.2 KiB
Python
222 lines
7.2 KiB
Python
"""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 <jwt>`` 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"
|