beam: standalone authorship-aware counterparty signature tool

wizard-work/beam/counterparty_sig.py

@@ -0,0 +1,90 @@
+#!/usr/bin/env python3
+# Authorship-aware repository signature (Q15 monitor logic, standalone).
+import hashlib
+import subprocess
+import sys
+
+
+def run(repo, *args):
+    r = subprocess.run(["git", "-C", repo, *args], capture_output=True, text=True)
+    if r.returncode != 0:
+        raise RuntimeError("git %s: %s" % (" ".join(args), r.stderr.strip()))
+    return r.stdout
+
+
+def commits(repo, n=200):
+    out = run(repo, "log", "-%d" % n, "--format=%H%x1f%an%x1f%ae")
+    rows = []
+    for line in out.splitlines():
+        if not line.strip():
+            continue
+        parts = line.split("\x1f")
+        if len(parts) != 3:
+            continue
+        rows.append((parts[0], parts[1], parts[2]))
+    return rows
+
+
+def is_self(author_name, author_email, self_name):
+    return author_name == self_name or author_email == self_name
+
+
+def newest_commit_for_path(repo, path, self_name, cache):
+    if path in cache:
+        return cache[path]
+    out = run(repo, "log", "-1", "--format=%H%x1f%an%x1f%ae", "--", path)
+    line = out.strip()
+    if not line:
+        cache[path] = None
+        return None
+    parts = line.split("\x1f")
+    sha, name, email = parts[0], parts[1], parts[2]
+    res = (sha, name, email, not is_self(name, email, self_name))
+    cache[path] = res
+    return res
+
+
+def signature(repo, self_name, folders):
+    head = None
+    for sha, name, email in [(c[0], c[1], c[2]) for c in commits(repo)]:
+        if not is_self(name, email, self_name):
+            head = sha
+            break
+    if head is None:
+        raise RuntimeError("no counterparty commit found")
+
+    files = []
+    for folder in folders:
+        out = run(repo, "ls-tree", "-r", "--name-only", head, folder)
+        for p in out.splitlines():
+            if p.strip():
+                files.append(p.strip())
+    files = sorted(set(files))
+
+    cache = {}
+    lines = ["head %s" % head[:12]]
+    for path in files:
+        info = newest_commit_for_path(repo, path, self_name, cache)
+        counterparty = True if info is None else info[3]
+        if not counterparty:
+            continue
+        blob = run(repo, "rev-parse", "%s:%s" % (head, path)).strip()
+        lines.append("%s %s" % (path, blob[:12]))
+
+    text = "\n".join(lines) + "\n"
+    digest = hashlib.sha256(text.encode()).hexdigest()
+    return text, digest
+
+
+def main():
+    if len(sys.argv) < 4:
+        print("usage: counterparty_sig.py <repo> <self_name> <folder> [folder ...]", file=sys.stderr)
+        sys.exit(2)
+    repo, self_name, folders = sys.argv[1], sys.argv[2], sys.argv[3:]
+    text, digest = signature(repo, self_name, folders)
+    sys.stdout.write(text)
+    print("sha256 %s" % digest)
+
+
+if __name__ == "__main__":
+    main()