diff --git a/sylk/applications/xmppgateway/__init__.py b/sylk/applications/xmppgateway/__init__.py index 064070f..c69487f 100644 --- a/sylk/applications/xmppgateway/__init__.py +++ b/sylk/applications/xmppgateway/__init__.py @@ -1,544 +1,540 @@ # Copyright (C) 2012 AG Projects. See LICENSE for details # from application.notification import IObserver, NotificationCenter from application.python import Null from sipsimple.core import SIPURI, SIPCoreError from sipsimple.payloads import ParserError from sipsimple.payloads.iscomposing import IsComposingDocument, IsComposingMessage from sipsimple.streams.applications.chat import CPIMMessage, CPIMParserError from sipsimple.threading.green import run_in_green_thread from zope.interface import implements from sylk.applications import SylkApplication from sylk.applications.xmppgateway.configuration import XMPPGatewayConfig from sylk.applications.xmppgateway.datatypes import Identity, FrozenURI, generate_sylk_resource, decode_resource from sylk.applications.xmppgateway.im import SIPMessageSender, SIPMessageError, ChatSessionHandler from sylk.applications.xmppgateway.logger import log from sylk.applications.xmppgateway.presence import S2XPresenceHandler, X2SPresenceHandler from sylk.applications.xmppgateway.media import MediaSessionHandler from sylk.applications.xmppgateway.muc import X2SMucInvitationHandler, S2XMucInvitationHandler, X2SMucHandler from sylk.applications.xmppgateway.util import format_uri from sylk.applications.xmppgateway.xmpp import XMPPManager from sylk.applications.xmppgateway.xmpp.session import XMPPChatSession from sylk.applications.xmppgateway.xmpp.stanzas import ChatMessage, ChatComposingIndication, NormalMessage class XMPPGatewayApplication(SylkApplication): implements(IObserver) def __init__(self): self.xmpp_manager = XMPPManager() self.pending_sessions = {} self.chat_sessions = set() self.media_sessions = set() self.s2x_muc_sessions = {} self.x2s_muc_sessions = {} self.s2x_presence_subscriptions = {} self.x2s_presence_subscriptions = {} self.s2x_muc_add_participant_handlers = {} self.x2s_muc_add_participant_handlers = {} def start(self): notification_center = NotificationCenter() notification_center.add_observer(self, sender=self.xmpp_manager) notification_center.add_observer(self, name='JingleSessionNewIncoming') self.xmpp_manager.start() def stop(self): notification_center = NotificationCenter() notification_center.remove_observer(self, sender=self.xmpp_manager) notification_center.add_observer(self, name='JingleSessionNewIncoming') self.xmpp_manager.stop() def incoming_session(self, session): stream_types = set([stream.type for stream in session.proposed_streams]) if 'chat' in stream_types: log.msg('New chat session from %s to %s' % (session.remote_identity.uri, session.local_identity.uri)) self.incoming_chat_session(session) elif 'audio' in stream_types: log.msg('New audio session from %s to %s' % (session.remote_identity.uri, session.local_identity.uri)) self.incoming_media_session(session) else: log.msg('New session from %s to %s rejected. Unsupported media: %s ' % (session.remote_identity.uri, session.local_identity.uri, stream_types)) session.reject(488) def incoming_chat_session(self, session): # Check if this session is really an invitation to add a participant to a conference room / muc if session.remote_identity.uri.host in self.xmpp_manager.muc_domains and 'isfocus' in session._invitation.remote_contact_header.parameters: try: referred_by_uri = SIPURI.parse(session.transfer_info.referred_by) except SIPCoreError: log.msg("SIP multiparty session invitation %s failed: invalid Referred-By header" % session._invitation.call_id) session.reject(488) return muc_uri = FrozenURI(session.remote_identity.uri.user, session.remote_identity.uri.host) inviter_uri = FrozenURI(referred_by_uri.user, referred_by_uri.host) recipient_uri = FrozenURI(session.local_identity.uri.user, session.local_identity.uri.host) sender = Identity(muc_uri) recipient = Identity(recipient_uri) inviter = Identity(inviter_uri) try: handler = self.s2x_muc_add_participant_handlers[(muc_uri, recipient_uri)] except KeyError: handler = S2XMucInvitationHandler(session, sender, recipient, inviter) self.s2x_muc_add_participant_handlers[(muc_uri, recipient_uri)] = handler NotificationCenter().add_observer(self, sender=handler) handler.start() else: log.msg("SIP multiparty session invitation %s failed: there is another invitation in progress from %s to %s" % (session._invitation.call_id, format_uri(inviter_uri, 'sip'), format_uri(recipient_uri, 'xmpp'))) session.reject(480) return # Check domain if session.remote_identity.uri.host not in XMPPGatewayConfig.domains: log.msg('Session rejected: From domain is not a local XMPP domain') session.reject(606, 'Not Acceptable') return # Get URI representing the SIP side contact_uri = session._invitation.remote_contact_header.uri if contact_uri.parameters.get('gr') is not None: sip_leg_uri = FrozenURI(contact_uri.user, contact_uri.host, contact_uri.parameters.get('gr')) else: tmp = session.remote_identity.uri sip_leg_uri = FrozenURI(tmp.user, tmp.host, generate_sylk_resource()) # Get URI representing the XMPP side request_uri = session._invitation.request_uri remote_resource = request_uri.parameters.get('gr', None) if remote_resource is not None: try: remote_resource = decode_resource(remote_resource) except (TypeError, UnicodeError): remote_resource = None xmpp_leg_uri = FrozenURI(request_uri.user, request_uri.host, remote_resource) try: handler = self.pending_sessions[(sip_leg_uri, xmpp_leg_uri)] except KeyError: pass else: # There is another pending session with same identifiers, can't accept this one log.msg('Session rejected: other session with same identifiers in progress') session.reject(488) return sip_identity = Identity(sip_leg_uri, session.remote_identity.display_name) handler = ChatSessionHandler.new_from_sip_session(sip_identity, session) NotificationCenter().add_observer(self, sender=handler) key = (sip_leg_uri, xmpp_leg_uri) self.pending_sessions[key] = handler if xmpp_leg_uri.resource is not None: # Incoming session target contained GRUU, so create XMPPChatSession immediately xmpp_session = XMPPChatSession(local_identity=handler.sip_identity, remote_identity=Identity(xmpp_leg_uri)) handler.xmpp_identity = xmpp_session.remote_identity handler.xmpp_session = xmpp_session def incoming_media_session(self, session): - if session.remote_identity.uri.host in self.xmpp_manager.muc_domains: - # Sessions to MUC are not allowed yet - session.reject(403) - return - if session.remote_identity.uri.host not in self.xmpp_manager.domains: + if session.remote_identity.uri.host not in self.xmpp_manager.domains|self.xmpp_manager.muc_domains: log.msg('Session rejected: From domain is not a local XMPP domain') - session.reject(606, 'Not Acceptable') + session.reject(403) return handler = MediaSessionHandler.new_from_sip_session(session) if handler is not None: NotificationCenter().add_observer(self, sender=handler) def incoming_subscription(self, subscribe_request, data): from_header = data.headers.get('From', Null) to_header = data.headers.get('To', Null) if Null in (from_header, to_header): subscribe_request.reject(400) return log.msg('SIP subscription from %s to %s' % (format_uri(from_header.uri, 'sip'), format_uri(to_header.uri, 'xmpp'))) if subscribe_request.event != 'presence': log.msg('SIP subscription rejected: only presence event is supported') subscribe_request.reject(489) return # Check domain remote_identity_uri = data.headers['From'].uri if remote_identity_uri.host not in XMPPGatewayConfig.domains: log.msg('SIP subscription rejected: From domain is not a local XMPP domain') subscribe_request.reject(606) return # Get URI representing the SIP side sip_leg_uri = FrozenURI(remote_identity_uri.user, remote_identity_uri.host) # Get URI representing the XMPP side request_uri = data.request_uri xmpp_leg_uri = FrozenURI(request_uri.user, request_uri.host) try: handler = self.s2x_presence_subscriptions[(sip_leg_uri, xmpp_leg_uri)] except KeyError: sip_identity = Identity(sip_leg_uri, data.headers['From'].display_name) xmpp_identity = Identity(xmpp_leg_uri) handler = S2XPresenceHandler(sip_identity, xmpp_identity) self.s2x_presence_subscriptions[(sip_leg_uri, xmpp_leg_uri)] = handler NotificationCenter().add_observer(self, sender=handler) handler.start() handler.add_sip_subscription(subscribe_request) def incoming_referral(self, refer_request, data): refer_request.reject(405) def incoming_sip_message(self, message_request, data): content_type = data.headers.get('Content-Type', Null).content_type from_header = data.headers.get('From', Null) to_header = data.headers.get('To', Null) if Null in (content_type, from_header, to_header): message_request.answer(400) return log.msg('New SIP Message from %s to %s' % (from_header.uri, to_header.uri)) # Check domain if from_header.uri.host not in XMPPGatewayConfig.domains: log.msg('Message rejected: From domain is not a local XMPP domain') message_request.answer(606) return if content_type == 'message/cpim': try: cpim_message = CPIMMessage.parse(data.body) except CPIMParserError: log.msg('Message rejected: CPIM parse error') message_request.answer(400) return else: body = cpim_message.body content_type = cpim_message.content_type sender = cpim_message.sender or from_header from_uri = sender.uri else: body = data.body from_uri = from_header.uri to_uri = str(to_header.uri) message_request.answer(200) if from_uri.parameters.get('gr', None) is None: from_uri = SIPURI.new(from_uri) from_uri.parameters['gr'] = generate_sylk_resource() sender = Identity(FrozenURI.parse(from_uri)) recipient = Identity(FrozenURI.parse(to_uri)) if content_type in ('text/plain', 'text/html'): if content_type == 'text/plain': html_body = None else: html_body = body body = None if XMPPGatewayConfig.use_msrp_for_chat: message = NormalMessage(sender, recipient, body, html_body, use_receipt=False) self.xmpp_manager.send_stanza(message) else: message = ChatMessage(sender, recipient, body, html_body, use_receipt=False) self.xmpp_manager.send_stanza(message) elif content_type == IsComposingDocument.content_type: if not XMPPGatewayConfig.use_msrp_for_chat: try: msg = IsComposingMessage.parse(body) except ParserError: pass else: state = 'composing' if msg.state == 'active' else 'paused' message = ChatComposingIndication(sender, recipient, state, use_receipt=False) self.xmpp_manager.send_stanza(message) def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) # Out of band XMPP stanza handling @run_in_green_thread def _NH_XMPPGotChatMessage(self, notification): # This notification is only processed here untill the ChatSessionHandler # has both (SIP and XMPP) sessions established message = notification.data.message sender = message.sender recipient = message.recipient if XMPPGatewayConfig.use_msrp_for_chat: if recipient.uri.resource is None: # If recipient resource is not set the session is started from # the XMPP side sip_leg_uri = FrozenURI.new(recipient.uri) xmpp_leg_uri = FrozenURI.new(sender.uri) try: handler = self.pending_sessions[(sip_leg_uri, xmpp_leg_uri)] except KeyError: # Not found, need to create a new handler and a outgoing SIP session xmpp_identity = Identity(xmpp_leg_uri) handler = ChatSessionHandler.new_from_xmpp_stanza(xmpp_identity, sip_leg_uri) key = (sip_leg_uri, xmpp_leg_uri) self.pending_sessions[key] = handler NotificationCenter().add_observer(self, sender=handler) handler.enqueue_xmpp_message(message) else: # Find handler pending XMPP confirmation sip_leg_uri = FrozenURI.new(recipient.uri) xmpp_leg_uri = FrozenURI(sender.uri.user, sender.uri.host) try: handler = self.pending_sessions[(sip_leg_uri, xmpp_leg_uri)] except KeyError: # Find handler pending XMPP confirmation sip_leg_uri = FrozenURI(recipient.uri.user, recipient.uri.host) xmpp_leg_uri = FrozenURI.new(sender.uri) try: handler = self.pending_sessions[(sip_leg_uri, xmpp_leg_uri)] except KeyError: # It's a new XMPP session to a full JID, disregard the full JID and start a new SIP session to the bare JID xmpp_identity = Identity(xmpp_leg_uri) handler = ChatSessionHandler.new_from_xmpp_stanza(xmpp_identity, sip_leg_uri) key = (sip_leg_uri, xmpp_leg_uri) self.pending_sessions[key] = handler NotificationCenter().add_observer(self, sender=handler) handler.enqueue_xmpp_message(message) else: # Found handle, create XMPP session and establish session session = XMPPChatSession(local_identity=recipient, remote_identity=sender) handler.enqueue_xmpp_message(message) handler.xmpp_identity = session.remote_identity handler.xmpp_session = session else: sip_message_sender = SIPMessageSender(message) try: sip_message_sender.send().wait() except SIPMessageError as e: # TODO report back an error stanza log.error('Error sending SIP Message: %s' % e) @run_in_green_thread def _NH_XMPPGotNormalMessage(self, notification): message = notification.data.message sip_message_sender = SIPMessageSender(message) try: sip_message_sender.send().wait() except SIPMessageError as e: # TODO report back an error stanza log.error('Error sending SIP Message: %s' % e) @run_in_green_thread def _NH_XMPPGotComposingIndication(self, notification): composing_indication = notification.data.composing_indication sender = composing_indication.sender recipient = composing_indication.recipient if not XMPPGatewayConfig.use_msrp_for_chat: state = 'active' if composing_indication.state == 'composing' else 'idle' body = IsComposingMessage(state=state, refresh=composing_indication.interval or 30).toxml() message = NormalMessage(sender, recipient, body, IsComposingDocument.content_type) sip_message_sender = SIPMessageSender(message) try: sip_message_sender.send().wait() except SIPMessageError as e: # TODO report back an error stanza log.error('Error sending SIP Message: %s' % e) def _NH_XMPPGotPresenceSubscriptionRequest(self, notification): stanza = notification.data.stanza # Disregard the resource part, the presence request could be a probe instead of a subscribe sender_uri = stanza.sender.uri sender_uri_bare = FrozenURI(sender_uri.user, sender_uri.host) try: handler = self.x2s_presence_subscriptions[(sender_uri_bare, stanza.recipient.uri)] except KeyError: xmpp_identity = stanza.sender xmpp_identity.uri = sender_uri_bare sip_identity = stanza.recipient handler = X2SPresenceHandler(sip_identity, xmpp_identity) self.x2s_presence_subscriptions[(sender_uri_bare, stanza.recipient.uri)] = handler notification.center.add_observer(self, sender=handler) handler.start() def _NH_XMPPGotMucJoinRequest(self, notification): stanza = notification.data.stanza muc_uri = FrozenURI(stanza.recipient.uri.user, stanza.recipient.uri.host) nickname = stanza.recipient.uri.resource try: handler = self.x2s_muc_sessions[(stanza.sender.uri, muc_uri)] except KeyError: xmpp_identity = stanza.sender sip_identity = stanza.recipient sip_identity.uri = muc_uri handler = X2SMucHandler(sip_identity, xmpp_identity, nickname) handler._first_stanza = stanza notification.center.add_observer(self, sender=handler) handler.start() # Check if there was a pending join request on the SIP side try: handler = self.s2x_muc_add_participant_handlers[(muc_uri, FrozenURI(stanza.sender.uri.user, stanza.sender.uri.host))] except KeyError: pass else: handler.stop() def _NH_XMPPGotMucAddParticipantRequest(self, notification): sender = notification.data.sender recipient = notification.data.recipient participant = notification.data.participant muc_uri = FrozenURI(recipient.uri.user, recipient.uri.host) sender_uri = FrozenURI(sender.uri.user, sender.uri.host) participant_uri = FrozenURI(participant.uri.user, participant.uri.host) sender = Identity(sender_uri) recipient = Identity(muc_uri) participant = Identity(participant_uri) try: handler = self.x2s_muc_add_participant_handlers[(muc_uri, participant_uri)] except KeyError: handler = X2SMucInvitationHandler(sender, recipient, participant) self.x2s_muc_add_participant_handlers[(muc_uri, participant_uri)] = handler notification.center.add_observer(self, sender=handler) handler.start() # Chat session handling def _NH_ChatSessionDidStart(self, notification): handler = notification.sender log.msg('Chat session established sip:%s <--> xmpp:%s' % (handler.sip_identity.uri, handler.xmpp_identity.uri)) for k,v in self.pending_sessions.items(): if v is handler: del self.pending_sessions[k] break self.chat_sessions.add(handler) def _NH_ChatSessionDidEnd(self, notification): handler = notification.sender log.msg('Chat session ended sip:%s <--> xmpp:%s' % (handler.sip_identity.uri, handler.xmpp_identity.uri)) self.chat_sessions.remove(handler) notification.center.remove_observer(self, sender=handler) def _NH_ChatSessionDidFail(self, notification): handler = notification.sender uris = None for k,v in self.pending_sessions.items(): if v is handler: uris = k del self.pending_sessions[k] break sip_uri, xmpp_uri = uris log.msg('Chat session failed sip:%s <--> xmpp:%s (%s)' % (sip_uri, xmpp_uri, notification.data.reason)) notification.center.remove_observer(self, sender=handler) # Presence handling def _NH_S2XPresenceHandlerDidStart(self, notification): handler = notification.sender log.msg('Presence flow 0x%x established %s --> %s' % (id(handler), format_uri(handler.sip_identity.uri, 'sip'), format_uri(handler.xmpp_identity.uri, 'xmpp'))) log.msg('%d SIP --> XMPP and %d XMPP --> SIP presence flows are active' % (len(self.s2x_presence_subscriptions.keys()), len(self.x2s_presence_subscriptions.keys()))) def _NH_S2XPresenceHandlerDidEnd(self, notification): handler = notification.sender self.s2x_presence_subscriptions.pop((handler.sip_identity.uri, handler.xmpp_identity.uri), None) notification.center.remove_observer(self, sender=handler) log.msg('Presence flow 0x%x ended %s --> %s' % (id(handler), format_uri(handler.sip_identity.uri, 'sip'), format_uri(handler.xmpp_identity.uri, 'xmpp'))) log.msg('%d SIP --> XMPP and %d XMPP --> SIP presence flows are active' % (len(self.s2x_presence_subscriptions.keys()), len(self.x2s_presence_subscriptions.keys()))) def _NH_X2SPresenceHandlerDidStart(self, notification): handler = notification.sender log.msg('Presence flow 0x%x established %s --> %s' % (id(handler), format_uri(handler.xmpp_identity.uri, 'xmpp'), format_uri(handler.sip_identity.uri, 'sip'))) log.msg('%d SIP --> XMPP and %d XMPP --> SIP presence flows are active' % (len(self.s2x_presence_subscriptions.keys()), len(self.x2s_presence_subscriptions.keys()))) def _NH_X2SPresenceHandlerDidEnd(self, notification): handler = notification.sender self.x2s_presence_subscriptions.pop((handler.xmpp_identity.uri, handler.sip_identity.uri), None) notification.center.remove_observer(self, sender=handler) log.msg('Presence flow 0x%x ended %s --> %s' % (id(handler), format_uri(handler.xmpp_identity.uri, 'xmpp'), format_uri(handler.sip_identity.uri, 'sip'))) log.msg('%d SIP --> XMPP and %d XMPP --> SIP presence flows are active' % (len(self.s2x_presence_subscriptions.keys()), len(self.x2s_presence_subscriptions.keys()))) # MUC handling def _NH_X2SMucHandlerDidStart(self, notification): handler = notification.sender log.msg('Multiparty session established xmpp:%s --> sip:%s' % (handler.xmpp_identity.uri, handler.sip_identity.uri)) self.x2s_muc_sessions[(handler.xmpp_identity.uri, handler.sip_identity.uri)] = handler def _NH_X2SMucHandlerDidEnd(self, notification): handler = notification.sender log.msg('Multiparty session ended xmpp:%s --> sip:%s' % (handler.xmpp_identity.uri, handler.sip_identity.uri)) self.x2s_muc_sessions.pop((handler.xmpp_identity.uri, handler.sip_identity.uri), None) notification.center.remove_observer(self, sender=handler) def _NH_X2SMucInvitationHandlerDidStart(self, notification): handler = notification.sender sender_uri = handler.sender.uri muc_uri = handler.recipient.uri participant_uri = handler.participant.uri log.msg('%s invited %s to multiparty chat %s' % (format_uri(sender_uri, 'xmpp'), format_uri(participant_uri), format_uri(muc_uri, 'sip'))) def _NH_X2SMucInvitationHandlerDidEnd(self, notification): handler = notification.sender sender_uri = handler.sender.uri muc_uri = handler.recipient.uri participant_uri = handler.participant.uri log.msg('%s added %s to multiparty chat %s' % (format_uri(sender_uri, 'xmpp'), format_uri(participant_uri), format_uri(muc_uri, 'sip'))) del self.x2s_muc_add_participant_handlers[(muc_uri, participant_uri)] notification.center.remove_observer(self, sender=handler) def _NH_X2SMucInvitationHandlerDidFail(self, notification): handler = notification.sender sender_uri = handler.sender.uri muc_uri = handler.recipient.uri participant_uri = handler.participant.uri log.msg('%s could not add %s to multiparty chat %s: %s' % (format_uri(sender_uri, 'xmpp'), format_uri(participant_uri), format_uri(muc_uri, 'sip'), notification.data.failure)) del self.x2s_muc_add_participant_handlers[(muc_uri, participant_uri)] notification.center.remove_observer(self, sender=handler) def _NH_S2XMucInvitationHandlerDidStart(self, notification): handler = notification.sender muc_uri = handler.sender.uri inviter_uri = handler.inviter.uri recipient_uri = handler.recipient.uri log.msg("%s invited %s to multiparty chat %s" % (format_uri(inviter_uri, 'sip'), format_uri(recipient_uri, 'xmpp'), format_uri(muc_uri, 'sip'))) def _NH_S2XMucInvitationHandlerDidEnd(self, notification): handler = notification.sender muc_uri = handler.sender.uri inviter_uri = handler.inviter.uri recipient_uri = handler.recipient.uri log.msg('%s added %s to multiparty chat %s' % (format_uri(inviter_uri, 'sip'), format_uri(recipient_uri, 'xmpp'), format_uri(muc_uri, 'sip'))) del self.s2x_muc_add_participant_handlers[(muc_uri, recipient_uri)] notification.center.remove_observer(self, sender=handler) def _NH_S2XMucInvitationHandlerDidFail(self, notification): handler = notification.sender muc_uri = handler.sender.uri inviter_uri = handler.inviter.uri recipient_uri = handler.recipient.uri log.msg('%s could not add %s to multiparty chat %s: %s' % (format_uri(inviter_uri, 'sip'), format_uri(recipient_uri, 'xmpp'), format_uri(muc_uri, 'sip'), str(notification.data.failure))) del self.s2x_muc_add_participant_handlers[(muc_uri, recipient_uri)] notification.center.remove_observer(self, sender=handler) # Media sessions def _NH_JingleSessionNewIncoming(self, notification): session = notification.sender handler = MediaSessionHandler.new_from_jingle_session(session) if handler is not None: notification.center.add_observer(self, sender=handler) def _NH_MediaSessionHandlerDidStart(self, notification): handler = notification.sender log.msg('Media session started sip:%s <--> xmpp:%s' % (handler.sip_identity.uri, handler.xmpp_identity.uri)) self.media_sessions.add(handler) def _NH_MediaSessionHandlerDidEnd(self, notification): handler = notification.sender log.msg('Media session ended sip:%s <--> xmpp:%s' % (handler.sip_identity.uri, handler.xmpp_identity.uri)) self.media_sessions.remove(handler) notification.center.remove_observer(self, sender=handler) def _NH_MediaSessionHandlerDidFail(self, notification): handler = notification.sender log.msg('Media session failed sip:%s <--> xmpp:%s' % (handler.sip_identity.uri, handler.xmpp_identity.uri)) notification.center.remove_observer(self, sender=handler) diff --git a/sylk/applications/xmppgateway/media.py b/sylk/applications/xmppgateway/media.py index 46fe1aa..72e0dc0 100644 --- a/sylk/applications/xmppgateway/media.py +++ b/sylk/applications/xmppgateway/media.py @@ -1,315 +1,329 @@ # Copyright (C) 2013 AG Projects. See LICENSE for details # from application.notification import IObserver, NotificationCenter, NotificationData from application.python import Null from eventlib.twistedutil import block_on from sipsimple.account import AccountManager from sipsimple.conference import AudioConference from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.core import ContactHeader, FromHeader, ToHeader -from sipsimple.core import Engine, SIPURI +from sipsimple.core import Engine, SIPURI, SIPCoreError from sipsimple.lookup import DNSLookup, DNSLookupError from sipsimple.streams import MediaStreamRegistry as SIPMediaStreamRegistry from sipsimple.threading import run_in_twisted_thread from sipsimple.threading.green import run_in_green_thread from zope.interface import implements from sylk.applications.xmppgateway.datatypes import Identity, FrozenURI, generate_sylk_resource, encode_resource, decode_resource from sylk.applications.xmppgateway.logger import log from sylk.applications.xmppgateway.xmpp import XMPPManager from sylk.applications.xmppgateway.xmpp.jingle.session import JingleSession from sylk.applications.xmppgateway.xmpp.jingle.streams import MediaStreamRegistry as JingleMediaStreamRegistry from sylk.applications.xmppgateway.xmpp.stanzas import jingle from sylk.configuration import SIPConfig from sylk.session import Session __all__ = ['MediaSessionHandler'] class MediaSessionHandler(object): implements(IObserver) def __init__(self): self.started = False self.ended = False self._sip_identity = None self._xmpp_identity = None self._audio_bidge = AudioConference() self.sip_session = None self.jingle_session = None @classmethod def new_from_sip_session(cls, session): proposed_stream_types = set([stream.type for stream in session.proposed_streams]) streams = [] for stream_type in proposed_stream_types: try: klass = JingleMediaStreamRegistry().get(stream_type) except Exception: continue streams.append(klass()) if not streams: session.reject(488) return None session.send_ring_indication() instance = cls() NotificationCenter().add_observer(instance, sender=session) # Get URI representing the SIP side contact_uri = session._invitation.remote_contact_header.uri if contact_uri.parameters.get('gr') is not None: sip_leg_uri = FrozenURI(contact_uri.user, contact_uri.host, contact_uri.parameters.get('gr')) else: tmp = session.remote_identity.uri sip_leg_uri = FrozenURI(tmp.user, tmp.host, generate_sylk_resource()) instance._sip_identity = Identity(sip_leg_uri) # Get URI representing the XMPP side request_uri = session._invitation.request_uri remote_resource = request_uri.parameters.get('gr', None) if remote_resource is not None: try: remote_resource = decode_resource(remote_resource) except (TypeError, UnicodeError): remote_resource = None xmpp_leg_uri = FrozenURI(request_uri.user, request_uri.host, remote_resource) instance._xmpp_identity = Identity(xmpp_leg_uri) instance.sip_session = session instance._start_outgoing_jingle_session(streams) return instance @classmethod def new_from_jingle_session(cls, session): proposed_stream_types = set([stream.type for stream in session.proposed_streams]) streams = [] for stream_type in proposed_stream_types: try: klass = SIPMediaStreamRegistry().get(stream_type) except Exception: continue streams.append(klass()) if not streams: session.reject('unsupported-applications') return None session.send_ring_indication() instance = cls() NotificationCenter().add_observer(instance, sender=session) instance._xmpp_identity = session.remote_identity instance._sip_identity = session.local_identity instance.jingle_session = session instance._start_outgoing_sip_session(streams) return instance @property def sip_identity(self): return self._sip_identity @property def xmpp_identity(self): return self._xmpp_identity def _set_started(self, value): old_value = self.__dict__.get('started', False) self.__dict__['started'] = value if not old_value and value: NotificationCenter().post_notification('MediaSessionHandlerDidStart', sender=self) def _get_started(self): return self.__dict__['started'] started = property(_get_started, _set_started) del _get_started, _set_started @run_in_green_thread def _start_outgoing_sip_session(self, streams): notification_center = NotificationCenter() # self.xmpp_identity is our local identity on the SIP side from_uri = self.xmpp_identity.uri.as_sip_uri() - del from_uri.parameters['gr'] # no GRUU in From header + from_uri.parameters.pop('gr', None) # no GRUU in From header to_uri = self.sip_identity.uri.as_sip_uri() - del to_uri.parameters['gr'] # no GRUU in To header + to_uri.parameters.pop('gr', None) # no GRUU in To header # TODO: need to fix GRUU in the proxy #contact_uri = self.xmpp_identity.uri.as_sip_uri() #contact_uri.parameters['gr'] = encode_resource(contact_uri.parameters['gr'].decode('utf-8')) lookup = DNSLookup() settings = SIPSimpleSettings() account = AccountManager().sylkserver_account if account.sip.outbound_proxy is not None: uri = SIPURI(host=account.sip.outbound_proxy.host, port=account.sip.outbound_proxy.port, parameters={'transport': account.sip.outbound_proxy.transport}) else: uri = to_uri try: routes = lookup.lookup_sip_proxy(uri, settings.sip.transport_list).wait() except DNSLookupError: log.warning('DNS lookup error while looking for %s proxy' % uri) notification_center.post_notification('MedialSessionHandlerDidFail', sender=self, data=NotificationData(reason='DNS lookup error')) return route = routes.pop(0) from_header = FromHeader(from_uri) to_header = ToHeader(to_uri) transport = route.transport parameters = {} if transport=='udp' else {'transport': transport} contact_uri = SIPURI(user=account.contact.username, host=SIPConfig.local_ip.normalized, port=getattr(Engine(), '%s_port' % transport), parameters=parameters) contact_header = ContactHeader(contact_uri) self.sip_session = Session(account) notification_center.add_observer(self, sender=self.sip_session) self.sip_session.connect(from_header, to_header, contact_header=contact_header, routes=[route], streams=streams) @run_in_green_thread def _start_outgoing_jingle_session(self, streams): if self.xmpp_identity.uri.resource is not None: self.sip_session.reject() return xmpp_manager = XMPPManager() local_jid = self.sip_identity.uri.as_xmpp_jid() remote_jid = self.xmpp_identity.uri.as_xmpp_jid() + # If this was an invitation to a conference, use the information in the Referred-By header + if local_jid.host in xmpp_manager.muc_domains and self.sip_session.transfer_info and self.sip_session.transfer_info.referred_by: + try: + referred_by_uri = SIPURI.parse(self.sip_session.transfer_info.referred_by) + except SIPCoreError: + self.sip_session.reject(488) + return + else: + inviter_uri = FrozenURI(referred_by_uri.user, referred_by_uri.host) + local_jid = inviter_uri.as_xmpp_jid() + # Use disco to gather potential JIDs to call d = xmpp_manager.disco_client_protocol.requestItems(remote_jid, sender=local_jid) try: items = block_on(d) except Exception: items = [] if not items: self.sip_session.reject(480) return # Check which items support Jingle valid = [] for item in items: d = xmpp_manager.disco_client_protocol.requestInfo(item.entity, nodeIdentifier=item.nodeIdentifier, sender=local_jid) try: info = block_on(d) except Exception: continue if jingle.NS_JINGLE in info.features and jingle.NS_JINGLE_APPS_RTP in info.features: valid.append(item.entity) if not valid: self.sip_session.reject(480) return # TODO: start multiple sessions? self._xmpp_identity = Identity(FrozenURI.parse(valid[0])) notification_center = NotificationCenter() - self.jingle_session = JingleSession(xmpp_manager.jingle_protocol) + if local_jid.host in xmpp_manager.muc_domains: + self.jingle_session = JingleSession(xmpp_manager.jingle_coin_protocol) + else: + self.jingle_session = JingleSession(xmpp_manager.jingle_protocol) notification_center.add_observer(self, sender=self.jingle_session) self.jingle_session.connect(self.sip_identity, self.xmpp_identity, streams) def end(self): if self.ended: return notification_center = NotificationCenter() if self.sip_session is not None: notification_center.remove_observer(self, sender=self.sip_session) if self.sip_session.direction == 'incoming' and not self.started: self.sip_session.reject() else: self.sip_session.end() self.sip_session = None if self.jingle_session is not None: notification_center.remove_observer(self, sender=self.jingle_session) if self.jingle_session.direction == 'incoming' and not self.started: self.jingle_session.reject() else: self.jingle_session.end() self.jingle_session = None self.ended = True if self.started: notification_center.post_notification('MediaSessionHandlerDidEnd', sender=self) else: notification_center.post_notification('MediaSessionHandlerDidFail', sender=self) @run_in_twisted_thread def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPSessionDidStart(self, notification): log.msg("SIP session %s started" % notification.sender._invitation.call_id) if self.sip_session.direction == 'outgoing': # Time to accept the Jingle session and bridge them together try: audio_stream = next(stream for stream in self.sip_session.streams if stream.type=='audio') except StopIteration: pass else: self._audio_bidge.add(audio_stream) self.jingle_session.accept(self.jingle_session.proposed_streams) else: # Both sessions have been accepted now self.started = True try: audio_stream = next(stream for stream in self.sip_session.streams if stream.type=='audio') except StopIteration: pass else: self._audio_bidge.add(audio_stream) def _NH_SIPSessionDidEnd(self, notification): log.msg("SIP session %s ended" % notification.sender._invitation.call_id) notification.center.remove_observer(self, sender=self.sip_session) self.sip_session = None self.end() def _NH_SIPSessionDidFail(self, notification): log.msg("SIP session %s failed (%s)" % (notification.sender._invitation.call_id, notification.data.reason)) notification.center.remove_observer(self, sender=self.sip_session) self.sip_session = None self.end() def _NH_SIPSessionGotProposal(self, notification): self.sip_session.reject_proposal() def _NH_SIPSessionTransferNewIncoming(self, notification): self.sip_session.reject_transfer(403) def _NH_SIPSessionDidChangeHoldState(self, notification): if notification.data.originator == 'remote': if notification.data.on_hold: self.jingle_session.hold() else: self.jingle_session.unhold() def _NH_JingleSessionDidStart(self, notification): log.msg("Jingle session %s started" % notification.sender.id) if self.jingle_session.direction == 'incoming': # Both sessions have been accepted now self.started = True try: audio_stream = next(stream for stream in self.jingle_session.streams if stream.type=='audio') except StopIteration: pass else: self._audio_bidge.add(audio_stream) else: # Time to accept the Jingle session and bridge them together try: audio_stream = next(stream for stream in self.jingle_session.streams if stream.type=='audio') except StopIteration: pass else: self._audio_bidge.add(audio_stream) self.sip_session.accept(self.sip_session.proposed_streams) def _NH_JingleSessionDidEnd(self, notification): log.msg("Jingle session %s ended" % notification.sender.id) notification.center.remove_observer(self, sender=self.jingle_session) self.jingle_session = None self.end() def _NH_JingleSessionDidFail(self, notification): log.msg("Jingle session %s failed (%s)" % (notification.sender.id, notification.data.reason)) notification.center.remove_observer(self, sender=self.jingle_session) self.jingle_session = None self.end() def _NH_JingleSessionDidChangeHoldState(self, notification): if notification.data.originator == 'remote': if notification.data.on_hold: self.sip_session.hold() else: self.sip_session.unhold() diff --git a/sylk/applications/xmppgateway/xmpp/__init__.py b/sylk/applications/xmppgateway/xmpp/__init__.py index 162ccd2..2a861e1 100644 --- a/sylk/applications/xmppgateway/xmpp/__init__.py +++ b/sylk/applications/xmppgateway/xmpp/__init__.py @@ -1,319 +1,325 @@ # Copyright (C) 2012 AG Projects. See LICENSE for details # from application.notification import IObserver, NotificationCenter, NotificationData from application.python import Null from application.python.types import Singleton from twisted.internet import reactor from wokkel.disco import DiscoClientProtocol from wokkel.generic import FallbackHandler, VersionHandler from wokkel.ping import PingHandler from wokkel.server import ServerService from zope.interface import implements from sylk import __version__ as SYLK_VERSION from sylk.applications.xmppgateway.configuration import XMPPGatewayConfig from sylk.applications.xmppgateway.datatypes import FrozenURI from sylk.applications.xmppgateway.logger import log from sylk.applications.xmppgateway.xmpp.jingle.session import JingleSession, JingleSessionManager -from sylk.applications.xmppgateway.xmpp.protocols import DiscoProtocol, JingleProtocol, MessageProtocol, MUCServerProtocol, PresenceProtocol +from sylk.applications.xmppgateway.xmpp.protocols import DiscoProtocol, JingleProtocol, MessageProtocol, MUCServerProtocol, MUCPresenceProtocol, PresenceProtocol from sylk.applications.xmppgateway.xmpp.server import SylkInternalComponent, SylkRouter, SylkS2SServerFactory, xmpp_logger from sylk.applications.xmppgateway.xmpp.session import XMPPChatSessionManager, XMPPMucSessionManager from sylk.applications.xmppgateway.xmpp.subscription import XMPPSubscriptionManager class XMPPManager(object): __metaclass__ = Singleton implements(IObserver) def __init__(self): config = XMPPGatewayConfig self.stopped = False self.domains = set(config.domains) self.muc_domains = set(['%s.%s' % (config.muc_prefix, domain) for domain in self.domains]) router = SylkRouter() self._server_service = ServerService(router) self._server_service.domains = self.domains | self.muc_domains self._server_service.logTraffic = False # done manually self._s2s_factory = SylkS2SServerFactory(self._server_service) self._s2s_factory.logTraffic = False # done manually # Setup internal components self._internal_component = SylkInternalComponent(router) self._internal_component.domains = self.domains self._internal_component.manager = self self._muc_component = SylkInternalComponent(router) self._muc_component.domains = self.muc_domains self._muc_component.manager = self # Setup protocols self.message_protocol = MessageProtocol() self.message_protocol.setHandlerParent(self._internal_component) self.presence_protocol = PresenceProtocol() self.presence_protocol.setHandlerParent(self._internal_component) self.disco_protocol = DiscoProtocol() self.disco_protocol.setHandlerParent(self._internal_component) self.disco_client_protocol = DiscoClientProtocol() self.disco_client_protocol.setHandlerParent(self._internal_component) self.muc_protocol = MUCServerProtocol() self.muc_protocol.setHandlerParent(self._muc_component) + self.muc_presence_protocol = MUCPresenceProtocol() + self.muc_presence_protocol.setHandlerParent(self._muc_component) + self.disco_muc_protocol = DiscoProtocol() self.disco_muc_protocol.setHandlerParent(self._muc_component) self.version_protocol = VersionHandler('SylkServer', SYLK_VERSION) self.version_protocol.setHandlerParent(self._internal_component) self.fallback_protocol = FallbackHandler() self.fallback_protocol.setHandlerParent(self._internal_component) self.fallback_muc_protocol = FallbackHandler() self.fallback_muc_protocol.setHandlerParent(self._muc_component) self.ping_protocol = PingHandler() self.ping_protocol.setHandlerParent(self._internal_component) self.jingle_protocol = JingleProtocol() self.jingle_protocol.setHandlerParent(self._internal_component) + self.jingle_coin_protocol = JingleProtocol() + self.jingle_coin_protocol.setHandlerParent(self._muc_component) + self._s2s_listener = None self.chat_session_manager = XMPPChatSessionManager() self.muc_session_manager = XMPPMucSessionManager() self.subscription_manager = XMPPSubscriptionManager() self.jingle_session_manager = JingleSessionManager() def start(self): self.stopped = False xmpp_logger.start() config = XMPPGatewayConfig if config.trace_xmpp and xmpp_logger._xmpptrace_filename is not None: log.msg('Logging XMPP trace to file "%s"' % xmpp_logger._xmpptrace_filename) self._s2s_listener = reactor.listenTCP(config.local_port, self._s2s_factory, interface=config.local_ip) listen_address = self._s2s_listener.getHost() log.msg("XMPP listener started on %s:%d" % (listen_address.host, listen_address.port)) self.chat_session_manager.start() self.muc_session_manager.start() self.subscription_manager.start() self.jingle_session_manager.start() notification_center = NotificationCenter() notification_center.add_observer(self, sender=self._internal_component) notification_center.add_observer(self, sender=self._muc_component) self._internal_component.startService() self._muc_component.startService() def stop(self): self.stopped = True self._s2s_listener.stopListening() self.jingle_session_manager.stop() self.subscription_manager.stop() self.muc_session_manager.stop() self.chat_session_manager.stop() self._internal_component.stopService() self._muc_component.stopService() notification_center = NotificationCenter() notification_center.remove_observer(self, sender=self._internal_component) notification_center.remove_observer(self, sender=self._muc_component) xmpp_logger.stop() def send_stanza(self, stanza): if self.stopped: return self._internal_component.send(stanza.to_xml_element()) def send_muc_stanza(self, stanza): if self.stopped: return self._muc_component.send(stanza.to_xml_element()) def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) # Process message stanzas def _NH_XMPPGotChatMessage(self, notification): message = notification.data.message try: session = self.chat_session_manager.sessions[(message.recipient.uri, message.sender.uri)] except KeyError: notification.center.post_notification('XMPPGotChatMessage', sender=self, data=notification.data) else: session.channel.send(message) def _NH_XMPPGotNormalMessage(self, notification): notification.center.post_notification('XMPPGotNormalMessage', sender=self, data=notification.data) def _NH_XMPPGotComposingIndication(self, notification): composing_indication = notification.data.composing_indication try: session = self.chat_session_manager.sessions[(composing_indication.recipient.uri, composing_indication.sender.uri)] except KeyError: notification.center.post_notification('XMPPGotComposingIndication', sender=self, data=notification.data) else: session.channel.send(composing_indication) def _NH_XMPPGotErrorMessage(self, notification): error_message = notification.data.error_message try: session = self.chat_session_manager.sessions[(error_message.recipient.uri, error_message.sender.uri)] except KeyError: notification.center.post_notification('XMPPGotErrorMessage', sender=self, data=notification.data) else: session.channel.send(error_message) def _NH_XMPPGotReceipt(self, notification): receipt = notification.data.receipt try: session = self.chat_session_manager.sessions[(receipt.recipient.uri, receipt.sender.uri)] except KeyError: pass else: session.channel.send(receipt) # Process presence stanzas def _NH_XMPPGotPresenceAvailability(self, notification): stanza = notification.data.presence_stanza if stanza.recipient.uri.resource is not None: # Skip directed presence return sender_uri = stanza.sender.uri sender_uri_bare = FrozenURI(sender_uri.user, sender_uri.host) try: subscription = self.subscription_manager.outgoing_subscriptions[(stanza.recipient.uri, sender_uri_bare)] except KeyError: # Ignore incoming presence stanzas if there is no subscription pass else: subscription.channel.send(stanza) def _NH_XMPPGotPresenceSubscriptionStatus(self, notification): stanza = notification.data.presence_stanza if stanza.sender.uri.resource is not None or stanza.recipient.uri.resource is not None: # Skip directed presence return if stanza.type in ('subscribed', 'unsubscribed'): try: subscription = self.subscription_manager.outgoing_subscriptions[(stanza.recipient.uri, stanza.sender.uri)] except KeyError: pass else: subscription.channel.send(stanza) elif stanza.type in ('subscribe', 'unsubscribe'): try: subscription = self.subscription_manager.incoming_subscriptions[(stanza.recipient.uri, stanza.sender.uri)] except KeyError: if stanza.type == 'subscribe': notification.center.post_notification('XMPPGotPresenceSubscriptionRequest', sender=self, data=NotificationData(stanza=stanza)) else: subscription.channel.send(stanza) def _NH_XMPPGotPresenceProbe(self, notification): stanza = notification.data.presence_stanza if stanza.recipient.uri.resource is not None: # Skip directed presence return sender_uri = stanza.sender.uri sender_uri_bare = FrozenURI(sender_uri.user, sender_uri.host) try: subscription = self.subscription_manager.incoming_subscriptions[(stanza.recipient.uri, sender_uri_bare)] except KeyError: notification.center.post_notification('XMPPGotPresenceSubscriptionRequest', sender=self, data=NotificationData(stanza=stanza)) else: subscription.channel.send(stanza) # Process muc stanzas def _NH_XMPPMucGotGroupChat(self, notification): message = notification.data.message muc_uri = FrozenURI(message.recipient.uri.user, message.recipient.uri.host) try: session = self.muc_session_manager.incoming[(muc_uri, message.sender.uri)] except KeyError: # Ignore groupchat messages if there was no session created pass else: session.channel.send(message) def _NH_XMPPMucGotPresenceAvailability(self, notification): stanza = notification.data.presence_stanza if not stanza.sender.uri.resource: return muc_uri = FrozenURI(stanza.recipient.uri.user, stanza.recipient.uri.host) try: session = self.muc_session_manager.incoming[(muc_uri, stanza.sender.uri)] except KeyError: if stanza.available: notification.center.post_notification('XMPPGotMucJoinRequest', sender=self, data=NotificationData(stanza=stanza)) else: notification.center.post_notification('XMPPGotMucLeaveRequest', sender=self, data=NotificationData(stanza=stanza)) else: session.channel.send(stanza) def _NH_XMPPMucGotInvitation(self, notification): invitation = notification.data.invitation data = NotificationData(sender=invitation.sender, recipient=invitation.recipient, participant=invitation.invited_user) notification.center.post_notification('XMPPGotMucAddParticipantRequest', sender=self, data=data) # Jingle def _NH_XMPPGotJingleSessionInitiate(self, notification): stanza = notification.data.stanza try: self.jingle_session_manager.sessions[stanza.jingle.sid] except KeyError: - session = JingleSession(self.jingle_protocol) + session = JingleSession(notification.data.protocol) session.init_incoming(stanza) session.send_ring_indication() def _NH_XMPPGotJingleSessionTerminate(self, notification): stanza = notification.data.stanza try: session = self.jingle_session_manager.sessions[stanza.jingle.sid] except KeyError: return session.handle_notification(notification) def _NH_XMPPGotJingleSessionInfo(self, notification): stanza = notification.data.stanza try: session = self.jingle_session_manager.sessions[stanza.jingle.sid] except KeyError: return session.handle_notification(notification) def _NH_XMPPGotJingleSessionAccept(self, notification): stanza = notification.data.stanza try: session = self.jingle_session_manager.sessions[stanza.jingle.sid] except KeyError: return session.handle_notification(notification) def _NH_XMPPGotJingleDescriptionInfo(self, notification): stanza = notification.data.stanza try: session = self.jingle_session_manager.sessions[stanza.jingle.sid] except KeyError: return session.handle_notification(notification) def _NH_XMPPGotJingleTransportInfo(self, notification): stanza = notification.data.stanza try: session = self.jingle_session_manager.sessions[stanza.jingle.sid] except KeyError: return session.handle_notification(notification) diff --git a/sylk/applications/xmppgateway/xmpp/protocols.py b/sylk/applications/xmppgateway/xmpp/protocols.py index d848a0a..2f5e2f6 100644 --- a/sylk/applications/xmppgateway/xmpp/protocols.py +++ b/sylk/applications/xmppgateway/xmpp/protocols.py @@ -1,343 +1,381 @@ # Copyright (C) 2012 AG Projects. See LICENSE for details # from application.notification import NotificationCenter, NotificationData from twisted.internet import defer, reactor from twisted.words.protocols.jabber.error import StanzaError from twisted.words.protocols.jabber.jid import JID -from wokkel import disco, ping -from wokkel.muc import UserPresence -from wokkel.xmppim import BasePresenceProtocol, MessageProtocol, PresenceProtocol +from wokkel import disco, muc, ping, xmppim from sylk.applications.xmppgateway.configuration import XMPPGatewayConfig from sylk.applications.xmppgateway.datatypes import Identity, FrozenURI from sylk.applications.xmppgateway.xmpp.stanzas import (RECEIPTS_NS, CHATSTATES_NS, MUC_USER_NS, ErrorStanza, NormalMessage, MessageReceipt, ChatMessage, ChatComposingIndication, AvailabilityPresence, SubscriptionPresence, ProbePresence, MUCAvailabilityPresence, GroupChatMessage, IncomingInvitationMessage) from sylk.applications.xmppgateway.xmpp.stanzas import jingle -__all__ = ['DiscoProtocol', 'JingleProtocol', 'MessageProtocol', 'MUCServerProtocol', 'PresenceProtocol'] +__all__ = ['DiscoProtocol', 'JingleProtocol', 'MessageProtocol', 'MUCServerProtocol', 'MUCPresenceProtocol', 'PresenceProtocol'] -class MessageProtocol(MessageProtocol): +class MessageProtocol(xmppim.MessageProtocol): messageTypes = None, 'normal', 'chat', 'headline', 'groupchat', 'error' def _onMessage(self, message): if message.handled: return messageType = message.getAttribute("type") if messageType not in self.messageTypes: message["type"] = 'normal' self.onMessage(message) def onMessage(self, msg): notification_center = NotificationCenter() sender_uri = FrozenURI.parse('xmpp:'+msg['from']) sender = Identity(sender_uri) recipient_uri = FrozenURI.parse('xmpp:'+msg['to']) recipient = Identity(recipient_uri) msg_type = msg.getAttribute('type') msg_id = msg.getAttribute('id', None) is_empty = msg.body is None and msg.html is None if msg_type == 'error': error_type = msg.error['type'] conditions = [(child.name, child.defaultUri) for child in msg.error.elements()] error_message = ErrorStanza('message', sender, recipient, error_type, conditions, id=msg_id) notification_center.post_notification('XMPPGotErrorMessage', sender=self.parent, data=NotificationData(error_message=error_message)) return if msg_type in (None, 'normal', 'chat') and not is_empty: body = None html_body = None if msg.html is not None: html_body = msg.html.toXml() if msg.body is not None: body = unicode(msg.body) try: elem = next(c for c in msg.elements() if c.uri == RECEIPTS_NS) except StopIteration: use_receipt = False else: use_receipt = elem.name == u'request' if msg_type == 'chat': message = ChatMessage(sender, recipient, body, html_body, id=msg_id, use_receipt=use_receipt) notification_center.post_notification('XMPPGotChatMessage', sender=self.parent, data=NotificationData(message=message)) else: message = NormalMessage(sender, recipient, body, html_body, id=msg_id, use_receipt=use_receipt) notification_center.post_notification('XMPPGotNormalMessage', sender=self.parent, data=NotificationData(message=message)) return # Check if it's a composing indication if msg_type == 'chat' and is_empty: for elem in msg.elements(): try: elem = next(c for c in msg.elements() if c.uri == CHATSTATES_NS) except StopIteration: pass else: composing_indication = ChatComposingIndication(sender, recipient, elem.name, id=msg_id) notification_center.post_notification('XMPPGotComposingIndication', sender=self.parent, data=NotificationData(composing_indication=composing_indication)) return # Check if it's a receipt acknowledgement if is_empty: try: elem = next(c for c in msg.elements() if c.uri == RECEIPTS_NS) except StopIteration: pass else: if elem.name == u'received' and msg_id is not None: receipt = MessageReceipt(sender, recipient, msg_id) notification_center.post_notification('XMPPGotReceipt', sender=self.parent, data=NotificationData(receipt=receipt)) -class PresenceProtocol(PresenceProtocol): +class PresenceProtocol(xmppim.PresenceProtocol): def availableReceived(self, stanza): sender_uri = FrozenURI.parse('xmpp:'+stanza.element['from']) sender = Identity(sender_uri) recipient_uri = FrozenURI.parse('xmpp:'+stanza.element['to']) recipient = Identity(recipient_uri) id = stanza.element.getAttribute('id') show = stanza.show statuses = stanza.statuses presence_stanza = AvailabilityPresence(sender, recipient, available=True, show=show, statuses=statuses, id=id) NotificationCenter().post_notification('XMPPGotPresenceAvailability', sender=self.parent, data=NotificationData(presence_stanza=presence_stanza)) def unavailableReceived(self, stanza): sender_uri = FrozenURI.parse('xmpp:'+stanza.element['from']) sender = Identity(sender_uri) recipient_uri = FrozenURI.parse('xmpp:'+stanza.element['to']) recipient = Identity(recipient_uri) id = stanza.element.getAttribute('id') presence_stanza = AvailabilityPresence(sender, recipient, available=False, id=id) NotificationCenter().post_notification('XMPPGotPresenceAvailability', sender=self.parent, data=NotificationData(presence_stanza=presence_stanza)) def _process_subscription_stanza(self, stanza): sender_uri = FrozenURI.parse('xmpp:'+stanza.element['from']) sender = Identity(sender_uri) recipient_uri = FrozenURI.parse('xmpp:'+stanza.element['to']) recipient = Identity(recipient_uri) id = stanza.element.getAttribute('id') type = stanza.element.getAttribute('type') presence_stanza = SubscriptionPresence(sender, recipient, type, id=id) NotificationCenter().post_notification('XMPPGotPresenceSubscriptionStatus', sender=self.parent, data=NotificationData(presence_stanza=presence_stanza)) def subscribedReceived(self, stanza): self._process_subscription_stanza(stanza) def unsubscribedReceived(self, stanza): self._process_subscription_stanza(stanza) def subscribeReceived(self, stanza): self._process_subscription_stanza(stanza) def unsubscribeReceived(self, stanza): self._process_subscription_stanza(stanza) def probeReceived(self, stanza): sender_uri = FrozenURI.parse('xmpp:'+stanza.element['from']) sender = Identity(sender_uri) recipient_uri = FrozenURI.parse('xmpp:'+stanza.element['to']) recipient = Identity(recipient_uri) id = stanza.element.getAttribute('id') presence_stanza = ProbePresence(sender, recipient, id=id) NotificationCenter().post_notification('XMPPGotPresenceProbe', sender=self.parent, data=NotificationData(presence_stanza=presence_stanza)) -class MUCServerProtocol(BasePresenceProtocol): +class MUCServerProtocol(xmppim.BasePresenceProtocol): messageTypes = None, 'normal', 'chat', 'groupchat' - presenceTypeParserMap = {'available': UserPresence, - 'unavailable': UserPresence} + presenceTypeParserMap = {'available': muc.UserPresence, + 'unavailable': muc.UserPresence} def connectionInitialized(self): - BasePresenceProtocol.connectionInitialized(self) + self.xmlstream.addObserver('/presence/x[@xmlns="%s"]' % muc.NS_MUC, self._onPresence) self.xmlstream.addObserver('/message', self._onMessage) def _onMessage(self, message): if message.handled: return messageType = message.getAttribute("type") if messageType == 'error': return if messageType not in self.messageTypes: message['type'] = 'normal' if messageType == 'groupchat': self.onGroupChat(message) else: to_uri = FrozenURI.parse('xmpp:'+message['to']) if to_uri.host in self.parent.domains: # Check if it's an invitation if message.x is not None and message.x.invite is not None and message.x.invite.uri == MUC_USER_NS: self.onInvitation(message) else: # TODO: give error, private messages not supported pass def onGroupChat(self, msg): sender_uri = FrozenURI.parse('xmpp:'+msg['from']) sender = Identity(sender_uri) recipient_uri = FrozenURI.parse('xmpp:'+msg['to']) recipient = Identity(recipient_uri) body = None html_body = None if msg.html is not None: html_body = msg.html.toXml() if msg.body is not None: body = unicode(msg.body) message = GroupChatMessage(sender, recipient, body, html_body, id=msg.getAttribute('id', None)) NotificationCenter().post_notification('XMPPMucGotGroupChat', sender=self.parent, data=NotificationData(message=message)) def onInvitation(self, msg): sender_uri = FrozenURI.parse('xmpp:'+msg['from']) sender = Identity(sender_uri) recipient_uri = FrozenURI.parse('xmpp:'+msg['to']) recipient = Identity(recipient_uri) invited_user_uri = FrozenURI.parse('xmpp:'+msg.x.invite['to']) invited_user = Identity(invited_user_uri) if msg.x.invite.reason is not None and msg.x.invite.reason.uri == MUC_USER_NS: reason = unicode(msg.x.invite.reason) else: reason = None invitation = IncomingInvitationMessage(sender, recipient, invited_user=invited_user, reason=reason, id=msg.getAttribute('id', None)) NotificationCenter().post_notification('XMPPMucGotInvitation', sender=self.parent, data=NotificationData(invitation=invitation)) def availableReceived(self, stanza): sender_uri = FrozenURI.parse('xmpp:'+stanza.element['from']) sender = Identity(sender_uri) recipient_uri = FrozenURI.parse('xmpp:'+stanza.element['to']) recipient = Identity(recipient_uri) id = stanza.element.getAttribute('id') presence_stanza = MUCAvailabilityPresence(sender, recipient, available=True, id=id) NotificationCenter().post_notification('XMPPMucGotPresenceAvailability', sender=self.parent, data=NotificationData(presence_stanza=presence_stanza)) def unavailableReceived(self, stanza): sender_uri = FrozenURI.parse('xmpp:'+stanza.element['from']) sender = Identity(sender_uri) recipient_uri = FrozenURI.parse('xmpp:'+stanza.element['to']) recipient = Identity(recipient_uri) id = stanza.element.getAttribute('id') presence_stanza = MUCAvailabilityPresence(sender, recipient, available=False, id=id) NotificationCenter().post_notification('XMPPMucGotPresenceAvailability', sender=self.parent, data=NotificationData(presence_stanza=presence_stanza)) class DiscoProtocol(disco.DiscoHandler): def info(self, requestor, target, nodeIdentifier): """ Gather data for a disco info request. @param requestor: The entity that sent the request. @type requestor: L{JID} @param target: The entity the request was sent to. @type target: L{JID} @param nodeIdentifier: The optional node being queried, or C{''}. @type nodeIdentifier: C{unicode} @return: Deferred with the gathered results from sibling handlers. @rtype: L{defer.Deferred} """ xmpp_manager = self.parent.manager if target.host not in xmpp_manager.domains | xmpp_manager.muc_domains: return defer.fail(StanzaError('service-unavailable')) elements = [] elements.append(disco.DiscoFeature(disco.NS_DISCO_INFO)) elements.append(disco.DiscoFeature(disco.NS_DISCO_ITEMS)) elements.append(disco.DiscoFeature('http://sylkserver.com')) if target.host in xmpp_manager.muc_domains: elements.append(disco.DiscoIdentity('conference', 'text', 'SylkServer Chat Service')) elements.append(disco.DiscoFeature('http://jabber.org/protocol/muc')) + elements.append(disco.DiscoFeature('urn:ietf:rfc:3264')) + elements.append(disco.DiscoFeature(jingle.NS_JINGLE)) + elements.append(disco.DiscoFeature(jingle.NS_JINGLE_APPS_RTP)) + elements.append(disco.DiscoFeature(jingle.NS_JINGLE_APPS_RTP_AUDIO)) + #elements.append(disco.DiscoFeature(jingle.NS_JINGLE_APPS_RTP_VIDEO)) + elements.append(disco.DiscoFeature(jingle.NS_JINGLE_ICE_UDP_TRANSPORT)) + elements.append(disco.DiscoFeature(jingle.NS_JINGLE_RAW_UDP_TRANSPORT)) if target.user: # We can't say much more here, because the actual conference may end up on a different server elements.append(disco.DiscoFeature('muc_temporary')) elements.append(disco.DiscoFeature('muc_unmoderated')) else: elements.append(disco.DiscoFeature(ping.NS_PING)) if not target.user: elements.append(disco.DiscoIdentity('gateway', 'simple', 'SylkServer')) elements.append(disco.DiscoIdentity('server', 'im', 'SylkServer')) else: elements.append(disco.DiscoIdentity('client', 'pc')) - elements.append(disco.DiscoFeature('urn:ietf:rfc:3264')) elements.append(disco.DiscoFeature('http://jabber.org/protocol/caps')) elements.append(disco.DiscoFeature('http://jabber.org/protocol/chatstates')) + elements.append(disco.DiscoFeature('urn:ietf:rfc:3264')) elements.append(disco.DiscoFeature(jingle.NS_JINGLE)) elements.append(disco.DiscoFeature(jingle.NS_JINGLE_APPS_RTP)) elements.append(disco.DiscoFeature(jingle.NS_JINGLE_APPS_RTP_AUDIO)) #elements.append(disco.DiscoFeature(jingle.NS_JINGLE_APPS_RTP_VIDEO)) elements.append(disco.DiscoFeature(jingle.NS_JINGLE_ICE_UDP_TRANSPORT)) elements.append(disco.DiscoFeature(jingle.NS_JINGLE_RAW_UDP_TRANSPORT)) return defer.succeed(elements) def items(self, requestor, target, nodeIdentifier): """ Gather data for a disco items request. @param requestor: The entity that sent the request. @type requestor: L{JID} @param target: The entity the request was sent to. @type target: L{JID} @param nodeIdentifier: The optional node being queried, or C{''}. @type nodeIdentifier: C{unicode} @return: Deferred with the gathered results from sibling handlers. @rtype: L{defer.Deferred} """ xmpp_manager = self.parent.manager items = [] if not target.user and target.host in xmpp_manager.domains: items.append(disco.DiscoItem(JID('%s.%s' % (XMPPGatewayConfig.muc_prefix, target.host)), name='Multi-User Chat')) return defer.succeed(items) class JingleProtocol(jingle.JingleHandler): # Functions here need to return immediately so that the IQ result is sent, so schedule them in the reactor # TODO: review and remove this, just post notifications? def onSessionInitiate(self, request): reactor.callLater(0, NotificationCenter().post_notification, 'XMPPGotJingleSessionInitiate', sender=self.parent, - data=NotificationData(stanza=request)) + data=NotificationData(stanza=request, protocol=self)) def onSessionTerminate(self, request): reactor.callLater(0, NotificationCenter().post_notification, 'XMPPGotJingleSessionTerminate', sender=self.parent, data=NotificationData(stanza=request)) def onSessionAccept(self, request): reactor.callLater(0, NotificationCenter().post_notification, 'XMPPGotJingleSessionAccept', sender=self.parent, data=NotificationData(stanza=request)) def onSessionInfo(self, request): reactor.callLater(0, NotificationCenter().post_notification, 'XMPPGotJingleSessionInfo', sender=self.parent, data=NotificationData(stanza=request)) def onDescriptionInfo(self, request): reactor.callLater(0, NotificationCenter().post_notification, 'XMPPGotJingleDescriptionInfo', sender=self.parent, data=NotificationData(stanza=request)) def onTransportInfo(self, request): reactor.callLater(0, NotificationCenter().post_notification, 'XMPPGotJingleTransportInfo', sender=self.parent, data=NotificationData(stanza=request)) + +class MUCPresenceProtocol(xmppim.PresenceProtocol): + """Protocol implementation to handle presence subscription to MUC URIs + """ + + def subscribeReceived(self, stanza): + """ + Subscription request was received. + """ + self.subscribed(stanza.sender, sender=stanza.recipient) + self.send_available(stanza) + + def unsubscribeReceived(self, stanza): + """ + Unsubscription request was received. + """ + self.unsubscribed(stanza.sender, sender=stanza.recipient) + + def probeReceived(self, stanza): + """ + Probe presence was received. + """ + self.send_available(stanza) + + def send_available(self, stanza): + sender_uri = FrozenURI.parse('xmpp:'+stanza.element['from']) + sender = Identity(sender_uri) + recipient_uri = FrozenURI.parse('xmpp:'+stanza.element['to']) + recipient = Identity(recipient_uri) + + available = AvailabilityPresence(sender=recipient, recipient=sender) + self.send(available.to_xml_element()) + diff --git a/sylk/applications/xmppgateway/xmpp/stanzas/jingle.py b/sylk/applications/xmppgateway/xmpp/stanzas/jingle.py index 794be29..89cbbd0 100644 --- a/sylk/applications/xmppgateway/xmpp/stanzas/jingle.py +++ b/sylk/applications/xmppgateway/xmpp/stanzas/jingle.py @@ -1,649 +1,650 @@ # Copyright (c) AG Projects # Copyright (c) Uday Verma # Copyright (c) Ralph Meijer. # """ XMPP Jingle Protocol. This protocol is specified in * XEP-0166 - http://xmpp.org/extensions/xep-0166.html * XEP-0167 - http://xmpp.org/extensions/xep-0167.html * XEP-0176 - http://xmpp.org/extensions/xep-0176.html * XEP-0177 - http://xmpp.org/extensions/xep-0177.html """ from twisted.words.xish import domish from twisted.words.protocols.jabber import error from wokkel.generic import Request from wokkel.subprotocols import IQHandlerMixin, XMPPHandler NS_JINGLE_BASE = 'urn:xmpp:jingle' NS_JINGLE = NS_JINGLE_BASE + ':1' NS_JINGLE_ERRORS = NS_JINGLE_BASE + ':errors:1' +NS_JINGLE_APPS_BASE = NS_JINGLE_BASE + ':apps' -NS_JINGLE_APPS_RTP = NS_JINGLE_BASE + ':apps:rtp:1' -NS_JINGLE_APPS_RTP_INFO = NS_JINGLE_BASE + ':apps:rtp:info:1' -NS_JINGLE_APPS_RTP_AUDIO = NS_JINGLE_BASE + ':apps:rtp:audio' -NS_JINGLE_APPS_RTP_VIDEO = NS_JINGLE_BASE + ':apps:rtp:video' +NS_JINGLE_APPS_RTP = NS_JINGLE_APPS_BASE + ':rtp:1' +NS_JINGLE_APPS_RTP_INFO = NS_JINGLE_APPS_BASE + ':rtp:info:1' +NS_JINGLE_APPS_RTP_AUDIO = NS_JINGLE_APPS_BASE + ':rtp:audio' +NS_JINGLE_APPS_RTP_VIDEO = NS_JINGLE_APPS_BASE + ':rtp:video' NS_JINGLE_ICE_UDP_TRANSPORT = NS_JINGLE_BASE + ':transports:ice-udp:1' NS_JINGLE_RAW_UDP_TRANSPORT = NS_JINGLE_BASE + ':transports:raw-udp:1' # XPath for Jingle IQ requests IQ_JINGLE_REQUEST = '/iq[@type="get" or @type="set"]/jingle[@xmlns="' + NS_JINGLE + '"]' class Parameter(object): """ A class representing a payload parameter """ def __init__(self, name, value): self.name, self.value = name, value @classmethod def fromElement(cls, element): return cls(element.getAttribute('name'), element.getAttribute('value')) def toElement(self, defaultUri=None): element = domish.Element((defaultUri, 'parameter')) element['name'] = self.name element['value'] = self.value or '' return element class Crypto(object): """ A crypto method which makes up the encryption to be used """ def __init__(self, crypto_suite, key_params, tag, session_params=None): self.crypto_suite, self.key_params, self.tag, self.session_params = crypto_suite, key_params, tag, session_params @classmethod def fromElement(cls, element): return cls(element.getAttribute('crypto-suite'), element.getAttribute('key-params'), element.getAttribute('tag'), element.getAttribute('session-params')) def toElement(self, defaultUri=None): element = domish.Element((defaultUri, 'crypto')) element['crypto-suite'] = self.crypto_suite element['key-params'] = self.key_params if self.session_params: element['session-params'] = self.session_params element['tag'] = self.tag return element class Encryption(object): """ A class representing encryption method """ def __init__(self, required=False, cryptos=None): self.required, self.cryptos = required, cryptos or [] @classmethod def fromElement(cls, element): cryptos = [] for child in element.elements(): if child.name == 'crypto': cryptos.append(Crypto.fromElement(child)) # TODO: parse ZRTP elements required = element.hasAttribute('required') and (element.getAttribute('required').lower() in ['true', '1']) return cls(required, cryptos) def toElement(self, defaultUri=None): element = domish.Element((defaultUri, 'encryption')) if self.required: element['required'] = '1' for c in self.cryptos: element.addChild(c.toElement(defaultUri)) return element class Bandwidth(object): """ A class representing the bandwidth element """ def __init__(self, typ, value): self.typ, self.value = typ, value @classmethod def fromElement(cls, element): return cls(element.getAttribute('type'), str(element)) def toElement(self, defaultUri=None): element = domish.Element((defaultUri, 'bandwidth')) element['type'] = self.typ element.addContent(self.value) return element class PayloadType(object): """ A class representing payload type """ def __init__(self, id, name, clockrate=0, channels=0, maxptime=None, ptime=None, parameters=None): self.id, self.name, self.clockrate, self.channels, \ self.maxptime, self.ptime, self.parameters = \ id, name, clockrate, channels, maxptime, ptime, parameters or [] @classmethod def fromElement(cls, element): def _sga(v, t): """ SafeGetAttribute """ try: return t(element.getAttribute(v)) except (TypeError, ValueError): return None params = [] for c in element.children: params.append(Parameter.fromElement(c)) return cls(int(element.getAttribute('id')), element.getAttribute('name'), _sga('clockrate', int) or 0, _sga('channels', int) or 0, _sga('maxptime', int) or 0, _sga('ptime', int) or 0, params) def toElement(self, defaultUri=None): element = domish.Element((defaultUri, 'payload-type')) def _aiv(k, v): """ AppendIfValid """ if v: element[k] = str(v) element['id'] = str(self.id) _aiv('name', self.name) _aiv('clockrate', self.clockrate) _aiv('channels', self.channels) _aiv('maxptime', self.maxptime) _aiv('ptime', self.ptime) for p in self.parameters: element.addChild(p.toElement()) return element class ICECandidate(object): """ A class representing an ICE candidate """ def __init__(self, component, foundation, generation, id, ip, network, port, priority, protocol, typ, related_addr=None, related_port=0): self.component, self.foundation, self.generation, \ self.id, self.ip, self.network, self.port, self.priority, \ self.protocol, self.typ, self.related_addr, self.related_port = \ component, foundation, generation, \ id, ip, network, port, priority, protocol, typ, \ related_addr, related_port @classmethod def fromElement(cls, element): def _gas(*names): """ GetAttributeS """ def default_val(t): return None if t is str else t() return [(t(element.getAttribute(name)) if element.hasAttribute(name) else default_val(t)) for name, t in names] return cls(*_gas(('component', int), ('foundation', int), ('generation', int), ('id', str), ('ip', str), ('network', int), ('port', int), ('priority', int), ('protocol', str), ('type', str), ('rel-addr', str), ('rel-port', int))) def toElement(self, defaultUri=None): element = domish.Element((defaultUri, 'candidate')) def _aas(*names): """ AddAttributeS """ for n, v in names: if v is not None: element[n] = str(v) _aas(*[('component', self.component), ('foundation', self.foundation), ('generation', self.generation), ('id', self.id), ('ip', self.ip), ('network', self.network), ('port', self.port), ('priority', self.priority), ('protocol', self.protocol), ('type', self.typ), ('rel-addr', self.related_addr), ('rel-port', self.related_port)]) return element class UDPCandidate(object): """ A class representing a UDP candidate """ def __init__(self, component, generation, id_, ip, port, protocol, type=None): self.component = component self.generation = generation self.id = id_ self.ip = ip self.port = port self.protocol = protocol self.type = type @classmethod def fromElement(cls, element): def _gas(*names): """ GetAttributeS """ def default_val(t): return None if t is str else t() return [(t(element.getAttribute(name)) if element.hasAttribute(name) else default_val(t)) for name, t in names] return cls(*_gas(('component', int), ('generation', int), ('id', str), ('ip', str), ('port', int), ('protocol', str), ('type', str))) def toElement(self, defaultUri=None): element = domish.Element((defaultUri, 'candidate')) def _aas(*names): """ AddAttributeS """ for n, v in names: if v: element[n] = str(v) _aas(*[('component', self.component), ('generation', self.generation), ('id', self.id), ('ip', self.ip), ('port', self.port), ('protocol', self.protocol), ('type', self.type)]) return element class ICERemoteCandidate(object): """ A class represeting a remote candidate entity """ def __init__(self, component, ip, port): self.component, self.ip, self.port = component, ip, port @classmethod def fromElement(cls, element): return cls(int(element.getAttribute('component') or '0'), element.getAttribute('ip'), int(element.getAttribute('port') or '0')) def toElement(self, defaultUri=None): element = domish.Element((defaultUri, 'remote-candidate')) element['component'] = str(self.component) element['ip'] = self.ip element['port'] = str(self.port) return element class IceUdpTransport(object): """ Represents the ICE-UDP transport type """ def __init__(self, pwd=None, ufrag=None, candidates=None, remote_candidate=None): self.password, self.ufrag, self.candidates, self.remote_candidate = \ pwd, ufrag, candidates or [], remote_candidate @classmethod def fromElement(cls, element): password = element.getAttribute('pwd') or None ufrag = element.getAttribute('ufrag') or None candidates = [] remote_candidate = None for child in element.elements(): if child.name == 'remote-candidate' and remote_candidate is None: remote_candidate = ICERemoteCandidate.fromElement(child) elif child.name == 'candidate': candidates.append(ICECandidate.fromElement(child)) return cls(pwd=password, ufrag=ufrag, candidates=candidates, remote_candidate=remote_candidate) def toElement(self, defaultUri=None): element = domish.Element((defaultUri or NS_JINGLE_ICE_UDP_TRANSPORT, 'transport')) if self.password: element['pwd'] = self.password if self.ufrag: element['ufrag'] = self.ufrag if self.remote_candidate: element.addChild(self.remote_candidate.toElement()) elif self.candidates: for c in self.candidates: element.addChild(c.toElement()) return element class RawUdpTransport(object): """ Represents the Raw-UDP transport type """ def __init__(self, candidates=None): self.candidates = candidates or [] @classmethod def fromElement(cls, element): candidates = [] for child in element.elements(): if child.name == 'candidate': candidates.append(UDPCandidate.fromElement(child)) return cls(candidates=candidates) def toElement(self, defaultUri=None): element = domish.Element((defaultUri or NS_JINGLE_RAW_UDP_TRANSPORT, 'transport')) for c in self.candidates: element.addChild(c.toElement()) return element class RTPDescription(object): """ A class representing a RTP description """ def __init__(self, name=None, media=None, ssrc=None, payloads=None, encryption=None, bandwidth=None): self.name, self.media, self.ssrc, self.payloads, \ self.encryption, self.bandwidth = \ name, media, ssrc, payloads or [], encryption, bandwidth @classmethod def fromElement(cls, element): plds = [] encryption, bandwidth = None, None for child in element.elements(): if child.name == 'payload-type': plds.append(PayloadType.fromElement(child)) if child.name == 'encryption': encryption = Encryption.fromElement(child) if child.name == 'bandwidth': bandwidth = Bandwidth.fromElement(child) return cls(element.getAttribute('name'), element.getAttribute('media'), element.getAttribute('ssrc'), plds, encryption, bandwidth) def toElement(self, defaultUri=None): element = domish.Element((defaultUri or NS_JINGLE_APPS_RTP, 'description')) if self.name: element['name'] = self.name if self.media: element['media'] = self.media for p in self.payloads: element.addChild(p.toElement(defaultUri)) if self.encryption: element.addChild(self.encryption.toElement(defaultUri)) if self.bandwidth: element.addChild(self.bandwidth.toElement(defaultUri)) return element class Content(object): """ A class indicating a single content item within a jingle request. """ def __init__(self, creator, name, disposition=None, senders=None): self.creator, self.name, self.disposition, self.senders = \ creator, name, disposition, senders self.description = None self.transport = None @classmethod def fromElement(cls, element): creator = element.getAttribute('creator') name = element.getAttribute('name') disposition = element.getAttribute('disposition') senders = element.getAttribute('senders') description, transport = None, None for c in element.elements(): if c.name == 'description' and c.uri == NS_JINGLE_APPS_RTP: description = RTPDescription.fromElement(c) elif c.name == 'transport' and c.uri == NS_JINGLE_ICE_UDP_TRANSPORT: transport = IceUdpTransport.fromElement(c) elif c.name == 'transport' and c.uri == NS_JINGLE_RAW_UDP_TRANSPORT: transport = RawUdpTransport.fromElement(c) ret = cls(creator, name, disposition, senders) ret.description = description ret.transport = transport return ret def toElement(self): element = domish.Element((None, 'content')) element['creator'] = self.creator element['name'] = self.name if self.disposition: element['disposition'] = self.disposition if self.senders: element['senders'] = self.senders if self.description: element.addChild(self.description.toElement()) if self.transport: element.addChild(self.transport.toElement()) return element class EmptyType(unicode): @classmethod def fromElement(cls, element): return cls(element.name) def toElement(self): return domish.Element((None, self)) class ReasonType(EmptyType): pass class AlternativeSessionReason(unicode): def __new__(cls, value): obj = unicode.__new__(cls, 'alternative-session') obj.sid = value return obj @classmethod def fromElement(cls, element): return cls(element.firstChildElement().children[0]) def toElement(self): element = domish.Element((None, self)) element.addElement('sid', content=self.sid) return element class Reason(object): def __init__(self, reason, text=None): self.value = reason self.text = text @classmethod def fromElement(cls, element): reason = None text = None for c in element.children: if c.name == 'text': text = c.children[0] elif c.name == 'alternative-session': reason = AlternativeSessionReason.fromElement(c) else: reason = ReasonType.fromElement(c) return cls(reason, text) def toElement(self): element = domish.Element((None, 'reason')) element.addChild(self.value.toElement()) if self.text: element.addElement('text', content=self.text) return element class Info(unicode): @classmethod def fromElement(cls, element): return cls(element.name) def toElement(self): return domish.Element((NS_JINGLE_APPS_RTP_INFO, self)) class MuteInfo(Info): def __new__(cls, value, creator, name): obj = unicode.__new__(cls, value) obj.creator = creator obj.name = name return obj @classmethod def fromElement(cls, element): return cls(element.name, element['creator'], element['name']) def toElement(self): element = super(MuteInfo, self).toElement() element['creator'] = self.creator element['name'] = self.name return element class Jingle(object): """ A class representing a Jingle element within an IQ request """ def __init__(self, action, sid, initiator=None, responder=None, content=None, reason=None, info=None): self.action = action self.sid = sid self.initiator = initiator self.responder = responder self.reason = reason self.info = info if not hasattr(content, '__iter__'): if content is not None: self.content = [content] else: self.content = [] else: self.content = content @classmethod def fromElement(cls, element): action = element.getAttribute('action') initiator = element.getAttribute('initiator') responder = element.getAttribute('responder') sid = element.getAttribute('sid') content = [] reason = None info = None for c in element.elements(): if c.name == 'content': content.append(Content.fromElement(c)) elif c.name == 'reason': reason = Reason.fromElement(c) elif c.uri == NS_JINGLE_APPS_RTP_INFO: if c.name in ('mute', 'unmute'): info = MuteInfo.fromElement(c) else: info = Info.fromElement(c) return cls(action, sid, initiator, responder, content=content, reason=reason, info=info) def toElement(self): element = domish.Element((NS_JINGLE, 'jingle')) element['action'] = self.action element['sid'] = self.sid if self.initiator: element['initiator'] = self.initiator if self.responder: element['responder'] = self.responder for c in self.content: element.addChild(c.toElement()) if self.reason: element.addChild(self.reason.toElement()) if self.info: element.addChild(self.info.toElement()) return element class JingleIq(Request): stanzaKind = 'iq' stanzaType = 'set' timeout = None childParsers = {(NS_JINGLE, 'jingle'): '_parseJingleElement'} def __init__(self, sender=None, recipient=None, jingle=None): Request.__init__(self, recipient, sender, self.stanzaType) self.jingle = jingle def _parseJingleElement(self, element): self.jingle = Jingle.fromElement(element) def toElement(self): element = Request.toElement(self) element.addChild(self.jingle.toElement()) return element class JingleHandler(XMPPHandler, IQHandlerMixin): iqHandlers = {IQ_JINGLE_REQUEST: '_onJingleRequest'} def connectionInitialized(self): self.xmlstream.addObserver(IQ_JINGLE_REQUEST, self.handleRequest) def sessionTerminate(self, sender, recipient, sid, reason=None): jingle = Jingle('session-terminate', sid, reason=reason) return JingleIq(sender=sender, recipient=recipient, jingle=jingle) def sessionInfo(self, sender, recipient, sid, info=None): jingle = Jingle('session-info', sid, info=info) return JingleIq(sender=sender, recipient=recipient, jingle=jingle) def sessionAccept(self, sender, recipient, payload): payload.action = 'session-accept' return JingleIq(sender=sender, recipient=recipient, jingle=payload) def sessionInitiate(self, sender, recipient, payload): payload.action = 'session-initiate' return JingleIq(sender=sender, recipient=recipient, jingle=payload) def _onJingleRequest(self, iq): request = JingleIq.fromElement(iq) method_name = 'on'+''.join(item.capitalize() for item in request.jingle.action.lower().split('-')) handler = getattr(self, method_name, None) if callable(handler): handler(request) else: raise error.StanzaError('bad-request')