diff --git a/scripts/finalize_evidence.py b/scripts/finalize_evidence.py new file mode 100755 index 0000000..ad2e76d --- /dev/null +++ b/scripts/finalize_evidence.py @@ -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 `. + +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 + + +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()) \ No newline at end of file