chore: re-run ci

.env

@@ -202,3 +202,8 @@ LANGCHAIN_PROJECT=""
 RSS_FEED_SOURCES='[{"id":"prowler-releases","name":"Prowler Releases","type":"github_releases","url":"https://github.com/prowler-cloud/prowler/releases.atom","enabled":true}]'
 # Example with multiple sources (no trailing comma after last item):
 # RSS_FEED_SOURCES='[{"id":"prowler-releases","name":"Prowler Releases","type":"github_releases","url":"https://github.com/prowler-cloud/prowler/releases.atom","enabled":true},{"id":"prowler-blog","name":"Prowler Blog","type":"blog","url":"https://prowler.com/blog/rss","enabled":false}]'
+
+# Comma-separated IPs and CIDRs the SSRF guard must not block, for scanning
+# Kubernetes, IaC or OpenStack targets that live on a private network.
+# Read by the worker, which runs the scan, not by the API.
+PROWLER_ALLOWED_PRIVATE_NETWORKS=""

docs/user-guide/providers/kubernetes/misc.mdx

@@ -43,3 +43,32 @@ prowler kubernetes ...
 This will allow Prowler to connect to the cluster even if the proxy uses a self-signed certificate.
 
 These environment variables are supported both when using an external `kubeconfig` and in in-cluster mode.
