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

[pgp] Port to Gtk4 and add type annotations

Browse Source
This commit is contained in:
Philipp Hörist
2025-01-28 23:22:05 +01:00
parent 6fe2da67c3
commit 3a5816259c
11 changed files with 1323 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pgp/backend/python_gnupg.py
+36 -27
View File
@@ -20,6 +20,8 @@
# You should have received a copy of the GNU General Public License
# along with PGP Gajim Plugin. If not, see <http://www.gnu.org/licenses/>.
from typing import Any
import logging
import os
from functools import lru_cache
@@ -37,15 +39,14 @@ if logger.getEffectiveLevel() == logging.DEBUG:
logger.setLevel(logging.DEBUG)
class PGP(gnupg.GPG, metaclass=Singleton):
def __init__(self, binary, encoding=None):
super().__init__(gpgbinary=binary, use_agent=True)
class PGP(metaclass=Singleton):
def __init__(self) -> None:
self._pgp = gnupg.GPG(use_agent=True)
self._pgp.decode_errors = "replace"
if encoding is not None:
self.encoding = encoding
self.decode_errors = "replace"
def encrypt(self, payload, recipients, always_trust=False):
def encrypt(
self, data: str, recipients: list[str], always_trust: bool = False
) -> tuple[str, str]:
if not always_trust:
# check that we'll be able to encrypt
result = self.get_key(recipients[0])
@@ -53,8 +54,8 @@ class PGP(gnupg.GPG, metaclass=Singleton):
if key["trust"] not in ("f", "u"):
return "", "NOT_TRUSTED " + key["keyid"][-8:]
result = super().encrypt(
payload.encode("utf8"), recipients, always_trust=always_trust
result = self._pgp.encrypt(
data.encode("utf8"), recipients, always_trust=always_trust
)
if result.ok:
@@ -64,23 +65,25 @@ class PGP(gnupg.GPG, metaclass=Singleton):
return self._strip_header_footer(str(result)), error
def decrypt(self, payload):
data = self._add_header_footer(payload, "MESSAGE")
result = super().decrypt(data.encode("utf8"))
def encrypt_file(self, file: Any, recipients: list[str]) -> gnupg.Crypt:
return self._pgp.encrypt_file(file, recipients)
return result.data.decode("utf8")
def decrypt(self, payload: str) -> str:
data = self._add_header_footer(payload, "MESSAGE")
result = self._pgp.decrypt(data.encode("utf8"))
return str(result)
@lru_cache(maxsize=8)
def sign(self, payload, key_id):
def sign(self, payload: str | None, key_id: str) -> str:
if payload is None:
payload = ""
result = super().sign(payload.encode("utf8"), keyid=key_id, detach=True)
result = self._pgp.sign(payload.encode("utf8"), keyid=key_id, detach=True)
if result.fingerprint:
if result:
return self._strip_header_footer(str(result))
raise SignError(result.status)
def verify(self, payload, signed):
def verify(self, payload: str | None, signed: str) -> str | None:
# Hash algorithm is not transferred in the signed
# presence stanza so try all algorithms.
# Text name for hash algorithms from RFC 4880 - section 9.4
@@ -99,24 +102,30 @@ class PGP(gnupg.GPG, metaclass=Singleton):
self._add_header_footer(signed, "SIGNATURE"),
]
)
result = super().verify(data.encode("utf8"))
if result.valid:
result = self._pgp.verify(data.encode("utf8"))
if result:
return result.fingerprint
def get_key(self, key_id):
return super().list_keys(keys=[key_id])
def get_key(self, key_id: str) -> gnupg.ListKeys:
return self._pgp.list_keys(keys=[key_id])
def get_keys(self, secret=False):
keys = {}
result = super().list_keys(secret=secret)
def get_keys(self, secret: bool = False) -> dict[str, str]:
keys: dict[str, str] = {}
result = self._pgp.list_keys(secret=secret)
for key in result:
# Take first not empty uid
keys[key["fingerprint"]] = next(uid for uid in key["uids"] if uid)
return keys
def list_keys(
self, secret: bool = False, keys: list[str] | None = None, sigs: bool = False
) -> list[str]:
res = self._pgp.list_keys(secret, keys, sigs)
return res.fingerprints
@staticmethod
def _strip_header_footer(data):
def _strip_header_footer(data: str) -> str:
"""
Remove header and footer from data
"""
@@ -137,7 +146,7 @@ class PGP(gnupg.GPG, metaclass=Singleton):
return line
@staticmethod
def _add_header_footer(data, type_):
def _add_header_footer(data: str, type_: str) -> str:
"""
Add header and footer from data
"""
pgp/backend/store.py
+35 -21
View File
@@ -14,9 +14,17 @@
# You should have received a copy of the GNU General Public License
# along with PGP Gajim Plugin. If not, see <http://www.gnu.org/licenses/>.
from __future__ import annotations
from typing import Any
import json
import logging
from collections.abc import Callable
from pathlib import Path
from nbxmpp import JID
from gajim.common import app
from gajim.common import configpaths
@@ -28,7 +36,13 @@ class KeyResolveError(Exception):
class KeyStore:
def __init__(self, account, own_jid, log, list_keys_func):
def __init__(
self,
account: str,
own_jid: JID,
log: logging.LoggerAdapter[Any],
list_keys_func: Callable[..., list[str]],
) -> None:
self._list_keys_func = list_keys_func
self._log = log
self._account = account
@@ -66,15 +80,15 @@ class KeyStore:
self._save_store()
@staticmethod
def _empty_store():
def _empty_store() -> dict[str, Any]:
return {
"_version": CURRENT_STORE_VERSION,
"own_key_data": None,
"contact_key_data": {},
}
def _migrate_v1_store(self):
keys = {}
def _migrate_v1_store(self) -> None:
keys: dict[str, str] = {}
attached_keys = app.settings.get_account_setting(
self._account, "attached_gpg_keys"
)
@@ -98,7 +112,7 @@ class KeyStore:
)
self._log.info("Migration from store v1 was successful")
def _migrate_v2_store(self):
def _migrate_v2_store(self) -> None:
own_key_data = self.get_own_key_data()
if own_key_data is not None:
own_key_id, own_key_user = (
@@ -111,7 +125,7 @@ class KeyStore:
except KeyResolveError:
self._set_own_key_data_nosync(None)
prune_list = []
prune_list: list[str] = []
for dict_key, key_data in self._store["contact_key_data"].items():
try:
@@ -125,20 +139,18 @@ class KeyStore:
self._store["_version"] = CURRENT_STORE_VERSION
self._log.info("Migration from store v2 was successful")
def _save_store(self):
def _save_store(self) -> None:
with self._store_path.open("w") as file:
json.dump(self._store, file)
def _get_dict_key(self, jid):
def _get_dict_key(self, jid: str) -> str:
return "%s-%s" % (self._account, jid)
def _resolve_short_id(self, short_id, has_secret=False):
candidates = self._list_keys_func(
secret=has_secret, keys=(short_id,)
).fingerprints
if len(candidates) == 1:
return candidates[0]
elif len(candidates) > 1:
def _resolve_short_id(self, short_id: str, has_secret: bool = False) -> str:
fingerprints = self._list_keys_func(secret=has_secret, keys=[short_id])
if len(fingerprints) == 1:
return fingerprints[0]
elif len(fingerprints) > 1:
self._log.critical(
"Key collision during migration. Key ID is %s. Removing binding...",
repr(short_id),
@@ -150,11 +162,11 @@ class KeyStore:
)
raise KeyResolveError
def set_own_key_data(self, key_data):
def set_own_key_data(self, key_data: tuple[str, str] | None) -> None:
self._set_own_key_data_nosync(key_data)
self._save_store()
def _set_own_key_data_nosync(self, key_data):
def _set_own_key_data_nosync(self, key_data: tuple[str, str] | None) -> None:
if key_data is None:
self._store["own_key_data"] = None
else:
@@ -163,19 +175,21 @@ class KeyStore:
"key_user": key_data[1],
}
def get_own_key_data(self):
def get_own_key_data(self) -> dict[str, str] | None:
return self._store["own_key_data"]
def get_contact_key_data(self, jid):
def get_contact_key_data(self, jid: str) -> dict[str, str] | None:
key_ids = self._store["contact_key_data"]
dict_key = self._get_dict_key(jid)
return key_ids.get(dict_key)
def set_contact_key_data(self, jid, key_data):
def set_contact_key_data(self, jid: str, key_data: tuple[str, str] | None) -> None:
self._set_contact_key_data_nosync(jid, key_data)
self._save_store()
def _set_contact_key_data_nosync(self, jid, key_data):
def _set_contact_key_data_nosync(
self, jid: str, key_data: tuple[str, str] | None
) -> None:
key_ids = self._store["contact_key_data"]
dict_key = self._get_dict_key(jid)
if key_data is None: