railiance-platform/scripts/informed_decision_requester_metadata.py

"""Silent, attended, compare-before-write metadata child for CCR-2026-0032.

Does not read/write KV values, change issuer custody, or deploy Kubernetes objects.
"""
import argparse
from datetime import datetime, timezone
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import sys
import uuid

ROOT = Path(__file__).resolve().parents[1]
CCR = ROOT / 'credential-change-requests/CCR-2026-0032.yaml'
POLICY = 'workload-kv-read-informed-decision-sitting-requester-eso'
ROLE = 'auth/kubernetes/role/informed-decision-sitting-requester-eso'


class Refused(Exception):
    def __init__(self, code):
        self.code = code


def cli(arguments, payload=None, *, missing_ok=False):
    try:
        result = subprocess.run(['bao', *arguments], input=payload,
                                capture_output=True, timeout=30, check=False)
    except (OSError, subprocess.TimeoutExpired):
        raise Refused(4) from None
    if result.returncode:
        # bao read turns an empty 404 into this exact path-specific diagnostic.
        # Do not treat arbitrary stderr containing 404, denial or timeout as absence.
        if (missing_ok and arguments[:2] == ['read', '-format=json']
                and len(arguments) == 3 and result.returncode == 2
                and not result.stdout
                and result.stderr.strip() == ('No value found at ' + arguments[2]).encode()):
            return None
        raise Refused(4)
    if arguments[0] == 'write' and not result.stdout.strip():
        return {}
    try:
        return json.loads(result.stdout)
    except (ValueError, UnicodeError):
        raise Refused(5) from None


def role_matches(value):
    expected = {
        'bound_service_account_names': ['informed-decision-eso'],
        'bound_service_account_namespaces': ['informed-decision'],
        'audience': 'openbao', 'token_policies': [POLICY],
        'token_ttl': 900, 'token_max_ttl': 900,
        'token_explicit_max_ttl': 900, 'token_no_default_policy': True,
    }
    return (isinstance(value, dict)
            and all(type(value.get(k)) is type(v) and value[k] == v for k, v in expected.items())
            and value.get('token_period', 0) == 0
            and value.get('bound_service_account_namespace_selector', '') == ''
            and value.get('token_type', 'default') in ('default', 'service'))


def source(expected_digest):
    if hashlib.sha256(CCR.read_bytes()).hexdigest() != expected_digest:
        raise Refused(2)
    spec = importlib.util.spec_from_file_location('credential_change', ROOT / 'scripts/credential-change.py')
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    ccr, errors, _ = module.validate_ccr(CCR)
    if errors or ccr['status'] != 'approved' or ccr['openbao']['auth'].get('bound_claims_confirmed') is not True:
        raise Refused(2)
    bao = ccr['openbao']
    auth = bao['auth']
    if (bao['policy_name'] != POLICY or bao['kv_path'] != 'platform/workloads/informed-decision/sitting-requester'
            or bao['fields'] != ['CLIENT_SECRET'] or bao.get('metadata_read') is not False
            or bao.get('token_self_lifecycle', False) is not False
            or auth['method'] != 'kubernetes' or auth['mount'] != 'kubernetes'
            or auth['role'] != ROLE.rsplit('/', 1)[1]):
        raise Refused(2)
    payload = module.auth_payload(ccr)
    if payload != {'bound_service_account_names': ['informed-decision-eso'],
                   'bound_service_account_namespaces': ['informed-decision'],
                   'policies': POLICY, 'ttl': '15m', 'audience': 'openbao',
                   'token_max_ttl': '15m', 'token_explicit_max_ttl': '15m',
                   'token_no_default_policy': True}:
        raise Refused(2)
    policy = module.generated_policy_hcl(ccr)
    if (ROOT / bao['policy_file']).read_text() != policy:
        raise Refused(2)
    return policy, payload


def apply(policy, payload, *, command=cli, phase=lambda value: None):
    phase("read_policy")
    current_policy = command(['read', '-format=json', 'sys/policies/acl/' + POLICY], missing_ok=True)
    phase('read_role')
    current_role = command(['read', '-format=json', ROLE], missing_ok=True)
    # Check BOTH before mutating either; differing objects require a fresh review.
    if current_policy is not None and current_policy.get('data', {}).get('policy') != policy:
        raise Refused(3)
    if current_role is not None and not role_matches(current_role.get('data')):
        raise Refused(3)
    if current_policy is None:
        phase('write_policy')
        command(['write', '-format=json', 'sys/policies/acl/' + POLICY, '-'],
                json.dumps({'policy': policy}).encode())
    if current_role is None:
        phase('write_role')
        command(['write', '-format=json', ROLE, '-'], json.dumps(payload).encode())
    phase('verify_policy')
    verified_policy = command(['read', '-format=json', 'sys/policies/acl/' + POLICY])
    phase('verify_role')
    verified_role = command(['read', '-format=json', ROLE])
    if (verified_policy.get('data', {}).get('policy') != policy
            or not role_matches(verified_role.get('data'))):
        raise Refused(5)


def main():
    receipt = {'schema': 'requester-metadata-attempt.v1', 'ccr': 'CCR-2026-0032',
               'phase': 'arguments', 'exit_code': 2, 'native_delivery_verified': False}
    def phase(value):
        receipt['phase'] = value
    try:
        parser = argparse.ArgumentParser(add_help=False)
        parser.add_argument('--expected-ccr-sha256', required=True)
        # Parsing is deliberately silent even on malformed arguments.
        with open(os.devnull, 'w') as sink:
            original = sys.stderr
            try:
                sys.stderr = sink
                args = parser.parse_args()
            finally:
                sys.stderr = original
        phase('attended_boundary')
        if (Path.home().parent.name != '.warden-attended-login'
                or os.getenv('BAO_TOKEN') or os.getenv('VAULT_TOKEN')):
            raise Refused(2)
        phase('source')
        policy, payload = source(args.expected_ccr_sha256)
        phase('identity')
        identity = cli(['token', 'lookup', '-format=json']).get('data', {})
        policies = identity.get('policies', [])
        if 'platform-admin' not in policies or 'root' in policies:
            raise Refused(2)
        apply(policy, payload, phase=phase)
        phase('metadata_verified')
        receipt['exit_code'] = 0
        return 0
    except Refused as exc:
        receipt['exit_code'] = exc.code
        return exc.code
    except BaseException:
        return 2
    finally:
        # Fixed metadata only. Never persist CLI bytes, exception text or identity data.
        try:
            directory = ROOT / 'docs/evidence/requester-metadata'
            directory.mkdir(parents=True, exist_ok=True)
            receipt['at'] = datetime.now(timezone.utc).isoformat()
            path = directory / (uuid.uuid4().hex + '.json')
            fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            with os.fdopen(fd, 'w') as stream:
                json.dump(receipt, stream, sort_keys=True)
                stream.write('\n')
        except BaseException:
            return 6  # Receipt could not be retained; metadata may have changed.


if __name__ == '__main__':
    sys.exit(main())