server/controllers/tracker.ts

   1 import { Server as TrackerServer } from 'bittorrent-tracker'
   2 import express from 'express'
   3 import { createServer } from 'http'
   4 import proxyAddr from 'proxy-addr'
   5 import { WebSocketServer } from 'ws'
   6 import { Redis } from '@server/lib/redis'
   7 import { logger } from '../helpers/logger'
   8 import { CONFIG } from '../initializers/config'
   9 import { 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 let peersIps = {}
  16 let peersIpInfoHash = {}
  17 runPeersChecker()
  18
  19 const trackerServer = new TrackerServer({
  20   http: false,
  21   udp: false,
  22   ws: false,
  23   filter: async function (infoHash, params, cb) {
  24     if (CONFIG.TRACKER.ENABLED === false) {
  25       return cb(new Error('Tracker is disabled on this instance.'))
  26     }
  27
  28     let ip: string
  29
  30     if (params.type === 'ws') {
  31       ip = params.ip
  32     } else {
  33       ip = params.httpReq.ip
  34     }
  35
  36     const key = ip + '-' + infoHash
  37
  38     peersIps[ip] = peersIps[ip] ? peersIps[ip] + 1 : 1
  39     peersIpInfoHash[key] = peersIpInfoHash[key] ? peersIpInfoHash[key] + 1 : 1
  40
  41     if (CONFIG.TRACKER.REJECT_TOO_MANY_ANNOUNCES && peersIpInfoHash[key] > TRACKER_RATE_LIMITS.ANNOUNCES_PER_IP_PER_INFOHASH) {
  42       return cb(new Error(`Too many requests (${peersIpInfoHash[key]} of ip ${ip} for torrent ${infoHash}`))
  43     }
  44
  45     try {
  46       if (CONFIG.TRACKER.PRIVATE === false) return cb()
  47
  48       const videoFileExists = await VideoFileModel.doesInfohashExistCached(infoHash)
  49       if (videoFileExists === true) return cb()
  50
  51       const playlistExists = await VideoStreamingPlaylistModel.doesInfohashExistCached(infoHash)
  52       if (playlistExists === true) return cb()
  53
  54       cb(new Error(`Unknown infoHash ${infoHash} requested by ip ${ip}`))
  55
  56       // Close socket connection and block IP for a few time
  57       if (params.type === 'ws') {
  58         Redis.Instance.setTrackerBlockIP(ip)
  59           .catch(err => logger.error('Cannot set tracker block ip.', { err }))
  60
  61         // setTimeout to wait filter response
  62         setTimeout(() => params.socket.close(), 0)
  63       }
  64     } catch (err) {
  65       logger.error('Error in tracker filter.', { err })
  66       return cb(err)
  67     }
  68   }
  69 })
  70
  71 if (CONFIG.TRACKER.ENABLED !== false) {
  72   trackerServer.on('error', function (err) {
  73     logger.error('Error in tracker.', { err })
  74   })
  75
  76   trackerServer.on('warning', function (err) {
  77     const message = err.message || ''
  78
  79     if (CONFIG.LOG.LOG_TRACKER_UNKNOWN_INFOHASH === false && message.includes('Unknown infoHash')) {
  80       return
  81     }
  82
  83     logger.warn('Warning in tracker.', { err })
  84   })
  85 }
  86
  87 const onHttpRequest = trackerServer.onHttpRequest.bind(trackerServer)
  88 trackerRouter.get('/tracker/announce', (req, res) => onHttpRequest(req, res, { action: 'announce' }))
  89 trackerRouter.get('/tracker/scrape', (req, res) => onHttpRequest(req, res, { action: 'scrape' }))
  90
  91 function createWebsocketTrackerServer (app: express.Application) {
  92   const server = createServer(app)
  93   const wss = new WebSocketServer({ noServer: true })
  94
  95   wss.on('connection', function (ws, req) {
  96     ws['ip'] = proxyAddr(req, CONFIG.TRUST_PROXY)
  97
  98     trackerServer.onWebSocketConnection(ws)
  99   })
 100
 101   server.on('upgrade', (request: express.Request, socket, head) => {
 102     if (request.url === '/tracker/socket') {
 103       const ip = proxyAddr(request, CONFIG.TRUST_PROXY)
 104
 105       Redis.Instance.doesTrackerBlockIPExist(ip)
 106         .then(result => {
 107           if (result === true) {
 108             logger.debug('Blocking IP %s from tracker.', ip)
 109
 110             socket.write('HTTP/1.1 403 Forbidden\r\n\r\n')
 111             socket.destroy()
 112             return
 113           }
 114
 115           // FIXME: typings
 116           return wss.handleUpgrade(request, socket as any, head, ws => wss.emit('connection', ws, request))
 117         })
 118         .catch(err => logger.error('Cannot check if tracker block ip exists.', { err }))
 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 }