WIP: parakeeet

.gitlab-ci.yml

@@ -24,6 +24,17 @@ test-pyright:
     - pyright
   interruptible: true
 
+test-isort:
+  image: gajim-test
+  stage: test
+  rules:
+    - changes:
+      - "**/*.py"
+  script:
+    - isort --version
+    - isort --check .
+  interruptible: true
+
 test-ruff-lint:
   image: gajim-test
   stage: test

.pre-commit-config.yaml

@@ -1,12 +1,10 @@
 repos:
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: v0.15.7
+    rev: v0.14.8
     hooks:
       - id: ruff-check
         exclude: ".githooks/"
       - id: ruff-format
-      - id: ruff-check
-        args: [--select, I, --fix]
 
   - repo: https://github.com/codespell-project/codespell
     rev: v2.4.1

flatpak/org.gajim.Gajim.Plugin.openpgp.yaml

@@ -0,0 +1,57 @@
+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

openpgp/__init__.py

@@ -0,0 +1 @@
+from .pgpplugin import OpenPGPPlugin  # pyright: ignore # noqa: F401

openpgp/backend/base.py

@@ -0,0 +1,95 @@
+# 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

openpgp/backend/gpgme.py

@@ -0,0 +1,206 @@
+# 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

openpgp/backend/gpgme_types.py

@@ -0,0 +1,70 @@
+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]

openpgp/backend/pygpg.py

@@ -0,0 +1,177 @@
+# 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)

openpgp/backend/sql.py

@@ -0,0 +1,125 @@
+# 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()

openpgp/backend/util.py

@@ -0,0 +1,12 @@
+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)

openpgp/gtk/key.py

@@ -0,0 +1,250 @@
+# 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)

openpgp/gtk/style.css

@@ -0,0 +1,14 @@
+.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; }

openpgp/gtk/wizard.py

@@ -0,0 +1,252 @@
+# 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

openpgp/modules/key_store.py

@@ -0,0 +1,269 @@
+# 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

openpgp/modules/openpgp.py

@@ -0,0 +1,369 @@
+# 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"

openpgp/modules/util.py

@@ -0,0 +1,71 @@
+# 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

openpgp/org.gajim.Gajim.Plugin.openpgp.metainfo.xml

@@ -1,11 +1,12 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <component type="addon">
-  <id>org.gajim.Gajim.Plugin.stt_voice_messages</id>
+  <id>org.gajim.Gajim.Plugin.openpgp</id>
   <extends>org.gajim.Gajim</extends>
-  <name>STT Voice Messages Plugin</name>
-  <summary>Transcribes voice messages to text</summary>
+  <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>
+

openpgp/pgpplugin.py

@@ -0,0 +1,214 @@
+# 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)

openpgp/plugin-manifest.json

@@ -0,0 +1,20 @@
+{
+    "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"
+    ],
+    "short_name": "openpgp",
+    "version": "1.8.1"
+}

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "gajim-plugins"
-requires-python = ">=3.12"
+requires-python = ">=3.11"
 license = {text = "GPL-3.0-or-later"}
 version = "0.0.1"
 
