#!/usr/bin/env python3
"""
redact.py - single owner of "what is allowed to hit the Recall store".
WHY THIS EXISTS
---------------
The predecessor hooked PostToolUse with matcher "*". Every tool result on that machine -
vault reads, ssh sessions, curl output carrying bearer tokens - crossed a
third-party write path before landing in a 203MB SQLite file, with 180,548
operations staged in a cloud sync outbox. Nothing ever redacted anything.
Recall captures less by design, but "less" is not "none", so the write path
gets one owner and that owner gets a selftest.
THE RULE THAT MATTERS
---------------------
A redactor that cannot fire is worse than no redactor, because it reports clean.
Every pattern below is PROVEN able to fire by --selftest, and selftest failure is
exit 2 - the caller must treat that as "redaction is broken", not "input clean".
Usage:
redact.py --selftest # prove every pattern can fire; exit 0/2
from redact import scrub # scrub(text) -> (clean_text, n_hits, [labels])
"""
from __future__ import annotations
import re
import sys
PLACEHOLDER = "[REDACTED:{}]"
# (label, compiled pattern, group index to replace; 0 = whole match)
_RULES: list[tuple[str, re.Pattern, int]] = [
# --- private key material: whole block, never a fragment ---
("private-key", re.compile(
r"-----BEGIN[ A-Z]*PRIVATE KEY-----.*?-----END[ A-Z]*PRIVATE KEY-----",
re.DOTALL), 0),
# --- provider tokens with fixed, unmistakable prefixes ---
("anthropic-key", re.compile(r"sk-ant-[A-Za-z0-9_\-]{20,}"), 0),
("openai-key", re.compile(r"sk-(?!ant-)[A-Za-z0-9]{32,}"), 0),
("github-token", re.compile(r"gh[pousr]_[A-Za-z0-9]{20,}"), 0),
("slack-token", re.compile(r"xox[abprs]-[A-Za-z0-9-]{10,}"), 0),
("aws-akid", re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"), 0),
("google-key", re.compile(r"\bAIza[0-9A-Za-z_\-]{35}\b"), 0),
("telegram-bot", re.compile(r"\b\d{8,12}:AA[A-Za-z0-9_\-]{30,}"), 0),
("cloudflare", re.compile(r"\b[A-Za-z0-9_\-]{40}\b(?=[^A-Za-z0-9_\-]*(?i:cloudflare|CF_API))"), 0),
# --- bearer / basic auth headers: keep the header, kill the value ---
("bearer", re.compile(r"(?i)\b(authorization\s*:\s*bearer\s+)([A-Za-z0-9._\-]{12,})"), 2),
("basic", re.compile(r"(?i)\b(authorization\s*:\s*basic\s+)([A-Za-z0-9+/=]{12,})"), 2),
("apikey-header", re.compile(
r"(?i)\b(x-api-key\s*:\s*)([A-Za-z0-9._\-]{12,})"), 2),
# --- credentials embedded in URLs ---
("url-cred", re.compile(r"(?i)\b([a-z][a-z0-9+.\-]*://[^\s:/@]+:)([^\s@/]{3,})(@)"), 2),
# --- assignment forms: password=..., token: "...", secret => ... ---
("assigned-secret", re.compile(
r"(?i)\b((?:passwd|password|passphrase|secret|api[_\-]?key|access[_\-]?token"
r"|auth[_\-]?token|client[_\-]?secret|private[_\-]?key)\b\s*[:=]{1,2}>?\s*)"
r"([\"']?)([^\s\"',;]{6,})(\2)"), 3),
# --- JWTs ---
("jwt", re.compile(r"\beyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}"), 0),
]
# Values that look like secrets but are placeholders/examples - never worth a hit.
_ALLOW = re.compile(
r"(?i)^(?:x{3,}|\*{3,}|\.{3,}|<[^>]*>|\{\{.*\}\}|\$\{?[A-Z_]+\}?|"
r"redacted|changeme|example|placeholder|your[_\-]?\w+|none|null|true|false|"
r"\[REDACTED:[a-z-]+\])$")
def scrub(text: str | None) -> tuple[str, int, list[str]]:
"""Return (clean_text, hit_count, labels). Never raises on odd input."""
if not text:
return ("" if text is None else text), 0, []
if not isinstance(text, str):
text = str(text)
hits: list[str] = []
for label, pat, grp in _RULES:
def _sub(m: re.Match, _l=label, _g=grp) -> str:
captured = m.group(_g)
if captured and _ALLOW.match(captured.strip()):
return m.group(0) # known-safe placeholder, leave alone
hits.append(_l)
if _g == 0:
return PLACEHOLDER.format(_l)
# keep everything around the secret, swap only the secret itself
start, end = m.span(_g)
s0 = m.start(0)
return m.group(0)[:start - s0] + PLACEHOLDER.format(_l) + m.group(0)[end - s0:]
text = pat.sub(_sub, text)
return text, len(hits), sorted(set(hits))
# --- selftest ---------------------------------------------------------------
# One case per rule. If a rule stops firing, this fails loudly instead of the
# store quietly filling with cleartext.
_CASES: list[tuple[str, str]] = [
("private-key", "-----BEGIN RSA PRIVATE KEY-----\nMIIabc\n-----END RSA PRIVATE KEY-----"),
("private-key", "-----BEGIN OPENSSH PRIVATE KEY-----\nb3Blb\n-----END OPENSSH PRIVATE KEY-----"),
("anthropic-key", "key sk-ant-api03-AAAAAAAAAAAAAAAAAAAAAAAA here"),
("openai-key", "key sk-AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA here"),
("github-token", "ghp_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"),
("slack-token", "xoxb-1234567890-abcdefghij"),
("aws-akid", "AKIAIOSFODNN7EXAMPLE"),
("google-key", "AIza" + "Sy" + "A"*33),
("telegram-bot", "123456789:AAFakeTokenValueForTestingOnly12345"),
("cloudflare", "token abcdefghij0123456789abcdefghij0123456789 CF_API"),
("bearer", "Authorization: Bearer abcdef1234567890"),
("basic", "Authorization: Basic YWRtaW46aHVudGVyMg=="),
("apikey-header", "X-API-Key: abcdef1234567890"),
("url-cred", "postgres://admin:
[email protected]/app"),
("assigned-secret", 'password = "hunter2swordfish"'),
("jwt", "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dBjftJeZ4CVPmB92K27uhbUJU1p1r_wW1gFWFOEjXk"),
]
def selftest(verbose: bool = True) -> int:
failures: list[str] = []
for label, sample in _CASES:
clean, n, labels = scrub(sample)
if label not in labels:
failures.append(f"RULE DID NOT FIRE: {label!r} on {sample[:60]!r}")
elif "REDACTED" not in clean:
failures.append(f"RULE FIRED BUT LEFT CLEARTEXT: {label!r}")
# every rule must have a case - a rule with no case is an untested rule
covered = {c[0] for c in _CASES}
for label, _, _ in _RULES:
if label not in covered:
failures.append(f"RULE HAS NO SELFTEST CASE: {label!r}")
# must not maul ordinary text
for benign in (
"Restarted the service at 11:42 and health checked port 8080 ok.",
"pytest: pass: 21 fail: 0 skipped: 3",
"B4 fence pass: ok, mutation pass: caught",
"gate passed; all checks pass=true",
):
clean, n, _ = scrub(benign)
if n or clean != benign:
failures.append(f"FALSE POSITIVE on benign text: {benign!r} -> {clean!r}")
# placeholders must not be counted as secrets
for ph in ('password = "REDACTED"', 'api_key = "${VAULT_KEY}"', 'token = "<your-token>"'):
_, n, _ = scrub(ph)
if n:
failures.append(f"FALSE POSITIVE on placeholder: {ph!r}")
# IDEMPOTENCE. Re-scanning already-clean text must report ZERO hits, not the
# same hits again. Otherwise every maintenance run reports the same rows as
# fresh findings and people learn to ignore the report.
for sample in (s for _, s in _CASES):
once, n1, _ = scrub(sample)
twice, n2, _ = scrub(once)
if n2:
failures.append(f"NOT IDEMPOTENT: re-scan reported {n2} hit(s) on "
f"already-redacted text: {once[:60]!r}")
if twice != once:
failures.append(f"NOT STABLE: second pass changed the text: {twice[:60]!r}")
if failures:
if verbose:
print("REDACTION SELFTEST FAILED", file=sys.stderr)
for f in failures:
print(" " + f, file=sys.stderr)
return 2
if verbose:
print(f"redaction selftest OK - {len(_RULES)} rules, all proven able to fire")
return 0
if __name__ == "__main__":
if "--selftest" in sys.argv:
sys.exit(selftest())
data = sys.stdin.read()
clean, n, labels = scrub(data)
sys.stdout.write(clean)
if n:
print(f"\n[{n} redactions: {', '.join(labels)}]", file=sys.stderr)