feat(node): pair onboard computers with the Core fleet through UI

src/k1link/fleet/__init__.py

@@ -0,0 +1 @@
+"""Core-owned vehicle registry and private Node enrollment transport."""

src/k1link/fleet/registry.py

@@ -0,0 +1,399 @@
+from __future__ import annotations
+
+import json
+import secrets
+import sqlite3
+import threading
+import time
+from contextlib import suppress
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+from cryptography import x509
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+from missioncore_plugin_sdk.v0alpha2.identity import ExecutionBinding
+
+from .trust import SCHEMA, CoreTrust, PairingError, node_request, parse_invitation, pem, public_id
+
+
+class FleetRegistry:
+    """One writer lock and durable transactions; transport never owns vehicles."""
+
+    def __init__(self, root: Path):
+        root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        if root.is_symlink() or root.stat().st_mode & 0o077:
+            raise PairingError("Хранилище реестра требует закрытый каталог.")
+        self.started_at = time.time()
+        self.root = root
+        self.lock = threading.RLock()
+        self.trust = CoreTrust(root)
+        path = root / "fleet.sqlite3"
+        if path.is_symlink():
+            raise PairingError("Конфликт файла реестра.")
+        self.db = sqlite3.connect(path, check_same_thread=False)
+        path.chmod(0o600)
+        self.db.execute("PRAGMA journal_mode=WAL")
+        self.db.execute("PRAGMA synchronous=FULL")
+        self.db.execute("PRAGMA secure_delete=ON")
+        self.db.execute(
+            "CREATE TABLE IF NOT EXISTS vehicles (id TEXT PRIMARY KEY, "
+            "node_id TEXT UNIQUE NOT NULL, body TEXT NOT NULL)"
+        )
+        self.db.commit()
+        self.previews: dict[str, dict] = {}
+        self.listeners: dict[str, object] = {}
+        self.stop = threading.Event()
+        self.worker: threading.Thread | None = None
+
+    def rows(self):
+        return [
+            json.loads(row[0]) for row in self.db.execute("SELECT body FROM vehicles ORDER BY id")
+        ]
+
+    def save(self, row):
+        with self.db:
+            self.db.execute(
+                "INSERT INTO vehicles VALUES(?,?,?) "
+                "ON CONFLICT(id) DO UPDATE SET body=excluded.body",
+                (row["id"], row["node_id"], json.dumps(row)),
+            )
+
+    def find(self, identifier):
+        result = self.db.execute("SELECT body FROM vehicles WHERE id=?", (identifier,)).fetchone()
+        if not result:
+            raise PairingError("Аппарат не найден.")
+        return json.loads(result[0])
+
+    def start(self):
+        with self.lock:
+            for row in self.rows():
+                if row["enrollment"] in ("pending", "paired"):
+                    # Address absent: keep the binding and show offline.
+                    with suppress(OSError):
+                        self.listen(row["core_address"])
+            self.worker = threading.Thread(
+                target=self.reconcile, name="mission-core-fleet", daemon=True
+            )
+            self.worker.start()
+
+    def close(self):
+        self.stop.set()
+        if self.worker:
+            self.worker.join(timeout=20)
+        for server in self.listeners.values():
+            server.shutdown()
+            server.server_close()
+        with self.lock:
+            self.db.close()
+
+    def listen(self, address):
+        from .transport import NodeChannelServer
+
+        if address not in self.listeners:
+            server = NodeChannelServer(address, self)
+            self.listeners[address] = server
+            threading.Thread(
+                target=server.serve_forever, name="mission-core-node-channel", daemon=True
+            ).start()
+
+    def preview(self, code: str) -> dict:
+        invitation = parse_invitation(code.strip())
+        value, public_key, address = node_request(
+            invitation, "/v1/pair/inspect", {"id": invitation["id"], "secret": invitation["secret"]}
+        )
+        if value.get("schema") != SCHEMA or value.get("node_id") != invitation["node_id"]:
+            raise PairingError("БК вернул несовместимое подтверждение.")
+        with self.lock:
+            now = time.time()
+            self.previews = {
+                key: item for key, item in self.previews.items() if item["expires"] > now
+            }
+            if len(self.previews) >= 16:
+                raise PairingError("Завершите или закройте предыдущие приглашения.")
+            identifier = secrets.token_urlsafe(32)
+            self.previews[identifier] = {
+                "invitation": invitation,
+                "expires": min(now + 300, invitation["expires_at"]),
+                "public_key": public_key,
+                "core_address": address,
+                "node": value,
+            }
+        return {
+            "preview_id": identifier,
+            "node_id": invitation["node_id"],
+            "name": value["name"],
+            "host": value["host"],
+            "expires_at": invitation["expires_at"],
+            "endpoint": invitation["endpoint"],
+        }
+
+    def add(self, preview_id: str, name: str, platform: str) -> dict:
+        with self.lock:
+            preview = self.previews.get(preview_id)
+            if preview is None or preview["expires"] <= time.time():
+                raise PairingError("Проверка приглашения истекла. Вставьте код снова.")
+            if preview.get("created_id"):
+                return self.public(self.find(preview["created_id"]))
+            invitation = preview["invitation"]
+            existing = next(
+                (row for row in self.rows() if row["node_id"] == invitation["node_id"]), None
+            )
+            if existing and existing["enrollment"] in ("pending", "paired"):
+                if existing.get("invitation_id") == invitation["id"]:
+                    return self.public(existing)
+                raise PairingError("Этот БК уже добавлен. Сначала отзовите прежнюю привязку.")
+            if (
+                not name.strip()
+                or len(name.strip()) > 80
+                or any(ord(c) < 32 for c in name)
+                or platform not in ("ugv", "uav", "stationary", "other")
+            ):
+                raise PairingError("Проверьте название и класс аппарата.")
+            try:
+                self.listen(preview["core_address"])
+            except OSError:
+                raise PairingError(
+                    "Не удалось открыть частный канал Core. "
+                    "Проверьте сеть и доступность порта приложения."
+                ) from None
+            binding = {
+                "binding_id": secrets.token_urlsafe(32),
+                "core_id": self.trust.core_id,
+                "core_name": "Mission Core",
+                "endpoint": f"https://{preview['core_address']}:8782",
+                "ca_pem": pem(self.trust.ca),
+                "client_pem": self.trust.leaf(invitation["node_id"], preview["public_key"]),
+            }
+            row = {
+                "id": existing["id"] if existing else secrets.token_urlsafe(16),
+                "node_id": invitation["node_id"],
+                "name": name.strip(),
+                "platform": platform,
+                "enrollment": "pending",
+                "revision": (existing["revision"] + 1) if existing else 1,
+                "binding": binding,
+                "invitation_id": invitation["id"],
+                "invitation": invitation,
+                "receipt": None,
+                "core_address": preview["core_address"],
+                "last_seen": None,
+                "inventory": None,
+                "runtime": None,
+                "certificate_previous": None,
+                "notice": "Подтверждаем привязку с БК",
+                "created_at": time.time(),
+            }
+            self.save(row)
+            self.previews[preview_id] = {"created_id": row["id"], "expires": preview["expires"]}
+            return self.public(row)
+
+    def public(self, row):
+        online = (
+            row["enrollment"] == "paired"
+            and row["last_seen"] is not None
+            and row["last_seen"] >= self.started_at
+            and time.time() - row["last_seen"] < 20
+        )
+        return {
+            "id": row["id"],
+            "node_id": row["node_id"],
+            "name": row["name"],
+            "platform": row["platform"],
+            "enrollment": row["enrollment"],
+            "revision": row["revision"],
+            "connectivity": "online" if online else "offline",
+            "last_seen": row["last_seen"],
+            "host": row["inventory"],
+            "execution_binding": row["runtime"],
+            "devices": [],
+            "notice": row["notice"],
+            "core_endpoint": row["binding"]["endpoint"],
+        }
+
+    def listing(self):
+        with self.lock:
+            return {
+                "schema": "missioncore.fleet-registry/v1",
+                "core_id": self.trust.core_id,
+                "items": [self.public(row) for row in self.rows()],
+            }
+
+    def revoke(self, identifier: str):
+        with self.lock:
+            row = self.find(identifier)
+            if row["enrollment"] == "revoked":
+                return self.public(row)
+            row.update(
+                enrollment="revoked",
+                revision=row["revision"] + 1,
+                invitation=None,
+                receipt=None,
+                notice="Доверие отозвано",
+            )
+            self.save(row)
+            return self.public(row)
+
+    def advance(self, identifier):
+        # Hold no registry lock across network I/O. Every completion rechecks
+        # the exact binding and enrollment to avoid resurrecting a revoked row.
+        with self.lock:
+            row = self.find(identifier)
+            if row["enrollment"] != "pending":
+                return
+            invitation = row["invitation"]
+            if invitation["expires_at"] <= time.time():
+                row.update(
+                    enrollment="failed",
+                    notice="Приглашение истекло. Создайте новое в Node.",
+                    invitation=None,
+                    receipt=None,
+                )
+                self.save(row)
+                return
+        if row["receipt"] is None:
+            value, _, _ = node_request(
+                invitation,
+                "/v1/pair/offer",
+                {"id": invitation["id"], "secret": invitation["secret"], "binding": row["binding"]},
+            )
+            receipt = value.get("receipt")
+            if (
+                not isinstance(receipt, str)
+                or len(receipt) != 43
+                or value.get("node_id") != row["node_id"]
+            ):
+                raise PairingError("БК не подтвердил ожидаемую привязку.")
+            with self.lock:
+                current = self.find(identifier)
+                if (
+                    current["enrollment"] != "pending"
+                    or current["binding"]["binding_id"] != row["binding"]["binding_id"]
+                ):
+                    return
+                current["receipt"] = receipt
+                self.save(current)
+                row = current
+        node_request(
+            invitation,
+            "/v1/pair/commit",
+            {"id": row["binding"]["binding_id"], "receipt": row["receipt"]},
+        )
+        # An ack proves the binding. Online additionally needs the outgoing
+        # authenticated heartbeat; a lost ack is recovered by that heartbeat.
+        with self.lock:
+            current = self.find(identifier)
+            if (
+                current["enrollment"] == "pending"
+                and current["binding"]["binding_id"] == row["binding"]["binding_id"]
+            ):
+                current.update(
+                    enrollment="paired",
+                    invitation=None,
+                    receipt=None,
+                    notice="Ожидаем исходящее соединение БК",
+                )
+                self.save(current)
+
+    def reconcile(self):
+        while not self.stop.is_set():
+            with self.lock:
+                active = [row for row in self.rows() if row["enrollment"] in ("pending", "paired")]
+                for address in {row["core_address"] for row in active}:
+                    try:
+                        self.listen(address)
+                        server = self.listeners[address]
+                        if server.refresh_at <= time.time():
+                            server.refresh_context()
+                    except OSError:
+                        pass
+                identifiers = [row["id"] for row in active if row["enrollment"] == "pending"]
+                self.previews = {
+                    k: v for k, v in self.previews.items() if v["expires"] > time.time()
+                }
+            for identifier in identifiers:
+                if self.stop.is_set():
+                    break
+                try:
+                    self.advance(identifier)
+                except (PairingError, OSError, ValueError):
+                    with self.lock:
+                        row = self.find(identifier)
+                        if row["enrollment"] == "pending":
+                            row["notice"] = (
+                                "Подтверждение пока не получено. "
+                                "Повторяем до истечения приглашения."
+                            )
+                            self.save(row)
+            self.stop.wait(2)
+
+    def receive(self, certificate: bytes, path: str, value: dict):
+        cert = x509.load_der_x509_certificate(certificate)
+        key = cert.public_key()
+        if not isinstance(key, Ed25519PublicKey):
+            return 403, {"error": "Unauthorized Node"}
+        node_id = public_id("node_", key)
+        with self.lock:
+            row = next((item for item in self.rows() if item["node_id"] == node_id), None)
+            if (
+                row is None
+                or row["enrollment"] not in ("pending", "paired")
+                or value.get("node_id") != node_id
+                or value.get("binding_id") != row["binding"]["binding_id"]
+            ):
+                return 410, {"error": "Binding revoked"}
+            allowed = [row["binding"]["client_pem"]]
+            if (
+                row.get("certificate_previous")
+                and row["certificate_previous"]["until"] > time.time()
+            ):
+                allowed.append(row["certificate_previous"]["pem"])
+            if cert.serial_number not in [
+                x509.load_pem_x509_certificate(item.encode()).serial_number for item in allowed
+            ]:
+                return 403, {"error": "Certificate superseded"}
+            if value.get("schema") != SCHEMA:
+                return 400, {"error": "Incompatible protocol"}
+            if path == "/v1/node/unpair":
+                row.update(
+                    enrollment="revoked",
+                    revision=row["revision"] + 1,
+                    invitation=None,
+                    receipt=None,
+                    notice="БК отозвал привязку",
+                )
+                self.save(row)
+                return 200, {"ok": True}
+            if path != "/v1/node/heartbeat":
+                return 404, {"error": "Unknown operation"}
+            binding = ExecutionBinding.model_validate(value["execution_binding"])
+            if (
+                binding.node_id != node_id
+                or binding.platform.value != "linux"
+                or value.get("devices") != []
+            ):
+                return 400, {"error": "Invalid Node inventory"}
+            host = value.get("host")
+            if (
+                not isinstance(host, dict)
+                or len(host.get("usb", [])) > 256
+                or len(host.get("networks", [])) > 64
+            ):
+                return 400, {"error": "Invalid host inventory"}
+            row.update(
+                enrollment="paired",
+                invitation=None,
+                receipt=None,
+                last_seen=time.time(),
+                inventory=host,
+                runtime=binding.model_dump(mode="json"),
+                notice="",
+            )
+            current = x509.load_pem_x509_certificate(row["binding"]["client_pem"].encode())
+            if current.not_valid_after_utc - datetime.now(UTC) < timedelta(days=7):
+                row["certificate_previous"] = {
+                    "pem": row["binding"]["client_pem"],
+                    "until": time.time() + 3600,
+                }
+                row["binding"]["client_pem"] = self.trust.leaf(node_id, key)
+            self.save(row)
+            return 200, {"ok": True, "client_pem": row["binding"]["client_pem"]}

