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/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..36b401b8
--- /dev/null
+++ b/prowler/prowler/_core/cli_engine/adapters/binary_resolver.py
@@ -0,0 +1,30 @@
+"""Executable availability validation adapter."""
+
+import shutil
+from pathlib import Path
+
+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 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()
+        ):
+            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}"
+            )
+        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..5e963bd9
--- /dev/null
+++ b/prowler/prowler/_core/cli_engine/adapters/subprocess_executor.py
@@ -0,0 +1,48 @@
+"""Safe subprocess execution adapter."""
+
+import subprocess
+
+from pydantic import SecretStr
+
+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."""
+        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,
+                input=specification.input_bytes,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                cwd=specification.working_directory,
+                env=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=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
new file mode 100644
index 00000000..c4b46d90
--- /dev/null
+++ b/prowler/prowler/_core/cli_engine/contracts.py
@@ -0,0 +1,90 @@
+"""Immutable contracts for structured local process execution."""
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from typing import Any
+
+from pydantic import SecretStr
+
+EnvironmentValue = str | SecretStr
+
+
+@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, EnvironmentValue] | Sequence[tuple[str, EnvironmentValue]]
+    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, EnvironmentValue], ...]
+    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/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..136bf99a
--- /dev/null
+++ b/prowler/tests/behaviour/chk003_cli_engine/chk003_cli_engine.feature
@@ -0,0 +1,114 @@
+@cli-engine
+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: 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> 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: 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> boundary reports an expected failure
+    When the engine runs the command
+    Then CommandResult.error is a <error> 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 <parser>
+    When the engine runs the command
+    Then the parsed value is <value>
+
+    Examples:
+      | parser | value          |
+      | raw    | exact bytes    |
+      | text   | text           |
+      | json   | structured data|
+      | lines  | text lines     |
+      | regex  | captures       |
+
+# ---- 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
+    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: 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
new file mode 100644
index 00000000..5061e4e0
--- /dev/null
+++ b/prowler/tests/behaviour/chk003_cli_engine/conftest.py
@@ -0,0 +1,47 @@
+"""Fixtures local to CHK.003 behaviour tests."""
+
+# ruff: noqa: D102, D103
+
+from dataclasses import dataclass, field
+from typing import Any
+
+import pytest
+
+
+@dataclass
+class RecordingPorts:
+    """Deterministic ports recording identity, order, and exact payloads."""
+
+    events: list[str] = field(default_factory=list)
+    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) -> Any:  # noqa: D102
+        self.events.append("policy")
+        self.seen.append(specification)
+        return self.policy_error
+
+    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:  # noqa: D102
+        self.events.append("execution")
+        self.seen.append(specification)
+        return self.execution_result
+
+    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:
+            return self.parsing_result
+        return {"parser": specification.output.parser, "bytes": payload}
+
+
+@pytest.fixture
+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
new file mode 100644
index 00000000..60fe36d2
--- /dev/null
+++ b/prowler/tests/behaviour/chk003_cli_engine/test_chk003_cli_engine_bdd.py
@@ -0,0 +1,310 @@
+"""Executable behaviour contract for CHK.003."""
+
+# ruff: noqa: D103
+
+import importlib
+from dataclasses import FrozenInstanceError
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+from pydantic import SecretStr
+
+from prowler.models.configs.config_loader import ProwlerConfig
+
+from .conftest import RecordingPorts
+
+
+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 _request(api: Any, **changes: Any) -> Any:
+    values = {
+        "executable": "scanner",
+        "arguments": ["--format", "json"],
+        "environment": {"LANG": "C"},
+        "working_directory": "/work",
+        "input_bytes": b"input",
+        "output": api.OutputSpecification(parser="json"),
+        "timeout_seconds": 30.0,
+        "maximum_accepted_output_bytes": 4096,
+    }
+    values.update(changes)
+    return api.ValidatedCommandRequest(**values)
+
+
+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:  # noqa: D103
+    api = _api()
+    arguments = ["--format", "json"]
+    environment = {"LANG": "C"}
+    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")
+    environment["LANG"] = "changed"
+
+    assert specification.argv == ("scanner", "--format", "json")
+    assert specification.environment == (("LANG", "C"),)
+    assert specification.output == output
+    with pytest.raises(FrozenInstanceError):
+        specification.executable = "other"
+
+
+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"])
+
+    _engine(api, recording_ports).run(request)
+
+    assert recording_ports.seen[2].argv == (
+        "scanner",
+        "a; rm -rf /",
+        "$(touch nope)",
+        "x && y",
+    )
+    assert not hasattr(recording_ports.seen[2], "shell")
+
+
+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"")
+
+    result = _engine(api, recording_ports).run(_request(api))
+
+    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]
+
+
+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]
+) -> 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"),)
+
+
+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"),
+    [
+        ("policy", "PolicyError", ["policy"]),
+        ("resolution", "ResolutionError", ["policy", "resolution"]),
+        ("execution", "ExecutionError", ["policy", "resolution", "execution"]),
+        ("parsing", "ParsingError", ["policy", "resolution", "execution", "parsing"]),
+    ],
+)
+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,
+    )
+    if boundary == "execution":
+        recording_ports.execution_result = error
+
+    result = _engine(api, recording_ports).run(_request(api))
+
+    assert isinstance(result.error, error_type)
+    assert recording_ports.events == events
+
+
+def test_unsuccessful_outcome_retains_exact_bytes_and_skips_parser(  # noqa: D103
+    recording_ports: RecordingPorts,
+) -> None:
+    api = _api()
+    stdout = b"partial\x00\xff\n"
+    stderr = b"failure\x80\r\n"
+    recording_ports.execution_result = api.ProcessOutcome(17, stdout, stderr)
+
+    result = _engine(api, recording_ports).run(_request(api))
+
+    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_engine_replaces_parser_owned_evidence_and_preserves_context(  # noqa: D103
+    recording_ports: RecordingPorts,
+) -> None:
+    api = _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",
+    )
+
+    result = _engine(api, recording_ports).run(_request(api))
+
+    assert result.error == api.ParsingError(
+        message="invalid document",
+        stdout=stdout,
+        stderr=stderr,
+        context=(("line", "7"),),
+        cause="JSONDecodeError",
+    )
+
+
+def test_post_capture_output_size_classification_is_honest(  # noqa: D103
+    recording_ports: RecordingPorts,
+) -> None:
+    api = _api()
+    payload = b"x" * 5
+    recording_ports.execution_result = api.ProcessOutcome(0, payload, b"err")
+
+    result = _engine(api, recording_ports).run(
+        _request(api, maximum_accepted_output_bytes=4)
+    )
+
+    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_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)
+    recording_ports.parsing_result = payload
+
+    result = _engine(api, recording_ports).run(
+        _request(api, input_bytes=payload, output=api.OutputSpecification(parser="raw"))
+    )
+
+    assert recording_ports.seen[2].input_bytes == payload
+    assert (result.parsed, result.stdout, result.stderr) == (payload, payload, stderr)
+
+
+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"):
+        with pytest.raises(ModuleNotFoundError):
+            importlib.import_module(obsolete)
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_adapters.py b/prowler/tests/unit/chk003_cli_engine/test_adapters.py
new file mode 100644
index 00000000..854d2b32
--- /dev/null
+++ b/prowler/tests/unit/chk003_cli_engine/test_adapters.py
@@ -0,0 +1,190 @@
+"""Focused unit contract for CHK.003 production adapters."""
+
+# ruff: noqa: D103
+
+import importlib
+import subprocess
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+from pydantic import SecretStr
+
+
+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
+):  # noqa: D103
+    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_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)
+    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:  # noqa: D103
+    api = _api()
+    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="/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"),
+    [
+        (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(  # 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
+):  # noqa: D103
+    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