mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-09 21:14:22 +00:00
feat(sdk): discover external universal compliance frameworks via entry points (#11490)
This commit is contained in:
StylusFrost
@@ -14,6 +14,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
|
||||
- Support for external/custom providers, checks, and compliance frameworks without modifying core code [(#10700)](https://github.com/prowler-cloud/prowler/pull/10700)
|
||||
- `elbv2_alb_drop_invalid_header_fields_enabled` check for AWS provider, verifying Application Load Balancers have `routing.http.drop_invalid_header_fields.enabled` set to `true` to mitigate HTTP desync attacks (AWS FSBP ELB.4) [(#11471)](https://github.com/prowler-cloud/prowler/pull/11471)
|
||||
- `user`, `systemlog` and `idp` service for Okta provider with `user_inactivity_automation_35d_enabled`, `systemlog_streaming_enabled` and `idp_smart_card_dod_approved_ca` checks [(#11496)](https://github.com/prowler-cloud/prowler/pull/11496)
|
||||
- External multi-provider compliance frameworks can be registered via the `prowler.compliance.universal` entry point group [(#11490)](https://github.com/prowler-cloud/prowler/pull/11490)
|
||||
- AWS AI Security Framework support in the CLI dashboard [(#11475)](https://github.com/prowler-cloud/prowler/pull/11475)
|
||||
- `entra_service_principal_privileged_role_no_owners` check for M365 provider, failing when a service principal with a permanent Tier 0 directory role has owners on the service principal or its parent app registration [(#11070)](https://github.com/prowler-cloud/prowler/issues/11070)
|
||||
|
||||
|
||||
@@ -144,8 +144,7 @@ def get_available_compliance_frameworks(provider=None):
|
||||
continue
|
||||
if name not in available_compliance_frameworks:
|
||||
available_compliance_frameworks.append(name)
|
||||
# External compliance via entry points.
|
||||
# Multi-provider support for external plug-ins is tracked in PROWLER-1444.
|
||||
# External per-provider compliance via entry points.
|
||||
ep_dirs = _get_ep_compliance_dirs()
|
||||
for prov, path in ep_dirs.items():
|
||||
if provider and prov != provider:
|
||||
@@ -156,6 +155,32 @@ def get_available_compliance_frameworks(provider=None):
|
||||
name = file.name.removesuffix(".json")
|
||||
if name not in available_compliance_frameworks:
|
||||
available_compliance_frameworks.append(name)
|
||||
# External multi-provider frameworks via the dedicated universal group;
|
||||
# filtered by supports_provider when a provider is given.
|
||||
for ep in importlib.metadata.entry_points(group="prowler.compliance.universal"):
|
||||
try:
|
||||
module = ep.load()
|
||||
path = (
|
||||
module.__path__[0]
|
||||
if hasattr(module, "__path__")
|
||||
else os.path.dirname(module.__file__)
|
||||
)
|
||||
except Exception as error:
|
||||
logger.warning(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
continue
|
||||
if not os.path.isdir(path):
|
||||
continue
|
||||
for file in os.scandir(path):
|
||||
if file.is_file() and file.name.endswith(".json"):
|
||||
name = file.name.removesuffix(".json")
|
||||
if provider:
|
||||
framework = load_compliance_framework_universal(file.path)
|
||||
if framework is None or not framework.supports_provider(provider):
|
||||
continue
|
||||
if name not in available_compliance_frameworks:
|
||||
available_compliance_frameworks.append(name)
|
||||
return available_compliance_frameworks
|
||||
|
||||
|
||||
|
||||
@@ -478,9 +478,15 @@ class Compliance(BaseModel):
|
||||
compliance_framework_name
|
||||
not in bulk_compliance_frameworks
|
||||
):
|
||||
bulk_compliance_frameworks[
|
||||
compliance_framework_name
|
||||
] = load_compliance_framework(file_path)
|
||||
# External JSON: tolerate non-legacy
|
||||
# schemas (skip + warn) instead of aborting.
|
||||
framework = load_compliance_framework(
|
||||
file_path, fatal=False
|
||||
)
|
||||
if framework is not None:
|
||||
bulk_compliance_frameworks[
|
||||
compliance_framework_name
|
||||
] = framework
|
||||
except Exception as error:
|
||||
logger.warning(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
@@ -494,18 +500,26 @@ class Compliance(BaseModel):
|
||||
|
||||
# Testing Pending
|
||||
def load_compliance_framework(
|
||||
compliance_specification_file: str,
|
||||
) -> Compliance:
|
||||
"""load_compliance_framework loads and parse a Compliance Framework Specification"""
|
||||
compliance_specification_file: str, fatal: bool = True
|
||||
) -> Optional[Compliance]:
|
||||
"""load_compliance_framework loads and parse a Compliance Framework Specification.
|
||||
|
||||
With ``fatal=True`` (built-in JSONs) an invalid file aborts the run; with
|
||||
``fatal=False`` (external JSONs) it is skipped with a warning and ``None``
|
||||
is returned.
|
||||
"""
|
||||
try:
|
||||
compliance_framework = Compliance.parse_file(compliance_specification_file)
|
||||
return Compliance.parse_file(compliance_specification_file)
|
||||
except ValidationError as error:
|
||||
logger.critical(
|
||||
f"Compliance Framework Specification from {compliance_specification_file} is not valid: {error}"
|
||||
if fatal:
|
||||
logger.critical(
|
||||
f"Compliance Framework Specification from {compliance_specification_file} is not valid: {error}"
|
||||
)
|
||||
sys.exit(1)
|
||||
logger.warning(
|
||||
f"Skipping invalid compliance framework {compliance_specification_file}: {error}"
|
||||
)
|
||||
sys.exit(1)
|
||||
else:
|
||||
return compliance_framework
|
||||
return None
|
||||
|
||||
|
||||
# ─── Universal Compliance Schema Models (Phase 1-3) ─────────────────────────
|
||||
@@ -982,6 +996,25 @@ def get_bulk_compliance_frameworks_universal(provider: str) -> dict:
|
||||
if compliance_root and os.path.isdir(compliance_root):
|
||||
_load_jsons_from_dir(compliance_root, provider, bulk)
|
||||
|
||||
# External multi-provider frameworks via the dedicated universal entry
|
||||
# point group, kept separate from the per-provider `prowler.compliance`
|
||||
# group so the legacy loader never parses a universal JSON. Built-ins
|
||||
# (already in bulk) win on a name collision.
|
||||
for ep in importlib.metadata.entry_points(group="prowler.compliance.universal"):
|
||||
try:
|
||||
module = ep.load()
|
||||
ep_dir = (
|
||||
module.__path__[0]
|
||||
if hasattr(module, "__path__")
|
||||
else os.path.dirname(module.__file__)
|
||||
)
|
||||
if os.path.isdir(ep_dir):
|
||||
_load_jsons_from_dir(ep_dir, provider, bulk)
|
||||
except Exception as error:
|
||||
logger.warning(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}] -- {e}")
|
||||
return bulk
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic.v1 import ValidationError
|
||||
@@ -23,6 +25,7 @@ from prowler.lib.check.compliance_models import (
|
||||
TableLabels,
|
||||
UniversalComplianceRequirement,
|
||||
adapt_legacy_to_universal,
|
||||
get_bulk_compliance_frameworks_universal,
|
||||
load_compliance_framework_universal,
|
||||
)
|
||||
from tests.lib.outputs.compliance.fixtures import (
|
||||
@@ -1116,3 +1119,121 @@ class TestAttributesMetadataValidation:
|
||||
],
|
||||
attributes_metadata=self._metadata(enum=["high", "low"]),
|
||||
)
|
||||
|
||||
|
||||
class TestGetBulkUniversalEntryPoints:
|
||||
"""Entry-point discovery for universal (multi-provider) compliance frameworks."""
|
||||
|
||||
@staticmethod
|
||||
def _write_universal_json(directory, filename, framework, display_name):
|
||||
data = {
|
||||
"framework": framework,
|
||||
"name": display_name,
|
||||
"version": "1.0",
|
||||
"description": "External multi-provider framework",
|
||||
"requirements": [
|
||||
{
|
||||
"id": "1",
|
||||
"name": "Requirement 1",
|
||||
"description": "desc",
|
||||
"checks": {"fakeexternal": ["check_a"]},
|
||||
}
|
||||
],
|
||||
}
|
||||
with open(os.path.join(directory, filename), "w") as f:
|
||||
json.dump(data, f)
|
||||
|
||||
@staticmethod
|
||||
def _entry_point(path):
|
||||
module = MagicMock()
|
||||
module.__path__ = [path]
|
||||
ep = MagicMock()
|
||||
ep.name = "fakeexternal"
|
||||
ep.group = "prowler.compliance.universal"
|
||||
ep.load.return_value = module
|
||||
return ep
|
||||
|
||||
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
||||
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
||||
def test_includes_external_universal_framework(self, mock_list_modules, mock_ep):
|
||||
mock_list_modules.return_value = []
|
||||
with tempfile.TemporaryDirectory() as ep_dir:
|
||||
self._write_universal_json(
|
||||
ep_dir, "customuniversal_1.0.json", "CustomUniversal", "Custom"
|
||||
)
|
||||
mock_ep.return_value = [self._entry_point(ep_dir)]
|
||||
|
||||
bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
|
||||
|
||||
mock_ep.assert_called_with(group="prowler.compliance.universal")
|
||||
assert "customuniversal_1.0" in bulk
|
||||
assert bulk["customuniversal_1.0"].framework == "CustomUniversal"
|
||||
|
||||
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
||||
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
||||
def test_builtin_wins_over_external_on_name_collision(
|
||||
self, mock_list_modules, mock_ep
|
||||
):
|
||||
with (
|
||||
tempfile.TemporaryDirectory() as root,
|
||||
tempfile.TemporaryDirectory() as ep_dir,
|
||||
):
|
||||
builtin_sub = os.path.join(root, "builtinprov")
|
||||
os.makedirs(builtin_sub)
|
||||
self._write_universal_json(
|
||||
builtin_sub, "shared_1.0.json", "SharedFramework", "Built-in"
|
||||
)
|
||||
builtin_module = MagicMock()
|
||||
builtin_module.module_finder.path = root
|
||||
builtin_module.name = "prowler.compliance.builtinprov"
|
||||
mock_list_modules.return_value = [builtin_module]
|
||||
|
||||
self._write_universal_json(
|
||||
ep_dir, "shared_1.0.json", "SharedFramework", "External"
|
||||
)
|
||||
mock_ep.return_value = [self._entry_point(ep_dir)]
|
||||
|
||||
bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
|
||||
|
||||
assert "shared_1.0" in bulk
|
||||
assert bulk["shared_1.0"].name == "Built-in"
|
||||
|
||||
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
||||
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
||||
def test_loads_all_frameworks_in_a_single_entry_point_path(
|
||||
self, mock_list_modules, mock_ep
|
||||
):
|
||||
"""All JSONs in one entry-point directory are added, not collapsed to one."""
|
||||
mock_list_modules.return_value = []
|
||||
with tempfile.TemporaryDirectory() as ep_dir:
|
||||
self._write_universal_json(ep_dir, "fw_a_1.0.json", "FwA", "Framework A")
|
||||
self._write_universal_json(ep_dir, "fw_b_1.0.json", "FwB", "Framework B")
|
||||
mock_ep.return_value = [self._entry_point(ep_dir)]
|
||||
|
||||
bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
|
||||
|
||||
assert "fw_a_1.0" in bulk
|
||||
assert "fw_b_1.0" in bulk
|
||||
|
||||
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
||||
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
||||
def test_merges_frameworks_from_multiple_packages_same_provider(
|
||||
self, mock_list_modules, mock_ep
|
||||
):
|
||||
"""Two packages under the same provider name are both discovered."""
|
||||
mock_list_modules.return_value = []
|
||||
with (
|
||||
tempfile.TemporaryDirectory() as dir_a,
|
||||
tempfile.TemporaryDirectory() as dir_b,
|
||||
):
|
||||
self._write_universal_json(dir_a, "pkg_a_1.0.json", "PkgA", "Package A")
|
||||
self._write_universal_json(dir_b, "pkg_b_1.0.json", "PkgB", "Package B")
|
||||
mock_ep.return_value = [
|
||||
self._entry_point(dir_a),
|
||||
self._entry_point(dir_b),
|
||||
]
|
||||
|
||||
bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
|
||||
|
||||
assert "pkg_a_1.0" in bulk
|
||||
assert "pkg_b_1.0" in bulk
|
||||
|
||||
@@ -1218,6 +1218,48 @@ class TestCompliance:
|
||||
|
||||
assert "custom_1.0_ext" in frameworks
|
||||
|
||||
@patch("prowler.config.config.importlib.metadata.entry_points")
|
||||
def test_get_available_compliance_includes_external_universal(self, mock_ep):
|
||||
"""External universal frameworks under prowler.compliance.universal are
|
||||
listed, for a provider and for the provider=None case that feeds
|
||||
--compliance choices."""
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
from prowler.config.config import get_available_compliance_frameworks
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
framework = {
|
||||
"framework": "CustomUniversal",
|
||||
"name": "Custom Universal",
|
||||
"version": "1.0",
|
||||
"description": "Multi-provider",
|
||||
"requirements": [
|
||||
{
|
||||
"id": "1",
|
||||
"name": "r",
|
||||
"description": "d",
|
||||
"checks": {"aws": ["c"]},
|
||||
}
|
||||
],
|
||||
}
|
||||
with open(os.path.join(tmpdir, "customuniversal_1.0.json"), "w") as f:
|
||||
json.dump(framework, f)
|
||||
|
||||
module = MagicMock()
|
||||
module.__path__ = [tmpdir]
|
||||
ep = _make_entry_point(
|
||||
"anyname", "pkg.compliance", "prowler.compliance.universal"
|
||||
)
|
||||
ep.load.return_value = module
|
||||
mock_ep.side_effect = lambda group: (
|
||||
[ep] if group == "prowler.compliance.universal" else []
|
||||
)
|
||||
|
||||
assert "customuniversal_1.0" in get_available_compliance_frameworks("aws")
|
||||
assert "customuniversal_1.0" in get_available_compliance_frameworks(None)
|
||||
|
||||
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
||||
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
||||
def test_compliance_get_bulk_loads_external(self, mock_list_modules, mock_ep):
|
||||
@@ -1257,6 +1299,49 @@ class TestCompliance:
|
||||
assert "custom_1.0_fakeexternal" in bulk
|
||||
assert bulk["custom_1.0_fakeexternal"].Framework == "Custom"
|
||||
|
||||
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
||||
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
||||
def test_compliance_get_bulk_skips_non_legacy_external_json(
|
||||
self, mock_list_modules, mock_ep
|
||||
):
|
||||
"""A universal-schema JSON registered under prowler.compliance is skipped,
|
||||
not aborting the run via sys.exit."""
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
from prowler.lib.check.compliance_models import Compliance
|
||||
|
||||
mock_list_modules.return_value = []
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
json_data = {
|
||||
"framework": "Universal",
|
||||
"name": "Universal Framework",
|
||||
"version": "1.0",
|
||||
"description": "Multi-provider",
|
||||
"requirements": [
|
||||
{
|
||||
"id": "1",
|
||||
"name": "r",
|
||||
"description": "d",
|
||||
"checks": {"aws": ["c"]},
|
||||
}
|
||||
],
|
||||
}
|
||||
with open(os.path.join(tmpdir, "universal_1.0.json"), "w") as f:
|
||||
json.dump(json_data, f)
|
||||
|
||||
mock_module = MagicMock()
|
||||
mock_module.__path__ = [tmpdir]
|
||||
ep = _make_entry_point("aws", "pkg.compliance", "prowler.compliance")
|
||||
ep.load.return_value = mock_module
|
||||
mock_ep.return_value = [ep]
|
||||
|
||||
bulk = Compliance.get_bulk("aws")
|
||||
|
||||
assert "universal_1.0" not in bulk
|
||||
|
||||
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
||||
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
||||
def test_compliance_get_bulk_file_fallback(self, mock_list_modules, mock_ep):
|
||||
|
||||
Reference in New Issue
Block a user