create.rb 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411
  1. # frozen_string_literal: true
  2. class ActivityPub::Activity::Create < ActivityPub::Activity
  3. def perform
  4. return reject_payload! if unsupported_object_type? || invalid_origin?(@object['id']) || Tombstone.exists?(uri: @object['id']) || !related_to_local_activity?
  5. RedisLock.acquire(lock_options) do |lock|
  6. if lock.acquired?
  7. return if delete_arrived_first?(object_uri) || poll_vote?
  8. @status = find_existing_status
  9. if @status.nil?
  10. process_status
  11. elsif @options[:delivered_to_account_id].present?
  12. postprocess_audience_and_deliver
  13. end
  14. else
  15. raise Mastodon::RaceConditionError
  16. end
  17. end
  18. @status
  19. end
  20. private
  21. def process_status
  22. @tags = []
  23. @mentions = []
  24. @params = {}
  25. process_status_params
  26. process_tags
  27. process_audience
  28. ApplicationRecord.transaction do
  29. @status = Status.create!(@params)
  30. attach_tags(@status)
  31. end
  32. resolve_thread(@status)
  33. fetch_replies(@status)
  34. distribute(@status)
  35. forward_for_reply if @status.public_visibility? || @status.unlisted_visibility?
  36. end
  37. def find_existing_status
  38. status = status_from_uri(object_uri)
  39. status ||= Status.find_by(uri: @object['atomUri']) if @object['atomUri'].present?
  40. status
  41. end
  42. def process_status_params
  43. @params = begin
  44. {
  45. uri: @object['id'],
  46. url: object_url || @object['id'],
  47. account: @account,
  48. text: text_from_content || '',
  49. language: detected_language,
  50. spoiler_text: converted_object_type? ? '' : (text_from_summary || ''),
  51. created_at: @object['published'],
  52. override_timestamps: @options[:override_timestamps],
  53. reply: @object['inReplyTo'].present?,
  54. sensitive: @object['sensitive'] || false,
  55. visibility: visibility_from_audience,
  56. thread: replied_to_status,
  57. conversation: conversation_from_uri(@object['conversation']),
  58. media_attachment_ids: process_attachments.take(4).map(&:id),
  59. owned_poll: process_poll,
  60. }
  61. end
  62. end
  63. def process_audience
  64. (as_array(@object['to']) + as_array(@object['cc'])).uniq.each do |audience|
  65. next if audience == ActivityPub::TagManager::COLLECTIONS[:public]
  66. # Unlike with tags, there is no point in resolving accounts we don't already
  67. # know here, because silent mentions would only be used for local access
  68. # control anyway
  69. account = account_from_uri(audience)
  70. next if account.nil? || @mentions.any? { |mention| mention.account_id == account.id }
  71. @mentions << Mention.new(account: account, silent: true)
  72. # If there is at least one silent mention, then the status can be considered
  73. # as a limited-audience status, and not strictly a direct message, but only
  74. # if we considered a direct message in the first place
  75. next unless @params[:visibility] == :direct
  76. @params[:visibility] = :limited
  77. end
  78. # If the payload was delivered to a specific inbox, the inbox owner must have
  79. # access to it, unless they already have access to it anyway
  80. return if @options[:delivered_to_account_id].nil? || @mentions.any? { |mention| mention.account_id == @options[:delivered_to_account_id] }
  81. @mentions << Mention.new(account_id: @options[:delivered_to_account_id], silent: true)
  82. return unless @params[:visibility] == :direct
  83. @params[:visibility] = :limited
  84. end
  85. def postprocess_audience_and_deliver
  86. return if @status.mentions.find_by(account_id: @options[:delivered_to_account_id])
  87. delivered_to_account = Account.find(@options[:delivered_to_account_id])
  88. @status.mentions.create(account: delivered_to_account, silent: true)
  89. @status.update(visibility: :limited) if @status.direct_visibility?
  90. return unless delivered_to_account.following?(@account)
  91. FeedInsertWorker.perform_async(@status.id, delivered_to_account.id, :home)
  92. end
  93. def attach_tags(status)
  94. @tags.each do |tag|
  95. status.tags << tag
  96. TrendingTags.record_use!(tag, status.account, status.created_at) if status.public_visibility?
  97. end
  98. @mentions.each do |mention|
  99. mention.status = status
  100. mention.save
  101. end
  102. end
  103. def process_tags
  104. return if @object['tag'].nil?
  105. as_array(@object['tag']).each do |tag|
  106. if equals_or_includes?(tag['type'], 'Hashtag')
  107. process_hashtag tag
  108. elsif equals_or_includes?(tag['type'], 'Mention')
  109. process_mention tag
  110. elsif equals_or_includes?(tag['type'], 'Emoji')
  111. process_emoji tag
  112. end
  113. end
  114. end
  115. def process_hashtag(tag)
  116. return if tag['name'].blank?
  117. hashtag = tag['name'].gsub(/\A#/, '').mb_chars.downcase
  118. hashtag = Tag.where(name: hashtag).first_or_create!(name: hashtag)
  119. return if @tags.include?(hashtag)
  120. @tags << hashtag
  121. rescue ActiveRecord::RecordInvalid
  122. nil
  123. end
  124. def process_mention(tag)
  125. return if tag['href'].blank?
  126. account = account_from_uri(tag['href'])
  127. account = ::FetchRemoteAccountService.new.call(tag['href']) if account.nil?
  128. return if account.nil?
  129. @mentions << Mention.new(account: account, silent: false)
  130. end
  131. def process_emoji(tag)
  132. return if skip_download?
  133. return if tag['name'].blank? || tag['icon'].blank? || tag['icon']['url'].blank?
  134. shortcode = tag['name'].delete(':')
  135. image_url = tag['icon']['url']
  136. uri = tag['id']
  137. updated = tag['updated']
  138. emoji = CustomEmoji.find_by(shortcode: shortcode, domain: @account.domain)
  139. return unless emoji.nil? || image_url != emoji.image_remote_url || (updated && updated >= emoji.updated_at)
  140. emoji ||= CustomEmoji.new(domain: @account.domain, shortcode: shortcode, uri: uri)
  141. emoji.image_remote_url = image_url
  142. emoji.save
  143. end
  144. def process_attachments
  145. return [] if @object['attachment'].nil?
  146. media_attachments = []
  147. as_array(@object['attachment']).each do |attachment|
  148. next if attachment['url'].blank?
  149. href = Addressable::URI.parse(attachment['url']).normalize.to_s
  150. media_attachment = MediaAttachment.create(account: @account, remote_url: href, description: attachment['name'].presence, focus: attachment['focalPoint'])
  151. media_attachments << media_attachment
  152. next if unsupported_media_type?(attachment['mediaType']) || skip_download?
  153. media_attachment.file_remote_url = href
  154. media_attachment.save
  155. end
  156. media_attachments
  157. rescue Addressable::URI::InvalidURIError => e
  158. Rails.logger.debug e
  159. media_attachments
  160. end
  161. def process_poll
  162. return unless @object['type'] == 'Question' && (@object['anyOf'].is_a?(Array) || @object['oneOf'].is_a?(Array))
  163. expires_at = begin
  164. if @object['closed'].is_a?(String)
  165. @object['closed']
  166. elsif !@object['closed'].nil? && !@object['closed'].is_a?(FalseClass)
  167. Time.now.utc
  168. else
  169. @object['endTime']
  170. end
  171. end
  172. if @object['anyOf'].is_a?(Array)
  173. multiple = true
  174. items = @object['anyOf']
  175. else
  176. multiple = false
  177. items = @object['oneOf']
  178. end
  179. @account.polls.new(
  180. multiple: multiple,
  181. expires_at: expires_at,
  182. options: items.map { |item| item['name'].presence || item['content'] },
  183. cached_tallies: items.map { |item| item.dig('replies', 'totalItems') || 0 }
  184. )
  185. end
  186. def poll_vote?
  187. return false if replied_to_status.nil? || replied_to_status.poll.nil? || !replied_to_status.local? || !replied_to_status.poll.options.include?(@object['name'])
  188. replied_to_status.poll.votes.create!(account: @account, choice: replied_to_status.poll.options.index(@object['name']), uri: @object['id'])
  189. end
  190. def resolve_thread(status)
  191. return unless status.reply? && status.thread.nil? && Request.valid_url?(in_reply_to_uri)
  192. ThreadResolveWorker.perform_async(status.id, in_reply_to_uri)
  193. end
  194. def fetch_replies(status)
  195. collection = @object['replies']
  196. return if collection.nil?
  197. replies = ActivityPub::FetchRepliesService.new.call(status, collection, false)
  198. return if replies.present?
  199. uri = value_or_id(collection)
  200. ActivityPub::FetchRepliesWorker.perform_async(status.id, uri) unless uri.nil?
  201. end
  202. def conversation_from_uri(uri)
  203. return nil if uri.nil?
  204. return Conversation.find_by(id: OStatus::TagManager.instance.unique_tag_to_local_id(uri, 'Conversation')) if OStatus::TagManager.instance.local_id?(uri)
  205. Conversation.find_by(uri: uri) || Conversation.create(uri: uri)
  206. end
  207. def visibility_from_audience
  208. if equals_or_includes?(@object['to'], ActivityPub::TagManager::COLLECTIONS[:public])
  209. :public
  210. elsif equals_or_includes?(@object['cc'], ActivityPub::TagManager::COLLECTIONS[:public])
  211. :unlisted
  212. elsif equals_or_includes?(@object['to'], @account.followers_url)
  213. :private
  214. else
  215. :direct
  216. end
  217. end
  218. def audience_includes?(account)
  219. uri = ActivityPub::TagManager.instance.uri_for(account)
  220. equals_or_includes?(@object['to'], uri) || equals_or_includes?(@object['cc'], uri)
  221. end
  222. def replied_to_status
  223. return @replied_to_status if defined?(@replied_to_status)
  224. if in_reply_to_uri.blank?
  225. @replied_to_status = nil
  226. else
  227. @replied_to_status = status_from_uri(in_reply_to_uri)
  228. @replied_to_status ||= status_from_uri(@object['inReplyToAtomUri']) if @object['inReplyToAtomUri'].present?
  229. @replied_to_status
  230. end
  231. end
  232. def in_reply_to_uri
  233. value_or_id(@object['inReplyTo'])
  234. end
  235. def text_from_content
  236. return Formatter.instance.linkify([[text_from_name, text_from_summary.presence].compact.join("\n\n"), object_url || @object['id']].join(' ')) if converted_object_type?
  237. if @object['content'].present?
  238. @object['content']
  239. elsif content_language_map?
  240. @object['contentMap'].values.first
  241. end
  242. end
  243. def text_from_summary
  244. if @object['summary'].present?
  245. @object['summary']
  246. elsif summary_language_map?
  247. @object['summaryMap'].values.first
  248. end
  249. end
  250. def text_from_name
  251. if @object['name'].present?
  252. @object['name']
  253. elsif name_language_map?
  254. @object['nameMap'].values.first
  255. end
  256. end
  257. def detected_language
  258. if content_language_map?
  259. @object['contentMap'].keys.first
  260. elsif name_language_map?
  261. @object['nameMap'].keys.first
  262. elsif summary_language_map?
  263. @object['summaryMap'].keys.first
  264. elsif supported_object_type?
  265. LanguageDetector.instance.detect(text_from_content, @account)
  266. end
  267. end
  268. def object_url
  269. return if @object['url'].blank?
  270. url_candidate = url_to_href(@object['url'], 'text/html')
  271. if invalid_origin?(url_candidate)
  272. nil
  273. else
  274. url_candidate
  275. end
  276. end
  277. def summary_language_map?
  278. @object['summaryMap'].is_a?(Hash) && !@object['summaryMap'].empty?
  279. end
  280. def content_language_map?
  281. @object['contentMap'].is_a?(Hash) && !@object['contentMap'].empty?
  282. end
  283. def name_language_map?
  284. @object['nameMap'].is_a?(Hash) && !@object['nameMap'].empty?
  285. end
  286. def unsupported_media_type?(mime_type)
  287. mime_type.present? && !(MediaAttachment::IMAGE_MIME_TYPES + MediaAttachment::VIDEO_MIME_TYPES).include?(mime_type)
  288. end
  289. def skip_download?
  290. return @skip_download if defined?(@skip_download)
  291. @skip_download ||= DomainBlock.find_by(domain: @account.domain)&.reject_media?
  292. end
  293. def invalid_origin?(url)
  294. return true if unsupported_uri_scheme?(url)
  295. needle = Addressable::URI.parse(url).host
  296. haystack = Addressable::URI.parse(@account.uri).host
  297. !haystack.casecmp(needle).zero?
  298. end
  299. def reply_to_local?
  300. !replied_to_status.nil? && replied_to_status.account.local?
  301. end
  302. def related_to_local_activity?
  303. fetch? || followed_by_local_accounts? || requested_through_relay? ||
  304. responds_to_followed_account? || addresses_local_accounts?
  305. end
  306. def responds_to_followed_account?
  307. !replied_to_status.nil? && (replied_to_status.account.local? || replied_to_status.account.passive_relationships.exists?)
  308. end
  309. def addresses_local_accounts?
  310. return true if @options[:delivered_to_account_id]
  311. local_usernames = (as_array(@object['to']) + as_array(@object['cc'])).uniq.select { |uri| ActivityPub::TagManager.instance.local_uri?(uri) }.map { |uri| ActivityPub::TagManager.instance.uri_to_local_id(uri, :username) }
  312. return false if local_usernames.empty?
  313. Account.local.where(username: local_usernames).exists?
  314. end
  315. def forward_for_reply
  316. return unless @json['signature'].present? && reply_to_local?
  317. ActivityPub::RawDistributionWorker.perform_async(Oj.dump(@json), replied_to_status.account_id, [@account.preferred_inbox_url])
  318. end
  319. def lock_options
  320. { redis: Redis.current, key: "create:#{@object['id']}" }
  321. end
  322. end