feat(fleet): replace onboard computer without replacing apparatus

src/k1link/fleet/registry.py

@@ -52,6 +52,10 @@ class FleetRegistry:
             "CREATE TABLE IF NOT EXISTS vehicles (id TEXT PRIMARY KEY, "
             "node_id TEXT UNIQUE NOT NULL, body TEXT NOT NULL)"
         )
+        self.db.execute(
+            "CREATE TABLE IF NOT EXISTS board_history (vehicle_id TEXT NOT NULL, "
+            "revision INTEGER NOT NULL, body TEXT NOT NULL, PRIMARY KEY(vehicle_id, revision))"
+        )
         self.db.commit()
         from k1link.device_plugins.vesc.archive import Archive
 
@@ -74,11 +78,14 @@ class FleetRegistry:
             json.loads(row[0]) for row in self.db.execute("SELECT body FROM vehicles ORDER BY id")
         ]
 
-    def save(self, row):
+    def save(self, row, *, previous=None):
         with self.db:
+            if previous is not None:
+                self.db.execute("INSERT INTO board_history VALUES(?,?,?)",
+                                (previous["id"], previous["revision"], json.dumps(previous)))
             self.db.execute(
                 "INSERT INTO vehicles VALUES(?,?,?) "
-                "ON CONFLICT(id) DO UPDATE SET body=excluded.body",
+                "ON CONFLICT(id) DO UPDATE SET node_id=excluded.node_id, body=excluded.body",
                 (row["id"], row["node_id"], json.dumps(row)),
             )
         self.events.notify()
@@ -164,17 +171,31 @@ class FleetRegistry:
             "endpoint": invitation["endpoint"],
         }
 
-    def add(self, preview_id: str, name: str, platform: str) -> dict:
+    def add(self, preview_id: str, name: str, platform: str, *,
+            vehicle_id: str | None = None, expected_revision: int | None = None) -> 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"]))
+                current = self.find(preview["created_id"])
+                if (preview.get("target_id") != vehicle_id
+                        or current["binding"]["binding_id"] != preview.get("created_binding")):
+                    raise PairingError("Приглашение уже использовано. Проверьте БК заново.")
+                return self.public(current)
+            target = self.find(vehicle_id) if vehicle_id is not None else None
+            if target is not None:
+                if expected_revision != target["revision"]:
+                    raise PairingError("Привязка БК изменилась. Обновите аппарат и повторите проверку.")
+                from .replacement import check_replacement
+                check_replacement(self, target)
+                name, platform = target["name"], target["platform"]
             invitation = preview["invitation"]
             existing = next(
                 (row for row in self.rows() if row["node_id"] == invitation["node_id"]), None
             )
+            if target is not None and existing is not None and existing["id"] != target["id"]:
+                raise PairingError("Этот БК относится к другому аппарату. Выберите другое приглашение.")
             if existing and existing["enrollment"] in ("pending", "paired"):
                 if existing.get("invitation_id") == invitation["id"]:
                     return self.public(existing)
@@ -201,13 +222,14 @@ class FleetRegistry:
                 "ca_pem": pem(self.trust.ca),
                 "client_pem": self.trust.leaf(invitation["node_id"], preview["public_key"]),
             }
+            previous = target or existing
             row = {
-                "id": existing["id"] if existing else secrets.token_urlsafe(16),
+                "id": previous["id"] if previous 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,
+                "revision": (previous["revision"] + 1) if previous else 1,
                 "binding": binding,
                 "invitation_id": invitation["id"],
                 "invitation": invitation,
@@ -218,10 +240,11 @@ class FleetRegistry:
                 "runtime": None,
                 "certificate_previous": None,
                 "notice": "Подтверждаем привязку с БК",
-                "created_at": time.time(),
+                "created_at": previous["created_at"] if previous else time.time(),
             }
-            self.save(row)
-            self.previews[preview_id] = {"created_id": row["id"], "expires": preview["expires"]}
+            self.save(row, previous=previous)
+            self.previews[preview_id] = {"created_id": row["id"], "created_binding": binding["binding_id"],
+                                         "target_id": vehicle_id, "expires": preview["expires"]}
             return self.public(row)
 
     def public(self, row):

