informed-decision/tests/browser_fixture.py

"""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 contextlib import ExitStack
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), ExitStack() as components:
        controller.policy=policy
        lifecycle = None
        if os.environ.get('INFD_FLEX_AUTH_SOURCE'):
            from informed_decision.lifecycle import LifecycleRepository
            from informed_decision.lifecycle_service import LifecycleService
            lifecycle_directory = directory/'lifecycle-policy'; lifecycle_directory.mkdir()
            lifecycle_policy, _ = components.enter_context(flex_component(
                os.environ['INFD_FLEX_AUTH_BINARY'], lifecycle_directory, key,
                fixture=Path(os.environ['INFD_FLEX_AUTH_SOURCE'])/'examples/informed-decision-lifecycle'))
            lifecycle = LifecycleService(LifecycleRepository(controller.store), lifecycle_policy)
        pump=AuditPump(controller.store,sink,lifecycle_delivery=lifecycle is not None);pump.tick()
        assert pump.ready()
        app=App(login,controller,readiness=pump.ready,lifecycle=lifecycle)
        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,
                'lifecycle':lifecycle is not None,'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()