diff --git a/.github/workflows/k3s-p0-contract.yml b/.github/workflows/k3s-p0-contract.yml new file mode 100644 index 0000000000000..5cb9e735df5e4 --- /dev/null +++ b/.github/workflows/k3s-p0-contract.yml @@ -0,0 +1,26 @@ +name: K3s P0 offline contract +on: + pull_request: + paths: + - "examples/k3s-p0/**" + - ".github/workflows/k3s-p0-contract.yml" + push: + branches-ignore: [main] + paths: + - "examples/k3s-p0/**" + - ".github/workflows/k3s-p0-contract.yml" + workflow_dispatch: +jobs: + offline: + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Install YAML parser + run: python -m pip install "PyYAML==6.0.2" + - name: Validate bounded P0 manifest and digest renderer + run: python -m unittest discover -s examples/k3s-p0 -p 'test_*.py' -v diff --git a/examples/k3s-p0/Dockerfile b/examples/k3s-p0/Dockerfile new file mode 100644 index 0000000000000..3552af7432aaa --- /dev/null +++ b/examples/k3s-p0/Dockerfile @@ -0,0 +1,47 @@ +# P0 proof only: one persistent server, no worker provisioning. +# The released, self-contained CLI is verified against upstream SHA256SUMS. +# The Debian base tag is not digest-pinned; record/pin the built image digest before deployment. +FROM debian:bookworm-slim + +ARG TARGETARCH +ARG T3_VERSION=0.1.3 +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates curl dumb-init git openssh-client tar \ + && rm -rf /var/lib/apt/lists/* + +RUN set -eu; \ + case "${TARGETARCH}" in \ + amd64) t3_arch=x64 ;; \ + arm64) t3_arch=arm64 ;; \ + *) echo "Unsupported TARGETARCH: ${TARGETARCH}" >&2; exit 1 ;; \ + esac; \ + archive="t3-${T3_VERSION}-linux-${t3_arch}.tar.gz"; \ + base="https://github.com/pingdotgg/t3code/releases/download/v${T3_VERSION}"; \ + curl --fail --silent --show-error --location "${base}/SHA256SUMS" -o /tmp/SHA256SUMS; \ + curl --fail --silent --show-error --location "${base}/${archive}" -o /tmp/t3.tar.gz; \ + expected="$(awk -v name="${archive}" '{ n=$2; if (substr(n, 1, 1) == "*") n=substr(n, 2); if (n == name) { print $1; exit } }' /tmp/SHA256SUMS)"; \ + test -n "${expected}"; \ + printf '%s %s\n' "${expected}" /tmp/t3.tar.gz | sha256sum --check --status; \ + mkdir -p /opt/t3; \ + tar -xzf /tmp/t3.tar.gz --strip-components=1 -C /opt/t3; \ + /opt/t3/t3 --version; \ + rm -f /tmp/SHA256SUMS /tmp/t3.tar.gz + +RUN groupadd --gid 10001 t3 \ + && useradd --uid 10001 --gid 10001 --home-dir /data --no-create-home --shell /bin/sh t3 \ + && mkdir -p /data \ + && chown 10001:10001 /data + +ENV HOME=/data \ + T3CODE_HOME=/data/t3 \ + XDG_CACHE_HOME=/data/.cache \ + XDG_CONFIG_HOME=/data/.config +USER 10001:10001 +WORKDIR /data +EXPOSE 3773 + +# 'serve' prints headless pairing details; 'start --no-browser' does not need +# to print one-time pairing URLs into long-lived Kubernetes pod logs. +# Pair explicitly and transiently via `t3 auth pairing create`. +ENTRYPOINT ["/usr/bin/dumb-init", "--", "/opt/t3/t3"] +CMD ["start", "--no-browser", "--host", "0.0.0.0", "--port", "3773", "--base-dir", "/data/t3"] diff --git a/examples/k3s-p0/README.md b/examples/k3s-p0/README.md new file mode 100644 index 0000000000000..38e0d5928abc6 --- /dev/null +++ b/examples/k3s-p0/README.md @@ -0,0 +1,233 @@ +# Experimental P0: persistent T3 environment on self-owned K3s + +**Downstream-only, NOT deployed/tested on a live cluster.** This is one **owner** +T3 server with a stable /data PVC and private Tailscale HTTPS ingress, not a +worker broker, task scheduler, T3 Connect replacement, or permissioned peer link. +Nothing is applied automatically. Keep the user's current T3 deployment intact. + +## Source and version boundary + +- Reviewed upstream `pingdotgg/t3code` main at + `5211acfe95a0c6d82a60e2524b5e46b47fe67a3e` (2026-10-28). The image + instead packages the **released** self-contained CLI `v0.1.3`, published + 2026-10-27. A release is not proof that every latest-main linked-environment + beta behavior exists in it. P1 must pin and validate a compatible build. +- Dockerfile checks the downloaded release archive against GitHub Release + `SHA256SUMS`, on either linux/amd64 or linux/arm64. Its Debian base tag is + **mutable**; this is not a hermetic source build. Record the final pushed + **OCI image digest**, and deploy by digest, never `:latest`. +- The base image supplies T3 + Git/SSH, but **no provider CLI or login**. + Provider installation, writable homedir needs, browser libraries and actual + workload execution must be evaluated separately. Do not bake provider secrets. +- The existing independent [kind -> K3s CPU Job pilot](https://github.com/kvnloo/oss-factory/issues/127) + / [draft #128](https://github.com/kvnloo/oss-factory/pull/128) tests bounded + Kubernetes Job rendering, not T3 server identity or delegation; reuse its + offline lessons, not its scheduler. + +## Prerequisites + +A disposable/opt-in **single-node** K3s context, working default StorageClass, +Tailscale Kubernetes Operator with `tailscale` IngressClass and proxy namespace +`tailscale`, MagicDNS + HTTPS certs, and correctly restricted tailnet ACLs/grants +for this workload. No Tailscale Funnel. Verify the operator's actual proxy +namespace and tailnet policy before use; this example assumes the default name. +K3s must enforce NetworkPolicy for the ingress allowlist to work. Pods do +**not** provide VM-grade isolation for untrusted code; the manifest leaves +**egress unrestricted** for the owner process. Use a dedicated cluster/node +for the first experiment, not existing household infrastructure. + +## Read-only cluster preflight (before any apply) + +On an **owner-approved disposable test context**, check basic API prerequisites +without creating, deleting or restarting anything: + +```sh +python3 examples/k3s-p0/preflight.py --context YOUR_DISPOSABLE_K3S_CONTEXT +``` + +The probe uses explicit-context `kubectl get` requests only. It requires +a stable K3s build on Kubernetes 1.28+ (`/version.gitVersion` must end in +`+k3sN`; generic Kubernetes distributions are rejected), exactly one +Ready/un-tainted/schedulable node, an **absent** +`t3-k3s-p0` namespace (protecting existing state), the `tailscale` namespace +and IngressClass controller `tailscale.com/ts-ingress`, and exactly one named +default StorageClass. Missing RBAC or API data fails closed; errors never copy +kubectl stderr that could expose private endpoints. No credentials are exported. + +**This cannot prove a cluster is disposable or yours.** It also does not test +whether the Tailscale operator is healthy, certificates are issued, the +selected StorageClass can actually bind a PVC, NetworkPolicy is enforced by the +CNI, or tailnet ACLs/route identity are correct. Verify those separately, and +review the selected kubeconfig context and physical host before approving any +live apply. A PASS means *prerequisite metadata only*, not P0 acceptance. + +## Offline proof (safe) + +```sh +python3 -m pip install 'PyYAML==6.0.2' +python3 -m unittest discover -s examples/k3s-p0 -p 'test_*.py' -v +docker buildx build --platform linux/amd64 -f examples/k3s-p0/Dockerfile \ + -t ghcr.io/REPLACE_OWNER/t3-k3s-p0:0.1.3 --load . +``` + +For a real test, push to a registry you control, find its pushed **manifest** +digest (not the local image ID), and render a copy. The renderer rejects tags, +missing digests, YAML injection and direct modification of tracked source. +It stages a private mode-0600 file, then publishes it without overwriting an +existing path or following an output symlink. Use a fresh output path on reruns: + +```sh +python3 examples/k3s-p0/render.py \ + --image ghcr.io/YOUR_OWNER/t3-k3s-p0@sha256:YOUR_64_LOWERCASE_HEX_DIGEST \ + --output /tmp/t3-k3s-p0.yaml +kubectl --context YOUR_DISPOSABLE_K3S_CONTEXT apply --dry-run=client -f /tmp/t3-k3s-p0.yaml +kubectl --context YOUR_DISPOSABLE_K3S_CONTEXT diff -f /tmp/t3-k3s-p0.yaml +``` + +**Inspect diff and context before a deliberate live apply.** Never point these +commands at an existing T3 state/PVC, and never reuse credentials from the +home machine. The checked-in YAML intentionally has an unpullable image token. + +## Human-only deployment and acceptance + +1. Confirm cluster context/namespace do not already contain any T3 resources. + Apply the rendered YAML once: `kubectl --context YOUR_DISPOSABLE_K3S_CONTEXT + apply -f /tmp/t3-k3s-p0.yaml`. Check pod readiness, warnings, storage + events and Tailscale operator ingress address. A TCP probe only proves a + listening socket; it does not prove T3 auth, UI or provider readiness. +2. From a trusted admin shell, generate a **short-lived** pairing URL *on + demand*, without sending it to GitHub or storing it in pod logs: + `kubectl --context YOUR_DISPOSABLE_K3S_CONTEXT -n t3-k3s-p0 exec + deployment/t3-owner -- /opt/t3/t3 auth pairing create --base-dir /data/t3 + --base-url https://ACTUAL_INGRESS_FQDN --ttl 5m`. + Pair a web/desktop/mobile client reachable via the same tailnet. Verify + the server descriptor and its environment ID. Pairing a client does **not** + create an OAuth peer link; linked environments beta is separately consented. +3. Before/after restart, compare + `/data/t3/userdata/environment-id` (do not post its value publicly), T3 + project/thread data, client connection and the tailnet HTTPS endpoint. + Restart with `kubectl --context YOUR_DISPOSABLE_K3S_CONTEXT -n t3-k3s-p0 + rollout restart deployment/t3-owner`. Confirm the identity **and** data are + unchanged. The fixed pod hostname mitigates `os.hostname()` drift in + usage fingerprints (see [upstream #11849](https://github.com/pingdotgg/t3code/issues/11849)), + but is not itself proof of identity persistence. + + For **read-only, reproducible identity evidence**, capture a local baseline + after the first verified Ready deployment, then verify it **after** the + human-triggered restart above (never point at the existing home cluster): + + ```sh + python3 examples/k3s-p0/acceptance.py --context YOUR_DISPOSABLE_K3S_CONTEXT \ + --write-baseline /tmp/t3-p0-identity.json + # Human explicitly performs rollout restart; wait for Ready before continuing. + python3 examples/k3s-p0/acceptance.py --context YOUR_DISPOSABLE_K3S_CONTEXT \ + --check-baseline /tmp/t3-p0-identity.json + ``` + + From an **authorized tailnet client**, independently check that the + **actual HTTPS endpoint** answers as the same persisted environment. Use the + private v3 baseline created above, before **and again after** the human + restart. This second script sends only a credential-free GET to the public + T3 descriptor, `/.well-known/t3/environment`, with a fresh HTTPS connection, + OS-trusted certificate and hostname verification. It compares the returned + ID as a hash and prints no ID or pairing token. It does **not** use an + insecure TLS bypass, redirects, proxy settings or a bearer token. + + ```sh + python3 examples/k3s-p0/descriptor_probe.py --baseline /tmp/t3-p0-identity.json + ``` + + A PASS adds **remote HTTPS/T3 descriptor identity** evidence, not tailnet + ACL or CNI enforcement, client pairing/auth, UI functionality, project data + restoration or provider readiness. Perform **negative ACL tests from an + unauthorized identity** and verify a real authenticated web/mobile client + and previously saved thread separately. DNS/TLS failure must block P0; + do not work around it with `curl -k` or a public proxy. Run this only on + a disposable owner-approved cluster and never share the private baseline. + + The probe uses only read-only `kubectl get` and `kubectl exec ... cat` + requests. It also verifies actual Pod image/labels, PVC mount, no API-token + mount or hostPath, Service selector/port, private Ingress backend/TLS, + and live EndpointSlice target Pod UID, IP, readiness and T3 port 3773. + It additionally rejects an *admitted Pod* whose service account, host + process/network namespaces, stable hostname/host aliases, CPU/memory/ephemeral + requests and limits, non-root/seccomp settings, container privilege, + commands/environments, or exact PVC/tmp mounts differ from the reviewed + fixture. A safe Deployment template alone cannot attest the effective Pod + after admission. It also verifies the P0 ResourceQuota's exact hard + limits, reconciled status and absence of scopes, plus the ServiceAccount's + disabled token automount and absence of secret references. These API checks + do **not** prove Pod Security Admission or quota/NetworkPolicy enforcement; + test those on the disposable cluster. + It checks the **exact requested** `t3-k3s-p0..ts.net` Ingress + hostname and rejects additional reported ingress addresses. It checks + **all namespace Services and Ingresses** against the one approved owner + route, rejecting hidden secondary/public routes, ClusterIP + `tailscale.com/expose` annotations, unexpected controller annotations and + external Service load-balancer status. This strict inventory can reject + operator-added routes until a human reviews them; a snapshot cannot rule + out resources added after the read. It also verifies that + **the namespace has exactly one unchanged, narrow ingress NetworkPolicy**. + Kubernetes policies are additive, so a second broad allow would invalidate + the isolation claim. This is intentionally strict for this isolated fixture: + additional operator-injected policies require review instead of a green + acceptance result. The check needs permission to **list NetworkPolicies**, + **get ResourceQuotas/ServiceAccounts**, and `pods/exec` permission to read the environment ID; it is not a + read-only-RBAC-only Kubernetes probe. It also rejects an + unobserved/incomplete Deployment rollout. This closes + false-green checks where an unrelated/stale Pod or Service looks Ready + without actually selecting the intended T3 data. The API evidence is + **not** a dataplane/TLS/ACL or CNI enforcement probe; test those separately. + A matching Ingress name does not prove its DNS resolves to the expected + private tailnet destination: independently resolve the hostname, test + HTTPS certificate/identity from an approved client, and confirm denied + access from an unauthorized tailnet identity and unrelated Pod namespace. It requires + **one Running/Ready Pod with a new Pod UID** after the + human-triggered rollout, while checking unchanged hashed environment ID, + PVC UID, bound PV UID and reclaim policy, ingress hostname and pinned image + digest. It checks the PV's claimRef UID, ReadWriteOnce Filesystem mode, + 10Gi request, 10Gi reported PVC/PV capacity and matching storage class; + missing/undersized backing capacity or an orphaned/rebound PV fails closed. + A PV with a `Delete` reclaim policy can still erase data if its claim + is deleted. The probe does not change reclaim behavior or prove on-disk + durability; it needs cluster-scoped permission to `get persistentvolumes`. + Merely running the probe twice without restarting must fail. Each baseline + is format v3; discard offline v1/v2 baselines and collect a new sample. It never + prints the raw ID, grants access, restarts a server or applies/deletes + resources. Keep the private identity fingerprint out of Git. A PASS is **not** proof of + working HTTPS, ACL enforcement, client pairing, saved threads, app readiness + or provider execution; perform the manual checks separately. + +4. Verify pod/container image IDs, CPU/memory/ephemeral limits, absence of + mounted Kubernetes tokens, tailnet policy denial from an unauthorized device, + and **negative ingress test from a non-Tailscale namespace**. Inspect + `kubectl logs` for inadvertent pairing tokens *privately*. Reject this + deployment if tokens are logged. Record K3s version, CNI, storage class, + image digest, T3 version, latency, restarts and errors. +5. Stop without deleting state: `kubectl --context YOUR_DISPOSABLE_K3S_CONTEXT + -n t3-k3s-p0 scale deployment/t3-owner --replicas=0`. + To remove the public-looking endpoint independently, delete the private + `Ingress/t3-owner` in this namespace. **Do not** use + `kubectl delete -f /tmp/t3-k3s-p0.yaml` or delete the namespace as a + cleanup shortcut: this includes the PVC, and K3s local-path reclaim can + permanently delete the data. Retain/archive/delete PVC only after a + deliberate data-retention decision. + +## Gated next stage + +P1 will add an **external opt-in** broker that allocates separate worker +namespaces/PVCs using an idempotent lease, without changing T3 core or touching +the P0 owner. A client environment entry is not a peer authorization grant. +Require explicit `t3 environment link` OAuth consent, stable descriptor +verification, a pinned repository revision and one observed remote +`delegate_task` receipt. Upstream [#17051](https://github.com/pingdotgg/t3code/issues/17051) +still describes the record-before-launch gap: don't claim exactly-once +delegation until owner restart/lost-response tests prove it. Watch upstream +[#26475](https://github.com/pingdotgg/t3code/pull/26475) rather than using +plain `100.x` HTTP as a trust shortcut. No automatic worker enrollment into +client load balancing or shared worktrees. + +**Verification status:** offline contract tests and image/K3s/tailnet/client +acceptance must be separately recorded on [downstream issue #45](https://github.com/kvnloo/t3code/issues/45). +No live K3s, Docker build, provider, device, pairing or delegation result is +implied by this example. diff --git a/examples/k3s-p0/acceptance.py b/examples/k3s-p0/acceptance.py new file mode 100644 index 0000000000000..9dbfce0dea781 --- /dev/null +++ b/examples/k3s-p0/acceptance.py @@ -0,0 +1,602 @@ +#!/usr/bin/env python3 +"""Read-only, opt-in K3s P0 identity snapshot before/after a HUMAN restart. + +No cluster mutations, pairing grants, or network exposure. Never prints a raw +T3 environment ID. This does not establish functional client/agent readiness. +""" + +import argparse +import hashlib +import json +import os +from pathlib import Path +import subprocess +import stat + +NAMESPACE = "t3-k3s-p0" + + +class EvidenceError(RuntimeError): + """The observed cluster does not match the bounded P0 acceptance contract.""" + + +def kubectl(context: str, *args: str) -> str: + if not context.strip() or context.startswith("-"): + raise EvidenceError("an explicit kubectl context is required") + command = ["kubectl", "--context", context, *args] + try: + result = subprocess.run(command, capture_output=True, text=True, timeout=30, check=False) + except (OSError, subprocess.TimeoutExpired) as exc: + raise EvidenceError("kubectl read-only query failed or timed out") from exc + if result.returncode: + # Some kubectl errors include sensitive data; do not echo stderr. + raise EvidenceError("kubectl read-only query failed; inspect locally") + return result.stdout + + +def verify_admitted_pod_confinement(pod: dict) -> None: + """Reject unsafe admission mutations to the actual P0 owner Pod. + + A safe Deployment template or a Ready Pod alone cannot prove the running + Pod retained the intended process, credential, mount and network limits. + This remains API evidence, not proof the node/CNI enforces the settings. + """ + spec = pod.get("spec") + if not isinstance(spec, dict): + raise EvidenceError("P0 Pod spec is missing") + # Usage host identity depends on os.hostname(); a webhook may mutate the + # admitted Pod independently of the stable Deployment template. + if (spec.get("hostname") != "t3-owner" + or spec.get("subdomain") not in (None, "") + or spec.get("setHostnameAsFQDN") not in (None, False) + or spec.get("hostAliases")): + raise EvidenceError("P0 Pod hostname or host aliases drifted") + if (spec.get("serviceAccountName") != "t3-p0" + or spec.get("automountServiceAccountToken") is not False + or any(spec.get(flag) for flag in ("hostNetwork", "hostPID", "hostIPC", + "shareProcessNamespace")) + or spec.get("initContainers") or spec.get("ephemeralContainers") + or spec.get("imagePullSecrets")): + raise EvidenceError("P0 Pod has unsafe process, network or credential access") + pod_security = spec.get("securityContext") + expected_pod_security = { + "runAsNonRoot": True, "runAsUser": 10001, "runAsGroup": 10001, + "fsGroup": 10001, "seccompProfile": {"type": "RuntimeDefault"}, + } + if (not isinstance(pod_security, dict) + or any(pod_security.get(key) != value + for key, value in expected_pod_security.items()) + or pod_security.get("sysctls")): + raise EvidenceError("P0 Pod security context drifted from restricted template") + + containers = spec.get("containers") + if not isinstance(containers, list) or len(containers) != 1: + raise EvidenceError("P0 Pod must have exactly one owner container") + app = containers[0] + if not isinstance(app, dict): + raise EvidenceError("P0 owner container is invalid") + security = app.get("securityContext") + capabilities = security.get("capabilities") if isinstance(security, dict) else None + seccomp = security.get("seccompProfile", {}) if isinstance(security, dict) else None + if (not isinstance(security, dict) + or not isinstance(capabilities, dict) + or not isinstance(seccomp, dict) + or security.get("allowPrivilegeEscalation") is not False + or security.get("readOnlyRootFilesystem") is not True + or security.get("privileged") is True + or capabilities.get("drop") != ["ALL"] + or capabilities.get("add") + or seccomp.get("type", "RuntimeDefault") != "RuntimeDefault" + or security.get("procMount", "Default") != "Default" + # Container-level identity settings override the safe Pod defaults. + # The reviewed template has none; even equal-value injections are drift. + or any(field in security for field in ("runAsUser", "runAsGroup", + "runAsNonRoot")) + # An injected lifecycle hook can execute unreviewed commands. + or "lifecycle" in app + or any(app.get(field) for field in ("env", "envFrom", "command", "args"))): + raise EvidenceError("P0 owner container privilege or command drifted") + + resources = app.get("resources") + if (not isinstance(resources, dict) + or resources.get("requests") != { + "cpu": "500m", "memory": "1Gi", "ephemeral-storage": "512Mi"} + or resources.get("limits") != { + "cpu": "2", "memory": "4Gi", "ephemeral-storage": "2Gi"} + or set(resources) != {"requests", "limits"}): + raise EvidenceError("P0 admitted container resource bounds drifted") + + volumes = spec.get("volumes") + if (not isinstance(volumes, list) or len(volumes) != 2 + or any(not isinstance(volume, dict) for volume in volumes)): + raise EvidenceError("P0 Pod contains unexpected volumes") + named = {volume.get("name"): volume for volume in volumes} + data, tmp = named.get("data", {}), named.get("tmp", {}) + data_source = data.get("persistentVolumeClaim") + tmp_source = tmp.get("emptyDir") + if (set(named) != {"data", "tmp"} + or set(data) != {"name", "persistentVolumeClaim"} + or not isinstance(data_source, dict) + or data_source.get("claimName") != "t3-p0-data" + or data_source.get("readOnly") is True + or set(tmp) != {"name", "emptyDir"} + or not isinstance(tmp_source, dict) + or tmp_source.get("sizeLimit") != "1Gi" + or tmp_source.get("medium") not in (None, "")): + raise EvidenceError("P0 Pod volume source drifted") + mounts = app.get("volumeMounts") + expected_mounts = {("data", "/data"), ("tmp", "/tmp")} + if (not isinstance(mounts, list) or len(mounts) != 2 + or any(not isinstance(mount, dict) for mount in mounts) + or {(mount.get("name"), mount.get("mountPath")) for mount in mounts} != expected_mounts + or any(set(mount) - {"name", "mountPath", "readOnly"} + or mount.get("readOnly") is True for mount in mounts)): + raise EvidenceError("P0 owner container has unexpected mounts") + + +def verify_p0_wiring(pod: dict, deployment_image: str, service: dict, ingress: dict) -> None: + """Fail closed when Ready resources are not wired to this exact T3 owner. + + A listening TCP socket and a populated ingress hostname alone do not prove + that the ingress reaches the expected pod, image, and persistent volume. + This does not claim network-policy enforcement, authentication or pairing. + """ + meta = pod.get("metadata", {}) + spec = pod.get("spec", {}) + containers = spec.get("containers", []) + if (meta.get("labels", {}).get("app") != "t3-owner" + or len(containers) != 1 or containers[0].get("name") != "t3" + or containers[0].get("image") != deployment_image): + raise EvidenceError("selected Pod does not match pinned T3 Deployment") + if spec.get("automountServiceAccountToken") is not False: + raise EvidenceError("selected Pod must not mount a Kubernetes API token") + volumes = spec.get("volumes", []) + if any("hostPath" in volume for volume in volumes): + raise EvidenceError("selected Pod mounts an unexpected hostPath") + if (not any(mount.get("name") == "data" and mount.get("mountPath") == "/data" + for mount in containers[0].get("volumeMounts", [])) + or not any(volume.get("name") == "data" + and volume.get("persistentVolumeClaim", {}).get("claimName") == "t3-p0-data" + for volume in volumes)): + raise EvidenceError("selected Pod is not backed by the expected persistent T3 PVC") + + verify_admitted_pod_confinement(pod) + + # The Ingress is not the only Kubernetes route to this Service. Even a + # ClusterIP can gain an independent tailnet proxy via tailscale.com/expose. + # Require an exact opt-in route contract; reject injected controller hooks. + for kind, obj in (("Service", service), ("Ingress", ingress)): + metadata = obj.get("metadata") + annotations = metadata.get("annotations", {}) if isinstance(metadata, dict) else None + if (not isinstance(annotations, dict) + or set(annotations) - {"kubectl.kubernetes.io/last-applied-configuration"}): + raise EvidenceError(f"P0 {kind} has unapproved exposure annotations") + status = service.get("status", {}) + if (not isinstance(status, dict) + or not isinstance(status.get("loadBalancer", {}), dict) + or status.get("loadBalancer", {}).get("ingress")): + raise EvidenceError("P0 Service advertises an unexpected external route") + + svc = service.get("spec", {}) + if any(svc.get(field) for field in ("externalName", "loadBalancerIP", + "loadBalancerClass", "externalTrafficPolicy", + "healthCheckNodePort", "publishNotReadyAddresses", + "allocateLoadBalancerNodePorts")): + raise EvidenceError("P0 Service has unapproved exposure fields") + ports = svc.get("ports", []) + if (svc.get("type") != "ClusterIP" or svc.get("externalIPs") + or svc.get("selector") != {"app": "t3-owner"} + or len(ports) != 1 + or ports[0].get("name") != "http" + or ports[0].get("port") != 80 + or ports[0].get("targetPort") != "http"): + raise EvidenceError("P0 Service does not exclusively route to the T3 owner") + ing = ingress.get("spec", {}) + backend = ing.get("defaultBackend", {}).get("service", {}) + tls = ing.get("tls", []) + if (ing.get("rules") + or backend.get("name") != "t3-owner" + or backend.get("port", {}).get("name") != "http" + or len(tls) != 1 or tls[0].get("hosts") != ["t3-k3s-p0"]): + raise EvidenceError("P0 Ingress is not wired to the private T3 owner Service") + + +def verify_service_endpoints(pod: dict, endpoint_slices: dict) -> None: + """Prove Service's live EndpointSlices resolve only to this Ready Pod. + + This verifies API controller wiring, not actual CNI routing, TLS, or ACLs. + """ + meta = pod.get("metadata", {}) + status = pod.get("status", {}) + pod_name, pod_uid = meta.get("name"), meta.get("uid") + pod_ips = {status.get("podIP")} if status.get("podIP") else set() + for address in status.get("podIPs", []): + if isinstance(address, dict) and address.get("ip"): + pod_ips.add(address["ip"]) + if not pod_name or not pod_uid or not pod_ips: + raise EvidenceError("P0 Pod identity/IP is missing from its status") + slices = endpoint_slices.get("items") if isinstance(endpoint_slices, dict) else None + if not isinstance(slices, list) or not slices: + raise EvidenceError("P0 Service has no observed EndpointSlices") + for item in slices: + if not isinstance(item, dict) or item.get("metadata", {}).get("labels", {}).get( + "kubernetes.io/service-name") != "t3-owner": + raise EvidenceError("P0 EndpointSlice belongs to a different Service") + ports = item.get("ports", []) + if (not isinstance(ports, list) or len(ports) != 1 + or not isinstance(ports[0], dict) + or ports[0].get("name") != "http" + or ports[0].get("port") != 3773 + or ports[0].get("protocol", "TCP") != "TCP"): + raise EvidenceError("P0 EndpointSlice does not route to T3 port 3773") + endpoints = item.get("endpoints", []) + if not isinstance(endpoints, list) or len(endpoints) != 1: + raise EvidenceError("P0 EndpointSlice must select exactly one T3 Pod") + endpoint = endpoints[0] + ref = endpoint.get("targetRef", {}) + conditions = endpoint.get("conditions", {}) + if (ref.get("kind") != "Pod" or ref.get("name") != pod_name + or ref.get("uid") != pod_uid + or ref.get("namespace", "t3-k3s-p0") != "t3-k3s-p0" + or conditions.get("ready") is not True + or conditions.get("terminating") is True): + raise EvidenceError("P0 EndpointSlice selects an unready/stale Pod") + addresses = endpoint.get("addresses", []) + if (not isinstance(addresses, list) or not addresses + or any(not isinstance(ip, str) or ip not in pod_ips for ip in addresses)): + raise EvidenceError("P0 EndpointSlice routes to a stale/wrong Pod IP") + + + +def verify_private_ingress(ingress: dict) -> str: + """Reject unexpected public/second routes and mismatched tailnet identity. + + A plausible .ts.net suffix alone is insufficient: the DNS device label + must match the explicit Tailscale Ingress name in this P0 fixture. + This verifies API metadata, not TLS or tailnet ACL enforcement. + """ + addresses = ingress.get("status", {}).get("loadBalancer", {}).get("ingress") + if not isinstance(addresses, list) or len(addresses) != 1: + raise EvidenceError("P0 Ingress must advertise exactly one private hostname") + entry = addresses[0] + if not isinstance(entry, dict) or entry.get("ip"): + raise EvidenceError("P0 Ingress must not advertise a public/IP route") + hostname = entry.get("hostname") + labels = hostname.split(".") if isinstance(hostname, str) else [] + if (len(labels) != 4 or labels[0] != "t3-k3s-p0" + or labels[2:] != ["ts", "net"] + or not labels[1] + or labels[1].startswith("-") or labels[1].endswith("-") + or len(labels[1]) > 63 + or any(ch not in "abcdefghijklmnopqrstuvwxyz0123456789-" for ch in labels[1])): + raise EvidenceError("P0 Ingress hostname does not match the requested tailnet name") + return hostname + + +def verify_private_network_policy(policies: dict) -> None: + """Conservatively reject missing or extra additive policies for this P0 namespace. + + NetworkPolicies are additive: verifying one safe policy while another + allows all traffic would otherwise produce a false-green result. + CNI enforcement still requires a separately observed negative test. + """ + items = policies.get("items") if isinstance(policies, dict) else None + if not isinstance(items, list) or len(items) != 1: + raise EvidenceError("P0 requires exactly one isolated ingress NetworkPolicy") + item = items[0] + if not isinstance(item, dict) or item.get("metadata", {}).get("name") != "t3-p0-ingress": + raise EvidenceError("unexpected P0 NetworkPolicy identity") + expected = { + "podSelector": {"matchLabels": {"app": "t3-owner"}}, + "policyTypes": ["Ingress"], + "ingress": [{ + "from": [{"namespaceSelector": { + "matchLabels": {"kubernetes.io/metadata.name": "tailscale"}}}], + "ports": [{"protocol": "TCP", "port": 3773}], + }], + } + if item.get("spec") != expected: + raise EvidenceError("P0 NetworkPolicy allows a different ingress scope") + +def verify_exclusive_routes(service: dict, ingress: dict, + services: dict, ingresses: dict) -> None: + """Reject a second namespace route to the P0 owner, including public ones. + + API snapshots do not establish actual CNI/tailnet enforcement or prevent + later mutations; those require separate negative runtime probes. + """ + for name, listed, expected, list_kind, item_kind in ( + ("Service", services, service, "ServiceList", "Service"), + ("Ingress", ingresses, ingress, "IngressList", "Ingress"), + ): + items = listed.get("items") if isinstance(listed, dict) else None + if (not isinstance(listed, dict) or listed.get("kind") != list_kind + or not isinstance(items, list) or len(items) != 1 + or not isinstance(items[0], dict) + or items[0].get("kind") != item_kind + or not isinstance(items[0].get("metadata"), dict) + or items[0]["metadata"].get("name") != "t3-owner" + or items[0] != expected): + raise EvidenceError(f"P0 has an extra, missing, or changed {name} route") + + +def verify_persistent_volume(pvc: dict, pv: dict) -> tuple[str, str]: + """Verify exact owner PVC/PV binding, not on-disk durability or backups.""" + if not isinstance(pvc, dict) or not isinstance(pv, dict): + raise EvidenceError("P0 PVC/PV binding objects are invalid") + cm, cs, ct = (pvc.get(k) for k in ("metadata", "spec", "status")) + vm, vs, vt = (pv.get(k) for k in ("metadata", "spec", "status")) + if not all(isinstance(o, dict) for o in (cm, cs, ct, vm, vs, vt)): + raise EvidenceError("P0 PVC/PV binding fields are missing") + claim = vs.get("claimRef") + class_name = cs.get("storageClassName") + if (cm.get("name") != "t3-p0-data" or cm.get("namespace") != NAMESPACE + or not isinstance(cm.get("uid"), str) or not cm["uid"] + or not isinstance(vm.get("uid"), str) or not vm["uid"] + or cm.get("deletionTimestamp") or vm.get("deletionTimestamp") + or ct.get("phase") != "Bound" or vt.get("phase") != "Bound" + or vm.get("name") != cs.get("volumeName") + or not isinstance(claim, dict) + or claim.get("name") != "t3-p0-data" + or claim.get("namespace") != NAMESPACE + or claim.get("uid") != cm["uid"]): + raise EvidenceError("P0 PVC/PV is not bound to this exact owner claim") + if (cs.get("accessModes") != ["ReadWriteOnce"] + or vs.get("accessModes") != ["ReadWriteOnce"] + or cs.get("volumeMode") != "Filesystem" + or vs.get("volumeMode") != "Filesystem" + or cs.get("resources", {}).get("requests", {}).get("storage") != "10Gi" + or cs.get("dataSource") or cs.get("dataSourceRef") + or not isinstance(class_name, str) or not class_name + or vs.get("storageClassName") != class_name): + raise EvidenceError("P0 PVC/PV access, storage class or source drifted") + # The PVC request alone does not attest its bound backing capacity. + # This fixed-size P0 fixture requires both API-reported capacities to + # match 10Gi; actual bytes/durability still need physical validation. + pv_capacity = vs.get("capacity") + pvc_capacity = ct.get("capacity") + if (not isinstance(pv_capacity, dict) or pv_capacity.get("storage") != "10Gi" + or not isinstance(pvc_capacity, dict) + or pvc_capacity.get("storage") != "10Gi"): + raise EvidenceError("P0 PVC/PV bound capacity differs from 10Gi fixture") + reclaim = vs.get("persistentVolumeReclaimPolicy") + if reclaim not in ("Delete", "Retain"): + raise EvidenceError("P0 PV has an unrecognized reclaim policy") + return vm["uid"], reclaim + + +def verify_p0_namespace_controls(quota: dict, account: dict) -> None: + """Check admitted P0 namespace guardrails, not runtime admission enforcement.""" + expected_hard = { + "pods": "1", "persistentvolumeclaims": "1", + "requests.cpu": "2", "requests.memory": "4Gi", + "limits.cpu": "4", "limits.memory": "8Gi", + "requests.storage": "20Gi", "requests.ephemeral-storage": "2Gi", + "limits.ephemeral-storage": "4Gi", + } + if not isinstance(quota, dict) or not isinstance(account, dict): + raise EvidenceError("P0 namespace controls are not Kubernetes objects") + qmeta, qspec, qstatus = (quota.get(k) for k in ("metadata", "spec", "status")) + if (not all(isinstance(o, dict) for o in (qmeta, qspec, qstatus)) + or qmeta.get("name") != "t3-p0-quota" + or qmeta.get("namespace") != NAMESPACE + or qmeta.get("deletionTimestamp") + or set(qspec) != {"hard"} + or qspec.get("hard") != expected_hard + or qstatus.get("hard") != expected_hard): + raise EvidenceError("P0 ResourceQuota missing, widened, scoped, or unreconciled") + ameta = account.get("metadata") + if (not isinstance(ameta, dict) + or ameta.get("name") != "t3-p0" + or ameta.get("namespace") != NAMESPACE + or ameta.get("deletionTimestamp") + or account.get("automountServiceAccountToken") is not False + or account.get("secrets") + or account.get("imagePullSecrets")): + raise EvidenceError("P0 ServiceAccount allows credential or identity drift") + + +def collect(context: str, read=kubectl) -> dict[str, str]: + def get(kind: str, name: str) -> dict: + try: + obj = json.loads(read(context, "-n", NAMESPACE, "get", kind, name, "-o", "json")) + except (json.JSONDecodeError, TypeError) as exc: + raise EvidenceError("invalid kubectl JSON") from exc + # Even valid JSON can be a scalar/list. Never let a wrong API shape + # escape as AttributeError while collecting private restart evidence. + if not isinstance(obj, dict): + raise EvidenceError("invalid kubectl object") + return obj + + deployment = get("deployment", "t3-owner") + pod_spec = deployment.get("spec", {}).get("template", {}).get("spec", {}) + generation = deployment.get("metadata", {}).get("generation") + deployment_status = deployment.get("status", {}) + if (deployment.get("spec", {}).get("replicas") != 1 + or not isinstance(generation, int) or generation < 1 + or deployment_status.get("observedGeneration") != generation + or deployment_status.get("updatedReplicas") != 1 + or deployment_status.get("availableReplicas") != 1 + or deployment_status.get("readyReplicas") != 1 + or pod_spec.get("automountServiceAccountToken") is not False): + raise EvidenceError("P0 Deployment rollout not fully observed/Ready or mounts API token") + containers = pod_spec.get("containers", []) + if len(containers) != 1 or "@sha256:" not in containers[0].get("image", ""): + raise EvidenceError("P0 deployment must use exactly one digest-pinned container") + + # A Deployment can report Ready without having been restarted. Select one + # concrete, Ready Pod so a later comparison proves the rollout occurred. + try: + pod_list = json.loads(read(context, "-n", NAMESPACE, "get", "pods", + "-l", "app=t3-owner", "-o", "json")) + except (json.JSONDecodeError, TypeError) as exc: + raise EvidenceError("invalid kubectl pod list JSON") from exc + pods = pod_list.get("items") if isinstance(pod_list, dict) else None + if not isinstance(pods, list) or len(pods) != 1 or not isinstance(pods[0], dict): + raise EvidenceError("P0 must have exactly one selected Pod") + pod = pods[0] + pod_meta = pod.get("metadata", {}) + pod_status = pod.get("status", {}) + container_statuses = pod_status.get("containerStatuses", []) + pod_uid = pod_meta.get("uid") + pod_name = pod_meta.get("name") + if (not pod_uid or not pod_name or pod_meta.get("deletionTimestamp") + or pod_status.get("phase") != "Running" + or len(container_statuses) != 1 + or container_statuses[0].get("name") != "t3" + or container_statuses[0].get("ready") is not True): + raise EvidenceError("P0 selected Pod must be Running, Ready, and not deleting") + + pvc = get("pvc", "t3-p0-data") + # Do not pass an untrusted API volume name to kubectl as a CLI option. + pv_name = pvc.get("spec", {}).get("volumeName") + if (not isinstance(pv_name, str) or not (1 <= len(pv_name) <= 253) + or pv_name[0] not in "abcdefghijklmnopqrstuvwxyz0123456789" + or pv_name[-1] not in "abcdefghijklmnopqrstuvwxyz0123456789" + or any(ch not in "abcdefghijklmnopqrstuvwxyz0123456789.-" + for ch in pv_name)): + raise EvidenceError("P0 PVC has an invalid bound PV name") + try: + pv = json.loads(read(context, "get", "pv", pv_name, "-o", "json")) + except (json.JSONDecodeError, TypeError) as exc: + raise EvidenceError("invalid kubectl PV JSON") from exc + pv_uid, reclaim_policy = verify_persistent_volume(pvc, pv) + service = get("service", "t3-owner") + if service.get("spec", {}).get("type") != "ClusterIP": + raise EvidenceError("P0 Service is not private ClusterIP") + ingress = get("ingress", "t3-owner") + if ingress.get("spec", {}).get("ingressClassName") != "tailscale": + raise EvidenceError("P0 Ingress is not served by Tailscale") + if ingress.get("metadata", {}).get("annotations", {}).get("tailscale.com/funnel") == "true": + raise EvidenceError("Tailscale Funnel must stay disabled") + hostname = verify_private_ingress(ingress) + + verify_p0_wiring(pod, containers[0]["image"], service, ingress) + + # P0 has a dedicated namespace. A second Service or Ingress can expose + # the very same Pod despite a safe-looking named Tailscale Ingress. + try: + all_services = json.loads(read(context, "-n", NAMESPACE, "get", + "services", "-o", "json")) + all_ingresses = json.loads(read(context, "-n", NAMESPACE, "get", + "ingresses", "-o", "json")) + except (json.JSONDecodeError, TypeError) as exc: + raise EvidenceError("invalid kubectl route inventory JSON") from exc + verify_exclusive_routes(service, ingress, all_services, all_ingresses) + + # A matching Service selector is not proof that kube-proxy is directing + # traffic to this Pod. Reconcile the live EndpointSlices and Pod UID/IP. + try: + endpoint_slices = json.loads(read(context, "-n", NAMESPACE, "get", "endpointslices", + "-l", "kubernetes.io/service-name=t3-owner", + "-o", "json")) + except (json.JSONDecodeError, TypeError) as exc: + raise EvidenceError("invalid kubectl EndpointSlice JSON") from exc + verify_service_endpoints(pod, endpoint_slices) + + # NetworkPolicy isolation is additive: reject a missing/extra/widened + # policy instead of checking only the expected policy's name. + try: + network_policies = json.loads(read(context, "-n", NAMESPACE, "get", + "networkpolicies", "-o", "json")) + except (json.JSONDecodeError, TypeError) as exc: + raise EvidenceError("invalid kubectl NetworkPolicy JSON") from exc + verify_private_network_policy(network_policies) + + # A safe Pod alone cannot attest its namespace quota or ServiceAccount. + # Read both controls before fetching the persisted owner identity. + quota = get("resourcequota", "t3-p0-quota") + account = get("serviceaccount", "t3-p0") + verify_p0_namespace_controls(quota, account) + + # Explicitly read the owner's current identity; never log/print it. + raw_id = read(context, "-n", NAMESPACE, "exec", f"pod/{pod_name}", "--", + "cat", "/data/t3/userdata/environment-id").strip() + if not raw_id or len(raw_id) > 512 or "\n" in raw_id or "\r" in raw_id: + raise EvidenceError("missing or malformed persisted T3 environment ID") + return { + "format": "t3-k3s-p0-identity-v3", + "environmentIdHash": hashlib.sha256(raw_id.encode("utf-8")).hexdigest(), + "podUid": pod_uid, + "pvcUid": pvc["metadata"]["uid"], + "pvUid": pv_uid, + "pvReclaimPolicy": reclaim_policy, + "endpointHost": hostname, + "deploymentImage": containers[0]["image"], + } + + +def write_baseline(path: Path, record: dict) -> None: + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + fd = os.open(path, flags, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(record, handle, sort_keys=True) + handle.write("\n") + + +def read_private_baseline(path: Path) -> dict[str, str]: + """Bound and validate untrusted local evidence before comparing restart identity. + + Never follow a symlink or open a FIFO with a blocking read. A v3 baseline + is private local evidence, not a bearer token or authority to pair a peer. + """ + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0) + with os.fdopen(os.open(path, flags), "rb") as handle: + metadata = os.fstat(handle.fileno()) + if (not stat.S_ISREG(metadata.st_mode) + or stat.S_IMODE(metadata.st_mode) & 0o077 + or metadata.st_size > 4096): + raise EvidenceError("baseline must be a private regular file no larger than 4096 bytes") + payload = handle.read(4097) + if len(payload) > 4096: + raise EvidenceError("baseline exceeds 4096 bytes") + try: + old = json.loads(payload) + except (UnicodeError, json.JSONDecodeError, ValueError) as exc: + raise EvidenceError("baseline JSON is invalid") from exc + fields = {"format", "environmentIdHash", "podUid", "pvcUid", + "pvUid", "pvReclaimPolicy", "endpointHost", "deploymentImage"} + if (not isinstance(old, dict) or set(old) != fields + or old.get("format") != "t3-k3s-p0-identity-v3" + or any(not isinstance(old[key], str) or not old[key] + for key in fields - {"format"}) + or old["pvReclaimPolicy"] not in ("Retain", "Delete")): + raise EvidenceError("baseline v3 identity fields are invalid") + return old + + +def check_baseline(path: Path, record: dict) -> None: + old = read_private_baseline(path) + if old.get("format") != record["format"]: + raise EvidenceError("baseline format mismatch") + if not old.get("podUid") or old["podUid"] == record["podUid"]: + raise EvidenceError("P0 Pod UID unchanged; rollout restart not observed") + for field in ("environmentIdHash", "pvcUid", "pvUid", "pvReclaimPolicy", "endpointHost", "deploymentImage"): + if old.get(field) != record[field]: + raise EvidenceError(f"P0 restart changed {field}; do not claim persistence") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--context", required=True, help="explicit disposable K3s context") + modes = parser.add_mutually_exclusive_group(required=True) + modes.add_argument("--write-baseline", type=Path) + modes.add_argument("--check-baseline", type=Path) + args = parser.parse_args() + try: + record = collect(args.context) + if args.write_baseline: + write_baseline(args.write_baseline, record) + print("PASS: P0 route/PVC preflight observed; baseline saved with mode 0600") + else: + check_baseline(args.check_baseline, record) + print("PASS: fresh Ready Pod and route observed; persisted ID, PVC, endpoint and image unchanged") + print("NOT RUN: client pairing, auth, data/thread restoration, provider and policy enforcement") + except (EvidenceError, OSError, ValueError, KeyError) as exc: + parser.exit(1, f"FAIL: {exc}\n") + + +if __name__ == "__main__": + main() diff --git a/examples/k3s-p0/descriptor_probe.py b/examples/k3s-p0/descriptor_probe.py new file mode 100644 index 0000000000000..b7d01491a7104 --- /dev/null +++ b/examples/k3s-p0/descriptor_probe.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""Opt-in, credential-free HTTPS identity check from an authorized tailnet client. + +Run after acceptance.py saved a PRIVATE v3 Kubernetes identity baseline. +This does not read user tokens or grant access, and it never applies to K3s. +""" + +import argparse +from contextlib import closing +import hashlib +import http.client +import json +import os +from pathlib import Path +import re +import ssl +import stat + + +class EvidenceError(RuntimeError): + """Untrusted route or invalid persisted P0 baseline.""" + + +def verify_private_descriptor(hostname: str, expected_id_hash: str, + connect=http.client.HTTPSConnection) -> None: + """Verify the real HTTPS route answers as the persisted T3 environment. + + No auth is sent. TLS trust/hostname validation are mandatory. This proves + a read-only descriptor on one private origin, not pairing, ACLs, or a UI. + """ + if (not isinstance(hostname, str) + or not re.fullmatch(r"t3-k3s-p0\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.ts\.net", hostname) + or not re.fullmatch(r"[a-f0-9]{64}", expected_id_hash)): + raise EvidenceError("private HTTPS target or expected identity is invalid") + try: + with closing(connect(hostname, timeout=10, context=ssl.create_default_context())) as connection: + connection.request("GET", "/.well-known/t3/environment", headers={ + "Accept": "application/json", "Cache-Control": "no-store", + }) + response = connection.getresponse() + if response.status != 200: + raise EvidenceError("private HTTPS descriptor returned non-200 status") + content_type = response.getheader("Content-Type", "").split(";", 1)[0].strip().lower() + if content_type != "application/json": + raise EvidenceError("private HTTPS descriptor returned non-JSON content type") + payload = response.read(65537) + if len(payload) > 65536: + raise EvidenceError("private HTTPS descriptor exceeds response size limit") + except (OSError, TimeoutError, ssl.SSLError, http.client.HTTPException) as exc: + raise EvidenceError("private HTTPS descriptor unreachable or untrusted") from exc + try: + descriptor = json.loads(payload) + except (UnicodeError, json.JSONDecodeError, ValueError) as exc: + raise EvidenceError("private HTTPS descriptor contains invalid JSON") from exc + reported_id = descriptor.get("environmentId") if isinstance(descriptor, dict) else None + if (not isinstance(reported_id, str) or not reported_id + or len(reported_id) > 512 or "\n" in reported_id or "\r" in reported_id + or hashlib.sha256(reported_id.encode("utf-8")).hexdigest() != expected_id_hash): + raise EvidenceError("private HTTPS descriptor environment identity mismatch") + + +def read_baseline(path: Path) -> tuple[str, str]: + """Read the existing 0600 v3 baseline, not live credentials or T3 data.""" + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0) + with os.fdopen(os.open(path, flags), "rb") as handle: + metadata = os.fstat(handle.fileno()) + if (not stat.S_ISREG(metadata.st_mode) + or stat.S_IMODE(metadata.st_mode) & 0o077 + or metadata.st_size > 4096): + raise EvidenceError("baseline must be a private, regular, bounded file") + data = handle.read(4097) + if len(data) > 4096: + raise EvidenceError("baseline exceeds response size limit") + try: + record = json.loads(data) + except (UnicodeError, json.JSONDecodeError, ValueError) as exc: + raise EvidenceError("baseline JSON is invalid") from exc + if not isinstance(record, dict) or record.get("format") != "t3-k3s-p0-identity-v3": + raise EvidenceError("baseline format mismatch; collect a fresh v3 baseline") + if (not record.get("podUid") or not record.get("pvcUid") + or not record.get("pvUid") + or record.get("pvReclaimPolicy") not in ("Retain", "Delete")): + raise EvidenceError("baseline is missing Pod/PVC/PV evidence") + return record.get("endpointHost"), record.get("environmentIdHash") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--baseline", type=Path, required=True, + help="local mode-0600 acceptance.py v3 baseline; keep outside Git") + args = parser.parse_args() + try: + host, identity_hash = read_baseline(args.baseline) + verify_private_descriptor(host, identity_hash) + print("PASS: trusted tailnet HTTPS descriptor matches persisted T3 environment identity") + print("NOT RUN: tailnet ACL negative control, client pairing, thread data, or delegation") + except (EvidenceError, OSError, TypeError) as exc: + parser.exit(1, f"FAIL: {exc}\n") + + +if __name__ == "__main__": + main() diff --git a/examples/k3s-p0/k3s.yaml b/examples/k3s-p0/k3s.yaml new file mode 100644 index 0000000000000..b975ecb9b1ca6 --- /dev/null +++ b/examples/k3s-p0/k3s.yaml @@ -0,0 +1,173 @@ +# P0 only: dedicated, single-tenant, persistent T3 control environment. +# Does not provision workers, grant peer permissions, or touch an existing T3 install. +# Run render.py with a verified OCI digest before applying this template. +apiVersion: v1 +kind: Namespace +metadata: + name: t3-k3s-p0 + labels: + pod-security.kubernetes.io/enforce: restricted + pod-security.kubernetes.io/enforce-version: latest +--- +apiVersion: v1 +kind: ResourceQuota +metadata: + name: t3-p0-quota + namespace: t3-k3s-p0 +spec: + hard: + pods: "1" + persistentvolumeclaims: "1" + requests.cpu: "2" + requests.memory: 4Gi + limits.cpu: "4" + limits.memory: 8Gi + requests.storage: 20Gi + requests.ephemeral-storage: 2Gi + limits.ephemeral-storage: 4Gi +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: t3-p0 + namespace: t3-k3s-p0 +automountServiceAccountToken: false +--- +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: t3-p0-data + namespace: t3-k3s-p0 +spec: + accessModes: [ReadWriteOnce] + resources: + requests: + storage: 10Gi +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: t3-owner + namespace: t3-k3s-p0 +spec: + replicas: 1 + strategy: + type: Recreate + selector: + matchLabels: + app: t3-owner + template: + metadata: + labels: + app: t3-owner + spec: + serviceAccountName: t3-p0 + automountServiceAccountToken: false + hostname: t3-owner + terminationGracePeriodSeconds: 90 + securityContext: + runAsNonRoot: true + runAsUser: 10001 + runAsGroup: 10001 + fsGroup: 10001 + seccompProfile: + type: RuntimeDefault + containers: + - name: t3 + image: REPLACE_WITH_PINNED_IMAGE_DIGEST + imagePullPolicy: IfNotPresent + workingDir: /data + ports: + - name: http + containerPort: 3773 + protocol: TCP + resources: + requests: + cpu: 500m + memory: 1Gi + ephemeral-storage: 512Mi + limits: + cpu: "2" + memory: 4Gi + ephemeral-storage: 2Gi + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: [ALL] + startupProbe: + tcpSocket: + port: http + periodSeconds: 5 + failureThreshold: 60 + readinessProbe: + tcpSocket: + port: http + periodSeconds: 5 + volumeMounts: + - name: data + mountPath: /data + - name: tmp + mountPath: /tmp + volumes: + - name: data + persistentVolumeClaim: + claimName: t3-p0-data + - name: tmp + emptyDir: + sizeLimit: 1Gi +--- +apiVersion: v1 +kind: Service +metadata: + name: t3-owner + namespace: t3-k3s-p0 +spec: + type: ClusterIP + selector: + app: t3-owner + ports: + - name: http + port: 80 + targetPort: http + protocol: TCP +--- +# Tailscale Kubernetes Operator must already exist. TLS terminates at the +# tailnet ingress proxy. No public Ingress, LoadBalancer, NodePort or Funnel. +apiVersion: networking.k8s.io/v1 +kind: Ingress +metadata: + name: t3-owner + namespace: t3-k3s-p0 +spec: + ingressClassName: tailscale + tls: + - hosts: + - t3-k3s-p0 + defaultBackend: + service: + name: t3-owner + port: + name: http +--- +# This allows ingress only from pods in the operator's default "tailscale" +# namespace. It does NOT restrict egress or replace tailnet ACL grants. +# Some CNI configurations do not actually enforce NetworkPolicy; test it. +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: t3-p0-ingress + namespace: t3-k3s-p0 +spec: + podSelector: + matchLabels: + app: t3-owner + policyTypes: [Ingress] + ingress: + - from: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: tailscale + ports: + - protocol: TCP + port: 3773 diff --git a/examples/k3s-p0/preflight.py b/examples/k3s-p0/preflight.py new file mode 100644 index 0000000000000..56c47b078e263 --- /dev/null +++ b/examples/k3s-p0/preflight.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""Read-only suitability gate for a *disposable* K3s/Tailscale P0 test cluster. + +This intentionally cannot prove cluster ownership, CNI NetworkPolicy enforcement, +tailnet ACLs, certificate issuance, or client readiness. Never applies resources. +""" + +import argparse +import json +import re +import subprocess + + +class PreflightError(RuntimeError): + """The selected cluster is not yet eligible for the P0 acceptance test.""" + + +def query(context: str, *args: str, optional_absent: bool = False, + runner=subprocess.run): + if (not context or context != context.strip() or context.startswith("-") + or any(c in context for c in "\r\n\x00")): + raise PreflightError("provide an explicit, valid kubectl context") + command = ["kubectl", "--context", context, "--request-timeout=10s", *args] + try: + completed = runner(command, capture_output=True, text=True, + timeout=20, check=False) + except (OSError, subprocess.TimeoutExpired) as exc: + raise PreflightError("read-only Kubernetes query unavailable") from exc + if completed.returncode: + # Error text may reveal cluster endpoints or credentials: never repeat it. + raise PreflightError("read-only Kubernetes query rejected") + if not completed.stdout.strip() and optional_absent: + return None + try: + value = json.loads(completed.stdout) + except (ValueError, TypeError) as exc: + raise PreflightError("Kubernetes returned invalid JSON") from exc + if not isinstance(value, dict): + raise PreflightError("Kubernetes returned an unexpected object") + return value + + +def require_object(value, label: str) -> dict: + """Reject malformed resource fields without exposing response contents.""" + if not isinstance(value, dict): + raise PreflightError(f"Kubernetes returned malformed {label}") + return value + + +def verify(context: str, runner=subprocess.run) -> None: + """Check only prerequisites safe to prove using GET requests.""" + def read(*args, optional_absent=False): + return query(context, *args, optional_absent=optional_absent, + runner=runner) + + version = read("get", "--raw=/version") + git_version = version.get("gitVersion") + # Other Kubernetes distributions can pass every remaining prerequisite. + # This P0 proof must target a real, stable K3s build, not a generic cluster. + match = (re.fullmatch(r"v(\d+)\.(\d+)\.(\d+)\+k3s[1-9]\d*", git_version) + if isinstance(git_version, str) else None) + if not match or tuple(map(int, match.groups()[:2])) < (1, 28): + raise PreflightError("K3s server must be version 1.28+ with a stable +k3sN build") + + nodes = read("get", "nodes", "-o", "json").get("items") + if not isinstance(nodes, list) or len(nodes) != 1: + raise PreflightError("P0 proof requires exactly one disposable test node") + node = require_object(nodes[0], "Node") + spec = require_object(node.get("spec"), "Node spec") + status = require_object(node.get("status"), "Node status") + if spec.get("unschedulable") not in (None, False): + raise PreflightError("P0 test node is unschedulable") + taints = spec.get("taints", []) + if not isinstance(taints, list) or any(not isinstance(t, dict) for t in taints): + raise PreflightError("Kubernetes returned malformed Node taints") + if any(t.get("effect") in ("NoSchedule", "NoExecute") for t in taints): + raise PreflightError("P0 test node has an untolerated scheduling taint") + conditions = status.get("conditions") + if not isinstance(conditions, list) or any(not isinstance(c, dict) for c in conditions): + raise PreflightError("Kubernetes returned malformed Node conditions") + if not any(c.get("type") == "Ready" and c.get("status") == "True" + for c in conditions): + raise PreflightError("P0 test node is not Ready") + + namespace = read("get", "namespace", "t3-k3s-p0", "--ignore-not-found", + "-o", "json", optional_absent=True) + if namespace is not None: + raise PreflightError("P0 namespace already exists; refuse possible overwrite") + + tail_ns = read("get", "namespace", "tailscale", "-o", "json") + if require_object(tail_ns.get("metadata"), "Tailscale Namespace metadata").get("name") != "tailscale": + raise PreflightError("default Tailscale proxy namespace is not present") + ingress_class = read("get", "ingressclass", "tailscale", "-o", "json") + if require_object(ingress_class.get("spec"), "IngressClass spec").get("controller") != "tailscale.com/ts-ingress": + raise PreflightError("Tailscale IngressClass controller does not match") + + storage = read("get", "storageclasses", "-o", "json").get("items") + if not isinstance(storage, list): + raise PreflightError("StorageClass listing unavailable") + defaults = [] + for item in storage: + meta = require_object(require_object(item, "StorageClass").get("metadata"), + "StorageClass metadata") + annotations = meta.get("annotations", {}) + if not isinstance(annotations, dict): + raise PreflightError("Kubernetes returned malformed StorageClass annotations") + if annotations.get("storageclass.kubernetes.io/is-default-class") == "true": + defaults.append(meta) + if len(defaults) != 1 or not defaults[0].get("name"): + raise PreflightError("P0 requires exactly one named default StorageClass") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--context", required=True, help="explicit disposable K3s context") + args = parser.parse_args() + try: + verify(args.context) + except PreflightError as exc: + parser.exit(1, f"FAIL: {exc}\n") + print("PASS: read-only P0 Kubernetes prerequisite checks") + print("NOT RUN: disposable-cluster ownership, NetworkPolicy dataplane, tailnet TLS/ACL, K3s apply or T3 client") + + +if __name__ == "__main__": + main() diff --git a/examples/k3s-p0/render.py b/examples/k3s-p0/render.py new file mode 100644 index 0000000000000..693e28376a517 --- /dev/null +++ b/examples/k3s-p0/render.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 +"""Render the P0 example only with an immutable OCI image digest. + +No kubectl calls, deployment, credentials, environment modification or +network access. This is not a worker broker. +""" + +import argparse +from pathlib import Path +import os +import re +import tempfile + +PLACEHOLDER = "REPLACE_WITH_PINNED_IMAGE_DIGEST" +# Deliberately conservative; reject whitespace, tags without digests, +# shell metacharacters and YAML injection. A registry/namespace is required. +IMAGE = re.compile(r"(?=.{1,255}$)[a-z0-9][a-z0-9._:/-]*/[a-z0-9][a-z0-9._/-]*@sha256:[a-f0-9]{64}\Z") +HOST_LABEL = re.compile(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\Z") +REPOSITORY_COMPONENT = re.compile(r"[a-z0-9]+(?:(?:[._]|__|-+)[a-z0-9]+)*\Z") + + +def valid_repository(image: str) -> bool: + """Reject tag syntax in paths and invalid registry ports before deployment.""" + name = image.split("@sha256:", 1)[0] + registry, _, repository = name.partition("/") + host, sep, port = registry.partition(":") + return (bool(repository) + and all(HOST_LABEL.fullmatch(label) for label in host.split(".")) + and (not sep or (port.isascii() and port.isdecimal() + and 1 <= int(port) <= 65535)) + and all(REPOSITORY_COMPONENT.fullmatch(segment) + for segment in repository.split("/"))) + + +def render(source: str, image: str) -> str: + if (not IMAGE.fullmatch(image) or ".." in image or "//" in image + or not valid_repository(image)): + raise ValueError("image must be an OCI repository@sha256:<64 lowercase hex> reference") + if source.count(PLACEHOLDER) != 1: + raise ValueError("expected exactly one image placeholder") + return source.replace(PLACEHOLDER, image) + + +def write_new_output(path: Path, content: str) -> None: + """Publish complete private YAML without following or replacing an output path. + + Stage in the same directory, sync the complete file, then hard-link with + exclusive-create semantics. A failed link leaves the existing path intact. + """ + staged = None + try: + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=path.parent, + prefix=f".{path.name}.", suffix=".tmp", delete=False, + ) as handle: + staged = Path(handle.name) + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + os.link(staged, path) + finally: + if staged is not None: + staged.unlink(missing_ok=True) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--image", required=True, help="pushed image ref pinned at sha256 digest") + parser.add_argument("--output", required=True, type=Path, help="rendered YAML path outside source") + args = parser.parse_args() + source = Path(__file__).with_name("k3s.yaml") + if args.output.resolve() == source.resolve(): + parser.error("refusing to overwrite tracked template") + try: + output = render(source.read_text(encoding="utf-8"), args.image) + except ValueError as error: + parser.error(str(error)) + try: + write_new_output(args.output, output) + except OSError as error: + parser.error(f"refusing to replace output or could not create private file: {error.strerror}") + + +if __name__ == "__main__": + main() diff --git a/examples/k3s-p0/test_acceptance.py b/examples/k3s-p0/test_acceptance.py new file mode 100644 index 0000000000000..88f98edcb5113 --- /dev/null +++ b/examples/k3s-p0/test_acceptance.py @@ -0,0 +1,594 @@ +"""Offline fake-kubectl tests for read-only P0 cluster evidence collection.""" + +import importlib.util +import json +import os +from pathlib import Path +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parent +spec = importlib.util.spec_from_file_location("acceptance", ROOT / "acceptance.py") +assert spec and spec.loader +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) + + +def fake_reads(identity="stable-id", ready=True, host="t3-k3s-p0.tailxxx.ts.net", + funnel=False, pod_uid="pod-1", pod_ready=True, pod_count=1, deleting=False, + image=None): + docs = { + "deployment": {"metadata": {"generation": 7}, + "spec": {"replicas": 1, "template": {"spec": { + "automountServiceAccountToken": False, + "containers": [{"name": "t3", "image": image or "ghcr.io/me/t3@sha256:" + "a" * 64}]}}}, + "status": {"readyReplicas": 1 if ready else 0, + "availableReplicas": 1 if ready else 0, + "updatedReplicas": 1 if ready else 0, + "observedGeneration": 7}}, + "pods": {"items": [{ + "metadata": {"name": "t3-owner-test", "uid": pod_uid, + "labels": {"app": "t3-owner"}, + **({"deletionTimestamp": "2026-10-28T00:00:00Z"} if deleting else {})}, + "spec": {"serviceAccountName": "t3-p0", + "hostname": "t3-owner", + "automountServiceAccountToken": False, + "securityContext": { + "runAsNonRoot": True, "runAsUser": 10001, + "runAsGroup": 10001, "fsGroup": 10001, + "seccompProfile": {"type": "RuntimeDefault"}, + }, + "containers": [{"name": "t3", + "image": image or "ghcr.io/me/t3@sha256:" + "a" * 64, + "resources": { + "requests": {"cpu": "500m", "memory": "1Gi", + "ephemeral-storage": "512Mi"}, + "limits": {"cpu": "2", "memory": "4Gi", + "ephemeral-storage": "2Gi"}}, + "securityContext": { + "allowPrivilegeEscalation": False, + "readOnlyRootFilesystem": True, + "capabilities": {"drop": ["ALL"]}, + }, + "volumeMounts": [ + {"name": "data", "mountPath": "/data"}, + {"name": "tmp", "mountPath": "/tmp"}, + ]}], + "volumes": [{"name": "data", "persistentVolumeClaim": { + "claimName": "t3-p0-data"}}, + {"name": "tmp", "emptyDir": {"sizeLimit": "1Gi"}}]}, + "status": {"phase": "Running", "podIP": "10.42.0.9", + "podIPs": [{"ip": "10.42.0.9"}], + "containerStatuses": [ + {"name": "t3", "ready": pod_ready}]}, + } for _ in range(pod_count)]}, + "pvc": {"metadata": {"name": "t3-p0-data", "namespace": "t3-k3s-p0", + "uid": "stable-pvc-uid"}, + "spec": {"volumeName": "pvc-stable-volume", "storageClassName": "local-path", + "accessModes": ["ReadWriteOnce"], "volumeMode": "Filesystem", + "resources": {"requests": {"storage": "10Gi"}}}, + "status": {"phase": "Bound", "capacity": {"storage": "10Gi"}}}, + "pv": {"metadata": {"name": "pvc-stable-volume", "uid": "stable-pv-uid"}, + "spec": {"claimRef": {"name": "t3-p0-data", "namespace": "t3-k3s-p0", + "uid": "stable-pvc-uid"}, + "accessModes": ["ReadWriteOnce"], "volumeMode": "Filesystem", + "storageClassName": "local-path", + "persistentVolumeReclaimPolicy": "Delete", + "capacity": {"storage": "10Gi"}}, + "status": {"phase": "Bound"}}, + "endpointslices": {"items": [{ + "metadata": {"labels": {"kubernetes.io/service-name": "t3-owner"}}, + "ports": [{"name": "http", "port": 3773, "protocol": "TCP"}], + "endpoints": [{ + "conditions": {"ready": True}, + "targetRef": {"kind": "Pod", "name": "t3-owner-test", + "namespace": "t3-k3s-p0", "uid": pod_uid}, + "addresses": ["10.42.0.9"], + }], + }]}, + + "networkpolicies": {"items": [{ + "metadata": {"name": "t3-p0-ingress"}, + "spec": { + "podSelector": {"matchLabels": {"app": "t3-owner"}}, + "policyTypes": ["Ingress"], + "ingress": [{ + "from": [{"namespaceSelector": { + "matchLabels": {"kubernetes.io/metadata.name": "tailscale"}}}], + "ports": [{"protocol": "TCP", "port": 3773}], + }], + }, + }]}, + "resourcequota": { + "metadata": {"name": "t3-p0-quota", "namespace": "t3-k3s-p0"}, + "spec": {"hard": { + "pods": "1", "persistentvolumeclaims": "1", + "requests.cpu": "2", "requests.memory": "4Gi", + "limits.cpu": "4", "limits.memory": "8Gi", + "requests.storage": "20Gi", "requests.ephemeral-storage": "2Gi", + "limits.ephemeral-storage": "4Gi", + }}, + }, + "serviceaccount": { + "metadata": {"name": "t3-p0", "namespace": "t3-k3s-p0"}, + "automountServiceAccountToken": False, + }, + "service": {"spec": {"type": "ClusterIP", "selector": {"app": "t3-owner"}, + "ports": [{"name": "http", "port": 80, "targetPort": "http"}]}}, + "ingress": {"metadata": {"annotations": {"tailscale.com/funnel": "true"} if funnel else {}}, + "spec": {"ingressClassName": "tailscale", + "tls": [{"hosts": ["t3-k3s-p0"]}], + "defaultBackend": {"service": { + "name": "t3-owner", "port": {"name": "http"}}}}, + "status": {"loadBalancer": {"ingress": [{"hostname": host}]}}}, + } + docs["resourcequota"]["status"] = { + "hard": json.loads(json.dumps(docs["resourcequota"]["spec"]["hard"]))} + # GET and namespace LIST must agree on the same, unique routes. + for kind in ("service", "ingress"): + docs[kind]["kind"] = kind.capitalize() + docs[kind].setdefault("metadata", {}).update({ + "name": "t3-owner", "namespace": "t3-k3s-p0", + "uid": kind + "-uid", + }) + docs["services"] = {"kind": "ServiceList", "items": [ + json.loads(json.dumps(docs["service"]))]} + docs["ingresses"] = {"kind": "IngressList", "items": [ + json.loads(json.dumps(docs["ingress"]))]} + calls = [] + def read(context, *args): + calls.append((context, args)) + assert context == "disposable-test" + assert "apply" not in args and "delete" not in args and "rollout" not in args + if args[:2] == ("get", "pv"): + assert args[2:] == ("pvc-stable-volume", "-o", "json") + return json.dumps(docs["pv"]) + if "exec" in args: + assert args[3] == "pod/t3-owner-test" + return identity + "\n" + if args[3] == "pods": + assert args[4:] == ("-l", "app=t3-owner", "-o", "json") + if args[3] == "endpointslices": + assert args[4:] == ("-l", "kubernetes.io/service-name=t3-owner", + "-o", "json") + if args[3] == "networkpolicies": + assert args[4:] == ("-o", "json") + return json.dumps(docs[args[3]]) + return read, calls + + +class P0EvidenceTest(unittest.TestCase): + def test_rejects_non_object_named_api_responses(self): + # A syntactically valid scalar/list is not a Kubernetes resource. + # Each object response must fail closed, not raise AttributeError. + for kind in ("deployment", "pvc", "service", "ingress"): + for bad in ([], None, "not-an-object", 123): + with self.subTest(kind=kind, payload=repr(bad)): + valid, _ = fake_reads() + def malformed(context, *args): + if args[:4] == ("-n", module.NAMESPACE, "get", kind): + return json.dumps(bad) + return valid(context, *args) + with self.assertRaisesRegex(module.EvidenceError, "invalid kubectl object"): + module.collect("disposable-test", malformed) + + def test_rejects_quota_or_service_account_drift(self): + edits = [ + ("resourcequota", lambda d: d["spec"]["hard"].update(pods="9")), + ("resourcequota", lambda d: d["status"]["hard"].update(pods="9")), + ("resourcequota", lambda d: d["spec"].update(scopes=["BestEffort"])), + ("resourcequota", lambda d: d["spec"].update(scopeSelector={})), + ("resourcequota", lambda d: d["metadata"].update(deletionTimestamp="now")), + ("resourcequota", lambda d: d["metadata"].update(namespace="default")), + ("resourcequota", lambda d: d.update(spec=[])), + ("resourcequota", lambda d: d.pop("status")), + ("serviceaccount", lambda d: d.update(automountServiceAccountToken=True)), + ("serviceaccount", lambda d: d.pop("automountServiceAccountToken")), + ("serviceaccount", lambda d: d.update(imagePullSecrets=[{"name": "creds"}])), + ("serviceaccount", lambda d: d.update(secrets=[{"name": "token"}])), + ("serviceaccount", lambda d: d["metadata"].update(deletionTimestamp="now")), + ("serviceaccount", lambda d: d["metadata"].update(name="default")), + ("serviceaccount", lambda d: d.update(metadata=None)), + ] + for kind, edit in edits: + with self.subTest(kind=kind, case=str(edit)): + valid, _ = fake_reads() + def changed(context, *args): + raw = valid(context, *args) + if args[:4] != ("-n", module.NAMESPACE, "get", kind): + return raw + document = json.loads(raw) + edit(document) + return json.dumps(document) + with self.assertRaisesRegex(module.EvidenceError, "P0"): + module.collect("disposable-test", changed) + + def test_same_id_survives_restart_without_leaking_raw_id(self): + reader, calls = fake_reads(pod_uid="pod-before") + before = module.collect("disposable-test", reader) + after = module.collect("disposable-test", fake_reads(pod_uid="pod-after")[0]) + self.assertNotIn("stable-id", json.dumps(before)) + self.assertNotEqual(before["podUid"], after["podUid"]) + self.assertEqual( + {k: v for k, v in before.items() if k != "podUid"}, + {k: v for k, v in after.items() if k != "podUid"}) + self.assertEqual(len(calls), 11) + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "identity.json" + module.write_baseline(path, before) + self.assertEqual(path.stat().st_mode & 0o777, 0o600) + module.check_baseline(path, after) + with self.assertRaises(FileExistsError): + module.write_baseline(path, before) + + def test_change_in_environment_id_rejected(self): + first = module.collect("disposable-test", fake_reads("first", pod_uid="before")[0]) + second = module.collect("disposable-test", fake_reads("second", pod_uid="after")[0]) + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "identity.json" + module.write_baseline(path, first) + with self.assertRaisesRegex(module.EvidenceError, "environmentIdHash"): + module.check_baseline(path, second) + + def test_changed_pvc_or_endpoint_rejected(self): + original = module.collect("disposable-test", fake_reads()[0]) + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "identity.json" + module.write_baseline(path, original) + for change in ({"pvcUid": "new"}, {"pvUid": "new"}, + {"pvReclaimPolicy": "Retain"}, + {"endpointHost": "new.tail.ts.net"}): + with self.assertRaises(module.EvidenceError): + module.check_baseline(path, {**original, **change, "podUid": "after"}) + + def test_rejects_unsafe_or_corrupt_restart_baseline(self): + before = module.collect("disposable-test", fake_reads(pod_uid="before")[0]) + after = module.collect("disposable-test", fake_reads(pod_uid="after")[0]) + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "identity.json" + module.write_baseline(path, before) + module.check_baseline(path, after) + + path.chmod(0o644) + with self.assertRaisesRegex(module.EvidenceError, "private regular"): + module.check_baseline(path, after) + path.chmod(0o600) + + if hasattr(os, "O_NOFOLLOW"): + link = Path(folder) / "alias.json" + link.symlink_to(path) + with self.assertRaises(OSError): + module.check_baseline(link, after) + + for data, message in ( + (b"x" * 4097, "4096"), + (b"{", "JSON is invalid"), + (b"[]", "identity fields"), + (json.dumps({k: v for k, v in before.items() + if k != "pvUid"}).encode(), "identity fields"), + ): + with self.subTest(message=message): + path.write_bytes(data) + with self.assertRaisesRegex(module.EvidenceError, message): + module.check_baseline(path, after) + + if hasattr(os, "mkfifo"): + path.unlink() + os.mkfifo(path, 0o600) + with self.assertRaisesRegex(module.EvidenceError, "private regular"): + module.check_baseline(path, after) + + def test_not_ready_and_funnel_rejected(self): + for kwargs in ({"ready": False}, {"funnel": True}, {"host": "example.org"}, + {"pod_count": 0}, {"pod_count": 2}, {"pod_ready": False}, + {"pod_uid": ""}, {"deleting": True}): + with self.subTest(kwargs=kwargs): + with self.assertRaises(module.EvidenceError): + module.collect("disposable-test", fake_reads(**kwargs)[0]) + + def test_unchanged_pod_uid_cannot_claim_restart(self): + original = module.collect("disposable-test", fake_reads()[0]) + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "identity.json" + module.write_baseline(path, original) + with self.assertRaisesRegex(module.EvidenceError, "rollout restart not observed"): + module.check_baseline(path, original) + + def test_changed_image_rejected_even_with_new_pod(self): + original = module.collect("disposable-test", fake_reads(pod_uid="before")[0]) + changed = module.collect("disposable-test", fake_reads( + pod_uid="after", image="ghcr.io/me/t3@sha256:" + "b" * 64)[0]) + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "identity.json" + module.write_baseline(path, original) + with self.assertRaisesRegex(module.EvidenceError, "deploymentImage"): + module.check_baseline(path, changed) + + def test_no_implicit_kubectl_context(self): + with self.assertRaises(module.EvidenceError): + module.kubectl("-unsafe", "get", "pods") + + def test_rejects_hidden_routes_and_controller_exposure(self): + # A safe named Ingress alone is insufficient: a second Service, + # a second Ingress, or a ClusterIP annotation can expose the owner. + edits = [ + ("services", lambda d: d["items"].append( + {"kind": "Service", "metadata": {"name": "public"}, + "spec": {"type": "LoadBalancer"}})), + ("ingresses", lambda d: d["items"].append( + {"kind": "Ingress", "metadata": {"name": "public"}})), + ("services", lambda d: d["items"].clear()), + ("ingresses", lambda d: d["items"][0]["metadata"].update(uid="new-uid")), + ("ingresses", lambda d: d["items"][0].update(metadata=None)), + ("service", lambda d: d["metadata"].update( + annotations={"tailscale.com/expose": "true"})), + ("ingress", lambda d: d["metadata"].update( + annotations={"tailscale.com/funnel": "TRUE"})), + ("service", lambda d: d.update( + status={"loadBalancer": {"ingress": [{"ip": "203.0.113.4"}]}})), + ("service", lambda d: d["spec"].update( + externalName="public.example.invalid")), + ("service", lambda d: d["spec"].update( + publishNotReadyAddresses=True)), + ] + for index, (kind, edit) in enumerate(edits): + with self.subTest(case=index, kind=kind): + normal, _ = fake_reads() + + def altered(context, *args): + value = normal(context, *args) + if args[3] != kind: + return value + payload = json.loads(value) + edit(payload) + return json.dumps(payload) + + with self.assertRaises(module.EvidenceError): + module.collect("disposable-test", altered) + + def test_rejects_service_and_ingress_misdirection(self): + edits = [ + ("service", lambda doc: doc["spec"].update(selector={"app": "other"})), + ("service", lambda doc: doc["spec"]["ports"][0].update(targetPort="other")), + ("service", lambda doc: doc["spec"].update(externalIPs=["192.0.2.10"])), + ("ingress", lambda doc: doc["spec"]["defaultBackend"]["service"].update(name="other")), + ("ingress", lambda doc: doc["spec"].update(rules=[{"host": "public.example"}])), + ("ingress", lambda doc: doc["spec"]["tls"][0].update(hosts=["other"])), + ] + for kind, edit in edits: + with self.subTest(kind=kind, edit=edits.index((kind, edit))): + base, _ = fake_reads() + def altered(context, *args): + value = base(context, *args) + if args[3] != kind: + return value + doc = json.loads(value) + edit(doc) + return json.dumps(doc) + with self.assertRaisesRegex(module.EvidenceError, "Service|Ingress"): + module.collect("disposable-test", altered) + + def test_rejects_unsafe_admitted_pod_mutations(self): + # Kubernetes admission can mutate a safe Deployment template before + # the Pod starts; Ready status must not be a false security receipt. + edits = [ + lambda s: s.update(serviceAccountName="default"), + lambda s: s.update(automountServiceAccountToken=True), + lambda s: s.update(hostNetwork=True), + lambda s: s.update(hostname="injected-host"), + lambda s: s.pop("hostname"), + lambda s: s.update(subdomain="other"), + lambda s: s.update(setHostnameAsFQDN=True), + lambda s: s.update(hostAliases=[{"ip": "203.0.113.5", + "hostnames": ["other"]}]), + lambda s: s.update(hostPID=True), + lambda s: s.update(hostIPC=True), + lambda s: s.update(shareProcessNamespace=True), + lambda s: s.update(imagePullSecrets=[{"name": "credentials"}]), + lambda s: s.update(initContainers=[{"name": "init"}]), + lambda s: s.update(ephemeralContainers=[{"name": "debug"}]), + lambda s: s["securityContext"].update(runAsUser=0), + lambda s: s["securityContext"].update(seccompProfile={"type": "Unconfined"}), + lambda s: s["containers"][0]["securityContext"].update( + allowPrivilegeEscalation=True), + lambda s: s["containers"][0]["securityContext"].update( + readOnlyRootFilesystem=False), + lambda s: s["containers"][0]["securityContext"].update(runAsUser=0), + lambda s: s["containers"][0]["securityContext"].update(runAsUser=10001), + lambda s: s["containers"][0]["securityContext"].update(runAsGroup=0), + lambda s: s["containers"][0]["securityContext"].update(runAsNonRoot=False), + lambda s: s["containers"][0].update(lifecycle={ + "postStart": {"exec": {"command": ["/bin/sh", "-c", "id"]}} + }), + lambda s: s["containers"][0]["resources"]["requests"].update(cpu="0"), + lambda s: s["containers"][0]["resources"]["limits"].update(memory="8Gi"), + lambda s: s["containers"][0].update(resources={}), + lambda s: s["containers"][0]["securityContext"]["capabilities"].update( + add=["SYS_ADMIN"]), + lambda s: s["containers"][0]["securityContext"].update(capabilities=None), + lambda s: s["containers"][0]["securityContext"].update(seccompProfile=None), + lambda s: s["volumes"][0].update(persistentVolumeClaim=None), + lambda s: s["volumes"][1].update(emptyDir=None), + lambda s: s["containers"][0]["securityContext"].update( + seccompProfile={"type": "Unconfined"}), + lambda s: s["containers"][0].update(command=["/bin/sh"]), + lambda s: s["containers"][0].update(envFrom=[{"secretRef": {"name": "all"}}]), + lambda s: s["containers"][0]["volumeMounts"].append( + {"name": "token", "mountPath": "/var/run/secrets"}), + lambda s: s["volumes"].append({"name": "token", "secret": {"secretName": "all"}}), + lambda s: s["volumes"][1]["emptyDir"].update(sizeLimit="100Gi"), + ] + self.assertTrue(module.collect("disposable-test", fake_reads()[0])) + for index, edit in enumerate(edits): + with self.subTest(index=index): + reader, _ = fake_reads() + def altered(context, *args): + result = reader(context, *args) + if args[3] != "pods": + return result + doc = json.loads(result) + edit(doc["items"][0]["spec"]) + return json.dumps(doc) + with self.assertRaisesRegex(module.EvidenceError, "P0"): + module.collect("disposable-test", altered) + + def test_rejects_wrong_pod_image_pvc_or_token_mount(self): + edits = [ + lambda doc: doc["items"][0]["spec"]["containers"][0].update( + image="ghcr.io/other/t3@sha256:" + "b" * 64), + lambda doc: doc["items"][0]["spec"]["volumes"][0]["persistentVolumeClaim"].update( + claimName="another-pvc"), + lambda doc: doc["items"][0]["spec"].update(automountServiceAccountToken=True), + lambda doc: doc["items"][0]["spec"]["volumes"].append( + {"name": "escape", "hostPath": {"path": "/host"}}), + lambda doc: doc["items"][0]["metadata"].update(labels={"app": "other"}), + ] + for index, edit in enumerate(edits): + with self.subTest(index=index): + base, _ = fake_reads() + def altered(context, *args): + value = base(context, *args) + if args[3] != "pods": + return value + doc = json.loads(value) + edit(doc) + return json.dumps(doc) + with self.assertRaises(module.EvidenceError): + module.collect("disposable-test", altered) + + + def test_rejects_wrong_pv_binding_and_pvc_storage_drift(self): + cases = [ + ("pvc", lambda d: d["metadata"].update(uid="other-claim")), + ("pvc", lambda d: d["spec"].update(volumeName="--kubeconfig=/tmp/oops")), + ("pvc", lambda d: d["spec"].update(accessModes=["ReadWriteMany"])), + ("pvc", lambda d: d["spec"].update(volumeMode="Block")), + ("pvc", lambda d: d["spec"]["resources"]["requests"].update(storage="1Gi")), + ("pvc", lambda d: d["status"]["capacity"].update(storage="1Gi")), + ("pvc", lambda d: d["status"].pop("capacity")), + ("pv", lambda d: d["spec"]["capacity"].update(storage="1Gi")), + ("pv", lambda d: d["spec"].pop("capacity")), + ("pvc", lambda d: d["spec"].update(storageClassName="other-class")), + ("pvc", lambda d: d["spec"].update(dataSource={"name": "other"})), + ("pvc", lambda d: d["metadata"].update(deletionTimestamp="now")), + ("pv", lambda d: d["spec"]["claimRef"].update(uid="different-claim")), + ("pv", lambda d: d["spec"]["claimRef"].update(namespace="unrelated")), + ("pv", lambda d: d["spec"].update(accessModes=["ReadWriteMany"])), + ("pv", lambda d: d["spec"].update(storageClassName="other-class")), + ("pv", lambda d: d["spec"].update(persistentVolumeReclaimPolicy="Recycle")), + ("pv", lambda d: d["metadata"].update(deletionTimestamp="now")), + ("pv", lambda d: d["status"].update(phase="Released")), + ] + self.assertTrue(module.collect("disposable-test", fake_reads()[0])) + for i, (kind, mutate) in enumerate(cases): + with self.subTest(i=i, kind=kind): + read, _ = fake_reads() + def altered(context, *args): + raw = read(context, *args) + target = "pv" if args[:2] == ("get", "pv") else args[3] + if target != kind: + return raw + doc = json.loads(raw) + mutate(doc) + return json.dumps(doc) + with self.assertRaisesRegex(module.EvidenceError, "PV|PVC"): + module.collect("disposable-test", altered) + + def test_rejects_stale_service_endpoints(self): + edits = [ + lambda d: d.update(items=[]), + lambda d: d["items"][0]["metadata"]["labels"].update( + {"kubernetes.io/service-name": "unrelated"}), + lambda d: d["items"][0]["ports"][0].update(port=3774), + lambda d: d["items"][0]["endpoints"][0]["targetRef"].update(uid="old-uid"), + lambda d: d["items"][0]["endpoints"][0]["conditions"].update(ready=False), + lambda d: d["items"][0]["endpoints"][0]["conditions"].update(terminating=True), + lambda d: d["items"][0]["endpoints"][0].update(addresses=["10.42.0.99"]), + lambda d: d["items"][0]["endpoints"].append( + dict(d["items"][0]["endpoints"][0])), + ] + for index, edit in enumerate(edits): + with self.subTest(index=index): + base, _ = fake_reads() + def altered(context, *args): + value = base(context, *args) + if args[3] != "endpointslices": + return value + document = json.loads(value) + edit(document) + return json.dumps(document) + with self.assertRaisesRegex(module.EvidenceError, "EndpointSlice"): + module.collect("disposable-test", altered) + + def test_rejects_unobserved_or_unavailable_rollout(self): + edits = [ + lambda d: d["status"].update(observedGeneration=6), + lambda d: d["status"].update(availableReplicas=0), + lambda d: d["status"].update(updatedReplicas=0), + lambda d: d["metadata"].update(generation=8), + ] + for index, edit in enumerate(edits): + with self.subTest(index=index): + base, _ = fake_reads() + def altered(context, *args): + value = base(context, *args) + if args[3] != "deployment": + return value + document = json.loads(value) + edit(document) + return json.dumps(document) + with self.assertRaisesRegex(module.EvidenceError, "rollout"): + module.collect("disposable-test", altered) + + + def test_rejects_mismatched_or_multiple_tailnet_ingress_routes(self): + edits = [ + lambda d: d["status"]["loadBalancer"].update(ingress=[]), + lambda d: d["status"]["loadBalancer"]["ingress"][0].update( + hostname="unrelated.tailxxx.ts.net"), + lambda d: d["status"]["loadBalancer"]["ingress"][0].update( + hostname="t3-k3s-p0.tailxxx.ts.net.evil.example"), + lambda d: d["status"]["loadBalancer"]["ingress"][0].update( + hostname="t3-k3s-p0..ts.net"), + lambda d: d["status"]["loadBalancer"]["ingress"][0].update( + ip="203.0.113.19"), + lambda d: d["status"]["loadBalancer"]["ingress"].append( + {"hostname": "other.tailxxx.ts.net"}), + ] + for index, edit in enumerate(edits): + with self.subTest(index=index): + reader, _ = fake_reads() + def altered(context, *args): + raw = reader(context, *args) + if args[3] != "ingress": + return raw + document = json.loads(raw) + edit(document) + return json.dumps(document) + with self.assertRaisesRegex(module.EvidenceError, "Ingress"): + module.collect("disposable-test", altered) + + def test_rejects_missing_extra_or_widened_network_policy(self): + edits = [ + lambda d: d.update(items=[]), + lambda d: d["items"].append(dict(d["items"][0])), + lambda d: d["items"][0]["metadata"].update(name="unexpected"), + lambda d: d["items"][0]["spec"]["ingress"][0].update(**{"from": [{}]}), + lambda d: d["items"][0]["spec"]["ingress"][0]["ports"][0].update(port=3774), + lambda d: d["items"][0]["spec"].update(policyTypes=["Ingress", "Egress"]), + ] + for index, edit in enumerate(edits): + with self.subTest(index=index): + reader, _ = fake_reads() + def altered(context, *args): + raw = reader(context, *args) + if args[3] != "networkpolicies": + return raw + document = json.loads(raw) + edit(document) + return json.dumps(document) + with self.assertRaisesRegex(module.EvidenceError, "NetworkPolicy"): + module.collect("disposable-test", altered) + +if __name__ == "__main__": + unittest.main() diff --git a/examples/k3s-p0/test_contract.py b/examples/k3s-p0/test_contract.py new file mode 100644 index 0000000000000..b206c8ac302f0 --- /dev/null +++ b/examples/k3s-p0/test_contract.py @@ -0,0 +1,150 @@ +"""Offline P0 security and shape regression checks (requires PyYAML).""" + +import copy +import importlib.util +from pathlib import Path +import unittest + +import yaml + +ROOT = Path(__file__).resolve().parent +spec = importlib.util.spec_from_file_location("k3s_p0_render", ROOT / "render.py") +assert spec and spec.loader +renderer = importlib.util.module_from_spec(spec) +spec.loader.exec_module(renderer) +DOCS = list(yaml.safe_load_all((ROOT / "k3s.yaml").read_text(encoding="utf-8"))) +OBJECTS = {doc["kind"]: doc for doc in DOCS} + + +def assert_p0_contract(objects: dict) -> None: + expected = {"Namespace", "ResourceQuota", "ServiceAccount", + "PersistentVolumeClaim", "Deployment", "Service", "Ingress", "NetworkPolicy"} + assert set(objects) == expected, "unexpected cluster resources" + ns = objects["Namespace"] + assert ns["metadata"]["name"] == "t3-k3s-p0" + assert ns["metadata"]["labels"]["pod-security.kubernetes.io/enforce"] == "restricted" + for kind, doc in objects.items(): + if kind != "Namespace": + assert doc["metadata"]["namespace"] == "t3-k3s-p0" + quota = objects["ResourceQuota"]["spec"]["hard"] + assert quota["pods"] == "1" + assert quota["persistentvolumeclaims"] == "1" + assert quota["requests.storage"] == "20Gi" + sa = objects["ServiceAccount"] + assert sa["automountServiceAccountToken"] is False + pvc = objects["PersistentVolumeClaim"] + assert pvc["spec"]["accessModes"] == ["ReadWriteOnce"] + dep = objects["Deployment"]["spec"] + assert dep["replicas"] == 1 + assert dep["strategy"]["type"] == "Recreate" + pod = dep["template"]["spec"] + assert pod["serviceAccountName"] == "t3-p0" + assert pod["automountServiceAccountToken"] is False + assert pod["hostname"] == "t3-owner" + assert pod["securityContext"]["runAsNonRoot"] is True + assert pod["securityContext"]["runAsUser"] == 10001 + assert pod["securityContext"]["fsGroup"] == 10001 + assert pod["securityContext"]["seccompProfile"]["type"] == "RuntimeDefault" + assert len(pod["containers"]) == 1 + app = pod["containers"][0] + assert app["image"] == renderer.PLACEHOLDER + assert app["imagePullPolicy"] == "IfNotPresent" + assert app["securityContext"]["allowPrivilegeEscalation"] is False + assert app["securityContext"]["readOnlyRootFilesystem"] is True + assert app["securityContext"]["capabilities"]["drop"] == ["ALL"] + assert app["resources"]["limits"]["cpu"] == "2" + assert app["resources"]["limits"]["memory"] == "4Gi" + assert any(v["mountPath"] == "/data" for v in app["volumeMounts"]) + assert any(v["persistentVolumeClaim"]["claimName"] == "t3-p0-data" for v in pod["volumes"] + if "persistentVolumeClaim" in v) + assert all("hostPath" not in v for v in pod["volumes"]) + assert "privileged" not in app["securityContext"] + service = objects["Service"]["spec"] + assert service["type"] == "ClusterIP" + assert service["ports"][0]["targetPort"] == "http" + ingress = objects["Ingress"] + assert ingress["spec"]["ingressClassName"] == "tailscale" + assert ingress["spec"]["tls"][0]["hosts"] == ["t3-k3s-p0"] + assert "tailscale.com/funnel" not in ingress["metadata"].get("annotations", {}) + assert "rules" not in ingress["spec"] # default backend avoids public Host assumptions + policy = objects["NetworkPolicy"]["spec"] + assert policy["policyTypes"] == ["Ingress"] + assert policy["podSelector"]["matchLabels"] == {"app": "t3-owner"} + assert policy["ingress"][0]["from"][0]["namespaceSelector"]["matchLabels"] == { + "kubernetes.io/metadata.name": "tailscale" + } + assert policy["ingress"][0]["ports"] == [{"protocol": "TCP", "port": 3773}] + + +class P0ContractTest(unittest.TestCase): + def test_safe_declarative_template(self): + assert_p0_contract(OBJECTS) + + def test_rejects_public_ingress_mutation(self): + mutated = copy.deepcopy(OBJECTS) + mutated["Ingress"]["metadata"]["annotations"] = {"tailscale.com/funnel": "true"} + with self.assertRaises(AssertionError): + assert_p0_contract(mutated) + + def test_rejects_unsafe_pod_mutation(self): + mutated = copy.deepcopy(OBJECTS) + mutated["Deployment"]["spec"]["template"]["spec"]["containers"][0][ + "securityContext"]["allowPrivilegeEscalation"] = True + with self.assertRaises(AssertionError): + assert_p0_contract(mutated) + + def test_requires_digest_and_one_placeholder(self): + valid = "ghcr.io/example/t3-p0@sha256:" + "a" * 64 + source = (ROOT / "k3s.yaml").read_text(encoding="utf-8") + result = renderer.render(source, valid) + self.assertEqual(result.count(valid), 1) + self.assertNotIn(renderer.PLACEHOLDER, result) + for invalid in ( + "ghcr.io/example/t3-p0:latest", + "ghcr.io/example/t3-p0@sha256:not-a-digest", + "ghcr.io/example/t3-p0@sha256:" + "A" * 64, + "ghcr.io/example/t3-p0@sha256:" + "a" * 64 + "\nkind: Pod", + "ghcr.io/../t3-p0@sha256:" + "a" * 64, + ): + with self.subTest(invalid=invalid): + with self.assertRaises(ValueError): + renderer.render(source, invalid) + # A digest does not make an invalid repository path or registry port valid. + # Reject images that would otherwise render but fail Kubernetes image pulls. + for invalid_repository in ( + "ghcr.io/example:latest/t3-p0", + "ghcr.io/example/t3-p0:tag/extra", + "registry:bad/example/t3-p0", + "registry:65536/example/t3-p0", + "ghcr.io/foo-_/bar", + "ghcr.io/team.name-/t3", + ): + with self.subTest(invalid_repository=invalid_repository): + with self.assertRaises(ValueError): + renderer.render(source, invalid_repository + "@sha256:" + "a" * 64) + for valid_repository in ( + "ghcr.io/example/t3-p0", + "localhost:5000/example/t3-p0", + "registry.example:443/team/underscored__name", + "registry/team/name--next", + ): + with self.subTest(valid_repository=valid_repository): + image = valid_repository + "@sha256:" + "a" * 64 + self.assertIn(image, renderer.render(source, image)) + with self.assertRaises(ValueError): + renderer.render("no placeholder", valid) + with self.assertRaises(ValueError): + renderer.render(renderer.PLACEHOLDER * 2, valid) + + def test_docker_build_pins_t3_release_and_checks_archive(self): + dockerfile = (ROOT / "Dockerfile").read_text(encoding="utf-8") + self.assertIn("ARG T3_VERSION=0.1.3", dockerfile) + self.assertIn("/SHA256SUMS", dockerfile) + self.assertIn("sha256sum --check", dockerfile) + self.assertIn("USER 10001:10001", dockerfile) + self.assertIn('"start", "--no-browser"', dockerfile) + self.assertNotIn('"serve",', dockerfile) + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/k3s-p0/test_descriptor_probe.py b/examples/k3s-p0/test_descriptor_probe.py new file mode 100644 index 0000000000000..0033d6eee6275 --- /dev/null +++ b/examples/k3s-p0/test_descriptor_probe.py @@ -0,0 +1,194 @@ +"""Offline tests for the P0 read-only, unauthenticated tailnet descriptor probe.""" + +import hashlib +import importlib.util +import json +import os +import subprocess +import sys +from pathlib import Path +import ssl +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parent +spec = importlib.util.spec_from_file_location("descriptor_probe", ROOT / "descriptor_probe.py") +assert spec and spec.loader +probe = importlib.util.module_from_spec(spec) +spec.loader.exec_module(probe) + +HOST = "t3-k3s-p0.sample-tailnet.ts.net" +IDENTITY_HASH = hashlib.sha256(b"stable-id").hexdigest() + + +class FakeResponse: + def __init__(self, status=200, content_type="application/json", body=None): + self.status = status + self.content_type = content_type + self.body = body if body is not None else json.dumps( + {"environmentId": "stable-id"}).encode() + + def getheader(self, name, default=""): + return self.content_type + + def read(self, count): + assert count == 65537 + return self.body[:count] + + +class FakeTlsConnection: + def __init__(self, host, timeout, context, response=None): + assert host == HOST and timeout == 10 + assert context.check_hostname + assert context.verify_mode == ssl.CERT_REQUIRED + self.response = response if response is not None else FakeResponse() + self.calls = [] + self.closed = False + + def close(self): + self.closed = True + + def request(self, *args, **kwargs): + self.calls.append((args, kwargs)) + + def getresponse(self): + return self.response + + +def connection_for(response=None, error=None): + def connect(host, timeout, context): + if error is not None: + raise error + return FakeTlsConnection(host, timeout, context, response) + return connect + + +class P0TlsDescriptorTest(unittest.TestCase): + def test_checked_https_route_and_identity(self): + probe.verify_private_descriptor(HOST, IDENTITY_HASH, connection_for( + FakeResponse(content_type="application/json; charset=utf-8"))) + + def test_get_is_unauthenticated_and_does_not_redirect(self): + connection = FakeTlsConnection(HOST, 10, ssl.create_default_context()) + probe.verify_private_descriptor(HOST, IDENTITY_HASH, lambda *args, **kw: connection) + self.assertEqual(connection.calls, [ + (("GET", "/.well-known/t3/environment"), {"headers": { + "Accept": "application/json", "Cache-Control": "no-store"}}), + ]) + with self.assertRaisesRegex(probe.EvidenceError, "non-200"): + probe.verify_private_descriptor(HOST, IDENTITY_HASH, + connection_for(FakeResponse(status=302))) + + def test_plain_connection_closes_after_success_and_http_error(self): + # stdlib HTTPSConnection has close(), not __enter__/__exit__. + success = FakeTlsConnection(HOST, 10, ssl.create_default_context()) + self.assertFalse(hasattr(success, "__enter__")) + probe.verify_private_descriptor(HOST, IDENTITY_HASH, + lambda *args, **kwargs: success) + self.assertTrue(success.closed) + + failure = FakeTlsConnection(HOST, 10, ssl.create_default_context(), + response=FakeResponse(status=503)) + with self.assertRaisesRegex(probe.EvidenceError, "non-200"): + probe.verify_private_descriptor(HOST, IDENTITY_HASH, + lambda *args, **kwargs: failure) + self.assertTrue(failure.closed) + + def test_rejects_misdirected_hosts_and_invalid_hash(self): + for host in ("http://" + HOST, "t3-k3s-p0..ts.net", + "t3-k3s-p0.a.ts.net/secret", "t3-k3s-p0.a.ts.net.evil", + "t3-k3s-p0.-a.ts.net", "t3-k3s-p0.a-.ts.net", + "evil.tailnet.ts.net", "t3-k3s-p0.☺.ts.net"): + with self.subTest(host=host), self.assertRaises(probe.EvidenceError): + probe.verify_private_descriptor(host, IDENTITY_HASH, connection_for()) + with self.assertRaises(probe.EvidenceError): + probe.verify_private_descriptor(HOST, "bad-hash", connection_for()) + + def test_rejects_bad_http_body_and_mismatched_identity(self): + cases = [ + FakeResponse(status=401), FakeResponse(status=500), + FakeResponse(content_type="text/html"), FakeResponse(body=b"not-json"), + FakeResponse(body=b"[]"), FakeResponse(body=b"{}"), + FakeResponse(body=b'{"environmentId": "wrong"}'), + FakeResponse(body=b"a" * 65537), + ] + for index, response in enumerate(cases): + with self.subTest(index=index), self.assertRaises(probe.EvidenceError): + probe.verify_private_descriptor(HOST, IDENTITY_HASH, connection_for(response)) + + def test_rejects_tls_connect_timeout_and_cert_failure(self): + for error in (OSError("offline"), TimeoutError(), ssl.SSLError("certificate")): + with self.subTest(error=type(error).__name__): + with self.assertRaisesRegex(probe.EvidenceError, "unreachable or untrusted"): + probe.verify_private_descriptor(HOST, IDENTITY_HASH, + connection_for(error=error)) + + def test_requires_private_v3_baseline(self): + with tempfile.TemporaryDirectory() as temp: + path = Path(temp) / "evidence.json" + baseline = {"format": "t3-k3s-p0-identity-v3", "podUid": "pod-1", + "pvcUid": "pvc-1", "pvUid": "pv-1", + "pvReclaimPolicy": "Delete", "endpointHost": HOST, + "environmentIdHash": IDENTITY_HASH} + path.write_text(json.dumps(baseline), encoding="utf-8") + path.chmod(0o600) + self.assertEqual(probe.read_baseline(path), (HOST, IDENTITY_HASH)) + path.chmod(0o644) + with self.assertRaisesRegex(probe.EvidenceError, "private"): + probe.read_baseline(path) + path.chmod(0o600) + baseline["pvUid"] = "" + path.write_text(json.dumps(baseline), encoding="utf-8") + with self.assertRaisesRegex(probe.EvidenceError, "PV"): + probe.read_baseline(path) + baseline["pvUid"] = "pv-1" + baseline["format"] = "t3-k3s-p0-identity-v2" + path.write_text(json.dumps(baseline), encoding="utf-8") + with self.assertRaisesRegex(probe.EvidenceError, "format"): + probe.read_baseline(path) + path.write_bytes(b"x" * 4097) + with self.assertRaisesRegex(probe.EvidenceError, "baseline"): + probe.read_baseline(path) + + def test_rejects_symlink_baseline(self): + with tempfile.TemporaryDirectory() as temp: + original = Path(temp) / "original.json" + alias = Path(temp) / "evidence.json" + original.write_text("{}", encoding="utf-8") + original.chmod(0o600) + alias.symlink_to(original) + with self.assertRaises(OSError): + probe.read_baseline(alias) + + + @unittest.skipUnless(hasattr(os, "mkfifo") and hasattr(os, "O_NONBLOCK"), + "requires POSIX nonblocking FIFO support") + def test_rejects_fifo_baseline_without_blocking(self): + # A mode-0600 FIFO passes the permissions check, but a blocking open + # never reaches fstat unless another process opens the write end. + # Test in a subprocess so a regression fails on a timeout, not a hang. + with tempfile.TemporaryDirectory() as temp: + fifo = Path(temp) / "evidence.fifo" + os.mkfifo(fifo, 0o600) + script = """ +import importlib.util +from pathlib import Path +import sys +spec = importlib.util.spec_from_file_location("probe", sys.argv[1]) +probe = importlib.util.module_from_spec(spec) +spec.loader.exec_module(probe) +try: + probe.read_baseline(Path(sys.argv[2])) +except probe.EvidenceError: + sys.exit(0) +sys.exit(2) +""" + completed = subprocess.run( + [sys.executable, "-c", script, str(ROOT / "descriptor_probe.py"), str(fifo)], + check=False, capture_output=True, text=True, timeout=2, + ) + self.assertEqual(completed.returncode, 0, completed.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/k3s-p0/test_preflight.py b/examples/k3s-p0/test_preflight.py new file mode 100644 index 0000000000000..0f8ef4da29940 --- /dev/null +++ b/examples/k3s-p0/test_preflight.py @@ -0,0 +1,123 @@ +"""Read-only P0 preflight tests; no kubectl process or cluster required.""" + +import copy +import importlib.util +import json +from pathlib import Path +import subprocess +import unittest + +ROOT = Path(__file__).resolve().parent +spec = importlib.util.spec_from_file_location("p0_preflight", ROOT / "preflight.py") +assert spec and spec.loader +preflight = importlib.util.module_from_spec(spec) +spec.loader.exec_module(preflight) + +RESPONSES = { + ("get", "--raw=/version"): {"gitVersion": "v1.37.1+k3s1"}, + ("get", "nodes", "-o", "json"): {"items": [{ + "spec": {"unschedulable": False}, + "status": {"conditions": [{"type": "Ready", "status": "True"}]} + }]}, + ("get", "namespace", "t3-k3s-p0", "--ignore-not-found", "-o", "json"): None, + ("get", "namespace", "tailscale", "-o", "json"): {"metadata": {"name": "tailscale"}}, + ("get", "ingressclass", "tailscale", "-o", "json"): { + "spec": {"controller": "tailscale.com/ts-ingress"} + }, + ("get", "storageclasses", "-o", "json"): {"items": [{"metadata": { + "name": "local-path", "annotations": { + "storageclass.kubernetes.io/is-default-class": "true" + } + }}]}, +} + + +def fake_run(mapping, *, error=None): + def run(command, capture_output, text, timeout, check): + assert command[:4] == ["kubectl", "--context", "isolated-k3s", "--request-timeout=10s"] + assert all(x not in command for x in ("apply", "delete", "patch", "exec", "create", "rollout", "get secrets")) + assert capture_output and text and timeout == 20 and check is False + if error: + raise error + key = tuple(command[4:]) + assert key in mapping, f"unexpected kubectl query: {key}" + value = mapping[key] + return subprocess.CompletedProcess(command, 0, "" if value is None else json.dumps(value), "") + return run + + +class P0PreflightTests(unittest.TestCase): + def test_passes_disposable_ready_single_node(self): + preflight.verify("isolated-k3s", runner=fake_run(RESPONSES)) + + def test_rejects_non_k3s_and_malformed_server_versions(self): + # A generic Kubernetes cluster can meet every other read-only check. + # This P0 proof is explicitly K3s-only; never report it as K3s. + for version in ("v1.37.1", "v1.37.1+k0s1", "v1.37.1+k3s", + "v1.27.9+k3s1", None, 137): + mapping = copy.deepcopy(RESPONSES) + mapping[("get", "--raw=/version")] = {"gitVersion": version} + with self.subTest(version=version), self.assertRaises(preflight.PreflightError): + preflight.verify("isolated-k3s", runner=fake_run(mapping)) + + def test_rejects_missing_or_invalid_context(self): + for context in ("", "-danger", " isolated-k3s", "isolated-k3s\n", "a\x00b"): + with self.subTest(context=repr(context)), self.assertRaises(preflight.PreflightError): + preflight.verify(context, runner=fake_run(RESPONSES)) + + def test_rejects_conflicting_cluster_and_prerequisites(self): + scenarios = [ + (("get", "--raw=/version"), {"gitVersion": "v1.27.9+k3s1"}), + (("get", "nodes", "-o", "json"), {"items": []}), + (("get", "nodes", "-o", "json"), {"items": [{"spec": {}, "status": {}}]}), + (("get", "nodes", "-o", "json"), {"items": [{"spec": {"taints": [{"effect": "NoSchedule"}]}, "status": {"conditions": [{"type": "Ready", "status": "True"}]}}]}), + (("get", "namespace", "t3-k3s-p0", "--ignore-not-found", "-o", "json"), {"metadata": {"name": "t3-k3s-p0"}}), + (("get", "namespace", "tailscale", "-o", "json"), {"metadata": {"name": "another"}}), + (("get", "ingressclass", "tailscale", "-o", "json"), {"spec": {"controller": "nginx"}}), + (("get", "storageclasses", "-o", "json"), {"items": []}), + (("get", "storageclasses", "-o", "json"), {"items": [{"metadata": {"name": "a", "annotations": {"storageclass.kubernetes.io/is-default-class": "true"}}}, {"metadata": {"name": "b", "annotations": {"storageclass.kubernetes.io/is-default-class": "true"}}}]}), + ] + for key, value in scenarios: + values = copy.deepcopy(RESPONSES) + values[key] = value + with self.subTest(key=key, value=value), self.assertRaises(preflight.PreflightError): + preflight.verify("isolated-k3s", runner=fake_run(values)) + + def test_malformed_api_objects_fail_with_safe_preflight_error(self): + # JSON can be syntactically valid yet structurally incompatible with + # Kubernetes resources; never print a Python traceback or API contents. + cases = [ + (("get", "nodes", "-o", "json"), {"items": [{"spec": None, "status": {"conditions": [{"type": "Ready", "status": "True"}]}}]}), + (("get", "nodes", "-o", "json"), {"items": [{"spec": {"taints": None}, "status": {"conditions": [{"type": "Ready", "status": "True"}]}}]}), + (("get", "nodes", "-o", "json"), {"items": [{"spec": {"taints": ["bad"]}, "status": {"conditions": [{"type": "Ready", "status": "True"}]}}]}), + (("get", "nodes", "-o", "json"), {"items": [{"spec": {}, "status": {"conditions": ["bad"]}}]}), + (("get", "nodes", "-o", "json"), {"items": [{"spec": {}, "status": {"conditions": None}}]}), + (("get", "namespace", "tailscale", "-o", "json"), {"metadata": None}), + (("get", "ingressclass", "tailscale", "-o", "json"), {"spec": "unexpected"}), + (("get", "storageclasses", "-o", "json"), {"items": [None]}), + (("get", "storageclasses", "-o", "json"), {"items": [{"metadata": {"annotations": None}}]}), + ] + for key, response in cases: + values = copy.deepcopy(RESPONSES) + values[key] = response + with self.subTest(response=response), self.assertRaises(preflight.PreflightError): + preflight.verify("isolated-k3s", runner=fake_run(values)) + + def test_fails_closed_on_query_error_and_malformed_json(self): + with self.assertRaisesRegex(preflight.PreflightError, "unavailable"): + preflight.verify("isolated-k3s", runner=fake_run(RESPONSES, error=OSError("denied"))) + + def bad_json(command, capture_output, text, timeout, check): + return subprocess.CompletedProcess(command, 0, "not-json", "") + with self.assertRaisesRegex(preflight.PreflightError, "invalid JSON"): + preflight.verify("isolated-k3s", runner=bad_json) + + def error_json(command, capture_output, text, timeout, check): + return subprocess.CompletedProcess(command, 1, "", "private credential info") + with self.assertRaisesRegex(preflight.PreflightError, "query rejected") as exc: + preflight.verify("isolated-k3s", runner=error_json) + self.assertNotIn("credential", str(exc.exception)) + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/k3s-p0/test_render_output.py b/examples/k3s-p0/test_render_output.py new file mode 100644 index 0000000000000..72e7a48cf11fc --- /dev/null +++ b/examples/k3s-p0/test_render_output.py @@ -0,0 +1,48 @@ +import importlib.util +from pathlib import Path +import stat +import tempfile +import unittest +from unittest import mock + +ROOT = Path(__file__).resolve().parent +spec = importlib.util.spec_from_file_location("k3s_p0_render", ROOT / "render.py") +assert spec and spec.loader +render = importlib.util.module_from_spec(spec) +spec.loader.exec_module(render) + +class TestSafeRenderOutput(unittest.TestCase): + def test_output_is_complete_private_and_exclusive(self): + with tempfile.TemporaryDirectory() as root: + out = Path(root) / 'out.yaml' + render.write_new_output(out, 'hello\n') + self.assertEqual(out.read_text(), 'hello\n') + self.assertEqual(stat.S_IMODE(out.stat().st_mode), 0o600) + with self.assertRaises(FileExistsError): + render.write_new_output(out, 'unsafe replacement') + self.assertEqual(out.read_text(), 'hello\n') + self.assertEqual(set(Path(root).iterdir()), {out}) + + def test_symlink_does_not_clobber_target(self): + with tempfile.TemporaryDirectory() as root: + target = Path(root) / 'precious.txt' + target.write_text('user data') + link = Path(root) / 'output.yaml' + link.symlink_to(target) + with self.assertRaises(FileExistsError): + render.write_new_output(link, 'malicious data') + self.assertEqual(target.read_text(), 'user data') + self.assertTrue(link.is_symlink()) + self.assertFalse(any(Path(root).glob('.*.tmp'))) + + def test_publish_failure_cleans_staging_without_output(self): + with tempfile.TemporaryDirectory() as root: + out = Path(root) / 'output.yaml' + with mock.patch.object(render.os, 'link', side_effect=OSError('link unavailable')): + with self.assertRaises(OSError): + render.write_new_output(out, 'incomplete') + self.assertFalse(out.exists()) + self.assertEqual(list(Path(root).iterdir()), []) + +if __name__ == '__main__': + unittest.main()