fix(iac): validate redirects and stop echoing errors

prowler/changelog.d/iac-connection-test-ssrf.security.md

@@ -1 +1 @@
-IaC repository URLs resolving to loopback, private or otherwise non-public hosts rejected before both the connection test and the scan clone, with raw errors no longer echoed to the caller
+IaC repository URLs resolving to loopback, private or otherwise non-public hosts rejected before both the connection test and the scan clone, including every destination a redirect points to, with raw errors no longer echoed to the caller or the logs

prowler/providers/iac/iac_provider.py

@@ -35,6 +35,7 @@ from prowler.providers.iac.exceptions.exceptions import (
     IacScanError,
     IacTrivyNotFoundError,
 )
+from prowler.providers.iac.lib.git_transport import guarded_pool_manager, ls_remote
 
 
 class IacProvider(Provider):
@@ -348,6 +349,7 @@ class IacProvider(Provider):
                 )
 
             temporary_directory = tempfile.mkdtemp()
+            pool_manager = guarded_pool_manager()
             logger.info(
                 f"Cloning repository {original_url} into {temporary_directory}..."
             )
@@ -367,7 +369,10 @@ class IacProvider(Provider):
                         try:
                             bar.title = f"-> Cloning {original_url}..."
                             porcelain.clone(
-                                repository_url, temporary_directory, depth=1
+                                repository_url,
+                                temporary_directory,
+                                depth=1,
+                                pool_manager=pool_manager,
                             )
                             bar.title = "-> Repository cloned successfully!"
                         except Exception as clone_error:
@@ -376,12 +381,22 @@ class IacProvider(Provider):
                 else:
                     # No TTY, just clone without progress bar
                     logger.info(f"Cloning {original_url}...")
-                    porcelain.clone(repository_url, temporary_directory, depth=1)
+                    porcelain.clone(
+                        repository_url,
+                        temporary_directory,
+                        depth=1,
+                        pool_manager=pool_manager,
+                    )
                     logger.info("Repository cloned successfully!")
             except (AttributeError, OSError):
                 # Fallback if isatty() check fails
                 logger.info(f"Cloning {original_url}...")
-                porcelain.clone(repository_url, temporary_directory, depth=1)
+                porcelain.clone(
+                    repository_url,
+                    temporary_directory,
+                    depth=1,
+                    pool_manager=pool_manager,
+                )
                 logger.info("Repository cloned successfully!")
 
             # Detect the branch name from the cloned repository
@@ -689,16 +704,19 @@ class IacProvider(Provider):
                         # Public repository
                         auth_url = scan_repository_url
 
-                    # Use dulwich to test the connection
-                    porcelain.ls_remote(auth_url)
+                    # not porcelain.ls_remote: it cannot be given the pool manager
+                    # that validates redirect destinations
+                    ls_remote(auth_url)
 
                     return Connection(is_connected=True)
 
                 except Exception as e:
-                    # The raw error may carry the token-bearing URL or the
-                    # remote's response, so it is logged and never returned
+                    # A hostile remote can echo the Authorization header back in a git
+                    # ERR packet that dulwich copies into the exception text
                     error_msg = str(e)
