R10-02: original authenticated approval-State inventory and independent readback (nonpublishing)

.github/workflows/issue311-r10-approval-state-original.yml

@@ -0,0 +1,89 @@
+name: Issue 311 R10 original approval State gap (GET ONLY)
+on:
+  pull_request:
+    paths:
+      - 'scripts/issue311-r10-approval-state-original.py'
+      - 'tests/test-issue311-r10-approval-state-original.py'
+      - '.github/workflows/issue311-r10-approval-state-original.yml'
+  push:
+    branches: [main]
+    paths:
+      - 'scripts/issue311-r10-approval-state-original.py'
+      - 'tests/test-issue311-r10-approval-state-original.py'
+      - '.github/workflows/issue311-r10-approval-state-original.yml'
+permissions:
+  contents: read
+  actions: read
+jobs:
+  hostile:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - run: python tests/test-issue311-r10-approval-state-original.py
+  source:
+    if: github.repository == 'pgextwin/build' && github.ref == 'refs/heads/main' && github.event_name == 'push' && github.run_attempt == 1
+    runs-on: ubuntu-latest
+    timeout-minutes: 25
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - name: Original authenticated State GET with no App key
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: python scripts/issue311-r10-approval-state-original.py
+      - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: issue311-r10-approval-original-state
+          path: issue311-r10-approval-original-state.json
+          retention-days: 30
+          if-no-files-found: error
+  readback:
+    needs: source
+    runs-on: ubuntu-latest
+    timeout-minutes: 25
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - name: Download original JSON from same Run
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: gh run download "$GITHUB_RUN_ID" --repo "$GITHUB_REPOSITORY" --name issue311-r10-approval-original-state --dir .
+      - name: Rehash original outer ZIP bytes and verify exact inner JSON
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          python - <<'PY'
+          import hashlib,io,json,os,pathlib,subprocess,urllib.parse,urllib.request,zipfile
+          repo=os.environ['GITHUB_REPOSITORY']; run=os.environ['GITHUB_RUN_ID']
+          a=json.loads(subprocess.check_output(['gh','api',f'repos/{repo}/actions/runs/{run}/artifacts?per_page=100']))
+          hits=[v for v in a['artifacts'] if v.get('name')=='issue311-r10-approval-original-state']
+          assert len(hits)==1 and a['total_count']<=100
+          art=hits[0]
+          assert art['expired'] is False
+          request=urllib.request.Request(f'https://api.github.com/repos/{repo}/actions/artifacts/{art["id"]}/zip',headers={'Authorization':'Bearer '+os.environ['GH_TOKEN'],'Accept':'application/vnd.github+json'})
+          class SecureRedirect(urllib.request.HTTPRedirectHandler):
+              def redirect_request(self,request,fp,code,msg,headers,newurl):
+                  src=urllib.parse.urlparse(request.full_url); dst=urllib.parse.urlparse(newurl)
+                  assert dst.scheme=='https'
+                  if src.hostname!=dst.hostname:
+                      return urllib.request.Request(newurl,headers={'User-Agent':'pgextwin-issue311-readonly'})
+                  return super().redirect_request(request,fp,code,msg,headers,newurl)
+          with urllib.request.build_opener(SecureRedirect()).open(request,timeout=60) as r: raw=r.read()
+          sha=hashlib.sha256(raw).hexdigest()
+          assert art['digest']=='sha256:'+sha
+          with zipfile.ZipFile(io.BytesIO(raw)) as z:
+              assert z.testzip() is None
+              assert z.namelist()==['issue311-r10-approval-original-state.json']
+              inner=z.read(z.namelist()[0])
+          assert inner==pathlib.Path('issue311-r10-approval-original-state.json').read_bytes()
+          print('R10_APPROVAL_ORIGINAL_ZIP_SHA256',art['id'],len(raw),sha,'INNER',hashlib.sha256(inner).hexdigest())
+          PY
+      - name: Independently re-GET State (no App key)
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: python scripts/issue311-r10-approval-state-original.py --verify

