No OneTemporary
Actions

Authored By

Unknown

Size

32 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
	index e63b8ff1f..2d2c09f1c 100644
	--- a/lib/pleroma/web/activity_pub/publisher.ex
	+++ b/lib/pleroma/web/activity_pub/publisher.ex
	@@ -1,356 +1,390 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.Publisher do
	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.Delivery
	alias Pleroma.HTTP
	alias Pleroma.Instances
	alias Pleroma.Object
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Relay
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Workers.PublisherWorker

	require Pleroma.Constants

	import Pleroma.Web.ActivityPub.Visibility

	require Logger

	@moduledoc """
	ActivityPub outgoing federation module.
	"""

	@doc """
	Enqueue publishing a single activity.
	"""
	@spec enqueue_one(map(), Keyword.t()) :: {:ok, %Oban.Job{}}
	def enqueue_one(%{} = params, worker_args \\ []) do
	PublisherWorker.enqueue(
	"publish_one",
	%{"params" => params},
	worker_args
	)
	end

	@doc """
	Gathers a set of remote users given an IR envelope.
	"""
	def remote_users(%User{id: user_id}, %{data: %{"to" => to} = data}) do
	cc = Map.get(data, "cc", [])

	bcc =
	data
	\|> Map.get("bcc", [])
	\|> Enum.reduce([], fn ap_id, bcc ->
	case Pleroma.List.get_by_ap_id(ap_id) do
	%Pleroma.List{user_id: ^user_id} = list ->
	{:ok, following} = Pleroma.List.get_following(list)
	bcc ++ Enum.map(following, & &1.ap_id)

	_ ->
	bcc
	end
	end)

	[to, cc, bcc]
	\|> Enum.concat()
	\|> Enum.map(&User.get_cached_by_ap_id/1)
	\|> Enum.filter(fn user -> user && !user.local end)
	end

	@doc """
	Determine if an activity can be represented by running it through Transmogrifier.
	"""
	def representable?(%Activity{} = activity) do
	with {:ok, _data} <- Transmogrifier.prepare_outgoing(activity.data) do
	true
	else
	_e ->
	false
	end
	end

	@doc """
	- Publish a single message to a peer. Takes a struct with the following
	- parameters set:
	-
	+ Prepare an activity for publishing from an Oban job
	* `inbox`: the inbox to publish to
	* `activity_id`: the internal activity id
	* `cc`: the cc recipients relevant to this inbox (optional)
	"""
	- def publish_one(%{inbox: inbox, activity_id: activity_id} = params) do
	+ def prepare_one(%{inbox: inbox, activity_id: activity_id} = params) do
	activity = Activity.get_by_id_with_user_actor(activity_id)
	actor = activity.user_actor

	ap_id = activity.data["id"]
	Logger.debug("Federating #{ap_id} to #{inbox}")
	uri = %{path: path} = URI.parse(inbox)

	{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)

	cc = Map.get(params, :cc)

	json =
	data
	\|> Map.put("cc", cc)
	\|> Jason.encode!()

	digest = "SHA-256=" <> (:crypto.hash(:sha256, json) \|> Base.encode64())

	date = Pleroma.Signature.signed_date()

	signature =
	Pleroma.Signature.sign(actor, %{
	"(request-target)": "post #{path}",
	host: signature_host(uri),
	"content-length": byte_size(json),
	digest: digest,
	date: date
	})

	+ %{
	+ activity_id: activity_id,
	+ json: json,
	+ date: date,
	+ signature: signature,
	+ digest: digest,
	+ inbox: inbox,
	+ unreachable_since: params[:unreachable_since]
	+ }
	+ end
	+
	+ @doc """
	+ Publish a single message to a peer. Takes a struct with the following
	+ parameters set:
	+ * `activity_id`: the activity id
	+ * `json`: the json payload
	+ * `date`: the signed date from Pleroma.Signature.signed_date()
	+ * `signature`: the signature from Pleroma.Signature.sign/2
	+ * `digest`: base64 encoded the hash of the json payload prefixed with "SHA-256="
	+ * `inbox`: the inbox URI of this delivery
	+ * `unreachable_since`: timestamp the instance was marked unreachable
	+
	+ """
	+ def publish_one(%{
	+ activity_id: activity_id,
	+ json: json,
	+ date: date,
	+ signature: signature,
	+ digest: digest,
	+ inbox: inbox,
	+ unreachable_since: unreachable_since
	+ }) do
	with {:ok, %{status: code}} = result when code in 200..299 <-
	HTTP.post(
	inbox,
	json,
	[
	{"Content-Type", "application/activity+json"},
	{"Date", date},
	{"signature", signature},
	{"digest", digest}
	]
	) do
	- if not Map.has_key?(params, :unreachable_since) \|\| params[:unreachable_since] do
	- Instances.set_reachable(inbox)
	- end
	+ maybe_set_reachable(unreachable_since, inbox)

	result
	else
	{_post_result, %{status: code} = response} = e ->
	- unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
	+ maybe_set_unreachable(unreachable_since, inbox)
	Logger.metadata(activity: activity_id, inbox: inbox, status: code)
	Logger.error("Publisher failed to inbox #{inbox} with status #{code}")

	case response do
	%{status: 400} -> {:cancel, :bad_request}
	%{status: 403} -> {:cancel, :forbidden}
	%{status: 404} -> {:cancel, :not_found}
	%{status: 410} -> {:cancel, :not_found}
	_ -> {:error, e}
	end

	{:error, {:already_started, _}} ->
	Logger.debug("Publisher snoozing worker job due worker :already_started race condition")
	connection_pool_snooze()

	{:error, :pool_full} ->
	Logger.debug("Publisher snoozing worker job due to full connection pool")
	connection_pool_snooze()

	e ->
	- unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
	+ maybe_set_unreachable(unreachable_since, inbox)
	Logger.metadata(activity: activity_id, inbox: inbox)
	Logger.error("Publisher failed to inbox #{inbox} #{inspect(e)}")
	{:error, e}
	end
	end

	defp connection_pool_snooze, do: {:snooze, 3}

	+ defp maybe_set_reachable(%NaiveDateTime{}, inbox), do: Instances.set_reachable(inbox)
	+ defp maybe_set_reachable(_, _), do: :ok
	+
	+ defp maybe_set_unreachable(nil, inbox), do: Instances.set_unreachable(inbox)
	+ defp maybe_set_unreachable(%NaiveDateTime{}, _), do: :ok
	+
	defp signature_host(%URI{port: port, scheme: scheme, host: host}) do
	if port == URI.default_port(scheme) do
	host
	else
	"#{host}:#{port}"
	end
	end

	def should_federate?(nil, _), do: false
	def should_federate?(_, true), do: true

	def should_federate?(inbox, _) do
	%{host: host} = URI.parse(inbox)

	quarantined_instances =
	Config.get([:instance, :quarantined_instances], [])
	\|> Pleroma.Web.ActivityPub.MRF.instance_list_from_tuples()
	\|> Pleroma.Web.ActivityPub.MRF.subdomains_regex()

	!Pleroma.Web.ActivityPub.MRF.subdomain_match?(quarantined_instances, host)
	end

	@spec recipients(User.t(), Activity.t()) :: [[User.t()]]
	defp recipients(actor, activity) do
	followers =
	if actor.follower_address in activity.recipients do
	User.get_external_followers(actor)
	else
	[]
	end

	fetchers =
	with %Activity{data: %{"type" => "Delete"}} <- activity,
	%Object{id: object_id} <- Object.normalize(activity, fetch: false),
	fetchers <- User.get_delivered_users_by_object_id(object_id),
	_ <- Delivery.delete_all_by_object_id(object_id) do
	fetchers
	else
	_ ->
	[]
	end

	mentioned = remote_users(actor, activity)
	non_mentioned = (followers ++ fetchers) -- mentioned

	[mentioned, non_mentioned]
	end

	defp get_cc_ap_ids(ap_id, recipients) do
	host = Map.get(URI.parse(ap_id), :host)

	recipients
	\|> Enum.filter(fn %User{ap_id: ap_id} -> Map.get(URI.parse(ap_id), :host) == host end)
	\|> Enum.map(& &1.ap_id)
	end

	defp maybe_use_sharedinbox(%User{shared_inbox: nil, inbox: inbox}), do: inbox
	defp maybe_use_sharedinbox(%User{shared_inbox: shared_inbox}), do: shared_inbox

	@doc """
	Determine a user inbox to use based on heuristics. These heuristics
	are based on an approximation of the ``sharedInbox`` rules in the
	[ActivityPub specification][ap-sharedinbox].

	Please do not edit this function (or its children) without reading
	the spec, as editing the code is likely to introduce some breakage
	without some familiarity.

	[ap-sharedinbox]: https://www.w3.org/TR/activitypub/#shared-inbox-delivery
	"""
	def determine_inbox(
	%Activity{data: activity_data},
	%User{inbox: inbox} = user
	) do
	to = activity_data["to"] \|\| []
	cc = activity_data["cc"] \|\| []
	type = activity_data["type"]

	cond do
	type == "Delete" ->
	maybe_use_sharedinbox(user)

	Pleroma.Constants.as_public() in to \|\| Pleroma.Constants.as_public() in cc ->
	maybe_use_sharedinbox(user)

	length(to) + length(cc) > 1 ->
	maybe_use_sharedinbox(user)

	true ->
	inbox
	end
	end

	@doc """
	Publishes an activity with BCC to all relevant peers.
	"""

	def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity)
	when is_list(bcc) and bcc != [] do
	public = public?(activity)

	[priority_recipients, recipients] = recipients(actor, activity)

	inboxes =
	[priority_recipients, recipients]
	\|> Enum.map(fn recipients ->
	recipients
	\|> Enum.map(fn %User{} = user ->
	determine_inbox(activity, user)
	end)
	\|> Enum.uniq()
	\|> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
	\|> Instances.filter_reachable()
	end)

	Repo.checkout(fn ->
	Enum.each(inboxes, fn inboxes ->
	Enum.each(inboxes, fn {inbox, unreachable_since} ->
	%User{ap_id: ap_id} = Enum.find(recipients, fn actor -> actor.inbox == inbox end)

	# Get all the recipients on the same host and add them to cc. Otherwise, a remote
	# instance would only accept a first message for the first recipient and ignore the rest.
	cc = get_cc_ap_ids(ap_id, recipients)

	__MODULE__.enqueue_one(%{
	inbox: inbox,
	cc: cc,
	activity_id: activity.id,
	unreachable_since: unreachable_since
	})
	end)
	end)
	end)
	end

	# Publishes an activity to all relevant peers.
	def publish(%User{} = actor, %Activity{} = activity) do
	public = public?(activity)

	if public && Config.get([:instance, :allow_relay]) do
	Logger.debug(fn -> "Relaying #{activity.data["id"]} out" end)
	Relay.publish(activity)
	end

	[priority_inboxes, inboxes] =
	recipients(actor, activity)
	\|> Enum.map(fn recipients ->
	recipients
	\|> Enum.map(fn %User{} = user ->
	determine_inbox(activity, user)
	end)
	\|> Enum.uniq()
	\|> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
	end)

	inboxes = inboxes -- priority_inboxes

	[{priority_inboxes, 0}, {inboxes, 1}]
	\|> Enum.each(fn {inboxes, priority} ->
	inboxes
	\|> Instances.filter_reachable()
	\|> Enum.each(fn {inbox, unreachable_since} ->
	__MODULE__.enqueue_one(
	%{
	inbox: inbox,
	activity_id: activity.id,
	unreachable_since: unreachable_since
	},
	priority: priority
	)
	end)
	end)

	:ok
	end

	def gather_webfinger_links(%User{} = user) do
	[
	%{"rel" => "self", "type" => "application/activity+json", "href" => user.ap_id},
	%{
	"rel" => "self",
	"type" => "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
	"href" => user.ap_id
	},
	%{
	"rel" => "http://ostatus.org/schema/1.0/subscribe",
	"template" => "#{Pleroma.Web.Endpoint.url()}/ostatus_subscribe?acct={uri}"
	}
	]
	end

	def gather_nodeinfo_protocol_names, do: ["activitypub"]
	end
	diff --git a/lib/pleroma/web/federator.ex b/lib/pleroma/web/federator.ex
	index 3d3101d61..242cf4bfd 100644
	--- a/lib/pleroma/web/federator.ex
	+++ b/lib/pleroma/web/federator.ex
	@@ -1,123 +1,126 @@
	# 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

	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_ap_doc(%{params: _params, req_headers: _req_headers} = args) do
	job_args = Enum.into(args, %{}, fn {k, v} -> {Atom.to_string(k), v} end)

	ReceiverWorker.enqueue(
	"incoming_ap_doc",
	Map.put(job_args, "timeout", :timer.seconds(20)),
	priority: 2
	)
	end

	def incoming_ap_doc(%{"type" => "Delete"} = params) do
	ReceiverWorker.enqueue("incoming_ap_doc", %{"params" => params}, priority: 3, queue: :slow)
	end

	def incoming_ap_doc(params) do
	ReceiverWorker.enqueue("incoming_ap_doc", %{"params" => params})
	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.enqueue("publish", %{"activity_id" => activity.id},
	priority: publish_priority(type)
	)
	end

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

	# Job Worker Callbacks

	@spec perform(atom(), any()) :: {:ok, any()} \| {:error, any()}
	- def perform(:publish_one, params), do: Publisher.publish_one(params)
	+ 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()

	# 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)},
	nil <- Activity.normalize(params["id"]),
	{_, :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{} ->
	Logger.debug("Already had #{params["id"]}")
	{:error, :already_present}

	{:actor, e} ->
	Logger.debug("Unhandled actor #{actor}, #{inspect(e)}")
	{:error, e}

	{:error, {:validate_object, _}} = e ->
	Logger.error("Incoming AP doc validation error: #{inspect(e)}")
	Logger.debug(Jason.encode!(params, pretty: true))
	e

	e ->
	# Just drop those for now
	Logger.debug(fn -> "Unhandled activity\n" <> Jason.encode!(params, pretty: true) end)
	{:error, e}
	end
	end
	end
	diff --git a/test/pleroma/web/activity_pub/publisher_test.exs b/test/pleroma/web/activity_pub/publisher_test.exs
	index 1992fb228..370b4b4d2 100644
	--- a/test/pleroma/web/activity_pub/publisher_test.exs
	+++ b/test/pleroma/web/activity_pub/publisher_test.exs
	@@ -1,486 +1,495 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.PublisherTest do
	use Oban.Testing, repo: Pleroma.Repo
	use Pleroma.Web.ConnCase

	import ExUnit.CaptureLog
	import Pleroma.Factory
	import Tesla.Mock
	import Mock

	alias Pleroma.Activity
	alias Pleroma.Instances
	alias Pleroma.Object
	alias Pleroma.Web.ActivityPub.Publisher
	alias Pleroma.Web.CommonAPI

	@as_public "https://www.w3.org/ns/activitystreams#Public"

	setup do
	mock(fn env -> apply(HttpRequestMock, :request, [env]) end)
	:ok
	end

	setup_all do: clear_config([:instance, :federating], true)

	describe "should_federate?/1" do
	test "it returns false when the inbox is nil" do
	refute Publisher.should_federate?(nil, false)
	refute Publisher.should_federate?(nil, true)
	end

	test "it returns true when public is true" do
	assert Publisher.should_federate?(false, true)
	end
	end

	describe "gather_webfinger_links/1" do
	test "it returns links" do
	user = insert(:user)

	expected_links = [
	%{"href" => user.ap_id, "rel" => "self", "type" => "application/activity+json"},
	%{
	"href" => user.ap_id,
	"rel" => "self",
	"type" => "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\""
	},
	%{
	"rel" => "http://ostatus.org/schema/1.0/subscribe",
	"template" => "#{Pleroma.Web.Endpoint.url()}/ostatus_subscribe?acct={uri}"
	}
	]

	assert expected_links == Publisher.gather_webfinger_links(user)
	end
	end

	describe "determine_inbox/2" do
	test "it returns sharedInbox for messages involving as:Public in to" do
	user = insert(:user, %{shared_inbox: "http://example.com/inbox"})

	activity = %Activity{
	data: %{"to" => [@as_public], "cc" => [user.follower_address]}
	}

	assert Publisher.determine_inbox(activity, user) == "http://example.com/inbox"
	end

	test "it returns sharedInbox for messages involving as:Public in cc" do
	user = insert(:user, %{shared_inbox: "http://example.com/inbox"})

	activity = %Activity{
	data: %{"cc" => [@as_public], "to" => [user.follower_address]}
	}

	assert Publisher.determine_inbox(activity, user) == "http://example.com/inbox"
	end

	test "it returns sharedInbox for messages involving multiple recipients in to" do
	user = insert(:user, %{shared_inbox: "http://example.com/inbox"})
	user_two = insert(:user)
	user_three = insert(:user)

	activity = %Activity{
	data: %{"cc" => [], "to" => [user.ap_id, user_two.ap_id, user_three.ap_id]}
	}

	assert Publisher.determine_inbox(activity, user) == "http://example.com/inbox"
	end

	test "it returns sharedInbox for messages involving multiple recipients in cc" do
	user = insert(:user, %{shared_inbox: "http://example.com/inbox"})
	user_two = insert(:user)
	user_three = insert(:user)

	activity = %Activity{
	data: %{"to" => [], "cc" => [user.ap_id, user_two.ap_id, user_three.ap_id]}
	}

	assert Publisher.determine_inbox(activity, user) == "http://example.com/inbox"
	end

	test "it returns sharedInbox for messages involving multiple recipients in total" do
	user =
	insert(:user, %{
	shared_inbox: "http://example.com/inbox",
	inbox: "http://example.com/personal-inbox"
	})

	user_two = insert(:user)

	activity = %Activity{
	data: %{"to" => [user_two.ap_id], "cc" => [user.ap_id]}
	}

	assert Publisher.determine_inbox(activity, user) == "http://example.com/inbox"
	end

	test "it returns inbox for messages involving single recipients in total" do
	user =
	insert(:user, %{
	shared_inbox: "http://example.com/inbox",
	inbox: "http://example.com/personal-inbox"
	})

	activity = %Activity{
	data: %{"to" => [user.ap_id], "cc" => []}
	}

	assert Publisher.determine_inbox(activity, user) == "http://example.com/personal-inbox"
	end
	end

	describe "publish_one/1" do
	test "publish to url with with different ports" do
	inbox80 = "http://42.site/users/nick1/inbox"
	inbox42 = "http://42.site:42/users/nick1/inbox"
	activity = insert(:note_activity)

	mock(fn
	%{method: :post, url: "http://42.site:42/users/nick1/inbox"} ->
	{:ok, %Tesla.Env{status: 200, body: "port 42"}}

	%{method: :post, url: "http://42.site/users/nick1/inbox"} ->
	{:ok, %Tesla.Env{status: 200, body: "port 80"}}
	end)

	_actor = insert(:user)

	assert {:ok, %{body: "port 42"}} =
	- Publisher.publish_one(%{
	+ Publisher.prepare_one(%{
	inbox: inbox42,
	activity_id: activity.id,
	unreachable_since: true
	})
	+ \|> Publisher.publish_one()

	assert {:ok, %{body: "port 80"}} =
	- Publisher.publish_one(%{
	+ Publisher.prepare_one(%{
	inbox: inbox80,
	activity_id: activity.id,
	unreachable_since: true
	})
	+ \|> Publisher.publish_one()
	end

	test_with_mock "calls `Instances.set_reachable` on successful federation if `unreachable_since` is not specified",
	Instances,
	[:passthrough],
	[] do
	_actor = insert(:user)
	inbox = "http://200.site/users/nick1/inbox"
	activity = insert(:note_activity)

	assert {:ok, _} =
	- Publisher.publish_one(%{inbox: inbox, activity_id: activity.id})
	+ Publisher.prepare_one(%{inbox: inbox, activity_id: activity.id})
	+ \|> Publisher.publish_one()

	assert called(Instances.set_reachable(inbox))
	end

	test_with_mock "calls `Instances.set_reachable` on successful federation if `unreachable_since` is set",
	Instances,
	[:passthrough],
	[] do
	_actor = insert(:user)
	inbox = "http://200.site/users/nick1/inbox"
	activity = insert(:note_activity)

	assert {:ok, _} =
	- Publisher.publish_one(%{
	+ Publisher.prepare_one(%{
	inbox: inbox,
	activity_id: activity.id,
	unreachable_since: NaiveDateTime.utc_now()
	})
	+ \|> Publisher.publish_one()

	assert called(Instances.set_reachable(inbox))
	end

	test_with_mock "does NOT call `Instances.set_reachable` on successful federation if `unreachable_since` is nil",
	Instances,
	[:passthrough],
	[] do
	_actor = insert(:user)
	inbox = "http://200.site/users/nick1/inbox"
	activity = insert(:note_activity)

	assert {:ok, _} =
	- Publisher.publish_one(%{
	+ Publisher.prepare_one(%{
	inbox: inbox,
	activity_id: activity.id,
	unreachable_since: nil
	})
	+ \|> Publisher.publish_one()

	refute called(Instances.set_reachable(inbox))
	end

	test_with_mock "calls `Instances.set_unreachable` on target inbox on non-2xx HTTP response code",
	Instances,
	[:passthrough],
	[] do
	_actor = insert(:user)
	inbox = "http://404.site/users/nick1/inbox"
	activity = insert(:note_activity)

	assert {:cancel, _} =
	- Publisher.publish_one(%{inbox: inbox, activity_id: activity.id})
	+ Publisher.prepare_one(%{inbox: inbox, activity_id: activity.id})
	+ \|> Publisher.publish_one()

	assert called(Instances.set_unreachable(inbox))
	end

	test_with_mock "it calls `Instances.set_unreachable` on target inbox on request error of any kind",
	Instances,
	[:passthrough],
	[] do
	_actor = insert(:user)
	inbox = "http://connrefused.site/users/nick1/inbox"
	activity = insert(:note_activity)

	assert capture_log(fn ->
	assert {:error, _} =
	- Publisher.publish_one(%{
	+ Publisher.prepare_one(%{
	inbox: inbox,
	activity_id: activity.id
	})
	+ \|> Publisher.publish_one()
	end) =~ "connrefused"

	assert called(Instances.set_unreachable(inbox))
	end

	test_with_mock "does NOT call `Instances.set_unreachable` if target is reachable",
	Instances,
	[:passthrough],
	[] do
	_actor = insert(:user)
	inbox = "http://200.site/users/nick1/inbox"
	activity = insert(:note_activity)

	assert {:ok, _} =
	- Publisher.publish_one(%{inbox: inbox, activity_id: activity.id})
	+ Publisher.prepare_one(%{inbox: inbox, activity_id: activity.id})
	+ \|> Publisher.publish_one()

	refute called(Instances.set_unreachable(inbox))
	end

	test_with_mock "does NOT call `Instances.set_unreachable` if target instance has non-nil `unreachable_since`",
	Instances,
	[:passthrough],
	[] do
	_actor = insert(:user)
	inbox = "http://connrefused.site/users/nick1/inbox"
	activity = insert(:note_activity)

	assert capture_log(fn ->
	assert {:error, _} =
	- Publisher.publish_one(%{
	+ Publisher.prepare_one(%{
	inbox: inbox,
	activity_id: activity.id,
	unreachable_since: NaiveDateTime.utc_now()
	})
	+ \|> Publisher.publish_one()
	end) =~ "connrefused"

	refute called(Instances.set_unreachable(inbox))
	end
	end

	describe "publish/2" do
	test_with_mock "doesn't publish a non-public activity to quarantined instances.",
	Pleroma.Web.ActivityPub.Publisher,
	[:passthrough],
	[] do
	Config.put([:instance, :quarantined_instances], [{"domain.com", "some reason"}])

	follower =
	insert(:user, %{
	local: false,
	inbox: "https://domain.com/users/nick1/inbox"
	})

	actor = insert(:user, follower_address: follower.ap_id)

	{:ok, follower, actor} = Pleroma.User.follow(follower, actor)
	actor = refresh_record(actor)

	note_activity =
	insert(:followers_only_note_activity,
	user: actor,
	recipients: [follower.ap_id]
	)

	res = Publisher.publish(actor, note_activity)

	assert res == :ok

	refute_enqueued(
	worker: "Pleroma.Workers.PublisherWorker",
	args: %{
	"params" => %{
	inbox: "https://domain.com/users/nick1/inbox",
	activity_id: note_activity.id
	}
	}
	)
	end

	test_with_mock "Publishes a non-public activity to non-quarantined instances.",
	Pleroma.Web.ActivityPub.Publisher,
	[:passthrough],
	[] do
	Config.put([:instance, :quarantined_instances], [{"somedomain.com", "some reason"}])

	follower =
	insert(:user, %{
	local: false,
	inbox: "https://domain.com/users/nick1/inbox"
	})

	actor = insert(:user, follower_address: follower.ap_id)

	{:ok, follower, actor} = Pleroma.User.follow(follower, actor)
	actor = refresh_record(actor)

	note_activity =
	insert(:followers_only_note_activity,
	user: actor,
	recipients: [follower.ap_id]
	)

	res = Publisher.publish(actor, note_activity)

	assert res == :ok

	assert_enqueued(
	worker: "Pleroma.Workers.PublisherWorker",
	args: %{
	"params" => %{
	inbox: "https://domain.com/users/nick1/inbox",
	activity_id: note_activity.id
	}
	},
	priority: 1
	)
	end

	test_with_mock "Publishes to directly addressed actors with higher priority.",
	Pleroma.Web.ActivityPub.Publisher,
	[:passthrough],
	[] do
	note_activity = insert(:direct_note_activity)

	actor = Pleroma.User.get_by_ap_id(note_activity.data["actor"])

	res = Publisher.publish(actor, note_activity)

	assert res == :ok

	assert called(
	Publisher.enqueue_one(
	%{
	inbox: :_,
	activity_id: note_activity.id
	},
	priority: 0
	)
	)
	end

	test_with_mock "publishes an activity with BCC to all relevant peers.",
	Pleroma.Web.ActivityPub.Publisher,
	[:passthrough],
	[] do
	follower =
	insert(:user, %{
	local: false,
	inbox: "https://domain.com/users/nick1/inbox"
	})

	actor = insert(:user, follower_address: follower.ap_id)
	user = insert(:user)

	{:ok, follower, actor} = Pleroma.User.follow(follower, actor)

	note_activity =
	insert(:note_activity,
	recipients: [follower.ap_id],
	data_attrs: %{"bcc" => [user.ap_id]}
	)

	res = Publisher.publish(actor, note_activity)
	assert res == :ok

	assert_enqueued(
	worker: "Pleroma.Workers.PublisherWorker",
	args: %{
	"params" => %{
	inbox: "https://domain.com/users/nick1/inbox",
	activity_id: note_activity.id
	}
	}
	)
	end

	test_with_mock "publishes a delete activity to peers who signed fetch requests to the create acitvity/object.",
	Pleroma.Web.ActivityPub.Publisher,
	[:passthrough],
	[] do
	fetcher =
	insert(:user,
	local: false,
	inbox: "https://domain.com/users/nick1/inbox"
	)

	another_fetcher =
	insert(:user,
	local: false,
	inbox: "https://domain2.com/users/nick1/inbox"
	)

	actor = insert(:user)

	note_activity = insert(:note_activity, user: actor)
	object = Object.normalize(note_activity, fetch: false)

	activity_path = String.trim_leading(note_activity.data["id"], Pleroma.Web.Endpoint.url())
	object_path = String.trim_leading(object.data["id"], Pleroma.Web.Endpoint.url())

	build_conn()
	\|> put_req_header("accept", "application/activity+json")
	\|> assign(:user, fetcher)
	\|> get(object_path)
	\|> json_response(200)

	build_conn()
	\|> put_req_header("accept", "application/activity+json")
	\|> assign(:user, another_fetcher)
	\|> get(activity_path)
	\|> json_response(200)

	{:ok, delete} = CommonAPI.delete(note_activity.id, actor)

	res = Publisher.publish(actor, delete)
	assert res == :ok

	assert_enqueued(
	worker: "Pleroma.Workers.PublisherWorker",
	args: %{
	"params" => %{
	inbox: "https://domain.com/users/nick1/inbox",
	activity_id: delete.id
	}
	},
	priority: 1
	)

	assert_enqueued(
	worker: "Pleroma.Workers.PublisherWorker",
	args: %{
	"params" => %{
	inbox: "https://domain2.com/users/nick1/inbox",
	activity_id: delete.id
	}
	},
	priority: 1
	)
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Fri, Oct 30, 3:14 PM (19 h, 35 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1800240
Default Alt Text: (32 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions