More checks on remote downloads/views

server/core/helpers/custom-validators/activitypub/activity.ts

@@ -1,7 +1,8 @@
 import { Activity, ActivityType } from '@peertube/peertube-models'
+import { VIEW_LIFETIME } from '@server/initializers/constants.js'
 import validator from 'validator'
 import { isAbuseReasonValid } from '../abuses.js'
-import { exists } from '../misc.js'
+import { exists, isDateValid } from '../misc.js'
 import { sanitizeAndCheckActorObject } from './actor.js'
 import { isCacheFileObjectValid } from './cache-file.js'
 import { isActivityPubUrlValid, isBaseActivityValid, isObjectValid } from './misc.js'
@@ -11,14 +12,24 @@ import { sanitizeAndCheckVideoCommentObject } from './video-comments.js'
 import { sanitizeAndCheckVideoTorrentObject } from './videos.js'
 import { isWatchActionObjectValid } from './watch-action.js'
 
+const collectionTypes = new Set([ 'Collection', 'CollectionPage', 'OrderedCollection', 'OrderedCollectionPage' ])
+
 export function isRootActivityValid (activity: any) {
-  return isCollection(activity) || isActivity(activity)
+  if (!exists(activity)) return false
+
+  if (collectionTypes.has(activity.type)) return isCollection(activity)
+
+  return isActivity(activity)
 }
 
 function isCollection (activity: any) {
-  return (activity.type === 'Collection' || activity.type === 'OrderedCollection') &&
+  if (!validator.default.isInt(activity.totalItems + '', { min: 0 })) return false
-    validator.default.isInt(activity.totalItems, { min: 0 }) &&
+
-    Array.isArray(activity.items)
+  if (activity.type === 'Collection' || activity.type === 'CollectionPage') {
+    return Array.isArray(activity.items)
+  }
+
+  return Array.isArray(activity.orderedItems)
 }
 
 function isActivity (activity: any) {
@@ -77,9 +88,22 @@ export function isAnnounceActivityValid (activity: any) {
 }
 
 export function isViewActivityValid (activity: any) {
-  return isBaseActivityValid(activity, 'View') &&
+  if (!isBaseActivityValid(activity, 'View')) return false
-    isActivityPubUrlValid(activity.actor) &&
+  if (!isActivityPubUrlValid(activity.actor)) return false
-    isActivityPubUrlValid(activity.object)
+  if (!isActivityPubUrlValid(activity.object)) return false
+
+  if (exists(activity.expires)) {
+    // isDateValid() expects a string
+    if (typeof activity.expires !== 'string' || !isDateValid(activity.expires)) return false
+
+    const expires = new Date(activity.expires).getTime()
+    if (isNaN(expires)) return false
+
+    // Add a 10x margin to the max expires to allow for some clock drift or custom view lifetime between servers
+    if (expires > new Date().getTime() + (VIEW_LIFETIME.VIEWER_COUNTER * 10)) return false
+  }
+
+  return true
 }
 
 export function isCreateActivityValid (activity: any) {

server/core/initializers/constants.ts

@@ -565,10 +565,19 @@ export const VIEW_LIFETIME = {
 }
 export let VIEWER_SYNC_REDIS = 30000 // Sync viewer into redis
 
+export const MAX_REMOTE_VIEWERS_COUNTER = 1_000_000
+
 export const STATS_LIFETIME = {
   DOWNLOADS: 60000 * 60 // 1 hour
 }
 
+export const REMOTE_DOWNLOADS = {
+  DEDUPLICATION_LIFETIME: 60000 * 60 * 24, // 24 hours
+  RATE_LIMIT_LIFETIME: 60000 * 60, // 1 hour
+  // Max downloads of a specific video we accept from a specific instance in RATE_LIMIT_LIFETIME
+  MAX_PER_HOST_PER_VIDEO: 500
+}
+
 export const MAX_LOCAL_VIEWER_WATCH_SECTIONS = 100
 
 export let CONTACT_FORM_LIFETIME = 60000 * 60 // 1 hour

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

@@ -111,7 +111,7 @@ async function processCreateWatchAction (watchAction: WatchActionObject) {
   if (watchAction.actionStatus !== 'CompletedActionStatus') return
 
   const video = await VideoModel.loadByUrl(watchAction.object)
-  if (video.remote) return
+  if (!video || video.remote) return
 
   await sequelizeTypescript.transaction(async t => {
     return createOrUpdateLocalVideoViewer(watchAction, video, t)

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

@@ -1,12 +1,17 @@
-import { ActivityView } from '@peertube/peertube-models'
+import { ActivityDownload } from '@peertube/peertube-models'
+import { logger, loggerTagsFactory } from '@server/helpers/logger.js'
 import { VideoStatsManager } from '@server/lib/stats/video-stats-manager.js'
 import { APProcessorOptions } from '../../../types/activitypub-processor.model.js'
+import { MActorSignature } from '../../../types/models/index.js'
+import { checkUrlsSameHost } from '../url.js'
 import { getOrCreateAPVideo } from '../videos/index.js'
 
-async function processDownloadActivity (options: APProcessorOptions<ActivityView>) {
+const lTags = loggerTagsFactory('ap', 'download')
-  const { activity } = options
 
-  return processCreateDownload(activity)
+async function processDownloadActivity (options: APProcessorOptions<ActivityDownload>) {
+  const { activity, byActor } = options
+
+  return processCreateDownload(activity, byActor)
 }
 
 // ---------------------------------------------------------------------------
@@ -17,7 +22,7 @@ export {
 
 // ---------------------------------------------------------------------------
 
-async function processCreateDownload (activity: ActivityView) {
+async function processCreateDownload (activity: ActivityDownload, byActor: MActorSignature) {
   const videoObject = activity.object
 
   const { video } = await getOrCreateAPVideo({
@@ -26,5 +31,12 @@ async function processCreateDownload (activity: ActivityView) {
     allowRefresh: false
   })
 
-  await VideoStatsManager.Instance.processRemoteDownload({ video })
+  // An instance can tell us one of its users downloaded one of our videos
+  // But for a remote video, only its origin instance broadcasts download activities
+  if (!video.isLocal() && !checkUrlsSameHost(byActor.url, video.url)) {
+    logger.warn('Ignoring download activity %s of %s that does not come from the origin instance.', activity.id, video.url, lTags())
+    return
+  }
+
+  await VideoStatsManager.Instance.processRemoteDownload({ video, downloadId: activity.id, byActorUrl: byActor.url })
 }

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

@@ -1,8 +1,10 @@
 import { ActivityView } from '@peertube/peertube-models'
+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'
-import { MActorSignature } from '../../../types/models/index.js'
+import { MActorSignature, MVideo } from '../../../types/models/index.js'
 import { forwardVideoRelatedActivity } from '../send/shared/send-utils.js'
+import { checkUrlsSameHost } from '../url.js'
 import { getOrCreateAPVideo } from '../videos/index.js'
 
 async function processViewActivity (options: APProcessorOptions<ActivityView>) {
@@ -35,7 +37,7 @@ async function processCreateView (activity: ActivityView, byActor: MActorSignatu
     viewerExpires: activity.expires
       ? new Date(activity.expires)
       : undefined,
-    viewerResultCounter: getViewerResultCounter(activity)
+    viewerResultCounter: getViewerResultCounter(activity, video, byActor)
   })
 
   if (video.isLocal()) {
@@ -45,13 +47,16 @@ async function processCreateView (activity: ActivityView, byActor: MActorSignatu
   }
 }
 
-function getViewerResultCounter (activity: ActivityView) {
+function getViewerResultCounter (activity: ActivityView, video: MVideo, byActor: MActorSignature) {
   const result = activity.result
 
   if (!activity.expires || result?.interactionType !== 'WatchAction' || result?.type !== 'InteractionCounter') return undefined
 
+  // Only the origin instance of the video can send us a summary of all its viewers
+  if (!checkUrlsSameHost(byActor.url, video.url)) return undefined
+
   const counter = parseInt(result.userInteractionCount + '')
   if (isNaN(counter)) return undefined
 
-  return counter
+  return Math.min(Math.max(counter, 0), MAX_REMOTE_VIEWERS_COUNTER)
 }

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 { VIEW_LIFETIME } from '@server/initializers/constants.js'
+import { REMOTE_DOWNLOADS, 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,17 @@ export class VideoStats {
     ttl: VIEW_LIFETIME.VIEW
   })
 
+  // 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,
+    ttl: REMOTE_DOWNLOADS.DEDUPLICATION_LIFETIME
+  })
+
+  private readonly remoteDownloadsPerHostCache = new LRUCache<string, number>({
+    max: 10_000,
+    ttl: REMOTE_DOWNLOADS.RATE_LIMIT_LIFETIME
+  })
+
   // ---------------------------------------------------------------------------
   // Views
   // ---------------------------------------------------------------------------
@@ -116,10 +127,30 @@ export class VideoStats {
 
   async addRemoteDownload (options: {
     video: MVideoImmutable
+    downloadId: string
+    byActorUrl: string
   }) {
-    const { video } = options
+    const { video, downloadId, byActorUrl } = options
 
-    logger.debug('Adding remote download to video %s.', video.uuid, { ...lTags(video.uuid) })
+    logger.debug('Adding remote download to video %s.', video.uuid, { downloadId, ...lTags(video.uuid) })
+
+    if (this.remoteDownloadsCache.has(downloadId)) {
+      logger.debug('Ignoring already processed remote download %s.', downloadId, lTags(video.uuid))
+
+      return false
+    }
+    this.remoteDownloadsCache.set(downloadId, true)
+
+    // We can't check a remote instance really downloaded the video, so at least limit how much it can inflate our counter
+    const rateLimitKey = new URL(byActorUrl).host.toLowerCase() + '-' + video.id
+    const hostDownloads = (this.remoteDownloadsPerHostCache.get(rateLimitKey) || 0) + 1
+    this.remoteDownloadsPerHostCache.set(rateLimitKey, hostDownloads, { noUpdateTTL: true })
+
+    if (hostDownloads > REMOTE_DOWNLOADS.MAX_PER_HOST_PER_VIDEO) {
+      logger.warn('Too many remote downloads of video %s sent by %s, ignoring.', video.uuid, byActorUrl, lTags(video.uuid))
+
+      return false
+    }
 
     await this.addDownload(video)
 

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

@@ -149,6 +149,10 @@ export class VideoViewerCounters {
     let watchers = this.viewersPerVideo.get(video.id)
 
     if (!watchers || replaceCurrentViewers) {
+      for (const watcher of watchers || []) {
+        this.idToViewer.delete(watcher.id)
+      }
+
       watchers = []
       this.viewersPerVideo.set(video.id, watchers)
     }

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

@@ -110,12 +110,14 @@ export class VideoStatsManager {
 
   async processRemoteDownload (options: {
     video: MVideoImmutable
+    downloadId: string
+    byActorUrl: string
   }) {
-    const { video } = options
+    const { video, downloadId, byActorUrl } = options
 
     logger.debug('Processing remote download for %s.', video.url, lTags())
 
-    await this.videoStats.addRemoteDownload({ video })
+    await this.videoStats.addRemoteDownload({ video, downloadId, byActorUrl })
   }
 
   // ---------------------------------------------------------------------------