Govern an API through Istio

"""Build and run the complete Istio notes application in its own cluster."""
import json, os, shutil, subprocess, sys, tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent

def main():
    for program in ("docker", "git", "uv"):
        if not shutil.which(program):
            raise SystemExit(f"Install {program} before running this application.")
    subprocess.run(["docker", "info"], check=True, stdout=subprocess.DEVNULL)
    source = json.loads((ROOT / "SOURCE.json").read_text())
    dependencies = ROOT / ".deps"
    dependencies.mkdir(exist_ok=True)
    checkout = dependencies / "chio"
    if not checkout.exists():
        with tempfile.TemporaryDirectory(prefix="source-", dir=dependencies) as temporary:
            subprocess.run(["git", "init", "-q", temporary], check=True)
            subprocess.run(["git", "-C", temporary, "fetch", "--depth", "1",
                            source["repository"], source["revision"]], check=True)
            subprocess.run(["git", "-C", temporary, "checkout", "--detach", "FETCH_HEAD"], check=True)
            os.rename(temporary, checkout)
    actual = subprocess.check_output(["git", "-C", str(checkout), "rev-parse", "HEAD"], text=True).strip()
    if actual != source["revision"]:
        raise SystemExit("The existing build checkout differs from SOURCE.json. Inspect .deps/chio.")
    subprocess.run([sys.executable, str(ROOT / "install-kubernetes-tools.py")], check=True)
    subprocess.run(["uv", "run", "--locked", str(ROOT / "kubernetes.py"), "--source", str(checkout)],
                   cwd=ROOT, check=True)

if __name__ == "__main__":
    main()

"""A local notes API, with SQLite persistence and no Chio dependency."""
import argparse
import json
import sqlite3
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit


