104 lines
3.8 KiB
Python
Executable File
104 lines
3.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Mask a secret everywhere the chat history keeps it.
|
|
|
|
Read the secret on stdin, never as an argument — argv is visible in `ps` to any
|
|
local user, and it lands in the transcript a second time.
|
|
|
|
printf '%s' "$SECRET" | tools/mask-secret.py
|
|
|
|
Replaces the secret with asterisks of the SAME LENGTH. That matters: the live
|
|
session is appending to its transcript while this runs, so the file is opened
|
|
'r+b' and overwritten in place. No rename, no truncate, the inode survives and
|
|
the writer never notices. A `sed -i` would swap the inode out and the running
|
|
session's appends would vanish into an unlinked file.
|
|
|
|
WHAT THIS OWNS: Claude Code's own stores, on any machine —
|
|
`~/.claude/projects/**/*.jsonl`. Nothing else is universal.
|
|
|
|
ANYTHING ELSE IS A HOOK. A machine or platform that keeps its own copy of the
|
|
chat (a database, a log shipper, a mirror) drops an executable into
|
|
`mask-hooks.d/` beside this file, or wherever `MASK_HOOKS_DIR` points. Each hook
|
|
receives two lines on stdin — the secret, then the mask — and prints one line
|
|
saying what it did. Hooks are specific to a place; this file stays general.
|
|
|
|
One paste lands in a transcript about three times — the `user` message, a
|
|
`queue-operation` record, and a rolling `last-prompt` cache — plus once more for
|
|
every command that carries it. Do not assume one hit.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
PROJECTS = Path(os.environ.get("MASK_PROJECTS_DIR", Path.home() / ".claude" / "projects"))
|
|
HOOKS_DIR = Path(os.environ.get("MASK_HOOKS_DIR", Path(__file__).resolve().parent / "mask-hooks.d"))
|
|
|
|
|
|
def mask_file(path: Path, secret: bytes, mask: bytes) -> int:
|
|
"""Overwrite in place, preserving the inode. Returns occurrences masked."""
|
|
try:
|
|
with open(path, "r+b") as f:
|
|
data = f.read()
|
|
hits = [m.start() for m in re.finditer(re.escape(secret), data)]
|
|
for off in hits:
|
|
f.seek(off)
|
|
f.write(mask)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
return len(hits)
|
|
except (OSError, PermissionError) as err:
|
|
print(f" ! {path}: {err}", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
def run_hooks(secret: str, mask: str) -> None:
|
|
"""Every executable in HOOKS_DIR gets secret+mask on stdin."""
|
|
if not HOOKS_DIR.is_dir():
|
|
return
|
|
hooks = sorted(h for h in HOOKS_DIR.iterdir() if h.is_file() and os.access(h, os.X_OK))
|
|
if not hooks:
|
|
return
|
|
print(f"\nhooks ({HOOKS_DIR}):")
|
|
for hook in hooks:
|
|
try:
|
|
out = subprocess.run(
|
|
[str(hook)], input=f"{secret}\n{mask}\n",
|
|
capture_output=True, text=True, timeout=60,
|
|
)
|
|
line = (out.stdout.strip() or out.stderr.strip() or "(no output)").splitlines()[0]
|
|
status = "" if out.returncode == 0 else f" [exit {out.returncode}]"
|
|
print(f" {hook.name}: {line}{status}")
|
|
except subprocess.TimeoutExpired:
|
|
print(f" {hook.name}: TIMED OUT")
|
|
except OSError as err:
|
|
print(f" {hook.name}: failed to run ({err})")
|
|
|
|
|
|
def main() -> int:
|
|
secret = sys.stdin.read().strip()
|
|
if len(secret) < 12:
|
|
print("refusing: secret under 12 chars would mask too much", file=sys.stderr)
|
|
return 2
|
|
mask = "*" * len(secret)
|
|
|
|
total = 0
|
|
for path in sorted(PROJECTS.rglob("*.jsonl")):
|
|
n = mask_file(path, secret.encode(), mask.encode())
|
|
if n:
|
|
total += n
|
|
print(f" {n:>3} {path}")
|
|
print(f"transcripts: {total} occurrence(s) masked")
|
|
|
|
run_hooks(secret, mask)
|
|
|
|
print("\nStill holding the real value, deliberately:")
|
|
print(" - the tool's own config (that is how it authenticates)")
|
|
print(" - this session's live context, re-sent to the API each turn")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|