git.squeep.com Git - akkoma/blob - lib/pleroma/web/activity_pub/publisher.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.ActivityPub.Publisher do
   6   alias Pleroma.Activity
   7   alias Pleroma.Config
   8   alias Pleroma.Delivery
   9   alias Pleroma.HTTP
  10   alias Pleroma.Instances
  11   alias Pleroma.Object
  12   alias Pleroma.Repo
  13   alias Pleroma.User
  14   alias Pleroma.Web.ActivityPub.Relay
  15   alias Pleroma.Web.ActivityPub.Transmogrifier
  16   alias Pleroma.Web.FedSockets
  17
  18   require Pleroma.Constants
  19
  20   import Pleroma.Web.ActivityPub.Visibility
  21
  22   @behaviour Pleroma.Web.Federator.Publisher
  23
  24   require Logger
  25
  26   @moduledoc """
  27   ActivityPub outgoing federation module.
  28   """
  29
  30   @doc """
  31   Determine if an activity can be represented by running it through Transmogrifier.
  32   """
  33   def is_representable?(%Activity{} = activity) do
  34     with {:ok, _data} <- Transmogrifier.prepare_outgoing(activity.data) do
  35       true
  36     else
  37       _e ->
  38         false
  39     end
  40   end
  41
  42   @doc """
  43   Publish a single message to a peer.  Takes a struct with the following
  44   parameters set:
  45
  46   * `inbox`: the inbox to publish to
  47   * `json`: the JSON message body representing the ActivityPub message
  48   * `actor`: the actor which is signing the message
  49   * `id`: the ActivityStreams URI of the message
  50   """
  51   def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do
  52     Logger.debug("Federating #{id} to #{inbox}")
  53
  54     case FedSockets.get_or_create_fed_socket(inbox) do
  55       {:ok, fedsocket} ->
  56         Logger.debug("publishing via fedsockets - #{inspect(inbox)}")
  57         FedSockets.publish(fedsocket, json)
  58
  59       _ ->
  60         Logger.debug("publishing via http - #{inspect(inbox)}")
  61         http_publish(inbox, actor, json, params)
  62     end
  63   end
  64
  65   def publish_one(%{actor_id: actor_id} = params) do
  66     actor = User.get_cached_by_id(actor_id)
  67
  68     params
  69     |> Map.delete(:actor_id)
  70     |> Map.put(:actor, actor)
  71     |> publish_one()
  72   end
  73
  74   defp http_publish(inbox, actor, json, params) do
  75     uri = %{path: path} = URI.parse(inbox)
  76     digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64())
  77
  78     date = Pleroma.Signature.signed_date()
  79
  80     signature =
  81       Pleroma.Signature.sign(actor, %{
  82         "(request-target)": "post #{path}",
  83         host: signature_host(uri),
  84         "content-length": byte_size(json),
  85         digest: digest,
  86         date: date
  87       })
  88
  89     with {:ok, %{status: code}} when code in 200..299 <-
  90            result =
  91              HTTP.post(
  92                inbox,
  93                json,
  94                [
  95                  {"Content-Type", "application/activity+json"},
  96                  {"Date", date},
  97                  {"signature", signature},
  98                  {"digest", digest}
  99                ]
 100              ) do
 101       if not Map.has_key?(params, :unreachable_since) || params[:unreachable_since] do
 102         Instances.set_reachable(inbox)
 103       end
 104
 105       result
 106     else
 107       {_post_result, response} ->
 108         unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
 109         {:error, response}
 110     end
 111   end
 112
 113   defp signature_host(%URI{port: port, scheme: scheme, host: host}) do
 114     if port == URI.default_port(scheme) do
 115       host
 116     else
 117       "#{host}:#{port}"
 118     end
 119   end
 120
 121   defp should_federate?(inbox, public) do
 122     if public do
 123       true
 124     else
 125       %{host: host} = URI.parse(inbox)
 126
 127       quarantined_instances =
 128         Config.get([:instance, :quarantined_instances], [])
 129         |> Pleroma.Web.ActivityPub.MRF.subdomains_regex()
 130
 131       !Pleroma.Web.ActivityPub.MRF.subdomain_match?(quarantined_instances, host)
 132     end
 133   end
 134
 135   @spec recipients(User.t(), Activity.t()) :: list(User.t()) | []
 136   defp recipients(actor, activity) do
 137     followers =
 138       if actor.follower_address in activity.recipients do
 139         User.get_external_followers(actor)
 140       else
 141         []
 142       end
 143
 144     fetchers =
 145       with %Activity{data: %{"type" => "Delete"}} <- activity,
 146            %Object{id: object_id} <- Object.normalize(activity),
 147            fetchers <- User.get_delivered_users_by_object_id(object_id),
 148            _ <- Delivery.delete_all_by_object_id(object_id) do
 149         fetchers
 150       else
 151         _ ->
 152           []
 153       end
 154
 155     Pleroma.Web.Federator.Publisher.remote_users(actor, activity) ++ followers ++ fetchers
 156   end
 157
 158   defp get_cc_ap_ids(ap_id, recipients) do
 159     host = Map.get(URI.parse(ap_id), :host)
 160
 161     recipients
 162     |> Enum.filter(fn %User{ap_id: ap_id} -> Map.get(URI.parse(ap_id), :host) == host end)
 163     |> Enum.map(& &1.ap_id)
 164   end
 165
 166   defp maybe_use_sharedinbox(%User{shared_inbox: nil, inbox: inbox}), do: inbox
 167   defp maybe_use_sharedinbox(%User{shared_inbox: shared_inbox}), do: shared_inbox
 168
 169   @doc """
 170   Determine a user inbox to use based on heuristics.  These heuristics
 171   are based on an approximation of the ``sharedInbox`` rules in the
 172   [ActivityPub specification][ap-sharedinbox].
 173
 174   Please do not edit this function (or its children) without reading
 175   the spec, as editing the code is likely to introduce some breakage
 176   without some familiarity.
 177
 178      [ap-sharedinbox]: https://www.w3.org/TR/activitypub/#shared-inbox-delivery
 179   """
 180   def determine_inbox(
 181         %Activity{data: activity_data},
 182         %User{inbox: inbox} = user
 183       ) do
 184     to = activity_data["to"] || []
 185     cc = activity_data["cc"] || []
 186     type = activity_data["type"]
 187
 188     cond do
 189       type == "Delete" ->
 190         maybe_use_sharedinbox(user)
 191
 192       Pleroma.Constants.as_public() in to || Pleroma.Constants.as_public() in cc ->
 193         maybe_use_sharedinbox(user)
 194
 195       length(to) + length(cc) > 1 ->
 196         maybe_use_sharedinbox(user)
 197
 198       true ->
 199         inbox
 200     end
 201   end
 202
 203   @doc """
 204   Publishes an activity with BCC to all relevant peers.
 205   """
 206
 207   def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity)
 208       when is_list(bcc) and bcc != [] do
 209     public = is_public?(activity)
 210     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
 211
 212     recipients = recipients(actor, activity)
 213
 214     inboxes =
 215       recipients
 216       |> Enum.filter(&User.ap_enabled?/1)
 217       |> Enum.map(fn actor -> actor.inbox end)
 218       |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
 219       |> Instances.filter_reachable()
 220
 221     Repo.checkout(fn ->
 222       Enum.each(inboxes, fn {inbox, unreachable_since} ->
 223         %User{ap_id: ap_id} = Enum.find(recipients, fn actor -> actor.inbox == inbox end)
 224
 225         # Get all the recipients on the same host and add them to cc. Otherwise, a remote
 226         # instance would only accept a first message for the first recipient and ignore the rest.
 227         cc = get_cc_ap_ids(ap_id, recipients)
 228
 229         json =
 230           data
 231           |> Map.put("cc", cc)
 232           |> Jason.encode!()
 233
 234         Pleroma.Web.Federator.Publisher.enqueue_one(__MODULE__, %{
 235           inbox: inbox,
 236           json: json,
 237           actor_id: actor.id,
 238           id: activity.data["id"],
 239           unreachable_since: unreachable_since
 240         })
 241       end)
 242     end)
 243   end
 244
 245   @doc """
 246   Publishes an activity to all relevant peers.
 247   """
 248   def publish(%User{} = actor, %Activity{} = activity) do
 249     public = is_public?(activity)
 250
 251     if public && Config.get([:instance, :allow_relay]) do
 252       Logger.debug(fn -> "Relaying #{activity.data["id"]} out" end)
 253       Relay.publish(activity)
 254     end
 255
 256     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
 257     json = Jason.encode!(data)
 258
 259     recipients(actor, activity)
 260     |> Enum.filter(fn user -> User.ap_enabled?(user) end)
 261     |> Enum.map(fn %User{} = user ->
 262       determine_inbox(activity, user)
 263     end)
 264     |> Enum.uniq()
 265     |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
 266     |> Instances.filter_reachable()
 267     |> Enum.each(fn {inbox, unreachable_since} ->
 268       Pleroma.Web.Federator.Publisher.enqueue_one(
 269         __MODULE__,
 270         %{
 271           inbox: inbox,
 272           json: json,
 273           actor_id: actor.id,
 274           id: activity.data["id"],
 275           unreachable_since: unreachable_since
 276         }
 277       )
 278     end)
 279   end
 280
 281   def gather_webfinger_links(%User{} = user) do
 282     [
 283       %{"rel" => "self", "type" => "application/activity+json", "href" => user.ap_id},
 284       %{
 285         "rel" => "self",
 286         "type" => "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
 287         "href" => user.ap_id
 288       },
 289       %{
 290         "rel" => "http://ostatus.org/schema/1.0/subscribe",
 291         "template" => "#{Pleroma.Web.base_url()}/ostatus_subscribe?acct={uri}"
 292       }
 293     ]
 294   end
 295
 296   def gather_nodeinfo_protocol_names, do: ["activitypub"]
 297 end