Merge branch 'develop' of ssh://git.pleroma.social:2222/pleroma/pleroma into froth

CHANGELOG.md

@@ -56,6 +56,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

lib/pleroma/application.ex

@@ -94,7 +94,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) ++

lib/pleroma/web/admin_api/controllers/chat_controller.ex

@@ -8,7 +8,6 @@ defmodule Pleroma.Web.AdminAPI.ChatController do
   alias Pleroma.Activity
   alias Pleroma.Chat
   alias Pleroma.Chat.MessageReference
-  alias Pleroma.ModerationLog
   alias Pleroma.Pagination
   alias Pleroma.Web.AdminAPI
   alias Pleroma.Web.CommonAPI
@@ -42,12 +41,6 @@ def delete_message(%{assigns: %{user: user}} = conn, %{
          ^chat_id <- to_string(cm_ref.chat_id),
          %Activity{id: activity_id} <- Activity.get_create_by_object_ap_id(object_ap_id),
          {:ok, _} <- CommonAPI.delete(activity_id, user) do
-      ModerationLog.insert_log(%{
-        action: "chat_message_delete",
-        actor: user,
-        subject_id: message_id
-      })
-
       conn
       |> put_view(MessageReferenceView)
       |> render("show.json", chat_message_reference: cm_ref)

lib/pleroma/web/admin_api/controllers/status_controller.ex

@@ -65,12 +65,6 @@ def update(%{assigns: %{user: admin}, body_params: params} = conn, %{id: id}) do
 
   def delete(%{assigns: %{user: user}} = conn, %{id: id}) do
     with {:ok, %Activity{}} <- CommonAPI.delete(id, user) do
-      ModerationLog.insert_log(%{
-        action: "status_delete",
-        actor: user,
-        subject_id: id
-      })
-
       json(conn, %{})
     end
   end

lib/pleroma/web/common_api.ex

@@ -6,6 +6,7 @@ defmodule Pleroma.Web.CommonAPI do
   alias Pleroma.Activity
   alias Pleroma.Conversation.Participation
   alias Pleroma.Formatter
+  alias Pleroma.ModerationLog
   alias Pleroma.Object
   alias Pleroma.ThreadMute
   alias Pleroma.User
@@ -147,6 +148,21 @@ def delete(activity_id, user) do
          true <- User.superuser?(user) || user.ap_id == object.data["actor"],
          {:ok, delete_data, _} <- Builder.delete(user, object.data["id"]),
          {:ok, delete, _} <- Pipeline.common_pipeline(delete_data, local: true) do
+      if User.superuser?(user) and user.ap_id != object.data["actor"] do
+        action =
+          if object.data["type"] == "ChatMessage" do
+            "chat_message_delete"
+          else
+            "status_delete"
+          end
+
+        ModerationLog.insert_log(%{
+          action: action,
+          actor: user,
+          subject_id: activity_id
+        })
+      end
+
       {:ok, delete}
     else
       {:find_activity, _} ->

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, 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} ->
@@ -52,7 +53,7 @@ def websocket_init(state) do
       "#{__MODULE__} accepted websocket connection for user #{(state.user || %{id: "anonymous"}).id}, topic #{state.topic}"
     )
 
-    Streamer.add_socket(state.topic, state.user)
+    Streamer.add_socket(state.topic, state.oauth_token)
     {:ok, %{state | timer: timer()}}
   end
 
@@ -98,6 +99,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

lib/pleroma/web/o_auth/token/strategy/revoke.ex

@@ -21,6 +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)
+    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

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}
@@ -338,6 +338,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.

mix.exs

