|
#!/usr/bin/env -S uv run --script |
|
# /// script |
|
# requires-python = ">=3.9" |
|
# dependencies = [] |
|
# /// |
|
"""Redact secrets from coding-agent transcripts. |
|
|
|
Claude Code, Codex, Gemini CLI and Kimi Code all keep local JSONL transcripts of every |
|
session. Those transcripts capture whatever scrolled through the terminal, including |
|
`env` dumps, `cat .env`, `gh auth token` output and pasted credentials. This rewrites |
|
them in place, replacing each secret with a typed, hashed placeholder. |
|
|
|
uv run redact_agent_transcripts.py # dry run, report only |
|
uv run redact_agent_transcripts.py --show-context # see redacted samples |
|
uv run redact_agent_transcripts.py --apply # rewrite, with backups |
|
|
|
It does not model any agent's schema. It parses each line and walks every string value, |
|
so a new agent is a new glob rather than a new parser. |
|
""" |
|
from __future__ import annotations |
|
|
|
import argparse |
|
import hashlib |
|
import json |
|
import math |
|
import os |
|
import re |
|
import shutil |
|
import sys |
|
import tempfile |
|
import time |
|
from collections import Counter |
|
from dataclasses import dataclass, field |
|
from pathlib import Path |
|
from typing import Dict, Iterable, Iterator, List, Optional, Sequence, Set, Tuple |
|
|
|
__version__ = "1.0.0" |
|
|
|
# --------------------------------------------------------------------------- detectors |
|
|
|
@dataclass(frozen=True) |
|
class Detector: |
|
kind: str |
|
pattern: str |
|
group: int = 0 |
|
generic: bool = False # generic detectors are entropy-gated |
|
flags: int = 0 |
|
min_entropy: float = 0.0 # 0 means "use MIN_ENTROPY" |
|
require_mixed: bool = True # demand letters+digits (see looks_like_code_or_path) |
|
exclude: str = "" # if this matches the whole hit, it is not a credential |
|
|
|
|
|
DETECTORS: Sequence[Detector] = ( |
|
# Structured, prefixed credentials. High precision by shape, so no entropy gate: |
|
# gating these produces false negatives on low-entropy but entirely real keys. |
|
Detector("private-key", |
|
r"-----BEGIN [A-Z ]*PRIVATE KEY(?: BLOCK)?-----" |
|
r".*?-----END [A-Z ]*PRIVATE KEY(?: BLOCK)?-----", |
|
flags=re.S), |
|
# A key that was only partially printed (`head key.pem`) has no END line. The full |
|
# block above runs first and claims its span, so this only fires when there is none. |
|
Detector("private-key", |
|
r"-----BEGIN [A-Z ]*PRIVATE KEY(?: BLOCK)?-----[A-Za-z0-9+/=\s]{40,}", |
|
flags=re.S), |
|
Detector("anthropic-api-key", r"sk-ant-[A-Za-z0-9_-]{20,}"), |
|
Detector("openai-api-key", r"sk-proj-[A-Za-z0-9_-]{20,}"), |
|
Detector("openrouter-key", r"sk-or-v1-[A-Za-z0-9]{32,}"), |
|
Detector("github-token", r"github_pat_[A-Za-z0-9_]{50,}"), |
|
Detector("github-token", r"gh[pousr]_[A-Za-z0-9]{36,}"), |
|
Detector("gitlab-token", r"glpat-[A-Za-z0-9_-]{20,}"), |
|
Detector("aws-access-key-id", r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"), |
|
Detector("google-api-key", r"AIza[A-Za-z0-9_-]{30,}"), |
|
Detector("google-oauth-token", r"ya29\.[A-Za-z0-9_-]{30,}"), |
|
Detector("slack-token", r"xox[baprse]-[A-Za-z0-9-]{10,}"), |
|
Detector("atlassian-token", r"ATATT[A-Za-z0-9=_-]{20,}"), |
|
Detector("huggingface-token", r"hf_[A-Za-z0-9]{30,}"), |
|
Detector("npm-token", r"npm_[A-Za-z0-9]{30,}"), |
|
Detector("stripe-secret-key", r"(?:sk|rk)_live_[A-Za-z0-9]{20,}"), |
|
Detector("jwt", r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}"), |
|
Detector("openai-api-key", r"\bsk-[A-Za-z0-9]{48}\b"), |
|
# Unstructured. These carry real false-positive risk, so they are entropy-gated. |
|
Detector("bearer-token", |
|
r"[Aa]uthorization[\"']?\s*[:=]\s*[\"']?\s*(?:Bearer|Token|Basic)" |
|
r"[ \t]+([^\s\"',;}\])(&]{20,})", |
|
group=1, generic=True, require_mixed=False), |
|
# The keyword may be followed by more identifier before the separator, which is how |
|
# AWS_SECRET_ACCESS_KEY, DJANGO_SECRET_KEY and SECRET_KEY_BASE are spelled. Without |
|
# the trailing [a-z0-9_.-]* the most common shape in any `env` dump never matched. |
|
# No leading [a-z0-9_.-]* here. Matching the keyword anywhere in the identifier is |
|
# equivalent (AWS_SECRET_ACCESS_KEY matches at SECRET) but an unanchored leading |
|
# character class makes the engine scan a greedy run at every position in the line, |
|
# which cost a 12x slowdown across the corpus for zero extra matches. |
|
Detector("generic-secret-assignment", |
|
# [ \t]* rather than \s*: the value has to be on the same line. A YAML |
|
# "secrets:" at end of line, or a grep hit, otherwise captured the whole of |
|
# the FOLLOWING line as the credential. |
|
r"(?:api[_-]?key|private[_-]?key|secret|token|password|passwd" |
|
r"|credential)[a-z0-9_.-]*[\"']?[ \t]*[:=][ \t]*[\"']?([^\s\"',;}\])(&]{16,})", |
|
group=1, generic=True, flags=re.I, |
|
# Things that are shaped like a credential assignment but are not one. |
|
# Pagination cursors were the largest false-positive class in a real scan. |
|
exclude=r"(?i)(?:next_?page_?token|page_?token|continuation_?token|next_?token" |
|
r"|token_?count|tokens?_?used|token_?limit|tokenizer" |
|
r"|sha1|sha256|sha512|_hash|hash_|_digest|digest_|checksum" |
|
r"|fingerprint|_manager|_client|_service|_provider)"), |
|
# {0,64} because redis://:password@host has an empty user. The entropy floor is |
|
# lower here: a human-chosen password of 8 to 11 characters cannot reach 3.5. |
|
Detector("url-password", |
|
r"://[^\s:/@\"']{0,64}:([^\s:/@\"']{8,})@", |
|
group=1, generic=True, min_entropy=2.5, require_mixed=False), |
|
) |
|
|
|
# One combined pass used to skip lines that cannot contain a secret. It is a strict |
|
# superset of the detectors (the private-key entry is reduced to its opening marker so |
|
# that it matches a raw JSONL line, where the newlines are still escaped), so a line |
|
# that fails the prefilter provably has no match. This is what makes 1 GB tractable. |
|
_PREFILTER = re.compile( |
|
"|".join( |
|
r"-----BEGIN [A-Z ]*PRIVATE KEY" if d.kind == "private-key" |
|
# Word-boundary anchors are stripped here. In a raw JSONL line "\n" is the two |
|
# characters \ and n, so "n" before "AKIA" is not a word boundary and the |
|
# anchored pattern rejected the line before it was ever decoded. |
|
else re.sub(r"\\b", "", d.pattern) |
|
for d in DETECTORS |
|
), |
|
re.I, |
|
) |
|
|
|
_COMPILED = tuple( |
|
(d, re.compile(d.pattern, d.flags), re.compile(d.exclude) if d.exclude else None) |
|
for d in DETECTORS |
|
) |
|
|
|
# Strings that mark a match as documentation rather than a live credential. |
|
# Every marker here must be long enough that it will not occur by chance inside a real |
|
# credential. A bare "xxx" was tried and rejected: it appears in roughly 1 in 570 random |
|
# 60-character alphanumeric strings, so it silently drops genuine keys. |
|
_PLACEHOLDER_MARKERS = ("...", "<", ">", "redacted", "your-", "your_", "yourkey", |
|
"example", "changeme", "placeholder", "dummy", "put-your", |
|
"notreal", "fake-", "sample-", "replace-me") |
|
# Five or more identical alphanumeric characters in a row is filler (XXXXXXXX, |
|
# aaaaaaaa), and is the safe way to catch what a short "xxx" marker was reaching for. |
|
# Restricted to alphanumerics on purpose: the "-----" in a PEM header is structural, |
|
# not filler, and an unrestricted run rule swallows every private key. |
|
_FILLER_RUN = re.compile(r"([A-Za-z0-9])\1{4,}") |
|
|
|
MIN_ENTROPY = 3.5 |
|
|
|
# Characters that never appear inside a credential but are everywhere in source code. |
|
# Structural characters only. An earlier, wider set included ! # $ % @ * ? ^ ~ and threw |
|
# away genuine passwords like S3cr3t!Pass#2024 and URL-encoded p%40ssW0rd. |
|
_CODE_CHARS = set("()[]{}<>`\\;|") |
|
|
|
|
|
def looks_like_code_or_path(value: str, require_mixed: bool = True) -> bool: |
|
"""Reject source code and filesystem paths that the generic detectors pick up. |
|
|
|
This exists because of what a real scan turned up. `api_key = get_api_key()` was |
|
matching, capturing the function call as the secret. Replacement is line-wide, so |
|
that also rewrote `def get_api_key():` further along the same line: source code |
|
quoted in a transcript was being corrupted to scrub a secret that was never there. |
|
Before this guard, 1068 of 1249 findings on a real corpus were code identifiers. |
|
""" |
|
if any(c in _CODE_CHARS for c in value): |
|
return True |
|
if value.startswith("/") or "://" in value or value.endswith((".py", ".ts", ".js", ".json")): |
|
return True |
|
# Practically every real credential mixes letters and digits. An all-alphabetic |
|
# snake_case run is an identifier, and an all-numeric one is an id or a timestamp. |
|
# Only applied where a bare identifier is plausible, which is an assignment. The |
|
# password slot of a URL and the value after "Bearer" are structural credential |
|
# positions: a digitless password there is a password, not a variable name. |
|
if require_mixed and (not any(c.isdigit() for c in value) |
|
or not any(c.isalpha() for c in value)): |
|
return True |
|
return False |
|
|
|
|
|
def shannon_entropy(s: str) -> float: |
|
"""Bits of entropy per character.""" |
|
if not s: |
|
return 0.0 |
|
counts = Counter(s) |
|
n = len(s) |
|
return -sum((c / n) * math.log2(c / n) for c in counts.values()) |
|
|
|
|
|
def looks_like_placeholder(value: str) -> bool: |
|
"""True for documentation stand-ins: sk-ant-XXXX, <YOUR_KEY>, aaaa..., API_KEY=....""" |
|
low = value.lower() |
|
if any(m in low for m in _PLACEHOLDER_MARKERS): |
|
return True |
|
if _FILLER_RUN.search(value): |
|
return True |
|
# A value that is mostly one repeated character is a filler, not a credential. |
|
return max(Counter(value).values()) > len(value) / 2 |
|
|
|
|
|
@dataclass(frozen=True) |
|
class Finding: |
|
kind: str |
|
value: str |
|
|
|
|
|
def find_secrets(text: str, allowlist: Optional[Set[str]] = None) -> List[Finding]: |
|
"""Return the secrets in a single decoded string, in order, deduplicated. |
|
|
|
Detectors are applied in declaration order and a match is discarded if it overlaps |
|
a span already claimed, so `sk-ant-...` is reported once as an Anthropic key rather |
|
than also as a generic assignment. |
|
""" |
|
if not text or not _PREFILTER.search(text): |
|
return [] |
|
claimed: List[Tuple[int, int]] = [] |
|
out: List[Finding] = [] |
|
seen: Set[Tuple[str, str]] = set() |
|
for detector, rx, exclude_rx in _COMPILED: |
|
for m in rx.finditer(text): |
|
if exclude_rx is not None: |
|
# Widen to the whole identifier. The pattern starts at the keyword, so |
|
# the disqualifying part of "nextPageToken" sits before the match. |
|
back = m.start() |
|
while back > 0 and (text[back - 1].isalnum() or text[back - 1] in "_.-"): |
|
back -= 1 |
|
if exclude_rx.search(text[back:m.end()]): |
|
continue |
|
start, end = m.span(detector.group) |
|
if any(start < c_end and c_start < end for c_start, c_end in claimed): |
|
continue |
|
value = m.group(detector.group) |
|
if not value or looks_like_placeholder(value): |
|
continue |
|
# A "(" straight after the value means it is a function call, not a |
|
# credential. Minified identifiers like nonEmptyString$6 otherwise clear |
|
# the length, digit and entropy checks all at once. |
|
if detector.generic and text[end:end + 1] == "(": |
|
continue |
|
floor = detector.min_entropy or MIN_ENTROPY |
|
if detector.generic and (shannon_entropy(value) < floor |
|
or looks_like_code_or_path(value, detector.require_mixed)): |
|
continue |
|
if allowlist and value in allowlist: |
|
continue |
|
claimed.append(m.span()) |
|
key = (detector.kind, value) |
|
if key not in seen: |
|
seen.add(key) |
|
out.append(Finding(detector.kind, value)) |
|
return out |
|
|
|
|
|
# ------------------------------------------------------------------------ placeholders |
|
|
|
def load_salt(no_salt: bool = False) -> bytes: |
|
"""Per-machine random salt, so a truncated hash of a low-entropy secret cannot be |
|
brute-forced back. --no-salt gives IDs that are stable across machines instead.""" |
|
if no_salt: |
|
return b"" |
|
path = Path.home() / ".config" / "redact-transcripts" / "salt" |
|
if path.exists(): |
|
return path.read_bytes() |
|
path.parent.mkdir(parents=True, exist_ok=True) |
|
salt = os.urandom(16) |
|
fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) |
|
with os.fdopen(fd, "wb") as fh: |
|
fh.write(salt) |
|
return salt |
|
|
|
|
|
def placeholder(secret: str, kind: str, salt: bytes = b"") -> str: |
|
"""`[REDACTED:github-token:a3f91c]`. The digest keeps the transcript readable and |
|
lets you tell two different secrets apart, without revealing either.""" |
|
digest = hashlib.sha256(salt + secret.encode("utf-8", "surrogateescape")).hexdigest()[:6] |
|
return "[REDACTED:%s:%s]" % (kind, digest) |
|
|
|
|
|
def context_snippet(text: str, secret: str, kind: str, salt: bytes = b"", width: int = 40) -> str: |
|
"""Surrounding characters with the secret already replaced. Safe to print.""" |
|
i = text.find(secret) |
|
if i < 0: |
|
return placeholder(secret, kind, salt) |
|
lead, tail = text[max(0, i - width):i], text[i + len(secret):i + len(secret) + width] |
|
snippet = "%s%s%s" % (lead, placeholder(secret, kind, salt), tail) |
|
# Redact the window too. Secrets cluster: an exported key sits next to the token on |
|
# the next line, so showing raw context around one match leaks its neighbour. |
|
snippet = redact_text(snippet, salt)[0] |
|
snippet = snippet.replace("\n", "\\n").replace("\r", "\\r") |
|
return ("..." if i > width else "") + snippet + "..." |
|
|
|
|
|
# ---------------------------------------------------------------------- string walking |
|
|
|
def walk_strings(obj) -> Iterator[str]: |
|
"""Yield every string value in a parsed JSON document. Dict keys are skipped. |
|
|
|
Schema-agnostic on purpose. Claude's `toolUseResult` is an object, a bare string or |
|
an array depending on which tool ran, and `tool_result.content` is a string or an |
|
array, so targeting known paths breaks constantly. |
|
""" |
|
stack = [obj] |
|
while stack: |
|
cur = stack.pop() |
|
if isinstance(cur, str): |
|
yield cur |
|
elif isinstance(cur, dict): |
|
# Values only. Redacting a key would rename a field and change the |
|
# transcript's schema, and no agent stores a credential as a JSON key. |
|
stack.extend(cur.values()) |
|
elif isinstance(cur, (list, tuple)): |
|
stack.extend(cur) |
|
|
|
|
|
def find_secrets_in_obj(obj, allowlist: Optional[Set[str]] = None) -> List[Finding]: |
|
out: List[Finding] = [] |
|
seen: Set[Tuple[str, str]] = set() |
|
for s in walk_strings(obj): |
|
for f in find_secrets(s, allowlist): |
|
key = (f.kind, f.value) |
|
if key not in seen: |
|
seen.add(key) |
|
out.append(f) |
|
return out |
|
|
|
|
|
# -------------------------------------------------------------------- line redaction |
|
|
|
def _encodings(secret: str) -> List[str]: |
|
"""Every on-disk spelling of one secret, longest first. |
|
|
|
A secret found in a decoded string may appear in the raw line as-is, JSON-escaped |
|
(a private key's real newlines become `\\n`), or double-escaped (Codex stores tool |
|
arguments as a JSON string inside JSON). |
|
""" |
|
once = json.dumps(secret, ensure_ascii=False)[1:-1] |
|
twice = json.dumps(once, ensure_ascii=False)[1:-1] |
|
# Some encoders additionally escape the forward slash. Python's does not, so this |
|
# spelling has to be built by hand or a secret containing "/" is silently missed. |
|
variants = {secret, once, twice} |
|
variants |= {v.replace("/", "\\/") for v in list(variants) if "/" in v} |
|
return sorted(variants, key=len, reverse=True) |
|
|
|
|
|
def redact_text(text: str, salt: bytes = b"", allowlist: Optional[Set[str]] = None |
|
) -> Tuple[str, List[Finding]]: |
|
"""Redact a plain (non-JSON) string.""" |
|
findings = find_secrets(text, allowlist) |
|
for f in sorted(findings, key=lambda f: -len(f.value)): |
|
text = text.replace(f.value, placeholder(f.value, f.kind, salt)) |
|
return text, findings |
|
|
|
|
|
def redact_line(raw: str, salt: bytes = b"", allowlist: Optional[Set[str]] = None |
|
) -> Tuple[str, List[Finding]]: |
|
"""Redact one JSONL line, returning the new line and what was found. |
|
|
|
Secrets are located in the *decoded* value but substituted into the *raw* line, and |
|
the line is never re-serialized. Python's json.dumps does not reproduce another |
|
encoder's output byte for byte (`1e-07` vs `1e-7`, float formatting, ensure_ascii), |
|
so re-serializing would rewrite a gigabyte of lines that contain no secrets at all. |
|
Substituting into the raw text means clean lines stay byte-identical, line counts |
|
are preserved by construction, and a changed line differs only at the secret. |
|
""" |
|
if not raw.strip() or not _PREFILTER.search(raw): |
|
return raw, [] |
|
try: |
|
obj = json.loads(raw) |
|
except ValueError: |
|
return redact_text(raw, salt, allowlist) |
|
|
|
findings = find_secrets_in_obj(obj, allowlist) |
|
if not findings: |
|
return raw, [] |
|
|
|
out = raw |
|
# Longest first. A masked display such as "sk-ant-api03-abcd..." sits next to the |
|
# full key in `auth status` output; replacing the short prefix first consumed the |
|
# key's head and left the remaining 70 characters sitting in the file. |
|
for f in sorted(findings, key=lambda f: -len(f.value)): |
|
token = placeholder(f.value, f.kind, salt) |
|
for enc in _encodings(f.value): |
|
out = out.replace(enc, token) |
|
|
|
# Verify against the DECODED result, not the raw text. Raw substitution can miss a |
|
# spelling this code does not know about, and reporting a finding while leaving the |
|
# credential on disk is the worst possible outcome: the user believes it is clean. |
|
try: |
|
if find_secrets_in_obj(json.loads(out), allowlist): |
|
raise ValueError("secret survived raw substitution") |
|
except ValueError: |
|
# ensure_ascii=True on purpose: a transcript can carry a lone surrogate (what |
|
# JSON.stringify emits for a truncated emoji). Writing that raw produces bytes |
|
# that will not decode as UTF-8, or raises and kills the whole run. |
|
out = json.dumps(_redact_obj(obj, salt, allowlist), ensure_ascii=True) |
|
return out, findings |
|
|
|
|
|
def _redact_obj(obj, salt: bytes, allowlist: Optional[Set[str]]): |
|
if isinstance(obj, str): |
|
return redact_text(obj, salt, allowlist)[0] |
|
if isinstance(obj, dict): |
|
return {k: _redact_obj(v, salt, allowlist) for k, v in obj.items()} |
|
if isinstance(obj, list): |
|
return [_redact_obj(v, salt, allowlist) for v in obj] |
|
return obj |
|
|
|
|
|
# ------------------------------------------------------------------------- file layer |
|
|
|
@dataclass |
|
class FileStats: |
|
path: Path |
|
findings: int = 0 |
|
by_kind: Counter = field(default_factory=Counter) |
|
samples: List[str] = field(default_factory=list) |
|
changed: bool = False |
|
error: str = "" |
|
|
|
|
|
def redact_file(path: Path, apply: bool = False, backup_dir: Optional[Path] = None, |
|
salt: bytes = b"", allowlist: Optional[Set[str]] = None, |
|
collect_context: bool = False, max_samples: int = 3) -> FileStats: |
|
"""Scan one file, and rewrite it in place when `apply` is set. |
|
|
|
The file is only touched if something was found, so a clean transcript keeps its |
|
mtime. Rewrites are atomic and restore the original mode and mtime, because |
|
`/resume` orders sessions by mtime and distill.py's manifest is mtime-keyed. |
|
""" |
|
stats = FileStats(path=path) |
|
# Write through to the real file. os.replace on a symlink swaps out the LINK, which |
|
# would leave the actual transcript unredacted while looking like it worked. |
|
path = Path(os.path.realpath(str(path))) |
|
is_jsonl = path.suffix == ".jsonl" |
|
try: |
|
with open(path, "r", encoding="utf-8", errors="surrogateescape", newline="") as fh: |
|
original = fh.read() |
|
except (OSError, UnicodeError) as exc: |
|
stats.error = str(exc) |
|
return stats |
|
|
|
pieces: List[str] = [] |
|
if is_jsonl: |
|
# Split on "\n" only. str.splitlines() also breaks on U+2028, U+2029 and |
|
# U+0085, which are legal characters inside a JSON string, so it would chop a |
|
# valid record into fragments that no longer parse. |
|
raw_lines = original.split("\n") |
|
for idx, body in enumerate(raw_lines): |
|
sep = "\n" if idx < len(raw_lines) - 1 else "" |
|
new, found = redact_line(body, salt, allowlist) |
|
if found: |
|
_record(stats, found, body, salt, collect_context, max_samples) |
|
pieces.append(new + sep) |
|
updated = "".join(pieces) |
|
else: |
|
updated, found = redact_line(original, salt, allowlist) |
|
if found: |
|
_record(stats, found, original, salt, collect_context, max_samples) |
|
|
|
if not stats.findings or updated == original: |
|
return stats |
|
if not apply: |
|
return stats |
|
|
|
st = path.stat() |
|
tmp = None |
|
try: |
|
if backup_dir is not None: |
|
dest = backup_dir / Path(str(path).lstrip("/")) |
|
dest.parent.mkdir(parents=True, exist_ok=True) |
|
shutil.copy2(str(path), str(dest)) |
|
|
|
if st.st_nlink > 1: |
|
# Rewrite through the existing inode. os.replace would hand this path a new |
|
# one, and every sibling hardlink would quietly keep the original secret. |
|
with open(path, "w", encoding="utf-8", errors="surrogateescape", newline="") as fh: |
|
fh.write(updated) |
|
fh.flush() |
|
os.fsync(fh.fileno()) |
|
else: |
|
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".redact-", suffix=".tmp") |
|
with os.fdopen(fd, "w", encoding="utf-8", errors="surrogateescape", newline="") as fh: |
|
fh.write(updated) |
|
fh.flush() |
|
os.fsync(fh.fileno()) |
|
os.chmod(tmp, st.st_mode & 0o777) |
|
os.replace(tmp, str(path)) |
|
tmp = None |
|
os.utime(str(path), (st.st_atime, st.st_mtime)) |
|
except (OSError, UnicodeError) as exc: |
|
# One unwritable file must not abort a run across a thousand of them. |
|
if tmp and os.path.exists(tmp): |
|
try: |
|
os.unlink(tmp) |
|
except OSError: |
|
pass |
|
stats.error = str(exc) |
|
return stats |
|
stats.changed = True |
|
return stats |
|
|
|
|
|
def _record(stats: FileStats, found: List[Finding], text: str, salt: bytes, |
|
collect_context: bool, max_samples: int) -> None: |
|
stats.findings += len(found) |
|
for f in found: |
|
stats.by_kind[f.kind] += 1 |
|
if collect_context and len(stats.samples) < max_samples: |
|
stats.samples.append(context_snippet(text, f.value, f.kind, salt)) |
|
|
|
|
|
# ------------------------------------------------------------------- discovery layer |
|
|
|
TEXT_SUFFIXES = {".jsonl", ".json", ".md", ".txt", ".log"} |
|
|
|
# The exclusion principle: redact the record of what happened, never what the agent |
|
# reads in order to operate. Rewriting these breaks live tooling. `.claude.json` holds |
|
# `mcpServers.*.env` keys the MCP servers actually need; shell-snapshots are how Claude |
|
# Code restores the Bash environment. |
|
EXCLUDED_NAMES = { |
|
"auth.json", "auth.v2.file", "auth.v2.key", ".env", ".claude.json", |
|
"settings.json", "settings.local.json", "config.toml", "config.json", |
|
"credentials.json", ".credentials.json", "mcp_config.json", |
|
} |
|
EXCLUDED_SUFFIXES = {".key", ".pem", ".sh", ".bak", ".tmp"} |
|
EXCLUDED_DIR_PARTS = { |
|
"credentials", "oauth", "shell-snapshots", "node_modules", ".git", |
|
"agent-sdk-venv", "site-packages", ".venv", |
|
} |
|
|
|
|
|
def is_excluded(path: Path) -> bool: |
|
if path.name in EXCLUDED_NAMES or path.suffix in EXCLUDED_SUFFIXES: |
|
return True |
|
if path.name.endswith(("-key.sh", "-token.sh", ".meta.json")) and path.suffix != ".jsonl": |
|
if path.name.endswith(".meta.json"): |
|
return False |
|
return True |
|
return any(part in EXCLUDED_DIR_PARTS for part in path.parts) |
|
|
|
|
|
def iter_files(roots: Iterable[Path], skip_active_seconds: float = 0) -> Iterator[Path]: |
|
"""Yield every scannable file under `roots`, deduplicated by realpath. |
|
|
|
Deduplication is not an optimization. `~/.claude-personal`, `-work`, `-kimi`, |
|
`-opencode` and `-zai` are all symlinks to the same `projects/` directory, so |
|
without it the same gigabyte gets read and rewritten five times. Recursion matters |
|
too: Claude keeps 322 subagent transcripts at `<slug>/<session>/subagents/`, which |
|
a `*/*.jsonl` glob silently misses. |
|
""" |
|
seen: Set[str] = set() |
|
now = time.time() |
|
for root in roots: |
|
root = Path(root).expanduser() |
|
if not root.exists(): |
|
continue |
|
candidates = [root] if root.is_file() else sorted(root.rglob("*")) |
|
for p in candidates: |
|
try: |
|
if not p.is_file() or p.suffix not in TEXT_SUFFIXES or is_excluded(p): |
|
continue |
|
real = os.path.realpath(str(p)) |
|
if real in seen: |
|
continue |
|
if skip_active_seconds and (now - p.stat().st_mtime) < skip_active_seconds: |
|
continue |
|
seen.add(real) |
|
except OSError: |
|
continue |
|
yield p |
|
|
|
|
|
AGENT_ROOTS: Dict[str, List[str]] = { |
|
"claude": ["~/.claude/projects", "~/.claude/history.jsonl", |
|
"~/.claude-personal/paste-cache", "~/.claude-work/paste-cache"], |
|
"codex": ["~/.codex/sessions", "~/.codex/history.jsonl"], |
|
"gemini": ["~/.gemini/tmp", "~/.gemini/antigravity-cli/history.jsonl"], |
|
"kimi": ["~/.kimi-code/sessions", "~/.kimi-code/session_index.jsonl"], |
|
} |
|
|
|
# Known additional copies of the same conversations. Not scanned by default: they are |
|
# SQLite (with live WAL files) or very large binary-ish stores. Reported so you know |
|
# redacting the JSONL did not get everything. |
|
SECOND_COPIES = [ |
|
("~/.codex/thread_history_1.sqlite", "Codex mirrors every rollout item in thread_items.item_json"), |
|
("~/.claude/file-history/", "Claude keeps pre-edit copies of files it modified"), |
|
("~/.local/share/opencode/opencode.db", "OpenCode stores whole sessions in SQLite"), |
|
("~/.playwriter/cdp.jsonl", "Playwriter logs every CDP message, including revealed tokens"), |
|
("~/personal/claude-session-graph/distilled/", "Distilled markdown copies of Claude sessions"), |
|
] |
|
|
|
|
|
# ------------------------------------------------------------------------------- CLI |
|
|
|
def human(n: float) -> str: |
|
for unit in ("B", "KB", "MB", "GB"): |
|
if n < 1024 or unit == "GB": |
|
return "%.1f %s" % (n, unit) |
|
n /= 1024 |
|
return "%.1f GB" % n |
|
|
|
|
|
def run(roots: List[Path], apply: bool, backup_dir: Optional[Path], salt: bytes, |
|
allowlist: Set[str], show_context: bool, skip_active: float) -> dict: |
|
totals: Counter = Counter() |
|
files_with, files_changed, scanned, byte_count = [], 0, 0, 0 |
|
errors: List[Tuple[Path, str]] = [] |
|
started = time.time() |
|
for path in iter_files(roots, skip_active_seconds=skip_active): |
|
scanned += 1 |
|
try: |
|
byte_count += path.stat().st_size |
|
except OSError: |
|
pass |
|
st = redact_file(path, apply=apply, backup_dir=backup_dir, salt=salt, |
|
allowlist=allowlist, collect_context=show_context) |
|
if st.error: |
|
errors.append((path, st.error)) |
|
if st.findings: |
|
totals.update(st.by_kind) |
|
files_with.append(st) |
|
files_changed += 1 if st.changed else 0 |
|
return { |
|
"scanned_files": scanned, |
|
"scanned_bytes": byte_count, |
|
"files_with_findings": len(files_with), |
|
"files_rewritten": files_changed, |
|
"total_findings": sum(totals.values()), |
|
"by_kind": dict(totals.most_common()), |
|
"elapsed_seconds": round(time.time() - started, 2), |
|
"errors": ["%s: %s" % (p, e) for p, e in errors], |
|
"_details": files_with, |
|
} |
|
|
|
|
|
def self_test() -> int: |
|
"""Build a synthetic corpus in every supported envelope, then prove round-trip.""" |
|
import random, string as _s |
|
rng = random.Random(20260902) |
|
|
|
def body(n, alphabet=_s.ascii_letters + _s.digits): |
|
return "".join(rng.choice(alphabet) for _ in range(n)) |
|
|
|
secrets = ["sk-ant-api03-" + body(95), "ghp_" + body(36), "sk-proj-" + body(48), |
|
"AKIA" + body(16, _s.ascii_uppercase + _s.digits), |
|
"AIza" + body(35), "ATATT" + body(60), "hf_" + body(34), |
|
"eyJ" + body(30) + "." + body(40) + "." + body(43)] |
|
pem = ("-----BEGIN RSA PRIVATE KEY-----\n" |
|
+ "\n".join(body(64) for _ in range(4)) + "\n-----END RSA PRIVATE KEY-----") |
|
secrets.append(pem) |
|
|
|
envelopes = [ |
|
lambda s: {"type": "user", "message": {"role": "user", "content": [ |
|
{"type": "tool_result", "content": "$ env\nKEY=%s\n" % s}]}, |
|
"toolUseResult": {"stdout": "KEY=%s\n" % s, "stderr": ""}}, |
|
lambda s: {"toolUseResult": [{"type": "text", "text": "value %s" % s}]}, |
|
lambda s: {"toolUseResult": s}, |
|
lambda s: {"type": "response_item", "payload": { |
|
"type": "function_call", "arguments": json.dumps({"cmd": "export K=%s" % s})}}, |
|
lambda s: {"$set": {"messages": [{"content": [{"text": "use %s" % s}]}]}}, |
|
lambda s: {"event": {"result": {"output": "token: %s" % s}}, "f": 1e-7}, |
|
] |
|
|
|
tmp = Path(tempfile.mkdtemp(prefix="redact-selftest-")) |
|
failures: List[str] = [] |
|
try: |
|
target = tmp / "corpus.jsonl" |
|
lines, clean = [], {"note": "no secret here", "float": 1e-7, "u": "caf\u00e9"} |
|
for s in secrets: |
|
for env in envelopes: |
|
lines.append(json.dumps(env(s))) |
|
lines.append(json.dumps(clean)) |
|
target.write_text("\n".join(lines) + "\n", encoding="utf-8") |
|
before = target.read_text(encoding="utf-8").splitlines() |
|
os.utime(str(target), (0, 0)) |
|
|
|
st = redact_file(target, apply=True, backup_dir=tmp / "bk", salt=b"selftest") |
|
after = target.read_text(encoding="utf-8").splitlines() |
|
text = "\n".join(after) |
|
|
|
if len(before) != len(after): |
|
failures.append("line count changed: %d -> %d" % (len(before), len(after))) |
|
for i, line in enumerate(after): |
|
try: |
|
json.loads(line) |
|
except ValueError as exc: |
|
failures.append("line %d no longer valid JSON: %s" % (i, exc)) |
|
break |
|
for s in secrets: |
|
if s in text: |
|
failures.append("MISSED a %s" % find_secrets(s)[0].kind if find_secrets(s) else "MISSED a secret") |
|
if "BEGIN RSA PRIVATE KEY" in text: |
|
failures.append("private key block survived") |
|
expected = len(secrets) * len(envelopes) |
|
if st.findings != expected: |
|
failures.append("expected %d findings, got %d" % (expected, st.findings)) |
|
# clean lines must be byte-identical |
|
for a, b in zip(before, after): |
|
if "REDACTED" not in b and a != b: |
|
failures.append("a clean line was rewritten") |
|
break |
|
# idempotence |
|
st2 = redact_file(target, apply=True, backup_dir=tmp / "bk2", salt=b"selftest") |
|
if st2.findings: |
|
failures.append("not idempotent: %d findings on second pass" % st2.findings) |
|
# the backup must still hold the originals |
|
bk = list((tmp / "bk").rglob("corpus.jsonl")) |
|
if not bk or secrets[0] not in bk[0].read_text(encoding="utf-8"): |
|
failures.append("backup did not preserve the original") |
|
finally: |
|
shutil.rmtree(str(tmp), ignore_errors=True) |
|
|
|
if failures: |
|
for f in failures: |
|
print("FAIL: %s" % f, file=sys.stderr) |
|
return 1 |
|
print("self-test passed: %d secret shapes x %d envelope shapes, round-trip clean" |
|
% (len(secrets), len(envelopes))) |
|
return 0 |
|
|
|
|
|
def main(argv: Optional[List[str]] = None) -> int: |
|
ap = argparse.ArgumentParser( |
|
prog="redact_agent_transcripts.py", |
|
description="Redact secrets from Claude Code, Codex, Gemini CLI and Kimi Code transcripts.", |
|
formatter_class=argparse.RawDescriptionHelpFormatter, |
|
epilog="Default is a dry run. Nothing is written without --apply.") |
|
ap.add_argument("--agent", default="all", |
|
help="comma-separated: claude,codex,gemini,kimi,all (default: all)") |
|
ap.add_argument("--paths", nargs="+", metavar="PATH", |
|
help="scan these files/directories instead of the known agent stores") |
|
ap.add_argument("--apply", action="store_true", help="rewrite files in place (backs up first)") |
|
ap.add_argument("--backup-dir", type=Path, default=None, |
|
help="default: ~/.redact-backups/<timestamp>/") |
|
ap.add_argument("--show-context", action="store_true", |
|
help="print surrounding text with the secret already replaced") |
|
ap.add_argument("--allowlist", type=Path, help="file of literal values to never redact") |
|
ap.add_argument("--include-active", action="store_true", |
|
help="also scan files modified in the last 60s (the live session)") |
|
ap.add_argument("--no-salt", action="store_true", |
|
help="unsalted digests, stable across machines") |
|
ap.add_argument("--json", action="store_true", help="machine-readable summary on stdout") |
|
ap.add_argument("--self-test", action="store_true", help="run built-in round-trip checks") |
|
ap.add_argument("--version", action="version", version=__version__) |
|
args = ap.parse_args(argv) |
|
|
|
if args.self_test: |
|
return self_test() |
|
|
|
if args.paths: |
|
roots = [Path(p).expanduser() for p in args.paths] |
|
agents = ["custom"] |
|
else: |
|
agents = sorted(AGENT_ROOTS) if args.agent == "all" else \ |
|
[a.strip() for a in args.agent.split(",") if a.strip()] |
|
unknown = [a for a in agents if a not in AGENT_ROOTS] |
|
if unknown: |
|
ap.error("unknown agent(s): %s" % ", ".join(unknown)) |
|
roots = [Path(r).expanduser() for a in agents for r in AGENT_ROOTS[a]] |
|
|
|
allowlist = set() |
|
if args.allowlist: |
|
allowlist = {l.strip() for l in args.allowlist.read_text().splitlines() |
|
if l.strip() and not l.startswith("#")} |
|
|
|
salt = load_salt(args.no_salt) |
|
backup_dir = args.backup_dir |
|
if args.apply and backup_dir is None: |
|
backup_dir = Path.home() / ".redact-backups" / time.strftime("%Y%m%d-%H%M%S") |
|
|
|
result = run(roots, apply=args.apply, backup_dir=backup_dir, salt=salt, |
|
allowlist=allowlist, show_context=args.show_context, |
|
skip_active=0 if args.include_active else 60) |
|
details: List[FileStats] = result.pop("_details") |
|
|
|
if args.json: |
|
print(json.dumps(result, indent=2)) |
|
else: |
|
print("agents: %s" % ", ".join(agents)) |
|
print("scanned: %d files, %s in %.1fs" |
|
% (result["scanned_files"], human(result["scanned_bytes"]), |
|
result["elapsed_seconds"])) |
|
print("findings: %d in %d files" |
|
% (result["total_findings"], result["files_with_findings"])) |
|
if result["by_kind"]: |
|
print() |
|
width = max(len(k) for k in result["by_kind"]) |
|
for kind, n in result["by_kind"].items(): |
|
print(" %-*s %d" % (width, kind, n)) |
|
if args.show_context and details: |
|
print("\ncontext (secrets already replaced):") |
|
for st in details[:20]: |
|
print("\n %s" % st.path) |
|
for s in st.samples: |
|
print(" %s" % s) |
|
if details and not args.apply: |
|
print("\nfiles with findings:") |
|
for st in sorted(details, key=lambda s: -s.findings)[:15]: |
|
print(" %4d %s" % (st.findings, st.path)) |
|
if len(details) > 15: |
|
print(" ... and %d more" % (len(details) - 15)) |
|
|
|
extant = [(p, why) for p, why in SECOND_COPIES if Path(p).expanduser().exists()] |
|
if extant: |
|
print("\nnot scanned, but holds a second copy of the same conversations:") |
|
for p, why in extant: |
|
print(" %-42s %s" % (p, why)) |
|
print(" scan these with --paths, or accept that the JSONL pass missed them.") |
|
|
|
if result["errors"]: |
|
print("\ncould not be processed (%d):" % len(result["errors"])) |
|
for e in result["errors"][:10]: |
|
print(" %s" % e) |
|
|
|
if args.apply: |
|
print("\nrewrote %d files." % result["files_rewritten"]) |
|
print("If an agent was running, it may hold a session file open and rewrite") |
|
print("it from memory. Close your agents and re-run to be sure.") |
|
if backup_dir: |
|
print("BACKUPS AT %s STILL CONTAIN THE ORIGINAL SECRETS." % backup_dir) |
|
print("Delete them once you are satisfied: rm -rf %s" % backup_dir) |
|
elif result["total_findings"]: |
|
print("\nDry run. Re-run with --apply to rewrite these files.") |
|
|
|
return 1 if (result["total_findings"] and not args.apply) else 0 |
|
|
|
|
|
if __name__ == "__main__": |
|
sys.exit(main()) |
