1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Object.Fetcher do
7 alias Pleroma.Instances
10 alias Pleroma.Object.Containment
12 alias Pleroma.Signature
13 alias Pleroma.Web.ActivityPub.InternalFetchActor
14 alias Pleroma.Web.ActivityPub.ObjectValidator
15 alias Pleroma.Web.ActivityPub.Transmogrifier
16 alias Pleroma.Web.Federator
19 require Pleroma.Constants
21 defp touch_changeset(changeset) do
23 NaiveDateTime.utc_now()
24 |> NaiveDateTime.truncate(:second)
26 Ecto.Changeset.put_change(changeset, :updated_at, updated_at)
29 defp maybe_reinject_internal_fields(%{data: %{} = old_data}, new_data) do
31 %{"formerRepresentations" => %{"orderedItems" => list}} when is_list(list) -> true
35 internal_fields = Map.take(old_data, Pleroma.Constants.object_internal_fields())
37 remote_history_exists? = has_history?.(new_data)
39 # If the remote history exists, we treat that as the only source of truth.
41 if has_history?.(old_data) and not remote_history_exists? do
42 Map.put(new_data, "formerRepresentations", old_data["formerRepresentations"])
47 # If the remote does not have history information, we need to manage it ourselves
49 if not remote_history_exists? do
51 Pleroma.Constants.status_updatable_fields()
52 |> Enum.any?(fn field -> Map.get(old_data, field) != Map.get(new_data, field) end)
54 %{updated_object: updated_object} =
56 |> Object.Updater.maybe_update_history(old_data,
58 use_history_in_new_object?: false
66 Map.merge(new_data, internal_fields)
69 defp maybe_reinject_internal_fields(_, new_data), do: new_data
71 @spec reinject_object(struct(), map()) :: {:ok, Object.t()} | {:error, any()}
72 defp reinject_object(%Object{data: %{"type" => "Question"}} = object, new_data) do
73 Logger.debug("Reinjecting object #{new_data["id"]}")
75 with data <- maybe_reinject_internal_fields(object, new_data),
76 {:ok, data, _} <- ObjectValidator.validate(data, %{}),
77 changeset <- Object.change(object, %{data: data}),
78 changeset <- touch_changeset(changeset),
79 {:ok, object} <- Repo.insert_or_update(changeset),
80 {:ok, object} <- Object.set_cache(object) do
84 Logger.error("Error while processing object: #{inspect(e)}")
89 defp reinject_object(%Object{} = object, new_data) do
90 Logger.debug("Reinjecting object #{new_data["id"]}")
92 with new_data <- Transmogrifier.fix_object(new_data),
93 data <- maybe_reinject_internal_fields(object, new_data),
94 changeset <- Object.change(object, %{data: data}),
95 changeset <- touch_changeset(changeset),
96 {:ok, object} <- Repo.insert_or_update(changeset),
97 {:ok, object} <- Object.set_cache(object) do
101 Logger.error("Error while processing object: #{inspect(e)}")
106 def refetch_object(%Object{data: %{"id" => id}} = object) do
107 with {:local, false} <- {:local, Object.local?(object)},
108 {:ok, new_data} <- fetch_and_contain_remote_object_from_id(id),
109 {:ok, object} <- reinject_object(object, new_data) do
112 {:local, true} -> {:ok, object}
117 # Note: will create a Create activity, which we need internally at the moment.
118 def fetch_object_from_id(id, options \\ []) do
119 with {_, nil} <- {:fetch_object, Object.get_cached_by_ap_id(id)},
120 {_, true} <- {:allowed_depth, Federator.allowed_thread_distance?(options[:depth])},
121 {_, {:ok, data}} <- {:fetch, fetch_and_contain_remote_object_from_id(id)},
122 {_, nil} <- {:normalize, Object.normalize(data, fetch: false)},
123 params <- prepare_activity_params(data),
124 {_, :ok} <- {:containment, Containment.contain_origin(id, params)},
125 {_, {:ok, activity}} <-
126 {:transmogrifier, Transmogrifier.handle_incoming(params, options)},
127 {_, _data, %Object{} = object} <-
128 {:object, data, Object.normalize(activity, fetch: false)} do
131 {:allowed_depth, false} ->
132 {:error, "Max thread distance exceeded."}
135 {:error, "Object containment failed."}
137 {:transmogrifier, {:error, {:reject, e}}} ->
140 {:transmogrifier, {:reject, e}} ->
143 {:transmogrifier, _} = e ->
146 {:object, data, nil} ->
147 reinject_object(%Object{}, data)
149 {:normalize, object = %Object{}} ->
152 {:fetch_object, %Object{} = object} ->
155 {:fetch, {:error, error}} ->
163 defp prepare_activity_params(data) do
166 # Should we seriously keep this attributedTo thing?
167 "actor" => data["actor"] || data["attributedTo"],
170 |> Maps.put_if_present("to", data["to"])
171 |> Maps.put_if_present("cc", data["cc"])
172 |> Maps.put_if_present("bto", data["bto"])
173 |> Maps.put_if_present("bcc", data["bcc"])
176 def fetch_object_from_id!(id, options \\ []) do
177 with {:ok, object} <- fetch_object_from_id(id, options) do
180 {:error, %Tesla.Mock.Error{}} ->
183 {:error, {"Object has been deleted", _id, _code}} ->
187 Logger.debug("Rejected #{id} while fetching: #{inspect(reason)}")
191 Logger.error("Error while fetching #{id}: #{inspect(e)}")
196 defp make_signature(id, date) do
200 InternalFetchActor.get_actor()
202 "(request-target)": "get #{uri.path}",
207 {"signature", signature}
210 defp sign_fetch(headers, id, date) do
211 if Pleroma.Config.get([:activitypub, :sign_object_fetches]) do
212 [make_signature(id, date) | headers]
218 defp maybe_date_fetch(headers, date) do
219 if Pleroma.Config.get([:activitypub, :sign_object_fetches]) do
220 [{"date", date} | headers]
226 def fetch_and_contain_remote_object_from_id(id)
228 def fetch_and_contain_remote_object_from_id(%{"id" => id}),
229 do: fetch_and_contain_remote_object_from_id(id)
231 def fetch_and_contain_remote_object_from_id(id) when is_binary(id) do
232 Logger.debug("Fetching object #{id} via AP")
234 with {:scheme, true} <- {:scheme, String.starts_with?(id, "http")},
235 {:ok, body} <- get_object(id),
236 {:ok, data} <- safe_json_decode(body),
237 :ok <- Containment.contain_origin_from_id(id, data) do
238 unless Instances.reachable?(id) do
239 Instances.set_reachable(id)
245 {:error, "Unsupported URI scheme"}
255 def fetch_and_contain_remote_object_from_id(_id),
256 do: {:error, "id must be a string"}
258 defp get_object(id) do
259 date = Pleroma.Signature.signed_date()
262 [{"accept", "application/activity+json"}]
263 |> maybe_date_fetch(date)
264 |> sign_fetch(id, date)
266 case HTTP.get(id, headers) do
267 {:ok, %{body: body, status: code, headers: headers}} when code in 200..299 ->
268 case List.keyfind(headers, "content-type", 0) do
270 case Plug.Conn.Utils.media_type(content_type) do
271 {:ok, "application", "activity+json", _} ->
274 {:ok, "application", "ld+json",
275 %{"profile" => "https://www.w3.org/ns/activitystreams"}} ->
279 {:error, {:content_type, content_type}}
283 {:error, {:content_type, nil}}
286 {:ok, %{status: code}} when code in [404, 410] ->
287 {:error, {"Object has been deleted", id, code}}
297 defp safe_json_decode(nil), do: {:ok, nil}
298 defp safe_json_decode(json), do: Jason.decode(json)