#!/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