src/k1link/fleet/transport.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import json
+import ssl
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+from .trust import private_address
+
+
+class NodeChannelServer(ThreadingHTTPServer):
+    daemon_threads = True
+    block_on_close = False
+
+    def __init__(self, address, registry):
+        self.registry = registry
+        self.address = address
+        self.refresh_context()
+        self.capacity = threading.BoundedSemaphore(16)
+        super().__init__((address, 8782), NodeChannelHandler)
+
+    def refresh_context(self):
+        self.context = self.registry.trust.server_context(self.address)
+        self.refresh_at = time.time() + 7 * 86400
+
+    def process_request(self, request, client_address):
+        if not private_address(client_address[0]) or not self.capacity.acquire(blocking=False):
+            request.close()
+            return
+        super().process_request(request, client_address)
+
+    def process_request_thread(self, request, client_address):
+        try:
+            request.settimeout(8)
+            secured = self.context.wrap_socket(request, server_side=True)
+            super().process_request_thread(secured, client_address)
+        except (OSError, ssl.SSLError):
+            request.close()
+        finally:
+            self.capacity.release()
+
+    def handle_error(self, request, client_address):
+        # No tracebacks containing request fields, keys or local addresses.
+        pass
+
+
+class NodeChannelHandler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+
+    def log_message(self, *_args):
+        pass
+
+    def do_POST(self):
+        try:
+            size = int(self.headers.get("Content-Length", "0"))
+            if (
+                self.headers.get("Content-Type") != "application/json"
+                or self.headers.get("Origin")
+                or not 0 < size <= 65536
+            ):
+                raise ValueError
+            value = json.loads(self.rfile.read(size))
+            if not isinstance(value, dict):
+                raise ValueError
+            status, data = self.server.registry.receive(
+                self.connection.getpeercert(binary_form=True), self.path, value
+            )
+        except (ValueError, KeyError, TypeError):
+            self.close_connection = True
+            status, data = 400, {"error": "Invalid request"}
+        encoded = json.dumps(data).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(encoded)))
+        self.send_header("Cache-Control", "no-store")
+        self.end_headers()
+        self.wfile.write(encoded)

