fix(scan): add compliance info inside finding (#7247)

Co-authored-by: Pedro Martín <pedromarting3@gmail.com>
This commit is contained in:
Prowler Bot
2025-03-19 08:52:16 +01:00
committed by GitHub
parent 562921cd5e
commit ff4d16deb5
2 changed files with 49 additions and 28 deletions
+33 -12
View File
@@ -1,4 +1,5 @@
import datetime
from types import SimpleNamespace
from typing import Generator
from prowler.lib.check.check import (
@@ -22,7 +23,7 @@ from prowler.lib.scan.exceptions.exceptions import (
ScanInvalidSeverityError,
ScanInvalidStatusError,
)
from prowler.providers.common.models import Audit_Metadata
from prowler.providers.common.models import Audit_Metadata, ProviderOutputOptions
from prowler.providers.common.provider import Provider
@@ -38,6 +39,8 @@ class Scan:
_progress: float = 0.0
_duration: int = 0
_status: list[str] = None
_bulk_checks_metadata: dict[str, CheckMetadata]
_bulk_compliance_frameworks: dict
def __init__(
self,
@@ -88,18 +91,18 @@ class Scan:
raise ScanInvalidStatusError(f"Invalid status provided: {s}.")
# Load bulk compliance frameworks
bulk_compliance_frameworks = Compliance.get_bulk(provider.type)
self._bulk_compliance_frameworks = Compliance.get_bulk(provider.type)
# Get bulk checks metadata for the provider
bulk_checks_metadata = CheckMetadata.get_bulk(provider.type)
self._bulk_checks_metadata = CheckMetadata.get_bulk(provider.type)
# Complete checks metadata with the compliance framework specification
bulk_checks_metadata = update_checks_metadata_with_compliance(
bulk_compliance_frameworks, bulk_checks_metadata
self._bulk_checks_metadata = update_checks_metadata_with_compliance(
self._bulk_compliance_frameworks, self._bulk_checks_metadata
)
# Create a list of valid categories
valid_categories = set()
for check, metadata in bulk_checks_metadata.items():
for check, metadata in self._bulk_checks_metadata.items():
for category in metadata.Categories:
if category not in valid_categories:
valid_categories.add(category)
@@ -107,7 +110,7 @@ class Scan:
# Validate checks
if checks:
for check in checks:
if check not in bulk_checks_metadata.keys():
if check not in self._bulk_checks_metadata.keys():
raise ScanInvalidCheckError(f"Invalid check provided: {check}.")
# Validate services
@@ -121,7 +124,7 @@ class Scan:
# Validate compliances
if compliances:
for compliance in compliances:
if compliance not in bulk_compliance_frameworks.keys():
if compliance not in self._bulk_compliance_frameworks.keys():
raise ScanInvalidComplianceFrameworkError(
f"Invalid compliance provided: {compliance}."
)
@@ -147,8 +150,8 @@ class Scan:
# Load checks to execute
self._checks_to_execute = sorted(
load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metadata,
bulk_compliance_frameworks=bulk_compliance_frameworks,
bulk_checks_metadata=self._bulk_checks_metadata,
bulk_compliance_frameworks=self._bulk_compliance_frameworks,
check_list=checks,
service_list=services,
compliance_frameworks=compliances,
@@ -215,9 +218,17 @@ class Scan:
def duration(self) -> int:
return self._duration
@property
def bulk_checks_metadata(self) -> dict[str, CheckMetadata]:
return self._bulk_checks_metadata
@property
def bulk_compliance_frameworks(self) -> dict[str, CheckMetadata]:
return self._bulk_compliance_frameworks
def scan(
self,
custom_checks_metadata: dict = {},
custom_checks_metadata: dict = None,
) -> Generator[tuple[float, list[Finding]], None, None]:
"""
Executes the scan by iterating over the checks to execute and executing each check.
@@ -234,6 +245,14 @@ class Scan:
Exception: If any other error occurs during the execution of a check.
"""
try:
# Using SimpleNamespace to create a mocked object
arguments = SimpleNamespace()
output_options = ProviderOutputOptions(
arguments=arguments,
bulk_checks_metadata=self.bulk_checks_metadata,
)
checks_to_execute = self.checks_to_execute
# Initialize the Audit Metadata
# TODO: this should be done in the provider class
@@ -301,7 +320,9 @@ class Scan:
try:
findings.append(
Finding.generate_output(
self._provider, finding, output_options=None
self.provider,
finding,
output_options=output_options,
)
)
except Exception:
+16 -16
View File
@@ -28,34 +28,34 @@ class ProviderOutputOptions:
unix_timestamp: bool
def __init__(self, arguments, bulk_checks_metadata):
self.status = arguments.status
self.output_modes = arguments.output_formats
self.output_directory = arguments.output_directory
self.verbose = arguments.verbose
self.status = getattr(arguments, "status", None)
self.output_modes = getattr(arguments, "output_formats", None)
self.output_directory = getattr(arguments, "output_directory", None)
self.verbose = getattr(arguments, "verbose", None)
self.bulk_checks_metadata = bulk_checks_metadata
self.only_logs = arguments.only_logs
self.unix_timestamp = arguments.unix_timestamp
self.shodan_api_key = arguments.shodan
self.only_logs = getattr(arguments, "only_logs", None)
self.unix_timestamp = getattr(arguments, "unix_timestamp", None)
self.shodan_api_key = getattr(arguments, "shodan", None)
self.fixer = getattr(arguments, "fixer", None)
# Shodan API Key
if arguments.shodan:
if self.shodan_api_key:
# TODO: revisit this logic
provider = Provider.get_global_provider()
updated_audit_config = Provider.update_provider_config(
provider.audit_config, "shodan_api_key", arguments.shodan
provider.audit_config, "shodan_api_key", self.shodan_api_key
)
if updated_audit_config:
provider._audit_config = updated_audit_config
# Check output directory, if it is not created -> create it
if arguments.output_directory and not self.fixer:
if not isdir(arguments.output_directory):
if arguments.output_formats:
makedirs(arguments.output_directory, exist_ok=True)
if not isdir(arguments.output_directory + "/compliance"):
if arguments.output_formats:
makedirs(arguments.output_directory + "/compliance", exist_ok=True)
if self.output_directory and not self.fixer:
if not isdir(self.output_directory):
if self.output_modes:
makedirs(self.output_directory, exist_ok=True)
if not isdir(self.output_directory + "/compliance"):
if self.output_modes:
makedirs(self.output_directory + "/compliance", exist_ok=True)
@dataclass