diff --git a/prowler/README.md b/prowler/README.md index 72008312..b1b45387 100644 --- a/prowler/README.md +++ b/prowler/README.md @@ -51,3 +51,78 @@ host and must not contain user information, a query, a fragment, or whitespace. Paths and valid ports are allowed, including endpoints on localhost, private networks, and container services. The accepted value remains an ordinary string, and validation does not check network reachability. + +## Prowler 5.36 CLI compatibility + +CHK.004 targets the installed `prowler` distribution version 5.36.0. Its +Kubernetes parser registers `--context` for selecting a kubeconfig context; +`--kube-context` is not registered. The adapter therefore emits +`--kubeconfig-file --context `. This installed parser +evidence supersedes the stale proof-of-concept/contract spelling. + +## Contract credential-file lifecycle + +AWS and Azure credentials remain `SecretStr` environment values and are not +written to files. Prowler 5.36 requires filesystem paths for GCP service-account +JSON and Kubernetes kubeconfig input. Immediately before launching Prowler, one +contract execution therefore writes that plaintext credential to a randomly +named file inside a unique OS temporary directory. The file is closed before +the subprocess starts so native Windows can reopen it. It persists for the +Prowler command runtime and is deleted in `finally`, followed by its private +directory, whether execution returns or raises. The immutable command +specification and result may retain the now-stale temporary path, but never the +file content. + +On POSIX, the directory is mode `0700` and the file is mode `0600`. Native +Windows relies on the current user's temp-directory ACL. Python `chmod` cannot +guarantee POSIX-equivalent ACL semantics on Windows, so this injector does not +claim that Windows permissions are owner-only. Docker and Kubernetes pod +ephemeral storage can reduce exposure, but does not eliminate it. + +An abrupt interpreter crash, forced kill, host failure, or power loss can occur +before `finally` and leave plaintext residue in the OS temp location. Operators +must secure and preferably encrypt the temp volume and clean stale files under +their own retention policy. The injector deliberately performs no broad stale +cleanup that could delete unrelated files. + +Each created client is one-shot and releases its copied provider input on the +first terminal run path. Python immutable strings and copies cannot be +guaranteed to be zeroized; the upstream OpenAEV injection payload may retain +credential values until `process_message` returns. + +## Assessment output storage + +CHK.004 does not parse Prowler's console stream as OCSF. Each assessment owns a +unique controlled temporary output directory and tells Prowler to write the +single expected artifact as `findings.ocsf.json` (`--output-filename findings` +with `-M json-ocsf`). Console stdout and stderr remain bounded diagnostics; the +artifact is opened only at its exact path as a regular, non-symlink file and is +read incrementally to its separate 100 MiB limit. CHK.004 debug metadata reports +only the artifact byte size; record counting belongs to the downstream mapper. + +On Linux/POSIX, a writable directory at `/dev/shm` is preferred and labelled +`memory_tmpfs`, keeping normal output in memory-backed temporary storage. It is +selected only when a safe free-capacity probe reports at least the 100 MiB +artifact limit plus a 16 MiB safety margin for Prowler's additional temporary or +nested output. If `/dev/shm` is absent, unsuitable, undersized, cannot be probed, +or fails workspace creation, the injector makes one attempt in the portable +system temporary location labelled `filesystem_temp`, which may be disk-backed. +Windows always uses that system-temp fallback and relies on its native +temporary-directory ACL rather than making a POSIX permission claim. Owned +output directories use mode `0700` on POSIX. + +On native Windows, Python does not expose the POSIX `O_NOFOLLOW` guarantee used +to reject symlink substitution at open time. Regular-file checks before and +after open, plus device/inode identity checks, are therefore a best-effort +reparse-point defence inside the randomly named controlled directory. This +fallback does not claim atomic reparse-point exclusion on native Windows; +operators must secure the system temporary-directory ACL against untrusted +writers. + +The complete owned output tree, including any nested compliance output, is +recursively removed on every normal success or failure path. Cleanup is +idempotent and never scans or deletes sibling temporary paths. An abrupt crash, +forced kill, host failure, or power loss can still bypass controlled cleanup and +leave output residue. Operators must protect both `/dev/shm` and the portable +disk fallback according to the sensitivity of assessment findings and apply +their own stale-file policy after abnormal termination. diff --git a/prowler/prowler/_core/prowler_client/__init__.py b/prowler/prowler/_core/prowler_client/__init__.py new file mode 100644 index 00000000..39dd6f2b --- /dev/null +++ b/prowler/prowler/_core/prowler_client/__init__.py @@ -0,0 +1,47 @@ +"""Canonical synchronous Prowler client API.""" + +from .client import ( + DEFAULT_MAXIMUM_ACCEPTED_CONSOLE_BYTES, + DEFAULT_MAXIMUM_ACCEPTED_OUTPUT_BYTES, + DEFAULT_TIMEOUT_SECONDS, + ProwlerClient, + ProwlerClientConsumedError, +) +from .credentials import ( + CredentialCleanupError, + TemporaryCredentialLease, + TemporaryCredentialLeaseFactory, +) +from .factory import ProwlerClientFactory +from .output_workspace import ( + DEFAULT_MAXIMUM_ARTIFACT_BYTES, + DEFAULT_MEMORY_TMPFS_SAFETY_MARGIN_BYTES, + OUTPUT_ARTIFACT_BASENAME, + OUTPUT_ARTIFACT_FILENAME, + OutputArtifactError, + OutputWorkspaceCleanupError, + OutputWorkspacePreparationError, + TemporaryOutputWorkspace, + TemporaryOutputWorkspaceFactory, +) + +__all__ = [ + "DEFAULT_MAXIMUM_ACCEPTED_CONSOLE_BYTES", + "DEFAULT_MAXIMUM_ACCEPTED_OUTPUT_BYTES", + "DEFAULT_MAXIMUM_ARTIFACT_BYTES", + "DEFAULT_MEMORY_TMPFS_SAFETY_MARGIN_BYTES", + "DEFAULT_TIMEOUT_SECONDS", + "CredentialCleanupError", + "OUTPUT_ARTIFACT_BASENAME", + "OUTPUT_ARTIFACT_FILENAME", + "OutputArtifactError", + "OutputWorkspaceCleanupError", + "OutputWorkspacePreparationError", + "ProwlerClient", + "ProwlerClientConsumedError", + "ProwlerClientFactory", + "TemporaryCredentialLease", + "TemporaryCredentialLeaseFactory", + "TemporaryOutputWorkspace", + "TemporaryOutputWorkspaceFactory", +] diff --git a/prowler/prowler/_core/prowler_client/client.py b/prowler/prowler/_core/prowler_client/client.py new file mode 100644 index 00000000..1aa0c352 --- /dev/null +++ b/prowler/prowler/_core/prowler_client/client.py @@ -0,0 +1,236 @@ +"""Synchronous Prowler client over the safe CLI engine.""" + +import logging +from collections.abc import Sequence +from dataclasses import replace +from threading import Lock +from time import monotonic + +from prowler._core.cli_engine import ( + CommandResult, + OutputSpecification, + ValidatedCommandRequest, +) +from prowler.models.configs.config_loader import ProwlerConfig +from prowler.models.provider_inputs import ProviderInput + +from .contracts import CliEnginePort, OutputWorkspaceFactoryPort +from .credentials import CredentialCleanupError +from .output_workspace import ( + DEFAULT_MAXIMUM_ARTIFACT_BYTES, + OUTPUT_ARTIFACT_BASENAME, + OutputArtifactError, + OutputWorkspaceCleanupError, +) +from .provider_adapter import ProviderInvocationAdapter + +# A full multi-provider assessment may legitimately run for a substantial period. +DEFAULT_TIMEOUT_SECONDS = 3_600.0 +# Console output is diagnostic only; OCSF records have a separate artifact bound. +DEFAULT_MAXIMUM_ACCEPTED_CONSOLE_BYTES = 4 * 1024 * 1024 +# Compatibility alias retained for callers that imported the former console bound. +DEFAULT_MAXIMUM_ACCEPTED_OUTPUT_BYTES = DEFAULT_MAXIMUM_ACCEPTED_CONSOLE_BYTES + +_ALL_SEVERITIES = ( + "--severity", + "critical", + "high", + "medium", + "low", + "informational", +) +_OUTPUT_ARGUMENTS_PREFIX = ( + "--output-filename", + OUTPUT_ARTIFACT_BASENAME, + "-z", + "--only-logs", + "--no-color", +) +_OUTPUT_FORMAT_ARGUMENTS = ("-M", "json-ocsf") +_NARROWING_OPTIONS = frozenset( + { + "-c", + "--check", + "--checks", + "-s", + "--service", + "--services", + "--compliance", + } +) +_LOGGER = logging.getLogger(__name__) + + +def _safe_log(level: int, message: str, **metadata: object) -> None: + """Best-effort fixed logging whose failures cannot affect execution.""" + try: + _LOGGER.log(level, message, extra={"prowler_metadata": metadata}) + except BaseException: + return + + +class ProwlerClientConsumedError(RuntimeError): + """Reject reuse of a client whose provider input was already consumed.""" + + +class ProwlerClient: + """Run one frozen provider input synchronously through Prowler.""" + + def __init__( + self, + *, + config: ProwlerConfig, + provider: ProviderInput, + engine: CliEnginePort, + provider_adapter: ProviderInvocationAdapter, + output_workspace_factory: OutputWorkspaceFactoryPort, + ) -> None: + self._config = config.model_copy(deep=True) + self._provider: ProviderInput | None = provider.model_copy(deep=True) + self._engine = engine + self._provider_adapter = provider_adapter + self._output_workspace_factory = output_workspace_factory + self._consumption_lock = Lock() + + def run(self, check_filters: Sequence[str] = ()) -> CommandResult: + """Run one assessment and capture its controlled OCSF artifact.""" + with self._consumption_lock: + provider = self._provider + if provider is None: + raise ProwlerClientConsumedError( + "this Prowler client has already been consumed" + ) + self._provider = None + + invocation = None + workspace = None + result: CommandResult | None = None + primary_error: BaseException | None = None + try: + filters = tuple(check_filters) + if any(not isinstance(item, str) or not item.strip() for item in filters): + raise ValueError("check filters must be nonblank strings") + + _safe_log(logging.INFO, "Preparing Prowler output workspace") + workspace = self._output_workspace_factory.create() + _safe_log( + logging.DEBUG, + "Prowler output workspace metadata", + backend=workspace.backend, + provider=provider.provider, + check_selector=bool(filters), + ) + + invocation = self._provider_adapter.adapt(provider) + filter_arguments = ("-c", *filters) if filters else () + provider_and_selectors = (*invocation.arguments, *filter_arguments) + narrowed = any( + argument in _NARROWING_OPTIONS for argument in provider_and_selectors + ) + _safe_log( + logging.DEBUG, + "Prowler selector metadata", + selector_present=narrowed, + ) + severity_arguments = () if narrowed else _ALL_SEVERITIES + request = ValidatedCommandRequest( + executable=str(self._config.executable_path), + arguments=( + *provider_and_selectors, + *severity_arguments, + "--output-directory", + str(workspace.directory), + *_OUTPUT_ARGUMENTS_PREFIX, + *_OUTPUT_FORMAT_ARGUMENTS, + ), + environment=invocation.environment, + working_directory=None, + input_bytes=b"", + output=OutputSpecification(parser="raw"), + timeout_seconds=DEFAULT_TIMEOUT_SECONDS, + maximum_accepted_output_bytes=DEFAULT_MAXIMUM_ACCEPTED_CONSOLE_BYTES, + ) + started = monotonic() + try: + result = self._engine.run(request) + finally: + duration_ms = max(0, int((monotonic() - started) * 1000)) + _safe_log(logging.INFO, "Prowler process completed") + _safe_log( + logging.DEBUG, + "Prowler process metadata", + duration_ms=duration_ms, + return_code=(result.return_code if result is not None else None), + stdout_bytes=(len(result.stdout) if result is not None else 0), + stderr_bytes=(len(result.stderr) if result is not None else 0), + ) + del request + + if result.error is not None or result.return_code != 0: + return result + + try: + artifact = workspace.read_artifact( + maximum_bytes=DEFAULT_MAXIMUM_ARTIFACT_BYTES + ) + except OutputArtifactError as error: + _safe_log( + logging.ERROR, + "Prowler output artifact capture failed", + kind=error.kind, + ) + raise + _safe_log(logging.INFO, "Prowler output artifact captured") + _safe_log( + logging.DEBUG, + "Prowler output artifact metadata", + artifact_bytes=len(artifact), + ) + return replace(result, parsed=artifact) + except BaseException as error: + primary_error = error + raise + finally: + cleanup_failures: list[tuple[str, RuntimeError]] = [] + if invocation is not None: + for lease in invocation.credential_leases: + try: + lease.cleanup() + except BaseException: + cleanup_failures.append( + ("credential", CredentialCleanupError()) + ) + if workspace is not None: + try: + workspace.cleanup() + except BaseException: + cleanup_failures.append( + ("output_workspace", OutputWorkspaceCleanupError()) + ) + else: + _safe_log(logging.INFO, "Prowler output workspace cleaned") + + if cleanup_failures: + result_is_primary_failure = result is not None and ( + result.error is not None or result.return_code != 0 + ) + for resource, _cleanup_error in cleanup_failures: + _safe_log( + logging.WARNING, + "Secondary Prowler output cleanup failure", + resource=resource, + ) + if primary_error is not None: + for resource, _cleanup_error in cleanup_failures: + note = ( + "temporary credential cleanup also failed" + if resource == "credential" + else "temporary output workspace cleanup also failed" + ) + primary_error.add_note(note) + elif not result_is_primary_failure: + raise cleanup_failures[0][1] from None + + del provider + if invocation is not None: + del invocation diff --git a/prowler/prowler/_core/prowler_client/contracts.py b/prowler/prowler/_core/prowler_client/contracts.py new file mode 100644 index 00000000..f81abd90 --- /dev/null +++ b/prowler/prowler/_core/prowler_client/contracts.py @@ -0,0 +1,76 @@ +"""Internal ports and values for synchronous Prowler invocation.""" + +from dataclasses import dataclass +from pathlib import Path +from typing import Protocol + +from pydantic import SecretStr + +from prowler._core.cli_engine import CommandResult, ValidatedCommandRequest +from prowler._core.cli_engine.contracts import EnvironmentValue + + +class CliEnginePort(Protocol): + """Execute one validated command request.""" + + def run(self, request: ValidatedCommandRequest) -> CommandResult: + """Run the command synchronously.""" + + +class CliEngineFactoryPort(Protocol): + """Create CLI engines without executing them.""" + + def create(self) -> CliEnginePort: + """Create one engine.""" + + +class CredentialLeasePort(Protocol): + """Own the lifecycle of one temporary credential resource.""" + + @property + def path(self) -> Path: + """Return the temporary credential path.""" + + def cleanup(self) -> None: + """Idempotently remove the owned credential resources.""" + + +class CredentialLeaseFactoryPort(Protocol): + """Materialize one secret as a cross-platform temporary lease.""" + + def create(self, content: SecretStr, *, suffix: str) -> CredentialLeasePort: + """Return a newly materialized credential lease.""" + + +class OutputWorkspacePort(Protocol): + """Own one controlled Prowler output directory and artifact.""" + + @property + def directory(self) -> Path: + """Return the workspace directory supplied to Prowler.""" + + @property + def backend(self) -> str: + """Return the closed storage-backend label.""" + + def read_artifact(self, *, maximum_bytes: int) -> bytes: + """Securely read the exact bounded output artifact.""" + + def cleanup(self) -> None: + """Idempotently remove the recursively owned workspace.""" + + +class OutputWorkspaceFactoryPort(Protocol): + """Create controlled Prowler output workspaces without executing Prowler.""" + + def create(self) -> OutputWorkspacePort: + """Return one unique output workspace.""" + + +@dataclass(frozen=True) +class ProviderInvocation: + """Provider-specific arguments, environment, and credential resources.""" + + arguments: tuple[str, ...] + environment: tuple[tuple[str, EnvironmentValue], ...] + credential_leases: tuple[CredentialLeasePort, ...] = () diff --git a/prowler/prowler/_core/prowler_client/credentials.py b/prowler/prowler/_core/prowler_client/credentials.py new file mode 100644 index 00000000..ca51a0f3 --- /dev/null +++ b/prowler/prowler/_core/prowler_client/credentials.py @@ -0,0 +1,83 @@ +"""Cross-platform temporary credential leases.""" + +import os +import secrets +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from threading import Lock + +from pydantic import SecretStr + + +class CredentialCleanupError(RuntimeError): + """Report failed credential cleanup without exposing credential details.""" + + def __init__(self) -> None: + super().__init__("temporary credential cleanup failed") + + +@dataclass +class TemporaryCredentialLease: + """Own one credential path and its per-run temporary directory.""" + + path: Path + directory: Path + _cleaned: bool = field(default=False, init=False, repr=False) + _lock: Lock = field(default_factory=Lock, init=False, repr=False) + + def cleanup(self) -> None: + """Idempotently remove the credential file followed by its directory.""" + with self._lock: + if self._cleaned: + return + failed = False + try: + self.path.unlink(missing_ok=True) + except OSError: + failed = True + try: + self.directory.rmdir() + except FileNotFoundError: + pass + except OSError: + failed = True + if failed: + raise CredentialCleanupError() from None + self._cleaned = True + + +@dataclass(frozen=True) +class TemporaryCredentialLeaseFactory: + """Create closed credential files in unique per-run temporary directories.""" + + platform_name: str = os.name + temporary_root: Path | None = None + + def create(self, content: SecretStr, *, suffix: str) -> TemporaryCredentialLease: + """Materialize one credential and clean partial resources on failure.""" + directory = Path( + tempfile.mkdtemp( + prefix="openaev-prowler-credential-", + dir=self.temporary_root, + ) + ) + path = directory / f"{secrets.token_hex(16)}{suffix}" + lease = TemporaryCredentialLease(path=path, directory=directory) + try: + if self.platform_name != "nt": + os.chmod(directory, 0o700) + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + if self.platform_name == "nt": + descriptor = os.open(path, flags) + else: + descriptor = os.open(path, flags, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as credential_file: + credential_file.write(content.get_secret_value()) + except BaseException: + try: + lease.cleanup() + except CredentialCleanupError: + raise CredentialCleanupError() from None + raise + return lease diff --git a/prowler/prowler/_core/prowler_client/factory.py b/prowler/prowler/_core/prowler_client/factory.py new file mode 100644 index 00000000..bcacfd1c --- /dev/null +++ b/prowler/prowler/_core/prowler_client/factory.py @@ -0,0 +1,51 @@ +"""Composition root and quick-access API for synchronous Prowler clients.""" + +from collections.abc import Sequence +from dataclasses import dataclass + +from prowler._core.cli_engine import CliEngineFactory, CommandResult +from prowler.models.configs.config_loader import ProwlerConfig +from prowler.models.provider_inputs import ProviderInput + +from .client import ProwlerClient +from .contracts import ( + CliEngineFactoryPort, + CredentialLeaseFactoryPort, + OutputWorkspaceFactoryPort, +) +from .credentials import TemporaryCredentialLeaseFactory +from .output_workspace import TemporaryOutputWorkspaceFactory +from .provider_adapter import ProviderInvocationAdapter + + +@dataclass(frozen=True) +class ProwlerClientFactory: + """Create clients from safe defaults or explicitly injected test ports.""" + + engine_factory: CliEngineFactoryPort = CliEngineFactory() + credential_lease_factory: CredentialLeaseFactoryPort = ( + TemporaryCredentialLeaseFactory() + ) + output_workspace_factory: OutputWorkspaceFactoryPort = ( + TemporaryOutputWorkspaceFactory() + ) + + def create(self, config: ProwlerConfig, provider: ProviderInput) -> ProwlerClient: + """Create a client without executing Prowler.""" + return ProwlerClient( + config=config, + provider=provider, + engine=self.engine_factory.create(), + provider_adapter=ProviderInvocationAdapter(self.credential_lease_factory), + output_workspace_factory=self.output_workspace_factory, + ) + + def run( + self, + config: ProwlerConfig, + provider: ProviderInput, + *, + check_filters: Sequence[str] = (), + ) -> CommandResult: + """Create a client and synchronously run one assessment.""" + return self.create(config, provider).run(check_filters) diff --git a/prowler/prowler/_core/prowler_client/output_workspace.py b/prowler/prowler/_core/prowler_client/output_workspace.py new file mode 100644 index 00000000..aa719b20 --- /dev/null +++ b/prowler/prowler/_core/prowler_client/output_workspace.py @@ -0,0 +1,202 @@ +"""Controlled temporary workspaces for Prowler OCSF output artifacts.""" + +import errno +import os +import shutil +import stat +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from threading import Lock +from typing import Literal + +OUTPUT_ARTIFACT_BASENAME = "findings" +OUTPUT_ARTIFACT_FILENAME = "findings.ocsf.json" +DEFAULT_MAXIMUM_ARTIFACT_BYTES = 100 * 1024 * 1024 +# Reserve space for Prowler's nested/temporary output beyond the accepted artifact. +DEFAULT_MEMORY_TMPFS_SAFETY_MARGIN_BYTES = 16 * 1024 * 1024 +_READ_CHUNK_BYTES = 64 * 1024 + +OutputBackend = Literal["memory_tmpfs", "filesystem_temp"] +OutputArtifactErrorKind = Literal["missing", "nonregular", "unreadable", "oversized"] + +_ARTIFACT_MESSAGES: dict[OutputArtifactErrorKind, str] = { + "missing": "Prowler output artifact is missing", + "nonregular": "Prowler output artifact is not a regular file", + "unreadable": "Prowler output artifact could not be read", + "oversized": "Prowler output artifact exceeds the accepted size", +} + + +class OutputArtifactError(RuntimeError): + """Report a closed artifact failure without filesystem details.""" + + def __init__(self, kind: OutputArtifactErrorKind) -> None: + self.kind = kind + super().__init__(_ARTIFACT_MESSAGES[kind]) + + +class OutputWorkspacePreparationError(RuntimeError): + """Report output-workspace creation failure without filesystem details.""" + + def __init__(self) -> None: + super().__init__("temporary output workspace preparation failed") + + +class OutputWorkspaceCleanupError(RuntimeError): + """Report output-workspace cleanup failure without filesystem details.""" + + def __init__(self) -> None: + super().__init__("temporary output workspace cleanup failed") + + +@dataclass +class TemporaryOutputWorkspace: + """Own one unique temporary directory and its exact OCSF artifact path.""" + + directory: Path + backend: OutputBackend + _temporary_directory: tempfile.TemporaryDirectory[str] = field(repr=False) + _cleaned: bool = field(default=False, init=False, repr=False) + _lock: Lock = field(default_factory=Lock, init=False, repr=False) + + @property + def artifact_path(self) -> Path: + """Return the one accepted artifact path inside this workspace.""" + return self.directory / OUTPUT_ARTIFACT_FILENAME + + def read_artifact( + self, *, maximum_bytes: int = DEFAULT_MAXIMUM_ARTIFACT_BYTES + ) -> bytes: + """Read only the exact regular, non-symlink artifact up to its bound.""" + if maximum_bytes < 0: + raise ValueError("maximum artifact bytes must be nonnegative") + + try: + path_status = self.artifact_path.lstat() + except FileNotFoundError: + raise OutputArtifactError("missing") from None + except OSError: + raise OutputArtifactError("unreadable") from None + if stat.S_ISLNK(path_status.st_mode) or not stat.S_ISREG(path_status.st_mode): + raise OutputArtifactError("nonregular") + + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(self.artifact_path, flags) + except FileNotFoundError: + raise OutputArtifactError("missing") from None + except OSError as error: + kind: OutputArtifactErrorKind = ( + "nonregular" if error.errno == errno.ELOOP else "unreadable" + ) + raise OutputArtifactError(kind) from None + + chunks: list[bytes] = [] + total = 0 + read_failed = False + try: + try: + opened_status = os.fstat(descriptor) + except OSError: + raise OutputArtifactError("unreadable") from None + if not stat.S_ISREG(opened_status.st_mode): + raise OutputArtifactError("nonregular") + if (opened_status.st_dev, opened_status.st_ino) != ( + path_status.st_dev, + path_status.st_ino, + ): + raise OutputArtifactError("nonregular") + + while total <= maximum_bytes: + requested = min(_READ_CHUNK_BYTES, maximum_bytes + 1 - total) + try: + chunk = os.read(descriptor, requested) + except OSError: + read_failed = True + raise OutputArtifactError("unreadable") from None + if not chunk: + return b"".join(chunks) + chunks.append(chunk) + total += len(chunk) + raise OutputArtifactError("oversized") + finally: + try: + os.close(descriptor) + except OSError: + if not read_failed: + raise OutputArtifactError("unreadable") from None + + def cleanup(self) -> None: + """Idempotently remove only this workspace's recursively owned tree.""" + with self._lock: + if self._cleaned: + return + try: + self._temporary_directory.cleanup() + except BaseException: + raise OutputWorkspaceCleanupError() from None + self._cleaned = True + + +@dataclass(frozen=True) +class TemporaryOutputWorkspaceFactory: + """Create private memory-first workspaces with a portable temp fallback.""" + + platform_name: str = os.name + memory_root: Path = Path("/dev/shm") # noqa: S108 - approved Linux tmpfs root + temporary_root: Path | None = None + + def create(self) -> TemporaryOutputWorkspace: + """Create one unique controlled output workspace.""" + root, backend = self._select_root() + try: + temporary_directory = tempfile.TemporaryDirectory( + prefix="openaev-prowler-output-", dir=root + ) + except BaseException: + if backend != "memory_tmpfs": + raise OutputWorkspacePreparationError() from None + root, backend = self.temporary_root, "filesystem_temp" + try: + temporary_directory = tempfile.TemporaryDirectory( + prefix="openaev-prowler-output-", dir=root + ) + except BaseException: + raise OutputWorkspacePreparationError() from None + + directory = Path(temporary_directory.name) + try: + if self.platform_name != "nt": + os.chmod(directory, 0o700) + except BaseException: + try: + temporary_directory.cleanup() + except BaseException: # noqa: S110 - preserve the safe primary error + pass + raise OutputWorkspacePreparationError() from None + return TemporaryOutputWorkspace( + directory=directory, + backend=backend, + _temporary_directory=temporary_directory, + ) + + def _select_root(self) -> tuple[Path | None, OutputBackend]: + if self.platform_name != "nt": + try: + required_capacity = ( + DEFAULT_MAXIMUM_ARTIFACT_BYTES + + DEFAULT_MEMORY_TMPFS_SAFETY_MARGIN_BYTES + ) + memory_is_usable = ( + self.memory_root.exists() + and self.memory_root.is_dir() + and os.access(self.memory_root, os.W_OK) + and shutil.disk_usage(self.memory_root).free >= required_capacity + ) + except Exception: + memory_is_usable = False + if memory_is_usable: + return self.memory_root, "memory_tmpfs" + return self.temporary_root, "filesystem_temp" diff --git a/prowler/prowler/_core/prowler_client/provider_adapter.py b/prowler/prowler/_core/prowler_client/provider_adapter.py new file mode 100644 index 00000000..abf0c574 --- /dev/null +++ b/prowler/prowler/_core/prowler_client/provider_adapter.py @@ -0,0 +1,88 @@ +"""Translate frozen provider inputs into Prowler 5.36 CLI invocations.""" + +from pydantic import SecretStr + +from prowler._core.cli_engine.contracts import EnvironmentValue +from prowler.models.provider_inputs import ( + AwsProviderInput, + AzureProviderInput, + GcpProviderInput, + KubernetesProviderInput, + ProviderInput, +) + +from .contracts import CredentialLeaseFactoryPort, ProviderInvocation + + +class ProviderInvocationAdapter: + """Build credential-safe provider arguments and exact environments.""" + + def __init__(self, credential_leases: CredentialLeaseFactoryPort) -> None: + self._credential_leases = credential_leases + + def adapt(self, provider: ProviderInput) -> ProviderInvocation: + """Return the invocation for one validated provider input.""" + if isinstance(provider, AwsProviderInput): + environment: list[tuple[str, EnvironmentValue]] = [ + ("AWS_ACCESS_KEY_ID", SecretStr(provider.aws_access_key_id)), + ("AWS_SECRET_ACCESS_KEY", provider.aws_secret_access_key), + ] + if provider.aws_session_token is not None: + environment.append(("AWS_SESSION_TOKEN", provider.aws_session_token)) + if provider.aws_endpoint_url is not None: + environment.append(("AWS_ENDPOINT_URL", provider.aws_endpoint_url)) + return ProviderInvocation( + arguments=( + "aws", + "--region", + provider.aws_region, + ), + environment=tuple(environment), + ) + if isinstance(provider, AzureProviderInput): + return ProviderInvocation( + arguments=( + "azure", + "--sp-env-auth", + "--subscription-id", + provider.azure_subscription_id, + "--azure-region", + provider.azure_provider, + ), + environment=( + ("AZURE_TENANT_ID", SecretStr(provider.azure_tenant_id)), + ("AZURE_CLIENT_ID", SecretStr(provider.azure_client_id)), + ("AZURE_CLIENT_SECRET", provider.azure_client_secret), + ), + ) + if isinstance(provider, GcpProviderInput): + lease = self._credential_leases.create( + provider.gcp_service_account_json, suffix=".json" + ) + return ProviderInvocation( + arguments=( + "gcp", + "--credentials-file", + str(lease.path), + "--project-id", + provider.gcp_project_id, + ), + environment=(), + credential_leases=(lease,), + ) + if isinstance(provider, KubernetesProviderInput): + lease = self._credential_leases.create( + provider.kubernetes_kubeconfig, suffix=".yaml" + ) + return ProviderInvocation( + arguments=( + "kubernetes", + "--kubeconfig-file", + str(lease.path), + "--context", + provider.kubernetes_context, + ), + environment=(), + credential_leases=(lease,), + ) + raise TypeError("provider must be a supported provider input") diff --git a/prowler/tests/behaviour/chk004_prowler_client/__init__.py b/prowler/tests/behaviour/chk004_prowler_client/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/prowler/tests/behaviour/chk004_prowler_client/chk004_prowler_client.feature b/prowler/tests/behaviour/chk004_prowler_client/chk004_prowler_client.feature new file mode 100644 index 00000000..5b2fd473 --- /dev/null +++ b/prowler/tests/behaviour/chk004_prowler_client/chk004_prowler_client.feature @@ -0,0 +1,146 @@ +Feature: Synchronous Prowler CLI assessments + + Background: + Given a configured Prowler runtime and one supported provider input + + Scenario: Creating a client has no execution side effect + When the factory creates a provider client + Then no Prowler command has run + + Scenario: Running a full assessment captures OCSF from a controlled artifact + When the client runs with no check filters + Then Prowler receives the provider invocation without a check selector + And every severity is selected explicitly before the controlled output options + And OCSF is read from findings.ocsf.json rather than console output + And the process result and console streams are preserved + + Scenario: The factory quick-access run is equivalent to a created client run + When the same assessment is run through both public entry points + Then both entry points submit equivalent command requests + + Scenario: Check filters preserve order and duplicates + When the client runs checks check-z, check-a, and check-z + Then Prowler receives each check as a separate ordered argument + And the all-severity base-run override is absent + + Scenario: Provider and output options have one parser-safe order + When the client runs an assessment + Then provider options and selectors precede the controlled output directory and filename + And ignore-exit-code-3, log-only, and no-color controls are present + And json-ocsf is the final output-format argument + + Scenario Outline: Each provider uses its explicit authentication boundary + Given a provider input + When the client runs the provider assessment + Then only the required environment and arguments are submitted + + Examples: + | provider | + | AWS | + | Azure | + | GCP | + | Kubernetes | + + Scenario: A configured AWS provider endpoint is scoped to its invocation environment + Given an AWS provider input with an endpoint override and optional session token + When the client runs an AWS provider assessment + Then the endpoint is submitted as a plain AWS_ENDPOINT_URL environment value + And the optional session token remains in the exact AWS environment + And the endpoint is absent from command arguments + + Scenario: An unset AWS provider endpoint preserves the exact AWS environment + Given an AWS provider input without an endpoint override + When the client runs an AWS provider assessment + Then the exact AWS credential environment remains unchanged + And the endpoint is absent from command arguments + + Scenario: Kubernetes selects a kubeconfig context with Prowler 5.36 syntax + Given a Kubernetes provider input with a named context + When the client runs the provider assessment + Then Prowler receives the context through --context + And the stale --kube-context option is absent + + # ---- Constraints identified ---- + + Scenario: Blank check filters are rejected before execution + When the client receives a blank check filter + Then no Prowler command runs + + Scenario: An unset AWS provider endpoint ignores the ambient parent endpoint + Given AWS credentials without an optional session token + And the parent process has an ambient AWS_ENDPOINT_URL + When the client runs an AWS provider assessment without an endpoint override + Then only the exact required AWS credential environment is submitted + + Scenario: Configured executable and bounded raw execution are mandatory + When the client runs an assessment + Then the request uses the exact configured executable and raw byte parser + And stdin and working directory are empty with a four-MiB console limit + And the OCSF artifact has a distinct one-hundred-MiB limit + + Scenario: Successful execution requires one safe bounded artifact + Given Prowler reports success + When the exact output artifact is missing, nonregular, a symlink, unreadable, or oversized + Then the client fails with a typed closed artifact error + But an exact-limit regular artifact is accepted + + Scenario: Failed execution ignores partial output artifacts + When Prowler returns code 3, another nonzero code, or an engine error + Then the exact engine result is returned without reading a partial artifact + + Scenario: File-backed credentials are private and scoped to one run + Given a GCP or Kubernetes provider input + When the synchronous assessment succeeds, fails, or raises + Then the credential file exists in a unique private temporary directory only during execution + And the credential content is absent from command arguments and errors + + Scenario: Native platforms apply their available temporary-file protection + Given a GCP or Kubernetes provider input + When its credential lease is created on POSIX or Windows + Then POSIX applies directory mode 0700 and file mode 0600 + And Windows relies on the current user's temporary-directory ACL without claiming POSIX-equivalent permissions + + Scenario: Credential resources close before execution and clean deterministically + Given a file-backed provider input + When Prowler returns, rejects a result, fails to start, or times out + Then the closed credential file remains readable for the command runtime + And the file and its private directory are removed afterward + + Scenario: Credential creation cannot strand a partial lease + Given a credential file write fails + When the factory abandons the lease creation + Then its partial file and private directory are removed + + Scenario: Cleanup failure preserves a safe deterministic outcome + Given credential or output-workspace cleanup fails without exposing runtime material + When command execution otherwise returns + Then a safe cleanup error is raised + But when a process or artifact failure already exists it remains primary with only a safe secondary marker + + Scenario: Output workspaces prefer memory and clean only their owned tree + Given the runtime is POSIX with a writable /dev/shm directory + When concurrent assessments prepare unique private workspaces + Then each workspace uses the memory_tmpfs backend and POSIX mode 0700 + And every owned artifact and nested compliance directory is removed idempotently + But unrelated temporary paths are untouched + + Scenario: Output workspaces fall back portably + Given /dev/shm is missing, non-directory, or unwritable, or the runtime is Windows + When an assessment prepares its output workspace + Then the filesystem_temp backend uses the system temporary location + And Windows makes no POSIX permission claim + + Scenario: Runtime logs are safe and bounded + When preparation, process, artifact, and cleanup phases run + Then fixed phase messages and closed aggregate metadata are logged best-effort + And arguments, environments, credentials, paths, content, exceptions, and tracebacks are absent + + Scenario: A created client consumes its credential input once + Given a client retains a copied provider input before its first run + When the first run reaches any terminal path + Then the client releases its provider reference + And a second run is rejected with a safe consumed-client error + + Scenario: Operators are told the residual plaintext-file risk + When an operator reads the injector documentation + Then contract-runtime persistence and crash residue are disclosed for containers, pods, POSIX, and Windows diff --git a/prowler/tests/behaviour/chk004_prowler_client/conftest.py b/prowler/tests/behaviour/chk004_prowler_client/conftest.py new file mode 100644 index 00000000..b3796f7e --- /dev/null +++ b/prowler/tests/behaviour/chk004_prowler_client/conftest.py @@ -0,0 +1,122 @@ +"""Fixtures local to CHK.004 behaviour tests.""" + +# ruff: noqa: D102, D103 + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import pytest +from pydantic import SecretStr + +from prowler._core.cli_engine import CommandResult, ExecutionSpecification +from prowler.models.provider_inputs import ( + AwsProviderInput, + AzureProviderInput, + GcpProviderInput, + KubernetesProviderInput, +) + + +@dataclass +class RecordingEngine: + """Record requests and inspect temporary files during synchronous runs.""" + + requests: list[Any] = field(default_factory=list) + result: Any = None + raised: BaseException | None = None + inspect_paths: tuple[Path, ...] = () + observed_modes: list[int] = field(default_factory=list) + observed_directory_modes: list[int] = field(default_factory=list) + observed_contents: list[str] = field(default_factory=list) + artifact_bytes: bytes = b'{"raw":"artifact-ocsf"}\n' + console_stdout: bytes = b"\x1b[32mProwler completed\x1b[0m\n" + console_stderr: bytes = b"" + write_artifact: bool = True + create_nested_output: bool = False + observed_output_directories: list[Path] = field(default_factory=list) + + def run(self, request: Any) -> Any: + self.requests.append(request) + if "--output-directory" in request.arguments: + output_directory = Path( + request.arguments[request.arguments.index("--output-directory") + 1] + ) + self.observed_output_directories.append(output_directory) + if self.create_nested_output: + nested = output_directory / "compliance" / "nested" + nested.mkdir(parents=True) + (nested / "summary.json").write_text("fixture", encoding="utf-8") + if self.write_artifact: + (output_directory / "findings.ocsf.json").write_bytes( + self.artifact_bytes + ) + paths = list(self.inspect_paths) + for flag in ("--credentials-file", "--kubeconfig-file"): + if flag in request.arguments: + paths.append(Path(request.arguments[request.arguments.index(flag) + 1])) + for path in paths: + self.observed_modes.append(path.stat().st_mode & 0o777) + self.observed_directory_modes.append(path.parent.stat().st_mode & 0o777) + self.observed_contents.append(path.read_text(encoding="utf-8")) + if self.raised is not None: + raise self.raised + if self.result is None: + specification = ExecutionSpecification.from_request(request) + self.result = CommandResult( + specification=specification, + stdout=self.console_stdout, + stderr=self.console_stderr, + return_code=0, + parsed=self.console_stdout, + ) + return self.result + + +@dataclass +class RecordingEngineFactory: + """Return an injected recording engine.""" + + engine: RecordingEngine + create_calls: int = 0 + + def create(self) -> RecordingEngine: + self.create_calls += 1 + return self.engine + + +@pytest.fixture +def recording_engine() -> RecordingEngine: + return RecordingEngine() + + +@pytest.fixture +def provider_inputs() -> dict[str, Any]: + return { + "AWS": AwsProviderInput( + provider="aws", + aws_access_key_id="AKIA_TEST", + aws_secret_access_key=SecretStr("aws-secret"), + aws_session_token=SecretStr("aws-session"), + aws_account_id="123456789012", + aws_region="eu-west-1", + ), + "Azure": AzureProviderInput( + provider="azure", + azure_tenant_id="tenant-id", + azure_client_id="client-id", + azure_client_secret=SecretStr("azure-secret"), + azure_subscription_id="subscription-id", + azure_provider="AzureUSGovernment", + ), + "GCP": GcpProviderInput( + provider="gcp", + gcp_service_account_json=SecretStr('{"private_key":"gcp-secret"}'), + gcp_project_id="project-id", + ), + "Kubernetes": KubernetesProviderInput( + provider="kubernetes", + kubernetes_kubeconfig=SecretStr("kube-secret"), + kubernetes_context="cluster-context", + ), + } diff --git a/prowler/tests/behaviour/chk004_prowler_client/test_chk004_prowler_client_bdd.py b/prowler/tests/behaviour/chk004_prowler_client/test_chk004_prowler_client_bdd.py new file mode 100644 index 00000000..f277ff96 --- /dev/null +++ b/prowler/tests/behaviour/chk004_prowler_client/test_chk004_prowler_client_bdd.py @@ -0,0 +1,508 @@ +"""Executable behaviour contract for CHK.004.""" + +# ruff: noqa: D103 + +import importlib +import os +from pathlib import Path +from typing import Any + +import pytest +from pydantic import SecretStr + +from prowler._core.cli_engine import ( + CommandResult, + ExecutionSpecification, + OutputSpecification, +) +from prowler.models.configs.config_loader import ProwlerConfig + +from .conftest import RecordingEngine, RecordingEngineFactory + + +def _api() -> Any: + try: + return importlib.import_module("prowler._core.prowler_client") + except ModuleNotFoundError: + return importlib.import_module("prowler._core.client") + + +def _factory(engine: RecordingEngine) -> Any: + return _api().ProwlerClientFactory(engine_factory=RecordingEngineFactory(engine)) + + +def _config() -> ProwlerConfig: + return ProwlerConfig(executable_path="/opt/prowler/bin/prowler") + + +def _environment(request: Any) -> dict[str, Any]: + return dict(request.environment) + + +def _failed_result() -> CommandResult: + return CommandResult( + specification=ExecutionSpecification( + executable="/opt/prowler/bin/prowler", + arguments=(), + environment=(), + working_directory=None, + input_bytes=b"", + output=OutputSpecification(parser="raw"), + timeout_seconds=1, + maximum_accepted_output_bytes=1, + ), + return_code=1, + error="safe engine error", + ) + + +def test_create_does_not_execute( + recording_engine: RecordingEngine, provider_inputs: dict[str, Any] +) -> None: + factory = _factory(recording_engine) + + client = factory.create(_config(), provider_inputs["AWS"]) + + assert client is not None + assert recording_engine.requests == [] + + +def test_full_assessment_preserves_result_and_captures_artifact_without_check_selector( + recording_engine: RecordingEngine, provider_inputs: dict[str, Any] +) -> None: + client = _factory(recording_engine).create(_config(), provider_inputs["AWS"]) + + result = client.run() + + assert recording_engine.result is not None + assert result is not recording_engine.result + assert isinstance(result, CommandResult) + assert result.stdout == recording_engine.result.stdout + assert result.stderr == recording_engine.result.stderr + assert result.return_code == recording_engine.result.return_code + assert result.specification == recording_engine.result.specification + assert result.parsed == recording_engine.artifact_bytes + assert "-c" not in recording_engine.requests[0].arguments + + +def test_factory_run_matches_created_client_request( + recording_engine: RecordingEngine, provider_inputs: dict[str, Any] +) -> None: + factory = _factory(recording_engine) + + direct = factory.create(_config(), provider_inputs["AWS"]).run(("one", "two")) + quick = factory.run(_config(), provider_inputs["AWS"], check_filters=("one", "two")) + + assert direct.parsed == quick.parsed == recording_engine.artifact_bytes + first = recording_engine.requests[0] + second = recording_engine.requests[1] + first_arguments = list(first.arguments) + second_arguments = list(second.arguments) + first_arguments[first_arguments.index("--output-directory") + 1] = "" + second_arguments[second_arguments.index("--output-directory") + 1] = "" + assert tuple(first_arguments) == tuple(second_arguments) + assert first.environment == second.environment + + +def test_check_filters_are_separate_ordered_tokens( + recording_engine: RecordingEngine, provider_inputs: dict[str, Any] +) -> None: + client = _factory(recording_engine).create(_config(), provider_inputs["AWS"]) + + client.run(("check-z", "check-a", "check-z")) + + arguments = recording_engine.requests[0].arguments + selector_index = arguments.index("-c") + assert arguments[selector_index : selector_index + 4] == ( + "-c", + "check-z", + "check-a", + "check-z", + ) + assert selector_index < arguments.index("--output-directory") + + +@pytest.mark.parametrize( + ("provider_name", "expected_arguments", "expected_environment"), + [ + ( + "AWS", + ("aws", "--region", "eu-west-1"), + {"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN"}, + ), + ( + "Azure", + ( + "azure", + "--sp-env-auth", + "--subscription-id", + "subscription-id", + "--azure-region", + "AzureUSGovernment", + ), + {"AZURE_TENANT_ID", "AZURE_CLIENT_ID", "AZURE_CLIENT_SECRET"}, + ), + ( + "GCP", + ( + "gcp", + "--credentials-file", + "", + "--project-id", + "project-id", + ), + set(), + ), + ( + "Kubernetes", + ( + "kubernetes", + "--kubeconfig-file", + "", + "--context", + "cluster-context", + ), + set(), + ), + ], +) +def test_provider_invocation_is_explicit_and_secret_safe( + recording_engine: RecordingEngine, + provider_inputs: dict[str, Any], + provider_name: str, + expected_arguments: tuple[str, ...], + expected_environment: set[str], +) -> None: + _factory(recording_engine).run(_config(), provider_inputs[provider_name]) + + request = recording_engine.requests[0] + arguments = tuple( + ( + "" + if index in {2} and provider_name in {"GCP", "Kubernetes"} + else item + ) + for index, item in enumerate(request.arguments) + ) + assert arguments[: len(expected_arguments)] == expected_arguments + assert arguments[-2:] == ("-M", "json-ocsf") + assert set(_environment(request)) == expected_environment + assert all(isinstance(value, SecretStr) for value in _environment(request).values()) + rendered = repr(request.arguments) + assert all( + secret not in rendered + for secret in ("aws-secret", "azure-secret", "gcp-secret", "kube-secret") + ) + + +def test_configured_aws_endpoint_is_a_plain_aws_only_environment_value( + recording_engine: RecordingEngine, provider_inputs: dict[str, Any] +) -> None: + endpoint = "https://aws.internal.example:8443" + provider = provider_inputs["AWS"].model_copy(update={"aws_endpoint_url": endpoint}) + + _factory(recording_engine).run(_config(), provider) + + request = recording_engine.requests[0] + assert _environment(request) == { + "AWS_ACCESS_KEY_ID": SecretStr("AKIA_TEST"), + "AWS_SECRET_ACCESS_KEY": SecretStr("aws-secret"), + "AWS_SESSION_TOKEN": SecretStr("aws-session"), + "AWS_ENDPOINT_URL": endpoint, + } + assert isinstance(_environment(request)["AWS_ENDPOINT_URL"], str) + assert endpoint not in request.arguments + + +def test_unset_aws_endpoint_preserves_exact_aws_environment( + recording_engine: RecordingEngine, provider_inputs: dict[str, Any] +) -> None: + _factory(recording_engine).run(_config(), provider_inputs["AWS"]) + + request = recording_engine.requests[0] + assert _environment(request) == { + "AWS_ACCESS_KEY_ID": SecretStr("AKIA_TEST"), + "AWS_SECRET_ACCESS_KEY": SecretStr("aws-secret"), + "AWS_SESSION_TOKEN": SecretStr("aws-session"), + } + assert "AWS_ENDPOINT_URL" not in request.arguments + + +def test_unset_aws_endpoint_ignores_ambient_value_without_session_token( + monkeypatch: pytest.MonkeyPatch, + recording_engine: RecordingEngine, + provider_inputs: dict[str, Any], +) -> None: + monkeypatch.setenv("AWS_ENDPOINT_URL", "https://ambient.example.invalid") + provider = provider_inputs["AWS"].model_copy(update={"aws_session_token": None}) + + _factory(recording_engine).run(_config(), provider) + + request = recording_engine.requests[0] + assert _environment(request) == { + "AWS_ACCESS_KEY_ID": SecretStr("AKIA_TEST"), + "AWS_SECRET_ACCESS_KEY": SecretStr("aws-secret"), + } + assert "AWS_ENDPOINT_URL" not in request.arguments + + +def test_kubernetes_uses_prowler_536_context_flag_not_stale_alias( + recording_engine: RecordingEngine, provider_inputs: dict[str, Any] +) -> None: + _factory(recording_engine).run(_config(), provider_inputs["Kubernetes"]) + + arguments = recording_engine.requests[0].arguments + assert "--context" in arguments + assert arguments[arguments.index("--context") + 1] == "cluster-context" + assert "--kube-context" not in arguments + + +@pytest.mark.parametrize("filters", [("",), (" ",), ("ok", "\t")]) +def test_blank_filters_are_rejected_without_execution( + recording_engine: RecordingEngine, + provider_inputs: dict[str, Any], + filters: tuple[str, ...], +) -> None: + client = _factory(recording_engine).create(_config(), provider_inputs["AWS"]) + + with pytest.raises(ValueError, match="check filters must be nonblank strings"): + client.run(filters) + + assert recording_engine.requests == [] + + +def test_request_uses_exact_bounded_raw_execution_contract( + recording_engine: RecordingEngine, provider_inputs: dict[str, Any] +) -> None: + _factory(recording_engine).run(_config(), provider_inputs["AWS"]) + + request = recording_engine.requests[0] + assert request.executable == "/opt/prowler/bin/prowler" + assert request.output.parser == "raw" + assert request.input_bytes == b"" + assert request.working_directory is None + assert request.timeout_seconds == _api().DEFAULT_TIMEOUT_SECONDS + assert ( + request.maximum_accepted_output_bytes + == _api().DEFAULT_MAXIMUM_ACCEPTED_OUTPUT_BYTES + ) + + +@pytest.mark.parametrize("provider_name", ["GCP", "Kubernetes"]) +@pytest.mark.parametrize("outcome", ["success", "result_error", "exception"]) +def test_temporary_credentials_are_private_unique_and_always_removed( + recording_engine: RecordingEngine, + provider_inputs: dict[str, Any], + provider_name: str, + outcome: str, +) -> None: + if outcome == "result_error": + recording_engine.result = _failed_result() + elif outcome == "exception": + recording_engine.raised = RuntimeError("safe execution failure") + factory = _factory(recording_engine) + + try: + factory.run(_config(), provider_inputs[provider_name]) + except RuntimeError as error: + assert "secret" not in repr(error) + + request = recording_engine.requests[0] + path = Path(request.arguments[2]) + assert recording_engine.observed_modes == [0o600] + assert recording_engine.observed_directory_modes == [0o700] + assert recording_engine.observed_contents + assert not path.exists() + assert not path.parent.exists() + expected_suffix = ".json" if provider_name == "GCP" else ".yaml" + assert path.suffix == expected_suffix + assert recording_engine.observed_contents[0] not in repr(request.arguments) + + +def test_each_file_backed_run_uses_a_unique_private_directory( + recording_engine: RecordingEngine, provider_inputs: dict[str, Any] +) -> None: + factory = _factory(recording_engine) + + factory.run(_config(), provider_inputs["GCP"]) + factory.run(_config(), provider_inputs["GCP"]) + + paths = [Path(request.arguments[2]) for request in recording_engine.requests] + assert paths[0].parent != paths[1].parent + assert all(not path.parent.exists() for path in paths) + + +def test_windows_lease_uses_temp_acl_without_posix_permission_claim( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + api = _api() + chmod_calls: list[tuple[object, object]] = [] + monkeypatch.setattr( + os, + "chmod", + lambda path, mode: chmod_calls.append((path, mode)), + ) + factory = api.TemporaryCredentialLeaseFactory( + platform_name="nt", temporary_root=tmp_path + ) + + lease = factory.create(SecretStr("credential"), suffix=".json") + try: + assert lease.path.read_text(encoding="utf-8") == "credential" + assert chmod_calls == [] + finally: + lease.cleanup() + + +def test_credential_file_is_closed_before_engine_execution( + recording_engine: RecordingEngine, provider_inputs: dict[str, Any] +) -> None: + _factory(recording_engine).run(_config(), provider_inputs["Kubernetes"]) + + # Reading in RecordingEngine proves the writer released its handle before run(). + assert recording_engine.observed_contents == ["kube-secret"] + + +def test_credential_lease_cleanup_is_idempotent(tmp_path: Path) -> None: + factory_class = _api().TemporaryCredentialLeaseFactory + lease = factory_class(temporary_root=tmp_path).create( + SecretStr("credential"), suffix=".json" + ) + + lease.cleanup() + lease.cleanup() + + assert not lease.path.exists() + assert not lease.directory.exists() + + +def test_creation_write_failure_removes_partial_file_and_directory( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + factory_class = _api().TemporaryCredentialLeaseFactory + original_fdopen = os.fdopen + + class PartialWriteFailure: + def __init__(self, descriptor: int, *args: Any, **kwargs: Any) -> None: + self.wrapped = original_fdopen(descriptor, *args, **kwargs) + + def __enter__(self) -> "PartialWriteFailure": + return self + + def write(self, content: str) -> None: + self.wrapped.write(content[:1]) + self.wrapped.flush() + raise OSError("safe write failure") + + def __exit__(self, *_args: Any) -> None: + self.wrapped.close() + + monkeypatch.setattr(os, "fdopen", PartialWriteFailure) + + with pytest.raises(OSError, match="safe write failure"): + factory_class(temporary_root=tmp_path).create( + SecretStr("credential"), suffix=".json" + ) + + assert list(tmp_path.iterdir()) == [] + + +class _FailingCleanupLease: + def __init__(self, path: Path) -> None: + self.path = path + + def cleanup(self) -> None: + raise OSError("cleanup failure without path") + + +class _FailingCleanupFactory: + def __init__(self, path: Path) -> None: + self.path = path + + def create(self, _content: SecretStr, *, suffix: str) -> _FailingCleanupLease: + assert suffix in {".json", ".yaml"} + self.path.parent.mkdir(exist_ok=True) + self.path.write_text("non-secret fixture", encoding="utf-8") + return _FailingCleanupLease(self.path) + + +def test_cleanup_failure_after_success_raises_safe_cleanup_error( + recording_engine: RecordingEngine, + provider_inputs: dict[str, Any], + tmp_path: Path, +) -> None: + api = _api() + factory = api.ProwlerClientFactory( + engine_factory=RecordingEngineFactory(recording_engine), + credential_lease_factory=_FailingCleanupFactory(tmp_path / "credential.json"), + ) + + with pytest.raises(api.CredentialCleanupError) as caught: + factory.run(_config(), provider_inputs["GCP"]) + + rendered = repr(caught.value) + assert "credential.json" not in rendered + assert "gcp-secret" not in rendered + + +def test_cleanup_failure_preserves_primary_exception_with_safe_note( + recording_engine: RecordingEngine, + provider_inputs: dict[str, Any], + tmp_path: Path, +) -> None: + api = _api() + primary = RuntimeError("safe primary execution failure") + recording_engine.raised = primary + factory = api.ProwlerClientFactory( + engine_factory=RecordingEngineFactory(recording_engine), + credential_lease_factory=_FailingCleanupFactory(tmp_path / "credential.yaml"), + ) + + with pytest.raises(RuntimeError) as caught: + factory.run(_config(), provider_inputs["Kubernetes"]) + + assert caught.value is primary + assert caught.value.__notes__ == ["temporary credential cleanup also failed"] + assert "credential.yaml" not in repr(caught.value) + assert "kube-secret" not in repr(caught.value) + + +@pytest.mark.parametrize("terminal", ["success", "invalid_filter", "engine_error"]) +def test_client_releases_provider_and_rejects_second_run( + recording_engine: RecordingEngine, + provider_inputs: dict[str, Any], + terminal: str, +) -> None: + api = _api() + client = _factory(recording_engine).create(_config(), provider_inputs["AWS"]) + if terminal == "engine_error": + recording_engine.raised = RuntimeError("safe engine error") + + try: + client.run(("",) if terminal == "invalid_filter" else ()) + except (RuntimeError, ValueError): + pass + + assert client._provider is None + with pytest.raises(api.ProwlerClientConsumedError, match="already been consumed"): + client.run() + assert "aws-secret" not in repr(client) + + +def test_readme_discloses_plaintext_runtime_and_residual_risk() -> None: + readme = (Path(__file__).parents[3] / "README.md").read_text(encoding="utf-8") + required_phrases = ( + "plaintext", + "command runtime", + "deleted in `finally`", + "crash", + "power loss", + "ephemeral storage", + "Windows", + "ACL", + "encrypt", + "stale", + "zeroiz", + ) + + assert all(phrase.lower() in readme.lower() for phrase in required_phrases) diff --git a/prowler/tests/unit/chk004_prowler_client/__init__.py b/prowler/tests/unit/chk004_prowler_client/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/prowler/tests/unit/chk004_prowler_client/test_lease_lifecycle.py b/prowler/tests/unit/chk004_prowler_client/test_lease_lifecycle.py new file mode 100644 index 00000000..f1de8d92 --- /dev/null +++ b/prowler/tests/unit/chk004_prowler_client/test_lease_lifecycle.py @@ -0,0 +1,206 @@ +"""Regression tests for temporary credential lease lifetime.""" + +# ruff: noqa: D101, D102, D103 + +from dataclasses import dataclass +from pathlib import Path +from types import SimpleNamespace +from typing import Any, cast + +import pytest +from pydantic import SecretStr + +import prowler._core.prowler_client.client as client_module +from prowler._core.cli_engine import ( + CommandResult, + ExecutionSpecification, + OutputSpecification, +) +from prowler._core.prowler_client import ( + CredentialCleanupError, + ProwlerClient, + TemporaryCredentialLeaseFactory, +) +from prowler.models.configs.config_loader import ProwlerConfig +from prowler.models.provider_inputs import AwsProviderInput + + +@dataclass +class _Lease: + failure: BaseException | None = None + cleanup_calls: int = 0 + + def cleanup(self) -> None: + self.cleanup_calls += 1 + if self.failure is not None: + raise self.failure + + +@dataclass +class _Adapter: + lease: _Lease + + def adapt(self, _provider: AwsProviderInput) -> Any: + return SimpleNamespace( + arguments=("aws", "-M", "json-ocsf"), + environment={}, + credential_leases=(self.lease,), + ) + + +@dataclass +class _Engine: + result: CommandResult | None = None + failure: BaseException | None = None + + def run(self, _request: Any) -> CommandResult: + if self.failure is not None: + raise self.failure + assert self.result is not None + return self.result + + +@dataclass +class _Workspace: + directory: Path = Path("/controlled-workspace") + backend: str = "filesystem_temp" + cleanup_calls: int = 0 + + def read_artifact(self, *, maximum_bytes: int) -> bytes: + assert maximum_bytes > 0 + return b'{"artifact":"ocsf"}' + + def cleanup(self) -> None: + self.cleanup_calls += 1 + + +@dataclass(frozen=True) +class _WorkspaceFactory: + workspace: _Workspace + + def create(self) -> _Workspace: + return self.workspace + + +def _result() -> CommandResult: + return CommandResult( + specification=ExecutionSpecification( + executable="/opt/prowler/bin/prowler", + arguments=(), + environment=(), + working_directory=None, + input_bytes=b"", + output=OutputSpecification(parser="raw"), + timeout_seconds=1, + maximum_accepted_output_bytes=1, + ), + return_code=0, + ) + + +def _provider() -> AwsProviderInput: + return AwsProviderInput( + provider="aws", + aws_access_key_id="AKIA_TEST", + aws_secret_access_key=SecretStr("secret"), + aws_account_id="123456789012", + aws_region="eu-west-1", + ) + + +def _client( + lease: _Lease, engine: _Engine, workspace: _Workspace | None = None +) -> ProwlerClient: + selected_workspace = workspace or _Workspace() + return ProwlerClient( + config=ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), + provider=_provider(), + engine=engine, + provider_adapter=cast(Any, _Adapter(lease)), + output_workspace_factory=cast(Any, _WorkspaceFactory(selected_workspace)), + ) + + +def test_credential_file_is_created_owner_only(tmp_path: Path) -> None: + lease = TemporaryCredentialLeaseFactory(temporary_root=tmp_path).create( + SecretStr("credential"), suffix=".json" + ) + try: + assert lease.path.stat().st_mode & 0o777 == 0o600 + assert lease.directory.stat().st_mode & 0o777 == 0o700 + assert lease.path.read_text(encoding="utf-8") == "credential" + finally: + lease.cleanup() + + +def test_cleanup_removes_file_and_directory_idempotently(tmp_path: Path) -> None: + lease = TemporaryCredentialLeaseFactory(temporary_root=tmp_path).create( + SecretStr("credential"), suffix=".yaml" + ) + + lease.cleanup() + lease.cleanup() + + assert not lease.path.exists() + assert not lease.directory.exists() + + +def test_lease_is_released_when_engine_succeeds() -> None: + lease = _Lease() + result = _result() + + captured = _client(lease, _Engine(result=result)).run() + assert captured.parsed == b'{"artifact":"ocsf"}' + assert captured.return_code == result.return_code + assert lease.cleanup_calls == 1 + + +def test_lease_is_released_when_engine_fails() -> None: + lease = _Lease() + primary = RuntimeError("safe engine failure") + + with pytest.raises(RuntimeError) as caught: + _client(lease, _Engine(failure=primary)).run() + + assert caught.value is primary + assert lease.cleanup_calls == 1 + + +def test_lease_is_released_when_request_construction_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + lease = _Lease() + primary = RuntimeError("safe construction failure") + + def fail_construction(**_kwargs: Any) -> Any: + raise primary + + monkeypatch.setattr(client_module, "ValidatedCommandRequest", fail_construction) + workspace = _Workspace() + + with pytest.raises(RuntimeError) as caught: + _client(lease, _Engine(result=_result()), workspace).run() + + assert caught.value is primary + assert lease.cleanup_calls == 1 + assert workspace.cleanup_calls == 1 + + +def test_cleanup_failure_without_primary_error_is_safe() -> None: + lease = _Lease(failure=OSError("unsafe cleanup detail")) + + with pytest.raises(CredentialCleanupError) as caught: + _client(lease, _Engine(result=_result())).run() + + assert str(caught.value) == "temporary credential cleanup failed" + + +def test_cleanup_failure_preserves_primary_error_and_adds_safe_note() -> None: + lease = _Lease(failure=OSError("unsafe cleanup detail")) + primary = RuntimeError("safe primary failure") + + with pytest.raises(RuntimeError) as caught: + _client(lease, _Engine(failure=primary)).run() + + assert caught.value is primary + assert caught.value.__notes__ == ["temporary credential cleanup also failed"] diff --git a/prowler/tests/unit/chk004_prowler_client/test_output_capture.py b/prowler/tests/unit/chk004_prowler_client/test_output_capture.py new file mode 100644 index 00000000..d63b5c34 --- /dev/null +++ b/prowler/tests/unit/chk004_prowler_client/test_output_capture.py @@ -0,0 +1,416 @@ +"""Client-owned OCSF artifact capture tests for CHK.004.""" + +# ruff: noqa: D101, D102, D103 + +import json +import logging +from dataclasses import dataclass +from pathlib import Path +from types import SimpleNamespace +from typing import Any, cast + +import pytest +from pydantic import SecretStr + +from prowler._core.cli_engine import ( + CommandResult, + ExecutionSpecification, + OutputSpecification, +) +from prowler.models.configs.config_loader import ProwlerConfig +from prowler.models.provider_inputs import AwsProviderInput + + +def _api() -> Any: + from prowler._core import prowler_client + + return prowler_client + + +def _provider() -> AwsProviderInput: + return AwsProviderInput( + provider="aws", + aws_access_key_id="AKIA_TEST", + aws_secret_access_key=SecretStr("secret-canary"), + aws_account_id="123456789012", + aws_region="eu-west-1", + ) + + +def _result( + *, return_code: int | None = 0, error: object | None = None +) -> CommandResult: + return CommandResult( + specification=ExecutionSpecification( + executable="/opt/prowler/bin/prowler", + arguments=(), + environment=(), + working_directory=None, + input_bytes=b"", + output=OutputSpecification(parser="raw"), + timeout_seconds=1, + maximum_accepted_output_bytes=1, + ), + stdout=b"\x1b[31mconsole-not-json\x1b[0m", + stderr=b"console-stderr-canary", + return_code=return_code, + parsed=b"old-console-parser-value", + error=error, + ) + + +@dataclass +class _Engine: + result: CommandResult + artifact: bytes = b'[{"finding":"artifact"}]' + requests: list[Any] | None = None + create_nested: bool = False + failure: BaseException | None = None + + def run(self, request: Any) -> CommandResult: + if self.requests is None: + self.requests = [] + self.requests.append(request) + directory = Path( + request.arguments[request.arguments.index("--output-directory") + 1] + ) + (directory / "findings.ocsf.json").write_bytes(self.artifact) + if self.create_nested: + nested = directory / "compliance" / "provider" + nested.mkdir(parents=True) + (nested / "summary.csv").write_text("fixture", encoding="utf-8") + if self.failure is not None: + raise self.failure + return self.result + + +@dataclass(frozen=True) +class _EngineFactory: + engine: _Engine + + def create(self) -> _Engine: + return self.engine + + +@dataclass +class _CleanupReportingWorkspace: + wrapped: Any + + @property + def directory(self) -> Path: + return cast(Path, self.wrapped.directory) + + @property + def backend(self) -> str: + return cast(str, self.wrapped.backend) + + def read_artifact(self, *, maximum_bytes: int) -> bytes: + return cast(bytes, self.wrapped.read_artifact(maximum_bytes=maximum_bytes)) + + def cleanup(self) -> None: + self.wrapped.cleanup() + raise OSError("unsafe cleanup path canary") + + +@dataclass(frozen=True) +class _CleanupReportingWorkspaceFactory: + wrapped: Any + + def create(self) -> _CleanupReportingWorkspace: + return _CleanupReportingWorkspace(self.wrapped.create()) + + +def _factory(engine: _Engine, tmp_path: Path, **kwargs: Any) -> Any: + return _api().ProwlerClientFactory( + engine_factory=_EngineFactory(engine), + output_workspace_factory=_api().TemporaryOutputWorkspaceFactory( + platform_name="nt", temporary_root=tmp_path + ), + **kwargs, + ) + + +def test_success_uses_artifact_but_preserves_console_and_process_result( + tmp_path: Path, +) -> None: + original = _result() + engine = _Engine(original, create_nested=True) + + captured = _factory(engine, tmp_path).run( + ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), _provider() + ) + + assert captured is not original + assert captured.specification is original.specification + assert captured.stdout == original.stdout + assert captured.stderr == original.stderr + assert captured.return_code == original.return_code + assert captured.error is original.error + assert captured.parsed == engine.artifact + assert engine.requests is not None + assert not Path( + engine.requests[0].arguments[ + engine.requests[0].arguments.index("--output-directory") + 1 + ] + ).exists() + + +@pytest.mark.parametrize( + ("return_code", "error"), + [(3, None), (7, None), (0, "engine-error")], +) +def test_engine_failure_result_is_unchanged_and_ignores_partial_artifact( + tmp_path: Path, return_code: int, error: object | None +) -> None: + original = _result(return_code=return_code, error=error) + engine = _Engine(original, artifact=b"partial-invalid-artifact") + + captured = _factory(engine, tmp_path).run( + ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), _provider() + ) + + assert captured is original + + +def test_missing_success_artifact_raises_typed_error_and_cleans_workspace( + caplog: pytest.LogCaptureFixture, tmp_path: Path +) -> None: + caplog.set_level(logging.ERROR) + engine = _Engine(_result()) + + def no_artifact(request: Any) -> CommandResult: + if engine.requests is None: + engine.requests = [] + engine.requests.append(request) + return engine.result + + engine.run = no_artifact # type: ignore[method-assign] + with pytest.raises(_api().OutputArtifactError) as caught: + _factory(engine, tmp_path).run( + ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), _provider() + ) + assert caught.value.kind == "missing" + assert engine.requests is not None + workspace_path = Path( + engine.requests[0].arguments[ + engine.requests[0].arguments.index("--output-directory") + 1 + ] + ) + assert not workspace_path.exists() + assert [record.getMessage() for record in caplog.records] == [ + "Prowler output artifact capture failed" + ] + assert caplog.records[0].levelno == logging.ERROR + + +def test_request_has_small_console_limit_and_exact_ordered_output_controls( + tmp_path: Path, +) -> None: + engine = _Engine(_result()) + _factory(engine, tmp_path).run( + ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), _provider() + ) + assert engine.requests is not None + request = engine.requests[0] + arguments = request.arguments + output_index = arguments.index("--output-directory") + + assert request.maximum_accepted_output_bytes == 4 * 1024 * 1024 + assert arguments[:output_index] == ( + "aws", + "--region", + "eu-west-1", + "--severity", + "critical", + "high", + "medium", + "low", + "informational", + ) + assert arguments[output_index + 2 :] == ( + "--output-filename", + "findings", + "-z", + "--only-logs", + "--no-color", + "-M", + "json-ocsf", + ) + assert arguments[-2:] == ("-M", "json-ocsf") + assert "-z" in arguments + assert "--ignore-exit-code-3" not in arguments + + +@pytest.mark.parametrize( + "selector_arguments", + [("-c", "check"), ("--services", "s3"), ("--compliance", "cis_3.0_aws")], +) +def test_narrowed_runs_do_not_receive_all_severity_override( + tmp_path: Path, selector_arguments: tuple[str, str] +) -> None: + class Adapter: + def adapt(self, _provider: AwsProviderInput) -> Any: + return SimpleNamespace( + arguments=("aws", *selector_arguments), + environment=(), + credential_leases=(), + ) + + engine = _Engine(_result()) + workspace_factory = _api().TemporaryOutputWorkspaceFactory( + platform_name="nt", temporary_root=tmp_path + ) + client = _api().ProwlerClient( + config=ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), + provider=_provider(), + engine=engine, + provider_adapter=cast(Any, Adapter()), + output_workspace_factory=workspace_factory, + ) + + client.run() + assert engine.requests is not None + assert "--severity" not in engine.requests[0].arguments + + +def test_logs_are_fixed_phased_and_exclude_sensitive_canaries( + caplog: pytest.LogCaptureFixture, tmp_path: Path +) -> None: + caplog.set_level(logging.DEBUG) + engine = _Engine(_result(), artifact=b'[{"secret":"artifact-canary"}]') + + _factory(engine, tmp_path).run( + ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), _provider() + ) + + messages = [record.getMessage() for record in caplog.records] + assert { + "Preparing Prowler output workspace", + "Prowler process completed", + "Prowler output artifact captured", + "Prowler output workspace cleaned", + }.issubset(messages) + rendered = "\n".join( + f"{record.levelname} {record.getMessage()} {record.__dict__}" + for record in caplog.records + ) + assert "secret-canary" not in rendered + assert "artifact-canary" not in rendered + assert "console-stderr-canary" not in rendered + assert str(tmp_path) not in rendered + assert any(record.levelno == logging.DEBUG for record in caplog.records) + artifact_metadata = next( + cast(Any, record).prowler_metadata + for record in caplog.records + if record.getMessage() == "Prowler output artifact metadata" + ) + assert artifact_metadata == {"artifact_bytes": len(engine.artifact)} + + +def test_logging_failure_cannot_change_success( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + engine = _Engine(_result()) + monkeypatch.setattr( + "prowler._core.prowler_client.client._LOGGER.log", + lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("logger failed")), + ) + + captured = _factory(engine, tmp_path).run( + ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), _provider() + ) + assert captured.parsed == engine.artifact + + +@pytest.mark.parametrize("failure_type", [MemoryError, RecursionError]) +def test_debug_logging_never_decodes_artifact_or_changes_success( + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + failure_type: type[BaseException], +) -> None: + caplog.set_level(logging.DEBUG) + engine = _Engine(_result()) + decode_calls = 0 + + def fail_if_decoded(_payload: object, *_args: object, **_kwargs: object) -> object: + nonlocal decode_calls + decode_calls += 1 + raise failure_type("debug decoding must not run") + + monkeypatch.setattr(json, "loads", fail_if_decoded) + + captured = _factory(engine, tmp_path).run( + ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), _provider() + ) + + assert captured.parsed == engine.artifact + assert decode_calls == 0 + artifact_metadata = next( + cast(Any, record).prowler_metadata + for record in caplog.records + if record.getMessage() == "Prowler output artifact metadata" + ) + assert artifact_metadata == {"artifact_bytes": len(engine.artifact)} + + +def test_cleanup_failure_after_success_surfaces_only_safe_cleanup_error( + tmp_path: Path, +) -> None: + engine = _Engine(_result()) + normal_factory = _api().TemporaryOutputWorkspaceFactory( + platform_name="nt", temporary_root=tmp_path + ) + factory = _api().ProwlerClientFactory( + engine_factory=_EngineFactory(engine), + output_workspace_factory=_CleanupReportingWorkspaceFactory(normal_factory), + ) + + with pytest.raises(_api().OutputWorkspaceCleanupError) as caught: + factory.run( + ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), _provider() + ) + + assert str(caught.value) == "temporary output workspace cleanup failed" + assert "canary" not in repr(caught.value) + + +@pytest.mark.parametrize("primary_kind", ["exception", "result"]) +def test_cleanup_failure_never_masks_primary_and_emits_safe_warning( + caplog: pytest.LogCaptureFixture, tmp_path: Path, primary_kind: str +) -> None: + caplog.set_level(logging.WARNING) + primary = RuntimeError("safe primary failure") + result = _result(return_code=7) if primary_kind == "result" else _result() + engine = _Engine(result, failure=primary if primary_kind == "exception" else None) + normal_factory = _api().TemporaryOutputWorkspaceFactory( + platform_name="nt", temporary_root=tmp_path + ) + factory = _api().ProwlerClientFactory( + engine_factory=_EngineFactory(engine), + output_workspace_factory=_CleanupReportingWorkspaceFactory(normal_factory), + ) + + if primary_kind == "exception": + with pytest.raises(RuntimeError) as caught: + factory.run( + ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), + _provider(), + ) + assert caught.value is primary + assert caught.value.__notes__ == [ + "temporary output workspace cleanup also failed" + ] + else: + captured = factory.run( + ProwlerConfig(executable_path="/opt/prowler/bin/prowler"), _provider() + ) + assert captured is result + + warnings = [ + record for record in caplog.records if record.levelno == logging.WARNING + ] + assert [record.getMessage() for record in warnings] == [ + "Secondary Prowler output cleanup failure" + ] + assert "canary" not in "\n".join(str(record.__dict__) for record in warnings) diff --git a/prowler/tests/unit/chk004_prowler_client/test_output_workspace.py b/prowler/tests/unit/chk004_prowler_client/test_output_workspace.py new file mode 100644 index 00000000..867857a7 --- /dev/null +++ b/prowler/tests/unit/chk004_prowler_client/test_output_workspace.py @@ -0,0 +1,358 @@ +"""Output-workspace security and lifecycle tests for CHK.004.""" + +# ruff: noqa: D101, D102, D103 + +import os +import shutil +import stat +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest + + +def _api() -> Any: + from prowler._core import prowler_client + + return prowler_client + + +def test_posix_prefers_writable_dev_shm_equivalent( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + memory_root = tmp_path / "dev-shm" + memory_root.mkdir() + monkeypatch.setattr(os, "access", lambda path, mode: path == memory_root) + api = _api() + required_capacity = ( + api.DEFAULT_MAXIMUM_ARTIFACT_BYTES + + api.DEFAULT_MEMORY_TMPFS_SAFETY_MARGIN_BYTES + ) + monkeypatch.setattr( + shutil, + "disk_usage", + lambda _path: SimpleNamespace(free=required_capacity), + ) + + workspace = ( + _api() + .TemporaryOutputWorkspaceFactory( + platform_name="posix", + memory_root=memory_root, + temporary_root=tmp_path / "disk", + ) + .create() + ) + try: + assert workspace.backend == "memory_tmpfs" + assert workspace.directory.parent == memory_root + assert workspace.directory.stat().st_mode & 0o777 == 0o700 + assert workspace.artifact_path.name == "findings.ocsf.json" + finally: + workspace.cleanup() + + +def test_posix_rejects_tmpfs_without_artifact_capacity_plus_safety_margin( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + api = _api() + memory_root = tmp_path / "dev-shm" + memory_root.mkdir() + disk_root = tmp_path / "disk" + disk_root.mkdir() + monkeypatch.setattr(os, "access", lambda path, mode: path == memory_root) + required_capacity = ( + api.DEFAULT_MAXIMUM_ARTIFACT_BYTES + + api.DEFAULT_MEMORY_TMPFS_SAFETY_MARGIN_BYTES + ) + monkeypatch.setattr( + shutil, + "disk_usage", + lambda _path: SimpleNamespace(free=required_capacity - 1), + ) + + workspace = api.TemporaryOutputWorkspaceFactory( + platform_name="posix", + memory_root=memory_root, + temporary_root=disk_root, + ).create() + try: + assert workspace.backend == "filesystem_temp" + assert workspace.directory.parent == disk_root + finally: + workspace.cleanup() + + +def test_tmpfs_capacity_probe_failure_falls_back_to_system_temp( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + memory_root = tmp_path / "dev-shm" + memory_root.mkdir() + disk_root = tmp_path / "disk" + disk_root.mkdir() + monkeypatch.setattr(os, "access", lambda path, mode: path == memory_root) + monkeypatch.setattr( + shutil, + "disk_usage", + lambda _path: (_ for _ in ()).throw(OSError("probe failed")), + ) + + workspace = ( + _api() + .TemporaryOutputWorkspaceFactory( + platform_name="posix", + memory_root=memory_root, + temporary_root=disk_root, + ) + .create() + ) + try: + assert workspace.backend == "filesystem_temp" + assert workspace.directory.parent == disk_root + finally: + workspace.cleanup() + + +def test_tmpfs_creation_failure_falls_back_once_to_system_temp( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + api = _api() + memory_root = tmp_path / "dev-shm" + memory_root.mkdir() + disk_root = tmp_path / "disk" + disk_root.mkdir() + monkeypatch.setattr(os, "access", lambda path, mode: path == memory_root) + monkeypatch.setattr( + shutil, + "disk_usage", + lambda _path: SimpleNamespace(free=1024 * 1024 * 1024), + ) + original_temporary_directory = __import__("tempfile").TemporaryDirectory + creation_roots: list[Path | None] = [] + + def create_with_tmpfs_failure(*, prefix: str, dir: Path | None) -> Any: + creation_roots.append(dir) + if dir == memory_root: + raise OSError("tmpfs creation failed") + return original_temporary_directory(prefix=prefix, dir=dir) + + monkeypatch.setattr("tempfile.TemporaryDirectory", create_with_tmpfs_failure) + + workspace = api.TemporaryOutputWorkspaceFactory( + platform_name="posix", + memory_root=memory_root, + temporary_root=disk_root, + ).create() + try: + assert workspace.backend == "filesystem_temp" + assert workspace.directory.parent == disk_root + assert creation_roots == [memory_root, disk_root] + finally: + workspace.cleanup() + + +@pytest.mark.parametrize("memory_state", ["missing", "file", "unwritable"]) +def test_posix_falls_back_to_system_temp_when_memory_root_is_unsuitable( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, memory_state: str +) -> None: + memory_root = tmp_path / "dev-shm" + if memory_state == "file": + memory_root.write_text("not a directory", encoding="utf-8") + elif memory_state == "unwritable": + memory_root.mkdir() + monkeypatch.setattr(os, "access", lambda _path, _mode: False) + disk_root = tmp_path / "disk" + disk_root.mkdir() + + workspace = ( + _api() + .TemporaryOutputWorkspaceFactory( + platform_name="posix", memory_root=memory_root, temporary_root=disk_root + ) + .create() + ) + try: + assert workspace.backend == "filesystem_temp" + assert workspace.directory.parent == disk_root + finally: + workspace.cleanup() + + +def test_windows_uses_system_temp_without_posix_permission_claim( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + chmod_calls: list[tuple[object, object]] = [] + monkeypatch.setattr( + os, "chmod", lambda path, mode: chmod_calls.append((path, mode)) + ) + + workspace = ( + _api() + .TemporaryOutputWorkspaceFactory( + platform_name="nt", + memory_root=tmp_path / "dev-shm", + temporary_root=tmp_path, + ) + .create() + ) + try: + assert workspace.backend == "filesystem_temp" + assert chmod_calls == [] + finally: + workspace.cleanup() + + +def test_concurrent_workspaces_are_unique_and_cleanup_only_their_owned_tree( + tmp_path: Path, +) -> None: + factory = _api().TemporaryOutputWorkspaceFactory( + platform_name="nt", temporary_root=tmp_path + ) + with ThreadPoolExecutor(max_workers=8) as pool: + workspaces = list(pool.map(lambda _index: factory.create(), range(16))) + + sentinel = tmp_path / "unrelated" + sentinel.write_text("preserve", encoding="utf-8") + try: + assert len({workspace.directory for workspace in workspaces}) == 16 + for workspace in workspaces: + nested = workspace.directory / "compliance" / "nested" + nested.mkdir(parents=True) + (nested / "result.json").write_text("fixture", encoding="utf-8") + workspace.cleanup() + workspace.cleanup() + assert not workspace.directory.exists() + assert sentinel.read_text(encoding="utf-8") == "preserve" + finally: + for workspace in workspaces: + workspace.cleanup() + + +@pytest.mark.parametrize( + ("artifact_kind", "expected_kind"), + [ + ("missing", "missing"), + ("directory", "nonregular"), + ("symlink", "nonregular"), + ("oversized", "oversized"), + ], +) +def test_artifact_reader_fails_closed_without_path_or_exception_text( + tmp_path: Path, artifact_kind: str, expected_kind: str +) -> None: + workspace = ( + _api() + .TemporaryOutputWorkspaceFactory(platform_name="nt", temporary_root=tmp_path) + .create() + ) + canary = "path-canary-secret" + try: + if artifact_kind == "directory": + workspace.artifact_path.mkdir() + elif artifact_kind == "symlink": + target = workspace.directory / canary + target.write_bytes(b"{}") + workspace.artifact_path.symlink_to(target) + elif artifact_kind == "oversized": + workspace.artifact_path.write_bytes(b"12345") + + with pytest.raises(_api().OutputArtifactError) as caught: + workspace.read_artifact(maximum_bytes=4) + + assert caught.value.kind == expected_kind + assert canary not in repr(caught.value) + assert str(workspace.directory) not in repr(caught.value) + finally: + workspace.cleanup() + + +def test_artifact_reader_accepts_exact_limit_and_reads_incrementally( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + workspace = ( + _api() + .TemporaryOutputWorkspaceFactory(platform_name="nt", temporary_root=tmp_path) + .create() + ) + workspace.artifact_path.write_bytes(b"1234") + real_read = os.read + read_sizes: list[int] = [] + + def recording_read(descriptor: int, size: int) -> bytes: + read_sizes.append(size) + return real_read(descriptor, min(size, 2)) + + monkeypatch.setattr(os, "read", recording_read) + try: + assert workspace.read_artifact(maximum_bytes=4) == b"1234" + assert len(read_sizes) >= 3 + finally: + workspace.cleanup() + + +def test_artifact_read_failure_is_typed_and_closed( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + workspace = ( + _api() + .TemporaryOutputWorkspaceFactory(platform_name="nt", temporary_root=tmp_path) + .create() + ) + workspace.artifact_path.write_bytes(b"{}") + monkeypatch.setattr( + os, + "read", + lambda _descriptor, _size: (_ for _ in ()).throw(OSError("unsafe-canary")), + ) + try: + with pytest.raises(_api().OutputArtifactError) as caught: + workspace.read_artifact(maximum_bytes=4) + assert caught.value.kind == "unreadable" + assert "unsafe-canary" not in repr(caught.value) + finally: + workspace.cleanup() + + +def test_artifact_reader_preserves_preopen_and_opened_file_identity_check( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + workspace = ( + _api() + .TemporaryOutputWorkspaceFactory(platform_name="nt", temporary_root=tmp_path) + .create() + ) + workspace.artifact_path.write_bytes(b"{}") + path_status = workspace.artifact_path.lstat() + try: + with monkeypatch.context() as identity_patch: + identity_patch.setattr( + os, + "fstat", + lambda _descriptor: SimpleNamespace( + st_mode=stat.S_IFREG, + st_dev=path_status.st_dev, + st_ino=path_status.st_ino + 1, + ), + ) + with pytest.raises(_api().OutputArtifactError) as caught: + workspace.read_artifact(maximum_bytes=4) + assert caught.value.kind == "nonregular" + finally: + workspace.cleanup() + + +def test_windows_reparse_safety_is_documented_without_false_atomic_claim() -> None: + repository_root = Path(__file__).parents[3] + output_storage = ( + (repository_root / "README.md") + .read_text(encoding="utf-8") + .split("## Assessment output storage", maxsplit=1)[1] + ) + windows_safety = output_storage.lower() + + assert "best-effort" in windows_safety + assert "controlled directory" in windows_safety + assert "identity checks" in windows_safety + assert "does not claim atomic reparse-point exclusion" in windows_safety