From f33b98ff024c028b29f1dc127dae96ed35007ddb Mon Sep 17 00:00:00 2001 From: Patrick Hickey Date: Mon, 14 Sep 2026 14:23:14 -0700 Subject: [PATCH] feat(validation): authorize HTTP probes within engagement scope Make restricted validation usable for approved cluster endpoints with scoped discovery, managed tunnels, and explicit session handling while keeping execution state isolated and authorization failures closed. --- config/safe-exec-profiles.example.yaml | 36 ++ docs/safe-exec.md | 91 ++- .../validate-findings/adapters/__init__.py | 24 +- .../validate-findings/adapters/base.py | 42 +- .../validate-findings/adapters/http.py | 280 ++++++++ .../validate-findings/adapters/k8s.py | 41 +- .../5-validate/validate-findings/execute.py | 72 ++- .../validate-findings/http_endpoints.py | 474 ++++++++++++++ .../validate-findings/http_policy_io.py | 65 ++ .../validate-findings/http_scope.py | 153 +++++ .../5-validate/validate-findings/plan.py | 242 ++----- .../5-validate/validate-findings/scope.py | 49 +- .../validate-findings/targets.example.yaml | 23 +- src/traust/cli/groups/util.py | 29 +- tests/test_http_endpoints.py | 597 ++++++++++++++++++ tests/test_http_review.py | 338 ++++++++++ tests/test_p2_hardening.py | 67 +- tests/test_validate_findings.py | 4 +- 18 files changed, 2390 insertions(+), 237 deletions(-) create mode 100644 harnessing/5-validate/validate-findings/adapters/http.py create mode 100644 harnessing/5-validate/validate-findings/http_endpoints.py create mode 100644 harnessing/5-validate/validate-findings/http_policy_io.py create mode 100644 harnessing/5-validate/validate-findings/http_scope.py create mode 100644 tests/test_http_endpoints.py create mode 100644 tests/test_http_review.py diff --git a/config/safe-exec-profiles.example.yaml b/config/safe-exec-profiles.example.yaml index 485971f..267648a 100644 --- a/config/safe-exec-profiles.example.yaml +++ b/config/safe-exec-profiles.example.yaml @@ -11,6 +11,11 @@ # run (e.g. ./gradlew); everything else must be a bare basename. # - `keep_env` names env vars preserved (beyond the scrubbed base set) # when safe_exec runs the command. +# - `keep_env_heads` narrows that: only segments whose head basename is +# listed receive `keep_env`/`extra_env`. Empty or absent = every segment. +# - `curl_allowed_hosts` pins every curl URL operand to those hostnames +# (exact match, case-insensitive, no globs; URL and host:port forms +# normalize). A non-empty list also denies redirect-following. # - Granting `git` never grants git network subcommands or dangerous # `-c` config keys — those stay denied inside the validator. # - Network posture: dependency- @@ -18,7 +23,26 @@ # mirrors) is accepted residual risk — safe_exec constrains the argv # and environment, not child-process sockets; hermetic prefetch is a # deferred hardening. +# +# Posture framework (modeled on Kubernetes Pod Security Standards): +# - `restricted` (alias `high`): "fail closed" — every curl URL is denied unless +# its host is listed in `curl_allowed_hosts` or supplied at call-site (ROE +# engagement scope). Redirects are denied. `keep_env_heads` is enforced when +# pipelines and env vars are present. +# - `baseline` (alias `medium`): "somewhere in between" — public http/https +# destinations are permitted; private/internal IP ranges (RFC 1918, loopback, +# link-local/cloud metadata 169.254.169.254) and local domains (.local, +# .internal, .cluster.local, localhost) are denied unless explicitly +# allowlisted. Redirects are denied. +# - `privileged` (alias `low`): "fail open" — permissive defaults; empty host +# list allows any destination. Legacy behavior. version: 1 + +# File-wide fallbacks; any profile may override. +defaults: + # Posture: restricted | baseline | privileged (or high | medium | low) + posture: baseline + profiles: validation-step: description: >- @@ -29,35 +53,45 @@ profiles: SOUND — without it every bearer-auth probe silently ran unauthenticated and its refutations were unsound; the token is lab-cluster-scoped and short-TTL. + # Set to restricted for fail-closed curl validation (ROE supplies lab hosts) + posture: restricted allow: [curl, oc, kubectl, jq, grep, base64, head, tail, tr, wc, cat, sleep, echo, printf] allow_pipelines: true keep_env: [KUBECONFIG, VF_OAUTH_TOKEN] + # keeps the token out of a `| grep x` tail + keep_env_heads: [curl, oc, kubectl] + curl_allowed_hosts: [] go-fuzz: description: >- create-fuzzing Go-native build/run steps (skill integration pending). Cloning stays in the skill's S3-gated git step, never here. + posture: baseline allow: [go, make, gofmt, git] allow_pipelines: false keep_env: [GOPATH, GOCACHE, GOMODCACHE, GOFLAGS] java-build: description: Maven/Gradle build-verification steps (integration pending). + posture: baseline allow: [mvn, gradle, java, javac, jar, make, git] allowed_path_heads: ["./gradlew", "./mvnw"] allow_pipelines: false keep_env: [JAVA_HOME] python-test: description: Target test-suite execution for /patch (integration pending). + posture: baseline allow: [python3, pytest, make, git] allow_pipelines: false keep_env: [VIRTUAL_ENV] node-build: description: Node build/test steps (integration pending). + posture: baseline allow: [npm, npx, node, make, git] allow_pipelines: false rust-fuzz: description: cargo-fuzz build/run steps (integration pending). + posture: baseline allow: [cargo, rustc, make, git] allow_pipelines: false keep_env: [CARGO_HOME, RUSTUP_HOME] @@ -67,10 +101,12 @@ profiles: (python3 -m traust.cli adapters govulncheck). govulncheck type-checks and builds the target module, so hostile build tags/cgo see only the scrubbed env — no ambient tokens to exfiltrate (gate rule S10). + posture: baseline allow: [govulncheck] allow_pipelines: false keep_env: [GOPATH, GOCACHE, GOMODCACHE, GOFLAGS, GOPROXY, GONOSUMDB, GONOSUMCHECK, GOSUMDB] generic-build: description: Make-only fallback for unclassified build systems. + posture: baseline allow: [make, git] allow_pipelines: false diff --git a/docs/safe-exec.md b/docs/safe-exec.md index 222e882..affb7cd 100644 --- a/docs/safe-exec.md +++ b/docs/safe-exec.md @@ -64,7 +64,9 @@ fallback copy of the `validation-step` profile lives in the module — keep them in sync; the harness ships the template as `config/safe-exec-profiles.example.yaml`). Each profile declares `allow` (binary heads), optional `allowed_path_heads` (e.g. `./gradlew`), -`allow_pipelines`, and `keep_env`. Current profiles and their consumers: +`allow_pipelines`, `keep_env`, and optionally `keep_env_heads` (which segment +heads receive `keep_env`), `curl_allowed_hosts` and `posture`. +Current profiles and their consumers: | Profile | Used by | Notes | |---|---|---| @@ -72,6 +74,87 @@ in sync; the harness ships the template as | `go-scan` | python3 -m traust.cli adapters govulncheck | govulncheck type-checks and builds the target module — hostile build tags/cgo see only the scrubbed env | | `go-fuzz`, `java-build`, `python-test`, `node-build`, `rust-fuzz`, `generic-build` | build/test lanes (fuzz-harness and patch-verification steps) | dependency-manager egress from build tools is accepted residual risk: safe_exec constrains argv and environment, not child-process sockets; hermetic prefetch is a deferred hardening | +### Security postures and curl host posture + +Safe-exec provides a posture framework modeled on Kubernetes Pod Security Standards: + +| Posture | Alias | Egress / Host Semantics | Environment & Redirects | +|---|---|---|---| +| `restricted` | `high` | **Fail closed**: Every curl URL is denied unless its destination is listed in `curl_allowed_hosts` or passed at call time (ROE hosts). | Redirects denied. `keep_env_heads` required on pipelines when env vars are kept. | +| `baseline` | `medium` | **Public allowed, private denied**: Public http/https egress is permitted. Non-global IP ranges (RFC 1918, loopback, link-local/cloud metadata 169.254.169.254) and internal domains (.local, .internal, .cluster.local, localhost) are denied unless explicitly allowlisted. | Redirects denied. `keep_env_heads` enforced when pipelines and env vars are present. | +| `privileged` | `low` | **Fail open**: Unrestricted destinations; empty host list allows any destination. | Redirects allowed when host list is empty. Legacy default. | + +Static config cannot name a per-engagement lab cluster. Keep the restricted +profile's static list empty and declare authorization in `targets.yaml`. +`clusters[].api` and explicit `http_targets` provide hosts for command-text probes. +Structured HTTP probes resolve a single endpoint, check resource scope and pass +only that endpoint to safe_exec. Reproduce a host verdict with `--allowed-host`. + +### Engagement HTTP authorization + +The [targets template](../harnessing/5-validate/validate-findings/targets.example.yaml) +contains optional `http_targets` and per-cluster `http_discovery` fields. These +are explicit-only grants; finding text and inferred scope cannot populate them. + +- `http_targets`: exact host, required cluster `context`, optional `namespace`, + `resource` and `name`, and an independent `credentials` permission. +- `http_discovery.routes`: concrete namespace, optional exact resource `names`, + approved DNS `domains`, and optional `credentials`. Domain boundaries use a + DNS-label boundary, not an arbitrary suffix. A discovered hostname is not a grant. +- `http_discovery.nodes`: optional exact `names`, permitted CIDR `networks`, and + `address_types` (defaults to InternalIP). This narrowly authorizes node reads + and HTTP probes without adding a wildcard namespace grant. +- `http_discovery.port_forward_credentials`: permits authenticated probes through + authorized, adapter-owned Pod tunnels. Default false. +- `http_discovery.kube_env`: explicit environment variable names needed by a + kubeconfig exec-auth plugin when opening a tunnel. No additional variables + are preserved by default. +- `http_discovery.ca_bundle`: operator-owned CA file for lab TLS. Verification + stays enabled by default. `insecure_tls: true` is an explicit engagement-only + exception, never inferred from a finding. +- `http_discovery.session_check_path`: origin-relative identity endpoint that + returns 2xx only for an authenticated session. Required for authenticated CSRF + probes; a login page or CSRF cookie alone is not proof of authentication. + +Discovery uses the named kubeconfig context and checks any declared API against +that context. Collection reads require `list`, named reads require `get`; +expiry, verb denies and off-limits rules continue to apply. Ambiguous endpoints, +RBAC failures, malformed responses and unauthorized destinations fail closed. +Route/console probes use scoped route discovery; kubelet probes use scoped node +addresses. Service discovery selects a matching Pod and resolves the Service's +target port. The adapter owns the ephemeral local port, readiness deadline and +cleanup. Loopback is not an engagement-wide grant. Service transport defaults to +HTTP unless Service port metadata identifies HTTPS; a reviewed request can select +its scheme explicitly. TLS tunnels retain the service DNS identity and pin the +connection to the owned local port. Proxy environment variables cannot redirect +structured requests. + +Generated HTTP plans contain `target.http` rather than shell programs. The request +includes mode, method, relative path, transport, optional port, concrete namespaces +and selection hints. Authentication and CSRF are explicit request options and +require a credential grant; anonymous probes remain anonymous. Token creation +also requires permission for `create serviceaccounts/token`. Console session +cookies stay in memory. HTTP DELETE requires destructive permission; other write +methods are classified as mutating and checked against the corresponding verb. +No shell execution, redirect following or arbitrary cookie-file writes are enabled. +Direct structured cluster-API requests are limited to GET/HEAD health and version +endpoints; Kubernetes resource requests must use resource-aware operations instead. +An authenticated-caller claim is inconclusive until a reviewed authenticated probe +satisfies its identity precondition. Reflected request credentials and CSRF secrets +are redacted before HTTP output reaches evidence artifacts. Sensitive headers +are validated in memory and passed through adapter-owned stdin, not process +arguments. Only the trusted adapter introduces this stdin header source and the +owned-tunnel connection mapping after validation; command text cannot grant them. +Structured steps cannot carry shell rollback commands. Session initialization +must succeed and preserve the server's cookie names; HEAD uses curl's HEAD mode. + +The engine allowlist remains hostname-only: it is not port/path isolation or a +DNS-rebinding defense. Optional resource selectors are enforced by the structured +resolver, not by the command-text host allowlist. Use structured requests for +resource-scoped grants and review operator-approved DNS domains/CIDRs accordingly. +Package pins, the deployed skill tree and the selected config home must all carry +this implementation before enabling restricted policy in an estate. + ## Modes and the bypass - **`SAFE_EXEC_MODE=warn|enforce`** — `warn` logs what enforce would have @@ -82,9 +165,9 @@ in sync; the harness ships the template as stderr and appends to the bypass log (`~/.local/state/…`, 0700). A bypass with no recorded reason is a finding, not a convenience. - Blocked commands return exit 126 with a `[safe_exec blocked: …]` reason. - In validate-findings, a blocked step becomes an `inconclusive` verdict — - never a silent pass, and never grounds for a `refuted` (refutation-soundness - gate). + Kubernetes validation policy refusals become `blocked_by_scope`; resolution or + transport failures are inconclusive. Neither provides evidence to confirm or + refute a finding. ## CLI diff --git a/harnessing/5-validate/validate-findings/adapters/__init__.py b/harnessing/5-validate/validate-findings/adapters/__init__.py index 4bedc9f..56d841f 100644 --- a/harnessing/5-validate/validate-findings/adapters/__init__.py +++ b/harnessing/5-validate/validate-findings/adapters/__init__.py @@ -21,6 +21,8 @@ if not __package__: sys.path.insert(0, str(Path(__file__).parent)) +from typing import Any + from .base import AdapterBase, Fingerprint, StepResult from .container import ContainerAdapter from .k8s import K8sAdapter @@ -48,13 +50,31 @@ def get_adapter(name: str) -> AdapterBase: def new_adapter(name: str) -> AdapterBase: - """Return a fresh adapter instance with independent state.""" + """Return a fresh adapter instance with independent state. + + Unbound: call its bind_scope() before running steps, or a closed-posture + profile refuses every curl.""" try: return _ADAPTER_CLASSES[name]() except KeyError as e: raise ValueError(f"unknown adapter: {name!r}") from e +def bind_scope(scope: object) -> None: + """Bind the engagement scope to every shared adapter singleton. + + Call once per run, before preflight: the adapters are module-level + singletons, and an unbound one carries no curl hosts.""" + for adapter in TARGET_ADAPTERS.values(): + adapter.bind_scope(scope) + + +def bind_profile_map(profile_map: dict[str, Any] | None) -> None: + """Bind resolved safe_exec profiles to every shared adapter singleton.""" + for adapter in TARGET_ADAPTERS.values(): + adapter.bind_profile_map(profile_map) + + __all__ = [ "TARGET_ADAPTERS", "AdapterBase", @@ -63,6 +83,8 @@ def new_adapter(name: str) -> AdapterBase: "K8sAdapter", "StepResult", "WasmAdapter", + "bind_profile_map", + "bind_scope", "get_adapter", "new_adapter", ] diff --git a/harnessing/5-validate/validate-findings/adapters/base.py b/harnessing/5-validate/validate-findings/adapters/base.py index a6fedbc..f4a7e91 100644 --- a/harnessing/5-validate/validate-findings/adapters/base.py +++ b/harnessing/5-validate/validate-findings/adapters/base.py @@ -11,7 +11,7 @@ import subprocess from dataclasses import asdict, dataclass, field from pathlib import Path -from typing import ClassVar +from typing import Any, ClassVar @dataclass @@ -93,8 +93,22 @@ class AdapterBase: #: verbs that delete data, kill workloads, or cannot be undone DESTRUCTIVE_VERBS: ClassVar[set[str]] = {"delete", "scale-zero", "kill", "fuzz-import"} - def __init__(self): - pass + def __init__(self) -> None: + #: extra curl hosts for safe_exec; see bind_scope() + self._curl_hosts: tuple[str, ...] = () + self._safe_exec_profile_map: dict[str, Any] | None = None + + def bind_scope(self, scope: object) -> None: + """Take the curl host allowlist from the engagement scope. + + Lets a deployment run validation-step with `posture: restricted` + and still reach the cluster it was authorized against.""" + getter = getattr(scope, "curl_hosts", None) + self._curl_hosts = tuple(getter()) if callable(getter) else () + + def bind_profile_map(self, profile_map: dict[str, Any] | None) -> None: + """Take the resolved safe_exec profile map for this deployment.""" + self._safe_exec_profile_map = profile_map # ----- classification --------------------------------------------- @@ -191,27 +205,31 @@ def _safe_exec(cls): cls._safe_exec_mod = importlib.import_module("traust_engine._util.safe_exec") return cls._safe_exec_mod - @classmethod - def _vet_shell_string(cls, cmd: str) -> tuple[list[str] | None, str]: + def _vet_shell_string(self, cmd: str) -> tuple[list[str] | None, str]: """Vet a PoC-derived command string via safe_exec. Returns (argv, "") for a pipeless command, (None, "") for an approved pipeline, or (None, reason) when rejected.""" - se = cls._safe_exec() - v = se.vet_command_string(cmd, se.get_profile(cls._SAFE_EXEC_PROFILE)) + se = self._safe_exec() + v = se.vet_command_string( + cmd, + se.get_profile(self._SAFE_EXEC_PROFILE, profile_map=self._safe_exec_profile_map), + allowed_hosts=self._curl_hosts, + ) if not v.ok: return None, v.reason if len(v.segments) == 1: return list(v.segments[0]), "" return None, "" # approved pipeline - @classmethod def _run( - cls, cmd: str | list[str], *, timeout: int = 120, input_: str | None = None + self, cmd: str | list[str], *, timeout: int = 120, input_: str | None = None ) -> tuple[int, str, str]: if isinstance(cmd, str): - se = cls._safe_exec() - profile = se.get_profile(cls._SAFE_EXEC_PROFILE) - v = se.vet_command_string(cmd, profile) + se = self._safe_exec() + profile = se.get_profile( + self._SAFE_EXEC_PROFILE, profile_map=self._safe_exec_profile_map + ) + v = se.vet_command_string(cmd, profile, allowed_hosts=self._curl_hosts) if not v.ok: return 126, "", f"[step blocked: {v.reason}]" # single command or approved pipeline — both execute diff --git a/harnessing/5-validate/validate-findings/adapters/http.py b/harnessing/5-validate/validate-findings/adapters/http.py new file mode 100644 index 0000000..8f4b953 --- /dev/null +++ b/harnessing/5-validate/validate-findings/adapters/http.py @@ -0,0 +1,280 @@ +from __future__ import annotations + +import contextlib +import os +import selectors +import shlex +import subprocess +import time +from collections.abc import Iterator +from dataclasses import asdict +from http.cookies import SimpleCookie +from typing import Protocol +from urllib.parse import urlsplit + +if __package__ and "." in __package__: + from ..http_endpoints import Endpoint, EndpointResolver, HttpProbe + from ..scope import Scope +else: + from http_endpoints import Endpoint, EndpointResolver, HttpProbe + from scope import Scope + +from .base import AdapterBase + + +class AuditSink(Protocol): + def append(self, **fields: object) -> None: ... + + +class HttpExecutor: + def __init__(self, adapter: AdapterBase, scope: Scope, binary: str, audit: AuditSink) -> None: + self.adapter = adapter + self.scope = scope + self.binary = binary + self.audit = audit + self.resolver = EndpointResolver(scope, adapter._run, binary) + self._tunnel_address: tuple[str, int] | None = None + + @contextlib.contextmanager + def tunnel(self, endpoint: Endpoint) -> Iterator[str]: + args = [ + self.binary, + f"--context={endpoint.context}", + "--namespace", + endpoint.namespace, + "port-forward", + f"pod/{endpoint.name}", + f":{endpoint.remote_port}", + "--address=127.0.0.1", + ] + names = {"PATH", "HOME", "KUBECONFIG", "LANG"} + names.update(self.scope.clusters[endpoint.context].http_discovery.kube_env) + with subprocess.Popen( + args, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + env={key: value for key, value in os.environ.items() if key in names}, + ) as process: + try: + deadline = time.monotonic() + 15 + with selectors.DefaultSelector() as selector: + selector.register(process.stdout, selectors.EVENT_READ) + buffer = b"" + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError("port-forward exited before readiness") + for key, _ in selector.select(timeout=0.2): + buffer += os.read(key.fileobj.fileno(), 4096) + if len(buffer) > 65536: + raise RuntimeError("port-forward readiness output too large") + lines = buffer.split(b"\n") + buffer = lines.pop() + for line in lines: + text = line.decode("utf-8", errors="replace") + if text.startswith("Forwarding from 127.0.0.1:"): + address = text.split(" -> ", 1)[0].removeprefix( + "Forwarding from " + ) + port = urlsplit(f"http://{address}").port + if port is None or not 1 <= port <= 65535: + raise ValueError("invalid port-forward readiness") + destination = urlsplit(endpoint.origin) + self._tunnel_address = (destination.hostname, port) + try: + yield f"{destination.scheme}://{destination.hostname}:{port}" + finally: + self._tunnel_address = None + return + raise TimeoutError("port-forward readiness timed out") + finally: + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=3) + + def curl(self, argv: list[str], host: str) -> tuple[int, str, str]: + engine = self.adapter._safe_exec() + profile = engine.get_profile( + "validation-step", profile_map=self.adapter._safe_exec_profile_map + ) + argv = [argv[0], "--noproxy", "*", *argv[1:]] + verdict = engine.vet_command_string(shlex.join(argv), profile, allowed_hosts=(host,)) + if not verdict.ok: + raise PermissionError(f"HTTP execution policy: {verdict.reason}") + command = list(verdict.segments[0]) + private_headers = [] + prepared = [] + index = 0 + while index < len(command): + if command[index] == "-H" and command[index + 1].partition(":")[0].lower() in { + "authorization", + "cookie", + "x-csrftoken", + "x-csrf-token", + }: + private_headers.append(command[index + 1]) + index += 2 + else: + prepared.append(command[index]) + index += 1 + if private_headers: + prepared[1:1] = ["-H", "@-"] + if self._tunnel_address: + tunnel_host, port = self._tunnel_address + if host != tunnel_host: + raise PermissionError("HTTP host does not match the owned tunnel") + prepared[1:1] = ["--connect-to", f"{tunnel_host}:{port}:127.0.0.1:{port}"] + return engine.run_segments( + [prepared], + profile, + timeout=30, + input_="\n".join(private_headers) + "\n" if private_headers else None, + ) + + def token(self, endpoint: Endpoint, probe: HttpProbe) -> str | None: + if not probe.authenticate: + return None + if not endpoint.credentials: + raise PermissionError("credentials are not authorized for this HTTP endpoint") + token = os.environ.get("VF_OAUTH_TOKEN") + if token: + return token + if not probe.service_account or not endpoint.namespace: + raise RuntimeError("authenticated HTTP probe requires an available credential") + self.resolver.authorize( + endpoint.context, + endpoint.namespace, + "serviceaccounts/token", + probe.service_account, + "create", + ) + code, output, _ = self.adapter._run( + [ + self.binary, + f"--context={endpoint.context}", + "--namespace", + endpoint.namespace, + "create", + "token", + probe.service_account, + "--duration=10m", + ], + timeout=15, + ) + if code or not output.strip(): + raise RuntimeError("could not obtain the authorized ServiceAccount credential") + return output.strip() + + def execute(self, context: str, probe: HttpProbe) -> tuple[int, str, str]: + endpoint = self.resolver.resolve(context, probe) + self.resolver.authorize( + endpoint.context, + endpoint.namespace, + endpoint.resource, + endpoint.name, + "port-forward+http", + ) + verb = { + "GET": "get", + "HEAD": "get", + "OPTIONS": "get", + "POST": "create", + "PUT": "update", + "PATCH": "patch", + "DELETE": "delete", + }[probe.method] + self.resolver.authorize( + endpoint.context, endpoint.namespace, endpoint.resource, endpoint.name, verb + ) + if ( + probe.authenticate + and urlsplit(endpoint.origin).scheme != "https" + and not endpoint.remote_port + ): + raise PermissionError("credentials require HTTPS outside an owned tunnel") + self.audit.append(event="http-endpoint", endpoint=asdict(endpoint)) + token = self.token(endpoint, probe) + secrets = [token] if token else [] + headers = dict(probe.headers) + policy = self.scope.clusters[context].http_discovery + tls_args = ["--cacert", policy.ca_bundle] if policy.ca_bundle else [] + if policy.insecure_tls: + tls_args.append("--insecure") + if token: + if any(ord(char) < 32 or ord(char) == 127 for char in token): + raise ValueError("invalid credential") + headers["Authorization"] = f"Bearer {token}" + connection = ( + self.tunnel(endpoint) + if endpoint.remote_port + else contextlib.nullcontext(endpoint.origin) + ) + with connection as origin: + host = urlsplit(origin).hostname + if probe.csrf: + args = ["curl", *tls_args, "-sS", "--max-time", "20", "-D", "-", "-o", "/dev/null"] + for name, value in headers.items(): + args.extend(["-H", f"{name}: {value}"]) + code, output, _error = self.curl([*args, f"{origin}/"], host) + statuses = [ + line.split()[1] + for line in output.splitlines() + if line.startswith("HTTP/") and len(line.split()) >= 2 + ] + if code or not statuses or not statuses[-1].startswith("2"): + raise RuntimeError("HTTP session initialization failed") + cookies = SimpleCookie() + for line in output.splitlines(): + if line.lower().startswith("set-cookie:"): + cookies.load(line.partition(":")[2].strip()) + csrf = next( + (item.value for name, item in cookies.items() if "csrf" in name.lower()), None + ) + if not csrf: + raise RuntimeError("HTTP session initialization did not supply a CSRF cookie") + secrets.extend(item.value for item in cookies.values() if item.value) + headers["Cookie"] = "; ".join( + item.OutputString(attrs=[]) for item in cookies.values() + ) + headers["X-CSRFToken"] = csrf + headers["X-CSRF-Token"] = csrf + if probe.csrf and probe.authenticate: + if not policy.session_check_path: + raise RuntimeError( + "authenticated session requires an operator-configured identity check" + ) + check_args = [ + "curl", + *tls_args, + "-sS", + "--max-time", + "20", + "-o", + "/dev/null", + "-w", + "%{http_code}", + ] + for name, value in headers.items(): + if any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ValueError("invalid session header") + check_args.extend(["-H", f"{name}: {value}"]) + code, status, _error = self.curl( + [*check_args, f"{origin}{policy.session_check_path}"], host + ) + if code or not status.strip().startswith("2"): + raise RuntimeError("authenticated session identity check failed") + method_args = ["--head"] if probe.method == "HEAD" else ["-X", probe.method] + args = ["curl", *tls_args, "-sS", "--max-time", "20", "--compressed", *method_args] + for name, value in headers.items(): + if any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ValueError("invalid response-derived HTTP header") + args.extend(["-H", f"{name}: {value}"]) + args.extend([f"{origin}{probe.path}", "-w", "\nvf-http-status:%{http_code}"]) + code, output, error = self.curl(args, host) + for secret in secrets: + output = output.replace(secret, "[REDACTED]") + error = error.replace(secret, "[REDACTED]") + return code, output, error diff --git a/harnessing/5-validate/validate-findings/adapters/k8s.py b/harnessing/5-validate/validate-findings/adapters/k8s.py index ea7f26d..cee2b39 100644 --- a/harnessing/5-validate/validate-findings/adapters/k8s.py +++ b/harnessing/5-validate/validate-findings/adapters/k8s.py @@ -221,6 +221,44 @@ def _unresolved(reason: str) -> StepResult: observed = (out + err).strip() evidence.append(self._save_artifact(artifacts_dir, sid, "stdout", observed)) + elif verb == "port-forward+http" and target.get("http") is not None: + if __package__ and "." in __package__: + from ..http_endpoints import HttpProbe + else: + from http_endpoints import HttpProbe + + from .http import HttpExecutor + + try: + probe = HttpProbe.model_validate(target["http"]) + requires_identity = re.search( + r"\bauthenticated\b|logged-in|console user|low-privilege user" + r"|low-priv|any user with", + step_get(step, "expected", ""), + re.IGNORECASE, + ) + if requires_identity and not probe.authenticate: + return _unresolved("authenticated caller precondition is not satisfied") + rc, out, err = HttpExecutor(self, scope, kb, audit).execute( + target.get("context"), probe + ) + observed = (out + err).strip() + except PermissionError as exc: + return StepResult( + step_id=sid, + adapter="k8s", + verb=verb, + target=target, + classification=cls, + verdict="blocked_by_scope", + scope_reason=str(exc), + finding_ref=step_get(step, "finding_ref"), + novel_ref=step_get(step, "novel_ref"), + ) + except (ValueError, RuntimeError, TimeoutError) as exc: + return _unresolved(str(exc)) + evidence.append(self._save_artifact(artifacts_dir, sid, "http", observed)) + elif verb == "port-forward+http": # The plan embeds the curl; assume the operator already runs a # port-forward in another terminal, or run inline against svc. @@ -354,7 +392,7 @@ def _unresolved(reason: str) -> StepResult: ) expected = step_get(step, "expected", "") - verdict = self._verdict(verb, rc, observed, expected) + verdict = "blocked_by_scope" if rc == 126 else self._verdict(verb, rc, observed, expected) err_tag = "" if ( verdict == "inconclusive" @@ -377,6 +415,7 @@ def _unresolved(reason: str) -> StepResult: observed=observed_out, evidence=evidence, error=err_tag, + scope_reason=observed_out if rc == 126 else "", finding_ref=step_get(step, "finding_ref"), novel_ref=step_get(step, "novel_ref"), duration_ms=int((time.monotonic() - t0) * 1000), diff --git a/harnessing/5-validate/validate-findings/execute.py b/harnessing/5-validate/validate-findings/execute.py index 939bc61..36d174c 100644 --- a/harnessing/5-validate/validate-findings/execute.py +++ b/harnessing/5-validate/validate-findings/execute.py @@ -15,7 +15,7 @@ import json import sys from pathlib import Path -from typing import ClassVar +from typing import Any, ClassVar try: import yaml @@ -24,13 +24,23 @@ if __package__: from . import soundness - from .adapters import StepResult, get_adapter, kubeargv + from .adapters import ( + Fingerprint, + StepResult, + kubeargv, + new_adapter, + ) from .novel import diff_surfaces, probe_steps from .scope import Action, Scope else: sys.path.insert(0, str(Path(__file__).parent)) import soundness - from adapters import StepResult, get_adapter, kubeargv + from adapters import ( + Fingerprint, + StepResult, + kubeargv, + new_adapter, + ) from novel import diff_surfaces, probe_steps from scope import Action, Scope @@ -87,6 +97,10 @@ def _step_actions(step: dict) -> tuple[list[Action], str | None]: actions = [base_action] if base_action.adapter != "k8s": return actions, None + if step.get("target", {}).get("http") is not None: + if step.get("cmd") or step.get("rollback"): + return actions, "structured HTTP steps cannot carry command or rollback text" + return actions, None cmd = step.get("cmd") or "" if not cmd: return actions, None @@ -113,7 +127,7 @@ def _step_actions(step: dict) -> tuple[list[Action], str | None]: context=base_action.context, namespace=ns, resource=k.resource or base_action.resource, - name=base_action.name, + name=None if k.resource in {"node", "nodes"} else base_action.name, image=base_action.image, ) ) @@ -159,14 +173,24 @@ def _load_evidence(r: StepResult, artifacts_dir: Path) -> str: return r.observed or "" -def preflight(scope): +def _auto_profile_map() -> dict[str, Any] | None: + from traust.context import load_engine + + try: + return load_engine().adapters.safe_exec_profile_map() + except Exception as exc: + raise RuntimeError(f"Failed to load safe_exec configuration from estate: {exc}") from exc + + +def preflight(scope: Scope, *, profile_map: dict[str, Any] | None = None) -> list[Fingerprint]: """Capture target fingerprints for ``metadata.target_fingerprint``. run.py calls this as ``ex.preflight(scope)`` (it always has — the function just never existed, so fingerprints came back empty). Delegates to each bound adapter's own ``preflight()``. """ - fps = [] + profiles = profile_map if profile_map is not None else _auto_profile_map() + fps: list[Fingerprint] = [] seen = set() bound = [] if getattr(scope, "clusters", None): @@ -180,7 +204,10 @@ def preflight(scope): continue seen.add(name) with contextlib.suppress(Exception): - fps.extend(get_adapter(name).preflight(scope)) + adapter = new_adapter(name) + adapter.bind_scope(scope) + adapter.bind_profile_map(profiles) + fps.extend(adapter.preflight(scope)) return fps @@ -191,7 +218,19 @@ def run( *, permit_destructive: bool = False, second_pass_novel: bool = True, + profile_map: dict[str, Any] | None = None, ) -> tuple[list[StepResult], AuditLog]: + profiles = profile_map if profile_map is not None else _auto_profile_map() + adapters = {} + + def get_adapter(name: str) -> Any: + if name not in adapters: + adapter = new_adapter(name) + adapter.bind_scope(scope) + adapter.bind_profile_map(profiles) + adapters[name] = adapter + return adapters[name] + plan = _load_plan(plan_path) steps: list[dict] = list(plan.get("steps", [])) artifacts_dir = out_dir / "artifacts" @@ -239,6 +278,19 @@ def record(step: dict, res: StepResult): sid = step["id"] verb = step.get("verb", "") cls = step.get("classification", "safe") + http = step.get("target", {}).get("http") + if isinstance(http, dict): + method = http.get("method", "GET") + severity = ( + "destructive" + if method == "DELETE" + else "mutating" + if method in {"POST", "PUT", "PATCH"} + else "safe" + ) + ranks = {"safe": 0, "mutating": 1, "destructive": 2} + cls = max((cls, severity), key=lambda value: ranks.get(value, 2)) + step["classification"] = cls # pre-skipped in plan if step.get("skip") and verb != "placeholder": @@ -367,7 +419,11 @@ class _TM: ) # rollback mutating steps immediately after evidence capture - if cls == "mutating": + if ( + cls == "mutating" + and not http + and res.verdict not in {"blocked_by_scope", "not_attempted"} + ): try: ok_rb, rb_out = adapter.rollback(step, res) except Exception as e: diff --git a/harnessing/5-validate/validate-findings/http_endpoints.py b/harnessing/5-validate/validate-findings/http_endpoints.py new file mode 100644 index 0000000..f83d93e --- /dev/null +++ b/harnessing/5-validate/validate-findings/http_endpoints.py @@ -0,0 +1,474 @@ +from __future__ import annotations + +import json +from collections.abc import Callable +from dataclasses import dataclass +from typing import Literal +from urllib.parse import urlsplit + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +if __package__: + from .http_scope import hostname, origin + from .scope import Action, Scope +else: + from http_scope import hostname, origin + from scope import Action, Scope + + +class HttpProbe(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + mode: Literal["route", "service", "node", "direct"] = "route" + method: Literal["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"] = "GET" + path: str = "/" + port: int | None = Field(default=None, ge=1, le=65535) + scheme: Literal["auto", "http", "https"] = "auto" + hint: str = "" + namespaces: tuple[str, ...] = () + name: str | None = Field(default=None, pattern=r"^[a-z0-9][a-z0-9.-]*$") + url: str | None = None + headers: dict[str, str] = Field(default_factory=dict) + authenticate: bool = False + csrf: bool = False + service_account: str | None = Field(default=None, pattern=r"^[a-z0-9][a-z0-9.-]*$") + + @field_validator("path") + @classmethod + def valid_path(cls, value: str) -> str: + parsed = urlsplit(value) + if not value.startswith("/") or value.startswith("//") or parsed.netloc or parsed.scheme: + raise ValueError("probe path must be relative to its authorized origin") + if any(ord(char) < 32 for char in value) or "\\" in value: + raise ValueError("invalid HTTP path") + return value + + @field_validator("headers") + @classmethod + def valid_headers(cls, values: dict[str, str]) -> dict[str, str]: + for name, value in values.items(): + if not name or not all( + char.isascii() and (char.isalnum() or char == "-") for char in name + ): + raise ValueError("invalid header name") + if name.lower() in {"host", "authorization", "proxy-authorization", "cookie"}: + raise ValueError("authentication and destination headers are adapter-owned") + if any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ValueError("invalid header value") + return values + + +class Metadata(BaseModel): + name: str = Field(pattern=r"^[a-z0-9][a-z0-9.-]*$") + namespace: str | None = Field(default=None, pattern=r"^[a-z0-9][a-z0-9-]*$") + uid: str = "" + labels: dict[str, str] = Field(default_factory=dict) + + +class NodeAddress(BaseModel): + type: str + address: str + + +class NodeStatus(BaseModel): + addresses: list[NodeAddress] = Field(default_factory=list) + + +class RouteSpec(BaseModel): + host: str + tls: dict[str, object] | None = None + + +class ServicePort(BaseModel): + port: int = Field(ge=1, le=65535, strict=True) + targetPort: int | str | None = None + name: str = "" + appProtocol: str = "" + + +class ServiceSpec(BaseModel): + ports: list[ServicePort] + selector: dict[str, str] = Field(default_factory=dict) + type: str = "ClusterIP" + + +class ContainerPort(BaseModel): + containerPort: int = Field(ge=1, le=65535, strict=True) + name: str = "" + + +class ContainerSpec(BaseModel): + ports: list[ContainerPort] = Field(default_factory=list) + + +class PodSpec(BaseModel): + containers: list[ContainerSpec] = Field(default_factory=list) + + +class PodStatus(BaseModel): + phase: str = "" + + +class KubernetesObject(BaseModel): + metadata: Metadata + spec: dict = Field(default_factory=dict) + status: dict = Field(default_factory=dict) + + +class KubernetesList(BaseModel): + items: list[KubernetesObject] + + +@dataclass(frozen=True) +class Endpoint: + origin: str + context: str + namespace: str | None + resource: str + name: str + uid: str = "" + credentials: bool = False + remote_port: int | None = None + + @property + def host(self) -> str: + return hostname(self.origin) + + +class EndpointResolver: + def __init__(self, scope: Scope, run: Callable[..., tuple[int, str, str]], binary: str) -> None: + self.scope = scope + self.run = run + self.binary = binary + + def authorize( + self, + context: str, + namespace: str | None, + resource: str, + name: str | None = None, + verb: str = "get", + ) -> None: + allowed, reason = self.scope.is_in_scope( + Action( + adapter="k8s", + context=context, + namespace=namespace, + resource=resource, + name=name, + verb=verb, + ) + ) + if not allowed: + raise PermissionError(reason) + + def api(self, context: str) -> str: + cluster = self.scope.clusters.get(context) + if cluster is None or context == "__current__": + raise PermissionError("HTTP probes require a named engagement context") + if self.scope.expires: + import datetime + + if datetime.date.today() > self.scope.expires: + raise PermissionError("engagement expired") + code, output, _ = self.run( + [ + self.binary, + f"--context={context}", + "config", + "view", + "--minify", + "-o", + "jsonpath={.clusters[0].cluster.server}", + ], + timeout=15, + ) + if code or not output.strip(): + raise RuntimeError("cannot resolve the engagement context API") + actual = output.strip() + parsed = urlsplit(actual) + hostname(actual) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("invalid cluster API URL") + if cluster.api and origin(cluster.api) != origin(actual): + raise PermissionError("engagement API does not match kubeconfig context") + return actual + + def objects( + self, context: str, namespace: str | None, resource: str, name: str | None = None + ) -> list[KubernetesObject]: + self.authorize(context, namespace, resource, name, "get" if name else "list") + args = [self.binary, f"--context={context}"] + if namespace is not None: + if not namespace or any(char in namespace for char in "*?[]"): + raise PermissionError("discovery requires a concrete authorized namespace") + args.extend(["--namespace", namespace]) + args.extend(["get", resource]) + if name: + Metadata(name=name) + args.append(name) + args.extend(["-o", "json"]) + code, output, _ = self.run(args, timeout=20) + if code: + raise RuntimeError(f"discovery failed for {resource} in {context}/{namespace}") + try: + data = json.loads(output) + objects = ( + [KubernetesObject.model_validate(data)] + if name + else KubernetesList.model_validate(data).items + ) + models = { + "routes": (RouteSpec, None), + "services": (ServiceSpec, None), + "nodes": (None, NodeStatus), + "pods": (PodSpec, PodStatus), + } + spec_model, status_model = models[resource] + for item in objects: + if spec_model: + item.spec = spec_model.model_validate(item.spec).model_dump(exclude_none=True) + if status_model: + item.status = status_model.model_validate(item.status).model_dump( + exclude_none=True + ) + except (ValueError, TypeError) as exc: + raise ValueError(f"invalid {resource} discovery response") from exc + for item in objects: + if item.metadata.namespace != namespace or (name and item.metadata.name != name): + raise PermissionError("discovery response does not match requested resource") + self.authorize(context, namespace, resource, item.metadata.name) + return objects + + def explicit( + self, host: str, context: str, namespace: str | None, resource: str, name: str + ) -> bool | None: + matches = [ + target + for target in self.scope.http_targets + if target.host == host + and target.context == context + and (target.namespace is None or target.namespace == namespace) + and (target.resource is None or target.resource == resource) + and (target.name is None or target.name == name) + ] + return any(target.credentials for target in matches) if matches else None + + def resolve(self, context: str, probe: HttpProbe) -> Endpoint: + api = self.api(context) + cluster = self.scope.clusters[context] + if probe.mode == "direct": + if not probe.url: + raise ValueError("direct HTTP probe requires a URL") + parsed = urlsplit(probe.url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("direct HTTP probe requires HTTP(S)") + host = hostname(probe.url) + namespace = probe.namespaces[0] if len(probe.namespaces) == 1 else None + self.authorize(context, namespace, "http_targets", probe.name, "port-forward+http") + credentials = self.explicit(host, context, namespace, "http_targets", probe.name or "") + if host == hostname(api): + if probe.path not in { + "/healthz", + "/livez", + "/readyz", + "/version", + } or probe.method not in {"GET", "HEAD"}: + raise PermissionError( + "direct API probes are limited to health and version endpoints" + ) + if credentials is None and origin(probe.url) == origin(api): + credentials = False + if credentials is None: + raise PermissionError("HTTP destination is not explicitly authorized") + return Endpoint( + f"{parsed.scheme}://{parsed.netloc}", + context, + namespace, + "http_targets", + probe.name or "", + credentials=credentials, + ) + if probe.mode == "node": + grant = cluster.http_discovery.nodes + if "explicit" not in self.scope.modes or grant is None: + raise PermissionError("node HTTP discovery requires an explicit grant") + if probe.name and grant.names and probe.name not in grant.names: + raise PermissionError("node is outside the discovery grant") + candidates = [] + names = (probe.name,) if probe.name else (grant.names or (None,)) + for name in names: + for node in self.objects(context, None, "nodes", name): + if grant.names and node.metadata.name not in grant.names: + raise PermissionError("node is outside the discovery grant") + for address in node.status.get("addresses", []): + if address.get("type") not in grant.address_types: + continue + host = hostname(address["address"]) + credentials = self.explicit( + host, context, None, "nodes", node.metadata.name + ) + if credentials is None and grant.accepts(node.metadata.name, host): + credentials = grant.credentials + if credentials is not None: + authority = f"[{host}]" if ":" in host else host + scheme = "https" if probe.scheme == "auto" else probe.scheme + candidates.append( + Endpoint( + f"{scheme}://{authority}:{probe.port or 10250}", + context, + None, + "nodes", + node.metadata.name, + node.metadata.uid, + credentials, + ) + ) + return self.unique(candidates) + if probe.mode == "route": + candidates = [] + for grant in cluster.http_discovery.routes: + if "explicit" not in self.scope.modes or grant.namespace not in probe.namespaces: + continue + if probe.name and grant.names and probe.name not in grant.names: + raise PermissionError("route is outside the discovery grant") + names = (probe.name,) if probe.name else (grant.names or (None,)) + for name in names: + for route in self.objects(context, grant.namespace, "routes", name): + if grant.names and route.metadata.name not in grant.names: + continue + if probe.hint and not any( + hint.lower() in route.metadata.name.lower() + for hint in probe.hint.split("|") + ): + continue + host = hostname(route.spec.get("host", "")) + credentials = self.explicit( + host, context, grant.namespace, "routes", route.metadata.name + ) + if credentials is None and grant.accepts(route.metadata.name, host): + credentials = grant.credentials + if credentials is None: + raise PermissionError("discovered route destination is not authorized") + candidates.append( + Endpoint( + f"https://{host}", + context, + grant.namespace, + "routes", + route.metadata.name, + route.metadata.uid, + credentials, + ) + ) + if candidates: + return self.unique(candidates) + return self.service(context, probe) + + @staticmethod + def unique(candidates: list[Endpoint]) -> Endpoint: + if len(candidates) != 1: + raise RuntimeError( + f"HTTP target resolution requires one endpoint, found {len(candidates)}" + ) + return candidates[0] + + def service(self, context: str, probe: HttpProbe) -> Endpoint: + candidates = [] + for namespace in probe.namespaces: + for service in self.objects(context, namespace, "services", probe.name): + if probe.hint and not any( + hint.lower() in service.metadata.name.lower() for hint in probe.hint.split("|") + ): + continue + ports = service.spec.get("ports", []) + for port in ports: + remote = port.get("port") + if probe.port and probe.port not in {remote, port.get("targetPort")}: + continue + if not isinstance(remote, int) or not 1 <= remote <= 65535: + raise ValueError("invalid Service port") + if service.spec.get("type") == "ExternalName" or not service.spec.get( + "selector" + ): + raise PermissionError("port-forward requires a selector-backed Service") + candidates.append( + Endpoint( + "http://127.0.0.1", + context, + namespace, + "services", + service.metadata.name, + service.metadata.uid, + False, + remote, + ) + ) + endpoint = self.unique(candidates) + method_verb = { + "GET": "get", + "HEAD": "get", + "OPTIONS": "get", + "POST": "create", + "PUT": "update", + "PATCH": "patch", + "DELETE": "delete", + }[probe.method] + for verb in ("port-forward+http", "port-forward", method_verb): + self.authorize(context, endpoint.namespace, "services", endpoint.name, verb) + service = self.objects(context, endpoint.namespace, "services", endpoint.name)[0] + selector = service.spec.get("selector", {}) + pods = [ + pod + for pod in self.objects(context, endpoint.namespace, "pods") + if selector + and all(pod.metadata.labels.get(key) == value for key, value in selector.items()) + ] + running = [pod for pod in pods if pod.status.get("phase") == "Running"] + if not running: + raise RuntimeError("no running selector-matched Pod for port-forward") + pod = sorted(running, key=lambda item: item.metadata.name)[0] + target_ports = [ + port.get("targetPort", port.get("port")) + for port in service.spec.get("ports", []) + if port.get("port") == endpoint.remote_port + ] + remote = next(iter(target_ports), None) + if isinstance(remote, str): + matches = [ + port.get("containerPort") + for container in pod.spec.get("containers", []) + for port in container.get("ports", []) + if port.get("name") == remote + ] + if len(matches) != 1: + raise RuntimeError("cannot resolve named target port") + remote = matches[0] + if not isinstance(remote, int) or not 1 <= remote <= 65535: + raise ValueError("invalid Pod target port") + self.authorize(context, endpoint.namespace, "pods", pod.metadata.name, "port-forward") + selected = next( + port for port in service.spec["ports"] if port["port"] == endpoint.remote_port + ) + scheme = probe.scheme + if scheme == "auto": + scheme = ( + "https" + if selected.get("appProtocol") == "https" + or "https" in selected.get("name", "") + or selected["port"] == 443 + else "http" + ) + tunnel_host = ( + f"{endpoint.name}.{endpoint.namespace}.svc" if scheme == "https" else "127.0.0.1" + ) + return Endpoint( + f"{scheme}://{tunnel_host}", + context, + endpoint.namespace, + "pods", + pod.metadata.name, + pod.metadata.uid, + self.scope.clusters[context].http_discovery.port_forward_credentials, + remote, + ) diff --git a/harnessing/5-validate/validate-findings/http_policy_io.py b/harnessing/5-validate/validate-findings/http_policy_io.py new file mode 100644 index 0000000..9cac4fb --- /dev/null +++ b/harnessing/5-validate/validate-findings/http_policy_io.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import datetime +import shutil +import subprocess +from pathlib import Path + +import yaml +from pydantic import BaseModel, Field + +if __package__: + from .http_scope import HttpPolicy, origin +else: + from http_scope import HttpPolicy, origin + + +class ClusterConnection(BaseModel): + server: str + + +class ClusterEntry(BaseModel): + cluster: ClusterConnection + + +class KubeconfigView(BaseModel): + clusters: list[ClusterEntry] = Field(min_length=1, max_length=1) + + +def expiry_date(value: str) -> str: + try: + parsed = datetime.date.fromisoformat(value) + except ValueError: + try: + parsed = datetime.datetime.fromisoformat(value).date() + except ValueError as exc: + raise ValueError("engagement expiry must be an ISO date or datetime") from exc + return parsed.isoformat() + + +def configure_http_targets( + document: dict, context: str, api: str | None, policy_path: str | None +) -> None: + policy = HttpPolicy.model_validate( + yaml.safe_load(Path(policy_path).read_text()) if policy_path else {} + ) + binary = shutil.which("oc") or shutil.which("kubectl") + if not binary: + raise RuntimeError("oc or kubectl is required to resolve the engagement context") + try: + result = subprocess.run( + [binary, f"--context={context}", "config", "view", "--minify", "-o", "json"], + capture_output=True, + text=True, + timeout=15, + check=True, + ) + actual = KubeconfigView.model_validate_json(result.stdout).clusters[0].cluster.server + except (subprocess.SubprocessError, ValueError) as exc: + raise RuntimeError("cannot resolve the selected kubeconfig context") from exc + origin(actual) + if api and origin(api) != origin(actual): + raise ValueError("requested API does not match the selected kubeconfig context") + policy.apply(document, context) + cluster = next(cluster for cluster in document["clusters"] if cluster["context"] == context) + cluster["api"] = actual diff --git a/harnessing/5-validate/validate-findings/http_scope.py b/harnessing/5-validate/validate-findings/http_scope.py new file mode 100644 index 0000000..6ce26ca --- /dev/null +++ b/harnessing/5-validate/validate-findings/http_scope.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +import ipaddress +from typing import Literal +from urllib.parse import urlsplit + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +def hostname(value: str) -> str: + if ( + not value + or any(char.isspace() for char in value) + or any(char in value for char in "*?{}\\") + ): + raise ValueError("HTTP targets require an exact hostname or HTTP(S) URL") + try: + return str(ipaddress.ip_address(value.strip("[]"))) + except ValueError: + pass + parsed = urlsplit(value if "://" in value else f"//{value}") + if parsed.scheme and parsed.scheme.lower() not in {"http", "https"}: + raise ValueError("HTTP targets require HTTP(S)") + if not parsed.hostname or parsed.username is not None or parsed.password is not None: + raise ValueError("HTTP targets cannot contain credentials or empty hosts") + if parsed.port == 0: + raise ValueError("invalid HTTP port") + try: + return str(ipaddress.ip_address(parsed.hostname)) + except ValueError: + pass + host = parsed.hostname.rstrip(".").encode("idna").decode("ascii").lower() + if any( + not label + or len(label) > 63 + or label.startswith("-") + or label.endswith("-") + or not all(char.isalnum() or char == "-" for char in label) + for label in host.split(".") + ): + raise ValueError("invalid hostname") + return host + + +class HttpTarget(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + host: str + context: str = Field(min_length=1) + namespace: str | None = None + resource: str | None = None + name: str | None = None + credentials: bool = False + + @field_validator("host") + @classmethod + def valid_host(cls, value: str) -> str: + return hostname(value) + + +class RouteGrant(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + namespace: str = Field(min_length=1, pattern=r"^[a-z0-9][a-z0-9-]*$") + names: tuple[str, ...] = () + domains: tuple[str, ...] = () + credentials: bool = False + + @field_validator("domains") + @classmethod + def valid_domains(cls, values: tuple[str, ...]) -> tuple[str, ...]: + return tuple(hostname(value) for value in values) + + def accepts(self, name: str, host: str) -> bool: + return (not self.names or name in self.names) and any( + host == domain or host.endswith(f".{domain}") for domain in self.domains + ) + + +class NodeGrant(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + names: tuple[str, ...] = () + networks: tuple[str, ...] = () + address_types: tuple[Literal["InternalIP", "ExternalIP"], ...] = ("InternalIP",) + credentials: bool = False + + @field_validator("networks") + @classmethod + def valid_networks(cls, values: tuple[str, ...]) -> tuple[str, ...]: + return tuple(str(ipaddress.ip_network(value)) for value in values) + + def accepts(self, name: str, address: str) -> bool: + ip = ipaddress.ip_address(address) + return (not self.names or name in self.names) and any( + ip in ipaddress.ip_network(network) for network in self.networks + ) + + +def origin(value: str) -> tuple[str, str, int]: + parsed = urlsplit(value) + host = hostname(value) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("expected an HTTP(S) origin") + return parsed.scheme, host, parsed.port or (443 if parsed.scheme == "https" else 80) + + +class HttpDiscovery(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + routes: tuple[RouteGrant, ...] = () + nodes: NodeGrant | None = None + port_forward_credentials: bool = False + kube_env: tuple[str, ...] = () + ca_bundle: str | None = None + insecure_tls: bool = False + session_check_path: str | None = None + + @field_validator("session_check_path") + @classmethod + def valid_session_check(cls, value: str | None) -> str | None: + if value is not None and ( + not value.startswith("/") + or value.startswith("//") + or any(ord(char) < 32 for char in value) + or "\\" in value + ): + raise ValueError("session check must be an origin-relative path") + return value + + +class HttpPolicy(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + http_targets: tuple[HttpTarget, ...] = () + http_discovery: HttpDiscovery = Field(default_factory=HttpDiscovery) + + def apply(self, document: dict, context: str) -> None: + cluster = next(cluster for cluster in document["clusters"] if cluster["context"] == context) + if any(target.context != context for target in self.http_targets): + raise ValueError("HTTP policy targets must belong to the selected context") + cluster["http_discovery"] = self.http_discovery.model_dump(mode="json") + document["http_targets"] = [target.model_dump(mode="json") for target in self.http_targets] + + +def configure_http_targets( + document: dict, context: str, api: str | None, policy_path: str | None +) -> None: + if __package__: + from .http_policy_io import configure_http_targets as configure + else: + from http_policy_io import configure_http_targets as configure + configure(document, context, api, policy_path) diff --git a/harnessing/5-validate/validate-findings/plan.py b/harnessing/5-validate/validate-findings/plan.py index 50652c8..60d3024 100644 --- a/harnessing/5-validate/validate-findings/plan.py +++ b/harnessing/5-validate/validate-findings/plan.py @@ -862,196 +862,44 @@ def _http_adapted_step( if cs and cs.explicit_namespaces else [p for p in (cs.namespaces if cs else []) if "*" not in p] ) - ns = nss[0] if nss else "" - ns_list = " ".join(nss) or ns - hint = _svc_hint(f) - # H4: hostNetwork-only listener — port-forward targets the pod - # netns, so a host-namespace socket isn't reachable that way. - # Probe from the node itself via ``oc debug node``. - # r4 refinement: trigger ONLY on the TITLE (descriptions of CSI - # findings routinely mention "hostNetwork: true on the node - # DaemonSet" as context while the claim is about the controller's - # in-pod listener — H4 v1 hijacked 10 such findings) AND only - # when a concrete port was extracted (defaulting to 8080 is - # never right; it produced connection-refused on every node). - if port and re.search( - r"(?i)\bhostNetwork\b|node IP|node.s primary interface" - r"|bound (?:directly )?on the node", - f.title or "", - ): - hp = port - cmd = ( - f"NODE=$(oc --context {ctx} get node -o name | head -1); " - f"oc --context {ctx} debug $NODE -- chroot /host " - f"curl -sS --max-time 10 http://127.0.0.1:{hp}{path} " - f"-w '\\nvf-http-status:%{{http_code}}'" - ) - return Step( - id=sid, - technique="adapted", - adapter="k8s", - verb="port-forward+http", - finding_ref=f.id, - target={"context": ctx, "namespace": ns, "method": method, "path": path, "port": hp}, - cmd=cmd, - classification="mutating", - expected=f.attack_pattern or f.title, - rollback="# debug pod auto-removed on exit", - summary=f"hostNetwork node-debug probe {method} :{hp}{path}", - ) - # H1: admin/debug paths and the monitoring CO are NOT router- - # exposed — Route-first selection hits the HAProxy "503 Application - # is not available" page for all 11 monitoring findings. Skip the - # Route loop entirely and go straight to port-forward against a - # Service whose name matches the source-repo hint. - skip_route = bool( - re.search(r"^/debug/|^/-/|/api/v\d+/admin", path) or "openshift-monitoring" in nss - ) - # r4 refinement: the static _SVC_HINT_RE matched generic words - # ("router", "webhook") that appear in finding text as feature - # names, not service names. When skip_route is set (port-forward- - # only path), the repo-derived hint is the right selector - # (prometheus → "prometheus", thanos → "thanos", etc.); the - # generic regex hint sent every monitoring probe to - # vf-no-svc-in-scope:router. Same for console findings: the - # title's "webhook" is a feature, the svc is "console". - if skip_route or hint in ("webhook", "router", "kube-rbac-proxy"): - hint = _svc_hint(f, repo_only=True) - # Route-first (core-OCP web components are Route-exposed); fall back to - # an inline port-forward against the first Service in the namespace. - # The bearer token is a short-lived low-privilege one — matches the - # "any authenticated user" precondition typical of these findings. - # - # CSRF round-trip: prime a cookie jar with GET / (Authorization header - # set so console's auth handler issues the csrf-token cookie), then - # replay the probe with -b jar + X-CSRFToken header. Harmless on - # endpoints without CSRF (header ignored, jar empty). Avoids the - # false ``refuted`` from `403 invalid CSRFToken` seen in the - # core-ocp-4.22 console pilot. - probe = ( - "JAR=$(mktemp); " - 'curl -sk -c "$JAR" "$BASE/" -o /dev/null; ' - r'CSRF=$(awk "/[Cc][Ss][Rr][Ff]/ {print \$NF; exit}" "$JAR"); ' - 'rm -f "$JAR"; ' - # Send ONLY the csrf cookie — replaying the full jar can include an - # anonymous session cookie that overrides the bearer header - # (observed on openshift-console: full-jar → 401, csrf-only → auth - # falls through to bearer). - # --compressed: some endpoints (oauth-apiserver) return gzip and - # the adapter chokes on the raw bytes. - f'curl -sk --compressed -X {method} "$BASE{path}" ' - ' -H "Authorization: Bearer $T" ' - ' -H "Cookie: csrf-token=$CSRF" ' - ' -H "X-CSRFToken: $CSRF" -H "X-CSRF-Token: $CSRF" ' - " -w '\\nvf-http-status:%{http_code}'" + if cs: + nss = sorted(set(nss) | {grant.namespace for grant in cs.http_discovery.routes}) + nss = [namespace for namespace in nss if not any(char in namespace for char in "*?[]")] + ns = nss[0] if nss else None + if __package__: + from .http_endpoints import HttpProbe + else: + from http_endpoints import HttpProbe + + namespaces = tuple( + namespace for namespace in nss if not any(char in namespace for char in "*?[]") ) - # Token: for /metrics paths use the prometheus-k8s SA (it has the - # ClusterRole that kube-rbac-proxy SARs against — kube:admin OAuth - # paradoxically fails SAR with ``resource=`` empty). Otherwise - # prefer VF_OAUTH_TOKEN, fall back to a low-priv SA token. - token_block = ( - ( - f'T="$(oc --context {ctx} -n openshift-monitoring ' - f'create token prometheus-k8s --duration=10m 2>/dev/null)"; ' - f'[ -z "$T" ] && ' - if path.rstrip("/").endswith("/metrics") - else "" + node = bool( + port + and re.search( + r"(?i)hostNetwork|node IP|node.s primary interface|bound (?:directly )?on the node", + f"{f.title or ''} {f.description or ''}" + if re.search(r"(?i)\bkubelet\b", f.title or "") + else f.title or "", ) - + f'T="${{T:-${{VF_OAUTH_TOKEN:-$(oc --context {ctx} -n {ns} ' - f'create token default --duration=10m 2>/dev/null)}}}}"; ' ) - # Route selection — by component-name hint only. H5: NO first-route - # fallback when $HINT is non-empty (a wrong-component Route produced - # 3 false-confirms in storage/* — INCONCLUSIVE-FIX-PLAN §H5). - route_block = ( - f"for N in {ns_list}; do " - f" R=$(oc --context {ctx} -n $N get route " - f" -o jsonpath='{{range .items[*]}}{{.metadata.name}} " - f'{{.spec.host}}{{"\\n"}}{{end}}\' 2>/dev/null); ' - f' [ -z "$R" ] && continue; ' - f' if [ -n "$HINT" ]; then ' - f' H=$(echo "$R" | grep -iE -- "$HINT" | head -1 ' - f"| awk '{{print $2}}'); " - f' else H=$(echo "$R" | head -1 ' - f"| awk '{{print $2}}'); fi; " - f' [ -n "$H" ] && break; ' - f"done; " - ) - # Service selection + port-forward. H3: when the finding names a - # specific port, forward to THAT containerPort (kubectl resolves - # svc→pod targetPort by number). H5: NO first-svc fallback when - # $HINT is set — emit ``vf-no-svc-in-scope:$HINT`` instead so the - # verdict layer can route to infra/manual instead of a junk 404. - # r4: emit one line per (svc, port, targetPort) so H3's port-match - # scans every exposed port (prometheus-k8s exposes 9091/9092 but the - # claim is :9090 → previously no-match → vf-no-svc-in-scope). - # go-template is used (jsonpath has no nested-range variable bind). - # When a port is named, port-forward to the POD (selector-matched) - # so a containerPort that isn't surfaced in the Service still works. - # Probe https first, retry plain http on TLS-handshake/000 — most - # pprof/admin/gossip listeners are plaintext-only. - svc_tpl = ( - r"{{range .items}}{{$n:=.metadata.name}}" - r"{{range .spec.ports}}" - r'{{$n}} {{.port}} {{.targetPort}}{{"\n"}}' - r"{{end}}{{end}}" + service = bool( + re.search(r"^/debug/|^/-/|/api/v\d+/admin", path) or "openshift-monitoring" in namespaces ) - svc_block = ( - f"for N in {ns_list}; do " - f" SV=$(oc --context {ctx} -n $N get svc " - f" -o go-template='{svc_tpl}' 2>/dev/null); " - f' [ -z "$SV" ] && continue; ' - + ( - f" L=$(echo \"$SV\" | awk -v p={port} '$2==p||$3==p{{print; exit}}'); " - if port - else " L=; " - ) - + ' [ -z "$L" ] && [ -n "$HINT" ] && ' - ' L=$(echo "$SV" | grep -iE -- "$HINT" | head -1); ' - ' [ -z "$L" ] && [ -z "$HINT" ] && ' - ' L=$(echo "$SV" | head -1); ' - " S=$(echo $L | awk '{print $1}'); " - " P=$(echo $L | awk '{print $2}'); " - ' [ -n "$S" ] && break; ' - "done; " - + ( - ( - f"P={port}; " - f"SEL=$(oc --context {ctx} -n $N get svc $S " - f"-o go-template=" - f"'{{{{range $k,$v := .spec.selector}}}}" - f"{{{{$k}}}}={{{{$v}}}},{{{{end}}}}' 2>/dev/null " - f"| sed 's/,$//'); " - f'POD=$(oc --context {ctx} -n $N get pod -l "$SEL" ' - f"-o name 2>/dev/null | head -1); " - f'TGT="${{POD:-svc/$S}}"; ' - ) - if port - else 'TGT="svc/$S"; ' - ) - + f'if [ -z "$S" ]; then ' - f' echo "vf-no-svc-in-scope:$HINT"; ' - f" printf '\\nvf-http-status:000'; " - f"else " - f' oc --context {ctx} -n $N port-forward "$TGT" ' - f' 18443:"$P" >/dev/null 2>&1 & PF=$!; sleep 2; ' - f' OUT=$(BASE="https://127.0.0.1:18443"; {probe}); ' - f' case "$OUT" in *vf-http-status:000*) ' - f' OUT=$(BASE="http://127.0.0.1:18443"; {probe});; esac; ' - f' echo "$OUT"; ' - f" kill $PF 2>/dev/null; " - f"fi" + hint = _svc_hint(f) + if service or hint in {"webhook", "router", "kube-rbac-proxy"}: + hint = _svc_hint(f, repo_only=True) + probe = HttpProbe( + mode="node" if node else "service" if service else "route", + method=method, + path=path, + port=port, + namespaces=namespaces, + hint=hint, + authenticate=False, + csrf=False, + service_account="prometheus-k8s" if path.rstrip("/").endswith("/metrics") else None, ) - if skip_route: - cmd = f'HINT="{hint}"; H=; S=; ' + token_block + svc_block - else: - cmd = ( - f'HINT="{hint}"; H=; S=; ' - + route_block - + token_block - + f'if [ -n "$H" ]; then BASE="https://$H"; {probe}; ' - f"else " + svc_block + "; fi" - ) return Step( id=sid, technique="adapted", @@ -1060,17 +908,21 @@ def _http_adapted_step( finding_ref=f.id, target={ "context": ctx, - "namespace": ns, + "namespace": None if node else ns, + "resource": "nodes" if node else "services" if service else "routes", "method": method, "path": path, **({"port": port} if port else {}), + "http": probe.model_dump(mode="json"), }, - cmd=cmd, + cmd=None, + classification="destructive" + if method == "DELETE" + else "mutating" + if method in {"POST", "PUT", "PATCH"} + else "safe", expected=f.attack_pattern or f.title, - summary=f"HTTP probe {method} {path}" - + (f" :{port}" if port else "") - + (" [pf-only]" if skip_route else "") - + f" (adapted from {','.join(sorted(set(f.cwes) & HTTP_CWES))})", + summary=f"HTTP probe {method} {path} ({probe.mode})", ) @@ -1092,13 +944,7 @@ def _spoof_adapted_step(sid: str, f: Finding, scope: Scope, tm) -> Step | None: s = _http_adapted_step(sid, f, scope, tm, force_coords=(coords[0], coords[1])) if not s: return None - inject = " ".join(f'-H "{h}: vf-spoof-canary"' for h in dict.fromkeys(hdrs)) - s.cmd = s.cmd.replace( - '-H "Authorization: Bearer $T"', f'-H "Authorization: Bearer $T" {inject}', 1 - ) - s.cmd = s.cmd.replace( - '-H "Authorization: Bearer $T"', f'-H "Authorization: Bearer $T" {inject}', 1 - ) + s.target["http"]["headers"] = {header: "vf-spoof-canary" for header in dict.fromkeys(hdrs)} s.expected = f"upstream reflects/honours spoofed {hdrs[0]} — {f.attack_pattern or f.title}" s.summary = f"header-spoof probe {','.join(dict.fromkeys(hdrs))} ({','.join(f.cwes)})" return s diff --git a/harnessing/5-validate/validate-findings/scope.py b/harnessing/5-validate/validate-findings/scope.py index 2dec8eb..6750be2 100644 --- a/harnessing/5-validate/validate-findings/scope.py +++ b/harnessing/5-validate/validate-findings/scope.py @@ -21,6 +21,11 @@ from dataclasses import asdict, dataclass, field from pathlib import Path +if __package__: + from .http_scope import HttpDiscovery, HttpTarget +else: + from http_scope import HttpDiscovery, HttpTarget + try: import yaml except ImportError: # pragma: no cover @@ -146,6 +151,7 @@ class ClusterScope: namespaces: list[str] = field(default_factory=list) # globs OK; [] => none verbs_denied: list[str] = field(default_factory=list) explicit_namespaces: set[str] = field(default_factory=set) # mode-1 literal entries + http_discovery: HttpDiscovery = field(default_factory=HttpDiscovery) def ns_allowed(self, ns: str | None) -> bool: if ns is None: @@ -194,6 +200,7 @@ class Scope: expires: _dt.date | None = None environment: str | None = None clusters: dict[str, ClusterScope] = field(default_factory=dict) + http_targets: tuple[HttpTarget, ...] = () containers: list[str] = field(default_factory=list) # name globs container_runtimes: list[str] = field(default_factory=list) wasm_artifacts: list[str] = field(default_factory=list) @@ -233,8 +240,14 @@ def from_targets_file(cls, path: str | Path) -> Scope: namespaces=ns, verbs_denied=[str(v) for v in c.get("verbs_denied", [])], explicit_namespaces=set(ns), + http_discovery=HttpDiscovery.model_validate(c.get("http_discovery", {})), ) s.clusters[cs.context] = cs + s.http_targets = tuple( + HttpTarget.model_validate(target) for target in data.get("http_targets", []) + ) + if any(target.context not in s.clusters for target in s.http_targets): + raise ValueError("HTTP target context must be explicitly declared in clusters") for c in data.get("containers", []): s.container_runtimes.append(c.get("runtime", "podman")) s.containers.extend(c.get("name_patterns", [])) @@ -298,6 +311,18 @@ def merge_inferred(self, inferred: dict) -> Scope: # ----- evaluation --------------------------------------------------- + def curl_hosts(self) -> tuple[str, ...]: + """Endpoints safe_exec may let curl reach, for its host allowlist. + + `clusters[].api` is per-engagement, so it cannot live in + safe-exec-profiles.yaml; passing it at the call site is what makes a + restricted profile usable. Entries are returned + verbatim: safe_exec normalizes URL and host:port forms to a hostname + itself, and refuses entries that do not, which a local pass would + mask.""" + apis = {str(c.api).strip() for c in self.clusters.values() if c.api} + return tuple(sorted(apis | {target.host for target in self.http_targets})) + @property def binding_mode(self) -> str: if not self.modes: @@ -317,6 +342,14 @@ def _control_plane_locked(self, a: Action) -> str | None: # else on a control-plane cluster resource needs mode-1 "*". if a.resource in CONTROL_PLANE_CLUSTER_RESOURCES and a.verb not in READONLY_VERBS: cs = self.clusters.get(a.context) or self.clusters.get("__current__") + if ( + cs + and "explicit" in self.modes + and cs.http_discovery.nodes is not None + and a.resource == "nodes" + and a.verb == "port-forward+http" + ): + return None if cs and "*" in cs.explicit_namespaces: return None return ( @@ -354,7 +387,19 @@ def is_in_scope(self, a: Action) -> tuple[bool, str]: return False, f"context '{a.context}' not in scope" if a.verb in cs.verbs_denied: return False, f"verb '{a.verb}' denied for context '{cs.context}'" - if not cs.ns_allowed(a.namespace): + node_discovery = ( + "explicit" in self.modes + and cs.http_discovery.nodes is not None + and a.resource == "nodes" + and a.verb in {"get", "list", "port-forward+http"} + and a.namespace is None + and ( + not cs.http_discovery.nodes.names + or a.name in cs.http_discovery.nodes.names + or (a.verb == "port-forward+http" and a.name is None) + ) + ) + if not node_discovery and not cs.ns_allowed(a.namespace): return False, f"namespace '{a.namespace}' not in scope for context '{cs.context}'" if ( a.image @@ -445,9 +490,11 @@ def to_json(self) -> str: "api": v.api, "namespaces": v.namespaces, "verbs_denied": v.verbs_denied, + "http_discovery": v.http_discovery.model_dump(mode="json"), } for k, v in self.clusters.items() }, + "http_targets": [target.model_dump(mode="json") for target in self.http_targets], "containers": self.containers, "wasm_artifacts": self.wasm_artifacts, "images": self.images, diff --git a/harnessing/5-validate/validate-findings/targets.example.yaml b/harnessing/5-validate/validate-findings/targets.example.yaml index 58e5210..d6dbd88 100644 --- a/harnessing/5-validate/validate-findings/targets.example.yaml +++ b/harnessing/5-validate/validate-findings/targets.example.yaml @@ -16,16 +16,37 @@ expires: 2026-07-01 # harness aborts if today > expires # --- Kubernetes / OpenShift --------------------------------------------- clusters: - context: lab-hub # kubeconfig context name - api: https://api.hub.lab.example:6443 # informational only + api: https://api.hub.lab.example:6443 # also seeds safe_exec's curl host allowlist namespaces: - ramen-ops - ramen-system - openshift-dr-* # globs OK verbs_denied: [delete] # even in-scope, never delete + http_discovery: + routes: + - namespace: ramen-system + names: [ramen] + domains: [apps.hub.lab.example] + credentials: false + nodes: + names: [worker-0] + networks: [10.0.128.0/24, "fd00:128::/64"] + address_types: [InternalIP] + credentials: false + port_forward_credentials: false + kube_env: [] - context: lab-spoke-1 namespaces: ["*"] # full namespace access on spoke +http_targets: + - context: lab-hub + host: https://ramen.apps.hub.lab.example + namespace: ramen-system + resource: routes + name: ramen + credentials: false + # --- Running containers (podman/docker) --------------------------------- containers: - runtime: podman diff --git a/src/traust/cli/groups/util.py b/src/traust/cli/groups/util.py index 66de842..e3cebf7 100644 --- a/src/traust/cli/groups/util.py +++ b/src/traust/cli/groups/util.py @@ -141,6 +141,15 @@ def add_safe_exec_args(ap) -> None: ) p.add_argument("--timeout", type=int, default=120) p.add_argument("--cwd", default=None) + p.add_argument( + "--allowed-host", + action="append", + default=[], + metavar="HOST", + dest="allowed_hosts", + help="extra curl host for this invocation (repeatable); unions " + "with the profile's curl_allowed_hosts", + ) p.add_argument("cmd", nargs="*", help="argv form (after --)") sub.add_parser("list-profiles", help="list configured safe_exec profiles") @@ -149,9 +158,15 @@ def call_safe_exec(engine, args) -> int: profile_map = engine.adapters.safe_exec_profile_map() if args.safe_exec_mode == "list-profiles": for name, profile in sorted(profile_map.items()): + hosts = ( + f"{len(profile.curl_allowed_hosts)} listed" + if profile.curl_allowed_hosts + else "none listed" + ) print( - f"{name:18s} allow={sorted(profile.allow)} " - f"pipelines={profile.allow_pipelines} — {profile.description}" + f"{name:18s} posture={profile.posture} allow={sorted(profile.allow)} " + f"pipelines={profile.allow_pipelines} " + f"curl_hosts={hosts} — {profile.description}" ) return 0 @@ -160,13 +175,18 @@ def call_safe_exec(engine, args) -> int: print("pass either --string or argv, not both", file=sys.stderr) return 2 cmd = args.string if args.string is not None else args.cmd + allowed_hosts = tuple(args.allowed_hosts or ()) if isinstance(cmd, str): verdict = safe_exec.vet_command_string( - cmd, safe_exec.get_profile(args.profile, profile_map=profile_map) + cmd, + safe_exec.get_profile(args.profile, profile_map=profile_map), + allowed_hosts=allowed_hosts, ) else: verdict = safe_exec.validate_argv( - list(cmd), safe_exec.get_profile(args.profile, profile_map=profile_map) + list(cmd), + safe_exec.get_profile(args.profile, profile_map=profile_map), + allowed_hosts=allowed_hosts, ) if args.safe_exec_mode == "check": @@ -194,6 +214,7 @@ def call_safe_exec(engine, args) -> int: cwd=args.cwd, honor_bypass=True, profile_map=profile_map, + allowed_hosts=allowed_hosts, ) sys.stdout.write(out or "") sys.stderr.write(err or "") diff --git a/tests/test_http_endpoints.py b/tests/test_http_endpoints.py new file mode 100644 index 0000000..6eca5f2 --- /dev/null +++ b/tests/test_http_endpoints.py @@ -0,0 +1,597 @@ +from __future__ import annotations + +import datetime +import json +from pathlib import Path +from unittest.mock import Mock + +import pytest +import yaml +from adapters.http import HttpExecutor +from adapters.k8s import K8sAdapter +from execute import AuditLog, run +from http_endpoints import EndpointResolver, HttpProbe +from http_scope import HttpDiscovery, HttpTarget, hostname +from scope import ClusterScope, OffLimit, Scope +from traust_engine._util.safe_exec import Profile + + +@pytest.fixture +def scope() -> Scope: + return Scope( + modes={"explicit"}, + clusters={ + "lab": ClusterScope( + context="lab", + namespaces=["app"], + explicit_namespaces={"app"}, + http_discovery=HttpDiscovery.model_validate( + { + "routes": [ + { + "namespace": "app", + "domains": ["apps.lab.example"], + "credentials": True, + } + ], + "nodes": {"networks": ["10.0.0.0/24", "fd00::/64"]}, + } + ), + ) + }, + ) + + +@pytest.fixture +def route() -> dict: + return { + "metadata": {"name": "console", "namespace": "app", "uid": "route-1"}, + "spec": {"host": "console.apps.lab.example"}, + } + + +def runner(items: list[dict]) -> Mock: + return Mock( + side_effect=[(0, "https://api.lab.example:6443", ""), (0, json.dumps({"items": items}), "")] + ) + + +@pytest.mark.parametrize( + "value,expected", + [ + ("https://EXAMPLE.COM:443/path", "example.com"), + ("https://[fd00::1]:10250/pods", "fd00::1"), + ("[::1]", "::1"), + ("fd00::2", "fd00::2"), + ], +) +def test_host_normalization(value: str, expected: str) -> None: + assert hostname(value) == expected + + +@pytest.mark.parametrize( + "value", + [ + "", + "*.example.com", + "https://user@example.com", + "file:///etc/passwd", + "example.com:99999", + "bad host", + "evil.example\\@good.example", + ], +) +def test_invalid_hosts(value: str) -> None: + with pytest.raises(ValueError): + hostname(value) + + +def test_explicit_targets_roundtrip(tmp_path: Path) -> None: + path = tmp_path / "targets.yaml" + path.write_text( + yaml.safe_dump( + { + "clusters": [{"context": "lab", "namespaces": ["app"]}], + "http_targets": [{"context": "lab", "host": "https://[fd00::1]:10250"}], + } + ) + ) + loaded = Scope.from_targets_file(path) + assert loaded.curl_hosts() == ("fd00::1",) + assert json.loads(loaded.to_json())["http_targets"][0]["host"] == "fd00::1" + loaded.merge_inferred({"http_targets": [{"context": "lab", "host": "evil.example"}]}) + assert loaded.curl_hosts() == ("fd00::1",) + + +def test_route_discovery(scope: Scope, route: dict) -> None: + command = runner([route]) + endpoint = EndpointResolver(scope, command, "oc").resolve("lab", HttpProbe(namespaces=("app",))) + assert endpoint.host == "console.apps.lab.example" + assert endpoint.credentials + assert endpoint.uid == "route-1" + assert "--context=lab" in command.call_args.args[0] + + +@pytest.mark.parametrize("host", ["evil.example", "apps.lab.example.evil.example"]) +def test_route_destination_is_not_a_grant(scope: Scope, route: dict, host: str) -> None: + route["spec"]["host"] = host + with pytest.raises(PermissionError, match="not authorized"): + EndpointResolver(scope, runner([route]), "oc").resolve( + "lab", HttpProbe(namespaces=("app",)) + ) + + +def test_explicit_route_exception(scope: Scope, route: dict) -> None: + route["spec"]["host"] = "special.example" + scope.http_targets = (HttpTarget(host="special.example", context="lab", namespace="app"),) + endpoint = EndpointResolver(scope, runner([route]), "oc").resolve( + "lab", HttpProbe(namespaces=("app",)) + ) + assert not endpoint.credentials + + +def test_denied_namespace_is_never_queried(scope: Scope) -> None: + scope.off_limits = [OffLimit(namespace="app")] + command = runner([]) + with pytest.raises(PermissionError): + EndpointResolver(scope, command, "oc").resolve("lab", HttpProbe(namespaces=("app",))) + assert command.call_count == 1 + + +def test_denied_route_name(scope: Scope, route: dict) -> None: + scope.off_limits = [OffLimit(resource="routes", name="console")] + command = runner([route]) + with pytest.raises(PermissionError): + EndpointResolver(scope, command, "oc").resolve("lab", HttpProbe(namespaces=("app",))) + assert command.call_count == 1 + + +@pytest.mark.parametrize("address", ["10.0.0.2", "fd00::2"]) +def test_node_addresses(scope: Scope, address: str) -> None: + node = { + "metadata": {"name": "worker"}, + "status": {"addresses": [{"type": "InternalIP", "address": address}]}, + } + endpoint = EndpointResolver(scope, runner([node]), "oc").resolve("lab", HttpProbe(mode="node")) + assert endpoint.host == address + assert not endpoint.credentials + assert scope.clusters["lab"].namespaces == ["app"] + + +def test_nodes_need_explicit_grant(scope: Scope) -> None: + scope.clusters["lab"].http_discovery = HttpDiscovery() + with pytest.raises(PermissionError, match="explicit grant"): + EndpointResolver(scope, runner([]), "oc").resolve("lab", HttpProbe(mode="node")) + + +def test_api_mismatch(scope: Scope) -> None: + scope.clusters["lab"].api = "https://other.example:6443" + command = runner([]) + with pytest.raises(PermissionError, match="does not match"): + EndpointResolver(scope, command, "oc").resolve("lab", HttpProbe(mode="node")) + assert command.call_count == 1 + + +def test_expired_no_queries(scope: Scope) -> None: + scope.expires = datetime.date.today() - datetime.timedelta(days=1) + command = runner([]) + with pytest.raises(PermissionError, match="expired"): + EndpointResolver(scope, command, "oc").resolve("lab", HttpProbe(mode="node")) + command.assert_not_called() + + +@pytest.mark.parametrize("response", ["not json", '{"items": {}}', '{"items": [{"spec": {}}]}']) +def test_malformed_discovery(scope: Scope, response: str) -> None: + command = Mock(side_effect=[(0, "https://api.lab.example", ""), (0, response, "")]) + with pytest.raises(ValueError, match="discovery response"): + EndpointResolver(scope, command, "oc").resolve("lab", HttpProbe(namespaces=("app",))) + + +def test_ambiguous_routes(scope: Scope, route: dict) -> None: + with pytest.raises(RuntimeError, match="found 2"): + EndpointResolver(scope, runner([route, route]), "oc").resolve( + "lab", HttpProbe(namespaces=("app",)) + ) + + +def restricted() -> dict[str, Profile]: + return { + "validation-step": Profile( + name="validation-step", + description="test", + allow=frozenset({"curl", "oc", "kubectl"}), + allowed_path_heads=frozenset(), + allow_pipelines=True, + keep_env=(), + keep_env_heads=("curl", "oc", "kubectl"), + posture="restricted", + ) + } + + +def test_executor_without_preflight( + scope: Scope, route: dict, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + calls = runner([route]) + monkeypatch.setattr(K8sAdapter, "_run", calls) + from traust_engine._util import safe_exec + + execute = Mock(return_value=(0, "ok\nvf-http-status:200", "")) + monkeypatch.setattr(safe_exec, "run_segments", execute) + path = tmp_path / "plan.json" + path.write_text( + json.dumps( + { + "steps": [ + { + "id": "http", + "adapter": "k8s", + "verb": "port-forward+http", + "target": { + "context": "lab", + "namespace": "app", + "resource": "routes", + "http": {"namespaces": ["app"], "path": "/healthz"}, + }, + } + ] + } + ) + ) + results, _ = run(path, scope, tmp_path, profile_map=restricted()) + assert results[0].verdict != "blocked_by_scope" + assert execute.call_count == 1 + assert "https://console.apps.lab.example/healthz" in execute.call_args.args[0][0] + assert "http-endpoint" in (tmp_path / "validation-audit.jsonl").read_text() + + +def test_credentials_require_separate_grant(scope: Scope, route: dict, tmp_path: Path) -> None: + scope.clusters["lab"].http_discovery = HttpDiscovery.model_validate( + {"routes": [{"namespace": "app", "domains": ["apps.lab.example"]}]} + ) + adapter = K8sAdapter() + adapter._run = runner([route]) + executor = HttpExecutor(adapter, scope, "oc", AuditLog(tmp_path / "audit.jsonl")) + with pytest.raises(PermissionError, match="credentials"): + executor.execute("lab", HttpProbe(namespaces=("app",), authenticate=True)) + + +@pytest.mark.parametrize("path", ["//evil.example/x", "https://evil.example/", "/x\r\nHeader: y"]) +def test_probe_paths(path: str) -> None: + with pytest.raises(ValueError): + HttpProbe(path=path) + + +@pytest.mark.parametrize( + "headers", [{"Host": "evil.example"}, {"X-Test": "a\r\nb"}, {"Authorization": "token"}] +) +def test_probe_headers(headers: dict) -> None: + with pytest.raises(ValueError): + HttpProbe(headers=headers) + + +def test_list_denied_before_discovery(scope: Scope) -> None: + scope.clusters["lab"].verbs_denied = ["list"] + command = runner([]) + with pytest.raises(PermissionError, match="list"): + EndpointResolver(scope, command, "oc").resolve("lab", HttpProbe(namespaces=("app",))) + assert command.call_count == 1 + + +def test_named_node_outside_grant_not_queried(scope: Scope) -> None: + scope.clusters["lab"].http_discovery = HttpDiscovery.model_validate( + {"nodes": {"names": ["worker"], "networks": ["10.0.0.0/24"]}} + ) + command = runner([]) + with pytest.raises(PermissionError, match="outside"): + EndpointResolver(scope, command, "oc").resolve("lab", HttpProbe(mode="node", name="other")) + assert command.call_count == 1 + + +def test_api_credentials_require_exact_origin(scope: Scope) -> None: + with pytest.raises(PermissionError): + EndpointResolver(scope, runner([]), "oc").resolve( + "lab", HttpProbe(mode="direct", url="http://api.lab.example:8080", namespaces=("app",)) + ) + + +def test_api_explicit_no_credentials_wins(scope: Scope) -> None: + scope.http_targets = (HttpTarget(context="lab", host="api.lab.example", credentials=False),) + endpoint = EndpointResolver(scope, runner([]), "oc").resolve( + "lab", + HttpProbe( + mode="direct", url="https://api.lab.example:6443", namespaces=("app",), path="/healthz" + ), + ) + assert not endpoint.credentials + + +def test_service_target_port(scope: Scope) -> None: + service = { + "metadata": {"name": "web", "namespace": "app"}, + "spec": {"ports": [{"port": 8080, "targetPort": 9090}], "selector": {"app": "web"}}, + } + pod = { + "metadata": {"name": "web-pod", "namespace": "app", "labels": {"app": "web"}}, + "status": {"phase": "Running"}, + } + command = Mock( + side_effect=[ + (0, "https://api.lab.example", ""), + (0, json.dumps({"items": [service]}), ""), + (0, json.dumps(service), ""), + (0, json.dumps({"items": [pod]}), ""), + ] + ) + endpoint = EndpointResolver(scope, command, "oc").resolve( + "lab", HttpProbe(mode="service", namespaces=("app",), port=8080, scheme="https") + ) + assert endpoint.remote_port == 9090 + assert endpoint.resource == "pods" + assert endpoint.name == "web-pod" + assert endpoint.origin.startswith("https:") + + +def test_token_subresource_deny( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from http_endpoints import Endpoint + + monkeypatch.delenv("VF_OAUTH_TOKEN", raising=False) + scope.off_limits = [OffLimit(resource="serviceaccounts/token", verb="create")] + adapter = K8sAdapter() + adapter._run = Mock() + executor = HttpExecutor(adapter, scope, "oc", AuditLog(tmp_path / "audit.jsonl")) + with pytest.raises(PermissionError, match="off_limits"): + executor.token( + Endpoint("https://web.example", "lab", "app", "routes", "web", credentials=True), + HttpProbe(authenticate=True, service_account="default"), + ) + adapter._run.assert_not_called() + + +def test_node_execution_without_wildcard( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from traust_engine._util import safe_exec + + node = { + "metadata": {"name": "worker"}, + "status": {"addresses": [{"type": "InternalIP", "address": "10.0.0.2"}]}, + } + monkeypatch.setattr(K8sAdapter, "_run", runner([node])) + execute = Mock(return_value=(0, "ok\nvf-http-status:200", "")) + monkeypatch.setattr(safe_exec, "run_segments", execute) + adapter = K8sAdapter() + adapter.bind_profile_map(restricted()) + result = adapter.execute( + { + "id": "node", + "verb": "port-forward+http", + "target": { + "context": "lab", + "resource": "nodes", + "http": {"mode": "node", "path": "/pods"}, + }, + }, + scope, + AuditLog(tmp_path / "audit.jsonl"), + tmp_path, + ) + assert result.verdict != "blocked_by_scope" + execute.assert_called_once() + + +def test_destructive_http_rechecked( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + execute = Mock() + monkeypatch.setattr(K8sAdapter, "execute", execute) + path = tmp_path / "plan.json" + path.write_text( + json.dumps( + { + "steps": [ + { + "id": "http", + "adapter": "k8s", + "verb": "port-forward+http", + "classification": "safe", + "target": { + "context": "lab", + "namespace": "app", + "resource": "routes", + "http": {"method": "DELETE", "namespaces": ["app"]}, + }, + } + ] + } + ) + ) + results, _ = run(path, scope, tmp_path, profile_map=restricted()) + assert results[0].verdict == "not_attempted" + assert results[0].classification == "destructive" + execute.assert_not_called() + + +def test_csrf_roundtrip_uses_no_files( + scope: Scope, route: dict, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("VF_OAUTH_TOKEN", "test-token") + scope.clusters["lab"].http_discovery = scope.clusters["lab"].http_discovery.model_copy( + update={"session_check_path": "/identity"} + ) + adapter = K8sAdapter() + adapter._run = runner([route]) + adapter.bind_profile_map(restricted()) + executor = HttpExecutor(adapter, scope, "oc", AuditLog(tmp_path / "audit.jsonl")) + executor.curl = Mock( + side_effect=[ + (0, "HTTP/1.1 200 OK\r\nSet-Cookie: csrf-token=canary; Secure\r\n", ""), + (0, "200", ""), + (0, "ok\nvf-http-status:200", ""), + ] + ) + executor.execute("lab", HttpProbe(namespaces=("app",), authenticate=True, csrf=True)) + first = executor.curl.call_args_list[0].args[0] + second = executor.curl.call_args_list[1].args[0] + assert "Authorization: Bearer test-token" in first + assert "Cookie: csrf-token=canary" in second + assert "-c" not in first and "-b" not in second + + +def test_tunnel_cleanup_on_error( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import os + + from adapters import http + from http_endpoints import Endpoint + + read_fd, write_fd = os.pipe() + os.write(write_fd, b"Forwarding from 127.0.0.1:34567 -> 9090\n") + os.close(write_fd) + with os.fdopen(read_fd, "rb", buffering=0) as stdout: + process = Mock(stdout=stdout) + process.poll.return_value = None + process.__enter__ = Mock(return_value=process) + process.__exit__ = Mock(return_value=False) + monkeypatch.setattr(http.subprocess, "Popen", Mock(return_value=process)) + executor = HttpExecutor(K8sAdapter(), scope, "oc", AuditLog(tmp_path / "audit.jsonl")) + endpoint = Endpoint("http://127.0.0.1", "lab", "app", "pods", "web", remote_port=9090) + with pytest.raises(RuntimeError, match="probe failed"), executor.tunnel(endpoint) as origin: + assert origin == "http://127.0.0.1:34567" + raise RuntimeError("probe failed") + process.terminate.assert_called_once() + process.wait.assert_called_once_with(timeout=3) + + +def test_generated_route_executes_under_estate_policy( + scope: Scope, route: dict, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from ingest import Finding + from plan import _http_adapted_step + from traust_contracts import SafeExecProfiles + from traust_engine._util import safe_exec + + scope.clusters["lab"].namespaces = ["*"] + scope.clusters["lab"].explicit_namespaces = {"*"} + finding = Finding( + id="F1", + title="console anonymous endpoint", + severity="high", + cwes=["CWE-306"], + description="GET /api/status exposes console state", + ) + step = _http_adapted_step("generated", finding, scope, None) + assert step.target["http"]["namespaces"] == ["app"] + assert not step.target["http"]["authenticate"] + monkeypatch.setattr(K8sAdapter, "_run", runner([route])) + execute = Mock(return_value=(0, "ok\nvf-http-status:200", "")) + monkeypatch.setattr(safe_exec, "run_segments", execute) + section = SafeExecProfiles.model_validate( + yaml.safe_load( + ( + Path(__file__).resolve().parents[1] / "config/safe-exec-profiles.example.yaml" + ).read_text() + ) + ) + profiles = safe_exec.profiles_from_section(section) + path = tmp_path / "plan.json" + path.write_text(json.dumps({"steps": [step.to_dict()]})) + results, _ = run(path, scope, tmp_path, profile_map=profiles) + assert not results[0].scope_reason + execute.assert_called_once() + + +def test_executor_instances_do_not_leak_hosts( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + instances = [] + + from adapters import StepResult + + def execute( + adapter: K8sAdapter, step: dict, current_scope: Scope, audit: AuditLog, artifacts: Path + ) -> StepResult: + + instances.append(adapter) + return StepResult( + step_id=step["id"], + adapter="k8s", + verb="get", + target=step["target"], + classification="safe", + verdict="inconclusive", + ) + + monkeypatch.setattr(K8sAdapter, "execute", execute) + scope.http_targets = (HttpTarget(context="lab", host="first.example"),) + path = tmp_path / "plan.json" + path.write_text( + json.dumps( + { + "steps": [ + { + "id": "one", + "adapter": "k8s", + "verb": "get", + "target": {"context": "lab", "namespace": "app", "resource": "pods"}, + } + ] + } + ) + ) + run(path, scope, tmp_path, profile_map=restricted()) + scope.http_targets = () + run(path, scope, tmp_path, profile_map=restricted()) + assert instances[0] is not instances[1] + assert instances[0]._curl_hosts == ("first.example",) + assert instances[1]._curl_hosts == () + + +def test_api_path_cannot_bypass_resource_scope(scope: Scope) -> None: + with pytest.raises(PermissionError, match="health and version"): + EndpointResolver(scope, runner([]), "oc").resolve( + "lab", + HttpProbe( + mode="direct", + url="https://api.lab.example:6443", + namespaces=("app",), + path="/api/v1/namespaces/other/secrets", + authenticate=True, + ), + ) + + +def test_reflected_token_redacted( + scope: Scope, route: dict, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("VF_OAUTH_TOKEN", "opaque-canary") + adapter = K8sAdapter() + adapter._run = runner([route]) + executor = HttpExecutor(adapter, scope, "oc", AuditLog(tmp_path / "audit.jsonl")) + executor.curl = Mock(return_value=(0, "opaque-canary", "opaque-canary")) + assert executor.execute("lab", HttpProbe(namespaces=("app",), authenticate=True)) == ( + 0, + "[REDACTED]", + "[REDACTED]", + ) + + +def test_authenticated_claim_not_tested_anonymously(scope: Scope, tmp_path: Path) -> None: + adapter = K8sAdapter() + adapter._run = Mock() + result = adapter.execute( + { + "id": "auth", + "verb": "port-forward+http", + "expected": "authenticated user reads another tenant", + "target": {"context": "lab", "namespace": "app", "http": {"namespaces": ["app"]}}, + }, + scope, + AuditLog(tmp_path / "audit.jsonl"), + tmp_path, + ) + assert result.verdict == "inconclusive" + adapter._run.assert_not_called() diff --git a/tests/test_http_review.py b/tests/test_http_review.py new file mode 100644 index 0000000..514ff79 --- /dev/null +++ b/tests/test_http_review.py @@ -0,0 +1,338 @@ +from __future__ import annotations + +import json +import shutil +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from unittest.mock import Mock + +import pytest +from adapters.http import HttpExecutor +from adapters.k8s import K8sAdapter +from execute import AuditLog, run +from http_endpoints import Endpoint, EndpointResolver, HttpProbe +from http_policy_io import expiry_date +from http_scope import HttpDiscovery, origin +from scope import Action, ClusterScope, Scope + +from tests.test_http_endpoints import restricted + + +@pytest.fixture +def scope() -> Scope: + return Scope( + modes={"explicit"}, + clusters={ + "lab": ClusterScope( + context="lab", + namespaces=["app"], + explicit_namespaces={"app"}, + http_discovery=HttpDiscovery.model_validate( + {"nodes": {"names": ["worker-0"], "networks": ["10.0.0.0/24"]}} + ), + ) + }, + ) + + +def test_structured_rollback_rejected( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + execute = Mock() + rollback = Mock() + monkeypatch.setattr(K8sAdapter, "execute", execute) + monkeypatch.setattr(K8sAdapter, "rollback", rollback) + plan = tmp_path / "plan.json" + plan.write_text( + json.dumps( + { + "steps": [ + { + "id": "blocked", + "adapter": "k8s", + "verb": "port-forward+http", + "target": { + "context": "lab", + "namespace": "app", + "http": {"method": "POST"}, + }, + "rollback": "oc delete pods --all -n outside", + } + ] + } + ) + ) + results, _ = run(plan, scope, tmp_path, profile_map=restricted()) + assert results[0].verdict == "blocked_by_scope" + execute.assert_not_called() + rollback.assert_not_called() + + +def test_node_names_limit_general_scope(scope: Scope) -> None: + assert not scope.is_in_scope( + Action(adapter="k8s", context="lab", resource="nodes", name="other", verb="get") + )[0] + assert not scope.is_in_scope( + Action(adapter="k8s", context="lab", resource="nodes", verb="list") + )[0] + assert scope.is_in_scope( + Action(adapter="k8s", context="lab", resource="nodes", name="worker-0", verb="get") + )[0] + + +@pytest.mark.parametrize( + "value", ["null", "", "tomorrow", "2026-99-99", "2099-01-01\noff_limits: []"] +) +def test_invalid_expiry(value: str) -> None: + with pytest.raises(ValueError, match="ISO"): + expiry_date(value) + + +def test_origin_defaults() -> None: + assert origin("http://example.com") == origin("http://example.com:80") + assert origin("http://example.com") != origin("http://example.com:443") + + +def test_nested_data_validated(scope: Scope) -> None: + command = Mock( + side_effect=[ + (0, "https://api.example:6443", ""), + ( + 0, + json.dumps({"metadata": {"name": "worker-0"}, "status": {"addresses": [None]}}), + "", + ), + ] + ) + with pytest.raises(ValueError, match="discovery response"): + EndpointResolver(scope, command, "oc").resolve( + "lab", HttpProbe(mode="node", name="worker-0") + ) + + +@pytest.mark.parametrize("response", ["HTTP/1.1 403 Forbidden\r\n", "HTTP/1.1 200 OK\r\n"]) +def test_csrf_failure_stops_probe(scope: Scope, tmp_path: Path, response: str) -> None: + executor = HttpExecutor(K8sAdapter(), scope, "oc", AuditLog(tmp_path / "audit")) + executor.resolver.resolve = Mock( + return_value=Endpoint("https://web.example", "lab", "app", "routes", "web") + ) + executor.curl = Mock(return_value=(0, response, "")) + with pytest.raises(RuntimeError, match="session initialization"): + executor.execute("lab", HttpProbe(csrf=True)) + executor.curl.assert_called_once() + + +def test_private_headers_not_in_process_arguments( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from traust_engine._util import safe_exec + + adapter = K8sAdapter() + adapter.bind_profile_map(restricted()) + executor = HttpExecutor(adapter, scope, "oc", AuditLog(tmp_path / "audit")) + launch = Mock(return_value=(0, "", "")) + monkeypatch.setattr(safe_exec, "run_segments", launch) + executor.curl(["curl", "-H", "Authorization: Bearer canary", "http://127.0.0.1/"], "127.0.0.1") + argv = launch.call_args.args[0][0] + assert not any("canary" in value for value in argv) + assert "Authorization: Bearer canary" in launch.call_args.kwargs["input_"] + assert argv[argv.index("--noproxy") + 1] == "*" + + +@pytest.mark.skipif(not shutil.which("curl"), reason="curl is required") +def test_real_head_bypasses_proxy( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + class Handler(BaseHTTPRequestHandler): + def do_HEAD(self) -> None: + self.send_response(200) + self.send_header("Content-Length", "12345") + self.end_headers() + + def log_message(self, format: str, *args: object) -> None: + pass + + with ThreadingHTTPServer(("127.0.0.1", 0), Handler) as server: + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + monkeypatch.setenv("http_proxy", "http://127.0.0.1:1") + monkeypatch.setenv("no_proxy", "") + adapter = K8sAdapter() + adapter.bind_profile_map(restricted()) + executor = HttpExecutor(adapter, scope, "oc", AuditLog(tmp_path / "audit")) + executor.resolver.resolve = Mock( + return_value=Endpoint( + f"http://127.0.0.1:{server.server_port}", "lab", "app", "routes", "web" + ) + ) + code, output, _ = executor.execute("lab", HttpProbe(method="HEAD")) + assert code == 0 + assert "vf-http-status:200" in output + finally: + server.shutdown() + thread.join(timeout=3) + + +@pytest.mark.skipif( + not shutil.which("curl") or not shutil.which("openssl"), reason="TLS tools required" +) +def test_real_tls_ca_and_session( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import ssl + import subprocess + + certificate = tmp_path / "cert.pem" + key = tmp_path / "key.pem" + subprocess.run( + [ + "openssl", + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + str(key), + "-out", + str(certificate), + "-days", + "1", + "-subj", + "/CN=localhost", + "-addext", + "subjectAltName=DNS:web.app.svc", + ], + check=True, + capture_output=True, + ) + received = [] + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + received.append(dict(self.headers)) + authenticated = self.headers.get("Authorization") == "Bearer token-canary" + session_valid = "session=session-canary" in self.headers.get("Cookie", "") + self.send_response( + 200 if authenticated and (self.path == "/" or session_valid) else 403 + ) + self.send_header("Content-Length", str(len(b"token-canary session-canary csrf-canary"))) + if self.path == "/": + self.send_header("Set-Cookie", "custom-csrf=csrf-canary; Secure") + self.send_header("Set-Cookie", "session=session-canary; Secure") + self.end_headers() + self.wfile.write(b"token-canary session-canary csrf-canary") + + def log_message(self, format: str, *args: object) -> None: + pass + + with ThreadingHTTPServer(("127.0.0.1", 0), Handler) as server: + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(certificate, key) + server.socket = context.wrap_socket(server.socket, server_side=True) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + monkeypatch.setenv("VF_OAUTH_TOKEN", "token-canary") + scope.clusters["lab"].http_discovery = HttpDiscovery( + ca_bundle=str(certificate), session_check_path="/identity" + ) + adapter = K8sAdapter() + adapter.bind_profile_map(restricted()) + executor = HttpExecutor(adapter, scope, "oc", AuditLog(tmp_path / "audit")) + executor.resolver.resolve = Mock( + return_value=Endpoint( + f"https://web.app.svc:{server.server_port}", + "lab", + "app", + "routes", + "web", + credentials=True, + ) + ) + executor._tunnel_address = ("web.app.svc", server.server_port) + original_curl = executor.curl + + def checked_curl(argv: list[str], host: str) -> tuple[int, str, str]: + result = original_curl(argv, host) + assert result[0] == 0, result[2] + return result + + executor.curl = checked_curl + code, output, _ = executor.execute( + "lab", HttpProbe(path="/probe", authenticate=True, csrf=True) + ) + assert code == 0 + assert "canary" not in output + assert received[0]["Authorization"] == "Bearer token-canary" + assert "custom-csrf=csrf-canary" in received[1]["Cookie"] + assert "session=session-canary" in received[1]["Cookie"] + finally: + server.shutdown() + thread.join(timeout=3) + + +@pytest.mark.parametrize("failure", ["exit", "timeout", "kill"]) +def test_tunnel_failure_cleanup( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: str +) -> None: + import io + import subprocess + + from adapters import http + + process = Mock(stdout=io.BytesIO()) + process.__enter__ = Mock(return_value=process) + process.__exit__ = Mock(return_value=False) + process.poll.return_value = 1 if failure == "exit" else None + if failure == "kill": + process.wait.side_effect = [subprocess.TimeoutExpired("oc", 3), 0] + selector = Mock() + selector.__enter__ = Mock(return_value=selector) + selector.__exit__ = Mock(return_value=False) + monkeypatch.setattr(http.selectors, "DefaultSelector", Mock(return_value=selector)) + monkeypatch.setattr(http.subprocess, "Popen", Mock(return_value=process)) + monkeypatch.setattr( + http.time, "monotonic", Mock(side_effect=[0, 1 if failure == "exit" else 16]) + ) + executor = HttpExecutor(K8sAdapter(), scope, "oc", AuditLog(tmp_path / "audit")) + with ( + pytest.raises((RuntimeError, TimeoutError)), + executor.tunnel( + Endpoint("http://127.0.0.1", "lab", "app", "pods", "web", remote_port=8080) + ), + ): + pytest.fail("tunnel should not become ready") + if failure != "exit": + process.terminate.assert_called_once() + if failure == "kill": + process.kill.assert_called_once() + + +@pytest.mark.parametrize("command", ["oc get nodes other", "oc get nodes", "oc get node/other"]) +def test_declared_node_name_does_not_authorize_command( + scope: Scope, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, command: str +) -> None: + execute = Mock() + monkeypatch.setattr(K8sAdapter, "execute", execute) + plan = tmp_path / "plan.json" + plan.write_text( + json.dumps( + { + "steps": [ + { + "id": "node", + "adapter": "k8s", + "verb": "get", + "target": {"context": "lab", "resource": "nodes", "name": "worker-0"}, + "cmd": command, + } + ] + } + ) + ) + results, _ = run(plan, scope, tmp_path, profile_map=restricted()) + assert results[0].verdict == "blocked_by_scope" + execute.assert_not_called() diff --git a/tests/test_p2_hardening.py b/tests/test_p2_hardening.py index dc95047..db15d9a 100644 --- a/tests/test_p2_hardening.py +++ b/tests/test_p2_hardening.py @@ -28,24 +28,79 @@ ], ) def test_vet_shell_string_blocks(cmd, frag): - _argv, reason = AdapterBase._vet_shell_string(cmd) + _argv, reason = AdapterBase()._vet_shell_string(cmd) assert reason and frag in reason, (cmd, reason) def test_vet_shell_string_allows_legit(): - argv, reason = AdapterBase._vet_shell_string( - 'curl -sk https://api.x:6443/healthz -d \'{"a":"b;c"}\'' - ) + a = AdapterBase() + argv, reason = a._vet_shell_string('curl -sk https://api.x:6443/healthz -d \'{"a":"b;c"}\'') assert not reason and argv[0] == "curl" - argv, reason = AdapterBase._vet_shell_string("oc get pods -o json | jq '.items[0]'") + argv, reason = a._vet_shell_string("oc get pods -o json | jq '.items[0]'") assert not reason and argv is None # approved pipeline def test_run_blocks_instead_of_crashing(): - rc, _out, err = AdapterBase._run("python3 -c 'print(1)'") + rc, _out, err = AdapterBase()._run("python3 -c 'print(1)'") assert rc == 126 and "step blocked" in err +# ---- curl host allowlist plumbing ---------------------------------- + + +class _FakeScope: + def __init__(self, hosts): + self._hosts = tuple(hosts) + + def curl_hosts(self): + return self._hosts + + +def test_bind_scope_carries_roe_hosts_into_vetting(): + a = AdapterBase() + assert a._curl_hosts == () + a.bind_scope(_FakeScope(["api.hub.lab.example"])) + assert a._curl_hosts == ("api.hub.lab.example",) + + +def test_scope_curl_hosts_excludes_unowned_loopback(): + from scope import ClusterScope, Scope + + s = Scope(clusters={"c": ClusterScope(context="c", api="https://api.lab.example:6443")}) + hosts = s.curl_hosts() + assert "https://api.lab.example:6443" in hosts + assert not {"127.0.0.1", "localhost", "[::1]"} & set(hosts) + assert Scope().curl_hosts() == () + + +def test_bind_scope_tolerates_scope_without_curl_hosts(): + a = AdapterBase() + a.bind_scope(object()) + assert a._curl_hosts == () + + +def test_bind_profile_map_enforces_estate_profile(): + from traust_engine._util import safe_exec + + p = safe_exec.Profile( + name="validation-step", + description="test restricted", + allow=frozenset({"curl"}), + allowed_path_heads=frozenset(), + allow_pipelines=False, + keep_env=(), + posture="restricted", + ) + a = AdapterBase() + a.bind_profile_map({"validation-step": p}) + _argv, reason = a._vet_shell_string("curl https://evil.example/") + assert reason and "restricted" in reason + + a.bind_scope(_FakeScope(["api.hub.lab.example"])) + argv, reason = a._vet_shell_string("curl https://api.hub.lab.example/healthz") + assert not reason and argv is not None + + def test_classify_ifs_evasion_not_safe(): assert ( AdapterBase().classify("raw", cmd="rm${IFS}-rf /") != "safe" or True diff --git a/tests/test_validate_findings.py b/tests/test_validate_findings.py index 1aa506e..4731ead 100644 --- a/tests/test_validate_findings.py +++ b/tests/test_validate_findings.py @@ -1105,7 +1105,9 @@ def test_cwe_918_ssrf_with_path_adapted(self): assert step.verb == "port-forward+http" assert step.target["path"] == "/api/dev-console/webhooks/a" assert step.target["namespace"] == "openshift-console" - assert "vf-http-status" in step.cmd + assert step.cmd is None + assert step.target["http"]["method"] == "POST" + assert step.target["http"]["path"] == "/api/dev-console/webhooks/a" def test_unknown_cwe_no_step(self): f = Finding(id="F1", title="test", severity="low", cwes=["CWE-999"])