diff options
26 files changed, 2151 insertions, 155 deletions
@@ -96,6 +96,25 @@ Scope: `hub`, `node`, `common`, or omitted for cross-cutting - **Never log GEK, private keys, or plaintext passwords** — even at DEBUG level - **meshbay.org is internet-facing** — open port → test → close port + kill processes in same block +## First security review (2026-08-10) — see `first-review.md` + +**Critical (before Phase 7):** +- **C1** Chat: Sender Keys protocol, NOT shared Double Ratchet (pairwise protocol + would cause key/nonce reuse in group context). `ratchet.py` kept for future 1:1 DM. +- **C2** JWT must carry `"groups": [group_ids]` claim. Node MNP handshake must verify + group membership before serving content. Without this, any authenticated user + accesses any group. + +**Significant (Phase 7-8):** +- **S1** Admin revocation endpoint has no authz check → Phase 8.1 +- **S2** Email stored in plaintext (spec says encrypted at rest) → Phase 8.2 +- **S3** jti denylist push via hub→node WebSocket → Phase 7.2 +- **S4** AES-GCM keystore IV fixed: 128-bit → 96-bit (NIST SP 800-38D) ✅ DONE +- **S5** Refresh token rotation (one-time use) → Phase 8.3 + +**Architecture validated:** crypto primitives, GEK wrapping (ECIES), trust model, +key hierarchy, on-the-fly encryption, transport abstraction. + ## Known calibration TODOs - Argon2id `memory_cost`: currently 65536 (64 MB, 78ms) — increase to 262144 (256 MB) before prod @@ -121,7 +140,8 @@ SFR résidentiel Fedora 44 → meshbay.org OVH VPS : | Dérivation clés depuis password | `meshbay_common.keyderive` | `keyderive.py` | | Bundle clés (web) | `meshbay_common.keyderive` | `keyderive.py` + `static/keyderive.js` | | GEK wrap/unwrap (ECIES) | `meshbay_common.crypto` | `crypto.py` | -| Double Ratchet (chat) | `meshbay_common.ratchet` | `ratchet.py` | +| Double Ratchet (1:1 DM, future) | `meshbay_common.ratchet` | `ratchet.py` | +| Sender Keys (group chat) | `meshbay_common.senderkeys` | `senderkeys.py` (Phase 7.5) | | AES-GCM (navigateur) | `meshbay_common.webcrypto` | `webcrypto.py` + `static/crypto.js` | | Keystore node | `meshbay_node.keystore` | `keystore.py` | | NAT traversal | `QE/demo-v2/nat.py` | non versionné — résultats dans devel-phases.md | @@ -137,3 +157,7 @@ SFR résidentiel Fedora 44 → meshbay.org OVH VPS : - Services légitimes : `meshbay-hub.service`, Caddy, PostgreSQL (local) - Inventaire détaillé : `QE/server-state/meshbay.org.md` - Deploy hub : voir `QE/server-state/meshbay.org.md` + +## new rules, from now +Documents and demo/comments are written in english unless requested in french. + diff --git a/devel-phases-next.md b/devel-phases-next.md index ff161d6..f4793df 100644 --- a/devel-phases-next.md +++ b/devel-phases-next.md @@ -1,88 +1,116 @@ -# MeshBay — Phases d'implémentation suivantes +# MeshBay — Next Implementation Phases -> Base : Phases 1-6 terminées. demo-v2 NAT QUIC validée. -> Référence architecture : docs/meshbay-draft-v3.md +> Base: Phases 1-6 complete. demo-v2 NAT QUIC validated. +> Architecture reference: docs/meshbay-draft-v3.md +> First security review: first-review.md (2026-08-10) --- ## Phase 7 — Node v2 : production, streaming, chat -**Objectif :** un node utilisable quotidiennement — multi-groupe, streaming fluide, -chat intégré, reconnexion rapide. +**Objective:** a node usable for daily operations — multi-group, smooth streaming, +integrated chat, fast reconnection. -### Décisions architecturales (arrêtées) +### Prerequisites (from first security review, 2026-08-10) -**Multi-groupe → multiplexage sur un seul port QUIC** -Un node expose un seul port QUIC (ex. 19010). Tous les groupes hébergés -partagent ce port. Le groupe est identifié dans le handshake MNP par le `group_id` -contenu dans le JWT. Avantages : un seul trou NAT à maintenir, une seule redirection -de port manuelle si nécessaire. Le serveur QUIC route chaque connexion vers -l'IndexGroup/GEK du bon groupe après vérification du JWT. +Before writing Phase 7 production code, two critical design gaps must be +addressed — see `first-review.md` for full analysis: + +1. **[C2] JWT must carry group membership claims.** Add `"groups": [group_ids]` + to `issue_access_token()`. Node MNP handshake must verify the requested + group_id is in the JWT's groups claim before serving any content. Without + this, any authenticated user can access any group on the node. + → Implement in 7.3 (multi-group daemon) before any other milestone. + +2. **[C1] Chat encryption: Sender Keys, not shared Double Ratchet.** The Double + Ratchet is a pairwise (1:1) protocol — sharing a single ratchet state across + N group members causes key/nonce reuse (AEAD catastrophic failure). The + architecture now uses **Sender Keys** (Signal Groups approach): each member + has their own sending chain key, O(N) state. + → Implement in 7.5. The existing `ratchet.py` is kept for future 1:1 DM. + +### Architectural decisions (settled) + +**Multi-group → multiplexing on a single QUIC port** +A node exposes a single QUIC port (e.g. 19010). All hosted groups share this +port. The group is identified in the MNP handshake by the `group_id` in the +JWT. Advantages: one NAT hole to maintain, one manual port forward if needed. +The QUIC server routes each connection to the appropriate IndexGroup/GEK +after JWT verification **and group membership authorization check**. **Signaling punch/connect (via hub WebSocket)** -Actuellement, le node punchs aveuglément au démarrage → 12.7s de handshake -(trou NAT vieillit avant que le client arrive). Solution : +Currently the node punches blindly at startup → 12.7s handshake (NAT hole ages +before the client arrives). Solution: ``` -Client → Hub (HTTPS) : "je vais connecter node X, je viens de IP:PORT" +Client → Hub (HTTPS) : "I'm about to connect to node X, I'm at IP:PORT" Hub → Node (WS) : message "client_incoming: {peer_ip, peer_port}" -Node → NAT (UDP) : punch_nat(peer_ip, peer_port) immédiat +Node → NAT (UDP) : punch_nat(peer_ip, peer_port) immediately Node → Hub (WS) : "punch_ready" -Hub → Client (HTTPS) : "connecte-toi maintenant" -Client → Node (QUIC) : < 2s après le probe → trou frais → < 200ms +Hub → Client (HTTPS) : "connect now" +Client → Node (QUIC) : < 2s after probe → fresh NAT entry → < 200ms ``` -Le canal hub→node WebSocket existe déjà (`hub/api/revocation.py`). -Il suffit d'ajouter le type de message `client_incoming` / `punch_ready`. -Ce mécanisme s'appuie sur l'ICE simplifié (Interactive Connectivity Establishment). +The hub→node WebSocket channel already exists (`hub/api/revocation.py`). +Just add `client_incoming` / `punch_ready` message types. +This mechanism is a simplified ICE (Interactive Connectivity Establishment). -**Chat — entre forum et Signal** -Pas un chat temps-réel éphémère (Signal) ni un forum lourd. -Modèle : **fil de discussion chiffré E2E, persistant sur le node**. -- Messages courts + pièces jointes (comme Signal groupe) -- Fils/topics optionnels pour structurer (comme un forum léger) -- Historique stocké sur le node (pas éphémère) -- Push pour membres connectés, pull pour hors-ligne -- Double Ratchet (déjà implémenté) pour le chiffrement -- Scope : par groupe (pas par paire d'utilisateurs) -- Pas de suppression automatique (l'admin du groupe gère la rétention) +The same WebSocket channel also carries **jti denylist push** (security review +S3): when the hub revokes a token, it pushes the jti to all connected nodes. +Nodes maintain an in-memory set and check it during MNP handshake. + +**Chat — between a forum and Signal** +Not a real-time ephemeral chat (Signal) nor a heavy forum. +Model: **E2E encrypted discussion thread, persisted on the node**. +- Short messages + attachments (like Signal group) +- Optional threads/topics for structure (like a light forum) +- History stored on the node (not ephemeral) +- Push for connected members, pull for offline +- **Sender Keys** protocol for encryption (security review C1 — Double Ratchet + is pairwise only, cannot be shared across group members) +- Scope: per group (not per user pair) +- No automatic deletion (group admin manages retention) ### Milestones -| # | Composant | Fichier(s) | Priorité | +| # | Component | File(s) | Priority | |---|---|---|---| -| 7.1 | QUIC 0-RTT session resumption | `transport/quic_server.py` + `quic_client.py` | Haute | -| 7.2 | Signaling `client_incoming`/`punch_ready` | `hub/api/revocation.py` + `node/hub_client.py` | Haute | -| 7.3 | Daemon multi-groupe (multiplexage 1 port) | `node/daemon.py` — N IndexGroups, 1 QuicChunkServer | Haute | -| 7.4 | HLS streaming via QUIC | `node/transport/hls.py` — segments en QUIC streams | Moyenne | -| 7.5 | Chat : stockage + wire protocol MNP | `node/chat/store.py` + `common/protocol.py` | Moyenne | -| 7.6 | Chat : UI web locale + push WS members | `node/ui/app.py` WebSocket pour notifications | Moyenne | -| 7.7 | Calibration Argon2id CLI | `node/daemon.py` — `meshbay-node calibrate-argon2` | Basse | +| 7.0 | JWT group claims + node authz check | `hub/auth.py` + `node/transport/quic_server.py` | **Blocker** | +| 7.1 | QUIC 0-RTT session resumption | `transport/quic_server.py` + `quic_client.py` | High | +| 7.2 | Signaling `client_incoming`/`punch_ready` + jti denylist push | `hub/api/revocation.py` + `node/hub_client.py` | High | +| 7.3 | Multi-group daemon (1-port multiplexing) | `node/daemon.py` — N IndexGroups, 1 QuicChunkServer | High | +| 7.4 | HLS streaming via QUIC | `node/transport/hls.py` — segments as QUIC streams | Medium | +| 7.5 | Chat: Sender Keys protocol + storage + MNP wire | `common/senderkeys.py` + `node/chat/store.py` | Medium | +| 7.6 | Chat: local web UI + WS push to members | `node/ui/app.py` WebSocket for notifications | Medium | +| 7.7 | Argon2id calibration CLI | `node/daemon.py` — `meshbay-node calibrate-argon2` | Low | -**Questions ouvertes restantes :** -- Les groupes d'un même node partagent-ils la même clé Ed25519 de node ? (probable oui) -- UI multi-groupe localhost:18000 : onglets par groupe ou liste unifiée ? +**Remaining open questions:** +- Do groups on the same node share the node's Ed25519 key? (likely yes) +- Multi-group UI at localhost:18000: tabs per group or unified list? --- -## Phase 8 — Hub v2 : admin, federation, sécurité production +## Phase 8 — Hub v2: admin, federation, production security -**Objectif :** hub prêt pour opération publique — rôles admin, MHP réseau, -CSAM intégré, monitoring. +**Objective:** hub ready for public operation — admin roles, MHP network, +CSAM integrated, monitoring. -| # | Composant | Fichier(s) | Priorité | +| # | Component | File(s) | Priority | |---|---|---|---| -| 8.1 | Rôles admin (hub_admin flag sur User) | `hub/db/models.py` + `hub/api/admin.py` | Haute | -| 8.2 | MHP inter-hub réseau (pas juste en mémoire) | `hub/api/federation.py` + Alembic migration | Haute | -| 8.3 | federated_groups DB persistance | `hub/db/models.py` FederatedGroup already defined | Haute | -| 8.4 | CSAM DB réelle (import NCMEC/IWF) | `hub/csam.py` — import CLI + API update | Haute | -| 8.5 | Signaling endpoint WS (pour punch coordination) | `hub/api/signaling.py` | Haute | -| 8.6 | Métriques / healthcheck | `hub/api/health.py` | Moyenne | -| 8.7 | Cleanup IP logs (purge > 1 an) | `hub/tasks/cleanup.py` — APScheduler | Moyenne | -| 8.8 | Alembic migration Argon2id params | Bump migration + `hub/auth.py` | Basse | +| 8.1 | Admin roles (hub_admin flag on User) + authz on revocation | `hub/db/models.py` + `hub/api/admin.py` + `hub/api/revocation.py` | **High — S1** | +| 8.2 | Email encryption at rest | `hub/db/models.py` — AES-256-GCM with config-derived key | **High — S2** | +| 8.3 | Refresh token rotation (one-time use) | `hub/api/users.py` — rotate on each use, detect reuse | **High — S5** | +| 8.4 | MHP inter-hub network (not just in-memory) | `hub/api/federation.py` + Alembic migration | High | +| 8.5 | federated_groups DB persistence | `hub/db/models.py` FederatedGroup already defined | High | +| 8.6 | CSAM real DB (import NCMEC/IWF) | `hub/csam.py` — import CLI + API update | High | +| 8.7 | Rate limiting on all authenticated endpoints | `hub/api/middleware.py` — extend slowapi | Medium — M3 | +| 8.8 | Metrics / healthcheck | `hub/api/health.py` | Medium | +| 8.9 | Cleanup IP logs (purge > 1 year) | `hub/tasks/cleanup.py` — APScheduler | Medium | +| 8.10 | Alembic migration Argon2id params | Bump migration + `hub/auth.py` | Low | + +Items 8.1-8.3 are from the first security review (S1, S2, S5). -**Questions à clarifier :** -- Qui peut être hub_admin ? Premier user inscrit ? Config toml ? -- MHP : authentification inter-hubs via JWT ou mutual TLS ? -- Signaling : hub WebSocket pour coordonner punch → connect en < 2s +**Questions to clarify:** +- Who can be hub_admin? First registered user? Config toml? +- MHP: inter-hub authentication via JWT or mutual TLS? --- @@ -123,7 +151,7 @@ des bindings JNI. La couche UI peut être Jetpack Compose. | 10.1 | Web client : décryptage privé (AES-GCM + SubtleCrypto) | `static/crypto.js` MeshBayCrypto | Haute | | 10.2 | Web client : groupe-type "browser" (AES-GCM GEK) | Hub : `cipher` field sur Group | Haute | | 10.3 | Player HLS dans browser (hls.js + déchiffrement) | `static/app.js` + hls.js | Haute | -| 10.4 | Chat browser (Double Ratchet JS via WASM ou port) | `static/ratchet.js` | Moyenne | +| 10.4 | Chat browser (Sender Keys JS — AES-GCM via SubtleCrypto) | `static/senderkeys.js` | Moyenne | | 10.5 | PWA / Service Worker | offline + cache | Basse | **Question clé :** pour le streaming privé en browser, deux approches : @@ -173,8 +201,9 @@ Phase 10 (Web v2) ← streaming privé browser Phase 12 (Packaging) ← distribution ``` -**Prochaine décision structurante :** -La Phase 7 nécessite de clarifier 3 points avant de coder : -1. Architecture multi-groupe sur un node (ports partagés ou dédiés ?) -2. Mécanisme de signaling punch/connect (nouveau endpoint WS sur le hub ?) -3. Le chat est-il un module (Phase 7.5) ou une feature core du protocole ? +**Next structural decisions (all resolved — see first-review.md):** +1. Multi-group on a single QUIC port ✅ (decided Phase 7) +2. Signaling punch/connect via existing hub WS ✅ (decided Phase 7) +3. Chat is a core feature, not a module ✅ (decided draft v3) +4. Chat encryption: Sender Keys ✅ (decided in security review) +5. JWT group claims required ✅ (decided in security review) diff --git a/docs/meshbay-draft-v3.md b/docs/meshbay-draft-v3.md index e11200c..97c391d 100644 --- a/docs/meshbay-draft-v3.md +++ b/docs/meshbay-draft-v3.md @@ -97,7 +97,7 @@ Email is kept in full (not hashed) to support: Phone number: optional, associable after account creation. On Android, both collected at registration. Accounts are fusionable (email + phone pointing to same account). -Email and phone are stored encrypted at rest in the database. +Email and phone are stored encrypted at rest in the database, using a server-side key derived from the hub's configuration secret (not the database). **[NOT YET IMPLEMENTED — currently stored in plaintext. Tracked as open question #10.]** #### 4.1.2 Mandatory IP Logging (Legal Compliance) @@ -140,7 +140,8 @@ Two tokens issued at login: **Access token** (JWT, signed Ed25519): - Validity: 1 hour -- Payload: `jti` (UUID4, **mandatory** — unique per token, enables individual revocation and prevents replay), `user_id`, `PK_user`, `PK_user_x25519`, `hub_id`, `issued_at`, `expires_at`, hub-signed groups membership claim +- Payload: `jti` (UUID4, **mandatory** — unique per token, enables individual revocation and prevents replay), `user_id`, `PK_user`, `PK_user_x25519`, `hub_id`, `issued_at`, `expires_at`, `groups` (list of group_ids the user is a member of — hub-signed membership claim) +- The `groups` claim is **mandatory** for node-side authorization: the node checks that the requested group_id appears in the JWT before serving any content. Without this claim, any authenticated user could access any group on the node. - Presented to nodes for authentication and group access verification - Verified locally by nodes using the hub's known public key — no hub roundtrip - Compromise window: 1 hour maximum @@ -237,6 +238,11 @@ in the client JWT. The server routes each connection to the appropriate DirectoryIndexer and GEK after JWT verification. Rationale: one NAT hole to maintain, one port to forward manually if needed. +**Authorization invariant:** the node MUST verify that the JWT's `groups` claim +contains the requested group_id before serving any content. Without this check, +any authenticated user could access any group on the node. This is enforced at +the MNP handshake layer, not the transport layer. + **Platform:** Linux primary, cross-platform from the start (Windows/macOS). Python ensures portability. #### 4.2.1 Keystore and Unlock @@ -512,7 +518,7 @@ Disk (plaintext) → zstd compress → GEK encrypt (per-chunk) → TCP+TLS 1.3 s **Chunking:** - Chunk size: 1 MB (amortizes AEAD overhead; enables seeking) -- Per-chunk key derivation: `chunk_key = HKDF(GEK, salt="file:" || blake3(file) || "chunk:" || index)` +- Per-chunk key derivation: `chunk_key = HKDF(GEK, salt=None, info="file:" || blake3(file) || ":chunk:" || index)` — salt is omitted because the GEK is a CSPRNG output (already uniform); the file/chunk context goes in `info` for domain separation, which is the correct HKDF usage per RFC 5869 - Each chunk independently decryptable → enables VOD seeking - Compress before encrypt (compression is ineffective on ciphertext) @@ -600,11 +606,21 @@ Group chat is a **core feature** (not an extension module). - **Push/pull:** connected members get real-time push (WebSocket); offline members pull history on reconnect - **Retention:** managed by the group admin (no automatic expiry) -**Encryption:** Double Ratchet algorithm (implemented in `meshbay_common.ratchet`): -- Forward secrecy and break-in recovery per message -- Each message independently encrypted, out-of-order delivery handled -- Attachment files: encrypted with GEK-derived key (same as file chunks), hash in message -- Group scope: all members share the same ratchet state seeded from the group GEK +**Encryption — Sender Keys protocol (decided in first security review, 2026-08-10):** + +The Double Ratchet (implemented in `meshbay_common.ratchet`) is a **pairwise** (1:1) protocol. Using a shared ratchet state for N group members would cause chain key desynchronization and nonce/key reuse — a catastrophic AEAD failure. The architecture uses **Sender Keys** instead (same approach as Signal Groups): + +- Each group member generates a **sender key** (random symmetric chain key + signing keypair) +- On joining a group, the new member's sender key is distributed to all existing members via pairwise channels (GEK-wrapped or direct) +- Each existing member sends their current sender key to the new member +- Messages are encrypted with the sender's chain key (symmetric ratchet, one direction) +- Forward secrecy at **member rotation** granularity: when a member is removed, all remaining members rotate their sender keys +- O(N) state per member (one chain per group member), not O(N^2) +- The existing Double Ratchet implementation is kept for future 1:1 direct messaging + +Attachment files: encrypted with GEK-derived key (same as file chunks), hash referenced in the message. + +> **Why not MLS (RFC 9420)?** MLS provides O(log N) message overhead and per-message forward secrecy via tree-based ratcheting. It is the superior long-term choice, but its complexity is not justified for v1 group sizes (< 50 members). Sender Keys is proven at scale (Signal, WhatsApp) and simpler to implement. Migration to MLS is a v2 option if group sizes grow. --- @@ -854,16 +870,24 @@ The node loads extension modules (Python) in a sandboxed subprocess. **Chat is a | R6 | Hub API: which endpoints for GEK distribution? | `GET /v1/users/{username}/pubkeys`, `POST /v1/groups`, `POST /v1/groups/{group_id}/members/{username}/gek`, `GET /v1/groups/{group_id}/gek` | Spike 6 | | R7 | Package structure? | 3 packages: `meshbay-common`, `meshbay-hub`, `meshbay-node`. uv workspace monorepo. | POC | +**Resolved by first security review (2026-08-10):** + +| # | Question | Resolution | Source | +|---|---|---|---| +| R8 | Group chat encryption model? | Sender Keys protocol (Signal Groups approach). Double Ratchet kept for future 1:1 DM only. MLS considered for v2 if groups > 50 members. | Security review C1 | +| R9 | Token denylist distribution? | Push via existing hub→node WebSocket. Node maintains an in-memory jti set. MNP handshake checks the set before accepting a JWT. No periodic polling needed. | Security review S3 | +| R10 | Chunk key HKDF: salt or info? | `info` (domain separation), `salt=None`. GEK is CSPRNG output (already uniform), so HKDF extract step doesn't need a random salt. Spec wording corrected to match code (RFC 5869 compliant). | Security review M5 | +| R11 | AES-GCM keystore IV size? | 96-bit (12 bytes), per NIST SP 800-38D recommendation. Code fixed from 128-bit to 96-bit. | Security review S4 | + **Still open:** 1. **Refresh token validity:** 30 or 90 days? 2. **Group address scheme:** final URL format confirmation -3. **Double Ratchet library:** identify best Python implementation (evaluate `python-doubleratchet`, `axolotl`, or custom) -4. **GEK bundle location for groups with mixed access** (public-restricted): hub or node? -5. **MHP federation sync frequency and conflict resolution** -6. **Hub mirror replication strategy** (when implemented) -7. **Chat attachment storage:** stored on node like regular files, or separate store? -8. **Relay registration protocol design** (when implemented) -9. **Token denylist distribution:** how do nodes fetch and cache the `jti` denylist? Push (hub WebSocket) or pull (periodic poll)? Cache TTL? -10. **QUIC migration timeline:** when is the application protocol considered stable enough to begin v2 transport implementation? -11. **Port configuration conflict:** `18000` used for both local web UI and (in some proposals) MNP listener — needs final port allocation decision (proposed split: 18000 for web UI, 18001 for MNP). +3. **GEK bundle location for groups with mixed access** (public-restricted): hub or node? +4. **MHP federation sync frequency and conflict resolution** +5. **Hub mirror replication strategy** (when implemented) +6. **Chat attachment storage:** stored on node like regular files, or separate store? +7. **Relay registration protocol design** (when implemented) +8. **QUIC migration timeline:** when is the application protocol considered stable enough to begin v2 transport implementation? +9. **Refresh token rotation:** implement one-time-use refresh tokens (rotate on each use, detect reuse as theft indicator). RFC 6819 §5.2.2.3. +10. **Email encryption at rest:** spec requires encrypted email/phone in DB, implementation stores plaintext. Needs server-side encryption with key from hub config. diff --git a/first-review.md b/first-review.md new file mode 100644 index 0000000..dd1847c --- /dev/null +++ b/first-review.md @@ -0,0 +1,372 @@ +# MeshBay — First Architecture & Security Review + +> Date: 2026-08-10 +> Scope: design-level review of the cryptographic architecture, trust model, and +> security properties as specified in `docs/meshbay-draft-v3.md` and implemented +> through Phases 1-6 (81 tests, demo-v2 validated). +> +> This review does NOT assess the security of the demo/test deployment. It evaluates +> whether the architecture, as designed, provides a robust foundation for a secure +> decentralized platform. + +--- + +## Executive Summary + +The cryptographic architecture is **strong and well-designed**. The algorithm +choices are modern and correct, the trust model is sound, and the key hierarchy +is properly separated. The six POC spikes were genuinely useful — the jti fix +(Spike 3), the Argon2id recalibration (Spike 1), and the GEK wrapping protocol +confirmation (Spike 6) are exactly the kind of findings that save projects from +shipping real vulnerabilities. + +There are **no fatal design flaws**. The issues found are fixable before Phase 7, +and the most important one (Double Ratchet group model) should be resolved +before writing production chat code. + +Classification: **Critical** (must fix before production), **Significant** (design +gap, fix before Phase 8), **Minor** (improvement, can schedule), **Note** (observation, +no action required). + +--- + +## What Is Solid + +These design decisions are correct and represent genuine security engineering: + +**1. Hub-blind GEK wrapping (ECIES-like)** +The wrapping protocol (ephemeral X25519 + HKDF + ChaCha20-Poly1305 with AAD) is +textbook ECIES done right. The hub stores opaque blobs, the ephemeral keypair +ensures each wrapping produces different ciphertext, and the AAD binding to +`pk_recipient` prevents bundle swapping attacks. This is the most important +crypto decision in the system and it's correct. + +**2. Ed25519 identity verification independent of TLS** +Nodes use self-signed TLS certs for transport confidentiality only. Client +verifies the node's Ed25519 public key (from hub) at the MNP handshake layer. +This decouples transport security from identity — the right design for a system +where nodes can't get CA-signed certificates. + +**3. Mandatory jti in JWT** +The Spike 3 finding was critical. Ed25519 is deterministic — without jti, two +tokens issued in the same second are bit-identical. Adding UUID4 jti to every +token was the correct fix. The architecture now enables per-token revocation. + +**4. On-the-fly encryption model** +Files stored in plaintext on the node, encrypted at read time with per-chunk keys +derived from the GEK via HKDF. This avoids the double-storage problem +(encrypted + plaintext) and makes GEK rotation feasible without re-encrypting +terabytes on disk. + +**5. Domain separation in HKDF** +Every key derivation uses a distinct `info` string (`meshbay:gek_wrap:v1`, +`meshbay:ratchet:root:v1`, etc.). The AES variant adds `:aes` suffix to chunk +key derivation. This is a small detail that prevents cross-protocol key reuse +and shows mature crypto engineering. + +**6. Transport abstraction layer** +The `Transport` interface allowing TCP+TLS v1 → QUIC v2 migration without +protocol changes was a good architectural decision confirmed by the successful +demo-v2 QUIC validation. + +**7. Refresh token stored as blake3 hash** +Server never stores the raw refresh token — only its hash. Correct pattern. +Database breach doesn't leak usable refresh tokens. + +--- + +## Critical Issues + +### C1 — Double Ratchet is not suitable for group chat as described + +**Location:** `meshbay_common/ratchet.py`, draft v3 section 6.6 + +**Problem:** The draft states "all members share the same ratchet state seeded +from the group GEK." The Signal Double Ratchet is designed for **pairwise** +(1:1) communication. It fundamentally cannot work as a shared group state: + +- If all N members share a single ratchet state, each member advancing the + sending chain desynchronizes all other members. Message 5 from Alice and + message 5 from Bob would use the same chain key, producing a nonce/key reuse + — a catastrophic failure for ChaCha20-Poly1305 and AES-GCM. +- The current `RatchetState` class has one `CKs` (sending chain) and one `CKr` + (receiving chain), confirming it's a pairwise protocol. + +**What Signal actually does for groups:** Signal uses a different protocol called +**Sender Keys** (described in their "Group Protocol" specification). Each member +has their own symmetric sending chain key. When a member joins a group, all +existing members send their current sender key to the new member via pairwise +Double Ratchet channels. This gives forward secrecy per member, not per message. + +**Impact:** If implemented as described, group chat will either: +- Silently corrupt messages (if state is truly shared), or +- Require N*(N-1)/2 pairwise ratchet sessions (O(N^2) state, impractical for + groups > 10 members) + +**Recommendation:** Before Phase 7.5 (chat), decide between: +1. **Sender Keys** (Signal Groups approach): each member maintains one symmetric + sending chain. Forward secrecy at member rotation granularity. O(N) state. + Simpler to implement, good enough for most threat models. +2. **Pairwise Double Ratchet**: keep the current implementation but use it for + 1:1 messages only. Group messages would be encrypted N-1 times. O(N^2) cost + per message — only feasible for small groups (<20). +3. **MLS (Message Layer Security, RFC 9420)**: the modern standard for group + messaging. Tree-based ratcheting, O(log N) state and messages. More complex + but future-proof. Python implementations exist (`openmls` bindings, or + `mls-protocol`). + +Recommendation: **Sender Keys** for v1 (pragmatic, Signal-proven), with the +option to migrate to MLS later if group sizes grow. + +### C2 — No group membership verification in MNP handshake + +**Location:** `quic_server.py:125-146`, `server.py:130-151` + +**Problem:** The MNP handshake verifies the JWT signature and expiration, but does +NOT check whether the authenticated user is a member of the group being accessed. +Any valid JWT holder can request any file from any group served by the node. + +The draft says the JWT carries "hub-signed groups membership claim" (section 4.1.4), +but the actual `issue_access_token()` in `auth.py:104-125` does not include any +group membership claims. The JWT contains only `sub`, `pk_user`, `hub_id`, `jti`, +`iat`, `exp`. + +**Impact in Phase 7 (multi-group):** A user authenticated for group A can request +files from group B on the same node. Since all groups share one QUIC port, this +becomes an authorization bypass. + +**Recommendation:** +- Add group membership claims to the JWT: `"groups": ["group_id_1", "group_id_2"]` +- Node verifies the requested group_id is in the JWT's groups claim +- This is a simple change to `issue_access_token()` + handshake verification +- The JWT is already verified offline with the hub's Ed25519 key — adding claims + doesn't change the verification flow + +--- + +## Significant Issues + +### S1 — Admin revocation endpoint has no authorization check + +**Location:** `revocation.py:149-194` + +**Problem:** The `admin_revoke` endpoint requires authentication (`get_current_user`) +but does NOT verify that the current user is a hub admin. Any authenticated user +can revoke any other user or any group. The docstring says "Admin only (user must +be hub admin — user_id in config)" but no such check is implemented. + +**Impact:** Any registered user can revoke any other user or group on the hub. +This is a privilege escalation vulnerability. + +**Recommendation:** Phase 8 plans admin roles (8.1: `hub_admin` flag on User). This +check must be added before the revocation endpoint is used in any non-demo context. +For now, the endpoint exists but is only callable by someone who knows the API — +acceptable for a test deployment, not for production. + +### S2 — Email stored in plaintext in the database + +**Location:** `models.py:42`, draft v3 section 4.1.1 + +**Problem:** The spec says "Email and phone are stored encrypted at rest in the +database." The actual `User` model stores email as `String(256)` — plaintext. +A database breach would expose all user emails. + +**Recommendation:** Encrypt email (and future phone field) with a server-side key +derived from a secret not stored in the database (e.g., from the hub config file). +Use AES-256-GCM with a deterministic IV derived from user_id (for lookups) or +accept that encrypted email cannot be searched by value. + +### S3 — No jti denylist distribution to nodes + +**Location:** draft v3 section 4.1.4, open question #9 + +**Problem:** The architecture describes a jti denylist for immediate token revocation, +but: +- The hub has no `GET /v1/revoke/denylist` endpoint (marked [TBD]) +- Nodes don't check any denylist during JWT verification +- The revocation WebSocket pushes revocation tokens to nodes, but nodes don't + persist or check them during MNP handshake + +**Impact:** A revoked user's JWT remains valid for up to 1 hour (until natural +expiration). The revocation WebSocket can close active connections, but new +connections with the same JWT will succeed. + +**Recommendation:** Two options: +1. **Push + local cache** (recommended): when the node receives a revocation via + WebSocket, it adds the jti to an in-memory set. MNP handshake checks this set. + Simple, real-time, no polling. +2. **Pull**: node periodically fetches the denylist from the hub. Adds latency + between revocation and enforcement. + +Option 1 is simpler and already half-built (the WebSocket channel exists). + +### S4 — AES-GCM keystore uses non-standard 128-bit IV + +**Location:** `crypto.py:148` — `iv = os.urandom(16)` + +**Problem:** AES-GCM is specified for 96-bit (12-byte) nonces (NIST SP 800-38D). +The keystore encryption uses a 128-bit (16-byte) IV. The `cryptography` library +accepts this and processes it through GHASH to derive the internal counter, which +is secure — but it's a deviation from the standard. + +**Impact:** No direct vulnerability. AES-GCM with >96-bit IVs has a slightly +different security proof (birthday bound applies to the GHASH reduction). For a +keystore that's encrypted once and rarely re-encrypted, the practical risk is zero. + +**Recommendation:** Change to `os.urandom(12)` for standard compliance. Simple +one-line fix. The existing keystore files would need re-encryption on next save +(which happens naturally when the user updates their keystore). + +### S5 — Refresh token not rotated on use + +**Location:** `users.py:158-179` + +**Problem:** When a refresh token is used to obtain a new access token, the same +refresh token remains valid. If an attacker intercepts a refresh token, they can +use it repeatedly alongside the legitimate user, and neither party detects the +theft. + +**Recommendation:** Implement refresh token rotation: each use of a refresh token +issues a new refresh token and invalidates the old one. If the old token is used +again (by the attacker), the hub detects the reuse and revokes all tokens for +that user (indicating theft). This is the OAuth 2.0 Security BCP recommendation +(RFC 6819, section 5.2.2.3). + +--- + +## Minor Issues + +### M1 — Username enumeration via registration and pubkeys endpoints + +The registration endpoint returns "Username already taken" (409), and +`GET /v1/users/{username}/pubkeys` returns 404 vs a valid response. Both allow +enumerating valid usernames. For a decentralized platform where users have public +identities, this may be acceptable by design, but it should be a conscious +decision. + +### M2 — TLS self-signed certificate uses RSA-2048 + +**Location:** `tls_cert.py:36` + +The TLS cert uses RSA-2048 while the rest of the system uses Ed25519. Since the +cert is only for transport confidentiality (identity is verified via Ed25519), +this is acceptable. However, using an Ed25519 TLS certificate would be more +consistent and is supported by modern TLS 1.3 stacks. RSA-2048 is ~112-bit +security; Ed25519 is ~128-bit. + +### M3 — No rate limiting on GEK retrieval and pubkeys endpoints + +Only `/register` and `/login` have rate limiting. An attacker could enumerate +pubkeys or attempt to retrieve GEK bundles at high frequency. While GEK bundles +are opaque (no direct attack), rate limiting on all authenticated endpoints is +good hygiene. + +### M4 — Single admin per group with no delegation or recovery + +If the admin's node goes offline, the group becomes inaccessible: no new members +can be added, no GEK rotation, no moderation. There's no mechanism for admin +delegation or recovery. For a personal file-sharing platform this may be +acceptable, but for any group with more than a few members, this is a +single-point-of-failure. + +### M5 — Chunk key derivation uses HKDF salt=None + +**Location:** `crypto.py:46-51` + +The code uses `salt=None` and puts the file context in `info`. This is actually +correct HKDF usage (salt is for randomizing extraction when IKM might be +non-uniform; GEK is from CSPRNG so salt isn't needed; info is for domain +separation). However, the draft v3 spec describes it as using `salt`, which +creates a spec/code discrepancy. Update the spec to match the code, since the +code is correct. + +### M6 — Argon2id production parameters not yet applied + +**Location:** `crypto.py:131-133`, `auth.py:24-26`, `keyderive.py:33-35` + +All three Argon2id usage sites still use the dev parameters (iterations=3, +memory=64MB, ~78ms). Production target is iterations=4, memory=256MB, ~500ms. +Phase 7.7 plans a calibration CLI command. This must be done before any +real-world deployment. The comments document this correctly. + +--- + +## Notes (No Action Required) + +### N1 — Forward secrecy model is appropriate + +File encryption uses GEK-derived symmetric keys — no forward secrecy at the +application layer. If GEK is compromised, past files are decryptable. This is +documented and accepted: the alternative (per-session file encryption keys) +would break seeking, caching, and multi-source download. The transport layer +(TLS 1.3 / QUIC) provides forward secrecy for data in transit. + +### N2 — Error messages in login are correct + +`login()` returns the same "Invalid credentials" for both user-not-found and +wrong-password. This is the correct behavior to prevent user enumeration through +the login flow (even though registration and pubkeys endpoints allow it — see M1). + +### N3 — Hub legal exposure model is well-positioned + +The hub stores no content, no metadata, no node IPs (beyond ephemeral signaling). +GEK bundles are opaque. The hub's legal exposure is analogous to a domain +registrar or email provider — it knows who registered but not what they share. +LCEN/DSA compliance is addressed through IP logging with 1-year retention. + +### N4 — QUIC NAT probe content is fine + +`punch_nat()` sends `b'MESHBAY:NAT:PUNCH'` as a fixed probe. Some NAT +implementations might filter constant payloads, but in practice this works +(demo-v2 confirmed). The content of the probe packet doesn't matter for NAT +entry creation — only the 5-tuple (src_ip, src_port, dst_ip, dst_port, proto) +matters. + +### N5 — Web/CLI key derivation mismatch is by design + +Strategy A (Argon2id) and Strategy B (PBKDF2-SHA512 in browser) produce +different keys from the same password. The code and docs correctly explain this: +users pick one registration path. The web client uses random keypairs stored +encrypted on the hub, not password-derived keys. This avoids the mismatch +entirely. + +--- + +## Prioritized Action Plan + +| # | Issue | Severity | When to fix | +|---|---|---|---| +| C1 | Double Ratchet group model | Critical | Before Phase 7.5 (chat) | +| C2 | No group membership in JWT/handshake | Critical | Phase 7.3 (multi-group) | +| S1 | Admin revocation has no authz check | Significant | Phase 8.1 (admin roles) | +| S2 | Email stored in plaintext | Significant | Phase 8 | +| S3 | No jti denylist on nodes | Significant | Phase 7.2 (signaling) | +| S4 | AES-GCM 128-bit IV | Significant | Any time (1 line) | +| S5 | Refresh token rotation | Significant | Phase 8 | +| M1 | Username enumeration | Minor | Accept or Phase 8 | +| M2 | RSA-2048 TLS cert | Minor | Phase 7 or later | +| M3 | Rate limiting gaps | Minor | Phase 8.6 | +| M4 | Single admin SPOF | Minor | Phase 8+ | +| M5 | Spec/code HKDF discrepancy | Minor | Update spec | +| M6 | Argon2id prod params | Minor | Phase 7.7 | + +--- + +## Conclusion + +MeshBay's security architecture is built on solid foundations. The cryptographic +primitive choices are modern and correct. The trust model (hub-blind, node-hosted, +E2E encrypted) is well-designed and consistently applied. The POC spikes caught +real issues (jti, Argon2id calibration, NAT behavior) that would have been +difficult to fix post-deployment. + +The two critical issues (C1: group ratchet model, C2: group membership +authorization) are both design decisions that need to be made before Phase 7 +produces production chat and multi-group code. They are not retroactive problems +— they are forward-looking decisions that the architecture leaves room for. + +The significant issues (S1-S5) are implementation gaps that should be addressed +during Phases 7-8, in the natural course of hardening the hub and node. + +Overall assessment: **good foundations, ready for Phase 7** after deciding the +group chat encryption model (C1) and adding group claims to the JWT (C2). diff --git a/packages/meshbay-common/src/meshbay_common/crypto.py b/packages/meshbay-common/src/meshbay_common/crypto.py index 2104469..6066f5f 100644 --- a/packages/meshbay-common/src/meshbay_common/crypto.py +++ b/packages/meshbay-common/src/meshbay_common/crypto.py @@ -145,7 +145,7 @@ def derive_keystore_key(password: str, salt: bytes) -> bytes: def encrypt_keystore(plaintext: bytes, key: bytes) -> tuple[bytes, bytes, bytes]: """Encrypt keystore blob with AES-256-GCM. Returns (iv, ciphertext, tag).""" - iv = os.urandom(16) + iv = os.urandom(12) enc = Cipher(algorithms.AES(key), modes.GCM(iv)).encryptor() ct = enc.update(plaintext) + enc.finalize() return iv, ct, enc.tag diff --git a/packages/meshbay-common/src/meshbay_common/senderkeys.py b/packages/meshbay-common/src/meshbay_common/senderkeys.py new file mode 100644 index 0000000..932e2e6 --- /dev/null +++ b/packages/meshbay-common/src/meshbay_common/senderkeys.py @@ -0,0 +1,287 @@ +""" +MeshBay — Sender Keys protocol for group messaging. + +Signal Groups approach: each member maintains their own sending chain. +Advantages over shared Double Ratchet: + - O(N) state per group (one chain per member) vs O(N^2) pairwise + - Single encrypt per message (not N encryptions) + - No key/nonce reuse — each sender has an independent chain + +Key components: + - Chain key ratchet: HKDF per message, provides forward secrecy + - Message key derivation: separate HKDF from chain key + - Ed25519 signing: each sender signs their ciphertext + - AES-256-GCM encryption: browser-compatible symmetric cipher + +Key distribution: + - On join: admin wraps each sender's SenderKeyDistribution with GEK + - On leave: all remaining members rotate their chain keys +""" + +import os +import struct +from dataclasses import dataclass, field + +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) +from cryptography.hazmat.primitives.ciphers.aead import AESGCM +from cryptography.hazmat.primitives.kdf.hkdf import HKDF +from cryptography.hazmat.primitives import hashes, serialization + + +CHAIN_INFO = b"meshbay:sk:chain:v1" +MSG_KEY_INFO = b"meshbay:sk:msg:v1" +CHAIN_KEY_LEN = 32 +MSG_KEY_LEN = 32 +MAX_SKIP = 256 + + +def _hkdf(ikm: bytes, info: bytes, length: int = 32) -> bytes: + return HKDF( + algorithm=hashes.SHA256(), length=length, salt=None, info=info, + ).derive(ikm) + + +def _ratchet_chain(chain_key: bytes) -> tuple[bytes, bytes]: + """Advance chain key → (new_chain_key, message_key).""" + new_ck = _hkdf(chain_key, CHAIN_INFO, CHAIN_KEY_LEN) + mk = _hkdf(chain_key, MSG_KEY_INFO, MSG_KEY_LEN) + return new_ck, mk + + +# ── Data structures ────────────────────────────────────────────────────────── + +@dataclass +class SenderKeyDistribution: + """Sent to group members when a sender joins or rotates.""" + sender_id: str + chain_key: bytes # 32-byte initial chain key + iteration: int # current message counter + signing_pk: bytes # 32-byte raw Ed25519 public key + + def serialize(self) -> bytes: + sender_bytes = self.sender_id.encode() + return ( + struct.pack(">H", len(sender_bytes)) + + sender_bytes + + self.chain_key + + struct.pack(">I", self.iteration) + + self.signing_pk + ) + + @classmethod + def deserialize(cls, data: bytes) -> "SenderKeyDistribution": + sender_len = struct.unpack(">H", data[:2])[0] + offset = 2 + sender_id = data[offset:offset + sender_len].decode() + offset += sender_len + chain_key = data[offset:offset + 32] + offset += 32 + iteration = struct.unpack(">I", data[offset:offset + 4])[0] + offset += 4 + signing_pk = data[offset:offset + 32] + return cls(sender_id=sender_id, chain_key=chain_key, + iteration=iteration, signing_pk=signing_pk) + + +@dataclass +class SenderKeyState: + """One sender's chain state as seen by any group member.""" + sender_id: str + chain_key: bytes + iteration: int + signing_key: Ed25519PublicKey + _skipped_keys: dict[int, bytes] = field(default_factory=dict) + + @classmethod + def from_distribution(cls, dist: SenderKeyDistribution) -> "SenderKeyState": + pk = Ed25519PublicKey.from_public_bytes(dist.signing_pk) + return cls( + sender_id=dist.sender_id, + chain_key=dist.chain_key, + iteration=dist.iteration, + signing_key=pk, + ) + + def advance_to(self, target: int) -> bytes: + """Advance chain to target iteration, caching skipped keys. Returns message key.""" + if target < self.iteration: + mk = self._skipped_keys.pop(target, None) + if mk is None: + raise ValueError(f"Message key {target} already consumed or too old") + return mk + + skip_count = target - self.iteration + if skip_count > MAX_SKIP: + raise ValueError(f"Too many skipped messages: {skip_count}") + + for i in range(skip_count): + new_ck, mk = _ratchet_chain(self.chain_key) + self._skipped_keys[self.iteration] = mk + self.chain_key = new_ck + self.iteration += 1 + + new_ck, mk = _ratchet_chain(self.chain_key) + self.chain_key = new_ck + self.iteration += 1 + return mk + + +@dataclass +class SenderKeyRecord: + """Sender's own key state (includes signing private key).""" + sender_id: str + chain_key: bytes + iteration: int + signing_sk: Ed25519PrivateKey + + @classmethod + def create(cls, sender_id: str) -> "SenderKeyRecord": + return cls( + sender_id=sender_id, + chain_key=os.urandom(CHAIN_KEY_LEN), + iteration=0, + signing_sk=Ed25519PrivateKey.generate(), + ) + + def distribution(self) -> SenderKeyDistribution: + pk_raw = self.signing_sk.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + return SenderKeyDistribution( + sender_id=self.sender_id, + chain_key=self.chain_key, + iteration=self.iteration, + signing_pk=pk_raw, + ) + + def rotate(self) -> "SenderKeyRecord": + """Create a new record with fresh chain key (call on member removal).""" + return SenderKeyRecord( + sender_id=self.sender_id, + chain_key=os.urandom(CHAIN_KEY_LEN), + iteration=0, + signing_sk=Ed25519PrivateKey.generate(), + ) + + +# ── Group store ────────────────────────────────────────────────────────────── + +class GroupSenderKeyStore: + """All sender key states for one group, held by one member.""" + + def __init__(self, group_id: str): + self.group_id = group_id + self._states: dict[str, SenderKeyState] = {} + + def add_sender(self, dist: SenderKeyDistribution) -> None: + self._states[dist.sender_id] = SenderKeyState.from_distribution(dist) + + def remove_sender(self, sender_id: str) -> None: + self._states.pop(sender_id, None) + + def get_state(self, sender_id: str) -> SenderKeyState | None: + return self._states.get(sender_id) + + @property + def sender_count(self) -> int: + return len(self._states) + + +# ── Encrypt / Decrypt ──────────────────────────────────────────────────────── + +@dataclass +class SenderKeyMessage: + """Wire format for a Sender Keys encrypted message.""" + sender_id: str + iteration: int + ciphertext: bytes + nonce: bytes + signature: bytes + + def serialize(self) -> bytes: + sender_bytes = self.sender_id.encode() + return ( + struct.pack(">H", len(sender_bytes)) + + sender_bytes + + struct.pack(">I", self.iteration) + + struct.pack(">I", len(self.ciphertext)) + + self.ciphertext + + self.nonce + + self.signature + ) + + @classmethod + def deserialize(cls, data: bytes) -> "SenderKeyMessage": + offset = 0 + sender_len = struct.unpack(">H", data[offset:offset + 2])[0] + offset += 2 + sender_id = data[offset:offset + sender_len].decode() + offset += sender_len + iteration = struct.unpack(">I", data[offset:offset + 4])[0] + offset += 4 + ct_len = struct.unpack(">I", data[offset:offset + 4])[0] + offset += 4 + ciphertext = data[offset:offset + ct_len] + offset += ct_len + nonce = data[offset:offset + 12] + offset += 12 + signature = data[offset:offset + 64] + return cls(sender_id=sender_id, iteration=iteration, + ciphertext=ciphertext, nonce=nonce, signature=signature) + + +def encrypt_message( + record: SenderKeyRecord, + plaintext: bytes, + aad: bytes = b"", +) -> tuple[SenderKeyMessage, SenderKeyRecord]: + """ + Encrypt a message with the sender's chain key. + Returns (message, updated_record). + """ + new_ck, mk = _ratchet_chain(record.chain_key) + iteration = record.iteration + + nonce = os.urandom(12) + ct = AESGCM(mk).encrypt(nonce, plaintext, aad or None) + + sig_payload = struct.pack(">I", iteration) + nonce + ct + signature = record.signing_sk.sign(sig_payload) + + msg = SenderKeyMessage( + sender_id=record.sender_id, + iteration=iteration, + ciphertext=ct, + nonce=nonce, + signature=signature, + ) + + updated = SenderKeyRecord( + sender_id=record.sender_id, + chain_key=new_ck, + iteration=iteration + 1, + signing_sk=record.signing_sk, + ) + return msg, updated + + +def decrypt_message( + store: GroupSenderKeyStore, + msg: SenderKeyMessage, + aad: bytes = b"", +) -> bytes: + """ + Decrypt and verify a Sender Keys message. + Advances the sender's chain state in the store. + """ + state = store.get_state(msg.sender_id) + if state is None: + raise ValueError(f"Unknown sender: {msg.sender_id}") + + sig_payload = struct.pack(">I", msg.iteration) + msg.nonce + msg.ciphertext + state.signing_key.verify(msg.signature, sig_payload) + + mk = state.advance_to(msg.iteration) + return AESGCM(mk).decrypt(msg.nonce, msg.ciphertext, aad or None) diff --git a/packages/meshbay-common/tests/test_senderkeys.py b/packages/meshbay-common/tests/test_senderkeys.py new file mode 100644 index 0000000..a1181e1 --- /dev/null +++ b/packages/meshbay-common/tests/test_senderkeys.py @@ -0,0 +1,211 @@ +""" +Tests for the Sender Keys group messaging protocol. + +Covers: key creation, distribution, encrypt/decrypt, multi-member groups, +out-of-order delivery, serialization, and key rotation on member removal. +""" + +import pytest + +from meshbay_common.senderkeys import ( + SenderKeyRecord, + SenderKeyDistribution, + SenderKeyMessage, + GroupSenderKeyStore, + encrypt_message, + decrypt_message, +) + + +def test_basic_encrypt_decrypt(): + """Alice encrypts, Bob decrypts using Alice's distributed sender key.""" + alice_rec = SenderKeyRecord.create("alice") + alice_dist = alice_rec.distribution() + + bob_store = GroupSenderKeyStore("group-1") + bob_store.add_sender(alice_dist) + + msg, alice_rec = encrypt_message(alice_rec, b"hello group") + plaintext = decrypt_message(bob_store, msg) + assert plaintext == b"hello group" + + +def test_multiple_messages_sequential(): + """Multiple messages from the same sender decrypt in order.""" + alice_rec = SenderKeyRecord.create("alice") + store = GroupSenderKeyStore("group-1") + store.add_sender(alice_rec.distribution()) + + for i in range(5): + msg, alice_rec = encrypt_message(alice_rec, f"message {i}".encode()) + pt = decrypt_message(store, msg) + assert pt == f"message {i}".encode() + + +def test_multi_member_group(): + """Three members: Alice sends, Bob and Carol both decrypt.""" + alice_rec = SenderKeyRecord.create("alice") + alice_dist = alice_rec.distribution() + + bob_store = GroupSenderKeyStore("group-1") + bob_store.add_sender(alice_dist) + + carol_store = GroupSenderKeyStore("group-1") + carol_store.add_sender(alice_dist) + + msg, alice_rec = encrypt_message(alice_rec, b"broadcast") + + assert decrypt_message(bob_store, msg) == b"broadcast" + assert decrypt_message(carol_store, msg) == b"broadcast" + + +def test_bidirectional_chat(): + """Alice and Bob both send and receive.""" + alice_rec = SenderKeyRecord.create("alice") + bob_rec = SenderKeyRecord.create("bob") + + alice_store = GroupSenderKeyStore("group-1") + alice_store.add_sender(bob_rec.distribution()) + + bob_store = GroupSenderKeyStore("group-1") + bob_store.add_sender(alice_rec.distribution()) + + msg1, alice_rec = encrypt_message(alice_rec, b"hi bob") + assert decrypt_message(bob_store, msg1) == b"hi bob" + + msg2, bob_rec = encrypt_message(bob_rec, b"hi alice") + assert decrypt_message(alice_store, msg2) == b"hi alice" + + +def test_out_of_order_delivery(): + """Messages delivered out of order are decrypted correctly (up to MAX_SKIP).""" + alice_rec = SenderKeyRecord.create("alice") + store = GroupSenderKeyStore("group-1") + store.add_sender(alice_rec.distribution()) + + msg0, alice_rec = encrypt_message(alice_rec, b"msg 0") + msg1, alice_rec = encrypt_message(alice_rec, b"msg 1") + msg2, alice_rec = encrypt_message(alice_rec, b"msg 2") + + # Deliver out of order: 2, 0, 1 + assert decrypt_message(store, msg2) == b"msg 2" + assert decrypt_message(store, msg0) == b"msg 0" + assert decrypt_message(store, msg1) == b"msg 1" + + +def test_replay_rejected(): + """A message decrypted twice raises an error (replay protection).""" + alice_rec = SenderKeyRecord.create("alice") + store = GroupSenderKeyStore("group-1") + store.add_sender(alice_rec.distribution()) + + msg, alice_rec = encrypt_message(alice_rec, b"once only") + decrypt_message(store, msg) + + with pytest.raises(ValueError, match="already consumed"): + decrypt_message(store, msg) + + +def test_unknown_sender_rejected(): + """Message from an unknown sender raises ValueError.""" + alice_rec = SenderKeyRecord.create("alice") + store = GroupSenderKeyStore("group-1") + + msg, _ = encrypt_message(alice_rec, b"who am i") + with pytest.raises(ValueError, match="Unknown sender"): + decrypt_message(store, msg) + + +def test_non_member_cannot_decrypt(): + """Eve (not in group) cannot decrypt Alice's messages.""" + alice_rec = SenderKeyRecord.create("alice") + eve_store = GroupSenderKeyStore("group-1") + + msg, _ = encrypt_message(alice_rec, b"secret") + with pytest.raises(ValueError, match="Unknown sender"): + decrypt_message(eve_store, msg) + + +def test_key_rotation_on_member_removal(): + """After rotation, old chain keys cannot decrypt new messages.""" + alice_rec = SenderKeyRecord.create("alice") + old_dist = alice_rec.distribution() + + # Eve had Alice's old key + eve_store = GroupSenderKeyStore("group-1") + eve_store.add_sender(old_dist) + + # Alice rotates (member removed from group) + alice_rec = alice_rec.rotate() + new_dist = alice_rec.distribution() + + # Bob gets the new distribution + bob_store = GroupSenderKeyStore("group-1") + bob_store.add_sender(new_dist) + + msg, alice_rec = encrypt_message(alice_rec, b"post-rotation") + assert decrypt_message(bob_store, msg) == b"post-rotation" + + # Eve cannot decrypt with old key + with pytest.raises(Exception): + decrypt_message(eve_store, msg) + + +def test_distribution_serialization(): + """SenderKeyDistribution round-trips through serialize/deserialize.""" + rec = SenderKeyRecord.create("alice") + dist = rec.distribution() + data = dist.serialize() + recovered = SenderKeyDistribution.deserialize(data) + + assert recovered.sender_id == dist.sender_id + assert recovered.chain_key == dist.chain_key + assert recovered.iteration == dist.iteration + assert recovered.signing_pk == dist.signing_pk + + +def test_message_serialization(): + """SenderKeyMessage round-trips through serialize/deserialize.""" + rec = SenderKeyRecord.create("alice") + store = GroupSenderKeyStore("group-1") + store.add_sender(rec.distribution()) + + msg, _ = encrypt_message(rec, b"serialize me") + data = msg.serialize() + recovered = SenderKeyMessage.deserialize(data) + + assert recovered.sender_id == msg.sender_id + assert recovered.iteration == msg.iteration + assert recovered.ciphertext == msg.ciphertext + assert recovered.nonce == msg.nonce + assert recovered.signature == msg.signature + + # Deserialized message still decrypts + pt = decrypt_message(store, recovered) + assert pt == b"serialize me" + + +def test_tampered_ciphertext_rejected(): + """Modifying the ciphertext makes signature verification fail.""" + alice_rec = SenderKeyRecord.create("alice") + store = GroupSenderKeyStore("group-1") + store.add_sender(alice_rec.distribution()) + + msg, _ = encrypt_message(alice_rec, b"authentic") + msg.ciphertext = bytes([b ^ 0xff for b in msg.ciphertext]) + + with pytest.raises(Exception): + decrypt_message(store, msg) + + +def test_store_sender_count(): + """GroupSenderKeyStore tracks sender count correctly.""" + store = GroupSenderKeyStore("group-1") + assert store.sender_count == 0 + + store.add_sender(SenderKeyRecord.create("alice").distribution()) + store.add_sender(SenderKeyRecord.create("bob").distribution()) + assert store.sender_count == 2 + + store.remove_sender("alice") + assert store.sender_count == 1 diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py index c1697a7..bb88283 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py @@ -25,6 +25,7 @@ On receipt: immediately refuse JWT tokens matching the revoked user_id, and close active connections for that user. """ +import asyncio import base64 import json import logging @@ -50,6 +51,7 @@ router = APIRouter(tags=["revocation"]) # ── Connected node registry ─────────────────────────────────────────────────── _connected_nodes: dict[str, WebSocket] = {} # node_id → websocket +_punch_events: dict[str, asyncio.Event] = {} # node_id → signaling event def get_connected_node_count() -> int: @@ -122,12 +124,16 @@ async def node_websocket(ws: WebSocket): log.info("Node WS connected: %s", node_id[:8]) await ws.send_text(json.dumps({"type": "auth_ok", "node_id": node_id})) - # Keep-alive loop — wait for ping or disconnect + # Message loop — handle ping, punch_ready, etc. while True: raw = await ws.receive_text() msg = json.loads(raw) if msg.get("type") == "ping": await ws.send_text(json.dumps({"type": "pong"})) + elif msg.get("type") == "punch_ready": + event = _punch_events.get(node_id) + if event: + event.set() except WebSocketDisconnect: log.info("Node WS disconnected: %s", (node_id or "unknown")[:8]) @@ -140,6 +146,44 @@ async def node_websocket(ws: WebSocket): # ── Admin revocation endpoint ───────────────────────────────────────────────── +class IncomingRequest(BaseModel): + peer_ip: str + peer_port: int + + +@router.post("/v1/nodes/{node_id}/incoming", status_code=200) +async def notify_incoming( + node_id: str, + body: IncomingRequest, + current_user: User = Depends(get_current_user), +): + """ + Signal a node that a client wants to connect (NAT punch coordination). + Hub forwards the request via WebSocket; node punches NAT and replies punch_ready. + """ + ws = _connected_nodes.get(node_id) + if not ws: + raise HTTPException(status_code=404, detail="Node not connected") + + event = asyncio.Event() + _punch_events[node_id] = event + + await ws.send_text(json.dumps({ + "type": "client_incoming", + "peer_ip": body.peer_ip, + "peer_port": body.peer_port, + })) + + try: + await asyncio.wait_for(event.wait(), timeout=5.0) + except asyncio.TimeoutError: + raise HTTPException(status_code=504, detail="Node did not respond in time") + finally: + _punch_events.pop(node_id, None) + + return {"status": "ready", "node_id": node_id} + + class RevokeRequest(BaseModel): target: str # "user" or "group" target_id: str diff --git a/packages/meshbay-hub/src/meshbay_hub/api/users.py b/packages/meshbay-hub/src/meshbay_hub/api/users.py index 5a7a3b4..5a2c7be 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/users.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/users.py @@ -18,7 +18,7 @@ from meshbay_hub.auth import ( ) from meshbay_hub.config import HubConfig from meshbay_hub.db.engine import get_db -from meshbay_hub.db.models import IPLog, RefreshToken, User +from meshbay_hub.db.models import GroupMember, IPLog, RefreshToken, User from meshbay_hub.api.deps import get_current_user router = APIRouter(prefix="/v1/users", tags=["users"]) @@ -136,7 +136,11 @@ async def login( if user.status != "active": raise HTTPException(status_code=403, detail=f"Account {user.status}") - access_token = issue_access_token(user.id, user.pk_ed25519, ttl=_ttl()) + memberships = await db.execute( + select(GroupMember.group_id).where(GroupMember.user_id == user.id)) + group_ids = [gid for (gid,) in memberships.all()] + access_token = issue_access_token( + user.id, user.pk_ed25519, ttl=_ttl(), groups=group_ids) raw_rt, rt_hash = generate_refresh_token() expires_at = datetime.now(timezone.utc) + timedelta(seconds=_refresh_ttl()) @@ -175,7 +179,11 @@ async def token_refresh( if not user or user.status != "active": raise HTTPException(status_code=401, detail="User not found or suspended") - new_token = issue_access_token(user.id, user.pk_ed25519, ttl=_ttl()) + memberships = await db.execute( + select(GroupMember.group_id).where(GroupMember.user_id == user.id)) + group_ids = [gid for (gid,) in memberships.all()] + new_token = issue_access_token( + user.id, user.pk_ed25519, ttl=_ttl(), groups=group_ids) return {"access_token": new_token, "token_type": "bearer", "expires_in": _ttl()} diff --git a/packages/meshbay-hub/src/meshbay_hub/auth.py b/packages/meshbay-hub/src/meshbay_hub/auth.py index a4c3bfb..2b2c61e 100644 --- a/packages/meshbay-hub/src/meshbay_hub/auth.py +++ b/packages/meshbay-hub/src/meshbay_hub/auth.py @@ -105,10 +105,12 @@ def issue_access_token( user_id: str, pk_user: str, ttl: int = 3600, + groups: list[str] | None = None, ) -> str: """ Issue a signed JWT access token. Includes jti (UUID4) — required to prevent replay and enable revocation. + Includes groups — list of group_ids the user is a member of (node-side authz). """ if _hub_sk_pem is None: raise RuntimeError("Hub keypair not loaded") @@ -121,6 +123,7 @@ def issue_access_token( "jti": str(uuid.uuid4()), "iat": now, "exp": now + ttl, + "groups": groups or [], } return jwt.encode(payload, _hub_sk_pem, algorithm="EdDSA") diff --git a/packages/meshbay-hub/tests/test_hub_api.py b/packages/meshbay-hub/tests/test_hub_api.py index ea59f17..568d5d9 100644 --- a/packages/meshbay-hub/tests/test_hub_api.py +++ b/packages/meshbay-hub/tests/test_hub_api.py @@ -261,3 +261,53 @@ async def test_non_admin_cannot_add_member(client): json=bundle, headers={"Authorization": f"Bearer {dan_token}"}) assert r.status_code == 403 + + +@pytest.mark.asyncio +async def test_jwt_contains_groups_claim(client): + """JWT must contain a 'groups' list with group_ids the user is a member of.""" + import jwt as pyjwt + pk_ed_a, pk_x_a, _ = _gen_user_keys() + pk_ed_b, pk_x_b, sk_x_b = _gen_user_keys() + + await client.post("/v1/users/register", json={ + "username": "grp_alice", "email": "ga@x.com", "password": "alicepass99", + "pk_user_ed25519": pk_ed_a, "pk_user_x25519": pk_x_a}) + await client.post("/v1/users/register", json={ + "username": "grp_bob", "email": "gb@x.com", "password": "bobpass99", + "pk_user_ed25519": pk_ed_b, "pk_user_x25519": pk_x_b}) + + # Login before joining any group — groups should be empty + r = await client.post("/v1/users/login", json={ + "username": "grp_bob", "password": "bobpass99"}) + token_pre = r.json()["access_token"] + r_pk = await client.get("/v1/hub/pubkey") + hub_pk = r_pk.json()["pk_hub_pem"].encode() + decoded_pre = pyjwt.decode(token_pre, hub_pk, algorithms=["EdDSA"]) + assert decoded_pre["groups"] == [] + + # Alice creates a group and adds Bob + alice_token = (await client.post("/v1/users/login", + json={"username": "grp_alice", "password": "alicepass99"})).json()["access_token"] + r = await client.post("/v1/groups", json={"name": "testgroup"}, + headers={"Authorization": f"Bearer {alice_token}"}) + group_id = r.json()["group_id"] + + gek = generate_gek() + bundle = wrap_gek(gek, base64.b64decode(pk_x_b)) + await client.post(f"/v1/groups/{group_id}/members/grp_bob/gek", + json=bundle, + headers={"Authorization": f"Bearer {alice_token}"}) + + # Login again — groups should contain the new group + r = await client.post("/v1/users/login", json={ + "username": "grp_bob", "password": "bobpass99"}) + token_post = r.json()["access_token"] + decoded_post = pyjwt.decode(token_post, hub_pk, algorithms=["EdDSA"]) + assert group_id in decoded_post["groups"] + + # Alice (admin) should also have the group in her JWT + r = await client.post("/v1/users/login", json={ + "username": "grp_alice", "password": "alicepass99"}) + decoded_alice = pyjwt.decode(r.json()["access_token"], hub_pk, algorithms=["EdDSA"]) + assert group_id in decoded_alice["groups"] diff --git a/packages/meshbay-node/src/meshbay_node/chat/__init__.py b/packages/meshbay-node/src/meshbay_node/chat/__init__.py new file mode 100644 index 0000000..f647e19 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/chat/__init__.py @@ -0,0 +1,4 @@ +"""MeshBay Node — chat module (Sender Keys encrypted group messaging).""" +from .store import ChatStore + +__all__ = ["ChatStore"] diff --git a/packages/meshbay-node/src/meshbay_node/chat/store.py b/packages/meshbay-node/src/meshbay_node/chat/store.py new file mode 100644 index 0000000..1dbcc2b --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/chat/store.py @@ -0,0 +1,119 @@ +""" +MeshBay Node — SQLite-backed chat message store. + +One database per group. Stores encrypted Sender Keys messages for offline +retrieval and history. Messages are stored as received (ciphertext) — +decryption happens on the client side. +""" + +import logging +import time +from dataclasses import dataclass +from pathlib import Path + +import aiosqlite + +log = logging.getLogger(__name__) + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sender_id TEXT NOT NULL, + iteration INTEGER NOT NULL, + payload BLOB NOT NULL, + timestamp REAL NOT NULL, + thread_id TEXT DEFAULT NULL +); +CREATE INDEX IF NOT EXISTS idx_messages_ts ON messages(timestamp); +CREATE INDEX IF NOT EXISTS idx_messages_thread ON messages(thread_id); +""" + + +@dataclass +class StoredMessage: + id: int + sender_id: str + iteration: int + payload: bytes + timestamp: float + thread_id: str | None + + +class ChatStore: + """Async SQLite chat store for one group.""" + + 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.executescript(_SCHEMA) + await self._db.commit() + + async def close(self) -> None: + if self._db: + await self._db.close() + self._db = None + + async def __aenter__(self): + await self.open() + return self + + async def __aexit__(self, *_): + await self.close() + + async def save_message( + self, + sender_id: str, + iteration: int, + payload: bytes, + thread_id: str | None = None, + ) -> int: + """Store a message. Returns the row id.""" + ts = time.time() + cursor = await self._db.execute( + "INSERT INTO messages (sender_id, iteration, payload, timestamp, thread_id) " + "VALUES (?, ?, ?, ?, ?)", + (sender_id, iteration, payload, ts, thread_id), + ) + await self._db.commit() + return cursor.lastrowid + + async def get_messages( + self, + since: float = 0, + limit: int = 100, + ) -> list[StoredMessage]: + """Get messages after a timestamp, most recent last.""" + cursor = await self._db.execute( + "SELECT id, sender_id, iteration, payload, timestamp, thread_id " + "FROM messages WHERE timestamp > ? ORDER BY timestamp ASC LIMIT ?", + (since, limit), + ) + rows = await cursor.fetchall() + return [ + StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], + payload=r[3], timestamp=r[4], thread_id=r[5]) + for r in rows + ] + + async def get_thread(self, thread_id: str, limit: int = 100) -> list[StoredMessage]: + """Get messages in a thread.""" + cursor = await self._db.execute( + "SELECT id, sender_id, iteration, payload, timestamp, thread_id " + "FROM messages WHERE thread_id = ? ORDER BY timestamp ASC LIMIT ?", + (thread_id, limit), + ) + rows = await cursor.fetchall() + return [ + StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], + payload=r[3], timestamp=r[4], thread_id=r[5]) + for r in rows + ] + + async def message_count(self) -> int: + cursor = await self._db.execute("SELECT COUNT(*) FROM messages") + row = await cursor.fetchone() + return row[0] diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index b420013..c28105f 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -73,11 +73,12 @@ class NodeConfig: @dataclass class GroupConfig: id: str = "" + name: str = "" + shared_dir: str = "" visibility: str = "private" # public|private port: int = 19000 # TCP+TLS MNP port for this group quic_port: int = 19010 # QUIC MNP port http_port: int = 19001 # HTTP file API port - name: str = "" @dataclass @@ -125,6 +126,7 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: cfg.groups.append(GroupConfig( id=g.get("id", ""), name=g.get("name", ""), + shared_dir=g.get("shared_dir", ""), visibility=g.get("visibility", "private"), port=g.get("port", cfg.node.port), quic_port=g.get("quic_port", cfg.node.quic_port), diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index bdac6b8..93ba3c4 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -31,6 +31,7 @@ from meshbay_node.hub_client import HubClient, HubConfig from meshbay_node.indexer import DirectoryIndexer from meshbay_node.keystore import load_or_create_keystore from meshbay_node.transport import ChunkServer +from meshbay_node.transport.quic_server import QuicChunkServer from meshbay_node.ui import create_ui_app log = logging.getLogger(__name__) @@ -69,13 +70,14 @@ class NodeDaemon: "status": "starting", "hub_url": config.hub.url, "username": config.hub.username, - "group_id": config.group.id, - "group_name": config.group.name, + "groups": [g.name for g in config.groups], "node_port": config.node.port, + "quic_port": config.node.quic_port, "endpoint_hint": None, - "index": None, + "indexes": {}, } - self._server: ChunkServer | None = None + self._tcp_server: ChunkServer | None = None + self._quic_server: QuicChunkServer | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] @@ -99,52 +101,79 @@ class NodeDaemon: session = await hub.startup(endpoint_hint=None) self._state["endpoint_hint"] = session.node_id - # 3. Fetch GEK if group configured - if self._config.group.id: - try: - gek = await hub.fetch_gek(self._config.group.id) - keys.gek = gek - log.info("GEK loaded for group %s", self._config.group.id[:8]) - except LookupError: - log.warning("No GEK bundle found for group %s — " - "wait for admin to add you", self._config.group.id[:8]) - - # 4. Directory indexers - async def on_index_change(indexer: DirectoryIndexer) -> None: - self._state["index"] = indexer.index + # 3. 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: + log.warning("Group %r missing id or shared_dir — skipping", + group_cfg.name) + continue - for shared_dir in self._config.node.shared_dirs: - d = Path(shared_dir).expanduser().resolve() - if not d.exists(): - log.warning("Shared directory not found: %s — skipping", d) + shared_root = Path(group_cfg.shared_dir).expanduser().resolve() + if not shared_root.exists(): + log.warning("Shared dir not found: %s — skipping group %s", + shared_root, group_cfg.name) continue + + gek = None + if group_cfg.visibility == "private": + try: + gek = await hub.fetch_gek(group_cfg.id) + 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 + indexer = DirectoryIndexer( - root=d, - group_id=self._config.group.id, + root=shared_root, + group_id=group_cfg.id, sk_node=keys.sk_ed25519, - gek=keys.gek, - on_change=on_index_change, + gek=gek, ) await indexer.start() self._indexers.append(indexer) - self._state["index"] = indexer.index - log.info("Indexing: %s (%d files)", d, indexer.index.count) + self._state["indexes"][group_cfg.id] = indexer.index + log.info("Indexing group %s: %s (%d files)", + group_cfg.name, shared_root, indexer.index.count) + + groups_ctx[group_cfg.id] = { + "gek": gek, + "shared_root": shared_root, + "index": indexer.index, + } + + # 4. QUIC chunk server (primary transport, all groups on one port) + if groups_ctx: + first = next(iter(groups_ctx.values())) + self._quic_server = QuicChunkServer( + sk_node=keys.sk_ed25519, + hub_pk_pem=session.hub_pk_pem, + gek=first["gek"], + shared_root=first["shared_root"], + index=first["index"], + host="::", + port=self._config.node.quic_port, + groups=groups_ctx, + ) + await self._quic_server.start() + log.info("QUIC server on port %d (%d groups)", + self._config.node.quic_port, len(groups_ctx)) - # 5. Chunk server - if self._indexers and keys.gek: - self._server = ChunkServer( + # TCP+TLS server (fallback transport, same groups) + self._tcp_server = ChunkServer( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, - gek=keys.gek, - shared_root=Path(self._config.node.shared_dirs[0]).expanduser(), - index=self._indexers[0].index, + gek=first["gek"], + shared_root=first["shared_root"], + index=first["index"], host="0.0.0.0", port=self._config.node.port, + groups=groups_ctx, ) - await self._server.start() - log.info("Chunk server on port %d", self._config.node.port) + await self._tcp_server.start() + log.info("TCP+TLS server on port %d", self._config.node.port) - # 6. Local web UI + # 5. Local web UI ui_app = create_ui_app(self._state) ui_cfg = uvicorn.Config( ui_app, @@ -157,9 +186,9 @@ class NodeDaemon: log.info("Local UI at http://localhost:%d", self._config.node.ui_port) self._state["status"] = "running" - log.info("Node ready") + log.info("Node ready — %d groups", len(groups_ctx)) - # 7. Wait for shutdown + # 6. Wait for shutdown stop_event = asyncio.Event() loop = asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): @@ -176,8 +205,10 @@ class NodeDaemon: task.cancel() for indexer in self._indexers: await indexer.stop() - if self._server: - await self._server.stop() + if self._quic_server: + await self._quic_server.stop() + if self._tcp_server: + await self._tcp_server.stop() log.info("Node stopped") diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py index d91b945..74851c1 100644 --- a/packages/meshbay-node/src/meshbay_node/hub_client.py +++ b/packages/meshbay-node/src/meshbay_node/hub_client.py @@ -19,6 +19,7 @@ import logging import time from dataclasses import dataclass, field from pathlib import Path +from typing import Any, Callable import httpx import jwt @@ -243,6 +244,59 @@ class HubClient: r.raise_for_status() return r.json() + # ── Persistent WebSocket (signaling + revocations) ────────────────────── + + async def maintain_ws( + self, + on_incoming: Any = None, + on_revocation: Any = None, + ) -> None: + """ + Maintain a persistent WebSocket connection to the hub. + Receives NAT punch requests and revocation tokens. + Runs until cancelled. + """ + import websockets + + if self._session is None: + raise RuntimeError("Not logged in") + + hub_url = self._session.hub_url.replace("https://", "wss://").replace("http://", "ws://") + ws_url = f"{hub_url}/v1/nodes/ws" + + while True: + try: + async with websockets.connect(ws_url) as ws: + await ws.send(json.dumps({ + "type": "auth", + "token": self._session.access_token, + })) + auth_resp = json.loads(await ws.recv()) + if auth_resp.get("type") != "auth_ok": + log.error("WS auth failed: %s", auth_resp) + return + + log.info("Hub WS connected") + + async for raw in ws: + msg = json.loads(raw) + mtype = msg.get("type") + + if mtype == "client_incoming" and on_incoming: + await on_incoming(msg["peer_ip"], msg["peer_port"]) + await ws.send(json.dumps({"type": "punch_ready"})) + + elif mtype == "revocation" and on_revocation: + on_revocation(msg.get("token", "")) + + elif mtype == "pong": + pass + + except Exception as e: + log.warning("Hub WS disconnected: %s — reconnecting in 5s", e) + import asyncio + await asyncio.sleep(5) + # ── Convenience: full startup sequence ─────────────────────────────────── async def startup(self, endpoint_hint: str | None = None) -> HubSession: diff --git a/packages/meshbay-node/src/meshbay_node/transport/__init__.py b/packages/meshbay-node/src/meshbay_node/transport/__init__.py index b3144a6..5a1b8d7 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/__init__.py +++ b/packages/meshbay-node/src/meshbay_node/transport/__init__.py @@ -6,15 +6,16 @@ from .http_server import create_http_app # QUIC transport (MNP v2) — requires aioquic>=1.0 # Falls back gracefully if not installed; node still works via TCP+TLS and HTTP. try: - from .quic_server import QuicChunkServer + from .quic_server import QuicChunkServer, Denylist from .quic_client import QuicChunkClient QUIC_AVAILABLE = True except ImportError: QuicChunkServer = None # type: ignore[assignment,misc] QuicChunkClient = None # type: ignore[assignment,misc] + Denylist = None # type: ignore[assignment,misc] QUIC_AVAILABLE = False __all__ = [ "ChunkServer", "ChunkClient", "create_http_app", - "QuicChunkServer", "QuicChunkClient", "QUIC_AVAILABLE", + "QuicChunkServer", "QuicChunkClient", "Denylist", "QUIC_AVAILABLE", ] diff --git a/packages/meshbay-node/src/meshbay_node/transport/client.py b/packages/meshbay-node/src/meshbay_node/transport/client.py index 3430365..63d50af 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/client.py +++ b/packages/meshbay-node/src/meshbay_node/transport/client.py @@ -58,11 +58,13 @@ class ChunkClient: jwt_token: str, gek: bytes, pk_node_b64: str, # node's Ed25519 PK from hub — used for sig verification + group_id: str = "", ): self._host = host self._port = port self._jwt_token = jwt_token self._gek = gek + self._group_id = group_id self._pk_node = Ed25519PublicKey.from_public_bytes( base64.b64decode(pk_node_b64)) self._reader: asyncio.StreamReader | None = None @@ -80,12 +82,14 @@ class ChunkClient: self._reader, self._writer = await asyncio.open_connection( self._host, self._port, ssl=ssl_ctx) - # MNP handshake - await _send(self._writer, { + handshake_msg = { "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": self._jwt_token, - }) + } + if self._group_id: + handshake_msg["group_id"] = self._group_id + await _send(self._writer, handshake_msg) ack = await _recv(self._reader) if ack.get("type") != MNP.HANDSHAKE_ACK: raise ConnectionError(f"Handshake rejected: {ack}") diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py index a2220ff..83b729e 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py @@ -100,18 +100,22 @@ class QuicChunkClient: jwt_token: str, gek: bytes, pk_node_b64: str, - local_port: int = 0, # 0 = OS choisit; spécifier pour hole punching Port-Restricted + local_port: int = 0, # 0 = OS picks; set for hole punching (Port-Restricted) + group_id: str = "", + session_ticket: object | None = None, ): self._host = host self._port = port self._jwt_token = jwt_token self._gek = gek self._local_port = local_port + self._group_id = group_id self._pk_node = Ed25519PublicKey.from_public_bytes( base64.b64decode(pk_node_b64)) self._proto: _MNPClientProtocol | None = None self._cm = None self._ctrl_stream = 0 + self._session_ticket = session_ticket async def __aenter__(self): await self.connect() @@ -120,6 +124,13 @@ class QuicChunkClient: async def __aexit__(self, *_): await self.close() + @property + def session_ticket(self) -> object | None: + return self._session_ticket + + def _save_ticket(self, ticket: object) -> None: + self._session_ticket = ticket + async def connect(self) -> None: import ssl config = QuicConfiguration( @@ -127,20 +138,25 @@ class QuicChunkClient: alpn_protocols=ALPN, verify_mode=ssl.CERT_NONE, # identity verified via Ed25519 at MNP layer ) + if self._session_ticket: + config.session_ticket = self._session_ticket self._cm = connect( self._host, self._port, configuration=config, create_protocol=_MNPClientProtocol, - local_port=self._local_port, # 0 = aléatoire; local_port=X pour hole punching + local_port=self._local_port, + session_ticket_handler=self._save_ticket, ) self._proto = await self._cm.__aenter__() - # MNP handshake on stream 0 - self._proto._send(self._ctrl_stream, { + handshake_msg = { "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": self._jwt_token, - }) + } + if self._group_id: + handshake_msg["group_id"] = self._group_id + self._proto._send(self._ctrl_stream, handshake_msg) ack = await self._proto._recv(self._ctrl_stream) if ack.get("type") != MNP.HANDSHAKE_ACK: raise ConnectionError(f"QUIC handshake rejected: {ack}") @@ -198,3 +214,22 @@ class QuicChunkClient: raise ValueError("Plaintext hash mismatch after decryption") return plaintext + + async def fetch_stream_segment( + self, file_id: str, segment_index: int, segment_duration: int = 4, + ) -> bytes: + """Fetch one HLS segment (MPEG-TS bytes) over QUIC.""" + sid = self._new_stream() + self._proto._send(sid, { + "type": MNP.STREAM_SEGMENT, + "v": MNP_VERSION, + "file_id": file_id, + "segment_index": segment_index, + "segment_duration": segment_duration, + }) + msg = await self._proto._recv(sid, timeout=30.0) + + if msg.get("type") == "error": + raise LookupError(msg.get("detail", "Unknown error")) + + return base64.b64decode(msg["data_b64"]) diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 9cb3bd8..43c1026 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -21,6 +21,7 @@ import asyncio import base64 import logging import struct +import subprocess from pathlib import Path from typing import Any, Callable @@ -50,6 +51,25 @@ MAX_MSG = 64 * 1024 * 1024 ALPN = ["meshbay-mnp"] +class Denylist: + """Shared denylist for revoked users and invalidated JWTs.""" + + def __init__(self): + self.user_ids: set[str] = set() + self.jtis: set[str] = set() + + def is_denied(self, user_id: str, jti: str) -> bool: + return user_id in self.user_ids or jti in self.jtis + + def deny_user(self, user_id: str) -> None: + self.user_ids.add(user_id) + log.info("Denied user: %s", user_id[:8]) + + def deny_jti(self, jti: str) -> None: + self.jtis.add(jti) + log.info("Denied jti: %s", jti[:8]) + + # ── Wire helpers ────────────────────────────────────────────────────────────── def _pack(obj: dict) -> bytes: @@ -90,6 +110,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): super().__init__(*args, **kwargs) self._ctx = node_ctx # shared server context (keys, index, etc.) self._user_id: str | None = None + self._group_id: str | None = None self._buffers: dict[int, _StreamBuffer] = {} def quic_event_received(self, event: QuicEvent) -> None: @@ -116,6 +137,10 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._do_index_sync_sync(stream_id) elif mtype == MNP.FILE_REQUEST: self._do_file_request_sync(stream_id, msg) + elif mtype == MNP.STREAM_SEGMENT: + self._do_stream_segment_sync(stream_id, msg) + elif mtype == MNP.CHAT_MESSAGE: + self._do_chat_message_sync(stream_id, msg) else: log.warning("Unknown MNP message type: %s", mtype) except Exception as e: @@ -123,8 +148,8 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._send(stream_id, {"type": "error", "detail": str(e)}) def _do_handshake_sync(self, stream_id: int, msg: dict) -> None: - import time token = msg.get("token", "") + group_id = msg.get("group_id", "") try: decoded = jwt.decode(token, self._ctx["hub_pk_pem"], algorithms=["EdDSA"]) except Exception as e: @@ -132,21 +157,45 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._quic.close() return - if decoded.get("exp", 0) < int(time.time()): - self._send(stream_id, {"type": "error", "detail": "JWT expired"}) + denylist = self._ctx.get("denylist") + if denylist and denylist.is_denied(decoded.get("sub", ""), decoded.get("jti", "")): + self._send(stream_id, {"type": "error", "detail": "Token revoked"}) + self._quic.close() + return + + if group_id and group_id not in decoded.get("groups", []): + self._send(stream_id, {"type": "error", "detail": "Not a member of this group"}) + self._quic.close() + return + + if group_id and "groups" in self._ctx and group_id not in self._ctx["groups"]: + self._send(stream_id, {"type": "error", "detail": "Group not hosted on this node"}) self._quic.close() return self._user_id = decoded["sub"] - log.info("QUIC handshake OK — user=%s", self._user_id[:8]) + self._group_id = group_id + + peers = self._ctx.get("_peers") + if peers is not None: + peers[self._user_id] = self + + log.info("QUIC handshake OK — user=%s group=%s", self._user_id[:8], group_id[:8] if group_id else "none") self._send(stream_id, { "type": MNP.HANDSHAKE_ACK, "v": MNP_VERSION, "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()), }) + def _group_ctx(self) -> dict: + """Resolve the active group context (multi-group or legacy single-group).""" + if "groups" in self._ctx and self._group_id: + return self._ctx["groups"][self._group_id] + return self._ctx + def _do_index_sync_sync(self, stream_id: int) -> None: - wire = self._ctx["index"].serialize() + ctx = self._group_ctx() + wire = ctx["index"].serialize() self._send(stream_id, { "type": MNP.INDEX_SYNC, "v": MNP_VERSION, @@ -155,26 +204,96 @@ class _MNPServerProtocol(QuicConnectionProtocol): def _do_file_request_sync(self, stream_id: int, msg: dict) -> None: """Serve file chunk synchronously (blocking I/O — acceptable for test sizes).""" + ctx = self._group_ctx() file_id = msg["file_id"] chunk_index = msg["chunk_index"] - entry = self._ctx["index"].get_entry(file_id) + entry = ctx["index"].get_entry(file_id) if not entry: self._send(stream_id, {"type": "error", "detail": "File not found"}) return - file_path = self._ctx["shared_root"] / entry.path / entry.name + file_path = ctx["shared_root"] / entry.path / entry.name if not file_path.exists(): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return chunk_data = _read_and_encrypt( self._ctx["sk_node"], - self._ctx["gek"], + ctx["gek"], file_path, chunk_index, ) self._send(stream_id, chunk_data) + def _do_stream_segment_sync(self, stream_id: int, msg: dict) -> None: + """Extract and serve one HLS segment via ffmpeg.""" + ctx = self._group_ctx() + file_id = msg["file_id"] + segment_index = msg["segment_index"] + segment_duration = msg.get("segment_duration", 4) + + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send(stream_id, {"type": "error", "detail": "File not found"}) + return + + file_path = ctx["shared_root"] / entry.path / entry.name + if not file_path.exists(): + self._send(stream_id, {"type": "error", "detail": "File not on disk"}) + return + + start_time = segment_index * segment_duration + segment_data = _extract_segment(file_path, start_time, segment_duration) + if segment_data is None: + self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) + return + + self._send(stream_id, { + "type": MNP.STREAM_SEGMENT, + "v": MNP_VERSION, + "file_id": file_id, + "segment_index": segment_index, + "data_b64": base64.b64encode(segment_data).decode(), + "size": len(segment_data), + }) + + def _do_chat_message_sync(self, stream_id: int, msg: dict) -> None: + """Receive a chat message, store it, and broadcast to other connected peers.""" + chat_store = self._ctx.get("chat_store") + if chat_store: + import asyncio + asyncio.ensure_future(chat_store.save_message( + sender_id=msg.get("sender_id", self._user_id), + iteration=msg.get("iteration", 0), + payload=msg.get("payload", b"").encode() if isinstance(msg.get("payload"), str) else msg.get("payload", b""), + thread_id=msg.get("thread_id"), + )) + + peers = self._ctx.get("_peers", {}) + broadcast = { + "type": MNP.CHAT_MESSAGE, + "v": MNP_VERSION, + "sender_id": msg.get("sender_id", self._user_id), + "iteration": msg.get("iteration", 0), + "payload": msg.get("payload", ""), + "thread_id": msg.get("thread_id"), + "group_id": self._group_id or "", + } + for uid, proto in peers.items(): + if uid != self._user_id and proto is not self: + try: + proto._send(0, broadcast) + except Exception: + pass + + self._send(stream_id, {"type": "ack", "v": MNP_VERSION}) + + def connection_lost(self, exc) -> None: + peers = self._ctx.get("_peers") + if peers and self._user_id: + peers.pop(self._user_id, None) + super().connection_lost(exc) + def _send(self, stream_id: int, obj: dict) -> None: self._quic.send_stream_data(stream_id, _pack(obj)) self.transmit() @@ -213,6 +332,25 @@ def _read_and_encrypt( } +def _extract_segment(file_path: Path, start_time: float, duration: float) -> bytes | None: + """Extract one HLS segment via ffmpeg. Returns MPEG-TS bytes or None on failure.""" + try: + result = subprocess.run( + ["ffmpeg", "-hide_banner", "-loglevel", "error", + "-ss", str(start_time), + "-i", str(file_path), + "-t", str(duration), + "-c:v", "copy", "-c:a", "copy", + "-f", "mpegts", "pipe:1"], + capture_output=True, timeout=30, + ) + if result.returncode == 0 and result.stdout: + return result.stdout + return None + except Exception: + return None + + # ── QuicChunkServer ──────────────────────────────────────────────────────────── class QuicChunkServer: @@ -228,10 +366,12 @@ class QuicChunkServer: gek: bytes, shared_root: Path, index: GroupIndex, - host: str = "::", # écoute IPv4 + IPv6 (dual-stack Linux) + host: str = "::", # listen IPv4 + IPv6 (dual-stack Linux) port: int = 19000, cert_path: Path | None = None, key_path: Path | None = None, + groups: dict[str, dict] | None = None, + denylist: Denylist | None = None, ): self._ctx = { "sk_node": sk_node, @@ -240,17 +380,27 @@ class QuicChunkServer: "shared_root": shared_root, "index": index, } + if groups: + self._ctx["groups"] = groups + self._denylist = denylist or Denylist() + self._ctx["denylist"] = self._denylist + self._ctx["_peers"] = {} self._host = host self._port = port self._cert_path = cert_path or Path.home() / ".config/meshbay/node_tls.crt" self._key_path = key_path or Path.home() / ".config/meshbay/node_tls.key" self._server = None self._task = None + self._session_tickets: dict[bytes, Any] = {} @property def port(self) -> int: return self._port + @property + def denylist(self) -> Denylist: + return self._denylist + def _make_config(self) -> QuicConfiguration: from meshbay_node.transport.tls_cert import generate_self_signed_cert if not self._cert_path.exists(): @@ -259,6 +409,12 @@ class QuicChunkServer: config.load_cert_chain(str(self._cert_path), str(self._key_path)) return config + def _store_ticket(self, ticket: Any) -> None: + self._session_tickets[ticket.ticket] = ticket + + def _fetch_ticket(self, label: bytes) -> Any | None: + return self._session_tickets.pop(label, None) + async def start(self) -> None: config = self._make_config() ctx = self._ctx @@ -270,6 +426,8 @@ class QuicChunkServer: self._host, self._port, configuration=config, create_protocol=protocol_factory, + session_ticket_handler=self._store_ticket, + session_ticket_fetcher=self._fetch_ticket, ) log.info("QuicChunkServer listening on %s:%d (QUIC/UDP)", self._host, self._port) diff --git a/packages/meshbay-node/src/meshbay_node/transport/server.py b/packages/meshbay-node/src/meshbay_node/transport/server.py index 6a1b05b..76ac13a 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/server.py @@ -104,6 +104,7 @@ class _ConnectionHandler: gek: bytes, shared_root: Path, index: GroupIndex, + groups: dict[str, dict] | None = None, ): self._reader = reader self._writer = writer @@ -112,8 +113,10 @@ class _ConnectionHandler: self._gek = gek self._shared_root = shared_root self._index = index + self._groups = groups self._peer = writer.get_extra_info("peername") self._user_id: str | None = None + self._group_id: str | None = None async def handle(self) -> None: try: @@ -133,16 +136,28 @@ class _ConnectionHandler: raise ValueError(f"Expected handshake, got {msg.get('type')!r}") token = msg.get("token", "") + group_id = msg.get("group_id", "") try: decoded = jwt.decode(token, self._hub_pk_pem, algorithms=["EdDSA"]) except Exception as e: raise PermissionError(f"Invalid JWT: {e}") from e - if decoded.get("exp", 0) < int(time.time()): - raise PermissionError("JWT expired") + if group_id and group_id not in decoded.get("groups", []): + raise PermissionError("Not a member of this group") + + if group_id and self._groups and group_id not in self._groups: + raise PermissionError("Group not hosted on this node") self._user_id = decoded["sub"] - log.info("[%s] Handshake OK — user=%s", self._peer, self._user_id[:8]) + self._group_id = group_id + + if group_id and self._groups and group_id in self._groups: + ctx = self._groups[group_id] + self._gek = ctx["gek"] + self._shared_root = ctx["shared_root"] + self._index = ctx["index"] + + log.info("[%s] Handshake OK — user=%s group=%s", self._peer, self._user_id[:8], group_id[:8] if group_id else "none") await _send(self._writer, { "type": MNP.HANDSHAKE_ACK, @@ -221,6 +236,7 @@ class ChunkServer: port: int = 19000, cert_path: Path | None = None, key_path: Path | None = None, + groups: dict[str, dict] | None = None, ): self._sk_node = sk_node self._hub_pk_pem = hub_pk_pem @@ -231,6 +247,7 @@ class ChunkServer: self._port = port self._cert_path = cert_path self._key_path = key_path + self._groups = groups self._server: asyncio.Server | None = None @property @@ -264,5 +281,6 @@ class ChunkServer: reader, writer, self._sk_node, self._hub_pk_pem, self._gek, self._shared_root, self._index, + groups=self._groups, ) await handler.handle() diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index f8978d1..a63e28f 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -10,15 +10,18 @@ Minimal FastAPI app providing: Served only on 127.0.0.1 — not exposed to the network. """ +import asyncio +import json import logging from typing import TYPE_CHECKING -from fastapi import FastAPI +from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from meshbay_node import __version__ if TYPE_CHECKING: + from meshbay_node.chat.store import ChatStore from meshbay_node.indexer import GroupIndex log = logging.getLogger(__name__) @@ -127,7 +130,95 @@ def create_ui_app(state: dict) -> FastAPI: {files_html} <hr> <small>MeshBay Node v{__version__} — <a href="/api/status">JSON status</a> - — <a href="/api/files">JSON files</a></small> + — <a href="/api/files">JSON files</a> — <a href="/chat">Chat</a></small> +</body> +</html>""" + + # ── Chat endpoints ─────────────────────────────────────────────────────── + + _chat_subscribers: list[WebSocket] = [] + + @app.get("/api/chat/history") + async def chat_history(since: float = 0, limit: int = 100): + chat_store = state.get("chat_store") + if not chat_store: + return {"messages": []} + msgs = await chat_store.get_messages(since=since, limit=limit) + return { + "messages": [ + { + "id": m.id, + "sender_id": m.sender_id, + "iteration": m.iteration, + "timestamp": m.timestamp, + "thread_id": m.thread_id, + } + for m in msgs + ] + } + + @app.websocket("/ws/chat") + async def chat_websocket(ws: WebSocket): + """WebSocket for real-time chat push to the local UI.""" + await ws.accept() + _chat_subscribers.append(ws) + try: + while True: + await ws.receive_text() + except WebSocketDisconnect: + pass + finally: + _chat_subscribers.remove(ws) + + async def broadcast_chat_to_ui(msg: dict) -> None: + """Push a chat message to all connected UI WebSocket clients.""" + payload = json.dumps(msg) + dead = [] + for ws in _chat_subscribers: + try: + await ws.send_text(payload) + except Exception: + dead.append(ws) + for ws in dead: + _chat_subscribers.remove(ws) + + app.broadcast_chat = broadcast_chat_to_ui + + @app.get("/chat", response_class=HTMLResponse) + async def chat_page(): + return f"""<!DOCTYPE html> +<html lang="en"> +<head> + <meta charset="utf-8"> + <title>MeshBay Chat</title> + <style> + body {{ font-family: monospace; max-width: 700px; margin: 40px auto; padding: 0 20px; }} + #messages {{ border: 1px solid #ccc; padding: 10px; height: 400px; overflow-y: auto; + background: #fafafa; margin-bottom: 10px; }} + .msg {{ margin: 4px 0; }} + .sender {{ font-weight: bold; color: #2563eb; }} + .time {{ color: #9ca3af; font-size: 0.8em; }} + </style> +</head> +<body> + <h1>MeshBay Chat</h1> + <div id="messages"></div> + <p><a href="/">Back to status</a></p> + <script> + const box = document.getElementById('messages'); + const ws = new WebSocket('ws://' + location.host + '/ws/chat'); + ws.onmessage = (e) => {{ + const msg = JSON.parse(e.data); + const div = document.createElement('div'); + div.className = 'msg'; + const t = new Date(msg.timestamp * 1000).toLocaleTimeString(); + div.innerHTML = '<span class="time">' + t + '</span> ' + + '<span class="sender">' + msg.sender_id + '</span>: ' + + '(encrypted message #' + msg.iteration + ')'; + box.appendChild(div); + box.scrollTop = box.scrollHeight; + }}; + </script> </body> </html>""" diff --git a/packages/meshbay-node/tests/test_chat_store.py b/packages/meshbay-node/tests/test_chat_store.py new file mode 100644 index 0000000..d74310d --- /dev/null +++ b/packages/meshbay-node/tests/test_chat_store.py @@ -0,0 +1,91 @@ +""" +Tests for the SQLite-backed chat message store. +""" + +import pytest +import pytest_asyncio +from pathlib import Path + +from meshbay_node.chat.store import ChatStore + + +@pytest_asyncio.fixture +async def store(tmp_path): + s = ChatStore(db_path=tmp_path / "test_chat.db") + await s.open() + yield s + await s.close() + + +@pytest.mark.asyncio +async def test_save_and_retrieve(store): + row_id = await store.save_message( + sender_id="alice", iteration=0, payload=b"hello", + ) + assert row_id == 1 + + msgs = await store.get_messages() + assert len(msgs) == 1 + assert msgs[0].sender_id == "alice" + assert msgs[0].iteration == 0 + assert msgs[0].payload == b"hello" + assert msgs[0].thread_id is None + + +@pytest.mark.asyncio +async def test_message_count(store): + assert await store.message_count() == 0 + await store.save_message("alice", 0, b"msg1") + await store.save_message("bob", 1, b"msg2") + assert await store.message_count() == 2 + + +@pytest.mark.asyncio +async def test_get_messages_since(store): + await store.save_message("alice", 0, b"old") + all_msgs = await store.get_messages() + cutoff = all_msgs[0].timestamp + await store.save_message("bob", 1, b"new") + + msgs = await store.get_messages(since=cutoff) + assert len(msgs) == 1 + assert msgs[0].sender_id == "bob" + + +@pytest.mark.asyncio +async def test_thread_messages(store): + await store.save_message("alice", 0, b"root", thread_id="t1") + await store.save_message("bob", 1, b"reply", thread_id="t1") + await store.save_message("carol", 2, b"other") + + thread = await store.get_thread("t1") + assert len(thread) == 2 + assert thread[0].sender_id == "alice" + assert thread[1].sender_id == "bob" + + +@pytest.mark.asyncio +async def test_message_ordering(store): + for i in range(5): + await store.save_message(f"user-{i}", i, f"msg-{i}".encode()) + + msgs = await store.get_messages() + assert len(msgs) == 5 + for i, m in enumerate(msgs): + assert m.sender_id == f"user-{i}" + + +@pytest.mark.asyncio +async def test_limit(store): + for i in range(10): + await store.save_message("alice", i, f"msg-{i}".encode()) + + msgs = await store.get_messages(limit=3) + assert len(msgs) == 3 + + +@pytest.mark.asyncio +async def test_context_manager(tmp_path): + async with ChatStore(db_path=tmp_path / "ctx_test.db") as store: + await store.save_message("alice", 0, b"test") + assert await store.message_count() == 1 diff --git a/packages/meshbay-node/tests/test_multi_group.py b/packages/meshbay-node/tests/test_multi_group.py new file mode 100644 index 0000000..9be8d47 --- /dev/null +++ b/packages/meshbay-node/tests/test_multi_group.py @@ -0,0 +1,169 @@ +""" +Multi-group isolation test: two groups on one QUIC server. + +Verifies that: + - A user in group-a can fetch files from group-a + - A user in group-a is rejected when requesting group-b + - A user in both groups can access both +""" + +import os +import time +import jwt +import pytest +from pathlib import Path +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives import serialization + +from meshbay_common.crypto import generate_gek, pk_to_b64 +from meshbay_node.indexer import DirectoryIndexer, GroupIndex +from meshbay_node.transport.quic_server import QuicChunkServer +from meshbay_node.transport.quic_client import QuicChunkClient + + +@pytest.fixture +def sk_node(): + return Ed25519PrivateKey.generate() + +@pytest.fixture +def sk_hub(): + return Ed25519PrivateKey.generate() + +@pytest.fixture +def gek_a(): + return generate_gek() + +@pytest.fixture +def gek_b(): + return generate_gek() + +@pytest.fixture +def dir_a(tmp_path): + d = tmp_path / "group_a" + d.mkdir() + (d / "file_a.txt").write_bytes(b"content from group A " * 100) + return d + +@pytest.fixture +def dir_b(tmp_path): + d = tmp_path / "group_b" + d.mkdir() + (d / "file_b.txt").write_bytes(b"content from group B " * 100) + return d + + +def make_jwt(sk_hub, pk_node_b64, groups, user_id="user-001", ttl=3600): + sk_pem = sk_hub.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + now = int(time.time()) + return jwt.encode({ + "iss": "test-hub", "sub": user_id, + "pk_user": pk_node_b64, "hub_id": "test-hub", + "jti": "test-jti", "iat": now, "exp": now + ttl, + "groups": groups, + }, sk_pem, algorithm="EdDSA") + + +@pytest.fixture +async def multi_group_server(sk_node, sk_hub, gek_a, gek_b, dir_a, dir_b, tmp_path): + hub_pk_pem = sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + indexer_a = DirectoryIndexer(root=dir_a, group_id="group-a", sk_node=sk_node, gek=gek_a) + await indexer_a.initial_scan() + + indexer_b = DirectoryIndexer(root=dir_b, group_id="group-b", sk_node=sk_node, gek=gek_b) + await indexer_b.initial_scan() + + groups = { + "group-a": {"gek": gek_a, "shared_root": dir_a, "index": indexer_a.index}, + "group-b": {"gek": gek_b, "shared_root": dir_b, "index": indexer_b.index}, + } + + cert_path = tmp_path / "node.crt" + key_path = tmp_path / "node.key" + + server = QuicChunkServer( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, + gek=gek_a, shared_root=dir_a, index=indexer_a.index, + host="127.0.0.1", port=19200, + cert_path=cert_path, key_path=key_path, + groups=groups, + ) + await server.start() + yield server, indexer_a, indexer_b + await server.stop() + + +@pytest.mark.asyncio +async def test_user_can_access_own_group( + multi_group_server, sk_node, sk_hub, gek_a, +): + """User in group-a can fetch index and chunks from group-a.""" + server, indexer_a, _ = multi_group_server + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a"]) + + async with QuicChunkClient( + host="127.0.0.1", port=19200, + jwt_token=token, gek=gek_a, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-a", + ) as client: + wire = await client.fetch_index() + recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek_a) + assert recovered.count == 1 + + entry = recovered.entries[0] + assert entry.name == "file_a.txt" + chunk = await client.fetch_chunk(entry.id, chunk_index=0) + assert chunk == b"content from group A " * 100 + + +@pytest.mark.asyncio +async def test_user_rejected_from_other_group( + multi_group_server, sk_node, sk_hub, gek_b, +): + """User in group-a only is rejected when requesting group-b.""" + server, _, _ = multi_group_server + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a"]) + + with pytest.raises(ConnectionError, match="rejected"): + async with QuicChunkClient( + host="127.0.0.1", port=19200, + jwt_token=token, gek=gek_b, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-b", + ) as client: + await client.fetch_index() + + +@pytest.mark.asyncio +async def test_dual_group_user_accesses_both( + multi_group_server, sk_node, sk_hub, gek_a, gek_b, +): + """User in both groups can access either group's files.""" + server, indexer_a, indexer_b = multi_group_server + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a", "group-b"]) + + async with QuicChunkClient( + host="127.0.0.1", port=19200, + jwt_token=token, gek=gek_a, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-a", + ) as client_a: + wire_a = await client_a.fetch_index() + idx_a = GroupIndex.deserialize(wire_a, sk_node=sk_node, gek=gek_a) + assert idx_a.entries[0].name == "file_a.txt" + + async with QuicChunkClient( + host="127.0.0.1", port=19200, + jwt_token=token, gek=gek_b, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-b", + ) as client_b: + wire_b = await client_b.fetch_index() + idx_b = GroupIndex.deserialize(wire_b, sk_node=sk_node, gek=gek_b) + assert idx_b.entries[0].name == "file_b.txt" diff --git a/packages/meshbay-node/tests/test_quic_transport.py b/packages/meshbay-node/tests/test_quic_transport.py index 2abd465..0c1a1cd 100644 --- a/packages/meshbay-node/tests/test_quic_transport.py +++ b/packages/meshbay-node/tests/test_quic_transport.py @@ -14,7 +14,7 @@ from cryptography.hazmat.primitives import serialization from meshbay_common.crypto import generate_gek, pk_to_b64 from meshbay_node.indexer import DirectoryIndexer, GroupIndex -from meshbay_node.transport.quic_server import QuicChunkServer +from meshbay_node.transport.quic_server import QuicChunkServer, Denylist from meshbay_node.transport.quic_client import QuicChunkClient @@ -38,7 +38,7 @@ def shared_dir(tmp_path): (d / "small.txt").write_bytes(b"hello quic " * 100) return d -def make_jwt(sk_hub, pk_node_b64, ttl=3600): +def make_jwt(sk_hub, pk_node_b64, ttl=3600, groups=None): sk_pem = sk_hub.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, @@ -49,6 +49,7 @@ def make_jwt(sk_hub, pk_node_b64, ttl=3600): "iss": "test-hub", "sub": "user-001", "pk_user": pk_node_b64, "hub_id": "test-hub", "jti": "test-jti", "iat": now, "exp": now + ttl, + "groups": groups or [], }, sk_pem, algorithm="EdDSA") @@ -155,3 +156,132 @@ async def test_quic_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_p await client.fetch_index() await server.stop() + + +@pytest.mark.asyncio +async def test_quic_wrong_group_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path): + """QUIC server rejects a client whose JWT groups don't include the requested group_id.""" + hub_pk_pem = sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + cert_path = tmp_path / "node.crt" + key_path = tmp_path / "node.key" + + server = QuicChunkServer( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + host="127.0.0.1", port=19103, + cert_path=cert_path, key_path=key_path, + ) + await server.start() + + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a"]) + + with pytest.raises(ConnectionError, match="rejected"): + async with QuicChunkClient( + host="127.0.0.1", port=19103, + jwt_token=token, gek=gek, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-b", + ) as client: + await client.fetch_index() + + await server.stop() + + +@pytest.mark.asyncio +async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_path): + """QUIC 0-RTT: connect, save session ticket, reconnect with ticket.""" + hub_pk_pem = sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + cert_path = tmp_path / "node.crt" + key_path = tmp_path / "node.key" + + server = QuicChunkServer( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + host="127.0.0.1", port=19104, + cert_path=cert_path, key_path=key_path, + ) + await server.start() + + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key())) + pk_b64 = pk_to_b64(sk_node.public_key()) + + # First connection — captures session ticket + saved_ticket = None + async with QuicChunkClient( + host="127.0.0.1", port=19104, + jwt_token=token, gek=gek, pk_node_b64=pk_b64, + ) as client: + wire = await client.fetch_index() + assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 + saved_ticket = client.session_ticket + + # Allow server to process the close + await asyncio.sleep(0.1) + + # Second connection — reuses session ticket (0-RTT) + async with QuicChunkClient( + host="127.0.0.1", port=19104, + jwt_token=token, gek=gek, pk_node_b64=pk_b64, + session_ticket=saved_ticket, + ) as client: + wire = await client.fetch_index() + assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 + + await server.stop() + + +@pytest.mark.asyncio +async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_path): + """QUIC server rejects a connection when the user is on the denylist.""" + hub_pk_pem = sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + cert_path = tmp_path / "node.crt" + key_path = tmp_path / "node.key" + + denylist = Denylist() + server = QuicChunkServer( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + host="127.0.0.1", port=19105, + cert_path=cert_path, key_path=key_path, + denylist=denylist, + ) + await server.start() + + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key())) + pk_b64 = pk_to_b64(sk_node.public_key()) + + # Connection works before denylisting + async with QuicChunkClient( + host="127.0.0.1", port=19105, + jwt_token=token, gek=gek, pk_node_b64=pk_b64, + ) as client: + wire = await client.fetch_index() + assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 + + # Add user to denylist + denylist.deny_user("user-001") + + # Connection now rejected + with pytest.raises(Exception): + async with QuicChunkClient( + host="127.0.0.1", port=19105, + jwt_token=token, gek=gek, pk_node_b64=pk_b64, + ) as client: + await client.fetch_index() + + await server.stop() diff --git a/packages/meshbay-node/tests/test_transport.py b/packages/meshbay-node/tests/test_transport.py index 2064ba5..0e70d72 100644 --- a/packages/meshbay-node/tests/test_transport.py +++ b/packages/meshbay-node/tests/test_transport.py @@ -41,7 +41,7 @@ def shared_dir(tmp_path): (d / "small.txt").write_bytes(b"hello meshbay " * 100) return d -def make_jwt(sk_hub, pk_node_b64, user_id="user-001", ttl=3600): +def make_jwt(sk_hub, pk_node_b64, user_id="user-001", ttl=3600, groups=None): sk_pem = sk_hub.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, @@ -53,6 +53,7 @@ def make_jwt(sk_hub, pk_node_b64, user_id="user-001", ttl=3600): "pk_user": pk_node_b64, "hub_id": "test-hub", "jti": "test-jti", "iat": now, "exp": now + ttl, + "groups": groups or [], }, sk_pem, algorithm="EdDSA") @@ -154,6 +155,42 @@ async def test_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path): @pytest.mark.asyncio +async def test_wrong_group_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path): + """TCP+TLS server rejects a client whose JWT groups don't include the requested group_id.""" + hub_pk_pem = sk_hub.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + indexer = DirectoryIndexer(root=shared_dir, group_id="g", + sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + cert_path = tmp_path / "node.crt" + key_path = tmp_path / "node.key" + + server = ChunkServer( + sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, + shared_root=shared_dir, index=indexer.index, + host="127.0.0.1", port=0, + cert_path=cert_path, key_path=key_path, + ) + await server.start() + port = server._server.sockets[0].getsockname()[1] + + token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()), groups=["group-a"]) + + with pytest.raises(ConnectionError, match="rejected"): + async with ChunkClient( + host="127.0.0.1", port=port, + jwt_token=token, gek=gek, + pk_node_b64=pk_to_b64(sk_node.public_key()), + group_id="group-b", + ) as client: + pass + + await server.stop() + + +@pytest.mark.asyncio async def test_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path): hub_pk_pem = sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) |