feat(prowler): chunk3 build the CLI engine (#422)

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

prowler/prowler/_core/__init__.py

@@ -0,0 +1 @@
+"""Shared Prowler injector infrastructure."""

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",
+]

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"]

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

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}",
+            )

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)

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

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)