Jon Chery
7a93a92c68
---ci---
phase: 4
milestone: v1.0
status: execute
persona: lead-developer
task: T-4.4
requirements:
covered: [REQ-10, REQ-12]
review:
p0: 1
p1: 0
p0_items:
- id: P0-001
location: scripts/finalize_evidence.py _request()
issue: urllib.error.URLError (connection refused, DNS, timeout) was uncaught and stack-traced instead of returning clean JSON
severity: P0 (the pipeline would crash on a Gitea outage instead of writing a clean failure event)
fix: catch URLError in _request(); return (0, 'URLError: <reason>') so callers report cleanly
---/ci---
Wave 2, task T-4.4. scripts/verify_phase04.sh checks: (1) typecheck; (2)
pipeline.yml structure (3 inputs + 4 jobs + correct if: conditions +
finalize.needs=prod-gate); (3) pipeline.yml references all 5 core scripts
+ branch-pin doc; (4) issue-to-contract.yml structure (issues[opened] +
parse-and-trigger); (5) issue-to-contract.yml references (l3b_agent_stub
+ dispatch endpoint + contract-ref + issue number + GITEA_TOKEN +
new_branch); (6) finalize_evidence.py --help + missing file + missing
token (all exit 1 clean, no stack trace); (7) finalize_evidence.py
dead-host dry-run (exit 1, no stack trace). All 12 checks PASS.
The dead-host check caught a P0: finalize_evidence.py did not catch
urllib.error.URLError, so a connection-refused would stack-trace in the
pipeline. Fixed by catching URLError in _request() and returning
(0, 'URLError: <reason>') so callers report a clean JSON failure.
182 lines
7.7 KiB
Python
Executable File
182 lines
7.7 KiB
Python
Executable File
#!/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
|
|
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()) |