vphttp.py

   1 #!/usr/bin/env python2
   2 # -*- coding: utf-8 -*-
   3 '''
   4 VPProxy: HTTP/HLS Stream to HTTP Multiplexing Proxy
   5
   6 Based on AceProxy (https://github.com/ValdikSS/AceProxy) design
   7 '''
   8 import traceback
   9 import gevent
  10 import gevent.monkey
  11 # Monkeypatching and all the stuff
  12 gevent.monkey.patch_all()
  13 # Startup delay for daemon restart
  14 gevent.sleep(5)
  15 import glob
  16 import os
  17 import signal
  18 import sys
  19 import logging
  20 import psutil
  21 import BaseHTTPServer
  22 import SocketServer
  23 from socket import error as SocketException
  24 from socket import SHUT_RDWR
  25 import urllib2
  26 import hashlib
  27 import vpconfig
  28 from vpconfig import VPConfig
  29 import vlcclient
  30 import gc
  31 import plugins.modules.ipaddr as ipaddr
  32 from clientcounter import ClientCounter
  33 from plugins.modules.PluginInterface import VPProxyPlugin
  34 try:
  35     import pwd
  36     import grp
  37 except ImportError:
  38     pass
  39
  40
  41
  42 class HTTPHandler(BaseHTTPServer.BaseHTTPRequestHandler):
  43
  44     requestlist = []
  45
  46     def handle_one_request(self):
  47         '''
  48         Add request to requestlist, handle request and remove from the list
  49         '''
  50         HTTPHandler.requestlist.append(self)
  51         BaseHTTPServer.BaseHTTPRequestHandler.handle_one_request(self)
  52         HTTPHandler.requestlist.remove(self)
  53
  54     def closeConnection(self):
  55         '''
  56         Disconnecting client
  57         '''
  58         if self.clientconnected:
  59             self.clientconnected = False
  60             try:
  61                 self.wfile.close()
  62                 self.rfile.close()
  63                 self.connection.shutdown(SHUT_RDWR)
  64             except:
  65                 pass
  66
  67     def dieWithError(self, errorcode=500):
  68         '''
  69         Close connection with error
  70         '''
  71         logging.warning("Dying with error")
  72         if self.clientconnected:
  73             self.send_error(errorcode)
  74             self.end_headers()
  75             self.closeConnection()
  76
  77     def proxyReadWrite(self):
  78         '''
  79         Read video stream and send it to client
  80         '''
  81         logger = logging.getLogger('http_proxyReadWrite')
  82         logger.debug("Started")
  83
  84         self.vlcstate = True
  85         self.streamstate = True
  86
  87         try:
  88             while True:
  89
  90                 if not self.clientconnected:
  91                     logger.debug("Client is not connected, terminating")
  92                     break
  93
  94                 data = self.video.read(4096)
  95                 if data and self.clientconnected:
  96                     self.wfile.write(data)
  97                 else:
  98                     logger.warning("Video connection closed")
  99                     break
 100
 101         except SocketException:
 102             # Video connection dropped
 103             logger.warning("Video connection dropped")
 104         finally:
 105             self.video.close()
 106             self.closeConnection()
 107
 108     def hangDetector(self):
 109         '''
 110         Detect client disconnection while in the middle of something
 111         or just normal connection close.
 112         '''
 113         logger = logging.getLogger('http_hangDetector')
 114         try:
 115             while True:
 116                 if not self.rfile.read():
 117                     break
 118         except:
 119             pass
 120         finally:
 121             self.clientconnected = False
 122             logger.debug("Client disconnected")
 123             try:
 124                 self.requestgreenlet.kill()
 125             except:
 126                 pass
 127             finally:
 128                 gevent.sleep()
 129             return
 130
 131     def do_HEAD(self):
 132         return self.do_GET(headers_only=True)
 133
 134     def do_GET(self, headers_only=False):
 135         '''
 136         GET request handler
 137         '''
 138         logger = logging.getLogger('http_HTTPHandler')
 139         self.clientconnected = True
 140         # Don't wait videodestroydelay if error happened
 141         self.errorhappened = True
 142         # Headers sent flag for fake headers UAs
 143         self.headerssent = False
 144         # Current greenlet
 145         self.requestgreenlet = gevent.getcurrent()
 146         # Connected client IP address
 147         self.clientip = self.request.getpeername()[0]
 148
 149         if VPConfig.firewall:
 150             # If firewall enabled
 151             self.clientinrange = any(map(lambda i: ipaddr.IPAddress(self.clientip) \
 152                                 in ipaddr.IPNetwork(i), VPConfig.firewallnetranges))
 153
 154             if (VPConfig.firewallblacklistmode and self.clientinrange) or \
 155                 (not VPConfig.firewallblacklistmode and not self.clientinrange):
 156                     logger.info('Dropping connection from ' + self.clientip + ' due to ' + \
 157                                 'firewall rules')
 158                     self.dieWithError(403)  # 403 Forbidden
 159                     return
 160
 161         logger.info("Accepted connection from " + self.clientip + " path " + self.path)
 162
 163         try:
 164             self.splittedpath = self.path.split('/')
 165             self.reqtype = self.splittedpath[1].lower()
 166             # If first parameter is 'pid' or 'torrent' or it should be handled
 167             # by plugin
 168             if not (self.reqtype in ('get','mp4','ogg','ogv') or self.reqtype in VPStuff.pluginshandlers):
 169                 self.dieWithError(400)  # 400 Bad Request
 170                 return
 171         except IndexError:
 172             self.dieWithError(400)  # 400 Bad Request
 173             return
 174
 175         # Handle request with plugin handler
 176         if self.reqtype in VPStuff.pluginshandlers:
 177             try:
 178                 VPStuff.pluginshandlers.get(self.reqtype).handle(self)
 179             except Exception as e:
 180                 logger.error('Plugin exception: ' + repr(e))
 181                 logger.error(traceback.format_exc())
 182                 self.dieWithError()
 183             finally:
 184                 self.closeConnection()
 185                 return
 186         self.handleRequest(headers_only)
 187
 188     def handleRequest(self, headers_only):
 189
 190         # Limit concurrent connections
 191         if 0 < VPConfig.maxconns <= VPStuff.clientcounter.total:
 192             logger.debug("Maximum connections reached, can't serve this")
 193             self.dieWithError(503)  # 503 Service Unavailable
 194             return
 195
 196         # Pretend to work fine with Fake UAs or HEAD request.
 197         useragent = self.headers.get('User-Agent')
 198         logger.debug("HTTP User Agent:"+useragent)
 199         fakeua = useragent and useragent in VPConfig.fakeuas
 200         if headers_only or fakeua:
 201             if fakeua:
 202                 logger.debug("Got fake UA: " + self.headers.get('User-Agent'))
 203             # Return 200 and exit
 204             self.send_response(200)
 205             self.send_header("Content-Type", "video/mpeg")
 206             self.end_headers()
 207             self.closeConnection()
 208             return
 209
 210         self.path_unquoted = urllib2.unquote('/'.join(self.splittedpath[2:]))
 211         # Make list with parameters
 212         self.params = list()
 213         for i in xrange(3, 8):
 214             try:
 215                 self.params.append(int(self.splittedpath[i]))
 216             except (IndexError, ValueError):
 217                 self.params.append('0')
 218
 219         # Adding client to clientcounter
 220         clients = VPStuff.clientcounter.add(self.reqtype+'\\'+self.path_unquoted, self.clientip)
 221         # If we are the one client, but sucessfully got vp instance from clientcounter,
 222         # then somebody is waiting in the videodestroydelay state
 223
 224         # Check if we are first client
 225         if VPStuff.clientcounter.get(self.reqtype+'\\'+self.path_unquoted)==1:
 226             logger.debug("First client, should create VLC session")
 227             shouldcreatevp = True
 228         else:
 229             logger.debug("Can reuse existing session")
 230             shouldcreatevp = False
 231
 232         self.vlcid = hashlib.md5(self.reqtype+'\\'+self.path_unquoted).hexdigest()
 233
 234         # Send fake headers if this User-Agent is in fakeheaderuas tuple
 235         if fakeua:
 236             logger.debug(
 237                 "Sending fake headers for " + useragent)
 238             self.send_response(200)
 239             self.send_header('Cache-Control','no-cache');
 240             if self.reqtype in ("ogg","ogv"):
 241                 self.send_header("Content-Type", "video/ogg")
 242             else:
 243                 self.send_header("Content-Type", "video/mpeg")
 244             self.end_headers()
 245             # Do not send real headers at all
 246             self.headerssent = True
 247
 248         try:
 249             self.hanggreenlet = gevent.spawn(self.hangDetector)
 250             logger.debug("hangDetector spawned")
 251             gevent.sleep()
 252
 253             # Getting URL
 254             self.errorhappened = False
 255
 256             if shouldcreatevp:
 257                 logger.debug("Got url " + self.path_unquoted)
 258                 # Force ffmpeg demuxing if set in config
 259                 if VPConfig.vlcforceffmpeg:
 260                     self.vlcprefix = 'http/ffmpeg://'
 261                 else:
 262                     self.vlcprefix = ''
 263
 264                 logger.info("Starting broadcasting "+self.path)
 265                 VPStuff.vlcclient.startBroadcast(
 266                     self.vlcid, self.vlcprefix + self.path_unquoted, VPConfig.vlcmux, VPConfig.vlcpreaccess, self.reqtype)
 267                 # Sleep a bit, because sometimes VLC doesn't open port in
 268                 # time
 269                 gevent.sleep(0.5)
 270
 271             # Building new VLC url
 272             self.url = 'http://' + VPConfig.vlchost + \
 273                 ':' + str(VPConfig.vlcoutport) + '/' + self.vlcid
 274             logger.debug("VLC url " + self.url)
 275
 276             # Sending client headers to videostream
 277             self.video = urllib2.Request(self.url)
 278             for key in self.headers.dict:
 279                 self.video.add_header(key, self.headers.dict[key])
 280
 281             self.video = urllib2.urlopen(self.video)
 282
 283             # Sending videostream headers to client
 284             if not self.headerssent:
 285                 self.send_response(self.video.getcode())
 286                 if self.video.info().dict.has_key('connection'):
 287                     del self.video.info().dict['connection']
 288                 if self.video.info().dict.has_key('server'):
 289                     del self.video.info().dict['server']
 290                 if self.video.info().dict.has_key('transfer-encoding'):
 291                     del self.video.info().dict['transfer-encoding']
 292                 if self.video.info().dict.has_key('content-type'):
 293                     del self.video.info().dict['content-type']
 294                 if self.video.info().dict.has_key('keep-alive'):
 295                     del self.video.info().dict['keep-alive']
 296
 297                 for key in self.video.info().dict:
 298                     self.send_header(key, self.video.info().dict[key])
 299
 300                 self.send_header('Cache-Control','no-cache');
 301
 302                 if self.reqtype=="ogg":
 303                     self.send_header("Content-Type", "video/ogg")
 304                 else:
 305                     self.send_header("Content-Type", "video/mpeg")
 306
 307                 # End headers. Next goes video data
 308                 self.end_headers()
 309                 logger.debug("Headers sent")
 310
 311             # Run proxyReadWrite
 312             self.proxyReadWrite()
 313
 314             # Waiting until hangDetector is joined
 315             self.hanggreenlet.join()
 316             logger.debug("Request handler finished")
 317
 318         except (vpclient.VPException, vlcclient.VlcException, urllib2.URLError) as e:
 319             logger.error("Exception: " + repr(e))
 320             self.errorhappened = True
 321             self.dieWithError()
 322         except gevent.GreenletExit:
 323             # hangDetector told us about client disconnection
 324             pass
 325         except Exception:
 326             # Unknown exception
 327             logger.error(traceback.format_exc())
 328             self.errorhappened = True
 329             self.dieWithError()
 330         finally:
 331             logger.debug("END REQUEST")
 332             logger.info("Closed connection from " + self.clientip + " path " + self.path)
 333             VPStuff.clientcounter.delete(self.reqtype+'\\'+self.path_unquoted, self.clientip)
 334             if not VPStuff.clientcounter.get(self.reqtype+'\\'+self.path_unquoted):
 335                 try:
 336                     logger.debug("That was the last client, destroying VPClient")
 337                     logger.info("Stopping broadcasting " + self.path)
 338                     VPStuff.vlcclient.stopBroadcast(self.vlcid)
 339                 except:
 340                     pass
 341                 self.vp.destroy()
 342             if not self.headersent:
 343                 logger.error("Problem receiving video stream, no headers!")
 344                 if VPStuff.clientcounter.total == 0:
 345                     logger.error("Probably VLC hang")
 346                     VPStuff.vlc.kill()
 347
 348 class HTTPServer(SocketServer.ThreadingMixIn, BaseHTTPServer.HTTPServer):
 349
 350     def handle_error(self, request, client_address):
 351         # Do not print HTTP tracebacks
 352         pass
 353
 354
 355 class VPStuff(object):
 356     '''
 357     Inter-class interaction class
 358     '''
 359     vlcclient=None
 360
 361 # taken from http://stackoverflow.com/questions/2699907/dropping-root-permissions-in-python
 362 def drop_privileges(uid_name, gid_name='nogroup'):
 363
 364     # Get the uid/gid from the name
 365     running_uid = pwd.getpwnam(uid_name).pw_uid
 366     running_uid_home = pwd.getpwnam(uid_name).pw_dir
 367     running_gid = grp.getgrnam(gid_name).gr_gid
 368
 369     # Remove group privileges
 370     os.setgroups([])
 371
 372     # Try setting the new uid/gid
 373     os.setgid(running_gid)
 374     os.setuid(running_uid)
 375
 376     # Ensure a very conservative umask
 377     old_umask = os.umask(077)
 378
 379     if os.getuid() == running_uid and os.getgid() == running_gid:
 380         # could be useful
 381         os.environ['HOME'] = running_uid_home
 382         return True
 383     return False
 384
 385 logging.basicConfig(
 386     filename=VPConfig.logpath + 'vphttp.log' if VPConfig.loggingtoafile else None,
 387     format='%(asctime)s %(levelname)s %(name)s: %(message)s', datefmt='%d.%m.%Y %H:%M:%S', level=VPConfig.debug)
 388 logger = logging.getLogger('INIT')
 389
 390 # Loading plugins
 391 # Trying to change dir (would fail in freezed state)
 392 try:
 393     os.chdir(os.path.dirname(os.path.realpath(__file__)))
 394 except:
 395     pass
 396 # Creating dict of handlers
 397 VPStuff.pluginshandlers = dict()
 398 # And a list with plugin instances
 399 VPStuff.pluginlist = list()
 400 pluginsmatch = glob.glob('plugins/*_plugin.py')
 401 sys.path.insert(0, 'plugins')
 402 pluginslist = [os.path.splitext(os.path.basename(x))[0] for x in pluginsmatch]
 403 for i in pluginslist:
 404     plugin = __import__(i)
 405     plugname = i.split('_')[0].capitalize()
 406     try:
 407         plugininstance = getattr(plugin, plugname)(VPConfig, VPStuff)
 408     except Exception as e:
 409         logger.error("Cannot load plugin " + plugname + ": " + repr(e))
 410         continue
 411     logger.debug('Plugin loaded: ' + plugname)
 412     for j in plugininstance.handlers:
 413         logger.info("Registering handler '" + j +"'")
 414         VPStuff.pluginshandlers[j] = plugininstance
 415     VPStuff.pluginlist.append(plugininstance)
 416
 417 # Check whether we can bind to the defined port safely
 418 if os.getuid() != 0 and VPConfig.httpport <= 1024:
 419     logger.error("Cannot bind to port " + str(VPConfig.httpport) + " without root privileges")
 420     sys.exit(1)
 421
 422 server = HTTPServer((VPConfig.httphost, VPConfig.httpport), HTTPHandler)
 423 logger = logging.getLogger('HTTP')
 424
 425 # Dropping root privileges if needed
 426 if VPConfig.vpproxyuser and os.getuid() == 0:
 427     if drop_privileges(VPConfig.vpproxyuser):
 428         logger.info("Dropped privileges to user " + VPConfig.vpproxyuser)
 429     else:
 430         logger.error("Cannot drop privileges to user " + VPConfig.vpproxyuser)
 431         sys.exit(1)
 432
 433 # Creating ClientCounter
 434 VPStuff.clientcounter = ClientCounter()
 435
 436 DEVNULL = open(os.devnull, 'wb')
 437
 438 # Spawning procedures
 439 def spawnVLC(cmd, delay = 0):
 440     try:
 441         VPStuff.vlc = psutil.Popen(cmd) #, stdout=DEVNULL, stderr=DEVNULL)
 442         gevent.sleep(delay)
 443         return True
 444     except:
 445         return False
 446
 447 def connectVLC():
 448     try:
 449         VPStuff.vlcclient = vlcclient.VlcClient(
 450             host=VPConfig.vlchost, port=VPConfig.vlcport, password=VPConfig.vlcpass,
 451             out_port=VPConfig.vlcoutport)
 452         return True
 453     except vlcclient.VlcException as e:
 454         return False
 455
 456 def isRunning(process):
 457     if psutil.version_info[0] >= 2:
 458         if process.is_running() and process.status() != psutil.STATUS_ZOMBIE:
 459             return True
 460     else:  # for older versions of psutil
 461         if process.is_running() and process.status != psutil.STATUS_ZOMBIE:
 462             return True
 463     return False
 464
 465 def findProcess(name):
 466     for proc in psutil.process_iter():
 467         try:
 468             pinfo = proc.as_dict(attrs=['pid', 'name'])
 469             if pinfo['name'] == name:
 470                 return pinfo['pid']
 471         except psutil.AccessDenied:
 472             # System process
 473             pass
 474         except psutil.NoSuchProcess:
 475             # Process terminated
 476             pass
 477     return None
 478
 479 def clean_proc():
 480     # Trying to close all spawned processes gracefully
 481     if isRunning(VPStuff.vlc):
 482         if VPStuff.vlcclient:
 483             VPStuff.vlcclient.destroy()
 484         gevent.sleep(1)
 485     if isRunning(VPStuff.vlc):
 486         # or not :)
 487         VPStuff.vlc.kill()
 488
 489 # This is what we call to stop the server completely
 490 def shutdown(signum = 0, frame = 0):
 491     logger.info("Stopping server...")
 492     # Closing all client connections
 493     for connection in server.RequestHandlerClass.requestlist:
 494         try:
 495             # Set errorhappened to prevent waiting for videodestroydelay
 496             connection.errorhappened = True
 497             connection.closeConnection()
 498         except:
 499             logger.warning("Cannot kill a connection!")
 500     clean_proc()
 501     server.server_close()
 502     sys.exit()
 503
 504 def _reloadconfig(signum=None, frame=None):
 505     '''
 506     Reload configuration file.
 507     SIGHUP handler.
 508     '''
 509     global VPConfig
 510
 511     logger = logging.getLogger('reloadconfig')
 512     reload(vpconfig)
 513     from vpconfig import VPConfig
 514     logger.info('Config reloaded')
 515
 516 # setting signal handlers
 517 try:
 518     gevent.signal(signal.SIGHUP, _reloadconfig)
 519     gevent.signal(signal.SIGTERM, shutdown)
 520 except AttributeError:
 521     pass
 522
 523 name = 'vlc'
 524 VPStuff.vlcProc = VPConfig.vlccmd.split()
 525 if spawnVLC(VPStuff.vlcProc, VPConfig.vlcspawntimeout) and connectVLC():
 526     logger.info("VLC spawned with pid " + str(VPStuff.vlc.pid))
 527 else:
 528     logger.error('Cannot spawn or connect to VLC!')
 529     clean_proc()
 530     sys.exit(1)
 531
 532 try:
 533     logger.info("Using gevent %s" % gevent.__version__)
 534     logger.info("Using psutil %s" % psutil.__version__)
 535     logger.info("Using VLC %s" % VPStuff.vlcclient._vlcver)
 536     logger.info("Server started.")
 537     while True:
 538         if not isRunning(VPStuff.vlc):
 539             del VPStuff.vlc
 540             if spawnVLC(VPStuff.vlcProc, VPConfig.vlcspawntimeout) and connectVLC():
 541                 logger.info("VLC died, respawned it with pid " + str(VPStuff.vlc.pid))
 542             else:
 543                 logger.error("Cannot spawn VLC!")
 544                 clean_proc()
 545                 sys.exit(1)
 546         # Return to our server tasks
 547         server.handle_request()
 548 except (KeyboardInterrupt, SystemExit):
 549     shutdown()