tegwick
3e537d117f
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e6f0-a6f6-7b93-9124-55ef584930e2
127 lines
6 KiB
Python
127 lines
6 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import ipaddress
|
|
import json
|
|
import ssl
|
|
import threading
|
|
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
from cryptography.x509.oid import NameOID
|
|
import pytest
|
|
|
|
from railiance_clock.client import Clock, FileTrust
|
|
from railiance_clock.core import ClockError
|
|
|
|
|
|
@pytest.fixture
|
|
def endpoint(setup):
|
|
path, _, timer, transport = setup
|
|
servers = []
|
|
def create(*, status=200, media='application/jose', encoding=None, tls=None):
|
|
calls = []
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *_):
|
|
pass
|
|
def do_GET(self):
|
|
calls.append(self.path)
|
|
self.send_response(500); self.end_headers()
|
|
def do_POST(self):
|
|
calls.append(self.path)
|
|
body = transport(json.loads(self.rfile.read(int(self.headers['Content-Length']))))
|
|
self.send_response(status)
|
|
self.send_header('Content-Type', media)
|
|
self.send_header('Content-Length', str(len(body)))
|
|
if 300 <= status < 400:
|
|
self.send_header('Location', '/must-not-follow')
|
|
if encoding:
|
|
self.send_header('Content-Encoding', encoding)
|
|
self.end_headers(); self.wfile.write(body)
|
|
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
|
|
if tls:
|
|
server.socket = tls.wrap_socket(server.socket, server_side=True)
|
|
thread = threading.Thread(target=server.serve_forever); thread.start()
|
|
servers.append((server, thread))
|
|
data = json.loads(path.read_text())
|
|
data['endpoint'] = f'{"https" if tls else "http"}://127.0.0.1:{server.server_port}/v1/time-samples'
|
|
data['transport'] = 'https' if tls else 'admitted-loopback'
|
|
path.write_text(json.dumps(data))
|
|
return Clock(FileTrust(path), timer=timer), calls
|
|
yield create
|
|
for server, thread in servers:
|
|
server.shutdown(); server.server_close(); thread.join()
|
|
|
|
|
|
def test_http_response_positive_control(endpoint):
|
|
clock, calls = endpoint()
|
|
assert clock.read().authority_id == 'railiance01'
|
|
assert calls == ['/v1/time-samples']
|
|
|
|
|
|
@pytest.mark.parametrize('status', [301, 302, 303, 307, 308])
|
|
def test_redirect_is_never_followed(endpoint, status):
|
|
clock, calls = endpoint(status=status)
|
|
with pytest.raises(ClockError):
|
|
clock.read()
|
|
assert calls == ['/v1/time-samples']
|
|
|
|
|
|
@pytest.mark.parametrize('options', [{'media': 'application/json'}, {'media': 'text/plain'},
|
|
{'encoding': 'gzip'}, {'status': 201}, {'status': 503}])
|
|
def test_unexpected_http_response_refused(endpoint, options):
|
|
clock, calls = endpoint(**options)
|
|
with pytest.raises(ClockError):
|
|
clock.read()
|
|
assert calls == ['/v1/time-samples']
|
|
|
|
|
|
@pytest.mark.parametrize('case', ['valid', 'expired', 'future', 'wrong-host', 'untrusted'])
|
|
def test_tls_certificate_validation(endpoint, tmp_path, monkeypatch, case):
|
|
now = datetime.now(timezone.utc)
|
|
ca_key = ec.generate_private_key(ec.SECP256R1())
|
|
ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'Disposable test CA')])
|
|
ca = (x509.CertificateBuilder().subject_name(ca_name).issuer_name(ca_name)
|
|
.public_key(ca_key.public_key()).serial_number(x509.random_serial_number())
|
|
.not_valid_before(now - timedelta(days=2)).not_valid_after(now + timedelta(days=2))
|
|
.add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
|
|
.sign(ca_key, hashes.SHA256()))
|
|
key = ec.generate_private_key(ec.SECP256R1())
|
|
start, end = now - timedelta(hours=1), now + timedelta(hours=1)
|
|
if case == 'expired':
|
|
start, end = now - timedelta(days=1), now - timedelta(hours=1)
|
|
if case == 'future':
|
|
start, end = now + timedelta(hours=1), now + timedelta(days=1)
|
|
san = x509.DNSName('wrong.invalid') if case == 'wrong-host' else x509.IPAddress(ipaddress.ip_address('127.0.0.1'))
|
|
leaf_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'Disposable test server')])
|
|
cert = (x509.CertificateBuilder().subject_name(leaf_name).issuer_name(ca_name)
|
|
.public_key(key.public_key()).serial_number(x509.random_serial_number())
|
|
.not_valid_before(start).not_valid_after(end)
|
|
.add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
|
|
.add_extension(x509.SubjectAlternativeName([san]), critical=False)
|
|
.sign(ca_key, hashes.SHA256()))
|
|
ca_file, cert_file, key_file = (tmp_path / name for name in ('ca.pem', 'cert.pem', 'key.pem'))
|
|
ca_file.write_bytes(ca.public_bytes(serialization.Encoding.PEM))
|
|
cert_file.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
|
|
key_file.write_bytes(key.private_bytes(serialization.Encoding.PEM,
|
|
serialization.PrivateFormat.PKCS8, serialization.NoEncryption()))
|
|
key_file.chmod(0o600)
|
|
empty = tmp_path / 'empty-ca-directory'; empty.mkdir()
|
|
# Exercise urllib's real default verification; no permissive SSL context.
|
|
monkeypatch.setenv('SSL_CERT_FILE', str(ca_file if case != 'untrusted' else empty / 'missing.pem'))
|
|
monkeypatch.setenv('SSL_CERT_DIR', str(empty))
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
context.load_cert_chain(cert_file, key_file)
|
|
clock, calls = endpoint(tls=context)
|
|
if case == 'valid':
|
|
assert clock.read().authority_id == 'railiance01'
|
|
assert calls == ['/v1/time-samples']
|
|
else:
|
|
with pytest.raises(ClockError) as exc:
|
|
clock.read()
|
|
reason = exc.value.__cause__.reason
|
|
assert isinstance(reason, ssl.SSLCertVerificationError)
|
|
expected = {'expired': 'expired', 'future': 'not yet valid',
|
|
'wrong-host': 'IP address mismatch', 'untrusted': 'issuer certificate'}
|
|
assert expected[case] in reason.verify_message
|
|
assert calls == [] # Refused during TLS, before application data.
|