diff --git a/sipsimple/account/__init__.py b/sipsimple/account/__init__.py index a6934ac4..8c5f082d 100644 --- a/sipsimple/account/__init__.py +++ b/sipsimple/account/__init__.py @@ -1,888 +1,889 @@ """ Implements a SIP Account management system that allows the definition of multiple SIP accounts and their properties. """ __all__ = ['Account', 'BonjourAccount', 'AccountManager'] from itertools import chain from threading import Lock from application.notification import IObserver, NotificationCenter, NotificationData from application.python import Null from application.python.descriptor import classproperty from application.python.types import Singleton from application.system import host as Host from eventlib import coros, proc from gnutls.crypto import X509Certificate, X509PrivateKey from gnutls.errors import GNUTLSError from gnutls.interfaces.twisted import X509Credentials from zope.interface import implementer from sipsimple.account.bonjour import BonjourServices, _bonjour from sipsimple.account.publication import PresencePublisher, DialogPublisher from sipsimple.account.registration import Registrar from sipsimple.account.subscription import MWISubscriber, PresenceWinfoSubscriber, DialogWinfoSubscriber, PresenceSubscriber, SelfPresenceSubscriber, DialogSubscriber from sipsimple.account.xcap import XCAPManager from sipsimple.core import Credentials, SIPURI, ContactURIFactory from sipsimple.configuration import ConfigurationManager, Setting, SettingsGroup, SettingsObject, SettingsObjectID from sipsimple.configuration.datatypes import AudioCodecList, MSRPConnectionModel, MSRPRelayAddress, MSRPTransport, NonNegativeInteger, Path, SIPAddress, SIPProxyAddress, SRTPKeyNegotiation, STUNServerAddressList, VideoCodecList, XCAPRoot from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.payloads import ParserError from sipsimple.payloads.messagesummary import MessageSummary from sipsimple.payloads.pidf import PIDFDocument from sipsimple.payloads.rlsnotify import RLSNotify from sipsimple.payloads.watcherinfo import WatcherInfoDocument from sipsimple.threading import call_in_thread from sipsimple.threading.green import call_in_green_thread, run_in_green_thread from sipsimple.util import user_info, execute_once class AuthSettings(SettingsGroup): username = Setting(type=str, default=None, nillable=True) password = Setting(type=str, default='') class SIPSettings(SettingsGroup): always_use_my_proxy = Setting(type=bool, default=True) outbound_proxy = Setting(type=SIPProxyAddress, default=None, nillable=True) register = Setting(type=bool, default=True) register_interval = Setting(type=NonNegativeInteger, default=600) subscribe_interval = Setting(type=NonNegativeInteger, default=600) publish_interval = Setting(type=NonNegativeInteger, default=600) + tls_name = Setting(type=str, default=None, nillable=True) class SRTPEncryptionSettings(SettingsGroup): enabled = Setting(type=bool, default=True) key_negotiation = Setting(type=SRTPKeyNegotiation, default='opportunistic') class RTPSettings(SettingsGroup): audio_codec_list = Setting(type=AudioCodecList, default=None, nillable=True) video_codec_list = Setting(type=VideoCodecList, default=None, nillable=True) encryption = SRTPEncryptionSettings class NATTraversalSettings(SettingsGroup): use_ice = Setting(type=bool, default=False) stun_server_list = Setting(type=STUNServerAddressList, default=None, nillable=True) msrp_relay = Setting(type=MSRPRelayAddress, default=None, nillable=True) use_msrp_relay_for_outbound = Setting(type=bool, default=False) class MessageSummarySettings(SettingsGroup): enabled = Setting(type=bool, default=False) voicemail_uri = Setting(type=SIPAddress, default=None, nillable=True) class XCAPSettings(SettingsGroup): enabled = Setting(type=bool, default=False) discovered = Setting(type=bool, default=False) xcap_root = Setting(type=XCAPRoot, default=None, nillable=True) xcap_diff = Setting(type=bool, default=True) class PresenceSettings(SettingsGroup): enabled = Setting(type=bool, default=False) class TLSSettings(SettingsGroup): ca_list = Setting(type=Path, default=None, nillable=True) certificate = Setting(type=Path, default=None, nillable=True) verify_server = Setting(type=bool, default=False) class MSRPSettings(SettingsGroup): transport = Setting(type=MSRPTransport, default='tls') connection_model = Setting(type=MSRPConnectionModel, default='relay') @implementer(IObserver) class Account(SettingsObject): """ Object representing a SIP account. Contains configuration settings and attributes for accessing SIP related objects. When the account is active, it will register, publish its presence and subscribe to watcher-info events depending on its settings. If the object is un-pickled and its enabled flag was set, it will automatically activate. When the save method is called, depending on the value of the enabled flag, the account will activate/deactivate. Notifications sent by instances of Account: * CFGSettingsObjectWasCreated * CFGSettingsObjectWasActivated * CFGSettingsObjectWasDeleted * CFGSettingsObjectDidChange * SIPAccountWillActivate * SIPAccountDidActivate * SIPAccountWillDeactivate * SIPAccountDidDeactivate """ __group__ = 'Accounts' __id__ = SettingsObjectID(type=SIPAddress) id = __id__ enabled = Setting(type=bool, default=False) display_name = Setting(type=str, default=None, nillable=True) auth = AuthSettings sip = SIPSettings rtp = RTPSettings nat_traversal = NATTraversalSettings message_summary = MessageSummarySettings msrp = MSRPSettings presence = PresenceSettings xcap = XCAPSettings tls = TLSSettings def __new__(cls, id): #with AccountManager.load.lock: # if not AccountManager.load.called: # raise RuntimeError("cannot instantiate %s before calling AccountManager.load" % cls.__name__) return SettingsObject.__new__(cls, id) def __init__(self, id): self.contact = ContactURIFactory() self.xcap_manager = XCAPManager(self) self._started = False self._deleted = False self._active = False self._activation_lock = coros.Semaphore(1) self._registrar = Registrar(self) self._mwi_subscriber = MWISubscriber(self) self._pwi_subscriber = PresenceWinfoSubscriber(self) self._dwi_subscriber = DialogWinfoSubscriber(self) self._presence_subscriber = PresenceSubscriber(self) self._self_presence_subscriber = SelfPresenceSubscriber(self) self._dialog_subscriber = DialogSubscriber(self) self._presence_publisher = PresencePublisher(self) self._dialog_publisher = DialogPublisher(self) self._mwi_voicemail_uri = None self._pwi_version = None self._dwi_version = None self._presence_version = None self._dialog_version = None self.trusted_cas = [] self.ca_list = None def start(self): if self._started or self._deleted: return self._started = True notification_center = NotificationCenter() notification_center.add_observer(self, name='CFGSettingsObjectDidChange', sender=self) notification_center.add_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) notification_center.add_observer(self, name='XCAPManagerDidDiscoverServerCapabilities', sender=self.xcap_manager) notification_center.add_observer(self, sender=self._mwi_subscriber) notification_center.add_observer(self, sender=self._pwi_subscriber) notification_center.add_observer(self, sender=self._dwi_subscriber) notification_center.add_observer(self, sender=self._presence_subscriber) notification_center.add_observer(self, sender=self._self_presence_subscriber) notification_center.add_observer(self, sender=self._dialog_subscriber) self.xcap_manager.init() if self.enabled: self._activate() def stop(self): if not self._started: return self._started = False self._deactivate() notification_center = NotificationCenter() notification_center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=self) notification_center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) notification_center.remove_observer(self, name='XCAPManagerDidDiscoverServerCapabilities', sender=self.xcap_manager) notification_center.remove_observer(self, sender=self._mwi_subscriber) notification_center.remove_observer(self, sender=self._pwi_subscriber) notification_center.remove_observer(self, sender=self._dwi_subscriber) notification_center.remove_observer(self, sender=self._presence_subscriber) notification_center.remove_observer(self, sender=self._self_presence_subscriber) notification_center.remove_observer(self, sender=self._dialog_subscriber) def delete(self): if self._deleted: return self._deleted = True self.stop() self._registrar = None self._mwi_subscriber = None self._pwi_subscriber = None self._dwi_subscriber = None self._presence_subscriber = None self._self_presence_subscriber = None self._dialog_subscriber = None self._presence_publisher = None self._dialog_publisher = None self.xcap_manager = None SettingsObject.delete(self) @run_in_green_thread def reregister(self): if self._started: self._registrar.reregister() @run_in_green_thread def resubscribe(self): if self._started: self._mwi_subscriber.resubscribe() self._pwi_subscriber.resubscribe() self._dwi_subscriber.resubscribe() self._presence_subscriber.resubscribe() self._self_presence_subscriber.resubscribe() self._dialog_subscriber.resubscribe() @property def credentials(self): username = self.auth.username or self.id.username username = username.encode() if username else None password = self.auth.password.encode() if self.auth.password else None return Credentials(username, password) @property def registered(self): try: return self._registrar.registered except AttributeError: return False @property def mwi_active(self): try: return self._mwi_subscriber.subscribed except AttributeError: return False @property def tls_credentials(self): # This property can be optimized to cache the credentials it loads from disk, # however this is not a time consuming operation (~ 3000 req/sec). -Luci settings = SIPSimpleSettings() tls_certificate = self.tls.certificate or settings.tls.certificate certificate = None private_key = None if tls_certificate is not None: try: certificate_data = open(tls_certificate.normalized).read() certificate = X509Certificate(certificate_data) private_key = X509PrivateKey(certificate_data) except (FileNotFoundError, GNUTLSError, UnicodeDecodeError): pass trusted_cas = [] ca_list = self.tls.ca_list or settings.tls.ca_list if ca_list is not None: if len(self.trusted_cas) > 0: trusted_cas = self.trusted_cas else: crt = None start = False try: ca_text = open(ca_list.normalized).read() except (FileNotFoundError, GNUTLSError, UnicodeDecodeError): ca_text = '' for line in ca_text.split("\n"): if "BEGIN CERT" in line: start = True crt = line + "\n" elif "END CERT" in line: crt = crt + line + "\n" end = True start = False try: trusted_cas.append(X509Certificate(crt)) except (GNUTLSError, ValueError) as e: continue elif start: crt = crt + line + "\n" self.trusted_cas = trusted_cas self.ca_list = ca_list credentials = X509Credentials(certificate, private_key, trusted_cas) credentials.verify_peer = self.tls.verify_server or settings.tls.certificate return credentials @property def uri(self): return SIPURI(user=self.id.username, host=self.id.domain) @property def voicemail_uri(self): return self._mwi_voicemail_uri or self.message_summary.voicemail_uri @property def presence_state(self): try: return self._presence_publisher.state except AttributeError: return None @presence_state.setter def presence_state(self, state): try: self._presence_publisher.state = state except AttributeError: pass @property def dialog_state(self): try: return self._dialog_publisher.state except AttributeError: return None @dialog_state.setter def dialog_state(self, state): try: self._dialog_publisher.state = state except AttributeError: pass def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) @run_in_green_thread def _NH_CFGSettingsObjectDidChange(self, notification): if self._started and 'enabled' in notification.data.modified: if self.enabled: self._activate() else: self._deactivate() def _NH_XCAPManagerDidDiscoverServerCapabilities(self, notification): if self._started and self.xcap.discovered is False: self.xcap.discovered = True self.save() notification.center.post_notification('SIPAccountDidDiscoverXCAPSupport', sender=self) def _NH_MWISubscriberDidDeactivate(self, notification): self._mwi_voicemail_uri = None def _NH_MWISubscriptionGotNotify(self, notification): body = notification.data.body.decode() if notification.data.body else None if body and notification.data.content_type == MessageSummary.content_type: try: message_summary = MessageSummary.parse(body) except ParserError: pass else: self._mwi_voicemail_uri = message_summary.message_account and SIPAddress(message_summary.message_account.replace('sip:', '', 1)) or None notification.center.post_notification('SIPAccountGotMessageSummary', sender=self, data=NotificationData(message_summary=message_summary)) def _NH_PresenceWinfoSubscriptionGotNotify(self, notification): body = notification.data.body.decode() if notification.data.body else None if body and notification.data.content_type == WatcherInfoDocument.content_type: try: watcher_info = WatcherInfoDocument.parse(body) watcher_list = watcher_info['sip:' + self.id] except (ParserError, KeyError): pass else: if watcher_list.package != 'presence': return if self._pwi_version is None: if watcher_info.state == 'partial': self._pwi_subscriber.resubscribe() elif watcher_info.version <= self._pwi_version: return elif watcher_info.state == 'partial' and watcher_info.version > self._pwi_version + 1: self._pwi_subscriber.resubscribe() self._pwi_version = watcher_info.version data = NotificationData(version=watcher_info.version, state=watcher_info.state, watcher_list=watcher_list) notification.center.post_notification('SIPAccountGotPresenceWinfo', sender=self, data=data) def _NH_PresenceWinfoSubscriptionDidEnd(self, notification): self._pwi_version = None def _NH_PresenceWinfoSubscriptionDidFail(self, notification): self._pwi_version = None def _NH_DialogWinfoSubscriptionGotNotify(self, notification): body = notification.data.body.decode() if notification.data.body else None if body and notification.data.content_type == WatcherInfoDocument.content_type: try: watcher_info = WatcherInfoDocument.parse(body) watcher_list = watcher_info['sip:' + self.id] except (ParserError, KeyError): pass else: if watcher_list.package != 'dialog': return if self._dwi_version is None: if watcher_info.state == 'partial': self._dwi_subscriber.resubscribe() elif watcher_info.version <= self._dwi_version: return elif watcher_info.state == 'partial' and watcher_info.version > self._dwi_version + 1: self._dwi_subscriber.resubscribe() self._dwi_version = watcher_info.version data = NotificationData(version=watcher_info.version, state=watcher_info.state, watcher_list=watcher_list) notification.center.post_notification('SIPAccountGotDialogWinfo', sender=self, data=data) def _NH_DialogWinfoSubscriptionDidEnd(self, notification): self._dwi_version = None def _NH_DialogWinfoSubscriptionDidFail(self, notification): self._dwi_version = None def _NH_PresenceSubscriptionGotNotify(self, notification): body = notification.data.body.decode() if notification.data.body else None if body and notification.data.content_type == RLSNotify.content_type: try: rls_notify = RLSNotify.parse('{content_type}\r\n\r\n{body}'.format(content_type=notification.data.headers['Content-Type'], body=body)) except ParserError: pass else: if rls_notify.uri != self.xcap_manager.rls_presence_uri: return if self._presence_version is None: if not rls_notify.full_state: self._presence_subscriber.resubscribe() elif rls_notify.version <= self._presence_version: return elif not rls_notify.full_state and rls_notify.version > self._presence_version + 1: self._presence_subscriber.resubscribe() self._presence_version = rls_notify.version data = NotificationData(version=rls_notify.version, full_state=rls_notify.full_state, resource_map=dict((str(resource.uri), resource) for resource in rls_notify)) notification.center.post_notification('SIPAccountGotPresenceState', sender=self, data=data) def _NH_PresenceSubscriptionDidEnd(self, notification): self._presence_version = None def _NH_PresenceSubscriptionDidFail(self, notification): self._presence_version = None def _NH_SelfPresenceSubscriptionGotNotify(self, notification): body = notification.data.body.decode() if notification.data.body else None if body and notification.data.content_type == PIDFDocument.content_type: try: pidf_doc = PIDFDocument.parse(body) except ParserError: pass else: if pidf_doc.entity.partition('sip:')[2] != self.id: return notification.center.post_notification('SIPAccountGotSelfPresenceState', sender=self, data=NotificationData(pidf=pidf_doc)) def _NH_DialogSubscriptionGotNotify(self, notification): body = notification.data.body.decode() if notification.data.body else None if body and notification.data.content_type == RLSNotify.content_type: try: rls_notify = RLSNotify.parse('{content_type}\r\n\r\n{body}'.format(content_type=notification.data.headers['Content-Type'], body=body)) except ParserError: pass else: if rls_notify.uri != self.xcap_manager.rls_dialog_uri: return if self._dialog_version is None: if not rls_notify.full_state: self._dialog_subscriber.resubscribe() elif rls_notify.version <= self._dialog_version: return elif not rls_notify.full_state and rls_notify.version > self._dialog_version + 1: self._dialog_subscriber.resubscribe() self._dialog_version = rls_notify.version data = NotificationData(version=rls_notify.version, full_state=rls_notify.full_state, resource_map=dict((resource.uri, resource) for resource in rls_notify)) notification.center.post_notification('SIPAccountGotDialogState', sender=self, data=data) def _NH_DialogSubscriptionDidEnd(self, notification): self._dialog_version = None def _NH_DialogSubscriptionDidFail(self, notification): self._dialog_version = None def _activate(self): with self._activation_lock: if self._active: return notification_center = NotificationCenter() notification_center.post_notification('SIPAccountWillActivate', sender=self) self._active = True self._registrar.start() self._mwi_subscriber.start() self._pwi_subscriber.start() self._dwi_subscriber.start() self._presence_subscriber.start() self._self_presence_subscriber.start() self._dialog_subscriber.start() self._presence_publisher.start() self._dialog_publisher.start() if self.xcap.enabled: self.xcap_manager.start() notification_center.post_notification('SIPAccountDidActivate', sender=self) def _deactivate(self): with self._activation_lock: if not self._active: return notification_center = NotificationCenter() notification_center.post_notification('SIPAccountWillDeactivate', sender=self) self._active = False handlers = [self._registrar, self._mwi_subscriber, self._pwi_subscriber, self._dwi_subscriber, self._presence_subscriber, self._self_presence_subscriber, self._dialog_subscriber, self._presence_publisher, self._dialog_publisher, self.xcap_manager] proc.waitall([proc.spawn(handler.stop) for handler in handlers]) notification_center.post_notification('SIPAccountDidDeactivate', sender=self) def __repr__(self): return '%s(%r)' % (self.__class__.__name__, self.id) def __setstate__(self, data): # This restores the password from its previous location as a top level setting # after it was moved under the auth group. SettingsObject.__setstate__(self, data) if not data.get('auth', {}).get('password') and data.get('password'): self.auth.password = data.pop('password') self.save() class BonjourMSRPSettings(SettingsGroup): transport = Setting(type=MSRPTransport, default='tls') class BonjourAccountEnabledSetting(Setting): def __get__(self, obj, objtype): if obj is None: return self return _bonjour.available and self.values.get(obj, self.default) def __set__(self, obj, value): if not _bonjour.available: raise RuntimeError('mdns support is not available') Setting.__set__(self, obj, value) @implementer(IObserver) class BonjourAccount(SettingsObject): """ Object representing a bonjour account. Contains configuration settings and attributes for accessing bonjour related options. When the account is active, it will send broadcast its contact address on the LAN. If the object is un-pickled and its enabled flag was set, it will automatically activate. When the save method is called, depending on the value of the enabled flag, the account will activate/deactivate. Notifications sent by instances of Account: * CFGSettingsObjectWasCreated * CFGSettingsObjectWasActivated * CFGSettingsObjectWasDeleted * CFGSettingsObjectDidChange * SIPAccountWillActivate * SIPAccountDidActivate * SIPAccountWillDeactivate * SIPAccountDidDeactivate """ __group__ = 'Accounts' __id__ = SIPAddress('bonjour@local') id = property(lambda self: self.__id__) enabled = BonjourAccountEnabledSetting(type=bool, default=True) display_name = Setting(type=str, default=user_info.fullname, nillable=False) msrp = BonjourMSRPSettings presence = PresenceSettings rtp = RTPSettings tls = TLSSettings def __new__(cls): # with AccountManager.load.lock: # if not AccountManager.load.called: # raise RuntimeError("cannot instantiate %s before calling AccountManager.load" % cls.__name__) return SettingsObject.__new__(cls) def __init__(self): self.contact = ContactURIFactory() self.credentials = None self._started = False self._active = False self._activation_lock = coros.Semaphore(1) self._bonjour_services = BonjourServices(self) # initialize fake settings (these are here to make the bonjour account quack like a duck) self.nat_traversal = NATTraversalSettings() self.nat_traversal.use_ice = False self.nat_traversal.msrp_relay = None self.nat_traversal.use_msrp_relay_for_outbound = False self.xcap = XCAPSettings() self.xcap.enabled = False self.xcap.discovered = False self.xcap.xcap_root = None def __repr__(self): return '%s()' % self.__class__.__name__ def start(self): if self._started: return self._started = True notification_center = NotificationCenter() notification_center.add_observer(self, name='CFGSettingsObjectDidChange', sender=self) notification_center.add_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) self._bonjour_services.start() if self.enabled: self._activate() def stop(self): if not self._started: return self._started = False self._deactivate() self._bonjour_services.stop() notification_center = NotificationCenter() notification_center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=self) notification_center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) @classproperty def mdns_available(cls): return _bonjour.available @property def registered(self): return False @property def tls_credentials(self): # This property can be optimized to cache the credentials it loads from disk, # however this is not a time consuming operation (~ 3000 req/sec). -Luci settings = SIPSimpleSettings() tls_certificate = self.tls.certificate or settings.tls.certificate if tls_certificate is not None: certificate_data = open(tls_certificate.normalized).read() certificate = X509Certificate(certificate_data) private_key = X509PrivateKey(certificate_data) else: certificate = None private_key = None credentials = X509Credentials(certificate, private_key, []) credentials.verify_peer = False return credentials @property def uri(self): return SIPURI(user=self.contact.username, host=Host.default_ip or '127.0.0.1') @property def presence_state(self): return self._bonjour_services.presence_state @presence_state.setter def presence_state(self, state): self._bonjour_services.presence_state = state def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) @run_in_green_thread def _NH_CFGSettingsObjectDidChange(self, notification): if self._started: if 'enabled' in notification.data.modified: if self.enabled: self._activate() else: self._deactivate() elif self.enabled: if 'display_name' in notification.data.modified: self._bonjour_services.update_registrations() if {'sip.transport_list', 'tls.certificate'}.intersection(notification.data.modified): self._bonjour_services.update_registrations() self._bonjour_services.restart_discovery() def _activate(self): with self._activation_lock: if self._active: return notification_center = NotificationCenter() notification_center.post_notification('SIPAccountWillActivate', sender=self) self._active = True self._bonjour_services.activate() notification_center.post_notification('SIPAccountDidActivate', sender=self) def _deactivate(self): with self._activation_lock: if not self._active: return notification_center = NotificationCenter() notification_center.post_notification('SIPAccountWillDeactivate', sender=self) self._active = False self._bonjour_services.deactivate() notification_center.post_notification('SIPAccountDidDeactivate', sender=self) @implementer(IObserver) class AccountManager(object, metaclass=Singleton): """ This is a singleton object which manages all the SIP accounts. It is also used to manage the default account (the one used for outbound sessions) using the default_account attribute: manager = AccountManager() manager.default_account = manager.get_account('alice@example.net') The following notifications are sent: * SIPAccountManagerDidRemoveAccount * SIPAccountManagerDidAddAccount * SIPAccountManagerDidChangeDefaultAccount """ def __init__(self): self._lock = Lock() self.accounts = {} notification_center = NotificationCenter() notification_center.add_observer(self, name='CFGSettingsObjectWasActivated') notification_center.add_observer(self, name='CFGSettingsObjectWasCreated') @execute_once def load(self): """ Load all accounts from the configuration. The accounts will not be started until the start method is called. """ configuration = ConfigurationManager() bonjour_account = BonjourAccount() names = configuration.get_names([Account.__group__]) [Account(id) for id in names if id != bonjour_account.id] default_account = self.default_account if default_account is None or not default_account.enabled: try: self.default_account = next((account for account in list(self.accounts.values()) if account.enabled)) except StopIteration: self.default_account = None def start(self): """ Start the accounts, which will determine the ones with the enabled flag set to activate. """ notification_center = NotificationCenter() notification_center.post_notification('SIPAccountManagerWillStart', sender=self) proc.waitall([proc.spawn(account.start) for account in list(self.accounts.values())]) notification_center.post_notification('SIPAccountManagerDidStart', sender=self) def stop(self): """ Stop the accounts, which will determine the ones that were enabled to deactivate. This method returns only once the accounts were stopped successfully or they timed out trying. """ notification_center = NotificationCenter() notification_center.post_notification('SIPAccountManagerWillEnd', sender=self) proc.waitall([proc.spawn(account.stop) for account in list(self.accounts.values())]) notification_center.post_notification('SIPAccountManagerDidEnd', sender=self) def has_account(self, id): return id in self.accounts def get_account(self, id): return self.accounts[id] def get_accounts(self): return list(self.accounts.values()) def iter_accounts(self): return iter(list(self.accounts.values())) def find_account(self, contact_uri): # compare contact_address with account contact exact_matches = (account for account in list(self.accounts.values()) if account.enabled and account.contact.username==contact_uri.user) # compare username in contact URI with account username loose_matches = (account for account in list(self.accounts.values()) if account.enabled and account.id.username==contact_uri.user) return next(chain(exact_matches, loose_matches, [None])) def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_CFGSettingsObjectWasActivated(self, notification): if isinstance(notification.sender, Account) or (isinstance(notification.sender, BonjourAccount) and _bonjour.available): account = notification.sender self.accounts[account.id] = account notification.center.add_observer(self, sender=account, name='CFGSettingsObjectDidChange') notification.center.add_observer(self, sender=account, name='CFGSettingsObjectWasDeleted') notification.center.post_notification('SIPAccountManagerDidAddAccount', sender=self, data=NotificationData(account=account)) from sipsimple.application import SIPApplication if SIPApplication.running: call_in_green_thread(account.start) def _NH_CFGSettingsObjectWasCreated(self, notification): if isinstance(notification.sender, Account): account = notification.sender if account.enabled and self.default_account is None: self.default_account = account def _NH_CFGSettingsObjectWasDeleted(self, notification): account = notification.sender del self.accounts[account.id] notification.center.remove_observer(self, sender=account, name='CFGSettingsObjectDidChange') notification.center.remove_observer(self, sender=account, name='CFGSettingsObjectWasDeleted') notification.center.post_notification('SIPAccountManagerDidRemoveAccount', sender=self, data=NotificationData(account=account)) def _NH_CFGSettingsObjectDidChange(self, notification): account = notification.sender if '__id__' in notification.data.modified: modified_id = notification.data.modified['__id__'] self.accounts[modified_id.new] = self.accounts.pop(modified_id.old) if 'enabled' in notification.data.modified: if account.enabled and self.default_account is None: self.default_account = account elif not account.enabled and self.default_account is account: try: self.default_account = next((account for account in list(self.accounts.values()) if account.enabled)) except StopIteration: self.default_account = None @property def default_account(self): settings = SIPSimpleSettings() return self.accounts.get(settings.default_account, None) @default_account.setter def default_account(self, account): if account is not None and not account.enabled: raise ValueError("account %s is not enabled" % account.id) notification_center = NotificationCenter() settings = SIPSimpleSettings() with self._lock: old_account = self.accounts.get(settings.default_account, None) if account is old_account: return if account is None: settings.default_account = None else: settings.default_account = account.id settings.save() # we need to post the notification in the file-io thread in order to have it serialized after the # SIPAccountManagerDidAddAccount notification that is triggered when the account is saved the first # time, because save is executed in the file-io thread while this runs in the current thread. -Dan call_in_thread('file-io', notification_center.post_notification, 'SIPAccountManagerDidChangeDefaultAccount', sender=self, data=NotificationData(old_account=old_account, account=account)) diff --git a/sipsimple/account/publication.py b/sipsimple/account/publication.py index d25649dd..c4c7782b 100644 --- a/sipsimple/account/publication.py +++ b/sipsimple/account/publication.py @@ -1,398 +1,398 @@ """Implements the publisher handlers""" __all__ = ['Publisher', 'PresencePublisher', 'DialogPublisher'] import random from abc import ABCMeta, abstractproperty from threading import Lock from time import time from application.notification import IObserver, NotificationCenter, NotificationData from application.python import Null, limit from application.python.types import MarkerType from application.system import host as Host from eventlib import coros, proc from twisted.internet import reactor from zope.interface import implementer from sipsimple.core import FromHeader, Publication, PublicationETagError, RouteHeader, SIPURI, SIPCoreError from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.lookup import DNSLookup, DNSLookupError from sipsimple.payloads.dialoginfo import DialogInfoDocument from sipsimple.payloads.pidf import PIDFDocument from sipsimple.threading import run_in_twisted_thread from sipsimple.threading.green import Command, run_in_green_thread Command.register_defaults('publish', refresh_interval=None) class SameState(metaclass=MarkerType): pass class SIPPublicationDidFail(Exception): def __init__(self, data): self.data = data class SIPPublicationDidNotEnd(Exception): def __init__(self, data): self.data = data class PublicationError(Exception): def __init__(self, error, retry_after, refresh_interval=None): self.error = error self.retry_after = retry_after self.refresh_interval = refresh_interval class PublisherNickname(dict): def __missing__(self, name): return self.setdefault(name, name[:-9] if name.endswith('Publisher') else name) def __get__(self, obj, objtype): return self[objtype.__name__] def __set__(self, obj, value): raise AttributeError('cannot set attribute') def __delete__(self, obj): raise AttributeError('cannot delete attribute') @implementer(IObserver) class Publisher(object, metaclass=ABCMeta): __nickname__ = PublisherNickname() __transports__ = frozenset(['tls', 'tcp', 'udp']) def __init__(self, account): self.account = account self.started = False self.active = False self.publishing = False self._lock = Lock() self._command_proc = None self._command_channel = coros.queue() self._data_channel = coros.queue() self._publication = None self._dns_wait = 1 self._publish_wait = 1 self._publication_timer = None self.__dict__['state'] = None @abstractproperty def event(self): return None @abstractproperty def payload_type(self): return None @property def extra_headers(self): return [] @property def state(self): return self.__dict__['state'] @state.setter def state(self, state): if state is not None and not isinstance(state, self.payload_type.root_element): raise ValueError("state must be a %s document or None" % self.payload_type.root_element.__name__) with self._lock: old_state = self.__dict__['state'] self.__dict__['state'] = state if state == old_state: return self._publish(state) def start(self): if self.started: return self.started = True notification_center = NotificationCenter() notification_center.add_observer(self, sender=self) notification_center.post_notification(self.__class__.__name__ + 'WillStart', sender=self) notification_center.add_observer(self, name='CFGSettingsObjectDidChange', sender=self.account) notification_center.add_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) notification_center.add_observer(self, name='NetworkConditionsDidChange') self._command_proc = proc.spawn(self._run) notification_center.post_notification(self.__class__.__name__ + 'DidStart', sender=self) notification_center.remove_observer(self, sender=self) def stop(self): if not self.started: return self.started = False self.active = False notification_center = NotificationCenter() notification_center.add_observer(self, sender=self) notification_center.post_notification(self.__class__.__name__ + 'WillEnd', sender=self) notification_center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=self.account) notification_center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) notification_center.remove_observer(self, name='NetworkConditionsDidChange') command = Command('terminate') self._command_channel.send(command) command.wait() self._command_proc = None notification_center.post_notification(self.__class__.__name__ + 'DidDeactivate', sender=self) notification_center.post_notification(self.__class__.__name__ + 'DidEnd', sender=self) notification_center.remove_observer(self, sender=self) def activate(self): if not self.started: raise RuntimeError("not started") self.active = True self._command_channel.send(Command('publish', state=self.state)) notification_center = NotificationCenter() notification_center.post_notification(self.__class__.__name__ + 'DidActivate', sender=self) def deactivate(self): if not self.started: raise RuntimeError("not started") self.active = False self._command_channel.send(Command('unpublish')) notification_center = NotificationCenter() notification_center.post_notification(self.__class__.__name__ + 'DidDeactivate', sender=self) @run_in_twisted_thread def _publish(self, state): if not self.active: return if state is None: self._command_channel.send(Command('unpublish')) else: self._command_channel.send(Command('publish', state=state)) def _run(self): while True: command = self._command_channel.wait() handler = getattr(self, '_CH_%s' % command.name) handler(command) def _CH_publish(self, command): if command.state is None or self._publication is None and command.state is SameState: command.signal() return notification_center = NotificationCenter() settings = SIPSimpleSettings() if self._publication_timer is not None and self._publication_timer.active(): self._publication_timer.cancel() self._publication_timer = None if self._publication is None: duration = command.refresh_interval or self.account.sip.publish_interval from_header = FromHeader(self.account.uri, self.account.display_name) self._publication = Publication(from_header, self.event, self.payload_type.content_type, credentials=self.account.credentials, duration=duration, extra_headers=self.extra_headers) notification_center.add_observer(self, sender=self._publication) notification_center.post_notification(self.__class__.__name__ + 'WillPublish', sender=self, data=NotificationData(state=command.state, duration=duration)) else: notification_center.post_notification(self.__class__.__name__ + 'WillRefresh', sender=self, data=NotificationData(state=command.state)) try: if Host.default_ip is None: raise PublicationError('No IP address', retry_after=60) # Lookup routes valid_transports = self.__transports__.intersection(settings.sip.transport_list) if self.account.sip.outbound_proxy is not None and self.account.sip.outbound_proxy.transport in valid_transports: 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 = DNSLookup() try: - routes = lookup.lookup_sip_proxy(uri, valid_transports).wait() + routes = lookup.lookup_sip_proxy(uri, valid_transports, tls_name=self.account.sip.tls_name).wait() except DNSLookupError as e: retry_after = random.uniform(self._dns_wait, 2*self._dns_wait) self._dns_wait = limit(2*self._dns_wait, max=30) raise PublicationError('DNS lookup failed: %s' % e, retry_after=retry_after) else: self._dns_wait = 1 body = None if command.state is SameState else command.state.toxml() # Publish by trying each route in turn publish_timeout = time() + 30 for route in routes: if Host.default_ip is None: raise PublicationError('No IP address', retry_after=60) remaining_time = publish_timeout-time() if remaining_time > 0: try: try: self._publication.publish(body, RouteHeader(route.uri), timeout=limit(remaining_time, min=1, max=10)) except ValueError as e: # this happens for an initial PUBLISH with body=None raise PublicationError(str(e), retry_after=0) except PublicationETagError: state = self.state # access self.state only once to avoid race conditions if state is not None: self._publication.publish(state.toxml(), RouteHeader(route.uri), timeout=limit(remaining_time, min=1, max=10)) else: command.signal() return except SIPCoreError: raise PublicationError('Internal error', retry_after=5) try: while True: notification = self._data_channel.wait() if notification.name == 'SIPPublicationDidSucceed': break if notification.name == 'SIPPublicationDidEnd': raise PublicationError('Publication expired', retry_after=random.uniform(60, 120)) # publication expired while we were trying to re-publish except SIPPublicationDidFail as e: if e.data.code == 407: # Authentication failed, so retry the publication in some time raise PublicationError('Authentication failed', retry_after=random.uniform(60, 120)) elif e.data.code == 412: raise PublicationError('Conditional request failed', retry_after=0) elif e.data.code == 423: # Get the value of the Min-Expires header if e.data.min_expires is not None and e.data.min_expires > self.account.sip.publish_interval: refresh_interval = e.data.min_expires else: refresh_interval = None raise PublicationError('Interval too short', retry_after=random.uniform(60, 120), refresh_interval=refresh_interval) elif e.data.code in (405, 406, 489): raise PublicationError('Method or event not supported', retry_after=3600) else: # Otherwise just try the next route continue else: self.publishing = True self._publish_wait = 1 command.signal() break else: # There are no more routes to try, reschedule the publication retry_after = random.uniform(self._publish_wait, 2*self._publish_wait) self._publish_wait = limit(self._publish_wait*2, max=30) raise PublicationError('No more routes to try', retry_after=retry_after) except PublicationError as e: self.publishing = False notification_center.remove_observer(self, sender=self._publication) def publish(e): if self.active: self._command_channel.send(Command('publish', event=command.event, state=self.state, refresh_interval=e.refresh_interval)) else: command.signal() self._publication_timer = None self._publication_timer = reactor.callLater(e.retry_after, publish, e) self._publication = None notification_center.post_notification(self.__nickname__ + 'PublicationDidFail', sender=self, data=NotificationData(reason=e.error)) else: notification_center.post_notification(self.__nickname__ + 'PublicationDidSucceed', sender=self) def _CH_unpublish(self, command): # Cancel any timer which would restart the publication process if self._publication_timer is not None and self._publication_timer.active(): self._publication_timer.cancel() self._publication_timer = None publishing = self.publishing self.publishing = False if self._publication is not None: notification_center = NotificationCenter() if publishing: self._publication.end(timeout=2) try: while True: notification = self._data_channel.wait() if notification.name == 'SIPPublicationDidEnd': break except (SIPPublicationDidFail, SIPPublicationDidNotEnd): notification_center.post_notification(self.__nickname__ + 'PublicationDidNotEnd', sender=self) else: notification_center.post_notification(self.__nickname__ + 'PublicationDidEnd', sender=self) notification_center.remove_observer(self, sender=self._publication) self._publication = None command.signal() def _CH_terminate(self, command): self._CH_unpublish(command) raise proc.ProcExit @run_in_twisted_thread def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPPublicationDidSucceed(self, notification): if notification.sender is self._publication: self._data_channel.send(notification) def _NH_SIPPublicationDidFail(self, notification): if notification.sender is self._publication: self._data_channel.send_exception(SIPPublicationDidFail(notification.data)) def _NH_SIPPublicationDidEnd(self, notification): if notification.sender is self._publication: self._data_channel.send(notification) def _NH_SIPPublicationDidNotEnd(self, notification): if notification.sender is self._publication: self._data_channel.send_exception(SIPPublicationDidNotEnd(notification.data)) def _NH_SIPPublicationWillExpire(self, notification): if notification.sender is self._publication: self._publish(SameState) @run_in_green_thread def _NH_CFGSettingsObjectDidChange(self, notification): if not self.started: return if 'enabled' in notification.data.modified: return # global account activation is handled separately by the account itself elif 'presence.enabled' in notification.data.modified: if self.account.presence.enabled: self.activate() else: self.deactivate() elif self.active and {'__id__', 'auth.password', 'auth.username', 'sip.outbound_proxy', 'sip.transport_list', 'sip.publish_interval'}.intersection(notification.data.modified): self._command_channel.send(Command('unpublish')) self._command_channel.send(Command('publish', state=self.state)) def _NH_NetworkConditionsDidChange(self, notification): if self.active: self._command_channel.send(Command('unpublish')) self._command_channel.send(Command('publish', state=self.state)) class PresencePublisher(Publisher): """A publisher for presence state""" @property def event(self): return 'presence' @property def payload_type(self): return PIDFDocument def _NH_PresencePublisherDidStart(self, notification): if self.account.presence.enabled: self.activate() class DialogPublisher(Publisher): """A publisher for dialog info state""" @property def event(self): return 'dialog' @property def payload_type(self): return DialogInfoDocument def _NH_DialogPublisherDidStart(self, notification): if self.account.presence.enabled: self.activate() diff --git a/sipsimple/account/registration.py b/sipsimple/account/registration.py index 3798c57a..3b955709 100644 --- a/sipsimple/account/registration.py +++ b/sipsimple/account/registration.py @@ -1,324 +1,324 @@ """Implements the registration handler""" __all__ = ['Registrar'] import random from time import time from application.notification import IObserver, NotificationCenter, NotificationData from application.python import Null, limit from application.system import host as Host from eventlib import coros, proc from twisted.internet import reactor from zope.interface import implementer from sipsimple.core import ContactHeader, FromHeader, Header, Registration, RouteHeader, SIPURI, SIPCoreError, NoGRUU from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.lookup import DNSLookup, DNSLookupError from sipsimple.threading import run_in_twisted_thread from sipsimple.threading.green import Command, run_in_green_thread Command.register_defaults('register', refresh_interval=None) class SIPRegistrationDidFail(Exception): def __init__(self, data): self.data = data class SIPRegistrationDidNotEnd(Exception): def __init__(self, data): self.data = data class RegistrationError(Exception): def __init__(self, error, retry_after, refresh_interval=None): self.error = error self.retry_after = retry_after self.refresh_interval = refresh_interval @implementer(IObserver) class Registrar(object): def __init__(self, account): self.account = account self.started = False self.active = False self.registered = False self._command_proc = None self._command_channel = coros.queue() self._data_channel = coros.queue() self._registration = None self._dns_wait = 1 self._register_wait = 1 self._registration_timer = None def start(self): if self.started: return self.started = True notification_center = NotificationCenter() notification_center.add_observer(self, name='CFGSettingsObjectDidChange', sender=self.account) notification_center.add_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) notification_center.add_observer(self, name='NetworkConditionsDidChange') self._command_proc = proc.spawn(self._run) if self.account.sip.register: self.activate() def stop(self): if not self.started: return self.started = False self.active = False notification_center = NotificationCenter() notification_center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=self.account) notification_center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) notification_center.remove_observer(self, name='NetworkConditionsDidChange') command = Command('terminate') self._command_channel.send(command) command.wait() self._command_proc = None def activate(self): if not self.started: raise RuntimeError("not started") self.active = True self._command_channel.send(Command('register')) def deactivate(self): if not self.started: raise RuntimeError("not started") self.active = False self._command_channel.send(Command('unregister')) def reregister(self): if self.active: self._command_channel.send(Command('unregister')) self._command_channel.send(Command('register')) def _run(self): while True: command = self._command_channel.wait() #print('Registrar for %s got command %s' % (self.account.id, command.name)) handler = getattr(self, '_CH_%s' % command.name) handler(command) def _CH_register(self, command): notification_center = NotificationCenter() settings = SIPSimpleSettings() if self._registration_timer is not None and self._registration_timer.active(): self._registration_timer.cancel() self._registration_timer = None try: if Host.default_ip is None: raise RegistrationError('No IP address', retry_after=60) # Initialize the registration if self._registration is None: duration = command.refresh_interval or self.account.sip.register_interval try: self._registration = Registration(FromHeader(self.account.uri, self.account.display_name), credentials=self.account.credentials, duration=duration, extra_headers=[Header('Supported', 'gruu')]) except Exception as e: raise RegistrationError('Cannot create registration: %s' % str(e), retry_after=120) notification_center.add_observer(self, sender=self._registration) notification_center.post_notification('SIPAccountWillRegister', sender=self.account) else: notification_center.post_notification('SIPAccountRegistrationWillRefresh', sender=self.account) # Lookup routes if self.account.sip.outbound_proxy is not None and self.account.sip.outbound_proxy.transport in settings.sip.transport_list: 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 = DNSLookup() try: - routes = lookup.lookup_sip_proxy(uri, settings.sip.transport_list).wait() + routes = lookup.lookup_sip_proxy(uri, settings.sip.transport_list, tls_name=self.account.sip.tls_name).wait() except DNSLookupError as e: retry_after = int(random.uniform(self._dns_wait, 2*self._dns_wait)) self._dns_wait = limit(2*self._dns_wait, max=30) raise RegistrationError('DNS lookup failed: %s' % e, retry_after=retry_after) else: self._dns_wait = 1 # Register by trying each route in turn register_timeout = time() + 30 i = 0 for route in routes: i += 1 remaining_time = register_timeout-time() if remaining_time > 0: try: contact_uri = self.account.contact[NoGRUU, route] except KeyError: continue contact_header = ContactHeader(contact_uri) instance_id = '"<%s>"' % settings.instance_id contact_header.parameters[b"+sip.instance"] = instance_id.encode() if self.account.nat_traversal.use_ice: contact_header.parameters[b"+sip.ice"] = None route_header = RouteHeader(route.uri) try: self._registration.register(contact_header, route_header, timeout=limit(remaining_time, min=1, max=10)) except SIPCoreError: raise RegistrationError('Internal error', retry_after=5) try: while True: notification = self._data_channel.wait() if notification.name == 'SIPRegistrationDidSucceed': break if notification.name == 'SIPRegistrationDidEnd': raise RegistrationError('Registration expired', retry_after=int(random.uniform(60, 120))) # registration expired while we were trying to re-register except SIPRegistrationDidFail as e: notification_data = NotificationData(code=e.data.code, reason=e.data.reason, registration=self._registration, registrar=route) notification_center.post_notification('SIPAccountRegistrationGotAnswer', sender=self.account, data=notification_data) if e.data.code == 401: # Authentication failed, so retry the registration in some time raise RegistrationError('Authentication failed', retry_after=int(random.uniform(60, 120))) elif e.data.code == 408: # Timeout raise RegistrationError('Request timeout', retry_after=int(random.uniform(15, 40))) elif e.data.code == 423: # Get the value of the Min-Expires header if e.data.min_expires is not None and e.data.min_expires > self.account.sip.register_interval: refresh_interval = e.data.min_expires else: refresh_interval = None raise RegistrationError('Interval too short', retry_after=int(random.uniform(60, 120)), refresh_interval=refresh_interval) else: if i == len(routes): raise RegistrationError(e.data.reason, retry_after=int(random.uniform(15, 40))) else: # Otherwise just try the next route continue else: notification_data = NotificationData(code=notification.data.code, reason=notification.data.reason, registration=self._registration, registrar=route) notification_center.post_notification('SIPAccountRegistrationGotAnswer', sender=self.account, data=notification_data) self.registered = True # Save GRUU try: header = next(header for header in notification.data.contact_header_list if header.parameters.get('+sip.instance', '').strip('"<>') == settings.instance_id) except StopIteration: self.account.contact.public_gruu = None self.account.contact.temporary_gruu = None else: public_gruu = header.parameters.get('pub-gruu', None) temporary_gruu = header.parameters.get('temp-gruu', None) try: self.account.contact.public_gruu = SIPURI.parse(public_gruu.strip('"')) except (AttributeError, SIPCoreError): self.account.contact.public_gruu = None try: self.account.contact.temporary_gruu = SIPURI.parse(temporary_gruu.strip('"')) except (AttributeError, SIPCoreError): self.account.contact.temporary_gruu = None notification_data = NotificationData(contact_header=notification.data.contact_header, contact_header_list=notification.data.contact_header_list, expires=notification.data.expires_in, registrar=route) notification_center.post_notification('SIPAccountRegistrationDidSucceed', sender=self.account, data=notification_data) self._register_wait = 1 command.signal() break else: # There are no more routes to try, reschedule the registration retry_after = int(random.uniform(self._register_wait, 2*self._register_wait)) self._register_wait = limit(self._register_wait*2, max=30) raise RegistrationError('No more routes to try', retry_after=retry_after) except RegistrationError as e: self.registered = False notification_center.discard_observer(self, sender=self._registration) notification_center.post_notification('SIPAccountRegistrationDidFail', sender=self.account, data=NotificationData(error=e.error, retry_after=e.retry_after)) def register(e): if self.active: self._command_channel.send(Command('register', command.event, refresh_interval=e.refresh_interval)) self._registration_timer = None self._registration_timer = reactor.callLater(e.retry_after, register, e) self._registration = None self.account.contact.public_gruu = None self.account.contact.temporary_gruu = None def _CH_unregister(self, command): # Cancel any timer which would restart the registration process if self._registration_timer is not None and self._registration_timer.active(): self._registration_timer.cancel() self._registration_timer = None registered = self.registered self.registered = False if self._registration is not None: notification_center = NotificationCenter() if registered: self._registration.end(timeout=2) try: while True: notification = self._data_channel.wait() if notification.name == 'SIPRegistrationDidEnd': break except (SIPRegistrationDidFail, SIPRegistrationDidNotEnd) as e: notification_center.post_notification('SIPAccountRegistrationDidNotEnd', sender=self.account, data=NotificationData(code=e.data.code, reason=e.data.reason, registration=self._registration)) else: notification_center.post_notification('SIPAccountRegistrationDidEnd', sender=self.account, data=NotificationData(registration=self._registration)) notification_center.remove_observer(self, sender=self._registration) self._registration = None self.account.contact.public_gruu = None self.account.contact.temporary_gruu = None command.signal() def _CH_terminate(self, command): self._CH_unregister(command) raise proc.ProcExit @run_in_twisted_thread def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPRegistrationDidSucceed(self, notification): if notification.sender is self._registration: self._data_channel.send(notification) def _NH_SIPRegistrationDidFail(self, notification): if notification.sender is self._registration: self._data_channel.send_exception(SIPRegistrationDidFail(notification.data)) def _NH_SIPRegistrationDidEnd(self, notification): if notification.sender is self._registration: self._data_channel.send(notification) def _NH_SIPRegistrationDidNotEnd(self, notification): if notification.sender is self._registration: self._data_channel.send_exception(SIPRegistrationDidNotEnd(notification.data)) def _NH_SIPRegistrationWillExpire(self, notification): if self.active: self._command_channel.send(Command('register')) @run_in_green_thread def _NH_CFGSettingsObjectDidChange(self, notification): if not self.started: return if 'enabled' in notification.data.modified: return # global account activation is handled separately by the account itself elif 'sip.register' in notification.data.modified: if self.account.sip.register: self.activate() else: self.deactivate() elif self.active and {'__id__', 'auth.password', 'auth.username', 'nat_traversal.use_ice', 'sip.outbound_proxy', 'sip.transport_list', 'sip.register_interval'}.intersection(notification.data.modified): self._command_channel.send(Command('unregister')) self._command_channel.send(Command('register')) def _NH_NetworkConditionsDidChange(self, notification): if self.active: self._command_channel.send(Command('unregister')) self._command_channel.send(Command('register')) diff --git a/sipsimple/account/subscription.py b/sipsimple/account/subscription.py index a88dbbc9..058ca5bc 100644 --- a/sipsimple/account/subscription.py +++ b/sipsimple/account/subscription.py @@ -1,502 +1,502 @@ """Implements the subscription handlers""" __all__ = ['Subscriber', 'MWISubscriber', 'PresenceWinfoSubscriber', 'DialogWinfoSubscriber', 'PresenceSubscriber', 'SelfPresenceSubscriber', 'DialogSubscriber'] import random from abc import ABCMeta, abstractproperty from time import time from application.notification import IObserver, NotificationCenter, NotificationData from application.python import Null, limit from application.system import host as Host from eventlib import coros, proc from twisted.internet import reactor from zope.interface import implementer from sipsimple.core import ContactHeader, FromHeader, Header, RouteHeader, SIPURI, Subscription, ToHeader, SIPCoreError, NoGRUU from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.lookup import DNSLookup, DNSLookupError from sipsimple.threading import run_in_twisted_thread from sipsimple.threading.green import Command, run_in_green_thread Command.register_defaults('subscribe', refresh_interval=None) class SIPSubscriptionDidFail(Exception): def __init__(self, data): self.data = data class SubscriptionError(Exception): def __init__(self, error, retry_after, refresh_interval=None): self.error = error self.retry_after = retry_after self.refresh_interval = refresh_interval class InterruptSubscription(Exception): pass class TerminateSubscription(Exception): pass class Content(object): def __init__(self, body, type): self.body = body self.type = type class SubscriberNickname(dict): def __missing__(self, name): return self.setdefault(name, name[:-10] if name.endswith('Subscriber') else name) def __get__(self, obj, objtype): return self[objtype.__name__] def __set__(self, obj, value): raise AttributeError('cannot set attribute') def __delete__(self, obj): raise AttributeError('cannot delete attribute') @implementer(IObserver) class Subscriber(object, metaclass=ABCMeta): __nickname__ = SubscriberNickname() __transports__ = frozenset(['tls', 'tcp', 'udp']) def __init__(self, account): self.account = account self.started = False self.active = False self.subscribed = False self._command_proc = None self._command_channel = coros.queue() self._data_channel = coros.queue() self._subscription = None self._subscription_proc = None self._subscription_timer = None @abstractproperty def event(self): return None @property def subscription_uri(self): return self.account.id @property def content(self): return Content(None, None) @property def extra_headers(self): return [] def start(self): if self.started: return self.started = True notification_center = NotificationCenter() notification_center.add_observer(self, sender=self) notification_center.post_notification(self.__class__.__name__ + 'WillStart', sender=self) notification_center.add_observer(self, name='NetworkConditionsDidChange') self._command_proc = proc.spawn(self._run) notification_center.post_notification(self.__class__.__name__ + 'DidStart', sender=self) notification_center.remove_observer(self, sender=self) def stop(self): if not self.started: return self.started = False self.active = False notification_center = NotificationCenter() notification_center.add_observer(self, sender=self) notification_center.post_notification(self.__class__.__name__ + 'WillEnd', sender=self) notification_center.remove_observer(self, name='NetworkConditionsDidChange') command = Command('terminate') self._command_channel.send(command) command.wait() self._command_proc = None notification_center.post_notification(self.__class__.__name__ + 'DidDeactivate', sender=self) notification_center.post_notification(self.__class__.__name__ + 'DidEnd', sender=self) notification_center.remove_observer(self, sender=self) def activate(self): if not self.started: raise RuntimeError("not started") self.active = True self._command_channel.send(Command('subscribe')) notification_center = NotificationCenter() notification_center.post_notification(self.__class__.__name__ + 'DidActivate', sender=self) def deactivate(self): if not self.started: raise RuntimeError("not started") self.active = False self._command_channel.send(Command('unsubscribe')) notification_center = NotificationCenter() notification_center.post_notification(self.__class__.__name__ + 'DidDeactivate', sender=self) def resubscribe(self): if self.active: self._command_channel.send(Command('subscribe')) def _run(self): while True: command = self._command_channel.wait() handler = getattr(self, '_CH_%s' % command.name) handler(command) def _CH_subscribe(self, command): if self._subscription_timer is not None and self._subscription_timer.active(): self._subscription_timer.cancel() self._subscription_timer = None if self._subscription_proc is not None: subscription_proc = self._subscription_proc subscription_proc.kill(InterruptSubscription) subscription_proc.wait() self._subscription_proc = proc.spawn(self._subscription_handler, command) def _CH_unsubscribe(self, command): # Cancel any timer which would restart the subscription process if self._subscription_timer is not None and self._subscription_timer.active(): self._subscription_timer.cancel() self._subscription_timer = None if self._subscription_proc is not None: subscription_proc = self._subscription_proc subscription_proc.kill(TerminateSubscription) subscription_proc.wait() self._subscription_proc = None command.signal() def _CH_terminate(self, command): self._CH_unsubscribe(command) raise proc.ProcExit def _subscription_handler(self, command): notification_center = NotificationCenter() settings = SIPSimpleSettings() subscription_uri = self.subscription_uri refresh_interval = command.refresh_interval or self.account.sip.subscribe_interval valid_transports = self.__transports__.intersection(settings.sip.transport_list) try: if Host.default_ip is None: raise SubscriptionError('No IP address', retry_after=60) # Lookup routes if self.account.sip.outbound_proxy is not None and self.account.sip.outbound_proxy.transport in valid_transports: 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 = SIPURI(host=subscription_uri.domain) lookup = DNSLookup() try: - routes = lookup.lookup_sip_proxy(uri, valid_transports).wait() + routes = lookup.lookup_sip_proxy(uri, valid_transports, tls_name=self.account.sip.tls_name).wait() except DNSLookupError as e: raise SubscriptionError('DNS lookup failed: %s' % e, retry_after=random.uniform(15, 30)) subscription_uri = SIPURI(user=subscription_uri.username, host=subscription_uri.domain) content = self.content timeout = time() + 30 for route in routes: if Host.default_ip is None: raise SubscriptionError('No IP address', retry_after=60) remaining_time = timeout - time() if remaining_time > 0: try: contact_uri = self.account.contact[NoGRUU, route] except KeyError: continue subscription = Subscription(subscription_uri, FromHeader(self.account.uri, self.account.display_name), ToHeader(subscription_uri), ContactHeader(contact_uri), self.event.encode(), RouteHeader(route.uri), credentials=self.account.credentials, refresh=refresh_interval) notification_center.add_observer(self, sender=subscription) try: subscription.subscribe(body=content.body, content_type=content.type, extra_headers=self.extra_headers, timeout=limit(remaining_time, min=1, max=5)) except SIPCoreError: notification_center.remove_observer(self, sender=subscription) raise SubscriptionError('Internal error', retry_after=5) self._subscription = subscription try: while True: notification = self._data_channel.wait() if notification.name == 'SIPSubscriptionDidStart': break except SIPSubscriptionDidFail as e: notification_center.remove_observer(self, sender=subscription) self._subscription = None if e.data.code == 407: # Authentication failed, so retry the subscription in some time raise SubscriptionError('Authentication failed', retry_after=random.uniform(60, 120)) elif e.data.code == 423: # Get the value of the Min-Expires header if e.data.min_expires is not None and e.data.min_expires > self.account.sip.subscribe_interval: refresh_interval = e.data.min_expires else: refresh_interval = None raise SubscriptionError('Interval too short', retry_after=random.uniform(60, 120), refresh_interval=refresh_interval) elif e.data.code in (405, 406, 489): raise SubscriptionError('Method or event not supported', retry_after=3600) elif e.data.code == 1400: raise SubscriptionError(e.data.reason, retry_after=3600) else: # Otherwise just try the next route continue else: self.subscribed = True command.signal() break else: # There are no more routes to try, reschedule the subscription raise SubscriptionError('No more routes to try', retry_after=random.uniform(60, 180)) # At this point it is subscribed. Handle notifications and ending/failures. notification_center.post_notification(self.__nickname__ + 'SubscriptionDidStart', sender=self) try: while True: notification = self._data_channel.wait() if notification.name == 'SIPSubscriptionGotNotify': notification_center.post_notification(self.__nickname__ + 'SubscriptionGotNotify', sender=self, data=notification.data) elif notification.name == 'SIPSubscriptionDidEnd': notification_center.post_notification(self.__nickname__ + 'SubscriptionDidEnd', sender=self, data=NotificationData(originator='remote')) if self.active: self._command_channel.send(Command('subscribe')) break except SIPSubscriptionDidFail: notification_center.post_notification(self.__nickname__ + 'SubscriptionDidFail', sender=self) if self.active: self._command_channel.send(Command('subscribe')) notification_center.remove_observer(self, sender=self._subscription) except InterruptSubscription as e: if not self.subscribed: command.signal(e) if self._subscription is not None: notification_center.remove_observer(self, sender=self._subscription) try: self._subscription.end(timeout=2) except SIPCoreError: pass finally: notification_center.post_notification(self.__nickname__ + 'SubscriptionDidEnd', sender=self, data=NotificationData(originator='local')) except TerminateSubscription as e: if not self.subscribed: command.signal(e) if self._subscription is not None: try: self._subscription.end(timeout=2) except SIPCoreError: pass else: try: while True: notification = self._data_channel.wait() if notification.name == 'SIPSubscriptionDidEnd': break except SIPSubscriptionDidFail: pass finally: notification_center.remove_observer(self, sender=self._subscription) notification_center.post_notification(self.__nickname__ + 'SubscriptionDidEnd', sender=self, data=NotificationData(originator='local')) except SubscriptionError as e: def subscribe(e): if self.active: self._command_channel.send(Command('subscribe', command.event, refresh_interval=e.refresh_interval)) self._subscription_timer = None self._subscription_timer = reactor.callLater(e.retry_after, subscribe, e) notification_center.post_notification(self.__nickname__ + 'SubscriptionDidFail', sender=self) finally: self.subscribed = False self._subscription = None self._subscription_proc = None @run_in_twisted_thread def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPSubscriptionDidStart(self, notification): if notification.sender is self._subscription: self._data_channel.send(notification) def _NH_SIPSubscriptionDidEnd(self, notification): if notification.sender is self._subscription: self._data_channel.send(notification) def _NH_SIPSubscriptionDidFail(self, notification): if notification.sender is self._subscription: self._data_channel.send_exception(SIPSubscriptionDidFail(notification.data)) def _NH_SIPSubscriptionGotNotify(self, notification): if notification.sender is self._subscription: self._data_channel.send(notification) def _NH_NetworkConditionsDidChange(self, notification): if self.active: self._command_channel.send(Command('subscribe')) class MWISubscriber(Subscriber): """Message Waiting Indicator subscriber""" @property def event(self): return 'message-summary' @property def subscription_uri(self): return self.account.message_summary.voicemail_uri or self.account.id def _NH_MWISubscriberWillStart(self, notification): notification.center.add_observer(self, name='CFGSettingsObjectDidChange', sender=self.account) notification.center.add_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) def _NH_MWISubscriberWillEnd(self, notification): notification.center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=self.account) notification.center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) def _NH_MWISubscriberDidStart(self, notification): if self.account.message_summary.enabled: self.activate() @run_in_green_thread def _NH_CFGSettingsObjectDidChange(self, notification): if not self.started: return if 'enabled' in notification.data.modified: return # global account activation is handled separately by the account itself elif 'message_summary.enabled' in notification.data.modified: if self.account.message_summary.enabled: self.activate() else: self.deactivate() elif self.active and {'__id__', 'auth.password', 'auth.username', 'message_summary.voicemail_uri', 'sip.always_use_my_proxy', 'sip.outbound_proxy', 'sip.subscribe_interval', 'sip.transport_list'}.intersection(notification.data.modified): self._command_channel.send(Command('subscribe')) class AbstractPresenceSubscriber(Subscriber): """Abstract class defining behavior for all presence subscribers""" __transports__ = frozenset(['tls', 'tcp']) def _NH_AbstractPresenceSubscriberWillStart(self, notification): notification.center.add_observer(self, name='SIPAccountDidDiscoverXCAPSupport', sender=self.account) notification.center.add_observer(self, name='CFGSettingsObjectDidChange', sender=self.account) notification.center.add_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) def _NH_AbstractPresenceSubscriberWillEnd(self, notification): notification.center.remove_observer(self, name='SIPAccountDidDiscoverXCAPSupport', sender=self.account) notification.center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=self.account) notification.center.remove_observer(self, name='CFGSettingsObjectDidChange', sender=SIPSimpleSettings()) def _NH_AbstractPresenceSubscriberDidStart(self, notification): if self.account.presence.enabled and self.account.xcap.discovered: self.activate() def _NH_SIPAccountDidDiscoverXCAPSupport(self, notification): if self.account.presence.enabled and not self.active: self.activate() @run_in_green_thread def _NH_CFGSettingsObjectDidChange(self, notification): if not self.started or not self.account.xcap.discovered: return if 'enabled' in notification.data.modified: return # global account activation is handled separately by the account itself elif 'presence.enabled' in notification.data.modified: if self.account.presence.enabled: self.activate() else: self.deactivate() elif self.active and {'__id__', 'auth.password', 'auth.username', 'sip.always_use_my_proxy', 'sip.outbound_proxy', 'sip.subscribe_interval', 'sip.transport_list'}.intersection(notification.data.modified): self._command_channel.send(Command('subscribe')) class PresenceWinfoSubscriber(AbstractPresenceSubscriber): """Presence Watcher Info subscriber""" _NH_PresenceWinfoSubscriberWillStart = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberWillStart _NH_PresenceWinfoSubscriberWillEnd = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberWillEnd _NH_PresenceWinfoSubscriberDidStart = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberDidStart @property def event(self): return 'presence.winfo' class DialogWinfoSubscriber(AbstractPresenceSubscriber): """Dialog Watcher Info subscriber""" _NH_DialogWinfoSubscriberWillStart = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberWillStart _NH_DialogWinfoSubscriberWillEnd = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberWillEnd _NH_DialogWinfoSubscriberDidStart = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberDidStart @property def event(self): return 'dialog.winfo' class PresenceSubscriber(AbstractPresenceSubscriber): """Presence subscriber""" _NH_PresenceSubscriberWillStart = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberWillStart _NH_PresenceSubscriberWillEnd = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberWillEnd _NH_PresenceSubscriberDidStart = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberDidStart @property def event(self): return 'presence' @property def subscription_uri(self): return self.account.xcap_manager.rls_presence_uri @property def extra_headers(self): return [Header('Supported', 'eventlist')] class SelfPresenceSubscriber(AbstractPresenceSubscriber): """Self presence subscriber""" _NH_SelfPresenceSubscriberWillStart = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberWillStart _NH_SelfPresenceSubscriberWillEnd = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberWillEnd _NH_SelfPresenceSubscriberDidStart = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberDidStart @property def event(self): return 'presence' @property def subscription_uri(self): return self.account.id class DialogSubscriber(AbstractPresenceSubscriber): """Dialog subscriber""" _NH_DialogSubscriberWillStart = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberWillStart _NH_DialogSubscriberWillEnd = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberWillEnd _NH_DialogSubscriberDidStart = AbstractPresenceSubscriber._NH_AbstractPresenceSubscriberDidStart @property def event(self): return 'dialog' @property def subscription_uri(self): return self.account.xcap_manager.rls_dialog_uri @property def extra_headers(self): return [Header('Supported', 'eventlist')] diff --git a/sipsimple/lookup.py b/sipsimple/lookup.py index 27672da7..4093fb74 100644 --- a/sipsimple/lookup.py +++ b/sipsimple/lookup.py @@ -1,575 +1,577 @@ """ Implements DNS lookups in the context of SIP, STUN and MSRP relay based on RFC3263 and related standards. This can be used to determine the next hop(s) and failover for routing of SIP messages and reservation of network resources prior the starting of a SIP session. """ import re from itertools import chain from time import time from urllib.parse import urlparse # patch dns.entropy module which is not thread-safe import dns import sys from functools import partial from random import randint, randrange dns.entropy = dns.__class__('dns.entropy') dns.entropy.__file__ = dns.__file__.replace('__init__.py', 'entropy.py') dns.entropy.__builtins__ = dns.__builtins__ dns.entropy.random_16 = partial(randrange, 2**16) dns.entropy.between = randint sys.modules['dns.entropy'] = dns.entropy del partial, randint, randrange, sys # replace standard select and socket modules with versions from eventlib from eventlib import coros, proc from eventlib.green import select from eventlib.green import socket import dns.name import dns.resolver import dns.query dns.resolver.socket = socket dns.query.socket = socket dns.query.select = select #TODO3: dnspython newer than 0.20 has a new API if ('_set_polling_backend' in dir(dns.query)): dns.query._set_polling_backend(dns.query._select_for) from application.notification import IObserver, NotificationCenter, NotificationData from application.python import Null, limit from application.python.decorator import decorator, preserve_signature from application.python.types import Singleton from dns import exception, rdatatype from twisted.internet import reactor from zope.interface import implementer from sipsimple.core import Route from sipsimple.threading import run_in_twisted_thread from sipsimple.threading.green import Command, InterruptCommand, run_in_waitable_green_thread def domain_iterator(domain): """ A generator which returns the domain and its parent domains. """ while domain not in ('.', ''): yield domain domain = (domain.split('.', 1)+[''])[1] @decorator def post_dns_lookup_notifications(func): @preserve_signature(func) def wrapper(obj, *args, **kwargs): notification_center = NotificationCenter() try: result = func(obj, *args, **kwargs) except DNSLookupError as e: notification_center.post_notification('DNSLookupDidFail', sender=obj, data=NotificationData(error=str(e))) raise else: notification_center.post_notification('DNSLookupDidSucceed', sender=obj, data=NotificationData(result=result)) return result return wrapper class DNSLookupError(Exception): """ The error raised by DNSLookup when a lookup cannot be performed. """ class DNSCache(object): """ A simple DNS cache which uses twisted's timers to invalidate its expired data. """ def __init__(self): self.data = {} def get(self, key): return self.data.get(key, None) def put(self, key, value): expiration = value.expiration-time() if expiration > 0: self.data[key] = value reactor.callLater(limit(expiration, max=3600), self.data.pop, key, None) def flush(self, key=None): if key is not None: self.data.pop(key, None) else: self.data = {} class InternalResolver(dns.resolver.Resolver): def __init__(self, *args, **kw): super(InternalResolver, self).__init__(*args, **kw) if self.domain.to_text().endswith('local.'): self.domain = dns.name.root self.search = [item for item in self.search if not item.to_text().endswith('local.')] class DNSResolver(dns.resolver.Resolver): """ The resolver used by DNSLookup. The lifetime setting on it applies to all the queries made on this resolver. Each time a query is performed, its duration is subtracted from the lifetime value. """ def __init__(self): dns.resolver.Resolver.__init__(self, configure=False) dns_manager = DNSManager() self.search = dns_manager.search self.domain = dns_manager.domain self.nameservers = dns_manager.nameservers def query(self, *args, **kw): start_time = time() try: return dns.resolver.Resolver.query(self, *args, **kw) finally: self.lifetime -= min(self.lifetime, time()-start_time) class SRVResult(object): """ Internal object used to save the result of SRV queries. """ def __init__(self, priority, weight, port, address): self.priority = priority self.weight = weight self.port = port self.address = address class NAPTRResult(object): """ Internal object used to save the result of NAPTR queries. """ def __init__(self, service, order, preference, priority, weight, port, address): self.service = service self.order = order self.preference = preference self.priority = priority self.weight = weight self.port = port self.address = address class DNSLookup(object): cache = DNSCache() @run_in_waitable_green_thread @post_dns_lookup_notifications def lookup_service(self, uri, service, timeout=3.0, lifetime=15.0): """ Performs an SRV query to determine the servers used for the specified service from the domain in uri.host. If this fails and falling back is supported, also performs an A query on uri.host, returning the default port of the service along with the IP addresses in the answer. The services supported are `stun' and 'msrprelay'. The DNSLookupDidSucceed notification contains a result attribute which is a list of (address, port) tuples. The DNSLookupDidFail notification contains an error attribute describing the error encountered. """ service_srv_record_map = {"stun": ("_stun._udp", 3478, False), "msrprelay": ("_msrps._tcp", 2855, True)} log_context = dict(context='lookup_service', service=service, uri=uri) try: service_prefix, service_port, service_fallback = service_srv_record_map[service] except KeyError: raise DNSLookupError("Unknown service: %s" % service) try: # If the host part of the URI is an IP address, we will not do any lookup if re.match("^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", uri.host.decode()): return [(uri.host.decode(), uri.port or service_port)] resolver = DNSResolver() resolver.cache = self.cache resolver.timeout = timeout resolver.lifetime = lifetime record_name = '%s.%s' % (service_prefix, uri.host.decode()) services = self._lookup_srv_records(resolver, [record_name], log_context=log_context) if services[record_name]: return [(result.address, result.port) for result in services[record_name]] elif service_fallback: addresses = self._lookup_a_records(resolver, [uri.host.decode()], log_context=log_context) if addresses[uri.host.decode()]: return [(addr, service_port) for addr in addresses[uri.host.decode()]] except dns.resolver.Timeout: raise DNSLookupError('Timeout in lookup for %s servers for domain %s' % (service, uri.host.decode())) else: raise DNSLookupError('No %s servers found for domain %s' % (service, uri.host.decode())) @run_in_waitable_green_thread @post_dns_lookup_notifications - def lookup_sip_proxy(self, uri, supported_transports, timeout=3.0, lifetime=15.0): + def lookup_sip_proxy(self, uri, supported_transports, timeout=3.0, lifetime=15.0, tls_name=None): """ Performs an RFC 3263 compliant lookup of transport/ip/port combinations for a particular SIP URI. As arguments it takes a SIPURI object and a list of supported transports, in order of preference of the application. It returns a list of Route objects that can be used in order of preference. The DNSLookupDidSucceed notification contains a result attribute which is a list of Route objects. The DNSLookupDidFail notification contains an error attribute describing the error encountered. + + Set tls_name to the Common Name the server must present """ naptr_service_transport_map = {"sips+d2t": "tls", "sip+d2t": "tcp", "sip+d2u": "udp"} transport_service_map = {"udp": "_sip._udp", "tcp": "_sip._tcp", "tls": "_sips._tcp"} log_context = dict(context='lookup_sip_proxy', uri=uri) if not supported_transports: raise DNSLookupError("No transports are supported") supported_transports = [transport.lower() for transport in supported_transports] unknown_transports = set(supported_transports).difference(transport_service_map) if unknown_transports: raise DNSLookupError("Unknown transports: %s" % ', '.join(unknown_transports)) try: # If the host part of the URI is an IP address, we will not do any lookup transport = uri.transport.decode() if isinstance(uri.transport, bytes) else uri.transport if re.match("^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", uri.host.decode()): transport = 'tls' if uri.secure else transport.lower() if transport not in supported_transports: raise DNSLookupError("IP transport %s dictated by URI is not supported" % transport) port = uri.port or (5061 if transport=='tls' else 5060) route = [Route(address=uri.host, port=port, transport=transport)] return route resolver = DNSResolver() resolver.cache = self.cache resolver.timeout = timeout resolver.lifetime = lifetime # If the port is specified in the URI, we will only do an A lookup if uri.port: transport = 'tls' if uri.secure else transport.lower() if transport not in supported_transports: raise DNSLookupError("Host transport %s dictated by URI is not supported" % transport) addresses = self._lookup_a_records(resolver, [uri.host.decode()], log_context=log_context) if addresses[uri.host.decode()]: - return [Route(address=addr, port=uri.port, transport=transport, tls_name=uri.host) for addr in addresses[uri.host.decode()]] + return [Route(address=addr, port=uri.port, transport=transport, tls_name=tls_name or uri.host) for addr in addresses[uri.host.decode()]] # If the transport was already set as a parameter on the SIP URI, only do SRV lookups elif 'transport' in uri.parameters: transport = uri.parameters['transport'].lower() if transport not in supported_transports: raise DNSLookupError("Requested lookup for URI with %s transport, but it is not supported" % transport) if uri.secure and transport != 'tls': raise DNSLookupError("Requested lookup for SIPS URI, but with %s transport parameter" % transport) record_name = '%s.%s' % (transport_service_map[transport], uri.host.decode()) services = self._lookup_srv_records(resolver, [record_name], log_context=log_context) if services[record_name]: - return [Route(address=result.address, port=result.port, transport=transport, tls_name=uri.host) for result in services[record_name]] + return [Route(address=result.address, port=result.port, transport=transport, tls_name=tls_name or uri.host) for result in services[record_name]] else: # If SRV lookup fails, try A lookup addresses = self._lookup_a_records(resolver, [uri.host.decode()], log_context=log_context) port = 5061 if transport=='tls' else 5060 if addresses[uri.host.decode()]: - return [Route(address=addr, port=port, transport=transport, tls_name=uri.host) for addr in addresses[uri.host.decode()]] + return [Route(address=addr, port=port, transport=transport, tls_name=tls_name or uri.host) for addr in addresses[uri.host.decode()]] # Otherwise, it means we don't have a numeric IP address, a port isn't specified and neither is a transport. So we have to do a full NAPTR lookup else: # If the URI is a SIPS URI, we only support the TLS transport. if uri.secure: if 'tls' not in supported_transports: raise DNSLookupError("Requested lookup for SIPS URI, but TLS transport is not supported") supported_transports = ['tls'] # First try NAPTR lookup naptr_services = [service for service, transport in list(naptr_service_transport_map.items()) if transport in supported_transports] try: pointers = self._lookup_naptr_record(resolver, uri.host.decode(), naptr_services, log_context=log_context) except dns.resolver.Timeout: pointers = [] if pointers: - return [Route(address=result.address, port=result.port, transport=naptr_service_transport_map[result.service], tls_name=uri.host) for result in pointers] + return [Route(address=result.address, port=result.port, transport=naptr_service_transport_map[result.service], tls_name=tls_name or uri.host) for result in pointers] else: # If that fails, try SRV lookup routes = [] for transport in supported_transports: record_name = '%s.%s' % (transport_service_map[transport], uri.host.decode()) try: services = self._lookup_srv_records(resolver, [record_name], log_context=log_context) except dns.resolver.Timeout: continue if services[record_name]: - routes.extend(Route(address=result.address, port=result.port, transport=transport, tls_name=uri.host) for result in services[record_name]) + routes.extend(Route(address=result.address, port=result.port, transport=transport, tls_name=tls_name or uri.host) for result in services[record_name]) if routes: return routes else: # If SRV lookup fails, try A lookup transport = 'tls' if uri.secure else 'udp' if transport in supported_transports: addresses = self._lookup_a_records(resolver, [uri.host.decode()], log_context=log_context) port = 5061 if transport=='tls' else 5060 if addresses[uri.host.decode()]: - return [Route(address=addr, port=port, transport=transport, tls_name=uri.host) for addr in addresses[uri.host.decode()]] + return [Route(address=addr, port=port, transport=transport, tls_name=tls_name or uri.host) for addr in addresses[uri.host.decode()]] except dns.resolver.Timeout: raise DNSLookupError("Timeout in lookup for routes for SIP URI %s" % uri) else: raise DNSLookupError("No routes found for SIP URI %s" % uri) @run_in_waitable_green_thread @post_dns_lookup_notifications def lookup_xcap_server(self, uri, timeout=3.0, lifetime=15.0): """ Performs a TXT query against xcap. and returns all results that look like HTTP URIs. """ log_context = dict(context='lookup_xcap_server', uri=uri) notification_center = NotificationCenter() try: # If the host part of the URI is an IP address, we cannot not do any lookup if re.match("^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", uri.host.decode()): raise DNSLookupError("Cannot perform DNS query because the host is an IP address") resolver = DNSResolver() resolver.cache = self.cache resolver.timeout = timeout resolver.lifetime = lifetime record_name = 'xcap.%s' % uri.host.decode() results = [] try: answer = resolver.query(record_name, rdatatype.TXT) except dns.resolver.Timeout as e: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='TXT', query_name=str(record_name), nameservers=resolver.nameservers, answer=None, error=e, **log_context)) raise except exception.DNSException as e: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='TXT', query_name=str(record_name), nameservers=resolver.nameservers, answer=None, error=e, **log_context)) else: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='TXT', query_name=str(record_name), nameservers=resolver.nameservers, answer=answer, error=None, **log_context)) for result_uri in list(chain(*(r.strings for r in answer.rrset))): parsed_uri = urlparse(result_uri.decode()) if parsed_uri.scheme in ('http', 'https') and parsed_uri.netloc: results.append(result_uri.decode()) if not results: raise DNSLookupError('No XCAP servers found for domain %s' % uri.host.decode()) return results except dns.resolver.Timeout: raise DNSLookupError('Timeout in lookup for XCAP servers for domain %s' % uri.host.decode()) def _lookup_a_records(self, resolver, hostnames, additional_records=[], log_context={}): notification_center = NotificationCenter() additional_addresses = dict((rset.name.to_text(), rset) for rset in additional_records if rset.rdtype == rdatatype.A) addresses = {} for hostname in hostnames: if hostname in additional_addresses: addresses[hostname] = [r.address for r in additional_addresses[hostname]] else: try: answer = resolver.query(hostname, rdatatype.A) except dns.resolver.Timeout as e: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='A', query_name=str(hostname), nameservers=resolver.nameservers, answer=None, error=e, **log_context)) raise except exception.DNSException as e: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='A', query_name=str(hostname), nameservers=resolver.nameservers, answer=None, error=e, **log_context)) addresses[hostname] = [] else: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='A', query_name=str(hostname), nameservers=resolver.nameservers, answer=answer, error=None, **log_context)) addresses[hostname] = [r.address for r in answer.rrset] return addresses def _lookup_srv_records(self, resolver, srv_names, additional_records=[], log_context={}): notification_center = NotificationCenter() additional_services = dict((rset.name.to_text(), rset) for rset in additional_records if rset.rdtype == rdatatype.SRV) services = {} for srv_name in srv_names: services[srv_name] = [] if srv_name in additional_services: addresses = self._lookup_a_records(resolver, [r.target.to_text() for r in additional_services[srv_name]], additional_records) for record in additional_services[srv_name]: services[srv_name].extend(SRVResult(record.priority, record.weight, record.port, addr) for addr in addresses.get(record.target.to_text(), ())) else: try: answer = resolver.query(srv_name, rdatatype.SRV) except dns.resolver.Timeout as e: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='SRV', query_name=str(srv_name), nameservers=resolver.nameservers, answer=None, error=e, **log_context)) raise except exception.DNSException as e: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='SRV', query_name=str(srv_name), nameservers=resolver.nameservers, answer=None, error=e, **log_context)) else: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='SRV', query_name=str(srv_name), nameservers=resolver.nameservers, answer=answer, error=None, **log_context)) addresses = self._lookup_a_records(resolver, [r.target.to_text() for r in answer.rrset], answer.response.additional, log_context) for record in answer.rrset: services[srv_name].extend(SRVResult(record.priority, record.weight, record.port, addr) for addr in addresses.get(record.target.to_text(), ())) services[srv_name].sort(key=lambda result: (result.priority, -result.weight)) return services def _lookup_naptr_record(self, resolver, domain, services, log_context={}): notification_center = NotificationCenter() pointers = [] try: answer = resolver.query(domain, rdatatype.NAPTR) except dns.resolver.Timeout as e: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='NAPTR', query_name=str(domain), nameservers=resolver.nameservers, answer=None, error=e, **log_context)) raise except exception.DNSException as e: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='NAPTR', query_name=str(domain), nameservers=resolver.nameservers, answer=None, error=e, **log_context)) else: notification_center.post_notification('DNSLookupTrace', sender=self, data=NotificationData(query_type='NAPTR', query_name=str(domain), nameservers=resolver.nameservers, answer=answer, error=None, **log_context)) records = [r for r in answer.rrset if r.service.decode().lower() in services] services = self._lookup_srv_records(resolver, [r.replacement.to_text() for r in records], answer.response.additional, log_context) for record in records: pointers.extend(NAPTRResult(record.service.decode().lower(), record.order, record.preference, r.priority, r.weight, r.port, r.address) for r in services.get(record.replacement.to_text(), ())) pointers.sort(key=lambda result: (result.order, result.preference)) return pointers @implementer(IObserver) class DNSManager(object, metaclass=Singleton): def __init__(self): try: default_resolver = InternalResolver() except dns.resolver.NoResolverConfiguration: default_resolver = Null self.search = default_resolver.search self.domain = default_resolver.domain self.google_nameservers = ['8.8.8.8', '8.8.4.4'] self.nameservers = default_resolver.nameservers or [] self.probed_domain = 'sip2sip.info.' self._channel = coros.queue() self._proc = None self._timer = None self._wakeup_timer = None notification_center = NotificationCenter() notification_center.add_observer(self, name='SystemIPAddressDidChange') notification_center.add_observer(self, name='SystemDidWakeUpFromSleep') @property def nameservers(self): return self.__dict__['nameservers'] @nameservers.setter def nameservers(self, value): old_value = self.__dict__.get('nameservers', Null) self.__dict__['nameservers'] = value if old_value is Null: NotificationCenter().post_notification('DNSResolverDidInitialize', sender=self, data=NotificationData(nameservers=value)) elif value != old_value: NotificationCenter().post_notification('DNSNameserversDidChange', sender=self, data=NotificationData(nameservers=value)) def start(self): self._proc = proc.spawn(self._run) self._channel.send(Command('probe_dns')) def stop(self): if self._proc is not None: self._proc.kill() self._proc = None if self._timer is not None and self._timer.active(): self._timer.cancel() self._timer = None if self._wakeup_timer is not None and self._wakeup_timer.active(): self._wakeup_timer.cancel() self._wakeup_timer = None def _run(self): while True: try: command = self._channel.wait() handler = getattr(self, '_CH_%s' % command.name) handler(command) except InterruptCommand: pass def _CH_probe_dns(self, command): if self._timer is not None and self._timer.active(): self._timer.cancel() self._timer = None try: resolver = InternalResolver() except dns.resolver.NoResolverConfiguration as e: self._timer = reactor.callLater(15, self._channel.send, Command('probe_dns')) return self.domain = resolver.domain self.search = resolver.search local_nameservers = resolver.nameservers # probe local resolver resolver.timeout = 1 resolver.lifetime = 3 try: answer = resolver.query(self.probed_domain, rdatatype.NAPTR) if not any(record.rdtype == rdatatype.NAPTR for record in answer.rrset): raise exception.DNSException("No NAPTR records found") answer = resolver.query("_sip._udp.%s" % self.probed_domain, rdatatype.SRV) if not any(record.rdtype == rdatatype.SRV for record in answer.rrset): raise exception.DNSException("No SRV records found") except (dns.resolver.Timeout, exception.DNSException): pass else: self.nameservers = resolver.nameservers return # local resolver failed. probe google resolver resolver.nameservers = self.google_nameservers resolver.timeout = 2 resolver.lifetime = 4 try: answer = resolver.query(self.probed_domain, rdatatype.NAPTR) if not any(record.rdtype == rdatatype.NAPTR for record in answer.rrset): raise exception.DNSException("No NAPTR records found") except (dns.resolver.Timeout, exception.DNSException): pass else: self.nameservers = resolver.nameservers return # google resolver failed. fallback to local resolver and schedule another probe for later self.nameservers = local_nameservers self._timer = reactor.callLater(15, self._channel.send, Command('probe_dns')) @run_in_twisted_thread def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SystemIPAddressDidChange(self, notification): self._proc.kill(InterruptCommand) self._channel.send(Command('probe_dns')) def _NH_SystemDidWakeUpFromSleep(self, notification): if self._wakeup_timer is None: def wakeup_action(): self._proc.kill(InterruptCommand) self._channel.send(Command('probe_dns')) self._wakeup_timer = None self._wakeup_timer = reactor.callLater(5, wakeup_action) # wait for system to stabilize diff --git a/sipsimple/session.py b/sipsimple/session.py index b4427239..0e3ebbc2 100644 --- a/sipsimple/session.py +++ b/sipsimple/session.py @@ -1,2751 +1,2754 @@ """ Implements an asynchronous notification based mechanism for establishment, modification and termination of sessions using Session Initiation Protocol (SIP) standardized in RFC3261. """ __all__ = ['Session', 'SessionManager'] import random from threading import RLock from time import time from application.notification import IObserver, Notification, NotificationCenter, NotificationData from application.python import Null, limit from application.python.decorator import decorator, preserve_signature from application.python.types import Singleton from application.system import host from eventlib import api, coros, proc from twisted.internet import reactor from zope.interface import implementer from sipsimple import log from sipsimple.account import AccountManager, BonjourAccount from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.core import DialogID, Engine, Invitation, Referral, Subscription, PJSIPError, SIPCoreError, SIPCoreInvalidStateError, SIPURI, sip_status_messages, sipfrag_re from sipsimple.core import ContactHeader, FromHeader, Header, ReasonHeader, ReferToHeader, ReplacesHeader, RouteHeader, ToHeader, WarningHeader from sipsimple.core import SDPConnection, SDPMediaStream, SDPSession from sipsimple.core import PublicGRUU, PublicGRUUIfAvailable, NoGRUU from sipsimple.lookup import DNSLookup, DNSLookupError from sipsimple.payloads import ParserError from sipsimple.payloads.conference import ConferenceDocument from sipsimple.streams import MediaStreamRegistry, InvalidStreamError, UnknownStreamError from sipsimple.threading import run_in_twisted_thread from sipsimple.threading.green import Command, run_in_green_thread from sipsimple.util import ISOTimestamp class InvitationDisconnectedError(Exception): def __init__(self, invitation, data): self.invitation = invitation self.data = data class MediaStreamDidNotInitializeError(Exception): def __init__(self, stream, data): self.stream = stream self.data = data class MediaStreamDidFailError(Exception): def __init__(self, stream, data): self.stream = stream self.data = data class SubscriptionError(Exception): def __init__(self, error, timeout, **attributes): self.error = error self.timeout = timeout self.attributes = attributes class SIPSubscriptionDidFail(Exception): def __init__(self, data): self.data = data class InterruptSubscription(Exception): pass class TerminateSubscription(Exception): pass class ReferralError(Exception): def __init__(self, error, code=0): self.error = error self.code = code class TerminateReferral(Exception): pass class SIPReferralDidFail(Exception): def __init__(self, data): self.data = data class IllegalStateError(RuntimeError): pass class IllegalDirectionError(RuntimeError): pass class SIPInvitationTransferDidFail(Exception): def __init__(self, data): self.data = data @decorator def transition_state(required_state, new_state): def state_transitioner(func): @preserve_signature(func) def wrapper(obj, *args, **kwargs): with obj._lock: if obj.state != required_state: raise IllegalStateError('cannot call %s in %s state' % (func.__name__, obj.state)) obj.state = new_state return func(obj, *args, **kwargs) return wrapper return state_transitioner @decorator def check_state(required_states): def state_checker(func): @preserve_signature(func) def wrapper(obj, *args, **kwargs): if obj.state not in required_states: raise IllegalStateError('cannot call %s in %s state' % (func.__name__, obj.state)) return func(obj, *args, **kwargs) return wrapper return state_checker @decorator def check_transfer_state(direction, state): def state_checker(func): @preserve_signature(func) def wrapper(obj, *args, **kwargs): if obj.transfer_handler.direction != direction: raise IllegalDirectionError('cannot transfer in %s direction' % obj.transfer_handler.direction) if obj.transfer_handler.state != state: raise IllegalStateError('cannot transfer in %s state' % obj.transfer_handler.state) return func(obj, *args, **kwargs) return wrapper return state_checker class AddParticipantOperation(object): pass class RemoveParticipantOperation(object): pass @implementer(IObserver) class ReferralHandler(object): def __init__(self, session, participant_uri, operation): self.participant_uri = participant_uri if not isinstance(self.participant_uri, SIPURI): if not self.participant_uri.startswith(('sip:', 'sips:')): self.participant_uri = 'sip:%s' % self.participant_uri try: self.participant_uri = SIPURI.parse(self.participant_uri) except SIPCoreError: notification_center = NotificationCenter() if operation is AddParticipantOperation: notification_center.post_notification('SIPConferenceDidNotAddParticipant', sender=session, data=NotificationData(participant=self.participant_uri, code=0, reason='invalid participant URI')) else: notification_center.post_notification('SIPConferenceDidNotRemoveParticipant', sender=session, data=NotificationData(participant=self.participant_uri, code=0, reason='invalid participant URI')) return self.session = session self.operation = operation self.active = False self.route = None self._channel = coros.queue() self._referral = None def start(self): notification_center = NotificationCenter() if not self.session.remote_focus: if self.operation is AddParticipantOperation: notification_center.post_notification('SIPConferenceDidNotAddParticipant', sender=self.session, data=NotificationData(participant=self.participant_uri, code=0, reason='remote endpoint is not a focus')) else: notification_center.post_notification('SIPConferenceDidNotRemoveParticipant', sender=self.session, data=NotificationData(participant=self.participant_uri, code=0, reason='remote endpoint is not a focus')) self.session = None return notification_center.add_observer(self, sender=self.session) notification_center.add_observer(self, name='NetworkConditionsDidChange') proc.spawn(self._run) def _run(self): notification_center = NotificationCenter() settings = SIPSimpleSettings() try: # Lookup routes account = self.session.account if account is BonjourAccount(): uri = SIPURI.new(self.session._invitation.remote_contact_header.uri) elif account.sip.outbound_proxy is not None and account.sip.outbound_proxy.transport in settings.sip.transport_list: uri = SIPURI(host=account.sip.outbound_proxy.host, port=account.sip.outbound_proxy.port, parameters={'transport': account.sip.outbound_proxy.transport}) elif account.sip.always_use_my_proxy: uri = SIPURI(host=account.id.domain) else: uri = SIPURI.new(self.session.remote_identity.uri) lookup = DNSLookup() try: - routes = lookup.lookup_sip_proxy(uri, settings.sip.transport_list).wait() + tls_name = account.sip.tls_name if account is not BonjourAccount() else None + routes = lookup.lookup_sip_proxy(uri, settings.sip.transport_list, tls_name=tls_name).wait() except DNSLookupError as e: timeout = random.uniform(15, 30) raise ReferralError(error='DNS lookup failed: %s' % e) target_uri = SIPURI.new(self.session.remote_identity.uri) timeout = time() + 30 for route in routes: self.route = route remaining_time = timeout - time() if remaining_time > 0: try: contact_uri = account.contact[NoGRUU, route] except KeyError: continue refer_to_header = ReferToHeader(str(self.participant_uri)) refer_to_header.parameters['method'] = 'INVITE' if self.operation is AddParticipantOperation else 'BYE' referral = Referral(target_uri, FromHeader(account.uri, account.display_name), ToHeader(target_uri), refer_to_header, ContactHeader(contact_uri), RouteHeader(route.uri), account.credentials) notification_center.add_observer(self, sender=referral) try: referral.send_refer(timeout=limit(remaining_time, min=1, max=5)) except SIPCoreError: notification_center.remove_observer(self, sender=referral) timeout = 5 raise ReferralError(error='Internal error') self._referral = referral try: while True: notification = self._channel.wait() if notification.name == 'SIPReferralDidStart': break except SIPReferralDidFail as e: notification_center.remove_observer(self, sender=referral) self._referral = None if e.data.code in (403, 405): raise ReferralError(error=sip_status_messages[e.data.code], code=e.data.code) else: # Otherwise just try the next route continue else: break else: self.route = None raise ReferralError(error='No more routes to try') # At this point it is subscribed. Handle notifications and ending/failures. try: self.active = True while True: notification = self._channel.wait() if notification.name == 'SIPReferralGotNotify': if notification.data.event == 'refer' and notification.data.body: match = sipfrag_re.match(notification.data.body) if match: code = int(match.group('code')) reason = match.group('reason') if code/100 > 2: continue if self.operation is AddParticipantOperation: notification_center.post_notification('SIPConferenceGotAddParticipantProgress', sender=self.session, data=NotificationData(participant=self.participant_uri, code=code, reason=reason)) else: notification_center.post_notification('SIPConferenceGotRemoveParticipantProgress', sender=self.session, data=NotificationData(participant=self.participant_uri, code=code, reason=reason)) elif notification.name == 'SIPReferralDidEnd': break except SIPReferralDidFail as e: notification_center.remove_observer(self, sender=self._referral) raise ReferralError(error=e.data.reason, code=e.data.code) else: notification_center.remove_observer(self, sender=self._referral) if self.operation is AddParticipantOperation: notification_center.post_notification('SIPConferenceDidAddParticipant', sender=self.session, data=NotificationData(participant=self.participant_uri)) else: notification_center.post_notification('SIPConferenceDidRemoveParticipant', sender=self.session, data=NotificationData(participant=self.participant_uri)) finally: self.active = False except TerminateReferral: if self._referral is not None: try: self._referral.end(timeout=2) except SIPCoreError: pass else: try: while True: notification = self._channel.wait() if notification.name == 'SIPReferralDidEnd': break except SIPReferralDidFail: pass finally: notification_center.remove_observer(self, sender=self._referral) if self.operation is AddParticipantOperation: notification_center.post_notification('SIPConferenceDidNotAddParticipant', sender=self.session, data=NotificationData(participant=self.participant_uri, code=0, reason='error')) else: notification_center.post_notification('SIPConferenceDidNotRemoveParticipant', sender=self.session, data=NotificationData(participant=self.participant_uri, code=0, reason='error')) except ReferralError as e: if self.operation is AddParticipantOperation: notification_center.post_notification('SIPConferenceDidNotAddParticipant', sender=self.session, data=NotificationData(participant=self.participant_uri, code=e.code, reason=e.error)) else: notification_center.post_notification('SIPConferenceDidNotRemoveParticipant', sender=self.session, data=NotificationData(participant=self.participant_uri, code=e.code, reason=e.error)) finally: notification_center.remove_observer(self, sender=self.session) notification_center.remove_observer(self, name='NetworkConditionsDidChange') self.session = None self._referral = None def _refresh(self): try: contact_header = ContactHeader(self.session.account.contact[NoGRUU, self.route]) except KeyError: pass else: try: self._referral.refresh(contact_header=contact_header, timeout=2) except (SIPCoreError, SIPCoreInvalidStateError): pass @run_in_twisted_thread def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPReferralDidStart(self, notification): self._channel.send(notification) def _NH_SIPReferralDidEnd(self, notification): self._channel.send(notification) def _NH_SIPReferralDidFail(self, notification): self._channel.send_exception(SIPReferralDidFail(notification.data)) def _NH_SIPReferralGotNotify(self, notification): self._channel.send(notification) def _NH_SIPSessionDidFail(self, notification): self._channel.send_exception(TerminateReferral()) def _NH_SIPSessionWillEnd(self, notification): self._channel.send_exception(TerminateReferral()) def _NH_NetworkConditionsDidChange(self, notification): if self.active: self._refresh() @implementer(IObserver) class ConferenceHandler(object): def __init__(self, session): self.session = session self.active = False self.subscribed = False self._command_proc = None self._command_channel = coros.queue() self._data_channel = coros.queue() self._subscription = None self._subscription_proc = None self._subscription_timer = None notification_center = NotificationCenter() notification_center.add_observer(self, sender=self.session) notification_center.add_observer(self, name='NetworkConditionsDidChange') self._command_proc = proc.spawn(self._run) @run_in_green_thread def add_participant(self, participant_uri): referral_handler = ReferralHandler(self.session, participant_uri, AddParticipantOperation) referral_handler.start() @run_in_green_thread def remove_participant(self, participant_uri): referral_handler = ReferralHandler(self.session, participant_uri, RemoveParticipantOperation) referral_handler.start() def _run(self): while True: command = self._command_channel.wait() handler = getattr(self, '_CH_%s' % command.name) handler(command) def _activate(self): self.active = True command = Command('subscribe') self._command_channel.send(command) return command def _deactivate(self): self.active = False command = Command('unsubscribe') self._command_channel.send(command) return command def _resubscribe(self): command = Command('subscribe') self._command_channel.send(command) return command def _terminate(self): notification_center = NotificationCenter() notification_center.remove_observer(self, sender=self.session) notification_center.remove_observer(self, name='NetworkConditionsDidChange') self._deactivate() command = Command('terminate') self._command_channel.send(command) command.wait() self.session = None def _CH_subscribe(self, command): if self._subscription_timer is not None and self._subscription_timer.active(): self._subscription_timer.cancel() self._subscription_timer = None if self._subscription_proc is not None: subscription_proc = self._subscription_proc subscription_proc.kill(InterruptSubscription) subscription_proc.wait() self._subscription_proc = proc.spawn(self._subscription_handler, command) def _CH_unsubscribe(self, command): # Cancel any timer which would restart the subscription process if self._subscription_timer is not None and self._subscription_timer.active(): self._subscription_timer.cancel() self._subscription_timer = None if self._subscription_proc is not None: subscription_proc = self._subscription_proc subscription_proc.kill(TerminateSubscription) subscription_proc.wait() self._subscription_proc = None command.signal() def _CH_terminate(self, command): command.signal() raise proc.ProcExit() def _subscription_handler(self, command): notification_center = NotificationCenter() settings = SIPSimpleSettings() try: # Lookup routes account = self.session.account if account is BonjourAccount(): uri = SIPURI.new(self.session._invitation.remote_contact_header.uri) elif account.sip.outbound_proxy is not None and account.sip.outbound_proxy.transport in settings.sip.transport_list: uri = SIPURI(host=account.sip.outbound_proxy.host, port=account.sip.outbound_proxy.port, parameters={'transport': account.sip.outbound_proxy.transport}) elif account.sip.always_use_my_proxy: uri = SIPURI(host=account.id.domain) else: uri = SIPURI.new(self.session.remote_identity.uri) lookup = DNSLookup() try: - routes = lookup.lookup_sip_proxy(uri, settings.sip.transport_list).wait() + tls_name = account.sip.tls_name if account is not BonjourAccount() else None + routes = lookup.lookup_sip_proxy(uri, settings.sip.transport_list, tls_name=tls_name).wait() except DNSLookupError as e: timeout = random.uniform(15, 30) raise SubscriptionError(error='DNS lookup failed: %s' % e, timeout=timeout) target_uri = SIPURI.new(self.session.remote_identity.uri) default_interval = 600 if account is BonjourAccount() else account.sip.subscribe_interval refresh_interval = getattr(command, 'refresh_interval', default_interval) timeout = time() + 30 for route in routes: remaining_time = timeout - time() if remaining_time > 0: try: contact_uri = account.contact[NoGRUU, route] except KeyError: continue subscription = Subscription(target_uri, FromHeader(account.uri, account.display_name), ToHeader(target_uri), ContactHeader(contact_uri), b'conference', RouteHeader(route.uri), credentials=account.credentials, refresh=refresh_interval) notification_center.add_observer(self, sender=subscription) try: subscription.subscribe(timeout=limit(remaining_time, min=1, max=5)) except SIPCoreError as e: notification_center.remove_observer(self, sender=subscription) timeout = 5 raise SubscriptionError(error='Internal error %s' % str(e), timeout=timeout) self._subscription = subscription try: while True: notification = self._data_channel.wait() if notification.sender is subscription: break except SIPSubscriptionDidFail as e: notification_center.remove_observer(self, sender=subscription) self._subscription = None if e.data.code == 407: # Authentication failed, so retry the subscription in some time timeout = random.uniform(60, 120) raise SubscriptionError(error='Authentication failed', timeout=timeout) elif e.data.code == 423: # Get the value of the Min-Expires header timeout = random.uniform(60, 120) if e.data.min_expires is not None and e.data.min_expires > refresh_interval: raise SubscriptionError(error='Interval too short', timeout=timeout, min_expires=e.data.min_expires) else: raise SubscriptionError(error='Interval too short', timeout=timeout) elif e.data.code in (405, 406, 489, 1400): command.signal(e) return else: # Otherwise just try the next route continue else: self.subscribed = True command.signal() break else: # There are no more routes to try, reschedule the subscription timeout = random.uniform(60, 180) raise SubscriptionError(error='No more routes to try', timeout=timeout) # At this point it is subscribed. Handle notifications and ending/failures. try: while True: notification = self._data_channel.wait() if notification.sender is not self._subscription: continue if notification.name == 'SIPSubscriptionGotNotify': if notification.data.event == 'conference' and notification.data.body: try: conference_info = ConferenceDocument.parse(notification.data.body) except ParserError: pass else: notification_center.post_notification('SIPSessionGotConferenceInfo', sender=self.session, data=NotificationData(conference_info=conference_info)) elif notification.name == 'SIPSubscriptionDidEnd': break except SIPSubscriptionDidFail: self._command_channel.send(Command('subscribe')) notification_center.remove_observer(self, sender=self._subscription) except InterruptSubscription as e: if not self.subscribed: command.signal(e) if self._subscription is not None: notification_center.remove_observer(self, sender=self._subscription) try: self._subscription.end(timeout=2) except SIPCoreError: pass except TerminateSubscription as e: if not self.subscribed: command.signal(e) if self._subscription is not None: try: self._subscription.end(timeout=2) except SIPCoreError: pass else: try: while True: notification = self._data_channel.wait() if notification.sender is self._subscription and notification.name == 'SIPSubscriptionDidEnd': break except SIPSubscriptionDidFail: pass finally: notification_center.remove_observer(self, sender=self._subscription) except SubscriptionError as e: if 'min_expires' in e.attributes: command = Command('subscribe', command.event, refresh_interval=e.attributes['min_expires']) else: command = Command('subscribe', command.event) self._subscription_timer = reactor.callLater(e.timeout, self._command_channel.send, command) finally: self.subscribed = False self._subscription = None self._subscription_proc = None @run_in_twisted_thread def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPSubscriptionDidStart(self, notification): self._data_channel.send(notification) def _NH_SIPSubscriptionDidEnd(self, notification): self._data_channel.send(notification) def _NH_SIPSubscriptionDidFail(self, notification): self._data_channel.send_exception(SIPSubscriptionDidFail(notification.data)) def _NH_SIPSubscriptionGotNotify(self, notification): self._data_channel.send(notification) def _NH_SIPSessionDidStart(self, notification): if self.session.remote_focus: self._activate() @run_in_green_thread def _NH_SIPSessionDidFail(self, notification): self._terminate() @run_in_green_thread def _NH_SIPSessionDidEnd(self, notification): self._terminate() def _NH_SIPSessionDidRenegotiateStreams(self, notification): if self.session.remote_focus and not self.active: self._activate() elif not self.session.remote_focus and self.active: self._deactivate() def _NH_NetworkConditionsDidChange(self, notification): if self.active: self._resubscribe() class TransferInfo(object): def __init__(self, referred_by=None, replaced_dialog_id=None): self.referred_by = referred_by self.replaced_dialog_id = replaced_dialog_id @implementer(IObserver) class TransferHandler(object): def __init__(self, session): self.state = None self.direction = None self.new_session = None self.session = session notification_center = NotificationCenter() notification_center.add_observer(self, sender=self.session) notification_center.add_observer(self, sender=self.session._invitation) self._command_channel = coros.queue() self._data_channel = coros.queue() self._proc = proc.spawn(self._run) self.completed = False def _run(self): while True: command = self._command_channel.wait() handler = getattr(self, '_CH_%s' % command.name) handler(command) self.direction = None self.state = None def _CH_incoming_transfer(self, command): self.direction = 'incoming' notification_center = NotificationCenter() refer_to_hdr = command.data.headers.get('Refer-To') target = SIPURI.parse(refer_to_hdr.uri) referred_by_hdr = command.data.headers.get('Referred-By', None) if referred_by_hdr is not None: origin = referred_by_hdr.body else: origin = str(self.session.remote_identity.uri) try: while True: try: notification = self._data_channel.wait() except SIPInvitationTransferDidFail: self.state = 'failed' return else: if notification.name == 'SIPInvitationTransferDidStart': self.state = 'starting' refer_to_uri = SIPURI.new(target) refer_to_uri.headers = {} refer_to_uri.parameters = {} notification_center.post_notification('SIPSessionTransferNewIncoming', self.session, NotificationData(transfer_destination=refer_to_uri)) elif notification.name == 'SIPSessionTransferDidStart': break elif notification.name == 'SIPSessionTransferDidFail': self.state = 'failed' try: self.session._invitation.notify_transfer_progress(notification.data.code, notification.data.reason) except SIPCoreError: return while True: try: notification = self._data_channel.wait() except SIPInvitationTransferDidFail: return self.state = 'started' transfer_info = TransferInfo(referred_by=origin) try: replaces_hdr = target.headers.pop('Replaces') call_id, rest = replaces_hdr.split(';', 1) params = dict((item.split('=') for item in rest.split(';'))) to_tag = params.get('to-tag') from_tag = params.get('from-tag') except (KeyError, ValueError): pass else: transfer_info.replaced_dialog_id = DialogID(call_id, local_tag=from_tag, remote_tag=to_tag) settings = SIPSimpleSettings() account = self.session.account if account is BonjourAccount(): uri = target elif account.sip.outbound_proxy is not None and account.sip.outbound_proxy.transport in settings.sip.transport_list: uri = SIPURI(host=account.sip.outbound_proxy.host, port=account.sip.outbound_proxy.port, parameters={'transport': account.sip.outbound_proxy.transport}) elif account.sip.always_use_my_proxy: uri = SIPURI(host=account.id.domain) else: uri = target lookup = DNSLookup() try: - routes = lookup.lookup_sip_proxy(uri, settings.sip.transport_list).wait() + tls_name = account.sip.tls_name if account is not BonjourAccount() else None + routes = lookup.lookup_sip_proxy(uri, settings.sip.transport_list, tls_name=tls_name).wait() except DNSLookupError as e: self.state = 'failed' notification_center.post_notification('SIPSessionTransferDidFail', sender=self.session, data=NotificationData(code=0, reason="DNS lookup failed: {}".format(e))) try: self.session._invitation.notify_transfer_progress(480) except SIPCoreError: return while True: try: self._data_channel.wait() except SIPInvitationTransferDidFail: return return self.new_session = Session(account) notification_center = NotificationCenter() notification_center.add_observer(self, sender=self.new_session) self.new_session.connect(ToHeader(target), routes=routes, streams=[MediaStreamRegistry.AudioStream()], transfer_info=transfer_info) while True: try: notification = self._data_channel.wait() except SIPInvitationTransferDidFail: return if notification.name == 'SIPInvitationTransferDidEnd': return except proc.ProcExit: if self.new_session is not None: notification_center.remove_observer(self, sender=self.new_session) self.new_session = None raise def _CH_outgoing_transfer(self, command): self.direction = 'outgoing' notification_center = NotificationCenter() self.completed = False self.state = 'starting' while True: try: notification = self._data_channel.wait() except SIPInvitationTransferDidFail as e: self.state = None self.direction = None notification_center.post_notification('SIPSessionTransferDidFail', sender=self.session, data=NotificationData(code=e.data.code, reason=e.data.reason)) return if notification.name == 'SIPInvitationTransferDidStart': self.state = 'started' notification_center.post_notification('SIPSessionTransferDidStart', sender=self.session) elif notification.name == 'SIPInvitationTransferDidEnd': self.state = None self.direction = None self.session.end() notification_center.post_notification('SIPSessionTransferDidEnd', sender=self.session) return def _terminate(self): notification_center = NotificationCenter() notification_center.remove_observer(self, sender=self.session._invitation) notification_center.remove_observer(self, sender=self.session) self._proc.kill() self._proc = None self._command_channel = None self._data_channel = None self.session = None @run_in_twisted_thread def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPInvitationTransferNewIncoming(self, notification): self._command_channel.send(Command('incoming_transfer', data=notification.data)) def _NH_SIPInvitationTransferNewOutgoing(self, notification): self._command_channel.send(Command('outgoing_transfer', data=notification.data)) def _NH_SIPInvitationTransferDidStart(self, notification): self._data_channel.send(notification) def _NH_SIPInvitationTransferDidFail(self, notification): self.direction = None if not self.completed: self._data_channel.send_exception(SIPInvitationTransferDidFail(notification.data)) def _NH_SIPInvitationTransferDidEnd(self, notification): self.direction = None self._data_channel.send(notification) def _NH_SIPInvitationTransferGotNotify(self, notification): if notification.data.event == 'refer' and notification.data.body: match = sipfrag_re.match(notification.data.body) if match: code = int(match.group('code')) reason = match.group('reason') notification.center.post_notification('SIPSessionTransferGotProgress', sender=self.session, data=NotificationData(code=code, reason=reason)) if code == 200: self.completed = True self.direction = None self.state = None notification.center.post_notification('SIPSessionTransferDidEnd', sender=self.session) elif code >= 400: self.state = None self.direction = None notification.center.post_notification('SIPSessionTransferDidFail', sender=self.session, data=NotificationData(code=code, reason=reason)) def _NH_SIPSessionTransferDidStart(self, notification): if notification.sender is self.session and self.state == 'starting': self._data_channel.send(notification) def _NH_SIPSessionTransferDidFail(self, notification): if notification.sender is self.session and self.state == 'starting': self._data_channel.send(notification) def _NH_SIPSessionGotRingIndication(self, notification): if notification.sender is self.new_session and self.session is not None: try: self.session._invitation.notify_transfer_progress(180) except SIPCoreError: pass def _NH_SIPSessionGotProvisionalResponse(self, notification): if notification.sender is self.new_session and self.session is not None: try: self.session._invitation.notify_transfer_progress(notification.data.code, notification.data.reason) except SIPCoreError: pass def _NH_SIPSessionDidStart(self, notification): if notification.sender is self.new_session: notification.center.remove_observer(self, sender=notification.sender) self.new_session = None if self.session is not None: notification.center.post_notification('SIPSessionTransferDidEnd', sender=self.session) if self.state == 'started': try: self.session._invitation.notify_transfer_progress(200) except SIPCoreError: pass self.state = 'ended' self.session.end() def _NH_SIPSessionDidEnd(self, notification): if notification.sender is self.new_session: # If any stream fails to start we won't get SIPSessionDidFail, we'll get here instead notification.center.remove_observer(self, sender=notification.sender) self.new_session = None if self.session is not None: notification.center.post_notification('SIPSessionTransferDidFail', sender=self.session, data=NotificationData(code=500, reason='internal error')) if self.state == 'started': try: self.session._invitation.notify_transfer_progress(500) except SIPCoreError: pass self.state = 'failed' else: self._terminate() def _NH_SIPSessionDidFail(self, notification): if notification.sender is self.new_session: notification.center.remove_observer(self, sender=notification.sender) self.new_session = None if self.session is not None: notification.center.post_notification('SIPSessionTransferDidFail', sender=self.session, data=NotificationData(code=notification.data.code or 500, reason=notification.data.reason)) if self.state == 'started': try: self.session._invitation.notify_transfer_progress(notification.data.code or 500, notification.data.reason) except SIPCoreError: pass self.state = 'failed' else: self._terminate() class OptionalTag(str): def __eq__(self, other): return other is None or super(OptionalTag, self).__eq__(other) def __ne__(self, other): return not self == other def __repr__(self): return '{}({})'.format(self.__class__.__name__, super(OptionalTag, self).__repr__()) @implementer(IObserver) class SessionReplaceHandler(object): def __init__(self, session): self.session = session def start(self): notification_center = NotificationCenter() notification_center.add_observer(self, sender=self.session) notification_center.add_observer(self, sender=self.session.replaced_session) def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPSessionDidStart(self, notification): notification.center.remove_observer(self, sender=self.session) notification.center.remove_observer(self, sender=self.session.replaced_session) self.session.replaced_session.end() self.session.replaced_session = None self.session = None def _NH_SIPSessionDidFail(self, notification): if notification.sender is self.session: notification.center.remove_observer(self, sender=self.session) notification.center.remove_observer(self, sender=self.session.replaced_session) self.session.replaced_session = None self.session = None _NH_SIPSessionDidEnd = _NH_SIPSessionDidFail @implementer(IObserver) class Session(object): media_stream_timeout = 15 short_reinvite_timeout = 5 def __init__(self, account): self.account = account self.direction = None self.end_time = None self.on_hold = False self.proposed_streams = None self.route = None self.state = None self.start_time = None self.streams = None self.transport = None self.local_focus = False self.remote_focus = False self.greenlet = None self.conference = None self.replaced_session = None self.transfer_handler = None self.transfer_info = None self._channel = coros.queue() self._hold_in_progress = False self._invitation = None self._local_identity = None self._remote_identity = None self._lock = RLock() def init_incoming(self, invitation, data): notification_center = NotificationCenter() remote_sdp = invitation.sdp.proposed_remote self.proposed_streams = [] if remote_sdp: for index, media_stream in enumerate(remote_sdp.media): if media_stream.port != 0: for stream_type in MediaStreamRegistry: try: stream = stream_type.new_from_sdp(self, remote_sdp, index) except UnknownStreamError as e: continue except InvalidStreamError as e: log.error("Invalid stream: {}".format(e)) break except Exception as e: log.exception("Exception occurred while setting up stream from SDP: {}".format(e)) break else: stream.index = index self.proposed_streams.append(stream) break self.direction = 'incoming' self.state = 'incoming' self.transport = invitation.transport.lower() self._invitation = invitation self.conference = ConferenceHandler(self) self.transfer_handler = TransferHandler(self) if 'isfocus' in invitation.remote_contact_header.parameters: self.remote_focus = True if 'Referred-By' in data.headers or 'Replaces' in data.headers: self.transfer_info = TransferInfo() if 'Referred-By' in data.headers: self.transfer_info.referred_by = data.headers['Referred-By'].body if 'Replaces' in data.headers: replaces_header = data.headers.get('Replaces') # Because we only allow the remote tag to be optional, it can only match established dialogs and early outgoing dialogs, but not early incoming dialogs, # which according to RFC3891 should be rejected with 481 (which will happen automatically by never matching them). if replaces_header.early_only or replaces_header.from_tag == '0': replaced_dialog_id = DialogID(replaces_header.call_id, local_tag=replaces_header.to_tag, remote_tag=OptionalTag(replaces_header.from_tag)) else: replaced_dialog_id = DialogID(replaces_header.call_id, local_tag=replaces_header.to_tag, remote_tag=replaces_header.from_tag) session_manager = SessionManager() try: replaced_session = next(session for session in session_manager.sessions if session.dialog_id == replaced_dialog_id) except StopIteration: invitation.send_response(481) return else: # Any matched dialog at this point is either established, terminated or early outgoing. if replaced_session.state in ('terminating', 'terminated'): invitation.send_response(603) return elif replaced_session.dialog_id.remote_tag is not None and replaces_header.early_only: # The replaced dialog is established, but the early-only flag is set invitation.send_response(486) return self.replaced_session = replaced_session self.transfer_info.replaced_dialog_id = replaced_dialog_id replace_handler = SessionReplaceHandler(self) replace_handler.start() notification_center.add_observer(self, sender=invitation) notification_center.post_notification('SIPSessionNewIncoming', sender=self, data=NotificationData(streams=self.proposed_streams[:], headers=data.headers)) @transition_state(None, 'connecting') @run_in_green_thread def connect(self, to_header, routes, streams, is_focus=False, transfer_info=None, extra_headers=None): self.greenlet = api.getcurrent() notification_center = NotificationCenter() settings = SIPSimpleSettings() connected = False received_code = 0 received_reason = None unhandled_notifications = [] extra_headers = extra_headers or [] if {'to', 'from', 'via', 'contact', 'route', 'record-route'}.intersection(header.name.lower() for header in extra_headers): raise RuntimeError('invalid header in extra_headers: To, From, Via, Contact, Route and Record-Route headers are not allowed') self.direction = 'outgoing' self.proposed_streams = streams self.route = routes[0] self.transport = self.route.transport.lower() self.local_focus = is_focus self._invitation = Invitation() self._local_identity = FromHeader(self.account.uri, self.account.display_name) self._remote_identity = to_header self.conference = ConferenceHandler(self) self.transfer_handler = TransferHandler(self) self.transfer_info = transfer_info notification_center.add_observer(self, sender=self._invitation) notification_center.post_notification('SIPSessionNewOutgoing', sender=self, data=NotificationData(streams=streams[:])) for stream in self.proposed_streams: notification_center.add_observer(self, sender=stream) stream.initialize(self, direction='outgoing') try: wait_count = len(self.proposed_streams) while wait_count > 0: notification = self._channel.wait() if notification.name == 'MediaStreamDidInitialize': wait_count -= 1 try: contact_uri = self.account.contact[PublicGRUUIfAvailable, self.route] local_ip = host.outgoing_ip_for(self.route.address) if local_ip is None: raise ValueError("could not get outgoing IP address") except (KeyError, ValueError) as e: for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self._fail(originator='local', code=480, reason=sip_status_messages[480], error=str(e)) return connection = SDPConnection(local_ip.encode()) local_sdp = SDPSession(local_ip.encode(), name=settings.user_agent.encode()) for index, stream in enumerate(self.proposed_streams): stream.index = index media = stream.get_local_media(remote_sdp=None, index=index) if media.connection is None or (media.connection is not None and not media.has_ice_attributes and not media.has_ice_candidates): media.connection = connection local_sdp.media.append(media) from_header = FromHeader(self.account.uri, self.account.display_name) route_header = RouteHeader(self.route.uri) contact_header = ContactHeader(contact_uri) if is_focus: contact_header.parameters['isfocus'] = None if self.transfer_info is not None: if self.transfer_info.referred_by is not None: extra_headers.append(Header('Referred-By', self.transfer_info.referred_by)) if self.transfer_info.replaced_dialog_id is not None: dialog_id = self.transfer_info.replaced_dialog_id extra_headers.append(ReplacesHeader(dialog_id.call_id, dialog_id.local_tag, dialog_id.remote_tag)) self._invitation.send_invite(to_header.uri, from_header, to_header, route_header, contact_header, local_sdp, self.account.credentials, extra_headers) try: with api.timeout(settings.sip.invite_timeout): while True: notification = self._channel.wait() if notification.name == 'SIPInvitationGotSDPUpdate': if notification.data.succeeded: local_sdp = notification.data.local_sdp remote_sdp = notification.data.remote_sdp break else: for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self._fail(originator='remote', code=0, reason=None, error='SDP negotiation failed: %s' % notification.data.error) return elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'early': if notification.data.code == 180: notification_center.post_notification('SIPSessionGotRingIndication', self) notification_center.post_notification('SIPSessionGotProvisionalResponse', self, NotificationData(code=notification.data.code, reason=notification.data.reason)) elif notification.data.state == 'connecting': received_code = notification.data.code received_reason = notification.data.reason elif notification.data.state == 'connected': if not connected: connected = True notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=received_code, reason=received_reason)) else: unhandled_notifications.append(notification) elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) except api.TimeoutError: self.end() return notification_center.post_notification('SIPSessionWillStart', sender=self) stream_map = dict((stream.index, stream) for stream in self.proposed_streams) for index, local_media in enumerate(local_sdp.media): remote_media = remote_sdp.media[index] stream = stream_map[index] if remote_media.port: # TODO: check if port is also 0 in local_sdp. In that case PJSIP disabled the stream because # negotiation failed. If there are more streams, however, the negotiation is considered successful as a # whole, so while we built a normal SDP, PJSIP modified it and sent it to the other side. That's kind io # OK, but we cannot really start the stream. -Saul stream.start(local_sdp, remote_sdp, index) else: notification_center.remove_observer(self, sender=stream) self.proposed_streams.remove(stream) del stream_map[stream.index] stream.deactivate() stream.end() removed_streams = [stream for stream in self.proposed_streams if stream.index >= len(local_sdp.media)] for stream in removed_streams: notification_center.remove_observer(self, sender=stream) self.proposed_streams.remove(stream) del stream_map[stream.index] stream.deactivate() stream.end() invitation_notifications = [] with api.timeout(self.media_stream_timeout): wait_count = len(self.proposed_streams) while wait_count > 0: notification = self._channel.wait() if notification.name == 'MediaStreamDidStart': wait_count -= 1 elif notification.name == 'SIPInvitationChangedState': invitation_notifications.append(notification) for notification in invitation_notifications: self._channel.send(notification) while not connected or self._channel: notification = self._channel.wait() if notification.name == 'SIPInvitationChangedState': if notification.data.state == 'early': if notification.data.code == 180: notification_center.post_notification('SIPSessionGotRingIndication', self) notification_center.post_notification('SIPSessionGotProvisionalResponse', self, NotificationData(code=notification.data.code, reason=notification.data.reason)) elif notification.data.state == 'connecting': received_code = notification.data.code received_reason = notification.data.reason elif notification.data.state == 'connected': if not connected: connected = True notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=received_code, reason=received_reason)) else: unhandled_notifications.append(notification) elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) except (MediaStreamDidNotInitializeError, MediaStreamDidFailError, api.TimeoutError) as e: for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() if isinstance(e, api.TimeoutError): error = 'media stream timed-out while starting' elif isinstance(e, MediaStreamDidNotInitializeError): error = 'media stream did not initialize: %s' % e.data.reason else: error = '%s media stream failed: %s' % (e.stream.type, e.data.reason) self._fail(originator='local', code=0, reason=None, error=error) except InvitationDisconnectedError as e: notification_center.remove_observer(self, sender=self._invitation) for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self.state = 'terminated' # As weird as it may sound, PJSIP accepts a BYE even without receiving a final response to the INVITE if e.data.prev_state in ('connecting', 'connected') or getattr(e.data, 'method', None) == 'BYE': notification_center.post_notification('SIPSessionWillEnd', self, NotificationData(originator=e.data.originator)) if e.data.originator == 'remote': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method=e.data.method, code=200, reason=sip_status_messages[200])) self.end_time = ISOTimestamp.now() notification_center.post_notification('SIPSessionDidEnd', self, NotificationData(originator=e.data.originator, end_reason=e.data.disconnect_reason)) else: if e.data.originator == 'remote': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=e.data.code, reason=e.data.reason)) code = e.data.code reason = e.data.reason elif e.data.disconnect_reason == 'timeout': code = 408 reason = 'timeout' else: # TODO: we should know *exactly* when there are set -Saul code = getattr(e.data, 'code', 0) reason = getattr(e.data, 'reason', 'Session disconnected') if e.data.originator == 'remote' and code // 100 == 3: redirect_identities = e.data.headers.get('Contact', []) else: redirect_identities = None notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator=e.data.originator, code=code, reason=reason, failure_reason=e.data.disconnect_reason, redirect_identities=redirect_identities)) self.greenlet = None except SIPCoreError as e: for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self._fail(originator='local', code=0, reason=None, error='SIP core error: %s' % str(e)) else: self.greenlet = None self.state = 'connected' self.streams = self.proposed_streams self.proposed_streams = None self.start_time = ISOTimestamp.now() any_stream_ice = any(getattr(stream, 'ice_active', False) for stream in self.streams) if any_stream_ice: self._reinvite_after_ice() notification_center.post_notification('SIPSessionDidStart', self, NotificationData(streams=self.streams[:])) for notification in unhandled_notifications: self.handle_notification(notification) if self._hold_in_progress: self._send_hold() def _reinvite_after_ice(self): # This function does not do any error checking, it's designed to be called at the end of connect and add_stream self.state = 'sending_proposal' self.greenlet = api.getcurrent() notification_center = NotificationCenter() local_sdp = SDPSession.new(self._invitation.sdp.active_local) local_sdp.version += 1 for index, stream in enumerate(self.streams): local_sdp.media[index] = stream.get_local_media(remote_sdp=None, index=index) self._invitation.send_reinvite(sdp=local_sdp) received_invitation_state = False received_sdp_update = False try: with api.timeout(self.short_reinvite_timeout): while not received_invitation_state or not received_sdp_update: notification = self._channel.wait() if notification.name == 'SIPInvitationGotSDPUpdate': received_sdp_update = True if notification.data.succeeded: local_sdp = notification.data.local_sdp remote_sdp = notification.data.remote_sdp for index, stream in enumerate(self.streams): stream.update(local_sdp, remote_sdp, index) else: return elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': received_invitation_state = True notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=notification.data.code, reason=notification.data.reason)) elif notification.data.state == 'disconnected': self.end() return except Exception: pass finally: self.state = 'connected' self.greenlet = None @check_state(['incoming', 'received_proposal']) @run_in_green_thread def send_ring_indication(self): try: self._invitation.send_response(180) except SIPCoreInvalidStateError: pass # The INVITE session might have already been cancelled; ignore the error @transition_state('incoming', 'accepting') @run_in_green_thread def accept(self, streams, is_focus=False, extra_headers=None): self.greenlet = api.getcurrent() notification_center = NotificationCenter() settings = SIPSimpleSettings() self.local_focus = is_focus connected = False unhandled_notifications = [] extra_headers = extra_headers or [] if {'to', 'from', 'via', 'contact', 'route', 'record-route'}.intersection(header.name.lower() for header in extra_headers): raise RuntimeError('invalid header in extra_headers: To, From, Via, Contact, Route and Record-Route headers are not allowed') if self.proposed_streams: for stream in self.proposed_streams: if stream in streams: notification_center.add_observer(self, sender=stream) stream.initialize(self, direction='incoming') else: for index, stream in enumerate(streams): notification_center.add_observer(self, sender=stream) stream.index = index stream.initialize(self, direction='outgoing') self.proposed_streams = streams wait_count = len(self.proposed_streams) try: while wait_count > 0: notification = self._channel.wait() if notification.name == 'MediaStreamDidInitialize': wait_count -= 1 remote_sdp = self._invitation.sdp.proposed_remote sdp_connection = remote_sdp.connection or next((media.connection for media in remote_sdp.media if media.connection is not None)) sdp_ip = sdp_connection.address.decode() if isinstance(sdp_connection.address, bytes) else sdp_connection.address local_ip = host.outgoing_ip_for(sdp_ip) if sdp_ip != '0.0.0.0' else sdp_ip if local_ip is None: for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self._fail(originator='local', code=500, reason=sip_status_messages[500], error='could not get local IP address') return connection = SDPConnection(local_ip.encode()) local_sdp = SDPSession(local_ip.encode(), name=settings.user_agent.encode()) if remote_sdp: stream_map = dict((stream.index, stream) for stream in self.proposed_streams) for index, media in enumerate(remote_sdp.media): stream = stream_map.get(index, None) if stream is not None: # TODO: broken for RTP streams here media = stream.get_local_media(remote_sdp=remote_sdp, index=index) if not media.has_ice_attributes and not media.has_ice_candidates: media.connection = connection else: media = SDPMediaStream.new(media) media.connection = connection media.port = 0 media.attributes = [] media.bandwidth_info = [] local_sdp.media.append(media) else: for index, stream in enumerate(self.proposed_streams): stream.index = index media = stream.get_local_media(remote_sdp=None, index=index) if media.connection is None or (media.connection is not None and not media.has_ice_attributes and not media.has_ice_candidates): media.connection = connection local_sdp.media.append(media) contact_header = ContactHeader.new(self._invitation.local_contact_header) try: local_contact_uri = self.account.contact[PublicGRUU, self._invitation.transport] except KeyError: pass else: contact_header.uri = local_contact_uri if is_focus: contact_header.parameters['isfocus'] = None self._invitation.send_response(200, contact_header=contact_header, sdp=local_sdp, extra_headers=extra_headers) notification_center.post_notification('SIPSessionWillStart', sender=self) # Local and remote SDPs will be set after the 200 OK is sent while True: notification = self._channel.wait() if notification.name == 'SIPInvitationGotSDPUpdate': if notification.data.succeeded: local_sdp = notification.data.local_sdp remote_sdp = notification.data.remote_sdp break else: if not connected: # we could not have got a SIPInvitationGotSDPUpdate if we did not get an ACK connected = True notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=200, reason=sip_status_messages[200], ack_received=True)) for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self._fail(originator='remote', code=0, reason=None, error='SDP negotiation failed: %s' % notification.data.error) return elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected': if not connected: connected = True notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=200, reason=sip_status_messages[200], ack_received=True)) elif notification.data.prev_state == 'connected': unhandled_notifications.append(notification) elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) wait_count = 0 stream_map = dict((stream.index, stream) for stream in self.proposed_streams) for index, local_media in enumerate(local_sdp.media): remote_media = remote_sdp.media[index] stream = stream_map.get(index, None) if stream is not None: if remote_media.port: wait_count += 1 stream.start(local_sdp, remote_sdp, index) else: notification_center.remove_observer(self, sender=stream) self.proposed_streams.remove(stream) del stream_map[stream.index] stream.deactivate() stream.end() removed_streams = [stream for stream in self.proposed_streams if stream.index >= len(local_sdp.media)] for stream in removed_streams: notification_center.remove_observer(self, sender=stream) self.proposed_streams.remove(stream) del stream_map[stream.index] stream.deactivate() stream.end() with api.timeout(self.media_stream_timeout): while wait_count > 0 or not connected or self._channel: notification = self._channel.wait() if notification.name == 'MediaStreamDidStart': wait_count -= 1 elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected': if not connected: connected = True notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=200, reason='OK', ack_received=True)) elif notification.data.prev_state == 'connected': unhandled_notifications.append(notification) elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) else: unhandled_notifications.append(notification) except (MediaStreamDidNotInitializeError, MediaStreamDidFailError, api.TimeoutError) as e: if self._invitation.state == 'connecting': ack_received = False if isinstance(e, api.TimeoutError) and wait_count == 0 else 'unknown' # pjsip's invite session object does not inform us whether the ACK was received or not notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=200, reason='OK', ack_received=ack_received)) elif self._invitation.state == 'connected' and not connected: # we didn't yet get to process the SIPInvitationChangedState (state -> connected) notification notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=200, reason='OK', ack_received=True)) for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() reason_header = None if isinstance(e, api.TimeoutError): if wait_count > 0: error = 'media stream timed-out while starting' else: error = 'No ACK received' reason_header = ReasonHeader('SIP') reason_header.cause = 500 reason_header.text = 'Missing ACK' elif isinstance(e, MediaStreamDidNotInitializeError): error = 'media stream did not initialize: %s' % e.data.reason reason_header = ReasonHeader('SIP') reason_header.cause = 500 reason_header.text = 'media stream did not initialize' else: error = '%s media stream failed: %s' % (e.stream.type, e.data.reason) reason_header = ReasonHeader('SIP') reason_header.cause = 500 reason_header.text = 'media stream failed to start' self.start_time = ISOTimestamp.now() if self._invitation.state in ('incoming', 'early'): self._fail(originator='local', code=500, reason=sip_status_messages[500], error=error, reason_header=reason_header) else: self._fail(originator='local', code=0, reason=None, error=error, reason_header=reason_header) except InvitationDisconnectedError as e: notification_center.remove_observer(self, sender=self._invitation) for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self.state = 'terminated' if e.data.prev_state in ('incoming', 'early'): notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=487, reason='Session Cancelled', ack_received='unknown')) notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator='remote', code=487, reason='Session Cancelled', failure_reason=e.data.disconnect_reason, redirect_identities=None)) elif e.data.prev_state == 'connecting' and e.data.disconnect_reason == 'missing ACK': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=200, reason='OK', ack_received=False)) notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator='local', code=200, reason=sip_status_messages[200], failure_reason=e.data.disconnect_reason, redirect_identities=None)) else: notification_center.post_notification('SIPSessionWillEnd', self, NotificationData(originator='remote')) notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method=getattr(e.data, 'method', 'INVITE'), code=200, reason='OK')) self.end_time = ISOTimestamp.now() notification_center.post_notification('SIPSessionDidEnd', self, NotificationData(originator='remote', end_reason=e.data.disconnect_reason)) self.greenlet = None except SIPCoreInvalidStateError: # the only reason for which this error can be thrown is if invitation.send_response was called after the INVITE session was cancelled by the remote party notification_center.remove_observer(self, sender=self._invitation) for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self.greenlet = None self.state = 'terminated' notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=487, reason='Session Cancelled', ack_received='unknown')) notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator='remote', code=487, reason='Session Cancelled', failure_reason='user request', redirect_identities=None)) except SIPCoreError as e: for stream in self.proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self._fail(originator='local', code=500, reason=sip_status_messages[500], error='SIP core error: %s' % str(e)) else: self.greenlet = None self.state = 'connected' self.streams = self.proposed_streams self.proposed_streams = None self.start_time = ISOTimestamp.now() notification_center.post_notification('SIPSessionDidStart', self, NotificationData(streams=self.streams[:])) for notification in unhandled_notifications: self.handle_notification(notification) if self._hold_in_progress: self._send_hold() finally: self.greenlet = None @transition_state('incoming', 'terminating') @run_in_green_thread def reject(self, code=603, reason=None): self.greenlet = api.getcurrent() notification_center = NotificationCenter() try: self._invitation.send_response(code, reason) with api.timeout(1): while True: notification = self._channel.wait() if notification.name == 'SIPInvitationChangedState': if notification.data.state == 'disconnected': ack_received = notification.data.disconnect_reason != 'missing ACK' notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=code, reason=sip_status_messages[code], ack_received=ack_received)) break except SIPCoreInvalidStateError: # the only reason for which this error can be thrown is if invitation.send_response was called after the INVITE session was cancelled by the remote party self.greenlet = None except SIPCoreError as e: self._fail(originator='local', code=500, reason=sip_status_messages[500], error='SIP core error: %s' % str(e)) except api.TimeoutError: notification_center.remove_observer(self, sender=self._invitation) self.greenlet = None self.state = 'terminated' notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=code, reason=sip_status_messages[code], ack_received=False)) notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator='local', code=code, reason=sip_status_messages[code], failure_reason='timeout', redirect_identities=None)) else: notification_center.remove_observer(self, sender=self._invitation) self.greenlet = None self.state = 'terminated' notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator='local', code=code, reason=sip_status_messages[code], failure_reason='user request', redirect_identities=None)) finally: self.greenlet = None @transition_state('received_proposal', 'accepting_proposal') @run_in_green_thread def accept_proposal(self, streams): self.greenlet = api.getcurrent() notification_center = NotificationCenter() unhandled_notifications = [] streams = [stream for stream in streams if stream in self.proposed_streams] for stream in streams: notification_center.add_observer(self, sender=stream) stream.initialize(self, direction='incoming') try: wait_count = len(streams) while wait_count > 0: notification = self._channel.wait() if notification.name == 'MediaStreamDidInitialize': wait_count -= 1 local_sdp = SDPSession.new(self._invitation.sdp.active_local) local_sdp.version += 1 remote_sdp = self._invitation.sdp.proposed_remote connection = SDPConnection(local_sdp.address) stream_map = dict((stream.index, stream) for stream in streams) for index, media in enumerate(remote_sdp.media): stream = stream_map.get(index, None) if stream is not None: media = stream.get_local_media(remote_sdp=remote_sdp, index=index) if not media.has_ice_attributes and not media.has_ice_candidates: media.connection = connection if index < len(local_sdp.media): local_sdp.media[index] = media else: local_sdp.media.append(media) elif index >= len(local_sdp.media): # actually == is sufficient media = SDPMediaStream.new(media) media.connection = connection media.port = 0 media.attributes = [] media.bandwidth_info = [] local_sdp.media.append(media) self._invitation.send_response(200, sdp=local_sdp) prev_on_hold_streams = set(stream for stream in self.streams if stream.hold_supported and stream.on_hold_by_remote) received_invitation_state = False received_sdp_update = False while not received_invitation_state or not received_sdp_update: notification = self._channel.wait() if notification.name == 'SIPInvitationGotSDPUpdate': received_sdp_update = True if notification.data.succeeded: local_sdp = notification.data.local_sdp remote_sdp = notification.data.remote_sdp for stream in self.streams: stream.update(local_sdp, remote_sdp, stream.index) else: self._fail_proposal(originator='remote', error='SDP negotiation failed: %s' % notification.data.error) return elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=200, reason=sip_status_messages[200], ack_received='unknown')) received_invitation_state = True elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) on_hold_streams = set(stream for stream in self.streams if stream.hold_supported and stream.on_hold_by_remote) if on_hold_streams != prev_on_hold_streams: hold_supported_streams = (stream for stream in self.streams if stream.hold_supported) notification_center.post_notification('SIPSessionDidChangeHoldState', self, NotificationData(originator='remote', on_hold=bool(on_hold_streams), partial=bool(on_hold_streams) and any(not stream.on_hold_by_remote for stream in hold_supported_streams))) for stream in streams: # TODO: check if port is 0 in local_sdp. In that case PJSIP disabled the stream because # negotiation failed. If there are more streams, however, the negotiation is considered successful as a # whole, so while we built a normal SDP, PJSIP modified it and sent it to the other side. That's kind of # OK, but we cannot really start the stream. -Saul stream.start(local_sdp, remote_sdp, stream.index) with api.timeout(self.media_stream_timeout): wait_count = len(streams) while wait_count > 0 or self._channel: notification = self._channel.wait() if notification.name == 'MediaStreamDidStart': wait_count -= 1 else: unhandled_notifications.append(notification) except api.TimeoutError: self._fail_proposal(originator='remote', error='media stream timed-out while starting') except MediaStreamDidNotInitializeError as e: self._fail_proposal(originator='remote', error='media stream did not initialize: {.data.reason}'.format(e)) except MediaStreamDidFailError as e: self._fail_proposal(originator='remote', error='media stream failed: {.data.reason}'.format(e)) except InvitationDisconnectedError as e: self._fail_proposal(originator='remote', error='session ended') notification = Notification('SIPInvitationChangedState', e.invitation, e.data) notification.center = notification_center self.handle_notification(notification) except SIPCoreError as e: self._fail_proposal(originator='remote', error='SIP core error: %s' % str(e)) else: proposed_streams = self.proposed_streams self.proposed_streams = None self.streams = self.streams + streams self.greenlet = None self.state = 'connected' notification_center.post_notification('SIPSessionProposalAccepted', self, NotificationData(originator='remote', accepted_streams=streams, proposed_streams=proposed_streams)) notification_center.post_notification('SIPSessionDidRenegotiateStreams', self, NotificationData(originator='remote', added_streams=streams, removed_streams=[])) for notification in unhandled_notifications: self.handle_notification(notification) if self._hold_in_progress: self._send_hold() finally: self.greenlet = None @transition_state('received_proposal', 'rejecting_proposal') @run_in_green_thread def reject_proposal(self, code=488, reason=None): self.greenlet = api.getcurrent() notification_center = NotificationCenter() try: self._invitation.send_response(code, reason) with api.timeout(1, None): while True: notification = self._channel.wait() if notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=code, reason=sip_status_messages[code], ack_received='unknown')) break except SIPCoreError as e: self._fail_proposal(originator='remote', error='SIP core error: %s' % str(e)) else: proposed_streams = self.proposed_streams self.proposed_streams = None self.greenlet = None self.state = 'connected' notification_center.post_notification('SIPSessionProposalRejected', self, NotificationData(originator='remote', code=code, reason=sip_status_messages[code], proposed_streams=proposed_streams)) if self._hold_in_progress: self._send_hold() finally: self.greenlet = None def add_stream(self, stream): self.add_streams([stream]) @transition_state('connected', 'sending_proposal') @run_in_green_thread def add_streams(self, streams): streams = list(set(streams).difference(self.streams)) if not streams: self.state = 'connected' return self.greenlet = api.getcurrent() notification_center = NotificationCenter() settings = SIPSimpleSettings() unhandled_notifications = [] self.proposed_streams = streams for stream in self.proposed_streams: notification_center.add_observer(self, sender=stream) stream.initialize(self, direction='outgoing') try: wait_count = len(self.proposed_streams) while wait_count > 0: notification = self._channel.wait() if notification.name == 'MediaStreamDidInitialize': wait_count -= 1 elif notification.name == 'SIPInvitationChangedState': # This is actually the only reason for which this notification could be received if notification.data.state == 'connected' and notification.data.sub_state == 'received_proposal': self._fail_proposal(originator='local', error='received stream proposal') self.handle_notification(notification) return local_sdp = SDPSession.new(self._invitation.sdp.active_local) local_sdp.version += 1 for stream in self.proposed_streams: # Try to reuse a disabled media stream to avoid an ever-growing SDP try: index = next(index for index, media in enumerate(local_sdp.media) if media.port == 0) reuse_media = True except StopIteration: index = len(local_sdp.media) reuse_media = False stream.index = index media = stream.get_local_media(remote_sdp=None, index=index) if reuse_media: local_sdp.media[index] = media else: local_sdp.media.append(media) self._invitation.send_reinvite(sdp=local_sdp) notification_center.post_notification('SIPSessionNewProposal', sender=self, data=NotificationData(originator='local', proposed_streams=self.proposed_streams[:])) received_invitation_state = False received_sdp_update = False try: with api.timeout(settings.sip.invite_timeout): while not received_invitation_state or not received_sdp_update: notification = self._channel.wait() if notification.name == 'SIPInvitationGotSDPUpdate': received_sdp_update = True if notification.data.succeeded: local_sdp = notification.data.local_sdp remote_sdp = notification.data.remote_sdp for s in self.streams: s.update(local_sdp, remote_sdp, s.index) else: self._fail_proposal(originator='local', error='SDP negotiation failed: %s' % notification.data.error) return elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': received_invitation_state = True notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=notification.data.code, reason=notification.data.reason)) if notification.data.code >= 300: proposed_streams = self.proposed_streams for stream in proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self.proposed_streams = None self.greenlet = None self.state = 'connected' notification_center.post_notification('SIPSessionProposalRejected', self, NotificationData(originator='local', code=notification.data.code, reason=notification.data.reason, proposed_streams=proposed_streams)) return elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) except api.TimeoutError: self.cancel_proposal() return accepted_streams = [] for stream in self.proposed_streams: try: remote_media = remote_sdp.media[stream.index] except IndexError: self._fail_proposal(originator='local', error='SDP media missing in answer') return else: if remote_media.port: stream.start(local_sdp, remote_sdp, stream.index) accepted_streams.append(stream) else: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() with api.timeout(self.media_stream_timeout): wait_count = len(accepted_streams) while wait_count > 0: notification = self._channel.wait() if notification.name == 'MediaStreamDidStart': wait_count -= 1 except api.TimeoutError: self._fail_proposal(originator='local', error='media stream timed-out while starting') except MediaStreamDidNotInitializeError as e: self._fail_proposal(originator='local', error='media stream did not initialize: {.data.reason}'.format(e)) except MediaStreamDidFailError as e: self._fail_proposal(originator='local', error='media stream failed: {.data.reason}'.format(e)) except InvitationDisconnectedError as e: self._fail_proposal(originator='local', error='session ended') notification = Notification('SIPInvitationChangedState', e.invitation, e.data) notification.center = notification_center self.handle_notification(notification) except SIPCoreError as e: self._fail_proposal(originator='local', error='SIP core error: %s' % str(e)) else: self.greenlet = None self.state = 'connected' self.streams += accepted_streams proposed_streams = self.proposed_streams self.proposed_streams = None any_stream_ice = any(getattr(stream, 'ice_active', False) for stream in accepted_streams) if any_stream_ice: self._reinvite_after_ice() notification_center.post_notification('SIPSessionProposalAccepted', self, NotificationData(originator='local', accepted_streams=accepted_streams, proposed_streams=proposed_streams)) notification_center.post_notification('SIPSessionDidRenegotiateStreams', self, NotificationData(originator='local', added_streams=accepted_streams, removed_streams=[])) for notification in unhandled_notifications: self.handle_notification(notification) if self._hold_in_progress: self._send_hold() finally: self.greenlet = None def remove_stream(self, stream): self.remove_streams([stream]) @transition_state('connected', 'sending_proposal') @run_in_green_thread def remove_streams(self, streams): streams = list(set(streams).intersection(self.streams)) if not streams: self.state = 'connected' return self.greenlet = api.getcurrent() notification_center = NotificationCenter() unhandled_notifications = [] try: local_sdp = SDPSession.new(self._invitation.sdp.active_local) local_sdp.version += 1 for stream in streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() self.streams.remove(stream) media = local_sdp.media[stream.index] media.port = 0 media.attributes = [] media.bandwidth_info = [] self._invitation.send_reinvite(sdp=local_sdp) received_invitation_state = False received_sdp_update = False with api.timeout(self.short_reinvite_timeout): while not received_invitation_state or not received_sdp_update: notification = self._channel.wait() if notification.name == 'SIPInvitationGotSDPUpdate': received_sdp_update = True if notification.data.succeeded: local_sdp = notification.data.local_sdp remote_sdp = notification.data.remote_sdp for s in self.streams: s.update(local_sdp, remote_sdp, s.index) elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': received_invitation_state = True notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=notification.data.code, reason=notification.data.reason)) if not (200 <= notification.data.code < 300): break elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) except InvitationDisconnectedError as e: for stream in streams: stream.end() self.greenlet = None notification = Notification('SIPInvitationChangedState', e.invitation, e.data) notification.center = notification_center self.handle_notification(notification) except (api.TimeoutError, MediaStreamDidFailError, SIPCoreError): for stream in streams: stream.end() self.end() else: for stream in streams: stream.end() self.greenlet = None self.state = 'connected' notification_center.post_notification('SIPSessionDidRenegotiateStreams', self, NotificationData(originator='local', added_streams=[], removed_streams=streams)) for notification in unhandled_notifications: self.handle_notification(notification) if self._hold_in_progress: self._send_hold() finally: self.greenlet = None @transition_state('sending_proposal', 'cancelling_proposal') @run_in_green_thread def cancel_proposal(self): if self.greenlet is not None: api.kill(self.greenlet, api.GreenletExit()) self.greenlet = api.getcurrent() notification_center = NotificationCenter() try: self._invitation.cancel_reinvite() while True: try: notification = self._channel.wait() except MediaStreamDidFailError: continue if notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=notification.data.code, reason=notification.data.reason)) if notification.data.code == 487: proposed_streams = self.proposed_streams or [] for stream in proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self.proposed_streams = None self.state = 'connected' notification_center.post_notification('SIPSessionProposalRejected', self, NotificationData(originator='local', code=notification.data.code, reason=notification.data.reason, proposed_streams=proposed_streams)) elif notification.data.code == 200: self.end() elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) break except SIPCoreError as e: notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', code=0, reason=None, failure_reason='SIP core error: %s' % str(e), redirect_identities=None)) proposed_streams = self.proposed_streams or [] for stream in proposed_streams: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self.proposed_streams = None self.greenlet = None self.state = 'connected' notification_center.post_notification('SIPSessionProposalRejected', self, NotificationData(originator='local', code=0, reason='SIP core error: %s' % str(e), proposed_streams=proposed_streams)) except InvitationDisconnectedError as e: for stream in self.proposed_streams or []: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self.proposed_streams = None self.greenlet = None notification = Notification('SIPInvitationChangedState', e.invitation, e.data) notification.center = notification_center self.handle_notification(notification) else: for stream in self.proposed_streams or []: notification_center.remove_observer(self, sender=stream) stream.deactivate() stream.end() self.proposed_streams = None self.greenlet = None self.state = 'connected' finally: self.greenlet = None if self._hold_in_progress: self._send_hold() @run_in_green_thread def hold(self): if self.on_hold or self._hold_in_progress: return self._hold_in_progress = True streams = (self.streams or []) + (self.proposed_streams or []) if not streams: return for stream in streams: stream.hold() if self.state == 'connected': self._send_hold() @run_in_green_thread def unhold(self): if not self.on_hold and not self._hold_in_progress: return self._hold_in_progress = False streams = (self.streams or []) + (self.proposed_streams or []) if not streams: return for stream in streams: stream.unhold() if self.state == 'connected': self._send_unhold() @run_in_green_thread def end(self): if self.state in (None, 'terminating', 'terminated'): return if self.greenlet is not None: api.kill(self.greenlet, api.GreenletExit()) self.greenlet = None notification_center = NotificationCenter() if self._invitation is None: # The invitation was not yet constructed self.state = 'terminated' notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator='local', code=487, reason='Session Cancelled', failure_reason='user request', redirect_identities=None)) return elif self._invitation.state is None: # The invitation was built but never sent streams = (self.streams or []) + (self.proposed_streams or []) for stream in streams[:]: try: notification_center.remove_observer(self, sender=stream) except KeyError: streams.remove(stream) else: stream.deactivate() stream.end() self.state = 'terminated' notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator='local', code=487, reason='Session Cancelled', failure_reason='user request', redirect_identities=None)) return invitation_state = self._invitation.state if invitation_state in ('disconnecting', 'disconnected'): return self.greenlet = api.getcurrent() self.state = 'terminating' if invitation_state == 'connected': notification_center.post_notification('SIPSessionWillEnd', self, NotificationData(originator='local')) streams = (self.streams or []) + (self.proposed_streams or []) for stream in streams[:]: try: notification_center.remove_observer(self, sender=stream) except KeyError: streams.remove(stream) else: stream.deactivate() cancelling = invitation_state != 'connected' and self.direction == 'outgoing' try: self._invitation.end(timeout=1) while True: try: notification = self._channel.wait() except MediaStreamDidFailError: continue if notification.name == 'SIPInvitationChangedState' and notification.data.state == 'disconnected': if notification.data.disconnect_reason in ('internal error', 'missing ACK'): pass elif notification.data.disconnect_reason == 'timeout': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local' if self.direction=='outgoing' else 'remote', method='INVITE', code=408, reason='Timeout')) elif cancelling: notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=notification.data.code, reason=notification.data.reason)) elif hasattr(notification.data, 'method'): notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method=notification.data.method, code=200, reason=sip_status_messages[200])) elif notification.data.disconnect_reason == 'user request': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='BYE', code=notification.data.code, reason=notification.data.reason)) break except SIPCoreError as e: if cancelling: notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator='local', code=0, reason=None, failure_reason='SIP core error: %s' % str(e), redirect_identities=None)) else: self.end_time = ISOTimestamp.now() notification_center.post_notification('SIPSessionDidEnd', self, NotificationData(originator='local', end_reason='SIP core error: %s' % str(e))) except InvitationDisconnectedError as e: # As it weird as it may sound, PJSIP accepts a BYE even without receiving a final response to the INVITE if e.data.prev_state == 'connected': if e.data.originator == 'remote': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator=e.data.originator, method=e.data.method, code=200, reason=sip_status_messages[200])) self.end_time = ISOTimestamp.now() notification_center.post_notification('SIPSessionDidEnd', self, NotificationData(originator=e.data.originator, end_reason=e.data.disconnect_reason)) elif getattr(e.data, 'method', None) == 'BYE' and e.data.originator == 'remote': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator=e.data.originator, method=e.data.method, code=200, reason=sip_status_messages[200])) notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator=e.data.originator, code=0, reason=None, failure_reason=e.data.disconnect_reason, redirect_identities=None)) else: if e.data.originator == 'remote': code = e.data.code reason = e.data.reason elif e.data.disconnect_reason == 'timeout': code = 408 reason = 'timeout' else: code = 0 reason = None if e.data.originator == 'remote' and code // 100 == 3: redirect_identities = e.data.headers.get('Contact', []) else: redirect_identities = None notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=code, reason=reason)) notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator=e.data.originator, code=code, reason=reason, failure_reason=e.data.disconnect_reason, redirect_identities=redirect_identities)) else: if cancelling: notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator='local', code=487, reason='Session Cancelled', failure_reason='user request', redirect_identities=None)) else: self.end_time = ISOTimestamp.now() notification_center.post_notification('SIPSessionDidEnd', self, NotificationData(originator='local', end_reason='user request')) finally: for stream in streams: stream.end() notification_center.remove_observer(self, sender=self._invitation) self.greenlet = None self.state = 'terminated' @check_state(['connected']) @check_transfer_state(None, None) @run_in_twisted_thread def transfer(self, target_uri, replaced_session=None): notification_center = NotificationCenter() notification_center.post_notification('SIPSessionTransferNewOutgoing', self, NotificationData(transfer_destination=target_uri)) try: self._invitation.transfer(target_uri, replaced_session.dialog_id if replaced_session is not None else None) except SIPCoreError as e: notification_center.post_notification('SIPSessionTransferDidFail', sender=self, data=NotificationData(code=500, reason=str(e))) @check_state(['connected', 'received_proposal', 'sending_proposal', 'accepting_proposal', 'rejecting_proposal', 'cancelling_proposal']) @check_transfer_state('incoming', 'starting') def accept_transfer(self): notification_center = NotificationCenter() notification_center.post_notification('SIPSessionTransferDidStart', sender=self) @check_state(['connected', 'received_proposal', 'sending_proposal', 'accepting_proposal', 'rejecting_proposal', 'cancelling_proposal']) @check_transfer_state('incoming', 'starting') def reject_transfer(self, code=603, reason=None): notification_center = NotificationCenter() notification_center.post_notification('SIPSessionTransferDidFail', self, NotificationData(code=code, reason=reason or sip_status_messages[code])) @property def dialog_id(self): return self._invitation.dialog_id if self._invitation is not None else None @property def local_identity(self): if self._invitation is not None and self._invitation.local_identity is not None: return self._invitation.local_identity else: return self._local_identity @property def peer_address(self): return self._invitation.peer_address if self._invitation is not None else None @property def remote_identity(self): if self._invitation is not None and self._invitation.remote_identity is not None: return self._invitation.remote_identity else: return self._remote_identity @property def remote_user_agent(self): return self._invitation.remote_user_agent if self._invitation is not None else None def _cancel_hold(self): notification_center = NotificationCenter() try: self._invitation.cancel_reinvite() while True: try: notification = self._channel.wait() except MediaStreamDidFailError: continue if notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=notification.data.code, reason=notification.data.reason)) if notification.data.code == 200: self.end() return False elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) break except SIPCoreError as e: notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', code=0, reason=None, failure_reason='SIP core error: %s' % str(e), redirect_identities=None)) except InvitationDisconnectedError as e: self.greenlet = None notification = Notification('SIPInvitationChangedState', e.invitation, e.data) notification.center = notification_center self.handle_notification(notification) return False return True def _send_hold(self): self.state = 'sending_proposal' self.greenlet = api.getcurrent() notification_center = NotificationCenter() unhandled_notifications = [] try: local_sdp = SDPSession.new(self._invitation.sdp.active_local) local_sdp.version += 1 for stream in self.streams: local_sdp.media[stream.index] = stream.get_local_media(remote_sdp=None, index=stream.index) self._invitation.send_reinvite(sdp=local_sdp) received_invitation_state = False received_sdp_update = False with api.timeout(self.short_reinvite_timeout): while not received_invitation_state or not received_sdp_update: notification = self._channel.wait() if notification.name == 'SIPInvitationGotSDPUpdate': received_sdp_update = True if notification.data.succeeded: local_sdp = notification.data.local_sdp remote_sdp = notification.data.remote_sdp for stream in self.streams: stream.update(local_sdp, remote_sdp, stream.index) elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': received_invitation_state = True notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=notification.data.code, reason=notification.data.reason)) elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) except InvitationDisconnectedError as e: self.greenlet = None notification = Notification('SIPInvitationChangedState', e.invitation, e.data) notification.center = notification_center self.handle_notification(notification) return except api.TimeoutError: if not self._cancel_hold(): return except SIPCoreError: pass self.greenlet = None self.on_hold = True self.state = 'connected' hold_supported_streams = (stream for stream in self.streams if stream.hold_supported) notification_center.post_notification('SIPSessionDidChangeHoldState', self, NotificationData(originator='local', on_hold=True, partial=any(not stream.on_hold_by_local for stream in hold_supported_streams))) for notification in unhandled_notifications: self.handle_notification(notification) if self._hold_in_progress: self._hold_in_progress = False else: for stream in self.streams: stream.unhold() self._send_unhold() def _send_unhold(self): self.state = 'sending_proposal' self.greenlet = api.getcurrent() notification_center = NotificationCenter() unhandled_notifications = [] try: local_sdp = SDPSession.new(self._invitation.sdp.active_local) local_sdp.version += 1 for stream in self.streams: local_sdp.media[stream.index] = stream.get_local_media(remote_sdp=None, index=stream.index) self._invitation.send_reinvite(sdp=local_sdp) received_invitation_state = False received_sdp_update = False with api.timeout(self.short_reinvite_timeout): while not received_invitation_state or not received_sdp_update: notification = self._channel.wait() if notification.name == 'SIPInvitationGotSDPUpdate': received_sdp_update = True if notification.data.succeeded: local_sdp = notification.data.local_sdp remote_sdp = notification.data.remote_sdp for stream in self.streams: stream.update(local_sdp, remote_sdp, stream.index) elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': received_invitation_state = True notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=notification.data.code, reason=notification.data.reason)) elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) except InvitationDisconnectedError as e: self.greenlet = None notification = Notification('SIPInvitationChangedState', e.invitation, e.data) notification.center = notification_center self.handle_notification(notification) return except api.TimeoutError: if not self._cancel_hold(): return except SIPCoreError: pass self.greenlet = None self.on_hold = False self.state = 'connected' notification_center.post_notification('SIPSessionDidChangeHoldState', self, NotificationData(originator='local', on_hold=False, partial=False)) for notification in unhandled_notifications: self.handle_notification(notification) if self._hold_in_progress: for stream in self.streams: stream.hold() self._send_hold() def _fail(self, originator, code, reason, error, reason_header=None): notification_center = NotificationCenter() prev_inv_state = self._invitation.state self.state = 'terminating' if prev_inv_state not in (None, 'incoming', 'outgoing', 'early', 'connecting'): notification_center.post_notification('SIPSessionWillEnd', self, NotificationData(originator=originator)) if self._invitation.state not in (None, 'disconnecting', 'disconnected'): try: if self._invitation.direction == 'incoming' and self._invitation.state in ('incoming', 'early'): if 400<=code<=699 and reason is not None: self._invitation.send_response(code, extra_headers=[reason_header] if reason_header is not None else []) else: self._invitation.end(extra_headers=[reason_header] if reason_header is not None else []) with api.timeout(1): while True: notification = self._channel.wait() if notification.name == 'SIPInvitationChangedState' and notification.data.state == 'disconnected': if prev_inv_state in ('connecting', 'connected'): if notification.data.disconnect_reason in ('timeout', 'missing ACK'): sip_code = 200 sip_reason = 'OK' originator = 'local' elif hasattr(notification.data, 'method'): sip_code = 200 sip_reason = 'OK' originator = 'remote' else: sip_code = notification.data.code sip_reason = notification.data.reason originator = 'local' notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator=originator, method='BYE', code=sip_code, reason=sip_reason)) elif self._invitation.direction == 'incoming' and prev_inv_state in ('incoming', 'early'): ack_received = notification.data.disconnect_reason != 'missing ACK' notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=code, reason=reason, ack_received=ack_received)) elif self._invitation.direction == 'outgoing' and prev_inv_state in ('outgoing', 'early'): notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='INVITE', code=487, reason='Session Cancelled')) break except SIPCoreError: pass except api.TimeoutError: if prev_inv_state in ('connecting', 'connected'): notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='local', method='BYE', code=408, reason=sip_status_messages[408])) notification_center.remove_observer(self, sender=self._invitation) self.state = 'terminated' notification_center.post_notification('SIPSessionDidFail', self, NotificationData(originator=originator, code=code, reason=reason, failure_reason=error, redirect_identities=None)) self.greenlet = None def _fail_proposal(self, originator, error): notification_center = NotificationCenter() for stream in self.proposed_streams: try: notification_center.remove_observer(self, sender=stream) except KeyError: # _fail_proposal can be called from reject_proposal, which means the stream will # not have been initialized or the session registered as an observer for it. pass else: stream.deactivate() stream.end() if originator == 'remote' and self._invitation.sub_state == 'received_proposal': try: self._invitation.send_response(488 if self.proposed_streams else 500) except SIPCoreError: pass else: notification_center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=500, reason=sip_status_messages[500], ack_received='unknown')) notification_center.post_notification('SIPSessionHadProposalFailure', self, NotificationData(originator=originator, failure_reason=error, proposed_streams=self.proposed_streams[:])) self.state = 'connected' self.proposed_streams = None self.greenlet = None @run_in_green_thread def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPInvitationChangedState(self, notification): if self.state == 'terminated': return if notification.data.originator == 'remote' and notification.data.state not in ('disconnecting', 'disconnected'): contact_header = notification.data.headers.get('Contact', None) if contact_header and 'isfocus' in contact_header[0].parameters: self.remote_focus = True if self.greenlet is not None: if notification.data.state == 'disconnected' and notification.data.prev_state != 'disconnecting': self._channel.send_exception(InvitationDisconnectedError(notification.sender, notification.data)) else: self._channel.send(notification) else: self.greenlet = api.getcurrent() unhandled_notifications = [] try: if notification.data.state == 'connected' and notification.data.sub_state == 'received_proposal': self.state = 'received_proposal' try: proposed_remote_sdp = self._invitation.sdp.proposed_remote active_remote_sdp = self._invitation.sdp.active_remote if len(proposed_remote_sdp.media) < len(active_remote_sdp.media): engine = Engine() self._invitation.send_response(488, extra_headers=[WarningHeader(399, engine.user_agent, 'Streams cannot be deleted from the SDP')]) self.state = 'connected' notification.center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=488, reason=sip_status_messages[488], ack_received='unknown')) return for stream in self.streams: if not stream.validate_update(proposed_remote_sdp, stream.index): engine = Engine() self._invitation.send_response(488, extra_headers=[WarningHeader(399, engine.user_agent, 'Failed to update media stream index %d' % stream.index)]) self.state = 'connected' notification.center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=488, reason=sip_status_messages[488], ack_received='unknown')) return added_media_indexes = set() removed_media_indexes = set() reused_media_indexes = set() for index, media_stream in enumerate(proposed_remote_sdp.media): if index >= len(active_remote_sdp.media): added_media_indexes.add(index) elif media_stream.port == 0 and active_remote_sdp.media[index].port > 0: removed_media_indexes.add(index) elif media_stream.port > 0 and active_remote_sdp.media[index].port == 0: reused_media_indexes.add(index) elif media_stream.media != active_remote_sdp.media[index].media: added_media_indexes.add(index) removed_media_indexes.add(index) if added_media_indexes | reused_media_indexes and removed_media_indexes: engine = Engine() self._invitation.send_response(488, extra_headers=[WarningHeader(399, engine.user_agent, 'Both removing AND adding a media stream is currently not supported')]) self.state = 'connected' notification.center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=488, reason=sip_status_messages[488], ack_received='unknown')) return elif added_media_indexes | reused_media_indexes: self.proposed_streams = [] for index in added_media_indexes | reused_media_indexes: media_stream = proposed_remote_sdp.media[index] if media_stream.port != 0: for stream_type in MediaStreamRegistry: try: stream = stream_type.new_from_sdp(self, proposed_remote_sdp, index) except UnknownStreamError: continue except InvalidStreamError as e: log.error("Invalid stream: {}".format(e)) break except Exception as e: log.exception("Exception occurred while setting up stream from SDP: {}".format(e)) break else: stream.index = index self.proposed_streams.append(stream) break if self.proposed_streams: self._invitation.send_response(100) notification.center.post_notification('SIPSessionNewProposal', sender=self, data=NotificationData(originator='remote', proposed_streams=self.proposed_streams[:])) else: self._invitation.send_response(488) self.state = 'connected' notification.center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=488, reason=sip_status_messages[488], ack_received='unknown')) return else: local_sdp = SDPSession.new(self._invitation.sdp.active_local) local_sdp.version += 1 removed_streams = [stream for stream in self.streams if stream.index in removed_media_indexes] prev_on_hold_streams = set(stream for stream in self.streams if stream.hold_supported and stream.on_hold_by_remote) for stream in removed_streams: notification.center.remove_observer(self, sender=stream) stream.deactivate() media = local_sdp.media[stream.index] media.port = 0 media.attributes = [] media.bandwidth_info = [] for stream in self.streams: local_sdp.media[stream.index] = stream.get_local_media(remote_sdp=proposed_remote_sdp, index=stream.index) try: self._invitation.send_response(200, sdp=local_sdp) except PJSIPError: for stream in removed_streams: self.streams.remove(stream) stream.end() if removed_streams: self.end() return else: try: self._invitation.send_response(488) except PJSIPError: self.end() return else: for stream in removed_streams: self.streams.remove(stream) stream.end() notification.center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=200, reason=sip_status_messages[200], ack_received='unknown')) received_invitation_state = False received_sdp_update = False while not received_sdp_update or not received_invitation_state or self._channel: notification = self._channel.wait() if notification.name == 'SIPInvitationGotSDPUpdate': received_sdp_update = True if notification.data.succeeded: local_sdp = notification.data.local_sdp remote_sdp = notification.data.remote_sdp for stream in self.streams: stream.update(local_sdp, remote_sdp, stream.index) elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': received_invitation_state = True elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) else: unhandled_notifications.append(notification) else: unhandled_notifications.append(notification) on_hold_streams = set(stream for stream in self.streams if stream.hold_supported and stream.on_hold_by_remote) if on_hold_streams != prev_on_hold_streams: hold_supported_streams = (stream for stream in self.streams if stream.hold_supported) notification.center.post_notification('SIPSessionDidChangeHoldState', self, NotificationData(originator='remote', on_hold=bool(on_hold_streams), partial=bool(on_hold_streams) and any(not stream.on_hold_by_remote for stream in hold_supported_streams))) if removed_media_indexes: notification.center.post_notification('SIPSessionDidRenegotiateStreams', self, NotificationData(originator='remote', added_streams=[], removed_streams=removed_streams)) except InvitationDisconnectedError as e: self.greenlet = None self.state = 'connected' notification = Notification('SIPInvitationChangedState', e.invitation, e.data) notification.center = NotificationCenter() self.handle_notification(notification) except SIPCoreError: self.end() else: self.state = 'connected' elif notification.data.state == 'connected' and notification.data.sub_state == 'received_proposal_request': self.state = 'received_proposal_request' try: # An empty proposal was received, generate an offer self._invitation.send_response(100) local_sdp = SDPSession.new(self._invitation.sdp.active_local) local_sdp.version += 1 connection_address = host.outgoing_ip_for(self._invitation.peer_address.ip) if local_sdp.connection is not None: local_sdp.connection.address = connection_address for index, stream in enumerate(self.streams): stream.reset(index) media = stream.get_local_media(remote_sdp=None, index=index) if media.connection is not None: media.connection.address = connection_address local_sdp.media[stream.index] = media self._invitation.send_response(200, sdp=local_sdp) notification.center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=200, reason=sip_status_messages[200], ack_received='unknown')) received_invitation_state = False received_sdp_update = False while not received_sdp_update or not received_invitation_state or self._channel: notification = self._channel.wait() if notification.name == 'SIPInvitationGotSDPUpdate': received_sdp_update = True if notification.data.succeeded: local_sdp = notification.data.local_sdp remote_sdp = notification.data.remote_sdp for stream in self.streams: stream.update(local_sdp, remote_sdp, stream.index) elif notification.name == 'SIPInvitationChangedState': if notification.data.state == 'connected' and notification.data.sub_state == 'normal': received_invitation_state = True elif notification.data.state == 'disconnected': raise InvitationDisconnectedError(notification.sender, notification.data) else: unhandled_notifications.append(notification) else: unhandled_notifications.append(notification) except InvitationDisconnectedError as e: self.greenlet = None self.state = 'connected' notification = Notification('SIPInvitationChangedState', e.invitation, e.data) notification.center = NotificationCenter() self.handle_notification(notification) except SIPCoreError: raise # FIXME else: self.state = 'connected' elif notification.data.prev_state == notification.data.state == 'connected' and notification.data.prev_sub_state == 'received_proposal' and notification.data.sub_state == 'normal': if notification.data.originator == 'local' and notification.data.code == 487: proposed_streams = self.proposed_streams self.proposed_streams = None self.state = 'connected' notification.center.post_notification('SIPSessionProposalRejected', self, NotificationData(originator='remote', code=notification.data.code, reason=notification.data.reason, proposed_streams=proposed_streams)) if self._hold_in_progress: self._send_hold() elif notification.data.state == 'disconnected': if self.state == 'incoming': self.state = 'terminated' if notification.data.originator == 'remote': notification.center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator='remote', method='INVITE', code=487, reason='Session Cancelled', ack_received='unknown')) notification.center.post_notification('SIPSessionDidFail', self, NotificationData(originator='remote', code=487, reason='Session Cancelled', failure_reason=notification.data.disconnect_reason, redirect_identities=None)) else: # There must have been an error involved notification.center.post_notification('SIPSessionDidFail', self, NotificationData(originator='local', code=0, reason=None, failure_reason=notification.data.disconnect_reason, redirect_identities=None)) else: self.state = 'terminated' notification.center.post_notification('SIPSessionWillEnd', self, NotificationData(originator=notification.data.originator)) for stream in self.streams: notification.center.remove_observer(self, sender=stream) stream.deactivate() stream.end() if notification.data.originator == 'remote': if hasattr(notification.data, 'method'): notification.center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator=notification.data.originator, method=notification.data.method, code=200, reason=sip_status_messages[200])) else: notification.center.post_notification('SIPSessionDidProcessTransaction', self, NotificationData(originator=notification.data.originator, method='INVITE', code=notification.data.code, reason=notification.data.reason)) self.end_time = ISOTimestamp.now() notification.center.post_notification('SIPSessionDidEnd', self, NotificationData(originator=notification.data.originator, end_reason=notification.data.disconnect_reason)) notification.center.remove_observer(self, sender=self._invitation) finally: self.greenlet = None for notification in unhandled_notifications: self.handle_notification(notification) def _NH_SIPInvitationGotSDPUpdate(self, notification): if self.greenlet is not None: self._channel.send(notification) def _NH_MediaStreamDidInitialize(self, notification): if self.greenlet is not None: self._channel.send(notification) def _NH_RTPStreamDidEnableEncryption(self, notification): if notification.sender.type != 'audio': return audio_stream = notification.sender if audio_stream.encryption.type == 'ZRTP': # start ZRTP on the video stream, if applicable try: video_stream = next(stream for stream in self.streams or [] if stream.type=='video') except StopIteration: return if video_stream.encryption.type == 'ZRTP' and not video_stream.encryption.active: video_stream.encryption.zrtp._enable(audio_stream) def _NH_MediaStreamDidStart(self, notification): stream = notification.sender if stream.type == 'audio' and stream.encryption.type == 'ZRTP': stream.encryption.zrtp._enable() elif stream.type == 'video' and stream.encryption.type == 'ZRTP': # start ZRTP on the video stream, if applicable try: audio_stream = next(stream for stream in self.streams or [] if stream.type=='audio') except StopIteration: pass else: if audio_stream.encryption.type == 'ZRTP' and audio_stream.encryption.active: stream.encryption.zrtp._enable(audio_stream) if self.greenlet is not None: self._channel.send(notification) def _NH_MediaStreamDidNotInitialize(self, notification): if self.greenlet is not None and self.state not in ('terminating', 'terminated'): self._channel.send_exception(MediaStreamDidNotInitializeError(notification.sender, notification.data)) def _NH_MediaStreamDidFail(self, notification): if self.greenlet is not None: if self.state not in ('terminating', 'terminated'): self._channel.send_exception(MediaStreamDidFailError(notification.sender, notification.data)) else: stream = notification.sender if self.streams == [stream]: self.end() else: try: self.remove_stream(stream) except IllegalStateError: self.end() @implementer(IObserver) class SessionManager(object, metaclass=Singleton): def __init__(self): self.sessions = [] self.state = None self._channel = coros.queue() def start(self): self.state = 'starting' notification_center = NotificationCenter() notification_center.post_notification('SIPSessionManagerWillStart', sender=self) notification_center.add_observer(self, 'SIPInvitationChangedState') notification_center.add_observer(self, 'SIPSessionNewIncoming') notification_center.add_observer(self, 'SIPSessionNewOutgoing') notification_center.add_observer(self, 'SIPSessionDidFail') notification_center.add_observer(self, 'SIPSessionDidEnd') self.state = 'started' notification_center.post_notification('SIPSessionManagerDidStart', sender=self) def stop(self): self.state = 'stopping' notification_center = NotificationCenter() notification_center.post_notification('SIPSessionManagerWillEnd', sender=self) for session in self.sessions: session.end() while self.sessions: self._channel.wait() notification_center.remove_observer(self, 'SIPInvitationChangedState') notification_center.remove_observer(self, 'SIPSessionNewIncoming') notification_center.remove_observer(self, 'SIPSessionNewOutgoing') notification_center.remove_observer(self, 'SIPSessionDidFail') notification_center.remove_observer(self, 'SIPSessionDidEnd') self.state = 'stopped' notification_center.post_notification('SIPSessionManagerDidEnd', sender=self) @run_in_twisted_thread def handle_notification(self, notification): if notification.name == 'SIPInvitationChangedState' and notification.data.state == 'incoming': account_manager = AccountManager() account = account_manager.find_account(notification.data.request_uri) if account is None: notification.sender.send_response(404) return notification.sender.send_response(100) session = Session(account) session.init_incoming(notification.sender, notification.data) elif notification.name in ('SIPSessionNewIncoming', 'SIPSessionNewOutgoing'): self.sessions.append(notification.sender) elif notification.name in ('SIPSessionDidFail', 'SIPSessionDidEnd'): self.sessions.remove(notification.sender) if self.state == 'stopping': self._channel.send(notification)