hueso/gajim-plugins
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[openpgp] Port to Gtk4

Browse Source
This commit is contained in:
Philipp Hörist
2025-02-08 16:10:31 +01:00
parent 3a5816259c
commit c67e7f341b
19 changed files with 935 additions and 1111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
openpgp/modules/key_store.py
View File

@@ -14,8 +14,17 @@
# You should have received a copy of the GNU General Public License
# along with OpenPGP Gajim Plugin. If not, see <http://www.gnu.org/licenses/>.
import logging
from __future__ import annotations
import logging
from collections.abc import Iterator
from nbxmpp.protocol import JID
from nbxmpp.structs import PGPKeyMetadata
from openpgp.backend.base import BasePGPBackend
from openpgp.backend.sql import ContactRow
from openpgp.backend.sql import Storage
from openpgp.modules.util import Trust
log = logging.getLogger("gajim.p.openpgp.store")
@@ -26,45 +35,34 @@ class KeyData:
Holds all data related to a certain key
"""
def __init__(self, contact_data):
def __init__(
self,
contact_data: ContactData,
fingerprint: str,
active: bool,
trust: Trust,
timestamp: float,
):
self._contact_data = contact_data
self.fingerprint = None
self.active = False
self._trust = Trust.UNKNOWN
self.timestamp = None
self.fingerprint = fingerprint
self.active = active
self._trust = trust
self.timestamp = timestamp
self.comment = None
self.has_pubkey = False
@property
def trust(self):
def trust(self) -> Trust:
return self._trust
@trust.setter
def trust(self, value):
def trust(self, value: Trust) -> None:
if value not in (Trust.NOT_TRUSTED, Trust.UNKNOWN, Trust.BLIND, Trust.VERIFIED):
raise ValueError("Trust value not allowed: %s" % value)
self._trust = value
self._contact_data.set_trust(self.fingerprint, self._trust)
@classmethod
def from_key(cls, contact_data, key, trust):
keydata = cls(contact_data)
keydata.fingerprint = key.fingerprint
keydata.timestamp = key.date
keydata.active = True
keydata._trust = trust
return keydata
@classmethod
def from_row(cls, contact_data, row):
keydata = cls(contact_data)
keydata.fingerprint = row.fingerprint
keydata.timestamp = row.timestamp
keydata.comment = row.comment
keydata._trust = row.trust
keydata.active = row.active
return keydata
def delete(self):
self._contact_data.delete_key(self.fingerprint)
@@ -74,9 +72,9 @@ class ContactData:
Holds all data related to a contact
"""
def __init__(self, jid, storage, pgp):
def __init__(self, jid: JID, storage: Storage, pgp: BasePGPBackend) -> None:
self.jid = jid
self._key_store = {}
self._key_store: dict[str, KeyData] = {}
self._storage = storage
self._pgp = pgp
@@ -87,13 +85,13 @@ class ContactData:
return "xmpp:%s" % self.jid
@property
def default_trust(self):
def default_trust(self) -> Trust:
for key in self._key_store.values():
if key.trust in (Trust.NOT_TRUSTED, Trust.BLIND):
return Trust.UNKNOWN
return Trust.BLIND
def db_values(self):
def db_values(self) -> Iterator[tuple[JID, str, bool, Trust, float]]:
for key in self._key_store.values():
yield (
self.jid,
@@ -101,28 +99,39 @@ class ContactData:
key.active,
key.trust,
key.timestamp,
key.comment,
)
def add_from_key(self, key):
def add_from_key(self, key: PGPKeyMetadata) -> KeyData:
try:
keydata = self._key_store[key.fingerprint]
except KeyError:
keydata = KeyData.from_key(self, key, self.default_trust)
keydata = KeyData(
self,
key.fingerprint,
True,
self.default_trust,
key.date,
)
self._key_store[key.fingerprint] = keydata
log.info("Add from key: %s %s", self.jid, keydata.fingerprint)
return keydata
def add_from_db(self, row):
def add_from_db(self, row: ContactRow) -> KeyData:
try:
keydata = self._key_store[row.fingerprint]
except KeyError:
keydata = KeyData.from_row(self, row)
keydata = KeyData(
self,
row.fingerprint,
row.active,
row.trust,
row.timestamp,
)
self._key_store[row.fingerprint] = keydata
log.info("Add from row: %s %s", self.jid, row.fingerprint)
return keydata
def process_keylist(self, keylist):
def process_keylist(self, keylist: list[PGPKeyMetadata] | None) -> list[str]:
log.info("Process keylist: %s %s", self.jid, keylist)
if keylist is None:
@@ -131,8 +140,8 @@ class ContactData:
self._storage.save_contact(self.db_values())
return []
missing_pub_keys = []
fingerprints = set([key.fingerprint for key in keylist])
missing_pub_keys: list[str] = []
fingerprints = {key.fingerprint for key in keylist}
if fingerprints == self._key_store.keys():
log.info("No updates found")
for key in self._key_store.values():
@@ -156,7 +165,7 @@ class ContactData:
self._storage.save_contact(self.db_values())
return missing_pub_keys
def set_public_key(self, fingerprint):
def set_public_key(self, fingerprint: str) -> None:
try:
keydata = self._key_store[fingerprint]
except KeyError:
@@ -167,7 +176,7 @@ class ContactData:
keydata.has_pubkey = True
log.info("Set public key: %s %s", self.jid, fingerprint)
def get_keys(self, only_trusted=True):
def get_keys(self, only_trusted: bool = True) -> list[KeyData]:
keys = list(self._key_store.values())
if not only_trusted:
return keys
@@ -175,13 +184,13 @@ class ContactData:
k for k in keys if k.active and k.trust in (Trust.VERIFIED, Trust.BLIND)
]
def get_key(self, fingerprint):
def get_key(self, fingerprint: str) -> KeyData | None:
return self._key_store.get(fingerprint, None)
def set_trust(self, fingerprint, trust):
def set_trust(self, fingerprint: str, trust: Trust) -> None:
self._storage.set_trust(self.jid, fingerprint, trust)
def delete_key(self, fingerprint):
def delete_key(self, fingerprint: str) -> None:
self._storage.delete_key(self.jid, fingerprint)
self._pgp.delete_key(fingerprint)
del self._key_store[fingerprint]
@@ -192,8 +201,8 @@ class PGPContacts:
Holds all contacts available for PGP encryption
"""
def __init__(self, pgp, storage):
self._contacts = {}
def __init__(self, pgp: BasePGPBackend, storage: Storage) -> None:
self._contacts: dict[JID, ContactData] = {}
self._storage = storage
self._pgp = pgp
self._load_from_storage()
@@ -204,14 +213,12 @@ class PGPContacts:
keyring = self._pgp.get_keys()
for key in keyring:
log.info("Found: %s %s", key.jid, key.fingerprint)
assert key.jid is not None
self.set_public_key(key.jid, key.fingerprint)
def _load_from_storage(self):
log.info("Load contacts from storage")
rows = self._storage.load_contacts()
if rows is None:
return
for row in rows:
log.info("Found: %s %s", row.jid, row.fingerprint)
try:
@@ -223,7 +230,9 @@ class PGPContacts:
else:
contact_data.add_from_db(row)
def process_keylist(self, jid, keylist):
def process_keylist(
self, jid: JID, keylist: list[PGPKeyMetadata] | None
) -> list[str]:
try:
contact_data = self._contacts[jid]
except KeyError:
@@ -235,7 +244,7 @@ class PGPContacts:
return missing_pub_keys
def set_public_key(self, jid, fingerprint):
def set_public_key(self, jid: JID, fingerprint: str) -> None:
try:
contact_data = self._contacts[jid]
except KeyError:
@@ -243,14 +252,14 @@ class PGPContacts:
else:
contact_data.set_public_key(fingerprint)
def get_keys(self, jid, only_trusted=True):
def get_keys(self, jid: JID, only_trusted: bool = True) -> list[KeyData]:
try:
contact_data = self._contacts[jid]
return contact_data.get_keys(only_trusted=only_trusted)
except KeyError:
return []
def get_trust(self, jid, fingerprint):
def get_trust(self, jid: JID, fingerprint: str) -> Trust:
contact_data = self._contacts.get(jid, None)
if contact_data is None:
return Trust.UNKNOWN

115
openpgp/modules/openpgp.py
View File

@@ -14,37 +14,47 @@
# You should have received a copy of the GNU General Public License
# along with OpenPGP Gajim Plugin. If not, see <http://www.gnu.org/licenses/>.
from typing import Any
from typing import cast
import logging
import sys
import time
from collections.abc import Callable
from pathlib import Path
from nbxmpp import Node
from nbxmpp import StanzaMalformed
from nbxmpp.client import Client as nbxmppClient
from nbxmpp.errors import MalformedStanzaError
from nbxmpp.errors import StanzaError
from nbxmpp.exceptions import StanzaDecrypted
from nbxmpp.modules.openpgp import create_message_stanza
from nbxmpp.modules.openpgp import create_signcrypt_node
from nbxmpp.modules.openpgp import parse_signcrypt
from nbxmpp.modules.openpgp import PGPKeyMetadata
from nbxmpp.namespaces import Namespace
from nbxmpp.protocol import JID
from nbxmpp.protocol import Message
from nbxmpp.structs import EncryptionData
from nbxmpp.structs import MessageProperties
from nbxmpp.structs import PGPKeyMetadata
from nbxmpp.structs import PGPPublicKey
from nbxmpp.structs import StanzaHandler
from nbxmpp.task import Task
from gajim.common import app
from gajim.common import configpaths
from gajim.common.client import Client
from gajim.common.events import MessageNotSent
from gajim.common.modules.base import BaseModule
from gajim.common.modules.util import event_node
from gajim.common.structs import OutgoingMessage
from openpgp.backend.sql import Storage
from openpgp.modules.key_store import KeyData
from openpgp.modules.key_store import PGPContacts
from openpgp.modules.util import DecryptionFailed
from openpgp.modules.util import ENCRYPTION_NAME
from openpgp.modules.util import Key
from openpgp.modules.util import NOT_ENCRYPTED_TAGS
from openpgp.modules.util import prepare_stanza
from openpgp.modules.util import Trust
@@ -52,7 +62,7 @@ from openpgp.modules.util import Trust
if sys.platform == "win32":
from openpgp.backend.pygpg import PythonGnuPG as PGPBackend
else:
from openpgp.backend.gpgme import GPGME as PGPBackend
from openpgp.backend.gpgme import GPGMe as PGPBackend
log = logging.getLogger("gajim.p.openpgp")
@@ -75,7 +85,7 @@ class OpenPGP(BaseModule):
"request_secret_key",
]
def __init__(self, client):
def __init__(self, client: Client):
BaseModule.__init__(self, client)
self.handlers = [
@@ -103,67 +113,88 @@ class OpenPGP(BaseModule):
log.info("Own Fingerprint at start: %s", self._fingerprint)
@property
def secret_key_available(self):
def secret_key_available(self) -> bool:
return self._fingerprint is not None
def get_own_key_details(self):
def get_own_key_details(self) -> tuple[str | None, int | None]:
self._fingerprint, self._date = self._pgp.get_own_key_details()
return self._fingerprint, self._date
def generate_key(self):
def generate_key(self) -> None:
self._pgp.generate_key()
def set_public_key(self):
def set_public_key(self) -> None:
log.info("%s => Publish public key", self._account)
assert self._fingerprint is not None
assert self._date is not None
key = self._pgp.export_key(self._fingerprint)
assert key is not None
self._nbxmpp("OpenPGP").set_public_key(key, self._fingerprint, self._date)
def request_public_key(self, jid, fingerprint):
def request_public_key(self, jid: JID, fingerprint: str) -> None:
log.info("%s => Request public key %s - %s", self._account, fingerprint, jid)
self._nbxmpp("OpenPGP").request_public_key(
jid, fingerprint, callback=self._public_key_received, user_data=fingerprint
)
def _public_key_received(self, task):
def _public_key_received(self, task: Task) -> None:
fingerprint = task.get_user_data()
try:
result = task.finish()
result = cast(PGPPublicKey | None, task.finish())
except (StanzaError, MalformedStanzaError) as error:
log.error("%s => Public Key not found: %s", self._account, error)
return
if result is None:
log.error("%s => Public Key Node is empty", self._account)
return
imported_key = self._pgp.import_key(result.key, result.jid)
if imported_key is not None:
self._contacts.set_public_key(result.jid, fingerprint)
def set_keylist(self, keylist=None):
def set_keylist(self, keylist: list[PGPKeyMetadata] | None = None) -> None:
if keylist is None:
keylist = [PGPKeyMetadata(None, self._fingerprint, self._date)]
assert self._fingerprint is not None
assert self._date is not None
keylist = [PGPKeyMetadata(self.own_jid, self._fingerprint, self._date)]
log.info("%s => Publish keylist", self._account)
self._nbxmpp("OpenPGP").set_keylist(keylist)
@event_node(Namespace.OPENPGP_PK)
def _keylist_notification_received(self, _con, _stanza, properties):
def _keylist_notification_received(
self, _client: nbxmppClient, _stanza: Node, properties: MessageProperties
) -> None:
assert properties.pubsub_event is not None
if properties.pubsub_event.retracted:
return
keylist = properties.pubsub_event.data or []
assert properties.jid is not None
keylist: list[PGPKeyMetadata] = []
if properties.pubsub_event.data:
keylist = cast(list[PGPKeyMetadata], properties.pubsub_event.data)
self._process_keylist(keylist, properties.jid)
def request_keylist(self, jid=None):
def request_keylist(self, jid: JID | None = None) -> None:
if jid is None:
jid = self.own_jid
log.info("%s => Fetch keylist %s", self._account, jid)
self._nbxmpp("OpenPGP").request_keylist(
jid, callback=self._keylist_received, user_data=jid
)
def _keylist_received(self, task):
jid = task.get_user_data()
def _keylist_received(self, task: Task) -> None:
jid = cast(JID, task.get_user_data())
try:
keylist = task.finish()
keylist = cast(list[PGPKeyMetadata] | None, task.finish())
except (StanzaError, MalformedStanzaError) as error:
log.error("%s => Keylist query failed: %s", self._account, error)
if self.own_jid.bare_match(jid) and self._fingerprint is not None:
@@ -173,7 +204,9 @@ class OpenPGP(BaseModule):
log.info("Keylist received from %s", jid)
self._process_keylist(keylist, jid)
def _process_keylist(self, keylist, from_jid):
def _process_keylist(
self, keylist: list[PGPKeyMetadata] | None, from_jid: JID
) -> None:
if not keylist:
log.warning("%s => Empty keylist received from %s", self._account, from_jid)
self._contacts.process_keylist(self.own_jid, keylist)
@@ -185,14 +218,19 @@ class OpenPGP(BaseModule):
log.info("Received own keylist")
for key in keylist:
log.info(key.fingerprint)
for key in keylist:
# Check if own fingerprint is published
if key.fingerprint == self._fingerprint:
log.info("Own key found in keys list")
return
log.info("Own key not published")
if self._fingerprint is not None:
keylist.append(Key(self._fingerprint, self._date))
assert self._date is not None
keylist.append(
PGPKeyMetadata(self.own_jid, self._fingerprint, self._date)
)
self.set_keylist(keylist)
return
@@ -204,10 +242,14 @@ class OpenPGP(BaseModule):
for fingerprint in missing_pub_keys:
self.request_public_key(from_jid, fingerprint)
def decrypt_message(self, _con, stanza, properties: MessageProperties):
def decrypt_message(
self, _client: nbxmppClient, stanza: Message, properties: MessageProperties
) -> None:
if not properties.is_openpgp:
return
assert properties.openpgp is not None
remote_jid = properties.remote_jid
assert remote_jid is not None
@@ -249,7 +291,9 @@ class OpenPGP(BaseModule):
raise StanzaDecrypted
def encrypt_message(self, message: OutgoingMessage, callback):
def encrypt_message(
self, message: OutgoingMessage, callback: Callable[[OutgoingMessage], None]
) -> None:
remote_jid = message.contact.jid
keys = self._contacts.get_keys(remote_jid)
@@ -257,12 +301,17 @@ class OpenPGP(BaseModule):
log.error("Dropping stanza to %s, because we have no key", remote_jid)
return
assert self._fingerprint is not None
keys += self._contacts.get_keys(self.own_jid)
keys += [Key(self._fingerprint, None)]
keys += [
KeyData(None, self._fingerprint, True, Trust.VERIFIED, 0) # pyright: ignore
]
payload = create_signcrypt_node(
message.get_stanza(), [remote_jid], NOT_ENCRYPTED_TAGS
)
payload = str(payload).encode("utf8")
encrypted_payload, error = self._pgp.encrypt(payload, keys)
if error:
@@ -279,6 +328,8 @@ class OpenPGP(BaseModule):
)
return
assert encrypted_payload is not None
create_message_stanza(
message.get_stanza(), encrypted_payload, bool(message.get_text())
)
@@ -292,7 +343,7 @@ class OpenPGP(BaseModule):
callback(message)
@staticmethod
def print_msg_to_log(stanza):
def print_msg_to_log(stanza: Node) -> None:
"""Prints a stanza in a fancy way to the log"""
log.debug("-" * 15)
stanzastr = "\n" + stanza.__str__(fancy=True)
@@ -300,19 +351,21 @@ class OpenPGP(BaseModule):
log.debug(stanzastr)
log.debug("-" * 15)
def get_keys(self, jid=None, only_trusted=True):
def get_keys(
self, jid: JID | None = None, only_trusted: bool = True
) -> list[KeyData]:
if jid is None:
jid = self.own_jid
return self._contacts.get_keys(jid, only_trusted=only_trusted)
def clear_fingerprints(self):
def clear_fingerprints(self) -> None:
self.set_keylist()
def cleanup(self):
def cleanup(self) -> None:
self._storage.cleanup()
self._pgp = None
self._contacts = None
del self._pgp
del self._contacts
def get_instance(*args, **kwargs):
def get_instance(*args: Any, **kwargs: Any) -> tuple[Any, str]:
return OpenPGP(*args, **kwargs), "OpenPGP"

22
openpgp/modules/util.py
View File

@@ -14,9 +14,9 @@
# You should have received a copy of the GNU General Public License
# along with OpenPGP Gajim Plugin. If not, see <http://www.gnu.org/licenses/>.
from collections import namedtuple
from enum import IntEnum
from nbxmpp import Node
from nbxmpp.namespaces import Namespace
ENCRYPTION_NAME = "OpenPGP"
@@ -27,11 +27,9 @@ NOT_ENCRYPTED_TAGS = [
("no-copy", Namespace.HINTS),
("no-permanent-store", Namespace.HINTS),
("origin-id", Namespace.SID),
("thread", None),
("thread", ""),
]
Key = namedtuple("Key", "fingerprint date")
class Trust(IntEnum):
NOT_TRUSTED = 0
@@ -40,19 +38,23 @@ class Trust(IntEnum):
VERIFIED = 3
def prepare_stanza(stanza, payload):
def prepare_stanza(stanza: Node, payload: list[Node | str]) -> None:
delete_nodes(stanza, "openpgp", Namespace.OPENPGP)
delete_nodes(stanza, "body")
nodes = [(node.getName(), node.getNamespace()) for node in payload]
for name, namespace in nodes:
delete_nodes(stanza, name, namespace)
nodes: list[Node] = []
for node in payload:
if isinstance(node, str):
continue
name, namespace = node.getName(), node.getNamespace()
delete_nodes(stanza, name, namespace)
nodes.append(node)
for node in nodes:
stanza.addChild(node=node)
def delete_nodes(stanza, name, namespace=None):
def delete_nodes(stanza: Node, name: str, namespace: str | None = None) -> None:
attrs = None
if namespace is not None:
attrs = {"xmlns": Namespace.OPENPGP}