#!/usr/bin/env python3 """Inventory everything that reads the forge without credentials (ORGREF-WP-0002-T02). Read-only. Combines the runtime capture (ArgoCD sources and workload images) with a fresh scan of local working trees for forge URLs, classifies each hit by how it reads the forge, and groups the result into consumers: the repo or workload that reads, what it reads, and how. A consumer that reads over SSH with a personal key is recorded but is not anonymous. python3 tools/access_consumers.py --root /home/worsch \ --runtime inventory/runtime-2026-09-28.json \ --output inventory/access-consumers-2026-10-08.json """ import argparse import collections import csv import json import os import re import stat import subprocess from pathlib import Path from inventory import BINARY, SENSITIVE FORGE_HOSTS = ("forgejo.coulomb.social", "gitea.coulomb.social") URL = re.compile( r"(?:https?://)?(?Pforgejo\.coulomb\.social|gitea\.coulomb\.social)" r"(?P/[A-Za-z0-9._~%/@:+-]*)?") SKIP_DIRS = {".git", "node_modules", "vendor", "history", "archived", "__pycache__", ".venv", "venv", "dist", "build", ".terraform", "rescued"} SKIP_FILES = {"WORK-RECORDS.md", ".custodian-brief.md"} MAX_BYTES = 1_000_000 # How a reference reads the forge. First match wins; order matters. KINDS = [ ("record", lambda f, p, line: f.startswith(("docs/evidence/", "evidence/", "tests/", "test/")) or "/evidence/" in f or "/tests/" in f or f.endswith(("PKG-INFO", ".sum"))), ("raw-fetch", lambda f, p, line: "/raw/" in p or "/media/" in p), ("package-consumer", lambda f, p, line: p.startswith(("/api/packages/", "/api/v1/packages/"))), ("api-client", lambda f, p, line: p.startswith("/api/")), ("ci-action", lambda f, p, line: ".forgejo/workflows" in f or ".gitea/workflows" in f or ".github/workflows" in f), ("package-consumer", lambda f, p, line: Path(f).name in { "go.mod", "go.sum", "pyproject.toml", "requirements.txt", "package.json", "uv.lock", "Cargo.toml", "setup.cfg"} or "/api/packages/" in p), ("gitops-source", lambda f, p, line: "repoURL" in line or p.endswith(".git") and f.endswith((".yaml", ".yml"))), ("image-pull", lambda f, p, line: bool(re.search(r"(image|repository|FROM)\b", line)) or "@sha256:" in line or Path(f).name.startswith(("Dockerfile", "Containerfile"))), ("doc-link", lambda f, p, line: f.endswith((".md", ".rst", ".txt", ".html"))), ("build-tooling", lambda f, p, line: Path(f).name in {"Makefile", "justfile"} or f.endswith((".sh", ".mk"))), ("code-reference", lambda f, p, line: f.endswith((".py", ".go", ".ts", ".js", ".rs"))), ("config-reference", lambda f, p, line: f.endswith((".yaml", ".yml", ".json", ".toml", ".ini", ".service", ".template", ".example"))), ] def classify(file, path, line): for kind, test in KINDS: if test(file, path or "", line): return kind return "other" def target_repo(path): """org/repo named by a forge URL path, if any.""" parts = [p for p in (path or "").split("/") if p] if parts[:2] == ["api", "v1"] and len(parts) >= 5 and parts[2] == "repos": parts = parts[3:] elif parts and parts[0] == "api": # Package paths name an owner/index, not a repository. Other API # endpoints also cannot establish a repository identity. return None elif parts and parts[0] == "v2": parts = parts[1:] if len(parts) >= 2 and parts[0] not in {"user", "explore", "assets"}: return f"{parts[0]}/{parts[1].removesuffix('.git').split('@')[0].split(':')[0]}" return None def scan_repo(repo, diagnostics=None): hits = [] diagnostics = diagnostics if diagnostics is not None else [] def excluded(path, reason): diagnostics.append({"repository": repo.name, "path": path, "reason": reason}) def walk_error(error): excluded(str(Path(error.filename).relative_to(repo)), "unreadable-directory") for dirpath, dirnames, filenames in os.walk(repo, onerror=walk_error): kept = [] for d in sorted(dirnames): rel = str((Path(dirpath) / d).relative_to(repo)) if (Path(dirpath) / d).is_symlink(): excluded(rel, "symlink-not-followed") elif d in SKIP_DIRS or (d.startswith(".") and d not in {".forgejo", ".gitea", ".github"}): excluded(rel, "excluded-directory") else: kept.append(d) dirnames[:] = kept for name in sorted(filenames): full = Path(dirpath) / name rel = str(full.relative_to(repo)) if name in SKIP_FILES: excluded(rel, "generated-work-record") continue if SENSITIVE.search(rel): excluded(rel, "sensitive-or-encrypted") continue if full.suffix.lower() in BINARY: excluded(rel, "binary-extension") continue try: if full.is_symlink(): excluded(rel, "symlink-not-followed") continue info = full.stat() if not stat.S_ISREG(info.st_mode): excluded(rel, "non-regular-file") continue if info.st_size > MAX_BYTES: excluded(rel, "over-1MB") continue text = full.read_text(encoding="utf-8") except UnicodeDecodeError: excluded(rel, "non-utf8-content") continue except OSError: excluded(rel, "unreadable-file") continue if "\0" in text: excluded(rel, "binary-content") continue for number, line in enumerate(text.splitlines(), 1): for m in URL.finditer(line): hits.append({ "file": rel, "line": number, "host": m.group("host"), "kind": classify(rel, m.group("path") or "", line), "target": target_repo(m.group("path")), }) return hits def remote_form(repo): try: url = subprocess.run(["git", "-C", str(repo), "remote", "get-url", "origin"], capture_output=True, text=True, check=True).stdout.strip() except (subprocess.CalledProcessError, OSError): return None if url.startswith(("forgejo-remote:", "ssh://", "git@")): return "ssh-key" if any(h in url for h in FORGE_HOSTS): return "https" return "other" def runtime_consumers(runtime): out = [] for app in runtime.get("applications", []): for src in app.get("sources", []): u = src.get("repoURL", "") m = URL.search(u) if m: out.append({"consumer": f"argocd:{app['name']}", "class": "cluster-gitops", "host": m.group("host"), "kind": "gitops-source", "target": target_repo(m.group("path")), "credential": "unknown (ArgoCD repo credentials not inspected)"}) for w in runtime.get("workloads", []): for image in w.get("images", []): m = URL.search(image) if m: secrets = w.get("image_pull_secret_names") or [] out.append({"consumer": f"k3s:{w['namespace']}/{w['kind']}/{w['name']}", "class": "cluster-image-pull", "kind": "image-pull", "host": m.group("host"), "target": target_repo(m.group("path")), "credential": "pull secret named: " + ",".join(secrets) if secrets else "unknown (no imagePullSecrets; node registry auth not inspected)"}) return out CLASS_OF_KIND = { "raw-fetch": "raw-file-fetch", "api-client": "api-client", "ci-action": "ci-runner", "package-consumer": "package-consumer", "gitops-source": "repo-gitops", "image-pull": "repo-image-reference", "doc-link": "doc-link", "other": "other", "record": "record-or-test", "build-tooling": "build-tooling", "code-reference": "code-reference", "config-reference": "config-reference", } def main(): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--root", default="/home/worsch") ap.add_argument("--runtime", required=True) ap.add_argument("--assignments", default="taxonomy/assignments.csv") ap.add_argument("--output", required=True) args = ap.parse_args() stewards = {} with open(args.assignments, newline="") as fh: for row in csv.DictReader(fh): stewards[row["repository"]] = row["proposed_org"] runtime = json.load(open(args.runtime)) consumers = runtime_consumers(runtime) remotes = collections.Counter() scanned = [] diagnostics = [] coordinator = Path(__file__).resolve().parents[1] for repo in sorted(Path(args.root).iterdir()): if not (repo / ".git").exists() or repo.name.startswith("rescued-") or repo.resolve() == coordinator: continue scanned.append(repo.name) remotes[remote_form(repo) or "none"] += 1 grouped = collections.defaultdict(list) for h in scan_repo(repo, diagnostics): grouped[(h["host"], h["kind"], h["target"])].append(f"{h['file']}:{h['line']}") for (host, kind, target), where in sorted(grouped.items(), key=lambda kv: (kv[0][0], kv[0][1], str(kv[0][2]))): consumers.append({"consumer": f"repo:{repo.name}", "class": CLASS_OF_KIND[kind], "host": host, "kind": kind, "target": target, "locations": where[:10], "location_count": len(where)}) for c in consumers: name = (c.get("target") or "").split("/")[-1] c["target_steward"] = stewards.get(name, "unassigned") summary = collections.Counter(c["class"] for c in consumers) report = { "schema_version": "0.2", "workplan": "ORGREF-WP-0002-T02", "runtime_capture": os.path.basename(args.runtime), "scanned_repositories": len(scanned), "local_remote_forms": dict(remotes), "summary_by_class": dict(sorted(summary.items())), "image_pulls_without_named_credentials": sum(1 for c in consumers if c["class"] == "cluster-image-pull" and c["credential"].startswith("unknown")), "consumers": consumers, "scan_exclusions": diagnostics, "scan_error_count": sum(d["reason"].startswith("unreadable") for d in diagnostics), "limits": [ "Runtime classes come from the runtime capture named above, not a live read.", "ArgoCD repository credentials and pull-secret contents were not inspected; " "a named pull secret is not proof that it authenticates to the forge.", "Missing imagePullSecrets do not prove anonymous pulls: node-level registry " "auth and service-account defaults require owner evidence.", "Sensitive/encrypted, binary, large, symlinked and excluded paths are recorded " "in scan_exclusions; unreadable paths prevent a completeness claim.", "The coordinator is excluded to prevent generated inventory self-reference.", "Hostnames are preserved so legacy-host references can be reviewed separately.", "Package indexes name a namespace, not a verified repository; their target is null.", "Only literal forge hostnames are found; computed URLs and consumers outside " "these working trees (external sites, Telegram, other people's clones) are not.", "history/ and archived/ directories are skipped; they record, they do not read.", ], } Path(args.output).write_text(json.dumps(report, indent=1, sort_keys=False) + "\n") print(json.dumps({k: report[k] for k in ("scanned_repositories", "local_remote_forms", "summary_by_class", "image_pulls_without_named_credentials", "scan_error_count")}, indent=1)) if __name__ == "__main__": main()