fix(kubernetes): reject non-public kubeconfig servers

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,138 @@
+"""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"
+
+_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 _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
+    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:
+        raise OutboundURLNotAllowedError(f"Could not read a host from URL {url!r}")
+    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.
+    """
+    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 reached"
+            )
+
+
+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. 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/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()