feat: bind each realtime ws ticket to the session that minted it and accept it once while that session is live, require a second-factor session to join admin:platform, confine API key and OAuth sockets to their own workspace with member permissions narrowed to their read scopes, and refuse those credentials in the socket URL unless REALTIME_ALLOW_QUERY_CREDENTIALS=true

AGENTS.md

@@ -418,7 +418,7 @@ Web conventions:
 
 ### Developer WebSocket
 
-API keys with the `REALTIME_SUBSCRIBE` permission (bit 11) can connect to the same socket, presenting the key (or a `wmat_` OAuth token) in the `x-warmbly-token` header; the query-string form for them is deprecated and kept only for existing clients, while the 10-minute ws ticket rides the query string. Connection spam is bounded by per-user concurrent-connection caps (plan-based, default 10), per-IP (50), a global cap, join rate limits, and per-key IP restrictions. Documented in `docs/content/docs/api/realtime.mdx` — keep that page in sync with channel/limit changes.
+API keys with the `REALTIME_SUBSCRIBE` permission (bit 11) can connect to the same socket, presenting the key (or a `wmat_` OAuth token) in the `x-warmbly-token` header; in the query string they are refused unless the realtime service sets `REALTIME_ALLOW_QUERY_CREDENTIALS=true`. Such a socket reaches only the credential's own workspace, with member permissions narrowed to its read scopes (`Auth.confine`). The 10-minute ws ticket rides the query string, opens one socket, and names its session, which must still be live; `admin:platform` additionally requires that session to have verified a second factor. Connection spam is bounded by per-user concurrent-connection caps (plan-based, default 10), per-IP (50), a global cap, join rate limits, and per-key IP restrictions. Documented in `docs/content/docs/api/realtime.mdx` — keep that page in sync with channel/limit changes.
 
 ## System Shape
 

docs/content/docs/api/realtime.mdx

@@ -25,9 +25,9 @@ Three token types are accepted:
 
 - **API key** (`wmbly_…`): send a key with the `REALTIME_SUBSCRIBE` permission (bit 11) in the `X-Warmbly-Token` request header of the WebSocket handshake. See [Permissions](/api/permissions/) for how to grant it.
 - **OAuth access token** (`wmat_…`): send an access token whose grant includes the realtime scope (the lowercased permission, `realtime_subscribe`) in the same `X-Warmbly-Token` header. It connects on the granting user's behalf with exactly the scopes they approved. See [OAuth](/api/oauth/).
-- **Short-lived JWT**: browser sessions call the authenticated socket endpoint to mint a 10-minute connection token and pass it as the `token` query parameter. This is what the dashboard itself uses.
+- **Short-lived JWT**: browser sessions call the authenticated socket endpoint to mint a 10-minute connection token and pass it as the `token` query parameter. This is what the dashboard itself uses. Each token opens one socket, and only while the session that minted it is still signed in, so mint a fresh one for every connection attempt.
 
-Send API keys and OAuth access tokens in the `x-warmbly-token` header. Putting one in the URL as `token` still works but is deprecated: URLs end up in proxy and access logs, and these credentials outlive any session, so it will stop working in a later API version. The 10-minute connection token belongs in the query string. Browsers cannot set headers on a WebSocket, so connect with an API key or OAuth token from a server-side client.
+Send API keys and OAuth access tokens in the `x-warmbly-token` header. An API key or OAuth token in the URL as `token` is refused with code `4003`: URLs end up in proxy and access logs, and these credentials outlive any session. A self-hosted instance with an older client that cannot set the header can accept them there again with `REALTIME_ALLOW_QUERY_CREDENTIALS=true` on the realtime service (see [Configuration](/development/configuration/)). The 10-minute connection token belongs in the query string. Browsers cannot set headers on a WebSocket, so connect with an API key or OAuth token from a server-side client.
 
 `POST /v1/getaway` continues returning the direct gateway `url` and `expires_in`. It also returns `proxy_path` when a gateway is configured. The dashboard resolves that path against its versioned API base and uses the API origin for its WebSocket, so a self-host does not depend on realtime CSP discovery succeeding during dashboard startup. Failed connection attempts alternate with the direct gateway for older proxy configurations. The proxy accepts only short-lived WebSocket tickets, not access tokens, API keys or OAuth credentials; developer clients continue connecting directly. It preserves the dashboard Origin and realtime authorization and connection limits. The backend signs a ticket-bound client-address proof so per-IP limits do not treat every proxied user as the same client.
 
@@ -35,6 +35,18 @@ A credential stops working on the socket as soon as it does on the REST API: a r
 
 In short, both developer auth methods work here: an API key with `REALTIME_SUBSCRIBE`, or an OAuth access token with the realtime scope. They authenticate through the same gate, so everything below applies to either.
 
+A key or token reaches only the workspace it was created in, and there only the events its read scopes cover, never more than its holder's own member permissions:
+
+| Scope | Event families it receives |
+| --- | --- |
+| `READ_EMAILS` | Mailbox, warmup and mailbox import events, and the `account:` channel |
+| `READ_CAMPAIGNS` | Campaign activity, task progress and send, open, click and reply pulses |
+| `READ_CONTACTS` | Contact, page hit, form and research events |
+| `READ_UNIBOX` | Inbox events and AI reply drafts |
+| `READ_ANALYTICS` | Placement tests and direct email events |
+
+Team, billing and settings events never reach a key or token. Joining another workspace's `org:`, `campaign:` or `account:` topic is refused with `not_a_member`, and bulk operation progress arrives only for operations in the credential's own workspace.
+
 If the key (or OAuth app) has IP restrictions configured, they are enforced on the socket handshake as well.
 
 ## Channels
@@ -354,8 +366,8 @@ Both paths use the same reason codes:
 
 | Code | Meaning | Client action |
 | --- | --- | --- |
-| 4003 | Not authenticated (token missing) | Add a valid `token` to the connection URL |
-| 4004 | Authentication failed: token expired, or invalid key | Re-mint the token (or check the key), then reconnect |
+| 4003 | Not authenticated (token missing, or an API key or OAuth token sent in the URL) | Add a valid `token` to the connection URL, or move the key to the `x-warmbly-token` header |
+| 4004 | Authentication failed: token expired, already used, its session signed out, or invalid key | Re-mint the token (or check the key), then reconnect |
 | 4005 | Malformed topic, for example an id that is not a UUID | Fix the topic; retrying will not help |
 | 4007 | Rate limited | Wait out `retry_after_ms`, then retry |
 | 4009 | Connection limit exceeded | Reduce concurrent sockets |

