WIP bonfire-networks/bonfire-app#912 and bonfire-networks/bonfire-app…

…#899

config/config.exs

@@ -64,6 +64,8 @@ config :activity_pub, :instance,
   hostname: "localhost",
   federation_publisher_modules: [ActivityPub.Federator.APPublisher],
   federation_reachability_timeout_days: 7,
+  # Max. depth of reply-to and reply activities fetching on incoming federation, to prevent out-of-memory situations while fetching very long threads.
+  federation_incoming_max_recursion: 10,
   rewrite_policy: [],
   handle_unknown_activities: false
 

lib/federator/fetcher.ex

@@ -75,6 +75,7 @@ defmodule ActivityPub.Federator.Fetcher do
     end)
   end
 
+  @doc "Fetch a list of objects within recursion limits. Used for reply_to/context, and replies or similar collections."
   def maybe_fetch(entries, opts \\ [])
   def maybe_fetch([], _opts), do: nil
 
@@ -93,6 +94,8 @@ defmodule ActivityPub.Federator.Fetcher do
             entry_depth = depth + index
 
             if allowed_recursion?(entry_depth, max_items) do
+              info(id, "fetch recursed (async)")
+
               enqueue_fetch(
                 id,
                 Enum.into(opts[:worker_attrs] || %{}, %{
@@ -103,13 +106,13 @@ defmodule ActivityPub.Federator.Fetcher do
             end
           end
 
-          nil
-
         true ->
           for {id, index} <- Enum.with_index(entries) do
             entry_depth = depth + index
 
             if allowed_recursion?(entry_depth, max_items) do
+              info(id, "fetch recursed (inline)")
+
               fetch_object_from_id(id,
                 depth: entry_depth
               )
@@ -712,6 +715,7 @@ defmodule ActivityPub.Federator.Fetcher do
   def allowed_recursion?(distance, max_recursion \\ nil) do
     max_distance = max_recursion || max_recursion()
 
+    debug(distance, "distance")
     debug(max_distance, "max_distance")
 
     if is_number(distance) and is_number(max_distance) and max_distance >= 0 do

lib/federator/transformer.ex

@@ -284,7 +284,7 @@ defmodule ActivityPub.Federator.Transformer do
   def fix_in_reply_to(%{"inReplyTo" => in_reply_to} = object, options)
       when not is_nil(in_reply_to) do
     with in_reply_to_id when is_binary(in_reply_to_id) <- Utils.single_ap_id(in_reply_to),
-         _ <- Fetcher.maybe_fetch(in_reply_to_id, options) do
+         _ <- Fetcher.maybe_fetch(in_reply_to_id, options) |> info("fetched reply_to?") do
       object
       |> Map.put("inReplyTo", in_reply_to_id)
       # |> Map.put("context", replied_object.data["context"] || object["conversation"]) # TODO as an update when we get the async inReplyTo?
@@ -343,7 +343,9 @@ defmodule ActivityPub.Federator.Transformer do
     context = object["context"] || object["conversation"] || object["inReplyTo"]
 
     with context when is_binary(context) <- Utils.single_ap_id(context),
-         _ <- Fetcher.maybe_fetch(context, options) do
+         _ <-
+           Fetcher.maybe_fetch(context, options)
+           |> info("fetched context?") do
       object
       |> Map.put("context", context)
     else
@@ -591,27 +593,35 @@ defmodule ActivityPub.Federator.Transformer do
   def fix_replies(%{"replies" => replies} = data, options)
       when is_list(replies) and replies != [] do
     Fetcher.maybe_fetch(replies, options)
+    |> info("fetched replies?")
+
     # TODO: update the data with only IDs in case we have full objects?
     data
   end
 
   def fix_replies(%{"replies" => %{"items" => replies}} = data, options)
       when is_list(replies) and replies != [] do
     Fetcher.maybe_fetch(replies, options)
+    |> info("fetched replies?")
+
     # TODO: update the data with only IDs in case we have full objects?
     Map.put(data, "replies", replies)
   end
 
   def fix_replies(%{"replies" => %{"first" => replies}} = data, options)
       when is_list(replies) and replies != [] do
     Fetcher.maybe_fetch(replies, options)
+    |> info("fetched replies?")
+
     # TODO: update the data with only IDs in case we have full objects?
     Map.put(data, "replies", replies)
   end
 
   def fix_replies(%{"replies" => %{"first" => %{"items" => replies}}} = data, options)
       when is_list(replies) and replies != [] do
     Fetcher.maybe_fetch(replies, options)
+    |> info("fetched replies?")
+
     # TODO: update the data with only IDs in case we have full objects?
     Map.put(data, "replies", replies)
   end
@@ -634,7 +644,7 @@ defmodule ActivityPub.Federator.Transformer do
       )
       |> debug("opts")
     )
-    |> debug()
+    |> info("fetched collection?")
 
     data
   end

lib/federator/workers/remote_fetcher_worker.ex

@@ -3,7 +3,10 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule ActivityPub.Federator.Workers.RemoteFetcherWorker do
-  use ActivityPub.Federator.Worker, queue: "remote_fetcher"
+  use ActivityPub.Federator.Worker,
+    queue: "remote_fetcher",
+    unique: [fields: [:args], keys: [:op, :id]]
+
   alias ActivityPub.Federator.Fetcher
 
   @impl Oban.Worker

lib/federator/workers/worker.ex

@@ -13,12 +13,11 @@ defmodule ActivityPub.Federator.Worker do
   defmacro __using__(opts) do
     caller_module = __CALLER__.module
     queue = Keyword.fetch!(opts, :queue)
+    opts = Keyword.put_new(opts, :max_attempts, 3)
 
     quote do
       # Note: `max_attempts` is intended to be overridden in `new/2` call
-      use Oban.Worker,
-        queue: unquote(queue),
-        max_attempts: 3
+      use Oban.Worker, unquote(opts)
 
       def enqueueable(op, params, worker_args \\ []) do
         params = Map.merge(%{"op" => op, "repo" => ActivityPub.Utils.repo()}, params)

lib/fixtures.ex

@@ -350,6 +350,14 @@ defmodule ActivityPub.Fixtures do
             headers: ActivityPub.Utils.activitypub_object_headers()
           }
         end,
