From d01a6ebf8526414b1ab605352a0305dca01d122c Mon Sep 17 00:00:00 2001
From: everbarry <alessandro.preiti@protonmail.com>
Date: Fri, 17 Apr 2026 14:10:52 +0200
Subject: [PATCH] v0.3

---
 cli/README.md                               | 175 +++----
 cli/pyproject.toml                          |  17 +-
 cli/zkac_cli/__init__.py                    |   3 -
 cli/zkac_cli/client.py                      | 368 +++++++++++++
 cli/zkac_cli/client_ops.py                  |  64 ---
 cli/zkac_cli/creds.py                       |  84 ---
 cli/zkac_cli/issuance_util.py               |  26 -
 cli/zkac_cli/main.py                        | 552 +++++++-------------
 cli/zkac_cli/paths.py                       |  12 +-
 cli/zkac_cli/registry_local.py              |  68 ---
 cli/zkac_cli/registry_log.py                |  59 ---
 cli/zkac_cli/server.py                      | 282 ++++++++++
 cli/zkac_cli/server_app.py                  | 272 ----------
 cli/zkac_cli/store.py                       | 181 +++++++
 cli/zkac_node.egg-info/PKG-INFO             |   5 +
 cli/zkac_node.egg-info/SOURCES.txt          |  14 +
 cli/zkac_node.egg-info/dependency_links.txt |   1 +
 cli/zkac_node.egg-info/entry_points.txt     |   2 +
 cli/zkac_node.egg-info/requires.txt         |   1 +
 cli/zkac_node.egg-info/top_level.txt        |   1 +
 docs/SECURITY.md                            |  57 +-
 python/zkac/tcp.py                          |  44 ++
 src/node.rs                                 |  34 ++
 src/python.rs                               |  31 ++
 tests/test_zkac_tcp.py                      |  58 ++
 25 files changed, 1350 insertions(+), 1061 deletions(-)
 create mode 100644 cli/zkac_cli/client.py
 delete mode 100644 cli/zkac_cli/client_ops.py
 delete mode 100644 cli/zkac_cli/creds.py
 delete mode 100644 cli/zkac_cli/issuance_util.py
 delete mode 100644 cli/zkac_cli/registry_local.py
 delete mode 100644 cli/zkac_cli/registry_log.py
 create mode 100644 cli/zkac_cli/server.py
 delete mode 100644 cli/zkac_cli/server_app.py
 create mode 100644 cli/zkac_cli/store.py
 create mode 100644 cli/zkac_node.egg-info/PKG-INFO
 create mode 100644 cli/zkac_node.egg-info/SOURCES.txt
 create mode 100644 cli/zkac_node.egg-info/dependency_links.txt
 create mode 100644 cli/zkac_node.egg-info/entry_points.txt
 create mode 100644 cli/zkac_node.egg-info/requires.txt
 create mode 100644 cli/zkac_node.egg-info/top_level.txt

diff --git a/cli/README.md b/cli/README.md
index 741fb30..95c3865 100644
--- a/cli/README.md
+++ b/cli/README.md
@@ -1,121 +1,108 @@
 # zkac-node CLI
 
-Command-line interface for [ZKAC](../README.md) using the **Python bindings only** (`zkac` package). It runs a **registry-capable server** (management + client-managed registries + optional issuance relay) and **per-user** material under `~/.zkac/` (or `$ZKAC_HOME`).
-
-## Prerequisites
-
-- Python ≥ 3.9  
-- The **`zkac`** extension built and installed from the repository root, for example:
-
-  ```bash
-  cd /path/to/ZKAC
-  maturin develop
-  # or: pip install -e .
-  ```
-
-## Installation
+Install the `zkac` wheel from the repo root first (`maturin develop` or `pip install .`), then:
 
 ```bash
-cd /path/to/ZKAC/cli
-pip install -e .
+pip install -e ./cli
+zkac-node --help
 ```
 
-This installs the **`zkac-node`** console script.
+## Quick start
 
-## Environment
+```bash
+# 1. Create identities (one per machine / actor)
+zkac-node identity init          # on admin machine
+zkac-node identity init          # on recipient machine (separate ~/.zkac)
 
-| Variable    | Meaning |
-|------------|---------|
-| `ZKAC_HOME` | Base directory for users (default: `~/.zkac`). Each user lives at `$ZKAC_HOME/<userid>/`. |
+# Recipient shares their issuance public key out-of-band:
+zkac-node identity show          # prints issuance pk (hex)
 
-## Server vs client
+# 2. Start the server (separate machine or same, different data-dir)
+zkac-node serve --data-dir /var/lib/zkac --port 9800 &
 
-- **Server** (`zkac-node serve`): a node that can **accept registry create/update** from an operator with the **`zkac.mgmt`** credential. It also serves **managed** sessions (BBS+ auth against stored client-managed registries) and optionally a **relay** port for blind issuance queues.
-- **Client**: a **userid** with files under `$ZKAC_HOME/<userid>/` (transport key, registries, credentials).
+# 3. Pin the server's public key (printed at startup)
+zkac-node server pin localhost:9800 --key <SERVER_PK_HEX>
 
-## Ports (defaults)
+# 4. Create a registry (admin side)
+zkac-node registry create localhost:9800 --roles analyst,operator
 
-| Port role   | Default | Purpose |
-|------------|---------|---------|
-| Management | 7400    | ZKAC + static role `zkac.mgmt`; JSON commands (create/update registry, issuance peek/grant). |
-| Managed    | 7401    | ZKAC + `RegistryManager`; member proves a role in a client-managed registry. |
-| Relay      | 7402    | Optional **plaintext** JSON line protocol for enqueue/poll of issuance requests. Use `--relay-port 0` on `serve` to disable. Binds with `--relay-bind` (default `127.0.0.1`). |
+# 5. Grant recipient the 'analyst' role (only needs their public key)
+zkac-node grant --server localhost:9800 \
+    --registry <REGISTRY_ID> --role analyst --to <RECIPIENT_PK_HEX>
 
-## Layout on disk
+# 6. Recipient lists pending credentials
+zkac-node credentials list --server localhost:9800
 
-**Per user** (`$ZKAC_HOME/<userid>/`):
+# 7. Recipient collects (host:port:registry_id:role)
+zkac-node collect localhost:9800:<REGISTRY_ID>:analyst
 
-- `transport.json` — Ristretto **client** transport keypair (`zkac.Keypair`).
-- `profile.json` — `userid` and metadata.
-- `registries/<slug>/` — one directory per logical registry:
-  - `admin.json` / `registry.json` — produced by `registry-init` (admin issuer material + public state + state cert).
-  - `roles/<name>.json` — member credential payloads for `connect`.
-  - `issued/` — files from `issue-member` (handoff).
-- `pending/<request_id>.json` — saved by `issuance-request` until `issuance-poll` finalizes.
-- `servers/<label>/server.json` — from `pin-server` (pinned server pubkey + host/ports).
-
-**Server data** (`serve --data <dir>`):
-
-- `transport.json` — server transport keypair.
-- `mgmt_issuer.json` — BBS issuer secret for the static `zkac.mgmt` role only.
-- `mgmt_member.json` — member credential for `zkac.mgmt` (give to operators who push registries).
-- `registry_events.json` — append-only log to rebuild `RegistryManager` after restart (registry state; **not** the issuance queue).
+# 8. Recipient authenticates anonymously
+zkac-node auth --registry <REGISTRY_ID> --role analyst --server localhost:9800
+```
 
 ## Commands
 
-Run `zkac-node --help` and `zkac-node <command> --help` for full flags.
-
 | Command | Description |
 |---------|-------------|
-| `serve` | Start server; `--init` creates keys and mgmt credential in `--data`. |
-| `user-create` | Create a new userid directory with a client transport key. |
-| `pin-server` | Save server pubkey + host/ports for a user (from server’s `transport.json`). |
-| `registry-init` | Offline: build a client-managed registry (`RegistryState` + state cert) under the user’s `registries/<slug>/`. |
-| `registry-push` | Push create (or `--update`) to the server over the **management** port using `mgmt_member.json` from the server data dir. |
-| `registry-import-public` | Copy `registry.json` from another user (public metadata for peers). |
-| `connect` | Open a **managed** ZKAC session and send a JSON command (`--command`, default `whoami`). |
-| `issue-member` | Admin: blind-issue a role credential to a file under `issued/` (out-of-band handoff). |
-| `import-credential` | Install a member JSON into `roles/<name>.json`. |
-| `issuance-request` | Enqueue a blind commitment on the **relay** (see caveats below). |
-| `issuance-grant` | Admin: list pending via mgmt peek, `issue_blind`, `grant_credential`. |
-| `issuance-poll` | Poll relay for blind signature; finalize credential into `roles/<role>.json` when ready. |
+| `identity init` | Generate issuance keypair under `~/.zkac/` |
+| `identity show` | Show issuance pk + owned registries + credentials |
+| `serve --data-dir D` | Run as a ZKAC server storing data in D |
+| `server pin <host:port> --key <hex>` | Pin a server's public key |
+| `registry create <server> --roles r1,r2` | Create a new registry (fresh BBS+ issuer) |
+| `registry update <server> --registry R --add-roles r3` | Add roles to a registry you own |
+| `registry get <server> --registry R` | Fetch registry state from a server |
+| `registry list` | List locally owned registries |
+| `grant --server S --registry R --role X --to <pk>` | Issue credential encrypted to recipient's pk |
+| `credentials list [--server S ...]` | Show local credentials + pending grants |
+| `collect <host:port:registry:role>` | Fetch + finalize one pending credential |
+| `auth --registry R --role X [--server S]` | Authenticate via ZKAC handshake |
 
-## Typical flows
+## Protocol
 
-**Operator: first-time server**
+All connections use a single encrypted channel:
 
