tegwick
576caa2665
AUDIT-WP-0004-T04, closing the workplan. Decision (Bernd): default to redaction, allow rejection per sender. Losing an audit record over one field is worse than storing it masked, but a higher-assurance channel must be able to refuse rather than mask. secret_policy is set per sender identity in AUDIT_CORE_SENDERS and defaults to redact. Detection now covers the whole payload at any depth, including lists, rather than only the top level of data. Under redaction the value is masked and the key is preserved: dropping the key would hide that the sender transmitted the field at all, which is exactly what an operator needs in order to stop it. The stored record carries details.redaction with policy and affected paths, so a reader never has to infer whether what they see is what was sent. Idempotency is unaffected - the payload hash is taken over the original request body, so redaction is deterministic and a resubmission still reconciles as a duplicate. Both outcomes are counted durably by sender, source, action and field path, exposed at GET /v1/secret-findings. Per-path aggregation is the point: the actionable unit is "stop emitting data.auth.token on membership.added", not "there were 47 redactions". Counters survive restart because the fix they drive lives in another service. Contract doc updated to match. Tests 46 -> 50. WP-0004 is finished. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
"""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))
|