From 07dc898b0d2c1807292af2bbd8331de0500b43a9 Mon Sep 17 00:00:00 2001 From: Lothnic Date: Mon, 3 Aug 2026 15:52:11 +0530 Subject: [PATCH] feat(control): enforce node cert identity on the control session (P0) Add require_mtls: post-enrollment routes (sessions, heartbeat, desired-state, observations) now verify a forwarded node cert whose SAN node_id matches the path node_id, re-checking the CA chain and validity. NodeCA gains verify_cert + node_id_from_cert. Default off for dev/private networks; the edge terminates mTLS and forwards the PEM. Closes the control-plane half of the mTLS gate (DESIGN.md 15.6). Co-Authored-By: Claude Opus 4.8 --- meridian_control/ca.py | 28 ++++++++ meridian_control/config.py | 8 +++ meridian_control/routes.py | 20 ++++-- meridian_control/service.py | 24 ++++++- tests/control/test_mtls.py | 138 ++++++++++++++++++++++++++++++++++++ 5 files changed, 213 insertions(+), 5 deletions(-) create mode 100644 tests/control/test_mtls.py diff --git a/meridian_control/ca.py b/meridian_control/ca.py index 1aba372..c301c2e 100644 --- a/meridian_control/ca.py +++ b/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") diff --git a/meridian_control/config.py b/meridian_control/config.py index 637227a..a879d4f 100644 --- a/meridian_control/config.py +++ b/meridian_control/config.py @@ -14,6 +14,12 @@ 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": @@ -21,4 +27,6 @@ def from_env(cls) -> "ControlConfig": 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), ) diff --git a/meridian_control/routes.py b/meridian_control/routes.py index 0b7a63e..b316a8e 100644 --- a/meridian_control/routes.py +++ b/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 ----------------------------------------------------- diff --git a/meridian_control/service.py b/meridian_control/service.py index 560ab2b..e74838f 100644 --- a/meridian_control/service.py +++ b/meridian_control/service.py @@ -13,6 +13,7 @@ import hashlib import json import secrets +import urllib.parse from datetime import datetime, timedelta, timezone from typing import Callable, Optional, cast @@ -20,7 +21,7 @@ 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: diff --git a/tests/control/test_mtls.py b/tests/control/test_mtls.py new file mode 100644 index 0000000..74f813e --- /dev/null +++ b/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_")