diff --git a/prowler/lib/network/ssrf.py b/prowler/lib/network/ssrf.py index d039c58084..30f5e9b861 100644 --- a/prowler/lib/network/ssrf.py +++ b/prowler/lib/network/ssrf.py @@ -73,6 +73,10 @@ def _ip_is_non_public(address: str) -> bool: 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) diff --git a/tests/lib/network/ssrf_test.py b/tests/lib/network/ssrf_test.py new file mode 100644 index 0000000000..4b8005d0e3 --- /dev/null +++ b/tests/lib/network/ssrf_test.py @@ -0,0 +1,173 @@ +import ipaddress +import socket +from unittest import mock + +import pytest + +from prowler.lib.network.ssrf import ( + ALLOWED_PRIVATE_NETWORKS_ENV, + OutboundURLNotAllowedError, + allowed_private_networks, + extract_host, + 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 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") + + +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")