Page MenuHomePhorge
Files F85825742

No OneTemporary
Actions

Authored By
Unknown
Size
17 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/changelog.d/fix-duplicate-inbox-deliveries.fix b/changelog.d/fix-duplicate-inbox-deliveries.fix
new file mode 100644
index 000000000..e69de29bb
diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
index cc47b3d27..fb7c6f005 100644
--- a/lib/pleroma/web/activity_pub/publisher.ex
+++ b/lib/pleroma/web/activity_pub/publisher.ex
@@ -1,348 +1,354 @@
# 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.t(), 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 is_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:
* `inbox`: the inbox to publish to
* `json`: the JSON message body representing the ActivityPub message
* `actor`: the actor which is signing the message
* `id`: the ActivityStreams URI of the message
"""
def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do
Logger.debug("Federating #{id} to #{inbox}")
uri = %{path: path} = URI.parse(inbox)
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
})
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
result
else
{_post_result, %{status: code} = response} = e ->
unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
Logger.metadata(activity: id, inbox: inbox, status: code)
Logger.error("Publisher failed to inbox #{inbox} with status #{code}")
case response do
%{status: 403} -> {:discard, :forbidden}
%{status: 404} -> {:discard, :not_found}
%{status: 410} -> {:discard, :not_found}
_ -> {:error, e}
end
e ->
unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
Logger.metadata(activity: id, inbox: inbox)
Logger.error("Publisher failed to inbox #{inbox} #{inspect(e)}")
{:error, e}
end
end
def publish_one(%{actor_id: actor_id} = params) do
actor = User.get_cached_by_id(actor_id)
params
|> Map.delete(:actor_id)
|> Map.put(:actor, actor)
|> publish_one()
end
defp signature_host(%URI{port: port, scheme: scheme, host: host}) do
if port == URI.default_port(scheme) do
host
else
"#{host}:#{port}"
end
end
defp should_federate?(inbox, public) do
if public do
true
else
%{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
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 = is_public?(activity)
{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
[priority_recipients, recipients] = recipients(actor, activity)
inboxes =
[priority_recipients, recipients]
|> Enum.map(fn recipients ->
recipients
- |> Enum.map(fn actor -> actor.inbox end)
+ |> 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)
json =
data
|> Map.put("cc", cc)
|> Jason.encode!()
__MODULE__.enqueue_one(%{
inbox: inbox,
json: json,
actor_id: actor.id,
id: activity.data["id"],
unreachable_since: unreachable_since
})
end)
end)
end)
end
# Publishes an activity to all relevant peers.
def publish(%User{} = actor, %Activity{} = activity) do
public = is_public?(activity)
if public && Config.get([:instance, :allow_relay]) do
Logger.debug(fn -> "Relaying #{activity.data["id"]} out" end)
Relay.publish(activity)
end
{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
json = Jason.encode!(data)
[priority_inboxes, inboxes] =
recipients(actor, activity)
|> Enum.map(fn recipients ->
recipients
- |> Enum.map(fn actor -> actor.inbox end)
+ |> 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,
json: json,
actor_id: actor.id,
id: activity.data["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/test/pleroma/web/federator_test.exs b/test/pleroma/web/federator_test.exs
index 6826e6c2f..4a398f239 100644
--- a/test/pleroma/web/federator_test.exs
+++ b/test/pleroma/web/federator_test.exs
@@ -1,173 +1,211 @@
# 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.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
})
dt = NaiveDateTime.utc_now()
Instances.set_unreachable(inbox1, dt)
Instances.set_consistently_unreachable(URI.parse(inbox2).host)
{:ok, _activity} =
CommonAPI.post(user, %{status: "HI @nick1@domain.com, @nick2@domain2.com!"})
expected_dt = NaiveDateTime.to_iso8601(dt)
ObanHelpers.perform(all_enqueued(worker: PublisherWorker))
assert ObanHelpers.member?(
%{
"op" => "publish_one",
"params" => %{"inbox" => inbox1, "unreachable_since" => expected_dt}
},
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 "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

File Metadata

Mime Type
text/x-diff
Expires
Fri, Oct 30, 2:50 AM (1 d, 4 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1799252
Default Alt Text
(17 KB)

Event Timeline