More robust AP handlers

server/core/initializers/constants.ts

@@ -578,6 +578,10 @@ export const REMOTE_DOWNLOADS = {
   MAX_PER_HOST_PER_VIDEO: 500
 }
 
+export const REMOTE_VIEWS = {
+  DEDUPLICATION_LIFETIME: 60000 * 60 * 24 // 24 hours
+}
+
 export const MAX_LOCAL_VIEWER_WATCH_SECTIONS = 100
 
 export let CONTACT_FORM_LIFETIME = 60000 * 60 // 1 hour

server/core/lib/activitypub/crawl.ts

@@ -30,7 +30,7 @@ export async function crawlCollectionPage<T> (argUrl: string, handler: HandlerFu
     if (typeof nextLink === 'string') {
       // Don't crawl ourselves
       const remoteHost = new URL(nextLink).host
-      if (remoteHost === WEBSERVER.HOST) continue
+      if (remoteHost === WEBSERVER.HOST) break
 
       url = nextLink
 

server/core/lib/activitypub/process/process-announce.ts

@@ -1,12 +1,14 @@
 import { ActivityAnnounce } from '@peertube/peertube-models'
 import { getAPId } from '@server/lib/activitypub/activity.js'
 import { retryTransactionWrapper } from '../../../helpers/database-utils.js'
+import { logger } from '../../../helpers/logger.js'
 import { sequelizeTypescript } from '../../../initializers/database.js'
 import { VideoShareModel } from '../../../models/video/video-share.js'
 import { APProcessorOptions } from '../../../types/activitypub-processor.model.js'
 import { MActorSignature } from '../../../types/models/index.js'
 import { Notifier } from '../../notifier/index.js'
 import { forwardVideoRelatedActivity } from '../send/shared/send-utils.js'
+import { checkUrlsSameHost } from '../url.js'
 import { maybeGetOrCreateAPVideo } from '../videos/index.js'
 
 async function processAnnounceActivity (options: APProcessorOptions<ActivityAnnounce>) {
@@ -31,6 +33,12 @@ export {
 async function processVideoShare (actorAnnouncer: MActorSignature, activity: ActivityAnnounce, notify: boolean) {
   const objectUri = getAPId(activity.object)
 
+  // The share is identified by the announce URL, so don't let an actor use (and so squat) the announce URL of another host
+  if (checkUrlsSameHost(activity.id, actorAnnouncer.url) !== true) {
+    logger.warn('Ignoring announce %s that has not the same host than actor %s.', activity.id, actorAnnouncer.url)
+    return
+  }
+
   const { video, created: videoCreated } = await maybeGetOrCreateAPVideo({ videoObject: objectUri })
   if (!video) return
 

server/core/lib/activitypub/process/process-create.ts

@@ -27,6 +27,7 @@ import { createOrUpdateLocalVideoViewer } from '../local-video-viewer.js'
 import { createOrUpdateVideoPlaylist } from '../playlists/index.js'
 import { sendReplyApproval } from '../send/send-reply-approval.js'
 import { forwardVideoRelatedActivity } from '../send/shared/send-utils.js'
+import { checkUrlsSameHost, getLocalApproveReplyActivityPubUrl } from '../url.js'
 import { resolveThread } from '../video-comments.js'
 import { canVideoBeFederated, getOrCreateAPVideo } from '../videos/index.js'
 
@@ -50,7 +51,10 @@ async function processCreateActivity (options: APProcessorOptions<ActivityCreate
   }
 
   if (activityType === 'WatchAction') {
-    return retryTransactionWrapper(processCreateWatchAction, activityObject)
+    // Watch actions are only sent to the inbox of the video origin, so we never have to process a fetched one
+    if (options.fromFetch) return
+
+    return retryTransactionWrapper(processCreateWatchAction, activityObject, byActor)
   }
 
   if (activityType === 'CacheFile') {
@@ -107,9 +111,14 @@ async function processCreateCacheFile (
   }
 }
 
-async function processCreateWatchAction (watchAction: WatchActionObject) {
+async function processCreateWatchAction (watchAction: WatchActionObject, byActor: MActorSignature) {
   if (watchAction.actionStatus !== 'CompletedActionStatus') return
 
+  if (checkUrlsSameHost(watchAction.id, byActor.url) !== true) {
+    logger.warn('Ignoring watch action %s that has not the same host than actor %s.', watchAction.id, byActor.url)
+    return
+  }
+
   const video = await VideoModel.loadByUrl(watchAction.object)
   if (!video || video.remote) return
 
@@ -170,7 +179,7 @@ async function processCreateVideoComment (
     }
 
     // New comment or re-sent after an approval -> forward comment
-    if (comment.heldForReview === false && (created || commentObject.replyApproval)) {
+    if (comment.heldForReview === false && (created || await consumeReplyApproval(video, comment, commentObject))) {
       // Don't resend the activity to the sender
       const exceptions = [ byActor ]
 
@@ -181,6 +190,31 @@ async function processCreateVideoComment (
   if (created) Notifier.Instance.notifyOnNewComment(comment)
 }
 
+// The origin instance re-sends us the comment when we approved the reply
+// Ensure it's the approval we sent and that we didn't already process it
+async function consumeReplyApproval (
+  video: MVideoAccountLightBlacklistAllFiles,
+  comment: MCommentOwnerVideo,
+  commentObject: VideoCommentObject
+) {
+  if (!commentObject.replyApproval) return false
+
+  const expectedApproval = getLocalApproveReplyActivityPubUrl(video, comment)
+
+  if (commentObject.replyApproval !== expectedApproval) {
+    logger.warn('Do not forward comment %s that has an unknown reply approval %s.', comment.url, commentObject.replyApproval)
+    return false
+  }
+
+  // We already forwarded this comment after this approval
+  if (comment.replyApproval === expectedApproval) return false
+
+  comment.replyApproval = expectedApproval
+  await comment.save()
+
+  return true
+}
+
 async function processCreatePlaylist (
   activity: ActivityCreate<PlaylistObject | string>,
   playlistObject: PlaylistObject,

server/core/lib/activitypub/process/process-reply-approval.ts

@@ -12,8 +12,12 @@ export function processReplyApprovalFactory (type: Extract<ActivityType, 'Approv
     const { activity, byActor } = options
     const comment = await VideoCommentModel.loadByUrlAndPopulateAccountAndVideoAndReply(activity.object)
 
-    if (!comment || comment.isDeleted()) {
-      throw new Error(`Cannot process reply approval on comment ${comment.url} that doesn't exist`)
+    if (!comment) {
+      throw new Error(`Cannot process reply approval on comment ${activity.object} that doesn't exist`)
+    }
+
+    if (comment.isDeleted()) {
+      throw new Error(`Cannot process reply approval on deleted comment ${comment.url}`)
     }
 
     if (comment.isLocal() !== true) {

server/core/lib/activitypub/process/process-undo.ts

@@ -173,6 +173,11 @@ function processUndoAnnounce (byActor: MActorSignature, announceActivity: Activi
 function processUndoFollow (follower: MActorSignature, followActivity: ActivityFollow) {
   return sequelizeTypescript.transaction(async t => {
     const following = await ActorModel.loadByUrlAndPopulateAccountAndChannel(followActivity.object, t)
+    if (!following) {
+      logger.warn('Unknown actor %s to undo the follow of %s.', followActivity.object, follower.url)
+      return
+    }
+
     const actorFollow = await ActorFollowModel.loadByActorAndTarget(follower.id, following.id, t)
 
     if (!actorFollow) {

server/core/lib/activitypub/process/process-view.ts

@@ -1,4 +1,5 @@
 import { ActivityView } from '@peertube/peertube-models'
+import { logger, loggerTagsFactory } from '@server/helpers/logger.js'
 import { MAX_REMOTE_VIEWERS_COUNTER } from '@server/initializers/constants.js'
 import { VideoStatsManager } from '@server/lib/stats/video-stats-manager.js'
 import { APProcessorOptions } from '../../../types/activitypub-processor.model.js'
@@ -7,6 +8,8 @@ import { forwardVideoRelatedActivity } from '../send/shared/send-utils.js'
 import { checkUrlsSameHost } from '../url.js'
 import { getOrCreateAPVideo } from '../videos/index.js'
 
+const lTags = loggerTagsFactory('ap', 'view')
+
 async function processViewActivity (options: APProcessorOptions<ActivityView>) {
   const { activity, byActor } = options
 
@@ -24,13 +27,19 @@ export {
 async function processCreateView (activity: ActivityView, byActor: MActorSignature) {
   const videoObject = activity.object
 
+  // The view URL is built from the actor URL, so don't accept a view of another host
+  if (checkUrlsSameHost(activity.id, byActor.url) !== true) {
+    logger.warn('Ignoring view %s that has not the same host than actor %s.', activity.id, byActor.url, lTags())
+    return
+  }
+
   const { video } = await getOrCreateAPVideo({
     videoObject,
     fetchType: 'with-blacklist',
     allowRefresh: false
   })
 
-  await VideoStatsManager.Instance.processRemoteView({
+  const isNew = await VideoStatsManager.Instance.processRemoteView({
     video,
     viewerId: activity.id,
 
@@ -40,7 +49,8 @@ async function processCreateView (activity: ActivityView, byActor: MActorSignatu
     viewerResultCounter: getViewerResultCounter(activity, video, byActor)
   })
 
-  if (video.isLocal()) {
+  // Only forward a view we didn't already know, so a replayed activity can't be amplified to all our followers
+  if (isNew && video.isLocal()) {
     // Forward the view but don't resend the activity to the sender
     const exceptions = [ byActor ]
     await forwardVideoRelatedActivity({ activity, transaction: undefined, followersException: exceptions, video, parallelizable: true })

server/core/lib/activitypub/process/process.ts

@@ -53,12 +53,22 @@ export async function processActivities (
   const actorsCache: { [ url: string ]: MActorSignature } = {}
 
   for (const activity of activities) {
+    if (!activity || typeof activity !== 'object') {
+      logger.warn('Cannot process invalid activity.', { activity })
+      continue
+    }
+
     if (!signatureActor && [ 'Create', 'Announce', 'Like' ].includes(activity.type) === false) {
       logger.error('Cannot process activity %s (type: %s) without the actor signature.', activity.id, activity.type)
       continue
     }
 
+    let byActor: MActorSignature
+
+    // Activities fetched from a remote instance are not validated, so a malformed one must not prevent processing the others
+    try {
       const actorUrl = getAPId(activity.actor)
+      if (!actorUrl) throw new Error('Cannot find the actor URL of the activity')
 
       // When we fetch remote data, we don't have signature
       if (signatureActor && actorUrl !== signatureActor.url) {
@@ -71,8 +81,12 @@ export async function processActivities (
         continue
       }
 
-    const byActor = signatureActor || actorsCache[actorUrl] || await getOrCreateAPActor(actorUrl)
+      byActor = signatureActor || actorsCache[actorUrl] || await getOrCreateAPActor(actorUrl)
       actorsCache[actorUrl] = byActor
+    } catch (err) {
+      logger.warn('Cannot get the actor of activity %s, skipping.', activity.id, { err })
+      continue
+    }
 
     const activityProcessor = processActivity[activity.type]
     if (activityProcessor === undefined) {

server/core/lib/stats/shared/video-stats.ts

@@ -1,7 +1,7 @@
 import { buildUUID } from '@peertube/peertube-node-utils'
 import { logger, loggerTagsFactory } from '@server/helpers/logger.js'
 import { CONFIG } from '@server/initializers/config.js'
-import { REMOTE_DOWNLOADS, VIEW_LIFETIME } from '@server/initializers/constants.js'
+import { REMOTE_DOWNLOADS, REMOTE_VIEWS, VIEW_LIFETIME } from '@server/initializers/constants.js'
 import { sendDownload } from '@server/lib/activitypub/send/send-download.js'
 import { sendView } from '@server/lib/activitypub/send/send-view.js'
 import { getCachedVideoDuration } from '@server/lib/video.js'
@@ -18,6 +18,11 @@ export class VideoStats {
     ttl: VIEW_LIFETIME.VIEW
   })
 
+  private readonly remoteViewsCache = new LRUCache<string, boolean>({
+    max: 50_000,
+    ttl: REMOTE_VIEWS.DEDUPLICATION_LIFETIME
+  })
+
   // Remote instances are trusted to report downloads of our videos, so guard against duplicated/flooded activities
   private readonly remoteDownloadsCache = new LRUCache<string, boolean>({
     max: 50_000,
@@ -58,10 +63,21 @@ export class VideoStats {
 
   async addRemoteView (options: {
     video: MVideo
+    viewerId: string | null
   }) {
-    const { video } = options
+    const { video, viewerId } = options
 
-    logger.debug('Adding remote view to video %s.', video.uuid, { ...lTags(video.uuid) })
+    logger.debug('Adding remote view to video %s.', video.uuid, { viewerId, ...lTags(video.uuid) })
+
+    if (viewerId) {
+      if (this.remoteViewsCache.has(viewerId)) {
+        logger.debug('Ignoring already processed remote view %s.', viewerId, lTags(video.uuid))
+
+        return false
+      }
+
+      this.remoteViewsCache.set(viewerId, true)
+    }
 
     await this.addView(video)
 

server/core/lib/stats/video-stats-manager.ts

@@ -71,6 +71,7 @@ export class VideoStatsManager {
     return { successView, successViewer }
   }
 
+  // Returns false if we already know this viewer/view, so the caller can avoid processing it again
   async processRemoteView (options: {
     video: MVideo
     viewerId: string | null
@@ -84,16 +85,14 @@ export class VideoStatsManager {
     // Viewer
     if (viewerExpires) {
       if (video.remote === false) {
-        this.videoViewerCounters.addRemoteViewerOnLocalVideo({ video, viewerId, viewerExpires })
-        return
+        return this.videoViewerCounters.addRemoteViewerOnLocalVideo({ video, viewerId, viewerExpires })
       }
 
-      this.videoViewerCounters.addRemoteViewerOnRemoteVideo({ video, viewerId, viewerExpires, viewerResultCounter })
-      return
+      return this.videoViewerCounters.addRemoteViewerOnRemoteVideo({ video, viewerId, viewerExpires, viewerResultCounter })
     }
 
     // Just a view
-    await this.videoStats.addRemoteView({ video })
+    return this.videoStats.addRemoteView({ video, viewerId })
   }
 
   // ---------------------------------------------------------------------------