cx-dgsp: retire the legacy bearer wire subprotocol

Packages/macOS/CmuxNext/Sources/CmuxNextFeed/Wire/CloudFeedSource.swift

@@ -142,31 +142,13 @@ public final class CloudFeedSource: FeedSource {
     /// This app's version (`CFBundleShortVersionString`), sent on every wire connect.
     static let clientVersion = Bundle.main.object(forInfoDictionaryKey: "CFBundleShortVersionString") as? String
 
-    /// A Worker that reads the upgrade's Authorization header names this header on every wire 401.
-    static let wireAuthHeader = "x-cmux-wire-auth"
-
-    /// The upgrade's 401 lacked `x-cmux-wire-auth`: a Worker from before cx-dgsp, which reads only the legacy subprotocol.
-    private struct LegacyWorker: Error {}
-
-    /// The token goes in the Authorization header, which Workers invocation logs never record
-    /// (cx-dgsp). Only a Worker that does not read it yet gets the legacy `bearer.` subprotocol, once.
+    /// The token goes only in the Authorization header, which Workers invocation logs never record
+    /// (cx-dgsp). The legacy `bearer.` subprotocol is retired: a 401 is final.
     private func session(onSnapshot: () -> Void) async throws {
         let bearer = try await token()
-        do {
-            try await session(bearer: bearer, legacy: false, onSnapshot: onSnapshot)
-        } catch is LegacyWorker {
-            try await session(bearer: bearer, legacy: true, onSnapshot: onSnapshot)
-        }
-    }
-
-    private func session(bearer: String, legacy: Bool, onSnapshot: () -> Void) async throws {
         var request = URLRequest(url: wireURL)
-        if legacy {
-            request.setValue("cmux.wire.v1, bearer.\(bearer)", forHTTPHeaderField: "Sec-WebSocket-Protocol")
-        } else {
-            request.setValue("cmux.wire.v1", forHTTPHeaderField: "Sec-WebSocket-Protocol")
-            request.setValue("Bearer \(bearer)", forHTTPHeaderField: "Authorization")
-        }
+        request.setValue("cmux.wire.v1", forHTTPHeaderField: "Sec-WebSocket-Protocol")
+        request.setValue("Bearer \(bearer)", forHTTPHeaderField: "Authorization")
         // updates.minimumVersion (enterprise P17-4): the server refuses older clients with client.too_old.
         if let version = Self.clientVersion { request.setValue(version, forHTTPHeaderField: "x-cmux-client-version") }
         let task = URLSession.shared.webSocketTask(with: request)
@@ -175,20 +157,9 @@ public final class CloudFeedSource: FeedSource {
         socket = task
         task.resume()
         defer { task.cancel(with: .goingAway, reason: nil) }
-        var first = true
         // wakeup-allow: each turn awaits the next frame; a closed socket throws and ends the loop.
         while !Task.isCancelled {
-            let message: URLSessionWebSocketTask.Message
-            do {
-                message = try await task.receive()
-            } catch {
-                if first, !legacy, let response = task.response as? HTTPURLResponse, response.statusCode == 401,
-                   response.value(forHTTPHeaderField: Self.wireAuthHeader) == nil {
-                    throw LegacyWorker()
-                }
-                throw error
-            }
-            first = false
+            let message = try await task.receive()
             let data: Data
             switch message {
             case let .string(text): data = Data(text.utf8)

backend/apps/api/src/index.ts

@@ -70,12 +70,12 @@ const gateUnreachable = (e: unknown) => {
 /**
  * WebSocket gateway: `GET /v1/wire/{user|team|feed|cloud}` and `/v1/wire/conv/<conversation>` with subprotocol
  * `cmux.wire.v1`. The credential is the Authorization header (native clients), a `ticket.<ticket>`
- * subprotocol from POST /v1/wire/ticket (browsers), or the legacy `bearer.<token>` subprotocol
- * (wire-ticket.ts; invocation logs record every header but Authorization). The Worker
+ * subprotocol from POST /v1/wire/ticket (browsers); the legacy `bearer.<token>` subprotocol is
+ * retired (wire-ticket.ts; invocation logs record every header but Authorization). The Worker
  * authenticates and passes the principal to the owner DO; frames never carry identity.
  */
 const wire = async (request: Request, env: Env, spent: SpentTicket | undefined, scope: string, conversation?: string /* or agent for mux */): Promise<Response> => {
-  const credential = await wireCredential(request, env, scope, spent)
+  const credential = await wireCredential(request, env, spent)
   if (!credential.ok) return credential.response
   const { principal: authenticatedToken, protocols } = credential
   if (!authenticatedToken?.user || !authenticatedToken.team) return wireUnauthenticated()

backend/apps/api/src/wire-ticket.ts

@@ -14,12 +14,14 @@ import { isMachineInstallKind } from "./machine-installs.ts"
  *    the Authorization header mints the ticket: 32 random bytes, single use, 30 s, 60 per minute
  *    per user. The router entry spends it on any path, before any other check, so a ticket in a
  *    log is always spent.
- * 3. Subprotocol `bearer.<token>`: the legacy path, kept until every client is updated. Each use
- *    logs `wire.legacy_bearer` (no value), so the remaining clients can be counted.
+ *
+ * The legacy `bearer.<token>` subprotocol is retired: it authenticates nothing, and it is still
+ * stripped from the subprotocols forwarded to the owner object, so a token never travels further.
  */
 
 export const TICKET_PREFIX = "ticket."
-const LEGACY_PREFIX = "bearer."
+/** The retired credential subprotocol: never read, always stripped. */
+const RETIRED_PREFIX = "bearer."
 /** A ticket lives 30 s: a client mints it and upgrades at once; 30 s covers slow cellular and TLS. */
 export const WIRE_TICKET_TTL_MS = 30_000
 /** Each 401 from a wire upgrade names this header, so a native client knows the server reads Authorization. */
@@ -128,15 +130,12 @@ export const spendTicketAtEntry = async (request: Request, env: Env): Promise<Sp
  * Resolves the upgrade's principal and returns the subprotocols without any credential, so the
  * request forwarded to the owner object carries none. `spent` is the router entry's ticket result.
  */
-export const wireCredential = async (request: Request, env: Env, scope: string, spent: SpentTicket | undefined): Promise<WireCredential> => {
+export const wireCredential = async (request: Request, env: Env, spent: SpentTicket | undefined): Promise<WireCredential> => {
   const all = subprotocols(request)
-  const protocols = all.filter((p) => !p.startsWith(TICKET_PREFIX) && !p.startsWith(LEGACY_PREFIX))
+  const protocols = all.filter((p) => !p.startsWith(TICKET_PREFIX) && !p.startsWith(RETIRED_PREFIX))
   if (spent !== undefined) return spent.principal ? { ok: true, principal: spent.principal, protocols } : { ok: false, response: unauthenticated("unknown or spent ticket") }
   const header = bearerOf(request)
-  if (header !== undefined) return { ok: true, principal: await authenticate(env, header), protocols }
-  const legacy = all.find((p) => p.startsWith(LEGACY_PREFIX))?.slice(LEGACY_PREFIX.length)
-  if (legacy !== undefined) console.log(JSON.stringify({ msg: "wire.legacy_bearer", scope }))
-  return { ok: true, principal: await authenticate(env, legacy), protocols }
+  return { ok: true, principal: header === undefined ? undefined : await authenticate(env, header), protocols }
 }
 
 /** The forwarded upgrade: the request's headers without Authorization and with credential-free subprotocols. */

backend/apps/api/test/api.test.ts

@@ -46,7 +46,7 @@ const mintToken = async (user: string, install: string, pair: CryptoKeyPair) =>
 }
 
 const openWire = async (scope: "user" | "team", token: string) => {
-  const res = await worker.fetch(`https://api.test/v1/wire/${scope}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${token}` } })
+  const res = await worker.fetch(`https://api.test/v1/wire/${scope}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${token}` } })
   expect(res.status).toBe(101)
   const ws = res.webSocket!
   const frames: Array<any> = []

backend/apps/api/test/chief-brain-live.test.ts

@@ -42,7 +42,7 @@ const ownerWithChief = async (sub: string) => {
 }
 
 const connect = async (id: string, token: string) => {
-  const res = await worker.fetch(`https://api.test/v1/wire/conv/${id}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${token}` } })
+  const res = await worker.fetch(`https://api.test/v1/wire/conv/${id}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${token}` } })
   expect(res.status).toBe(101)
   const ws = res.webSocket!
   const frames: Array<any> = []

backend/apps/api/test/chief-brain-socket-recheck.test.ts

@@ -31,7 +31,7 @@ const placedChief = async (name: string) => {
 
 /** The chief's socket on its main conversation: op frames, their answers, and a close. */
 const chiefSocket = async (conversation: string, token: string) => {
-  const res = await worker.fetch(`https://api.test/v1/wire/conv/${conversation}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${token}` } })
+  const res = await worker.fetch(`https://api.test/v1/wire/conv/${conversation}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${token}` } })
   expect(res.status).toBe(101)
   const ws = res.webSocket!
   const frames: Array<any> = []
