Ingestion of remote collection items (#38106)

app/lib/activitypub/activity/add.rb

@@ -12,6 +12,9 @@ class ActivityPub::Activity::Add < ActivityPub::Activity
       else
         add_featured
       end
+    else
+      @collection = @account.collections.find_by(uri: @json['target'])
+      add_collection_item if @collection && Mastodon::Feature.collections_federation_enabled?
     end
   end
 
@@ -30,4 +33,8 @@ class ActivityPub::Activity::Add < ActivityPub::Activity
 
     FeaturedTag.create!(account: @account, name: name) if name.present?
   end
+
+  def add_collection_item
+    ActivityPub::ProcessFeaturedItemService.new.call(@collection, @object)
+  end
 end

app/models/collection_item.rb

@@ -29,7 +29,7 @@ class CollectionItem < ApplicationRecord
 
   validates :position, numericality: { only_integer: true, greater_than: 0 }
   validates :activity_uri, presence: true, if: :local_item_with_remote_account?
-  validates :approval_uri, absence: true, unless: :local?
+  validates :approval_uri, presence: true, unless: -> { local? || account&.local? }
   validates :account, presence: true, if: :accepted?
   validates :object_uri, presence: true, if: -> { account.nil? }
   validates :uri, presence: true, if: :remote?

app/services/activitypub/process_featured_item_service.rb

@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+class ActivityPub::ProcessFeaturedItemService
+  include JsonLdHelper
+  include Lockable
+  include Redisable
+
+  def call(collection, uri_or_object)
+    item_json = uri_or_object.is_a?(String) ? fetch_resource(uri_or_object, true) : uri_or_object
+    return if non_matching_uri_hosts?(collection.uri, item_json['id'])
+
+    with_redis_lock("collection_item:#{item_json['id']}") do
+      return if collection.collection_items.exists?(uri: item_json['id'])
+
+      @collection_item = collection.collection_items.create!(
+        uri: item_json['id'],
+        object_uri: item_json['featuredObject'],
+        approval_uri: item_json['featureAuthorization']
+      )
+
+      verify_authorization!
+
+      @collection_item
+    end
+  end
+
+  private
+
+  def verify_authorization!
+    ActivityPub::VerifyFeaturedItemService.new.call(@collection_item)
+  rescue Mastodon::RecursionLimitExceededError, Mastodon::UnexpectedResponseError, *Mastodon::HTTP_CONNECTION_ERRORS
+    ActivityPub::VerifyFeaturedItemWorker.perform_in(rand(30..600).seconds, @collection_item.id)
+  end
+end

app/services/activitypub/verify_featured_item_service.rb

@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+class ActivityPub::VerifyFeaturedItemService
+  include JsonLdHelper
+
+  def call(collection_item)
+    @collection_item = collection_item
+    @authorization = fetch_resource(@collection_item.approval_uri, true, raise_on_error: :temporary)
+
+    if @authorization.nil?
+      @collection_item.update!(state: :rejected)
+      return
+    end
+
+    return if non_matching_uri_hosts?(@collection_item.approval_uri, @authorization['interactionTarget'])
+    return unless matching_type? && matching_collection_uri?
+
+    account = Account.where(uri: @collection_item.object_uri).first
+    account ||= ActivityPub::FetchRemoteAccountService.new.call(@collection_item.object_uri)
+    return if account.blank?
+
+    @collection_item.update!(account:, state: :accepted)
+  end
+
+  private
+
+  def matching_type?
+    supported_context?(@authorization) && equals_or_includes?(@authorization['type'], 'FeatureAuthorization')
+  end
+
+  def matching_collection_uri?
+    @collection_item.collection.uri == @authorization['interactingObject']
+  end
+end

app/workers/activitypub/verify_featured_item_worker.rb

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+class ActivityPub::VerifyFeaturedItemWorker
+  include Sidekiq::Worker
+  include ExponentialBackoff
+  include JsonLdHelper
+
+  sidekiq_options queue: 'pull', retry: 5
+
+  def perform(collection_item_id)
+    collection_item = CollectionItem.find(collection_item_id)
+
+    ActivityPub::VerifyFeaturedItemService.new.call(collection_item)
+  rescue ActiveRecord::RecordNotFound
+    # Do nothing
+    nil
+  rescue Mastodon::UnexpectedResponseError => e
+    raise e unless response_error_unsalvageable?(e.response)
+  end
+end

spec/lib/activitypub/activity/add_spec.rb

@@ -3,78 +3,118 @@
 require 'rails_helper'
 
 RSpec.describe ActivityPub::Activity::Add do
-  let(:sender) { Fabricate(:account, featured_collection_url: 'https://example.com/featured', domain: 'example.com') }
-  let(:status) { Fabricate(:status, account: sender, visibility: :private) }
+  context 'when the target is the featured collection' do
+    let(:sender) { Fabricate(:account, featured_collection_url: 'https://example.com/featured', domain: 'example.com') }
+    let(:status) { Fabricate(:status, account: sender, visibility: :private) }
 
-  let(:json) do
-    {
-      '@context': 'https://www.w3.org/ns/activitystreams',
-      id: 'foo',
-      type: 'Add',
-      actor: ActivityPub::TagManager.instance.uri_for(sender),
-      object: ActivityPub::TagManager.instance.uri_for(status),
-      target: sender.featured_collection_url,
-    }.with_indifferent_access
-  end
-
-  describe '#perform' do
-    subject { described_class.new(json, sender) }
-
-    it 'creates a pin' do
-      subject.perform
-      expect(sender.pinned?(status)).to be true
+    let(:json) do
+      {
+        '@context': 'https://www.w3.org/ns/activitystreams',
+        id: 'foo',
+        type: 'Add',
+        actor: ActivityPub::TagManager.instance.uri_for(sender),
+        object: ActivityPub::TagManager.instance.uri_for(status),
+        target: sender.featured_collection_url,
+      }.with_indifferent_access
     end
 
-    context 'when status was not known before' do
-      let(:service_stub) { instance_double(ActivityPub::FetchRemoteStatusService) }
+    describe '#perform' do
+      subject { described_class.new(json, sender) }
 
-      let(:json) do
-        {
-          '@context': 'https://www.w3.org/ns/activitystreams',
-          id: 'foo',
-          type: 'Add',
-          actor: ActivityPub::TagManager.instance.uri_for(sender),
-          object: 'https://example.com/unknown',
-          target: sender.featured_collection_url,
-        }.with_indifferent_access
+      it 'creates a pin' do
+        subject.perform
+        expect(sender.pinned?(status)).to be true
       end
 
-      before do
-        allow(ActivityPub::FetchRemoteStatusService).to receive(:new).and_return(service_stub)
-      end
+      context 'when status was not known before' do
+        let(:service_stub) { instance_double(ActivityPub::FetchRemoteStatusService) }
+
+        let(:json) do
+          {
+            '@context': 'https://www.w3.org/ns/activitystreams',
+            id: 'foo',
+            type: 'Add',
+            actor: ActivityPub::TagManager.instance.uri_for(sender),
+            object: 'https://example.com/unknown',
+            target: sender.featured_collection_url,
+          }.with_indifferent_access
+        end
 
-      context 'when there is a local follower' do
         before do
-          account = Fabricate(:account)
-          account.follow!(sender)
+          allow(ActivityPub::FetchRemoteStatusService).to receive(:new).and_return(service_stub)
         end
 
-        it 'fetches the status and pins it' do
-          allow(service_stub).to receive(:call) do |uri, id: true, on_behalf_of: nil, **|
-            expect(uri).to eq 'https://example.com/unknown'
-            expect(id).to be true
-            expect(on_behalf_of&.following?(sender)).to be true
-            status
+        context 'when there is a local follower' do
+          before do
+            account = Fabricate(:account)
+            account.follow!(sender)
           end
-          subject.perform
-          expect(service_stub).to have_received(:call)
-          expect(sender.pinned?(status)).to be true
-        end
-      end
 
-      context 'when there is no local follower' do
-        it 'tries to fetch the status' do
-          allow(service_stub).to receive(:call) do |uri, id: true, on_behalf_of: nil, **|
-            expect(uri).to eq 'https://example.com/unknown'
-            expect(id).to be true
-            expect(on_behalf_of).to be_nil
-            nil
+          it 'fetches the status and pins it' do
+            allow(service_stub).to receive(:call) do |uri, id: true, on_behalf_of: nil, **|
+              expect(uri).to eq 'https://example.com/unknown'
+              expect(id).to be true
+              expect(on_behalf_of&.following?(sender)).to be true
+              status
+            end
+            subject.perform
+            expect(service_stub).to have_received(:call)
+            expect(sender.pinned?(status)).to be true
+          end
+        end
+
+        context 'when there is no local follower' do
+          it 'tries to fetch the status' do
+            allow(service_stub).to receive(:call) do |uri, id: true, on_behalf_of: nil, **|
+              expect(uri).to eq 'https://example.com/unknown'
+              expect(id).to be true
+              expect(on_behalf_of).to be_nil
+              nil
+            end
+            subject.perform
+            expect(service_stub).to have_received(:call)
+            expect(sender.pinned?(status)).to be false
           end
-          subject.perform
-          expect(service_stub).to have_received(:call)
-          expect(sender.pinned?(status)).to be false
         end
       end
     end
   end
+
+  context 'when the target is a collection', feature: :collections_federation do
+    subject { described_class.new(activity_json, collection.account) }
+
+    let(:collection) { Fabricate(:remote_collection) }
+    let(:featured_item_json) do
+      {
+        '@context' => 'https://www.w3.org/ns/activitystreams',
+        'id' => 'https://other.example.com/featured_item/1',
+        'type' => 'FeaturedItem',
+        'featuredObject' => 'https://example.com/actor/1',
+        'featuredObjectType' => 'Person',
+        'featureAuthorization' => 'https://example.com/auth/1',
+      }
+    end
+    let(:activity_json) do
+      {
+        '@context' => 'https://www.w3.org/ns/activitystreams',
+        'type' => 'Add',
+        'actor' => collection.account.uri,
+        'target' => collection.uri,
+        'object' => featured_item_json,
+      }
+    end
+    let(:stubbed_service) do
+      instance_double(ActivityPub::ProcessFeaturedItemService, call: true)
+    end
+
+    before do
+      allow(ActivityPub::ProcessFeaturedItemService).to receive(:new).and_return(stubbed_service)
+    end
+
+    it 'determines the correct collection and calls the service' do
+      subject.perform
+
+      expect(stubbed_service).to have_received(:call).with(collection, featured_item_json)
+    end
+  end
 end

spec/models/collection_item_spec.rb

@@ -21,9 +21,13 @@ RSpec.describe CollectionItem do
 
       let(:remote_collection) { Fabricate.build(:collection, local: false) }
 
-      it { is_expected.to validate_absence_of(:approval_uri) }
-
       it { is_expected.to validate_presence_of(:uri) }
+
+      context 'when account is not present' do
+        subject { Fabricate.build(:collection_item, collection: remote_collection, account: nil) }
+
+        it { is_expected.to validate_presence_of(:approval_uri) }
+      end
     end
 
     context 'when account is not present' do

spec/services/activitypub/process_featured_item_service_spec.rb

@@ -0,0 +1,82 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe ActivityPub::ProcessFeaturedItemService do
+  subject { described_class.new }
+
+  let(:collection) { Fabricate(:remote_collection, uri: 'https://other.example.com/collection/1') }
+  let(:featured_item_json) do
+    {
+      '@context' => 'https://www.w3.org/ns/activitystreams',
+      'id' => 'https://other.example.com/featured_item/1',
+      'type' => 'FeaturedItem',
+      'featuredObject' => 'https://example.com/actor/1',
+      'featuredObjectType' => 'Person',
+      'featureAuthorization' => 'https://example.com/auth/1',
+    }
+  end
+  let(:stubbed_service) do
+    instance_double(ActivityPub::VerifyFeaturedItemService, call: true)
+  end
+
+  before do
+    allow(ActivityPub::VerifyFeaturedItemService).to receive(:new).and_return(stubbed_service)
+  end
+
+  shared_examples 'non-matching URIs' do
+    context "when the item's URI does not match the collection's" do
+      let(:collection) { Fabricate(:remote_collection) }
+
+      it 'does not create a collection item and returns `nil`' do
+        expect do
+          expect(subject.call(collection, object)).to be_nil
+        end.to_not change(CollectionItem, :count)
+      end
+    end
+  end
+
+  context 'when the collection item is inlined' do
+    let(:object) { featured_item_json }
+
+    it_behaves_like 'non-matching URIs'
+
+    it 'creates and verifies the item' do
+      expect { subject.call(collection, object) }.to change(collection.collection_items, :count).by(1)
+
+      expect(stubbed_service).to have_received(:call)
+
+      new_item = collection.collection_items.last
+      expect(new_item.object_uri).to eq 'https://example.com/actor/1'
+      expect(new_item.approval_uri).to eq 'https://example.com/auth/1'
+    end
+  end
+
+  context 'when only the id of the collection item is given' do
+    let(:object) { featured_item_json['id'] }
+    let(:featured_item_request) do
+      stub_request(:get, object)
+        .to_return_json(
+          status: 200,
+          body: featured_item_json,
+          headers: { 'Content-Type' => 'application/activity+json' }
+        )
+    end
+
+    before do
+      featured_item_request
+    end
+
+    it_behaves_like 'non-matching URIs'
+
+    it 'fetches the collection item' do
+      expect { subject.call(collection, object) }.to change(collection.collection_items, :count).by(1)
+
+      expect(featured_item_request).to have_been_requested
+
+      new_item = collection.collection_items.last
+      expect(new_item.object_uri).to eq 'https://example.com/actor/1'
+      expect(new_item.approval_uri).to eq 'https://example.com/auth/1'
+    end
+  end
+end

spec/services/activitypub/verify_featured_item_service_spec.rb

@@ -0,0 +1,86 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe ActivityPub::VerifyFeaturedItemService do
+  subject { described_class.new }
+
+  let(:collection) { Fabricate(:remote_collection) }
+  let(:collection_item) do
+    Fabricate(:collection_item,
+              collection:,
+              account: nil,
+              state: :pending,
+              uri: 'https://other.example.com/items/1',
+              object_uri: 'https://example.com/actor/1',
+              approval_uri: verification_json['id'])
+  end
+  let(:verification_json) do
+    {
+      '@context' => 'https://www.w3.org/ns/activitystreams',
+      'type' => 'FeatureAuthorization',
+      'id' => 'https://example.com/auth/1',
+      'interactionTarget' => 'https://example.com/actor/1',
+      'interactingObject' => collection.uri,
+    }
+  end
+  let(:verification_request) do
+    stub_request(:get, 'https://example.com/auth/1')
+      .to_return_json(
+        status: 200,
+        body: verification_json,
+        headers: { 'Content-Type' => 'application/activity+json' }
+      )
+  end
+  let(:featured_account) { Fabricate(:remote_account, uri: 'https://example.com/actor/1') }
+
+  before { verification_request }
+
+  context 'when the authorization can be verified' do
+    context 'when the featured account is known' do
+      before { featured_account }
+
+      it 'verifies and creates the item' do
+        subject.call(collection_item)
+
+        expect(verification_request).to have_been_requested
+
+        expect(collection_item.account_id).to eq featured_account.id
+        expect(collection_item).to be_accepted
+      end
+    end
+
+    context 'when the featured account is not known' do
+      let(:stubbed_service) { instance_double(ActivityPub::FetchRemoteAccountService) }
+
+      before do
+        allow(stubbed_service).to receive(:call).with('https://example.com/actor/1') { featured_account }
+        allow(ActivityPub::FetchRemoteAccountService).to receive(:new).and_return(stubbed_service)
+      end
+
+      it 'fetches the actor and creates the item' do
+        subject.call(collection_item)
+
+        expect(stubbed_service).to have_received(:call)
+        expect(verification_request).to have_been_requested
+
+        expect(collection_item.account_id).to eq featured_account.id
+        expect(collection_item).to be_accepted
+      end
+    end
+  end
+
+  context 'when the authorization cannot be verified' do
+    let(:verification_request) do
+      stub_request(:get, 'https://example.com/auth/1')
+        .to_return(status: 404)
+    end
+
+    it 'creates item without attached account and in proper state' do
+      subject.call(collection_item)
+
+      expect(collection_item.account_id).to be_nil
+      expect(collection_item).to be_rejected
+    end
+  end
+end

spec/workers/activitypub/verify_featured_item_worker_spec.rb

@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe ActivityPub::VerifyFeaturedItemWorker do
+  let(:worker) { described_class.new }
+  let(:service) { instance_double(ActivityPub::VerifyFeaturedItemService, call: true) }
+
+  describe '#perform' do
+    let(:collection_item) { Fabricate(:unverified_remote_collection_item) }
+
+    before { stub_service }
+
+    it 'sends the status to the service' do
+      worker.perform(collection_item.id)
+
+      expect(service).to have_received(:call).with(collection_item)
+    end
+
+    it 'returns nil for non-existent record' do
+      result = worker.perform(123_123_123)
+
+      expect(result).to be_nil
+    end
+  end
+
+  def stub_service
+    allow(ActivityPub::VerifyFeaturedItemService)
+      .to receive(:new)
+      .and_return(service)
+  end
+end