No OneTemporary
Actions

Authored By

Unknown

Size

15 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/workers/receiver_worker.ex b/lib/pleroma/workers/receiver_worker.ex
	index e2c950967..507e099c2 100644
	--- a/lib/pleroma/workers/receiver_worker.ex
	+++ b/lib/pleroma/workers/receiver_worker.ex
	@@ -1,97 +1,111 @@
	# 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.Signature
	alias Pleroma.User
	alias Pleroma.Web.Federator
	+ alias Pleroma.Web.Plugs.MappedSignatureToIdentityPlug

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

	@impl true

	def perform(%Job{
	args: %{
	"op" => "incoming_ap_doc",
	"method" => method,
	"params" => params,
	"req_headers" => req_headers,
	"request_path" => request_path,
	"query_string" => query_string
	}
	}) do
	# Oban's serialization converts our tuple headers to lists.
	# Revert it for the signature validation.
	req_headers = Enum.into(req_headers, [], &List.to_tuple(&1))

	conn_data = %Plug.Conn{
	+ assigns: %{valid_signature: true},
	method: method,
	params: params,
	req_headers: req_headers,
	request_path: request_path,
	query_string: query_string
	}

	with {:ok, %User{}} <- User.get_or_fetch_by_ap_id(conn_data.params["actor"]),
	{:ok, _public_key} <- Signature.refetch_public_key(conn_data),
	{:signature, true} <- {:signature, Signature.validate_signature(conn_data)},
	+ {:same_actor, true} <- {:same_actor, validate_same_actor(conn_data)},
	{: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

	def perform(%Job{args: %{"op" => "incoming_ap_doc", "params" => 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

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

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

	+ defp validate_same_actor(conn_data) do
	+ case MappedSignatureToIdentityPlug.call(conn_data, []) do
	+ %Plug.Conn{assigns: %{valid_signature: true}} ->
	+ true
	+
	+ _ ->
	+ false
	+ end
	+ end
	+
	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}
	# 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}
	# Catchall
	{:error, _} = e -> e
	e -> {:error, e}
	end
	end
	end
	diff --git a/test/pleroma/workers/receiver_worker_test.exs b/test/pleroma/workers/receiver_worker_test.exs
	index 1c886a74e..bc027ad4c 100644
	--- a/test/pleroma/workers/receiver_worker_test.exs
	+++ b/test/pleroma/workers/receiver_worker_test.exs
	@@ -1,349 +1,349 @@
	# 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.Web.CommonAPI
	alias Pleroma.Web.Federator
	alias Pleroma.Workers.ReceiverWorker

	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

	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")

	{:ok, oban_job} =
	Federator.incoming_ap_doc(%{
	method: "POST",
	req_headers: [],
	request_path: "/inbox",
	params: params,
	query_string: ""
	})

	assert {:cancel, {:error, :forbidden}} = ReceiverWorker.perform(oban_job)
	end

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

	{:ok, oban_job} =
	Federator.incoming_ap_doc(%{
	method: "POST",
	req_headers: [],
	request_path: "/inbox",
	params: params,
	query_string: ""
	})

	assert {:cancel, {:error, :not_found}} = ReceiverWorker.perform(oban_job)
	end

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

	{:ok, oban_job} =
	Federator.incoming_ap_doc(%{
	method: "POST",
	req_headers: [],
	request_path: "/inbox",
	params: params,
	query_string: ""
	})

	assert {:cancel, {:error, :not_found}} = ReceiverWorker.perform(oban_job)
	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)

	{:ok, oban_job} =
	Federator.incoming_ap_doc(%{
	method: "POST",
	req_headers: [],
	request_path: "/inbox",
	params: params,
	query_string: ""
	})

	assert {:cancel, {:user_active, false}} = ReceiverWorker.perform(oban_job)
	end
	end

	test "it can validate the signature" do
	Tesla.Mock.mock(fn
	%{url: "https://phpc.social/users/denniskoch"} ->
	%Tesla.Env{
	status: 200,
	body: File.read!("test/fixtures/denniskoch.json"),
	headers: [{"content-type", "application/activity+json"}]
	}

	%{url: "https://phpc.social/users/denniskoch/collections/featured"} ->
	%Tesla.Env{
	status: 200,
	headers: [{"content-type", "application/activity+json"}],
	body:
	File.read!("test/fixtures/users_mock/masto_featured.json")
	\|> String.replace("{{domain}}", "phpc.social")
	\|> String.replace("{{nickname}}", "denniskoch")
	}
	end)

	params =
	File.read!("test/fixtures/receiver_worker_signature_activity.json") \|> Jason.decode!()

	req_headers = [
	["accept-encoding", "gzip"],
	["content-length", "5184"],
	["content-type", "application/activity+json"],
	["date", "Thu, 25 Jul 2024 13:33:31 GMT"],
	["digest", "SHA-256=ouge/6HP2/QryG6F3JNtZ6vzs/hSwMk67xdxe87eH7A="],
	["host", "bikeshed.party"],
	[
	"signature",
	"keyId=\"https://mastodon.social/users/bastianallgeier#main-key\",algorithm=\"rsa-sha256\",headers=\"(request-target) host date digest content-type\",signature=\"ymE3vn5Iw50N6ukSp8oIuXJB5SBjGAGjBasdTDvn+ahZIzq2SIJfmVCsIIzyqIROnhWyQoTbavTclVojEqdaeOx+Ejz2wBnRBmhz5oemJLk4RnnCH0lwMWyzeY98YAvxi9Rq57Gojuv/1lBqyGa+rDzynyJpAMyFk17XIZpjMKuTNMCbjMDy76ILHqArykAIL/v1zxkgwxY/+ELzxqMpNqtZ+kQ29znNMUBB3eVZ/mNAHAz6o33Y9VKxM2jw+08vtuIZOusXyiHbRiaj2g5HtN2WBUw1MzzfRfHF2/yy7rcipobeoyk5RvP5SyHV3WrIeZ3iyoNfmv33y8fxllF0EA==\""
	],
	[
	"user-agent",
	"http.rb/5.2.0 (Mastodon/4.3.0-nightly.2024-07-25; +https://mastodon.social/)"
	]
	]

	{:ok, oban_job} =
	Federator.incoming_ap_doc(%{
	method: "POST",
	req_headers: req_headers,
	request_path: "/inbox",
	params: params,
	query_string: ""
	})

	assert {:ok, %Pleroma.Activity{}} = ReceiverWorker.perform(oban_job)
	end

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

	{:ok, oban_job} =
	Federator.incoming_ap_doc(%{
	method: "POST",
	req_headers: [],
	request_path: "/inbox",
	params: params,
	query_string: ""
	})

	assert {:cancel, :origin_containment_failed} = ReceiverWorker.perform(oban_job)
	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)

	{:ok, oban_job} =
	Federator.incoming_ap_doc(%{
	method: "POST",
	req_headers: [],
	request_path: "/inbox",
	params: params,
	query_string: ""
	})

	assert {:cancel, _} = ReceiverWorker.perform(oban_job)
	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

	test "cancels when signature actor does not match payload actor" do
	- alice = insert(:user, local: false, ap_id: "https://example.com/users/alice")
	+ _alice = insert(:user, local: false, ap_id: "https://example.com/users/alice")
	bob = insert(:user, local: false, ap_id: "https://example.com/users/bob")

	note = insert(:note, user: bob, object_local: false)

	update = %{
	"type" => "Update",
	"actor" => bob.ap_id,
	"id" => "https://example.com/activities/malicious-update",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"cc" => [],
	"object" => note.data
	}

	req_headers = [
	["host", "example.com"],
	["date", "Thu, 25 Jul 2024 13:33:31 GMT"],
	["digest", "SHA-256=fake-digest"],
	["content-type", "application/activity+json"],
	[
	"signature",
	"keyId=\"https://example.com/users/alice#main-key\",algorithm=\"rsa-sha256\",headers=\"(request-target) host date digest content-type\",signature=\"fake-signature\""
	]
	]

	oban_job = %Oban.Job{
	args: %{
	"op" => "incoming_ap_doc",
	"method" => "POST",
	"params" => update,
	"req_headers" => req_headers,
	"request_path" => "/inbox",
	"query_string" => ""
	}
	}

	with_mock Pleroma.Signature, [:passthrough],
	refetch_public_key: fn _conn -> {:ok, :fake_public_key} end,
	validate_signature: fn _conn -> true end do
	- assert {:cancel, :invalid_signature} = ReceiverWorker.perform(oban_job)
	+ assert {:cancel, :actor_signature_mismatch} = ReceiverWorker.perform(oban_job)
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Mon, Nov 2, 10:02 AM (3 h, 13 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1804364
Default Alt Text: (15 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions