f15ff61362
A read:repository-only PAT gets HTTP 403 on GET /orgs/coulomb/repos (needs read:organization). The 2026-10-03 attended run failed there and rolled back (PAT deleted, nothing written to OpenBao). The source token is still verified against the probe exactly as before: branch, archive, /admin/users and Actions secrets denials. Also records the delegated metadata apply evidence. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 3775@bnt-lap001 Assistant-Session: cfacb6a5-ad5b-4f79-ac67-053e3b0d34a3
389 lines
15 KiB
Python
Executable file
389 lines
15 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Attended, idempotent bootstrap for CCR-2026-0031 (silent on success).
|
|
|
|
Creates or verifies the restricted Forgejo service user ``core-hub-source`` and
|
|
its all-repository ``repo.code`` read team in org ``coulomb``, mints one PAT with
|
|
scope exactly ``read:repository``, verifies it, and writes it to OpenBao from
|
|
stdin. The value never reaches argv, stdout, stderr, Git, State Hub or files.
|
|
|
|
Idempotency: if a ``core-hub-source-read-*`` PAT and the OpenBao field both
|
|
exist, the run is a verified no-op. A PAT without the OpenBao value (or the
|
|
reverse) is drift and fails closed; use the CCR rotation procedure.
|
|
|
|
The Forgejo admin token reaches the inner process only through
|
|
``warden access forgejo-admin-api-token --exec``; OpenBao authority is the
|
|
ambient platform-admin session of the surrounding governed lane.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
import subprocess
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
CCR_ID = "CCR-2026-0031"
|
|
API_BASE_URL = "https://forgejo.coulomb.social"
|
|
API_ROOT = API_BASE_URL + "/api/v1"
|
|
BAO_PATH = "platform/workloads/core-hub/forgejo-source-read"
|
|
BAO_FIELD = "FORGEJO_SOURCE_TOKEN"
|
|
ORG = "coulomb"
|
|
USER = "core-hub-source"
|
|
TEAM = "core-hub-source-readers"
|
|
TOKEN_PREFIX = "core-hub-source-read-"
|
|
SCOPES = ["read:repository"]
|
|
ACTOR = "attended operator via governed platform-admin lane"
|
|
REPO_DIR = Path(__file__).resolve().parent.parent
|
|
DIAGNOSTIC_PATH = Path("/tmp/core-hub-source-bootstrap-diagnostic.json")
|
|
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 write_diagnostic(stage: str, error: Exception | None = None) -> None:
|
|
detail = str(error) if isinstance(error, ProvisionError) else "unexpected internal error"
|
|
DIAGNOSTIC_PATH.write_text(json.dumps({"stage": stage, "detail": detail}, sort_keys=True) + "\n")
|
|
DIAGNOSTIC_PATH.chmod(0o600)
|
|
|
|
|
|
def sanitize_warden_failure(result: subprocess.CompletedProcess[bytes]) -> str:
|
|
text = ((result.stdout or b"") + b"\n" + (result.stderr or b"")).decode("utf-8", errors="replace")
|
|
text = re.sub(r"[A-Za-z0-9_=.:-]{20,}", "<redacted>", text)
|
|
lines = [line.strip() for line in text.splitlines() if line.strip()]
|
|
return " | ".join(lines[-8:])[:600] or f"nested Warden exit {result.returncode}"
|
|
|
|
|
|
def api_request(
|
|
token: str,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
payload: dict[str, object] | None = None,
|
|
expected: tuple[int, ...] = (200,),
|
|
read_body: bool = True,
|
|
authorization: str | None = None,
|
|
) -> 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": authorization or 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.split('?')[0]} 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.split('?')[0]} 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 bao_has_token() -> bool:
|
|
return bao(["kv", "get", f"-field={BAO_FIELD}", BAO_PATH]).returncode == 0
|
|
|
|
|
|
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": f"{USER}@coulomb.social",
|
|
"full_name": "Core Hub repo-manager 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 team_satisfies_contract(team: dict[str, object]) -> bool:
|
|
units_map = team.get("units_map") or {}
|
|
if not isinstance(units_map, dict):
|
|
return False
|
|
return (
|
|
team.get("permission") == "read"
|
|
and team.get("includes_all_repositories") is True
|
|
and team.get("can_create_org_repo") is False
|
|
and units_map.get("repo.code") == "read"
|
|
and all(units_map.get(unit, "none") == "none" for unit in NON_CODE_UNITS)
|
|
)
|
|
|
|
|
|
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": f"Read-only source acquisition for the Core Hub repo-manager publisher ({CCR_ID})",
|
|
"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")
|
|
if not team_satisfies_contract(team):
|
|
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 find_private_probe(admin_token: str) -> str:
|
|
"""Pick a private org repository to probe.
|
|
|
|
Listing org repositories needs read:organization, which the least-privilege
|
|
source token deliberately lacks (Forgejo returns 403), so the admin token
|
|
chooses the probe and the source token is then tested against it.
|
|
"""
|
|
for page in range(1, 6):
|
|
_, repos = api_request(admin_token, "GET", f"/orgs/{ORG}/repos?limit=50&page={page}")
|
|
if not isinstance(repos, list):
|
|
raise ProvisionError("Forgejo returned an invalid repository list")
|
|
for repo in repos:
|
|
if isinstance(repo, dict) and repo.get("private") is True and isinstance(repo.get("name"), str):
|
|
return str(repo["name"])
|
|
if len(repos) < 50:
|
|
break
|
|
raise ProvisionError("no private repository found in the org to probe")
|
|
|
|
|
|
def verify_source_token(source_token: str, scopes: object, probe: str) -> str:
|
|
if scopes != SCOPES:
|
|
raise ProvisionError("Forgejo returned scopes other than exactly read:repository")
|
|
_, repo = api_request(source_token, "GET", f"/repos/{ORG}/{probe}")
|
|
branch_name = repo.get("default_branch") if isinstance(repo, dict) else None
|
|
if not isinstance(branch_name, str) or not branch_name:
|
|
raise ProvisionError("source token could not resolve the probe default branch")
|
|
_, branch = api_request(
|
|
source_token, "GET", f"/repos/{ORG}/{probe}/branches/{urllib.parse.quote(branch_name)}"
|
|
)
|
|
commit = branch.get("commit") if isinstance(branch, dict) else None
|
|
revision = commit.get("id") if isinstance(commit, dict) else None
|
|
if not isinstance(revision, str) or len(revision) != 40:
|
|
raise ProvisionError("source token returned an invalid probe revision")
|
|
api_request(
|
|
source_token,
|
|
"GET",
|
|
f"/repos/{ORG}/{probe}/archive/{urllib.parse.quote(revision)}.tar.gz",
|
|
read_body=False,
|
|
)
|
|
api_request(source_token, "GET", "/admin/users", expected=(403, 404))
|
|
api_request(source_token, "GET", f"/repos/{ORG}/{probe}/actions/secrets?limit=1", expected=(403, 404))
|
|
return probe
|
|
|
|
|
|
def verify_effective_permission(admin_token: str, repo: 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(token_name: str, probe: str) -> None:
|
|
detail = (
|
|
f"Restricted user {USER} and all-repository repo.code-read team {TEAM} verified; PAT {token_name} "
|
|
f"reports exactly read:repository; private probe {ORG}/{probe} branch and archive read passed; "
|
|
"effective permission read-only; Actions-secret and instance-admin probes denied; OpenBao field "
|
|
"written from stdin and read back without value output."
|
|
)
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable, "scripts/credential-change.py", "record-evidence", CCR_ID,
|
|
"--actor", ACTOR, "--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) -> str:
|
|
"""Return "noop" when already provisioned, "minted" after a fresh mint."""
|
|
ensure_user(admin_token)
|
|
ensure_team(admin_token)
|
|
_, existing = api_request(admin_token, "GET", f"/users/{USER}/tokens?limit=100")
|
|
if not isinstance(existing, list):
|
|
raise ProvisionError("Forgejo returned an invalid token list")
|
|
has_pat = any(str(item.get("name", "")).startswith(TOKEN_PREFIX) for item in existing)
|
|
has_bao = bao_has_token()
|
|
if has_pat and has_bao:
|
|
return "noop"
|
|
if has_pat or has_bao:
|
|
raise ProvisionError("PAT and OpenBao value disagree; use the CCR rotation procedure")
|
|
|
|
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 = ""
|
|
service_password = ""
|
|
basic_authorization = ""
|
|
bao_written = False
|
|
committed = False
|
|
try:
|
|
service_password = secrets.token_urlsafe(48)
|
|
api_request(
|
|
admin_token,
|
|
"PATCH",
|
|
f"/admin/users/{USER}",
|
|
payload={
|
|
"password": service_password, "must_change_password": False, "restricted": True,
|
|
"admin": False, "allow_create_organization": False, "allow_git_hook": False,
|
|
"allow_import_local": False, "max_repo_creation": 0, "prohibit_login": False,
|
|
"active": True, "visibility": "private",
|
|
},
|
|
)
|
|
basic_authorization = "Basic " + base64.b64encode(f"{USER}:{service_password}".encode()).decode()
|
|
_, created = api_request(
|
|
"", "POST", f"/users/{USER}/tokens",
|
|
payload={"name": token_name, "scopes": SCOPES},
|
|
expected=(201,), authorization=basic_authorization,
|
|
)
|
|
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")
|
|
probe = verify_source_token(source_token, created.get("scopes"), find_private_probe(admin_token))
|
|
verify_effective_permission(admin_token, probe)
|
|
|
|
stored = bao(
|
|
[
|
|
"kv", "put", BAO_PATH, f"{BAO_FIELD}=-", f"API_USER={USER}",
|
|
f"API_BASE_URL={API_BASE_URL}", "TOKEN_SCOPES=read:repository",
|
|
f"GENERATED_AT={timestamp}",
|
|
],
|
|
stdin=source_token.encode(),
|
|
)
|
|
bao_written = True
|
|
if stored.returncode != 0:
|
|
raise ProvisionError("OpenBao source-token write failed")
|
|
readback = bao(["kv", "get", f"-field={BAO_FIELD}", BAO_PATH])
|
|
if readback.returncode != 0 or readback.stdout.strip() != source_token.encode():
|
|
raise ProvisionError("OpenBao source-token readback mismatch")
|
|
committed = True
|
|
record_evidence(token_name, probe)
|
|
return "minted"
|
|
except Exception:
|
|
if not committed:
|
|
if token_id is not None:
|
|
try:
|
|
api_request(
|
|
"", "DELETE", f"/users/{USER}/tokens/{token_id}",
|
|
expected=(204, 404), authorization=basic_authorization,
|
|
)
|
|
except Exception:
|
|
pass
|
|
if bao_written:
|
|
bao(["kv", "delete", BAO_PATH])
|
|
raise
|
|
finally:
|
|
source_token = ""
|
|
service_password = ""
|
|
basic_authorization = ""
|
|
|
|
|
|
def outer() -> int:
|
|
DIAGNOSTIC_PATH.unlink(missing_ok=True)
|
|
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,
|
|
)
|
|
if result.returncode != 0 and not DIAGNOSTIC_PATH.exists():
|
|
write_diagnostic("forgejo-admin-route", ProvisionError(sanitize_warden_failure(result)))
|
|
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:
|
|
write_diagnostic("inner", ProvisionError("Warden did not provide API_TOKEN"))
|
|
return 1
|
|
try:
|
|
provision(admin_token)
|
|
except Exception as error:
|
|
write_diagnostic("provision", error)
|
|
raise
|
|
DIAGNOSTIC_PATH.unlink(missing_ok=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except Exception:
|
|
# Never surface provider bodies or credential material to the Warden lane.
|
|
raise SystemExit(1) from None
|