git.squeep.com Git - akkoma/blob - lib/pleroma/object/fetcher.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2021 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 data <- maybe_reinject_internal_fields(object, new_data),
  40          {:ok, data, _} <- ObjectValidator.validate(data, %{}),
  41          changeset <- Object.change(object, %{data: data}),
  42          changeset <- touch_changeset(changeset),
  43          {:ok, object} <- Repo.insert_or_update(changeset),
  44          {:ok, object} <- Object.set_cache(object) do
  45       {:ok, object}
  46     else
  47       e ->
  48         Logger.error("Error while processing object: #{inspect(e)}")
  49         {:error, e}
  50     end
  51   end
  52
  53   defp reinject_object(%Object{} = object, new_data) do
  54     Logger.debug("Reinjecting object #{new_data["id"]}")
  55
  56     with new_data <- Transmogrifier.fix_object(new_data),
  57          data <- maybe_reinject_internal_fields(object, new_data),
  58          changeset <- Object.change(object, %{data: data}),
  59          changeset <- touch_changeset(changeset),
  60          {:ok, object} <- Repo.insert_or_update(changeset),
  61          {:ok, object} <- Object.set_cache(object) do
  62       {:ok, object}
  63     else
  64       e ->
  65         Logger.error("Error while processing object: #{inspect(e)}")
  66         {:error, e}
  67     end
  68   end
  69
  70   def refetch_object(%Object{data: %{"id" => id}} = object) do
  71     with {:local, false} <- {:local, Object.local?(object)},
  72          {:ok, new_data} <- fetch_and_contain_remote_object_from_id(id),
  73          {:ok, object} <- reinject_object(object, new_data) do
  74       {:ok, object}
  75     else
  76       {:local, true} -> {:ok, object}
  77       e -> {:error, e}
  78     end
  79   end
  80
  81   # Note: will create a Create activity, which we need internally at the moment.
  82   def fetch_object_from_id(id, options \\ []) do
  83     with {_, nil} <- {:fetch_object, Object.get_cached_by_ap_id(id)},
  84          {_, true} <- {:allowed_depth, Federator.allowed_thread_distance?(options[:depth])},
  85          {_, {:ok, data}} <- {:fetch, fetch_and_contain_remote_object_from_id(id)},
  86          {_, nil} <- {:normalize, Object.normalize(data, fetch: false)},
  87          params <- prepare_activity_params(data),
  88          {_, :ok} <- {:containment, Containment.contain_origin(id, params)},
  89          {_, {:ok, activity}} <-
  90            {:transmogrifier, Transmogrifier.handle_incoming(params, options)},
  91          {_, _data, %Object{} = object} <-
  92            {:object, data, Object.normalize(activity, fetch: false)} do
  93       {:ok, object}
  94     else
  95       {:allowed_depth, false} ->
  96         {:error, "Max thread distance exceeded."}
  97
  98       {:containment, _} ->
  99         {:error, "Object containment failed."}
 100
 101       {:transmogrifier, {:error, {:reject, e}}} ->
 102         {:reject, e}
 103
 104       {:transmogrifier, _} = e ->
 105         {:error, e}
 106
 107       {:object, data, nil} ->
 108         reinject_object(%Object{}, data)
 109
 110       {:normalize, object = %Object{}} ->
 111         {:ok, object}
 112
 113       {:fetch_object, %Object{} = object} ->
 114         {:ok, object}
 115
 116       {:fetch, {:error, error}} ->
 117         {:error, error}
 118
 119       e ->
 120         e
 121     end
 122   end
 123
 124   defp prepare_activity_params(data) do
 125     %{
 126       "type" => "Create",
 127       "to" => data["to"] || [],
 128       "cc" => data["cc"] || [],
 129       # Should we seriously keep this attributedTo thing?
 130       "actor" => data["actor"] || data["attributedTo"],
 131       "object" => data
 132     }
 133   end
 134
 135   def fetch_object_from_id!(id, options \\ []) do
 136     with {:ok, object} <- fetch_object_from_id(id, options) do
 137       object
 138     else
 139       {:error, %Tesla.Mock.Error{}} ->
 140         nil
 141
 142       {:error, "Object has been deleted"} ->
 143         nil
 144
 145       {:reject, reason} ->
 146         Logger.info("Rejected #{id} while fetching: #{inspect(reason)}")
 147         nil
 148
 149       e ->
 150         Logger.error("Error while fetching #{id}: #{inspect(e)}")
 151         nil
 152     end
 153   end
 154
 155   defp make_signature(id, date) do
 156     uri = URI.parse(id)
 157
 158     signature =
 159       InternalFetchActor.get_actor()
 160       |> Signature.sign(%{
 161         "(request-target)": "get #{uri.path}",
 162         host: uri.host,
 163         date: date
 164       })
 165
 166     {"signature", signature}
 167   end
 168
 169   defp sign_fetch(headers, id, date) do
 170     if Pleroma.Config.get([:activitypub, :sign_object_fetches]) do
 171       [make_signature(id, date) | headers]
 172     else
 173       headers
 174     end
 175   end
 176
 177   defp maybe_date_fetch(headers, date) do
 178     if Pleroma.Config.get([:activitypub, :sign_object_fetches]) do
 179       [{"date", date} | headers]
 180     else
 181       headers
 182     end
 183   end
 184
 185   def fetch_and_contain_remote_object_from_id(id)
 186
 187   def fetch_and_contain_remote_object_from_id(%{"id" => id}),
 188     do: fetch_and_contain_remote_object_from_id(id)
 189
 190   def fetch_and_contain_remote_object_from_id(id) when is_binary(id) do
 191     Logger.debug("Fetching object #{id} via AP")
 192
 193     with {:scheme, true} <- {:scheme, String.starts_with?(id, "http")},
 194          {:ok, body} <- get_object(id),
 195          {:ok, data} <- safe_json_decode(body),
 196          :ok <- Containment.contain_origin_from_id(id, data) do
 197       {:ok, data}
 198     else
 199       {:scheme, _} ->
 200         {:error, "Unsupported URI scheme"}
 201
 202       {:error, e} ->
 203         {:error, e}
 204
 205       e ->
 206         {:error, e}
 207     end
 208   end
 209
 210   def fetch_and_contain_remote_object_from_id(_id),
 211     do: {:error, "id must be a string"}
 212
 213   defp get_object(id) do
 214     date = Pleroma.Signature.signed_date()
 215
 216     headers =
 217       [{"accept", "application/activity+json"}]
 218       |> maybe_date_fetch(date)
 219       |> sign_fetch(id, date)
 220
 221     case HTTP.get(id, headers) do
 222       {:ok, %{body: body, status: code, headers: headers}} when code in 200..299 ->
 223         case List.keyfind(headers, "content-type", 0) do
 224           {_, content_type} ->
 225             case Plug.Conn.Utils.media_type(content_type) do
 226               {:ok, "application", "activity+json", _} ->
 227                 {:ok, body}
 228
 229               {:ok, "application", "ld+json",
 230                %{"profile" => "https://www.w3.org/ns/activitystreams"}} ->
 231                 {:ok, body}
 232
 233               _ ->
 234                 {:error, {:content_type, content_type}}
 235             end
 236
 237           _ ->
 238             {:error, {:content_type, nil}}
 239         end
 240
 241       {:ok, %{status: code}} when code in [404, 410] ->
 242         {:error, "Object has been deleted"}
 243
 244       {:error, e} ->
 245         {:error, e}
 246
 247       e ->
 248         {:error, e}
 249     end
 250   end
 251
 252   defp safe_json_decode(nil), do: {:ok, nil}
 253   defp safe_json_decode(json), do: Jason.decode(json)
 254 end