+      "https://mastodon.local/users/admin/statuses/8511" =>
+        fn "https://mastodon.local/users/admin/statuses/8511", _, _, _ ->
+          %Tesla.Env{
+            status: 200,
+            body: file("fixtures/mastodon/mastodon-note-object-reply.json"),
+            headers: ActivityPub.Utils.activitypub_object_headers()
+          }
+        end,
       "https://mocked.local/users/karen" => fn "https://mocked.local/users/karen", _, _, _ ->
         %Tesla.Env{status: 200, body: file("fixtures/pleroma_user_actor.json")}
       end,

test/activity_pub/federator/fetcher_recursion_test.exs

@@ -0,0 +1,91 @@
+defmodule ActivityPub.Federator.FetcherRecursionTest do
+  use ActivityPub.DataCase, async: false
+  import Tesla.Mock
+  import Mock
+
+  alias ActivityPub.Federator.Fetcher
+  alias ActivityPub.Federator.WebFinger
+
+  alias ActivityPub.Object, as: Activity
+  alias ActivityPub.Instances
+  alias ActivityPub.Object
+
+  alias ActivityPub.Test.HttpRequestMock
+
+  setup_all do
+    mock_global(fn
+      env ->
+        HttpRequestMock.request(env)
+    end)
+
+    :ok
+  end
+
+  describe "max thread distance restriction" do
+    @ap_id "https://mastodon.local/@admin/99512778738411822"
+    @reply_ap_id "https://mastodon.local/users/admin/statuses/8511"
+    @reply_2_ap_id "https://mocked.local/objects/eb3b1181-38cc-4eaf-ba1b-3f5431fa9779"
+
+    setup do: clear_config([:instance, :federation_incoming_max_recursion])
+
+    test "it returns error if thread depth is exceeded" do
+      clear_config([:instance, :federation_incoming_max_recursion], 0)
+
+      assert {:error, "Stopping to avoid too much recursion"} =
+               Fetcher.fetch_object_from_id(@ap_id, depth: 1)
+
+      assert {:error, :not_found} = ActivityPub.Object.get_cached(ap_id: @ap_id)
+    end
+
+    test "it fetches object if max thread depth is restricted to 0 and depth is not specified" do
+      clear_config([:instance, :federation_incoming_max_recursion], 0)
+
+      assert {:ok, _} = Fetcher.fetch_object_from_id(@ap_id)
+    end
+
+    test "it fetches object if requested depth does not exceed max thread depth" do
+      clear_config([:instance, :federation_incoming_max_recursion], 10)
+
+      assert {:ok, _} = Fetcher.fetch_object_from_id(@ap_id, depth: 10)
+    end
+
+    test "it fetches reply_to and replies if thread depth is not exceeded" do
+      clear_config([:instance, :federation_incoming_max_recursion], 4)
+
+      assert {:ok, _} =
+               Fetcher.fetch_object_from_id(@reply_ap_id, depth: 1)
+
+      Oban.drain_queue(queue: :remote_fetcher, with_recursion: true)
+
+      assert {:ok, _} = ActivityPub.Object.get_cached(ap_id: @reply_ap_id)
+      assert {:ok, _} = ActivityPub.Object.get_cached(ap_id: @ap_id)
+      assert {:ok, _} = ActivityPub.Object.get_cached(ap_id: @reply_2_ap_id)
+    end
+
+    test "it fetches reply_to if thread depth is not exceeded" do
+      clear_config([:instance, :federation_incoming_max_recursion], 2)
+
+      assert {:ok, _} =
+               Fetcher.fetch_object_from_id(@reply_ap_id, depth: 1)
+
+      Oban.drain_queue(queue: :remote_fetcher, with_recursion: true)
+
+      assert {:ok, _} = ActivityPub.Object.get_cached(ap_id: @reply_ap_id)
+      assert {:ok, _} = ActivityPub.Object.get_cached(ap_id: @ap_id)
+      assert {:error, :not_found} = ActivityPub.Object.get_cached(ap_id: @reply_2_ap_id)
+    end
+
+    test "it does not fetch reply_to if thread depth is exceeded" do
+      clear_config([:instance, :federation_incoming_max_recursion], 3)
+
+      assert {:ok, _} =
+               Fetcher.fetch_object_from_id(@reply_ap_id, depth: 3)
+
+      Oban.drain_queue(queue: :remote_fetcher, with_recursion: true)
+
+      assert {:ok, _} = ActivityPub.Object.get_cached(ap_id: @reply_ap_id)
+      assert {:error, :not_found} = ActivityPub.Object.get_cached(ap_id: @ap_id)
+      assert {:error, :not_found} = ActivityPub.Object.get_cached(ap_id: @reply_2_ap_id)
+    end
+  end
+end