docs/content/docs/development/configuration.mdx

@@ -959,6 +959,7 @@ The Elixir websocket service. Its runtime configuration is read only when the re
 | `PORT` | Listen port | `4000` |
 | `CHECK_ORIGIN_HOSTS` | Origins a browser may open a socket from, comma separated, e.g. `https://app.example.com,https://admin.example.com`. Not this service's own host | unset |
 | `CHECK_ORIGIN` | Legacy host-based form, consulted only when `CHECK_ORIGIN_HOSTS` is empty. `true` accepts an upgrade only from `PHX_HOST` | `false` |
+| `REALTIME_ALLOW_QUERY_CREDENTIALS` | `true` accepts an API key or OAuth token in the socket URL's `token` parameter, for an older client that cannot send the `x-warmbly-token` header. URLs reach proxy and access logs, so leave it off unless such a client still depends on it. Read in every environment, not only `prod` | `false` |
 | `PUBSUB_ENABLED` | `true` swaps the Redis bridge for Google Pub/Sub | `false` |
 | `GCP_PROJECT_ID` | Required when `PUBSUB_ENABLED=true`; the service refuses to boot without it | unset |
 | `MAX_CONNECTIONS_PER_USER` | Concurrent sockets one account may hold. The caller's plan limit applies too, whichever is lower | `10` |

internal/api/handler/socket.go

@@ -27,7 +27,13 @@ func (h *Handler) GenerateWebsocket(c *gin.Context) {
 		return
 	}
 
-	token, xerr := h.SocketService.GenerateWebsocketToken(c.Request.Context(), uid)
+	session := middleware.GetSession(c)
+	if session == nil {
+		errx.Handle(c, errx.ErrUnauthorized)
+		return
+	}
+
+	token, xerr := h.SocketService.GenerateWebsocketToken(c.Request.Context(), uid, session.ID)
 	if xerr != nil {
 		errx.Handle(c, xerr)
 		return

internal/api/handler/socket_test.go

@@ -13,13 +13,15 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
+	"github.com/warmbly/warmbly/internal/api/middleware"
 	"github.com/warmbly/warmbly/internal/app/token"
 	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/models"
 )
 
 type socketTicketStub struct{ ticket string }
 
