libervia-backend/sat/plugins/plugin_xep_0373.py

#!/usr/bin/env python3

# Libervia plugin for OpenPGP for XMPP
# Copyright (C) 2022-2022 Tim Henkes (me@syndace.dev)

# 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 abc import ABC, abstractmethod
import base64
from datetime import datetime, timezone
import enum
import secrets
import string
from typing import Any, Dict, Iterable, List, Literal, Optional, Set, Tuple, cast
from xml.sax.saxutils import quoteattr

from typing_extensions import Final, NamedTuple, Never, assert_never
from wokkel import muc, pubsub
from wokkel.disco import DiscoFeature, DiscoInfo
import xmlschema

from sat.core import exceptions
from sat.core.constants import Const as C
from sat.core.core_types import SatXMPPEntity
from sat.core.i18n import _, D_
from sat.core.log import getLogger, Logger
from sat.core.sat_main import SAT
from sat.core.xmpp import SatXMPPClient
from sat.memory import persistent
from sat.plugins.plugin_xep_0045 import XEP_0045
from sat.plugins.plugin_xep_0060 import XEP_0060
from sat.plugins.plugin_xep_0163 import XEP_0163
from sat.tools.xmpp_datetime import format_datetime, parse_datetime
from sat.tools import xml_tools
from twisted.internet import defer
from twisted.words.protocols.jabber import jid
from twisted.words.xish import domish

try:
    import gpg
except ImportError as import_error:
    raise exceptions.MissingModule(
        "You are missing the 'gpg' package required by the OX plugin. The recommended"
        " installation method is via your operating system's package manager, since the"
        " version of the library has to match the version of your GnuPG installation. See"
        " https://wiki.python.org/moin/GnuPrivacyGuard#Accessing_GnuPG_via_gpgme"
    ) from import_error


__all__ = [  # pylint: disable=unused-variable
    "PLUGIN_INFO",
    "NS_OX",
    "XEP_0373",
    "VerificationError",
    "XMPPInteractionFailed",
    "InvalidPacket",
    "DecryptionFailed",
    "VerificationFailed",
    "UnknownKey",
    "GPGProviderError",
    "GPGPublicKey",
    "GPGSecretKey",
    "GPGProvider",
    "PublicKeyMetadata",
    "gpg_provider",
    "TrustLevel"
]


log = cast(Logger, getLogger(__name__))  # type: ignore[no-untyped-call]


PLUGIN_INFO = {
    C.PI_NAME: "XEP-0373",
    C.PI_IMPORT_NAME: "XEP-0373",
    C.PI_TYPE: "SEC",
    C.PI_PROTOCOLS: [ "XEP-0373" ],
    C.PI_DEPENDENCIES: [ "XEP-0060", "XEP-0163" ],
    C.PI_RECOMMENDATIONS: [],
    C.PI_MAIN: "XEP_0373",
    C.PI_HANDLER: "no",
    C.PI_DESCRIPTION: D_("Implementation of OpenPGP for XMPP"),
}


NS_OX: Final = "urn:xmpp:openpgp:0"


PARAM_CATEGORY = "Security"
PARAM_NAME = "ox_policy"
STR_KEY_PUBLIC_KEYS_METADATA = "/public-keys-metadata/{}"


class VerificationError(Exception):
    """
    Raised by verifying methods of :class:`XEP_0373` on semantical verification errors.
    """


class XMPPInteractionFailed(Exception):
    """
    Raised by methods of :class:`XEP_0373` on XMPP interaction failure. The reason this
    exception exists is that the exceptions raised by XMPP interactions are not properly
    documented for the most part, thus all exceptions are caught and wrapped in instances
    of this class.
    """


class InvalidPacket(ValueError):
    """
    Raised by methods of :class:`GPGProvider` when an invalid packet is encountered.
    """


class DecryptionFailed(Exception):
    """
    Raised by methods of :class:`GPGProvider` on decryption failures.
    """


class VerificationFailed(Exception):
    """
    Raised by methods of :class:`GPGProvider` on verification failures.
    """


class UnknownKey(ValueError):
    """
    Raised by methods of :class:`GPGProvider` when an unknown key is referenced.
    """


class GPGProviderError(Exception):
    """
    Raised by methods of :class:`GPGProvider` on internal errors.
    """


class GPGPublicKey(ABC):
    """
    Interface describing a GPG public key.
    """

    @property
    @abstractmethod
    def fingerprint(self) -> str:
        """
        @return: The OpenPGP v4 fingerprint string of this public key.
        """


class GPGSecretKey(ABC):
    """
    Interface descibing a GPG secret key.
    """

    @property
    @abstractmethod
    def public_key(self) -> GPGPublicKey:
        """
        @return: The public key corresponding to this secret key.
        """


