Control hardening: enforce revocation everywhere + CRL endpoint

meridian_control/README.md

@@ -42,6 +42,7 @@ meridian-control mint-token --auto-approve
 | `POST /admin/tokens` · `/admin/claims/{id}/approve` · `/admin/nodes/{id}/desired` · `/admin/nodes/{id}/stop-authorize` · `/admin/nodes/{id}/revoke` · `/admin/restore` | Operator actions |
 | `GET /admin/projection` | Serving projection (gateway consumes read-only) |
 | `POST /admin/place` | Capacity-aware placement: pick a node/device with enough allocatable VRAM |
+| `GET /admin/crl` | Certificate revocation list (revoked node ids + cert serials) for the edge |
 
 ## Restore safety
 

meridian_control/routes.py

@@ -130,6 +130,11 @@ def projection(request: Request):
     return {"endpoints": _svc(request).serving_projection()}
 
 
+@router.get("/admin/crl")
+def crl(request: Request):
+    return {"revoked": _svc(request).crl()}
+
+
 @router.post("/admin/place")
 def place(request: Request, body: dict = Body(...)):
     placement = _svc(request).select_placement(int(body.get("required_vram_bytes", 0)))

meridian_control/service.py

@@ -337,11 +337,20 @@ def heartbeat(self, node_id: str, request: dict) -> dict:
                 "stop_authorized_engine_ids": sorted(stop_ids),
             }
 
+    def _active_node(self, s, node_id: str) -> Node:
+        """Fetch a node, rejecting unknown and revoked ones. Revocation is thus
+        enforced on every post-enrollment call, independent of the mTLS gate."""
+        node = s.get(Node, node_id)
+        if node is None:
+            raise ControlServiceError("NODE_NOT_FOUND", "unknown node", http_status=404)
+        if node.revoked:
+            raise ControlServiceError("NODE_NOT_AUTHORIZED", "node certificate revoked", http_status=403)
+        return cast(Node, node)
+
     # --- desired state / observations ----------------------------------
     def fetch_desired_state(self, node_id: str, generation: int) -> dict:
         with self._sf() as s:
-            if s.get(Node, node_id) is None:
-                raise ControlServiceError("NODE_NOT_FOUND", "unknown node", http_status=404)
+            self._active_node(s, node_id)
             row = s.scalar(
                 select(DesiredSnapshotRow).where(DesiredSnapshotRow.node_id == node_id)
                 .order_by(DesiredSnapshotRow.generation.desc())
@@ -352,8 +361,7 @@ def fetch_desired_state(self, node_id: str, generation: int) -> dict:
 
     def post_observation(self, node_id: str, observation: dict) -> dict:
         with self._sf() as s:
-            if s.get(Node, node_id) is None:
-                raise ControlServiceError("NODE_NOT_FOUND", "unknown node", http_status=404)
+            self._active_node(s, node_id)
             seq = int(observation.get("sequence", 0))
             row = s.get(ObservationRow, node_id)
             if row is None:
@@ -387,6 +395,23 @@ def select_placement(self, required_vram_bytes: int) -> Optional[dict]:
             return None
         return {"node_id": best[1], "device_id": best[2], "allocatable_vram_bytes": best[0]}
 
+    def crl(self) -> list[dict]:
+        """Certificate revocation list: revoked nodes with their cert serial, for
+        the TLS-terminating edge to reject at the handshake (DESIGN.md 15.6)."""
+        from cryptography import x509
+
+        out: list[dict] = []
+        with self._sf() as s:
+            for node in s.scalars(select(Node).where(Node.revoked.is_(True))):
+                serial: Optional[str] = None
+                if node.certificate_pem:
+                    try:
+                        serial = str(x509.load_pem_x509_certificate(node.certificate_pem.encode()).serial_number)
+                    except ValueError:
+                        serial = None
+                out.append({"node_id": node.node_id, "serial": serial})
+        return out
+
     # --- read models (operator / gateway projection) -------------------
     def serving_projection(self) -> list[dict]:
         """Read-only projection the gateway consumes (DESIGN.md 17). Routable

tests/control/test_service.py

@@ -125,6 +125,28 @@ def test_revoked_node_cannot_heartbeat(tmp_path, clock, node_key):
     assert exc.value.code == "NODE_NOT_AUTHORIZED"
 
 
+def test_revocation_enforced_on_desired_and_observation(tmp_path, clock, node_key):
+    svc = make_service(tmp_path, clock)
+    node_id = _enroll_auto(svc, node_key)["node_id"]
+    svc.set_desired(node_id, {"node_id": node_id, "generation": 1, "snapshot_hash": "sha256:x", "assignments": []})
+    svc.revoke_node(node_id)
+    for call in (lambda: svc.fetch_desired_state(node_id, 1),
+                 lambda: svc.post_observation(node_id, {"sequence": 1, "engines": []})):
+        with pytest.raises(ControlServiceError) as exc:
+            call()
+        assert exc.value.code == "NODE_NOT_AUTHORIZED"
+
+
+def test_crl_lists_revoked_nodes_with_serial(tmp_path, clock, node_key):
+    svc = make_service(tmp_path, clock)
+    node_id = _enroll_auto(svc, node_key)["node_id"]
+    assert svc.crl() == []
+    svc.revoke_node(node_id)
+    crl = svc.crl()
+    assert len(crl) == 1 and crl[0]["node_id"] == node_id
+    assert crl[0]["serial"] and crl[0]["serial"].isdigit()
+
+
 def test_desired_state_generation_and_fetch(tmp_path, clock, node_key):
     svc = make_service(tmp_path, clock)
     node_id = _enroll_auto(svc, node_key)["node_id"]