feat(P04): add finalize_evidence.py helper for D-028 state persistence (T-4.1)

---ci--- phase: 4 milestone: v1.0 status: execute persona: backend-engineer task: T-4.1 requirements: covered: [REQ-10] ---/ci--- Wave 1, task T-4.1. Adds scripts/finalize_evidence.py, a stdlib-only helper that uploads a local audit.json to the acdl-evidence Gitea repo via the file-contents API. It GETs the existing file to discover its sha (update mode) and falls back to create mode on 404, then PUTs/POSTs the base64-encoded content with the Authorization: token <token> header. This is the per-stage state-persistence mechanism (D-028) used by the pipeline jobs between re-dispatches, and the finalize-step writer (D-029).
2026-07-21 13:37:51 +00:00
parent 88945fed71
commit 3bcebd8d14
1 changed files with 177 additions and 0 deletions
@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+"""finalize_evidence.py — REQ-10 / D-028 / D-029
+
+Uploads (PUT or POST) a local `audit.json` to the `acdl-evidence` repo on
+Gitea via the file-contents API. Used by the pipeline workflow steps to
+persist the hash-chained audit trail to `acdl-evidence` between dispatches
+(D-028 state-persistence across re-dispatches; D-029 finalize step).
+
+Uses only the Python standard library (urllib.request) so it has no
+external dependency on `requests`. Auth header: `Authorization: token <token>`.
+
+Input (argv flags):
+  --audit <path>           (required) local audit.json file to upload
+  --owner <org>            (optional, default continuous-intelligence)
+  --repo <name>            (optional, default acdl-evidence)
+  --branch <name>         (optional, default main)
+  --path <remote path>     (optional, default audit.json) path in the repo
+  --token-env <env var>    (optional, default ACDL_GITEA_TOKEN)
+  --host <url>             (optional, default https://git.cloudinit.dev)
+  --message <commit msg>   (optional, default chore(evidence): update audit.json)
+
+Behavior:
+  1. Read the token from os.environ[token_env]. Missing -> stderr + exit 1.
+  2. Read the local audit file; base64-encode it.
+  3. GET the current file at .../contents/<path>?ref=<branch> to discover
+     the existing `sha`. 200 -> capture sha (update mode). 404 -> no sha
+     (create mode). Other errors -> exit 1.
+  4. If sha set: PUT with body {content, message, branch, sha}.
+     If no sha: POST with body {content, message, branch}.
+  5. Print {"uploaded": true, "path": "<path>", "sha": "<new sha>"} to
+     stdout and exit 0.
+  6. On any HTTP error: print
+     {"uploaded": false, "status": <code>, "body": "<body>"} to stdout
+     and exit 1.
+"""
+import argparse
+import base64
+import json
+import os
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+def _request(method: str, url: str, token: str, body: dict = None):
+    """Perform an HTTP request with the Gitea auth header. Returns
+    (status_code, response_body_text). Raises URLError on network failure."""
+    data = None
+    headers = {"Authorization": f"token {token}",
+               "Accept": "application/json"}
+    if body is not None:
+        data = json.dumps(body).encode("utf-8")
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, method=method, headers=headers)
+    try:
+        with urllib.request.urlopen(req) as resp:
+            return resp.getcode(), resp.read().decode("utf-8", "replace")
+    except urllib.error.HTTPError as exc:
+        # HTTPError carries the response body
+        try:
+            body_text = exc.read().decode("utf-8", "replace")
+        except Exception:
+            body_text = ""
+        return exc.code, body_text
+
+
+def get_existing_sha(host: str, owner: str, repo: str, path: str,
+                     branch: str, token: str):
+    """Return (sha-or-None, error_status_or_None). On 200 returns the sha.
+    On 404 returns (None, None). Other codes return (None, (status, body))."""
+    qs = urllib.parse.urlencode({"ref": branch})
+    url = f"{host}/api/v1/repos/{owner}/{repo}/contents/{path}?{qs}"
+    status, body = _request("GET", url, token)
+    if status == 200:
+        try:
+            data = json.loads(body)
+            return data.get("sha"), None
+        except (ValueError, TypeError):
+            return None, (status, body)
+    if status == 404:
+        return None, None
+    return None, (status, body)
+
+
+def upload(host: str, owner: str, repo: str, path: str, branch: str,
+           message: str, content_b64: str, sha, token: str):
+    """PUT (update) or POST (create) the file. Returns (new_sha, None) on
+    success or (None, (status, body)) on HTTP error."""
+    url = f"{host}/api/v1/repos/{owner}/{repo}/contents/{path}"
+    if sha:
+        body = {"content": content_b64, "message": message,
+                "branch": branch, "sha": sha}
+        status, resp = _request("PUT", url, token, body)
+    else:
+        body = {"content": content_b64, "message": message, "branch": branch}
+        status, resp = _request("POST", url, token, body)
+    if status in (200, 201):
+        try:
+            data = json.loads(resp)
+            # The file-contents API returns the new content object either at
+            # top-level `content` (POST create) or `content` (PUT update).
+            new_sha = None
+            if isinstance(data, dict):
+                content_obj = data.get("content") or data
+                if isinstance(content_obj, dict):
+                    new_sha = content_obj.get("sha")
+            return new_sha, None
+        except (ValueError, TypeError):
+            return None, None
+    return None, (status, resp)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="Upload a local audit.json to the acdl-evidence Gitea "
+                    "repo via the file-contents API (D-028/D-029).")
+    parser.add_argument("--audit", required=True,
+                        help="Local audit.json file to upload")
+    parser.add_argument("--owner", default="continuous-intelligence",
+                        help="Gitea org (default: continuous-intelligence)")
+    parser.add_argument("--repo", default="acdl-evidence",
+                        help="Gitea repo (default: acdl-evidence)")
+    parser.add_argument("--branch", default="main",
+                        help="Target branch (default: main)")
+    parser.add_argument("--path", default="audit.json",
+                        help="Remote path in the repo (default: audit.json)")
+    parser.add_argument("--token-env", default="ACDL_GITEA_TOKEN",
+                        help="Env var name holding the Gitea token "
+                             "(default: ACDL_GITEA_TOKEN)")
+    parser.add_argument("--host", default="https://git.cloudinit.dev",
+                        help="Gitea host URL (default: https://git.cloudinit.dev)")
+    parser.add_argument("--message", default="chore(evidence): update audit.json",
+                        help="Commit message (default: chore(evidence): "
+                             "update audit.json)")
+    args = parser.parse_args()
+
+    token = os.environ.get(args.token_env)
+    if not token:
+        print(f"finalize_evidence: required env var {args.token_env} is not "
+              f"set", file=sys.stderr)
+        return 1
+
+    # Read + base64-encode the local audit file. Missing/unreadable file is
+    # a clean exit 1 (no stack trace).
+    try:
+        with open(args.audit, "rb") as fh:
+            raw = fh.read()
+    except OSError as exc:
+        print(f"finalize_evidence: cannot read {args.audit}: {exc}",
+              file=sys.stderr)
+        return 1
+    content_b64 = base64.b64encode(raw).decode("ascii")
+
+    # Discover existing sha (update vs create).
+    sha, err = get_existing_sha(args.host, args.owner, args.repo,
+                                args.path, args.branch, token)
+    if err is not None:
+        status, body = err
+        print(json.dumps({"uploaded": False, "status": status, "body": body}))
+        return 1
+
+    # Upload (PUT if sha, POST otherwise).
+    new_sha, err = upload(args.host, args.owner, args.repo, args.path,
+                          args.branch, args.message, content_b64, sha, token)
+    if err is not None:
+        status, body = err
+        print(json.dumps({"uploaded": False, "status": status, "body": body}))
+        return 1
+
+    print(json.dumps({"uploaded": True, "path": args.path,
+                      "sha": new_sha}))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())