From eddc09008eae36a5d3c9140f53980c82564254ab Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Sat, 18 Jul 2026 15:17:46 +0300 Subject: [PATCH] feat(k1): add physical acceptance transport --- README.md | 11 +- docs/01_IMPLEMENTATION_PLAN.md | 18 +- docs/05_K1_MQTT_STREAM_PROFILE.md | 21 +- docs/06_K1_LIVE_VIEWER.md | 9 +- docs/07_MISSION_CORE_MONOREPO.md | 9 +- docs/09_OBSERVATION_SESSIONS.md | 4 +- ...n-observation-runtime-and-k1-extraction.md | 4 +- .../0012-device-bound-k1-command-authority.md | 27 +- ...IPHONE_LOCAL_PROTOCOL_20260716.redacted.md | 14 + plugins/xgrids-k1/README.md | 9 +- plugins/xgrids-k1/profiles/README.md | 11 +- .../profiles/fw-3.0.2/direct-lan.v1.json | 4 +- src/k1link/device_plugins/xgrids_k1/cli.py | 42 ++ .../protocol/application_acceptance.py | 188 ++++++ .../protocol/application_authority.py | 72 +++ .../xgrids_k1/protocol/application_mqtt.py | 587 ++++++++++++++++++ .../xgrids_k1/protocol/application_publish.py | 8 + tests/test_cli.py | 35 ++ tests/test_xgrids_application_acceptance.py | 243 ++++++++ tests/test_xgrids_application_authority.py | 65 ++ tests/test_xgrids_application_mqtt.py | 225 +++++++ tests/test_xgrids_application_publish.py | 1 + 22 files changed, 1574 insertions(+), 33 deletions(-) create mode 100644 src/k1link/device_plugins/xgrids_k1/protocol/application_acceptance.py create mode 100644 src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py create mode 100644 tests/test_xgrids_application_acceptance.py create mode 100644 tests/test_xgrids_application_mqtt.py diff --git a/README.md b/README.md index dbe0487..0585fc9 100644 --- a/README.md +++ b/README.md @@ -151,9 +151,14 @@ orchestrator passes all retained response barriers. The installed publisher boundary is structurally write-disabled and cannot call its sink. Plugin v0.4.0 wires a dormant coordinator into the facade: explicit shadow arm holds authority for at most 300 seconds and expiry/disarm/reprovision/acquisition/shutdown revoke -it together with the orchestrator. No emission method or live sink exists, no UI -button claims device control, `vendor_writes_enabled` remains false, and no -application command is sent. +it together with the orchestrator. A separate, uninstalled physical-acceptance +transport now reproduces MQTT 3.1.1 `clean_session=false`, keepalive 60, exact +response subscriptions, QoS2 completion and the five response-gated batches. +It consumes operation keys before publish and poisons unknown outcomes without +retry. An interactive Keychain administration command provisions through +Apple's hidden prompt without putting the authority in argv. Neither component +is wired to facade/UI; `vendor_writes_enabled` remains false and no application +command has been sent. This locked bootstrap is repeatable in the current workspace, not yet a standalone release install. The frontend consumes sibling `file:` packages from diff --git a/docs/01_IMPLEMENTATION_PLAN.md b/docs/01_IMPLEMENTATION_PLAN.md index 6ab2fd6..60c1db5 100644 --- a/docs/01_IMPLEMENTATION_PLAN.md +++ b/docs/01_IMPLEMENTATION_PLAN.md @@ -21,7 +21,7 @@ Each gate produces evidence and an explicit GO, PAUSE or BLOCKED result. | Stage 7 observation archive | GO (point/pose/telemetry contract) — durable catalog, recovery, capture-clock-bounded RRD preparation, archived metric time series, saved-session timeline and atomic playback are implemented | | Stage 7 recorded cameras | GO (contract), acceptance pending — acquisition-owned fMP4 archive and player are implemented/tested; one real archived K1 camera plus point-cloud session has not passed playback yet | | Plugin isolation | GO (laboratory control plane) — vendor backend/frontend and optional scene controls are plugin-owned; manifest/runtime descriptor parity, versioned handshake, lifecycle health and transport correlation fail closed while execution remains in-process | -| K1 application control | PAUSE (dormant runtime gate complete) — plugin v0.4.0 exposes operator-confirmed shadow arm/disarm/state; authority is held in a 15–300 s non-exportable lease and revoked on expiry, reprovision, acquisition preparation or shutdown; orchestrator is facade-owned, but no emission method/live sink exists and vendor writes remain false | +| K1 application control | PAUSE (physical-acceptance software gate complete) — plugin v0.4.0 runtime remains dormant/write-disabled; a separate uninstalled transport now enforces the retained MQTT session, exact response subscriptions, QoS2 one-shot operation keys, five response-gated batches and poison-on-unknown behavior; Keychain provisioning and physical START/STOP remain unexecuted | | Stage 8 product storage | PAUSE — retention, replication, encryption, capacity monitoring and long-run browser/WASM stress remain deployment gates | USB project copying remains optional ground truth rather than a blocker for the @@ -46,9 +46,15 @@ is now wired into the XGRIDS facade and plugin manifest. Explicit operator arm loads the Keychain value into a bounded 15–300 second process-memory lease; expiry, disarm, reprovision, acquisition preparation and shutdown drop both the lease and orchestrator. Public state contains only lease timing/booleans. There -is deliberately no batch-emission method, live MQTT sink or UI button. Keychain -item provisioning, a separately reviewed write-capable type and durable -post-stop save remain unproven. The physical button is the known-safe fallback. +is deliberately no batch-emission method, live MQTT sink or UI button in the +installed runtime. A separate physical-acceptance-only transport and dialogue +executor now implement the reviewed MQTT exchange without facade/UI wiring. +Each operation key is consumed before publish, automatic reconnect and +application retry are forbidden, and any unknown outcome poisons the transport. +The one-time `k1link authority provision` command delegates secret entry to the +macOS Keychain TTY prompt and validates through the production loader without +receiving the value in argv/environment/file/browser state. The command has not +been run. Physical START/STOP remain untested; the button is the safe fallback. The Stage 6 live path uses a bounded raw-first bridge: loss in the visualization queue cannot discard MQTT evidence. The queue holds four preview messages; @@ -267,7 +273,7 @@ useful stream is decoded or structurally identified. - firmware or internal-Linux analysis; - physical end-to-end shared-timeline playback for a newly archived session containing point cloud plus one selected K1 camera; -- operator-owned Keychain item provisioning, separately reviewed emission transport - and stable-artifact proof after the observed stop lifecycle; +- operator-owned Keychain item provisioning, operator-present physical use of + the reviewed acceptance transport and stable-artifact proof after STOP; - long-running large-session WebViewer/WASM memory telemetry; - production retention, replication, encryption and cross-platform packaging. diff --git a/docs/05_K1_MQTT_STREAM_PROFILE.md b/docs/05_K1_MQTT_STREAM_PROFILE.md index 86864c3..5e1a19b 100644 --- a/docs/05_K1_MQTT_STREAM_PROFILE.md +++ b/docs/05_K1_MQTT_STREAM_PROFILE.md @@ -211,7 +211,26 @@ operator confirmation, a connected/attested K1, idle runtime and no active acquisition. Its non-exportable process-memory lease is limited to 15–300 seconds and is revoked by expiry, disarm, reprovision, acquisition preparation or shutdown. The current one-shot boundary is structurally write-disabled, -exposes no emission method and cannot touch its injected sink. The complete +exposes no emission method and cannot touch its injected sink. + +A separate physical-acceptance-only implementation now exists outside the +facade/runtime. Retained-PCAP re-audit fixes its MQTT contract at 3.1.1, +`clean_session=false`, keepalive 60, the three ordered 9/5/42-topic control +subscription groups, separation from the point-cloud client, QoS 2 +request/PUBCOMP completion and +`lixel/application/response/modeling`. It performs one connection attempt, +consumes every ordinal/action operation key before publish, never reconnects or +retries, rejects stale/duplicate/unexpected responses and poisons every unknown +post-publish outcome. A 15–120 second permit requires explicit operator-present, +owner-controlled-device, LixelGO-closed, battery/storage and expected-state +confirmations. START additionally requires the complete five-batch bootstrap; +STOP requires its own separate permit. This code is not installed as a plugin +action and has never emitted to the K1. + +The one-time Keychain administration command uses `security -w` as the final +argument so Apple's own TTY prompt receives the authority. Mission Core never +accepts it through CLI argv and validates the resulting fixed item through the +production loader. The command has not yet been run. The complete stop/save sequence has not physically proved a durable vendor project. A correlated STOP success, stream quiescence, return to `READY`/steady-green hardware state and appearance of a reusable native project diff --git a/docs/06_K1_LIVE_VIEWER.md b/docs/06_K1_LIVE_VIEWER.md index 813030d..c4bf48c 100644 --- a/docs/06_K1_LIVE_VIEWER.md +++ b/docs/06_K1_LIVE_VIEWER.md @@ -150,8 +150,9 @@ publishing anything. One private application-level OpenAPI authority is kept separate from the transient vendor ID/serial returned by the BLE-selected K1. The fixed Keychain loader, bounded execution lease and dormant facade orchestrator now exist. No emission method or live MQTT sink is installed, and -durable save completion after stop remains open, so `vendor_writes_enabled` is -false and the physical-button workflow is canonical. +an acceptance-only transport exists outside the facade. Keychain provisioning, +physical START/STOP and durable save completion remain open, so +`vendor_writes_enabled` is false and the physical-button workflow is canonical. ## Automatic Rerun source and lifecycle @@ -333,8 +334,8 @@ listener and its process memory must be closed unconditionally. separate generic media path. Historical sessions predating that archive have no recoverable video. - Physical double-click remains the K1 scan start/stop control. Any MQTT command - publisher needs reviewed Keychain item provisioning, a new write-capable - emission type, response-gated one-shot behavior and durable-save proof. + promotion needs reviewed Keychain item provisioning, operator-present use of + the uninstalled response-gated transport and durable-save proof. - No terrain map, elevation model, obstacle segmentation, localization fusion, mission planner or vehicle control is implemented by this viewer milestone. - Exact coordinate axes and the scanner-to-vehicle transform remain a mounting diff --git a/docs/07_MISSION_CORE_MONOREPO.md b/docs/07_MISSION_CORE_MONOREPO.md index 1845e75..88e04b5 100644 --- a/docs/07_MISSION_CORE_MONOREPO.md +++ b/docs/07_MISSION_CORE_MONOREPO.md @@ -150,8 +150,9 @@ is one private application-level authority; vendor ID and serial come from the BLE-selected live K1 and are cross-checked against status. No MQTT publisher exists. A fixed Keychain loader, bounded lease and dormant facade orchestrator are present, but there is no emission method or live MQTT sink. Keychain item -provisioning and stable-artifact proof after STOP remain unresolved, so -`vendor_writes_enabled=false`. +provisioning and stable-artifact proof after STOP remain unresolved. A separate +acceptance-only transport is not imported by the facade, so +`vendor_writes_enabled=false` remains authoritative. ## Semantic lifecycle @@ -224,8 +225,8 @@ views. read-only RTSP/H.264 adapter for each target OS, add disk-backed sealed media caching and evolve same-host MSE delivery toward an authenticated Edge media plane. Keep the modeling-command publisher disabled until operator-owned - Keychain item provisioning, reviewed emission and durable-save confirmation - gate. + Keychain item provisioning, physical acceptance of the reviewed uninstalled + transport and durable-save confirmation gate. Complex equipment will likely be assembled from separately useful component plugins into configured hardware packs, while standalone equipment remains diff --git a/docs/09_OBSERVATION_SESSIONS.md b/docs/09_OBSERVATION_SESSIONS.md index b3f4210..d604a92 100644 --- a/docs/09_OBSERVATION_SESSIONS.md +++ b/docs/09_OBSERVATION_SESSIONS.md @@ -513,5 +513,5 @@ viewer/WASM payload. That is a packaging optimization item, not a failed gate. No retained physical K1 session contains the new canonical camera archive, so a real point-cloud plus one-camera recorded playback remains an explicit hardware acceptance test. Automated protocol tests also do not authorize K1 modeling -publishing: operator-owned Keychain item provisioning, reviewed emission and -durable save remain separate physical/security gates. +publishing: operator-owned Keychain item provisioning, operator-present +physical acceptance and durable save remain separate physical/security gates. diff --git a/docs/adr/0009-device-plugin-observation-runtime-and-k1-extraction.md b/docs/adr/0009-device-plugin-observation-runtime-and-k1-extraction.md index 23f9428..6b1c8ad 100644 --- a/docs/adr/0009-device-plugin-observation-runtime-and-k1-extraction.md +++ b/docs/adr/0009-device-plugin-observation-runtime-and-k1-extraction.md @@ -114,8 +114,8 @@ The operator project name is normalized and validated by the K1 contribution, stored as display/catalog metadata and never used as a path component. This ADR does not claim that it reaches the scanner: the inert application-control codec has no publisher, and automatic K1 writes remain disabled pending reviewed -operator-owned Keychain item provisioning, reviewed emission and durable -post-stop save evidence. +operator-owned Keychain item provisioning, physical acceptance of the +uninstalled transport and durable post-stop save evidence. ADR 0011 subsequently places the action control plane behind a versioned descriptor/handshake/health transport seam. Observation discovery and export diff --git a/docs/adr/0012-device-bound-k1-command-authority.md b/docs/adr/0012-device-bound-k1-command-authority.md index 1d0b235..3df877e 100644 --- a/docs/adr/0012-device-bound-k1-command-authority.md +++ b/docs/adr/0012-device-bound-k1-command-authority.md @@ -103,10 +103,33 @@ shadow arm/disarm/state actions and redacted snapshots. The coordinator has no request-emission method and the installed sink is explicitly uninstalled; this runtime wiring does not widen write authority. +`application_mqtt.py` and `application_acceptance.py` provide the next, still +uninstalled physical-acceptance boundary. The MQTT type repeats the retained +3.1.1 session (`clean_session=false`, keepalive 60), subscribes only to exact +known response topics, admits only reviewed request topics and QoS2/retain-false +envelopes, and consumes explicit bootstrap ordinals/modeling actions before the +first publish call. It performs one connect attempt with no reconnect. PUBCOMP +and every required application response are separate barriers; stale, +duplicate, unexpected, oversized or timed-out responses poison the transport. +An unknown post-publish outcome can only be reconciled physically/status-wise; +the same operation key can never be attempted again by that transport. + +The acceptance executor requires a 15–120 second single-action permit backed by +five explicit operator confirmations. START cannot skip the five recovered +bootstrap batches; STOP uses a separate permit. These types are not imported by +the facade, declared as plugin actions or represented by UI controls. + +`MacOSKeychainApplicationAuthorityProvisioner` invokes Apple's interactive +`security add-generic-password ... -w` prompt with `-w` last. The private value +therefore never enters Mission Core argv, environment, files, logs or browser +state, and the resulting item is immediately validated by the fixed loader. +The repository CLI exposes this as an explicit one-time admin command, but the +command has not been executed in this milestone. + ## Promotion gate -A future publisher requires separate review and an operator-present physical -acceptance: +Promotion now requires operator-present physical acceptance of the uninstalled +transport: 1. provision the reviewed private authority into the fixed local Keychain item; 2. confirm battery/storage and select one K1 over BLE; diff --git a/docs/lab/002_LIXELGO_IPHONE_LOCAL_PROTOCOL_20260716.redacted.md b/docs/lab/002_LIXELGO_IPHONE_LOCAL_PROTOCOL_20260716.redacted.md index 0b1808b..9dae49a 100644 --- a/docs/lab/002_LIXELGO_IPHONE_LOCAL_PROTOCOL_20260716.redacted.md +++ b/docs/lab/002_LIXELGO_IPHONE_LOCAL_PROTOCOL_20260716.redacted.md @@ -65,6 +65,20 @@ Observed K1 transports: No LixelGO UDP flow was observed. +The retained clean-cycle control connection was subsequently re-audited without +capturing new traffic. MQTT CONNECT used protocol level 4 / MQTT 3.1.1, +`clean_session=false` and keepalive 60. Control and point-cloud subscriptions +used separate 19-byte client IDs; their values remain private. The control +client issued three ordered subscription groups containing 9, 5 and 42 topics +before issuing requests. The separate point-cloud client then issued groups of +1 and 1 topic; their values are the already documented legacy/current point +channels. The control group order, topic names and requested QoS values are now +locked by public synthetic regression without exposing client IDs or payloads. +The modeling acknowledgement topic is +`lixel/application/response/modeling`; request QoS was 2 while the K1 delivered +the observed application response PUBLISH packets at QoS 0. These redacted +facts bound the acceptance transport but do not authorize it. + ## Start and static initialization The application start transaction is now physical wire evidence rather than diff --git a/plugins/xgrids-k1/README.md b/plugins/xgrids-k1/README.md index eee3983..3818053 100644 --- a/plugins/xgrids-k1/README.md +++ b/plugins/xgrids-k1/README.md @@ -83,9 +83,12 @@ uv run python plugins/xgrids-k1/profile_loader.py Plugin v0.4.0 does not widen device authority. Its application-control coordinator has no request-emission method, no live MQTT sink and no UI control; -`vendor_writes_enabled` remains false. Keychain item provisioning and a separate -review of any future write-capable transport are still required before a -physical command can be considered. +`vendor_writes_enabled` remains false. A separate uninstalled acceptance-only +transport now implements exact subscriptions, QoS2 completion, response +barriers and no-retry poisoning. The admin CLI can provision the fixed Keychain +item through Apple's hidden prompt without accepting the value as an argument. +Neither path is imported by the facade; Keychain provisioning and a controlled +physical START/STOP acceptance are still required. The optional owner-controlled iPhone/LixelGO observation tool lives under [`lab/iphone-capture/`](lab/iphone-capture/). It pins `pymobiledevice3` in a diff --git a/plugins/xgrids-k1/profiles/README.md b/plugins/xgrids-k1/profiles/README.md index fddbc22..0d3f9ed 100644 --- a/plugins/xgrids-k1/profiles/README.md +++ b/plugins/xgrids-k1/profiles/README.md @@ -40,10 +40,13 @@ literal `{device_id}:ModelingRequest` session relation, retained start settings and numeric success code. Retained client/wire evidence identifies OpenAPI as one private application-level value rather than a per-scanner profile. A fixed read-only macOS Keychain loader, bounded in-memory lease and dormant facade -orchestrator now exist. No request-emission method or live MQTT sink exists; -operator-owned Keychain item provisioning, reviewed emission, durable save -completion, timeout and rollback behavior remain unresolved. Acquisition -therefore stays `operator-manual` through the verified physical double-click. +orchestrator now exist. The installed runtime has no request-emission method or +live MQTT sink. A separate uninstalled acceptance transport implements the +retained MQTT session, exact response subscriptions and poison-on-unknown +one-shot behavior. Operator-owned Keychain item provisioning, physical +acceptance, durable save completion and rollback evidence remain unresolved. +Acquisition therefore stays `operator-manual` through the verified physical +double-click. The standalone encoder models the recovered wire schema, including enum values outside the retained request. It is not an authorization policy: any future publisher must enforce the exact profile mapping (`2/1/0`, omitted diff --git a/plugins/xgrids-k1/profiles/fw-3.0.2/direct-lan.v1.json b/plugins/xgrids-k1/profiles/fw-3.0.2/direct-lan.v1.json index ec3e889..c06f877 100644 --- a/plugins/xgrids-k1/profiles/fw-3.0.2/direct-lan.v1.json +++ b/plugins/xgrids-k1/profiles/fw-3.0.2/direct-lan.v1.json @@ -359,7 +359,7 @@ }, "success_result_code": 302252033, "required_unresolved_context": [ - "operator-owned Keychain item provisioning and reviewed request emission", + "operator-owned Keychain item provisioning and physical acceptance of the uninstalled reviewed transport", "authorization policy for any setting outside the retained request", "timeout, rejection and rollback contract" ], @@ -406,7 +406,7 @@ "request_fields": {}, "success_result_code": 302252033, "required_unresolved_context": [ - "operator-owned Keychain item provisioning and reviewed request emission", + "operator-owned Keychain item provisioning and physical acceptance of the uninstalled reviewed transport", "save-completion and final-standby state mapping", "timeout and rollback contract" ], diff --git a/src/k1link/device_plugins/xgrids_k1/cli.py b/src/k1link/device_plugins/xgrids_k1/cli.py index 297e68a..e8a0ac3 100644 --- a/src/k1link/device_plugins/xgrids_k1/cli.py +++ b/src/k1link/device_plugins/xgrids_k1/cli.py @@ -36,6 +36,10 @@ from k1link.device_plugins.xgrids_k1.mqtt import ( capture_mqtt, ) from k1link.device_plugins.xgrids_k1.net.snapshot import snapshot +from k1link.device_plugins.xgrids_k1.protocol.application_authority import ( + ApplicationAuthorityLoadError, + MacOSKeychainApplicationAuthorityProvisioner, +) from k1link.device_plugins.xgrids_k1.usb.snapshot import snapshot as usb_snapshot from k1link.macos_credentials import CredentialDialogError, prompt_wifi_credentials @@ -49,10 +53,15 @@ ble_app = typer.Typer(help="Bluetooth LE discovery and metadata commands.", no_a net_app = typer.Typer(help="Read-only local network observation commands.", no_args_is_help=True) usb_app = typer.Typer(help="Read-only macOS USB metadata commands.", no_args_is_help=True) analyze_app = typer.Typer(help="Bounded offline evidence analysis commands.", no_args_is_help=True) +authority_app = typer.Typer( + help="One-time local authority administration; never sends a K1 command.", + no_args_is_help=True, +) app.add_typer(ble_app, name="ble") app.add_typer(net_app, name="net") app.add_typer(usb_app, name="usb") app.add_typer(analyze_app, name="analyze") +app.add_typer(authority_app, name="authority") class ToolStatus(TypedDict): @@ -101,6 +110,39 @@ def _command_output(args: list[str]) -> str | None: return result.stdout.strip() +@authority_app.command("provision") +def authority_provision( + confirm_reviewed_authority: Annotated[ + bool, + typer.Option( + "--confirm-reviewed-authority", + help="Confirm this is the reviewed private LixelGO FW 3.0.2 authority.", + ), + ] = False, +) -> None: + """Prompt in macOS Keychain without receiving or printing the authority.""" + + if not confirm_reviewed_authority: + console.print( + "[red]Authority provisioning not confirmed.[/red] " + "Add --confirm-reviewed-authority only for the reviewed private value." + ) + raise typer.Exit(code=2) + console.print( + "macOS Keychain will prompt for the reviewed authority. Mission Core will not " + "receive it through argv, environment, a file, browser state or logs." + ) + try: + source = MacOSKeychainApplicationAuthorityProvisioner().provision_interactively() + except ApplicationAuthorityLoadError as exc: + console.print(f"[red]Authority provisioning failed:[/red] {exc}") + raise typer.Exit(code=2) from exc + console.print( + f"[green]Keychain item validated.[/green] service={source.service!r}; " + f"account={source.account!r}. No K1 command was sent." + ) + + def _default_route_interface() -> str | None: output = _command_output(["route", "-n", "get", "default"]) if output is None: diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_acceptance.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_acceptance.py new file mode 100644 index 0000000..0958ae4 --- /dev/null +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_acceptance.py @@ -0,0 +1,188 @@ +from __future__ import annotations + +import math +import threading +import time +from collections.abc import Callable, Collection, Sequence +from dataclasses import dataclass +from typing import Literal, Protocol + +from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( + LiveDeviceControlBinding, + ShadowApplicationBootstrapOrchestrator, +) +from k1link.device_plugins.xgrids_k1.protocol.application_mqtt import ( + MODELING_RESPONSE_TOPIC, +) +from k1link.device_plugins.xgrids_k1.protocol.application_publish import ( + OneShotPublishEnvelope, +) +from k1link.device_plugins.xgrids_k1.protocol.modeling_control import ( + ModelingAction, + ModelingResponse, + correlate_modeling_response, +) +from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import ( + ShadowModelingCommand, +) + +MIN_ACCEPTANCE_PERMIT_SECONDS = 15.0 +MAX_ACCEPTANCE_PERMIT_SECONDS = 120.0 + + +class ApplicationAcceptanceError(RuntimeError): + """The operator-present physical acceptance contract failed closed.""" + + +class ApplicationBatchExchange(Protocol): + def exchange_batch_once( + self, + envelopes: Sequence[OneShotPublishEnvelope], + *, + required_response_topics: Collection[str], + ) -> dict[str, bytes]: ... + + +@dataclass(frozen=True, slots=True) +class PhysicalAcceptanceChecklist: + action: ModelingAction + operator_present: Literal[True] + owner_controlled_device: Literal[True] + lixelgo_closed: Literal[True] + battery_storage_confirmed: Literal[True] + expected_physical_state_confirmed: Literal[True] + + def __post_init__(self) -> None: + confirmations = ( + self.operator_present, + self.owner_controlled_device, + self.lixelgo_closed, + self.battery_storage_confirmed, + self.expected_physical_state_confirmed, + ) + if any(value is not True for value in confirmations): + raise ApplicationAcceptanceError( + "every physical acceptance confirmation must be explicitly true" + ) + if type(self.action) is not ModelingAction: + raise ApplicationAcceptanceError("acceptance action must be an explicit ModelingAction") + + +class PhysicalAcceptancePermit: + """Short, single-action capability that is consumed before MQTT publish.""" + + def __init__( + self, + checklist: PhysicalAcceptanceChecklist, + *, + ttl_seconds: float = 60.0, + monotonic: Callable[[], float] = time.monotonic, + ) -> None: + if not isinstance(ttl_seconds, (int, float)) or isinstance(ttl_seconds, bool): + raise ApplicationAcceptanceError("acceptance permit TTL must be numeric") + if not math.isfinite(ttl_seconds) or not ( + MIN_ACCEPTANCE_PERMIT_SECONDS <= ttl_seconds <= MAX_ACCEPTANCE_PERMIT_SECONDS + ): + raise ApplicationAcceptanceError( + "acceptance permit TTL is outside the reviewed 15-120 second range" + ) + self._lock = threading.Lock() + self._checklist = checklist + self._monotonic = monotonic + self._expires_at = monotonic() + float(ttl_seconds) + self._consumed = False + + @property + def action(self) -> ModelingAction: + return self._checklist.action + + def consume(self, action: ModelingAction) -> None: + with self._lock: + if self._consumed: + raise ApplicationAcceptanceError("physical acceptance permit was already consumed") + if self._monotonic() >= self._expires_at: + raise ApplicationAcceptanceError("physical acceptance permit expired") + if action is not self._checklist.action: + raise ApplicationAcceptanceError("physical acceptance permit action mismatch") + self._consumed = True + + def snapshot(self) -> dict[str, object]: + with self._lock: + remaining = max(0.0, self._expires_at - self._monotonic()) + return { + "mode": "operator-present-physical-acceptance", + "action": self._checklist.action.name.casefold(), + "consumed": self._consumed, + "remaining_seconds": round(remaining, 3), + "automatic_retry": False, + } + + +class PhysicalAcceptanceDialogueExecutor: + """Drive the recovered barriers and exactly one permitted START or STOP.""" + + def __init__( + self, + transport: ApplicationBatchExchange, + permit: PhysicalAcceptancePermit, + ) -> None: + self._transport = transport + self._permit = permit + self._bootstrap_complete = False + self._command_complete = False + + def run_bootstrap( + self, + orchestrator: ShadowApplicationBootstrapOrchestrator, + ) -> LiveDeviceControlBinding: + if self._bootstrap_complete or self._command_complete: + raise ApplicationAcceptanceError("physical acceptance bootstrap was already attempted") + if self._permit.action is not ModelingAction.START: + raise ApplicationAcceptanceError("bootstrap is admitted only by a START permit") + + while not orchestrator.snapshot().bootstrap_complete: + batch = orchestrator.next_batch() + required_topics = { + request.response_topic for request in batch if request.response_required + } + responses = self._transport.exchange_batch_once( + [OneShotPublishEnvelope.from_bootstrap_request(request) for request in batch], + required_response_topics=required_topics, + ) + for request in batch: + if request.response_required: + orchestrator.accept_response( + request.response_topic, + responses[request.response_topic], + ) + + binding = orchestrator.binding + if binding is None: + raise ApplicationAcceptanceError("bootstrap completed without a live device binding") + self._bootstrap_complete = True + return binding + + def execute_modeling(self, command: ShadowModelingCommand) -> ModelingResponse: + if self._command_complete: + raise ApplicationAcceptanceError("physical acceptance command was already attempted") + if command.action is ModelingAction.START and not self._bootstrap_complete: + raise ApplicationAcceptanceError("START requires the complete response-gated bootstrap") + self._permit.consume(command.action) + self._command_complete = True + responses = self._transport.exchange_batch_once( + [OneShotPublishEnvelope.from_modeling_command(command)], + required_response_topics={MODELING_RESPONSE_TOPIC}, + ) + return correlate_modeling_response( + responses[MODELING_RESPONSE_TOPIC], + command.command, + ) + + def snapshot(self) -> dict[str, object]: + return { + "mode": "physical-acceptance-only", + "bootstrap_complete": self._bootstrap_complete, + "command_complete": self._command_complete, + "permit": self._permit.snapshot(), + "automatic_retry": False, + } diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_authority.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_authority.py index a259e35..cf4d1d8 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/application_authority.py +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_authority.py @@ -3,6 +3,7 @@ from __future__ import annotations import platform import shutil import subprocess +import sys from dataclasses import dataclass from typing import Protocol @@ -14,6 +15,7 @@ from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( KEYCHAIN_SERVICE = "NODEDC Mission Core XGRIDS K1 OpenAPI" KEYCHAIN_ACCOUNT = "lixelgo-application-fw-3.0.2" KEYCHAIN_TIMEOUT_SECONDS = 5.0 +KEYCHAIN_INTERACTIVE_TIMEOUT_SECONDS = 300.0 class ApplicationAuthorityLoadError(RuntimeError): @@ -31,6 +33,16 @@ class CommandRunner(Protocol): ) -> subprocess.CompletedProcess[bytes]: ... +class InteractiveCommandRunner(Protocol): + def __call__( + self, + args: list[str], + *, + check: bool, + timeout: float, + ) -> subprocess.CompletedProcess[bytes]: ... + + @dataclass(frozen=True, slots=True) class ApplicationAuthoritySourceSnapshot: provider: str = "macos-keychain" @@ -116,3 +128,63 @@ class MacOSKeychainApplicationAuthorityLoader: finally: for index in range(len(secret_buffer)): secret_buffer[index] = 0 + + +class MacOSKeychainApplicationAuthorityProvisioner: + """Create or replace the fixed item through Apple's own hidden prompt. + + ``security add-generic-password`` receives ``-w`` as its final option and + prompts on the controlling TTY. The authority therefore never enters a + Python string, subprocess argv, environment variable, file, log or browser + request. Provisioning is an explicit one-time operator/admin action and is + intentionally unavailable without an interactive terminal. + """ + + def __init__( + self, + *, + runner: InteractiveCommandRunner = subprocess.run, + loader: MacOSKeychainApplicationAuthorityLoader | None = None, + ) -> None: + self._runner = runner + self._loader = loader or MacOSKeychainApplicationAuthorityLoader() + + def provision_interactively(self) -> ApplicationAuthoritySourceSnapshot: + if platform.system() != "Darwin": + raise ApplicationAuthorityLoadError( + "application authority provisioning is supported only through macOS Keychain" + ) + if not sys.stdin.isatty() or not sys.stdout.isatty(): + raise ApplicationAuthorityLoadError( + "application authority provisioning requires an interactive terminal" + ) + security = shutil.which("security") + if security != "/usr/bin/security": + raise ApplicationAuthorityLoadError("trusted macOS security binary is unavailable") + + try: + completed = self._runner( + [ + security, + "add-generic-password", + "-U", + "-s", + KEYCHAIN_SERVICE, + "-a", + KEYCHAIN_ACCOUNT, + "-w", + ], + check=False, + timeout=KEYCHAIN_INTERACTIVE_TIMEOUT_SECONDS, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise ApplicationAuthorityLoadError( + "macOS Keychain authority provisioning failed" + ) from exc + if completed.returncode != 0: + raise ApplicationAuthorityLoadError("macOS Keychain authority was not provisioned") + + # Re-open through the production loader so a wrong value fails before + # any Mission Core control lease can be armed. + self._loader.load() + return self._loader.snapshot() diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py new file mode 100644 index 0000000..853787a --- /dev/null +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py @@ -0,0 +1,587 @@ +from __future__ import annotations + +import math +import secrets +import threading +import time +from collections import deque +from collections.abc import Callable, Collection, Sequence +from dataclasses import dataclass + +import paho.mqtt.client as mqtt +from paho.mqtt.enums import CallbackAPIVersion +from paho.mqtt.properties import Properties +from paho.mqtt.reasoncodes import ReasonCode + +from k1link.device_plugins.xgrids_k1.ble.wifi_provisioning import AP_FALLBACK_IPV4 +from k1link.device_plugins.xgrids_k1.mqtt import validate_private_ipv4 +from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( + DEVICE_CONFIG_RESPONSE_TOPIC, + DEVICE_INFO_RESPONSE_TOPIC, + GET_CLOUD_CONFIG_RESPONSE_TOPIC, + GET_NTRIP_PROFILE_RESPONSE_TOPIC, + GET_RTK_ADVANCE_RESPONSE_TOPIC, + MODELING_STATUS_RESPONSE_TOPIC, +) +from k1link.device_plugins.xgrids_k1.protocol.application_publish import ( + OneShotPublishEnvelope, +) +from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import ( + MODELING_REQUEST_TOPIC, +) + +MODELING_RESPONSE_TOPIC = "lixel/application/response/modeling" +CONTROL_KEEPALIVE_SECONDS = 60 +CONTROL_CONNECT_TIMEOUT_SECONDS = 10.0 +CONTROL_EXCHANGE_TIMEOUT_SECONDS = 5.0 +CONTROL_LOOP_INTERVAL_SECONDS = 0.05 +MAX_CONTROL_RESPONSE_BYTES = 64 * 1024 + +# The retained LixelGO control connection issued these three SUBSCRIBE packets +# in this exact order. Point-cloud subscriptions belonged to a separate client. +CONTROL_SUBSCRIPTION_GROUPS: tuple[tuple[tuple[str, int], ...], ...] = ( + ( + ("RealtimePath", 0), + ("RtkLioFusion", 0), + ("RtkPoseFusion", 0), + ("PrePathArray", 0), + ("ScanStatus", 0), + ("lixel/application/report/lio_pose", 0), + ("lixel/application/report/device_status", 0), + ("lixel/application/report/system_error", 0), + (DEVICE_INFO_RESPONSE_TOPIC, 0), + ), + ( + ("SystemStatus", 1), + ("DeviceStatus", 1), + ("StatisticalData", 1), + ("lixel/application/report/modeling", 1), + ("lixel/application/report/upgrade", 1), + ), + ( + ("AbnormalStatus", 2), + ("ActionApp2DeviceResponse", 2), + ("ActionAppSyncTimeResponse", 2), + ("DeviceInfoResponse", 2), + ("DeviceActivateResponse", 2), + ("CumulativeResponse", 2), + ("DebugStatusResponse", 2), + ("ModelingInfoResponse", 2), + ("ControlPointResponse", 2), + ("GetNtripConfigResponse", 2), + ("SetNtripConfigResponse", 2), + ("ProjectOperationResponse", 2), + ("PotreeConvertResponse", 2), + ("SetProjectNoteResponse", 2), + ("GetAppConfigResponse", 2), + ("SetAppConfigResponse", 2), + ("GetMqttConfigResponse", 2), + ("SetMqttConfigResponse", 2), + ("lixel/application/report/heartbeat", 2), + ("lixel/calibration/response/lidar", 2), + ("lixel/application/response/cameraConfig", 2), + ("lixel/application/response/service", 2), + ("lixel/application/report/pgo_progress", 2), + ("lixel/application/response/rtk/set_satellite_system", 2), + ("lixel/application/report/modeling/status", 2), + ("lixel/application/response/project_file", 2), + (GET_CLOUD_CONFIG_RESPONSE_TOPIC, 2), + ("lixel/application/response/set_cloud_server_config", 2), + (GET_NTRIP_PROFILE_RESPONSE_TOPIC, 2), + ("lixel/application/response/set_ntrip_profile", 2), + (GET_RTK_ADVANCE_RESPONSE_TOPIC, 2), + ("lixel/application/response/set_rtk_advance", 2), + (MODELING_STATUS_RESPONSE_TOPIC, 2), + ("lixel/application/response/control_point", 2), + ("lixel/application/response/service_control", 2), + ("lixel/application/response/project_list", 2), + ("lixel/application/response/device_control", 2), + (DEVICE_CONFIG_RESPONSE_TOPIC, 2), + (MODELING_RESPONSE_TOPIC, 2), + ("lixel/application/report/control_point", 2), + ("lixel/application/response/measure_point", 2), + ("lixel/application/response/upgrade", 2), + ), +) + +APPLICATION_RESPONSE_TOPICS = frozenset( + { + DEVICE_INFO_RESPONSE_TOPIC, + MODELING_STATUS_RESPONSE_TOPIC, + GET_RTK_ADVANCE_RESPONSE_TOPIC, + DEVICE_CONFIG_RESPONSE_TOPIC, + GET_NTRIP_PROFILE_RESPONSE_TOPIC, + GET_CLOUD_CONFIG_RESPONSE_TOPIC, + MODELING_RESPONSE_TOPIC, + } +) + +CONTROL_REPORT_TOPICS = frozenset( + topic + for group in CONTROL_SUBSCRIPTION_GROUPS + for topic, _qos in group + if "Response" not in topic and "/response/" not in topic +) + +APPLICATION_REQUEST_ALLOWLIST = frozenset( + { + "lixel/application/request/device_info", + "lixel/application/request/modeling_status", + "lixel/application/request/get_rtk_advance", + "lixel/application/request/device_config", + "lixel/application/request/get_ntrip_profile", + "lixel/application/request/get_cloud_server_config", + MODELING_REQUEST_TOPIC, + } +) + + +class ApplicationMqttTransportError(RuntimeError): + """The reviewed control transport failed before a request was attempted.""" + + +class ApplicationCommandOutcomeUnknown(RuntimeError): + """A request may have reached the K1 and must never be retried automatically.""" + + +@dataclass(frozen=True, slots=True) +class ApplicationMqttTransportSnapshot: + state: str + connect_attempts: int + subscribe_attempts: int + publish_attempts: int + qos2_completions: int + correlated_responses: int + ignored_known_responses: int + operation_keys_consumed: int + clean_session: bool = False + keepalive_seconds: int = CONTROL_KEEPALIVE_SECONDS + automatic_reconnect: bool = False + automatic_retry: bool = False + + def as_dict(self) -> dict[str, object]: + return { + "mode": "physical-acceptance-only", + "state": self.state, + "connect_attempts": self.connect_attempts, + "subscribe_attempts": self.subscribe_attempts, + "publish_attempts": self.publish_attempts, + "qos2_completions": self.qos2_completions, + "correlated_responses": self.correlated_responses, + "ignored_known_responses": self.ignored_known_responses, + "operation_keys_consumed": self.operation_keys_consumed, + "clean_session": self.clean_session, + "keepalive_seconds": self.keepalive_seconds, + "automatic_reconnect": self.automatic_reconnect, + "automatic_retry": self.automatic_retry, + } + + +class ReviewedApplicationMqttTransport: + """Exact-profile MQTT exchange for an operator-present physical acceptance. + + This type is deliberately not installed in the facade or plugin runtime. + It performs one connection attempt, never reconnects, consumes every + operation key before calling ``publish``, and permanently poisons itself + after any unknown post-publish outcome. + """ + + def __init__( + self, + host: str, + *, + port: int = 1883, + connect_timeout_seconds: float = CONTROL_CONNECT_TIMEOUT_SECONDS, + exchange_timeout_seconds: float = CONTROL_EXCHANGE_TIMEOUT_SECONDS, + client_factory: Callable[[], mqtt.Client] | None = None, + monotonic: Callable[[], float] = time.monotonic, + ) -> None: + self._target_ipv4 = validate_private_ipv4(host) + if self._target_ipv4 == AP_FALLBACK_IPV4: + raise ValueError("K1 access-point fallback address is not a direct-LAN control target") + if not 1 <= port <= 65535: + raise ValueError("port must be between 1 and 65535") + for name, value in ( + ("connect_timeout_seconds", connect_timeout_seconds), + ("exchange_timeout_seconds", exchange_timeout_seconds), + ): + if not math.isfinite(value) or value <= 0: + raise ValueError(f"{name} must be finite and greater than zero") + self._port = port + self._connect_timeout_seconds = connect_timeout_seconds + self._exchange_timeout_seconds = exchange_timeout_seconds + self._monotonic = monotonic + self._client_factory = client_factory + self._client: mqtt.Client | None = None + self._lock = threading.Lock() + self._state = "new" + self._connected = False + self._subscribed = False + self._closing = False + self._subscription_mid: int | None = None + self._subscription_group_index = 0 + self._completed_publish_mids: set[int] = set() + self._messages: deque[tuple[str, bytes]] = deque() + self._callback_error: str | None = None + self._consumed_operation_keys: set[str] = set() + self._connect_attempts = 0 + self._subscribe_attempts = 0 + self._publish_attempts = 0 + self._qos2_completions = 0 + self._correlated_responses = 0 + self._ignored_known_responses = 0 + + def open(self) -> ApplicationMqttTransportSnapshot: + with self._lock: + if self._state != "new": + raise ApplicationMqttTransportError("control transport can be opened only once") + self._state = "connecting" + self._connect_attempts = 1 + client = self._new_client() + self._install_callbacks(client) + self._client = client + try: + result = client.connect( + self._target_ipv4, + port=self._port, + keepalive=CONTROL_KEEPALIVE_SECONDS, + ) + except (OSError, RuntimeError, ValueError) as exc: + self._fail_before_publish("control MQTT connect call failed", exc) + if result != mqtt.MQTT_ERR_SUCCESS: + self._fail_before_publish("control MQTT connect call was rejected") + + deadline = self._monotonic() + self._connect_timeout_seconds + self._drive_until(lambda: self._subscribed, deadline, post_publish=False) + with self._lock: + self._state = "ready" + return self.snapshot() + + def exchange_batch_once( + self, + envelopes: Sequence[OneShotPublishEnvelope], + *, + required_response_topics: Collection[str], + ) -> dict[str, bytes]: + batch = tuple(envelopes) + required = frozenset(required_response_topics) + if not batch: + raise ValueError("control exchange batch must not be empty") + if not required or not required <= APPLICATION_RESPONSE_TOPICS: + raise ValueError("required response topics exceed the reviewed allowlist") + operation_keys = tuple(envelope.operation_key for envelope in batch) + if len(set(operation_keys)) != len(operation_keys): + raise ValueError("control exchange batch repeats an operation key") + if any(envelope.topic not in APPLICATION_REQUEST_ALLOWLIST for envelope in batch): + raise ValueError("control exchange batch exceeds the reviewed request allowlist") + allowed_non_barrier_responses = ( + frozenset({MODELING_STATUS_RESPONSE_TOPIC}) + if any( + envelope.topic == "lixel/application/request/modeling_status" for envelope in batch + ) + else frozenset() + ) + + with self._lock: + if self._consumed_operation_keys.intersection(operation_keys): + raise ApplicationCommandOutcomeUnknown( + "control operation key was already consumed; retry is forbidden" + ) + if self._state != "ready" or not self._connected or not self._subscribed: + raise ApplicationMqttTransportError("control transport is not ready") + stale_response = bool(self._messages) + if stale_response: + self._poison_locked("stale response preceded the request batch") + self._consumed_operation_keys.update(operation_keys) + if stale_response: + self.close() + raise ApplicationCommandOutcomeUnknown( + "stale control response makes the next command outcome ambiguous" + ) + + client = self._require_client() + publish_mids: set[int] = set() + for envelope in batch: + with self._lock: + self._publish_attempts += 1 + try: + info = client.publish( + envelope.topic, + payload=envelope.payload, + qos=envelope.qos, + retain=envelope.retain, + ) + except (OSError, RuntimeError, ValueError) as exc: + self._fail_after_publish("control MQTT publish call failed", exc) + if info.rc != mqtt.MQTT_ERR_SUCCESS or info.mid is None: + self._fail_after_publish("control MQTT publish call returned an unsafe result") + if info.mid in publish_mids: + self._fail_after_publish("control MQTT reused a packet identifier") + publish_mids.add(info.mid) + + responses: dict[str, bytes] = {} + deadline = self._monotonic() + self._exchange_timeout_seconds + + def complete() -> bool: + self._drain_responses(required, allowed_non_barrier_responses, responses) + with self._lock: + return publish_mids <= self._completed_publish_mids and required <= responses.keys() + + self._drive_until(complete, deadline, post_publish=True) + self._drain_responses(required, allowed_non_barrier_responses, responses) + with self._lock: + self._correlated_responses += len(responses) + return responses + + def close(self) -> None: + with self._lock: + if self._state == "closed": + return + self._closing = True + client = self._client + if client is not None: + try: + if self._subscribed: + client.unsubscribe( + [topic for group in CONTROL_SUBSCRIPTION_GROUPS for topic, _qos in group] + ) + client.disconnect() + except (OSError, RuntimeError, ValueError): + pass + with self._lock: + self._connected = False + self._subscribed = False + self._messages.clear() + if self._state not in {"poisoned", "failed"}: + self._state = "closed" + + def snapshot(self) -> ApplicationMqttTransportSnapshot: + with self._lock: + return ApplicationMqttTransportSnapshot( + state=self._state, + connect_attempts=self._connect_attempts, + subscribe_attempts=self._subscribe_attempts, + publish_attempts=self._publish_attempts, + qos2_completions=self._qos2_completions, + correlated_responses=self._correlated_responses, + ignored_known_responses=self._ignored_known_responses, + operation_keys_consumed=len(self._consumed_operation_keys), + ) + + def _new_client(self) -> mqtt.Client: + if self._client_factory is not None: + return self._client_factory() + client_id = f"mck1-{secrets.token_hex(7)}" + return mqtt.Client( + callback_api_version=CallbackAPIVersion.VERSION2, + client_id=client_id, + clean_session=False, + protocol=mqtt.MQTTv311, + reconnect_on_failure=False, + ) + + def _install_callbacks(self, client: mqtt.Client) -> None: + def on_connect( + callback_client: mqtt.Client, + _userdata: object, + _flags: mqtt.ConnectFlags, + reason_code: ReasonCode, + _properties: Properties | None, + ) -> None: + if reason_code.is_failure: + self._set_callback_error("control MQTT broker rejected connection") + return + with self._lock: + self._connected = True + self._subscribe_next_group(callback_client) + + def on_subscribe( + _callback_client: mqtt.Client, + _userdata: object, + mid: int, + reason_codes: list[ReasonCode], + _properties: Properties | None, + ) -> None: + with self._lock: + expected_mid = self._subscription_mid + group_index = self._subscription_group_index + if group_index >= len(CONTROL_SUBSCRIPTION_GROUPS): + self._set_callback_error("control MQTT received a surplus SUBACK") + return + expected_group = CONTROL_SUBSCRIPTION_GROUPS[group_index] + if mid != expected_mid or len(reason_codes) != len(expected_group): + self._set_callback_error("control MQTT received an unexpected SUBACK") + return + if any(reason_code.is_failure for reason_code in reason_codes): + self._set_callback_error("control MQTT broker rejected a response subscription") + return + with self._lock: + self._subscription_group_index += 1 + complete = self._subscription_group_index == len(CONTROL_SUBSCRIPTION_GROUPS) + self._subscribed = complete + if not complete: + self._subscribe_next_group(_callback_client) + + def on_publish( + _callback_client: mqtt.Client, + _userdata: object, + mid: int, + reason_code: ReasonCode, + _properties: Properties | None, + ) -> None: + if reason_code.is_failure: + self._set_callback_error("control MQTT QoS2 transaction failed") + return + with self._lock: + self._completed_publish_mids.add(mid) + self._qos2_completions += 1 + + def on_message( + _callback_client: mqtt.Client, + _userdata: object, + message: mqtt.MQTTMessage, + ) -> None: + allowed = {topic for group in CONTROL_SUBSCRIPTION_GROUPS for topic, _qos in group} + if message.topic not in allowed: + self._set_callback_error("control MQTT received an unreviewed subscribed topic") + return + payload = bytes(message.payload) + if len(payload) > MAX_CONTROL_RESPONSE_BYTES: + self._set_callback_error("control MQTT response exceeds the reviewed bound") + return + with self._lock: + if message.topic in CONTROL_REPORT_TOPICS: + self._ignored_known_responses += 1 + else: + self._messages.append((message.topic, payload)) + + def on_disconnect( + _callback_client: mqtt.Client, + _userdata: object, + _flags: mqtt.DisconnectFlags, + reason_code: ReasonCode, + _properties: Properties | None, + ) -> None: + with self._lock: + expected = self._closing + self._connected = False + if not expected or reason_code.is_failure: + self._set_callback_error("control MQTT connection ended unexpectedly") + + client.on_connect = on_connect + client.on_subscribe = on_subscribe + client.on_publish = on_publish + client.on_message = on_message + client.on_disconnect = on_disconnect + + def _subscribe_next_group(self, client: mqtt.Client) -> None: + with self._lock: + group_index = self._subscription_group_index + if group_index < len(CONTROL_SUBSCRIPTION_GROUPS): + group = CONTROL_SUBSCRIPTION_GROUPS[group_index] + self._subscribe_attempts += 1 + else: + group = None + if group is None: + self._set_callback_error("control MQTT subscription sequence overflowed") + return + try: + result, mid = client.subscribe(list(group)) + except (OSError, RuntimeError, ValueError): + self._set_callback_error("control MQTT response subscription failed") + return + if result != mqtt.MQTT_ERR_SUCCESS or mid is None: + self._set_callback_error("control MQTT response subscription was rejected") + return + with self._lock: + self._subscription_mid = mid + + def _drive_until( + self, + predicate: Callable[[], bool], + deadline: float, + *, + post_publish: bool, + ) -> None: + client = self._require_client() + while not predicate(): + with self._lock: + callback_error = self._callback_error + if callback_error is not None: + if post_publish: + self._fail_after_publish(callback_error) + self._fail_before_publish(callback_error) + if self._monotonic() >= deadline: + if post_publish: + self._fail_after_publish("control MQTT response barrier timed out") + self._fail_before_publish("control MQTT connection/subscription timed out") + try: + result = client.loop(timeout=CONTROL_LOOP_INTERVAL_SECONDS) + except (OSError, RuntimeError, ValueError) as exc: + if post_publish: + self._fail_after_publish("control MQTT network loop failed", exc) + self._fail_before_publish("control MQTT network loop failed", exc) + if result != mqtt.MQTT_ERR_SUCCESS: + if post_publish: + self._fail_after_publish("control MQTT network loop returned an error") + self._fail_before_publish("control MQTT network loop returned an error") + + def _drain_responses( + self, + required: frozenset[str], + allowed_non_barrier: frozenset[str], + responses: dict[str, bytes], + ) -> None: + ambiguous_message: str | None = None + with self._lock: + while self._messages: + topic, payload = self._messages.popleft() + if topic not in required: + if topic in allowed_non_barrier: + self._ignored_known_responses += 1 + continue + self._poison_locked("unexpected response made correlation ambiguous") + ambiguous_message = "unexpected control response made command outcome ambiguous" + break + if topic in responses: + self._poison_locked("duplicate required response made correlation ambiguous") + ambiguous_message = "duplicate control response made command outcome ambiguous" + break + responses[topic] = payload + if ambiguous_message is not None: + self.close() + raise ApplicationCommandOutcomeUnknown(ambiguous_message) + + def _set_callback_error(self, message: str) -> None: + with self._lock: + if self._callback_error is None: + self._callback_error = message + + def _fail_before_publish(self, message: str, cause: BaseException | None = None) -> None: + with self._lock: + self._state = "failed" + self.close() + error = ApplicationMqttTransportError(message) + if cause is not None: + raise error from cause + raise error + + def _fail_after_publish(self, message: str, cause: BaseException | None = None) -> None: + with self._lock: + self._poison_locked(message) + self.close() + error = ApplicationCommandOutcomeUnknown( + f"{message}; automatic retry is forbidden until physical/status reconciliation" + ) + if cause is not None: + raise error from cause + raise error + + def _poison_locked(self, _reason: str) -> None: + self._state = "poisoned" + + def _require_client(self) -> mqtt.Client: + client = self._client + if client is None: + raise ApplicationMqttTransportError("control MQTT client is not installed") + return client diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_publish.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_publish.py index 5a5673b..567def0 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/application_publish.py +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_publish.py @@ -45,6 +45,7 @@ class UninstalledApplicationPublishSink: @dataclass(frozen=True, slots=True) class OneShotPublishEnvelope: + operation_key: str topic: str payload: bytes = field(repr=False) payload_sha256: str @@ -59,6 +60,7 @@ class OneShotPublishEnvelope: request: EncodedApplicationRequest, ) -> OneShotPublishEnvelope: return cls( + operation_key=f"bootstrap:{request.ordinal}:{request.message_type}", topic=request.topic, payload=request.payload, payload_sha256=request.payload_sha256, @@ -73,6 +75,7 @@ class OneShotPublishEnvelope: command: ShadowModelingCommand, ) -> OneShotPublishEnvelope: return cls( + operation_key=f"modeling:{command.action.name.casefold()}", topic=command.topic, payload=command.command.payload, payload_sha256=command.payload_sha256, @@ -82,6 +85,10 @@ class OneShotPublishEnvelope: ) def __post_init__(self) -> None: + if not self.operation_key or any( + ord(character) <= 0x20 or ord(character) > 0x7E for character in self.operation_key + ): + raise ValueError("publish envelope operation key is invalid") if self.qos != 2 or self.retain or self.automatic_retry: raise ValueError("publish envelope differs from the reviewed one-shot contract") if self.payload_bytes != len(self.payload): @@ -91,6 +98,7 @@ class OneShotPublishEnvelope: def as_dict(self) -> dict[str, object]: return { + "operation_key": self.operation_key, "topic": self.topic, "payload_sha256": self.payload_sha256, "payload_bytes": self.payload_bytes, diff --git a/tests/test_cli.py b/tests/test_cli.py index 2e3a885..f99f32f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -25,6 +25,41 @@ def test_doctor_json() -> None: assert any(item["name"] == "tcpdump" for item in payload["tools"]) +def test_authority_provision_requires_explicit_reviewed_value_confirmation() -> None: + result = runner.invoke(app, ["authority", "provision"]) + + assert result.exit_code == 2 + assert "provisioning not confirmed" in result.stdout + + +def test_authority_provision_never_accepts_the_secret_as_a_cli_value(monkeypatch: Any) -> None: + calls = 0 + + class FakeSnapshot: + service = "fixed-service" + account = "fixed-account" + + class FakeProvisioner: + def provision_interactively(self) -> FakeSnapshot: + nonlocal calls + calls += 1 + return FakeSnapshot() + + monkeypatch.setattr( + "k1link.device_plugins.xgrids_k1.cli.MacOSKeychainApplicationAuthorityProvisioner", + FakeProvisioner, + ) + result = runner.invoke( + app, + ["authority", "provision", "--confirm-reviewed-authority"], + ) + + assert result.exit_code == 0 + assert calls == 1 + assert "Keychain item validated" in result.stdout + assert "No K1 command was sent" in " ".join(result.stdout.split()) + + def test_mqtt_capture_requires_owned_device_confirmation(tmp_path: Path) -> None: result = runner.invoke( app, diff --git a/tests/test_xgrids_application_acceptance.py b/tests/test_xgrids_application_acceptance.py new file mode 100644 index 0000000..4e3b2f6 --- /dev/null +++ b/tests/test_xgrids_application_acceptance.py @@ -0,0 +1,243 @@ +from __future__ import annotations + +from collections.abc import Collection, Sequence + +import pytest + +from k1link.device_plugins.xgrids_k1.protocol.application_acceptance import ( + ApplicationAcceptanceError, + PhysicalAcceptanceChecklist, + PhysicalAcceptanceDialogueExecutor, + PhysicalAcceptancePermit, +) +from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( + ApplicationControlAuthority, + ShadowApplicationBootstrapOrchestrator, +) +from k1link.device_plugins.xgrids_k1.protocol.application_mqtt import ( + MODELING_RESPONSE_TOPIC, +) +from k1link.device_plugins.xgrids_k1.protocol.application_publish import ( + OneShotPublishEnvelope, +) +from k1link.device_plugins.xgrids_k1.protocol.modeling_control import ( + OPENAPI_SUCCESS, + CommandHeaderIdentity, + ModelingAction, + MountType, + RecordMode, + ScanMode, + encode_modeling_start, + encode_modeling_stop, +) +from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import ( + ShadowModelingCommand, +) + +APPLICATION_KEY = "11111111-2222-3333-4444-555555555555" +VENDOR_DEVICE_ID = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" + + +def _varint(value: int) -> bytes: + encoded = bytearray() + while value > 0x7F: + encoded.append((value & 0x7F) | 0x80) + value >>= 7 + encoded.append(value) + return bytes(encoded) + + +def _uint(number: int, value: int) -> bytes: + return _varint(number << 3) + _varint(value) + + +def _bytes(number: int, value: bytes) -> bytes: + return _varint((number << 3) | 2) + _varint(len(value)) + value + + +def _text(number: int, value: str) -> bytes: + return _bytes(number, value.encode()) + + +def _header(session_id: str) -> bytes: + return _text(4, VENDOR_DEVICE_ID) + _text(5, session_id) + _text(6, APPLICATION_KEY) + + +def _device_info_response(session_id: str) -> bytes: + base_info = b"".join( + ( + _text(2, "V3.0.2-20260101-release"), + _text(3, "V3.0.2"), + _text(6, "LixelKity K1"), + _text(7, "K1SERIAL01"), + _text(8, "K1"), + ) + ) + working_status = _uint(1, 1) + device_info = _bytes(2, base_info) + _bytes(7, working_status) + error = _uint(1, OPENAPI_SUCCESS) + return _bytes(1, _header(session_id)) + _bytes(2, device_info) + _bytes(15, error) + + +def _generic_response(session_id: str) -> bytes: + return _bytes(1, _header(session_id)) + _bytes(15, _uint(1, OPENAPI_SUCCESS)) + + +def _modeling_response(action: ModelingAction) -> bytes: + session = f"{VENDOR_DEVICE_ID}:ModelingRequest" + return _bytes(1, _header(session)) + _uint(2, action) + _bytes(15, _uint(1, OPENAPI_SUCCESS)) + + +class SyntheticAcceptanceTransport: + def __init__(self) -> None: + self.batches: list[tuple[str, ...]] = [] + + def exchange_batch_once( + self, + envelopes: Sequence[OneShotPublishEnvelope], + *, + required_response_topics: Collection[str], + ) -> dict[str, bytes]: + self.batches.append(tuple(envelope.operation_key for envelope in envelopes)) + responses: dict[str, bytes] = {} + if MODELING_RESPONSE_TOPIC in required_response_topics: + action = ( + ModelingAction.START + if envelopes[0].operation_key == "modeling:start" + else ModelingAction.STOP + ) + responses[MODELING_RESPONSE_TOPIC] = _modeling_response(action) + return responses + + for envelope in envelopes: + prefix, ordinal_text, message_type = envelope.operation_key.split(":", 2) + assert prefix == "bootstrap" + ordinal = int(ordinal_text) + if message_type == "DeviceInfoRequest": + session = ( + ":DeviceInfoRequest" + if ordinal == 1 + else f"{VENDOR_DEVICE_ID}:DeviceInfoRequest" + ) + response = _device_info_response(session) + else: + session = ( + f":{message_type}" if ordinal <= 3 else f"{VENDOR_DEVICE_ID}:{message_type}" + ) + if message_type == "DeviceConfigRequest": + session += ":Publish_Proto_DeviceConfig_SetTime" + response = _generic_response(session) + for topic in required_response_topics: + if topic.endswith(_response_suffix(message_type)): + responses[topic] = response + return responses + + +def _response_suffix(message_type: str) -> str: + values = { + "DeviceInfoRequest": "device_info", + "ModelingStatusRequest": "modeling_status", + "GetRtkAdvanceRequest": "get_rtk_advance", + "DeviceConfigRequest": "device_config", + "GetNtripProfileRequest": "get_ntrip_profile", + "GetCloudServerConfigRequest": "get_cloud_server_config", + } + return values[message_type] + + +def _checklist(action: ModelingAction) -> PhysicalAcceptanceChecklist: + return PhysicalAcceptanceChecklist( + action=action, + operator_present=True, + owner_controlled_device=True, + lixelgo_closed=True, + battery_storage_confirmed=True, + expected_physical_state_confirmed=True, + ) + + +def test_start_acceptance_requires_full_bootstrap_and_consumes_one_short_permit() -> None: + transport = SyntheticAcceptanceTransport() + permit = PhysicalAcceptancePermit(_checklist(ModelingAction.START)) + executor = PhysicalAcceptanceDialogueExecutor(transport, permit) + orchestrator = ShadowApplicationBootstrapOrchestrator( + ApplicationControlAuthority(openapi_key=APPLICATION_KEY), + epoch_seconds=1_752_680_000, + timezone_name="Europe/Moscow", + ) + + binding = executor.run_bootstrap(orchestrator) + command = ShadowModelingCommand.from_command( + encode_modeling_start( + CommandHeaderIdentity( + device_id=binding.vendor_device_id, + openapi_key=APPLICATION_KEY, + ), + project_name="SAFE_PROJECT", + record_mode=RecordMode.RECORD_AND_CALCULATE, + scan_mode=ScanMode.LCC, + mount_type=MountType.HANDHELD, + ) + ) + response = executor.execute_modeling(command) + + assert response.action is ModelingAction.START + assert [len(batch) for batch in transport.batches] == [1, 2, 3, 1, 3, 1] + assert permit.snapshot()["consumed"] is True + assert executor.snapshot()["bootstrap_complete"] is True + assert executor.snapshot()["command_complete"] is True + with pytest.raises(ApplicationAcceptanceError, match="already attempted"): + executor.execute_modeling(command) + + +def test_start_cannot_skip_bootstrap_and_stop_uses_a_separate_action_permit() -> None: + transport = SyntheticAcceptanceTransport() + start = PhysicalAcceptanceDialogueExecutor( + transport, + PhysicalAcceptancePermit(_checklist(ModelingAction.START)), + ) + start_command = ShadowModelingCommand.from_command( + encode_modeling_start( + CommandHeaderIdentity(device_id=VENDOR_DEVICE_ID, openapi_key=APPLICATION_KEY), + project_name="SAFE_PROJECT", + record_mode=RecordMode.RECORD_AND_CALCULATE, + scan_mode=ScanMode.LCC, + mount_type=MountType.HANDHELD, + ) + ) + with pytest.raises(ApplicationAcceptanceError, match="requires.*bootstrap"): + start.execute_modeling(start_command) + + stop = PhysicalAcceptanceDialogueExecutor( + transport, + PhysicalAcceptancePermit(_checklist(ModelingAction.STOP)), + ) + stop_command = ShadowModelingCommand.from_command( + encode_modeling_stop( + CommandHeaderIdentity(device_id=VENDOR_DEVICE_ID, openapi_key=APPLICATION_KEY) + ) + ) + assert stop.execute_modeling(stop_command).action is ModelingAction.STOP + assert transport.batches[-1] == ("modeling:stop",) + + +def test_acceptance_permit_rejects_implicit_or_expired_authority() -> None: + with pytest.raises(ApplicationAcceptanceError, match="explicitly true"): + PhysicalAcceptanceChecklist( + action=ModelingAction.START, + operator_present=False, # type: ignore[arg-type] + owner_controlled_device=True, + lixelgo_closed=True, + battery_storage_confirmed=True, + expected_physical_state_confirmed=True, + ) + + now = [100.0] + permit = PhysicalAcceptancePermit( + _checklist(ModelingAction.STOP), + ttl_seconds=15.0, + monotonic=lambda: now[0], + ) + now[0] = 115.0 + with pytest.raises(ApplicationAcceptanceError, match="expired"): + permit.consume(ModelingAction.STOP) diff --git a/tests/test_xgrids_application_authority.py b/tests/test_xgrids_application_authority.py index e3082d4..22075bc 100644 --- a/tests/test_xgrids_application_authority.py +++ b/tests/test_xgrids_application_authority.py @@ -10,6 +10,7 @@ from k1link.device_plugins.xgrids_k1.protocol.application_authority import ( KEYCHAIN_SERVICE, ApplicationAuthorityLoadError, MacOSKeychainApplicationAuthorityLoader, + MacOSKeychainApplicationAuthorityProvisioner, ) PRIVATE_AUTHORITY = b"11111111-2222-3333-4444-555555555555\n" @@ -108,3 +109,67 @@ def test_authority_loader_rejects_wrong_profile_length_without_reflection( MacOSKeychainApplicationAuthorityLoader(runner=runner).load() assert invalid_secret.decode() not in str(error.value) + + +@patch("sys.stdout.isatty", return_value=True) +@patch("sys.stdin.isatty", return_value=True) +@patch("platform.system", return_value="Darwin") +@patch("shutil.which", return_value="/usr/bin/security") +def test_authority_provisioning_uses_hidden_security_prompt_and_validates_item( + _which: object, + _system: object, + _stdin_tty: object, + _stdout_tty: object, +) -> None: + provisioning_calls: list[list[str]] = [] + + def interactive_runner( + args: list[str], + *, + check: bool, + timeout: float, + ) -> subprocess.CompletedProcess[bytes]: + provisioning_calls.append(args) + assert check is False + assert timeout == 300.0 + return subprocess.CompletedProcess(args, 0) + + def lookup_runner( + args: list[str], + **_kwargs: object, + ) -> subprocess.CompletedProcess[bytes]: + return subprocess.CompletedProcess(args, 0, stdout=PRIVATE_AUTHORITY, stderr=b"") + + provisioner = MacOSKeychainApplicationAuthorityProvisioner( + runner=interactive_runner, + loader=MacOSKeychainApplicationAuthorityLoader(runner=lookup_runner), + ) + snapshot = provisioner.provision_interactively() + + assert provisioning_calls == [ + [ + "/usr/bin/security", + "add-generic-password", + "-U", + "-s", + KEYCHAIN_SERVICE, + "-a", + KEYCHAIN_ACCOUNT, + "-w", + ] + ] + assert PRIVATE_AUTHORITY.decode().strip() not in str(provisioning_calls) + assert snapshot.secret_cached is False + assert snapshot.secret_exportable is False + + +@patch("sys.stdout.isatty", return_value=False) +@patch("sys.stdin.isatty", return_value=False) +@patch("platform.system", return_value="Darwin") +def test_authority_provisioning_requires_an_interactive_terminal( + _system: object, + _stdin_tty: object, + _stdout_tty: object, +) -> None: + with pytest.raises(ApplicationAuthorityLoadError, match="interactive terminal"): + MacOSKeychainApplicationAuthorityProvisioner().provision_interactively() diff --git a/tests/test_xgrids_application_mqtt.py b/tests/test_xgrids_application_mqtt.py new file mode 100644 index 0000000..0b3be30 --- /dev/null +++ b/tests/test_xgrids_application_mqtt.py @@ -0,0 +1,225 @@ +from __future__ import annotations + +import hashlib +from collections import deque +from types import SimpleNamespace +from typing import Any, cast + +import paho.mqtt.client as mqtt +import pytest + +from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( + DEVICE_INFO_RESPONSE_TOPIC, +) +from k1link.device_plugins.xgrids_k1.protocol.application_mqtt import ( + CONTROL_SUBSCRIPTION_GROUPS, + ApplicationCommandOutcomeUnknown, + ReviewedApplicationMqttTransport, +) +from k1link.device_plugins.xgrids_k1.protocol.application_publish import ( + OneShotPublishEnvelope, +) + + +class FakeReasonCode: + def __init__(self, *, failure: bool = False) -> None: + self.is_failure = failure + + +class FakeClock: + def __init__(self) -> None: + self.now = 100.0 + + def __call__(self) -> float: + return self.now + + +class FakeControlClient: + def __init__(self, *, emit_exchange: bool = True, clock: FakeClock | None = None) -> None: + self.on_connect: Any = None + self.on_subscribe: Any = None + self.on_publish: Any = None + self.on_message: Any = None + self.on_disconnect: Any = None + self.emit_exchange = emit_exchange + self.clock = clock + self.events: deque[tuple[str, object]] = deque() + self.connect_calls: list[tuple[str, int, int]] = [] + self.subscribe_calls: list[list[tuple[str, int]]] = [] + self.publish_calls: list[tuple[str, bytes, int, bool]] = [] + self.unsubscribe_calls: list[list[str]] = [] + self.disconnect_calls = 0 + self.next_mid = 20 + + def connect(self, host: str, port: int, keepalive: int) -> mqtt.MQTTErrorCode: + self.connect_calls.append((host, port, keepalive)) + self.events.append(("connect", FakeReasonCode())) + return mqtt.MQTT_ERR_SUCCESS + + def subscribe(self, topics: list[tuple[str, int]]) -> tuple[mqtt.MQTTErrorCode, int]: + self.subscribe_calls.append(topics) + mid = 6 + len(self.subscribe_calls) + self.events.append(("subscribe", mid)) + return mqtt.MQTT_ERR_SUCCESS, mid + + def publish( + self, + topic: str, + payload: bytes, + qos: int, + retain: bool, + ) -> SimpleNamespace: + self.publish_calls.append((topic, payload, qos, retain)) + mid = self.next_mid + self.next_mid += 1 + if self.emit_exchange: + self.events.append(("publish", mid)) + self.events.append(("message", (DEVICE_INFO_RESPONSE_TOPIC, b"response"))) + return SimpleNamespace(rc=mqtt.MQTT_ERR_SUCCESS, mid=mid) + + def loop(self, timeout: float) -> mqtt.MQTTErrorCode: + del timeout + if self.clock is not None: + self.clock.now += 1.0 + if not self.events: + return mqtt.MQTT_ERR_SUCCESS + kind, value = self.events.popleft() + if kind == "connect": + self.on_connect(self, None, SimpleNamespace(), value, None) + elif kind == "subscribe": + group = CONTROL_SUBSCRIPTION_GROUPS[int(value) - 7] + self.on_subscribe( + self, + None, + value, + [FakeReasonCode() for _item in group], + None, + ) + elif kind == "publish": + self.on_publish(self, None, value, FakeReasonCode(), None) + elif kind == "message": + topic, payload = value + self.on_message(self, None, SimpleNamespace(topic=topic, payload=payload)) + return mqtt.MQTT_ERR_SUCCESS + + def unsubscribe(self, topics: list[str]) -> tuple[mqtt.MQTTErrorCode, int]: + self.unsubscribe_calls.append(topics) + return mqtt.MQTT_ERR_SUCCESS, 50 + + def disconnect(self) -> mqtt.MQTTErrorCode: + self.disconnect_calls += 1 + return mqtt.MQTT_ERR_SUCCESS + + +def _envelope(operation_key: str = "bootstrap:1:DeviceInfoRequest") -> OneShotPublishEnvelope: + payload = b"synthetic-reviewed-request" + return OneShotPublishEnvelope( + operation_key=operation_key, + topic="lixel/application/request/device_info", + payload=payload, + payload_sha256=hashlib.sha256(payload).hexdigest(), + payload_bytes=len(payload), + qos=2, + retain=False, + ) + + +def test_acceptance_transport_rejects_the_k1_access_point_fallback() -> None: + with pytest.raises(ValueError, match="not a direct-LAN"): + ReviewedApplicationMqttTransport("192.168.56.1") + + +def test_retained_control_subscription_batches_remain_exact_and_separate_from_points() -> None: + assert [len(group) for group in CONTROL_SUBSCRIPTION_GROUPS] == [9, 5, 42] + assert CONTROL_SUBSCRIPTION_GROUPS[0][-1] == (DEVICE_INFO_RESPONSE_TOPIC, 0) + assert ("lixel/application/response/modeling", 2) in CONTROL_SUBSCRIPTION_GROUPS[2] + assert all( + topic not in {"RealtimePointcloud", "lixel/application/report/lio_pcl"} + for group in CONTROL_SUBSCRIPTION_GROUPS + for topic, _qos in group + ) + + +def test_acceptance_transport_connects_once_and_completes_one_qos2_exchange() -> None: + fake = FakeControlClient() + transport = ReviewedApplicationMqttTransport( + "192.168.1.20", + client_factory=lambda: cast(mqtt.Client, fake), + ) + + opened = transport.open().as_dict() + responses = transport.exchange_batch_once( + [_envelope()], + required_response_topics={DEVICE_INFO_RESPONSE_TOPIC}, + ) + transport.close() + + assert fake.connect_calls == [("192.168.1.20", 1883, 60)] + assert fake.subscribe_calls == [list(group) for group in CONTROL_SUBSCRIPTION_GROUPS] + assert fake.publish_calls == [ + ( + "lixel/application/request/device_info", + b"synthetic-reviewed-request", + 2, + False, + ) + ] + assert responses == {DEVICE_INFO_RESPONSE_TOPIC: b"response"} + assert opened["clean_session"] is False + assert opened["automatic_reconnect"] is False + assert opened["automatic_retry"] is False + snapshot = transport.snapshot().as_dict() + assert snapshot["state"] == "closed" + assert snapshot["publish_attempts"] == 1 + assert snapshot["subscribe_attempts"] == 3 + assert snapshot["qos2_completions"] == 1 + assert snapshot["correlated_responses"] == 1 + assert snapshot["operation_keys_consumed"] == 1 + + +def test_consumed_operation_key_can_never_be_published_again() -> None: + fake = FakeControlClient() + transport = ReviewedApplicationMqttTransport( + "192.168.1.20", + client_factory=lambda: cast(mqtt.Client, fake), + ) + transport.open() + transport.exchange_batch_once( + [_envelope()], + required_response_topics={DEVICE_INFO_RESPONSE_TOPIC}, + ) + + with pytest.raises(ApplicationCommandOutcomeUnknown, match="already consumed"): + transport.exchange_batch_once( + [_envelope()], + required_response_topics={DEVICE_INFO_RESPONSE_TOPIC}, + ) + + assert len(fake.publish_calls) == 1 + + +def test_post_publish_timeout_poisoned_transport_never_retries() -> None: + clock = FakeClock() + fake = FakeControlClient(emit_exchange=False, clock=clock) + transport = ReviewedApplicationMqttTransport( + "192.168.1.20", + exchange_timeout_seconds=2.0, + client_factory=lambda: cast(mqtt.Client, fake), + monotonic=clock, + ) + transport.open() + + with pytest.raises(ApplicationCommandOutcomeUnknown, match="automatic retry is forbidden"): + transport.exchange_batch_once( + [_envelope()], + required_response_topics={DEVICE_INFO_RESPONSE_TOPIC}, + ) + + assert transport.snapshot().state == "poisoned" + assert len(fake.publish_calls) == 1 + with pytest.raises(ApplicationCommandOutcomeUnknown, match="already consumed"): + transport.exchange_batch_once( + [_envelope()], + required_response_topics={DEVICE_INFO_RESPONSE_TOPIC}, + ) + assert len(fake.publish_calls) == 1 diff --git a/tests/test_xgrids_application_publish.py b/tests/test_xgrids_application_publish.py index 4eaacc8..fde3c9a 100644 --- a/tests/test_xgrids_application_publish.py +++ b/tests/test_xgrids_application_publish.py @@ -79,6 +79,7 @@ def test_publish_envelope_redacts_payload_and_rejects_retry_semantics() -> None: assert PRIVATE_AUTHORITY not in str(envelope.as_dict()) with pytest.raises(ValueError, match="one-shot contract"): OneShotPublishEnvelope( + operation_key=envelope.operation_key, topic=envelope.topic, payload=envelope.payload, payload_sha256=envelope.payload_sha256,