forked from treehouse/mastodon
Merge pull request #1465 from ThibG/glitch-soc/merge-upstream
Merge upstream changesrebase/4.0.0rc2
commit
db01f8b942
|
@ -103,7 +103,7 @@ class Api::BaseController < ApplicationController
|
||||||
elsif !current_user.functional?
|
elsif !current_user.functional?
|
||||||
render json: { error: 'Your login is currently disabled' }, status: 403
|
render json: { error: 'Your login is currently disabled' }, status: 403
|
||||||
else
|
else
|
||||||
set_user_activity
|
update_user_sign_in
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|
|
@ -7,6 +7,7 @@ class Auth::SessionsController < Devise::SessionsController
|
||||||
|
|
||||||
skip_before_action :require_no_authentication, only: [:create]
|
skip_before_action :require_no_authentication, only: [:create]
|
||||||
skip_before_action :require_functional!
|
skip_before_action :require_functional!
|
||||||
|
skip_before_action :update_user_sign_in
|
||||||
|
|
||||||
prepend_before_action :set_pack
|
prepend_before_action :set_pack
|
||||||
|
|
||||||
|
@ -26,6 +27,7 @@ class Auth::SessionsController < Devise::SessionsController
|
||||||
|
|
||||||
def create
|
def create
|
||||||
super do |resource|
|
super do |resource|
|
||||||
|
resource.update_sign_in!(request, new_sign_in: true)
|
||||||
remember_me(resource)
|
remember_me(resource)
|
||||||
flash.delete(:notice)
|
flash.delete(:notice)
|
||||||
end
|
end
|
||||||
|
@ -59,7 +61,7 @@ class Auth::SessionsController < Devise::SessionsController
|
||||||
|
|
||||||
def find_user
|
def find_user
|
||||||
if session[:attempt_user_id]
|
if session[:attempt_user_id]
|
||||||
User.find(session[:attempt_user_id])
|
User.find_by(id: session[:attempt_user_id])
|
||||||
else
|
else
|
||||||
user = User.authenticate_with_ldap(user_params) if Devise.ldap_authentication
|
user = User.authenticate_with_ldap(user_params) if Devise.ldap_authentication
|
||||||
user ||= User.authenticate_with_pam(user_params) if Devise.pam_authentication
|
user ||= User.authenticate_with_pam(user_params) if Devise.pam_authentication
|
||||||
|
@ -92,6 +94,7 @@ class Auth::SessionsController < Devise::SessionsController
|
||||||
|
|
||||||
def require_no_authentication
|
def require_no_authentication
|
||||||
super
|
super
|
||||||
|
|
||||||
# Delete flash message that isn't entirely useful and may be confusing in
|
# Delete flash message that isn't entirely useful and may be confusing in
|
||||||
# most cases because /web doesn't display/clear flash messages.
|
# most cases because /web doesn't display/clear flash messages.
|
||||||
flash.delete(:alert) if flash[:alert] == I18n.t('devise.failure.already_authenticated')
|
flash.delete(:alert) if flash[:alert] == I18n.t('devise.failure.already_authenticated')
|
||||||
|
@ -113,13 +116,30 @@ class Auth::SessionsController < Devise::SessionsController
|
||||||
|
|
||||||
def home_paths(resource)
|
def home_paths(resource)
|
||||||
paths = [about_path]
|
paths = [about_path]
|
||||||
|
|
||||||
if single_user_mode? && resource.is_a?(User)
|
if single_user_mode? && resource.is_a?(User)
|
||||||
paths << short_account_path(username: resource.account)
|
paths << short_account_path(username: resource.account)
|
||||||
end
|
end
|
||||||
|
|
||||||
paths
|
paths
|
||||||
end
|
end
|
||||||
|
|
||||||
def continue_after?
|
def continue_after?
|
||||||
truthy_param?(:continue)
|
truthy_param?(:continue)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
def restart_session
|
||||||
|
clear_attempt_from_session
|
||||||
|
redirect_to new_user_session_path, alert: I18n.t('devise.failure.timeout')
|
||||||
|
end
|
||||||
|
|
||||||
|
def set_attempt_session(user)
|
||||||
|
session[:attempt_user_id] = user.id
|
||||||
|
session[:attempt_user_updated_at] = user.updated_at.to_s
|
||||||
|
end
|
||||||
|
|
||||||
|
def clear_attempt_from_session
|
||||||
|
session.delete(:attempt_user_id)
|
||||||
|
session.delete(:attempt_user_updated_at)
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
|
@ -18,7 +18,9 @@ module SignInTokenAuthenticationConcern
|
||||||
def authenticate_with_sign_in_token
|
def authenticate_with_sign_in_token
|
||||||
user = self.resource = find_user
|
user = self.resource = find_user
|
||||||
|
|
||||||
if user_params[:sign_in_token_attempt].present? && session[:attempt_user_id]
|
if user.present? && session[:attempt_user_id].present? && session[:attempt_user_updated_at] != user.updated_at.to_s
|
||||||
|
restart_session
|
||||||
|
elsif user_params.key?(:sign_in_token_attempt) && session[:attempt_user_id]
|
||||||
authenticate_with_sign_in_token_attempt(user)
|
authenticate_with_sign_in_token_attempt(user)
|
||||||
elsif user.present? && user.external_or_valid_password?(user_params[:password])
|
elsif user.present? && user.external_or_valid_password?(user_params[:password])
|
||||||
prompt_for_sign_in_token(user)
|
prompt_for_sign_in_token(user)
|
||||||
|
@ -27,7 +29,7 @@ module SignInTokenAuthenticationConcern
|
||||||
|
|
||||||
def authenticate_with_sign_in_token_attempt(user)
|
def authenticate_with_sign_in_token_attempt(user)
|
||||||
if valid_sign_in_token_attempt?(user)
|
if valid_sign_in_token_attempt?(user)
|
||||||
session.delete(:attempt_user_id)
|
clear_attempt_from_session
|
||||||
remember_me(user)
|
remember_me(user)
|
||||||
sign_in(user)
|
sign_in(user)
|
||||||
else
|
else
|
||||||
|
@ -42,11 +44,11 @@ module SignInTokenAuthenticationConcern
|
||||||
UserMailer.sign_in_token(user, request.remote_ip, request.user_agent, Time.now.utc.to_s).deliver_later!
|
UserMailer.sign_in_token(user, request.remote_ip, request.user_agent, Time.now.utc.to_s).deliver_later!
|
||||||
end
|
end
|
||||||
|
|
||||||
set_locale do
|
set_attempt_session(user)
|
||||||
session[:attempt_user_id] = user.id
|
|
||||||
use_pack 'auth'
|
use_pack 'auth'
|
||||||
|
|
||||||
@body_classes = 'lighter'
|
@body_classes = 'lighter'
|
||||||
render :sign_in_token
|
|
||||||
end
|
set_locale { render :sign_in_token }
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
|
@ -37,9 +37,11 @@ module TwoFactorAuthenticationConcern
|
||||||
def authenticate_with_two_factor
|
def authenticate_with_two_factor
|
||||||
user = self.resource = find_user
|
user = self.resource = find_user
|
||||||
|
|
||||||
if user.webauthn_enabled? && user_params[:credential].present? && session[:attempt_user_id]
|
if user.present? && session[:attempt_user_id].present? && session[:attempt_user_updated_at] != user.updated_at.to_s
|
||||||
|
restart_session
|
||||||
|
elsif user.webauthn_enabled? && user_params.key?(:credential) && session[:attempt_user_id]
|
||||||
authenticate_with_two_factor_via_webauthn(user)
|
authenticate_with_two_factor_via_webauthn(user)
|
||||||
elsif user_params[:otp_attempt].present? && session[:attempt_user_id]
|
elsif user_params.key?(:otp_attempt) && session[:attempt_user_id]
|
||||||
authenticate_with_two_factor_via_otp(user)
|
authenticate_with_two_factor_via_otp(user)
|
||||||
elsif user.present? && user.external_or_valid_password?(user_params[:password])
|
elsif user.present? && user.external_or_valid_password?(user_params[:password])
|
||||||
prompt_for_two_factor(user)
|
prompt_for_two_factor(user)
|
||||||
|
@ -50,7 +52,7 @@ module TwoFactorAuthenticationConcern
|
||||||
webauthn_credential = WebAuthn::Credential.from_get(user_params[:credential])
|
webauthn_credential = WebAuthn::Credential.from_get(user_params[:credential])
|
||||||
|
|
||||||
if valid_webauthn_credential?(user, webauthn_credential)
|
if valid_webauthn_credential?(user, webauthn_credential)
|
||||||
session.delete(:attempt_user_id)
|
clear_attempt_from_session
|
||||||
remember_me(user)
|
remember_me(user)
|
||||||
sign_in(user)
|
sign_in(user)
|
||||||
render json: { redirect_path: root_path }, status: :ok
|
render json: { redirect_path: root_path }, status: :ok
|
||||||
|
@ -61,7 +63,7 @@ module TwoFactorAuthenticationConcern
|
||||||
|
|
||||||
def authenticate_with_two_factor_via_otp(user)
|
def authenticate_with_two_factor_via_otp(user)
|
||||||
if valid_otp_attempt?(user)
|
if valid_otp_attempt?(user)
|
||||||
session.delete(:attempt_user_id)
|
clear_attempt_from_session
|
||||||
remember_me(user)
|
remember_me(user)
|
||||||
sign_in(user)
|
sign_in(user)
|
||||||
else
|
else
|
||||||
|
@ -71,17 +73,20 @@ module TwoFactorAuthenticationConcern
|
||||||
end
|
end
|
||||||
|
|
||||||
def prompt_for_two_factor(user)
|
def prompt_for_two_factor(user)
|
||||||
set_locale do
|
set_attempt_session(user)
|
||||||
session[:attempt_user_id] = user.id
|
|
||||||
use_pack 'auth'
|
use_pack 'auth'
|
||||||
|
|
||||||
@body_classes = 'lighter'
|
@body_classes = 'lighter'
|
||||||
@webauthn_enabled = user.webauthn_enabled?
|
@webauthn_enabled = user.webauthn_enabled?
|
||||||
@scheme_type = if user.webauthn_enabled? && user_params[:otp_attempt].blank?
|
@scheme_type = begin
|
||||||
|
if user.webauthn_enabled? && user_params[:otp_attempt].blank?
|
||||||
'webauthn'
|
'webauthn'
|
||||||
else
|
else
|
||||||
'totp'
|
'totp'
|
||||||
end
|
end
|
||||||
render :two_factor
|
end
|
||||||
end
|
|
||||||
|
set_locale { render :two_factor }
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
|
@ -6,14 +6,13 @@ module UserTrackingConcern
|
||||||
UPDATE_SIGN_IN_HOURS = 24
|
UPDATE_SIGN_IN_HOURS = 24
|
||||||
|
|
||||||
included do
|
included do
|
||||||
before_action :set_user_activity
|
before_action :update_user_sign_in
|
||||||
end
|
end
|
||||||
|
|
||||||
private
|
private
|
||||||
|
|
||||||
def set_user_activity
|
def update_user_sign_in
|
||||||
return unless user_needs_sign_in_update?
|
current_user.update_sign_in!(request) if user_needs_sign_in_update?
|
||||||
current_user.update_tracked_fields!(request)
|
|
||||||
end
|
end
|
||||||
|
|
||||||
def user_needs_sign_in_update?
|
def user_needs_sign_in_update?
|
||||||
|
|
|
@ -0,0 +1,17 @@
|
||||||
|
# frozen_string_literal: true
|
||||||
|
|
||||||
|
module AccessTokenExtension
|
||||||
|
extend ActiveSupport::Concern
|
||||||
|
|
||||||
|
included do
|
||||||
|
after_commit :push_to_streaming_api
|
||||||
|
end
|
||||||
|
|
||||||
|
def revoke(clock = Time)
|
||||||
|
update(revoked_at: clock.now.utc)
|
||||||
|
end
|
||||||
|
|
||||||
|
def push_to_streaming_api
|
||||||
|
Redis.current.publish("timeline:access_token:#{id}", Oj.dump(event: :kill)) if revoked? || destroyed?
|
||||||
|
end
|
||||||
|
end
|
|
@ -70,12 +70,16 @@ class SessionActivation < ApplicationRecord
|
||||||
end
|
end
|
||||||
|
|
||||||
def assign_access_token
|
def assign_access_token
|
||||||
superapp = Doorkeeper::Application.find_by(superapp: true)
|
self.access_token = Doorkeeper::AccessToken.create!(access_token_attributes)
|
||||||
|
end
|
||||||
|
|
||||||
self.access_token = Doorkeeper::AccessToken.create!(application_id: superapp&.id,
|
def access_token_attributes
|
||||||
|
{
|
||||||
|
application_id: Doorkeeper::Application.find_by(superapp: true)&.id,
|
||||||
resource_owner_id: user_id,
|
resource_owner_id: user_id,
|
||||||
scopes: 'read write follow',
|
scopes: 'read write follow',
|
||||||
expires_in: Doorkeeper.configuration.access_token_expires_in,
|
expires_in: Doorkeeper.configuration.access_token_expires_in,
|
||||||
use_refresh_token: Doorkeeper.configuration.refresh_token_enabled?)
|
use_refresh_token: Doorkeeper.configuration.refresh_token_enabled?,
|
||||||
|
}
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
|
@ -63,7 +63,7 @@ class User < ApplicationRecord
|
||||||
devise :two_factor_backupable,
|
devise :two_factor_backupable,
|
||||||
otp_number_of_backup_codes: 10
|
otp_number_of_backup_codes: 10
|
||||||
|
|
||||||
devise :registerable, :recoverable, :rememberable, :trackable, :validatable,
|
devise :registerable, :recoverable, :rememberable, :validatable,
|
||||||
:confirmable
|
:confirmable
|
||||||
|
|
||||||
include Omniauthable
|
include Omniauthable
|
||||||
|
@ -165,6 +165,24 @@ class User < ApplicationRecord
|
||||||
prepare_new_user! if new_user && approved?
|
prepare_new_user! if new_user && approved?
|
||||||
end
|
end
|
||||||
|
|
||||||
|
def update_sign_in!(request, new_sign_in: false)
|
||||||
|
old_current, new_current = current_sign_in_at, Time.now.utc
|
||||||
|
self.last_sign_in_at = old_current || new_current
|
||||||
|
self.current_sign_in_at = new_current
|
||||||
|
|
||||||
|
old_current, new_current = current_sign_in_ip, request.remote_ip
|
||||||
|
self.last_sign_in_ip = old_current || new_current
|
||||||
|
self.current_sign_in_ip = new_current
|
||||||
|
|
||||||
|
if new_sign_in
|
||||||
|
self.sign_in_count ||= 0
|
||||||
|
self.sign_in_count += 1
|
||||||
|
end
|
||||||
|
|
||||||
|
save(validate: false) unless new_record?
|
||||||
|
prepare_returning_user!
|
||||||
|
end
|
||||||
|
|
||||||
def pending?
|
def pending?
|
||||||
!approved?
|
!approved?
|
||||||
end
|
end
|
||||||
|
@ -196,11 +214,6 @@ class User < ApplicationRecord
|
||||||
prepare_new_user!
|
prepare_new_user!
|
||||||
end
|
end
|
||||||
|
|
||||||
def update_tracked_fields!(request)
|
|
||||||
super
|
|
||||||
prepare_returning_user!
|
|
||||||
end
|
|
||||||
|
|
||||||
def otp_enabled?
|
def otp_enabled?
|
||||||
otp_required_for_login
|
otp_required_for_login
|
||||||
end
|
end
|
||||||
|
|
|
@ -5,7 +5,8 @@ class AccountDeletionWorker
|
||||||
|
|
||||||
sidekiq_options queue: 'pull'
|
sidekiq_options queue: 'pull'
|
||||||
|
|
||||||
def perform(account_id, reserve_username: true)
|
def perform(account_id, options = {})
|
||||||
|
reserve_username = options.with_indifferent_access.fetch(:reserve_username, true)
|
||||||
DeleteAccountService.new.call(Account.find(account_id), reserve_username: reserve_username, reserve_email: false)
|
DeleteAccountService.new.call(Account.find(account_id), reserve_username: reserve_username, reserve_email: false)
|
||||||
rescue ActiveRecord::RecordNotFound
|
rescue ActiveRecord::RecordNotFound
|
||||||
true
|
true
|
||||||
|
|
|
@ -140,6 +140,7 @@ module Mastodon
|
||||||
Doorkeeper::AuthorizationsController.layout 'modal'
|
Doorkeeper::AuthorizationsController.layout 'modal'
|
||||||
Doorkeeper::AuthorizedApplicationsController.layout 'admin'
|
Doorkeeper::AuthorizedApplicationsController.layout 'admin'
|
||||||
Doorkeeper::Application.send :include, ApplicationExtension
|
Doorkeeper::Application.send :include, ApplicationExtension
|
||||||
|
Doorkeeper::AccessToken.send :include, AccessTokenExtension
|
||||||
Devise::FailureApp.send :include, AbstractController::Callbacks
|
Devise::FailureApp.send :include, AbstractController::Callbacks
|
||||||
Devise::FailureApp.send :include, HttpAcceptLanguage::EasyAccess
|
Devise::FailureApp.send :include, HttpAcceptLanguage::EasyAccess
|
||||||
Devise::FailureApp.send :include, Localized
|
Devise::FailureApp.send :include, Localized
|
||||||
|
|
|
@ -219,7 +219,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
|
||||||
|
|
||||||
context 'using a valid OTP' do
|
context 'using a valid OTP' do
|
||||||
before do
|
before do
|
||||||
post :create, params: { user: { otp_attempt: user.current_otp } }, session: { attempt_user_id: user.id }
|
post :create, params: { user: { otp_attempt: user.current_otp } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'redirects to home' do
|
it 'redirects to home' do
|
||||||
|
@ -234,7 +234,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
|
||||||
context 'when the server has an decryption error' do
|
context 'when the server has an decryption error' do
|
||||||
before do
|
before do
|
||||||
allow_any_instance_of(User).to receive(:validate_and_consume_otp!).and_raise(OpenSSL::Cipher::CipherError)
|
allow_any_instance_of(User).to receive(:validate_and_consume_otp!).and_raise(OpenSSL::Cipher::CipherError)
|
||||||
post :create, params: { user: { otp_attempt: user.current_otp } }, session: { attempt_user_id: user.id }
|
post :create, params: { user: { otp_attempt: user.current_otp } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'shows a login error' do
|
it 'shows a login error' do
|
||||||
|
@ -248,7 +248,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
|
||||||
|
|
||||||
context 'using a valid recovery code' do
|
context 'using a valid recovery code' do
|
||||||
before do
|
before do
|
||||||
post :create, params: { user: { otp_attempt: recovery_codes.first } }, session: { attempt_user_id: user.id }
|
post :create, params: { user: { otp_attempt: recovery_codes.first } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'redirects to home' do
|
it 'redirects to home' do
|
||||||
|
@ -262,7 +262,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
|
||||||
|
|
||||||
context 'using an invalid OTP' do
|
context 'using an invalid OTP' do
|
||||||
before do
|
before do
|
||||||
post :create, params: { user: { otp_attempt: 'wrongotp' } }, session: { attempt_user_id: user.id }
|
post :create, params: { user: { otp_attempt: 'wrongotp' } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'shows a login error' do
|
it 'shows a login error' do
|
||||||
|
@ -334,7 +334,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
|
||||||
before do
|
before do
|
||||||
@controller.session[:webauthn_challenge] = challenge
|
@controller.session[:webauthn_challenge] = challenge
|
||||||
|
|
||||||
post :create, params: { user: { credential: fake_credential } }, session: { attempt_user_id: user.id }
|
post :create, params: { user: { credential: fake_credential } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'instructs the browser to redirect to home' do
|
it 'instructs the browser to redirect to home' do
|
||||||
|
@ -383,7 +383,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
|
||||||
context 'using a valid sign in token' do
|
context 'using a valid sign in token' do
|
||||||
before do
|
before do
|
||||||
user.generate_sign_in_token && user.save
|
user.generate_sign_in_token && user.save
|
||||||
post :create, params: { user: { sign_in_token_attempt: user.sign_in_token } }, session: { attempt_user_id: user.id }
|
post :create, params: { user: { sign_in_token_attempt: user.sign_in_token } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'redirects to home' do
|
it 'redirects to home' do
|
||||||
|
@ -397,7 +397,7 @@ RSpec.describe Auth::SessionsController, type: :controller do
|
||||||
|
|
||||||
context 'using an invalid sign in token' do
|
context 'using an invalid sign in token' do
|
||||||
before do
|
before do
|
||||||
post :create, params: { user: { sign_in_token_attempt: 'wrongotp' } }, session: { attempt_user_id: user.id }
|
post :create, params: { user: { sign_in_token_attempt: 'wrongotp' } }, session: { attempt_user_id: user.id, attempt_user_updated_at: user.updated_at.to_s }
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'shows a login error' do
|
it 'shows a login error' do
|
||||||
|
|
|
@ -294,7 +294,7 @@ const startWorker = (workerId) => {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
client.query('SELECT oauth_access_tokens.resource_owner_id, users.account_id, users.chosen_languages, oauth_access_tokens.scopes, devices.device_id FROM oauth_access_tokens INNER JOIN users ON oauth_access_tokens.resource_owner_id = users.id LEFT OUTER JOIN devices ON oauth_access_tokens.id = devices.access_token_id WHERE oauth_access_tokens.token = $1 AND oauth_access_tokens.revoked_at IS NULL LIMIT 1', [token], (err, result) => {
|
client.query('SELECT oauth_access_tokens.id, oauth_access_tokens.resource_owner_id, users.account_id, users.chosen_languages, oauth_access_tokens.scopes, devices.device_id FROM oauth_access_tokens INNER JOIN users ON oauth_access_tokens.resource_owner_id = users.id LEFT OUTER JOIN devices ON oauth_access_tokens.id = devices.access_token_id WHERE oauth_access_tokens.token = $1 AND oauth_access_tokens.revoked_at IS NULL LIMIT 1', [token], (err, result) => {
|
||||||
done();
|
done();
|
||||||
|
|
||||||
if (err) {
|
if (err) {
|
||||||
|
@ -310,6 +310,7 @@ const startWorker = (workerId) => {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
req.accessTokenId = result.rows[0].id;
|
||||||
req.scopes = result.rows[0].scopes.split(' ');
|
req.scopes = result.rows[0].scopes.split(' ');
|
||||||
req.accountId = result.rows[0].account_id;
|
req.accountId = result.rows[0].account_id;
|
||||||
req.chosenLanguages = result.rows[0].chosen_languages;
|
req.chosenLanguages = result.rows[0].chosen_languages;
|
||||||
|
@ -451,6 +452,55 @@ const startWorker = (workerId) => {
|
||||||
});
|
});
|
||||||
};
|
};
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @typedef SystemMessageHandlers
|
||||||
|
* @property {function(): void} onKill
|
||||||
|
*/
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param {any} req
|
||||||
|
* @param {SystemMessageHandlers} eventHandlers
|
||||||
|
* @return {function(string): void}
|
||||||
|
*/
|
||||||
|
const createSystemMessageListener = (req, eventHandlers) => {
|
||||||
|
return message => {
|
||||||
|
const json = parseJSON(message);
|
||||||
|
|
||||||
|
if (!json) return;
|
||||||
|
|
||||||
|
const { event } = json;
|
||||||
|
|
||||||
|
log.silly(req.requestId, `System message for ${req.accountId}: ${event}`);
|
||||||
|
|
||||||
|
if (event === 'kill') {
|
||||||
|
log.verbose(req.requestId, `Closing connection for ${req.accountId} due to expired access token`);
|
||||||
|
eventHandlers.onKill();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param {any} req
|
||||||
|
* @param {any} res
|
||||||
|
*/
|
||||||
|
const subscribeHttpToSystemChannel = (req, res) => {
|
||||||
|
const systemChannelId = `timeline:access_token:${req.accessTokenId}`;
|
||||||
|
|
||||||
|
const listener = createSystemMessageListener(req, {
|
||||||
|
|
||||||
|
onKill () {
|
||||||
|
res.end();
|
||||||
|
},
|
||||||
|
|
||||||
|
});
|
||||||
|
|
||||||
|
res.on('close', () => {
|
||||||
|
unsubscribe(`${redisPrefix}${systemChannelId}`, listener);
|
||||||
|
});
|
||||||
|
|
||||||
|
subscribe(`${redisPrefix}${systemChannelId}`, listener);
|
||||||
|
};
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param {any} req
|
* @param {any} req
|
||||||
* @param {any} res
|
* @param {any} res
|
||||||
|
@ -463,6 +513,8 @@ const startWorker = (workerId) => {
|
||||||
}
|
}
|
||||||
|
|
||||||
accountFromRequest(req, alwaysRequireAuth).then(() => checkScopes(req, channelNameFromPath(req))).then(() => {
|
accountFromRequest(req, alwaysRequireAuth).then(() => checkScopes(req, channelNameFromPath(req))).then(() => {
|
||||||
|
subscribeHttpToSystemChannel(req, res);
|
||||||
|
}).then(() => {
|
||||||
next();
|
next();
|
||||||
}).catch(err => {
|
}).catch(err => {
|
||||||
next(err);
|
next(err);
|
||||||
|
@ -538,7 +590,9 @@ const startWorker = (workerId) => {
|
||||||
|
|
||||||
const listener = message => {
|
const listener = message => {
|
||||||
const json = parseJSON(message);
|
const json = parseJSON(message);
|
||||||
|
|
||||||
if (!json) return;
|
if (!json) return;
|
||||||
|
|
||||||
const { event, payload, queued_at } = json;
|
const { event, payload, queued_at } = json;
|
||||||
|
|
||||||
const transmit = () => {
|
const transmit = () => {
|
||||||
|
@ -924,6 +978,28 @@ const startWorker = (workerId) => {
|
||||||
socket.send(JSON.stringify({ error: err.toString() }));
|
socket.send(JSON.stringify({ error: err.toString() }));
|
||||||
});
|
});
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param {WebSocketSession} session
|
||||||
|
*/
|
||||||
|
const subscribeWebsocketToSystemChannel = ({ socket, request, subscriptions }) => {
|
||||||
|
const systemChannelId = `timeline:access_token:${request.accessTokenId}`;
|
||||||
|
|
||||||
|
const listener = createSystemMessageListener(request, {
|
||||||
|
|
||||||
|
onKill () {
|
||||||
|
socket.close();
|
||||||
|
},
|
||||||
|
|
||||||
|
});
|
||||||
|
|
||||||
|
subscribe(`${redisPrefix}${systemChannelId}`, listener);
|
||||||
|
|
||||||
|
subscriptions[systemChannelId] = {
|
||||||
|
listener,
|
||||||
|
stopHeartbeat: () => {},
|
||||||
|
};
|
||||||
|
};
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param {string|string[]} arrayOrString
|
* @param {string|string[]} arrayOrString
|
||||||
* @return {string}
|
* @return {string}
|
||||||
|
@ -970,7 +1046,9 @@ const startWorker = (workerId) => {
|
||||||
|
|
||||||
ws.on('message', data => {
|
ws.on('message', data => {
|
||||||
const json = parseJSON(data);
|
const json = parseJSON(data);
|
||||||
|
|
||||||
if (!json) return;
|
if (!json) return;
|
||||||
|
|
||||||
const { type, stream, ...params } = json;
|
const { type, stream, ...params } = json;
|
||||||
|
|
||||||
if (type === 'subscribe') {
|
if (type === 'subscribe') {
|
||||||
|
@ -982,6 +1060,8 @@ const startWorker = (workerId) => {
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
subscribeWebsocketToSystemChannel(session);
|
||||||
|
|
||||||
if (location.query.stream) {
|
if (location.query.stream) {
|
||||||
subscribeWebsocketToChannel(session, firstParam(location.query.stream), location.query);
|
subscribeWebsocketToChannel(session, firstParam(location.query.stream), location.query);
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue