audit-core/audit_core/redaction.py

"""Secret-shaped field detection, redaction, and rejection.

AUDIT-WP-0004-T04. The default is to **redact and accept**: rejecting an
otherwise legitimate event loses the audit record entirely, which is a worse
outcome than storing it with one field masked. A sender that needs the
stricter posture can be switched to **reject** per identity, so a
higher-assurance channel can refuse rather than mask.

Both outcomes are counted per field path, because the goal is not to redact
efficiently — it is to stop producing such fields at the source. A counter
that only says "47 redactions" tells an operator nothing actionable; one that
says ``data.credentials.password`` on ``membership.added`` from ``user-engine``
names the thing to fix.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

SECRET_FRAGMENTS = ("password", "secret", "token", "credential", "private_key")

REDACTED = "[redacted]"

POLICY_REDACT = "redact"
POLICY_REJECT = "reject"
POLICIES = (POLICY_REDACT, POLICY_REJECT)


@dataclass(frozen=True)
class Finding:
    """One secret-shaped field, located by dotted path."""

    path: str
    in_persisted_data: bool


class SecretFieldRejection(ValueError):
    """Raised under the reject policy. Carries the findings for counting."""

    def __init__(self, findings: list[Finding]) -> None:
        self.findings = findings
        # The message stays the stable error code senders already key on.
        super().__init__("secret_shaped_field")


def _is_secret_key(key: Any) -> bool:
    lowered = str(key).lower()
    return any(fragment in lowered for fragment in SECRET_FRAGMENTS)


def scan(value: Any, prefix: str = "") -> list[Finding]:
    """Locate every secret-shaped key, by dotted path.

    Detection is by key name across the whole structure, including nested
    dicts and lists. Values are not inspected: a value-shape heuristic on an
    audit payload produces false positives on legitimate identifiers, and a
    false positive here silently mangles an audit record.
    """
    findings: list[Finding] = []
    _walk(value, prefix, findings)
    return findings


def _walk(value: Any, prefix: str, findings: list[Finding]) -> None:
    if isinstance(value, dict):
        for key, item in value.items():
            path = f"{prefix}.{key}" if prefix else str(key)
            if _is_secret_key(key):
                findings.append(Finding(path=path, in_persisted_data=_persisted(path)))
                continue  # do not descend into a field already flagged
            _walk(item, path, findings)
    elif isinstance(value, list):
        for index, item in enumerate(value):
            _walk(item, f"{prefix}[{index}]", findings)


def _persisted(path: str) -> bool:
    """Whether this path lands in the stored record.

    Only ``data`` is persisted into the event details. A secret-shaped key
    elsewhere in the envelope is still reported, but it is dropped by
    normalization rather than stored, so it needs no masking.
    """
    return path == "data" or path.startswith("data.") or path.startswith("data[")


def redact(value: Any) -> Any:
    """Return a copy with every secret-shaped field's value masked.

    Keys are preserved. Removing them would hide the fact that the sender
    transmitted the field at all, which is exactly what the operator needs to
    see in order to stop it.
    """
    if isinstance(value, dict):
        return {
            key: (REDACTED if _is_secret_key(key) else redact(item))
            for key, item in value.items()
        }
    if isinstance(value, list):
        return [redact(item) for item in value]
    return value


def apply_policy(payload: dict[str, Any], policy: str) -> tuple[Any, list[Finding]]:
    """Apply ``policy`` to ``payload['data']``.

    Returns the data to persist and every finding across the whole payload.
    Raises :class:`SecretFieldRejection` under the reject policy.
    """
    findings = scan(payload)
    data = payload.get("data")
    if not findings:
        return data, findings
    if policy == POLICY_REJECT:
        raise SecretFieldRejection(findings)
    return redact(data), findings


def finding_from_path(path: str) -> Finding:
    """Rebuild a Finding from its path.

    ``in_persisted_data`` is a pure function of the path, so a finding recorded
    on an event record can be reconstructed for counting without carrying the
    flag through the stored details.
    """
    return Finding(path=path, in_persisted_data=_persisted(path))