From ef1f9e8d4e4bd2bae0483523c7d31a20c086207e Mon Sep 17 00:00:00 2001 From: William Pitcock Date: Sun, 12 May 2019 03:09:05 +0000 Subject: [PATCH] activitypub: split out outgoing federation into a federation module --- lib/pleroma/web/activity_pub/activity_pub.ex | 87 ------------ lib/pleroma/web/activity_pub/publisher.ex | 131 +++++++++++++++++++ 2 files changed, 131 insertions(+), 87 deletions(-) create mode 100644 lib/pleroma/web/activity_pub/publisher.ex diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex index 8f8c23a9b..11777c220 100644 --- a/lib/pleroma/web/activity_pub/activity_pub.ex +++ b/lib/pleroma/web/activity_pub/activity_pub.ex @@ -5,7 +5,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do alias Pleroma.Activity alias Pleroma.Conversation - alias Pleroma.Instances alias Pleroma.Notification alias Pleroma.Object alias Pleroma.Object.Fetcher @@ -15,7 +14,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do alias Pleroma.User alias Pleroma.Web.ActivityPub.MRF alias Pleroma.Web.ActivityPub.Transmogrifier - alias Pleroma.Web.Federator alias Pleroma.Web.WebFinger import Ecto.Query @@ -24,8 +22,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do require Logger - @httpoison Application.get_env(:pleroma, :httpoison) - # For Announce activities, we filter the recipients based on following status for any actors # that match actual users. See issue #164 for more information about why this is necessary. defp get_recipients(%{"type" => "Announce"} = data) do @@ -961,89 +957,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do end end - def should_federate?(inbox, public) do - if public do - true - else - inbox_info = URI.parse(inbox) - !Enum.member?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host) - end - end - - def publish(actor, activity) do - remote_followers = - if actor.follower_address in activity.recipients do - {:ok, followers} = User.get_followers(actor) - followers |> Enum.filter(&(!&1.local)) - else - [] - end - - public = is_public?(activity) - - {:ok, data} = Transmogrifier.prepare_outgoing(activity.data) - json = Jason.encode!(data) - - (Pleroma.Web.Salmon.remote_users(activity) ++ remote_followers) - |> Enum.filter(fn user -> User.ap_enabled?(user) end) - |> Enum.map(fn %{info: %{source_data: data}} -> - (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"] - end) - |> Enum.uniq() - |> Enum.filter(fn inbox -> should_federate?(inbox, public) end) - |> Instances.filter_reachable() - |> Enum.each(fn {inbox, unreachable_since} -> - Federator.publish_single_ap(%{ - inbox: inbox, - json: json, - actor: actor, - id: activity.data["id"], - unreachable_since: unreachable_since - }) - end) - end - - def publish_one(%{inbox: inbox, json: json, actor: actor, id: id} = params) do - Logger.info("Federating #{id} to #{inbox}") - host = URI.parse(inbox).host - - digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64()) - - date = - NaiveDateTime.utc_now() - |> Timex.format!("{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT") - - signature = - Pleroma.Web.HTTPSignatures.sign(actor, %{ - host: host, - "content-length": byte_size(json), - digest: digest, - date: date - }) - - with {:ok, %{status: code}} when code in 200..299 <- - result = - @httpoison.post( - inbox, - json, - [ - {"Content-Type", "application/activity+json"}, - {"Date", date}, - {"signature", signature}, - {"digest", digest} - ] - ) do - if !Map.has_key?(params, :unreachable_since) || params[:unreachable_since], - do: Instances.set_reachable(inbox) - - result - else - {_post_result, response} -> - unless params[:unreachable_since], do: Instances.set_unreachable(inbox) - {:error, response} - end - end - # filter out broken threads def contain_broken_threads(%Activity{} = activity, %User{} = user) do entire_thread_visible_for_user?(activity, user) diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex new file mode 100644 index 000000000..ee9f0fdd3 --- /dev/null +++ b/lib/pleroma/web/activity_pub/publisher.ex @@ -0,0 +1,131 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Web.ActivityPub.Publisher do + alias Pleroma.Activity + alias Pleroma.Instances + alias Pleroma.User + alias Pleroma.Web.ActivityPub.Transmogrifier + + import Pleroma.Web.ActivityPub.Visibility + + @behaviour Pleroma.Web.Federator.Publisher + + require Logger + + @httpoison Application.get_env(:pleroma, :httpoison) + + @moduledoc """ + ActivityPub outgoing federation module. + """ + + @doc """ + Determine if an activity can be represented by running it through Transmogrifier. + """ + def is_representable?(%Activity{} = activity) do + with %{} = _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.info("Federating #{id} to #{inbox}") + host = URI.parse(inbox).host + + digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64()) + + date = + NaiveDateTime.utc_now() + |> Timex.format!("{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT") + + signature = + Pleroma.Web.HTTPSignatures.sign(actor, %{ + host: host, + "content-length": byte_size(json), + digest: digest, + date: date + }) + + with {:ok, %{status: code}} when code in 200..299 <- + result = + @httpoison.post( + inbox, + json, + [ + {"Content-Type", "application/activity+json"}, + {"Date", date}, + {"signature", signature}, + {"digest", digest} + ] + ) do + if !Map.has_key?(params, :unreachable_since) || params[:unreachable_since], + do: Instances.set_reachable(inbox) + + result + else + {_post_result, response} -> + unless params[:unreachable_since], do: Instances.set_unreachable(inbox) + {:error, response} + end + end + + defp should_federate?(inbox, public) do + if public do + true + else + inbox_info = URI.parse(inbox) + !Enum.member?(Pleroma.Config.get([:instance, :quarantined_instances], []), inbox_info.host) + end + end + + @doc """ + Publishes an activity to all relevant peers. + """ + def publish(%User{} = actor, %Activity{} = activity) do + remote_followers = + if actor.follower_address in activity.recipients do + {:ok, followers} = User.get_followers(actor) + followers |> Enum.filter(&(!&1.local)) + else + [] + end + + public = is_public?(activity) + + {:ok, data} = Transmogrifier.prepare_outgoing(activity.data) + json = Jason.encode!(data) + + (Pleroma.Web.Salmon.remote_users(activity) ++ remote_followers) + |> Enum.filter(fn user -> User.ap_enabled?(user) end) + |> Enum.map(fn %{info: %{source_data: data}} -> + (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"] + end) + |> Enum.uniq() + |> Enum.filter(fn inbox -> should_federate?(inbox, public) end) + |> Instances.filter_reachable() + |> Enum.each(fn {inbox, unreachable_since} -> + Pleroma.Web.Federator.Publisher.enqueue_one( + __MODULE__, + %{ + inbox: inbox, + json: json, + actor: actor, + id: activity.data["id"], + unreachable_since: unreachable_since + } + ) + end) + end +end