diff --git a/sip-settings b/sip-settings index 22e3599..658082a 100755 --- a/sip-settings +++ b/sip-settings @@ -1,374 +1,373 @@ #!/usr/bin/env python3 import fcntl import re import struct import sys import termios from collections import deque from optparse import OptionParser from sipsimple.account import Account, BonjourAccount, AccountManager from sipsimple.application import SIPApplication from sipsimple.configuration import ConfigurationError, ConfigurationManager, DefaultValue, Setting, SettingsGroupMeta from sipsimple.configuration.datatypes import List, STUNServerAddress from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.storage import FileStorage from sipsimple.threading import ThreadManager from sipclient.configuration import config_directory from sipclient.configuration.account import AccountExtension from sipclient.configuration.settings import SIPSimpleSettingsExtension def format_child(obj, attrname, maxchars): linebuf = attrname if isinstance(getattr(type(obj), attrname, None), Setting): attr = getattr(obj, attrname) string = str(attr) if maxchars is not None: maxchars -= len(attrname)+4 if len(string) > maxchars: string = string[:maxchars-3]+'...' linebuf += ' = ' + string return linebuf def display_object(obj, name): # get terminal width if sys.stdout.isatty(): width = struct.unpack('HHHH', fcntl.ioctl(sys.stdout.fileno(), termios.TIOCGWINSZ, struct.pack('HHHH', 0, 0, 0, 0)))[1] else: width = None children = deque([child for child in dir(type(obj)) if isinstance(getattr(type(obj), child, None), Setting)] + \ [child for child in dir(type(obj)) if isinstance(getattr(type(obj), child, None), SettingsGroupMeta)]) # display first line linebuf = ' '*(len(name)+3) + '+' if children: linebuf += '-- ' + format_child(obj, children.popleft(), width-(len(name)+7) if width is not None else None) print(linebuf) # display second line linebuf = name + ' --|' if children: linebuf += '-- ' + format_child(obj, children.popleft(), width-(len(name)+7) if width is not None else None) print(linebuf) # display the rest of the lines if children: while children: child = children.popleft() linebuf = ' '*(len(name)+3) + ('|' if children else '+') + '-- ' + format_child(obj, child, width-(len(name)+7) if width is not None else None) print(linebuf) else: linebuf = ' '*(len(name)+3) + '+' print(linebuf) print() [display_object(getattr(obj, child), child) for child in dir(type(obj)) if isinstance(getattr(type(obj), child, None), SettingsGroupMeta)] class SettingsParser(object): @classmethod def parse_default(cls, type, value): if issubclass(type, List): values = re.split(r'\s*,\s*', value) return values elif issubclass(type, bool): if value.lower() == 'true': return True else: return False else: return value @classmethod def parse_MSRPRelayAddress(cls, type, value): return type.from_description(value) @classmethod def parse_SIPProxyAddress(cls, type, value): return type.from_description(value) @classmethod def parse_STUNServerAddress(cls, type, value): return type.from_description(value) @classmethod def parse_STUNServerAddressList(cls, type, value): values = re.split(r'\s*,\s*', value) return [STUNServerAddress.from_description(v) for v in values] @classmethod def parse_PortRange(cls, type, value): return type(*value.split(':', 1)) @classmethod def parse_Resolution(cls, type, value): return type(*value.split('x', 1)) @classmethod def parse_SoundFile(cls, type, value): if ',' in value: path, volume = value.split(',', 1) else: path, volume = value, 100 return type(path, volume) @classmethod def parse_AccountSoundFile(cls, type, value): if ',' in value: path, volume = value.split(',', 1) else: path, volume = value, 100 return type(path, volume) @classmethod def parse(cls, type, value): if value == 'None': return None if value == 'DEFAULT': return DefaultValue parser = getattr(cls, 'parse_%s' % type.__name__, cls.parse_default) return parser(type, value) class AccountConfigurator(object): def __init__(self): Account.register_extension(AccountExtension) BonjourAccount.register_extension(AccountExtension) self.configuration_manager = ConfigurationManager() self.configuration_manager.start() self.account_manager = AccountManager() self.account_manager.load() def list(self): print('Accounts:') bonjour_account = BonjourAccount() accounts = [account for account in self.account_manager.get_accounts() if account.id != bonjour_account.id] accounts.sort(key=lambda x: x.id) accounts.append(bonjour_account) for account in accounts: print(' %s (%s)%s' % (account.id, 'enabled' if account.enabled else 'disabled', ' - default_account' if account is self.account_manager.default_account else '')) def add(self, sip_address, password): if self.account_manager.has_account(sip_address): print('Account %s already exists' % sip_address) return try: account = Account(sip_address) except ValueError as e: print('Cannot add SIP account: %s' % str(e)) return account.auth.password = password account.enabled = True account.save() print('Account added') def delete(self, sip_address): if sip_address != 'ALL': possible_accounts = [account for account in self.account_manager.iter_accounts() if sip_address in account.id] if len(possible_accounts) > 1: print("More than one account exists which matches %s: %s" % (sip_address, ", ".join(sorted(account.id for account in possible_accounts)))) return if len(possible_accounts) == 0: print('Account %s does not exist' % sip_address) return account = possible_accounts[0] if account == BonjourAccount(): print('Cannot delete bonjour account') return account.delete() print('Account deleted') else: for account in self.account_manager.get_accounts(): account.delete() print('Accounts deleted') def show(self, sip_address=None): if sip_address is None: accounts = [self.account_manager.default_account] if accounts[0] is None: print("No accounts configured") return else: if sip_address != 'ALL': accounts = [account for account in self.account_manager.iter_accounts() if sip_address in account.id] else: accounts = self.account_manager.get_accounts() if not accounts: print('No accounts which match %s' % sip_address) return for account in accounts: print('Account %s:' % account.id) display_object(account, 'account') def set(self, *args): if not args: raise TypeError("set must receive at least one argument") if '=' in args[0]: accounts = [self.account_manager.default_account] if accounts[0] is None: print("No accounts configured") return else: sip_address = args[0] args = args[1:] if sip_address != 'ALL': accounts = [account for account in self.account_manager.iter_accounts() if sip_address in account.id] else: accounts = self.account_manager.get_accounts() if not accounts: print('No accounts which match %s' % sip_address) return try: settings = dict(arg.split('=', 1) for arg in args) except ValueError: print('Illegal arguments: %s' % ' '.join(args)) return for account in accounts: for attrname, value in settings.items(): object = account name = attrname while '.' in name: local_name, name = name.split('.', 1) - print(dir(account)) try: object = getattr(object, local_name) except AttributeError: print('Unknown setting: %s' % attrname) object = None break if object is not None: try: attribute = getattr(type(object), name) value = SettingsParser.parse(attribute.type, value) setattr(object, name, value) except AttributeError: print('Unknown setting: %s' % attrname) except ValueError as e: print('%s: %s' % (attrname, str(e))) account.save() print('Account%s updated' % ('s' if len(accounts) > 1 else '')) def default(self, sip_address): possible_accounts = [account for account in self.account_manager.iter_accounts() if sip_address in account.id] if len(possible_accounts) > 1: print("More than one account exists which matches %s: %s" % (sip_address, ", ".join(sorted(account.id for account in possible_accounts)))) return if len(possible_accounts) == 0: print('Account %s does not exist' % sip_address) return account = possible_accounts[0] try: self.account_manager.default_account = account except ValueError as e: print(str(e)) return print('Account %s is now default account' % account.id) class SIPSimpleConfigurator(object): def __init__(self): SIPSimpleSettings.register_extension(SIPSimpleSettingsExtension) self.configuration_manager = ConfigurationManager() self.configuration_manager.start() SIPSimpleSettings() def show(self): print('SIP SIMPLE settings:') display_object(SIPSimpleSettings(), 'SIP SIMPLE') def set(self, *args): sipsimple_settings = SIPSimpleSettings() try: settings = dict(arg.split('=', 1) for arg in args) except ValueError: print('Illegal arguments: %s' % ' '.join(args)) return for attrname, value in settings.items(): object = sipsimple_settings name = attrname while '.' in name: local_name, name = name.split('.', 1) try: object = getattr(object, local_name) except AttributeError: print('Unknown setting: %s' % attrname) object = None break if object is not None: try: attribute = getattr(type(object), name) value = SettingsParser.parse(attribute.type, value) setattr(object, name, value) except AttributeError: print('Unknown setting: %s' % attrname) except ValueError as e: print('%s: %s' % (attrname, str(e))) sipsimple_settings.save() print('SIP SIMPLE general settings updated') if __name__ == '__main__': description = "This script manages the SIP SIMPLE client SDK settings." usage = """%prog [--general|--account] [options] command [arguments] %prog --general show %prog --general set key1=value1 [key2=value2 ...] %prog --account list %prog --account add user@domain password %prog --account delete user@domain|ALL %prog --account show [user@domain|ALL] %prog --account set [user@domain|ALL] key1=value1|DEFAULT [key2=value2|DEFAULT ...] %prog --account default user@domain""" parser = OptionParser(usage=usage, description=description) parser.print_usage = parser.print_help parser.add_option('-c', '--config-directory', type='string', dest='config_directory', help='The configuration directory to use. This overrides the default location.') parser.add_option("-a", "--account", action="store_true", dest="account", help="Manage SIP accounts' settings") parser.add_option("-g", "--general", action="store_true", dest="general", help="Manage general SIP SIMPLE middleware settings") options, args = parser.parse_args() # exactly one of -a or -g must be specified if (not (options.account or options.general)) or (options.account and options.general): parser.print_usage() sys.exit(1) # there must be at least one command if not args: sys.stderr.write("Error: no command specified\n") parser.print_usage() sys.exit(1) SIPApplication.storage = FileStorage(options.config_directory or config_directory) thread_manager = ThreadManager() thread_manager.start() # execute the handlers try: if options.account: object = AccountConfigurator() else: object = SIPSimpleConfigurator() except ConfigurationError as e: sys.stderr.write("Failed to load sipclient's configuration: %s\n" % str(e)) sys.stderr.write("If an old configuration file is in place, delete it or move it and recreate the configuration using the sip_settings script.\n") else: command, args = args[0], args[1:] handler = getattr(object, command, None) if handler is None or not callable(handler): sys.stderr.write("Error: illegal command: %s\n" % command) parser.print_usage() sys.exit(1) try: handler(*args) except TypeError: sys.stderr.write("Error: illegal usage of command %s\n" % command) parser.print_usage() sys.exit(1) finally: thread_manager.stop() diff --git a/sip-subscribe-mwi b/sip-subscribe-mwi index ecb0ff9..cde2624 100755 --- a/sip-subscribe-mwi +++ b/sip-subscribe-mwi @@ -1,380 +1,380 @@ #!/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, Header, Route, RouteHeader, SIPCoreError, SIPURI, Subscription, ToHeader from sipsimple.lookup import DNSLookup from sipsimple.payloads.messagesummary import MessageSummary 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 SubscriptionApplication(object): def __init__(self, account_name, target, trace_sip, trace_pjsip, trace_notifications): self.account_name = account_name self.target = target 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(AccountExtension) 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 message summary subscription") elif not self.account.message_summary.enabled: raise RuntimeError("Message summary is not enabled 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={'message-summary': ['application/simple-message-summary']}, 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=self.account.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(self.account.tls.certificate) if self.account.tls.certificate else None, tls_privkey_file=os.path.expanduser(self.account.tls.certificate) if self.account.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 ) if self.target is None: self.target = ToHeader(SIPURI(user=self.account.id.username, host=self.account.id.domain)) else: if '@' not in self.target: self.target = '%s@%s' % (self.target, self.account.id.domain) if not self.target.startswith('sip:') and not self.target.startswith('sips:'): self.target = 'sip:' + self.target try: self.target = ToHeader(SIPURI.parse(self.target)) except SIPCoreError: self.output.put('Illegal SIP URI: %s' % self.target) engine.stop() return 1 self.output.put('Subscribing to %s for the message-summary event' % self.target.uri) # 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 += ' n: toggle notifications 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.target.uri, FromHeader(self.account.uri, self.account.display_name), self.target, ContactHeader(self.account.contact[route]), - "message-summary", + b"message-summary", route_header, credentials=self.account.credentials, refresh=self.account.sip.subscribe_interval) notification_center.add_observer(self, sender=self.subscription) self.subscription.subscribe(extra_headers=[Header('Supported', 'eventlist')], timeout=5) def _NH_SIPSubscriptionGotNotify(self, notification): if notification.data.body: ms = MessageSummary.parse(notification.data.body) self.output.put('\nReceived NOTIFY:\n' + ms.to_string()) self.print_help() 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.target.uri, FromHeader(self.account.uri, self.account.display_name), self.target, ContactHeader(self.account.contact[route]), - "message-summary", + b"message-summary", 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(extra_headers=[Header('Supported', 'eventlist')], 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}) elif self.account.sip.always_use_my_proxy: uri = SIPURI(host=self.account.id.domain) else: uri = self.target.uri lookup.lookup_sip_proxy(uri, settings.sip.transport_list) if __name__ == "__main__": description = "This script subscribes to the message summary event package for the specified SIP target. When a NOTIFY is received with the message summary information it will be displayed. 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 = SubscriptionApplication(options.account_name, args[0] if args else None, 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) diff --git a/sip-subscribe-presence b/sip-subscribe-presence index d1d86e2..e2cb027 100755 --- a/sip-subscribe-presence +++ b/sip-subscribe-presence @@ -1,625 +1,625 @@ #!/usr/bin/env python3 import datetime import os import random import select import sys import termios import urllib.request, urllib.parse, urllib.error 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.lookup import DNSLookup, DNSManager from sipsimple.configuration import ConfigurationError, ConfigurationManager from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.core import ContactHeader, Engine, FromHeader, RouteHeader, SIPCoreError, SIPURI, Subscription, ToHeader, Route, PJSIPError from sipsimple.payloads import ParserError from sipsimple.payloads import rpid # needed to register RPID extensions from sipsimple.payloads.pidf import Device, Person, Service, PIDF, PIDFDocument 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 SubscriptionApplication(object): def __init__(self, account_name, target, trace_sip, trace_pjsip, trace_notifications): self.account_name = account_name self.target = target self.input = InputThread(self) self.output = EventQueue(self._write) 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 _write(self, message): if isinstance(message, str): message = message.encode(sys.getfilesystemencoding()) sys.stdout.write(message.decode()+'\n') def run(self): account_manager = AccountManager() configuration = ConfigurationManager() self.engine = Engine() # start output thread self.output.start() # startup configuration Account.register_extension(AccountExtension) BonjourAccount.register_extension(AccountExtension) 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 presence subscription") elif not self.account.presence.enabled: raise RuntimeError("Presence is not enabled 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 self.engine.start( auto_sound=False, - events={b"presence": [PIDFDocument.content_type.encode()]}, + events={"presence": [PIDFDocument.content_type.encode()]}, 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=self.account.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(self.account.tls.certificate) if self.account.tls.certificate else None, tls_privkey_file=os.path.expanduser(self.account.tls.certificate) if self.account.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 ) if self.target is None: self.target = ToHeader(SIPURI(user=self.account.id.username, host=self.account.id.domain)) else: if '@' not in self.target: self.target = '%s@%s' % (self.target, self.account.id.domain) if not self.target.startswith('sip:') and not self.target.startswith('sips:'): self.target = 'sip:' + self.target try: self.target = ToHeader(SIPURI.parse(self.target)) except SIPCoreError: self.output.put('Illegal SIP URI: %s' % self.target) self.engine.stop() return 1 self.output.put('Subscribing to %s for the presence event' % self.target.uri) # 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: self.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 += ' n: toggle notifications 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_SIPEngineDidStart(self, notification): dns_manager = DNSManager() self.engine.set_nameservers(dns_manager.nameservers) 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.target.uri, FromHeader(self.account.uri, self.account.display_name), self.target, ContactHeader(self.account.contact[route]), b"presence", 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 == PIDFDocument.content_type: self.output.put('Received NOTIFY:') try: pidf = PIDF.parse(notification.data.body) except ParserError as e: self.output.put('Got illegal PIDF document: %s\n%s' % (str(e), notification.data.body)) else: self._display_pidf(pidf) self.print_help() 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.target.uri, FromHeader(self.account.uri, self.account.display_name), self.target, ContactHeader(self.account.contact[route]), b"presence", route_header, credentials=self.account.credentials, refresh=self.account.sip.subscribe_interval) notification_center = NotificationCenter() notification_center.add_observer(self, sender=self.subscription) try: self.subscription.subscribe(timeout=5) except PJSIPError as e: self.output.put('Subscription failed: %s' % str(e)) 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): settings = SIPSimpleSettings() key = notification.data.input if key == 't': self.logger.sip_to_stdout = not self.logger.sip_to_stdout self.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 self.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}) elif self.account.sip.always_use_my_proxy: uri = SIPURI(host=self.account.id.domain) else: uri = self.target.uri lookup.lookup_sip_proxy(uri, settings.sip.transport_list) def _format_note(self, note): text = "Note" if note.lang is not None: text += "(%s)" % note.lang text += ": %s" % note return text def _format_person(self, person, pidf): buf = [] # display class if person.rpid_class is not None: buf.append(" Class: %s" % person.rpid_class) # display timestamp if person.timestamp is not None: buf.append(" Timestamp: %s" % person.timestamp) # display notes if person.notes: for note in person.notes: buf.append(" %s" % self._format_note(note)) elif pidf.notes: for note in pidf.notes: buf.append(" %s" % self._format_note(note)) # display activities if person.activities is not None: activities = list(person.activities) if len(activities) > 0: text = " Activities" if person.activities.since is not None or person.activities.until is not None: text += " valid" if person.activities.since is not None: text += " from %s" % person.activities.since if person.activities.until is not None: text += " until %s" % person.activities.until text += ": %s" % ', '.join(str(activity) for activity in activities) buf.append(text) if len(person.activities.notes) > 0: for note in person.activities.notes: buf.append(" %s" % self._format_note(note)) elif len(person.activities.notes) > 0: buf.append(" Activities") for note in person.activities.notes: buf.append(" %s" % self._format_note(note)) # display mood if person.mood is not None: moods = list(person.mood) if len(moods) > 0: text = " Mood" if person.mood.since is not None or person.mood.until is not None: text += " valid" if person.mood.since is not None: text += " from %s" % person.mood.since if person.mood.until is not None: text += " until %s" % person.mood.until text += ": %s" % ', '.join(str(mood) for mood in moods) buf.append(text) if len(person.mood.notes) > 0: for note in person.mood.notes: buf.append(" %s" % self._format_note(note)) # display place is if person.place_is is not None: place_info = ', '.join('%s %s' % (key.capitalize(), getattr(person.place_is, key).value) for key in ('audio', 'video', 'text') if getattr(person.place_is, key) and getattr(person.place_is, key).value) if place_info != '': buf.append(" Place information: " + place_info) # display privacy if person.privacy is not None: text = " Private conversation possible with: " private = [] if person.privacy.audio: private.append("Audio") if person.privacy.video: private.append("Video") if person.privacy.text: private.append("Text") if len(private) > 0: text += ", ".join(private) else: text += "None" buf.append(text) # display sphere if person.sphere is not None: timeinfo = [] if person.sphere.since is not None: timeinfo.append('from %s' % str(person.sphere.since)) if person.sphere.until is not None: timeinfo.append('until %s' % str(person.sphere.until)) if len(timeinfo) != 0: timeinfo = ' (' + ', '.join(timeinfo) + ')' else: timeinfo = '' buf.append(" Current sphere%s: %s" % (timeinfo, person.sphere.value)) # display status icon if person.status_icon is not None: buf.append(" Status icon: %s" % person.status_icon) # display time and time offset if person.time_offset is not None: ctime = datetime.datetime.utcnow() + datetime.timedelta(minutes=int(person.time_offset)) time_offset = int(person.time_offset)/60.0 if time_offset == int(time_offset): offset_info = '(UTC+%d%s)' % (time_offset, (person.time_offset.description is not None and (' (%s)' % person.time_offset.description) or '')) else: offset_info = '(UTC+%.1f%s)' % (time_offset, (person.time_offset.description is not None and (' (%s)' % person.time_offset.description) or '')) buf.append(" Current user time: %s %s" % (ctime.strftime("%H:%M"), offset_info)) # display user input if person.user_input is not None: buf.append(" User is %s" % person.user_input) if person.user_input.last_input: buf.append(" Last input at: %s" % person.user_input.last_input) if person.user_input.idle_threshold: buf.append(" Idle threshold: %s seconds" % person.user_input.idle_threshold) return buf def _format_service(self, service, pidf): buf = [] # display class if service.rpid_class is not None: buf.append(" Class: %s" % service.rpid_class) # display timestamp if service.timestamp is not None: buf.append(" Timestamp: %s" % service.timestamp) # display notes for note in service.notes: buf.append(" %s" % self._format_note(note)) # display status if service.status is not None: if service.status.basic is not None: buf.append(" Basic status: %s" % service.status.basic) if service.status.extended is not None: buf.append(" Extended status: %s" % service.status.extended) # display contact if service.contact is not None: buf.append(" Contact%s: %s" % ((service.contact.priority is not None) and (' priority %s' % service.contact.priority) or '', urllib.parse.unquote(service.contact.value))) # display device ID if service.device_info is not None: description = " (%s)" % urllib.parse.unquote(service.device_info.description.value).decode('utf-8') if service.device_info.description else "" buf.append(" Service offered by device: %s%s" % (service.device_info.id, description)) # display relationship if service.relationship is not None: buf.append(" Relationship: %s" % service.relationship.value) # display service-class if service.service_class is not None: buf.append(" Service class: %s" % service.service_class.value) # display status icon if service.status_icon is not None: buf.append(" Status icon: %s" % service.status_icon) # display user input if service.user_input is not None: buf.append(" Service is %s" % service.user_input) if service.user_input.last_input: buf.append(" Last input at: %s" % service.user_input.last_input) if service.user_input.idle_threshold: buf.append(" Idle threshold: %s seconds" % service.user_input.idle_threshold) return buf def _format_device(self, device, pidf): buf = [] # display device ID if device.device_id is not None: buf.append(" Device id: %s" % device.device_id) # display class if device.rpid_class is not None: buf.append(" Class: %s" % device.rpid_class) # display timestamp if device.timestamp is not None: buf.append(" Timestamp: %s" % device.timestamp) # display notes for note in device.notes: buf.append(" %s" % self._format_note(note)) # display user input if device.user_input is not None: buf.append(" Device is %s" % device.user_input) if device.user_input.last_input: buf.append(" Last input at: %s" % device.user_input.last_input) if device.user_input.idle_threshold: buf.append(" Idle threshold: %s seconds" % device.user_input.idle_threshold) return buf def _display_pidf(self, pidf): buf = ["-"*16] buf.append("Presence for %s:" % urllib.parse.unquote(pidf.entity)) persons = {} devices = {} services = {} printed_sep = True for child in pidf: if isinstance(child, Person): persons[child.id] = child elif isinstance(child, Device): devices[child.id] = child elif isinstance(child, Service): services[child.id] = child # handle person information if len(persons) == 0: if list(pidf.notes): buf.append(" Person information:") for note in pidf.notes: buf.append(" %s" % self._format_note(note)) printed_sep = False else: for person in list(persons.values()): buf.append(" Person: %s" % person.id) buf.extend(self._format_person(person, pidf)) printed_sep = False # handle services informaation if len(services) > 0: if not printed_sep: buf.append(" " + "-"*3) for service in list(services.values()): buf.append(" Service: %s" % service.id) buf.extend(self._format_service(service, pidf)) # handle devices informaation if len(devices) > 0: if not printed_sep: buf.append(" " + "-"*3) for device in list(devices.values()): buf.append(" Device: %s" % device.id) buf.extend(self._format_device(device, pidf)) buf.append("-"*16) # push the data self.output.put('\n'.join(buf)) if __name__ == "__main__": description = "This script subscribes to the presence event package published by the specified SIP target. If a SIP target is not specified, it will subscribe to its own address. It will then interprete PIDF bodies contained in NOTIFYs and display their meaning. 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 = SubscriptionApplication(options.account_name, args[0] if args else None, 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) diff --git a/sip-subscribe-rls b/sip-subscribe-rls index 18e2a8e..e5d5bb1 100755 --- a/sip-subscribe-rls +++ b/sip-subscribe-rls @@ -1,376 +1,376 @@ #!/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 implements +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, Header, RouteHeader, SIPCoreError, SIPURI, Subscription, ToHeader, Route from sipsimple.lookup import DNSLookup 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 SubscriptionApplication(object): def __init__(self, account_name, target, trace_sip, trace_pjsip, trace_notifications): self.account_name = account_name self.target = target 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(AccountExtension) 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 presence subscription") elif not self.account.presence.enabled: raise RuntimeError("Presence is not enabled 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': ['multipart/related', 'application/rlmi+xml', 'application/pidf+xml']}, 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=self.account.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(self.account.tls.certificate) if self.account.tls.certificate else None, tls_privkey_file=os.path.expanduser(self.account.tls.certificate) if self.account.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 ) if self.target is None: self.target = ToHeader(SIPURI(user='%s+presence' % self.account.id.username, host=self.account.id.domain)) else: if '@' not in self.target: self.target = '%s@%s' % (self.target, self.account.id.domain) if not self.target.startswith('sip:') and not self.target.startswith('sips:'): self.target = 'sip:' + self.target try: self.target = ToHeader(SIPURI.parse(self.target)) except SIPCoreError: self.output.put('Illegal SIP URI: %s' % self.target) engine.stop() return 1 self.output.put('Subscribing to %s for the presence event' % self.target.uri) # 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 += ' n: toggle notifications 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.target.uri, FromHeader(self.account.uri, self.account.display_name), self.target, ContactHeader(self.account.contact[route]), - "presence", + b"presence", route_header, credentials=self.account.credentials, refresh=self.account.sip.subscribe_interval) notification_center.add_observer(self, sender=self.subscription) self.subscription.subscribe(extra_headers=[Header('Supported', 'eventlist')], timeout=5) def _NH_SIPSubscriptionGotNotify(self, notification): if notification.data.body: self.output.put('Received NOTIFY:\n' + notification.data.body) self.print_help() 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.target.uri, FromHeader(self.account.uri, self.account.display_name), self.target, ContactHeader(self.account.contact[route]), - "presence", + b"presence", 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(extra_headers=[Header('Supported', 'eventlist')], 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 = self.target.uri lookup.lookup_sip_proxy(uri, settings.sip.transport_list) if __name__ == "__main__": description = "This script subscribes to the presence event package published by the specified SIP target assuming it is a resource list handled by a RLS server. The RLS server will then SUBSCRIBE in behalf of the account, collect NOTIFYs with the presence information of the recipients and provide periodically aggregated NOTIFYs back to the subscriber. If a target address is not specified, it will subscribe to the address 'username-buddies@domain.com', where username and domain are taken from the account's SIP address. It will then interprete PIDF bodies contained in NOTIFYs and display their meaning. 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 = SubscriptionApplication(options.account_name, args[0] if args else None, 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) diff --git a/sip-subscribe-winfo b/sip-subscribe-winfo index 7ef9557..c541790 100755 --- a/sip-subscribe-winfo +++ b/sip-subscribe-winfo @@ -1,393 +1,393 @@ #!/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(AccountExtension) 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=self.account.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(self.account.tls.certificate) if self.account.tls.certificate else None, tls_privkey_file=os.path.expanduser(self.account.tls.certificate) if self.account.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]), - "presence.winfo", + 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]), - "presence.winfo", + 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) 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) diff --git a/sip-subscribe-xcap-diff b/sip-subscribe-xcap-diff index 0cbab29..f4faafa 100755 --- a/sip-subscribe-xcap-diff +++ b/sip-subscribe-xcap-diff @@ -1,411 +1,411 @@ #!/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.xcapdiff import XCAPDiffDocument, Document, Element, Attribute from sipsimple.payloads.resourcelists import ResourceListsDocument, ResourceLists, List, Entry 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 SubscriptionApplication(object): def __init__(self, account_name, trace_sip, trace_pjsip, trace_notifications): self.account_name = account_name self.target = None 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.body = None self.content_type = None 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(AccountExtension) 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 not self.account.enabled: raise RuntimeError("Account %s is not enabled" % self.account.id) elif self.account == BonjourAccount(): raise RuntimeError("Cannot use bonjour account for presence subscription") 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() # generate the body list = List() resource_lists = ResourceLists([list]) list.add(Entry('resource-lists/users/sip:%s/index' % self.account.id)) list.add(Entry('rls-services/users/sip:%s/index' % self.account.id)) list.add(Entry('pres-rules/users/sip:%s/index' % self.account.id)) self.body = resource_lists.toxml(pretty_print=True) self.content_type = ResourceListsDocument.content_type # start logging self.logger.start() # start the engine engine.start( auto_sound=False, events={'xcap-diff': ['application/xcap-diff+xml']}, 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=self.account.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(self.account.tls.certificate) if self.account.tls.certificate else None, tls_privkey_file=os.path.expanduser(self.account.tls.certificate) if self.account.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.target = ToHeader(SIPURI(user=self.account.id.username, host=self.account.id.domain)) self.output.put('Subscribing to %s for the xcap-diff event' % self.target.uri) # 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 += ' n: toggle notifications 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, 489) 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.target.uri, FromHeader(self.account.uri, self.account.display_name), self.target, ContactHeader(self.account.contact[route]), - "xcap-diff", + b"xcap-diff", route_header, credentials=self.account.credentials, refresh=self.account.sip.subscribe_interval) notification_center.add_observer(self, sender=self.subscription) - self.subscription.subscribe(body=self.body, content_type=self.content_type, timeout=5) + self.subscription.subscribe(body=self.body.encode(), content_type=self.content_type, timeout=5) def _NH_SIPSubscriptionGotNotify(self, notification): if notification.data.content_type == XCAPDiffDocument.content_type: try: xcap_diff = XCAPDiffDocument.parse(notification.data.body) except ParserError as e: self.output.put("xcap-diff document is invalid: %s" % str(e)) else: self._display_xcapdiff(xcap_diff) self.print_help() 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.target.uri, FromHeader(self.account.uri, self.account.display_name), self.target, ContactHeader(self.account.contact[route]), - "xcap-diff", + b"xcap-diff", 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(body=self.body, content_type=self.content_type, timeout=5) + self.subscription.subscribe(body=self.body.encode(), content_type=self.content_type, 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 = self.target.uri lookup.lookup_sip_proxy(uri, settings.sip.transport_list) def _display_xcapdiff(self, xcap_diff): message = [] message.append('XCAP diff for XCAP root %s' % xcap_diff.xcap_root) for child in xcap_diff: if isinstance(child, Document): message.append(' %s document %s for AUID %s changed' % ('Global' if child.selector.globaltree is not None else "User's %s" % child.selector.userstree, child.selector.document, child.selector.auid)) message.append(' URL: %s' % child.selector) if child.previous_etag: message.append(' Previous ETag: %s' % child.previous_etag) if child.new_etag: message.append(' New ETag: %s' % child.new_etag) if child.empty_body: message.append(' Body did not change') elif isinstance(child, Element): message.append(' %s element %s in document %s for AUID %s changed' % ('Global' if child.selector.globaltree is not None else "User's %s" % child.selector.userstree, child.selector.node, child.selector.document, child.selector.auid)) message.append(' URL: %s/%s' % (xcap_diff.xcap_root, child.selector)) elif isinstance(child, Attribute): message.append(' %s attribute %s in document %s for AUID %s changed' % ('Global' if child.selector.globaltree is not None else "User's %s" % child.selector.userstree, child.selector.node, child.selector.document, child.selector.auid)) message.append(' URL: %s/%s' % (xcap_diff.xcap_root, child.selector)) if child.value: message.append(' New value: %s' % child.value) self.output.put('\n'.join(message)) if __name__ == "__main__": description = "This script subscribes to the xcap-diff event package for the given SIP account. The program will un-SUBSCRIBE and quit when CTRL+D is pressed." usage = "%prog [options]" 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 = SubscriptionApplication(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)