-                    logger.error(f"IaC repository connection test failed: {error_msg}")
+                    logger.error(
+                        f"IaC repository connection test failed: {e.__class__.__name__}"
+                    )
                     if "authentication" in error_msg.lower() or "401" in error_msg:
                         return Connection(
                             is_connected=False,

prowler/providers/iac/lib/git_transport.py

@@ -0,0 +1,55 @@
+"""dulwich HTTP transport that keeps redirect destinations inside the SSRF guard."""
+
+from __future__ import annotations
+
+import os
+
+import urllib3
+from dulwich.client import default_urllib3_manager, get_transport_and_path
+from dulwich.config import Config, StackedConfig, env_config
+
+from prowler.lib.network.ssrf import validate_outbound_url
+
+
+class _GuardedRedirects:
+    def urlopen(self, method, url, redirect=True, **kwargs):
+        # urllib3 recurses through self.urlopen once per hop, so validating here
+        # covers the whole redirect chain and not only the supplied URL. dulwich
+        # adopts a redirect's destination as the new repository base.
+        validate_outbound_url(url)
+        return super().urlopen(method, url, redirect=redirect, **kwargs)
+
+
+class GuardedPoolManager(_GuardedRedirects, urllib3.PoolManager):
+    """Pool manager that validates every destination it is asked to open."""
+
+
+class GuardedProxyManager(_GuardedRedirects, urllib3.ProxyManager):
+    """Proxy manager that validates every destination it is asked to open."""
+
+
+def git_config() -> Config:
+    """Stacked git configuration, with the GIT_* environment overrides porcelain applies."""
+    config = StackedConfig.default()
+    override = env_config(os.environ)
+    if override is not None:
+        config.backends.insert(0, override)
+    return config
+
+
+def guarded_pool_manager(config: Config | None = None) -> urllib3.PoolManager:
+    """dulwich HTTP manager whose every request, redirects included, is validated."""
+    return default_urllib3_manager(
+        config if config is not None else git_config(),
+        pool_manager_cls=GuardedPoolManager,
+        proxy_manager_cls=GuardedProxyManager,
+    )
+
+
+def ls_remote(url: str):
+    """porcelain.ls_remote with the guarded manager, which porcelain cannot be given."""
+    config = git_config()
+    client, path = get_transport_and_path(
+        url, config=config, pool_manager=guarded_pool_manager(config)
+    )
+    return client.get_refs(path.encode() if isinstance(path, str) else path)

tests/providers/iac/iac_provider_test.py

@@ -785,7 +785,9 @@ class TestIacProvider:
         url = "https://github.com/user/repo.git"
         with mock.patch.object(provider, "_detect_branch_name", return_value="main"):
             temp_dir, branch_name = provider._clone_repository(url)
-        mock_clone.assert_called_with(url, "/tmp/fake-dir", depth=1)
+        mock_clone.assert_called_with(
+            url, "/tmp/fake-dir", depth=1, pool_manager=mock.ANY
+        )
         assert temp_dir == "/tmp/fake-dir"
         assert branch_name == "main"
 
@@ -799,7 +801,9 @@ class TestIacProvider:
                 url, github_username="user", personal_access_token="token123"
             )
         expected_url = "https://user:token123@github.com/user/repo.git"
-        mock_clone.assert_called_with(expected_url, "/tmp/fake-dir", depth=1)
+        mock_clone.assert_called_with(
+            expected_url, "/tmp/fake-dir", depth=1, pool_manager=mock.ANY
+        )
         assert temp_dir == "/tmp/fake-dir"
         assert branch_name == "develop"
 
@@ -813,10 +817,30 @@ class TestIacProvider:
                 url, oauth_app_token="oauth456"
             )
         expected_url = "https://oauth2:oauth456@github.com/user/repo.git"
-        mock_clone.assert_called_with(expected_url, "/tmp/fake-dir", depth=1)
+        mock_clone.assert_called_with(
+            expected_url, "/tmp/fake-dir", depth=1, pool_manager=mock.ANY
+        )
         assert temp_dir == "/tmp/fake-dir"
         assert branch_name == "master"
 
+    @mock.patch("prowler.providers.iac.iac_provider.porcelain.clone")
+    @mock.patch("tempfile.mkdtemp", return_value="/tmp/fake-dir")
+    def test_clone_repository_validates_redirect_destinations(
+        self, _mock_mkdtemp, mock_clone
+    ):
+        """dulwich adopts a redirect's destination, so the clone needs the guarded manager."""
+        provider = IacProvider()
+        with (
+            mock.patch.object(provider, "_detect_branch_name", return_value="main"),
+            mock.patch(
+                "prowler.providers.iac.iac_provider.guarded_pool_manager",
+                return_value="the-guarded-manager",
+            ),
+        ):
+            provider._clone_repository("https://github.com/user/repo.git")
+
+        assert mock_clone.call_args.kwargs["pool_manager"] == "the-guarded-manager"
+
     @mock.patch("prowler.providers.iac.iac_provider.porcelain.clone")
     @mock.patch("tempfile.mkdtemp", return_value="/tmp/fake-dir")
     def test_clone_repository_failure_raises(self, _mock_mkdtemp, mock_clone):
@@ -881,9 +905,7 @@ class TestIacProvider:
         assert branch_name == "main"
 
     def test_test_connection_rejects_loopback_url(self):
-        with patch(
-            "prowler.providers.iac.iac_provider.porcelain.ls_remote"
-        ) as mock_ls_remote:
+        with patch("prowler.providers.iac.iac_provider.ls_remote") as mock_ls_remote:
             connection = IacProvider.test_connection(
                 scan_repository_url="https://127.0.0.1/user/repo.git"
             )
@@ -893,9 +915,7 @@ class TestIacProvider:
         mock_ls_remote.assert_not_called()
 
     def test_test_connection_rejects_private_range_url(self):
