railiance-platform/scripts/provision-policy-nexus-forgejo-source.py

#!/usr/bin/env python3
"""Attended first bootstrap for CCR-2026-0014 (silent on success)."""

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


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
DIAGNOSTIC_PATH = Path("/tmp/policy-nexus-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 classify_warden_failure(result: subprocess.CompletedProcess[bytes]) -> str:
    output = (result.stdout or b"") + b"\n" + (result.stderr or b"")
    text = output.decode("utf-8", errors="replace").lower()
    checks = (
        ("routing catalog", "routing-catalog"),
        ("caller auth", "caller-auth"),
        ("permission denied", "owner-permission"),
        ("fetch failed", "owner-fetch"),
        ("policy denied", "policy-denied"),
        ("founder_required", "founder-required"),
        ("not valid", "invalid-invocation"),
        ("requires --", "missing-argument"),
    )
    for needle, classification in checks:
        if needle in text:
            return classification
    return "unclassified"


def sanitize_warden_failure(result: subprocess.CompletedProcess[bytes]) -> str:
    output = (result.stdout or b"") + b"\n" + (result.stderr or b"")
    text = output.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,
    extra_headers: dict[str, str] | None = None,
    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 {}),
            **(extra_headers or {}),
        },
    )
    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 = ""
    service_password = ""
    basic_authorization = ""
    bao_written = False
    activated = 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",
            },
            expected=(200,),
        )
        basic = base64.b64encode(f"{USER}:{service_password}".encode()).decode()
        basic_authorization = f"Basic {basic}"
        basic = ""
        _, 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")
        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(
                    "",
                    "DELETE",
                    f"/users/{USER}/tokens/{token_id}",
                    expected=(204, 404),
                    authorization=basic_authorization,
                )
            except Exception:
                pass
        if bao_written and not activated:
            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(
            f"forgejo-admin-route:{classify_warden_failure(result)}",
            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:
        raise ProvisionError("Warden did not provide API_TOKEN")
    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:
        # The surrounding Warden lane records only the failure and never receives
        # provider response bodies or credential material.
        raise SystemExit(1) from None