@@ -4,7 +4,7 @@ defmodule Pleroma.Mixfile do
   def project do
     [
       app: :pleroma,
-      version: version("2.4.52"),
+      version: version("2.4.53"),
       elixir: "~> 1.10",
       elixirc_paths: elixirc_paths(Mix.env()),
       compilers: [:phoenix, :gettext] ++ Mix.compilers(),

priv/gettext/errors.pot

@@ -111,7 +111,7 @@ msgid "Can't display this activity"
 msgstr ""
 
 #, elixir-autogen, elixir-format
-#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:325
+#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:327
 msgid "Can't find user"
 msgstr ""
 
@@ -173,7 +173,7 @@ msgid "Invalid CAPTCHA"
 msgstr ""
 
 #, elixir-autogen, elixir-format
-#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:144
+#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:146
 #: lib/pleroma/web/o_auth/o_auth_controller.ex:631
 msgid "Invalid credentials"
 msgstr ""
@@ -199,7 +199,7 @@ msgid "Invalid password."
 msgstr ""
 
 #, elixir-autogen, elixir-format
-#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:255
+#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:257
 msgid "Invalid request"
 msgstr ""
 
@@ -209,7 +209,7 @@ msgid "Kocaptcha service unavailable"
 msgstr ""
 
 #, elixir-autogen, elixir-format
-#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:140
+#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:142
 msgid "Missing parameters"
 msgstr ""
 
@@ -236,6 +236,7 @@ msgid "Poll's author can't vote"
 msgstr ""
 
 #, elixir-autogen, elixir-format
+#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:492
 #: lib/pleroma/web/mastodon_api/controllers/fallback_controller.ex:20
 #: lib/pleroma/web/mastodon_api/controllers/poll_controller.ex:39
 #: lib/pleroma/web/mastodon_api/controllers/poll_controller.ex:51
@@ -438,7 +439,7 @@ msgid "List not found"
 msgstr ""
 
 #, elixir-autogen, elixir-format
-#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:151
+#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:153
 msgid "Missing parameter: %{name}"
 msgstr ""
 
@@ -557,7 +558,7 @@ msgid "Access denied"
 msgstr ""
 
 #, elixir-autogen, elixir-format
-#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:322
+#: lib/pleroma/web/mastodon_api/controllers/account_controller.ex:324
 msgid "This API requires an authenticated user"
 msgstr ""
 

test/pleroma/integration/mastodon_websocket_test.exs

@@ -93,7 +93,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
@@ -130,5 +130,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

test/pleroma/web/admin_api/controllers/chat_controller_test.exs

@@ -53,7 +53,7 @@ test "it deletes a message from the chat", %{conn: conn, admin: admin} do
       log_entry = Repo.one(ModerationLog)
 
       assert ModerationLog.get_log_entry_message(log_entry) ==
-               "@#{admin.nickname} deleted chat message ##{cm_ref.id}"
+               "@#{admin.nickname} deleted chat message ##{message.id}"
 
       assert result["id"] == cm_ref.id
       refute MessageReference.get_by_id(cm_ref.id)

test/pleroma/web/mastodon_api/controllers/status_controller_test.exs

@@ -8,6 +8,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusControllerTest do
 
   alias Pleroma.Activity
   alias Pleroma.Conversation.Participation
+  alias Pleroma.ModerationLog
   alias Pleroma.Object
   alias Pleroma.Repo
   alias Pleroma.ScheduledActivity
@@ -970,30 +971,40 @@ test "when you didn't create it" do
       assert Activity.get_by_id(activity.id) == activity
     end
 
-    test "when you're an admin or moderator", %{conn: conn} do
-      activity1 = insert(:note_activity)
-      activity2 = insert(:note_activity)
-      admin = insert(:user, is_admin: true)
-      moderator = insert(:user, is_moderator: true)
+    test "when you're an admin", %{conn: conn} do
+      activity = insert(:note_activity)
+      user = insert(:user, is_admin: true)
 
       res_conn =
         conn
-        |> assign(:user, admin)
-        |> assign(:token, insert(:oauth_token, user: admin, scopes: ["write:statuses"]))
-        |> delete("/api/v1/statuses/#{activity1.id}")
+        |> assign(:user, user)
+        |> assign(:token, insert(:oauth_token, user: user, scopes: ["write:statuses"]))
+        |> delete("/api/v1/statuses/#{activity.id}")
 
       assert %{} = json_response_and_validate_schema(res_conn, 200)
 
+      assert ModerationLog |> Repo.one() |> ModerationLog.get_log_entry_message() ==
+               "@#{user.nickname} deleted status ##{activity.id}"
+
+      refute Activity.get_by_id(activity.id)
+    end
+
+    test "when you're a moderator", %{conn: conn} do
+      activity = insert(:note_activity)
+      user = insert(:user, is_moderator: true)
+
       res_conn =
         conn
-        |> assign(:user, moderator)
-        |> assign(:token, insert(:oauth_token, user: moderator, scopes: ["write:statuses"]))
-        |> delete("/api/v1/statuses/#{activity2.id}")
+        |> assign(:user, user)
+        |> assign(:token, insert(:oauth_token, user: user, scopes: ["write:statuses"]))
+        |> delete("/api/v1/statuses/#{activity.id}")
 
       assert %{} = json_response_and_validate_schema(res_conn, 200)
 
-      refute Activity.get_by_id(activity1.id)
-      refute Activity.get_by_id(activity2.id)
+      assert ModerationLog |> Repo.one() |> ModerationLog.get_log_entry_message() ==
+               "@#{user.nickname} deleted status ##{activity.id}"
+
+      refute Activity.get_by_id(activity.id)
     end
   end
 

test/pleroma/web/streamer_test.exs

@@ -887,4 +887,105 @@ 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
+    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"])
+
+      post_user = insert(:user)
+      CommonAPI.follow(user, post_user)
+      CommonAPI.follow(user2, post_user)
+
+      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"
+        })
+
+      send_all.(tasks, {StreamerTest, :ready})
+
+      Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)
+
+      send_all.(tasks, {StreamerTest, :revoked})
+
+      Enum.each(tasks, &Task.await/1)
+    end
+
+    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)
+
+      tasks = [
+        Task.async(child_proc.(starter.(user, token), hit)),
+        Task.async(child_proc.(starter.(user, token), hit))
+      ]
+
+      {:ok, _} =
+        CommonAPI.post(post_user, %{
+          status: "hi"
+        })
+
+      send_all.(tasks, {StreamerTest, :ready})
+
+      Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)
+
+      send_all.(tasks, {StreamerTest, :revoked})
+
+      Enum.each(tasks, &Task.await/1)
+    end
+  end
 end

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