fix: on bad payload, block join

Author: filipecabaco

lib/realtime_web/channels/realtime_channel.ex

@@ -5,8 +5,8 @@ defmodule RealtimeWeb.RealtimeChannel do
   use RealtimeWeb, :channel
   use RealtimeWeb.RealtimeChannel.Logging
 
-  alias RealtimeWeb.SocketDisconnect
   alias DBConnection.Backoff
+  alias Phoenix.Socket
 
   alias Realtime.Crypto
   alias Realtime.GenCounter
@@ -22,11 +22,14 @@ defmodule RealtimeWeb.RealtimeChannel do
   alias Realtime.Tenants.Connect
 
   alias RealtimeWeb.Channels.Payloads.Join
+  alias RealtimeWeb.Channels.Payloads.Config
+  alias RealtimeWeb.Channels.Payloads.PostgresChange
   alias RealtimeWeb.ChannelsAuthorization
   alias RealtimeWeb.RealtimeChannel.BroadcastHandler
   alias RealtimeWeb.RealtimeChannel.MessageDispatcher
   alias RealtimeWeb.RealtimeChannel.PresenceHandler
   alias RealtimeWeb.RealtimeChannel.Tracker
+  alias RealtimeWeb.SocketDisconnect
 
   @confirm_token_ms_interval :timer.minutes(5)
 
@@ -47,20 +50,11 @@ defmodule RealtimeWeb.RealtimeChannel do
     Logger.metadata(external_id: tenant_id, project: tenant_id)
     Logger.put_process_level(self(), log_level)
 
-    socket =
-      socket
-      |> assign_access_token(params)
-      |> assign_counter()
-      |> assign_presence_counter()
-      |> assign(:private?, !!params["config"]["private"])
-      |> assign(:policies, nil)
-
-    case Join.validate(params) do
-      {:ok, _join} -> nil
-      {:error, :invalid_join_payload, errors} -> log_error(socket, "InvalidJoinPayload", errors)
-    end
+    # We always need to assign the access token so we can get the logs metadata working as expected
+    socket = assign_access_token(socket, params)
 
-    with :ok <- SignalHandler.shutdown_in_progress?(),
+    with {:ok, %Socket{} = socket, %Join{} = configuration} <- configure_socket(socket, params),
+         :ok <- SignalHandler.shutdown_in_progress?(),
          :ok <- only_private?(tenant_id, socket),
          :ok <- limit_joins(socket),
          :ok <- limit_channels(socket),
@@ -70,7 +64,6 @@ defmodule RealtimeWeb.RealtimeChannel do
          {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant_id),
          {:ok, socket} <- maybe_assign_policies(sub_topic, db_conn, socket) do
       tenant_topic = Tenants.tenant_topic(tenant_id, sub_topic, !socket.assigns.private?)
-
       # fastlane subscription
       metadata =
         MessageDispatcher.fastlane_metadata(transport_pid, serializer, topic, socket.assigns.log_level, tenant_id)
@@ -79,15 +72,11 @@ defmodule RealtimeWeb.RealtimeChannel do
 
       Phoenix.PubSub.subscribe(Realtime.PubSub, "realtime:operations:" <> tenant_id)
 
-      is_new_api = new_api?(params)
-      # TODO: Default will be moved to false in the future
-      presence_enabled? =
-        case get_in(params, ["config", "presence", "enabled"]) do
-          enabled when is_boolean(enabled) -> enabled
-          _ -> true
-        end
+      is_new_api = new_api?(configuration)
 
-      pg_change_params = pg_change_params(is_new_api, params, channel_pid, claims, sub_topic)
+      presence_enabled? = Join.presence_enabled?(configuration)
+
+      pg_change_params = pg_change_params(is_new_api, configuration, channel_pid, claims, sub_topic)
 
       opts = %{
         is_new_api: is_new_api,
@@ -104,13 +93,13 @@ defmodule RealtimeWeb.RealtimeChannel do
       state = %{postgres_changes: add_id_to_postgres_changes(pg_change_params)}
 
       assigns = %{
-        ack_broadcast: !!params["config"]["broadcast"]["ack"],
+        ack_broadcast: Join.ack_broadcast?(configuration),
         confirm_token_ref: confirm_token_ref,
         is_new_api: is_new_api,
         pg_sub_ref: nil,
         pg_change_params: pg_change_params,
-        presence_key: presence_key(params),
-        self_broadcast: !!params["config"]["broadcast"]["self"],
+        presence_key: Join.presence_key(configuration),
+        self_broadcast: Join.self_broadcast?(configuration),
         tenant_topic: tenant_topic,
         channel_name: sub_topic,
         presence_enabled?: presence_enabled?
@@ -124,6 +113,9 @@ defmodule RealtimeWeb.RealtimeChannel do
 
       {:ok, state, assign(socket, assigns)}
     else
+      {:error, :invalid_join_payload, errors, socket} ->
+        log_error(socket, "InvalidJoinPayload", errors)
+
       {:error, :expired_token, msg} ->
         maybe_log_warning(socket, "InvalidJWTToken", msg)
 
@@ -200,6 +192,23 @@ defmodule RealtimeWeb.RealtimeChannel do
     end
   end
 
+  defp configure_socket(socket, params) do
+    case Join.validate(params) do
+      {:ok, configuration} ->
+        socket =
+          socket
+          |> assign_counter()
+          |> assign_presence_counter()
+          |> assign(:private?, Join.private?(configuration))
+          |> assign(:policies, nil)
+
+        {:ok, socket, configuration}
+
+      {:error, :invalid_join_payload, errors} ->
+        {:error, :invalid_join_payload, errors, socket}
+    end
+  end
+
   @impl true
   def handle_info(:update_rate_counter, %{assigns: %{limits: %{max_events_per_second: max}}} = socket) do
     count(socket)
@@ -537,40 +546,24 @@ defmodule RealtimeWeb.RealtimeChannel do
 
   defp count(%{assigns: %{rate_counter: counter}}), do: GenCounter.add(counter.id)
 
-  defp presence_key(params) do
-    case params["config"]["presence"]["key"] do
-      key when is_binary(key) and key != "" -> key
-      _ -> UUID.uuid1()
-    end
-  end
-
-  defp assign_access_token(%{assigns: %{headers: headers}} = socket, params) do
-    access_token = Map.get(params, "access_token") || Map.get(params, "user_token")
+  defp assign_access_token(socket, params) do
+    %{assigns: %{tenant_token: tenant_token, headers: headers}} = socket
     {_, header} = Enum.find(headers, {nil, nil}, fn {k, _} -> k == "x-api-key" end)
 
-    case access_token do
-      nil -> assign(socket, :access_token, header)
-      "sb_" <> _ -> assign(socket, :access_token, header)
-      _ -> handle_access_token(socket, params)
-    end
-  end
-
-  defp assign_access_token(socket, params), do: handle_access_token(socket, params)
+    access_token = Map.get(params, "access_token")
+    user_token = Map.get(params, "user_token")
 
-  defp handle_access_token(%{assigns: %{tenant_token: _tenant_token}} = socket, %{"user_token" => user_token})
-       when is_binary(user_token) do
-    assign(socket, :access_token, user_token)
-  end
+    access_token =
+      cond do
+        is_binary(access_token) and !String.starts_with?(access_token, "sb_") -> access_token
+        is_binary(user_token) and !String.starts_with?(user_token, "sb_") -> user_token
+        is_binary(tenant_token) and !String.starts_with?(tenant_token, "sb_") -> tenant_token
+        true -> header
+      end
 
-  defp handle_access_token(%{assigns: %{tenant_token: _tenant_token}} = socket, %{"access_token" => access_token})
-       when is_binary(access_token) do
     assign(socket, :access_token, access_token)
   end
 
-  defp handle_access_token(%{assigns: %{tenant_token: tenant_token}} = socket, _params) when is_binary(tenant_token) do
-    assign(socket, :access_token, tenant_token)
-  end
-
   defp confirm_token(%{assigns: assigns}) do
     %{jwt_secret: jwt_secret, access_token: access_token} = assigns
 
@@ -637,28 +630,30 @@ defmodule RealtimeWeb.RealtimeChannel do
     })
   end
 
-  defp new_api?(%{"config" => _}), do: true
+  defp new_api?(%Join{config: config}) when not is_nil(config), do: true
   defp new_api?(_), do: false
 
-  defp pg_change_params(true, params, channel_pid, claims, _) do
-    case get_in(params, ["config", "postgres_changes"]) do
-      [_ | _] = params_list ->
-        params_list
-        |> Enum.reject(&is_nil/1)
-        |> Enum.map(fn params ->
-          %{
-            id: UUID.uuid1(),
-            channel_pid: channel_pid,
-            claims: claims,
-            params: params
-          }
-        end)
-
-      _ ->
-        []
-    end
+  defp pg_change_params(true, %Join{config: %Config{postgres_changes: postgres_changes}}, channel_pid, claims, _)
+       when not is_nil(postgres_changes) do
+    postgres_changes
+    |> Enum.reject(&is_nil/1)
+    |> Enum.map(fn %PostgresChange{table: table, event: event, schema: schema, filter: filter} ->
+      params =
+        %{"table" => table, "filter" => filter, "schema" => schema, "event" => event}
+        |> Enum.reject(fn {_, v} -> is_nil(v) end)
+        |> Map.new()
+
+      %{
+        id: UUID.uuid1(),
+        channel_pid: channel_pid,
+        claims: claims,
+        params: params
+      }
+    end)
   end
 
+  defp pg_change_params(true, _, _, _, _), do: []
+
   defp pg_change_params(false, _, channel_pid, claims, sub_topic) do
     params =
       case String.split(sub_topic, ":", parts: 3) do

test/integration/rt_channel_test.exs

@@ -420,28 +420,18 @@ defmodule Realtime.Integration.RtChannelTest do
                      500
     end
 
-    test "handle nil postgres changes params as empty param changes", %{tenant: tenant} do
+    test "nil postgres changes params identified as error", %{tenant: tenant} do
       {socket, _} = get_connection(tenant)
       topic = "realtime:any"
       config = %{postgres_changes: [nil]}
 
-      WebsocketClient.join(socket, topic, %{config: config})
-
-      assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^topic}, 200
-      assert_receive %Phoenix.Socket.Message{event: "presence_state", payload: %{}, topic: ^topic}, 500
+      log =
+        capture_log(fn ->
+          WebsocketClient.join(socket, topic, %{config: config})
+          Process.sleep(500)
+        end)
 
-      refute_receive %Message{
-                       event: "system",
-                       payload: %{
-                         "channel" => "any",
-                         "extension" => "postgres_changes",
-                         "message" => "Subscribed to PostgreSQL",
-                         "status" => "ok"
-                       },
-                       ref: nil,
-                       topic: ^topic
-                     },
-                     1000
+      assert log =~ "InvalidJoinPayload"
     end
   end
 

test/realtime_web/channels/realtime_channel_test.exs

@@ -503,12 +503,9 @@ defmodule RealtimeWeb.RealtimeChannelTest do
 
     test "expired jwt returns a error with sub data if available log_level=warning", %{tenant: tenant} do
       sub = random_string()
-
       api_key = Generators.generate_jwt_token(tenant)
-
-      jwt =
-        Generators.generate_jwt_token(tenant, %{role: "authenticated", exp: System.system_time(:second) - 1, sub: sub})
-
+      claims = %{role: "authenticated", exp: System.system_time(:second) - 1, sub: sub}
+      jwt = Generators.generate_jwt_token(tenant, claims)
       assert {:ok, socket} = connect(UserSocket, %{"log_level" => "warning"}, conn_opts(tenant, api_key))
 
       log =
@@ -675,6 +672,44 @@ defmodule RealtimeWeb.RealtimeChannelTest do
     end
   end
 
+  describe "join payload validations" do
+    test "valid payload allows join", %{tenant: tenant} do
+      jwt = Generators.generate_jwt_token(tenant)
+      {:ok, %Socket{} = socket} = connect(UserSocket, %{"log_level" => "warning"}, conn_opts(tenant, jwt))
+
+      config = %{
+        "config" => %{
+          "private" => false,
+          "broadcast" => %{"ack" => false, "self" => false},
+          "presence" => %{"enabled" => true, "key" => "potato"},
+          "postgres_changes" => [
+            %{"event" => "INSERT", "schema" => "public", "table" => "users", "filter" => "id=eq.1"},
+            %{"event" => "DELETE", "schema" => "public", "table" => "users", "filter" => "id=eq.2"},
+            %{"event" => "UPDATE", "schema" => "public", "table" => "users", "filter" => "id=eq.3"}
+          ]
+        },
+        "access_token" => jwt
+      }
+
+      assert {:ok, _, %Socket{}} = subscribe_and_join(socket, "realtime:test", config)
+    end
+
+    test "invalid payload returns error", %{tenant: tenant} do
+      jwt = Generators.generate_jwt_token(tenant)
+      {:ok, %Socket{} = socket} = connect(UserSocket, %{"log_level" => "warning"}, conn_opts(tenant, jwt))
+
+      log =
+        capture_log(fn ->
+          assert {:error, %{reason: reason}} =
+                   subscribe_and_join(socket, "realtime:test", %{"config" => "potato"})
+
+          assert reason =~ "unable to parse, expected a map"
+        end)
+
+      assert log =~ "InvalidJoinPayload"
+    end
+  end
+
   test "registers transport pid and channel pid per tenant", %{tenant: tenant} do
     jwt = Generators.generate_jwt_token(tenant)
     {:ok, %Socket{} = socket} = connect(UserSocket, %{"log_level" => "warning"}, conn_opts(tenant, jwt))