test-driver/tests/test_http_boundaries.py

"""Authenticated traffic stays on origin, including standalone descendants."""
from contextlib import contextmanager
from dataclasses import replace
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
import json

import pytest

from scenarios.alice_bob_carol import build, USE_CASE
from testdriver import Runner, Verdict, SurfaceNotPermitted
from testdriver.browser import Session
from testdriver.classification import classify
from testdriver.crystallization import Trajectory, generate_test_module, assess_stability
from testdriver.http import _origin


@contextmanager
def server():
    receipts = []
    routes = {}

    class Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            receipts.append((self.path, self.headers.get('Authorization')))
            code, target = routes.get(self.path, (200, None))
            self.send_response(code)
            if target:
                self.send_header('Location', target)
            self.end_headers()

        do_POST = do_GET

        def log_message(self, *args):
            pass

    http = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
    thread = Thread(target=lambda: http.serve_forever(poll_interval=0.01), daemon=True)
    thread.start()
    try:
        yield f'http://127.0.0.1:{http.server_port}', routes, receipts
    finally:
        http.shutdown()
        http.server_close()
        thread.join()


@pytest.fixture(params=['session', 'generated', 'checked_in'])
def post(request):
    if request.param == 'session':
        return lambda base, path: Session(base, 'synthetic-test-token').post_form(path, {})[0]
    if request.param == 'checked_in':
        from crystallized.test_grant_access import _post
    else:
        source = generate_test_module(
            trajectory=Trajectory('s2', 'grant_access', 'browser', '/grant', ()),
            action_args={}, ancestor_id='a', ancestor_maturity='T1', descendant_id='b',
            runs=3, sut_version='test', claims=USE_CASE.claims[:1],
            claims_module='scenarios.alice_bob_carol')
        namespace = {}
        exec(compile(source, '<generated>', 'exec'), namespace)
        _post = namespace['_post']
    return lambda base, path: _post(base, 'synthetic-test-token', path, {})


@pytest.mark.parametrize('network_path', [False, True])
def test_external_targets_are_rejected_before_any_credential_is_sent(post, network_path):
    with server() as (base, _, original), server() as (other, _, receiver):
        target = other + '/grant'
        if network_path:
            target = target.removeprefix('http:')
        with pytest.raises((SurfaceNotPermitted, ValueError), match="authenticated"):
            post(base, target)
        assert receiver == original == []


@pytest.mark.parametrize('code', [301, 302, 303, 307, 308])
def test_redirects_cannot_forward_credentials_to_another_origin(post, code):
    with server() as (base, routes, original), server() as (other, _, receiver):
        routes['/start'] = (code, other + '/capture')
        with pytest.raises((SurfaceNotPermitted, ValueError), match="authenticated"):
            post(base, '/start')
        assert original == [('/start', 'Bearer synthetic-test-token')]
        assert receiver == []


def test_same_origin_targets_and_redirects_still_work(post):
    with server() as (base, routes, receipts):
        routes['/start'] = (302, '/next')
        routes['/next'] = (302, base + '/finish')
        assert post(base, base + '/start') == 200
        assert receipts == [(path, 'Bearer synthetic-test-token')
                            for path in ('/start', '/next', '/finish')]


@pytest.mark.parametrize('target', ['https://example.test/path', 'file:///tmp/nope',
                                    'http://user:pass@example.test/path'])
def test_scheme_changes_and_userinfo_are_not_allowed(post, target):
    with pytest.raises((SurfaceNotPermitted, ValueError), match="authenticated"):
        post('http://example.test', target)


def test_default_ports_and_host_case_have_normal_origin_semantics():
    assert _origin('https://EXAMPLE.test/path') == _origin('https://example.test:443/')
    assert _origin('http://example.test') == _origin('http://example.test:80')
    assert _origin('http://example.test:0') != _origin('http://example.test')


def test_runner_rejects_reported_surface_even_when_driver_omits_its_guard():
    packs = []
    for _ in range(3):
        world, driver, observer, asset, oracle = build()
        calls = []

        class UncheckedDriver:
            def realize(self, actor, action):
                calls.append(action.name)
                return driver.realize(actor, replace(action, permitted_surfaces=frozenset({'api'})))

        asset.scenario = replace(asset.scenario, steps=tuple(
            replace(step, action=replace(step.action, permitted_surfaces=frozenset({'browser'})))
            for step in asset.scenario.steps))
        result = Runner(world, UncheckedDriver(), observer, oracle).run(asset)
        assert result.verdict is Verdict.INCONCLUSIVE
        assert len(calls) == 1
        assert all(j.verdict is Verdict.INCONCLUSIVE for j in result.judgments)
        assert any(o.kind == 'surface_violation' for o in result.evidence.observations)
        packs.append(json.loads(result.evidence.to_json()))
    assert not classify(packs[0], packs[1]).safe_to_accept
    assert not assess_stability(packs).stable