@@ -121,7 +121,7 @@ describe("placed chief rights on an open socket", { timeout: 60_000 }, () => {
     const thread = await call("/v1/read", c.token, { op: "conversation.thread", params: { conversation, root: (msg as unknown as { id: string }).id } })
     expect(thread.status, JSON.stringify(thread.json)).toBe(200)
     // Its wake queue socket and mux.ack.
-    const res = await worker.fetch(`https://api.test/v1/wire/mux/${c.chief.id}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${c.token}` } })
+    const res = await worker.fetch(`https://api.test/v1/wire/mux/${c.chief.id}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${c.token}` } })
     expect(res.status).toBe(101)
     res.webSocket!.accept()
     res.webSocket!.close()

backend/apps/api/test/chief-brain.test.ts

@@ -70,7 +70,7 @@ const waitFor = async (frames: Array<any>, pred: (f: any) => boolean) => {
 
 /** A conversation socket with a frame log. */
 const connect = async (id: string, token: string) => {
-  const res = await worker.fetch(`https://api.test/v1/wire/conv/${id}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${token}` } })
+  const res = await worker.fetch(`https://api.test/v1/wire/conv/${id}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${token}` } })
   expect(res.status).toBe(101)
   const ws = res.webSocket!
   const frames: Array<any> = []

