railiance-platform/scripts/provision-reuse-surface-forgejo-source.py

#!/usr/bin/env python3
"""Silent attended CCR-2026-0046 bootstrap; never runs without exact approval.

Only this seven-repository identity/PAT/KV path is mutable. No policy, ESO,
consumer rollout, or registration mutations occur here. A receipt is evidence,
not authorization. Uncertain requests are recorded and reconciled, never retried
in the same execution; orphan PAT drift requires attended recovery.
"""
from __future__ import annotations
import argparse
import base64
import datetime as dt
import fcntl
import hashlib
import json
import os
from pathlib import Path
import re
import secrets
import subprocess
import sys
import urllib.error
import urllib.request
import yaml

ROOT = Path(__file__).resolve().parent.parent
CCR = 'CCR-2026-0046'
CCR_FILE = 'credential-change-requests/CCR-2026-0046-reuse-surface-forgejo-source-read.yaml'
PLAN_FILE = 'docs/proposals/2026-10-15-reuse-private-reader/bootstrap-contract.json'
SCRIPT_FILE = 'scripts/provision-reuse-surface-forgejo-source.py'
API = 'https://forgejo.coulomb.social/api/v1'
USER = 'reuse-surface-source'
TEAM = 'reuse-surface-source-readers'
ORG = 'coulomb'
KV = 'platform/workloads/reuse/reuse-surface/forgejo-source-read'
FIELD = 'FORGEJO_SOURCE_TOKEN'
PREFIX = 'reuse-surface-source-read-'
REPOS = frozenset('rapp-openbao rapp-policy-nexus rapp-postgres rapp-sbom-nexus rapp-telemetry rapp-tenant-engine rapp-user-engine'.split())
SCOPES = ['read:repository']
UNITS = {'repo.code': 'read', **{x: 'none' for x in ('repo.actions', 'repo.packages', 'repo.issues', 'repo.ext_issues', 'repo.wiki', 'repo.pulls', 'repo.releases', 'repo.projects', 'repo.ext_wiki')}}

class Refused(Exception):
    """Messages must be fixed non-secret codes, never provider text."""

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        raise Refused('redirect_refused')

class Journal:
    def __init__(self, path):
        flags = os.O_RDWR | os.O_CREAT | os.O_APPEND | os.O_NOFOLLOW
        self.fd = os.open(path, flags, 0o600)
        if os.fstat(self.fd).st_mode & 0o077:
            os.close(self.fd)
            raise Refused('receipt_permissions')
        fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        self.attempt = secrets.token_hex(12)
    def put(self, event, **fields):
        # Only locally authored operation tags, scalar status/IDs are accepted.
        allowed = {'operation', 'status', 'http_status', 'token_id', 'team_id', 'outcome', 'error_code'}
        if set(fields) - allowed:
            raise Refused('receipt_fields')
        record = {'at': dt.datetime.now(dt.timezone.utc).isoformat(), 'ccr': CCR,
                  'attempt': self.attempt, 'event': event, **fields}
        data = (json.dumps(record, sort_keys=True) + '\n').encode()
        while data:
            written = os.write(self.fd, data)
            if written <= 0:
                raise Refused('receipt_write_failed')
            data = data[written:]
        os.fsync(self.fd)
    def close(self):
        os.close(self.fd)