class GPGProvider(ABC):
    """
    Interface describing a GPG provider, i.e. a library or framework providing GPG
    encryption, signing and key management.

    All methods may raise :class:`GPGProviderError` in addition to those exception types
    listed explicitly.

    # TODO: Check keys for revoked, disabled and expired everywhere and exclude those (?)
    """

    @abstractmethod
    def export_public_key(self, public_key: GPGPublicKey) -> bytes:
        """Export a public key in a key material packet according to RFC 4880 §5.5.

        Do not use OpenPGP's ASCII Armor.

        @param public_key: The public key to export.
        @return: The packet containing the exported public key.
        @raise UnknownKey: if the public key is not available.
        """

    @abstractmethod
    def import_public_key(self, packet: bytes) -> GPGPublicKey:
        """Import a public key from a key material packet according to RFC 4880 §5.5.

        OpenPGP's ASCII Armor is not used.

        @param packet: A packet containing an exported public key.
        @return: The public key imported from the packet.
        @raise InvalidPacket: if the packet is either syntactically or semantically deemed
            invalid.

        @warning: Only packets of version 4 or higher may be accepted, packets below
            version 4 MUST be rejected.
        """

    @abstractmethod
    def backup_secret_key(self, secret_key: GPGSecretKey) -> bytes:
        """Export a secret key for transfer according to RFC 4880 §11.1.

        Do not encrypt the secret data, i.e. set the octet indicating string-to-key usage
        conventions to zero in the corresponding secret-key packet according to RFC 4880
        §5.5.3. Do not use OpenPGP's ASCII Armor.

        @param secret_key: The secret key to export.
        @return: The binary blob containing the exported secret key.
        @raise UnknownKey: if the secret key is not available.
        """

    @abstractmethod
    def restore_secret_keys(self, data: bytes) -> Set[GPGSecretKey]:
        """Restore secret keys exported for transfer according to RFC 4880 §11.1.

        The secret data is not encrypted, i.e. the octet indicating string-to-key usage
        conventions in the corresponding secret-key packets according to RFC 4880 §5.5.3
        are set to zero. OpenPGP's ASCII Armor is not used.

        @param data: Concatenation of one or more secret keys exported for transfer.
        @return: The secret keys imported from the data.
        @raise InvalidPacket: if the data or one of the packets included in the data is
            either syntactically or semantically deemed invalid.

        @warning: Only packets of version 4 or higher may be accepted, packets below
            version 4 MUST be rejected.
        """

    @abstractmethod
    def encrypt_symmetrically(self, plaintext: bytes, password: str) -> bytes:
        """Encrypt data symmetrically according to RFC 4880 §5.3.

        The password is used to build a Symmetric-Key Encrypted Session Key packet which
        precedes the Symmetrically Encrypted Data packet that holds the encrypted data.

        @param plaintext: The data to encrypt.
        @param password: The password to encrypt the data with.
        @return: The encrypted data.
        """

    @abstractmethod
    def decrypt_symmetrically(self, ciphertext: bytes, password: str) -> bytes:
        """Decrypt data symmetrically according to RFC 4880 §5.3.

        The ciphertext consists of a Symmetrically Encrypted Data packet that holds the
        encrypted data, preceded by a Symmetric-Key Encrypted Session Key packet using the
        password.

        @param ciphertext: The ciphertext.
        @param password: The password to decrypt the data with.
        @return: The plaintext.
        @raise DecryptionFailed: on decryption failure.
        """

    @abstractmethod
    def sign(self, data: bytes, secret_keys: Set[GPGSecretKey]) -> bytes:
        """Sign some data.

        OpenPGP's ASCII Armor is not used.

        @param data: The data to sign.
        @param secret_keys: The secret keys to sign the data with.
        @return: The OpenPGP message carrying the signed data.
        """

    @abstractmethod
    def sign_detached(self, data: bytes, secret_keys: Set[GPGSecretKey]) -> bytes:
        """Sign some data. Create the signature detached from the data.

        OpenPGP's ASCII Armor is not used.

        @param data: The data to sign.
        @param secret_keys: The secret keys to sign the data with.
        @return: The OpenPGP message carrying the detached signature.
        """

    @abstractmethod
    def verify(self, signed_data: bytes, public_keys: Set[GPGPublicKey]) -> bytes:
        """Verify signed data.

        OpenPGP's ASCII Armor is not used.

        @param signed_data: The signed data as an OpenPGP message.
        @param public_keys: The public keys to verify the signature with.
        @return: The verified and unpacked data.
        @raise VerificationFailed: if the data could not be verified.

        @warning: For implementors: it has to be confirmed that a valid signature by one
            of the public keys is available.
        """

    @abstractmethod
    def verify_detached(
        self,
        data: bytes,
        signature: bytes,
        public_keys: Set[GPGPublicKey]
    ) -> None:
        """Verify signed data, where the signature was created detached from the data.

        OpenPGP's ASCII Armor is not used.

        @param data: The data.
        @param signature: The signature as an OpenPGP message.
        @param public_keys: The public keys to verify the signature with.
        @raise VerificationFailed: if the data could not be verified.

        @warning: For implementors: it has to be confirmed that a valid signature by one
            of the public keys is available.
        """

    @abstractmethod
    def encrypt(
        self,
        plaintext: bytes,
        public_keys: Set[GPGPublicKey],
        signing_keys: Optional[Set[GPGSecretKey]] = None
    ) -> bytes:
        """Encrypt and optionally sign some data.

        OpenPGP's ASCII Armor is not used.

        @param plaintext: The data to encrypt and optionally sign.
        @param public_keys: The public keys to encrypt the data for.
        @param signing_keys: The secret keys to sign the data with.
        @return: The OpenPGP message carrying the encrypted and optionally signed data.
        """

    @abstractmethod
    def decrypt(
        self,
        ciphertext: bytes,
        secret_keys: Set[GPGSecretKey],
        public_keys: Optional[Set[GPGPublicKey]] = None
    ) -> bytes:
        """Decrypt and optionally verify some data.

        OpenPGP's ASCII Armor is not used.

        @param ciphertext: The encrypted and optionally signed data as an OpenPGP message.
        @param secret_keys: The secret keys to attempt decryption with.
        @param public_keys: The public keys to verify the optional signature with.
        @return: The decrypted, optionally verified and unpacked data.
        @raise DecryptionFailed: on decryption failure.
        @raise VerificationFailed: if the data could not be verified.

        @warning: For implementors: it has to be confirmed that the data was decrypted
            using one of the secret keys and that a valid signature by one of the public
            keys is available in case the data is signed.
        """

    @abstractmethod
    def list_public_keys(self, user_id: str) -> Set[GPGPublicKey]:
        """List public keys.

        @param user_id: The user id.
        @return: The set of public keys available for this user id.
        """

    @abstractmethod
    def list_secret_keys(self, user_id: str) -> Set[GPGSecretKey]:
        """List secret keys.

        @param user_id: The user id.
        @return: The set of secret keys available for this user id.
        """

    @abstractmethod
    def can_sign(self, public_key: GPGPublicKey) -> bool:
        """
        @return: Whether the public key belongs to a key pair capable of signing.
        """

    @abstractmethod
    def can_encrypt(self, public_key: GPGPublicKey) -> bool:
        """
        @return: Whether the public key belongs to a key pair capable of encryption.
        """

    @abstractmethod
    def create_key(self, user_id: str) -> GPGSecretKey:
        """Create a new GPG key, capable of signing and encryption.

        The key is generated without password protection and without expiration. If a key
        with the same user id already exists, a new key is created anyway.

        @param user_id: The user id to assign to the new key.
        @return: The new key.
        """


class GPGME_GPGPublicKey(GPGPublicKey):
    """
    GPG public key implementation based on GnuPG Made Easy (GPGME).
    """

    def __init__(self, key_obj: Any) -> None:
        """
        @param key_obj: The GPGME key object.
        """

        self.__key_obj = key_obj

    @property
    def fingerprint(self) -> str:
        return self.__key_obj.fpr

    @property
    def key_obj(self) -> Any:
        return self.__key_obj


class GPGME_GPGSecretKey(GPGSecretKey):
    """
    GPG secret key implementation based on GnuPG Made Easy (GPGME).
    """

    def __init__(self, public_key: GPGME_GPGPublicKey) -> None:
        """
        @param public_key: The public key corresponding to this secret key.
        """

        self.__public_key = public_key

    @property
    def public_key(self) -> GPGME_GPGPublicKey:
        return self.__public_key


