
'''
 Implementation of a mediaproxy request dispatcher for SER.

 The dispatcher implements an architecture where SER will work with multiple
mediaproxy servers (local or remote). The dispatcher is implemented around the
concept of choosing which mediaproxy server to use based on the domain of the
sender or destination. If no mediaproxy server is defined for the given domain
it will fallback to a default mediaproxy server (if configured to use one).

 Mediaproxy servers for a domain are defined using DNS SRV records.
Multiple mediaproxy servers can be defined for a domain by using the priority
and weight attributes of the SRV records. This permits both load balancing the
traffic between multiple mediaproxy servers (if they are defined at the same
priority but with different weights) or mediaproxy server fallbacks (if none
of the servers defined at a priority responds, it will fallback to servers
defined at a lower priority).


Copyright 2004-2005 Dan Pascu
'''

import os
import sys
import grp
import time
import random
import atexit
import socket
import errno
import cPickle
import traceback
import threading
from SocketServer import ThreadingUnixStreamServer, ThreadingTCPServer, BaseRequestHandler

from utilities import *
from datatypes import *
from request import Request, InvalidRequestError
from accounting import accounting, StopRecordSerializer, UnserializeError
from configuration import readSettings, ConfigSection


##
## dispatcher configuration
##
class ProxydispatcherAddress(MandatoryNetworkAddress):
    _defaultPort = 25061
    _name = 'proxydispatcher address'

class DispatcherConfig(ConfigSection):
    _dataTypes = {'defaultProxy': MediaProxyAddress,
                  'listen':       ProxydispatcherAddress}
    socket        = '/var/run/proxydispatcher.sock'
    listen        = ProxydispatcherAddress('default')
    group         = 'openser'
    defaultProxy  = '/var/run/mediaproxy.sock'
    timeout       = 24*60*60 ## remove sessions older than 24 hours
    checkInterval = 600      ## check this often for expired sessions

## We use this to overwrite some of the settings above on a local basis if needed
readSettings('Dispatcher', DispatcherConfig)
# todo: embed logic of encoding/decoding address to/from string into StructureInfo -Dan
DispatcherConfig._encoded_listen = address2str(DispatcherConfig.listen)

try:
    from accounting import accounting
except ImportError:
    warning("accounting is enabled but the accounting module is missing. accounting is not available!")
    accounting = Null()


## Internal state of dispatcher (DON'T TOUCH!!!)

Sessions = {}

## Sessions above should only be accessed after aquiring this lock
sessionLock = threading.RLock()

maxDataSize = 8192

## Fix the environment

## Find where the dispatcher is located
##
## Add path to foreign python modules that we include with mediaproxy for
## convenience as they are not standard. Some are also needed for python < 2.3

ModulesPath   = os.path.realpath((os.path.dirname(__file__) or '.'))
sys.path[1:1] = [ModulesPath + '/external']

import DNS
DNS.DiscoverNameServers()

## check if we have python 2.3 or newer. Else use some 3rd party module for timeout
## to work with sockets
try:
    socket.setdefaulttimeout
except:
    import timeoutsocket
    socket.setdefaulttimeout = timeoutsocket.setDefaultSocketTimeout
    socket.getdefaulttimeout = timeoutsocket.getDefaultSocketTimeout
    socket.timeout = timeoutsocket.Timeout
    del(timeoutsocket)


## Use this function to get the dispatcher rolling
def start():
    ## restore previous sessions if any
    restoreSessions()
    ## a timer to expire stale sessions (runs in a separate thread)
    timer = SessionTimer(DispatcherConfig.checkInterval)
    timer.start()
    ## the mediaproxy media timeout handler
    mp = MediaproxyCommandHandler()
    ts = ThreadedServer(mp, 'MediaTimeoutHandler')
    ts.start()
    ## the main request handler that does the request dispatching
    server = SERCommandHandler()
    server.serve_forever()

def saveSessions():
    sessionLock.acquire()
    try:
        if Sessions:
            sessionsFile = '%s/%s' % (ModulesPath, '.dispatcher-sessions.dat')
            try:
                f = open(sessionsFile, 'w')
            except:
                pass
            else:
                failedDump = False
                try:
                    try:
                        cPickle.dump(Sessions, f)
                    except:
                        failedDump = True
                finally:
                    f.close()
                if failedDump:
                    try:    os.unlink(sessionsFile)
                    except: pass
    finally:
        sessionLock.release()

def restoreSessions():
    sessionLock.acquire()
    try:
        sessionsFile = '%s/%s' % (ModulesPath, '.dispatcher-sessions.dat')
        try:
            f = open(sessionsFile, 'r')
        except:
            pass
        else:
            global Sessions
            try:
                Sessions = cPickle.load(f)
            except:
                pass
            f.close()
            try:    os.unlink(sessionsFile)
            except: pass
    finally:
        sessionLock.release()

## Helper functions

## Classes

class DispatcherError(Exception): pass


class SessionTimer:
    def __init__(self, period):
        self.__period = period
        self.__started = False
        self.__lock = threading.RLock()
        self.__thread = None
    def start(self):
        self.__lock.acquire()
        try:
            assert not self.__started, "timer already started"
            self.__started = True
            self.__run()
        finally:
            self.__lock.release()
    def cancel(self):
        self.__lock.acquire()
        try:
            if self.__thread:
                self.__thread.cancel()
                self.__thread = None
            self.__started = False
        finally:
            self.__lock.release()
    def __run(self):
        ## Check sessions and timeout staled ones
        count = 0
        now = time.time()
        sessionLock.acquire()
        try:
            for session in Sessions.values():
                if now >= session.expire:
                    try:
                        del Sessions[session._id]
                    except KeyError:
                        pass
                    else:
                        count += 1
        finally:
            sessionLock.release()
        if count > 0:
            info("expired %d staled session(s)" % count)
        self.__lock.acquire()
        try:
            if self.__started:
                timer = threading.Timer(self.__period, self.__run)
                timer.setName('SessionTimer')
                timer.setDaemon(True)
                timer.start()
                self.__thread = timer
            else:
                self.__thread = None
        finally:
            self.__lock.release()


class MediaproxyCommandHandler(ThreadingTCPServer):
    def __init__(self):
        try:
            ThreadingTCPServer.__init__(self, DispatcherConfig.listen, MediaproxyRequestHandler)
        except socket.error, why:
            raise RuntimeError, "couldn't create server (%s)" % why[1]
    def server_bind(self):
        self.allow_reuse_address = True
        ThreadingTCPServer.server_bind(self)

class MediaproxyRequestHandler(BaseRequestHandler):
    '''Handles a command received through the mediaproxy control socket'''
    def handle(self):
        line = self.request.recv(maxDataSize)
        print line.rstrip()
        try:
            request = Request(line)
        except InvalidRequestError, msg:
            error(msg)
        else:
            try:
                result = self.process(request)
                if result is not None:
                    self.request.send(result + '\n')
            except InvalidRequestError, msg:
                error(msg)
        self.request.close() # this is done by close_request() anyway

    def process(self, request):
        cmd = request.cmd

        if cmd == 'timeout':
            sessionId = request.sessionId
            sessionLock.acquire()
            try:
                session = Sessions.get(sessionId, None)
                if session is None:
                    return 'Not found'
                ## check if requesting peer owns the session
                peer = self.client_address[0]
                if session.remote:
                    mediaproxy = session.address[0]
                else:
                    mediaproxy = session.address
                if peer != mediaproxy:
                    error("%s tried to timeout a call it doesn't own (owner is %s)" % (peer, mediaproxy))
                    return 'Error'
                ## deserialize the stop record
                recordSerializer = StopRecordSerializer()
                try:
                    record = recordSerializer.load(request.stopRecord)
                except UnserializeError, why:
                    error("couldn't unserialize stop record: %s" % why)
                    return 'Error'
                del(Sessions[sessionId])
            finally:
                sessionLock.release()
            accounting.log(record)
            # notify callcontrol
            return 'Ok'
        else:
            raise InvalidRequestError, "unknown command: `%s'" % cmd

class ThreadedServer(threading.Thread):
    def __init__(self, server, name='Server'):
        threading.Thread.__init__(self, name=name)
        self.setDaemon(True)
        self.server = server
    def run(self):
        self.server.serve_forever()


