#!/usr/bin/env python3
"""MASON-WP-0009-T05 attended, value-free lane verifier; no KV GETs.

The existing operator/ESO policies do NOT grant capabilities-self. An existing
read-only ops-mason-survey grant queries their tokens via sys/capabilities.
This file prepares a founder-run procedure, not a new authorization surface.
"""
import argparse
import hashlib
import json
import os
from pathlib import Path
import subprocess
import selectors
import signal
import sys
import time
import urllib.request
import urllib.error

DATA = 'platform/data/workloads/keycape/secrets-engine-openbao'
OP_ROLE = 'keycape-secrets-engine-openbao-operator-workload-kv-read'
ESO_ROLE = 'external-secrets-keycape-secrets-engine-openbao'
POLICY = 'workload-kv-read-keycape-secrets-engine-openbao-'
PATHS = [DATA, DATA.replace('/data/', '/metadata/'),
         'platform/data/workloads/keycape', 'platform/data/workloads/keycape/',
         DATA + '-other', 'platform/data/workloads/keycape/other',
         'platform/data/workloads/keycape/*', 'sys/policies/acl/ops-mason-survey',
         'auth/token/create', 'auth/token/revoke-self']
BASE = 'http://127.0.0.1:18200'

class Refused(Exception):
    pass

class RecoveredLoginRefused(Refused):
    """Retain only an already recovered session for finally cleanup, never proof."""
    def __init__(self, code, auth):
        super().__init__(code)
        token = auth.get('client_token') if isinstance(auth, dict) else None
        self.token = token if isinstance(token, str) and 0 < len(token) <= 4096 and not any(c.isspace() for c in token) else ''

class Transport(Exception):
    pass

class Api:
    """Hard-coded endpoint allowlist prevents secret GETs, role/policy writes."""
    def __init__(self):
        self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}),
                                                  NoRedirect())
    def call(self, endpoint, token='', body=None):
        allowed = {'auth/token/lookup-self', 'auth/token/revoke-self',
                   'sys/capabilities', 'sys/capabilities-self',
                   'auth/kubernetes/login'}
        if endpoint not in allowed:
            raise Refused('endpoint_not_allowlisted')
        req = urllib.request.Request(BASE + '/v1/' + endpoint,
            data=None if body is None else json.dumps(body).encode(),
            headers={'Content-Type': 'application/json', 'X-Vault-Token': token},
            method='GET' if endpoint == 'auth/token/lookup-self' else 'POST')
        try:
            with self.opener.open(req, timeout=20) as response:
                if endpoint == 'auth/token/revoke-self':
                    # OpenBao's native self-revoke success is 204 with no body.
                    if response.status != 204 or response.read(1):
                        raise Refused('unconfirmed_revoke_response')
                    return {}
                if response.status != 200 or response.headers.get_content_type() != 'application/json':
                    raise Refused('invalid_status_or_content_type')
                length = response.headers.get('Content-Length')
                if length is not None and (not length.isdecimal() or int(length) > 65536):
                    raise Refused('oversized_or_invalid_content_length')
                raw = response.read(65537)
                if len(raw) > 65536:
                    raise Refused('oversized_response')
                result = json.loads(raw)
                if not isinstance(result, dict):
                    raise Refused('invalid_response_envelope')
                if 'errors' in result:
                    if endpoint == 'auth/kubernetes/login':
                        raise RecoveredLoginRefused('native_error_envelope', result.get('auth'))
                    raise Refused('native_error_envelope')
                key = 'auth' if endpoint == 'auth/kubernetes/login' else 'data'
                if not isinstance(result.get(key), dict):
                    raise Refused('invalid_endpoint_response_shape')
                return result
        except urllib.error.HTTPError as error:
            error.close()  # never read/print server errors
            raise Refused('http_' + str(error.code)) from None
        except (OSError, ValueError, urllib.error.URLError):
            raise Transport('transport_or_invalid_response') from None

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

def digest(value):
    return hashlib.sha256(value.encode()).hexdigest()

def accessor_digest(value):
    if not isinstance(value, str) or not 1 <= len(value) <= 256 or any(c.isspace() for c in value):
        raise Refused('missing_or_invalid_native_accessor')
    return digest(value)

def checked_caps(response, expected):
    data = response.get('data', response)
    # Only batch-shaped named results accepted, never substitute a default.
    if not isinstance(data, dict) or not all(path in data for path in expected):
        raise Refused('incomplete_capability_matrix')
    result = {}
    for path, caps in expected.items():
        actual = data[path]
        if not isinstance(actual, list) or sorted(actual) != sorted(caps):
            raise Refused('capability_mismatch')
        result[path] = actual
    return result

