diff --git a/libervia/backend/core/exceptions.py b/libervia/backend/core/exceptions.py index 546e3ab0..274918fb 100644 --- a/libervia/backend/core/exceptions.py +++ b/libervia/backend/core/exceptions.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 -# SàT Exceptions +# Libervia Exceptions # Copyright (C) 2011 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify diff --git a/libervia/backend/plugins/plugin_sec_pubsub_signing.py b/libervia/backend/plugins/plugin_sec_pubsub_signing.py index 0280ba3b..1e3fda16 100644 --- a/libervia/backend/plugins/plugin_sec_pubsub_signing.py +++ b/libervia/backend/plugins/plugin_sec_pubsub_signing.py @@ -37,7 +37,7 @@ from libervia.backend.core.log import getLogger from libervia.backend.tools import utils from libervia.backend.tools.common import data_format -from .plugin_xep_0373 import VerificationFailed +from .plugin_xep_0373.exceptions import VerificationFailed log = getLogger(__name__) diff --git a/libervia/backend/plugins/plugin_xep_0373.py b/libervia/backend/plugins/plugin_xep_0373/__init__.py similarity index 61% rename from libervia/backend/plugins/plugin_xep_0373.py rename to libervia/backend/plugins/plugin_xep_0373/__init__.py index d19a22dc..6d4e4f2f 100644 --- a/libervia/backend/plugins/plugin_xep_0373.py +++ b/libervia/backend/plugins/plugin_xep_0373/__init__.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 # Libervia plugin for OpenPGP for XMPP +# Copyright (C) 2025-2026 Jérôme Poisosn (goffi@goffi.org) # Copyright (C) 2022-2024 Tim Henkes (me@syndace.dev) # This program is free software: you can redistribute it and/or modify @@ -16,66 +17,58 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -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 import Any, Iterable, Literal, cast from typing import Final, NamedTuple, Never, assert_never -from wokkel import muc, pubsub -from wokkel.disco import DiscoFeature, DiscoInfo -import xmlschema +from xml.sax.saxutils import quoteattr from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C from libervia.backend.core.core_types import SatXMPPEntity -from libervia.backend.core.i18n import _, D_ -from libervia.backend.core.log import getLogger, Logger +from libervia.backend.core.i18n import D_, _ +from libervia.backend.core.log import Logger, getLogger from libervia.backend.core.main import LiberviaBackend from libervia.backend.core.xmpp import SatXMPPClient from libervia.backend.memory import persistent from libervia.backend.plugins.plugin_xep_0045 import XEP_0045 from libervia.backend.plugins.plugin_xep_0060 import XEP_0060 from libervia.backend.plugins.plugin_xep_0163 import XEP_0163 -from libervia.backend.tools.xmpp_datetime import format_datetime, parse_datetime from libervia.backend.tools import xml_tools +from libervia.backend.tools.xmpp_datetime import format_datetime, parse_datetime from twisted.internet import defer from twisted.words.protocols.jabber import jid from twisted.words.xish import domish +from wokkel import muc, pubsub +from wokkel.disco import DiscoFeature, DiscoInfo +import xmlschema + +from .exceptions import VerificationError, XMPPInteractionFailed +from .models import GPGProviderBase, GPGPublicKeyBase, GPGSecretKeyBase try: + # GPG (GPG-ME) is the official Python binding, using system packages 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", - "TrustLevel", -] - +except ImportError: + # If not availble, we fallback on python-gnupg, which an unofficial binding which can + # be installed without system packages (gpg binary still needs to be installed). + try: + import gnupg + except ImportError: + raise exceptions.MissingModule( + "You are missing the required GPG backend. You need either the 'gpg' package" + " (recommended, requires 'gpgme' library) or the 'gnupg' package (Python" + " wrapper for gpg command). See" + " https://wiki.python.org/moin/GnuPrivacyGuard#Accessing_GnuPG_via_gpgme for" + " the gpg package and https://gnupg.readthedocs.io/ for gnupg." + ) + else: + from .gnupg_backend import GPGPublicKey, GPGSecretKey, GPGProvider +else: + from .gpg_backend import GPGPublicKey, GPGSecretKey, GPGProvider log = cast(Logger, getLogger(__name__)) # type: ignore[no-untyped-call] @@ -157,712 +150,6 @@ def binary_to_ascii_armor( return "\n".join(armor) -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). - """ - - ALGORITHM = "ed25519/cert,sign+cv25519/encr" - - 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 - - kwargs = {"home_dir": self.__home_dir, "signers": signers} - - with gpg.Context(**kwargs) 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: - # A single ECC (primary) key with signing and encryption capabilities is not supported, a - # subkey is required for encryption. - # Create a primary key only capable of certifying subkeys - primary_create_result = c.create_key( - user_id, - algorithm=self.ALGORITHM, - expires=False, - certify=True, - force=True, - ) - - primary_key_obj = c.get_key(primary_create_result.fpr, secret=True) - - c.create_subkey( - primary_key_obj, algorithm=self.ALGORITHM, expires=False, sign=True - ) - - c.create_subkey( - primary_key_obj, algorithm=self.ALGORITHM, expires=False, encrypt=True - ) - except gpg.errors.GPGMEError as e: - raise GPGProviderError("Internal GPGME error") from e - except gpg.errors.KeyNotFound as e: - raise GPGProviderError("Newly created key not found") from e - - return GPGME_GPGSecretKey(GPGME_GPGPublicKey(primary_key_obj)) - - class PublicKeyMetadata(NamedTuple): """ Metadata about a published public key. @@ -896,23 +183,20 @@ class TrustLevel(enum.Enum): DISTRUSTED: str = "DISTRUSTED" -OPENPGP_SCHEMA = xmlschema.XMLSchema( - """ +OPENPGP_SCHEMA = xmlschema.XMLSchema(""" -""" -) +""") # 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( - """ +CONTENT_SCHEMA = xmlschema.XMLSchema11(""" @@ -972,13 +256,11 @@ CONTENT_SCHEMA = xmlschema.XMLSchema11( -""" -) +""") PUBLIC_KEYS_LIST_NODE = "urn:xmpp:openpgp:0:public-keys" -PUBLIC_KEYS_LIST_SCHEMA = xmlschema.XMLSchema( - """ +PUBLIC_KEYS_LIST_SCHEMA = xmlschema.XMLSchema(""" @@ -998,12 +280,10 @@ PUBLIC_KEYS_LIST_SCHEMA = xmlschema.XMLSchema( -""" -) +""") -PUBKEY_SCHEMA = xmlschema.XMLSchema( - """ +PUBKEY_SCHEMA = xmlschema.XMLSchema(""" @@ -1019,32 +299,29 @@ PUBKEY_SCHEMA = xmlschema.XMLSchema( -""" -) +""") -SECRETKEY_SCHEMA = xmlschema.XMLSchema( - """ +SECRETKEY_SCHEMA = xmlschema.XMLSchema(""" -""" -) +""") DEFAULT_TRUST_MODEL_PARAM = f""" - + - @@ -1053,15 +330,14 @@ DEFAULT_TRUST_MODEL_PARAM = f""" """ -def get_gpg_provider(sat: LiberviaBackend, client: SatXMPPEntity) -> GPGProvider: +def get_gpg_provider(host: LiberviaBackend, client: SatXMPPEntity) -> GPGProviderBase: """Get the GPG provider for a client. - @param sat: The SAT instance. + @param host: The LiberviaBackend instance. @param client: The client. @return: The GPG provider specifically for that client. """ - - return GPGME_GPGProvider(str(sat.get_local_path(client, "gnupg-home"))) + return GPGProvider(str(host.get_local_path(client, "gnupg-home"))) def generate_passphrase() -> str: @@ -1093,10 +369,10 @@ class XEP_0373: # model host.memory.update_params(DEFAULT_TRUST_MODEL_PARAM) - self.__xep_0045 = cast(Optional[XEP_0045], host.plugins.get("XEP-0045")) + self.__xep_0045 = cast(XEP_0045 | None, host.plugins.get("XEP-0045")) self.__xep_0060 = cast(XEP_0060, host.plugins["XEP-0060"]) - self.__storage: Dict[str, persistent.LazyPersistentBinaryDict] = {} + self.__storage: dict[str, persistent.LazyPersistentBinaryDict] = {} xep_0163 = cast(XEP_0163, host.plugins["XEP-0163"]) xep_0163.add_pep_event( @@ -1140,7 +416,7 @@ class XEP_0373: client = self.host.get_client(profile) sender = cast(jid.JID, items_event.sender) - items = cast(List[domish.Element], items_event.items) + items = cast(list[domish.Element], items_event.items) if len(items) > 1: log.warning("Ignoring public keys list update with more than one element.") @@ -1152,11 +428,11 @@ class XEP_0373: return public_keys_list_elt = cast( - Optional[domish.Element], + domish.Element | None, next(item_elt.elements(NS_OX, "public-keys-list"), None), ) - pubkey_metadata_elts: Optional[List[domish.Element]] = None + pubkey_metadata_elts: list[domish.Element] | None = None if public_keys_list_elt is not None: try: @@ -1239,7 +515,9 @@ class XEP_0373: storage_key, [pkm.to_dict() for pkm in new_public_keys_metadata] ) - def list_public_keys(self, client: SatXMPPClient, jid: jid.JID) -> Set[GPGPublicKey]: + def list_public_keys( + self, client: SatXMPPClient, jid: jid.JID + ) -> set[GPGPublicKeyBase]: """List GPG public keys available for a JID. @param client: The client to perform this operation with. @@ -1251,7 +529,7 @@ class XEP_0373: return gpg_provider.list_public_keys(f"xmpp:{jid.userhost()}") - def list_secret_keys(self, client: SatXMPPClient) -> Set[GPGSecretKey]: + def list_secret_keys(self, client: SatXMPPClient) -> set[GPGSecretKeyBase]: """List GPG secret keys available for a JID. @param client: The client to perform this operation with. @@ -1262,7 +540,7 @@ class XEP_0373: return gpg_provider.list_secret_keys(f"xmpp:{client.jid.userhost()}") - async def create_key(self, client: SatXMPPClient) -> GPGSecretKey: + async def create_key(self, client: SatXMPPClient) -> GPGSecretKeyBase: """Create a new GPG key, capable of signing and encryption. The key is generated without password protection and without expiration. @@ -1304,7 +582,7 @@ class XEP_0373: element_name: Literal["signcrypt", "sign", "crypt"], recipient_jids: Iterable[jid.JID], include_rpad: bool, - ) -> Tuple[domish.Element, domish.Element]: + ) -> tuple[domish.Element, domish.Element]: """Build a content element. @param element_name: The name of the content element. @@ -1338,8 +616,8 @@ class XEP_0373: @staticmethod def build_signcrypt_element( - recipient_jids: Iterable[jid.JID], - ) -> Tuple[domish.Element, domish.Element]: + recipient_jids: set[jid.JID], + ) -> tuple[domish.Element, domish.Element]: """Build a ```` content element. @param recipient_jids: The intended recipients of this content element. Can be @@ -1355,8 +633,8 @@ class XEP_0373: @staticmethod def build_sign_element( - recipient_jids: Iterable[jid.JID], include_rpad: bool - ) -> Tuple[domish.Element, domish.Element]: + recipient_jids: set[jid.JID], include_rpad: bool + ) -> tuple[domish.Element, domish.Element]: """Build a ```` content element. @param recipient_jids: The intended recipients of this content element. Can be @@ -1375,7 +653,7 @@ class XEP_0373: @staticmethod def build_crypt_element( recipient_jids: Iterable[jid.JID], - ) -> Tuple[domish.Element, domish.Element]: + ) -> tuple[domish.Element, domish.Element]: """Build a ```` content element. @param recipient_jids: The intended recipients of this content element. Specifying @@ -1391,7 +669,7 @@ class XEP_0373: self, client: SatXMPPClient, content_elt: domish.Element, - recipient_jids: Set[jid.JID], + recipient_jids: set[jid.JID], ) -> domish.Element: """Build an ```` element. @@ -1412,7 +690,7 @@ class XEP_0373: ) ) - encryption_keys: Set[GPGPublicKey] = set() + encryption_keys: set[GPGPublicKeyBase] = set() for recipient_jid in recipient_jids: # import all keys of the recipient @@ -1445,7 +723,7 @@ class XEP_0373: openpgp_elt: domish.Element, element_name: Literal["signcrypt", "sign", "crypt"], sender_jid: jid.JID, - ) -> Tuple[domish.Element, datetime]: + ) -> tuple[domish.Element, datetime]: """Verify, decrypt and unpack an ```` element. @param client: The client to perform this operation with. @@ -1529,7 +807,7 @@ class XEP_0373: ): raise VerificationError( f"Recipient list in <{element_name}/> element does not list our (bare)" - f" JID." + " JID." ) time_elt = next(content_elt.elements(NS_OX, "time")) @@ -1541,7 +819,7 @@ class XEP_0373: return payload_elt, timestamp async def publish_public_key( - self, client: SatXMPPClient, public_key: GPGPublicKey + self, client: SatXMPPClient, public_key: GPGPublicKeyBase ) -> None: """Publish a public key. @@ -1582,7 +860,7 @@ class XEP_0373: async def import_all_public_keys( self, client: SatXMPPClient, entity_jid: jid.JID - ) -> Set[GPGPublicKey]: + ) -> set[GPGPublicKeyBase]: """import all public keys of a JID that have not been imported before. @param client: The client. @@ -1638,7 +916,7 @@ class XEP_0373: async def import_public_key( self, client: SatXMPPClient, jid: jid.JID, fingerprint: str - ) -> GPGPublicKey: + ) -> GPGPublicKeyBase: """import a public key. @param client: The client. @@ -1677,13 +955,13 @@ class XEP_0373: ) from e pubkey_elt = cast( - Optional[domish.Element], next(item_elt.elements(NS_OX, "pubkey"), None) + domish.Element | None, 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." + " element." ) try: @@ -1691,7 +969,7 @@ class XEP_0373: except xmlschema.XMLSchemaValidationError as e: raise exceptions.ParsingError( f"Publish-Subscribe item of JID {jid.userhost()} doesn't pass pubkey" - f" schema validation." + " schema validation." ) from e public_key = gpg_provider.import_public_key( @@ -1701,7 +979,7 @@ class XEP_0373: return public_key async def publish_public_keys_list( - self, client: SatXMPPClient, public_keys_list: Iterable[PublicKeyMetadata] + self, client: SatXMPPClient, public_keys_list: set[PublicKeyMetadata] ) -> None: """Publish/update the own public keys list. @@ -1747,7 +1025,7 @@ class XEP_0373: async def download_public_keys_list( self, client: SatXMPPClient, jid: jid.JID - ) -> Optional[Set[PublicKeyMetadata]]: + ) -> set[PublicKeyMetadata] | None: """Download the public keys list of a JID. @param client: The client. @@ -1775,7 +1053,7 @@ class XEP_0373: return None public_keys_list_elt = cast( - Optional[domish.Element], + domish.Element | None, next(item_elt.elements(NS_OX, "public-keys-list"), None), ) @@ -1787,7 +1065,7 @@ class XEP_0373: 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." + " list schema validation." ) from e return { @@ -1802,7 +1080,7 @@ class XEP_0373: async def __prepare_secret_key_synchronization( self, client: SatXMPPClient - ) -> Optional[domish.Element]: + ) -> domish.Element | None: """Prepare for secret key synchronization. Makes sure the relative protocols and protocol extensions are supported by the @@ -1827,8 +1105,8 @@ class XEP_0373: "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) + 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.") @@ -1915,7 +1193,7 @@ class XEP_0373: return backup_code - async def download_secret_keys(self, client: SatXMPPClient) -> Optional[bytes]: + async def download_secret_keys(self, client: SatXMPPClient) -> bytes | None: """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 @@ -1936,7 +1214,7 @@ class XEP_0373: return None secretkey_elt = cast( - Optional[domish.Element], next(item_elt.elements(NS_OX, "secretkey"), None) + domish.Element | None, next(item_elt.elements(NS_OX, "secretkey"), None) ) if secretkey_elt is None: @@ -1953,7 +1231,7 @@ class XEP_0373: def import_secret_keys( self, client: SatXMPPClient, ciphertext: bytes, backup_code: str - ) -> Set[GPGSecretKey]: + ) -> set[GPGSecretKeyBase]: """import previously downloaded secret keys. The downloading and importing steps are separate since a backup code is required @@ -1979,7 +1257,7 @@ class XEP_0373: @staticmethod def __get_joined_muc_users( client: SatXMPPClient, xep_0045: XEP_0045, room_jid: jid.JID - ) -> Set[jid.JID]: + ) -> set[jid.JID]: """ @param client: The client. @param xep_0045: A MUC plugin instance. @@ -1990,7 +1268,7 @@ class XEP_0373: """ # TODO: This should probably be a global helper somewhere - bare_jids: Set[jid.JID] = set() + bare_jids: set[jid.JID] = set() try: room = cast(muc.Room, xep_0045.get_room(client, room_jid)) @@ -1999,11 +1277,11 @@ class XEP_0373: "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) + for user in cast(dict[str, muc.User], room.roster).values(): + entity = cast(SatXMPPEntity | None, user.entity) if entity is None: raise exceptions.InternalError( - f"Participant list of MUC requested, but the entity information of" + "Participant list of MUC requested, but the entity information of" f" the participant {user} is not available." ) @@ -2012,7 +1290,7 @@ class XEP_0373: return bare_jids async def get_trust( - self, client: SatXMPPClient, public_key: GPGPublicKey, owner: jid.JID + self, client: SatXMPPClient, public_key: GPGPublicKeyBase, owner: jid.JID ) -> TrustLevel: """Query the trust level of a public key. @@ -2032,7 +1310,7 @@ class XEP_0373: async def set_trust( self, client: SatXMPPClient, - public_key: GPGPublicKey, + public_key: GPGPublicKeyBase, owner: jid.JID, trust_level: TrustLevel, ) -> None: @@ -2061,7 +1339,7 @@ class XEP_0373: if entity.resource: raise ValueError("A bare JID is expected.") - bare_jids: Set[jid.JID] + bare_jids: set[jid.JID] if self.__xep_0045 is not None and self.__xep_0045.is_joined_room(client, entity): bare_jids = self.__get_joined_muc_users(client, self.__xep_0045, entity) else: @@ -2075,8 +1353,9 @@ class XEP_0373: ) async def callback( - data: Any, profile: str # pylint: disable=unused-argument - ) -> Dict[Never, Never]: + 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. @@ -2088,7 +1367,7 @@ class XEP_0373: return {} data_form_result = cast( - Dict[str, str], xml_tools.xmlui_result_2_data_form_result(data) + dict[str, str], xml_tools.xmlui_result_2_data_form_result(data) ) for key, value in data_form_result.items(): if not key.startswith("trust_"): @@ -2108,7 +1387,9 @@ class XEP_0373: submit_id = self.host.register_callback(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 + 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 diff --git a/libervia/backend/plugins/plugin_xep_0373/exceptions.py b/libervia/backend/plugins/plugin_xep_0373/exceptions.py new file mode 100644 index 00000000..2f57efb8 --- /dev/null +++ b/libervia/backend/plugins/plugin_xep_0373/exceptions.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 + +# Libervia plugin for OpenPGP for XMPP +# Copyright (C) 2025-2026 Jérôme Poisosn (goffi@goffi.org) +# Copyright (C) 2022-2024 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 . + + +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. + """ diff --git a/libervia/backend/plugins/plugin_xep_0373/gnupg_backend.py b/libervia/backend/plugins/plugin_xep_0373/gnupg_backend.py new file mode 100644 index 00000000..79cb175a --- /dev/null +++ b/libervia/backend/plugins/plugin_xep_0373/gnupg_backend.py @@ -0,0 +1,375 @@ +#!/usr/bin/env python3 + +# Libervia plugin for OpenPGP for XMPP +# Copyright (C) 2025-2026 Jérôme Poisosn (goffi@goffi.org) +# Copyright (C) 2022-2024 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 . + +import os +import tempfile +from typing import Any + +import gnupg + +from .exceptions import ( + DecryptionFailed, + GPGProviderError, + InvalidPacket, + UnknownKey, + VerificationFailed, +) +from .models import GPGProviderBase, GPGPublicKeyBase, GPGSecretKeyBase + + +class GPGPublicKey(GPGPublicKeyBase): + """ + GPG public key implementation based on python-gnupg. + """ + + def __init__(self, key_obj: dict[str, Any]) -> None: + """ + @param key_obj: The python-gnupg key dictionary. + """ + + self._key_obj = key_obj + + @property + def fingerprint(self) -> str: + return self._key_obj["fingerprint"] + + @property + def key_obj(self) -> dict[str, Any]: + return self._key_obj + + +class GPGSecretKey(GPGSecretKeyBase): + """ + GPG secret key implementation based on python-gnupg. + """ + + def __init__(self, public_key: GPGPublicKey) -> None: + """ + @param public_key: The public key corresponding to this secret key. + """ + + self._public_key = public_key + + @property + def public_key(self) -> GPGPublicKey: + return self._public_key + + +class GPGProvider(GPGProviderBase): + """ + GPG provider implementation based on python-gnupg. + """ + + ALGORITHM = "ed25519" + + def __init__(self, home_dir: str | None = None) -> None: + """ + @param home_dir: Optional GPG home directory path to use for all operations. + """ + + self._home_dir = home_dir + self._gpg = gnupg.GPG(gnupghome=self._home_dir) + + def export_public_key(self, public_key: GPGPublicKeyBase) -> bytes: + assert isinstance(public_key, GPGPublicKey) + + try: + result = self._gpg.export_keys(public_key.fingerprint, armor=False) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result: + raise UnknownKey(f"Public key {public_key.fingerprint} not found.") + + return result + + def import_public_key(self, packet: bytes) -> GPGPublicKeyBase: + try: + result = self._gpg.import_keys(packet) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result or not hasattr(result, "fingerprints") or not result.fingerprints: + raise InvalidPacket("Data not considered for public key import") + + if len(result.fingerprints) != 1: + raise InvalidPacket( + "Public key packet does not contain exactly one public key (not" + " counting subkeys)." + ) + + fingerprint = result.fingerprints[0] + key_list = self._gpg.list_keys(keys=fingerprint) + if not key_list: + raise GPGProviderError("Newly imported public key not found") + + return GPGPublicKey(key_list[0]) + + def backup_secret_key(self, secret_key: GPGSecretKeyBase) -> bytes: + assert isinstance(secret_key, GPGSecretKey) + + try: + result = self._gpg.export_keys( + secret_key.public_key.fingerprint, secret=True, armor=False + ) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result: + raise UnknownKey(f"Secret key {secret_key.public_key.fingerprint} not found.") + + return result + + def restore_secret_keys(self, data: bytes) -> set[GPGSecretKeyBase]: + try: + result = self._gpg.import_keys(data) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result or not hasattr(result, "fingerprints") or not result.fingerprints: + raise InvalidPacket("Data not considered for secret key import") + + secret_keys = set() + for fingerprint in result.fingerprints: + key_list = self._gpg.list_keys(keys=fingerprint, secret=True) + if not key_list: + raise GPGProviderError("Newly imported secret key not found") + + secret_keys.add(GPGSecretKey(GPGPublicKey(key_list[0]))) + + return secret_keys + + def encrypt_symmetrically(self, plaintext: bytes, password: str) -> bytes: + try: + result = self._gpg.encrypt( + plaintext, None, passphrase=password, symmetric=True, armor=False + ) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result or not hasattr(result, "data") or not result.data: + raise DecryptionFailed("Encryption failed") + + return result.data + + def decrypt_symmetrically(self, ciphertext: bytes, password: str) -> bytes: + try: + result = self._gpg.decrypt(ciphertext, passphrase=password) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result or not hasattr(result, "data") or not result.data: + raise DecryptionFailed("Decryption failed") + + return result.data + + def sign(self, data: bytes, secret_keys: set[GPGSecretKeyBase]) -> bytes: + if not secret_keys: + raise GPGProviderError("No secret keys provided for signing") + + secret_key = next(iter(secret_keys)) + assert isinstance(secret_key, GPGSecretKey) + + try: + result = self._gpg.sign( + data, + keyid=secret_key.public_key.fingerprint, + detach=False, + binary=True, + ) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result or not hasattr(result, "data") or not result.data: + raise GPGProviderError("Signing failed") + + return result.data + + def sign_detached(self, data: bytes, secret_keys: set[GPGSecretKeyBase]) -> bytes: + if not secret_keys: + raise GPGProviderError("No secret keys provided for signing") + + secret_key = next(iter(secret_keys)) + assert isinstance(secret_key, GPGSecretKey) + + try: + result = self._gpg.sign( + data, + keyid=secret_key.public_key.fingerprint, + detach=True, + binary=True, + ) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result or not hasattr(result, "data") or not result.data: + raise GPGProviderError("Detached signing failed") + + return result.data + + def verify(self, signed_data: bytes, public_keys: set[GPGPublicKeyBase]) -> bytes: + try: + result = self._gpg.verify(signed_data) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + for public_key in public_keys: + assert isinstance(public_key, GPGPublicKey) + if result.pubkey_fingerprint == public_key.fingerprint: + break + else: + raise VerificationFailed("Data not signed by one of the expected public keys") + + return result.data + + def verify_detached( + self, data: bytes, signature: bytes, public_keys: set[GPGPublicKeyBase] + ) -> None: + try: + # FIXME: There is no method in gnupg to have signature and data in memory, we + # have to write signature to a file. This is probably due to the limitation + # of the gpg CLI. + with tempfile.NamedTemporaryFile(delete=False) as signature_file: + signature_file.write(signature) + signature_path = signature_file.name + try: + result = self._gpg.verify_data(signature_path, data) + finally: + os.unlink(signature_path) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + for public_key in public_keys: + assert isinstance(public_key, GPGPublicKey) + if result.pubkey_fingerprint == public_key.fingerprint: + break + else: + raise VerificationFailed("Data not signed by one of the expected public keys") + + def encrypt( + self, + plaintext: bytes, + public_keys: set[GPGPublicKeyBase], + signing_keys: set[GPGSecretKeyBase] | None = None, + ) -> bytes: + recipients = [] + for public_key in public_keys: + assert isinstance(public_key, GPGPublicKey) + recipients.append(public_key.fingerprint) + + kwargs: dict[str, Any] = {"armor": False} + if signing_keys: + secret_key = next(iter(signing_keys)) + assert isinstance(secret_key, GPGSecretKey) + kwargs["sign"] = secret_key.public_key.fingerprint + + try: + result = self._gpg.encrypt(plaintext, recipients, **kwargs) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result or not hasattr(result, "data") or not result.data: + raise GPGProviderError("Encryption failed") + + return result.data + + def decrypt( + self, + ciphertext: bytes, + secret_keys: set[GPGSecretKeyBase], + public_keys: set[GPGPublicKeyBase] | None = None, + ) -> bytes: + verify = public_keys is not None + + try: + result = self._gpg.decrypt(ciphertext) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result or not hasattr(result, "data"): + raise DecryptionFailed("Decryption failed") + + if verify: + valid_signature_found = False + for public_key in public_keys: + assert isinstance(public_key, GPGPublicKey) + + if result.pubkey_fingerprint == public_key.fingerprint: + valid_signature_found = True + break + + if not valid_signature_found: + raise VerificationFailed( + "Data not signed by one of the expected public keys" + ) + + return result.data + + def list_public_keys(self, user_id: str) -> set[GPGPublicKeyBase]: + try: + key_list = self._gpg.list_keys(keys=user_id) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + return {GPGPublicKey(key) for key in key_list if key.get("type") == "pub"} + + def list_secret_keys(self, user_id: str) -> set[GPGSecretKeyBase]: + try: + key_list = self._gpg.list_keys(keys=user_id, secret=True) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + return { + GPGSecretKey(GPGPublicKey(key)) + for key in key_list + if key.get("type") == "sec" + } + + def can_sign(self, public_key: GPGPublicKeyBase) -> bool: + assert isinstance(public_key, GPGPublicKey) + + key_obj = public_key.key_obj + return "s" in key_obj.get("cap", "").lower() + + def can_encrypt(self, public_key: GPGPublicKeyBase) -> bool: + assert isinstance(public_key, GPGPublicKey) + + key_obj = public_key.key_obj + return "c" in key_obj.get("cap", "").lower() + + def create_key(self, user_id: str) -> GPGSecretKeyBase: + key_input = self._gpg.gen_key_input( + key_type=self.ALGORITHM, + name_real=user_id, + no_protection=True, + expire_data=0, + ) + try: + result = self._gpg.gen_key(key_input) + except Exception as e: + raise GPGProviderError("Internal python-gnupg error") from e + + if not result or not hasattr(result, "fingerprint") or not result.fingerprint: + raise GPGProviderError("Key creation failed") + + key_list = self._gpg.list_keys(keys=result.fingerprint, secret=True) + if not key_list: + raise GPGProviderError("Newly created key not found") + + return GPGSecretKey(GPGPublicKey(key_list[0])) diff --git a/libervia/backend/plugins/plugin_xep_0373/gpg_backend.py b/libervia/backend/plugins/plugin_xep_0373/gpg_backend.py new file mode 100644 index 00000000..70c6d3a1 --- /dev/null +++ b/libervia/backend/plugins/plugin_xep_0373/gpg_backend.py @@ -0,0 +1,441 @@ +#!/usr/bin/env python3 + +# Libervia plugin for OpenPGP for XMPP +# Copyright (C) 2025-2026 Jérôme Poisosn (goffi@goffi.org) +# Copyright (C) 2022-2024 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 . + +from typing import Any + +import gpg + +from .exceptions import ( + DecryptionFailed, + GPGProviderError, + InvalidPacket, + UnknownKey, + VerificationFailed, +) +from .models import GPGProviderBase, GPGPublicKeyBase, GPGSecretKeyBase + + +class GPGPublicKey(GPGPublicKeyBase): + """ + 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 GPGSecretKey(GPGSecretKeyBase): + """ + GPG secret key implementation based on GnuPG Made Easy (GPGME). + """ + + def __init__(self, public_key: GPGPublicKey) -> None: + """ + @param public_key: The public key corresponding to this secret key. + """ + + self.__public_key = public_key + + @property + def public_key(self) -> GPGPublicKey: + return self.__public_key + + +class GPGProvider(GPGProviderBase): + """ + GPG provider implementation based on GnuPG Made Easy (GPGME). + """ + + ALGORITHM = "ed25519/cert,sign+cv25519/encr" + + def __init__(self, home_dir: str | None = 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: GPGPublicKeyBase) -> bytes: + assert isinstance(public_key, 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) -> GPGPublicKeyBase: + # 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 GPGPublicKey(key_obj) + + def backup_secret_key(self, secret_key: GPGSecretKeyBase) -> bytes: + assert isinstance(secret_key, 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[GPGSecretKeyBase]: + # 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(GPGSecretKey(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[GPGSecretKeyBase]) -> bytes: + signers = [] + for secret_key in secret_keys: + assert isinstance(secret_key, 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[GPGSecretKeyBase]) -> bytes: + signers = [] + for secret_key in secret_keys: + assert isinstance(secret_key, 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[GPGPublicKeyBase]) -> 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, 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[GPGPublicKeyBase] + ) -> 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, 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[GPGPublicKeyBase], + signing_keys: set[GPGSecretKeyBase] | None = None, + ) -> bytes: + recipients = [] + for public_key in public_keys: + assert isinstance(public_key, GPGPublicKey) + + recipients.append(public_key.key_obj) + + signers = [] + if signing_keys is not None: + for secret_key in signing_keys: + assert isinstance(secret_key, GPGSecretKey) + + signers.append(secret_key.public_key.key_obj) + + sign = signing_keys is not None + + kwargs = {"home_dir": self.__home_dir, "signers": signers} + + with gpg.Context(**kwargs) 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[GPGSecretKeyBase], + public_keys: set[GPGPublicKeyBase] | None = 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, 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[GPGPublicKeyBase]: + with gpg.Context(home_dir=self.__home_dir) as c: + try: + return { + 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[GPGSecretKeyBase]: + with gpg.Context(home_dir=self.__home_dir) as c: + try: + return { + GPGSecretKey(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: GPGPublicKeyBase) -> bool: + assert isinstance(public_key, GPGPublicKey) + + return any(subkey.can_sign for subkey in public_key.key_obj.subkeys) + + def can_encrypt(self, public_key: GPGPublicKeyBase) -> bool: + assert isinstance(public_key, GPGPublicKey) + + return any(subkey.can_encrypt for subkey in public_key.key_obj.subkeys) + + def create_key(self, user_id: str) -> GPGSecretKeyBase: + with gpg.Context(home_dir=self.__home_dir) as c: + try: + # A single ECC (primary) key with signing and encryption capabilities is + # not supported, a subkey is required for encryption. Create a primary key + # only capable of certifying subkeys + primary_create_result = c.create_key( + user_id, + algorithm=self.ALGORITHM, + expires=False, + certify=True, + force=True, + ) + + primary_key_obj = c.get_key(primary_create_result.fpr, secret=True) + + c.create_subkey( + primary_key_obj, algorithm=self.ALGORITHM, expires=False, sign=True + ) + + c.create_subkey( + primary_key_obj, + algorithm=self.ALGORITHM, + expires=False, + encrypt=True, + ) + except gpg.errors.GPGMEError as e: + raise GPGProviderError("Internal GPGME error") from e + except gpg.errors.KeyNotFound as e: + raise GPGProviderError("Newly created key not found") from e + + return GPGSecretKey(GPGPublicKey(primary_key_obj)) diff --git a/libervia/backend/plugins/plugin_xep_0373/models.py b/libervia/backend/plugins/plugin_xep_0373/models.py new file mode 100644 index 00000000..0fa7a15b --- /dev/null +++ b/libervia/backend/plugins/plugin_xep_0373/models.py @@ -0,0 +1,273 @@ +#!/usr/bin/env python3 + +# Libervia plugin for OpenPGP for XMPP +# Copyright (C) 2025-2026 Jérôme Poisosn (goffi@goffi.org) +# Copyright (C) 2022-2024 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 . + +from abc import ABC, abstractmethod + + +class GPGPublicKeyBase(ABC): + """ + Interface describing a GPG public key. + """ + + @property + @abstractmethod + def fingerprint(self) -> str: + """ + @return: The OpenPGP v4 fingerprint string of this public key. + """ + + +class GPGSecretKeyBase(ABC): + """ + Interface descibing a GPG secret key. + """ + + @property + @abstractmethod + def public_key(self) -> GPGPublicKeyBase: + """ + @return: The public key corresponding to this secret key. + """ + + +class GPGProviderBase(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: GPGPublicKeyBase) -> 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) -> GPGPublicKeyBase: + """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: GPGSecretKeyBase) -> 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[GPGSecretKeyBase]: + """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[GPGSecretKeyBase]) -> 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[GPGSecretKeyBase]) -> 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[GPGPublicKeyBase]) -> 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[GPGPublicKeyBase] + ) -> 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[GPGPublicKeyBase], + signing_keys: set[GPGSecretKeyBase] | None = 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[GPGSecretKeyBase], + public_keys: set[GPGPublicKeyBase] | None = 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[GPGPublicKeyBase]: + """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[GPGSecretKeyBase]: + """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: GPGPublicKeyBase) -> bool: + """ + @return: Whether the public key belongs to a key pair capable of signing. + """ + + @abstractmethod + def can_encrypt(self, public_key: GPGPublicKeyBase) -> bool: + """ + @return: Whether the public key belongs to a key pair capable of encryption. + """ + + @abstractmethod + def create_key(self, user_id: str) -> GPGSecretKeyBase: + """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. + """