diff --git a/AGENTS.md b/AGENTS.md index 65080fe5e..2a657e42c 100644 --- a/AGENTS.md +++ b/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 diff --git a/docs/content/docs/api/realtime.mdx b/docs/content/docs/api/realtime.mdx index e9b5be423..59b8513e3 100644 --- a/docs/content/docs/api/realtime.mdx +++ b/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 | diff --git a/docs/content/docs/development/configuration.mdx b/docs/content/docs/development/configuration.mdx index fa7cfbddf..b1eb23cdd 100644 --- a/docs/content/docs/development/configuration.mdx +++ b/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` | diff --git a/internal/api/handler/socket.go b/internal/api/handler/socket.go index 185a5bd07..e8ef0754a 100644 --- a/internal/api/handler/socket.go +++ b/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 diff --git a/internal/api/handler/socket_test.go b/internal/api/handler/socket_test.go index 822d88035..17b78f8c9 100644 --- a/internal/api/handler/socket_test.go +++ b/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 { diff --git a/internal/app/socket/gen.go b/internal/app/socket/gen.go index 9709a2362..24e30073e 100644 --- a/internal/app/socket/gen.go +++ b/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() diff --git a/internal/app/socket/service.go b/internal/app/socket/service.go index aef26e7d8..9a0ff017f 100644 --- a/internal/app/socket/service.go +++ b/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 { diff --git a/internal/app/token/gen.go b/internal/app/token/gen.go index abe5e8e31..5cc9471e8 100644 --- a/internal/app/token/gen.go +++ b/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() diff --git a/internal/app/token/purpose_test.go b/internal/app/token/purpose_test.go index 2216c0206..806a4bbdb 100644 --- a/internal/app/token/purpose_test.go +++ b/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. diff --git a/internal/app/token/service.go b/internal/app/token/service.go index 60487c175..39bc8894c 100644 --- a/internal/app/token/service.go +++ b/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. diff --git a/realtime/config/runtime.exs b/realtime/config/runtime.exs index fd6891111..0293a1f64 100644 --- a/realtime/config/runtime.exs +++ b/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 = diff --git a/realtime/config/test.exs b/realtime/config/test.exs index e6f3a66d9..8c7189419 100644 --- a/realtime/config/test.exs +++ b/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 diff --git a/realtime/lib/realtime/api_key.ex b/realtime/lib/realtime/api_key.ex index cdbad32b0..1ada8c472 100644 --- a/realtime/lib/realtime/api_key.ex +++ b/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} diff --git a/realtime/lib/realtime/auth.ex b/realtime/lib/realtime/auth.ex index 796cf11ab..588179efd 100644 --- a/realtime/lib/realtime/auth.ex +++ b/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 diff --git a/realtime/lib/realtime/oauth_token.ex b/realtime/lib/realtime/oauth_token.ex index 69b2e4eae..6dad9af24 100644 --- a/realtime/lib/realtime/oauth_token.ex +++ b/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 diff --git a/realtime/lib/realtime/redis.ex b/realtime/lib/realtime/redis.ex index a4747c5e4..9373387c5 100644 --- a/realtime/lib/realtime/redis.ex +++ b/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. diff --git a/realtime/lib/realtime_web/channels/account_channel.ex b/realtime/lib/realtime_web/channels/account_channel.ex index 4bd5dff60..293044397 100644 --- a/realtime/lib/realtime_web/channels/account_channel.ex +++ b/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 ) diff --git a/realtime/lib/realtime_web/channels/admin_channel.ex b/realtime/lib/realtime_web/channels/admin_channel.ex index 476c78865..8ed3b4fe7 100644 --- a/realtime/lib/realtime_web/channels/admin_channel.ex +++ b/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 diff --git a/realtime/lib/realtime_web/channels/bulk_channel.ex b/realtime/lib/realtime_web/channels/bulk_channel.ex index 8d37eab88..99ac7796e 100644 --- a/realtime/lib/realtime_web/channels/bulk_channel.ex +++ b/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 diff --git a/realtime/lib/realtime_web/channels/campaign_channel.ex b/realtime/lib/realtime_web/channels/campaign_channel.ex index 6bb0c2e5b..67c1f3bc6 100644 --- a/realtime/lib/realtime_web/channels/campaign_channel.ex +++ b/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 ) diff --git a/realtime/lib/realtime_web/channels/org_channel.ex b/realtime/lib/realtime_web/channels/org_channel.ex index 33a1eb014..fc1a9ec71 100644 --- a/realtime/lib/realtime_web/channels/org_channel.ex +++ b/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 diff --git a/realtime/lib/realtime_web/channels/user_channel.ex b/realtime/lib/realtime_web/channels/user_channel.ex index 1e19461a1..b9942b5e4 100644 --- a/realtime/lib/realtime_web/channels/user_channel.ex +++ b/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 = diff --git a/realtime/lib/realtime_web/channels/user_socket.ex b/realtime/lib/realtime_web/channels/user_socket.ex index bf6e6ba38..6b002bdaf 100644 --- a/realtime/lib/realtime_web/channels/user_socket.ex +++ b/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 diff --git a/realtime/test/realtime/credential_confine_test.exs b/realtime/test/realtime/credential_confine_test.exs new file mode 100644 index 000000000..b7bb74204 --- /dev/null +++ b/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 diff --git a/realtime/test/realtime/ws_ticket_test.exs b/realtime/test/realtime/ws_ticket_test.exs new file mode 100644 index 000000000..135c4a5cb --- /dev/null +++ b/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 diff --git a/realtime/test/realtime_web/channels/admin_channel_test.exs b/realtime/test/realtime_web/channels/admin_channel_test.exs index 53107ab7d..8c92fd45a 100644 --- a/realtime/test/realtime_web/channels/admin_channel_test.exs +++ b/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 diff --git a/realtime/test/realtime_web/channels/user_socket_credential_test.exs b/realtime/test/realtime_web/channels/user_socket_credential_test.exs index b2d9d1609..6a0564c2e 100644 --- a/realtime/test/realtime_web/channels/user_socket_credential_test.exs +++ b/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 diff --git a/realtime/test/support/counter.ex b/realtime/test/support/counter.ex index 314249444..9514adb64 100644 --- a/realtime/test/support/counter.ex +++ b/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 ->