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