def survey(api, token):
    info = api.call('auth/token/lookup-self', token)['data']
    if info.get('identity_policies'):
        raise Refused('query_identity_policy_additions')
    policies = set(info.get('policies', []))
    if policies not in ({'ops-mason-survey'}, {'ops-mason-survey', 'default'}):
        raise Refused('query_policy_mismatch')
    if info.get('display_name') != 'token-claude-MASON-WP-0009-T05' or (type(info.get('ttl')) is not int or not 0 < info['ttl'] <= 2700):
        raise Refused('query_identity_or_ttl_mismatch')
    expected = {DATA: ['deny'], 'sys/policies/acl/ops-mason-survey': ['deny'],
                'auth/token/create': ['deny'], 'sys/capabilities': ['update']}
    checked_caps(api.call('sys/capabilities-self', token, {'paths': list(expected)}), expected)
    return info

def safe_env(home):
    # Never inherit credential variables, token helpers, proxies, CLI plugins.
    return {'PATH': '/home/worsch/.local/bin:/usr/local/bin:/usr/bin:/bin', 'HOME': home,
            'BAO_ADDR': BASE, 'VAULT_ADDR': BASE,
            'DISPLAY': os.environ.get('DISPLAY', ''),
            'WSL_INTEROP': os.environ.get('WSL_INTEROP', ''),
            'WSL_DISTRO_NAME': os.environ.get('WSL_DISTRO_NAME', '')}

def run(argv, env):
    # Capture each stream with a real memory bound while the child runs; never
    # retain an unbounded PIPE response and only check its length afterwards.
    process = None
    try:
        process = subprocess.Popen(argv, env=env, stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True)
        buffers = {process.stdout: bytearray(), process.stderr: bytearray()}
        deadline = time.monotonic() + 300
        with selectors.DefaultSelector() as selector:
            for stream in buffers:
                selector.register(stream, selectors.EVENT_READ)
            while selector.get_map():
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    raise Transport('native_transport_or_timeout')
                for key, _ in selector.select(min(remaining, 1)):
                    chunk = os.read(key.fileobj.fileno(), 8192)
                    if not chunk:
                        selector.unregister(key.fileobj)
                        continue
                    target = buffers[key.fileobj]
                    if len(target) + len(chunk) > 65536:
                        raise Refused('oversized_native_response')
                    target.extend(chunk)
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise Transport('native_transport_or_timeout')
        code = process.wait(timeout=remaining)
        return code, bytes(buffers[process.stdout]), bytes(buffers[process.stderr])
    except (OSError, subprocess.TimeoutExpired):
        raise Transport('native_transport_or_timeout') from None
    finally:
        if process is not None:
            if process.poll() is None:
                try:
                    os.killpg(process.pid, signal.SIGKILL)
                except ProcessLookupError:
                    pass
                process.wait()
            process.stdout.close()
            process.stderr.close()

def oidc(env, runner=run):
    # -no-store is mandatory; output/errors stay inside the founder's process.
    code, raw, err = runner(['/home/worsch/.local/bin/bao', 'login', '-no-store', '-format=json',
                             '-method=oidc', '-path=netkingdom', 'role=' + OP_ROLE], env)
    if code:
        # A candidate refusal still needs independent issuer + correlated
        # OpenBao owner evidence. HTTP 403, cancellation or timeout never pass.
        claim_refusal = b'claim "groups" does not match' in err or b'claim "groups" not found' in err
        return None, 'group_refusal_candidate' if claim_refusal else 'unproved_login_failure'
    try:
        result = json.loads(raw)
        if not isinstance(result, dict) or not isinstance(result.get('auth'), dict):
            raise Refused('malformed_native_login')
        if 'errors' in result:
            raise RecoveredLoginRefused('native_error_envelope', result['auth'])
        return result['auth'], 'login_succeeded'
    except (ValueError, KeyError, TypeError):
        raise Refused('malformed_native_login') from None

def eso(api, runner=run):
    # Existing owner SSH certificate, no kubeconfig/Secret reads. Separate
    # TokenRequest (10m, openbao audience); no ESO controller session revoked.
    env = {'PATH': '/usr/local/bin:/usr/bin:/bin', 'HOME': os.path.expanduser('~')}
    rc, out, _ = runner(['ssh', '-oBatchMode=yes', 'railiance01',
                         'kubectl get nodes -o json'], env)
    if rc:
        raise Transport('cluster_identity_unproved')
    try:
        nodes = json.loads(out)['items']
        addresses = {a['address'] for n in nodes for a in n['status']['addresses']}
        if len(nodes) != 1 or '92.205.62.239' not in addresses:
            raise Refused('wrong_cluster')
    except (KeyError, TypeError, ValueError):
        raise Refused('invalid_cluster_identity') from None
    rc, jwt, _ = runner(['ssh', '-oBatchMode=yes', 'railiance01',
        'kubectl -n external-secrets create token external-secrets --audience=openbao --duration=10m'], env)
    if rc or not jwt.strip():
        raise Transport('tokenrequest_failed')
    return api.call('auth/kubernetes/login', body={'role': ESO_ROLE,
                      'jwt': jwt.decode().strip()})['auth']

