diff options
Diffstat (limited to 'packages')
22 files changed, 1620 insertions, 73 deletions
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) |