Enforce unauth restrictions for public streaming endpoints

lib/pleroma/web/streamer.ex

@@ -25,6 +25,7 @@ defmodule Pleroma.Web.Streamer do
   def registry, do: @registry
 
   @public_streams ["public", "public:local", "public:media", "public:local:media"]
+  @local_streams ["public:local", "public:local:media"]
   @user_streams ["user", "user:notification", "direct"]
 
   @doc "Expands and authorizes a stream, and registers the process for streaming."
@@ -41,14 +42,37 @@ def get_topic_and_add_socket(stream, user, oauth_token, params \\ %{}) do
     end
   end
 
+  defp can_access_stream(user, oauth_token, kind) do
+    with {_, true} <- {:restrict?, Config.restrict_unauthenticated_access?(:timelines, kind)},
+         {_, %User{id: user_id}, %Token{user_id: user_id}} <- {:user, user, oauth_token},
+         {_, true} <-
+           {:scopes,
+            OAuthScopesPlug.filter_descendants(["read:statuses"], oauth_token.scopes) != []} do
+      true
+    else
+      {:restrict?, _} ->
+        true
+
+      _ ->
+        false
+    end
+  end
+
   @doc "Expand and authorizes a stream"
   @spec get_topic(stream :: String.t(), User.t() | nil, Token.t() | nil, Map.t()) ::
           {:ok, topic :: String.t()} | {:error, :bad_topic}
   def get_topic(stream, user, oauth_token, params \\ %{})
 
-  # Allow all public steams.
+  # Allow all public steams if the instance allows unauthenticated access.
-  def get_topic(stream, _user, _oauth_token, _params) when stream in @public_streams do
+  # Otherwise, only allow users with valid oauth tokens.
+  def get_topic(stream, user, oauth_token, _params) when stream in @public_streams do
+    kind = if stream in @local_streams, do: :local, else: :federated
+
+    if can_access_stream(user, oauth_token, kind) do
       {:ok, stream}
+    else
+      {:error, :unauthorized}
+    end
   end
 
   # Allow all hashtags streams.
@@ -57,12 +81,20 @@ def get_topic("hashtag", _user, _oauth_token, %{"tag" => tag} = _params) do
   end
 
   # Allow remote instance streams.
-  def get_topic("public:remote", _user, _oauth_token, %{"instance" => instance} = _params) do
+  def get_topic("public:remote", user, oauth_token, %{"instance" => instance} = _params) do
+    if can_access_stream(user, oauth_token, :federated) do
       {:ok, "public:remote:" <> instance}
+    else
+      {:error, :unauthorized}
+    end
   end
 
-  def get_topic("public:remote:media", _user, _oauth_token, %{"instance" => instance} = _params) do
+  def get_topic("public:remote:media", user, oauth_token, %{"instance" => instance} = _params) do
+    if can_access_stream(user, oauth_token, :federated) do
       {:ok, "public:remote:media:" <> instance}
+    else
+      {:error, :unauthorized}
+    end
   end
 
   # Expand user streams.

test/pleroma/web/streamer_test.exs

@@ -25,6 +25,26 @@ test "allows public" do
       assert {:ok, "public:local:media"} = Streamer.get_topic("public:local:media", nil, nil)
     end
 
+    test "rejects local public streams if restricted_unauthenticated is on" do
+      clear_config([:restrict_unauthenticated, :timelines, :local], true)
+
+      assert {:error, :unauthorized} = Streamer.get_topic("public:local", nil, nil)
+      assert {:error, :unauthorized} = Streamer.get_topic("public:local:media", nil, nil)
+    end
+
+    test "rejects remote public streams if restricted_unauthenticated is on" do
+      clear_config([:restrict_unauthenticated, :timelines, :federated], true)
+
+      assert {:error, :unauthorized} = Streamer.get_topic("public", nil, nil)
+      assert {:error, :unauthorized} = Streamer.get_topic("public:media", nil, nil)
+
+      assert {:error, :unauthorized} =
+               Streamer.get_topic("public:remote", nil, nil, %{"instance" => "lain.com"})
+
+      assert {:error, :unauthorized} =
+               Streamer.get_topic("public:remote:media", nil, nil, %{"instance" => "lain.com"})
+    end
+
     test "allows instance streams" do
       assert {:ok, "public:remote:lain.com"} =
                Streamer.get_topic("public:remote", nil, nil, %{"instance" => "lain.com"})
@@ -65,6 +85,63 @@ test "allows public streams (regardless of OAuth token scopes)", %{
       end
     end
 
+    test "allows local public streams if restricted_unauthenticated is on", %{
+      user: user,
+      token: oauth_token
+    } do
+      clear_config([:restrict_unauthenticated, :timelines, :local], true)
+
+      %{token: read_notifications_token} = oauth_access(["read:notifications"], user: user)
+      %{token: badly_scoped_token} = oauth_access(["irrelevant:scope"], user: user)
+
+      assert {:ok, "public:local"} = Streamer.get_topic("public:local", user, oauth_token)
+
+      assert {:ok, "public:local:media"} =
+               Streamer.get_topic("public:local:media", user, oauth_token)
+
+      for token <- [read_notifications_token, badly_scoped_token] do
+        assert {:error, :unauthorized} = Streamer.get_topic("public:local", user, token)
+
+        assert {:error, :unauthorized} = Streamer.get_topic("public:local:media", user, token)
+      end
+    end
+
+    test "allows remote public streams if restricted_unauthenticated is on", %{
+      user: user,
+      token: oauth_token
+    } do
+      clear_config([:restrict_unauthenticated, :timelines, :federated], true)
+
+      %{token: read_notifications_token} = oauth_access(["read:notifications"], user: user)
+      %{token: badly_scoped_token} = oauth_access(["irrelevant:scope"], user: user)
+
+      assert {:ok, "public"} = Streamer.get_topic("public", user, oauth_token)
+      assert {:ok, "public:media"} = Streamer.get_topic("public:media", user, oauth_token)
+
+      assert {:ok, "public:remote:lain.com"} =
+               Streamer.get_topic("public:remote", user, oauth_token, %{"instance" => "lain.com"})
+
+      assert {:ok, "public:remote:media:lain.com"} =
+               Streamer.get_topic("public:remote:media", user, oauth_token, %{
+                 "instance" => "lain.com"
+               })
+
+      for token <- [read_notifications_token, badly_scoped_token] do
+        assert {:error, :unauthorized} = Streamer.get_topic("public", user, token)
+        assert {:error, :unauthorized} = Streamer.get_topic("public:media", user, token)
+
+        assert {:error, :unauthorized} =
+                 Streamer.get_topic("public:remote", user, token, %{
+                   "instance" => "lain.com"
+                 })
+
+        assert {:error, :unauthorized} =
+                 Streamer.get_topic("public:remote:media", user, token, %{
+                   "instance" => "lain.com"
+                 })
+      end
+    end
+
     test "allows user streams (with proper OAuth token scopes)", %{
       user: user,
       token: read_oauth_token