"""Authenticated HTTP control channel for the RF-001 RX laptop. The RX laptop always initiates the connection. The server exposes only a small JSON API, never a filesystem, and accepts only fixed RF-001 task kinds. RF capture is receive-only and requires an exact local authorization phrase. RF transmission is deliberately absent from the network protocol. """ from __future__ import annotations import argparse from collections import OrderedDict from dataclasses import dataclass from datetime import datetime, timezone import hashlib import hmac from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import os from pathlib import Path import re import secrets import sys import threading import time from typing import Any, Mapping from urllib.error import HTTPError from urllib.parse import parse_qs, quote, urlsplit from urllib.request import ProxyHandler, Request, build_opener try: from tools.rf001_agent_update import ( RELEASE_ID_PATTERN, ReleaseValidationError, inspect_release_archive, sign_update_manifest, ) except ModuleNotFoundError: # Direct execution from tools/. from rf001_agent_update import ( # type: ignore[no-redef] RELEASE_ID_PATTERN, ReleaseValidationError, inspect_release_archive, sign_update_manifest, ) ROOT = Path(__file__).resolve().parents[1] PROTOCOL_VERSION = 1 MAX_JSON_BYTES = 256 * 1024 AUTH_WINDOW_SECONDS = 60 TASK_LEASE_SECONDS = 30 INVENTORY_TASK = "INVENTORY" OFFLINE_VERIFY_TASK = "OFFLINE_VERIFY" RX_CAPTURE_START_TASK = "RX_CAPTURE_START" RX_CAPTURE_STATUS_TASK = "RX_CAPTURE_STATUS" CAPTURE_DECODE_TASK = "CAPTURE_DECODE" CAPTURE_DIAGNOSTICS_TASK = "CAPTURE_DIAGNOSTICS" STAIRCASE_SEARCH_TASK = "STAIRCASE_SEARCH" UPDATE_AGENT_TASK = "UPDATE_AGENT" RX_AUTHORIZATION_PHRASE = "AUTHORIZE RF RX" FIXED_RX_SERIAL = "000000000000000097c063dc2e5327d3" TASK_ID_PATTERN = re.compile(r"^[A-Z0-9][A-Z0-9_-]{2,95}$") AGENT_ID_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{2,63}$") BOOTSTRAP_RELEASE_ID_PATTERN = re.compile( r"^rf001-bootstrap-[0-9]{4}\.[0-9]{2}\.[0-9]{2}\.[0-9]+$" ) GIT_HEAD_PATTERN = re.compile(r"^[0-9a-f]{40}$") SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") SESSION_ID_PATTERN = re.compile(r"^RF001A_[0-9]{8}T[0-9]{6}Z_[0-9A-F]{8}$") DEFAULT_STATE_PATH = ROOT / ".local" / "rf001_http_server_state.json" DEFAULT_TOKEN_PATH = ROOT / ".local" / "rf001_http_token.txt" DEFAULT_SERVER_URL = "http://192.168.255.185:8765" DEFAULT_RELEASES_ROOT = ROOT / ".local" / "releases" def redirect_output(path: Path) -> None: """Send service-mode stdout and stderr to one line-buffered UTF-8 log.""" path.parent.mkdir(parents=True, exist_ok=True) stream = path.open("a", encoding="utf-8", buffering=1) sys.stdout = stream sys.stderr = stream class ProtocolError(ValueError): """The peer sent an invalid or unsupported protocol message.""" class AuthenticationError(ProtocolError): """Request authentication failed.""" def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def canonical_json(value: Mapping[str, Any]) -> bytes: return json.dumps( value, ensure_ascii=False, separators=(",", ":"), sort_keys=True, ).encode("utf-8") def read_token(path: Path) -> bytes: token = path.read_text(encoding="ascii").strip().encode("ascii") if len(token) < 32: raise ValueError("token must contain at least 32 ASCII bytes") return token def initialize_token(path: Path) -> str: """Create a token without overwriting an existing credential.""" path.parent.mkdir(parents=True, exist_ok=True) token = secrets.token_urlsafe(48) try: with path.open("x", encoding="ascii", newline="\n") as stream: stream.write(token + "\n") except FileExistsError as error: raise FileExistsError(f"token already exists: {path}") from error return token def _signature_payload( method: str, target: str, timestamp: str, nonce: str, body: bytes, ) -> bytes: body_digest = hashlib.sha256(body).hexdigest() fields = ( str(PROTOCOL_VERSION), timestamp, nonce, method.upper(), target, body_digest, ) return "\n".join(fields).encode("ascii") def signed_headers( token: bytes, method: str, target: str, body: bytes, *, timestamp: int | None = None, nonce: str | None = None, ) -> dict[str, str]: timestamp_text = str(int(time.time()) if timestamp is None else timestamp) nonce_text = secrets.token_hex(16) if nonce is None else nonce if not re.fullmatch(r"[0-9a-f]{32}", nonce_text): raise ValueError("nonce must be 32 lowercase hexadecimal characters") payload = _signature_payload( method, target, timestamp_text, nonce_text, body, ) signature = hmac.new(token, payload, hashlib.sha256).hexdigest() return { "Content-Type": "application/json", "X-SDR-Version": str(PROTOCOL_VERSION), "X-SDR-Timestamp": timestamp_text, "X-SDR-Nonce": nonce_text, "X-SDR-Signature": signature, } class RequestAuthenticator: """Verify request signatures and reject nonce replay within the time window.""" def __init__( self, token: bytes, *, window_seconds: int = AUTH_WINDOW_SECONDS, ) -> None: self.token = token self.window_seconds = window_seconds self._nonces: OrderedDict[str, int] = OrderedDict() self._lock = threading.Lock() def verify( self, method: str, target: str, body: bytes, headers: Mapping[str, str], *, now: int | None = None, ) -> None: version = headers.get("X-SDR-Version", "") timestamp_text = headers.get("X-SDR-Timestamp", "") nonce = headers.get("X-SDR-Nonce", "") signature = headers.get("X-SDR-Signature", "") if version != str(PROTOCOL_VERSION): raise AuthenticationError("unsupported authentication version") if not timestamp_text.isdecimal(): raise AuthenticationError("invalid authentication timestamp") if not re.fullmatch(r"[0-9a-f]{32}", nonce): raise AuthenticationError("invalid authentication nonce") if not re.fullmatch(r"[0-9a-f]{64}", signature): raise AuthenticationError("invalid authentication signature") current = int(time.time()) if now is None else now request_time = int(timestamp_text) if abs(current - request_time) > self.window_seconds: raise AuthenticationError("authentication timestamp is outside the window") expected = hmac.new( self.token, _signature_payload(method, target, timestamp_text, nonce, body), hashlib.sha256, ).hexdigest() if not hmac.compare_digest(signature, expected): raise AuthenticationError("authentication signature mismatch") cutoff = current - self.window_seconds with self._lock: while self._nonces: oldest_nonce, oldest_time = next(iter(self._nonces.items())) if oldest_time >= cutoff: break self._nonces.pop(oldest_nonce) if nonce in self._nonces: raise AuthenticationError("authentication nonce was already used") self._nonces[nonce] = request_time def _validate_agent_id(value: Any) -> str: if not isinstance(value, str) or AGENT_ID_PATTERN.fullmatch(value) is None: raise ProtocolError("invalid agent_id") return value def _validate_task_id(value: Any) -> str: if not isinstance(value, str) or TASK_ID_PATTERN.fullmatch(value) is None: raise ProtocolError("invalid task_id") return value def _require_mapping(value: Any, field: str) -> dict[str, Any]: if not isinstance(value, dict): raise ProtocolError(f"{field} must be a JSON object") return value def _validate_session_id(value: Any) -> str: if not isinstance(value, str) or SESSION_ID_PATTERN.fullmatch(value) is None: raise ProtocolError("invalid session_id") return value def _validate_timestamp(value: Any, field: str) -> str: if not isinstance(value, str): raise ProtocolError(f"{field} must be a timestamp string") try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as error: raise ProtocolError(f"{field} is not an ISO timestamp") from error if parsed.tzinfo is None: raise ProtocolError(f"{field} must contain a timezone") return parsed.astimezone(timezone.utc).isoformat() @dataclass(frozen=True) class InventoryTask: task_id: str agent_id: str created_at: str include_hackrf_info: bool def as_mapping(self) -> dict[str, Any]: return { "protocol_version": PROTOCOL_VERSION, "task_id": self.task_id, "kind": INVENTORY_TASK, "agent_id": self.agent_id, "created_at": self.created_at, "include_hackrf_info": self.include_hackrf_info, "rf_tx": 0, } @dataclass(frozen=True) class OfflineVerifyTask: task_id: str agent_id: str created_at: str expected_head: str def as_mapping(self) -> dict[str, Any]: return { "protocol_version": PROTOCOL_VERSION, "task_id": self.task_id, "kind": OFFLINE_VERIFY_TASK, "agent_id": self.agent_id, "created_at": self.created_at, "expected_head": self.expected_head, "rf_tx": 0, } @dataclass(frozen=True) class RxCaptureStartTask: task_id: str agent_id: str created_at: str expected_head: str session_id: str def as_mapping(self) -> dict[str, Any]: return { "protocol_version": PROTOCOL_VERSION, "task_id": self.task_id, "kind": RX_CAPTURE_START_TASK, "agent_id": self.agent_id, "created_at": self.created_at, "expected_head": self.expected_head, "session_id": self.session_id, "rx_serial": FIXED_RX_SERIAL, "rx_authorized": True, "automatic_retry": False, "rf_tx": 0, } @dataclass(frozen=True) class RxCaptureStatusTask: task_id: str agent_id: str created_at: str session_id: str def as_mapping(self) -> dict[str, Any]: return { "protocol_version": PROTOCOL_VERSION, "task_id": self.task_id, "kind": RX_CAPTURE_STATUS_TASK, "agent_id": self.agent_id, "created_at": self.created_at, "session_id": self.session_id, "rf_tx": 0, } @dataclass(frozen=True) class CaptureDecodeTask: task_id: str agent_id: str created_at: str session_id: str tx_started_at: str tx_finished_at: str def as_mapping(self) -> dict[str, Any]: return { "protocol_version": PROTOCOL_VERSION, "task_id": self.task_id, "kind": CAPTURE_DECODE_TASK, "agent_id": self.agent_id, "created_at": self.created_at, "session_id": self.session_id, "tx_started_at": self.tx_started_at, "tx_finished_at": self.tx_finished_at, "rf_tx": 0, } @dataclass(frozen=True) class CaptureDiagnosticsTask: task_id: str agent_id: str created_at: str session_id: str tx_started_at: str tx_finished_at: str def as_mapping(self) -> dict[str, Any]: return { "protocol_version": PROTOCOL_VERSION, "task_id": self.task_id, "kind": CAPTURE_DIAGNOSTICS_TASK, "agent_id": self.agent_id, "created_at": self.created_at, "session_id": self.session_id, "tx_started_at": self.tx_started_at, "tx_finished_at": self.tx_finished_at, "rf_tx": 0, } @dataclass(frozen=True) class StaircaseSearchTask: task_id: str agent_id: str created_at: str session_id: str expected_rx_sha256: str expected_reference_sha256: str def as_mapping(self) -> dict[str, Any]: return { "protocol_version": PROTOCOL_VERSION, "task_id": self.task_id, "kind": STAIRCASE_SEARCH_TASK, "agent_id": self.agent_id, "created_at": self.created_at, "session_id": self.session_id, "expected_rx_sha256": self.expected_rx_sha256, "expected_reference_sha256": self.expected_reference_sha256, "hardware_accessed": False, "automatic_retry": False, "rf_tx": 0, } @dataclass(frozen=True) class UpdateAgentTask: task_id: str agent_id: str created_at: str release_id: str agent_version: str package_sha256: str package_bytes: int content_manifest_sha256: str compatible_protocol_version: int approval_expires_at: str def as_mapping(self) -> dict[str, Any]: return { "protocol_version": PROTOCOL_VERSION, "task_id": self.task_id, "kind": UPDATE_AGENT_TASK, "agent_id": self.agent_id, "created_at": self.created_at, "release_id": self.release_id, "agent_version": self.agent_version, "package_sha256": self.package_sha256, "package_bytes": self.package_bytes, "content_manifest_sha256": self.content_manifest_sha256, "compatible_protocol_version": self.compatible_protocol_version, "approval_expires_at": self.approval_expires_at, "hardware_accessed": False, "automatic_retry": False, "rf_tx": 0, } class ControlState: """Thread-safe single-agent task state persisted as atomic JSON.""" def __init__(self, path: Path) -> None: self.path = path self._lock = threading.Lock() self._data = self._load() def _load(self) -> dict[str, Any]: if not self.path.exists(): return { "protocol_version": PROTOCOL_VERSION, "agents": {}, "tasks": [], "results": {}, "releases": {}, "bootstrap_approvals": {}, } raw = json.loads(self.path.read_text(encoding="utf-8-sig")) data = _require_mapping(raw, "state") if data.get("protocol_version") != PROTOCOL_VERSION: raise ProtocolError("unsupported saved state version") if not isinstance(data.get("agents"), dict): raise ProtocolError("saved agents state is invalid") if not isinstance(data.get("tasks"), list): raise ProtocolError("saved tasks state is invalid") if not isinstance(data.get("results"), dict): raise ProtocolError("saved results state is invalid") data.setdefault("releases", {}) data.setdefault("bootstrap_approvals", {}) if not isinstance(data["releases"], dict): raise ProtocolError("saved releases state is invalid") if not isinstance(data["bootstrap_approvals"], dict): raise ProtocolError("saved bootstrap approvals state is invalid") return data def _save(self) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) temporary = self.path.with_suffix(self.path.suffix + ".tmp") temporary.write_bytes(canonical_json(self._data) + b"\n") os.replace(temporary, self.path) def enqueue_inventory( self, agent_id: str, *, include_hackrf_info: bool, ) -> dict[str, Any]: agent = _validate_agent_id(agent_id) created_at = utc_now() compact_time = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") task_id = f"INVENTORY_{compact_time}_{secrets.token_hex(4).upper()}" task = InventoryTask( task_id=task_id, agent_id=agent, created_at=created_at, include_hackrf_info=include_hackrf_info, ).as_mapping() record = { **task, "state": "QUEUED", "leased_at_epoch": None, "completed_at": None, } with self._lock: self._data["tasks"].append(record) self._save() return task def enqueue_offline_verify( self, agent_id: str, *, expected_head: str, ) -> dict[str, Any]: agent = _validate_agent_id(agent_id) if GIT_HEAD_PATTERN.fullmatch(expected_head) is None: raise ProtocolError("expected_head must be a lowercase 40-character hash") created_at = utc_now() compact_time = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") task_id = f"OFFLINE_{compact_time}_{secrets.token_hex(4).upper()}" task = OfflineVerifyTask( task_id=task_id, agent_id=agent, created_at=created_at, expected_head=expected_head, ).as_mapping() record = { **task, "state": "QUEUED", "leased_at_epoch": None, "completed_at": None, } with self._lock: self._data["tasks"].append(record) self._save() return task def _append_task(self, task: dict[str, Any]) -> dict[str, Any]: record = { **task, "state": "QUEUED", "leased_at_epoch": None, "completed_at": None, } with self._lock: self._data["tasks"].append(record) self._save() return task def enqueue_rx_capture( self, agent_id: str, *, expected_head: str, authorization: str, ) -> dict[str, Any]: agent = _validate_agent_id(agent_id) if GIT_HEAD_PATTERN.fullmatch(expected_head) is None: raise ProtocolError("expected_head must be a lowercase 40-character hash") if authorization != RX_AUTHORIZATION_PHRASE: raise ProtocolError("exact RX authorization phrase is required") compact_time = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") suffix = secrets.token_hex(4).upper() task = RxCaptureStartTask( task_id=f"RXSTART_{compact_time}_{suffix}", agent_id=agent, created_at=utc_now(), expected_head=expected_head, session_id=f"RF001A_{compact_time}_{suffix}", ).as_mapping() return self._append_task(task) def enqueue_rx_status( self, agent_id: str, *, session_id: str, ) -> dict[str, Any]: agent = _validate_agent_id(agent_id) session = _validate_session_id(session_id) compact_time = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") task = RxCaptureStatusTask( task_id=f"RXSTATUS_{compact_time}_{secrets.token_hex(4).upper()}", agent_id=agent, created_at=utc_now(), session_id=session, ).as_mapping() return self._append_task(task) def enqueue_capture_decode( self, agent_id: str, *, session_id: str, tx_started_at: str, tx_finished_at: str, ) -> dict[str, Any]: agent = _validate_agent_id(agent_id) session = _validate_session_id(session_id) started = _validate_timestamp(tx_started_at, "tx_started_at") finished = _validate_timestamp(tx_finished_at, "tx_finished_at") if datetime.fromisoformat(finished) < datetime.fromisoformat(started): raise ProtocolError("tx_finished_at precedes tx_started_at") compact_time = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") task = CaptureDecodeTask( task_id=f"DECODE_{compact_time}_{secrets.token_hex(4).upper()}", agent_id=agent, created_at=utc_now(), session_id=session, tx_started_at=started, tx_finished_at=finished, ).as_mapping() return self._append_task(task) def enqueue_capture_diagnostics( self, agent_id: str, *, session_id: str, tx_started_at: str, tx_finished_at: str, ) -> dict[str, Any]: agent = _validate_agent_id(agent_id) session = _validate_session_id(session_id) started = _validate_timestamp(tx_started_at, "tx_started_at") finished = _validate_timestamp(tx_finished_at, "tx_finished_at") if datetime.fromisoformat(finished) < datetime.fromisoformat(started): raise ProtocolError("tx_finished_at precedes tx_started_at") compact_time = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") task = CaptureDiagnosticsTask( task_id=f"DIAG_{compact_time}_{secrets.token_hex(4).upper()}", agent_id=agent, created_at=utc_now(), session_id=session, tx_started_at=started, tx_finished_at=finished, ).as_mapping() return self._append_task(task) def enqueue_staircase_search( self, agent_id: str, *, session_id: str, expected_rx_sha256: str, expected_reference_sha256: str, ) -> dict[str, Any]: agent = _validate_agent_id(agent_id) session = _validate_session_id(session_id) if SHA256_PATTERN.fullmatch(expected_rx_sha256) is None: raise ProtocolError("expected_rx_sha256 is invalid") if SHA256_PATTERN.fullmatch(expected_reference_sha256) is None: raise ProtocolError("expected_reference_sha256 is invalid") compact_time = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") task = StaircaseSearchTask( task_id=( f"STAIRCASE_{compact_time}_{secrets.token_hex(4).upper()}" ), agent_id=agent, created_at=utc_now(), session_id=session, expected_rx_sha256=expected_rx_sha256, expected_reference_sha256=expected_reference_sha256, ).as_mapping() return self._append_task(task) @staticmethod def _release_path(package_path: str) -> Path: candidate = Path(package_path).resolve() releases_root = DEFAULT_RELEASES_ROOT.resolve() try: candidate.relative_to(releases_root) except ValueError as error: raise ProtocolError("release package must be inside .local/releases") from error if not candidate.is_file(): raise ProtocolError("release package is absent") return candidate def register_agent_release( self, package_path: str, *, token: bytes, ) -> dict[str, Any]: candidate = self._release_path(package_path) try: inspection = inspect_release_archive(candidate) except (OSError, ReleaseValidationError) as error: raise ProtocolError(f"release package is invalid: {error}") from error manifest = {**inspection.descriptor(), "package_kind": "AGENT_RELEASE"} signature = sign_update_manifest(token, manifest) record = { "manifest": manifest, "manifest_signature": signature, "package_path": str(candidate), "registered_at": utc_now(), } with self._lock: previous = self._data["releases"].get(inspection.release_id) if previous is not None and canonical_json(previous) != canonical_json(record): raise ProtocolError("release_id is already registered differently") self._data["releases"][inspection.release_id] = record self._save() return { "release_id": inspection.release_id, "agent_version": inspection.agent_version, "package_sha256": inspection.package_sha256, "package_bytes": inspection.package_bytes, "package_kind": "AGENT_RELEASE", "registered": True, "rf_tx": 0, } def register_bootstrap_release( self, package_path: str, *, release_id: str, agent_version: str, content_manifest_sha256: str, token: bytes, ) -> dict[str, Any]: candidate = self._release_path(package_path) if BOOTSTRAP_RELEASE_ID_PATTERN.fullmatch(release_id) is None: raise ProtocolError("release_id is invalid") if re.fullmatch(r"[0-9]{4}\.[0-9]{2}\.[0-9]{2}\.[0-9]+", agent_version) is None: raise ProtocolError("agent_version is invalid") if SHA256_PATTERN.fullmatch(content_manifest_sha256) is None: raise ProtocolError("content_manifest_sha256 is invalid") digest = hashlib.sha256() with candidate.open("rb") as stream: while block := stream.read(4 * 1024 * 1024): digest.update(block) manifest = { "schema_version": 1, "release_id": release_id, "agent_version": agent_version, "protocol_version": PROTOCOL_VERSION, "package_kind": "BOOTSTRAP", "package_sha256": digest.hexdigest(), "package_bytes": candidate.stat().st_size, "content_manifest_sha256": content_manifest_sha256, } record = { "manifest": manifest, "manifest_signature": sign_update_manifest(token, manifest), "package_path": str(candidate), "registered_at": utc_now(), } with self._lock: previous = self._data["releases"].get(release_id) if previous is not None and canonical_json(previous) != canonical_json(record): raise ProtocolError("release_id is already registered differently") self._data["releases"][release_id] = record self._save() return {**manifest, "registered": True, "rf_tx": 0} def enqueue_agent_update( self, agent_id: str, *, release_id: str, validity_seconds: int, ) -> dict[str, Any]: agent = _validate_agent_id(agent_id) if not 60 <= validity_seconds <= 86_400: raise ProtocolError("update approval validity must be 60 to 86400 seconds") with self._lock: release = self._data["releases"].get(release_id) if not isinstance(release, dict): raise ProtocolError("release_id is not registered") manifest = _require_mapping(release.get("manifest"), "release manifest") if manifest.get("package_kind") != "AGENT_RELEASE": raise ProtocolError("bootstrap package cannot be used for remote update") if manifest.get("protocol_version") != PROTOCOL_VERSION: raise ProtocolError("release protocol version is incompatible") existing = [ task for task in self._data["tasks"] if task.get("kind") == UPDATE_AGENT_TASK and task.get("agent_id") == agent and task.get("release_id") == release_id and task.get("state") not in {"CANCELLED", "LEASE_EXPIRED"} ] if existing: raise ProtocolError("this release already has an update task") compact_time = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") expires = datetime.fromtimestamp( time.time() + validity_seconds, timezone.utc, ).isoformat() task = UpdateAgentTask( task_id=f"UPDATE_{compact_time}_{secrets.token_hex(4).upper()}", agent_id=agent, created_at=utc_now(), release_id=release_id, agent_version=str(manifest["agent_version"]), package_sha256=str(manifest["package_sha256"]), package_bytes=int(manifest["package_bytes"]), content_manifest_sha256=str( manifest["content_manifest_sha256"] ), compatible_protocol_version=int(manifest["protocol_version"]), approval_expires_at=expires, ).as_mapping() record = { **task, "state": "QUEUED", "leased_at_epoch": None, "completed_at": None, } self._data["tasks"].append(record) self._save() return task def approve_bootstrap_download( self, release_id: str, *, validity_seconds: int, ) -> dict[str, Any]: if not 60 <= validity_seconds <= 3_600: raise ProtocolError("bootstrap validity must be 60 to 3600 seconds") ticket = secrets.token_urlsafe(32) with self._lock: release = self._data["releases"].get(release_id) if not isinstance(release, dict): raise ProtocolError("release_id is not registered") manifest = _require_mapping(release.get("manifest"), "release manifest") if manifest.get("package_kind") != "BOOTSTRAP": raise ProtocolError("release is not a bootstrap package") self._data["bootstrap_approvals"][release_id] = { "ticket_sha256": hashlib.sha256(ticket.encode("ascii")).hexdigest(), "expires_at_epoch": int(time.time()) + validity_seconds, "attempts": 0, "maximum_attempts": 2, "created_at": utc_now(), } self._save() return { "release_id": release_id, "ticket": ticket, "expires_in_seconds": validity_seconds, "maximum_download_attempts": 2, "rf_tx": 0, } def acquire_bootstrap_download( self, release_id: str, ticket: str, ) -> dict[str, Any]: with self._lock: approval = self._data["bootstrap_approvals"].get(release_id) release = self._data["releases"].get(release_id) if not isinstance(approval, dict) or not isinstance(release, dict): raise AuthenticationError("bootstrap download is not approved") expected = str(approval.get("ticket_sha256", "")) actual = hashlib.sha256(ticket.encode("ascii", errors="ignore")).hexdigest() if not hmac.compare_digest(expected, actual): raise AuthenticationError("bootstrap download ticket is invalid") if int(approval.get("expires_at_epoch", 0)) < int(time.time()): raise AuthenticationError("bootstrap download ticket expired") attempts = int(approval.get("attempts", 0)) maximum = int(approval.get("maximum_attempts", 0)) if attempts >= maximum: raise AuthenticationError("bootstrap download attempt limit reached") approval["attempts"] = attempts + 1 approval["last_attempt_at"] = utc_now() self._save() return json.loads(canonical_json(release)) def approved_agent_release( self, agent_id: str, release_id: str, ) -> dict[str, Any]: agent = _validate_agent_id(agent_id) with self._lock: approved = [ task for task in self._data["tasks"] if task.get("kind") == UPDATE_AGENT_TASK and task.get("agent_id") == agent and task.get("release_id") == release_id and task.get("state") in {"QUEUED", "LEASED"} and datetime.fromisoformat(str(task.get("approval_expires_at"))) >= datetime.now(timezone.utc) ] release = self._data["releases"].get(release_id) if len(approved) != 1 or not isinstance(release, dict): raise AuthenticationError("agent release download is not approved") return json.loads(canonical_json(release)) def check_in(self, payload: Mapping[str, Any]) -> dict[str, Any]: if payload.get("protocol_version") != PROTOCOL_VERSION: raise ProtocolError("unsupported protocol version") agent_id = _validate_agent_id(payload.get("agent_id")) if payload.get("rf_tx") != 0: raise ProtocolError("agent check-in must report rf_tx = 0") inventory = _require_mapping(payload.get("inventory"), "inventory") active_payload = payload.get("active_task") active_task: dict[str, Any] | None = None if active_payload is not None: active_task = _require_mapping(active_payload, "active_task") _validate_task_id(active_task.get("task_id")) if active_task.get("state") not in { "STARTING", "RUNNING", "RESULT_PENDING", }: raise ProtocolError("active task state is invalid") processed = active_task.get("processed_samples") total = active_task.get("total_samples") if type(processed) is not int or type(total) is not int: raise ProtocolError("active task progress must use integers") if processed < 0 or total <= 0 or processed > total: raise ProtocolError("active task progress is invalid") if active_task.get("rf_tx") != 0: raise ProtocolError("active task must report rf_tx = 0") now_epoch = int(time.time()) with self._lock: self._data["agents"][agent_id] = { "last_seen_at": utc_now(), "remote_inventory": inventory, "active_task": active_task, "rf_tx": 0, } selected: dict[str, Any] | None = None active_matched = False if active_task is not None: matches = [ task for task in self._data["tasks"] if task.get("task_id") == active_task.get("task_id") and task.get("agent_id") == agent_id ] if len(matches) != 1 or matches[0].get("state") != "LEASED": raise ProtocolError("active task does not match a leased task") matches[0]["leased_at_epoch"] = now_epoch matches[0]["progress"] = active_task active_matched = True for pending in self._data["tasks"]: if ( pending.get("kind") == UPDATE_AGENT_TASK and pending.get("agent_id") == agent_id and pending.get("state") == "QUEUED" ): pending["update_status"] = "WAITING_FOR_IDLE" pending["blocked_by_task_id"] = active_task["task_id"] pending["status_updated_at"] = utc_now() for task in ([] if active_matched else self._data["tasks"]): if task.get("agent_id") != agent_id: continue state = task.get("state") leased_at = task.get("leased_at_epoch") expired = ( state == "LEASED" and isinstance(leased_at, int) and now_epoch - leased_at > TASK_LEASE_SECONDS ) if expired and task.get("automatic_retry") is False: task["state"] = "LEASE_EXPIRED" task["lease_expired_at"] = utc_now() task["completed_at"] = task["lease_expired_at"] continue if state == "QUEUED" or expired: task["state"] = "LEASED" task["leased_at_epoch"] = now_epoch selected = { key: value for key, value in task.items() if key not in {"state", "leased_at_epoch", "completed_at"} } break self._save() return { "protocol_version": PROTOCOL_VERSION, "server_time": utc_now(), "task": selected, "rf_tx": 0, } def record_result(self, payload: Mapping[str, Any]) -> dict[str, Any]: if payload.get("protocol_version") != PROTOCOL_VERSION: raise ProtocolError("unsupported protocol version") task_id = _validate_task_id(payload.get("task_id")) agent_id = _validate_agent_id(payload.get("agent_id")) if payload.get("rf_tx") != 0: raise ProtocolError("result must report rf_tx = 0") if payload.get("status") not in {"COMPLETE", "FAILED"}: raise ProtocolError("invalid result status") _require_mapping(payload.get("result"), "result") errors = payload.get("errors") if not isinstance(errors, list) or not all( isinstance(item, str) for item in errors ): raise ProtocolError("errors must be a list of strings") serialized = canonical_json(payload) if len(serialized) > MAX_JSON_BYTES: raise ProtocolError("result is too large") with self._lock: matching = [ task for task in self._data["tasks"] if task.get("task_id") == task_id ] if len(matching) != 1 or matching[0].get("agent_id") != agent_id: raise ProtocolError("result does not match a known task") if matching[0].get("state") == "CANCELLED": raise ProtocolError("result belongs to a cancelled task") previous = self._data["results"].get(task_id) if previous is not None: if canonical_json(previous) != serialized: raise ProtocolError("conflicting duplicate result") return {"accepted": True, "duplicate": True, "rf_tx": 0} self._data["results"][task_id] = dict(payload) matching[0]["state"] = "COMPLETE" matching[0]["completed_at"] = utc_now() self._save() return {"accepted": True, "duplicate": False, "rf_tx": 0} def cancel_task(self, task_id: str, *, reason: str) -> dict[str, Any]: task_identifier = _validate_task_id(task_id) if not isinstance(reason, str) or not reason.strip() or len(reason) > 500: raise ProtocolError("cancellation reason must contain 1 to 500 characters") with self._lock: matching = [ task for task in self._data["tasks"] if task.get("task_id") == task_identifier ] if len(matching) != 1: raise ProtocolError("task to cancel was not found") task = matching[0] if task.get("state") == "COMPLETE": raise ProtocolError("completed task cannot be cancelled") if task.get("state") != "CANCELLED": task["state"] = "CANCELLED" task["cancelled_at"] = utc_now() task["cancellation_reason"] = reason.strip() task["completed_at"] = task["cancelled_at"] self._save() return { "task_id": task_identifier, "state": "CANCELLED", "cancellation_reason": task.get("cancellation_reason"), "rf_tx": 0, } def snapshot(self) -> dict[str, Any]: with self._lock: return json.loads(canonical_json(self._data)) class ControlHTTPServer(ThreadingHTTPServer): daemon_threads = True def __init__( self, server_address: tuple[str, int], *, state: ControlState, token: bytes, allowed_agent_ip: str, ) -> None: super().__init__(server_address, ControlRequestHandler) self.state = state self.token = token self.authenticator = RequestAuthenticator(token) self.allowed_agent_ip = allowed_agent_ip class ControlRequestHandler(BaseHTTPRequestHandler): server: ControlHTTPServer def log_message(self, format: str, *args: object) -> None: message = format % args print(f"HTTP {self.client_address[0]} {message}") def _send_json(self, status: HTTPStatus, payload: Mapping[str, Any]) -> None: body = canonical_json(payload) + b"\n" self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _send_package(self, release: Mapping[str, Any]) -> None: package_path = Path(str(release.get("package_path", ""))) manifest = _require_mapping(release.get("manifest"), "release manifest") expected_size = int(manifest.get("package_bytes", -1)) if not package_path.is_file() or package_path.stat().st_size != expected_size: raise ProtocolError("registered release package is unavailable") self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "application/zip") self.send_header("Content-Length", str(expected_size)) self.send_header("Cache-Control", "no-store") self.send_header( "Content-Disposition", f'attachment; filename="{manifest["release_id"]}.zip"', ) self.end_headers() with package_path.open("rb") as stream: while block := stream.read(1024 * 1024): self.wfile.write(block) def _read_body(self) -> bytes: length_text = self.headers.get("Content-Length", "") if not length_text.isdecimal(): raise ProtocolError("valid Content-Length is required") length = int(length_text) if length > MAX_JSON_BYTES: raise ProtocolError("request body is too large") return self.rfile.read(length) def _read_authenticated_json(self) -> dict[str, Any]: body = self._read_body() self.server.authenticator.verify( self.command, self.path, body, self.headers, ) try: decoded = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as error: raise ProtocolError("request body is not valid UTF-8 JSON") from error return _require_mapping(decoded, "request") def _require_agent_source(self) -> None: if self.client_address[0] != self.server.allowed_agent_ip: raise AuthenticationError("agent source IP is not allowed") def _require_local_source(self) -> None: bound_address = self.server.server_address[0] allowed = {"127.0.0.1", "::1"} if bound_address not in {"0.0.0.0", "::"}: allowed.add(bound_address) if self.client_address[0] not in allowed: raise AuthenticationError("local endpoint requires the server host") def do_GET(self) -> None: # noqa: N802 try: split = urlsplit(self.path) parts = split.path.strip("/").split("/") if self.path == "/api/v1/health": self._send_json( HTTPStatus.OK, { "status": "OK", "protocol_version": PROTOCOL_VERSION, "server_time": utc_now(), "rf_tx": 0, }, ) return if self.path == "/api/v1/local/status": self._require_local_source() self.server.authenticator.verify( self.command, self.path, b"", self.headers, ) self._send_json(HTTPStatus.OK, self.server.state.snapshot()) return if ( len(parts) == 6 and parts[:4] == ["api", "v1", "agent", "releases"] and parts[5] in {"manifest", "package"} ): self._require_agent_source() self.server.authenticator.verify( self.command, self.path, b"", self.headers, ) release_id = parts[4] query = parse_qs(split.query, strict_parsing=True) agent_values = query.get("agent_id", []) if len(agent_values) != 1: raise ProtocolError("agent_id query parameter is required") release = self.server.state.approved_agent_release( agent_values[0], release_id, ) if parts[5] == "manifest": self._send_json( HTTPStatus.OK, { "manifest": release["manifest"], "manifest_signature": release["manifest_signature"], "rf_tx": 0, }, ) else: self._send_package(release) return if ( len(parts) == 6 and parts[:4] == ["api", "v1", "bootstrap", "releases"] and parts[5] == "package" ): self._require_agent_source() query = parse_qs(split.query, strict_parsing=True) tickets = query.get("ticket", []) if len(tickets) != 1: raise AuthenticationError("bootstrap ticket is required") release = self.server.state.acquire_bootstrap_download( parts[4], tickets[0], ) self._send_package(release) return self._send_json(HTTPStatus.NOT_FOUND, {"error": "not found"}) except AuthenticationError as error: self._send_json(HTTPStatus.UNAUTHORIZED, {"error": str(error)}) except ProtocolError as error: self._send_json(HTTPStatus.BAD_REQUEST, {"error": str(error)}) def do_POST(self) -> None: # noqa: N802 try: if self.path == "/api/v1/agent/check-in": self._require_agent_source() payload = self._read_authenticated_json() response = self.server.state.check_in(payload) self._send_json(HTTPStatus.OK, response) return if self.path == "/api/v1/agent/result": self._require_agent_source() payload = self._read_authenticated_json() response = self.server.state.record_result(payload) self._send_json(HTTPStatus.OK, response) return if self.path == "/api/v1/local/tasks/inventory": self._require_local_source() payload = self._read_authenticated_json() include = payload.get("include_hackrf_info", True) if type(include) is not bool: raise ProtocolError("include_hackrf_info must be boolean") task = self.server.state.enqueue_inventory( _validate_agent_id(payload.get("agent_id")), include_hackrf_info=include, ) self._send_json(HTTPStatus.CREATED, task) return if self.path == "/api/v1/local/tasks/offline-verify": self._require_local_source() payload = self._read_authenticated_json() task = self.server.state.enqueue_offline_verify( _validate_agent_id(payload.get("agent_id")), expected_head=str(payload.get("expected_head", "")), ) self._send_json(HTTPStatus.CREATED, task) return if self.path == "/api/v1/local/tasks/rx-capture": self._require_local_source() payload = self._read_authenticated_json() task = self.server.state.enqueue_rx_capture( _validate_agent_id(payload.get("agent_id")), expected_head=str(payload.get("expected_head", "")), authorization=str(payload.get("authorization", "")), ) self._send_json(HTTPStatus.CREATED, task) return if self.path == "/api/v1/local/tasks/rx-status": self._require_local_source() payload = self._read_authenticated_json() task = self.server.state.enqueue_rx_status( _validate_agent_id(payload.get("agent_id")), session_id=str(payload.get("session_id", "")), ) self._send_json(HTTPStatus.CREATED, task) return if self.path == "/api/v1/local/tasks/capture-decode": self._require_local_source() payload = self._read_authenticated_json() task = self.server.state.enqueue_capture_decode( _validate_agent_id(payload.get("agent_id")), session_id=str(payload.get("session_id", "")), tx_started_at=str(payload.get("tx_started_at", "")), tx_finished_at=str(payload.get("tx_finished_at", "")), ) self._send_json(HTTPStatus.CREATED, task) return if self.path == "/api/v1/local/tasks/capture-diagnostics": self._require_local_source() payload = self._read_authenticated_json() task = self.server.state.enqueue_capture_diagnostics( _validate_agent_id(payload.get("agent_id")), session_id=str(payload.get("session_id", "")), tx_started_at=str(payload.get("tx_started_at", "")), tx_finished_at=str(payload.get("tx_finished_at", "")), ) self._send_json(HTTPStatus.CREATED, task) return if self.path == "/api/v1/local/tasks/staircase-search": self._require_local_source() payload = self._read_authenticated_json() task = self.server.state.enqueue_staircase_search( _validate_agent_id(payload.get("agent_id")), session_id=str(payload.get("session_id", "")), expected_rx_sha256=str( payload.get("expected_rx_sha256", "") ), expected_reference_sha256=str( payload.get("expected_reference_sha256", "") ), ) self._send_json(HTTPStatus.CREATED, task) return if self.path == "/api/v1/local/releases/register-agent": self._require_local_source() payload = self._read_authenticated_json() release = self.server.state.register_agent_release( str(payload.get("package_path", "")), token=self.server.token, ) self._send_json(HTTPStatus.CREATED, release) return if self.path == "/api/v1/local/releases/register-bootstrap": self._require_local_source() payload = self._read_authenticated_json() release = self.server.state.register_bootstrap_release( str(payload.get("package_path", "")), release_id=str(payload.get("release_id", "")), agent_version=str(payload.get("agent_version", "")), content_manifest_sha256=str( payload.get("content_manifest_sha256", "") ), token=self.server.token, ) self._send_json(HTTPStatus.CREATED, release) return if self.path == "/api/v1/local/tasks/update-agent": self._require_local_source() payload = self._read_authenticated_json() validity = payload.get("validity_seconds") if type(validity) is not int: raise ProtocolError("validity_seconds must be an integer") task = self.server.state.enqueue_agent_update( _validate_agent_id(payload.get("agent_id")), release_id=str(payload.get("release_id", "")), validity_seconds=validity, ) self._send_json(HTTPStatus.CREATED, task) return if self.path == "/api/v1/local/releases/approve-bootstrap": self._require_local_source() payload = self._read_authenticated_json() validity = payload.get("validity_seconds") if type(validity) is not int: raise ProtocolError("validity_seconds must be an integer") approval = self.server.state.approve_bootstrap_download( str(payload.get("release_id", "")), validity_seconds=validity, ) host, port = self.server.server_address[:2] approval["download_url"] = ( f"http://{host}:{port}/api/v1/bootstrap/releases/" f"{approval['release_id']}/package?ticket=" f"{quote(str(approval['ticket']), safe='')}" ) self._send_json(HTTPStatus.CREATED, approval) return if self.path == "/api/v1/local/tasks/cancel": self._require_local_source() payload = self._read_authenticated_json() response = self.server.state.cancel_task( str(payload.get("task_id", "")), reason=str(payload.get("reason", "")), ) self._send_json(HTTPStatus.OK, response) return self._send_json(HTTPStatus.NOT_FOUND, {"error": "not found"}) except AuthenticationError as error: self._send_json(HTTPStatus.UNAUTHORIZED, {"error": str(error)}) except ProtocolError as error: self._send_json(HTTPStatus.BAD_REQUEST, {"error": str(error)}) except Exception as error: # pragma: no cover - defensive server boundary self._send_json( HTTPStatus.INTERNAL_SERVER_ERROR, {"error": f"internal server error: {type(error).__name__}"}, ) def request_json( server_url: str, token: bytes, method: str, target: str, payload: Mapping[str, Any] | None = None, *, timeout: float = 10.0, ) -> dict[str, Any]: body = b"" if payload is None else canonical_json(payload) headers = signed_headers(token, method, target, body) request = Request( server_url.rstrip("/") + target, data=body if method.upper() != "GET" else None, headers=headers, method=method.upper(), ) opener = build_opener(ProxyHandler({})) try: with opener.open(request, timeout=timeout) as response: response_body = response.read(MAX_JSON_BYTES + 1) except HTTPError as error: detail = error.read(MAX_JSON_BYTES).decode("utf-8", errors="replace") raise RuntimeError(f"HTTP {error.code}: {detail}") from error if len(response_body) > MAX_JSON_BYTES: raise ProtocolError("server response is too large") decoded = json.loads(response_body.decode("utf-8")) return _require_mapping(decoded, "response") def request_file( server_url: str, token: bytes, target: str, destination: Path, *, expected_bytes: int, timeout: float = 30.0, ) -> None: if expected_bytes <= 0: raise ProtocolError("expected download size must be positive") headers = signed_headers(token, "GET", target, b"") request = Request( server_url.rstrip("/") + target, headers=headers, method="GET", ) opener = build_opener(ProxyHandler({})) destination.parent.mkdir(parents=True, exist_ok=True) try: with opener.open(request, timeout=timeout) as response: length = response.headers.get("Content-Length", "") if not length.isdecimal() or int(length) != expected_bytes: raise ProtocolError("download Content-Length mismatch") written = 0 with destination.open("xb") as stream: while block := response.read(min(1024 * 1024, expected_bytes - written + 1)): written += len(block) if written > expected_bytes: raise ProtocolError("download exceeded approved size") stream.write(block) if written != expected_bytes: raise ProtocolError("download ended before the approved size") except HTTPError as error: detail = error.read(MAX_JSON_BYTES).decode("utf-8", errors="replace") raise RuntimeError(f"HTTP {error.code}: {detail}") from error def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) token_parser = subparsers.add_parser("init-token") token_parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_PATH) serve_parser = subparsers.add_parser("serve") serve_parser.add_argument("--bind", default="192.168.255.185") serve_parser.add_argument("--port", type=int, default=8765) serve_parser.add_argument("--allowed-agent-ip", default="192.168.255.56") serve_parser.add_argument("--state-file", type=Path, default=DEFAULT_STATE_PATH) serve_parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_PATH) serve_parser.add_argument("--log-file", type=Path) serve_parser.add_argument("--bind-retry-seconds", type=float, default=0.0) enqueue_parser = subparsers.add_parser("enqueue-inventory") enqueue_parser.add_argument("--server", default=DEFAULT_SERVER_URL) enqueue_parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_PATH) enqueue_parser.add_argument("--agent-id", required=True) enqueue_parser.add_argument( "--without-hackrf-info", action="store_true", ) offline_parser = subparsers.add_parser("enqueue-offline-verify") offline_parser.add_argument("--server", default=DEFAULT_SERVER_URL) offline_parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_PATH) offline_parser.add_argument("--agent-id", required=True) offline_parser.add_argument("--expected-head", required=True) rx_parser = subparsers.add_parser("enqueue-rx-capture") rx_parser.add_argument("--server", default=DEFAULT_SERVER_URL) rx_parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_PATH) rx_parser.add_argument("--agent-id", required=True) rx_parser.add_argument("--expected-head", required=True) rx_parser.add_argument("--authorize-rx", required=True) rx_status_parser = subparsers.add_parser("enqueue-rx-status") rx_status_parser.add_argument("--server", default=DEFAULT_SERVER_URL) rx_status_parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_PATH) rx_status_parser.add_argument("--agent-id", required=True) rx_status_parser.add_argument("--session-id", required=True) decode_parser = subparsers.add_parser("enqueue-capture-decode") decode_parser.add_argument("--server", default=DEFAULT_SERVER_URL) decode_parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_PATH) decode_parser.add_argument("--agent-id", required=True) decode_parser.add_argument("--session-id", required=True) decode_parser.add_argument("--tx-started-at", required=True) decode_parser.add_argument("--tx-finished-at", required=True) diagnostics_parser = subparsers.add_parser("enqueue-capture-diagnostics") diagnostics_parser.add_argument("--server", default=DEFAULT_SERVER_URL) diagnostics_parser.add_argument( "--token-file", type=Path, default=DEFAULT_TOKEN_PATH ) diagnostics_parser.add_argument("--agent-id", required=True) diagnostics_parser.add_argument("--session-id", required=True) diagnostics_parser.add_argument("--tx-started-at", required=True) diagnostics_parser.add_argument("--tx-finished-at", required=True) staircase_parser = subparsers.add_parser("enqueue-staircase-search") staircase_parser.add_argument("--server", default=DEFAULT_SERVER_URL) staircase_parser.add_argument( "--token-file", type=Path, default=DEFAULT_TOKEN_PATH ) staircase_parser.add_argument("--agent-id", required=True) staircase_parser.add_argument("--session-id", required=True) staircase_parser.add_argument("--expected-rx-sha256", required=True) staircase_parser.add_argument("--expected-reference-sha256", required=True) register_release_parser = subparsers.add_parser("register-agent-release") register_release_parser.add_argument("--server", default=DEFAULT_SERVER_URL) register_release_parser.add_argument( "--token-file", type=Path, default=DEFAULT_TOKEN_PATH ) register_release_parser.add_argument("--package", required=True, type=Path) register_bootstrap_parser = subparsers.add_parser("register-bootstrap-release") register_bootstrap_parser.add_argument("--server", default=DEFAULT_SERVER_URL) register_bootstrap_parser.add_argument( "--token-file", type=Path, default=DEFAULT_TOKEN_PATH ) register_bootstrap_parser.add_argument("--package", required=True, type=Path) register_bootstrap_parser.add_argument("--release-id", required=True) register_bootstrap_parser.add_argument("--agent-version", required=True) register_bootstrap_parser.add_argument( "--content-manifest-sha256", required=True ) update_parser = subparsers.add_parser("enqueue-agent-update") update_parser.add_argument("--server", default=DEFAULT_SERVER_URL) update_parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_PATH) update_parser.add_argument("--agent-id", required=True) update_parser.add_argument("--release-id", required=True) update_parser.add_argument("--validity-seconds", type=int, default=3600) bootstrap_parser = subparsers.add_parser("approve-bootstrap-download") bootstrap_parser.add_argument("--server", default=DEFAULT_SERVER_URL) bootstrap_parser.add_argument( "--token-file", type=Path, default=DEFAULT_TOKEN_PATH ) bootstrap_parser.add_argument("--release-id", required=True) bootstrap_parser.add_argument("--validity-seconds", type=int, default=1800) cancel_parser = subparsers.add_parser("cancel-task") cancel_parser.add_argument("--server", default=DEFAULT_SERVER_URL) cancel_parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_PATH) cancel_parser.add_argument("--task-id", required=True) cancel_parser.add_argument("--reason", required=True) status_parser = subparsers.add_parser("status") status_parser.add_argument("--server", default=DEFAULT_SERVER_URL) status_parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_PATH) return parser def main() -> None: arguments = build_parser().parse_args() if arguments.command == "serve" and arguments.log_file is not None: redirect_output(arguments.log_file) if arguments.command == "init-token": initialize_token(arguments.token_file) print(f"TOKEN_FILE = {arguments.token_file}") print("TOKEN_CREATED = YES") print("RF TX = 0") return token = read_token(arguments.token_file) if arguments.command == "serve": if arguments.bind_retry_seconds < 0: raise ValueError("bind retry interval cannot be negative") state = ControlState(arguments.state_file) deadline = time.monotonic() + arguments.bind_retry_seconds waiting_for_address = False while True: try: server = ControlHTTPServer( (arguments.bind, arguments.port), state=state, token=token, allowed_agent_ip=arguments.allowed_agent_ip, ) break except OSError as error: address_unavailable = getattr(error, "winerror", None) == 10049 retry_open = time.monotonic() < deadline if not address_unavailable or not retry_open: raise if not waiting_for_address: print( f"WAITING FOR LOCAL ADDRESS = {arguments.bind} " f"({arguments.bind_retry_seconds:g} seconds maximum)" ) print("RF TX = 0") waiting_for_address = True time.sleep(min(5.0, max(0.1, deadline - time.monotonic()))) print(f"SERVER = http://{arguments.bind}:{arguments.port}") print(f"ALLOWED AGENT IP = {arguments.allowed_agent_ip}") print( "TASK KINDS = INVENTORY, OFFLINE_VERIFY, RX_CAPTURE_START, " "RX_CAPTURE_STATUS, CAPTURE_DECODE, CAPTURE_DIAGNOSTICS, " "STAIRCASE_SEARCH, UPDATE_AGENT" ) print("RF TX = 0") try: server.serve_forever(poll_interval=0.2) except KeyboardInterrupt: pass finally: server.server_close() return if arguments.command == "enqueue-inventory": response = request_json( arguments.server, token, "POST", "/api/v1/local/tasks/inventory", { "agent_id": arguments.agent_id, "include_hackrf_info": not arguments.without_hackrf_info, }, ) elif arguments.command == "enqueue-offline-verify": response = request_json( arguments.server, token, "POST", "/api/v1/local/tasks/offline-verify", { "agent_id": arguments.agent_id, "expected_head": arguments.expected_head, }, ) elif arguments.command == "enqueue-rx-capture": response = request_json( arguments.server, token, "POST", "/api/v1/local/tasks/rx-capture", { "agent_id": arguments.agent_id, "expected_head": arguments.expected_head, "authorization": arguments.authorize_rx, }, ) elif arguments.command == "enqueue-rx-status": response = request_json( arguments.server, token, "POST", "/api/v1/local/tasks/rx-status", { "agent_id": arguments.agent_id, "session_id": arguments.session_id, }, ) elif arguments.command == "enqueue-capture-decode": response = request_json( arguments.server, token, "POST", "/api/v1/local/tasks/capture-decode", { "agent_id": arguments.agent_id, "session_id": arguments.session_id, "tx_started_at": arguments.tx_started_at, "tx_finished_at": arguments.tx_finished_at, }, ) elif arguments.command == "enqueue-capture-diagnostics": response = request_json( arguments.server, token, "POST", "/api/v1/local/tasks/capture-diagnostics", { "agent_id": arguments.agent_id, "session_id": arguments.session_id, "tx_started_at": arguments.tx_started_at, "tx_finished_at": arguments.tx_finished_at, }, ) elif arguments.command == "enqueue-staircase-search": response = request_json( arguments.server, token, "POST", "/api/v1/local/tasks/staircase-search", { "agent_id": arguments.agent_id, "session_id": arguments.session_id, "expected_rx_sha256": arguments.expected_rx_sha256, "expected_reference_sha256": ( arguments.expected_reference_sha256 ), }, ) elif arguments.command == "register-agent-release": response = request_json( arguments.server, token, "POST", "/api/v1/local/releases/register-agent", {"package_path": str(arguments.package.resolve())}, ) elif arguments.command == "register-bootstrap-release": response = request_json( arguments.server, token, "POST", "/api/v1/local/releases/register-bootstrap", { "package_path": str(arguments.package.resolve()), "release_id": arguments.release_id, "agent_version": arguments.agent_version, "content_manifest_sha256": ( arguments.content_manifest_sha256 ), }, ) elif arguments.command == "enqueue-agent-update": response = request_json( arguments.server, token, "POST", "/api/v1/local/tasks/update-agent", { "agent_id": arguments.agent_id, "release_id": arguments.release_id, "validity_seconds": arguments.validity_seconds, }, ) elif arguments.command == "approve-bootstrap-download": response = request_json( arguments.server, token, "POST", "/api/v1/local/releases/approve-bootstrap", { "release_id": arguments.release_id, "validity_seconds": arguments.validity_seconds, }, ) elif arguments.command == "cancel-task": response = request_json( arguments.server, token, "POST", "/api/v1/local/tasks/cancel", { "task_id": arguments.task_id, "reason": arguments.reason, }, ) else: response = request_json( arguments.server, token, "GET", "/api/v1/local/status", ) print(json.dumps(response, ensure_ascii=False, indent=2, sort_keys=True)) print("RF TX = 0") if __name__ == "__main__": main()