railiance-platform/tests/task_execution_trusted_servers.py

"""Actual v2 provider fixture; no generic policy, fabricated claim or live identity."""
from contextlib import contextmanager
from dataclasses import replace
from datetime import datetime,timedelta,timezone
import base64
import hashlib
import json
import os
from pathlib import Path
import socket
import ssl
import subprocess
import threading
import time
from types import SimpleNamespace
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
from wsgiref.simple_server import make_server

from cryptography import x509
from cryptography.hazmat.primitives import serialization,hashes
from cryptography.hazmat.primitives.asymmetric import ed25519,rsa
from cryptography.x509.oid import NameOID

from task_execution_source_servers import QuietWSGI,private,free_port,http
from task_execution_journal import Binding
from task_execution_native_fixture import FixtureAdmission,NativeFixtureAdapter
import task_execution_plan as plan_source
import task_execution_grant as grant

ACCESS=Path(os.environ.get('TRUSTED_ACCESS_SOURCE_ROOT','/tmp/access-trusted-b2-b3-r3'))
SECRETS=Path(os.environ.get('SECRETS_ENGINE_SOURCE_ROOT','/tmp/netkingdom-secrets-trusted-successor'))
BINARY=Path(os.environ.get('TRUSTED_ACCESS_BINARY','/tmp/netkingdom-access-trusted-r3'))
VERIFIER=Path(os.environ.get('TRUSTED_PDP_VERIFIER','/tmp/secrets-trusted-successor-verifier'))
POLICY_DIGEST=os.environ.get('TRUSTED_POLICY_DIGEST','sha256:de5683e5a34161b9349b10cd8e4abef704e0136e3b14b97719fcd7a79850fc4d')
PRINCIPAL='system:serviceaccount:secrets-engine:secrets-engine'


def certificate(directory):
    key=rsa.generate_private_key(public_exponent=65537,key_size=2048)
    name=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,'localhost')])
    now=datetime.now(timezone.utc)
    cert=(x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(key.public_key())
          .serial_number(x509.random_serial_number()).not_valid_before(now-timedelta(minutes=1))
          .not_valid_after(now+timedelta(hours=1))
          .add_extension(x509.SubjectAlternativeName([x509.DNSName('localhost')]),critical=False)
          .sign(key,hashes.SHA256()))
    ca=directory/'trusted-ca.pem';ca.write_bytes(cert.public_bytes(serialization.Encoding.PEM));ca.chmod(0o600)
    private_key=directory/'trusted-tls.key'
    private_key.write_bytes(key.private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.PKCS8,serialization.NoEncryption()));private_key.chmod(0o600)
    context=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER);context.load_cert_chain(ca,private_key)
    return ca,context


def serve(server,*,tls=None):
    if tls is not None:server.socket=tls.wrap_socket(server.socket,server_side=True)
    thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
    return thread


