gitea/nas
1
0
Fork 2
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Add support for editing for published statuses (#16697)

Browse source 
* Add support for editing for published statuses

* Fix references to stripped-out code

* Various fixes and improvements

* Further fixes and improvements

* Fix updates being potentially sent to unauthorized recipients

* Various fixes and improvements

* Fix wrong words in test

* Fix notifying accounts that were tagged but were not in the audience

* Fix mistake
This commit is contained in:
Eugen Rochko 2022-01-19 22:37:27 +01:00 • committed by GitHub
parent 2d1f082bb6
commit 1060666c58
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
56 changed files with 1415 additions and 574 deletions
Download patch file Download diff file Expand all files Collapse all files

2
app/services/activitypub/fetch_remote_poll_service.rb
View file

@ -8,6 +8,6 @@ class ActivityPub::FetchRemotePollService < BaseService
return unless supported_context?(json)
ActivityPub::ProcessPollService.new.call(poll, json)
ActivityPub::ProcessStatusUpdateService.new.call(poll.status, json)
end
end

64
app/services/activitypub/process_poll_service.rb
View file

@ -1,64 +0,0 @@
# frozen_string_literal: true
class ActivityPub::ProcessPollService < BaseService
include JsonLdHelper
def call(poll, json)
@json = json
return unless expected_type?
previous_expires_at = poll.expires_at
expires_at = begin
if @json['closed'].is_a?(String)
@json['closed']
elsif !@json['closed'].nil? && !@json['closed'].is_a?(FalseClass)
Time.now.utc
else
@json['endTime']
end
end
items = begin
if @json['anyOf'].is_a?(Array)
@json['anyOf']
else
@json['oneOf']
end
end
voters_count = @json['votersCount']
latest_options = items.filter_map { |item| item['name'].presence || item['content'] }
# If for some reasons the options were changed, it invalidates all previous
# votes, so we need to remove them
poll.votes.delete_all if latest_options != poll.options
begin
poll.update!(
last_fetched_at: Time.now.utc,
expires_at: expires_at,
options: latest_options,
cached_tallies: items.map { |item| item.dig('replies', 'totalItems') || 0 },
voters_count: voters_count
)
rescue ActiveRecord::StaleObjectError
poll.reload
retry
end
# If the poll had no expiration date set but now has, and people have voted,
# schedule a notification.
if previous_expires_at.nil? && poll.expires_at.present? && poll.votes.exists?
PollExpirationNotifyWorker.perform_at(poll.expires_at + 5.minutes, poll.id)
end
end
private
def expected_type?
equals_or_includes_any?(@json['type'], %w(Question))
end
end

275
app/services/activitypub/process_status_update_service.rb Normal file
View file

@ -0,0 +1,275 @@
# frozen_string_literal: true
class ActivityPub::ProcessStatusUpdateService < BaseService
include JsonLdHelper
def call(status, json)
@json = json
@status_parser = ActivityPub::Parser::StatusParser.new(@json)
@uri = @status_parser.uri
@status = status
@account = status.account
@media_attachments_changed = false
# Only native types can be updated at the moment
return if !expected_type? || already_updated_more_recently?
# Only allow processing one create/update per status at a time
RedisLock.acquire(lock_options) do |lock|
if lock.acquired?
Status.transaction do
create_previous_edit!
update_media_attachments!
update_poll!
update_immediate_attributes!
update_metadata!
create_edit!
end
queue_poll_notifications!
reset_preview_card!
broadcast_updates!
else
raise Mastodon::RaceConditionError
end
end
end
private
def update_media_attachments!
previous_media_attachments = @status.media_attachments.to_a
next_media_attachments = []
as_array(@json['attachment']).each do |attachment|
media_attachment_parser = ActivityPub::Parser::MediaAttachmentParser.new(attachment)
next if media_attachment_parser.remote_url.blank? || next_media_attachments.size > 4
begin
media_attachment = previous_media_attachments.find { |previous_media_attachment| previous_media_attachment.remote_url == media_attachment_parser.remote_url }
media_attachment ||= MediaAttachment.new(account: @account, remote_url: media_attachment_parser.remote_url)
# If a previously existing media attachment was significantly updated, mark
# media attachments as changed even if none were added or removed
if media_attachment_parser.significantly_changes?(media_attachment)
@media_attachments_changed = true
end
media_attachment.description = media_attachment_parser.description
media_attachment.focus = media_attachment_parser.focus
media_attachment.thumbnail_remote_url = media_attachment_parser.thumbnail_remote_url
media_attachment.blurhash = media_attachment_parser.blurhash
media_attachment.save!
next_media_attachments << media_attachment
next if unsupported_media_type?(media_attachment_parser.file_content_type) || skip_download?
RedownloadMediaWorker.perform_async(media_attachment.id) if media_attachment.remote_url_previously_changed? || media_attachment.thumbnail_remote_url_previously_changed?
rescue Addressable::URI::InvalidURIError => e
Rails.logger.debug "Invalid URL in attachment: #{e}"
end
end
removed_media_attachments = previous_media_attachments - next_media_attachments
added_media_attachments = next_media_attachments - previous_media_attachments
MediaAttachment.where(id: removed_media_attachments.map(&:id)).update_all(status_id: nil)
MediaAttachment.where(id: added_media_attachments.map(&:id)).update_all(status_id: @status.id)
@media_attachments_changed = true if removed_media_attachments.positive? || added_media_attachments.positive?
end
def update_poll!
previous_poll = @status.preloadable_poll
@previous_expires_at = previous_poll&.expires_at
poll_parser = ActivityPub::Parser::PollParser.new(@json)
if poll_parser.valid?
poll = previous_poll || @account.polls.new(status: @status)
# If for some reasons the options were changed, it invalidates all previous
# votes, so we need to remove them
if poll_parser.significantly_changes?(poll)
@media_attachments_changed = true
poll.votes.delete_all unless poll.new_record?
end
poll.last_fetched_at = Time.now.utc
poll.options = poll_parser.options
poll.multiple = poll_parser.multiple
poll.expires_at = poll_parser.expires_at
poll.voters_count = poll_parser.voters_count
poll.cached_tallies = poll_parser.cached_tallies
poll.save!
@status.poll_id = poll.id
elsif previous_poll.present?
previous_poll.destroy!
@media_attachments_changed = true
@status.poll_id = nil
end
end
def update_immediate_attributes!
@status.text = @status_parser.text || ''
@status.spoiler_text = @status_parser.spoiler_text || ''
@status.sensitive = @account.sensitized? || @status_parser.sensitive || false
@status.language = @status_parser.language || detected_language
@status.edited_at = @status_parser.edited_at || Time.now.utc
@status.save!
end
def update_metadata!
@raw_tags = []
@raw_mentions = []
@raw_emojis = []
as_array(@json['tag']).each do |tag|
if equals_or_includes?(tag['type'], 'Hashtag')
@raw_tags << tag['name']
elsif equals_or_includes?(tag['type'], 'Mention')
@raw_mentions << tag['href']
elsif equals_or_includes?(tag['type'], 'Emoji')
@raw_emojis << tag
end
end
update_tags!
update_mentions!
update_emojis!
end
def update_tags!
@status.tags = Tag.find_or_create_by_names(@raw_tags)
end
def update_mentions!
previous_mentions = @status.active_mentions.includes(:account).to_a
current_mentions = []
@raw_mentions.each do |href|
next if href.blank?
account = ActivityPub::TagManager.instance.uri_to_resource(href, Account)
account ||= ActivityPub::FetchRemoteAccountService.new.call(href)
next if account.nil?
mention = previous_mentions.find { |x| x.account_id == account.id }
mention ||= account.mentions.new(status: @status)
current_mentions << mention
end
current_mentions.each do |mention|
mention.save if mention.new_record?
end
# If previous mentions are no longer contained in the text, convert them
# to silent mentions, since withdrawing access from someone who already
# received a notification might be more confusing
removed_mentions = previous_mentions - current_mentions
Mention.where(id: removed_mentions.map(&:id)).update_all(silent: true) unless removed_mentions.empty?
end
def update_emojis!
return if skip_download?
@raw_emojis.each do |raw_emoji|
custom_emoji_parser = ActivityPub::Parser::CustomEmojiParser.new(raw_emoji)
next if custom_emoji_parser.shortcode.blank? || custom_emoji_parser.image_remote_url.blank?
emoji = CustomEmoji.find_by(shortcode: custom_emoji_parser.shortcode, domain: @account.domain)
next unless emoji.nil? || custom_emoji_parser.image_remote_url != emoji.image_remote_url || (custom_emoji_parser.updated_at && custom_emoji_parser.updated_at >= emoji.updated_at)
begin
emoji ||= CustomEmoji.new(domain: @account.domain, shortcode: custom_emoji_parser.shortcode, uri: custom_emoji_parser.uri)
emoji.image_remote_url = custom_emoji_parser.image_remote_url
emoji.save
rescue Seahorse::Client::NetworkingError => e
Rails.logger.warn "Error storing emoji: #{e}"
end
end
end
def expected_type?
equals_or_includes_any?(@json['type'], %w(Note Question))
end
def lock_options
{ redis: Redis.current, key: "create:#{@uri}", autorelease: 15.minutes.seconds }
end
def detected_language
LanguageDetector.instance.detect(@status_parser.text, @account)
end
def create_previous_edit!
# We only need to create a previous edit when no previous edits exist, e.g.
# when the status has never been edited. For other cases, we always create
# an edit, so the step can be skipped
return if @status.edits.any?
@status.edits.create(
text: @status.text,
spoiler_text: @status.spoiler_text,
media_attachments_changed: false,
account_id: @account.id,
created_at: @status.created_at
)
end
def create_edit!
return unless @status.text_previously_changed? || @status.spoiler_text_previously_changed? || @media_attachments_changed
@status_edit = @status.edits.create(
text: @status.text,
spoiler_text: @status.spoiler_text,
media_attachments_changed: @media_attachments_changed,
account_id: @account.id,
created_at: @status.edited_at
)
end
def skip_download?
return @skip_download if defined?(@skip_download)
@skip_download ||= DomainBlock.reject_media?(@account.domain)
end
def unsupported_media_type?(mime_type)
mime_type.present? && !MediaAttachment.supported_mime_types.include?(mime_type)
end
def already_updated_more_recently?
@status.edited_at.present? && @status_parser.edited_at.present? && @status.edited_at > @status_parser.edited_at
end
def reset_preview_card!
@status.preview_cards.clear if @status.text_previously_changed? || @status.spoiler_text.present?
LinkCrawlWorker.perform_in(rand(1..59).seconds, @status.id) if @status.spoiler_text.blank?
end
def broadcast_updates!
::DistributionWorker.perform_async(@status.id, update: true)
end
def queue_poll_notifications!
poll = @status.preloadable_poll
# If the poll had no expiration date set but now has, or now has a sooner
# expiration date, and people have voted, schedule a notification
return unless poll.present? && poll.expires_at.present? && poll.votes.exists?
PollExpirationNotifyWorker.remove_from_scheduled(poll.id) if @previous_expires_at.present? && @previous_expires_at > poll.expires_at
PollExpirationNotifyWorker.perform_at(poll.expires_at + 5.minutes, poll.id)
end
end

157
app/services/fan_out_on_write_service.rb
View file

@ -3,107 +3,126 @@
class FanOutOnWriteService < BaseService
# Push a status into home and mentions feeds
# @param [Status] status
def call(status)
raise Mastodon::RaceConditionError if status.visibility.nil?
# @param [Hash] options
# @option options [Boolean] update
# @option options [Array<Integer>] silenced_account_ids
def call(status, options = {})
@status = status
@account = status.account
@options = options
deliver_to_self(status) if status.account.local?
check_race_condition!
if status.direct_visibility?
deliver_to_mentioned_followers(status)
deliver_to_own_conversation(status)
elsif status.limited_visibility?
deliver_to_mentioned_followers(status)
else
deliver_to_followers(status)
deliver_to_lists(status)
end
return if status.account.silenced? || !status.public_visibility? || status.reblog?
render_anonymous_payload(status)
deliver_to_hashtags(status)
return if status.reply? && status.in_reply_to_account_id != status.account_id
deliver_to_public(status)
deliver_to_media(status) if status.media_attachments.any?
fan_out_to_local_recipients!
fan_out_to_public_streams! if broadcastable?
end
private
def deliver_to_self(status)
Rails.logger.debug "Delivering status #{status.id} to author"
FeedManager.instance.push_to_home(status.account, status)
def check_race_condition!
# I don't know why but at some point we had an issue where
# this service was being executed with status objects
# that had a null visibility - which should not be possible
# since the column in the database is not nullable.
#
# This check re-queues the service to be run at a later time
# with the full object, if something like it occurs
raise Mastodon::RaceConditionError if @status.visibility.nil?
end
def deliver_to_followers(status)
Rails.logger.debug "Delivering status #{status.id} to followers"
def fan_out_to_local_recipients!
deliver_to_self!
notify_mentioned_accounts!
status.account.followers_for_local_distribution.select(:id).reorder(nil).find_in_batches do |followers|
case @status.visibility.to_sym
when :public, :unlisted, :private
deliver_to_all_followers!
deliver_to_lists!
when :limited
deliver_to_mentioned_followers!
else
deliver_to_mentioned_followers!
deliver_to_conversation!
end
end
def fan_out_to_public_streams!
broadcast_to_hashtag_streams!
broadcast_to_public_streams!
end
def deliver_to_self!
FeedManager.instance.push_to_home(@account, @status, update: update?) if @account.local?
end
def notify_mentioned_accounts!
@status.active_mentions.where.not(id: @options[:silenced_account_ids] || []).joins(:account).merge(Account.local).select(:id, :account_id).reorder(nil).find_in_batches do |mentions|
LocalNotificationWorker.push_bulk(mentions) do |mention|
[mention.account_id, mention.id, 'Mention', :mention]
end
end
end
def deliver_to_all_followers!
@account.followers_for_local_distribution.select(:id).reorder(nil).find_in_batches do |followers|
FeedInsertWorker.push_bulk(followers) do |follower|
[status.id, follower.id, :home]
[@status.id, follower.id, :home, update: update?]
end
end
end
def deliver_to_lists(status)
Rails.logger.debug "Delivering status #{status.id} to lists"
status.account.lists_for_local_distribution.select(:id).reorder(nil).find_in_batches do |lists|
def deliver_to_lists!
@account.lists_for_local_distribution.select(:id).reorder(nil).find_in_batches do |lists|
FeedInsertWorker.push_bulk(lists) do |list|
[status.id, list.id, :list]
[@status.id, list.id, :list, update: update?]
end
end
end
def deliver_to_mentioned_followers(status)
Rails.logger.debug "Delivering status #{status.id} to limited followers"
status.mentions.joins(:account).merge(status.account.followers_for_local_distribution).select(:id, :account_id).reorder(nil).find_in_batches do |mentions|
def deliver_to_mentioned_followers!
@status.mentions.joins(:account).merge(@account.followers_for_local_distribution).select(:id, :account_id).reorder(nil).find_in_batches do |mentions|
FeedInsertWorker.push_bulk(mentions) do |mention|
[status.id, mention.account_id, :home]
[@status.id, mention.account_id, :home, update: update?]
end
end
end
def render_anonymous_payload(status)
@payload = InlineRenderer.render(status, nil, :status)
@payload = Oj.dump(event: :update, payload: @payload)
end
def deliver_to_hashtags(status)
Rails.logger.debug "Delivering status #{status.id} to hashtags"
status.tags.pluck(:name).each do |hashtag|
Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", @payload)
Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", @payload) if status.local?
def broadcast_to_hashtag_streams!
@status.tags.pluck(:name).each do |hashtag|
Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", anonymous_payload)
Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", anonymous_payload) if @status.local?
end
end
def deliver_to_public(status)
Rails.logger.debug "Delivering status #{status.id} to public timeline"
def broadcast_to_public_streams!
return if @status.reply? && @status.in_reply_to_account_id != @account.id
Redis.current.publish('timeline:public', @payload)
if status.local?
Redis.current.publish('timeline:public:local', @payload)
else
Redis.current.publish('timeline:public:remote', @payload)
Redis.current.publish('timeline:public', anonymous_payload)
Redis.current.publish(@status.local? ? 'timeline:public:local' : 'timeline:public:remote', anonymous_payload)
if @status.media_attachments.any?
Redis.current.publish('timeline:public:media', anonymous_payload)
Redis.current.publish(@status.local? ? 'timeline:public:local:media' : 'timeline:public:remote:media', anonymous_payload)
end
end
def deliver_to_media(status)
Rails.logger.debug "Delivering status #{status.id} to media timeline"
Redis.current.publish('timeline:public:media', @payload)
if status.local?
Redis.current.publish('timeline:public:local:media', @payload)
else
Redis.current.publish('timeline:public:remote:media', @payload)
end
def deliver_to_conversation!
AccountConversation.add_status(@account, @status) unless update?
end
def deliver_to_own_conversation(status)
AccountConversation.add_status(status.account, status)
def anonymous_payload
@anonymous_payload ||= Oj.dump(
event: update? ? :'status.update' : :update,
payload: InlineRenderer.render(@status, nil, :status)
)
end
def update?
@is_update
end
def broadcastable?
@status.public_visibility? && !@status.reblog? && !@account.silenced?
end
end

69
app/services/process_mentions_service.rb
View file

@ -8,12 +8,23 @@ class ProcessMentionsService < BaseService
# remote users
# @param [Status] status
def call(status)
return unless status.local?
@status = status
@status = status
mentions = []
return unless @status.local?
status.text = status.text.gsub(Account::MENTION_RE) do |match|
@previous_mentions = @status.active_mentions.includes(:account).to_a
@current_mentions = []
Status.transaction do
scan_text!
assign_mentions!
end
end
private
def scan_text!
@status.text = @status.text.gsub(Account::MENTION_RE) do |match|
username, domain = Regexp.last_match(1).split('@')
domain = begin
@ -26,49 +37,45 @@ class ProcessMentionsService < BaseService
mentioned_account = Account.find_remote(username, domain)
# If the account cannot be found or isn't the right protocol,
# first try to resolve it
if mention_undeliverable?(mentioned_account)
begin
mentioned_account = resolve_account_service.call(Regexp.last_match(1))
mentioned_account = ResolveAccountService.new.call(Regexp.last_match(1))
rescue Webfinger::Error, HTTP::Error, OpenSSL::SSL::SSLError, Mastodon::UnexpectedResponseError
mentioned_account = nil
end
end
# If after resolving it still isn't found or isn't the right
# protocol, then give up
next match if mention_undeliverable?(mentioned_account) || mentioned_account&.suspended?
mention = mentioned_account.mentions.new(status: status)
mentions << mention if mention.save
mention = @previous_mentions.find { |x| x.account_id == mentioned_account.id }
mention ||= mentioned_account.mentions.new(status: @status)
@current_mentions << mention
"@#{mentioned_account.acct}"
end
status.save!
mentions.each { |mention| create_notification(mention) }
@status.save!
end
private
def assign_mentions!
@current_mentions.each do |mention|
mention.save if mention.new_record?
end
# If previous mentions are no longer contained in the text, convert them
# to silent mentions, since withdrawing access from someone who already
# received a notification might be more confusing
removed_mentions = @previous_mentions - @current_mentions
Mention.where(id: removed_mentions.map(&:id)).update_all(silent: true) unless removed_mentions.empty?
end
def mention_undeliverable?(mentioned_account)
mentioned_account.nil? || (!mentioned_account.local? && mentioned_account.ostatus?)
end
def create_notification(mention)
mentioned_account = mention.account
if mentioned_account.local?
LocalNotificationWorker.perform_async(mentioned_account.id, mention.id, mention.class.name, :mention)
elsif mentioned_account.activitypub?
ActivityPub::DeliveryWorker.perform_async(activitypub_json, mention.status.account_id, mentioned_account.inbox_url, { synchronize_followers: !mention.status.distributable? })
end
end
def activitypub_json
return @activitypub_json if defined?(@activitypub_json)
@activitypub_json = Oj.dump(serialize_payload(ActivityPub::ActivityPresenter.from_status(@status), ActivityPub::ActivitySerializer, signer: @status.account))
end
def resolve_account_service
ResolveAccountService.new
mentioned_account.nil? || (!mentioned_account.local? && !mentioned_account.activitypub?)
end
end

2
app/services/remove_status_service.rb
View file

@ -87,7 +87,7 @@ class RemoveStatusService < BaseService
# the author and wouldn't normally receive the delete
# notification - so here, we explicitly send it to them
status_reach_finder = StatusReachFinder.new(@status)
status_reach_finder = StatusReachFinder.new(@status, unsafe: true)
ActivityPub::DeliveryWorker.push_bulk(status_reach_finder.inboxes) do |inbox_url|
[signed_activity_json, @account.id, inbox_url]