server/controllers/tracker.ts

   1 import { Server as TrackerServer } from 'bittorrent-tracker'
   2 import express from 'express'
   3 import { createServer } from 'http'
   4 import LRUCache from 'lru-cache'
   5 import proxyAddr from 'proxy-addr'
   6 import { WebSocketServer } from 'ws'
   7 import { logger } from '../helpers/logger'
   8 import { CONFIG } from '../initializers/config'
   9 import { LRU_CACHE, TRACKER_RATE_LIMITS } from '../initializers/constants'
  10 import { VideoFileModel } from '../models/video/video-file'
  11 import { VideoStreamingPlaylistModel } from '../models/video/video-streaming-playlist'
  12
  13 const trackerRouter = express.Router()
  14
  15 const blockedIPs = new LRUCache<string, boolean>({
  16   max: LRU_CACHE.TRACKER_IPS.MAX_SIZE,
  17   ttl: TRACKER_RATE_LIMITS.BLOCK_IP_LIFETIME
  18 })
  19
  20 let peersIps = {}
  21 let peersIpInfoHash = {}
  22 runPeersChecker()
  23
  24 const trackerServer = new TrackerServer({
  25   http: false,
  26   udp: false,
  27   ws: false,
  28   filter: async function (infoHash, params, cb) {
  29     if (CONFIG.TRACKER.ENABLED === false) {
  30       return cb(new Error('Tracker is disabled on this instance.'))
  31     }
  32
  33     let ip: string
  34
  35     if (params.type === 'ws') {
  36       ip = params.ip
  37     } else {
  38       ip = params.httpReq.ip
  39     }
  40
  41     const key = ip + '-' + infoHash
  42
  43     peersIps[ip] = peersIps[ip] ? peersIps[ip] + 1 : 1
  44     peersIpInfoHash[key] = peersIpInfoHash[key] ? peersIpInfoHash[key] + 1 : 1
  45
  46     if (CONFIG.TRACKER.REJECT_TOO_MANY_ANNOUNCES && peersIpInfoHash[key] > TRACKER_RATE_LIMITS.ANNOUNCES_PER_IP_PER_INFOHASH) {
  47       return cb(new Error(`Too many requests (${peersIpInfoHash[key]} of ip ${ip} for torrent ${infoHash}`))
  48     }
  49
  50     try {
  51       if (CONFIG.TRACKER.PRIVATE === false) return cb()
  52
  53       const videoFileExists = await VideoFileModel.doesInfohashExistCached(infoHash)
  54       if (videoFileExists === true) return cb()
  55
  56       const playlistExists = await VideoStreamingPlaylistModel.doesInfohashExistCached(infoHash)
  57       if (playlistExists === true) return cb()
  58
  59       cb(new Error(`Unknown infoHash ${infoHash} requested by ip ${ip}`))
  60
  61       // Close socket connection and block IP for a few time
  62       if (params.type === 'ws') {
  63         blockedIPs.set(ip, true)
  64
  65         // setTimeout to wait filter response
  66         setTimeout(() => params.socket.close(), 0)
  67       }
  68     } catch (err) {
  69       logger.error('Error in tracker filter.', { err })
  70       return cb(err)
  71     }
  72   }
  73 })
  74
  75 if (CONFIG.TRACKER.ENABLED !== false) {
  76   trackerServer.on('error', function (err) {
  77     logger.error('Error in tracker.', { err })
  78   })
  79
  80   trackerServer.on('warning', function (err) {
  81     const message = err.message || ''
  82
  83     if (CONFIG.LOG.LOG_TRACKER_UNKNOWN_INFOHASH === false && message.includes('Unknown infoHash')) {
  84       return
  85     }
  86
  87     logger.warn('Warning in tracker.', { err })
  88   })
  89 }
  90
  91 const onHttpRequest = trackerServer.onHttpRequest.bind(trackerServer)
  92 trackerRouter.get('/tracker/announce', (req, res) => onHttpRequest(req, res, { action: 'announce' }))
  93 trackerRouter.get('/tracker/scrape', (req, res) => onHttpRequest(req, res, { action: 'scrape' }))
  94
  95 function createWebsocketTrackerServer (app: express.Application) {
  96   const server = createServer(app)
  97   const wss = new WebSocketServer({ noServer: true })
  98
  99   wss.on('connection', function (ws, req) {
 100     ws['ip'] = proxyAddr(req, CONFIG.TRUST_PROXY)
 101
 102     trackerServer.onWebSocketConnection(ws)
 103   })
 104
 105   server.on('upgrade', (request: express.Request, socket, head) => {
 106     if (request.url === '/tracker/socket') {
 107       const ip = proxyAddr(request, CONFIG.TRUST_PROXY)
 108
 109       if (blockedIPs.has(ip)) {
 110         logger.debug('Blocking IP %s from tracker.', ip)
 111
 112         socket.write('HTTP/1.1 403 Forbidden\r\n\r\n')
 113         socket.destroy()
 114         return
 115       }
 116
 117       // FIXME: typings
 118       return wss.handleUpgrade(request, socket as any, head, ws => wss.emit('connection', ws, request))
 119     }
 120
 121     // Don't destroy socket, we have Socket.IO too
 122   })
 123
 124   return server
 125 }
 126
 127 // ---------------------------------------------------------------------------
 128
 129 export {
 130   trackerRouter,
 131   createWebsocketTrackerServer
 132 }
 133
 134 // ---------------------------------------------------------------------------
 135
 136 function runPeersChecker () {
 137   setInterval(() => {
 138     logger.debug('Checking peers.')
 139
 140     for (const ip of Object.keys(peersIpInfoHash)) {
 141       if (peersIps[ip] > TRACKER_RATE_LIMITS.ANNOUNCES_PER_IP) {
 142         logger.warn('Peer %s made abnormal requests (%d).', ip, peersIps[ip])
 143       }
 144     }
 145
 146     peersIpInfoHash = {}
 147     peersIps = {}
 148   }, TRACKER_RATE_LIMITS.INTERVAL)
 149 }