aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node
diff options
context:
space:
mode:
Diffstat (limited to 'packages/meshbay-node')
-rw-r--r--packages/meshbay-node/src/meshbay_node/hub_client.py256
-rw-r--r--packages/meshbay-node/tests/test_hub_client.py207
2 files changed, 463 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py
new file mode 100644
index 0000000..d91b945
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/hub_client.py
@@ -0,0 +1,256 @@
+"""
+MeshBay Node — Hub client.
+
+Handles all communication from the node to a Mesh Hub:
+ - User registration (first run)
+ - Login → JWT (access token + refresh token)
+ - JWT offline verification and auto-refresh
+ - Node announcement (endpoint_hint)
+ - GEK bundle retrieval for a group
+ - User public key lookup (for GEK wrapping)
+
+JWT verification is done locally using the hub's cached Ed25519 public key.
+The hub is only contacted for login and refresh — not for every request.
+"""
+
+import base64
+import json
+import logging
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import httpx
+import jwt
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+from cryptography.hazmat.primitives import serialization
+
+from meshbay_common.crypto import pk_to_b64, unwrap_gek
+from meshbay_node.keystore import NodeKeys
+
+log = logging.getLogger(__name__)
+
+TOKEN_REFRESH_MARGIN = 300 # refresh access token 5 min before expiry
+
+
+@dataclass
+class HubSession:
+ hub_url: str
+ username: str
+ user_id: str
+ access_token: str
+ refresh_token: str
+ hub_pk_pem: bytes # cached hub Ed25519 public key
+ node_id: str = ""
+ _token_exp: int = 0
+
+ @property
+ def auth_headers(self) -> dict:
+ return {"Authorization": f"Bearer {self.access_token}"}
+
+ @property
+ def token_expires_in(self) -> int:
+ return max(0, self._token_exp - int(time.time()))
+
+ @property
+ def token_needs_refresh(self) -> bool:
+ return self.token_expires_in < TOKEN_REFRESH_MARGIN
+
+
+@dataclass
+class HubConfig:
+ hub_url: str
+ username: str
+ password: str
+ cache_dir: Path = field(default_factory=lambda: Path.home() / ".config" / "meshbay")
+
+ @property
+ def hub_pk_cache_path(self) -> Path:
+ safe = self.hub_url.replace("://", "_").replace("/", "_").replace(":", "_")
+ return self.cache_dir / f"hub_pk_{safe}.pem"
+
+
+# ── Hub client ────────────────────────────────────────────────────────────────
+
+class HubClient:
+ """Async hub client. Use as async context manager or call close() explicitly."""
+
+ def __init__(self, config: HubConfig, keys: NodeKeys):
+ self._config = config
+ self._keys = keys
+ self._http = httpx.AsyncClient(timeout=15, base_url=config.hub_url)
+ self._session: HubSession | None = None
+
+ async def __aenter__(self):
+ return self
+
+ async def __aexit__(self, *_):
+ await self.close()
+
+ async def close(self):
+ await self._http.aclose()
+
+ # ── Hub public key ────────────────────────────────────────────────────────
+
+ async def _fetch_hub_pk(self) -> bytes:
+ """Fetch and cache hub Ed25519 public key PEM."""
+ cache = self._config.hub_pk_cache_path
+ if cache.exists():
+ log.debug("Hub PK loaded from cache: %s", cache)
+ return cache.read_bytes()
+
+ r = await self._http.get("/v1/hub/pubkey")
+ r.raise_for_status()
+ pem = r.json()["pk_hub_pem"].encode()
+
+ cache.parent.mkdir(parents=True, exist_ok=True)
+ cache.write_bytes(pem)
+ log.info("Hub PK fetched and cached: %s", cache)
+ return pem
+
+ # ── Registration ──────────────────────────────────────────────────────────
+
+ async def register(self) -> str:
+ """Register this node's user on the hub. Returns user_id. Idempotent (409 ok)."""
+ r = await self._http.post("/v1/users/register", json={
+ "username": self._config.username,
+ "password": self._config.password,
+ "pk_user_ed25519": self._keys.pk_ed25519_b64,
+ "pk_user_x25519": self._keys.pk_x25519_b64,
+ })
+ if r.status_code == 201:
+ log.info("Registered user '%s' on hub", self._config.username)
+ return r.json()["user_id"]
+ if r.status_code == 409:
+ log.debug("User '%s' already registered", self._config.username)
+ return ""
+ r.raise_for_status()
+ return ""
+
+ # ── Login ─────────────────────────────────────────────────────────────────
+
+ async def login(self) -> HubSession:
+ """Login, verify JWT offline, return HubSession."""
+ hub_pk_pem = await self._fetch_hub_pk()
+
+ r = await self._http.post("/v1/users/login", json={
+ "username": self._config.username,
+ "password": self._config.password,
+ })
+ r.raise_for_status()
+ data = r.json()
+
+ access_token = data["access_token"]
+ refresh_token = data["refresh_token"]
+
+ # Verify offline — if this passes, the hub's identity is confirmed
+ decoded = jwt.decode(access_token, hub_pk_pem, algorithms=["EdDSA"])
+ assert decoded["pk_user"] == self._keys.pk_ed25519_b64, \
+ "Hub returned token for wrong public key"
+ assert "jti" in decoded, "Hub token missing jti — hub is outdated"
+
+ self._session = HubSession(
+ hub_url=self._config.hub_url,
+ username=self._config.username,
+ user_id=decoded["sub"],
+ access_token=access_token,
+ refresh_token=refresh_token,
+ hub_pk_pem=hub_pk_pem,
+ _token_exp=decoded["exp"],
+ )
+ log.info("Logged in as '%s' (exp in %ds)", self._config.username,
+ self._session.token_expires_in)
+ return self._session
+
+ async def refresh_token(self) -> None:
+ """Refresh the access token using the refresh token."""
+ if self._session is None:
+ raise RuntimeError("Not logged in")
+
+ r = await self._http.post("/v1/users/token/refresh", json={
+ "refresh_token": self._session.refresh_token,
+ })
+ r.raise_for_status()
+ new_token = r.json()["access_token"]
+
+ decoded = jwt.decode(new_token, self._session.hub_pk_pem, algorithms=["EdDSA"])
+ self._session.access_token = new_token
+ self._session._token_exp = decoded["exp"]
+ log.debug("Access token refreshed (exp in %ds)", self._session.token_expires_in)
+
+ async def ensure_fresh_token(self) -> None:
+ """Auto-refresh token if close to expiry."""
+ if self._session and self._session.token_needs_refresh:
+ await self.refresh_token()
+
+ # ── Node announcement ─────────────────────────────────────────────────────
+
+ async def announce_node(self, endpoint_hint: str | None = None) -> str:
+ """Announce this node to the hub. Returns node_id."""
+ if self._session is None:
+ raise RuntimeError("Not logged in")
+ await self.ensure_fresh_token()
+
+ r = await self._http.post("/v1/nodes/announce", json={
+ "pk_node": self._keys.pk_ed25519_b64,
+ "endpoint_hint": endpoint_hint,
+ }, headers=self._session.auth_headers)
+ r.raise_for_status()
+ node_id = r.json()["node_id"]
+ self._session.node_id = node_id
+ log.info("Node announced: %s (hint=%s)", node_id[:8], endpoint_hint)
+ return node_id
+
+ # ── GEK retrieval ─────────────────────────────────────────────────────────
+
+ async def fetch_gek(self, group_id: str) -> bytes:
+ """
+ Fetch and unwrap the GEK bundle for a group.
+ Returns the raw GEK bytes.
+ """
+ if self._session is None:
+ raise RuntimeError("Not logged in")
+ await self.ensure_fresh_token()
+
+ r = await self._http.get(f"/v1/groups/{group_id}/gek",
+ headers=self._session.auth_headers)
+ if r.status_code == 404:
+ raise LookupError(f"No GEK bundle found for group {group_id!r}")
+ r.raise_for_status()
+
+ bundle = r.json()
+ sk_x_raw = self._keys.sk_x25519.private_bytes(
+ serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
+ serialization.NoEncryption())
+ pk_x_raw = base64.b64decode(self._keys.pk_x25519_b64)
+
+ gek = unwrap_gek(bundle, sk_x_raw, pk_x_raw)
+ log.info("GEK unwrapped for group %s", group_id[:8])
+ return gek
+
+ # ── User pubkey lookup ────────────────────────────────────────────────────
+
+ async def get_user_pubkeys(self, username: str) -> dict:
+ """Return {'pk_ed25519': str, 'pk_x25519': str} for a user."""
+ if self._session is None:
+ raise RuntimeError("Not logged in")
+ await self.ensure_fresh_token()
+
+ r = await self._http.get(f"/v1/users/{username}/pubkeys",
+ headers=self._session.auth_headers)
+ if r.status_code == 404:
+ raise LookupError(f"User not found: {username!r}")
+ r.raise_for_status()
+ return r.json()
+
+ # ── Convenience: full startup sequence ───────────────────────────────────
+
+ async def startup(self, endpoint_hint: str | None = None) -> HubSession:
+ """
+ Full startup sequence: register (idempotent) → login → announce node.
+ Returns an active HubSession.
+ """
+ await self.register()
+ session = await self.login()
+ await self.announce_node(endpoint_hint)
+ return session
diff --git a/packages/meshbay-node/tests/test_hub_client.py b/packages/meshbay-node/tests/test_hub_client.py
new file mode 100644
index 0000000..fe8a2af
--- /dev/null
+++ b/packages/meshbay-node/tests/test_hub_client.py
@@ -0,0 +1,207 @@
+"""
+Tests for meshbay_node.hub_client — uses httpx.MockTransport to avoid network.
+"""
+
+import base64
+import json
+import os
+import time
+import pytest
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import httpx
+import jwt
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+from cryptography.hazmat.primitives import serialization
+
+from meshbay_common.crypto import generate_gek, pk_to_b64, wrap_gek
+from meshbay_node.hub_client import HubClient, HubConfig, HubSession
+from meshbay_node.keystore import NodeKeys
+
+
+# ── Test fixtures ──────────────────────────────────────────────────────────────
+
+@pytest.fixture
+def hub_keys():
+ """Fake hub Ed25519 keypair for signing test JWTs."""
+ sk = Ed25519PrivateKey.generate()
+ sk_pem = sk.private_bytes(
+ serialization.Encoding.PEM,
+ serialization.PrivateFormat.PKCS8,
+ serialization.NoEncryption(),
+ )
+ pk_pem = sk.public_key().public_bytes(
+ serialization.Encoding.PEM,
+ serialization.PublicFormat.SubjectPublicKeyInfo,
+ )
+ return sk, sk_pem, pk_pem
+
+
+@pytest.fixture
+def node_keys():
+ sk_ed = Ed25519PrivateKey.generate()
+ sk_x = X25519PrivateKey.generate()
+ return NodeKeys(sk_ed25519=sk_ed, sk_x25519=sk_x)
+
+
+@pytest.fixture
+def hub_config(tmp_path):
+ return HubConfig(
+ hub_url="http://fake-hub",
+ username="testuser",
+ password="testpass99",
+ cache_dir=tmp_path,
+ )
+
+
+def make_token(sk_pem, user_id, pk_user_b64, hub_id="fake-hub", ttl=3600):
+ now = int(time.time())
+ return jwt.encode({
+ "iss": hub_id, "sub": user_id, "pk_user": pk_user_b64,
+ "hub_id": hub_id, "jti": "test-jti",
+ "iat": now, "exp": now + ttl,
+ }, sk_pem, algorithm="EdDSA")
+
+
+# ── Tests ─────────────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_login_verifies_jwt_offline(hub_keys, node_keys, hub_config):
+ sk_hub, sk_hub_pem, pk_hub_pem = hub_keys
+ user_id = "user-uuid-001"
+ token = make_token(sk_hub_pem, user_id, node_keys.pk_ed25519_b64)
+
+ def handler(request):
+ if request.url.path == "/v1/hub/pubkey":
+ return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()})
+ if request.url.path == "/v1/users/login":
+ return httpx.Response(200, json={
+ "access_token": token, "refresh_token": "rt-abc", "expires_in": 3600})
+ return httpx.Response(404)
+
+ transport = httpx.MockTransport(handler)
+ client = HubClient(hub_config, node_keys)
+ client._http = httpx.AsyncClient(transport=transport, base_url="http://fake-hub")
+
+ session = await client.login()
+ assert session.user_id == user_id
+ assert session.access_token == token
+ assert session.hub_pk_pem == pk_hub_pem
+ assert session.token_expires_in > 3500
+ assert not session.token_needs_refresh
+
+
+@pytest.mark.asyncio
+async def test_login_rejects_missing_jti(hub_keys, node_keys, hub_config):
+ sk_hub, sk_hub_pem, pk_hub_pem = hub_keys
+ # Token without jti
+ bad_token = jwt.encode({
+ "iss": "fake-hub", "sub": "uid", "pk_user": node_keys.pk_ed25519_b64,
+ "hub_id": "fake-hub", "iat": int(time.time()), "exp": int(time.time()) + 3600,
+ }, sk_hub_pem, algorithm="EdDSA")
+
+ def handler(request):
+ if request.url.path == "/v1/hub/pubkey":
+ return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()})
+ if request.url.path == "/v1/users/login":
+ return httpx.Response(200, json={
+ "access_token": bad_token, "refresh_token": "rt", "expires_in": 3600})
+ return httpx.Response(404)
+
+ transport = httpx.MockTransport(handler)
+ client = HubClient(hub_config, node_keys)
+ client._http = httpx.AsyncClient(transport=transport, base_url="http://fake-hub")
+
+ with pytest.raises(AssertionError, match="jti"):
+ await client.login()
+
+
+@pytest.mark.asyncio
+async def test_register_idempotent(hub_keys, node_keys, hub_config):
+ def handler(request):
+ if request.url.path == "/v1/users/register":
+ return httpx.Response(409, json={"detail": "Username already taken"})
+ return httpx.Response(404)
+
+ transport = httpx.MockTransport(handler)
+ client = HubClient(hub_config, node_keys)
+ client._http = httpx.AsyncClient(transport=transport, base_url="http://fake-hub")
+
+ # Should not raise on 409
+ result = await client.register()
+ assert result == ""
+
+
+@pytest.mark.asyncio
+async def test_token_needs_refresh(hub_keys, node_keys, hub_config):
+ sk_hub, sk_hub_pem, pk_hub_pem = hub_keys
+ # Token expiring in 60s (< TOKEN_REFRESH_MARGIN of 300s)
+ short_token = make_token(sk_hub_pem, "uid", node_keys.pk_ed25519_b64, ttl=60)
+
+ def handler(request):
+ if request.url.path == "/v1/hub/pubkey":
+ return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()})
+ if request.url.path == "/v1/users/login":
+ return httpx.Response(200, json={
+ "access_token": short_token, "refresh_token": "rt", "expires_in": 60})
+ return httpx.Response(404)
+
+ transport = httpx.MockTransport(handler)
+ client = HubClient(hub_config, node_keys)
+ client._http = httpx.AsyncClient(transport=transport, base_url="http://fake-hub")
+
+ session = await client.login()
+ assert session.token_needs_refresh
+
+
+@pytest.mark.asyncio
+async def test_fetch_gek(hub_keys, node_keys, hub_config):
+ """Admin wraps GEK for this node; client fetches and unwraps."""
+ sk_hub, sk_hub_pem, pk_hub_pem = hub_keys
+ gek = generate_gek()
+
+ # Simulate admin wrapping GEK for this node
+ pk_x_raw = base64.b64decode(node_keys.pk_x25519_b64)
+ bundle = wrap_gek(gek, pk_x_raw)
+
+ token = make_token(sk_hub_pem, "uid", node_keys.pk_ed25519_b64)
+
+ def handler(request):
+ if request.url.path == "/v1/hub/pubkey":
+ return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()})
+ if request.url.path == "/v1/users/login":
+ return httpx.Response(200, json={
+ "access_token": token, "refresh_token": "rt", "expires_in": 3600})
+ if "/v1/groups/" in request.url.path and request.url.path.endswith("/gek"):
+ return httpx.Response(200, json=bundle)
+ return httpx.Response(404)
+
+ transport = httpx.MockTransport(handler)
+ client = HubClient(hub_config, node_keys)
+ client._http = httpx.AsyncClient(transport=transport, base_url="http://fake-hub")
+
+ await client.login()
+ recovered = await client.fetch_gek("group-abc")
+ assert recovered == gek
+
+
+@pytest.mark.asyncio
+async def test_hub_pk_cached(hub_keys, node_keys, hub_config, tmp_path):
+ _, _, pk_hub_pem = hub_keys
+ call_count = {"n": 0}
+
+ def handler(request):
+ if request.url.path == "/v1/hub/pubkey":
+ call_count["n"] += 1
+ return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()})
+ return httpx.Response(404)
+
+ transport = httpx.MockTransport(handler)
+ client = HubClient(hub_config, node_keys)
+ client._http = httpx.AsyncClient(transport=transport, base_url="http://fake-hub")
+
+ await client._fetch_hub_pk()
+ await client._fetch_hub_pk() # second call should use cache
+ assert call_count["n"] == 1, "Hub PK should be fetched only once"
generated by cgit v1.2.3 (git 2.51.0) at 2026-09-25 21:04:07 +0000