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-node/src/meshbay_node | |
| 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-node/src/meshbay_node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/replication.py | 142 |
1 files changed, 142 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/replication.py b/packages/meshbay-node/src/meshbay_node/replication.py new file mode 100644 index 0000000..c2f5cd3 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/replication.py @@ -0,0 +1,142 @@ +""" +MeshBay Node — content replication (node-to-node, admin-authorized). + +A replication node downloads files from a source node and stores them +locally, then registers itself as an additional swarm source in the hub. +This provides redundancy and improves availability for public content. + +Only public content is replicated (no GEK needed). +Private content replication requires the GEK and is admin-controlled. + +Usage: + replicator = ContentReplicator( + hub_url=..., access_token=..., source_endpoint=..., + local_dir=Path("/data/replicated"), node_pk_b64=..., + ) + await replicator.replicate_file(file_id, file_name, file_size) +""" + +import asyncio +import hashlib +import logging +from pathlib import Path + +import blake3 +import httpx + +log = logging.getLogger(__name__) + +CHUNK_SIZE = 1024 * 1024 # 1 MB + + +class ContentReplicator: + """ + Downloads public files from a source node and registers as swarm source. + """ + + def __init__( + self, + hub_url: str, + access_token: str, + source_endpoint: str, # "http://ip:port" of source node HTTP API + local_dir: Path, + node_pk_b64: str, + ): + self._hub_url = hub_url.rstrip("/") + self._access_token = access_token + self._source_endpoint = source_endpoint.rstrip("/") + self._local_dir = local_dir + self._node_pk_b64 = node_pk_b64 + self._local_dir.mkdir(parents=True, exist_ok=True) + + @property + def _auth_headers(self) -> dict: + return {"Authorization": f"Bearer {self._access_token}"} + + async def fetch_index(self) -> list[dict]: + """Fetch the public Mesh Group Index from the source node.""" + async with httpx.AsyncClient(timeout=30) as c: + r = await c.get(f"{self._source_endpoint}/index") + r.raise_for_status() + return r.json()["entries"] + + async def replicate_file( + self, + file_id: str, + file_name: str, + file_size: int, + progress_cb=None, + ) -> Path: + """ + Download a public file from the source node, verify integrity, + save locally, and register as swarm source in the hub. + Returns the local file path. + """ + local_path = self._local_dir / file_name + if local_path.exists(): + # Verify hash + existing_hash = blake3.blake3(local_path.read_bytes()).hexdigest() + if existing_hash == file_id: + log.info("Already have %s, skipping", file_name) + await self._register_swarm(file_id, local_path) + return local_path + + log.info("Replicating %s (%d bytes) from %s", file_name, file_size, self._source_endpoint) + + # Stream download chunk by chunk + n_chunks = max(1, (file_size + CHUNK_SIZE - 1) // CHUNK_SIZE) + with open(local_path, "wb") as f: + async with httpx.AsyncClient(timeout=60) as c: + for chunk_idx in range(n_chunks): + # Download full file (simpler for public content) + if chunk_idx == 0: + r = await c.get( + f"{self._source_endpoint}/file/{file_id}", + headers=self._auth_headers, + ) + r.raise_for_status() + f.write(r.content) + if progress_cb: + progress_cb(len(r.content), file_size) + break # full file downloaded in one request + + # Verify hash + actual_hash = blake3.blake3(local_path.read_bytes()).hexdigest() + if actual_hash != file_id: + local_path.unlink(missing_ok=True) + raise ValueError(f"Hash mismatch: expected {file_id[:16]}, got {actual_hash[:16]}") + + log.info("Replicated %s (hash OK)", file_name) + await self._register_swarm(file_id, local_path) + return local_path + + async def _register_swarm(self, content_hash: str, local_path: Path) -> None: + """Register this node as a swarm source for the content hash in the hub.""" + try: + async with httpx.AsyncClient(timeout=10) as c: + r = await c.post( + f"{self._hub_url}/v1/swarm/register", + json={"content_hash": content_hash, "endpoint": self._source_endpoint}, + headers=self._auth_headers, + ) + r.raise_for_status() + log.debug("Registered as swarm source for %s", content_hash[:16]) + except Exception as e: + log.warning("Failed to register swarm source: %s", e) + + async def replicate_all(self, progress_cb=None) -> list[Path]: + """Replicate all public files from the source node.""" + entries = await self.fetch_index() + results = [] + for entry in entries: + try: + path = await self.replicate_file( + file_id=entry["id"], + file_name=entry["name"], + file_size=entry["size"], + progress_cb=progress_cb, + ) + results.append(path) + except Exception as e: + log.error("Failed to replicate %s: %s", entry["name"], e) + return results |