108 lines
3.9 KiB
Python
Executable File
108 lines
3.9 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.
|
|
|
|
Covers both stores a human can actually read back:
|
|
- Claude Code transcripts ~/.claude/projects/**/*.jsonl
|
|
- Postgres chat_session_events (what the web chat replays)
|
|
|
|
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.home() / ".claude" / "projects"
|
|
PGDB = os.environ.get("MASK_PGDB", "officer_dev")
|
|
PGHOST = os.environ.get("MASK_PGHOST", "127.0.0.1")
|
|
PGUSER = os.environ.get("MASK_PGUSER", "postgres")
|
|
|
|
|
|
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 mask_postgres(secret: str, mask: str) -> int:
|
|
"""Replace inside the jsonb payload. Returns rows updated, -1 if unavailable."""
|
|
sql = (
|
|
"update chat_session_events set event = "
|
|
"replace(event::text, %(s)s, %(m)s)::jsonb "
|
|
"where event::text like %(l)s;"
|
|
)
|
|
# psql has no bind params, so quote by doubling any single quotes. The secret
|
|
# is hex/base64 in practice, but never build SQL from unchecked input blindly.
|
|
q = lambda v: "'" + v.replace("'", "''") + "'"
|
|
stmt = sql.replace("%(s)s", q(secret)).replace("%(m)s", q(mask)).replace("%(l)s", q(f"%{secret}%"))
|
|
env = {**os.environ, "PGPASSWORD": os.environ.get("MASK_PGPASSWORD", "postgres")}
|
|
try:
|
|
out = subprocess.run(
|
|
["psql", "-h", PGHOST, "-U", PGUSER, "-d", PGDB, "-tAc", stmt],
|
|
capture_output=True, text=True, env=env, timeout=30,
|
|
)
|
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
|
return -1
|
|
if out.returncode != 0:
|
|
print(f" ! postgres: {out.stderr.strip()[:120]}", file=sys.stderr)
|
|
return -1
|
|
return int(out.stdout.strip().split()[-1]) if out.stdout.strip() else 0
|
|
|
|
|
|
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)
|
|
sb, mb = secret.encode(), mask.encode()
|
|
|
|
total = 0
|
|
for path in sorted(PROJECTS.rglob("*.jsonl")):
|
|
n = mask_file(path, sb, mb)
|
|
if n:
|
|
total += n
|
|
print(f" {n:>3} {path}")
|
|
print(f"transcripts: {total} occurrence(s) masked")
|
|
|
|
rows = mask_postgres(secret, mask)
|
|
if rows < 0:
|
|
print("postgres: skipped (psql unavailable or errored)")
|
|
else:
|
|
print(f"postgres: {rows} row(s) updated")
|
|
|
|
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())
|