diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index cf95212c37..e19ef05621 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -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) diff --git a/prowler/providers/vercel/services/project/project_service.py b/prowler/providers/vercel/services/project/project_service.py index 9b3b24859d..e139a2a939 100644 --- a/prowler/providers/vercel/services/project/project_service.py +++ b/prowler/providers/vercel/services/project/project_service.py @@ -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) diff --git a/prowler/providers/vercel/services/security/security_service.py b/prowler/providers/vercel/services/security/security_service.py index eadac856b9..3c79f5c2f5 100644 --- a/prowler/providers/vercel/services/security/security_service.py +++ b/prowler/providers/vercel/services/security/security_service.py @@ -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.""" diff --git a/tests/providers/vercel/services/project/project_service_test.py b/tests/providers/vercel/services/project/project_service_test.py new file mode 100644 index 0000000000..b82588ddd9 --- /dev/null +++ b/tests/providers/vercel/services/project/project_service_test.py @@ -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 diff --git a/tests/providers/vercel/services/security/security_service_test.py b/tests/providers/vercel/services/security/security_service_test.py new file mode 100644 index 0000000000..d0a690b1db --- /dev/null +++ b/tests/providers/vercel/services/security/security_service_test.py @@ -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 == []