No OneTemporary
Actions

Authored By

Unknown

Size

30 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/federator.ex b/lib/pleroma/web/federator.ex
	index 88b267331..003fdc00c 100644
	--- a/lib/pleroma/web/federator.ex
	+++ b/lib/pleroma/web/federator.ex
	@@ -1,146 +1,181 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Federator do
	alias Pleroma.Activity
	alias Pleroma.Object.Containment
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Publisher
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.ActivityPub.Utils
	alias Pleroma.Workers.PublisherWorker
	alias Pleroma.Workers.ReceiverWorker
	alias Pleroma.Workers.SignatureRetryWorker

	require Logger

	@behaviour Pleroma.Web.Federator.Publishing

	@doc """
	Returns `true` if the distance to target object does not exceed max configured value.
	Serves to prevent fetching of very long threads, especially useful on smaller instances.
	Addresses [memory leaks on recursive replies fetching](https://git.pleroma.social/pleroma/pleroma/issues/161).
	Applies to fetching of both ancestor (reply-to) and child (reply) objects.
	"""
	# credo:disable-for-previous-line Credo.Check.Readability.MaxLineLength
	def allowed_thread_distance?(distance) do
	max_distance = Pleroma.Config.get([:instance, :federation_incoming_replies_max_depth])

	if max_distance && max_distance >= 0 do
	# Default depth is 0 (an object has zero distance from itself in its thread)
	(distance \|\| 0) <= max_distance
	else
	true
	end
	end

	# Client API
	def incoming_failed_signature_ap_doc(%{
	method: method,
	params: params,
	req_headers: req_headers,
	request_path: request_path,
	query_string: query_string
	}) do
	SignatureRetryWorker.new(
	%{
	"op" => "incoming_failed_signature_ap_doc",
	"method" => method,
	"req_headers" => req_headers,
	"params" => params,
	"request_path" => request_path,
	"query_string" => query_string,
	"timeout" => :timer.seconds(20)
	},
	priority: 2
	)
	\|> Oban.insert()
	end

	- def incoming_ap_doc(%{"type" => "Delete"} = params) do
	- ReceiverWorker.new(%{"op" => "incoming_ap_doc", "params" => params},
	- priority: 3,
	- queue: :slow
	- )
	- \|> Oban.insert()
	- end
	-
	def incoming_ap_doc(params) do
	- ReceiverWorker.new(%{"op" => "incoming_ap_doc", "params" => params})
	+ options =
	+ if delete_activity?(params) do
	+ [
	+ priority: 3,
	+ queue: :slow,
	+ unique: [period: :infinity, states: [:available, :scheduled, :executing, :retryable]]
	+ ]
	+ else
	+ []
	+ end
	+
	+ ReceiverWorker.new(%{"op" => "incoming_ap_doc", "params" => params}, options)
	\|> Oban.insert()
	end

	@impl true
	def publish(%{id: "pleroma:fakeid"} = activity) do
	perform(:publish, activity)
	end

	@impl true
	def publish(%Pleroma.Activity{data: %{"type" => type}} = activity) do
	PublisherWorker.new(%{"op" => "publish", "activity_id" => activity.id},
	priority: publish_priority(type)
	)
	\|> Oban.insert()
	end

	defp publish_priority("Delete"), do: 3
	defp publish_priority(_), do: 0

	# Job Worker Callbacks

	@impl true
	@spec perform(atom(), any()) :: :ok \| {:ok, any()} \| {:error, any()}
	def perform(:publish_one, params) do
	Publisher.prepare_one(params)
	\|> Publisher.publish_one()
	end

	def perform(:publish, activity) do
	Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end)

	%User{} = actor = User.get_cached_by_ap_id(activity.data["actor"])
	Publisher.publish(actor, activity)
	end

	def perform(:incoming_ap_doc, params) do
	Logger.debug("Handling incoming AP activity")

	actor =
	params
	\|> Map.get("actor")
	\|> Utils.get_ap_id()

	+ existing_activity = Activity.normalize(params["id"])
	+
	# NOTE: we use the actor ID to do the containment, this is fine because an
	# actor shouldn't be acting on objects outside their own AP server.
	with {_, {:ok, user}} <- {:actor, User.get_or_fetch_by_ap_id(actor)},
	- {:user_active, true} <- {:user_active, match?(true, user.is_active)},
	- nil <- Activity.normalize(params["id"]),
	+ {:activity_state, :ok} <-
	+ {:activity_state, incoming_activity_state(user, existing_activity, params)},
	{_, :ok} <-
	{:correct_origin?, Containment.contain_origin_from_id(actor, params)},
	{:ok, activity} <- Transmogrifier.handle_incoming(params) do
	{:ok, activity}
	else
	{:correct_origin?, _} ->
	Logger.debug("Origin containment failure for #{params["id"]}")
	{:error, :origin_containment_failed}

	- %Activity{} ->
	+ {:activity_state, :already_present} ->
	Logger.debug("Already had #{params["id"]}")
	{:error, :already_present}

	+ {:activity_state, {:user_active, false} = reason} ->
	+ {:error, reason}
	+
	{:actor, e} ->
	Logger.debug("Unhandled actor #{actor}, #{inspect(e)}")
	{:error, e}

	{:reject, reason} = e ->
	Logger.debug("Rejected by MRF: #{inspect(reason)}")
	{:error, e}

	e ->
	# Just drop those for now
	Logger.debug(fn -> "Unhandled activity\n" <> Jason.encode!(params, pretty: true) end)
	{:error, e}
	end
	end
	+
	+ defp incoming_activity_state(user, %Activity{data: existing_data}, params) do
	+ cond do
	+ exact_delete_replay?(existing_data, params) -> :ok
	+ user.is_active != true -> {:user_active, false}
	+ true -> :already_present
	+ end
	+ end
	+
	+ defp incoming_activity_state(%User{is_active: true}, nil, _params), do: :ok
	+ defp incoming_activity_state(_user, nil, _params), do: {:user_active, false}
	+
	+ defp exact_delete_replay?(existing_data, params) do
	+ existing_actor = Utils.get_ap_id(existing_data["actor"])
	+ incoming_actor = Utils.get_ap_id(params["actor"])
	+ existing_object = Utils.get_ap_id(existing_data["object"])
	+ incoming_object = Utils.get_ap_id(params["object"])
	+
	+ delete_activity?(existing_data) and delete_activity?(params) and
	+ is_binary(existing_data["id"]) and existing_data["id"] == params["id"] and
	+ is_binary(existing_actor) and existing_actor == incoming_actor and
	+ is_binary(existing_object) and existing_object == incoming_object
	+ end
	+
	+ defp delete_activity?(%{"type" => "Delete"}), do: true
	+ defp delete_activity?(%{"type" => types}) when is_list(types), do: "Delete" in types
	+ defp delete_activity?(_), do: false
	end
	diff --git a/lib/pleroma/workers/receiver_worker.ex b/lib/pleroma/workers/receiver_worker.ex
	index 3afbe138d..3825ace68 100644
	--- a/lib/pleroma/workers/receiver_worker.ex
	+++ b/lib/pleroma/workers/receiver_worker.ex
	@@ -1,87 +1,88 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Workers.ReceiverWorker do
	alias Pleroma.Instances
	alias Pleroma.Web.Federator
	alias Pleroma.Workers.SignatureRetryWorker

	use Oban.Worker, queue: :federator_incoming, max_attempts: 5, unique: [period: :infinity]

	@impl true
	def perform(%Job{args: %{"op" => "incoming_ap_doc", "params" => params} = args} = job) do
	if signature_retry_job?(args) do
	perform_signature_retry(job)
	else
	perform_incoming(params)
	end
	end

	def perform(%Job{args: %{"op" => "incoming_ap_doc"} = args} = job) do
	if signature_retry_job?(args) do
	perform_signature_retry(job)
	else
	process_errors(:missing_incoming_ap_doc_params)
	end
	end

	defp perform_signature_retry(%Job{args: args} = job) do
	SignatureRetryWorker.perform(%Job{
	job
	\| args: Map.put(args, "op", "incoming_failed_signature_ap_doc")
	})
	end

	defp perform_incoming(params) do
	with {:ok, res} <- Federator.perform(:incoming_ap_doc, params) do
	unless Instances.reachable?(params["actor"]) do
	domain = URI.parse(params["actor"]).host
	Oban.insert(Pleroma.Workers.ReachabilityWorker.new(%{"domain" => domain}))
	end

	{:ok, res}
	else
	e -> process_errors(e)
	end
	end

	defp signature_retry_job?(args) do
	Enum.any?(~w(method req_headers request_path query_string), &Map.has_key?(args, &1))
	end

	@impl true
	def timeout(%_{args: %{"timeout" => timeout}}), do: timeout

	def timeout(_job), do: :timer.seconds(5)

	defp process_errors({:error, {:error, _} = error}), do: process_errors(error)

	defp process_errors(errors) do
	case errors do
	# User fetch failures
	{:error, :not_found} = reason -> {:cancel, reason}
	{:error, :forbidden} = reason -> {:cancel, reason}
	# Inactive user
	{:error, {:user_active, false} = reason} -> {:cancel, reason}
	# Validator will error and return a changeset error
	# e.g., duplicate activities or if the object was deleted
	{:error, {:validate, {:error, _changeset} = reason}} -> {:cancel, reason}
	# Duplicate detection during Normalization
	{:error, :already_present} -> {:cancel, :already_present}
	+ {:error, :already_deleted} -> {:cancel, :already_deleted}
	# MRFs will return a reject
	{:error, {:reject, _} = reason} -> {:cancel, reason}
	# HTTP Sigs
	{:signature, false} -> {:cancel, :invalid_signature}
	{:same_actor, false} -> {:cancel, :actor_signature_mismatch}
	# Origin / URL validation failed somewhere possibly due to spoofing
	{:error, :origin_containment_failed} -> {:cancel, :origin_containment_failed}
	# Unclear if this can be reached
	{:error, {:side_effects, {:error, :no_object_actor}} = reason} -> {:cancel, reason}
	:missing_incoming_ap_doc_params -> {:cancel, :missing_incoming_ap_doc_params}
	# Catchall
	{:error, _} = e -> e
	e -> {:error, e}
	end
	end
	end
	diff --git a/test/pleroma/web/federator_test.exs b/test/pleroma/web/federator_test.exs
	index 16fe1066a..86821c6df 100644
	--- a/test/pleroma/web/federator_test.exs
	+++ b/test/pleroma/web/federator_test.exs
	@@ -1,206 +1,233 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.FederatorTest do
	alias Pleroma.Instances
	+ alias Pleroma.Repo
	alias Pleroma.Tests.ObanHelpers
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.Federator
	alias Pleroma.Workers.PublisherWorker

	use Pleroma.DataCase
	use Oban.Testing, repo: Pleroma.Repo

	import Pleroma.Factory
	import Mock

	setup_all do
	Tesla.Mock.mock_global(fn env -> apply(HttpRequestMock, :request, [env]) end)

	:ok
	end

	setup_all do: clear_config([:instance, :federating], true)
	setup do: clear_config([:instance, :allow_relay])
	setup do: clear_config([:mrf, :policies])
	setup do: clear_config([:mrf_keyword])

	describe "Publish an activity" do
	setup do
	user = insert(:user)
	{:ok, activity} = CommonAPI.post(user, %{status: "HI"})

	relay_mock = {
	Pleroma.Web.ActivityPub.Relay,
	[],
	[publish: fn _activity -> send(self(), :relay_publish) end]
	}

	%{activity: activity, relay_mock: relay_mock}
	end

	test "to shared inbox when multiple actors from same instance are recipients" do
	user = insert(:user)

	shared_inbox = "https://domain.com/inbox"

	follower_one =
	insert(:user, %{
	local: false,
	nickname: "nick1@domain.com",
	ap_id: "https://domain.com/users/nick1",
	inbox: "https://domain.com/users/nick1/inbox",
	shared_inbox: shared_inbox
	})

	follower_two =
	insert(:user, %{
	local: false,
	nickname: "nick2@domain.com",
	ap_id: "https://domain.com/users/nick2",
	inbox: "https://domain.com/users/nick2/inbox",
	shared_inbox: shared_inbox
	})

	{:ok, _, _} = Pleroma.User.follow(follower_one, user)
	{:ok, _, _} = Pleroma.User.follow(follower_two, user)

	{:ok, _activity} = CommonAPI.post(user, %{status: "Happy Friday everyone!"})

	ObanHelpers.perform(all_enqueued(worker: PublisherWorker))

	inboxes =
	all_enqueued(worker: PublisherWorker)
	\|> Enum.filter(&(get_in(&1, [Access.key(:args), Access.key("op")]) == "publish_one"))
	\|> Enum.map(&get_in(&1, [Access.key(:args), Access.key("params"), Access.key("inbox")]))

	assert [shared_inbox] == inboxes
	end

	test "with relays active, it publishes to the relay", %{
	activity: activity,
	relay_mock: relay_mock
	} do
	with_mocks([relay_mock]) do
	Federator.publish(activity)
	ObanHelpers.perform(all_enqueued(worker: PublisherWorker))
	end

	assert_received :relay_publish
	end

	test "with relays deactivated, it does not publish to the relay", %{
	activity: activity,
	relay_mock: relay_mock
	} do
	clear_config([:instance, :allow_relay], false)

	with_mocks([relay_mock]) do
	Federator.publish(activity)
	ObanHelpers.perform(all_enqueued(worker: PublisherWorker))
	end

	refute_received :relay_publish
	end
	end

	describe "Targets reachability filtering in `publish`" do
	test "it federates only to reachable instances via AP" do
	user = insert(:user)

	{inbox1, inbox2} =
	{"https://domain.com/users/nick1/inbox", "https://domain2.com/users/nick2/inbox"}

	insert(:user, %{
	local: false,
	nickname: "nick1@domain.com",
	ap_id: "https://domain.com/users/nick1",
	inbox: inbox1
	})

	insert(:user, %{
	local: false,
	nickname: "nick2@domain2.com",
	ap_id: "https://domain2.com/users/nick2",
	inbox: inbox2
	})

	Instances.set_unreachable(URI.parse(inbox2).host)

	{:ok, _activity} =
	CommonAPI.post(user, %{status: "HI @nick1@domain.com, @nick2@domain2.com!"})

	ObanHelpers.perform(all_enqueued(worker: PublisherWorker))

	assert ObanHelpers.member?(
	%{
	"op" => "publish_one",
	"params" => %{"inbox" => inbox1}
	},
	all_enqueued(worker: PublisherWorker)
	)
	end
	end

	describe "Receive an activity" do
	test "successfully processes incoming AP docs with correct origin" do
	params = %{
	"@context" => "https://www.w3.org/ns/activitystreams",
	"actor" => "http://mastodon.example.org/users/admin",
	"type" => "Create",
	"id" => "http://mastodon.example.org/users/admin/activities/1",
	"object" => %{
	"type" => "Note",
	"content" => "hi world!",
	"id" => "http://mastodon.example.org/users/admin/objects/1",
	"attributedTo" => "http://mastodon.example.org/users/admin",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"]
	},
	"to" => ["https://www.w3.org/ns/activitystreams#Public"]
	}

	assert {:ok, job} = Federator.incoming_ap_doc(params)
	assert {:ok, _activity} = ObanHelpers.perform(job)

	assert {:ok, job} = Federator.incoming_ap_doc(params)
	assert {:cancel, :already_present} = ObanHelpers.perform(job)
	end

	+ test "keeps in-flight Delete jobs unique and allows completed jobs to be enqueued again" do
	+ for type <- ["Delete", ["Delete"]] do
	+ params = %{
	+ "type" => type,
	+ "id" => "https://example.com/activities/delete/#{inspect(type)}",
	+ "actor" => "https://example.com/users/alice",
	+ "object" => "https://example.com/objects/note"
	+ }
	+
	+ assert {:ok, first_job} = Federator.incoming_ap_doc(params)
	+ refute first_job.conflict?
	+
	+ assert {:ok, duplicate_job} = Federator.incoming_ap_doc(params)
	+ assert duplicate_job.conflict?
	+
	+ assert {:ok, completed_job} =
	+ first_job
	+ \|> Ecto.Changeset.change(state: "completed")
	+ \|> Repo.update()
	+
	+ assert completed_job.state == "completed"
	+ assert {:ok, replay_job} = Federator.incoming_ap_doc(params)
	+ refute replay_job.conflict?
	+ end
	+ end
	+
	test "rejects incoming AP docs with incorrect origin" do
	params = %{
	"@context" => "https://www.w3.org/ns/activitystreams",
	"actor" => "https://niu.moe/users/rye",
	"type" => "Create",
	"id" => "http://mastodon.example.org/users/admin/activities/1",
	"object" => %{
	"type" => "Note",
	"content" => "hi world!",
	"id" => "http://mastodon.example.org/users/admin/objects/1",
	"attributedTo" => "http://mastodon.example.org/users/admin",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"]
	},
	"to" => ["https://www.w3.org/ns/activitystreams#Public"]
	}

	assert {:ok, job} = Federator.incoming_ap_doc(params)
	assert {:cancel, :origin_containment_failed} = ObanHelpers.perform(job)
	end

	test "it does not crash if MRF rejects the post" do
	clear_config([:mrf_keyword, :reject], ["lain"])

	clear_config(
	[:mrf, :policies],
	Pleroma.Web.ActivityPub.MRF.KeywordPolicy
	)

	params =
	File.read!("test/fixtures/mastodon-post-activity.json")
	\|> Jason.decode!()

	assert {:ok, job} = Federator.incoming_ap_doc(params)
	assert {:cancel, _} = ObanHelpers.perform(job)
	end
	end
	end
	diff --git a/test/pleroma/workers/receiver_worker_test.exs b/test/pleroma/workers/receiver_worker_test.exs
	index ea05f38f1..72b4d149e 100644
	--- a/test/pleroma/workers/receiver_worker_test.exs
	+++ b/test/pleroma/workers/receiver_worker_test.exs
	@@ -1,324 +1,382 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Workers.ReceiverWorkerTest do
	use Pleroma.DataCase, async: false
	use Oban.Testing, repo: Pleroma.Repo

	import Mock
	import Pleroma.Factory

	alias Pleroma.User
	+ alias Pleroma.Activity
	+ alias Pleroma.Object
	+ alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Workers.ReceiverWorker

	defp signature_headers_for(%User{} = signer) do
	[
	{"host", "local.test"},
	{"date", "Thu, 25 Jul 2024 13:33:31 GMT"},
	{"digest", "SHA-256=fake-digest"},
	{"content-type", "application/activity+json"},
	{
	"signature",
	"keyId=\"#{signer.ap_id}#main-key\",algorithm=\"rsa-sha256\",headers=\"(request-target) host date digest content-type\",signature=\"fake-signature\""
	}
	]
	end

	defp perform_incoming(params) do
	ReceiverWorker.perform(%Oban.Job{
	args: %{"op" => "incoming_ap_doc", "params" => params}
	})
	end

	test "it does not retry MRF reject" do
	params = insert(:note).data

	with_mock Pleroma.Web.ActivityPub.Transmogrifier,
	handle_incoming: fn _ -> {:reject, "MRF"} end do
	assert {:cancel, {:reject, "MRF"}} =
	ReceiverWorker.perform(%Oban.Job{
	args: %{"op" => "incoming_ap_doc", "params" => params}
	})
	end
	end

	test "it does not retry ObjectValidator reject" do
	params =
	insert(:note_activity).data
	\|> Map.put("id", Pleroma.Web.ActivityPub.Utils.generate_activity_id())
	\|> Map.put("object", %{
	"type" => "Note",
	"id" => Pleroma.Web.ActivityPub.Utils.generate_object_id()
	})

	with_mock Pleroma.Web.ActivityPub.ObjectValidator, [:passthrough],
	validate: fn _, _ -> {:error, %Ecto.Changeset{}} end do
	assert {:cancel, {:error, %Ecto.Changeset{}}} =
	ReceiverWorker.perform(%Oban.Job{
	args: %{"op" => "incoming_ap_doc", "params" => params}
	})
	end
	end

	test "it does not retry duplicates" do
	params = insert(:note_activity).data

	assert {:cancel, :already_present} =
	ReceiverWorker.perform(%Oban.Job{
	args: %{"op" => "incoming_ap_doc", "params" => params}
	})
	end

	+ test "it does not retry Deletes for an already deleted target" do
	+ params = %{
	+ "type" => "Delete",
	+ "id" => "https://example.com/activities/delete-duplicate",
	+ "actor" => "https://example.com/users/alice",
	+ "object" => "https://example.com/objects/note"
	+ }
	+
	+ with_mock Pleroma.Web.Federator,
	+ perform: fn :incoming_ap_doc, ^params -> {:error, {:error, :already_deleted}} end do
	+ assert {:cancel, :already_deleted} = perform_incoming(params)
	+ end
	+ end
	+
	+ test "it completes an exact replay of a persisted Delete" do
	+ user =
	+ insert(:user,
	+ local: false,
	+ ap_id: "http://mastodon.example.org/users/admin"
	+ )
	+
	+ object =
	+ insert(:note,
	+ user: user,
	+ data: %{"id" => "http://mastodon.example.org/objects/recover-delete"}
	+ )
	+
	+ activity = insert(:note_activity, user: user, note: object, local: false)
	+ object_id = activity.data["object"]
	+
	+ params =
	+ File.read!("test/fixtures/mastodon-delete.json")
	+ \|> Jason.decode!()
	+ \|> Map.put("actor", user.ap_id)
	+ \|> put_in(["object", "id"], object_id)
	+
	+ assert {:ok, persisted_delete, _meta} = ActivityPub.persist(params, local: false)
	+ assert %Object{data: %{"type" => "Note"}} = Object.get_by_ap_id(object_id)
	+ assert {:ok, %User{}} = User.set_activation(user, false)
	+
	+ replay_params = Map.put(params, "type", ["Delete"])
	+
	+ assert {:ok, %Activity{id: delete_id}} = perform_incoming(replay_params)
	+ assert delete_id == persisted_delete.id
	+ assert %Object{data: %{"type" => "Tombstone"}} = Object.get_by_ap_id(object_id)
	+ end
	+
	+ test "keeps the inactive-actor error for existing non-Delete activities" do
	+ user = insert(:user)
	+ activity = insert(:note_activity, user: user)
	+ assert {:ok, %User{}} = User.set_activation(user, false)
	+
	+ assert {:cancel, {:user_active, false}} = perform_incoming(activity.data)
	+ end
	+
	describe "cancels on a failed user fetch" do
	setup do
	Tesla.Mock.mock(fn
	%{url: "https://springfield.social/users/bart"} ->
	%Tesla.Env{
	status: 403,
	body: ""
	}

	%{url: "https://springfield.social/users/troymcclure"} ->
	%Tesla.Env{
	status: 404,
	body: ""
	}

	%{url: "https://springfield.social/users/hankscorpio"} ->
	%Tesla.Env{
	status: 410,
	body: ""
	}
	end)
	end

	test "when request returns a 403" do
	params =
	insert(:note_activity).data
	\|> Map.put("actor", "https://springfield.social/users/bart")

	assert {:cancel, {:error, :forbidden}} = perform_incoming(params)
	end

	test "when request returns a 404" do
	params =
	insert(:note_activity).data
	\|> Map.put("actor", "https://springfield.social/users/troymcclure")

	assert {:cancel, {:error, :not_found}} = perform_incoming(params)
	end

	test "when request returns a 410" do
	params =
	insert(:note_activity).data
	\|> Map.put("actor", "https://springfield.social/users/hankscorpio")

	assert {:cancel, {:error, :not_found}} = perform_incoming(params)
	end

	test "when user account is disabled" do
	user = insert(:user)

	fake_activity = URI.parse(user.ap_id) \|> Map.put(:path, "/fake-activity") \|> to_string

	params =
	insert(:note_activity, user: user).data
	\|> Map.put("id", fake_activity)

	{:ok, %User{}} = User.set_activation(user, false)

	assert {:cancel, {:user_active, false}} = perform_incoming(params)
	end
	end

	test "cancels due to origin containment" do
	params =
	insert(:note_activity).data
	\|> Map.put("id", "https://notorigindomain.com/activity")

	assert {:cancel, :origin_containment_failed} = perform_incoming(params)
	end

	test "canceled due to deleted object" do
	params =
	insert(:announce_activity).data
	\|> Map.put("object", "http://localhost:4001/deleted")

	Tesla.Mock.mock(fn
	%{url: "http://localhost:4001/deleted"} ->
	%Tesla.Env{
	status: 404,
	body: ""
	}
	end)

	assert {:cancel, _} = perform_incoming(params)
	end

	test "delegates legacy failed-signature metadata jobs instead of processing them as trusted" do
	alice = insert(:user, local: false, ap_id: "https://one.com/users/alice")
	bob = insert(:user, local: false, ap_id: "https://two.com/users/bob")
	object_id = "https://two.com/objects/legacy-forged-note"

	create = %{
	"type" => "Create",
	"actor" => bob.ap_id,
	"id" => "https://two.com/activities/legacy-forged-create",
	"context" => "https://two.com/contexts/legacy-forged-create",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"cc" => [],
	"object" => %{
	"type" => "Note",
	"id" => object_id,
	"actor" => bob.ap_id,
	"attributedTo" => bob.ap_id,
	"context" => "https://two.com/contexts/legacy-forged-create",
	"content" => "forged post",
	"published" => "2024-07-25T13:33:31Z",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"cc" => []
	}
	}

	assert {:cancel, :actor_signature_mismatch} =
	ReceiverWorker.perform(%Oban.Job{
	args: %{
	"op" => "incoming_ap_doc",
	"method" => "POST",
	"params" => create,
	"req_headers" => signature_headers_for(alice),
	"request_path" => "/inbox",
	"query_string" => ""
	}
	})

	refute Pleroma.Activity.get_by_ap_id(create["id"])
	refute Pleroma.Object.get_by_ap_id(object_id)
	end

	test "fails closed for the old persisted failed-signature job shape" do
	alice = insert(:user, local: false, ap_id: "https://one.com/users/alice")
	bob = insert(:user, local: false, ap_id: "https://two.com/users/bob")
	object_id = "https://two.com/objects/old-shape-forged-note"

	create = %{
	"type" => "Create",
	"actor" => bob.ap_id,
	"id" => "https://two.com/activities/old-shape-forged-create",
	"context" => "https://two.com/contexts/old-shape-forged-create",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"cc" => [],
	"object" => %{
	"type" => "Note",
	"id" => object_id,
	"actor" => bob.ap_id,
	"attributedTo" => bob.ap_id,
	"context" => "https://two.com/contexts/old-shape-forged-create",
	"content" => "forged post",
	"published" => "2024-07-25T13:33:31Z",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"cc" => []
	}
	}

	assert {:cancel, :missing_signature_retry_metadata} =
	ReceiverWorker.perform(%Oban.Job{
	args: %{
	"op" => "incoming_ap_doc",
	"params" => create,
	"req_headers" => signature_headers_for(alice),
	"timeout" => 20_000
	}
	})

	refute Pleroma.Activity.get_by_ap_id(create["id"])
	refute Pleroma.Object.get_by_ap_id(object_id)
	end

	test "fails closed for legacy retry jobs missing one metadata field" do
	alice = insert(:user, local: false, ap_id: "https://one.com/users/alice")
	params = insert(:note_activity).data

	assert {:cancel, :missing_signature_retry_metadata} =
	ReceiverWorker.perform(%Oban.Job{
	args: %{
	"op" => "incoming_ap_doc",
	"method" => "POST",
	"params" => params,
	"req_headers" => signature_headers_for(alice),
	"request_path" => "/inbox"
	}
	})
	end

	test "fails closed for malformed legacy metadata jobs without params" do
	assert {:cancel, :missing_signature_retry_metadata} =
	ReceiverWorker.perform(%Oban.Job{
	args: %{
	"op" => "incoming_ap_doc",
	"req_headers" => [],
	"timeout" => 20_000
	}
	})
	end

	describe "Server reachability:" do
	setup do
	user = insert(:user)
	remote_user = insert(:user, local: false, ap_id: "https://example.com/users/remote")
	{:ok, _, _} = Pleroma.User.follow(user, remote_user)
	{:ok, activity} = CommonAPI.post(remote_user, %{status: "Test post"})

	%{
	user: user,
	remote_user: remote_user,
	activity: activity
	}
	end

	test "schedules ReachabilityWorker if host is unreachable", %{activity: activity} do
	with_mocks [
	{Pleroma.Web.ActivityPub.Transmogrifier, [],
	[handle_incoming: fn _ -> {:ok, activity} end]},
	{Pleroma.Instances, [], [reachable?: fn _ -> false end]},
	{Pleroma.Web.Federator, [], [perform: fn :incoming_ap_doc, _params -> {:ok, nil} end]}
	] do
	job = %Oban.Job{
	args: %{
	"op" => "incoming_ap_doc",
	"params" => activity.data
	}
	}

	Pleroma.Workers.ReceiverWorker.perform(job)

	assert_enqueued(
	worker: Pleroma.Workers.ReachabilityWorker,
	args: %{"domain" => "example.com"}
	)
	end
	end

	test "does not schedule ReachabilityWorker if host is reachable", %{activity: activity} do
	with_mocks [
	{Pleroma.Web.ActivityPub.Transmogrifier, [],
	[handle_incoming: fn _ -> {:ok, activity} end]},
	{Pleroma.Instances, [], [reachable?: fn _ -> true end]},
	{Pleroma.Web.Federator, [], [perform: fn :incoming_ap_doc, _params -> {:ok, nil} end]}
	] do
	job = %Oban.Job{
	args: %{
	"op" => "incoming_ap_doc",
	"params" => activity.data
	}
	}

	Pleroma.Workers.ReceiverWorker.perform(job)

	refute_enqueued(worker: Pleroma.Workers.ReachabilityWorker)
	end
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sat, Oct 31, 5:21 AM (18 h, 49 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1801077
Default Alt Text: (30 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions