From dd7ef0dc41584089a97444d8192bc61505108e6c Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Tue, 8 Aug 2017 21:52:15 +0200
Subject: Add ActivityPub inbox (#4216)

* Add ActivityPub inbox

* Handle ActivityPub deletes

* Handle ActivityPub creates

* Handle ActivityPub announces

* Stubs for handling all activities that need to be handled

* Add ActivityPub actor resolving

* Handle conversation URI passing in ActivityPub

* Handle content language in ActivityPub

* Send accept header when fetching actor, handle JSON parse errors

* Test for ActivityPub::FetchRemoteAccountService

* Handle public key and icon/image when embedded/as array/as resolvable URI

* Implement ActivityPub::FetchRemoteStatusService

* Add stubs for more interactions

* Undo activities implemented

* Handle out of order activities

* Hook up ActivityPub to ResolveRemoteAccountService, handle
Update Account activities

* Add fragment IDs to all transient activity serializers

* Add tests and fixes

* Add stubs for missing tests

* Add more tests

* Add more tests
---
 app/workers/activitypub/processing_worker.rb     | 11 +++++++++++
 app/workers/activitypub/thread_resolve_worker.rb | 17 +++++++++++++++++
 2 files changed, 28 insertions(+)
 create mode 100644 app/workers/activitypub/processing_worker.rb
 create mode 100644 app/workers/activitypub/thread_resolve_worker.rb

(limited to 'app/workers')

diff --git a/app/workers/activitypub/processing_worker.rb b/app/workers/activitypub/processing_worker.rb
new file mode 100644
index 000000000..7656ab56a
--- /dev/null
+++ b/app/workers/activitypub/processing_worker.rb
@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+class ActivityPub::ProcessingWorker
+  include Sidekiq::Worker
+
+  sidekiq_options backtrace: true
+
+  def perform(account_id, body)
+    ProcessCollectionService.new.call(body, Account.find(account_id))
+  end
+end
diff --git a/app/workers/activitypub/thread_resolve_worker.rb b/app/workers/activitypub/thread_resolve_worker.rb
new file mode 100644
index 000000000..4ef762d06
--- /dev/null
+++ b/app/workers/activitypub/thread_resolve_worker.rb
@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+class ActivityPub::ThreadResolveWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'pull', retry: false
+
+  def perform(child_status_id, parent_uri)
+    child_status  = Status.find(child_status_id)
+    parent_status = ActivityPub::FetchRemoteStatusService.new.call(parent_uri)
+
+    return if parent_status.nil?
+
+    child_status.thread = parent_status
+    child_status.save!
+  end
+end
-- 
cgit 


From 3d47154c206fcea8fc46e024ee11a40063a05023 Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Sat, 12 Aug 2017 02:54:54 +0200
Subject: Only PuSH-resubscribe to OStatus accounts (#4583)

---
 app/workers/scheduler/subscriptions_scheduler.rb | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

(limited to 'app/workers')

diff --git a/app/workers/scheduler/subscriptions_scheduler.rb b/app/workers/scheduler/subscriptions_scheduler.rb
index 7bfd002f4..5ddfaed18 100644
--- a/app/workers/scheduler/subscriptions_scheduler.rb
+++ b/app/workers/scheduler/subscriptions_scheduler.rb
@@ -8,13 +8,12 @@ class Scheduler::SubscriptionsScheduler
 
   def perform
     logger.info 'Queueing PuSH re-subscriptions'
-
     Pubsubhubbub::SubscribeWorker.push_bulk(expiring_accounts.pluck(:id))
   end
 
   private
 
   def expiring_accounts
-    Account.expiring(1.day.from_now).partitioned
+    Account.where(protocol: :ostatus).expiring(1.day.from_now).partitioned
   end
 end
-- 
cgit 


From b7370ac8baa643d93ea727699b3b11f9d3a55bea Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Sun, 13 Aug 2017 00:44:41 +0200
Subject: ActivityPub delivery (#4566)

* Deliver ActivityPub Like

* Deliver ActivityPub Undo-Like

* Deliver ActivityPub Create/Announce activities

* Deliver ActivityPub creates from mentions

* Deliver ActivityPub Block/Undo-Block

* Deliver ActivityPub Accept/Reject-Follow

* Deliver ActivityPub Undo-Follow

* Deliver ActivityPub Follow

* Deliver ActivityPub Delete activities

Incidentally fix #889

* Adjust BatchedRemoveStatusService for ActivityPub

* Add tests for ActivityPub workers

* Add tests for FollowService

* Add tests for FavouriteService, UnfollowService and PostStatusService

* Add tests for ReblogService, BlockService, UnblockService, ProcessMentionsService

* Add tests for AuthorizeFollowService, RejectFollowService, RemoveStatusService

* Add tests for BatchedRemoveStatusService

* Deliver updates to a local account to ActivityPub followers

* Minor adjustments
---
 .../api/v1/accounts/credentials_controller.rb      |  3 +-
 app/controllers/settings/profiles_controller.rb    |  1 +
 app/lib/activitypub/activity.rb                    |  2 +-
 app/models/account.rb                              |  4 ++
 app/services/authorize_follow_service.rb           | 19 ++++++-
 app/services/batched_remove_status_service.rb      | 43 ++++++++++++--
 app/services/block_service.rb                      | 19 ++++++-
 app/services/favourite_service.rb                  | 28 ++++++---
 app/services/follow_service.rb                     | 14 ++++-
 app/services/post_status_service.rb                |  1 +
 app/services/process_mentions_service.rb           | 28 ++++++---
 app/services/reblog_service.rb                     | 28 +++++++--
 app/services/reject_follow_service.rb              | 19 ++++++-
 app/services/remove_status_service.rb              | 49 +++++++++++++---
 app/services/unblock_service.rb                    | 19 ++++++-
 app/services/unfavourite_service.rb                | 22 +++++++-
 app/services/unfollow_service.rb                   | 19 ++++++-
 app/workers/activitypub/delivery_worker.rb         | 37 ++++++++++++
 app/workers/activitypub/distribution_worker.rb     | 38 +++++++++++++
 app/workers/activitypub/processing_worker.rb       |  2 +-
 .../activitypub/update_distribution_worker.rb      | 31 ++++++++++
 .../api/v1/accounts/credentials_controller_spec.rb |  6 ++
 .../settings/profiles_controller_spec.rb           |  2 +
 spec/services/authorize_follow_service_spec.rb     | 24 +++++++-
 .../services/batched_remove_status_service_spec.rb |  7 +++
 spec/services/block_service_spec.rb                | 19 ++++++-
 spec/services/favourite_service_spec.rb            | 22 +++++++-
 spec/services/follow_service_spec.rb               | 25 ++++++--
 spec/services/post_status_service_spec.rb          |  8 ++-
 spec/services/process_mentions_service_spec.rb     | 46 +++++++++++----
 spec/services/reblog_service_spec.rb               | 49 ++++++++++++----
 spec/services/reject_follow_service_spec.rb        | 24 +++++++-
 spec/services/remove_status_service_spec.rb        |  8 +++
 .../resolve_remote_account_service_spec.rb         | 66 ++++++++++++----------
 spec/services/unblock_service_spec.rb              | 22 +++++++-
 spec/services/unfollow_service_spec.rb             | 22 +++++++-
 spec/workers/activitypub/delivery_worker_spec.rb   | 23 ++++++++
 .../activitypub/distribution_worker_spec.rb        | 48 ++++++++++++++++
 spec/workers/activitypub/processing_worker_spec.rb | 15 +++++
 .../activitypub/thread_resolve_worker_spec.rb      | 16 ++++++
 .../activitypub/update_distribution_worker_spec.rb | 20 +++++++
 41 files changed, 785 insertions(+), 113 deletions(-)
 create mode 100644 app/workers/activitypub/delivery_worker.rb
 create mode 100644 app/workers/activitypub/distribution_worker.rb
 create mode 100644 app/workers/activitypub/update_distribution_worker.rb
 create mode 100644 spec/workers/activitypub/delivery_worker_spec.rb
 create mode 100644 spec/workers/activitypub/distribution_worker_spec.rb
 create mode 100644 spec/workers/activitypub/processing_worker_spec.rb
 create mode 100644 spec/workers/activitypub/thread_resolve_worker_spec.rb
 create mode 100644 spec/workers/activitypub/update_distribution_worker_spec.rb

(limited to 'app/workers')

diff --git a/app/controllers/api/v1/accounts/credentials_controller.rb b/app/controllers/api/v1/accounts/credentials_controller.rb
index 073808532..90a580c33 100644
--- a/app/controllers/api/v1/accounts/credentials_controller.rb
+++ b/app/controllers/api/v1/accounts/credentials_controller.rb
@@ -10,8 +10,9 @@ class Api::V1::Accounts::CredentialsController < Api::BaseController
   end
 
   def update
-    current_account.update!(account_params)
     @account = current_account
+    @account.update!(account_params)
+    ActivityPub::UpdateDistributionWorker.perform_async(@account.id)
     render json: @account, serializer: REST::CredentialAccountSerializer
   end
 
diff --git a/app/controllers/settings/profiles_controller.rb b/app/controllers/settings/profiles_controller.rb
index 0367e3593..c751c64ae 100644
--- a/app/controllers/settings/profiles_controller.rb
+++ b/app/controllers/settings/profiles_controller.rb
@@ -15,6 +15,7 @@ class Settings::ProfilesController < ApplicationController
 
   def update
     if @account.update(account_params)
+      ActivityPub::UpdateDistributionWorker.perform_async(@account.id)
       redirect_to settings_profile_path, notice: I18n.t('generic.changes_saved_msg')
     else
       render :show
diff --git a/app/lib/activitypub/activity.rb b/app/lib/activitypub/activity.rb
index 5debe023a..f8de8060c 100644
--- a/app/lib/activitypub/activity.rb
+++ b/app/lib/activitypub/activity.rb
@@ -93,7 +93,7 @@ class ActivityPub::Activity
   end
 
   def distribute_to_followers(status)
-    DistributionWorker.perform_async(status.id)
+    ::DistributionWorker.perform_async(status.id)
   end
 
   def delete_arrived_first?(uri)
diff --git a/app/models/account.rb b/app/models/account.rb
index 163bd1c0e..a7264353e 100644
--- a/app/models/account.rb
+++ b/app/models/account.rb
@@ -171,6 +171,10 @@ class Account < ApplicationRecord
       reorder(nil).pluck('distinct accounts.domain')
     end
 
+    def inboxes
+      reorder(nil).where(protocol: :activitypub).pluck("distinct coalesce(nullif(accounts.shared_inbox_url, ''), accounts.inbox_url)")
+    end
+
     def triadic_closures(account, limit: 5, offset: 0)
       sql = <<-SQL.squish
         WITH first_degree AS (
diff --git a/app/services/authorize_follow_service.rb b/app/services/authorize_follow_service.rb
index 41815a393..db35b6030 100644
--- a/app/services/authorize_follow_service.rb
+++ b/app/services/authorize_follow_service.rb
@@ -4,11 +4,28 @@ class AuthorizeFollowService < BaseService
   def call(source_account, target_account)
     follow_request = FollowRequest.find_by!(account: source_account, target_account: target_account)
     follow_request.authorize!
-    NotificationWorker.perform_async(build_xml(follow_request), target_account.id, source_account.id) unless source_account.local?
+    create_notification(follow_request) unless source_account.local?
+    follow_request
   end
 
   private
 
+  def create_notification(follow_request)
+    if follow_request.account.ostatus?
+      NotificationWorker.perform_async(build_xml(follow_request), follow_request.target_account_id, follow_request.account_id)
+    elsif follow_request.account.activitypub?
+      ActivityPub::DeliveryWorker.perform_async(build_json(follow_request), follow_request.target_account_id, follow_request.account.inbox_url)
+    end
+  end
+
+  def build_json(follow_request)
+    ActiveModelSerializers::SerializableResource.new(
+      follow_request,
+      serializer: ActivityPub::AcceptFollowSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+
   def build_xml(follow_request)
     OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.authorize_follow_request_salmon(follow_request))
   end
diff --git a/app/services/batched_remove_status_service.rb b/app/services/batched_remove_status_service.rb
index ab810c628..e6c8c9208 100644
--- a/app/services/batched_remove_status_service.rb
+++ b/app/services/batched_remove_status_service.rb
@@ -15,9 +15,11 @@ class BatchedRemoveStatusService < BaseService
     @mentions = statuses.map { |s| [s.id, s.mentions.includes(:account).to_a] }.to_h
     @tags     = statuses.map { |s| [s.id, s.tags.pluck(:name)] }.to_h
 
-    @stream_entry_batches = []
-    @salmon_batches       = []
-    @json_payloads        = statuses.map { |s| [s.id, Oj.dump(event: :delete, payload: s.id)] }.to_h
+    @stream_entry_batches  = []
+    @salmon_batches        = []
+    @activity_json_batches = []
+    @json_payloads         = statuses.map { |s| [s.id, Oj.dump(event: :delete, payload: s.id)] }.to_h
+    @activity_json         = {}
 
     # Ensure that rendered XML reflects destroyed state
     Status.where(id: statuses.map(&:id)).in_batches.destroy_all
@@ -27,7 +29,11 @@ class BatchedRemoveStatusService < BaseService
       account = account_statuses.first.account
 
       unpush_from_home_timelines(account_statuses)
-      batch_stream_entries(account_statuses) if account.local?
+
+      if account.local?
+        batch_stream_entries(account_statuses)
+        batch_activity_json(account, account_statuses)
+      end
     end
 
     # Cannot be batched
@@ -38,6 +44,7 @@ class BatchedRemoveStatusService < BaseService
 
     Pubsubhubbub::DistributionWorker.push_bulk(@stream_entry_batches) { |batch| batch }
     NotificationWorker.push_bulk(@salmon_batches) { |batch| batch }
+    ActivityPub::DeliveryWorker.push_bulk(@activity_json_batches) { |batch| batch }
   end
 
   private
@@ -50,6 +57,22 @@ class BatchedRemoveStatusService < BaseService
     end
   end
 
+  def batch_activity_json(account, statuses)
+    account.followers.inboxes.each do |inbox_url|
+      statuses.each do |status|
+        @activity_json_batches << [build_json(status), account.id, inbox_url]
+      end
+    end
+
+    statuses.each do |status|
+      other_recipients = (status.mentions + status.reblogs).map(&:account).reject(&:local?).select(&:activitypub?).uniq(&:id)
+
+      other_recipients.each do |target_account|
+        @activity_json_batches << [build_json(status), account.id, target_account.inbox_url]
+      end
+    end
+  end
+
   def unpush_from_home_timelines(statuses)
     account    = statuses.first.account
     recipients = account.followers.local.pluck(:id)
@@ -79,7 +102,7 @@ class BatchedRemoveStatusService < BaseService
     return if @mentions[status.id].empty?
 
     payload    = stream_entry_to_xml(status.stream_entry.reload)
-    recipients = @mentions[status.id].map(&:account).reject(&:local?).uniq(&:domain).map(&:id)
+    recipients = @mentions[status.id].map(&:account).reject(&:local?).select(&:ostatus?).uniq(&:domain).map(&:id)
 
     recipients.each do |recipient_id|
       @salmon_batches << [payload, status.account_id, recipient_id]
@@ -111,4 +134,14 @@ class BatchedRemoveStatusService < BaseService
   def redis
     Redis.current
   end
+
+  def build_json(status)
+    return @activity_json[status.id] if @activity_json.key?(status.id)
+
+    @activity_json[status.id] = ActiveModelSerializers::SerializableResource.new(
+      status,
+      serializer: ActivityPub::DeleteSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
 end
diff --git a/app/services/block_service.rb b/app/services/block_service.rb
index 5d7bf6a3b..f2253226b 100644
--- a/app/services/block_service.rb
+++ b/app/services/block_service.rb
@@ -12,11 +12,28 @@ class BlockService < BaseService
     block = account.block!(target_account)
 
     BlockWorker.perform_async(account.id, target_account.id)
-    NotificationWorker.perform_async(build_xml(block), account.id, target_account.id) unless target_account.local?
+    create_notification(block) unless target_account.local?
+    block
   end
 
   private
 
+  def create_notification(block)
+    if block.target_account.ostatus?
+      NotificationWorker.perform_async(build_xml(block), block.account_id, block.target_account_id)
+    elsif block.target_account.activitypub?
+      ActivityPub::DeliveryWorker.perform_async(build_json(block), block.account_id, block.target_account.inbox_url)
+    end
+  end
+
+  def build_json(block)
+    ActiveModelSerializers::SerializableResource.new(
+      block,
+      serializer: ActivityPub::BlockSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+
   def build_xml(block)
     OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.block_salmon(block))
   end
diff --git a/app/services/favourite_service.rb b/app/services/favourite_service.rb
index 291f9e56e..4aa935170 100644
--- a/app/services/favourite_service.rb
+++ b/app/services/favourite_service.rb
@@ -15,18 +15,32 @@ class FavouriteService < BaseService
     return favourite unless favourite.nil?
 
     favourite = Favourite.create!(account: account, status: status)
-
-    if status.local?
-      NotifyService.new.call(favourite.status.account, favourite)
-    else
-      NotificationWorker.perform_async(build_xml(favourite), account.id, status.account_id)
-    end
-
+    create_notification(favourite)
     favourite
   end
 
   private
 
+  def create_notification(favourite)
+    status = favourite.status
+
+    if status.account.local?
+      NotifyService.new.call(status.account, favourite)
+    elsif status.account.ostatus?
+      NotificationWorker.perform_async(build_xml(favourite), favourite.account_id, status.account_id)
+    elsif status.account.activitypub?
+      ActivityPub::DeliveryWorker.perform_async(build_json(favourite), favourite.account_id, status.account.inbox_url)
+    end
+  end
+
+  def build_json(favourite)
+    ActiveModelSerializers::SerializableResource.new(
+      favourite,
+      serializer: ActivityPub::LikeSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+
   def build_xml(favourite)
     OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.favourite_salmon(favourite))
   end
diff --git a/app/services/follow_service.rb b/app/services/follow_service.rb
index 3155feaa4..2be625cd8 100644
--- a/app/services/follow_service.rb
+++ b/app/services/follow_service.rb
@@ -14,7 +14,7 @@ class FollowService < BaseService
 
     return if source_account.following?(target_account)
 
-    if target_account.locked?
+    if target_account.locked? || target_account.activitypub?
       request_follow(source_account, target_account)
     else
       direct_follow(source_account, target_account)
@@ -28,9 +28,11 @@ class FollowService < BaseService
 
     if target_account.local?
       NotifyService.new.call(target_account, follow_request)
-    else
+    elsif target_account.ostatus?
       NotificationWorker.perform_async(build_follow_request_xml(follow_request), source_account.id, target_account.id)
       AfterRemoteFollowRequestWorker.perform_async(follow_request.id)
+    elsif target_account.activitypub?
+      ActivityPub::DeliveryWorker.perform_async(build_json(follow_request), source_account.id, target_account.inbox_url)
     end
 
     follow_request
@@ -63,4 +65,12 @@ class FollowService < BaseService
   def build_follow_xml(follow)
     OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.follow_salmon(follow))
   end
+
+  def build_json(follow_request)
+    ActiveModelSerializers::SerializableResource.new(
+      follow_request,
+      serializer: ActivityPub::FollowSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
 end
diff --git a/app/services/post_status_service.rb b/app/services/post_status_service.rb
index 951a38e19..5ff93f21e 100644
--- a/app/services/post_status_service.rb
+++ b/app/services/post_status_service.rb
@@ -39,6 +39,7 @@ class PostStatusService < BaseService
     LinkCrawlWorker.perform_async(status.id) unless status.spoiler_text?
     DistributionWorker.perform_async(status.id)
     Pubsubhubbub::DistributionWorker.perform_async(status.stream_entry.id)
+    ActivityPub::DistributionWorker.perform_async(status.id)
 
     if options[:idempotency].present?
       redis.setex("idempotency:status:#{account.id}:#{options[:idempotency]}", 3_600, status.id)
diff --git a/app/services/process_mentions_service.rb b/app/services/process_mentions_service.rb
index 438033d22..407fa8c18 100644
--- a/app/services/process_mentions_service.rb
+++ b/app/services/process_mentions_service.rb
@@ -28,18 +28,32 @@ class ProcessMentionsService < BaseService
     end
 
     status.mentions.includes(:account).each do |mention|
-      mentioned_account = mention.account
-
-      if mentioned_account.local?
-        NotifyService.new.call(mentioned_account, mention)
-      else
-        NotificationWorker.perform_async(stream_entry_to_xml(status.stream_entry), status.account_id, mentioned_account.id)
-      end
+      create_notification(status, mention)
     end
   end
 
   private
 
+  def create_notification(status, mention)
+    mentioned_account = mention.account
+
+    if mentioned_account.local?
+      NotifyService.new.call(mentioned_account, mention)
+    elsif mentioned_account.ostatus?
+      NotificationWorker.perform_async(stream_entry_to_xml(status.stream_entry), status.account_id, mentioned_account.id)
+    elsif mentioned_account.activitypub?
+      ActivityPub::DeliveryWorker.perform_async(build_json(mention.status), mention.status.account_id, mentioned_account.inbox_url)
+    end
+  end
+
+  def build_json(status)
+    ActiveModelSerializers::SerializableResource.new(
+      status,
+      serializer: ActivityPub::ActivitySerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+
   def follow_remote_account_service
     @follow_remote_account_service ||= ResolveRemoteAccountService.new
   end
diff --git a/app/services/reblog_service.rb b/app/services/reblog_service.rb
index ba24b1f9d..7f886af7c 100644
--- a/app/services/reblog_service.rb
+++ b/app/services/reblog_service.rb
@@ -21,13 +21,31 @@ class ReblogService < BaseService
 
     DistributionWorker.perform_async(reblog.id)
     Pubsubhubbub::DistributionWorker.perform_async(reblog.stream_entry.id)
+    ActivityPub::DistributionWorker.perform_async(reblog.id)
 
-    if reblogged_status.local?
-      NotifyService.new.call(reblog.reblog.account, reblog)
-    else
-      NotificationWorker.perform_async(stream_entry_to_xml(reblog.stream_entry), account.id, reblog.reblog.account_id)
+    create_notification(reblog)
+    reblog
+  end
+
+  private
+
+  def create_notification(reblog)
+    reblogged_status = reblog.reblog
+
+    if reblogged_status.account.local?
+      NotifyService.new.call(reblogged_status.account, reblog)
+    elsif reblogged_status.account.ostatus?
+      NotificationWorker.perform_async(stream_entry_to_xml(reblog.stream_entry), reblog.account_id, reblogged_status.account_id)
+    elsif reblogged_status.account.activitypub?
+      ActivityPub::DeliveryWorker.perform_async(build_json(reblog), reblog.account_id, reblogged_status.account.inbox_url)
     end
+  end
 
-    reblog
+  def build_json(reblog)
+    ActiveModelSerializers::SerializableResource.new(
+      reblog,
+      serializer: ActivityPub::ActivitySerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
   end
 end
diff --git a/app/services/reject_follow_service.rb b/app/services/reject_follow_service.rb
index fd7e66c23..a91266aa4 100644
--- a/app/services/reject_follow_service.rb
+++ b/app/services/reject_follow_service.rb
@@ -4,11 +4,28 @@ class RejectFollowService < BaseService
   def call(source_account, target_account)
     follow_request = FollowRequest.find_by!(account: source_account, target_account: target_account)
     follow_request.reject!
-    NotificationWorker.perform_async(build_xml(follow_request), target_account.id, source_account.id) unless source_account.local?
+    create_notification(follow_request) unless source_account.local?
+    follow_request
   end
 
   private
 
+  def create_notification(follow_request)
+    if follow_request.account.ostatus?
+      NotificationWorker.perform_async(build_xml(follow_request), follow_request.target_account_id, follow_request.account_id)
+    elsif follow_request.account.activitypub?
+      ActivityPub::DeliveryWorker.perform_async(build_json(follow_request), follow_request.target_account_id, follow_request.account.inbox_url)
+    end
+  end
+
+  def build_json(follow_request)
+    ActiveModelSerializers::SerializableResource.new(
+      follow_request,
+      serializer: ActivityPub::RejectFollowSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+
   def build_xml(follow_request)
     OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.reject_follow_request_salmon(follow_request))
   end
diff --git a/app/services/remove_status_service.rb b/app/services/remove_status_service.rb
index a5281f586..fcccbaa24 100644
--- a/app/services/remove_status_service.rb
+++ b/app/services/remove_status_service.rb
@@ -22,8 +22,10 @@ class RemoveStatusService < BaseService
 
     return unless @account.local?
 
-    remove_from_mentioned(@stream_entry.reload)
-    Pubsubhubbub::DistributionWorker.perform_async(@stream_entry.id)
+    @stream_entry = @stream_entry.reload
+
+    remove_from_remote_followers
+    remove_from_remote_affected
   end
 
   private
@@ -38,15 +40,48 @@ class RemoveStatusService < BaseService
     end
   end
 
-  def remove_from_mentioned(stream_entry)
-    salmon_xml       = stream_entry_to_xml(stream_entry)
-    target_accounts  = @mentions.map(&:account).reject(&:local?).uniq(&:domain)
+  def remove_from_remote_affected
+    # People who got mentioned in the status, or who
+    # reblogged it from someone else might not follow
+    # the author and wouldn't normally receive the
+    # delete notification - so here, we explicitly
+    # send it to them
+
+    target_accounts = (@mentions.map(&:account).reject(&:local?) + @reblogs.map(&:account).reject(&:local?)).uniq(&:id)
+
+    # Ostatus
+    NotificationWorker.push_bulk(target_accounts.select(&:ostatus?).uniq(&:domain)) do |target_account|
+      [salmon_xml, @account.id, target_account.id]
+    end
+
+    # ActivityPub
+    ActivityPub::DeliveryWorker.push_bulk(target_accounts.select(&:activitypub?).uniq(&:inbox_url)) do |inbox_url|
+      [activity_json, @account.id, inbox_url]
+    end
+  end
+
+  def remove_from_remote_followers
+    # OStatus
+    Pubsubhubbub::DistributionWorker.perform_async(@stream_entry.id)
 
-    NotificationWorker.push_bulk(target_accounts) do |target_account|
-      [salmon_xml, stream_entry.account_id, target_account.id]
+    # ActivityPub
+    ActivityPub::DeliveryWorker.push_bulk(@account.followers.inboxes) do |inbox_url|
+      [activity_json, @account.id, inbox_url]
     end
   end
 
+  def salmon_xml
+    @salmon_xml ||= stream_entry_to_xml(@stream_entry)
+  end
+
+  def activity_json
+    @activity_json ||= ActiveModelSerializers::SerializableResource.new(
+      @status,
+      serializer: ActivityPub::DeleteSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+
   def remove_reblogs
     # We delete reblogs of the status before the original status,
     # because once original status is gone, reblogs will disappear
diff --git a/app/services/unblock_service.rb b/app/services/unblock_service.rb
index ff15c7275..72fc5ab15 100644
--- a/app/services/unblock_service.rb
+++ b/app/services/unblock_service.rb
@@ -5,11 +5,28 @@ class UnblockService < BaseService
     return unless account.blocking?(target_account)
 
     unblock = account.unblock!(target_account)
-    NotificationWorker.perform_async(build_xml(unblock), account.id, target_account.id) unless target_account.local?
+    create_notification(unblock) unless target_account.local?
+    unblock
   end
 
   private
 
+  def create_notification(unblock)
+    if unblock.target_account.ostatus?
+      NotificationWorker.perform_async(build_xml(unblock), unblock.account_id, unblock.target_account_id)
+    elsif unblock.target_account.activitypub?
+      ActivityPub::DeliveryWorker.perform_async(build_json(unblock), unblock.account_id, unblock.target_account.inbox_url)
+    end
+  end
+
+  def build_json(unblock)
+    ActiveModelSerializers::SerializableResource.new(
+      unblock,
+      serializer: ActivityPub::UndoBlockSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+
   def build_xml(block)
     OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.unblock_salmon(block))
   end
diff --git a/app/services/unfavourite_service.rb b/app/services/unfavourite_service.rb
index 564aaee46..e53798e66 100644
--- a/app/services/unfavourite_service.rb
+++ b/app/services/unfavourite_service.rb
@@ -4,14 +4,30 @@ class UnfavouriteService < BaseService
   def call(account, status)
     favourite = Favourite.find_by!(account: account, status: status)
     favourite.destroy!
-
-    NotificationWorker.perform_async(build_xml(favourite), account.id, status.account_id) unless status.local?
-
+    create_notification(favourite) unless status.local?
     favourite
   end
 
   private
 
+  def create_notification(favourite)
+    status = favourite.status
+
+    if status.account.ostatus?
+      NotificationWorker.perform_async(build_xml(favourite), favourite.account_id, status.account_id)
+    elsif status.account.activitypub?
+      ActivityPub::DeliveryWorker.perform_async(build_json(favourite), favourite.account_id, status.account.inbox_url)
+    end
+  end
+
+  def build_json(favourite)
+    ActiveModelSerializers::SerializableResource.new(
+      favourite,
+      serializer: ActivityPub::UndoLikeSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+
   def build_xml(favourite)
     OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.unfavourite_salmon(favourite))
   end
diff --git a/app/services/unfollow_service.rb b/app/services/unfollow_service.rb
index 388909586..10af75146 100644
--- a/app/services/unfollow_service.rb
+++ b/app/services/unfollow_service.rb
@@ -7,12 +7,29 @@ class UnfollowService < BaseService
   def call(source_account, target_account)
     follow = source_account.unfollow!(target_account)
     return unless follow
-    NotificationWorker.perform_async(build_xml(follow), source_account.id, target_account.id) unless target_account.local?
+    create_notification(follow) unless target_account.local?
     UnmergeWorker.perform_async(target_account.id, source_account.id)
+    follow
   end
 
   private
 
+  def create_notification(follow)
+    if follow.target_account.ostatus?
+      NotificationWorker.perform_async(build_xml(follow), follow.account_id, follow.target_account_id)
+    elsif follow.target_account.activitypub?
+      ActivityPub::DeliveryWorker.perform_async(build_json(follow), follow.account_id, follow.target_account.inbox_url)
+    end
+  end
+
+  def build_json(follow)
+    ActiveModelSerializers::SerializableResource.new(
+      follow,
+      serializer: ActivityPub::UndoFollowSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+
   def build_xml(follow)
     OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.unfollow_salmon(follow))
   end
diff --git a/app/workers/activitypub/delivery_worker.rb b/app/workers/activitypub/delivery_worker.rb
new file mode 100644
index 000000000..cd67b6710
--- /dev/null
+++ b/app/workers/activitypub/delivery_worker.rb
@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+class ActivityPub::DeliveryWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'push', retry: 5, dead: false
+
+  HEADERS = { 'Content-Type' => 'application/activity+json' }.freeze
+
+  def perform(json, source_account_id, inbox_url)
+    @json           = json
+    @source_account = Account.find(source_account_id)
+    @inbox_url      = inbox_url
+
+    perform_request
+
+    raise Mastodon::UnexpectedResponseError, @response unless response_successful?
+  rescue => e
+    raise e.class, "Delivery failed for #{inbox_url}: #{e.message}"
+  end
+
+  private
+
+  def build_request
+    request = Request.new(:post, @inbox_url, body: @json)
+    request.on_behalf_of(@source_account, :uri)
+    request.add_headers(HEADERS)
+  end
+
+  def perform_request
+    @response = build_request.perform
+  end
+
+  def response_successful?
+    @response.code > 199 && @response.code < 300
+  end
+end
diff --git a/app/workers/activitypub/distribution_worker.rb b/app/workers/activitypub/distribution_worker.rb
new file mode 100644
index 000000000..004dd25d1
--- /dev/null
+++ b/app/workers/activitypub/distribution_worker.rb
@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+class ActivityPub::DistributionWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'push'
+
+  def perform(status_id)
+    @status  = Status.find(status_id)
+    @account = @status.account
+
+    return if skip_distribution?
+
+    ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
+      [payload, @account.id, inbox_url]
+    end
+  rescue ActiveRecord::RecordNotFound
+    true
+  end
+
+  private
+
+  def skip_distribution?
+    @status.direct_visibility?
+  end
+
+  def inboxes
+    @inboxes ||= @account.followers.inboxes
+  end
+
+  def payload
+    @payload ||= ActiveModelSerializers::SerializableResource.new(
+      @status,
+      serializer: ActivityPub::ActivitySerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+end
diff --git a/app/workers/activitypub/processing_worker.rb b/app/workers/activitypub/processing_worker.rb
index 7656ab56a..bb9adf64b 100644
--- a/app/workers/activitypub/processing_worker.rb
+++ b/app/workers/activitypub/processing_worker.rb
@@ -6,6 +6,6 @@ class ActivityPub::ProcessingWorker
   sidekiq_options backtrace: true
 
   def perform(account_id, body)
-    ProcessCollectionService.new.call(body, Account.find(account_id))
+    ActivityPub::ProcessCollectionService.new.call(body, Account.find(account_id))
   end
 end
diff --git a/app/workers/activitypub/update_distribution_worker.rb b/app/workers/activitypub/update_distribution_worker.rb
new file mode 100644
index 000000000..f3377dcec
--- /dev/null
+++ b/app/workers/activitypub/update_distribution_worker.rb
@@ -0,0 +1,31 @@
+# frozen_string_literal: true
+
+class ActivityPub::UpdateDistributionWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'push'
+
+  def perform(account_id)
+    @account = Account.find(account_id)
+
+    ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
+      [payload, @account.id, inbox_url]
+    end
+  rescue ActiveRecord::RecordNotFound
+    true
+  end
+
+  private
+
+  def inboxes
+    @inboxes ||= @account.followers.inboxes
+  end
+
+  def payload
+    @payload ||= ActiveModelSerializers::SerializableResource.new(
+      @account,
+      serializer: ActivityPub::UpdateSerializer,
+      adapter: ActivityPub::Adapter
+    ).to_json
+  end
+end
diff --git a/spec/controllers/api/v1/accounts/credentials_controller_spec.rb b/spec/controllers/api/v1/accounts/credentials_controller_spec.rb
index 4a3100348..bc89772b9 100644
--- a/spec/controllers/api/v1/accounts/credentials_controller_spec.rb
+++ b/spec/controllers/api/v1/accounts/credentials_controller_spec.rb
@@ -20,6 +20,8 @@ describe Api::V1::Accounts::CredentialsController do
   describe 'PATCH #update' do
     describe 'with valid data' do
       before do
+        allow(ActivityPub::UpdateDistributionWorker).to receive(:perform_async)
+
         patch :update, params: {
           display_name: "Alice Isn't Dead",
           note: "Hi!\n\nToot toot!",
@@ -40,6 +42,10 @@ describe Api::V1::Accounts::CredentialsController do
         expect(user.account.avatar).to exist
         expect(user.account.header).to exist
       end
+
+      it 'queues up an account update distribution' do
+        expect(ActivityPub::UpdateDistributionWorker).to have_received(:perform_async).with(user.account_id)
+      end
     end
 
     describe 'with invalid data' do
diff --git a/spec/controllers/settings/profiles_controller_spec.rb b/spec/controllers/settings/profiles_controller_spec.rb
index e502dbda7..ee3315be6 100644
--- a/spec/controllers/settings/profiles_controller_spec.rb
+++ b/spec/controllers/settings/profiles_controller_spec.rb
@@ -17,11 +17,13 @@ RSpec.describe Settings::ProfilesController, type: :controller do
 
   describe 'PUT #update' do
     it 'updates the user profile' do
+      allow(ActivityPub::UpdateDistributionWorker).to receive(:perform_async)
       account = Fabricate(:account, user: @user, display_name: 'Old name')
 
       put :update, params: { account: { display_name: 'New name' } }
       expect(account.reload.display_name).to eq 'New name'
       expect(response).to redirect_to(settings_profile_path)
+      expect(ActivityPub::UpdateDistributionWorker).to have_received(:perform_async).with(account.id)
     end
   end
 end
diff --git a/spec/services/authorize_follow_service_spec.rb b/spec/services/authorize_follow_service_spec.rb
index 3f3a2bc56..d74eb41a2 100644
--- a/spec/services/authorize_follow_service_spec.rb
+++ b/spec/services/authorize_follow_service_spec.rb
@@ -22,7 +22,7 @@ RSpec.describe AuthorizeFollowService do
     end
   end
 
-  describe 'remote' do
+  describe 'remote OStatus' do
     let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com')).account }
 
     before do
@@ -46,4 +46,26 @@ RSpec.describe AuthorizeFollowService do
       }).to have_been_made.once
     end
   end
+
+  describe 'remote ActivityPub' do
+    let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', domain: 'example.com', protocol: :activitypub, inbox_url: 'http://example.com/inbox')).account }
+
+    before do
+      FollowRequest.create(account: bob, target_account: sender)
+      stub_request(:post, bob.inbox_url).to_return(status: 200)
+      subject.call(bob, sender)
+    end
+
+    it 'removes follow request' do
+      expect(bob.requested?(sender)).to be false
+    end
+
+    it 'creates follow relation' do
+      expect(bob.following?(sender)).to be true
+    end
+
+    it 'sends an accept activity' do
+      expect(a_request(:post, bob.inbox_url)).to have_been_made.once
+    end
+  end
 end
diff --git a/spec/services/batched_remove_status_service_spec.rb b/spec/services/batched_remove_status_service_spec.rb
index c20085e25..2484d4b58 100644
--- a/spec/services/batched_remove_status_service_spec.rb
+++ b/spec/services/batched_remove_status_service_spec.rb
@@ -6,6 +6,7 @@ RSpec.describe BatchedRemoveStatusService do
   let!(:alice)  { Fabricate(:account) }
   let!(:bob)    { Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://example.com/salmon') }
   let!(:jeff)   { Fabricate(:account) }
+  let!(:hank)   { Fabricate(:account, username: 'hank', protocol: :activitypub, domain: 'example.com', inbox_url: 'http://example.com/inbox') }
 
   let(:status1) { PostStatusService.new.call(alice, 'Hello @bob@example.com') }
   let(:status2) { PostStatusService.new.call(alice, 'Another status') }
@@ -15,9 +16,11 @@ RSpec.describe BatchedRemoveStatusService do
 
     stub_request(:post, 'http://example.com/push').to_return(status: 200, body: '', headers: {})
     stub_request(:post, 'http://example.com/salmon').to_return(status: 200, body: '', headers: {})
+    stub_request(:post, 'http://example.com/inbox').to_return(status: 200)
 
     Fabricate(:subscription, account: alice, callback_url: 'http://example.com/push', confirmed: true, expires_at: 30.days.from_now)
     jeff.follow!(alice)
+    hank.follow!(alice)
 
     status1
     status2
@@ -58,4 +61,8 @@ RSpec.describe BatchedRemoveStatusService do
       xml.match(TagManager::VERBS[:delete])
     }).to have_been_made.once
   end
+
+  it 'sends delete activity to followers' do
+    expect(a_request(:post, 'http://example.com/inbox')).to have_been_made.at_least_once
+  end
 end
diff --git a/spec/services/block_service_spec.rb b/spec/services/block_service_spec.rb
index 2a54e032e..bd2ab3d53 100644
--- a/spec/services/block_service_spec.rb
+++ b/spec/services/block_service_spec.rb
@@ -17,7 +17,7 @@ RSpec.describe BlockService do
     end
   end
 
-  describe 'remote' do
+  describe 'remote OStatus' do
     let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com')).account }
 
     before do
@@ -36,4 +36,21 @@ RSpec.describe BlockService do
       }).to have_been_made.once
     end
   end
+
+  describe 'remote ActivityPub' do
+    let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', protocol: :activitypub, domain: 'example.com', inbox_url: 'http://example.com/inbox')).account }
+
+    before do
+      stub_request(:post, 'http://example.com/inbox').to_return(status: 200)
+      subject.call(sender, bob)
+    end
+
+    it 'creates a blocking relation' do
+      expect(sender.blocking?(bob)).to be true
+    end
+
+    it 'sends a block activity' do
+      expect(a_request(:post, 'http://example.com/inbox')).to have_been_made.once
+    end
+  end
 end
diff --git a/spec/services/favourite_service_spec.rb b/spec/services/favourite_service_spec.rb
index 36f1b64d4..2ab1f32ca 100644
--- a/spec/services/favourite_service_spec.rb
+++ b/spec/services/favourite_service_spec.rb
@@ -18,8 +18,8 @@ RSpec.describe FavouriteService do
     end
   end
 
-  describe 'remote' do
-    let(:bob)    { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com')).account }
+  describe 'remote OStatus' do
+    let(:bob)    { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', protocol: :ostatus, domain: 'example.com', salmon_url: 'http://salmon.example.com')).account }
     let(:status) { Fabricate(:status, account: bob, uri: 'tag:example.com:blahblah') }
 
     before do
@@ -38,4 +38,22 @@ RSpec.describe FavouriteService do
       }).to have_been_made.once
     end
   end
+
+  describe 'remote ActivityPub' do
+    let(:bob)    { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, protocol: :activitypub, username: 'bob', domain: 'example.com', inbox_url: 'http://example.com/inbox')).account }
+    let(:status) { Fabricate(:status, account: bob) }
+
+    before do
+      stub_request(:post, "http://example.com/inbox").to_return(:status => 200, :body => "", :headers => {})
+      subject.call(sender, status)
+    end
+
+    it 'creates a favourite' do
+      expect(status.favourites.first).to_not be_nil
+    end
+
+    it 'sends a like activity' do
+      expect(a_request(:post, "http://example.com/inbox")).to have_been_made.once
+    end
+  end
 end
diff --git a/spec/services/follow_service_spec.rb b/spec/services/follow_service_spec.rb
index 32dedb3ad..1e2378031 100644
--- a/spec/services/follow_service_spec.rb
+++ b/spec/services/follow_service_spec.rb
@@ -44,9 +44,9 @@ RSpec.describe FollowService do
     end
   end
 
-  context 'remote account' do
+  context 'remote OStatus account' do
     describe 'locked account' do
-      let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, locked: true, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com')).account }
+      let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, protocol: :ostatus, locked: true, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com')).account }
 
       before do
         stub_request(:post, "http://salmon.example.com/").to_return(:status => 200, :body => "", :headers => {})
@@ -66,7 +66,7 @@ RSpec.describe FollowService do
     end
 
     describe 'unlocked account' do
-      let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com', hub_url: 'http://hub.example.com')).account }
+      let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, protocol: :ostatus, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com', hub_url: 'http://hub.example.com')).account }
 
       before do
         stub_request(:post, "http://salmon.example.com/").to_return(:status => 200, :body => "", :headers => {})
@@ -91,7 +91,7 @@ RSpec.describe FollowService do
     end
 
     describe 'already followed account' do
-      let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com', hub_url: 'http://hub.example.com')).account }
+      let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, protocol: :ostatus, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com', hub_url: 'http://hub.example.com')).account }
 
       before do
         sender.follow!(bob)
@@ -111,4 +111,21 @@ RSpec.describe FollowService do
       end
     end
   end
+
+  context 'remote ActivityPub account' do
+    let(:bob) { Fabricate(:user, account: Fabricate(:account, username: 'bob', domain: 'example.com', protocol: :activitypub, inbox_url: 'http://example.com/inbox')).account }
+
+    before do
+      stub_request(:post, "http://example.com/inbox").to_return(:status => 200, :body => "", :headers => {})
+      subject.call(sender, bob.acct)
+    end
+
+    it 'creates follow request' do
+      expect(FollowRequest.find_by(account: sender, target_account: bob)).to_not be_nil
+    end
+
+    it 'sends a follow activity to the inbox' do
+      expect(a_request(:post, 'http://example.com/inbox')).to have_been_made.once
+    end
+  end
 end
diff --git a/spec/services/post_status_service_spec.rb b/spec/services/post_status_service_spec.rb
index 57876dcc2..4182c4e1f 100644
--- a/spec/services/post_status_service_spec.rb
+++ b/spec/services/post_status_service_spec.rb
@@ -100,16 +100,18 @@ RSpec.describe PostStatusService do
     expect(hashtags_service).to have_received(:call).with(status)
   end
 
-  it 'pings PuSH hubs' do
+  it 'gets distributed' do
     allow(DistributionWorker).to receive(:perform_async)
     allow(Pubsubhubbub::DistributionWorker).to receive(:perform_async)
+    allow(ActivityPub::DistributionWorker).to receive(:perform_async)
+
     account = Fabricate(:account)
 
     status = subject.call(account, "test status update")
 
     expect(DistributionWorker).to have_received(:perform_async).with(status.id)
-    expect(Pubsubhubbub::DistributionWorker).
-      to have_received(:perform_async).with(status.stream_entry.id)
+    expect(Pubsubhubbub::DistributionWorker).to have_received(:perform_async).with(status.stream_entry.id)
+    expect(ActivityPub::DistributionWorker).to have_received(:perform_async).with(status.id)
   end
 
   it 'crawls links' do
diff --git a/spec/services/process_mentions_service_spec.rb b/spec/services/process_mentions_service_spec.rb
index 984d13746..09f8fa45b 100644
--- a/spec/services/process_mentions_service_spec.rb
+++ b/spec/services/process_mentions_service_spec.rb
@@ -1,22 +1,44 @@
 require 'rails_helper'
 
 RSpec.describe ProcessMentionsService do
-  let(:account)     { Fabricate(:account, username: 'alice') }
-  let(:remote_user) { Fabricate(:account, username: 'remote_user', domain: 'example.com', salmon_url: 'http://salmon.example.com') }
-  let(:status)      { Fabricate(:status, account: account, text: "Hello @#{remote_user.acct}") }
+  let(:account) { Fabricate(:account, username: 'alice') }
+  let(:status)  { Fabricate(:status, account: account, text: "Hello @#{remote_user.acct}") }
 
-  subject { ProcessMentionsService.new }
+  context 'OStatus' do
+    let(:remote_user) { Fabricate(:account, username: 'remote_user', protocol: :ostatus, domain: 'example.com', salmon_url: 'http://salmon.example.com') }
 
-  before do
-    stub_request(:post, remote_user.salmon_url)
-    subject.(status)
-  end
+    subject { ProcessMentionsService.new }
+
+    before do
+      stub_request(:post, remote_user.salmon_url)
+      subject.call(status)
+    end
 
-  it 'creates a mention' do
-    expect(remote_user.mentions.where(status: status).count).to eq 1
+    it 'creates a mention' do
+      expect(remote_user.mentions.where(status: status).count).to eq 1
+    end
+
+    it 'posts to remote user\'s Salmon end point' do
+      expect(a_request(:post, remote_user.salmon_url)).to have_been_made.once
+    end
   end
 
-  it 'posts to remote user\'s Salmon end point' do
-    expect(a_request(:post, remote_user.salmon_url)).to have_been_made
+  context 'ActivityPub' do
+    let(:remote_user) { Fabricate(:account, username: 'remote_user', protocol: :activitypub, domain: 'example.com', inbox_url: 'http://example.com/inbox') }
+
+    subject { ProcessMentionsService.new }
+
+    before do
+      stub_request(:post, remote_user.inbox_url)
+      subject.call(status)
+    end
+
+    it 'creates a mention' do
+      expect(remote_user.mentions.where(status: status).count).to eq 1
+    end
+
+    it 'sends activity to the inbox' do
+      expect(a_request(:post, remote_user.inbox_url)).to have_been_made.once
+    end
   end
 end
diff --git a/spec/services/reblog_service_spec.rb b/spec/services/reblog_service_spec.rb
index 5f89169e9..0ad5c5f6b 100644
--- a/spec/services/reblog_service_spec.rb
+++ b/spec/services/reblog_service_spec.rb
@@ -2,22 +2,49 @@ require 'rails_helper'
 
 RSpec.describe ReblogService do
   let(:alice)  { Fabricate(:account, username: 'alice') }
-  let(:bob)    { Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com') }
-  let(:status) { Fabricate(:status, account: bob, uri: 'tag:example.com;something:something') }
 
-  subject { ReblogService.new }
+  context 'OStatus' do
+    let(:bob)    { Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com') }
+    let(:status) { Fabricate(:status, account: bob, uri: 'tag:example.com;something:something') }
 
-  before do
-    stub_request(:post, 'http://salmon.example.com')
+    subject { ReblogService.new }
 
-    subject.(alice, status)
-  end
+    before do
+      stub_request(:post, 'http://salmon.example.com')
+      subject.call(alice, status)
+    end
+
+    it 'creates a reblog' do
+      expect(status.reblogs.count).to eq 1
+    end
 
-  it 'creates a reblog' do
-    expect(status.reblogs.count).to eq 1
+    it 'sends a Salmon slap for a remote reblog' do
+      expect(a_request(:post, 'http://salmon.example.com')).to have_been_made
+    end
   end
 
-  it 'sends a Salmon slap for a remote reblog' do
-    expect(a_request(:post, 'http://salmon.example.com')).to have_been_made
+  context 'ActivityPub' do
+    let(:bob)    { Fabricate(:account, username: 'bob', protocol: :activitypub, domain: 'example.com', inbox_url: 'http://example.com/inbox') }
+    let(:status) { Fabricate(:status, account: bob) }
+
+    subject { ReblogService.new }
+
+    before do
+      stub_request(:post, bob.inbox_url)
+      allow(ActivityPub::DistributionWorker).to receive(:perform_async)
+      subject.call(alice, status)
+    end
+
+    it 'creates a reblog' do
+      expect(status.reblogs.count).to eq 1
+    end
+
+    it 'distributes to followers' do
+      expect(ActivityPub::DistributionWorker).to have_received(:perform_async)
+    end
+
+    it 'sends an announce activity to the author' do
+      expect(a_request(:post, bob.inbox_url)).to have_been_made.once
+    end
   end
 end
diff --git a/spec/services/reject_follow_service_spec.rb b/spec/services/reject_follow_service_spec.rb
index 50749b633..2e06345b3 100644
--- a/spec/services/reject_follow_service_spec.rb
+++ b/spec/services/reject_follow_service_spec.rb
@@ -22,7 +22,7 @@ RSpec.describe RejectFollowService do
     end
   end
 
-  describe 'remote' do
+  describe 'remote OStatus' do
     let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com')).account }
 
     before do
@@ -46,4 +46,26 @@ RSpec.describe RejectFollowService do
       }).to have_been_made.once
     end
   end
+
+  describe 'remote ActivityPub' do
+    let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', domain: 'example.com', protocol: :activitypub, inbox_url: 'http://example.com/inbox')).account }
+
+    before do
+      FollowRequest.create(account: bob, target_account: sender)
+      stub_request(:post, bob.inbox_url).to_return(status: 200)
+      subject.call(bob, sender)
+    end
+
+    it 'removes follow request' do
+      expect(bob.requested?(sender)).to be false
+    end
+
+    it 'does not create follow relation' do
+      expect(bob.following?(sender)).to be false
+    end
+
+    it 'sends a reject activity' do
+      expect(a_request(:post, bob.inbox_url)).to have_been_made.once
+    end
+  end
 end
diff --git a/spec/services/remove_status_service_spec.rb b/spec/services/remove_status_service_spec.rb
index a3bce7613..dc6b350cb 100644
--- a/spec/services/remove_status_service_spec.rb
+++ b/spec/services/remove_status_service_spec.rb
@@ -6,13 +6,17 @@ RSpec.describe RemoveStatusService do
   let!(:alice)  { Fabricate(:account) }
   let!(:bob)    { Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://example.com/salmon') }
   let!(:jeff)   { Fabricate(:account) }
+  let!(:hank)   { Fabricate(:account, username: 'hank', protocol: :activitypub, domain: 'example.com', inbox_url: 'http://example.com/inbox') }
 
   before do
     stub_request(:post, 'http://example.com/push').to_return(status: 200, body: '', headers: {})
     stub_request(:post, 'http://example.com/salmon').to_return(status: 200, body: '', headers: {})
+    stub_request(:post, 'http://example.com/inbox').to_return(status: 200)
 
     Fabricate(:subscription, account: alice, callback_url: 'http://example.com/push', confirmed: true, expires_at: 30.days.from_now)
     jeff.follow!(alice)
+    hank.follow!(alice)
+
     @status = PostStatusService.new.call(alice, 'Hello @bob@example.com')
     subject.call(@status)
   end
@@ -31,6 +35,10 @@ RSpec.describe RemoveStatusService do
     }).to have_been_made
   end
 
+  it 'sends delete activity to followers' do
+    expect(a_request(:post, 'http://example.com/inbox')).to have_been_made.twice
+  end
+
   it 'sends Salmon slap to previously mentioned users' do
     expect(a_request(:post, "http://example.com/salmon").with { |req|
       xml = OStatus2::Salmon.new.unpack(req.body)
diff --git a/spec/services/resolve_remote_account_service_spec.rb b/spec/services/resolve_remote_account_service_spec.rb
index c3b902b34..d0eab2310 100644
--- a/spec/services/resolve_remote_account_service_spec.rb
+++ b/spec/services/resolve_remote_account_service_spec.rb
@@ -1,7 +1,7 @@
 require 'rails_helper'
 
 RSpec.describe ResolveRemoteAccountService do
-  subject { ResolveRemoteAccountService.new }
+  subject { described_class.new }
 
   before do
     stub_request(:get, "https://quitter.no/.well-known/host-meta").to_return(request_fixture('.host-meta.txt'))
@@ -29,29 +29,6 @@ RSpec.describe ResolveRemoteAccountService do
     expect(subject.call('catsrgr8@example.com')).to be_nil
   end
 
-  it 'returns an already existing remote account' do
-    old_account      = Fabricate(:account, username: 'gargron', domain: 'quitter.no')
-    returned_account = subject.call('gargron@quitter.no')
-
-    expect(old_account.id).to eq returned_account.id
-  end
-
-  it 'returns a new remote account' do
-    account = subject.call('gargron@quitter.no')
-
-    expect(account.username).to eq 'gargron'
-    expect(account.domain).to eq 'quitter.no'
-    expect(account.remote_url).to eq 'https://quitter.no/api/statuses/user_timeline/7477.atom'
-  end
-
-  it 'follows a legitimate account redirection' do
-    account = subject.call('gargron@redirected.com')
-
-    expect(account.username).to eq 'gargron'
-    expect(account.domain).to eq 'quitter.no'
-    expect(account.remote_url).to eq 'https://quitter.no/api/statuses/user_timeline/7477.atom'
-  end
-
   it 'prevents hijacking existing accounts' do
     account = subject.call('hacker1@redirected.com')
     expect(account.salmon_url).to_not eq 'https://hacker.com/main/salmon/user/7477'
@@ -61,12 +38,41 @@ RSpec.describe ResolveRemoteAccountService do
     expect(subject.call('hacker2@redirected.com')).to be_nil
   end
 
-  it 'returns a new remote account' do
-    account = subject.call('foo@localdomain.com')
+  context 'with an OStatus account' do
+    it 'returns an already existing remote account' do
+      old_account      = Fabricate(:account, username: 'gargron', domain: 'quitter.no')
+      returned_account = subject.call('gargron@quitter.no')
+
+      expect(old_account.id).to eq returned_account.id
+    end
+
+    it 'returns a new remote account' do
+      account = subject.call('gargron@quitter.no')
+
+      expect(account.username).to eq 'gargron'
+      expect(account.domain).to eq 'quitter.no'
+      expect(account.remote_url).to eq 'https://quitter.no/api/statuses/user_timeline/7477.atom'
+    end
+
+    it 'follows a legitimate account redirection' do
+      account = subject.call('gargron@redirected.com')
+
+      expect(account.username).to eq 'gargron'
+      expect(account.domain).to eq 'quitter.no'
+      expect(account.remote_url).to eq 'https://quitter.no/api/statuses/user_timeline/7477.atom'
+    end
+
+    it 'returns a new remote account' do
+      account = subject.call('foo@localdomain.com')
+
+      expect(account.username).to eq 'foo'
+      expect(account.domain).to eq 'localdomain.com'
+      expect(account.remote_url).to eq 'https://webdomain.com/users/foo.atom'
+    end
+  end
 
-    expect(account.username).to eq 'foo'
-    expect(account.domain).to eq 'localdomain.com'
-    expect(account.remote_url).to eq 'https://webdomain.com/users/foo.atom'
+  context 'with an ActivityPub account' do
+    pending
   end
 
   it 'processes one remote account at a time using locks' do
@@ -78,7 +84,7 @@ RSpec.describe ResolveRemoteAccountService do
       Thread.new do
         true while wait_for_start
         begin
-          return_values << ResolveRemoteAccountService.new.call('foo@localdomain.com')
+          return_values << described_class.new.call('foo@localdomain.com')
         rescue ActiveRecord::RecordNotUnique
           fail_occurred = true
         end
diff --git a/spec/services/unblock_service_spec.rb b/spec/services/unblock_service_spec.rb
index 1b9ae1239..def4981e7 100644
--- a/spec/services/unblock_service_spec.rb
+++ b/spec/services/unblock_service_spec.rb
@@ -18,7 +18,7 @@ RSpec.describe UnblockService do
     end
   end
 
-  describe 'remote' do
+  describe 'remote OStatus' do
     let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com')).account }
 
     before do
@@ -28,7 +28,7 @@ RSpec.describe UnblockService do
     end
 
     it 'destroys the blocking relation' do
-      expect(sender.following?(bob)).to be false
+      expect(sender.blocking?(bob)).to be false
     end
 
     it 'sends an unblock salmon slap' do
@@ -38,4 +38,22 @@ RSpec.describe UnblockService do
       }).to have_been_made.once
     end
   end
+
+  describe 'remote ActivityPub' do
+    let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', protocol: :activitypub, domain: 'example.com', inbox_url: 'http://example.com/inbox')).account }
+
+    before do
+      sender.block!(bob)
+      stub_request(:post, 'http://example.com/inbox').to_return(status: 200)
+      subject.call(sender, bob)
+    end
+
+    it 'destroys the blocking relation' do
+      expect(sender.blocking?(bob)).to be false
+    end
+
+    it 'sends an unblock activity' do
+      expect(a_request(:post, 'http://example.com/inbox')).to have_been_made.once
+    end
+  end
 end
diff --git a/spec/services/unfollow_service_spec.rb b/spec/services/unfollow_service_spec.rb
index 8ec2148a1..29040431e 100644
--- a/spec/services/unfollow_service_spec.rb
+++ b/spec/services/unfollow_service_spec.rb
@@ -18,8 +18,8 @@ RSpec.describe UnfollowService do
     end
   end
 
-  describe 'remote' do
-    let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', domain: 'example.com', salmon_url: 'http://salmon.example.com')).account }
+  describe 'remote OStatus' do
+    let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', protocol: :ostatus, domain: 'example.com', salmon_url: 'http://salmon.example.com')).account }
 
     before do
       sender.follow!(bob)
@@ -38,4 +38,22 @@ RSpec.describe UnfollowService do
       }).to have_been_made.once
     end
   end
