Trust Control Walkthrough

"""Larkfield's pricing API over HTTP, with SQLite persistence and no Chio dependency."""
import argparse
from decimal import ROUND_HALF_UP, Decimal
import json
import sqlite3
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit

PRICES = {"sku-118": Decimal("24.00"), "sku-204": Decimal("58.00"), "sku-301": Decimal("140.00")}
CENT = Decimal("0.01")
PREFIX = "/v1/quotes/"


def row_to_quote(row):
    return {"id": row[0], "sku": row[1], "percent": row[2], "price": row[3], "currency": "USD"}


class Pricing(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 == "/v1/quotes":
                rows = database.execute("SELECT id, sku, percent, price FROM quotes ORDER BY id LIMIT 100").fetchall()
                return self.respond(200, {"quotes": [row_to_quote(row) for row in rows]})
            if path.startswith(PREFIX) and path[len(PREFIX):].isdigit():
                row = database.execute("SELECT id, sku, percent, price FROM quotes WHERE id = ?", (int(path[len(PREFIX):]),)).fetchone()
                if row:
                    return self.respond(200, row_to_quote(row))
        self.respond(404, {"error": "quote_not_found"})

    def do_POST(self):
        if urlsplit(self.path).path != "/v1/quote":
            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)
            if not isinstance(payload, dict) or set(payload) != {"sku", "percent"}:
                return self.respond(400, {"error": "body_must_contain_sku_and_percent"})
            sku, percent = payload["sku"], payload["percent"]
            if sku not in PRICES:
                return self.respond(400, {"error": "unknown_sku"})
            if isinstance(percent, bool) or not isinstance(percent, (int, float)) or not 0 <= percent <= 100:
                return self.respond(400, {"error": "percent_must_be_0_to_100"})
        except (ValueError, UnicodeError):
            return self.respond(400, {"error": "invalid_json_body"})
        price = str((PRICES[sku] * (1 - Decimal(str(percent)) / 100)).quantize(CENT, ROUND_HALF_UP))
        with sqlite3.connect(self.server.database) as database:
            cursor = database.execute("INSERT INTO quotes(sku, percent, price) VALUES (?, ?, ?)", (sku, percent, price))
            quote_id = cursor.lastrowid
        self.respond(201, {"id": quote_id, "sku": sku, "percent": percent, "price": price, "currency": "USD"})

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


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

{
  "id": "01a12f17-db6b-7f20-bb9d-e94cde41d740",
  "request_id": "01a12f17-db5c-7310-bbb9-0b029cbb53e6",
  "route_pattern": "/v1/quote",
  "method": "POST",
  "caller_identity_hash": "5a45e80d1ab1a73e8c2644f3f6c6e53974d0a986c2fc86692372d3f130248c9c",
  "verdict": {
    "verdict": "allow"
  },
  "evidence": [
    {
      "guard_name": "CapabilityGuard",
      "verdict": true,
      "details": "valid capability token presented"
    }
  ],
  "response_status": 201,
  "timestamp": 1791791455,
  "content_hash": "67bcc3e408e5bca9eea78d7a74e951678c8267159500e6529978658236dc9b77",
  "policy_hash": "9d644d6ad425e9840795f666f3868141d067f06f6c86de72c3713ba8fa3a868a",
  "capability_id": "cap-01a12f17-da8d-7820-ad45-bafd8a01f4c6",
  "metadata": {
    "chio_decision_receipt_id": "01a12f17-db5e-7480-af88-595ae5f2813b",
    "chio_http_status_scope": "final",
    "chio_kernel_receipt_id": "rcpt-01a12f17-db5e-7480-af88-594a8a2c0784",
    "route_selection": {
      "candidates": [
        {
          "available": true,
          "routeId": "native-route",
          "selectedProtocols": [
            "http",
            "native"
          ],
          "targetProtocol": "native"
        }
      ],
      "decision": "select",
      "requestedTargetProtocol": "native",
      "routeSelectionId": "8b9a5d95f044f00a7566024c5ff2b6cf38b65828f523c719b27bcaab0df7bfab",
      "selectedProtocols": [
        "http",
        "native"
      ],
      "selectedRouteId": "native-route",
      "selectedTargetProtocol": "native",
      "sourceProtocol": "http"
    }
  },
  "kernel_key": "6d430942a78dff59221c9abbd466d0ebd1ac7f106b2ff76ac68977ade20ba3e9",
  "signature": "7c7c2dcb0a262b83da7ee5290d8a80a05841c008e4d8e1c7ac62edba63db11634472973629c6fa87554fc71bb2d99d85dda83b115a691df85c4bff196ae48a0b"
}