def lane_matrix(api, query, auth, mode):
    token = auth.get('client_token')
    if not isinstance(token, str) or not token or any(c.isspace() for c in token):
        raise Refused('malformed_lane_token')
    metadata = auth.get('metadata', {})
    expected_role = OP_ROLE if mode == 'operator' else ESO_ROLE
    if metadata.get('role') != expected_role:
        raise Refused('lane_role_mismatch')
    if mode == 'eso' and (metadata.get('service_account_name') != 'external-secrets' or
                          metadata.get('service_account_namespace') != 'external-secrets'):
        raise Refused('eso_identity_mismatch')
    if auth.get('identity_policies'):
        raise Refused('lane_identity_policy_additions')
    expected_policy = POLICY + ('operator' if mode == 'operator' else 'eso')
    if set(auth.get('policies', [])) != {expected_policy}:
        raise Refused('lane_policy_mismatch')
    if (type(auth.get('lease_duration')) is not int or not 0 < auth['lease_duration'] <= (300 if mode == 'operator' else 900)):
        raise Refused('lane_ttl_mismatch')
    expected = {p: ['deny'] for p in PATHS}
    expected[DATA] = ['read']
    expected['auth/token/revoke-self'] = ['update']
    return checked_caps(api.call('sys/capabilities', query,
                        {'token': token, 'paths': PATHS}), expected)

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('mode', choices=['operator', 'eso', 'nonmember'])
    parser.add_argument('--survey-token-fd', type=int, required=True)
    parser.add_argument('--approval-reference', required=True)
    parser.add_argument('--approved-source-sha256', required=True)
    parser.add_argument('--receipt', type=Path, required=True)
    parser.add_argument('--attended', action='store_true')
    args = parser.parse_args()
    if not args.attended or not sys.stdin.isatty() or args.survey_token_fd < 3:
        parser.error('founder terminal attendance and separate survey token FD required')
    if any(os.environ.get(k) for k in ('BAO_TOKEN', 'VAULT_TOKEN', 'BAO_NAMESPACE', 'VAULT_NAMESPACE')):
        parser.error('ambient authority refused')
    source_sha256 = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
    if args.approved_source_sha256 != source_sha256:
        parser.error('approved source digest differs; no session requested')
    # Reservation prevents overwrites/symlink substitution before credentials.
    fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
    receipt = {'schema': 'mason-client-custody-native/v1', 'task': 'MASON-WP-0009-T05',
        'mode': args.mode, 'source_sha256': source_sha256,
        'approval_reference': args.approval_reference, 'at': int(time.time()),
        'secret_reads': 0, 'result': 'unproved', 'owner_acceptance': 'pending',
        'query_revoke': False, 'lane_revoke': False}
    api = Api()
    query = ''
    query_valid = False
    lane = ''
    try:
        with os.fdopen(args.survey_token_fd, 'r') as handle:
            query = handle.read(4097).strip()
        if not query or len(query) > 4096 or any(c.isspace() for c in query):
            raise Refused('malformed_query_token')
        query_info = survey(api, query)
        query_valid = True
        receipt['query_accessor_sha256'] = accessor_digest(query_info.get('accessor'))
        if args.mode == 'eso':
            auth = eso(api)
        else:
            import tempfile
            with tempfile.TemporaryDirectory(prefix='mason-oidc-') as home:
                auth, result = oidc(safe_env(home))
            receipt['native_login_result'] = result
            if auth is None:
                # Never pronounce the nonmember check passed from CLI text.
                raise Refused(result)
        lane = auth.get('client_token', '')
        receipt['lane_accessor_sha256'] = accessor_digest(auth.get('accessor'))
        if args.mode == 'nonmember':
            raise Refused('nonmember_obtained_token')
        receipt['capabilities'] = lane_matrix(api, query, auth, args.mode)
        receipt['result'] = 'capability_matrix_observed'
    except RecoveredLoginRefused as error:
        lane = error.token
        receipt['failure_class'] = str(error)
    except (Refused, Transport) as error:
        receipt['failure_class'] = str(error)
    except Exception:
        receipt['failure_class'] = 'invalid_or_unexpected_response'
    finally:
        if lane:
            try:
                api.call('auth/token/revoke-self', lane, {})
                receipt['lane_revoke'] = True
            except Exception:
                receipt['result'] = 'unproved'
        if query_valid:
            try:
                api.call('auth/token/revoke-self', query, {})
                receipt['query_revoke'] = True
            except Exception:
                receipt['result'] = 'unproved'
        query = lane = ''
        with os.fdopen(fd, 'w') as handle:
            json.dump(receipt, handle, indent=2, sort_keys=True)
            handle.write('\n')
    return 0 if receipt['result'] == 'capability_matrix_observed' else 1

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