fix: Implement private only check on socket connect (#1193)

Implement private only check on socket connect

lib/realtime/api/tenant.ex

@@ -74,7 +74,8 @@ defmodule Realtime.Api.Tenant do
       :max_channels_per_client,
       :max_joins_per_second,
       :suspend,
-      :notify_private_alpha
+      :notify_private_alpha,
+      :private_only
     ])
     |> validate_required([
       :external_id,

lib/realtime/tenants.ex

@@ -278,7 +278,7 @@ defmodule Realtime.Tenants do
     :notify_private_alpha
     :private_only
   """
-  @spec get_tenant_by_external_id(String.t()) :: Tenant.t() | nil
+  @spec update_management(String.t(), map()) :: Tenant.t() | nil
   def update_management(tenant_id, attrs) do
     tenant_id
     |> Cache.get_tenant_by_external_id()

lib/realtime_web/channels/realtime_channel.ex

@@ -49,6 +49,7 @@ defmodule RealtimeWeb.RealtimeChannel do
     start_db_rate_counter(tenant_id)
 
     with false <- SignalHandler.shutdown_in_progress?(),
+         :ok <- only_private?(tenant_id, socket),
          :ok <- limit_joins(socket.assigns),
          :ok <- limit_channels(socket),
          :ok <- limit_max_users(socket.assigns),
@@ -131,6 +132,13 @@ defmodule RealtimeWeb.RealtimeChannel do
           "Token expiration time is invalid"
         )
 
+      {:error, :private_only} ->
+        Logging.log_error_message(
+          :error,
+          "PrivateOnly",
+          "This project only allows private channels"
+        )
+
       {:error, error} ->
         Logging.log_error_message(:error, "UnknownError", error)
 
@@ -676,4 +684,14 @@ defmodule RealtimeWeb.RealtimeChannel do
   defp maybe_assign_policies(_, _, _, _, socket) do
     {:ok, assign(socket, policies: nil)}
   end
+
+  @spec only_private?(String.t(), map()) :: :ok | {:error, :private_only}
+  defp only_private?(tenant_id, %{assigns: %{check_authorization?: check_authorization?}}) do
+    tenant = Tenants.get_tenant_by_external_id(tenant_id)
+
+    cond do
+      tenant.private_only and !check_authorization? -> {:error, :private_only}
+      true -> :ok
+    end
+  end
 end

lib/realtime_web/controllers/tenant_controller.ex

@@ -291,7 +291,7 @@ defmodule RealtimeWeb.TenantController do
       %Tenant{} = tenant ->
         render(conn, "show.json", tenant: tenant)
 
-      {:error, :tenant_not_found} ->
+      nil ->
         Helpers.log_error("TenantNotFound", "Tenant not found")
 
         conn

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.33.22",
+      version: "2.33.23",
       elixir: "~> 1.16.0",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,

test/integration/rt_channel_test.exs

@@ -801,6 +801,56 @@ defmodule Realtime.Integration.RtChannelTest do
     end
   end
 
+  describe "only private channels" do
+    setup [:rls_context]
+
+    @tag policies: [
+           :authenticated_read_broadcast_and_presence,
+           :authenticated_write_broadcast_and_presence
+         ]
+    test "user with only private channels enabled will not be able to join public channels", %{
+      topic: topic
+    } do
+      Realtime.Tenants.update_management(@external_id, %{private_only: true})
+      :timer.sleep(100)
+      {socket, _} = get_connection("authenticated")
+      config = %{broadcast: %{self: true}, private: false}
+      topic = "realtime:#{topic}"
+      WebsocketClient.join(socket, topic, %{config: config})
+
+      assert_receive %Phoenix.Socket.Message{
+                       event: "phx_reply",
+                       payload: %{
+                         "response" => %{"reason" => "This project only allows private channels"},
+                         "status" => "error"
+                       }
+                     },
+                     500
+
+      Realtime.Tenants.update_management(@external_id, %{private_only: false})
+      :timer.sleep(100)
+    end
+
+    @tag policies: [
+           :authenticated_read_broadcast_and_presence,
+           :authenticated_write_broadcast_and_presence
+         ]
+    test "user with only private channels enabled will be able to join private channels", %{
+      topic: topic
+    } do
+      Realtime.Tenants.update_management(@external_id, %{private_only: true})
+      :timer.sleep(100)
+      {socket, _} = get_connection("authenticated")
+      config = %{broadcast: %{self: true}, private: true}
+      topic = "realtime:#{topic}"
+      WebsocketClient.join(socket, topic, %{config: config})
+
+      assert_receive %Phoenix.Socket.Message{event: "phx_reply"}, 500
+      Realtime.Tenants.update_management(@external_id, %{private_only: false})
+      :timer.sleep(100)
+    end
+  end
+
   defp token_valid(role), do: generate_token(%{role: role})
   defp token_no_role(), do: generate_token()
 

test/realtime/repo_test.exs

@@ -154,7 +154,7 @@ defmodule Realtime.RepoTest do
     test "fetches one entry and loads a given struct", %{db_conn: db_conn, tenant: tenant} do
       message_1 = message_fixture(tenant)
       _message_2 = message_fixture(tenant)
-      query = from c in Message, where: c.id == ^message_1.id
+      query = from(c in Message, where: c.id == ^message_1.id)
       assert {:ok, ^message_1} = Repo.one(db_conn, query, Message)
       assert Ecto.get_meta(message_1, :state) == :loaded
     end
@@ -169,19 +169,23 @@ defmodule Realtime.RepoTest do
     end
 
     test "if not found, returns not found error", %{db_conn: db_conn} do
-      query = from c in Message, where: c.topic == "potato"
+      query = from(c in Message, where: c.topic == "potato")
       assert {:error, :not_found} = Repo.one(db_conn, query, Message)
     end
 
     test "handles exceptions", %{db_conn: db_conn} do
       Process.unlink(db_conn)
       Process.exit(db_conn, :kill)
-      query = from c in Message, where: c.topic == "potato"
+      query = from(c in Message, where: c.topic == "potato")
       assert {:error, :postgrex_exception} = Repo.one(db_conn, query, Message)
     end
   end
 
   describe "insert/3" do
+    setup %{db_conn: db_conn} do
+      Realtime.Tenants.Connect.CreatePartitions.run(%{db_conn_pid: db_conn})
+    end
+
     test "inserts a new entry with a given changeset and returns struct", %{db_conn: db_conn} do
       changeset = Message.changeset(%Message{}, %{topic: "foo", extension: :presence})
 
@@ -271,7 +275,7 @@ defmodule Realtime.RepoTest do
 
     test "raises error on bad queries", %{db_conn: db_conn} do
       # wrong id type
-      query = from c in Message, where: c.id == "potato"
+      query = from(c in Message, where: c.id == "potato")
 
       assert_raise Ecto.QueryError, fn ->
         Repo.del(db_conn, query)