class GPGME_GPGProvider(GPGProvider):
    """
    GPG provider implementation based on GnuPG Made Easy (GPGME).
    """

    def __init__(self, home_dir: Optional[str] = None) -> None:
        """
        @param home_dir: Optional GPG home directory path to use for all operations.
        """

        self.__home_dir = home_dir

    def export_public_key(self, public_key: GPGPublicKey) -> bytes:
        assert isinstance(public_key, GPGME_GPGPublicKey)

        pattern = public_key.fingerprint

        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                result = c.key_export_minimal(pattern)
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e

            if result is None:
                raise UnknownKey(f"Public key {pattern} not found.")

            return result

    def import_public_key(self, packet: bytes) -> GPGPublicKey:
        # TODO
        # - Reject packets older than version 4
        # - Check whether it's actually a public key (through packet inspection?)

        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                result = c.key_import(packet)
            except gpg.errors.GPGMEError as e:
                # From looking at the code, `key_import` never raises. The documentation
                # says it does though, so this is included for future-proofness.
                raise GPGProviderError("Internal GPGME error") from e

            if not hasattr(result, "considered"):
                raise InvalidPacket(
                    f"Data not considered for public key import: {result}"
                )

            if len(result.imports) != 1:
                raise InvalidPacket(
                    "Public key packet does not contain exactly one public key (not"
                    " counting subkeys)."
                )

            try:
                key_obj = c.get_key(result.imports[0].fpr, secret=False)
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e
            except gpg.errors.KeyError as e:
                raise GPGProviderError("Newly imported public key not found") from e

            return GPGME_GPGPublicKey(key_obj)

    def backup_secret_key(self, secret_key: GPGSecretKey) -> bytes:
        assert isinstance(secret_key, GPGME_GPGSecretKey)
        # TODO
        # - Handle password protection/pinentry
        # - Make sure the key is exported unencrypted

        pattern = secret_key.public_key.fingerprint

        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                result = c.key_export_secret(pattern)
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e

            if result is None:
                raise UnknownKey(f"Secret key {pattern} not found.")

            return result

    def restore_secret_keys(self, data: bytes) -> Set[GPGSecretKey]:
        # TODO
        # - Reject packets older than version 4
        # - Check whether it's actually secret keys (through packet inspection?)

        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                result = c.key_import(data)
            except gpg.errors.GPGMEError as e:
                # From looking at the code, `key_import` never raises. The documentation
                # says it does though, so this is included for future-proofness.
                raise GPGProviderError("Internal GPGME error") from e

            if not hasattr(result, "considered"):
                raise InvalidPacket(
                    f"Data not considered for secret key import: {result}"
                )

            if len(result.imports) == 0:
                raise InvalidPacket("Secret key packet does not contain a secret key.")

            secret_keys = set()
            for import_status in result.imports:
                try:
                    key_obj = c.get_key(import_status.fpr, secret=True)
                except gpg.errors.GPGMEError as e:
                    raise GPGProviderError("Internal GPGME error") from e
                except gpg.errors.KeyError as e:
                    raise GPGProviderError("Newly imported secret key not found") from e

                secret_keys.add(GPGME_GPGSecretKey(GPGME_GPGPublicKey(key_obj)))

            return secret_keys

    def encrypt_symmetrically(self, plaintext: bytes, password: str) -> bytes:
        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                ciphertext, __, __ = c.encrypt(plaintext, passphrase=password, sign=False)
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e

            return ciphertext

    def decrypt_symmetrically(self, ciphertext: bytes, password: str) -> bytes:
        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                plaintext, __, __ = c.decrypt(
                    ciphertext,
                    passphrase=password,
                    verify=False
                )
            except gpg.errors.GPGMEError as e:
                # TODO: Find out what kind of error is raised if the password is wrong and
                # re-raise it as DecryptionFailed instead.
                raise GPGProviderError("Internal GPGME error") from e
            except gpg.UnsupportedAlgorithm as e:
                raise DecryptionFailed("Unsupported algorithm") from e

            return plaintext

    def sign(self, data: bytes, secret_keys: Set[GPGSecretKey]) -> bytes:
        signers = []
        for secret_key in secret_keys:
            assert isinstance(secret_key, GPGME_GPGSecretKey)

            signers.append(secret_key.public_key.key_obj)

        with gpg.Context(home_dir=self.__home_dir, signers=signers) as c:
            try:
                signed_data, __ = c.sign(data)
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e
            except gpg.errors.InvalidSigners as e:
                raise GPGProviderError(
                    "At least one of the secret keys is invalid for signing"
                ) from e

            return signed_data

    def sign_detached(self, data: bytes, secret_keys: Set[GPGSecretKey]) -> bytes:
        signers = []
        for secret_key in secret_keys:
            assert isinstance(secret_key, GPGME_GPGSecretKey)

            signers.append(secret_key.public_key.key_obj)

        with gpg.Context(home_dir=self.__home_dir, signers=signers) as c:
            try:
                signature, __ = c.sign(data, mode=gpg.constants.sig.mode.DETACH)
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e
            except gpg.errors.InvalidSigners as e:
                raise GPGProviderError(
                    "At least one of the secret keys is invalid for signing"
                ) from e

            return signature

    def verify(self, signed_data: bytes, public_keys: Set[GPGPublicKey]) -> bytes:
        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                data, result = c.verify(signed_data)
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e
            except gpg.errors.BadSignatures as e:
                raise VerificationFailed("Bad signatures on signed data") from e

            valid_signature_found = False
            for public_key in public_keys:
                assert isinstance(public_key, GPGME_GPGPublicKey)

                for subkey in public_key.key_obj.subkeys:
                    for sig in result.signatures:
                        if subkey.can_sign and subkey.fpr == sig.fpr:
                            valid_signature_found = True

            if not valid_signature_found:
                raise VerificationFailed(
                    "Data not signed by one of the expected public keys"
                )

            return data

    def verify_detached(
        self,
        data: bytes,
        signature: bytes,
        public_keys: Set[GPGPublicKey]
    ) -> None:
        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                __, result = c.verify(data, signature=signature)
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e
            except gpg.errors.BadSignatures as e:
                raise VerificationFailed("Bad signatures on signed data") from e

            valid_signature_found = False
            for public_key in public_keys:
                assert isinstance(public_key, GPGME_GPGPublicKey)

                for subkey in public_key.key_obj.subkeys:
                    for sig in result.signatures:
                        if subkey.can_sign and subkey.fpr == sig.fpr:
                            valid_signature_found = True

            if not valid_signature_found:
                raise VerificationFailed(
                    "Data not signed by one of the expected public keys"
                )

    def encrypt(
        self,
        plaintext: bytes,
        public_keys: Set[GPGPublicKey],
        signing_keys: Optional[Set[GPGSecretKey]] = None
    ) -> bytes:
        recipients = []
        for public_key in public_keys:
            assert isinstance(public_key, GPGME_GPGPublicKey)

            recipients.append(public_key.key_obj)

        signers = []
        if signing_keys is not None:
            for secret_key in signing_keys:
                assert isinstance(secret_key, GPGME_GPGSecretKey)

                signers.append(secret_key.public_key.key_obj)

        sign = signing_keys is not None

        with gpg.Context(home_dir=self.__home_dir, signers=signers) as c:
            try:
                ciphertext, __, __ = c.encrypt(
                    plaintext,
                    recipients=recipients,
                    sign=sign,
                    always_trust=True,
                    add_encrypt_to=True
                )
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e
            except gpg.errors.InvalidRecipients as e:
                raise GPGProviderError(
                    "At least one of the public keys is invalid for encryption"
                ) from e
            except gpg.errors.InvalidSigners as e:
                raise GPGProviderError(
                    "At least one of the signing keys is invalid for signing"
                ) from e

            return ciphertext

    def decrypt(
        self,
        ciphertext: bytes,
        secret_keys: Set[GPGSecretKey],
        public_keys: Optional[Set[GPGPublicKey]] = None
    ) -> bytes:
        verify = public_keys is not None

        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                plaintext, result, verify_result = c.decrypt(
                    ciphertext,
                    verify=verify
                )
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e
            except gpg.UnsupportedAlgorithm as e:
                raise DecryptionFailed("Unsupported algorithm") from e

            # TODO: Check whether the data was decrypted using one of the expected secret
            # keys

            if public_keys is not None:
                valid_signature_found = False
                for public_key in public_keys:
                    assert isinstance(public_key, GPGME_GPGPublicKey)

                    for subkey in public_key.key_obj.subkeys:
                        for sig in verify_result.signatures:
                            if subkey.can_sign and subkey.fpr == sig.fpr:
                                valid_signature_found = True

                if not valid_signature_found:
                    raise VerificationFailed(
                        "Data not signed by one of the expected public keys"
                    )

            return plaintext

    def list_public_keys(self, user_id: str) -> Set[GPGPublicKey]:
        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                return {
                    GPGME_GPGPublicKey(key)
                    for key
                    in c.keylist(pattern=user_id, secret=False)
                }
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e

    def list_secret_keys(self, user_id: str) -> Set[GPGSecretKey]:
        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                return {
                    GPGME_GPGSecretKey(GPGME_GPGPublicKey(key))
                    for key
                    in c.keylist(pattern=user_id, secret=True)
                }
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e

    def can_sign(self, public_key: GPGPublicKey) -> bool:
        assert isinstance(public_key, GPGME_GPGPublicKey)

        return any(subkey.can_sign for subkey in public_key.key_obj.subkeys)

    def can_encrypt(self, public_key: GPGPublicKey) -> bool:
        assert isinstance(public_key, GPGME_GPGPublicKey)

        return any(subkey.can_encrypt for subkey in public_key.key_obj.subkeys)

    def create_key(self, user_id: str) -> GPGSecretKey:
        with gpg.Context(home_dir=self.__home_dir) as c:
            try:
                result = c.create_key(
                    user_id,
                    expires=False,
                    sign=True,
                    encrypt=True,
                    certify=False,
                    authenticate=False,
                    force=True
                )

                key_obj = c.get_key(result.fpr, secret=True)
            except gpg.errors.GPGMEError as e:
                raise GPGProviderError("Internal GPGME error") from e
            except gpg.errors.KeyError as e:
                raise GPGProviderError("Newly created key not found") from e

            return GPGME_GPGSecretKey(GPGME_GPGPublicKey(key_obj))


class PublicKeyMetadata(NamedTuple):
    """
    Metadata about a published public key.
    """

    fingerprint: str
    timestamp: datetime


@enum.unique
class TrustLevel(enum.Enum):
    """
    The trust levels required for BTBV and manual trust.
    """

    TRUSTED: str = "TRUSTED"
    BLINDLY_TRUSTED: str = "BLINDLY_TRUSTED"
    UNDECIDED: str = "UNDECIDED"
    DISTRUSTED: str = "DISTRUSTED"


OPENPGP_SCHEMA = xmlschema.XMLSchema("""<?xml version="1.0" encoding="utf8"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
    targetNamespace="urn:xmpp:openpgp:0"
    xmlns="urn:xmpp:openpgp:0">

    <xs:element name="openpgp" type="xs:base64Binary"/>
</xs:schema>
""")