src/k1link/fleet/replacement.py

@@ -0,0 +1,29 @@
+"""Admission for replacing the computer of an existing vehicle, never its identity."""
+import time
+from datetime import datetime
+
+from .trust import PairingError
+
+
+def check_replacement(fleet, row):
+    if row["enrollment"] == "pending":
+        raise PairingError("Дождитесь завершения текущей привязки БК или отзовите её.")
+    control = fleet.rover_control.view(row["node_id"])
+    if control["controlling"] or (control["fresh"] and control["snapshot"].get("state") in
+                                  ("preparing", "ready", "driving", "stopping")):
+        raise PairingError("Сначала завершите управление аппаратом и дождитесь остановки.")
+    fleet.device_enrollment.prune()
+    if any(entry["binding"] == row["binding"]["binding_id"]
+           and entry["public"]["state"] in ("queued", "running")
+           for (vehicle_id, _), entry in fleet.device_enrollment.pending.items()
+           if vehicle_id == row["id"]):
+        raise PairingError("Дождитесь завершения подключения беспроводного устройства.")
+    for receipt in row.get("sensor_commands", {}).values():
+        if receipt.get("state") not in ("queued", "running"):
+            continue
+        try:
+            deadline = datetime.fromisoformat(receipt["command"]["deadline_at"].replace("Z", "+00:00")).timestamp()
+        except (KeyError, ValueError, TypeError):
+            deadline = float("inf")
+        if deadline > time.time():
+            raise PairingError("Дождитесь завершения операции с устройствами БК.")

src/k1link/web/fleet_api.py

@@ -48,6 +48,12 @@ class AddRequest(BaseModel):
     platform: str = Field(pattern="^(ugv|uav|stationary|other)$")
 
 
+class AttachBoardRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    preview_id: str = Field(min_length=43, max_length=43)
+    expected_revision: int = Field(ge=1, strict=True)
+
+
 router = APIRouter(prefix="/api/v1/fleet", tags=["fleet"])
 
 