-```bash
-export ZKAC_HOME=~/.zkac   # optional
-zkac-node serve --data ./myserver --init --mgmt-port 7400 --managed-port 7401 --relay-port 7402
+1. **Anonymous handshake** (X25519 ephemeral DH + Schnorr server identity proof
+   verified against a pinned public key) establishes an encrypted session.
+2. The first encrypted frame selects the mode:
+   - `{"op": "mgmt"}` — management commands (JSON request/reply loop)
+   - `{"op": "auth", ...}` — BBS+ role authentication
+
+Admin-only commands (`post_grant`) require a BBS+ presentation of the registry's
+`__admin__` credential bound to the session transcript hash — unlinkable across
+grants by construction.
+
+## Threat model
+
+The server is a **trustless, zero-information** relay:
+
+- All traffic is encrypted and server-authenticated (same handshake as role auth).
+- Registry state is opaque bytes with a BBS+ state certificate. The server cannot
+  mutate state or impersonate the admin without the BBS+ issuer secret.
+- Grants are end-to-end encrypted from admin to recipient using ephemeral X25519
+  ECDH. The server stores only `(recipient_pk, eph_pk, ciphertext)` per grant —
+  no registry ID, role name, or credential material.
+- Admin identity is unlinkable across `grant` calls (BBS+ presentations are
+  rerandomized per call).
+- Recipient's authenticated session uses a fresh ephemeral transport key + an
+  unlinkable BBS+ presentation, so the server cannot correlate sessions with
+  the mailbox pk used at collect time.
+- No user IDs exist on the server. Clients are identified only by ephemeral
+  keys (handshake) or issuance public keys (mailbox addressing).
+
+## Storage layout
+
+Client (`~/.zkac/`):
+```
+identity.json                       issuance keypair (long-term secret)
+admin/<registry_id>.json            BBS+ issuer + admin credential per owned registry
+credentials/<rid>_<role>.json       BBS+ credentials granted to this identity
+servers/<host_port>.json            pinned server public keys
 ```
 
-Distribute **`myserver/transport.json`** (public key) and **`myserver/mgmt_member.json`** (sensitive) to admins who push registries.
-
-**Admin: create registry offline and push**
-
-```bash
-zkac-node user-create alice
-zkac-node registry-init --user alice --slug demo --roles analyst reader
-zkac-node registry-push --user alice --slug demo --server-data ./myserver --host 127.0.0.1 --mgmt-port 7400
+Server (`--data-dir`):
 ```
-
-**Member: pin server, import public registry + credential, connect**
-
-```bash
-zkac-node pin-server --user bob --name prod --transport-file ./myserver/transport.json \
-  --host 10.0.0.1 --mgmt-port 7400 --managed-port 7401 --relay-port 7402
-zkac-node registry-import-public --from-user alice --to-user bob --slug demo
-# copy or issue credential file into bob/registries/demo/roles/analyst.json
-zkac-node connect --user bob --slug demo --credential roles/analyst.json \
-  --server-file ~/.zkac/bob/servers/prod/server.json --host 10.0.0.1 --managed-port 7401
+server_key.json                     Schnorr keypair (long-term server secret)
+registries/<rid>.state              raw RegistryState bytes
+registries/<rid>.cert               raw state cert bytes
+mailbox/<recipient_pk_hex>.json     [{grant_id, eph_pk_b64, ciphertext_b64}, ...]
 ```
-
-## Caveats
-
-1. **Issuance relay queue is in-memory.** Pending grants are lost if `serve` restarts before `issuance-grant` / `issuance-poll` complete. Registry snapshots are persisted via `registry_events.json`; the relay queue is not.
-2. **Relay is a localhost-oriented, plaintext JSON line protocol** (not the full E2E design in the Rust layer). Use `--relay-bind 127.0.0.1` and do not expose the relay port untrusted networks without additional protection.
-3. **`zkac.mgmt`** is a **separate** BBS role from the client-managed registry admin; it only authorizes **management RPCs** to the server (push registry, peek/grant issuance). Registry state integrity still comes from BBS+ state certificates inside `RegistryManager`.
-
-## Development
-
-The CLI package name is **`zkac-node-cli`** (see `pyproject.toml`). It declares no PyPI dependency on `zkac`; install the `zkac` wheel from the parent project first.
diff --git a/cli/pyproject.toml b/cli/pyproject.toml
index ab97ed7..4946e42 100644
--- a/cli/pyproject.toml
+++ b/cli/pyproject.toml
@@ -1,17 +1,12 @@
-[build-system]
-requires = ["hatchling"]
-build-backend = "hatchling.build"
-
 [project]
-name = "zkac-node-cli"
+name = "zkac-node"
 version = "0.1.0"
-description = "ZKAC node CLI (server + client) using the zkac Python bindings"
-readme = "README.md"
-requires-python = ">=3.9"
-dependencies = []
+requires-python = ">=3.10"
+dependencies = ["zkac"]
 
 [project.scripts]
 zkac-node = "zkac_cli.main:main"
 
-[tool.hatch.build.targets.wheel]
-packages = ["zkac_cli"]
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
diff --git a/cli/zkac_cli/__init__.py b/cli/zkac_cli/__init__.py
index d7e6f28..e69de29 100644
--- a/cli/zkac_cli/__init__.py
+++ b/cli/zkac_cli/__init__.py
@@ -1,3 +0,0 @@
-"""ZKAC node CLI — server (registry-capable) and client commands."""
-
-__version__ = "0.1.0"
diff --git a/cli/zkac_cli/client.py b/cli/zkac_cli/client.py
new file mode 100644
index 0000000..83a85b4
--- /dev/null
+++ b/cli/zkac_cli/client.py
@@ -0,0 +1,368 @@
+"""Client-side operations over a unified encrypted channel.
+
+All management and grant traffic uses the same anonymous handshake
+(X25519 + Schnorr server identity proof) as authenticated sessions.
+
+Grant flow (admin-initiated, server is trustless):
+  1. Admin knows recipient's X25519 issuance public key (out-of-band).
+  2. Admin locally prepares a blind request and issues the blind signature.
+  3. Admin encrypts the full credential material to the recipient's pk.
+  4. Admin posts the opaque ciphertext to the server's mailbox, authenticated
+     by a BBS+ admin presentation bound to the session transcript.
+  5. Recipient trial-decrypts each pending entry to find matching grants.
+
+The server sees only: recipient_pk, eph_pk, ciphertext, grant_id.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import socket
+
+import zkac
+from zkac.tcp import FramedSession, client_handshake_anon, client_handshake_managed
+
+from . import store
+
+
+def _b64(data: bytes) -> str:
+    return base64.b64encode(data).decode()
+
+
+def _unb64(s: str) -> bytes:
+    return base64.b64decode(s)
+
+
+def _parse_server(server: str) -> tuple[str, int]:
+    host, _, port = server.rpartition(":")
+    return host or "127.0.0.1", int(port)
+
+
+def parse_spec(spec: str) -> tuple[str, str, str]:
+    """Parse 'host:port:registry_id:role' into (server, registry_id, role)."""
+    parts = spec.rsplit(":", 2)
+    if len(parts) != 3:
+        raise ValueError(f"invalid spec {spec!r}, expected host:port:registry_id:role")
+    return parts[0], parts[1], parts[2]
+
+
+# ── Encrypted management session ─────────────────────────────────────
+
+def _resolve_server_pk(server: str) -> zkac.PublicKey:
+    """Load pinned server public key or fail."""
+    pin = store.load_server_pin(server)
+    if pin is None:
+        raise RuntimeError(
+            f"no pinned key for {server}; run: zkac-node server pin {server} --key <hex>"
+        )
+    return zkac.PublicKey.from_bytes(_unb64(pin["server_public_key_b64"]))
+
+
+def _mgmt_connect(server: str) -> tuple[socket.socket, FramedSession]:
+    """Open an encrypted management session to the server."""
+    host, port = _parse_server(server)
+    sock = socket.create_connection((host, port))
+    server_pk = _resolve_server_pk(server)
+    node = zkac.Node(zkac.Keypair())
+    session = client_handshake_anon(sock, node, server_pk)
+    framed = FramedSession(sock, session)
+    framed.send(json.dumps({"op": "mgmt"}).encode())
+    return sock, framed
+
+
+def _mgmt_cmd(framed: FramedSession, cmd: dict) -> dict:
+    framed.send(json.dumps(cmd).encode())
+    return json.loads(framed.recv())
+
+
+def _ok(resp: dict) -> dict:
+    if resp.get("error"):
+        raise RuntimeError(resp["error"])
+    return resp
+
+
+# ── Registry operations ──────────────────────────────────────────────
+
+def create_registry(server: str, role_names: list[str]) -> str:
+    identity = store.load_identity()
+    admin_mat = store.new_admin_material()
+    bbs_issuer, bbs_pk, admin_cred = store.reconstruct_admin(admin_mat)
+
+    role_entries = [(zkac.role_id(name), bbs_pk, 1) for name in role_names]
+    state = zkac.RegistryState.build(
+        bbs_pk, identity["issuance_pk"], 1, b"\x00" * 32, role_entries,
+    )
+    state_bytes = state.serialize()
+    state_cert = state.certify(admin_cred)
+    registry_id = state.registry_id()
+
+    sock, framed = _mgmt_connect(server)
+    try:
+        resp = _ok(_mgmt_cmd(framed, {
+            "cmd": "create_registry",
+            "state_bytes_b64": _b64(state_bytes),
+            "state_cert_b64": _b64(bytes(state_cert)),
+        }))
+    finally:
+        sock.close()
+
+    rid_hex = resp["registry_id"]
+    store.save_admin(rid_hex, {
+        "server": server,
+        "roles": role_names,
+        **admin_mat,
+    })
+    return rid_hex
+
+
+def update_registry(server: str, registry_id_hex: str, add_roles: list[str]):
+    admin_data = store.load_admin(registry_id_hex)
+    bbs_issuer, bbs_pk, admin_cred = store.reconstruct_admin(admin_data)
+    identity = store.load_identity()
+
+    sock, framed = _mgmt_connect(server)
+    try:
+        cur = _ok(_mgmt_cmd(framed, {
+            "cmd": "get_registry", "registry_id": registry_id_hex,
+        }))
+
+        old_state = zkac.RegistryState.deserialize(_unb64(cur["state_bytes_b64"]))
+        prev_hash = old_state.state_hash()
+        new_version = old_state.version() + 1
+
+        old_roles = admin_data.get("roles", [])
+        all_roles = list(old_roles) + [r for r in add_roles if r not in old_roles]
+        role_entries = [(zkac.role_id(name), bbs_pk, 1) for name in all_roles]
+
+        new_state = zkac.RegistryState.build(
+            bbs_pk, identity["issuance_pk"], new_version, bytes(prev_hash), role_entries,
+        )
+        new_cert = new_state.certify(admin_cred)
+
+        _ok(_mgmt_cmd(framed, {
+            "cmd": "update_registry",
+            "registry_id": registry_id_hex,
+            "state_bytes_b64": _b64(new_state.serialize()),
+            "state_cert_b64": _b64(bytes(new_cert)),
+        }))
+    finally:
+        sock.close()
+
+    admin_data["roles"] = all_roles
+    store.save_admin(registry_id_hex, admin_data)
+
+
+def get_registry(server: str, registry_id_hex: str) -> dict:
+    sock, framed = _mgmt_connect(server)
+    try:
+        return _ok(_mgmt_cmd(framed, {
+            "cmd": "get_registry", "registry_id": registry_id_hex,
+        }))
+    finally:
+        sock.close()
+
+
+def list_own_registries() -> list[dict]:
+    result = []
+    for rid in store.list_admin_registries():
+        data = store.load_admin(rid)
+        result.append({
+            "registry_id": rid,
+            "server": data.get("server", "?"),
+            "roles": data.get("roles", []),
+        })
+    return result
+
+
+# ── Admin-initiated grant ────────────────────────────────────────────
+
+def grant(server: str, registry_id_hex: str, role_name: str,
+          recipient_pk_hex: str) -> str:
+    """Admin issues a credential and posts it (encrypted) to the recipient's mailbox."""
+    admin_data = store.load_admin(registry_id_hex)
+    roles = admin_data.get("roles", [])
+    if role_name not in roles:
+        raise RuntimeError(f"role {role_name!r} not in registry (have: {roles})")
+
+    bbs_issuer, bbs_pk, admin_cred = store.reconstruct_admin(admin_data)
+    role_rid = zkac.role_id(role_name)
+    epoch = 1
+
+    req = zkac.prepare_blind_request()
+    blind_sig = bbs_issuer.issue_blind(req.commitment_with_proof(), role_rid, epoch)
+
+    payload = json.dumps({
+        "registry_id": registry_id_hex,
+        "role_name": role_name,
+        "epoch": epoch,
+        "issuer_pk_b64": _b64(bbs_pk.to_bytes()),
+        "blind_sig_b64": _b64(blind_sig),
+        "member_secret_b64": _b64(req.member_secret()),
+        "prover_blind_b64": _b64(req.prover_blind()),
+    }).encode()
+
+    recipient_pk = bytes.fromhex(recipient_pk_hex)
+    eph_kp = zkac.IssuanceKeypair()
+    ciphertext = eph_kp.encrypt(recipient_pk, payload)
+
+    sock, framed = _mgmt_connect(server)
+    try:
+        transcript_hash = bytes(framed.session.transcript_hash())
+        admin_proof = admin_cred.present(transcript_hash)
+        resp = _ok(_mgmt_cmd(framed, {
+            "cmd": "post_grant",
+            "registry_id": registry_id_hex,
+            "recipient_pk_hex": recipient_pk_hex,
+            "eph_pk_b64": _b64(eph_kp.public_key_bytes()),
+            "ciphertext_b64": _b64(ciphertext),
+            "admin_proof_b64": _b64(admin_proof),
+        }))
+    finally:
+        sock.close()
+
+    return resp["grant_id"]
+
+
+# ── Receiver: list + collect (trial-decrypt) ─────────────────────────
+
+def list_pending(server: str) -> list[dict]:
+    identity = store.load_identity()
+    pk_hex = identity["issuance_pk"].hex()
+    sock, framed = _mgmt_connect(server)
+    try:
+        resp = _ok(_mgmt_cmd(framed, {"cmd": "list_grants", "recipient_pk_hex": pk_hex}))
+    finally:
+        sock.close()
+
+    receiver_kp = zkac.IssuanceKeypair.from_secret(identity["issuance_sk"])
+    results = []
+    for entry in resp["grants"]:
+        try:
+            eph_pk = _unb64(entry["eph_pk_b64"])
+            ct = _unb64(entry["ciphertext_b64"])
+            plaintext = json.loads(receiver_kp.decrypt(eph_pk, ct))
+            results.append({
+                "grant_id": entry["grant_id"],
+                "registry_id": plaintext.get("registry_id", "?"),
+                "role_name": plaintext.get("role_name", "?"),
+            })
+        except Exception:
+            results.append({
+                "grant_id": entry["grant_id"],
+                "registry_id": "?",
+                "role_name": "(undecryptable)",
+            })
+    return results
+
+
+def collect(spec: str) -> dict:
+    """Fetch and finalize a pending credential described by 'host:port:registry_id:role'."""
+    server, registry_id_hex, role_name = parse_spec(spec)
+    identity = store.load_identity()
+    pk_hex = identity["issuance_pk"].hex()
+    receiver_kp = zkac.IssuanceKeypair.from_secret(identity["issuance_sk"])
+
+    sock, framed = _mgmt_connect(server)
+    try:
+        pending = _ok(_mgmt_cmd(framed, {
+            "cmd": "list_grants", "recipient_pk_hex": pk_hex,
+        }))["grants"]
+
+        # Trial-decrypt to find matching grant
+        target_grant_id = None
+        target_payload = None
+        for entry in pending:
+            try:
+                eph_pk = _unb64(entry["eph_pk_b64"])
+                ct = _unb64(entry["ciphertext_b64"])
+                plaintext = json.loads(receiver_kp.decrypt(eph_pk, ct))
+                if (plaintext.get("registry_id") == registry_id_hex and
+                        plaintext.get("role_name") == role_name):
+                    target_grant_id = entry["grant_id"]
+                    target_payload = plaintext
+                    break
+            except Exception:
+                continue
+
+        if target_grant_id is None:
+            raise RuntimeError(f"no pending grant for {spec}")
+
+        _ok(_mgmt_cmd(framed, {
+            "cmd": "claim_grant",
+            "recipient_pk_hex": pk_hex,
+            "grant_id": target_grant_id,
+        }))
+
+        reg_info = _ok(_mgmt_cmd(framed, {
+            "cmd": "get_registry", "registry_id": registry_id_hex,
+        }))
+    finally:
+        sock.close()
+
+    cred_data = {
+        "blind_sig_b64": target_payload["blind_sig_b64"],
+        "member_secret_b64": target_payload["member_secret_b64"],
+        "prover_blind_b64": target_payload["prover_blind_b64"],
+        "role_name": role_name,
+        "epoch": target_payload["epoch"],
+        "issuer_pk_b64": target_payload["issuer_pk_b64"],
+    }
+    cred = store.reconstruct_credential(cred_data)
+    cred.present(b"self-test")
+
+    store.save_credential(registry_id_hex, role_name, cred_data)
+
+    # Pin server + stash registry metadata locally
+    pin = store.load_server_pin(server)
+    if pin:
+        server_pk_b64 = pin["server_public_key_b64"]
+    else:
+        server_pk_b64 = _b64(b"\x00" * 32)
+    store.pin_server(server, server_pk_b64)
+
+    return {"registry_id": registry_id_hex, "role": role_name, "server": server}
+
+
+# ── Authenticated session ─────────────────────────────────────────────
+
+def authenticate(registry_id_hex: str, role_name: str,
+                 server: str | None = None) -> dict:
+    admin_data = None
+    try:
+        admin_data = store.load_admin(registry_id_hex)
+    except FileNotFoundError:
+        pass
+
+    if server is None:
+        if admin_data and admin_data.get("server"):
+            server = admin_data["server"]
+        else:
+            raise RuntimeError("server address required (--server host:port)")
+
+    cred_data = store.load_credential_data(registry_id_hex, role_name)
+    cred = store.reconstruct_credential(cred_data)
+
+    server_pk = _resolve_server_pk(server)
+    node = zkac.Node(zkac.Keypair())
+    host, port = _parse_server(server)
+
+    sock = socket.create_connection((host, port))
+    try:
+        session = client_handshake_anon(sock, node, server_pk)
+        framed = FramedSession(sock, session)
+
+        transcript_hash = bytes(session.transcript_hash())
+        auth_proof = cred.present(transcript_hash)
+        role_rid = zkac.role_id(role_name)
+
+        framed.send(json.dumps({
+            "op": "auth",
+            "registry_id": registry_id_hex,
+            "role_id": role_rid.hex(),
+            "bbs_auth_b64": _b64(auth_proof),
+        }).encode())
+
+        return json.loads(framed.recv())
+    finally:
+        sock.close()
diff --git a/cli/zkac_cli/client_ops.py b/cli/zkac_cli/client_ops.py
deleted file mode 100644
index 183cd61..0000000
--- a/cli/zkac_cli/client_ops.py
+++ /dev/null
@@ -1,64 +0,0 @@
-"""TCP clients for management, managed, and relay channels."""
-
-from __future__ import annotations
-
-import json
-import socket
-from typing import Any
-
-import zkac
-from zkac.tcp import FramedSession, client_handshake, client_handshake_managed
-
-
-def mgmt_call(
-    host: str,
-    port: int,
-    server_pk: zkac.PublicKey,
-    mgmt_cred: zkac.Credential,
-    cmd: dict[str, Any],
-) -> dict[str, Any]:
-    node = zkac.Node(zkac.Keypair())
-    sock = socket.create_connection((host, port))
-    try:
-        session = client_handshake(sock, node, server_pk, mgmt_cred)
-        framed = FramedSession(sock, session)
-        framed.send(json.dumps(cmd).encode("utf-8"))
-        return json.loads(framed.recv().decode("utf-8"))
-    finally:
-        sock.close()
-
-
-def managed_call(
-    host: str,
-    port: int,
-    server_pk: zkac.PublicKey,
-    cred: zkac.Credential,
-    registry_id: bytes,
-    cmd: dict[str, Any],
-) -> dict[str, Any]:
-    node = zkac.Node(zkac.Keypair())
-    sock = socket.create_connection((host, port))
-    try:
-        session = client_handshake_managed(sock, node, server_pk, cred, registry_id)
-        framed = FramedSession(sock, session)
-        framed.send(json.dumps(cmd).encode("utf-8"))
-        return json.loads(framed.recv().decode("utf-8"))
-    finally:
-        sock.close()
-
-
-def relay_line(host: str, port: int, obj: dict[str, Any]) -> dict[str, Any]:
-    payload = (json.dumps(obj) + "\n").encode("utf-8")
-    sock = socket.create_connection((host, port))
-    try:
-        sock.sendall(payload)
-        buf = b""
-        while b"\n" not in buf:
-            chunk = sock.recv(4096)
-            if not chunk:
-                break
-            buf += chunk
-        line = buf.split(b"\n", 1)[0]
-        return json.loads(line.decode("utf-8"))
-    finally:
-        sock.close()
diff --git a/cli/zkac_cli/creds.py b/cli/zkac_cli/creds.py
deleted file mode 100644
index 1d903f9..0000000
--- a/cli/zkac_cli/creds.py
+++ /dev/null
@@ -1,84 +0,0 @@
-"""Serialize / restore BBS+ member credentials and transport keys."""
-
-from __future__ import annotations
-
-import base64
-import json
-from pathlib import Path
-from typing import Any
-
-import zkac
-
-
-def save_json(path: Path, obj: Any) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(obj, indent=2), encoding="utf-8")
-
-
-def load_json(path: Path) -> Any:
-    return json.loads(path.read_text(encoding="utf-8"))
-
-
-def member_payload(
-    blind_sig: bytes,
-    req: zkac.BlindRequest,
-    role_id: bytes,
-    epoch: int,
-    issuer_pk: zkac.BbsPublicKey,
-) -> dict[str, Any]:
-    return {
-        "role_id_hex": role_id.hex(),
-        "epoch": epoch,
-        "blind_sig_b64": base64.b64encode(blind_sig).decode(),
-        "member_secret_b64": base64.b64encode(req.member_secret()).decode(),
-        "prover_blind_b64": base64.b64encode(req.prover_blind()).decode(),
-        "issuer_public_key_b64": base64.b64encode(issuer_pk.to_bytes()).decode(),
-    }
-
-
-def load_member_credential(data: dict[str, Any]) -> zkac.Credential:
-    pk = zkac.BbsPublicKey.from_bytes(base64.b64decode(data["issuer_public_key_b64"]))
-    rid = bytes.fromhex(data["role_id_hex"])
-    return zkac.Credential.finalize(
-        base64.b64decode(data["blind_sig_b64"]),
-        base64.b64decode(data["member_secret_b64"]),
-        base64.b64decode(data["prover_blind_b64"]),
-        rid,
-        int(data["epoch"]),
-        pk,
-    )
-
-
-def save_transport_keypair(path: Path, kp: zkac.Keypair) -> None:
-    save_json(
-        path,
-        {
-            "secret_key_b64": base64.b64encode(kp.secret_key_bytes()).decode(),
-            "public_key_b64": base64.b64encode(kp.public_key().to_bytes()).decode(),
-        },
-    )
-
-
-def load_transport_keypair(path: Path) -> zkac.Keypair:
-    t = load_json(path)
-    return zkac.Keypair.from_secret_key(base64.b64decode(t["secret_key_b64"]))
-
-
-def save_server_transport(path: Path, kp: zkac.Keypair) -> None:
-    save_json(
-        path,
-        {
-            "server_secret_key_b64": base64.b64encode(kp.secret_key_bytes()).decode(),
-            "server_public_key_b64": base64.b64encode(kp.public_key().to_bytes()).decode(),
-        },
-    )
-
-
-def load_server_public_key(path: Path) -> zkac.PublicKey:
-    t = load_json(path)
-    return zkac.PublicKey.from_bytes(base64.b64decode(t["server_public_key_b64"]))
-
-
-def load_server_keypair(path: Path) -> zkac.Keypair:
-    t = load_json(path)
-    return zkac.Keypair.from_secret_key(base64.b64decode(t["server_secret_key_b64"]))
diff --git a/cli/zkac_cli/issuance_util.py b/cli/zkac_cli/issuance_util.py
deleted file mode 100644
index f59eb97..0000000
--- a/cli/zkac_cli/issuance_util.py
+++ /dev/null
@@ -1,26 +0,0 @@
-"""Helpers around RegistryManager issuance queues (Python API drains on take)."""
-
-from __future__ import annotations
-
-import zkac
-
-
-def peek_pending_requests(
-    mgr: zkac.RegistryManager, registry_id: bytes
-) -> list[tuple[bytes, bytes, bytes, bytes]]:
-    """Return pending (request_id, role_id, eph_pk, ciphertext) and re-queue them."""
-    items = mgr.take_pending_requests(registry_id)
-    out: list[tuple[bytes, bytes, bytes, bytes]] = []
-    for tup in items:
-        req_id, role_id, eph_pk, enc = tup
-        out.append((req_id, role_id, eph_pk, enc))
-        mgr.queue_issuance_request(registry_id, req_id, role_id, eph_pk, enc)
-    return out
-
-
-def take_pending_requests(
-    mgr: zkac.RegistryManager, registry_id: bytes
-) -> list[tuple[bytes, bytes, bytes, bytes]]:
-    """Drain pending queue (admin processes and must grant or lose requests)."""
-    items = mgr.take_pending_requests(registry_id)
-    return [(a, b, c, d) for a, b, c, d in items]
diff --git a/cli/zkac_cli/main.py b/cli/zkac_cli/main.py
index 0a3e003..953fcc1 100644
--- a/cli/zkac_cli/main.py
+++ b/cli/zkac_cli/main.py
@@ -1,414 +1,246 @@
-"""zkac-node CLI entrypoint."""
+"""CLI entry point for zkac-node."""
 
 from __future__ import annotations
 
 import argparse
-import base64
 import json
-import secrets
-import shutil
 import sys
-from pathlib import Path
 
-import zkac
-
-from zkac_cli import creds
-from zkac_cli import client_ops
-from zkac_cli import paths
-from zkac_cli import registry_local
-from zkac_cli import server_app
+from . import client, store
 
 
-MGMT_ROLE = server_app.MGMT_ROLE
+# ── identity ──────────────────────────────────────────────────────────
+
+def _cmd_identity_init(_args):
+    if store.identity_exists():
+        print("identity already exists")
+        ident = store.load_identity()
+        print(f"  issuance public key: {ident['issuance_pk'].hex()}")
+        return
+    path = store.create_identity()
+    ident = store.load_identity()
+    print("created identity")
+    print(f"  issuance public key: {ident['issuance_pk'].hex()}")
+    print(f"  (share this key out-of-band to receive credentials)")
+    print(f"  stored in {path}")
 
 
-def cmd_serve(args: argparse.Namespace) -> None:
-    data = Path(args.data).resolve()
-    if args.init or not (data / "transport.json").is_file():
-        data.mkdir(parents=True, exist_ok=True)
-        transport = zkac.Keypair()
-        issuer = zkac.BbsIssuer()
-        pk = issuer.public_key()
-        mg_rid = zkac.role_id(MGMT_ROLE)
-        req = zkac.prepare_blind_request()
-        sig = issuer.issue_blind(req.commitment_with_proof(), mg_rid, 1)
-        _ = zkac.Credential.finalize(
-            sig, req.member_secret(), req.prover_blind(), mg_rid, 1, pk
-        )
-        creds.save_server_transport(data / "transport.json", transport)
-        creds.save_json(
-            data / "mgmt_issuer.json",
-            {"issuer_secret_b64": base64.b64encode(issuer.secret_key_bytes()).decode()},
-        )
-        m = creds.member_payload(sig, req, mg_rid, 1, pk)
-        creds.save_json(data / "mgmt_member.json", m)
-        print(f"Initialized server data in {data}", flush=True)
-        print(
-            "Copy mgmt_member.json for operators; distribute server_public_key from transport.json",
-            flush=True,
-        )
-    relay = None if args.relay_port == 0 else args.relay_port
-    relay_bind = args.relay_bind
-    print(
-        f"Starting: mgmt={args.mgmt_port} managed={args.managed_port} relay={relay!r} bind={relay_bind}",
-        flush=True,
-    )
-    server_app.serve(data, args.mgmt_port, args.managed_port, relay, relay_bind)
+def _cmd_identity_show(_args):
+    ident = store.load_identity()
+    print(f"  issuance pk: {ident['issuance_pk'].hex()}")
+
+    regs = client.list_own_registries()
+    if regs:
+        print(f"  registries owned: {len(regs)}")
+        for r in regs:
+            print(f"    {r['registry_id']}  @ {r['server']}  roles={r['roles']}")
+
+    creds = store.list_credentials()
+    if creds:
+        print(f"  credentials: {len(creds)}")
+        for reg_hex, role in creds:
+            print(f"    {reg_hex[:16]}… / {role}")
 
 
-def cmd_user_create(args: argparse.Namespace) -> None:
-    uid = args.userid
-    d = paths.ensure_user(uid)
-    kp = zkac.Keypair()
-    creds.save_transport_keypair(d / "transport.json", kp)
-    creds.save_json(d / "profile.json", {"userid": uid})
-    print(f"User {uid!r}: wrote {d / 'transport.json'}")
+# ── server ────────────────────────────────────────────────────────────
+
+def _cmd_serve(args):
+    from .server import serve
+    serve(args.data_dir, args.host, args.port)
 
 
-def cmd_registry_init(args: argparse.Namespace) -> None:
-    base = paths.ensure_user(args.user)
-    out = (base / "registries" / args.slug).resolve()
-    r = registry_local.create_registry_bundle(args.slug, list(args.roles), out)
-    print(json.dumps(r, indent=2))
+def _cmd_server_pin(args):
+    pk_hex = args.key
+    import base64
+    pk_bytes = bytes.fromhex(pk_hex)
+    store.pin_server(args.server, base64.b64encode(pk_bytes).decode())
+    print(f"pinned {args.server} -> {pk_hex[:16]}…")
 
 
-def cmd_registry_push(args: argparse.Namespace) -> None:
-    server_data = Path(args.server_data).resolve()
-    reg_dir = (paths.user_dir(args.user) / "registries" / args.slug).resolve()
-    reg = creds.load_json(reg_dir / "registry.json")
-    mg = creds.load_json(server_data / "mgmt_member.json")
-    t = creds.load_json(server_data / "transport.json")
-    server_pk = zkac.PublicKey.from_bytes(base64.b64decode(t["server_public_key_b64"]))
-    mgmt_cred = creds.load_member_credential(mg)
-    cmd = {"cmd": "create_registry", "state_b64": reg["state_bytes_b64"], "state_cert_b64": reg["state_cert_b64"]}
-    if args.update:
-        cmd = {"cmd": "update_registry", "state_b64": reg["state_bytes_b64"], "state_cert_b64": reg["state_cert_b64"]}
-    out = client_ops.mgmt_call(args.host, args.mgmt_port, server_pk, mgmt_cred, cmd)
-    print(json.dumps(out, indent=2))
+# ── registry ──────────────────────────────────────────────────────────
+
+def _cmd_registry_create(args):
+    roles = [r.strip() for r in args.roles.split(",")]
+    rid = client.create_registry(args.server, roles)
+    print(f"registry created: {rid}")
+    print(f"  roles: {', '.join(roles)}")
 
 
-def cmd_connect(args: argparse.Namespace) -> None:
-    reg_dir = (paths.user_dir(args.user) / "registries" / args.slug).resolve()
-    reg_path = reg_dir / "registry.json"
-    if reg_path.is_file():
-        reg = creds.load_json(reg_path)
-    elif args.registry_id:
-        reg = {"registry_id_hex": args.registry_id}
-    else:
-        print(
-            "Missing registry.json. Copy from the admin (public) or pass --registry-id.",
-            file=sys.stderr,
-        )
-        raise SystemExit(1)
-    cred_path = (reg_dir / args.credential).resolve()
-    m = creds.load_json(cred_path)
-    credential = creds.load_member_credential(m)
-    server = creds.load_json(Path(args.server_file).expanduser().resolve())
-    server_pk = zkac.PublicKey.from_bytes(base64.b64decode(server["server_public_key_b64"]))
-    registry_id = bytes.fromhex(reg["registry_id_hex"])
-    body = client_ops.managed_call(
-        args.host,
-        args.managed_port,
-        server_pk,
-        credential,
-        registry_id,
-        {"cmd": args.command},
-    )
-    print(json.dumps(body, indent=2))
+def _cmd_registry_update(args):
+    add = [r.strip() for r in args.add_roles.split(",")]
+    client.update_registry(args.server, args.registry, add)
+    print(f"registry updated: {args.registry[:16]}…")
+    print(f"  added roles: {', '.join(add)}")
 
 
-def cmd_issuance_request(args: argparse.Namespace) -> None:
-    """Queue a blind issuance request (payload is commitment bytes; localhost relay only)."""
-    reg = creds.load_json(paths.user_dir(args.user) / "registries" / args.slug / "registry.json")
-    issuance_pk = base64.b64decode(reg["issuance_public_key_b64"])
-    req = zkac.prepare_blind_request()
-    payload = req.commitment_with_proof()
-    req_id = secrets.token_bytes(32)
-    eph_pk = secrets.token_bytes(32)
-    # Store locally for finalize
-    pending = {
-        "request_id_hex": req_id.hex(),
-        "role_name": args.role,
-        "member_secret_b64": base64.b64encode(req.member_secret()).decode(),
-        "prover_blind_b64": base64.b64encode(req.prover_blind()).decode(),
-        "issuer_public_key_b64": reg["admin_issuer_public_key_b64"],
-        "epoch": 1,
-    }
-    pdir = paths.user_dir(args.user) / "pending"
-    pdir.mkdir(parents=True, exist_ok=True)
-    creds.save_json(pdir / f"{req_id.hex()}.json", pending)
-
-    obj = {
-        "cmd": "enqueue",
-        "registry_id_hex": reg["registry_id_hex"],
-        "role_name": args.role,
-        "request_id_hex": req_id.hex(),
-        "eph_pk_hex": eph_pk.hex(),
-        "payload_b64": base64.b64encode(payload).decode(),
-    }
-    out = client_ops.relay_line(args.host, args.relay_port, obj)
-    print(json.dumps(out, indent=2))
-    print(f"Saved pending material to {pdir / (req_id.hex() + '.json')}")
+def _cmd_registry_get(args):
+    info = client.get_registry(args.server, args.registry)
+    print(f"registry: {args.registry}")
+    print(f"  state bytes: {len(info.get('state_bytes_b64', ''))} chars (b64)")
 
 
-def cmd_issuance_poll(args: argparse.Namespace) -> None:
-    reg = creds.load_json(paths.user_dir(args.user) / "registries" / args.slug / "registry.json")
-    obj = {
-        "cmd": "poll",
-        "registry_id_hex": reg["registry_id_hex"],
-        "request_id_hex": args.request_id,
-    }
-    out = client_ops.relay_line(args.host, args.relay_port, obj)
-    print(json.dumps(out, indent=2))
-    if out.get("status") == "ready":
-        pend = creds.load_json(paths.user_dir(args.user) / "pending" / f"{args.request_id}.json")
-        pk = zkac.BbsPublicKey.from_bytes(base64.b64decode(pend["issuer_public_key_b64"]))
-        rid = zkac.role_id(pend["role_name"])
-        blind_sig = base64.b64decode(out["blind_sig_b64"])
-        zkac.Credential.finalize(
-            blind_sig,
-            base64.b64decode(pend["member_secret_b64"]),
-            base64.b64decode(pend["prover_blind_b64"]),
-            rid,
-            int(pend["epoch"]),
-            pk,
-        )
-        out_dir = paths.user_dir(args.user) / "registries" / args.slug / "roles"
-        out_dir.mkdir(parents=True, exist_ok=True)
-        member = {
-            "role_id_hex": rid.hex(),
-            "epoch": int(pend["epoch"]),
-            "blind_sig_b64": base64.b64encode(blind_sig).decode(),
-            "member_secret_b64": pend["member_secret_b64"],
-            "prover_blind_b64": pend["prover_blind_b64"],
-            "issuer_public_key_b64": pend["issuer_public_key_b64"],
-        }
-        dest = out_dir / f"{pend['role_name']}.json"
-        creds.save_json(dest, member)
-        print(f"You have role {pend['role_name']!r}; credential saved to {dest}")
-        print("Use: zkac-node connect ... --credential roles/{0}.json".format(pend["role_name"]))
+def _cmd_registry_list(_args):
+    regs = client.list_own_registries()
+    if not regs:
+        print("no registries")
+        return
+    for r in regs:
+        print(f"  {r['registry_id']}  @ {r['server']}  roles={r['roles']}")
 
 
-def cmd_issuance_grant(args: argparse.Namespace) -> None:
-    server_data = Path(args.server_data).resolve()
-    reg_dir = (paths.user_dir(args.admin_user) / "registries" / args.slug).resolve()
-    admin = creds.load_json(reg_dir / "admin.json")
-    issuer = zkac.BbsIssuer.from_secret_key(base64.b64decode(admin["admin_issuer_secret_b64"]))
-    t = creds.load_json(server_data / "transport.json")
-    mg = creds.load_json(server_data / "mgmt_member.json")
-    server_pk = zkac.PublicKey.from_bytes(base64.b64decode(t["server_public_key_b64"]))
-    mgmt_cred = creds.load_member_credential(mg)
+# ── grant (admin-initiated) ──────────────────────────────────────────
 
-    peek = client_ops.mgmt_call(
-        args.host,
-        args.mgmt_port,
-        server_pk,
-        mgmt_cred,
-        {"cmd": "issuance_peek", "registry_id_hex": args.registry_id},
-    )
-    pending = peek.get("pending") or []
-    target = None
-    for p in pending:
-        if p["request_id_hex"] == args.request_id:
-            target = p
-            break
-    if target is None:
-        print("request_id not in pending queue", file=sys.stderr)
-        raise SystemExit(1)
-    commit = base64.b64decode(target["payload_b64"])
-    role_id = bytes.fromhex(target["role_id_hex"])
-    blind = issuer.issue_blind(commit, role_id, 1)
-    out = client_ops.mgmt_call(
-        args.host,
-        args.mgmt_port,
-        server_pk,
-        mgmt_cred,
-        {
-            "cmd": "issuance_grant",
-            "registry_id_hex": args.registry_id,
-            "request_id_hex": args.request_id,
-            "blind_sig_b64": base64.b64encode(blind).decode(),
-        },
-    )
-    print(json.dumps(out, indent=2))
+def _cmd_grant(args):
+    gid = client.grant(args.server, args.registry, args.role, args.to)
+    print(f"granted {args.role!r} to {args.to[:16]}…")
+    print(f"  grant id: {gid}")
+    print(f"  recipient can collect with:")
+    print(f"    zkac-node collect {args.server}:{args.registry}:{args.role}")
 
 
-def cmd_issue_member_file(args: argparse.Namespace) -> None:
-    """Admin issues a role credential locally (out-of-band handoff)."""
-    reg_dir = (paths.user_dir(args.admin_user) / "registries" / args.slug).resolve()
-    admin = creds.load_json(reg_dir / "admin.json")
-    issuer = zkac.BbsIssuer.from_secret_key(base64.b64decode(admin["admin_issuer_secret_b64"]))
-    pk = issuer.public_key()
-    rid = zkac.role_id(args.role)
-    req = zkac.prepare_blind_request()
-    sig = issuer.issue_blind(req.commitment_with_proof(), rid, 1)
-    payload = creds.member_payload(sig, req, rid, 1, pk)
-    out_path = reg_dir / "issued" / f"{args.role}_{args.target_user}.json"
-    creds.save_json(out_path, payload)
-    print(f"Wrote {out_path}")
+# ── credentials list / collect ────────────────────────────────────────
+
+def _cmd_credentials_list(args):
+    local = store.list_credentials()
+    print("local credentials:")
+    if not local:
+        print("  (none)")
+    for reg_hex, role in local:
+        print(f"  {reg_hex}:{role}")
+
+    servers = list(args.server or [])
+    for s in store.list_pinned_servers():
+        if s not in servers:
+            servers.append(s)
+
+    if not servers:
+        print("\n(no servers to query; pass --server host:port to check for pending)")
+        return
+
+    print("\npending grants:")
+    any_pending = False
+    for srv in servers:
+        try:
+            grants = client.list_pending(srv)
+        except Exception as exc:
+            print(f"  [{srv}] error: {exc}")
+            continue
+        for g in grants:
+            any_pending = True
+            rid = g.get("registry_id", "?")
+            role = g.get("role_name", "?")
+            if rid != "?" and role != "?" and store.has_credential(rid, role):
+                note = "  (already collected locally)"
+            else:
+                note = ""
+            print(f"  {srv}:{rid}:{role}{note}")
+    if not any_pending:
+        print("  (none)")
 
 
-def cmd_registry_import_public(args: argparse.Namespace) -> None:
-    """Copy public registry.json (state + cert summary) from another local user."""
-    src = (paths.user_dir(args.from_user) / "registries" / args.slug / "registry.json").resolve()
-    dst_dir = paths.ensure_user(args.to_user) / "registries" / args.slug
-    dst_dir.mkdir(parents=True, exist_ok=True)
-    shutil.copy(src, dst_dir / "registry.json")
-    print(f"Copied {src} -> {dst_dir / 'registry.json'}")
+def _cmd_collect(args):
+    result = client.collect(args.spec)
+    print("collected credential")
+    print(f"  registry: {result['registry_id']}")
+    print(f"  role:     {result['role']}")
+    print(f"  server:   {result['server']}")
 
 
-def cmd_import_credential(args: argparse.Namespace) -> None:
-    src = Path(args.file).resolve()
-    data = creds.load_json(src)
-    dest = paths.user_dir(args.user) / "registries" / args.slug / "roles" / f"{args.name}.json"
-    creds.save_json(dest, data)
-    print(f"Imported credential to {dest}")
+# ── auth ──────────────────────────────────────────────────────────────
+
+def _cmd_auth(args):
+    resp = client.authenticate(args.registry, args.role, server=args.server)
+    print(json.dumps(resp, indent=2))
 
 
-def cmd_pin_server(args: argparse.Namespace) -> None:
-    """Save server public key + ports for a user (from server's transport.json)."""
-    tpath = Path(args.transport_file).resolve()
-    t = creds.load_json(tpath)
-    d = paths.ensure_user(args.user) / "servers" / args.name
-    d.mkdir(parents=True, exist_ok=True)
-    creds.save_json(
-        d / "server.json",
-        {
-            "server_public_key_b64": t["server_public_key_b64"],
-            "host": args.host,
-            "mgmt_port": args.mgmt_port,
-            "managed_port": args.managed_port,
-            "relay_port": args.relay_port,
-        },
-    )
-    print(f"Pinned server {args.name!r} for user {args.user!r} -> {d / 'server.json'}")
+# ── argparse wiring ───────────────────────────────────────────────────
 
+def main():
+    p = argparse.ArgumentParser(prog="zkac-node")
+    sub = p.add_subparsers(dest="group", required=True)
 
-def build_parser() -> argparse.ArgumentParser:
-    p = argparse.ArgumentParser(
-        prog="zkac-node",
-        description="ZKAC node CLI (server + client). "
-        "Requires the 'zkac' package from this repo (maturin develop). "
-        "Set ZKAC_HOME to override ~/.zkac/. "
-        "Issuance relay queues are in-memory — use the same server process between request and grant.",
-    )
-    sub = p.add_subparsers(dest="command", required=True)
+    # identity
+    id_p = sub.add_parser("identity", help="manage local identity")
+    id_sub = id_p.add_subparsers(dest="action", required=True)
 
-    ps = sub.add_parser("serve", help="Run server (registry-capable node)")
-    ps.add_argument("--data", type=Path, required=True, help="Server data directory")
-    ps.add_argument("--init", action="store_true", help="Initialize data dir (transport + mgmt issuer)")
-    ps.add_argument("--mgmt-port", type=int, default=7400)
-    ps.add_argument("--managed-port", type=int, default=7401)
-    ps.add_argument(
-        "--relay-port",
-        type=int,
-        default=7402,
-        help="Plaintext relay for issuance queue (localhost only). Use 0 to disable.",
-    )
-    ps.add_argument("--relay-bind", default="127.0.0.1")
-    ps.set_defaults(func=cmd_serve)
+    c = id_sub.add_parser("init", help="generate a new identity")
+    c.set_defaults(func=_cmd_identity_init)
 
-    su = sub.add_parser("user-create", help="Create a user directory under ~/.zkac/")
-    su.add_argument("userid")
-    su.set_defaults(func=cmd_user_create)
+    c = id_sub.add_parser("show", help="show identity + registries + credentials")
+    c.set_defaults(func=_cmd_identity_show)
 
-    pi = sub.add_parser("pin-server", help="Save server connection info for a user")
-    pi.add_argument("--user", required=True)
-    pi.add_argument("--name", required=True, help="Label for this server")
-    pi.add_argument("--transport-file", required=True, help="Path to server's transport.json")
-    pi.add_argument("--host", required=True)
-    pi.add_argument("--mgmt-port", type=int, default=7400)
-    pi.add_argument("--managed-port", type=int, default=7401)
-    pi.add_argument("--relay-port", type=int, default=7402)
-    pi.set_defaults(func=cmd_pin_server)
+    # serve
+    c = sub.add_parser("serve", help="run as a ZKAC server node")
+    c.add_argument("--data-dir", required=True, help="server data directory")
+    c.add_argument("--host", default="127.0.0.1")
+    c.add_argument("--port", type=int, default=9800)
+    c.set_defaults(func=_cmd_serve)
 
-    ri = sub.add_parser("registry-init", help="Create a client-managed registry offline")
-    ri.add_argument("--user", required=True)
-    ri.add_argument("--slug", required=True)
-    ri.add_argument("--roles", nargs="+", required=True)
-    ri.set_defaults(func=cmd_registry_init)
+    # server pin
+    srv_p = sub.add_parser("server", help="manage server pins")
+    srv_sub = srv_p.add_subparsers(dest="action", required=True)
 
-    rp = sub.add_parser("registry-push", help="Push registry snapshot to server (mgmt channel)")
-    rp.add_argument("--user", required=True)
-    rp.add_argument("--slug", required=True)
-    rp.add_argument("--server-data", type=Path, required=True, help="Server data dir (transport + mgmt_member)")
-    rp.add_argument("--host", default="127.0.0.1")
-    rp.add_argument("--mgmt-port", type=int, default=7400)
-    rp.add_argument("--update", action="store_true", help="Send update_registry instead of create")
-    rp.set_defaults(func=cmd_registry_push)
+    c = srv_sub.add_parser("pin", help="pin a server's public key")
+    c.add_argument("server", help="host:port")
+    c.add_argument("--key", required=True, help="server public key (hex)")
+    c.set_defaults(func=_cmd_server_pin)
 
-    cc = sub.add_parser("connect", help="Managed ZKAC session test (whoami/get_registry)")
-    cc.add_argument("--user", required=True)
-    cc.add_argument("--slug", required=True)
-    cc.add_argument("--credential", default="roles/member.json", help="Relative path under registries/<slug>/")
-    cc.add_argument(
-        "--registry-id",
-        default="",
-        help="If registry.json is absent, hex registry id (pin public registry first)",
-    )
-    cc.add_argument("--server-file", required=True, help="Pinned server.json from pin-server")
-    cc.add_argument("--host", required=True)
-    cc.add_argument("--managed-port", type=int, default=7401)
-    cc.add_argument("--command", default="whoami")
-    cc.set_defaults(func=cmd_connect)
+    # registry
+    reg_p = sub.add_parser("registry", help="manage registries")
+    reg_sub = reg_p.add_subparsers(dest="action", required=True)
 
-    ir = sub.add_parser(
-        "issuance-request",
-        help="Enqueue blind commitment on relay (localhost; see help text)",
-    )
-    ir.add_argument("--user", required=True)
-    ir.add_argument("--slug", required=True)
-    ir.add_argument("--role", required=True)
-    ir.add_argument("--host", default="127.0.0.1")
-    ir.add_argument("--relay-port", type=int, default=7402)
-    ir.set_defaults(func=cmd_issuance_request)
+    c = reg_sub.add_parser("create", help="create a new registry on a server")
+    c.add_argument("server", help="host:port")
+    c.add_argument("--roles", required=True, help="comma-separated role names")
+    c.set_defaults(func=_cmd_registry_create)
 
-    ip = sub.add_parser("issuance-poll", help="Poll relay for blind signature")
-    ip.add_argument("--user", required=True)
-    ip.add_argument("--slug", required=True)
-    ip.add_argument("--request-id", required=True, dest="request_id")
-    ip.add_argument("--host", default="127.0.0.1")
-    ip.add_argument("--relay-port", type=int, default=7402)
-    ip.set_defaults(func=cmd_issuance_poll)
+    c = reg_sub.add_parser("update", help="add roles to a registry you own")
+    c.add_argument("server", help="host:port")
+    c.add_argument("--registry", required=True)
+    c.add_argument("--add-roles", required=True, help="comma-separated new roles")
+    c.set_defaults(func=_cmd_registry_update)
 
-    ig = sub.add_parser("issuance-grant", help="Admin: issue blind sig and push grant to server")
-    ig.add_argument("--admin-user", required=True)
-    ig.add_argument("--slug", required=True)
-    ig.add_argument("--server-data", type=Path, required=True)
-    ig.add_argument("--registry-id", required=True, dest="registry_id")
-    ig.add_argument("--request-id", required=True, dest="request_id")
-    ig.add_argument("--host", default="127.0.0.1")
-    ig.add_argument("--mgmt-port", type=int, default=7400)
-    ig.set_defaults(func=cmd_issuance_grant)
+    c = reg_sub.add_parser("get", help="fetch registry info from server")
+    c.add_argument("server", help="host:port")
+    c.add_argument("--registry", required=True)
+    c.set_defaults(func=_cmd_registry_get)
 
-    im = sub.add_parser("issue-member", help="Admin: issue credential to file for handoff")
-    im.add_argument("--admin-user", required=True)
-    im.add_argument("--slug", required=True)
-    im.add_argument("--role", required=True)
-    im.add_argument("--target-user", required=True)
-    im.set_defaults(func=cmd_issue_member_file)
+    c = reg_sub.add_parser("list", help="list local registries (owned)")
+    c.set_defaults(func=_cmd_registry_list)
 
-    irp = sub.add_parser(
-        "registry-import-public",
-        help="Copy registry.json from another user (public state + cert metadata)",
-    )
-    irp.add_argument("--from-user", required=True, dest="from_user")
-    irp.add_argument("--to-user", required=True, dest="to_user")
-    irp.add_argument("--slug", required=True)
-    irp.set_defaults(func=cmd_registry_import_public)
+    # grant
+    c = sub.add_parser("grant", help="issue a credential to a recipient (admin)")
+    c.add_argument("--server", required=True, help="host:port")
+    c.add_argument("--registry", required=True)
+    c.add_argument("--role", required=True)
+    c.add_argument("--to", required=True,
+                   help="recipient's issuance public key (hex)")
+    c.set_defaults(func=_cmd_grant)
 
-    ii = sub.add_parser("import-credential", help="Copy a member json into a user's registry folder")
-    ii.add_argument("--user", required=True)
-    ii.add_argument("--slug", required=True)
-    ii.add_argument("--name", required=True, help="Filename base (e.g. analyst)")
-    ii.add_argument("file", type=Path)
-    ii.set_defaults(func=cmd_import_credential)
+    # credentials list
+    cred_p = sub.add_parser("credentials", help="credentials (local + pending)")
+    cred_sub = cred_p.add_subparsers(dest="action", required=True)
+    c = cred_sub.add_parser("list", help="show local + pending credentials")
+    c.add_argument("--server", action="append",
+                   help="server to query (host:port); may be repeated")
+    c.set_defaults(func=_cmd_credentials_list)
 
-    return p
+    # collect
+    c = sub.add_parser("collect", help="fetch and finalize a pending credential")
+    c.add_argument("spec", help="host:port:registry_id:role")
+    c.set_defaults(func=_cmd_collect)
 
+    # auth
+    c = sub.add_parser("auth", help="authenticate to a server with a credential")
+    c.add_argument("--registry", required=True)
+    c.add_argument("--role", required=True)
+    c.add_argument("--server", default=None, help="host:port (optional if known)")
+    c.set_defaults(func=_cmd_auth)
 
-def main() -> None:
-    args = build_parser().parse_args()
+    args = p.parse_args()
+    if not hasattr(args, "func"):
+        p.print_help()
+        sys.exit(1)
     args.func(args)
 
 
diff --git a/cli/zkac_cli/paths.py b/cli/zkac_cli/paths.py
index 4748614..a161d00 100644
--- a/cli/zkac_cli/paths.py
+++ b/cli/zkac_cli/paths.py
@@ -8,12 +8,6 @@ def zkac_home() -> Path:
     return Path(os.environ.get("ZKAC_HOME", Path.home() / ".zkac"))
 
 
-def user_dir(userid: str) -> Path:
-    return zkac_home() / userid
-
-
-def ensure_user(userid: str) -> Path:
-    zkac_home().mkdir(parents=True, exist_ok=True)
-    d = user_dir(userid)
-    d.mkdir(parents=True, exist_ok=True)
-    return d
+def data_dir() -> Path:
+    """Server data directory (overridable via ZKAC_DATA_DIR)."""
+    return Path(os.environ.get("ZKAC_DATA_DIR", zkac_home()))
diff --git a/cli/zkac_cli/registry_local.py b/cli/zkac_cli/registry_local.py
deleted file mode 100644
index 5bfde57..0000000
--- a/cli/zkac_cli/registry_local.py
+++ /dev/null
@@ -1,68 +0,0 @@
-"""Offline client-managed registry creation (same structure as demo/setup_managed_demo)."""
-
-from __future__ import annotations
-
-import base64
-import json
-from pathlib import Path
-
-import zkac
-
-from zkac_cli import creds
-
-
-def create_registry_bundle(
-    slug: str,
-    role_names: list[str],
-    out_dir: Path,
-) -> dict[str, str]:
-    """
-    Build a new registry with admin issuer = role issuer for all listed roles.
-    Writes admin.json, registry.json under out_dir.
-    """
-    out_dir.mkdir(parents=True, exist_ok=True)
-    admin_issuer = zkac.BbsIssuer()
-    admin_pk = admin_issuer.public_key()
-    admin_rid = zkac.admin_role_id()
-
-    req = zkac.prepare_blind_request()
-    sig = admin_issuer.issue_blind(req.commitment_with_proof(), admin_rid, 0)
-    admin_cred = zkac.Credential.finalize(
-        sig, req.member_secret(), req.prover_blind(), admin_rid, 0, admin_pk
-    )
-
-    issuance_kp = zkac.IssuanceKeypair()
-    role_entries = [(zkac.role_id(name), admin_pk, 1) for name in role_names]
-
-    state = zkac.RegistryState.build(
-        admin_pk, issuance_kp.public_key_bytes(), 1, b"\x00" * 32, role_entries
-    )
-    state_bytes = state.serialize()
-    state_cert = state.certify(admin_cred)
-    registry_id = state.registry_id()
-
-    admin_payload = {
-        "slug": slug,
-        "admin_issuer_secret_b64": base64.b64encode(admin_issuer.secret_key_bytes()).decode(),
-        "admin_issuer_public_key_b64": base64.b64encode(admin_pk.to_bytes()).decode(),
-        "admin_member_secret_b64": base64.b64encode(req.member_secret()).decode(),
-        "admin_prover_blind_b64": base64.b64encode(req.prover_blind()).decode(),
-        "admin_blind_sig_b64": base64.b64encode(sig).decode(),
-        "issuance_secret_b64": base64.b64encode(issuance_kp.secret_bytes()).decode(),
-        "issuance_public_key_b64": base64.b64encode(issuance_kp.public_key_bytes()).decode(),
-        "registry_id_hex": registry_id.hex(),
-    }
-    creds.save_json(out_dir / "admin.json", admin_payload)
-
-    reg_payload = {
-        "slug": slug,
-        "registry_id_hex": registry_id.hex(),
-        "state_bytes_b64": base64.b64encode(state_bytes).decode(),
-        "state_cert_b64": base64.b64encode(bytes(state_cert)).decode(),
-        "admin_issuer_public_key_b64": base64.b64encode(admin_pk.to_bytes()).decode(),
-        "issuance_public_key_b64": base64.b64encode(issuance_kp.public_key_bytes()).decode(),
-        "roles": role_names,
-    }
-    creds.save_json(out_dir / "registry.json", reg_payload)
-
-    return {"registry_id_hex": registry_id.hex(), "path": str(out_dir)}
diff --git a/cli/zkac_cli/registry_log.py b/cli/zkac_cli/registry_log.py
deleted file mode 100644
index ac67709..0000000
--- a/cli/zkac_cli/registry_log.py
+++ /dev/null
@@ -1,59 +0,0 @@
-"""Persist RegistryManager by replaying create/update events."""
-
-from __future__ import annotations
-
-import base64
-import json
-from pathlib import Path
-from typing import Any
-
-import zkac
-
-
-def save_events(path: Path, events: list[dict[str, Any]]) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps({"events": events}, indent=2), encoding="utf-8")
-
-
-def load_events(path: Path) -> list[dict[str, Any]]:
-    if not path.is_file():
-        return []
-    data = json.loads(path.read_text(encoding="utf-8"))
-    return list(data.get("events", []))
-
-
-def replay_manager(events: list[dict[str, Any]]) -> zkac.RegistryManager:
-    """Rebuild manager from persisted events (multiple registries supported)."""
-    mgr = zkac.RegistryManager()
-    by_rid: dict[bytes, list[dict[str, Any]]] = {}
-    for e in events:
-        st = base64.b64decode(e["state_b64"])
-        st_obj = zkac.RegistryState.deserialize(st)
-        rid = st_obj.registry_id()
-        by_rid.setdefault(rid, []).append(e)
-    for rid, evs in by_rid.items():
-        evs.sort(
-            key=lambda e: zkac.RegistryState.deserialize(
-                base64.b64decode(e["state_b64"])
-            ).version()
-        )
-        for i, e in enumerate(evs):
-            st = base64.b64decode(e["state_b64"])
-            cert = base64.b64decode(e["state_cert_b64"])
-            if i == 0:
-                mgr.create(st, cert)
-            else:
-                mgr.update(rid, st, cert)
-    return mgr
-
-
-def append_event(path: Path, op: str, state_bytes: bytes, state_cert: bytes) -> None:
-    events = load_events(path)
-    events.append(
-        {
-            "op": op,
-            "state_b64": base64.b64encode(state_bytes).decode(),
-            "state_cert_b64": base64.b64encode(state_cert).decode(),
-        }
-    )
-    save_events(path, events)
diff --git a/cli/zkac_cli/server.py b/cli/zkac_cli/server.py
new file mode 100644
index 0000000..c271f92
--- /dev/null
+++ b/cli/zkac_cli/server.py
@@ -0,0 +1,282 @@
+"""ZKAC server: all traffic over a single encrypted, server-authenticated channel.
+
+Every connection performs an anonymous handshake (X25519 + Schnorr server
+identity proof).  The first encrypted frame selects the mode:
+
+  {"op": "mgmt"}   management commands (create_registry, post_grant, ...)
+  {"op": "auth", "registry_id": hex, "bbs_auth_b64": ...}  role authentication
+
+The server stores only cryptographically verified opaque blobs:
+  <data_dir>/server_key.json          Schnorr keypair
+  <data_dir>/registries/<rid>.state   raw RegistryState bytes
+  <data_dir>/registries/<rid>.cert    raw state cert bytes
+  <data_dir>/mailbox/<pk_hex>.json    [{grant_id, eph_pk_b64, ciphertext_b64}, ...]
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import os
+import socket
+import threading
+import traceback
+from pathlib import Path
+
+import zkac
+from zkac.tcp import FramedSession, server_handshake_anon
+
+
+def _b64(data: bytes) -> str:
+    return base64.b64encode(data).decode()
+
+
+def _unb64(s: str) -> bytes:
+    return base64.b64decode(s)
+
+
+# ── Opaque server storage ─────────────────────────────────────────────
+
+class _ServerStore:
+    """Thread-safe, opaque persistence for registries and mailbox."""
+
+    def __init__(self, data_dir: Path):
+        self._dir = data_dir
+        self._reg_dir = data_dir / "registries"
+        self._mbox_dir = data_dir / "mailbox"
+        self._reg_dir.mkdir(parents=True, exist_ok=True)
+        self._mbox_dir.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.Lock()
+
+    # ── server key ────────────────────────────────────────────────────
+
+    def load_or_create_keypair(self) -> zkac.Keypair:
+        kf = self._dir / "server_key.json"
+        if kf.exists():
+            data = json.loads(kf.read_text())
+            return zkac.Keypair.from_secret_key(_unb64(data["secret_b64"]))
+        kp = zkac.Keypair()
+        kf.write_text(json.dumps({
+            "secret_b64": _b64(kp.secret_key_bytes()),
+            "public_b64": _b64(kp.public_key().to_bytes()),
+        }, indent=2))
+        return kp
+
+    # ── registries ────────────────────────────────────────────────────
+
+    def save_registry(self, rid_hex: str, state_bytes: bytes, cert_bytes: bytes):
+        with self._lock:
+            (self._reg_dir / f"{rid_hex}.state").write_bytes(state_bytes)
+            (self._reg_dir / f"{rid_hex}.cert").write_bytes(cert_bytes)
+
+    def load_all_registries(self, mgr: zkac.RegistryManager) -> int:
+        count = 0
+        for p in sorted(self._reg_dir.glob("*.state")):
+            rid_hex = p.stem
+            cert_path = self._reg_dir / f"{rid_hex}.cert"
+            if not cert_path.exists():
+                continue
+            try:
+                mgr.create(p.read_bytes(), cert_path.read_bytes())
+                count += 1
+            except Exception as exc:
+                print(f"[server] skip registry {rid_hex}: {exc}")
+        return count
+
+    # ── mailbox ───────────────────────────────────────────────────────
+
+    def _mbox_path(self, pk_hex: str) -> Path:
+        return self._mbox_dir / f"{pk_hex}.json"
+
+    def post_grant(self, recipient_pk_hex: str, entry: dict) -> str:
+        grant_id = os.urandom(16).hex()
+        entry = {"grant_id": grant_id, **entry}
+        with self._lock:
+            p = self._mbox_path(recipient_pk_hex)
+            entries = json.loads(p.read_text()) if p.exists() else []
+            entries.append(entry)
+            p.write_text(json.dumps(entries, indent=2))
+        return grant_id
+
+    def list_grants(self, recipient_pk_hex: str) -> list[dict]:
+        with self._lock:
+            p = self._mbox_path(recipient_pk_hex)
+            if not p.exists():
+                return []
+            return json.loads(p.read_text())
+
+    def claim_grant(self, recipient_pk_hex: str, grant_id: str) -> dict | None:
+        with self._lock:
+            p = self._mbox_path(recipient_pk_hex)
+            if not p.exists():
+                return None
+            entries = json.loads(p.read_text())
+            for i, e in enumerate(entries):
+                if e["grant_id"] == grant_id:
+                    del entries[i]
+                    if entries:
+                        p.write_text(json.dumps(entries, indent=2))
+                    else:
+                        p.unlink(missing_ok=True)
+                    return e
+            return None
+
+
+# ── Command dispatch (inside encrypted session) ──────────────────────
+
+def _dispatch(cmd: dict, mgr: zkac.RegistryManager, store: _ServerStore,
+              server_pk_b64: str, transcript_hash: bytes) -> dict:
+    try:
+        action = cmd.get("cmd")
+
+        if action == "server_info":
+            return {"ok": True, "server_public_key_b64": server_pk_b64}
+
+        if action == "create_registry":
+            state_bytes = _unb64(cmd["state_bytes_b64"])
+            state_cert = _unb64(cmd["state_cert_b64"])
+            rid = mgr.create(state_bytes, state_cert)
+            store.save_registry(rid.hex(), state_bytes, state_cert)
+            return {"ok": True, "registry_id": rid.hex()}
+
+        if action == "get_registry":
+            rid = bytes.fromhex(cmd["registry_id"])
+            state_bytes, state_cert = mgr.get(rid)
+            return {
+                "ok": True,
+                "state_bytes_b64": _b64(state_bytes),
+                "state_cert_b64": _b64(state_cert),
+            }
+
+        if action == "update_registry":
+            rid = bytes.fromhex(cmd["registry_id"])
+            state_bytes = _unb64(cmd["state_bytes_b64"])
+            state_cert = _unb64(cmd["state_cert_b64"])
+            mgr.update(rid, state_bytes, state_cert)
+            store.save_registry(cmd["registry_id"], state_bytes, state_cert)
+            return {"ok": True}
+
+        if action == "post_grant":
+            rid = bytes.fromhex(cmd["registry_id"])
+            proof = _unb64(cmd["admin_proof_b64"])
+            if not mgr.verify_admin(rid, proof, transcript_hash):
+                return {"error": "admin proof failed"}
+            entry = {
+                "eph_pk_b64": cmd["eph_pk_b64"],
+                "ciphertext_b64": cmd["ciphertext_b64"],
+            }
+            gid = store.post_grant(cmd["recipient_pk_hex"], entry)
+            return {"ok": True, "grant_id": gid}
+
+        if action == "list_grants":
+            entries = store.list_grants(cmd["recipient_pk_hex"])
+            return {"ok": True, "grants": entries}
+
+        if action == "claim_grant":
+            entry = store.claim_grant(cmd["recipient_pk_hex"], cmd["grant_id"])
+            if entry is None:
+                return {"error": "grant not found"}
+            return {"ok": True, "grant": entry}
+
+        return {"error": f"unknown command: {action}"}
+
+    except Exception as exc:
+        return {"error": str(exc)}
+
+
+# ── Connection handler ────────────────────────────────────────────────
+
+def _handle_conn(conn: socket.socket, addr: tuple, node: zkac.Node,
+                 mgr: zkac.RegistryManager, store: _ServerStore,
+                 server_pk_b64: str):
+    peer = f"{addr[0]}:{addr[1]}"
+    try:
+        session = server_handshake_anon(conn, node)
+        framed = FramedSession(conn, session)
+        transcript_hash = bytes(session.transcript_hash())
+
+        hello = json.loads(framed.recv())
+        op = hello.get("op")
+
+        if op == "mgmt":
+            while True:
+                try:
+                    data = framed.recv()
+                except (ConnectionError, OSError):
+                    break
+                cmd = json.loads(data)
+                resp = _dispatch(cmd, mgr, store, server_pk_b64, transcript_hash)
+                framed.send(json.dumps(resp).encode())
+
+        elif op == "auth":
+            registry_id = bytes.fromhex(hello["registry_id"])
+            role_id = bytes.fromhex(hello["role_id"])
+            proof_bytes = _unb64(hello["bbs_auth_b64"])
+
+            ok = mgr.verify_presentation(
+                registry_id, role_id, proof_bytes, transcript_hash,
+            )
+            if not ok:
+                framed.send(json.dumps({"error": "auth failed"}).encode())
+                return
+
+            resp = {
+                "status": "authenticated",
+                "registry_id": registry_id.hex(),
+                "role_id": role_id.hex(),
+            }
+            framed.send(json.dumps(resp).encode())
+
+            # keep session open for app traffic
+            while True:
+                try:
+                    data = framed.recv()
+                except (ConnectionError, OSError):
+                    break
+                framed.send(data)
+        else:
+            framed.send(json.dumps({"error": f"unknown op: {op}"}).encode())
+
+    except (ConnectionError, BrokenPipeError, OSError):
+        pass
+    except Exception as exc:
+        print(f"[server] {peer} error: {exc}")
+        traceback.print_exc()
+    finally:
+        conn.close()
+
+
+# ── Public entry point ────────────────────────────────────────────────
+
+def serve(data_dir: str, host: str = "127.0.0.1", port: int = 9800):
+    dd = Path(data_dir)
+    dd.mkdir(parents=True, exist_ok=True)
+
+    store = _ServerStore(dd)
+    kp = store.load_or_create_keypair()
+    server_pk_b64 = _b64(kp.public_key().to_bytes())
+    node = zkac.Node(kp)
+
+    mgr = zkac.RegistryManager()
+    n = store.load_all_registries(mgr)
+
+    print(f"server public key: {_unb64(server_pk_b64).hex()}")
+    print(f"loaded {n} registries")
+    print(f"listening on {host}:{port}")
+
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind((host, port))
+    sock.listen(8)
+    try:
+        while True:
+            conn, addr = sock.accept()
+            threading.Thread(
+                target=_handle_conn,
+                args=(conn, addr, node, mgr, store, server_pk_b64),
+                daemon=True,
+            ).start()
+    except KeyboardInterrupt:
+        print("\nshutdown")
+    finally:
+        sock.close()
diff --git a/cli/zkac_cli/server_app.py b/cli/zkac_cli/server_app.py
deleted file mode 100644
index d43346c..0000000
--- a/cli/zkac_cli/server_app.py
+++ /dev/null
@@ -1,272 +0,0 @@
-"""TCP servers: ZKAC management, ZKAC managed app, localhost relay (credential queue)."""
-
-from __future__ import annotations
-
-import base64
-import json
-import socket
-import threading
-import traceback
-from pathlib import Path
-from typing import Callable
-
-import zkac
-from zkac.tcp import FramedSession, server_handshake, server_handshake_managed
-
-from zkac_cli import issuance_util
-from zkac_cli import registry_log
-
-
-MGMT_ROLE = "zkac.mgmt"
-
-
-def build_static_registry(issuer_pk: zkac.BbsPublicKey) -> zkac.RoleRegistry:
-    reg = zkac.RoleRegistry()
-    reg.register_role(zkac.role_id(MGMT_ROLE), issuer_pk, 1)
-    return reg
-
-
-def handle_mgmt_session(
-    conn: socket.socket,
-    data_dir: Path,
-    mgr: zkac.RegistryManager,
-    registry_ids: set[bytes],
-    events_path: Path,
-    save_registry: Callable[[], None],
-) -> None:
-    from zkac_cli import creds
-
-    framed = None
-    try:
-        sk = creds.load_server_keypair(data_dir / "transport.json")
-        node = zkac.Node(sk)
-        issuer = zkac.BbsIssuer.from_secret_key(
-            base64.b64decode(creds.load_json(data_dir / "mgmt_issuer.json")["issuer_secret_b64"])
-        )
-        static_reg = build_static_registry(issuer.public_key())
-        session, role_id = server_handshake(conn, node, static_reg)
-        if role_id != zkac.role_id(MGMT_ROLE):
-            return
-        framed = FramedSession(conn, session)
-        raw = framed.recv()
-        msg = json.loads(raw.decode("utf-8"))
-        cmd = msg["cmd"]
-        if cmd == "create_registry":
-            st = base64.b64decode(msg["state_b64"])
-            cert = base64.b64decode(msg["state_cert_b64"])
-            rid = mgr.create(st, cert)
-            registry_ids.add(rid)
-            registry_log.append_event(events_path, "create", st, cert)
-            save_registry()
-            out = {"ok": True, "registry_id_hex": rid.hex()}
-        elif cmd == "update_registry":
-            st = base64.b64decode(msg["state_b64"])
-            cert = base64.b64decode(msg["state_cert_b64"])
-            st_o = zkac.RegistryState.deserialize(st)
-            rid = st_o.registry_id()
-            mgr.update(rid, st, cert)
-            registry_log.append_event(events_path, "update", st, cert)
-            save_registry()
-            out = {"ok": True, "registry_id_hex": rid.hex()}
-        elif cmd == "get_registry":
-            rid = bytes.fromhex(msg["registry_id_hex"])
-            st, c = mgr.get(rid)
-            out = {
-                "ok": True,
-                "state_b64": base64.b64encode(st).decode(),
-                "state_cert_b64": base64.b64encode(c).decode(),
-            }
-        elif cmd == "list_registries":
-            out = {"ok": True, "registry_ids_hex": [h.hex() for h in sorted(registry_ids)]}
-        elif cmd == "issuance_peek":
-            rid = bytes.fromhex(msg["registry_id_hex"])
-            pending = issuance_util.peek_pending_requests(mgr, rid)
-            out = {
-                "ok": True,
-                "pending": [
-                    {
-                        "request_id_hex": a.hex(),
-                        "role_id_hex": b.hex(),
-                        "eph_pk_hex": c.hex(),
-                        "payload_b64": base64.b64encode(d).decode(),
-                    }
-                    for a, b, c, d in pending
-                ],
-            }
-        elif cmd == "issuance_grant":
-            rid = bytes.fromhex(msg["registry_id_hex"])
-            req_id = bytes.fromhex(msg["request_id_hex"])
-            blind = base64.b64decode(msg["blind_sig_b64"])
-            mgr.grant_credential(rid, req_id, blind)
-            save_registry()
-            out = {"ok": True}
-        else:
-            out = {"ok": False, "error": f"unknown cmd {cmd}"}
-        framed.send(json.dumps(out).encode("utf-8"))
-    except Exception:
-        traceback.print_exc()
-    finally:
-        conn.close()
-
-
-def run_managed_handler(
-    conn: socket.socket,
-    data_dir: Path,
-    mgr: zkac.RegistryManager,
-) -> None:
-    from zkac_cli import creds
-
-    try:
-        sk = creds.load_server_keypair(data_dir / "transport.json")
-        node = zkac.Node(sk)
-        session, registry_id, role_id = server_handshake_managed(conn, node, mgr)
-        framed = FramedSession(conn, session)
-        raw = framed.recv()
-        msg = json.loads(raw.decode("utf-8"))
-        cmd = msg.get("cmd", "ping")
-        if cmd == "get_registry":
-            rid = bytes.fromhex(msg["registry_id_hex"])
-            st, c = mgr.get(rid)
-            body = {
-                "ok": True,
-                "state_b64": base64.b64encode(st).decode(),
-                "state_cert_b64": base64.b64encode(c).decode(),
-            }
-        elif cmd == "whoami":
-            body = {
-                "ok": True,
-                "registry_id_hex": registry_id.hex(),
-                "role_id_hex": role_id.hex(),
-            }
-        else:
-            body = {
-                "ok": True,
-                "registry_id_hex": registry_id.hex(),
-                "role_id_hex": role_id.hex(),
-                "note": "authenticated managed session",
-            }
-        framed.send(json.dumps(body).encode("utf-8"))
-    except Exception:
-        traceback.print_exc()
-    finally:
-        conn.close()
-
-
-def handle_relay_session(
-    conn: socket.socket,
-    mgr: zkac.RegistryManager,
-    save_registry: Callable[[], None],
-) -> None:
-    try:
-        buf = b""
-        while b"\n" not in buf:
-            chunk = conn.recv(4096)
-            if not chunk:
-                return
-            buf += chunk
-        line, _, _ = buf.partition(b"\n")
-        msg = json.loads(line.decode("utf-8"))
-        cmd = msg["cmd"]
-        if cmd == "enqueue":
-            rid = bytes.fromhex(msg["registry_id_hex"])
-            role = zkac.role_id(msg["role_name"])
-            req_id = bytes.fromhex(msg["request_id_hex"])
-            eph = bytes.fromhex(msg["eph_pk_hex"])
-            blob = base64.b64decode(msg["payload_b64"])
-            mgr.queue_issuance_request(rid, req_id, role, eph, blob)
-            save_registry()
-            out = {"ok": True, "status": "queued"}
-        elif cmd == "poll":
-            rid = bytes.fromhex(msg["registry_id_hex"])
-            req_id = bytes.fromhex(msg["request_id_hex"])
-            g = mgr.take_granted_credential(rid, req_id)
-            save_registry()
-            if g is None:
-                out = {"ok": True, "status": "pending"}
-            else:
-                out = {
-                    "ok": True,
-                    "status": "ready",
-                    "blind_sig_b64": base64.b64encode(g).decode(),
-                }
-        else:
-            out = {"ok": False, "error": "unknown relay cmd"}
-        conn.sendall((json.dumps(out) + "\n").encode("utf-8"))
-    except Exception:
-        traceback.print_exc()
-    finally:
-        conn.close()
-
-
-def serve(
-    data_dir: Path,
-    mgmt_port: int,
-    managed_port: int,
-    relay_port: int | None,
-    relay_bind: str,
-) -> None:
-    events_path = data_dir / "registry_events.json"
-    registry_ids: set[bytes] = set()
-    if events_path.is_file():
-        mgr = registry_log.replay_manager(registry_log.load_events(events_path))
-        for e in registry_log.load_events(events_path):
-            st = base64.b64decode(e["state_b64"])
-            registry_ids.add(zkac.RegistryState.deserialize(st).registry_id())
-    else:
-        mgr = zkac.RegistryManager()
-
-    def save_registry() -> None:
-        return
-
-    def mgmt_loop() -> None:
-        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        s.bind(("0.0.0.0", mgmt_port))
-        s.listen(8)
-        print(f"[mgmt] ZKAC listening on 0.0.0.0:{mgmt_port}")
-        while True:
-            c, a = s.accept()
-            print(f"[mgmt] connect {a}")
-            threading.Thread(
-                target=handle_mgmt_session,
-                args=(c, data_dir, mgr, registry_ids, events_path, save_registry),
-                daemon=True,
-            ).start()
-
-    def managed_loop() -> None:
-        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        s.bind(("0.0.0.0", managed_port))
-        s.listen(8)
-        print(f"[managed] ZKAC listening on 0.0.0.0:{managed_port}")
-        while True:
-            c, a = s.accept()
-            print(f"[managed] connect {a}")
-            threading.Thread(
-                target=run_managed_handler,
-                args=(c, data_dir, mgr),
-                daemon=True,
-            ).start()
-
-    def relay_loop() -> None:
-        if relay_port is None:
-            return
-        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        s.bind((relay_bind, relay_port))
-        s.listen(8)
-        print(f"[relay] plaintext JSON lines on {relay_bind}:{relay_port}")
-        while True:
-            c, a = s.accept()
-            print(f"[relay] connect {a}")
-            threading.Thread(
-                target=handle_relay_session,
-                args=(c, mgr, save_registry),
-                daemon=True,
-            ).start()
-
-    threading.Thread(target=mgmt_loop, daemon=True).start()
-    threading.Thread(target=managed_loop, daemon=True).start()
-    if relay_port is not None:
-        threading.Thread(target=relay_loop, daemon=True).start()
-    threading.Event().wait()
diff --git a/cli/zkac_cli/store.py b/cli/zkac_cli/store.py
new file mode 100644
index 0000000..7c7c913
--- /dev/null
+++ b/cli/zkac_cli/store.py
@@ -0,0 +1,181 @@
+"""Persistent storage for ZKAC identities, admin bundles, credentials, and server pins.
+
+Layout (single identity, no userid):
+  ~/.zkac/identity.json                   issuance keypair (long-term client secret)
+  ~/.zkac/admin/<registry_id>.json        BBS+ issuer + admin credential per owned registry
+  ~/.zkac/credentials/<rid>_<role>.json   BBS+ credentials granted to this identity
+  ~/.zkac/servers/<host_port>.json        pinned server public keys
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+from pathlib import Path
+
+import zkac
+
+from .paths import zkac_home
+
+
+def _b64(data: bytes) -> str:
+    return base64.b64encode(data).decode()
+
+
+def _unb64(s: str) -> bytes:
+    return base64.b64decode(s)
+
+
+def _home() -> Path:
+    d = zkac_home()
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+# ── Identity (single, no userid) ─────────────────────────────────────
+
+def create_identity() -> Path:
+    d = _home()
+    p = d / "identity.json"
+    if p.exists():
+        raise FileExistsError(f"identity already exists at {p}")
+
+    issuance_kp = zkac.IssuanceKeypair()
+    identity = {
+        "issuance_secret_b64": _b64(issuance_kp.secret_bytes()),
+        "issuance_public_b64": _b64(issuance_kp.public_key_bytes()),
+    }
+    p.write_text(json.dumps(identity, indent=2))
+    for sub in ("admin", "credentials", "servers"):
+        (d / sub).mkdir(exist_ok=True)
+    return d
+
+
+def load_identity() -> dict:
+    data = json.loads((_home() / "identity.json").read_text())
+    return {
+        "issuance_sk": _unb64(data["issuance_secret_b64"]),
+        "issuance_pk": _unb64(data["issuance_public_b64"]),
+    }
+
+
+def identity_exists() -> bool:
+    return (_home() / "identity.json").exists()
+
+
+# ── Server pins ───────────────────────────────────────────────────────
+
+def _server_key(server: str) -> str:
+    return server.replace(":", "_")
+
+
+def pin_server(server: str, server_pk_b64: str):
+    d = _home() / "servers"
+    d.mkdir(exist_ok=True)
+    (d / f"{_server_key(server)}.json").write_text(
+        json.dumps({"server_public_key_b64": server_pk_b64}, indent=2)
+    )
+
+
+def load_server_pin(server: str) -> dict | None:
+    p = _home() / "servers" / f"{_server_key(server)}.json"
+    if not p.exists():
+        return None
+    return json.loads(p.read_text())
+
+
+def list_pinned_servers() -> list[str]:
+    d = _home() / "servers"
+    if not d.exists():
+        return []
+    return sorted(p.stem.replace("_", ":") for p in d.glob("*.json"))
+
+
+# ── Admin bundles (per registry) ──────────────────────────────────────
+
+def new_admin_material() -> dict:
+    bbs_issuer = zkac.BbsIssuer()
+    bbs_pk = bbs_issuer.public_key()
+    admin_rid = zkac.admin_role_id()
+
+    req = zkac.prepare_blind_request()
+    sig = bbs_issuer.issue_blind(req.commitment_with_proof(), admin_rid, 0)
+
+    return {
+        "bbs_issuer_secret_b64": _b64(bbs_issuer.secret_key_bytes()),
+        "bbs_issuer_public_b64": _b64(bbs_pk.to_bytes()),
+        "admin_blind_sig_b64": _b64(sig),
+        "admin_member_secret_b64": _b64(req.member_secret()),
+        "admin_prover_blind_b64": _b64(req.prover_blind()),
+    }
+
+
+def reconstruct_admin(data: dict) -> tuple:
+    bbs_issuer = zkac.BbsIssuer.from_secret_key(_unb64(data["bbs_issuer_secret_b64"]))
+    bbs_pk = bbs_issuer.public_key()
+    admin_cred = zkac.Credential.finalize(
+        _unb64(data["admin_blind_sig_b64"]),
+        _unb64(data["admin_member_secret_b64"]),
+        _unb64(data["admin_prover_blind_b64"]),
+        zkac.admin_role_id(), 0, bbs_pk,
+    )
+    return bbs_issuer, bbs_pk, admin_cred
+
+
+def save_admin(registry_id_hex: str, info: dict):
+    d = _home() / "admin"
+    d.mkdir(exist_ok=True)
+    (d / f"{registry_id_hex}.json").write_text(json.dumps(info, indent=2))
+
+
+def load_admin(registry_id_hex: str) -> dict:
+    return json.loads((_home() / "admin" / f"{registry_id_hex}.json").read_text())
+
+
+def list_admin_registries() -> list[str]:
+    d = _home() / "admin"
+    if not d.exists():
+        return []
+    return sorted(p.stem for p in d.glob("*.json"))
+
+
+# ── Credentials ───────────────────────────────────────────────────────
+
+def save_credential(registry_id_hex: str, role_name: str, cred_data: dict):
+    d = _home() / "credentials"
+    d.mkdir(exist_ok=True)
+    (d / f"{registry_id_hex}_{role_name}.json").write_text(json.dumps(cred_data, indent=2))
+
+
+def load_credential_data(registry_id_hex: str, role_name: str) -> dict:
+    return json.loads(
+        (_home() / "credentials" / f"{registry_id_hex}_{role_name}.json").read_text()
+    )
+
+
+def has_credential(registry_id_hex: str, role_name: str) -> bool:
+    return (_home() / "credentials" / f"{registry_id_hex}_{role_name}.json").exists()
+
+
+def reconstruct_credential(cred_data: dict) -> zkac.Credential:
+    pk = zkac.BbsPublicKey.from_bytes(_unb64(cred_data["issuer_pk_b64"]))
+    return zkac.Credential.finalize(
+        _unb64(cred_data["blind_sig_b64"]),
+        _unb64(cred_data["member_secret_b64"]),
+        _unb64(cred_data["prover_blind_b64"]),
+        zkac.role_id(cred_data["role_name"]),
+        cred_data["epoch"],
+        pk,
+    )
+
+
+def list_credentials() -> list[tuple[str, str]]:
+    d = _home() / "credentials"
+    if not d.exists():
+        return []
+    result = []
+    for p in sorted(d.glob("*.json")):
+        parts = p.stem.rsplit("_", 1)
+        if len(parts) == 2:
+            result.append((parts[0], parts[1]))
+    return result
diff --git a/cli/zkac_node.egg-info/PKG-INFO b/cli/zkac_node.egg-info/PKG-INFO
new file mode 100644
index 0000000..43800ac
--- /dev/null
+++ b/cli/zkac_node.egg-info/PKG-INFO
@@ -0,0 +1,5 @@
+Metadata-Version: 2.4
+Name: zkac-node
+Version: 0.1.0
+Requires-Python: >=3.10
+Requires-Dist: zkac
diff --git a/cli/zkac_node.egg-info/SOURCES.txt b/cli/zkac_node.egg-info/SOURCES.txt
new file mode 100644
index 0000000..75468e6
--- /dev/null
+++ b/cli/zkac_node.egg-info/SOURCES.txt
@@ -0,0 +1,14 @@
+README.md
+pyproject.toml
+zkac_cli/__init__.py
+zkac_cli/client.py
+zkac_cli/main.py
+zkac_cli/paths.py
+zkac_cli/server.py
+zkac_cli/store.py
+zkac_node.egg-info/PKG-INFO
+zkac_node.egg-info/SOURCES.txt
+zkac_node.egg-info/dependency_links.txt
+zkac_node.egg-info/entry_points.txt
+zkac_node.egg-info/requires.txt
+zkac_node.egg-info/top_level.txt
\ No newline at end of file
diff --git a/cli/zkac_node.egg-info/dependency_links.txt b/cli/zkac_node.egg-info/dependency_links.txt
new file mode 100644
index 0000000..8b13789
--- /dev/null
+++ b/cli/zkac_node.egg-info/dependency_links.txt
@@ -0,0 +1 @@
+
diff --git a/cli/zkac_node.egg-info/entry_points.txt b/cli/zkac_node.egg-info/entry_points.txt
new file mode 100644
index 0000000..e0a11c4
--- /dev/null
+++ b/cli/zkac_node.egg-info/entry_points.txt
@@ -0,0 +1,2 @@
+[console_scripts]
+zkac-node = zkac_cli.main:main
diff --git a/cli/zkac_node.egg-info/requires.txt b/cli/zkac_node.egg-info/requires.txt
new file mode 100644
index 0000000..f2d352c
--- /dev/null
+++ b/cli/zkac_node.egg-info/requires.txt
@@ -0,0 +1 @@
+zkac
diff --git a/cli/zkac_node.egg-info/top_level.txt b/cli/zkac_node.egg-info/top_level.txt
new file mode 100644
index 0000000..03e77dd
--- /dev/null
+++ b/cli/zkac_node.egg-info/top_level.txt
@@ -0,0 +1 @@
+zkac_cli
diff --git a/docs/SECURITY.md b/docs/SECURITY.md
index 1cb3637..105b302 100644
--- a/docs/SECURITY.md
+++ b/docs/SECURITY.md
@@ -1,4 +1,4 @@
-# Security model and audit notes (ZKAC 0.2)
+# Security model and audit notes (ZKAC 0.3)
 
 This document summarizes the design, residual risks, and recommendations for operators integrating **ZKAC**. It is not a substitute for independent review before high-assurance deployment.
 
@@ -6,10 +6,11 @@ This document summarizes the design, residual risks, and recommendations for ope
 
 - **Authentication:** Only holders of a valid BBS+ credential for a registered role can complete `verify_auth` for that role.
 - **Server identity:** The server proves its long-term identity to the client via a Schnorr signature over the session transcript; clients verify against a pinned public key. This prevents MITM attacks without requiring TLS.
-- **Confidentiality & integrity:** Session payloads are authenticated-encrypted (ChaCha20-Poly1305) with a key derived from an ephemeral X25519 handshake.
+- **Confidentiality & integrity:** All traffic (management and authenticated sessions) is authenticated-encrypted (ChaCha20-Poly1305) with keys derived from an ephemeral X25519 handshake.
 - **Replay resistance:** Duplicate ciphertexts in a direction are rejected (sliding window + monotonic counter).
 - **Unlinkability (credential layer):** BBS+ presentations are unlinkable across sessions when the presentation header (the session transcript hash) differs; the verifier learns only the disclosed attributes (opaque `role_id`, epoch) and validity. Client anonymity is preserved: the client never reveals its long-term key during the handshake.
 - **Server cannot forge credentials:** The server stores only the issuer **public** key per role; forging requires the issuer secret key.
+- **Opaque server:** The server stores only cryptographically verified state blobs and opaque grant ciphertexts. No user identities, role names, or credential material are stored or visible to the server.
 
 ## Cryptographic components
 
@@ -19,9 +20,12 @@ This document summarizes the design, residual risks, and recommendations for ope
 | Identity | Schnorr on Ristretto255, BLAKE2b-512 challenge | Server identity binding |
 | Credentials | BBS+ on BLS12-381 (zkryptium), SHAKE256 ciphersuite | Blind issuance, ZK presentations |
 | Role IDs | BLAKE2b-512 (truncated to 32 bytes) | Opaque role identifiers |
+| Grant delivery | X25519 static/ephemeral DH, HKDF-SHA256, ChaCha20-Poly1305 | E2E-encrypted credential grants |
 
 ## Protocol flow
 
+### Unified channel (all connections)
+
 ```
 Client                                Server
   |--- init_msg (eph_pk) ------------>|
