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     if (CONFIG.LOG.LOG_TRACKER_UNKNOWN_INFOHASH) {
  78       const message = err.message || ''
  79       if (message.includes('Unknown infoHash')) return
  80     }
  81
  82     logger.warn('Warning in tracker.', { err })
  83   })
  84 }
  85
  86 const onHttpRequest = trackerServer.onHttpRequest.bind(trackerServer)
  87 trackerRouter.get('/tracker/announce', (req, res) => onHttpRequest(req, res, { action: 'announce' }))
  88 trackerRouter.get('/tracker/scrape', (req, res) => onHttpRequest(req, res, { action: 'scrape' }))
  89
  90 function createWebsocketTrackerServer (app: express.Application) {
  91   const server = createServer(app)
  92   const wss = new WebSocketServer({ noServer: true })
  93
  94   wss.on('connection', function (ws, req) {
  95     ws['ip'] = proxyAddr(req, CONFIG.TRUST_PROXY)
  96
  97     trackerServer.onWebSocketConnection(ws)
  98   })
  99
 100   server.on('upgrade', (request: express.Request, socket, head) => {
 101     if (request.url === '/tracker/socket') {
 102       const ip = proxyAddr(request, CONFIG.TRUST_PROXY)
 103
 104       Redis.Instance.doesTrackerBlockIPExist(ip)
 105         .then(result => {
 106           if (result === true) {
 107             logger.debug('Blocking IP %s from tracker.', ip)
 108
 109             socket.write('HTTP/1.1 403 Forbidden\r\n\r\n')
 110             socket.destroy()
 111             return
 112           }
 113
 114           // FIXME: typings
 115           return wss.handleUpgrade(request, socket as any, head, ws => wss.emit('connection', ws, request))
 116         })
 117         .catch(err => logger.error('Cannot check if tracker block ip exists.', { err }))
 118     }
 119
 120     // Don't destroy socket, we have Socket.IO too
 121   })
 122
 123   return server
 124 }
 125
 126 // ---------------------------------------------------------------------------
 127
 128 export {
 129   trackerRouter,
 130   createWebsocketTrackerServer
 131 }
 132
 133 // ---------------------------------------------------------------------------
 134
 135 function runPeersChecker () {
 136   setInterval(() => {
 137     logger.debug('Checking peers.')
 138
 139     for (const ip of Object.keys(peersIpInfoHash)) {
 140       if (peersIps[ip] > TRACKER_RATE_LIMITS.ANNOUNCES_PER_IP) {
 141         logger.warn('Peer %s made abnormal requests (%d).', ip, peersIps[ip])
 142       }
 143     }
 144
 145     peersIpInfoHash = {}
 146     peersIps = {}
 147   }, TRACKER_RATE_LIMITS.INTERVAL)
 148 }