src/k1link/fleet/trust.py

@@ -0,0 +1,254 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import http.client
+import ipaddress
+import json
+import os
+import re
+import ssl
+import tempfile
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from urllib.parse import urlsplit
+
+from cryptography import x509
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
+from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID
+
+SCHEMA = "missioncore.node-pairing/v1"
+TOKEN = re.compile(r"^[A-Za-z0-9_-]{43}$")
+NODE_ID = re.compile(r"^node_[a-f0-9]{64}$")
+
+
+class PairingError(ValueError):
+    pass
+
+
+def private_address(value: str) -> bool:
+    try:
+        ip = ipaddress.IPv4Address(value)
+        return any(
+            ip in ipaddress.ip_network(cidr)
+            for cidr in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "100.64.0.0/10")
+        )
+    except ipaddress.AddressValueError:
+        return False
+
+
+def endpoint(value: str, port: int) -> tuple[str, int]:
+    try:
+        u = urlsplit(value)
+        if (
+            u.scheme == "https"
+            and u.username is None
+            and not u.path
+            and not u.query
+            and not u.fragment
+            and u.port == port
+            and private_address(u.hostname or "")
+        ):
+            return u.hostname, port
+    except (ValueError, TypeError):
+        pass
+    raise PairingError("В приглашении указан недопустимый адрес частной сети.")
+
+
+def parse_invitation(code: str) -> dict:
+    try:
+        if not code.startswith("MCN1.") or len(code) > 4096:
+            raise ValueError
+        encoded = code[5:]
+        data = json.loads(
+            base64.b64decode(encoded + "=" * (-len(encoded) % 4), altchars=b"-_", validate=True)
+        )
+        if set(data) != {"schema", "node_id", "id", "endpoint", "expires_at", "secret"}:
+            raise ValueError
+        if (
+            data["schema"] != SCHEMA
+            or not NODE_ID.fullmatch(data["node_id"])
+            or not TOKEN.fullmatch(data["id"])
+            or not TOKEN.fullmatch(data["secret"])
+        ):
+            raise ValueError
+        now = datetime.now(UTC).timestamp()
+        if type(data["expires_at"]) is not int or not now < data["expires_at"] <= now + 630:
+            raise PairingError("Приглашение просрочено. Создайте новое в Node.")
+        endpoint(data["endpoint"], 8781)
+        return data
+    except (ValueError, TypeError, KeyError, UnicodeError) as error:
+        if isinstance(error, PairingError):
+            raise
+        raise PairingError("Код приглашения повреждён или имеет неподдерживаемую версию.") from None
+
+
+def atomic_private(path: Path, data: bytes) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    if path.is_symlink():
+        raise PairingError("Обнаружен конфликт файла доверия.")
+    fd, name = tempfile.mkstemp(prefix=".fleet-", dir=path.parent)
+    try:
+        with os.fdopen(fd, "wb") as f:
+            f.write(data)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(name, path)
+        directory = os.open(path.parent, os.O_RDONLY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+    finally:
+        Path(name).unlink(missing_ok=True)
+
+
+def public_id(prefix: str, key: Ed25519PublicKey) -> str:
+    return prefix + hashlib.sha256(key.public_bytes_raw()).hexdigest()
+
+
+def pem(cert: x509.Certificate) -> str:
+    return cert.public_bytes(serialization.Encoding.PEM).decode()
+
+
+class CoreTrust:
+    def __init__(self, root: Path):
+        self.root = root
+        path = root / "core-identity.json"
+        if path.exists():
+            if path.is_symlink() or path.stat().st_mode & 0o077:
+                raise PairingError("Права хранилища Core должны быть закрытыми.")
+            value = json.loads(path.read_text())
+            self.key = Ed25519PrivateKey.from_private_bytes(base64.b64decode(value["key"]))
+            self.ca = x509.load_pem_x509_certificate(value["ca"].encode())
+            if self.ca.public_key().public_bytes_raw() != self.key.public_key().public_bytes_raw():
+                raise PairingError("Идентичность Core повреждена; автоматическая замена запрещена.")
+        else:
+            self.key = Ed25519PrivateKey.generate()
+            name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Mission Core authority")])
+            self.ca = (
+                self.builder(name, self.key.public_key(), 3650)
+                .issuer_name(name)
+                .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
+                .add_extension(
+                    x509.KeyUsage(True, False, False, False, False, True, True, None, None),
+                    critical=True,
+                )
+                .sign(self.key, None)
+            )
+            atomic_private(
+                path,
+                json.dumps(
+                    {
+                        "key": base64.b64encode(self.key.private_bytes_raw()).decode(),
+                        "ca": pem(self.ca),
+                    }
+                ).encode(),
+            )
+        self.core_id = public_id("core_", self.key.public_key())
+
+    @staticmethod
+    def builder(subject, public_key, days):
+        now = datetime.now(UTC)
+        return (
+            x509.CertificateBuilder()
+            .subject_name(subject)
+            .public_key(public_key)
+            .serial_number(x509.random_serial_number())
+            .not_valid_before(now - timedelta(minutes=1))
+            .not_valid_after(now + timedelta(days=days))
+        )
+
+    def leaf(self, node_id: str, public_key: Ed25519PublicKey) -> str:
+        subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Mission Core Node")])
+        return pem(
+            self.builder(subject, public_key, 30)
+            .issuer_name(self.ca.subject)
+            .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
+            .add_extension(
+                x509.KeyUsage(True, False, False, False, False, False, False, None, None),
+                critical=True,
+            )
+            .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.CLIENT_AUTH]), critical=False)
+            .sign(self.key, None)
+        )
+
+    def server_context(self, address: str) -> ssl.SSLContext:
+        if not private_address(address):
+            raise PairingError("Сервер Core должен использовать частный адрес.")
+        cert = (
+            self.builder(self.ca.subject, self.key.public_key(), 30)
+            .issuer_name(self.ca.subject)
+            .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
+            .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False)
+            .add_extension(
+                x509.SubjectAlternativeName([x509.IPAddress(ipaddress.ip_address(address))]),
+                critical=False,
+            )
+            .sign(self.key, None)
+        )
+        cert_path, key_path = self.root / f"server-{address}.pem", self.root / "server-key.pem"
+        atomic_private(cert_path, (pem(cert) + pem(self.ca)).encode())
+        atomic_private(
+            key_path,
+            self.key.private_bytes(
+                serialization.Encoding.PEM,
+                serialization.PrivateFormat.PKCS8,
+                serialization.NoEncryption(),
+            ),
+        )
+        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        context.minimum_version = ssl.TLSVersion.TLSv1_3
+        context.load_cert_chain(cert_path, key_path)
+        context.verify_mode = ssl.CERT_REQUIRED
+        context.load_verify_locations(cadata=pem(self.ca))
+        return context
+
+
+def node_request(invitation: dict, path: str, payload: dict) -> tuple[dict, Ed25519PublicKey, str]:
+    address, port = endpoint(invitation["endpoint"], 8781)
+    # The one-use secret is sent only AFTER authenticating the Node identity.
+    # No ambient proxy, DNS lookup, redirect or system CA replaces this pin.
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    context.minimum_version = ssl.TLSVersion.TLSv1_3
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+    connection = http.client.HTTPSConnection(address, port, timeout=5, context=context)
+    try:
+        connection.connect()
+        cert = x509.load_der_x509_certificate(connection.sock.getpeercert(binary_form=True))
+        key = cert.public_key()
+        now = datetime.now(UTC)
+        if (
+            not isinstance(key, Ed25519PublicKey)
+            or public_id("node_", key) != invitation["node_id"]
+            or not cert.not_valid_before_utc <= now < cert.not_valid_after_utc
+        ):
+            raise PairingError("Идентичность БК не совпала с приглашением.")
+        key.verify(cert.signature, cert.tbs_certificate_bytes)
+        local = connection.sock.getsockname()[0]
+        if not private_address(local):
+            raise PairingError("Core не получил частный обратный адрес.")
+        connection.request("POST", path, json.dumps(payload), {"Content-Type": "application/json"})
+        response = connection.getresponse()
+        if response.status != 200:
+            raise PairingError(
+                "Node отклонил запрос: приглашение истекло, отменено или уже использовано."
+                if response.status in (409, 410)
+                else "Node не смог подтвердить привязку."
+            )
+        value = json.loads(response.read(65537))
+        if not isinstance(value, dict):
+            raise ValueError("invalid response")
+        return value, key, local
+    except PairingError:
+        raise
+    except Exception:
+        # No invitation, peer-provided body, certificate or OS exception in API/logs.
+        raise PairingError(
+            "БК недоступен или не подтвердил защищённое соединение. "
+            "Проверьте частную сеть и повторите."
+        ) from None
+    finally:
+        connection.close()

