Audit which repositories Grok Build collected and whether xAI still lists the archives

grok-upload-audit.py

#!/usr/bin/env python3
"""Find repositories collected by Grok Build.

Usage:
  python3 grok-upload-audit.py           # read local Grok logs only
  python3 grok-upload-audit.py --remote  # also ask xAI if archive IDs exist

The remote check sends only logged archive IDs to xAI. It never uploads or
downloads source code and never prints credentials or archive IDs.

Labels:
  COLLECTED-ONLY  Grok's collector started; no matching queue event was found.
  QUEUED          An archive reached Grok's persistent upload queue.
  REMOTE-PRESENT  xAI's API still recognizes at least one exact archive ID.
  NOT-LISTED      xAI reports the IDs absent. This does not prove deletion from
                  backups, replicas, derived data, or other storage systems.
"""

import json
import os
import sys
import urllib.error
import urllib.request
from collections import defaultdict
from pathlib import Path

REMOTE = "--remote" in sys.argv
LOG = Path(os.getenv("GROK_LOG", "~/.grok/logs/unified.jsonl")).expanduser()
AUTH = Path(os.getenv("GROK_AUTH", "~/.grok/auth.json")).expanduser()
API = "https://cli-chat-proxy.grok.com/v1/storage/batch_exists"

if "--help" in sys.argv or "-h" in sys.argv:
    raise SystemExit(__doc__)
if not LOG.is_file():
    raise SystemExit(f"Grok log not found: {LOG}")

starts = {}
repos = defaultdict(lambda: {"attempts": 0, "archives": [], "last": ""})

for line in LOG.open(encoding="utf-8", errors="replace"):
    try:
        event = json.loads(line)
    except json.JSONDecodeError:
        continue

    ctx = event.get("ctx") or {}
    msg, sid = event.get("msg"), event.get("sid")
    turn, timestamp = ctx.get("turn_number"), event.get("ts") or ""

    if msg == "repo_state.upload.start" and isinstance(ctx.get("repo_path"), str):
        repo = ctx["repo_path"]
        starts[sid, turn, ctx.get("phase")] = repo
        repos[repo]["attempts"] += 1
        repos[repo]["last"] = max(repos[repo]["last"], timestamp)

    elif msg == "repo_state.upload.enqueued" and isinstance(ctx.get("gcs_path"), str):
        archive = ctx["gcs_path"]
        name = archive.rsplit("/", 1)[-1]
        if name not in {"before_codebase.tar.gz", "after_codebase.tar.gz"}:
            continue
        phase = name.removesuffix(".tar.gz")
        repo = starts.get((sid, turn, phase), "<unknown: earlier log missing>")
        repos[repo]["archives"].append(archive)
        repos[repo]["last"] = max(repos[repo]["last"], timestamp)

paths = list(dict.fromkeys(p for row in repos.values() for p in row["archives"]))
present, absent = set(), set()

if REMOTE and paths:
    auth = json.loads(AUTH.read_text(encoding="utf-8"))
    entries = [v for v in auth.values() if isinstance(v, dict) and v.get("key")]
    if not entries:
        raise SystemExit("No cached token found; run `grok login` first.")
    token = max(entries, key=lambda v: v.get("expires_at", ""))["key"]

    for offset in range(0, len(paths), 100):
        body = json.dumps({"paths": paths[offset : offset + 100]}).encode()
        request = urllib.request.Request(
            API,
            data=body,
            method="POST",
            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        )
        try:
            with urllib.request.urlopen(request, timeout=20) as response:
                result = json.load(response)
        except urllib.error.HTTPError as error:
            raise SystemExit(f"Storage check failed with HTTP {error.code}.") from None
        except urllib.error.URLError as error:
            raise SystemExit(f"Storage check failed: {error.reason}") from None
        present.update(result.get("exists", []))
        absent.update(result.get("missing", []))

home = str(Path.home())
print("STATUS\tARCHIVES\tATTEMPTS\tLAST EVENT\tREPOSITORY")

for repo, row in sorted(repos.items()):
    archives = row["archives"]
    if not archives:
        status, summary = "COLLECTED-ONLY", "0"
    elif not REMOTE:
        status, summary = "QUEUED", str(len(archives))
    else:
        yes = sum(p in present for p in archives)
        no = sum(p in absent for p in archives)
        unknown = len(archives) - yes - no
        status = "REMOTE-PRESENT" if yes else "NOT-LISTED" if no == len(archives) else "UNKNOWN"
        summary = f"{yes} present/{no} absent/{unknown} unknown"

    shown = "~" + repo[len(home) :] if repo.startswith(home + "/") else repo
    print(f"{status}\t{summary}\t{row['attempts']}\t{row['last'] or '-'}\t{shown}")

print("\nReview output before sharing: repository paths may be private.")
print("NOT-LISTED is not proof of deletion from backups, replicas, or derived data.")

PROMPT.md

Grok Build upload audit prompt

Paste this into Claude Code, Codex, or another local agent with filesystem and network access:

Perform a read-only audit of Grok Build codebase collection and retention on
this machine. Do the investigation with shell/Python; do not just give me
instructions. Do not launch Grok, send it a prompt, upload/download archives,
invoke /privacy, delete anything, or change configuration/account state. Never
print tokens, session IDs, or storage object IDs. Do not create or delete even
temporary files; process everything in memory or through stdin/stdout.

Inspect ~/.grok/logs/unified*.jsonl plus rotated logs beside it. Find
`repo_state.upload.start` and `repo_state.upload.enqueued`. Correlate events by
`sid`, `ctx.turn_number`, and phase; derive an enqueued event's phase from its
`gcs_path` basename (`before_codebase.tar.gz` or `after_codebase.tar.gz`). Map
each archive to the start event's `ctx.repo_path`.

For every repository, count collection attempts and queued archives and show
first/last timestamps. Then perform a metadata-only existence check: read the
newest cached `key` from ~/.grok/auth.json without displaying it and POST the
exact queued paths as `{"paths":[...]}` batches to:
https://cli-chat-proxy.grok.com/v1/storage/batch_exists
with `Authorization: Bearer <token>`. Never call download or follow signed URLs.

Print the UTC check time and this table:
STATUS | PRESENT/ABSENT/UNKNOWN | ATTEMPTS | FIRST/LAST | REPOSITORY

Classify COLLECTED-ONLY when collection started without a matching queue event;
QUEUED when an archive was enqueued; REMOTE-PRESENT when any exact ID is in
`exists`; NOT-LISTED only when all its IDs are in `missing`; otherwise UNKNOWN.

Finish with totals and caveats: queueing alone does not prove server retention;
REMOTE-PRESENT proves only that this API still recognizes the object;
NOT-LISTED does not prove deletion from backups, replicas, derived data, or
other stores; missing logs can cause false negatives. Warn that repository
paths are sensitive before I share the output.