# The following schema needs verion 1.1 of XML Schema, which is not supported by lxml.
# Luckily, xmlschema exists, which is a clean, well maintained, cross-platform
# implementation of XML Schema, including version 1.1.
CONTENT_SCHEMA = xmlschema.XMLSchema11("""<?xml version="1.1" encoding="utf8"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
    targetNamespace="urn:xmpp:openpgp:0"
    xmlns="urn:xmpp:openpgp:0">

    <xs:element name="signcrypt">
        <xs:complexType>
            <xs:all>
                <xs:element ref="to" maxOccurs="unbounded"/>
                <xs:element ref="time"/>
                <xs:element ref="rpad" minOccurs="0"/>
                <xs:element ref="payload"/>
            </xs:all>
        </xs:complexType>
    </xs:element>

    <xs:element name="sign">
        <xs:complexType>
            <xs:all>
                <xs:element ref="to" maxOccurs="unbounded"/>
                <xs:element ref="time"/>
                <xs:element ref="rpad" minOccurs="0"/>
                <xs:element ref="payload"/>
            </xs:all>
        </xs:complexType>
    </xs:element>

    <xs:element name="crypt">
        <xs:complexType>
            <xs:all>
                <xs:element ref="to" minOccurs="0" maxOccurs="unbounded"/>
                <xs:element ref="time"/>
                <xs:element ref="rpad" minOccurs="0"/>
                <xs:element ref="payload"/>
            </xs:all>
        </xs:complexType>
    </xs:element>

    <xs:element name="to">
        <xs:complexType>
            <xs:attribute name="jid" type="xs:string"/>
        </xs:complexType>
    </xs:element>

    <xs:element name="time">
        <xs:complexType>
            <xs:attribute name="stamp" type="xs:dateTime"/>
        </xs:complexType>
    </xs:element>

    <xs:element name="rpad" type="xs:string"/>

    <xs:element name="payload">
        <xs:complexType>
            <xs:sequence>
                <xs:any minOccurs="0" maxOccurs="unbounded" processContents="skip"/>
            </xs:sequence>
        </xs:complexType>
    </xs:element>
</xs:schema>
""")


PUBLIC_KEYS_LIST_NODE = "urn:xmpp:openpgp:0:public-keys"
PUBLIC_KEYS_LIST_SCHEMA = xmlschema.XMLSchema("""<?xml version="1.0" encoding="utf8"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
    targetNamespace="urn:xmpp:openpgp:0"
    xmlns="urn:xmpp:openpgp:0">

    <xs:element name="public-keys-list">
        <xs:complexType>
            <xs:sequence>
                <xs:element ref="pubkey-metadata" minOccurs="0" maxOccurs="unbounded"/>
            </xs:sequence>
        </xs:complexType>
    </xs:element>

    <xs:element name="pubkey-metadata">
        <xs:complexType>
            <xs:attribute name="v4-fingerprint" type="xs:string"/>
            <xs:attribute name="date" type="xs:dateTime"/>
        </xs:complexType>
    </xs:element>
</xs:schema>
""")


PUBKEY_SCHEMA = xmlschema.XMLSchema("""<?xml version="1.0" encoding="utf8"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
    targetNamespace="urn:xmpp:openpgp:0"
    xmlns="urn:xmpp:openpgp:0">

    <xs:element name="pubkey">
        <xs:complexType>
            <xs:all>
                <xs:element ref="data"/>
            </xs:all>
            <xs:anyAttribute processContents="skip"/>
        </xs:complexType>
    </xs:element>

    <xs:element name="data" type="xs:base64Binary"/>
</xs:schema>
""")


SECRETKEY_SCHEMA = xmlschema.XMLSchema("""<?xml version="1.0" encoding="utf8"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
    targetNamespace="urn:xmpp:openpgp:0"
    xmlns="urn:xmpp:openpgp:0">

    <xs:element name="secretkey" type="xs:base64Binary"/>
</xs:schema>
""")


DEFAULT_TRUST_MODEL_PARAM = f"""
<params>
<individual>
<category name="{PARAM_CATEGORY}" label={quoteattr(D_('Security'))}>
    <param name="{PARAM_NAME}"
        label={quoteattr(D_('OMEMO default trust policy'))}
        type="list" security="3">
        <option value="manual" label={quoteattr(D_('Manual trust (more secure)'))} />
        <option value="btbv"
            label={quoteattr(D_('Blind Trust Before Verification (more user friendly)'))}
            selected="true" />
    </param>
</category>
</individual>
</params>
"""


def get_gpg_provider(sat: SAT, client: SatXMPPClient) -> GPGProvider:
    """Get the GPG provider for a client.

    @param sat: The SAT instance.
    @param client: The client.
    @return: The GPG provider specifically for that client.
    """

    return GPGME_GPGProvider(str(sat.get_local_path(client, "gnupg-home")))


def generate_passphrase() -> str:
    """Generate a secure passphrase for symmetric encryption.

    @return: The passphrase.
    """

    return "-".join("".join(
        secrets.choice("123456789ABCDEFGHIJKLMNPQRSTUVWXYZ") for __ in range(4)
    ) for __ in range(6))


