mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-04-16 17:47:47 +00:00
Compare commits
5 Commits
master
...
feat/prowl
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7a64c07287 | ||
|
|
ef6fd91071 | ||
|
|
3cfd4b8e0c | ||
|
|
318e1d1d71 | ||
|
|
96f94aba43 |
@@ -112,7 +112,7 @@ Every AWS provider scan will enqueue an Attack Paths ingestion job automatically
|
||||
| M365 | 89 | 9 | 4 | 5 | Official | UI, API, CLI |
|
||||
| OCI | 48 | 13 | 3 | 10 | Official | UI, API, CLI |
|
||||
| Alibaba Cloud | 61 | 9 | 3 | 9 | Official | UI, API, CLI |
|
||||
| Cloudflare | 29 | 2 | 0 | 5 | Official | UI, API, CLI |
|
||||
| Cloudflare | 30 | 3 | 0 | 5 | Official | UI, API, CLI |
|
||||
| IaC | [See `trivy` docs.](https://trivy.dev/latest/docs/coverage/iac/) | N/A | N/A | N/A | Official | UI, API, CLI |
|
||||
| MongoDB Atlas | 10 | 3 | 0 | 8 | Official | UI, API, CLI |
|
||||
| LLM | [See `promptfoo` docs.](https://www.promptfoo.dev/docs/red-team/plugins/) | N/A | N/A | N/A | Official | CLI |
|
||||
|
||||
@@ -30,6 +30,18 @@ Prowler requires read-only access to Cloudflare zones and their settings. The fo
|
||||
Ensure the API Token has access to all zones targeted for scanning. Missing permissions may cause some checks to fail or return incomplete results.
|
||||
</Warning>
|
||||
|
||||
### Checks Requiring API Key and Email Authentication
|
||||
|
||||
Some Cloudflare API endpoints are restricted to user-level authentication and cannot be accessed with scoped API Tokens. The following checks require [API Key and Email](#api-key-and-email-legacy) authentication:
|
||||
|
||||
| Check | Description |
|
||||
|-------|-------------|
|
||||
| `api_token_ip_restriction_enabled` | Verifies that API tokens have client IP address filtering configured |
|
||||
|
||||
<Note>
|
||||
When using API Token authentication, these checks are automatically skipped without affecting the rest of the scan.
|
||||
</Note>
|
||||
|
||||
---
|
||||
|
||||
## API Token (Recommended)
|
||||
|
||||
@@ -17,6 +17,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
|
||||
- `entra_conditional_access_policy_mfa_enforced_for_guest_users` check for M365 provider [(#10616)](https://github.com/prowler-cloud/prowler/pull/10616)
|
||||
- `entra_conditional_access_policy_corporate_device_sign_in_frequency_enforced` check for m365 provider [(#10618)](https://github.com/prowler-cloud/prowler/pull/10618)
|
||||
- `entra_conditional_access_policy_block_unknown_device_platforms` check for m365 provider [(#10615)](https://github.com/prowler-cloud/prowler/pull/10615)
|
||||
- `api_token_ip_restriction_enabled` check for cloudflare provider [(#10624)](https://github.com/prowler-cloud/prowler/pull/10624)
|
||||
|
||||
### 🔄 Changed
|
||||
|
||||
|
||||
4
prowler/providers/cloudflare/services/api/api_client.py
Normal file
4
prowler/providers/cloudflare/services/api/api_client.py
Normal file
@@ -0,0 +1,4 @@
|
||||
from prowler.providers.cloudflare.services.api.api_service import API
|
||||
from prowler.providers.common.provider import Provider
|
||||
|
||||
api_client = API(Provider.get_global_provider())
|
||||
67
prowler/providers/cloudflare/services/api/api_service.py
Normal file
67
prowler/providers/cloudflare/services/api/api_service.py
Normal file
@@ -0,0 +1,67 @@
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.providers.cloudflare.lib.service.service import CloudflareService
|
||||
|
||||
|
||||
class API(CloudflareService):
|
||||
"""Retrieve Cloudflare API tokens for the authenticated user."""
|
||||
|
||||
def __init__(self, provider):
|
||||
super().__init__(__class__.__name__, provider)
|
||||
self.tokens: list["CloudflareAPIToken"] = []
|
||||
self._list_api_tokens()
|
||||
|
||||
def _list_api_tokens(self) -> None:
|
||||
"""List all API tokens for the authenticated user."""
|
||||
logger.info("API - Listing API tokens...")
|
||||
try:
|
||||
seen_token_ids: set[str] = set()
|
||||
for token in self.client.user.tokens.list():
|
||||
token_id = getattr(token, "id", None)
|
||||
if not token_id:
|
||||
continue
|
||||
if token_id in seen_token_ids:
|
||||
continue
|
||||
seen_token_ids.add(token_id)
|
||||
|
||||
# Extract IP condition details
|
||||
condition = getattr(token, "condition", None)
|
||||
request_ip = getattr(condition, "request_ip", None) if condition else None
|
||||
ip_in = getattr(request_ip, "in_", None) if request_ip else None
|
||||
ip_not_in = getattr(request_ip, "not_in", None) if request_ip else None
|
||||
|
||||
self.tokens.append(
|
||||
CloudflareAPIToken(
|
||||
id=token_id,
|
||||
name=getattr(token, "name", None),
|
||||
status=getattr(token, "status", None),
|
||||
ip_allow_list=ip_in or [],
|
||||
ip_deny_list=ip_not_in or [],
|
||||
expires_on=getattr(token, "expires_on", None),
|
||||
issued_on=getattr(token, "issued_on", None),
|
||||
last_used_on=getattr(token, "last_used_on", None),
|
||||
modified_on=getattr(token, "modified_on", None),
|
||||
)
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
|
||||
class CloudflareAPIToken(BaseModel):
|
||||
"""Cloudflare API token representation."""
|
||||
|
||||
id: str
|
||||
name: Optional[str] = None
|
||||
status: Optional[str] = None
|
||||
ip_allow_list: list[str] = Field(default_factory=list)
|
||||
ip_deny_list: list[str] = Field(default_factory=list)
|
||||
expires_on: Optional[datetime] = None
|
||||
issued_on: Optional[datetime] = None
|
||||
last_used_on: Optional[datetime] = None
|
||||
modified_on: Optional[datetime] = None
|
||||
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"Provider": "cloudflare",
|
||||
"CheckID": "api_token_ip_restriction_enabled",
|
||||
"CheckTitle": "API token has client IP address filtering configured",
|
||||
"CheckType": [],
|
||||
"ServiceName": "api",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "APIToken",
|
||||
"ResourceGroup": "IAM",
|
||||
"Description": "**Cloudflare API tokens** with **client IP address filtering** restrict token usage to requests originating from trusted network locations, reducing exposure if a token is compromised.",
|
||||
"Risk": "Without **IP address filtering**, a compromised API token can be used from **any network location**.\n- **Confidentiality**: attackers with a stolen token can access account resources from any IP.\n- **Integrity**: unauthorized modifications can be made from uncontrolled networks.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://developers.cloudflare.com/fundamentals/api/get-started/create-token/"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Log in to the Cloudflare dashboard.\n2. Go to My Profile > API Tokens.\n3. Select the token to edit and click the three-dot menu > Edit.\n4. Under Client IP Address Filtering, add the IP addresses or CIDR ranges that should be allowed or denied.\n5. Save the token.",
|
||||
"Terraform": "```hcl\nresource \"cloudflare_api_token\" \"example_resource\" {\n name = \"example_resource\"\n\n policy {\n permission_groups = [\"<PERMISSION_GROUP_ID>\"]\n resources = { \"com.cloudflare.api.account.*\" = \"*\" }\n }\n\n # Restrict token to specific IP ranges\n condition {\n request_ip {\n in = [\"192.0.2.0/24\", \"198.51.100.0/24\"]\n }\n }\n}\n```"
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Configure **client IP address filtering** following the **principle of least privilege**. Restricting credentials to known IP ranges reduces the impact of credential compromise by ensuring they cannot be used from unauthorized networks.",
|
||||
"Url": "https://hub.prowler.com/checks/cloudflare/api_token_ip_restriction_enabled"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"identity-access"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "API tokens can have both allow lists and deny lists for IP filtering. This check requires API Key and Email authentication because the Cloudflare API does not expose user token listing to scoped API Tokens."
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
from prowler.lib.check.models import Check, CheckReportCloudflare
|
||||
from prowler.providers.cloudflare.services.api.api_client import api_client
|
||||
|
||||
|
||||
class api_token_ip_restriction_enabled(Check):
|
||||
"""Ensure that Cloudflare API tokens have client IP address filtering configured.
|
||||
|
||||
API tokens with IP address filtering restrict token usage to requests originating
|
||||
from specific IP addresses or CIDR ranges. Without IP filtering, a compromised
|
||||
token can be used from any network location.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[CheckReportCloudflare]:
|
||||
"""Execute the API token IP restriction check.
|
||||
|
||||
Iterates through all Cloudflare API tokens and verifies that each token
|
||||
has client IP address filtering configured via allow or deny lists.
|
||||
|
||||
Returns:
|
||||
A list of CheckReportCloudflare objects with PASS status if IP
|
||||
filtering is configured, or FAIL status if no IP restrictions exist.
|
||||
"""
|
||||
findings = []
|
||||
for token in api_client.tokens:
|
||||
report = CheckReportCloudflare(
|
||||
metadata=self.metadata(),
|
||||
resource=token,
|
||||
)
|
||||
|
||||
if token.ip_allow_list or token.ip_deny_list:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"API token {token.name or token.id} has client IP address filtering configured."
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"API token {token.name or token.id} does not have client IP address filtering configured."
|
||||
findings.append(report)
|
||||
return findings
|
||||
0
tests/providers/cloudflare/services/api/__init__.py
Normal file
0
tests/providers/cloudflare/services/api/__init__.py
Normal file
@@ -0,0 +1,211 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.cloudflare.services.api.api_service import (
|
||||
CloudflareAPIToken,
|
||||
)
|
||||
from tests.providers.cloudflare.cloudflare_fixtures import (
|
||||
set_mocked_cloudflare_provider,
|
||||
)
|
||||
|
||||
TOKEN_ID = "test-token-id"
|
||||
TOKEN_NAME = "Test API Token"
|
||||
|
||||
|
||||
class Test_api_token_ip_restriction_enabled:
|
||||
"""Tests for the api_token_ip_restriction_enabled check."""
|
||||
|
||||
def test_no_tokens(self):
|
||||
api_client = mock.MagicMock
|
||||
api_client.tokens = []
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled.api_client",
|
||||
new=api_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled import (
|
||||
api_token_ip_restriction_enabled,
|
||||
)
|
||||
|
||||
check = api_token_ip_restriction_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_token_with_ip_allow_list(self):
|
||||
api_client = mock.MagicMock
|
||||
api_client.tokens = [
|
||||
CloudflareAPIToken(
|
||||
id=TOKEN_ID,
|
||||
name=TOKEN_NAME,
|
||||
status="active",
|
||||
ip_allow_list=["192.0.2.0/24", "198.51.100.0/24"],
|
||||
ip_deny_list=[],
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled.api_client",
|
||||
new=api_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled import (
|
||||
api_token_ip_restriction_enabled,
|
||||
)
|
||||
|
||||
check = api_token_ip_restriction_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_id == TOKEN_ID
|
||||
assert result[0].resource_name == TOKEN_NAME
|
||||
assert result[0].status == "PASS"
|
||||
assert "has client IP address filtering configured" in result[0].status_extended
|
||||
|
||||
def test_token_with_ip_deny_list(self):
|
||||
api_client = mock.MagicMock
|
||||
api_client.tokens = [
|
||||
CloudflareAPIToken(
|
||||
id=TOKEN_ID,
|
||||
name=TOKEN_NAME,
|
||||
status="active",
|
||||
ip_allow_list=[],
|
||||
ip_deny_list=["10.0.0.0/8"],
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled.api_client",
|
||||
new=api_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled import (
|
||||
api_token_ip_restriction_enabled,
|
||||
)
|
||||
|
||||
check = api_token_ip_restriction_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_id == TOKEN_ID
|
||||
assert result[0].status == "PASS"
|
||||
assert "has client IP address filtering configured" in result[0].status_extended
|
||||
|
||||
def test_token_with_both_allow_and_deny_lists(self):
|
||||
api_client = mock.MagicMock
|
||||
api_client.tokens = [
|
||||
CloudflareAPIToken(
|
||||
id=TOKEN_ID,
|
||||
name=TOKEN_NAME,
|
||||
status="active",
|
||||
ip_allow_list=["192.0.2.0/24"],
|
||||
ip_deny_list=["10.0.0.0/8"],
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled.api_client",
|
||||
new=api_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled import (
|
||||
api_token_ip_restriction_enabled,
|
||||
)
|
||||
|
||||
check = api_token_ip_restriction_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert "has client IP address filtering configured" in result[0].status_extended
|
||||
|
||||
def test_token_without_ip_restriction(self):
|
||||
api_client = mock.MagicMock
|
||||
api_client.tokens = [
|
||||
CloudflareAPIToken(
|
||||
id=TOKEN_ID,
|
||||
name=TOKEN_NAME,
|
||||
status="active",
|
||||
ip_allow_list=[],
|
||||
ip_deny_list=[],
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled.api_client",
|
||||
new=api_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled import (
|
||||
api_token_ip_restriction_enabled,
|
||||
)
|
||||
|
||||
check = api_token_ip_restriction_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_id == TOKEN_ID
|
||||
assert result[0].resource_name == TOKEN_NAME
|
||||
assert result[0].status == "FAIL"
|
||||
assert "does not have client IP address filtering configured" in result[0].status_extended
|
||||
|
||||
def test_multiple_tokens_mixed(self):
|
||||
api_client = mock.MagicMock
|
||||
api_client.tokens = [
|
||||
CloudflareAPIToken(
|
||||
id="token-1",
|
||||
name="Restricted Token",
|
||||
status="active",
|
||||
ip_allow_list=["192.0.2.0/24"],
|
||||
ip_deny_list=[],
|
||||
),
|
||||
CloudflareAPIToken(
|
||||
id="token-2",
|
||||
name="Unrestricted Token",
|
||||
status="active",
|
||||
ip_allow_list=[],
|
||||
ip_deny_list=[],
|
||||
),
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled.api_client",
|
||||
new=api_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.api.api_token_ip_restriction_enabled.api_token_ip_restriction_enabled import (
|
||||
api_token_ip_restriction_enabled,
|
||||
)
|
||||
|
||||
check = api_token_ip_restriction_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 2
|
||||
assert result[0].resource_id == "token-1"
|
||||
assert result[0].status == "PASS"
|
||||
assert result[1].resource_id == "token-2"
|
||||
assert result[1].status == "FAIL"
|
||||
@@ -0,0 +1,251 @@
|
||||
from datetime import datetime, timezone
|
||||
from unittest import mock
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from prowler.providers.cloudflare.services.api.api_service import (
|
||||
API,
|
||||
CloudflareAPIToken,
|
||||
)
|
||||
from tests.providers.cloudflare.cloudflare_fixtures import (
|
||||
set_mocked_cloudflare_provider,
|
||||
)
|
||||
|
||||
|
||||
class Test_CloudflareAPIToken_Model:
|
||||
def test_cloudflare_api_token_model(self):
|
||||
token = CloudflareAPIToken(
|
||||
id="token-123",
|
||||
name="My API Token",
|
||||
status="active",
|
||||
ip_allow_list=["192.0.2.0/24"],
|
||||
ip_deny_list=["10.0.0.0/8"],
|
||||
expires_on=datetime(2026, 12, 31, tzinfo=timezone.utc),
|
||||
issued_on=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
last_used_on=datetime(2026, 4, 1, tzinfo=timezone.utc),
|
||||
modified_on=datetime(2026, 3, 15, tzinfo=timezone.utc),
|
||||
)
|
||||
assert token.id == "token-123"
|
||||
assert token.name == "My API Token"
|
||||
assert token.status == "active"
|
||||
assert token.ip_allow_list == ["192.0.2.0/24"]
|
||||
assert token.ip_deny_list == ["10.0.0.0/8"]
|
||||
assert token.expires_on == datetime(2026, 12, 31, tzinfo=timezone.utc)
|
||||
assert token.issued_on == datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
assert token.last_used_on == datetime(2026, 4, 1, tzinfo=timezone.utc)
|
||||
assert token.modified_on == datetime(2026, 3, 15, tzinfo=timezone.utc)
|
||||
|
||||
def test_cloudflare_api_token_defaults(self):
|
||||
token = CloudflareAPIToken(id="token-456")
|
||||
assert token.id == "token-456"
|
||||
assert token.name is None
|
||||
assert token.status is None
|
||||
assert token.ip_allow_list == []
|
||||
assert token.ip_deny_list == []
|
||||
assert token.expires_on is None
|
||||
assert token.issued_on is None
|
||||
assert token.last_used_on is None
|
||||
assert token.modified_on is None
|
||||
|
||||
def test_cloudflare_api_token_with_ip_allow_list_only(self):
|
||||
token = CloudflareAPIToken(
|
||||
id="token-789",
|
||||
name="Restricted Token",
|
||||
status="active",
|
||||
ip_allow_list=["192.0.2.0/24", "198.51.100.0/24"],
|
||||
)
|
||||
assert token.ip_allow_list == ["192.0.2.0/24", "198.51.100.0/24"]
|
||||
assert token.ip_deny_list == []
|
||||
|
||||
def test_cloudflare_api_token_with_ip_deny_list_only(self):
|
||||
token = CloudflareAPIToken(
|
||||
id="token-101",
|
||||
name="Deny-list Token",
|
||||
status="active",
|
||||
ip_deny_list=["10.0.0.0/8", "172.16.0.0/12"],
|
||||
)
|
||||
assert token.ip_allow_list == []
|
||||
assert token.ip_deny_list == ["10.0.0.0/8", "172.16.0.0/12"]
|
||||
|
||||
def test_cloudflare_api_token_disabled_status(self):
|
||||
token = CloudflareAPIToken(
|
||||
id="token-disabled",
|
||||
name="Disabled Token",
|
||||
status="disabled",
|
||||
)
|
||||
assert token.status == "disabled"
|
||||
|
||||
def test_cloudflare_api_token_expired_status(self):
|
||||
token = CloudflareAPIToken(
|
||||
id="token-expired",
|
||||
name="Expired Token",
|
||||
status="expired",
|
||||
expires_on=datetime(2025, 1, 1, tzinfo=timezone.utc),
|
||||
)
|
||||
assert token.status == "expired"
|
||||
assert token.expires_on == datetime(2025, 1, 1, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
class Test_API_Service:
|
||||
"""Tests for the API service _list_api_tokens method."""
|
||||
|
||||
def _make_sdk_token(self, **kwargs):
|
||||
"""Create a mock Cloudflare SDK token object."""
|
||||
token = MagicMock()
|
||||
for key, value in kwargs.items():
|
||||
setattr(token, key, value)
|
||||
return token
|
||||
|
||||
def test_list_api_tokens_with_ip_conditions(self):
|
||||
mock_provider = set_mocked_cloudflare_provider()
|
||||
request_ip = MagicMock()
|
||||
request_ip.in_ = ["192.0.2.0/24"]
|
||||
request_ip.not_in = ["10.0.0.0/8"]
|
||||
condition = MagicMock()
|
||||
condition.request_ip = request_ip
|
||||
|
||||
sdk_token = self._make_sdk_token(
|
||||
id="token-1",
|
||||
name="Test Token",
|
||||
status="active",
|
||||
condition=condition,
|
||||
expires_on=None,
|
||||
issued_on=None,
|
||||
last_used_on=None,
|
||||
modified_on=None,
|
||||
)
|
||||
mock_provider.session.client.user.tokens.list.return_value = [sdk_token]
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mock_provider,
|
||||
):
|
||||
service = API(mock_provider)
|
||||
|
||||
assert len(service.tokens) == 1
|
||||
assert service.tokens[0].id == "token-1"
|
||||
assert service.tokens[0].name == "Test Token"
|
||||
assert service.tokens[0].ip_allow_list == ["192.0.2.0/24"]
|
||||
assert service.tokens[0].ip_deny_list == ["10.0.0.0/8"]
|
||||
|
||||
def test_list_api_tokens_without_condition(self):
|
||||
mock_provider = set_mocked_cloudflare_provider()
|
||||
sdk_token = self._make_sdk_token(
|
||||
id="token-2",
|
||||
name="No Condition Token",
|
||||
status="active",
|
||||
condition=None,
|
||||
expires_on=None,
|
||||
issued_on=None,
|
||||
last_used_on=None,
|
||||
modified_on=None,
|
||||
)
|
||||
mock_provider.session.client.user.tokens.list.return_value = [sdk_token]
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mock_provider,
|
||||
):
|
||||
service = API(mock_provider)
|
||||
|
||||
assert len(service.tokens) == 1
|
||||
assert service.tokens[0].ip_allow_list == []
|
||||
assert service.tokens[0].ip_deny_list == []
|
||||
|
||||
def test_list_api_tokens_skips_duplicates(self):
|
||||
mock_provider = set_mocked_cloudflare_provider()
|
||||
sdk_token_1 = self._make_sdk_token(
|
||||
id="token-dup",
|
||||
name="First",
|
||||
status="active",
|
||||
condition=None,
|
||||
expires_on=None,
|
||||
issued_on=None,
|
||||
last_used_on=None,
|
||||
modified_on=None,
|
||||
)
|
||||
sdk_token_2 = self._make_sdk_token(
|
||||
id="token-dup",
|
||||
name="Duplicate",
|
||||
status="active",
|
||||
condition=None,
|
||||
expires_on=None,
|
||||
issued_on=None,
|
||||
last_used_on=None,
|
||||
modified_on=None,
|
||||
)
|
||||
sdk_token_3 = self._make_sdk_token(
|
||||
id="token-other",
|
||||
name="Other",
|
||||
status="active",
|
||||
condition=None,
|
||||
expires_on=None,
|
||||
issued_on=None,
|
||||
last_used_on=None,
|
||||
modified_on=None,
|
||||
)
|
||||
mock_provider.session.client.user.tokens.list.return_value = [
|
||||
sdk_token_1,
|
||||
sdk_token_2,
|
||||
sdk_token_3,
|
||||
]
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mock_provider,
|
||||
):
|
||||
service = API(mock_provider)
|
||||
|
||||
assert len(service.tokens) == 2
|
||||
assert service.tokens[0].id == "token-dup"
|
||||
assert service.tokens[0].name == "First"
|
||||
assert service.tokens[1].id == "token-other"
|
||||
|
||||
def test_list_api_tokens_skips_none_id(self):
|
||||
mock_provider = set_mocked_cloudflare_provider()
|
||||
sdk_token_no_id = self._make_sdk_token(
|
||||
id=None,
|
||||
name="No ID",
|
||||
status="active",
|
||||
condition=None,
|
||||
expires_on=None,
|
||||
issued_on=None,
|
||||
last_used_on=None,
|
||||
modified_on=None,
|
||||
)
|
||||
sdk_token_valid = self._make_sdk_token(
|
||||
id="token-valid",
|
||||
name="Valid",
|
||||
status="active",
|
||||
condition=None,
|
||||
expires_on=None,
|
||||
issued_on=None,
|
||||
last_used_on=None,
|
||||
modified_on=None,
|
||||
)
|
||||
mock_provider.session.client.user.tokens.list.return_value = [
|
||||
sdk_token_no_id,
|
||||
sdk_token_valid,
|
||||
]
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mock_provider,
|
||||
):
|
||||
service = API(mock_provider)
|
||||
|
||||
assert len(service.tokens) == 1
|
||||
assert service.tokens[0].id == "token-valid"
|
||||
|
||||
def test_list_api_tokens_handles_exception(self):
|
||||
mock_provider = set_mocked_cloudflare_provider()
|
||||
mock_provider.session.client.user.tokens.list.side_effect = Exception(
|
||||
"API error"
|
||||
)
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=mock_provider,
|
||||
):
|
||||
service = API(mock_provider)
|
||||
|
||||
assert service.tokens == []
|
||||
Reference in New Issue
Block a user