# /// script
# requires-python = ">=3.11"
# dependencies = ["pynacl>=1.5,<2", "chio-sdk"]
# [tool.uv.sources]
# chio-sdk = { path = "sdk/chio-py" }
# ///
"""Issue a quote, revoke the caller's actual capability, then verify the refused repeated quote."""

import argparse
import hashlib
import json
import os
import secrets
import shutil
import socket
import sqlite3
import subprocess
import sys
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 nacl.signing import SigningKey

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


def http(base, path, body=None, *, bearer=None, capability=None):
    headers = {"Content-Type": "application/json"}
    if bearer:
        headers["Authorization"] = "Bearer " + bearer
    if capability:
        headers["X-Chio-Capability"] = json.dumps(capability, separators=(",", ":"))
    request = urllib.request.Request(
        base + path, headers=headers, data=None if body is None else json.dumps(body).encode()
    )
    try:
        response = urllib.request.urlopen(request, timeout=15)
    except urllib.error.HTTPError as error:
        response = error
    with response:
        payload = json.loads(response.read())
        return response.status, response.headers.get("X-Chio-Receipt-Id"), payload


def seed(path):
    if not path.exists():
        descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(descriptor, "w") as output:
            output.write(SigningKey.generate().encode().hex() + "\n")
    return SigningKey(bytes.fromhex(path.read_text().strip()))


