P0: enforce node cert identity on the control session

meridian_control/ca.py

@@ -81,3 +81,31 @@ def issue_node_cert(self, node_id: str, node_public_key: bytes, lifetime_hours:
 
     def trust_bundle(self) -> str:
         return self._cert.public_bytes(serialization.Encoding.PEM).decode()
+
+    def verify_cert(self, cert_pem: str) -> x509.Certificate:
+        """Verify a presented node cert chains to this CA and is currently valid.
+        Raises ValueError on any failure. Defense-in-depth behind the edge that
+        already terminated mTLS (DESIGN.md 15.6)."""
+        try:
+            cert = x509.load_pem_x509_certificate(cert_pem.encode())
+        except ValueError as e:
+            raise ValueError(f"unparseable client certificate: {e}") from e
+        now = dt.datetime.now(dt.timezone.utc)
+        if now < cert.not_valid_before_utc or now > cert.not_valid_after_utc:
+            raise ValueError("client certificate is expired or not yet valid")
+        pub = self._cert.public_key()
+        assert isinstance(pub, Ed25519PublicKey)
+        try:
+            pub.verify(cert.signature, cert.tbs_certificate_bytes)
+        except Exception as e:  # InvalidSignature
+            raise ValueError("client certificate is not signed by this CA") from e
+        return cert
+
+
+def node_id_from_cert(cert: x509.Certificate) -> str:
+    """Extract the node_id from the cert SAN URI `meridian-node://{node_id}`."""
+    san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
+    for uri in san.get_values_for_type(x509.UniformResourceIdentifier):
+        if uri.startswith("meridian-node://"):
+            return uri[len("meridian-node://"):]
+    raise ValueError("client certificate has no meridian-node SAN")

meridian_control/config.py

@@ -14,11 +14,19 @@ class ControlConfig:
     lease_ttl_seconds: int = 30
     heartbeat_interval_seconds: int = 10
     cert_lifetime_hours: int = 24
+    # When true, post-enrollment calls must carry a CA-issued node cert whose SAN
+    # node_id matches the path (DESIGN.md 15.6). The TLS-terminating edge verifies
+    # the chain and forwards the PEM (url-escaped) in `client_cert_header`; the app
+    # must be reachable only through that edge. Default off for dev/private nets.
+    require_mtls: bool = False
+    client_cert_header: str = "x-client-cert"
 
     @classmethod
     def from_env(cls) -> "ControlConfig":
         return cls(
             db_url=os.environ.get("MERIDIAN_CONTROL_DB_URL", cls.db_url),
             ca_dir=Path(os.environ.get("MERIDIAN_CONTROL_CA_DIR", "./ca")),
             lease_ttl_seconds=int(os.environ.get("MERIDIAN_CONTROL_LEASE_TTL", "30")),
+            require_mtls=os.environ.get("MERIDIAN_CONTROL_REQUIRE_MTLS", "").lower() in ("1", "true", "yes"),
+            client_cert_header=os.environ.get("MERIDIAN_CONTROL_CLIENT_CERT_HEADER", cls.client_cert_header),
         )

meridian_control/routes.py

@@ -40,24 +40,36 @@ def get_claim(request: Request, claim_id: str, x_possession_proof: Optional[str]
     return svc.resolve_claim(claim_id, _b64url_decode(x_possession_proof))
 
 
+def _client_cert(request: Request) -> Optional[str]:
+    return request.headers.get(_svc(request)._config.client_cert_header)
+
+
 @router.post("/control/v1/nodes/{node_id}/sessions")
 def establish_session(request: Request, node_id: str, body: dict = Body(...)):
-    return _svc(request).establish_session(node_id, body)
+    svc = _svc(request)
+    svc.verify_node_identity(node_id, _client_cert(request))
+    return svc.establish_session(node_id, body)
 
 
 @router.post("/control/v1/nodes/{node_id}/heartbeat")
 def heartbeat(request: Request, node_id: str, body: dict = Body(...)):
-    return _svc(request).heartbeat(node_id, body)
+    svc = _svc(request)
+    svc.verify_node_identity(node_id, _client_cert(request))
+    return svc.heartbeat(node_id, body)
 
 
 @router.get("/control/v1/nodes/{node_id}/desired-state")
 def desired_state(request: Request, node_id: str, generation: int = Query(0)):
-    return _svc(request).fetch_desired_state(node_id, generation)
+    svc = _svc(request)
+    svc.verify_node_identity(node_id, _client_cert(request))
+    return svc.fetch_desired_state(node_id, generation)
 
 
 @router.post("/control/v1/nodes/{node_id}/observations")
 def observations(request: Request, node_id: str, body: dict = Body(...)):
-    return _svc(request).post_observation(node_id, body)
+    svc = _svc(request)
+    svc.verify_node_identity(node_id, _client_cert(request))
+    return svc.post_observation(node_id, body)
 
 
 # --- operator admin -----------------------------------------------------

meridian_control/service.py

@@ -13,14 +13,15 @@
 import hashlib
 import json
 import secrets
+import urllib.parse
 from datetime import datetime, timedelta, timezone
 from typing import Callable, Optional, cast
 
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 from sqlalchemy import select
 
-from .ca import NodeCA
+from .ca import NodeCA, node_id_from_cert
 from .config import ControlConfig
 from .models import (
     AuditEvent,
@@ -228,6 +229,27 @@ def _approved_response(self, s, node_id: str) -> dict:
             },
         }
 
+    # --- transport auth -------------------------------------------------
+    def verify_node_identity(self, node_id: str, cert_header: Optional[str]) -> None:
+        """Bind the presented node certificate to the path node_id (DESIGN.md 15.6).
+        No-op unless `require_mtls`. The edge terminates mTLS and forwards the
+        url-escaped PEM; here we re-verify the chain and match the SAN node_id."""
+        if not self._config.require_mtls:
+            return
+        if not cert_header:
+            raise ControlServiceError(
+                "NODE_NOT_AUTHORIZED", "client certificate required", http_status=401
+            )
+        try:
+            cert = self._ca.verify_cert(urllib.parse.unquote(cert_header))
+            cert_node_id = node_id_from_cert(cert)
+        except ValueError as e:
+            raise ControlServiceError("NODE_NOT_AUTHORIZED", str(e), http_status=403) from e
+        if cert_node_id != node_id:
+            raise ControlServiceError(
+                "NODE_NOT_AUTHORIZED", "certificate identity does not match node", http_status=403
+            )
+
     # --- session --------------------------------------------------------
     def establish_session(self, node_id: str, request: dict) -> dict:
         with self._sf() as s:

tests/control/test_mtls.py

@@ -0,0 +1,138 @@
+"""mTLS identity binding (DESIGN.md 15.6): post-enrollment calls must carry a
+CA-issued node cert whose SAN node_id matches the path. Covers the server-side
+`verify_node_identity` logic and an end-to-end pass/reject through the real app.
+"""
+
+from __future__ import annotations
+
+import socket
+import threading
+import time
+import urllib.parse
+
+import pytest
+from conftest import MutableClock
+from cryptography.hazmat.primitives import serialization
+
+from meridian_control.app import create_app
+from meridian_control.ca import NodeCA
+from meridian_control.config import ControlConfig
+from meridian_control.db import make_session_factory
+from meridian_control.service import ControlService, ControlServiceError
+
+
+def _mtls_service(tmp_path, require_mtls=True):
+    cfg = ControlConfig(
+        db_url=f"sqlite:///{tmp_path}/control.db", ca_dir=tmp_path / "ca",
+        lease_ttl_seconds=30, require_mtls=require_mtls,
+    )
+    ca = NodeCA.load_or_create(cfg.ca_dir)
+    return ControlService(make_session_factory(cfg.db_url), ca, cfg, now=MutableClock()), ca
+
+
+def _enroll(svc, node_key):
+    token = svc.create_token(auto_approve=True)
+    resp = svc.enroll(token, {"node_public_key": node_key.public_b64url()})
+    return resp["node_id"], resp["certificate"]
+
+
+def test_matching_cert_is_accepted(tmp_path, node_key):
+    svc, _ = _mtls_service(tmp_path)
+    node_id, cert = _enroll(svc, node_key)
+    svc.verify_node_identity(node_id, urllib.parse.quote(cert))  # no raise
+
+
+def test_missing_cert_rejected_when_required(tmp_path, node_key):
+    svc, _ = _mtls_service(tmp_path)
+    node_id, _ = _enroll(svc, node_key)
+    with pytest.raises(ControlServiceError) as exc:
+        svc.verify_node_identity(node_id, None)
+    assert exc.value.code == "NODE_NOT_AUTHORIZED"
+
+
+def test_cert_for_another_node_rejected(tmp_path, node_key):
+    svc, _ = _mtls_service(tmp_path)
+    node_a, cert_a = _enroll(svc, node_key)
+    node_b, _ = _enroll(svc, node_key)
+    assert node_a != node_b
+    with pytest.raises(ControlServiceError) as exc:
+        svc.verify_node_identity(node_b, urllib.parse.quote(cert_a))
+    assert exc.value.code == "NODE_NOT_AUTHORIZED"
+
+
+def test_cert_from_foreign_ca_rejected(tmp_path, node_key):
+    svc, _ = _mtls_service(tmp_path)
+    node_id, _ = _enroll(svc, node_key)
+    foreign = NodeCA.load_or_create(tmp_path / "foreign-ca")
+    raw_pub = node_key.key.public_key().public_bytes(
+        encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw
+    )
+    forged = foreign.issue_node_cert(node_id, raw_pub)
+    with pytest.raises(ControlServiceError) as exc:
+        svc.verify_node_identity(node_id, urllib.parse.quote(forged))
+    assert exc.value.code == "NODE_NOT_AUTHORIZED"
+
+
+def test_disabled_is_noop(tmp_path, node_key):
+    svc, _ = _mtls_service(tmp_path, require_mtls=False)
+    node_id, _ = _enroll(svc, node_key)
+    svc.verify_node_identity(node_id, None)  # no raise even with no cert
+
+
+# --- end-to-end through the real app ------------------------------------
+def _free_port() -> int:
+    s = socket.socket()
+    s.bind(("127.0.0.1", 0))
+    port = s.getsockname()[1]
+    s.close()
+    return port
+
+
+@pytest.fixture
+def live_mtls(tmp_path):
+    import uvicorn
+
+    cfg = ControlConfig(
+        db_url=f"sqlite:///{tmp_path}/x.db", ca_dir=tmp_path / "ca",
+        lease_ttl_seconds=30, require_mtls=True,
+    )
+    app = create_app(cfg)
+    port = _free_port()
+    server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning"))
+    thread = threading.Thread(target=server.run, daemon=True)
+    thread.start()
+    deadline = time.time() + 10
+    while not server.started and time.time() < deadline:
+        time.sleep(0.02)
+    assert server.started
+    try:
+        yield app, f"http://127.0.0.1:{port}"
+    finally:
+        server.should_exit = True
+        thread.join(timeout=5)
+
+
+def test_agent_presents_cert_end_to_end(live_mtls, tmp_path):
+    pytest.importorskip("meridian_node")
+    from meridian_node.agent import Agent
+    from meridian_node.config import Config
+    from meridian_node.control import ControlError
+    from meridian_node.http_transport import HttpTransport
+    from meridian_node.wiring import upgrade_transport
+
+    app, url = live_mtls
+    token = app.state.control_service.create_token(auto_approve=True)
+    cfg = Config(control_plane_url=url, state_dir=tmp_path / "node", mode="observe-only")
+    agent = Agent(cfg, HttpTransport(url, enrollment_token=token))
+    node_id = agent.ensure_enrolled()
+
+    # Without the forwarded cert header the session is rejected.
+    with pytest.raises(ControlError) as exc:
+        agent.establish_session()
+    assert exc.value.code == "NODE_NOT_AUTHORIZED"
+
+    # After upgrading to the cert transport, the same call is authorized.
+    upgrade_transport(cfg, agent)
+    agent.establish_session()
+    assert agent.heartbeat_once()["accepted_sequence"] == 1
+    assert node_id.startswith("node_")