git.squeep.com Git - akkoma/blob - lib/pleroma/object/fetcher.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.Object.Fetcher do
   6   alias Pleroma.HTTP
   7   alias Pleroma.Object
   8   alias Pleroma.Object.Containment
   9   alias Pleroma.Repo
  10   alias Pleroma.Signature
  11   alias Pleroma.Web.ActivityPub.InternalFetchActor
  12   alias Pleroma.Web.ActivityPub.ObjectValidator
  13   alias Pleroma.Web.ActivityPub.Transmogrifier
  14   alias Pleroma.Web.Federator
  15
  16   require Logger
  17   require Pleroma.Constants
  18
  19   defp touch_changeset(changeset) do
  20     updated_at =
  21       NaiveDateTime.utc_now()
  22       |> NaiveDateTime.truncate(:second)
  23
  24     Ecto.Changeset.put_change(changeset, :updated_at, updated_at)
  25   end
  26
  27   defp maybe_reinject_internal_fields(%{data: %{} = old_data}, new_data) do
  28     internal_fields = Map.take(old_data, Pleroma.Constants.object_internal_fields())
  29
  30     Map.merge(new_data, internal_fields)
  31   end
  32
  33   defp maybe_reinject_internal_fields(_, new_data), do: new_data
  34
  35   @spec reinject_object(struct(), map()) :: {:ok, Object.t()} | {:error, any()}
  36   defp reinject_object(%Object{data: %{"type" => "Question"}} = object, new_data) do
  37     Logger.debug("Reinjecting object #{new_data["id"]}")
  38
  39     with new_data <- Transmogrifier.fix_object(new_data),
  40          data <- maybe_reinject_internal_fields(object, new_data),
  41          {:ok, data, _} <- ObjectValidator.validate(data, %{}),
  42          changeset <- Object.change(object, %{data: data}),
  43          changeset <- touch_changeset(changeset),
  44          {:ok, object} <- Repo.insert_or_update(changeset),
  45          {:ok, object} <- Object.set_cache(object) do
  46       {:ok, object}
  47     else
  48       e ->
  49         Logger.error("Error while processing object: #{inspect(e)}")
  50         {:error, e}
  51     end
  52   end
  53
  54   defp reinject_object(%Object{} = object, new_data) do
  55     Logger.debug("Reinjecting object #{new_data["id"]}")
  56
  57     with new_data <- Transmogrifier.fix_object(new_data),
  58          data <- maybe_reinject_internal_fields(object, new_data),
  59          changeset <- Object.change(object, %{data: data}),
  60          changeset <- touch_changeset(changeset),
  61          {:ok, object} <- Repo.insert_or_update(changeset),
  62          {:ok, object} <- Object.set_cache(object) do
  63       {:ok, object}
  64     else
  65       e ->
  66         Logger.error("Error while processing object: #{inspect(e)}")
  67         {:error, e}
  68     end
  69   end
  70
  71   def refetch_object(%Object{data: %{"id" => id}} = object) do
  72     with {:local, false} <- {:local, Object.local?(object)},
  73          {:ok, new_data} <- fetch_and_contain_remote_object_from_id(id),
  74          {:ok, object} <- reinject_object(object, new_data) do
  75       {:ok, object}
  76     else
  77       {:local, true} -> {:ok, object}
  78       e -> {:error, e}
  79     end
  80   end
  81
  82   # Note: will create a Create activity, which we need internally at the moment.
  83   def fetch_object_from_id(id, options \\ []) do
  84     with {_, nil} <- {:fetch_object, Object.get_cached_by_ap_id(id)},
  85          {_, true} <- {:allowed_depth, Federator.allowed_thread_distance?(options[:depth])},
  86          {_, {:ok, data}} <- {:fetch, fetch_and_contain_remote_object_from_id(id)},
  87          {_, nil} <- {:normalize, Object.normalize(data, false)},
  88          params <- prepare_activity_params(data),
  89          {_, :ok} <- {:containment, Containment.contain_origin(id, params)},
  90          {_, {:ok, activity}} <-
  91            {:transmogrifier, Transmogrifier.handle_incoming(params, options)},
  92          {_, _data, %Object{} = object} <-
  93            {:object, data, Object.normalize(activity, false)} do
  94       {:ok, object}
  95     else
  96       {:allowed_depth, false} ->
  97         {:error, "Max thread distance exceeded."}
  98
  99       {:containment, _} ->
 100         {:error, "Object containment failed."}
 101
 102       {:transmogrifier, {:error, {:reject, nil}}} ->
 103         {:reject, nil}
 104
 105       {:transmogrifier, _} = e ->
 106         {:error, e}
 107
 108       {:object, data, nil} ->
 109         reinject_object(%Object{}, data)
 110
 111       {:normalize, object = %Object{}} ->
 112         {:ok, object}
 113
 114       {:fetch_object, %Object{} = object} ->
 115         {:ok, object}
 116
 117       {:fetch, {:error, error}} ->
 118         {:error, error}
 119
 120       e ->
 121         e
 122     end
 123   end
 124
 125   defp prepare_activity_params(data) do
 126     %{
 127       "type" => "Create",
 128       "to" => data["to"],
 129       "cc" => data["cc"],
 130       # Should we seriously keep this attributedTo thing?
 131       "actor" => data["actor"] || data["attributedTo"],
 132       "object" => data
 133     }
 134   end
 135
 136   def fetch_object_from_id!(id, options \\ []) do
 137     with {:ok, object} <- fetch_object_from_id(id, options) do
 138       object
 139     else
 140       {:error, %Tesla.Mock.Error{}} ->
 141         nil
 142
 143       {:error, "Object has been deleted"} ->
 144         nil
 145
 146       {:reject, reason} ->
 147         Logger.info("Rejected #{id} while fetching: #{inspect(reason)}")
 148         nil
 149
 150       e ->
 151         Logger.error("Error while fetching #{id}: #{inspect(e)}")
 152         nil
 153     end
 154   end
 155
 156   defp make_signature(id, date) do
 157     uri = URI.parse(id)
 158
 159     signature =
 160       InternalFetchActor.get_actor()
 161       |> Signature.sign(%{
 162         "(request-target)": "get #{uri.path}",
 163         host: uri.host,
 164         date: date
 165       })
 166
 167     [{"signature", signature}]
 168   end
 169
 170   defp sign_fetch(headers, id, date) do
 171     if Pleroma.Config.get([:activitypub, :sign_object_fetches]) do
 172       headers ++ make_signature(id, date)
 173     else
 174       headers
 175     end
 176   end
 177
 178   defp maybe_date_fetch(headers, date) do
 179     if Pleroma.Config.get([:activitypub, :sign_object_fetches]) do
 180       headers ++ [{"date", date}]
 181     else
 182       headers
 183     end
 184   end
 185
 186   def fetch_and_contain_remote_object_from_id(id) when is_binary(id) do
 187     Logger.debug("Fetching object #{id} via AP")
 188
 189     date = Pleroma.Signature.signed_date()
 190
 191     headers =
 192       [{"accept", "application/activity+json"}]
 193       |> maybe_date_fetch(date)
 194       |> sign_fetch(id, date)
 195
 196     Logger.debug("Fetch headers: #{inspect(headers)}")
 197
 198     with {:scheme, true} <- {:scheme, String.starts_with?(id, "http")},
 199          {:ok, %{body: body, status: code}} when code in 200..299 <- HTTP.get(id, headers),
 200          {:ok, data} <- Jason.decode(body),
 201          :ok <- Containment.contain_origin_from_id(id, data) do
 202       {:ok, data}
 203     else
 204       {:ok, %{status: code}} when code in [404, 410] ->
 205         {:error, "Object has been deleted"}
 206
 207       {:scheme, _} ->
 208         {:error, "Unsupported URI scheme"}
 209
 210       {:error, e} ->
 211         {:error, e}
 212
 213       e ->
 214         {:error, e}
 215     end
 216   end
 217
 218   def fetch_and_contain_remote_object_from_id(%{"id" => id}),
 219     do: fetch_and_contain_remote_object_from_id(id)
 220
 221   def fetch_and_contain_remote_object_from_id(_id), do: {:error, "id must be a string"}
 222 end