From 1b85a3ef3d9a7f0b85fdc8af419c135a91477d99 Mon Sep 17 00:00:00 2001 From: codex Date: Mon, 31 Aug 2026 23:24:05 +0200 Subject: [PATCH] Add attended Policy Nexus source bootstrap Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a058f3-8ba0-7692-a042-9a870fc3d663 --- ...0014-policy-nexus-forgejo-source-read.yaml | 1 + .../openbao-bootstrap-policy-nexus-source.sh | 9 + .../provision-policy-nexus-forgejo-source.py | 375 ++++++++++++++++++ ...t_policy_nexus_forgejo_source_provision.py | 85 ++++ 4 files changed, 470 insertions(+) create mode 100755 scripts/openbao-bootstrap-policy-nexus-source.sh create mode 100755 scripts/provision-policy-nexus-forgejo-source.py create mode 100644 tests/test_policy_nexus_forgejo_source_provision.py diff --git a/credential-change-requests/CCR-2026-0014-policy-nexus-forgejo-source-read.yaml b/credential-change-requests/CCR-2026-0014-policy-nexus-forgejo-source-read.yaml index 98d0d92..db22f73 100644 --- a/credential-change-requests/CCR-2026-0014-policy-nexus-forgejo-source-read.yaml +++ b/credential-change-requests/CCR-2026-0014-policy-nexus-forgejo-source-read.yaml @@ -79,6 +79,7 @@ access_frontdoor: resolvable: false delivery: surface: forgejo-actions-secret + bootstrap_command: warden access openbao-platform-admin-login --exec -- scripts/openbao-bootstrap-policy-nexus-source.sh target: Repository Actions secret FORGEJO_SOURCE_TOKEN on coulomb/policy-nexus. Delivery is attended and must not expose the value in command output, process arguments, Git, State Hub, or workflow logs. diff --git a/scripts/openbao-bootstrap-policy-nexus-source.sh b/scripts/openbao-bootstrap-policy-nexus-source.sh new file mode 100755 index 0000000..744904c --- /dev/null +++ b/scripts/openbao-bootstrap-policy-nexus-source.sh @@ -0,0 +1,9 @@ +#!/usr/bin/env bash +# Exact silent owner command for the CCR-2026-0014 attended bootstrap. +set -euo pipefail + +REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "${REPO_DIR}" + +scripts/openbao-apply-policy-nexus-source-ccr.sh +python3 scripts/provision-policy-nexus-forgejo-source.py >/dev/null 2>&1 diff --git a/scripts/provision-policy-nexus-forgejo-source.py b/scripts/provision-policy-nexus-forgejo-source.py new file mode 100755 index 0000000..3ca5f13 --- /dev/null +++ b/scripts/provision-policy-nexus-forgejo-source.py @@ -0,0 +1,375 @@ +#!/usr/bin/env python3 +"""Attended first bootstrap for CCR-2026-0014 (silent on success).""" + +from __future__ import annotations + +import argparse +import datetime as dt +import json +import os +from pathlib import Path +import secrets +import subprocess +import sys +import urllib.error +import urllib.parse +import urllib.request + + +API_ROOT = "https://forgejo.coulomb.social/api/v1" +BAO_PATH = "platform/workloads/policy-nexus/forgejo-source-read" +ORG = "coulomb" +REPO = "policy-nexus" +USER = "policy-nexus-source" +TEAM = "policy-nexus-source-readers" +SECRET = "FORGEJO_SOURCE_TOKEN" +TOKEN_PREFIX = "policy-nexus-source-read-" +SCOPES = ["read:repository"] +REPO_DIR = Path(__file__).resolve().parent.parent +NON_CODE_UNITS = ( + "repo.actions", + "repo.packages", + "repo.issues", + "repo.ext_issues", + "repo.wiki", + "repo.pulls", + "repo.releases", + "repo.projects", + "repo.ext_wiki", +) + + +class ProvisionError(RuntimeError): + pass + + +def api_request( + token: str, + method: str, + path: str, + *, + payload: dict[str, object] | None = None, + expected: tuple[int, ...] = (200,), + read_body: bool = True, +) -> tuple[int, object | None]: + data = None if payload is None else json.dumps(payload).encode() + request = urllib.request.Request( + API_ROOT + path, + data=data, + method=method, + headers={ + "Authorization": f"token {token}", + "Accept": "application/json", + **({"Content-Type": "application/json"} if data is not None else {}), + }, + ) + try: + with urllib.request.urlopen(request, timeout=30) as response: + status = response.status + body = response.read() if read_body else response.read(1) + except urllib.error.HTTPError as error: + status = error.code + body = b"" + if status not in expected: + raise ProvisionError(f"{method} {path} returned HTTP {status}") + if not read_body or not body: + return status, None + try: + return status, json.loads(body) + except json.JSONDecodeError as error: + raise ProvisionError(f"{method} {path} returned invalid JSON") from error + + +def bao(args: list[str], *, stdin: bytes | None = None) -> subprocess.CompletedProcess[bytes]: + return subprocess.run( + ["bao", *args], + cwd=REPO_DIR, + input=stdin, + capture_output=True, + check=False, + timeout=60, + ) + + +def require_empty_destinations(admin_token: str, existing_tokens: list[dict[str, object]]) -> None: + if any(str(item.get("name", "")).startswith(TOKEN_PREFIX) for item in existing_tokens): + raise ProvisionError("a Policy Nexus source PAT already exists; use the rotation procedure") + _, action_secrets = api_request( + admin_token, + "GET", + f"/repos/{ORG}/{REPO}/actions/secrets?limit=100", + ) + if not isinstance(action_secrets, list): + raise ProvisionError("Forgejo returned an invalid Actions-secret list") + if any(item.get("name") == SECRET for item in action_secrets): + raise ProvisionError("FORGEJO_SOURCE_TOKEN already exists; use the rotation procedure") + if bao(["kv", "get", "-field=FORGEJO_SOURCE_TOKEN", BAO_PATH]).returncode == 0: + raise ProvisionError("the OpenBao destination already contains a source token") + + +def ensure_user(admin_token: str) -> None: + status, user = api_request(admin_token, "GET", f"/users/{USER}", expected=(200, 404)) + if status == 404: + _, user = api_request( + admin_token, + "POST", + "/admin/users", + payload={ + "username": USER, + "email": "policy-nexus-source@coulomb.social", + "full_name": "Policy Nexus source reader", + "password": secrets.token_urlsafe(48), + "restricted": True, + "must_change_password": False, + "send_notify": False, + "visibility": "private", + }, + expected=(201,), + ) + if not isinstance(user, dict): + raise ProvisionError("Forgejo returned an invalid service user") + if user.get("login") != USER or user.get("restricted") is not True or user.get("is_admin") is True: + raise ProvisionError("service user does not satisfy the restricted non-admin contract") + + +def ensure_team(admin_token: str) -> int: + _, teams = api_request(admin_token, "GET", f"/orgs/{ORG}/teams?limit=100") + if not isinstance(teams, list): + raise ProvisionError("Forgejo returned an invalid team list") + team = next((item for item in teams if item.get("name") == TEAM), None) + if team is None: + units_map = {unit: "none" for unit in NON_CODE_UNITS} + units_map["repo.code"] = "read" + _, team = api_request( + admin_token, + "POST", + f"/orgs/{ORG}/teams", + payload={ + "name": TEAM, + "description": "Read-only source acquisition for Policy Nexus publication", + "permission": "read", + "includes_all_repositories": True, + "can_create_org_repo": False, + "units": ["repo.code"], + "units_map": units_map, + }, + expected=(201,), + ) + if not isinstance(team, dict) or not isinstance(team.get("id"), int): + raise ProvisionError("Forgejo returned an invalid source-reader team") + units_map = team.get("units_map") or {} + if ( + team.get("permission") != "read" + or team.get("includes_all_repositories") is not True + or team.get("can_create_org_repo") is not False + or units_map.get("repo.code") != "read" + or any(units_map.get(unit, "none") != "none" for unit in NON_CODE_UNITS) + ): + raise ProvisionError("source-reader team does not satisfy the least-privilege contract") + team_id = int(team["id"]) + api_request( + admin_token, + "PUT", + f"/teams/{team_id}/members/{USER}", + expected=(204,), + ) + return team_id + + +def verify_source_token(source_token: str, scopes: object) -> None: + if scopes != SCOPES: + raise ProvisionError("Forgejo returned scopes broader than read:repository") + _, branch = api_request(source_token, "GET", f"/repos/{ORG}/{REPO}/branches/main") + if not isinstance(branch, dict) or not isinstance(branch.get("commit"), dict): + raise ProvisionError("source token could not resolve the main branch") + revision = branch["commit"].get("id") + if not isinstance(revision, str) or len(revision) != 40: + raise ProvisionError("source token returned an invalid main revision") + api_request( + source_token, + "GET", + f"/repos/{ORG}/{REPO}/archive/{urllib.parse.quote(revision)}.tar.gz", + expected=(200,), + read_body=False, + ) + api_request(source_token, "GET", "/admin/users", expected=(403, 404)) + api_request( + source_token, + "GET", + f"/repos/{ORG}/{REPO}/actions/secrets?limit=1", + expected=(403, 404), + ) + + +def verify_effective_permission(admin_token: str) -> None: + _, permission = api_request( + admin_token, + "GET", + f"/repos/{ORG}/{REPO}/collaborators/{USER}/permission", + ) + if not isinstance(permission, dict) or permission.get("permission") != "read": + raise ProvisionError("service user effective repository permission is not read-only") + + +def record_evidence(run_id: int | None, token_name: str) -> None: + detail = ( + f"Restricted user {USER} and all-repository repo.code-read team {TEAM} verified; " + f"PAT {token_name} reports exactly read:repository; archive read passed; effective repository " + f"permission is read-only; Actions-secret read and instance-admin probes were denied; OpenBao and repository " + f"Actions secret were populated without value output; workflow dispatch run_id={run_id}." + ) + result = subprocess.run( + [ + sys.executable, + "scripts/credential-change.py", + "record-evidence", + "CCR-2026-0014", + "--actor", + "attended operator via governed platform-admin lane", + "--kind", + "forgejo_source_bootstrap", + "--result", + "passed", + "--detail", + detail, + "--status", + "applied", + "--record-state-hub", + ], + cwd=REPO_DIR, + capture_output=True, + check=False, + timeout=120, + ) + if result.returncode != 0: + raise ProvisionError("credential evidence recording failed") + + +def provision(admin_token: str) -> None: + ensure_user(admin_token) + ensure_team(admin_token) + verify_effective_permission(admin_token) + _, existing_tokens = api_request(admin_token, "GET", f"/users/{USER}/tokens?limit=100") + if not isinstance(existing_tokens, list): + raise ProvisionError("Forgejo returned an invalid token list") + require_empty_destinations(admin_token, existing_tokens) + + timestamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ") + token_name = TOKEN_PREFIX + timestamp + token_id: int | None = None + source_token = "" + bao_written = False + activated = False + try: + _, created = api_request( + admin_token, + "POST", + f"/users/{USER}/tokens", + payload={"name": token_name, "scopes": SCOPES}, + expected=(201,), + ) + if not isinstance(created, dict) or not isinstance(created.get("id"), int): + raise ProvisionError("Forgejo returned an invalid access-token record") + token_id = int(created["id"]) + source_token = str(created.get("sha1") or "") + if not source_token: + raise ProvisionError("Forgejo did not return the new access token") + verify_source_token(source_token, created.get("scopes")) + + stored = bao( + [ + "kv", + "put", + BAO_PATH, + "FORGEJO_SOURCE_TOKEN=-", + f"API_USER={USER}", + "API_BASE_URL=https://forgejo.coulomb.social", + "TOKEN_SCOPES=read:repository", + f"GENERATED_AT={timestamp}", + ], + stdin=source_token.encode(), + ) + if stored.returncode != 0: + raise ProvisionError("OpenBao source-token write failed") + bao_written = True + readback = bao(["kv", "get", "-field=FORGEJO_SOURCE_TOKEN", BAO_PATH]) + if readback.returncode != 0 or readback.stdout.strip() != source_token.encode(): + raise ProvisionError("OpenBao source-token readback mismatch") + + api_request( + admin_token, + "PUT", + f"/repos/{ORG}/{REPO}/actions/secrets/{SECRET}", + payload={"data": source_token}, + expected=(201, 204), + ) + activated = True + _, dispatch = api_request( + admin_token, + "POST", + f"/repos/{ORG}/{REPO}/actions/workflows/publish-image.yaml/dispatches", + payload={"ref": "main", "return_run_info": True}, + expected=(201, 204), + ) + run_id = dispatch.get("id") if isinstance(dispatch, dict) else None + record_evidence(run_id if isinstance(run_id, int) else None, token_name) + except Exception: + if not activated and token_id is not None: + try: + api_request( + admin_token, + "DELETE", + f"/users/{USER}/tokens/{token_id}", + expected=(204, 404), + ) + except Exception: + pass + if bao_written and not activated: + bao(["kv", "delete", BAO_PATH]) + raise + finally: + source_token = "" + + +def outer() -> int: + result = subprocess.run( + [ + "warden", + "access", + "forgejo-admin-api-token", + "--field", + "API_TOKEN", + "--exec", + "--", + sys.executable, + str(Path(__file__).resolve()), + "--inner", + ], + cwd=REPO_DIR, + capture_output=True, + check=False, + timeout=900, + ) + return result.returncode + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--inner", action="store_true", help=argparse.SUPPRESS) + args = parser.parse_args() + if not args.inner: + return outer() + admin_token = os.environ.get("API_TOKEN", "") + if not admin_token: + raise ProvisionError("Warden did not provide API_TOKEN") + provision(admin_token) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception: + # The surrounding Warden lane records only the failure and never receives + # provider response bodies or credential material. + raise SystemExit(1) from None diff --git a/tests/test_policy_nexus_forgejo_source_provision.py b/tests/test_policy_nexus_forgejo_source_provision.py new file mode 100644 index 0000000..e731a8f --- /dev/null +++ b/tests/test_policy_nexus_forgejo_source_provision.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path +import subprocess +import unittest +from unittest import mock + + +SCRIPT = ( + Path(__file__).resolve().parents[1] + / "scripts" + / "provision-policy-nexus-forgejo-source.py" +) +SPEC = importlib.util.spec_from_file_location("policy_nexus_source_provision", SCRIPT) +assert SPEC is not None and SPEC.loader is not None +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) + + +class PolicyNexusForgejoSourceProvisionTests(unittest.TestCase): + def test_contract_is_exactly_read_repository_and_code_only(self) -> None: + self.assertEqual(MODULE.SCOPES, ["read:repository"]) + self.assertNotIn("repo.code", MODULE.NON_CODE_UNITS) + self.assertIn("repo.actions", MODULE.NON_CODE_UNITS) + self.assertIn("repo.packages", MODULE.NON_CODE_UNITS) + + def test_broader_returned_scope_is_rejected_before_network_checks(self) -> None: + with mock.patch.object(MODULE, "api_request") as request: + with self.assertRaises(MODULE.ProvisionError): + MODULE.verify_source_token("candidate", ["read:repository", "write:repository"]) + request.assert_not_called() + + def test_actions_failure_revokes_pat_and_removes_new_kv_value(self) -> None: + calls: list[tuple[str, str]] = [] + + def fake_api(token, method, path, **kwargs): + calls.append((method, path)) + if path == "/users/policy-nexus-source/tokens?limit=100": + return 200, [] + if path.endswith("/actions/secrets?limit=100"): + return 200, [] + if method == "POST" and path == "/users/policy-nexus-source/tokens": + return 201, { + "id": 73, + "name": "candidate", + "sha1": "source-token-value", + "scopes": ["read:repository"], + } + if method == "PUT" and path.endswith("/actions/secrets/FORGEJO_SOURCE_TOKEN"): + raise MODULE.ProvisionError("injected Actions failure") + if method == "DELETE" and path == "/users/policy-nexus-source/tokens/73": + return 204, None + raise AssertionError((method, path, kwargs)) + + bao_calls: list[tuple[list[str], bytes | None]] = [] + + def fake_bao(args, *, stdin=None): + bao_calls.append((args, stdin)) + if args[:3] == ["kv", "get", "-field=FORGEJO_SOURCE_TOKEN"]: + if stdin is None and len(bao_calls) == 1: + return subprocess.CompletedProcess(args, 2, b"", b"") + return subprocess.CompletedProcess(args, 0, b"source-token-value\n", b"") + return subprocess.CompletedProcess(args, 0, b"", b"") + + with ( + mock.patch.object(MODULE, "ensure_user"), + mock.patch.object(MODULE, "ensure_team", return_value=7), + mock.patch.object(MODULE, "verify_effective_permission"), + mock.patch.object(MODULE, "verify_source_token"), + mock.patch.object(MODULE, "api_request", side_effect=fake_api), + mock.patch.object(MODULE, "bao", side_effect=fake_bao), + ): + with self.assertRaises(MODULE.ProvisionError): + MODULE.provision("admin-token") + + self.assertIn(("DELETE", "/users/policy-nexus-source/tokens/73"), calls) + self.assertTrue(any(args == ["kv", "delete", MODULE.BAO_PATH] for args, _ in bao_calls)) + put = next((args, stdin) for args, stdin in bao_calls if args[:2] == ["kv", "put"]) + self.assertEqual(put[1], b"source-token-value") + self.assertNotIn("source-token-value", " ".join(put[0])) + + +if __name__ == "__main__": + unittest.main()