+
+## Private Cluster Endpoints
+
+Clusters reached over a private network are a normal setup.
+
+**Prowler CLI scans them with no extra configuration.** On the CLI the operator supplies the `kubeconfig` themselves, so there is nothing to defend against and the check below does not apply.
+
+**Prowler App rejects a `kubeconfig` whose `clusters[*].cluster.server` resolves to a non-public address**, as an SSRF defense. There the kube-config arrives from a tenant and the worker must not be used to reach the internal network it runs in. To scan a private cluster from Prowler App, declare the trusted ranges explicitly:
+
+```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. A kubeconfig declaring several clusters is rejected when any one of them resolves to a non-public address outside the allowlist. 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 IaC and OpenStack providers. The Image provider has its own, `PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS`.
+
+The check resolves the cluster hostname locally, before the Kubernetes client connects. If egress is only possible through `HTTPS_PROXY` and the hostname cannot be resolved locally, make the name resolvable to the scanning process. An allowlisted range does not help here: a host that does not resolve is rejected before any address is compared against the allowlist.

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/kubernetes-kubeconfig-server-ssrf.security.md

@@ -0,0 +1 @@
+Kubernetes kubeconfig cluster servers pointing at loopback, private or otherwise non-public addresses rejected before the client connects

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/kubernetes/exceptions/exceptions.py

@@ -34,6 +34,10 @@ class KubernetesBaseException(ProwlerException):
             "message": "The provided kube-config is invalid.",
             "remediation": "Review the kube-config and the attached error to get more details. Please, refer to the Kubernetes config documentation: https://kubernetes.io/docs/reference/config-api/kubeconfig.v1/#Config",
         },
+        (4007, "KubernetesKubeConfigServerNotAllowedError"): {
+            "message": "The provided kube-config points to a cluster server that is not an allowed destination.",
+            "remediation": "Make sure every cluster server in the kube-config is a public HTTP or HTTPS endpoint. To scan a cluster that lives on a private network, declare the trusted ranges in the PROWLER_ALLOWED_PRIVATE_NETWORKS environment variable of the process running the scan. Please, refer to the Kubernetes config documentation: https://kubernetes.io/docs/reference/config-api/kubeconfig.v1/#Config",
+        },
     }
 
     def __init__(
@@ -89,3 +93,8 @@ class KubernetesInvalidProviderIdError(KubernetesBaseException):
 class KubernetesInvalidKubeConfigFileError(KubernetesBaseException):
     def __init__(self, file=None, original_exception=None, message=None):
         super().__init__(4006, file, original_exception, message)
+
+
+class KubernetesKubeConfigServerNotAllowedError(KubernetesBaseException):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(4007, file, original_exception, message)

prowler/providers/kubernetes/kubernetes_provider.py

@@ -15,6 +15,7 @@ from prowler.config.config import (
     load_and_validate_config_file,
 )
 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.providers.common.models import Audit_Metadata, Connection
 from prowler.providers.common.provider import Provider
@@ -24,6 +25,7 @@ from prowler.providers.kubernetes.exceptions.exceptions import (
     KubernetesError,
     KubernetesInvalidKubeConfigFileError,
     KubernetesInvalidProviderIdError,
+    KubernetesKubeConfigServerNotAllowedError,
     KubernetesSetUpSessionError,
     KubernetesTimeoutError,
 )
@@ -255,6 +257,7 @@ class KubernetesProvider(Provider):
             if kubeconfig_content:
                 logger.info("Using kubeconfig content...")
                 config_data = safe_load(kubeconfig_content)
+                KubernetesProvider.validate_cluster_servers(config_data)
                 config.load_kube_config_from_dict(config_data, context=context)
                 if context:
                     contexts = config_data.get("contexts", [])
@@ -327,6 +330,11 @@ class KubernetesProvider(Provider):
                     api_client=ApiClient(configuration), context=context
                 )
 
+        except OutboundURLNotAllowedError as server_error:
+            logger.warning(f"Rejected kubeconfig cluster server: {server_error}")
+            raise KubernetesKubeConfigServerNotAllowedError(
+                file=os.path.abspath(__file__)
+            )
         except parser.ParserError as parser_error:
             logger.critical(
                 f"{parser_error.__class__.__name__}[{parser_error.__traceback__.tb_lineno}]: {parser_error}"
@@ -356,6 +364,14 @@ class KubernetesProvider(Provider):
                 original_exception=error, file=os.path.abspath(__file__)
             )
 
+    @staticmethod
+    def validate_cluster_servers(config_data: dict) -> None:
+        """Reject any cluster server URL that is not a public HTTP(S) endpoint."""
+        for cluster in (config_data or {}).get("clusters") or []:
+            server = ((cluster or {}).get("cluster") or {}).get("server")
+            if server:
+                validate_outbound_url(server, allowed_schemes=("http", "https"))
+
     @staticmethod
     def test_connection(
         kubeconfig_file: str = "~/.kube/config",
@@ -440,6 +456,13 @@ class KubernetesProvider(Provider):
             if raise_on_exception:
                 raise invalid_provider_id_error
             return Connection(error=invalid_provider_id_error)
+        except KubernetesKubeConfigServerNotAllowedError as server_not_allowed_error:
+            logger.critical(
+                f"KubernetesKubeConfigServerNotAllowedError[{server_not_allowed_error.__traceback__.tb_lineno}]: {server_not_allowed_error}"
+            )
+            if raise_on_exception:
+                raise server_not_allowed_error
+            return Connection(error=server_not_allowed_error)
         except KubernetesSetUpSessionError as setup_session_error:
             logger.critical(
                 f"KubernetesSetUpSessionError[{setup_session_error.__traceback__.tb_lineno}]: {setup_session_error}"

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/kubernetes/kubernetes_provider_test.py

@@ -1,6 +1,9 @@
+import socket
 from argparse import Namespace
 from unittest.mock import patch
 
+import pytest
+import yaml
 from kubernetes.config.config_exception import ConfigException
 
 from kubernetes import client
@@ -10,6 +13,7 @@ from prowler.config.config import (
     load_and_validate_config_file,
 )
 from prowler.providers.kubernetes.exceptions.exceptions import (
+    KubernetesKubeConfigServerNotAllowedError,
     KubernetesSetUpSessionError,
 )
 from prowler.providers.kubernetes.kubernetes_provider import KubernetesProvider
@@ -28,6 +32,35 @@ def mock_get_context_user_roles(*_):
     return []
 
 
+@pytest.fixture(autouse=True)
+def _default_dns_resolves_public(monkeypatch):
+    """Resolve every kubeconfig host to a public IP so the guard never hits real DNS."""
+
+    def _stub(_host, *_a, **_kw):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("8.8.8.8", 0))]
+
+    monkeypatch.setattr("prowler.lib.network.ssrf.socket.getaddrinfo", _stub)
+
+
+def build_kubeconfig(servers: list) -> dict:
+    return {
+        "apiVersion": "v1",
+        "kind": "Config",
+        "clusters": [
+            {"cluster": {"server": server}, "name": f"cluster-{index}"}
+            for index, server in enumerate(servers)
+        ],
+        "contexts": [
+            {
+                "context": {"cluster": "cluster-0", "user": "example-user"},
+                "name": "example-context",
+            }
+        ],
+        "current-context": "example-context",
+        "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}],
+    }
+
+
 class TestKubernetesProvider:
     def test_kubernetes_provider_no_namespaces(
         self,
@@ -673,3 +706,107 @@ class TestKubernetesProvider:
 
             assert config.proxy == proxy_url
             assert config.verify_ssl is False
+
+
+class TestKubernetesProviderClusterServerGuard:
+    @patch(
+        "prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
+    )
+    @patch("kubernetes.config.load_kube_config_from_dict")
+    def test_public_cluster_server_is_allowed(
+        self, mock_load_kube_config_from_dict, mock_list_namespace
+    ):
+        mock_list_namespace.return_value.items = []
+
+        connection = KubernetesProvider.test_connection(
+            kubeconfig_file=None,
+            kubeconfig_content=yaml.safe_dump(
+                build_kubeconfig(["https://kubernetes.example.com"])
+            ),
+            provider_id="example-context",
+            raise_on_exception=False,
+        )
+
+        assert connection.is_connected
+        assert connection.error is None
+        mock_load_kube_config_from_dict.assert_called_once()
+
+    @pytest.mark.parametrize(
+        "server",
+        [
+            "https://127.0.0.1:6443",
+            "https://10.0.0.5:6443",
+            "https://[::1]:6443",
+            "https://169.254.169.254/latest/meta-data",
+        ],
+    )
+    @patch("kubernetes.config.load_kube_config_from_dict")
+    def test_non_public_cluster_server_is_rejected_before_loading(
+        self, mock_load_kube_config_from_dict, server
+    ):
+        connection = KubernetesProvider.test_connection(
+            kubeconfig_file=None,
+            kubeconfig_content=yaml.safe_dump(build_kubeconfig([server])),
+            provider_id="example-context",
+            raise_on_exception=False,
+        )
+
+        assert not connection.is_connected
+        assert isinstance(connection.error, KubernetesKubeConfigServerNotAllowedError)
+        assert server not in str(connection.error)
+        assert connection.error.original_exception is None
+        mock_load_kube_config_from_dict.assert_not_called()
+
+    @patch("kubernetes.config.load_kube_config_from_dict")
+    def test_multi_cluster_kubeconfig_with_one_internal_server_is_rejected(
+        self, mock_load_kube_config_from_dict
+    ):
+        connection = KubernetesProvider.test_connection(
+            kubeconfig_file=None,
+            kubeconfig_content=yaml.safe_dump(
+                build_kubeconfig(
+                    ["https://kubernetes.example.com", "https://192.168.1.10:6443"]
+                )
+            ),
+            provider_id="example-context",
+            raise_on_exception=False,
+        )
+
+        assert not connection.is_connected
+        assert isinstance(connection.error, KubernetesKubeConfigServerNotAllowedError)
+        assert "192.168.1.10" not in str(connection.error)
+        mock_load_kube_config_from_dict.assert_not_called()
+
+    @patch("kubernetes.config.load_kube_config_from_dict")
+    def test_non_http_cluster_server_scheme_is_rejected(
+        self, mock_load_kube_config_from_dict
+    ):
+        connection = KubernetesProvider.test_connection(
+            kubeconfig_file=None,
+            kubeconfig_content=yaml.safe_dump(
+                build_kubeconfig(["file:///etc/kubernetes/admin.conf"])
+            ),
+            provider_id="example-context",
+            raise_on_exception=False,
+        )
+
+        assert not connection.is_connected
+        assert isinstance(connection.error, KubernetesKubeConfigServerNotAllowedError)
+        mock_load_kube_config_from_dict.assert_not_called()
+
+    @patch("kubernetes.config.load_kube_config_from_dict")
+    def test_non_public_cluster_server_raises_when_requested(
+        self, mock_load_kube_config_from_dict
+    ):
+        with pytest.raises(KubernetesKubeConfigServerNotAllowedError) as error:
+            KubernetesProvider.test_connection(
+                kubeconfig_file=None,
+                kubeconfig_content=yaml.safe_dump(
+                    build_kubeconfig(["https://127.0.0.1:6443"])
+                ),
+                provider_id="example-context",
+                raise_on_exception=True,
+            )
+
+        assert "127.0.0.1" not in str(error.value)
+        mock_load_kube_config_from_dict.assert_not_called()