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 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()