src/k1link/web/app.py

@@ -112,6 +112,7 @@ from k1link.simulation.projects import SimulationProjectService, SimulationProje
 from k1link.web.advanced_laboratory_api import build_advanced_laboratory_router
 from k1link.web.artifact_health_api import build_artifact_health_router
 from k1link.web.compute_contour_api import build_compute_contour_router
+from k1link.web.fleet_api import router as fleet_router
 from k1link.web.device_plugin_composition import load_installed_device_plugins
 from k1link.web.e30_engineering_api import build_e30_engineering_router
 from k1link.web.e30_human_review_api import build_e30_human_review_router
@@ -897,7 +898,24 @@ async def _recorded_blueprint_resource_reaper() -> None:
 
 
 @asynccontextmanager
-async def app_lifespan(_: FastAPI) -> AsyncIterator[None]:
+async def app_lifespan(application: FastAPI) -> AsyncIterator[None]:
+    import logging
+    import sqlite3
+
+    from k1link.fleet.registry import FleetRegistry
+
+    fleet = None
+    application.state.fleet = None
+    try:
+        fleet = FleetRegistry(session_store.data_dir / "fleet")
+        fleet.start()
+        application.state.fleet = fleet
+    except (OSError, ValueError, sqlite3.Error):
+        # Fail closed for pairing without taking down unrelated operator work.
+        if fleet is not None:
+            fleet.close()
+            fleet = None
+        logging.getLogger(__name__).error("Fleet trust storage unavailable; pairing disabled")
     reconciler: asyncio.Task[None] | None = None
     publication_reconciler: asyncio.Task[None] | None = None
     blueprint_reaper: asyncio.Task[None] | None = None
