feat(api): canary token CRUD router (/api/v1/canary) + tests

Two sub-routers under /api/v1/canary: blobs (operator-uploaded artifacts, deduped by sha256): - POST /blobs (multipart upload; admin) - GET /blobs (list with token_count; admin) - DELETE /blobs/{uuid} (refcount-aware; 409 when referenced; admin) tokens (per-decky planted artifacts): - POST /tokens (generate or instrument + plant; admin) - GET /tokens?decky_name=&kind=&state= (filter; viewer) - GET /tokens/{uuid} (detail; viewer) - GET /tokens/{uuid}/preview (instrumented bytes; admin) - GET /tokens/{uuid}/triggers (paged callback log; viewer) - DELETE /tokens/{uuid} (revoke + bus event; admin) XOR validation: exactly one of blob_uuid / generator must be set. Path validation rejects relative/NUL/newlines/.. segments. Every body-bearing route documents 400 plus 401/403/404 as applicable. Stdlib MIME sniffer (no python-magic dep) covers PNG/JPEG/GIF/PDF/ HTML/XML/DOCX/XLSX/JSON/YAML/TOML/text/plain; everything else falls through to passthrough. Tests run end-to-end through the live FastAPI app (planter docker exec is patched); 17 cases covering dedup, refcount, lifecycle, XOR validation, path validation, and 404 paths.
2026-04-27 13:18:00 -04:00
parent f9513bb7dd
commit 6c4ea706f8
6 changed files with 881 additions and 0 deletions
--- a/decnet/web/router/init.py
+++ b/decnet/web/router/init.py
@@ -47,6 +47,7 @@ from .swarm_updates import swarm_updates_router
 from .swarm_mgmt import swarm_mgmt_router
 from .system import system_router
 from .topology import topology_router
