fix(evidence): bind HTTP validators to final URL

harness/tests/test_http_evidence_source.py

@@ -11,6 +11,9 @@ from tht.ports.evidence 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"):
@@ -46,6 +49,24 @@ class Handler(BaseHTTPRequestHandler):
                 "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")
@@ -60,6 +81,9 @@ class Handler(BaseHTTPRequestHandler):
 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()
@@ -246,3 +270,62 @@ def test_http_closes_response_when_streaming_fails():
     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