fix: Ensure token refresh checks write policy

mix.exs

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

test/integration/rt_channel_test.exs

@@ -565,10 +565,9 @@
            :authenticated_read_broadcast_and_presence,
            :authenticated_write_broadcast_and_presence
          ]
-    test "on new access_token and channel is private policies are reevaluated",
+    test "on new access_token and channel is private policies are reevaluated for read policy",
          %{topic: topic} do
       {socket, access_token} = get_connection("authenticated")
-      {:ok, new_token} = token_valid("anon")
 
       realtime_topic = "realtime:#{topic}"
 
@@ -580,6 +579,8 @@
       assert_receive %Message{event: "phx_reply"}, 500
       assert_receive %Message{event: "presence_state"}, 500
 
+      {:ok, new_token} = token_valid("anon")
+
       WebsocketClient.send_event(socket, realtime_topic, "access_token", %{
         "access_token" => new_token
       })
@@ -601,6 +602,65 @@
       assert_receive %Message{event: "phx_close", topic: ^realtime_topic}
     end
 
+    @tag policies: [
+           :authenticated_read_broadcast_and_presence,
+           :authenticated_write_broadcast_and_presence
+         ]
+    test "on new access_token and channel is private policies are reevaluated for write policy",
+         %{topic: topic, tenant: tenant} do
+      {socket, access_token} = get_connection("authenticated")
+      realtime_topic = "realtime:#{topic}"
+
+      WebsocketClient.join(socket, realtime_topic, %{
+        config: %{broadcast: %{self: true}, private: true},
+        access_token: access_token
+      })
+
+      assert_receive %Message{event: "phx_reply"}, 500
+      assert_receive %Message{event: "presence_state"}, 500
+      # Checks first send which will set write policy to true
+      payload = %{"event" => "TEST", "payload" => %{"msg" => 1}, "type" => "broadcast"}
+      WebsocketClient.send_event(socket, realtime_topic, "broadcast", payload)
+      Process.sleep(1000)
+
+      assert_receive %Message{
+                       event: "broadcast",
+                       payload: ^payload,
+                       topic: ^realtime_topic
+                     },
+                     500
+
+      # RLS policies changed to only allow read
+      {:ok, db_conn} = Database.connect(tenant, "realtime_test")
+      clean_table(db_conn, "realtime", "messages")
+      create_rls_policies(db_conn, [:authenticated_read_broadcast_and_presence], %{topic: topic})
+
+      # Set new token to recheck policies
+      {:ok, new_token} =
+        generate_token(%{
+          exp: System.system_time(:second) + 1000,
+          role: "authenticated",
+          sub: random_string()
+        })
+
+      WebsocketClient.send_event(socket, realtime_topic, "access_token", %{
+        "access_token" => new_token
+      })
+
+      # Send message to be ignored
+      payload = %{"event" => "TEST", "payload" => %{"msg" => 1}, "type" => "broadcast"}
+      WebsocketClient.send_event(socket, realtime_topic, "broadcast", payload)
+
+      Process.sleep(1000)
+
+      refute_receive %Message{
+                       event: "broadcast",
+                       payload: ^payload,
+                       topic: ^realtime_topic
+                     },
+                     500
+    end
+
     test "on new access_token and channel is public policies are not reevaluated",
          %{topic: topic} do
       {socket, access_token} = get_connection("authenticated")
@@ -1115,7 +1175,7 @@
     } do
       change_tenant_configuration(:private_only, true)
 
       Realtime.Tenants.Cache.invalidate_tenant_cache(@external_id)
 
       Process.sleep(100)
 
@@ -1453,10 +1513,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)"
@@ -1493,7 +1553,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)
@@ -1506,9 +1566,9 @@
   defp change_tenant_configuration(limit, value) do
     @external_id
     |> Realtime.Tenants.get_tenant_by_external_id()
     |> Realtime.Api.Tenant.changeset(%{limit => value})
     |> Realtime.Repo.update!()
 
     Realtime.Tenants.Cache.invalidate_tenant_cache(@external_id)
   end
 end