+from .canary import canary_router
 from .webhooks import webhooks_router

 api_router = APIRouter(
@@ -148,5 +149,9 @@ api_router.include_router(system_router)
 # MazeNET Topologies (nested topology CRUD + mutation queue)
 api_router.include_router(topology_router)

+# Canary tokens — operator-facing CRUD (worker hosts the
+# attacker-facing surface separately via `decnet canary`).
+api_router.include_router(canary_router)
+
 # External webhook subscriptions (SIEM/SOAR egress)
 api_router.include_router(webhooks_router)
--- a/decnet/web/router/canary/init.py
+++ b/decnet/web/router/canary/init.py
@@ -0,0 +1,23 @@
+"""Canary tokens — operator-facing CRUD.
+
+Mounted under ``/api/v1/canary``.  Covers:
+
+* ``POST /blobs`` — upload an artifact (multipart);
+  ``GET /blobs``, ``DELETE /blobs/{id}`` — listing + cleanup
+* ``POST /tokens`` — generate + plant a token on a target decky;
+  ``GET /tokens``, ``GET /tokens/{id}``, ``DELETE /tokens/{id}``
+  — listing + detail + revoke
+* ``GET /tokens/{id}/preview`` — instrumented bytes for sanity-check
+* ``GET /tokens/{id}/triggers`` — paged callback log
+
+The ``decnet canary`` worker runs the ATTACKER-facing surface (HTTP
+slug + DNS); this module is the OPERATOR-facing surface only.
+"""
+from fastapi import APIRouter
+
+from .api_blobs import router as blobs_router
+from .api_tokens import router as tokens_router
+
+canary_router = APIRouter(prefix="/canary")
+canary_router.include_router(blobs_router)
+canary_router.include_router(tokens_router)
--- a/decnet/web/router/canary/api_blobs.py
+++ b/decnet/web/router/canary/api_blobs.py
@@ -0,0 +1,172 @@
+"""Operator-uploaded canary blob CRUD.
+
+Three endpoints:
+
+* ``POST /blobs`` — multipart upload; sniffs MIME from the magic
+  bytes (no python-magic dependency), persists to disk under the
+  sha256 hash, returns the (possibly pre-existing) row.
+* ``GET /blobs`` — list all blobs with their live token reference
+  count.
+* ``DELETE /blobs/{uuid}`` — refcount-aware delete; returns 409 if
+  any token still references the blob.
+
+Admin-gated: blobs are operator-supplied content that may carry
+sensitive material (real-looking financial reports, etc.); listing
+them and deleting them is an admin operation.  Reading them via the
+preview path is also admin-gated.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
+
+from decnet.canary import storage
+from decnet.logging import get_logger
+from decnet.web.db.models import (
+    CanaryBlobResponse,
+    CanaryBlobsResponse,
+    MessageResponse,
+)
+from decnet.web.dependencies import repo, require_admin
+
+log = get_logger("api.canary.blobs")
+
+router = APIRouter(prefix="/blobs", tags=["Canary"])
+
+
+# --- MIME sniffing (stdlib-only, replaces python-magic) -------------------
+#
+# The DOCX/XLSX/PDF/PNG/JPEG/GIF/HTML/JSON/YAML space covers everything
+# our instrumenters know how to mutate.  Anything else falls through to
+# ``application/octet-stream`` and the API routes the token to the
+# ``passthrough`` instrumenter.
+
+_MAGIC_TABLE: tuple[tuple[bytes, str], ...] = (
+    (b"\x89PNG\r\n\x1a\n", "image/png"),
+    (b"\xff\xd8\xff", "image/jpeg"),
+    (b"GIF87a", "image/gif"),
+    (b"GIF89a", "image/gif"),
+    (b"%PDF-", "application/pdf"),
+    # OOXML (DOCX/XLSX) starts with PK\x03\x04 but so do plain zips.
+    # We disambiguate by Content_Types entry below.
+    (b"<!DOCTYPE", "text/html"),
+    (b"<html", "text/html"),
+    (b"<HTML", "text/html"),
+    (b"<?xml", "application/xml"),
+)
+
+
+def _sniff_mime(filename: str, head: bytes) -> str:
+    for marker, mime in _MAGIC_TABLE:
+        if head.startswith(marker):
+            return mime
+    if head[:4] == b"PK\x03\x04":
+        # OOXML alias detection: peek for the document-specific Override
+        # in [Content_Types].xml. We only need to look at the first
+        # block; the central directory comes later.
+        if b"wordprocessingml" in head:
+            return "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
+        if b"spreadsheetml" in head:
+            return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
+        return "application/zip"
+    # Plaintext heuristic: if the head decodes as printable utf-8 we
+    # call it text/plain — that's good enough to route to the plain
+    # instrumenter, which also handles json/yaml/toml.
+    try:
+        head.decode("utf-8")
+        if all(b in (0x09, 0x0A, 0x0D) or b >= 0x20 for b in head[:128]):
+            lf = filename.lower()
+            if lf.endswith((".json",)):
+                return "application/json"
+            if lf.endswith((".yaml", ".yml")):
+                return "application/yaml"
+            if lf.endswith((".toml",)):
+                return "application/toml"
+            return "text/plain"
+    except UnicodeDecodeError:
+        pass
+    return "application/octet-stream"
+
+
+def _row_to_response(row: dict[str, Any]) -> CanaryBlobResponse:
+    return CanaryBlobResponse(**row)
+
+
+@router.post(
+    "",
+    response_model=CanaryBlobResponse,
+    status_code=201,
+    responses={
+        400: {"description": "Empty file or unreadable upload"},
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+    },
+)
+async def api_upload_blob(
+    file: UploadFile = File(...),
+    admin: dict = Depends(require_admin),
+) -> CanaryBlobResponse:
+    content = await file.read()
+    if not content:
+        raise HTTPException(status_code=400, detail="uploaded file is empty")
+    sniffed = _sniff_mime(file.filename or "", content[:1024])
+    sha, _path, size = storage.write_blob(content)
+    row = await repo.upsert_canary_blob({
+        "sha256": sha,
+        "filename": file.filename or "(unnamed)",
+        "content_type": sniffed,
+        "size_bytes": size,
+        "uploaded_by": admin.get("uuid", "unknown"),
+        "uploaded_at": datetime.now(timezone.utc),
+    })
+    row.setdefault("token_count", 0)
+    return _row_to_response(row)
+
+
+@router.get(
+    "",
+    response_model=CanaryBlobsResponse,
+    responses={
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+    },
+)
+async def api_list_blobs(
+    admin: dict = Depends(require_admin),
+) -> CanaryBlobsResponse:
+    rows = await repo.list_canary_blobs()
+    return CanaryBlobsResponse(
+        blobs=[_row_to_response(r) for r in rows],
+        total=len(rows),
+    )
+
+
+@router.delete(
+    "/{uuid}",
+    response_model=MessageResponse,
+    responses={
+        404: {"description": "Blob not found"},
+        409: {"description": "Blob still referenced by a token"},
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+    },
+)
+async def api_delete_blob(
+    uuid: str,
+    admin: dict = Depends(require_admin),
+) -> MessageResponse:
+    existing = await repo.get_canary_blob(uuid)
+    if existing is None:
+        raise HTTPException(status_code=404, detail="blob not found")
+    deleted = await repo.delete_canary_blob(uuid)
+    if not deleted:
+        raise HTTPException(
+            status_code=409,
+            detail="blob is still referenced by one or more tokens",
+        )
+    # DB row is gone; best-effort unlink the bytes on disk.  A failure
+    # here leaves a recoverable orphan, never a dangling DB ref.
+    storage.unlink_blob(existing["sha256"])
+    return MessageResponse(message="ok")
--- a/decnet/web/router/canary/api_tokens.py
+++ b/decnet/web/router/canary/api_tokens.py
@@ -0,0 +1,318 @@
+"""Operator-facing canary token CRUD.
+
+Every body-bearing route documents the 400 error per
+:mod:`feedback_schemathesis_400`.  Auth deps:
+
+* writes (POST, DELETE) → :func:`require_admin`
+* reads (GET, preview)  → :func:`require_viewer`
+
+The router resolves blobs / instrumenters / generators here, builds
+the :class:`CanaryArtifact`, and hands it to the planter.  The
+worker is a separate process; it doesn't see this code path.
+"""
+from __future__ import annotations
+
+from secrets import token_urlsafe
+from typing import Any
+from uuid import uuid4
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+
+from decnet.canary import (
+    CanaryContext,
+    get_generator,
+    get_instrumenter,
+    pick_instrumenter_for_mime,
+    storage,
+)
+from decnet.canary.base import InstrumenterRejectedError
+from decnet.canary.factory import KNOWN_GENERATORS
+from decnet.canary.paths import normalize_placement
+from decnet.canary import planter
+from decnet.logging import get_logger
+from decnet.web.db.models import (
+    CanaryTokenCreateRequest,
+    CanaryTokenResponse,
+    CanaryTokensResponse,
+    CanaryTriggerResponse,
+    CanaryTriggersResponse,
+    MessageResponse,
+)
+from decnet.web.dependencies import repo, require_admin, require_viewer
+
+log = get_logger("api.canary.tokens")
+
+router = APIRouter(prefix="/tokens", tags=["Canary"])
+
+
+def _http_base() -> str:
+    import os
+    return os.environ.get(
+        "DECNET_CANARY_HTTP_BASE", "http://localhost:8088",
+    ).rstrip("/")
+
+
+def _dns_zone() -> str:
+    import os
+    return os.environ.get("DECNET_CANARY_DNS_ZONE", "").strip(".").lower()
+
+
+def _row_to_response(row: dict[str, Any]) -> CanaryTokenResponse:
+    return CanaryTokenResponse(**row)
+
+
+def _trigger_row_to_response(row: dict[str, Any]) -> CanaryTriggerResponse:
+    # Decode raw_headers JSON for the response shape.
+    headers = row.get("raw_headers") or "{}"
+    try:
+        import json
+        decoded = json.loads(headers) if isinstance(headers, str) else headers
+        if not isinstance(decoded, dict):
+            decoded = {}
+    except (ValueError, TypeError):
+        decoded = {}
+    out = dict(row)
+    out["headers"] = decoded
+    out.pop("raw_headers", None)
+    return CanaryTriggerResponse(**out)
+
+
+# ---------------------------------------------------------- create
+
+@router.post(
+    "",
+    response_model=CanaryTokenResponse,
+    status_code=201,
+    responses={
+        400: {"description": "Invalid token request (missing/conflicting fields, bad path, instrumenter rejection)"},
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+        404: {"description": "Referenced blob not found"},
+    },
+)
+async def api_create_token(
+    req: CanaryTokenCreateRequest,
+    admin: dict = Depends(require_admin),
+) -> CanaryTokenResponse:
+    # Exactly one of blob_uuid / generator must be set.
+    if bool(req.blob_uuid) == bool(req.generator):
+        raise HTTPException(
+            status_code=400,
+            detail="provide exactly one of blob_uuid or generator",
+        )
+    try:
+        placement_path = normalize_placement(req.placement_path)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+
+    slug = token_urlsafe(16)
+    ctx = CanaryContext(
+        callback_token=slug, http_base=_http_base(), dns_zone=_dns_zone(),
+    )
+
+    if req.generator:
+        if req.generator not in KNOWN_GENERATORS:
+            raise HTTPException(
+                status_code=400,
+                detail=f"unknown generator: {req.generator!r}",
+            )
+        generator = get_generator(req.generator)
+        artifact = generator.generate(ctx)
+        instrumenter_name = None
+    else:
+        # Upload-driven token.
+        blob = await repo.get_canary_blob(req.blob_uuid)
+        if blob is None:
+            raise HTTPException(status_code=404, detail="blob not found")
+        try:
+            blob_bytes = storage.read_blob(blob["sha256"])
+        except FileNotFoundError as e:
+            raise HTTPException(
+                status_code=410,
+                detail="blob bytes missing on disk; please re-upload",
+            ) from e
+        instrumenter_name = pick_instrumenter_for_mime(blob["content_type"])
+        ins = get_instrumenter(instrumenter_name)
+        try:
+            artifact = ins.instrument(blob_bytes, ctx, target_path=placement_path)
+        except InstrumenterRejectedError as e:
+            raise HTTPException(status_code=400, detail=str(e)) from e
+
+    artifact.path = placement_path
+    token_uuid = str(uuid4())
+    kind = req.kind
+    await repo.create_canary_token({
+        "uuid": token_uuid,
+        "kind": kind,
+        "decky_name": req.decky_name,
+        "blob_uuid": req.blob_uuid,
+        "instrumenter": instrumenter_name,
+        "generator": req.generator,
+        "placement_path": placement_path,
+        "callback_token": slug,
+        "secret_seed": slug,
+        "created_by": admin.get("uuid", "unknown"),
+        "state": "planted",
+    })
+    await planter.plant(req.decky_name, artifact, token_uuid=token_uuid, repo=repo)
+    row = await repo.get_canary_token(token_uuid)
+    return _row_to_response(row)
+
+
+# ---------------------------------------------------------- list / detail
+
+@router.get(
+    "",
+    response_model=CanaryTokensResponse,
+    responses={
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+    },
+)
+async def api_list_tokens(
+    decky_name: str | None = Query(default=None),
+    state: str | None = Query(default=None),
+    kind: str | None = Query(default=None),
+    viewer: dict = Depends(require_viewer),
+) -> CanaryTokensResponse:
+    rows = await repo.list_canary_tokens(
+        decky_name=decky_name, state=state, kind=kind,
+    )
+    return CanaryTokensResponse(
+        tokens=[_row_to_response(r) for r in rows],
+        total=len(rows),
+    )
+
+
+@router.get(
+    "/{uuid}",
+    response_model=CanaryTokenResponse,
+    responses={
+        404: {"description": "Token not found"},
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+    },
+)
+async def api_get_token(
+    uuid: str,
+    viewer: dict = Depends(require_viewer),
+) -> CanaryTokenResponse:
+    row = await repo.get_canary_token(uuid)
+    if row is None:
+        raise HTTPException(status_code=404, detail="token not found")
+    return _row_to_response(row)
+
+
+# ---------------------------------------------------------- preview
+
+@router.get(
+    "/{uuid}/preview",
+    response_class=Response,
+    responses={
+        200: {"description": "Instrumented bytes (raw)"},
+        404: {"description": "Token not found"},
+        409: {"description": "Token has no preview-able bytes (passive aws_creds, etc.)"},
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+    },
+)
+async def api_preview_token(
+    uuid: str,
+    admin: dict = Depends(require_admin),
+) -> Response:
+    """Return the instrumented bytes the planter dropped on the decky.
+
+    Re-derived deterministically from the row's ``secret_seed`` —
+    we don't store the rendered bytes server-side.  Lets operators
+    diff-check what we wrote without ``docker exec``-ing into the
+    container.
+    """
+    row = await repo.get_canary_token(uuid)
+    if row is None:
+        raise HTTPException(status_code=404, detail="token not found")
+    ctx = CanaryContext(
+        callback_token=row["callback_token"],
+        http_base=_http_base(),
+        dns_zone=_dns_zone(),
+    )
+    if row["generator"]:
+        artifact = get_generator(row["generator"]).generate(ctx)
+    elif row["blob_uuid"] and row["instrumenter"]:
+        blob = await repo.get_canary_blob(row["blob_uuid"])
+        if blob is None:
+            raise HTTPException(
+                status_code=409,
+                detail="blob has been deleted; preview unavailable",
+            )
+        try:
+            blob_bytes = storage.read_blob(blob["sha256"])
+        except FileNotFoundError as e:
+            raise HTTPException(
+                status_code=409,
+                detail="blob bytes missing on disk",
+            ) from e
+        ins = get_instrumenter(row["instrumenter"])
+        try:
+            artifact = ins.instrument(
+                blob_bytes, ctx, target_path=row["placement_path"],
+            )
+        except InstrumenterRejectedError as e:
+            raise HTTPException(status_code=409, detail=str(e)) from e
+    else:
+        raise HTTPException(
+            status_code=409,
+            detail="token has neither generator nor instrumenter — nothing to preview",
+        )
+    return Response(content=artifact.content, media_type="application/octet-stream")
+
+
+# ---------------------------------------------------------- triggers
+
+@router.get(
+    "/{uuid}/triggers",
+    response_model=CanaryTriggersResponse,
+    responses={
+        404: {"description": "Token not found"},
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+    },
+)
+async def api_list_triggers(
+    uuid: str,
+    limit: int = Query(default=100, ge=1, le=500),
+    offset: int = Query(default=0, ge=0),
+    viewer: dict = Depends(require_viewer),
+) -> CanaryTriggersResponse:
+    row = await repo.get_canary_token(uuid)
+    if row is None:
+        raise HTTPException(status_code=404, detail="token not found")
+    rows = await repo.list_canary_triggers(uuid, limit=limit, offset=offset)
+    return CanaryTriggersResponse(
+        triggers=[_trigger_row_to_response(r) for r in rows],
+        total=len(rows),
+    )
+
+
+# ---------------------------------------------------------- revoke
+
+@router.delete(
+    "/{uuid}",
+    response_model=MessageResponse,
+    responses={
+        404: {"description": "Token not found"},
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+    },
+)
+async def api_revoke_token(
+    uuid: str,
+    admin: dict = Depends(require_admin),
+) -> MessageResponse:
+    row = await repo.get_canary_token(uuid)
+    if row is None:
+        raise HTTPException(status_code=404, detail="token not found")
+    await planter.revoke(
+        row["decky_name"], row["placement_path"],
+        token_uuid=uuid, repo=repo,
+    )
+    return MessageResponse(message="ok")
--- a/tests/api/canary/init.py
+++ b/tests/api/canary/init.py
--- a/tests/api/canary/test_canary_tokens_api.py
+++ b/tests/api/canary/test_canary_tokens_api.py
@@ -0,0 +1,363 @@
+"""End-to-end coverage for /api/v1/canary/* via the live FastAPI app.
+
+The planter's docker-exec call is patched so we don't need a real
+docker daemon; everything else (DB, repo, instrumenters, generators,
+storage) runs for real.
+"""
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import patch
+
+import httpx
+import pytest
+
+
+_BASE = "/api/v1/canary"
+
+
+class _FakeProc:
+    def __init__(self, rc: int = 0, stderr: bytes = b"") -> None:
+        self.returncode = rc
+        self._stderr = stderr
+
+    async def communicate(self) -> tuple[bytes, bytes]:
+        return b"", self._stderr
+
+    def kill(self) -> None:  # pragma: no cover
+        pass
+
+
+def _patch_subprocess(rc: int = 0, stderr: bytes = b""):
+    async def _fake(*argv, **kw):  # noqa: ANN001
+        return _FakeProc(rc, stderr)
+    return patch.object(asyncio, "create_subprocess_exec", _fake)
+
+
+def _hdr(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+# ---------------- blob upload ---------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_blob_upload_dedupes(
+    client: httpx.AsyncClient, auth_token: str, tmp_path, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_BLOB_DIR", str(tmp_path))
+    files = {"file": ("notes.txt", b"hello canary", "text/plain")}
+    res = await client.post(f"{_BASE}/blobs", files=files, headers=_hdr(auth_token))
+    assert res.status_code == 201, res.text
+    first = res.json()
+    # Re-uploading the same bytes returns the same uuid.
+    files2 = {"file": ("notes-rename.txt", b"hello canary", "text/plain")}
+    res2 = await client.post(f"{_BASE}/blobs", files=files2, headers=_hdr(auth_token))
+    assert res2.status_code == 201
+    assert res2.json()["uuid"] == first["uuid"]
+
+
+@pytest.mark.asyncio
+async def test_blob_upload_rejects_empty(
+    client: httpx.AsyncClient, auth_token: str, tmp_path, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_BLOB_DIR", str(tmp_path))
+    files = {"file": ("empty.txt", b"", "text/plain")}
+    res = await client.post(f"{_BASE}/blobs", files=files, headers=_hdr(auth_token))
+    assert res.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_blob_list_carries_token_count(
+    client: httpx.AsyncClient, auth_token: str, tmp_path, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_BLOB_DIR", str(tmp_path))
+    monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "https://canary.test")
+    files = {"file": ("x.txt", b"some text", "text/plain")}
+    blob = (await client.post(
+        f"{_BASE}/blobs", files=files, headers=_hdr(auth_token),
+    )).json()
+    # Initially zero references.
+    res = await client.get(f"{_BASE}/blobs", headers=_hdr(auth_token))
+    assert res.status_code == 200
+    body = res.json()
+    assert body["total"] == 1 and body["blobs"][0]["token_count"] == 0
+    # Bind a token to bump the count.
+    with _patch_subprocess(rc=0):
+        tok_res = await client.post(
+            f"{_BASE}/tokens",
+            json={
+                "decky_name": "web1", "kind": "http",
+                "placement_path": "/etc/x.conf", "blob_uuid": blob["uuid"],
+            },
+            headers=_hdr(auth_token),
+        )
+    assert tok_res.status_code == 201, tok_res.text
+    res = await client.get(f"{_BASE}/blobs", headers=_hdr(auth_token))
+    assert res.json()["blobs"][0]["token_count"] == 1
+
+
+@pytest.mark.asyncio
+async def test_blob_delete_refuses_when_referenced(
+    client: httpx.AsyncClient, auth_token: str, tmp_path, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_BLOB_DIR", str(tmp_path))
+    monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "https://canary.test")
+    files = {"file": ("x.txt", b"more text", "text/plain")}
+    blob = (await client.post(
+        f"{_BASE}/blobs", files=files, headers=_hdr(auth_token),
+    )).json()
+    with _patch_subprocess(rc=0):
+        await client.post(
+            f"{_BASE}/tokens",
+            json={
+                "decky_name": "web1", "kind": "http",
+                "placement_path": "/etc/x.conf", "blob_uuid": blob["uuid"],
+            },
+            headers=_hdr(auth_token),
+        )
+    res = await client.delete(
+        f"{_BASE}/blobs/{blob['uuid']}", headers=_hdr(auth_token),
+    )
+    assert res.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_blob_delete_404_for_missing(
+    client: httpx.AsyncClient, auth_token: str
+) -> None:
+    res = await client.delete(
+        f"{_BASE}/blobs/00000000-0000-0000-0000-000000000000",
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 404
+
+
+# ---------------- token lifecycle ----------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_create_token_requires_xor_blob_or_generator(
+    client: httpx.AsyncClient, auth_token: str
+) -> None:
+    res = await client.post(
+        f"{_BASE}/tokens",
+        json={"decky_name": "w", "kind": "http", "placement_path": "/x"},
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 400
+    res = await client.post(
+        f"{_BASE}/tokens",
+        json={
+            "decky_name": "w", "kind": "http", "placement_path": "/x",
+            "generator": "aws_creds", "blob_uuid": "u",
+        },
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_create_token_rejects_relative_path(
+    client: httpx.AsyncClient, auth_token: str
+) -> None:
+    res = await client.post(
+        f"{_BASE}/tokens",
+        json={
+            "decky_name": "w", "kind": "http",
+            "placement_path": "relative/path", "generator": "env_file",
+        },
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_create_token_with_unknown_generator(
+    client: httpx.AsyncClient, auth_token: str
+) -> None:
+    res = await client.post(
+        f"{_BASE}/tokens",
+        json={
+            "decky_name": "w", "kind": "http",
+            "placement_path": "/x", "generator": "bogus",
+        },
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_create_token_with_missing_blob(
+    client: httpx.AsyncClient, auth_token: str
+) -> None:
+    res = await client.post(
+        f"{_BASE}/tokens",
+        json={
+            "decky_name": "w", "kind": "http",
+            "placement_path": "/x",
+            "blob_uuid": "00000000-0000-0000-0000-000000000000",
+        },
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_token_list_filter_by_decky(
+    client: httpx.AsyncClient, auth_token: str, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "https://canary.test")
+    with _patch_subprocess(rc=0):
+        for decky in ("web1", "web2"):
+            await client.post(
+                f"{_BASE}/tokens",
+                json={
+                    "decky_name": decky, "kind": "http",
+                    "placement_path": "/x", "generator": "env_file",
+                },
+                headers=_hdr(auth_token),
+            )
+    res = await client.get(
+        f"{_BASE}/tokens?decky_name=web1", headers=_hdr(auth_token),
+    )
+    assert res.status_code == 200
+    body = res.json()
+    assert body["total"] == 1
+    assert body["tokens"][0]["decky_name"] == "web1"
+
+
+@pytest.mark.asyncio
+async def test_token_detail_404(
+    client: httpx.AsyncClient, auth_token: str
+) -> None:
+    res = await client.get(
+        f"{_BASE}/tokens/00000000-0000-0000-0000-000000000000",
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_revoke_token_404(
+    client: httpx.AsyncClient, auth_token: str
+) -> None:
+    res = await client.delete(
+        f"{_BASE}/tokens/00000000-0000-0000-0000-000000000000",
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_revoke_token_succeeds(
+    client: httpx.AsyncClient, auth_token: str, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "https://canary.test")
+    with _patch_subprocess(rc=0):
+        created = (await client.post(
+            f"{_BASE}/tokens",
+            json={
+                "decky_name": "web1", "kind": "http",
+                "placement_path": "/etc/x.env", "generator": "env_file",
+            },
+            headers=_hdr(auth_token),
+        )).json()
+        res = await client.delete(
+            f"{_BASE}/tokens/{created['uuid']}",
+            headers=_hdr(auth_token),
+        )
+    assert res.status_code == 200, res.text
+    detail = (await client.get(
+        f"{_BASE}/tokens/{created['uuid']}", headers=_hdr(auth_token),
+    )).json()
+    assert detail["state"] == "revoked"
+
+
+# ---------------- preview -------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_preview_synthesised_token(
+    client: httpx.AsyncClient, auth_token: str, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "https://canary.test")
+    with _patch_subprocess(rc=0):
+        created = (await client.post(
+            f"{_BASE}/tokens",
+            json={
+                "decky_name": "web1", "kind": "http",
+                "placement_path": "/etc/x.env", "generator": "env_file",
+            },
+            headers=_hdr(auth_token),
+        )).json()
+    res = await client.get(
+        f"{_BASE}/tokens/{created['uuid']}/preview",
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 200
+    # Slug round-trips into the previewed bytes (env_file embeds it
+    # in API_BASE_URL).
+    assert created["callback_token"].encode() in res.content
+
+
+@pytest.mark.asyncio
+async def test_preview_404(
+    client: httpx.AsyncClient, auth_token: str,
+) -> None:
+    res = await client.get(
+        f"{_BASE}/tokens/00000000-0000-0000-0000-000000000000/preview",
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 404
+
+
+# ---------------- triggers list ------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_triggers_list_for_token(
+    client: httpx.AsyncClient, auth_token: str, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "https://canary.test")
+    with _patch_subprocess(rc=0):
+        created = (await client.post(
+            f"{_BASE}/tokens",
+            json={
+                "decky_name": "web1", "kind": "http",
+                "placement_path": "/etc/x.env", "generator": "env_file",
+            },
+            headers=_hdr(auth_token),
+        )).json()
+    # No triggers yet.
+    res = await client.get(
+        f"{_BASE}/tokens/{created['uuid']}/triggers",
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 200
+    assert res.json()["total"] == 0
+    # 404 for a missing token.
+    res = await client.get(
+        f"{_BASE}/tokens/00000000-0000-0000-0000-000000000000/triggers",
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 404
+
+
+# ---------------- auth ----------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_unauthenticated_writes_rejected(
+    client: httpx.AsyncClient,
+) -> None:
+    for path, method in [
+        (f"{_BASE}/tokens", "POST"),
+        (f"{_BASE}/blobs", "POST"),
+    ]:
+        res = await client.request(
+            method, path, json={}, files={} if method == "POST" else None,
+        )
+        # Either 401 from the auth dep or 422 from missing body — the
+        # important property is "not anonymous".
+        assert res.status_code in (401, 403, 422), f"{path} {method} -> {res.status_code}"