From c62a4f1c173490ad64fdfbab0c005ca3523b6013 Mon Sep 17 00:00:00 2001
From: Tusooa Zhu <tusooa@kazv.moe>
Date: Fri, 19 Aug 2022 13:19:38 -0400
Subject: Disconnect streaming sessions when token is revoked

---
 lib/pleroma/web/mastodon_api/websocket_handler.ex |  8 +++-
 lib/pleroma/web/o_auth/token/strategy/revoke.ex   |  1 +
 lib/pleroma/web/streamer.ex                       | 24 ++++++++--
 test/pleroma/web/streamer_test.exs                | 54 +++++++++++++++++++++++
 4 files changed, 81 insertions(+), 6 deletions(-)

diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
index 0d1faffbd..ffbc2c4de 100644
--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
@@ -32,7 +32,7 @@ def init(%{qs: qs} = req, state) do
           req
         end
 
-      {:cowboy_websocket, req, %{user: user, topic: topic, count: 0, timer: nil},
+      {:cowboy_websocket, req, %{user: user, topic: topic, oauth_token: oauth_token, count: 0, timer: nil},
        %{idle_timeout: @timeout}}
     else
       {:error, :bad_topic} ->
@@ -54,7 +54,7 @@ def websocket_init(state) do
       }, topic #{state.topic}"
     )
 
-    Streamer.add_socket(state.topic, state.user)
+    Streamer.add_socket(state.topic, state.oauth_token)
     {:ok, %{state | timer: timer()}}
   end
 
@@ -100,6 +100,10 @@ def websocket_info(:tick, state) do
     {:reply, :ping, %{state | timer: nil, count: 0}, :hibernate}
   end
 
+  def websocket_info(:close, state) do
+    {:stop, state}
+  end
+
   # State can be `[]` only in case we terminate before switching to websocket,
   # we already log errors for these cases in `init/1`, so just do nothing here
   def terminate(_reason, _req, []), do: :ok
diff --git a/lib/pleroma/web/o_auth/token/strategy/revoke.ex b/lib/pleroma/web/o_auth/token/strategy/revoke.ex
index 8d6572704..03a0b91ae 100644
--- a/lib/pleroma/web/o_auth/token/strategy/revoke.ex
+++ b/lib/pleroma/web/o_auth/token/strategy/revoke.ex
@@ -22,5 +22,6 @@ def revoke(%App{} = app, %{"token" => token} = _attrs) do
   @spec revoke(Token.t()) :: {:ok, Token.t()} | {:error, Ecto.Changeset.t()}
   def revoke(%Token{} = token) do
     Repo.delete(token)
+    Pleroma.Web.Streamer.close_streams_by_oauth_token(token)
   end
 end
diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex
index fc3bbb130..8bf70d99b 100644
--- a/lib/pleroma/web/streamer.ex
+++ b/lib/pleroma/web/streamer.ex
@@ -37,7 +37,7 @@ def registry, do: @registry
           {:ok, topic :: String.t()} | {:error, :bad_topic} | {:error, :unauthorized}
   def get_topic_and_add_socket(stream, user, oauth_token, params \\ %{}) do
     with {:ok, topic} <- get_topic(stream, user, oauth_token, params) do
-      add_socket(topic, user)
+      add_socket(topic, oauth_token)
     end
   end
 
@@ -120,10 +120,10 @@ def get_topic(_stream, _user, _oauth_token, _params) do
   end
 
   @doc "Registers the process for streaming. Use `get_topic/3` to get the full authorized topic."
-  def add_socket(topic, user) do
+  def add_socket(topic, oauth_token) do
     if should_env_send?() do
-      auth? = if user, do: true
-      Registry.register(@registry, topic, auth?)
+      oauth_token_id = if oauth_token, do: oauth_token.id, else: false
+      Registry.register(@registry, topic, oauth_token_id)
     end
 
     {:ok, topic}
@@ -320,6 +320,22 @@ defp thread_containment(activity, user) do
     end
   end
 
+  def close_streams_by_oauth_token(oauth_token) do
+    if should_env_send?() do
+      Registry.select(
+        @registry,
+        [
+          {
+            {:"$1", :"$2", :"$3"},
+            [{:==, :"$3", oauth_token.id}],
+            [:"$2"]
+          }
+        ]
+      )
+      |> Enum.each(fn pid -> send(pid, :close) end)
+    end
+  end
+
   # In test environement, only return true if the registry is started.
   # In benchmark environment, returns false.
   # In any other environment, always returns true.
