test_intake_redaction.py

#!/usr/bin/env python3
"""Test intake redaction: secrets, env dumps, oversized payloads, empty body.

Covers v2.13.0 release blocker requirement #7:
- empty body
- oversized body
- secret-like strings
- invalid JSON
- duplicate submission
- e2e fixture
"""
import json
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from scripts.intake_redact import is_env_dump, redact_payload, redact_text, redaction_summary


# ── redact_text ──

class TestRedactText:
    def test_no_secrets(self):
        assert redact_text("pip install times out") == "pip install times out"

    def test_github_token(self):
        result = redact_text("found ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij12 in config")
        assert "[REDACTED:github_token]" in result
        assert "ghp_" not in result

    def test_github_token_with_key_prefix(self):
        result = redact_text("token: ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij12")
        # "token:" triggers the generic credential pattern first — either way, redacted
        assert "[REDACTED:" in result
        assert "ghp_" not in result

    def test_slack_token(self):
        result = redact_text("slack: xoxb-1234567890-1234567890123-abc")
        assert "[REDACTED:slack_token]" in result

    def test_generic_api_key(self):
        result = redact_text("sk-abcdefghijklmnopqrstuvwx")
        assert "[REDACTED:api_key]" in result

    def test_password_key_value(self):
        result = redact_text("password=SuperSecret123 other=keep")
        assert "[REDACTED:credential]" in result
        assert "other=keep" in result

    def test_token_in_header(self):
        result = redact_text('Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.signature')
        assert "[REDACTED:bearer_token]" in result

    def test_private_key_pem(self):
        pem = "-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEA\n-----END RSA PRIVATE KEY-----"
        result = redact_text(pem)
        assert "[REDACTED:private_key]" in result
        assert "PRIVATE KEY" not in result

    def test_aws_key(self):
        result = redact_text("AKIAIOSFODNN7EXAMPLE is my key")
        assert "[REDACTED:aws_key]" in result

    def test_credit_card(self):
        result = redact_text("card: 4111 1111 1111 1111")
        assert "[REDACTED:card_number]" in result

    def test_truncation(self):
        long_text = "a" * 5000
        result = redact_text(long_text, max_length=2000)
        assert len(result) == 2000

    def test_empty_input(self):
        assert redact_text("") == ""
        assert redact_text(None) == ""


# ── is_env_dump ──

class TestEnvDump:
    def test_normal_text(self):
        assert not is_env_dump("pip install fails behind proxy")

    def test_env_dump_detected(self):
        dump = "export AWS_SECRET_ACCESS_KEY=abc\nexport DATABASE_URL=postgres://..."
        assert is_env_dump(dump)

    def test_single_env_var_not_dump(self):
        assert not is_env_dump("MY_VAR=something")


# ── redact_payload ──

class TestRedactPayload:
    def test_normal_payload_unchanged(self):
        record = {"message": "search returns 0 results", "source": "curl"}
        safe = redact_payload(record)
        assert safe["message"] == "search returns 0 results"
        assert safe["source"] == "curl"

    def test_secret_in_message_redacted(self):
        record = {"message": "my token is ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij12"}
        safe = redact_payload(record)
        assert "ghp_" not in safe["message"]
        assert "[REDACTED:github_token]" in safe["message"]

    def test_context_dict_redacted(self):
        record = {"message": "test", "context": {"env": "DATABASE_URL=postgres://secret"}}
        safe = redact_payload(record)
        assert "[REDACTED:credential]" in safe["context"]["env"]

    def test_env_dump_flagged(self):
        record = {"message": "export AWS_SECRET_KEY=abc\nexport DATABASE_URL=pg://x"}
        safe = redact_payload(record)
        assert safe.get("_env_dump_detected") is True

    def test_non_dict_input(self):
        assert redact_payload(None) == {}
        assert redact_payload("string") == {}

    def test_max_length_enforced(self):
        record = {"message": "x" * 5000}
        safe = redact_payload(record)
        assert len(safe["message"]) == 2000


# ── redaction_summary ──

class TestRedactionSummary:
    def test_no_redactions(self):
        record = {"message": "clean text"}
        safe = redact_payload(record)
        summary = redaction_summary(record, safe)
        assert summary["total"] == 0

    def test_counts_redactions(self):
        record = {"message": "ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij12 and sk-abcdefghijklmnopqrstuvwx"}
        safe = redact_payload(record)
        summary = redaction_summary(record, safe)
        assert summary["total"] >= 2
        assert summary.get("message", 0) >= 2


# ── Edge case: empty body ──

class TestEmptyBody:
    def test_empty_dict(self):
        safe = redact_payload({})
        assert safe == {}

    def test_none_fields(self):
        record = {"message": None, "context": None}
        safe = redact_payload(record)
        # None fields should not crash
        assert "message" in safe


# ── Edge case: oversized body ──

class TestOversizedBody:
    def test_large_message_truncated(self):
        record = {"message": "A" * 10000}
        safe = redact_payload(record)
        assert len(safe["message"]) == 2000

    def test_large_context_truncated(self):
        record = {"message": "ok", "context": {"dump": "B" * 5000}}
        safe = redact_payload(record)
        assert len(safe["context"]["dump"]) == 500  # context fields capped at 500


# ── Edge case: invalid JSON (handled at caller level) ──

class TestInvalidJSON:
    def test_non_string_input(self):
        """Non-string input is coerced to string."""
        result = redact_text(12345)
        assert result == "12345"


# ── Edge case: duplicate-like content ──

class TestDuplicateLike:
    def test_repeated_secrets_each_redacted(self):
        record = {"message": "key1: ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij12 key2: ghp_XYABCDEFGHIJ1234567890abcdefghij12345678"}
        safe = redact_payload(record)
        assert safe["message"].count("[REDACTED:") >= 2


# ── E2E fixture ──

class TestE2EFixture:
    def test_full_intake_redaction_flow(self):
        """Simulate a realistic intake payload through the full redaction pipeline."""
        raw = {
            "type": "diagnostic",
            "source": "curl",
            "message": "pip install fails with timeout. My token is ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij12 and DATABASE_URL=postgres://user:pass@host/db",
            "context": {
                "tool": "fatal-guard",
                "version": "0.3.0",
                "platform": "linux",
                "env": "STAGING",
            },
            "consent": "private_only",
        }
        safe = redact_payload(raw)
        summary = redaction_summary(raw, safe)

        # Secrets removed
        assert "ghp_" not in safe["message"]
        assert "pass@host" not in safe["message"]
        assert "[REDACTED:" in safe["message"]

        # Non-secret fields preserved
        assert safe["type"] == "diagnostic"
        assert safe["source"] == "curl"
        assert safe["consent"] == "private_only"
        assert safe["context"]["tool"] == "fatal-guard"
        assert safe["context"]["version"] == "0.3.0"

        # Summary counts
        assert summary["total"] >= 2


if __name__ == "__main__":
    pytest.main([__file__, "-v"])