From 00b5b2a1220bc560bf6f63dce29b9eefc7e24df2 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Thu, 27 Aug 2026 09:35:52 +0200 Subject: [PATCH 1/4] test(prowler): define CHK005 OCSF mapping behavior (#422) --- prowler/prowler/models/__init__.py | 3 +- prowler/prowler/models/findings.py | 65 +++++ .../behaviour/chk005_ocsf_mapping/__init__.py | 1 + .../chk005_ocsf_mapping.feature | 79 ++++++ .../behaviour/chk005_ocsf_mapping/conftest.py | 46 ++++ .../test_chk005_ocsf_mapping_bdd.py | 226 ++++++++++++++++++ .../unit/chk005_ocsf_mapping/__init__.py | 1 + .../unit/chk005_ocsf_mapping/conftest.py | 29 +++ .../test_decode_and_paths.py | 113 +++++++++ 9 files changed, 562 insertions(+), 1 deletion(-) create mode 100644 prowler/prowler/models/findings.py create mode 100644 prowler/tests/behaviour/chk005_ocsf_mapping/__init__.py create mode 100644 prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature create mode 100644 prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py create mode 100644 prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py create mode 100644 prowler/tests/unit/chk005_ocsf_mapping/__init__.py create mode 100644 prowler/tests/unit/chk005_ocsf_mapping/conftest.py create mode 100644 prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py diff --git a/prowler/prowler/models/__init__.py b/prowler/prowler/models/__init__.py index 2292b3f3..4c0c0ad0 100644 --- a/prowler/prowler/models/__init__.py +++ b/prowler/prowler/models/__init__.py @@ -1,5 +1,6 @@ """Prowler injector models.""" from prowler.models.configs import ConfigLoader +from prowler.models.findings import OpenAevFinding -__all__ = ["ConfigLoader"] +__all__ = ["ConfigLoader", "OpenAevFinding"] diff --git a/prowler/prowler/models/findings.py b/prowler/prowler/models/findings.py new file mode 100644 index 00000000..d5e2806c --- /dev/null +++ b/prowler/prowler/models/findings.py @@ -0,0 +1,65 @@ +"""CHK.005 Prowler OCSF to OpenAEV finding boundary.""" + +from dataclasses import dataclass +from typing import Any + +from pydantic import BaseModel, ConfigDict + +from prowler._core.cli_engine import CommandResult + + +@dataclass(frozen=True) +class OcsfDecodeError(ValueError): + """Safe structured failure while decoding raw OCSF output.""" + + code: str + message: str + record_index: int | None = None + + +@dataclass(frozen=True) +class OcsfMappingError(ValueError): + """Safe structured failure while mapping one OCSF record.""" + + code: str + message: str + record_index: int | None = None + source_path: str | None = None + + +class OpenAevFinding(BaseModel): + """Project boundary model; pyoaev 2.3.5 has no public finding model.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + type: str + value: str + expectation_result: str + severity: str + severity_weight: int + asset_reference: str + asset_name: str + cloud_provider: str + region: str + cloud_account: str + compliance_tags: tuple[str, ...] + remediation: str + remediation_url: str | None + description: str + + +def decode_ocsf_output(payload: bytes | str) -> tuple[dict[str, Any], ...]: + """Decode raw Prowler JSON array or JSON Lines output.""" + raise NotImplementedError("CHK.005 RED: raw OCSF decoding is not implemented") + + +def map_ocsf_finding( + record: dict[str, Any], *, record_index: int = 0 +) -> OpenAevFinding: + """Map one decoded OCSF record to the local OpenAEV boundary model.""" + raise NotImplementedError("CHK.005 RED: OCSF mapping is not implemented") + + +def map_command_result(result: CommandResult) -> tuple[OpenAevFinding, ...]: + """Map one successful CHK.004 result without executing another command.""" + raise NotImplementedError("CHK.005 RED: command result mapping is not implemented") diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/__init__.py b/prowler/tests/behaviour/chk005_ocsf_mapping/__init__.py new file mode 100644 index 00000000..20f2a7d1 --- /dev/null +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/__init__.py @@ -0,0 +1 @@ +"""CHK.005 OCSF mapping behaviour tests.""" diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature b/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature new file mode 100644 index 00000000..c22834bd --- /dev/null +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature @@ -0,0 +1,79 @@ +Feature: Map raw Prowler OCSF output to OpenAEV findings + A successful Prowler assessment is translated without executing another command. + + Scenario: Map an OCSF JSON array in source order + Given successful raw Prowler output containing two OCSF findings + When the output is mapped to OpenAEV findings + Then two immutable findings are returned in source order + And every finding declares exactly the 14 OpenAEV fields + + Scenario: Map OCSF JSON Lines while ignoring blank lines + Given successful raw Prowler JSON Lines output separated by blank lines + When the output is mapped to OpenAEV findings + Then each object becomes one finding in source order + + Scenario Outline: Normalize finding status + Given an OCSF finding with status "" + When the finding is mapped + Then expectation_result is "" + + Examples: + | source_status | result_status | + | PASS | SUCCESS | + | passed | SUCCESS | + | FAIL | FAILED | + | failed | FAILED | + | MUTED | IGNORED | + | manual | IGNORED | + | SUPPRESSED | IGNORED | + | ERROR | MUTED | + | UNKNOWN | MUTED | + + Scenario Outline: Normalize finding severity + Given an OCSF finding with severity "" + When the finding is mapped + Then severity is "" + And severity_weight is + + Examples: + | source_severity | normalized_severity | weight | + | critical | CRITICAL | 4 | + | HIGH | HIGH | 3 | + | medium | MEDIUM | 2 | + | LOW | LOW | 1 | + | informational | INFO | 0 | + | unknown | INFO | 0 | + + Scenario: Preserve Prowler 5.36 compliance value order + Given an OCSF finding with ordered compliance values under unmapped.compliance + When the finding is mapped + Then compliance_tags contains every value in source order + + Scenario: Use safe fallbacks for optional values + Given an OCSF finding without severity or compliance and with no remediation reference + When the finding is mapped + Then severity is INFO with weight zero + And compliance_tags is empty + And remediation_url is absent + + # ---- Constraints identified ---- + + Scenario: Reject malformed raw output without disclosing it + Given raw output containing malformed or non-UTF-8 sensitive data + When the output is decoded + Then a structured safe decode error is returned without the raw data + + Scenario: Reject a missing required source path without a lookup exception + Given an OCSF finding missing a required mapped source path + When the finding is mapped + Then a structured mapping error identifies the record index and source path + + Scenario: Reject an empty required resource collection + Given an OCSF finding with no resources + When the finding is mapped + Then a structured mapping error identifies resources[0] + + Scenario: Reject whitespace-padded status rather than silently normalizing it + Given an OCSF finding with whitespace around a recognized status + When the finding is mapped + Then expectation_result is MUTED diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py b/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py new file mode 100644 index 00000000..31389e53 --- /dev/null +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py @@ -0,0 +1,46 @@ +"""Local fixtures for CHK.005 behaviour.""" + +from copy import deepcopy +from typing import Any + +import pytest + + +@pytest.fixture +def ocsf_record() -> dict[str, Any]: + """Return one representative Prowler 5.36 detection finding.""" + return { + "finding_info": { + "uid": "prowler.aws.iam.root_user_access_key", + "title": "Root user access keys should be removed", + "desc": "The root user has active access keys.", + }, + "status": "PASS", + "severity": "High", + "resources": [{"uid": "arn:aws:iam::123456789012:root", "name": "root"}], + "cloud": { + "provider": "aws", + "region": "eu-west-1", + "account": {"uid": "123456789012"}, + }, + "unmapped": { + "compliance": { + "CIS-1.5": ["1.1", "1.2"], + "ENS-RD2022": "op.acc.6", + } + }, + "remediation": { + "desc": "Delete the root user access keys.", + "references": ["https://example.test/remediation"], + }, + } + + +@pytest.fixture +def copy_record(ocsf_record: dict[str, Any]): + """Return a factory that isolates mutable source records.""" + + def factory() -> dict[str, Any]: + return deepcopy(ocsf_record) + + return factory diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py b/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py new file mode 100644 index 00000000..8fdb0538 --- /dev/null +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py @@ -0,0 +1,226 @@ +"""Raw-pytest bindings for the CHK.005 feature.""" + +import json +from dataclasses import FrozenInstanceError +from typing import Any, Callable + +import pytest +from pydantic import ValidationError + +from prowler._core.cli_engine import ( + CommandResult, + ExecutionSpecification, + OutputSpecification, +) +from prowler.models.findings import ( + OcsfDecodeError, + OcsfMappingError, + OpenAevFinding, + decode_ocsf_output, + map_command_result, + map_ocsf_finding, +) + + +def _success(payload: bytes) -> CommandResult: + specification = ExecutionSpecification( + executable="/usr/local/bin/prowler", + arguments=("aws", "-M", "json-ocsf"), + environment=(), + working_directory=None, + input_bytes=b"", + output=OutputSpecification(parser="raw"), + timeout_seconds=3600, + maximum_accepted_output_bytes=100 * 1024 * 1024, + ) + return CommandResult( + specification=specification, + stdout=payload, + return_code=0, + parsed=payload, + ) + + +def test_maps_json_array_to_exact_immutable_findings_in_order( + copy_record: Callable[[], dict[str, Any]], +) -> None: + first = copy_record() + second = copy_record() + second["finding_info"]["uid"] = "second" + payload = json.dumps([first, second]).encode() + + findings = map_command_result(_success(payload)) + + assert tuple(finding.type for finding in findings) == ( + "prowler.aws.iam.root_user_access_key", + "second", + ) + assert tuple(OpenAevFinding.model_fields) == ( + "type", + "value", + "expectation_result", + "severity", + "severity_weight", + "asset_reference", + "asset_name", + "cloud_provider", + "region", + "cloud_account", + "compliance_tags", + "remediation", + "remediation_url", + "description", + ) + with pytest.raises((ValidationError, FrozenInstanceError)): + findings[0].severity = "LOW" # type: ignore[misc] + + +def test_maps_json_lines_and_ignores_blank_lines( + copy_record: Callable[[], dict[str, Any]], +) -> None: + first = copy_record() + second = copy_record() + second["finding_info"]["uid"] = "second" + payload = f"{json.dumps(first)}\n\n \n{json.dumps(second)}\n".encode() + + findings = map_command_result(_success(payload)) + + assert tuple(finding.type for finding in findings) == ( + "prowler.aws.iam.root_user_access_key", + "second", + ) + + +@pytest.mark.parametrize( + ("source", "expected"), + [ + ("PASS", "SUCCESS"), + ("passed", "SUCCESS"), + ("FAIL", "FAILED"), + ("failed", "FAILED"), + ("MUTED", "IGNORED"), + ("manual", "IGNORED"), + ("SUPPRESSED", "IGNORED"), + ("ERROR", "MUTED"), + ("UNKNOWN", "MUTED"), + (" PASS ", "MUTED"), + ], +) +def test_normalizes_status_without_trimming( + copy_record: Callable[[], dict[str, Any]], source: str, expected: str +) -> None: + record = copy_record() + record["status"] = source + + assert map_ocsf_finding(record, record_index=7).expectation_result == expected + + +@pytest.mark.parametrize( + ("source", "label", "weight"), + [ + ("critical", "CRITICAL", 4), + ("HIGH", "HIGH", 3), + ("medium", "MEDIUM", 2), + ("LOW", "LOW", 1), + ("informational", "INFO", 0), + ("unknown", "INFO", 0), + ], +) +def test_normalizes_severity_case_insensitively( + copy_record: Callable[[], dict[str, Any]], + source: str, + label: str, + weight: int, +) -> None: + record = copy_record() + record["severity"] = source + + finding = map_ocsf_finding(record) + + assert (finding.severity, finding.severity_weight) == (label, weight) + + +def test_maps_all_fields_and_preserves_compliance_values( + ocsf_record: dict[str, Any], +) -> None: + finding = map_ocsf_finding(ocsf_record) + + assert finding.model_dump() == { + "type": "prowler.aws.iam.root_user_access_key", + "value": "Root user access keys should be removed", + "expectation_result": "SUCCESS", + "severity": "HIGH", + "severity_weight": 3, + "asset_reference": "arn:aws:iam::123456789012:root", + "asset_name": "root", + "cloud_provider": "aws", + "region": "eu-west-1", + "cloud_account": "123456789012", + "compliance_tags": ("1.1", "1.2", "op.acc.6"), + "remediation": "Delete the root user access keys.", + "remediation_url": "https://example.test/remediation", + "description": "The root user has active access keys.", + } + + +def test_optional_values_have_safe_fallbacks( + copy_record: Callable[[], dict[str, Any]], +) -> None: + record = copy_record() + del record["severity"] + del record["unmapped"]["compliance"] + record["remediation"]["references"] = [] + + finding = map_ocsf_finding(record) + + assert (finding.severity, finding.severity_weight) == ("INFO", 0) + assert finding.compliance_tags == () + assert finding.remediation_url is None + + +@pytest.mark.parametrize("payload", [b"\xffsecret", b'[{"token":"secret"}']) +def test_decode_errors_do_not_echo_sensitive_payload(payload: bytes) -> None: + with pytest.raises(OcsfDecodeError) as caught: + decode_ocsf_output(payload) + + assert caught.value.code in {"invalid_utf8", "invalid_json"} + assert caught.value.message == "unable to decode Prowler OCSF output" + assert "secret" not in str(caught.value) + + +def test_non_object_record_has_safe_index() -> None: + with pytest.raises(OcsfDecodeError) as caught: + decode_ocsf_output(b'[{"ok": true}, "sensitive"]') + + assert caught.value.code == "non_object_record" + assert caught.value.record_index == 1 + assert "sensitive" not in str(caught.value) + + +def test_missing_required_path_has_structured_mapping_error( + copy_record: Callable[[], dict[str, Any]], +) -> None: + record = copy_record() + del record["cloud"]["account"]["uid"] + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=4) + + assert caught.value.code == "missing_source_path" + assert caught.value.record_index == 4 + assert caught.value.source_path == "cloud.account.uid" + assert "123456789012" not in str(caught.value) + + +def test_empty_resources_has_structured_mapping_error( + copy_record: Callable[[], dict[str, Any]], +) -> None: + record = copy_record() + record["resources"] = [] + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=2) + + assert caught.value.code == "missing_source_path" + assert caught.value.source_path == "resources[0]" + assert caught.value.record_index == 2 diff --git a/prowler/tests/unit/chk005_ocsf_mapping/__init__.py b/prowler/tests/unit/chk005_ocsf_mapping/__init__.py new file mode 100644 index 00000000..72b26e9e --- /dev/null +++ b/prowler/tests/unit/chk005_ocsf_mapping/__init__.py @@ -0,0 +1 @@ +"""CHK.005 mapping unit tests.""" diff --git a/prowler/tests/unit/chk005_ocsf_mapping/conftest.py b/prowler/tests/unit/chk005_ocsf_mapping/conftest.py new file mode 100644 index 00000000..9c73c7ab --- /dev/null +++ b/prowler/tests/unit/chk005_ocsf_mapping/conftest.py @@ -0,0 +1,29 @@ +"""Local fixtures for CHK.005 mapping units.""" + +from copy import deepcopy +from typing import Any + +import pytest + + +@pytest.fixture +def copy_record(): + """Return isolated representative Prowler 5.36 records.""" + record: dict[str, Any] = { + "finding_info": {"uid": "uid", "title": "title", "desc": "description"}, + "status": "PASS", + "severity": "high", + "resources": [{"uid": "resource", "name": "resource name"}], + "cloud": { + "provider": "aws", + "region": "eu-west-1", + "account": {"uid": "account"}, + }, + "unmapped": {"compliance": {}}, + "remediation": {"desc": "fix", "references": ["https://example.test"]}, + } + + def factory() -> dict[str, Any]: + return deepcopy(record) + + return factory diff --git a/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py b/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py new file mode 100644 index 00000000..5a37145a --- /dev/null +++ b/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py @@ -0,0 +1,113 @@ +"""Focused parser and source-path edge cases.""" + +import json +from typing import Any, Callable + +import pytest + +from prowler._core.cli_engine import CommandResult, ExecutionSpecification +from prowler.models.findings import ( + OcsfDecodeError, + OcsfMappingError, + decode_ocsf_output, + map_command_result, + map_ocsf_finding, +) + + +def test_empty_json_array_and_blank_json_lines_are_empty() -> None: + assert decode_ocsf_output(b"[]") == () + assert decode_ocsf_output(b" \n\n\t") == () + + +def test_single_json_line_object_is_accepted() -> None: + assert decode_ocsf_output('{"finding_info": {}}') == ({"finding_info": {}},) + + +@pytest.mark.parametrize("payload", [b"null", b"42", b'"text"', b"{}\n[]"]) +def test_top_level_or_jsonl_non_objects_are_rejected(payload: bytes) -> None: + with pytest.raises(OcsfDecodeError) as caught: + decode_ocsf_output(payload) + + assert caught.value.code in {"invalid_top_level", "non_object_record"} + + +def test_command_result_must_be_success() -> None: + result = CommandResult( + specification=ExecutionSpecification( + executable="prowler", + arguments=(), + environment=(), + working_directory=None, + input_bytes=b"", + output=object(), # type: ignore[arg-type] + timeout_seconds=1, + maximum_accepted_output_bytes=1, + ), + return_code=1, + error=RuntimeError("sensitive failure"), + ) + + with pytest.raises(OcsfMappingError) as caught: + map_command_result(result) + + assert caught.value.code == "command_not_successful" + assert "sensitive" not in str(caught.value) + + +@pytest.mark.parametrize( + ("compliance", "expected"), + [ + ({"A": "x", "B": ["y", "x"]}, ("x", "y", "x")), + (["a", "b"], ("a", "b")), + ("one", ("one",)), + ({"A": {"first": "x", "second": ["y"]}}, ("x", "y")), + ], +) +def test_supported_compliance_shapes_preserve_values_and_duplicates( + copy_record: Callable[[], dict[str, Any]], + compliance: object, + expected: tuple[str, ...], +) -> None: + record = copy_record() + record["unmapped"]["compliance"] = compliance + + assert map_ocsf_finding(record).compliance_tags == expected + + +def test_invalid_compliance_leaf_has_structured_error( + copy_record: Callable[[], dict[str, Any]], +) -> None: + record = copy_record() + record["unmapped"]["compliance"] = {"CIS": ["1.1", 42]} + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=3) + + assert caught.value.code == "invalid_source_value" + assert caught.value.record_index == 3 + assert caught.value.source_path == "unmapped.compliance.CIS[1]" + + +@pytest.mark.parametrize( + "path_mutation", + [ + lambda record: record.update(finding_info="wrong"), + lambda record: record.update(status=7), + lambda record: record["resources"].__setitem__(0, "wrong"), + lambda record: record["remediation"].update(references="wrong"), + ], +) +def test_wrong_source_types_are_structured_errors( + copy_record: Callable[[], dict[str, Any]], + path_mutation: Callable[[dict[str, Any]], None], +) -> None: + record = copy_record() + path_mutation(record) + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record) + + assert caught.value.code == "invalid_source_value" + assert caught.value.source_path + assert json.dumps(record) not in str(caught.value) From 24c7aa162504f77b75141110951b5c462e3f4fd5 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Thu, 27 Aug 2026 09:57:31 +0200 Subject: [PATCH 2/4] feat(prowler): map raw OCSF findings (#422) --- prowler/prowler/models/findings.py | 259 +++++++++++++++++- .../test_chk005_ocsf_mapping_bdd.py | 10 + .../test_decode_and_paths.py | 7 + 3 files changed, 273 insertions(+), 3 deletions(-) diff --git a/prowler/prowler/models/findings.py b/prowler/prowler/models/findings.py index d5e2806c..54594399 100644 --- a/prowler/prowler/models/findings.py +++ b/prowler/prowler/models/findings.py @@ -1,5 +1,7 @@ """CHK.005 Prowler OCSF to OpenAEV finding boundary.""" +import json +from collections.abc import Mapping, Sequence from dataclasses import dataclass from typing import Any @@ -16,6 +18,15 @@ class OcsfDecodeError(ValueError): message: str record_index: int | None = None + def __str__(self) -> str: + """Render only safe structural context.""" + suffix = ( + f" (record_index={self.record_index})" + if self.record_index is not None + else "" + ) + return f"{self.code}: {self.message}{suffix}" + @dataclass(frozen=True) class OcsfMappingError(ValueError): @@ -26,6 +37,16 @@ class OcsfMappingError(ValueError): record_index: int | None = None source_path: str | None = None + def __str__(self) -> str: + """Render only safe structural context.""" + context = [] + if self.record_index is not None: + context.append(f"record_index={self.record_index}") + if self.source_path is not None: + context.append(f"source_path={self.source_path}") + suffix = f" ({', '.join(context)})" if context else "" + return f"{self.code}: {self.message}{suffix}" + class OpenAevFinding(BaseModel): """Project boundary model; pyoaev 2.3.5 has no public finding model.""" @@ -48,18 +69,250 @@ class OpenAevFinding(BaseModel): description: str +_STATUS_MAP = { + "PASS": "SUCCESS", + "PASSED": "SUCCESS", + "FAIL": "FAILED", + "FAILED": "FAILED", + "MUTED": "IGNORED", + "MANUAL": "IGNORED", + "SUPPRESSED": "IGNORED", +} + +_SEVERITY_MAP = { + "CRITICAL": ("CRITICAL", 4), + "HIGH": ("HIGH", 3), + "MEDIUM": ("MEDIUM", 2), + "LOW": ("LOW", 1), + "INFORMATIONAL": ("INFO", 0), +} + +_DECODE_MESSAGE = "unable to decode Prowler OCSF output" +_MAPPING_MESSAGE = "unable to map Prowler OCSF record" + + def decode_ocsf_output(payload: bytes | str) -> tuple[dict[str, Any], ...]: """Decode raw Prowler JSON array or JSON Lines output.""" - raise NotImplementedError("CHK.005 RED: raw OCSF decoding is not implemented") + if isinstance(payload, bytes): + try: + text = payload.decode("utf-8") + except UnicodeDecodeError as error: + raise OcsfDecodeError("invalid_utf8", _DECODE_MESSAGE) from error + elif isinstance(payload, str): + text = payload + else: + raise OcsfDecodeError("invalid_payload_type", _DECODE_MESSAGE) + + if not text.strip(): + return () + + try: + decoded = json.loads(text) + except json.JSONDecodeError: + return _decode_json_lines(text) + + if isinstance(decoded, dict): + return (decoded,) + if not isinstance(decoded, list): + raise OcsfDecodeError("invalid_top_level", _DECODE_MESSAGE) + return _validate_records(decoded) def map_ocsf_finding( record: dict[str, Any], *, record_index: int = 0 ) -> OpenAevFinding: """Map one decoded OCSF record to the local OpenAEV boundary model.""" - raise NotImplementedError("CHK.005 RED: OCSF mapping is not implemented") + finding_info = _required_mapping(record, "finding_info", record_index) + resources = _required_sequence(record, "resources", record_index) + if not resources: + raise _mapping_error("missing_source_path", record_index, "resources[0]") + resource = _mapping_value(resources[0], "resources[0]", record_index) + cloud = _required_mapping(record, "cloud", record_index) + account = _required_mapping(cloud, "cloud.account", record_index, key="account") + remediation = _required_mapping(record, "remediation", record_index) + + status = _required_string(record, "status", record_index) + severity_value = record.get("severity") + if severity_value is None: + severity = ("INFO", 0) + elif isinstance(severity_value, str): + severity = _SEVERITY_MAP.get(severity_value.upper(), ("INFO", 0)) + else: + raise _mapping_error("invalid_source_value", record_index, "severity") + + references_value = remediation.get("references", ()) + if not isinstance(references_value, Sequence) or isinstance( + references_value, (str, bytes) + ): + raise _mapping_error( + "invalid_source_value", record_index, "remediation.references" + ) + remediation_url = None + if references_value: + remediation_url = _string_value( + references_value[0], "remediation.references[0]", record_index + ) + + return OpenAevFinding( + type=_required_string( + finding_info, "finding_info.uid", record_index, key="uid" + ), + value=_required_string( + finding_info, "finding_info.title", record_index, key="title" + ), + expectation_result=_STATUS_MAP.get(status.upper(), "MUTED"), + severity=severity[0], + severity_weight=severity[1], + asset_reference=_required_string( + resource, "resources[0].uid", record_index, key="uid" + ), + asset_name=_required_string( + resource, "resources[0].name", record_index, key="name" + ), + cloud_provider=_required_string( + cloud, "cloud.provider", record_index, key="provider" + ), + region=_required_string(cloud, "cloud.region", record_index, key="region"), + cloud_account=_required_string( + account, "cloud.account.uid", record_index, key="uid" + ), + compliance_tags=_compliance_tags(record, record_index), + remediation=_required_string( + remediation, "remediation.desc", record_index, key="desc" + ), + remediation_url=remediation_url, + description=_required_string( + finding_info, "finding_info.desc", record_index, key="desc" + ), + ) def map_command_result(result: CommandResult) -> tuple[OpenAevFinding, ...]: """Map one successful CHK.004 result without executing another command.""" - raise NotImplementedError("CHK.005 RED: command result mapping is not implemented") + if result.error is not None or result.return_code != 0: + raise OcsfMappingError( + "command_not_successful", + "Prowler command result is not successful", + ) + payload = ( + result.parsed if isinstance(result.parsed, (bytes, str)) else result.stdout + ) + records = decode_ocsf_output(payload) + return tuple( + map_ocsf_finding(record, record_index=index) + for index, record in enumerate(records) + ) + + +def _decode_json_lines(text: str) -> tuple[dict[str, Any], ...]: + records: list[dict[str, Any]] = [] + for record_index, line in enumerate( + line for line in text.splitlines() if line.strip() + ): + try: + decoded = json.loads(line) + except json.JSONDecodeError as error: + raise OcsfDecodeError( + "invalid_json", _DECODE_MESSAGE, record_index + ) from error + if not isinstance(decoded, dict): + raise OcsfDecodeError("non_object_record", _DECODE_MESSAGE, record_index) + records.append(decoded) + return tuple(records) + + +def _validate_records(records: list[Any]) -> tuple[dict[str, Any], ...]: + validated = [] + for record_index, record in enumerate(records): + if not isinstance(record, dict): + raise OcsfDecodeError("non_object_record", _DECODE_MESSAGE, record_index) + validated.append(record) + return tuple(validated) + + +def _mapping_error(code: str, record_index: int, source_path: str) -> OcsfMappingError: + return OcsfMappingError(code, _MAPPING_MESSAGE, record_index, source_path) + + +def _mapping_value( + value: object, source_path: str, record_index: int +) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise _mapping_error("invalid_source_value", record_index, source_path) + return value + + +def _required_mapping( + parent: Mapping[str, Any], + source_path: str, + record_index: int, + *, + key: str | None = None, +) -> Mapping[str, Any]: + lookup_key = key or source_path + if lookup_key not in parent: + raise _mapping_error("missing_source_path", record_index, source_path) + return _mapping_value(parent[lookup_key], source_path, record_index) + + +def _required_sequence( + parent: Mapping[str, Any], source_path: str, record_index: int +) -> Sequence[Any]: + if source_path not in parent: + raise _mapping_error("missing_source_path", record_index, source_path) + value = parent[source_path] + if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): + raise _mapping_error("invalid_source_value", record_index, source_path) + return value + + +def _string_value(value: object, source_path: str, record_index: int) -> str: + if not isinstance(value, str): + raise _mapping_error("invalid_source_value", record_index, source_path) + return value + + +def _required_string( + parent: Mapping[str, Any], + source_path: str, + record_index: int, + *, + key: str | None = None, +) -> str: + lookup_key = key or source_path + if lookup_key not in parent: + raise _mapping_error("missing_source_path", record_index, source_path) + return _string_value(parent[lookup_key], source_path, record_index) + + +def _compliance_tags(record: Mapping[str, Any], record_index: int) -> tuple[str, ...]: + unmapped = record.get("unmapped") + if unmapped is None: + return () + unmapped_mapping = _mapping_value(unmapped, "unmapped", record_index) + compliance = unmapped_mapping.get("compliance") + if compliance is None: + return () + return tuple(_flatten_compliance(compliance, "unmapped.compliance", record_index)) + + +def _flatten_compliance( + value: object, source_path: str, record_index: int +) -> list[str]: + if isinstance(value, str): + return [value] + if isinstance(value, Mapping): + flattened = [] + for key, nested in value.items(): + flattened.extend( + _flatten_compliance(nested, f"{source_path}.{key}", record_index) + ) + return flattened + if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): + flattened = [] + for index, nested in enumerate(value): + flattened.extend( + _flatten_compliance(nested, f"{source_path}[{index}]", record_index) + ) + return flattened + raise _mapping_error("invalid_source_value", record_index, source_path) diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py b/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py index 8fdb0538..3064ecd2 100644 --- a/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py @@ -44,6 +44,7 @@ def _success(payload: bytes) -> CommandResult: def test_maps_json_array_to_exact_immutable_findings_in_order( copy_record: Callable[[], dict[str, Any]], ) -> None: + """Map each array object to one frozen 14-field finding in order.""" first = copy_record() second = copy_record() second["finding_info"]["uid"] = "second" @@ -78,6 +79,7 @@ def test_maps_json_array_to_exact_immutable_findings_in_order( def test_maps_json_lines_and_ignores_blank_lines( copy_record: Callable[[], dict[str, Any]], ) -> None: + """Ignore blank JSONL lines without disturbing record order.""" first = copy_record() second = copy_record() second["finding_info"]["uid"] = "second" @@ -109,6 +111,7 @@ def test_maps_json_lines_and_ignores_blank_lines( def test_normalizes_status_without_trimming( copy_record: Callable[[], dict[str, Any]], source: str, expected: str ) -> None: + """Normalize status case but not unapproved surrounding whitespace.""" record = copy_record() record["status"] = source @@ -132,6 +135,7 @@ def test_normalizes_severity_case_insensitively( label: str, weight: int, ) -> None: + """Normalize declared severity labels and weights without case sensitivity.""" record = copy_record() record["severity"] = source @@ -143,6 +147,7 @@ def test_normalizes_severity_case_insensitively( def test_maps_all_fields_and_preserves_compliance_values( ocsf_record: dict[str, Any], ) -> None: + """Map every authoritative path and retain compliance encounter order.""" finding = map_ocsf_finding(ocsf_record) assert finding.model_dump() == { @@ -166,6 +171,7 @@ def test_maps_all_fields_and_preserves_compliance_values( def test_optional_values_have_safe_fallbacks( copy_record: Callable[[], dict[str, Any]], ) -> None: + """Use approved fallbacks for absent severity, compliance, and URL.""" record = copy_record() del record["severity"] del record["unmapped"]["compliance"] @@ -180,6 +186,7 @@ def test_optional_values_have_safe_fallbacks( @pytest.mark.parametrize("payload", [b"\xffsecret", b'[{"token":"secret"}']) def test_decode_errors_do_not_echo_sensitive_payload(payload: bytes) -> None: + """Keep malformed payload content out of structured decode errors.""" with pytest.raises(OcsfDecodeError) as caught: decode_ocsf_output(payload) @@ -189,6 +196,7 @@ def test_decode_errors_do_not_echo_sensitive_payload(payload: bytes) -> None: def test_non_object_record_has_safe_index() -> None: + """Identify a non-object array member by index without echoing it.""" with pytest.raises(OcsfDecodeError) as caught: decode_ocsf_output(b'[{"ok": true}, "sensitive"]') @@ -200,6 +208,7 @@ def test_non_object_record_has_safe_index() -> None: def test_missing_required_path_has_structured_mapping_error( copy_record: Callable[[], dict[str, Any]], ) -> None: + """Report the missing required path and source record index.""" record = copy_record() del record["cloud"]["account"]["uid"] @@ -215,6 +224,7 @@ def test_missing_required_path_has_structured_mapping_error( def test_empty_resources_has_structured_mapping_error( copy_record: Callable[[], dict[str, Any]], ) -> None: + """Treat the required first resource as a path-aware mapping failure.""" record = copy_record() record["resources"] = [] diff --git a/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py b/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py index 5a37145a..ab2e8ef9 100644 --- a/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py +++ b/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py @@ -16,16 +16,19 @@ def test_empty_json_array_and_blank_json_lines_are_empty() -> None: + """Decode valid empty representations to an empty immutable collection.""" assert decode_ocsf_output(b"[]") == () assert decode_ocsf_output(b" \n\n\t") == () def test_single_json_line_object_is_accepted() -> None: + """Accept one object as a one-record JSON Lines document.""" assert decode_ocsf_output('{"finding_info": {}}') == ({"finding_info": {}},) @pytest.mark.parametrize("payload", [b"null", b"42", b'"text"', b"{}\n[]"]) def test_top_level_or_jsonl_non_objects_are_rejected(payload: bytes) -> None: + """Reject scalar top levels and non-object JSONL records.""" with pytest.raises(OcsfDecodeError) as caught: decode_ocsf_output(payload) @@ -33,6 +36,7 @@ def test_top_level_or_jsonl_non_objects_are_rejected(payload: bytes) -> None: def test_command_result_must_be_success() -> None: + """Reject a CHK.004 error envelope without exposing its details.""" result = CommandResult( specification=ExecutionSpecification( executable="prowler", @@ -69,6 +73,7 @@ def test_supported_compliance_shapes_preserve_values_and_duplicates( compliance: object, expected: tuple[str, ...], ) -> None: + """Flatten evidenced compliance forms in order without deduplication.""" record = copy_record() record["unmapped"]["compliance"] = compliance @@ -78,6 +83,7 @@ def test_supported_compliance_shapes_preserve_values_and_duplicates( def test_invalid_compliance_leaf_has_structured_error( copy_record: Callable[[], dict[str, Any]], ) -> None: + """Report the exact unsupported compliance leaf path.""" record = copy_record() record["unmapped"]["compliance"] = {"CIS": ["1.1", 42]} @@ -102,6 +108,7 @@ def test_wrong_source_types_are_structured_errors( copy_record: Callable[[], dict[str, Any]], path_mutation: Callable[[dict[str, Any]], None], ) -> None: + """Convert wrong source container and scalar types to safe errors.""" record = copy_record() path_mutation(record) From 124868827b2144879edd40bb4cff6aadaf365bca Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Thu, 27 Aug 2026 11:28:15 +0200 Subject: [PATCH 3/4] feat(prowler): map nested Prowler 3.x OCSF Security Finding shape in CHK.005 (#422) --- prowler/prowler/models/findings.py | 149 ++++++++-- .../chk005_ocsf_mapping.feature | 51 ++++ .../behaviour/chk005_ocsf_mapping/conftest.py | 137 +++++++++ .../test_chk005_dual_shape_bdd.py | 268 ++++++++++++++++++ .../test_dual_shape_paths.py | 254 +++++++++++++++++ 5 files changed, 837 insertions(+), 22 deletions(-) create mode 100644 prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_dual_shape_bdd.py create mode 100644 prowler/tests/unit/chk005_ocsf_mapping/test_dual_shape_paths.py diff --git a/prowler/prowler/models/findings.py b/prowler/prowler/models/findings.py index 54594399..18ed8dcc 100644 --- a/prowler/prowler/models/findings.py +++ b/prowler/prowler/models/findings.py @@ -74,6 +74,7 @@ class OpenAevFinding(BaseModel): "PASSED": "SUCCESS", "FAIL": "FAILED", "FAILED": "FAILED", + "FAILURE": "FAILED", "MUTED": "IGNORED", "MANUAL": "IGNORED", "SUPPRESSED": "IGNORED", @@ -122,14 +123,19 @@ def map_ocsf_finding( record: dict[str, Any], *, record_index: int = 0 ) -> OpenAevFinding: """Map one decoded OCSF record to the local OpenAEV boundary model.""" - finding_info = _required_mapping(record, "finding_info", record_index) + block_key, finding_block = _selected_finding_block(record, record_index) resources = _required_sequence(record, "resources", record_index) if not resources: raise _mapping_error("missing_source_path", record_index, "resources[0]") resource = _mapping_value(resources[0], "resources[0]", record_index) cloud = _required_mapping(record, "cloud", record_index) account = _required_mapping(cloud, "cloud.account", record_index, key="account") - remediation = _required_mapping(record, "remediation", record_index) + if "remediation" in record: + remediation: Mapping[str, Any] | None = _required_mapping( + record, "remediation", record_index + ) + else: + remediation = None status = _required_string(record, "status", record_index) severity_value = record.get("severity") @@ -140,25 +146,28 @@ def map_ocsf_finding( else: raise _mapping_error("invalid_source_value", record_index, "severity") - references_value = remediation.get("references", ()) - if not isinstance(references_value, Sequence) or isinstance( - references_value, (str, bytes) - ): - raise _mapping_error( - "invalid_source_value", record_index, "remediation.references" - ) - remediation_url = None - if references_value: - remediation_url = _string_value( - references_value[0], "remediation.references[0]", record_index - ) + if remediation is not None: + references_value = remediation.get("references", ()) + if not isinstance(references_value, Sequence) or isinstance( + references_value, (str, bytes) + ): + raise _mapping_error( + "invalid_source_value", record_index, "remediation.references" + ) + remediation_url: str | None = None + if references_value: + remediation_url = _string_value( + references_value[0], "remediation.references[0]", record_index + ) + else: + remediation_url = None return OpenAevFinding( type=_required_string( - finding_info, "finding_info.uid", record_index, key="uid" + finding_block, f"{block_key}.uid", record_index, key="uid" ), value=_required_string( - finding_info, "finding_info.title", record_index, key="title" + finding_block, f"{block_key}.title", record_index, key="title" ), expectation_result=_STATUS_MAP.get(status.upper(), "MUTED"), severity=severity[0], @@ -177,12 +186,18 @@ def map_ocsf_finding( account, "cloud.account.uid", record_index, key="uid" ), compliance_tags=_compliance_tags(record, record_index), - remediation=_required_string( - remediation, "remediation.desc", record_index, key="desc" + remediation=( + _required_string(remediation, "remediation.desc", record_index, key="desc") + if remediation is not None + else _block_remediation(finding_block, block_key, record_index)[0] + ), + remediation_url=( + remediation_url + if remediation is not None + else _block_remediation(finding_block, block_key, record_index)[1] ), - remediation_url=remediation_url, description=_required_string( - finding_info, "finding_info.desc", record_index, key="desc" + finding_block, f"{block_key}.desc", record_index, key="desc" ), ) @@ -255,6 +270,21 @@ def _required_mapping( return _mapping_value(parent[lookup_key], source_path, record_index) +def _selected_finding_block( + record: Mapping[str, Any], record_index: int +) -> tuple[str, Mapping[str, Any]]: + """Select the one present finding block and its source path key.""" + if "finding_info" in record: + block_key = "finding_info" + elif "finding" in record: + block_key = "finding" + else: + raise _mapping_error( + "missing_source_path", record_index, "finding_info|finding" + ) + return block_key, _mapping_value(record[block_key], block_key, record_index) + + def _required_sequence( parent: Mapping[str, Any], source_path: str, record_index: int ) -> Sequence[Any]: @@ -285,17 +315,76 @@ def _required_string( return _string_value(parent[lookup_key], source_path, record_index) +def _block_remediation( + block: Mapping[str, Any], block_key: str, record_index: int +) -> tuple[str, str | None]: + """Map the finding-block remediation to its desc and URL value.""" + if "remediation" not in block: + raise _mapping_error( + "missing_source_path", + record_index, + f"remediation|{block_key}.remediation", + ) + remediation = _mapping_value( + block["remediation"], f"{block_key}.remediation", record_index + ) + desc = _required_string( + remediation, f"{block_key}.remediation.desc", record_index, key="desc" + ) + if "references" in remediation: + return desc, _block_remediation_url( + remediation, block_key, "references", record_index + ) + if "kb_articles" in remediation: + return desc, _block_remediation_url( + remediation, block_key, "kb_articles", record_index + ) + return desc, None + + +def _block_remediation_url( + remediation: Mapping[str, Any], + block_key: str, + key: str, + record_index: int, +) -> str | None: + """Resolve one present block URL list to its unvalidated first entry.""" + source_path = f"{block_key}.remediation.{key}" + value = remediation[key] + if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): + raise _mapping_error("invalid_source_value", record_index, source_path) + if not value: + return None + return _string_value(value[0], f"{source_path}[0]", record_index) + + def _compliance_tags(record: Mapping[str, Any], record_index: int) -> tuple[str, ...]: unmapped = record.get("unmapped") if unmapped is None: - return () + return _top_level_compliance_tags(record, record_index) unmapped_mapping = _mapping_value(unmapped, "unmapped", record_index) compliance = unmapped_mapping.get("compliance") if compliance is None: - return () + return _top_level_compliance_tags(record, record_index) return tuple(_flatten_compliance(compliance, "unmapped.compliance", record_index)) +def _top_level_compliance_tags( + record: Mapping[str, Any], record_index: int +) -> tuple[str, ...]: + """Map a present top-level compliance object to requirement tags.""" + compliance = record.get("compliance") + if compliance is None: + return () + compliance_mapping = _mapping_value(compliance, "compliance", record_index) + requirements = compliance_mapping.get("requirements") + if requirements is None: + return () + return tuple( + _flatten_requirements(requirements, "compliance.requirements", record_index) + ) + + def _flatten_compliance( value: object, source_path: str, record_index: int ) -> list[str]: @@ -316,3 +405,19 @@ def _flatten_compliance( ) return flattened raise _mapping_error("invalid_source_value", record_index, source_path) + + +def _flatten_requirements( + value: object, source_path: str, record_index: int +) -> list[str]: + """Flatten a strict string-or-list requirements value in encounter order.""" + if isinstance(value, str): + return [value] + if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): + flattened = [] + for index, nested in enumerate(value): + flattened.extend( + _flatten_requirements(nested, f"{source_path}[{index}]", record_index) + ) + return flattened + raise _mapping_error("invalid_source_value", record_index, source_path) diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature b/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature index c22834bd..646564cf 100644 --- a/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature @@ -28,6 +28,11 @@ Feature: Map raw Prowler OCSF output to OpenAEV findings | SUPPRESSED | IGNORED | | ERROR | MUTED | | UNKNOWN | MUTED | + | FAILURE | FAILED | + | failure | FAILED | + | Muted | IGNORED | + | Manual | IGNORED | + | Suppressed | IGNORED | Scenario Outline: Normalize finding severity Given an OCSF finding with severity "" @@ -77,3 +82,49 @@ Feature: Map raw Prowler OCSF output to OpenAEV findings Given an OCSF finding with whitespace around a recognized status When the finding is mapped Then expectation_result is MUTED + + # ---- Prowler 3.x nested dual-shape contract ---- + + Scenario: Map a Prowler 3.11.3 nested detection finding + Given an OCSF finding with a finding block, nested remediation, and top-level compliance + When the finding is mapped + Then all 14 OpenAEV fields are mapped from the nested shape + And compliance_tags keeps the compound requirement strings in order without splitting or deduplication + And remediation_url is the first kb_articles entry without URL validation + And unmapped nested fields such as severity_id, status_detail, and state have no effect + + Scenario: Select the present finding block by precedence + Given an OCSF finding with both finding_info and finding blocks + When the finding is mapped + Then finding_info is selected and the finding block is never read + And an empty present finding_info reports missing finding_info.uid without fallback + And a null finding_info is an invalid value at finding_info without fallback + + Scenario: Resolve remediation across top level and finding blocks + Given an OCSF finding whose selected finding block carries its own remediation + When the finding is mapped + Then a present top-level remediation always wins with its exact flat mapping + And kb_articles inside a top-level remediation is ignored leaving remediation_url absent + And references presence wins over kb_articles within one block even when references is empty + And an empty kb_articles without references leaves remediation_url absent + And neither location present is a structured error at remediation|finding.remediation + + Scenario: Resolve compliance across unmapped and top level + Given an OCSF finding with top-level compliance requirements + When the finding is mapped + Then a present non-null unmapped.compliance always wins with its flat flattening + And a null unmapped.compliance defers to top-level compliance requirements + And a present empty unmapped.compliance wins with empty tags + And string or list requirements become tags in encounter order without splitting or deduplication + And a non-string requirements leaf or a mapping is a precise-path structured error + And absent or null requirements produce empty tags + + Scenario: Report a missing finding block with a union source path + Given an OCSF finding without finding_info or finding + When the finding is mapped + Then a structured mapping error uses source path finding_info|finding + + Scenario: Keep dual-shape error coordinates content-free + Given a nested OCSF finding violating a required or typed nested path + When the finding is mapped + Then the structured mapping error carries only code, record index, and source path diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py b/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py index 31389e53..ca945626 100644 --- a/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py @@ -44,3 +44,140 @@ def factory() -> dict[str, Any]: return deepcopy(ocsf_record) return factory + + +@pytest.fixture +def ocsf_nested_record() -> dict[str, Any]: + """Return one isolated Prowler 3.11.3 nested detection finding.""" + return deepcopy( + { + "finding": { + "title": "Check S3 Account Level Public Access Block.", + "desc": "Check S3 Account Level Public Access Block.", + "supporting_data": { + "Risk": ( + "Public access policies may be applied to sensitive data buckets." + ), + "Notes": "", + }, + "remediation": { + "kb_articles": [ + "https://docs.bridgecrew.io/docs/bc_aws_s3_21#cloudformation", + "https://docs.bridgecrew.io/docs/bc_aws_s3_21#terraform", + "aws s3control put-public-access-block " + "--public-access-block-configuration " + "BlockPublicAcls=true,IgnorePublicAcls=true," + "BlockPublicPolicy=true,RestrictPublicBuckets=true " + "--account-id ", + "https://github.com/cloudmatos/matos/tree/master/" + "remediations/aws/s3/s3control/block-public-access", + "https://docs.aws.amazon.com/AmazonS3/latest/" + "userguide/access-control-block-public-access.html", + ], + "desc": ( + "You can enable Public Access Block at the account level to " + "prevent the exposure of your data stored in S3." + ), + }, + "types": ["Data Protection"], + "src_url": "", + "uid": ( + "prowler-aws-s3_account_level_public_access_blocks-" + "123456789012-us-east-1-123456789012" + ), + "related_events": [], + }, + "resources": [ + { + "group": {"name": "s3"}, + "region": "us-east-1", + "name": "123456789012", + "uid": "arn:aws:iam::123456789012:root", + "labels": [], + "type": "AwsS3Bucket", + "details": "", + } + ], + "status_detail": ( + "Block Public Access is not configured for the account " + "123456789012." + ), + "compliance": { + "status": "Failure", + "requirements": [ + "NIST-800-53-Revision-5: ac_2_6, ac_3, ac_3_7, ac_4_21, ac_6, " + "ac_17_b, ac_17_1, ac_17_4_a, ac_17_9, ac_17_10, cm_6_a, cm_9_b, " + "mp_2, sc_7_2, sc_7_3, sc_7_7, sc_7_9_a, sc_7_11, sc_7_12, " + "sc_7_16, sc_7_20, sc_7_21, sc_7_24_b, sc_7_25, sc_7_26, " + "sc_7_27, sc_7_28, sc_7_a, sc_7_b, sc_7_c, sc_25", + "GxP-21-CFR-Part-11: 11.10-d, 11.10-g", + "CIS-1.4: 2.1.5", + "AWS-Well-Architected-Framework-Security-Pillar: SEC03-BP07", + "FFIEC: d3-pc-im-b-1", + "FedRamp-Moderate-Revision-4: ac-3, ac-6, ac-17-1, ac-21-b, " + "cm-2, sc-4, sc-7-3, sc-7", + "CIS-2.0: 2.1.5", + "AWS-Foundational-Security-Best-Practices: s3", + "NIST-800-53-Revision-4: sc_7_3, sc_7", + "CISA: your-systems-3, your-data-2", + "NIST-800-171-Revision-2: 3_1_1, 3_1_2, 3_1_3, 3_1_14, " + "3_1_20, 3_3_8, 3_4_6, 3_13_2, 3_13_5", + "FedRAMP-Low-Revision-4: ac-3, ac-17, cm-2, sc-7", + "NIST-CSF-1.1: ac_3, ac_5, ds_5, ip_8, pt_3", + "MITRE-ATTACK: T1530", + "CIS-1.5: 2.1.5", + "HIPAA: 164_308_a_1_ii_b, 164_308_a_3_i", + ], + "status_detail": ( + "Block Public Access is not configured for the account " + "123456789012." + ), + }, + "message": ( + "Block Public Access is not configured for the account " + "123456789012." + ), + "severity_id": 4, + "severity": "High", + "cloud": { + "account": {"name": "", "uid": "123456789012"}, + "region": "us-east-1", + "org": {"uid": "", "name": ""}, + "provider": "aws", + "project_uid": "", + }, + "time": "2026-05-26 15:14:30.983675", + "metadata": { + "original_time": "2026-05-26T15:14:30.983675", + "profiles": ["default"], + "product": { + "language": "en", + "name": "Prowler", + "version": "3.11.3", + "vendor_name": "Prowler/ProwlerPro", + "feature": { + "name": "s3_account_level_public_access_blocks", + "uid": "s3_account_level_public_access_blocks", + "version": "3.11.3", + }, + }, + "version": "1.0.0-rc.3", + }, + "state_id": 0, + "state": "New", + "status_id": 2, + "status": "Failure", + "type_uid": 200101, + "type_name": "Security Finding: Create", + "impact_id": 0, + "impact": "Unknown", + "confidence_id": 0, + "confidence": "Unknown", + "activity_id": 1, + "activity_name": "Create", + "category_uid": 2, + "category_name": "Findings", + "class_uid": 2001, + "class_name": "Security Finding", + } + ) diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_dual_shape_bdd.py b/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_dual_shape_bdd.py new file mode 100644 index 00000000..bb731e34 --- /dev/null +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_dual_shape_bdd.py @@ -0,0 +1,268 @@ +"""Raw-pytest bindings for the CHK.005 dual-shape (Prowler 3.x nested) contract.""" + +from copy import deepcopy +from typing import Any, Callable + +import pytest + +from prowler.models.findings import ( + OcsfMappingError, + OpenAevFinding, + map_ocsf_finding, +) + +_CANARY = "CANARY-SECRET-VALUE" + + +def _mapped(record: dict[str, Any], record_index: int = 0) -> OpenAevFinding: + """Map a record expected to succeed, surfacing absent behavior clearly. + + The frozen OcsfMappingError dataclass cannot survive the contextlib + traceback reassignment in this host's pytest runner, so an escaped + structured error is converted into an explicit failure carrying its + rendered code, record index, and source path. + """ + try: + return map_ocsf_finding(record, record_index=record_index) + except OcsfMappingError as error: + pytest.fail(f"dual-shape behavior absent: {error}") + + +def test_maps_prowler_3_11_3_nested_shape_to_exact_immutable_findings( + ocsf_nested_record: dict[str, Any], +) -> None: + """H1: map the fully pinned nested record to the exact 14-field vector.""" + finding = _mapped(ocsf_nested_record) + + assert finding.model_dump() == { + "type": ( + "prowler-aws-s3_account_level_public_access_blocks-" + "123456789012-us-east-1-123456789012" + ), + "value": "Check S3 Account Level Public Access Block.", + "expectation_result": "FAILED", + "severity": "HIGH", + "severity_weight": 3, + "asset_reference": "arn:aws:iam::123456789012:root", + "asset_name": "123456789012", + "cloud_provider": "aws", + "region": "us-east-1", + "cloud_account": "123456789012", + "compliance_tags": ( + "NIST-800-53-Revision-5: ac_2_6, ac_3, ac_3_7, ac_4_21, ac_6, " + "ac_17_b, ac_17_1, ac_17_4_a, ac_17_9, ac_17_10, cm_6_a, cm_9_b, " + "mp_2, sc_7_2, sc_7_3, sc_7_7, sc_7_9_a, sc_7_11, sc_7_12, " + "sc_7_16, sc_7_20, sc_7_21, sc_7_24_b, sc_7_25, sc_7_26, " + "sc_7_27, sc_7_28, sc_7_a, sc_7_b, sc_7_c, sc_25", + "GxP-21-CFR-Part-11: 11.10-d, 11.10-g", + "CIS-1.4: 2.1.5", + "AWS-Well-Architected-Framework-Security-Pillar: SEC03-BP07", + "FFIEC: d3-pc-im-b-1", + "FedRamp-Moderate-Revision-4: ac-3, ac-6, ac-17-1, ac-21-b, " + "cm-2, sc-4, sc-7-3, sc-7", + "CIS-2.0: 2.1.5", + "AWS-Foundational-Security-Best-Practices: s3", + "NIST-800-53-Revision-4: sc_7_3, sc_7", + "CISA: your-systems-3, your-data-2", + "NIST-800-171-Revision-2: 3_1_1, 3_1_2, 3_1_3, 3_1_14, " + "3_1_20, 3_3_8, 3_4_6, 3_13_2, 3_13_5", + "FedRAMP-Low-Revision-4: ac-3, ac-17, cm-2, sc-7", + "NIST-CSF-1.1: ac_3, ac_5, ds_5, ip_8, pt_3", + "MITRE-ATTACK: T1530", + "CIS-1.5: 2.1.5", + "HIPAA: 164_308_a_1_ii_b, 164_308_a_3_i", + ), + "remediation": ( + "You can enable Public Access Block at the account level to " + "prevent the exposure of your data stored in S3." + ), + "remediation_url": "https://docs.bridgecrew.io/docs/bc_aws_s3_21#cloudformation", + "description": "Check S3 Account Level Public Access Block.", + } + + +@pytest.mark.parametrize( + ("source", "expected"), + [ + ("FAILURE", "FAILED"), + ("failure", "FAILED"), + ("Muted", "IGNORED"), + ("Manual", "IGNORED"), + ("Suppressed", "IGNORED"), + ], +) +def test_normalizes_prowler_3x_status_rows( + copy_record: Callable[[], dict[str, Any]], source: str, expected: str +) -> None: + """Additive 3.x status rows; all existing status rows stay unchanged.""" + record = copy_record() + record["status"] = source + + assert _mapped(record, record_index=7).expectation_result == expected + + +def test_finding_info_block_wins_when_both_blocks_present( + copy_record: Callable[[], dict[str, Any]], +) -> None: + """Select finding_info and never read the finding block's content.""" + record = copy_record() + record["finding"] = {"uid": _CANARY, "title": _CANARY, "desc": _CANARY} + + finding = _mapped(record) + + assert finding.type == "prowler.aws.iam.root_user_access_key" + assert finding.value == "Root user access keys should be removed" + assert finding.description == "The root user has active access keys." + assert _CANARY not in finding.model_dump_json() + + +def test_top_level_remediation_wins_over_block_remediation( + copy_record: Callable[[], dict[str, Any]], + ocsf_nested_record: dict[str, Any], +) -> None: + """A present top-level remediation keeps its exact flat mapping.""" + record = copy_record() + record["finding"] = deepcopy(ocsf_nested_record["finding"]) + record["message"] = _CANARY + + finding = _mapped(record) + + assert finding.remediation == "Delete the root user access keys." + assert finding.remediation_url == "https://example.test/remediation" + assert _CANARY not in finding.model_dump_json() + + +def test_unmapped_compliance_wins_over_erroring_top_level_compliance( + ocsf_nested_record: dict[str, Any], +) -> None: + """A non-null unmapped.compliance wins even when top-level would error.""" + record = ocsf_nested_record + record["unmapped"] = {"compliance": {"CIS-1.5": ["1.1", "1.2"]}} + record["compliance"] = 42 + + finding = _mapped(record) + + assert finding.compliance_tags == ("1.1", "1.2") + + +def test_block_remediation_references_precede_kb_articles( + ocsf_nested_record: dict[str, Any], +) -> None: + """Within one block remediation, references presence wins over kb_articles.""" + record = ocsf_nested_record + record["finding"]["remediation"]["references"] = ["https://ref.test"] + + finding = _mapped(record) + + assert finding.remediation_url == "https://ref.test" + + +def test_empty_block_references_win_presence_over_kb_articles( + ocsf_nested_record: dict[str, Any], +) -> None: + """A present empty references list resolves the URL to absent.""" + record = ocsf_nested_record + record["finding"]["remediation"]["references"] = [] + + assert _mapped(record).remediation_url is None + + +def test_empty_present_finding_info_never_defers_to_finding( + ocsf_nested_record: dict[str, Any], +) -> None: + """C1: an empty present finding_info errors at finding_info.uid.""" + record = ocsf_nested_record + record["finding_info"] = {} + record["message"] = _CANARY + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=5) + + assert caught.value.code == "missing_source_path" + assert caught.value.record_index == 5 + assert caught.value.source_path == "finding_info.uid" + assert _CANARY not in str(caught.value) + + +def test_null_finding_info_is_invalid_value_without_finding_fallback( + ocsf_nested_record: dict[str, Any], +) -> None: + """C2: a null finding_info is invalid_source_value at finding_info.""" + record = ocsf_nested_record + record["finding_info"] = None + record["message"] = _CANARY + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=6) + + assert caught.value.code == "invalid_source_value" + assert caught.value.record_index == 6 + assert caught.value.source_path == "finding_info" + assert _CANARY not in str(caught.value) + + +def test_empty_block_kb_articles_without_references_leaves_url_absent( + ocsf_nested_record: dict[str, Any], +) -> None: + """C3: selected block remediation with kb_articles=[] and no references.""" + record = ocsf_nested_record + record["finding"]["remediation"]["kb_articles"] = [] + + finding = _mapped(record) + + assert finding.remediation_url is None + assert finding.remediation.startswith("You can enable Public Access Block") + + +def test_null_unmapped_compliance_defers_to_top_level_requirements( + ocsf_nested_record: dict[str, Any], +) -> None: + """C4: a null unmapped.compliance defers to the 16 requirement strings.""" + record = ocsf_nested_record + record["unmapped"] = {"compliance": None} + + finding = _mapped(record) + + assert finding.compliance_tags == tuple(record["compliance"]["requirements"]) + assert len(finding.compliance_tags) == 16 + + +def test_present_empty_unmapped_compliance_wins_with_empty_tags( + ocsf_nested_record: dict[str, Any], +) -> None: + """C5: {} is a valid present value; precedence keeps tags empty.""" + record = ocsf_nested_record + record["unmapped"] = {"compliance": {}} + + assert _mapped(record).compliance_tags == () + + +def test_top_level_remediation_ignores_kb_articles( + copy_record: Callable[[], dict[str, Any]], +) -> None: + """C6: kb_articles inside a top-level remediation is ignored.""" + record = copy_record() + del record["remediation"]["references"] + record["remediation"]["kb_articles"] = ["https://kb.test"] + + assert _mapped(record).remediation_url is None + + +def test_present_top_level_compliance_without_requirements_maps_empty_tags( + ocsf_nested_record: dict[str, Any], +) -> None: + """A present top-level compliance without requirements maps to empty tags.""" + record = ocsf_nested_record + record["compliance"] = {"status": "Failure"} + + assert _mapped(record).compliance_tags == () + + +def test_present_top_level_compliance_with_null_requirements_maps_empty_tags( + ocsf_nested_record: dict[str, Any], +) -> None: + """A present top-level compliance with requirements=None maps to empty tags.""" + record = ocsf_nested_record + record["compliance"] = {"requirements": None} + + assert _mapped(record).compliance_tags == () diff --git a/prowler/tests/unit/chk005_ocsf_mapping/test_dual_shape_paths.py b/prowler/tests/unit/chk005_ocsf_mapping/test_dual_shape_paths.py new file mode 100644 index 00000000..46ea4b47 --- /dev/null +++ b/prowler/tests/unit/chk005_ocsf_mapping/test_dual_shape_paths.py @@ -0,0 +1,254 @@ +"""Path-level dual-shape error coordinates for the CHK.005 mapper.""" + +from typing import Any + +import pytest + +from prowler.models.findings import ( + OcsfMappingError, + OpenAevFinding, + map_ocsf_finding, +) + +_CANARY = "CANARY-SECRET-VALUE" + + +def _mapped(record: dict[str, Any], record_index: int = 0) -> OpenAevFinding: + """Map a record expected to succeed, surfacing absent behavior clearly. + + The frozen OcsfMappingError dataclass cannot survive the contextlib + traceback reassignment in this host's pytest runner, so an escaped + structured error is converted into an explicit failure carrying its + rendered code, record index, and source path. + """ + try: + return map_ocsf_finding(record, record_index=record_index) + except OcsfMappingError as error: + pytest.fail(f"dual-shape behavior absent: {error}") + + +def _nested_record() -> dict[str, Any]: + """One nested 3.11.3-style record with canary-tainted content everywhere.""" + return { + "finding": { + "uid": "prowler-aws-s3-x-123456789012-us-east-1-123456789012", + "title": "Check.", + "desc": "Check.", + "remediation": { + "desc": "Enable the block.", + "kb_articles": ["https://kb.test"], + }, + }, + "resources": [ + {"uid": "arn:aws:iam::123456789012:root", "name": "123456789012"} + ], + "cloud": { + "provider": "aws", + "region": "us-east-1", + "account": {"uid": "123456789012"}, + }, + "status": "Failure", + "severity": "High", + "compliance": { + "status": _CANARY, + "requirements": ["NIST-800-53-Revision-5: ac_3"], + "status_detail": _CANARY, + }, + "message": _CANARY, + "severity_id": 4, + "state": "New", + "state_id": 0, + } + + +def test_missing_both_finding_blocks_reports_union_path() -> None: + """E1: neither finding block → missing_source_path finding_info|finding.""" + record = _nested_record() + del record["finding"] + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=4) + + assert caught.value.code == "missing_source_path" + assert caught.value.record_index == 4 + assert caught.value.source_path == "finding_info|finding" + assert _CANARY not in str(caught.value) + + +def test_present_non_object_finding_info_has_structured_error() -> None: + """E2a: a present non-object finding_info → invalid_source_value finding_info.""" + record = _nested_record() + record["finding_info"] = _CANARY + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=9) + + assert caught.value.code == "invalid_source_value" + assert caught.value.record_index == 9 + assert caught.value.source_path == "finding_info" + assert _CANARY not in str(caught.value) + + +def test_present_non_object_finding_block_has_structured_error() -> None: + """E2b: a present non-object finding block → invalid_source_value finding.""" + record = _nested_record() + record["finding"] = _CANARY + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=9) + + assert caught.value.code == "invalid_source_value" + assert caught.value.record_index == 9 + assert caught.value.source_path == "finding" + assert _CANARY not in str(caught.value) + + +def test_missing_remediation_everywhere_reports_union_path() -> None: + """E3: no top-level and no block remediation → union-path error.""" + record = _nested_record() + del record["finding"]["remediation"] + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=3) + + assert caught.value.code == "missing_source_path" + assert caught.value.record_index == 3 + assert caught.value.source_path == "remediation|finding.remediation" + assert _CANARY not in str(caught.value) + + +def test_block_remediation_without_desc_reports_block_desc_path() -> None: + """E4: block remediation with kb_articles but no desc → desc path error.""" + record = _nested_record() + del record["finding"]["remediation"]["desc"] + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=8) + + assert caught.value.code == "missing_source_path" + assert caught.value.record_index == 8 + assert caught.value.source_path == "finding.remediation.desc" + assert _CANARY not in str(caught.value) + + +@pytest.mark.parametrize( + ("mutation", "source_path"), + [ + ( + lambda record: record["finding"]["remediation"].update(kb_articles=42), + "finding.remediation.kb_articles", + ), + ( + lambda record: record["finding"]["remediation"].update(kb_articles=[42]), + "finding.remediation.kb_articles[0]", + ), + ( + lambda record: record["finding"]["remediation"].update(references=42), + "finding.remediation.references", + ), + ( + lambda record: record["finding"]["remediation"].update(references=[42]), + "finding.remediation.references[0]", + ), + ], +) +def test_block_remediation_type_errors_report_precise_paths( + mutation: Any, source_path: str +) -> None: + """E5: non-list lists and non-string heads → precise invalid_source_value.""" + record = _nested_record() + mutation(record) + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=2) + + assert caught.value.code == "invalid_source_value" + assert caught.value.record_index == 2 + assert caught.value.source_path == source_path + assert _CANARY not in str(caught.value) + + +def test_non_mapping_top_level_compliance_reports_compliance_path() -> None: + """E6a: top-level compliance that is not a mapping → invalid at compliance.""" + record = _nested_record() + record["compliance"] = _CANARY + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=11) + + assert caught.value.code == "invalid_source_value" + assert caught.value.record_index == 11 + assert caught.value.source_path == "compliance" + assert _CANARY not in str(caught.value) + + +def test_string_requirements_are_accepted_as_single_tag() -> None: + """E6b: a single-string requirements value is accepted as one tag.""" + record = _nested_record() + record["compliance"]["requirements"] = "one-string" + + assert _mapped(record).compliance_tags == ("one-string",) + + +@pytest.mark.parametrize( + ("requirements", "source_path"), + [ + (42, "compliance.requirements"), + (["a", 42], "compliance.requirements[1]"), + ({"K": ["v"]}, "compliance.requirements"), + ], +) +def test_invalid_requirements_grammar_reports_precise_paths( + requirements: Any, source_path: str +) -> None: + """E6c: non-string leaves and mappings → precise invalid_source_value.""" + record = _nested_record() + record["compliance"]["requirements"] = requirements + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=13) + + assert caught.value.code == "invalid_source_value" + assert caught.value.record_index == 13 + assert caught.value.source_path == source_path + assert _CANARY not in str(caught.value) + + +def test_missing_flat_remediation_everywhere_reports_union_path() -> None: + """E7: flat finding_info record with remediation nowhere → union-path error.""" + record = { + "finding_info": { + "uid": "prowler.aws.iam.root_user_access_key", + "title": "Root user access keys should be removed.", + "desc": "The root user has active access keys.", + }, + "resources": [ + {"uid": "arn:aws:iam::123456789012:root", "name": "root"} + ], + "cloud": { + "provider": "aws", + "region": "eu-west-1", + "account": {"uid": "123456789012"}, + }, + "status": "PASS", + "severity": "High", + "message": _CANARY, + } + + with pytest.raises(OcsfMappingError) as caught: + map_ocsf_finding(record, record_index=14) + + assert caught.value.code == "missing_source_path" + assert caught.value.record_index == 14 + assert caught.value.source_path == "remediation|finding_info.remediation" + assert _CANARY not in str(caught.value) + + +def test_block_kb_articles_head_is_used_without_url_validation() -> None: + """E8: a non-URL kb_articles head maps through without URL validation.""" + record = _nested_record() + record["finding"]["remediation"]["kb_articles"] = ["not a url"] + + finding = _mapped(record) + + assert finding.remediation_url == "not a url" From 343b9dcc23960441a476de9427ed9a42a7541fe1 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Thu, 27 Aug 2026 12:25:25 +0200 Subject: [PATCH 4/4] fix(prowler): restore CHK005 OCSF mapping compatibility (#422) --- prowler/prowler/models/findings.py | 67 ++++++++++++++----- .../chk005_ocsf_mapping.feature | 47 +++++++------ .../behaviour/chk005_ocsf_mapping/conftest.py | 18 ++--- .../test_chk005_dual_shape_bdd.py | 3 +- .../test_chk005_ocsf_mapping_bdd.py | 55 +++++++++++---- .../unit/chk005_ocsf_mapping/conftest.py | 7 +- .../test_decode_and_paths.py | 5 +- .../test_dual_shape_paths.py | 4 +- 8 files changed, 138 insertions(+), 68 deletions(-) diff --git a/prowler/prowler/models/findings.py b/prowler/prowler/models/findings.py index 18ed8dcc..a0c509b8 100644 --- a/prowler/prowler/models/findings.py +++ b/prowler/prowler/models/findings.py @@ -69,16 +69,14 @@ class OpenAevFinding(BaseModel): description: str -_STATUS_MAP = { +_STATUS_CODE_MAP = { "PASS": "SUCCESS", "PASSED": "SUCCESS", "FAIL": "FAILED", "FAILED": "FAILED", "FAILURE": "FAILED", - "MUTED": "IGNORED", - "MANUAL": "IGNORED", - "SUPPRESSED": "IGNORED", } +_IGNORED_LIFECYCLE_STATUSES = {"SUPPRESSED", "MUTED", "MANUAL"} _SEVERITY_MAP = { "CRITICAL": ("CRITICAL", 4), @@ -92,6 +90,13 @@ class OpenAevFinding(BaseModel): _MAPPING_MESSAGE = "unable to map Prowler OCSF record" +def _expectation_result(status: str, status_code: str) -> str: + """Keep lifecycle suppression separate from the check result code.""" + if status.upper() in _IGNORED_LIFECYCLE_STATUSES: + return "IGNORED" + return _STATUS_CODE_MAP.get(status_code.upper(), "IGNORED") + + def decode_ocsf_output(payload: bytes | str) -> tuple[dict[str, Any], ...]: """Decode raw Prowler JSON array or JSON Lines output.""" if isinstance(payload, bytes): @@ -128,8 +133,27 @@ def map_ocsf_finding( if not resources: raise _mapping_error("missing_source_path", record_index, "resources[0]") resource = _mapping_value(resources[0], "resources[0]", record_index) - cloud = _required_mapping(record, "cloud", record_index) - account = _required_mapping(cloud, "cloud.account", record_index, key="account") + if "cloud" in record: + cloud = _required_mapping(record, "cloud", record_index) + account = _required_mapping(cloud, "cloud.account", record_index, key="account") + cloud_provider = _required_string( + cloud, "cloud.provider", record_index, key="provider" + ) + region = _required_string(cloud, "cloud.region", record_index, key="region") + cloud_account = _required_string( + account, "cloud.account.uid", record_index, key="uid" + ) + else: + unmapped = _required_mapping(record, "unmapped", record_index) + cloud_provider = _required_string( + unmapped, "unmapped.provider", record_index, key="provider" + ) + cloud_account = _required_string( + unmapped, "unmapped.provider_uid", record_index, key="provider_uid" + ) + region = _required_string( + resource, "resources[0].namespace", record_index, key="namespace" + ) if "remediation" in record: remediation: Mapping[str, Any] | None = _required_mapping( record, "remediation", record_index @@ -138,6 +162,11 @@ def map_ocsf_finding( remediation = None status = _required_string(record, "status", record_index) + status_code = ( + _required_string(record, "status_code", record_index) + if "status_code" in record + else status + ) severity_value = record.get("severity") if severity_value is None: severity = ("INFO", 0) @@ -169,7 +198,7 @@ def map_ocsf_finding( value=_required_string( finding_block, f"{block_key}.title", record_index, key="title" ), - expectation_result=_STATUS_MAP.get(status.upper(), "MUTED"), + expectation_result=_expectation_result(status, status_code), severity=severity[0], severity_weight=severity[1], asset_reference=_required_string( @@ -178,13 +207,9 @@ def map_ocsf_finding( asset_name=_required_string( resource, "resources[0].name", record_index, key="name" ), - cloud_provider=_required_string( - cloud, "cloud.provider", record_index, key="provider" - ), - region=_required_string(cloud, "cloud.region", record_index, key="region"), - cloud_account=_required_string( - account, "cloud.account.uid", record_index, key="uid" - ), + cloud_provider=cloud_provider, + region=region, + cloud_account=cloud_account, compliance_tags=_compliance_tags(record, record_index), remediation=( _required_string(remediation, "remediation.desc", record_index, key="desc") @@ -393,9 +418,17 @@ def _flatten_compliance( if isinstance(value, Mapping): flattened = [] for key, nested in value.items(): - flattened.extend( - _flatten_compliance(nested, f"{source_path}.{key}", record_index) - ) + if nested is True: + flattened.append(str(key)) + elif nested is False or nested is None: + continue + else: + flattened.extend( + f"{key}:{tag}" + for tag in _flatten_compliance( + nested, f"{source_path}.{key}", record_index + ) + ) return flattened if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): flattened = [] diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature b/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature index 646564cf..f8adea98 100644 --- a/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/chk005_ocsf_mapping.feature @@ -13,26 +13,23 @@ Feature: Map raw Prowler OCSF output to OpenAEV findings Then each object becomes one finding in source order Scenario Outline: Normalize finding status - Given an OCSF finding with status "" + Given an OCSF finding with lifecycle status "" and result status code "" When the finding is mapped Then expectation_result is "" Examples: - | source_status | result_status | - | PASS | SUCCESS | - | passed | SUCCESS | - | FAIL | FAILED | - | failed | FAILED | - | MUTED | IGNORED | - | manual | IGNORED | - | SUPPRESSED | IGNORED | - | ERROR | MUTED | - | UNKNOWN | MUTED | - | FAILURE | FAILED | - | failure | FAILED | - | Muted | IGNORED | - | Manual | IGNORED | - | Suppressed | IGNORED | + | status | status_code | result_status | + | New | PASS | SUCCESS | + | New | pass | SUCCESS | + | New | FAIL | FAILED | + | New | failed | FAILED | + | Suppressed | FAIL | IGNORED | + | suppressed | PASS | IGNORED | + | MUTED | FAIL | IGNORED | + | manual | PASS | IGNORED | + | New | ERROR | IGNORED | + | New | UNKNOWN | IGNORED | + | New | PASS | IGNORED | Scenario Outline: Normalize finding severity Given an OCSF finding with severity "" @@ -61,6 +58,11 @@ Feature: Map raw Prowler OCSF output to OpenAEV findings And compliance_tags is empty And remediation_url is absent + Scenario: Map a cloudless Kubernetes finding + Given an OCSF finding without cloud and with provider identity under unmapped + When the finding is mapped + Then provider and account come from unmapped and region comes from the resource namespace + # ---- Constraints identified ---- Scenario: Reject malformed raw output without disclosing it @@ -78,10 +80,10 @@ Feature: Map raw Prowler OCSF output to OpenAEV findings When the finding is mapped Then a structured mapping error identifies resources[0] - Scenario: Reject whitespace-padded status rather than silently normalizing it - Given an OCSF finding with whitespace around a recognized status + Scenario: Reject whitespace-padded status code rather than silently normalizing it + Given an OCSF finding with whitespace around a recognized status code When the finding is mapped - Then expectation_result is MUTED + Then expectation_result is IGNORED # ---- Prowler 3.x nested dual-shape contract ---- @@ -93,6 +95,11 @@ Feature: Map raw Prowler OCSF output to OpenAEV findings And remediation_url is the first kb_articles entry without URL validation And unmapped nested fields such as severity_id, status_detail, and state have no effect + Scenario: Preserve a nested Prowler 3.x result when status_code is absent + Given a nested OCSF finding whose legacy result is carried by status + When the finding is mapped + Then the legacy status vocabulary maps without changing the nested field shape + Scenario: Select the present finding block by precedence Given an OCSF finding with both finding_info and finding blocks When the finding is mapped @@ -112,7 +119,7 @@ Feature: Map raw Prowler OCSF output to OpenAEV findings Scenario: Resolve compliance across unmapped and top level Given an OCSF finding with top-level compliance requirements When the finding is mapped - Then a present non-null unmapped.compliance always wins with its flat flattening + Then a present non-null unmapped.compliance always wins with framework-preserving flattening And a null unmapped.compliance defers to top-level compliance requirements And a present empty unmapped.compliance wins with empty tags And string or list requirements become tags in encounter order without splitting or deduplication diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py b/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py index ca945626..2d31c480 100644 --- a/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/conftest.py @@ -1,7 +1,7 @@ """Local fixtures for CHK.005 behaviour.""" from copy import deepcopy -from typing import Any +from typing import Any, Callable import pytest @@ -15,7 +15,8 @@ def ocsf_record() -> dict[str, Any]: "title": "Root user access keys should be removed", "desc": "The root user has active access keys.", }, - "status": "PASS", + "status": "New", + "status_code": "PASS", "severity": "High", "resources": [{"uid": "arn:aws:iam::123456789012:root", "name": "root"}], "cloud": { @@ -37,7 +38,9 @@ def ocsf_record() -> dict[str, Any]: @pytest.fixture -def copy_record(ocsf_record: dict[str, Any]): +def copy_record( + ocsf_record: dict[str, Any], +) -> Callable[[], dict[str, Any]]: """Return a factory that isolates mutable source records.""" def factory() -> dict[str, Any]: @@ -56,7 +59,8 @@ def ocsf_nested_record() -> dict[str, Any]: "desc": "Check S3 Account Level Public Access Block.", "supporting_data": { "Risk": ( - "Public access policies may be applied to sensitive data buckets." + "Public access policies may be applied to sensitive " + "data buckets." ), "Notes": "", }, @@ -99,8 +103,7 @@ def ocsf_nested_record() -> dict[str, Any]: } ], "status_detail": ( - "Block Public Access is not configured for the account " - "123456789012." + "Block Public Access is not configured for the account " "123456789012." ), "compliance": { "status": "Failure", @@ -134,8 +137,7 @@ def ocsf_nested_record() -> dict[str, Any]: ), }, "message": ( - "Block Public Access is not configured for the account " - "123456789012." + "Block Public Access is not configured for the account " "123456789012." ), "severity_id": 4, "severity": "High", diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_dual_shape_bdd.py b/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_dual_shape_bdd.py index bb731e34..1cc46ab0 100644 --- a/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_dual_shape_bdd.py +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_dual_shape_bdd.py @@ -96,6 +96,7 @@ def test_normalizes_prowler_3x_status_rows( ) -> None: """Additive 3.x status rows; all existing status rows stay unchanged.""" record = copy_record() + del record["status_code"] record["status"] = source assert _mapped(record, record_index=7).expectation_result == expected @@ -142,7 +143,7 @@ def test_unmapped_compliance_wins_over_erroring_top_level_compliance( finding = _mapped(record) - assert finding.compliance_tags == ("1.1", "1.2") + assert finding.compliance_tags == ("CIS-1.5:1.1", "CIS-1.5:1.2") def test_block_remediation_references_precede_kb_articles( diff --git a/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py b/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py index 3064ecd2..e637cab5 100644 --- a/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py +++ b/prowler/tests/behaviour/chk005_ocsf_mapping/test_chk005_ocsf_mapping_bdd.py @@ -94,26 +94,31 @@ def test_maps_json_lines_and_ignores_blank_lines( @pytest.mark.parametrize( - ("source", "expected"), + ("status", "status_code", "expected"), [ - ("PASS", "SUCCESS"), - ("passed", "SUCCESS"), - ("FAIL", "FAILED"), - ("failed", "FAILED"), - ("MUTED", "IGNORED"), - ("manual", "IGNORED"), - ("SUPPRESSED", "IGNORED"), - ("ERROR", "MUTED"), - ("UNKNOWN", "MUTED"), - (" PASS ", "MUTED"), + ("New", "PASS", "SUCCESS"), + ("New", "pass", "SUCCESS"), + ("New", "FAIL", "FAILED"), + ("New", "failed", "FAILED"), + ("Suppressed", "FAIL", "IGNORED"), + ("suppressed", "PASS", "IGNORED"), + ("MUTED", "FAIL", "IGNORED"), + ("manual", "PASS", "IGNORED"), + ("New", "ERROR", "IGNORED"), + ("New", "UNKNOWN", "IGNORED"), + ("New", " PASS ", "IGNORED"), ], ) def test_normalizes_status_without_trimming( - copy_record: Callable[[], dict[str, Any]], source: str, expected: str + copy_record: Callable[[], dict[str, Any]], + status: str, + status_code: str, + expected: str, ) -> None: """Normalize status case but not unapproved surrounding whitespace.""" record = copy_record() - record["status"] = source + record["status"] = status + record["status_code"] = status_code assert map_ocsf_finding(record, record_index=7).expectation_result == expected @@ -161,7 +166,11 @@ def test_maps_all_fields_and_preserves_compliance_values( "cloud_provider": "aws", "region": "eu-west-1", "cloud_account": "123456789012", - "compliance_tags": ("1.1", "1.2", "op.acc.6"), + "compliance_tags": ( + "CIS-1.5:1.1", + "CIS-1.5:1.2", + "ENS-RD2022:op.acc.6", + ), "remediation": "Delete the root user access keys.", "remediation_url": "https://example.test/remediation", "description": "The root user has active access keys.", @@ -184,6 +193,24 @@ def test_optional_values_have_safe_fallbacks( assert finding.remediation_url is None +def test_cloudless_finding_uses_unmapped_provider_and_resource_namespace( + copy_record: Callable[[], dict[str, Any]], +) -> None: + """Map Kubernetes/provider identity when the OCSF cloud object is absent.""" + record = copy_record() + del record["cloud"] + record["unmapped"].update(provider="kubernetes", provider_uid="cluster-production") + record["resources"][0]["namespace"] = "payments" + + finding = map_ocsf_finding(record) + + assert ( + finding.cloud_provider, + finding.cloud_account, + finding.region, + ) == ("kubernetes", "cluster-production", "payments") + + @pytest.mark.parametrize("payload", [b"\xffsecret", b'[{"token":"secret"}']) def test_decode_errors_do_not_echo_sensitive_payload(payload: bytes) -> None: """Keep malformed payload content out of structured decode errors.""" diff --git a/prowler/tests/unit/chk005_ocsf_mapping/conftest.py b/prowler/tests/unit/chk005_ocsf_mapping/conftest.py index 9c73c7ab..b02dee1e 100644 --- a/prowler/tests/unit/chk005_ocsf_mapping/conftest.py +++ b/prowler/tests/unit/chk005_ocsf_mapping/conftest.py @@ -1,17 +1,18 @@ """Local fixtures for CHK.005 mapping units.""" from copy import deepcopy -from typing import Any +from typing import Any, Callable import pytest @pytest.fixture -def copy_record(): +def copy_record() -> Callable[[], dict[str, Any]]: """Return isolated representative Prowler 5.36 records.""" record: dict[str, Any] = { "finding_info": {"uid": "uid", "title": "title", "desc": "description"}, - "status": "PASS", + "status": "New", + "status_code": "PASS", "severity": "high", "resources": [{"uid": "resource", "name": "resource name"}], "cloud": { diff --git a/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py b/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py index ab2e8ef9..f5203014 100644 --- a/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py +++ b/prowler/tests/unit/chk005_ocsf_mapping/test_decode_and_paths.py @@ -62,10 +62,11 @@ def test_command_result_must_be_success() -> None: @pytest.mark.parametrize( ("compliance", "expected"), [ - ({"A": "x", "B": ["y", "x"]}, ("x", "y", "x")), + ({"A": "x", "B": ["y", "x"]}, ("A:x", "B:y", "B:x")), (["a", "b"], ("a", "b")), ("one", ("one",)), - ({"A": {"first": "x", "second": ["y"]}}, ("x", "y")), + ({"A": {"first": "x", "second": ["y"]}}, ("A:first:x", "A:second:y")), + ({"A": True, "B": False, "C": None}, ("A",)), ], ) def test_supported_compliance_shapes_preserve_values_and_duplicates( diff --git a/prowler/tests/unit/chk005_ocsf_mapping/test_dual_shape_paths.py b/prowler/tests/unit/chk005_ocsf_mapping/test_dual_shape_paths.py index 46ea4b47..aae2e781 100644 --- a/prowler/tests/unit/chk005_ocsf_mapping/test_dual_shape_paths.py +++ b/prowler/tests/unit/chk005_ocsf_mapping/test_dual_shape_paths.py @@ -222,9 +222,7 @@ def test_missing_flat_remediation_everywhere_reports_union_path() -> None: "title": "Root user access keys should be removed.", "desc": "The root user has active access keys.", }, - "resources": [ - {"uid": "arn:aws:iam::123456789012:root", "name": "root"} - ], + "resources": [{"uid": "arn:aws:iam::123456789012:root", "name": "root"}], "cloud": { "provider": "aws", "region": "eu-west-1",