fix(fetch-page): check robots policy before each redirect (#680)

* fix(fetch-page): check robots policy before each redirect

* changelog: place the #680 entry under [Unreleased]

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Mads Lorentzen <madslorentzen_17@hotmail.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_fetch_page.py

@@ -39,12 +39,52 @@ CONTENT_TYPES = {"/euc-kr.html": "text/html; charset=euc-kr"}
 
 class _Handler(http.server.BaseHTTPRequestHandler):
     def do_GET(self):  # noqa: N802 (http.server naming)
-        if self.path == "/ok.html":
+        self.server.paths.append(self.path)
+        if self.path in ("/ok.html", "/ok.html?q=a%20b", "/ok.html?t=%C3%A6%C3%B8%C3%A5") or (
+            self.path.startswith("/unicode/") and self.path.endswith("?page=2")
+        ):
             body = PAGE.encode("utf-8")
             self.send_response(200)
+        elif self.path.startswith("/unicode/"):
+            self.send_response(302)
+            self.send_header("Location", "?page=2")
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
+        elif self.path == "/redirect-utf8":
+            # http.server writes header values as latin-1; preserve raw UTF-8 bytes.
+            self.send_response(302)
+            self.send_header("Location", "/ok.html?t=" + "æøå".encode("utf-8").decode("latin-1"))
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
         elif self.path == "/euc-kr.html":
             body = KOREAN_PAGE.encode("cp949")
             self.send_response(200)
+        elif self.path == "/robots.txt":
+            body = b"User-agent: *\nAllow: /redirect\nDisallow: /blocked\n"
+            self.send_response(200)
+        elif self.path in ("/redirect", "/redirect-allowed"):
+            self.send_response(302)
+            self.send_header("Location", "/blocked" if self.path == "/redirect" else "/ok.html")
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
+        elif self.path == "/redirect-space":
+            self.send_response(302)
+            self.send_header("Location", "/ok.html?q=a b")
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
+        elif self.path == "/redirect-other":
+            self.send_response(302)
+            self.send_header("Location", self.server.redirect_to)
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
+        elif self.path == "/blocked":
+            body = b"<html><body>blocked posting</body></html>"
+            self.send_response(200)
         elif self.path == "/forbidden":
             body = b"<html><body>denied</body></html>"
             self.send_response(403)
@@ -75,6 +115,7 @@ class FetchPageFixture(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
+        cls.server.paths = []
         cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
         cls.thread.start()
         cls.base = f"http://127.0.0.1:{cls.server.server_address[1]}"
@@ -134,6 +175,90 @@ class RobotsGate(FetchPageFixture):
         self.run_main(f"{self.base}/ok.html", "--out", str(self.out))
         self.assertEqual(self.gate_calls, [f"{self.base}/ok.html"])
 
+    def test_redirect_target_is_checked_before_fetching(self):
+        # The documented retry follows redirects. An allowed posting URL can
+        # redirect to a path this site's robots.txt explicitly disallows.
+        fetch_page.gate = self._orig_gate
+        for path in ("curl", "urllib"):
+            with self.subTest(path=path):
+                if path == "urllib":
+                    original_which = fetch_page.shutil.which
+                    fetch_page.shutil.which = lambda name: None
+                    self.addCleanup(setattr, fetch_page.shutil, "which", original_which)
+                self.server.paths.clear()
+                code, out, err = self.run_main(f"{self.base}/redirect", "--out", str(self.out))
+                self.assertEqual(code, 1, out + err)
+                self.assertIn("DISALLOWED", out)
+                self.assertNotIn("/blocked", self.server.paths)
+                self.assertFalse(self.out.exists())
+
+    def test_allowed_redirect_still_fetches_the_posting(self):
+        fetch_page.gate = self._orig_gate
+        for path in ("curl", "urllib"):
+            with self.subTest(path=path):
+                if path == "urllib":
+                    original_which = fetch_page.shutil.which
+                    fetch_page.shutil.which = lambda name: None
+                    self.addCleanup(setattr, fetch_page.shutil, "which", original_which)
+                self.server.paths.clear()
+                code, out, err = self.run_main(f"{self.base}/redirect-allowed", "--out", str(self.out))
+                self.assertEqual(code, 0, out + err)
+                self.assertEqual(self.out.read_text(encoding="utf-8"), PAGE)
+                self.assertIn("/ok.html", self.server.paths)
+                self.assertGreaterEqual(self.server.paths.count("/robots.txt"), 2)
+                self.out.unlink()
+
+    def test_raw_space_in_location_is_encoded_on_both_paths(self):
+        for path in ("curl", "urllib"):
+            with self.subTest(path=path):
+                if path == "urllib":
+                    original_which = fetch_page.shutil.which
+                    fetch_page.shutil.which = lambda name: None
+                    self.addCleanup(setattr, fetch_page.shutil, "which", original_which)
+                self.server.paths.clear()
+                code, out, err = self.run_main(f"{self.base}/redirect-space", "--out", str(self.out))
+                self.assertEqual(code, 0, out + err)
+                self.assertIn("/ok.html?q=a%20b", self.server.paths)
+                self.assertEqual(self.out.read_text(encoding="utf-8"), PAGE)
+                self.out.unlink()
+
+    def test_raw_utf8_in_location_keeps_its_bytes_on_both_paths(self):
+        for path in ("curl", "urllib"):
+            with self.subTest(path=path):
+                if path == "urllib":
+                    original_which = fetch_page.shutil.which
+                    fetch_page.shutil.which = lambda name: None
+                    self.addCleanup(setattr, fetch_page.shutil, "which", original_which)
+                self.server.paths.clear()
+                code, out, err = self.run_main(f"{self.base}/redirect-utf8", "--out", str(self.out))
+                self.assertEqual(code, 0, out + err)
+                self.assertIn("/ok.html?t=%C3%A6%C3%B8%C3%A5", self.server.paths)
+                self.out.unlink()
+
+    @unittest.skipUnless(shutil.which("curl"), "urllib cannot send a URL with raw non-ASCII characters")
+    def test_relative_redirect_from_a_unicode_url(self):
+        self.server.paths.clear()
+        code, out, err = self.run_main(f"{self.base}/unicode/日本", "--out", str(self.out))
+        self.assertEqual(code, 0, out + err)
+        self.assertTrue(any(p.endswith("?page=2") for p in self.server.paths), self.server.paths)
+
+    def test_cross_origin_redirect_uses_destination_policy(self):
+        destination = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
+        destination.paths = []
+        thread = threading.Thread(target=destination.serve_forever, daemon=True)
+        thread.start()
+        self.addCleanup(destination.server_close)
+        self.addCleanup(destination.shutdown)
+        self.server.redirect_to = f"http://127.0.0.1:{destination.server_address[1]}/blocked"
+        fetch_page.gate = self._orig_gate
+
+        code, out, err = self.run_main(f"{self.base}/redirect-other", "--out", str(self.out))
+
+        self.assertEqual(code, 1, out + err)
+        self.assertIn("/robots.txt", destination.paths)
+        self.assertNotIn("/blocked", destination.paths)
+        self.assertFalse(self.out.exists())
+
 
 class Fetching(FetchPageFixture):
     def test_fetches_and_reports_status(self):
@@ -232,7 +357,7 @@ class CurlArgvShape(unittest.TestCase):
 
     def test_curl_argv_is_get_only_and_terminates_options(self):
         url = "-https://example.com/x"  # a URL that would read as a flag
-        argv = fetch_page.build_curl_argv(url, Path("out.html"))
+        argv = fetch_page.build_curl_argv(url, Path("out.html"), Path("headers.txt"), 45)
         self.assertEqual(argv[0], "curl")
         self.assertEqual(argv[-2:], ["--", url])
         for forbidden in ("-d", "--data", "--data-binary", "--data-raw", "-F", "--form",
@@ -240,6 +365,8 @@ class CurlArgvShape(unittest.TestCase):
             self.assertNotIn(forbidden, argv[:-1])
         self.assertIn("--max-filesize", argv)
         self.assertIn("--max-redirs", argv)
+        self.assertEqual(argv[argv.index("--max-redirs") + 1], "0")
+        self.assertNotIn("-L", argv)
         self.assertIn("-A", argv)
         self.assertEqual(argv[argv.index("-A") + 1], fetch_page.BROWSER)