From 8150130dc5ca3a8137b63fddf8983605dd195333 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 22:57:32 +0200 Subject: [PATCH 01/13] test(prowler): define CLI engine behavior (#422) --- .../behaviour/chk003_cli_engine/__init__.py | 1 + .../chk003_cli_engine.feature | 49 ++++ .../behaviour/chk003_cli_engine/conftest.py | 54 +++++ .../test_chk003_cli_engine_bdd.py | 213 ++++++++++++++++++ .../tests/unit/chk003_cli_engine/__init__.py | 1 + .../test_subprocess_executor.py | 45 ++++ 6 files changed, 363 insertions(+) create mode 100644 prowler/tests/behaviour/chk003_cli_engine/__init__.py create mode 100644 prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature create mode 100644 prowler/tests/behaviour/chk003_cli_engine/conftest.py create mode 100644 prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py create mode 100644 prowler/tests/unit/chk003_cli_engine/__init__.py create mode 100644 prowler/tests/unit/chk003_cli_engine/test_subprocess_executor.py diff --git a/prowler/tests/behaviour/chk003_cli_engine/__init__.py b/prowler/tests/behaviour/chk003_cli_engine/__init__.py new file mode 100644 index 00000000..5341f52d --- /dev/null +++ b/prowler/tests/behaviour/chk003_cli_engine/__init__.py @@ -0,0 +1 @@ +"""CHK.003 CLI engine behaviour tests.""" diff --git a/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature new file mode 100644 index 00000000..e014cbca --- /dev/null +++ b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature @@ -0,0 +1,49 @@ +@cli-engine +Feature: Safe CLI engine orchestration + + Scenario: Create an immutable execution specification + Given a validated CLI request and resolvable command inputs + When the engine creates an execution specification + Then its executable, ordered arguments, environment, working context, input bytes, and parser selection are fixed + + Scenario: Invoke without shell interpolation + Given an allowed request containing shell metacharacters as arguments + When the engine runs the request + Then the executable and every argument are supplied as separate structured values + + Scenario: Reject a disallowed request + Given a request that violates command policy + When the engine runs the request + Then it returns a policy error before resolution or execution + + Scenario: Report an unresolved command value + Given an allowed request whose command cannot be resolved + When the engine runs the request + Then it returns a resolution error before execution + + Scenario: Report a process failure + Given a resolved execution specification + When process startup fails or the process outcome is unsuccessful + Then the engine returns an execution error retaining captured output bytes + + Scenario: Report unparseable output + Given a successful process whose output cannot be parsed + When the engine parses the output + Then it returns a parsing error retaining the original output bytes + + Scenario: Preserve bytes through execution and parsing + Given input, output, and error streams containing arbitrary bytes + When the engine transfers the payload and returns a successful result + Then every byte remains unchanged and the parsed result is returned with captured streams + +# ---- Constraints identified ---- + + Scenario: Do not parse an unsuccessful process outcome + Given a process returns an unsuccessful outcome + When the engine runs the request + Then parsing is not attempted + + Scenario: Preserve empty structured boundaries + Given an allowed executable with no arguments, environment entries, working context, or input bytes + When the engine runs the request + Then the empty values remain distinct and unchanged diff --git a/prowler/tests/behaviour/chk003_cli_engine/conftest.py b/prowler/tests/behaviour/chk003_cli_engine/conftest.py new file mode 100644 index 00000000..b2dde792 --- /dev/null +++ b/prowler/tests/behaviour/chk003_cli_engine/conftest.py @@ -0,0 +1,54 @@ +"""Fixtures local to CHK.003 CLI engine behaviour.""" + +from dataclasses import dataclass, field +from typing import Any + +import pytest + + +@dataclass +class RecordingPorts: + """Deterministic port bundle recording boundary order and values.""" + + events: list[str] = field(default_factory=list) + allowed: bool = True + resolvable: bool = True + outcome: Any = None + parse_error: Exception | None = None + invocation: Any = None + parsed_payload: bytes | None = None + + def check(self, request: Any) -> None: + """Record policy evaluation and reject when configured.""" + self.events.append("policy") + if not self.allowed: + raise request.errors.PolicyError("request rejected") + + def resolve(self, request: Any) -> Any: + """Record resolution and return the prepared specification.""" + self.events.append("resolution") + if not self.resolvable: + raise request.errors.ResolutionError("value unresolved") + return request.specification + + def execute(self, specification: Any) -> Any: + """Record execution and return the configured outcome.""" + self.events.append("execution") + self.invocation = specification + if isinstance(self.outcome, Exception): + raise self.outcome + return self.outcome + + def parse(self, parser: str, payload: bytes) -> object: + """Record parsing and return deterministic parsed data.""" + self.events.append("parsing") + self.parsed_payload = payload + if self.parse_error is not None: + raise self.parse_error + return {"parser": parser, "size": len(payload)} + + +@pytest.fixture +def recording_ports() -> RecordingPorts: + """Return fresh deterministic CLI boundary ports.""" + return RecordingPorts() diff --git a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py new file mode 100644 index 00000000..a66aa2da --- /dev/null +++ b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py @@ -0,0 +1,213 @@ +"""Executable behaviour contract for the generic CHK.003 CLI engine.""" + +import importlib +from dataclasses import FrozenInstanceError +from typing import Any + +import pytest + +from .conftest import RecordingPorts + + +def _api() -> Any: + try: + return importlib.import_module("prowler.cli_engine") + except ModuleNotFoundError: + pytest.fail("CHK.003 generic CLI engine behaviour is absent") + + +def _request(api: Any, **changes: Any) -> Any: + values = { + "executable": "scanner", + "arguments": ("--format", "json"), + "environment": (("LANG", "C"),), + "working_directory": "/work", + "input_bytes": b"input", + "parser": "json", + } + values.update(changes) + return api.ValidatedCliRequest(**values) + + +def _engine(api: Any, ports: RecordingPorts) -> Any: + return api.CliEngine(policy=ports, resolver=ports, executor=ports, parser=ports) + + +def _attach(api: Any, ports: RecordingPorts, request: Any) -> Any: + object.__setattr__(request, "errors", api) + object.__setattr__( + request, "specification", api.ExecutionSpecification.from_request(request) + ) + return request + + +def test_execution_specification_is_deeply_immutable() -> None: + """Copy nested mutable request values into immutable structures.""" + api = _api() + arguments = ["--format", "json"] + environment = {"LANG": "C"} + request = api.ValidatedCliRequest( + executable="scanner", + arguments=arguments, + environment=environment, + working_directory="/work", + input_bytes=b"input", + parser="json", + ) + specification = api.ExecutionSpecification.from_request(request) + arguments.append("--changed") + environment["LANG"] = "changed" + + assert specification.arguments == ("--format", "json") + assert specification.environment == (("LANG", "C"),) + with pytest.raises(FrozenInstanceError): + specification.executable = "other" + + +def test_structured_arguments_keep_shell_metacharacters_inert( + recording_ports: RecordingPorts, +) -> None: + """Keep shell syntax as inert individual arguments.""" + api = _api() + request = _attach( + api, + recording_ports, + _request(api, arguments=("value; rm -rf /", "$(unsafe)", "a && b")), + ) + recording_ports.outcome = api.ProcessOutcome(0, b"ok", b"") + + _engine(api, recording_ports).run(request) + + assert recording_ports.invocation.argv == ( + "scanner", + "value; rm -rf /", + "$(unsafe)", + "a && b", + ) + assert not hasattr(recording_ports.invocation, "command") + + +@pytest.mark.parametrize( + ("allowed", "resolvable", "error_type", "events"), + [ + (False, True, "PolicyError", ["policy"]), + (True, False, "ResolutionError", ["policy", "resolution"]), + ], +) +def test_failures_short_circuit_in_boundary_order( + recording_ports: RecordingPorts, + allowed: bool, + resolvable: bool, + error_type: str, + events: list[str], +) -> None: + """Stop at the first failed boundary in required order.""" + api = _api() + recording_ports.allowed = allowed + recording_ports.resolvable = resolvable + request = _attach(api, recording_ports, _request(api)) + + with pytest.raises(getattr(api, error_type)): + _engine(api, recording_ports).run(request) + + assert recording_ports.events == events + + +def test_nonzero_outcome_retains_bytes_and_skips_parsing( + recording_ports: RecordingPorts, +) -> None: + """Retain failed process streams without invoking the parser.""" + api = _api() + stdout = b"partial\x00\xff\n" + stderr = b"failure\x80\r\n" + recording_ports.outcome = api.ProcessOutcome(7, stdout, stderr) + request = _attach(api, recording_ports, _request(api)) + + with pytest.raises(api.ExecutionError) as caught: + _engine(api, recording_ports).run(request) + + assert (caught.value.stdout, caught.value.stderr) == (stdout, stderr) + assert recording_ports.events == ["policy", "resolution", "execution"] + + +def test_executor_exception_becomes_execution_error( + recording_ports: RecordingPorts, +) -> None: + """Translate process-start exceptions into execution errors.""" + api = _api() + recording_ports.outcome = OSError("cannot start") + request = _attach(api, recording_ports, _request(api)) + + with pytest.raises(api.ExecutionError) as caught: + _engine(api, recording_ports).run(request) + + assert caught.value.stdout == b"" + assert caught.value.stderr == b"" + + +def test_parser_exception_retains_original_process_bytes( + recording_ports: RecordingPorts, +) -> None: + """Retain exact process streams when parsing raises.""" + api = _api() + stdout = b"\xff\x00not-json\n" + stderr = b"warning\x80" + recording_ports.outcome = api.ProcessOutcome(0, stdout, stderr) + recording_ports.parse_error = ValueError("invalid") + request = _attach(api, recording_ports, _request(api)) + + with pytest.raises(api.ParsingError) as caught: + _engine(api, recording_ports).run(request) + + assert (caught.value.stdout, caught.value.stderr) == (stdout, stderr) + assert recording_ports.parsed_payload == stdout + + +def test_success_preserves_all_bytes_and_returns_parsed_result( + recording_ports: RecordingPorts, +) -> None: + """Return parsed data with exact captured streams.""" + api = _api() + payload = b"\x00\xffline1\r\nline2\n" + stderr = b"\x80warning\x00" + request = _attach(api, recording_ports, _request(api, input_bytes=payload)) + recording_ports.outcome = api.ProcessOutcome(0, payload, stderr) + + result = _engine(api, recording_ports).run(request) + + assert recording_ports.invocation.input_bytes == payload + assert recording_ports.parsed_payload == payload + assert result == api.ExecutionSuccess( + parsed={"parser": "json", "size": len(payload)}, + stdout=payload, + stderr=stderr, + ) + assert recording_ports.events == ["policy", "resolution", "execution", "parsing"] + + +def test_empty_values_remain_structured(recording_ports: RecordingPorts) -> None: + """Preserve valid empty boundaries without coercion.""" + api = _api() + request = _attach( + api, + recording_ports, + _request( + api, + arguments=(), + environment=(), + working_directory=None, + input_bytes=b"", + ), + ) + recording_ports.outcome = api.ProcessOutcome(0, b"", b"") + + _engine(api, recording_ports).run(request) + + assert recording_ports.invocation == api.ExecutionSpecification( + executable="scanner", + arguments=(), + environment=(), + working_directory=None, + input_bytes=b"", + parser="json", + ) diff --git a/prowler/tests/unit/chk003_cli_engine/__init__.py b/prowler/tests/unit/chk003_cli_engine/__init__.py new file mode 100644 index 00000000..d68f99bb --- /dev/null +++ b/prowler/tests/unit/chk003_cli_engine/__init__.py @@ -0,0 +1 @@ +"""Focused CHK.003 CLI engine unit tests.""" diff --git a/prowler/tests/unit/chk003_cli_engine/test_subprocess_executor.py b/prowler/tests/unit/chk003_cli_engine/test_subprocess_executor.py new file mode 100644 index 00000000..0ba18027 --- /dev/null +++ b/prowler/tests/unit/chk003_cli_engine/test_subprocess_executor.py @@ -0,0 +1,45 @@ +"""Unit tests for the structured subprocess adapter.""" + +import importlib +from typing import Any +from unittest.mock import patch + +import pytest + + +def _api() -> Any: + try: + return importlib.import_module("prowler.cli_engine") + except ModuleNotFoundError: + pytest.fail("CHK.003 subprocess executor is absent") + + +def test_subprocess_adapter_forces_shell_false_and_preserves_bytes() -> None: + """Invoke subprocess with structured values and no shell.""" + api = _api() + specification = api.ExecutionSpecification( + executable="tool", + arguments=("a;b",), + environment=(("KEY", "value"),), + working_directory="/work", + input_bytes=b"\x00\xff", + parser="raw", + ) + completed = __import__("subprocess").CompletedProcess( + args=specification.argv, returncode=0, stdout=b"\xff", stderr=b"\x00" + ) + + with patch("subprocess.run", return_value=completed) as run: + outcome = api.SubprocessExecutor().execute(specification) + + run.assert_called_once_with( + ("tool", "a;b"), + input=b"\x00\xff", + stdout=-1, + stderr=-1, + cwd="/work", + env={"KEY": "value"}, + shell=False, + check=False, + ) + assert outcome == api.ProcessOutcome(0, b"\xff", b"\x00") From 19aa0ab3659b053c769d24ce21acae193f0ad6bc Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:00:59 +0200 Subject: [PATCH 02/13] feat(prowler): add generic CLI engine (#422) --- prowler/prowler/cli_engine.py | 175 ++++++++++++++++++ prowler/prowler/cli_engine_errors.py | 34 ++++ .../behaviour/chk003_cli_engine/conftest.py | 14 +- .../test_chk003_cli_engine_bdd.py | 38 ++-- 4 files changed, 229 insertions(+), 32 deletions(-) create mode 100644 prowler/prowler/cli_engine.py create mode 100644 prowler/prowler/cli_engine_errors.py diff --git a/prowler/prowler/cli_engine.py b/prowler/prowler/cli_engine.py new file mode 100644 index 00000000..81ffca64 --- /dev/null +++ b/prowler/prowler/cli_engine.py @@ -0,0 +1,175 @@ +"""Generic hexagonal engine for structured local process execution.""" + +from __future__ import annotations + +import subprocess +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from typing import Protocol + +from prowler import cli_engine_errors as errors + +ExecutionError = errors.ExecutionError +ParsingError = errors.ParsingError +PolicyError = errors.PolicyError +ResolutionError = errors.ResolutionError + + +def _freeze_environment( + environment: Mapping[str, str] | Sequence[tuple[str, str]], +) -> tuple[tuple[str, str], ...]: + """Copy environment entries into a stable ordered tuple.""" + if isinstance(environment, Mapping): + return tuple(environment.items()) + return tuple(environment) + + +@dataclass(frozen=True) +class ValidatedCliRequest: + """Validated command inputs used to derive an execution specification.""" + + executable: str + arguments: Sequence[str] + environment: Mapping[str, str] | Sequence[tuple[str, str]] + working_directory: str | None + input_bytes: bytes + parser: str + + +@dataclass(frozen=True) +class ExecutionSpecification: + """Deeply immutable structured process and parser specification.""" + + executable: str + arguments: tuple[str, ...] + environment: tuple[tuple[str, str], ...] + working_directory: str | None + input_bytes: bytes + parser: str + + @classmethod + def from_request(cls, request: ValidatedCliRequest) -> ExecutionSpecification: + """Copy a validated request into immutable nested values.""" + return cls( + executable=request.executable, + arguments=tuple(request.arguments), + environment=_freeze_environment(request.environment), + working_directory=request.working_directory, + input_bytes=request.input_bytes, + parser=request.parser, + ) + + @property + def argv(self) -> tuple[str, ...]: + """Return executable and ordered arguments as structured values.""" + return (self.executable, *self.arguments) + + +@dataclass(frozen=True) +class ProcessOutcome: + """Exact process outcome returned by an executor port.""" + + return_code: int + stdout: bytes + stderr: bytes + + +@dataclass(frozen=True) +class ExecutionSuccess: + """Parsed result accompanied by exact captured process streams.""" + + parsed: object + stdout: bytes + stderr: bytes + + +class PolicyPort(Protocol): + """Authorize one immutable execution specification.""" + + def check(self, specification: ExecutionSpecification) -> None: + """Raise PolicyError when execution is not permitted.""" + + +class ResolverPort(Protocol): + """Resolve values needed by one immutable execution specification.""" + + def resolve(self, specification: ExecutionSpecification) -> ExecutionSpecification: + """Return the resolved specification or raise ResolutionError.""" + + +class ExecutorPort(Protocol): + """Execute one immutable specification without shell interpretation.""" + + def execute(self, specification: ExecutionSpecification) -> ProcessOutcome: + """Return exact process bytes and return code.""" + + +class ParserPort(Protocol): + """Parse exact successful stdout bytes.""" + + def parse(self, parser: str, payload: bytes) -> object: + """Return the selected parser's result.""" + + +class CliEngine: + """Orchestrate policy, resolution, execution, and parsing in order.""" + + def __init__( + self, + *, + policy: PolicyPort, + resolver: ResolverPort, + executor: ExecutorPort, + parser: ParserPort, + ) -> None: + """Bind injected ports without selecting provider-specific behavior.""" + self._policy = policy + self._resolver = resolver + self._executor = executor + self._parser = parser + + def run(self, request: ValidatedCliRequest) -> ExecutionSuccess: + """Run a validated request through each boundary exactly in order.""" + specification = ExecutionSpecification.from_request(request) + self._policy.check(specification) + specification = self._resolver.resolve(specification) + try: + outcome = self._executor.execute(specification) + except ExecutionError: + raise + except OSError as error: + raise ExecutionError(str(error)) from error + if outcome.return_code != 0: + raise ExecutionError( + "process returned an unsuccessful outcome", + stdout=outcome.stdout, + stderr=outcome.stderr, + return_code=outcome.return_code, + ) + try: + parsed = self._parser.parse(specification.parser, outcome.stdout) + except ParsingError: + raise + except Exception as error: + raise ParsingError( + str(error), stdout=outcome.stdout, stderr=outcome.stderr + ) from error + return ExecutionSuccess(parsed, outcome.stdout, outcome.stderr) + + +class SubprocessExecutor: + """Subprocess-backed executor using structured argv and shell=False.""" + + def execute(self, specification: ExecutionSpecification) -> ProcessOutcome: + """Execute a specification and preserve all process bytes exactly.""" + completed = subprocess.run( # noqa: S603 - policy-approved structured argv + specification.argv, + input=specification.input_bytes, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=specification.working_directory, + env=dict(specification.environment), + shell=False, + check=False, + ) + return ProcessOutcome(completed.returncode, completed.stdout, completed.stderr) diff --git a/prowler/prowler/cli_engine_errors.py b/prowler/prowler/cli_engine_errors.py new file mode 100644 index 00000000..19e3a144 --- /dev/null +++ b/prowler/prowler/cli_engine_errors.py @@ -0,0 +1,34 @@ +"""Distinct failure types for the generic CLI engine.""" + +from dataclasses import dataclass + + +class CliEngineError(Exception): + """Base class for distinct generic CLI engine failures.""" + + +class PolicyError(CliEngineError): + """The policy boundary rejected the immutable specification.""" + + +class ResolutionError(CliEngineError): + """The resolution boundary could not resolve the specification.""" + + +@dataclass(frozen=True) +class ExecutionError(CliEngineError): + """Process failure retaining exact captured streams.""" + + message: str + stdout: bytes = b"" + stderr: bytes = b"" + return_code: int | None = None + + +@dataclass(frozen=True) +class ParsingError(CliEngineError): + """Parser failure retaining the original process streams.""" + + message: str + stdout: bytes + stderr: bytes diff --git a/prowler/tests/behaviour/chk003_cli_engine/conftest.py b/prowler/tests/behaviour/chk003_cli_engine/conftest.py index b2dde792..8703b5b1 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/conftest.py +++ b/prowler/tests/behaviour/chk003_cli_engine/conftest.py @@ -18,18 +18,22 @@ class RecordingPorts: invocation: Any = None parsed_payload: bytes | None = None - def check(self, request: Any) -> None: + def check(self, specification: Any) -> None: """Record policy evaluation and reject when configured.""" self.events.append("policy") if not self.allowed: - raise request.errors.PolicyError("request rejected") + from prowler.cli_engine import PolicyError - def resolve(self, request: Any) -> Any: + raise PolicyError("request rejected") + + def resolve(self, specification: Any) -> Any: """Record resolution and return the prepared specification.""" self.events.append("resolution") if not self.resolvable: - raise request.errors.ResolutionError("value unresolved") - return request.specification + from prowler.cli_engine import ResolutionError + + raise ResolutionError("value unresolved") + return specification def execute(self, specification: Any) -> Any: """Record execution and return the configured outcome.""" diff --git a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py index a66aa2da..5512617f 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py +++ b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py @@ -33,14 +33,6 @@ def _engine(api: Any, ports: RecordingPorts) -> Any: return api.CliEngine(policy=ports, resolver=ports, executor=ports, parser=ports) -def _attach(api: Any, ports: RecordingPorts, request: Any) -> Any: - object.__setattr__(request, "errors", api) - object.__setattr__( - request, "specification", api.ExecutionSpecification.from_request(request) - ) - return request - - def test_execution_specification_is_deeply_immutable() -> None: """Copy nested mutable request values into immutable structures.""" api = _api() @@ -69,11 +61,7 @@ def test_structured_arguments_keep_shell_metacharacters_inert( ) -> None: """Keep shell syntax as inert individual arguments.""" api = _api() - request = _attach( - api, - recording_ports, - _request(api, arguments=("value; rm -rf /", "$(unsafe)", "a && b")), - ) + request = _request(api, arguments=("value; rm -rf /", "$(unsafe)", "a && b")) recording_ports.outcome = api.ProcessOutcome(0, b"ok", b"") _engine(api, recording_ports).run(request) @@ -105,7 +93,7 @@ def test_failures_short_circuit_in_boundary_order( api = _api() recording_ports.allowed = allowed recording_ports.resolvable = resolvable - request = _attach(api, recording_ports, _request(api)) + request = _request(api) with pytest.raises(getattr(api, error_type)): _engine(api, recording_ports).run(request) @@ -121,7 +109,7 @@ def test_nonzero_outcome_retains_bytes_and_skips_parsing( stdout = b"partial\x00\xff\n" stderr = b"failure\x80\r\n" recording_ports.outcome = api.ProcessOutcome(7, stdout, stderr) - request = _attach(api, recording_ports, _request(api)) + request = _request(api) with pytest.raises(api.ExecutionError) as caught: _engine(api, recording_ports).run(request) @@ -136,7 +124,7 @@ def test_executor_exception_becomes_execution_error( """Translate process-start exceptions into execution errors.""" api = _api() recording_ports.outcome = OSError("cannot start") - request = _attach(api, recording_ports, _request(api)) + request = _request(api) with pytest.raises(api.ExecutionError) as caught: _engine(api, recording_ports).run(request) @@ -154,7 +142,7 @@ def test_parser_exception_retains_original_process_bytes( stderr = b"warning\x80" recording_ports.outcome = api.ProcessOutcome(0, stdout, stderr) recording_ports.parse_error = ValueError("invalid") - request = _attach(api, recording_ports, _request(api)) + request = _request(api) with pytest.raises(api.ParsingError) as caught: _engine(api, recording_ports).run(request) @@ -170,7 +158,7 @@ def test_success_preserves_all_bytes_and_returns_parsed_result( api = _api() payload = b"\x00\xffline1\r\nline2\n" stderr = b"\x80warning\x00" - request = _attach(api, recording_ports, _request(api, input_bytes=payload)) + request = _request(api, input_bytes=payload) recording_ports.outcome = api.ProcessOutcome(0, payload, stderr) result = _engine(api, recording_ports).run(request) @@ -188,16 +176,12 @@ def test_success_preserves_all_bytes_and_returns_parsed_result( def test_empty_values_remain_structured(recording_ports: RecordingPorts) -> None: """Preserve valid empty boundaries without coercion.""" api = _api() - request = _attach( + request = _request( api, - recording_ports, - _request( - api, - arguments=(), - environment=(), - working_directory=None, - input_bytes=b"", - ), + arguments=(), + environment=(), + working_directory=None, + input_bytes=b"", ) recording_ports.outcome = api.ProcessOutcome(0, b"", b"") From 9fe57e31f4bb082cafdc805dad2ef2c3f03f2ebc Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:02:42 +0200 Subject: [PATCH 03/13] test(prowler): expose CLI engine trust bypasses (#422) --- .../behaviour/chk003_cli_engine/conftest.py | 9 +++- .../test_chk003_cli_engine_bdd.py | 44 +++++++++++++++++++ 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/prowler/tests/behaviour/chk003_cli_engine/conftest.py b/prowler/tests/behaviour/chk003_cli_engine/conftest.py index 8703b5b1..fe007662 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/conftest.py +++ b/prowler/tests/behaviour/chk003_cli_engine/conftest.py @@ -15,24 +15,31 @@ class RecordingPorts: resolvable: bool = True outcome: Any = None parse_error: Exception | None = None + resolver_return: Any = None + policy_specification: Any = None + resolution_specification: Any = None invocation: Any = None parsed_payload: bytes | None = None def check(self, specification: Any) -> None: """Record policy evaluation and reject when configured.""" self.events.append("policy") + self.policy_specification = specification if not self.allowed: from prowler.cli_engine import PolicyError raise PolicyError("request rejected") def resolve(self, specification: Any) -> Any: - """Record resolution and return the prepared specification.""" + """Record resolution without replacing the immutable specification.""" self.events.append("resolution") + self.resolution_specification = specification if not self.resolvable: from prowler.cli_engine import ResolutionError raise ResolutionError("value unresolved") + if self.resolver_return is not None: + return self.resolver_return return specification def execute(self, specification: Any) -> Any: diff --git a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py index 5512617f..72815a96 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py +++ b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py @@ -75,6 +75,28 @@ def test_structured_arguments_keep_shell_metacharacters_inert( assert not hasattr(recording_ports.invocation, "command") +def test_resolver_cannot_replace_policy_approved_specification( + recording_ports: RecordingPorts, +) -> None: + """Keep the exact policy-approved specification through execution.""" + api = _api() + request = _request(api) + replacement = api.ExecutionSpecification.from_request( + _request(api, executable="unapproved", arguments=("--bypass",)) + ) + recording_ports.resolver_return = replacement + recording_ports.outcome = api.ProcessOutcome(0, b"ok", b"") + + _engine(api, recording_ports).run(request) + + assert ( + recording_ports.policy_specification is recording_ports.resolution_specification + ) + assert recording_ports.policy_specification is recording_ports.invocation + assert recording_ports.invocation.executable == "scanner" + assert recording_ports.invocation.arguments == ("--format", "json") + + @pytest.mark.parametrize( ("allowed", "resolvable", "error_type", "events"), [ @@ -151,6 +173,28 @@ def test_parser_exception_retains_original_process_bytes( assert recording_ports.parsed_payload == stdout +def test_parser_parsing_error_is_rebuilt_with_actual_process_bytes( + recording_ports: RecordingPorts, +) -> None: + """Replace parser-owned evidence while retaining its failure as the cause.""" + api = _api() + stdout = b"actual stdout\x00\xff" + stderr = b"actual stderr\x80" + parser_error = api.ParsingError( + "invalid provider output", stdout=b"forged", stderr=b"" + ) + recording_ports.outcome = api.ProcessOutcome(0, stdout, stderr) + recording_ports.parse_error = parser_error + + with pytest.raises(api.ParsingError) as caught: + _engine(api, recording_ports).run(_request(api)) + + assert caught.value is not parser_error + assert caught.value.message == "invalid provider output" + assert (caught.value.stdout, caught.value.stderr) == (stdout, stderr) + assert caught.value.__cause__ is parser_error + + def test_success_preserves_all_bytes_and_returns_parsed_result( recording_ports: RecordingPorts, ) -> None: From 5fbc2e92bc5914e02320a24a55831e66a5695810 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:04:43 +0200 Subject: [PATCH 04/13] fix(prowler): preserve CLI engine trust boundaries (#422) --- prowler/prowler/cli_engine.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/prowler/prowler/cli_engine.py b/prowler/prowler/cli_engine.py index 81ffca64..6d5a6db3 100644 --- a/prowler/prowler/cli_engine.py +++ b/prowler/prowler/cli_engine.py @@ -93,8 +93,8 @@ def check(self, specification: ExecutionSpecification) -> None: class ResolverPort(Protocol): """Resolve values needed by one immutable execution specification.""" - def resolve(self, specification: ExecutionSpecification) -> ExecutionSpecification: - """Return the resolved specification or raise ResolutionError.""" + def resolve(self, specification: ExecutionSpecification) -> None: + """Resolve required values or raise ResolutionError.""" class ExecutorPort(Protocol): @@ -132,7 +132,7 @@ def run(self, request: ValidatedCliRequest) -> ExecutionSuccess: """Run a validated request through each boundary exactly in order.""" specification = ExecutionSpecification.from_request(request) self._policy.check(specification) - specification = self._resolver.resolve(specification) + self._resolver.resolve(specification) try: outcome = self._executor.execute(specification) except ExecutionError: @@ -148,11 +148,10 @@ def run(self, request: ValidatedCliRequest) -> ExecutionSuccess: ) try: parsed = self._parser.parse(specification.parser, outcome.stdout) - except ParsingError: - raise except Exception as error: + message = error.message if isinstance(error, ParsingError) else str(error) raise ParsingError( - str(error), stdout=outcome.stdout, stderr=outcome.stderr + message, stdout=outcome.stdout, stderr=outcome.stderr ) from error return ExecutionSuccess(parsed, outcome.stdout, outcome.stderr) From d58a4d07d3720285074eaf435abf13435b509d88 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:06:57 +0200 Subject: [PATCH 05/13] test(prowler): define core CLI engine envelope (#422) --- .../chk003_cli_engine.feature | 112 ++++---- .../behaviour/chk003_cli_engine/conftest.py | 60 ++--- .../test_chk003_cli_engine_bdd.py | 245 +++++++----------- .../unit/chk003_cli_engine/test_adapters.py | 110 ++++++++ .../test_subprocess_executor.py | 45 ---- 5 files changed, 295 insertions(+), 277 deletions(-) create mode 100644 prowler/tests/unit/chk003_cli_engine/test_adapters.py delete mode 100644 prowler/tests/unit/chk003_cli_engine/test_subprocess_executor.py diff --git a/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature index e014cbca..30a28dce 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature +++ b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature @@ -1,49 +1,73 @@ @cli-engine -Feature: Safe CLI engine orchestration - - Scenario: Create an immutable execution specification - Given a validated CLI request and resolvable command inputs - When the engine creates an execution specification - Then its executable, ordered arguments, environment, working context, input bytes, and parser selection are fixed - - Scenario: Invoke without shell interpolation - Given an allowed request containing shell metacharacters as arguments - When the engine runs the request - Then the executable and every argument are supplied as separate structured values - - Scenario: Reject a disallowed request - Given a request that violates command policy - When the engine runs the request - Then it returns a policy error before resolution or execution - - Scenario: Report an unresolved command value - Given an allowed request whose command cannot be resolved - When the engine runs the request - Then it returns a resolution error before execution - - Scenario: Report a process failure - Given a resolved execution specification - When process startup fails or the process outcome is unsuccessful - Then the engine returns an execution error retaining captured output bytes - - Scenario: Report unparseable output - Given a successful process whose output cannot be parsed - When the engine parses the output - Then it returns a parsing error retaining the original output bytes - - Scenario: Preserve bytes through execution and parsing - Given input, output, and error streams containing arbitrary bytes - When the engine transfers the payload and returns a successful result - Then every byte remains unchanged and the parsed result is returned with captured streams +Feature: Safe local CLI engine orchestration + The CLI engine executes one policy-approved immutable specification and + reports expected failures in a result envelope. + + Scenario: Preserve one deeply immutable execution specification + Given mutable validated command inputs + When an execution specification is constructed + Then executable, argv, environment, cwd, stdin, parser, timeout, and output acceptance limit are fixed + + Scenario: Keep shell syntax inert + Given an allowed structured command containing shell metacharacters + When the engine runs the command + Then every value reaches the executor as a separate argument + + Scenario: Preserve the policy-approved specification through every boundary + Given a policy-approved immutable execution specification + When resolution validates and the engine runs it + Then policy, resolution, execution, and parsing observe that exact specification + + Scenario Outline: Return distinct expected failures without raising + Given the boundary reports an expected failure + When the engine runs the command + Then CommandResult.error is a error + And later boundaries are not called + + Examples: + | boundary | error | + | policy | PolicyError | + | resolution | ResolutionError | + | execution | ExecutionError | + | parsing | ParsingError | + + Scenario: Retain exact bytes for an unsuccessful process + Given a process returns nonzero with arbitrary stdout and stderr bytes + When the engine runs the command + Then its execution error retains the return code and exact process bytes + + Scenario: Own parser-error evidence at the engine boundary + Given a parser reports forged or missing process evidence + When the engine handles the parser failure + Then its parsing error contains actual process stdout and stderr + And safe parser context and cause detail are retained + + Scenario Outline: Parse supported process output + Given successful output for parser + When the engine runs the command + Then the parsed value is + + Examples: + | parser | value | + | raw | exact bytes | + | text | text | + | json | structured data| + | lines | text lines | + | regex | captures | # ---- Constraints identified ---- - Scenario: Do not parse an unsuccessful process outcome - Given a process returns an unsuccessful outcome - When the engine runs the request - Then parsing is not attempted + Scenario: Classify oversized captured output honestly + Given process output has already been captured beyond the accepted size + When the engine handles the process outcome + Then it returns an output_too_large_after_capture execution error with exact bytes + + Scenario: Preserve arbitrary bytes without decoding + Given stdin, stdout, and stderr contain invalid UTF-8, NULs, and newlines + When execution succeeds with the raw parser + Then all bytes remain exact - Scenario: Preserve empty structured boundaries - Given an allowed executable with no arguments, environment entries, working context, or input bytes - When the engine runs the request - Then the empty values remain distinct and unchanged + Scenario: Expose only the core CLI engine API + Given CHK.003 is installed + Then prowler._core.cli_engine is importable + And prowler.cli_engine and prowler.cli_engine_errors are absent diff --git a/prowler/tests/behaviour/chk003_cli_engine/conftest.py b/prowler/tests/behaviour/chk003_cli_engine/conftest.py index fe007662..8e7ff303 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/conftest.py +++ b/prowler/tests/behaviour/chk003_cli_engine/conftest.py @@ -1,4 +1,4 @@ -"""Fixtures local to CHK.003 CLI engine behaviour.""" +"""Fixtures local to CHK.003 behaviour tests.""" from dataclasses import dataclass, field from typing import Any @@ -8,58 +8,38 @@ @dataclass class RecordingPorts: - """Deterministic port bundle recording boundary order and values.""" + """Deterministic ports recording identity, order, and exact payloads.""" events: list[str] = field(default_factory=list) - allowed: bool = True - resolvable: bool = True - outcome: Any = None - parse_error: Exception | None = None - resolver_return: Any = None - policy_specification: Any = None - resolution_specification: Any = None - invocation: Any = None - parsed_payload: bytes | None = None + seen: list[Any] = field(default_factory=list) + policy_error: Any = None + resolution_error: Any = None + execution_result: Any = None + parsing_result: Any = None - def check(self, specification: Any) -> None: - """Record policy evaluation and reject when configured.""" + def check(self, specification: Any) -> Any: self.events.append("policy") - self.policy_specification = specification - if not self.allowed: - from prowler.cli_engine import PolicyError + self.seen.append(specification) + return self.policy_error - raise PolicyError("request rejected") - - def resolve(self, specification: Any) -> Any: - """Record resolution without replacing the immutable specification.""" + def validate(self, specification: Any) -> Any: self.events.append("resolution") - self.resolution_specification = specification - if not self.resolvable: - from prowler.cli_engine import ResolutionError - - raise ResolutionError("value unresolved") - if self.resolver_return is not None: - return self.resolver_return - return specification + self.seen.append(specification) + return self.resolution_error def execute(self, specification: Any) -> Any: - """Record execution and return the configured outcome.""" self.events.append("execution") - self.invocation = specification - if isinstance(self.outcome, Exception): - raise self.outcome - return self.outcome + self.seen.append(specification) + return self.execution_result - def parse(self, parser: str, payload: bytes) -> object: - """Record parsing and return deterministic parsed data.""" + def parse(self, specification: Any, payload: bytes) -> Any: self.events.append("parsing") - self.parsed_payload = payload - if self.parse_error is not None: - raise self.parse_error - return {"parser": parser, "size": len(payload)} + self.seen.append(specification) + if self.parsing_result is not None: + return self.parsing_result + return {"parser": specification.output.parser, "bytes": payload} @pytest.fixture def recording_ports() -> RecordingPorts: - """Return fresh deterministic CLI boundary ports.""" return RecordingPorts() diff --git a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py index 72815a96..f3a05816 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py +++ b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py @@ -1,4 +1,4 @@ -"""Executable behaviour contract for the generic CHK.003 CLI engine.""" +"""Executable behaviour contract for CHK.003.""" import importlib from dataclasses import FrozenInstanceError @@ -11,231 +11,180 @@ def _api() -> Any: try: - return importlib.import_module("prowler.cli_engine") + return importlib.import_module("prowler._core.cli_engine") except ModuleNotFoundError: - pytest.fail("CHK.003 generic CLI engine behaviour is absent") + pytest.fail("canonical prowler._core.cli_engine API is absent") def _request(api: Any, **changes: Any) -> Any: values = { "executable": "scanner", - "arguments": ("--format", "json"), - "environment": (("LANG", "C"),), + "arguments": ["--format", "json"], + "environment": {"LANG": "C"}, "working_directory": "/work", "input_bytes": b"input", - "parser": "json", + "output": api.OutputSpecification(parser="json"), + "timeout_seconds": 30.0, + "maximum_accepted_output_bytes": 4096, } values.update(changes) - return api.ValidatedCliRequest(**values) + return api.ValidatedCommandRequest(**values) def _engine(api: Any, ports: RecordingPorts) -> Any: return api.CliEngine(policy=ports, resolver=ports, executor=ports, parser=ports) -def test_execution_specification_is_deeply_immutable() -> None: - """Copy nested mutable request values into immutable structures.""" +def test_specification_is_deeply_immutable() -> None: api = _api() arguments = ["--format", "json"] environment = {"LANG": "C"} - request = api.ValidatedCliRequest( - executable="scanner", - arguments=arguments, - environment=environment, - working_directory="/work", - input_bytes=b"input", - parser="json", - ) + output = api.OutputSpecification(parser="regex", pattern=r"id=(\d+)") + request = _request(api, arguments=arguments, environment=environment, output=output) + specification = api.ExecutionSpecification.from_request(request) - arguments.append("--changed") + arguments.append("changed") environment["LANG"] = "changed" - assert specification.arguments == ("--format", "json") + assert specification.argv == ("scanner", "--format", "json") assert specification.environment == (("LANG", "C"),) + assert specification.output == output with pytest.raises(FrozenInstanceError): specification.executable = "other" -def test_structured_arguments_keep_shell_metacharacters_inert( - recording_ports: RecordingPorts, -) -> None: - """Keep shell syntax as inert individual arguments.""" +def test_metacharacters_are_inert_structured_arguments(recording_ports: RecordingPorts) -> None: api = _api() - request = _request(api, arguments=("value; rm -rf /", "$(unsafe)", "a && b")) - recording_ports.outcome = api.ProcessOutcome(0, b"ok", b"") + recording_ports.execution_result = api.ProcessOutcome(0, b"ok", b"") + request = _request(api, arguments=["a; rm -rf /", "$(touch nope)", "x && y"]) _engine(api, recording_ports).run(request) - assert recording_ports.invocation.argv == ( + assert recording_ports.seen[2].argv == ( "scanner", - "value; rm -rf /", - "$(unsafe)", - "a && b", + "a; rm -rf /", + "$(touch nope)", + "x && y", ) - assert not hasattr(recording_ports.invocation, "command") + assert not hasattr(recording_ports.seen[2], "shell") -def test_resolver_cannot_replace_policy_approved_specification( - recording_ports: RecordingPorts, -) -> None: - """Keep the exact policy-approved specification through execution.""" +def test_same_exact_specification_crosses_all_boundaries(recording_ports: RecordingPorts) -> None: api = _api() - request = _request(api) - replacement = api.ExecutionSpecification.from_request( - _request(api, executable="unapproved", arguments=("--bypass",)) - ) - recording_ports.resolver_return = replacement - recording_ports.outcome = api.ProcessOutcome(0, b"ok", b"") + recording_ports.execution_result = api.ProcessOutcome(0, b"ok", b"") - _engine(api, recording_ports).run(request) + result = _engine(api, recording_ports).run(_request(api)) - assert ( - recording_ports.policy_specification is recording_ports.resolution_specification - ) - assert recording_ports.policy_specification is recording_ports.invocation - assert recording_ports.invocation.executable == "scanner" - assert recording_ports.invocation.arguments == ("--format", "json") + assert recording_ports.events == ["policy", "resolution", "execution", "parsing"] + assert len({id(value) for value in recording_ports.seen}) == 1 + assert result.specification is recording_ports.seen[0] @pytest.mark.parametrize( - ("allowed", "resolvable", "error_type", "events"), + ("boundary", "error_name", "events"), [ - (False, True, "PolicyError", ["policy"]), - (True, False, "ResolutionError", ["policy", "resolution"]), + ("policy", "PolicyError", ["policy"]), + ("resolution", "ResolutionError", ["policy", "resolution"]), + ("execution", "ExecutionError", ["policy", "resolution", "execution"]), + ("parsing", "ParsingError", ["policy", "resolution", "execution", "parsing"]), ], ) -def test_failures_short_circuit_in_boundary_order( - recording_ports: RecordingPorts, - allowed: bool, - resolvable: bool, - error_type: str, - events: list[str], +def test_expected_failures_use_result_envelope( + recording_ports: RecordingPorts, boundary: str, error_name: str, events: list[str] ) -> None: - """Stop at the first failed boundary in required order.""" api = _api() - recording_ports.allowed = allowed - recording_ports.resolvable = resolvable - request = _request(api) + error_type = getattr(api, error_name) + error = error_type(message=f"{boundary} failed") + recording_ports.execution_result = api.ProcessOutcome(0, b"output", b"warning") + setattr(recording_ports, f"{boundary}_error" if boundary != "parsing" else "parsing_result", error) + if boundary == "execution": + recording_ports.execution_result = error - with pytest.raises(getattr(api, error_type)): - _engine(api, recording_ports).run(request) + result = _engine(api, recording_ports).run(_request(api)) + assert isinstance(result.error, error_type) assert recording_ports.events == events -def test_nonzero_outcome_retains_bytes_and_skips_parsing( +def test_unsuccessful_outcome_retains_exact_bytes_and_skips_parser( recording_ports: RecordingPorts, ) -> None: - """Retain failed process streams without invoking the parser.""" api = _api() stdout = b"partial\x00\xff\n" stderr = b"failure\x80\r\n" - recording_ports.outcome = api.ProcessOutcome(7, stdout, stderr) - request = _request(api) + recording_ports.execution_result = api.ProcessOutcome(17, stdout, stderr) - with pytest.raises(api.ExecutionError) as caught: - _engine(api, recording_ports).run(request) + result = _engine(api, recording_ports).run(_request(api)) - assert (caught.value.stdout, caught.value.stderr) == (stdout, stderr) + assert result.stdout == stdout and result.stderr == stderr + assert result.error == api.ExecutionError( + message="process returned an unsuccessful outcome", + kind="unsuccessful_process", + stdout=stdout, + stderr=stderr, + return_code=17, + ) assert recording_ports.events == ["policy", "resolution", "execution"] -def test_executor_exception_becomes_execution_error( +def test_engine_replaces_parser_owned_evidence_and_preserves_context( recording_ports: RecordingPorts, ) -> None: - """Translate process-start exceptions into execution errors.""" api = _api() - recording_ports.outcome = OSError("cannot start") - request = _request(api) + stdout, stderr = b"actual\x00\xff", b"warning\x80\n" + recording_ports.execution_result = api.ProcessOutcome(0, stdout, stderr) + recording_ports.parsing_result = api.ParsingError( + message="invalid document", + stdout=b"forged", + stderr=b"forged", + context=(("line", "7"),), + cause="JSONDecodeError", + ) - with pytest.raises(api.ExecutionError) as caught: - _engine(api, recording_ports).run(request) + result = _engine(api, recording_ports).run(_request(api)) - assert caught.value.stdout == b"" - assert caught.value.stderr == b"" + assert result.error == api.ParsingError( + message="invalid document", + stdout=stdout, + stderr=stderr, + context=(("line", "7"),), + cause="JSONDecodeError", + ) -def test_parser_exception_retains_original_process_bytes( +def test_post_capture_output_size_classification_is_honest( recording_ports: RecordingPorts, ) -> None: - """Retain exact process streams when parsing raises.""" api = _api() - stdout = b"\xff\x00not-json\n" - stderr = b"warning\x80" - recording_ports.outcome = api.ProcessOutcome(0, stdout, stderr) - recording_ports.parse_error = ValueError("invalid") - request = _request(api) - - with pytest.raises(api.ParsingError) as caught: - _engine(api, recording_ports).run(request) - - assert (caught.value.stdout, caught.value.stderr) == (stdout, stderr) - assert recording_ports.parsed_payload == stdout + payload = b"x" * 5 + recording_ports.execution_result = api.ProcessOutcome(0, payload, b"err") - -def test_parser_parsing_error_is_rebuilt_with_actual_process_bytes( - recording_ports: RecordingPorts, -) -> None: - """Replace parser-owned evidence while retaining its failure as the cause.""" - api = _api() - stdout = b"actual stdout\x00\xff" - stderr = b"actual stderr\x80" - parser_error = api.ParsingError( - "invalid provider output", stdout=b"forged", stderr=b"" + result = _engine(api, recording_ports).run( + _request(api, maximum_accepted_output_bytes=4) ) - recording_ports.outcome = api.ProcessOutcome(0, stdout, stderr) - recording_ports.parse_error = parser_error - - with pytest.raises(api.ParsingError) as caught: - _engine(api, recording_ports).run(_request(api)) - assert caught.value is not parser_error - assert caught.value.message == "invalid provider output" - assert (caught.value.stdout, caught.value.stderr) == (stdout, stderr) - assert caught.value.__cause__ is parser_error + assert result.error.kind == "output_too_large_after_capture" + assert result.stdout == payload and result.error.stdout == payload + assert recording_ports.events == ["policy", "resolution", "execution"] -def test_success_preserves_all_bytes_and_returns_parsed_result( - recording_ports: RecordingPorts, -) -> None: - """Return parsed data with exact captured streams.""" +def test_arbitrary_bytes_remain_exact(recording_ports: RecordingPorts) -> None: api = _api() - payload = b"\x00\xffline1\r\nline2\n" - stderr = b"\x80warning\x00" - request = _request(api, input_bytes=payload) - recording_ports.outcome = api.ProcessOutcome(0, payload, stderr) - - result = _engine(api, recording_ports).run(request) - - assert recording_ports.invocation.input_bytes == payload - assert recording_ports.parsed_payload == payload - assert result == api.ExecutionSuccess( - parsed={"parser": "json", "size": len(payload)}, - stdout=payload, - stderr=stderr, - ) - assert recording_ports.events == ["policy", "resolution", "execution", "parsing"] + payload, stderr = b"\x00\xffline\n", b"\x80warn\r\n" + recording_ports.execution_result = api.ProcessOutcome(0, payload, stderr) + recording_ports.parsing_result = payload - -def test_empty_values_remain_structured(recording_ports: RecordingPorts) -> None: - """Preserve valid empty boundaries without coercion.""" - api = _api() - request = _request( - api, - arguments=(), - environment=(), - working_directory=None, - input_bytes=b"", + result = _engine(api, recording_ports).run( + _request(api, input_bytes=payload, output=api.OutputSpecification(parser="raw")) ) - recording_ports.outcome = api.ProcessOutcome(0, b"", b"") - _engine(api, recording_ports).run(request) + assert recording_ports.seen[2].input_bytes == payload + assert (result.parsed, result.stdout, result.stderr) == (payload, payload, stderr) - assert recording_ports.invocation == api.ExecutionSpecification( - executable="scanner", - arguments=(), - environment=(), - working_directory=None, - input_bytes=b"", - parser="json", - ) + +def test_only_core_package_is_public() -> None: + api = _api() + assert api.CliEngine + for obsolete in ("prowler.cli_engine", "prowler.cli_engine_errors"): + with pytest.raises(ModuleNotFoundError): + importlib.import_module(obsolete) diff --git a/prowler/tests/unit/chk003_cli_engine/test_adapters.py b/prowler/tests/unit/chk003_cli_engine/test_adapters.py new file mode 100644 index 00000000..6eaa91a7 --- /dev/null +++ b/prowler/tests/unit/chk003_cli_engine/test_adapters.py @@ -0,0 +1,110 @@ +"""Focused unit contract for CHK.003 production adapters.""" + +import importlib +import subprocess +from typing import Any +from unittest.mock import patch + +import pytest + + +def _api() -> Any: + try: + return importlib.import_module("prowler._core.cli_engine") + except ModuleNotFoundError: + pytest.fail("canonical prowler._core.cli_engine API is absent") + + +def _spec(api: Any, **changes: Any) -> Any: + values = { + "executable": "tool", + "arguments": ("a;b",), + "environment": (("KEY", "value"),), + "working_directory": "/work", + "input_bytes": b"\x00\xff", + "output": api.OutputSpecification(parser="raw"), + "timeout_seconds": 2.5, + "maximum_accepted_output_bytes": 100, + } + values.update(changes) + return api.ExecutionSpecification(**values) + + +def test_subprocess_executor_forces_shell_false_and_preserves_bytes() -> None: + api = _api() + specification = _spec(api) + completed = subprocess.CompletedProcess( + args=specification.argv, returncode=0, stdout=b"\xff", stderr=b"\x00" + ) + + with patch("subprocess.run", return_value=completed) as run: + outcome = api.SubprocessExecutor().execute(specification) + + run.assert_called_once_with( + ("tool", "a;b"), + input=b"\x00\xff", + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd="/work", + env={"KEY": "value"}, + timeout=2.5, + shell=False, + check=False, + ) + assert outcome == api.ProcessOutcome(0, b"\xff", b"\x00") + + +def test_subprocess_start_and_timeout_errors_are_enveloped() -> None: + api = _api() + specification = _spec(api) + executor = api.SubprocessExecutor() + with patch("subprocess.run", side_effect=OSError("missing")): + started = executor.execute(specification) + with patch( + "subprocess.run", + side_effect=subprocess.TimeoutExpired( + specification.argv, 2.5, output=b"partial\xff", stderr=b"slow\x00" + ), + ): + timed_out = executor.execute(specification) + + assert started.kind == "process_start_failed" + assert started.stdout == b"" and started.stderr == b"" + assert timed_out.kind == "timeout" + assert (timed_out.stdout, timed_out.stderr) == (b"partial\xff", b"slow\x00") + + +def test_binary_resolver_only_validates_exact_executable() -> None: + api = _api() + specification = _spec(api, executable="scanner") + resolver = api.WhichBinaryResolver() + with patch("shutil.which", return_value="/different/scanner") as which: + result = resolver.validate(specification) + assert result is None + which.assert_called_once_with("scanner", path="value" if False else None) + assert specification.executable == "scanner" + + +@pytest.mark.parametrize( + ("output", "specification", "expected"), + [ + (b"\x00\xff", ("raw", None), b"\x00\xff"), + (b"hello\n", ("text", None), "hello\n"), + (b'{"ok": true}', ("json", None), {"ok": True}), + (b"a\nb\n", ("lines", None), ["a", "b"]), + (b"id=42", ("regex", r"id=(\d+)"), "42"), + ], +) +def test_output_parsers(output: bytes, specification: tuple[str, str | None], expected: Any) -> None: + api = _api() + spec = _spec(api, output=api.OutputSpecification(*specification)) + assert api.OutputParserAdapter().parse(spec, output) == expected + + +def test_parser_failure_has_safe_context_without_claiming_process_evidence() -> None: + api = _api() + spec = _spec(api, output=api.OutputSpecification(parser="json")) + error = api.OutputParserAdapter().parse(spec, b"{bad") + assert isinstance(error, api.ParsingError) + assert error.stdout == b"" and error.stderr == b"" + assert error.context diff --git a/prowler/tests/unit/chk003_cli_engine/test_subprocess_executor.py b/prowler/tests/unit/chk003_cli_engine/test_subprocess_executor.py deleted file mode 100644 index 0ba18027..00000000 --- a/prowler/tests/unit/chk003_cli_engine/test_subprocess_executor.py +++ /dev/null @@ -1,45 +0,0 @@ -"""Unit tests for the structured subprocess adapter.""" - -import importlib -from typing import Any -from unittest.mock import patch - -import pytest - - -def _api() -> Any: - try: - return importlib.import_module("prowler.cli_engine") - except ModuleNotFoundError: - pytest.fail("CHK.003 subprocess executor is absent") - - -def test_subprocess_adapter_forces_shell_false_and_preserves_bytes() -> None: - """Invoke subprocess with structured values and no shell.""" - api = _api() - specification = api.ExecutionSpecification( - executable="tool", - arguments=("a;b",), - environment=(("KEY", "value"),), - working_directory="/work", - input_bytes=b"\x00\xff", - parser="raw", - ) - completed = __import__("subprocess").CompletedProcess( - args=specification.argv, returncode=0, stdout=b"\xff", stderr=b"\x00" - ) - - with patch("subprocess.run", return_value=completed) as run: - outcome = api.SubprocessExecutor().execute(specification) - - run.assert_called_once_with( - ("tool", "a;b"), - input=b"\x00\xff", - stdout=-1, - stderr=-1, - cwd="/work", - env={"KEY": "value"}, - shell=False, - check=False, - ) - assert outcome == api.ProcessOutcome(0, b"\xff", b"\x00") From 4f97907c2ef8db06a00823750b0169ee30f7027f Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:09:22 +0200 Subject: [PATCH 06/13] feat(prowler): port safe core CLI engine (#422) --- prowler/prowler/_core/__init__.py | 1 + prowler/prowler/_core/cli_engine/__init__.py | 44 +++++ .../_core/cli_engine/adapters/__init__.py | 7 + .../cli_engine/adapters/binary_resolver.py | 19 ++ .../cli_engine/adapters/output_parser.py | 53 ++++++ .../adapters/subprocess_executor.py | 42 +++++ prowler/prowler/_core/cli_engine/contracts.py | 86 +++++++++ prowler/prowler/_core/cli_engine/engine.py | 90 +++++++++ prowler/prowler/_core/cli_engine/errors.py | 47 +++++ prowler/prowler/_core/cli_engine/factory.py | 27 +++ prowler/prowler/_core/cli_engine/policy.py | 28 +++ prowler/prowler/_core/cli_engine/ports.py | 36 ++++ prowler/prowler/cli_engine.py | 174 ------------------ prowler/prowler/cli_engine_errors.py | 34 ---- .../behaviour/chk003_cli_engine/conftest.py | 12 +- .../test_chk003_cli_engine_bdd.py | 32 +++- .../unit/chk003_cli_engine/test_adapters.py | 18 +- 17 files changed, 522 insertions(+), 228 deletions(-) create mode 100644 prowler/prowler/_core/__init__.py create mode 100644 prowler/prowler/_core/cli_engine/__init__.py create mode 100644 prowler/prowler/_core/cli_engine/adapters/__init__.py create mode 100644 prowler/prowler/_core/cli_engine/adapters/binary_resolver.py create mode 100644 prowler/prowler/_core/cli_engine/adapters/output_parser.py create mode 100644 prowler/prowler/_core/cli_engine/adapters/subprocess_executor.py create mode 100644 prowler/prowler/_core/cli_engine/contracts.py create mode 100644 prowler/prowler/_core/cli_engine/engine.py create mode 100644 prowler/prowler/_core/cli_engine/errors.py create mode 100644 prowler/prowler/_core/cli_engine/factory.py create mode 100644 prowler/prowler/_core/cli_engine/policy.py create mode 100644 prowler/prowler/_core/cli_engine/ports.py delete mode 100644 prowler/prowler/cli_engine.py delete mode 100644 prowler/prowler/cli_engine_errors.py diff --git a/prowler/prowler/_core/__init__.py b/prowler/prowler/_core/__init__.py new file mode 100644 index 00000000..f2763b4e --- /dev/null +++ b/prowler/prowler/_core/__init__.py @@ -0,0 +1 @@ +"""Shared Prowler injector infrastructure.""" diff --git a/prowler/prowler/_core/cli_engine/__init__.py b/prowler/prowler/_core/cli_engine/__init__.py new file mode 100644 index 00000000..99eba540 --- /dev/null +++ b/prowler/prowler/_core/cli_engine/__init__.py @@ -0,0 +1,44 @@ +"""Canonical public API for safe local CLI execution.""" + +from .adapters import OutputParserAdapter, SubprocessExecutor, WhichBinaryResolver +from .contracts import ( + CommandResult, + ExecutionSpecification, + OutputSpecification, + ProcessOutcome, + ValidatedCommandRequest, +) +from .engine import CliEngine +from .errors import ( + CliEngineError, + ExecutionError, + ParsingError, + PolicyError, + ResolutionError, +) +from .factory import CliEngineFactory +from .policy import ExecutionPolicy +from .ports import BinaryResolverPort, ExecutorPort, OutputParserPort, PolicyPort + +__all__ = [ + "BinaryResolverPort", + "CliEngine", + "CliEngineError", + "CliEngineFactory", + "CommandResult", + "ExecutionError", + "ExecutionPolicy", + "ExecutionSpecification", + "ExecutorPort", + "OutputParserAdapter", + "OutputParserPort", + "OutputSpecification", + "ParsingError", + "PolicyError", + "PolicyPort", + "ProcessOutcome", + "ResolutionError", + "SubprocessExecutor", + "ValidatedCommandRequest", + "WhichBinaryResolver", +] diff --git a/prowler/prowler/_core/cli_engine/adapters/__init__.py b/prowler/prowler/_core/cli_engine/adapters/__init__.py new file mode 100644 index 00000000..c267aa83 --- /dev/null +++ b/prowler/prowler/_core/cli_engine/adapters/__init__.py @@ -0,0 +1,7 @@ +"""Safe production adapters.""" + +from .binary_resolver import WhichBinaryResolver +from .output_parser import OutputParserAdapter +from .subprocess_executor import SubprocessExecutor + +__all__ = ["OutputParserAdapter", "SubprocessExecutor", "WhichBinaryResolver"] diff --git a/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py b/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py new file mode 100644 index 00000000..7e873cba --- /dev/null +++ b/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py @@ -0,0 +1,19 @@ +"""Executable availability validation adapter.""" + +import shutil + +from ..contracts import ExecutionSpecification +from ..errors import ResolutionError + + +class WhichBinaryResolver: + """Validate the exact executable without replacing it.""" + + def validate(self, specification: ExecutionSpecification) -> ResolutionError | None: + """Check PATH from the exact environment when supplied.""" + path = dict(specification.environment).get("PATH") + if shutil.which(specification.executable, path=path) is None: + return ResolutionError( + f"executable is unavailable: {specification.executable}" + ) + return None diff --git a/prowler/prowler/_core/cli_engine/adapters/output_parser.py b/prowler/prowler/_core/cli_engine/adapters/output_parser.py new file mode 100644 index 00000000..3f14cf9e --- /dev/null +++ b/prowler/prowler/_core/cli_engine/adapters/output_parser.py @@ -0,0 +1,53 @@ +"""Output parsing adapters.""" + +import json +import re +from typing import Any + +from ..contracts import ExecutionSpecification +from ..errors import ParsingError + + +class OutputParserAdapter: + """Parse exact stdout according to the immutable specification.""" + + def parse( + self, specification: ExecutionSpecification, payload: bytes + ) -> Any | ParsingError: + """Return parsed output or safe parser context without process evidence.""" + parser = specification.output.parser.lower() + if parser == "raw": + return payload + try: + text = payload.decode("utf-8") + if parser == "text": + return text + if parser == "json": + return json.loads(text) + if parser == "lines": + return text.splitlines() + if parser == "regex": + if specification.output.pattern is None: + raise ValueError("regex parser requires a pattern") + match = re.search(specification.output.pattern, text) + if match is None: + raise ValueError("output did not match the regular expression") + if match.groupdict(): + return match.groupdict() + if len(match.groups()) == 1: + return match.group(1) + return match.groups() or match.group(0) + raise ValueError( + f"unsupported output parser: {specification.output.parser}" + ) + except ( + UnicodeDecodeError, + json.JSONDecodeError, + re.error, + ValueError, + ) as error: + return ParsingError( + "unable to parse process output", + context=(("parser", specification.output.parser),), + cause=f"{type(error).__name__}: {error}", + ) diff --git a/prowler/prowler/_core/cli_engine/adapters/subprocess_executor.py b/prowler/prowler/_core/cli_engine/adapters/subprocess_executor.py new file mode 100644 index 00000000..e170fdc4 --- /dev/null +++ b/prowler/prowler/_core/cli_engine/adapters/subprocess_executor.py @@ -0,0 +1,42 @@ +"""Safe subprocess execution adapter.""" + +import subprocess + +from ..contracts import ExecutionSpecification, ProcessOutcome +from ..errors import ExecutionError + + +class SubprocessExecutor: + """Execute structured argv with shell disabled.""" + + def execute( + self, specification: ExecutionSpecification + ) -> ProcessOutcome | ExecutionError: + """Capture exact bytes and envelope startup and timeout failures.""" + try: + completed = subprocess.run( # noqa: S603 - policy-approved structured argv + specification.argv, + input=specification.input_bytes, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=specification.working_directory, + env=dict(specification.environment), + timeout=specification.timeout_seconds, + shell=False, + check=False, + ) + except subprocess.TimeoutExpired as error: + return ExecutionError( + "process timed out", + kind="timeout", + stdout=error.output or b"", + stderr=error.stderr or b"", + cause=type(error).__name__, + ) + except OSError as error: + return ExecutionError( + "process could not be started", + kind="process_start_failed", + cause=f"{type(error).__name__}: {error}", + ) + return ProcessOutcome(completed.returncode, completed.stdout, completed.stderr) diff --git a/prowler/prowler/_core/cli_engine/contracts.py b/prowler/prowler/_core/cli_engine/contracts.py new file mode 100644 index 00000000..7e946b3e --- /dev/null +++ b/prowler/prowler/_core/cli_engine/contracts.py @@ -0,0 +1,86 @@ +"""Immutable contracts for structured local process execution.""" + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class OutputSpecification: + """Select one parser and its optional regular expression.""" + + parser: str = "raw" + pattern: str | None = None + + +@dataclass(frozen=True) +class ValidatedCommandRequest: + """Validated caller input copied into an execution specification.""" + + executable: str + arguments: Sequence[str] + environment: Mapping[str, str] | Sequence[tuple[str, str]] + working_directory: str | None + input_bytes: bytes + output: OutputSpecification + timeout_seconds: float + maximum_accepted_output_bytes: int + + +@dataclass(frozen=True) +class ExecutionSpecification: + """Deeply immutable policy and execution unit.""" + + executable: str + arguments: tuple[str, ...] + environment: tuple[tuple[str, str], ...] + working_directory: str | None + input_bytes: bytes + output: OutputSpecification + timeout_seconds: float + maximum_accepted_output_bytes: int + + @classmethod + def from_request(cls, request: ValidatedCommandRequest) -> "ExecutionSpecification": + """Defensively copy nested request values.""" + environment = ( + tuple(request.environment.items()) + if isinstance(request.environment, Mapping) + else tuple(request.environment) + ) + return cls( + executable=request.executable, + arguments=tuple(request.arguments), + environment=environment, + working_directory=request.working_directory, + input_bytes=request.input_bytes, + output=request.output, + timeout_seconds=request.timeout_seconds, + maximum_accepted_output_bytes=request.maximum_accepted_output_bytes, + ) + + @property + def argv(self) -> tuple[str, ...]: + """Return structured argv without shell interpolation.""" + return (self.executable, *self.arguments) + + +@dataclass(frozen=True) +class ProcessOutcome: + """Raw process completion values.""" + + return_code: int + stdout: bytes + stderr: bytes + + +@dataclass(frozen=True) +class CommandResult: + """Result envelope for success and expected failures.""" + + specification: ExecutionSpecification + stdout: bytes = b"" + stderr: bytes = b"" + return_code: int | None = None + parsed: Any | None = None + error: Any | None = None diff --git a/prowler/prowler/_core/cli_engine/engine.py b/prowler/prowler/_core/cli_engine/engine.py new file mode 100644 index 00000000..9085f119 --- /dev/null +++ b/prowler/prowler/_core/cli_engine/engine.py @@ -0,0 +1,90 @@ +"""Ordered CLI-engine orchestration.""" + +from dataclasses import replace + +from .contracts import ( + CommandResult, + ExecutionSpecification, + ValidatedCommandRequest, +) +from .errors import ExecutionError, ParsingError +from .ports import BinaryResolverPort, ExecutorPort, OutputParserPort, PolicyPort + + +class CliEngine: + """Run policy, resolution validation, execution, then parsing.""" + + def __init__( + self, + *, + policy: PolicyPort, + resolver: BinaryResolverPort, + executor: ExecutorPort, + parser: OutputParserPort, + ) -> None: + self._policy = policy + self._resolver = resolver + self._executor = executor + self._parser = parser + + def run(self, request: ValidatedCommandRequest) -> CommandResult: + """Return all expected outcomes in one envelope.""" + specification = ExecutionSpecification.from_request(request) + policy_error = self._policy.check(specification) + if policy_error is not None: + return CommandResult(specification, error=policy_error) + resolution_error = self._resolver.validate(specification) + if resolution_error is not None: + return CommandResult(specification, error=resolution_error) + execution = self._executor.execute(specification) + if isinstance(execution, ExecutionError): + return CommandResult( + specification, + stdout=execution.stdout, + stderr=execution.stderr, + return_code=execution.return_code, + error=execution, + ) + result = CommandResult( + specification, + stdout=execution.stdout, + stderr=execution.stderr, + return_code=execution.return_code, + ) + if execution.return_code != 0: + return replace( + result, + error=ExecutionError( + "process returned an unsuccessful outcome", + kind="unsuccessful_process", + stdout=execution.stdout, + stderr=execution.stderr, + return_code=execution.return_code, + ), + ) + if ( + max(len(execution.stdout), len(execution.stderr)) + > specification.maximum_accepted_output_bytes + ): + return replace( + result, + error=ExecutionError( + "captured process output exceeds the accepted size", + kind="output_too_large_after_capture", + stdout=execution.stdout, + stderr=execution.stderr, + return_code=execution.return_code, + ), + ) + parsed = self._parser.parse(specification, execution.stdout) + if isinstance(parsed, ParsingError): + owned_error = ParsingError( + parsed.message, + kind=parsed.kind, + stdout=execution.stdout, + stderr=execution.stderr, + context=parsed.context, + cause=parsed.cause, + ) + return replace(result, error=owned_error) + return replace(result, parsed=parsed) diff --git a/prowler/prowler/_core/cli_engine/errors.py b/prowler/prowler/_core/cli_engine/errors.py new file mode 100644 index 00000000..ba70b693 --- /dev/null +++ b/prowler/prowler/_core/cli_engine/errors.py @@ -0,0 +1,47 @@ +"""Distinct expected CLI-engine failures carried by CommandResult.""" + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class CliEngineError: + """Base data contract for expected failures.""" + + message: str + kind: str = "cli_engine_error" + + +@dataclass(frozen=True) +class PolicyError(CliEngineError): + """Policy rejected the exact immutable specification.""" + + kind: str = "policy_rejected" + + +@dataclass(frozen=True) +class ResolutionError(CliEngineError): + """The exact executable could not be validated.""" + + kind: str = "resolution_failed" + + +@dataclass(frozen=True) +class ExecutionError(CliEngineError): + """Process startup or completion failed with exact available bytes.""" + + kind: str = "execution_failed" + stdout: bytes = b"" + stderr: bytes = b"" + return_code: int | None = None + cause: str | None = None + + +@dataclass(frozen=True) +class ParsingError(CliEngineError): + """Parsing failed; engine-owned fields carry process evidence.""" + + kind: str = "parsing_failed" + stdout: bytes = b"" + stderr: bytes = b"" + context: tuple[tuple[str, str], ...] = () + cause: str | None = None diff --git a/prowler/prowler/_core/cli_engine/factory.py b/prowler/prowler/_core/cli_engine/factory.py new file mode 100644 index 00000000..c6a80ad3 --- /dev/null +++ b/prowler/prowler/_core/cli_engine/factory.py @@ -0,0 +1,27 @@ +"""Composition root for production and injected CLI engines.""" + +from dataclasses import dataclass + +from .adapters import OutputParserAdapter, SubprocessExecutor, WhichBinaryResolver +from .engine import CliEngine +from .policy import ExecutionPolicy +from .ports import BinaryResolverPort, ExecutorPort, OutputParserPort, PolicyPort + + +@dataclass(frozen=True) +class CliEngineFactory: + """Construct an engine from explicit ports or safe defaults.""" + + policy: PolicyPort = ExecutionPolicy() + resolver: BinaryResolverPort = WhichBinaryResolver() + executor: ExecutorPort = SubprocessExecutor() + parser: OutputParserPort = OutputParserAdapter() + + def create(self) -> CliEngine: + """Assemble a CLI engine.""" + return CliEngine( + policy=self.policy, + resolver=self.resolver, + executor=self.executor, + parser=self.parser, + ) diff --git a/prowler/prowler/_core/cli_engine/policy.py b/prowler/prowler/_core/cli_engine/policy.py new file mode 100644 index 00000000..78556a7e --- /dev/null +++ b/prowler/prowler/_core/cli_engine/policy.py @@ -0,0 +1,28 @@ +"""Default execution policy.""" + +from collections.abc import Callable +from dataclasses import dataclass + +from .contracts import ExecutionSpecification +from .errors import PolicyError + + +@dataclass(frozen=True) +class ExecutionPolicy: + """Authorize specifications with an optional predicate.""" + + allow: Callable[[ExecutionSpecification], bool] | None = None + + def check(self, specification: ExecutionSpecification) -> PolicyError | None: + """Return a policy result without raising expected failures.""" + if self.allow is None: + return None + try: + permitted = self.allow(specification) + except Exception: # predicates are untrusted policy extensions + return PolicyError( + "policy evaluation failed", kind="policy_evaluation_failed" + ) + if not permitted: + return PolicyError("execution specification rejected") + return None diff --git a/prowler/prowler/_core/cli_engine/ports.py b/prowler/prowler/_core/cli_engine/ports.py new file mode 100644 index 00000000..e7dc7ace --- /dev/null +++ b/prowler/prowler/_core/cli_engine/ports.py @@ -0,0 +1,36 @@ +"""Hexagonal ports for CLI-engine boundaries.""" + +from typing import Any, Protocol + +from .contracts import ExecutionSpecification, ProcessOutcome +from .errors import ExecutionError, ParsingError, PolicyError, ResolutionError + + +class PolicyPort(Protocol): + """Authorize an exact specification.""" + + def check(self, specification: ExecutionSpecification) -> PolicyError | None: ... + + +class BinaryResolverPort(Protocol): + """Validate availability without replacing the executable.""" + + def validate( + self, specification: ExecutionSpecification + ) -> ResolutionError | None: ... + + +class ExecutorPort(Protocol): + """Execute structured argv.""" + + def execute( + self, specification: ExecutionSpecification + ) -> ProcessOutcome | ExecutionError: ... + + +class OutputParserPort(Protocol): + """Parse stdout under the exact specification.""" + + def parse( + self, specification: ExecutionSpecification, payload: bytes + ) -> Any | ParsingError: ... diff --git a/prowler/prowler/cli_engine.py b/prowler/prowler/cli_engine.py deleted file mode 100644 index 6d5a6db3..00000000 --- a/prowler/prowler/cli_engine.py +++ /dev/null @@ -1,174 +0,0 @@ -"""Generic hexagonal engine for structured local process execution.""" - -from __future__ import annotations - -import subprocess -from collections.abc import Mapping, Sequence -from dataclasses import dataclass -from typing import Protocol - -from prowler import cli_engine_errors as errors - -ExecutionError = errors.ExecutionError -ParsingError = errors.ParsingError -PolicyError = errors.PolicyError -ResolutionError = errors.ResolutionError - - -def _freeze_environment( - environment: Mapping[str, str] | Sequence[tuple[str, str]], -) -> tuple[tuple[str, str], ...]: - """Copy environment entries into a stable ordered tuple.""" - if isinstance(environment, Mapping): - return tuple(environment.items()) - return tuple(environment) - - -@dataclass(frozen=True) -class ValidatedCliRequest: - """Validated command inputs used to derive an execution specification.""" - - executable: str - arguments: Sequence[str] - environment: Mapping[str, str] | Sequence[tuple[str, str]] - working_directory: str | None - input_bytes: bytes - parser: str - - -@dataclass(frozen=True) -class ExecutionSpecification: - """Deeply immutable structured process and parser specification.""" - - executable: str - arguments: tuple[str, ...] - environment: tuple[tuple[str, str], ...] - working_directory: str | None - input_bytes: bytes - parser: str - - @classmethod - def from_request(cls, request: ValidatedCliRequest) -> ExecutionSpecification: - """Copy a validated request into immutable nested values.""" - return cls( - executable=request.executable, - arguments=tuple(request.arguments), - environment=_freeze_environment(request.environment), - working_directory=request.working_directory, - input_bytes=request.input_bytes, - parser=request.parser, - ) - - @property - def argv(self) -> tuple[str, ...]: - """Return executable and ordered arguments as structured values.""" - return (self.executable, *self.arguments) - - -@dataclass(frozen=True) -class ProcessOutcome: - """Exact process outcome returned by an executor port.""" - - return_code: int - stdout: bytes - stderr: bytes - - -@dataclass(frozen=True) -class ExecutionSuccess: - """Parsed result accompanied by exact captured process streams.""" - - parsed: object - stdout: bytes - stderr: bytes - - -class PolicyPort(Protocol): - """Authorize one immutable execution specification.""" - - def check(self, specification: ExecutionSpecification) -> None: - """Raise PolicyError when execution is not permitted.""" - - -class ResolverPort(Protocol): - """Resolve values needed by one immutable execution specification.""" - - def resolve(self, specification: ExecutionSpecification) -> None: - """Resolve required values or raise ResolutionError.""" - - -class ExecutorPort(Protocol): - """Execute one immutable specification without shell interpretation.""" - - def execute(self, specification: ExecutionSpecification) -> ProcessOutcome: - """Return exact process bytes and return code.""" - - -class ParserPort(Protocol): - """Parse exact successful stdout bytes.""" - - def parse(self, parser: str, payload: bytes) -> object: - """Return the selected parser's result.""" - - -class CliEngine: - """Orchestrate policy, resolution, execution, and parsing in order.""" - - def __init__( - self, - *, - policy: PolicyPort, - resolver: ResolverPort, - executor: ExecutorPort, - parser: ParserPort, - ) -> None: - """Bind injected ports without selecting provider-specific behavior.""" - self._policy = policy - self._resolver = resolver - self._executor = executor - self._parser = parser - - def run(self, request: ValidatedCliRequest) -> ExecutionSuccess: - """Run a validated request through each boundary exactly in order.""" - specification = ExecutionSpecification.from_request(request) - self._policy.check(specification) - self._resolver.resolve(specification) - try: - outcome = self._executor.execute(specification) - except ExecutionError: - raise - except OSError as error: - raise ExecutionError(str(error)) from error - if outcome.return_code != 0: - raise ExecutionError( - "process returned an unsuccessful outcome", - stdout=outcome.stdout, - stderr=outcome.stderr, - return_code=outcome.return_code, - ) - try: - parsed = self._parser.parse(specification.parser, outcome.stdout) - except Exception as error: - message = error.message if isinstance(error, ParsingError) else str(error) - raise ParsingError( - message, stdout=outcome.stdout, stderr=outcome.stderr - ) from error - return ExecutionSuccess(parsed, outcome.stdout, outcome.stderr) - - -class SubprocessExecutor: - """Subprocess-backed executor using structured argv and shell=False.""" - - def execute(self, specification: ExecutionSpecification) -> ProcessOutcome: - """Execute a specification and preserve all process bytes exactly.""" - completed = subprocess.run( # noqa: S603 - policy-approved structured argv - specification.argv, - input=specification.input_bytes, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - cwd=specification.working_directory, - env=dict(specification.environment), - shell=False, - check=False, - ) - return ProcessOutcome(completed.returncode, completed.stdout, completed.stderr) diff --git a/prowler/prowler/cli_engine_errors.py b/prowler/prowler/cli_engine_errors.py deleted file mode 100644 index 19e3a144..00000000 --- a/prowler/prowler/cli_engine_errors.py +++ /dev/null @@ -1,34 +0,0 @@ -"""Distinct failure types for the generic CLI engine.""" - -from dataclasses import dataclass - - -class CliEngineError(Exception): - """Base class for distinct generic CLI engine failures.""" - - -class PolicyError(CliEngineError): - """The policy boundary rejected the immutable specification.""" - - -class ResolutionError(CliEngineError): - """The resolution boundary could not resolve the specification.""" - - -@dataclass(frozen=True) -class ExecutionError(CliEngineError): - """Process failure retaining exact captured streams.""" - - message: str - stdout: bytes = b"" - stderr: bytes = b"" - return_code: int | None = None - - -@dataclass(frozen=True) -class ParsingError(CliEngineError): - """Parser failure retaining the original process streams.""" - - message: str - stdout: bytes - stderr: bytes diff --git a/prowler/tests/behaviour/chk003_cli_engine/conftest.py b/prowler/tests/behaviour/chk003_cli_engine/conftest.py index 8e7ff303..5061e4e0 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/conftest.py +++ b/prowler/tests/behaviour/chk003_cli_engine/conftest.py @@ -1,5 +1,7 @@ """Fixtures local to CHK.003 behaviour tests.""" +# ruff: noqa: D102, D103 + from dataclasses import dataclass, field from typing import Any @@ -17,22 +19,22 @@ class RecordingPorts: execution_result: Any = None parsing_result: Any = None - def check(self, specification: Any) -> Any: + def check(self, specification: Any) -> Any: # noqa: D102 self.events.append("policy") self.seen.append(specification) return self.policy_error - def validate(self, specification: Any) -> Any: + def validate(self, specification: Any) -> Any: # noqa: D102 self.events.append("resolution") self.seen.append(specification) return self.resolution_error - def execute(self, specification: Any) -> Any: + def execute(self, specification: Any) -> Any: # noqa: D102 self.events.append("execution") self.seen.append(specification) return self.execution_result - def parse(self, specification: Any, payload: bytes) -> Any: + def parse(self, specification: Any, payload: bytes) -> Any: # noqa: D102 self.events.append("parsing") self.seen.append(specification) if self.parsing_result is not None: @@ -41,5 +43,5 @@ def parse(self, specification: Any, payload: bytes) -> Any: @pytest.fixture -def recording_ports() -> RecordingPorts: +def recording_ports() -> RecordingPorts: # noqa: D103 return RecordingPorts() diff --git a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py index f3a05816..98b3bfcd 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py +++ b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py @@ -1,5 +1,7 @@ """Executable behaviour contract for CHK.003.""" +# ruff: noqa: D103 + import importlib from dataclasses import FrozenInstanceError from typing import Any @@ -35,7 +37,7 @@ def _engine(api: Any, ports: RecordingPorts) -> Any: return api.CliEngine(policy=ports, resolver=ports, executor=ports, parser=ports) -def test_specification_is_deeply_immutable() -> None: +def test_specification_is_deeply_immutable() -> None: # noqa: D103 api = _api() arguments = ["--format", "json"] environment = {"LANG": "C"} @@ -53,7 +55,9 @@ def test_specification_is_deeply_immutable() -> None: specification.executable = "other" -def test_metacharacters_are_inert_structured_arguments(recording_ports: RecordingPorts) -> None: +def test_metacharacters_are_inert_structured_arguments( # noqa: D103 + recording_ports: RecordingPorts, +) -> None: api = _api() recording_ports.execution_result = api.ProcessOutcome(0, b"ok", b"") request = _request(api, arguments=["a; rm -rf /", "$(touch nope)", "x && y"]) @@ -69,7 +73,9 @@ def test_metacharacters_are_inert_structured_arguments(recording_ports: Recordin assert not hasattr(recording_ports.seen[2], "shell") -def test_same_exact_specification_crosses_all_boundaries(recording_ports: RecordingPorts) -> None: +def test_same_exact_specification_crosses_all_boundaries( # noqa: D103 + recording_ports: RecordingPorts, +) -> None: api = _api() recording_ports.execution_result = api.ProcessOutcome(0, b"ok", b"") @@ -89,14 +95,18 @@ def test_same_exact_specification_crosses_all_boundaries(recording_ports: Record ("parsing", "ParsingError", ["policy", "resolution", "execution", "parsing"]), ], ) -def test_expected_failures_use_result_envelope( +def test_expected_failures_use_result_envelope( # noqa: D103 recording_ports: RecordingPorts, boundary: str, error_name: str, events: list[str] ) -> None: api = _api() error_type = getattr(api, error_name) error = error_type(message=f"{boundary} failed") recording_ports.execution_result = api.ProcessOutcome(0, b"output", b"warning") - setattr(recording_ports, f"{boundary}_error" if boundary != "parsing" else "parsing_result", error) + setattr( + recording_ports, + f"{boundary}_error" if boundary != "parsing" else "parsing_result", + error, + ) if boundary == "execution": recording_ports.execution_result = error @@ -106,7 +116,7 @@ def test_expected_failures_use_result_envelope( assert recording_ports.events == events -def test_unsuccessful_outcome_retains_exact_bytes_and_skips_parser( +def test_unsuccessful_outcome_retains_exact_bytes_and_skips_parser( # noqa: D103 recording_ports: RecordingPorts, ) -> None: api = _api() @@ -127,7 +137,7 @@ def test_unsuccessful_outcome_retains_exact_bytes_and_skips_parser( assert recording_ports.events == ["policy", "resolution", "execution"] -def test_engine_replaces_parser_owned_evidence_and_preserves_context( +def test_engine_replaces_parser_owned_evidence_and_preserves_context( # noqa: D103 recording_ports: RecordingPorts, ) -> None: api = _api() @@ -152,7 +162,7 @@ def test_engine_replaces_parser_owned_evidence_and_preserves_context( ) -def test_post_capture_output_size_classification_is_honest( +def test_post_capture_output_size_classification_is_honest( # noqa: D103 recording_ports: RecordingPorts, ) -> None: api = _api() @@ -168,7 +178,9 @@ def test_post_capture_output_size_classification_is_honest( assert recording_ports.events == ["policy", "resolution", "execution"] -def test_arbitrary_bytes_remain_exact(recording_ports: RecordingPorts) -> None: +def test_arbitrary_bytes_remain_exact( + recording_ports: RecordingPorts, +) -> None: # noqa: D103 api = _api() payload, stderr = b"\x00\xffline\n", b"\x80warn\r\n" recording_ports.execution_result = api.ProcessOutcome(0, payload, stderr) @@ -182,7 +194,7 @@ def test_arbitrary_bytes_remain_exact(recording_ports: RecordingPorts) -> None: assert (result.parsed, result.stdout, result.stderr) == (payload, payload, stderr) -def test_only_core_package_is_public() -> None: +def test_only_core_package_is_public() -> None: # noqa: D103 api = _api() assert api.CliEngine for obsolete in ("prowler.cli_engine", "prowler.cli_engine_errors"): diff --git a/prowler/tests/unit/chk003_cli_engine/test_adapters.py b/prowler/tests/unit/chk003_cli_engine/test_adapters.py index 6eaa91a7..72f9ea9d 100644 --- a/prowler/tests/unit/chk003_cli_engine/test_adapters.py +++ b/prowler/tests/unit/chk003_cli_engine/test_adapters.py @@ -1,5 +1,7 @@ """Focused unit contract for CHK.003 production adapters.""" +# ruff: noqa: D103 + import importlib import subprocess from typing import Any @@ -30,7 +32,9 @@ def _spec(api: Any, **changes: Any) -> Any: return api.ExecutionSpecification(**values) -def test_subprocess_executor_forces_shell_false_and_preserves_bytes() -> None: +def test_subprocess_executor_forces_shell_false_and_preserves_bytes() -> ( + None +): # noqa: D103 api = _api() specification = _spec(api) completed = subprocess.CompletedProcess( @@ -54,7 +58,7 @@ def test_subprocess_executor_forces_shell_false_and_preserves_bytes() -> None: assert outcome == api.ProcessOutcome(0, b"\xff", b"\x00") -def test_subprocess_start_and_timeout_errors_are_enveloped() -> None: +def test_subprocess_start_and_timeout_errors_are_enveloped() -> None: # noqa: D103 api = _api() specification = _spec(api) executor = api.SubprocessExecutor() @@ -74,7 +78,7 @@ def test_subprocess_start_and_timeout_errors_are_enveloped() -> None: assert (timed_out.stdout, timed_out.stderr) == (b"partial\xff", b"slow\x00") -def test_binary_resolver_only_validates_exact_executable() -> None: +def test_binary_resolver_only_validates_exact_executable() -> None: # noqa: D103 api = _api() specification = _spec(api, executable="scanner") resolver = api.WhichBinaryResolver() @@ -95,13 +99,17 @@ def test_binary_resolver_only_validates_exact_executable() -> None: (b"id=42", ("regex", r"id=(\d+)"), "42"), ], ) -def test_output_parsers(output: bytes, specification: tuple[str, str | None], expected: Any) -> None: +def test_output_parsers( # noqa: D103 + output: bytes, specification: tuple[str, str | None], expected: Any +) -> None: api = _api() spec = _spec(api, output=api.OutputSpecification(*specification)) assert api.OutputParserAdapter().parse(spec, output) == expected -def test_parser_failure_has_safe_context_without_claiming_process_evidence() -> None: +def test_parser_failure_has_safe_context_without_claiming_process_evidence() -> ( + None +): # noqa: D103 api = _api() spec = _spec(api, output=api.OutputSpecification(parser="json")) error = api.OutputParserAdapter().parse(spec, b"{bad") From dca62a79788d3d136fc279e83c79e07c7322a2e9 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:11:56 +0200 Subject: [PATCH 07/13] test(prowler): expose resolver environment mismatch (#422) --- .../chk003_cli_engine.feature | 15 +++++++ .../test_chk003_cli_engine_bdd.py | 44 +++++++++++++++++++ .../unit/chk003_cli_engine/test_adapters.py | 33 +++++++++++++- 3 files changed, 90 insertions(+), 2 deletions(-) diff --git a/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature index 30a28dce..d12a9e5c 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature +++ b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature @@ -18,6 +18,21 @@ Feature: Safe local CLI engine orchestration When resolution validates and the engine runs it Then policy, resolution, execution, and parsing observe that exact specification + Scenario Outline: Reject a relative executable without a usable specification PATH + Given a relative executable and a PATH in the specification environment + When resolution runs while the parent environment can find that executable + Then resolution fails before execution without consulting the parent environment + + Examples: + | path | + | missing | + | blank | + + Scenario: Validate an absolute executable without PATH + Given an absolute executable and no PATH in the specification environment + When resolution validates that exact executable + Then execution continues with the unchanged specification environment + Scenario Outline: Return distinct expected failures without raising Given the boundary reports an expected failure When the engine runs the command diff --git a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py index 98b3bfcd..31778b9c 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py +++ b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py @@ -5,6 +5,7 @@ import importlib from dataclasses import FrozenInstanceError from typing import Any +from unittest.mock import patch import pytest @@ -86,6 +87,49 @@ def test_same_exact_specification_crosses_all_boundaries( # noqa: D103 assert result.specification is recording_ports.seen[0] +@pytest.mark.parametrize("environment", [{"LANG": "C"}, {"PATH": ""}]) +def test_relative_executable_requires_usable_specification_path( # noqa: D103 + recording_ports: RecordingPorts, environment: dict[str, str] +) -> None: + api = _api() + engine = api.CliEngine( + policy=recording_ports, + resolver=api.WhichBinaryResolver(), + executor=recording_ports, + parser=recording_ports, + ) + + with patch("shutil.which", return_value="/parent/bin/scanner") as which: + result = engine.run(_request(api, environment=environment)) + + assert isinstance(result.error, api.ResolutionError) + assert recording_ports.events == ["policy"] + which.assert_not_called() + + +def test_absolute_executable_does_not_require_path( # noqa: D103 + recording_ports: RecordingPorts, +) -> None: + api = _api() + executable = "/opt/tools/scanner" + recording_ports.execution_result = api.ProcessOutcome(0, b"ok", b"") + engine = api.CliEngine( + policy=recording_ports, + resolver=api.WhichBinaryResolver(), + executor=recording_ports, + parser=recording_ports, + ) + + with patch("shutil.which", return_value=executable) as which: + result = engine.run( + _request(api, executable=executable, environment={"LANG": "C"}) + ) + + assert result.error is None + which.assert_called_once_with(executable, path="") + assert recording_ports.seen[1].environment == (("LANG", "C"),) + + @pytest.mark.parametrize( ("boundary", "error_name", "events"), [ diff --git a/prowler/tests/unit/chk003_cli_engine/test_adapters.py b/prowler/tests/unit/chk003_cli_engine/test_adapters.py index 72f9ea9d..f389df63 100644 --- a/prowler/tests/unit/chk003_cli_engine/test_adapters.py +++ b/prowler/tests/unit/chk003_cli_engine/test_adapters.py @@ -80,15 +80,44 @@ def test_subprocess_start_and_timeout_errors_are_enveloped() -> None: # noqa: D def test_binary_resolver_only_validates_exact_executable() -> None: # noqa: D103 api = _api() - specification = _spec(api, executable="scanner") + specification = _spec( + api, executable="scanner", environment=(("PATH", "/spec/bin"),) + ) resolver = api.WhichBinaryResolver() with patch("shutil.which", return_value="/different/scanner") as which: result = resolver.validate(specification) assert result is None - which.assert_called_once_with("scanner", path="value" if False else None) + which.assert_called_once_with("scanner", path="/spec/bin") assert specification.executable == "scanner" +@pytest.mark.parametrize("environment", [(), (("PATH", ""),)]) +def test_binary_resolver_rejects_relative_executable_without_usable_path( + environment: tuple[tuple[str, str], ...], +) -> None: # noqa: D103 + api = _api() + specification = _spec(api, executable="scanner", environment=environment) + + with patch("shutil.which", return_value="/parent/bin/scanner") as which: + result = api.WhichBinaryResolver().validate(specification) + + assert isinstance(result, api.ResolutionError) + which.assert_not_called() + + +def test_binary_resolver_validates_absolute_executable_without_path() -> ( + None +): # noqa: D103 + api = _api() + specification = _spec(api, executable="/opt/tools/scanner", environment=()) + + with patch("shutil.which", return_value="/opt/tools/scanner") as which: + result = api.WhichBinaryResolver().validate(specification) + + assert result is None + which.assert_called_once_with("/opt/tools/scanner", path="") + + @pytest.mark.parametrize( ("output", "specification", "expected"), [ From ce62d424fbd89c2109628578edeb74c5bf3f01ad Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:14:12 +0200 Subject: [PATCH 08/13] feat(prowler): bind resolution to spec environment (#422) --- .../prowler/_core/cli_engine/adapters/binary_resolver.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py b/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py index 7e873cba..030a8209 100644 --- a/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py +++ b/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py @@ -1,6 +1,7 @@ """Executable availability validation adapter.""" import shutil +from pathlib import Path from ..contracts import ExecutionSpecification from ..errors import ResolutionError @@ -10,9 +11,13 @@ class WhichBinaryResolver: """Validate the exact executable without replacing it.""" def validate(self, specification: ExecutionSpecification) -> ResolutionError | None: - """Check PATH from the exact environment when supplied.""" + """Check the executable using only the specification environment.""" path = dict(specification.environment).get("PATH") - if shutil.which(specification.executable, path=path) is None: + if not Path(specification.executable).is_absolute() and not path: + return ResolutionError( + "relative executable requires a non-blank PATH in the specification" + ) + if shutil.which(specification.executable, path=path or "") is None: return ResolutionError( f"executable is unavailable: {specification.executable}" ) From 51db3b137d26ea8522b885605293130c3ea459c6 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:22:38 +0200 Subject: [PATCH 09/13] test(prowler): reject whitespace-only resolver PATH (#422) --- .../behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py index 31778b9c..772f606d 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py +++ b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py @@ -87,7 +87,7 @@ def test_same_exact_specification_crosses_all_boundaries( # noqa: D103 assert result.specification is recording_ports.seen[0] -@pytest.mark.parametrize("environment", [{"LANG": "C"}, {"PATH": ""}]) +@pytest.mark.parametrize("environment", [{"LANG": "C"}, {"PATH": ""}, {"PATH": " \t "}]) def test_relative_executable_requires_usable_specification_path( # noqa: D103 recording_ports: RecordingPorts, environment: dict[str, str] ) -> None: From b19f1665cbacf1d517d57102e11684b5abdc1cb2 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:26:08 +0200 Subject: [PATCH 10/13] fix(prowler): reject blank resolver PATH (#422) --- prowler/prowler/_core/cli_engine/adapters/binary_resolver.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py b/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py index 030a8209..e19e4c86 100644 --- a/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py +++ b/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py @@ -13,7 +13,9 @@ class WhichBinaryResolver: def validate(self, specification: ExecutionSpecification) -> ResolutionError | None: """Check the executable using only the specification environment.""" path = dict(specification.environment).get("PATH") - if not Path(specification.executable).is_absolute() and not path: + if not Path(specification.executable).is_absolute() and ( + not path or not path.strip() + ): return ResolutionError( "relative executable requires a non-blank PATH in the specification" ) From 5003143f7a4610d6ebc00884aedbd96603e54f6c Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:31:34 +0200 Subject: [PATCH 11/13] test(prowler): preserve configured CLI executable (#422) --- prowler/README.md | 5 +++++ .../chk003_cli_engine/chk003_cli_engine.feature | 5 +++++ .../test_chk003_cli_engine_bdd.py | 16 ++++++++++++++++ 3 files changed, 26 insertions(+) diff --git a/prowler/README.md b/prowler/README.md index 64416517..72008312 100644 --- a/prowler/README.md +++ b/prowler/README.md @@ -23,6 +23,11 @@ the equivalent environment variables. Never commit real tokens. require the file to exist; executable resolution happens immediately before a future assessment execution. +CHK.003 accepts this configured absolute path at its validated command-request +boundary and preserves it in the immutable execution specification. The later +Prowler adapter must pass `str(config.prowler.executable_path)` into that request; +the generic engine does not hardcode a Prowler binary location. + ## Run ```shell diff --git a/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature index d12a9e5c..86115397 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature +++ b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature @@ -33,6 +33,11 @@ Feature: Safe local CLI engine orchestration When resolution validates that exact executable Then execution continues with the unchanged specification environment + Scenario: Preserve the configured Prowler executable in the command specification + Given Prowler configuration selects an absolute executable path + When a caller builds a validated command request from that configuration + Then the immutable execution specification uses that exact executable path + Scenario Outline: Return distinct expected failures without raising Given the boundary reports an expected failure When the engine runs the command diff --git a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py index 772f606d..a9b36ff6 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py +++ b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py @@ -9,6 +9,8 @@ import pytest +from prowler.models.configs.config_loader import ProwlerConfig + from .conftest import RecordingPorts @@ -130,6 +132,20 @@ def test_absolute_executable_does_not_require_path( # noqa: D103 assert recording_ports.seen[1].environment == (("LANG", "C"),) +def test_configured_prowler_executable_becomes_immutable_specification() -> None: + api = _api() + config = ProwlerConfig(executable_path="/opt/prowler/bin/prowler") + + specification = api.ExecutionSpecification.from_request( + _request(api, executable=str(config.executable_path)) + ) + + assert specification.executable == "/opt/prowler/bin/prowler" + assert specification.argv[0] == "/opt/prowler/bin/prowler" + with pytest.raises(FrozenInstanceError): + specification.executable = "other" + + @pytest.mark.parametrize( ("boundary", "error_name", "events"), [ From 4ebb78809ae26dc5eb11db3099866445ec803001 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:40:06 +0200 Subject: [PATCH 12/13] test(prowler): protect secret CLI environments (#422) --- .../chk003_cli_engine.feature | 21 ++++++++ .../test_chk003_cli_engine_bdd.py | 48 +++++++++++++++++++ .../unit/chk003_cli_engine/test_adapters.py | 43 +++++++++++++++++ 3 files changed, 112 insertions(+) diff --git a/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature index 86115397..136bf99a 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature +++ b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature @@ -18,6 +18,16 @@ Feature: Safe local CLI engine orchestration When resolution validates and the engine runs it Then policy, resolution, execution, and parsing observe that exact specification + Scenario: Preserve secret environment values through trusted boundaries + Given a validated command environment contains an ordinary value and a secret value + When the immutable specification crosses policy and resolver boundaries + Then the secret remains wrapped and ordinary diagnostics remain redacted + + Scenario: Reject a secret PATH without using it for resolution + Given a relative executable and a secret PATH in the specification environment + When resolution runs + Then resolution fails without unwrapping PATH or searching for the executable + Scenario Outline: Reject a relative executable without a usable specification PATH Given a relative executable and a PATH in the specification environment When resolution runs while the parent environment can find that executable @@ -77,6 +87,17 @@ Feature: Safe local CLI engine orchestration # ---- Constraints identified ---- + Scenario: Unwrap secrets only at subprocess execution + Given an immutable environment contains an ordinary value and a secret value + When the subprocess executor starts the structured command + Then a fresh exact environment containing the unwrapped secret reaches subprocess execution + And shell execution and inherited environment merging remain disabled + + Scenario: Keep environment values out of execution errors + Given process startup fails with an operating-system diagnostic containing an environment value + When the subprocess executor envelopes the failure + Then the execution error does not expose that value + Scenario: Classify oversized captured output honestly Given process output has already been captured beyond the accepted size When the engine handles the process outcome diff --git a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py index a9b36ff6..60fe36d2 100644 --- a/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py +++ b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py @@ -8,6 +8,7 @@ from unittest.mock import patch import pytest +from pydantic import SecretStr from prowler.models.configs.config_loader import ProwlerConfig @@ -89,6 +90,53 @@ def test_same_exact_specification_crosses_all_boundaries( # noqa: D103 assert result.specification is recording_ports.seen[0] +def test_secret_environment_stays_wrapped_and_redacted_across_boundaries( + recording_ports: RecordingPorts, +) -> None: + api = _api() + environment_value = "chk003-boundary-secret" + secret = SecretStr(environment_value) + recording_ports.execution_result = api.ProcessOutcome(0, b"ok", b"") + + result = _engine(api, recording_ports).run( + _request(api, environment={"LANG": "C", "TOKEN": secret}) + ) + + assert result.error is None + assert all( + dict(specification.environment)["TOKEN"] is secret + for specification in recording_ports.seen + ) + assert environment_value not in repr(result.specification) + assert environment_value not in str(result.specification) + assert environment_value not in repr(result) + assert environment_value not in str(result) + + +def test_secret_path_is_rejected_without_lookup_or_unwrapping( + recording_ports: RecordingPorts, +) -> None: + api = _api() + secret_path = SecretStr("/secret/bin") + engine = api.CliEngine( + policy=recording_ports, + resolver=api.WhichBinaryResolver(), + executor=recording_ports, + parser=recording_ports, + ) + + with patch.object(secret_path, "get_secret_value") as unwrap, patch( + "shutil.which", return_value="/secret/bin/scanner" + ) as which: + result = engine.run(_request(api, environment={"PATH": secret_path})) + + assert isinstance(result.error, api.ResolutionError) + assert "secret" not in result.error.message.lower() + assert recording_ports.events == ["policy"] + unwrap.assert_not_called() + which.assert_not_called() + + @pytest.mark.parametrize("environment", [{"LANG": "C"}, {"PATH": ""}, {"PATH": " \t "}]) def test_relative_executable_requires_usable_specification_path( # noqa: D103 recording_ports: RecordingPorts, environment: dict[str, str] diff --git a/prowler/tests/unit/chk003_cli_engine/test_adapters.py b/prowler/tests/unit/chk003_cli_engine/test_adapters.py index f389df63..854d2b32 100644 --- a/prowler/tests/unit/chk003_cli_engine/test_adapters.py +++ b/prowler/tests/unit/chk003_cli_engine/test_adapters.py @@ -8,6 +8,7 @@ from unittest.mock import patch import pytest +from pydantic import SecretStr def _api() -> Any: @@ -58,6 +59,48 @@ def test_subprocess_executor_forces_shell_false_and_preserves_bytes() -> ( assert outcome == api.ProcessOutcome(0, b"\xff", b"\x00") +def test_subprocess_executor_alone_unwraps_secret_into_fresh_exact_environment() -> ( + None +): + api = _api() + source_environment = (("LANG", "C"), ("TOKEN", SecretStr("exec-secret"))) + specification = _spec(api, environment=source_environment) + completed = subprocess.CompletedProcess( + args=specification.argv, returncode=0, stdout=b"", stderr=b"" + ) + + with patch("subprocess.run", return_value=completed) as run: + outcome = api.SubprocessExecutor().execute(specification) + + passed_environment = run.call_args.kwargs["env"] + assert passed_environment == {"LANG": "C", "TOKEN": "exec-secret"} + assert passed_environment is not source_environment + assert dict(specification.environment)["TOKEN"].get_secret_value() == "exec-secret" + assert run.call_args.kwargs["shell"] is False + assert outcome == api.ProcessOutcome(0, b"", b"") + + +def test_subprocess_start_error_does_not_expose_environment_values() -> None: + api = _api() + environment_value = "error-secret-value" + specification = _spec( + api, + environment=( + ("VISIBLE", environment_value), + ("TOKEN", SecretStr(environment_value)), + ), + ) + + with patch( + "subprocess.run", side_effect=OSError(f"failed near {environment_value}") + ): + error = api.SubprocessExecutor().execute(specification) + + assert environment_value not in repr(error) + assert environment_value not in str(error) + assert environment_value not in (error.cause or "") + + def test_subprocess_start_and_timeout_errors_are_enveloped() -> None: # noqa: D103 api = _api() specification = _spec(api) From 331c18c17c5ea0d53b2a7e29b7e9c4f7b0c7d614 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 26 Aug 2026 23:46:43 +0200 Subject: [PATCH 13/13] fix(prowler): contain secret CLI environments (#422) --- .../_core/cli_engine/adapters/binary_resolver.py | 4 ++++ .../_core/cli_engine/adapters/subprocess_executor.py | 10 ++++++++-- prowler/prowler/_core/cli_engine/contracts.py | 8 ++++++-- 3 files changed, 18 insertions(+), 4 deletions(-) diff --git a/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py b/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py index e19e4c86..36b401b8 100644 --- a/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py +++ b/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py @@ -13,6 +13,10 @@ class WhichBinaryResolver: def validate(self, specification: ExecutionSpecification) -> ResolutionError | None: """Check the executable using only the specification environment.""" path = dict(specification.environment).get("PATH") + if path is not None and not isinstance(path, str): + return ResolutionError( + "PATH must be an ordinary non-blank string when provided" + ) if not Path(specification.executable).is_absolute() and ( not path or not path.strip() ): diff --git a/prowler/prowler/_core/cli_engine/adapters/subprocess_executor.py b/prowler/prowler/_core/cli_engine/adapters/subprocess_executor.py index e170fdc4..5e963bd9 100644 --- a/prowler/prowler/_core/cli_engine/adapters/subprocess_executor.py +++ b/prowler/prowler/_core/cli_engine/adapters/subprocess_executor.py @@ -2,6 +2,8 @@ import subprocess +from pydantic import SecretStr + from ..contracts import ExecutionSpecification, ProcessOutcome from ..errors import ExecutionError @@ -13,6 +15,10 @@ def execute( self, specification: ExecutionSpecification ) -> ProcessOutcome | ExecutionError: """Capture exact bytes and envelope startup and timeout failures.""" + environment = { + name: value.get_secret_value() if isinstance(value, SecretStr) else value + for name, value in specification.environment + } try: completed = subprocess.run( # noqa: S603 - policy-approved structured argv specification.argv, @@ -20,7 +26,7 @@ def execute( stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=specification.working_directory, - env=dict(specification.environment), + env=environment, timeout=specification.timeout_seconds, shell=False, check=False, @@ -37,6 +43,6 @@ def execute( return ExecutionError( "process could not be started", kind="process_start_failed", - cause=f"{type(error).__name__}: {error}", + cause=type(error).__name__, ) return ProcessOutcome(completed.returncode, completed.stdout, completed.stderr) diff --git a/prowler/prowler/_core/cli_engine/contracts.py b/prowler/prowler/_core/cli_engine/contracts.py index 7e946b3e..c4b46d90 100644 --- a/prowler/prowler/_core/cli_engine/contracts.py +++ b/prowler/prowler/_core/cli_engine/contracts.py @@ -4,6 +4,10 @@ from dataclasses import dataclass from typing import Any +from pydantic import SecretStr + +EnvironmentValue = str | SecretStr + @dataclass(frozen=True) class OutputSpecification: @@ -19,7 +23,7 @@ class ValidatedCommandRequest: executable: str arguments: Sequence[str] - environment: Mapping[str, str] | Sequence[tuple[str, str]] + environment: Mapping[str, EnvironmentValue] | Sequence[tuple[str, EnvironmentValue]] working_directory: str | None input_bytes: bytes output: OutputSpecification @@ -33,7 +37,7 @@ class ExecutionSpecification: executable: str arguments: tuple[str, ...] - environment: tuple[tuple[str, str], ...] + environment: tuple[tuple[str, EnvironmentValue], ...] working_directory: str | None input_bytes: bytes output: OutputSpecification