scripts/issue311-r10-approval-state-original.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+"""R10-02 authenticated GET-only State approval-evidence inventory; no writer."""
+from __future__ import annotations
+import base64
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+
+REPO = "pgextwin/build"
+STATE_REF = "issue311-state"
+DIR = "checkpoints/request-lifecycle"
+SHA40 = re.compile(r"[0-9a-f]{40}\Z")
+SHA64 = re.compile(r"[0-9a-f]{64}\Z")
+ID = re.compile(r"r-[0-9]{8}-[a-z0-9-]{8,56}\Z")
+TERMINALS = {None, "NO_UPDATE", "QUARANTINED", "FAILED", "BLOCKED_APPROVAL", "SUCCESS"}
+
+def need(ok, why):
+    if not ok:
+        raise ValueError("R10_APPROVAL_ORIGINAL_DENY " + why)
+
+def blob_sha(raw):
+    return hashlib.sha1(b"blob " + str(len(raw)).encode() + b"\0" + raw).hexdigest()
+
+def unique_json(raw):
+    def pairs(items):
+        result = {}
+        for key, value in items:
+            need(key not in result, "duplicate JSON field")
+            result[key] = value
+        return result
+    return json.loads(raw, object_pairs_hook=pairs)
+
+def inspect(name, sha, raw):
+    need(type(name) is str and name.endswith(".json") and ID.fullmatch(name[:-5]), "bad original filename")
+    need(SHA40.fullmatch(sha or "") and blob_sha(raw) == sha, "Git blob/bytes mismatch")
+    state = unique_json(raw)
+    need(type(state) is dict and state.get("schemaVersion") == 1
+         and state.get("scope") == "AUDIT_ONLY_READONLY"
+         and state.get("requestId") == name[:-5]
+         and SHA64.fullmatch(state.get("requestSha256") or "")
+         and SHA40.fullmatch(state.get("sourceHeadSha") or "")
+         and type(state.get("revision")) is int and state["revision"] >= 0
+         and state.get("publisherAuthorized") is False
+         and state.get("writeOutcomeUnknown") is False
+         and state.get("terminal") in TERMINALS,
+         "unsafe/invalid State")
+    targets, outcomes = state.get("targetScope"), state.get("targets")
+    need(type(targets) is list and 1 <= len(targets) <= 17
+         and len(set(targets)) == len(targets)
+         and all(type(t) is str and t.startswith("pgextwin/") for t in targets)
+         and type(outcomes) is dict and set(outcomes) == set(targets),
+         "missing/duplicate target")
+    blocked = [t for t in targets if outcomes[t] == "BLOCKED_APPROVAL"]
+    need(not blocked or state["terminal"] == "BLOCKED_APPROVAL", "blocked target with wrong terminal")
+    need(state["terminal"] != "BLOCKED_APPROVAL" or bool(blocked), "terminal without blocked target")
+    for key in ("sourceRun", "completionRun"):
+        run = state.get(key)
+        need(run is None or (type(run) is dict and type(run.get("id")) is int
+             and run["id"] > 0 and type(run.get("attempt")) is int
+             and 1 <= run["attempt"] <= 3), "invalid original Run locator")
+    return {"requestId": name[:-5], "blobSha": sha, "revision": state["revision"],
+            "terminal": state["terminal"], "targetCount": len(targets),
+            "blockedTargetCount": len(blocked), "sourceRun": state.get("sourceRun"),
+            "sourceHeadSha": state["sourceHeadSha"],
+            "humanApprovalEvidenceVerified": False, "realApprovalCasReceiptVerified": False,
+            "r10_02Accepted": False}
+
+def snapshot(commit, listing, reader):
+    need(SHA40.fullmatch(commit or ""), "invalid State ref")
+    need(type(listing) is list and len(listing) >= 29, "historical 29 checkpoint scope incomplete")
+    seen = set()
+    entries = []
+    for item in sorted(listing, key=lambda v: v.get("name", "")):
+        name = item.get("name")
+        need(item.get("type") == "file" and name not in seen
+             and item.get("path") == DIR + "/" + name, "missing/duplicate/nested original")
+        seen.add(name)
+        entries.append(inspect(name, item.get("sha"), reader(item, commit)))
+    count = sum(int(row["blockedTargetCount"] > 0) for row in entries)
+    return {"schemaVersion": 1, "mode": "AUTHENTICATED_GET_ONLY_NONPUBLISHING",
+            "scope": "R10_02_ACTUAL_APPROVAL_EVIDENCE_GAP",
+            "stateRef": STATE_REF, "stateCommitSha": commit,
+            "checkpointCount": len(entries), "blockedApprovalCheckpointCount": count,
+            "blockedApprovalTargetCount": sum(row["blockedTargetCount"] for row in entries),
+            "gap": "NO_ACTUAL_BLOCKED_APPROVAL" if count == 0 else
+                   "BLOCKED_STATE_HUMAN_APPROVAL_AND_ORIGINAL_CAS_STILL_UNVERIFIED",
+            "r10_02Accepted": False, "publicationAuthorized": False,
+            "writerUsed": False, "cronChanged": False, "checkpoints": entries}
+
+def gh(path):
+    return unique_json(subprocess.check_output(["gh", "api", path], timeout=90))
+
+def reader(item, commit):
+    record = gh("repos/" + REPO + "/contents/" + item["path"] + "?ref=" + commit)
+    need(record.get("sha") == item["sha"] and record.get("path") == item["path"]
+         and record.get("encoding") == "base64", "Contents GET mismatch")
+    raw = base64.b64decode(record["content"], validate=False)
+    need(len(raw) == record.get("size"), "State original length mismatch")
+    return raw
+
+def observe():
+    ref = "repos/" + REPO + "/git/ref/heads/" + STATE_REF
+    commit = gh(ref)["object"]["sha"]
+    items = gh("repos/" + REPO + "/contents/" + DIR + "?ref=" + commit)
+    proof = snapshot(commit, items, reader)
+    need(gh(ref)["object"]["sha"] == commit, "State changed during snapshot")
+    return proof
+
+def serial(doc):
+    return (json.dumps(doc, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + "\n").encode("ascii")
+
+def main():
+    need(os.environ.get("GITHUB_REPOSITORY") == REPO
+         and os.environ.get("GITHUB_REF") == "refs/heads/main"
+         and os.environ.get("GITHUB_EVENT_NAME") == "push"
+         and os.environ.get("GITHUB_RUN_ATTEMPT") == "1"
+         and bool(os.environ.get("GH_TOKEN")), "not authenticated original push")
+    output = Path("issue311-r10-approval-original-state.json")
+    raw = serial(observe())
+    if "--verify" in sys.argv[1:]:
+        need(output.read_bytes() == raw, "separate Runner original State mismatch")
+    else:
+        need(not output.exists(), "never overwrite receipt")
+        output.write_bytes(raw)
+    data = unique_json(raw)
+    print("R10_APPROVAL_STATE_ORIGINAL", data["stateCommitSha"],
+          "checkpoints", data["checkpointCount"],
+          "blocked", data["blockedApprovalCheckpointCount"],
+          "JSON_SHA256", hashlib.sha256(raw).hexdigest(), "R10_02_PENDING")
+
+if __name__ == "__main__":
+    main()

tests/test-issue311-r10-approval-state-original.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+"""R10 genuine approval evidence-gap hostile fixtures; no network or App key."""
+import copy
+import importlib.util
+import json
+from pathlib import Path
+
+p = Path(__file__).resolve().parents[1] / "scripts/issue311-r10-approval-state-original.py"
+spec = importlib.util.spec_from_file_location("r10approval", p)
+m = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(m)
+name = "r-20261031-r10approvalinventory01.json"
+s = {"schemaVersion": 1, "scope": "AUDIT_ONLY_READONLY",
+     "requestId": name[:-5], "requestSha256": "b"*64,
+     "sourceHeadSha": "a"*40, "revision": 2,
+     "publisherAuthorized": False, "writeOutcomeUnknown": False,
+     "terminal": "QUARANTINED",
+     "targetScope": ["pgextwin/orafce"],
+     "targets": {"pgextwin/orafce": "QUARANTINED"},
+     "sourceRun": {"id": 42, "attempt": 1}, "completionRun": None}
+
+def raw(obj):
+    return (json.dumps(obj, sort_keys=True) + "\n").encode()
+
+def deny(fn):
+    try:
+        fn()
+    except (ValueError, KeyError, TypeError):
+        return
+    raise AssertionError("R10_FALSE_APPROVAL_ACCEPTED")
+
+original = raw(s)
+assert m.inspect(name, m.blob_sha(original), original)["blockedTargetCount"] == 0
+blocked = copy.deepcopy(s)
+blocked["targets"]["pgextwin/orafce"] = "BLOCKED_APPROVAL"
+deny(lambda: m.inspect(name, m.blob_sha(raw(blocked)), raw(blocked)))
+blocked["terminal"] = "BLOCKED_APPROVAL"
+record = m.inspect(name, m.blob_sha(raw(blocked)), raw(blocked))
+assert record["blockedTargetCount"] == 1
+assert record["humanApprovalEvidenceVerified"] is False
+assert record["r10_02Accepted"] is False
+for key, value in (("publisherAuthorized", True), ("writeOutcomeUnknown", True),
+                   ("sourceHeadSha", "forged"), ("scope", "PUBLISH_ELIGIBLE"),
+                   ("revision", -1)):
+    bad = copy.deepcopy(s)
+    bad[key] = value
+    deny(lambda bad=bad: m.inspect(name, m.blob_sha(raw(bad)), raw(bad)))
+deny(lambda: m.inspect(name, "f"*40, original))
+deny(lambda: m.inspect("forged.json", m.blob_sha(original), original))
+listing, content = [], {}
+for index in range(29):
+    entry = copy.deepcopy(s)
+    filename = f"r-20261031-r10original{index:03d}.json"
+    entry["requestId"] = filename[:-5]
+    value = raw(entry)
+    path = m.DIR + "/" + filename
+    listing.append({"name": filename, "path": path, "type": "file",
+                    "sha": m.blob_sha(value)})
+    content[path] = value
+proof = m.snapshot("a"*40, listing, lambda item, commit: content[item["path"]])
+assert proof["checkpointCount"] == 29 and proof["blockedApprovalCheckpointCount"] == 0
+assert proof["gap"] == "NO_ACTUAL_BLOCKED_APPROVAL"
+assert proof["r10_02Accepted"] is False and proof["writerUsed"] is False
+deny(lambda: m.snapshot("a"*40, listing[:-1], lambda item, commit: content[item["path"]]))
+bad_list = copy.deepcopy(listing)
+bad_list[0]["path"] = "forged/path.json"
+deny(lambda: m.snapshot("a"*40, bad_list, lambda item, commit: content[item["path"]]))
+print("R10_02_ORIGINAL_STATE_HOSTILE_POLICY_PASS")