backend/apps/api/test/chief-token.test.ts

@@ -55,7 +55,7 @@ describe("chief tokens", { timeout: 60_000 }, () => {
     // A chief token cannot change the owner's account.
     expect((await op(tok.json.access_token, "install.revoke", { install: a.install })).json.ok).toBe(false)
     // Archive the chief: the same token is refused on the next request, and its open sockets close.
-    const muxRes = await worker.fetch(`https://api.test/v1/wire/mux/${chief.id}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${tok.json.access_token}` } })
+    const muxRes = await worker.fetch(`https://api.test/v1/wire/mux/${chief.id}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${tok.json.access_token}` } })
     expect(muxRes.status).toBe(101)
     const ws = muxRes.webSocket!
     let closed: number | undefined
@@ -87,7 +87,7 @@ describe("chief tokens", { timeout: 60_000 }, () => {
     expect(JSON.stringify(chiefList.json)).toContain("auth.forbidden")
     // The cloud socket (the stream behind the daemon's cloud-inbox-subscribe).
     const subscribe = async (token: string) => {
-      const res = await worker.fetch("https://api.test/v1/wire/user", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${token}` } })
+      const res = await worker.fetch("https://api.test/v1/wire/user", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${token}` } })
       expect(res.status).toBe(101)
       const ws = res.webSocket!
       const frames: Array<any> = []
@@ -123,7 +123,7 @@ describe("chief tokens", { timeout: 60_000 }, () => {
     expect((hist.json.value.messages as Array<{ client_msg_id: string; seq: number }>).some((m) => m.client_msg_id === "to-chief")).toBe(true)
     const seq = (hist.json.value.messages as Array<{ client_msg_id: string; seq: number }>).find((m) => m.client_msg_id === "to-chief")!.seq
     const firstFrame = async (path: string) => {
-      const res = await worker.fetch(`https://api.test${path}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${tok}` } })
+      const res = await worker.fetch(`https://api.test${path}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${tok}` } })
       expect(res.status).toBe(101)
       const ws = res.webSocket!
       const frames: Array<any> = []

backend/apps/api/test/cloud-routes.test.ts

