key-cape/scripts/openbao_negative_private_launcher.py

#!/usr/bin/env python3
"""Founder-held, source-pinned direct SSH launcher; no credentials on normal IO.
Ready marker is a procedural readiness check, NEVER authenticated live authority.
Use only after exact source/host/transport/client/custody/file allowance admission.
"""
import argparse
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import re
import secrets
import shlex
import subprocess
import stat
import time

ADAPTER = Path(__file__).with_name('openbao_negative_private_handoff.py')
spec = importlib.util.spec_from_file_location('private_handoff', ADAPTER)
handoff = importlib.util.module_from_spec(spec)
spec.loader.exec_module(handoff)
PROPOSAL = 'KEY-WP-0039-T02/attended-lifecycle-v9-ready-input'
TARGET = 'sso/keycape'
PREPARATION_TOTAL_SECONDS = 5
MAX_PUBLIC_BYTES = 65536
MAX_SOURCE_BYTES = 262144
CONTRACT_KEYS = {'schema','proposal_id','target','remote_host','remote_wrapper_path',
    'remote_wrapper_sha256','remote_child_path','remote_child_sha256','remote_envelope_path',
    'local_envelope_path','remote_receipt_path','founder_decision_id',
    'private_tmpfs_file_allowance_receipt','local_launcher_sha256','local_adapter_sha256'}

