diff options
Diffstat (limited to 'packages/meshbay-node/src')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/bundle_store.py | 105 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/config.py | 12 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 160 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/hub_client.py | 128 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 353 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ui/app.py | 163 |
6 files changed, 768 insertions, 153 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/bundle_store.py b/packages/meshbay-node/src/meshbay_node/bundle_store.py new file mode 100644 index 0000000..e7c6981 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/bundle_store.py @@ -0,0 +1,105 @@ +""" +Bundle store — SQLite-backed storage for GEK bundles and keypair bundles. + +GEK bundles: ECIES-wrapped GEK targeted at a specific user's X25519 key. +Keypair bundles: AES-GCM encrypted (Ed25519 + X25519) private keys, encrypted +with the user's password-derived bundle_key. Opaque to the node. + +Both are stored and served over the P2P DataChannel during MNP handshake. +""" + +import logging +from pathlib import Path + +import aiosqlite + +log = logging.getLogger(__name__) + +_SCHEMA_GEK = """\ +CREATE TABLE IF NOT EXISTS gek_bundles ( + group_id TEXT NOT NULL, + user_id TEXT NOT NULL, + pk_eph_b64 TEXT NOT NULL, + nonce_b64 TEXT NOT NULL, + wrapped_b64 TEXT NOT NULL, + stored_at TEXT NOT NULL DEFAULT (datetime('now')), + PRIMARY KEY (group_id, user_id) +); +""" + +_SCHEMA_KEYPAIR = """\ +CREATE TABLE IF NOT EXISTS keypair_bundles ( + user_id TEXT PRIMARY KEY, + bundle_enc TEXT NOT NULL, + stored_at TEXT NOT NULL DEFAULT (datetime('now')) +); +""" + + +class BundleStore: + def __init__(self, db_path: Path): + self._db_path = db_path + self._db: aiosqlite.Connection | None = None + + async def open(self) -> None: + self._db_path.parent.mkdir(parents=True, exist_ok=True) + self._db = await aiosqlite.connect(str(self._db_path)) + await self._db.execute(_SCHEMA_GEK) + await self._db.execute(_SCHEMA_KEYPAIR) + await self._db.commit() + + async def store( + self, + group_id: str, + user_id: str, + pk_eph_b64: str, + nonce_b64: str, + wrapped_b64: str, + ) -> None: + assert self._db + await self._db.execute( + "INSERT OR REPLACE INTO gek_bundles " + "(group_id, user_id, pk_eph_b64, nonce_b64, wrapped_b64, stored_at) " + "VALUES (?, ?, ?, ?, ?, datetime('now'))", + (group_id, user_id, pk_eph_b64, nonce_b64, wrapped_b64), + ) + await self._db.commit() + + async def fetch(self, group_id: str, user_id: str) -> dict | None: + assert self._db + async with self._db.execute( + "SELECT pk_eph_b64, nonce_b64, wrapped_b64 FROM gek_bundles " + "WHERE group_id = ? AND user_id = ?", + (group_id, user_id), + ) as cursor: + row = await cursor.fetchone() + if not row: + return None + return { + "pk_eph_b64": row[0], + "nonce_b64": row[1], + "wrapped_b64": row[2], + } + + async def store_keypair(self, user_id: str, bundle_enc: str) -> None: + assert self._db + await self._db.execute( + "INSERT OR REPLACE INTO keypair_bundles " + "(user_id, bundle_enc, stored_at) VALUES (?, ?, datetime('now'))", + (user_id, bundle_enc), + ) + await self._db.commit() + + async def fetch_keypair(self, user_id: str) -> str | None: + assert self._db + async with self._db.execute( + "SELECT bundle_enc FROM keypair_bundles WHERE user_id = ?", + (user_id,), + ) as cursor: + row = await cursor.fetchone() + return row[0] if row else None + + async def close(self) -> None: + if self._db: + await self._db.close() + self._db = None diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index 9e6a391..a7a0785 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -52,6 +52,11 @@ visibility = "public" [keystore] # unlock_file = "~/.config/meshbay/unlock.key" # or set MESHBAY_UNLOCK_KEY env var + +# Node sovereignty: pin the operator's Ed25519 public key (base64, 32 bytes raw). +# Admin operations (file delete) require cryptographic proof of this key. +# Auto-pinned on first startup from the node operator's keystore. +# admin_pk_ed25519 = "base64-encoded-32-bytes" """ @@ -59,7 +64,6 @@ visibility = "public" class HubConfig: url: str = "https://meshbay.org" username: str = "" - password: str = "" # loaded from keystore or env; never written to TOML @dataclass @@ -94,6 +98,7 @@ class Config: groups: list[GroupConfig] = field(default_factory=list) keystore: KeystoreConfig = field(default_factory=KeystoreConfig) data_dir: Path = field(default_factory=lambda: Path.home() / ".local" / "share" / "meshbay") + admin_pk_ed25519: str = "" # base64 raw Ed25519 public key pinned locally # Back-compat: single-group access @property @@ -144,6 +149,9 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: if "data_dir" in raw: cfg.data_dir = Path(raw["data_dir"]).expanduser().resolve() + if "admin_pk_ed25519" in raw: + cfg.admin_pk_ed25519 = raw["admin_pk_ed25519"] + ks = raw.get("keystore", {}) if "path" in ks: cfg.keystore.path = Path(ks["path"]).expanduser() @@ -155,8 +163,6 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: cfg.hub.url = url if user := os.environ.get("MESHBAY_USERNAME"): cfg.hub.username = user - if pwd := os.environ.get("MESHBAY_PASSWORD"): - cfg.hub.password = pwd if port := os.environ.get("MESHBAY_PORT"): cfg.node.port = int(port) diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 5851b34..fe12909 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -23,6 +23,7 @@ Usage: """ import asyncio +import base64 import json import logging import signal @@ -30,12 +31,14 @@ import sys from pathlib import Path import uvicorn +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from meshbay_common import MNP_VERSION from meshbay_common.protocol import MNP from meshbay_node.audit import AuditStore +from meshbay_node.bundle_store import BundleStore from meshbay_node.chat.store import ChatStore -from meshbay_node.config import Config, load_config, write_example_config +from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config from meshbay_node.hub_client import HubClient, HubConfig from meshbay_node.indexer import DirectoryIndexer from meshbay_node.keystore import NodeKeys, load_or_create_keystore @@ -111,6 +114,7 @@ class NodeDaemon: self._denylist = Denylist() if Denylist else None self._chat_stores: dict[str, ChatStore] = {} self._audit_store: AuditStore | None = None + self._bundle_store: BundleStore | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] self._hub: HubClient | None = None @@ -126,18 +130,46 @@ class NodeDaemon: ) log.info("Keys loaded: %s", keys.pk_ed25519_b64[:16]) - # 2. Hub connection + # 2. Start admin UI early (so operator can copy node key before hub login) + self._state["pk_node_ed25519"] = keys.pk_ed25519_b64 + self._state["config"] = self._config + from meshbay_node.ui import create_ui_app + ui_app = create_ui_app(self._state) + ui_cfg = uvicorn.Config( + ui_app, + host="127.0.0.1", + port=self._config.node.ui_port, + log_level="warning", + ) + ui_server = uvicorn.Server(ui_cfg) + self._tasks.append(asyncio.create_task(ui_server.serve())) + log.info("Admin UI at http://localhost:%d", self._config.node.ui_port) + + # 3. Hub connection (Ed25519 auth — retries until node key is linked) hub_cfg = HubConfig( hub_url=self._config.hub.url, username=self._config.hub.username, - password=self._config.hub.password, ) async with HubClient(hub_cfg, keys) as hub: self._hub = hub - session = await hub.startup(endpoint_hint=None) + session = await self._login_with_retry(hub) self._state["endpoint_hint"] = session.node_id - # 3. Build per-group contexts + # 4. Bundle store (P2P GEK bundles) + data_dir = self._config.data_dir + data_dir.mkdir(parents=True, exist_ok=True) + self._bundle_store = BundleStore(db_path=data_dir / "bundles.db") + await self._bundle_store.open() + log.info("Bundle store opened: %s", data_dir / "bundles.db") + + # X25519 key material for GEK unwrapping + from cryptography.hazmat.primitives import serialization + sk_x_raw = keys.sk_x25519.private_bytes( + serialization.Encoding.Raw, serialization.PrivateFormat.Raw, + serialization.NoEncryption()) + pk_x_raw = base64.b64decode(keys.pk_x25519_b64) + + # 4. Build per-group contexts groups_ctx: dict[str, dict] = {} for group_cfg in self._config.groups: if not group_cfg.id or not group_cfg.shared_dir: @@ -153,12 +185,13 @@ class NodeDaemon: gek = None if group_cfg.visibility == "private": - try: - gek = await hub.fetch_gek(group_cfg.id) + gek = await self._load_gek( + group_cfg.id, session.user_id, sk_x_raw, pk_x_raw) + if gek: log.info("GEK loaded for group %s", group_cfg.id[:8]) - except LookupError: - log.warning("No GEK for group %s — skipping", group_cfg.name) - continue + else: + log.info("No GEK yet for group %s — will accept first setup", + group_cfg.name) indexer = DirectoryIndexer( root=shared_root, @@ -183,9 +216,7 @@ class NodeDaemon: log.error("No valid groups configured — exiting") return - # 4. Chat stores (one SQLite DB per group) - data_dir = self._config.data_dir - data_dir.mkdir(parents=True, exist_ok=True) + # 5. Chat stores (one SQLite DB per group) for gid in groups_ctx: chat_db = data_dir / gid[:16] / "chat.db" store = ChatStore(db_path=chat_db) @@ -194,7 +225,7 @@ class NodeDaemon: groups_ctx[gid]["chat_store"] = store log.info("Chat stores opened: %d groups", len(self._chat_stores)) - # 4b. Audit store (legal compliance — IP + action logging) + # 6. Audit store (legal compliance — IP + action logging) audit_db = data_dir / "audit.db" self._audit_store = AuditStore(db_path=audit_db) await self._audit_store.open() @@ -219,6 +250,17 @@ class NodeDaemon: self._webrtc._ctx["hub_ws"] = _WsSender(hub) self._webrtc._ctx["node_user_id"] = session.user_id self._webrtc._ctx["audit_store"] = self._audit_store + self._webrtc._ctx["bundle_store"] = self._bundle_store + self._webrtc._ctx["sk_x25519_raw"] = sk_x_raw + self._webrtc._ctx["pk_x25519_raw"] = pk_x_raw + self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64 + + admin_pk = self._resolve_admin_pk(keys) + if admin_pk: + self._webrtc._ctx["admin_pk_ed25519"] = admin_pk + log.info("Admin Ed25519 key pinned for node sovereignty") + else: + log.warning("No admin_pk_ed25519 — admin operations disabled") log.info("WebRTC transport ready") else: log.warning("WebRTC not available (aiortc not installed)") @@ -323,23 +365,13 @@ class NodeDaemon: log.info("HTTP API on port %d for group %s", group_cfg.http_port, group_cfg.name) - # 10. Local web UI + # 10. Update admin UI state (UI already running from step 2) self._state["groups_ctx"] = groups_ctx - self._state["config"] = self._config self._state["audit_store"] = self._audit_store + self._state["bundle_store"] = self._bundle_store self._state["webrtc"] = self._webrtc self._state["hub"] = hub - from meshbay_node.ui import create_ui_app - ui_app = create_ui_app(self._state) - ui_cfg = uvicorn.Config( - ui_app, - host="127.0.0.1", - port=self._config.node.ui_port, - log_level="warning", - ) - ui_server = uvicorn.Server(ui_cfg) - self._tasks.append(asyncio.create_task(ui_server.serve())) - log.info("Local UI at http://localhost:%d", self._config.node.ui_port) + self._state["pk_x25519_raw"] = pk_x_raw self._state["status"] = "running" log.info("Node ready — %d groups, WebRTC=%s, QUIC=%s", @@ -363,6 +395,75 @@ class NodeDaemon: await self._shutdown() + async def _login_with_retry(self, hub: HubClient): + """Login to hub, retrying if the node key hasn't been linked yet.""" + import httpx as _httpx + while True: + try: + return await hub.startup(endpoint_hint=None) + except _httpx.HTTPStatusError as e: + body = e.response.text if hasattr(e.response, 'text') else '' + if e.response.status_code == 401 and "No node key" in body: + self._state["status"] = "waiting_for_node_key" + log.warning( + "Node key not linked — open admin UI at " + "http://localhost:%d, copy the key, and paste it in " + "Settings > Link Node on the hub. Retrying in 30s...", + self._config.node.ui_port, + ) + await asyncio.sleep(30) + else: + raise + except Exception as e: + log.warning("Hub login failed: %s — retrying in 10s", e) + await asyncio.sleep(10) + + async def _load_gek( + self, + group_id: str, + node_user_id: str, + sk_x_raw: bytes, + pk_x_raw: bytes, + ) -> bytes | None: + """Load GEK from local bundle store (node-only, hub never touches crypto).""" + from meshbay_common.crypto import unwrap_gek_aes + + if not self._bundle_store: + return None + + # Try node-specific bundle first (stored by init_gek for daemon reload), + # then fall back to operator's user bundle (legacy / pre-dual-key) + for user_key in [f"_node_{node_user_id}", node_user_id]: + bundle = await self._bundle_store.fetch(group_id, user_key) + if not bundle: + continue + try: + gek = unwrap_gek_aes(bundle, sk_x_raw, pk_x_raw) + log.info("GEK loaded from local bundle store for group %s (key=%s)", + group_id[:8], user_key[:16]) + return gek + except Exception as e: + log.debug("Failed to unwrap GEK bundle (key=%s): %s", user_key[:16], e) + + log.warning("No unwrappable GEK bundle found for group %s", group_id[:8]) + return None + + def _resolve_admin_pk(self, keys: NodeKeys) -> Ed25519PublicKey | None: + """Resolve the admin Ed25519 public key: config → auto-pin from node keystore.""" + if self._config.admin_pk_ed25519: + try: + raw = base64.b64decode(self._config.admin_pk_ed25519) + return Ed25519PublicKey.from_public_bytes(raw) + except Exception as e: + log.error("Invalid admin_pk_ed25519 in config: %s", e) + return None + + pk = keys.sk_ed25519.public_key() + from meshbay_common.crypto import pk_to_b64 + pk_b64 = pk_to_b64(pk) + log.info("Auto-pinning admin key from node keystore: %s", pk_b64[:16]) + return pk + async def _on_index_change(self, indexer: DirectoryIndexer) -> None: """Called when a DirectoryIndexer detects file changes.""" group_id = indexer.group_id @@ -429,6 +530,9 @@ class NodeDaemon: if self._audit_store: await self._audit_store.close() + if self._bundle_store: + await self._bundle_store.close() + for store in self._chat_stores.values(): await store.close() @@ -475,7 +579,7 @@ def main() -> None: calibrate_argon2() return - cfg = load_config(args.config) + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) if not cfg.hub.username: print("Error: hub.username not set in config. Run: meshbay-node init") sys.exit(1) diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py index ba9d3ff..432af0a 100644 --- a/packages/meshbay-node/src/meshbay_node/hub_client.py +++ b/packages/meshbay-node/src/meshbay_node/hub_client.py @@ -2,15 +2,15 @@ MeshBay Node — Hub client. Handles all communication from the node to a Mesh Hub: - - User registration (first run) - - Login → JWT (access token + refresh token) + - Ed25519 authentication (node-scoped JWT, no password material on node) - JWT offline verification and auto-refresh - Node announcement (endpoint_hint) - - GEK bundle retrieval for a group - User public key lookup (for GEK wrapping) + - Swarm hash registration -JWT verification is done locally using the hub's cached Ed25519 public key. -The hub is only contacted for login and refresh — not for every request. +The node authenticates via Ed25519 challenge-response (/v1/nodes/auth). +No auth_key or password is ever stored on or transmitted from the node. +The hub issues a node-scoped JWT that cannot manage group membership. """ import base64 @@ -23,10 +23,7 @@ from typing import Any, Callable import httpx import jwt -from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey -from cryptography.hazmat.primitives import serialization -from meshbay_common.crypto import pk_to_b64, unwrap_gek from meshbay_node.keystore import NodeKeys log = logging.getLogger(__name__) @@ -62,7 +59,6 @@ class HubSession: class HubConfig: hub_url: str username: str - password: str cache_dir: Path = field(default_factory=lambda: Path.home() / ".config" / "meshbay") @property @@ -110,80 +106,54 @@ class HubClient: log.info("Hub PK fetched and cached: %s", cache) return pem - # ── Registration ────────────────────────────────────────────────────────── - - async def register(self) -> str: - """Register this node's user on the hub. Returns user_id. Idempotent (409 ok).""" - r = await self._http.post("/v1/users/register", json={ - "username": self._config.username, - "password": self._config.password, - "pk_user_ed25519": self._keys.pk_ed25519_b64, - "pk_user_x25519": self._keys.pk_x25519_b64, - }) - if r.status_code == 201: - log.info("Registered user '%s' on hub", self._config.username) - return r.json()["user_id"] - if r.status_code == 409: - log.debug("User '%s' already registered", self._config.username) - return "" - r.raise_for_status() - return "" - - # ── Login ───────────────────────────────────────────────────────────────── + # ── Ed25519 authentication ─────────────────────────────────────────────── async def login(self) -> HubSession: - """Login, verify JWT offline, return HubSession.""" + """Authenticate via Ed25519 challenge-response. Returns node-scoped HubSession.""" hub_pk_pem = await self._fetch_hub_pk() - r = await self._http.post("/v1/users/login", json={ - "username": self._config.username, - "password": self._config.password, + timestamp = int(time.time()) + message = f"meshbay:node_auth:{self._config.username}:{timestamp}".encode() + signature = self._keys.sk_ed25519.sign(message) + + r = await self._http.post("/v1/nodes/auth", json={ + "username": self._config.username, + "timestamp": timestamp, + "signature": base64.b64encode(signature).decode(), }) r.raise_for_status() data = r.json() - access_token = data["access_token"] - refresh_token = data["refresh_token"] + access_token = data["access_token"] - # Verify offline — if this passes, the hub's identity is confirmed decoded = jwt.decode(access_token, hub_pk_pem, algorithms=["EdDSA"]) assert decoded["pk_user"] == self._keys.pk_ed25519_b64, \ "Hub returned token for wrong public key" assert "jti" in decoded, "Hub token missing jti — hub is outdated" + assert decoded.get("scope") == "node", \ + "Expected node-scoped token" - self._session = HubSession( - hub_url=self._config.hub_url, - username=self._config.username, - user_id=decoded["sub"], - access_token=access_token, - refresh_token=refresh_token, - hub_pk_pem=hub_pk_pem, - _token_exp=decoded["exp"], - ) + if self._session: + self._session.access_token = access_token + self._session._token_exp = decoded["exp"] + else: + self._session = HubSession( + hub_url=self._config.hub_url, + username=self._config.username, + user_id=decoded["sub"], + access_token=access_token, + refresh_token="", + hub_pk_pem=hub_pk_pem, + _token_exp=decoded["exp"], + ) log.info("Logged in as '%s' (exp in %ds)", self._config.username, self._session.token_expires_in) return self._session - async def refresh_token(self) -> None: - """Refresh the access token using the refresh token.""" - if self._session is None: - raise RuntimeError("Not logged in") - - r = await self._http.post("/v1/users/token/refresh", json={ - "refresh_token": self._session.refresh_token, - }) - r.raise_for_status() - new_token = r.json()["access_token"] - - decoded = jwt.decode(new_token, self._session.hub_pk_pem, algorithms=["EdDSA"]) - self._session.access_token = new_token - self._session._token_exp = decoded["exp"] - log.debug("Access token refreshed (exp in %ds)", self._session.token_expires_in) - async def ensure_fresh_token(self) -> None: - """Auto-refresh token if close to expiry.""" + """Re-authenticate with Ed25519 if token is close to expiry.""" if self._session and self._session.token_needs_refresh: - await self.refresh_token() + await self.login() # ── Node announcement ───────────────────────────────────────────────────── @@ -203,33 +173,6 @@ class HubClient: log.info("Node announced: %s (hint=%s)", node_id[:8], endpoint_hint) return node_id - # ── GEK retrieval ───────────────────────────────────────────────────────── - - async def fetch_gek(self, group_id: str) -> bytes: - """ - Fetch and unwrap the GEK bundle for a group. - Returns the raw GEK bytes. - """ - if self._session is None: - raise RuntimeError("Not logged in") - await self.ensure_fresh_token() - - r = await self._http.get(f"/v1/groups/{group_id}/gek", - headers=self._session.auth_headers) - if r.status_code == 404: - raise LookupError(f"No GEK bundle found for group {group_id!r}") - r.raise_for_status() - - bundle = r.json() - sk_x_raw = self._keys.sk_x25519.private_bytes( - serialization.Encoding.Raw, serialization.PrivateFormat.Raw, - serialization.NoEncryption()) - pk_x_raw = base64.b64decode(self._keys.pk_x25519_b64) - - gek = unwrap_gek(bundle, sk_x_raw, pk_x_raw) - log.info("GEK unwrapped for group %s", group_id[:8]) - return gek - # ── User pubkey lookup ──────────────────────────────────────────────────── async def get_user_pubkeys(self, username: str) -> dict: @@ -354,10 +297,9 @@ class HubClient: async def startup(self, endpoint_hint: str | None = None) -> HubSession: """ - Full startup sequence: register (idempotent) → login → announce node. - Returns an active HubSession. + Full startup sequence: Ed25519 login → announce node. + The operator must register separately (browser or setup script). """ - await self.register() session = await self.login() await self.announce_node(endpoint_hint) return session diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index e692c80..13e90c8 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -24,7 +24,10 @@ Signaling flow (handled externally by the hub): import asyncio import base64 +import hashlib +import hmac import logging +import os import struct from pathlib import Path from typing import Any @@ -32,7 +35,10 @@ from typing import Any import jwt import msgpack from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel -from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) from meshbay_common import MNP_VERSION from meshbay_common.crypto import pk_to_b64 @@ -46,6 +52,15 @@ CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 +def _extract_dtls_fingerprint(sdp: str) -> bytes: + """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes.""" + for line in sdp.splitlines(): + if line.startswith("a=fingerprint:sha-256 "): + hex_str = line.split(" ", 1)[1].replace(":", "") + return bytes.fromhex(hex_str) + return b"" + + STREAM_SEGMENT_SIZE = 256 * 1024 _H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} @@ -153,6 +168,9 @@ class WebRTCPeerSession: self._peer_id: str = peer_id self._remote_ip: str = "" self._username: str = "" + self._pk_user: str = "" + self._gek_challenge: bytes | None = None + self._admin_challenges: dict[str, bytes] = {} def _setup_channel(self, channel: RTCDataChannel) -> None: self._channel = channel @@ -171,6 +189,12 @@ class WebRTCPeerSession: try: if mtype == MNP.HANDSHAKE: self._do_handshake(msg) + elif mtype == MNP.HANDSHAKE_RESPONSE: + self._do_handshake_response(msg) + elif mtype == MNP.GEK_BUNDLE_FETCH and self._gek_challenge is not None: + asyncio.ensure_future(self._do_gek_bundle_fetch()) + elif mtype == MNP.KEYPAIR_BUNDLE_FETCH and self._gek_challenge is not None: + asyncio.ensure_future(self._do_keypair_bundle_fetch()) elif self._user_id is None: self._send({"type": "error", "detail": "Handshake required"}) elif mtype == MNP.INDEX_SYNC: @@ -179,8 +203,6 @@ class WebRTCPeerSession: self._do_file_request(msg) elif mtype == MNP.STREAM_SEGMENT: self._do_stream_segment(msg) - elif mtype == MNP.GEK_REQUEST: - self._do_gek_request() elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message(msg) elif mtype == MNP.CHAT_HISTORY: @@ -189,6 +211,12 @@ class WebRTCPeerSession: self._do_file_upload(msg) elif mtype == MNP.FILE_DELETE: self._do_file_delete(msg) + elif mtype == MNP.ADMIN_RESPONSE: + self._do_admin_response(msg) + elif mtype == MNP.GEK_BUNDLE_STORE: + asyncio.ensure_future(self._do_gek_bundle_store(msg)) + elif mtype == MNP.KEYPAIR_BUNDLE_STORE: + asyncio.ensure_future(self._do_keypair_bundle_store(msg)) elif mtype == MNP.STREAM_REQUEST: asyncio.ensure_future(self._stream_video(msg)) else: @@ -234,23 +262,235 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "Group not hosted on this node"}) return - self._user_id = decoded["sub"] - self._group_id = group_id - self._username = decoded.get("username", "") + # Store decoded JWT data but DO NOT set self._user_id yet — + # the user is not authenticated until they prove GEK possession. + self._pending_sub = decoded["sub"] + self._pending_group = group_id + self._pending_username = decoded.get("username", "") + self._pending_pk_user = decoded.get("pk_user", "") + + ctx = self._ctx + if "groups" in ctx and group_id: + gctx = ctx["groups"].get(group_id, ctx) + else: + gctx = ctx + gek = gctx.get("gek") + + nonce = os.urandom(32) + self._gek_challenge = nonce + challenge = { + "type": MNP.HANDSHAKE_CHALLENGE, + "v": MNP_VERSION, + "nonce": base64.b64encode(nonce).decode(), + } + if not gek: + self._send({ + "type": "error", + "detail": "Group encryption not initialized — contact node operator", + }) + return + self._send(challenge) + + def _do_handshake_response(self, msg: dict) -> None: + if not self._gek_challenge or not hasattr(self, "_pending_sub"): + self._send({"type": "error", "detail": "No pending handshake challenge"}) + return + + group_id = self._pending_group + ctx = self._ctx + if "groups" in ctx and group_id: + gctx = ctx["groups"].get(group_id, ctx) + else: + gctx = ctx + gek = gctx.get("gek") + + if not gek: + self._send({"type": "error", "detail": "Group encryption not initialized"}) + self._gek_challenge = None + return + + proof = msg.get("proof", "") + try: + proof_bytes = base64.b64decode(proof) + except Exception: + self._send({"type": "error", "detail": "Invalid proof encoding"}) + return + + offer_fp = b"" + answer_fp = b"" + if self._pc.remoteDescription: + offer_fp = _extract_dtls_fingerprint(self._pc.remoteDescription.sdp) + if self._pc.localDescription: + answer_fp = _extract_dtls_fingerprint(self._pc.localDescription.sdp) + + data = self._gek_challenge + offer_fp + answer_fp + expected = hmac.new(gek, data, hashlib.sha256).digest() + if not hmac.compare_digest(proof_bytes, expected): + self._send({"type": "error", "detail": "GEK proof failed"}) + self._gek_challenge = None + self._audit_auth_failed(group_id, "GEK HMAC mismatch") + return + + self._gek_challenge = None + self._complete_handshake() + + def _complete_handshake(self) -> None: + self._user_id = self._pending_sub + self._group_id = self._pending_group + self._username = self._pending_username + self._pk_user = self._pending_pk_user peers = self._ctx.get("_peers") if peers is not None: peers[self._user_id] = self + node_user_id = self._ctx.get("node_user_id") log.info("WebRTC handshake OK — user=%s group=%s", - self._user_id[:8], group_id[:8] if group_id else "none") - self._send({ + self._user_id[:8], + self._group_id[:8] if self._group_id else "none") + ack = { "type": MNP.HANDSHAKE_ACK, "v": MNP_VERSION, "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()), - }) + "is_node_admin": bool(node_user_id and self._user_id == node_user_id), + } + if node_user_id: + ack["node_user_id"] = node_user_id + pk_x_b64 = self._ctx.get("pk_x25519_b64") + if pk_x_b64: + ack["node_pk_x25519"] = pk_x_b64 + self._send(ack) self._audit("handshake") + async def _do_gek_bundle_fetch(self) -> None: + """Serve the caller's wrapped GEK bundle during the handshake window.""" + bundle_store = self._ctx.get("bundle_store") + if not bundle_store: + self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) + return + + group_id = getattr(self, "_pending_group", "") + user_id = getattr(self, "_pending_sub", "") + if not group_id or not user_id: + self._send({"type": "error", "detail": "No pending handshake"}) + return + + bundle = await bundle_store.fetch(group_id, user_id) + if bundle: + self._send({ + "type": MNP.GEK_BUNDLE_RESP, + "v": MNP_VERSION, + "found": True, + "pk_eph_b64": bundle["pk_eph_b64"], + "nonce_b64": bundle["nonce_b64"], + "wrapped_b64": bundle["wrapped_b64"], + }) + else: + self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) + + async def _do_gek_bundle_store(self, msg: dict) -> None: + """Store a wrapped GEK bundle for a target user (admin operation).""" + bundle_store = self._ctx.get("bundle_store") + if not bundle_store: + self._send({"type": "error", "detail": "Bundle store not available"}) + return + + target_user_id = msg.get("user_id", "") + group_id = msg.get("group_id") or self._group_id + pk_eph = msg.get("pk_eph_b64", "") + nonce = msg.get("nonce_b64", "") + wrapped = msg.get("wrapped_b64", "") + + if not target_user_id or not pk_eph or not nonce or not wrapped or not group_id: + self._send({"type": "error", "detail": "Missing bundle fields"}) + return + + await bundle_store.store(group_id, target_user_id, pk_eph, nonce, wrapped) + log.info("GEK bundle stored: group=%s user=%s", group_id[:8], target_user_id[:8]) + self._audit("gek_bundle_store", f"target={target_user_id[:8]}") + + self._send({ + "type": "ack", "v": MNP_VERSION, + "detail": "gek_bundle_stored", + "user_id": target_user_id, + }) + + # Auto-activate GEK if the bundle is for the node operator + node_user_id = self._ctx.get("node_user_id") + if node_user_id and target_user_id == node_user_id and group_id: + await self._try_activate_gek(group_id, target_user_id) + + async def _try_activate_gek(self, group_id: str, user_id: str) -> None: + """Unwrap and activate GEK for the node when the operator's bundle arrives.""" + from meshbay_common.crypto import unwrap_gek_aes + + bundle_store = self._ctx.get("bundle_store") + sk_x_raw = self._ctx.get("sk_x25519_raw") + pk_x_raw = self._ctx.get("pk_x25519_raw") + if not bundle_store or not sk_x_raw or not pk_x_raw: + return + + bundle = await bundle_store.fetch(group_id, user_id) + if not bundle: + return + + try: + gek = unwrap_gek_aes(bundle, sk_x_raw, pk_x_raw) + except Exception as e: + log.warning("Failed to unwrap GEK for auto-activation: %s", e) + return + + groups = self._ctx.get("groups") + if groups and group_id in groups: + groups[group_id]["gek"] = gek + log.info("GEK auto-activated for group %s", group_id[:8]) + elif "gek" in self._ctx: + self._ctx["gek"] = gek + log.info("GEK auto-activated (single-group mode)") + + async def _do_keypair_bundle_fetch(self) -> None: + """Serve the caller's encrypted keypair bundle during the handshake window.""" + bundle_store = self._ctx.get("bundle_store") + if not bundle_store: + self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) + return + + user_id = getattr(self, "_pending_sub", "") + if not user_id: + self._send({"type": "error", "detail": "No pending handshake"}) + return + + bundle_enc = await bundle_store.fetch_keypair(user_id) + if bundle_enc: + self._send({ + "type": MNP.KEYPAIR_BUNDLE_RESP, + "v": MNP_VERSION, + "found": True, + "bundle_enc": bundle_enc, + }) + else: + self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) + + async def _do_keypair_bundle_store(self, msg: dict) -> None: + """Store an encrypted keypair bundle (user backs up their own keys on node).""" + bundle_store = self._ctx.get("bundle_store") + if not bundle_store: + self._send({"type": "error", "detail": "Bundle store not available"}) + return + + bundle_enc = msg.get("bundle_enc", "") + if not bundle_enc: + self._send({"type": "error", "detail": "Missing bundle_enc"}) + return + + await bundle_store.store_keypair(self._user_id, bundle_enc) + log.info("Keypair bundle stored for user=%s", self._user_id[:8]) + self._audit("keypair_bundle_store") + self._send({ + "type": "ack", "v": MNP_VERSION, + "detail": "keypair_bundle_stored", + }) + def _audit_auth_failed(self, group_id: str, reason: str) -> None: audit = self._ctx.get("audit_store") if audit: @@ -275,6 +515,7 @@ class WebRTCPeerSession: { "id": e.id, "name": e.name, "path": e.path, "size": e.size, "type": e.type, "added_at": e.added_at, + "uploader_id": e.uploader_id, } for e in idx.entries ] @@ -286,18 +527,6 @@ class WebRTCPeerSession: "entries": entries, }) - def _do_gek_request(self) -> None: - ctx = self._group_ctx() - gek = ctx.get("gek") - if not gek: - self._send({"type": "error", "detail": "No GEK available"}) - return - self._send({ - "type": MNP.GEK_RESPONSE, - "v": MNP_VERSION, - "gek_b64": base64.b64encode(gek).decode(), - }) - def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg["file_id"] @@ -378,7 +607,7 @@ class WebRTCPeerSession: if chat_store: raw = payload.encode() if isinstance(payload, str) else payload asyncio.ensure_future(chat_store.save_message( - sender_id=msg.get("sender_id", self._user_id), + sender_id=self._user_id, iteration=msg.get("iteration", 0), payload=raw, thread_id=msg.get("thread_id"), @@ -389,7 +618,7 @@ class WebRTCPeerSession: broadcast = { "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, - "sender_id": msg.get("sender_id", self._user_id), + "sender_id": self._user_id, "sender_name": sender_name, "payload": payload, "thread_id": msg.get("thread_id"), @@ -493,6 +722,18 @@ class WebRTCPeerSession: tmp_path.rename(final_path) log.info("Upload complete: %s (%d chunks)", safe_name, total_chunks) self._audit("file_upload", safe_name) + self._register_uploader(ctx, safe_name) + + def _register_uploader(self, ctx: dict, filename: str) -> None: + """Tag the index entry with the uploader's user_id after upload completes.""" + idx = ctx.get("index") + if not idx: + return + for entry in idx.entries: + if entry.name == filename and entry.path == "": + entry.uploader_id = self._user_id + entry.uploader_pk = self._pk_user + return def _do_file_delete(self, msg: dict) -> None: ctx = self._group_ctx() @@ -501,16 +742,76 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "Missing file_id"}) return - node_user_id = self._ctx.get("node_user_id") - if node_user_id and self._user_id != node_user_id: - self._send({"type": "error", "detail": "Only node admin can delete files"}) + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + + admin_pk = self._ctx.get("admin_pk_ed25519") + has_uploader_pk = bool(entry.uploader_pk) + if not admin_pk and not has_uploader_pk: + self._send({"type": "error", "detail": "No authorized key for deletion"}) + return + + challenge = os.urandom(32) + self._admin_challenges[file_id] = challenge + self._send({ + "type": MNP.ADMIN_CHALLENGE, + "v": MNP_VERSION, + "challenge": base64.b64encode(challenge).decode(), + "file_id": file_id, + }) + + def _do_admin_response(self, msg: dict) -> None: + file_id = msg.get("file_id", "") + sig_b64 = msg.get("signature", "") + + challenge = self._admin_challenges.pop(file_id, None) + if not challenge: + self._send({"type": "error", "detail": "No pending admin challenge"}) + return + + try: + sig_bytes = base64.b64decode(sig_b64) + except Exception: + self._send({"type": "error", "detail": "Invalid signature encoding"}) return + ctx = self._group_ctx() entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return + verified = False + + # Try admin key (locally pinned) + admin_pk = self._ctx.get("admin_pk_ed25519") + if admin_pk: + try: + admin_pk.verify(sig_bytes, challenge) + verified = True + except Exception: + pass + + # Try uploader key (stored at upload time) + if not verified and entry.uploader_pk: + try: + uploader_key = Ed25519PublicKey.from_public_bytes( + base64.b64decode(entry.uploader_pk)) + uploader_key.verify(sig_bytes, challenge) + verified = True + except Exception: + pass + + if not verified: + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") + return + + self._exec_file_delete(ctx, file_id, entry) + + def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: file_path = ctx["shared_root"] / entry.path / entry.name if file_path.exists(): file_path.unlink() diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index 5e77ed8..b4885af 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -12,15 +12,17 @@ Served only on 127.0.0.1 — not exposed to the network. No authentication required (localhost only). """ +import base64 import json import logging import time from pathlib import Path from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query -from fastapi.responses import HTMLResponse +from fastapi.responses import HTMLResponse, JSONResponse from meshbay_node import __version__ +from meshbay_common.crypto import generate_gek, wrap_gek_aes log = logging.getLogger(__name__) @@ -52,6 +54,7 @@ def create_ui_app(state: dict) -> FastAPI: "group_count": len(groups_ctx), "total_files": total_files, "webrtc_peers": webrtc.active_peers if webrtc else 0, + "pk_node_ed25519": state.get("pk_node_ed25519", ""), } @app.get("/api/groups") @@ -167,6 +170,95 @@ def create_ui_app(state: dict) -> FastAPI: ], } + # ── GEK initialization (operator only, localhost) ────────────────────── + + @app.post("/api/groups/{group_id}/gek") + async def init_gek(group_id: str): + """Generate GEK, wrap for all group members, store, and activate.""" + groups_ctx = state.get("groups_ctx", {}) + if group_id not in groups_ctx: + return JSONResponse({"error": "Group not hosted on this node"}, 404) + + hub = state.get("hub") + if not hub or not hub._session: + return JSONResponse({"error": "Hub not connected"}, 503) + + bundle_store = state.get("bundle_store") + if not bundle_store: + return JSONResponse({"error": "Bundle store not available"}, 503) + + await hub.ensure_fresh_token() + session = hub._session + members_resp = await hub._http.get( + f"/v1/groups/{group_id}/members", + headers=session.auth_headers, + ) + if not members_resp.is_success: + return JSONResponse( + {"error": f"Failed to fetch members: {members_resp.status_code}"}, 502) + members = members_resp.json().get("members", []) + if not members: + return JSONResponse({"error": "No members in group"}, 400) + + existing_gek = groups_ctx[group_id].get("gek") + gek = existing_gek or generate_gek() + + wrapped_count = 0 + errors = [] + for member in members: + username = member["username"] + user_id = member["user_id"] + try: + pk_data = await hub.get_user_pubkeys(username) + pk_x_raw = base64.b64decode(pk_data["pk_x25519"]) + bundle = wrap_gek_aes(gek, pk_x_raw) + await bundle_store.store( + group_id, user_id, + bundle["pk_eph_b64"], bundle["nonce_b64"], bundle["wrapped_b64"], + ) + wrapped_count += 1 + log.info("GEK wrapped for %s (%s)", username, user_id[:8]) + except Exception as e: + errors.append(f"{username}: {e}") + log.warning("Failed to wrap GEK for %s: %s", username, e) + + if wrapped_count == 0: + return JSONResponse( + {"error": "Failed to wrap GEK for any member", "details": errors}, 500) + + # Also store a copy wrapped for the node keystore X25519 key + # so the daemon can reload GEK on restart without the operator's browser keys + config = state.get("config") + node_user_id = hub._session.user_id if hub._session else None + pk_x_node_raw = state.get("pk_x25519_raw") + if pk_x_node_raw and node_user_id: + try: + node_bundle = wrap_gek_aes(gek, pk_x_node_raw) + await bundle_store.store( + group_id, f"_node_{node_user_id}", + node_bundle["pk_eph_b64"], node_bundle["nonce_b64"], + node_bundle["wrapped_b64"], + ) + log.info("GEK also wrapped for node keystore (daemon reload)") + except Exception as e: + log.warning("Failed to wrap GEK for node keystore: %s", e) + + groups_ctx[group_id]["gek"] = gek + log.info("GEK initialized for group %s — wrapped for %d/%d members", + group_id[:8], wrapped_count, len(members)) + + webrtc = state.get("webrtc") + if webrtc and "groups" in webrtc._ctx and group_id in webrtc._ctx["groups"]: + webrtc._ctx["groups"][group_id]["gek"] = gek + + return { + "status": "ok", + "group_id": group_id, + "wrapped_count": wrapped_count, + "total_members": len(members), + "errors": errors, + } + # ── Chat endpoints ─────────────────────────────────────────────────────── _chat_subscribers: list[WebSocket] = [] @@ -246,7 +338,10 @@ def _render_page(state: dict) -> str: webrtc = state.get("webrtc") total_files = sum(idx.count for idx in indexes.values()) peer_count = webrtc.active_peers if webrtc else 0 - status_color = {"running": "#22c55e", "error": "#ef4444"}.get(status, "#f59e0b") + status_color = { + "running": "#22c55e", "error": "#ef4444", + "waiting_for_node_key": "#f97316", + }.get(status, "#f59e0b") # Groups section groups_html = "" @@ -269,13 +364,34 @@ def _render_page(state: dict) -> str: f"<td>{_fmt_size(e.size)}</td><td>{e.path or '/'}</td></tr>" ) + has_gek = bool(ctx.get("gek")) + gek_badge = ( + '<span class="badge" style="background:#22c55e">GEK active</span>' + if has_gek + else '<span class="badge" style="background:#ef4444">No GEK</span>' + ) + gek_label = "Re-wrap GEK for all members" if has_gek else "Initialize GEK" + gek_color = "#3b82f6" if has_gek else "#22c55e" + gek_action = f""" + <div style="margin:10px 0"> + <button onclick="initGEK('{gid}')" + id="gek-btn-{gid[:8]}" + style="padding:8px 16px;background:{gek_color};color:#fff;border:none; + border-radius:6px;cursor:pointer;font-size:0.85em"> + {gek_label} + </button> + <span id="gek-status-{gid[:8]}" class="muted" style="margin-left:8px"></span> + </div>""" + groups_html += f""" <div class="card"> <h3>{name} <span class="badge" style="background:#6366f1">{vis}</span> + {gek_badge} </h3> <p><b>Directory:</b> <code>{shared}</code></p> <p><b>Files:</b> {fcount} — <b>Total:</b> {_fmt_size(total_size)}</p> + {gek_action} <p class="muted">ID: {gid}</p> <details><summary>File list</summary> <table> @@ -382,6 +498,21 @@ def _render_page(state: dict) -> str: <p><b>Node ID:</b> <code>{state.get("endpoint_hint") or "—"}</code></p> </div> + <h2>Link Node to Hub Account</h2> + <div class="card"> + <p>To connect to your group from a browser, link this node to your hub account. + Copy the key below and paste it in <b>Settings > Link Node</b> on the hub.</p> + <div style="margin:12px 0;display:flex;align-items:center;gap:8px"> + <code id="nodeKey" style="flex:1;padding:8px;word-break:break-all;background:var(--border); + border-radius:4px;font-size:0.9em;user-select:all">{state.get("pk_node_ed25519", "—")}</code> + <button onclick="navigator.clipboard.writeText(document.getElementById('nodeKey').textContent).then(()=>{{this.textContent='Copied!';setTimeout(()=>this.textContent='Copy',2000)}})" + style="padding:8px 16px;background:var(--accent);color:#fff;border:none;border-radius:6px; + cursor:pointer;font-size:0.85em;white-space:nowrap">Copy</button> + </div> + <p class="muted">This is the node's Ed25519 public key. It's safe to share — it identifies + this node but cannot be used to impersonate it.</p> + </div> + <div class="footer"> MeshBay Node v{__version__} — localhost only — <a href="/api/status">status</a> · @@ -392,7 +523,33 @@ def _render_page(state: dict) -> str: — auto-refresh 10s </div> </div> -<script>setTimeout(()=>location.reload(), 10000);</script> +<script> +async function initGEK(groupId) {{ + const btn = document.getElementById('gek-btn-' + groupId.slice(0,8)); + const status = document.getElementById('gek-status-' + groupId.slice(0,8)); + if (btn) btn.disabled = true; + if (status) status.textContent = 'Initializing...'; + try {{ + const resp = await fetch('/api/groups/' + groupId + '/gek', {{ method: 'POST' }}); + const data = await resp.json(); + if (resp.ok) {{ + if (status) status.textContent = 'GEK initialized — wrapped for ' + + data.wrapped_count + '/' + data.total_members + ' members'; + if (status) status.style.color = '#22c55e'; + setTimeout(() => location.reload(), 2000); + }} else {{ + if (status) status.textContent = data.error || 'Failed'; + if (status) status.style.color = '#ef4444'; + if (btn) btn.disabled = false; + }} + }} catch (e) {{ + if (status) status.textContent = 'Error: ' + e.message; + if (status) status.style.color = '#ef4444'; + if (btn) btn.disabled = false; + }} +}} +setTimeout(()=>location.reload(), 10000); +</script> </body> </html>""" |