feat(node): add hub client with JWT offline verify and GEK fetch

Register/login/announce/token-refresh/GEK-unwrap. JWT jti and
pk_user verified at login. Hub PK cached after first fetch.
6/6 tests passing with httpx.MockTransport (no network).

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

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"