
'''
Handle media stream relaying and provide control ports to receive and
process requests from SER (SIP Express Router) or the proxydispatcher.

Copyright 2003-2005 Dan Pascu
'''

import asyncore
import socket
import sys, os, grp, errno, traceback, atexit
import time
import struct
import re
import weakref

from utilities import *
from datatypes import *
from version import release
from request import Request, InvalidRequestError
from configuration import readSettings, ConfigSection, Boolean, dumpSection


class ProxyConfig(ConfigSection):
    _dataTypes = {'socket':     ControlSocket,
                  'proxyIP':    IPAddress,
                  'portRange':  PortRange,
                  'TOS':        HexNumber,
                  'listen':     NetworkAddress,
                  'allow':      NetworkRangeList}
    socket      = '/var/run/mediaproxy.sock'
    group       = 'openser'
    proxyIP     = thisHostIP ## defined in utilities
    portRange   = (60000, 65000)
    TOS         = 0xb8
    listen      = None
    allow       = None
    idleTimeout = 60
    holdTimeout = 60*60
    forceClose  = 0


## We use this to overwrite some of the settings above on a local basis if needed
readSettings('MediaProxy', ProxyConfig)
#dumpSection(ProxyConfig)

if ProxyConfig.allow and NetworkRange('Any') not in ProxyConfig.allow:
    ## Always add ourselves to the accept list
    myself = NetworkRange('127.0.0.1')
    if myself not in ProxyConfig.allow:
        ProxyConfig.allow.append(myself)

maxDataSize = 8*1024

try:
    from accounting import accounting, StopRecord, StopRecordSerializer, QueuedItemProcessingThread
except ImportError:
    warning("accounting is enabled but the accounting module is missing. accounting is not available!")
    accounting = Null()
    dispatcher = Null()
    class StopRecord(object):
        def __new__(typ, *args, **kwargs):
            return None
else:
    class DispatcherNotifyThread(QueuedItemProcessingThread):
        '''
        Notify the proxydispatcher of calls that did timeout and provide
        the dispatcher with the accounting information for them.
        '''
        def __init__(self):
            QueuedItemProcessingThread.__init__(self, name='DispatcherNotify', file='.dispatcher-notify.dat')

        def process(self, record):
            address = record['dispatcher']
            expires = record['expires']
            del record['dispatcher'], record['expires']

            msg = 'timeout %s %s\n' % (record['id'], StopRecordSerializer().dump(record))

            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            try:
                s.settimeout(5)
            except AttributeError:
                pass
            try:
                start = time.time()
                try:
                    s.connect(address)
                    s.send(msg)
                    result = s.recv(maxDataSize).strip()
                except (socket.error, socket.gaierror, socket.herror, socket.timeout), why:
                    try:               motive = why[1]
                    except IndexError: motive = why
                    now = time.time()
                    if now < expires:
                        warning("couldn't notify dispatcher at %s (%s). will retry later." % (address[0], motive))
                        record['dispatcher'] = address
                        record['expires'] = expires
                        # eventually use a different queue only for failures
                        delay = 1
                        duration = now - start
                        if duration < delay:
                            time.sleep(delay-duration)
                        self.put(record) ## put back in the queue and process later
                    else:
                        error("couldn't notify dispatcher at %s (%s). giving up." % (address[0], motive))
                        # eventually log the record to syslog or a file
                    return
                if result.lower() == 'ok':
                    info("notified dispatcher at %s about expired session %s" % (address[0], record['id']))
                elif result.lower() == 'not found':
                    info("session %s already closed by the dispatcher at %s" % (record['id'], address[0]))
                else:
                    error("failed to notify dispatcher at %s (returned `%s')." % (address[0], result))
                    # eventually log the record to syslog or a file
            finally:
                s.close()

        def notify(self, session):
            if not session.dispatcher:
                return ## no dispatcher to notify.
            record = StopRecord(session)
            record['dispatcher'] = session.dispatcher
            record['expires'] = time.time() + 3600 ## try for an hour
            self.put(record)

    dispatcher = DispatcherNotifyThread()
    dispatcher.start()

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

## A few shortcuts, for faster access
proxyIP = ProxyConfig.proxyIP
minPort = ProxyConfig.portRange[0]
maxPort = ProxyConfig.portRange[1]
crtPort = minPort

Sessions = {}

## For computing traffic through mediaproxy
byteBucket = [0, 0, 0]
Traffic    = [0, 0, 0]

## Non public networks. This includes RFC1918 networks, localnet (127.0.0.0/8),
## broadcast networks (224.0.0.0/4) and 0.0.0.0/8
nonPublicNetworks = [
    {'name': '0.0.0.0',     'value': 0x00000000L, 'mask': 0xff000000L},
    {'name': '10.0.0.0',    'value': 0x0a000000L, 'mask': 0xff000000L},
    {'name': '127.0.0.0',   'value': 0x7f000000L, 'mask': 0xff000000L},
    {'name': '172.16.0.0',  'value': 0xac100000L, 'mask': 0xfff00000L},
    {'name': '192.168.0.0', 'value': 0xc0a80000L, 'mask': 0xffff0000L},
    {'name': '224.0.0.0',   'value': 0xe0000000L, 'mask': 0xf0000000L}
]

