sip-subscribe-winfo3

#!/usr/bin/env python3

import os
import random
import select
import sys
import termios

from collections import deque
from optparse import OptionParser
from threading import Thread
from time import time

from application import log
from application.notification import IObserver, NotificationCenter, NotificationData
from application.python.queue import EventQueue
from eventlib.twistedutil import join_reactor
from twisted.internet import reactor
from twisted.internet.error import ReactorNotRunning
from zope.interface import implementer

from sipsimple.account import Account, AccountManager, BonjourAccount
from sipsimple.application import SIPApplication
from sipsimple.configuration import ConfigurationError, ConfigurationManager
from sipsimple.configuration.settings import SIPSimpleSettings
from sipsimple.core import ContactHeader, Engine, FromHeader, Route, RouteHeader, SIPCoreError, SIPURI, Subscription, ToHeader
from sipsimple.lookup import DNSLookup
from sipsimple.payloads import ParserError
from sipsimple.payloads.watcherinfo import WatcherInfoDocument
from sipsimple.storage import FileStorage
from sipsimple.threading import run_in_twisted_thread

from sipclient.configuration import config_directory
from sipclient.configuration.account import AccountExtension
from sipclient.configuration.settings import SIPSimpleSettingsExtension
from sipclient.log import Logger


class InputThread(Thread):
    def __init__(self, application):
        Thread.__init__(self)
        self.application = application
        self.daemon = True
        self._old_terminal_settings = None

    def run(self):
        notification_center = NotificationCenter()
        while True:
            for char in self._getchars():
                if char == "\x04":
                    self.application.stop()
                    return
                else:
                    notification_center.post_notification('SAInputWasReceived', sender=self, data=NotificationData(input=char))

    def stop(self):
        self._termios_restore()

    def _termios_restore(self):
        if self._old_terminal_settings is not None:
            termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, self._old_terminal_settings)

    def _getchars(self):
        fd = sys.stdin.fileno()
        if os.isatty(fd):
            self._old_terminal_settings = termios.tcgetattr(fd)
            new = termios.tcgetattr(fd)
            new[3] = new[3] & ~termios.ICANON & ~termios.ECHO
            new[6][termios.VMIN] = b'\000'
            try:
                termios.tcsetattr(fd, termios.TCSADRAIN, new)
                if select.select([fd], [], [], None)[0]:
                    return sys.stdin.read(4192)
            finally:
                self._termios_restore()
        else:
            return os.read(fd, 4192)


