Refactor server for user isolation

CONTRIBUTING.md

@@ -47,6 +47,10 @@ Keep boundaries clear:
 ### Connection Tracking
 The Server keeps active Valkey clients in an in-memory Client_Map maintained by the Connection_Manager (`apps/server/src/connection.ts`). Each Client_Map entry is keyed by `buildConnectionId(host, port, db)` exported from `common/src/connection-id.ts`. Do not reconstruct this identifier elsewhere.
 
+All per-connection server state is scoped by **owner identity**, the Valkey identity the connection authenticated as (`ownerOf` in `apps/server/src/owner.ts`). Each owner has its own `OwnerState`: clients, cluster topology, collector credentials, collectors and their keys. CPU baselines, config sessions and node watchers carry the owner key too. Handlers get the requesting owner's maps through `Deps`. Do not read the module-level maps in `metrics-orchestrator.ts` from a user path: they are the env owner's state (the `VALKEY_*` identity).
+
+A WebSocket may act only on connections it opened with credentials Valkey accepted. `resolveAction` in `apps/server/src/socket-binding.ts` resolves every action to its owner or rejects it. Every connect first proves the credentials with a short-lived client (`createAuthProbeClient`), under the failed-AUTH limit in `apps/server/src/auth-limiter.ts`.
+
 ### Per-Node Config Fan-Out & Reply Contract
 Monitor-configuration updates fan out to every cluster node and collect a per-node outcome rather than treating the cluster as all-or-nothing. When adding or changing this flow:
 

apps/server/src/__integration__/README.md

@@ -4,6 +4,8 @@ End-to-end tests that drive the server's WebSocket API against a live Valkey sta
 
 ## Quick run
 
