Merge branch 'release/2.0.5' into 'stable'v2.0.5

Release/2.0.5

See merge request pleroma/secteam/pleroma!4

2 files changed, 69 insertions, 15 deletions

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