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.Maps
   8   alias Pleroma.Object
   9   alias Pleroma.Object.Containment
  10   alias Pleroma.Repo
  11   alias Pleroma.Signature
  12   alias Pleroma.Web.ActivityPub.InternalFetchActor
  13   alias Pleroma.Web.ActivityPub.ObjectValidator
  14   alias Pleroma.Web.ActivityPub.Transmogrifier
  15   alias Pleroma.Web.Federator
  16
  17   require Logger
  18   require Pleroma.Constants
  19
  20   defp touch_changeset(changeset) do
  21     updated_at =
  22       NaiveDateTime.utc_now()
  23       |> NaiveDateTime.truncate(:second)
  24
  25     Ecto.Changeset.put_change(changeset, :updated_at, updated_at)
  26   end
  27
  28   defp maybe_reinject_internal_fields(%{data: %{} = old_data}, new_data) do
  29     has_history? = fn
  30       %{"formerRepresentations" => %{"orderedItems" => list}} when is_list(list) -> true
  31       _ -> false
  32     end
  33
  34     internal_fields = Map.take(old_data, Pleroma.Constants.object_internal_fields())
  35
  36     remote_history_exists? = has_history?.(new_data)
  37
  38     # If the remote history exists, we treat that as the only source of truth.
  39     new_data =
  40       if has_history?.(old_data) and not remote_history_exists? do
  41         Map.put(new_data, "formerRepresentations", old_data["formerRepresentations"])
  42       else
  43         new_data
  44       end
  45
  46     # If the remote does not have history information, we need to manage it ourselves
  47     new_data =
  48       if not remote_history_exists? do
  49         changed? =
  50           Pleroma.Constants.status_updatable_fields()
  51           |> Enum.any?(fn field -> Map.get(old_data, field) != Map.get(new_data, field) end)
  52
  53         %{updated_object: updated_object} =
  54           new_data
  55           |> Object.Updater.maybe_update_history(old_data,
  56             updated: changed?,
  57             use_history_in_new_object?: false
  58           )
  59
  60         updated_object
  61       else
  62         new_data
  63       end
  64
  65     Map.merge(new_data, internal_fields)
  66   end
  67
  68   defp maybe_reinject_internal_fields(_, new_data), do: new_data
  69
  70   @spec reinject_object(struct(), map()) :: {:ok, Object.t()} | {:error, any()}
  71   defp reinject_object(%Object{data: %{"type" => "Question"}} = object, new_data) do
  72     Logger.debug("Reinjecting object #{new_data["id"]}")
  73
  74     with data <- maybe_reinject_internal_fields(object, new_data),
  75          {:ok, data, _} <- ObjectValidator.validate(data, %{}),
  76          changeset <- Object.change(object, %{data: data}),
  77          changeset <- touch_changeset(changeset),
  78          {:ok, object} <- Repo.insert_or_update(changeset),
  79          {:ok, object} <- Object.set_cache(object) do
  80       {:ok, object}
  81     else
  82       e ->
  83         Logger.error("Error while processing object: #{inspect(e)}")
  84         {:error, e}
  85     end
  86   end
  87
  88   defp reinject_object(%Object{} = object, new_data) do
  89     Logger.debug("Reinjecting object #{new_data["id"]}")
  90
  91     with new_data <- Transmogrifier.fix_object(new_data),
  92          data <- maybe_reinject_internal_fields(object, new_data),
  93          changeset <- Object.change(object, %{data: data}),
  94          changeset <- touch_changeset(changeset),
  95          {:ok, object} <- Repo.insert_or_update(changeset),
  96          {:ok, object} <- Object.set_cache(object) do
  97       {:ok, object}
  98     else
  99       e ->
 100         Logger.error("Error while processing object: #{inspect(e)}")
 101         {:error, e}
 102     end
 103   end
 104
 105   def refetch_object(%Object{data: %{"id" => id}} = object) do
 106     with {:local, false} <- {:local, Object.local?(object)},
 107          {:ok, new_data} <- fetch_and_contain_remote_object_from_id(id),
 108          {:ok, object} <- reinject_object(object, new_data) do
 109       {:ok, object}
 110     else
 111       {:local, true} -> {:ok, object}
 112       e -> {:error, e}
 113     end
 114   end
 115
 116   # Note: will create a Create activity, which we need internally at the moment.
 117   def fetch_object_from_id(id, options \\ []) do
 118     with {_, nil} <- {:fetch_object, Object.get_cached_by_ap_id(id)},
 119          {_, true} <- {:allowed_depth, Federator.allowed_thread_distance?(options[:depth])},
 120          {_, {:ok, data}} <- {:fetch, fetch_and_contain_remote_object_from_id(id)},
 121          {_, nil} <- {:normalize, Object.normalize(data, fetch: false)},
 122          params <- prepare_activity_params(data),
 123          {_, :ok} <- {:containment, Containment.contain_origin(id, params)},
 124          {_, {:ok, activity}} <-
 125            {:transmogrifier, Transmogrifier.handle_incoming(params, options)},
 126          {_, _data, %Object{} = object} <-
 127            {:object, data, Object.normalize(activity, fetch: false)} do
 128       {:ok, object}
 129     else
 130       {:allowed_depth, false} ->
 131         {:error, "Max thread distance exceeded."}
 132
 133       {:containment, _} ->
 134         {:error, "Object containment failed."}
 135
 136       {:transmogrifier, {:error, {:reject, e}}} ->
 137         {:reject, e}
 138
 139       {:transmogrifier, {:reject, e}} ->
 140         {:reject, e}
 141
 142       {:transmogrifier, _} = e ->
 143         {:error, e}
 144
 145       {:object, data, nil} ->
 146         reinject_object(%Object{}, data)
 147
 148       {:normalize, object = %Object{}} ->
 149         {:ok, object}
 150
 151       {:fetch_object, %Object{} = object} ->
 152         {:ok, object}
 153
 154       {:fetch, {:error, error}} ->
 155         {:error, error}
 156
 157       e ->
 158         e
 159     end
 160   end
 161
 162   defp prepare_activity_params(data) do
 163     %{
 164       "type" => "Create",
 165       # Should we seriously keep this attributedTo thing?
 166       "actor" => data["actor"] || data["attributedTo"],
 167       "object" => data
 168     }
 169     |> Maps.put_if_present("to", data["to"])
 170     |> Maps.put_if_present("cc", data["cc"])
 171     |> Maps.put_if_present("bto", data["bto"])
 172     |> Maps.put_if_present("bcc", data["bcc"])
 173   end
 174
 175   def fetch_object_from_id!(id, options \\ []) do
 176     with {:ok, object} <- fetch_object_from_id(id, options) do
 177       object
 178     else
 179       {:error, %Tesla.Mock.Error{}} ->
 180         nil
 181
 182       {:error, "Object has been deleted"} ->
 183         nil
 184
 185       {:reject, reason} ->
 186         Logger.info("Rejected #{id} while fetching: #{inspect(reason)}")
 187         nil
 188
 189       e ->
 190         Logger.error("Error while fetching #{id}: #{inspect(e)}")
 191         nil
 192     end
 193   end
 194
 195   defp make_signature(id, date) do
 196     uri = URI.parse(id)
 197
 198     signature =
 199       InternalFetchActor.get_actor()
 200       |> Signature.sign(%{
 201         "(request-target)": "get #{uri.path}",
 202         host: uri.host,
 203         date: date
 204       })
 205
 206     {"signature", signature}
 207   end
 208
 209   defp sign_fetch(headers, id, date) do
 210     if Pleroma.Config.get([:activitypub, :sign_object_fetches]) do
 211       [make_signature(id, date) | headers]
 212     else
 213       headers
 214     end
 215   end
 216
 217   defp maybe_date_fetch(headers, date) do
 218     if Pleroma.Config.get([:activitypub, :sign_object_fetches]) do
 219       [{"date", date} | headers]
 220     else
 221       headers
 222     end
 223   end
 224
 225   def fetch_and_contain_remote_object_from_id(id)
 226
 227   def fetch_and_contain_remote_object_from_id(%{"id" => id}),
 228     do: fetch_and_contain_remote_object_from_id(id)
 229
 230   def fetch_and_contain_remote_object_from_id(id) when is_binary(id) do
 231     Logger.debug("Fetching object #{id} via AP")
 232
 233     with {:scheme, true} <- {:scheme, String.starts_with?(id, "http")},
 234          {:ok, body} <- get_object(id),
 235          {:ok, data} <- safe_json_decode(body),
 236          :ok <- Containment.contain_origin_from_id(id, data) do
 237       {:ok, data}
 238     else
 239       {:scheme, _} ->
 240         {:error, "Unsupported URI scheme"}
 241
 242       {:error, e} ->
 243         {:error, e}
 244
 245       e ->
 246         {:error, e}
 247     end
 248   end
 249
 250   def fetch_and_contain_remote_object_from_id(_id),
 251     do: {:error, "id must be a string"}
 252
 253   defp get_object(id) do
 254     date = Pleroma.Signature.signed_date()
 255
 256     headers =
 257       [{"accept", "application/activity+json"}]
 258       |> maybe_date_fetch(date)
 259       |> sign_fetch(id, date)
 260
 261     case HTTP.get(id, headers) do
 262       {:ok, %{body: body, status: code, headers: headers}} when code in 200..299 ->
 263         case List.keyfind(headers, "content-type", 0) do
 264           {_, content_type} ->
 265             case Plug.Conn.Utils.media_type(content_type) do
 266               {:ok, "application", "activity+json", _} ->
 267                 {:ok, body}
 268
 269               {:ok, "application", "ld+json",
 270                %{"profile" => "https://www.w3.org/ns/activitystreams"}} ->
 271                 {:ok, body}
 272
 273               _ ->
 274                 {:error, {:content_type, content_type}}
 275             end
 276
 277           _ ->
 278             {:error, {:content_type, nil}}
 279         end
 280
 281       {:ok, %{status: code}} when code in [404, 410] ->
 282         {:error, "Object has been deleted"}
 283
 284       {:error, e} ->
 285         {:error, e}
 286
 287       e ->
 288         {:error, e}
 289     end
 290   end
 291
 292   defp safe_json_decode(nil), do: {:ok, nil}
 293   defp safe_json_decode(json), do: Jason.decode(json)
 294 end