@@ -121,7 +121,7 @@ describe("cloud ops through the Worker", { timeout: 60_000 }, () => {
   it("publishes cloud.machine.upsert and cloud.machine.removed on the cloud wire with revision = stream seq", async () => {
     const { t, team } = await signedIn("cloud-route-3")
     const created = await op(t, "cloud.machine.create", { name: "watched", size: { cpu: 2, memory_mb: 4096, disk_mb: 16384 } })
-    const res = await worker.fetch("https://api.test/v1/wire/cloud", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${t}` } })
+    const res = await worker.fetch("https://api.test/v1/wire/cloud", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${t}` } })
     expect(res.status).toBe(101)
     const ws = res.webSocket!
     const frames: Array<any> = []

backend/apps/api/test/cloud-snapshots.test.ts

@@ -47,7 +47,7 @@ describe("snapshots", { timeout: 60_000 }, () => {
     const machine = (await post("/v1/ops", a.session, { op: "cloud.machine.create", params: { size: SIZE }, idempotency_key: crypto.randomUUID(), origin: "user" })).body.value.machine.id as string
     const { json } = await bindFile(cloudStub(a.team), machine)
     await post("/v1/cloud/bind", undefined, { team: a.team, machine, bind_token: json.bind_token, wg_public_key: WG_KEY, daemon: DAEMON, install_public_jwk: (await vmKey()).jwk })
-    const res = await worker.fetch("https://api.test/v1/wire/cloud", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${a.session}` } })
+    const res = await worker.fetch("https://api.test/v1/wire/cloud", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${a.session}` } })
     const ws = res.webSocket!
     const frames: Array<any> = []
     ws.addEventListener("message", (e) => frames.push(JSON.parse(e.data as string)))

backend/apps/api/test/cloud-vm-install.test.ts

@@ -71,7 +71,7 @@ describe("VM install at bind", { timeout: 60_000 }, () => {
 
   it("event.emit: v1 kinds reach the team's subscribers as an ephemeral event; bad kinds, big data and floods are refused", async () => {
     const s = await vmSetup("cloud-bind-4")
-    const res = await worker.fetch("https://api.test/v1/wire/cloud", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${s.a.session}` } })
+    const res = await worker.fetch("https://api.test/v1/wire/cloud", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${s.a.session}` } })
     const ws = res.webSocket!
     const frames: Array<any> = []
     ws.addEventListener("message", (e) => frames.push(JSON.parse(e.data as string)))

backend/apps/api/test/cloud-vm-isolation.test.ts

@@ -35,7 +35,7 @@ describe("VM install isolation", { timeout: 60_000 }, () => {
     for (const op of ["install.list", "team.directory", "team.members.list"]) expect((await read(s.vmToken, op, {})).status, op).toBe(403)
     expect((await post("/v1/ops", s.vmToken, { op: "install.rename", params: { install: s.install.id, name: "x" }, idempotency_key: crypto.randomUUID(), origin: "cli" })).status).toBe(403)
     for (const scope of ["user", "team", "feed", "cloud"]) {
-      const res = await worker.fetch(`https://api.test/v1/wire/${scope}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${s.vmToken}` } })
+      const res = await worker.fetch(`https://api.test/v1/wire/${scope}`, { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${s.vmToken}` } })
       expect(res.status, scope).toBe(403)
     }
     const pk = await worker.fetch("https://api.test/v1/presence-key", { method: "POST", headers: { "content-type": "application/json", authorization: `Bearer ${s.vmToken}` }, body: "{}" })
@@ -59,7 +59,7 @@ describe("VM install isolation", { timeout: 60_000 }, () => {
 
   it("event times are clamped to the server clock and URL queries are stripped in any case (review P3)", async () => {
     const s = await vmSetup("cloud-bind-2")
-    const res = await worker.fetch("https://api.test/v1/wire/cloud", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${s.a.session}` } })
+    const res = await worker.fetch("https://api.test/v1/wire/cloud", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${s.a.session}` } })
     const ws = res.webSocket!
     const frames: Array<any> = []
     ws.addEventListener("message", (e) => frames.push(JSON.parse(e.data as string)))

backend/apps/api/test/feed.test.ts

@@ -46,7 +46,7 @@ const setup = async (stackUser: string) => {
 }
 
 const openFeed = async (token: string) => {
-  const res = await worker.fetch("https://api.test/v1/wire/feed", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": `cmux.wire.v1, bearer.${token}` } })
+  const res = await worker.fetch("https://api.test/v1/wire/feed", { headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "cmux.wire.v1", authorization: `Bearer ${token}` } })
   expect(res.status).toBe(101)
   const ws = res.webSocket!
   const frames: Array<any> = []