def port():
    with socket.socket() as listener:
        listener.bind(("127.0.0.1", 0))
        return listener.getsockname()[1]


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("sku", nargs="?", default="sku-204", choices=["sku-118", "sku-204", "sku-301"])
    parser.add_argument("percent", nargs="?", type=float, default=15)
    args = parser.parse_args()
    binary = os.environ.get("CHIO_BIN") or shutil.which("chio")
    if not binary:
        parser.error("Install the Chio CLI selected in README.md, or set CHIO_BIN")
    binary = str(Path(binary).resolve())
    state = ROOT / ".state"
    state.mkdir(mode=0o700, exist_ok=True)
    state.chmod(0o700)
    # One operator process owns issuance and revocation. Concurrent starts must
    # not reuse the same state or impersonate a listener from another run.
    import fcntl

    lock = (state / "operator.lock").open("a")
    try:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        parser.error("This application is already running; wait for it to finish")
    caller = seed(state / "caller.seed")
    issuer = seed(state / "trust.seed")
    mediator = seed(state / "mediator.seed")
    signer = mediator.verify_key.encode().hex()
    run = state / "runs" / str(uuid.uuid4())
    run.mkdir(parents=True, mode=0o700)
    (run / "trusted-kernel-key.txt").write_text(signer + "\n")
    operator_token = secrets.token_hex(32)
    api, control, proxy = [f"http://127.0.0.1:{port()}" for _ in range(3)]
    children = []
    logs = []

    def start(name, command, environment=None):
        log = (run / (name + ".log")).open("w")
        logs.append(log)
        process = subprocess.Popen(
            command, cwd=ROOT, env=environment, stdin=subprocess.DEVNULL, stdout=log, stderr=log
        )
        children.append(process)
        return process

    def stop(process):
        process.terminate()
        try:
            process.wait(timeout=10)
        except subprocess.TimeoutExpired:
            process.kill()
            process.wait()
        children.remove(process)

    def ready(base, path):
        for _ in range(150):
            if any(child.poll() is not None for child in children):
                raise RuntimeError(f"A service stopped; inspect {run.relative_to(ROOT)}/*.log")
            try:
                if http(base, path)[0] == 200:
                    return
            except (OSError, ValueError):
                pass
            time.sleep(0.1)
        raise RuntimeError("Service startup timed out; inspect the retained logs")

    try:
        start(
            "pricing",
            [sys.executable, "pricing_api.py", "--port", api.rsplit(":", 1)[1], "--data", str(state)],
        )
        trust_command = [
            binary,
            "trust",
            "serve",
            "--listen",
            control.removeprefix("http://"),
            "--service-token",
            operator_token,
            "--authority-seed-file",
            str(state / "trust.seed"),
            "--session-db",
            str(state / "authority.db"),
            "--receipt-db",
            str(state / "trust-receipts.db"),
        ]
        trust_process = start("trust", trust_command)
        ready(api, "/healthz")
        ready(control, "/health")
        mediator_command = [
            binary,
            "--control-url",
            control,
            "--control-token",
            operator_token,
            "api",
            "protect",
            "--upstream",
            api,
            "--spec",
            "openapi.yaml",
            "--listen",
            proxy.removeprefix("http://"),
            "--authority-seed-file",
            str(state / "mediator.seed"),
            "--receipt-store",
            str(state / "receipts.db"),
        ]
        mediator_environment = dict(
            os.environ, CHIO_TRUSTED_ISSUER_KEY=issuer.verify_key.encode().hex()
        )
        mediator_process = start("mediator", mediator_command, mediator_environment)
        ready(proxy, "/chio/health")
        issuance = {
            "subjectPublicKey": caller.verify_key.encode().hex(),
            "ttlSeconds": 300,
            "scope": {
                "grants": [
                    {
                        "server_id": "chio_http_authority",
                        "tool_name": "authorize_http_request",
                        "operations": ["invoke"],
                        "constraints": [],
                    }
                ],
                "resource_grants": [],
                "prompt_grants": [],
            },
        }
        status, _, issued = http(control, "/v1/capabilities/issue", issuance, bearer=operator_token)
        if status != 200:
            raise RuntimeError(f"Capability issuance failed: {status}: {issued}")
        capability = issued["capability"]
        if capability["subject"] != issuance["subjectPublicKey"]:
            raise ValueError("Issued capability has the wrong subject")
        (run / "capability.json").write_text(json.dumps(capability, indent=2))
        (run / "capability.json").chmod(0o600)
        before = http(proxy, "/v1/quotes")[2]["quotes"]
        percent = int(args.percent) if args.percent == int(args.percent) else args.percent
        body = {"sku": args.sku, "percent": percent}
        status, allowed_id, saved = http(proxy, "/v1/quote", body, capability=capability)
        if status != 201 or not allowed_id:
            raise ValueError(f"Authorized quote failed: {status}: {saved}")
        after = http(proxy, "/v1/quotes")[2]["quotes"]
        if len(after) != len(before) + 1 or saved["id"] not in [quote["id"] for quote in after]:
            raise ValueError("The allowed quote was not recorded")
        status, _, revocation = http(
            control, "/v1/revocations", {"capabilityId": capability["id"]}, bearer=operator_token
        )
        if status != 200:
            raise ValueError(f"Revocation failed: {status}: {revocation}")
        status, denied_id, denied = http(proxy, "/v1/quote", body, capability=capability)
        if status != 403 or not denied_id:
            raise ValueError(f"The revoked grant was not refused: {status}: {denied}")
        if http(proxy, "/v1/quotes")[2]["quotes"] != after:
            raise ValueError("The refused request changed application data")
        associations = [
            ("allowed", allowed_id, "allow", capability["id"]),
            ("revoked", denied_id, "deny", capability["id"]),
        ]
        stop(mediator_process)
        stop(trust_process)
        trust_process = start("trust-restarted", trust_command)
        ready(control, "/health")
        mediator_process = start("mediator-restarted", mediator_command, mediator_environment)
        ready(proxy, "/chio/health")
        restarted_status, restarted_id, _ = http(proxy, "/v1/quote", body, capability=capability)
        if restarted_status != 403 or not restarted_id:
            raise ValueError("Restart lost the revocation")
        associations.append(("after_restart", restarted_id, "deny", capability["id"]))
        live_status, _, live_issued = http(
            control, "/v1/capabilities/issue", issuance, bearer=operator_token
        )
        if live_status != 200:
            raise ValueError("Could not issue the outage-control grant")
        stop(trust_process)
        outage_status, outage_id, _ = http(
            proxy, "/v1/quote", body, capability=live_issued["capability"]
        )
        if outage_status != 503 or not outage_id:
            raise ValueError("Unavailable revocation authority did not fail closed")
        associations.append(
            ("authority_unavailable", outage_id, "deny", live_issued["capability"]["id"])
        )
        if http(api, "/v1/quotes")[2]["quotes"] != after:
            raise ValueError("Restart or outage caused an unauthorized write")
        connection = sqlite3.connect(
            (state / "receipts.db").resolve().as_uri() + "?mode=ro", uri=True
        )
        verified = []
        try:
            for case, receipt_id, verdict, expected_capability in associations:
                row = connection.execute(
                    "SELECT receipt_json FROM http_receipts WHERE id=?", (receipt_id,)
                ).fetchone()
                if not row:
                    raise ValueError("The operation receipt is missing from its store")
                receipt = json.loads(row[0])
                checks = verify_http_receipt_with_trusted_signers(receipt, [signer])
                if not checks["ok"] or receipt["verdict"]["verdict"] != verdict:
                    raise ValueError("Receipt verification failed")
                if receipt["capability_id"] != expected_capability:
                    raise ValueError("Receipt names a different capability")
                if (
                    receipt["method"] != "POST"
                    or receipt["route_pattern"] != "/v1/quote"
                    or receipt.get("metadata", {}).get("chio_http_status_scope") != "final"
                ):
                    raise ValueError("Receipt does not describe this final HTTP operation")
                binding = {
                    "method": "POST",
                    "route_pattern": "/v1/quote",
                    "path": "/v1/quote",
                    "query": {},
                    "body_hash": hashlib.sha256(json.dumps(body).encode()).hexdigest(),
                }
                if receipt["content_hash"] != sha256_hex_utf8(canonicalize_json(binding)):
                    raise ValueError("Receipt is bound to different request bytes")
                verified.append({"case": case, "receipt": receipt, "verification": checks})
        finally:
            connection.close()
        result = {
            "subject": capability["subject"],
            "capability_id": capability["id"],
            "quote": saved,
            "before_count": len(before),
            "after_count": len(before) + 1,
            "repeated_request": body,
            "revocation": revocation,
            "restart_refusal_status": restarted_status,
            "outage_refusal_status": outage_status,
            "refusal_status": status,
            "refusal": denied,
            "receipts": verified,
        }
        (run / "verification.json").write_text(json.dumps(result, indent=2))
        print(
            f"Quote {saved['id']}: {saved['sku']} at {saved['percent']}% off, USD {saved['price']}\nRevoked {capability['id']}\nRepeated quote refused: HTTP 403; recorded quotes unchanged\nRevocation survives restart; unavailable authority refuses a live grant\nFour signed HTTP receipts verified against the operator-selected key\nEvidence: {run.relative_to(ROOT)}/verification.json"
        )
    finally:
        for child in reversed(children):
            if child.poll() is None:
                child.terminate()
        for child in reversed(children):
            try:
                child.wait(timeout=10)
            except subprocess.TimeoutExpired:
                child.kill()
                child.wait()
        for log in logs:
            log.close()
        lock.close()


if __name__ == "__main__":
    main()