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

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.Federator do
   6   alias Pleroma.Activity
   7   alias Pleroma.Object.Containment
   8   alias Pleroma.User
   9   alias Pleroma.Web.ActivityPub.ActivityPub
  10   alias Pleroma.Web.ActivityPub.Transmogrifier
  11   alias Pleroma.Web.ActivityPub.Utils
  12   alias Pleroma.Web.Federator.Publisher
  13   alias Pleroma.Workers.PublisherWorker
  14   alias Pleroma.Workers.ReceiverWorker
  15
  16   require Logger
  17
  18   @doc "Addresses [memory leaks on recursive replies fetching](https://git.pleroma.social/pleroma/pleroma/issues/161)"
  19   # credo:disable-for-previous-line Credo.Check.Readability.MaxLineLength
  20   def allowed_incoming_reply_depth?(depth) do
  21     max_replies_depth = Pleroma.Config.get([:instance, :federation_incoming_replies_max_depth])
  22
  23     if max_replies_depth do
  24       (depth || 1) <= max_replies_depth
  25     else
  26       true
  27     end
  28   end
  29
  30   # Client API
  31
  32   def incoming_ap_doc(params) do
  33     ReceiverWorker.enqueue("incoming_ap_doc", %{"params" => params})
  34   end
  35
  36   def publish(%{id: "pleroma:fakeid"} = activity) do
  37     perform(:publish, activity)
  38   end
  39
  40   def publish(activity) do
  41     PublisherWorker.enqueue("publish", %{"activity_id" => activity.id})
  42   end
  43
  44   # Job Worker Callbacks
  45
  46   @spec perform(atom(), module(), any()) :: {:ok, any()} | {:error, any()}
  47   def perform(:publish_one, module, params) do
  48     apply(module, :publish_one, [params])
  49   end
  50
  51   def perform(:publish, activity) do
  52     Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end)
  53
  54     with %User{} = actor <- User.get_cached_by_ap_id(activity.data["actor"]),
  55          {:ok, actor} <- User.ensure_keys_present(actor) do
  56       Publisher.publish(actor, activity)
  57     end
  58   end
  59
  60   def perform(:incoming_ap_doc, params) do
  61     Logger.info("Handling incoming AP activity")
  62
  63     params = Utils.normalize_params(params)
  64
  65     # NOTE: we use the actor ID to do the containment, this is fine because an
  66     # actor shouldn't be acting on objects outside their own AP server.
  67     with {:ok, _user} <- ap_enabled_actor(params["actor"]),
  68          nil <- Activity.normalize(params["id"]),
  69          :ok <- Containment.contain_origin_from_id(params["actor"], params),
  70          {:ok, activity} <- Transmogrifier.handle_incoming(params) do
  71       {:ok, activity}
  72     else
  73       %Activity{} ->
  74         Logger.info("Already had #{params["id"]}")
  75         :error
  76
  77       _e ->
  78         # Just drop those for now
  79         Logger.info("Unhandled activity")
  80         Logger.info(Jason.encode!(params, pretty: true))
  81         :error
  82     end
  83   end
  84
  85   def ap_enabled_actor(id) do
  86     user = User.get_cached_by_ap_id(id)
  87
  88     if User.ap_enabled?(user) do
  89       {:ok, user}
  90     else
  91       ActivityPub.make_user_from_ap_id(id)
  92     end
  93   end
  94 end