#!/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 `. Input (argv flags): --audit (required) local audit.json file to upload --owner (optional, default continuous-intelligence) --repo (optional, default acdl-evidence) --branch (optional, default main) --path (optional, default audit.json) path in the repo --token-env (optional, default ACDL_GITEA_TOKEN) --host (optional, default https://git.cloudinit.dev) --message (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/?ref= 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": "", "sha": ""} to stdout and exit 0. 6. On any HTTP error: print {"uploaded": false, "status": , "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 except urllib.error.URLError as exc: # Network-level failure (connection refused, DNS, timeout). Return # a synthetic 0 status + the reason so callers can report cleanly # without a stack trace. return 0, f"URLError: {exc.reason}" 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())