From 3bbd551f0fd0c2052e69fd37f77ca525481711cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20Mart=C3=ADn?= Date: Mon, 26 Oct 2026 13:37:46 +0100 Subject: [PATCH] fix(openstack): block SSRF via auth_url in connection test (#12962) --- .../providers/openstack/authentication.mdx | 34 ++++ .../openstack-auth-url-ssrf.security.md | 1 + .../openstack/exceptions/exceptions.py | 16 ++ .../providers/openstack/openstack_provider.py | 13 ++ .../openstack/openstack_provider_test.py | 171 ++++++++++++++++++ 5 files changed, 235 insertions(+) create mode 100644 prowler/changelog.d/openstack-auth-url-ssrf.security.md diff --git a/docs/user-guide/providers/openstack/authentication.mdx b/docs/user-guide/providers/openstack/authentication.mdx index 8c122f5e26..642ab707b6 100644 --- a/docs/user-guide/providers/openstack/authentication.mdx +++ b/docs/user-guide/providers/openstack/authentication.mdx @@ -611,6 +611,40 @@ If specific checks fail due to insufficient permissions: Using Public Cloud credentials can limit Keystone API access, so the command above may not work. Verify permissions in the provider's control panel instead. +### Private Keystone Endpoints + +Most OpenStack deployments keep Keystone on an internal network, which is the normal case rather than the exception. + +**Prowler CLI scans a private cloud with no extra configuration:** + +```console +export OS_AUTH_URL="https://keystone.internal:5000/v3" +prowler openstack +``` + +On the CLI the operator supplies the `auth_url` themselves, so there is nothing to defend against and the check below does not apply. + +**Prowler App rejects an `auth_url` that resolves to a non-public address**, as an SSRF defense, before the SDK connects and on both the connection test and the scan. There the URL arrives from a tenant and the worker must not be used to reach the internal network it runs in. To scan a private cloud 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 IaC providers. The Image provider has its own, `PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS`. + ## Next Steps - [Getting Started with OpenStack](/user-guide/providers/openstack/getting-started-openstack) - Run your first scan diff --git a/prowler/changelog.d/openstack-auth-url-ssrf.security.md b/prowler/changelog.d/openstack-auth-url-ssrf.security.md new file mode 100644 index 0000000000..f50f7e0ec2 --- /dev/null +++ b/prowler/changelog.d/openstack-auth-url-ssrf.security.md @@ -0,0 +1 @@ +OpenStack auth_url resolving to loopback, private or otherwise non-public hosts rejected before the SDK connects, on both the connection test and the scan diff --git a/prowler/providers/openstack/exceptions/exceptions.py b/prowler/providers/openstack/exceptions/exceptions.py index f5b7dc9a7d..00b4f9fe43 100644 --- a/prowler/providers/openstack/exceptions/exceptions.py +++ b/prowler/providers/openstack/exceptions/exceptions.py @@ -54,6 +54,10 @@ class OpenStackBaseException(ProwlerException): "message": "Ambiguous region configuration in clouds.yaml", "remediation": "Use either 'region_name' or 'regions' in your cloud configuration, not both.", }, + (17012, "OpenStackAuthUrlNotAllowedError"): { + "message": "OpenStack auth_url points at a destination Prowler must not reach", + "remediation": "Use an http or https auth_url that resolves to a public address, or allow the private network through PROWLER_ALLOWED_PRIVATE_NETWORKS.", + }, } def __init__(self, code, file=None, original_exception=None, message=None): @@ -212,3 +216,15 @@ class OpenStackAmbiguousRegionError(OpenStackBaseException): original_exception=original_exception, message=message, ) + + +class OpenStackAuthUrlNotAllowedError(OpenStackBaseException): + """Exception for an auth_url rejected by the outbound URL guard""" + + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + code=17012, + file=file, + original_exception=original_exception, + message=message, + ) diff --git a/prowler/providers/openstack/openstack_provider.py b/prowler/providers/openstack/openstack_provider.py index e11c4f7909..0b9edbf57c 100644 --- a/prowler/providers/openstack/openstack_provider.py +++ b/prowler/providers/openstack/openstack_provider.py @@ -14,12 +14,14 @@ 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 from prowler.providers.openstack.exceptions.exceptions import ( OpenStackAmbiguousRegionError, OpenStackAuthenticationError, + OpenStackAuthUrlNotAllowedError, OpenStackCloudNotFoundError, OpenStackConfigFileNotFoundError, OpenStackCredentialsError, @@ -454,6 +456,15 @@ class OpenstackProvider(Provider): message=f"Failed to load clouds.yaml configuration: {error}", ) + @staticmethod + def _validate_auth_url(auth_url: str) -> None: + """Reject an auth_url the worker must not reach; the detail stays in the log.""" + try: + validate_outbound_url(auth_url, allowed_schemes=("http", "https")) + except OutboundURLNotAllowedError as error: + logger.warning(f"OpenStack auth_url rejected: {error}") + raise OpenStackAuthUrlNotAllowedError() + @staticmethod def _create_connection( session: OpenStackSession, @@ -471,6 +482,7 @@ class OpenstackProvider(Provider): region: Optional region override — when given, the connection is scoped to this specific region instead of the session default. """ + OpenstackProvider._validate_auth_url(session.auth_url) try: # Don't load from clouds.yaml or environment variables, we configure this in setup_session() conn = connect( @@ -636,6 +648,7 @@ class OpenstackProvider(Provider): except ( OpenStackCredentialsError, OpenStackAuthenticationError, + OpenStackAuthUrlNotAllowedError, OpenStackSessionError, OpenStackConfigFileNotFoundError, OpenStackCloudNotFoundError, diff --git a/tests/providers/openstack/openstack_provider_test.py b/tests/providers/openstack/openstack_provider_test.py index b36fcd97a5..b61c9ca2e6 100644 --- a/tests/providers/openstack/openstack_provider_test.py +++ b/tests/providers/openstack/openstack_provider_test.py @@ -16,6 +16,7 @@ from prowler.providers.common.models import Connection from prowler.providers.openstack.exceptions.exceptions import ( OpenStackAmbiguousRegionError, OpenStackAuthenticationError, + OpenStackAuthUrlNotAllowedError, OpenStackCloudNotFoundError, OpenStackConfigFileNotFoundError, OpenStackCredentialsError, @@ -26,6 +27,23 @@ from prowler.providers.openstack.exceptions.exceptions import ( from prowler.providers.openstack.models import OpenStackIdentityInfo, OpenStackSession from prowler.providers.openstack.openstack_provider import OpenstackProvider +PUBLIC_IP = "8.8.8.8" + + +def _fake_getaddrinfo(host_to_ip: dict): + def _getaddrinfo(host, *_args, **_kwargs): + return [(None, None, None, None, (host_to_ip.get(host, PUBLIC_IP), 0))] + + return _getaddrinfo + + +@pytest.fixture(autouse=True) +def _dns_resolves_public(monkeypatch): + """Keep the outbound URL guard offline: every hostname resolves to a public IP.""" + monkeypatch.setattr( + "prowler.lib.network.ssrf.socket.getaddrinfo", _fake_getaddrinfo({}) + ) + class TestOpenstackProvider: """Test suite for OpenStack Provider initialization.""" @@ -1620,3 +1638,156 @@ clouds: assert result.is_connected is False assert isinstance(result.error, OpenStackInvalidProviderIdError) + + +class TestOpenstackProviderAuthUrlGuard: + """Test suite for the outbound URL guard applied to auth_url in test_connection.""" + + CLOUDS_YAML = """ +clouds: + test-cloud: + auth: + auth_url: {auth_url} + username: test-user + password: test-password + project_id: test-project-id + region_name: RegionOne +""" + + @staticmethod + def _test_connection(**kwargs) -> tuple[Connection, MagicMock]: + with patch( + "prowler.providers.openstack.openstack_provider.connect" + ) as mock_connect: + mock_connect.return_value = MagicMock() + result = OpenstackProvider.test_connection( + username="test-user", + password="test-password", + project_id="test-project-id", + region_name="RegionOne", + raise_on_exception=False, + **kwargs, + ) + return result, mock_connect + + @pytest.mark.parametrize( + "auth_url", + [ + "https://127.0.0.1:5000/v3", + "https://[::1]:5000/v3", + "https://10.0.0.5:5000/v3", + "https://172.16.0.5:5000/v3", + "https://192.168.1.5:5000/v3", + "http://169.254.169.254/latest/meta-data", + "ftp://keystone.example.com:5000/v3", + ], + ) + def test_test_connection_rejects_non_public_auth_url(self, auth_url): + result, mock_connect = self._test_connection(auth_url=auth_url) + + assert result.is_connected is False + assert isinstance(result.error, OpenStackAuthUrlNotAllowedError) + mock_connect.assert_not_called() + + def test_test_connection_rejects_hostname_resolving_to_private_ip( + self, monkeypatch + ): + monkeypatch.setattr( + "prowler.lib.network.ssrf.socket.getaddrinfo", + _fake_getaddrinfo({"keystone.example.com": "10.0.0.5"}), + ) + + result, mock_connect = self._test_connection( + auth_url="https://keystone.example.com:5000/v3" + ) + + assert result.is_connected is False + assert isinstance(result.error, OpenStackAuthUrlNotAllowedError) + mock_connect.assert_not_called() + + def test_test_connection_rejection_does_not_echo_the_target(self): + result, _ = self._test_connection( + auth_url="https://placeholder-user:placeholder-token@10.0.0.5:5000/v3" # trufflehog:ignore + ) + + assert isinstance(result.error, OpenStackAuthUrlNotAllowedError) + assert result.error.original_exception is None + for fragment in ("10.0.0.5", "placeholder-token", "non-public", "resolves"): + assert fragment not in str(result.error) + + def test_test_connection_rejection_raises_when_requested(self): + with patch("prowler.providers.openstack.openstack_provider.connect"): + with pytest.raises(OpenStackAuthUrlNotAllowedError): + OpenstackProvider.test_connection( + auth_url="https://127.0.0.1:5000/v3", + username="test-user", + password="test-password", + project_id="test-project-id", + region_name="RegionOne", + raise_on_exception=True, + ) + + def test_scan_initialisation_rejects_a_non_public_auth_url(self): + with patch( + "prowler.providers.openstack.openstack_provider.connect" + ) as mock_connect: + with pytest.raises(OpenStackAuthUrlNotAllowedError): + OpenstackProvider( + auth_url="https://169.254.169.254:5000/v3", + username="test-user", + password="test-password", + project_id="test-project-id", + region_name="RegionOne", + ) + + mock_connect.assert_not_called() + + def test_scan_initialisation_rejects_shared_address_space(self): + with patch( + "prowler.providers.openstack.openstack_provider.connect" + ) as mock_connect: + with pytest.raises(OpenStackAuthUrlNotAllowedError): + OpenstackProvider( + auth_url="https://100.100.100.200:5000/v3", + username="test-user", + password="test-password", + project_id="test-project-id", + region_name="RegionOne", + ) + + mock_connect.assert_not_called() + + def test_test_connection_allows_public_auth_url(self): + result, mock_connect = self._test_connection( + auth_url="https://openstack.example.com:5000/v3" + ) + + assert result.is_connected is True + assert result.error is None + mock_connect.assert_called_once() + + def test_test_connection_rejects_private_auth_url_from_clouds_yaml_content( + self, + ): + result, mock_connect = self._test_connection( + clouds_yaml_content=self.CLOUDS_YAML.format( + auth_url="https://127.0.0.1:5000/v3" + ), + clouds_yaml_cloud="test-cloud", + ) + + assert result.is_connected is False + assert isinstance(result.error, OpenStackAuthUrlNotAllowedError) + mock_connect.assert_not_called() + + def test_test_connection_allows_public_auth_url_from_clouds_yaml_content(self): + result, mock_connect = self._test_connection( + clouds_yaml_content=self.CLOUDS_YAML.format( + auth_url="https://openstack.example.com:5000/v3" + ), + clouds_yaml_cloud="test-cloud", + ) + + assert result.is_connected is True + assert result.error is None + mock_connect.assert_called_once()