#!/usr/bin/env python3 """Save one real session as a card-gallery scenario (dev tooling; stdlib only). export-scenario.py --id lead-issue-3 --title "SWE Lead · issue #3" [--base http://127.0.0.1:8765/v1] [--token-file PATH] [--note TEXT] [--out src/gallery/scenarios] [--max-tool-chars 1200] [--replace OLD=NEW ...] Reads the session row, its workers, the stored transcript, the pending Inbox items and the board from a running server (`--base` is the API root up to and including /v1; for a session on a joined machine, pass that machine's proxy root). The token, when one is needed, is read from a FILE so it never appears in argv or shell history. What is written is scrubbed: secrets and bearer tokens, email addresses, home directories, AWS account ids, and long tool output (truncated). The session id is replaced by a scenario-local one. `scenarios.test.ts` re-checks every file, so a leak fails the unit tests too. Exports are for THIS machine: name them `real-…` and they are gitignored (`src/gallery/scenarios/real-*.json`). Scrubbing catches patterns, not meaning, so a real session is never committed; write a hand-made scenario for anything shared. """ import argparse import json import os import re import sys import urllib.parse import urllib.request SECRET_PATTERNS = [ (re.compile(r"sk-[A-Za-z0-9_-]{16,}"), "sk-REDACTED"), (re.compile(r"gh[pousr]_[A-Za-z0-9]{20,}"), "gh_REDACTED"), (re.compile(r"github_pat_[A-Za-z0-9_]{20,}"), "gh_REDACTED"), (re.compile(r"xox[abpr]-[A-Za-z0-9-]{10,}"), "xox-REDACTED"), (re.compile(r"AKIA[0-9A-Z]{16}"), "AKIAREDACTED"), (re.compile(r"(?i)(bearer\s+)[A-Za-z0-9._~+/=-]{20,}"), r"\1REDACTED"), (re.compile(r"eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}"), "JWT_REDACTED"), (re.compile(r"(?i)((?:password|passwd|secret|api[_-]?key|token)\"?\s*[:=]\s*\"?)[^\s\"',}]{8,}"), r"\1REDACTED"), (re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----", re.S), "PRIVATE_KEY_REDACTED"), ] EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}") HOME = re.compile(r"/(?:Users|home)/[A-Za-z0-9._-]+") AWS_ACCOUNT = re.compile(r"(? str: for pattern, repl in SECRET_PATTERNS: text = pattern.sub(repl, text) text = EMAIL.sub("user@example.com", text) text = HOME.sub("/home/user", text) return AWS_ACCOUNT.sub("123456789012", text) def scrub(value): if isinstance(value, str): return scrub_text(value) if isinstance(value, list): return [scrub(v) for v in value] if isinstance(value, dict): return {k: scrub(v) for k, v in value.items()} return value def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("session_id") ap.add_argument("--id", required=True, help="scenario id (file name and URL slug)") ap.add_argument("--title", required=True) ap.add_argument("--note", default="") ap.add_argument("--base", default="http://127.0.0.1:8765/v1") ap.add_argument("--token-file", default="") ap.add_argument("--out", default=os.path.join(os.path.dirname(__file__), "..", "src", "gallery", "scenarios")) ap.add_argument("--max-tool-chars", type=int, default=1200) ap.add_argument("--replace", action="append", default=[], metavar="OLD=NEW", help="replace a name everywhere (a GitHub handle, a customer); repeatable") args = ap.parse_args() if not re.fullmatch(r"[a-z0-9-]+", args.id): ap.error("--id must be lowercase letters, digits and dashes") if not args.id.startswith("real-"): args.id = "real-" + args.id # the gitignored prefix: an export never gets committed headers = {} if args.token_file: with open(os.path.expanduser(args.token_file)) as fh: headers["Authorization"] = "Bearer " + fh.read().rstrip() def get(path: str, default=None): try: req = urllib.request.Request(args.base.rstrip("/") + path, headers=headers) return json.load(urllib.request.urlopen(req, timeout=60)) except Exception as exc: # noqa: BLE001 - optional endpoints may be missing on older servers if default is None: raise SystemExit(f"GET {path} failed: {exc}") return default sid = args.session_id quoted = urllib.parse.quote(sid) rows = get("/sessions") rows = rows if isinstance(rows, list) else rows.get("sessions", []) row = next((s for s in rows if s.get("session_id") == sid), None) if row is None: raise SystemExit(f"no session {sid} on {args.base}") workers = [s for s in rows if (s.get("team") or {}).get("lead_session") == sid] messages = get(f"/sessions/{quoted}/messages").get("messages", []) pending = get(f"/inbox?state=pending&session_id={quoted}", {"items": []}).get("items", []) board = get(f"/sessions/{quoted}/board", {}) unattended = get(f"/sessions/{quoted}/unattended", {}).get("unattended") for m in messages: # tool output is the bulk of a transcript and the likeliest leak if m.get("role") == "tool" and isinstance(m.get("content"), str) and len(m["content"]) > args.max_tool_chars: cut = len(m["content"]) - args.max_tool_chars m["content"] = m["content"][: args.max_tool_chars] + f"\n… [{cut} characters cut by export-scenario]" local_sid = "scn-" + args.id team = row.get("team") or {} scenario = { "id": args.id, "title": args.title, "note": args.note, "exported": {"from_session": sid[:8] + "…", "messages": len(messages)}, "session": { "session_id": local_sid, "title": row.get("title") or "", "agent": row.get("agent") or "", "model": row.get("model") or "", "mode": row.get("mode") or "interactive", "workspace": row.get("workspace") or "", # No answer (older server): a session with parked items is, in effect, unattended. "unattended": bool(unattended) if unattended is not None else bool(pending), }, "messages": messages, "pending": pending, } if workers: scenario["team"] = { "team_id": team.get("team_id") or "team", "chat_enabled": bool(team.get("chat_enabled")), "workers": [ { "name": (w.get("team") or {}).get("actor") or w.get("title") or "worker", "persona": w.get("agent") or "", "model": w.get("model") or "", "status": (w.get("team") or {}).get("status") or "idle", "current_item": (w.get("team") or {}).get("current_item") or "", "usage": w.get("usage") or {}, } for w in workers ], } if isinstance(board, dict) and board.get("items"): scenario["board"] = board # The real session id appears inside items, board rows and sometimes tool output. text = json.dumps(scrub(scenario), indent=2, ensure_ascii=False).replace(sid, local_sid) for pair in args.replace: old, _, new = pair.partition("=") if old: text = text.replace(old, new) out = os.path.join(args.out, args.id + ".json") with open(out, "w") as fh: fh.write(text + "\n") print(f"wrote {os.path.normpath(out)}: {len(messages)} messages, {len(pending)} pending, {len(workers)} workers") print("Gitignored (real-*.json): it stays on this machine.") return 0 if __name__ == "__main__": sys.exit(main())