# TODO: Handle the user id mess
class XEP_0373:
    """
    Implementation of XEP-0373: OpenPGP for XMPP under namespace ``urn:xmpp:openpgp:0``.
    """

    def __init__(self, host: SAT) -> None:
        """
        @param sat: The SAT instance.
        """

        self.host = host

        # Add configuration option to choose between manual trust and BTBV as the trust
        # model
        host.memory.updateParams(DEFAULT_TRUST_MODEL_PARAM)

        self.__xep_0045 = cast(Optional[XEP_0045], host.plugins.get("XEP-0045"))
        self.__xep_0060 = cast(XEP_0060, host.plugins["XEP-0060"])

        self.__storage: Dict[str, persistent.LazyPersistentBinaryDict] = {}

        xep_0163 = cast(XEP_0163, host.plugins["XEP-0163"])
        xep_0163.addPEPEvent(
            "OX_PUBLIC_KEYS_LIST",
            PUBLIC_KEYS_LIST_NODE,
            lambda items_event, profile: defer.ensureDeferred(
                self.__on_public_keys_list_update(items_event, profile)
            )
        )

    async def profileConnecting(self, client):
        client.gpg_provider = get_gpg_provider(self.host, client)

    async def profileConnected(  # pylint: disable=invalid-name
        self,
        client: SatXMPPClient
    ) -> None:
        """
        @param client: The client.
        """

        profile = cast(str, client.profile)

        if not profile in self.__storage:
            self.__storage[profile] = \
                persistent.LazyPersistentBinaryDict("XEP-0373", client.profile)

        if len(self.list_secret_keys(client)) == 0:
            log.debug(f"Generating first GPG key for {client.jid.userhost()}.")
            await self.create_key(client)

    async def __on_public_keys_list_update(
        self,
        items_event: pubsub.ItemsEvent,
        profile: str
    ) -> None:
        """Handle public keys list updates fired by PEP.

        @param items_event: The event.
        @param profile: The profile this event belongs to.
        """

        client = self.host.getClient(profile)

        sender = cast(jid.JID, items_event.sender)
        items = cast(List[domish.Element], items_event.items)

        if len(items) > 1:
            log.warning("Ignoring public keys list update with more than one element.")
            return

        item_elt = next(iter(items), None)
        if item_elt is None:
            log.debug("Ignoring empty public keys list update.")
            return

        public_keys_list_elt = cast(
            Optional[domish.Element],
            next(item_elt.elements(NS_OX, "public-keys-list"), None)
        )

        pubkey_metadata_elts: Optional[List[domish.Element]] = None

        if public_keys_list_elt is not None:
            try:
                PUBLIC_KEYS_LIST_SCHEMA.validate(public_keys_list_elt.toXml())
            except xmlschema.XMLSchemaValidationError:
                pass
            else:
                pubkey_metadata_elts = \
                    list(public_keys_list_elt.elements(NS_OX, "pubkey-metadata"))

        if pubkey_metadata_elts is None:
            log.warning(f"Malformed public keys list update item: {item_elt.toXml()}")
            return

        new_public_keys_metadata = { PublicKeyMetadata(
            fingerprint=cast(str, pubkey_metadata_elt["v4-fingerprint"]),
            timestamp=parse_datetime(cast(str, pubkey_metadata_elt["date"]))
        ) for pubkey_metadata_elt in pubkey_metadata_elts }

        storage_key = STR_KEY_PUBLIC_KEYS_METADATA.format(sender.userhost())

        local_public_keys_metadata = cast(
            Set[PublicKeyMetadata],
            await self.__storage[profile].get(storage_key, set())
        )

        unchanged_keys = new_public_keys_metadata & local_public_keys_metadata
        changed_or_new_keys = new_public_keys_metadata - unchanged_keys
        available_keys = self.list_public_keys(client, sender)

        for key_metadata in changed_or_new_keys:
            # Check whether the changed or new key has been imported before
            if any(key.fingerprint == key_metadata.fingerprint for key in available_keys):
                try:
                    # If it has been imported before, try to update it
                    await self.import_public_key(client, sender, key_metadata.fingerprint)
                except Exception as e:
                    log.warning(f"Public key import failed: {e}")

                    # If the update fails, remove the key from the local metadata list
                    # such that the update is attempted again next time
                    new_public_keys_metadata.remove(key_metadata)

        # Check whether this update was for our account and make sure all of our keys are
        # included in the update
        if sender.userhost() == client.jid.userhost():
            secret_keys = self.list_secret_keys(client)
            missing_keys = set(filter(lambda secret_key: all(
                key_metadata.fingerprint != secret_key.public_key.fingerprint
                for key_metadata
                in new_public_keys_metadata
            ), secret_keys))

            if len(missing_keys) > 0:
                log.warning(
                    "Public keys list update did not contain at least one of our keys."
                    f" {new_public_keys_metadata}"
                )

                for missing_key in missing_keys:
                    log.warning(missing_key.public_key.fingerprint)
                    new_public_keys_metadata.add(PublicKeyMetadata(
                        fingerprint=missing_key.public_key.fingerprint,
                        timestamp=datetime.now(timezone.utc)
                    ))

                await self.publish_public_keys_list(client, new_public_keys_metadata)

        await self.__storage[profile].force(storage_key, new_public_keys_metadata)

    def list_public_keys(self, client: SatXMPPClient, jid: jid.JID) -> Set[GPGPublicKey]:
        """List GPG public keys available for a JID.

        @param client: The client to perform this operation with.
        @param jid: The JID. Can be a bare JID.
        @return: The set of public keys available for this JID.
        """

        gpg_provider = get_gpg_provider(self.host, client)

        return gpg_provider.list_public_keys(f"xmpp:{jid.userhost()}")

    def list_secret_keys(self, client: SatXMPPClient) -> Set[GPGSecretKey]:
        """List GPG secret keys available for a JID.

        @param client: The client to perform this operation with.
        @return: The set of secret keys available for this JID.
        """

        gpg_provider = get_gpg_provider(self.host, client)

        return gpg_provider.list_secret_keys(f"xmpp:{client.jid.userhost()}")

    async def create_key(self, client: SatXMPPClient) -> GPGSecretKey:
        """Create a new GPG key, capable of signing and encryption.

        The key is generated without password protection and without expiration.

        @param client: The client to perform this operation with.
        @return: The new key.
        """

        gpg_provider = get_gpg_provider(self.host, client)

        secret_key = gpg_provider.create_key(f"xmpp:{client.jid.userhost()}")

        await self.publish_public_key(client, secret_key.public_key)

        storage_key = STR_KEY_PUBLIC_KEYS_METADATA.format(client.jid.userhost())

        public_keys_list = cast(
            Set[PublicKeyMetadata],
            await self.__storage[client.profile].get(storage_key, set())
        )

        public_keys_list.add(PublicKeyMetadata(
            fingerprint=secret_key.public_key.fingerprint,
            timestamp=datetime.now(timezone.utc)
        ))

        await self.publish_public_keys_list(client, public_keys_list)

        await self.__storage[client.profile].force(storage_key, public_keys_list)

        return secret_key

    @staticmethod
    def __build_content_element(
        element_name: Literal["signcrypt", "sign", "crypt"],
        recipient_jids: Iterable[jid.JID],
        include_rpad: bool
    ) -> Tuple[domish.Element, domish.Element]:
        """Build a content element.

        @param element_name: The name of the content element.
        @param recipient_jids: The intended recipients of this content element. Can be
            bare JIDs.
        @param include_rpad: Whether to include random-length random-content padding.
        @return: The content element and the ``<payload/>`` element to add the stanza
            extension elements to.
        """

        content_elt = domish.Element((NS_OX, element_name))

        for recipient_jid in recipient_jids:
            content_elt.addElement("to")["jid"] = recipient_jid.userhost()

        content_elt.addElement("time")["stamp"] = format_datetime()

        if include_rpad:
            # XEP-0373 doesn't specify bounds for the length of the random padding. This
            # uses the bounds specified in XEP-0420 for the closely related rpad affix.
            rpad_length = secrets.randbelow(201)
            rpad_content = "".join(
                secrets.choice(string.digits + string.ascii_letters + string.punctuation)
                for __
                in range(rpad_length)
            )
            content_elt.addElement("rpad", content=rpad_content)

        payload_elt = content_elt.addElement("payload")

        return content_elt, payload_elt

    @staticmethod
    def build_signcrypt_element(
        recipient_jids: Iterable[jid.JID]
    ) -> Tuple[domish.Element, domish.Element]:
        """Build a ``<signcrypt/>`` content element.

        @param recipient_jids: The intended recipients of this content element. Can be
            bare JIDs.
        @return: The ``<signcrypt/>`` element and the ``<payload/>`` element to add the
            stanza extension elements to.
        """

        if len(recipient_jids) == 0:
            raise ValueError("Recipient JIDs must be provided.")

        return XEP_0373.__build_content_element("signcrypt", recipient_jids, True)

    @staticmethod
    def build_sign_element(
        recipient_jids: Iterable[jid.JID],
        include_rpad: bool
    ) -> Tuple[domish.Element, domish.Element]:
        """Build a ``<sign/>`` content element.

        @param recipient_jids: The intended recipients of this content element. Can be
            bare JIDs.
        @param include_rpad: Whether to include random-length random-content padding,
            which is OPTIONAL for the ``<sign/>`` content element.
        @return: The ``<sign/>`` element and the ``<payload/>`` element to add the stanza
            extension elements to.
        """

        if len(recipient_jids) == 0:
            raise ValueError("Recipient JIDs must be provided.")

        return XEP_0373.__build_content_element("sign", recipient_jids, include_rpad)

    @staticmethod
    def build_crypt_element(
        recipient_jids: Iterable[jid.JID]
    ) -> Tuple[domish.Element, domish.Element]:
        """Build a ``<crypt/>`` content element.

        @param recipient_jids: The intended recipients of this content element. Specifying
            the intended recipients is OPTIONAL for the ``<crypt/>`` content element. Can
            be bare JIDs.
        @return: The ``<crypt/>`` element and the ``<payload/>`` element to add the stanza
            extension elements to.
        """

        return XEP_0373.__build_content_element("crypt", recipient_jids, True)

    async def build_openpgp_element(
        self,
        client: SatXMPPClient,
        content_elt: domish.Element,
        recipient_jids: Set[jid.JID]
    ) -> domish.Element:
        """Build an ``<openpgp/>`` element.

        @param client: The client to perform this operation with.
        @param content_elt: The content element to contain in the ``<openpgp/>`` element.
        @param recipient_jids: The recipient's JIDs. Can be bare JIDs.
        @return: The ``<openpgp/>`` element.
        """

        gpg_provider = get_gpg_provider(self.host, client)

        # TODO: I'm not sure whether we want to sign with all keys by default or choose
        # just one key/a subset of keys to sign with.
        signing_keys = set(filter(
            lambda secret_key: gpg_provider.can_sign(secret_key.public_key),
            self.list_secret_keys(client)
        ))

        encryption_keys: Set[GPGPublicKey] = set()

        for recipient_jid in recipient_jids:
            # Import all keys of the recipient
            all_public_keys = await self.import_all_public_keys(client, recipient_jid)

            # Filter for keys that can encrypt
            encryption_keys |= set(filter(gpg_provider.can_encrypt, all_public_keys))

        # TODO: Handle trust

        content = content_elt.toXml().encode("utf-8")
        data: bytes

        if content_elt.name == "signcrypt":
            data = gpg_provider.encrypt(content, encryption_keys, signing_keys)
        elif content_elt.name == "sign":
            data = gpg_provider.sign(content, signing_keys)
        elif content_elt.name == "crypt":
            data = gpg_provider.encrypt(content, encryption_keys)
        else:
            raise ValueError(f"Unknown content element <{content_elt.name}/>")

        openpgp_elt = domish.Element((NS_OX, "openpgp"))
        openpgp_elt.addContent(base64.b64encode(data).decode("ASCII"))
        return openpgp_elt

    async def unpack_openpgp_element(
        self,
        client: SatXMPPClient,
        openpgp_elt: domish.Element,
        element_name: Literal["signcrypt", "sign", "crypt"],
        sender_jid: jid.JID
    ) -> Tuple[domish.Element, datetime]:
        """Verify, decrypt and unpack an ``<openpgp/>`` element.

        @param client: The client to perform this operation with.
        @param openpgp_elt: The ``<openpgp/>`` element.
        @param element_name: The name of the content element.
        @param sender_jid: The sender's JID. Can be a bare JID.
        @return: The ``<payload/>`` element containing the decrypted/verified stanza
            extension elements carried by this ``<openpgp/>`` element, and the timestamp
            contained in the content element.
        @raise exceptions.ParsingError: on syntactical verification errors.
        @raise VerificationError: on semantical verification errors accoding to XEP-0373.
        @raise DecryptionFailed: on decryption failure.
        @raise VerificationFailed: if the data could not be verified.

        @warning: The timestamp is not verified for plausibility; this SHOULD be done by
            the calling code.
        """

        gpg_provider = get_gpg_provider(self.host, client)

        decryption_keys = set(filter(
            lambda secret_key: gpg_provider.can_encrypt(secret_key.public_key),
            self.list_secret_keys(client)
        ))

        # Import all keys of the sender
        all_public_keys = await self.import_all_public_keys(client, sender_jid)

        # Filter for keys that can sign
        verification_keys = set(filter(gpg_provider.can_sign, all_public_keys))

        # TODO: Handle trust

        try:
            OPENPGP_SCHEMA.validate(openpgp_elt.toXml())
        except xmlschema.XMLSchemaValidationError as e:
            raise exceptions.ParsingError(
                "<openpgp/> element doesn't pass schema validation."
            ) from e

        openpgp_message = base64.b64decode(str(openpgp_elt))
        content: bytes

        if element_name == "signcrypt":
            content = gpg_provider.decrypt(
                openpgp_message,
                decryption_keys,
                public_keys=verification_keys
            )
        elif element_name == "sign":
            content = gpg_provider.verify(openpgp_message, verification_keys)
        elif element_name == "crypt":
            content = gpg_provider.decrypt(openpgp_message, decryption_keys)
        else:
            assert_never(element_name)

        try:
            content_elt = cast(
                domish.Element,
                xml_tools.ElementParser()(content.decode("utf-8"))
            )
        except UnicodeDecodeError as e:
            raise exceptions.ParsingError("UTF-8 decoding error") from e

        try:
            CONTENT_SCHEMA.validate(content_elt.toXml())
        except xmlschema.XMLSchemaValidationError as e:
            raise exceptions.ParsingError(
                f"<{element_name}/> element doesn't pass schema validation."
            ) from e

        if content_elt.name != element_name:
            raise exceptions.ParsingError(f"Not a <{element_name}/> element.")

        recipient_jids = \
            { jid.JID(to_elt["jid"]) for to_elt in content_elt.elements(NS_OX, "to") }

        if (
            client.jid.userhostJID() not in { jid.userhostJID() for jid in recipient_jids }
            and element_name != "crypt"
        ):
            raise VerificationError(
                f"Recipient list in <{element_name}/> element does not list our (bare)"
                f" JID."
            )

        time_elt = next(content_elt.elements(NS_OX, "time"))

        timestamp = parse_datetime(time_elt["stamp"])

        payload_elt = next(content_elt.elements(NS_OX, "payload"))

        return payload_elt, timestamp

    async def publish_public_key(
        self,
        client: SatXMPPClient,
        public_key: GPGPublicKey
    ) -> None:
        """Publish a public key.

        @param client: The client.
        @param public_key: The public key to publish.
        @raise XMPPInteractionFailed: if any interaction via XMPP failed.
        """

        gpg_provider = get_gpg_provider(self.host, client)

        packet = gpg_provider.export_public_key(public_key)

        node = f"urn:xmpp:openpgp:0:public-keys:{public_key.fingerprint}"

        pubkey_elt = domish.Element((NS_OX, "pubkey"))

        pubkey_elt.addElement("data", content=base64.b64encode(packet).decode("ASCII"))

        try:
            await self.__xep_0060.sendItem(
                client,
                client.jid.userhostJID(),
                node,
                pubkey_elt,
                format_datetime(),
                extra={
                    XEP_0060.EXTRA_PUBLISH_OPTIONS: {
                        XEP_0060.OPT_PERSIST_ITEMS: "true",
                        XEP_0060.OPT_ACCESS_MODEL: "open",
                        XEP_0060.OPT_MAX_ITEMS: 1
                    },
                    # TODO: Do we really want publish_without_options here?
                    XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "publish_without_options"
                }
            )
        except Exception as e:
            raise XMPPInteractionFailed("Publishing the public key failed.") from e

    async def import_all_public_keys(
        self,
        client: SatXMPPClient,
        entity_jid: jid.JID
    ) -> Set[GPGPublicKey]:
        """Import all public keys of a JID that have not been imported before.

        @param client: The client.
        @param jid: The JID. Can be a bare JID.
        @return: The public keys.
        @note: Failure to import a key simply results in the key not being included in the
            result.
        """

        available_public_keys = self.list_public_keys(client, entity_jid)

        storage_key = STR_KEY_PUBLIC_KEYS_METADATA.format(entity_jid.userhost())

        public_keys_metadata = cast(
            Set[PublicKeyMetadata],
            await self.__storage[client.profile].get(storage_key, set())
        )
        if not public_keys_metadata:
            public_keys_metadata = await self.download_public_keys_list(
                client, entity_jid
            )
            if not public_keys_metadata:
                raise exceptions.NotFound(
                    f"Can't find public keys for {entity_jid}"
                )
            else:
                await self.__storage[client.profile].aset(
                    storage_key, public_keys_metadata
                )


        missing_keys = set(filter(lambda public_key_metadata: all(
            public_key_metadata.fingerprint != public_key.fingerprint
            for public_key
            in available_public_keys
        ), public_keys_metadata))

        for missing_key in missing_keys:
            try:
                available_public_keys.add(
                    await self.import_public_key(client, entity_jid, missing_key.fingerprint)
                )
            except Exception as e:
                log.warning(
                    f"Import of public key {missing_key.fingerprint} owned by"
                    f" {entity_jid.userhost()} failed, ignoring: {e}"
                )

        return available_public_keys

    async def import_public_key(
        self,
        client: SatXMPPClient,
        jid: jid.JID,
        fingerprint: str
    ) -> GPGPublicKey:
        """Import a public key.

        @param client: The client.
        @param jid: The JID owning the public key. Can be a bare JID.
        @param fingerprint: The fingerprint of the public key.
        @return: The public key.
        @raise exceptions.NotFound: if the public key was not found.
        @raise exceptions.ParsingError: on XML-level parsing errors.
        @raise InvalidPacket: if the packet is either syntactically or semantically deemed
            invalid.
        @raise XMPPInteractionFailed: if any interaction via XMPP failed.
        """

        gpg_provider = get_gpg_provider(self.host, client)

        node = f"urn:xmpp:openpgp:0:public-keys:{fingerprint}"

        try:
            items, __ = await self.__xep_0060.getItems(
                client,
                jid.userhostJID(),
                node,
                max_items=1
            )
        except exceptions.NotFound as e:
            raise exceptions.NotFound(
                f"No public key with fingerprint {fingerprint} published by JID"
                f" {jid.userhost()}."
            ) from e
        except Exception as e:
            raise XMPPInteractionFailed("Fetching the public keys list failed.") from e

        try:
            item_elt = cast(domish.Element, items[0])
        except IndexError as e:
            raise exceptions.NotFound(
                f"No public key with fingerprint {fingerprint} published by JID"
                f" {jid.userhost()}."
            ) from e

        pubkey_elt = cast(
            Optional[domish.Element],
            next(item_elt.elements(NS_OX, "pubkey"), None)
        )

        if pubkey_elt is None:
            raise exceptions.ParsingError(
                f"Publish-Subscribe item of JID {jid.userhost()} doesn't contain pubkey"
                f" element."
            )

        try:
            PUBKEY_SCHEMA.validate(pubkey_elt.toXml())
        except xmlschema.XMLSchemaValidationError as e:
            raise exceptions.ParsingError(
                f"Publish-Subscribe item of JID {jid.userhost()} doesn't pass pubkey"
                f" schema validation."
            ) from e

        public_key = gpg_provider.import_public_key(base64.b64decode(str(
            next(pubkey_elt.elements(NS_OX, "data"))
        )))

        return public_key

    async def publish_public_keys_list(
        self,
        client: SatXMPPClient,
        public_keys_list: Iterable[PublicKeyMetadata]
    ) -> None:
        """Publish/update the own public keys list.

        @param client: The client.
        @param public_keys_list: The public keys list.
        @raise XMPPInteractionFailed: if any interaction via XMPP failed.

        @warning: All public keys referenced in the public keys list MUST be published
            beforehand.
        """

        if len({ pkm.fingerprint for pkm in public_keys_list }) != len(public_keys_list):
            raise ValueError("Public keys list contains duplicate fingerprints.")

        node = "urn:xmpp:openpgp:0:public-keys"

        public_keys_list_elt = domish.Element((NS_OX, "public-keys-list"))

        for public_key_metadata in public_keys_list:
            pubkey_metadata_elt = public_keys_list_elt.addElement("pubkey-metadata")
            pubkey_metadata_elt["v4-fingerprint"] = public_key_metadata.fingerprint
            pubkey_metadata_elt["date"] = format_datetime(public_key_metadata.timestamp)

        try:
            await self.__xep_0060.sendItem(
                client,
                client.jid.userhostJID(),
                node,
                public_keys_list_elt,
                item_id=XEP_0060.ID_SINGLETON,
                extra={
                    XEP_0060.EXTRA_PUBLISH_OPTIONS: {
                        XEP_0060.OPT_PERSIST_ITEMS: "true",
                        XEP_0060.OPT_ACCESS_MODEL: "open",
                        XEP_0060.OPT_MAX_ITEMS: 1
                    },
                    # TODO: Do we really want publish_without_options here?
                    XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "publish_without_options"
                }
            )
        except Exception as e:
            raise XMPPInteractionFailed("Publishing the public keys list failed.") from e

    async def download_public_keys_list(
        self,
        client: SatXMPPClient,
        jid: jid.JID
    ) -> Optional[Set[PublicKeyMetadata]]:
        """Download the public keys list of a JID.

        @param client: The client.
        @param jid: The JID. Can be a bare JID.
        @return: The public keys list or ``None`` if the JID hasn't published a public
            keys list. An empty list means the JID has published an empty list.
        @raise exceptions.ParsingError: on XML-level parsing errors.
        @raise XMPPInteractionFailed: if any interaction via XMPP failed.
        """

        node = "urn:xmpp:openpgp:0:public-keys"

        try:
            items, __ = await self.__xep_0060.getItems(
                client,
                jid.userhostJID(),
                node,
                max_items=1
            )
        except exceptions.NotFound:
            return None
        except Exception as e:
            raise XMPPInteractionFailed() from e

        try:
            item_elt = cast(domish.Element, items[0])
        except IndexError:
            return None

        public_keys_list_elt = cast(
            Optional[domish.Element],
            next(item_elt.elements(NS_OX, "public-keys-list"), None)
        )

        if public_keys_list_elt is None:
            return None

        try:
            PUBLIC_KEYS_LIST_SCHEMA.validate(public_keys_list_elt.toXml())
        except xmlschema.XMLSchemaValidationError as e:
            raise exceptions.ParsingError(
                f"Publish-Subscribe item of JID {jid.userhost()} doesn't pass public keys"
                f" list schema validation."
            ) from e

        return {
            PublicKeyMetadata(
                fingerprint=pubkey_metadata_elt["v4-fingerprint"],
                timestamp=parse_datetime(pubkey_metadata_elt["date"])
            )
            for pubkey_metadata_elt
            in public_keys_list_elt.elements(NS_OX, "pubkey-metadata")
        }

    async def __prepare_secret_key_synchronization(
        self,
        client: SatXMPPClient
    ) -> Optional[domish.Element]:
        """Prepare for secret key synchronization.

        Makes sure the relative protocols and protocol extensions are supported by the
        server and makes sure that the PEP node for secret synchronization exists and is
        configured correctly. The node is created if necessary.

        @param client: The client.
        @return: As part of the preparations, the secret key synchronization PEP node is
            fetched. The result of that fetch is returned here.
        @raise exceptions.FeatureNotFound: if the server lacks support for the required
            protocols or protocol extensions.
        @raise XMPPInteractionFailed: if any interaction via XMPP failed.
        """

        try:
            infos = cast(DiscoInfo, await self.host.memory.disco.getInfos(
                client,
                client.jid.userhostJID()
            ))
        except Exception as e:
            raise XMPPInteractionFailed(
                "Error performing service discovery on the own bare JID."
            ) from e

        identities = cast(Dict[Tuple[str, str], str], infos.identities)
        features = cast(Set[DiscoFeature], infos.features)

        if ("pubsub", "pep") not in identities:
            raise exceptions.FeatureNotFound("Server doesn't support PEP.")

        if "http://jabber.org/protocol/pubsub#access-whitelist" not in features:
            raise exceptions.FeatureNotFound(
                "Server doesn't support the whitelist access model."
            )

        persistent_items_supported = \
            "http://jabber.org/protocol/pubsub#persistent-items" in features

        # TODO: persistent-items is a SHOULD, how do we handle the feature missing?

        node = "urn:xmpp:openpgp:0:secret-key"

        try:
            items, __ = await self.__xep_0060.getItems(
                client,
                client.jid.userhostJID(),
                node,
                max_items=1
            )
        except exceptions.NotFound:
            try:
                await self.__xep_0060.createNode(
                    client,
                    client.jid.userhostJID(),
                    node,
                    {
                        XEP_0060.OPT_PERSIST_ITEMS: "true",
                        XEP_0060.OPT_ACCESS_MODEL: "whitelist",
                        XEP_0060.OPT_MAX_ITEMS: "1"
                    }
                )
            except Exception as e:
                raise XMPPInteractionFailed(
                    "Error creating the secret key synchronization node."
                ) from e
        except Exception as e:
            raise XMPPInteractionFailed(
                "Error fetching the secret key synchronization node."
            ) from e

        try:
            return cast(domish.Element, items[0])
        except IndexError:
            return None

    async def export_secret_keys(
        self,
        client: SatXMPPClient,
        secret_keys: Iterable[GPGSecretKey]
    ) -> str:
        """Export secret keys to synchronize them with other devices.

        @param client: The client.
        @param secret_keys: The secret keys to export.
        @return: The backup code needed to decrypt the exported secret keys.
        @raise exceptions.FeatureNotFound: if the server lacks support for the required
            protocols or protocol extensions.
        @raise XMPPInteractionFailed: if any interaction via XMPP failed.
        """

        gpg_provider = get_gpg_provider(self.host, client)

        await self.__prepare_secret_key_synchronization(client)

        backup_code = generate_passphrase()

        plaintext = b"".join(
            gpg_provider.backup_secret_key(secret_key) for secret_key in secret_keys
        )

        ciphertext = gpg_provider.encrypt_symmetrically(plaintext, backup_code)

        node = "urn:xmpp:openpgp:0:secret-key"

        secretkey_elt = domish.Element((NS_OX, "secretkey"))
        secretkey_elt.addContent(base64.b64encode(ciphertext).decode("ASCII"))

        try:
            await self.__xep_0060.sendItem(
                client,
                client.jid.userhostJID(),
                node,
                secretkey_elt
            )
        except Exception as e:
            raise XMPPInteractionFailed("Publishing the secret keys failed.") from e

        return backup_code

    async def download_secret_keys(self, client: SatXMPPClient) -> Optional[bytes]:
        """Download previously exported secret keys to import them in a second step.

        The downloading and importing steps are separate since a backup code is required
        for the import and it should be possible to try multiple backup codes without
        redownloading the data every time. The second half of the import procedure is
        provided by :meth:`import_secret_keys`.

        @param client: The client.
        @return: The encrypted secret keys previously exported, if any.
        @raise exceptions.FeatureNotFound: if the server lacks support for the required
            protocols or protocol extensions.
        @raise exceptions.ParsingError: on XML-level parsing errors.
        @raise XMPPInteractionFailed: if any interaction via XMPP failed.
        """

        item_elt = await self.__prepare_secret_key_synchronization(client)
        if item_elt is None:
            return None

        secretkey_elt = cast(
            Optional[domish.Element],
            next(item_elt.elements(NS_OX, "secretkey"), None)
        )

        if secretkey_elt is None:
            return None

        try:
            SECRETKEY_SCHEMA.validate(secretkey_elt.toXml())
        except xmlschema.XMLSchemaValidationError as e:
            raise exceptions.ParsingError(
                "Publish-Subscribe item doesn't pass secretkey schema validation."
            ) from e

        return base64.b64decode(str(secretkey_elt))

    def import_secret_keys(
        self,
        client: SatXMPPClient,
        ciphertext: bytes,
        backup_code: str
    ) -> Set[GPGSecretKey]:
        """Import previously downloaded secret keys.

        The downloading and importing steps are separate since a backup code is required
        for the import and it should be possible to try multiple backup codes without
        redownloading the data every time. The first half of the import procedure is
        provided by :meth:`download_secret_keys`.

        @param client: The client to perform this operation with.
        @param ciphertext: The ciphertext, i.e. the data returned by
            :meth:`download_secret_keys`.
        @param backup_code: The backup code needed to decrypt the data.
        @raise InvalidPacket: if one of the GPG packets building the secret key data is
            either syntactically or semantically deemed invalid.
        @raise DecryptionFailed: on decryption failure.
        """

        gpg_provider = get_gpg_provider(self.host, client)

        return gpg_provider.restore_secret_keys(gpg_provider.decrypt_symmetrically(
            ciphertext,
            backup_code
        ))

    @staticmethod
    def __get_joined_muc_users(
        client: SatXMPPClient,
        xep_0045: XEP_0045,
        room_jid: jid.JID
    ) -> Set[jid.JID]:
        """
        @param client: The client.
        @param xep_0045: A MUC plugin instance.
        @param room_jid: The room JID.
        @return: A set containing the bare JIDs of the MUC participants.
        @raise InternalError: if the MUC is not joined or the entity information of a
            participant isn't available.
        """
        # TODO: This should probably be a global helper somewhere

        bare_jids: Set[jid.JID] = set()

        try:
            room = cast(muc.Room, xep_0045.getRoom(client, room_jid))
        except exceptions.NotFound as e:
            raise exceptions.InternalError(
                "Participant list of unjoined MUC requested."
            ) from e

        for user in cast(Dict[str, muc.User], room.roster).values():
            entity = cast(Optional[SatXMPPEntity], user.entity)
            if entity is None:
                raise exceptions.InternalError(
                    f"Participant list of MUC requested, but the entity information of"
                    f" the participant {user} is not available."
                )

            bare_jids.add(entity.jid.userhostJID())

        return bare_jids

    async def get_trust(
        self,
        client: SatXMPPClient,
        public_key: GPGPublicKey,
        owner: jid.JID
    ) -> TrustLevel:
        """Query the trust level of a public key.

        @param client: The client to perform this operation under.
        @param public_key: The public key.
        @param owner: The owner of the public key. Can be a bare JID.
        @return: The trust level.
        """

        key = f"/trust/{owner.userhost()}/{public_key.fingerprint}"

        try:
            return TrustLevel(await self.__storage[client.profile][key])
        except KeyError:
            return TrustLevel.UNDECIDED

    async def set_trust(
        self,
        client: SatXMPPClient,
        public_key: GPGPublicKey,
        owner: jid.JID,
        trust_level: TrustLevel
    ) -> None:
        """Set the trust level of a public key.

        @param client: The client to perform this operation under.
        @param public_key: The public key.
        @param owner: The owner of the public key. Can be a bare JID.
        @param trust_leve: The trust level.
        """

        key = f"/trust/{owner.userhost()}/{public_key.fingerprint}"

        await self.__storage[client.profile].force(key, trust_level.name)

    async def getTrustUI(  # pylint: disable=invalid-name
        self,
        client: SatXMPPClient,
        entity: jid.JID
    ) -> xml_tools.XMLUI:
        """
        @param client: The client.
        @param entity: The entity whose device trust levels to manage.
        @return: An XMLUI instance which opens a form to manage the trust level of all
            devices belonging to the entity.
        """

        if entity.resource:
            raise ValueError("A bare JID is expected.")

        bare_jids: Set[jid.JID]
        if self.__xep_0045 is not None and self.__xep_0045.isJoinedRoom(client, entity):
            bare_jids = self.__get_joined_muc_users(client, self.__xep_0045, entity)
        else:
            bare_jids = { entity.userhostJID() }

        all_public_keys = list({
            bare_jid: list(self.list_public_keys(client, bare_jid))
            for bare_jid
            in bare_jids
        }.items())

        async def callback(
            data: Any,
            profile: str  # pylint: disable=unused-argument
        ) -> Dict[Never, Never]:
            """
            @param data: The XMLUI result produces by the trust UI form.
            @param profile: The profile.
            @return: An empty dictionary. The type of the return value was chosen
                conservatively since the exact options are neither known not needed here.
            """

            if C.bool(data.get("cancelled", "false")):
                return {}

            data_form_result = cast(
                Dict[str, str],
                xml_tools.XMLUIResult2DataFormResult(data)
            )
            for key, value in data_form_result.items():
                if not key.startswith("trust_"):
                    continue

                outer_index, inner_index = key.split("_")[1:]

                owner, public_keys = all_public_keys[int(outer_index)]
                public_key = public_keys[int(inner_index)]
                trust = TrustLevel(value)

                if (await self.get_trust(client, public_key, owner)) is not trust:
                    await self.set_trust(client, public_key, owner, value)

            return {}

        submit_id = self.host.registerCallback(callback, with_data=True, one_shot=True)

        result = xml_tools.XMLUI(
            panel_type=C.XMLUI_FORM,
            title=D_("OX trust management"),
            submit_id=submit_id
        )
        # Casting this to Any, otherwise all calls on the variable cause type errors
        # pylint: disable=no-member
        trust_ui = cast(Any, result)
        trust_ui.addText(D_(
            "This is OX trusting system. You'll see below the GPG keys of your "
            "contacts, and a list selection to trust them or not. A trusted key "
            "can read your messages in plain text, so be sure to only validate "
            "keys that you are sure are belonging to your contact. It's better "
            "to do this when you are next to your contact, so "
            "you can check the \"fingerprint\" of the key "
            "yourself. Do *not* validate a key if the fingerprint is wrong!"
        ))

        own_secret_keys = self.list_secret_keys(client)

        trust_ui.changeContainer("label")
        for index, secret_key in enumerate(own_secret_keys):
            trust_ui.addLabel(D_(f"Own secret key {index} fingerprint"))
            trust_ui.addText(secret_key.public_key.fingerprint)
            trust_ui.addEmpty()
            trust_ui.addEmpty()

        for outer_index, [ owner, public_keys ] in enumerate(all_public_keys):
            for inner_index, public_key in enumerate(public_keys):
                trust_ui.addLabel(D_("Contact"))
                trust_ui.addJid(jid.JID(owner))
                trust_ui.addLabel(D_("Fingerprint"))
                trust_ui.addText(public_key.fingerprint)
                trust_ui.addLabel(D_("Trust this device?"))

                current_trust_level = await self.get_trust(client, public_key, owner)
                avaiable_trust_levels = \
                    { TrustLevel.DISTRUSTED, TrustLevel.TRUSTED, current_trust_level }

                trust_ui.addList(
                    f"trust_{outer_index}_{inner_index}",
                    options=[ trust_level.name for trust_level in avaiable_trust_levels ],
                    selected=current_trust_level.name,
                    styles=[ "inline" ]
                )

                trust_ui.addEmpty()
                trust_ui.addEmpty()

        return result