test/fixtures/mastodon/mastodon-note-object-reply.json

@@ -0,0 +1,45 @@
+{
+   "@context" : [
+      "https://www.w3.org/ns/activitystreams",
+      "https://w3id.org/security/v1",
+      {
+         "Emoji" : "toot:Emoji",
+         "Hashtag" : "as:Hashtag",
+         "atomUri" : "ostatus:atomUri",
+         "conversation" : "ostatus:conversation",
+         "inReplyToAtomUri" : "ostatus:inReplyToAtomUri",
+         "manuallyApprovesFollowers" : "as:manuallyApprovesFollowers",
+         "movedTo" : "as:movedTo",
+         "ostatus" : "http://ostatus.org#",
+         "sensitive" : "as:sensitive",
+         "toot" : "http://joinmastodon.org/ns#"
+      }
+   ],
+   "atomUri" : "https://mastodon.local/users/admin/statuses/8511",
+   "attachment" : [
+      {
+         "mediaType" : "image/jpeg",
+         "name" : null,
+         "type" : "Document",
+         "url" : "https://mastodon.local/system/media_attachments/files/000/000/002/original/334ce029e7bfb920.jpg"
+      }
+   ],
+   "attributedTo" : "https://mastodon.local/users/admin",
+   "cc" : [
+      "https://mastodon.local/users/admin/followers"
+   ],
+   "content" : "<p>yeah.</p>",
+   "conversation" : "tag:mastodon.local,2018-02-17:objectId=59:objectType=Conversation",
+   "id" : "https://mastodon.local/users/admin/statuses/8511",
+   "inReplyTo" : "https://mastodon.local/users/admin/statuses/99512778738411822",
+   "inReplyToAtomUri" : null,
+   "published" : "2018-02-17T17:46:20Z",
+   "sensitive" : false,
+   "summary" : null,
+   "tag" : [],
+   "to" : [
+      "https://www.w3.org/ns/activitystreams#Public"
+   ],
+   "type" : "Note",
+   "url" : "https://mastodon.local/@admin/8511"
+}

test/fixtures/mastodon/mastodon-note-object.json

@@ -34,6 +34,19 @@
    "inReplyTo" : null,
    "inReplyToAtomUri" : null,
    "published" : "2018-02-17T17:46:20Z",
+   "replies": {
+      "id": "https://mastodon.local/users/admin/statuses/99512778738411822/replies",
+      "type": "Collection",
+      "first": {
+         "type": "CollectionPage",
+         "next": "https://mastodon.local/users/admin/statuses/99512778738411822/replies?min_id=99512778738411824&page=true",
+         "partOf": "https://mastodon.local/users/admin/statuses/99512778738411822/replies",
+         "items": [
+            "https://mastodon.local/users/admin/statuses/8511",
+            "https://mocked.local/objects/eb3b1181-38cc-4eaf-ba1b-3f5431fa9779"
+         ]
+      }
+   },
    "sensitive" : false,
    "summary" : null,
    "tag" : [],

test/fixtures/mastodon/mastodon-post-activity.json

@@ -43,8 +43,8 @@
                 "next": "https://mastodon.local/users/admin/statuses/99512778738411822/replies?min_id=99512778738411824&page=true",
                 "partOf": "https://mastodon.local/users/admin/statuses/99512778738411822/replies",
                 "items": [
-                    "https://mastodon.local/users/admin/statuses/99512778738411823",
-                    "https://mastodon.local/users/admin/statuses/99512778738411824"
+                    "https://mastodon.local/users/admin/statuses/8511",
+                    "https://mocked.local/objects/eb3b1181-38cc-4eaf-ba1b-3f5431fa9779"
                 ]
             }
         },