7 files changed, 104 insertions, 77 deletions

diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 2f0333da0..3e8f19e30 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -501,7 +501,15 @@ defmodule Pleroma.User do
 
     params = Map.put(params, :last_refreshed_at, NaiveDateTime.utc_now())
 
-    params = if remote?, do: truncate_fields_param(params), else: params
+    params =
+      if remote? do
+        params
+        |> truncate_fields_param()
+        |> truncate_if_exists(:name, name_limit)
+        |> truncate_if_exists(:bio, bio_limit)
+      else
+        params
+      end
 
     struct
     |> cast(
diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 5f895406d..c4f83f9e1 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
@@ -604,7 +604,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   end
 
   defp do_block(blocker, blocked, activity_id, local) do
-    outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
     unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
 
     if unfollow_blocked do
@@ -612,8 +611,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
       if follow_activity, do: unfollow(blocker, blocked, nil, local)
     end
 
-    with true <- outgoing_blocks,
-         block_data <- make_block_data(blocker, blocked, activity_id),
+    with block_data <- make_block_data(blocker, blocked, activity_id),
          {:ok, activity} <- insert(block_data, local),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
diff --git a/lib/pleroma/web/activity_pub/transmogrifier.ex b/lib/pleroma/web/activity_pub/transmogrifier.ex
index 831739c5f..3fc4762d6 100644
--- a/lib/pleroma/web/activity_pub/transmogrifier.ex
+++ b/lib/pleroma/web/activity_pub/transmogrifier.ex
@@ -544,6 +544,9 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
          {:ok, follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "accept"),
          %User{local: true} = follower <- User.get_cached_by_ap_id(follow_activity.data["actor"]),
          {:ok, _relationship} <- FollowingRelationship.update(follower, followed, :follow_accept) do
+      User.update_follower_count(followed)
+      User.update_following_count(follower)
+
       ActivityPub.accept(%{
         to: follow_activity.data["to"],
         type: "Accept",
@@ -553,7 +556,8 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
         activity_id: id
       })
     else
-      _e -> :error
+      _e ->
+        :error
     end
   end
 
diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex
index 15dd2ed45..a49cfa35e 100644
--- a/lib/pleroma/web/activity_pub/utils.ex
+++ b/lib/pleroma/web/activity_pub/utils.ex
@@ -6,6 +6,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   alias Ecto.Changeset
   alias Ecto.UUID
   alias Pleroma.Activity
+  alias Pleroma.Config
   alias Pleroma.Notification
   alias Pleroma.Object
   alias Pleroma.Repo
@@ -169,8 +170,11 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   Enqueues an activity for federation if it's local
   """
   @spec maybe_federate(any()) :: :ok
-  def maybe_federate(%Activity{local: true} = activity) do
-    if Pleroma.Config.get!([:instance, :federating]) do
+  def maybe_federate(%Activity{local: true, data: %{"type" => type}} = activity) do
+    outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
+
+    with true <- Config.get!([:instance, :federating]),
+         true <- type != "Block" || outgoing_blocks do
       Pleroma.Web.Federator.publish(activity)
     end
 
diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
index 5652a37c1..b1aebe014 100644
--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
@@ -12,29 +12,15 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
 
   @behaviour :cowboy_websocket
 
-  @streams [
-    "public",
-    "public:local",
-    "public:media",
-    "public:local:media",
-    "user",
-    "user:notification",
-    "direct",
-    "list",
-    "hashtag"
-  ]
-  @anonymous_streams ["public", "public:local", "hashtag"]
-
   # Handled by periodic keepalive in Pleroma.Web.Streamer.Ping.
   @timeout :infinity
 
   def init(%{qs: qs} = req, state) do
-    with params <- :cow_qs.parse_qs(qs),
+    with params <- Enum.into(:cow_qs.parse_qs(qs), %{}),
          sec_websocket <- :cowboy_req.header("sec-websocket-protocol", req, nil),
-         access_token <- List.keyfind(params, "access_token", 0),
-         {_, stream} <- List.keyfind(params, "stream", 0),
-         {:ok, user} <- allow_request(stream, [access_token, sec_websocket]),
-         topic when is_binary(topic) <- expand_topic(stream, params) do
+         access_token <- Map.get(params, "access_token"),
+         {:ok, user} <- authenticate_request(access_token, sec_websocket),
+         {:ok, topic} <- Streamer.get_topic(Map.get(params, "stream"), user, params) do
       req =
         if sec_websocket do
           :cowboy_req.set_resp_header("sec-websocket-protocol", sec_websocket, req)
@@ -44,14 +30,14 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
 
       {:cowboy_websocket, req, %{user: user, topic: topic}, %{idle_timeout: @timeout}}
     else
-      {:error, code} ->
-        Logger.debug("#{__MODULE__} denied connection: #{inspect(code)} - #{inspect(req)}")
-        {:ok, req} = :cowboy_req.reply(code, req)
+      {:error, :bad_topic} ->
+        Logger.debug("#{__MODULE__} bad topic #{inspect(req)}")
+        {:ok, req} = :cowboy_req.reply(404, req)
         {:ok, req, state}
 
-      error ->
-        Logger.debug("#{__MODULE__} denied connection: #{inspect(error)} - #{inspect(req)}")
-        {:ok, req} = :cowboy_req.reply(400, req)
+      {:error, :unauthorized} ->
+        Logger.debug("#{__MODULE__} authentication error: #{inspect(req)}")
+        {:ok, req} = :cowboy_req.reply(401, req)
         {:ok, req, state}
     end
   end
@@ -93,50 +79,23 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   end
 
   # Public streams without authentication.
-  defp allow_request(stream, [nil, nil]) when stream in @anonymous_streams do
+  defp authenticate_request(nil, nil) do
     {:ok, nil}
   end
 
   # Authenticated streams.
-  defp allow_request(stream, [access_token, sec_websocket]) when stream in @streams do
-    token =
-      with {"access_token", token} <- access_token do
-        token
-      else
-        _ -> sec_websocket
-      end
+  defp authenticate_request(access_token, sec_websocket) do
+    token = access_token || sec_websocket
 
     with true <- is_bitstring(token),
          %Token{user_id: user_id} <- Repo.get_by(Token, token: token),
          user = %User{} <- User.get_cached_by_id(user_id) do
       {:ok, user}
     else
-      _ -> {:error, 403}
-    end
-  end
-
-  # Not authenticated.
-  defp allow_request(stream, _) when stream in @streams, do: {:error, 403}
-
-  # No matching stream.
-  defp allow_request(_, _), do: {:error, 404}
-
-  defp expand_topic("hashtag", params) do
-    case List.keyfind(params, "tag", 0) do
-      {_, tag} -> "hashtag:#{tag}"
-      _ -> nil
+      _ -> {:error, :unauthorized}
     end
   end
 
-  defp expand_topic("list", params) do
-    case List.keyfind(params, "list", 0) do
-      {_, list} -> "list:#{list}"
-      _ -> nil
-    end
-  end
-
-  defp expand_topic(topic, _), do: topic
-
   defp streamer_socket(state) do
     %{transport_pid: self(), assigns: state}
   end
diff --git a/lib/pleroma/web/streamer/state.ex b/lib/pleroma/web/streamer/state.ex
index 999550b88..4eb462a1a 100644
--- a/lib/pleroma/web/streamer/state.ex
+++ b/lib/pleroma/web/streamer/state.ex
@@ -36,30 +36,28 @@ defmodule Pleroma.Web.Streamer.State do
   end
 
   def handle_call({:add, topic, socket}, _from, %{sockets: sockets} = state) do
-    internal_topic = internal_topic(topic, socket)
     stream_socket = StreamerSocket.from_socket(socket)
 
     sockets_for_topic =
       sockets
-      |> Map.get(internal_topic, [])
+      |> Map.get(topic, [])
       |> List.insert_at(0, stream_socket)
       |> Enum.uniq()
 
-    state = put_in(state, [:sockets, internal_topic], sockets_for_topic)
+    state = put_in(state, [:sockets, topic], sockets_for_topic)
     Logger.debug("Got new conn for #{topic}")
     {:reply, state, state}
   end
 
   def handle_call({:remove, topic, socket}, _from, %{sockets: sockets} = state) do
-    internal_topic = internal_topic(topic, socket)
     stream_socket = StreamerSocket.from_socket(socket)
 
     sockets_for_topic =
       sockets
-      |> Map.get(internal_topic, [])
+      |> Map.get(topic, [])
       |> List.delete(stream_socket)
 
-    state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic)
+    state = Kernel.put_in(state, [:sockets, topic], sockets_for_topic)
     {:reply, state, state}
   end
 
@@ -70,13 +68,4 @@ defmodule Pleroma.Web.Streamer.State do
   defp do_remove_socket(_env, topic, socket) do
     GenServer.call(__MODULE__, {:remove, topic, socket})
   end
-
-  defp internal_topic(topic, socket)
-       when topic in ~w[user user:notification direct] do
-    "#{topic}:#{socket.assigns[:user].id}"
-  end
-
-  defp internal_topic(topic, _) do
-    topic
-  end
 end
diff --git a/lib/pleroma/web/streamer/streamer.ex b/lib/pleroma/web/streamer/streamer.ex
index 814d5a729..b7294d084 100644
--- a/lib/pleroma/web/streamer/streamer.ex
+++ b/lib/pleroma/web/streamer/streamer.ex
@@ -3,12 +3,77 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.Streamer do
+  alias Pleroma.User
   alias Pleroma.Web.Streamer.State
   alias Pleroma.Web.Streamer.Worker
 
   @timeout 60_000
   @mix_env Mix.env()
 
+  @public_streams ["public", "public:local", "public:media", "public:local:media"]
+  @user_streams ["user", "user:notification", "direct"]
+
+  @doc "Expands and authorizes a stream, and registers the process for streaming."
+  @spec get_topic_and_add_socket(stream :: String.t(), State.t(), Map.t() | nil) ::
+          {:ok, topic :: String.t()} | {:error, :bad_topic} | {:error, :unauthorized}
+  def get_topic_and_add_socket(stream, socket, params \\ %{}) do
+    user =
+      case socket do
+        %{assigns: %{user: user}} -> user
+        _ -> nil
+      end
+
+    case get_topic(stream, user, params) do
+      {:ok, topic} ->
+        add_socket(topic, socket)
+        {:ok, topic}
+
+      error ->
+        error
+    end
+  end
+
+  @doc "Expand and authorizes a stream"
+  @spec get_topic(stream :: String.t(), User.t() | nil, Map.t()) ::
+          {:ok, topic :: String.t()} | {:error, :bad_topic}
+  def get_topic(stream, user, params \\ %{})
+
+  # Allow all public steams.
+  def get_topic(stream, _, _) when stream in @public_streams do
+    {:ok, stream}
+  end
+
+  # Allow all hashtags streams.
+  def get_topic("hashtag", _, %{"tag" => tag}) do
+    {:ok, "hashtag:" <> tag}
+  end
+
+  # Expand user streams.
+  def get_topic(stream, %User{} = user, _) when stream in @user_streams do
+    {:ok, stream <> ":" <> to_string(user.id)}
+  end
+
+  def get_topic(stream, _, _) when stream in @user_streams do
+    {:error, :unauthorized}
+  end
+
+  # List streams.
+  def get_topic("list", %User{} = user, %{"list" => id}) do
+    if Pleroma.List.get(id, user) do
+      {:ok, "list:" <> to_string(id)}
+    else
+      {:error, :bad_topic}
+    end
+  end
+
+  def get_topic("list", _, _) do
+    {:error, :unauthorized}
+  end
+
+  def get_topic(_, _, _) do
+    {:error, :bad_topic}
+  end
+
   def add_socket(topic, socket) do
     State.add_socket(topic, socket)
   end