+With Podman, use `podman compose` in place of `docker compose`. On Apple silicon, set `DOCKER_PLATFORM=linux/arm64`.
+
 ```bash
 docker compose -f docker/docker-compose.test.yml up -d --build --wait
 npm run test:integration
@@ -13,7 +15,8 @@ docker compose -f docker/docker-compose.test.yml down -v
 ## Stack
 
 - **Cluster:** `valkey-7001..7006` on ports `7001..7006`. Initialized by the `cluster-init` service and seeded by `populate`. Used by `cluster-topology`, `connection`, `key-browser`, `monitoring`, and `send-command` integration tests via `defaultConnectionDetails()`.
-- **Standalone:** `valkey-standalone` on port `6379`. No ACL, no cluster mode. Used by the "two databases" test in `connection.integration.test.ts` via `defaultStandaloneConnectionDetails(db)` to verify that `(host, port, db)` triples produce isolated clients.
+- **Standalone:** `valkey-standalone` on port `6379`. No cluster mode. Used by the "two databases" test in `connection.integration.test.ts` via `defaultStandaloneConnectionDetails(db)` to verify that `(host, port, db)` triples produce isolated clients.
+- **ACL users:** Every node has the users `appuser`/`admin` (all commands) and `restricted`/`restricted` (all commands except `ACL` and `COMMANDLOG`). The standalone node also keeps the passwordless `default` user. `cluster_init.sh` sets the cluster users on every run, and the standalone node gets them from its `--user` arguments. `identity-isolation.integration.test.ts` uses the two users to check that users who connect as different Valkey users share no client and no metrics collector.
 
 ## Note
 

apps/server/src/__integration__/harness/fixture.ts

@@ -26,3 +26,23 @@ export const defaultStandaloneConnectionDetails = (db: number = 0) => {
     db,
   }
 }
+
+/**
+ * A second ACL user with every command except `ACL` and `COMMANDLOG`. Created
+ * on the cluster nodes by `cluster_init.sh` and on the standalone node by its
+ * `--user` arguments in `docker/docker-compose.test.yml`.
+ */
+export const RESTRICTED_USER = { username: "restricted", password: "restricted" }
+
+export const restrictedClusterConnectionDetails = () => ({
+  ...defaultConnectionDetails(),
+  ...RESTRICTED_USER,
+})
+
+export const standaloneConnectionDetailsAs = (
+  user: { username: string; password: string },
+  db: number = 0,
+) => ({
+  ...defaultStandaloneConnectionDetails(db),
+  ...user,
+})

apps/server/src/__integration__/identity-isolation.integration.test.ts

@@ -0,0 +1,189 @@
+import { after, describe, it } from "node:test"
+import assert from "node:assert/strict"
+import { buildConnectionId, COMMANDLOG_TYPE, MONITOR_ACTION, toNodeId, VALKEY } from "valkey-common"
+import { WsClient, type InboundMessage } from "./harness/wsClient"
+import {
+  defaultConnectionDetails,
+  defaultStandaloneConnectionDetails,
+  RESTRICTED_USER,
+  restrictedClusterConnectionDetails,
+  standaloneConnectionDetailsAs,
+  WS_URL
+} from "./harness/fixture"
+
+/**
+ * Per-identity isolation (Phase 1 of discussion #580).
+ *
+ * Each WsClient is one browser socket. `admin` connects as `appuser` (all
+ * commands). `limited` connects as `restricted` (no ACL, no COMMANDLOG). The
+ * two users must never share a Valkey client or a metrics collector, and a
+ * socket may act only on connections it opened itself.
+ *
+ * A rejected action produces no reply at all: the guard returns without
+ * sending. So the absence of a reply within a window is the assertion.
+ */
+describe("integration / identity isolation", async () => {
+  const admin = await WsClient.connect(WS_URL)
+  const limited = await WsClient.connect(WS_URL)
+  const stranger = await WsClient.connect(WS_URL)
+
+  const clusterAdmin = defaultConnectionDetails()
+  const clusterId0 = buildConnectionId(clusterAdmin.host, clusterAdmin.port, 0)
+  const standaloneAdmin = standaloneConnectionDetailsAs({ username: "appuser", password: "admin" })
+  const standaloneLimited = standaloneConnectionDetailsAs(RESTRICTED_USER)
+  const standaloneId = buildConnectionId(standaloneAdmin.host, standaloneAdmin.port, 0)
+
+  let clusterId: string | undefined
+
+  const run = async (ws: WsClient, connectionId: string, command: string): Promise<InboundMessage> => {
+    ws.send({ type: VALKEY.COMMAND.sendRequested, payload: { connectionId, command } })
+    const reply = await Promise.race([
+      ws.waitFor(VALKEY.COMMAND.sendFulfilled, 10000),
+      ws.waitFor(VALKEY.COMMAND.sendFailed, 10000),
+    ])
+    return reply
+  }
+
+  const silentFor = async (ws: WsClient, types: string[], windowMs = 3000) => {
+    const results = await Promise.all(types.map((type, i) => ws.collectFor(type, i === 0 ? windowMs : 100)))
+    return results.reduce((n, r) => n + r.length, 0)
+  }
+
+  after(async () => {
+    await admin.close()
+    await limited.close()
+    await stranger.close()
+  })
+
+  it("connects two Valkey users to the same cluster node and the same standalone node", async () => {
+    admin.send({ type: VALKEY.CONNECTION.connectPending, payload: { connectionId: clusterId0, connectionDetails: clusterAdmin } })
+    const adminCluster = await admin.waitFor(VALKEY.CONNECTION.clusterConnectFulfilled, 30000)
+    clusterId = adminCluster.payload?.connectionDetails?.clusterId
+    assert.ok(clusterId)
+
+    limited.send({
+      type: VALKEY.CONNECTION.connectPending,
+      payload: { connectionId: clusterId0, connectionDetails: restrictedClusterConnectionDetails() },
+    })
+    await limited.waitFor(VALKEY.CONNECTION.clusterConnectFulfilled, 30000)
+
+    admin.send({ type: VALKEY.CONNECTION.connectPending, payload: { connectionId: standaloneId, connectionDetails: standaloneAdmin } })
+    await admin.waitFor(VALKEY.CONNECTION.standaloneConnectFulfilled, 30000)
+    limited.send({ type: VALKEY.CONNECTION.connectPending, payload: { connectionId: standaloneId, connectionDetails: standaloneLimited } })
+    await limited.waitFor(VALKEY.CONNECTION.standaloneConnectFulfilled, 30000)
+  })
+
+  it("runs each user's commands as that user (the reported ACL LIST bug)", async () => {
+    for (const connectionId of [clusterId0, standaloneId]) {
+      const adminWho = await run(admin, connectionId, "ACL WHOAMI")
+      assert.equal(adminWho.payload, "appuser", `admin on ${connectionId}`)
+      const limitedWho = await run(limited, connectionId, "ACL WHOAMI")
+      assert.equal(limitedWho.type, VALKEY.COMMAND.sendFailed, "restricted has no ACL permission")
+      assert.match(JSON.stringify(limitedWho.payload), /NOPERM|no permissions/i)
+      assert.doesNotMatch(JSON.stringify(limitedWho.payload), /appuser/)
+
+      const adminList = await run(admin, connectionId, "ACL LIST")
+      assert.equal(adminList.type, VALKEY.COMMAND.sendFulfilled, `admin ACL LIST on ${connectionId}`)
+    }
+  })
+
+  it("gives each user a metrics collector that runs as that user", async () => {
+    // Collectors register a moment after connect; poll CLIENT LIST for both users.
+    let users = new Set<string>()
+    for (let attempt = 0; attempt < 30 && users.size < 2; attempt++) {
+      const list = await run(admin, standaloneId, "CLIENT LIST")
+      // The server parses `key:value` replies into an object; rebuild the lines.
+      const text = typeof list.payload === "string"
+        ? list.payload
+        : Object.entries(list.payload ?? {}).map(([k, v]) => `${k}:${String(v)}`).join("\n")
+      users = new Set(
+        text
+          .split("\n")
+          .filter((line) => line.includes("valkey_admin_metrics"))
+          .map((line) => / user=(\S+)/.exec(line)?.[1] ?? ""),
+      )
+      if (users.size < 2) await new Promise((r) => setTimeout(r, 1000))
+    }
+    assert.ok(users.has("appuser"), `appuser collector missing: ${[...users].join(",")}`)
+    assert.ok(users.has(RESTRICTED_USER.username), `restricted collector missing: ${[...users].join(",")}`)
+  })
+
+  it("rejects a wrong password for an existing owner and keeps that owner's socket working", async () => {
+    stranger.send({
+      type: VALKEY.CONNECTION.connectPending,
+      payload: { connectionId: standaloneId, connectionDetails: { ...standaloneAdmin, password: "wrong" } },
+    })
+    const rejected = await stranger.waitFor(VALKEY.CONNECTION.connectRejected, 15000)
+    assert.equal(rejected.payload?.connectionId, standaloneId)
+
+    const stillWorks = await run(admin, standaloneId, "PING")
+    assert.equal(stillWorks.payload, "PONG")
+  })
+
+  it("rejects actions on a connection the socket did not open", async () => {
+    stranger.send({ type: VALKEY.COMMAND.sendRequested, payload: { connectionId: standaloneId, command: "PING" } })
+    assert.equal(await silentFor(stranger, [VALKEY.COMMAND.sendFulfilled, VALKEY.COMMAND.sendFailed]), 0)
+  })
+
+  it("rejects a clusterId with no connectionId", async () => {
+    assert.ok(clusterId)
+    stranger.send({
+      type: VALKEY.COMMANDLOGS.commandLogsRequested,
+      payload: { clusterId, commandLogType: COMMANDLOG_TYPE.SLOW },
+    })
+    assert.equal(await silentFor(stranger, [VALKEY.COMMANDLOGS.commandLogsFulfilled, VALKEY.COMMANDLOGS.commandLogsError]), 0)
+  })
+
+  it("rejects a foreign clusterId paired with an owned connectionId", async () => {
+    assert.ok(clusterId)
+    const standaloneOnly = await WsClient.connect(WS_URL)
+    try {
+      const details = defaultStandaloneConnectionDetails(0)
+      standaloneOnly.send({ type: VALKEY.CONNECTION.connectPending, payload: { connectionId: standaloneId, connectionDetails: details } })
+      await standaloneOnly.waitFor(VALKEY.CONNECTION.standaloneConnectFulfilled, 30000)
+      standaloneOnly.send({
+        type: VALKEY.COMMANDLOGS.commandLogsRequested,
+        payload: { connectionId: standaloneId, clusterId, commandLogType: COMMANDLOG_TYPE.SLOW },
+      })
+      assert.equal(
+        await silentFor(standaloneOnly, [VALKEY.COMMANDLOGS.commandLogsFulfilled, VALKEY.COMMANDLOGS.commandLogsError]),
+        0,
+      )
+    } finally {
+      await standaloneOnly.close()
+    }
+  })
+
+  it("accepts a db-less nodeId with its own clusterId (monitor banner path)", async () => {
+    assert.ok(clusterId)
+    admin.send({
+      type: VALKEY.MONITOR.monitorRequested,
+      payload: { connectionId: toNodeId(clusterId0), clusterId, monitorAction: MONITOR_ACTION.STATUS },
+    })
+    const replies = await admin.collectFor(VALKEY.MONITOR.monitorFulfilled, 10000)
+    assert.ok(replies.length >= 1, `expected at least one node to answer STATUS; got ${replies.length}`)
+  })
+
+  it("accepts an empty-string clusterId on a standalone connection", async () => {
+    admin.send({
+      type: VALKEY.COMMANDLOGS.commandLogsRequested,
+      payload: { connectionId: standaloneId, clusterId: "", commandLogType: COMMANDLOG_TYPE.SLOW },
+    })
+    const fulfilled = await admin.collectFor(VALKEY.COMMANDLOGS.commandLogsFulfilled, 15000)
+    const errored = await admin.collectFor(VALKEY.COMMANDLOGS.commandLogsError, 100)
+    assert.ok(fulfilled.length + errored.length >= 1)
+  })
+
+  it("asks for the password when a password connection resumes without one", async () => {
+    stranger.send({
+      type: VALKEY.CONNECTION.connectPending,
+      payload: {
+        connectionId: standaloneId,
+        connectionDetails: { ...standaloneAdmin, password: undefined },
+        isResume: true,
+      },
+    })
+    const rejected = await stranger.waitFor(VALKEY.CONNECTION.connectRejected, 10000)
+    assert.equal(rejected.payload?.requiresAuth, true)
+  })
+})