rtpPayloads = {
     0: 'G711u', 1: '1016',  2: 'G721',  3: 'GSM',  4: 'G723',  5: 'DVI4', 6: 'DVI4',
     7: 'LPC',   8: 'G711a', 9: 'G722', 10: 'L16', 11: 'L16',  14: 'MPA', 15: 'G728',
    18: 'G729', 25: 'CelB', 26: 'JPEG', 28: 'nv',  31: 'H261', 32: 'MPV', 33: 'MP2T',
    34: 'H263'
}

## Helper functions

def isPublicIP(address):
    try:
        adr = socket.inet_aton(address)
    except:
        if address:
            print >>sys.stderr, "Invalid IP address: '%s'" % address
        return False
    for net in nonPublicNetworks:
        netaddr = struct.unpack('!L', adr)[0] & net['mask']
        if (netaddr == net['value']):
            return False
    return True

## We overwrite the loop in the asyncore module to be able to call an extra function
## at the end of the loop (to let us timeout idle RTP sessions).
## While timeout can be a fractional number, we only use integers.
def loop(timeout=1, map=None):
    if map is None:
        map = asyncore.socket_map
    timer = SessionTimer(timeout)
    while map:
        try:
            asyncore.poll3(timeout, map)
        except OverflowError:
            ## This is a very weird error, which happens rarely and unpredictably.
            ## Its backtrace doesn't make any sense because conversion from long
            ## to int is not needed at all, so I guess the error message is
            ## misleading and the error is somewhere else, most likely in the
            ## underlying C select module. 
            ##
            ## The backtrace reads like:
            ##     File "/usr/lib/python2.3/asyncore.py", line 174, in poll3 
            ##       for fd, flags in r: 
            ##   OverflowError: long int too large to convert to int
            ##
            warning("captured OverflowError in main loop")
            pass
        timeout = timer.run()

## Classes

class CommandError(Exception):     pass
class MediaProxyError(Exception):  pass
class RTPSessionError(Exception):  pass
class MediaStreamError(Exception): pass
class RTPStreamError(Exception):   pass


class SessionTimer:
    def __init__(self, period):
        self.start = time.time()
        self.period = period
        self.next = self.start + period
        #self.before = self.start

    def run(self):
        global byteBucket, Traffic
        now = time.time()
        if now >= self.next:
            period = self.period
            forceClose = ProxyConfig.forceClose
            ## Check sessions and timeout idle ones
            for session in Sessions.values():
                for stream in session.mediaStreams:
                    stream.idleTime += period
                if session.didTimeout:
                    session.end()
                elif forceClose and session.duration >= forceClose:
                    session.forceClosed = True
                    session.end()
            Traffic    = [int(val/period) for val in byteBucket]
            byteBucket = [0, 0, 0]
            self.next += period
            #print "Loop time: %5.3f ms" % ((time.time() - now)*1000)
            return self.next - time.time()
        return self.next - now