class SERCommandHandler(ThreadingUnixStreamServer):
    def __init__(self, path=None, group=None):
        ## If we would have passed these as default values they would have been
        ## locked to the values they had when the dispatcher module is loaded
        ## That way we can change the default values after dispatcher is loaded
        ## and we get the updated values
        path = path or DispatcherConfig.socket
        group = group or DispatcherConfig.group
        self.path = path
        self.group = group
        try:
            os.unlink(path)
        except OSError:
            pass
        try:
            ThreadingUnixStreamServer.__init__(self, path, SERRequestHandler)
        except socket.error, why:
            raise DispatcherError, "couldn't create server (%s)" % why[1]
        atexit.register(self.handle_close)

    def server_bind(self):
        ThreadingUnixStreamServer.server_bind(self)
        path = self.path
        try:
            groupId = grp.getgrnam(self.group)[2]
        except KeyError, IndexError:
            ## Make it world writable
            try:
                os.chmod(path, 0777)
            except OSError:
                warning("couldn't set access rights for %s." % path)
                warning("SER may not be able to communicate with us!")
        else:
            ## Make it writable only to the SIP proxy group members
            try:
                os.chown(path, -1, groupId)
                os.chmod(path, 0770)
                #os.chmod(path, S_IRUSR|S_IWUSR|S_IXUSR|S_IRGRP|S_IWGRP|S_IXGRP)
            except OSError:
                warning("couldn't set access rights for %s." % path)
                warning("SER may not be able to communicate with us!")

    def handle_close(self):
        self.socket.close()
        try:
            os.unlink(self.path)
        except OSError:
            pass


class SERRequestHandler(BaseRequestHandler):
    '''Handles a command received through the control socket'''

    def handle(self):
        line = self.request.recv(maxDataSize)
        print line.rstrip()
        try:
            request = Request(line)
        except InvalidRequestError, msg:
            error(msg)
        else:
            try:
                s = time.time()
                result = self.process(request)
                print "execution time: %5.2f ms" % ((time.time()-s)*1000)
                if result is not None:
                    self.request.send(result)
            except InvalidRequestError, msg:
                error(msg)
        self.request.close() # this is done by close_request() anyway

    def process(self, request):
        cmd = request.cmd

        if cmd == 'request':
            sessionId = request.sessionId
            sessionLock.acquire()
            try:
                session = Sessions.get(sessionId, None)
            finally:
                sessionLock.release()
            if session is not None:
                if session.remote:
                    request.info.dispatcher = DispatcherConfig._encoded_listen
                return session.forward(request)
            domain = (request.localCaller and request.callerDomain) or (request.localCalled and request.calledDomain) or None
            if domain is not None:
                proxy = MediaProxy('domain://%s' % domain).address
                if proxy is None:
                    proxy = MediaProxy(DispatcherConfig.defaultProxy).address
            else:
                proxy = MediaProxy(DispatcherConfig.defaultProxy).address
            if proxy is None:
                error("No mediaproxy servers found for the default domain.")
                return None
            sessionLock.acquire()
            try:
                try:
                    session = Sessions[sessionId]
                except KeyError:
                    if request.info.totag:
                        ## Don't create a new session on re-INVITEs
                        return None
                    if not proxy:
                        warning("no proxy available for call: %(callerDomain)s -> %(calledDomain)s" % request)
                        return None
                    session = MediaSession(sessionId, proxy)
                    Sessions[sessionId] = session
            finally:
                sessionLock.release()
            if session.remote:
                request.info.dispatcher = DispatcherConfig._encoded_listen
            return session.forward(request)
        elif cmd == 'lookup':
            sessionLock.acquire()
            try:
                session = Sessions.get(request.sessionId, None)
                if session is None:
                    #warning("trying to lookup session with non-existent id: `%(sessionId)s'" % request)
                    return None
            finally:
                sessionLock.release()
            #if session.remote:
                #request.info.dispatcher = DispatcherConfig._encoded_listen
            return session.forward(request)
        elif cmd == 'delete':
            sessionLock.acquire()
            try:
                session = Sessions.get(request.sessionId, None)
                if session is None:
                    return None
            finally:
                sessionLock.release()
            #if session.remote:
                #request.info.dispatcher = DispatcherConfig._encoded_listen
            result = session.forward(request)
            if result is not None:
                result = result.rstrip()
            if result:
                try:
                    session.stopRecord = StopRecordSerializer().load(result)
                except (TypeError, UnserializeError), why:
                    error("received unparsable stop record from mediaproxy: %s" % why)
                    warning("accounting records will not be updated")
            session.end()
            return ''
        else:
            raise InvalidRequestError, "unknown command: `%s'" % cmd