diff --git a/test/pleroma/web/streamer_test.exs b/test/pleroma/web/streamer_test.exs
index b788a9138..5426467e5 100644
--- a/test/pleroma/web/streamer_test.exs
+++ b/test/pleroma/web/streamer_test.exs
@@ -813,4 +813,58 @@ test "it sends conversation update to the 'direct' stream when a message is dele
       assert last_status["id"] == to_string(create_activity.id)
     end
   end
+
+  describe "stop streaming if token got revoked" do
+    test "do not revoke other tokens" do
+      %{user: user, token: token} = oauth_access(["read"])
+      %{token: token2} = oauth_access(["read"], user: user)
+      %{user: user2, token: user2_token} = oauth_access(["read"])
+
+      post_user = insert(:user)
+      CommonAPI.follow(user, post_user)
+      CommonAPI.follow(user2, post_user)
+
+      Streamer.get_topic_and_add_socket("user", user, token)
+      Streamer.get_topic_and_add_socket("user", user, token2)
+      Streamer.get_topic_and_add_socket("user", user2, user2_token)
+
+      {:ok, _} =
+        CommonAPI.post(post_user, %{
+          status: "hi"
+        })
+
+      assert_receive {:render_with_user, _, "update.json", _}
+      assert_receive {:render_with_user, _, "update.json", _}
+      assert_receive {:render_with_user, _, "update.json", _}
+
+      Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)
+
+      assert_receive :close
+      refute_receive :close
+    end
+
+    test "revoke all streams for this token" do
+      %{user: user, token: token} = oauth_access(["read"])
+
+      post_user = insert(:user)
+      CommonAPI.follow(user, post_user)
+
+      Streamer.get_topic_and_add_socket("user", user, token)
+      Streamer.get_topic_and_add_socket("user", user, token)
+
+      {:ok, _} =
+        CommonAPI.post(post_user, %{
+          status: "hi"
+        })
+
+      assert_receive {:render_with_user, _, "update.json", _}
+      assert_receive {:render_with_user, _, "update.json", _}
+
+      Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)
+
+      assert_receive :close
+      assert_receive :close
+      refute_receive :close
+    end
+  end
 end
-- 
cgit v1.2.3


From eb42e90c4f9ca35a6dc0e84e6f87b6f4b680173c Mon Sep 17 00:00:00 2001
From: Tusooa Zhu <tusooa@kazv.moe>
Date: Fri, 19 Aug 2022 13:56:39 -0400
Subject: Use Websockex to replace websocket_client

---
 mix.exs                                            |  2 +-
 mix.lock                                           |  2 +-
 .../integration/mastodon_websocket_test.exs        | 14 +++++------
 test/support/websocket_client.ex                   | 28 ++++++++++------------
 4 files changed, 22 insertions(+), 24 deletions(-)

diff --git a/mix.exs b/mix.exs
index 927f39975..46c9fcaa2 100644
--- a/mix.exs
+++ b/mix.exs
@@ -210,7 +210,7 @@ defp deps do
       {:excoveralls, "0.12.3", only: :test},
       {:hackney, "~> 1.18.0", override: true},
       {:mox, "~> 1.0", only: :test},
-      {:websocket_client, git: "https://github.com/jeremyong/websocket_client.git", only: :test}
+      {:websockex, "~> 0.4.3", only: :test}
     ] ++ oauth_deps()
   end
 
diff --git a/mix.lock b/mix.lock
index 821c397b4..1fe713e8e 100644
--- a/mix.lock
+++ b/mix.lock
@@ -126,5 +126,5 @@
   "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.0", "bc84380c9ab48177092f43ac89e4dfa2c6d62b40b8bd132b1059ecc7232f9a78", [:rebar3], [], "hexpm", "25eee6d67df61960cf6a794239566599b09e17e668d3700247bc498638152521"},
   "unsafe": {:hex, :unsafe, "1.0.1", "a27e1874f72ee49312e0a9ec2e0b27924214a05e3ddac90e91727bc76f8613d8", [:mix], [], "hexpm", "6c7729a2d214806450d29766abc2afaa7a2cbecf415be64f36a6691afebb50e5"},
   "web_push_encryption": {:git, "https://github.com/lanodan/elixir-web-push-encryption.git", "026a043037a89db4da8f07560bc8f9c68bcf0cc0", [branch: "bugfix/otp-24"]},