@contextmanager
def trusted_providers(directory):
    from approval_engine.api import App
    from approval_engine.auth import Identity,StaticTokenAuthenticator
    from approval_engine.store import Engine
    ca,tls=certificate(directory)
    state=SimpleNamespace(principal=PRINCIPAL,caller_valid=True)
    expiry=int(time.time())+600
    encoded=base64.urlsafe_b64encode(json.dumps({'exp':expiry}).encode()).decode().rstrip('=')
    caller='synthetic.'+encoded+'.synthetic';reviewer='synthetic-trusted-reviewer';calls=[]
    class Review(BaseHTTPRequestHandler):
        def log_message(self,*args):pass
        def do_POST(self):
            raw=self.rfile.read(int(self.headers.get('Content-Length','0')));body=json.loads(raw)
            valid=(state.caller_valid and self.path=='/apis/authentication.k8s.io/v1/tokenreviews'
                   and self.headers.get('Authorization')=='Bearer '+reviewer
                   and body.get('spec',{}).get('token')==caller
                   and 'flex-auth' in body.get('spec',{}).get('audiences',[]))
            calls.append(('review',valid))
            response={'status':{'authenticated':valid,'audiences':['flex-auth'],'user':{'username':state.principal}}}
            payload=json.dumps(response).encode();self.send_response(200);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(payload)));self.end_headers();self.wfile.write(payload)
    review=ThreadingHTTPServer(('127.0.0.1',0),Review);review_thread=serve(review,tls=tls)
    def identity(subject,kind,scopes):
        return Identity(subject,'synthetic-whole-source',('approval-engine',),kind,'tenant:platform',
                        frozenset(),frozenset(scopes),{'level':'aal2','methods':['synthetic']},
                        'fixture:whole-source',tenant_source='directory' if kind=='human' else 'registration')
    service='synthetic-whole-service';reader='synthetic-whole-reader';human='synthetic-whole-human'
    authenticator=StaticTokenAuthenticator({service:identity('secrets-engine','service',['approval:create','approval:read','approval:consume']),
        reader:identity('access-source-reader','service',['approval:read']),
        human:identity('whole-human','human',['approval:approve','approval:revoke'])})
    # Explicit synthetic owner observation contract, not an API default change.
    engine=Engine(str(directory/'whole-approval.sqlite'),freshness_ttl=120)
    app=App(engine,authenticator)
    def logged(channel):
        def handle(environ,start_response):
            calls.append((channel,environ['REQUEST_METHOD'],environ['PATH_INFO']))
            return app(environ,start_response)
        return handle
    source=make_server('127.0.0.1',0,logged('source'),handler_class=QuietWSGI)
    consumer=make_server('127.0.0.1',0,logged('consumer'),handler_class=QuietWSGI)
    source_thread=serve(source,tls=tls);consumer_thread=serve(consumer)
    reader_file=private(directory/'approval-reader.token',reader)
    review_file=private(directory/'reviewer.token',reviewer)
    cfg={'source_id':'synthetic-whole-actual-approval','source_revision':'approval-3485415-precision-74eb0ff6-fixture',
         'origin':'https://localhost:'+str(source.server_port),'store_tenant':'tenant:platform',
         'ca_file':str(ca),'token_file':str(reader_file)}
    config=directory/'approval-source.json';private(config,json.dumps(cfg))
    seed_key=ed25519.Ed25519PrivateKey.generate()
    b64=lambda value:base64.urlsafe_b64encode(value).decode().rstrip('=')
    seed=private(directory/'whole-signing.seed',b64(seed_key.private_bytes(serialization.Encoding.Raw,serialization.PrivateFormat.Raw,serialization.NoEncryption())))
    public=b64(seed_key.public_key().public_bytes(serialization.Encoding.Raw,serialization.PublicFormat.Raw))
    registry=directory/'whole-registry.json';registry.write_bytes((ACCESS/'examples/secrets-engine/registry_snapshot.json').read_bytes())
    policy=ACCESS/'examples/secrets-engine-service-task-execution-v2/policy_package.md'
    record=directory/'whole-access-record.jsonl';port=free_port();origin='http://127.0.0.1:'+str(port)
    args=[str(BINARY),'serve','--addr','127.0.0.1:'+str(port),'--registry',str(registry),'--policy',str(policy),
          '--log',str(record),'--envelope-signing-key-file',str(seed),'--envelope-signing-kid','synthetic-whole-key',
          '--caller-auth-mode','enforce','--caller-audience','flex-auth',
          '--caller-kubernetes-url','https://localhost:'+str(review.server_port),
          '--caller-reviewer-token-file',str(review_file),'--caller-ca-file',str(ca),
          '--caller-binding','secrets-engine='+PRINCIPAL,'--approval-source-file',str(config)]
    process=subprocess.Popen(args,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,env={})
    try:
        for _ in range(100):
            if process.poll() is not None:raise RuntimeError('actual trusted Access source exited')
            try:
                with socket.create_connection(('127.0.0.1',port),timeout=.1):break
            except OSError:time.sleep(.025)
        else:raise RuntimeError('actual trusted Access source startup unconfirmed')
        trust={'schema':'secrets-engine.pdp-trust.v1','evaluator':'flex-auth/local','mode':'standalone',
               'policy_package':'secrets-engine.task-execution.service-pilot','policy_version':'v2',
               'policy_package_digest':POLICY_DIGEST,'keys':{'synthetic-whole-key':public}}
        trustfile=private(directory/'whole-pdp-trust.json',json.dumps(trust))
        approval_token=private(directory/'whole-consumer-approval.token',service)
        pdp_token=private(directory/'whole-consumer-pdp.token',caller)
        from task_execution_mediator import ConfirmationConfig
        cfg=SimpleNamespace(approval_url='http://127.0.0.1:'+str(consumer.server_port),approval_token_file=approval_token,
            approval_client_secret_file=None,pdp_url=origin,pdp_token_file=pdp_token,pdp_trust_file=trustfile,pdp_verifier=VERIFIER,
            authorization_subject_id='secrets-engine',authorization_subject_type='service',
            authorization_policy_package=trust['policy_package'],authorization_policy_version='v2',clock_trust_file=None)
        def restart_registry():
            nonlocal process
            process.terminate();process.wait(timeout=5)
            changed=json.loads(registry.read_bytes())
            for subject in changed['subjects']:
                if subject['id']=='secrets-engine':subject['metadata']['description']+=' synthetic revision drift'
            registry.write_text(json.dumps(changed))
            process=subprocess.Popen(args,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,env={})
            for _ in range(100):
                if process.poll() is not None:raise RuntimeError('changed-registry source startup refused')
                try:
                    with socket.create_connection(('127.0.0.1',port),timeout=.1):break
                except OSError:time.sleep(.025)
            else:raise RuntimeError('changed-registry source startup unconfirmed')
        yield SimpleNamespace(cfg=cfg,engine=engine,calls=calls,state=state,record=record,process=process,
                              registry=registry,restart_registry=restart_registry,
                              confirmation=ConfirmationConfig(cfg.approval_url,approval_token),human_token=human,
                              source_config=config,reader_file=reader_file)
    finally:
        process.terminate();process.wait(timeout=5)
        for server in [source,consumer,review]:server.shutdown();server.server_close()
        for thread in [source_thread,consumer_thread,review_thread]:thread.join(5)
        engine.close()


