"""Ed25519 request signing for the ``meshstack-device-v2`` edge function. The canonical string and header names here are not our invention: they mirror ``supabase/functions/meshstack-device-v2/index.ts`` in ``ttracx/sadie-mesh``, which builds POST\\n{action}\\n{timestamp}\\n{nonce}\\n{sha256_hex(raw_body)} and verifies it against ``meshstack_devices_v2.identity_public_key``. Any drift between this module and that function is a silent authentication failure, so the canonical form is asserted by ``tests/test_signing.py`` against a vector taken from the edge function's own construction. Two properties matter for correctness: * The body hash is over the **exact bytes sent on the wire**. We therefore serialise once and sign that same buffer, rather than re-serialising, because ``json.dumps`` key order or separator changes would invalidate the signature. * The timestamp must land inside the backend's +/-5 minute window, so the node clock is checked at startup rather than failing obscurely on first register. """ from __future__ import annotations import base64 import hashlib import json import os import secrets import time from dataclasses import dataclass from typing import Any, Mapping from .errors import ConfigError try: from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, Ed25519PublicKey, ) except ImportError as _import_error: # pragma: no cover - environment dependent # Re-raised with instructions rather than letting a bare ModuleNotFoundError # surface from deep inside `import thoxmesh_node`, where it reads as a broken # package rather than one missing dependency. raise ConfigError( "the 'cryptography' package is required for mesh device signing. Install with:\n" " pip install cryptography\n" "It ships prebuilt wheels for every platform this node targets, so this " "should not need a compiler." ) from _import_error #: Header names expected by ``meshstack-device-v2``. HEADER_DEVICE_ID = "x-thox-device-id" HEADER_TIMESTAMP = "x-thox-timestamp" HEADER_NONCE = "x-thox-nonce" HEADER_SIGNATURE = "x-thox-signature" #: The backend rejects requests more than 5 minutes from its own clock. CLOCK_SKEW_LIMIT_SECONDS = 5 * 60 def sha256_hex(payload: bytes) -> str: """Hex SHA-256, matching the edge function's ``sha256Hex``.""" return hashlib.sha256(payload).hexdigest() def canonical_string(action: str, timestamp_ms: str, nonce: str, body_hash: str) -> str: """Build the exact string the backend reconstructs and verifies. Kept as a standalone function so the test suite can pin the format without needing a private key. """ return f"POST\n{action}\n{timestamp_ms}\n{nonce}\n{body_hash}" @dataclass(frozen=True) class SignedRequest: """A ready-to-send signed request: the exact body bytes plus headers.""" body: bytes headers: dict[str, str] class DeviceIdentity: """A mesh device identity: a device UUID plus its Ed25519 signing key. The private key is a real secret. It is loaded from the environment or a file and never logged, never serialised into telemetry, and never committed. In Colab it is pasted into a form field; in a Space it is a Space secret. """ def __init__(self, device_id: str, private_key: Ed25519PrivateKey) -> None: if not device_id: raise ConfigError("device_id must not be empty") self._device_id = device_id self._private_key = private_key # ── Construction ───────────────────────────────────────────────────── @classmethod def from_base64_seed(cls, device_id: str, seed_b64: str) -> "DeviceIdentity": """Load from a base64 32-byte Ed25519 seed (the private key material). Accepts standard or URL-safe base64, with or without padding, because these strings get copy-pasted through notebooks and Space secret fields where padding is routinely mangled. """ raw = _b64decode_forgiving(seed_b64, field="private key seed") if len(raw) == 64: # Some tools emit the libsodium 64-byte "secret key" (seed || public). raw = raw[:32] if len(raw) != 32: raise ConfigError( f"Ed25519 private key seed must decode to 32 bytes, got {len(raw)}" ) return cls(device_id, Ed25519PrivateKey.from_private_bytes(raw)) @classmethod def from_pem_file(cls, device_id: str, path: str) -> "DeviceIdentity": """Load from an unencrypted PKCS#8 PEM private key file.""" try: with open(path, "rb") as handle: key = serialization.load_pem_private_key(handle.read(), password=None) except (OSError, ValueError) as exc: raise ConfigError(f"could not read Ed25519 private key from {path}: {exc}") from exc if not isinstance(key, Ed25519PrivateKey): raise ConfigError(f"{path} is not an Ed25519 private key") return cls(device_id, key) @classmethod def generate(cls, device_id: str = "00000000-0000-0000-0000-000000000000") -> "DeviceIdentity": """Generate an ephemeral identity. For tests and key-provisioning only. A generated key is useless against the live mesh until its public half is registered on ``meshstack_devices_v2`` via a pairing session, which requires an authenticated mesh owner. """ return cls(device_id, Ed25519PrivateKey.generate()) # ── Accessors ──────────────────────────────────────────────────────── @property def device_id(self) -> str: return self._device_id def public_key_base64(self) -> str: """Base64 raw public key, the format stored in ``identity_public_key``.""" raw = self._private_key.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) return base64.b64encode(raw).decode("ascii") def private_seed_base64(self) -> str: """Base64 raw private seed. Only used by the key-provisioning command.""" raw = self._private_key.private_bytes( encoding=serialization.Encoding.Raw, format=serialization.PrivateFormat.Raw, encryption_algorithm=serialization.NoEncryption(), ) return base64.b64encode(raw).decode("ascii") # ── Signing ────────────────────────────────────────────────────────── def sign_request(self, action: str, payload: Mapping[str, Any]) -> SignedRequest: """Serialise, hash, sign and return body + headers for one mesh call. ``action`` is signed separately from the body even though the body also carries it; that is the backend's construction and we match it exactly. """ if not action: raise ConfigError("action must not be empty") body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8") timestamp_ms = str(int(time.time() * 1000)) nonce = secrets.token_hex(16) message = canonical_string(action, timestamp_ms, nonce, sha256_hex(body)) signature = self._private_key.sign(message.encode("utf-8")) headers = { "content-type": "application/json", HEADER_DEVICE_ID: self._device_id, HEADER_TIMESTAMP: timestamp_ms, HEADER_NONCE: nonce, HEADER_SIGNATURE: base64.b64encode(signature).decode("ascii"), } return SignedRequest(body=body, headers=headers) def __repr__(self) -> str: # pragma: no cover - never leak key material return f"DeviceIdentity(device_id={self._device_id!r})" def verify_signature(public_key_b64: str, message: str, signature_b64: str) -> bool: """Verify a canonical string against a base64 raw public key. Present so the smoke test can prove the signature the agent produces is the one the backend would accept, without needing the live mesh. """ try: public = Ed25519PublicKey.from_public_bytes(_b64decode_forgiving(public_key_b64, field="public key")) public.verify(_b64decode_forgiving(signature_b64, field="signature"), message.encode("utf-8")) except Exception: return False return True def check_clock_skew(server_time_ms: int, *, now_ms: int | None = None) -> float: """Return skew in seconds, raising if it exceeds the backend's window. Called after the first successful heartbeat, which echoes ``server_time``. Colab VMs and Space containers occasionally drift, and a drifted clock presents as ``request_timestamp_out_of_window`` on every subsequent call — an error whose text does not obviously mean "your clock is wrong". """ current = now_ms if now_ms is not None else int(time.time() * 1000) skew = (current - server_time_ms) / 1000.0 if abs(skew) > CLOCK_SKEW_LIMIT_SECONDS: raise ConfigError( f"node clock is {skew:.0f}s from mesh time; the mesh rejects skew over " f"{CLOCK_SKEW_LIMIT_SECONDS}s. Fix the node clock before registering." ) return skew def _b64decode_forgiving(value: str, *, field: str) -> bytes: """Decode base64 tolerating URL-safe alphabet and missing padding.""" if not isinstance(value, str) or not value.strip(): raise ConfigError(f"{field} must be a non-empty base64 string") text = value.strip().replace("-", "+").replace("_", "/") text += "=" * (-len(text) % 4) try: return base64.b64decode(text, validate=False) except Exception as exc: raise ConfigError(f"{field} is not valid base64: {exc}") from exc def load_identity_from_env(env: Mapping[str, str] | None = None) -> DeviceIdentity | None: """Build a ``DeviceIdentity`` from environment, or ``None`` if unconfigured. Returning ``None`` rather than raising lets the agent fall back to controller transport, which needs no device credentials and is what makes a same-day demo possible before a device has been paired. """ source = env if env is not None else os.environ device_id = (source.get("THOX_MESH_DEVICE_ID") or "").strip() seed = (source.get("THOX_MESH_DEVICE_KEY") or "").strip() key_file = (source.get("THOX_MESH_DEVICE_KEY_FILE") or "").strip() if not device_id: return None if seed: return DeviceIdentity.from_base64_seed(device_id, seed) if key_file: return DeviceIdentity.from_pem_file(device_id, key_file) raise ConfigError( "THOX_MESH_DEVICE_ID is set but no key: provide THOX_MESH_DEVICE_KEY " "(base64 seed) or THOX_MESH_DEVICE_KEY_FILE (PEM path)" )