@@ -920,6 +938,9 @@ async def app_lifespan(_: FastAPI) -> AsyncIterator[None]:
     finally:
         from k1link.viewer.recorded import recorded_blueprint_sessions
 
+        application.state.fleet = None
+        if fleet is not None:
+            await asyncio.to_thread(fleet.close)
         if blueprint_reaper is not None:
             blueprint_reaper.cancel()
             with suppress(asyncio.CancelledError):
@@ -950,6 +971,9 @@ app = FastAPI(
 app.add_middleware(GZipMiddleware, minimum_size=1_024, compresslevel=5)
 
 
+app.include_router(fleet_router)
+
+
 @app.exception_handler(RequestValidationError)
 async def request_validation_error_handler(
     _: Request,

src/k1link/web/fleet_api.py

@@ -0,0 +1,85 @@
+"""Operator-only fleet admission; the separate private mTLS listener is in fleet."""
+
+from __future__ import annotations
+
+import ipaddress
+from typing import Annotated
+from urllib.parse import urlsplit
+
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from pydantic import BaseModel, ConfigDict, Field
+
+from k1link.fleet.registry import FleetRegistry
+from k1link.fleet.trust import PairingError
+
+
+def local_operator(request: Request) -> FleetRegistry:
+    try:
+        peer = ipaddress.ip_address(request.client.host)
+        host = urlsplit(f"http://{request.headers.get('host', '')}")
+        if not peer.is_loopback or host.hostname not in ("127.0.0.1", "localhost", "::1"):
+            raise ValueError
+        origin = request.headers.get("origin")
+        if origin and origin != f"http://{request.headers['host']}":
+            raise ValueError
+        if request.headers.get("sec-fetch-site") == "cross-site":
+            raise ValueError
+    except (ValueError, AttributeError, KeyError):
+        raise HTTPException(403, "Откройте Mission Core на компьютере оператора.") from None
+    registry = getattr(request.app.state, "fleet", None)
+    if registry is None:
+        raise HTTPException(503, "Реестр аппаратов недоступен. Повторите подключение.")
+    return registry
+
+
+class PreviewRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    code: str = Field(min_length=1, max_length=4096)
+
+
+class AddRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    preview_id: str = Field(min_length=43, max_length=43)
+    name: str = Field(min_length=1, max_length=80)
+    platform: str = Field(pattern="^(ugv|uav|stationary|other)$")
+
+
+router = APIRouter(prefix="/api/v1/fleet", tags=["fleet"])
+
+
+@router.get("")
+def fleet_list(response: Response, fleet: Annotated[FleetRegistry, Depends(local_operator)]):
+    response.headers["Cache-Control"] = "no-store"
+    return fleet.listing()
+
+
+@router.post("/preview")
+def fleet_preview(
+    body: PreviewRequest,
+    response: Response,
+    fleet: Annotated[FleetRegistry, Depends(local_operator)],
+):
+    response.headers["Cache-Control"] = "no-store"
+    try:
+        return fleet.preview(body.code)
+    except PairingError as error:
+        raise HTTPException(409, str(error)) from None
+
+
+@router.post("")
+def fleet_add(
+    body: AddRequest, response: Response, fleet: Annotated[FleetRegistry, Depends(local_operator)]
+):
+    response.headers["Cache-Control"] = "no-store"
+    try:
+        return fleet.add(body.preview_id, body.name, body.platform)
+    except PairingError as error:
+        raise HTTPException(409, str(error)) from None
+
+
+@router.delete("/{vehicle_id}")
+def fleet_revoke(vehicle_id: str, fleet: Annotated[FleetRegistry, Depends(local_operator)]):
+    try:
+        return fleet.revoke(vehicle_id)
+    except PairingError as error:
+        raise HTTPException(404, str(error)) from None