STT plugin based on onnx-asr

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