@@ -30,10 +34,26 @@ Client                                Server
   |<-- response_msg + identity_pkt ---|
   | complete DH                       |
   | decrypt + verify server sig       |
-  | encrypt BBS+ auth                 |
-  |--- encrypted BBS+ auth ---------> |
-  |                                   | verify_auth()
-  |===== encrypted session ===========>|
+  |===== encrypted session ==========>|
+  |--- {op: "mgmt"} or {op: "auth"}->|
+```
+
+Management commands (`create_registry`, `post_grant`, etc.) and BBS+ role authentication both run inside the same encrypted, server-authenticated channel. There is no unencrypted management path.
+
+### Grant delivery (admin → recipient, through server)
+
+```
+Admin                  Server (opaque relay)        Recipient
+  |-- post_grant ------->|                            |
+  |   (admin_proof,      | stores only:               |
+  |    recipient_pk,     |  {eph_pk, ciphertext}      |
+  |    eph_pk,           |  keyed by recipient_pk     |
+  |    ciphertext)       |                            |
+  |                      |<-- list_grants ------------|
+  |                      |--- [{eph_pk, ct}, ...] --->|
+  |                      |                            | trial-decrypt
+  |                      |<-- claim_grant ------------|
+  |                      |  (removes entry)           |
 ```
 
 ## Threats considered
@@ -41,20 +61,26 @@ Client                                Server
 ### Network attacker (passive)
 
 - Observes ciphertexts; cannot break ChaCha20-Poly1305 or derive session keys without breaking X25519 / HKDF under standard assumptions.
+- Management traffic is indistinguishable from auth traffic at the wire level (same handshake, same framing).
 
 ### Network attacker (active / MITM)
 
 - **Server impersonation:** The server signs the session transcript hash with its long-term Ristretto255 key (`prove_identity`). The client verifies this signature against the **pinned** server public key. A MITM running a separate DH exchange produces a different transcript; it cannot forge the server's signature. The client aborts on mismatch.
 - **Client impersonation:** The BBS+ presentation is bound to the session transcript hash. A MITM cannot relay a presentation from one session to another (different transcripts) or forge one (requires a valid credential from the issuer).
 - **Relay attack:** A MITM that relays the real server's identity proof to a client fails because the proof is encrypted under the MITM-to-server session keys (not the client-to-MITM keys), and the signature is over the wrong transcript.
+- **Management channel:** All management commands (registry creation, grants) are protected by the same encrypted channel, eliminating the previous plaintext management path.
 
 ### Malicious server
 
 - Can **learn** opaque `role_id`, current epoch, and that *some* valid member authenticated.
+- Sees `registry_id` values (needed for routing) but not role names or registry contents beyond opaque state bytes.
+- Sees `recipient_pk` for mailbox addressing, plus `eph_pk` and ciphertext per grant, but cannot decrypt grant payloads.
 - **Cannot** forge BBS+ credentials without the issuer secret key.
 - **Cannot** learn `member_secret` from presentations under the BBS+ security assumptions.
 - **Cannot** distinguish which specific member authenticated among valid credential holders (unlinkability holds against the verifier for distinct presentation headers).
-- **Cannot** learn the client's long-term public key — it is never transmitted.
+- **Cannot** learn the client's long-term public key — it is never transmitted during handshake or auth.
+- **Cannot** perform admin operations (registry updates, grant posting) without a valid admin BBS+ credential.
+- **Cannot** correlate a recipient's mailbox identity with their authenticated sessions (different keys, unlinkable proofs).
 
 ### Malicious client
 
@@ -73,7 +99,7 @@ The server's long-term `PublicKey` (32-byte Ristretto255 point) functions as a *
 
 Recommended strategies:
 
-1. **Static configuration** (default): embed the server public key in client config, environment variable, or CLI flag. Equivalent to WireGuard's `[Peer] PublicKey = ...`.
+1. **Static configuration** (default): embed the server public key in client config or CLI pin command (`zkac-node server pin`). Equivalent to WireGuard's `[Peer] PublicKey = ...`.
 2. **Trust On First Use (TOFU):** accept the server's key on first connection, pin it for subsequent sessions. Risk: first connection is vulnerable.
 3. **Out-of-band verification:** compare public key fingerprints over a trusted side channel (phone, in-person, encrypted messaging).
 4. **Key registry / directory:** a trusted service maps names to public keys. Shifts trust to the registry and its authentication channel.
@@ -81,28 +107,36 @@ Recommended strategies:
 ## Operational requirements
 
 1. **Issuer secret key:** Protect `BbsIssuer` secret material (HSM, KMS, or encrypted at rest). Compromise = ability to issue arbitrary credentials for that role.
-2. **Server long-term key:** Protect the `Node` `Keypair` secret. Compromise = ability to impersonate the server. Rotate the key and distribute the new public key to clients if compromised.
+2. **Server long-term key:** Protect the server's `server_key.json`. Compromise = ability to impersonate the server. Rotate the key and distribute the new public key to clients if compromised.
 3. **Member storage:** `member_secret` and finalized `Credential` material must be protected; loss = re-enrollment required.
 4. **Epoch revocation:** On compromise or policy change, call `set_epoch` and re-issue credentials only to legitimate members; old credentials become invalid at verification time.
-5. **Registry integrity:** The server's `(role_id → public key, epoch)` mapping must be integrity-protected (trusted storage or signed updates), or attackers could swap keys or epochs.
+5. **Registry integrity:** Registry state is integrity-protected by BBS+ state certificates (admin must sign updates). The server verifies these certificates before accepting changes.
 6. **Role ID privacy:** `role_id` is a hash of the role name only if you use `role_id("myrole")`; treat role names as secrets if enumeration is a concern, or derive role IDs with an additional secret salt known to members.
+7. **Client identity:** The only persistent client identifier is the issuance public key used for grant mailbox addressing. This key is shared out-of-band with admins; it is not linked to any transport key or BBS+ presentation.
 
 ## Implementation notes (audit checklist)
 
 - [x] BBS+ proof verification uses the same header and presentation binding as proof generation (`verify_presentation` in Rust).
 - [x] Session transcript is included in the presentation via `present(transcript_hash)`.
-- [x] Server identity proof: Schnorr signature over `transcript_hash`, verified against pinned public key before BBS+ auth proceeds.
+- [x] Server identity proof: Schnorr signature over `transcript_hash`, verified against pinned public key before any traffic.
 - [x] Schnorr nonce is deterministic (`H(sk || msg)`) — no dependence on RNG quality at signing time.
 - [x] Replay protection is symmetric per direction in `Session`.
 - [x] Constant-time comparisons are used where critical in transport/replay paths (`subtle` crate).
 - [x] Client long-term key is never transmitted, preserving BBS+ unlinkability.
+- [x] Management and auth channels use the same encrypted handshake (no plaintext management path).
+- [x] Admin proofs in management commands are bound to the session transcript hash (no separate nonce).
+- [x] Server stores only opaque state bytes, state certs, and encrypted grant blobs (no role names, no user IDs).
 - [ ] **External:** Python bindings surface raw bytes; callers must not log secrets (`secret_key_bytes`, `member_secret`, `prover_blind`).
 - [ ] **External:** Use secure randomness from the OS (library uses OS RNG for key generation paths exposed in Rust).
 
 ## Design decisions
 
+- **Unified encrypted channel:** All traffic (management and auth) uses the same anonymous handshake. This eliminates the attack surface of an unencrypted management path and simplifies the protocol to a single mode.
+- **Anonymous handshake (`complete_connect_anon`):** The client verifies the server's identity but does not authenticate itself during the handshake. BBS+ auth is sent as an application-layer message inside the encrypted session, not as part of the handshake. This allows the same channel for both anonymous management and authenticated role access.
 - **Server-only identity proof:** Only the server signs the transcript. Adding client long-term signing would break BBS+ unlinkability (the server could correlate sessions by client public key). Client authentication is handled entirely by the anonymous BBS+ credential.
 - **Deterministic Schnorr nonces:** The signing nonce is derived as `H("zkac-schnorr-nonce" || sk || msg)`, eliminating a class of RNG-failure attacks (cf. PS3 ECDSA, Sony 2010). Same key + same message = same signature.
+- **Opaque mailbox:** Grant entries on the server contain only `(eph_pk, ciphertext)` — no registry ID or role name. Recipients find their grants by trial-decrypting. This prevents the server from learning which registry or role a grant is for.
+- **No user IDs on server:** The server has no concept of user accounts. It is a stateless relay authenticated only by cryptographic proofs.
 
 ## Known limitations
 
@@ -110,6 +144,7 @@ Recommended strategies:
 - **Epoch granularity:** Revocation is coarse (epoch bump); plan issuance and rotation policy accordingly.
 - **zkryptium dependency:** Security follows the underlying crate and BLS12-381/BBS+ standards; keep dependencies updated.
 - **Key distribution:** The library provides the cryptographic mechanism; initial key distribution is an application-layer responsibility.
+- **Mailbox metadata:** The server sees `recipient_pk` as a mailbox address and the number/size/timing of grants. This is inherent to the delivery mechanism.
 
 ## Reporting issues
 
diff --git a/python/zkac/tcp.py b/python/zkac/tcp.py
index 1e79aa2..1cf4e67 100644
--- a/python/zkac/tcp.py
+++ b/python/zkac/tcp.py
@@ -157,6 +157,50 @@ def server_handshake_managed(
     return session, registry_id, role_id
 
 
+def client_handshake_anon(
+    sock: socket.socket,
+    node: Node,
+    expected_server_pk: PublicKey,
+) -> Session:
+    """
+    Anonymous client handshake: verify server identity only, no BBS+ auth.
+
+    Returns an encrypted :class:`Session` for management traffic.
+    """
+    pending, init_msg = node.connect()
+    write_frame(sock, init_msg)
+
+    bundle = read_frame(sock)
+    if len(bundle) < _HANDSHAKE_MSG_LEN:
+        raise ValueError("server handshake bundle too short")
+    response_msg = bundle[:_HANDSHAKE_MSG_LEN]
+    identity_proof = bundle[_HANDSHAKE_MSG_LEN:]
+
+    session = node.complete_connect_anon(
+        pending, response_msg, identity_proof, expected_server_pk,
+    )
+    return session
+
+
+def server_handshake_anon(
+    sock: socket.socket,
+    node: Node,
+) -> Session:
+    """
+    Server-side anonymous handshake: prove identity, no BBS+ verification.
+
+    Returns an encrypted :class:`Session`.
+    """
+    init_msg = read_frame(sock)
+    if len(init_msg) != _HANDSHAKE_MSG_LEN:
+        raise ValueError("init_msg must be 32 bytes")
+
+    session, response_msg = node.accept(init_msg)
+    identity_proof = node.prove_identity(session)
+    write_frame(sock, response_msg + identity_proof)
+    return session
+
+
 class FramedSession:
     """
     One ZKAC ciphertext per TCP frame: encrypt before send, decrypt after recv.
diff --git a/src/node.rs b/src/node.rs
index 40c5636..806e3a2 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -103,6 +103,40 @@ impl Node {
         Ok((session, encrypted_auth))
     }
 
+    /// Complete the connection verifying only the server's identity proof.
+    /// No BBS+ presentation is produced — the session is anonymous on the
+    /// client side.  Useful for management channels that still need an
+    /// encrypted, server-authenticated transport.
+    pub fn complete_connect_anon(
+        &self,
+        pending: PendingConnect,
+        response_msg: &[u8; HANDSHAKE_MSG_LEN],
+        identity_proof: &[u8],
+        expected_server_pk: &PublicKey,
+    ) -> Result<Session> {
+        let mut session = pending.handshake.complete(response_msg)?;
+
+        let id_payload = session.decrypt(identity_proof)?;
+        if id_payload.len() != IDENTITY_PROOF_LEN {
+            return Err(Error::IdentityVerificationFailed("identity proof wrong length"));
+        }
+
+        let mut pk_bytes = [0u8; 32];
+        pk_bytes.copy_from_slice(&id_payload[..32]);
+        let server_pk = PublicKey::from_bytes(pk_bytes)?;
+
+        if server_pk != *expected_server_pk {
+            return Err(Error::IdentityVerificationFailed("server public key mismatch"));
+        }
+
+        let mut sig_bytes = [0u8; SIGNATURE_LEN];
+        sig_bytes.copy_from_slice(&id_payload[32..]);
+        let sig = Signature::from_bytes(&sig_bytes)?;
+        server_pk.verify(session.transcript_hash(), &sig)?;
+
+        Ok(session)
+    }
+
     // ── Responder (server) side ──────────────────────────────────────
 
     /// Accept an incoming handshake initiation.
diff --git a/src/python.rs b/src/python.rs
index 573fbc2..116db0d 100644
--- a/src/python.rs
+++ b/src/python.rs
@@ -743,6 +743,37 @@ impl PyNode {
         Ok(PyBytes::new(py, &rid))
     }
 
+    /// Complete handshake verifying server identity only (no BBS+ auth).
+    fn complete_connect_anon(
+        &self,
+        pending: &mut PyPendingConnect,
+        response_msg: &[u8],
+        identity_proof: &[u8],
+        expected_server_pk: &PyPublicKey,
+    ) -> PyResult<PySession> {
+        let p = pending
+            .inner
+            .take()
+            .ok_or_else(|| PyValueError::new_err("PendingConnect already consumed"))?;
+
+        if response_msg.len() != HANDSHAKE_MSG_LEN {
+            return Err(PyValueError::new_err("response_msg must be 32 bytes"));
+        }
+        let msg: [u8; HANDSHAKE_MSG_LEN] = response_msg.try_into().unwrap();
+
+        let session = self
+            .inner
+            .complete_connect_anon(
+                p,
+                &msg,
+                identity_proof,
+                &expected_server_pk.inner,
+            )
+            .map_err(to_py_err)?;
+
+        Ok(PySession { inner: session })
+    }
+
     /// Complete handshake for a client-managed registry. Includes
     /// registry_id in the auth packet.
     fn complete_connect_managed(
diff --git a/tests/test_zkac_tcp.py b/tests/test_zkac_tcp.py
index 1ca7900..79022e3 100644
--- a/tests/test_zkac_tcp.py
+++ b/tests/test_zkac_tcp.py
@@ -8,8 +8,10 @@ from zkac.tcp import (
     FramedSession,
     MAX_TCP_FRAME_BYTES,
     client_handshake,
+    client_handshake_anon,
     read_frame,
     server_handshake,
+    server_handshake_anon,
     write_frame,
 )
 
@@ -81,6 +83,62 @@ class TestHandshakeOverTcp:
             assert not t.is_alive()
 
 
+class TestAnonHandshake:
+    def test_anon_handshake_no_bbs(self):
+        """Anonymous handshake: server identity verified, no BBS+ auth."""
+        client_sock, server_sock = socket.socketpair()
+        server_kp = zkac.Keypair()
+        server_pk = server_kp.public_key()
+
+        def run_server():
+            try:
+                srv = zkac.Node(server_kp)
+                session = server_handshake_anon(server_sock, srv)
+                framed = FramedSession(server_sock, session)
+                msg = framed.recv()
+                framed.send(b"echo:" + msg)
+            finally:
+                server_sock.close()
+
+        t = threading.Thread(target=run_server)
+        t.start()
+        try:
+            cli = zkac.Node(zkac.Keypair())
+            session = client_handshake_anon(client_sock, cli, server_pk)
+            framed = FramedSession(client_sock, session)
+            framed.send(b"hello")
+            assert framed.recv() == b"echo:hello"
+        finally:
+            client_sock.close()
+            t.join(timeout=5)
+            assert not t.is_alive()
+
+    def test_anon_wrong_server_key_rejected(self):
+        """Anonymous handshake rejects wrong pinned server key."""
+        client_sock, server_sock = socket.socketpair()
+        server_kp = zkac.Keypair()
+        wrong_pk = zkac.Keypair().public_key()
+
+        def run_server():
+            try:
+                srv = zkac.Node(server_kp)
+                server_handshake_anon(server_sock, srv)
+            except Exception:
+                pass
+            finally:
+                server_sock.close()
+
+        t = threading.Thread(target=run_server)
+        t.start()
+        try:
+            cli = zkac.Node(zkac.Keypair())
+            with pytest.raises(ValueError):
+                client_handshake_anon(client_sock, cli, wrong_pk)
+        finally:
+            client_sock.close()
+            t.join(timeout=5)
+
+
 class TestFramedSession:
     def test_framed_encrypt_roundtrip(self):
         _, pk, rid, cred = _make_credential()