class Notes(BaseHTTPRequestHandler):
    def setup(self):
        super().setup()
        self.connection.settimeout(10)

    def respond(self, status, payload):
        body = json.dumps(payload, separators=(",", ":")).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        path = urlsplit(self.path).path
        with sqlite3.connect(self.server.database) as database:
            if path == "/healthz":
                return self.respond(200, {"status": "ready"})
            if path == "/notes":
                rows = database.execute("SELECT id, text FROM notes ORDER BY id LIMIT 100").fetchall()
                return self.respond(200, {"notes": [{"id": row[0], "text": row[1]} for row in rows]})
            if path.startswith("/notes/") and path[7:].isdigit():
                row = database.execute("SELECT id, text FROM notes WHERE id = ?", (int(path[7:]),)).fetchone()
                if row:
                    return self.respond(200, {"id": row[0], "text": row[1]})
        self.respond(404, {"error": "note_not_found"})

    def do_POST(self):
        if urlsplit(self.path).path != "/notes":
            return self.respond(404, {"error": "route_not_found"})
        if self.headers.get("Transfer-Encoding"):
            return self.respond(400, {"error": "content_length_required"})
        try:
            length = int(self.headers.get("Content-Length", "0"))
            if not 0 < length <= 8192:
                return self.respond(413, {"error": "body_must_be_1_to_8192_bytes"})
            raw = self.rfile.read(length)
            if len(raw) != length:
                return self.respond(400, {"error": "incomplete_body"})
            payload = json.loads(raw)
            text = payload.get("text") if isinstance(payload, dict) else None
            if not isinstance(text, str) or not 1 <= len(text.strip()) <= 1000:
                return self.respond(400, {"error": "text_must_be_1_to_1000_characters"})
        except (ValueError, UnicodeError):
            return self.respond(400, {"error": "invalid_json_body"})
        with sqlite3.connect(self.server.database) as database:
            cursor = database.execute("INSERT INTO notes(text) VALUES (?)", (text,))
            note_id = cursor.lastrowid
        self.respond(201, {"id": note_id, "text": text})

    def do_DELETE(self):
        path = urlsplit(self.path).path
        if not path.startswith("/notes/") or not path[7:].isdigit():
            return self.respond(404, {"error": "route_not_found"})
        with sqlite3.connect(self.server.database) as database:
            cursor = database.execute("DELETE FROM notes WHERE id = ?", (int(path[7:]),))
        self.respond(200 if cursor.rowcount else 404, {"deleted": cursor.rowcount == 1})


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--port", type=int, default=8087)
    parser.add_argument("--host", default="127.0.0.1", help="Bind address; Kubernetes uses 0.0.0.0")
    parser.add_argument("--data", type=Path, default=Path(".notes"))
    args = parser.parse_args()
    args.data.mkdir(mode=0o700, parents=True, exist_ok=True)
    database = args.data / "notes.db"
    with sqlite3.connect(database) as connection:
        connection.execute("CREATE TABLE IF NOT EXISTS notes(id INTEGER PRIMARY KEY, text TEXT NOT NULL)")
    server = ThreadingHTTPServer((args.host, args.port), Notes)
    server.database = database
    print(f"Notes API: http://{args.host}:{args.port}; data: {database}", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()

FROM rust:1.94.1-bookworm AS build
RUN apt-get update && apt-get install -y --no-install-recommends clang cmake pkg-config libssl-dev protobuf-compiler && rm -rf /var/lib/apt/lists/*
ENV CARGO_BUILD_JOBS=2
WORKDIR /source
COPY . .
RUN cargo build --locked --profile docker-release -p chio-envoy-ext-authz --features runtime --bin chio-envoy-ext-authz
FROM build AS authority-build
RUN cargo build --locked --profile docker-release -p chio-cli --bin chio
FROM python:3.12.13-slim-bookworm@sha256:4766d8b510c428e595d74b9cc5bbb2fae8e26316fffb4adc89908d79aacd58a2 AS authority
COPY --from=authority-build /source/target/docker-release/chio /usr/local/bin/chio
USER 65532:65532
EXPOSE 9097
ENTRYPOINT ["chio"]
FROM debian:bookworm-slim AS runtime
RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates libssl3 && rm -rf /var/lib/apt/lists/*
COPY --from=build /source/target/docker-release/chio-envoy-ext-authz /usr/local/bin/chio-envoy-ext-authz
USER 65532:65532
EXPOSE 9091 9092
ENTRYPOINT ["chio-envoy-ext-authz"]

# The same persistent notes API used by the local Envoy application.
# Its real paths (/notes and /healthz) require no VirtualService rewrite.
apiVersion: v1
kind: Namespace
metadata:
  name: agent-tools
  labels:
    istio-injection: enabled
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: demo-tool-source
  namespace: agent-tools
data:
  notes.py: |
    """A local notes API, with SQLite persistence and no Chio dependency."""
    import argparse
    import json
    import sqlite3
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
    from pathlib import Path
    from urllib.parse import urlsplit


    class Notes(BaseHTTPRequestHandler):
        def setup(self):
            super().setup()
            self.connection.settimeout(10)

        def respond(self, status, payload):
            body = json.dumps(payload, separators=(",", ":")).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def do_GET(self):
            path = urlsplit(self.path).path
            with sqlite3.connect(self.server.database) as database:
                if path == "/healthz":
                    return self.respond(200, {"status": "ready"})
                if path == "/notes":
                    rows = database.execute("SELECT id, text FROM notes ORDER BY id LIMIT 100").fetchall()
                    return self.respond(200, {"notes": [{"id": row[0], "text": row[1]} for row in rows]})
                if path.startswith("/notes/") and path[7:].isdigit():
                    row = database.execute("SELECT id, text FROM notes WHERE id = ?", (int(path[7:]),)).fetchone()
                    if row:
                        return self.respond(200, {"id": row[0], "text": row[1]})
            self.respond(404, {"error": "note_not_found"})

        def do_POST(self):
            if urlsplit(self.path).path != "/notes":
                return self.respond(404, {"error": "route_not_found"})
            if self.headers.get("Transfer-Encoding"):
                return self.respond(400, {"error": "content_length_required"})
            try:
                length = int(self.headers.get("Content-Length", "0"))
                if not 0 < length <= 8192:
                    return self.respond(413, {"error": "body_must_be_1_to_8192_bytes"})
                raw = self.rfile.read(length)
                if len(raw) != length:
                    return self.respond(400, {"error": "incomplete_body"})
                payload = json.loads(raw)
                text = payload.get("text") if isinstance(payload, dict) else None
                if not isinstance(text, str) or not 1 <= len(text.strip()) <= 1000:
                    return self.respond(400, {"error": "text_must_be_1_to_1000_characters"})
            except (ValueError, UnicodeError):
                return self.respond(400, {"error": "invalid_json_body"})
            with sqlite3.connect(self.server.database) as database:
                cursor = database.execute("INSERT INTO notes(text) VALUES (?)", (text,))
                note_id = cursor.lastrowid
            self.respond(201, {"id": note_id, "text": text})

        def do_DELETE(self):
            path = urlsplit(self.path).path
            if not path.startswith("/notes/") or not path[7:].isdigit():
                return self.respond(404, {"error": "route_not_found"})
            with sqlite3.connect(self.server.database) as database:
                cursor = database.execute("DELETE FROM notes WHERE id = ?", (int(path[7:]),))
            self.respond(200 if cursor.rowcount else 404, {"deleted": cursor.rowcount == 1})


    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description=__doc__)
        parser.add_argument("--port", type=int, default=8087)
        parser.add_argument("--host", default="127.0.0.1", help="Bind address; Kubernetes uses 0.0.0.0")
        parser.add_argument("--data", type=Path, default=Path(".notes"))
        args = parser.parse_args()
        args.data.mkdir(mode=0o700, parents=True, exist_ok=True)
        database = args.data / "notes.db"
        with sqlite3.connect(database) as connection:
            connection.execute("CREATE TABLE IF NOT EXISTS notes(id INTEGER PRIMARY KEY, text TEXT NOT NULL)")
        server = ThreadingHTTPServer((args.host, args.port), Notes)
        server.database = database
        print(f"Notes API: http://{args.host}:{args.port}; data: {database}", flush=True)
        try:
            server.serve_forever()
        except KeyboardInterrupt:
            pass
        finally:
            server.server_close()
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: demo-tool-notes
  namespace: agent-tools
spec:
  accessModes: [ReadWriteOnce]
  resources:
    requests:
      storage: 64Mi
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: demo-tool
  namespace: agent-tools
spec:
  replicas: 1
  strategy:
    type: Recreate
  selector:
    matchLabels:
      app.kubernetes.io/name: demo-tool
  template:
    metadata:
      labels:
        app.kubernetes.io/name: demo-tool
        chio.world/secured: "true"
    spec:
      automountServiceAccountToken: false
      securityContext:
        runAsNonRoot: true
        runAsUser: 65532
        runAsGroup: 65532
        fsGroup: 65532
        seccompProfile:
          type: RuntimeDefault
      containers:
        - name: notes
          image: python:3.12.13-slim-bookworm@sha256:4766d8b510c428e595d74b9cc5bbb2fae8e26316fffb4adc89908d79aacd58a2
          command: [python3, /app/notes.py, --host, 0.0.0.0, --data, /data]
          ports:
            - name: http
              containerPort: 8087
          readinessProbe:
            httpGet:
              path: /healthz
              port: http
          livenessProbe:
            httpGet:
              path: /healthz
              port: http
          resources:
            requests: {cpu: 50m, memory: 64Mi}
            limits: {cpu: 200m, memory: 128Mi}
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop: [ALL]
          volumeMounts:
            - {name: source, mountPath: /app, readOnly: true}
            - {name: data, mountPath: /data}
      volumes:
        - name: source
          configMap:
            name: demo-tool-source
        - name: data
          persistentVolumeClaim:
            claimName: demo-tool-notes
---
apiVersion: v1
kind: Service
metadata:
  name: demo-tool
  namespace: agent-tools
spec:
  selector:
    app.kubernetes.io/name: demo-tool
  ports:
    - {name: http, port: 80, targetPort: http, appProtocol: http}

# /// script
# requires-python = ">=3.11"
# dependencies = ["chio-sdk"]
# [tool.uv.sources]
# chio-sdk = { path = "sdk/chio-py" }
# ///
"""Qualify the supplied Istio deployment in an owned, temporary kind cluster.

Requires Docker, kind, kubectl and istioctl. Keeps evidence under .kubernetes/;
removes only the cluster and registry created by this invocation. Every kubectl
and Istio command uses this run's separate kubeconfig.
"""

import argparse
import hashlib
import json
import os
import shutil
import subprocess
import time
import urllib.error
import urllib.request
import uuid
from pathlib import Path

from chio.invariants import (
    canonicalize_json,
    sha256_hex_utf8,
    verify_http_receipt_with_trusted_signers,
)
from deploy import resources

ROOT = Path(__file__).resolve().parent
REPOSITORY = ROOT.parents[1]
PYTHON_IMAGE = "python:3.12.13-slim-bookworm@sha256:4766d8b510c428e595d74b9cc5bbb2fae8e26316fffb4adc89908d79aacd58a2"
NODE_IMAGE = (
    "kindest/node:v1.36.4@sha256:099e049362a1526b2db71494e1947aae99bd16290d7c895f2b7ea312e3cbfaed"
)
REGISTRY_IMAGE = (
    "registry:2.8.3@sha256:a3d8aaa63ed8681a604f1dea0aa03f100d5895b6a58ace528858a7b332415373"
)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--source", type=Path, help="Chio source checkout used as the Docker build context"
    )
    args = parser.parse_args()
    global REPOSITORY
    REPOSITORY = (args.source or REPOSITORY).resolve()
    if not (REPOSITORY / "crates/protocol/chio-envoy-ext-authz/Cargo.toml").is_file():
        parser.error("--source must name a Chio source checkout containing the Envoy adapter")
    installed = ROOT / ".tools"
    os.environ["PATH"] = str(installed) + os.pathsep + os.environ.get("PATH", "")
    for program in ("docker", "kind", "kubectl", "istioctl"):
        if not shutil.which(program):
            raise SystemExit(f"Install {program} using the prerequisites in README.md")
    os.umask(0o077)
    name = "chio-notes-" + uuid.uuid4().hex[:12]
    registry = name + "-registry"
    state = ROOT / ".kubernetes" / name
    state.mkdir(parents=True)
    kubeconfig = str(state / "kubeconfig")
    children = []
    forward_log = None
    cluster_created = registry_created = False
    observations = []
    with (state / "commands.log").open("w") as log:

        def run(command, *, data=None, capture=False, check=True, timeout=900):
            result = subprocess.run(
                command,
                cwd=REPOSITORY,
                input=data,
                text=True,
                stdout=subprocess.PIPE if capture else log,
                stderr=log,
                timeout=timeout,
                check=check,
            )
            return result.stdout.strip() if capture else result.returncode

        def kube(*arguments, **options):
            return run(["kubectl", "--kubeconfig", kubeconfig, *arguments], **options)

        def apply(value):
            kube("apply", "-f", "-", data=json.dumps(value))

        def rollout(namespace, deployment):
            kube("-n", namespace, "rollout", "status", "deployment/" + deployment, "--timeout=300s")

        def authority(path, body=None):
            request = urllib.request.Request(
                "http://127.0.0.1:" + local_port + path,
                data=None if body is None else json.dumps(body).encode(),
                headers={"Content-Type": "application/json"},
            )
            with urllib.request.urlopen(request, timeout=10) as response:
                return json.load(response)

        def client(path="/notes", body=None, grant=None, expected=200):
            # The request originates in a separate pod and enters the notes
            # pod through its actual Istio inbound proxy. Port-forwarding the
            # notes container would bypass that interception boundary.
            request = {"path": path, "body": body, "grant": grant}
            program = """import json,sys,urllib.request,urllib.error
p=json.load(sys.stdin);headers={'Content-Type':'application/json'}
if p['grant'] is not None: headers['X-Chio-Capability']=json.dumps(p['grant'],separators=(',',':'))
r=urllib.request.Request('http://demo-tool.agent-tools.svc.cluster.local'+p['path'],data=None if p['body'] is None else json.dumps(p['body']).encode(),headers=headers)
try: response=urllib.request.urlopen(r,timeout=15)
except urllib.error.HTTPError as error: response=error
with response:
 raw=response.read()
 try: body=json.loads(raw)
 except ValueError: body=raw.decode()
 print(json.dumps({'status':response.status,'receipt_id':response.headers.get('x-chio-receipt-id'),'body':body}))
"""
            response = json.loads(
                kube(
                    "-n",
                    "chio-system",
                    "exec",
                    "-i",
                    "client",
                    "--",
                    "python3",
                    "-c",
                    program,
                    data=json.dumps(request),
                    capture=True,
                )
            )
            if response["status"] != expected:
                raise AssertionError(f"Expected {expected}: {response}")
            if path != "/healthz" and expected in (200, 201, 403) and not response["receipt_id"]:
                raise AssertionError("Protected request is missing its admission receipt ID")
            observations.append(
                {
                    "path": path,
                    "request_body": body,
                    "capability_id": None if grant is None else grant["id"],
                    **response,
                }
            )
            return response["body"]

        try:
            print("Building the authorization and authority images", flush=True)
            for target in ("runtime", "authority"):
                run(
                    [
                        "docker",
                        "build",
                        "--target",
                        target,
                        "-f",
                        str(ROOT / "Dockerfile"),
                        "-t",
                        name + ":" + target,
                        ".",
                    ],
                    timeout=7200,
                )
            run(
                ["docker", "run", "-d", "--name", registry, "-p", "127.0.0.1::5000", REGISTRY_IMAGE]
            )
            registry_created = True
            port_mapping = json.loads(
                run(
                    ["docker", "inspect", registry, "--format", "{{json .NetworkSettings.Ports}}"],
                    capture=True,
                )
            )
            registry_port = port_mapping["5000/tcp"][0]["HostPort"]
            deadline = time.monotonic() + 30
            while True:
                try:
                    with urllib.request.urlopen(
                        f"http://127.0.0.1:{registry_port}/v2/", timeout=2
                    ) as response:
                        if response.status == 200:
                            break
                except (OSError, urllib.error.URLError) as error:
                    if time.monotonic() >= deadline:
                        raise RuntimeError(
                            "The owned image registry did not become ready"
                        ) from error
                    time.sleep(0.2)
            image_references = {}
            for target in ("runtime", "authority"):
                tag = f"localhost:{registry_port}/chio-{target}:{name}"
                run(["docker", "tag", name + ":" + target, tag])
                run(["docker", "push", tag])
                refs = json.loads(
                    run(
                        ["docker", "image", "inspect", tag, "--format", "{{json .RepoDigests}}"],
                        capture=True,
                    )
                )
                image_references[target] = next(
                    ref
                    for ref in refs
                    if ref.startswith(f"localhost:{registry_port}/chio-{target}@sha256:")
                )
            (state / "images.json").write_text(json.dumps(image_references, indent=2) + "\n")
            print("Creating an isolated kind cluster and installing Istio", flush=True)
            # Register ownership before creation so failed bootstrap is cleaned.
            cluster_created = True
            run(
                [
                    "kind",
                    "create",
                    "cluster",
                    "--name",
                    name,
                    "--kubeconfig",
                    kubeconfig,
                    "--image",
                    NODE_IMAGE,
                    "--wait",
                    "180s",
                ]
            )
            run(["docker", "network", "connect", "kind", registry])
            nodes = run(["kind", "get", "nodes", "--name", name], capture=True).splitlines()
            for node in nodes:
                registry_directory = f"/etc/containerd/certs.d/localhost:{registry_port}"
                run(["docker", "exec", node, "mkdir", "-p", registry_directory])
                run(
                    [
                        "docker",
                        "exec",
                        "-i",
                        node,
                        "cp",
                        "/dev/stdin",
                        registry_directory + "/hosts.toml",
                    ],
                    data=f'[host."http://{registry}:5000"]\n  capabilities = ["pull", "resolve"]\n',
                )
            run(
                [
                    "istioctl",
                    "--kubeconfig",
                    kubeconfig,
                    "install",
                    "-y",
                    "--set",
                    "profile=minimal",
                    "-f",
                    str(ROOT / "01-meshconfig-patch.yaml"),
                ]
            )
            apply(
                {
                    "apiVersion": "v1",
                    "kind": "Namespace",
                    "metadata": {"name": "chio-system", "labels": {"istio-injection": "disabled"}},
                }
            )
            kube("apply", "-f", str(ROOT / "03-demo-workload.yaml"))
            rollout("agent-tools", "demo-tool")
            apply(
                {
                    "apiVersion": "v1",
                    "kind": "ConfigMap",
                    "metadata": {"name": "notes-api-contract", "namespace": "chio-system"},
                    "data": {"openapi.yaml": (ROOT / "local/openapi.yaml").read_text()},
                }
            )
            apply(
                {
                    "apiVersion": "v1",
                    "kind": "PersistentVolumeClaim",
                    "metadata": {"name": "notes-authority", "namespace": "chio-system"},
                    "spec": {
                        "accessModes": ["ReadWriteOnce"],
                        "resources": {"requests": {"storage": "128Mi"}},
                    },
                }
            )
            apply(
                {
                    "apiVersion": "apps/v1",
                    "kind": "Deployment",
                    "metadata": {"name": "notes-authority", "namespace": "chio-system"},
                    "spec": {
                        "replicas": 1,
                        "strategy": {"type": "Recreate"},
                        "selector": {"matchLabels": {"app": "notes-authority"}},
                        "template": {
                            "metadata": {"labels": {"app": "notes-authority"}},
                            "spec": {
                                "automountServiceAccountToken": False,
                                "securityContext": {
                                    "runAsNonRoot": True,
                                    "runAsUser": 65532,
                                    "runAsGroup": 65532,
                                    "fsGroup": 65532,
                                },
                                "containers": [
                                    {
                                        "name": "authority",
                                        "workingDir": "/data",
                                        "image": image_references["authority"],
                                        "args": [
                                            "--authority-seed-file",
                                            "/data/authority.hex",
                                            "api",
                                            "protect",
                                            "--upstream",
                                            "http://demo-tool.agent-tools.svc.cluster.local",
                                            "--spec",
                                            "/contract/openapi.yaml",
                                            "--listen",
                                            "0.0.0.0:9097",
                                            "--receipt-store",
                                            "/data/receipts.db",
                                        ],
                                        "ports": [{"name": "http", "containerPort": 9097}],
                                        "readinessProbe": {
                                            "httpGet": {"path": "/chio/health", "port": "http"}
                                        },
                                        "volumeMounts": [
                                            {"name": "data", "mountPath": "/data"},
                                            {
                                                "name": "contract",
                                                "mountPath": "/contract",
                                                "readOnly": True,
                                            },
                                        ],
                                    }
                                ],
                                "volumes": [
                                    {
                                        "name": "data",
                                        "persistentVolumeClaim": {"claimName": "notes-authority"},
                                    },
                                    {
                                        "name": "contract",
                                        "configMap": {"name": "notes-api-contract"},
                                    },
                                ],
                            },
                        },
                    },
                }
            )
            apply(
                {
                    "apiVersion": "v1",
                    "kind": "Service",
                    "metadata": {"name": "notes-authority", "namespace": "chio-system"},
                    "spec": {
                        "selector": {"app": "notes-authority"},
                        "ports": [{"port": 9097, "targetPort": "http"}],
                    },
                }
            )
            rollout("chio-system", "notes-authority")
            # Let kubectl choose an available loopback port; its diagnostic is
            # read only from this owned subprocess, never an unrelated listener.
            forward_log = (state / "authority-forward.log").open("w")
            forwarding = subprocess.Popen(
                [
                    "kubectl",
                    "--kubeconfig",
                    kubeconfig,
                    "-n",
                    "chio-system",
                    "port-forward",
                    "service/notes-authority",
                    ":9097",
                    "--address",
                    "127.0.0.1",
                ],
                stdout=forward_log,
                stderr=forward_log,
            )
            children.append(forwarding)
            deadline = time.monotonic() + 30
            while True:
                lines = (state / "authority-forward.log").read_text().splitlines()
                found = next(
                    (line for line in lines if line.startswith("Forwarding from 127.0.0.1:")), None
                )
                if found:
                    local_port = found.split(":")[1].split()[0]
                    break
                if forwarding.poll() is not None or time.monotonic() > deadline:
                    raise RuntimeError("Authority port forwarding did not become ready")
                time.sleep(0.1)
            issuance = {
                "subject": "istio-notes-writer",
                "job_uid": name,
                "ttl_seconds": 300,
                "scopes": ["tool:chio_http_authority:authorize_http_request:invoke"],
            }
            token = authority("/v1/capabilities/mint", issuance)["capability"]
            signer = token["issuer"]
            (state / "trusted-kernel-key.txt").write_text(signer + "\n")
            config = {
                "image": image_references["runtime"],
                "authority_url": "http://notes-authority.chio-system.svc.cluster.local:9097",
                "trusted_kernel_key": signer,
            }
            (state / "deployment.json").write_text(json.dumps(config, indent=2) + "\n")
            apply(resources(config))
            rollout("chio-system", "chio-sidecar")
            kube("apply", "-f", str(ROOT / "02-authorization-policy.yaml"))
            apply(
                {
                    "apiVersion": "v1",
                    "kind": "Pod",
                    "metadata": {"name": "client", "namespace": "chio-system"},
                    "spec": {
                        "automountServiceAccountToken": False,
                        "containers": [
                            {
                                "name": "client",
                                "image": PYTHON_IMAGE,
                                "command": ["python3", "-c", "import time; time.sleep(3600)"],
                            }
                        ],
                    },
                }
            )
            kube(
                "-n", "chio-system", "wait", "--for=condition=Ready", "pod/client", "--timeout=180s"
            )
            run(["istioctl", "--kubeconfig", kubeconfig, "proxy-status"], capture=False)
            # Establish policy propagation with a read-only request. A 200 from
            # an unprotected route is insufficient: admission must supply its ID.
            deadline = time.monotonic() + 60
            while True:
                try:
                    client()
                    break
                except AssertionError:
                    if time.monotonic() >= deadline:
                        raise
                    time.sleep(0.5)
            print("Testing the actual Istio request path", flush=True)
            client("/healthz")
            before = client()["notes"]
            client(body={"text": "ungranted write must not appear"}, expected=403)
            created = client(
                body={"text": "Review the Istio deployment"}, grant=token, expected=201
            )
            after = client()["notes"]
            assert len(after) == len(before) + 1 and any(
                note["id"] == created["id"] for note in after
            )
            authority("/v1/capabilities/release", {"capability_id": token["id"]})
            client(body={"text": "revoked write must not appear"}, grant=token, expected=403)
            assert client()["notes"] == after
            # Export current records through SQLite itself, including committed
            # WAL data; copying a live database file would lose that consistency.
            export = "import sqlite3,json; c=sqlite3.connect('file:/data/receipts.db?mode=ro',uri=True); print(json.dumps([json.loads(r[0]) for r in c.execute('SELECT receipt_json FROM http_receipts')]))"
            records = json.loads(
                kube(
                    "-n",
                    "chio-system",
                    "exec",
                    "deployment/notes-authority",
                    "--",
                    "python3",
                    "-c",
                    export,
                    capture=True,
                )
            )
            by_id = {record["id"]: record for record in records}
            for observation in observations:
                if observation["path"] == "/healthz":
                    continue
                receipt = by_id[observation["receipt_id"]]
                assert verify_http_receipt_with_trusted_signers(receipt, [signer])["ok"]
                body = observation["request_body"]
                binding = {
                    "method": "GET" if body is None else "POST",
                    "route_pattern": "/notes",
                    "path": "/notes",
                    "query": {},
                    "body_hash": None,
                }
                if body is not None:
                    binding["body_hash"] = hashlib.sha256(json.dumps(body).encode()).hexdigest()
                assert receipt["content_hash"] == sha256_hex_utf8(canonicalize_json(binding)), {
                    "receipt_id": receipt["id"],
                    "expected_binding": binding,
                    "observed_hash": receipt["content_hash"],
                }
                expected_verdict = "deny" if observation["status"] == 403 else "allow"
                assert receipt["verdict"]["verdict"] == expected_verdict
                assert receipt.get("capability_id") == observation["capability_id"]
                assert receipt["method"] == binding["method"]
                assert receipt["route_pattern"] == "/notes"
                assert receipt.get("metadata", {}).get("chio_http_status_scope") == "decision"
                observation["receipt"] = receipt
            kube("-n", "chio-system", "scale", "deployment/notes-authority", "--replicas=0")
            kube(
                "-n",
                "chio-system",
                "wait",
                "--for=delete",
                "pod",
                "-l",
                "app=notes-authority",
                "--timeout=90s",
            )
            client(body={"text": "outage write must not appear"}, grant=token, expected=503)
            client("/healthz")
            # Read the data in the owning workload, without asserting that an
            # unavailable authorization service should admit a diagnostic read.
            check = "import sqlite3,json; c=sqlite3.connect('file:/data/notes.db?mode=ro',uri=True); print(json.dumps([{'id':r[0],'text':r[1]} for r in c.execute('SELECT id,text FROM notes ORDER BY id')]))"
            retained = json.loads(
                kube(
                    "-n",
                    "agent-tools",
                    "exec",
                    "deployment/demo-tool",
                    "-c",
                    "notes",
                    "--",
                    "python3",
                    "-c",
                    check,
                    capture=True,
                )
            )
            assert retained == after
            (state / "verification.json").write_text(
                json.dumps(
                    {
                        "ok": True,
                        "kubernetes_image": NODE_IMAGE,
                        "images": image_references,
                        "trusted_kernel_key": signer,
                        "observations": observations,
                        "retained_notes": retained,
                    },
                    indent=2,
                )
                + "\n"
            )
            print(
                "Passed: Istio allow, missing grant, revocation, exact signed bodies, and outage without another effect.",
                flush=True,
            )
            print("Evidence:", state.relative_to(ROOT), flush=True)
        finally:
            for process in children:
                if process.poll() is None:
                    process.terminate()
                try:
                    process.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    process.kill()
                    process.wait()
            if forward_log is not None:
                forward_log.close()
            if cluster_created:
                kube("get", "pods", "-A", check=False, timeout=30)
                run(
                    ["kind", "export", "logs", "--name", name, str(state / "cluster-logs")],
                    check=False,
                    timeout=120,
                )
                run(
                    ["kind", "delete", "cluster", "--name", name, "--kubeconfig", kubeconfig],
                    check=False,
                    timeout=120,
                )
            if registry_created:
                run(["docker", "rm", "-f", registry], check=False, timeout=30)


if __name__ == "__main__":
    main()

"""Render Kubernetes resources from an operator's explicit deployment configuration.

Input JSON fields: image (registry/image@sha256:...), authority_url (HTTP(S)
origin), trusted_kernel_key (64 hexadecimal characters). The public key must
come from your authority operator's trusted configuration. This command renders
resources; it does not change your cluster or replace Istio MeshConfig.
"""

import argparse
import json
from pathlib import Path
import re
from urllib.parse import urlparse


def resources(config):
    if not isinstance(config, dict) or set(config) != {
        "image",
        "authority_url",
        "trusted_kernel_key",
    }:
        raise ValueError("Configuration needs exactly image, authority_url and trusted_kernel_key")
    image = config["image"]
    key = config["trusted_kernel_key"]
    if not isinstance(config["authority_url"], str):
        raise ValueError("authority_url must be an HTTP(S) origin")
    origin = urlparse(config["authority_url"])
    if not isinstance(image, str) or not re.fullmatch(r"[^\s]+@sha256:[0-9a-f]{64}", image):
        raise ValueError("image must select your built registry image by its sha256 digest")
    if not isinstance(key, str) or not re.fullmatch(r"[0-9a-fA-F]{64}", key):
        raise ValueError(
            "trusted_kernel_key must be the operator-selected 32-byte public key in hex"
        )
    if (
        origin.scheme not in ("http", "https")
        or not origin.hostname
        or origin.username
        or origin.password
        or origin.query
        or origin.fragment
        or origin.path not in ("", "/")
    ):
        raise ValueError("authority_url must be an HTTP(S) origin without credentials or a path")
    namespace = "chio-system"
    labels = {"app.kubernetes.io/name": "chio-sidecar"}
    metadata = lambda name: {"name": name, "namespace": namespace, "labels": labels}
    container = {
        "name": "authorization",
        "image": image,
        "args": [
            "--listen",
            "0.0.0.0:9091",
            "--health-listen",
            "0.0.0.0:9092",
            "--authority-url",
            config["authority_url"],
            "--trusted-kernel-key-file",
            "/etc/chio/kernel-key.txt",
        ],
        "ports": [
            {"name": "grpc", "containerPort": 9091},
            {"name": "health", "containerPort": 9092},
        ],
        "volumeMounts": [{"name": "trust", "mountPath": "/etc/chio", "readOnly": True}],
        "securityContext": {
            "allowPrivilegeEscalation": False,
            "readOnlyRootFilesystem": True,
            "capabilities": {"drop": ["ALL"]},
        },
        "resources": {
            "requests": {"cpu": "100m", "memory": "128Mi"},
            "limits": {"cpu": "1", "memory": "512Mi"},
        },
        "startupProbe": {
            "httpGet": {"path": "/healthz", "port": "health"},
            "periodSeconds": 1,
            "failureThreshold": 30,
        },
        "livenessProbe": {"httpGet": {"path": "/healthz", "port": "health"}, "periodSeconds": 10},
        "readinessProbe": {"httpGet": {"path": "/readyz", "port": "health"}, "periodSeconds": 5},
    }
    return {
        "apiVersion": "v1",
        "kind": "List",
        "items": [
            {
                "apiVersion": "v1",
                "kind": "Namespace",
                "metadata": {"name": namespace, "labels": {"istio-injection": "disabled"}},
            },
            {
                "apiVersion": "v1",
                "kind": "ConfigMap",
                "metadata": metadata("chio-authority-trust"),
                "data": {"kernel-key.txt": key.lower() + "\n"},
            },
            {
                "apiVersion": "apps/v1",
                "kind": "Deployment",
                "metadata": metadata("chio-sidecar"),
                "spec": {
                    "replicas": 2,
                    "selector": {"matchLabels": labels},
                    "template": {
                        "metadata": {"labels": labels},
                        "spec": {
                            "automountServiceAccountToken": False,
                            "securityContext": {
                                "runAsNonRoot": True,
                                "runAsUser": 65532,
                                "runAsGroup": 65532,
                                "seccompProfile": {"type": "RuntimeDefault"},
                            },
                            "containers": [container],
                            "volumes": [
                                {
                                    "name": "trust",
                                    "configMap": {
                                        "name": "chio-authority-trust",
                                        "defaultMode": 292,
                                    },
                                }
                            ],
                        },
                    },
                },
            },
            {
                "apiVersion": "v1",
                "kind": "Service",
                "metadata": metadata("chio-sidecar"),
                "spec": {
                    "selector": labels,
                    "ports": [
                        {"name": "grpc", "port": 9091, "targetPort": "grpc", "appProtocol": "grpc"},
                        {
                            "name": "health",
                            "port": 9092,
                            "targetPort": "health",
                            "appProtocol": "http",
                        },
                    ],
                },
            },
        ],
    }


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("config", type=Path)
    args = parser.parse_args()
    try:
        print(json.dumps(resources(json.loads(args.config.read_text())), indent=2))
    except (ValueError, TypeError, OSError) as error:
        parser.error(str(error))

static_resources:
  listeners:
    - name: notes
      address:
        socket_address: {address: 127.0.0.1, port_value: 10000}
      filter_chains:
        - filters:
            - name: envoy.filters.network.http_connection_manager
              typed_config:
                '@type': type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
                stat_prefix: governed_notes
                route_config:
                  name: notes
                  virtual_hosts:
                    - name: notes
                      domains: ['*']
                      routes:
                        - match: {prefix: /}
                          route: {cluster: notes}
                http_filters:
                  - name: envoy.filters.http.ext_authz
                    typed_config:
                      '@type': type.googleapis.com/envoy.extensions.filters.http.ext_authz.v3.ExtAuthz
                      transport_api_version: V3
                      failure_mode_allow: false
                      clear_route_cache: false
                      with_request_body:
                        max_request_bytes: 65536
                        allow_partial_message: false
                        pack_as_bytes: true
                      grpc_service:
                        envoy_grpc: {cluster_name: chio_authorization}
                        timeout: 3s
                  - name: envoy.filters.http.router
                    typed_config:
                      '@type': type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
  clusters:
    - name: notes
      type: STATIC
      connect_timeout: 1s
      load_assignment:
        cluster_name: notes
        endpoints:
          - lb_endpoints:
              - endpoint:
                  address:
                    socket_address: {address: 127.0.0.1, port_value: 8087}
    - name: chio_authorization
      type: STATIC
      connect_timeout: 1s
      typed_extension_protocol_options:
        envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
          '@type': type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
          explicit_http_config: {http2_protocol_options: {}}
      load_assignment:
        cluster_name: chio_authorization
        endpoints:
          - lb_endpoints:
              - endpoint:
                  address:
                    socket_address: {address: 127.0.0.1, port_value: 9091}

"""Install the example's checksum-pinned cluster tools into .tools/.

Uses official kind, Kubernetes and Istio releases. Does not change PATH,
system installations, Docker configuration or any existing cluster.
"""

import hashlib
import io
import json
import os
from pathlib import Path
import platform
import tarfile
import tempfile
import urllib.request

ROOT = Path(__file__).resolve().parent


def main():
    system = platform.system().lower()
    architecture = {"x86_64": "amd64", "aarch64": "arm64", "arm64": "arm64"}.get(platform.machine())
    releases = json.loads((ROOT / "kubernetes-tools.json").read_text())
    selection = releases.get(f"{system}-{architecture}")
    if selection is None:
        raise SystemExit("This installer supports Linux and macOS on x86_64 or arm64.")
    destination = ROOT / ".tools"
    destination.mkdir(exist_ok=True)
    for name, release in selection.items():
        print(f"Downloading {name} from {release['url']}", flush=True)
        with urllib.request.urlopen(release["url"], timeout=120) as response:
            content = response.read(200 * 1024 * 1024 + 1)
        if hashlib.sha256(content).hexdigest() != release["sha256"]:
            raise SystemExit(f"{name}: release checksum does not match; nothing installed.")
        if name == "istioctl":
            # Read the exact regular-file member, never extract archive paths.
            with tarfile.open(fileobj=io.BytesIO(content)) as archive:
                member = archive.getmember("istio-1.31.0/bin/istioctl")
                if not member.isfile() or member.size > 200 * 1024 * 1024:
                    raise SystemExit("The Istio archive does not contain the expected binary.")
                source = archive.extractfile(member)
                if source is None:
                    raise SystemExit("The Istio binary could not be read.")
                content = source.read()
        temporary = None
        try:
            with tempfile.NamedTemporaryFile(dir=destination, delete=False) as output:
                temporary = Path(output.name)
                output.write(content)
                output.flush()
                os.fsync(output.fileno())
            temporary.chmod(0o755)
            os.replace(temporary, destination / name)
        finally:
            if temporary is not None:
                temporary.unlink(missing_ok=True)
    print("Installed kind 0.33.0, Kubernetes CLI 1.36.4 and Istio 1.31.0 into .tools/.")


if __name__ == "__main__":
    main()