Mercurial > libervia-backend
view libervia/backend/tools/common/tls.py @ 4240:79c8a70e1813
backend, frontend: prepare remote control:
This is a series of changes necessary to prepare the implementation of remote control
feature:
- XEP-0166: add a `priority` attribute to `ApplicationData`: this is needed when several
applications are working in a same session, to know which one must be handled first.
Will be used to make Remote Control have precedence over Call content.
- XEP-0166: `_call_plugins` is now async and is not used with `DeferredList` anymore: the
benefit to have methods called in parallels is very low, and it cause a lot of trouble
as we can't predict order. Methods are now called sequentially so workflow can be
predicted.
- XEP-0167: fix `senders` XMPP attribute <=> SDP mapping
- XEP-0234: preflight acceptance key is now `pre-accepted` instead of `file-accepted`, so
the same key can be used with other jingle applications.
- XEP-0167, XEP-0343: move some method to XEP-0167
- XEP-0353: use new `priority` feature to call preflight methods of applications according
to it.
- frontend (webrtc): refactor the sources/sink handling with a more flexible mechanism
based on Pydantic models. It is now possible to have has many Data Channel as necessary,
to have them in addition to A/V streams, to specify manually GStreamer sources and
sinks, etc.
- frontend (webrtc): rework of the pipeline to reduce latency.
- frontend: new `portal_desktop` method. Screenshare portal handling has been moved there,
and RemoteDesktop portal has been added.
- frontend (webrtc): fix `extract_ufrag_pwd` method.
rel 436
author | Goffi <goffi@goffi.org> |
---|---|
date | Sat, 11 May 2024 13:52:41 +0200 |
parents | 4b842c1fb686 |
children | 0d7bb4df2343 |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia: an XMPP client # 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 <http://www.gnu.org/licenses/>. """TLS handling with twisted""" from libervia.backend.core.log import getLogger from libervia.backend.core import exceptions from libervia.backend.tools import config as tools_config try: import OpenSSL from twisted.internet import ssl except ImportError: ssl = None log = getLogger(__name__) def get_options_from_config(config, section=""): options = {} for option in ('tls_certificate', 'tls_private_key', 'tls_chain'): options[option] = tools_config.config_get(config, section, option) return options def tls_options_check(options): """Check options coherence if TLS is activated, and update missing values Must be called only if TLS is activated """ if not options["tls_certificate"]: raise exceptions.ConfigError( "a TLS certificate is needed to activate HTTPS connection") if not options["tls_private_key"]: options["tls_private_key"] = options["tls_certificate"] def load_certificates(f): """Read a .pem file with a list of certificates @param f (file): file obj (opened .pem file) @return (list[OpenSSL.crypto.X509]): list of certificates @raise OpenSSL.crypto.Error: error while parsing the file """ # XXX: didn't found any method to load a .pem file with several certificates # so the certificates split is done here certificates = [] buf = [] while True: line = f.readline() buf.append(line) if "-----END CERTIFICATE-----" in line: certificates.append( OpenSSL.crypto.load_certificate( OpenSSL.crypto.FILETYPE_PEM, "".join(buf) ) ) buf = [] elif not line: log.debug(f"{len(certificates)} certificate(s) found") return certificates def load_p_key(f): """Read a private key from a .pem file @param f (file): file obj (opened .pem file) @return (list[OpenSSL.crypto.PKey]): private key object @raise OpenSSL.crypto.Error: error while parsing the file """ return OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, f.read()) def load_certificate(f): """Read a public certificate from a .pem file @param f (file): file obj (opened .pem file) @return (list[OpenSSL.crypto.X509]): public certificate @raise OpenSSL.crypto.Error: error while parsing the file """ return OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, f.read()) def get_tls_context_factory(options): """Load TLS certificate and build the context factory needed for listenSSL""" if ssl is None: raise ImportError("Python module pyOpenSSL is not installed!") cert_options = {} for name, option, method in [ ("privateKey", "tls_private_key", load_p_key), ("certificate", "tls_certificate", load_certificate), ("extraCertChain", "tls_chain", load_certificates), ]: path = options[option] if not path: assert option == "tls_chain" continue log.debug(f"loading {option} from {path}") try: with open(path) as f: cert_options[name] = method(f) except IOError as e: raise exceptions.DataError( f"Error while reading file {path} for option {option}: {e}" ) except OpenSSL.crypto.Error: raise exceptions.DataError( f"Error while parsing file {path} for option {option}, are you sure " f"it is a valid .pem file?" ) if ( option == "tls_private_key" and options["tls_certificate"] == path ): raise exceptions.ConfigError( f"You are using the same file for private key and public " f"certificate, make sure that both a in {path} or use " f"--tls_private_key option" ) return ssl.CertificateOptions(**cert_options)