class Client:
    def __init__(self, admin, journal):
        self.admin, self.journal = admin, journal
        self.opener = urllib.request.build_opener(NoRedirect())
    def request(self, method, path, payload=None, expected=(200,), *, token=None, sudo=False, basic=None, operation=None):
        if not path.startswith('/') or '://' in path:
            raise Refused('invalid_api_path')
        headers = {'Authorization': basic or 'token ' + (self.admin if token is None else token), 'Accept': 'application/json'}
        if sudo:
            headers['Sudo'] = USER
        if payload is not None:
            headers['Content-Type'] = 'application/json'
        if method != 'GET':
            if not operation:
                raise Refused('mutation_without_intent')
            self.journal.put('intent', operation=operation)
        req = urllib.request.Request(API + path, data=None if payload is None else json.dumps(payload).encode(), headers=headers, method=method)
        self.last_total_count = None
        try:
            with self.opener.open(req, timeout=30) as response:
                status, body = response.status, response.read(2_000_001)
                raw_total = response.headers.get('X-Total-Count')
                if raw_total is not None:
                    if not re.fullmatch(r'[0-9]+', raw_total):
                        raise Refused('invalid_total_count')
                    self.last_total_count = int(raw_total)
        except urllib.error.HTTPError as error:
            status, body = error.code, b''
        except Exception:
            if operation:
                self.journal.put('outcome', operation=operation, status='ambiguous')
            raise Refused('request_observation_failed') from None
        if operation:
            self.journal.put('outcome', operation=operation, status='observed', http_status=status)
        if status not in expected:
            raise Refused('unexpected_http_status')
        if len(body) > 2_000_000:
            raise Refused('response_bound_exceeded')
        if not body:
            return status, None
        try:
            return status, json.loads(body)
        except Exception:
            raise Refused('invalid_json') from None
    def pages(self, path, *, sudo=False):
        """Page until observed empty; requested size may be server-clamped.

        Forgejo11.0.3 default MaxResponseItems=50, so a response shorter than
        our requested100 proves nothing. Detect unstable/duplicate IDs and
        reconcile X-Total-Count whenever available; exceeding bound refuses.
        """
        items, seen, expected_total = [], set(), None
        for page in range(1, 21):
            separator = '&' if '?' in path else '?'
            _, batch = self.request('GET', path + separator + f'limit=100&page={page}', sudo=sudo)
            if not isinstance(batch, list) or any(not isinstance(x, dict) or type(x.get('id')) is not int or x['id'] <= 0 for x in batch):
                raise Refused('invalid_list')
            observed_total = getattr(self, 'last_total_count', None)
            if observed_total is not None:
                if expected_total is not None and observed_total != expected_total:
                    raise Refused('unstable_total_count')
                expected_total = observed_total
            if not batch:
                if expected_total is not None and len(items) != expected_total:
                    raise Refused('total_count_mismatch')
                return items
            for item in batch:
                if item['id'] in seen:
                    raise Refused('duplicate_page_id')
                seen.add(item['id'])
            items.extend(batch)
            if expected_total is not None and len(items) > expected_total:
                raise Refused('total_count_mismatch')
        raise Refused('pagination_bound_exceeded')

class Bao:
    @staticmethod
    def run(args, stdin=None):
        try:
            return subprocess.run(['bao', *args], input=stdin, capture_output=True, timeout=60, check=False, cwd=ROOT)
        except Exception:
            raise Refused('bao_observation_failed') from None
    def get(self):
        result = self.run(['kv', 'get', '-format=json', KV])
        if result.returncode != 0:
            # Distinguish an actual not-found response from denied/unavailable.
            # Never include captured raw errors in any evidence.
            if re.search(rb'Code:\s*404\b', result.stderr or b'') or b'No value found at ' in (result.stderr or b''):
                return None
            raise Refused('bao_preflight_failed')
        try:
            data = json.loads(result.stdout)['data']['data']
        except Exception:
            raise Refused('bao_invalid_record') from None
        if set(data) != {FIELD, 'API_USER', 'API_BASE_URL', 'TOKEN_SCOPES', 'GENERATED_AT', 'CCR_ID'} or data['API_USER'] != USER or data['API_BASE_URL'] != 'https://forgejo.coulomb.social' or data['TOKEN_SCOPES'] != 'read:repository' or data['CCR_ID'] != CCR or not isinstance(data[FIELD], str) or not data[FIELD]:
            raise Refused('bao_record_conflict')
        return data
    def put(self, token, journal):
        journal.put('intent', operation='kv_create')
        timestamp = dt.datetime.now(dt.timezone.utc).isoformat()
        try:
            result = self.run(['kv', 'put', '-cas=0', KV, FIELD + '=-', 'API_USER=' + USER,
                               'API_BASE_URL=https://forgejo.coulomb.social', 'TOKEN_SCOPES=read:repository',
                               'GENERATED_AT=' + timestamp, 'CCR_ID=' + CCR], token.encode())
        except Refused:
            journal.put('outcome', operation='kv_create', status='ambiguous')
            raise
        journal.put('outcome', operation='kv_create', status='observed' if result.returncode == 0 else 'ambiguous')
        # Readback reconciles even a response failure after successful write.
        stored = self.get()
        if stored is None or not secrets.compare_digest(stored[FIELD], token):
            raise Refused('bao_write_requires_reconciliation')
        journal.put('verified', operation='kv_create', status='matched')