@implementer(IObserver)
class WinfoApplication(object):

    def __init__(self, account_name, trace_sip, trace_pjsip, trace_notifications):
        self.account_name = account_name
        self.input = InputThread(self)
        self.output = EventQueue(lambda event: sys.stdout.write(event+'\n'))
        self.logger = Logger(sip_to_stdout=trace_sip, pjsip_to_stdout=trace_pjsip, notifications_to_stdout=trace_notifications)
        self.success = False
        self.account = None
        self.subscription = None
        self.stopping = False

        self._subscription_routes = None
        self._subscription_timeout = 0.0
        self._subscription_wait = 0.5

        account_manager = AccountManager()
        engine = Engine()
        notification_center = NotificationCenter()
        notification_center.add_observer(self, sender=account_manager)
        notification_center.add_observer(self, sender=engine)
        notification_center.add_observer(self, sender=self.input)

        log.level.current = log.level.WARNING

    def run(self):
        account_manager = AccountManager()
        configuration = ConfigurationManager()
        engine = Engine()
        
        # start output thread
        self.output.start()
    
        # startup configuration
        Account.register_extension(AccountExtension)
        BonjourAccount.register_extension(BonjourAccountExtension)
        SIPSimpleSettings.register_extension(SIPSimpleSettingsExtension)
        SIPApplication.storage = FileStorage(config_directory)
        try:
            configuration.start()
        except ConfigurationError as e:
            raise RuntimeError("Failed to load sipclient's configuration: %s\nIf an old configuration file is in place, delete it or move it and recreate the configuration using the sip_settings script." % str(e))
        account_manager.load()
        if self.account_name is None:
            self.account = account_manager.default_account
        else:
            possible_accounts = [account for account in account_manager.iter_accounts() if self.account_name in account.id and account.enabled]
            if len(possible_accounts) > 1:
                raise RuntimeError("More than one account exists which matches %s: %s" % (self.account_name, ", ".join(sorted(account.id for account in possible_accounts))))
            if len(possible_accounts) == 0:
                raise RuntimeError("No enabled account that matches %s was found. Available and enabled accounts: %s" % (self.account_name, ", ".join(sorted(account.id for account in account_manager.get_accounts() if account.enabled))))
            self.account = possible_accounts[0]
        if self.account is None:
            raise RuntimeError("Unknown account %s. Available accounts: %s" % (self.account_name, ', '.join(account.id for account in account_manager.iter_accounts())))
        elif self.account == BonjourAccount():
            raise RuntimeError("Cannot use bonjour account for watcherinfo subscription")
        elif not self.account.presence.enabled:
            raise RuntimeError("Presence is not enabled for account %s" % self.account.id)
        elif not self.account.xcap.enabled:
            raise RuntimeError("XCAP is not enabled for account %s" % self.account.id)
        elif self.account.xcap.xcap_root is None:
            raise RuntimeError("XCAP root is not defined for account %s" % self.account.id)
        for account in account_manager.iter_accounts():
            if account == self.account:
                account.sip.register = False
            else:
                account.enabled = False
        self.output.put('Using account %s' % self.account.id)
        settings = SIPSimpleSettings()

        # start logging
        self.logger.start()

        # start the engine
        engine.start(
            auto_sound=False,
            events={'presence.winfo': [WatcherInfoDocument.content_type]},
            udp_port=settings.sip.udp_port if "udp" in settings.sip.transport_list else None,
            tcp_port=settings.sip.tcp_port if "tcp" in settings.sip.transport_list else None,
            tls_port=settings.sip.tls_port if "tls" in settings.sip.transport_list else None,
            tls_verify_server=settings.tls.verify_server,
            tls_ca_file=os.path.expanduser(settings.tls.ca_list) if settings.tls.ca_list else None,
            tls_cert_file=os.path.expanduser(settings.tls.certificate) if settings.tls.certificate else None,
            tls_privkey_file=os.path.expanduser(settings.tls.certificate) if settings.tls.certificate else None,
            user_agent=settings.user_agent,
            sample_rate=settings.audio.sample_rate,
            rtp_port_range=(settings.rtp.port_range.start, settings.rtp.port_range.end),
            trace_sip=settings.logs.trace_sip or self.logger.sip_to_stdout,
            log_level=settings.logs.pjsip_level if (settings.logs.trace_pjsip or self.logger.pjsip_to_stdout) else 0
        )

        self.output.put('Subscribing to the presence.winfo event')

        # start the input thread
        self.input.start()

        reactor.callLater(0, self._subscribe)

        # start twisted
        try:
            reactor.run()
        finally:
            self.input.stop()
        
        # stop the output
        self.output.stop()
        self.output.join()
        
        self.logger.stop()

        return 0 if self.success else 1

    def stop(self):
        self.stopping = True
        if self.subscription is not None and self.subscription.state.lower() in ('accepted', 'pending', 'active'):
            self.subscription.end(timeout=1)
        else:
            engine = Engine()
            engine.stop()

    def print_help(self):
        message  = 'Available control keys:\n'
        message += '  t: toggle SIP trace on the console\n'
        message += '  j: toggle PJSIP trace on the console\n'
        message += '  Ctrl-d: quit the program\n'
        message += '  ?: display this help message\n'
        self.output.put('\n'+message)
        
    def handle_notification(self, notification):
        handler = getattr(self, '_NH_%s' % notification.name, None)
        if handler is not None:
            handler(notification)

    def _NH_SIPSubscriptionDidStart(self, notification):
        route = Route(notification.sender.route_header.uri.host, notification.sender.route_header.uri.port, notification.sender.route_header.uri.parameters.get('transport', 'udp'))
        self._subscription_routes = None
        self._subscription_wait = 0.5
        self.output.put('Subscription succeeded at %s:%d;transport=%s' % (route.address, route.port, route.transport))
        self.success = True

    def _NH_SIPSubscriptionChangedState(self, notification):
        route = Route(notification.sender.route_header.uri.host, notification.sender.route_header.uri.port, notification.sender.route_header.uri.parameters.get('transport', 'udp'))
        if notification.data.state.lower() == "pending":
            self.output.put('Subscription pending at %s:%d;transport=%s' % (route.address, route.port, route.transport))
        elif notification.data.state.lower() == "active":
            self.output.put('Subscription active at %s:%d;transport=%s' % (route.address, route.port, route.transport))

    def _NH_SIPSubscriptionDidEnd(self, notification):
        notification_center = NotificationCenter()
        notification_center.remove_observer(self, sender=notification.sender)
        self.subscription = None
        route = Route(notification.sender.route_header.uri.host, notification.sender.route_header.uri.port, notification.sender.route_header.uri.parameters.get('transport', 'udp'))
        self.output.put('Unsubscribed from %s:%d;transport=%s' % (route.address, route.port, route.transport))
        self.stop()

    def _NH_SIPSubscriptionDidFail(self, notification):
        notification_center = NotificationCenter()
        notification_center.remove_observer(self, sender=notification.sender)
        self.subscription = None
        route = Route(notification.sender.route_header.uri.host, notification.sender.route_header.uri.port, notification.sender.route_header.uri.parameters.get('transport', 'udp'))
        if notification.data.code:
            status = ': %d %s' % (notification.data.code, notification.data.reason)
        else:
            status = ': %s' % notification.data.reason
        self.output.put('Subscription failed at %s:%d;transport=%s%s' % (route.address, route.port, route.transport, status))
        if self.stopping or notification.data.code in (401, 403, 407) or self.success:
            self.success = False
            self.stop()
        else:
            if not self._subscription_routes or time() > self._subscription_timeout:
                self._subscription_wait = min(self._subscription_wait*2, 30)
                timeout = random.uniform(self._subscription_wait, 2*self._subscription_wait)
                reactor.callFromThread(reactor.callLater, timeout, self._subscribe)
            else:
                route = self._subscription_routes.popleft()
                route_header = RouteHeader(route.uri)
                self.subscription = Subscription(self.account.uri,
                                                 FromHeader(self.account.uri, self.account.display_name),
                                                 ToHeader(self.account.uri, self.account.display_name),
                                                 ContactHeader(self.account.contact[route]),
                                                 b"presence.winfo",
                                                 route_header,
                                                 credentials=self.account.credentials,
                                                 refresh=self.account.sip.subscribe_interval)
                notification_center.add_observer(self, sender=self.subscription)
                self.subscription.subscribe(timeout=5)

    def _NH_SIPSubscriptionGotNotify(self, notification):
        if notification.data.content_type == WatcherInfoDocument.content_type:
            self._handle_winfo(notification.data.body)

    def _NH_DNSLookupDidSucceed(self, notification):
        # create subscription and register to get notifications from it
        self._subscription_routes = deque(notification.data.result)
        route = self._subscription_routes.popleft()
        route_header = RouteHeader(route.uri)
        self.subscription = Subscription(self.account.uri,
                                         FromHeader(self.account.uri, self.account.display_name),
                                         ToHeader(self.account.uri, self.account.display_name),
                                         ContactHeader(self.account.contact[route]),
                                         b"presence.winfo",
                                         route_header,
                                         credentials=self.account.credentials,
                                         refresh=self.account.sip.subscribe_interval)
        notification_center = NotificationCenter()
        notification_center.add_observer(self, sender=self.subscription)
        self.subscription.subscribe(timeout=5)

    def _NH_DNSLookupDidFail(self, notification):
        self.output.put('DNS lookup failed: %s' % notification.data.error)
        timeout = random.uniform(1.0, 2.0)
        reactor.callLater(timeout, self._subscribe)

    def _NH_SAInputWasReceived(self, notification):
        engine = Engine()
        settings = SIPSimpleSettings()
        key = notification.data.input
        if key == 't':
            self.logger.sip_to_stdout = not self.logger.sip_to_stdout
            engine.trace_sip = self.logger.sip_to_stdout or settings.logs.trace_sip
            self.output.put('SIP tracing to console is now %s.' % ('activated' if self.logger.sip_to_stdout else 'deactivated'))
        elif key == 'j':
            self.logger.pjsip_to_stdout = not self.logger.pjsip_to_stdout
            engine.log_level = settings.logs.pjsip_level if (self.logger.pjsip_to_stdout or settings.logs.trace_pjsip) else 0
            self.output.put('PJSIP tracing to console is now %s.' % ('activated' if self.logger.pjsip_to_stdout else 'deactivated'))
        elif key == 'n':
            self.logger.notifications_to_stdout = not self.logger.notifications_to_stdout
            self.output.put('Notification tracing to console is now %s.' % ('activated' if self.logger.notifications_to_stdout else 'deactivated'))
        elif key == '?':
            self.print_help()

    @run_in_twisted_thread
    def _NH_SIPEngineDidEnd(self, notification):
        self._stop_reactor()

    @run_in_twisted_thread
    def _NH_SIPEngineDidFail(self, notification):
        self.output.put('Engine failed.')
        self._stop_reactor()

    def _NH_SIPEngineGotException(self, notification):
        self.output.put('An exception occured within the SIP core:\n'+notification.data.traceback)

    def _stop_reactor(self):
        try:
            reactor.stop()
        except ReactorNotRunning:
            pass
    
    def _subscribe(self):
        settings = SIPSimpleSettings()
        
        self._subscription_timeout = time()+30

        lookup = DNSLookup()
        notification_center = NotificationCenter()
        notification_center.add_observer(self, sender=lookup)
        if self.account.sip.outbound_proxy is not None:
            uri = SIPURI(host=self.account.sip.outbound_proxy.host, port=self.account.sip.outbound_proxy.port, parameters={'transport': self.account.sip.outbound_proxy.transport})
        else:
            uri = SIPURI(host=self.account.id.domain)
        lookup.lookup_sip_proxy(uri, settings.sip.transport_list, tls_name=self.account.sip.tls_name)

    def _handle_winfo(self, body):
        try:
            watcher_info = WatcherInfoDocument.parse(body)
        except ParserError as e:
            self.output.put("Got illegal winfo document: %s\n%s" % (str(e), body))
        else:
            try:
                wlist = watcher_info['sip:' + self.account.id]
            except KeyError:
                self.output.put("Expected an entry for account %s in the winfo document" % self.account.id)
            else:
                buf = ["Received NOTIFY:", "----"]
                buf.append("Active watchers:")
                for watcher in wlist.active:
                    buf.append("  %s" % watcher)
                buf.append("Terminated watchers:")
                for watcher in wlist.terminated:
                    buf.append("  %s" % watcher)
                buf.append("Pending watchers:")
                for watcher in wlist.pending:
                    buf.append("  %s" % watcher)
                buf.append("Waiting watchers:")
                for watcher in wlist.waiting:
                    buf.append("  %s" % watcher)
                buf.append("----")
                self.output.put('\n'.join(buf))


