diff options
Diffstat (limited to 'packages/meshbay-node/src')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/config.py | 16 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 229 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/roster.py | 372 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 411 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/ui/app.py | 110 |
5 files changed, 979 insertions, 159 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index b681355..04cb1d3 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -44,7 +44,10 @@ id = "" name = "Public Archive" shared_dir = "/home/user/Archive" quic_port = 19012 -visibility = "public" +visibility = "public" # discoverable on the hub +# join_policy = "open" # anyone the hub says is a member gets the group key, + # with no pairing code. Only for groups where that is + # genuinely intended: it means the hub can join too. [keystore] # unlock_file = "~/.config/meshbay/unlock.key" @@ -74,7 +77,15 @@ class GroupConfig: id: str = "" name: str = "" shared_dir: str = "" - visibility: str = "private" # public|private + visibility: str = "private" # public|private — discoverability, not admission + # Admission. "invite" (default) means a newcomer needs a one-time pairing code + # before the node wraps the group key for them; "open" means the node pins + # whoever turns up first (TOFU) and serves them. + # + # Deliberately read from THIS file and never from the hub: a hub that could + # declare a group open would walk into any group it liked. Being findable + # (`visibility`) and being open (`join_policy`) are different questions. + join_policy: str = "invite" # invite|open quic_port: int = 19010 # QUIC MNP port @@ -127,6 +138,7 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: name=g.get("name", ""), shared_dir=g.get("shared_dir", ""), visibility=g.get("visibility", "private"), + join_policy=g.get("join_policy", "invite"), quic_port=g.get("quic_port", cfg.node.quic_port), )) # Back-compat: single [group] section diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 46a1716..7f84abe 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -45,7 +45,8 @@ from meshbay_node.chat.store import ChatStore 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 +from meshbay_node.keystore import load_or_create_keystore +from meshbay_node.roster import Roster from meshbay_node.transport import ( Denylist, QUIC_AVAILABLE, @@ -117,6 +118,7 @@ class NodeDaemon: self._chat_stores: dict[str, ChatStore] = {} self._audit_store: AuditStore | None = None self._bundle_store: BundleStore | None = None + self._roster: Roster | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] self._hub: HubClient | None = None @@ -174,6 +176,14 @@ class NodeDaemon: await self._bundle_store.open() log.info("Bundle store opened: %s", data_dir / "bundles.db") + # 4b. Roster — who this node recognises and which keys are theirs. + # Node authority is established here, locally, and never learned from + # the hub: a hub that could name the operator's key could install + # itself as node administrator. + self._roster = Roster(db_path=data_dir / "roster.db") + await self._roster.open() + await self._roster.purge_expired() + # X25519 key material for GEK unwrapping from cryptography.hazmat.primitives import serialization sk_x_raw = keys.sk_x25519.private_bytes( @@ -223,6 +233,9 @@ class NodeDaemon: "shared_root": shared_root, "index": indexer.index, "visibility": group_cfg.visibility, + # Admission policy comes from node.toml, never from the hub: + # a hub that could declare a group open would be handed its key. + "join_policy": group_cfg.join_policy, } if not groups_ctx: @@ -272,12 +285,20 @@ class NodeDaemon: 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) + self._webrtc._ctx["roster"] = self._roster + admin_pk = self._legacy_admin_pk() + paired = await self._roster.has_operator() if self._roster else False if admin_pk: self._webrtc._ctx["admin_pk_ed25519"] = admin_pk - log.info("Admin Ed25519 key pinned for node sovereignty") + self._webrtc._ctx["has_admin_authority"] = paired + if paired or admin_pk: + sources = ([] if not paired else ["paired operator"]) + \ + ([] if not admin_pk else ["node.toml admin_pk"]) + log.info("Node authority: %s", " + ".join(sources)) else: - log.warning("No admin_pk_ed25519 — admin operations disabled") + log.warning( + "No operator paired — invites and file deletion are " + "refused. Run: meshbay-node operator pair") log.info("WebRTC transport ready") else: log.warning("WebRTC not available (aiortc not installed)") @@ -358,6 +379,8 @@ class NodeDaemon: self._state["groups_ctx"] = groups_ctx self._state["audit_store"] = self._audit_store self._state["bundle_store"] = self._bundle_store + self._state["roster"] = self._roster + self._state["node_user_id"] = session.user_id self._state["webrtc"] = self._webrtc self._state["hub"] = hub self._state["pk_x25519_raw"] = pk_x_raw @@ -457,21 +480,25 @@ class NodeDaemon: 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 + def _legacy_admin_pk(self) -> Ed25519PublicKey | None: + """ + The pre-roster way of naming the operator: `admin_pk_ed25519` in node.toml. - 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 + Still honoured so a deployment configured that way keeps working, but no + longer the only path — and the auto-pin that used to stand in for it is + gone. It pinned the node's *keystore* key while the browser signed with the + user's *identity* key, so admin operations failed closed with a signature + error that looked like a bug elsewhere (finding M3). An operator now pairs + a browser with `meshbay-node operator pair`. + """ + if not self._config.admin_pk_ed25519: + return None + 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 async def _on_index_change(self, indexer: DirectoryIndexer) -> None: """Called when a DirectoryIndexer detects file changes.""" @@ -554,6 +581,9 @@ class NodeDaemon: if self._bundle_store: await self._bundle_store.close() + if self._roster: + await self._roster.close() + for store in self._chat_stores.values(): await store.close() @@ -570,6 +600,60 @@ class NodeDaemon: log.info("Node stopped") +# ── CLI helpers ─────────────────────────────────────────────────────────────── + +def _daemon_api(cfg: Config, path: str, method: str = "GET", + timeout: int = 30) -> dict: + """ + Call the daemon's loopback API. + + The daemon owns the roster, the hub session and the live group contexts, so + the CLI asks it to act rather than opening its databases behind its back. It + also means every operator action goes through the same authorization as the + admin UI (the per-run session token, 11.5.3). + """ + import json as _json + import urllib.error + import urllib.parse + import urllib.request + + token_file = cfg.data_dir / "ui-token" + if not token_file.exists(): + print("Node is not running — start it with: meshbay-node") + sys.exit(1) + + sep = "&" if "?" in path else "?" + url = (f"http://127.0.0.1:{cfg.node.ui_port}{path}" + f"{sep}t={token_file.read_text().strip()}") + try: + req = urllib.request.Request(url, method=method) + with urllib.request.urlopen(req, timeout=timeout) as r: + return _json.loads(r.read()) + except urllib.error.HTTPError as e: + body = e.read().decode()[:300] + try: + detail = _json.loads(body).get("error", body) + except Exception: + detail = body + print(f"failed: {detail}") + sys.exit(1) + except Exception as e: + print(f"failed: {e}") + sys.exit(1) + + +def _resolve_group(cfg: Config, group: str | None) -> str: + """The group argument, or the only configured one.""" + if group: + return group + configured = [g.id for g in cfg.groups if g.id] + if len(configured) == 1: + return configured[0] + print("--group is required (several groups configured)" + if configured else "no group configured in node.toml") + sys.exit(1) + + # ── Entry point ─────────────────────────────────────────────────────────────── def main() -> None: @@ -577,19 +661,23 @@ def main() -> None: parser = argparse.ArgumentParser(description="MeshBay Node daemon") parser.add_argument("command", nargs="?", - choices=["init", "status", "ui", "gek-init", "calibrate-argon2"], + choices=["init", "status", "ui", "gek-init", "operator", + "calibrate-argon2"], help="init: write example config | status: node state and keys " - "| ui: print the admin UI URL | calibrate-argon2: benchmark") + "| ui: print the admin UI URL | operator pair: pair a " + "browser with this node | calibrate-argon2: benchmark") + parser.add_argument("subcommand", nargs="?", + help="'pair' for the operator command") parser.add_argument("--config", type=Path, default=None, help="Config file path") parser.add_argument("--group", default=None, - help="group id for gek-init (optional if only one is configured)") + help="group id (optional if only one is configured)") parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) args = parser.parse_args() # Query commands print a report; library logging would interleave with it. - quiet = args.command in ("status", "ui", "gek-init") + quiet = args.command in ("status", "ui", "gek-init", "operator") logging.basicConfig( level=logging.ERROR if quiet else getattr(logging, args.log_level), format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", @@ -649,50 +737,79 @@ def main() -> None: for g in cfg.groups: print(f" group {g.name} [{g.visibility}] {g.id or '<no id>'}") print(f" {g.shared_dir or '<no shared_dir>'}") - if not cfg.admin_pk_ed25519: - print("admin key NOT pinned — file deletion and member invites will be") - print(" refused (node.toml: admin_pk_ed25519)") - return - - if args.command == "gek-init": - import json as _json - import urllib.error - import urllib.request + # Node authority: the roster is the source of truth, node.toml the legacy + # form. Read the DB directly so this reports correctly while the daemon is + # stopped — the state an operator is most often in when checking. + import asyncio as _asyncio - cfg = load_config(args.config or DEFAULT_CONFIG_PATH) - group_id = args.group - if not group_id: - if len(cfg.groups) == 1: - group_id = cfg.groups[0].id - else: - print("--group is required (several groups configured)") - sys.exit(1) + from meshbay_node.roster import Roster as _Roster - token_file = cfg.data_dir / "ui-token" - if not token_file.exists(): - print("Node is not running — start it with: meshbay-node") - sys.exit(1) + async def _read_roster() -> tuple[list, int]: + r = _Roster(db_path=cfg.data_dir / "roster.db") + await r.open() + try: + return (await r.list_members()), len(await r.list_invites()) + finally: + await r.close() - # The daemon holds the hub session and the live group contexts, so the CLI - # asks it to do the work rather than duplicating it. Same operation as the - # admin UI button — an operator on a headless host should never need a - # browser on that host to initialise a group key. - url = (f"http://127.0.0.1:{cfg.node.ui_port}/api/groups/{group_id}/gek" - f"?t={token_file.read_text().strip()}") try: - req = urllib.request.Request(url, method="POST") - with urllib.request.urlopen(req, timeout=60) as r: - out = _json.loads(r.read()) - except urllib.error.HTTPError as e: - print(f"failed: {e.code} {e.read().decode()[:300]}") - sys.exit(1) + members, pending = _asyncio.run(_read_roster()) + except Exception as e: + members, pending = [], 0 + print(f"roster <unreadable: {e}>") + + operators = [m for m in members if m["role"] == "operator" + and m["status"] == "active"] + if operators: + for op in operators: + print(f"operator {op.get('username') or op['user_id'][:8]}" + f" key {(op.get('pk_ed25519') or '')[:16]}…" + f" paired {op.get('pinned_at', '?')}") + elif cfg.admin_pk_ed25519: + print("operator node.toml admin_pk_ed25519 (legacy)") + print(" run `meshbay-node operator pair` to replace it") + else: + print("operator NONE PAIRED — file deletion and member invites are") + print(" refused. Run: meshbay-node operator pair") + if pending: + print(f"invites {pending} pending code(s)") + return + + if args.command == "gek-init": + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + group_id = _resolve_group(cfg, args.group) + out = _daemon_api(cfg, f"/api/groups/{group_id}/gek", + method="POST", timeout=60) print(f"GEK ready for {group_id}") - print(f" wrapped for {out.get('wrapped_count')}/{out.get('total_members')} members") + print(f" {out.get('authorized_members', 0)} authorized member(s) — each " + f"receives the key on connect") for err in out.get("errors") or []: print(f" ! {err}") return + if args.command == "operator": + if args.subcommand != "pair": + print("usage: meshbay-node operator pair") + sys.exit(1) + + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + out = _daemon_api(cfg, "/api/operator/pair", method="POST") + + from meshbay_node.roster import write_code_file + path = write_code_file(cfg.data_dir, out["code"], out.get("expires_at", "")) + + print(f"PAIRING CODE {out['code']}") + print(f"valid until {out.get('expires_at', '?')}") + print() + print("Sign in to the web app as this node's operator, open one of your") + print("groups, go to the Members tab and enter the code there.") + print("It works once, for that account only, and authorizes invites and") + print("file deletion from that browser.") + print() + print(f"also written to {path}") + return + if args.command == "ui": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) token_file = cfg.data_dir / "ui-token" diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py new file mode 100644 index 0000000..f231792 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -0,0 +1,372 @@ +""" +Node roster — who this node recognises, and which keys are theirs. + +The node keeps its own answer to "may this person have the group key", derived from +what the operator authorized locally. It is deliberately NOT derived from the hub: +the hub decides group membership, and a hub that invents an account and mints a +token for it would otherwise collect the GEK on connect. Hub membership is an input +to the decision; it is not the decision. + +Three tables: + + identities — one row per person, not per group. Someone paired for one group + needs no code for the next one on the same node. + members — role and status per (group, user). + invites — one-time pairing codes, stored as a hash. The code itself exists + only in the operator's hands and the invitee's. + +The code is what binds a public key to an account without asking the hub +(finding H3). See `docs/invite-pairing-v1.md`. +""" + +from __future__ import annotations + +import hashlib +import logging +import os +import secrets +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import aiosqlite + +log = logging.getLogger(__name__) + +# Crockford base32 without I, L, O and U: no character pair a human can confuse +# when reading a code aloud or typing it from a phone screen. +_ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" +CODE_LEN = 8 # 8 × 5 bits = 40 bits of entropy +DEFAULT_INVITE_TTL = 24 * 3600 # seconds + +_SCHEMA = """\ +CREATE TABLE IF NOT EXISTS identities ( + user_id TEXT PRIMARY KEY, + username TEXT NOT NULL, + pk_ed25519 TEXT NOT NULL, + pk_x25519 TEXT NOT NULL, + pinned_at TEXT NOT NULL, + pinned_via TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS members ( + group_id TEXT NOT NULL, + user_id TEXT NOT NULL, + role TEXT NOT NULL, + status TEXT NOT NULL, + approved_by TEXT NOT NULL, + approved_at TEXT NOT NULL, + PRIMARY KEY (group_id, user_id) +); + +CREATE TABLE IF NOT EXISTS invites ( + code_hash TEXT PRIMARY KEY, + group_id TEXT NOT NULL, + user_id TEXT NOT NULL, + role TEXT NOT NULL, + created_by TEXT NOT NULL, + created_at TEXT NOT NULL, + expires_at TEXT NOT NULL, + used_at TEXT +); +""" + + +def generate_code() -> str: + """A fresh pairing code, formatted for a human to read out: XXXX-XXXX.""" + raw = "".join(secrets.choice(_ALPHABET) for _ in range(CODE_LEN)) + return f"{raw[:4]}-{raw[4:]}" + + +def normalize_code(code: str) -> str: + """ + Fold what a human typed onto what was generated. + + Crockford's rules: case-insensitive, dashes and spaces are decoration, and the + excluded letters map onto the digits they resemble. Someone reading a code over + the phone should not be able to get it wrong in a way we could have absorbed. + """ + out = [] + for ch in code.upper(): + if ch in "- \t": + continue + if ch in "IL": + out.append("1") + elif ch == "O": + out.append("0") + elif ch == "U": + out.append("V") + else: + out.append(ch) + return "".join(out) + + +def hash_code(code: str) -> str: + """ + Store codes hashed: a stolen roster DB must not yield usable invitations. + + SHA-256 rather than a password KDF on purpose — the input is 40 bits of + uniformly random secret, not a human-chosen string, so there is nothing for a + slow hash to defend. + """ + return hashlib.sha256(normalize_code(code).encode()).hexdigest() + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +class Roster: + 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)) + self._db.row_factory = aiosqlite.Row + # WAL: the CLI writes invites (`operator pair`) while the daemon reads them. + await self._db.execute("PRAGMA journal_mode=WAL") + await self._db.executescript(_SCHEMA) + await self._db.commit() + + async def close(self) -> None: + if self._db: + await self._db.close() + self._db = None + + # ── Identities ─────────────────────────────────────────────────────────── + + async def pin_identity( + self, + user_id: str, + username: str, + pk_ed25519: str, + pk_x25519: str, + via: str, + ) -> None: + assert self._db + await self._db.execute( + "INSERT OR REPLACE INTO identities " + "(user_id, username, pk_ed25519, pk_x25519, pinned_at, pinned_via) " + "VALUES (?, ?, ?, ?, ?, ?)", + (user_id, username, pk_ed25519, pk_x25519, _now(), via), + ) + await self._db.commit() + + async def get_identity(self, user_id: str) -> dict | None: + assert self._db + async with self._db.execute( + "SELECT * FROM identities WHERE user_id = ?", (user_id,) + ) as cur: + row = await cur.fetchone() + return dict(row) if row else None + + async def unpin(self, user_id: str) -> bool: + assert self._db + cur = await self._db.execute( + "DELETE FROM identities WHERE user_id = ?", (user_id,)) + await self._db.commit() + return cur.rowcount > 0 + + async def list_identities(self) -> list[dict]: + assert self._db + async with self._db.execute( + "SELECT * FROM identities ORDER BY pinned_at" + ) as cur: + return [dict(r) for r in await cur.fetchall()] + + # ── Authority ──────────────────────────────────────────────────────────── + + async def operator_pks(self) -> list[str]: + """ + Base64 Ed25519 keys allowed to authorize admin operations on this node. + + Read fresh on every check rather than cached: an unpin must take effect at + once, and this runs only on admin operations, which are rare. + """ + assert self._db + async with self._db.execute( + "SELECT i.pk_ed25519 FROM identities i " + "JOIN members m ON m.user_id = i.user_id " + "WHERE m.role = 'operator' AND m.status = 'active'" + ) as cur: + return [r["pk_ed25519"] for r in await cur.fetchall()] + + async def has_operator(self) -> bool: + return bool(await self.operator_pks()) + + async def is_authorized(self, group_id: str, user_id: str) -> bool: + """ + May this person be handed the group key? + + The node's own answer, not the hub's. Hub membership is what lets someone + reach the node; this is what decides whether the key is wrapped for them — + otherwise a hub that invents an account and mints a token for it would be + served the GEK on connect. + + An operator is authorized for every group this node hosts: their authority + is node-wide and is recorded with an empty group_id. + """ + assert self._db + async with self._db.execute( + "SELECT 1 FROM members WHERE user_id = ? AND status = 'active' " + "AND (group_id = ? OR (group_id = '' AND role = 'operator')) LIMIT 1", + (user_id, group_id), + ) as cur: + return await cur.fetchone() is not None + + # ── Members ────────────────────────────────────────────────────────────── + + async def set_member( + self, + group_id: str, + user_id: str, + role: str, + status: str, + approved_by: str, + ) -> None: + assert self._db + await self._db.execute( + "INSERT OR REPLACE INTO members " + "(group_id, user_id, role, status, approved_by, approved_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + (group_id, user_id, role, status, approved_by, _now()), + ) + await self._db.commit() + + async def get_member(self, group_id: str, user_id: str) -> dict | None: + assert self._db + async with self._db.execute( + "SELECT * FROM members WHERE group_id = ? AND user_id = ?", + (group_id, user_id), + ) as cur: + row = await cur.fetchone() + return dict(row) if row else None + + async def list_members(self, group_id: str | None = None) -> list[dict]: + assert self._db + sql = ( + "SELECT m.*, i.username, i.pk_ed25519, i.pinned_at, i.pinned_via " + "FROM members m LEFT JOIN identities i ON i.user_id = m.user_id" + ) + args: tuple = () + if group_id is not None: + sql += " WHERE m.group_id = ?" + args = (group_id,) + async with self._db.execute(sql + " ORDER BY m.approved_at", args) as cur: + return [dict(r) for r in await cur.fetchall()] + + async def set_status(self, group_id: str, user_id: str, status: str) -> bool: + assert self._db + cur = await self._db.execute( + "UPDATE members SET status = ? WHERE group_id = ? AND user_id = ?", + (status, group_id, user_id), + ) + await self._db.commit() + return cur.rowcount > 0 + + # ── Invites ────────────────────────────────────────────────────────────── + + async def create_invite( + self, + group_id: str, + user_id: str, + role: str, + created_by: str, + ttl: int = DEFAULT_INVITE_TTL, + ) -> str: + """ + Issue a one-time code. Returns it in the clear — this is the only moment it + exists outside the operator's hands; only its hash is kept. + + Any earlier unused invite for the same person and group is dropped, so + re-inviting supersedes rather than accumulating valid codes. + """ + assert self._db + await self._db.execute( + "DELETE FROM invites WHERE group_id = ? AND user_id = ? AND used_at IS NULL", + (group_id, user_id), + ) + code = generate_code() + expires = datetime.now(timezone.utc) + timedelta(seconds=ttl) + await self._db.execute( + "INSERT INTO invites " + "(code_hash, group_id, user_id, role, created_by, created_at, expires_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (hash_code(code), group_id, user_id, role, created_by, _now(), + expires.isoformat(timespec="seconds")), + ) + await self._db.commit() + return code + + async def consume_invite(self, code: str, user_id: str) -> dict | None: + """ + Redeem a code for `user_id`, or return None. + + Single use is enforced by the UPDATE's WHERE clause: two connections racing + the same code cannot both see `used_at IS NULL`, so exactly one wins. + """ + assert self._db + code_hash = hash_code(code) + async with self._db.execute( + "SELECT * FROM invites WHERE code_hash = ?", (code_hash,) + ) as cur: + row = await cur.fetchone() + if not row: + return None + + invite = dict(row) + if invite["used_at"] is not None: + return None + # A code is valid for exactly one account, so a leaked code cannot be + # redeemed by whoever finds it first. + if invite["user_id"] != user_id: + return None + if datetime.fromisoformat(invite["expires_at"]) < datetime.now(timezone.utc): + return None + + cur = await self._db.execute( + "UPDATE invites SET used_at = ? WHERE code_hash = ? AND used_at IS NULL", + (_now(), code_hash), + ) + await self._db.commit() + if cur.rowcount == 0: + return None + return invite + + async def list_invites(self, include_used: bool = False) -> list[dict]: + assert self._db + sql = "SELECT * FROM invites" + if not include_used: + sql += " WHERE used_at IS NULL" + async with self._db.execute(sql + " ORDER BY created_at") as cur: + return [dict(r) for r in await cur.fetchall()] + + async def purge_expired(self) -> int: + assert self._db + cur = await self._db.execute( + "DELETE FROM invites WHERE used_at IS NULL AND expires_at < ?", + (_now(),), + ) + await self._db.commit() + return cur.rowcount + + +async def open_roster(data_dir: Path) -> Roster: + roster = Roster(data_dir / "roster.db") + await roster.open() + return roster + + +def write_code_file(data_dir: Path, code: str, expires_at: str) -> Path: + """ + Leave the code in a file as well as on stdout. + + An operator working over SSH may not be able to copy out of their terminal, + and a code that can only be read off a scrolled-away screen is a dead end. + """ + path = data_dir / "pair-code" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(f"{code}\nexpires {expires_at}\n") + os.chmod(path, 0o600) + return path 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 34a96bd..fe3ee2e 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -57,10 +57,16 @@ from meshbay_common.handshake import ( from meshbay_common.adminop import ( ADMIN_CHALLENGE_TTL, OP_FILE_DELETE, - OP_GEK_BUNDLE_STORE, + OP_INVITE_CREATE, admin_transcript, ) -from meshbay_common.crypto import pk_to_b64 +from meshbay_common.crypto import pk_to_b64, wrap_gek_aes +from meshbay_common.join import ( + JOIN_TTL, + ROLE_MEMBER, + ROLE_OPERATOR, + join_transcript, +) from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex @@ -85,6 +91,14 @@ MAX_CONCURRENT_TRANSCODES = 2 # Bundle fetches are served in the pre-proof window (C4). Bounded and audited # until the native client removes remote keypair bundles entirely. MAX_PRE_PROOF_FETCHES = 4 +# Pairing codes carry 40 bits and are single-use, but a connection must not be +# allowed to sit there guessing. Failures are audited, so a grind is visible. +MAX_JOIN_ATTEMPTS = 5 +# Per-connection limits alone would not bind an attacker who can open connections +# at will — and the adversary who can mint tokens for any account is the hub. So +# failed pairings are also counted node-wide over a window. +MAX_JOIN_FAILURES_WINDOW = 20 +JOIN_FAILURE_WINDOW = 600 # seconds UPLOAD_DIR_NAME = ".uploads" # Conservative allowlist: also what keeps markup out of filenames, which the node admin # UI used to render unescaped (finding H2). @@ -218,6 +232,11 @@ class WebRTCPeerSession: self._username: str = "" self._pk_user: str = "" self._gek_challenge: bytes | None = None + # Same value as the GEK challenge, but kept for the life of the connection: + # a join_request is signed over it, and it must stay verifiable after the + # handshake clears the challenge (an operator pairs while already connected). + self._nonce_node: bytes = b"" + self._join_attempts = 0 self._nonce_client: bytes = b"" self._admin_ops: dict[str, dict] = {} # op_id → pending admin operation self._uploads: dict[str, dict] = {} # filename → {next_index, bytes} @@ -259,6 +278,12 @@ class WebRTCPeerSession: asyncio.ensure_future(self._do_gek_bundle_fetch()) else: asyncio.ensure_future(self._do_keypair_bundle_fetch()) + elif mtype == MNP.JOIN_REQUEST and self._nonce_node: + # Valid both before the GEK proof (a new member has no GEK to prove + # with) and after it (an operator pairing a browser is already + # connected). Authority comes from the pairing code and the + # signature, never from the session state. + asyncio.ensure_future(self._do_join_request(msg)) elif self._user_id is None: self._send({"type": "error", "detail": "Handshake required"}) elif mtype == MNP.INDEX_SYNC: @@ -277,8 +302,8 @@ class WebRTCPeerSession: self._do_file_delete(msg) elif mtype == MNP.ADMIN_RESPONSE: self._do_admin_response(msg) - elif mtype == MNP.GEK_BUNDLE_STORE: - self._do_gek_bundle_store(msg) + elif mtype == MNP.INVITE_CREATE: + self._do_invite_create(msg) elif mtype == MNP.KEYPAIR_BUNDLE_STORE: asyncio.ensure_future(self._do_keypair_bundle_store(msg)) elif mtype == MNP.STREAM_REQUEST: @@ -359,6 +384,7 @@ class WebRTCPeerSession: return self._gek_challenge = os.urandom(NONCE_LEN) + self._nonce_node = self._gek_challenge self._send({ "type": MNP.HANDSHAKE_CHALLENGE, "v": MNP_VERSION, @@ -472,49 +498,40 @@ class WebRTCPeerSession: else: self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) - def _do_gek_bundle_store(self, msg: dict) -> None: + def _do_invite_create(self, msg: dict) -> None: """ - Request to store a wrapped GEK bundle for a target user. - - Finding C5b: this used to write whatever any authenticated member sent, with - INSERT OR REPLACE semantics, and then auto-activate the bundle if it was - addressed to the node operator. Since the operator's X25519 public key is - public — the node even hands it out in handshake_ack — any member could wrap - a GEK of their own choosing for the operator and make the node adopt it, - locking every legitimate member out of the group and taking over the key. + Issue a one-time pairing code for someone the operator wants to admit. - Storing a bundle is now a node-operator operation gated by an Ed25519 - challenge, and nothing arriving over MNP can activate a GEK: activation - happens only through the local admin UI or the CLI. + Replaces the old invite path, where the inviter fetched the invitee's + public key from the hub and wrapped the group key for whatever came back + (H3). The node now needs nothing but a name: it will wrap the key itself, + later, for a key the invitee proves they hold. """ - bundle_store = self._ctx.get("bundle_store") - if not bundle_store: - self._send({"type": "error", "detail": "Bundle store not available"}) + roster = self._ctx.get("roster") + if roster is None: + self._send({"type": "error", "detail": "Roster not available"}) return - target_user_id = msg.get("user_id", "") + invitee_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"}) + if not invitee_id or not group_id: + self._send({"type": "error", "detail": "Missing user_id or group_id"}) + return + if group_id != self._group_id: + self._send({"type": "error", "detail": "Wrong group for this session"}) return - if not self._ctx.get("admin_pk_ed25519"): + if not self._has_admin_authority(): self._send({ "type": "error", - "detail": "No admin key pinned — bundle storage refused", + "detail": "No operator paired — run `meshbay-node operator pair`", }) return - self._issue_admin_challenge(OP_GEK_BUNDLE_STORE, target_user_id, { + self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, { "group_id": group_id, - "user_id": target_user_id, - "pk_eph_b64": pk_eph, - "nonce_b64": nonce, - "wrapped_b64": wrapped, + "user_id": invitee_id, + "username": str(msg.get("username", ""))[:64], }) async def _do_keypair_bundle_fetch(self) -> None: @@ -560,6 +577,240 @@ class WebRTCPeerSession: "detail": "keypair_bundle_stored", }) + # ── Pairing and join (H3, M3) ──────────────────────────────────────────── + + def _join_refuse(self, reason: str, audit_detail: str = "") -> None: + self._join_attempts += 1 + # Node-wide window, shared across connections: reconnecting must not reset + # the budget. + now = time.time() + failures = [t for t in self._ctx.get("join_failures", []) + if now - t < JOIN_FAILURE_WINDOW] + failures.append(now) + self._ctx["join_failures"] = failures + self._audit_join("join_refused", audit_detail or reason) + self._send({ + "type": MNP.JOIN_RESULT, + "v": MNP_VERSION, + "ok": False, + "reason": reason, + }) + + def _audit_join(self, event: str, detail: str) -> None: + audit = self._ctx.get("audit_store") + if not audit: + return + self._remote_ip = self._remote_ip or _get_remote_ip(self._pc) + asyncio.ensure_future(audit.log_event( + user_id=self._user_id or getattr(self, "_pending_sub", "unknown"), + event=event, + ip=self._remote_ip, + username=self._username or getattr(self, "_pending_username", ""), + group_id=self._group_id or getattr(self, "_pending_group", "") or "", + detail=detail, + )) + + async def _do_join_request(self, msg: dict) -> None: + """ + Pin an identity, or recognise one already pinned. + + The client signs its own Ed25519 and X25519 keys together with the node's + nonce, so the identity key vouches for the encryption key — that is what + will make it safe for the node to wrap the GEK for a key that arrived over + the wire instead of one fetched from the hub's directory (H3). + + A first pairing needs a one-time code, which the hub never sees. Afterwards + the pin is the credential and a changed key is refused outright, the same + rule the client applies to `pk_node` (11.5.8). + """ + roster = self._ctx.get("roster") + if roster is None: + self._send({"type": "error", "detail": "Roster not available"}) + return + + if self._join_attempts >= MAX_JOIN_ATTEMPTS: + self._send({"type": "error", "detail": "Too many attempts"}) + return + + now = time.time() + recent = [t for t in self._ctx.get("join_failures", []) + if now - t < JOIN_FAILURE_WINDOW] + if len(recent) >= MAX_JOIN_FAILURES_WINDOW: + self._audit_join("join_throttled", f"{len(recent)} failures in window") + self._send({"type": "error", "detail": "Pairing temporarily locked"}) + return + + user_id = self._user_id or getattr(self, "_pending_sub", "") + username = self._username or getattr(self, "_pending_username", "") + if not user_id: + self._send({"type": "error", "detail": "Handshake required"}) + return + + pk_ed_b64 = msg.get("pk_ed25519", "") + pk_x_b64 = msg.get("pk_x25519", "") + code = msg.get("code", "") + ts = msg.get("ts", 0) + + try: + pk_ed_raw = base64.b64decode(pk_ed_b64) + pk_x_raw = base64.b64decode(pk_x_b64) + if len(pk_ed_raw) != 32 or len(pk_x_raw) != 32: + raise ValueError + pk_ed = Ed25519PublicKey.from_public_bytes(pk_ed_raw) + except Exception: + self._join_refuse("invalid_keys") + return + + if not isinstance(ts, int) or abs(time.time() - ts) > JOIN_TTL: + self._join_refuse("stale_request") + return + + # An empty group_id means operator pairing, which is node-wide. Anything + # else must be the group this connection authenticated to — a signature + # obtained for one group must not name another. + group_id = msg.get("group_id", "") or "" + session_group = self._group_id or getattr(self, "_pending_group", "") or "" + if group_id and group_id != session_group: + self._join_refuse("group_mismatch") + return + + transcript = join_transcript( + node_pk_b64=self._node_pk_b64(), + group_id=group_id, + user_id=user_id, + pk_ed25519_b64=pk_ed_b64, + pk_x25519_b64=pk_x_b64, + nonce_node=self._nonce_node, + ts=ts, + ) + try: + sig = base64.b64decode(msg.get("sig", "")) + except Exception: + self._join_refuse("invalid_signature_encoding") + return + if not self._verify_sig(pk_ed, transcript, sig): + self._join_refuse("signature_invalid") + return + + known = await roster.get_identity(user_id) + if known: + if known["pk_ed25519"] != pk_ed_b64 or known["pk_x25519"] != pk_x_b64: + # The blocking warning, raised where it matters: whoever this is + # holds a different key than the person the operator paired. + self._join_refuse( + "key_changed", + f"pinned={known['pk_ed25519'][:16]} presented={pk_ed_b64[:16]}") + return + member = await roster.get_member(group_id, user_id) + await self._join_ok( + user_id, pk_x_raw, session_group, + role=member["role"] if member else "", + recognised=True, + ) + return + + if not code: + if self._group_join_policy(session_group) == "open": + # An open-join group admits anyone the hub calls a member, so a + # code would protect nothing — the hub can walk in through the + # front door. Pin what turns up and say so in the audit log. + await self._pin_and_admit( + roster, user_id, username, pk_ed_b64, pk_x_b64, + group_id=session_group, role=ROLE_MEMBER, + approved_by="open-join", via="tofu") + await self._join_ok(user_id, pk_x_raw, session_group, + role=ROLE_MEMBER, recognised=False) + return + self._join_refuse("code_required") + return + + invite = await roster.consume_invite(code, user_id) + if not invite: + self._join_refuse("code_invalid") + return + + await self._pin_and_admit( + roster, user_id, username, pk_ed_b64, pk_x_b64, + group_id=invite["group_id"], role=invite["role"], + approved_by=invite["created_by"], via="code") + await self._join_ok(user_id, pk_x_raw, invite["group_id"], + role=invite["role"], recognised=False) + + def _group_join_policy(self, group_id: str) -> str: + """ + Admission policy for a group, read from the node's own configuration. + + Never from the hub: a hub that could declare a group open would be handed + the key to it (§3.4 of docs/invite-pairing-v1.md). + """ + gctx = (self._ctx.get("groups") or {}).get(group_id) or {} + return gctx.get("join_policy", "invite") + + async def _pin_and_admit( + self, roster, user_id: str, username: str, pk_ed_b64: str, pk_x_b64: str, + *, group_id: str, role: str, approved_by: str, via: str, + ) -> None: + await roster.pin_identity( + user_id=user_id, username=username, + pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via=via, + ) + await roster.set_member( + group_id=group_id, user_id=user_id, role=role, + status="active", approved_by=approved_by, + ) + if role == ROLE_OPERATOR: + self._ctx["has_admin_authority"] = True + + log.info("Identity pinned (%s): user=%s role=%s", via, user_id[:8], role) + self._audit_join("join_pinned", f"role={role} via={via}") + + async def _join_ok( + self, user_id: str, pk_x_raw: bytes, group_id: str, + *, role: str, recognised: bool, + ) -> None: + """ + Answer a join, wrapping the group key for the key the caller just proved. + + This is the H3 fix. The inviter used to fetch the invitee's public key from + the hub and wrap the GEK for whatever came back, so a hub that answered + with its own key was handed the group key by an honest member following the + protocol exactly. The node now wraps for a key that arrived from its owner + over an authenticated channel, bound to a pinned identity. + """ + reply = { + "type": MNP.JOIN_RESULT, + "v": MNP_VERSION, + "ok": True, + "recognised": recognised, + "role": role, + } + + roster = self._ctx["roster"] + if group_id and not await roster.is_authorized(group_id, user_id): + # Pinned on this node, but not admitted to this group. Hub membership + # alone must not produce a key. + reply["gek"] = False + reply["reason"] = "not_authorized_for_group" + self._send(reply) + self._audit_join("join_no_gek", f"group={group_id[:8]} not authorized") + return + + gctx = (self._ctx.get("groups") or {}).get(group_id) or {} + gek = gctx.get("gek") + if not gek: + reply["gek"] = False + reply["reason"] = "no_gek" + self._send(reply) + return + + bundle = wrap_gek_aes(gek, pk_x_raw) + reply["gek"] = True + reply["pk_eph_b64"] = bundle["pk_eph_b64"] + reply["nonce_b64"] = bundle["nonce_b64"] + reply["wrapped_b64"] = bundle["wrapped_b64"] + self._send(reply) + self._audit_join("gek_wrapped", f"group={group_id[:8]}") + def _audit_pre_proof_fetch(self, mtype: str) -> None: """Record bundle access made before the GEK proof (C4).""" audit = self._ctx.get("audit_store") @@ -903,9 +1154,8 @@ class WebRTCPeerSession: 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: + if not self._has_admin_authority() and not has_uploader_pk: self._send({"type": "error", "detail": "No authorized key for deletion"}) return @@ -956,6 +1206,43 @@ class WebRTCPeerSession: except Exception: return False + def _has_admin_authority(self) -> bool: + """ + Cheap synchronous pre-check: is there anyone who could authorize this? + + Only decides whether to issue a challenge at all — the gate is + `_verify_admin_sig`. The flag is set at startup and refreshed in-process + when an operator pairs. + """ + return bool(self._ctx.get("admin_pk_ed25519") + or self._ctx.get("has_admin_authority")) + + async def _verify_admin_sig(self, transcript: bytes, sig: bytes) -> bool: + """ + Check a signature against every key holding node-operator authority. + + Read from the roster on each call rather than cached: revoking a paired + browser must take effect immediately, and admin operations are rare enough + that a SQLite read costs nothing. `admin_pk_ed25519` in node.toml is still + honoured so an existing deployment keeps working until its operator pairs + (M3) — it is the legacy form of the same statement. + """ + legacy = self._ctx.get("admin_pk_ed25519") + if self._verify_sig(legacy, transcript, sig): + return True + + roster = self._ctx.get("roster") + if roster is None: + return False + for pk_b64 in await roster.operator_pks(): + try: + pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_b64)) + except Exception: + continue + if self._verify_sig(pk, transcript, sig): + return True + return False + def _do_admin_response(self, msg: dict) -> None: op_id = msg.get("op_id", "") sig_b64 = msg.get("signature", "") @@ -985,14 +1272,15 @@ class WebRTCPeerSession: ) if pending["op"] == OP_FILE_DELETE: - self._admin_exec_file_delete(pending, transcript, sig_bytes) - elif pending["op"] == OP_GEK_BUNDLE_STORE: asyncio.ensure_future( - self._admin_exec_bundle_store(pending, transcript, sig_bytes)) + self._admin_exec_file_delete(pending, transcript, sig_bytes)) + elif pending["op"] == OP_INVITE_CREATE: + asyncio.ensure_future( + self._admin_exec_invite_create(pending, transcript, sig_bytes)) else: self._send({"type": "error", "detail": "Unknown admin operation"}) - def _admin_exec_file_delete( + async def _admin_exec_file_delete( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: file_id = pending["subject"] @@ -1012,7 +1300,7 @@ class WebRTCPeerSession: # Node operator, or the user who uploaded this file — verified by the key # recorded at upload time, never by a JWT claim (the hub controls those). - if not (self._verify_sig(self._ctx.get("admin_pk_ed25519"), transcript, sig) + if not (await self._verify_admin_sig(transcript, sig) or self._verify_sig(uploader_pk, transcript, sig)): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") @@ -1020,33 +1308,46 @@ class WebRTCPeerSession: self._exec_file_delete(ctx, file_id, entry) - async def _admin_exec_bundle_store( + async def _admin_exec_invite_create( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: # Node operator only. A group admin who does not run the node has no - # authority over what this node stores (draft-v4 §4.2.x, deny by default). - if not self._verify_sig(self._ctx.get("admin_pk_ed25519"), transcript, sig): + # authority over who this node admits (deny by default). Delegation is + # designed but deferred — see §6.2 of docs/invite-pairing-v1.md. + if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) - self._audit("admin_auth_failed", f"gek_bundle_store:{pending['subject'][:16]}") + self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}") return - payload = pending["payload"] - bundle_store = self._ctx.get("bundle_store") - if not bundle_store: - self._send({"type": "error", "detail": "Bundle store not available"}) + roster = self._ctx.get("roster") + if roster is None: + self._send({"type": "error", "detail": "Roster not available"}) return - await bundle_store.store( - payload["group_id"], payload["user_id"], - payload["pk_eph_b64"], payload["nonce_b64"], payload["wrapped_b64"], + payload = pending["payload"] + code = await roster.create_invite( + group_id=payload["group_id"], + user_id=payload["user_id"], + role=ROLE_MEMBER, + created_by=self._user_id or "", ) - log.info("GEK bundle stored: group=%s user=%s", + invites = await roster.list_invites() + expires = next( + (i["expires_at"] for i in invites + if i["user_id"] == payload["user_id"] + and i["group_id"] == payload["group_id"]), "") + + log.info("Invite created: group=%s user=%s", payload["group_id"][:8], payload["user_id"][:8]) - self._audit("gek_bundle_store", f"target={payload['user_id'][:8]}") + self._audit("invite_create", f"target={payload['user_id'][:8]}") + # The code exists in the clear exactly here and in the operator's hands. self._send({ - "type": "ack", "v": MNP_VERSION, - "detail": "gek_bundle_stored", + "type": MNP.INVITE_RESULT, + "v": MNP_VERSION, + "code": code, + "expires_at": expires, "user_id": payload["user_id"], + "username": payload.get("username", ""), }) def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index 21e4445..e671b72 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -24,6 +24,7 @@ from fastapi.responses import HTMLResponse, JSONResponse from meshbay_node import __version__ from meshbay_common.crypto import generate_gek, wrap_gek_aes +from meshbay_common.join import ROLE_OPERATOR log = logging.getLogger(__name__) @@ -217,11 +218,61 @@ def create_ui_app(state: dict) -> FastAPI: ], } + # ── Operator pairing (localhost only) ────────────────────────────────── + + @app.post("/api/operator/pair") + async def operator_pair(): + """ + Issue a one-time code that pairs a browser as this node's operator. + + The code is the whole point: it binds the operator's browser identity key + to their account without asking the hub, which is what stops a hub from + naming itself node administrator (M3, and the same substitution as H3). + It is returned once and stored only as a hash. + """ + roster = state.get("roster") + user_id = state.get("node_user_id") + if not roster or not user_id: + return JSONResponse({"error": "Node not connected to hub yet"}, 503) + + code = await roster.create_invite( + group_id="", # operator authority is node-wide + user_id=user_id, + role=ROLE_OPERATOR, + created_by="local-cli", + ) + invites = await roster.list_invites() + expires = next((i["expires_at"] for i in invites + if i["user_id"] == user_id and i["role"] == ROLE_OPERATOR), "") + return {"code": code, "expires_at": expires, "user_id": user_id} + + @app.get("/api/roster") + async def api_roster(): + roster = state.get("roster") + if not roster: + return {"identities": [], "members": [], "pending_invites": 0} + return { + "identities": await roster.list_identities(), + "members": await roster.list_members(), + "pending_invites": len(await roster.list_invites()), + } + # ── 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.""" + """ + Generate the group key and activate it. + + It used to be wrapped here for every member, using public keys fetched from + the hub — which is H3 with the node as the victim instead of the inviter: a + hub answering with its own key was handed the group key by the node itself. + + Nothing is pre-wrapped for members now. Each member's copy is produced when + they connect, for a key they proved they hold (`join_request`). Only the + node's own copy is stored, so the daemon can reload the key across restarts + without the operator's browser. + """ groups_ctx = state.get("groups_ctx", {}) if group_id not in groups_ctx: return JSONResponse({"error": "Group not hosted on this node"}, 404) @@ -234,48 +285,15 @@ def create_ui_app(state: dict) -> FastAPI: 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() + errors: list[str] = [] - 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) + roster = state.get("roster") + authorized = len(await roster.list_members(group_id)) if roster else 0 - 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") + # Store a copy wrapped for the node keystore X25519 key so the daemon can + # reload the GEK on restart without the operator's browser keys. 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: @@ -286,13 +304,14 @@ def create_ui_app(state: dict) -> FastAPI: node_bundle["pk_eph_b64"], node_bundle["nonce_b64"], node_bundle["wrapped_b64"], ) - log.info("GEK also wrapped for node keystore (daemon reload)") + log.info("GEK wrapped for node keystore (daemon reload)") except Exception as e: + errors.append(f"node keystore: {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)) + log.info("GEK initialized for group %s — %d authorized member(s) will " + "receive it on connect", group_id[:8], authorized) webrtc = state.get("webrtc") if webrtc and "groups" in webrtc._ctx and group_id in webrtc._ctx["groups"]: @@ -301,8 +320,7 @@ def create_ui_app(state: dict) -> FastAPI: return { "status": "ok", "group_id": group_id, - "wrapped_count": wrapped_count, - "total_members": len(members), + "authorized_members": authorized, "errors": errors, } @@ -585,8 +603,8 @@ async function initGEK(groupId) {{ const resp = await fetch('/api/groups/' + groupId + '/gek?t=' + TOKEN, {{ 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.textContent = 'GEK initialized — ' + + data.authorized_members + ' authorized member(s) get it on connect'; if (status) status.style.color = '#22c55e'; setTimeout(() => location.reload(), 2000); }} else {{ |