-        with patch(
-            "prowler.providers.iac.iac_provider.porcelain.ls_remote"
-        ) as mock_ls_remote:
+        with patch("prowler.providers.iac.iac_provider.ls_remote") as mock_ls_remote:
             connection = IacProvider.test_connection(
                 scan_repository_url="https://10.0.0.1/user/repo.git"
             )
@@ -944,9 +964,7 @@ class TestIacProvider:
                 "prowler.lib.network.ssrf.socket.getaddrinfo",
                 return_value=[(None, None, None, None, ("140.82.121.4", 0))],
             ) as mock_getaddrinfo,
-            patch(
-                "prowler.providers.iac.iac_provider.porcelain.ls_remote"
-            ) as mock_ls_remote,
+            patch("prowler.providers.iac.iac_provider.ls_remote") as mock_ls_remote,
         ):
             connection = IacProvider.test_connection(
                 scan_repository_url="https://github.com/user/repo.git"
@@ -963,7 +981,7 @@ class TestIacProvider:
                 return_value=[(None, None, None, None, ("140.82.121.4", 0))],
             ),
             patch(
-                "prowler.providers.iac.iac_provider.porcelain.ls_remote",
+                "prowler.providers.iac.iac_provider.ls_remote",
                 side_effect=Exception(
                     "https://x-access-token:SENTINEL_TOKEN@github.com/user/repo.git refused"
                 ),

tests/providers/iac/lib/git_transport_test.py

@@ -0,0 +1,92 @@
+import http.server
+import threading
+
+import pytest
+
+from prowler.lib.network.ssrf import (
+    ALLOWED_PRIVATE_NETWORKS_ENV,
+    OutboundURLNotAllowedError,
+)
+from prowler.providers.iac.lib.git_transport import ls_remote
+
+LINK_LOCAL_TARGET = "http://169.254.169.254/evil/info/refs?service=git-upload-pack"
+
+
+class _Recorder:
+    def __init__(self):
+        self.requests = []
+        self.redirect_to = None
+
+
+def _server(recorder):
+    class Handler(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            recorder.requests.append((self.path, self.headers.get("Authorization")))
+            if recorder.redirect_to:
+                self.send_response(302)
+                self.send_header("Location", recorder.redirect_to)
+            else:
+                self.send_response(404)
+            self.end_headers()
+
+        def log_message(self, *_args):
+            pass
+
+    httpd = http.server.HTTPServer(("127.0.0.1", 0), Handler)
+    threading.Thread(target=httpd.serve_forever, daemon=True).start()
+    return httpd
+
+
+@pytest.fixture
+def origin():
+    recorder = _Recorder()
+    httpd = _server(recorder)
+    recorder.port = httpd.server_address[1]
+    recorder.url = f"http://127.0.0.1:{recorder.port}"
+    yield recorder
+    httpd.shutdown()
+
+
+@pytest.fixture
+def final():
+    recorder = _Recorder()
+    httpd = _server(recorder)
+    recorder.port = httpd.server_address[1]
+    recorder.url = f"http://127.0.0.1:{recorder.port}"
+    yield recorder
+    httpd.shutdown()
+
+
+@pytest.fixture(autouse=True)
+def allow_loopback(monkeypatch):
+    # the servers run on loopback, so the guard has to be told to allow it;
+    # the redirect targets under test are judged by the same allowlist
+    monkeypatch.setenv(ALLOWED_PRIVATE_NETWORKS_ENV, "127.0.0.1/32")
+
+
+class TestLsRemote:
+    def test_rejects_a_redirect_to_a_non_public_address(self, origin):
+        origin.redirect_to = LINK_LOCAL_TARGET
+
+        with pytest.raises(OutboundURLNotAllowedError):
+            ls_remote(f"{origin.url}/repo.git")
+
+        assert len(origin.requests) == 1
+
+    def test_follows_a_redirect_to_an_allowed_address(self, origin, final):
+        origin.redirect_to = f"{final.url}/moved/info/refs?service=git-upload-pack"
+
+        # the final server answers 404, so dulwich reports not a repository
+        # rather than a rejection: the hop was allowed and followed
+        with pytest.raises(Exception) as outcome:
+            ls_remote(f"{origin.url}/repo.git")
+
+        assert not isinstance(outcome.value, OutboundURLNotAllowedError)
+        assert len(final.requests) == 1
+
+    def test_keeps_sending_the_url_credentials(self, origin):
+        """A supplied pool manager must not cost dulwich its userinfo authentication."""
+        with pytest.raises(Exception):
+            ls_remote(f"http://x-access-token:a-token@127.0.0.1:{origin.port}/repo.git")
+
+        assert origin.requests[0][1] is not None