0c09c5297bb8e040d32460a1201c517f473544ac
[vpproxy.git] / 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, no-store, must-revalidate');
240             self.send_header('Pragma','no-cache');
241             if self.reqtype in ("ogg","ogv"):
242                 self.send_header("Content-Type", "video/ogg")
243             else:
244                 self.send_header("Content-Type", "video/mpeg")
245             self.end_headers()
246             # Do not send real headers at all
247             self.headerssent = True
248
249         try:
250             self.hanggreenlet = gevent.spawn(self.hangDetector)
251             logger.debug("hangDetector spawned")
252             gevent.sleep()
253
254             # Getting URL
255             self.errorhappened = False
256
257             if shouldcreatevp:
258                 logger.debug("Got url " + self.path_unquoted)
259                 # Force ffmpeg demuxing if set in config
260                 if VPConfig.vlcforceffmpeg:
261                     self.vlcprefix = 'http/ffmpeg://'
262                 else:
263                     self.vlcprefix = ''
264
265                 logger.info("Starting broadcasting "+self.path)                    
266                 VPStuff.vlcclient.startBroadcast(
267                     self.vlcid, self.vlcprefix + self.path_unquoted, VPConfig.vlcmux, VPConfig.vlcpreaccess, self.reqtype)
268                 # Sleep a bit, because sometimes VLC doesn't open port in
269                 # time
270                 gevent.sleep(0.5)
271
272             # Building new VLC url
273             self.url = 'http://' + VPConfig.vlchost + \
274                 ':' + str(VPConfig.vlcoutport) + '/' + self.vlcid
275             logger.debug("VLC url " + self.url)
276
277             # Sending client headers to videostream
278             self.video = urllib2.Request(self.url)
279             for key in self.headers.dict:
280                 self.video.add_header(key, self.headers.dict[key])
281
282             self.video = urllib2.urlopen(self.video)
283
284             # Sending videostream headers to client
285             if not self.headerssent:
286                 self.send_response(self.video.getcode())
287                 if self.video.info().dict.has_key('connection'):
288                     del self.video.info().dict['connection']
289                 if self.video.info().dict.has_key('server'):
290                     del self.video.info().dict['server']
291                 if self.video.info().dict.has_key('transfer-encoding'):
292                     del self.video.info().dict['transfer-encoding']
293                 if self.video.info().dict.has_key('content-type'):
294                     del self.video.info().dict['content-type']
295                 if self.video.info().dict.has_key('keep-alive'):
296                     del self.video.info().dict['keep-alive']
297
298                 for key in self.video.info().dict:
299                     self.send_header(key, self.video.info().dict[key])
300
301                 self.send_header('Cache-Control','no-cache, no-store, must-revalidate');
302                 self.send_header('Pragma','no-cache');
303
304                 if self.reqtype=="ogg":
305                     self.send_header("Content-Type", "video/ogg")
306                 else:
307                     self.send_header("Content-Type", "video/mpeg")
308
309                 # End headers. Next goes video data
310                 self.end_headers()
311                 logger.debug("Headers sent")
312
313             # Run proxyReadWrite
314             self.proxyReadWrite()
315
316             # Waiting until hangDetector is joined
317             self.hanggreenlet.join()
318             logger.debug("Request handler finished")
319
320         except (vpclient.VPException, vlcclient.VlcException, urllib2.URLError) as e:
321             logger.error("Exception: " + repr(e))
322             self.errorhappened = True
323             self.dieWithError()
324         except gevent.GreenletExit:
325             # hangDetector told us about client disconnection
326             pass
327         except Exception:
328             # Unknown exception
329             logger.error(traceback.format_exc())
330             self.errorhappened = True
331             self.dieWithError()
332         finally:
333             logger.debug("END REQUEST")
334             logger.info("Closed connection from " + self.clientip + " path " + self.path)
335             VPStuff.clientcounter.delete(self.reqtype+'\\'+self.path_unquoted, self.clientip)
336             if not VPStuff.clientcounter.get(self.reqtype+'\\'+self.path_unquoted):
337                 try:
338                     logger.debug("That was the last client, destroying VPClient")
339                     logger.info("Stopping broadcasting " + self.path)
340                     VPStuff.vlcclient.stopBroadcast(self.vlcid)
341                 except:
342                     pass
343                 self.vp.destroy()
344             if not self.headerssent:
345                 logger.error("Problem receiving video stream, no headers!")
346                 if VPStuff.clientcounter.total == 0:
347                     logger.error("Probably VLC hang")
348                     VPStuff.vlc.kill()
349
350 class HTTPServer(SocketServer.ThreadingMixIn, BaseHTTPServer.HTTPServer):
351
352     def handle_error(self, request, client_address):
353         # Do not print HTTP tracebacks
354         pass
355
356
357 class VPStuff(object):
358     '''
359     Inter-class interaction class
360     '''
361     vlcclient=None
362
363 # taken from http://stackoverflow.com/questions/2699907/dropping-root-permissions-in-python
364 def drop_privileges(uid_name, gid_name='nogroup'):
365
366     # Get the uid/gid from the name
367     running_uid = pwd.getpwnam(uid_name).pw_uid
368     running_uid_home = pwd.getpwnam(uid_name).pw_dir
369     running_gid = grp.getgrnam(gid_name).gr_gid
370
371     # Remove group privileges
372     os.setgroups([])
373
374     # Try setting the new uid/gid
375     os.setgid(running_gid)
376     os.setuid(running_uid)
377
378     # Ensure a very conservative umask
379     old_umask = os.umask(077)
380
381     if os.getuid() == running_uid and os.getgid() == running_gid:
382         # could be useful
383         os.environ['HOME'] = running_uid_home
384         return True
385     return False
386
387 logging.basicConfig(
388     filename=VPConfig.logpath + 'vphttp.log' if VPConfig.loggingtoafile else None,
389     format='%(asctime)s %(levelname)s %(name)s: %(message)s', datefmt='%d.%m.%Y %H:%M:%S', level=VPConfig.debug)
390 logger = logging.getLogger('INIT')
391
392 # Loading plugins
393 # Trying to change dir (would fail in freezed state)
394 try:
395     os.chdir(os.path.dirname(os.path.realpath(__file__)))
396 except:
397     pass
398 # Creating dict of handlers
399 VPStuff.pluginshandlers = dict()
400 # And a list with plugin instances
401 VPStuff.pluginlist = list()
402 pluginsmatch = glob.glob('plugins/*_plugin.py')
403 sys.path.insert(0, 'plugins')
404 pluginslist = [os.path.splitext(os.path.basename(x))[0] for x in pluginsmatch]
405 for i in pluginslist:
406     plugin = __import__(i)
407     plugname = i.split('_')[0].capitalize()
408     try:
409         plugininstance = getattr(plugin, plugname)(VPConfig, VPStuff)
410     except Exception as e:
411         logger.error("Cannot load plugin " + plugname + ": " + repr(e))
412         continue
413     logger.debug('Plugin loaded: ' + plugname)
414     for j in plugininstance.handlers:
415         logger.info("Registering handler '" + j +"'")
416         VPStuff.pluginshandlers[j] = plugininstance
417     VPStuff.pluginlist.append(plugininstance)
418
419 # Check whether we can bind to the defined port safely
420 if os.getuid() != 0 and VPConfig.httpport <= 1024:
421     logger.error("Cannot bind to port " + str(VPConfig.httpport) + " without root privileges")
422     sys.exit(1)
423
424 server = HTTPServer((VPConfig.httphost, VPConfig.httpport), HTTPHandler)
425 logger = logging.getLogger('HTTP')
426
427 # Dropping root privileges if needed
428 if VPConfig.vpproxyuser and os.getuid() == 0:
429     if drop_privileges(VPConfig.vpproxyuser):
430         logger.info("Dropped privileges to user " + VPConfig.vpproxyuser)
431     else:
432         logger.error("Cannot drop privileges to user " + VPConfig.vpproxyuser)
433         sys.exit(1)
434
435 # Creating ClientCounter
436 VPStuff.clientcounter = ClientCounter()
437
438 DEVNULL = open(os.devnull, 'wb')
439
440 # Spawning procedures
441 def spawnVLC(cmd, delay = 0):
442     try:
443         VPStuff.vlc = psutil.Popen(cmd) #, stdout=DEVNULL, stderr=DEVNULL)
444         gevent.sleep(delay)
445         return True
446     except:
447         return False
448
449 def connectVLC():
450     try:
451         VPStuff.vlcclient = vlcclient.VlcClient(
452             host=VPConfig.vlchost, port=VPConfig.vlcport, password=VPConfig.vlcpass,
453             out_port=VPConfig.vlcoutport)
454         return True
455     except vlcclient.VlcException as e:
456         return False
457
458 def isRunning(process):
459     if psutil.version_info[0] >= 2:
460         if process.is_running() and process.status() != psutil.STATUS_ZOMBIE:
461             return True
462     else:  # for older versions of psutil
463         if process.is_running() and process.status != psutil.STATUS_ZOMBIE:
464             return True
465     return False
466
467 def findProcess(name):
468     for proc in psutil.process_iter():
469         try:
470             pinfo = proc.as_dict(attrs=['pid', 'name'])
471             if pinfo['name'] == name:
472                 return pinfo['pid']
473         except psutil.AccessDenied:
474             # System process
475             pass
476         except psutil.NoSuchProcess:
477             # Process terminated
478             pass
479     return None
480
481 def clean_proc():
482     # Trying to close all spawned processes gracefully
483     if isRunning(VPStuff.vlc):
484         if VPStuff.vlcclient:
485             VPStuff.vlcclient.destroy()
486         gevent.sleep(1)
487     if isRunning(VPStuff.vlc):
488         # or not :)
489         VPStuff.vlc.kill()
490
491 # This is what we call to stop the server completely
492 def shutdown(signum = 0, frame = 0):
493     logger.info("Stopping server...")
494     # Closing all client connections
495     for connection in server.RequestHandlerClass.requestlist:
496         try:
497             # Set errorhappened to prevent waiting for videodestroydelay
498             connection.errorhappened = True
499             connection.closeConnection()
500         except:
501             logger.warning("Cannot kill a connection!")
502     clean_proc()
503     server.server_close()
504     sys.exit()
505
506 def _reloadconfig(signum=None, frame=None):
507     '''
508     Reload configuration file.
509     SIGHUP handler.
510     '''
511     global VPConfig
512
513     logger = logging.getLogger('reloadconfig')
514     reload(vpconfig)
515     from vpconfig import VPConfig
516     logger.info('Config reloaded')
517
518 # setting signal handlers
519 try:
520     gevent.signal(signal.SIGHUP, _reloadconfig)
521     gevent.signal(signal.SIGTERM, shutdown)
522 except AttributeError:
523     pass
524
525 name = 'vlc'
526 VPStuff.vlcProc = VPConfig.vlccmd.split()
527 if spawnVLC(VPStuff.vlcProc, VPConfig.vlcspawntimeout) and connectVLC():
528     logger.info("VLC spawned with pid " + str(VPStuff.vlc.pid))
529 else:
530     logger.error('Cannot spawn or connect to VLC!')
531     clean_proc()
532     sys.exit(1)
533
534 try:
535     logger.info("Using gevent %s" % gevent.__version__)
536     logger.info("Using psutil %s" % psutil.__version__)
537     logger.info("Using VLC %s" % VPStuff.vlcclient._vlcver)
538     logger.info("Server started.")
539     while True:
540         if not isRunning(VPStuff.vlc):
541             del VPStuff.vlc
542             if spawnVLC(VPStuff.vlcProc, VPConfig.vlcspawntimeout) and connectVLC():
543                 logger.info("VLC died, respawned it with pid " + str(VPStuff.vlc.pid))
544             else:
545                 logger.error("Cannot spawn VLC!")
546                 clean_proc()
547                 sys.exit(1)
548         # Return to our server tasks
549         server.handle_request()
550 except (KeyboardInterrupt, SystemExit):
551     shutdown()