Remove openpgp plugin
It has been integrated into Gajim 2.4.7
This commit is contained in:
@@ -1,57 +0,0 @@
|
|||||||
id: org.gajim.Gajim.Devel.Plugin.openpgp
|
|
||||||
runtime: org.gajim.Gajim.Devel
|
|
||||||
sdk: org.gnome.Sdk//50
|
|
||||||
build-extension: true
|
|
||||||
separate-locales: false
|
|
||||||
appstream-compose: false
|
|
||||||
|
|
||||||
build-options:
|
|
||||||
env:
|
|
||||||
PIP_TARGET: /app/plugins/openpgp/site-packages
|
|
||||||
PIP_DISABLE_PIP_VERSION_CHECK: "1"
|
|
||||||
|
|
||||||
modules:
|
|
||||||
- name: swig
|
|
||||||
cleanup:
|
|
||||||
- "*"
|
|
||||||
build-options:
|
|
||||||
prefix: ${FLATPAK_DEST}
|
|
||||||
sources:
|
|
||||||
- type: archive
|
|
||||||
url: https://downloads.sourceforge.net/swig/swig-4.4.1.tar.gz
|
|
||||||
sha256: 40162a706c56f7592d08fd52ef5511cb7ac191f3593cf07306a0a554c6281fcf
|
|
||||||
|
|
||||||
- name: python3-gpgme
|
|
||||||
buildsystem: simple
|
|
||||||
build-options:
|
|
||||||
env:
|
|
||||||
SWIG_LIB: "/app/plugins/openpgp/share/swig/4.4.1"
|
|
||||||
prepend-path: "/app/plugins/openpgp/bin:"
|
|
||||||
build-commands:
|
|
||||||
- sed -e 's|, \"swig\"||' -i pyproject.toml
|
|
||||||
- autoreconf -fiv
|
|
||||||
- ./configure # Required to set the version number
|
|
||||||
- mv src gpg
|
|
||||||
- pip3 install --no-build-isolation .
|
|
||||||
sources:
|
|
||||||
- type: git
|
|
||||||
url: https://dev.gnupg.org/source/gpgmepy.git
|
|
||||||
tag: gpgmepy-2.0.0
|
|
||||||
commit: 22e1cd5fdea1f62690fba55cf616bde97c1ac5d3
|
|
||||||
|
|
||||||
- name: gajim-openpgp
|
|
||||||
buildsystem: simple
|
|
||||||
build-commands:
|
|
||||||
- 'cp -R -t ${FLATPAK_DEST} *'
|
|
||||||
post-install:
|
|
||||||
- install -Dm644 --target-directory=${FLATPAK_DEST}/share/metainfo org.gajim.Gajim.Plugin.openpgp.metainfo.xml
|
|
||||||
- appstreamcli compose --components=${FLATPAK_ID} --prefix=/ --origin=${FLATPAK_ID}
|
|
||||||
--result-root=${FLATPAK_DEST} --data-dir=${FLATPAK_DEST}/share/app-info/xmls ${FLATPAK_DEST}
|
|
||||||
sources:
|
|
||||||
- type: archive
|
|
||||||
strip-components: 0
|
|
||||||
url: https://ftp.gajim.org/plugins/master/openpgp/openpgp_1.8.1.zip
|
|
||||||
sha256: 3a166ce42f84afc591200759eaa3e26f7b0e74b10bbf8a0c25813039be28b1d5
|
|
||||||
- type: shell
|
|
||||||
commands:
|
|
||||||
- sed -i "s+org.gajim.Gajim+org.gajim.Gajim.Devel+" org.gajim.Gajim.Plugin.*.metainfo.xml
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
from .pgpplugin import OpenPGPPlugin # pyright: ignore # noqa: F401
|
|
||||||
@@ -1,95 +0,0 @@
|
|||||||
# Copyright (C) 2025 Philipp Hörist <philipp AT hoerist.com>
|
|
||||||
#
|
|
||||||
# This file is part of the OpenPGP Gajim Plugin.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation; version 3 only.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin 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 General Public License for more details.
|
|
||||||
#
|
|
||||||
# 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 __future__ import annotations
|
|
||||||
|
|
||||||
from typing import TYPE_CHECKING
|
|
||||||
|
|
||||||
from collections.abc import Sequence
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from nbxmpp.protocol import JID
|
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
|
||||||
from openpgp.modules.key_store import KeyData
|
|
||||||
|
|
||||||
|
|
||||||
class BaseKeyringItem:
|
|
||||||
def __init__(self) -> None:
|
|
||||||
self._uid = self._get_uid()
|
|
||||||
|
|
||||||
@property
|
|
||||||
def is_xmpp_key(self) -> bool:
|
|
||||||
try:
|
|
||||||
return self.jid is not None
|
|
||||||
except Exception:
|
|
||||||
return False
|
|
||||||
|
|
||||||
def is_valid(self, jid: JID) -> bool:
|
|
||||||
if not self.is_xmpp_key:
|
|
||||||
return False
|
|
||||||
return jid == self.jid
|
|
||||||
|
|
||||||
def _get_uid(self) -> str | None:
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
@property
|
|
||||||
def fingerprint(self) -> str:
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
@property
|
|
||||||
def uid(self):
|
|
||||||
if self._uid is not None:
|
|
||||||
return self._uid
|
|
||||||
|
|
||||||
@property
|
|
||||||
def jid(self) -> JID | None:
|
|
||||||
if self._uid is not None:
|
|
||||||
return JID.from_string(self._uid)
|
|
||||||
|
|
||||||
def __hash__(self):
|
|
||||||
return hash(self.fingerprint)
|
|
||||||
|
|
||||||
|
|
||||||
class BasePGPBackend:
|
|
||||||
def __init__(self, jid: str, gnupghome: Path) -> None:
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
def generate_key(self) -> None:
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
def encrypt(
|
|
||||||
self, payload: bytes, keys: list[KeyData]
|
|
||||||
) -> tuple[bytes | None, str | None]:
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
def decrypt(self, payload: bytes) -> tuple[str, str]:
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
def get_keys(self) -> Sequence[BaseKeyringItem]:
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
def import_key(self, data: bytes, jid: JID) -> BaseKeyringItem | None:
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
def get_own_key_details(self) -> tuple[str | None, int | None]:
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
def export_key(self, fingerprint: str) -> bytes | None:
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
def delete_key(self, fingerprint: str) -> None:
|
|
||||||
raise NotImplementedError
|
|
||||||
@@ -1,206 +0,0 @@
|
|||||||
# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>
|
|
||||||
#
|
|
||||||
# This file is part of the OpenPGP Gajim Plugin.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation; version 3 only.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin 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 General Public License for more details.
|
|
||||||
#
|
|
||||||
# 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
|
|
||||||
from collections.abc import Iterator
|
|
||||||
from collections.abc import Sequence
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
import gpg
|
|
||||||
from gpg.errors import KeyNotFound
|
|
||||||
from gpg.results import GenkeyResult
|
|
||||||
from gpg.results import ImportResult
|
|
||||||
from nbxmpp.protocol import JID
|
|
||||||
|
|
||||||
from openpgp.backend.base import BaseKeyringItem
|
|
||||||
from openpgp.backend.base import BasePGPBackend
|
|
||||||
from openpgp.backend.gpgme_types import Key
|
|
||||||
from openpgp.backend.util import parse_uid
|
|
||||||
from openpgp.modules.key_store import KeyData
|
|
||||||
from openpgp.modules.util import DecryptionFailed
|
|
||||||
|
|
||||||
log = logging.getLogger("gajim.p.openpgp.gpgme")
|
|
||||||
|
|
||||||
|
|
||||||
class KeyringItem(BaseKeyringItem):
|
|
||||||
def __init__(self, key: Key) -> None:
|
|
||||||
self._key = key
|
|
||||||
BaseKeyringItem.__init__(self)
|
|
||||||
|
|
||||||
def _get_uid(self) -> str | None:
|
|
||||||
for uid in self._key.uids:
|
|
||||||
try:
|
|
||||||
return parse_uid(uid.uid)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
@property
|
|
||||||
def fingerprint(self) -> str:
|
|
||||||
return self._key.fpr
|
|
||||||
|
|
||||||
|
|
||||||
class GPGMe(BasePGPBackend):
|
|
||||||
def __init__(self, jid: str, gnuhome: Path) -> None:
|
|
||||||
self._jid = jid
|
|
||||||
self._home_dir = str(gnuhome)
|
|
||||||
|
|
||||||
def _get_context(self) -> gpg.Context:
|
|
||||||
return gpg.Context(armor=False, offline=True, home_dir=self._home_dir)
|
|
||||||
|
|
||||||
def generate_key(self) -> None:
|
|
||||||
with self._get_context() as context:
|
|
||||||
result = cast(
|
|
||||||
GenkeyResult,
|
|
||||||
context.create_key(
|
|
||||||
f"xmpp:{self._jid}",
|
|
||||||
algorithm="default",
|
|
||||||
expires=False,
|
|
||||||
passphrase=None,
|
|
||||||
force=False,
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
log.info("Generated new key: %s", result.fpr)
|
|
||||||
|
|
||||||
def _get_key(self, fingerprint: str) -> Key | None:
|
|
||||||
with self._get_context() as context:
|
|
||||||
try:
|
|
||||||
return cast(Key, context.get_key(fingerprint))
|
|
||||||
except KeyNotFound as error:
|
|
||||||
log.warning("key not found: %s", error.keystr)
|
|
||||||
return
|
|
||||||
|
|
||||||
except Exception as error:
|
|
||||||
log.warning("get_key() error: %s", error)
|
|
||||||
return
|
|
||||||
|
|
||||||
def get_own_key_details(self) -> tuple[str | None, int | None]:
|
|
||||||
with self._get_context() as context:
|
|
||||||
keys = cast(list[Key], list(context.keylist(secret=True)))
|
|
||||||
if not keys:
|
|
||||||
return None, None
|
|
||||||
|
|
||||||
key = keys[0]
|
|
||||||
for subkey in key.subkeys:
|
|
||||||
if subkey.fpr == key.fpr:
|
|
||||||
return subkey.fpr, subkey.timestamp
|
|
||||||
|
|
||||||
return None, None
|
|
||||||
|
|
||||||
def get_keys(self) -> Sequence[KeyringItem]:
|
|
||||||
keys: list[KeyringItem] = []
|
|
||||||
with self._get_context() as context:
|
|
||||||
for key in cast(Iterator[Key], context.keylist(secret=False)):
|
|
||||||
keyring_item = KeyringItem(key)
|
|
||||||
if not keyring_item.is_xmpp_key:
|
|
||||||
log.warning("Key not suited for xmpp: %s", key.fpr)
|
|
||||||
self.delete_key(keyring_item.fingerprint)
|
|
||||||
continue
|
|
||||||
|
|
||||||
keys.append(keyring_item)
|
|
||||||
|
|
||||||
return keys
|
|
||||||
|
|
||||||
def export_key(self, fingerprint: str) -> bytes | None:
|
|
||||||
with self._get_context() as context:
|
|
||||||
return context.key_export_minimal(pattern=fingerprint)
|
|
||||||
|
|
||||||
# def encrypt_decrypt_files(self):
|
|
||||||
# c = gpg.Context()
|
|
||||||
# recipient = c.get_key("fingerprint of recipient's key")
|
|
||||||
|
|
||||||
# # Encrypt
|
|
||||||
# with open('foo.txt', 'r') as input_file:
|
|
||||||
# with open('foo.txt.gpg', 'wb') as output_file:
|
|
||||||
# c.encrypt([recipient], 0, input_file, output_file)
|
|
||||||
|
|
||||||
# # Decrypt
|
|
||||||
# with open('foo.txt.gpg', 'rb') as input_file:
|
|
||||||
# with open('foo2.txt', 'w') as output_file:
|
|
||||||
# c.decrypt(input_file, output_file)
|
|
||||||
|
|
||||||
def encrypt(
|
|
||||||
self, payload: bytes, keys: list[KeyData]
|
|
||||||
) -> tuple[bytes | None, str | None]:
|
|
||||||
recipients: list[Any] = []
|
|
||||||
with self._get_context() as context:
|
|
||||||
for key in keys:
|
|
||||||
key = cast(Key | None, context.get_key(key.fingerprint))
|
|
||||||
if key is not None:
|
|
||||||
recipients.append(key)
|
|
||||||
|
|
||||||
if not recipients:
|
|
||||||
return None, "No keys found to encrypt to"
|
|
||||||
|
|
||||||
with self._get_context() as context:
|
|
||||||
result = context.encrypt(payload, recipients, always_trust=True)
|
|
||||||
|
|
||||||
ciphertext, result, _sign_result = result
|
|
||||||
return ciphertext, None
|
|
||||||
|
|
||||||
raise RuntimeError
|
|
||||||
|
|
||||||
def decrypt(self, payload: bytes) -> tuple[str, str]:
|
|
||||||
with self._get_context() as context:
|
|
||||||
try:
|
|
||||||
result = context.decrypt(payload)
|
|
||||||
except Exception as error:
|
|
||||||
raise DecryptionFailed("Decryption failed: %s" % error)
|
|
||||||
|
|
||||||
plaintext, result, verify_result = result
|
|
||||||
plaintext = plaintext.decode()
|
|
||||||
|
|
||||||
fingerprints = [sig.fpr for sig in verify_result.signatures]
|
|
||||||
if not fingerprints or len(fingerprints) > 1:
|
|
||||||
log.error(result)
|
|
||||||
log.error(verify_result)
|
|
||||||
raise DecryptionFailed("Verification failed")
|
|
||||||
|
|
||||||
return plaintext, fingerprints[0]
|
|
||||||
|
|
||||||
raise RuntimeError
|
|
||||||
|
|
||||||
def import_key(self, data: bytes, jid: JID) -> KeyringItem | None:
|
|
||||||
log.info("Import key from %s", jid)
|
|
||||||
item = None
|
|
||||||
with self._get_context() as context:
|
|
||||||
result = context.key_import(data)
|
|
||||||
if not isinstance(result, ImportResult) or result.imported != 1:
|
|
||||||
log.error("Key import failed: %s", jid)
|
|
||||||
log.error(result)
|
|
||||||
return
|
|
||||||
|
|
||||||
fingerprint = result.imports[0].fpr
|
|
||||||
key = self._get_key(fingerprint)
|
|
||||||
assert key is not None
|
|
||||||
item = KeyringItem(key)
|
|
||||||
if not item.is_valid(jid):
|
|
||||||
log.warning("Invalid key found")
|
|
||||||
log.warning(key)
|
|
||||||
self.delete_key(item.fingerprint)
|
|
||||||
return
|
|
||||||
|
|
||||||
return item
|
|
||||||
|
|
||||||
def delete_key(self, fingerprint: str) -> None:
|
|
||||||
log.info("Delete Key: %s", fingerprint)
|
|
||||||
key = self._get_key(fingerprint)
|
|
||||||
assert key is not None
|
|
||||||
with self._get_context() as context:
|
|
||||||
context.op_delete(key, True) # pyright: ignore
|
|
||||||
@@ -1,70 +0,0 @@
|
|||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
|
|
||||||
class UID:
|
|
||||||
address: Any | None
|
|
||||||
comment: str
|
|
||||||
email: str
|
|
||||||
invalid: int
|
|
||||||
last_update: int
|
|
||||||
name: str
|
|
||||||
origin: int
|
|
||||||
revoked: int
|
|
||||||
signatures: list[Any]
|
|
||||||
thisown: bool
|
|
||||||
tofu: list[Any]
|
|
||||||
uid: str
|
|
||||||
uidhash: str
|
|
||||||
validity: int
|
|
||||||
|
|
||||||
|
|
||||||
class SubKey:
|
|
||||||
can_authenticate: int
|
|
||||||
can_certify: int
|
|
||||||
can_encrypt: int
|
|
||||||
can_sign: int
|
|
||||||
card_number: Any | None
|
|
||||||
curve: Any | None
|
|
||||||
disabled: int
|
|
||||||
expired: int
|
|
||||||
expires: int
|
|
||||||
fpr: str
|
|
||||||
invalid: int
|
|
||||||
is_cardkey: int
|
|
||||||
is_de_vs: int
|
|
||||||
is_qualified: int
|
|
||||||
keygrip: Any | None
|
|
||||||
keyid: str
|
|
||||||
length: int
|
|
||||||
pubkey_algo: int
|
|
||||||
revoked: int
|
|
||||||
secret: int
|
|
||||||
thisown: bool
|
|
||||||
timestamp: int
|
|
||||||
|
|
||||||
|
|
||||||
class Key:
|
|
||||||
can_authenticate: int
|
|
||||||
can_certify: int
|
|
||||||
can_encrypt: int
|
|
||||||
can_sign: int
|
|
||||||
chain_id: Any | None
|
|
||||||
disabled: int
|
|
||||||
expired: int
|
|
||||||
fpr: str
|
|
||||||
invalid: int
|
|
||||||
is_qualified: int
|
|
||||||
issuer_name: str | None
|
|
||||||
issuer_serial: str | None
|
|
||||||
keylist_mode: int
|
|
||||||
last_update: int
|
|
||||||
origin: int
|
|
||||||
owner_trust: int
|
|
||||||
protocol: int
|
|
||||||
revoked: int
|
|
||||||
secret: int
|
|
||||||
subkeys: list[SubKey]
|
|
||||||
thisown: bool
|
|
||||||
uids: list[UID]
|
|
||||||
@@ -1,177 +0,0 @@
|
|||||||
# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>
|
|
||||||
#
|
|
||||||
# This file is part of the OpenPGP Gajim Plugin.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation; version 3 only.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin 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 General Public License for more details.
|
|
||||||
#
|
|
||||||
# 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
|
|
||||||
|
|
||||||
import logging
|
|
||||||
from collections.abc import Sequence
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
import gnupg
|
|
||||||
from nbxmpp.protocol import JID
|
|
||||||
|
|
||||||
from openpgp.backend.base import BaseKeyringItem
|
|
||||||
from openpgp.backend.base import BasePGPBackend
|
|
||||||
from openpgp.backend.util import parse_uid
|
|
||||||
from openpgp.modules.key_store import KeyData
|
|
||||||
from openpgp.modules.util import DecryptionFailed
|
|
||||||
|
|
||||||
log = logging.getLogger("gajim.p.openpgp.pygnupg")
|
|
||||||
if log.getEffectiveLevel() == logging.DEBUG:
|
|
||||||
log = logging.getLogger("gnupg")
|
|
||||||
log.addHandler(logging.StreamHandler())
|
|
||||||
log.setLevel(logging.DEBUG)
|
|
||||||
|
|
||||||
|
|
||||||
class KeyringItem(BaseKeyringItem):
|
|
||||||
def __init__(self, key: dict[Any, Any]) -> None:
|
|
||||||
self._key = key
|
|
||||||
BaseKeyringItem.__init__(self)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def keyid(self) -> str:
|
|
||||||
return self._key["keyid"]
|
|
||||||
|
|
||||||
def _get_uid(self) -> str | None:
|
|
||||||
for uid in self._key["uids"]:
|
|
||||||
try:
|
|
||||||
return parse_uid(uid)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
@property
|
|
||||||
def fingerprint(self) -> str:
|
|
||||||
return self._key["fingerprint"]
|
|
||||||
|
|
||||||
|
|
||||||
class PythonGnuPG(BasePGPBackend):
|
|
||||||
def __init__(self, jid: str, gnupghome: Path) -> None:
|
|
||||||
self._gnupg = gnupg.GPG(gpgbinary="gpg", gnupghome=str(gnupghome))
|
|
||||||
self._jid = jid
|
|
||||||
self._own_fingerprint = None
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _get_key_params(jid: str) -> str:
|
|
||||||
"""
|
|
||||||
Generate --gen-key input
|
|
||||||
"""
|
|
||||||
|
|
||||||
params = {
|
|
||||||
"Key-Type": "RSA",
|
|
||||||
"Key-Length": 2048,
|
|
||||||
"Name-Real": "xmpp:%s" % jid,
|
|
||||||
}
|
|
||||||
|
|
||||||
out = "Key-Type: %s\n" % params.pop("Key-Type")
|
|
||||||
for key, val in list(params.items()):
|
|
||||||
out += "%s: %s\n" % (key, val)
|
|
||||||
out += "%no-protection\n"
|
|
||||||
out += "%commit\n"
|
|
||||||
return out
|
|
||||||
|
|
||||||
def generate_key(self) -> None:
|
|
||||||
self._gnupg.gen_key(self._get_key_params(self._jid))
|
|
||||||
|
|
||||||
def encrypt(
|
|
||||||
self, payload: bytes, keys: list[KeyData]
|
|
||||||
) -> tuple[bytes | None, str | None]:
|
|
||||||
recipients = [key.fingerprint for key in keys]
|
|
||||||
log.info("encrypt to:")
|
|
||||||
for fingerprint in recipients:
|
|
||||||
log.info(fingerprint)
|
|
||||||
|
|
||||||
result = self._gnupg.encrypt(
|
|
||||||
payload,
|
|
||||||
recipients,
|
|
||||||
armor=False,
|
|
||||||
sign=self._own_fingerprint,
|
|
||||||
always_trust=True,
|
|
||||||
)
|
|
||||||
|
|
||||||
if result.ok:
|
|
||||||
error = ""
|
|
||||||
else:
|
|
||||||
error = result.status
|
|
||||||
|
|
||||||
return result.data, error
|
|
||||||
|
|
||||||
def decrypt(self, payload: bytes) -> tuple[str, str]:
|
|
||||||
result = self._gnupg.decrypt(payload, always_trust=True)
|
|
||||||
if not result.ok:
|
|
||||||
raise DecryptionFailed(result.status)
|
|
||||||
|
|
||||||
assert result.fingerprint is not None
|
|
||||||
return result.data.decode("utf8"), result.fingerprint
|
|
||||||
|
|
||||||
def _get_key(self, fingerprint: str) -> gnupg.ListKeys:
|
|
||||||
return self._gnupg.list_keys(keys=[fingerprint])
|
|
||||||
|
|
||||||
def get_keys(self) -> Sequence[KeyringItem]:
|
|
||||||
result = self._gnupg.list_keys(secret=False)
|
|
||||||
keys: list[KeyringItem] = []
|
|
||||||
for key in result:
|
|
||||||
item = KeyringItem(key)
|
|
||||||
if not item.is_xmpp_key:
|
|
||||||
log.warning("Invalid key found, deleting key")
|
|
||||||
log.warning(key)
|
|
||||||
self.delete_key(item.fingerprint)
|
|
||||||
continue
|
|
||||||
keys.append(item)
|
|
||||||
return keys
|
|
||||||
|
|
||||||
def import_key(self, data: bytes, jid: JID) -> KeyringItem | None:
|
|
||||||
log.info("Import key from %s", jid)
|
|
||||||
result = self._gnupg.import_keys(data)
|
|
||||||
if not result:
|
|
||||||
log.error("Could not import key")
|
|
||||||
log.error(result)
|
|
||||||
return
|
|
||||||
|
|
||||||
fpr = result.results[0]["fingerprint"]
|
|
||||||
assert fpr is not None
|
|
||||||
|
|
||||||
key = self._get_key(fpr)
|
|
||||||
item = KeyringItem(key[0])
|
|
||||||
if not item.is_valid(jid):
|
|
||||||
log.warning("Invalid key found, deleting key")
|
|
||||||
log.warning(key)
|
|
||||||
self.delete_key(item.fingerprint)
|
|
||||||
return
|
|
||||||
|
|
||||||
return item
|
|
||||||
|
|
||||||
def get_own_key_details(self) -> tuple[str | None, int | None]:
|
|
||||||
result = self._gnupg.list_keys(secret=True)
|
|
||||||
if not result:
|
|
||||||
return None, None
|
|
||||||
|
|
||||||
if len(result) > 1:
|
|
||||||
log.error("More than one secret key found")
|
|
||||||
return None, None
|
|
||||||
|
|
||||||
self._own_fingerprint = result[0]["fingerprint"]
|
|
||||||
return self._own_fingerprint, int(result[0]["date"])
|
|
||||||
|
|
||||||
def export_key(self, fingerprint: str) -> bytes | None:
|
|
||||||
key = self._gnupg.export_keys(
|
|
||||||
fingerprint, secret=False, armor=False, minimal=True
|
|
||||||
)
|
|
||||||
assert isinstance(key, bytes | None)
|
|
||||||
return key
|
|
||||||
|
|
||||||
def delete_key(self, fingerprint: str) -> None:
|
|
||||||
log.info("Delete Key: %s", fingerprint)
|
|
||||||
self._gnupg.delete_keys(fingerprint)
|
|
||||||
@@ -1,125 +0,0 @@
|
|||||||
# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>
|
|
||||||
#
|
|
||||||
# This file is part of the OpenPGP Gajim Plugin.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation; version 3 only.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin 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 General Public License for more details.
|
|
||||||
#
|
|
||||||
# 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 NamedTuple
|
|
||||||
|
|
||||||
import logging
|
|
||||||
import sqlite3
|
|
||||||
from collections import namedtuple
|
|
||||||
from collections.abc import Iterator
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from nbxmpp.protocol import JID
|
|
||||||
|
|
||||||
from openpgp.modules.util import Trust
|
|
||||||
|
|
||||||
log = logging.getLogger("gajim.p.openpgp.sql")
|
|
||||||
|
|
||||||
TABLE_LAYOUT = """
|
|
||||||
CREATE TABLE contacts (
|
|
||||||
jid TEXT,
|
|
||||||
fingerprint TEXT,
|
|
||||||
active BOOLEAN,
|
|
||||||
trust INTEGER,
|
|
||||||
timestamp INTEGER,
|
|
||||||
comment TEXT
|
|
||||||
);
|
|
||||||
CREATE UNIQUE INDEX jid_fingerprint ON contacts (jid, fingerprint);"""
|
|
||||||
|
|
||||||
|
|
||||||
class ContactRow(NamedTuple):
|
|
||||||
jid: JID
|
|
||||||
fingerprint: str
|
|
||||||
active: bool
|
|
||||||
trust: Trust
|
|
||||||
timestamp: float
|
|
||||||
|
|
||||||
|
|
||||||
class Storage:
|
|
||||||
def __init__(self, folder_path: Path) -> None:
|
|
||||||
self._con = sqlite3.connect(
|
|
||||||
str(folder_path / "contacts.db"), detect_types=sqlite3.PARSE_COLNAMES
|
|
||||||
)
|
|
||||||
|
|
||||||
self._con.row_factory = self._namedtuple_factory
|
|
||||||
self._create_database()
|
|
||||||
self._migrate_database()
|
|
||||||
self._con.execute("PRAGMA synchronous=FULL;")
|
|
||||||
self._con.commit()
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _namedtuple_factory(cursor: sqlite3.Cursor, row: Any) -> Any:
|
|
||||||
fields = [col[0] for col in cursor.description]
|
|
||||||
Row = namedtuple("Row", fields) # pyright: ignore
|
|
||||||
named_row = Row(*row)
|
|
||||||
return named_row
|
|
||||||
|
|
||||||
def _user_version(self) -> int:
|
|
||||||
return self._con.execute("PRAGMA user_version").fetchone()[0]
|
|
||||||
|
|
||||||
def _create_database(self) -> None:
|
|
||||||
if not self._user_version():
|
|
||||||
log.info("Create contacts.db")
|
|
||||||
self._execute_query(TABLE_LAYOUT)
|
|
||||||
|
|
||||||
def _execute_query(self, query: str) -> None:
|
|
||||||
transaction = """
|
|
||||||
BEGIN TRANSACTION;
|
|
||||||
%s
|
|
||||||
PRAGMA user_version=1;
|
|
||||||
END TRANSACTION;
|
|
||||||
""" % (query)
|
|
||||||
self._con.executescript(transaction)
|
|
||||||
|
|
||||||
def _migrate_database(self) -> None:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def load_contacts(self) -> list[ContactRow]:
|
|
||||||
sql = """SELECT jid as "jid [jid]",
|
|
||||||
fingerprint,
|
|
||||||
active,
|
|
||||||
trust,
|
|
||||||
timestamp
|
|
||||||
FROM contacts"""
|
|
||||||
|
|
||||||
return self._con.execute(sql).fetchall()
|
|
||||||
|
|
||||||
def save_contact(
|
|
||||||
self, db_values: Iterator[tuple[JID, str, bool, Trust, float]]
|
|
||||||
) -> None:
|
|
||||||
sql = """REPLACE INTO
|
|
||||||
contacts(jid, fingerprint, active, trust, timestamp)
|
|
||||||
VALUES(?, ?, ?, ?, ?)"""
|
|
||||||
for values in db_values:
|
|
||||||
log.info("Store key: %s", values)
|
|
||||||
self._con.execute(sql, values)
|
|
||||||
self._con.commit()
|
|
||||||
|
|
||||||
def set_trust(self, jid: JID, fingerprint: str, trust: Trust) -> None:
|
|
||||||
sql = "UPDATE contacts SET trust = ? WHERE jid = ? AND fingerprint = ?"
|
|
||||||
log.info("Set Trust: %s %s %s", trust, jid, fingerprint)
|
|
||||||
self._con.execute(sql, (trust, jid, fingerprint))
|
|
||||||
self._con.commit()
|
|
||||||
|
|
||||||
def delete_key(self, jid: JID, fingerprint: str) -> None:
|
|
||||||
sql = "DELETE from contacts WHERE jid = ? AND fingerprint = ?"
|
|
||||||
log.info("Delete Key: %s %s", jid, fingerprint)
|
|
||||||
self._con.execute(sql, (jid, fingerprint))
|
|
||||||
self._con.commit()
|
|
||||||
|
|
||||||
def cleanup(self) -> None:
|
|
||||||
self._con.close()
|
|
||||||
@@ -1,12 +0,0 @@
|
|||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
|
|
||||||
def parse_uid(uid: str, compat: bool = False) -> str:
|
|
||||||
if uid.startswith("xmpp:"):
|
|
||||||
return uid[5:]
|
|
||||||
|
|
||||||
# Compat with uids of form "Name <xmpp:my@jid.com>"
|
|
||||||
if compat and "<xmpp:" in uid and uid.endswith(">"):
|
|
||||||
return uid[:-1].split("<xmpp:", maxsplit=1)[1]
|
|
||||||
|
|
||||||
raise ValueError("Unknown UID format: %s" % uid)
|
|
||||||
@@ -1,250 +0,0 @@
|
|||||||
# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>
|
|
||||||
#
|
|
||||||
# This file is part of the OpenPGP Gajim Plugin.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation; version 3 only.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin 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 General Public License for more details.
|
|
||||||
#
|
|
||||||
# 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 __future__ import annotations
|
|
||||||
|
|
||||||
from typing import cast
|
|
||||||
|
|
||||||
import logging
|
|
||||||
import time
|
|
||||||
|
|
||||||
from gi.repository import Gtk
|
|
||||||
from nbxmpp import JID
|
|
||||||
|
|
||||||
from gajim.common import app
|
|
||||||
from gajim.gtk.alert import ConfirmationAlertDialog
|
|
||||||
from gajim.gtk.util.misc import container_remove_all
|
|
||||||
from gajim.gtk.window import GajimAppWindow
|
|
||||||
from gajim.plugins.plugins_i18n import _
|
|
||||||
|
|
||||||
from openpgp.modules.key_store import KeyData
|
|
||||||
from openpgp.modules.openpgp import OpenPGP
|
|
||||||
from openpgp.modules.util import Trust
|
|
||||||
|
|
||||||
log = logging.getLogger("gajim.p.openpgp.keydialog")
|
|
||||||
|
|
||||||
TRUST_DATA = {
|
|
||||||
Trust.NOT_TRUSTED: ("dialog-error-symbolic", _("Not Trusted"), "error-color"),
|
|
||||||
Trust.UNKNOWN: ("security-low-symbolic", _("Not Decided"), "warning-color"),
|
|
||||||
Trust.BLIND: ("security-medium-symbolic", _("Blind Trust"), "encrypted-color"),
|
|
||||||
Trust.VERIFIED: ("security-high-symbolic", _("Verified"), "encrypted-color"),
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
class KeyDialog(GajimAppWindow):
|
|
||||||
def __init__(self, account: str, jid: JID, transient: Gtk.Window) -> None:
|
|
||||||
GajimAppWindow.__init__(
|
|
||||||
self,
|
|
||||||
name="PGPKeyDialog",
|
|
||||||
title=_("Public Keys for %s") % jid,
|
|
||||||
default_width=450,
|
|
||||||
default_height=400,
|
|
||||||
transient_for=transient,
|
|
||||||
modal=True,
|
|
||||||
add_window_padding=True,
|
|
||||||
header_bar=True,
|
|
||||||
)
|
|
||||||
|
|
||||||
self.add_css_class("openpgp-key-dialog")
|
|
||||||
|
|
||||||
self._client = app.get_client(account)
|
|
||||||
|
|
||||||
self._listbox = Gtk.ListBox()
|
|
||||||
self._listbox.set_selection_mode(Gtk.SelectionMode.NONE)
|
|
||||||
|
|
||||||
self._scrolled = Gtk.ScrolledWindow(hexpand=True)
|
|
||||||
self._scrolled.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
|
|
||||||
self._scrolled.set_child(self._listbox)
|
|
||||||
|
|
||||||
self.set_child(self._scrolled)
|
|
||||||
|
|
||||||
open_pgp_module = cast(OpenPGP, self._client.get_module("OpenPGP")) # type: ignore
|
|
||||||
keys = open_pgp_module.get_keys(jid, only_trusted=False)
|
|
||||||
for key in keys:
|
|
||||||
log.info("Load: %s", key.fingerprint)
|
|
||||||
self._listbox.append(KeyRow(key, self))
|
|
||||||
|
|
||||||
def _cleanup(self) -> None:
|
|
||||||
del self._client
|
|
||||||
del self._listbox
|
|
||||||
del self._scrolled
|
|
||||||
|
|
||||||
|
|
||||||
class KeyRow(Gtk.ListBoxRow):
|
|
||||||
def __init__(self, key: KeyData, dialog: GajimAppWindow):
|
|
||||||
Gtk.ListBoxRow.__init__(self)
|
|
||||||
self.set_activatable(False)
|
|
||||||
|
|
||||||
self._dialog = dialog
|
|
||||||
self.key = key
|
|
||||||
|
|
||||||
box = Gtk.Box()
|
|
||||||
box.set_spacing(12)
|
|
||||||
|
|
||||||
self._trust_button = Gtk.MenuButton()
|
|
||||||
self._trust_button.set_popover(TrustPopver(self))
|
|
||||||
self._update_button_state()
|
|
||||||
box.append(self._trust_button)
|
|
||||||
|
|
||||||
label_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
|
|
||||||
fingerprint = Gtk.Label(label=self._format_fingerprint(key.fingerprint))
|
|
||||||
fingerprint.get_style_context().add_class("openpgp-mono")
|
|
||||||
if not key.active:
|
|
||||||
fingerprint.get_style_context().add_class("openpgp-inactive-color")
|
|
||||||
fingerprint.set_selectable(True)
|
|
||||||
fingerprint.set_halign(Gtk.Align.START)
|
|
||||||
fingerprint.set_valign(Gtk.Align.START)
|
|
||||||
fingerprint.set_hexpand(True)
|
|
||||||
label_box.append(fingerprint)
|
|
||||||
|
|
||||||
date = Gtk.Label(label=self._format_timestamp(key.timestamp))
|
|
||||||
date.set_halign(Gtk.Align.START)
|
|
||||||
date.get_style_context().add_class("openpgp-mono")
|
|
||||||
if not key.active:
|
|
||||||
date.get_style_context().add_class("openpgp-inactive-color")
|
|
||||||
label_box.append(date)
|
|
||||||
|
|
||||||
box.append(label_box)
|
|
||||||
self.set_child(box)
|
|
||||||
|
|
||||||
def _update_button_state(self) -> None:
|
|
||||||
icon_name, tooltip, css_class = TRUST_DATA[self.key.trust]
|
|
||||||
self._trust_button.set_icon_name(icon_name)
|
|
||||||
|
|
||||||
for css_cls in self._trust_button.get_css_classes():
|
|
||||||
if css_cls.startswith("openpgp"):
|
|
||||||
self._trust_button.remove_css_class(css_cls)
|
|
||||||
|
|
||||||
if not self.key.active:
|
|
||||||
css_class = "inactive-color"
|
|
||||||
tooltip = "%s - %s" % (_("Inactive"), tooltip)
|
|
||||||
|
|
||||||
self._trust_button.add_css_class(f"openpgp-{css_class}")
|
|
||||||
self._trust_button.set_tooltip_text(tooltip)
|
|
||||||
|
|
||||||
def delete_fingerprint(self):
|
|
||||||
def _remove():
|
|
||||||
listbox = cast(Gtk.ListBox, self.get_parent())
|
|
||||||
listbox.remove(self)
|
|
||||||
self.key.delete()
|
|
||||||
|
|
||||||
ConfirmationAlertDialog(
|
|
||||||
_("Delete Public Key?"),
|
|
||||||
_("This will permanently delete this public key"),
|
|
||||||
confirm_label=_("_Delete"),
|
|
||||||
appearance="destructive",
|
|
||||||
callback=_remove,
|
|
||||||
)
|
|
||||||
|
|
||||||
def set_trust(self, trust: Trust) -> None:
|
|
||||||
self.key.trust = trust
|
|
||||||
self._update_button_state()
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _format_fingerprint(fingerprint: str) -> str:
|
|
||||||
fplen = len(fingerprint)
|
|
||||||
wordsize = fplen // 8
|
|
||||||
buf = ""
|
|
||||||
for w in range(0, fplen, wordsize):
|
|
||||||
buf += f"{fingerprint[w : w + wordsize]} "
|
|
||||||
return buf.rstrip()
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _format_timestamp(timestamp: float) -> str:
|
|
||||||
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timestamp))
|
|
||||||
|
|
||||||
|
|
||||||
class TrustPopver(Gtk.Popover):
|
|
||||||
def __init__(self, row: KeyRow):
|
|
||||||
Gtk.Popover.__init__(self)
|
|
||||||
self._row = row
|
|
||||||
self._listbox = Gtk.ListBox()
|
|
||||||
self._listbox.set_selection_mode(Gtk.SelectionMode.NONE)
|
|
||||||
if row.key.trust != Trust.VERIFIED:
|
|
||||||
self._listbox.append(VerifiedOption())
|
|
||||||
if row.key.trust != Trust.NOT_TRUSTED:
|
|
||||||
self._listbox.append(NotTrustedOption())
|
|
||||||
self._listbox.append(DeleteOption())
|
|
||||||
self.set_child(self._listbox)
|
|
||||||
self._listbox.connect("row-activated", self._activated)
|
|
||||||
self.add_css_class("openpgp-trust-popover")
|
|
||||||
|
|
||||||
def _activated(self, listbox: Gtk.ListBox, row: MenuOption) -> None:
|
|
||||||
self.popdown()
|
|
||||||
if row.type_ is None:
|
|
||||||
self._row.delete_fingerprint()
|
|
||||||
else:
|
|
||||||
self._row.set_trust(row.type_)
|
|
||||||
self.update()
|
|
||||||
|
|
||||||
def update(self):
|
|
||||||
container_remove_all(self._listbox)
|
|
||||||
if self._row.key.trust != Trust.VERIFIED:
|
|
||||||
self._listbox.append(VerifiedOption())
|
|
||||||
if self._row.key.trust != Trust.NOT_TRUSTED:
|
|
||||||
self._listbox.append(NotTrustedOption())
|
|
||||||
self._listbox.append(DeleteOption())
|
|
||||||
|
|
||||||
|
|
||||||
class MenuOption(Gtk.ListBoxRow):
|
|
||||||
type_: Trust | None
|
|
||||||
icon: str
|
|
||||||
label: str
|
|
||||||
color: str
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
Gtk.ListBoxRow.__init__(self)
|
|
||||||
box = Gtk.Box()
|
|
||||||
box.set_spacing(6)
|
|
||||||
|
|
||||||
image = Gtk.Image.new_from_icon_name(self.icon)
|
|
||||||
if self.color:
|
|
||||||
image.add_css_class(self.color)
|
|
||||||
|
|
||||||
label = Gtk.Label(label=self.label)
|
|
||||||
box.append(image)
|
|
||||||
box.append(label)
|
|
||||||
self.set_child(box)
|
|
||||||
|
|
||||||
|
|
||||||
class VerifiedOption(MenuOption):
|
|
||||||
type_ = Trust.VERIFIED
|
|
||||||
icon = "security-high-symbolic"
|
|
||||||
label = _("Verified")
|
|
||||||
color = "encrypted-color"
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
MenuOption.__init__(self)
|
|
||||||
|
|
||||||
|
|
||||||
class NotTrustedOption(MenuOption):
|
|
||||||
type_ = Trust.NOT_TRUSTED
|
|
||||||
icon = "dialog-error-symbolic"
|
|
||||||
label = _("Not Trusted")
|
|
||||||
color = "error-color"
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
MenuOption.__init__(self)
|
|
||||||
|
|
||||||
|
|
||||||
class DeleteOption(MenuOption):
|
|
||||||
type_ = None
|
|
||||||
icon = "user-trash-symbolic"
|
|
||||||
label = _("Delete")
|
|
||||||
color = ""
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
MenuOption.__init__(self)
|
|
||||||
@@ -1,14 +0,0 @@
|
|||||||
.openpgp-inactive-color button > box > image { color: @unfocused_borders; }
|
|
||||||
.openpgp-error-color button > box > image { color: @error_color; }
|
|
||||||
.openpgp-warning-color button > box > image { color: @warning_color; }
|
|
||||||
.openpgp-encrypted-color button > box > image { color: rgb(75, 181, 67); }
|
|
||||||
.openpgp-mono { font-size: 12px; font-family: monospace; }
|
|
||||||
.openpgp-key-dialog > box { margin: 12px; }
|
|
||||||
.openpgp-key-dialog scrolledwindow row {
|
|
||||||
border-bottom: 1px solid;
|
|
||||||
border-color: @unfocused_borders;
|
|
||||||
padding: 10px 20px 10px 10px;
|
|
||||||
}
|
|
||||||
.openpgp-key-dialog scrolledwindow row:last-child { border-bottom: 0px; }
|
|
||||||
.openpgp-key-dialog scrolledwindow { border: 1px solid; border-color:@unfocused_borders; }
|
|
||||||
.openpgp-trust-popover row { padding: 10px 15px 10px 10px; }
|
|
||||||
@@ -1,252 +0,0 @@
|
|||||||
# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>
|
|
||||||
#
|
|
||||||
# This file is part of the OpenPGP Gajim Plugin.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation; version 3 only.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin 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 General Public License for more details.
|
|
||||||
#
|
|
||||||
# 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 __future__ import annotations
|
|
||||||
|
|
||||||
from typing import cast
|
|
||||||
|
|
||||||
import logging
|
|
||||||
import threading
|
|
||||||
from enum import IntEnum
|
|
||||||
|
|
||||||
from gi.repository import GLib
|
|
||||||
from gi.repository import Gtk
|
|
||||||
|
|
||||||
from gajim.common import app
|
|
||||||
from gajim.common.client import Client
|
|
||||||
from gajim.gtk.control import ChatControl
|
|
||||||
from gajim.plugins.plugins_i18n import _
|
|
||||||
|
|
||||||
from ..pgpplugin import OpenPGPPlugin
|
|
||||||
|
|
||||||
log = logging.getLogger("gajim.p.openpgp.wizard")
|
|
||||||
|
|
||||||
|
|
||||||
class Page(IntEnum):
|
|
||||||
WELCOME = 0
|
|
||||||
NEWKEY = 1
|
|
||||||
SUCCESS = 2
|
|
||||||
ERROR = 3
|
|
||||||
|
|
||||||
|
|
||||||
class KeyWizard(Gtk.Assistant):
|
|
||||||
def __init__(
|
|
||||||
self, plugin: OpenPGPPlugin, account: str, chat_control: ChatControl
|
|
||||||
) -> None:
|
|
||||||
Gtk.Assistant.__init__(self)
|
|
||||||
|
|
||||||
self._client = app.get_client(account)
|
|
||||||
self._plugin = plugin
|
|
||||||
self._account = account
|
|
||||||
self._data_form_widget = None
|
|
||||||
self._is_form = None
|
|
||||||
self._chat_control = chat_control
|
|
||||||
|
|
||||||
self.set_application(app.app)
|
|
||||||
self.set_transient_for(app.window)
|
|
||||||
self.set_resizable(True)
|
|
||||||
|
|
||||||
self.set_default_size(600, 400)
|
|
||||||
self.get_style_context().add_class("dialog-margin")
|
|
||||||
|
|
||||||
self._add_page(WelcomePage())
|
|
||||||
# self._add_page(BackupKeyPage())
|
|
||||||
self._add_page(NewKeyPage(self, self._client))
|
|
||||||
# self._add_page(SaveBackupCodePage())
|
|
||||||
self._add_page(SuccessfulPage())
|
|
||||||
self._add_page(ErrorPage())
|
|
||||||
|
|
||||||
self.connect("prepare", self._on_page_change)
|
|
||||||
self.connect("cancel", self._on_cancel)
|
|
||||||
self.connect("close", self._on_cancel)
|
|
||||||
|
|
||||||
self._remove_sidebar()
|
|
||||||
self.show()
|
|
||||||
|
|
||||||
def _add_page(self, page: PagesT) -> None:
|
|
||||||
self.append_page(page)
|
|
||||||
self.set_page_type(page, page.type_)
|
|
||||||
self.set_page_title(page, page.title)
|
|
||||||
self.set_page_complete(page, page.complete)
|
|
||||||
|
|
||||||
def _remove_sidebar(self) -> None:
|
|
||||||
main_box = cast(Gtk.Box, self.get_child())
|
|
||||||
sidebar = main_box.get_first_child()
|
|
||||||
assert sidebar is not None
|
|
||||||
main_box.remove(sidebar)
|
|
||||||
|
|
||||||
def _activate_encryption(self):
|
|
||||||
action = app.window.lookup_action("set-encryption")
|
|
||||||
assert action is not None
|
|
||||||
action.activate(GLib.Variant("s", self._plugin.encryption_name))
|
|
||||||
|
|
||||||
def _on_page_change(self, assistant: Gtk.Assistant, page: PagesT) -> None:
|
|
||||||
if self.get_current_page() == Page.NEWKEY:
|
|
||||||
page = cast(NewKeyPage, page)
|
|
||||||
if self._client.get_module("OpenPGP").secret_key_available: # pyright: ignore
|
|
||||||
self.set_current_page(Page.SUCCESS)
|
|
||||||
else:
|
|
||||||
page.generate()
|
|
||||||
elif self.get_current_page() == Page.SUCCESS:
|
|
||||||
self._activate_encryption()
|
|
||||||
|
|
||||||
def _on_cancel(self, widget: Gtk.Assistant):
|
|
||||||
self.destroy()
|
|
||||||
|
|
||||||
|
|
||||||
class WelcomePage(Gtk.Box):
|
|
||||||
type_ = Gtk.AssistantPageType.INTRO
|
|
||||||
title = _("Welcome")
|
|
||||||
complete = True
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
super().__init__(orientation=Gtk.Orientation.VERTICAL)
|
|
||||||
self.set_spacing(18)
|
|
||||||
title_label = Gtk.Label(label=_("Setup OpenPGP"))
|
|
||||||
text_label = Gtk.Label(label=_("Gajim will now try to setup OpenPGP for you"))
|
|
||||||
self.append(title_label)
|
|
||||||
self.append(text_label)
|
|
||||||
|
|
||||||
|
|
||||||
class RequestPage(Gtk.Box):
|
|
||||||
type_ = Gtk.AssistantPageType.INTRO
|
|
||||||
title = _("Request OpenPGP Key")
|
|
||||||
complete = False
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
super().__init__(orientation=Gtk.Orientation.VERTICAL)
|
|
||||||
self.set_spacing(18)
|
|
||||||
spinner = Gtk.Spinner()
|
|
||||||
self.append(spinner)
|
|
||||||
spinner.start()
|
|
||||||
|
|
||||||
|
|
||||||
# class BackupKeyPage(Gtk.Box):
|
|
||||||
|
|
||||||
# type_ = Gtk.AssistantPageType.INTRO
|
|
||||||
# title = _('Supply Backup Code')
|
|
||||||
# complete = True
|
|
||||||
|
|
||||||
# def __init__(self):
|
|
||||||
# super().__init__(orientation=Gtk.Orientation.VERTICAL)
|
|
||||||
# self.set_spacing(18)
|
|
||||||
# title_label = Gtk.Label(label=_('Backup Code'))
|
|
||||||
# text_label = Gtk.Label(
|
|
||||||
# label=_('We found a backup Code, please supply your password'))
|
|
||||||
# self.add(title_label)
|
|
||||||
# self.add(text_label)
|
|
||||||
# entry = Gtk.Entry()
|
|
||||||
# self.add(entry)
|
|
||||||
|
|
||||||
|
|
||||||
class NewKeyPage(RequestPage):
|
|
||||||
type_ = Gtk.AssistantPageType.PROGRESS
|
|
||||||
title = _("Generating new Key")
|
|
||||||
complete = False
|
|
||||||
|
|
||||||
def __init__(self, assistant: Gtk.Assistant, client: Client) -> None:
|
|
||||||
super().__init__()
|
|
||||||
self._assistant = assistant
|
|
||||||
self._client = client
|
|
||||||
|
|
||||||
def generate(self):
|
|
||||||
log.info("Creating Key")
|
|
||||||
thread = threading.Thread(target=self.worker)
|
|
||||||
thread.start()
|
|
||||||
|
|
||||||
def worker(self):
|
|
||||||
text = None
|
|
||||||
try:
|
|
||||||
self._client.get_module("OpenPGP").generate_key() # pyright: ignore
|
|
||||||
except Exception as error:
|
|
||||||
text = str(error)
|
|
||||||
|
|
||||||
GLib.idle_add(self.finished, text)
|
|
||||||
|
|
||||||
def finished(self, error: str | None) -> None:
|
|
||||||
if error is None:
|
|
||||||
self._client.get_module("OpenPGP").get_own_key_details() # pyright: ignore
|
|
||||||
self._client.get_module("OpenPGP").set_public_key() # pyright: ignore
|
|
||||||
self._client.get_module("OpenPGP").request_keylist() # pyright: ignore
|
|
||||||
self._assistant.set_current_page(Page.SUCCESS)
|
|
||||||
else:
|
|
||||||
error_page = cast(ErrorPage, self._assistant.get_nth_page(Page.ERROR))
|
|
||||||
error_page.set_text(error)
|
|
||||||
self._assistant.set_current_page(Page.ERROR)
|
|
||||||
|
|
||||||
|
|
||||||
# class SaveBackupCodePage(RequestPage):
|
|
||||||
|
|
||||||
# type_ = Gtk.AssistantPageType.PROGRESS
|
|
||||||
# title = _('Save this code')
|
|
||||||
# complete = False
|
|
||||||
|
|
||||||
# def __init__(self):
|
|
||||||
# super().__init__(orientation=Gtk.Orientation.VERTICAL)
|
|
||||||
# self.set_spacing(18)
|
|
||||||
# title_label = Gtk.Label(label=_('Backup Code'))
|
|
||||||
# text_label = Gtk.Label(
|
|
||||||
# label=_('This is your backup code, you need it if you reinstall Gajim'))
|
|
||||||
# self.add(title_label)
|
|
||||||
# self.add(text_label)
|
|
||||||
|
|
||||||
|
|
||||||
class SuccessfulPage(Gtk.Box):
|
|
||||||
type_ = Gtk.AssistantPageType.SUMMARY
|
|
||||||
title = _("Setup successful")
|
|
||||||
complete = True
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
super().__init__(orientation=Gtk.Orientation.VERTICAL)
|
|
||||||
self.set_spacing(12)
|
|
||||||
self.set_homogeneous(True)
|
|
||||||
|
|
||||||
icon = Gtk.Image.new_from_icon_name("object-select-symbolic")
|
|
||||||
icon.add_css_class("success-color")
|
|
||||||
icon.set_valign(Gtk.Align.END)
|
|
||||||
label = Gtk.Label(label=_("Setup successful"))
|
|
||||||
label.add_css_class("bold16")
|
|
||||||
label.set_valign(Gtk.Align.START)
|
|
||||||
|
|
||||||
self.append(icon)
|
|
||||||
self.append(label)
|
|
||||||
|
|
||||||
|
|
||||||
class ErrorPage(Gtk.Box):
|
|
||||||
type_ = Gtk.AssistantPageType.SUMMARY
|
|
||||||
title = _("Setup failed")
|
|
||||||
complete = True
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
super().__init__(orientation=Gtk.Orientation.VERTICAL)
|
|
||||||
self.set_spacing(12)
|
|
||||||
self.set_homogeneous(True)
|
|
||||||
|
|
||||||
icon = Gtk.Image.new_from_icon_name("dialog-error-symbolic")
|
|
||||||
icon.get_style_context().add_class("error-color")
|
|
||||||
icon.set_valign(Gtk.Align.END)
|
|
||||||
self._label = Gtk.Label()
|
|
||||||
self._label.get_style_context().add_class("bold16")
|
|
||||||
self._label.set_valign(Gtk.Align.START)
|
|
||||||
|
|
||||||
self.append(icon)
|
|
||||||
self.append(self._label)
|
|
||||||
|
|
||||||
def set_text(self, text: str) -> None:
|
|
||||||
self._label.set_text(text)
|
|
||||||
|
|
||||||
|
|
||||||
PagesT = WelcomePage | RequestPage | NewKeyPage | SuccessfulPage | ErrorPage
|
|
||||||
@@ -1,269 +0,0 @@
|
|||||||
# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>
|
|
||||||
#
|
|
||||||
# This file is part of the OpenPGP Gajim Plugin.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation; version 3 only.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin 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 General Public License for more details.
|
|
||||||
#
|
|
||||||
# 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 __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")
|
|
||||||
|
|
||||||
|
|
||||||
class KeyData:
|
|
||||||
"""
|
|
||||||
Holds all data related to a certain key
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
contact_data: ContactData,
|
|
||||||
fingerprint: str,
|
|
||||||
active: bool,
|
|
||||||
trust: Trust,
|
|
||||||
timestamp: float,
|
|
||||||
):
|
|
||||||
self._contact_data = contact_data
|
|
||||||
self.fingerprint = fingerprint
|
|
||||||
self.active = active
|
|
||||||
self._trust = trust
|
|
||||||
self.timestamp = timestamp
|
|
||||||
self.comment = None
|
|
||||||
self.has_pubkey = False
|
|
||||||
|
|
||||||
@property
|
|
||||||
def trust(self) -> Trust:
|
|
||||||
return self._trust
|
|
||||||
|
|
||||||
@trust.setter
|
|
||||||
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)
|
|
||||||
|
|
||||||
def delete(self):
|
|
||||||
self._contact_data.delete_key(self.fingerprint)
|
|
||||||
|
|
||||||
|
|
||||||
class ContactData:
|
|
||||||
"""
|
|
||||||
Holds all data related to a contact
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, jid: JID, storage: Storage, pgp: BasePGPBackend) -> None:
|
|
||||||
self.jid = jid
|
|
||||||
self._key_store: dict[str, KeyData] = {}
|
|
||||||
self._storage = storage
|
|
||||||
self._pgp = pgp
|
|
||||||
|
|
||||||
@property
|
|
||||||
def userid(self):
|
|
||||||
assert self.jid is not None, "JID not set"
|
|
||||||
return "xmpp:%s" % self.jid
|
|
||||||
|
|
||||||
@property
|
|
||||||
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) -> Iterator[tuple[JID, str, bool, Trust, float]]:
|
|
||||||
for key in self._key_store.values():
|
|
||||||
yield (
|
|
||||||
self.jid,
|
|
||||||
key.fingerprint,
|
|
||||||
key.active,
|
|
||||||
key.trust,
|
|
||||||
key.timestamp,
|
|
||||||
)
|
|
||||||
|
|
||||||
def add_from_key(self, key: PGPKeyMetadata) -> KeyData:
|
|
||||||
try:
|
|
||||||
keydata = self._key_store[key.fingerprint]
|
|
||||||
except KeyError:
|
|
||||||
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: ContactRow) -> KeyData:
|
|
||||||
try:
|
|
||||||
keydata = self._key_store[row.fingerprint]
|
|
||||||
except KeyError:
|
|
||||||
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: list[PGPKeyMetadata] | None) -> list[str]:
|
|
||||||
log.info("Process keylist: %s %s", self.jid, keylist)
|
|
||||||
|
|
||||||
if keylist is None:
|
|
||||||
for keydata in self._key_store.values():
|
|
||||||
keydata.active = False
|
|
||||||
self._storage.save_contact(self.db_values())
|
|
||||||
return []
|
|
||||||
|
|
||||||
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():
|
|
||||||
if not key.has_pubkey:
|
|
||||||
missing_pub_keys.append(key.fingerprint)
|
|
||||||
return missing_pub_keys
|
|
||||||
|
|
||||||
for keydata in self._key_store.values():
|
|
||||||
keydata.active = False
|
|
||||||
|
|
||||||
for key in keylist:
|
|
||||||
try:
|
|
||||||
keydata = self._key_store[key.fingerprint]
|
|
||||||
keydata.active = True
|
|
||||||
if not keydata.has_pubkey:
|
|
||||||
missing_pub_keys.append(keydata.fingerprint)
|
|
||||||
except KeyError:
|
|
||||||
keydata = self.add_from_key(key)
|
|
||||||
missing_pub_keys.append(keydata.fingerprint)
|
|
||||||
|
|
||||||
self._storage.save_contact(self.db_values())
|
|
||||||
return missing_pub_keys
|
|
||||||
|
|
||||||
def set_public_key(self, fingerprint: str) -> None:
|
|
||||||
try:
|
|
||||||
keydata = self._key_store[fingerprint]
|
|
||||||
except KeyError:
|
|
||||||
log.warning(
|
|
||||||
"Set public key on unknown fingerprint: %s %s", self.jid, fingerprint
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
keydata.has_pubkey = True
|
|
||||||
log.info("Set public key: %s %s", self.jid, fingerprint)
|
|
||||||
|
|
||||||
def get_keys(self, only_trusted: bool = True) -> list[KeyData]:
|
|
||||||
keys = list(self._key_store.values())
|
|
||||||
if not only_trusted:
|
|
||||||
return keys
|
|
||||||
return [
|
|
||||||
k for k in keys if k.active and k.trust in (Trust.VERIFIED, Trust.BLIND)
|
|
||||||
]
|
|
||||||
|
|
||||||
def get_key(self, fingerprint: str) -> KeyData | None:
|
|
||||||
return self._key_store.get(fingerprint, None)
|
|
||||||
|
|
||||||
def set_trust(self, fingerprint: str, trust: Trust) -> None:
|
|
||||||
self._storage.set_trust(self.jid, fingerprint, trust)
|
|
||||||
|
|
||||||
def delete_key(self, fingerprint: str) -> None:
|
|
||||||
self._storage.delete_key(self.jid, fingerprint)
|
|
||||||
self._pgp.delete_key(fingerprint)
|
|
||||||
del self._key_store[fingerprint]
|
|
||||||
|
|
||||||
|
|
||||||
class PGPContacts:
|
|
||||||
"""
|
|
||||||
Holds all contacts available for PGP encryption
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, pgp: BasePGPBackend, storage: Storage) -> None:
|
|
||||||
self._contacts: dict[JID, ContactData] = {}
|
|
||||||
self._storage = storage
|
|
||||||
self._pgp = pgp
|
|
||||||
self._load_from_storage()
|
|
||||||
self._load_from_keyring()
|
|
||||||
|
|
||||||
def _load_from_keyring(self):
|
|
||||||
log.info("Load keys from keyring")
|
|
||||||
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()
|
|
||||||
for row in rows:
|
|
||||||
log.info("Found: %s %s", row.jid, row.fingerprint)
|
|
||||||
try:
|
|
||||||
contact_data = self._contacts[row.jid]
|
|
||||||
except KeyError:
|
|
||||||
contact_data = ContactData(row.jid, self._storage, self._pgp)
|
|
||||||
contact_data.add_from_db(row)
|
|
||||||
self._contacts[row.jid] = contact_data
|
|
||||||
else:
|
|
||||||
contact_data.add_from_db(row)
|
|
||||||
|
|
||||||
def process_keylist(
|
|
||||||
self, jid: JID, keylist: list[PGPKeyMetadata] | None
|
|
||||||
) -> list[str]:
|
|
||||||
try:
|
|
||||||
contact_data = self._contacts[jid]
|
|
||||||
except KeyError:
|
|
||||||
contact_data = ContactData(jid, self._storage, self._pgp)
|
|
||||||
missing_pub_keys = contact_data.process_keylist(keylist)
|
|
||||||
self._contacts[jid] = contact_data
|
|
||||||
else:
|
|
||||||
missing_pub_keys = contact_data.process_keylist(keylist)
|
|
||||||
|
|
||||||
return missing_pub_keys
|
|
||||||
|
|
||||||
def set_public_key(self, jid: JID, fingerprint: str) -> None:
|
|
||||||
try:
|
|
||||||
contact_data = self._contacts[jid]
|
|
||||||
except KeyError:
|
|
||||||
log.warning("ContactData not found: %s %s", jid, fingerprint)
|
|
||||||
else:
|
|
||||||
contact_data.set_public_key(fingerprint)
|
|
||||||
|
|
||||||
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: JID, fingerprint: str) -> Trust:
|
|
||||||
contact_data = self._contacts.get(jid, None)
|
|
||||||
if contact_data is None:
|
|
||||||
return Trust.UNKNOWN
|
|
||||||
|
|
||||||
key = contact_data.get_key(fingerprint)
|
|
||||||
if key is None:
|
|
||||||
return Trust.UNKNOWN
|
|
||||||
return key.trust
|
|
||||||
@@ -1,369 +0,0 @@
|
|||||||
# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>
|
|
||||||
#
|
|
||||||
# This file is part of the OpenPGP Gajim Plugin.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation; version 3 only.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin 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 General Public License for more details.
|
|
||||||
#
|
|
||||||
# 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.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 NOT_ENCRYPTED_TAGS
|
|
||||||
from openpgp.modules.util import prepare_stanza
|
|
||||||
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
|
|
||||||
|
|
||||||
|
|
||||||
log = logging.getLogger("gajim.p.openpgp")
|
|
||||||
|
|
||||||
|
|
||||||
# Module name
|
|
||||||
name = ENCRYPTION_NAME
|
|
||||||
|
|
||||||
|
|
||||||
class OpenPGP(BaseModule):
|
|
||||||
_nbxmpp_extends = "OpenPGP"
|
|
||||||
_nbxmpp_methods = [
|
|
||||||
"set_keylist",
|
|
||||||
"request_keylist",
|
|
||||||
"set_public_key",
|
|
||||||
"request_public_key",
|
|
||||||
"set_secret_key",
|
|
||||||
"request_secret_key",
|
|
||||||
]
|
|
||||||
|
|
||||||
def __init__(self, client: Client):
|
|
||||||
BaseModule.__init__(self, client)
|
|
||||||
|
|
||||||
self.handlers = [
|
|
||||||
StanzaHandler(
|
|
||||||
name="message",
|
|
||||||
callback=self.decrypt_message,
|
|
||||||
ns=Namespace.OPENPGP,
|
|
||||||
priority=9,
|
|
||||||
),
|
|
||||||
]
|
|
||||||
|
|
||||||
self._register_pubsub_handler(self._keylist_notification_received)
|
|
||||||
|
|
||||||
self.own_jid = self._client.get_own_jid()
|
|
||||||
|
|
||||||
own_bare_jid = self.own_jid.bare
|
|
||||||
path = Path(configpaths.get("MY_DATA")) / "openpgp" / own_bare_jid
|
|
||||||
if not path.exists():
|
|
||||||
path.mkdir(mode=0o700, parents=True)
|
|
||||||
|
|
||||||
self._pgp = PGPBackend(own_bare_jid, path)
|
|
||||||
self._storage = Storage(path)
|
|
||||||
self._contacts = PGPContacts(self._pgp, self._storage)
|
|
||||||
self._fingerprint, self._date = self.get_own_key_details()
|
|
||||||
log.info("Own Fingerprint at start: %s", self._fingerprint)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def secret_key_available(self) -> bool:
|
|
||||||
return self._fingerprint is not None
|
|
||||||
|
|
||||||
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) -> None:
|
|
||||||
self._pgp.generate_key()
|
|
||||||
|
|
||||||
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: 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: Task) -> None:
|
|
||||||
fingerprint = task.get_user_data()
|
|
||||||
try:
|
|
||||||
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: list[PGPKeyMetadata] | None = None) -> None:
|
|
||||||
if keylist is None:
|
|
||||||
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, _client: nbxmppClient, _stanza: Node, properties: MessageProperties
|
|
||||||
) -> None:
|
|
||||||
assert properties.pubsub_event is not None
|
|
||||||
|
|
||||||
if properties.pubsub_event.retracted:
|
|
||||||
return
|
|
||||||
|
|
||||||
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: 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: Task) -> None:
|
|
||||||
jid = cast(JID, task.get_user_data())
|
|
||||||
try:
|
|
||||||
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:
|
|
||||||
self.set_keylist()
|
|
||||||
return
|
|
||||||
|
|
||||||
log.info("Keylist received from %s", jid)
|
|
||||||
self._process_keylist(keylist, 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)
|
|
||||||
if self.own_jid.bare_match(from_jid) and self._fingerprint is not None:
|
|
||||||
self.set_keylist()
|
|
||||||
return
|
|
||||||
|
|
||||||
if self.own_jid.bare_match(from_jid):
|
|
||||||
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:
|
|
||||||
assert self._date is not None
|
|
||||||
keylist.append(
|
|
||||||
PGPKeyMetadata(self.own_jid, self._fingerprint, self._date)
|
|
||||||
)
|
|
||||||
self.set_keylist(keylist)
|
|
||||||
return
|
|
||||||
|
|
||||||
missing_pub_keys = self._contacts.process_keylist(from_jid, keylist)
|
|
||||||
|
|
||||||
for key in keylist:
|
|
||||||
log.info(key.fingerprint)
|
|
||||||
|
|
||||||
for fingerprint in missing_pub_keys:
|
|
||||||
self.request_public_key(from_jid, fingerprint)
|
|
||||||
|
|
||||||
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
|
|
||||||
|
|
||||||
try:
|
|
||||||
payload, fingerprint = self._pgp.decrypt(properties.openpgp)
|
|
||||||
except DecryptionFailed as error:
|
|
||||||
log.warning(error)
|
|
||||||
return
|
|
||||||
|
|
||||||
signcrypt = Node(node=payload)
|
|
||||||
|
|
||||||
try:
|
|
||||||
payload, recipients, _timestamp = parse_signcrypt(signcrypt)
|
|
||||||
except StanzaMalformed as error:
|
|
||||||
log.warning("Decryption failed: %s", error)
|
|
||||||
log.warning(payload)
|
|
||||||
return
|
|
||||||
|
|
||||||
if not any(map(self.own_jid.bare_match, recipients)):
|
|
||||||
log.warning("to attr not valid")
|
|
||||||
log.warning(signcrypt)
|
|
||||||
return
|
|
||||||
|
|
||||||
keys = self._contacts.get_keys(remote_jid)
|
|
||||||
fingerprints = [key.fingerprint for key in keys]
|
|
||||||
if fingerprint not in fingerprints:
|
|
||||||
log.warning("Invalid fingerprint on message: %s", fingerprint)
|
|
||||||
log.warning("Expected: %s", fingerprints)
|
|
||||||
return
|
|
||||||
|
|
||||||
log.info("Received OpenPGP message from: %s", properties.jid)
|
|
||||||
prepare_stanza(stanza, payload)
|
|
||||||
|
|
||||||
trust = self._contacts.get_trust(remote_jid, fingerprint)
|
|
||||||
|
|
||||||
properties.encrypted = EncryptionData(
|
|
||||||
protocol=ENCRYPTION_NAME, key=fingerprint, trust=trust
|
|
||||||
)
|
|
||||||
|
|
||||||
raise StanzaDecrypted
|
|
||||||
|
|
||||||
def encrypt_message(
|
|
||||||
self, message: OutgoingMessage, callback: Callable[[OutgoingMessage], None]
|
|
||||||
) -> None:
|
|
||||||
remote_jid = message.contact.jid
|
|
||||||
|
|
||||||
keys = self._contacts.get_keys(remote_jid)
|
|
||||||
if not keys:
|
|
||||||
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 += [
|
|
||||||
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:
|
|
||||||
log.error("Error: %s", error)
|
|
||||||
text = message.get_text(with_fallback=False) or ""
|
|
||||||
app.ged.raise_event(
|
|
||||||
MessageNotSent(
|
|
||||||
client=self._client,
|
|
||||||
jid=str(remote_jid),
|
|
||||||
message=text,
|
|
||||||
error=error,
|
|
||||||
time=time.time(),
|
|
||||||
)
|
|
||||||
)
|
|
||||||
return
|
|
||||||
|
|
||||||
assert encrypted_payload is not None
|
|
||||||
|
|
||||||
create_message_stanza(
|
|
||||||
message.get_stanza(), encrypted_payload, bool(message.get_text())
|
|
||||||
)
|
|
||||||
|
|
||||||
message.set_encryption(
|
|
||||||
EncryptionData(
|
|
||||||
protocol=ENCRYPTION_NAME, key="Unknown", trust=Trust.VERIFIED
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
callback(message)
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
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)
|
|
||||||
stanzastr = stanzastr[0:-1]
|
|
||||||
log.debug(stanzastr)
|
|
||||||
log.debug("-" * 15)
|
|
||||||
|
|
||||||
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) -> None:
|
|
||||||
self.set_keylist()
|
|
||||||
|
|
||||||
def cleanup(self) -> None:
|
|
||||||
self._storage.cleanup()
|
|
||||||
del self._pgp
|
|
||||||
del self._contacts
|
|
||||||
|
|
||||||
|
|
||||||
def get_instance(*args: Any, **kwargs: Any) -> tuple[Any, str]:
|
|
||||||
return OpenPGP(*args, **kwargs), "OpenPGP"
|
|
||||||
@@ -1,71 +0,0 @@
|
|||||||
# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>
|
|
||||||
#
|
|
||||||
# This file is part of the OpenPGP Gajim Plugin.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation; version 3 only.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin 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 General Public License for more details.
|
|
||||||
#
|
|
||||||
# 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 enum import IntEnum
|
|
||||||
|
|
||||||
from nbxmpp import Node
|
|
||||||
from nbxmpp.namespaces import Namespace
|
|
||||||
|
|
||||||
ENCRYPTION_NAME = "OpenPGP"
|
|
||||||
|
|
||||||
NOT_ENCRYPTED_TAGS = [
|
|
||||||
("no-store", Namespace.HINTS),
|
|
||||||
("store", Namespace.HINTS),
|
|
||||||
("no-copy", Namespace.HINTS),
|
|
||||||
("no-permanent-store", Namespace.HINTS),
|
|
||||||
("origin-id", Namespace.SID),
|
|
||||||
("thread", ""),
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
class Trust(IntEnum):
|
|
||||||
NOT_TRUSTED = 0
|
|
||||||
UNKNOWN = 1
|
|
||||||
BLIND = 2
|
|
||||||
VERIFIED = 3
|
|
||||||
|
|
||||||
|
|
||||||
def prepare_stanza(stanza: Node, payload: list[Node | str]) -> None:
|
|
||||||
delete_nodes(stanza, "openpgp", Namespace.OPENPGP)
|
|
||||||
delete_nodes(stanza, "body")
|
|
||||||
|
|
||||||
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: Node, name: str, namespace: str | None = None) -> None:
|
|
||||||
attrs = None
|
|
||||||
if namespace is not None:
|
|
||||||
attrs = {"xmlns": Namespace.OPENPGP}
|
|
||||||
nodes = stanza.getTags(name, attrs)
|
|
||||||
for node in nodes:
|
|
||||||
stanza.delChild(node)
|
|
||||||
|
|
||||||
|
|
||||||
class VerifyFailed(Exception):
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
class DecryptionFailed(Exception):
|
|
||||||
pass
|
|
||||||
@@ -1,12 +0,0 @@
|
|||||||
<?xml version="1.0" encoding="UTF-8"?>
|
|
||||||
<component type="addon">
|
|
||||||
<id>org.gajim.Gajim.Plugin.openpgp</id>
|
|
||||||
<extends>org.gajim.Gajim</extends>
|
|
||||||
<name>OpenPGP Plugin</name>
|
|
||||||
<summary>Experimental OpenPGP XEP-0373 Implementation</summary>
|
|
||||||
<url type="homepage">https://gajim.org/</url>
|
|
||||||
<metadata_license>CC-BY-SA-3.0</metadata_license>
|
|
||||||
<project_license>GPL-3.0-only</project_license>
|
|
||||||
<update_contact>gajim-devel_AT_gajim.org</update_contact>
|
|
||||||
</component>
|
|
||||||
|
|
||||||
@@ -1,214 +0,0 @@
|
|||||||
# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>
|
|
||||||
#
|
|
||||||
# This file is part of the OpenPGP Gajim Plugin.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation; version 3 only.
|
|
||||||
#
|
|
||||||
# OpenPGP Gajim Plugin 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 General Public License for more details.
|
|
||||||
#
|
|
||||||
# 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 __future__ import annotations
|
|
||||||
|
|
||||||
from typing import Any
|
|
||||||
from typing import TYPE_CHECKING
|
|
||||||
|
|
||||||
import logging
|
|
||||||
from collections.abc import Callable
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from gi.repository import Gdk
|
|
||||||
from gi.repository import GLib
|
|
||||||
from gi.repository import Gtk
|
|
||||||
from nbxmpp import JID
|
|
||||||
from nbxmpp.namespaces import Namespace
|
|
||||||
|
|
||||||
from gajim.common import app
|
|
||||||
from gajim.common import configpaths
|
|
||||||
from gajim.common import ged
|
|
||||||
from gajim.common.client import Client
|
|
||||||
from gajim.common.const import CSSPriority
|
|
||||||
from gajim.common.events import SignedIn
|
|
||||||
from gajim.common.structs import OutgoingMessage
|
|
||||||
from gajim.gtk.alert import InformationAlertDialog
|
|
||||||
from gajim.gtk.control import ChatControl
|
|
||||||
from gajim.plugins import GajimPlugin
|
|
||||||
from gajim.plugins.plugins_i18n import _
|
|
||||||
|
|
||||||
from openpgp.modules.util import ENCRYPTION_NAME
|
|
||||||
|
|
||||||
try:
|
|
||||||
from openpgp.modules import openpgp
|
|
||||||
except (ImportError, OSError) as e:
|
|
||||||
error_msg = str(e)
|
|
||||||
else:
|
|
||||||
error_msg = None
|
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
|
||||||
from openpgp.modules.openpgp import OpenPGP
|
|
||||||
|
|
||||||
log = logging.getLogger("gajim.p.openpgp")
|
|
||||||
|
|
||||||
|
|
||||||
class OpenPGPPlugin(GajimPlugin):
|
|
||||||
def init(self):
|
|
||||||
if error_msg:
|
|
||||||
self.activatable = False
|
|
||||||
self.available_text = error_msg
|
|
||||||
self.config_dialog = None
|
|
||||||
return
|
|
||||||
|
|
||||||
self.events_handlers = {
|
|
||||||
"signed-in": (ged.PRECORE, self._on_signed_in),
|
|
||||||
}
|
|
||||||
|
|
||||||
self.modules = [openpgp] # type: ignore
|
|
||||||
|
|
||||||
self.encryption_name = ENCRYPTION_NAME
|
|
||||||
self.config_dialog = None
|
|
||||||
self.gui_extension_points = {
|
|
||||||
"encrypt" + self.encryption_name: (self._encrypt_message, None),
|
|
||||||
"send_message" + self.encryption_name: (self._before_sendmessage, None),
|
|
||||||
"encryption_dialog" + self.encryption_name: (
|
|
||||||
self._on_encryption_button_clicked,
|
|
||||||
None,
|
|
||||||
),
|
|
||||||
"encryption_state" + self.encryption_name: (
|
|
||||||
self._get_encryption_state,
|
|
||||||
None,
|
|
||||||
),
|
|
||||||
"update_caps": (self._update_caps, None),
|
|
||||||
}
|
|
||||||
|
|
||||||
self._create_paths()
|
|
||||||
self._load_css()
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def get_openpgp_module(account: str) -> OpenPGP:
|
|
||||||
return app.get_client(account).get_module("OpenPGP") # pyright: ignore
|
|
||||||
|
|
||||||
def _load_css(self) -> None:
|
|
||||||
path = Path(__file__).parent / "gtk" / "style.css"
|
|
||||||
try:
|
|
||||||
with path.open("r") as f:
|
|
||||||
css = f.read()
|
|
||||||
except Exception as exc:
|
|
||||||
log.error("Error loading css: %s", exc)
|
|
||||||
return
|
|
||||||
|
|
||||||
display = Gdk.Display.get_default()
|
|
||||||
assert display is not None
|
|
||||||
|
|
||||||
try:
|
|
||||||
provider = Gtk.CssProvider()
|
|
||||||
provider.load_from_bytes(GLib.Bytes.new(css.encode("utf-8")))
|
|
||||||
Gtk.StyleContext.add_provider_for_display(
|
|
||||||
display, provider, CSSPriority.DEFAULT_THEME
|
|
||||||
)
|
|
||||||
except Exception:
|
|
||||||
log.exception("Error loading application css")
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _create_paths() -> None:
|
|
||||||
keyring_path = Path(configpaths.get("MY_DATA")) / "openpgp"
|
|
||||||
if not keyring_path.exists():
|
|
||||||
keyring_path.mkdir()
|
|
||||||
|
|
||||||
def _on_signed_in(self, event: SignedIn) -> None:
|
|
||||||
openpgp = self.get_openpgp_module(event.account)
|
|
||||||
if openpgp.secret_key_available:
|
|
||||||
log.info(
|
|
||||||
"%s => Publish keylist and public key after sign in", event.account
|
|
||||||
)
|
|
||||||
openpgp.request_keylist()
|
|
||||||
openpgp.set_public_key()
|
|
||||||
|
|
||||||
def activate(self) -> None:
|
|
||||||
for account in app.settings.get_active_accounts():
|
|
||||||
client = app.get_client(account)
|
|
||||||
client.get_module("Caps").update_caps()
|
|
||||||
if app.account_is_connected(account):
|
|
||||||
openpgp = self.get_openpgp_module(account)
|
|
||||||
if openpgp.secret_key_available:
|
|
||||||
log.info(
|
|
||||||
"%s => Publish keylist and public key after plugin activation",
|
|
||||||
account,
|
|
||||||
)
|
|
||||||
openpgp.request_keylist()
|
|
||||||
openpgp.set_public_key()
|
|
||||||
|
|
||||||
def deactivate(self) -> None:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def activate_encryption(self, chat_control: ChatControl) -> bool:
|
|
||||||
account = chat_control.account
|
|
||||||
assert chat_control.contact is not None
|
|
||||||
jid = chat_control.contact.jid
|
|
||||||
openpgp = self.get_openpgp_module(account)
|
|
||||||
if openpgp.secret_key_available:
|
|
||||||
keys = openpgp.get_keys(jid, only_trusted=False)
|
|
||||||
if not keys:
|
|
||||||
openpgp.request_keylist(JID.from_string(jid))
|
|
||||||
return True
|
|
||||||
|
|
||||||
from openpgp.gtk.wizard import KeyWizard
|
|
||||||
|
|
||||||
KeyWizard(self, account, chat_control)
|
|
||||||
return False
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _update_caps(_account: str, features: list[str]) -> None:
|
|
||||||
features.append("%s+notify" % Namespace.OPENPGP_PK)
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _get_encryption_state(
|
|
||||||
_chat_control: ChatControl, state: dict[str, Any]
|
|
||||||
) -> None:
|
|
||||||
state["authenticated"] = True
|
|
||||||
state["visible"] = True
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _on_encryption_button_clicked(chat_control: ChatControl) -> None:
|
|
||||||
account = chat_control.account
|
|
||||||
assert chat_control.contact is not None
|
|
||||||
jid = chat_control.contact.jid
|
|
||||||
|
|
||||||
from openpgp.gtk.key import KeyDialog
|
|
||||||
|
|
||||||
KeyDialog(account, jid, app.window)
|
|
||||||
|
|
||||||
def _before_sendmessage(self, chat_control: ChatControl) -> None:
|
|
||||||
account = chat_control.account
|
|
||||||
assert chat_control.contact is not None
|
|
||||||
jid = chat_control.contact.jid
|
|
||||||
openpgp = self.get_openpgp_module(account)
|
|
||||||
|
|
||||||
if not openpgp.secret_key_available:
|
|
||||||
from openpgp.gtk.wizard import KeyWizard
|
|
||||||
|
|
||||||
KeyWizard(self, account, chat_control)
|
|
||||||
return
|
|
||||||
|
|
||||||
keys = openpgp.get_keys(jid)
|
|
||||||
if not keys:
|
|
||||||
InformationAlertDialog(
|
|
||||||
_("Not Trusted"), _("There was no trusted and active key found")
|
|
||||||
)
|
|
||||||
chat_control.sendmessage = False
|
|
||||||
|
|
||||||
def _encrypt_message(
|
|
||||||
self,
|
|
||||||
client: Client,
|
|
||||||
message: OutgoingMessage,
|
|
||||||
callback: Callable[[OutgoingMessage], None],
|
|
||||||
) -> None:
|
|
||||||
openpgp = self.get_openpgp_module(client.account)
|
|
||||||
if not openpgp.secret_key_available:
|
|
||||||
return
|
|
||||||
openpgp.encrypt_message(message, callback)
|
|
||||||
@@ -1,21 +0,0 @@
|
|||||||
{
|
|
||||||
"authors": [
|
|
||||||
"Philipp Hörist <philipp@hoerist.com>"
|
|
||||||
],
|
|
||||||
"description": "Experimental OpenPGP (XEP-0373) implementation.",
|
|
||||||
"homepage": "https://dev.gajim.org/gajim/gajim-plugins/wikis/OpenPGPplugin",
|
|
||||||
"config_dialog": false,
|
|
||||||
"name": "OpenPGP",
|
|
||||||
"platforms": [
|
|
||||||
"others",
|
|
||||||
"linux",
|
|
||||||
"darwin",
|
|
||||||
"win32"
|
|
||||||
],
|
|
||||||
"requirements": [
|
|
||||||
"gajim>=2.4.2",
|
|
||||||
"gajim<=2.4.5"
|
|
||||||
],
|
|
||||||
"short_name": "openpgp",
|
|
||||||
"version": "1.8.2"
|
|
||||||
}
|
|
||||||
@@ -20,7 +20,6 @@ PLUGIN_DIRS = [
|
|||||||
"length_notifier",
|
"length_notifier",
|
||||||
"message_box_size",
|
"message_box_size",
|
||||||
"now_listen",
|
"now_listen",
|
||||||
"openpgp",
|
|
||||||
"pgp",
|
"pgp",
|
||||||
"plugins_translations",
|
"plugins_translations",
|
||||||
"quick_replies",
|
"quick_replies",
|
||||||
|
|||||||
Reference in New Issue
Block a user