ship: phase-04 pipeline-and-approval-gates (v1.0.4)
Squash merge of phase/04-pipeline-and-approval-gates; pipeline.yml + issue-to-contract.yml + finalize_evidence.py; verify_phase04.sh green; 1 P0 fixed (shell injection).
This commit was merged in pull request #4.
This commit is contained in:
Executable
+182
@@ -0,0 +1,182 @@
|
||||
#!/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())
|
||||
Executable
+186
@@ -0,0 +1,186 @@
|
||||
#!/usr/bin/env bash
|
||||
# Phase 04 verification script.
|
||||
# Confirms the pipeline workflow + issue trigger + finalize_evidence.py
|
||||
# conform to the Phase 04 plan and the Gitea Actions topology in
|
||||
# ARCHITECTURE.md. Does NOT execute a real Gitea Actions run (act_runner
|
||||
# is not registered in this environment); validates structure + syntax
|
||||
# + a dry-run of finalize_evidence.py against a dead host.
|
||||
#
|
||||
# Usage: scripts/verify_phase04.sh
|
||||
# Exit codes: 0 = all checks passed; 1 = one or more checks failed.
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
cd "$ROOT"
|
||||
|
||||
fail_count=0
|
||||
pass() { printf ' [PASS] %s\n' "$1"; }
|
||||
fail() { printf ' [FAIL] %s\n' "$1"; fail_count=$((fail_count + 1)); }
|
||||
|
||||
echo "== Phase 04 verification =="
|
||||
echo "Root: ${ROOT}"
|
||||
echo
|
||||
|
||||
# --- Check 1: typecheck ---
|
||||
echo "-- Check 1: typecheck --"
|
||||
if bash -n scripts/finalize_evidence.py 2>/dev/null || python3 -m py_compile scripts/finalize_evidence.py 2>/dev/null; then
|
||||
pass "py_compile finalize_evidence.py"
|
||||
else
|
||||
fail "py_compile finalize_evidence.py"
|
||||
fi
|
||||
if python3 -c "import yaml; yaml.safe_load(open('.gitea/workflows/pipeline.yml')); yaml.safe_load(open('contracts-repo/.gitea/workflows/issue-to-contract.yml'))" 2>/dev/null; then
|
||||
pass "yaml load both workflows"
|
||||
else
|
||||
fail "yaml load workflows"
|
||||
fi
|
||||
|
||||
# --- Check 2: pipeline.yml structure ---
|
||||
echo "-- Check 2: pipeline.yml structure (D-027, D-028) --"
|
||||
p_struct=$(python3 << 'PYEOF'
|
||||
import yaml, sys
|
||||
try:
|
||||
d = yaml.safe_load(open('.gitea/workflows/pipeline.yml'))
|
||||
on = d.get('on', d.get(True)) or {}
|
||||
assert 'workflow_dispatch' in on, 'no workflow_dispatch trigger'
|
||||
inputs = on['workflow_dispatch']['inputs']
|
||||
assert set(inputs.keys()) == {'contract-ref', 'approve_qa', 'approve_prod'}, f'inputs: {set(inputs.keys())}'
|
||||
assert inputs['contract-ref']['type'] == 'string', 'contract-ref type'
|
||||
assert inputs['approve_qa']['type'] == 'boolean', 'approve_qa type'
|
||||
assert inputs['approve_prod']['type'] == 'boolean', 'approve_prod type'
|
||||
jobs = d['jobs']
|
||||
assert set(jobs.keys()) == {'dev', 'qa-gate', 'prod-gate', 'finalize'}, f'jobs: {set(jobs.keys())}'
|
||||
dev_if = jobs['dev'].get('if', '')
|
||||
assert 'approve_qa' in dev_if and 'approve_prod' in dev_if, f'dev.if: {dev_if}'
|
||||
qa_if = jobs['qa-gate'].get('if', '')
|
||||
assert 'approve_qa' in qa_if, f'qa-gate.if: {qa_if}'
|
||||
prod_if = jobs['prod-gate'].get('if', '')
|
||||
assert 'approve_prod' in prod_if, f'prod-gate.if: {prod_if}'
|
||||
fin_needs = jobs['finalize'].get('needs', [])
|
||||
assert fin_needs == ['prod-gate'] or fin_needs == 'prod-gate', f'finalize.needs: {fin_needs}'
|
||||
# All jobs runs-on ubuntu-latest
|
||||
for name, job in jobs.items():
|
||||
assert job.get('runs-on') == 'ubuntu-latest', f'{name} runs-on: {job.get("runs-on")}'
|
||||
print('OK')
|
||||
except AssertionError as ex:
|
||||
print(f'FAIL: {ex}')
|
||||
sys.exit(1)
|
||||
except Exception as ex:
|
||||
print(f'FAIL: {ex}')
|
||||
sys.exit(1)
|
||||
PYEOF
|
||||
)
|
||||
if [ "$p_struct" = "OK" ]; then
|
||||
pass "pipeline.yml: 3 inputs + 4 jobs + correct if: conditions + finalize.needs=prod-gate"
|
||||
else
|
||||
fail "pipeline.yml structure: $p_struct"
|
||||
fi
|
||||
|
||||
# --- Check 3: pipeline.yml references core scripts ---
|
||||
echo "-- Check 3: pipeline.yml references core scripts (D-029) --"
|
||||
text=$(cat .gitea/workflows/pipeline.yml)
|
||||
missing=""
|
||||
for ref in policy_checker.py confidence_signal.py mock_executor.sh evidence_writer.py finalize_evidence.py; do
|
||||
if ! echo "$text" | grep -qF "$ref"; then
|
||||
missing="$missing $ref"
|
||||
fi
|
||||
done
|
||||
if [ -z "$missing" ]; then
|
||||
pass "pipeline.yml references all 5 core scripts"
|
||||
else
|
||||
fail "pipeline.yml missing references:$missing"
|
||||
fi
|
||||
# Branch-pin documentation
|
||||
if echo "$text" | grep -q 'milestone/v1.0-initial'; then
|
||||
pass "pipeline.yml documents branch-pin to milestone/v1.0-initial"
|
||||
else
|
||||
fail "pipeline.yml missing branch-pin reference"
|
||||
fi
|
||||
|
||||
# --- Check 4: issue-to-contract.yml structure ---
|
||||
echo "-- Check 4: issue-to-contract.yml structure (D-030) --"
|
||||
i_struct=$(python3 << 'PYEOF'
|
||||
import yaml, sys
|
||||
try:
|
||||
d = yaml.safe_load(open('contracts-repo/.gitea/workflows/issue-to-contract.yml'))
|
||||
on = d.get('on', d.get(True)) or {}
|
||||
assert 'issues' in on, 'no issues trigger'
|
||||
assert on['issues']['types'] == ['opened'], f'types: {on["issues"]["types"]}'
|
||||
assert 'parse-and-trigger' in d['jobs'], 'no parse-and-trigger job'
|
||||
assert d['jobs']['parse-and-trigger'].get('runs-on') == 'ubuntu-latest', 'runs-on'
|
||||
print('OK')
|
||||
except AssertionError as ex:
|
||||
print(f'FAIL: {ex}')
|
||||
sys.exit(1)
|
||||
PYEOF
|
||||
)
|
||||
if [ "$i_struct" = "OK" ]; then
|
||||
pass "issue-to-contract.yml: issues[opened] + parse-and-trigger job"
|
||||
else
|
||||
fail "issue-to-contract.yml structure: $i_struct"
|
||||
fi
|
||||
|
||||
# --- Check 5: issue-to-contract.yml references + dispatch endpoint ---
|
||||
echo "-- Check 5: issue-to-contract.yml references + dispatch (D-014, D-030) --"
|
||||
text=$(cat contracts-repo/.gitea/workflows/issue-to-contract.yml)
|
||||
missing=""
|
||||
for ref in l3b_agent_stub.py 'actions/workflows/pipeline.yml/dispatches' 'contract-ref' 'gitea.event.issue.number' 'GITEA_TOKEN' 'new_branch'; do
|
||||
if ! echo "$text" | grep -qF "$ref"; then
|
||||
missing="$missing $ref"
|
||||
fi
|
||||
done
|
||||
if [ -z "$missing" ]; then
|
||||
pass "issue-to-contract.yml: l3b_agent_stub + dispatch + contract-ref + issue number + token + new_branch"
|
||||
else
|
||||
fail "issue-to-contract.yml missing references:$missing"
|
||||
fi
|
||||
|
||||
# --- Check 6: finalize_evidence.py --help + clean failure ---
|
||||
echo "-- Check 6: finalize_evidence.py CLI + clean failure modes ---"
|
||||
out=$(python3 scripts/finalize_evidence.py --help 2>&1); rc=$?
|
||||
if [ "$rc" = "0" ] && echo "$out" | grep -qi 'usage\|--audit\|--owner'; then
|
||||
pass "finalize_evidence.py --help exits 0 with usage"
|
||||
else
|
||||
fail "finalize_evidence.py --help: rc=$rc"
|
||||
fi
|
||||
|
||||
# Missing audit file (with a fake token so it gets past the env check) → exit 1, no stack trace
|
||||
out=$(ACDL_GITEA_TOKEN=fake python3 scripts/finalize_evidence.py --audit /tmp/definitely_nonexistent_audit.json 2>&1); rc=$?
|
||||
if [ "$rc" = "1" ] && ! echo "$out" | grep -q 'Traceback'; then
|
||||
pass "finalize_evidence.py missing file → exit 1, no stack trace"
|
||||
else
|
||||
fail "finalize_evidence.py missing file: rc=$rc, out='$out'"
|
||||
fi
|
||||
|
||||
# Missing token env (audit file present) → exit 1, no stack trace
|
||||
printf '[]\n' > /tmp/empty_audit.json
|
||||
out=$(env -u ACDL_GITEA_TOKEN python3 scripts/finalize_evidence.py --audit /tmp/empty_audit.json 2>&1); rc=$?
|
||||
if [ "$rc" = "1" ] && ! echo "$out" | grep -q 'Traceback'; then
|
||||
pass "finalize_evidence.py missing token env → exit 1, no stack trace"
|
||||
else
|
||||
fail "finalize_evidence.py missing token: rc=$rc, out='$out'"
|
||||
fi
|
||||
|
||||
# --- Check 7: finalize_evidence.py dry-run against a dead host (clean failure) ---
|
||||
echo "-- Check 7: finalize_evidence.py dry-run against dead host ---"
|
||||
# Use a real audit.json but point at a host that will refuse the connection.
|
||||
printf '[{"seq":0,"ts":"2026-07-21T00:00:00Z","stage":"genesis","event":"init","prev_hash":"GENESIS","hash":"x"}]\n' > /tmp/real_audit.json
|
||||
out=$(ACDL_GITEA_TOKEN=fake GITEA_HOST=http://127.0.0.1:0 python3 scripts/finalize_evidence.py --audit /tmp/real_audit.json --host http://127.0.0.1:0 2>&1); rc=$?
|
||||
if [ "$rc" = "1" ] && ! echo "$out" | grep -q 'Traceback'; then
|
||||
pass "finalize_evidence.py dead host → exit 1, no stack trace (clean API failure)"
|
||||
else
|
||||
fail "finalize_evidence.py dead host: rc=$rc, out='$out'"
|
||||
fi
|
||||
|
||||
# Cleanup
|
||||
rm -f /tmp/empty_audit.json /tmp/real_audit.json
|
||||
|
||||
echo
|
||||
echo "== Summary =="
|
||||
if [ "$fail_count" -eq 0 ]; then
|
||||
echo "Phase 04 verification PASSED (pipeline + issue trigger + finalize helper, all checks ok)"
|
||||
exit 0
|
||||
else
|
||||
echo "Phase 04 verification FAILED (${fail_count} check(s) failed)"
|
||||
exit 1
|
||||
fi
|
||||
Reference in New Issue
Block a user