def read_stable(path, limit):
    # Always called under the shared preparation timer in the actual CLI graph.
    # NONBLOCK + regular-file check refuses FIFO/device. Resolution refuses
    # observed symlink ancestors; metadata compares detect observed source drift.
    p = Path(path)
    handoff.need(p.is_absolute() and p.resolve() == p, 'source_regular_path_required')
    fd = os.open(p,os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
    try:
        before = os.fstat(fd)
        handoff.need(stat.S_ISREG(before.st_mode) and before.st_nlink == 1
                     and not stat.S_IMODE(before.st_mode) & 0o022
                     and 0 < before.st_size <= limit, 'source_file_invalid')
        chunks,size = [],0
        while size <= limit:
            chunk = os.read(fd,min(4096,limit+1-size))
            if not chunk:
                break
            size += len(chunk)
            handoff.need(size <= limit,'source_file_oversized')
            chunks.append(chunk)
        after = os.fstat(fd)
        fields = lambda v:(v.st_dev,v.st_ino,v.st_uid,v.st_mode,v.st_nlink,
                         v.st_size,v.st_mtime_ns,v.st_ctime_ns)
        handoff.need(fields(before) == fields(after) and size == before.st_size,
                     'source_file_changed')
        return b''.join(chunks)
    finally:
        os.close(fd)

def digest(path):
    return hashlib.sha256(read_stable(path,MAX_SOURCE_BYTES)).hexdigest()

def remaining_preparation(preparation_deadline):
    handoff.need(handoff.finite(preparation_deadline),'preparation_window_invalid')
    remaining = preparation_deadline-time.monotonic()
    handoff.need(0 < remaining <= PREPARATION_TOTAL_SECONDS,'preparation_window_expired')
    return remaining

def pin(value):
    return isinstance(value,str) and re.fullmatch('[0-9a-f]{64}',value) is not None

def remote_path(value):
    handoff.need(isinstance(value,str) and len(value) <= 256
                 and re.fullmatch('/[A-Za-z0-9_./-]+',value) is not None
                 and '..' not in Path(value).parts and not value.endswith('/'), 'remote_path_invalid')
    return value

def read_public(path):
    raw = read_stable(path,MAX_PUBLIC_BYTES)
    obj = handoff.decode(raw)
    handoff.need(isinstance(obj,dict),'source_contract_invalid')
    return raw,obj

def validate_contract(contract):
    handoff.need(isinstance(contract,dict) and set(contract) == CONTRACT_KEYS
                 and contract['schema'] == 'key39-private-launch-contract/v1'
                 and contract['proposal_id'] == PROPOSAL and contract['target'] == TARGET
                 and contract['remote_host'] == 'railiance01','source_contract_invalid')
    for field in ('remote_wrapper_path','remote_child_path','remote_envelope_path','remote_receipt_path'):
        remote_path(contract[field])
    handoff.need(len({contract[key] for key in ('remote_wrapper_path','remote_child_path',
                 'remote_envelope_path','remote_receipt_path')}) == 4,'remote_paths_must_differ')
    for field in ('remote_wrapper_sha256','remote_child_sha256','local_launcher_sha256','local_adapter_sha256'):
        handoff.need(pin(contract[field]),'source_pin_required')
    handoff.need(digest(__file__) == contract['local_launcher_sha256']
                 and digest(ADAPTER) == contract['local_adapter_sha256'],'local_source_pin_changed')
    handoff.need(all(isinstance(contract[k],str) and 0 < len(contract[k]) <= 256
                 for k in ('founder_decision_id','private_tmpfs_file_allowance_receipt')),
                 'founder_receipt_required')
    raw,envelope = read_public(contract['local_envelope_path'])
    handoff.need(envelope.get('proposal_id') == PROPOSAL and envelope.get('target') == TARGET
                 and envelope.get('disposition') == 'approved'
                 and envelope.get('founder_decision_id') == contract['founder_decision_id'],
                 'source_envelope_mismatch')
    expiry = envelope.get('valid_until_epoch')
    window = envelope.get('maximum_session_seconds')
    handoff.need(handoff.finite(expiry) and type(window) is int and 420 <= window <= 900
                 and 0 < expiry-time.time() <= window,'source_window_invalid')
    health = envelope.get('bao_health_identity_sha256')
    handoff.need(pin(health),'source_health_pin_required')
    hashes = envelope.get('source_hashes',{})
    handoff.need(isinstance(hashes,dict) and
                 hashes.get(contract['remote_child_path']) == contract['remote_child_sha256'] and
                 hashes.get(contract['remote_wrapper_path']) == contract['remote_wrapper_sha256'],
                 'remote_source_pin_missing')
    owners = envelope.get('owner_acknowledgments',{})
    handoff.need(isinstance(owners,dict) and all(isinstance(owners.get(k),str) and owners[k]
                 for k in ('key-cape','net-kingdom','railiance-platform')),'owner_source_receipts_required')
    return hashlib.sha256(raw).hexdigest(),envelope

def launch_and_deliver(contract,human_path,agent_path,*,consume_owner_files=False,
                       preparation_deadline=None):
    receipt = {'native_accepted':False,'authenticated_authority_established_by_helper':False,
               'child_ready_matched':False,'private_input_delivery_confirmed':False,
               'consumer_process_completed':False,'consumer_outcome_unknown':False}
    process = None
    owned_fds = set()
    if preparation_deadline is None:
        preparation_deadline = time.monotonic() + PREPARATION_TOTAL_SECONDS
    try:
        with handoff.deadline(remaining_preparation(preparation_deadline)):
            envelope_hash,envelope = validate_contract(contract)
            nonce = secrets.token_hex(32)
            expected = {'session_nonce':nonce,'proposal_id':PROPOSAL,'target':TARGET,
                'executor_sha256':contract['remote_child_sha256'],'envelope_sha256':envelope_hash,
                'bao_health_identity_sha256':envelope['bao_health_identity_sha256'],
                'valid_until_epoch':envelope['valid_until_epoch']}
            command = ['python3',contract['remote_wrapper_path'],
                '--child',contract['remote_child_path'],'--child-sha256',contract['remote_child_sha256'],
                '--envelope',contract['remote_envelope_path'],'--envelope-sha256',envelope_hash,
                '--receipt',contract['remote_receipt_path'],'--session-nonce',nonce]
            argv = ['ssh','-T','-oBatchMode=yes','-oStrictHostKeyChecking=yes',
                '-oForwardAgent=no','-oClearAllForwardings=yes','-oConnectTimeout=5',
                '-oServerAliveInterval=5','-oServerAliveCountMax=1',
                'railiance01',shlex.join(command)]
            # Do not pass arbitrary confidential producer environment to SSH.
            allowed = ('PATH','HOME','USER','LOGNAME','LANG','LC_ALL','SSH_AUTH_SOCK')
            environment = {key:os.environ[key] for key in allowed if key in os.environ}
            process = subprocess.Popen(argv,stdin=subprocess.PIPE,stdout=subprocess.PIPE,
                stderr=subprocess.DEVNULL,env=environment,close_fds=True)
        remaining_preparation(preparation_deadline)
        # Distinct process-owned pipes; no arbitrary caller ready file or boolean.
        write_fd = os.dup(process.stdin.fileno()); owned_fds.add(write_fd)
        ready_fd = os.dup(process.stdout.fileno()); owned_fds.add(ready_fd)
        process.stdin.close()
        # Close parent's duplicate read endpoint; only owned ready_fd is used.
        process.stdout.close()
        delivery = handoff.deliver(human_path,agent_path,write_fd,ready_fd=ready_fd,
            expected=expected,consume_owner_files=consume_owner_files,
            budget=remaining_preparation(preparation_deadline))
        owned_fds.clear()  # deliver owns and closes both endpoints
        receipt.update(delivery)
        # Private bytes may already have reached the remote child even if local
        # write completion/EOF confirmation failed. Local SSH teardown proves no
        # remote no-effects or cleanup outcome; never clear this unknown here.
        if delivery['private_input_delivery_unknown']:
            receipt['consumer_outcome_unknown'] = True
            receipt['consumer_process_completed'] = False
        if not delivery['private_input_delivery_confirmed']:
            return receipt
        # Value-free exit status only; child receipt and native acceptance remain owner duties.
        wait_budget = max(0,min(envelope['valid_until_epoch']-time.time(),
                               envelope['maximum_session_seconds']-180)) + 330
        try:
            rc = process.wait(timeout=wait_budget)
        except subprocess.TimeoutExpired:
            receipt.update(failure_code='consumer_completion_unconfirmed',consumer_outcome_unknown=True)
            return receipt
        receipt['consumer_process_completed'] = rc == 0
        receipt['consumer_outcome_unknown'] = rc != 0
        if rc != 0:
            receipt['failure_code'] = 'consumer_completion_unconfirmed'
    except handoff.Refused as error:
        receipt['failure_code'] = str(error)
    except Exception:
        receipt['failure_code'] = 'private_launcher_incomplete'
        if receipt['private_input_delivery_confirmed']:
            receipt['consumer_outcome_unknown'] = True
    finally:
        for fd in owned_fds:
            try:
                os.close(fd)
            except OSError:
                pass
        if process is not None:
            # Local teardown never claims remote process/custody reconciliation.
            try:
                if process.poll() is None:
                    process.terminate()
                    try:
                        process.wait(timeout=1)
                    except subprocess.TimeoutExpired:
                        process.kill()
                        process.wait(timeout=1)
                for stream in (process.stdin,process.stdout):
                    if stream is not None:
                        stream.close()
            except Exception:
                receipt['transport_cleanup_unconfirmed'] = True
                receipt['consumer_outcome_unknown'] = True
    return receipt

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--contract',required=True)
    parser.add_argument('--human-file',required=True)
    parser.add_argument('--agent-file',required=True)
    parser.add_argument('--consume-owner-files',action='store_true')
    args = parser.parse_args()
    preparation_deadline = time.monotonic() + PREPARATION_TOTAL_SECONDS
    try:
        with handoff.deadline(remaining_preparation(preparation_deadline)):
            _,contract = read_public(args.contract)
        receipt = launch_and_deliver(contract,args.human_file,args.agent_file,
            consume_owner_files=args.consume_owner_files,preparation_deadline=preparation_deadline)
    except handoff.Refused as error:
        receipt = {'native_accepted':False,'private_input_delivery_confirmed':False,
                   'failure_code':str(error)}
    except Exception:
        receipt = {'native_accepted':False,'private_input_delivery_confirmed':False,
                   'failure_code':'private_launcher_contract_invalid'}
    print(json.dumps(receipt,separators=(',',':')))
    return 0 if receipt.get('consumer_process_completed') is True else 1

if __name__ == '__main__':
    raise SystemExit(main())