diff options
| author | Christophe Besson <cbesson@gmail.com> | 2026-08-09 05:31:05 +0200 |
|---|---|---|
| committer | Christophe Besson <cbesson@gmail.com> | 2026-08-09 05:31:05 +0200 |
| commit | 3b2dd318477eb268e6821fb000aeadfe60d85987 (patch) | |
| tree | 0c46c3ea663490bbee847a0eec5a41c41ad6561b /packages/meshbay-hub/src/meshbay_hub/api | |
| parent | 6d64da7816aec2cc58bb1a6f0aceb9c9a921129f (diff) | |
| download | meshbay-3b2dd318477eb268e6821fb000aeadfe60d85987.tar.gz | |
feat: Phase 6 complete — chat, multi-group, federation, replication, webcrypto
6.1 Double Ratchet (meshbay_common/ratchet.py):
Forward secrecy, break-in recovery, out-of-order delivery.
Signal-spec KDF_RK/KDF_CK via HKDF-SHA256. 11/11 tests.
6.2 Multi-group node (config.py):
[[groups]] TOML array, per-group ports, back-compat [group].
6.3 MHP federation persistence (db/models.py FederatedGroup + SwarmSource):
receive_directory() now persists to federated_groups table.
list_public_groups() includes federated results with source attribution.
6.4 Content replication (node/replication.py + hub SwarmSource):
ContentReplicator: fetch-index, download, hash-verify, register-swarm.
Hub: POST /v1/swarm/register, GET /v1/swarm/{hash} for multi-source.
6.5 Browser private group (webcrypto.py + static/crypto.js):
AES-256-GCM variant of GEK for WebCrypto-compatible groups.
crypto.js: SubtleCrypto importGEK + deriveChunkKey + decryptChunk.
Keys distinct from ChaCha20 via :aes HKDF info suffix. 4/4 tests.
74/74 tests total.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-hub/src/meshbay_hub/api')
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/api/federation.py | 28 | ||||
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/api/groups.py | 97 |
2 files changed, 106 insertions, 19 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/federation.py b/packages/meshbay-hub/src/meshbay_hub/api/federation.py index 5a7f45f..88bb645 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/federation.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/federation.py @@ -36,6 +36,7 @@ from meshbay_hub.api.deps import get_current_user from meshbay_hub.auth import _hub_id, _hub_sk_pem, hub_public_key_pem from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import Group, User +from sqlalchemy.ext.asyncio import AsyncSession log = logging.getLogger(__name__) @@ -141,19 +142,38 @@ class DirectoryPayload(BaseModel): async def receive_directory( body: DirectoryPayload, authorization: str = Header(...), + db: AsyncSession = Depends(get_db), ): """ Receive a Mesh Directory update from a peer hub. - The directory is stored in memory (production: DB table federated_groups). + Persists groups to federated_groups table for cross-hub search. """ try: _verify_mhp_token(authorization.removeprefix("Bearer ")) except Exception as e: raise HTTPException(status_code=401, detail=str(e)) - log.info("Received %d groups from hub %s", len(body.groups), body.hub_id[:16]) - # TODO Phase 5+: persist to federated_groups table, make searchable - return {"accepted": len(body.groups), "from_hub": body.hub_id} + from meshbay_hub.db.models import FederatedGroup + from datetime import datetime, timezone + now = datetime.now(timezone.utc) + count = 0 + for g in body.groups: + existing = await db.get(FederatedGroup, g["id"]) + if existing: + existing.name = g.get("name", existing.name) + existing.join_policy = g.get("join_policy", existing.join_policy) + existing.updated_at = now + else: + db.add(FederatedGroup( + id=g["id"], + name=g.get("name", ""), + source_hub=body.hub_id, + join_policy=g.get("join_policy", "invite"), + )) + count += 1 + await db.commit() + log.info("Persisted %d groups from hub %s", count, body.hub_id[:16]) + return {"accepted": count, "from_hub": body.hub_id} # ── Revocation propagation ──────────────────────────────────────────────────── diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py index 0092fbd..2ae2c93 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py @@ -7,7 +7,10 @@ from sqlalchemy.ext.asyncio import AsyncSession from meshbay_hub.api.deps import get_current_user from meshbay_hub.db.engine import get_db -from meshbay_hub.db.models import GEKBundle, Group, GroupMember, IPLog, User +from meshbay_hub.db.models import ( + FederatedGroup, GEKBundle, Group, GroupMember, + IPLog, SwarmSource, User, +) router = APIRouter(prefix="/v1/groups", tags=["groups"]) @@ -17,27 +20,91 @@ async def list_public_groups( db: AsyncSession = Depends(get_db), limit: int = 50, offset: int = 0, + include_federated: bool = True, ): - """List public groups — browsable without authentication.""" + """List public groups — local and optionally federated. No auth required.""" result = await db.execute( select(Group) .where(Group.visibility == "public", Group.status == "active") .order_by(Group.created_at.desc()) - .limit(limit) - .offset(offset) + .limit(limit).offset(offset) ) - groups = result.scalars().all() + local = result.scalars().all() + groups = [ + { + "id": g.id, "name": g.name, "join_policy": g.join_policy, + "created_at": g.created_at.isoformat(), "source": "local", + } + for g in local + ] + + if include_federated: + fed_result = await db.execute( + select(FederatedGroup) + .order_by(FederatedGroup.updated_at.desc()) + .limit(limit) + ) + for fg in fed_result.scalars().all(): + groups.append({ + "id": fg.id, "name": fg.name, "join_policy": fg.join_policy, + "updated_at": fg.updated_at.isoformat(), "source": fg.source_hub, + }) + + return {"groups": groups, "total": len(groups)} + + +# ── Swarm (content replication) ─────────────────────────────────────────────── + +class SwarmRegisterRequest(BaseModel): + content_hash: str # blake3 hex + endpoint: str # "ip:port" + + +@router.post("/v1/swarm/register", status_code=201) +async def swarm_register( + body: SwarmRegisterRequest, + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + """Node registers itself as a source for a content hash (public swarm).""" + from datetime import datetime, timezone + node_result = await db.execute( + select(User).where(User.id == current_user.id)) + # Use current_user.id as node_id for simplicity + existing = await db.get(SwarmSource, (body.content_hash, current_user.id)) + now = datetime.now(timezone.utc) + if existing: + existing.endpoint = body.endpoint + existing.last_seen = now + else: + db.add(SwarmSource( + content_hash=body.content_hash, + node_id=current_user.id, + endpoint=body.endpoint, + )) + await db.commit() + return {"status": "registered", "hash": body.content_hash} + + +@router.get("/v1/swarm/{content_hash}") +async def swarm_sources( + content_hash: str, + db: AsyncSession = Depends(get_db), +): + """Return list of nodes that can serve a content hash.""" + from datetime import datetime, timezone, timedelta + cutoff = datetime.now(timezone.utc) - timedelta(minutes=30) + result = await db.execute( + select(SwarmSource) + .where( + SwarmSource.content_hash == content_hash, + SwarmSource.last_seen > cutoff, + ) + ) + sources = result.scalars().all() return { - "groups": [ - { - "id": g.id, - "name": g.name, - "join_policy": g.join_policy, - "created_at": g.created_at.isoformat(), - } - for g in groups - ], - "total": len(groups), + "hash": content_hash, + "sources": [{"node_id": s.node_id, "endpoint": s.endpoint} for s in sources], } |