332 lines
12 KiB
Python
332 lines
12 KiB
Python
import threading
|
|
import socket
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
import pytest
|
|
|
|
from tht.evidence.adapters import HttpManifestEvidenceSource
|
|
from tht.evidence.contracts import EvidenceSourceError
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
etag_requests = 0
|
|
etag_body_responses = 0
|
|
redirect_target = "/redirected-v1"
|
|
redirect_request_validators = []
|
|
final_request_validators = []
|
|
|
|
def do_GET(self):
|
|
if self.path.startswith("/etag"):
|
|
type(self).etag_requests += 1
|
|
if self.headers.get("If-None-Match") == '"abc"':
|
|
self.send_response(304)
|
|
self.end_headers()
|
|
return
|
|
self.send_response(200)
|
|
self.send_header("ETag", '"abc"')
|
|
self.send_header("Content-Type", "text/markdown")
|
|
self.end_headers()
|
|
type(self).etag_body_responses += 1
|
|
self.wfile.write(b"hello")
|
|
elif self.path == "/large":
|
|
self.send_response(200)
|
|
self.send_header("Content-Length", "20")
|
|
self.end_headers()
|
|
self.wfile.write(b"x" * 20)
|
|
elif self.path == "/busy":
|
|
self.send_response(503)
|
|
self.end_headers()
|
|
elif self.path == "/missing":
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
elif self.path == "/redirect-private":
|
|
self.send_response(302)
|
|
self.send_header("Location", f"http://127.0.0.1:{self.server.server_port}/etag")
|
|
self.end_headers()
|
|
elif self.path == "/redirect-userinfo":
|
|
self.send_response(302)
|
|
self.send_header(
|
|
"Location", f"http://user:password@127.0.0.1:{self.server.server_port}/etag"
|
|
)
|
|
self.end_headers()
|
|
elif self.path == "/stable-redirect":
|
|
type(self).redirect_request_validators.append(self.headers.get("If-None-Match"))
|
|
self.send_response(302)
|
|
self.send_header("Location", type(self).redirect_target)
|
|
self.end_headers()
|
|
elif self.path in {"/redirected-v1", "/redirected-v2"}:
|
|
type(self).final_request_validators.append(
|
|
(self.path, self.headers.get("If-None-Match"))
|
|
)
|
|
etag = '"v1"' if self.path.endswith("v1") else '"v2"'
|
|
if self.headers.get("If-None-Match") == etag:
|
|
self.send_response(304)
|
|
self.end_headers()
|
|
return
|
|
self.send_response(200)
|
|
self.send_header("ETag", etag)
|
|
self.end_headers()
|
|
self.wfile.write(self.path.encode())
|
|
else:
|
|
self.send_response(200)
|
|
self.send_header("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT")
|
|
self.end_headers()
|
|
self.wfile.write(b"fallback")
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def server_url():
|
|
Handler.etag_requests = 0
|
|
Handler.etag_body_responses = 0
|
|
Handler.redirect_target = "/redirected-v1"
|
|
Handler.redirect_request_validators = []
|
|
Handler.final_request_validators = []
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield f"http://127.0.0.1:{server.server_port}"
|
|
finally:
|
|
server.shutdown()
|
|
thread.join()
|
|
|
|
|
|
def test_http_uses_etag_and_strips_query_from_provenance(server_url):
|
|
source = HttpManifestEvidenceSource(
|
|
[f"{server_url}/etag?token=secret"], allow_private_hosts=True
|
|
)
|
|
item = next(iter(source.discover()))
|
|
|
|
assert item.fingerprint.startswith("etag:")
|
|
assert item.fingerprint != "etag:abc"
|
|
assert item.uri == f"{server_url}/etag"
|
|
assert "secret" not in item.model_dump_json()
|
|
assert source.acquire(item).content == b"hello"
|
|
|
|
|
|
def test_http_uses_last_modified_then_content_hash(server_url):
|
|
modified = next(iter(HttpManifestEvidenceSource(
|
|
[f"{server_url}/modified"], allow_private_hosts=True
|
|
).discover()))
|
|
assert modified.fingerprint.startswith("last-modified:")
|
|
|
|
class NoValidators(Handler):
|
|
def do_GET(self):
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(b"content")
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), NoValidators)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
item = next(iter(HttpManifestEvidenceSource(
|
|
[f"http://127.0.0.1:{server.server_port}/doc"], allow_private_hosts=True
|
|
).discover()))
|
|
assert item.fingerprint.startswith("sha256:")
|
|
finally:
|
|
server.shutdown()
|
|
thread.join()
|
|
|
|
|
|
@pytest.mark.parametrize("path,retryable", [("/busy", True), ("/missing", False)])
|
|
def test_http_classifies_status_errors(server_url, path, retryable):
|
|
with pytest.raises(EvidenceSourceError) as caught:
|
|
list(HttpManifestEvidenceSource(
|
|
[server_url + path], allow_private_hosts=True
|
|
).discover())
|
|
assert caught.value.retryable is retryable
|
|
assert server_url not in str(caught.value)
|
|
|
|
|
|
def test_http_rejects_oversize_and_private_redirect(server_url):
|
|
with pytest.raises(EvidenceSourceError) as large:
|
|
list(HttpManifestEvidenceSource(
|
|
[server_url + "/large"], max_bytes=10, allow_private_hosts=True
|
|
).discover())
|
|
assert not large.value.retryable
|
|
|
|
with pytest.raises(EvidenceSourceError) as redirect:
|
|
list(HttpManifestEvidenceSource([server_url + "/redirect-private"]).discover())
|
|
assert not redirect.value.retryable
|
|
|
|
|
|
def test_http_rejects_unsupported_manifest_scheme():
|
|
with pytest.raises(ValueError, match="http"):
|
|
HttpManifestEvidenceSource(["file:///tmp/secret"])
|
|
|
|
|
|
def test_http_conditional_discovery_reuses_cached_verified_bytes(server_url):
|
|
source = HttpManifestEvidenceSource([server_url + "/etag"], allow_private_hosts=True)
|
|
first = next(iter(source.discover()))
|
|
second = next(iter(source.discover()))
|
|
|
|
assert second == first
|
|
assert source.acquire(second).content == b"hello"
|
|
assert Handler.etag_requests == 3
|
|
assert Handler.etag_body_responses == 1
|
|
|
|
|
|
def test_http_rejects_mixed_public_private_dns_answers(monkeypatch):
|
|
monkeypatch.setattr(socket, "getaddrinfo", lambda *args, **kwargs: [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 80)),
|
|
(socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("::1", 80, 0, 0)),
|
|
])
|
|
with pytest.raises(EvidenceSourceError) as caught:
|
|
list(HttpManifestEvidenceSource(["http://example.test/doc"]).discover())
|
|
assert not caught.value.retryable
|
|
|
|
|
|
def test_http_rejects_userinfo_redirect(server_url):
|
|
with pytest.raises(EvidenceSourceError) as caught:
|
|
list(HttpManifestEvidenceSource(
|
|
[server_url + "/redirect-userinfo"], allow_private_hosts=True
|
|
).discover())
|
|
assert not caught.value.retryable
|
|
|
|
|
|
class FakeSocket:
|
|
def __init__(self, address):
|
|
self.address = address
|
|
|
|
def getpeername(self):
|
|
return (self.address, 443)
|
|
|
|
|
|
class FakeResponse:
|
|
status_code = 200
|
|
headers = {}
|
|
is_redirect = False
|
|
|
|
def __init__(self, *, peer="127.0.0.1", stream_error=None, location=None):
|
|
connection = type("Connection", (), {"sock": FakeSocket(peer)})()
|
|
self.raw = type("Raw", (), {"_connection": connection})()
|
|
self.stream_error = stream_error
|
|
self.closed = False
|
|
if location:
|
|
self.is_redirect = True
|
|
self.status_code = 302
|
|
self.headers = {"Location": location}
|
|
else:
|
|
self.is_redirect = False
|
|
self.status_code = 200
|
|
self.headers = {}
|
|
|
|
def iter_content(self, chunk_size):
|
|
if self.stream_error:
|
|
raise self.stream_error
|
|
yield b"ok"
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, response):
|
|
self.response = response
|
|
|
|
def get(self, *args, **kwargs):
|
|
return self.response
|
|
|
|
|
|
def test_http_rejects_public_to_private_rebind(monkeypatch):
|
|
monkeypatch.setattr(socket, "getaddrinfo", lambda *args, **kwargs: [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)),
|
|
])
|
|
source = HttpManifestEvidenceSource(["https://example.test/doc"])
|
|
response = FakeResponse(peer="127.0.0.1")
|
|
source._session = FakeSession(response)
|
|
with pytest.raises(EvidenceSourceError):
|
|
list(source.discover())
|
|
assert response.closed
|
|
|
|
|
|
def test_http_rejects_public_redirect_to_private_destination(monkeypatch):
|
|
monkeypatch.setattr(socket, "getaddrinfo", lambda host, *args, **kwargs: [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", (
|
|
"93.184.216.34" if host == "example.test" else "127.0.0.1", 443
|
|
)),
|
|
])
|
|
source = HttpManifestEvidenceSource(["https://example.test/doc"])
|
|
response = FakeResponse(
|
|
peer="93.184.216.34", location="https://private.test/secret"
|
|
)
|
|
source._session = FakeSession(response)
|
|
with pytest.raises(EvidenceSourceError) as caught:
|
|
list(source.discover())
|
|
assert not caught.value.retryable
|
|
assert response.closed
|
|
|
|
|
|
def test_http_closes_response_when_streaming_fails():
|
|
source = HttpManifestEvidenceSource(
|
|
["https://example.test/doc"], allow_private_hosts=True
|
|
)
|
|
response = FakeResponse(stream_error=socket.timeout("read timed out"))
|
|
source._session = FakeSession(response)
|
|
with pytest.raises(EvidenceSourceError):
|
|
list(source.discover())
|
|
assert response.closed
|
|
|
|
|
|
def test_http_binds_validators_to_exact_final_redirect_url(server_url):
|
|
source = HttpManifestEvidenceSource(
|
|
[server_url + "/stable-redirect"], allow_private_hosts=True
|
|
)
|
|
first = next(iter(source.discover()))
|
|
second = next(iter(source.discover()))
|
|
|
|
assert first == second
|
|
assert Handler.redirect_request_validators == [None, None]
|
|
assert Handler.final_request_validators == [
|
|
("/redirected-v1", None),
|
|
("/redirected-v1", '"v1"'),
|
|
]
|
|
|
|
|
|
def test_http_redirect_path_change_fetches_and_replaces_body(server_url):
|
|
source = HttpManifestEvidenceSource(
|
|
[server_url + "/stable-redirect"], allow_private_hosts=True
|
|
)
|
|
first = next(iter(source.discover()))
|
|
assert source.acquire(first).content == b"/redirected-v1"
|
|
|
|
Handler.redirect_target = "/redirected-v2"
|
|
with pytest.raises(EvidenceSourceError):
|
|
source.acquire(first)
|
|
current = next(iter(source.discover()))
|
|
|
|
assert source.acquire(current).content == b"/redirected-v2"
|
|
assert ("/redirected-v2", None) in Handler.final_request_validators
|
|
|
|
|
|
def test_http_rejects_unsolicited_304_without_bound_validator(monkeypatch):
|
|
monkeypatch.setattr(socket, "getaddrinfo", lambda *args, **kwargs: [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)),
|
|
])
|
|
source = HttpManifestEvidenceSource(["https://example.test/doc"])
|
|
response = FakeResponse(peer="93.184.216.34")
|
|
response.status_code = 304
|
|
source._session = FakeSession(response)
|
|
|
|
with pytest.raises(EvidenceSourceError) as caught:
|
|
list(source.discover())
|
|
assert not caught.value.retryable
|
|
assert response.closed
|
|
|
|
|
|
def test_http_rejects_cross_origin_304_for_cached_provenance(server_url):
|
|
source = HttpManifestEvidenceSource([server_url + "/etag"], allow_private_hosts=True)
|
|
item = next(iter(source.discover()))
|
|
response = FakeResponse()
|
|
response.status_code = 304
|
|
source._session = FakeSession(response)
|
|
|
|
with pytest.raises(EvidenceSourceError) as caught:
|
|
source._download("https://other.example/doc", item.uri)
|
|
assert not caught.value.retryable
|
|
assert response.closed
|