Skip to content
Back to top

# Source code for slixmpp.plugins.xep_0257.client_cert_management

    # Slixmpp: The Slick XMPP Library
    # Copyright (C) 2012 Nathanael C. Fritz, Lance J.T. Stout
    # This file is part of Slixmpp.
    # See the file LICENSE for copying permission.
    import logging

    from typing import Awaitable

    from slixmpp import Iq, JID
    from slixmpp.plugins import BasePlugin
    from slixmpp.plugins.xep_0257 import stanza

    log = logging.getLogger(__name__)

    class XEP_0257(BasePlugin):

        name = 'xep_0257'
        description = 'XEP-0257: Client Certificate Management for SASL EXTERNAL'
        dependencies = {'xep_0030'}
        stanza = stanza

        def plugin_init(self) -> None:
            stanza.register_plugins()

        async def get_certs(self, ifrom: JID | None = None,
                            timeout: float | None = None) -> set[tuple[str, str, tuple[str, ...]]]:
            """
            Return the list of cert items.

            :param ifrom: JID to send the stanza from (for components).
            :param timeout: Timeout of the query (in seconds)
            :returns: A list of certificate items as tuple of (name, cert, active users).
            :raises IqTimeout: When the query timeouts
            :raises IqError: If the server encounters an error while processing
                             the query.
            """
            iq = self.xmpp.Iq()
            iq['type'] = 'get'
            iq['from'] = ifrom
            iq.enable('sasl_certs')
            certs = await iq.send(timeout=timeout)
            return {
                (cert['name'], cert['x509cert'], tuple(cert['users']))
                for cert in certs['sasl_certs']['items']
            }

        def add_cert(self, name: str, cert: str,
                     allow_management: bool = True,
                     ifrom: JID | None = None,
                     timeout: float | None = None) -> Awaitable[Iq]:
            """
            Register a cert with the server.

            :param name: Name of the cert.
            :param cert: Base64 of the cert’s DER.
            :param allow_management: Allow management of this cert by a client
                                     logged in with it.
            :param ifrom: JID to send the stanza from (for components).
            :param timeout: Timeeout of the query (in seconds).
            """
            iq = self.xmpp.Iq()
            iq['type'] = 'set'
            iq['from'] = ifrom
            iq['sasl_cert_append']['name'] = name
            iq['sasl_cert_append']['x509cert'] = cert
            iq['sasl_cert_append']['cert_management'] = allow_management
            return iq.send(timeout=timeout)

        def disable_cert(self, name: str, ifrom: JID | None = None,
                         timeout: float | None = None) -> Awaitable[Iq]:
            """
            Disable a cert. Clients using this cert are not immediately disconnected.

            :param name: Name of the cert.
            :param ifrom: JID to send the stanza from (for components).
            :param timeout: Timeeout of the query (in seconds).
            """
            iq = self.xmpp.Iq()
            iq['type'] = 'set'
            iq['from'] = ifrom
            iq['sasl_cert_disable']['name'] = name
            return iq.send(timeout=timeout)

        def revoke_cert(self, name: str, ifrom: JID | None = None,
                        timeout: float | None = None) -> Awaitable[Iq]:
            """
            Revoke a cert. Clients using this cert are immediately disconnected.

            :param name: Name of the cert.
            :param ifrom: JID to send the stanza from (for components).
            :param timeout: Timeeout of the query (in seconds).
            """
            iq = self.xmpp.Iq()
            iq['type'] = 'set'
            iq['from'] = ifrom
            iq['sasl_cert_revoke']['name'] = name
            return iq.send(timeout=timeout)