+
+  describe 'remote ActivityPub' do
+    let(:bob) { Fabricate(:user, email: 'bob@example.com', account: Fabricate(:account, username: 'bob', protocol: :activitypub, domain: 'example.com', inbox_url: 'http://example.com/inbox')).account }
+
+    before do
+      sender.follow!(bob)
+      stub_request(:post, 'http://example.com/inbox').to_return(status: 200)
+      subject.call(sender, bob)
+    end
+
+    it 'destroys the following relation' do
+      expect(sender.following?(bob)).to be false
+    end
+
+    it 'sends an unfollow activity' do
+      expect(a_request(:post, 'http://example.com/inbox')).to have_been_made.once
+    end
+  end
 end
diff --git a/spec/workers/activitypub/delivery_worker_spec.rb b/spec/workers/activitypub/delivery_worker_spec.rb
new file mode 100644
index 000000000..351be185c
--- /dev/null
+++ b/spec/workers/activitypub/delivery_worker_spec.rb
@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+describe ActivityPub::DeliveryWorker do
+  subject { described_class.new }
+
+  let(:sender)  { Fabricate(:account) }
+  let(:payload) { 'test' }
+
+  describe 'perform' do
+    it 'performs a request' do
+      stub_request(:post, 'https://example.com/api').to_return(status: 200)
+      subject.perform(payload, sender.id, 'https://example.com/api')
+      expect(a_request(:post, 'https://example.com/api')).to have_been_made.once
+    end
+
+    it 'raises when request fails' do
+      stub_request(:post, 'https://example.com/api').to_return(status: 500)
+      expect { subject.perform(payload, sender.id, 'https://example.com/api') }.to raise_error Mastodon::UnexpectedResponseError
+    end
+  end
+end
diff --git a/spec/workers/activitypub/distribution_worker_spec.rb b/spec/workers/activitypub/distribution_worker_spec.rb
new file mode 100644
index 000000000..368ca025a
--- /dev/null
+++ b/spec/workers/activitypub/distribution_worker_spec.rb
@@ -0,0 +1,48 @@
+require 'rails_helper'
+
+describe ActivityPub::DistributionWorker do
+  subject { described_class.new }
+
+  let(:status)   { Fabricate(:status) }
+  let(:follower) { Fabricate(:account, protocol: :activitypub, inbox_url: 'http://example.com') }
+
+  describe '#perform' do
+    before do
+      allow(ActivityPub::DeliveryWorker).to receive(:push_bulk)
+      follower.follow!(status.account)
+    end
+
+    context 'with public status' do
+      before do
+        status.update(visibility: :public)
+      end
+
+      it 'delivers to followers' do
+        subject.perform(status.id)
+        expect(ActivityPub::DeliveryWorker).to have_received(:push_bulk).with(['http://example.com'])
+      end
+    end
+
+    context 'with private status' do
+      before do
+        status.update(visibility: :private)
+      end
+
+      it 'delivers to followers' do
+        subject.perform(status.id)
+        expect(ActivityPub::DeliveryWorker).to have_received(:push_bulk).with(['http://example.com'])
+      end
+    end
+
+    context 'with direct status' do
+      before do
+        status.update(visibility: :direct)
+      end
+
+      it 'does nothing' do
+        subject.perform(status.id)
+        expect(ActivityPub::DeliveryWorker).to_not have_received(:push_bulk)
+      end
+    end
+  end
+end
diff --git a/spec/workers/activitypub/processing_worker_spec.rb b/spec/workers/activitypub/processing_worker_spec.rb
new file mode 100644
index 000000000..b42c0bdbc
--- /dev/null
+++ b/spec/workers/activitypub/processing_worker_spec.rb
@@ -0,0 +1,15 @@
+require 'rails_helper'
+
+describe ActivityPub::ProcessingWorker do
+  subject { described_class.new }
+
+  let(:account) { Fabricate(:account) }
+
+  describe '#perform' do
+    it 'delegates to ActivityPub::ProcessCollectionService' do
+      allow(ActivityPub::ProcessCollectionService).to receive(:new).and_return(double(:service, call: nil))
+      subject.perform(account.id, '')
+      expect(ActivityPub::ProcessCollectionService).to have_received(:new)
+    end
+  end
+end
diff --git a/spec/workers/activitypub/thread_resolve_worker_spec.rb b/spec/workers/activitypub/thread_resolve_worker_spec.rb
new file mode 100644
index 000000000..b954cb62c
--- /dev/null
+++ b/spec/workers/activitypub/thread_resolve_worker_spec.rb
@@ -0,0 +1,16 @@
+require 'rails_helper'
+
+describe ActivityPub::ThreadResolveWorker do
+  subject { described_class.new }
+
+  let(:status) { Fabricate(:status) }
+  let(:parent) { Fabricate(:status) }
+
+  describe '#perform' do
+    it 'gets parent from ActivityPub::FetchRemoteStatusService and glues them together' do
+      allow(ActivityPub::FetchRemoteStatusService).to receive(:new).and_return(double(:service, call: parent))
+      subject.perform(status.id, 'http://example.com/123')
+      expect(status.reload.in_reply_to_id).to eq parent.id
+    end
+  end
+end
diff --git a/spec/workers/activitypub/update_distribution_worker_spec.rb b/spec/workers/activitypub/update_distribution_worker_spec.rb
new file mode 100644
index 000000000..688a424d5
--- /dev/null
+++ b/spec/workers/activitypub/update_distribution_worker_spec.rb
@@ -0,0 +1,20 @@
+require 'rails_helper'
+
+describe ActivityPub::UpdateDistributionWorker do
+  subject { described_class.new }
+
+  let(:account)  { Fabricate(:account) }
+  let(:follower) { Fabricate(:account, protocol: :activitypub, inbox_url: 'http://example.com') }
+
+  describe '#perform' do
+    before do
+      allow(ActivityPub::DeliveryWorker).to receive(:push_bulk)
+      follower.follow!(account)
+    end
+
+    it 'delivers to followers' do
+      subject.perform(account.id)
+      expect(ActivityPub::DeliveryWorker).to have_received(:push_bulk).with(['http://example.com'])
+    end
+  end
+end
-- 
cgit 


