tegwick
c80a944105
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
122 lines
4.9 KiB
Python
122 lines
4.9 KiB
Python
"""Exercise the real HTTP boundary with synthetic credentials and local servers."""
|
|
from contextlib import contextmanager
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from threading import Thread
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from secrets_engine import approval_consume
|
|
from secrets_engine.decision_check import check_decision
|
|
from secrets_engine.errors import DecisionError
|
|
|
|
|
|
@contextmanager
|
|
def server(*, status=200, body=b'{}', location=None):
|
|
seen = []
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def handle_request(self):
|
|
# Record only a boolean about our synthetic credential.
|
|
seen.append(self.headers.get('Authorization') == 'Bearer synthetic-pdp-credential')
|
|
self.rfile.read(int(self.headers.get('Content-Length', '0')))
|
|
self.send_response(status)
|
|
if location:
|
|
self.send_header('Location', location)
|
|
self.send_header('Content-Length', str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
do_POST = do_GET = handle_request
|
|
|
|
def log_message(self, *_args):
|
|
pass
|
|
|
|
httpd = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
|
|
thread = Thread(target=httpd.serve_forever, kwargs={'poll_interval': 0.01}, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield f'http://127.0.0.1:{httpd.server_port}', seen
|
|
finally:
|
|
httpd.shutdown()
|
|
httpd.server_close()
|
|
thread.join(timeout=2)
|
|
|
|
|
|
@pytest.fixture
|
|
def credential(tmp_path):
|
|
path = tmp_path / 'synthetic.token'
|
|
path.write_text('synthetic-pdp-credential')
|
|
path.chmod(0o600)
|
|
return path
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_proxies(monkeypatch):
|
|
# Tests must not contact proxies configured on the developer's machine.
|
|
import os
|
|
import urllib.request
|
|
monkeypatch.setattr(urllib.request, "_opener", None)
|
|
for name in os.environ:
|
|
if name.lower().endswith('_proxy'):
|
|
monkeypatch.delenv(name)
|
|
|
|
|
|
@pytest.mark.parametrize('status', [301, 302, 303, 307, 308])
|
|
def test_redirect_never_contacts_another_origin(credential, status):
|
|
with server() as (destination, received):
|
|
with server(status=status, location=destination + '/sink') as (source, sent):
|
|
with pytest.raises(DecisionError):
|
|
check_decision(allow_unsigned_test_fixture=True, base_url=source, token_file=credential, request={})
|
|
assert sent == [True]
|
|
assert received == []
|
|
|
|
|
|
def test_environment_proxy_never_receives_request(credential, monkeypatch):
|
|
with server() as (proxy, proxied):
|
|
with server() as (source, direct):
|
|
monkeypatch.setenv('http_proxy', proxy)
|
|
monkeypatch.setenv('no_proxy', '')
|
|
assert check_decision(allow_unsigned_test_fixture=True, base_url=source, token_file=credential, request={}) == {}
|
|
assert direct == [True]
|
|
assert proxied == []
|
|
|
|
|
|
from tests.signature_support import signature_tools, admit_tools
|
|
|
|
|
|
def test_approval_flow_uses_the_same_safe_transport(credential, monkeypatch, signature_tools):
|
|
observation = SimpleNamespace(request={})
|
|
monkeypatch.setattr(approval_consume, 'resolve_approval_observation', lambda *a, **kw: observation)
|
|
with server() as (destination, received):
|
|
with server(status=302, location=destination + '/sink') as (source, sent):
|
|
cfg = SimpleNamespace(pdp_url=source, pdp_token_file=credential,
|
|
authorization_policy_package='test', authorization_policy_version='v1')
|
|
admit_tools(cfg, credential.parent, signature_tools)
|
|
with pytest.raises(DecisionError):
|
|
approval_consume.authorize_action(cfg, object(), 'rotate', None)
|
|
assert sent == [True]
|
|
assert received == []
|
|
|
|
|
|
@pytest.mark.parametrize('url', [
|
|
'http://localhost:8080', 'http://localhost.:8080',
|
|
'http://user:synthetic-password@127.0.0.1:8080',
|
|
'http://127.0.0.1:8080?query=value', 'http://127.0.0.1:8080#fragment',
|
|
'http://127.0.0.1:8080/path', 'http://127.0.0.1:0',
|
|
'http://127.0.0.1:wrong', 'http://[::1', 'http://127.0.0.1:65536',
|
|
'http://127.0.0.1:\n8080', ' http://127.0.0.1:8080', 'file://127.0.0.1',
|
|
])
|
|
def test_invalid_address_refused_before_credential_read(url, tmp_path):
|
|
with pytest.raises(DecisionError, match='URL'):
|
|
check_decision(allow_unsigned_test_fixture=True, base_url=url, token_file=tmp_path / 'missing', request={})
|
|
|
|
|
|
@pytest.mark.parametrize('body,reason', [
|
|
(b'{}' + b' ' * (512 * 1024), 'too large'),
|
|
(b'\xff', 'non-JSON'), (b'[]', 'non-object'),
|
|
], ids=['oversized', 'invalid-utf8', 'array'])
|
|
def test_response_limits_and_encoding_fail_closed(credential, body, reason):
|
|
with server(body=body) as (source, _):
|
|
with pytest.raises(DecisionError, match=reason):
|
|
check_decision(allow_unsigned_test_fixture=True, base_url=source, token_file=credential, request={})
|