-func (s socketTicketStub) GenerateWebsocketToken(context.Context, uuid.UUID) (string, *errx.Error) {
+func (s socketTicketStub) GenerateWebsocketToken(context.Context, uuid.UUID, uuid.UUID) (string, *errx.Error) {
 	return s.ticket, nil
 }
 
@@ -31,6 +33,7 @@ func TestSocketTicketNormalizesLegacyURL(t *testing.T) {
 		c, _ := gin.CreateTestContext(w)
 		c.Request = httptest.NewRequest(http.MethodPost, "/v1/getaway", nil)
 		c.Set("user_id", uuid.New().String())
+		c.Set(middleware.SessionKey, &models.Session{ID: uuid.New()})
 		h.GenerateWebsocket(c)
 		var response map[string]any
 		if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {

internal/app/socket/gen.go

@@ -5,13 +5,15 @@ import (
 	"time"
 
 	"github.com/google/uuid"
-	"github.com/warmbly/warmbly/internal/app/token"
 	"github.com/warmbly/warmbly/internal/errx"
 	"github.com/warmbly/warmbly/internal/observability/errs"
 	"github.com/warmbly/warmbly/internal/pkg/crypt"
 )
 
-func (s *socketService) GenerateWebsocketToken(ctx context.Context, userID uuid.UUID) (string, *errx.Error) {
+func (s *socketService) GenerateWebsocketToken(ctx context.Context, userID, sessionID uuid.UUID) (string, *errx.Error) {
+	if sessionID == uuid.Nil {
+		return "", errx.ErrUnauthorized
+	}
 	issuedAt := time.Now()
 	expiresAt := issuedAt.Add(SocketTTL)
 	id := uuid.New()
@@ -21,7 +23,7 @@ func (s *socketService) GenerateWebsocketToken(ctx context.Context, userID uuid.
 		return "", errx.InternalError()
 	}
 
-	wsToken, err := s.tokenService.GenerateTokenFor(token.PurposeWebSocket, userID, id, "", nonce, issuedAt, expiresAt)
+	wsToken, err := s.tokenService.GenerateWebsocketTicket(userID, id, sessionID, nonce, issuedAt, expiresAt)
 	if err != nil {
 		errs.CaptureException(err)
 		return "", errx.InternalError()

internal/app/socket/service.go

@@ -10,7 +10,8 @@ import (
 )
 
 type SocketService interface {
-	GenerateWebsocketToken(ctx context.Context, userID uuid.UUID) (string, *errx.Error)
+	// GenerateWebsocketToken mints a short-lived socket ticket bound to the caller's session.
+	GenerateWebsocketToken(ctx context.Context, userID, sessionID uuid.UUID) (string, *errx.Error)
 }
 
 type socketService struct {

internal/app/token/gen.go

@@ -25,6 +25,8 @@ type TokenClaims struct {
 	// password-reset token from being accepted as a session.
 	Purpose  string `json:"purpose,omitempty"`
 	ClientIP string `json:"client_ip,omitempty"`
+	// TicketSession is the session a websocket ticket was minted from; sid is the ticket's own id.
+	TicketSession string `json:"session_id,omitempty"`
 	jwt.RegisteredClaims
 }
 
@@ -72,6 +74,20 @@ func (s *tokenService) VerifyToken(tokenStr string) (*TokenClaims, *errx.Error)
 	return claims, nil
 }
 
+// GenerateWebsocketTicket mints a socket ticket the realtime service accepts once,
+// and only while the session it names is live.
+func (s *tokenService) GenerateWebsocketTicket(userID, ticketID, sessionID uuid.UUID, nonce string, issuedAt, expiresAt time.Time) (string, error) {
+	claims := TokenClaims{
+		UserID: userID, SessionID: ticketID, Nonce: nonce,
+		Purpose: PurposeWebSocket, TicketSession: sessionID.String(),
+		RegisteredClaims: jwt.RegisteredClaims{
+			ExpiresAt: jwt.NewNumericDate(expiresAt),
+			IssuedAt:  jwt.NewNumericDate(issuedAt),
+		},
+	}
+	return jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString([]byte(s.AuthSecret))
+}
+
 // The proof is bound to one ticket and is never accepted as a socket credential.
 func (s *tokenService) GenerateWebsocketProxyProof(ticket *TokenClaims, clientIP string) (string, error) {
 	now := time.Now()

internal/app/token/purpose_test.go

@@ -7,6 +7,28 @@ import (
 	"github.com/google/uuid"
 )
 
+// A socket ticket names the session it was minted from, so the realtime service can require that session to be live.
+func TestWebsocketTicketNamesItsSession(t *testing.T) {
+	s := &tokenService{AuthSecret: "test-secret-at-least-32-characters-long"}
+	userID, ticketID, sessionID := uuid.New(), uuid.New(), uuid.New()
+	now := time.Now()
+
+	tok, err := s.GenerateWebsocketTicket(userID, ticketID, sessionID, "nonce", now, now.Add(time.Minute))
+	if err != nil {
+		t.Fatal(err)
+	}
+	claims, xerr := s.VerifyTokenFor(PurposeWebSocket, tok)
+	if xerr != nil {
+		t.Fatalf("ticket should verify as a socket ticket: %v", xerr)
+	}
+	if claims.UserID != userID || claims.SessionID != ticketID || claims.TicketSession != sessionID.String() || claims.Nonce != "nonce" {
+		t.Fatalf("ticket claims: %+v", claims)
+	}
+	if _, xerr := s.VerifyTokenFor(PurposeAccess, tok); xerr == nil {
+		t.Fatal("a socket ticket must not verify as an access token")
+	}
+}
+
 // A token minted for one flow must not be spendable on another. This is the
 // regression for the realtime socket accepting a password-reset link token,
 // and for anything else that shares the signing key.

internal/app/token/service.go

@@ -20,6 +20,8 @@ type TokenService interface {
 	// minted for one flow cannot be spent on another.
 	GenerateTokenFor(purpose string, userID, sessionID uuid.UUID, email, nonce string, issuedAt, expiresAt time.Time) (string, error)
 	GenerateWebsocketProxyProof(ticket *TokenClaims, clientIP string) (string, error)
+	// GenerateWebsocketTicket mints a socket ticket bound to the session it was minted from.
+	GenerateWebsocketTicket(userID, ticketID, sessionID uuid.UUID, nonce string, issuedAt, expiresAt time.Time) (string, error)
 	VerifyToken(tokenStr string) (*TokenClaims, *errx.Error)
 	// VerifyTokenFor also requires the token to have been minted for this
 	// purpose, so one flow's token cannot be spent on another.

realtime/config/runtime.exs

@@ -1,5 +1,13 @@
 import Config
 
+# API keys and OAuth tokens are read from the x-warmbly-token header; the query
+# string carries them only when an operator turns this on for an older client.
+if config_env() != :test do
+  config :realtime,
+    allow_query_credentials:
+      String.trim(System.get_env("REALTIME_ALLOW_QUERY_CREDENTIALS", "")) == "true"
+end
+
 if config_env() == :prod do
   # Required environment variables
   jwt_secret =

realtime/config/test.exs

@@ -10,7 +10,8 @@ config :realtime,
   jwt_secret: "test_jwt_secret",
   pubsub_enabled: false,
   # Count rate-limit hits in memory: the suite must run with no Redis.
-  rate_limit_counter: Realtime.Test.Counter
+  rate_limit_counter: Realtime.Test.Counter,
+  ticket_store: Realtime.Test.Counter
 
 config :realtime, auth_repo: Realtime.Test.AuthRepo
 config :realtime, subscription_repo: Realtime.Test.AuthRepo

realtime/lib/realtime/api_key.ex

@@ -37,10 +37,10 @@ defmodule Realtime.ApiKey do
   end
 
   @doc """
-  Validate an API key and return the user_id if valid.
+  Validate an API key and return its holder and grant if valid.
 
   Returns:
-  - {:ok, user_id} if valid
+  - {:ok, user_id, %{organization_id, scopes}} if valid
   - {:error, reason} if invalid
 
   Checks:
@@ -57,14 +57,15 @@ defmodule Realtime.ApiKey do
          :ok <- check_expiration(key_data),
          :ok <- check_permission(key_data),
          :ok <- check_ip_restriction(key_data, client_ip),
-         :ok <- check_live(key_data) do
-      {:ok, key_data.user_id}
+         {:ok, grant} <- check_live(key_data) do
+      {:ok, key_data.user_id, grant}
     end
   end
 
-  # Uncached, so a revoked key, a banned holder or one who left the workspace stops at once.
+  # Uncached, so a revoked key, a banned holder or one who left the workspace stops at once,
+  # and the workspace and scopes the socket is confined to are the key's current ones.
   @live_query """
-  SELECT 1 FROM api_keys k JOIN users u ON u.id = k.user_id
+  SELECT k.organization_id::text, k.permissions FROM api_keys k JOIN users u ON u.id = k.user_id
   WHERE k.id = $1 AND k.status = 'active' AND (u.ban_scope & 1) = 0
     AND EXISTS (SELECT 1 FROM organization_members m
                 WHERE m.organization_id = k.organization_id AND m.user_id = k.user_id
@@ -73,8 +74,8 @@ defmodule Realtime.ApiKey do
 
   defp check_live(%{id: id}) do
     with {:ok, bin} <- Ecto.UUID.dump(id),
-         {:ok, %{rows: [_ | _]}} <- Repo.query(@live_query, [bin]) do
-      :ok
+         {:ok, %{rows: [[org_id, scopes] | _]}} <- Repo.query(@live_query, [bin]) do
+      {:ok, %{organization_id: org_id, scopes: scopes}}
     else
       {:ok, %{rows: []}} -> {:error, :key_inactive}
       _ -> {:error, :database_error}

realtime/lib/realtime/auth.ex

@@ -25,7 +25,11 @@ defmodule Realtime.Auth do
   - Otherwise = JWT token
 
   Returns:
-  - {:ok, user_id, :jwt}, {:ok, user_id, :api_key} or {:ok, user_id, :oauth} on success
+  - {:ok, user_id, auth_type, context} on success, where auth_type is :jwt,
+    :api_key or :oauth. For a ticket, context carries `session_id` and
+    `mfa_verified`; for an API key or OAuth token it carries `credential`,
+    the workspace the credential belongs to and the member permissions its
+    scopes reach.
   - {:error, reason} on failure
   """
   def verify_token(token, opts \\ [])
@@ -62,9 +66,10 @@ defmodule Realtime.Auth do
 
     case JOSE.JWT.verify_strict(jwk(secret), ["HS256"], token) do
       {true, %JOSE.JWT{fields: fields}, _} ->
-        case validate_claims(fields) do
-          {:ok, user_id} -> {:ok, user_id, :jwt}
-          error -> error
+        with {:ok, user_id} <- validate_claims(fields),
+             {:ok, session} <- ticket_session(fields, user_id),
+             :ok <- claim_ticket(fields) do
+          {:ok, user_id, :jwt, session}
         end
 
       {false, _, _} ->
@@ -118,13 +123,59 @@ defmodule Realtime.Auth do
 
   def proxy_ip(_, _), do: :error
 
+  @live_session_query """
+  SELECT mfa_verified FROM sessions
+  WHERE id = $1 AND user_id = $2 AND revoked_at IS NULL
+    AND (expires_at IS NULL OR expires_at > now())
+  """
+
+  # A ticket naming its session is accepted only while that session is live.
+  defp ticket_session(%{"session_id" => session_id}, user_id)
+       when is_binary(session_id) and session_id != "" do
+    case dump_and_query(@live_session_query, [session_id, user_id]) do
+      {:ok, %{rows: [[mfa_verified] | _]}} ->
+        {:ok, %{session_id: session_id, mfa_verified: mfa_verified == true}}
+
+      {:ok, %{rows: []}} ->
+        {:error, :session_revoked}
+
+      {:error, _reason} ->
+        {:error, :database_error}
+    end
+  end
+
+  # Tickets minted by a backend that predates the claim: no session, so no second factor.
+  defp ticket_session(_claims, _user_id), do: {:ok, %{session_id: nil, mfa_verified: false}}
+
+  # Each ticket opens one socket; the marker outlives the ticket itself.
+  defp claim_ticket(%{"sid" => ticket_id, "exp" => exp})
+       when is_binary(ticket_id) and ticket_id != "" and is_integer(exp) do
+    ttl = max(exp - System.system_time(:second), 1)
+
+    case ticket_store().set_nx("ws_ticket_used:" <> ticket_id, ttl) do
+      {:ok, "OK"} ->
+        :ok
+
+      {:ok, nil} ->
+        {:error, :ticket_used}
+
+      {:error, reason} ->
+        Logger.warning("Socket ticket claim unavailable, accepting: #{inspect(reason)}")
+        :ok
+    end
+  end
+
+  defp claim_ticket(_claims), do: {:error, :invalid_claims}
+
+  defp ticket_store, do: Application.get_env(:realtime, :ticket_store, Realtime.Redis)
+
   @doc """
   Verify an API key.
   """
   def verify_api_key(api_key, opts \\ []) do
     case ApiKey.validate(api_key, opts) do
-      {:ok, user_id} ->
-        {:ok, user_id, :api_key}
+      {:ok, user_id, credential} ->
+        {:ok, user_id, :api_key, %{credential: credential_grant(credential)}}
 
       {:error, reason} ->
         Logger.warning("API key verification failed: #{inspect(reason)}")
@@ -137,8 +188,8 @@ defmodule Realtime.Auth do
   """
   def verify_oauth_token(token, opts \\ []) do
     case OAuthToken.validate(token, opts) do
-      {:ok, user_id} ->
-        {:ok, user_id, :oauth}
+      {:ok, user_id, credential} ->
+        {:ok, user_id, :oauth, %{credential: credential_grant(credential)}}
 
       {:error, reason} ->
         Logger.warning("OAuth token verification failed: #{inspect(reason)}")
@@ -162,6 +213,8 @@ defmodule Realtime.Auth do
   def error_code(:key_inactive), do: 4004
   def error_code(:key_expired), do: 4004
   def error_code(:token_revoked), do: 4004
+  def error_code(:session_revoked), do: 4004
+  def error_code(:ticket_used), do: 4004
   def error_code(:database_error), do: 4004
   def error_code(:permission_denied), do: 4010
   def error_code(:ip_not_allowed), do: 4010
@@ -194,6 +247,8 @@ defmodule Realtime.Auth do
   def error_message(:key_inactive), do: "API key inactive"
   def error_message(:key_expired), do: "API key expired"
   def error_message(:token_revoked), do: "Access token revoked"
+  def error_message(:session_revoked), do: "Authentication failed"
+  def error_message(:ticket_used), do: "Authentication failed"
   def error_message(:database_error), do: "Authentication failed"
   def error_message(:permission_denied), do: "Permission denied"
   def error_message(:ip_not_allowed), do: "IP address not allowed"
@@ -311,6 +366,57 @@ defmodule Realtime.Auth do
   def in_scope?(%{scope: _}, _kind, _id), do: false
   def in_scope?(_member, _kind, _id), do: true
 
+  # API scope bit => the member permission it lets a credential socket use.
+  @scope_grants [
+    {0, :manage_emails},
+    {1, :view_campaigns},
+    {2, :view_contacts},
+    {3, :access_unibox},
+    {4, :view_analytics}
+  ]
+
+  @doc """
+  The workspace an API key or OAuth token belongs to, and the member
+  permissions its scopes reach. Team, billing and settings events need a
+  scope no credential has, so they never reach a credential socket.
+  """
+  def credential_grant(%{organization_id: org_id, scopes: scopes}) do
+    scopes = if is_integer(scopes), do: scopes, else: 0
+
+    mask =
+      Enum.reduce(@scope_grants, 0, fn {bit, perm}, acc ->
+        if Bitwise.band(scopes, Bitwise.bsl(1, bit)) != 0,
+          do: Bitwise.bor(acc, permission(perm)),
+          else: acc
+      end)
+
+    %{organization_id: org_id, permissions: mask}
+  end
+
+  @doc """
+  Narrow a membership lookup to what the socket's credential reaches. A
+  ticket socket (no credential) keeps the member's own permissions; an API
+  key or OAuth token reaches only its own workspace, and there only the
+  member permissions its scopes grant.
+  """
+  def confine(result, nil), do: result
+
+  def confine({:ok, %{organization_id: member_org} = member}, %{
+        organization_id: credential_org,
+        permissions: mask
+      })
+      when not is_nil(member_org) and not is_nil(credential_org) do
+    with {:ok, id} <- dump_uuid(credential_org),
+         {:ok, ^id} <- dump_uuid(member_org) do
+      {:ok, %{member | permissions: Bitwise.band(member.permissions || 0, mask)}}
+    else
+      _ -> {:error, :not_a_member}
+    end
+  end
+
+  def confine({:ok, _member}, _credential), do: {:error, :not_a_member}
+  def confine(error, _credential), do: error
+
   @doc false
   def matches_org?(_member, org_id) when org_id in [nil, ""], do: true
 

realtime/lib/realtime/oauth_token.ex

@@ -34,10 +34,10 @@ defmodule Realtime.OAuthToken do
   end
 
   @doc """
-  Validate an OAuth2 access token and return the user_id if valid.
+  Validate an OAuth2 access token and return its holder and grant if valid.
 
   Returns:
-  - {:ok, user_id} if valid
+  - {:ok, user_id, %{organization_id, scopes}} if valid
   - {:error, reason} if invalid
 
   Checks:
@@ -51,7 +51,7 @@ defmodule Realtime.OAuthToken do
          :ok <- check_revoked(grant),
          :ok <- check_expiration(grant),
          :ok <- check_scope(grant) do
-      {:ok, grant.user_id}
+      {:ok, grant.user_id, %{organization_id: grant.organization_id, scopes: grant.scopes}}
     end
   end
 
@@ -61,7 +61,7 @@ defmodule Realtime.OAuthToken do
   # The holder must still be a member, not banned from signing in, and the app usable,
   # the same conditions the API applies when it resolves the token.
   @lookup_query """
-  SELECT g.user_id::text, g.scopes, g.access_expires_at, g.revoked_at
+  SELECT g.user_id::text, g.scopes, g.access_expires_at, g.revoked_at, g.organization_id::text
   FROM oauth_access_grants g
   WHERE g.access_token_hash = $1
     AND EXISTS (SELECT 1 FROM organization_members m
@@ -73,10 +73,11 @@ defmodule Realtime.OAuthToken do
 
   defp lookup_token(token) do
     case Repo.query(@lookup_query, [hash_token(token)]) do
-      {:ok, %{rows: [[user_id, scopes, expires_at, revoked_at] | _]}} ->
+      {:ok, %{rows: [[user_id, scopes, expires_at, revoked_at, org_id] | _]}} ->
         {:ok,
          %{
            user_id: user_id,
+           organization_id: org_id,
            scopes: scopes,
            access_expires_at: expires_at,
            revoked_at: revoked_at

realtime/lib/realtime/redis.ex

@@ -105,6 +105,13 @@ defmodule Realtime.Redis do
     end
   end
 
+  @doc """
+  Set a key only if it is absent. {:ok, "OK"} when set, {:ok, nil} when it was already there.
+  """
+  def set_nx(key, ttl_seconds) do
+    command(["SET", key, "1", "NX", "EX", ttl_seconds])
+  end
+
   @doc """
   Increment a counter with TTL. Used for rate limiting.
   Returns the new count or nil on failure.

realtime/lib/realtime_web/channels/account_channel.ex

@@ -31,7 +31,9 @@ defmodule RealtimeWeb.AccountChannel do
       # Verify user has access to this email account via organization membership
       checked_at = System.monotonic_time(:millisecond)
 
-      case Auth.check_email_account_access(user_id, account_id) do
+      case user_id
+           |> Auth.check_email_account_access(account_id)
+           |> Auth.confine(socket.assigns[:credential]) do
         {:ok, member} ->
           Logger.debug("User #{user_id} joined account:#{account_id}")
 
@@ -90,7 +92,9 @@ defmodule RealtimeWeb.AccountChannel do
     ChannelGuard.refresh_authorization(
       socket,
       fn ->
-        Auth.check_email_account_access(socket.assigns.user_id, socket.assigns.account_id)
+        socket.assigns.user_id
+        |> Auth.check_email_account_access(socket.assigns.account_id)
+        |> Auth.confine(socket.assigns[:credential])
       end,
       force
     )

realtime/lib/realtime_web/channels/admin_channel.ex

@@ -7,7 +7,8 @@ defmodule RealtimeWeb.AdminChannel do
   to connected admins unfiltered, so the admin app can render a live event
   feed and invalidate its queries without polling.
 
-  Join is gated hard: JWT sockets only (no API keys) and the user must have
+  Join is gated hard: JWT sockets only (no API keys), from a session that
+  verified a second factor, and the user must have
   `users.admin_permissions > 0`. There is no presence tracking here — this
   is an internal operations surface, not a collaboration space.
   """
@@ -36,25 +37,35 @@ defmodule RealtimeWeb.AdminChannel do
   defp authorize_join(socket) do
     user_id = socket.assigns.user_id
 
-    if Map.get(socket.assigns, :auth_type) != :jwt do
-      ChannelGuard.join_error("jwt_required")
-    else
-      case Auth.check_admin(user_id) do
-        {:ok, admin} ->
-          Logger.debug("Admin #{user_id} joined the platform event channel")
+    cond do
+      Map.get(socket.assigns, :auth_type) != :jwt ->
+        ChannelGuard.join_error("jwt_required")
 
-          socket = assign(socket, :admin_permissions, Map.get(admin, :permissions, 0))
+      # Same rule as the admin API: only a session that verified a second factor.
+      Map.get(socket.assigns, :mfa_verified) != true ->
+        ChannelGuard.join_error("admin_mfa_required")
 
-          {:ok,
-           %{
-             heartbeat_interval_ms: 25_000,
-             server_timeout_ms: 60_000
-           }, socket}
+      true ->
+        authorize_admin(socket, user_id)
+    end
+  end
 
-        {:error, reason} ->
-          Logger.warning("Admin channel join rejected for #{user_id}: #{inspect(reason)}")
-          ChannelGuard.join_error("not_an_admin")
-      end
+  defp authorize_admin(socket, user_id) do
+    case Auth.check_admin(user_id) do
+      {:ok, admin} ->
+        Logger.debug("Admin #{user_id} joined the platform event channel")
+
+        socket = assign(socket, :admin_permissions, Map.get(admin, :permissions, 0))
+
+        {:ok,
+         %{
+           heartbeat_interval_ms: 25_000,
+           server_timeout_ms: 60_000
+         }, socket}
+
+      {:error, reason} ->
+        Logger.warning("Admin channel join rejected for #{user_id}: #{inspect(reason)}")
+        ChannelGuard.join_error("not_an_admin")
     end
   end
 

realtime/lib/realtime_web/channels/bulk_channel.ex

@@ -16,7 +16,9 @@ defmodule RealtimeWeb.BulkChannel do
 
   require Logger
 
+  alias Realtime.Auth
   alias RealtimeWeb.ChannelGuard
+  alias RealtimeWeb.OrgChannel
 
   @impl true
   def join("bulk:" <> operation_id, _params, socket) do
@@ -51,13 +53,30 @@ defmodule RealtimeWeb.BulkChannel do
 
   @doc false
   # An event with no owner belongs to nobody and is never delivered here.
-  def own_event?(%{"user_id" => user_id}, socket)
+  def own_event?(%{"user_id" => user_id} = event, socket)
       when is_binary(user_id) and user_id != "" do
-    user_id == socket.assigns.user_id
+    user_id == socket.assigns.user_id and credential_reaches?(event, socket)
   end
 
   def own_event?(_event, _socket), do: false
 
+  # An API key or OAuth token sees only its own workspace's operations, within its scopes.
+  defp credential_reaches?(event, socket) do
+    case socket.assigns[:credential] do
+      nil ->
+        true
+
+      %{organization_id: credential_org, permissions: mask} ->
+        org_id = event["org_id"] || event["organization_id"]
+
+        is_binary(org_id) and org_id != "" and
+          Auth.matches_org?(%{organization_id: credential_org}, org_id) and
+          socket
+          |> ChannelGuard.remember_authorization(%{permissions: mask})
+          |> OrgChannel.can_see_event?(event)
+    end
+  end
+
   defp valid_uuid?(id) do
     case UUID.info(id) do
       {:ok, _} -> true

realtime/lib/realtime_web/channels/campaign_channel.ex

@@ -31,7 +31,9 @@ defmodule RealtimeWeb.CampaignChannel do
       # Verify user has access to this campaign via organization membership
       checked_at = System.monotonic_time(:millisecond)
 
-      case Auth.check_campaign_access(user_id, campaign_id) do
+      case user_id
+           |> Auth.check_campaign_access(campaign_id)
+           |> Auth.confine(socket.assigns[:credential]) do
         {:ok, member} ->
           Logger.debug("User #{user_id} joined campaign:#{campaign_id}")
 
@@ -90,7 +92,9 @@ defmodule RealtimeWeb.CampaignChannel do
     ChannelGuard.refresh_authorization(
       socket,
       fn ->
-        Auth.check_campaign_access(socket.assigns.user_id, socket.assigns.campaign_id)
+        socket.assigns.user_id
+        |> Auth.check_campaign_access(socket.assigns.campaign_id)
+        |> Auth.confine(socket.assigns[:credential])
       end,
       force
     )

realtime/lib/realtime_web/channels/org_channel.ex

@@ -57,7 +57,9 @@ defmodule RealtimeWeb.OrgChannel do
     user_id = socket.assigns.user_id
     checked_at = System.monotonic_time(:millisecond)
 
-    case Auth.check_org_membership(user_id, org_id) do
+    case user_id
+         |> Auth.check_org_membership(org_id)
+         |> Auth.confine(socket.assigns[:credential]) do
       {:ok, member} ->
         Logger.debug("User #{user_id} joined org channel #{org_id}")
 
@@ -280,7 +282,9 @@ defmodule RealtimeWeb.OrgChannel do
     case ChannelGuard.refresh_authorization(
            socket,
            fn ->
-             Auth.check_org_membership(socket.assigns.user_id, socket.assigns.org_id)
+             socket.assigns.user_id
+             |> Auth.check_org_membership(socket.assigns.org_id)
+             |> Auth.confine(socket.assigns[:credential])
            end,
            force
          ) do

realtime/lib/realtime_web/channels/user_channel.ex

@@ -128,7 +128,9 @@ defmodule RealtimeWeb.UserChannel do
   end
 
   defp reread_event_authorization(socket, event, key, cache, now) do
-    case lookup_event_authorization(socket.assigns.user_id, key) do
+    case socket.assigns.user_id
+         |> lookup_event_authorization(key)
+         |> Auth.confine(socket.assigns[:credential]) do
       {:ok, member} ->
         if ChannelGuard.fresh?(now, System.monotonic_time(:millisecond)) do
           cache =

realtime/lib/realtime_web/channels/user_socket.ex

@@ -25,8 +25,8 @@ defmodule RealtimeWeb.UserSocket do
   channel("org:*", RealtimeWeb.OrgChannel)
   channel("admin:*", RealtimeWeb.AdminChannel)
 
-  # API keys and OAuth tokens belong in this header; the query string, which
-  # proxies record, is a deprecated fallback for them kept for existing clients.
+  # API keys and OAuth tokens belong in this header. The query string, which
+  # proxies record, carries them only where REALTIME_ALLOW_QUERY_CREDENTIALS is set.
   @credential_header "x-warmbly-token"
 
   @impl true
@@ -47,11 +47,15 @@ defmodule RealtimeWeb.UserSocket do
         {:ok, header}
 
       present?(query) and Auth.long_lived?(query) ->
-        Logger.info(
-          "Socket credential in the query string; send it in the x-warmbly-token header"
-        )
+        if Application.get_env(:realtime, :allow_query_credentials, false) do
+          Logger.info(
+            "Socket credential in the query string; send it in the x-warmbly-token header"
+          )
 
-        {:ok, query}
+          {:ok, query}
+        else
+          {:error, :credential_in_query}
+        end
 
       present?(query) ->
         {:ok, query}
@@ -75,7 +79,7 @@ defmodule RealtimeWeb.UserSocket do
   defp authenticate(token, socket, connect_info) do
     ip = client_ip(token, connect_info)
 
-    with {:ok, user_id, auth_type} <- Auth.verify_token(token, ip: ip),
+    with {:ok, user_id, auth_type, context} <- Auth.verify_token(token, ip: ip),
          limits <- Subscription.get_limits(user_id),
          :ok <- check_connection_limit(user_id, ip, limits),
          :ok <- check_rate_limit(user_id, limits) do
@@ -87,6 +91,10 @@ defmodule RealtimeWeb.UserSocket do
         |> assign(:ip_address, ip)
         |> assign(:auth_type, auth_type)
         |> assign(:rate_limits, limits)
+        # nil for a ticket; an API key or OAuth token's workspace and scope grant otherwise.
+        |> assign(:credential, Map.get(context, :credential))
+        |> assign(:session_id, Map.get(context, :session_id))
+        |> assign(:mfa_verified, Map.get(context, :mfa_verified) == true)
 
       {:ok, socket}
     else

realtime/test/realtime/credential_confine_test.exs

@@ -0,0 +1,140 @@
+defmodule Realtime.CredentialConfineTest do
+  @moduledoc """
+  An API key or OAuth token socket reaches its own workspace, and there only
+  what its scopes grant.
+  """
+
+  use RealtimeWeb.ChannelCase, async: false
+
+  alias Realtime.Auth
+  alias Realtime.Test.AuthRepo
+  alias RealtimeWeb.BulkChannel
+  alias RealtimeWeb.OrgChannel
+
+  @org "0b7d5c1e-6a1f-4c1e-9d5a-2f0c8e4b7a10"
+  @other_org "1c8e6d2f-7b2a-4d2f-8e6b-3a1d9f5c8b21"
+
+  # Every member permission, so only the credential narrows anything.
+  @all_member_permissions 8191
+
+  @read_emails 1
+  @read_campaigns 2
+  @read_contacts 4
+  @read_unibox 8
+  @read_analytics 16
+  @realtime_subscribe 2048
+
+  defp grant(scopes, org \\ @org),
+    do: Auth.credential_grant(%{organization_id: org, scopes: scopes})
+
+  defp member(org \\ @org),
+    do: %{organization_id: org, permissions: @all_member_permissions, scope: nil}
+
+  test "each read scope reaches the member permission behind it" do
+    perm = fn scopes -> grant(scopes).permissions end
+
+    assert perm.(@realtime_subscribe) == 0
+    assert perm.(@read_emails) == Auth.permission(:manage_emails)
+    assert perm.(@read_campaigns) == Auth.permission(:view_campaigns)
+    assert perm.(@read_contacts) == Auth.permission(:view_contacts)
+    assert perm.(@read_unibox) == Auth.permission(:access_unibox)
+    assert perm.(@read_analytics) == Auth.permission(:view_analytics)
+  end
+
+  test "no scope reaches team, billing or settings" do
+    mask = grant(0xFFFFFFFF).permissions
+
+    for perm <- [:manage_team, :manage_billing, :manage_settings, :transfer_ownership] do
+      refute Auth.has_permission?(%{permissions: mask}, Auth.permission(perm))
+    end
+  end
+
+  test "a credential keeps the member permissions its scopes grant, in its own workspace" do
+    credential = grant(@read_campaigns)
+
+    assert {:ok, %{permissions: permissions}} = Auth.confine({:ok, member()}, credential)
+    assert permissions == Auth.permission(:view_campaigns)
+  end
+
+  test "a credential never reaches more than the member holds" do
+    credential = grant(@read_campaigns + @read_unibox)
+    viewer = %{member() | permissions: Auth.permission(:view_campaigns)}
+
+    assert {:ok, %{permissions: permissions}} = Auth.confine({:ok, viewer}, credential)
+    assert permissions == Auth.permission(:view_campaigns)
+  end
+
+  test "a credential reaches nothing in another workspace" do
+    assert {:error, :not_a_member} =
+             Auth.confine({:ok, member(@other_org)}, grant(@read_campaigns))
+  end
+
+  test "a credential reaches nothing outside a workspace" do
+    assert {:error, :not_a_member} =
+             Auth.confine({:ok, %{permissions: 65535, scope: nil}}, grant(@read_campaigns))
+  end
+
+  test "a ticket socket keeps the member's own permissions" do
+    assert Auth.confine({:ok, member(@other_org)}, nil) == {:ok, member(@other_org)}
+    assert Auth.confine({:error, :forbidden}, grant(@read_campaigns)) == {:error, :forbidden}
+  end
+
+  describe "org channel" do
+    setup do
+      start_supervised!(AuthRepo)
+
+      AuthRepo.respond_with(fn _sql, _params ->
+        {:ok, %{rows: [[uuid(), "admin", @all_member_permissions, true, true, nil]]}}
+      end)
+
+      :ok
+    end
+
+    defp credential_socket(scopes, org \\ @org),
+      do: user_socket(uuid(), %{auth_type: :api_key, credential: grant(scopes, org)})
+
+    test "a credential joins only its own workspace's channel" do
+      assert {:ok, _, _} =
+               subscribe_and_join(credential_socket(@read_campaigns), OrgChannel, "org:#{@org}")
+
+      assert {:error, %{reason: "Not a member of this organization"}} =
+               subscribe_and_join(
+                 credential_socket(@read_campaigns),
+                 OrgChannel,
+                 "org:#{@other_org}"
+               )
+    end
+
+    test "a credential receives only the event families its scopes grant" do
+      {:ok, _, socket} =
+        subscribe_and_join(credential_socket(@read_campaigns), OrgChannel, "org:#{@org}")
+
+      assert socket.assigns.permissions == Auth.permission(:view_campaigns)
+      assert OrgChannel.can_see_event?(socket, %{"event_type" => "EMAIL_SENT"})
+      refute OrgChannel.can_see_event?(socket, %{"event_type" => "INBOX_EMAIL_RECEIVED"})
+      refute OrgChannel.can_see_event?(socket, %{"event_type" => "CONTACT_CREATED"})
+      refute OrgChannel.can_see_event?(socket, %{"event_type" => "MEMBER_JOINED"})
+      refute OrgChannel.can_see_event?(socket, %{"event_type" => "BILLING_UPDATED"})
+    end
+  end
+
+  describe "bulk channel" do
+    defp bulk_event(user_id, org) do
+      %{"event_type" => "BULK_PROGRESS", "user_id" => user_id, "org_id" => org}
+    end
+
+    test "a credential receives its holder's operations in its own workspace only" do
+      user_id = uuid()
+      socket = user_socket(user_id, %{auth_type: :api_key, credential: grant(@read_contacts)})
+
+      assert BulkChannel.own_event?(bulk_event(user_id, @org), socket)
+      refute BulkChannel.own_event?(bulk_event(user_id, @other_org), socket)
+      refute BulkChannel.own_event?(Map.delete(bulk_event(user_id, @org), "org_id"), socket)
+    end
+
+    test "a ticket socket receives its own operations wherever they ran" do
+      user_id = uuid()
+      assert BulkChannel.own_event?(bulk_event(user_id, @other_org), user_socket(user_id))
+    end
+  end
+end

realtime/test/realtime/ws_ticket_test.exs

@@ -0,0 +1,87 @@
+defmodule Realtime.WsTicketTest do
+  @moduledoc """
+  A socket ticket opens one socket, and only while the session it names is live.
+  """
+
+  use ExUnit.Case, async: false
+
+  alias Realtime.Auth
+  alias Realtime.Test.AuthRepo
+  alias Realtime.Test.Counter
+
+  @user "0b7d5c1e-6a1f-4c1e-9d5a-2f0c8e4b7a10"
+  @session "1c8e6d2f-7b2a-4d2f-8e6b-3a1d9f5c8b21"
+
+  setup do
+    Counter.reset()
+    start_supervised!(AuthRepo)
+    :ok
+  end
+
+  defp ticket(extra \\ %{}) do
+    claims =
+      Map.merge(
+        %{
+          "sub" => @user,
+          "sid" => UUID.uuid4(),
+          "nonce" => "nonce",
+          "purpose" => "ws",
+          "exp" => System.system_time(:second) + 600
+        },
+        extra
+      )
+
+    JOSE.JWT.sign(JOSE.JWK.from_oct("test_jwt_secret"), %{"alg" => "HS256"}, claims)
+    |> JOSE.JWS.compact()
+    |> elem(1)
+  end
+
+  defp live_session(mfa_verified) do
+    AuthRepo.respond_with(fn sql, _params ->
+      if String.contains?(sql, "FROM sessions"),
+        do: {:ok, %{rows: [[mfa_verified]]}},
+        else: {:ok, %{rows: []}}
+    end)
+  end
+
+  test "a ticket from a live session carries that session's second factor" do
+    live_session(true)
+
+    assert {:ok, @user, :jwt, %{session_id: @session, mfa_verified: true}} =
+             Auth.verify_jwt(ticket(%{"session_id" => @session}))
+
+    live_session(false)
+
+    assert {:ok, @user, :jwt, %{mfa_verified: false}} =
+             Auth.verify_jwt(ticket(%{"session_id" => @session}))
+  end
+
+  test "a ticket from a revoked or expired session is refused" do
+    AuthRepo.respond_with(fn _sql, _params -> {:ok, %{rows: []}} end)
+
+    assert {:error, :session_revoked} = Auth.verify_jwt(ticket(%{"session_id" => @session}))
+  end
+
+  test "a session lookup that fails refuses the ticket" do
+    AuthRepo.respond_with(fn _sql, _params -> {:error, :timeout} end)
+
+    assert {:error, :database_error} = Auth.verify_jwt(ticket(%{"session_id" => @session}))
+  end
+
+  test "a ticket opens one socket" do
+    live_session(true)
+    token = ticket(%{"session_id" => @session})
+
+    assert {:ok, @user, :jwt, _} = Auth.verify_jwt(token)
+    assert {:error, :ticket_used} = Auth.verify_jwt(token)
+  end
+
+  test "a ticket that names no session reaches no second factor" do
+    assert {:ok, @user, :jwt, %{session_id: nil, mfa_verified: false}} = Auth.verify_jwt(ticket())
+    assert AuthRepo.queries() == []
+  end
+
+  test "a ticket with no id is refused" do
+    assert {:error, :invalid_claims} = Auth.verify_jwt(ticket(%{"sid" => nil}))
+  end
+end

realtime/test/realtime_web/channels/admin_channel_test.exs

@@ -12,7 +12,7 @@ defmodule RealtimeWeb.AdminChannelTest do
 
   test "each distinct event is delivered once and leaving removes the subscription" do
     Process.flag(:trap_exit, true)
-    {:ok, _, socket} = subscribe_and_join(user_socket(uuid()), AdminChannel, "admin:platform")
+    {:ok, _, socket} = subscribe_and_join(admin_socket(), AdminChannel, "admin:platform")
     :sys.get_state(socket.channel_pid)
 
     assert Enum.count(Registry.lookup(Realtime.PubSub, socket.topic), fn {pid, _} ->
@@ -37,13 +37,29 @@ defmodule RealtimeWeb.AdminChannelTest do
     AuthRepo.respond_with(fn _sql, _params -> {:ok, %{rows: []}} end)
 
     assert {:error, %{reason: "not_an_admin"}} =
-             subscribe_and_join(user_socket(uuid()), AdminChannel, "admin:platform")
+             subscribe_and_join(admin_socket(), AdminChannel, "admin:platform")
 
     assert {:error, %{reason: "jwt_required"}} =
              subscribe_and_join(
-               user_socket(uuid(), %{auth_type: :api_key}),
+               user_socket(uuid(), %{auth_type: :api_key, mfa_verified: true}),
                AdminChannel,
                "admin:platform"
              )
   end
+
+  test "an admin whose session did not verify a second factor cannot join" do
+    assert {:error, %{reason: "admin_mfa_required"}} =
+             subscribe_and_join(user_socket(uuid()), AdminChannel, "admin:platform")
+
+    assert {:error, %{reason: "admin_mfa_required"}} =
+             subscribe_and_join(
+               user_socket(uuid(), %{mfa_verified: false}),
+               AdminChannel,
+               "admin:platform"
+             )
+
+    assert AuthRepo.queries() == []
+  end
+
+  defp admin_socket, do: user_socket(uuid(), %{mfa_verified: true})
 end

realtime/test/realtime_web/channels/user_socket_credential_test.exs

@@ -1,5 +1,5 @@
 defmodule RealtimeWeb.UserSocketCredentialTest do
-  use ExUnit.Case, async: true
+  use ExUnit.Case, async: false
 
   alias RealtimeWeb.UserSocket
 
@@ -9,7 +9,18 @@ defmodule RealtimeWeb.UserSocketCredentialTest do
     assert UserSocket.credential(%{"token" => @ticket}, %{}) == {:ok, @ticket}
   end
 
-  test "an API key or OAuth token in the query string is still accepted, deprecated" do
+  test "an API key or OAuth token in the query string is refused by default" do
+    assert UserSocket.credential(%{"token" => "wmbly_abc"}, %{}) ==
+             {:error, :credential_in_query}
+
+    assert UserSocket.credential(%{"token" => "wmat_abc"}, %{}) ==
+             {:error, :credential_in_query}
+  end
+
+  test "an API key in the query string is read when the operator allows it" do
+    Application.put_env(:realtime, :allow_query_credentials, true)
+    on_exit(fn -> Application.delete_env(:realtime, :allow_query_credentials) end)
+
     assert UserSocket.credential(%{"token" => "wmbly_abc"}, %{}) == {:ok, "wmbly_abc"}
     assert UserSocket.credential(%{"token" => "wmat_abc"}, %{}) == {:ok, "wmat_abc"}
   end

realtime/test/support/counter.ex

@@ -26,6 +26,18 @@ defmodule Realtime.Test.Counter do
     Agent.get(__MODULE__, &get_in(&1, [:counts, key])) || 0
   end
 
+  def set_nx(key, _ttl_seconds) do
+    Agent.get_and_update(__MODULE__, fn
+      %{down: true} = state ->
+        {{:error, :redis_unavailable}, state}
+
+      %{counts: counts} = state ->
+        if Map.has_key?(counts, key),
+          do: {{:ok, nil}, state},
+          else: {{:ok, "OK"}, %{state | counts: Map.put(counts, key, 1)}}
+    end)
+  end
+
   def incr_with_ttl(key, _ttl_seconds) do
     Agent.get_and_update(__MODULE__, fn
       %{down: true} = state ->