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
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import sys

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:
        # Only a structured API 404 is absence. Denial/transport failure is not.
        if missing_ok and result.returncode == 2 and b'Code: 404' in result.stderr:
            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):
    current_policy = command(['read', '-format=json', 'sys/policies/acl/' + POLICY], missing_ok=True)
    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:
        command(['write', '-format=json', 'sys/policies/acl/' + POLICY, '-'],
                json.dumps({'policy': policy}).encode())
    if current_role is None:
        command(['write', '-format=json', ROLE, '-'], json.dumps(payload).encode())
    verified_policy = command(['read', '-format=json', 'sys/policies/acl/' + POLICY])
    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():
    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
        if (Path.home().parent.name != '.warden-attended-login'
                or os.getenv('BAO_TOKEN') or os.getenv('VAULT_TOKEN')):
            raise Refused(2)
        policy, payload = source(args.expected_ccr_sha256)
        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)
        return 0
    except Refused as exc:
        return exc.code
    except BaseException:
        return 2


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