feat(api): discover compliance frameworks from available providers

- Source compliance template, checks mapping, and overview filter
  validation from the SDK's available providers
- Replace static provider enum loops with get_available_providers()
- Cover dynamic external-provider compliance discovery with tests
This commit is contained in:
StylusFrost committed 2026-06-01 13:44:56 +02:00
1 parent b8d3312577
commit 58fb424ed7
3 files changed
+59 -14

No files matched your search

+5 -4
View File
@@ -3,6 +3,7 @@ from collections.abc import Iterable, Mapping
from api.models import Provider
from prowler.lib.check.compliance_models import Compliance
from prowler.lib.check.models import CheckMetadata
from prowler.providers.common.provider import Provider as SDKProvider
AVAILABLE_COMPLIANCE_FRAMEWORKS = {}
@@ -12,7 +13,7 @@ class LazyComplianceTemplate(Mapping):
def __init__(self, provider_types: Iterable[str] | None = None) -> None:
if provider_types is None:
provider_types = Provider.ProviderChoices.values
provider_types = SDKProvider.get_available_providers()
self._provider_types = tuple(provider_types)
self._provider_types_set = set(self._provider_types)
self._cache: dict[str, dict] = {}
@@ -53,7 +54,7 @@ class LazyChecksMapping(Mapping):
def __init__(self, provider_types: Iterable[str] | None = None) -> None:
if provider_types is None:
provider_types = Provider.ProviderChoices.values
provider_types = SDKProvider.get_available_providers()
self._provider_types = tuple(provider_types)
self._provider_types_set = set(self._provider_types)
self._cache: dict[str, dict] = {}
@@ -201,7 +202,7 @@ def load_prowler_checks(
"""
checks = {}
if provider_types is None:
provider_types = Provider.ProviderChoices.values
provider_types = SDKProvider.get_available_providers()
for provider_type in provider_types:
checks[provider_type] = {
check_id: set() for check_id in get_prowler_provider_checks(provider_type)
@@ -280,7 +281,7 @@ def generate_compliance_overview_template(
"""
template = {}
if provider_types is None:
provider_types = Provider.ProviderChoices.values
provider_types = SDKProvider.get_available_providers()
for provider_type in provider_types:
provider_compliance = template.setdefault(provider_type, {})
compliance_data_dict = prowler_compliance.get(provider_type, {})
+51 -8
View File
@@ -4,6 +4,8 @@ import pytest
from api import compliance as compliance_module
from api.compliance import (
LazyChecksMapping,
LazyComplianceTemplate,
generate_compliance_overview_template,
generate_scan_compliance,
get_compliance_frameworks,
@@ -40,12 +42,10 @@ class TestCompliance:
mock_compliance.get_bulk.assert_called_once_with(provider_type)
@patch("api.compliance.get_prowler_provider_checks")
@patch("api.models.Provider.ProviderChoices")
@patch("api.compliance.SDKProvider.get_available_providers", return_value=["aws"])
def test_load_prowler_checks(
self, mock_provider_choices, mock_get_prowler_provider_checks
self, _mock_get_available_providers, mock_get_prowler_provider_checks
):
mock_provider_choices.values = ["aws"]
mock_get_prowler_provider_checks.return_value = ["check1", "check2", "check3"]
prowler_compliance = {
@@ -163,10 +163,8 @@ class TestCompliance:
is None
)
@patch("api.models.Provider.ProviderChoices")
def test_generate_compliance_overview_template(self, mock_provider_choices):
mock_provider_choices.values = ["aws"]
@patch("api.compliance.SDKProvider.get_available_providers", return_value=["aws"])
def test_generate_compliance_overview_template(self, _mock_get_available_providers):
requirement1 = MagicMock(
Id="requirement1",
Name="Requirement 1",
@@ -310,3 +308,48 @@ class TestGetComplianceFrameworks:
f"get_compliance_frameworks({provider_type!r}) returned names not "
f"loadable by Compliance.get_bulk: {sorted(missing)}"
)
class TestDynamicProviderDiscovery:
"""Compliance discovery is sourced from the SDK's available providers, so an
externally-registered provider absent from the legacy static enum is
discovered without any change to the API."""
EXTERNAL = "acme-external"
@patch(
"api.compliance.SDKProvider.get_available_providers",
return_value=["aws", EXTERNAL],
)
def test_lazy_template_includes_external_provider(self, _mock_available):
template = LazyComplianceTemplate()
assert self.EXTERNAL in template
assert set(template) == {"aws", self.EXTERNAL}
@patch(
"api.compliance.SDKProvider.get_available_providers",
return_value=["aws", EXTERNAL],
)
def test_lazy_checks_mapping_includes_external_provider(self, _mock_available):
checks_mapping = LazyChecksMapping()
assert self.EXTERNAL in checks_mapping
assert set(checks_mapping) == {"aws", self.EXTERNAL}
@patch(
"api.compliance.SDKProvider.get_available_providers",
return_value=[EXTERNAL],
)
def test_overview_template_iterates_dynamic_providers(self, _mock_available):
template = generate_compliance_overview_template({self.EXTERNAL: {}})
assert self.EXTERNAL in template
@patch("api.compliance.get_prowler_provider_checks", return_value=["check1"])
@patch(
"api.compliance.SDKProvider.get_available_providers",
return_value=[EXTERNAL],
)
def test_load_checks_iterates_dynamic_providers(
self, _mock_available, _mock_checks
):
checks = load_prowler_checks({self.EXTERNAL: {}})
assert self.EXTERNAL in checks
+3 -2
View File
@@ -324,6 +324,7 @@ from prowler.providers.aws.exceptions.exceptions import (
from prowler.providers.aws.lib.cloudtrail_timeline.cloudtrail_timeline import (
CloudTrailTimeline,
)
from prowler.providers.common.provider import Provider as SDKProvider
logger = logging.getLogger(BackendLogger.API)
@@ -4993,7 +4994,7 @@ class ComplianceOverviewViewSet(BaseRLSViewSet, TaskManagementMixin):
# If we couldn't determine from database, try each provider type
if not provider_type:
for pt in Provider.ProviderChoices.values:
for pt in SDKProvider.get_available_providers():
if compliance_id in get_compliance_frameworks(pt):
provider_type = pt
break
@@ -5333,7 +5334,7 @@ class OverviewViewSet(BaseRLSViewSet):
"""Extract and validate provider filters from query params."""
params = self.request.query_params
filters = {}
valid_provider_types = {c[0] for c in Provider.ProviderChoices.choices}
valid_provider_types = set(SDKProvider.get_available_providers())
provider_id = params.get("filter[provider_id]")
if provider_id: