fix: Match error on handle_info when token is checked for exp

lib/realtime_web/channels/realtime_channel.ex

@@ -316,9 +316,11 @@ defmodule RealtimeWeb.RealtimeChannel do
       {:error, :missing_claims} ->
         shutdown_response(socket, "Fields `role` and `exp` are required in JWT")
 
+      {:error, :expired_token, msg} ->
+        shutdown_response(socket, msg)
+
       {:error, error} ->
-        message = "Access token has expired: " <> to_log(error)
-        shutdown_response(socket, message)
+        shutdown_response(socket, to_log(error))
     end
   end
 
@@ -410,20 +412,20 @@ defmodule RealtimeWeb.RealtimeChannel do
       {:error, :expired_token, msg} ->
         shutdown_response(socket, msg)
 
-      {:error, error} ->
-        msg = "Received an invalid access token from client: " <> inspect(error)
+      {:error, :missing_claims} ->
+        shutdown_response(socket, "Fields `role` and `exp` are required in JWT")
 
-        shutdown_response(socket, msg)
+      {:error, :expected_claims_map} ->
+        shutdown_response(socket, "Token claims must be a map")
+
+      {:error, error} ->
+        shutdown_response(socket, inspect(error))
     end
   end
 
-  def handle_in("broadcast", payload, socket) do
-    BroadcastHandler.call(payload, socket)
-  end
+  def handle_in("broadcast", payload, socket), do: BroadcastHandler.call(payload, socket)
 
-  def handle_in("presence", payload, socket) do
-    PresenceHandler.call(payload, socket)
-  end
+  def handle_in("presence", payload, socket), do: PresenceHandler.call(payload, socket)
 
   def handle_in(type, payload, socket) do
     socket = count(socket)

mix.exs

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

test/integration/rt_channel_test.exs

@@ -511,15 +511,25 @@
     end
   end
 
-  test "token required the role key" do
-    {:ok, token} = token_no_role()
+  describe "token handling" do
+    setup [:rls_context]
 
-    assert {:error, %{status_code: 403}} =
-             WebsocketClient.connect(self(), @uri, @serializer, [{"x-api-key", token}])
-  end
+    @tag policies: [
+           :authenticated_read_broadcast_and_presence,
+           :authenticated_write_broadcast_and_presence
+         ]
+    test "invalid JWT with expired token" do
+      assert capture_log(fn ->
+               get_connection("authenticated", %{:exp => System.system_time(:second) - 1000})
+             end) =~ "InvalidJWTToken: Token as expired 1000 seconds ago"
+    end
 
-  describe "handle refresh token messages" do
-    setup [:rls_context]
+    test "token required the role key" do
+      {:ok, token} = token_no_role()
+
+      assert {:error, %{status_code: 403}} =
+               WebsocketClient.connect(self(), @uri, @serializer, [{"x-api-key", token}])
+    end
 
     @tag policies: [
            :authenticated_read_broadcast_and_presence,
@@ -598,7 +608,7 @@
         event: "system",
         payload: %{
           "extension" => "system",
-          "message" => "Received an invalid access token from client: :expected_claims_map",
+          "message" => "Token claims must be a map",
           "status" => "error"
         }
       }
@@ -635,6 +645,74 @@
              end) =~
                "ChannelShutdown: Token as expired 1000 seconds ago"
     end
+
+    test "missing claims close connection",
+         %{topic: topic} do
+      {socket, access_token} =
+        get_connection("authenticated")
+
+      config = %{broadcast: %{self: true}, private: false}
+      realtime_topic = "realtime:#{topic}"
+
+      WebsocketClient.join(socket, realtime_topic, %{config: config, access_token: access_token})
+
+      assert_receive %Phoenix.Socket.Message{event: "phx_reply"}, 500
+      assert_receive %Phoenix.Socket.Message{event: "presence_state"}, 500
+      {:ok, token} = generate_token(%{:exp => System.system_time(:second) + 2000})
+      # Update token to be a near expiring token
+      WebsocketClient.send_event(socket, realtime_topic, "access_token", %{
+        "access_token" => token
+      })
+
+      assert_receive %Phoenix.Socket.Message{
+                       event: "system",
+                       payload: %{
+                         "extension" => "system",
+                         "message" => "Fields `role` and `exp` are required in JWT",
+                         "status" => "error"
+                       }
+                     },
+                     500
+
+      assert_receive %Phoenix.Socket.Message{event: "phx_close"}
+    end
+
+    test "checks token periodically",
+         %{topic: topic} do
+      {socket, access_token} =
+        get_connection("authenticated")
+
+      config = %{broadcast: %{self: true}, private: false}
+      realtime_topic = "realtime:#{topic}"
+
+      WebsocketClient.join(socket, realtime_topic, %{config: config, access_token: access_token})
+
+      assert_receive %Phoenix.Socket.Message{event: "phx_reply"}, 500
+      assert_receive %Phoenix.Socket.Message{event: "presence_state"}, 500
+
+      {:ok, token} =
+        generate_token(%{:exp => System.system_time(:second) + 2, role: "authenticated"})
+
+      # Update token to be a near expiring token
+      WebsocketClient.send_event(socket, realtime_topic, "access_token", %{
+        "access_token" => token
+      })
+
+      # Awaits to see if connection closes automatically
+      assert_receive %Phoenix.Socket.Message{
+                       event: "system",
+                       payload: %{
+                         "extension" => "system",
+                         "message" => msg,
+                         "status" => "error"
+                       }
+                     },
+                     3000
+
+      assert_receive %Phoenix.Socket.Message{event: "phx_close"}
+
+      assert msg =~ "Token as expired"
+    end
   end
 
   describe "handle broadcast changes" do
@@ -886,7 +964,7 @@
     } do
       set_private_only(true)
 
       Realtime.Tenants.Cache.invalidate_tenant_cache(@external_id)
 
       Process.sleep(100)
 
@@ -978,20 +1056,6 @@
     end
   end
 
-  describe "invalid jwt handling" do
-    setup [:rls_context]
-
-    @tag policies: [
-           :authenticated_read_broadcast_and_presence,
-           :authenticated_write_broadcast_and_presence
-         ]
-    test "invalid JWT with expired token" do
-      assert capture_log(fn ->
-               get_connection("authenticated", %{:exp => System.system_time(:second) - 1000})
-             end) =~ "InvalidJWTToken: Token as expired 1000 seconds ago"
-    end
-  end
-
   test "handle empty topic by closing the socket" do
     {socket, _} = get_connection("authenticated")
     config = %{broadcast: %{self: true}, private: false}
@@ -1048,10 +1112,10 @@
   end
 
   def setup_trigger(%{tenant: tenant, topic: topic} = context) do
     Realtime.Tenants.Connect.shutdown(@external_id)
     Process.sleep(500)
 
     {:ok, db_conn} = Realtime.Tenants.Connect.connect(@external_id)
 
     random_name = String.downcase("test_#{random_string()}")
     query = "CREATE TABLE #{random_name} (id serial primary key, details text)"
@@ -1088,7 +1152,7 @@
       {:ok, db_conn} = Database.connect(tenant, "realtime_test", :stop)
       query = "DROP TABLE #{random_name} CASCADE"
       Postgrex.query!(db_conn, query, [])
       Realtime.Tenants.Connect.shutdown(db_conn)
 
       Process.sleep(500)
     end)
@@ -1101,9 +1165,9 @@
   defp set_private_only(value) do
     @external_id
     |> Realtime.Tenants.get_tenant_by_external_id()
     |> Realtime.Api.Tenant.changeset(%{private_only: value})
     |> Realtime.Repo.update!()
 
     Realtime.Tenants.Cache.invalidate_tenant_cache(@external_id)
   end
 end