-  "websocket_client": {:git, "https://github.com/jeremyong/websocket_client.git", "9a6f65d05ebf2725d62fb19262b21f1805a59fbf", []},
+  "websockex": {:hex, :websockex, "0.4.3", "92b7905769c79c6480c02daacaca2ddd49de936d912976a4d3c923723b647bf0", [:mix], [], "hexpm", "95f2e7072b85a3a4cc385602d42115b73ce0b74a9121d0d6dbbf557645ac53e4"},
 }
diff --git a/test/pleroma/integration/mastodon_websocket_test.exs b/test/pleroma/integration/mastodon_websocket_test.exs
index 43ec57893..1e0319144 100644
--- a/test/pleroma/integration/mastodon_websocket_test.exs
+++ b/test/pleroma/integration/mastodon_websocket_test.exs
@@ -33,16 +33,16 @@ def start_socket(qs \\ nil, headers \\ []) do
 
   test "refuses invalid requests" do
     capture_log(fn ->
-      assert {:error, {404, _}} = start_socket()
-      assert {:error, {404, _}} = start_socket("?stream=ncjdk")
+      assert {:error, %WebSockex.RequestError{code: 404}} = start_socket()
+      assert {:error, %WebSockex.RequestError{code: 404}} = start_socket("?stream=ncjdk")
       Process.sleep(30)
     end)
   end
 
   test "requires authentication and a valid token for protected streams" do
     capture_log(fn ->
-      assert {:error, {401, _}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
-      assert {:error, {401, _}} = start_socket("?stream=user")
+      assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
+      assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user")
       Process.sleep(30)
     end)
   end
@@ -102,7 +102,7 @@ test "accepts the 'user' stream", %{token: token} = _state do
       assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")
 
       capture_log(fn ->
-        assert {:error, {401, _}} = start_socket("?stream=user")
+        assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user")
         Process.sleep(30)
       end)
     end
@@ -111,7 +111,7 @@ test "accepts the 'user:notification' stream", %{token: token} = _state do
       assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
 
       capture_log(fn ->
-        assert {:error, {401, _}} = start_socket("?stream=user:notification")
+        assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user:notification")
         Process.sleep(30)
       end)
     end
@@ -120,7 +120,7 @@ test "accepts valid token on Sec-WebSocket-Protocol header", %{token: token} do
       assert {:ok, _} = start_socket("?stream=user", [{"Sec-WebSocket-Protocol", token.token}])
 
       capture_log(fn ->
-        assert {:error, {401, _}} =
+        assert {:error, %WebSockex.RequestError{code: 401}} =
                  start_socket("?stream=user", [{"Sec-WebSocket-Protocol", "I am a friend"}])
 
         Process.sleep(30)
diff --git a/test/support/websocket_client.ex b/test/support/websocket_client.ex
index 34b955474..2660f6151 100644
--- a/test/support/websocket_client.ex
+++ b/test/support/websocket_client.ex
@@ -5,18 +5,17 @@
 defmodule Pleroma.Integration.WebsocketClient do
   # https://github.com/phoenixframework/phoenix/blob/master/test/support/websocket_client.exs
 
+  use WebSockex
+
   @doc """
   Starts the WebSocket server for given ws URL. Received Socket.Message's
   are forwarded to the sender pid
   """
   def start_link(sender, url, headers \\ []) do
-    :crypto.start()
-    :ssl.start()
-
-    :websocket_client.start_link(
-      String.to_charlist(url),
+    WebSockex.start_link(
+      url,
       __MODULE__,
-      [sender],
+      %{ sender: sender },
       extra_headers: headers
     )
   end
@@ -36,27 +35,26 @@ def send_text(server_pid, msg) do
   end
 
   @doc false
-  def init([sender], _conn_state) do
-    {:ok, %{sender: sender}}
-  end
-
-  @doc false
-  def websocket_handle(frame, _conn_state, state) do
+  @impl true
+  def handle_frame(frame, state) do
     send(state.sender, frame)
     {:ok, state}
   end
 
   @doc false
-  def websocket_info({:text, msg}, _conn_state, state) do
+  @impl true
+  def handle_info({:text, msg}, state) do
     {:reply, {:text, msg}, state}
   end
 
-  def websocket_info(:close, _conn_state, _state) do
+  @impl true
+  def handle_info(:close, _state) do
     {:close, <<>>, "done"}
   end
 
   @doc false
-  def websocket_terminate(_reason, _conn_state, _state) do
+  @impl true
+  def terminate(_reason, _state) do
     :ok
   end
 end
-- 
cgit v1.2.3


From 3522852c6196cafa63804240f52dd593e09ba694 Mon Sep 17 00:00:00 2001
From: Tusooa Zhu <tusooa@kazv.moe>
Date: Fri, 19 Aug 2022 14:09:42 -0400
Subject: Test that server will disconnect websocket upon token revocation

---
 test/pleroma/integration/mastodon_websocket_test.exs | 18 +++++++++++++++++-
 test/support/websocket_client.ex                     |  6 ++++++
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/test/pleroma/integration/mastodon_websocket_test.exs b/test/pleroma/integration/mastodon_websocket_test.exs
index 1e0319144..adb2d7004 100644
--- a/test/pleroma/integration/mastodon_websocket_test.exs
+++ b/test/pleroma/integration/mastodon_websocket_test.exs
@@ -91,7 +91,7 @@ test "receives well formatted events" do
 
       {:ok, token} = OAuth.Token.exchange_token(app, auth)
 
-      %{user: user, token: token}
+      %{app: app, user: user, token: token}
     end
 
     test "accepts valid tokens", state do
@@ -126,5 +126,21 @@ test "accepts valid token on Sec-WebSocket-Protocol header", %{token: token} do
         Process.sleep(30)
       end)
     end
+
+    test "disconnect when token is revoked", %{app: app, user: user, token: token} do
+      assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
+      assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")
+
+      {:ok, auth} = OAuth.Authorization.create_authorization(app, user)
+
+      {:ok, token2} = OAuth.Token.exchange_token(app, auth)
+      assert {:ok, _} = start_socket("?stream=user&access_token=#{token2.token}")
+
+      OAuth.Token.Strategy.Revoke.revoke(token)
+
+      assert_receive {:close, _}
+      assert_receive {:close, _}
+      refute_receive {:close, _}
+    end
   end
 end
diff --git a/test/support/websocket_client.ex b/test/support/websocket_client.ex
index 2660f6151..abe7d5eda 100644
--- a/test/support/websocket_client.ex
+++ b/test/support/websocket_client.ex
@@ -41,6 +41,12 @@ def handle_frame(frame, state) do
     {:ok, state}
   end
 
+  @impl true
+  def handle_disconnect(conn_status, state) do
+    send(state.sender, {:close, conn_status})
+    {:ok, state}
+  end
+
   @doc false
   @impl true
   def handle_info({:text, msg}, state) do
-- 
cgit v1.2.3


From f459c1260b43396fb7173e97e29ccef441a615ec Mon Sep 17 00:00:00 2001
From: Tusooa Zhu <tusooa@kazv.moe>
Date: Fri, 19 Aug 2022 14:10:07 -0400
Subject: Lint

---
 lib/pleroma/web/mastodon_api/websocket_handler.ex    | 3 ++-
 test/pleroma/integration/mastodon_websocket_test.exs | 8 ++++++--
 test/support/websocket_client.ex                     | 2 +-
 3 files changed, 9 insertions(+), 4 deletions(-)

diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
index ffbc2c4de..930e9eb29 100644
--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
@@ -32,7 +32,8 @@ def init(%{qs: qs} = req, state) do
           req
         end
 
-      {:cowboy_websocket, req, %{user: user, topic: topic, oauth_token: oauth_token, count: 0, timer: nil},
+      {:cowboy_websocket, req,
+       %{user: user, topic: topic, oauth_token: oauth_token, count: 0, timer: nil},
        %{idle_timeout: @timeout}}
     else
       {:error, :bad_topic} ->
diff --git a/test/pleroma/integration/mastodon_websocket_test.exs b/test/pleroma/integration/mastodon_websocket_test.exs
index adb2d7004..d44033842 100644
--- a/test/pleroma/integration/mastodon_websocket_test.exs
+++ b/test/pleroma/integration/mastodon_websocket_test.exs
@@ -41,7 +41,9 @@ test "refuses invalid requests" do
 
   test "requires authentication and a valid token for protected streams" do
     capture_log(fn ->
-      assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
+      assert {:error, %WebSockex.RequestError{code: 401}} =
+               start_socket("?stream=user&access_token=aaaaaaaaaaaa")
+
       assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user")
       Process.sleep(30)
     end)
@@ -111,7 +113,9 @@ test "accepts the 'user:notification' stream", %{token: token} = _state do
       assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
 
       capture_log(fn ->
-        assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user:notification")
+        assert {:error, %WebSockex.RequestError{code: 401}} =
+                 start_socket("?stream=user:notification")
+
         Process.sleep(30)
       end)
     end
diff --git a/test/support/websocket_client.ex b/test/support/websocket_client.ex
index abe7d5eda..70d331999 100644
--- a/test/support/websocket_client.ex
+++ b/test/support/websocket_client.ex
@@ -15,7 +15,7 @@ def start_link(sender, url, headers \\ []) do
     WebSockex.start_link(
       url,
       __MODULE__,
-      %{ sender: sender },
+      %{sender: sender},
       extra_headers: headers
     )
   end
-- 
cgit v1.2.3


From a31d6bb52c8856c71f20d49aec8948573dacba68 Mon Sep 17 00:00:00 2001
From: Tusooa Zhu <tusooa@kazv.moe>
Date: Fri, 19 Aug 2022 14:58:57 -0400
Subject: Execute session disconnect in background

---
 lib/pleroma/application.ex                      |  3 ++-
 lib/pleroma/web/o_auth/token/strategy/revoke.ex | 15 +++++++++++++--
 2 files changed, 15 insertions(+), 3 deletions(-)

diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index 9824e0a4a..92d143665 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -89,7 +89,8 @@ def start(_type, _args) do
         Pleroma.Repo,
         Config.TransferTask,
         Pleroma.Emoji,
-        Pleroma.Web.Plugs.RateLimiter.Supervisor
+        Pleroma.Web.Plugs.RateLimiter.Supervisor,
+        {Task.Supervisor, name: Pleroma.TaskSupervisor}
       ] ++
         cachex_children() ++
         http_children(adapter, @mix_env) ++
diff --git a/lib/pleroma/web/o_auth/token/strategy/revoke.ex b/lib/pleroma/web/o_auth/token/strategy/revoke.ex
index 03a0b91ae..de99bc137 100644
--- a/lib/pleroma/web/o_auth/token/strategy/revoke.ex
+++ b/lib/pleroma/web/o_auth/token/strategy/revoke.ex
@@ -21,7 +21,18 @@ def revoke(%App{} = app, %{"token" => token} = _attrs) do
   @doc "Revokes access token"
   @spec revoke(Token.t()) :: {:ok, Token.t()} | {:error, Ecto.Changeset.t()}
   def revoke(%Token{} = token) do
-    Repo.delete(token)
-    Pleroma.Web.Streamer.close_streams_by_oauth_token(token)
+    with {:ok, token} <- Repo.delete(token) do
+      Task.Supervisor.start_child(
+        Pleroma.TaskSupervisor,
+        Pleroma.Web.Streamer,
+        :close_streams_by_oauth_token,
+        [token],
+        restart: :transient
+      )
+
+      {:ok, token}
+    else
+      result -> result
+    end
   end
 end
-- 
cgit v1.2.3


From 5a2c8ef4ccfbcc996fb812779730c78e2a3fbdcd Mon Sep 17 00:00:00 2001
From: Tusooa Zhu <tusooa@kazv.moe>
Date: Fri, 19 Aug 2022 19:58:16 -0400
Subject: Refactor streamer test

---
 test/pleroma/web/streamer_test.exs | 81 ++++++++++++++++++++++++++++++--------
 1 file changed, 64 insertions(+), 17 deletions(-)

diff --git a/test/pleroma/web/streamer_test.exs b/test/pleroma/web/streamer_test.exs
index 5426467e5..7c4b9e288 100644
--- a/test/pleroma/web/streamer_test.exs
+++ b/test/pleroma/web/streamer_test.exs
@@ -815,7 +815,47 @@ test "it sends conversation update to the 'direct' stream when a message is dele
   end
 
   describe "stop streaming if token got revoked" do
-    test "do not revoke other tokens" do
+    setup do
+      child_proc = fn start, finalize ->
+        fn ->
+          start.()
+
+          receive do
+            {StreamerTest, :ready} ->
+              assert_receive {:render_with_user, _, "update.json", _}
+
+              receive do
+                {StreamerTest, :revoked} -> finalize.()
+              end
+          end
+        end
+      end
+
+      starter = fn user, token ->
+        fn -> Streamer.get_topic_and_add_socket("user", user, token) end
+      end
+
+      hit = fn -> assert_receive :close end
+      miss = fn -> refute_receive :close end
+
+      send_all = fn tasks, thing -> Enum.each(tasks, &send(&1.pid, thing)) end
+
+      %{
+        child_proc: child_proc,
+        starter: starter,
+        hit: hit,
+        miss: miss,
+        send_all: send_all
+      }
+    end
+
+    test "do not revoke other tokens", %{
+      child_proc: child_proc,
+      starter: starter,
+      hit: hit,
+      miss: miss,
+      send_all: send_all
+    } do
       %{user: user, token: token} = oauth_access(["read"])
       %{token: token2} = oauth_access(["read"], user: user)
       %{user: user2, token: user2_token} = oauth_access(["read"])
@@ -824,47 +864,54 @@ test "do not revoke other tokens" do
       CommonAPI.follow(user, post_user)
       CommonAPI.follow(user2, post_user)
 
-      Streamer.get_topic_and_add_socket("user", user, token)
-      Streamer.get_topic_and_add_socket("user", user, token2)
-      Streamer.get_topic_and_add_socket("user", user2, user2_token)
+      tasks = [
+        Task.async(child_proc.(starter.(user, token), hit)),
+        Task.async(child_proc.(starter.(user, token2), miss)),
+        Task.async(child_proc.(starter.(user2, user2_token), miss))
+      ]
 
       {:ok, _} =
         CommonAPI.post(post_user, %{
           status: "hi"
         })
 
-      assert_receive {:render_with_user, _, "update.json", _}
-      assert_receive {:render_with_user, _, "update.json", _}
-      assert_receive {:render_with_user, _, "update.json", _}
+      send_all.(tasks, {StreamerTest, :ready})
 
       Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)
 
-      assert_receive :close
-      refute_receive :close
+      send_all.(tasks, {StreamerTest, :revoked})
+
+      Enum.each(tasks, &Task.await/1)
     end
 
-    test "revoke all streams for this token" do
+    test "revoke all streams for this token", %{
+      child_proc: child_proc,
+      starter: starter,
+      hit: hit,
+      send_all: send_all
+    } do
       %{user: user, token: token} = oauth_access(["read"])
 
       post_user = insert(:user)
       CommonAPI.follow(user, post_user)
 
-      Streamer.get_topic_and_add_socket("user", user, token)
-      Streamer.get_topic_and_add_socket("user", user, token)
+      tasks = [
+        Task.async(child_proc.(starter.(user, token), hit)),
+        Task.async(child_proc.(starter.(user, token), hit))
+      ]
 
       {:ok, _} =
         CommonAPI.post(post_user, %{
           status: "hi"
         })
 
-      assert_receive {:render_with_user, _, "update.json", _}
-      assert_receive {:render_with_user, _, "update.json", _}
+      send_all.(tasks, {StreamerTest, :ready})
 
       Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)
 