def valid_user(user):
    return isinstance(user, dict) and user.get('login') == USER and user.get('restricted') is True and user.get('is_admin') is False and user.get('visibility') == 'private'

def valid_team(team):
    return isinstance(team, dict) and isinstance(team.get('id'), int) and team.get('name') == TEAM and team.get('permission') == 'read' and team.get('includes_all_repositories') is False and team.get('can_create_org_repo') is False and team.get('units_map') == UNITS

def repo_names(items):
    names = set()
    for repo in items:
        if repo.get('owner', {}).get('login') != ORG or repo.get('name') not in REPOS:
            raise Refused('unexpected_team_repository')
        names.add(repo['name'])
    if len(names) != len(items):
        raise Refused('duplicate_repository')
    return names

def inspect(client, bao):
    """Read-only preflight before mutations.

    Forgejo11.0.3 ListMyRepos uses OwnerID plus Collaborate=None; model
    ValueOrDefault(true) includes direct access and team grants (not only owned).
    Deployed-server API compatibility and live scope proof remain separate gates.
    """
    for repo in sorted(REPOS):
        _, meta = client.request('GET', f'/repos/{ORG}/{repo}')
        if not isinstance(meta, dict) or meta.get('private') is not True or meta.get('full_name') != ORG + '/' + repo:
            raise Refused('source_repository_conflict')
    _, probe = client.request('GET', '/repos/coulomb/rapp-core-hub')
    if not isinstance(probe, dict) or probe.get('private') is not True:
        raise Refused('negative_probe_not_private')
    _, user = client.request('GET', '/users/' + USER, expected=(200, 404))
    if user is not None and not valid_user(user):
        raise Refused('user_conflict')
    teams = [x for x in client.pages(f'/orgs/{ORG}/teams') if x.get('name') == TEAM]
    if len(teams) > 1:
        raise Refused('duplicate_team')
    team = teams[0] if teams else None
    members, repos = [], set()
    if team is not None:
        if not valid_team(team):
            raise Refused('team_conflict')
        members = client.pages('/teams/' + str(team['id']) + '/members')
        if len(members) > 1 or any(x.get('login') != USER for x in members):
            raise Refused('team_members_conflict')
        repos = repo_names(client.pages('/teams/' + str(team['id']) + '/repos'))
    tokens = []
    if user is not None:
        orgs = client.pages('/user/orgs', sudo=True)
        if any(x.get('username') != ORG for x in orgs):
            raise Refused('extra_organization_affiliation')
        grants = client.pages('/user/teams', sudo=True)
        if any(x.get('name') != TEAM or x.get('organization', {}).get('username') != ORG for x in grants):
            raise Refused('extra_team_grant')
        for meta in client.pages('/user/repos?order_by=id', sudo=True):
            if meta.get('private') is True and meta.get('full_name') not in {ORG + '/' + x for x in REPOS}:
                raise Refused('extra_private_repository_grant')
            if meta.get('owner', {}).get('login') == USER:
                raise Refused('service_user_owned_repository')
        tokens = client.pages('/users/' + USER + '/tokens')
        if len(tokens) > 1 or any(not str(x.get('name', '')).startswith(PREFIX) or x.get('scopes') != SCOPES or not isinstance(x.get('id'), int) for x in tokens):
            raise Refused('token_conflict')
    stored = bao.get()
    if bool(tokens) != bool(stored):
        raise Refused('orphan_token_or_kv_requires_recovery')
    if tokens and (not isinstance(tokens[0].get('token_last_eight'), str) or len(tokens[0]['token_last_eight']) != 8 or not secrets.compare_digest(tokens[0]['token_last_eight'], stored[FIELD][-8:])):
        raise Refused('pat_kv_binding_unverifiable')
    if tokens and (not team or repos != REPOS or len(members) != 1):
        raise Refused('active_lane_grant_drift')
    return user, team, repos, members, tokens, stored