@@ -165,6 +171,17 @@ def fleet_revoke(vehicle_id: str, fleet: Annotated[FleetRegistry, Depends(local_
         raise HTTPException(404, str(error)) from None
 
 
+@router.post("/{vehicle_id}/computer")
+def fleet_attach_board(vehicle_id: str, body: AttachBoardRequest, response: Response,
+                       fleet: Annotated[FleetRegistry, Depends(local_operator)]):
+    response.headers["Cache-Control"] = "no-store"
+    try:
+        return fleet.add(body.preview_id, "", "", vehicle_id=vehicle_id,
+                         expected_revision=body.expected_revision)
+    except PairingError as error:
+        raise HTTPException(409, str(error)) from None
+
+
 @router.post("/{vehicle_id}/devices/operations")
 def sensor_command(
     vehicle_id: str, body: dict, fleet: Annotated[FleetRegistry, Depends(local_operator)]

src/k1link/web/rover_api.py

@@ -27,6 +27,8 @@ def rover_state(vehicle_id: str, response: Response,
 def rover_arm(vehicle_id: str, body: dict,
               fleet: Annotated[FleetRegistry, Depends(local_operator)]):
     try:
+        # Admission and node resolution must not race a computer replacement.
+        with fleet.lock:
             return fleet.rover_control.arm(node(fleet, vehicle_id), body)
     except (ValueError, PairingError) as error:
         raise HTTPException(409, str(error)) from None
@@ -35,6 +37,7 @@ def rover_arm(vehicle_id: str, body: dict,
 def rover_command(vehicle_id: str, body: dict,
                   fleet: Annotated[FleetRegistry, Depends(local_operator)]):
     try:
+        with fleet.lock:
             return fleet.rover_control.command(node(fleet, vehicle_id), body)
     except (ValueError, PairingError) as error:
         raise HTTPException(409, str(error)) from None

tests/fleet/test_replacement.py

@@ -0,0 +1,162 @@
+"""Replacement uses synthetic identities; no device or network access."""
+import json
+import secrets
+import sqlite3
+import time
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from test_pairing import setup, create, code, cert, heartbeat
+from k1link.fleet import board_layout
+from k1link.fleet.registry import FleetRegistry
+from k1link.fleet.trust import SCHEMA, PairingError, public_id
+from k1link.web.fleet_api import router
+
+
+def candidate(fleet, monkeypatch):
+    key = Ed25519PrivateKey.generate()
+    invite = dict(schema=SCHEMA, node_id=public_id('node_', key.public_key()),
+                  id=secrets.token_urlsafe(32), secret=secrets.token_urlsafe(32),
+                  expires_at=int(time.time())+600, endpoint='https://192.168.20.8:8781')
+    def request(invitation, path, payload):
+        return dict(schema=SCHEMA, node_id=invite['node_id'], name='Replacement board',
+                    host={'os':'Linux'}, receipt=secrets.token_urlsafe(32)), key.public_key(), '192.168.20.5'
+    monkeypatch.setattr('k1link.fleet.registry.node_request', request)
+    return fleet.preview(code(invite))
+
+
+def current(setup):
+    fleet = setup[0]
+    public, _ = create(setup)
+    fleet.advance(public['id'])
+    return fleet.find(public['id'])
+
+
+def attach(fleet, old, preview, **kwargs):
+    return fleet.add(preview['preview_id'], '', '', vehicle_id=old['id'],
+                     expected_revision=kwargs.get('revision', old['revision']))
+
+
+def test_replacement_keeps_vehicle_and_layout_but_not_old_authority(setup, monkeypatch):
+    fleet = setup[0]
+    old = current(setup)
+    board_layout.update(fleet.root, old['id'], 'settings', False)
+    before_layout = board_layout.read(fleet.root, old['id'])
+    preview = candidate(fleet, monkeypatch)
+    new = attach(fleet, old, preview)
+    assert new['id'] == old['id'] and new['name'] == old['name'] and new['platform'] == old['platform']
+    assert new['node_id'] == preview['node_id'] != old['node_id']
+    assert new['revision'] == old['revision']+1 and new['enrollment'] == 'pending'
+    assert new['host'] is None and new['sensor_state']['items'] == []
+    assert board_layout.read(fleet.root, old['id']) == before_layout
+    assert fleet.find(old['id'])['created_at'] == old['created_at']
+    archived = fleet.db.execute('SELECT body FROM board_history WHERE vehicle_id=?', (old['id'],)).fetchone()[0]
+    assert json.loads(archived) == old
+    assert fleet.db.execute('SELECT node_id FROM vehicles').fetchone()[0] == preview['node_id']
+    assert fleet.receive(cert(old), '/v1/node/heartbeat', heartbeat(old))[0] == 410
+    assert attach(fleet, old, preview) == new  # A lost operator response is retried safely.
+    fresh = fleet.find(old['id'])
+    assert fleet.receive(cert(fresh), '/v1/node/heartbeat', heartbeat(fresh))[0] == 200
+    assert fleet.public(fleet.find(old['id']))['connectivity'] == 'online'
+    reopened = FleetRegistry(fleet.root)
+    try:
+        assert reopened.find(old['id'])['node_id'] == preview['node_id']
+        assert reopened.db.execute('SELECT count(*) FROM board_history').fetchone()[0] == 1
+    finally:
+        reopened.close()
+
+
+def test_stale_and_expired_preview_do_not_touch_existing_binding(setup, monkeypatch):
+    fleet = setup[0]
+    old = current(setup)
+    preview = candidate(fleet, monkeypatch)
+    with pytest.raises(PairingError, match='изменилась'):
+        attach(fleet, old, preview, revision=old['revision']+1)
+    fleet.previews[preview['preview_id']]['expires'] = 0
+    with pytest.raises(PairingError, match='истекла'):
+        attach(fleet, old, preview)
+    assert fleet.find(old['id']) == old
+    assert fleet.db.execute('SELECT count(*) FROM board_history').fetchone()[0] == 0
+
+
+def test_other_apparatus_computer_is_never_silently_transferred(setup, monkeypatch):
+    fleet = setup[0]
+    old = current(setup)
+    preview = candidate(fleet, monkeypatch)
+    verified = dict(fleet.previews[preview['preview_id']])
+    other = fleet.add(preview['preview_id'], 'Other apparatus', 'ugv')
+    with pytest.raises(PairingError):
+        attach(fleet, old, preview)
+    # Another verified preview of the same computer cannot transfer its owner.
+    preview['preview_id'] = secrets.token_urlsafe(32)
+    fleet.previews[preview['preview_id']] = verified
+    with pytest.raises(PairingError, match='другому аппарату'):
+        attach(fleet, old, preview)
+    assert fleet.find(old['id']) == old
+    assert len(fleet.rows()) == 2
+
+
+@pytest.mark.parametrize('activity', ['pairing', 'control', 'device', 'wireless'])
+def test_replacement_rejects_active_work(setup, monkeypatch, activity):
+    fleet = setup[0]
+    old = current(setup)
+    if activity == 'pairing':
+        old['enrollment'] = 'pending'
+    elif activity == 'control':
+        fleet.rover_control.exchange(old['node_id'], {'relay_id':'a'*32, 'rover':{'instance':'test','supported':True,'state':'idle'}})
+        fleet.rover_control.arm(old['node_id'], {'standstill_confirmed':True,'current_a':5,'max_erpm':1000})
+    elif activity == 'device':
+        old['sensor_commands'] = {'synthetic':{'state':'running','command':{'deadline_at':(datetime.now(UTC)+timedelta(seconds=60)).isoformat()}}}
+    else:
+        fleet.device_enrollment.pending[(old['id'],'synthetic')] = dict(binding=old['binding']['binding_id'], created=time.time(),deadline=time.time()+60,payload={},public={'state':'running'})
+    fleet.save(old)
+    preview = candidate(fleet, monkeypatch)
+    with pytest.raises(PairingError):
+        attach(fleet, old, preview)
+    assert fleet.find(old['id']) == old
+
+
+def test_failed_transaction_preserves_old_binding_and_history(setup, monkeypatch):
+    fleet = setup[0]
+    old = current(setup)
+    preview = candidate(fleet, monkeypatch)
+    fleet.db.execute("CREATE TRIGGER fail_replace BEFORE UPDATE ON vehicles BEGIN SELECT RAISE(ABORT, 'synthetic failure'); END")
+    with pytest.raises(sqlite3.IntegrityError):
+        attach(fleet, old, preview)
+    assert fleet.find(old['id']) == old
+    assert fleet.db.execute('SELECT count(*) FROM board_history').fetchone()[0] == 0
+
+
+def test_listener_failure_keeps_original_and_preview_usable(setup, monkeypatch):
+    fleet = setup[0]
+    old = current(setup)
+    preview = candidate(fleet, monkeypatch)
+    def fail(_):
+        raise OSError('synthetic')
+    monkeypatch.setattr(fleet, 'listen', fail)
+    with pytest.raises(PairingError, match='частный канал'):
+        attach(fleet, old, preview)
+    assert fleet.find(old['id']) == old
+    monkeypatch.setattr(fleet, 'listen', lambda _:None)
+    assert attach(fleet, old, preview)['id'] == old['id']
+
+
+def test_attach_endpoint_scoped_local_revision_and_payload(setup, monkeypatch):
+    fleet = setup[0]
+    old = current(setup)
+    preview = candidate(fleet, monkeypatch)
+    app = FastAPI(); app.state.fleet = fleet; app.include_router(router)
+    body = {'preview_id':preview['preview_id'],'expected_revision':old['revision']}
+    with TestClient(app, base_url='http://127.0.0.1:8000', client=('127.0.0.1',55555)) as client:
+        path = '/api/v1/fleet/'+old['id']+'/computer'
+        assert client.post(path, json=body, headers={'Origin':'https://attacker.test'}).status_code == 403
+        assert client.post(path, json={**body,'name':'injected'}).status_code == 422
+        assert client.post(path, json={**body,'expected_revision':0}).status_code == 422
+        response = client.post(path, json=body)
+        assert response.status_code == 200
+        assert response.json()['id'] == old['id']
+        assert old['binding']['binding_id'] not in response.text