diff --git a/changelog.d/oauth-user-preload.change b/changelog.d/oauth-user-preload.change new file mode 100644 index 000000000..43f5ea06b --- /dev/null +++ b/changelog.d/oauth-user-preload.change @@ -0,0 +1 @@ +Avoid a duplicate user query during OAuth authentication. diff --git a/changelog.d/publisher-worker-atoms.fix b/changelog.d/publisher-worker-atoms.fix new file mode 100644 index 000000000..2df6608ba --- /dev/null +++ b/changelog.d/publisher-worker-atoms.fix @@ -0,0 +1 @@ +Avoid creating atoms from serialized PublisherWorker parameters. diff --git a/changelog.d/stream-domain-block-following.fix b/changelog.d/stream-domain-block-following.fix new file mode 100644 index 000000000..84890c680 --- /dev/null +++ b/changelog.d/stream-domain-block-following.fix @@ -0,0 +1 @@ +Keep followed users visible in streams when their domain is blocked. diff --git a/changelog.d/user-cache-keys.fix b/changelog.d/user-cache-keys.fix new file mode 100644 index 000000000..4af21dd1f --- /dev/null +++ b/changelog.d/user-cache-keys.fix @@ -0,0 +1 @@ +Fix inconsistent User cache keys and invalidate numeric ID aliases. diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex index 75da41da9..768a6f3c2 100644 --- a/lib/pleroma/user.ex +++ b/lib/pleroma/user.ex @@ -272,13 +272,13 @@ defmodule Pleroma.User do end def cached_blocked_users_ap_ids(user) do - @cachex.fetch!(:user_cache, "blocked_users_ap_ids:#{user.ap_id}", fn _ -> + @cachex.fetch!(:user_cache, cache_key(:blocked_users_ap_ids, user), fn _ -> blocked_users_ap_ids(user) end) end def cached_muted_users_ap_ids(user) do - @cachex.fetch!(:user_cache, "muted_users_ap_ids:#{user.ap_id}", fn _ -> + @cachex.fetch!(:user_cache, cache_key(:muted_users_ap_ids, user), fn _ -> muted_users_ap_ids(user) end) end @@ -1253,13 +1253,18 @@ defmodule Pleroma.User do get_cached_by_nickname(nickname) end + defp cache_key(:id, %User{id: id}), do: cache_key(:id, id) + defp cache_key(:nickname, %User{nickname: nickname}), do: cache_key(:nickname, nickname) + defp cache_key(namespace, %User{ap_id: ap_id}), do: cache_key(namespace, ap_id) + defp cache_key(namespace, value), do: "#{namespace}:#{value}" + def set_cache({:ok, user}), do: set_cache(user) def set_cache({:error, err}), do: {:error, err} def set_cache(%User{} = user) do - @cachex.put(:user_cache, "ap_id:#{user.ap_id}", user) - @cachex.put(:user_cache, "nickname:#{user.nickname}", user) - @cachex.put(:user_cache, "friends_ap_ids:#{user.nickname}", get_user_friends_ap_ids(user)) + @cachex.put(:user_cache, cache_key(:ap_id, user), user) + @cachex.put(:user_cache, cache_key(:nickname, user), user) + @cachex.put(:user_cache, cache_key(:friends_ap_ids, user), get_user_friends_ap_ids(user)) {:ok, user} end @@ -1287,22 +1292,23 @@ defmodule Pleroma.User do @spec get_cached_user_friends_ap_ids(User.t()) :: [String.t()] def get_cached_user_friends_ap_ids(user) do - @cachex.fetch!(:user_cache, "friends_ap_ids:#{user.ap_id}", fn _ -> + @cachex.fetch!(:user_cache, cache_key(:friends_ap_ids, user), fn _ -> get_user_friends_ap_ids(user) end) end def invalidate_cache(user) do - @cachex.del(:user_cache, "ap_id:#{user.ap_id}") - @cachex.del(:user_cache, "nickname:#{user.nickname}") - @cachex.del(:user_cache, "friends_ap_ids:#{user.ap_id}") - @cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}") - @cachex.del(:user_cache, "muted_users_ap_ids:#{user.ap_id}") + @cachex.del(:user_cache, cache_key(:ap_id, user)) + @cachex.del(:user_cache, cache_key(:nickname, user)) + @cachex.del(:user_cache, cache_key(:id, user)) + @cachex.del(:user_cache, cache_key(:friends_ap_ids, user)) + @cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user)) + @cachex.del(:user_cache, cache_key(:muted_users_ap_ids, user)) end @spec get_cached_by_ap_id(String.t()) :: User.t() | nil def get_cached_by_ap_id(ap_id) do - key = "ap_id:#{ap_id}" + key = cache_key(:ap_id, ap_id) with {:ok, nil} <- @cachex.get(:user_cache, key), user when not is_nil(user) <- get_by_ap_id(ap_id), @@ -1315,14 +1321,14 @@ defmodule Pleroma.User do end def get_cached_by_id(id) do - key = "id:#{id}" + key = cache_key(:id, id) ap_id = @cachex.fetch!(:user_cache, key, fn _ -> user = get_by_id(id) if user do - @cachex.put(:user_cache, "ap_id:#{user.ap_id}", user) + @cachex.put(:user_cache, cache_key(:ap_id, user), user) {:commit, user.ap_id} else {:ignore, ""} @@ -1333,7 +1339,7 @@ defmodule Pleroma.User do end def get_cached_by_nickname(nickname) do - key = "nickname:#{nickname}" + key = cache_key(:nickname, nickname) @cachex.fetch!(:user_cache, key, fn _ -> case get_or_fetch_by_nickname(nickname) do @@ -1651,7 +1657,7 @@ defmodule Pleroma.User do |> Oban.insert() end - @cachex.del(:user_cache, "muted_users_ap_ids:#{muter.ap_id}") + @cachex.del(:user_cache, cache_key(:muted_users_ap_ids, muter)) {:ok, Enum.filter([user_mute, user_notification_mute], & &1)} end @@ -1661,7 +1667,7 @@ defmodule Pleroma.User do with {:ok, user_mute} <- UserRelationship.delete_mute(muter, mutee), {:ok, user_notification_mute} <- UserRelationship.delete_notification_mute(muter, mutee) do - @cachex.del(:user_cache, "muted_users_ap_ids:#{muter.ap_id}") + @cachex.del(:user_cache, cache_key(:muted_users_ap_ids, muter)) {:ok, [user_mute, user_notification_mute]} end end @@ -2815,7 +2821,7 @@ defmodule Pleroma.User do {:ok, UserRelationship.t()} | {:error, Ecto.Changeset.t()} defp add_to_block(%User{} = user, %User{} = blocked, expires_at) do with {:ok, relationship} <- UserRelationship.create_block(user, blocked, expires_at) do - @cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}") + @cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user)) {:ok, relationship} end end @@ -2824,7 +2830,7 @@ defmodule Pleroma.User do {:ok, UserRelationship.t()} | {:ok, nil} | {:error, Ecto.Changeset.t()} defp remove_from_block(%User{} = user, %User{} = blocked) do with {:ok, relationship} <- UserRelationship.delete_block(user, blocked) do - @cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}") + @cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user)) {:ok, relationship} end end diff --git a/lib/pleroma/web/federator.ex b/lib/pleroma/web/federator.ex index 90cd2e54a..88b267331 100644 --- a/lib/pleroma/web/federator.ex +++ b/lib/pleroma/web/federator.ex @@ -89,7 +89,8 @@ defmodule Pleroma.Web.Federator do # Job Worker Callbacks - @spec perform(atom(), any()) :: {:ok, any()} | {:error, any()} + @impl true + @spec perform(atom(), any()) :: :ok | {:ok, any()} | {:error, any()} def perform(:publish_one, params) do Publisher.prepare_one(params) |> Publisher.publish_one() diff --git a/lib/pleroma/web/federator/publishing.ex b/lib/pleroma/web/federator/publishing.ex index 3a242b8dc..889b6b83e 100644 --- a/lib/pleroma/web/federator/publishing.ex +++ b/lib/pleroma/web/federator/publishing.ex @@ -3,5 +3,6 @@ # SPDX-License-Identifier: AGPL-3.0-only defmodule Pleroma.Web.Federator.Publishing do + @callback perform(atom(), any()) :: :ok | {:ok, any()} | {:error, any()} @callback publish(map()) :: any() end diff --git a/lib/pleroma/web/mastodon_api/controllers/status_controller.ex b/lib/pleroma/web/mastodon_api/controllers/status_controller.ex index 26e38aac8..6cd7469f9 100644 --- a/lib/pleroma/web/mastodon_api/controllers/status_controller.ex +++ b/lib/pleroma/web/mastodon_api/controllers/status_controller.ex @@ -716,7 +716,10 @@ defmodule Pleroma.Web.MastodonAPI.StatusController do end end - defp put_application(params, %{assigns: %{token: %Token{user: %User{} = user} = token}} = _conn) do + defp put_application( + params, + %{assigns: %{token: %Token{} = token, user: %User{} = user}} = _conn + ) do if user.disclose_client do %{client_name: client_name, website: website} = Repo.preload(token, :app).app Map.put(params, :generator, %{type: "Application", name: client_name, url: website}) diff --git a/lib/pleroma/web/plugs/o_auth_plug.ex b/lib/pleroma/web/plugs/o_auth_plug.ex index 488968691..b59ac9d3e 100644 --- a/lib/pleroma/web/plugs/o_auth_plug.ex +++ b/lib/pleroma/web/plugs/o_auth_plug.ex @@ -52,7 +52,7 @@ defmodule Pleroma.Web.Plugs.OAuthPlug do where: t.token == ^token ) - with %Token{user_id: user_id} = token_record <- Repo.one(token_query) |> Repo.preload(:user), + with %Token{user_id: user_id} = token_record <- Repo.one(token_query), false <- is_nil(user_id), %User{} = user <- User.get_cached_by_id(user_id) do {:ok, user, token_record} diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex index 14d942c9f..3c154288b 100644 --- a/lib/pleroma/web/streamer.ex +++ b/lib/pleroma/web/streamer.ex @@ -195,6 +195,9 @@ defmodule Pleroma.Web.Streamer do recipients = MapSet.new(item.recipients) domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks) + following_ap_ids = + if user.domain_blocks == [], do: [], else: User.get_cached_user_friends_ap_ids(user) + with parent <- Object.normalize(item, fetch: false) || item, true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)), true <- item.data["type"] != "Announce" || item.actor not in reblog_muted_ap_ids, @@ -205,8 +208,12 @@ defmodule Pleroma.Web.Streamer do true <- MapSet.disjoint?(recipients, recipient_blocks), %{host: item_host} <- URI.parse(item.actor), %{host: parent_host} <- URI.parse(parent.data["actor"]), - false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host), - false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host), + false <- + item.actor not in following_ap_ids and + Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host), + false <- + parent.data["actor"] not in following_ap_ids and + Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host), true <- thread_containment(item, user), false <- CommonAPI.thread_muted?(parent, user) do false diff --git a/lib/pleroma/workers/publisher_worker.ex b/lib/pleroma/workers/publisher_worker.ex index f799af77a..186f6f4ef 100644 --- a/lib/pleroma/workers/publisher_worker.ex +++ b/lib/pleroma/workers/publisher_worker.ex @@ -4,25 +4,39 @@ defmodule Pleroma.Workers.PublisherWorker do alias Pleroma.Activity + alias Pleroma.Config alias Pleroma.Instances alias Pleroma.Web.Federator use Oban.Worker, queue: :federator_outgoing, max_attempts: 13 + @publish_one_param_keys %{ + "activity_id" => :activity_id, + "cc" => :cc, + "inbox" => :inbox, + "unreachable_since" => :unreachable_since + } + @impl true def perform(%Job{args: %{"op" => "publish", "activity_id" => activity_id}}) do activity = Activity.get_by_id(activity_id) - Federator.perform(:publish, activity) + federator().perform(:publish, activity) end def perform(%Job{args: %{"op" => "publish_one", "params" => params}} = job) do - params = Map.new(params, fn {k, v} -> {String.to_atom(k), v} end) + params = + Enum.reduce(params, %{}, fn {key, value}, acc -> + case @publish_one_param_keys do + %{^key => atom_key} -> Map.put(acc, atom_key, value) + _ -> acc + end + end) # Cancel / skip the job if this server believed to be unreachable now if not Instances.reachable?(params.inbox) do {:cancel, :unreachable} else - case Federator.perform(:publish_one, params) do + case federator().perform(:publish_one, params) do {:ok, _} -> :ok @@ -44,6 +58,8 @@ defmodule Pleroma.Workers.PublisherWorker do @impl true def timeout(_job), do: :timer.seconds(10) + defp federator, do: Config.get([:pipeline, :federator], Federator) + @base_backoff 15 @pow 5 @impl true diff --git a/test/pleroma/user_test.exs b/test/pleroma/user_test.exs index de735cdb7..9ae9cce26 100644 --- a/test/pleroma/user_test.exs +++ b/test/pleroma/user_test.exs @@ -1988,14 +1988,26 @@ defmodule Pleroma.UserTest do end describe "caching" do + test "set_cache stores friends under the AP ID cache key" do + user = insert(:user) + + User.set_cache(user) + + assert {:ok, []} = Cachex.get(:user_cache, "friends_ap_ids:#{user.ap_id}") + assert {:ok, nil} = Cachex.get(:user_cache, "friends_ap_ids:#{user.nickname}") + end + test "invalidate_cache works" do user = insert(:user) User.set_cache(user) + User.get_cached_by_id(user.id) User.invalidate_cache(user) {:ok, nil} = Cachex.get(:user_cache, "ap_id:#{user.ap_id}") {:ok, nil} = Cachex.get(:user_cache, "nickname:#{user.nickname}") + {:ok, nil} = Cachex.get(:user_cache, "friends_ap_ids:#{user.ap_id}") + {:ok, nil} = Cachex.get(:user_cache, "id:#{user.id}") end test "User.delete() plugs any possible zombie objects" do diff --git a/test/pleroma/web/plugs/o_auth_plug_test.exs b/test/pleroma/web/plugs/o_auth_plug_test.exs index 8917eadd1..2509e4203 100644 --- a/test/pleroma/web/plugs/o_auth_plug_test.exs +++ b/test/pleroma/web/plugs/o_auth_plug_test.exs @@ -44,6 +44,18 @@ defmodule Pleroma.Web.Plugs.OAuthPlugTest do assert conn.assigns[:user] == opts[:user] end + test "with a valid token, it does not preload the duplicate user association", %{ + conn: conn, + token: token + } do + conn = + conn + |> put_req_header("authorization", "bearer #{token.token}") + |> OAuthPlug.call(%{}) + + assert %Ecto.Association.NotLoaded{} = conn.assigns.token.user + end + test "with valid token (downcase) in url parameters, it assigns the user", opts do conn = :get diff --git a/test/pleroma/web/streamer_test.exs b/test/pleroma/web/streamer_test.exs index b26bd1847..7610b3379 100644 --- a/test/pleroma/web/streamer_test.exs +++ b/test/pleroma/web/streamer_test.exs @@ -475,6 +475,20 @@ defmodule Pleroma.Web.StreamerTest do assert Streamer.filtered_by_user?(user, favorite_activity) end + test "it does not filter followed users on blocked domains", %{user: user} do + followed = + insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/friend"}) + + {:ok, user} = User.block_domain(user, "hecking-lewd-place.com") + {:ok, activity} = CommonAPI.post(followed, %{status: "still a friend"}) + + assert Streamer.filtered_by_user?(user, activity) + + {:ok, user, _followed} = User.follow(user, followed) + + refute Streamer.filtered_by_user?(user, activity) + end + test "it sends follow activities to the 'user:notification' stream", %{ user: user, token: oauth_token diff --git a/test/pleroma/workers/publisher_worker_test.exs b/test/pleroma/workers/publisher_worker_test.exs index 4519864a5..ddbbba910 100644 --- a/test/pleroma/workers/publisher_worker_test.exs +++ b/test/pleroma/workers/publisher_worker_test.exs @@ -6,8 +6,8 @@ defmodule Pleroma.Workers.PublisherWorkerTest do use Pleroma.DataCase, async: false use Oban.Testing, repo: Pleroma.Repo + import Mox import Pleroma.Factory - import Mock alias Pleroma.Instances alias Pleroma.Object @@ -15,6 +15,8 @@ defmodule Pleroma.Workers.PublisherWorkerTest do alias Pleroma.Web.ActivityPub.Builder alias Pleroma.Web.CommonAPI alias Pleroma.Web.Federator + alias Pleroma.Web.FederatorMock + alias Pleroma.Workers.PublisherWorker describe "Oban job priority:" do setup do @@ -55,49 +57,80 @@ defmodule Pleroma.Workers.PublisherWorkerTest do end test "marks server as unreachable only on final failure", %{activity: activity} do - with_mock Pleroma.Web.Federator, - perform: fn :publish_one, _params -> {:error, :connection_error} end do - # First attempt - job = %Oban.Job{ - args: %{ - "op" => "publish_one", - "params" => %{ - "inbox" => "https://example.com/inbox", - "activity_id" => activity.id - } - }, - attempt: 1, - max_attempts: 5 - } + expect(FederatorMock, :perform, 2, fn :publish_one, _params -> + {:error, :connection_error} + end) - assert {:error, :connection_error} = Pleroma.Workers.PublisherWorker.perform(job) - assert Instances.reachable?("https://example.com/inbox") + # First attempt + job = %Oban.Job{ + args: %{ + "op" => "publish_one", + "params" => %{ + "inbox" => "https://example.com/inbox", + "activity_id" => activity.id + } + }, + attempt: 1, + max_attempts: 5 + } - # Final attempt - job = %{job | attempt: 5} - assert {:error, :connection_error} = Pleroma.Workers.PublisherWorker.perform(job) - refute Instances.reachable?("https://example.com/inbox") - end + assert {:error, :connection_error} = PublisherWorker.perform(job) + assert Instances.reachable?("https://example.com/inbox") + + # Final attempt + job = %{job | attempt: 5} + assert {:error, :connection_error} = PublisherWorker.perform(job) + refute Instances.reachable?("https://example.com/inbox") end test "does not mark server as unreachable on successful publish", %{activity: activity} do - with_mock Pleroma.Web.Federator, - perform: fn :publish_one, _params -> {:ok, %{status: 200}} end do - job = %Oban.Job{ - args: %{ - "op" => "publish_one", - "params" => %{ - "inbox" => "https://example.com/inbox", - "activity_id" => activity.id - } - }, - attempt: 1, - max_attempts: 5 - } + expect(FederatorMock, :perform, fn :publish_one, _params -> {:ok, %{status: 200}} end) - assert :ok = Pleroma.Workers.PublisherWorker.perform(job) - assert Instances.reachable?("https://example.com/inbox") - end + job = %Oban.Job{ + args: %{ + "op" => "publish_one", + "params" => %{ + "inbox" => "https://example.com/inbox", + "activity_id" => activity.id + } + }, + attempt: 1, + max_attempts: 5 + } + + assert :ok = PublisherWorker.perform(job) + assert Instances.reachable?("https://example.com/inbox") + end + + test "does not create atoms from unexpected job params", %{activity: activity} do + unknown_key = "unknown_#{System.unique_integer([:positive])}" + + assert_raise ArgumentError, fn -> String.to_existing_atom(unknown_key) end + + expect(FederatorMock, :perform, fn :publish_one, params -> + assert params == %{ + inbox: "https://example.com/inbox", + activity_id: activity.id + } + + {:ok, %{status: 200}} + end) + + job = %Oban.Job{ + args: %{ + "op" => "publish_one", + "params" => %{ + "inbox" => "https://example.com/inbox", + "activity_id" => activity.id, + unknown_key => "ignored" + } + }, + attempt: 1, + max_attempts: 5 + } + + assert :ok = PublisherWorker.perform(job) + assert_raise ArgumentError, fn -> String.to_existing_atom(unknown_key) end end test "cancels job if server is unreachable", %{activity: activity} do @@ -117,7 +150,7 @@ defmodule Pleroma.Workers.PublisherWorkerTest do max_attempts: 5 } - assert {:cancel, :unreachable} = Pleroma.Workers.PublisherWorker.perform(job) + assert {:cancel, :unreachable} = PublisherWorker.perform(job) end end end