fix(iac): block SSRF in repository connection test (#12960)

docs/user-guide/providers/iac/authentication.mdx

@@ -11,3 +11,28 @@ Prowler's Infrastructure as Code (IaC) provider enables you to scan local or rem
     - [**GitHub Username and Personal Access Token (PAT)**](https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/managing-your-personal-access-tokens#creating-a-personal-access-token-classic)
     - [**GitHub OAuth App Token**](https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/managing-your-personal-access-tokens#creating-a-fine-grained-personal-access-token)
     - [**Git URL**](https://git-scm.com/docs/git-clone#_git_urls)
+
+### Private Repository Hosts
+
+**Prowler CLI scans a repository on an internal network with no extra configuration.** On the CLI the operator supplies the repository URL themselves, so there is nothing to defend against and the check below does not apply.
+
+**Prowler App rejects a repository URL that resolves to a non-public address**, as an SSRF defense. There the URL arrives from a tenant and the worker must not be used to reach the internal network it runs in. The check covers both the connection test and the scan clone. To scan a self-hosted GitLab, Gitea or Bitbucket from Prowler App, declare the trusted ranges:
+
+```console
+PROWLER_ALLOWED_PRIVATE_NETWORKS="10.20.0.0/16,192.168.65.254/32"
+```
+
+The value is a comma-separated list of IPs and CIDRs. A resolved address inside an allowlisted range is permitted; every other non-public address stays blocked, so link-local (`169.254.169.254`), loopback and the rest of the internal network remain protected. Malformed entries are rejected, and a non-empty allowlist is logged as a relaxed security control.
+
+The scan runs in the worker, not the API, so the variable has to reach the worker:
+
+- **Docker Compose**: set it in the root `.env`; it reaches the worker through the shared environment file.
+- **Helm**: add it under `api.djangoConfig`, which is rendered into the API ConfigMap that the worker inherits through `envFrom`.
+
+```yaml
+api:
+  djangoConfig:
+    PROWLER_ALLOWED_PRIVATE_NETWORKS: "10.20.0.0/16"
+```
+
+Setting it only on the API container has no effect. The same variable applies to the Kubernetes and OpenStack providers. The Image provider has its own, `PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS`.

prowler/__main__.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
+import os
 import sys
 import tempfile
 from os import environ
@@ -55,6 +56,7 @@ from prowler.lib.check.custom_checks_metadata import (
 from prowler.lib.check.models import CheckMetadata
 from prowler.lib.cli.parser import ProwlerArgumentParser
 from prowler.lib.logger import logger, set_logging_config
+from prowler.lib.network.ssrf import SKIP_OUTBOUND_CHECK_ENV
 from prowler.lib.outputs.asff.asff import ASFF
 from prowler.lib.outputs.compliance.asd_essential_eight.asd_essential_eight_aws import (
     ASDEssentialEightAWS,
@@ -236,6 +238,11 @@ def _send_ocsf_to_cloud(file_path: str) -> dict | None:
 
 
 def prowler():
+    # On the CLI the operator supplies the target themselves, so the outbound host
+    # check defends nothing and would refuse their own private network. setdefault,
+    # so an operator who set the variable explicitly keeps their choice.
+    os.environ.setdefault(SKIP_OUTBOUND_CHECK_ENV, "true")
+
     # Parse Arguments
     # Refactor(CLI)
     parser = ProwlerArgumentParser()

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

@@ -0,0 +1 @@
+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/lib/network/ssrf.py

@@ -0,0 +1,166 @@
+"""Outbound URL validation for provider connection tests."""
+
+from __future__ import annotations
+
+import ipaddress
+import os
+import re
+import socket
+from urllib.parse import urlparse
+
+from prowler.lib.logger import logger
+
+ALLOWED_PRIVATE_NETWORKS_ENV = "PROWLER_ALLOWED_PRIVATE_NETWORKS"
+
+# The check stays on unless a host process opts out. `prowler/__main__.py` opts the
+# CLI out, because there the operator and the person supplying the URL are the same
+# and there is no SSRF boundary to defend. Anything else -- Prowler App's API and
+# worker above all -- keeps the check, so a missing setting fails safe.
+SKIP_OUTBOUND_CHECK_ENV = "PROWLER_SKIP_OUTBOUND_HOST_CHECK"
+
+_TRUTHY = {"1", "true", "yes", "on"}
+
+_NON_PUBLIC_IP_PROPERTIES = (
+    "is_private",
+    "is_loopback",
+    "is_link_local",
+    "is_multicast",
+    "is_reserved",
+    "is_unspecified",
+)
+
+# scp-like git remotes (user@host:path) carry no scheme, so urlparse cannot read them
+_SCP_LIKE_REMOTE = re.compile(r"^(?:[^@/]+@)?(?P<host>[^:/]+):(?!//)")
+
+_NAT64_WELL_KNOWN_PREFIX = ipaddress.IPv6Network("64:ff9b::/96")
+
+
+class OutboundURLNotAllowedError(Exception):
+    """A supplied URL points at a destination the worker must not reach."""
+
+
+def _parse_allowed_networks(raw: str | None) -> tuple:
+    if not raw or not raw.strip():
+        return ()
+    networks = []
+    for entry in raw.split(","):
+        entry = entry.strip()
+        if not entry:
+            continue
+        try:
+            networks.append(ipaddress.ip_network(entry, strict=False))
+        except ValueError as error:
+            raise OutboundURLNotAllowedError(
+                f"Malformed entry {entry!r} in {ALLOWED_PRIVATE_NETWORKS_ENV}: {error}"
+            )
+    return tuple(networks)
+
+
+def allowed_private_networks() -> tuple:
+    """Operator-configured private networks the SSRF guard must not block."""
+    networks = _parse_allowed_networks(os.environ.get(ALLOWED_PRIVATE_NETWORKS_ENV))
+    if networks:
+        logger.warning(
+            f"{ALLOWED_PRIVATE_NETWORKS_ENV} is set — SSRF protection relaxed for private networks: "
+            + ", ".join(str(network) for network in networks)
+        )
+    return networks
+
+
+def outbound_check_skipped() -> bool:
+    """Whether this process asked to skip the non-public destination check."""
+    return os.environ.get(SKIP_OUTBOUND_CHECK_ENV, "").strip().lower() in _TRUTHY
+
+
+def _unwrap_ipv6(address: ipaddress._BaseAddress) -> ipaddress._BaseAddress:
+    if not isinstance(address, ipaddress.IPv6Address):
+        return address
+    embedded = address.ipv4_mapped or address.sixtofour
+    if embedded is None and address in _NAT64_WELL_KNOWN_PREFIX:
+        embedded = ipaddress.IPv4Address(int(address) & 0xFFFFFFFF)
+    return embedded or address
+
+
+def _ip_is_non_public(address: str) -> bool:
+    try:
+        parsed = _unwrap_ipv6(ipaddress.ip_address(address))
+    except ValueError:
+        return False
+    # is_global is the broad check; the properties stay because some multicast
+    # ranges report is_global and would otherwise slip through
+    if not parsed.is_global:
+        return True
+    return any(getattr(parsed, prop) for prop in _NON_PUBLIC_IP_PROPERTIES)
+
+
+def _ip_is_allowlisted(address: str, networks: tuple) -> bool:
+    try:
+        parsed = ipaddress.ip_address(address)
+    except ValueError:
+        return False
+    return any(
+        parsed.version == network.version and parsed in network for network in networks
+    )
+
+
+def _resolve(host: str) -> set:
+    try:
+        return {sockaddr[0] for *_, sockaddr in socket.getaddrinfo(host, None)}
+    except socket.gaierror as error:
+        raise OutboundURLNotAllowedError(f"Could not resolve host {host!r}: {error}")
+
+
+def extract_host(url: str) -> str:
+    """Host of a URL, accepting scp-like git remotes that carry no scheme."""
+    scp_like = _SCP_LIKE_REMOTE.match(url)
+    if scp_like and "://" not in url:
+        return scp_like.group("host")
+    host = urlparse(url).hostname
+    if not host:
+        # The URL itself stays out of the message: a configured repository or auth
+        # URL can carry credentials in its userinfo, and this message is logged
+        raise OutboundURLNotAllowedError("Could not read a host from the supplied URL")
+    return host
+
+
+def validate_outbound_host(host: str) -> None:
+    """Reject a host that is, or resolves to, a non-public address.
+
+    Resolution happens here and again inside the client that connects, so a
+    hostile DNS server can still answer differently the second time.
+    """
+    if outbound_check_skipped():
+        logger.warning(
+            f"{SKIP_OUTBOUND_CHECK_ENV} is set — destination check disabled for host {host!r}"
+        )
+        return
+
+    networks = allowed_private_networks()
+
+    try:
+        ipaddress.ip_address(host)
+    except ValueError:
+        addresses = _resolve(host)
+    else:
+        addresses = {host}
+
+    for address in addresses:
+        if _ip_is_non_public(address) and not _ip_is_allowlisted(address, networks):
+            raise OutboundURLNotAllowedError(
+                f"Host {host!r} resolves to non-public address {address} and cannot be "
+                f"reached. To scan a target on a private network, list the trusted "
+                f"ranges in the {ALLOWED_PRIVATE_NETWORKS_ENV} environment variable of "
+                f"the process running the scan"
+            )
+
+
+def validate_outbound_url(
+    url: str, *, allowed_schemes: tuple = ("http", "https")
+) -> None:
+    """Reject a URL whose scheme is not allowed or whose host is not public."""
+    scheme = urlparse(url).scheme
+    if scheme and scheme not in allowed_schemes:
+        raise OutboundURLNotAllowedError(
+            f"Disallowed URL scheme {scheme!r}. Allowed: {', '.join(allowed_schemes)}"
+        )
+    validate_outbound_host(extract_host(url))

prowler/providers/iac/iac_provider.py

@@ -17,6 +17,10 @@ from prowler.config.config import (
 )
 from prowler.lib.check.models import CheckReportIAC
 from prowler.lib.logger import logger
+from prowler.lib.network.ssrf import (
+    OutboundURLNotAllowedError,
+    validate_outbound_url,
+)
 from prowler.lib.utils.utils import print_boxes
 from prowler.lib.utils.vulnerability_references import (
     build_finding_reference_url,
@@ -31,6 +35,14 @@ from prowler.providers.iac.exceptions.exceptions import (
     IacScanError,
     IacTrivyNotFoundError,
 )
+from prowler.providers.iac.lib.git_transport import (
+    ALLOWED_SCHEMES,
+    effective_url,
+    git_config,
+    guarded_pool_manager,
+    ls_remote,
+    proxy_base_url,
+)
 
 
 class IacProvider(Provider):
@@ -326,6 +338,11 @@ class IacProvider(Provider):
         Returns:
             tuple[str, str]: (temporary_directory, branch_name)
         """
+        # one config for both: git rewrites the URL again inside porcelain, and the
+        # guard must judge the same destination dulwich will connect to
+        config = git_config()
+        target = effective_url(config, repository_url)
+        validate_outbound_url(target, allowed_schemes=ALLOWED_SCHEMES)
         try:
             original_url = repository_url
 
@@ -341,6 +358,7 @@ class IacProvider(Provider):
                 )
 
             temporary_directory = tempfile.mkdtemp()
+            pool_manager = guarded_pool_manager(config, proxy_base_url(target))
             logger.info(
                 f"Cloning repository {original_url} into {temporary_directory}..."
             )
@@ -360,7 +378,11 @@ class IacProvider(Provider):
                         try:
                             bar.title = f"-> Cloning {original_url}..."
                             porcelain.clone(
-                                repository_url, temporary_directory, depth=1
+                                repository_url,
+                                temporary_directory,
+                                depth=1,
+                                config=config,
+                                pool_manager=pool_manager,
                             )
                             bar.title = "-> Repository cloned successfully!"
                         except Exception as clone_error:
@@ -369,12 +391,24 @@ 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,
+                        config=config,
+                        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,
+                    config=config,
+                    pool_manager=pool_manager,
+                )
                 logger.info("Repository cloned successfully!")
 
             # Detect the branch name from the cloned repository
@@ -382,11 +416,14 @@ class IacProvider(Provider):
 
             return temporary_directory, branch_name
         except Exception as error:
+            # the authenticated URL embeds the token, and ProwlerException puts
+            # original_exception into its str(), which the API returns verbatim
             logger.critical(
-                f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
+                f"{error.__class__.__name__}:{error.__traceback__.tb_lineno}"
             )
             raise IacRepositoryCloneError(
-                file=__file__, original_exception=error
+                file=__file__,
+                message=f"Unable to clone the repository to scan ({error.__class__.__name__})",
             ) from error
 
     def run(self) -> List[CheckReportIAC]:
@@ -652,6 +689,17 @@ class IacProvider(Provider):
                     is_connected=False, error="Repository URL is required"
                 )
 
+            try:
+                validate_outbound_url(
+                    scan_repository_url, allowed_schemes=ALLOWED_SCHEMES
+                )
+            except OutboundURLNotAllowedError as error:
+                logger.warning(f"Rejected IaC repository URL: {error}")
+                return Connection(
+                    is_connected=False,
+                    error="Repository URL is not an allowed destination.",
+                )
+
             # Try to clone the repository to test the connection
             with tempfile.TemporaryDirectory():
                 try:
@@ -670,13 +718,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:
+                    # 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: {e.__class__.__name__}"
+                    )
                     if "authentication" in error_msg.lower() or "401" in error_msg:
                         return Connection(
                             is_connected=False,
@@ -690,13 +744,14 @@ class IacProvider(Provider):
                     else:
                         return Connection(
                             is_connected=False,
-                            error=f"Failed to connect to repository: {error_msg}",
+                            error="Failed to connect to repository.",
                         )
 
         except Exception as error:
             if raise_on_exception:
                 raise
+            logger.error(f"Unexpected error testing IaC repository connection: {error}")
             return Connection(
                 is_connected=False,
-                error=f"Unexpected error testing connection: {str(error)}",
+                error="Unexpected error testing connection.",
             )

prowler/providers/iac/lib/git_transport.py

@@ -0,0 +1,98 @@
+"""dulwich HTTP transport that keeps redirect destinations inside the SSRF guard."""
+
+from __future__ import annotations
+
+import os
+from urllib.parse import urlparse, urlunparse
+
+import urllib3
+from dulwich.client import default_urllib3_manager, get_transport_and_path
+from dulwich.config import Config, StackedConfig, apply_instead_of, env_config
+
+from prowler.lib.network.ssrf import validate_outbound_url
+
+# shared so the connection test and the clone cannot disagree about what a
+# tenant is allowed to configure
+ALLOWED_SCHEMES = ("http", "https", "ssh", "git+ssh", "git")
+
+
+class _GuardedRedirects:
+    def urlopen(self, method, url, redirect=True, **kwargs):
+        # urllib3 recurses through self.urlopen once per hop, so validating here
+        # covers the whole chain; dulwich adopts a hop 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_* overrides porcelain applies."""
+    config = StackedConfig.default()
+    override = env_config(os.environ)
+    if override is not None:
+        config.backends.insert(0, override)
+    return config
+
+
+def effective_url(config: Config, url: str) -> str:
+    """URL after git's ``url.*.insteadOf`` rewriting, which is what picks the transport.
+
+    A rewrite can turn a validated public URL into an ``ssh://`` one on a private
+    host, so the guard has to see this form and not the supplied one.
+    """
+    return apply_instead_of(config, url, push=False)
+
+
+def proxy_base_url(url: str) -> str:
+    """URL without credentials, for dulwich's ``no_proxy`` and ``http.<url>.*`` matching.
+
+    Only the userinfo is removed. The path stays, because an ``http.<url>.*``
+    section can be path-specific and dulwich ranks matches by path length.
+    """
+    parsed = urlparse(url)
+    if parsed.username is None and parsed.password is None:
+        return url
+    host = parsed.hostname or ""
+    # urlparse drops the brackets of an IPv6 literal and they are not optional
+    netloc = f"[{host}]" if ":" in host else host
+    if parsed.port:
+        netloc = f"{netloc}:{parsed.port}"
+    return urlunparse(parsed._replace(netloc=netloc))
+
+
+def guarded_pool_manager(config: Config, base_url: str) -> urllib3.PoolManager:
+    """dulwich HTTP manager whose every request, redirects included, is validated.
+
+    ``base_url`` is what dulwich uses to honour ``no_proxy`` and the per-URL
+    ``http.*`` settings, so building the manager without it silently drops them.
+    """
+    return default_urllib3_manager(
+        config,
+        base_url=base_url,
+        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.
+
+    The supplied URL is handed to dulwich, not the rewritten one: ``insteadOf`` is a
+    single substitution, so rewriting twice can land on a third destination.
+    """
+    config = git_config()
+    target = effective_url(config, url)
+    validate_outbound_url(target, allowed_schemes=ALLOWED_SCHEMES)
+    client, path = get_transport_and_path(
+        url,
+        config=config,
+        pool_manager=guarded_pool_manager(config, proxy_base_url(target)),
+    )
+    return client.get_refs(path.encode() if isinstance(path, str) else path)

tests/lib/network/ssrf_test.py

@@ -0,0 +1,257 @@
+import ipaddress
+import os
+import socket
+from unittest import mock
+
+import pytest
+
+from prowler.lib.network.ssrf import (
+    ALLOWED_PRIVATE_NETWORKS_ENV,
+    SKIP_OUTBOUND_CHECK_ENV,
+    OutboundURLNotAllowedError,
+    allowed_private_networks,
+    extract_host,
+    outbound_check_skipped,
+    validate_outbound_host,
+    validate_outbound_url,
+)
+
+
+def _resolves_to(*addresses):
+    return mock.patch.object(
+        socket,
+        "getaddrinfo",
+        return_value=[(2, 1, 6, "", (address, 0)) for address in addresses],
+    )
+
+
+class TestValidateOutboundHost:
+    @pytest.mark.parametrize(
+        "address",
+        [
+            "127.0.0.1",
+            "10.0.0.5",
+            "192.168.1.1",
+            "172.16.0.1",
+            "169.254.169.254",
+            "0.0.0.0",
+            "224.0.0.1",
+            "198.18.0.1",
+            "203.0.113.1",
+            "::1",
+            "fe80::1",
+            "fc00::1",
+            "ff02::1",
+            "2001:db8::1",
+        ],
+    )
+    def test_rejects_non_public_literals(self, address):
+        with pytest.raises(OutboundURLNotAllowedError):
+            validate_outbound_host(address)
+
+    @pytest.mark.parametrize(
+        "address", ["100.64.0.1", "100.100.100.200", "100.127.255.254"]
+    )
+    def test_rejects_shared_address_space(self, address):
+        with pytest.raises(OutboundURLNotAllowedError):
+            validate_outbound_host(address)
+
+    @pytest.mark.parametrize(
+        "address", ["8.8.8.8", "1.1.1.1", "140.82.121.4", "2606:4700:4700::1111"]
+    )
+    def test_allows_public_literals(self, address):
+        validate_outbound_host(address)
+
+    @pytest.mark.parametrize(
+        "address",
+        ["::ffff:169.254.169.254", "2002:a00:5::", "64:ff9b::a00:5"],
+    )
+    def test_rejects_ipv6_embedding_a_non_public_ipv4(self, address):
+        with pytest.raises(OutboundURLNotAllowedError):
+            validate_outbound_host(address)
+
+    def test_rejects_a_hostname_resolving_to_a_private_address(self):
+        with _resolves_to("10.0.0.5"):
+            with pytest.raises(OutboundURLNotAllowedError):
+                validate_outbound_host("registry.internal")
+
+    def test_rejects_a_hostname_with_one_non_public_answer(self):
+        with _resolves_to("8.8.8.8", "127.0.0.1"):
+            with pytest.raises(OutboundURLNotAllowedError):
+                validate_outbound_host("split-horizon.example.com")
+
+    def test_allows_a_hostname_resolving_to_a_public_address(self):
+        with _resolves_to("140.82.121.4"):
+            validate_outbound_host("github.com")
+
+    def test_rejects_a_hostname_that_does_not_resolve(self):
+        with mock.patch.object(
+            socket, "getaddrinfo", side_effect=socket.gaierror("no such host")
+        ):
+            with pytest.raises(OutboundURLNotAllowedError, match="Could not resolve"):
+                validate_outbound_host("nowhere.invalid")
+
+
+class TestOutboundCheckOptOut:
+    def test_is_enforced_when_the_variable_is_unset(self, monkeypatch):
+        monkeypatch.delenv(SKIP_OUTBOUND_CHECK_ENV, raising=False)
+        with pytest.raises(OutboundURLNotAllowedError):
+            validate_outbound_host("169.254.169.254")
+
+    @pytest.mark.parametrize("raw", ["1", "true", "TRUE", "yes", "on"])
+    def test_is_skipped_when_the_variable_is_truthy(self, monkeypatch, raw):
+        monkeypatch.setenv(SKIP_OUTBOUND_CHECK_ENV, raw)
+        validate_outbound_host("169.254.169.254")
+
+    @pytest.mark.parametrize("raw", ["", "0", "false", "no", "off", "maybe"])
+    def test_is_enforced_for_anything_else(self, monkeypatch, raw):
+        monkeypatch.setenv(SKIP_OUTBOUND_CHECK_ENV, raw)
+        with pytest.raises(OutboundURLNotAllowedError):
+            validate_outbound_host("169.254.169.254")
+
+    def test_logs_the_skip_so_an_operator_sees_the_disabled_control(self, monkeypatch):
+        monkeypatch.setenv(SKIP_OUTBOUND_CHECK_ENV, "true")
+
+        with mock.patch("prowler.lib.network.ssrf.logger") as logged:
+            validate_outbound_host("169.254.169.254")
+
+        assert logged.warning.call_count == 1
+        assert SKIP_OUTBOUND_CHECK_ENV in logged.warning.call_args.args[0]
+
+    def test_the_cli_entrypoint_opts_out(self, monkeypatch):
+        """prowler() must set the opt-out before it does anything else."""
+        # a throwaway mapping, so the variable prowler() sets cannot leak into
+        # the tests that assert the check is enforced
+        monkeypatch.setattr(os, "environ", dict(os.environ))
+        os.environ.pop(SKIP_OUTBOUND_CHECK_ENV, None)
+
+        import prowler.__main__ as prowler_main
+
+        class _Stop(Exception):
+            pass
+
+        def _parser_that_stops():
+            # raised from the first statement after the opt-out, so reaching here
+            # proves the variable was already set
+            raise _Stop
+
+        monkeypatch.setattr(prowler_main, "ProwlerArgumentParser", _parser_that_stops)
+
+        with pytest.raises(_Stop):
+            prowler_main.prowler()
+
+        assert outbound_check_skipped()
+
+    @pytest.mark.parametrize("chosen", ["false", "0", "off", ""])
+    def test_the_cli_entrypoint_respects_an_explicit_value(self, monkeypatch, chosen):
+        """setdefault, not assignment: an operator who opted in keeps the check."""
+        monkeypatch.setattr(os, "environ", dict(os.environ))
+        os.environ[SKIP_OUTBOUND_CHECK_ENV] = chosen
+
+        import prowler.__main__ as prowler_main
+
+        class _Stop(Exception):
+            pass
+
+        def _parser_that_stops():
+            raise _Stop
+
+        monkeypatch.setattr(prowler_main, "ProwlerArgumentParser", _parser_that_stops)
+
+        with pytest.raises(_Stop):
+            prowler_main.prowler()
+
+        assert os.environ[SKIP_OUTBOUND_CHECK_ENV] == chosen
+        assert not outbound_check_skipped()
+
+
+class TestAllowedPrivateNetworks:
+    def test_is_empty_when_unset(self, monkeypatch):
+        monkeypatch.delenv(ALLOWED_PRIVATE_NETWORKS_ENV, raising=False)
+        assert allowed_private_networks() == ()
+
+    @pytest.mark.parametrize("raw", ["", "   ", ",", " , "])
+    def test_is_empty_for_blank_values(self, monkeypatch, raw):
+        monkeypatch.setenv(ALLOWED_PRIVATE_NETWORKS_ENV, raw)
+        assert allowed_private_networks() == ()
+
+    def test_parses_addresses_and_cidrs(self, monkeypatch):
+        monkeypatch.setenv(
+            ALLOWED_PRIVATE_NETWORKS_ENV, "10.20.0.0/16, 192.168.65.254 ,fc00::/7"
+        )
+        assert allowed_private_networks() == (
+            ipaddress.ip_network("10.20.0.0/16"),
+            ipaddress.ip_network("192.168.65.254/32"),
+            ipaddress.ip_network("fc00::/7"),
+        )
+
+    def test_rejects_a_malformed_entry(self, monkeypatch):
+        monkeypatch.setenv(ALLOWED_PRIVATE_NETWORKS_ENV, "10.20.0.0/16,not-a-network")
+        with pytest.raises(OutboundURLNotAllowedError, match="Malformed entry"):
+            allowed_private_networks()
+
+    def test_allows_an_address_inside_an_allowlisted_range(self, monkeypatch):
+        monkeypatch.setenv(ALLOWED_PRIVATE_NETWORKS_ENV, "10.20.0.0/16")
+        validate_outbound_host("10.20.1.5")
+
+    def test_still_rejects_an_address_outside_the_allowlisted_range(self, monkeypatch):
+        monkeypatch.setenv(ALLOWED_PRIVATE_NETWORKS_ENV, "10.20.0.0/16")
+        with pytest.raises(OutboundURLNotAllowedError):
+            validate_outbound_host("169.254.169.254")
+
+    def test_does_not_match_an_allowlist_entry_of_another_family(self, monkeypatch):
+        monkeypatch.setenv(ALLOWED_PRIVATE_NETWORKS_ENV, "fc00::/7")
+        with pytest.raises(OutboundURLNotAllowedError):
+            validate_outbound_host("10.20.1.5")
+
+
+class TestExtractHost:
+    @pytest.mark.parametrize(
+        "url, expected",
+        [
+            ("https://github.com/org/repo", "github.com"),
+            (
+                "https://user:token@github.com/org/repo",  # trufflehog:ignore
+                "github.com",
+            ),
+            ("https://github.com:8443/org/repo", "github.com"),
+            ("git@github.com:org/repo.git", "github.com"),
+            ("github.com:org/repo.git", "github.com"),
+            ("ssh://git@github.com/org/repo.git", "github.com"),
+        ],
+    )
+    def test_reads_the_host(self, url, expected):
+        assert extract_host(url) == expected
+
+    def test_rejects_a_url_without_a_host(self):
+        with pytest.raises(OutboundURLNotAllowedError, match="Could not read a host"):
+            extract_host("not a url")
+
+    def test_keeps_the_url_out_of_the_rejection_message(self):
+        # the message is logged, and a configured URL can carry credentials
+        with pytest.raises(OutboundURLNotAllowedError) as rejection:
+            extract_host("https://user:s3cr3t@/org/repo.git")
+
+        assert "s3cr3t" not in str(rejection.value)
+        assert "user" not in str(rejection.value)
+
+
+class TestValidateOutboundURL:
+    def test_rejects_a_disallowed_scheme(self):
+        with pytest.raises(OutboundURLNotAllowedError, match="Disallowed URL scheme"):
+            validate_outbound_url("file:///etc/passwd")
+
+    def test_allows_an_explicitly_permitted_scheme(self):
+        with _resolves_to("140.82.121.4"):
+            validate_outbound_url(
+                "ssh://git@github.com/org/repo.git",
+                allowed_schemes=("http", "https", "ssh", "git"),
+            )
+
+    def test_rejects_a_non_public_host_on_an_allowed_scheme(self):
+        with pytest.raises(OutboundURLNotAllowedError):
+            validate_outbound_url("http://169.254.169.254/latest/meta-data")
+
+    def test_rejects_shared_address_space_on_an_allowed_scheme(self):
+        with pytest.raises(OutboundURLNotAllowedError):
+            validate_outbound_url("http://100.100.100.200/latest/meta-data")

tests/providers/iac/iac_provider_test.py

@@ -7,6 +7,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from prowler.lib.check.models import CheckReportIAC
+from prowler.lib.network.ssrf import OutboundURLNotAllowedError
 from prowler.providers.iac.exceptions.exceptions import (
     IacRepositoryCloneError,
     IacScanError,
@@ -784,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, config=mock.ANY, pool_manager=mock.ANY
+        )
         assert temp_dir == "/tmp/fake-dir"
         assert branch_name == "main"
 
@@ -798,7 +801,13 @@ 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,
+            config=mock.ANY,
+            pool_manager=mock.ANY,
+        )
         assert temp_dir == "/tmp/fake-dir"
         assert branch_name == "develop"
 
@@ -812,10 +821,34 @@ 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,
+            config=mock.ANY,
+            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):
@@ -828,13 +861,19 @@ class TestIacProvider:
         lets the API report the failure as a normal task error instead of a
         `SystemExit` escaping the Celery worker.
         """
-        mock_clone.side_effect = Exception("repository not found")
+        mock_clone.side_effect = Exception(
+            "https://x-access-token:SENTINEL_TOKEN@github.com/user/repo.git refused"
+        )
 
         with pytest.raises(IacRepositoryCloneError) as exc_info:
             IacProvider(scan_repository_url="https://github.com/user/repo.git")
 
-        assert "repository not found" in str(exc_info.value)
+        # ProwlerException formats original_exception into its str(), and the API
+        # returns that, so the authenticated URL must not reach it
+        assert "SENTINEL_TOKEN" not in str(exc_info.value)
+        assert "Exception" in str(exc_info.value)
         assert exc_info.value.code == 21000
+        assert isinstance(exc_info.value.__cause__, Exception)
 
     def test_detect_branch_name_main(self):
         """Test detecting 'main' branch from .git/HEAD"""
@@ -878,3 +917,95 @@ class TestIacProvider:
         # Pass a non-existent directory
         branch_name = provider._detect_branch_name("/non/existent/path")
         assert branch_name == "main"
+
+    def test_test_connection_rejects_loopback_url(self):
+        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"
+            )
+
+        assert connection.is_connected is False
+        assert connection.error == "Repository URL is not an allowed destination."
+        mock_ls_remote.assert_not_called()
+
+    def test_test_connection_rejects_private_range_url(self):
+        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"
+            )
+
+        assert connection.is_connected is False
+        assert connection.error == "Repository URL is not an allowed destination."
+        mock_ls_remote.assert_not_called()
+
+    def test_clone_repository_rejects_a_non_public_url(self):
+        provider = IacProvider.__new__(IacProvider)
+        with patch("prowler.providers.iac.iac_provider.porcelain.clone") as mock_clone:
+            with pytest.raises(OutboundURLNotAllowedError):
+                provider._clone_repository("https://169.254.169.254/org/repo.git")
+
+        mock_clone.assert_not_called()
+
+    def test_clone_repository_rejects_shared_address_space(self):
+        provider = IacProvider.__new__(IacProvider)
+        with patch("prowler.providers.iac.iac_provider.porcelain.clone") as mock_clone:
+            with pytest.raises(OutboundURLNotAllowedError):
+                provider._clone_repository("https://100.100.100.200/org/repo.git")
+
+        mock_clone.assert_not_called()
+
+    def test_clone_repository_does_not_validate_the_token_bearing_url(self):
+        provider = IacProvider.__new__(IacProvider)
+        with patch(
+            "prowler.lib.network.ssrf.socket.getaddrinfo",
+            return_value=[(None, None, None, None, ("140.82.121.4", 0))],
+        ) as mock_getaddrinfo:
+            with patch("prowler.providers.iac.iac_provider.porcelain.clone"):
+                with patch.object(
+                    IacProvider, "_detect_branch_name", return_value="main"
+                ):
+                    provider._clone_repository(
+                        "https://github.com/org/repo.git",
+                        github_username="user",
+                        personal_access_token="token",
+                    )
+
+        assert mock_getaddrinfo.call_args[0][0] == "github.com"
+
+    def test_test_connection_allows_public_url(self):
+        with (
+            patch(
+                "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.ls_remote") as mock_ls_remote,
+        ):
+            connection = IacProvider.test_connection(
+                scan_repository_url="https://github.com/user/repo.git"
+            )
+
+        assert connection.is_connected is True
+        mock_getaddrinfo.assert_called_once_with("github.com", None)
+        mock_ls_remote.assert_called_once_with("https://github.com/user/repo.git")
+
+    def test_test_connection_does_not_echo_raw_error(self):
+        with (
+            patch(
+                "prowler.lib.network.ssrf.socket.getaddrinfo",
+                return_value=[(None, None, None, None, ("140.82.121.4", 0))],
+            ),
+            patch(
+                "prowler.providers.iac.iac_provider.ls_remote",
+                side_effect=Exception(
+                    "https://x-access-token:SENTINEL_TOKEN@github.com/user/repo.git refused"
+                ),
+            ),
+        ):
+            connection = IacProvider.test_connection(
+                scan_repository_url="https://github.com/user/repo.git",
+                oauth_app_token="SENTINEL_TOKEN",
+            )
+
+        assert connection.is_connected is False
+        assert connection.error == "Failed to connect to repository."
+        assert "SENTINEL_TOKEN" not in connection.error

tests/providers/iac/lib/git_transport_test.py

@@ -0,0 +1,168 @@
+import http.server
+import socket
+import threading
+from unittest import mock
+
+import pytest
+from dulwich.config import ConfigDict
+
+from prowler.lib.network.ssrf import (
+    ALLOWED_PRIVATE_NETWORKS_ENV,
+    OutboundURLNotAllowedError,
+    validate_outbound_url,
+)
+from prowler.providers.iac.lib import git_transport
+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"
+# Kept apart from the URLs so the fixtures do not read as leaked credentials.
+USERINFO = "user:a-token"
+
+
+def _resolves_to(*addresses):
+    return mock.patch.object(
+        socket,
+        "getaddrinfo",
+        return_value=[(2, 1, 6, "", (address, 0)) for address in addresses],
+    )
+
+
+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://{USERINFO}@127.0.0.1:{origin.port}/repo.git")
+
+        assert origin.requests[0][1] is not None
+
+
+class TestEffectiveDestination:
+    def test_rejects_a_private_host_an_insteadof_rule_rewrites_to(self, monkeypatch):
+        """git rewrites the URL before dulwich picks the transport.
+
+        The supplied host resolves public on purpose, so only the rewrite can
+        make this fail.
+        """
+        config = ConfigDict()
+        config.set(
+            (b"url", b"ssh://10.0.0.5/"), b"insteadOf", b"https://reg.example.com/"
+        )
+        monkeypatch.setattr(git_transport, "git_config", lambda: config)
+
+        with _resolves_to("140.82.121.4"):
+            with pytest.raises(OutboundURLNotAllowedError, match="non-public"):
+                git_transport.ls_remote("https://reg.example.com/org/repo.git")
+
+    @pytest.mark.parametrize(
+        "url,expected",
+        [
+            (
+                f"https://{USERINFO}@github.com:8443/o/r.git",
+                "https://github.com:8443/o/r.git",
+            ),
+            # the path stays: an http.<url>.* section can be path-specific
+            (f"https://{USERINFO}@github.com/o/r.git", "https://github.com/o/r.git"),
+            # urlparse drops the brackets of an IPv6 literal
+            (f"https://{USERINFO}@[::1]:8443/o/r.git", "https://[::1]:8443/o/r.git"),
+            # nothing to strip, so nothing is rewritten
+            (
+                "https://[2606:4700:4700::1111]/o/r.git",
+                "https://[2606:4700:4700::1111]/o/r.git",
+            ),
+        ],
+    )
+    def test_strips_only_the_credentials_from_the_proxy_base_url(self, url, expected):
+        assert git_transport.proxy_base_url(url) == expected
+
+    @pytest.mark.parametrize("scheme", ["http", "https", "ssh", "git+ssh", "git"])
+    def test_allows_every_scheme_dulwich_can_transport(self, scheme, monkeypatch):
+        """A scheme dulwich routes must reach the host check, not be refused on scheme."""
+        monkeypatch.setattr(git_transport, "git_config", ConfigDict)
+
+        with _resolves_to("140.82.121.4"):
+            # the scheme must not be what rejects it; a public host is accepted
+            validate_outbound_url(
+                f"{scheme}://host.example/o/r.git",
+                allowed_schemes=git_transport.ALLOWED_SCHEMES,
+            )
+
+    def test_honours_a_no_proxy_bypass(self, monkeypatch, origin):
+        """Built without base_url, dulwich cannot see no_proxy and uses the proxy."""
+        monkeypatch.setenv("http_proxy", "http://127.0.0.1:1")
+        monkeypatch.setenv("no_proxy", "127.0.0.1")
+
+        with pytest.raises(Exception):
+            git_transport.ls_remote(f"{origin.url}/repo.git")
+
+        assert len(origin.requests) == 1