From 137c6c8ed5ec0c8bd65a542d61c926337695cc61 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Tue, 29 Sep 2026 20:51:21 +0200 Subject: [PATCH] feat(plane): per-task egress policy; receiver by name through the gateway Agent Substrate's egress gateway denies an actor without an EgressPolicy, so no task could deliver its artifacts or reach a model. egress.py grants each task exactly: http/80 to ousast-receiver.ax-system.svc.cluster.local, and TLS passthrough/443 to the bound Workspaces' https Git hosts and the bound Model's openultrasast.io/egress-hosts (api.deepseek.com for deepseek-flash). No IPs, no wildcards, no https (MITM) rules. The reconciler writes the policy with kubectl-ate between ax apply and resume, reads it back, waits OUSAST_EGRESS_SETTLE (11 s, the gateway's cached deny), and after ax delete task confirms the policy went with its actor (kubectl-ate has no delete verb; the store cascades). The receiver is a selector-less Service (ops/ax/receiver-service.yaml.tmpl, applied by up.sh together with the no-build agentgateway egress gateway) on port 18090 of the host. Measured live: the gateway checks the policy on Host but connects to the address the actor dialled, so the runner dials the Service's ClusterIP (OUSAST_ARTIFACT_DIAL) with the receiver's name as Host. The runner now labels its delivery with the Run's task name (OUSAST_TASK), which is what the receiver accepts; the ax Task name was refused with 403. The Receiver moved to egress.py to keep reconciler.py at 460 lines. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo --- .kiro/specs/ai-service-plane/design.md | 28 +- .kiro/specs/ai-service-plane/tasks.md | 6 + .../measurements/2026-09-08-module-audit.json | 14 +- ops/ax/receiver-service.yaml.tmpl | 31 +++ ops/ax/smoke-run.sh | 6 +- ops/ax/up.sh | 14 + plane/models/deepseek-flash.yaml | 1 + src/openultrasast/plane/egress.py | 252 ++++++++++++++++++ src/openultrasast/plane/manifests.py | 16 ++ src/openultrasast/plane/reconciler.py | 98 +++---- src/openultrasast/plane/runner.py | 31 ++- tests/conftest.py | 48 ++++ tests/test_plane_egress.py | 199 ++++++++++++++ tests/test_plane_reconciler.py | 37 ++- tests/test_plane_runner.py | 4 + 15 files changed, 703 insertions(+), 82 deletions(-) create mode 100644 ops/ax/receiver-service.yaml.tmpl create mode 100644 src/openultrasast/plane/egress.py create mode 100644 tests/test_plane_egress.py diff --git a/.kiro/specs/ai-service-plane/design.md b/.kiro/specs/ai-service-plane/design.md index 986b085..a1fc2e3 100644 --- a/.kiro/specs/ai-service-plane/design.md +++ b/.kiro/specs/ai-service-plane/design.md @@ -110,8 +110,8 @@ new image can fail with `DeadlineExceeded` while Agent Substrate builds its gold `OUSAST_RESUME_TIMEOUT` (900 s). ax has no Completed phase and never reports that a command exited: completion is the artifact delivery. `Failed` after a resume went through, or the task disappearing, before delivery fails the task with ax's condition message; no delivery within `OUSAST_TASK_TIMEOUT` (7200 s) fails it too; `ax delete task` -follows either way. Artifacts: the reconciler serves a small HTTP receiver on the host (reachable from kind pods -via the docker bridge); the runner posts the output directory there after its command exits. Tasks sharing a +follows either way. Artifacts: the reconciler serves a small HTTP receiver on the host (reachable from actors +by name through the egress gateway, see Egress below); the runner posts the output directory there after its command exits. Tasks sharing a `serialize` label never overlap; everything else may run in parallel up to `--workers`. A task's outcome is read from its delivered `summary.json`: `done` only when the task says every unit finished; `unfinished` on a budget stop; `failed` on an account or authentication error, a crash, or a missing delivery. `status` prints @@ -131,6 +131,30 @@ bound Model's `spec.secretKey.key` names the variable (`DEEPSEEK_API_KEY` for `d reconciler's own environment (`ousast` loads `.env` at startup); a Model-bound task whose variable is unset fails before `ax apply`, and a task without a Model gets none. The value is never rendered, logged or written. +#### Egress (`egress.py`) + +Every actor connection except DNS goes through Agent Substrate's egress gateway (`atenet-egress`), which denies +an actor without an `EgressPolicy`. Three pieces make the plane work under that rule: + +- **Receiver by name.** `ops/ax/receiver-service.yaml.tmpl` (applied by `up.sh`) is a selector-less Service + `ousast-receiver` in `ax-system`, port 80, with an EndpointSlice on the kind gateway at `OUSAST_ARTIFACT_PORT` + (fixed, 18090), where the reconciler binds its receiver. Tasks get + `OUSAST_ARTIFACT_URL=http://ousast-receiver.ax-system.svc.cluster.local/` plus `OUSAST_ARTIFACT_DIAL=:80` (looked up with kubectl unless set): the actor cannot resolve cluster names, so the + runner connects to the dial address and sends the URL's host as `Host`. The gateway decides the policy on + `Host` but connects to the address the actor dialled (measured: dialling the kind gateway on port 80 was allowed + and then refused upstream), so the dial address is the ClusterIP, which kube-proxy maps to the host. + `OUSAST_ARTIFACT_HOST` set keeps the old direct URL (tests). +- **Per-task policy.** `policy_for` grants `http` on port 80 to the receiver name, and `tls_passthrough` on 443 + to the Git hosts of the bound Workspaces' `https` repos and to the bound Model's + `openultrasast.io/egress-hosts` (`api.deepseek.com` for `deepseek-flash`, validated as a plain DNS name). + Nothing else: no IP address or wildcard, and passthrough only -- an `https` rule intercepts TLS and the actor + would need the gateway CA, which ax templates do not mount. +- **Order.** `ax apply` creates the actor; the reconciler writes the policy with `kubectl ate create + egress-policy` (update when one exists), reads it back, waits `OUSAST_EGRESS_SETTLE` (11 s: the gateway caches a + missing policy as deny for 10 s), then resumes. kubectl-ate has no delete verb; Substrate's store deletes the + policy with its actor, so after `ax delete task` the reconciler confirms it is gone. + ### Budget (`budget.py`) Wraps a chat client: sums the provider's usage fields per call (`prompt_tokens`, `prompt_cache_hit_tokens`, diff --git a/.kiro/specs/ai-service-plane/tasks.md b/.kiro/specs/ai-service-plane/tasks.md index 3c35656..f2905b1 100644 --- a/.kiro/specs/ai-service-plane/tasks.md +++ b/.kiro/specs/ai-service-plane/tasks.md @@ -11,6 +11,12 @@ mypy are clean, and the commit is gated on pytest's own exit code. Script-level and wait for it to finish; `ops/ax/down.sh` tears it down. Record the measured idle footprint (RSS, CPU) in `ops/ax/README.md` next to the 7 GB host constraint. - `ousast plane doctor` (in `reconciler.py` or a sibling `doctor.py`): the four checks of Requirement 3.5. + - Egress: `up.sh` applies Substrate's agentgateway egress gateway (`atenet-egress`, prebuilt image, rendered + with `kubectl kustomize`) when absent, and the receiver Service `ousast-receiver.ax-system` + (`receiver-service.yaml.tmpl`, endpoint the kind gateway at port 18090). The reconciler writes each task's + egress policy (`egress.py`: receiver over HTTP, Git and Model hosts as TLS passthrough) between `ax apply` and + resume; the runner delivers to `OUSAST_ARTIFACT_DIAL` (the Service's ClusterIP, port 80) with the receiver's name as + `Host`. - Stop rule: if the cluster does not come up within one working session, report the exact failure and stop; the decision to work around it is the maintainer's. - _Requirements: 3.5, 3.6_ diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index d3f9cf0..325b8f1 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,7 +1,7 @@ { - "modules": 131, + "modules": 132, "counts": { - "load_bearing": 122, + "load_bearing": 123, "standalone_capability": 9 }, "orphaned": [], @@ -661,6 +661,7 @@ "reason": "ai-service-plane Req 1-4: the service plane package; runner and reconciler integration in tasks 3-4", "importers": [ "cli", + "plane.egress", "plane.generate", "plane.reconciler", "plane.router", @@ -686,6 +687,14 @@ "plane.reconciler" ] }, + { + "module": "plane.egress", + "classification": "load_bearing", + "reason": "reachable from gate, map_gate, pair_gate or the CLI", + "importers": [ + "plane.reconciler" + ] + }, { "module": "plane.generate", "classification": "load_bearing", @@ -699,6 +708,7 @@ "classification": "load_bearing", "reason": "ai-service-plane Req 1.1-1.4: ax-shaped Task/Workspace/Model and the Run kind, validated before anything runs", "importers": [ + "plane.egress", "plane.generate", "plane.reconciler", "plane.router", diff --git a/ops/ax/receiver-service.yaml.tmpl b/ops/ax/receiver-service.yaml.tmpl new file mode 100644 index 0000000..7729d8c --- /dev/null +++ b/ops/ax/receiver-service.yaml.tmpl @@ -0,0 +1,31 @@ +# The reconciler's artifact receiver by name (ai-service-plane task 0). Actors reach nothing but through Agent +# Substrate's egress gateway, and the gateway allows HTTP by hostname only (never an IP address): the runner dials +# this Service's ClusterIP on port 80 with Host `ousast-receiver.ax-system.svc.cluster.local`; the gateway allows +# it on that Host and connects to the ClusterIP, which kube-proxy maps to the endpoint below, the host's receiver +# on ${OUSAST_ARTIFACT_PORT} at the kind gateway. Rendered by up.sh. +apiVersion: v1 +kind: Service +metadata: + name: ousast-receiver + namespace: ax-system +spec: + ports: + - name: http + port: 80 + targetPort: ${OUSAST_ARTIFACT_PORT} + protocol: TCP +--- +apiVersion: discovery.k8s.io/v1 +kind: EndpointSlice +metadata: + name: ousast-receiver-host + namespace: ax-system + labels: + kubernetes.io/service-name: ousast-receiver +addressType: IPv4 +ports: + - name: http + port: ${OUSAST_ARTIFACT_PORT} + protocol: TCP +endpoints: + - addresses: ["${KIND_GATEWAY}"] diff --git a/ops/ax/smoke-run.sh b/ops/ax/smoke-run.sh index 60babd1..bb49312 100755 --- a/ops/ax/smoke-run.sh +++ b/ops/ax/smoke-run.sh @@ -4,8 +4,10 @@ set -euo pipefail export PATH="$HOME/go/bin:$HOME/.local/bin:$PATH" SRC="${OUSAST_AX_SRC:-$HOME/.cache/ousast/ax-src}" RUNNER_IMAGE="$(cat "$SRC/runner-image")"; export RUNNER_IMAGE -KIND_GW="$(docker network inspect kind --format '{{range .IPAM.Config}}{{if .Gateway}}{{.Gateway}} {{end}}{{end}}' | tr ' ' '\n' | grep -m1 '\.')" -export OUSAST_ARTIFACT_HOST="${OUSAST_ARTIFACT_HOST:-$KIND_GW}" # sandboxes reach the host via the kind gateway +# Actors reach the receiver by its Service name through the egress gateway (ops/ax/receiver-service.yaml.tmpl, +# applied by up.sh, endpoint this host at OUSAST_ARTIFACT_PORT): the runner dials the Service's ClusterIP on port +# 80 with that name as Host; the reconciler looks the ClusterIP up unless OUSAST_ARTIFACT_DIAL is set. +export OUSAST_ARTIFACT_PORT="${OUSAST_ARTIFACT_PORT:-18090}" envsubst < "$(dirname "$0")/smoke-run.yaml.tmpl" > "$SRC/smoke-run.yaml" REPO="$(cd "$(dirname "$0")/../.." && pwd)" "$REPO/.venv/bin/ousast" plane run "$SRC/smoke-run.yaml" "$@" diff --git a/ops/ax/up.sh b/ops/ax/up.sh index ef41785..a5a85c6 100755 --- a/ops/ax/up.sh +++ b/ops/ax/up.sh @@ -35,6 +35,15 @@ if ready ate-system; then echo "substrate ready; kept"; else fi kubectl --context "$CTX" -n ate-system get pods +step "egress gateway (atenet-egress, agentgateway variant: one prebuilt image, nothing built)" +# Every actor connection goes through this gateway; without it an actor has no network at all. The Substrate +# install can die before deploying it (e.g. a full disk), so it is applied on its own when absent. +if kubectl --context "$CTX" -n ate-system get deploy atenet-egress >/dev/null 2>&1; then echo "gateway present; kept"; else + kubectl kustomize --load-restrictor=LoadRestrictionsNone "$SRC/substrate/manifests/ate-install/agentgateway-egress" \ + | kubectl --context "$CTX" apply -f - + kubectl --context "$CTX" -n ate-system rollout status deploy/atenet-egress --timeout=300s +fi + step "ax control plane (ax-system)" if ready ax-system; then echo "ax ready; kept"; else (cd "$SRC/ax" && make deploy AX_IMAGE_REPO="$KO_DOCKER_REPO") @@ -42,6 +51,11 @@ if ready ax-system; then echo "ax ready; kept"; else fi kubectl --context "$CTX" -n ax-system get pods +step "artifact receiver Service (ousast-receiver.ax-system -> this host)" +KIND_GATEWAY="$(docker network inspect kind --format '{{range .IPAM.Config}}{{if .Gateway}}{{.Gateway}} {{end}}{{end}}' | tr ' ' '\n' | grep -m1 '\.')" +export KIND_GATEWAY OUSAST_ARTIFACT_PORT="${OUSAST_ARTIFACT_PORT:-18090}" # the reconciler's fixed receiver port +envsubst < "$(dirname "$0")/receiver-service.yaml.tmpl" | kubectl --context "$CTX" apply -f - + step "worker pool (gVisor)" SUBSTRATE_VERSION="$(kubectl --context "$CTX" get nodes -o jsonpath='{.items[0].metadata.labels.ate\.dev/substrate-version}')" export SUBSTRATE_VERSION diff --git a/plane/models/deepseek-flash.yaml b/plane/models/deepseek-flash.yaml index 4761c3f..5fe2e47 100644 --- a/plane/models/deepseek-flash.yaml +++ b/plane/models/deepseek-flash.yaml @@ -10,6 +10,7 @@ metadata: atespace: default annotations: openultrasast.io/provider-extension: "true" + openultrasast.io/egress-hosts: "api.deepseek.com" # model.endpoint.DEEPSEEK_BASE_URL; the task's only model egress spec: provider: deepseek model: deepseek-flash diff --git a/src/openultrasast/plane/egress.py b/src/openultrasast/plane/egress.py new file mode 100644 index 0000000..23b64bc --- /dev/null +++ b/src/openultrasast/plane/egress.py @@ -0,0 +1,252 @@ +"""Per-task network egress on Agent Substrate (ai-service-plane, task 0: gateway + receiver Service). + +Every TCP connection an actor opens (DNS aside) is redirected to Substrate's egress gateway +(``atenet-egress.ate-system``), which denies by default: an actor without an ``EgressPolicy`` gets no tunnel, and +a refused connection is accepted and then closed. :func:`policy_for` grants a task exactly what it needs: + +* ``http`` on port 80 to the artifact receiver's cluster name (``ousast-receiver.ax-system.svc.cluster.local``, a + selector-less Service whose endpoint is this host's receiver, ``ops/ax/receiver-service.yaml.tmpl``). The actor + cannot resolve cluster names, so the runner dials ``OUSAST_ARTIFACT_DIAL`` (the Service's ClusterIP, port 80) + with the receiver's name as ``Host``. The gateway decides the policy on that ``Host`` but connects to the + address the actor dialled (measured live: agentgateway's upstream is the CONNECT authority, not the Host), so + the dial address must be one the gateway pod reaches: the ClusterIP, which kube-proxy maps to the host. +* ``tls_passthrough`` on 443 for the Git hosts of the bound Workspaces' ``https`` repos (smart HTTP talks to the + repo's host only) and for the bound Model's ``openultrasast.io/egress-hosts`` (``api.deepseek.com``). + +Nothing else: no wildcard, no IP address (the gateway rejects them), no ``https`` rule (that one intercepts TLS and +the actor would have to trust the gateway CA, which ax templates do not mount). :class:`Egress` writes the policy +with ``kubectl ate create egress-policy`` after ``ax apply`` created the actor, reads it back, and before resume +waits ``OUSAST_EGRESS_SETTLE`` s (11) because the gateway caches a missing policy as deny for 10 s. The kubectl-ate +CLI has no delete verb: Substrate's store deletes the policy with its actor (``ON DELETE CASCADE``), so +:meth:`Egress.delete` runs after ``ax delete task`` and confirms the policy is gone. +""" + +from __future__ import annotations + +import io +import ipaddress +import json +import os +import subprocess +import tarfile +import threading +import time +from collections.abc import Iterable, Mapping +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit + +from .manifests import Model, Task, Workspace + +__all__ = [ + "RECEIVER_HOST", + "RECEIVER_PORT", + "Egress", + "Receiver", + "git_hosts", + "receiver_cluster_ip", + "policy_for", + "receiver_address", + "rules_of", +] + +RECEIVER_HOST = "ousast-receiver.ax-system.svc.cluster.local" +RECEIVER_PORT = 18090 # the receiver Service's targetPort; ops/ax/up.sh renders the Service with the same number +_NOT_FOUND = ("not found", "notfound") + + +def _is_ip(host: str) -> bool: + try: + ipaddress.ip_address(host.strip("[]")) + except ValueError: + return False + return True + + +def git_hosts(workspaces: Iterable[Workspace]) -> tuple[list[str], list[str]]: + """(https hosts, http hosts) of the Workspaces' ``spec.git[].repo`` URLs; ssh and IP-address repos add nothing.""" + tls, plain = set[str](), set[str]() + for ws in workspaces: + for source in ws.git: + parts = urlsplit(source.repo) + host = (parts.hostname or "").lower() + if ( + host + and not _is_ip(host) + and parts.scheme in ("https", "http") + and parts.port in (None, 443 if parts.scheme == "https" else 80) + ): + (tls if parts.scheme == "https" else plain).add(host) + return sorted(tls), sorted(plain) + + +def policy_for(task: Task, workspaces: Iterable[Workspace], model: Model | None, receiver_host: str | None) -> dict[str, Any]: + """The task's EgressPolicy body: the receiver over HTTP/80, Git and Model hosts as TLS passthrough on 443.""" + tls, plain = git_hosts(workspaces) + tls = sorted(set(tls) | set(model.egress_hosts if model else ())) + if receiver_host and not _is_ip(receiver_host): # an address cannot be allowed; only the direct test mode has one + plain = sorted({*plain, receiver_host.lower()}) + rules: list[dict[str, Any]] = [] + if plain: + rules.append({"http": {"hostnames": plain, "ports": {"numbers": [80]}}}) + if tls: + rules.append({"tlsPassthrough": {"hostnames": tls, "ports": {"numbers": [443]}}}) + return {"rules": rules} + + +def rules_of(policy: Mapping[str, Any]) -> list[dict[str, Any]]: + """The rules of a policy as ``kubectl ate get -o json`` prints them, normalised for comparison.""" + out = [] + for rule in policy.get("rules") or []: + for kind, body in rule.items(): + key = "tlsPassthrough" if kind in ("tls_passthrough", "tlsPassthrough") else kind + ports = body.get("ports") or {} + norm_ports: dict[str, Any] = {"all": {}} if "all" in ports else {"numbers": sorted(int(n) for n in ports.get("numbers") or [])} + out.append({key: {"hostnames": sorted(body.get("hostnames") or []), "ports": norm_ports}}) + return sorted(out, key=lambda r: json.dumps(r, sort_keys=True)) + + +def receiver_cluster_ip(context: str | None = None) -> str | None: + """The ClusterIP of the ``ousast-receiver`` Service (``kubectl get svc``); None when kubectl cannot say.""" + context = context or "kind-" + (os.environ.get("KIND_CLUSTER_NAME") or "ousast") + argv = ["kubectl", "--context", context, "-n", "ax-system", "get", "svc", "ousast-receiver", "-o", "jsonpath={.spec.clusterIP}"] + try: + proc = subprocess.run(argv, capture_output=True, text=True, check=False, timeout=20) + except (OSError, subprocess.TimeoutExpired): + return None + address = proc.stdout.strip() + return address if proc.returncode == 0 and _is_ip(address) else None + + +def receiver_address(port: int) -> tuple[str, str | None]: + """(``OUSAST_ARTIFACT_URL``, ``OUSAST_ARTIFACT_DIAL``) for a receiver bound on ``port``. + + ``OUSAST_ARTIFACT_HOST`` set: the old direct URL ``http://:/`` and no dial (tests, a host the actor + reaches without the gateway). Otherwise the receiver Service's name, dialled at ``OUSAST_ARTIFACT_DIAL`` or the + Service's ClusterIP on port 80: the gateway checks the policy against ``Host`` but connects to the address the + actor dialled (its CONNECT authority), so that address must reach the receiver from the gateway pod. + """ + host = os.environ.get("OUSAST_ARTIFACT_HOST") + if host: + return f"http://{host}:{port}/", None + dial = os.environ.get("OUSAST_ARTIFACT_DIAL") + if not dial: + address = receiver_cluster_ip() + dial = f"{address}:80" if address else None + return f"http://{RECEIVER_HOST}/", dial + + +class Egress: + """``kubectl ate`` for one actor's egress policy; the executable is injectable so tests use a fake.""" + + def __init__(self, executable: str | None = None, context: str | None = None) -> None: + self.executable = executable or os.environ.get("OUSAST_KUBECTL_ATE") or "kubectl-ate" + self.context = context or "kind-" + (os.environ.get("KIND_CLUSTER_NAME") or "ousast") + + def _run(self, *args: str, stdin: str | None = None) -> subprocess.CompletedProcess[str]: + argv = [self.executable, "--context", self.context, *args] + return subprocess.run(argv, input=stdin, capture_output=True, text=True, check=False) + + @staticmethod + def _error(proc: subprocess.CompletedProcess[str]) -> str: + return (proc.stderr.strip() or proc.stdout.strip() or f"exit {proc.returncode}")[-400:] + + def get(self, actor: str, atespace: str) -> dict[str, Any] | None: + """The stored policy, None when there is none (or no actor); RuntimeError when kubectl-ate fails otherwise.""" + proc = self._run("get", "egress-policy", actor, "-a", atespace, "-o", "json") + if proc.returncode != 0: + if "notfound" in self._error(proc).lower().replace(" ", ""): + return None + raise RuntimeError(f"kubectl ate get egress-policy {atespace}/{actor} failed: {self._error(proc)}") + if not proc.stdout.strip(): # an actor without a policy: exit 0, "has no egress policy" on stderr only + return None + try: + doc = json.loads(proc.stdout) + except ValueError as exc: + raise RuntimeError(f"kubectl ate get egress-policy {atespace}/{actor}: unreadable output ({exc})") from None + return doc if isinstance(doc, dict) else None + + def apply(self, actor: str, atespace: str, policy: Mapping[str, Any], *, timeout: float | None = None) -> None: + """Create (or replace) the policy, retrying while the actor does not exist yet, then read it back.""" + limit = float(os.environ.get("OUSAST_EGRESS_TIMEOUT") or 60) if timeout is None else timeout + deadline, backoff = time.monotonic() + limit, 0.5 + while True: + existing = self.get(actor, atespace) + if existing is None: + proc = self._run("create", "egress-policy", actor, "-a", atespace, "-f", "-", stdin=json.dumps(dict(policy))) + else: + meta = {k: existing.get("metadata", {}).get(k) for k in ("name", "atespace", "uid", "version")} + body = json.dumps({**dict(policy), "metadata": meta}) + proc = self._run("update", "egress-policy", actor, "-a", atespace, "-f", "-", stdin=body) + if proc.returncode == 0: + break + error = self._error(proc) + retry = any(word in error.lower() for word in (*_NOT_FOUND, "already exists", "conflict", "unavailable")) + if not retry or time.monotonic() + backoff > deadline: + raise RuntimeError(f"egress policy for {atespace}/{actor} not written: {error}") + time.sleep(backoff) + backoff = min(backoff * 2, 5.0) + stored = self.get(actor, atespace) + if stored is None or rules_of(stored) != rules_of(policy): + raise RuntimeError(f"egress policy for {atespace}/{actor} reads back {stored!r}, not {dict(policy)!r}") + + def delete(self, actor: str, atespace: str) -> str | None: + """After ``ax delete task``: None once the policy is gone with its actor, else why it is still there.""" + try: + stored = self.get(actor, atespace) + except RuntimeError as exc: + return str(exc) + return None if stored is None else f"egress policy for {atespace}/{actor} outlived its actor" + + @staticmethod + def settle() -> None: + """Wait out the gateway's cached deny (10 s) before the actor's first connection: ``OUSAST_EGRESS_SETTLE``.""" + time.sleep(float(os.environ.get("OUSAST_EGRESS_SETTLE") or 11)) + + +class Receiver(ThreadingHTTPServer): + """Accepts one tar per running task and extracts it under ``//``.""" + + daemon_threads = True + + def __init__(self, run_name: str, base: Path, port: int) -> None: + super().__init__(("0.0.0.0", port), _Handler) + self.run_name, self.base, self.running, self.delivered = run_name, base, set[str](), set[str]() + self.lock = threading.Lock() + self.url, self.dial = receiver_address(self.server_address[1]) + + def accept(self, headers: Any, body: bytes) -> tuple[int, str]: + task = headers.get("X-Ousast-Task", "") + if headers.get("X-Ousast-Run") != self.run_name or task not in self.running: + return 403, "unknown run or task" + if not str(headers.get("Content-Type", "")).startswith("application/x-tar"): + return 415, "expected application/x-tar" + dest = (self.base / task).resolve() + try: + with tarfile.open(fileobj=io.BytesIO(body), mode="r:") as tar: + members = tar.getmembers() + for m in members: + if m.issym() or m.islnk() or not (dest / m.name).resolve().is_relative_to(dest): + return 400, f"rejected member {m.name}" + dest.mkdir(parents=True, exist_ok=True) + tar.extractall(dest, members) # every member checked above + except tarfile.TarError as exc: + return 400, f"bad tar: {exc}" + with self.lock: + self.delivered.add(task) + return 200, "ok" + + +class _Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: + server: Receiver = self.server # type: ignore[assignment] + body = self.rfile.read(int(self.headers.get("Content-Length") or 0)) + code, text = server.accept(self.headers, body) + self.send_response(code) + self.send_header("Content-Type", "text/plain") + self.end_headers() + self.wfile.write(text.encode()) + + def log_message(self, format: str, *args: object) -> None: + return diff --git a/src/openultrasast/plane/manifests.py b/src/openultrasast/plane/manifests.py index f79f15b..3e448be 100644 --- a/src/openultrasast/plane/manifests.py +++ b/src/openultrasast/plane/manifests.py @@ -24,6 +24,7 @@ import yaml __all__ = [ "AX_API_VERSION", "AX_PROVIDERS", + "EGRESS_HOSTS_ANNOTATION", "EXTENSION_PROVIDERS", "GIT_COMMITS_ANNOTATION", "PROVIDER_EXTENSION_ANNOTATION", @@ -55,6 +56,8 @@ AX_API_VERSION = "ax.io/v1alpha1" RUN_API_VERSION = "openultrasast.io/v1alpha1" PROVIDER_EXTENSION_ANNOTATION = "openultrasast.io/provider-extension" GIT_COMMITS_ANNOTATION = "openultrasast.io/git-commits" +EGRESS_HOSTS_ANNOTATION = "openultrasast.io/egress-hosts" # the Model API hosts a bound task may reach +_DNS_NAME = re.compile(r"^(?=.{1,253}$)[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?(\.[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?)*$") AX_PROVIDERS = frozenset({"google", "anthropic"}) EXTENSION_PROVIDERS = frozenset({"deepseek"}) @@ -168,6 +171,15 @@ class Model: secret_key: SecretKey | None = None parameters: Mapping[str, Any] = field(default_factory=dict) + @property + def egress_hosts(self) -> tuple[str, ...]: + """DNS names from ``metadata.annotations["openultrasast.io/egress-hosts"]`` (comma-separated).""" + return _split_hosts(self.metadata.annotations.get(EGRESS_HOSTS_ANNOTATION, "")) + + +def _split_hosts(text: str) -> tuple[str, ...]: + return tuple(h.strip() for h in text.split(",") if h.strip()) + @dataclass(frozen=True) class Budget: @@ -426,6 +438,10 @@ def _parse_model(check: _Check, metadata: Metadata, spec: Mapping[str, Any]) -> ) elif provider not in AX_PROVIDERS: raise check.fail("spec.provider", f'"{provider}" is not one of {", ".join(sorted(AX_PROVIDERS | EXTENSION_PROVIDERS))}') + for host in _split_hosts(metadata.annotations.get(EGRESS_HOSTS_ANNOTATION, "")): + if not _DNS_NAME.match(host) or host.replace(".", "").isdigit(): + path = f'metadata.annotations["{EGRESS_HOSTS_ANNOTATION}"]' + raise check.fail(path, f"{host!r} is not a lowercase DNS name (no wildcard, IP address, port or scheme)") secret = check.optional_mapping(spec, "spec", "secretKey", ("name", "key")) parameters = spec.get("parameters") or {} if not isinstance(parameters, dict): diff --git a/src/openultrasast/plane/reconciler.py b/src/openultrasast/plane/reconciler.py index 387e370..7dca7ed 100644 --- a/src/openultrasast/plane/reconciler.py +++ b/src/openultrasast/plane/reconciler.py @@ -12,40 +12,39 @@ vanishing, a refused start or ``OUSAST_TASK_TIMEOUT`` s (7200) without delivery ax rejects unknown fields, so rendered documents carry only ax's fields; ``openultrasast.io/model`` reaches the task as ``OUSAST_MODEL`` plus ``OUSAST_MODEL_PARAMS`` (Req 7.1), each Workspace's ``openultrasast.io/git-commits`` as ``OUSAST_GIT_PINS``; ``OUSAST_INPUT_`` points into ``/workspace/.ousast-in/``. The receiver takes -the runner's tar (``X-Ousast-Run``, ``X-Ousast-Task``) on ``OUSAST_ARTIFACT_PORT``, advertised as -``OUSAST_ARTIFACT_HOST`` (default ``172.17.0.1``, the docker bridge). State: ``~/ousast-results/plane//`` -(``OUSAST_RESULTS``), polled every ``OUSAST_POLL_SECONDS`` (3). +the runner's tar (``X-Ousast-Run``, ``X-Ousast-Task``) on ``OUSAST_ARTIFACT_PORT`` (18090), reached by its Service +name through the egress gateway; between ``ax apply`` and resume the task's egress policy is written (``egress.py``). +State: ``~/ousast-results/plane//`` (``OUSAST_RESULTS``), polled every ``OUSAST_POLL_SECONDS`` (3). """ from __future__ import annotations import contextlib -import io import json import os import queue import re import subprocess -import tarfile +import sys import threading import time from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, field from datetime import UTC, datetime -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any +from urllib.parse import urlsplit import yaml from .doctor import doctor +from .egress import RECEIVER_PORT, Egress, Receiver, policy_for from .manifests import AX_API_VERSION, Manifests, Model, Run, RunTask, Task, Workspace, load_manifests from .router import Router, credentials, open_router MODEL_ANNOTATION = "openultrasast.io/model" OUTPUT_ROOT = "/workspace/.ousast-out" INPUTS_ROOT = "/workspace/.ousast-in" -DEFAULT_ARTIFACT_HOST = "172.17.0.1" TRANSIENT = re.compile(r"deadline ?exceeded|unavailable", re.IGNORECASE) # the golden-snapshot build outlives a resume _ROW = "{:<28} {:<20} {:>7} {:>10} {:>10} {:>9} {:>10}" _COLUMNS = ("calls", "prompt", "cache_hit", "output", "usd") @@ -134,13 +133,15 @@ def _input_files(entry: RunTask, base: Path) -> list[dict[str, str]]: return files -def render_task(run: Run, entry: RunTask, manifests: Manifests, base: Path, artifact_url: str) -> list[dict[str, Any]]: +def render_task( + run: Run, entry: RunTask, manifests: Manifests, base: Path, artifact_url: str, dial: str | None = None +) -> list[dict[str, Any]]: """The documents ``ax apply`` receives for one Run task: Task, its Model, its Workspaces, the inputs Workspace.""" task: Task = manifests.tasks[entry.task] ax_name = _ax_name(run.metadata.name, entry.name) env = {e.name: e.value for e in task.env} env.update(OUSAST_RUN=run.metadata.name, OUSAST_TASK=entry.name, OUSAST_OUTPUT_DIR=f"{OUTPUT_ROOT}/{entry.name}") - env["OUSAST_ARTIFACT_URL"] = artifact_url + env.update(OUSAST_ARTIFACT_URL=artifact_url, **({"OUSAST_ARTIFACT_DIAL": dial} if dial else {})) budget = _fields(entry.budget) if entry.budget else {} env.update({f"OUSAST_BUDGET_{k.upper()}": str(v) for k, v in budget.items()}) for input_name, ref in entry.inputs.items(): @@ -214,55 +215,6 @@ class Ax: self._run("delete", "task", name) -class Receiver(ThreadingHTTPServer): - """Accepts one tar per running task and extracts it under ``//``.""" - - daemon_threads = True - - def __init__(self, run_name: str, base: Path, port: int) -> None: - super().__init__(("0.0.0.0", port), _Handler) - self.run_name, self.base, self.running, self.delivered = run_name, base, set[str](), set[str]() - self.lock = threading.Lock() - - def url(self) -> str: - return f"http://{os.environ.get('OUSAST_ARTIFACT_HOST') or DEFAULT_ARTIFACT_HOST}:{self.server_address[1]}/" - - def accept(self, headers: Any, body: bytes) -> tuple[int, str]: - task = headers.get("X-Ousast-Task", "") - if headers.get("X-Ousast-Run") != self.run_name or task not in self.running: - return 403, "unknown run or task" - if not str(headers.get("Content-Type", "")).startswith("application/x-tar"): - return 415, "expected application/x-tar" - dest = (self.base / task).resolve() - try: - with tarfile.open(fileobj=io.BytesIO(body), mode="r:") as tar: - members = tar.getmembers() - for m in members: - if m.issym() or m.islnk() or not (dest / m.name).resolve().is_relative_to(dest): - return 400, f"rejected member {m.name}" - dest.mkdir(parents=True, exist_ok=True) - tar.extractall(dest, members) # every member checked above - except tarfile.TarError as exc: - return 400, f"bad tar: {exc}" - with self.lock: - self.delivered.add(task) - return 200, "ok" - - -class _Handler(BaseHTTPRequestHandler): - def do_POST(self) -> None: - server: Receiver = self.server # type: ignore[assignment] - body = self.rfile.read(int(self.headers.get("Content-Length") or 0)) - code, text = server.accept(self.headers, body) - self.send_response(code) - self.send_header("Content-Type", "text/plain") - self.end_headers() - self.wfile.write(text.encode()) - - def log_message(self, format: str, *args: object) -> None: - return - - # --- the run ------------------------------------------------------------------------------------------------------- @@ -347,35 +299,45 @@ def _await(ax: Ax, name: str, delivered: Callable[[], bool], start: Callable[[], return None -def _execute(ax: Ax, run: Run, entry: RunTask, manifests: Manifests, base: Path, receiver: Receiver, router: Router) -> tuple[str, str]: +def _execute( + ax: Ax, egress: Egress, run: Run, entry: RunTask, manifests: Manifests, base: Path, receiver: Receiver, router: Router +) -> tuple[str, str]: ax_name, task = _ax_name(run.metadata.name, entry.name), manifests.tasks[entry.task] - manifest = base / entry.name / "task.yaml" + atespace, manifest = task.metadata.atespace or "default", base / entry.name / "task.yaml" manifest.parent.mkdir(parents=True, exist_ok=True) - docs = render_task(run, entry, manifests, base, receiver.url()) + docs = render_task(run, entry, manifests, base, receiver.url, receiver.dial) manifest.write_text(yaml.safe_dump_all(docs, sort_keys=False), encoding="utf-8") + model = manifests.models.get(task.metadata.annotations.get(MODEL_ANNOTATION) or "") + policy = policy_for(task, [manifests.workspaces[b.name] for b in task.workspaces], model, urlsplit(receiver.url).hostname) try: # Req 4.6: the bound Model's credential is read here and only ever sent in the start request - secrets = credentials(manifests.models.get(task.metadata.annotations.get(MODEL_ANNOTATION) or "")) - start = router.starter(task.metadata.atespace or "default", ax_name, run.metadata.name, secrets) - ax.apply(manifest) + start = router.starter(atespace, ax_name, run.metadata.name, credentials(model)) + ax.apply(manifest) # creates the Substrate actor; its egress is denied until the policy exists + egress.apply(ax_name, atespace, policy) + egress.settle() except RuntimeError as exc: # StartError included: a Model-bound task whose secret is unset + ax.delete(ax_name) return "failed", str(exc) try: failure = _await(ax, ax_name, lambda: entry.name in receiver.delivered, start) finally: ax.delete(ax_name) + leftover = egress.delete(ax_name, atespace) # the policy goes with its actor; a survivor is reported + if leftover: + print(leftover, file=sys.stderr) return ("failed", failure) if failure else outcome_of(_read_json(base / entry.name / "summary.json")) -def run(run_manifest: Path, *, workers: int = 1, ax: str = "ax", results_root: Path | None = None) -> str: +def run(run_manifest: Path, *, workers: int = 1, ax: str = "ax", results_root: Path | None = None, kubectl_ate: str | None = None) -> str: """Execute a Run on ax and return its status: ``done``, ``unfinished`` or ``failed`` (Req 3.1-3.3, 2.2).""" run_spec, manifests = load_run(Path(run_manifest)) base = run_dir(run_spec.metadata.name, results_root) base.mkdir(parents=True, exist_ok=True) lock = _acquire_lock(base) state = _State(base / "state.json", _read_json(base / "state.json") or {"run": run_spec.metadata.name, "started": _now(), "tasks": {}}) - receiver = Receiver(run_spec.metadata.name, base, int(os.environ.get("OUSAST_ARTIFACT_PORT") or 0)) + port = os.environ.get("OUSAST_ARTIFACT_PORT") or (0 if os.environ.get("OUSAST_ARTIFACT_HOST") else RECEIVER_PORT) + receiver = Receiver(run_spec.metadata.name, base, int(port)) # the receiver Service targets RECEIVER_PORT threading.Thread(target=receiver.serve_forever, daemon=True).start() - cli, finished = Ax(ax), queue.Queue[tuple[str, str, str]]() + cli, egress, finished = Ax(ax), Egress(kubectl_ate), queue.Queue[tuple[str, str, str]]() pending = [t for t in run_spec.tasks if state.status(t.name) != "done"] for entry in run_spec.tasks: state.set(entry.name, **({"status": "pending"} if entry in pending else {})) @@ -386,7 +348,7 @@ def run(run_manifest: Path, *, workers: int = 1, ax: str = "ax", results_root: P if state.status(entry.name) in ("running", "suspended"): cli.delete(_ax_name(run_spec.metadata.name, entry.name)) # an interrupted attempt; ax must not keep it state.set(entry.name, status="running", started=_now(), finished=None) - finished.put((entry.name, *_execute(cli, run_spec, entry, manifests, base, receiver, router))) + finished.put((entry.name, *_execute(cli, egress, run_spec, entry, manifests, base, receiver, router))) try: with open_router() as router: # one port-forward to Substrate's router for the whole run diff --git a/src/openultrasast/plane/runner.py b/src/openultrasast/plane/runner.py index 0ac6145..1ef5c25 100644 --- a/src/openultrasast/plane/runner.py +++ b/src/openultrasast/plane/runner.py @@ -43,6 +43,8 @@ Environment the runner reads, all optional except the two ax variables: OUSAST_CASE_CACHE a directory of git checkouts (``benchmarks/independent`` cache layout) OUSAST_GIT_PINS JSON ``{"/": "<40-hex commit>"}`` from the reconciler OUSAST_ARTIFACT_URL, OUSAST_RUN where to deliver, and the run name carried in ``X-Ousast-Run`` + OUSAST_ARTIFACT_DIAL ``host:port`` to connect to instead of the URL's host, which is sent as + ``Host`` (the egress gateway decides on and resolves that name) OUSAST_DELIVERY_BACKOFF seconds before the second delivery attempt, doubled per retry The runner exports ``OUSAST_WORKSPACE_DIR`` (the first bound workspace) when the Task's env leaves it unset, @@ -72,6 +74,7 @@ from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any +from urllib.parse import urlsplit import yaml @@ -632,11 +635,26 @@ def build_tar(output_dir: Path) -> bytes: return buffer.getvalue() -def deliver(url: str, payload: bytes, headers: Mapping[str, str], *, attempts: int = DELIVERY_ATTEMPTS, backoff: float = 1.0) -> int: - """POST ``payload`` to ``url`` with ``headers``; retry with doubling backoff; return the 2xx status.""" - last = "no attempt made" +def deliver( + url: str, + payload: bytes, + headers: Mapping[str, str], + *, + attempts: int = DELIVERY_ATTEMPTS, + backoff: float = 1.0, + dial: str | None = None, +) -> int: + """POST ``payload`` to ``url`` with ``headers``; retry with doubling backoff; return the 2xx status. + + ``dial`` (``host:port``): connect there instead and send the URL's host as ``Host``, as the actor must -- it + cannot resolve the receiver's cluster name, and the egress gateway decides on ``Host`` and resolves it itself. + """ + last, target, headers = "no attempt made", url, dict(headers) + if dial: + parts = urlsplit(url) + target, headers["Host"] = parts._replace(netloc=dial).geturl(), parts.netloc for attempt in range(1, attempts + 1): - request = urllib.request.Request(url, data=payload, method="POST", headers=dict(headers)) + request = urllib.request.Request(target, data=payload, method="POST", headers=headers) try: with urllib.request.urlopen(request, timeout=60) as response: # noqa: S310 - the reconciler's URL status = int(response.status) @@ -658,12 +676,13 @@ def _deliver_output(task: Task, output_dir: Path, environ: Mapping[str, str]) -> url = environ.get("OUSAST_ARTIFACT_URL") if not url: return False - headers = {"Content-Type": "application/x-tar", "X-Ousast-Task": task.metadata.name} + # the Run's task name (``OUSAST_TASK``) is what the receiver knows; the ax Task is named ``-`` + headers = {"Content-Type": "application/x-tar", "X-Ousast-Task": environ.get("OUSAST_TASK") or task.metadata.name} run = environ.get("OUSAST_RUN") if run: headers["X-Ousast-Run"] = run backoff = float(environ.get("OUSAST_DELIVERY_BACKOFF", "1.0")) - deliver(url, build_tar(output_dir), headers, backoff=backoff) + deliver(url, build_tar(output_dir), headers, backoff=backoff, dial=environ.get("OUSAST_ARTIFACT_DIAL") or None) return True diff --git a/tests/conftest.py b/tests/conftest.py index 9562d7b..7abe394 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -62,3 +62,51 @@ def _scratch_under_tmp_path(monkeypatch: pytest.MonkeyPatch, tmp_path_factory: p scratch = tmp_path_factory.mktemp("scratch") monkeypatch.setattr(tempfile, "tempdir", str(scratch)) + + +FAKE_KUBECTL_ATE = r'''#!{python} +"""Fake kubectl-ate for egress policies: an actor exists while ``applied/.json`` exists and is not gone +(the fake ax's record); its policy goes with it, as Substrate's store cascades. Every call is logged to ax.log.""" +import json, sys, time +from pathlib import Path + +HERE = Path(__file__).resolve().parent +argv = sys.argv[1:] +if argv[:1] == ["--context"]: + argv = argv[2:] +verb, actor, space = argv[0], argv[2], argv[argv.index("-a") + 1] +with (HERE / "ax.log").open("a") as log: + log.write(f"{time.monotonic():.4f} egress {verb} {actor}\n") +record = HERE / "applied" / f"{actor}.json" +alive = record.exists() and not json.loads(record.read_text()).get("gone") +store = HERE / "policies" / f"{space}_{actor}.json" +if not alive: + store.unlink(missing_ok=True) + sys.exit(print(f'Error: actor "{actor}" in atespace "{space}" not found', file=sys.stderr) or 1) +if verb == "get": + if not store.exists(): # kubectl-ate: no policy is a valid state, exit 0 with nothing on stdout + sys.exit(print(f'actor "{actor}" in atespace "{space}" has no egress policy', file=sys.stderr) or 0) + print(store.read_text()) +elif verb in ("create", "update"): + body = json.loads(sys.stdin.read()) + old = json.loads(store.read_text()) if store.exists() else None + if verb == "create" and old is not None: + sys.exit(print("Error: already exists", file=sys.stderr) or 1) + version = 1 if old is None else old["metadata"]["version"] + 1 + body["metadata"] = {"name": "default", "atespace": space, "uid": "uid-1", "version": version} + store.parent.mkdir(exist_ok=True) + store.write_text(json.dumps(body)) + (HERE / "policies" / f"{space}_{actor}.written.json").write_text(json.dumps(body)) # survives the cascade + print(f"egress policy {verb}d") +''' + + +@pytest.fixture +def kubectl_ate(tmp_path, monkeypatch: pytest.MonkeyPatch): + """A fake ``kubectl-ate`` in ``tmp_path`` (shared with the fake ax there); no settle wait.""" + path = tmp_path / "kubectl-ate" + path.write_text(FAKE_KUBECTL_ATE.replace("{python}", sys.executable), encoding="utf-8") + path.chmod(0o755) + monkeypatch.setenv("OUSAST_KUBECTL_ATE", str(path)) + monkeypatch.setenv("OUSAST_EGRESS_SETTLE", "0") + return path diff --git a/tests/test_plane_egress.py b/tests/test_plane_egress.py new file mode 100644 index 0000000..e9b5749 --- /dev/null +++ b/tests/test_plane_egress.py @@ -0,0 +1,199 @@ +"""ai-service-plane task 0: the per-task egress policy, its kubectl-ate calls, and the runner's dialled delivery. + +Agent Substrate's egress gateway denies by default; :func:`egress.policy_for` must grant exactly the receiver (HTTP +by name), the bound Workspaces' Git hosts and the bound Model's API host (TLS passthrough), and nothing a gateway +would reject or that would intercept TLS: no IP address, no wildcard, no ``https`` rule. +""" + +from __future__ import annotations + +import ipaddress +import json +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import pytest + +from openultrasast.plane import egress, runner +from openultrasast.plane.manifests import ManifestError, load_manifests, parse_manifest + +MANIFESTS = """ +apiVersion: ax.io/v1alpha1 +kind: Model +metadata: + name: deepseek-flash + annotations: {openultrasast.io/provider-extension: "true", openultrasast.io/egress-hosts: api.deepseek.com} +spec: {provider: deepseek, model: deepseek-flash, secretKey: {name: deepseek, key: DEEPSEEK_API_KEY}} +--- +apiVersion: ax.io/v1alpha1 +kind: Workspace +metadata: {name: cases} +spec: + git: + - {name: a, repo: https://github.com/example-org/alpha} + - {name: b, repo: https://github.com/example-org/beta.git} + - {name: c, repo: "git@gitlab.example.org:group/repo.git"} + - {name: d, repo: https://10.0.0.7/mirror/repo.git} +--- +apiVersion: ax.io/v1alpha1 +kind: Task +metadata: {name: verify, annotations: {openultrasast.io/model: deepseek-flash}} +spec: {command: [verify], workspaces: [{name: cases, path: /workspace}]} +--- +apiVersion: ax.io/v1alpha1 +kind: Task +metadata: {name: facts} +spec: {command: [repo-facts]} +""" + + +@pytest.fixture +def manifests(tmp_path: Path): + path = tmp_path / "m.yaml" + path.write_text(MANIFESTS, encoding="utf-8") + return load_manifests([path]) + + +def hostnames(policy: dict) -> list[str]: + return [h for rule in policy["rules"] for body in rule.values() for h in body["hostnames"]] + + +def assert_tight(policy: dict) -> None: + for rule in policy["rules"]: + assert set(rule) <= {"http", "tlsPassthrough"}, f"only http and passthrough rules, got {rule}" + body = next(iter(rule.values())) + assert "all" not in body["ports"], "never every port" + for host in hostnames(policy): + assert "*" not in host, f"no wildcard: {host}" + with pytest.raises(ValueError): + ipaddress.ip_address(host) + + +def test_model_free_task_gets_only_the_receiver(manifests) -> None: + policy = egress.policy_for(manifests.tasks["facts"], [], None, egress.RECEIVER_HOST) + assert policy == {"rules": [{"http": {"hostnames": [egress.RECEIVER_HOST], "ports": {"numbers": [80]}}}]} + assert_tight(policy) + + +def test_model_bound_task_with_git_workspaces(manifests) -> None: + task = manifests.tasks["verify"] + policy = egress.policy_for(task, [manifests.workspaces["cases"]], manifests.models["deepseek-flash"], egress.RECEIVER_HOST) + assert policy["rules"] == [ + {"http": {"hostnames": [egress.RECEIVER_HOST], "ports": {"numbers": [80]}}}, + {"tlsPassthrough": {"hostnames": ["api.deepseek.com", "github.com"], "ports": {"numbers": [443]}}}, + ], "one github.com for two repos; the ssh and IP-address repos add nothing" + assert_tight(policy) + + +def test_an_ip_receiver_is_not_granted(manifests) -> None: + assert egress.policy_for(manifests.tasks["facts"], [], None, "127.0.0.1") == {"rules": []} + + +def test_deepseek_model_manifest_names_the_endpoint_host() -> None: + from openultrasast.model.endpoint import DEEPSEEK_BASE_URL + + model = load_manifests([Path("plane/models/deepseek-flash.yaml")]).models["deepseek-flash"] + assert model.egress_hosts == (DEEPSEEK_BASE_URL.removeprefix("https://"),) + + +@pytest.mark.parametrize("hosts", ["*.deepseek.com", "10.1.2.3", "https://api.deepseek.com", "api.deepseek.com:443", "API.deepseek.com"]) +def test_egress_hosts_annotation_is_validated(hosts: str) -> None: + doc = { + "apiVersion": "ax.io/v1alpha1", + "kind": "Model", + "metadata": {"name": "m", "annotations": {"openultrasast.io/provider-extension": "true", "openultrasast.io/egress-hosts": hosts}}, + "spec": {"provider": "deepseek", "model": "x"}, + } + with pytest.raises(ManifestError, match="egress-hosts"): + parse_manifest(doc) + + +# --- kubectl-ate --------------------------------------------------------------------------------------------------- + + +def actor(tmp_path: Path, name: str, gone: bool = False) -> None: + (tmp_path / "applied").mkdir(exist_ok=True) + (tmp_path / "applied" / f"{name}.json").write_text(json.dumps({"gone": gone}), encoding="utf-8") + + +def calls(tmp_path: Path) -> list[str]: + return [" ".join(line.split()[2:4]) for line in (tmp_path / "ax.log").read_text().splitlines()] + + +POLICY = {"rules": [{"http": {"hostnames": [egress.RECEIVER_HOST], "ports": {"numbers": [80]}}}]} + + +def test_apply_creates_reads_back_and_delete_confirms_the_cascade(tmp_path: Path, kubectl_ate: Path) -> None: + cli = egress.Egress(str(kubectl_ate), context="kind-test") + actor(tmp_path, "t1") + cli.apply("t1", "default", POLICY) + stored = cli.get("t1", "default") + assert stored is not None and egress.rules_of(stored) == egress.rules_of(POLICY) + assert cli.delete("t1", "default") == "egress policy for default/t1 outlived its actor" + actor(tmp_path, "t1", gone=True) # ax delete task: the actor, and with it the policy, is gone + assert cli.delete("t1", "default") is None + assert calls(tmp_path) == ["get t1", "create t1", "get t1", "get t1", "get t1", "get t1"] + + +def test_apply_replaces_an_existing_policy(tmp_path: Path, kubectl_ate: Path) -> None: + cli = egress.Egress(str(kubectl_ate)) + actor(tmp_path, "t2") + cli.apply("t2", "default", {"rules": []}) + cli.apply("t2", "default", POLICY) + assert calls(tmp_path) == ["get t2", "create t2", "get t2", "get t2", "update t2", "get t2"] + assert json.loads((tmp_path / "policies" / "default_t2.json").read_text())["metadata"]["version"] == 2 + + +def test_apply_waits_for_the_actor_then_gives_up(tmp_path: Path, kubectl_ate: Path) -> None: + cli = egress.Egress(str(kubectl_ate)) + with pytest.raises(RuntimeError, match="not found"): + cli.apply("missing", "default", POLICY, timeout=0.2) + timer = threading.Timer(0.3, actor, args=(tmp_path, "late")) + timer.start() + cli.apply("late", "default", POLICY, timeout=10) + timer.join() + + +def test_receiver_address_modes(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("OUSAST_ARTIFACT_HOST", "127.0.0.1") + assert egress.receiver_address(4242) == ("http://127.0.0.1:4242/", None) + monkeypatch.delenv("OUSAST_ARTIFACT_HOST") + monkeypatch.setenv("OUSAST_ARTIFACT_DIAL", "172.19.0.1:80") + assert egress.receiver_address(18090) == (f"http://{egress.RECEIVER_HOST}/", "172.19.0.1:80") + + +# --- the runner dials the gateway address with the receiver's name as Host ---------------------------------------- + + +class _Seen(ThreadingHTTPServer): + daemon_threads = True + seen: list[tuple[str, str | None, bytes]] + + +class _Handler(BaseHTTPRequestHandler): + server: _Seen + + def do_POST(self) -> None: # noqa: N802 + body = self.rfile.read(int(self.headers.get("Content-Length") or 0)) + self.server.seen.append((self.path, self.headers.get("Host"), body)) + self.send_response(200) + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, format: str, *args: object) -> None: # noqa: A002 + pass + + +def test_delivery_dials_the_override_with_the_url_host() -> None: + server = _Seen(("127.0.0.1", 0), _Handler) + server.seen = [] + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + dial = f"127.0.0.1:{server.server_address[1]}" + url = f"http://{egress.RECEIVER_HOST}/in" + assert runner.deliver(url, b"tar", {"Content-Type": "application/x-tar"}, backoff=0, dial=dial) == 200 + assert server.seen == [("/in", egress.RECEIVER_HOST, b"tar")] + finally: + server.shutdown() + server.server_close() diff --git a/tests/test_plane_reconciler.py b/tests/test_plane_reconciler.py index ced96e7..a6ac04f 100644 --- a/tests/test_plane_reconciler.py +++ b/tests/test_plane_reconciler.py @@ -156,7 +156,7 @@ apiVersion: ax.io/v1alpha1 kind: Model metadata: name: deepseek-flash - annotations: {openultrasast.io/provider-extension: "true"} + annotations: {openultrasast.io/provider-extension: "true", openultrasast.io/egress-hosts: api.deepseek.com} spec: provider: deepseek model: deepseek-chat @@ -269,6 +269,8 @@ class Fake: out.append((stamp, Path(words[2]).parent.name)) elif verb in ("delete", "resume") and words[:2] == [verb, "task"]: out.append((stamp, words[2])) + elif verb == "egress" and words[0] == verb: + out.append((stamp, f"{words[1]} {words[2]}")) elif verb in ("delivered", "rejected") and words[0] == verb: out.append((stamp, words[1])) return out @@ -313,7 +315,7 @@ class _RouterHandler(BaseHTTPRequestHandler): @pytest.fixture -def fake(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[Fake]: +def fake(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, kubectl_ate: Path) -> Iterator[Fake]: monkeypatch.setenv("OUSAST_RESULTS", str(tmp_path / "results")) monkeypatch.setenv("OUSAST_ARTIFACT_HOST", "127.0.0.1") monkeypatch.setenv("OUSAST_POLL_SECONDS", "0.02") @@ -747,3 +749,34 @@ def test_doctor_treats_an_empty_namespace_as_not_ready(monkeypatch: pytest.Monke assert doctor._pods_ready("ns", "ctx") == (True, "2 pods ready") monkeypatch.setattr(doctor, "_sh", lambda *a: (True, "a-1 0/1 Running 0 1m")) assert doctor._pods_ready("ns", "ctx")[0] is False + + +# --- egress (task 0: the per-task policy) -------------------------------------------------------------------------- + + +def test_egress_policy_between_apply_and_resume_and_checked_after_delete(fake: Fake, tmp_path: Path) -> None: + """apply -> policy -> resume -> start -> delivery -> delete task -> the policy is gone with its actor.""" + assert reconciler.run(write_run(tmp_path, "eg"), ax=str(fake.ax)) == "done" + lines = [" ".join(words) for _, words in fake.log()] + + def first(prefix: str, after: int = 0) -> int: + return next(i for i, line in enumerate(lines) if i >= after and line.startswith(prefix)) + + order = [first("apply -f " + str(reconciler.run_dir("eg") / "verify"))] + steps = ("egress create eg-verify", "resume task eg-verify", "started eg-verify 202", "delivered verify", "delete task eg-verify") + for prefix in (*steps, "egress get eg-verify"): # the last one: the check after the delete + order.append(first(prefix, order[-1] + 1)) + assert order == sorted(order) + assert not (tmp_path / "policies" / "default_eg-verify.json").exists(), "the policy went with its actor" + written = json.loads((tmp_path / "policies" / "default_eg-verify.written.json").read_text()) + tls = {"hostnames": ["api.deepseek.com", "example.invalid"], "ports": {"numbers": [443]}} + assert written["rules"] == [{"tlsPassthrough": tls}], "the Model host and the Git host, nothing else" + facts = json.loads((tmp_path / "policies" / "default_eg-repo-facts.written.json").read_text()) + assert facts["rules"] == [{"tlsPassthrough": {"hostnames": ["example.invalid"], "ports": {"numbers": [443]}}}] + + +def test_a_refused_egress_policy_fails_the_task_before_resume(fake: Fake, tmp_path: Path, kubectl_ate: Path) -> None: + kubectl_ate.write_text("#!/bin/sh\necho 'Error: permission denied' >&2\nexit 1\n", encoding="utf-8") + assert reconciler.run(write_run(tmp_path, "egfail"), ax=str(fake.ax)) == "failed" + assert "permission denied" in state_of("egfail")["tasks"]["repo-facts"]["reason"] + assert fake.events("resume") == [] and [n for _, n in fake.events("delete")] == ["egfail-repo-facts"] diff --git a/tests/test_plane_runner.py b/tests/test_plane_runner.py index ca4112d..06953c6 100644 --- a/tests/test_plane_runner.py +++ b/tests/test_plane_runner.py @@ -244,6 +244,10 @@ def test_output_dir_is_delivered_as_a_tar_with_the_headers( assert headers["Content-Type"] == "application/x-tar" assert headers["X-Ousast-Task"] == "stub" assert headers["X-Ousast-Run"] == "validation-46" + receiver.posts.clear() + again = {"OUSAST_ARTIFACT_URL": receiver.url, "OUSAST_TASK": "facts", "OUSAST_STATE_DIR": str(tmp_path / "again-state")} + code, _ = run_main(monkeypatch, tmp_path / "again", files_workspace_yaml(), **again) + assert code == 0 and receiver.posts[0][0]["X-Ousast-Task"] == "facts", "the Run's task name, not the ax Task's" with tarfile.open(fileobj=io.BytesIO(body), mode="r:") as archive: names = sorted(archive.getnames()) summary = archive.extractfile("summary.json")