Merge pull request from GHSA-7w3c-p9j8-mq3x

* Ensure destruction of OAuth Applications notifies streaming

Due to doorkeeper using a dependent: delete_all relationship, the destroy of an OAuth Application bypassed the existing AccessTokenExtension callbacks for announcing destructing of access tokens.

* Ensure password resets revoke access to Streaming API

* Improve performance of deleting OAuth tokens

---------

Co-authored-by: Emelia Smith <ThisIsMissEm@users.noreply.github.com>

3 files changed, 47 insertions, 6 deletions

diff --git a/app/lib/application_extension.rb b/app/lib/application_extension.rb
index 4de69c1eadd..d1222656b75 100644
--- a/app/lib/application_extension.rb
+++ b/app/lib/application_extension.rb
@@ -4,12 +4,32 @@ module ApplicationExtension
   extend ActiveSupport::Concern
 
   included do
+    include Redisable
+
     validates :name, length: { maximum: 60 }
     validates :website, url: true, length: { maximum: 2_000 }, if: :website?
     validates :redirect_uri, length: { maximum: 2_000 }
+
+    # The relationship used between Applications and AccessTokens is using
+    # dependent: delete_all, which means the ActiveRecord callback in
+    # AccessTokenExtension is not run, so instead we manually announce to
+    # streaming that these tokens are being deleted.
+    before_destroy :push_to_streaming_api, prepend: true
   end
 
   def confirmation_redirect_uri
     redirect_uri.lines.first.strip
   end
+
+  def push_to_streaming_api
+    # TODO: #28793 Combine into a single topic
+    payload = Oj.dump(event: :kill)
+    access_tokens.in_batches do |tokens|
+      redis.pipelined do |pipeline|
+        tokens.ids.each do |id|
+          pipeline.publish("timeline:access_token:#{id}", payload)
+        end
+      end
+    end
+  end
 end
diff --git a/app/models/user.rb b/app/models/user.rb
index 5790be6e91d..641850a1072 100644
--- a/app/models/user.rb
+++ b/app/models/user.rb
@@ -373,6 +373,25 @@ class User < ApplicationRecord
     super
   end
 
+  def revoke_access!
+    Doorkeeper::AccessGrant.by_resource_owner(self).update_all(revoked_at: Time.now.utc)
+
+    Doorkeeper::AccessToken.by_resource_owner(self).in_batches do |batch|
+      batch.update_all(revoked_at: Time.now.utc)
+      Web::PushSubscription.where(access_token_id: batch).delete_all
+
+      # Revoke each access token for the Streaming API, since `update_all``
+      # doesn't trigger ActiveRecord Callbacks:
+      # TODO: #28793 Combine into a single topic
+      payload = Oj.dump(event: :kill)
+      redis.pipelined do |pipeline|
+        batch.ids.each do |id|
+          pipeline.publish("timeline:access_token:#{id}", payload)
+        end
+      end
+    end
+  end
+
   def reset_password!
     # First, change password to something random and deactivate all sessions
     transaction do
@@ -381,12 +400,7 @@ class User < ApplicationRecord
     end
 
     # Then, remove all authorized applications and connected push subscriptions
-    Doorkeeper::AccessGrant.by_resource_owner(self).in_batches.update_all(revoked_at: Time.now.utc)
-
-    Doorkeeper::AccessToken.by_resource_owner(self).in_batches do |batch|
-      batch.update_all(revoked_at: Time.now.utc)
-      Web::PushSubscription.where(access_token_id: batch).delete_all
-    end
+    revoke_access!
 
     # Finally, send a reset password prompt to the user
     send_reset_password_instructions
diff --git a/spec/models/user_spec.rb b/spec/models/user_spec.rb
index a7da31e6064..0d6f1607079 100644
--- a/spec/models/user_spec.rb
+++ b/spec/models/user_spec.rb
@@ -313,7 +313,10 @@ RSpec.describe User, type: :model do
     let!(:access_token) { Fabricate(:access_token, resource_owner_id: user.id) }
     let!(:web_push_subscription) { Fabricate(:web_push_subscription, access_token: access_token) }
 
+    let(:redis_pipeline_stub) { instance_double(Redis::Namespace, publish: nil) }
+
     before do
+      allow(redis).to receive(:pipelined).and_yield(redis_pipeline_stub)
       user.reset_password!
     end
 
@@ -329,6 +332,10 @@ RSpec.describe User, type: :model do
       expect(Doorkeeper::AccessToken.active_for(user).count).to eq 0
     end
 
+    it 'revokes streaming access for all access tokens' do
+      expect(redis_pipeline_stub).to have_received(:publish).with("timeline:access_token:#{access_token.id}", Oj.dump(event: :kill)).once
+    end
+
     it 'removes push subscriptions' do
       expect(Web::PushSubscription.where(user: user).or(Web::PushSubscription.where(access_token: access_token)).count).to eq 0
     end