if __name__ == "__main__":
    description = "This script subscribes to the presence.winfo event package and shows the received watcher info document's content. The program will un-SUBSCRIBE and quit when CTRL+D is pressed."
    usage = "%prog [options] [target-user@target-domain.com]"
    parser = OptionParser(usage=usage, description=description)
    parser.print_usage = parser.print_help
    parser.add_option("-a", "--account-name", type="string", dest="account_name", help="The name of the account to use.")
    parser.add_option("-s", "--trace-sip", action="store_true", dest="trace_sip", default=False, help="Dump the raw contents of incoming and outgoing SIP messages (disabled by default).")
    parser.add_option("-j", "--trace-pjsip", action="store_true", dest="trace_pjsip", default=False, help="Print PJSIP logging output (disabled by default).")
    parser.add_option("-n", "--trace-notifications", action="store_true", dest="trace_notifications", default=False, help="Print all notifications (disabled by default).")
    options, args = parser.parse_args()

    try:
        application = WinfoApplication(options.account_name, options.trace_sip, options.trace_pjsip, options.trace_notifications)
        return_code = application.run()
    except RuntimeError as e:
        print("Error: %s" % str(e))
        sys.exit(1)
    except SIPCoreError as e:
        print("Error: %s" % str(e))
        sys.exit(1)
    else:
        sys.exit(return_code)