def verify(client, source_token, plan):
    for source in plan['sources']:
        _, repo = client.request('GET', '/repos/' + ORG + '/' + source['repo'], token=source_token)
        if repo.get('permissions') != {'admin': False, 'push': False, 'pull': True}:
            raise Refused('effective_permission_not_read')
        # Exact consented bytes fetched quietly; raw output is never retained.
        path = f"/repos/{ORG}/{source['repo']}/contents/registry/indexes/capabilities.yaml?ref={source['consented_commit']}"
        _, content = client.request('GET', path, token=source_token)
        try:
            raw = base64.b64decode(content['content'], validate=False)
        except Exception:
            raise Refused('invalid_index_encoding') from None
        if hashlib.sha256(raw).hexdigest() != source['index_sha256']:
            raise Refused('index_digest_mismatch')
        client.request('GET', f"/repos/{ORG}/{source['repo']}/actions/secrets?limit=1", token=source_token, expected=(403, 404))
    client.request('GET', '/admin/users', token=source_token, expected=(403, 404))
    # A known private excluded source is selected by reviewed contract.
    probe = plan['negative_private_probe']
    _, meta = client.request('GET', '/repos/' + ORG + '/' + probe)
    if not isinstance(meta, dict) or meta.get('private') is not True:
        raise Refused('negative_probe_not_private')
    client.request('GET', '/repos/' + ORG + '/' + probe, token=source_token, expected=(403, 404))
    client.journal.put('verified', operation='reader_scope', status='seven_exact_no_extra_private_grants')

def provision(client, bao, plan):
    user, team, repos, members, tokens, stored = inspect(client, bao)
    if tokens:
        verify(client, stored[FIELD], plan)
        client.journal.put('terminal', outcome='verified_noop')
        return 'noop'
    if user is None:
        _, user = client.request('POST', '/admin/users', {'username': USER, 'email': USER + '@coulomb.social', 'full_name': 'Reuse Surface seven-source reader', 'password': secrets.token_urlsafe(48), 'restricted': True, 'must_change_password': False, 'send_notify': False, 'visibility': 'private'}, expected=(201,), operation='user_create')
        if not valid_user(user):
            raise Refused('created_user_contract_failed')
    if team is None:
        _, team = client.request('POST', f'/orgs/{ORG}/teams', {'name': TEAM, 'description': CCR + ' exact seven consented source repos', 'permission': 'read', 'includes_all_repositories': False, 'can_create_org_repo': False, 'units': ['repo.code'], 'units_map': UNITS}, expected=(201,), operation='team_create')
        if not valid_team(team):
            raise Refused('created_team_contract_failed')
    team_id = team['id']
    for repo in sorted(REPOS - repos):
        client.request('PUT', f'/teams/{team_id}/repos/{ORG}/{repo}', expected=(204,), operation='team_repo_' + repo)
    if not members:
        client.request('PUT', f'/teams/{team_id}/members/{USER}', expected=(204,), operation='team_member_add')
    # Recheck complete grants before creating any PAT; refuse implicit grants.
    _, _, actual_repos, actual_members, _, _ = inspect(client, bao)
    if actual_repos != REPOS or len(actual_members) != 1:
        raise Refused('incomplete_team_scope')
    password = secrets.token_urlsafe(48)
    client.request('PATCH', '/admin/users/' + USER, {'password': password, 'restricted': True, 'admin': False, 'allow_create_organization': False, 'allow_git_hook': False, 'allow_import_local': False, 'max_repo_creation': 0, 'must_change_password': False, 'active': True, 'visibility': 'private'}, operation='service_password_randomize')
    basic = 'Basic ' + base64.b64encode((USER + ':' + password).encode()).decode()
    _, record = client.request('POST', '/users/' + USER + '/tokens', {'name': PREFIX + client.journal.attempt, 'scopes': SCOPES}, expected=(201,), basic=basic, operation='pat_create')
    if not isinstance(record, dict) or not isinstance(record.get('id'), int) or record.get('scopes') != SCOPES or not isinstance(record.get('sha1'), str) or not record['sha1']:
        raise Refused('created_pat_requires_recovery')
    client.journal.put('identified', operation='pat_create', token_id=record['id'], team_id=team_id)
    token = record['sha1']
    verify(client, token, plan)
    bao.put(token, client.journal)
    # Repeat complete grant inventory after custody; verify exactly one PAT.
    inspect(client, bao)
    client.journal.put('terminal', outcome='provisioned_verified_delivery_pending')
    return 'minted'

