ben.carrasco/prowler
1
0 Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-10-14 15:34:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(vercel): handle team-scoped firewall config responses (#10695)

Browse Source
This commit is contained in:
Daniel Barranquero authored and GitHub committed 2026-04-15 11:59:20 +02:00
1 parent e6fc59267b
commit c3acb818d9
5 files changed
+411 -20
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

prowler/CHANGELOG.md
+3
View File
@@ -23,6 +23,9 @@ All notable changes to the **Prowler SDK** are documented in this file.
- Bump Poetry to `2.3.4` and consolidate SDK workflows onto the `setup-python-poetry` composite action with opt-in lockfile regeneration [(#10681)](https://github.com/prowler-cloud/prowler/pull/10681)
- Normalize Conditional Access platform values in Entra models and simplify platform-based checks [(#10635)](https://github.com/prowler-cloud/prowler/pull/10635)
### 🐞 Fixed
- Vercel firewall config handling for team-scoped projects and current API response shapes [(#10695)](https://github.com/prowler-cloud/prowler/pull/10695)
---
## [5.23.0] (Prowler v5.23.0)
prowler/providers/vercel/services/project/project_service.py
+15
View File
@@ -55,6 +55,7 @@ class Project(VercelService):
# Parse password protection
pwd_protection = proj.get("passwordProtection")
security = proj.get("security", {}) or {}
self.projects[project_id] = VercelProject(
id=project_id,
@@ -75,6 +76,16 @@ class Project(VercelService):
git_fork_protection=proj.get("gitForkProtection", True),
git_repository=proj.get("link"),
secure_compute=proj.get("secureCompute"),
firewall_enabled=security.get("firewallEnabled"),
firewall_config_version=(
str(security.get("firewallConfigVersion"))
if security.get("firewallConfigVersion") is not None
else None
),
managed_rules=security.get(
"managedRules", security.get("managedRulesets")
),
bot_id_enabled=security.get("botIdEnabled"),
)
logger.info(f"Project - Found {len(self.projects)} project(s)")
@@ -160,4 +171,8 @@ class VercelProject(BaseModel):
git_fork_protection: bool = True
git_repository: Optional[dict] = None
secure_compute: Optional[dict] = None
firewall_enabled: Optional[bool] = None
firewall_config_version: Optional[str] = None
managed_rules: Optional[dict] = None
bot_id_enabled: Optional[bool] = None
environment_variables: list[VercelEnvironmentVariable] = Field(default_factory=list)
prowler/providers/vercel/services/security/security_service.py
+149 -20
View File
@@ -26,10 +26,7 @@ class Security(VercelService):
def _fetch_firewall_config(self, project):
"""Fetch WAF/Firewall config for a single project."""
try:
data = self._get(
"/v1/security/firewall/config",
params={"projectId": project.id},
)
data = self._read_firewall_config(project)
if data is None:
# 403 — plan limitation, store with managed_rulesets=None
@@ -44,39 +41,60 @@ class Security(VercelService):
)
return
# Parse firewall config
fw = data.get("firewallConfig", data) if isinstance(data, dict) else {}
fw = self._normalize_firewall_config(data)
# Determine if firewall is enabled
rules = fw.get("rules", []) or []
managed = fw.get("managedRules", fw.get("managedRulesets"))
if not fw:
fallback_firewall_enabled = self._fallback_firewall_enabled(project)
self.firewall_configs[project.id] = VercelFirewallConfig(
project_id=project.id,
project_name=project.name,
team_id=project.team_id,
firewall_enabled=(
fallback_firewall_enabled
if fallback_firewall_enabled is not None
else False
),
managed_rulesets=self._fallback_managed_rulesets(project),
name=project.name,
id=project.id,
)
return
rules = [
rule for rule in (fw.get("rules", []) or []) if self._is_active(rule)
]
managed = self._active_managed_rulesets(
fw.get("managedRules", fw.get("managedRulesets", fw.get("crs")))
)
custom_rules = []
ip_blocking = []
ip_blocking = list(fw.get("ips", []) or [])
rate_limiting = []
for rule in rules:
rule_action = rule.get("action", {})
action_type = (
rule_action.get("type", "")
if isinstance(rule_action, dict)
else str(rule_action)
)
mitigate_action = self._mitigate_action(rule)
if action_type == "rate_limit" or rule.get("rateLimit"):
if self._is_rate_limiting_rule(rule, mitigate_action):
rate_limiting.append(rule)
elif action_type in ("deny", "block") and self._is_ip_rule(rule):
elif self._is_ip_rule(rule):
ip_blocking.append(rule)
else:
custom_rules.append(rule)
firewall_enabled = bool(rules) or bool(managed)
firewall_enabled = fw.get("firewallEnabled")
if firewall_enabled is None:
firewall_enabled = self._fallback_firewall_enabled(project)
if firewall_enabled is None:
firewall_enabled = bool(rules) or bool(ip_blocking) or bool(managed)
if not managed:
managed = self._fallback_managed_rulesets(project)
self.firewall_configs[project.id] = VercelFirewallConfig(
project_id=project.id,
project_name=project.name,
team_id=project.team_id,
firewall_enabled=firewall_enabled,
managed_rulesets=managed if managed is not None else {},
managed_rulesets=managed,
custom_rules=custom_rules,
ip_blocking_rules=ip_blocking,
rate_limiting_rules=rate_limiting,
@@ -95,6 +113,117 @@ class Security(VercelService):
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _read_firewall_config(self, project):
"""Read the deployed firewall config via the documented endpoint.
See: https://vercel.com/docs/rest-api/security/read-firewall-configuration
"""
params = self._firewall_params(project)
config_version = getattr(project, "firewall_config_version", None)
endpoints = []
if config_version:
endpoints.append(f"/v1/security/firewall/config/{config_version}")
endpoints.append("/v1/security/firewall/config/active")
last_error = None
for endpoint in endpoints:
try:
return self._get(endpoint, params=params)
except Exception as error:
last_error = error
logger.warning(
f"Security - Firewall config read failed for project "
f"{project.id} (team={getattr(project, 'team_id', None)}) "
f"on {endpoint} with params={params}: "
f"{error.__class__.__name__}: {error}"
)
if last_error is not None:
logger.debug(
f"Security - Falling back to firewall config wrapper for "
f"{project.id} after {last_error.__class__.__name__}: {last_error}"
)
return self._get("/v1/security/firewall/config", params=params)
@staticmethod
def _firewall_params(project) -> dict:
"""Build firewall request params, preserving team scope for team projects."""
params = {"projectId": project.id}
team_id = getattr(project, "team_id", None)
if isinstance(team_id, str) and team_id:
params["teamId"] = team_id
return params
@staticmethod
def _normalize_firewall_config(data: dict) -> dict:
"""Normalize firewall responses across Vercel endpoint variants."""
if not isinstance(data, dict):
return {}
if "firewallConfig" in data and isinstance(data["firewallConfig"], dict):
return data["firewallConfig"]
if any(key in data for key in ("active", "draft", "versions")):
return data.get("active") or {}
return data
@staticmethod
def _active_managed_rulesets(managed_rules: dict | None) -> dict:
"""Return only active managed rulesets."""
if not isinstance(managed_rules, dict):
return {}
return {
ruleset: config
for ruleset, config in managed_rules.items()
if not isinstance(config, dict) or config.get("active", False)
}
@classmethod
def _fallback_managed_rulesets(cls, project) -> dict:
"""Return active managed rulesets from project metadata."""
return cls._active_managed_rulesets(getattr(project, "managed_rules", None))
@staticmethod
def _fallback_firewall_enabled(project) -> bool | None:
"""Return firewall enabled state from project metadata when available."""
return getattr(project, "firewall_enabled", None)
@staticmethod
def _mitigate_action(rule: dict) -> dict:
"""Extract the nested Vercel mitigation action payload for a rule."""
action = rule.get("action", {})
if not isinstance(action, dict):
return {}
mitigate = action.get("mitigate")
return mitigate if isinstance(mitigate, dict) else action
@staticmethod
def _is_active(rule: dict) -> bool:
"""Treat missing active flags as enabled for backwards compatibility."""
return rule.get("active", True) is not False
@classmethod
def _is_rate_limiting_rule(
cls, rule: dict, mitigate_action: dict | None = None
) -> bool:
"""Check if a firewall rule enforces rate limiting."""
if rule.get("rateLimit"):
return True
mitigate = (
mitigate_action
if isinstance(mitigate_action, dict)
else cls._mitigate_action(rule)
)
return bool(mitigate.get("rateLimit")) or mitigate.get("action") == "rate_limit"
@staticmethod
def _is_ip_rule(rule: dict) -> bool:
"""Check if a rule is an IP blocking rule based on conditions."""
tests/providers/vercel/services/project/project_service_test.py
+45
View File
@@ -0,0 +1,45 @@
from unittest import mock
from prowler.providers.vercel.services.project.project_service import Project
from tests.providers.vercel.vercel_fixtures import (
PROJECT_ID,
PROJECT_NAME,
TEAM_ID,
set_mocked_vercel_provider,
)
class TestProjectService:
def test_list_projects_parses_security_metadata(self):
service = Project.__new__(Project)
service.provider = set_mocked_vercel_provider()
service.projects = {}
service._paginate = mock.MagicMock(
return_value=[
{
"id": PROJECT_ID,
"name": PROJECT_NAME,
"accountId": TEAM_ID,
"security": {
"firewallEnabled": True,
"firewallConfigVersion": 42,
"managedRules": {
"owasp": {"active": True, "action": "log"},
"ai_bots": {"active": False, "action": "deny"},
},
"botIdEnabled": True,
},
}
]
)
service._list_projects()
project = service.projects[PROJECT_ID]
assert project.firewall_enabled is True
assert project.firewall_config_version == "42"
assert project.managed_rules == {
"owasp": {"active": True, "action": "log"},
"ai_bots": {"active": False, "action": "deny"},
}
assert project.bot_id_enabled is True
tests/providers/vercel/services/security/security_service_test.py
+199
View File
@@ -0,0 +1,199 @@
from unittest import mock
from prowler.providers.vercel.services.project.project_service import VercelProject
from prowler.providers.vercel.services.security.security_service import Security
from tests.providers.vercel.vercel_fixtures import PROJECT_ID, PROJECT_NAME, TEAM_ID
class TestSecurityService:
def test_fetch_firewall_config_reads_active_version_and_normalizes_response(self):
project = VercelProject(id=PROJECT_ID, name=PROJECT_NAME, team_id=TEAM_ID)
service = Security.__new__(Security)
service.firewall_configs = {}
service._get = mock.MagicMock(
return_value={
"active": {
"firewallEnabled": True,
"managedRules": {
"owasp": {"active": True, "action": "deny"},
"ai_bots": {"active": False, "action": "deny"},
},
"rules": [
{
"id": "rule-custom",
"name": "Block admin access",
"active": True,
"conditionGroup": [
{
"conditions": [
{
"type": "path",
"op": "pre",
"value": "/admin",
}
]
}
],
"action": {
"mitigate": {
"action": "deny",
}
},
},
{
"id": "rule-rate-limit",
"name": "Rate limit login",
"active": True,
"conditionGroup": [
{
"conditions": [
{
"type": "path",
"op": "eq",
"value": "/login",
}
]
}
],
"action": {
"mitigate": {
"action": "deny",
"rateLimit": {
"algo": "fixed_window",
"window": 60,
"limit": 10,
},
}
},
},
],
"ips": [
{
"id": "ip-rule",
"ip": "203.0.113.7",
"action": "deny",
}
],
},
"draft": None,
"versions": [1],
}
)
service._fetch_firewall_config(project)
service._get.assert_called_once_with(
"/v1/security/firewall/config/active",
params={"projectId": PROJECT_ID, "teamId": TEAM_ID},
)
config = service.firewall_configs[PROJECT_ID]
assert config.firewall_enabled is True
assert config.managed_rulesets == {"owasp": {"active": True, "action": "deny"}}
assert [rule["id"] for rule in config.custom_rules] == ["rule-custom"]
assert [rule["id"] for rule in config.rate_limiting_rules] == [
"rule-rate-limit"
]
assert [rule["id"] for rule in config.ip_blocking_rules] == ["ip-rule"]
def test_fetch_firewall_config_parses_crs_managed_rulesets(self):
project = VercelProject(
id=PROJECT_ID,
name=PROJECT_NAME,
team_id=TEAM_ID,
firewall_config_version="1",
)
service = Security.__new__(Security)
service.firewall_configs = {}
service._get = mock.MagicMock(
return_value={
"id": "waf_test",
"version": 1,
"firewallEnabled": True,
"crs": {
"gen": {"active": True, "action": "log"},
"xss": {"active": True, "action": "deny"},
"php": {"active": False, "action": "log"},
},
"rules": [],
"ips": [],
}
)
service._fetch_firewall_config(project)
config = service.firewall_configs[PROJECT_ID]
assert config.firewall_enabled is True
assert config.managed_rulesets == {
"gen": {"active": True, "action": "log"},
"xss": {"active": True, "action": "deny"},
}
def test_fetch_firewall_config_falls_back_to_wrapper_when_active_missing(self):
project = VercelProject(id=PROJECT_ID, name=PROJECT_NAME, team_id=TEAM_ID)
service = Security.__new__(Security)
service.firewall_configs = {}
service._get = mock.MagicMock(
side_effect=[
Exception("404 active config not found"),
{"active": None, "draft": None, "versions": []},
]
)
service._fetch_firewall_config(project)
assert service._get.call_args_list == [
mock.call(
"/v1/security/firewall/config/active",
params={"projectId": PROJECT_ID, "teamId": TEAM_ID},
),
mock.call(
"/v1/security/firewall/config",
params={"projectId": PROJECT_ID, "teamId": TEAM_ID},
),
]
config = service.firewall_configs[PROJECT_ID]
assert config.firewall_enabled is False
assert config.managed_rulesets == {}
assert config.custom_rules == []
assert config.rate_limiting_rules == []
assert config.ip_blocking_rules == []
def test_fetch_firewall_config_uses_project_security_metadata_when_config_empty(
self,
):
project = VercelProject(
id=PROJECT_ID,
name=PROJECT_NAME,
team_id=TEAM_ID,
firewall_enabled=True,
firewall_config_version="42",
managed_rules={
"owasp": {"active": True, "action": "log"},
"ai_bots": {"active": False, "action": "deny"},
},
)
service = Security.__new__(Security)
service.firewall_configs = {}
service._get = mock.MagicMock(
return_value={"active": None, "draft": None, "versions": []}
)
service._fetch_firewall_config(project)
service._get.assert_called_once_with(
"/v1/security/firewall/config/42",
params={"projectId": PROJECT_ID, "teamId": TEAM_ID},
)
config = service.firewall_configs[PROJECT_ID]
assert config.firewall_enabled is True
assert config.managed_rulesets == {"owasp": {"active": True, "action": "log"}}
assert config.custom_rules == []
assert config.rate_limiting_rules == []
assert config.ip_blocking_rules == []