From 9141922d5771a0f5a190fd51e99f3e1aa30941ad Mon Sep 17 00:00:00 2001 From: Goffi Date: Mon, 15 May 2023 16:23:11 +0200 Subject: [PATCH] plugin XEP-0166: refactoring, and various improvments: - add models for transport and applications handlers and linked data - split models into separate file - some type hints - some documentation comments - add actions to prepare confirmation, useful to do initial parsing of all contents - application arg/kwargs and some transport data can be initialised during Jingle `initiate` call, this is notably useful when a call is made with transport data (this is the call for A/V calls where codecs and ICE candidate can be specified when starting a call) - session data can be specified during Jingle `initiate` call - new `store_in_session` argument in `_parse_elements`, which can be used to avoid race-condition when a context element ( or ) is being parsed for an action while an other action happens (like `transport-info`) - don't sed `sid` in `transport_elt` during a `transport-info` action anymore in `build_action`: this is specific to Jingle File Transfer and has been moved there rel 419 --- sat/core/core_types.py | 1 + .../__init__.py} | 434 ++++++++++++------ sat/plugins/plugin_xep_0166/models.py | 183 ++++++++ sat/plugins/plugin_xep_0215.py | 2 +- sat/plugins/plugin_xep_0260.py | 6 + sat/plugins/plugin_xep_0353.py | 21 +- 6 files changed, 494 insertions(+), 153 deletions(-) rename sat/plugins/{plugin_xep_0166.py => plugin_xep_0166/__init__.py} (78%) create mode 100644 sat/plugins/plugin_xep_0166/models.py diff --git a/sat/core/core_types.py b/sat/core/core_types.py index 58ecbc45..e676708b 100644 --- a/sat/core/core_types.py +++ b/sat/core/core_types.py @@ -27,6 +27,7 @@ from twisted.words.xish import domish class SatXMPPEntity: + profile: str jid: t_jid.JID is_component: bool server_jid: t_jid.JID diff --git a/sat/plugins/plugin_xep_0166.py b/sat/plugins/plugin_xep_0166/__init__.py similarity index 78% rename from sat/plugins/plugin_xep_0166.py rename to sat/plugins/plugin_xep_0166/__init__.py index 4275fcf8..16aa8e2e 100644 --- a/sat/plugins/plugin_xep_0166.py +++ b/sat/plugins/plugin_xep_0166/__init__.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 -# SàT plugin for Jingle (XEP-0166) +# Libervia plugin for Jingle (XEP-0166) # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify @@ -17,9 +17,8 @@ # along with this program. If not, see . -from collections import namedtuple import time -from typing import Any, Dict, Tuple +from typing import Any, Callable, Dict, Final, List, Optional, Tuple import uuid from twisted.internet import defer @@ -40,20 +39,30 @@ from sat.core.log import getLogger from sat.tools import xml_tools from sat.tools import utils +from .models import ( + ApplicationData, + BaseApplicationHandler, + BaseTransportHandler, + ContentData, + TransportData, +) + log = getLogger(__name__) -IQ_SET = '/iq[@type="set"]' -NS_JINGLE = "urn:xmpp:jingle:1" -NS_JINGLE_ERROR = "urn:xmpp:jingle:errors:1" -JINGLE_REQUEST = IQ_SET + '/jingle[@xmlns="' + NS_JINGLE + '"]' -STATE_PENDING = "PENDING" -STATE_ACTIVE = "ACTIVE" -STATE_ENDED = "ENDED" -CONFIRM_TXT = D_("{entity} want to start a jingle session with you, do you accept ?") +IQ_SET : Final = '/iq[@type="set"]' +NS_JINGLE : Final = "urn:xmpp:jingle:1" +NS_JINGLE_ERROR : Final = "urn:xmpp:jingle:errors:1" +JINGLE_REQUEST : Final = f'{IQ_SET}/jingle[@xmlns="{NS_JINGLE}"]' +STATE_PENDING : Final = "PENDING" +STATE_ACTIVE : Final = "ACTIVE" +STATE_ENDED : Final = "ENDED" +CONFIRM_TXT : Final = D_( + "{entity} want to start a jingle session with you, do you accept ?" +) -PLUGIN_INFO = { +PLUGIN_INFO : Final = { C.PI_NAME: "Jingle", C.PI_IMPORT_NAME: "XEP-0166", C.PI_TYPE: "XEP", @@ -65,39 +74,51 @@ PLUGIN_INFO = { } -ApplicationData = namedtuple("ApplicationData", ("namespace", "handler")) -TransportData = namedtuple("TransportData", ("namespace", "handler", "priority")) - - class XEP_0166: - namespace = NS_JINGLE - ROLE_INITIATOR = "initiator" - ROLE_RESPONDER = "responder" - TRANSPORT_DATAGRAM = "UDP" - TRANSPORT_STREAMING = "TCP" - REASON_SUCCESS = "success" - REASON_DECLINE = "decline" - REASON_FAILED_APPLICATION = "failed-application" - REASON_FAILED_TRANSPORT = "failed-transport" - REASON_CONNECTIVITY_ERROR = "connectivity-error" - A_SESSION_INITIATE = "session-initiate" - A_SESSION_ACCEPT = "session-accept" - A_SESSION_TERMINATE = "session-terminate" - A_SESSION_INFO = "session-info" - A_TRANSPORT_REPLACE = "transport-replace" - A_TRANSPORT_ACCEPT = "transport-accept" - A_TRANSPORT_REJECT = "transport-reject" - A_TRANSPORT_INFO = "transport-info" + namespace : Final = NS_JINGLE + + ROLE_INITIATOR : Final = "initiator" + ROLE_RESPONDER : Final = "responder" + + TRANSPORT_DATAGRAM : Final = "UDP" + TRANSPORT_STREAMING : Final = "TCP" + + REASON_SUCCESS : Final = "success" + REASON_DECLINE : Final = "decline" + REASON_FAILED_APPLICATION : Final = "failed-application" + REASON_FAILED_TRANSPORT : Final = "failed-transport" + REASON_CONNECTIVITY_ERROR : Final = "connectivity-error" + + # standard actions + + A_SESSION_INITIATE : Final = "session-initiate" + A_SESSION_ACCEPT : Final = "session-accept" + A_SESSION_TERMINATE : Final = "session-terminate" + A_SESSION_INFO : Final = "session-info" + A_TRANSPORT_REPLACE : Final = "transport-replace" + A_TRANSPORT_ACCEPT : Final = "transport-accept" + A_TRANSPORT_REJECT : Final = "transport-reject" + A_TRANSPORT_INFO : Final = "transport-info" + # non standard actions - A_PREPARE_INITIATOR = "prepare-initiator" # initiator must prepare tranfer - A_PREPARE_RESPONDER = "prepare-responder" # responder must prepare tranfer - A_ACCEPTED_ACK = ( + + #: called before the confirmation request, first event for responder, useful for + #: parsing + A_PREPARE_CONFIRMATION : Final = "prepare-confirmation" + #: initiator must prepare tranfer + A_PREPARE_INITIATOR : Final = "prepare-initiator" + #: responder must prepare tranfer + A_PREPARE_RESPONDER : Final = "prepare-responder" + #; session accepted ack has been received from initiator + A_ACCEPTED_ACK : Final = ( "accepted-ack" - ) # session accepted ack has been received from initiator - A_START = "start" # application can start - A_DESTROY = ( + ) + A_START : Final = "start" # application can start + #: called when a transport is destroyed (e.g. because it is remplaced). Used to do + #: cleaning operations + A_DESTROY : Final = ( "destroy" - ) # called when a transport is destroyed (e.g. because it is remplaced). Used to do cleaning operations + ) def __init__(self, host): log.info(_("plugin Jingle initialization")) @@ -116,6 +137,22 @@ class XEP_0166: def get_handler(self, client): return XEP_0166_handler(self) + def get_session(self, client: SatXMPPEntity, session_id: str) -> dict: + """Retrieve session from its SID + + @param session_id: session ID + @return: found session + + @raise exceptions.NotFound: no session with this SID has been found + """ + try: + return client.jingle_sessions[session_id] + except KeyError: + raise exceptions.NotFound( + f"No session with SID {session_id} found" + ) + + def _del_session(self, client, sid): try: del client.jingle_sessions[sid] @@ -128,7 +165,12 @@ class XEP_0166: ## helpers methods to build stanzas ## - def _build_jingle_elt(self, client, session, action): + def _build_jingle_elt( + self, + client: SatXMPPEntity, + session: dict, + action: str + ) -> Tuple[xmlstream.IQ, domish.Element]: iq_elt = client.IQ("set") iq_elt["from"] = session['local_jid'].full() iq_elt["to"] = session["peer_jid"].full() @@ -224,43 +266,53 @@ class XEP_0166: ## methods used by other plugins ## - def register_application(self, namespace, handler): + def register_application( + self, + namespace: str, + handler: BaseApplicationHandler + ) -> None: """Register an application plugin @param namespace(unicode): application namespace managed by the plugin @param handler(object): instance of a class which manage the application. May have the following methods: - - requestConfirmation(session, desc_elt, client): + - request_confirmation(session, desc_elt, client): - if present, it is called on when session must be accepted. - if it return True the session is accepted, else rejected. A Deferred can be returned - if not present, a generic accept dialog will be used - - jingle_session_init(client, self, session, content_name[, *args, **kwargs]): must return the domish.Element used for initial content - - jingle_handler(client, self, action, session, content_name, transport_elt): + - jingle_session_init( + client, self, session, content_name[, *args, **kwargs] + ): must return the domish.Element used for initial content + - jingle_handler( + client, self, action, session, content_name, transport_elt + ): called on several action to negociate the application or transport - jingle_terminate: called on session terminate, with reason_elt May be used to clean session """ if namespace in self._applications: raise exceptions.ConflictError( - "Trying to register already registered namespace {}".format(namespace) + f"Trying to register already registered namespace {namespace}" ) self._applications[namespace] = ApplicationData( namespace=namespace, handler=handler ) log.debug("new jingle application registered") - def register_transport(self, namespace, transport_type, handler, priority=0): + def register_transport( + self, + namespace: str, + transport_type: str, + handler: BaseTransportHandler, + priority: int = 0 + ) -> None: """Register a transport plugin - @param namespace(unicode): the XML namespace used for this transport - @param transport_type(unicode): type of transport to use (see XEP-0166 §8) - @param handler(object): instance of a class which manage the application. - Must have the following methods: - - jingle_session_init(client, self, session, content_name[, *args, **kwargs]): must return the domish.Element used for initial content - - jingle_handler(client, self, action, session, content_name, transport_elt): - called on several action to negociate the application or transport - @param priority(int): priority of this transport + @param namespace: the XML namespace used for this transport + @param transport_type: type of transport to use (see XEP-0166 §8) + @param handler: instance of a class which manage the application. + @param priority: priority of this transport """ assert transport_type in ( XEP_0166.TRANSPORT_DATAGRAM, @@ -314,31 +366,50 @@ class XEP_0166: content_elt.addChild(transport_elt) iq_elt.send() - def build_action(self, client, action, session, content_name): + def build_action( + self, + client: SatXMPPEntity, + action: str, + session: dict, + content_name: str, + iq_elt: Optional[xmlstream.IQ] = None, + context_elt: Optional[domish.Element] = None + ) -> Tuple[xmlstream.IQ, domish.Element]: """Build an element according to requested action - @param action(unicode): a jingle action (see XEP-0166 §7.2), + @param action: a jingle action (see XEP-0166 §7.2), session-* actions are not managed here transport-replace is managed in the dedicated [transport_replace] method - @param session(dict): jingle session data - @param content_name(unicode): name of the content - @return (tuple[domish.Element, domish.Element]): parent element, or element, according to action + @param session: jingle session data + @param content_name: name of the content + @param iq_elt: use this IQ instead of creating a new one if provided + @param context_elt: use this element instead of creating a new one if provided + @return: parent element, or element, according to action """ # we first build iq, jingle and content element which are the same in every cases - iq_elt, jingle_elt = self._build_jingle_elt(client, session, action) + if iq_elt is not None: + try: + jingle_elt = next(iq_elt.elements(NS_JINGLE, "jingle")) + except StopIteration: + raise exceptions.InternalError( + "The element provided doesn't have a element" + ) + else: + iq_elt, jingle_elt = self._build_jingle_elt(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: + if context_elt is not None: + pass + elif 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("unmanaged action {}".format(action)) + raise exceptions.InternalError(f"unmanaged action {action}") return iq_elt, context_elt @@ -350,7 +421,7 @@ class XEP_0166: """ return self._build_jingle_elt(client, session, XEP_0166.A_SESSION_INFO) - def getApplication(self, namespace: str) -> object: + def get_application(self, namespace: str) -> ApplicationData: """Retreive application corresponding to a namespace @raise exceptions.NotFound if application can't be found @@ -362,39 +433,53 @@ class XEP_0166: f"No application registered for {namespace}" ) - def get_content_data(self, content: dict) -> Tuple[object, list, dict, str]: + def get_content_data(self, content: dict) -> ContentData: """"Retrieve application and its argument from content""" app_ns = content["app_ns"] try: - application = self.getApplication(app_ns) + application = self.get_application(app_ns) except exceptions.NotFound as e: raise exceptions.InternalError(str(e)) app_args = content.get("app_args", []) app_kwargs = content.get("app_kwargs", {}) + transport_data = content.get("transport_data", {}) try: content_name = content["name"] except KeyError: content_name = content["name"] = str(uuid.uuid4()) - return application, app_args, app_kwargs, content_name + return ContentData( + application, + app_args, + app_kwargs, + transport_data, + content_name + ) - async def initiate(self, client, peer_jid, contents, encrypted=False): + async def initiate( + self, + client: SatXMPPEntity, + peer_jid: jid.JID, + contents: List[dict], + encrypted: bool = False, + **extra_data: Any + ) -> str: """Send a session initiation request - @param peer_jid(jid.JID): jid to establith session with - @param contents(list[dict]): list of contents to use: + @param peer_jid: jid to establith session with + @param contents: list of contents to use: The dict must have the following keys: - - app_ns(unicode): namespace of the application + - app_ns(str): namespace of the application the following keys are optional: - - transport_type(unicode): type of transport to use (see XEP-0166 §8) + - transport_type(str): 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 + - name(str): name of the content + - senders(str): 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 encrypted: if True, session must be encrypted and "encryption" must be set to all content data of session - @return (unicode): jingle session id + @return: jingle session id """ assert contents # there must be at least one content if (peer_jid == client.jid @@ -412,24 +497,25 @@ class XEP_0166: "peer_jid": peer_jid, "started": time.time(), "contents": {}, + **extra_data, } if not await self.host.trigger.async_point( "XEP-0166_initiate", client, session, contents ): - return + return sid iq_elt, jingle_elt = self._build_jingle_elt( client, session, XEP_0166.A_SESSION_INITIATE ) jingle_elt["initiator"] = initiator.full() - contents_dict = session["contents"] + session_contents = session["contents"] for content in contents: # we get the application plugin - application, app_args, app_kwargs, content_name = self.get_content_data(content) + content_data = self.get_content_data(content) # and the transport plugin transport_type = content.get("transport_type", XEP_0166.TRANSPORT_STREAMING) @@ -440,25 +526,25 @@ class XEP_0166: "No transport registered for {}".format(transport_type) ) - # we build the session data - content_data = { - "application": application, + # we build the session data for this content + session_content = { + "application": content_data.application, "application_data": {}, "transport": transport, - "transport_data": {}, + "transport_data": content_data.transport_data, "creator": XEP_0166.ROLE_INITIATOR, "senders": content.get("senders", "both"), } - if content_name in contents_dict: + if content_data.content_name in session_contents: raise exceptions.InternalError( "There is already a content with this name" ) - contents_dict[content_name] = content_data + session_contents[content_data.content_name] = session_content # we construct the content element content_elt = jingle_elt.addElement("content") - content_elt["creator"] = content_data["creator"] - content_elt["name"] = content_name + content_elt["creator"] = session_content["creator"] + content_elt["name"] = content_data.content_name try: content_elt["senders"] = content["senders"] except KeyError: @@ -466,15 +552,16 @@ class XEP_0166: # then the description element desc_elt = await utils.as_deferred( - application.handler.jingle_session_init, - client, session, content_name, *app_args, **app_kwargs + content_data.application.handler.jingle_session_init, + client, session, content_data.content_name, + *content_data.app_args, **content_data.app_kwargs ) content_elt.addChild(desc_elt) # and the transport one transport_elt = await utils.as_deferred( transport.handler.jingle_session_init, - client, session, content_name + client, session, content_data.content_name, ) content_elt.addChild(transport_elt) @@ -482,7 +569,7 @@ class XEP_0166: "XEP-0166_initiate_elt_built", client, session, iq_elt, jingle_elt ): - return + return sid if encrypted: for content in session["contents"].values(): if "encryption" not in content: @@ -496,6 +583,7 @@ class XEP_0166: failure_ = failure.Failure(e) self._iq_error(failure_, sid, client) raise failure_ + return sid def delayed_content_terminate(self, *args, **kwargs): """Put content_terminate in queue but don't execute immediately @@ -624,48 +712,66 @@ class XEP_0166: elif action == XEP_0166.A_SESSION_TERMINATE: self.on_session_terminate(client, request, jingle_elt, session) elif action == XEP_0166.A_SESSION_ACCEPT: - self.on_session_accept(client, request, jingle_elt, session) + await self.on_session_accept(client, request, jingle_elt, session) elif action == XEP_0166.A_SESSION_INFO: self.on_session_info(client, request, jingle_elt, session) elif action == XEP_0166.A_TRANSPORT_INFO: self.on_transport_info(client, request, jingle_elt, session) elif action == XEP_0166.A_TRANSPORT_REPLACE: - self.on_transport_replace(client, request, jingle_elt, session) + await self.on_transport_replace(client, request, jingle_elt, session) elif action == XEP_0166.A_TRANSPORT_ACCEPT: self.on_transport_accept(client, request, jingle_elt, session) elif action == XEP_0166.A_TRANSPORT_REJECT: self.on_transport_reject(client, request, jingle_elt, session) else: - raise exceptions.InternalError("Unknown action {}".format(action)) + raise exceptions.InternalError(f"Unknown action {action}") ## Actions callbacks ## def _parse_elements( self, - jingle_elt, - session, - request, - client, - new=False, - creator=ROLE_INITIATOR, - with_application=True, - with_transport=True, - ): + jingle_elt: domish.Element, + session: dict, + request: domish.Element, + client: SatXMPPEntity, + new: bool = False, + creator: str = ROLE_INITIATOR, + with_application: bool =True, + with_transport: bool = True, + store_in_session: bool = True, + ) -> Dict[str, dict]: """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 element, containing one or more - @param session(dict): session data - @param request(domish.Element): the whole request + @param jingle_elt: parent element, containing one or more + @param session: session data + @param request: the whole request @param client: %(doc_client)s - @param new(bool): True if the content is new and must be created, + @param new: True if 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 element else ignore it - @param with_transport(bool): if True, raise an error if there is no element else ignore it - @raise exceptions.CancelError: the error is treated and the calling method can cancel the treatment (i.e. return) + @param creator: only used if new is True: creating pear (see § 7.3) + @param with_application: if True, raise an error if there is no + element else ignore it + @param with_transport: if True, raise an error if there is no element + else ignore it + @param store_in_session: if True, the ``session`` contents will be updated with + the parsed elements. + Use False when you parse an action which can happen at any time (e.g. + transport-info) and meaning that a parsed element may already be present in + the session (e.g. if an authorisation request is waiting for user answer), + This can't be used when ``new`` is set. + @return: contents_dict (from session, or a new one if "store_in_session" is False) + @raise exceptions.CancelError: the error is treated and the calling method can + cancel the treatment (i.e. return) """ - contents_dict = session["contents"] + if store_in_session: + contents_dict = session["contents"] + else: + if new: + raise exceptions.InternalError( + '"store_in_session" must not be used when "new" is set' + ) + contents_dict = {n: {} for n in session["contents"]} content_elts = jingle_elt.elements(NS_JINGLE, "content") for content_elt in content_elts: @@ -758,6 +864,8 @@ class XEP_0166: content_data["transport_elt"] = transport_elt + return contents_dict + def _ignore(self, client, action, session, content_name, elt): """Dummy method used when not exception must be raised if a method is not implemented in _call_plugins @@ -765,30 +873,39 @@ class XEP_0166: """ return elt - def _call_plugins(self, client, action, session, app_method_name="jingle_handler", - transp_method_name="jingle_handler", app_default_cb=None, - transp_default_cb=None, delete=True, elements=True, - force_element=None): + def _call_plugins( + self, + client: SatXMPPEntity, + action: str, + session: dict, + app_method_name: Optional[str] = "jingle_handler", + transp_method_name: Optional[str] = "jingle_handler", + app_default_cb: Optional[Callable] = None, + transp_default_cb: Optional[Callable] = None, + delete: bool = True, + elements: bool = True, + force_element: Optional[domish.Element] = None + ) -> List[defer.Deferred]: """Call application and transport plugin methods for all contents - @param action(unicode): jingle action name - @param session(dict): jingle session data - @param app_method_name(unicode, None): name of the method to call for applications + @param action: jingle action name + @param session: jingle session data + @param app_method_name: name of the method to call for applications None to ignore - @param transp_method_name(unicode, None): name of the method to call for transports + @param transp_method_name: name of the method to call for transports None to ignore - @param app_default_cb(callable, None): default callback to use if plugin has not app_method_name + @param app_default_cb: default callback to use if plugin has not app_method_name None to raise an exception instead - @param transp_default_cb(callable, None): default callback to use if plugin has not transp_method_name + @param transp_default_cb: default callback to use if plugin has not transp_method_name None to raise an exception instead - @param delete(bool): if True, remove desc_elt and transport_elt from session + @param delete: if True, remove desc_elt and transport_elt from session ignored if elements is False - @param elements(bool): True if elements(desc_elt and tranport_elt) must be managed - must be True if _call_plugins is used in a request, and False if it used after a request - (i.e. on result or error) - @param force_element(None, domish.Element, object): if elements is False, it is used as element parameter + @param elements: True if elements(desc_elt and tranport_elt) must be managed + must be True if _call_plugins is used in a request, and False if it is used + after a request (i.e. on result or error) + @param force_element: if elements is False, it is used as element parameter else it is ignored - @return (list[defer.Deferred]): list of launched Deferred + @return : list of launched Deferred @raise exceptions.NotFound: method is not implemented """ contents_dict = session["contents"] @@ -868,6 +985,13 @@ class XEP_0166: ): return + await defer.DeferredList(self._call_plugins( + client, + XEP_0166.A_PREPARE_CONFIRMATION, + session, + delete=False + )) + # we now request each application plugin confirmation # and if all are accepted, we can accept the session confirm_defers = self._call_plugins( @@ -991,7 +1115,7 @@ class XEP_0166: terminate_dlist.addCallback(lambda __: self._del_session(client, session["id"])) client.send(xmlstream.toResponse(request, "result")) - def on_session_accept(self, client, request, jingle_elt, session): + async def on_session_accept(self, client, request, jingle_elt, session): """Method called once session is accepted This method is only called for initiator @@ -1000,7 +1124,7 @@ class XEP_0166: @param jingle_elt(domish.Element): the element @param session(dict): session data """ - log.debug("Jingle session {} has been accepted".format(session["id"])) + log.debug(f"Jingle session {session['id']} has been accepted") try: self._parse_elements(jingle_elt, session, request, client) @@ -1012,6 +1136,13 @@ class XEP_0166: # and change the state session["state"] = STATE_ACTIVE + await defer.DeferredList(self._call_plugins( + client, + XEP_0166.A_PREPARE_INITIATOR, + session, + delete=False + )) + negociate_defers = [] negociate_defers = self._call_plugins(client, XEP_0166.A_SESSION_ACCEPT, session) @@ -1068,8 +1199,7 @@ class XEP_0166: dlist.addCallback(self._on_session_cb, client, request, jingle_elt, session) dlist.addErrback(self._on_session_cb, client, request, jingle_elt, session) - @defer.inlineCallbacks - def on_transport_replace(self, client, request, jingle_elt, session): + async def on_transport_replace(self, client, request, jingle_elt, session): """A transport change is requested The request is parsed, and jingle_handler is called on concerned transport plugin(s) @@ -1127,7 +1257,8 @@ class XEP_0166: ) for content_name, content_data, transport, transport_elt in to_replace: # we can now actually replace the transport - yield content_data["transport"].handler.jingle_handler( + await utils.as_deferred( + content_data["transport"].handler.jingle_handler, client, XEP_0166.A_DESTROY, session, content_name, None ) content_data["transport"] = transport @@ -1137,12 +1268,14 @@ class XEP_0166: content_elt["name"] = content_name content_elt["creator"] = content_data["creator"] # we notify the transport and insert its in the answer - accept_transport_elt = yield transport.handler.jingle_handler( + accept_transport_elt = await utils.as_deferred( + transport.handler.jingle_handler, client, XEP_0166.A_TRANSPORT_REPLACE, session, content_name, transport_elt ) content_elt.addChild(accept_transport_elt) # there is no confirmation needed here, so we can directly prepare it - yield transport.handler.jingle_handler( + await utils.as_deferred( + transport.handler.jingle_handler, client, XEP_0166.A_PREPARE_RESPONDER, session, content_name, None ) @@ -1194,20 +1327,28 @@ class XEP_0166: # this behaviour may change in the future self.terminate(client, "failed-transport", session) - def on_transport_info(self, client, request, jingle_elt, session): + def on_transport_info( + self, + client: SatXMPPEntity, + request: domish.Element, + jingle_elt: domish.Element, + session: dict + ) -> None: """Method called when a transport-info action is received from other peer - The request is parsed, and jingle_handler is called on concerned transport plugin(s) + The request is parsed, and jingle_handler is called on concerned transport + plugin(s) @param client: %(doc_client)s - @param request(domish.Element): full request - @param jingle_elt(domish.Element): the element - @param session(dict): session data + @param request: full request + @param jingle_elt: the element + @param session: session data """ - log.debug("Jingle session {} has been accepted".format(session["id"])) + log.debug(f"Jingle session {session['id']} has been accepted") try: - self._parse_elements( - jingle_elt, session, request, client, with_application=False + parsed_contents = self._parse_elements( + jingle_elt, session, request, client, with_application=False, + store_in_session=False ) except exceptions.CancelError: return @@ -1217,11 +1358,12 @@ class XEP_0166: for content_name, content_data in session["contents"].items(): try: - transport_elt = content_data.pop("transport_elt") + transport_elt = parsed_contents[content_name]["transport_elt"] except KeyError: continue else: - content_data["transport"].handler.jingle_handler( + utils.as_deferred( + content_data["transport"].handler.jingle_handler, client, XEP_0166.A_TRANSPORT_INFO, session, diff --git a/sat/plugins/plugin_xep_0166/models.py b/sat/plugins/plugin_xep_0166/models.py new file mode 100644 index 00000000..8be3fb16 --- /dev/null +++ b/sat/plugins/plugin_xep_0166/models.py @@ -0,0 +1,183 @@ +#!/usr/bin/env python3 + +# Libervia plugin for Jingle (XEP-0166) +# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + + +import abc +from dataclasses import dataclass +from typing import Awaitable, Callable, Union + +from twisted.internet import defer +from twisted.words.xish import domish + +from sat.core.core_types import SatXMPPEntity +from sat.core.i18n import _ + + +class BaseApplicationHandler(abc.ABC): + + @abc.abstractmethod + def jingle_request_confirmation( + self, + client: SatXMPPEntity, + action: str, + session: dict, + content_name: str, + desc_elt: domish.Element, + ) -> Union[ + Callable[..., Union[bool, defer.Deferred]], + Callable[..., Awaitable[bool]] + ]: + """ + If present, it is called on when session must be accepted. + If not present, a generic accept dialog will be used. + + @param session: Jingle Session + @param desc_elt: element + @return: True if the session is accepted. + A Deferred can be returned. + """ + pass + + @abc.abstractmethod + def jingle_session_init( + self, + client: SatXMPPEntity, + session: dict, + content_name: str, + *args, **kwargs + ) -> Union[ + Callable[..., domish.Element], + Callable[..., Awaitable[domish.Element]] + ]: + """ + Must return the domish.Element used for initial content. + + @param client: SatXMPPEntity instance + @param session: Jingle Session + @param content_name: Name of the content + @return: The domish.Element used for initial content + """ + pass + + @abc.abstractmethod + def jingle_handler( + self, + client: SatXMPPEntity, + action: str, + session: dict, + content_name: str, + transport_elt: domish.Element + ) -> Union[ + Callable[..., None], + Callable[..., Awaitable[None]] + ]: + """ + Called on several actions to negotiate the application or transport. + + @param client: SatXMPPEntity instance + @param action: Jingle action + @param session: Jingle Session + @param content_name: Name of the content + @param transport_elt: Transport element + """ + pass + + @abc.abstractmethod + def jingle_terminate( + self, + reason_elt: domish.Element + ) -> Union[ + Callable[..., None], + Callable[..., Awaitable[None]] + ]: + """ + Called on session terminate, with reason_elt. + May be used to clean session. + + @param reason_elt: Reason element + """ + pass + + +class BaseTransportHandler(abc.ABC): + + @abc.abstractmethod + def jingle_session_init( + self, + client: SatXMPPEntity, + session: dict, + content_name: str, + *args, **kwargs + ) -> Union[ + Callable[..., domish.Element], + Callable[..., Awaitable[domish.Element]] + ]: + """ + Must return the domish.Element used for initial content. + + @param client: SatXMPPEntity instance + @param session: Jingle Session + @param content_name: Name of the content + @return: The domish.Element used for initial content + """ + pass + + @abc.abstractmethod + def jingle_handler( + self, + client: SatXMPPEntity, + action: str, + session: dict, + content_name: str, + transport_elt: domish.Element + ) -> Union[ + Callable[..., None], + Callable[..., Awaitable[None]] + ]: + """ + Called on several actions to negotiate the application or transport. + + @param client: SatXMPPEntity instance + @param action: Jingle action + @param session: Jingle Session + @param content_name: Name of the content + @param transport_elt: Transport element + """ + pass + + +@dataclass(frozen=True) +class ApplicationData: + namespace: str + handler: BaseApplicationHandler + + +@dataclass(frozen=True) +class TransportData: + namespace: str + handler: BaseTransportHandler + priority: int + + +@dataclass(frozen=True) +class ContentData: + application: ApplicationData + app_args: list + app_kwargs: dict + transport_data: dict + content_name: str diff --git a/sat/plugins/plugin_xep_0215.py b/sat/plugins/plugin_xep_0215.py index c3295cd4..64b1fc41 100644 --- a/sat/plugins/plugin_xep_0215.py +++ b/sat/plugins/plugin_xep_0215.py @@ -167,7 +167,7 @@ class XEP_0215: Response is cached after first query - @param entity: XMPP entity to query. Defaut to our own server + @param entity: XMPP entity to query. Default to our own server @return: found services """ if entity is None: diff --git a/sat/plugins/plugin_xep_0260.py b/sat/plugins/plugin_xep_0260.py index 43b1eb35..a2e8f350 100644 --- a/sat/plugins/plugin_xep_0260.py +++ b/sat/plugins/plugin_xep_0260.py @@ -161,9 +161,11 @@ class XEP_0260(object): cf XEP-0260 § 2.4 """ # now that the proxy is activated, we have to inform other peer + content_data = session["contents"][content_name] iq_elt, transport_elt = self._j.build_action( client, self._j.A_TRANSPORT_INFO, session, content_name ) + transport_elt["sid"] = content_data["transport_data"]["sid"] activated_elt = transport_elt.addElement("activated") activated_elt["cid"] = candidate.id iq_elt.send() @@ -175,9 +177,11 @@ class XEP_0260(object): """ # TODO: fallback to IBB # now that the proxy is activated, we have to inform other peer + content_data = session["contents"][content_name] iq_elt, transport_elt = self._j.build_action( client, self._j.A_TRANSPORT_INFO, session, content_name ) + transport_elt["sid"] = content_data["transport_data"]["sid"] transport_elt.addElement("proxy-error") iq_elt.send() log.warning( @@ -200,6 +204,7 @@ class XEP_0260(object): @param client(unicode): %(doc_client)s """ + content_data = session["contents"][content_name] transport_data["best_candidate"] = candidate # we need to disconnect all non selected candidates before removing them for c in transport_data["peer_candidates"]: @@ -210,6 +215,7 @@ class XEP_0260(object): iq_elt, transport_elt = self._j.build_action( client, self._j.A_TRANSPORT_INFO, session, content_name ) + transport_elt["sid"] = content_data["transport_data"]["sid"] if candidate is None: log.warning("Can't connect to any peer candidate") candidate_elt = transport_elt.addElement("candidate-error") diff --git a/sat/plugins/plugin_xep_0353.py b/sat/plugins/plugin_xep_0353.py index 29405295..fde4629e 100644 --- a/sat/plugins/plugin_xep_0353.py +++ b/sat/plugins/plugin_xep_0353.py @@ -105,22 +105,30 @@ class XEP_0353: mess_data = self.build_message_data(client, peer_jid, "propose", session['id']) for content in contents: - application, app_args, app_kwargs, content_name = self._j.get_content_data( + content_data = self._j.get_content_data( content) try: - jingle_description_elt = application.handler.jingle_description_elt + jingle_description_elt = ( + content_data.application.handler.jingle_description_elt + ) except AttributeError: - log.debug(f"no jingle_description_elt set for {application.handler}") + log.debug( + "no jingle_description_elt set for " + f"{content_data.application.handler}" + ) description_elt = domish.Element((content["app_ns"], "description")) else: description_elt = await utils.as_deferred( jingle_description_elt, - client, session, content_name, *app_args, **app_kwargs + client, session, content_data.content_name, *content_data.app_args, + **content_data.app_kwargs ) mess_data["xml"].propose.addChild(description_elt) response_d = defer.Deferred() # we wait for 2 min before cancelling the session init - response_d.addTimeout(2*60, reactor) + # response_d.addTimeout(2*60, reactor) + # FIXME: let's application decide timeout? + response_d.addTimeout(2, reactor) client._xep_0353_pending_sessions[session['id']] = response_d await client.send_message_data(mess_data) try: @@ -158,7 +166,7 @@ class XEP_0353: if peer_jid.userhostJID() not in client.roster: app_ns = elt.description.uri try: - application = self._j.getApplication(app_ns) + application = self._j.get_application(app_ns) human_name = getattr(application.handler, "human_name", application.name) except (exceptions.NotFound, AttributeError): if app_ns.startswith("urn:xmpp:jingle:apps:"): @@ -193,6 +201,7 @@ class XEP_0353: else: await client.presence.available(peer_jid) session_id = elt["id"] + # FIXME: accept is not used anymore in new specification, check it and remove it mess_data = self.build_message_data( client, client.jid.userhostJID(), "accept", session_id) await client.send_message_data(mess_data)