@@ -8,6 +8,7 @@ version = "0.0.1"
 dev = [
     "codespell==2.4.1",
     "gajim@git+https://dev.gajim.org/gajim/gajim.git",
+    "isort==7.0.0",
     "nbxmpp@git+https://dev.gajim.org/gajim/python-nbxmpp.git",
     "pre-commit",
     "pygobject-stubs@git+https://github.com/pygobject/pygobject-stubs.git",
@@ -16,12 +17,17 @@ dev = [
     "ruff==0.14.8",
 ]
 
+[tool.black]
+line-length = 88
+target-version = ["py310", "py311", "py312"]
+required-version = "24.10.0"
+
 [tool.codespell]
 skip = "*__pycache__*,build,dist,test,./acronyms_expander/acronyms.py,.egg-info,.git,*.po,*.po~,*.pot,*.nsi,*.spec,*.svg"
 ignore-words-list = "THIRDPARTY,Toi,fpr"
 
 [tool.pyright]
-pythonVersion = "3.12"
+pythonVersion = "3.10"
 pythonPlatform = "All"
 typeCheckingMode = "strict"
 
@@ -37,9 +43,10 @@ exclude = [
 
 [tool.ruff]
 line-length = 88
-target-version = "py312"
+target-version = "py310"
 
 [tool.ruff.lint]
+
 select = [
     "A",     # flake8-builtins
     # "AIR",   # Airflow
@@ -65,7 +72,7 @@ select = [
     # "FIX",   # flake8-fixme
     # "FLY",   # flynt
     "G",     # flake8-logging-format
-    "I",     # isort
+    # "I",     # isort
     "ICN",   # flake8-import-conventions
     # "INP",   # flake8-no-pep420
     "INT",   # flake8-gettext
@@ -171,20 +178,20 @@ dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
 [tool.ruff.lint.mccabe]
 max-complexity = 15
 
-[tool.ruff.lint.isort]
-case-sensitive = false
-order-by-type = false
-force-single-line = true
-section-order = [
-  "future",
-  "typing",
-  "standard-library",
-  "third-party",
-  "gajim",
-  "first-party",
-  "local-folder"
+[tool.isort]
+profile = "black"
+force_alphabetical_sort_within_sections = true
+force_single_line = true
+group_by_package = true
+known_typing = ["typing"]
+known_gajim = ["gajim"]
+sections = [
+  "FUTURE",
+  "TYPING",
+  "STDLIB",
+  "THIRDPARTY",
+  "GAJIM",
+  "FIRSTPARTY",
+  "LOCALFOLDER"
 ]
-
-[tool.ruff.lint.isort.sections]
-gajim = ["gajim"]
-typing = ["_typeshed", "typing", "typing_extensions"]
+skip_gitignore = true

scripts/update_translations.py

@@ -20,6 +20,7 @@ PLUGIN_DIRS = [
     "length_notifier",
     "message_box_size",
     "now_listen",
+    "openpgp",
     "pgp",
     "plugins_translations",
     "quick_replies",

stt_voice_messages/README.md

@@ -0,0 +1,49 @@
+# About
+
+This plugin allows you in conjuction with a _general-purpose speech recognition model_ to transcribe your voice messages to text.
+
+In order to make use of this plugin, you need to have at least one of the following models installed:
+
+#### OpenAI Whisper
+- Website: https://github.com/openai/whisper
+- Installable by: `pip install -U openai-whisper`
+
+#### Faster Whisper
+- Website: https://github.com/SYSTRAN/faster-whisper
+- Installable by: `pip install -U faster-whisper`
+
+Additionally you have to checkout the following Gajim branch:
+https://dev.gajim.org/mesonium/gajim/-/tree/stt_voice_messages
+
+# Hint
+
+_**The plugin is very much POC at this stage!**_
+
+Currently a chosen model will be on first downloaded in the background, during which 
+Gajim's UI may not respond.
+
+Typical model sizes are in case of OpenAI Whisper:
+
+| Multi Langual Model | Download Size | 
+|---------------------|---------------| 
+| Tiny                | 70 MB         |
+| Base                | 140 MB        | 
+| Small               | 460 MB        |
+| Medium              | 1.4 GB        | 
+| Large               | 2.9 GB        | 
+
+# TODO
+
+- [x] Offer multiple models
+- [ ] Add various model settings
+- [ ] Model receiving
+  - [ ] Hint model download state
+  - [ ] Allow to change model download location
+  - [ ] Allow to use local models
+- [ ] Database Handling
+  - [ ] Store transcribed messages in a DB
+  - [ ] Option to delete DB
+- [ ] Update UI
+  - [ ] Make it more pretty
+  - [ ] Show progress bar
+  - [ ] Highlight words on playback

stt_voice_messages/__init__.py

@@ -1 +1 @@
-from .stt_voice_messages import STTVoiceMessagesPlugin  # pyright: ignore  # noqa: F401
+from .stt_voice_messages import STTVoiceMessagesPlugin  # type: ignore  # noqa: F401

stt_voice_messages/gtk/config_dialog.py

@@ -15,86 +15,277 @@
 
 from __future__ import annotations
 
-from typing import Any
-from typing import TYPE_CHECKING
+import logging
+import typing
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
 
-from gi.repository import Gtk
+try:
+    import onnx_asr
+except ModuleNotFoundError:
+    if typing.TYPE_CHECKING:
+        import onnx_asr
 
-from gajim.gtk.const import Setting
-from gajim.gtk.const import SettingKind
-from gajim.gtk.const import SettingType
-from gajim.gtk.settings import SettingsDialog
+from gi.repository import Adw, Gtk
+
+from gajim.gtk.const import Setting, SettingKind, SettingType
+from gajim.gtk.filechoosers import Filter
+from gajim.gtk.settings import GajimPreferencesGroup, SettingsDialog
 from gajim.plugins.plugins_i18n import _
 
+from ..models import stt
+from ..models.model_settings import OnnxAsrSettings
+
 if TYPE_CHECKING:
     from ..stt_voice_messages import STTVoiceMessagesPlugin
 
+log = logging.getLogger('gajim.p.sttvm_config_dialog')
+
+
+class Configuration:
+    def __init__(self, plugin: STTVoiceMessagesPlugin):
+        self._plugin = plugin
+        self._instance = None
+        self._main_model_row = None
+        self._preset_model_picker = None
+        self._custom_model_id_entry = None
+        self._local_model_file_picker = None
+        self._status_group = None
+        self._model_data: dict[str, str] = {}
+        self._instance = stt.OnnxAsrModel()
+        self._instance.set_config(OnnxAsrSettings(
+            model_id=self.plugin.config['model_id'],
+            model_path=self.plugin.config['model_path']
+        ))
+        self._model_data = self._steal_model_list()
+
+    @property
+    def plugin(self) -> STTVoiceMessagesPlugin:
+        return self._plugin
+
+    @property
+    def is_available(self) -> bool:
+        return self._instance is not None
+
+    def unload_model(self) -> None:
+        if self._instance is not None:
+            self._instance.unload_now()
+
+    def _steal_model_list(self) -> dict[str, str]:
+        # UGLY: Extract available model choices from onnx_asr type hints.
+        ann = onnx_asr.load_model.__annotations__.get('model')
+        return {
+            v: v for arg in typing.get_args(ann)
+            for v in typing.get_args(arg)
+            if isinstance(v, str)
+        }
+
+    def on_setting(self, value: Any, data: Any) -> None:
+        if isinstance(value, str):
+            value = value.strip()
+        self.plugin.config[data] = value
+
+    def on_preset_changed(self, value: str, data: Any) -> None:
+        if self._custom_model_id_entry is not None:
+            entry_text = self._custom_model_id_entry.entry.get_text().strip()
+            if entry_text:
+                self._update_model_status()
+                return  # custom entry overrides; ignore preset change
+        self._write_model_id(value)
+        self._update_model_status()
+
+    def on_custom_model_id_changed(self, value: str, data: Any) -> None:
+        value = value.strip()
+        if value:
+            self._write_model_id(value)
+        elif self._preset_model_picker is not None:
+            preset_key = self._preset_model_picker._dropdown.get_selected_key()
+            if preset_key is not None:
+                self._write_model_id(preset_key)
+        self._apply_sensitivity_state()
+        self._update_model_status()
+
+    def on_model_file_picked(self, value: str, data: Any) -> None:
+        self._write_model_path(str(Path(value).parent) if value else '')
+        self._apply_sensitivity_state()
+        self._update_model_status()
+
+    def _write_model_id(self, model_id: str) -> None:
+        if self.plugin.config['model_id'] == model_id:
+            return
+        self.plugin.config['model_id'] = model_id
+        if self._instance is not None:
+            self._instance.set_config(OnnxAsrSettings(
+                model_id=self.plugin.config['model_id'],
+                model_path=self.plugin.config['model_path']
+            ))
+
+    def _write_model_path(self, model_path: str) -> None:
+        if self.plugin.config['model_path'] == model_path:
+            return
+        self.plugin.config['model_path'] = model_path
+        if self._instance is not None:
+            self._instance.set_config(OnnxAsrSettings(
+                model_id=self.plugin.config['model_id'],
+                model_path=self.plugin.config['model_path']
+            ))
+
+    def sync_model_path_from_widget(self) -> None:
+        if self._local_model_file_picker is None:
+            return
+        button = self._local_model_file_picker.get_activatable_widget()
+        path = button.get_path()
+        new_path = str(path.parent) if path else ''
+        self._write_model_path(new_path)
+
+    def _apply_sensitivity_state(self) -> None:
+        if self._preset_model_picker is None:
+            return
+        has_local = bool(self.plugin.config['model_path'])
+        entry_text = (self._custom_model_id_entry.entry.get_text().strip()
+                      if self._custom_model_id_entry else '')
+        has_entry = bool(entry_text)
+        self._custom_model_id_entry.set_sensitive(not has_local)
+        self._preset_model_picker.set_sensitive(not has_local and not has_entry)
+
+    def _update_model_status(self) -> None:
+        if self._main_model_row is None:
+            return
+        entry_text = (self._custom_model_id_entry.entry.get_text().strip()
+                      if self._custom_model_id_entry else '')
+
+        if self.plugin.config['model_path']:
+            path = Path(self.plugin.config['model_path'])
+            summary = _('Local: {}').format(path.name or str(path))
+            description = _('Loading model files from {}').format(path)
+            if not (path / 'config.json').exists():
+                description += '\n' + _(
+                    'config.json not found in this directory — onnx-asr will'
+                    ' fall back to Model preset or Custom Model ID for the'
+                    ' architecture.')
+        elif entry_text:
+            summary = _('Custom: {}').format(entry_text)
+            description = _('Using custom model: {}').format(entry_text)
+        else:
+            preset_key = (self._preset_model_picker._dropdown.get_selected_key()
+                          if self._preset_model_picker else '')
+            summary = preset_key or _('(none)')
+            description = (_('Using preset: {}').format(preset_key)
+                           if preset_key else '')
+
+        self._main_model_row._label.set_text(summary)
+        if self._status_group is not None:
+            self._status_group.set_description(description)
+
 
 class STTVoiceMessagesConfigDialog(SettingsDialog):
-    def __init__(self, plugin: STTVoiceMessagesPlugin, parent: Gtk.Window) -> None:
+    def __init__(self, config: Configuration, parent: Gtk.Window) -> None:
+        self.config = config
+        self.plugin = self.config.plugin
+        if not config.is_available:
+            return
 
-        self.plugin = plugin
-        model = plugin.model
-
-        if not model.available():
-            status = _('onnx-asr is not installed. Run: pip install "onnx-asr[hub]"')
-        elif model.is_loaded:
-            status = _("Model is loaded.")
-        elif model.will_download:
-            status = _("Model files will be downloaded on first use.")
-        else:
-            status = _("Model files are downloaded and ready.")
-
-        settings = [
-            Setting(
-                SettingKind.SWITCH,
-                _("Transcribe automatically"),
-                SettingType.VALUE,
-                bool(plugin.config["auto_transcribe"]),
-                callback=self._on_setting,
-                data="auto_transcribe",
-                desc=_("Transcribe voice messages as soon as they are displayed"),
-            ),
-            Setting(
-                SettingKind.ENTRY,
-                _("Model"),
-                SettingType.VALUE,
-                str(plugin.config["model_id"]),
-                callback=self._on_setting,
-                data="model_id",
-                desc=_("onnx-asr model name or Hugging Face repository"),
-            ),
-            Setting(
-                SettingKind.DROPDOWN,
-                _("Quantization"),
-                SettingType.VALUE,
-                str(plugin.config["quantization"]),
-                callback=self._on_setting,
-                data="quantization",
-                props={
-                    "data": {
-                        "int8": _("int8 (fast, ~600 MB download)"),
-                        "fp32": _("Full precision (~2.4 GB download)"),
-                    }
-                },
-                desc=_("Applied the next time the model is loaded"),
-            ),
-            Setting(SettingKind.GENERIC, _("Status"), SettingType.VALUE, desc=status),
+        rows = [
+            Setting(SettingKind.SWITCH,
+                    _('Auto Transcribe'),
+                    SettingType.VALUE,
+                    value=self.plugin.config['auto_transcribe'],
+                    data='auto_transcribe',
+                    callback=config.on_setting,
+                    desc=_('Transcribe messages as they appear')),
+            Setting(SettingKind.SUBPAGE,
+                    _('Model'),
+                    SettingType.VALUE,
+                    value=None,
+                    name='main_model',
+                    props={'subpage': 'sttvm-model'}),
         ]
 
         SettingsDialog.__init__(
             self,
             parent,
-            _("STT Voice Messages Configuration"),
+            _('STT Voice Messages'),
             Gtk.DialogFlags.MODAL,
-            settings,
-            "",
+            rows,
+            '',
         )
 
-    def _on_setting(self, value: Any, data: Any) -> None:
-        self.plugin.config[data] = value
-        if data in ("model_id", "quantization"):
-            self.plugin.model.set_config(
-                str(self.plugin.config["model_id"]),
-                str(self.plugin.config["quantization"]),
-            )
+        config._main_model_row = self.get_setting('main_model')
+
+        use_custom = self.plugin.config['model_id'] not in config._model_data
+
+
+        subpage_rows: list[Setting] = [
+            Setting(SettingKind.DROPDOWN,
+                    _('Model'),
+                    SettingType.VALUE,
+                    value=self.plugin.config['model_id'],
+                    name='preset_model',
+                    callback=config.on_preset_changed,
+                    props={'data': config._model_data}),
+            Setting(SettingKind.ENTRY,
+                    _('Custom Model'),
+                    SettingType.VALUE,
+                    value=self.plugin.config['model_id'] if use_custom else '',
+                    name='custom_model',
+                    callback=config.on_custom_model_id_changed,
+                    desc=_('Custom HF model path or model ID')),
+            Setting(SettingKind.FILECHOOSER,
+                    _('Local File'),
+                    SettingType.VALUE,
+                    value='',
+                    name='local_model_file',
+                    callback=config.on_model_file_picked,
+                    desc=_('Model ID is taken from config.json if not set'),
+                    props={'filefilters': [
+                        Filter(_('ONNX model'), suffixes=['onnx'], default=True),
+                    ]}),
+        ]
+
+        controls_group = GajimPreferencesGroup('model_controls')
+        for s in subpage_rows:
+            controls_group.add_setting(s)
+
+        status_group = Adw.PreferencesGroup()
+
+        pref_page = Adw.PreferencesPage()
+        pref_page.add(controls_group)
+        pref_page.add(status_group)
+
+        toolbar = Adw.ToolbarView(content=pref_page)
+        toolbar.add_top_bar(Adw.HeaderBar())
+
+        page = Adw.NavigationPage(
+            tag='sttvm-model', title=_('Model'), child=toolbar)
+        self._nav.add(page)
+
+        config._preset_model_picker = controls_group.get_setting('preset_model')
+        config._custom_model_id_entry = controls_group.get_setting('custom_model')
+        config._local_model_file_picker = controls_group.get_setting(
+            'local_model_file')
+        config._status_group = status_group
+
+        config._custom_model_id_entry.entry.set_placeholder_text(
+            _('onnx-community/whisper-large-v3-turbo'))
+
+        button = config._local_model_file_picker.get_activatable_widget()
+        button._label_text = _('.oonx')
+        button.reset()
+
+        if self.plugin.config['model_path']:
+            onnx_in_dir = next(iter(Path(self.plugin.config['model_path']).glob('*.onnx')),
+                               None)
+            if onnx_in_dir is not None:
+                button.set_path(onnx_in_dir)
+
+        config._update_model_status()
+        config._apply_sensitivity_state()
+
+    def _cleanup(self) -> None:
+        self.config.sync_model_path_from_widget()
+        self.config._main_model_row = None
+        self.config._preset_model_picker = None
+        self.config._custom_model_id_entry = None
+        self.config._local_model_file_picker = None
+        self.config._status_group = None
+        SettingsDialog._cleanup(self)

stt_voice_messages/gtk/sttbox.py

@@ -15,141 +15,82 @@
 
 from __future__ import annotations
 
-from typing import Any
-
 import logging
 from pathlib import Path
 
-from gi.repository import Adw
-from gi.repository import Gio
-from gi.repository import GLib
-from gi.repository import GObject
-from gi.repository import Gtk
+from gi.repository import Gtk, Adw
 
+from gajim.plugins.gajimplugin import GajimPluginConfig
 from gajim.plugins.plugins_i18n import _
 
-from ..model import OnnxAsrModel
-
-log = logging.getLogger("gajim.p.stt_voice_messages_sttbox")
+from .. import helper
 
+log = logging.getLogger('gajim.p.stt_voice_messages_sttbox')
 
 class STTBox(Gtk.Box):
-    def __init__(self, model: OnnxAsrModel, audio_file: Path) -> None:
-        Gtk.Box.__init__(self, orientation=Gtk.Orientation.VERTICAL, spacing=6)
-        self.set_margin_top(6)
-        self.set_halign(Gtk.Align.CENTER)
-        self.set_visible(False)
+    def __init__(self,
+                 config: GajimPluginConfig,
+                 audio_file: Path,
+                 ) -> None:
 
-        self._model = model
+        Gtk.Box.__init__(self, orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
+
+        self._config = config
+        self._model = None
         self._audio_file = audio_file
-        self._task: Gio.Task | None = None
+        self._text = ''
 
-        self._transcribe_button = Gtk.Button.new_from_icon_name(
-            "lucide-square-pen-symbolic"
-        )
-        self._transcribe_button.set_tooltip_text(_("Transcribe voice message"))
-        self._transcribe_button.set_valign(Gtk.Align.CENTER)
-        self._transcribe_button.connect("clicked", self._on_transcribe_clicked)
+        self._transcribe_button = Gtk.Button.new_from_icon_name("lucide-captions-symbolic")
+        self._transcribe_button.set_tooltip_text(_('Transcribe voice message'))
 
-        self._spinner = Adw.Spinner(valign=Gtk.Align.CENTER, visible=False)
+        self._spinner = Adw.Spinner(valign=Gtk.Align.START, visible=False)
 
-        self._transcription_label = Gtk.Label()
+        self._transcription_label = Gtk.Label(
+            label=_('Nothing transcribed yet'))
         self._transcription_label.set_max_width_chars(40)
         self._transcription_label.set_wrap(True)
-        self._transcription_label.set_xalign(0)
-        self._transcription_label.set_selectable(True)
 
         self.append(self._spinner)
         self.append(self._transcription_label)
+        self._transcribe_button.connect('clicked', self._on_transcribe_clicked)
+        self._result = helper.Results('')
 
     @property
     def button(self) -> Gtk.Button:
         return self._transcribe_button
 
-    def transcribe(self) -> None:
-        self._on_transcribe_clicked(self._transcribe_button)
-
     def _on_transcribe_clicked(self, _button: Gtk.Button) -> None:
-        if not self._model.available():
-            self._show_status(
-                _('onnx-asr is not installed. Run: pip install "onnx-asr[hub]"'),
-                busy=False,
-            )
+        log.debug('config._instance = %s', self._config._instance)
+        self._model = self._config._instance
+        if self._model is None:
             return
 
         if self._model.is_loaded:
-            text = _("Transcribing…")
+            text = _('Transcribing…')
         elif self._model.will_download:
-            text = _("Downloading model…")
+            text = _('Downloading ') + self._model.model_id
         else:
-            text = _("Loading model…")
+            text = _('Loading model…')
+        self._transcription_label.set_text(text)
+        self._spinner.set_visible(True)
+        self._task = helper.BackgroundTask(
+            self._model.load, self._on_load_done)
+        self._task.start()
 
-        self._show_status(text, busy=True)
-        self._run_bg(self._model.load, self._on_load_done)
-
-    def _on_load_done(self) -> None:
-        try:
-            self._finish()
-        except Exception as e:
-            self._show_status(_("Error: {}").format(e), busy=False)
-            return
-
-        self._show_status(_("Transcribing…"), busy=True)
-        self._run_bg(
-            lambda: self._model.recognize(self._audio_file, self._on_partial),
+    def _on_load_done(self):
+        self._transcription_label.set_text(_('Transcribing…'))
+        self._task = helper.BackgroundTask(
+            lambda: self._model.recognize(
+                self._result, helper.load_audio(self._audio_file)),
             self._show_result,
         )
+        self._task.start()
 
-    def _on_partial(self, text: str) -> None:
-        GLib.idle_add(self._set_partial, text)
-
-    def _set_partial(self, text: str) -> None:
-        self._transcription_label.set_text(text)
-
-    def _show_result(self) -> None:
-        try:
-            text = self._finish()
-        except Exception as e:
-            self._show_status(_("Error: {}").format(e), busy=False)
-            return
-
-        text = text.strip()
-        if not text:
-            text = _("No speech detected.")
-        self._show_status(text, busy=False)
-
-    def _run_bg(self, fn, on_done) -> None:
-        self._bg_fn = fn
-
-        def _cb(_source: GObject.Object, _result: Gio.AsyncResult, _data: None) -> None:
-            on_done()
-
-        self._task = Gio.Task.new(self, None, _cb, None)
-        self._task.run_in_thread(self._thread_cb)
-
-    @staticmethod
-    def _thread_cb(
-        task: Gio.Task,
-        source: "STTBox",
-        _data: None,
-        _cancel: Gio.Cancellable,
-    ) -> None:
-        try:
-            task.return_value(source._bg_fn())
-        except Exception as e:
-            task.return_value(e)
-
-    def _finish(self) -> Any:
-        task = self._task
-        self._task = None
-        value = task.propagate_value().value
-        if isinstance(value, Exception):
-            raise value
-        return value
-
-    def _show_status(self, text: str, busy: bool) -> None:
-        self._transcription_label.set_text(text)
-        self._transcription_label.set_visible(True)
-        self.set_visible(True)
-        self._spinner.set_visible(busy)
-        self._transcribe_button.set_sensitive(not busy)
+    def _show_result(self):
+        assert self._model is not  None
+        self._text = self._result.text
+        if self._text.strip() != '':
+            self._transcription_label.set_text(self._text.strip())
+        else:
+            self._transcription_label.set_text(_('_Have not heard any word!_'))
+        self._spinner.set_visible(False)

stt_voice_messages/helper.py

@@ -0,0 +1,112 @@
+# This file is part of Gajim.
+#
+# Gajim 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Gajim 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 Gajim. If not, see <http://www.gnu.org/licenses/>.
+
+import logging
+import typing
+from dataclasses import dataclass
+from pathlib import Path
+
+import gi
+import numpy as np
+from gi.repository import Gio, GObject
+
+try:
+    gi.require_version('Gst', '1.0')
+    from gi.repository import Gst
+except Exception:
+    if typing.TYPE_CHECKING:
+        from gi.repository import Gst
+
+log = logging.getLogger('gajim.p.sttvm_helper')
+
+
+@dataclass
+class Results:
+    text: str
+
+
+def load_audio(path: Path, sample_rate: int = 16000) -> np.ndarray:
+    Gst.init(None)
+    pipeline = Gst.parse_launch(
+        'filesrc name=src ! decodebin ! audioconvert ! audioresample ! '
+        f'audio/x-raw,format=F32LE,rate={sample_rate},channels=1 ! '
+        'appsink name=sink sync=false'
+    )
+    pipeline.get_by_name('src').set_property('location', str(path))
+    sink = pipeline.get_by_name('sink')
+    chunks: list[np.ndarray] = []
+
+    pipeline.set_state(Gst.State.PLAYING)
+    while (sample := sink.emit('try-pull-sample', 10 * Gst.SECOND)) is not None:
+        buf = sample.get_buffer()
+        _, info = buf.map(Gst.MapFlags.READ)
+        chunks.append(np.frombuffer(bytes(info.data), dtype=np.float32))
+        buf.unmap(info)
+    pipeline.set_state(Gst.State.NULL)
+
+    if not chunks:
+        raise RuntimeError(f'Could not decode audio: {path}')
+    return np.concatenate(chunks)
+
+
+'''
+https://discourse.gnome.org/t/gtk-threading-problem-with-glib-idle-add/13597/5
+https://github.com/gdm-settings/gdm-settings/blob/f245d3000200fa6be2a35c7f6ac45b131dadb5d6/src/utils.py#L116..L162
+'''
+
+
+class BackgroundTask(GObject.Object):
+    __gtype_name__ = 'BackgroundTask'
+
+    def __init__(self, function, finish_callback, **kwargs):
+        super().__init__(**kwargs)
+
+        self.function = function
+        self.finish_callback = finish_callback
+        self._current = None
+
+    def start(self):
+        if self._current:
+            AlreadyRunningError('Task is already running')
+
+        finish_callback = lambda self, task, nothing: self.finish_callback()
+
+        task = Gio.Task.new(self, None, finish_callback, None)
+        task.run_in_thread(self._thread_cb)
+
+        self._current = task
+
+    @staticmethod
+    def _thread_cb(task, self, task_data, cancellable):
+        try:
+            retval = self.function()
+            task.return_value(retval)
+        except Exception as e:
+            log.exception('Background task failed')
+            task.return_value(e)
+
+    def finish(self):
+        task = self._current
+        self._current = None
+
+        if not Gio.Task.is_valid(task, self):
+            raise InvalidGioTaskError()
+
+        value = task.propagate_value().value
+
+        if isinstance(value, Exception):
+            raise value
+
+        return value

stt_voice_messages/model.py

@@ -1,164 +0,0 @@
-# This file is part of Gajim.
-#
-# Gajim 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, either version 3 of the License, or
-# (at your option) any later version.
-#
-# Gajim 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 Gajim. If not, see <http://www.gnu.org/licenses/>.
-
-from __future__ import annotations
-
-from typing import Any
-
-import gc
-import logging
-import threading
-from collections.abc import Callable
-from importlib.util import find_spec
-from pathlib import Path
-
-import numpy as np
-from gi.repository import GLib
-from gi.repository import Gst
-
-log = logging.getLogger("gajim.p.sttvm_model")
-
-_IDLE_UNLOAD_SECONDS = 300
-
-
-def load_audio(path: Path, sample_rate: int = 16000) -> np.ndarray:
-    Gst.init(None)
-    pipeline = Gst.parse_launch(
-        "filesrc name=src ! decodebin ! audioconvert ! audioresample ! "
-        f"audio/x-raw,format=F32LE,rate={sample_rate},channels=1 ! "
-        "appsink name=sink sync=false"
-    )
-
-    pipeline.get_by_name("src").set_property("location", str(path))
-    sink = pipeline.get_by_name("sink")
-
-    chunks: list[np.ndarray] = []
-    try:
-        pipeline.set_state(Gst.State.PLAYING)
-        while (sample := sink.emit("try-pull-sample", 10 * Gst.SECOND)) is not None:
-            buf = sample.get_buffer()
-            _, info = buf.map(Gst.MapFlags.READ)
-            chunks.append(np.frombuffer(bytes(info.data), dtype=np.float32))
-            buf.unmap(info)
-        pipeline.set_state(Gst.State.NULL)
-        # wait for async cleanup so decodebin/appsink release memory
-        pipeline.get_state(Gst.CLOCK_TIME_NONE)
-    finally:
-        del pipeline
-        gc.collect()
-
-    if not chunks:
-        raise RuntimeError(f"Could not decode audio: {path}")
-    return np.concatenate(chunks)
-
-
-class OnnxAsrModel:
-    def __init__(self, model_id: str, quantization: str) -> None:
-        self._model_id = model_id
-        self._quantization = quantization
-        self._model: Any = None
-        self._loaded = False
-        self._busy = False
-        self._recognize_lock = threading.Lock()
-        self._unload_source: int | None = None
-
-    @staticmethod
-    def available() -> bool:
-        return find_spec("onnx_asr") is not None
-
-    @property
-    def is_loaded(self) -> bool:
-        return self._loaded
-
-    @property
-    def model_id(self) -> str:
-        return self._model_id
-
-    @property
-    def will_download(self) -> bool:
-        if self._loaded or not self.available():
-            return False
-        from huggingface_hub import try_to_load_from_cache
-        from onnx_asr.resolver import model_repos
-
-        repo = model_repos.get(self._model_id, self._model_id)
-        if "/" not in repo:
-            # Local path, nothing to download
-            return False
-        return not isinstance(try_to_load_from_cache(repo, "config.json"), str)
-
-    def set_config(self, model_id: str, quantization: str) -> None:
-        if model_id == self._model_id and quantization == self._quantization:
-            return
-        self.unload_now()
-        self._model_id = model_id
-        self._quantization = quantization
-
-    def load(self) -> None:
-        if self._loaded:
-            self._schedule_unload()
-            return
-
-        import onnx_asr
-
-        log.debug("Loading model %s", self._model_id)
-        providers = ["CUDAExecutionProvider", "CPUExecutionProvider"]
-        quantization = None if self._quantization == "fp32" else self._quantization
-        model = onnx_asr.load_model(self._model_id, quantization=quantization, providers=providers)
-        self._model = model.with_vad(onnx_asr.load_vad("silero"))
-        self._loaded = True
-        self._schedule_unload()
-
-    def recognize(
-        self,
-        audio_file: Path,
-        on_partial: Callable[[str], None] | None = None,
-    ) -> str:
-        # global lock, one transcription at a time
-        with self._recognize_lock:
-            self.load()
-            self._busy = True
-            try:
-                audio = load_audio(audio_file)
-                parts: list[str] = []
-                text = ""
-                for segment in self._model.recognize(audio, sample_rate=16000):
-                    parts.append(segment.text.strip())
-                    text = " ".join(part for part in parts if part)
-                    if on_partial is not None:
-                        on_partial(text)
-            finally:
-                self._busy = False
-        self._schedule_unload()
-        return text
-
-    def unload_now(self) -> None:
-        self._model = None
-        self._loaded = False
-        gc.collect()
-
-    def _schedule_unload(self) -> None:
-        if self._unload_source is not None:
-            GLib.source_remove(self._unload_source)
-        self._unload_source = GLib.timeout_add_seconds(
-            _IDLE_UNLOAD_SECONDS, self._unload_idle
-        )
-
-    def _unload_idle(self) -> bool:
-        if self._busy:
-            return GLib.SOURCE_CONTINUE
-        self._unload_source = None
-        self.unload_now()
-        return GLib.SOURCE_REMOVE

stt_voice_messages/models/distill.py

@@ -0,0 +1,45 @@
+# This file is part of Gajim.
+#
+# Gajim 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Gajim 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 Gajim. If not, see <http://www.gnu.org/licenses/>.
+
+import typing
+from pathlib import Path
+
+from gajim.gtk.const import Setting
+
+from .model_template import Model
+
+try:
+    import ctranslate2
+    CTRANSLATE2_AVAILABLE = True
+except ModuleNotFoundError:
+    if typing.TYPE_CHECKING:
+        import ctranslate2
+
+
+class WhisperModel(Model):
+    def __init__(self):
+        # TODO
+        self._result: str = ''
+
+    @property
+    def result(self) -> str:
+        return self._result
+
+    def transcribe(self, audio_file: Path) -> str:
+        pass
+
+    def on_setting(self, setting: Setting):
+        pass
+

stt_voice_messages/models/faster_whisper.py

@@ -0,0 +1,60 @@
+# This file is part of Gajim.
+#
+# Gajim 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Gajim 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 Gajim. If not, see <http://www.gnu.org/licenses/>.
+
+import logging
+import typing
+from dataclasses import dataclass
+from pathlib import Path
+
+from ..helper import Results
+from .model_settings import FasterWhisperSettings
+from .model_template import Model
+
+log = logging.getLogger('gajim.p.sttvm_faster_whisper')
+
+try:
+    import faster_whisper
+except ModuleNotFoundError:
+    if typing.TYPE_CHECKING:
+        import faster_whisper
+
+@dataclass
+class Configuration:
+    model_size: str
+
+class FasterWhisperModel(Model):
+    def __init__(self):
+        self._result: str = ''
+        self._config = FasterWhisperSettings()
+
+    @property
+    def result(self) -> str:
+        return self._result
+
+    def transcribe(self, result: Results, audio_file: Path) -> None:
+        model = faster_whisper.WhisperModel(self._config.model_size, compute_type="float32")
+        log.debug('model size is used = %s', self._config.model_size)
+        segments, _ = model.transcribe(audio_file)
+        segments = list(segments)
+        result.text = ''
+        log.debug('segments = %s', segments)
+        for segment in segments:
+            result.text += segment.text
+            print("[%.2fs -> %.2fs] %s" % (
+            segment.start, segment.end, segment.text))
+
+    def set_config(self, config: FasterWhisperSettings) -> None:
+        self._config = config
+

stt_voice_messages/models/model_settings.py

@@ -0,0 +1,23 @@
+# This file is part of Gajim.
+#
+# Gajim 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Gajim 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 Gajim. If not, see <http://www.gnu.org/licenses/>.
+
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class OnnxAsrSettings:
+    model_id: str = field(default='nemo-parakeet-tdt-0.6b-v3', init=True)
+    model_path: str = ''

stt_voice_messages/models/model_template.py

@@ -0,0 +1,41 @@
+# This file is part of Gajim.
+#
+# Gajim 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Gajim 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 Gajim. If not, see <http://www.gnu.org/licenses/>.
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+import numpy as np
+
+from ..helper import Results
+
+
+class Model(ABC):
+
+    @property
+    @abstractmethod
+    def is_loaded(self) -> bool:
+        pass
+
+    @abstractmethod
+    def load(self) -> None:
+        pass
+
+    @abstractmethod
+    def recognize(self, result: Results, audio: np.ndarray) -> None:
+        pass
+
+    @abstractmethod
+    def set_config(self, config: Any) -> None:
+        pass

stt_voice_messages/models/openai_whisper.py

@@ -0,0 +1,54 @@
+# This file is part of Gajim.
+#
+# Gajim 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Gajim 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 Gajim. If not, see <http://www.gnu.org/licenses/>.
+
+import logging
+import typing
+from dataclasses import dataclass
+from pathlib import Path
+
+from ..helper import Results
+from .model_settings import OpenAIWhisperSettings
+from .model_template import Model
+
+log = logging.getLogger('gajim.p.sttvm_whisper')
+
+try:
+    import whisper
+    OPENAI_WHISPER_AVAILABLE = True
+except ModuleNotFoundError:
+    if typing.TYPE_CHECKING:
+        import whisper
+
+@dataclass
+class Configuration:
+    model_size: str
+
+class WhisperModel(Model):
+    def __init__(self):
+        self._result: str = ''
+        self._config = OpenAIWhisperSettings()
+
+    @property
+    def result(self) -> str:
+        return self._result
+
+    def transcribe(self, result: Results, audio_file: Path) -> None:
+        model = whisper.load_model(self._config.model_size)
+        log.debug('model size is used = %s', self._config.model_size)
+        result.text = model.transcribe(audio_file)['text']  # pyright: ignore [reportAttributeAccessIssue]
+
+    def set_config(self, config: OpenAIWhisperSettings) -> None:
+        self._config = config
+

stt_voice_messages/models/stt.py

@@ -0,0 +1,132 @@
+# This file is part of Gajim.
+#
+# Gajim 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Gajim 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 Gajim. If not, see <http://www.gnu.org/licenses/>.
+
+import logging
+import pickle
+import subprocess
+import sys
+from pathlib import Path
+
+import numpy as np
+from gi.repository import GLib
+
+from ..helper import Results
+from .model_settings import OnnxAsrSettings
+from .model_template import Model
+
+log = logging.getLogger('gajim.p.sttvm_onnx_asr')
+
+
+_IDLE_UNLOAD_SECONDS = 300
+
+class OnnxAsrModel(Model):
+    def __init__(self):
+        self._proc = None
+        self._loaded = False
+        self._config = OnnxAsrSettings()
+        self._unload_source = None
+
+    @property
+    def is_loaded(self) -> bool:
+        return self._loaded
+
+    @property
+    def will_download(self) -> bool:
+        if self.is_loaded or self._config.model_path:
+            return False
+        from huggingface_hub import try_to_load_from_cache
+        from onnx_asr.resolver import model_repos
+        repo = model_repos.get(self._config.model_id, self._config.model_id)
+        if '/' not in repo:
+            return False
+        return not isinstance(try_to_load_from_cache(repo, 'config.json'), str)
+
+    def load(self) -> None:
+        if self._loaded:
+            self._reschedule_unload()
+            return
+        log.debug('Loading model %s in worker', self._config.model_id)
+        self._send({
+            'op': 'load',
+            'model_id': self._config.model_id,
+            'model_path': self._config.model_path,
+        })
+        self._loaded = True
+        self._reschedule_unload()
+
+    def recognize(self, result: Results, audio: np.ndarray) -> None:
+        self.load()
+        response = self._send({'op': 'recognize', 'audio': audio})
+        result.text = response['text']
+        self._reschedule_unload()
+
+    def set_config(self, config: OnnxAsrSettings) -> None:
+        if (config.model_id != self._config.model_id
+                or config.model_path != self._config.model_path):
+            self.unload_now()
+        self._config = OnnxAsrSettings(
+            model_id=config.model_id, model_path=config.model_path)
+
+    def unload_now(self) -> None:
+        if self._unload_source is not None:
+            GLib.source_remove(self._unload_source)
+            self._unload_source = None
+        if self._proc is not None:
+            log.debug('Terminating STT worker subprocess')
+            try:
+                self._proc.stdin.close()
+                self._proc.wait(timeout=2)
+            except subprocess.TimeoutExpired:
+                self._proc.kill()
+                self._proc.wait()
+            self._proc = None
+        self._loaded = False
+
+    def _ensure_proc(self) -> None:
+        if self._proc is not None and self._proc.poll() is None:
+            return
+        log.debug('Starting STT worker subprocess')
+        self._proc = subprocess.Popen(
+            [sys.executable, str(Path(__file__).parent / 'stt_worker.py')],
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+        )
+        self._loaded = False
+
+    def _send(self, cmd: dict) -> dict:
+        self._ensure_proc()
+        pickle.dump(cmd, self._proc.stdin)
+        self._proc.stdin.flush()
+        try:
+            response = pickle.load(self._proc.stdout)
+        except EOFError as e:
+            self._proc = None
+            self._loaded = False
+            raise RuntimeError('Worker subprocess exited unexpectedly') from e
+        if not response.get('ok'):
+            raise RuntimeError(response.get('error', 'unknown worker error'))
+        return response
+
+    def _reschedule_unload(self) -> None:
+        if self._unload_source is not None:
+            GLib.source_remove(self._unload_source)
+        self._unload_source = GLib.timeout_add_seconds(
+            _IDLE_UNLOAD_SECONDS, self._on_idle_unload)
+
+    def _on_idle_unload(self) -> bool:
+        self._unload_source = None
+        log.debug('Idle unload after %ds', _IDLE_UNLOAD_SECONDS)
+        self.unload_now()
+        return GLib.SOURCE_REMOVE

stt_voice_messages/models/stt_worker.py

@@ -0,0 +1,54 @@
+# This file is part of Gajim.
+#
+# Gajim 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Gajim 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 Gajim. If not, see <http://www.gnu.org/licenses/>.
+
+import pickle
+import sys
+import traceback
+
+
+def _respond(response: dict) -> None:
+    pickle.dump(response, sys.stdout.buffer)
+    sys.stdout.buffer.flush()
+
+
+def main() -> None:
+    model = None
+    while True:
+        try:
+            cmd = pickle.load(sys.stdin.buffer)
+        except EOFError:
+            return
+        try:
+            op = cmd['op']
+            if op == 'load':
+                import onnx_asr
+                model = onnx_asr.load_model(
+                    cmd['model_id'], cmd.get('model_path') or None)
+                _respond({'ok': True})
+            elif op == 'recognize':
+                text = model.recognize(cmd['audio'])
+                _respond({'ok': True, 'text': text})
+            else:
+                _respond({'ok': False, 'error': f'unknown op: {op}'})
+        except Exception as e:
+            _respond({
+                'ok': False,
+                'error': f'{type(e).__name__}: {e}',
+                'traceback': traceback.format_exc(),
+            })
+
+
+if __name__ == '__main__':
+    main()

stt_voice_messages/plugin-manifest.json

@@ -1,6 +1,6 @@
 {
     "authors": [
-        "hueso"
+        "mesonium <mesonium@posteo.eu>"
     ],
     "description": "Transcribes voice messages to text.",
     "homepage": "https://dev.gajim.org/gajim/gajim-plugins/wikis/STTVoiceMessagesPlugin",
@@ -13,8 +13,8 @@
         "win32"
     ],
     "requirements": [
-        "gajim>=2.5.0"
+        "gajim>=2.0.0"
     ],
     "short_name": "stt_voice_messages",
-    "version": "0.1.0"
+    "version": "0.0.1"
 }

stt_voice_messages/stt_voice_messages.py

@@ -19,52 +19,85 @@ import logging
 from functools import partial
 from pathlib import Path
 
-from gi.repository import Gtk
+from gi.repository import GLib, Gtk
 
+from gajim.common import app
 from gajim.plugins import GajimPlugin
 from gajim.plugins.plugins_i18n import _
 
-from .gtk.config_dialog import STTVoiceMessagesConfigDialog
+from .gtk.config_dialog import Configuration, STTVoiceMessagesConfigDialog
 from .gtk.sttbox import STTBox
-from .model import OnnxAsrModel
+from .models.model_settings import OnnxAsrSettings
 
-log = logging.getLogger("gajim.p.stt_voice_messages")
+log = logging.getLogger('gajim.p.stt_voice_messages')
+
+_FOCUS_LOSS_UNLOAD_SECONDS = 30
 
 
 class STTVoiceMessagesPlugin(GajimPlugin):
     def init(self) -> None:
-        self.description = _("Transcribes voice messages to text.")
+        self.description = _('Transcribes voice messages to text.')
 
         self.config_default_values = {
-            "auto_transcribe": (False, ""),
-            "model_id": ("nemo-parakeet-tdt-0.6b-v3", ""),
-            "quantization": ("int8", ""),
+            'auto_transcribe': (False, ''),
+            'model_id': ('nemo-parakeet-tdt-0.6b-v3', ''),
+            'model_path': ('', ''),
         }
 
-        self._model = OnnxAsrModel(
-            str(self.config["model_id"]), str(self.config["quantization"])
-        )
-        self.config_dialog = partial(STTVoiceMessagesConfigDialog, self)
+        self._config = Configuration(self)
+        self.config_dialog = partial(STTVoiceMessagesConfigDialog, self._config)
 
         self.gui_extension_points = {
-            "preview_audio": (self._on_preview_audio, None),
+            'preview_audio': (self._on_preview_audio_created, None),
         }
 
-    @property
-    def model(self) -> OnnxAsrModel:
-        return self._model
+        self._active_handler_id = 0
+        self._focus_unload_source = None
+
+    def activate(self) -> None:
+        if app.window is not None and self._active_handler_id == 0:
+            self._active_handler_id = app.window.connect(
+                'notify::is-active', self._on_window_active_changed)
 
     def deactivate(self) -> None:
-        self._model.unload_now()
+        if self._focus_unload_source is not None:
+            GLib.source_remove(self._focus_unload_source)
+            self._focus_unload_source = None
+        if self._active_handler_id != 0 and app.window is not None:
+            app.window.disconnect(self._active_handler_id)
+            self._active_handler_id = 0
+        if self._config.is_available:
+            self._config.unload_model()
 
-    def _on_preview_audio(
-        self, drawing_box: Gtk.Box, control_box: Gtk.Box, audio_file: Path
-    ) -> None:
-        content_box = drawing_box.get_parent().get_parent()
+    def _on_window_active_changed(self,
+                                  window: Gtk.Window,
+                                  _pspec: object,
+                                  ) -> None:
+        if window.is_active():
+            if self._focus_unload_source is not None:
+                GLib.source_remove(self._focus_unload_source)
+                self._focus_unload_source = None
+        elif self._focus_unload_source is None:
+            self._focus_unload_source = GLib.timeout_add_seconds(
+                _FOCUS_LOSS_UNLOAD_SECONDS, self._on_focus_unload_fired)
 
-        stt_box = STTBox(self._model, audio_file)
-        control_box.append(stt_box.button)
-        content_box.append(stt_box)
+    def _on_focus_unload_fired(self) -> bool:
+        self._focus_unload_source = None
+        if self._config.is_available:
+            self._config.unload_model()
+        return GLib.SOURCE_REMOVE
 
-        if self.config["auto_transcribe"]:
-            stt_box.transcribe()
+    def _on_preview_audio_created(self,
+                                  drawing_box: Gtk.Box,
+                                  control_box: Gtk.Box,
+                                  audio_file: Path
+                                  ) -> None:
+        self._drawing_box = drawing_box;
+        self._control_box = control_box;
+        self._audio_file = audio_file.as_posix()
+        self._create_stt_box()
+
+    def _create_stt_box(self) -> None:
+        self._stt_box = STTBox(self._config, self._audio_file)
+        self._control_box.append(self._stt_box.button)
+        self._drawing_box.append(self._stt_box)