feat(cli+broker): three-tier peer removal: disconnect, kick, ban
Broker (apps/broker/src/index.ts)
- Unified disconnect/kick handler uses close code 1000 for disconnect
(CLI auto-reconnects) vs 4001 for kick (CLI exits, no reconnect).
- Ban now closes with code 4002.
- Hello handler: revoked members get a specific 'revoked' error with a
'Contact the mesh owner to rejoin' message, then ws.close(4002).
Previously banned users saw the generic 'unauthorized' error.
- list_bans handler returns { name, pubkey, revokedAt } for each
revoked member.
CLI (apps/cli)
- ws-client: close codes 4001 and 4002 set .closed = true and stash
.terminalClose so callers can surface a friendly message instead of
the low-level 'ws terminal close' error. Revoked error in hello is
also captured as a terminal close.
- withMesh catches terminalClose and prints:
4001 → 'Kicked from this mesh. Run claudemesh to rejoin.'
4002 → the broker's 'Contact the mesh owner to rejoin.' message
- kick.ts now exports runDisconnect + runKick with clear hints:
'disconnect' → 'They will auto-reconnect within seconds.'
'kick' → 'They can rejoin anytime by running claudemesh.'
- cli.ts adds 'disconnect' dispatch; HELP updated.
Semantics:
disconnect: session reset, no DB state, auto-reconnects
kick : session ends, no DB state, user must manually rejoin
ban : session ends + revokedAt set, cannot rejoin until unban
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alejandro Gutiérrez
@@ -1669,6 +1669,26 @@ async function handleHello(
|
||||
}
|
||||
const member = await findMemberByPubkey(hello.meshId, hello.pubkey);
|
||||
if (!member) {
|
||||
// Distinguish "revoked" from "never a member" so banned users get
|
||||
// a clear message ("contact admin") instead of generic unauthorized.
|
||||
const [revokedRow] = await db
|
||||
.select({ displayName: meshMember.displayName, revokedAt: meshMember.revokedAt })
|
||||
.from(meshMember)
|
||||
.where(and(eq(meshMember.meshId, hello.meshId), eq(meshMember.peerPubkey, hello.pubkey)))
|
||||
.limit(1);
|
||||
if (revokedRow?.revokedAt) {
|
||||
metrics.connectionsRejected.inc({ reason: "revoked" });
|
||||
const [m] = await db.select({ slug: mesh.slug, name: mesh.name }).from(mesh).where(eq(mesh.id, hello.meshId)).limit(1);
|
||||
const meshLabel = m?.name || m?.slug || hello.meshId;
|
||||
sendError(
|
||||
ws,
|
||||
"revoked",
|
||||
`You've been removed from "${meshLabel}". Contact the mesh owner to rejoin.`,
|
||||
);
|
||||
ws.close(4002, "banned");
|
||||
log.info("hello rejected: revoked", { mesh_id: hello.meshId, display_name: revokedRow.displayName });
|
||||
return null;
|
||||
}
|
||||
metrics.connectionsRejected.inc({ reason: "unauthorized" });
|
||||
sendError(ws, "unauthorized", "pubkey not found in mesh");
|
||||
ws.close(1008, "unauthorized");
|
||||
@@ -3900,57 +3920,63 @@ function handleConnection(ws: WebSocket): void {
|
||||
|
||||
// --- Kick / Ban / Unban ---
|
||||
|
||||
case "disconnect":
|
||||
case "kick": {
|
||||
const km = msg as { type: "kick"; target?: string; stale?: number; all?: boolean; _reqId?: string };
|
||||
// Authz: only owner or admin can kick.
|
||||
// disconnect: soft — WS closes with 1000, CLI auto-reconnects.
|
||||
// kick: hard — WS closes with 4001, CLI exits (no reconnect).
|
||||
// Same target semantics (<name> | --stale <ms> | --all). Only
|
||||
// the close code differs.
|
||||
const isKick = msg.type === "kick";
|
||||
const km = msg as { type: "kick" | "disconnect"; target?: string; stale?: number; all?: boolean; _reqId?: string };
|
||||
const closeCode = isKick ? 4001 : 1000;
|
||||
const closeReason = isKick ? "kicked" : "disconnected";
|
||||
const ackType = isKick ? "kick_ack" : "disconnect_ack";
|
||||
|
||||
// Authz: only owner or admin.
|
||||
const [kickMesh] = await db.select({ ownerUserId: mesh.ownerUserId }).from(mesh).where(eq(mesh.id, conn.meshId)).limit(1);
|
||||
const [kickMember] = await db.select({ role: meshMember.role, userId: meshMember.userId }).from(meshMember).where(eq(meshMember.id, conn.memberId)).limit(1);
|
||||
if (!kickMesh || (kickMesh.ownerUserId !== kickMember?.userId && kickMember?.role !== "admin")) {
|
||||
sendError(ws, "forbidden", "only owner or admin can kick", undefined, km._reqId);
|
||||
sendError(ws, "forbidden", `only owner or admin can ${closeReason}`, undefined, km._reqId);
|
||||
break;
|
||||
}
|
||||
|
||||
const kicked: string[] = [];
|
||||
const affected: string[] = [];
|
||||
const now = Date.now();
|
||||
|
||||
if (km.all) {
|
||||
// Kick everyone except caller
|
||||
for (const [pid, peer] of connections) {
|
||||
if (peer.meshId !== conn.meshId || pid === presenceId) continue;
|
||||
try { peer.ws.close(1000, "kicked"); } catch {}
|
||||
try { peer.ws.close(closeCode, closeReason); } catch {}
|
||||
connections.delete(pid);
|
||||
void disconnectPresence(pid);
|
||||
kicked.push(peer.displayName || pid);
|
||||
affected.push(peer.displayName || pid);
|
||||
}
|
||||
} else if (km.stale && typeof km.stale === "number") {
|
||||
// Kick peers idle longer than stale ms
|
||||
const cutoff = now - km.stale;
|
||||
for (const [pid, peer] of connections) {
|
||||
if (peer.meshId !== conn.meshId || pid === presenceId) continue;
|
||||
// Check last_ping_at from DB for accurate staleness
|
||||
const [pres] = await db.select({ lastPingAt: presence.lastPingAt }).from(presence).where(eq(presence.id, pid)).limit(1);
|
||||
if (pres && pres.lastPingAt && pres.lastPingAt.getTime() < cutoff) {
|
||||
try { peer.ws.close(1000, "kicked_stale"); } catch {}
|
||||
try { peer.ws.close(closeCode, `${closeReason}_stale`); } catch {}
|
||||
connections.delete(pid);
|
||||
void disconnectPresence(pid);
|
||||
kicked.push(peer.displayName || pid);
|
||||
affected.push(peer.displayName || pid);
|
||||
}
|
||||
}
|
||||
} else if (km.target) {
|
||||
// Kick specific peer by name or pubkey
|
||||
for (const [pid, peer] of connections) {
|
||||
if (peer.meshId !== conn.meshId) continue;
|
||||
if (peer.displayName === km.target || peer.memberPubkey === km.target || peer.memberPubkey.startsWith(km.target)) {
|
||||
try { peer.ws.close(1000, "kicked"); } catch {}
|
||||
try { peer.ws.close(closeCode, closeReason); } catch {}
|
||||
connections.delete(pid);
|
||||
void disconnectPresence(pid);
|
||||
kicked.push(peer.displayName || pid);
|
||||
affected.push(peer.displayName || pid);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
conn.ws.send(JSON.stringify({ type: "kick_ack", kicked, _reqId: km._reqId }));
|
||||
log.info("ws kick", { presence_id: presenceId, kicked_count: kicked.length, target: km.target ?? km.stale ?? "all" });
|
||||
conn.ws.send(JSON.stringify({ type: ackType, kicked: affected, affected, _reqId: km._reqId }));
|
||||
log.info(`ws ${closeReason}`, { presence_id: presenceId, count: affected.length, target: km.target ?? km.stale ?? "all" });
|
||||
break;
|
||||
}
|
||||
|
||||
@@ -3985,7 +4011,7 @@ function handleConnection(ws: WebSocket): void {
|
||||
// Kick all their connections
|
||||
for (const [pid, peer] of connections) {
|
||||
if (peer.meshId === conn.meshId && peer.memberPubkey === targetMember.peerPubkey) {
|
||||
try { peer.ws.close(1000, "banned"); } catch {}
|
||||
try { peer.ws.close(4002, "banned"); } catch {}
|
||||
connections.delete(pid);
|
||||
void disconnectPresence(pid);
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "claudemesh-cli",
|
||||
"version": "1.0.0",
|
||||
"version": "1.0.0-alpha.43",
|
||||
"description": "Peer mesh for Claude Code sessions — CLI + MCP server.",
|
||||
"keywords": [
|
||||
"claude-code",
|
||||
|
||||
@@ -75,6 +75,21 @@ export async function withMesh<T>(
|
||||
await client.connect();
|
||||
const result = await fn(client, mesh);
|
||||
return result;
|
||||
} catch (e) {
|
||||
// Terminal close from the broker (banned / kicked). Give the user
|
||||
// a clear message instead of the low-level ws error.
|
||||
if (client.terminalClose) {
|
||||
const { code, reason } = client.terminalClose;
|
||||
if (code === 4002) {
|
||||
console.error(`\n ✘ ${reason}\n`);
|
||||
} else if (code === 4001) {
|
||||
console.error(`\n ✘ Kicked from this mesh. Run \`claudemesh\` to rejoin.\n`);
|
||||
} else {
|
||||
console.error(`\n ✘ Broker closed connection: ${reason}\n`);
|
||||
}
|
||||
process.exit(1);
|
||||
}
|
||||
throw e;
|
||||
} finally {
|
||||
client.close();
|
||||
}
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
/**
|
||||
* `claudemesh kick` — disconnect peers from the mesh.
|
||||
* `claudemesh disconnect` — soft disconnect (session reset, auto-reconnects).
|
||||
* `claudemesh kick` — hard kick (session ends, no auto-reconnect).
|
||||
*
|
||||
* claudemesh kick <name> kick one peer (can reconnect)
|
||||
* claudemesh kick --stale 30m kick idle peers (> 30 min no activity)
|
||||
* claudemesh kick --all kick everyone except yourself
|
||||
* claudemesh disconnect <peer> # nudge, reconnects in seconds
|
||||
* claudemesh kick <peer> # stop session, user runs claudemesh to rejoin
|
||||
* claudemesh kick --stale 30m # kick peers idle > 30m
|
||||
* claudemesh kick --all # kick everyone except yourself
|
||||
*
|
||||
* Ban (permanent, revokes membership) is in ban.ts.
|
||||
*/
|
||||
|
||||
import { withMesh } from "./connect.js";
|
||||
@@ -22,6 +26,44 @@ function parseStaleMs(input: string): number | null {
|
||||
return null;
|
||||
}
|
||||
|
||||
function buildPayload(
|
||||
kind: "disconnect" | "kick",
|
||||
target: string | undefined,
|
||||
opts: { stale?: string; all?: boolean },
|
||||
): Record<string, unknown> | { error: string } {
|
||||
if (opts.all) return { type: kind, all: true };
|
||||
if (opts.stale) {
|
||||
const ms = parseStaleMs(opts.stale);
|
||||
if (!ms) return { error: `Invalid stale duration: "${opts.stale}". Use e.g. 30m, 1h, 300s.` };
|
||||
return { type: kind, stale: ms };
|
||||
}
|
||||
if (target) return { type: kind, target };
|
||||
return { error: `Usage: claudemesh ${kind} <peer> | --stale 30m | --all` };
|
||||
}
|
||||
|
||||
export async function runDisconnect(
|
||||
target: string | undefined,
|
||||
opts: { mesh?: string; stale?: string; all?: boolean } = {},
|
||||
): Promise<number> {
|
||||
const config = readConfig();
|
||||
const meshSlug = opts.mesh ?? config.meshes[0]?.slug;
|
||||
if (!meshSlug) { render.err("No mesh joined."); return EXIT.NOT_FOUND; }
|
||||
|
||||
const built = buildPayload("disconnect", target, opts);
|
||||
if ("error" in built) { render.err(String(built.error)); return EXIT.INVALID_ARGS; }
|
||||
|
||||
return await withMesh({ meshSlug }, async (client) => {
|
||||
const result = await client.sendAndWait(built as Record<string, unknown>) as { affected?: string[]; kicked?: string[] };
|
||||
const peers = result?.affected ?? result?.kicked ?? [];
|
||||
if (peers.length === 0) render.info("No peers matched.");
|
||||
else {
|
||||
render.ok(`Disconnected ${peers.length} peer(s): ${peers.join(", ")}`);
|
||||
render.hint("They will auto-reconnect within seconds. For a session-ending kick, use `claudemesh kick`.");
|
||||
}
|
||||
return EXIT.SUCCESS;
|
||||
});
|
||||
}
|
||||
|
||||
export async function runKick(
|
||||
target: string | undefined,
|
||||
opts: { mesh?: string; stale?: string; all?: boolean } = {},
|
||||
@@ -30,29 +72,16 @@ export async function runKick(
|
||||
const meshSlug = opts.mesh ?? config.meshes[0]?.slug;
|
||||
if (!meshSlug) { render.err("No mesh joined."); return EXIT.NOT_FOUND; }
|
||||
|
||||
const built = buildPayload("kick", target, opts);
|
||||
if ("error" in built) { render.err(String(built.error)); return EXIT.INVALID_ARGS; }
|
||||
|
||||
return await withMesh({ meshSlug }, async (client) => {
|
||||
let payload: Record<string, unknown>;
|
||||
|
||||
if (opts.all) {
|
||||
payload = { type: "kick", all: true };
|
||||
} else if (opts.stale) {
|
||||
const ms = parseStaleMs(opts.stale);
|
||||
if (!ms) { render.err(`Invalid stale duration: "${opts.stale}". Use e.g. 30m, 1h, 300s.`); return EXIT.INVALID_ARGS; }
|
||||
payload = { type: "kick", stale: ms };
|
||||
} else if (target) {
|
||||
payload = { type: "kick", target };
|
||||
} else {
|
||||
render.err("Usage: claudemesh kick <peer> | --stale 30m | --all");
|
||||
return EXIT.INVALID_ARGS;
|
||||
}
|
||||
|
||||
const result = await client.sendAndWait(payload) as { kicked?: string[] };
|
||||
const kicked = result?.kicked ?? [];
|
||||
|
||||
if (kicked.length === 0) {
|
||||
render.info("No peers matched.");
|
||||
} else {
|
||||
render.ok(`Kicked ${kicked.length} peer(s): ${kicked.join(", ")}`);
|
||||
const result = await client.sendAndWait(built as Record<string, unknown>) as { affected?: string[]; kicked?: string[] };
|
||||
const peers = result?.affected ?? result?.kicked ?? [];
|
||||
if (peers.length === 0) render.info("No peers matched.");
|
||||
else {
|
||||
render.ok(`Kicked ${peers.length} peer(s): ${peers.join(", ")}`);
|
||||
render.hint("Their Claude Code session ended. They can rejoin anytime by running `claudemesh`.");
|
||||
}
|
||||
return EXIT.SUCCESS;
|
||||
});
|
||||
|
||||
@@ -30,9 +30,10 @@ Mesh
|
||||
claudemesh delete [slug] delete a mesh (alias: rm)
|
||||
claudemesh rename <slug> <name> rename a mesh
|
||||
claudemesh share [email] share mesh (invite link / send email)
|
||||
claudemesh kick <peer> disconnect a peer (can reconnect)
|
||||
claudemesh kick --stale 30m disconnect idle peers (> duration)
|
||||
claudemesh kick --all disconnect everyone except you
|
||||
claudemesh disconnect <peer> soft disconnect (peer auto-reconnects)
|
||||
claudemesh kick <peer> end session (peer must manually rejoin)
|
||||
claudemesh kick --stale 30m kick peers idle > duration
|
||||
claudemesh kick --all kick everyone except yourself
|
||||
claudemesh ban <peer> kick + permanently revoke (can't rejoin)
|
||||
claudemesh unban <peer> lift a ban
|
||||
claudemesh bans list banned members
|
||||
@@ -139,6 +140,7 @@ async function main(): Promise<void> {
|
||||
case "delete": case "rm": { const { deleteMesh } = await import("~/commands/delete-mesh.js"); process.exit(await deleteMesh(positionals[0] ?? "", { yes: !!flags.y || !!flags.yes })); break; }
|
||||
case "rename": { const { rename } = await import("~/commands/rename.js"); process.exit(await rename(positionals[0] ?? "", positionals[1] ?? "")); break; }
|
||||
case "share": case "invite": { const { invite } = await import("~/commands/invite.js"); process.exit(await invite(positionals[0], { mesh: flags.mesh as string, json: !!flags.json })); break; }
|
||||
case "disconnect": { const { runDisconnect } = await import("~/commands/kick.js"); process.exit(await runDisconnect(positionals[0], { mesh: flags.mesh as string, stale: flags.stale as string, all: !!flags.all })); break; }
|
||||
case "kick": { const { runKick } = await import("~/commands/kick.js"); process.exit(await runKick(positionals[0], { mesh: flags.mesh as string, stale: flags.stale as string, all: !!flags.all })); break; }
|
||||
case "ban": { const { runBan } = await import("~/commands/ban.js"); process.exit(await runBan(positionals[0], { mesh: flags.mesh as string })); break; }
|
||||
case "unban": { const { runUnban } = await import("~/commands/ban.js"); process.exit(await runUnban(positionals[0], { mesh: flags.mesh as string })); break; }
|
||||
|
||||
@@ -166,6 +166,8 @@ export class BrokerClient {
|
||||
private _serviceCatalog: Array<{ name: string; description: string; status: string; tools: Array<{ name: string; description: string; inputSchema: object }>; deployed_by: string }> = [];
|
||||
get serviceCatalog() { return this._serviceCatalog; }
|
||||
private closed = false;
|
||||
/** Non-null when the broker closed us with a terminal code (4001/4002). */
|
||||
public terminalClose: { code: number; reason: string } | null = null;
|
||||
private reconnectAttempt = 0;
|
||||
private helloTimer: NodeJS.Timeout | null = null;
|
||||
private reconnectTimer: NodeJS.Timeout | null = null;
|
||||
@@ -321,10 +323,23 @@ export class BrokerClient {
|
||||
this.handleServerMessage(msg);
|
||||
};
|
||||
|
||||
const onClose = (): void => {
|
||||
const onClose = (code?: number, reasonBuf?: Buffer): void => {
|
||||
if (this.helloTimer) clearTimeout(this.helloTimer);
|
||||
this.helloTimer = null;
|
||||
if (this.ws === ws) this.ws = null;
|
||||
const reason = reasonBuf?.toString("utf-8") ?? "";
|
||||
// Terminal close codes — broker told us to stay gone.
|
||||
// 4001 = kicked (session ended, user must manually rejoin)
|
||||
// 4002 = banned (member revoked, cannot rejoin until unbanned)
|
||||
if (code === 4001 || code === 4002) {
|
||||
this.closed = true;
|
||||
this.setConnStatus("closed");
|
||||
this.terminalClose = { code, reason };
|
||||
if (this._status !== "open") {
|
||||
reject(new Error(`ws terminal close ${code}: ${reason || "session ended"}`));
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (this._status !== "open" && this._status !== "reconnecting") {
|
||||
reject(new Error("ws closed before hello_ack"));
|
||||
}
|
||||
@@ -2158,6 +2173,11 @@ export class BrokerClient {
|
||||
}
|
||||
if (msg.type === "error") {
|
||||
this.debug(`broker error: ${msg.code} ${msg.message}`);
|
||||
// Terminal errors from hello — broker will close us next. Capture
|
||||
// so the caller (launch/peers/etc.) can surface a friendly message.
|
||||
if (msg.code === "revoked") {
|
||||
this.terminalClose = { code: 4002, reason: String(msg.message ?? "revoked") };
|
||||
}
|
||||
const id = msg.id ? String(msg.id) : null;
|
||||
let handledByPendingSend = false;
|
||||
if (id) {
|
||||
|
||||
Reference in New Issue
Block a user