def approval(path):
    try:
        receipt = json.loads(Path(path).read_text())
        ccr = yaml.safe_load((ROOT / CCR_FILE).read_text())
        if receipt['ccr_id'] != CCR or receipt['decision'] != 'approved' or not receipt['supervisor'] or not receipt['decision_ref'] or ccr['status'] != 'approved':
            raise Refused('approval_missing')
        files = {CCR_FILE, PLAN_FILE, SCRIPT_FILE, 'openbao/policies/workload-kv-read-reuse-surface-runtime-eso.hcl'}
        if set(receipt['approved_artifacts']) != files:
            raise Refused('approval_artifact_set')
        for rel in files:
            if hashlib.sha256((ROOT / rel).read_bytes()).hexdigest() != receipt['approved_artifacts'][rel]:
                raise Refused('approval_revision_mismatch')
        plan = json.loads((ROOT / PLAN_FILE).read_text())
        if {x['repo'] for x in plan['sources']} != REPOS or len(plan['sources']) != 7 or plan['negative_private_probe'] != 'rapp-core-hub' or any(not re.fullmatch(r'[a-f0-9]{40}', x['consented_commit']) or not re.fullmatch(r'[a-f0-9]{64}', x['index_sha256']) for x in plan['sources']):
            raise Refused('plan_scope_conflict')
        return plan
    except Refused:
        raise
    except Exception:
        raise Refused('approval_invalid') from None

def main(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument('--approval', required=True)
    parser.add_argument('--receipt', required=True)
    parser.add_argument('--inner', action='store_true', help=argparse.SUPPRESS)
    args = parser.parse_args(argv)
    journal = None
    lane_lock = None
    try:
        plan = approval(args.approval)
        if not args.inner:
            # Governed OpenBao attended lane surrounds this command. Warden
            # supplies only Forgejo API_TOKEN to inner child, not via argv/files.
            result = subprocess.run(['warden', 'access', 'forgejo-admin-api-token', '--field', 'API_TOKEN', '--exec', '--', sys.executable, str(ROOT / SCRIPT_FILE), '--inner', '--approval', str(Path(args.approval).resolve()), '--receipt', str(Path(args.receipt).resolve())], capture_output=True, timeout=900, check=False, cwd=ROOT)
            return 0 if result.returncode == 0 else 1
        lane_lock = os.open('/tmp/' + CCR + '-reuse-reader-' + str(os.getuid()) + '.lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
        if os.fstat(lane_lock).st_uid != os.getuid() or os.fstat(lane_lock).st_mode & 0o077:
            raise Refused('lane_lock_owner_or_permissions')
        fcntl.flock(lane_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        journal = Journal(args.receipt)
        admin = os.environ.get('API_TOKEN')
        if not admin:
            raise Refused('admin_route_missing')
        provision(Client(admin, journal), Bao(), plan)
        return 0
    except Exception as error:
        if journal:
            journal.put('terminal', outcome='stopped_reconcile_before_retry', error_code=str(error) if isinstance(error, Refused) else 'internal_failure')
        return 1
    finally:
        if journal:
            journal.close()
        if lane_lock is not None:
            os.close(lane_lock)

if __name__ == '__main__':
    try:
        result = main()
    except Exception:
        result = 1
    raise SystemExit(result)