Mercurial > libervia-backend
changeset 4048:04e21d6c9188
plugin XEP-0320: Use of DTLS-SRTP in Jingle Sessions:
rel 421
author | Goffi <goffi@goffi.org> |
---|---|
date | Mon, 15 May 2023 17:02:40 +0200 (19 months ago) |
parents | f0b1279a53c3 |
children | b56bf0c6c064 |
files | sat/plugins/plugin_xep_0320.py |
diffstat | 1 files changed, 106 insertions(+), 0 deletions(-) [+] |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/sat/plugins/plugin_xep_0320.py Mon May 15 17:02:40 2023 +0200 @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 + + +# Libervia plugin +# Copyright (C) 2009-2023 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 <http://www.gnu.org/licenses/>. + +from twisted.words.protocols.jabber.xmlstream import XMPPHandler +from twisted.words.xish import domish +from wokkel import disco, iwokkel +from zope.interface import implementer + +from sat.core.constants import Const as C +from sat.core.i18n import _ +from sat.core.log import getLogger + + +log = getLogger(__name__) + +NS_JINGLE_DTLS = "urn:xmpp:jingle:apps:dtls:0" + +PLUGIN_INFO = { + C.PI_NAME: "Use of DTLS-SRTP in Jingle Sessions", + C.PI_IMPORT_NAME: "XEP-0320", + C.PI_TYPE: "XEP", + C.PI_MODES: C.PLUG_MODE_BOTH, + C.PI_PROTOCOLS: ["XEP-0320"], + C.PI_DEPENDENCIES: ["XEP-0176"], + C.PI_RECOMMENDATIONS: [], + C.PI_MAIN: "XEP_0320", + C.PI_HANDLER: "yes", + C.PI_DESCRIPTION: _("""Use of DTLS-SRTP with RTP (for e2ee of A/V calls)"""), +} + + +class XEP_0320: + def __init__(self, host): + log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") + host.trigger.add("XEP-0176_parse_transport", self._parse_transport_trigger) + host.trigger.add("XEP-0176_build_transport", self._build_transport_trigger) + + def get_handler(self, client): + return XEP_0320_handler() + + def _parse_transport_trigger( + self, transport_elt: domish.Element, ice_data: dict + ) -> bool: + """Parse the <fingerprint> element""" + fingerprint_elt = next( + transport_elt.elements(NS_JINGLE_DTLS, "fingerprint"), None + ) + if fingerprint_elt is not None: + try: + ice_data["fingerprint"] = { + "hash": fingerprint_elt["hash"], + "setup": fingerprint_elt["setup"], + "fingerprint": str(fingerprint_elt), + } + except KeyError as e: + log.warning( + f"invalid <fingerprint> (attribue {e} is missing): " + f"{fingerprint_elt.toXml()})" + ) + + return True + + def _build_transport_trigger( + self, tranport_elt: domish.Element, ice_data: dict + ) -> bool: + """Build the <fingerprint> element if possible""" + try: + fingerprint_data = ice_data["fingerprint"] + hash_ = fingerprint_data["hash"] + setup = fingerprint_data["setup"] + fingerprint = fingerprint_data["fingerprint"] + except KeyError: + pass + else: + fingerprint_elt = tranport_elt.addElement( + (NS_JINGLE_DTLS, "fingerprint"), content=fingerprint + ) + fingerprint_elt["hash"] = hash_ + fingerprint_elt["setup"] = setup + + return True + + +@implementer(iwokkel.IDisco) +class XEP_0320_handler(XMPPHandler): + def getDiscoInfo(self, requestor, target, nodeIdentifier=""): + return [disco.DiscoFeature(NS_JINGLE_DTLS)] + + def getDiscoItems(self, requestor, target, nodeIdentifier=""): + return []