def install_exact_owner_descriptor(owner,*,window_seconds=60):
    """All server startup precedes this exact proposal time; no remint/reissue."""
    native,journal,binding,call,compiler,descriptor=owner
    from secrets_engine.execution_contract import ExecutionContract
    descriptor=json.loads(json.dumps(descriptor))
    descriptor['backend_budget']={'ttl_seconds':30,'uses':12}
    descriptor['operation_deadline']=(datetime.now(timezone.utc)+timedelta(seconds=window_seconds)).replace(microsecond=0).isoformat().replace('+00:00','Z')
    ccr=compiler._derive()[0]
    sources={'engine_dispatcher':SECRETS/'src/secrets_engine/task_execution_handoff.py',
             'platform_mediator':Path(__file__).resolve().parents[1]/'scripts/task_execution_mediator.py',
             'typed_ccr':compiler.validator,'policy_auth':compiler.config.platform_repo/ccr['openbao']['policy_file']}
    descriptor['source_bundle']={role:{'path':str(path),'sha256':plan_source.tagged(path.read_bytes())} for role,path in sources.items()}
    # Exact source descriptor participates in the existing reviewed pure grant
    # projection. Synthetic fixture approval is not a production CCR admission.
    record=json.loads(json.dumps(__import__('yaml').safe_load(compiler.grant_record.read_text())))
    record['execution'],record['openbao']=compiler.grant_effects(compiler.compile(native.meta['request']),descriptor)
    compiler.grant_record.write_text(__import__('yaml').safe_dump(record,sort_keys=False))
    pins=dict(compiler.config.source_pins);pins[str(compiler.grant_record)]=plan_source.tagged(compiler.grant_record.read_bytes())
    compiler=plan_source.Compiler(replace(compiler.config,source_pins=tuple(pins.items())))
    descriptor['grant_immutable_sha256']=grant.immutable_digest(record)
    descriptor=ExecutionContract.parse(descriptor).as_dict();raw=plan_source.encoded(descriptor)
    plan=compiler.bind_execution(compiler.compile(native.meta['request']),descriptor,raw)
    binding=Binding(binding.approval_id,binding.request_digest,binding.operation_id,plan.content_digest)
    admission=replace(native.admission,descriptor=raw)
    native=NativeFixtureAdapter(compiler,plan,journal,binding,admission)
    return native,journal,binding,call,compiler,descriptor