"""Local-only browser exercise over actual components and synthetic identity. Run with all three INFD_* component source/binary variables, and pass an output directory. The fixture controls below are NOT part of informed_decision.web. """ import json import os from pathlib import Path import sys import threading import base64 import hashlib import ssl from datetime import datetime, timedelta, timezone from urllib.parse import parse_qs, urlsplit from wsgiref.simple_server import WSGIRequestHandler, make_server from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.x509.oid import NameOID from informed_decision.oidc import KeyCapeLogin from informed_decision.runtime import AuditPump from informed_decision.web import App from test_browser_auth import ISSUER, IssuerFixture, signing_key from test_approval_component import component from test_durable_component import receiver from test_review_controller import review from review_components import flex_component def main(): directory=Path(sys.argv[1]);directory.mkdir(mode=0o700,parents=True,exist_ok=True) key=signing_key.__wrapped__() class Quiet(WSGIRequestHandler): def log_message(self,*args):pass tls_server=make_server('127.0.0.1',0,lambda e,s:[],handler_class=Quiet) public_origin=f'https://decisions.coulomb.social:{tls_server.server_port}' issuer_origin=f'https://keycape.test:{tls_server.server_port}' # Test-only ports, real HTTPS and redirects. Exact production strings stay # pinned by the separate registration/login contract tests. import informed_decision.oidc as oidc_module import informed_decision.web as web_module import test_browser_auth as browser_auth import test_approval_component as approval_component oidc_module.ORIGIN=web_module.ORIGIN=public_origin oidc_module.CALLBACK=browser_auth.CALLBACK=public_origin+'/auth/callback' browser_auth.ISSUER=approval_component.ISSUER=issuer_origin now=datetime.now(timezone.utc) name=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,'Browser fixture')]) certificate=(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('decisions.coulomb.social'),x509.DNSName('keycape.test')]),critical=False).sign(key,hashes.SHA256())) cert_path=directory/'browser-cert.pem';cert_path.write_bytes(certificate.public_bytes(serialization.Encoding.PEM)) key_path=directory/'browser-key.pem';key_path.write_bytes(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,serialization.NoEncryption()));key_path.chmod(0o600) context=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER);context.load_cert_chain(cert_path,key_path) tls_server.socket=context.wrap_socket(tls_server.socket,server_side=True) spki=base64.b64encode(hashlib.sha256(key.public_key().public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo)).digest()).decode() engine_fixture=component.__wrapped__(key,directory);engine_parts=next(engine_fixture) sink,receiver_transport,backend=receiver.__wrapped__(directory) controller,_,_,memo,engine,transport=review.__wrapped__(engine_parts,directory) issuer=IssuerFixture(key);login=KeyCapeLogin(issuer_origin,transport=issuer) original_start=login.start def start(): url,browser=original_start() issuer.params={k:v[0] for k,v in parse_qs(urlsplit(url).query).items()} return url,browser login.start=start with flex_component(os.environ['INFD_FLEX_AUTH_BINARY'],directory,key) as (policy,calls): controller.policy=policy pump=AuditPump(controller.store,sink);pump.tick() assert pump.ready() app=App(login,controller,readiness=pump.ready) def fixture(environ,start_response): path=environ.get('PATH_INFO') if path=='/authorize': params={k:v[0] for k,v in parse_qs(environ.get('QUERY_STRING','')).items()} assert params['code_challenge_method']=='S256' assert params['redirect_uri']==oidc_module.CALLBACK start_response('303 See Other',[('Location',oidc_module.CALLBACK+'?state='+params['state']+'&code=synthetic-browser-code')]) return [b''] if path=='/_fixture/status': pump.tick() intents=[controller.store.intent_for('fixture','human-fixture')] payload={'entries':len(engine.get('fixture').entries),'intents':intents, 'outbox_states':[r['state'] for r in controller.store.outbox()], 'policy_outcomes':[r['outcome'] for r in controller.store.policy_observations()], 'decision_attributable':[r['decision_attributable'] for r in controller.store.policy_observations()], 'entry_posts':sum(m=='POST' for m,_ in transport.calls), 'consume_calls':sum(url.endswith('/consume') for _,url in transport.calls), 'tokenreviews':len(calls)} raw=json.dumps(payload).encode();start_response('200 OK',[('Content-Type','application/json')]);return [raw] if path=='/_fixture/refuse-caller': policy.token_provider=lambda:'synthetic-wrong-caller' start_response('200 OK',[('Content-Type','application/json')]);return [b'{}'] if path=='/_fixture/stop': threading.Thread(target=server.shutdown,daemon=True).start() start_response('200 OK',[('Content-Type','application/json')]);return [b'{}'] return app(environ,start_response) tls_server.set_app(fixture) tls_thread=threading.Thread(target=tls_server.serve_forever,daemon=True);tls_thread.start() with make_server('127.0.0.1',0,fixture,handler_class=Quiet) as server: (directory/'server.json').write_text(json.dumps({'origin':f'http://127.0.0.1:{server.server_port}', 'public_origin':public_origin,'issuer':issuer_origin,'certificate_spki':spki,'memo_id':memo.id,'fixture_only':True})+'\n') print('Browser fixture ready',flush=True) try:server.serve_forever() except KeyboardInterrupt:pass finally: tls_server.shutdown();tls_server.server_close();tls_thread.join(timeout=5) engine_fixture.close() if __name__=='__main__':main()