From d2ec9e95b1f42ebf3674d312c317a4ce12c3fee8 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Fri, 25 Sep 2026 21:38:13 +0300 Subject: [PATCH] feat(fleet): replace onboard computer without replacing apparatus --- src/k1link/fleet/registry.py | 41 ++++++-- src/k1link/fleet/replacement.py | 29 ++++++ src/k1link/web/fleet_api.py | 17 ++++ src/k1link/web/rover_api.py | 7 +- tests/fleet/test_replacement.py | 162 ++++++++++++++++++++++++++++++++ 5 files changed, 245 insertions(+), 11 deletions(-) create mode 100644 src/k1link/fleet/replacement.py create mode 100644 tests/fleet/test_replacement.py diff --git a/src/k1link/fleet/registry.py b/src/k1link/fleet/registry.py index f2ee526..f874ede 100644 --- a/src/k1link/fleet/registry.py +++ b/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): diff --git a/src/k1link/fleet/replacement.py b/src/k1link/fleet/replacement.py new file mode 100644 index 0000000..e514282 --- /dev/null +++ b/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("Дождитесь завершения операции с устройствами БК.") diff --git a/src/k1link/web/fleet_api.py b/src/k1link/web/fleet_api.py index 6c353e2..f130981 100644 --- a/src/k1link/web/fleet_api.py +++ b/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)] diff --git a/src/k1link/web/rover_api.py b/src/k1link/web/rover_api.py index edd4988..0e19eef 100644 --- a/src/k1link/web/rover_api.py +++ b/src/k1link/web/rover_api.py @@ -27,7 +27,9 @@ def rover_state(vehicle_id: str, response: Response, def rover_arm(vehicle_id: str, body: dict, fleet: Annotated[FleetRegistry, Depends(local_operator)]): try: - return fleet.rover_control.arm(node(fleet, vehicle_id), body) + # 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: - return fleet.rover_control.command(node(fleet, vehicle_id), body) + 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 diff --git a/tests/fleet/test_replacement.py b/tests/fleet/test_replacement.py new file mode 100644 index 0000000..02ef73d --- /dev/null +++ b/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