class ControlSocket(asyncore.dispatcher):
    '''Control MediaProxy through a local socket on the filesystem'''
    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 rtphandler module is loaded
        ## This way we can change the default values after rtphandler is loaded
        ## and we get the updated values
        path = path or ProxyConfig.socket
        group = group or ProxyConfig.group
        self.path = path
        try:
            os.unlink(path)
        except OSError:
            pass
        asyncore.dispatcher.__init__(self)
        self.create_socket(socket.AF_UNIX, socket.SOCK_STREAM)
        ## we don't need this for unix sockets #self.set_reuse_addr()
        try:
            self.bind(path)
            self.listen(5)
        except socket.error, why:
            raise MediaProxyError, "couldn't create command socket (%s)" % why[1]
        try:
            groupId = grp.getgrnam(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("the SIP proxy may be unable 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("the SIP proxy may be unable to communicate with us!")
        atexit.register(self.handle_close)

    def writable(self):
        return 0

    def handle_accept(self):
        try:
            connection, address = self.accept()
        except socket.error: ## Rare Linux error
            print >>sys.stderr, "Error accepting connection on control socket"
            return
        except TypeError: ## Rare FreeBSD3 error
            print >>sys.stderr, "EWOULDBLOCK exception on control socket"
            return
        cmdHandler = CommandHandler(connection, address)

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

    def handle_error(self):
        asyncore.dispatcher.handle_error(self)
        warning("control socket handler closed by an exception. creating new handler.")
        newhandler = ControlSocket()


class RemoteControl(asyncore.dispatcher):
    '''Control MediaProxy from remote through a TCP socket'''
    def __init__(self, address=None, allowedHosts=()):
        address = address or ProxyConfig.listen
        asyncore.dispatcher.__init__(self)
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.set_reuse_addr()
        try:
            self.bind(address)
            self.listen(5)
        except socket.error, why:
            raise MediaProxyError, "couldn't create command socket (%s)" % why[1]
        allowed = allowedHosts or ProxyConfig.allow
        if allowed and (0L, 0L) in allowed:
            self.allowedHostList = ()
        else:
            self.allowedHostList = allowed
        atexit.register(self.handle_close)
    
    def aclStr(self):
        '''return a string representation of the access list saying from where are connections allowed'''
        if self.allowedHostList is None:
            return "nowhere"
        elif self.allowedHostList is ():
            return "anywhere"
        else:
            from math import log
            strlist = []
            for adr in self.allowedHostList:
                if adr == NetworkRange('None'):
                    continue
                netbase = socket.inet_ntoa(struct.pack('!L', adr[0]))
                hostmask = 0xFFFFFFFFL ^ adr[1]
                bits = 32 - int(log(hostmask+1, 2))
                if bits == 32:
                    strlist.append(netbase)
                else:
                    strlist.append("%s/%d" % (netbase, bits))
            return ', '.join(strlist)

    def isAllowed(self, address):
        '''None means deny everything. () means allow everything'''
        if self.allowedHostList is None:
            return False
        if self.allowedHostList is ():
            return True
        try:
            adr = socket.inet_aton(address)
        except:
            return False
        for netbase, mask in self.allowedHostList:
            if (struct.unpack('!L', adr)[0] & mask) == netbase:
                return True
        return False

    def writable(self):
        return 0

    def handle_accept(self):
        try:
            connection, address = self.accept()
        except socket.error: ## Rare Linux error
            print >>sys.stderr, "Error accepting connection on control socket"
            return
        except TypeError: ## Rare FreeBSD3 error
            print >>sys.stderr, "EWOULDBLOCK exception on control socket"
            return
        if not self.isAllowed(address[0]):
            connection.close()
            print >>sys.stderr, "Connection from %s is not allowed" % address[0]
            return
        cmdHandler = CommandHandler(connection, address)

    def handle_close(self):
        self.close()

    def handle_error(self):
        asyncore.dispatcher.handle_error(self)
        warning("remote command handler closed by an exception. creating new handler.")
        newhandler = RemoteControl()


class CommandHandler(asyncore.dispatcher):
    '''Handles a command received through a control socket'''

    def __init__(self, connection, address):
        asyncore.dispatcher.__init__(self, connection)
        self.addr = address
        self.outBuffer = ''
        self.inBuffer = ''

    def readable(self):
        return True

    def writable(self):
        return len(self.outBuffer) > 0

    def read_input(self):
        command = self.recv(maxDataSize)
        if not command:
            self.close()
            self.inBuffer  = ''
            self.outBuffer = ''
            return
        if command.find('\n') < 0:
            self.inBuffer += command
            if len(self.inBuffer) > maxDataSize:
                ## Somebody is flooding
                self.close()
                self.inBuffer = ''
                self.outBuffer = ''
            return
        else:
            command = self.inBuffer + command
            self.inBuffer = ''
        return command

    def handle_read(self):
        line = self.read_input()
        if not line:
            return
        print line.rstrip()
        start = time.time()
        try:
            request = Request(line)
        except InvalidRequestError, msg:
            error(msg)
            self.close()
        else:
            try:
                result = self.process(request)
            except InvalidRequestError, msg:
                error(msg)
                self.close()
            else:
                if result is None:
                    self.close()
                else:
                    self.outBuffer = "%s\n" % result
        print "execution time: %5.2f ms" % ((time.time() - start) * 1000)

    def handle_write(self):
        sent = self.send(self.outBuffer)
        self.outBuffer = self.outBuffer[sent:]
        if not self.outBuffer:
            self.close()

    def handle_close(self):
        self.close()

    def process(self, request):
        global Sessions

        cmd = request.cmd

        if cmd == 'status':
            result = ["version %s" % release,
                      "proxy %s %s/%s/%s" % ((len(Sessions),) + tuple(Traffic))]
            for session in Sessions.values():
                result.append(session.statline)
                for stream in session.mediaStreams:
                    result.append(stream.statline)
            return '\n'.join(result)
        elif cmd == 'status2':
            result = ["version %s" % release, 
                      "proxy %s %s/%s/%s" % ((len(Sessions),) + tuple(Traffic))]
            for session in Sessions.values():
                for stream in session.mediaStreams:
                    rtp = stream.rtpStream
                    idleTime = stream.idleTime
                    try:    adr1 = rtp.caller.paddress
                    except: adr1 = '?.?.?.?:?'
                    try:    adr2 = rtp.called.paddress
                    except: adr2 = '?.?.?.?:?'
                    ## We test for hold first because a call can be put on hold
                    ## before RTP data is sent (if phones have silence detection
                    ## capabilities for example)
                    if session.onHold:     status = 'hold'
                    elif rtp.signedIn < 2: status = 'inactive'
                    elif idleTime > 2:     status = 'idle'
                    else:                  status = 'active'
                    pbytes = "%s/%s/%s" % stream.bytes
                    line = '%s %s %s %s %s %s %s %s %s %s %s %s %s %s' % \
                         (session._id, adr1, rtp.paddr, adr2, status,
                          session.duration, idleTime, pbytes, rtp.rtpCodec,
                          stream.type,
                          session.callerInfo['from'], session.callerInfo['to'],
                          session.encodedAgents[0], session.encodedAgents[1])
                    result.append(line)
            return '\n'.join(result)
        elif cmd == 'summary':
            return "%s %s/%s/%s" % ((len(Sessions),) + tuple(Traffic))
        elif cmd == 'version':
            return release
        elif cmd == 'request' or cmd == 'lookup':
            # callers are request, called are lookup
            caller = (cmd == 'request')

            try:
                session = Sessions[request.sessionId]
            except KeyError:
                #if request.info.totag:
                #    ## Don't create a new session on re-INVITEs
                #    ## (we may decided not to use mediaproxy on the 1st INVITE)
                #    print "request.info.totag"
                #    return None
                ## Create a brand new session
                # todo: embed logic of encoding/decoding address to/from string into StructureInfo -Dan
                if request.info.dispatcher is not None:
                    try:
                        address = str2address(request.info.dispatcher)
                    except ValueError:
                        warning('received invalid packed dispatcher address from %s' % ip)
                        request.info.dispatcher = None
                    else:
                        if address[0] == '0.0.0.0':
                            if type(self.addr) is tuple:
                                request.info.dispatcher = (self.addr[0], address[1])
                            else:
                                request.info.dispatcher = None ## received over /var/run/mediaproxy.sock
                        else:
                            request.info.dispatcher = address
                try:
                    session = RTPSession(request, caller)
                except RTPSessionError, why:
                    raise InvalidRequestError, "cannot create session for '%s': %s" % (request.sessionId, why)
            else:
                session.updateStreams(request, caller)

            # Mmmmm, I'm a bit dubious about this section...
            # I think the on hold status show be per RTP stream
            ips = [address.split(':')[0] for address in request.streams.split(',')]
            ## A session can be put on hold only after both parties have sent their
            ## contact data. However they don't have to sign in on RTP before
            ## they can put the call on hold (they may have silence detection)
            if session.callerComplete and session.calledComplete and '0.0.0.0' in ips:
                session.onHold  = True
                session.timeout = ProxyConfig.holdTimeout
            else:
                session.onHold  = False
                session.timeout = ProxyConfig.idleTimeout

            # remove (or make it show in debug mode only)
            #for stream in session.mediaStreams:
                #stream.rtpStream.caller.show('rtp', compact=1)
                #stream.rtcpStream.caller.show('rtcp', compact=1)
                #stream.rtpStream.called.show('rtp', compact=1)
                #stream.rtcpStream.called.show('rtcp', compact=1)
            # until here
            return session.endpointAddresses()

        elif cmd == 'delete':
            try:
                session = Sessions[request.sessionId]
            except KeyError:
                return None
            session.end()
            if session.dispatcher:
                return StopRecordSerializer().dump(StopRecord(session))
            else:
                accounting.log(session)
                return None
        else:
            raise InvalidRequestError, "unknown command: `%s'" % cmd


class RTPPeer(object):
    '''A RTPPeer represents one endpoint of a RTP stream. There will be 4 RTPPeer's
    in a RTP session (2 from the data stream + 2 from the control stream)'''

    class Values:
        '''Hold the actual parameter values for a RTPPeer'''
        def __init__(self):
            self._parent     = None
            self.contactIP   = None
            self.visibleIP   = None
            self.expectedIP  = None
            self.rtpPort     = 0
            self.asymmetric  = 0
            self.local       = 0
            self.address     = None
            self.paddress    = '?.?.?.?:?' ## printable form of self.address
            self.signedInAddress = None
            self.expectedAddress = None
        def updateContacts(self):
            signedInIP   = (self.signedInAddress and self.signedInAddress[0]) or None
            signedInPort = (self.signedInAddress and self.signedInAddress[1]) or 0
            publicContactIP = (isPublicIP(self.contactIP) and self.contactIP) or None
            probableIncomingIP = (self.local and self.visibleIP) or None
            # This is problematic. contactIP may be a public IP, yet still different
            # from visibleIP (if the client uses the address of a conference server
            # for example)
            #if self.local and publicContactIP and self.visibleIP != self.contactIP:
                ### Client is using public IP address behind NAT. Ignore.
                #publicContactIP = None
            ## add probableIncomingIP to contactIP if you want to start sending to a peer
            ## that is behind NAT before he signs in. This requires static port forwarding
            ## in the NAT box for asymmetric clients to work, while for symmetric clients
            ## the chances to reach them are dim and depend on the type of NAT and if the
            ## connection was before opened from inside and have not yet timed out.
            contactIP = signedInIP or publicContactIP #or probableIncomingIP
            if self.asymmetric:
                port = self.rtpPort
            else:
                port = signedInPort or self.rtpPort
            self.address = (contactIP and port and (contactIP, port)) or None
            self.paddress = "%s:%s" % (self.address or ('?.?.?.?', '?'))
            self.expectedIP = publicContactIP or probableIncomingIP
            self.expectedAddress = (self.expectedIP and port and (self.expectedIP, port)) or None

    def __init__(self, name, contactIP=None, rtpPort=0, visibleIP=None, local=0, asymmetric=0):
        self.__v = self.Values()
        self.__v._parent = weakref.proxy(self)
        self.__v.contactIP = contactIP
        self.__v.visibleIP = visibleIP
        self.__v.rtpPort = rtpPort
        self.__v.asymmetric = asymmetric
        self.__v.local = local
        self.__v.signedInAddress = None
        self.__v.updateContacts()
        self.name = name

    def update(self, contactIP=None, rtpPort=0, visibleIP=None, local=0, asymmetric=0):
        self.contactIP = contactIP
        self.rtpPort = rtpPort
        self.visibleIP = visibleIP
        self.local = local
        self.asymmetric = asymmetric

    def getci(self): return self.__v.contactIP
    def getvi(self): return self.__v.visibleIP
    def getei(self): return self.__v.expectedIP
    def getea(self): return self.__v.expectedAddress
    def getrp(self): return self.__v.rtpPort
    def getas(self): return self.__v.asymmetric
    def getlo(self): return self.__v.local
    def getca(self): return self.__v.address
    def getpa(self): return self.__v.paddress
    def getsa(self): return self.__v.signedInAddress
    def setci(self, val): self.__v.contactIP       = val; self.__v.updateContacts()
    def setvi(self, val): self.__v.visibleIP       = val; self.__v.updateContacts()
    def setrp(self, val): self.__v.rtpPort         = val; self.__v.updateContacts()
    def setas(self, val): self.__v.asymmetric      = val; self.__v.updateContacts()
    def setlo(self, val): self.__v.local           = val; self.__v.updateContacts()
    def setsa(self, val): self.__v.signedInAddress = val; self.__v.updateContacts()
    contactIP  = property(getci, setci)
    visibleIP  = property(getvi, setvi)
    expectedIP = property(getei)
    rtpPort    = property(getrp, setrp)
    asymmetric = property(getas, setas)
    local      = property(getlo, setlo)
    address    = property(getca)
    paddress   = property(getpa)  ## Printable form of self.address
    signedInAddress = property(getsa, setsa)
    expectedAddress = property(getea)
    del(getci, getvi, getei, getea, getrp, getas, getca, getpa, getsa, getlo)
    del(setci, setvi, setrp, setas, setsa, setlo)

    def show(self, type, compact=0):
        if compact:
            print "%s.%-4s: sendto=%s expectIP=%s port=%d local=%d asym=%d" % (
                self.name, type.lower(), str(self.address), self.expectedIP,
                self.rtpPort, self.local, self.asymmetric)
        else:
            print "Contact IP:  %s" % self.contactIP
            print "Visible IP:  %s" % self.visibleIP
            print "Expected IP: %s" % self.expectedIP
            print "Signed in A: %s" % str(self.signedInAddress)
            print "Contact Adr: %s" % str(self.address)
            print "Port       : %d" % self.rtpPort
            print "Local      : %d" % self.local
            print "Asymmetric : %d" % self.asymmetric


class RTPStream(asyncore.dispatcher):
    '''A RTPStream is an entity consisting in a RTP proxy and 2 RTP endpoints (RTPPeers).
    The RTP proxy is relaying the data between the 2 RTP endpoints both ways.
    The RTP proxy is intrinsic in the structure of the RTPStream class, while the RTPPeer's
    are implemented by an external class.'''

    def __init__(self, stream, sock, name, control=False):
        asyncore.dispatcher.__init__(self, sock)
        self.addr = sock.getsockname()
        self.paddr = "%s:%s" % self.addr
        self.stream = weakref.proxy(stream) ## The media stream we belong to
        self.name = name
        self.caller = None
        self.called = None
        self.bytes   = [0, 0, 0]  ## from caller, from called, relayed
        self.packets = [0, 0, 0]  ## from caller, from called, relayed
        if control:
            self.payload = 0x7f
            self.rtpCodec = 'NA'
        else:
            self.payload = None
            self.rtpCodec = 'Unknown'
        self.signedIn = 0
        self.isDataStream = not control
        self.warned3rdparty = 0

    def recvfrom(self, buffer_size):
        try:
            return self.socket.recvfrom(buffer_size)
        except socket.error, why:
            if why[0] != errno.EWOULDBLOCK:
                self.handle_close()
                raise socket.error, why
            return '', ('dummy', 0)

    def sendto(self, data, address):
        try:
            return self.socket.sendto(data, address)
        except socket.error, why:
            if why[0] != errno.EWOULDBLOCK:
                self.handle_close()
                raise socket.error, why
            return 0

    def writable(self):
        return 0

    def setCodec(self, data):
        try:
            payload = ord(data[1]) & 0x7f
        except IndexError:
            print >>sys.stderr, "warning: insufficient data in RTP packet to determine codec type"
        else:
            ## payload=13  is 'Confort noise'
            ## payload=101 is 'Telephone event'
            if payload not in (13, 101):
                self.payload = payload
                unknownName = ((payload>95 and 'Dynamic(%d)') or 'Unknown(%d)') % payload
                self.rtpCodec = rtpPayloads.get(payload, unknownName)

    def signIn(self, peer, address, data):
        peer.signedInAddress = address
        self.signedIn += 1
        self.setCodec(data)
        # eventually remove (or make it show only when debug is enabled)
        print "session %s: %s signed in from %s:%s (%s) (will return to %s:%s)" % \
              ((self.stream.session._id, peer.name) + address + (self.name,) + peer.address)

    def handle_read(self):
        global byteBucket

        data, address = self.recvfrom(65536)
        if not (data and self.stream.complete):
            return
        if self.payload is None:
            self.setCodec(data)
        bytes = len(data) + 28 ## 28 is from IP+UDP headers
        caller = self.caller
        called = self.called

        if address == caller.signedInAddress:
            sender = 0
            destination = called.address
        elif address == called.signedInAddress:
            sender = 1
            destination = caller.address
        elif not caller.signedInAddress and address == caller.expectedAddress:
            self.signIn(caller, address, data)
            sender = 0
            destination = called.address
        elif not called.signedInAddress and address == called.expectedAddress:
            self.signIn(called, address, data)
            sender = 1
            destination = caller.address
        elif not caller.signedInAddress and address[0] == caller.expectedIP:
            self.signIn(caller, address, data)
            sender = 0
            destination = called.address
        elif not called.signedInAddress and address[0] == called.expectedIP:
            self.signIn(called, address, data)
            sender = 1
            destination = caller.address
        elif not caller.signedInAddress and caller.local and address[0]==caller.visibleIP:
            self.signIn(caller, address, data)
            sender = 0
            destination = called.address
        elif not called.signedInAddress and called.local and address[0]==called.visibleIP:
            self.signIn(called, address, data)
            sender = 1
            destination = caller.address
        elif not caller.signedInAddress and not caller.expectedIP:
            self.signIn(caller, address, data)
            sender = 0
            destination = called.address
        elif not called.signedInAddress and not called.expectedIP:
            self.signIn(called, address, data)
            sender = 1
            destination = caller.address
        elif caller.signedInAddress and address[0]==caller.visibleIP and address[1]==caller.rtpPort:
            self.signIn(caller, address, data)
            sender = 0
            destination = called.address
        elif called.signedInAddress and address[0]==called.visibleIP and address[1]==called.rtpPort:
            self.signIn(called, address, data)
            sender = 1
            destination = caller.address
        else:
            ## Allow called party to change address (for example after a 183 from
            ## PSTN, if the call is not answered it is diverted to an announcement
            ## on a media server, which will try to sign in from its own address).
            # to make this more secure, we should only allow a change in the 
            # address if we have previously received a lookup command that changed
            # the contact details and only accept a new sign in from that address.
            self.signIn(called, address, data)
            sender = 1
            destination = caller.address
            #if not self.warned3rdparty:
                #self.warned3rdparty = 1
                #warning("Received packet from a third party: %s:%s" % address)
            #return
        byteBucket[sender]   += bytes
        self.bytes[sender]   += bytes
        self.packets[sender] += 1
        if destination:
            byteBucket[2]    += bytes
            self.bytes[2]    += bytes
            self.packets[2]  += 1
            self.sendto(data, destination)
        if self.isDataStream:
            ## control stream traffic alone doesn't reset the timeout
            self.stream.idleTime = 0

    def handle_close(self):
        self.close()


class MediaStream(object):
    def __init__(self, session, mediatype='Audio'):
        self.session = weakref.proxy(session)
        self.type = mediatype.capitalize()
        (dataSlot, ctrlSlot) = self.__findRTPSlot()
        self.rtpStream  = RTPStream(self, dataSlot, name='RTP')
        self.rtcpStream = RTPStream(self, ctrlSlot, name='RTCP', control=True)
        self.idleTime = 0
        self.complete = 0

    ## complete is when the media stream is described for both ends
    #def getco(self): return (self.rtpStream.caller and self.rtpStream.called)
    #complete = property(getco)
    #del(getco)

    def getby(self):
        di, do, dr = self.rtpStream.bytes
        ci, co, cr = self.rtcpStream.bytes
        return (di+ci, do+co, dr+cr)
    def getpa(self):
        di, do, dr = self.rtpStream.packets
        ci, co, cr = self.rtcpStream.packets
        return (di+ci, do+co, dr+cr)
    def getsl(self):
        rtp = self.rtpStream
        caller = rtp.caller
        called = rtp.called
        idleTime = self.idleTime
        try:    adr1 = caller.paddress
        except: adr1 = '?.?.?.?:?'
        try:    adr2 = called.paddress
        except: adr2 = '?.?.?.?:?'
        ## We test for hold first because a call can be put on hold before RTP
        ## data is sent (if phones have silence detection capabilities for example)
        if self.session.onHold: status = 'hold'
        elif rtp.signedIn < 2:  status = 'inactive'
        elif idleTime > 2:      status = 'idle'
        else:                   status = 'active'
        pbytes = "%s/%s/%s" % self.bytes
        return 'stream %s %s %s %s %s %s %s %s' % \
               (adr1, adr2, rtp.paddr, pbytes, status, rtp.rtpCodec,
                self.type, idleTime)

    bytes    = property(getby)
    packets  = property(getpa)
    statline = property(getsl)
    
    del(getby, getpa, getsl)
    
    def close(self):
        self.rtpStream.close()
        self.rtcpStream.close()

    def setCaller(self, address, visibleIP, asymmetric):
        ip = address[0]
        port = int(address[1] or 0)
        local = self.session.localCaller
        if self.rtpStream.caller:
            # update the RTPPeer
            self.rtpStream.caller.update(ip, port,   visibleIP, local, asymmetric)
            self.rtcpStream.caller.update(ip, port+1, visibleIP, local, asymmetric)
        else:
            rtpStream  = self.rtpStream
            rtcpStream = self.rtcpStream
            rtpStream.caller  = RTPPeer('caller', ip, port,   visibleIP, local, asymmetric)
            rtcpStream.caller = RTPPeer('caller', ip, port+1, visibleIP, local, asymmetric)
            self.complete = rtpStream.called is not None

    def setCalled(self, address, visibleIP, asymmetric):
        ip = address[0]
        port = int(address[1] or 0)
        local = self.session.localCalled
        if self.rtpStream.called:
            # update the RTPPeer
            self.rtpStream.called.update(ip, port,   visibleIP, local, asymmetric)
            self.rtcpStream.called.update(ip, port+1, visibleIP, local, asymmetric)
        else:
            rtpStream  = self.rtpStream
            rtcpStream = self.rtcpStream
            rtpStream.called  = RTPPeer('called', ip, port,   visibleIP, local, asymmetric)
            rtcpStream.called = RTPPeer('called', ip, port+1, visibleIP, local, asymmetric)
            self.complete = rtpStream.caller is not None

    def __findRTPSlot(self):
        '''Find a pair of consecutive ports to bind to and return
        two sockets bound to them. The first port must be even.'''

        def validateRTPPort(port):
            if (port & 0x01): port += 1
            if port < minPort or port >= maxPort: port = minPort
            return port

        global crtPort

        dsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        csock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            dsock.setsockopt(socket.IPPROTO_IP, socket.IP_TOS, ProxyConfig.TOS)
            csock.setsockopt(socket.IPPROTO_IP, socket.IP_TOS, ProxyConfig.TOS)
        except socket.error:
            pass ## Not critical
        crtPort = validateRTPPort(crtPort)
        for port in range(crtPort, maxPort, 2) + range(minPort, crtPort, 2):
            try:
                try:
                    dsock.bind((proxyIP, port))
                except socket.error, why:
                    if why[0] in (errno.EADDRINUSE, errno.EACCES):
                        continue
                    else:
                        raise
                else:
                    try:
                        csock.bind((proxyIP, port+1))
                    except socket.error, why:
                        if why[0] in (errno.EADDRINUSE, errno.EACCES):
                            dsock.close()
                            dsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                            continue
                        else:
                            raise
                    else:
                        crtPort = port + 2
                        return (dsock, csock)
            except:
                ## Print traceback
                info = sys.exc_info()
                try:
                    traceback.print_exception(*info)
                finally:
                    del info
                break
        dsock.close()
        csock.close()
        raise MediaStreamError, "cannot find a free pair of ports for a media stream"


class RTPSession(object):
    # Either the caller or called can initialise the RTPSession first
    # The order can go:
    # INVITE + SDP -> 200 + SDP -> ACK (Most common, caller initialises RTPSession)
    # INVITE -> 200 + SDP -> ACK + SDP (Still valid, called initialises RTPSession)
    def __init__(self, request, caller):
        global Sessions

        # These can be initialise by either end of the call
        self._id = request.sessionId
        self.callerDomain = request.callerDomain
        self.calledDomain = request.calledDomain
        self.localCaller  = request.localCaller
        self.localCalled  = request.localCalled
        self.startTime = round(time.time())
        self.conversationStart = None
        self.timeout = ProxyConfig.idleTimeout
        # onHold should probably be a per mediaStream setting...
        self.onHold = False
        self.forceClosed = False
        self.callerComplete = False
        self.calledComplete = False
        self.ended = False
        # These get initialise in updateStreams
        self.userAgents = ['Unknown', 'Unknown']
        self.encodedAgents = ['Unknown', 'Unknown']
        self.dispatcher = request.info.dispatcher
        self.callerInfo = request.info
        ## We need this info until the called party signs in, else if a CANCEL
        ## from the caller comes before the OK it will raise an exception because
        ## there is no calledInfo to compute the stats when the session ends
        self.calledInfo = request.info ## eventually make a copy of it
        self.mediaStreams = []
        self.updateStreams(request, caller)
        if not self.mediaStreams:
            raise RTPSessionError, "there are no media streams to process"

        Sessions[self._id] = self
        adr = self.endpointAddresses().split()
        print "session %s: started. listening on %s:%s" % (self._id, adr[0], ','.join(adr[1:]))

    ## Properties that depend on the state of the media streams
    def getit(self): return min([x.idleTime for x in self.mediaStreams] or [0])
    def getco(self): return [stream.rtpStream.rtpCodec for stream in self.mediaStreams]
    def getty(self): return [stream.type for stream in self.mediaStreams]
    def getta(self): return [self.calledInfo['fromtag'], self.calledInfo['totag']]
    def getby(self):
        bytes = [stream.bytes for stream in self.mediaStreams]
        return reduce(lambda x,y: (x[0]+y[0], x[1]+y[1], x[2]+y[2]), bytes)
    def getpa(self):
        packs = [stream.packets for stream in self.mediaStreams]
        return reduce(lambda x,y: (x[0]+y[0], x[1]+y[1], x[2]+y[2]), packs)

    ## Properties related to call duration
    def getdu(self): return int(round(time.time()) - self.startTime)
    def getsu(self):
        t = self.conversationStart or round(time.time())
        return int(t - self.startTime)
    def getcd(self):
        if self.conversationStart is None:
            return 0
        callDuration = int(round(time.time()) - self.conversationStart)
        if self.didTimeout:
            callDuration = max(callDuration - int(self.idleTime), 0)
        return callDuration

    ## State related properties
    def getdt(self): return (self.idleTime >= self.timeout)
    def getsl(self):
        return 'session %s %s %s %s %s %s' % \
               (self._id, self.callerInfo['from'], self.callerInfo['to'],
                self.encodedAgents[0], self.encodedAgents[1], self.duration)
    def getin(self): # not used anymore. kept for reference only. TODO: remove -Dan
        return {'id': self._id,
                'startTime': self.startTime,
                'conversationStart': self.conversationStart,
                'duration': self.duration,
                'callDuration': self.callDuration,
                'bytes': self.bytes,
                'packets': self.packets,
                'codecs': self.codecs,
                'streamTypes': self.streamTypes,
                'userAgents': self.userAgents,
                'tags': self.tags,
                'didTimeout': self.didTimeout,
                'forceClosed': self.forceClosed}

    codecs       = property(getco)
    streamTypes  = property(getty)
    tags         = property(getta)
    bytes        = property(getby)
    packets      = property(getpa)

    idleTime     = property(getit)
    callDuration = property(getcd)
    duration     = property(getdu)
    setupTime    = property(getsu)

    didTimeout   = property(getdt)
    statline     = property(getsl)
    info         = property(getin)
    
    del(getit, getdt, getco, getty, getta, getdu)
    del(getsu, getby, getpa, getcd, getsl, getin)

    def updateStreams(self, request, caller):
        streams = [s.split(':') for s in request.streams.split(',')]
        visibleIP = request.signalingIP
        asymmetric = None

        # initialise the caller or called specific settings
        if caller:
            asymmetric = self.callerInfo.flags.asymmetric
            if not self.callerComplete:
                self.userAgents[0] = request.userAgent.decode('quopri')
                self.encodedAgents[0] = request.userAgent
                self.callerInfo = request.info
                self.conversationStart = round(time.time())
                self.callerComplete = True
        else:
            asymmetric = self.calledInfo.flags.asymmetric
            if not self.calledComplete:
                self.userAgents[1] = request.userAgent.decode('quopri')
                self.encodedAgents[1] = request.userAgent
                self.calledInfo = request.info
                self.conversationStart = round(time.time())
                self.calledComplete = True

        # initialise the mediaStreams
        for i in range(len(streams)):
            sinfo = streams[i]
            try:
                stream = self.mediaStreams[i]
                # update the stream
                if caller:
                    stream.setCaller(sinfo, visibleIP, asymmetric)
                else:
                    stream.setCalled(sinfo, visibleIP, asymmetric)
            except IndexError:
                # add a new stream
                try:
                    mediatype = sinfo[2]
                except IndexError:
                    mediatype = 'Audio'
                    # remove at a later time
                    print("warning: SER is using an old mediaproxy.so module. "
                          "Please upgrade it to get correct media type information.")
                try:
                    stream = MediaStream(self, mediatype)
                except MediaStreamError:
                    # check this. also at session creation
                    #for stream in self.mediaStreams:
                        #stream.close()
                    raise RTPSessionError, "cannot find enough free ports for the media streams"
                if caller:
                    stream.setCaller(sinfo, visibleIP, asymmetric)
                else:
                    stream.setCalled(sinfo, visibleIP, asymmetric)
                #self.mediaStreams[i] = stream
                # Mmmm, python doesn't seem to like random insertions
                # into lists. append should always do what I want...
                self.mediaStreams.append(stream)
        
    def close(self):
        for stream in self.mediaStreams:
            stream.close()
    __del__ = close  ## Free sockets if a session is deleted without being ended first

    def end(self):
        if not self.ended:
            global Sessions

            self.close()
            try:
                del(Sessions[self._id])
            except KeyError:
                pass

            ## Show summary in syslog
            print("session %s: %s/%s/%s packets, %s/%s/%s bytes (caller/called/relayed)" %
                  ((self._id,) + self.packets + self.bytes))
            what = "ended"
            if self.didTimeout:
                what += " (did timeout)"
            elif self.forceClosed:
                what += " (force closed)"
            print "session %s: %s." % (self._id, what)

            if self.didTimeout or self.forceClosed:
                ## need to notify dispatcher or update database because the call was closed by us
                if self.dispatcher:
                    dispatcher.notify(self)
                else:
                    accounting.log(self)

            self.ended = True

    def endpointAddresses(self):
        '''Return the IP address and the RTP data port(s) that will be used by this session'''
        ports = [str(stream.rtpStream.addr[1]) for stream in self.mediaStreams]
        return " ".join([proxyIP] + ports)

