mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-30 07:14:20 +00:00
fix(openstack): block SSRF via auth_url in connection test (#12962)
This commit is contained in:
1 parent
be01a97789
commit
3bbd551f0f
5 files changed
+235
No files matched your search
@@ -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.
|
||||
</Warning>
|
||||
|
||||
### 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
|
||||
|
||||
@@ -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
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user