diff --git a/docs/user-guide/providers/iac/authentication.mdx b/docs/user-guide/providers/iac/authentication.mdx index cea25ec02f..e205b4323e 100644 --- a/docs/user-guide/providers/iac/authentication.mdx +++ b/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`. diff --git a/prowler/__main__.py b/prowler/__main__.py index b9de47c07b..669c1c81b9 100644 --- a/prowler/__main__.py +++ b/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() diff --git a/prowler/changelog.d/iac-connection-test-ssrf.security.md b/prowler/changelog.d/iac-connection-test-ssrf.security.md new file mode 100644 index 0000000000..70e571da6a --- /dev/null +++ b/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 diff --git a/prowler/lib/network/__init__.py b/prowler/lib/network/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/lib/network/ssrf.py b/prowler/lib/network/ssrf.py new file mode 100644 index 0000000000..691af46764 --- /dev/null +++ b/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[^:/]+):(?!//)") + +_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)) diff --git a/prowler/providers/iac/iac_provider.py b/prowler/providers/iac/iac_provider.py index f9477bc5ce..96de189725 100644 --- a/prowler/providers/iac/iac_provider.py +++ b/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.", ) diff --git a/prowler/providers/iac/lib/git_transport.py b/prowler/providers/iac/lib/git_transport.py new file mode 100644 index 0000000000..ef0ca09d9c --- /dev/null +++ b/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..*`` matching. + + Only the userinfo is removed. The path stays, because an ``http..*`` + 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) diff --git a/tests/lib/network/ssrf_test.py b/tests/lib/network/ssrf_test.py new file mode 100644 index 0000000000..6d3633b2bf --- /dev/null +++ b/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") diff --git a/tests/providers/iac/iac_provider_test.py b/tests/providers/iac/iac_provider_test.py index fe51aacb8c..6d3622dc72 100644 --- a/tests/providers/iac/iac_provider_test.py +++ b/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 diff --git a/tests/providers/iac/lib/git_transport_test.py b/tests/providers/iac/lib/git_transport_test.py new file mode 100644 index 0000000000..00851b8fec --- /dev/null +++ b/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..* 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