class MediaSession(object):
    def __init__(self, sessionId, address):
        self._id = sessionId
        self.address = address
        self.sessionStart = time.time()
        self.expire = self.sessionStart + DispatcherConfig.timeout
        self.stopRecord = None
    def getrm(self):
        return type(self.address) is tuple
    def getst(self):
        if type(self.address) is tuple:
            return socket.AF_INET
        else:
            return socket.AF_UNIX
    def getas(self):
        if type(self.address) is tuple:
            return "%s:%s" % self.address
        else:
            return self.address
    remote = property(getrm)
    socktype = property(getst)
    address_str = property(getas)
    del getrm, getst, getas
    
    def forward(self, request):
        if self.address is None:
            return None
        s = socket.socket(self.socktype, socket.SOCK_STREAM)
        try:
            s.settimeout(5)
        except AttributeError:
            pass
        result = None
        try:
            try:
                s.connect(self.address)
                s.send(str(request))
                result = s.recv(maxDataSize)
            except (socket.error, socket.gaierror, socket.herror, socket.timeout), why:
                try:               motive = why[1]
                except IndexError: motive = why
                error("couldn't forward command to %s (%s)" % (self.address_str, motive))
                return None
        finally:
            s.close()
        print "forwarding to mediaproxy on %s: got: '%s'" % (self.address_str, result.rstrip())
        return result

    def end(self):
        sessionLock.acquire()
        try:
            try:
                del(Sessions[self._id])
            except KeyError:
                pass
            else:
                if self.stopRecord:
                    accounting.log(self.stopRecord)
        finally:
            sessionLock.release()


class MediaProxy(object):
    '''
    Class that defines a single MediaProxy server or a collection of
    MediaProxy servers as defined by a domain (using SRV records)
    '''
    def __init__(self, url):
        try:
            url + ''
        except TypeError:
            if url is None or isinstance(url, tuple):
                self.address = url
                return
            else:
                raise TypeError, "invalid MediaProxy address type"
        domainId = 'domain://'
        if url.startswith(domainId):
            domain = url[len(domainId):]
            mediaproxyrec = "_mediaproxy._tcp.%s" % domain
            try:
                dns = DNS.DnsRequest()
                answers = dns.req(name=mediaproxyrec, qtype='SRV', timeout=7).answers
            except (DNS.Base.DNSError, socket.error), why:
                error("couldn't retrieve DNS SRV record '%s': %s" % (mediaproxyrec, why))
                self.address = None
                return
            if not answers:
                self.address = None
                return
            servers = {}
            for x in answers:
                try:
                    (priority, weight, port, address) = x['data']
                except:
                    continue
                array = servers.setdefault(priority, [])
                array.append((weight, address, port))
            priorities = servers.keys()
            priorities.sort()
            allproxies = []
            for p in priorities:
                all = servers[p] ## all servers at the same priority
                if len(all) == 1:
                    allproxies.append((all[0][1:]))
                else:
                    weights = [x[0] or 1 for x in all]
                    try:
                        totalWeight = sum(weights)
                    except NameError:
                        ## This is for python older than 2.3
                        totalWeight = reduce(lambda x,y: x+y, weights)
                    weights = [int(w*100/totalWeight) for w in weights]
                    array = reduce(lambda x,y: x+y, [weights[i]*[i] for i in
                                                     xrange(len(weights))])
                    while array:
                        index = random.choice(array)
                        allproxies.append(all[index][1:])
                        array = [x for x in array if x!=index]
                    del(array)
            ## Now determine which of the servers are actually active and can be used
            # Start all connections in non-blocking mode the revert to blocking and poll each.
            # That way we only wait for 5 seconds no matter how many, instead of n*5 as it is now.
            # Doesn't work, as we will not be able to use timeouts after reverting to blocking
            # mode. ev start first in blocking mode and the rest in nonblocking mode and check
            # them only if first did timeout.
            for proxy in allproxies:
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                s.settimeout(3)
                try:
                    try:
                        s.connect(proxy)
                    except (socket.error, socket.gaierror, socket.herror, socket.timeout):
                        continue
                    else:
                        s.settimeout(1)
                        try:
                            s.send('version\n')
                            res = s.recv(1024)
                        except (socket.error, socket.gaierror, socket.herror, socket.timeout):
                            continue
                        if res:
                            proxy = s.getpeername()
                            break
                finally:
                    s.close()
            else:
                warning("none of the proxies for domain %s have responded" % domain)
                proxy = None
            self.address = proxy
        else:
            self.address = MediaProxyAddress(url)
    def __setattr__(self, name, value):
        readonly = ('address',)
        if name in readonly and self.__dict__.has_key(name):
            raise AttributeError, "'%s' is a readonly attribute" % name
        self.__dict__[name] = value