-      assert_receive :close
-      assert_receive :close
-      refute_receive :close
+      send_all.(tasks, {StreamerTest, :revoked})
+
+      Enum.each(tasks, &Task.await/1)
     end
   end
 end
-- 
cgit v1.2.3


From 31fd41de0cbca28cd2461e96384460596e54e9e9 Mon Sep 17 00:00:00 2001
From: Tusooa Zhu <tusooa@kazv.moe>
Date: Fri, 19 Aug 2022 20:29:06 -0400
Subject: Release 2.4.4

---
 CHANGELOG.md | 5 +++++
 mix.exs      | 2 +-
 2 files changed, 6 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 95405bb60..bcbe3ba56 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,6 +14,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ### Removed
 
+## 2.4.4 - 2022-08-19
+
+### Security
+- Streaming API sessions will now properly disconnect if the corresponding token is revoked
+
 ## 2.4.3 - 2022-05-06
 
 ### Security
diff --git a/mix.exs b/mix.exs
index 46c9fcaa2..0e2834fc6 100644
--- a/mix.exs
+++ b/mix.exs
@@ -4,7 +4,7 @@ defmodule Pleroma.Mixfile do
   def project do
     [
       app: :pleroma,
-      version: version("2.4.3"),
+      version: version("2.4.4"),
       elixir: "~> 1.9",
       elixirc_paths: elixirc_paths(Mix.env()),
       compilers: [:phoenix, :gettext] ++ Mix.compilers(),
-- 
cgit v1.2.3