erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin XEP-0166, XEP-0234, XEP-0261:

Browse source 
- transport and application data are now managed in separate dictionaries
    - new client.IQ method is used
    - new buildAction helper method for applications and transports
    - "role" is now stored in session_data
    - "senders" is now stored in content_data
    - plugin XEP-0166: "transport-info" action is now managed
    - plugin XEP-0166: application namespace and handler are now managed in a namedtuple
    - plugin XEP-0234: <range/> element is added by responder if not already present
This commit is contained in:
Goffi 2015-11-02 22:02:41 +01:00
commit 170d697016
1 changed files with 180 additions and 86 deletions
Download patch file Download diff file Expand all files Collapse all files

213
src/plugins/plugin_xep_0166.py
View file

@ -28,7 +28,7 @@ from twisted.words.protocols.jabber import jid
# from twisted.words.xish import domish
from twisted.internet import defer
# from wokkel import disco, iwokkel, data_form, compat
from wokkel import disco, iwokkel, compat
from wokkel import disco, iwokkel
from twisted.words.protocols.jabber import error
from twisted.words.protocols.jabber import xmlstream
# from sat.core import exceptions
@ -49,8 +49,6 @@ JINGLE_REQUEST = IQ_SET + '/jingle[@xmlns="' + NS_JINGLE + '"]'
STATE_PENDING = "PENDING"
STATE_ACTIVE = "ACTIVE"
STATE_ENDED = "ENDED"
INITIATOR = "initiator"
RESPONDER = "responder"
CONFIRM_TXT = D_("{entity} want to start a jingle session with you, do you accept ?")
PLUGIN_INFO = {
@ -64,10 +62,13 @@ PLUGIN_INFO = {
}
ApplicationData = namedtuple('ApplicationData', ('namespace', 'handler'))
TransportData = namedtuple('TransportData', ('namespace', 'handler', 'priority'))
class XEP_0166(object):
ROLE_INITIATOR = "initiator"
ROLE_RESPONDER = "responder"
TRANSPORT_DATAGRAM='UDP'
TRANSPORT_STREAMING='TCP'
REASON_SUCCESS='success'
@ -77,6 +78,7 @@ class XEP_0166(object):
A_SESSION_INITIATE = "session-initiate"
A_SESSION_ACCEPT = "session-accept"
A_SESSION_TERMINATE = "session-terminate"
A_TRANSPORT_INFO = "transport-info"
# non standard actions
A_PREPARE_INITIATOR = "prepare-initiator" # initiator must prepare tranfer
A_PREPARE_RESPONDER = "prepare-responder" # responder must prepare tranfer
@ -111,7 +113,7 @@ class XEP_0166(object):
## helpers methods to build stanzas ##
def _buildJingleElt(self, client, session, action):
iq_elt = compat.IQ(client.xmlstream, 'set')
iq_elt = client.IQ('set')
iq_elt['from'] = client.jid.full()
iq_elt['to'] = session['to_jid'].full()
jingle_elt = iq_elt.addElement("jingle", NS_JINGLE)
@ -204,7 +206,8 @@ class XEP_0166(object):
"""
if namespace in self._applications:
raise exceptions.ConflictError(u"Trying to register already registered namespace {}".format(namespace))
self._applications[namespace] = handler
self._applications[namespace] = ApplicationData(namespace=namespace, handler=handler)
log.debug(u"new jingle application registered")
def registerTransport(self, namespace, transport_type, handler, priority=0):
"""Register a transport plugin
@ -223,10 +226,38 @@ class XEP_0166(object):
raise exceptions.ConflictError(u"Trying to register already registered namespace {}".format(namespace))
transport_data = TransportData(namespace=namespace, handler=handler, priority=priority)
self._type_transports[transport_type].append(transport_data)
self._type_transports[transport_type].sort(key=lambda transport_data: transport_data.priority)
self._type_transports[transport_type].sort(key=lambda transport_data: transport_data.priority, reverse=True)
self._transports[namespace] = transport_data
log.debug(u"new jingle transport registered")
def buildAction(self, action, session, content_name, profile=C.PROF_KEY_NONE):
"""Build an element according to requested action
@param action(unicode): a jingle action (see XEP-0166 §7.2),
session-* actions are not managed here
@param session(dict): jingle session data
@param content_name(unicode): name of the content terminated
@param profile: %(doc_profile)s
@return (tuple[domish.Element, domish.Element]): parent <iq> element, <transport> or <description> element, according to action
"""
client = self.host.getClient(profile)
# we first build iq, jingle and content element which are the same in every cases
iq_elt, jingle_elt = self._buildJingleElt(client, session, action)
# FIXME: XEP-0260 § 2.3 Ex 5 has an initiator attribute, but it should not according to XEP-0166 §7.1 table 1, must be checked
content_data= session['contents'][content_name]
content_elt = jingle_elt.addElement('content')
content_elt['name'] = content_name
content_elt['creator'] = content_data['creator']
if action == XEP_0166.A_TRANSPORT_INFO:
context_elt = transport_elt = content_elt.addElement('transport', content_data['transport'].namespace)
transport_elt['sid'] = content_data['transport_data']['sid']
else:
raise exceptions.InternalError(u"unmanaged action {}".format(action))
return iq_elt, context_elt
@defer.inlineCallbacks
def initiate(self, to_jid, contents, profile=C.PROF_KEY_NONE):
"""Send a session initiation request
@ -239,6 +270,8 @@ class XEP_0166(object):
- transport_type(unicode): type of transport to use (see XEP-0166 §8)
default to TRANSPORT_STREAMING
- name(unicode): name of the content
- senders(unicode): One of XEP_0166.ROLE_INITIATOR, XEP_0166.ROLE_RESPONDER, both or none
default to BOTH (see XEP-0166 §7.3)
- app_args(list): args to pass to the application plugin
- app_kwargs(dict): keyword args to pass to the application plugin
@param profile: %(doc_profile)s
@ -251,6 +284,7 @@ class XEP_0166(object):
session = client.jingle_sessions[sid] = {'id': sid,
'state': STATE_PENDING,
'initiator': initiator,
'role': XEP_0166.ROLE_INITIATOR,
'to_jid': to_jid,
'started': time.time(),
'contents': {}
@ -264,21 +298,24 @@ class XEP_0166(object):
# we get the application plugin
app_ns = content['app_ns']
try:
application_handler = self._applications[app_ns]
application = self._applications[app_ns]
except KeyError:
raise exceptions.InternalError(u"No application registered for {}".format(app_ns))
# and the transport plugin
transport_type = content.get('transport_type', XEP_0166.TRANSPORT_STREAMING)
try:
transport_handler = self._type_transports[transport_type][0].handler
transport = self._type_transports[transport_type][0]
except IndexError:
raise exceptions.InternalError(u"No transport registered for {}".format(transport_type))
# we build the session data
content_data = {'application': application_handler,
'transport': transport_handler,
'creator': INITIATOR,
content_data = {'application': application,
'application_data': {},
'transport': transport,
'transport_data': {},
'creator': XEP_0166.ROLE_INITIATOR,
'senders': content.get('senders', 'both'),
}
try:
content_name = content['name']
@ -293,16 +330,20 @@ class XEP_0166(object):
content_elt = jingle_elt.addElement('content')
content_elt['creator'] = content_data['creator']
content_elt['name'] = content_name
try:
content_elt['senders'] = content['senders']
except KeyError:
pass
# then the description element
app_args = content.get('app_args', [])
app_kwargs = content.get('app_kwargs', {})
app_kwargs['profile'] = profile
desc_elt = yield application_handler.jingleSessionInit(session, content_name, *app_args, **app_kwargs)
desc_elt = yield application.handler.jingleSessionInit(session, content_name, *app_args, **app_kwargs)
content_elt.addChild(desc_elt)
# and the transport one
transport_elt = yield transport_handler.jingleSessionInit(session, content_name, profile)
transport_elt = yield transport.handler.jingleSessionInit(session, content_name, profile)
content_elt.addChild(transport_elt)
d = iq_elt.send()
@ -325,7 +366,7 @@ class XEP_0166(object):
## defaults methods called when plugin doesn't have them ##
def jingleRequestConfirmationDefault(self, session, desc_elt, profile):
def jingleRequestConfirmationDefault(self, action, session, content_name, desc_elt, profile):
"""This method request confirmation for a jingle session"""
log.debug(u"Using generic jingle confirmation method")
return xml_tools.deferConfirm(self.host, _(CONFIRM_TXT).format(entity=session['to_jid'].full()), _('Confirm Jingle session'), profile=profile)
@ -373,6 +414,7 @@ class XEP_0166(object):
session = client.jingle_sessions[sid] = {'id': sid,
'state': STATE_PENDING,
'initiator': to_jid,
'role': XEP_0166.ROLE_RESPONDER,
'to_jid': to_jid,
'started': time.time(),
}
@ -392,20 +434,26 @@ class XEP_0166(object):
self.onSessionTerminate(client, request, jingle_elt, session)
elif action == XEP_0166.A_SESSION_ACCEPT:
self.onSessionAccept(client, request, jingle_elt, session)
elif action == XEP_0166.A_TRANSPORT_INFO:
self.onTransportInfo(client, request, jingle_elt, session)
else:
raise exceptions.InternalError(u"Unknown action {}".format(session['state']))
raise exceptions.InternalError(u"Unknown action {}".format(action))
## Actions callbacks ##
def _parseElements(self, jingle_elt, session, request, client, new=False, creator=INITIATOR):
def _parseElements(self, jingle_elt, session, request, client, new=False, creator=ROLE_INITIATOR, with_application=True, with_transport=True):
"""Parse contents elements and fill contents_dict accordingly
after the parsing, contents_dict will containt handlers, "desc_elt" and "transport_elt"
@param jingle_elt(domish.Element): parent <jingle> element, containing one or more <content>
@param contents_dict(dict): session data for contents, the key is the name of the content
@param session(dict): session data
@param request(domish.Element): the whole request
@param client: %(doc_client)s
@param new(bool): if new the content is new and must be created,
else the content must exists, and session data will be filled
@param creator(unicode): only used if new is True: creating pear (see § 7.3)
@param with_application(bool): if True, raise an error if there is no <description> element else ignore it
@param with_transport(bool): if True, raise an error if there is no <transport> element else ignore it
@raise exceptions.CancelError: the error is treated the calling method can cancel the treatment (i.e. return)
"""
contents_dict = session['contents']
@ -419,7 +467,9 @@ class XEP_0166(object):
if not name or name in contents_dict:
self.sendError('bad-request', session['id'], request, client.profile)
raise exceptions.CancelError
content_data = contents_dict[name] = {'creator': creator}
content_data = contents_dict[name] = {'creator': creator,
'senders': content_elt.attributes.get('senders', 'both'),
}
else:
# the content must exist, we check it
try:
@ -430,57 +480,61 @@ class XEP_0166(object):
raise exceptions.CancelError
# application
desc_elt = content_elt.description
if not desc_elt:
self.sendError('bad-request', session['id'], request, client.profile)
raise exceptions.CancelError
if new:
# the content is new, we need to check and link the application_handler
app_ns = desc_elt.uri
if not app_ns or app_ns == NS_JINGLE:
if with_application:
desc_elt = content_elt.description
if not desc_elt:
self.sendError('bad-request', session['id'], request, client.profile)
raise exceptions.CancelError
try:
application_handler = self._applications[app_ns]
except KeyError:
log.warning(u"Unmanaged application namespace [{}]".format(app_ns))
self.sendError('service-unavailable', session['id'], request, client.profile)
raise exceptions.CancelError
if new:
# the content is new, we need to check and link the application
app_ns = desc_elt.uri
if not app_ns or app_ns == NS_JINGLE:
self.sendError('bad-request', session['id'], request, client.profile)
raise exceptions.CancelError
content_data['application'] = application_handler
else:
# the content exists, we check that we have not a former desc_elt
if 'desc_elt' in content_data:
raise exceptions.InternalError(u"desc_elt should not exist at this point")
try:
application = self._applications[app_ns]
except KeyError:
log.warning(u"Unmanaged application namespace [{}]".format(app_ns))
self.sendError('service-unavailable', session['id'], request, client.profile)
raise exceptions.CancelError
content_data['desc_elt'] = desc_elt
content_data['application'] = application
content_data['application_data'] = {}
else:
# the content exists, we check that we have not a former desc_elt
if 'desc_elt' in content_data:
raise exceptions.InternalError(u"desc_elt should not exist at this point")
content_data['desc_elt'] = desc_elt
# transport
transport_elt = content_elt.transport
if not transport_elt:
self.sendError('bad-request', session['id'], request, client.profile)
raise exceptions.CancelError
if new:
# the content is new, we need to check and link the transport_handler
transport_ns = transport_elt.uri
if not app_ns or app_ns == NS_JINGLE:
if with_transport:
transport_elt = content_elt.transport
if not transport_elt:
self.sendError('bad-request', session['id'], request, client.profile)
raise exceptions.CancelError
try:
transport_handler = self._transports[transport_ns].handler
except KeyError:
raise exceptions.InternalError(u"No transport registered for namespace {}".format(transport_ns))
content_data['transport'] = transport_handler
else:
# the content exists, we check that we have not a former transport_elt
if 'transport_elt' in content_data:
raise exceptions.InternalError(u"desc_elt should not exist at this point")
if new:
# the content is new, we need to check and link the transport
transport_ns = transport_elt.uri
if not app_ns or app_ns == NS_JINGLE:
self.sendError('bad-request', session['id'], request, client.profile)
raise exceptions.CancelError
content_data['transport_elt'] = transport_elt
try:
transport = self._transports[transport_ns]
except KeyError:
raise exceptions.InternalError(u"No transport registered for namespace {}".format(transport_ns))
content_data['transport'] = transport
content_data['transport_data'] = {}
else:
# the content exists, we check that we have not a former transport_elt
if 'transport_elt' in content_data:
raise exceptions.InternalError(u"desc_elt should not exist at this point")
content_data['transport_elt'] = transport_elt
def _callPlugins(self, action, session, app_method_name='jingleHandler', transp_method_name='jingleHandler', app_default_cb=None, transp_default_cb=None, delete=True, elements=True, profile=C.PROF_KEY_NONE):
"""Call application and transport plugin methods for all contents
@ -511,7 +565,7 @@ class XEP_0166(object):
if method_name is None:
continue
handler = content_data[handler_key]
handler = content_data[handler_key].handler
try:
method = getattr(handler, method_name)
except AttributeError:
@ -546,7 +600,7 @@ class XEP_0166(object):
session['contents'] = contents_dict = {}
try:
self._parseElements(jingle_elt, session, request, client, True, INITIATOR)
self._parseElements(jingle_elt, session, request, client, True, XEP_0166.ROLE_INITIATOR)
except exceptions.CancelError:
return
@ -591,19 +645,19 @@ class XEP_0166(object):
for content_name, content_data in session['contents'].iteritems():
content_elt = jingle_elt.addElement('content')
content_elt['creator'] = INITIATOR
content_elt['creator'] = XEP_0166.ROLE_INITIATOR
content_elt['name'] = content_name
application_handler = content_data['application']
app_session_accept_cb = application_handler.jingleHandler
application = content_data['application']
app_session_accept_cb = application.handler.jingleHandler
app_d = defer.maybeDeferred(app_session_accept_cb,
XEP_0166.A_SESSION_INITIATE, session, content_name, content_data.pop('desc_elt'), client.profile)
app_d.addCallback(addElement, content_elt)
defers_list.append(app_d)
transport_handler = content_data['transport']
transport_session_accept_cb = transport_handler.jingleHandler
transport = content_data['transport']
transport_session_accept_cb = transport.handler.jingleHandler
transport_d = defer.maybeDeferred(transport_session_accept_cb,
XEP_0166.A_SESSION_INITIATE, session, content_name, content_data.pop('transport_elt'), client.profile)
@ -628,7 +682,7 @@ class XEP_0166(object):
client.xmlstream.send(xmlstream.toResponse(request, 'result'))
def onSessionAccept(self, client, request, jingle_elt, session):
"""Method called one sesion is accepted
"""Method called once session is accepted
This method is only called for initiator
@param client: %(doc_client)s
@ -656,6 +710,33 @@ class XEP_0166(object):
# after negociations we start the transfer
negociate_dlist.addCallback(lambda dummy: self._callPlugins(XEP_0166.A_START, session, app_method_name=None, elements=False, profile=client.profile))
def onTransportInfo(self, client, request, jingle_elt, session):
"""Method called when a transport-info action is received from other peer
The request is parsed, and jingleHandler is called on concerned transport plugin(s)
@param client: %(doc_client)s
@param request(domish.Element): full <iq> request
@param jingle_elt(domish.Element): the <jingle> element
@param session(dict): session data
"""
log.debug(u"Jingle session {} has been accepted".format(session['id']))
try:
self._parseElements(jingle_elt, session, request, client, with_application=False)
except exceptions.CancelError:
return
# The parsing was OK, we send the <iq> result
client.xmlstream.send(xmlstream.toResponse(request, 'result'))
for content_name, content_data in session['contents'].iteritems():
try:
transport_elt = content_data.pop('transport_elt')
except KeyError:
continue
else:
content_data['transport'].handler.jingleHandler(XEP_0166.A_TRANSPORT_INFO, session, content_name, transport_elt, client.profile)
class XEP_0166_handler(xmlstream.XMPPHandler):
implements(iwokkel.IDisco)