CUST-WP-0075: dedicated core-hub Forgejo read identity CCR draft

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
2026-10-03 08:27:56 +02:00 · 2026-10-03 08:27:56 +02:00 · 401765f184
commit 401765f184
parent 6bb2f54b5f
7 changed files with 846 additions and 0 deletions
--- a/credential-change-requests/CCR-2026-0031-core-hub-forgejo-source-read.yaml
+++ b/credential-change-requests/CCR-2026-0031-core-hub-forgejo-source-read.yaml
@ -0,0 +1,143 @@
+id: CCR-2026-0031
+kind: credential-change-request
+schema_version: 1
+request_type: workload-kv-read
+title: Core Hub repo-manager publisher Forgejo private-source read token lane
+status: proposed
+created: '2026-10-03'
+updated: '2026-10-03'
+requester:
+  agent: claude (rapp-core-hub repo agent, dispatched by the-custodian under CUST-WP-0075)
+  reason: RAPPCOREHUB-WP-0003-T04. The repo-manager publisher in namespace core-hub
+    reads the 123-entry repository registry from Forgejo anonymously; nine private
+    sources return HTTP 404, so admission cannot reach 123 accepted / 0 errors.
+    The Forgejo admin PAT, any personal SSH key or PAT, and the policy-nexus-source
+    token (CCR-2026-0014) are prohibited for this workload. A dedicated identity
+    with read:repository only is required.
+review:
+  required: true
+  required_approvers:
+  - platform-operator
+  - core-hub-owner
+  comments: []
+target:
+  domain: infotech
+  tenant: core-hub
+  workload: core-hub-repo-manager-publisher
+  rapp: rapp-core-hub
+  environment: production
+  purpose: Hold a dedicated Forgejo PAT scoped exactly read:repository and project
+    it through External Secrets into namespace core-hub as the Secret that the
+    rapp-core-hub publisher mounts read-only at /var/run/secrets/repo-manager/forgejo/token.
+openbao:
+  mount: platform
+  kv_path: platform/workloads/core-hub/forgejo-source-read
+  fields:
+  - FORGEJO_SOURCE_TOKEN
+  - API_USER
+  - API_BASE_URL
+  - TOKEN_SCOPES
+  - GENERATED_AT
+  policy_name: workload-kv-read-core-hub-forgejo-source
+  policy_file: openbao/policies/workload-kv-read-core-hub-forgejo-source.hcl
+  auth:
+    method: kubernetes
+    mount: kubernetes
+    role: external-secrets-core-hub-forgejo-source
+    bound_claims:
+      service_account_names:
+      - external-secrets
+      service_account_namespaces:
+      - external-secrets
+    bound_claims_confirmed: false
+    policies:
+    - workload-kv-read-core-hub-forgejo-source
+    ttl: 15m
+access_frontdoor:
+  type: ops-warden
+  catalog_id: core-hub-forgejo-source-read
+  selector: core hub repo-manager publisher Forgejo private source repository read token
+  readiness: pending-review
+  resolvable: false
+delivery:
+  surface: external-secrets
+  bootstrap_command: warden access openbao-platform-admin-login --exec -- scripts/openbao-bootstrap-core-hub-source.sh
+  target: ClusterSecretStore openbao-core-hub-forgejo-source (conditions limited to
+    namespace core-hub) feeding ExternalSecret core-hub/repo-manager-forgejo-source,
+    which materializes Secret repo-manager-forgejo-source with the single key token
+    (remoteRef workloads/core-hub/forgejo-source-read property FORGEJO_SOURCE_TOKEN),
+    explicit empty target metadata template, creationPolicy Owner, deletionPolicy
+    Retain. rapp-core-hub then sets publisher.existingSecret to repo-manager-forgejo-source
+    with tokenKey token. Both manifests are drafted in this CCR only and land in
+    argocd/platform-addons/openbao-secretstore after approval and metadata apply.
+  forgejo_identity: core-hub-source
+  forgejo_team: core-hub-source-readers
+  forgejo_team_contract: Restricted, non-admin, private
+    service user (random unrecorded password) in org coulomb; team permission read, includes_all_repositories
+    true, can_create_org_repo false, repo.code read, every non-code unit none. PAT
+    name prefix core-hub-source-read-, scope exactly read:repository.
+  repositories: All repositories of Forgejo org coulomb (the registry's 123 entries
+    are all coulomb/*). Per-repository team membership was rejected because the
+    registry changes weekly and each addition would need a new attended session.
+  naming_basis: Follows the CCR-2026-0014 precedent (<workload>-source user, <workload>-source-readers
+    team). repo-manager publishes no service-user naming canon; its outstanding
+    identity-canon correction is a registry/Forgejo identity mismatch for one repository
+    and does not constrain this name.
+  provisioning_script: scripts/provision-core-hub-forgejo-source.py (silent, idempotent;
+    admin token reaches it only through warden access forgejo-admin-api-token --exec)
+risk:
+  classification: high
+  notes:
+  - Read-only at Forgejo, but disclosure grants estate-wide private source read
+    for org coulomb. No package, repository-write, organization-admin, user-write,
+    Actions, cluster, or deployment authority.
+  - Distinct from CCR-2026-0014; the policy-nexus-source identity and its token are
+    never reused, so revoking either lane leaves the other intact.
+  - The publisher calls the in-cluster Forgejo service over plain HTTP; the token
+    stays on the cluster network, restricted by the core-hub NetworkPolicy egress
+    to the forgejo namespace. TLS to the in-cluster service is out of scope here.
+  - Any ExternalSecret in namespace core-hub can reference the store; the namespace
+    is single-tenant (rapp-core-hub) and the store is not shared with other namespaces.
+  - Forgejo PATs carry no server-side expiry; the expiry is enforced by the rotation
+    schedule below and by GENERATED_AT metadata.
+verification:
+  positive:
+  - The minted token metadata reports exactly read:repository; it resolves a branch
+    and fetches one archive of a private coulomb repository, checked without printing
+    the value.
+  - ExternalSecret core-hub/repo-manager-forgejo-source reports Ready and the
+    Secret exists (checked only with the safe secret_annotation_maintenance helper,
+    never by reading values).
+  - make verify-publisher in rapp-core-hub reports 123 accepted / 0 errors once
+    the separate identity-canon registry correction has shipped.
+  negative:
+  - The PAT cannot create, update, or delete repository content, write packages,
+    or read Actions secrets; /admin/users is denied.
+  - The effective permission of core-hub-source on a probe repository is read.
+  - The external-secrets role cannot read platform/workloads/policy-nexus/forgejo-source-read
+    or any sibling path; a default OpenBao identity cannot read this path.
+  - Removing the Secret makes publisher private-source reads fail with 404 again
+    (fail closed), not fall back to another credential.
+  activation_conditions:
+  - Platform operator and Core Hub owner approve this CCR.
+  - Kubernetes auth role binding for the external-secrets service account is confirmed
+    (bound_claims_confirmed set true by confirm-binding).
+  - Delegated metadata apply writes the policy and the auth role.
+  - Attended provisioning mints the PAT and writes it to OpenBao from stdin, without
+    logs, chat, Git, State Hub, argv, or persistent temp files.
+  - The ClusterSecretStore and ExternalSecret are committed and synced; rapp-core-hub
+    mounts the Secret and passes the positive and negative checks above.
+  - ops-warden catalog entry core-hub-forgejo-source-read is added and reviewed.
+  evidence: []
+lifecycle:
+  deactivate: Delete ExternalSecret and Secret in core-hub, unset publisher.existingSecret,
+    revoke the PAT, disable the auth role, and leave private sources failing closed.
+  rotate: Every 180 days or on owner change. Mint a replacement read:repository PAT
+    for core-hub-source, write it to OpenBao from stdin, let ESO refresh (5m), pass
+    make verify-publisher, then revoke the predecessor PAT and record evidence.
+  compromised: Revoke the PAT immediately, delete the Secret, inspect Forgejo access
+    logs for core-hub-source reads, rotate through this lane, record bounded incident
+    evidence.
+state_hub:
+  workplan_id: RAPPCOREHUB-WP-0003
+  task_id: RAPPCOREHUB-WP-0003-T04
--- a/docs/credential-lane-designs/README.md
+++ b/docs/credential-lane-designs/README.md
@ -13,6 +13,7 @@ KeyCape approval-client admission below, which carries executable CCRs
 | [State Hub preflight signing](state-hub-preflight-signing.md) | RPF-WP-0034 (finished) / RPF-WP-0035-T04 | FLEX-WP-0020-T05; STATE-WP-0085-T09 delivered | Deployment binding, owner-approved custody and rotation window |
 | [KeyCape approval-engine clients](keycape-approval-clients.md) | RPF-WP-0035-T05 | KEY-WP-0013-T02; GLAS-WP-0015 | Owner approval of CCR-2026-0017/0018, attended founder window, KeyCape reading build |
 | [Approval client-side readers](keycape-approval-client-side-readers.md) | RPF-WP-0035-T06 | SECRETS-WP-0009-T03; APPROVAL-WP-0002 | Reader 1: exact operator group claim; Reader 2: the presenting actor is unnamed by any owner source |
+| [Core Hub Forgejo source read](core-hub-forgejo-source-read.md) (CCR-2026-0031, `proposed`) | CUST-WP-0075 / RAPPCOREHUB-WP-0003-T04 | rapp-core-hub publisher private sources | Approval + K8s binding confirmation; repo-manager identity-canon correction for 123/0 |

 The completed design workplans are archived. RPF-WP-0035 is the canonical
 queue for owner review, implementation, and live acceptance still required.
--- a/docs/credential-lane-designs/core-hub-forgejo-source-read.md
+++ b/docs/credential-lane-designs/core-hub-forgejo-source-read.md
@ -0,0 +1,115 @@
+# Core Hub Forgejo source-read lane (CCR-2026-0031)
+
+Status: **proposed**, drafted 2026-10-03 under CUST-WP-0075 for
+`RAPPCOREHUB-WP-0003-T04`. Nothing here is approved, applied or provisioned.
+No Forgejo, OpenBao or cluster object was created and no owner message other
+than the CUST-WP-0075 reply was sent.
+
+## Why
+
+The repo-manager publisher in namespace `core-hub` reads the 123-entry
+repository registry (all `coulomb/*`) anonymously. Nine private sources return
+HTTP 404. T04 forbids the Forgejo admin PAT, personal keys or PATs, and the
+`policy-nexus-source` token from CCR-2026-0014. This lane gives the publisher
+its own read-only identity.
+
+## Shape
+
+| Item | Value |
+| --- | --- |
+| Forgejo user | `core-hub-source` (restricted, non-admin, private; random unrecorded password) |
+| Forgejo team | `coulomb/core-hub-source-readers`: permission read, all repositories, `repo.code` read, every other unit none, cannot create repos |
+| PAT | `core-hub-source-read-<UTC timestamp>`, scope exactly `read:repository` |
+| OpenBao | `platform/workloads/core-hub/forgejo-source-read` fields `FORGEJO_SOURCE_TOKEN`, `API_USER`, `API_BASE_URL`, `TOKEN_SCOPES`, `GENERATED_AT` |
+| Policy | `workload-kv-read-core-hub-forgejo-source` (`openbao/policies/…hcl`, read on data + metadata of that one path) |
+| Auth | Kubernetes role `external-secrets-core-hub-forgejo-source`, bound to SA `external-secrets/external-secrets`, TTL 15m |
+| Delivery | ESO store + ExternalSecret below → Secret `core-hub/repo-manager-forgejo-source` key `token` |
+| Consumer | rapp-core-hub `publisher.existingSecret: repo-manager-forgejo-source` (already supported by the chart; mounts read-only at `/var/run/secrets/repo-manager/forgejo/token`) |
+| Rotation | every 180 days or owner change; Forgejo PATs have no server-side expiry |
+| Front door | ops-warden `core-hub-forgejo-source-read`, follow-up after approval (see below) |
+
+Naming follows the CCR-2026-0014 precedent (`<workload>-source`,
+`<workload>-source-readers`). repo-manager has no service-user naming canon.
+Its outstanding `identity-canon` correction is a registry/Forgejo identity
+mismatch for one repository. It blocks the 123/0 acceptance, not this
+credential or its name.
+
+## Draft manifests (land in `argocd/platform-addons/openbao-secretstore/` only after metadata apply)
+
+```yaml
+# CCR-2026-0031. Do not apply until approval and confirm-binding.
+apiVersion: external-secrets.io/v1
+kind: ClusterSecretStore
+metadata:
+  name: openbao-core-hub-forgejo-source
+  labels:
+    app.kubernetes.io/part-of: railiance-gitops
+    railiance-platform/component: external-secrets
+    app.kubernetes.io/name: core-hub
+spec:
+  provider:
+    vault:
+      server: http://openbao.openbao.svc:8200
+      path: platform
+      version: v2
+      auth:
+        kubernetes:
+          mountPath: kubernetes
+          role: external-secrets-core-hub-forgejo-source
+          serviceAccountRef:
+            name: external-secrets
+            namespace: external-secrets
+  conditions:
+    - namespaces:
+        - core-hub
+---
+apiVersion: external-secrets.io/v1
+kind: ExternalSecret
+metadata:
+  name: repo-manager-forgejo-source
+  namespace: core-hub
+spec:
+  refreshInterval: 5m
+  secretStoreRef:
+    kind: ClusterSecretStore
+    name: openbao-core-hub-forgejo-source
+  target:
+    # Explicit target metadata prevents ESO copying source last-applied annotations.
+    template:
+      metadata: {}
+    name: repo-manager-forgejo-source
+    creationPolicy: Owner
+    deletionPolicy: Retain
+  data:
+    - secretKey: token
+      remoteRef:
+        key: workloads/core-hub/forgejo-source-read
+        property: FORGEJO_SOURCE_TOKEN
+```
+
+## Attended sequence (after approval)
+
+1. Reviewers approve and confirm the binding (human acts, recorded by the
+   CCR tooling): `credential-change.py approve CCR-2026-0031 …` and
+   `credential-change.py confirm-binding CCR-2026-0031 …`.
+2. Founder, in their own terminal (child is silent; read the diagnostic file
+   on failure, never the output):
+   `warden access openbao-platform-admin-login --exec -- scripts/openbao-bootstrap-core-hub-source.sh`.
+   This runs the delegated metadata apply (policy + auth role), then
+   `scripts/provision-core-hub-forgejo-source.py`, which gets the Forgejo admin
+   token only through `warden access forgejo-admin-api-token --exec`, mints the
+   PAT, verifies scope, private-repo read and denials, writes the value to
+   OpenBao from stdin, reads it back for equality, and records non-secret
+   evidence. Re-running when the PAT and the OpenBao value both exist is a no-op;
+   a half state fails closed and points to rotation. Failure diagnostics:
+   `/tmp/core-hub-source-bootstrap-diagnostic.json` (stage + bounded reason only).
+3. Commit the two manifests above and let Argo sync; check presence only with
+   `scripts/secret_annotation_maintenance.py` through the authorized admin path.
+4. rapp-core-hub sets `publisher.existingSecret`, deploys, runs
+   `make verify-publisher` (expects 123/0 once repo-manager ships the
+   identity-canon correction), then enables `connectCandidate`.
+5. Add the ops-warden catalog entry `core-hub-forgejo-source-read`
+   (risk high, owner railiance-platform, path above, rotation steps from the
+   CCR) and move the CCR to `verified`/`active` with evidence.
+
+Offline tests: `python3 -m unittest tests/test_core_hub_forgejo_source_provision.py`.
--- a/openbao/policies/workload-kv-read-core-hub-forgejo-source.hcl
+++ b/openbao/policies/workload-kv-read-core-hub-forgejo-source.hcl
@ -0,0 +1,9 @@
+# CCR-2026-0031 (proposed). External Secrets reader for the dedicated Core Hub
+# Forgejo read:repository token. Not applied until the CCR is approved.
+path "platform/data/workloads/core-hub/forgejo-source-read" {
+  capabilities = ["read"]
+}
+
+path "platform/metadata/workloads/core-hub/forgejo-source-read" {
+  capabilities = ["read"]
+}
--- a/scripts/openbao-bootstrap-core-hub-source.sh
+++ b/scripts/openbao-bootstrap-core-hub-source.sh
@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+# Exact silent owner command for the CCR-2026-0031 attended bootstrap.
+# Run only after CCR-2026-0031 is approved and its binding confirmed:
+#   warden access openbao-platform-admin-login --exec -- scripts/openbao-bootstrap-core-hub-source.sh
+# Both steps are idempotent; on failure read /tmp/core-hub-source-bootstrap-diagnostic.json.
+set -euo pipefail
+
+REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "${REPO_DIR}"
+
+python3 scripts/credential-change.py applier-apply CCR-2026-0031 \
+  --actor "attended operator via governed platform-admin lane" \
+  --confirm "DELEGATED APPLY CCR-2026-0031" \
+  --record-state-hub \
+  >/dev/null 2>&1
+python3 scripts/provision-core-hub-forgejo-source.py >/dev/null 2>&1
--- a/scripts/provision-core-hub-forgejo-source.py
+++ b/scripts/provision-core-hub-forgejo-source.py
@ -0,0 +1,384 @@
+#!/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(source_token: str) -> str:
+    for page in range(1, 6):
+        _, repos = api_request(source_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("source token sees no private repository in the org")
+
+
+def verify_source_token(source_token: str, scopes: object) -> str:
+    if scopes != SCOPES:
+        raise ProvisionError("Forgejo returned scopes other than exactly read:repository")
+    probe = find_private_probe(source_token)
+    _, 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"))
+        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
--- a/tests/test_core_hub_forgejo_source_provision.py
+++ b/tests/test_core_hub_forgejo_source_provision.py
@ -0,0 +1,178 @@
+"""Offline tests for the CCR-2026-0031 attended bootstrap. No network, no bao."""
+
+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-core-hub-forgejo-source.py"
+SPEC = importlib.util.spec_from_file_location("core_hub_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)
+
+SENTINEL = "NONPRODUCTION_SOURCE_TOKEN_SENTINEL"
+USER = MODULE.USER
+
+
+def completed(args, rc=0, out=b""):
+    return subprocess.CompletedProcess(args, rc, out, b"")
+
+
+class Harness:
+    """Fake Forgejo + OpenBao recording every call."""
+
+    def __init__(self, *, pat_exists=False, bao_has=False, put_rc=0, readback=None):
+        self.calls: list[tuple[str, str, dict]] = []
+        self.bao_calls: list[tuple[list[str], bytes | None]] = []
+        self.pat_exists = pat_exists
+        self.bao_has = bao_has
+        self.put_rc = put_rc
+        self.readback = SENTINEL.encode() if readback is None else readback
+
+    def api(self, token, method, path, **kwargs):
+        self.calls.append((method, path, kwargs))
+        if method == "GET" and path == f"/users/{USER}/tokens?limit=100":
+            return 200, ([{"name": MODULE.TOKEN_PREFIX + "20260101T000000Z"}] if self.pat_exists else [])
+        if method == "PATCH" and path == f"/admin/users/{USER}":
+            return 200, {"login": USER}
+        if method == "POST" and path == f"/users/{USER}/tokens":
+            return 201, {"id": 73, "sha1": SENTINEL, "scopes": ["read:repository"]}
+        if method == "DELETE" and path == f"/users/{USER}/tokens/73":
+            return 204, None
+        raise AssertionError((method, path))
+
+    def bao(self, args, *, stdin=None):
+        self.bao_calls.append((args, stdin))
+        if args[:2] == ["kv", "get"]:
+            if not any(a[:2] == ["kv", "put"] for a, _ in self.bao_calls):
+                return completed(args, 0 if self.bao_has else 2)
+            return completed(args, 0, self.readback + b"\n")
+        if args[:2] == ["kv", "put"]:
+            return completed(args, self.put_rc)
+        return completed(args)
+
+    def run(self):
+        with (
+            mock.patch.object(MODULE, "ensure_user"),
+            mock.patch.object(MODULE, "ensure_team", return_value=7),
+            mock.patch.object(MODULE, "verify_source_token", return_value="private-probe"),
+            mock.patch.object(MODULE, "verify_effective_permission"),
+            mock.patch.object(MODULE, "record_evidence") as evidence,
+            mock.patch.object(MODULE, "api_request", side_effect=self.api),
+            mock.patch.object(MODULE, "bao", side_effect=self.bao),
+        ):
+            self.evidence = evidence
+            return MODULE.provision("admin-token")
+
+    def revoked(self):
+        return any(m == "DELETE" for m, _, _ in self.calls)
+
+
+class ContractTests(unittest.TestCase):
+    def test_identity_is_dedicated_and_not_policy_nexus(self) -> None:
+        self.assertEqual(MODULE.SCOPES, ["read:repository"])
+        self.assertEqual(MODULE.USER, "core-hub-source")
+        self.assertNotIn("policy-nexus", MODULE.USER + MODULE.TEAM + MODULE.BAO_PATH + MODULE.TOKEN_PREFIX)
+        self.assertEqual(MODULE.BAO_PATH, "platform/workloads/core-hub/forgejo-source-read")
+        self.assertNotIn("repo.code", MODULE.NON_CODE_UNITS)
+
+    def test_team_contract_rejects_any_write_or_extra_unit(self) -> None:
+        good = {
+            "permission": "read",
+            "includes_all_repositories": True,
+            "can_create_org_repo": False,
+            "units_map": {"repo.code": "read", **{u: "none" for u in MODULE.NON_CODE_UNITS}},
+        }
+        self.assertTrue(MODULE.team_satisfies_contract(good))
+        for change in (
+            {"permission": "write"},
+            {"can_create_org_repo": True},
+            {"includes_all_repositories": False},
+            {"units_map": {**good["units_map"], "repo.packages": "read"}},
+        ):
+            with self.subTest(change=change):
+                self.assertFalse(MODULE.team_satisfies_contract({**good, **change}))
+
+    def test_broader_scope_rejected_before_network(self) -> None:
+        with mock.patch.object(MODULE, "api_request") as request:
+            with self.assertRaises(MODULE.ProvisionError):
+                MODULE.verify_source_token(SENTINEL, ["read:repository", "write:repository"])
+        request.assert_not_called()
+
+    def test_errors_never_carry_query_strings_or_unexpected_detail(self) -> None:
+        with mock.patch.object(MODULE.urllib.request, "urlopen") as urlopen:
+            urlopen.side_effect = MODULE.urllib.error.HTTPError("u", 500, "x", {}, None)
+            with self.assertRaises(MODULE.ProvisionError) as ctx:
+                MODULE.api_request("t", "GET", "/orgs/coulomb/repos?limit=50&page=1")
+        self.assertEqual(str(ctx.exception), "GET /orgs/coulomb/repos returned HTTP 500")
+        with mock.patch.object(MODULE, "DIAGNOSTIC_PATH") as path:
+            MODULE.write_diagnostic("provision", RuntimeError(SENTINEL))
+        self.assertNotIn(SENTINEL, path.write_text.call_args.args[0])
+
+    def test_private_probe_requires_a_private_repository(self) -> None:
+        with mock.patch.object(MODULE, "api_request", return_value=(200, [{"name": "pub", "private": False}])):
+            with self.assertRaises(MODULE.ProvisionError):
+                MODULE.find_private_probe(SENTINEL)
+        repos = [{"name": "pub", "private": False}, {"name": "priv", "private": True}]
+        with mock.patch.object(MODULE, "api_request", return_value=(200, repos)):
+            self.assertEqual(MODULE.find_private_probe(SENTINEL), "priv")
+
+
+class ProvisionFlowTests(unittest.TestCase):
+    def test_fresh_mint_writes_value_only_via_stdin(self) -> None:
+        h = Harness()
+        self.assertEqual(h.run(), "minted")
+        put_args, put_stdin = next((a, s) for a, s in h.bao_calls if a[:2] == ["kv", "put"])
+        self.assertEqual(put_stdin, SENTINEL.encode())
+        self.assertIn(f"{MODULE.BAO_FIELD}=-", put_args)
+        self.assertNotIn(SENTINEL, " ".join(put_args))
+        self.assertFalse(h.revoked())
+        h.evidence.assert_called_once()
+        self.assertNotIn(SENTINEL, repr(h.evidence.call_args))
+
+    def test_rerun_when_fully_provisioned_is_noop(self) -> None:
+        h = Harness(pat_exists=True, bao_has=True)
+        self.assertEqual(h.run(), "noop")
+        self.assertFalse(any(m in ("POST", "PATCH", "DELETE") for m, _, _ in h.calls))
+        self.assertFalse(any(a[:2] == ["kv", "put"] for a, _ in h.bao_calls))
+
+    def test_drift_fails_closed_without_mutation(self) -> None:
+        for pat, has in ((True, False), (False, True)):
+            with self.subTest(pat=pat, bao=has):
+                h = Harness(pat_exists=pat, bao_has=has)
+                with self.assertRaises(MODULE.ProvisionError):
+                    h.run()
+                self.assertFalse(any(m in ("POST", "PATCH") for m, _, _ in h.calls))
+
+    def test_bao_write_failure_revokes_pat_and_cleans_kv(self) -> None:
+        h = Harness(put_rc=2)
+        with self.assertRaises(MODULE.ProvisionError):
+            h.run()
+        self.assertTrue(h.revoked())
+        self.assertIn((["kv", "delete", MODULE.BAO_PATH], None), h.bao_calls)
+
+    def test_readback_mismatch_revokes_pat(self) -> None:
+        h = Harness(readback=b"other")
+        with self.assertRaises(MODULE.ProvisionError):
+            h.run()
+        self.assertTrue(h.revoked())
+
+    def test_inner_without_admin_token_fails_without_network(self) -> None:
+        with (
+            mock.patch.object(MODULE.sys, "argv", ["x", "--inner"]),
+            mock.patch.dict(MODULE.os.environ, {}, clear=True),
+            mock.patch.object(MODULE, "write_diagnostic") as diag,
+            mock.patch.object(MODULE, "provision") as provision,
+        ):
+            self.assertEqual(MODULE.main(), 1)
+        provision.assert_not_called()
+        diag.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()