plugin XEP-0373: refactor and implement python-gnupg backend:

- split the plugin in several classes, with models, exceptions, and backends separated.
- add a backend for `python-gnupg`, as it can be pip installed without having to use
  `--system-site-packages` in the virtual environment.
- use modern syntax for type hints.

rel 475
This commit is contained in:
Goffi 2026-02-23 12:56:42 +01:00
commit 1745f1e520
7 changed files with 1252 additions and 819 deletions

View file

@ -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

View file

@ -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__)

View file

@ -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 <http://www.gnu.org/licenses/>.
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.
"""

View file

@ -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 <http://www.gnu.org/licenses/>.
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]))

View file

@ -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 <http://www.gnu.org/licenses/>.
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))

View file

@ -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 <http://www.gnu.org/licenses/>.
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.
"""