From 412ea873060da4dc73236fdd63a2931d27dbfa40 Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Sat, 19 Aug 2017 18:44:48 +0200
Subject: Improve ActivityPub/OStatus compatibility (#4632)

*Note: OStatus URIs are invalid for ActivityPub. But we have them for
as long as we want to keep old OStatus-sourced content and as long as
we remain OStatus-compatible.*

- In Announce handling, if object URI is not a URL, fallback to object URL
- Do not use specialized ThreadResolveWorker, rely on generalized handling
- When serializing notes, if parent's URI is not a URL, use parent's URL
---
 app/lib/activitypub/activity/announce.rb               | 14 ++++++++++++--
 app/lib/activitypub/activity/create.rb                 |  2 +-
 app/serializers/activitypub/note_serializer.rb         |  8 +++++++-
 app/workers/activitypub/thread_resolve_worker.rb       | 17 -----------------
 spec/workers/activitypub/thread_resolve_worker_spec.rb | 16 ----------------
 5 files changed, 20 insertions(+), 37 deletions(-)
 delete mode 100644 app/workers/activitypub/thread_resolve_worker.rb
 delete mode 100644 spec/workers/activitypub/thread_resolve_worker_spec.rb

(limited to 'app/workers')

diff --git a/app/lib/activitypub/activity/announce.rb b/app/lib/activitypub/activity/announce.rb
index decf8f960..09fec28a0 100644
--- a/app/lib/activitypub/activity/announce.rb
+++ b/app/lib/activitypub/activity/announce.rb
@@ -2,8 +2,8 @@
 
 class ActivityPub::Activity::Announce < ActivityPub::Activity
   def perform
-    original_status = status_from_uri(object_uri)
-    original_status = ActivityPub::FetchRemoteStatusService.new.call(object_uri) if original_status.nil?
+    original_status   = status_from_uri(object_uri)
+    original_status ||= fetch_remote_original_status
 
     return if original_status.nil? || delete_arrived_first?(@json['id'])
 
@@ -11,4 +11,14 @@ class ActivityPub::Activity::Announce < ActivityPub::Activity
     distribute(status)
     status
   end
+
+  private
+
+  def fetch_remote_original_status
+    if object_uri.start_with?('http')
+      ActivityPub::FetchRemoteStatusService.new.call(object_uri)
+    elsif @object['url'].present?
+      ::FetchRemoteStatusService.new.call(@object['url'])
+    end
+  end
 end
diff --git a/app/lib/activitypub/activity/create.rb b/app/lib/activitypub/activity/create.rb
index 77d66fba3..154125759 100644
--- a/app/lib/activitypub/activity/create.rb
+++ b/app/lib/activitypub/activity/create.rb
@@ -91,7 +91,7 @@ class ActivityPub::Activity::Create < ActivityPub::Activity
 
   def resolve_thread(status)
     return unless status.reply? && status.thread.nil?
-    ActivityPub::ThreadResolveWorker.perform_async(status.id, @object['inReplyTo'])
+    ThreadResolveWorker.perform_async(status.id, @object['inReplyTo'])
   end
 
   def conversation_from_uri(uri)
diff --git a/app/serializers/activitypub/note_serializer.rb b/app/serializers/activitypub/note_serializer.rb
index bc8eb8a35..4061b9ce4 100644
--- a/app/serializers/activitypub/note_serializer.rb
+++ b/app/serializers/activitypub/note_serializer.rb
@@ -27,7 +27,13 @@ class ActivityPub::NoteSerializer < ActiveModel::Serializer
   end
 
   def in_reply_to
-    ActivityPub::TagManager.instance.uri_for(object.thread) if object.reply?
+    return unless object.reply?
+
+    if object.thread.uri.nil? || object.thread.uri.start_with?('http')
+      ActivityPub::TagManager.instance.uri_for(object.thread)
+    else
+      object.thread.url
+    end
   end
 
   def published
diff --git a/app/workers/activitypub/thread_resolve_worker.rb b/app/workers/activitypub/thread_resolve_worker.rb
deleted file mode 100644
index 4ef762d06..000000000
--- a/app/workers/activitypub/thread_resolve_worker.rb
+++ /dev/null
@@ -1,17 +0,0 @@
-# frozen_string_literal: true
-
-class ActivityPub::ThreadResolveWorker
-  include Sidekiq::Worker
-
-  sidekiq_options queue: 'pull', retry: false
-
-  def perform(child_status_id, parent_uri)
-    child_status  = Status.find(child_status_id)
-    parent_status = ActivityPub::FetchRemoteStatusService.new.call(parent_uri)
-
-    return if parent_status.nil?
-
-    child_status.thread = parent_status
-    child_status.save!
-  end
-end
diff --git a/spec/workers/activitypub/thread_resolve_worker_spec.rb b/spec/workers/activitypub/thread_resolve_worker_spec.rb
deleted file mode 100644
index b954cb62c..000000000
--- a/spec/workers/activitypub/thread_resolve_worker_spec.rb
+++ /dev/null
@@ -1,16 +0,0 @@
-require 'rails_helper'
-
-describe ActivityPub::ThreadResolveWorker do
-  subject { described_class.new }
-
-  let(:status) { Fabricate(:status) }
-  let(:parent) { Fabricate(:status) }
-
-  describe '#perform' do
-    it 'gets parent from ActivityPub::FetchRemoteStatusService and glues them together' do
-      allow(ActivityPub::FetchRemoteStatusService).to receive(:new).and_return(double(:service, call: parent))
-      subject.perform(status.id, 'http://example.com/123')
-      expect(status.reload.in_reply_to_id).to eq parent.id
-    end
-  end
-end
-- 
cgit 


From 6e9eda53319bc970b085c7c55277981320b2a835 Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Mon, 21 Aug 2017 01:14:40 +0200
Subject: ActivityPub migration procedure (#4617)

* ActivityPub migration procedure

Once one account is detected as going from OStatus to ActivityPub,
invalidate WebFinger cache for other accounts from the same domain

* Unsubscribe from PuSH updates once we receive an ActivityPub payload

* Re-subscribe to PuSH unless already unsubscribed, regardless of protocol
---
 app/controllers/activitypub/inboxes_controller.rb   |  6 ++++++
 app/controllers/admin/accounts_controller.rb        |  2 +-
 app/services/activitypub/process_account_service.rb |  8 +++++++-
 app/services/unsubscribe_service.rb                 |  2 +-
 app/workers/activitypub/post_upgrade_worker.rb      | 15 +++++++++++++++
 app/workers/pubsubhubbub/unsubscribe_worker.rb      | 15 +++++++++++++++
 app/workers/scheduler/subscriptions_scheduler.rb    |  2 +-
 lib/tasks/mastodon.rake                             |  5 +----
 8 files changed, 47 insertions(+), 8 deletions(-)
 create mode 100644 app/workers/activitypub/post_upgrade_worker.rb
 create mode 100644 app/workers/pubsubhubbub/unsubscribe_worker.rb

(limited to 'app/workers')

diff --git a/app/controllers/activitypub/inboxes_controller.rb b/app/controllers/activitypub/inboxes_controller.rb
index 0f49b4399..078494c20 100644
--- a/app/controllers/activitypub/inboxes_controller.rb
+++ b/app/controllers/activitypub/inboxes_controller.rb
@@ -7,6 +7,7 @@ class ActivityPub::InboxesController < Api::BaseController
 
   def create
     if signed_request_account
+      upgrade_account
       process_payload
       head 201
     else
@@ -24,6 +25,11 @@ class ActivityPub::InboxesController < Api::BaseController
     @body ||= request.body.read
   end
 
+  def upgrade_account
+    return unless signed_request_account.subscribed?
+    Pubsubhubbub::UnsubscribeWorker.perform_async(signed_request_account.id)
+  end
+
   def process_payload
     ActivityPub::ProcessingWorker.perform_async(signed_request_account.id, body.force_encoding('UTF-8'))
   end
diff --git a/app/controllers/admin/accounts_controller.rb b/app/controllers/admin/accounts_controller.rb
index 7bceee2cd..54c659e1b 100644
--- a/app/controllers/admin/accounts_controller.rb
+++ b/app/controllers/admin/accounts_controller.rb
@@ -17,7 +17,7 @@ module Admin
     end
 
     def unsubscribe
-      UnsubscribeService.new.call(@account)
+      Pubsubhubbub::UnsubscribeWorker.perform_async(@account.id)
       redirect_to admin_account_path(@account.id)
     end
 
diff --git a/app/services/activitypub/process_account_service.rb b/app/services/activitypub/process_account_service.rb
index 9fb7ebf9e..2f2dfd330 100644
--- a/app/services/activitypub/process_account_service.rb
+++ b/app/services/activitypub/process_account_service.rb
@@ -12,7 +12,8 @@ class ActivityPub::ProcessAccountService < BaseService
     @domain   = domain
     @account  = Account.find_by(uri: @uri)
 
-    create_account if @account.nil?
+    create_account  if @account.nil?
+    upgrade_account if @account.ostatus?
     update_account
 
     @account
@@ -24,6 +25,7 @@ class ActivityPub::ProcessAccountService < BaseService
 
   def create_account
     @account = Account.new
+    @account.protocol    = :activitypub
     @account.username    = @username
     @account.domain      = @domain
     @account.uri         = @uri
@@ -50,6 +52,10 @@ class ActivityPub::ProcessAccountService < BaseService
     @account.save!
   end
 
+  def upgrade_account
+    ActivityPub::PostUpgradeWorker.perform_async(@account.domain)
+  end
+
   def image_url(key)
     value = first_of_value(@json[key])
 
diff --git a/app/services/unsubscribe_service.rb b/app/services/unsubscribe_service.rb
index c5e0e73fe..865f783bc 100644
--- a/app/services/unsubscribe_service.rb
+++ b/app/services/unsubscribe_service.rb
@@ -2,7 +2,7 @@
 
 class UnsubscribeService < BaseService
   def call(account)
-    return unless account.ostatus?
+    return if account.hub_url.blank?
 
     @account  = account
     @response = build_request.perform
diff --git a/app/workers/activitypub/post_upgrade_worker.rb b/app/workers/activitypub/post_upgrade_worker.rb
new file mode 100644
index 000000000..4154b8582
--- /dev/null
+++ b/app/workers/activitypub/post_upgrade_worker.rb
@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+class ActivityPub::PostUpgradeWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'pull'
+
+  def perform(domain)
+    Account.where(domain: domain)
+           .where(protocol: :ostatus)
+           .where.not(last_webfingered_at: nil)
+           .in_batches
+           .update_all(last_webfingered_at: nil)
+  end
+end
diff --git a/app/workers/pubsubhubbub/unsubscribe_worker.rb b/app/workers/pubsubhubbub/unsubscribe_worker.rb
new file mode 100644
index 000000000..a271715b7
--- /dev/null
+++ b/app/workers/pubsubhubbub/unsubscribe_worker.rb
@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+class Pubsubhubbub::UnsubscribeWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'push', retry: false, unique: :until_executed, dead: false
+
+  def perform(account_id)
+    account = Account.find(account_id)
+    logger.debug "PuSH unsubscribing from #{account.acct}"
+    ::UnsubscribeService.new.call(account)
+  rescue ActiveRecord::RecordNotFound
+    true
+  end
+end
diff --git a/app/workers/scheduler/subscriptions_scheduler.rb b/app/workers/scheduler/subscriptions_scheduler.rb
index 5ddfaed18..35ecda2db 100644
--- a/app/workers/scheduler/subscriptions_scheduler.rb
+++ b/app/workers/scheduler/subscriptions_scheduler.rb
@@ -14,6 +14,6 @@ class Scheduler::SubscriptionsScheduler
   private
 
   def expiring_accounts
-    Account.where(protocol: :ostatus).expiring(1.day.from_now).partitioned
+    Account.expiring(1.day.from_now).partitioned
   end
 end
diff --git a/lib/tasks/mastodon.rake b/lib/tasks/mastodon.rake
index 226523554..870fd7e4e 100644
--- a/lib/tasks/mastodon.rake
+++ b/lib/tasks/mastodon.rake
@@ -111,10 +111,7 @@ namespace :mastodon do
   namespace :push do
     desc 'Unsubscribes from PuSH updates of feeds nobody follows locally'
     task clear: :environment do
-      Account.remote.without_followers.where.not(subscription_expires_at: nil).find_each do |a|
-        Rails.logger.debug "PuSH unsubscribing from #{a.acct}"
-        UnsubscribeService.new.call(a)
-      end
+      Pubsubhubbub::UnsubscribeWorker.push_bulk(Account.remote.without_followers.where.not(subscription_expires_at: nil).pluck(:id))
     end
 
     desc 'Re-subscribes to soon expiring PuSH subscriptions (deprecated)'
-- 
cgit 


From f391a4673adc6d79bb3a46b493d599a4bf6d558f Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Mon, 21 Aug 2017 22:56:33 +0200
Subject: Periodically remove expired PuSH subscribers (#4654)

---
 app/models/subscription.rb                               |  1 +
 app/workers/scheduler/feed_cleanup_scheduler.rb          |  2 --
 app/workers/scheduler/media_cleanup_scheduler.rb         |  1 -
 app/workers/scheduler/subscriptions_cleanup_scheduler.rb | 11 +++++++++++
 app/workers/scheduler/subscriptions_scheduler.rb         |  1 -
 config/sidekiq.yml                                       |  3 +++
 6 files changed, 15 insertions(+), 4 deletions(-)
 create mode 100644 app/workers/scheduler/subscriptions_cleanup_scheduler.rb

(limited to 'app/workers')

diff --git a/app/models/subscription.rb b/app/models/subscription.rb
index bf643c1f9..14f1a140c 100644
--- a/app/models/subscription.rb
+++ b/app/models/subscription.rb
@@ -26,6 +26,7 @@ class Subscription < ApplicationRecord
 
   scope :confirmed, -> { where(confirmed: true) }
   scope :future_expiration, -> { where(arel_table[:expires_at].gt(Time.now.utc)) }
+  scope :expired, -> { where(arel_table[:expires_at].lt(Time.now.utc)) }
   scope :active, -> { confirmed.future_expiration }
 
   def lease_seconds=(value)
diff --git a/app/workers/scheduler/feed_cleanup_scheduler.rb b/app/workers/scheduler/feed_cleanup_scheduler.rb
index 402eed7c6..dbebaa2c3 100644
--- a/app/workers/scheduler/feed_cleanup_scheduler.rb
+++ b/app/workers/scheduler/feed_cleanup_scheduler.rb
@@ -5,8 +5,6 @@ class Scheduler::FeedCleanupScheduler
   include Sidekiq::Worker
 
   def perform
-    logger.info 'Cleaning out home feeds of inactive users'
-
     redis.pipelined do
       inactive_users.pluck(:account_id).each do |account_id|
         redis.del(FeedManager.instance.key(:home, account_id))
diff --git a/app/workers/scheduler/media_cleanup_scheduler.rb b/app/workers/scheduler/media_cleanup_scheduler.rb
index a95f512be..ce32ce314 100644
--- a/app/workers/scheduler/media_cleanup_scheduler.rb
+++ b/app/workers/scheduler/media_cleanup_scheduler.rb
@@ -5,7 +5,6 @@ class Scheduler::MediaCleanupScheduler
   include Sidekiq::Worker
 
   def perform
-    logger.info 'Cleaning out unattached media attachments'
     unattached_media.find_each(&:destroy)
   end
 
diff --git a/app/workers/scheduler/subscriptions_cleanup_scheduler.rb b/app/workers/scheduler/subscriptions_cleanup_scheduler.rb
new file mode 100644
index 000000000..3b9211e81
--- /dev/null
+++ b/app/workers/scheduler/subscriptions_cleanup_scheduler.rb
@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+require 'sidekiq-scheduler'
+
+class Scheduler::SubscriptionsCleanupScheduler
+  include Sidekiq::Worker
+
+  def perform
+    Subscription.expired.in_batches.delete_all
+  end
+end
diff --git a/app/workers/scheduler/subscriptions_scheduler.rb b/app/workers/scheduler/subscriptions_scheduler.rb
index 35ecda2db..469a3d2a6 100644
--- a/app/workers/scheduler/subscriptions_scheduler.rb
+++ b/app/workers/scheduler/subscriptions_scheduler.rb
@@ -7,7 +7,6 @@ class Scheduler::SubscriptionsScheduler
   include Sidekiq::Worker
 
   def perform
-    logger.info 'Queueing PuSH re-subscriptions'
     Pubsubhubbub::SubscribeWorker.push_bulk(expiring_accounts.pluck(:id))
   end
 
diff --git a/config/sidekiq.yml b/config/sidekiq.yml
index 8273c1201..a502f5593 100644
--- a/config/sidekiq.yml
+++ b/config/sidekiq.yml
@@ -21,3 +21,6 @@
   user_cleanup_scheduler:
     cron: '4 5 * * *'
     class: Scheduler::UserCleanupScheduler
+  subscriptions_cleanup_scheduler:
+    cron: '2 2 * * 0'
+    class: Scheduler::SubscriptionsCleanupScheduler
-- 
cgit 


From cf615abbf9323f3b73681306090de48f9e13a6b9 Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Thu, 24 Aug 2017 17:51:32 +0200
Subject: Add configuration to disable private status federation over PuSH
 (#4582)

---
 app/services/process_mentions_service.rb           |  2 +-
 app/workers/pubsubhubbub/distribution_worker.rb    |  2 +-
 config/initializers/ostatus.rb                     |  3 +-
 .../pubsubhubbub/distribution_worker_spec.rb       | 64 +++++++++++++++++-----
 4 files changed, 55 insertions(+), 16 deletions(-)

(limited to 'app/workers')

diff --git a/app/services/process_mentions_service.rb b/app/services/process_mentions_service.rb
index 407fa8c18..2b8a77147 100644
--- a/app/services/process_mentions_service.rb
+++ b/app/services/process_mentions_service.rb
@@ -39,7 +39,7 @@ class ProcessMentionsService < BaseService
 
     if mentioned_account.local?
       NotifyService.new.call(mentioned_account, mention)
-    elsif mentioned_account.ostatus?
+    elsif mentioned_account.ostatus? && (Rails.configuration.x.use_ostatus_privacy || !status.stream_entry.hidden?)
       NotificationWorker.perform_async(stream_entry_to_xml(status.stream_entry), status.account_id, mentioned_account.id)
     elsif mentioned_account.activitypub?
       ActivityPub::DeliveryWorker.perform_async(build_json(mention.status), mention.status.account_id, mentioned_account.inbox_url)
diff --git a/app/workers/pubsubhubbub/distribution_worker.rb b/app/workers/pubsubhubbub/distribution_worker.rb
index ea246128d..2a5e60fa0 100644
--- a/app/workers/pubsubhubbub/distribution_worker.rb
+++ b/app/workers/pubsubhubbub/distribution_worker.rb
@@ -14,7 +14,7 @@ class Pubsubhubbub::DistributionWorker
     @subscriptions = active_subscriptions.to_a
 
     distribute_public!(stream_entries.reject(&:hidden?))
-    distribute_hidden!(stream_entries.select(&:hidden?))
+    distribute_hidden!(stream_entries.select(&:hidden?)) if Rails.configuration.x.use_ostatus_privacy
   end
 
   private
diff --git a/config/initializers/ostatus.rb b/config/initializers/ostatus.rb
index 342996dcd..a885545f8 100644
--- a/config/initializers/ostatus.rb
+++ b/config/initializers/ostatus.rb
@@ -5,7 +5,7 @@ host     = ENV.fetch('LOCAL_DOMAIN') { "localhost:#{port}" }
 web_host = ENV.fetch('WEB_DOMAIN') { host }
 https    = ENV['LOCAL_HTTPS'] == 'true'
 
-alternate_domains = ENV.fetch('ALTERNATE_DOMAINS') { "" }
+alternate_domains = ENV.fetch('ALTERNATE_DOMAINS') { '' }
 
 Rails.application.configure do
   config.x.local_domain = host
@@ -17,6 +17,7 @@ Rails.application.configure do
 
   config.action_mailer.default_url_options = { host: web_host, protocol: https ? 'https://' : 'http://', trailing_slash: false }
   config.x.streaming_api_base_url          = 'ws://localhost:4000'
+  config.x.use_ostatus_privacy             = true
 
   if Rails.env.production?
     config.x.streaming_api_base_url = ENV.fetch('STREAMING_API_BASE_URL') { "ws#{https ? 's' : ''}://#{web_host}" }
diff --git a/spec/workers/pubsubhubbub/distribution_worker_spec.rb b/spec/workers/pubsubhubbub/distribution_worker_spec.rb
index 89191c084..5c22e7fa8 100644
--- a/spec/workers/pubsubhubbub/distribution_worker_spec.rb
+++ b/spec/workers/pubsubhubbub/distribution_worker_spec.rb
@@ -22,24 +22,62 @@ describe Pubsubhubbub::DistributionWorker do
     end
   end
 
-  describe 'with private status' do
-    let(:status) { Fabricate(:status, account: alice, text: 'Hello', visibility: :private) }
+  context 'when OStatus privacy is used' do
+    around do |example|
+      before_val = Rails.configuration.x.use_ostatus_privacy
+      Rails.configuration.x.use_ostatus_privacy = true
+      example.run
+      Rails.configuration.x.use_ostatus_privacy = before_val
+    end
 
-    it 'delivers payload only to subscriptions with followers' do
-      allow(Pubsubhubbub::DeliveryWorker).to receive(:push_bulk)
-      subject.perform(status.stream_entry.id)
-      expect(Pubsubhubbub::DeliveryWorker).to have_received(:push_bulk).with([subscription_with_follower])
-      expect(Pubsubhubbub::DeliveryWorker).to_not have_received(:push_bulk).with([anonymous_subscription])
+    describe 'with private status' do
+      let(:status) { Fabricate(:status, account: alice, text: 'Hello', visibility: :private) }
+
+      it 'delivers payload only to subscriptions with followers' do
+        allow(Pubsubhubbub::DeliveryWorker).to receive(:push_bulk)
+        subject.perform(status.stream_entry.id)
+        expect(Pubsubhubbub::DeliveryWorker).to have_received(:push_bulk).with([subscription_with_follower])
+        expect(Pubsubhubbub::DeliveryWorker).to_not have_received(:push_bulk).with([anonymous_subscription])
+      end
+    end
+
+    describe 'with direct status' do
+      let(:status) { Fabricate(:status, account: alice, text: 'Hello', visibility: :direct) }
+
+      it 'does not deliver payload' do
+        allow(Pubsubhubbub::DeliveryWorker).to receive(:push_bulk)
+        subject.perform(status.stream_entry.id)
+        expect(Pubsubhubbub::DeliveryWorker).to_not have_received(:push_bulk)
+      end
     end
   end
 
-  describe 'with direct status' do
-    let(:status) { Fabricate(:status, account: alice, text: 'Hello', visibility: :direct) }
+  context 'when OStatus privacy is not used' do
+    around do |example|
+      before_val = Rails.configuration.x.use_ostatus_privacy
+      Rails.configuration.x.use_ostatus_privacy = false
+      example.run
+      Rails.configuration.x.use_ostatus_privacy = before_val
+    end
 
-    it 'does not deliver payload' do
-      allow(Pubsubhubbub::DeliveryWorker).to receive(:push_bulk)
-      subject.perform(status.stream_entry.id)
-      expect(Pubsubhubbub::DeliveryWorker).to_not have_received(:push_bulk)
+    describe 'with private status' do
+      let(:status) { Fabricate(:status, account: alice, text: 'Hello', visibility: :private) }
+
+      it 'does not deliver anything' do
+        allow(Pubsubhubbub::DeliveryWorker).to receive(:push_bulk)
+        subject.perform(status.stream_entry.id)
+        expect(Pubsubhubbub::DeliveryWorker).to_not have_received(:push_bulk)
+      end
+    end
+
+    describe 'with direct status' do
+      let(:status) { Fabricate(:status, account: alice, text: 'Hello', visibility: :direct) }
+
+      it 'does not deliver payload' do
+        allow(Pubsubhubbub::DeliveryWorker).to receive(:push_bulk)
+        subject.perform(status.stream_entry.id)
+        expect(Pubsubhubbub::DeliveryWorker).to_not have_received(:push_bulk)
+      end
     end
   end
 end
-- 
cgit 


From 649a20ab46eadf9ae3bfc30782ae62379383bd72 Mon Sep 17 00:00:00 2001
From: masarakki <masaki182@gmail.com>
Date: Sat, 26 Aug 2017 19:40:03 +0900
Subject: authorize-follow-requests-after-unlocking (#4658)

---
 .../api/v1/accounts/credentials_controller.rb       |  2 +-
 app/controllers/settings/profiles_controller.rb     |  2 +-
 app/services/update_account_service.rb              | 21 +++++++++++++++++++++
 app/workers/authorize_follow_worker.rb              | 14 ++++++++++++++
 4 files changed, 37 insertions(+), 2 deletions(-)
 create mode 100644 app/services/update_account_service.rb
 create mode 100644 app/workers/authorize_follow_worker.rb

(limited to 'app/workers')

diff --git a/app/controllers/api/v1/accounts/credentials_controller.rb b/app/controllers/api/v1/accounts/credentials_controller.rb
index bea83cd2a..da534d960 100644
--- a/app/controllers/api/v1/accounts/credentials_controller.rb
+++ b/app/controllers/api/v1/accounts/credentials_controller.rb
@@ -12,7 +12,7 @@ class Api::V1::Accounts::CredentialsController < Api::BaseController
 
   def update
     @account = current_account
-    @account.update!(account_params)
+    UpdateAccountService.new.call(@account, account_params, raise_error: true)
     ActivityPub::UpdateDistributionWorker.perform_async(@account.id)
     render json: @account, serializer: REST::CredentialAccountSerializer
   end
diff --git a/app/controllers/settings/profiles_controller.rb b/app/controllers/settings/profiles_controller.rb
index c751c64ae..28f78a4fb 100644
--- a/app/controllers/settings/profiles_controller.rb
+++ b/app/controllers/settings/profiles_controller.rb
@@ -14,7 +14,7 @@ class Settings::ProfilesController < ApplicationController
   def show; end
 
   def update
-    if @account.update(account_params)
+    if UpdateAccountService.new.call(@account, account_params)
       ActivityPub::UpdateDistributionWorker.perform_async(@account.id)
       redirect_to settings_profile_path, notice: I18n.t('generic.changes_saved_msg')
     else
diff --git a/app/services/update_account_service.rb b/app/services/update_account_service.rb
new file mode 100644
index 000000000..09ea377e7
--- /dev/null
+++ b/app/services/update_account_service.rb
@@ -0,0 +1,21 @@
+# frozen_string_literal: true
+
+class UpdateAccountService < BaseService
+  def call(account, params, raise_error: false)
+    was_locked = account.locked
+    update_method = raise_error ? :update! : :update
+    account.send(update_method, params).tap do |ret|
+      next unless ret
+      authorize_all_follow_requests(account) if was_locked && !account.locked
+    end
+  end
+
+  private
+
+  def authorize_all_follow_requests(account)
+    follow_requests = FollowRequest.where(target_account: account)
+    AuthorizeFollowWorker.push_bulk(follow_requests) do |req|
+      [req.account_id, req.target_account_id]
+    end
+  end
+end
diff --git a/app/workers/authorize_follow_worker.rb b/app/workers/authorize_follow_worker.rb
new file mode 100644
index 000000000..0d5014624
--- /dev/null
+++ b/app/workers/authorize_follow_worker.rb
@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+class AuthorizeFollowWorker
+  include Sidekiq::Worker
+
+  def perform(source_account_id, target_account_id)
+    source_account = Account.find(source_account_id)
+    target_account = Account.find(target_account_id)
+
+    AuthorizeFollowService.new.call(source_account, target_account)
+  rescue ActiveRecord::RecordNotFound
+    true
+  end
+end
-- 
cgit 


From 00840f4f2edb8d1d46638ccbc90a1f4462d0867a Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Sat, 26 Aug 2017 13:47:38 +0200
Subject: Add handling of Linked Data Signatures in payloads (#4687)

* Add handling of Linked Data Signatures in payloads

* Add a way to sign JSON, fix canonicalization of signature options

* Fix signatureValue encoding, send out signed JSON when distributing

* Add missing security context
---
 .rubocop.yml                                       |  1 +
 Gemfile                                            |  3 +
 Gemfile.lock                                       | 16 ++++
 app/helpers/jsonld_helper.rb                       | 13 ++++
 app/lib/activitypub/adapter.rb                     |  2 +-
 app/lib/activitypub/linked_data_signature.rb       | 56 ++++++++++++++
 .../activitypub/process_collection_service.rb      | 11 +++
 app/services/authorize_follow_service.rb           |  4 +-
 app/services/batched_remove_status_service.rb      |  8 +-
 app/services/block_service.rb                      |  4 +-
 app/services/favourite_service.rb                  |  4 +-
 app/services/follow_service.rb                     |  4 +-
 app/services/process_mentions_service.rb           |  4 +-
 app/services/reblog_service.rb                     |  4 +-
 app/services/reject_follow_service.rb              |  4 +-
 app/services/remove_status_service.rb              | 10 ++-
 app/services/unblock_service.rb                    |  4 +-
 app/services/unfavourite_service.rb                |  4 +-
 app/services/unfollow_service.rb                   |  4 +-
 app/workers/activitypub/distribution_worker.rb     |  8 +-
 config/initializers/json_ld.rb                     |  4 +
 lib/json_ld/identity.rb                            | 86 ++++++++++++++++++++++
 lib/json_ld/security.rb                            | 50 +++++++++++++
 spec/lib/activitypub/linked_data_signature_spec.rb | 86 ++++++++++++++++++++++
 .../activitypub/process_collection_service_spec.rb |  5 +-
 25 files changed, 369 insertions(+), 30 deletions(-)
 create mode 100644 app/lib/activitypub/linked_data_signature.rb
 create mode 100644 config/initializers/json_ld.rb
 create mode 100644 lib/json_ld/identity.rb
 create mode 100644 lib/json_ld/security.rb
 create mode 100644 spec/lib/activitypub/linked_data_signature_spec.rb

(limited to 'app/workers')

diff --git a/.rubocop.yml b/.rubocop.yml
index ae3697174..a36aa5cae 100644
--- a/.rubocop.yml
+++ b/.rubocop.yml
@@ -10,6 +10,7 @@ AllCops:
   - 'node_modules/**/*'
   - 'Vagrantfile'
   - 'vendor/**/*'
+  - 'lib/json_ld/*'
 
 Bundler/OrderedGems:
   Enabled: false
diff --git a/Gemfile b/Gemfile
index 52ac43b9a..ae90697f1 100644
--- a/Gemfile
+++ b/Gemfile
@@ -68,6 +68,9 @@ gem 'tzinfo-data', '~> 1.2017'
 gem 'webpacker', '~> 2.0'
 gem 'webpush'
 
+gem 'json-ld-preloaded', '~> 2.2.1'
+gem 'rdf-normalize', '~> 0.3.1'
+
 group :development, :test do
   gem 'fabrication', '~> 2.16'
   gem 'fuubar', '~> 2.2'
diff --git a/Gemfile.lock b/Gemfile.lock
index adc37f7de..cd4573637 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -179,6 +179,8 @@ GEM
       activesupport (>= 4.0.1)
       hamlit (>= 1.2.0)
       railties (>= 4.0.1)
+    hamster (3.0.0)
+      concurrent-ruby (~> 1.0)
     hashdiff (0.3.5)
     highline (1.7.8)
     hiredis (0.6.1)
@@ -211,6 +213,13 @@ GEM
     idn-ruby (0.1.0)
     jmespath (1.3.1)
     json (2.1.0)
+    json-ld (2.1.5)
+      multi_json (~> 1.12)
+      rdf (~> 2.2)
+    json-ld-preloaded (2.2.1)
+      json-ld (~> 2.1, >= 2.1.5)
+      multi_json (~> 1.11)
+      rdf (~> 2.2)
     jsonapi-renderer (0.1.3)
     jwt (1.5.6)
     kaminari (1.0.1)
@@ -348,6 +357,11 @@ GEM
     rainbow (2.2.2)
       rake
     rake (12.0.0)
+    rdf (2.2.8)
+      hamster (~> 3.0)
+      link_header (~> 0.0, >= 0.0.8)
+    rdf-normalize (0.3.2)
+      rdf (~> 2.0)
     redis (3.3.3)
     redis-actionpack (5.0.1)
       actionpack (>= 4.0, < 6)
@@ -531,6 +545,7 @@ DEPENDENCIES
   httplog (~> 0.99)
   i18n-tasks (~> 0.9)
   idn-ruby
+  json-ld-preloaded (~> 2.2.1)
   kaminari (~> 1.0)
   letter_opener (~> 1.4)
   letter_opener_web (~> 1.3)
@@ -560,6 +575,7 @@ DEPENDENCIES
   rails-controller-testing (~> 1.0)
   rails-i18n (~> 5.0)
   rails-settings-cached (~> 0.6)
+  rdf-normalize (~> 0.3.1)
   redis (~> 3.3)
   redis-namespace (~> 1.5)
   redis-rails (~> 5.0)
diff --git a/app/helpers/jsonld_helper.rb b/app/helpers/jsonld_helper.rb
index 8355eb055..09446c8be 100644
--- a/app/helpers/jsonld_helper.rb
+++ b/app/helpers/jsonld_helper.rb
@@ -17,6 +17,11 @@ module JsonLdHelper
     !json.nil? && equals_or_includes?(json['@context'], ActivityPub::TagManager::CONTEXT)
   end
 
+  def canonicalize(json)
+    graph = RDF::Graph.new << JSON::LD::API.toRdf(json)
+    graph.dump(:normalize)
+  end
+
   def fetch_resource(uri)
     response = build_request(uri).perform
     return if response.code != 200
@@ -29,6 +34,14 @@ module JsonLdHelper
     nil
   end
 
+  def merge_context(context, new_context)
+    if context.is_a?(Array)
+      context << new_context
+    else
+      [context, new_context]
+    end
+  end
+
   private
 
   def build_request(uri)
diff --git a/app/lib/activitypub/adapter.rb b/app/lib/activitypub/adapter.rb
index df132f019..92210579e 100644
--- a/app/lib/activitypub/adapter.rb
+++ b/app/lib/activitypub/adapter.rb
@@ -11,7 +11,7 @@ class ActivityPub::Adapter < ActiveModelSerializers::Adapter::Base
 
   def serializable_hash(options = nil)
     options = serialization_options(options)
-    serialized_hash = { '@context': ActivityPub::TagManager::CONTEXT }.merge(ActiveModelSerializers::Adapter::Attributes.new(serializer, instance_options).serializable_hash(options))
+    serialized_hash = { '@context': [ActivityPub::TagManager::CONTEXT, 'https://w3id.org/security/v1'] }.merge(ActiveModelSerializers::Adapter::Attributes.new(serializer, instance_options).serializable_hash(options))
     self.class.transform_key_casing!(serialized_hash, instance_options)
   end
 end
diff --git a/app/lib/activitypub/linked_data_signature.rb b/app/lib/activitypub/linked_data_signature.rb
new file mode 100644
index 000000000..7173aed19
--- /dev/null
+++ b/app/lib/activitypub/linked_data_signature.rb
@@ -0,0 +1,56 @@
+# frozen_string_literal: true
+
+class ActivityPub::LinkedDataSignature
+  include JsonLdHelper
+
+  CONTEXT = 'https://w3id.org/identity/v1'
+
+  def initialize(json)
+    @json = json
+  end
+
+  def verify_account!
+    return unless @json['signature'].is_a?(Hash)
+
+    type        = @json['signature']['type']
+    creator_uri = @json['signature']['creator']
+    signature   = @json['signature']['signatureValue']
+
+    return unless type == 'RsaSignature2017'
+
+    creator   = ActivityPub::TagManager.instance.uri_to_resource(creator_uri, Account)
+    creator ||= ActivityPub::FetchRemoteKeyService.new.call(creator_uri)
+
+    return if creator.nil?
+
+    options_hash   = hash(@json['signature'].without('type', 'id', 'signatureValue').merge('@context' => CONTEXT))
+    document_hash  = hash(@json.without('signature'))
+    to_be_verified = options_hash + document_hash
+
+    if creator.keypair.public_key.verify(OpenSSL::Digest::SHA256.new, Base64.decode64(signature), to_be_verified)
+      creator
+    end
+  end
+
+  def sign!(creator)
+    options = {
+      'type'    => 'RsaSignature2017',
+      'creator' => [ActivityPub::TagManager.instance.uri_for(creator), '#main-key'].join,
+      'created' => Time.now.utc.iso8601,
+    }
+
+    options_hash  = hash(options.without('type', 'id', 'signatureValue').merge('@context' => CONTEXT))
+    document_hash = hash(@json.without('signature'))
+    to_be_signed  = options_hash + document_hash
+
+    signature = Base64.strict_encode64(creator.keypair.sign(OpenSSL::Digest::SHA256.new, to_be_signed))
+
+    @json.merge('@context' => merge_context(@json['@context'], CONTEXT), 'signature' => options.merge('signatureValue' => signature))
+  end
+
+  private
+
+  def hash(obj)
+    Digest::SHA256.hexdigest(canonicalize(obj))
+  end
+end
diff --git a/app/services/activitypub/process_collection_service.rb b/app/services/activitypub/process_collection_service.rb
index cd861c075..2cf15553d 100644
--- a/app/services/activitypub/process_collection_service.rb
+++ b/app/services/activitypub/process_collection_service.rb
@@ -9,6 +9,8 @@ class ActivityPub::ProcessCollectionService < BaseService
 
     return if @account.suspended? || !supported_context?
 
+    verify_account! if different_actor?
+
     case @json['type']
     when 'Collection', 'CollectionPage'
       process_items @json['items']
@@ -23,6 +25,10 @@ class ActivityPub::ProcessCollectionService < BaseService
 
   private
 
+  def different_actor?
+    @json['actor'].present? && value_or_id(@json['actor']) != @account.uri && @json['signature'].present?
+  end
+
   def process_items(items)
     items.reverse_each.map { |item| process_item(item) }.compact
   end
@@ -35,4 +41,9 @@ class ActivityPub::ProcessCollectionService < BaseService
     activity = ActivityPub::Activity.factory(item, @account)
     activity&.perform
   end
+
+  def verify_account!
+    account  = ActivityPub::LinkedDataSignature.new(@json).verify_account!
+    @account = account unless account.nil?
+  end
 end
diff --git a/app/services/authorize_follow_service.rb b/app/services/authorize_follow_service.rb
index 6f036dc5a..b1bff8962 100644
--- a/app/services/authorize_follow_service.rb
+++ b/app/services/authorize_follow_service.rb
@@ -24,11 +24,11 @@ class AuthorizeFollowService < BaseService
   end
 
   def build_json(follow_request)
-    ActiveModelSerializers::SerializableResource.new(
+    Oj.dump(ActivityPub::LinkedDataSignature.new(ActiveModelSerializers::SerializableResource.new(
       follow_request,
       serializer: ActivityPub::AcceptFollowSerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json).sign!(follow_request.target_account))
   end
 
   def build_xml(follow_request)
diff --git a/app/services/batched_remove_status_service.rb b/app/services/batched_remove_status_service.rb
index e6c8c9208..c90f4401d 100644
--- a/app/services/batched_remove_status_service.rb
+++ b/app/services/batched_remove_status_service.rb
@@ -138,10 +138,14 @@ class BatchedRemoveStatusService < BaseService
   def build_json(status)
     return @activity_json[status.id] if @activity_json.key?(status.id)
 
-    @activity_json[status.id] = ActiveModelSerializers::SerializableResource.new(
+    @activity_json[status.id] = sign_json(status, ActiveModelSerializers::SerializableResource.new(
       status,
       serializer: ActivityPub::DeleteSerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json)
+  end
+
+  def sign_json(status, json)
+    Oj.dump(ActivityPub::LinkedDataSignature.new(json).sign!(status.account))
   end
 end
diff --git a/app/services/block_service.rb b/app/services/block_service.rb
index f2253226b..b39c3eef2 100644
--- a/app/services/block_service.rb
+++ b/app/services/block_service.rb
@@ -27,11 +27,11 @@ class BlockService < BaseService
   end
 
   def build_json(block)
-    ActiveModelSerializers::SerializableResource.new(
+    Oj.dump(ActivityPub::LinkedDataSignature.new(ActiveModelSerializers::SerializableResource.new(
       block,
       serializer: ActivityPub::BlockSerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json).sign!(block.account))
   end
 
   def build_xml(block)
diff --git a/app/services/favourite_service.rb b/app/services/favourite_service.rb
index 4aa935170..44df3ed13 100644
--- a/app/services/favourite_service.rb
+++ b/app/services/favourite_service.rb
@@ -34,11 +34,11 @@ class FavouriteService < BaseService
   end
 
   def build_json(favourite)
-    ActiveModelSerializers::SerializableResource.new(
+    Oj.dump(ActivityPub::LinkedDataSignature.new(ActiveModelSerializers::SerializableResource.new(
       favourite,
       serializer: ActivityPub::LikeSerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json).sign!(favourite.account))
   end
 
   def build_xml(favourite)
diff --git a/app/services/follow_service.rb b/app/services/follow_service.rb
index 2be625cd8..a92eb6b88 100644
--- a/app/services/follow_service.rb
+++ b/app/services/follow_service.rb
@@ -67,10 +67,10 @@ class FollowService < BaseService
   end
 
   def build_json(follow_request)
-    ActiveModelSerializers::SerializableResource.new(
+    Oj.dump(ActivityPub::LinkedDataSignature.new(ActiveModelSerializers::SerializableResource.new(
       follow_request,
       serializer: ActivityPub::FollowSerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json).sign!(follow_request.account))
   end
 end
diff --git a/app/services/process_mentions_service.rb b/app/services/process_mentions_service.rb
index 2b8a77147..f123bf869 100644
--- a/app/services/process_mentions_service.rb
+++ b/app/services/process_mentions_service.rb
@@ -47,11 +47,11 @@ class ProcessMentionsService < BaseService
   end
 
   def build_json(status)
-    ActiveModelSerializers::SerializableResource.new(
+    Oj.dump(ActivityPub::LinkedDataSignature.new(ActiveModelSerializers::SerializableResource.new(
       status,
       serializer: ActivityPub::ActivitySerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json).sign!(status.account))
   end
 
   def follow_remote_account_service
diff --git a/app/services/reblog_service.rb b/app/services/reblog_service.rb
index 7f886af7c..5ed16c64b 100644
--- a/app/services/reblog_service.rb
+++ b/app/services/reblog_service.rb
@@ -42,10 +42,10 @@ class ReblogService < BaseService
   end
 
   def build_json(reblog)
-    ActiveModelSerializers::SerializableResource.new(
+    Oj.dump(ActivityPub::LinkedDataSignature.new(ActiveModelSerializers::SerializableResource.new(
       reblog,
       serializer: ActivityPub::ActivitySerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json).sign!(reblog.account))
   end
 end
diff --git a/app/services/reject_follow_service.rb b/app/services/reject_follow_service.rb
index a91266aa4..c1f7bcb60 100644
--- a/app/services/reject_follow_service.rb
+++ b/app/services/reject_follow_service.rb
@@ -19,11 +19,11 @@ class RejectFollowService < BaseService
   end
 
   def build_json(follow_request)
-    ActiveModelSerializers::SerializableResource.new(
+    Oj.dump(ActivityPub::LinkedDataSignature.new(ActiveModelSerializers::SerializableResource.new(
       follow_request,
       serializer: ActivityPub::RejectFollowSerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json).sign!(follow_request.target_account))
   end
 
   def build_xml(follow_request)
diff --git a/app/services/remove_status_service.rb b/app/services/remove_status_service.rb
index fcccbaa24..62eea677f 100644
--- a/app/services/remove_status_service.rb
+++ b/app/services/remove_status_service.rb
@@ -56,7 +56,7 @@ class RemoveStatusService < BaseService
 
     # ActivityPub
     ActivityPub::DeliveryWorker.push_bulk(target_accounts.select(&:activitypub?).uniq(&:inbox_url)) do |inbox_url|
-      [activity_json, @account.id, inbox_url]
+      [signed_activity_json, @account.id, inbox_url]
     end
   end
 
@@ -66,7 +66,7 @@ class RemoveStatusService < BaseService
 
     # ActivityPub
     ActivityPub::DeliveryWorker.push_bulk(@account.followers.inboxes) do |inbox_url|
-      [activity_json, @account.id, inbox_url]
+      [signed_activity_json, @account.id, inbox_url]
     end
   end
 
@@ -74,12 +74,16 @@ class RemoveStatusService < BaseService
     @salmon_xml ||= stream_entry_to_xml(@stream_entry)
   end
 
+  def signed_activity_json
+    @signed_activity_json ||= Oj.dump(ActivityPub::LinkedDataSignature.new(activity_json).sign!(@account))
+  end
+
   def activity_json
     @activity_json ||= ActiveModelSerializers::SerializableResource.new(
       @status,
       serializer: ActivityPub::DeleteSerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json
   end
 
   def remove_reblogs
diff --git a/app/services/unblock_service.rb b/app/services/unblock_service.rb
index 72fc5ab15..869f62d1c 100644
--- a/app/services/unblock_service.rb
+++ b/app/services/unblock_service.rb
@@ -20,11 +20,11 @@ class UnblockService < BaseService
   end
 
   def build_json(unblock)
-    ActiveModelSerializers::SerializableResource.new(
+    Oj.dump(ActivityPub::LinkedDataSignature.new(ActiveModelSerializers::SerializableResource.new(
       unblock,
       serializer: ActivityPub::UndoBlockSerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json).sign!(unblock.account))
   end
 
   def build_xml(block)
diff --git a/app/services/unfavourite_service.rb b/app/services/unfavourite_service.rb
index e53798e66..2fda11bd6 100644
--- a/app/services/unfavourite_service.rb
+++ b/app/services/unfavourite_service.rb
@@ -21,11 +21,11 @@ class UnfavouriteService < BaseService
   end
 
   def build_json(favourite)
-    ActiveModelSerializers::SerializableResource.new(
+    Oj.dump(ActivityPub::LinkedDataSignature.new(ActiveModelSerializers::SerializableResource.new(
       favourite,
       serializer: ActivityPub::UndoLikeSerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json).sign!(favourite.account))
   end
 
   def build_xml(favourite)
diff --git a/app/services/unfollow_service.rb b/app/services/unfollow_service.rb
index 10af75146..bf151ee28 100644
--- a/app/services/unfollow_service.rb
+++ b/app/services/unfollow_service.rb
@@ -23,11 +23,11 @@ class UnfollowService < BaseService
   end
 
   def build_json(follow)
-    ActiveModelSerializers::SerializableResource.new(
+    Oj.dump(ActivityPub::LinkedDataSignature.new(ActiveModelSerializers::SerializableResource.new(
       follow,
       serializer: ActivityPub::UndoFollowSerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json).sign!(follow.account))
   end
 
   def build_xml(follow)
diff --git a/app/workers/activitypub/distribution_worker.rb b/app/workers/activitypub/distribution_worker.rb
index 004dd25d1..14bb933c0 100644
--- a/app/workers/activitypub/distribution_worker.rb
+++ b/app/workers/activitypub/distribution_worker.rb
@@ -12,7 +12,7 @@ class ActivityPub::DistributionWorker
     return if skip_distribution?
 
     ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
-      [payload, @account.id, inbox_url]
+      [signed_payload, @account.id, inbox_url]
     end
   rescue ActiveRecord::RecordNotFound
     true
@@ -28,11 +28,15 @@ class ActivityPub::DistributionWorker
     @inboxes ||= @account.followers.inboxes
   end
 
+  def signed_payload
+    @signed_payload ||= Oj.dump(ActivityPub::LinkedDataSignature.new(payload).sign!(@account))
+  end
+
   def payload
     @payload ||= ActiveModelSerializers::SerializableResource.new(
       @status,
       serializer: ActivityPub::ActivitySerializer,
       adapter: ActivityPub::Adapter
-    ).to_json
+    ).as_json
   end
 end
diff --git a/config/initializers/json_ld.rb b/config/initializers/json_ld.rb
new file mode 100644
index 000000000..408e6490d
--- /dev/null
+++ b/config/initializers/json_ld.rb
@@ -0,0 +1,4 @@
+# frozen_string_literal: true
+
+require_relative '../../lib/json_ld/identity'
+require_relative '../../lib/json_ld/security'
diff --git a/lib/json_ld/identity.rb b/lib/json_ld/identity.rb
new file mode 100644
index 000000000..cfe50b956
--- /dev/null
+++ b/lib/json_ld/identity.rb
@@ -0,0 +1,86 @@
+# -*- encoding: utf-8 -*-
+# frozen_string_literal: true
+# This file generated automatically from https://w3id.org/identity/v1
+require 'json/ld'
+class JSON::LD::Context
+  add_preloaded("https://w3id.org/identity/v1") do
+    new(processingMode: "json-ld-1.0", term_definitions: {
+      "Credential" => TermDefinition.new("Credential", id: "https://w3id.org/credentials#Credential", simple: true),
+      "CryptographicKey" => TermDefinition.new("CryptographicKey", id: "https://w3id.org/security#Key", simple: true),
+      "CryptographicKeyCredential" => TermDefinition.new("CryptographicKeyCredential", id: "https://w3id.org/credentials#CryptographicKeyCredential", simple: true),
+      "EncryptedMessage" => TermDefinition.new("EncryptedMessage", id: "https://w3id.org/security#EncryptedMessage", simple: true),
+      "GraphSignature2012" => TermDefinition.new("GraphSignature2012", id: "https://w3id.org/security#GraphSignature2012", simple: true),
+      "Group" => TermDefinition.new("Group", id: "https://www.w3.org/ns/activitystreams#Group", simple: true),
+      "Identity" => TermDefinition.new("Identity", id: "https://w3id.org/identity#Identity", simple: true),
+      "LinkedDataSignature2015" => TermDefinition.new("LinkedDataSignature2015", id: "https://w3id.org/security#LinkedDataSignature2015", simple: true),
+      "Organization" => TermDefinition.new("Organization", id: "http://schema.org/Organization", simple: true),
+      "Person" => TermDefinition.new("Person", id: "http://schema.org/Person", simple: true),
+      "PostalAddress" => TermDefinition.new("PostalAddress", id: "http://schema.org/PostalAddress", simple: true),
+      "about" => TermDefinition.new("about", id: "http://schema.org/about", type_mapping: "@id"),
+      "accessControl" => TermDefinition.new("accessControl", id: "https://w3id.org/permissions#accessControl", type_mapping: "@id"),
+      "address" => TermDefinition.new("address", id: "http://schema.org/address", type_mapping: "@id"),
+      "addressCountry" => TermDefinition.new("addressCountry", id: "http://schema.org/addressCountry", simple: true),
+      "addressLocality" => TermDefinition.new("addressLocality", id: "http://schema.org/addressLocality", simple: true),
+      "addressRegion" => TermDefinition.new("addressRegion", id: "http://schema.org/addressRegion", simple: true),
+      "cipherAlgorithm" => TermDefinition.new("cipherAlgorithm", id: "https://w3id.org/security#cipherAlgorithm", simple: true),
+      "cipherData" => TermDefinition.new("cipherData", id: "https://w3id.org/security#cipherData", simple: true),
+      "cipherKey" => TermDefinition.new("cipherKey", id: "https://w3id.org/security#cipherKey", simple: true),
+      "claim" => TermDefinition.new("claim", id: "https://w3id.org/credentials#claim", type_mapping: "@id"),
+      "comment" => TermDefinition.new("comment", id: "http://www.w3.org/2000/01/rdf-schema#comment", simple: true),
+      "created" => TermDefinition.new("created", id: "http://purl.org/dc/terms/created", type_mapping: "http://www.w3.org/2001/XMLSchema#dateTime"),
+      "creator" => TermDefinition.new("creator", id: "http://purl.org/dc/terms/creator", type_mapping: "@id"),
+      "cred" => TermDefinition.new("cred", id: "https://w3id.org/credentials#", simple: true, prefix: true),
+      "credential" => TermDefinition.new("credential", id: "https://w3id.org/credentials#credential", type_mapping: "@id"),
+      "dc" => TermDefinition.new("dc", id: "http://purl.org/dc/terms/", simple: true, prefix: true),
+      "description" => TermDefinition.new("description", id: "http://schema.org/description", simple: true),
+      "digestAlgorithm" => TermDefinition.new("digestAlgorithm", id: "https://w3id.org/security#digestAlgorithm", simple: true),
+      "digestValue" => TermDefinition.new("digestValue", id: "https://w3id.org/security#digestValue", simple: true),
+      "domain" => TermDefinition.new("domain", id: "https://w3id.org/security#domain", simple: true),
+      "email" => TermDefinition.new("email", id: "http://schema.org/email", simple: true),
+      "expires" => TermDefinition.new("expires", id: "https://w3id.org/security#expiration", type_mapping: "http://www.w3.org/2001/XMLSchema#dateTime"),
+      "familyName" => TermDefinition.new("familyName", id: "http://schema.org/familyName", simple: true),
+      "givenName" => TermDefinition.new("givenName", id: "http://schema.org/givenName", simple: true),
+      "id" => TermDefinition.new("id", id: "@id", simple: true),
+      "identity" => TermDefinition.new("identity", id: "https://w3id.org/identity#", simple: true, prefix: true),
+      "identityService" => TermDefinition.new("identityService", id: "https://w3id.org/identity#identityService", type_mapping: "@id"),
+      "idp" => TermDefinition.new("idp", id: "https://w3id.org/identity#idp", type_mapping: "@id"),
+      "image" => TermDefinition.new("image", id: "http://schema.org/image", type_mapping: "@id"),
+      "initializationVector" => TermDefinition.new("initializationVector", id: "https://w3id.org/security#initializationVector", simple: true),
+      "issued" => TermDefinition.new("issued", id: "https://w3id.org/credentials#issued", type_mapping: "http://www.w3.org/2001/XMLSchema#dateTime"),
+      "issuer" => TermDefinition.new("issuer", id: "https://w3id.org/credentials#issuer", type_mapping: "@id"),
+      "label" => TermDefinition.new("label", id: "http://www.w3.org/2000/01/rdf-schema#label", simple: true),
+      "member" => TermDefinition.new("member", id: "http://schema.org/member", type_mapping: "@id"),
+      "memberOf" => TermDefinition.new("memberOf", id: "http://schema.org/memberOf", type_mapping: "@id"),
+      "name" => TermDefinition.new("name", id: "http://schema.org/name", simple: true),
+      "nonce" => TermDefinition.new("nonce", id: "https://w3id.org/security#nonce", simple: true),
+      "normalizationAlgorithm" => TermDefinition.new("normalizationAlgorithm", id: "https://w3id.org/security#normalizationAlgorithm", simple: true),
+      "owner" => TermDefinition.new("owner", id: "https://w3id.org/security#owner", type_mapping: "@id"),
+      "password" => TermDefinition.new("password", id: "https://w3id.org/security#password", simple: true),
+      "paymentProcessor" => TermDefinition.new("paymentProcessor", id: "https://w3id.org/payswarm#processor", simple: true),
+      "perm" => TermDefinition.new("perm", id: "https://w3id.org/permissions#", simple: true, prefix: true),
+      "postalCode" => TermDefinition.new("postalCode", id: "http://schema.org/postalCode", simple: true),
+      "preferences" => TermDefinition.new("preferences", id: "https://w3id.org/payswarm#preferences", type_mapping: "@vocab"),
+      "privateKey" => TermDefinition.new("privateKey", id: "https://w3id.org/security#privateKey", type_mapping: "@id"),
+      "privateKeyPem" => TermDefinition.new("privateKeyPem", id: "https://w3id.org/security#privateKeyPem", simple: true),
+      "ps" => TermDefinition.new("ps", id: "https://w3id.org/payswarm#", simple: true, prefix: true),
+      "publicKey" => TermDefinition.new("publicKey", id: "https://w3id.org/security#publicKey", type_mapping: "@id"),
+      "publicKeyPem" => TermDefinition.new("publicKeyPem", id: "https://w3id.org/security#publicKeyPem", simple: true),
+      "publicKeyService" => TermDefinition.new("publicKeyService", id: "https://w3id.org/security#publicKeyService", type_mapping: "@id"),
+      "rdf" => TermDefinition.new("rdf", id: "http://www.w3.org/1999/02/22-rdf-syntax-ns#", simple: true, prefix: true),
+      "rdfs" => TermDefinition.new("rdfs", id: "http://www.w3.org/2000/01/rdf-schema#", simple: true, prefix: true),
+      "recipient" => TermDefinition.new("recipient", id: "https://w3id.org/credentials#recipient", type_mapping: "@id"),
+      "revoked" => TermDefinition.new("revoked", id: "https://w3id.org/security#revoked", type_mapping: "http://www.w3.org/2001/XMLSchema#dateTime"),
+      "schema" => TermDefinition.new("schema", id: "http://schema.org/", simple: true, prefix: true),
+      "sec" => TermDefinition.new("sec", id: "https://w3id.org/security#", simple: true, prefix: true),
+      "signature" => TermDefinition.new("signature", id: "https://w3id.org/security#signature", simple: true),
+      "signatureAlgorithm" => TermDefinition.new("signatureAlgorithm", id: "https://w3id.org/security#signatureAlgorithm", simple: true),
+      "signatureValue" => TermDefinition.new("signatureValue", id: "https://w3id.org/security#signatureValue", simple: true),
+      "streetAddress" => TermDefinition.new("streetAddress", id: "http://schema.org/streetAddress", simple: true),
+      "title" => TermDefinition.new("title", id: "http://purl.org/dc/terms/title", simple: true),
+      "type" => TermDefinition.new("type", id: "@type", simple: true),
+      "url" => TermDefinition.new("url", id: "http://schema.org/url", type_mapping: "@id"),
+      "writePermission" => TermDefinition.new("writePermission", id: "https://w3id.org/permissions#writePermission", type_mapping: "@id"),
+      "xsd" => TermDefinition.new("xsd", id: "http://www.w3.org/2001/XMLSchema#", simple: true, prefix: true)
+    })
+  end
+end
diff --git a/lib/json_ld/security.rb b/lib/json_ld/security.rb
new file mode 100644
index 000000000..1230206f0
--- /dev/null
+++ b/lib/json_ld/security.rb
@@ -0,0 +1,50 @@
+# -*- encoding: utf-8 -*-
+# frozen_string_literal: true
+# This file generated automatically from https://w3id.org/security/v1
+require 'json/ld'
+class JSON::LD::Context
+  add_preloaded("https://w3id.org/security/v1") do
+    new(processingMode: "json-ld-1.0", term_definitions: {
+      "CryptographicKey" => TermDefinition.new("CryptographicKey", id: "https://w3id.org/security#Key", simple: true),
+      "EcdsaKoblitzSignature2016" => TermDefinition.new("EcdsaKoblitzSignature2016", id: "https://w3id.org/security#EcdsaKoblitzSignature2016", simple: true),
+      "EncryptedMessage" => TermDefinition.new("EncryptedMessage", id: "https://w3id.org/security#EncryptedMessage", simple: true),
+      "GraphSignature2012" => TermDefinition.new("GraphSignature2012", id: "https://w3id.org/security#GraphSignature2012", simple: true),
+      "LinkedDataSignature2015" => TermDefinition.new("LinkedDataSignature2015", id: "https://w3id.org/security#LinkedDataSignature2015", simple: true),
+      "LinkedDataSignature2016" => TermDefinition.new("LinkedDataSignature2016", id: "https://w3id.org/security#LinkedDataSignature2016", simple: true),
+      "authenticationTag" => TermDefinition.new("authenticationTag", id: "https://w3id.org/security#authenticationTag", simple: true),
+      "canonicalizationAlgorithm" => TermDefinition.new("canonicalizationAlgorithm", id: "https://w3id.org/security#canonicalizationAlgorithm", simple: true),
+      "cipherAlgorithm" => TermDefinition.new("cipherAlgorithm", id: "https://w3id.org/security#cipherAlgorithm", simple: true),
+      "cipherData" => TermDefinition.new("cipherData", id: "https://w3id.org/security#cipherData", simple: true),
+      "cipherKey" => TermDefinition.new("cipherKey", id: "https://w3id.org/security#cipherKey", simple: true),
+      "created" => TermDefinition.new("created", id: "http://purl.org/dc/terms/created", type_mapping: "http://www.w3.org/2001/XMLSchema#dateTime"),
+      "creator" => TermDefinition.new("creator", id: "http://purl.org/dc/terms/creator", type_mapping: "@id"),
+      "dc" => TermDefinition.new("dc", id: "http://purl.org/dc/terms/", simple: true, prefix: true),
+      "digestAlgorithm" => TermDefinition.new("digestAlgorithm", id: "https://w3id.org/security#digestAlgorithm", simple: true),
+      "digestValue" => TermDefinition.new("digestValue", id: "https://w3id.org/security#digestValue", simple: true),
+      "domain" => TermDefinition.new("domain", id: "https://w3id.org/security#domain", simple: true),
+      "encryptionKey" => TermDefinition.new("encryptionKey", id: "https://w3id.org/security#encryptionKey", simple: true),
+      "expiration" => TermDefinition.new("expiration", id: "https://w3id.org/security#expiration", type_mapping: "http://www.w3.org/2001/XMLSchema#dateTime"),
+      "expires" => TermDefinition.new("expires", id: "https://w3id.org/security#expiration", type_mapping: "http://www.w3.org/2001/XMLSchema#dateTime"),
+      "id" => TermDefinition.new("id", id: "@id", simple: true),
+      "initializationVector" => TermDefinition.new("initializationVector", id: "https://w3id.org/security#initializationVector", simple: true),
+      "iterationCount" => TermDefinition.new("iterationCount", id: "https://w3id.org/security#iterationCount", simple: true),
+      "nonce" => TermDefinition.new("nonce", id: "https://w3id.org/security#nonce", simple: true),
+      "normalizationAlgorithm" => TermDefinition.new("normalizationAlgorithm", id: "https://w3id.org/security#normalizationAlgorithm", simple: true),
+      "owner" => TermDefinition.new("owner", id: "https://w3id.org/security#owner", type_mapping: "@id"),
+      "password" => TermDefinition.new("password", id: "https://w3id.org/security#password", simple: true),
+      "privateKey" => TermDefinition.new("privateKey", id: "https://w3id.org/security#privateKey", type_mapping: "@id"),
+      "privateKeyPem" => TermDefinition.new("privateKeyPem", id: "https://w3id.org/security#privateKeyPem", simple: true),
+      "publicKey" => TermDefinition.new("publicKey", id: "https://w3id.org/security#publicKey", type_mapping: "@id"),
+      "publicKeyPem" => TermDefinition.new("publicKeyPem", id: "https://w3id.org/security#publicKeyPem", simple: true),
+      "publicKeyService" => TermDefinition.new("publicKeyService", id: "https://w3id.org/security#publicKeyService", type_mapping: "@id"),
+      "revoked" => TermDefinition.new("revoked", id: "https://w3id.org/security#revoked", type_mapping: "http://www.w3.org/2001/XMLSchema#dateTime"),
+      "salt" => TermDefinition.new("salt", id: "https://w3id.org/security#salt", simple: true),
+      "sec" => TermDefinition.new("sec", id: "https://w3id.org/security#", simple: true, prefix: true),
+      "signature" => TermDefinition.new("signature", id: "https://w3id.org/security#signature", simple: true),
+      "signatureAlgorithm" => TermDefinition.new("signatureAlgorithm", id: "https://w3id.org/security#signingAlgorithm", simple: true),
+      "signatureValue" => TermDefinition.new("signatureValue", id: "https://w3id.org/security#signatureValue", simple: true),
+      "type" => TermDefinition.new("type", id: "@type", simple: true),
+      "xsd" => TermDefinition.new("xsd", id: "http://www.w3.org/2001/XMLSchema#", simple: true, prefix: true)
+    })
+  end
+end
diff --git a/spec/lib/activitypub/linked_data_signature_spec.rb b/spec/lib/activitypub/linked_data_signature_spec.rb
new file mode 100644
index 000000000..ee4b68028
--- /dev/null
+++ b/spec/lib/activitypub/linked_data_signature_spec.rb
@@ -0,0 +1,86 @@
+require 'rails_helper'
+
+RSpec.describe ActivityPub::LinkedDataSignature do
+  include JsonLdHelper
+
+  let!(:sender) { Fabricate(:account, uri: 'http://example.com/alice') }
+
+  let(:raw_json) do
+    {
+      '@context' => 'https://www.w3.org/ns/activitystreams',
+      'id' => 'http://example.com/hello-world',
+    }
+  end
+
+  let(:json) { raw_json.merge('signature' => signature) }
+
+  subject { described_class.new(json) }
+
+  describe '#verify_account!' do
+    context 'when signature matches' do
+      let(:raw_signature) do
+        {
+          'creator' => 'http://example.com/alice',
+          'created' => '2017-09-23T20:21:34Z',
+        }
+      end
+
+      let(:signature) { raw_signature.merge('type' => 'RsaSignature2017', 'signatureValue' => sign(sender, raw_signature, raw_json)) }
+
+      it 'returns creator' do
+        expect(subject.verify_account!).to eq sender
+      end
+    end
+
+    context 'when signature is missing' do
+      let(:signature) { nil }
+
+      it 'returns nil' do
+        expect(subject.verify_account!).to be_nil
+      end
+    end
+
+    context 'when signature is tampered' do
+      let(:raw_signature) do
+        {
+          'creator' => 'http://example.com/alice',
+          'created' => '2017-09-23T20:21:34Z',
+        }
+      end
+
+      let(:signature) { raw_signature.merge('type' => 'RsaSignature2017', 'signatureValue' => 's69F3mfddd99dGjmvjdjjs81e12jn121Gkm1') }
+
+      it 'returns nil' do
+        expect(subject.verify_account!).to be_nil
+      end
+    end
+  end
+
+  describe '#sign!' do
+    subject { described_class.new(raw_json).sign!(sender) }
+
+    it 'returns a hash' do
+      expect(subject).to be_a Hash
+    end
+
+    it 'contains signature context' do
+      expect(subject['@context']).to include('https://www.w3.org/ns/activitystreams', 'https://w3id.org/identity/v1')
+    end
+
+    it 'contains signature' do
+      expect(subject['signature']).to be_a Hash
+      expect(subject['signature']['signatureValue']).to be_present
+    end
+
+    it 'can be verified again' do
+      expect(described_class.new(subject).verify_account!).to eq sender
+    end
+  end
+
+  def sign(from_account, options, document)
+    options_hash   = Digest::SHA256.hexdigest(canonicalize(options.merge('@context' => ActivityPub::LinkedDataSignature::CONTEXT)))
+    document_hash  = Digest::SHA256.hexdigest(canonicalize(document))
+    to_be_verified = options_hash + document_hash
+    Base64.strict_encode64(from_account.keypair.sign(OpenSSL::Digest::SHA256.new, to_be_verified))
+  end
+end
diff --git a/spec/services/activitypub/process_collection_service_spec.rb b/spec/services/activitypub/process_collection_service_spec.rb
index 6486483f6..bf3bc82aa 100644
--- a/spec/services/activitypub/process_collection_service_spec.rb
+++ b/spec/services/activitypub/process_collection_service_spec.rb
@@ -1,9 +1,10 @@
 require 'rails_helper'
 
 RSpec.describe ActivityPub::ProcessCollectionService do
-  subject { ActivityPub::ProcessCollectionService.new }
+  subject { described_class.new }
 
   describe '#call' do
-    pending
+    context 'when actor is the sender'
+    context 'when actor differs from sender'
   end
 end
-- 
cgit 


From 0397c58b61d8a364ff488fc5631bfc751903c242 Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Sat, 26 Aug 2017 18:52:53 +0200
Subject: Forward ActivityPub deletes to followers of rebloggers (#4706)

---
 app/lib/activitypub/activity/delete.rb             | 19 +++++++++++++++++-
 app/workers/activitypub/raw_distribution_worker.rb | 23 ++++++++++++++++++++++
 2 files changed, 41 insertions(+), 1 deletion(-)
 create mode 100644 app/workers/activitypub/raw_distribution_worker.rb

(limited to 'app/workers')

diff --git a/app/lib/activitypub/activity/delete.rb b/app/lib/activitypub/activity/delete.rb
index a1598dfd3..e7eb53a02 100644
--- a/app/lib/activitypub/activity/delete.rb
+++ b/app/lib/activitypub/activity/delete.rb
@@ -8,7 +8,24 @@ class ActivityPub::Activity::Delete < ActivityPub::Activity
     if status.nil?
       delete_later!(object_uri)
     else
-      RemoveStatusService.new.call(status)
+      forward_for_reblogs(status)
+      delete_now!(status)
     end
   end
+
+  private
+
+  def forward_for_reblogs(status)
+    ActivityPub::RawDistributionWorker.push_bulk(status.reblogs.includes(:account).references(:account).merge(Account.local).pluck(:account_id)) do |account|
+      [payload, account.id]
+    end
+  end
+
+  def delete_now!(status)
+    RemoveStatusService.new.call(status)
+  end
+
+  def payload
+    @payload ||= Oj.dump(@json)
+  end
 end
diff --git a/app/workers/activitypub/raw_distribution_worker.rb b/app/workers/activitypub/raw_distribution_worker.rb
new file mode 100644
index 000000000..d73466f6e
--- /dev/null
+++ b/app/workers/activitypub/raw_distribution_worker.rb
@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+class ActivityPub::RawDistributionWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'push'
+
+  def perform(json, source_account_id)
+    @account = Account.find(source_account_id)
+
+    ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
+      [json, @account.id, inbox_url]
+    end
+  rescue ActiveRecord::RecordNotFound
+    true
+  end
+
+  private
+
+  def inboxes
+    @inboxes ||= @account.followers.inboxes
+  end
+end
-- 
cgit 


From 4c76402ba1d355061e7e208b7a2f8251388a38e1 Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Tue, 29 Aug 2017 16:11:05 +0200
Subject: Serialize ActivityPub alternate link into OStatus deletes, handle it
 (#4730)

Requires moving Atom rendering from DistributionWorker (where
`stream_entry.status` is already nil) to inline (where
`stream_entry.status.destroyed?` is true) and distributing that.

Unfortunately, such XML renderings can no longer be easily chained
together into one payload of n items.
---
 app/lib/ostatus/activity/deletion.rb               |  4 +++-
 app/lib/ostatus/atom_serializer.rb                 |  3 +++
 app/models/status.rb                               | 13 ++++++++++--
 app/services/batched_remove_status_service.rb      | 24 +++++++++++++---------
 app/services/remove_status_service.rb              |  4 +---
 .../pubsubhubbub/raw_distribution_worker.rb        | 22 ++++++++++++++++++++
 .../services/batched_remove_status_service_spec.rb |  7 +++----
 7 files changed, 57 insertions(+), 20 deletions(-)
 create mode 100644 app/workers/pubsubhubbub/raw_distribution_worker.rb

(limited to 'app/workers')

diff --git a/app/lib/ostatus/activity/deletion.rb b/app/lib/ostatus/activity/deletion.rb
index 860faf501..c98f5ee0a 100644
--- a/app/lib/ostatus/activity/deletion.rb
+++ b/app/lib/ostatus/activity/deletion.rb
@@ -3,7 +3,9 @@
 class OStatus::Activity::Deletion < OStatus::Activity::Base
   def perform
     Rails.logger.debug "Deleting remote status #{id}"
-    status = Status.find_by(uri: id, account: @account)
+
+    status   = Status.find_by(uri: id, account: @account)
+    status ||= Status.find_by(uri: activitypub_uri, account: @account) if activitypub_uri?
 
     if status.nil?
       redis.setex("delete_upon_arrival:#{@account.id}:#{id}", 6 * 3_600, id)
diff --git a/app/lib/ostatus/atom_serializer.rb b/app/lib/ostatus/atom_serializer.rb
index 92a16d228..81fae4140 100644
--- a/app/lib/ostatus/atom_serializer.rb
+++ b/app/lib/ostatus/atom_serializer.rb
@@ -79,6 +79,9 @@ class OStatus::AtomSerializer
 
     if stream_entry.status.nil?
       append_element(entry, 'content', 'Deleted status')
+    elsif stream_entry.status.destroyed?
+      append_element(entry, 'content', 'Deleted status')
+      append_element(entry, 'link', nil, rel: :alternate, type: 'application/activity+json', href: ActivityPub::TagManager.instance.uri_for(stream_entry.status)) if stream_entry.account.local?
     else
       serialize_status_attributes(entry, stream_entry.status)
     end
diff --git a/app/models/status.rb b/app/models/status.rb
index 3dc83ad1f..abd902cd7 100644
--- a/app/models/status.rb
+++ b/app/models/status.rb
@@ -51,6 +51,7 @@ class Status < ApplicationRecord
 
   has_one :notification, as: :activity, dependent: :destroy
   has_one :preview_card, dependent: :destroy
+  has_one :stream_entry, as: :activity, inverse_of: :status
 
   validates :uri, uniqueness: true, unless: :local?
   validates :text, presence: true, unless: :reblog?
@@ -90,7 +91,11 @@ class Status < ApplicationRecord
   end
 
   def verb
-    reblog? ? :share : :post
+    if destroyed?
+      :delete
+    else
+      reblog? ? :share : :post
+    end
   end
 
   def object_type
@@ -110,7 +115,11 @@ class Status < ApplicationRecord
   end
 
   def title
-    reblog? ? "#{account.acct} shared a status by #{reblog.account.acct}" : "New status by #{account.acct}"
+    if destroyed?
+      "#{account.acct} deleted status"
+    else
+      reblog? ? "#{account.acct} shared a status by #{reblog.account.acct}" : "New status by #{account.acct}"
+    end
   end
 
   def hidden?
diff --git a/app/services/batched_remove_status_service.rb b/app/services/batched_remove_status_service.rb
index e9e22298d..86eaa5735 100644
--- a/app/services/batched_remove_status_service.rb
+++ b/app/services/batched_remove_status_service.rb
@@ -20,9 +20,10 @@ class BatchedRemoveStatusService < BaseService
     @activity_json_batches = []
     @json_payloads         = statuses.map { |s| [s.id, Oj.dump(event: :delete, payload: s.id)] }.to_h
     @activity_json         = {}
+    @activity_xml          = {}
 
     # Ensure that rendered XML reflects destroyed state
-    Status.where(id: statuses.map(&:id)).in_batches.destroy_all
+    statuses.each(&:destroy)
 
     # Batch by source account
     statuses.group_by(&:account_id).each do |_, account_statuses|
@@ -31,7 +32,7 @@ class BatchedRemoveStatusService < BaseService
       unpush_from_home_timelines(account_statuses)
 
       if account.local?
-        batch_stream_entries(account_statuses)
+        batch_stream_entries(account, account_statuses)
         batch_activity_json(account, account_statuses)
       end
     end
@@ -42,18 +43,16 @@ class BatchedRemoveStatusService < BaseService
       batch_salmon_slaps(status) if status.local?
     end
 
-    Pubsubhubbub::DistributionWorker.push_bulk(@stream_entry_batches) { |batch| batch }
+    Pubsubhubbub::RawDistributionWorker.push_bulk(@stream_entry_batches) { |batch| batch }
     NotificationWorker.push_bulk(@salmon_batches) { |batch| batch }
     ActivityPub::DeliveryWorker.push_bulk(@activity_json_batches) { |batch| batch }
   end
 
   private
 
-  def batch_stream_entries(statuses)
-    stream_entry_ids = statuses.map { |s| s.stream_entry.id }
-
-    stream_entry_ids.each_slice(100) do |batch_of_stream_entry_ids|
-      @stream_entry_batches << [batch_of_stream_entry_ids]
+  def batch_stream_entries(account, statuses)
+    statuses.each do |status|
+      @stream_entry_batches << [build_xml(status.stream_entry), account.id]
     end
   end
 
@@ -101,11 +100,10 @@ class BatchedRemoveStatusService < BaseService
   def batch_salmon_slaps(status)
     return if @mentions[status.id].empty?
 
-    payload    = stream_entry_to_xml(status.stream_entry.reload)
     recipients = @mentions[status.id].map(&:account).reject(&:local?).select(&:ostatus?).uniq(&:domain).map(&:id)
 
     recipients.each do |recipient_id|
-      @salmon_batches << [payload, status.account_id, recipient_id]
+      @salmon_batches << [build_xml(status.stream_entry), status.account_id, recipient_id]
     end
   end
 
@@ -145,6 +143,12 @@ class BatchedRemoveStatusService < BaseService
     ).as_json)
   end
 
+  def build_xml(stream_entry)
+    return @activity_xml[stream_entry.id] if @activity_xml.key?(stream_entry.id)
+
+    @activity_xml[stream_entry.id] = stream_entry_to_xml(stream_entry)
+  end
+
   def sign_json(status, json)
     Oj.dump(ActivityPub::LinkedDataSignature.new(json).sign!(status.account))
   end
diff --git a/app/services/remove_status_service.rb b/app/services/remove_status_service.rb
index 7ddbd8906..83fc77043 100644
--- a/app/services/remove_status_service.rb
+++ b/app/services/remove_status_service.rb
@@ -22,8 +22,6 @@ class RemoveStatusService < BaseService
 
     return unless @account.local?
 
-    @stream_entry = @stream_entry.reload
-
     remove_from_remote_followers
     remove_from_remote_affected
   end
@@ -62,7 +60,7 @@ class RemoveStatusService < BaseService
 
   def remove_from_remote_followers
     # OStatus
-    Pubsubhubbub::DistributionWorker.perform_async(@stream_entry.id)
+    Pubsubhubbub::RawDistributionWorker.perform_async(salmon_xml, @account.id)
 
     # ActivityPub
     ActivityPub::DeliveryWorker.push_bulk(@account.followers.inboxes) do |inbox_url|
diff --git a/app/workers/pubsubhubbub/raw_distribution_worker.rb b/app/workers/pubsubhubbub/raw_distribution_worker.rb
new file mode 100644
index 000000000..16962a623
--- /dev/null
+++ b/app/workers/pubsubhubbub/raw_distribution_worker.rb
@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+class Pubsubhubbub::RawDistributionWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'push'
+
+  def perform(xml, source_account_id)
+    @account       = Account.find(source_account_id)
+    @subscriptions = active_subscriptions.to_a
+
+    Pubsubhubbub::DeliveryWorker.push_bulk(@subscriptions) do |subscription|
+      [subscription.id, xml]
+    end
+  end
+
+  private
+
+  def active_subscriptions
+    Subscription.where(account: @account).active.select('id, callback_url, domain')
+  end
+end
diff --git a/spec/services/batched_remove_status_service_spec.rb b/spec/services/batched_remove_status_service_spec.rb
index 2484d4b58..b1e9ac567 100644
--- a/spec/services/batched_remove_status_service_spec.rb
+++ b/spec/services/batched_remove_status_service_spec.rb
@@ -48,11 +48,10 @@ RSpec.describe BatchedRemoveStatusService do
     expect(Redis.current).to have_received(:publish).with('timeline:public', any_args).at_least(:once)
   end
 
-  it 'sends PuSH update to PuSH subscribers with two payloads united' do
+  it 'sends PuSH update to PuSH subscribers' do
     expect(a_request(:post, 'http://example.com/push').with { |req|
-      matches = req.body.scan(TagManager::VERBS[:delete])
-      matches.size == 2
-    }).to have_been_made
+      matches = req.body.match(TagManager::VERBS[:delete])
+    }).to have_been_made.at_least_once
   end
 
   it 'sends Salmon slap to previously mentioned users' do
-- 
cgit 


From 7b8f26284072120701289f90bc6602ce918e4304 Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Wed, 30 Aug 2017 15:37:02 +0200
Subject: Forward ActivityPub creates that reply to local statuses (#4709)

* Forward ActivityPub creates that reply to local statuses

* Fix test

* Fix wrong signers
---
 app/lib/activitypub/activity/create.rb             | 10 ++++++
 app/lib/activitypub/activity/delete.rb             |  2 ++
 app/services/post_status_service.rb                |  1 +
 .../activitypub/reply_distribution_worker.rb       | 42 ++++++++++++++++++++++
 spec/lib/activitypub/activity/delete_spec.rb       |  1 +
 5 files changed, 56 insertions(+)
 create mode 100644 app/workers/activitypub/reply_distribution_worker.rb

(limited to 'app/workers')

diff --git a/app/lib/activitypub/activity/create.rb b/app/lib/activitypub/activity/create.rb
index 114aed84f..2eea1827a 100644
--- a/app/lib/activitypub/activity/create.rb
+++ b/app/lib/activitypub/activity/create.rb
@@ -17,6 +17,7 @@ class ActivityPub::Activity::Create < ActivityPub::Activity
 
     resolve_thread(status)
     distribute(status)
+    forward_for_reply if status.public_visibility? || status.unlisted_visibility?
 
     status
   end
@@ -162,4 +163,13 @@ class ActivityPub::Activity::Create < ActivityPub::Activity
     return @skip_download if defined?(@skip_download)
     @skip_download ||= DomainBlock.find_by(domain: @account.domain)&.reject_media?
   end
+
+  def reply_to_local?
+    !replied_to_status.nil? && replied_to_status.account.local?
+  end
+
+  def forward_for_reply
+    return unless @json['signature'].present? && reply_to_local?
+    ActivityPub::RawDistributionWorker.perform_async(Oj.dump(@json), replied_to_status.account_id)
+  end
 end
diff --git a/app/lib/activitypub/activity/delete.rb b/app/lib/activitypub/activity/delete.rb
index 789ed58f1..afa9a8079 100644
--- a/app/lib/activitypub/activity/delete.rb
+++ b/app/lib/activitypub/activity/delete.rb
@@ -16,6 +16,8 @@ class ActivityPub::Activity::Delete < ActivityPub::Activity
   private
 
   def forward_for_reblogs(status)
+    return if @json['signature'].blank?
+
     ActivityPub::RawDistributionWorker.push_bulk(status.reblogs.includes(:account).references(:account).merge(Account.local).pluck(:account_id)) do |account_id|
       [payload, account_id]
     end
diff --git a/app/services/post_status_service.rb b/app/services/post_status_service.rb
index 5ff93f21e..568f5a9e7 100644
--- a/app/services/post_status_service.rb
+++ b/app/services/post_status_service.rb
@@ -40,6 +40,7 @@ class PostStatusService < BaseService
     DistributionWorker.perform_async(status.id)
     Pubsubhubbub::DistributionWorker.perform_async(status.stream_entry.id)
     ActivityPub::DistributionWorker.perform_async(status.id)
+    ActivityPub::ReplyDistributionWorker.perform_async(status.id) if status.reply? && status.thread.account.local?
 
     if options[:idempotency].present?
       redis.setex("idempotency:status:#{account.id}:#{options[:idempotency]}", 3_600, status.id)
diff --git a/app/workers/activitypub/reply_distribution_worker.rb b/app/workers/activitypub/reply_distribution_worker.rb
new file mode 100644
index 000000000..f9127340f
--- /dev/null
+++ b/app/workers/activitypub/reply_distribution_worker.rb
@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+
+class ActivityPub::ReplyDistributionWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'push'
+
+  def perform(status_id)
+    @status  = Status.find(status_id)
+    @account = @status.thread.account
+
+    return if skip_distribution?
+
+    ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
+      [signed_payload, @status.account_id, inbox_url]
+    end
+  rescue ActiveRecord::RecordNotFound
+    true
+  end
+
+  private
+
+  def skip_distribution?
+    @status.private_visibility? || @status.direct_visibility?
+  end
+
+  def inboxes
+    @inboxes ||= @account.followers.inboxes
+  end
+
+  def signed_payload
+    @signed_payload ||= Oj.dump(ActivityPub::LinkedDataSignature.new(payload).sign!(@status.account))
+  end
+
+  def payload
+    @payload ||= ActiveModelSerializers::SerializableResource.new(
+      @status,
+      serializer: ActivityPub::ActivitySerializer,
+      adapter: ActivityPub::Adapter
+    ).as_json
+  end
+end
diff --git a/spec/lib/activitypub/activity/delete_spec.rb b/spec/lib/activitypub/activity/delete_spec.rb
index 6601f7262..65e743abb 100644
--- a/spec/lib/activitypub/activity/delete_spec.rb
+++ b/spec/lib/activitypub/activity/delete_spec.rb
@@ -11,6 +11,7 @@ RSpec.describe ActivityPub::Activity::Delete do
       type: 'Delete',
       actor: ActivityPub::TagManager.instance.uri_for(sender),
       object: ActivityPub::TagManager.instance.uri_for(status),
+      signature: 'foo',
     }.with_indifferent_access
   end
 
-- 
cgit