Harden workspace manifest HTTPS redirects

cli/python/base_projects/tests/test_workspace_pull.py

@@ -7,8 +7,11 @@
 from contextlib import redirect_stderr, redirect_stdout
 from pathlib import Path
 from unittest import mock
+from urllib.request import Request
 
 from base_projects import engine
+from base_projects.workspace_manifest import WorkspaceManifestError
+from base_projects.workspace_pull import HTTPSOnlyRedirectHandler
 from base_projects.workspace_pull import MAX_WORKSPACE_MANIFEST_SOURCE_BYTES
 
 
@@ -182,7 +185,7 @@ def test_workspace_pull_rejects_cleartext_http_source_before_fetching(self) -> N
             home.mkdir()
             base_home.mkdir()
 
-            with mock.patch("base_projects.workspace_pull.urlopen") as urlopen:
+            with mock.patch("base_projects.workspace_pull.build_opener") as build_opener:
                 status, stdout, stderr = invoke_engine(
                     ["pull", "--source", "http://example.test/workspace.yaml", "--manifest", str(target)],
                     base_home,
@@ -192,7 +195,7 @@ def test_workspace_pull_rejects_cleartext_http_source_before_fetching(self) -> N
         self.assertEqual(status, 1)
         self.assertEqual(stdout, "")
         self.assertFalse(target.exists())
-        urlopen.assert_not_called()
+        build_opener.assert_not_called()
         self.assertIn("Insecure workspace manifest source", stderr)
         self.assertIn("Use https://, file://, or a local path", stderr)
 
@@ -207,8 +210,11 @@ def test_workspace_pull_accepts_https_source(self) -> None:
 
             response = mock.MagicMock()
             response.read.return_value = WORKSPACE_MANIFEST.encode("utf-8")
+            response.geturl.return_value = "https://example.test/workspace.yaml"
             response.__enter__.return_value = response
-            with mock.patch("base_projects.workspace_pull.urlopen", return_value=response) as urlopen:
+            opener = mock.MagicMock()
+            opener.open.return_value = response
+            with mock.patch("base_projects.workspace_pull.build_opener", return_value=opener) as build_opener:
                 status, stdout, stderr = invoke_engine(
                     ["pull", "--source", "https://example.test/workspace.yaml", "--manifest", str(target)],
                     base_home,
@@ -219,9 +225,38 @@ def test_workspace_pull_accepts_https_source(self) -> None:
         self.assertEqual(status, 0)
         self.assertEqual(stderr, "")
         self.assertEqual(target_content, WORKSPACE_MANIFEST)
-        urlopen.assert_called_once_with("https://example.test/workspace.yaml", timeout=30)
+        build_opener.assert_called_once_with(HTTPSOnlyRedirectHandler)
+        opener.open.assert_called_once_with("https://example.test/workspace.yaml", timeout=30)
         self.assertIn("Status: created", stdout)
 
+    def test_https_redirect_handler_rejects_http_target(self) -> None:
+        request = Request("https://example.test/workspace.yaml")
+
+        with self.assertRaisesRegex(WorkspaceManifestError, "Insecure workspace manifest redirect"):
+            HTTPSOnlyRedirectHandler().redirect_request(
+                request,
+                None,
+                302,
+                "Found",
+                {"Location": "http://example.test/workspace.yaml"},
+                "http://example.test/workspace.yaml",
+            )
+
+    def test_https_redirect_handler_allows_https_target(self) -> None:
+        request = Request("https://example.test/workspace.yaml")
+
+        redirect = HTTPSOnlyRedirectHandler().redirect_request(
+            request,
+            None,
+            302,
+            "Found",
+            {"Location": "https://cdn.example.test/workspace.yaml"},
+            "https://cdn.example.test/workspace.yaml",
+        )
+
+        self.assertIsNotNone(redirect)
+        self.assertEqual(redirect.full_url, "https://cdn.example.test/workspace.yaml")
+
     def test_workspace_pull_rejects_invalid_manifest_without_overwriting_target(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             root = Path(tmpdir)

cli/python/base_projects/workspace_pull.py

@@ -5,7 +5,9 @@
 from dataclasses import dataclass
 from pathlib import Path
 from urllib.parse import unquote, urlparse
-from urllib.request import urlopen
+from urllib.request import HTTPRedirectHandler
+from urllib.request import Request
+from urllib.request import build_opener
 
 from base_projects.workspace_manifest import WorkspaceManifest
 from base_projects.workspace_manifest import WorkspaceManifestError
@@ -15,6 +17,19 @@
 MAX_WORKSPACE_MANIFEST_SOURCE_BYTES = 2 * 1024 * 1024
 
 
+class HTTPSOnlyRedirectHandler(HTTPRedirectHandler):
+    def redirect_request(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+        self, req: Request, fp, code, msg, headers, newurl
+    ):  # type: ignore[no-untyped-def]
+        redirect = super().redirect_request(req, fp, code, msg, headers, newurl)
+        if redirect is not None and urlparse(redirect.full_url).scheme != "https":
+            raise WorkspaceManifestError(
+                f"Insecure workspace manifest redirect from '{req.full_url}' to '{redirect.full_url}'. "
+                "Use an https:// redirect target."
+            )
+        return redirect
+
+
 @dataclass(frozen=True)
 class WorkspaceManifestPullResult:
     source: str
@@ -70,7 +85,14 @@ def fetch_workspace_manifest_source(source: str) -> bytes:
     if parsed.scheme == "https":
         try:
             # This command fetches an explicit user-configured manifest source.
-            with urlopen(source, timeout=30) as response:  # nosec B310
+            opener = build_opener(HTTPSOnlyRedirectHandler)
+            with opener.open(source, timeout=30) as response:  # nosec B310
+                final_source = response.geturl()
+                if urlparse(final_source).scheme != "https":
+                    raise WorkspaceManifestError(
+                        f"Insecure workspace manifest redirect from '{source}' to '{final_source}'. "
+                        "Use an https:// redirect target."
+                    )
                 return enforce_workspace_manifest_source_size(
                     source,
                     response.read(MAX_WORKSPACE_MANIFEST_SOURCE_BYTES + 1),