GitLab

at 2023-05-03T22:01:20+02:00

refactor: Remove omemo.aes module

Use the one provided by omemo-dr

refactor: Remove common.omemo module

This module has been moved to omemo-dr

refactor: Adapt to omemo-dr changes

- Rename method and variables to snake case
- Let omemo-dr generate the keys
- Add needs_init()
- Fix return types for some storage methods
- Remove obsolete database methods

     python3-gi,

     python3-gi-cairo,

     python3-nbxmpp-nightly (>=20230405),

-    python3-omemo-dr (>=20230407),

+    python3-omemo-dr (>=20230503),

     python3-setuptools,

     python3-packaging,

     python3-cryptography (>=3.4.8),

     python3-gi (>= 3.42.0),

     python3-gi-cairo (>= 1.14.0~),

     python3-nbxmpp-nightly (>=20230405),

-    python3-omemo-dr (>=20230407),

+    python3-omemo-dr (>=20230503),

     gir1.2-pango-1.0 (>= 1.50.0),

     gir1.2-gtk-3.0 (>= 3.24.30),

     gir1.2-gtksource-4,

 

 import binascii

 import threading

+from pathlib import Path

 

 from gi.repository import GLib

 from nbxmpp.const import Affiliation

 from nbxmpp.structs import PresenceProperties

 from nbxmpp.structs import StanzaHandler

 from nbxmpp.task import Task

+from omemo_dr.aes import aes_encrypt_file

+from omemo_dr.const import OMEMOTrust

+from omemo_dr.exceptions import DecryptionFailed

+from omemo_dr.exceptions import DuplicateMessage

+from omemo_dr.exceptions import KeyExchangeMessage

+from omemo_dr.exceptions import MessageNotForDevice

+from omemo_dr.exceptions import SelfMessage

+from omemo_dr.session import OMEMOSession

+from omemo_dr.structs import OMEMOBundle

+from omemo_dr.structs import OMEMOConfig

 

 from gajim.common import app

+from gajim.common import configpaths

 from gajim.common import ged

 from gajim.common import types

 from gajim.common.const import EncryptionData

 from gajim.common.modules.util import as_task

 from gajim.common.modules.util import event_node

 from gajim.common.modules.util import prepare_stanza

-from gajim.common.omemo.aes import aes_encrypt_file

-from gajim.common.omemo.state import DecryptionFailed

-from gajim.common.omemo.state import DuplicateMessage

-from gajim.common.omemo.state import KeyExchangeMessage

-from gajim.common.omemo.state import MessageNotForDevice

-from gajim.common.omemo.state import OmemoState

-from gajim.common.omemo.state import SelfMessage

-from gajim.common.omemo.util import Trust

+from gajim.common.storage.omemo import OMEMOStorage

 from gajim.common.structs import OutgoingMessage

 

 ALLOWED_TAGS = [

         self.allow_groupchat = True

 

         self._own_jid = self._client.get_own_jid().bare

-        self._backend = OmemoState(self._account, self._own_jid, self)

+

+        data_dir = Path(configpaths.get('MY_DATA'))

+        db_path = data_dir / f'omemo_{self._own_jid}.db'

+        storage = OMEMOStorage(self._account, db_path, self._log)

+

+        omemo_config = OMEMOConfig(default_prekey_amount=100,

+                                   min_prekey_amount=80,

+                                   spk_archive_seconds=86400 * 15,

+                                   spk_cycle_seconds=86400,

+                                   unacknowledged_count=2000)

+

+        self._backend = OMEMOSession(self._own_jid, storage, omemo_config)

+        self._backend.register_signal('republish-bundle',

+                                      self._on_republish_bundle)

 

         self._omemo_groupchats: set[str] = set()

         self._muc_temp_store: dict[bytes, str] = {}

         if self.is_omemo_groupchat(jid):

             self.get_affiliation_list(jid)

 

+    def _on_republish_bundle(self,

+                             _session: OMEMOSession,

+                             _signal_name: str,

+                             bundle: OMEMOBundle

+                             ) -> None:

+

+        self.set_bundle(bundle=bundle)

+

     @property

-    def backend(self) -> OmemoState:

+    def backend(self) -> OMEMOSession:

         return self._backend

 

     def check_send_preconditions(self, contact: types.ChatContactT) -> bool:

                 return False

 

             missing = True

-            for member_jid in self.backend.get_muc_members(jid):

+            for member_jid in self.backend.get_group_members(jid):

                 if not self.are_keys_missing(member_jid):

                     missing = False

             if missing:

                 return False

 

             # check if bundles are missing for some devices

-            if self.backend.storage.hasUndecidedFingerprints(jid):

+            if self.backend.storage.has_undecided_fingerprints(jid):

                 self._log.info('%s => Undecided Fingerprints for %s',

                                contact.account, jid)

                 app.ged.raise_event(EncryptionInfo(

     def _new_fingerprints_available(self, contact: types.ChatContactT) -> bool:

         fingerprints: list[int] = []

         if contact.is_groupchat:

-            for member_jid in self.backend.get_muc_members(str(contact.jid),

+            for member_jid in self.backend.get_group_members(str(contact.jid),

                                                            without_self=False):

-                fingerprints = self.backend.storage.getNewFingerprints(

+                fingerprints = self.backend.storage.get_new_fingerprints(

                     member_jid)

                 if fingerprints:

                     break

 

         else:

-            fingerprints = self.backend.storage.getNewFingerprints(

+            fingerprints = self.backend.storage.get_new_fingerprints(

                 str(contact.jid))

 

         if not fingerprints:

         if not event.message:

             return False

 

-        omemo_message = self.backend.encrypt(str(event.jid), event.message)

+        client = app.get_client(self._account)

+        contact = client.get_module('Contacts').get_contact(event.jid)

+

+        omemo_message = self.backend.encrypt(str(event.jid),

+                                             event.message,

+                                             groupchat=contact.is_groupchat)

         if omemo_message is None:

             raise Exception('Encryption error')

 

             event.xhtml = None

             event.additional_data['encrypted'] = {

                 'name': 'OMEMO',

-                'trust': GajimTrust[Trust.VERIFIED.name]}

+                'trust': GajimTrust[OMEMOTrust.VERIFIED.name]}

 

         self._debug_print_stanza(event.stanza)

         return True

                 return

 

             plaintext = self._muc_temp_store[properties.omemo.payload]

-            fingerprint = self.backend.own_fingerprint

-            trust = Trust.VERIFIED

+            fingerprint = self.backend.get_own_fingerprint()

+            trust = OMEMOTrust.VERIFIED

             del self._muc_temp_store[properties.omemo.payload]

 

         except DecryptionFailed:

                           'but not for your device.')

             # Neither trust nor fingerprint can be verified if we didn't

             # successfully decrypt the message

-            trust = Trust.UNTRUSTED

+            trust = OMEMOTrust.UNTRUSTED

             fingerprint = None

 

         prepare_stanza(stanza, plaintext)

 

         assert isinstance(properties.omemo, OMEMOMessage)

         self._log.info('Groupchat: Last resort trying to find SID in DB')

-        from_jid = self.backend.storage.getJidFromDevice(properties.omemo.sid)

+        from_jid = self.backend.storage.get_jid_from_device(

+            properties.omemo.sid)

         if not from_jid:

             self._log.error(

                 "Can't decrypt GroupChat Message from %s", resource)

         jid = properties.muc_user.jid.bare

         if properties.muc_user.affiliation in (Affiliation.OUTCAST,

                                                Affiliation.NONE):

-            self.backend.remove_muc_member(room, jid)

+            self.backend.remove_group_member(room, jid)

         else:

-            self.backend.add_muc_member(room, jid)

+            self.backend.add_group_member(room, jid)

 

         if self.is_omemo_groupchat(room):

             if not self.is_contact_in_roster(jid):

 

         for user_jid in result.users:

             jid = str(user_jid)

-            self.backend.add_muc_member(room_jid, jid)

+            self.backend.add_group_member(room_jid, jid)

 

             if not self.is_contact_in_roster(jid):

                 # Query Devicelists from JIDs not in our Roster

             self._omemo_groupchats.discard(jid)

 

     def _check_for_missing_sessions(self, jid: str) -> None:

-        devices_without_session = self.backend.devices_without_sessions(jid)

+        devices_without_session = self.backend.get_devices_without_sessions(jid)

         for device_id in devices_without_session:

             if device_id in self._device_bundle_querys:

                 continue

         # Fetch Bundles of own other Devices

         if self._own_jid not in self._query_for_bundles:

 

-            devices_without_session = self.backend.devices_without_sessions(

+            devices_without_session = self.backend.get_devices_without_sessions(

                 self._own_jid)

 

             self._query_for_bundles.append(self._own_jid)

         # Fetch Bundles of contacts devices

         if contact_jid not in self._query_for_bundles:

 

-            devices_without_session = self.backend.devices_without_sessions(

+            devices_without_session = self.backend.get_devices_without_sessions(

                 contact_jid)

 

             self._query_for_bundles.append(contact_jid)

             return False

         return True

 

-    def set_bundle(self) -> None:

-        self._nbxmpp('OMEMO').set_bundle(self.backend.bundle,

-                                         self.backend.own_device)

+    def set_bundle(self, bundle: OMEMOBundle | None = None) -> None:

+        if bundle is None:

+            bundle = self.backend.get_bundle()

+        self._nbxmpp('OMEMO').set_bundle(bundle,

+                                         self.backend.get_own_device())

 

     @as_task

     def request_bundle(self, jid: str, device_id: int):

             message=EncryptionInfoMsg.UNDECIDED_FINGERPRINTS))

 

     def set_devicelist(self, devicelist: Optional[list[int]] = None) -> None:

-        devicelist_: set[int] = {self.backend.own_device}

+        devicelist_: set[int] = {self.backend.get_own_device()}

         if devicelist is not None:

             devicelist_.update(devicelist)

         self._log.info('Publishing own devicelist: %s', devicelist_)

 

     def clear_devicelist(self) -> None:

         self.backend.update_devicelist(

-            self._own_jid, [self.backend.own_device])

+            self._own_jid, [self.backend.get_own_device()])

         self.set_devicelist()

 

     @as_task

             self._query_for_bundles.remove(jid)

 

         if own_devices:

-            if not self.backend.is_own_device_published:

+            if not self.backend.is_own_device_published():

                 # Our own device_id is not in the list, it could be

                 # overwritten by some other client

                 self.set_devicelist(devicelist)

         stanzastr = '\n' + stanza.__str__(fancy=True)

         stanzastr = stanzastr[0:-1]

         self._log.debug(stanzastr)

+

+    def cleanup(self) -> None:

+        self._backend.destroy()

+        del self._backend
-# 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; version 3 only.

-#

-# 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 NamedTuple

-from typing import Union

-

-import logging

-import os

-

-from cryptography.hazmat.backends import default_backend

-from cryptography.hazmat.primitives.ciphers import algorithms

-from cryptography.hazmat.primitives.ciphers import Cipher

-from cryptography.hazmat.primitives.ciphers.modes import GCM

-

-log = logging.getLogger('gajim.c.omemo.aes')

-

-IV_SIZE = 12

-

-

-class EncryptionResult(NamedTuple):

-    payload: bytes

-    key: bytes

-    iv: bytes

-

-

-def _decrypt(key: bytes, iv: bytes, tag: bytes, data: bytes) -> bytes:

-    decryptor = Cipher(

-        algorithms.AES(key),

-        GCM(iv, tag=tag),

-        backend=default_backend()).decryptor()

-    return decryptor.update(data) + decryptor.finalize()

-

-

-def aes_decrypt(_key: bytes, iv: bytes, payload: bytes) -> str:

-    if len(_key) >= 32:

-        # XEP-0384

-        log.debug('XEP Compliant Key/Tag')

-        data = payload

-        key = _key[:16]

-        tag = _key[16:]

-    else:

-        # Legacy

-        log.debug('Legacy Key/Tag')

-        data = payload[:-16]

-        key = _key

-        tag = payload[-16:]

-

-    return _decrypt(key, iv, tag, data).decode()

-

-

-def aes_decrypt_file(key: bytes, iv: bytes, payload: bytes) -> bytes:

-    data = payload[:-16]

-    tag = payload[-16:]

-    return _decrypt(key, iv, tag, data)

-

-

-def _encrypt(data: Union[str, bytes],

-             key_size: int,

-             iv_size: int = IV_SIZE

-             ) -> tuple[bytes, bytes, bytes, bytes]:

-

-    if isinstance(data, str):

-        data = data.encode()

-    key = os.urandom(key_size)

-    iv = os.urandom(iv_size)

-    encryptor = Cipher(

-        algorithms.AES(key),

-        GCM(iv),

-        backend=default_backend()).encryptor()

-

-    payload = encryptor.update(data) + encryptor.finalize()

-    return key, iv, encryptor.tag, payload

-

-

-def aes_encrypt(plaintext: str) -> EncryptionResult:

-    key, iv, tag, payload = _encrypt(plaintext, 16)

-    key += tag

-    return EncryptionResult(payload=payload, key=key, iv=iv)

-

-

-def aes_encrypt_file(data: bytes) -> EncryptionResult:

-    key, iv, tag, payload, = _encrypt(data, 32)

-    payload += tag

-    return EncryptionResult(payload=payload, key=key, iv=iv)

-

-

-def get_new_key() -> bytes:

-    return os.urandom(16)

-

-

-def get_new_iv() -> bytes:

-    return os.urandom(IV_SIZE)
-# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>

-# Copyright (C) 2015 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>

-#

-# This file is part of OMEMO Gajim Plugin.

-#

-# OMEMO 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.

-#

-# OMEMO 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 OMEMO Gajim Plugin. If not, see <http://www.gnu.org/licenses/>.

-

-from __future__ import annotations

-

-from typing import Any

-from typing import Optional

-

-import time

-from collections import defaultdict

-from pathlib import Path

-

-from nbxmpp.structs import OMEMOBundle

-from nbxmpp.structs import OMEMOMessage

-from omemo_dr.ecc.djbec import CurvePublicKey

-from omemo_dr.exceptions import DuplicateMessageException

-from omemo_dr.identitykey import IdentityKey

-from omemo_dr.identitykeypair import IdentityKeyPair

-from omemo_dr.protocol.prekeywhispermessage import PreKeyWhisperMessage

-from omemo_dr.protocol.whispermessage import WhisperMessage

-from omemo_dr.sessionbuilder import SessionBuilder

-from omemo_dr.sessioncipher import SessionCipher

-from omemo_dr.state.prekeybundle import PreKeyBundle

-from omemo_dr.util.keyhelper import KeyHelper

-

-from gajim.common import app

-from gajim.common import configpaths

-from gajim.common import types

-from gajim.common.omemo.aes import aes_decrypt

-from gajim.common.omemo.aes import aes_encrypt

-from gajim.common.omemo.aes import get_new_iv

-from gajim.common.omemo.aes import get_new_key

-from gajim.common.omemo.util import DEFAULT_PREKEY_AMOUNT

-from gajim.common.omemo.util import get_fingerprint

-from gajim.common.omemo.util import MIN_PREKEY_AMOUNT

-from gajim.common.omemo.util import SPK_ARCHIVE_TIME

-from gajim.common.omemo.util import SPK_CYCLE_TIME

-from gajim.common.omemo.util import Trust

-from gajim.common.omemo.util import UNACKNOWLEDGED_COUNT

-from gajim.common.storage.omemo import OMEMOStorage

-

-

-class OmemoState:

-    def __init__(self,

-                 account: str,

-                 own_jid: str,

-                 xmpp_con: types.xmppClient

-                 ) -> None:

-

-        self._account = account

-        self._own_jid = own_jid

-        self._log = xmpp_con._log

-        self._session_ciphers: dict[

-            str, dict[int, SessionCipher]] = defaultdict(dict)

-

-        data_dir = Path(configpaths.get('MY_DATA'))

-        db_path = data_dir / f'omemo_{self._own_jid}.db'

-        self._storage = OMEMOStorage(account, db_path, self._log)

-

-        self.xmpp_con = xmpp_con

-

-        self._log.info('%s PreKeys available',

-                       self._storage.getPreKeyCount())

-

-        self._device_store: dict[str, set[int]] = defaultdict(set)

-        self._muc_member_store: dict[str, set[str]] = defaultdict(set)

-

-        reg_id = self._storage.getLocalRegistrationId()

-        if reg_id is None:

-            raise ValueError('No own device found')

-

-        self._own_device = reg_id

-        self.add_device(self._own_jid, self._own_device)

-        self._log.info('Our device id: %s', self._own_device)

-

-        for jid, device in self._storage.getActiveDeviceTuples():

-            self._log.info('Load device from storage: %s - %s', jid, device)

-            self.add_device(jid, device)

-

-    def build_session(self,

-                      jid: str,

-                      device_id: int,

-                      bundle: OMEMOBundle

-                      ) -> None:

-

-        session = SessionBuilder(self._storage, self._storage, self._storage,

-                                 self._storage, jid, device_id)

-

-        registration_id = self._storage.getLocalRegistrationId()

-

-        prekey = bundle.pick_prekey()

-        otpk = CurvePublicKey(prekey['key'][1:])

-

-        spk = CurvePublicKey(bundle.spk['key'][1:])

-        ik = IdentityKey(CurvePublicKey(bundle.ik[1:]))

-

-        prekey_bundle = PreKeyBundle(registration_id,

-                                     device_id,

-                                     prekey['id'],

-                                     otpk,

-                                     bundle.spk['id'],

-                                     spk,

-                                     bundle.spk_signature,

-                                     ik)

-

-        session.processPreKeyBundle(prekey_bundle)

-        self._get_session_cipher(jid, device_id)

-

-    @property

-    def storage(self) -> OMEMOStorage:

-        return self._storage

-

-    @property

-    def own_fingerprint(self) -> str:

-        return get_fingerprint(self._storage.getIdentityKeyPair())

-

-    @property

-    def bundle(self) -> OMEMOBundle:

-        self._check_pre_key_count()

-

-        bundle: dict[str, Any] = {'otpks': []}

-        for k in self._storage.loadPendingPreKeys():

-            key = k.getKeyPair().getPublicKey().serialize()

-            bundle['otpks'].append({'key': key, 'id': k.getId()})

-

-        ik_pair = self._storage.getIdentityKeyPair()

-        bundle['ik'] = ik_pair.getPublicKey().serialize()

-

-        self._cycle_signed_pre_key(ik_pair)

-

-        spk = self._storage.loadSignedPreKey(

-            self._storage.getCurrentSignedPreKeyId())

-        bundle['spk_signature'] = spk.getSignature()

-        bundle['spk'] = {'key': spk.getKeyPair().getPublicKey().serialize(),

-                         'id': spk.getId()}

-

-        return OMEMOBundle(**bundle)

-

-    def decrypt_message(self,

-                        omemo_message: OMEMOMessage,

-                        jid: str

-                        ) -> tuple[str, str, Trust]:

-

-        if omemo_message.sid == self.own_device:

-            self._log.info('Received previously sent message by us')

-            raise SelfMessage

-

-        try:

-            encrypted_key, prekey = omemo_message.keys[self.own_device]

-        except KeyError:

-            self._log.info('Received message not for our device')

-            raise MessageNotForDevice

-

-        try:

-            if prekey:

-                key, fingerprint, trust = self._process_pre_key_message(

-                    jid, omemo_message.sid, encrypted_key)

-            else:

-                key, fingerprint, trust = self._process_message(

-                    jid, omemo_message.sid, encrypted_key)

-

-        except DuplicateMessageException:

-            self._log.info('Received duplicated message')

-            raise DuplicateMessage

-

-        except Exception as error:

-            self._log.warning(error)

-            raise DecryptionFailed

-

-        if omemo_message.payload is None:

-            self._log.debug('Decrypted Key Exchange Message')

-            raise KeyExchangeMessage

-

-        try:

-            result = aes_decrypt(key, omemo_message.iv, omemo_message.payload)

-        except Exception as error:

-            self._log.warning(error)

-            raise DecryptionFailed

-

-        self._log.debug('Decrypted Message => %s', result)

-        return result, fingerprint, trust

-

-    def _get_whisper_message(self,

-                             jid: str,

-                             device: int,

-                             key: bytes

-                             ) -> tuple[bytes, bool]:

-

-        cipher = self._get_session_cipher(jid, device)

-        cipher_key = cipher.encrypt(key)

-        prekey = isinstance(cipher_key, PreKeyWhisperMessage)

-        return cipher_key.serialize(), prekey

-

-    def encrypt(self, jid: str, plaintext: str) -> Optional[OMEMOMessage]:

-        try:

-            devices_for_encryption = self.get_devices_for_encryption(jid)

-        except NoDevicesFound:

-            self._log.warning('No devices for encryption found for: %s', jid)

-            return

-

-        result = aes_encrypt(plaintext)

-        whisper_messages: dict[

-            str, dict[int, tuple[bytes, bool]]] = defaultdict(dict)

-

-        for jid_, device in devices_for_encryption:

-            count = self._storage.getUnacknowledgedCount(jid_, device)

-            if count >= UNACKNOWLEDGED_COUNT:

-                self._log.warning('Set device inactive %s because of %s '

-                                  'unacknowledged messages', device, count)

-                self.remove_device(jid_, device)

-

-            try:

-                whisper_messages[jid_][device] = self._get_whisper_message(

-                    jid_, device, result.key)

-            except Exception:

-                self._log.exception('Failed to encrypt')

-                continue

-

-        recipients = set(whisper_messages.keys())

-        if jid != self._own_jid:

-            recipients -= {self._own_jid}

-        if not recipients:

-            self._log.error('Encrypted keys empty')

-            return

-

-        encrypted_keys: dict[int, tuple[bytes, bool]] = {}

-        for jid_ in whisper_messages:

-            encrypted_keys.update(whisper_messages[jid_])

-

-        self._log.debug('Finished encrypting message')

-        return OMEMOMessage(sid=self.own_device,

-                            keys=encrypted_keys,

-                            iv=result.iv,

-                            payload=result.payload)

-

-    def encrypt_key_transport(self,

-                              jid: str,

-                              devices: list[int]

-                              ) -> Optional[OMEMOMessage]:

-

-        whisper_messages: dict[

-            str, dict[int, tuple[bytes, bool]]] = defaultdict(dict)

-        for device in devices:

-            try:

-                whisper_messages[jid][device] = self._get_whisper_message(

-                    jid, device, get_new_key())

-            except Exception:

-                self._log.exception('Failed to encrypt')

-                continue

-

-        if not whisper_messages[jid]:

-            self._log.error('Encrypted keys empty')

-            return

-

-        self._log.debug('Finished Key Transport message')

-        return OMEMOMessage(sid=self.own_device,

-                            keys=whisper_messages[jid],

-                            iv=get_new_iv(),

-                            payload=None)

-

-    def has_trusted_keys(self, jid: str) -> bool:

-        inactive = self._storage.getInactiveSessionsKeys(jid)

-        trusted = self._storage.getTrustedFingerprints(jid)

-        return bool(set(trusted) - set(inactive))

-

-    def devices_without_sessions(self, jid: str) -> list[int]:

-        known_devices = self.get_devices(jid, without_self=True)

-        missing_devices = [dev

-                           for dev in known_devices

-                           if not self._storage.containsSession(jid, dev)]

-        if missing_devices:

-            self._log.info('Missing device sessions for %s: %s',

-                           jid, missing_devices)

-        return missing_devices

-

-    def _get_session_cipher(self, jid: str, device_id: int) -> SessionCipher:

-        try:

-            return self._session_ciphers[jid][device_id]

-        except KeyError:

-            cipher = SessionCipher(self._storage, self._storage, self._storage,

-                                   self._storage, jid, device_id)

-            self._session_ciphers[jid][device_id] = cipher

-            return cipher

-

-    def _process_pre_key_message(self,

-                                 jid: str,

-                                 device: int,

-                                 key: bytes

-                                 ) -> tuple[bytes, str, Trust]:

-

-        self._log.info('Process pre key message from %s', jid)

-        pre_key_message = PreKeyWhisperMessage.from_bytes(key)

-        if not pre_key_message.getPreKeyId():

-            raise Exception('Received Pre Key Message '

-                            'without PreKey => %s' % jid)

-

-        session_cipher = self._get_session_cipher(jid, device)

-        key = session_cipher.decryptPkmsg(pre_key_message)

-

-        identity_key = pre_key_message.getIdentityKey()

-        trust = self._get_trust_from_identity_key(jid, identity_key)

-        fingerprint = get_fingerprint(identity_key)

-

-        self._storage.setIdentityLastSeen(jid, identity_key)

-

-        self.xmpp_con.set_bundle()

-        self.add_device(jid, device)

-        return key, fingerprint, trust

-

-    def _process_message(self,

-                         jid: str,

-                         device: int,

-                         key: bytes

-                         ) -> tuple[bytes, str, Trust]:

-

-        self._log.info('Process message from %s', jid)

-        message = WhisperMessage.from_bytes(key)

-

-        session_cipher = self._get_session_cipher(jid, device)

-        key = session_cipher.decryptMsg(message)

-

-        identity_key = self._get_identity_key_from_device(jid, device)

-        trust = self._get_trust_from_identity_key(jid, identity_key)

-        fingerprint = get_fingerprint(identity_key)

-

-        self._storage.setIdentityLastSeen(jid, identity_key)

-

-        self.add_device(jid, device)

-

-        return key, fingerprint, trust

-

-    @staticmethod

-    def _get_identity_key_from_pk_message(key):

-        pre_key_message = PreKeyWhisperMessage.from_bytes(key)

-        return pre_key_message.getIdentityKey()

-

-    def _get_identity_key_from_device(self,

-                                      jid: str,

-                                      device: int

-                                      ) -> Optional[IdentityKey]:

-

-        session_record = self._storage.loadSession(jid, device)

-        return session_record.getSessionState().getRemoteIdentityKey()

-

-    def _get_trust_from_identity_key(self,

-                                     jid: str,

-                                     identity_key: IdentityKey

-                                     ) -> Trust:

-

-        trust = self._storage.getTrustForIdentity(jid, identity_key)

-        return Trust(trust) if trust is not None else Trust.UNDECIDED

-

-    def _check_pre_key_count(self) -> None:

-        # Check if enough PreKeys are available

-        pre_key_count = self._storage.getPreKeyCount()

-        if pre_key_count < MIN_PREKEY_AMOUNT:

-            missing_count = DEFAULT_PREKEY_AMOUNT - pre_key_count

-            self._storage.generateNewPreKeys(missing_count)

-            self._log.info('%s PreKeys created', missing_count)

-

-    def _cycle_signed_pre_key(self, ik_pair: IdentityKeyPair) -> None:

-        # Publish every SPK_CYCLE_TIME a new SignedPreKey

-        # Delete all existing SignedPreKeys that are older

-        # then SPK_ARCHIVE_TIME

-

-        # Check if SignedPreKey exist and create if not

-        if not self._storage.getCurrentSignedPreKeyId():

-            spk = KeyHelper.generateSignedPreKey(

-                ik_pair, self._storage.getNextSignedPreKeyId())

-            self._storage.storeSignedPreKey(spk.getId(), spk)

-            self._log.debug('New SignedPreKey created, because none existed')

-

-        # if SPK_CYCLE_TIME is reached, generate a new SignedPreKey

-        now = int(time.time())

-        timestamp = self._storage.getSignedPreKeyTimestamp(

-            self._storage.getCurrentSignedPreKeyId())

-

-        if int(timestamp) < now - SPK_CYCLE_TIME:

-            spk = KeyHelper.generateSignedPreKey(

-                ik_pair, self._storage.getNextSignedPreKeyId())

-            self._storage.storeSignedPreKey(spk.getId(), spk)

-            self._log.debug('Cycled SignedPreKey')

-

-        # Delete all SignedPreKeys that are older than SPK_ARCHIVE_TIME

-        timestamp = now - SPK_ARCHIVE_TIME

-        self._storage.removeOldSignedPreKeys(timestamp)

-

-    def update_devicelist(self, jid: str, devicelist: list[int]) -> None:

-        for device in list(devicelist):

-            if device == self.own_device:

-                continue

-            count = self._storage.getUnacknowledgedCount(jid, device)

-            if count > UNACKNOWLEDGED_COUNT:

-                self._log.warning('Ignore device because of %s unacknowledged'

-                                  ' messages: %s %s', count, jid, device)

-                devicelist.remove(device)

-

-        self._device_store[jid] = set(devicelist)

-        self._log.info('Saved devices for %s', jid)

-        self._storage.setActiveState(jid, devicelist)

-

-    def add_muc_member(self, room_jid: str, jid: str) -> None:

-        self._log.info('Saved MUC member %s %s', room_jid, jid)

-        self._muc_member_store[room_jid].add(jid)

-

-    def remove_muc_member(self, room_jid: str, jid: str) -> None:

-        self._log.info('Removed MUC member %s %s', room_jid, jid)

-        self._muc_member_store[room_jid].discard(jid)

-

-    def get_muc_members(self,

-                        room_jid: str,

-                        without_self: bool = True

-                        ) -> set[str]:

-

-        members = set(self._muc_member_store[room_jid])

-        if without_self:

-            members.discard(self._own_jid)

-        return members

-

-    def add_device(self, jid: str, device: int) -> None:

-        self._device_store[jid].add(device)

-

-    def remove_device(self, jid: str, device: int) -> None:

-        self._device_store[jid].discard(device)

-        self._storage.setInactive(jid, device)

-

-    def get_devices(self, jid: str, without_self: bool = False) -> set[int]:

-        devices = set(self._device_store[jid])

-        if without_self:

-            devices.discard(self.own_device)

-        return devices

-

-    def get_devices_for_encryption(self, jid: str) -> set[tuple[str, int]]:

-        devices_for_encryption: list[tuple[str, int]] = []

-

-        client = app.get_client(self._account)

-        contact = client.get_module('Contacts').get_contact(jid)

-        if contact.is_groupchat:

-            devices_for_encryption = self._get_devices_for_muc_encryption(jid)

-        else:

-            devices_for_encryption = self._get_devices_for_encryption(jid)

-

-        if not devices_for_encryption:

-            raise NoDevicesFound

-

-        devices_for_encryption += self._get_own_devices_for_encryption()

-        return set(devices_for_encryption)

-

-    def _get_devices_for_muc_encryption(self,

-                                        jid: str

-                                        ) -> list[tuple[str, int]]:

-

-        devices_for_encryption: list[tuple[str, int]] = []

-        for jid_ in self._muc_member_store[jid]:

-            devices_for_encryption += self._get_devices_for_encryption(jid_)

-        return devices_for_encryption

-

-    def _get_own_devices_for_encryption(self) -> list[tuple[str, int]]:

-        devices_for_encryption: list[tuple[str, int]] = []

-        own_devices = self.get_devices(self._own_jid, without_self=True)

-        for device in own_devices:

-            if self._storage.isTrusted(self._own_jid, device):

-                devices_for_encryption.append((self._own_jid, device))

-        return devices_for_encryption

-

-    def _get_devices_for_encryption(self, jid: str) -> list[tuple[str, int]]:

-        devices_for_encryption: list[tuple[str, int]] = []

-        devices = self.get_devices(jid)

-

-        for device in devices:

-            if self._storage.isTrusted(jid, device):

-                devices_for_encryption.append((jid, device))

-        return devices_for_encryption

-

-    @property

-    def own_device(self) -> int:

-        return self._own_device

-

-    @property

-    def devices_for_publish(self) -> set[int]:

-        devices = self.get_devices(self._own_jid)

-        if self.own_device not in devices:

-            devices.add(self.own_device)

-        return devices

-

-    @property

-    def is_own_device_published(self) -> bool:

-        return self.own_device in self.get_devices(self._own_jid)

-

-

-class NoDevicesFound(Exception):

-    pass

-

-

-class NoValidSessions(Exception):

-    pass

-

-

-class SelfMessage(Exception):

-    pass

-

-

-class MessageNotForDevice(Exception):

-    pass

-

-

-class DecryptionFailed(Exception):

-    pass

-

-

-class KeyExchangeMessage(Exception):

-    pass

-

-

-class InvalidMessage(Exception):

-    pass

-

-

-class DuplicateMessage(Exception):

-    pass
-# Copyright (C) 2015 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>

-#

-# This file is part of OMEMO Gajim Plugin.

-#

-# OMEMO 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.

-#

-# OMEMO 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 OMEMO Gajim Plugin. If not, see <http://www.gnu.org/licenses/>.

-

-from __future__ import annotations

-

-import binascii

-import textwrap

-from enum import IntEnum

-

-from omemo_dr.identitykey import IdentityKey

-from omemo_dr.identitykeypair import IdentityKeyPair

-

-DEFAULT_PREKEY_AMOUNT = 100

-MIN_PREKEY_AMOUNT = 80

-SPK_ARCHIVE_TIME = 86400 * 15  # 15 Days

-SPK_CYCLE_TIME = 86400         # 24 Hours

-UNACKNOWLEDGED_COUNT = 2000

-

-

-class Trust(IntEnum):

-    UNTRUSTED = 0

-    VERIFIED = 1

-    UNDECIDED = 2

-    BLIND = 3

-

-

-def get_fingerprint(identity_key: IdentityKeyPair,

-                    formatted: bool = False

-                    ) -> str:

-

-    public_key = identity_key.getPublicKey().serialize()

-    fingerprint = binascii.hexlify(public_key).decode()[2:]

-    if not formatted:

-        return fingerprint

-    fplen = len(fingerprint)

-    wordsize = fplen // 8

-    buf = ''

-    for word in range(0, fplen, wordsize):

-        buf += '{0} '.format(fingerprint[word:word + wordsize])

-    buf = textwrap.fill(buf, width=36)

-    return buf.rstrip().upper()

-

-

-class IdentityKeyExtended(IdentityKey):

-    def __hash__(self) -> int:

-        return hash(self.publicKey.serialize())

-

-    def get_fingerprint(self, formatted: bool = False) -> str:

-        return get_fingerprint(self, formatted=formatted)
 from collections import namedtuple

 from pathlib import Path

 

+from omemo_dr.const import OMEMOTrust

 from omemo_dr.ecc.djbec import CurvePublicKey

 from omemo_dr.ecc.djbec import DjbECPrivateKey

 from omemo_dr.exceptions import InvalidKeyIdException

 from omemo_dr.identitykey import IdentityKey

 from omemo_dr.identitykeypair import IdentityKeyPair

-from omemo_dr.state.axolotlstore import AxolotlStore

 from omemo_dr.state.prekeyrecord import PreKeyRecord

 from omemo_dr.state.sessionrecord import SessionRecord

 from omemo_dr.state.signedprekeyrecord import SignedPreKeyRecord

-from omemo_dr.util.keyhelper import KeyHelper

+from omemo_dr.state.store import Store

 from omemo_dr.util.medium import Medium

 

 from gajim.common import app

 from gajim.common.modules.util import LogAdapter

-from gajim.common.omemo.util import DEFAULT_PREKEY_AMOUNT

-from gajim.common.omemo.util import IdentityKeyExtended

-from gajim.common.omemo.util import Trust

 

 

-def _convert_identity_key(key: bytes) -> Optional[IdentityKeyExtended]:

+def _convert_identity_key(key: bytes) -> Optional[IdentityKey]:

     if not key:

         return

-    return IdentityKeyExtended(CurvePublicKey(key[1:]))

+    return IdentityKey(CurvePublicKey(key[1:]))

 

 

 def _convert_record(record: bytes) -> SessionRecord:

 sqlite3.register_converter('session_record', _convert_record)

 

 

-class OMEMOStorage(AxolotlStore):

+class OMEMOStorage(Store):

     def __init__(self, account: str, db_path: Path, log: LogAdapter) -> None:

         self._log = log

         self._account = account

         self._con = sqlite3.connect(db_path,

                                     detect_types=sqlite3.PARSE_COLNAMES)

         self._con.row_factory = self._namedtuple_factory

-        self.createDb()

-        self.migrateDb()

+        self.create_db()

+        self.migrate_db()

 

         self._con.execute('PRAGMA secure_delete=1')

         self._con.execute('PRAGMA synchronous=NORMAL;')

             self._con.execute('PRAGMA journal_mode=MEMORY;')

         self._con.commit()

 

-        if not self.getLocalRegistrationId():

-            self._log.info('Generating OMEMO keys')

-            self._generate_axolotl_keys()

-

     def _is_blind_trust_enabled(self) -> bool:

         return app.settings.get_account_setting(self._account,

                                                 'omemo_blind_trust')

                 fields.append(col[0])

         return namedtuple('Row', fields)(*row)  # pyright: ignore

 

-    def _generate_axolotl_keys(self) -> None:

-        identity_key_pair = KeyHelper.generateIdentityKeyPair()

-        registration_id = KeyHelper.getRandomSequence(2147483647)

-        pre_keys = KeyHelper.generatePreKeys(

-            KeyHelper.getRandomSequence(4294967296),

-            DEFAULT_PREKEY_AMOUNT)

-        self.storeLocalData(registration_id, identity_key_pair)

-

-        signed_pre_key = KeyHelper.generateSignedPreKey(

-            identity_key_pair, KeyHelper.getRandomSequence(65536))

-

-        self.storeSignedPreKey(signed_pre_key.getId(), signed_pre_key)

-

-        for pre_key in pre_keys:

-            self.storePreKey(pre_key.getId(), pre_key)

-

     def user_version(self) -> int:

         return self._con.execute('PRAGMA user_version').fetchone()[0]

 

-    def createDb(self) -> None:

+    def create_db(self) -> None:

         if self.user_version() == 0:

 

             create_tables = '''

                 ''' % (create_tables)

             self._con.executescript(create_db_sql)

 

-    def migrateDb(self) -> None:

+    def migrate_db(self) -> None:

         ''' Migrates the DB

         '''

 

             self._con.execute('PRAGMA user_version=12')

             self._con.commit()

 

-    def loadSignedPreKey(self, signedPreKeyId: int) -> SignedPreKeyRecord:

+    def load_signed_pre_key(self, signed_pre_key_id: int) -> SignedPreKeyRecord:

         query = 'SELECT record FROM signed_prekeys WHERE prekey_id = ?'

-        result = self._con.execute(query, (signedPreKeyId, )).fetchone()

+        result = self._con.execute(query, (signed_pre_key_id, )).fetchone()

         if result is None:

             raise InvalidKeyIdException('No such signedprekeyrecord! %s ' %

-                                        signedPreKeyId)

+                                        signed_pre_key_id)

         return SignedPreKeyRecord.from_bytes(result.record)

 

-    def loadSignedPreKeys(self) -> list[SignedPreKeyRecord]:

+    def load_signed_pre_keys(self) -> list[SignedPreKeyRecord]:

         query = 'SELECT record FROM signed_prekeys'

         results = self._con.execute(query).fetchall()

         return [SignedPreKeyRecord.from_bytes(row.record) for row in results]

 

-    def storeSignedPreKey(self,

-                          signedPreKeyId: int,

-                          signedPreKeyRecord: SignedPreKeyRecord

-                          ) -> None:

+    def store_signed_pre_key(self,

+                             signed_pre_key_id: int,

+                             signed_pre_key_record: SignedPreKeyRecord

+                             ) -> None:

 

         query = 'INSERT INTO signed_prekeys (prekey_id, record) VALUES(?,?)'

-        self._con.execute(query, (signedPreKeyId,

-                                  signedPreKeyRecord.serialize()))

+        self._con.execute(query, (signed_pre_key_id,

+                                  signed_pre_key_record.serialize()))

         self._con.commit()

 

-    def containsSignedPreKey(self, signedPreKeyId: int) -> bool:

+    def contains_signed_pre_key(self, signed_pre_key_id: int) -> bool:

         query = 'SELECT record FROM signed_prekeys WHERE prekey_id = ?'

-        result = self._con.execute(query, (signedPreKeyId,)).fetchone()

+        result = self._con.execute(query, (signed_pre_key_id,)).fetchone()

         return result is not None

 

-    def removeSignedPreKey(self, signedPreKeyId: int) -> None:

+    def remove_signed_pre_key(self, signed_pre_key_id: int) -> None:

         query = 'DELETE FROM signed_prekeys WHERE prekey_id = ?'

-        self._con.execute(query, (signedPreKeyId,))

+        self._con.execute(query, (signed_pre_key_id,))

         self._con.commit()

 

-    def getNextSignedPreKeyId(self) -> int:

-        result = self.getCurrentSignedPreKeyId()

-        if result is None:

-            return 1  # StartId if no SignedPreKeys exist

+    def get_next_signed_pre_key_id(self) -> int:

+        result = self.get_current_signed_pre_key_id()

         return (result % (Medium.MAX_VALUE - 1)) + 1

 

-    def getCurrentSignedPreKeyId(self) -> Optional[int]:

+    def get_current_signed_pre_key_id(self) -> int:

         query = 'SELECT MAX(prekey_id) FROM signed_prekeys'

         result = self._con.execute(query).fetchone()

-        return result.max_prekey_id if result is not None else None

+        assert result is not None

+        return result.max_prekey_id

 

-    def getSignedPreKeyTimestamp(self, signedPreKeyId: int) -> int:

+    def get_signed_pre_key_timestamp(self, signed_pre_key_id: int) -> int:

         query = '''SELECT strftime('%s', timestamp) FROM

                    signed_prekeys WHERE prekey_id = ?'''

 

-        result = self._con.execute(query, (signedPreKeyId,)).fetchone()

+        result = self._con.execute(query, (signed_pre_key_id,)).fetchone()

         if result is None:

             raise InvalidKeyIdException('No such signedprekeyrecord! %s' %

-                                        signedPreKeyId)

+                                        signed_pre_key_id)

 

         return result.formated_time

 

-    def removeOldSignedPreKeys(self, timestamp: int) -> None:

+    def remove_old_signed_pre_keys(self, timestamp: int) -> None:

         query = '''DELETE FROM signed_prekeys

                    WHERE timestamp < datetime(?, "unixepoch")'''

         self._con.execute(query, (timestamp,))

         self._con.commit()

 

-    def loadSession(self, recipientId: str, deviceId: int) -> SessionRecord:

+    def load_session(self, recipient_id: str, device_id: int) -> SessionRecord:

         query = '''SELECT record as "record [session_record]"

                    FROM sessions WHERE recipient_id = ? AND device_id = ?'''

-        result = self._con.execute(query, (recipientId, deviceId)).fetchone()

+        result = self._con.execute(query, (recipient_id, device_id)).fetchone()

         return result.record if result is not None else SessionRecord()

 

-    def getJidFromDevice(self, device_id: int) -> Optional[str]:

+    def get_jid_from_device(self, device_id: int) -> Optional[str]:

         query = '''SELECT recipient_id

                    FROM sessions WHERE device_id = ?'''

         result = self._con.execute(query, (device_id, )).fetchone()

         return result.recipient_id if result is not None else None

 

-    def getActiveDeviceTuples(self):

+    def get_active_device_tuples(self):

         query = '''SELECT recipient_id, device_id

                    FROM sessions WHERE active = 1'''

         return self._con.execute(query).fetchall()

 

-    def storeSession(self,

-                     recipientId: str,

-                     deviceId: int,

-                     sessionRecord: SessionRecord

-                     ) -> None:

+    def store_session(self,

+                      recipient_id: str,

+                      device_id: int,

+                      session_record: SessionRecord

+                      ) -> None:

 

         query = '''INSERT INTO sessions(recipient_id, device_id, record)

                    VALUES(?,?,?)'''

         try:

-            self._con.execute(query, (recipientId,

-                                      deviceId,

-                                      sessionRecord.serialize()))

+            self._con.execute(query, (recipient_id,

+                                      device_id,

+                                      session_record.serialize()))

         except sqlite3.IntegrityError:

             query = '''UPDATE sessions SET record = ?

                        WHERE recipient_id = ? AND device_id = ?'''

-            self._con.execute(query, (sessionRecord.serialize(),

-                                      recipientId,

-                                      deviceId))

+            self._con.execute(query, (session_record.serialize(),

+                                      recipient_id,

+                                      device_id))

 

         self._con.commit()

 

-    def containsSession(self, recipientId: str, deviceId: int) -> bool:

+    def contains_session(self, recipient_id: str, device_id: int) -> bool:

         query = '''SELECT record FROM sessions

                    WHERE recipient_id = ? AND device_id = ?'''

-        result = self._con.execute(query, (recipientId, deviceId)).fetchone()

+        result = self._con.execute(query, (recipient_id, device_id)).fetchone()

         return result is not None

 

-    def deleteSession(self, recipientId: str, deviceId: int) -> None:

-        self._log.info('Delete session for %s %s', recipientId, deviceId)

+    def delete_session(self, recipient_id: str, device_id: int) -> None:

+        self._log.info('Delete session for %s %s', recipient_id, device_id)

         query = 'DELETE FROM sessions WHERE recipient_id = ? AND device_id = ?'

-        self._con.execute(query, (recipientId, deviceId))

+        self._con.execute(query, (recipient_id, device_id))

         self._con.commit()

 

-    def deleteAllSessions(self, recipientId: str) -> None:

+    def delete_all_sessions(self, recipient_id: str) -> None:

         query = 'DELETE FROM sessions WHERE recipient_id = ?'

-        self._con.execute(query, (recipientId,))

+        self._con.execute(query, (recipient_id,))

         self._con.commit()

 

-    def getSessionsFromJid(self, recipientId: str):

+    def get_sessions_from_jid(self, recipient_id: str):

         query = '''SELECT recipient_id,

                           device_id,

                           record as "record [session_record]",

                           active

                    FROM sessions WHERE recipient_id = ?'''

-        return self._con.execute(query, (recipientId,)).fetchall()

+        return self._con.execute(query, (recipient_id,)).fetchall()

 

-    def getSessionsFromJids(self, recipientIds: list[str]):

+    def get_sessions_from_jids(self, recipient_ids: list[str]):

         query = '''

         SELECT recipient_id,

                device_id,

                active

         FROM sessions

         WHERE recipient_id IN ({})'''.format(

-            ', '.join(['?'] * len(recipientIds)))

-        return self._con.execute(query, recipientIds).fetchall()

+            ', '.join(['?'] * len(recipient_ids)))

+        return self._con.execute(query, recipient_ids).fetchall()

 

-    def setActiveState(self, jid: str, devicelist: list[int]) -> None:

+    def set_active_state(self, address: str, devicelist: list[int]) -> None:

         query = '''

         UPDATE sessions SET active = 1

         WHERE recipient_id = ? AND device_id IN ({})'''.format(

             ', '.join(['?'] * len(devicelist)))

-        self._con.execute(query, (jid,) + tuple(devicelist))

+        self._con.execute(query, (address,) + tuple(devicelist))

 

         query = '''

         UPDATE sessions SET active = 0

         WHERE recipient_id = ? AND device_id NOT IN ({})'''.format(

             ', '.join(['?'] * len(devicelist)))

-        self._con.execute(query, (jid,) + tuple(devicelist))

+        self._con.execute(query, (address,) + tuple(devicelist))

         self._con.commit()

 

-    def setInactive(self, jid: str, device_id: int) -> None:

+    def set_inactive(self, address: str, device_id: int) -> None:

         query = '''UPDATE sessions SET active = 0

                    WHERE recipient_id = ? AND device_id = ?'''

-        self._con.execute(query, (jid, device_id))

+        self._con.execute(query, (address, device_id))

         self._con.commit()

 

-    def getInactiveSessionsKeys(self,

-                                recipientId: str

-                                ) -> list[IdentityKeyExtended]:

+    def get_inactive_sessions_keys(self,

+                                   recipient_id: str

+                                   ) -> list[IdentityKey]:

 

         query = '''SELECT record as "record [session_record]" FROM sessions

                    WHERE active = 0 AND recipient_id = ?'''

-        results = self._con.execute(query, (recipientId,)).fetchall()

+        results = self._con.execute(query, (recipient_id,)).fetchall()

 

-        keys: list[IdentityKeyExtended] = []

+        keys: list[IdentityKey] = []

         for result in results:

-            key = result.record.getSessionState().getRemoteIdentityKey()

-            keys.append(IdentityKeyExtended(key.getPublicKey()))

+            key = result.record.get_session_state().get_remote_identity_key()

+            keys.append(IdentityKey(key.get_public_key()))

         return keys

 

-    def loadPreKey(self, preKeyId: int) -> PreKeyRecord:

+    def load_pre_key(self, pre_key_id: int) -> PreKeyRecord:

         query = '''SELECT record FROM prekeys WHERE prekey_id = ?'''

 

-        result = self._con.execute(query, (preKeyId,)).fetchone()

+        result = self._con.execute(query, (pre_key_id,)).fetchone()

         if result is None:

             raise Exception('No such prekeyRecord!')

         return PreKeyRecord.from_bytes(result.record)

 

-    def loadPendingPreKeys(self) -> list[PreKeyRecord]:

+    def load_pending_pre_keys(self) -> list[PreKeyRecord]:

         query = '''SELECT record FROM prekeys'''

         result = self._con.execute(query).fetchall()

         return [PreKeyRecord.from_bytes(row.record) for row in result]

 

-    def storePreKey(self, preKeyId: int, preKeyRecord: PreKeyRecord) -> None:

+    def store_pre_key(self,

+                      pre_key_id: int,

+                      pre_key_record: PreKeyRecord

+                      ) -> None:

         query = 'INSERT INTO prekeys (prekey_id, record) VALUES(?,?)'

-        self._con.execute(query, (preKeyId, preKeyRecord.serialize()))

+        self._con.execute(query, (pre_key_id, pre_key_record.serialize()))

         self._con.commit()

 

-    def containsPreKey(self, preKeyId: int) -> bool:

+    def contains_pre_key(self, pre_key_id: int) -> bool:

         query = 'SELECT record FROM prekeys WHERE prekey_id = ?'

-        result = self._con.execute(query, (preKeyId,)).fetchone()

+        result = self._con.execute(query, (pre_key_id,)).fetchone()

         return result is not None

 

-    def removePreKey(self, preKeyId: int) -> None:

+    def remove_pre_key(self, pre_key_id: int) -> None:

         query = 'DELETE FROM prekeys WHERE prekey_id = ?'

-        self._con.execute(query, (preKeyId,))

+        self._con.execute(query, (pre_key_id,))

         self._con.commit()

 

-    def getCurrentPreKeyId(self) -> int:

+    def get_current_pre_key_id(self) -> Optional[int]:

         query = 'SELECT MAX(prekey_id) FROM prekeys'

-        return self._con.execute(query).fetchone().max_prekey_id

+        result = self._con.execute(query).fetchone()

+        return result.max_prekey_id if result is not None else None

 

-    def getPreKeyCount(self) -> int:

+    def get_pre_key_count(self) -> int:

         query = 'SELECT COUNT(prekey_id) FROM prekeys'

         return self._con.execute(query).fetchone().count_prekey_id

 

-    def generateNewPreKeys(self, count: int) -> None:

-        prekey_id = self.getCurrentPreKeyId() or 0

-        pre_keys = KeyHelper.generatePreKeys(prekey_id + 1, count)

-        for pre_key in pre_keys:

-            self.storePreKey(pre_key.getId(), pre_key)

-

-    def getIdentityKeyPair(self) -> IdentityKeyPair:

+    def get_identity_key_pair(self) -> IdentityKeyPair:

         query = '''SELECT public_key as "public_key [pk]", private_key

                    FROM secret LIMIT 1'''

         result = self._con.execute(query).fetchone()

         return IdentityKeyPair.new(result.public_key,

                                    DjbECPrivateKey(result.private_key))

 

-    def getLocalRegistrationId(self) -> Optional[int]:  # pyright: ignore

+    def get_local_registration_id(self) -> int:

         query = 'SELECT device_id FROM secret LIMIT 1'

         result = self._con.execute(query).fetchone()

-        return result.device_id if result is not None else None

+        assert result is not None

+        return result.device_id

 

-    def storeLocalData(self,

-                       device_id: int,

-                       identityKeyPair: IdentityKeyPair

-                       ) -> None:

+    def store_own_identity(self,

+                           device_id: int,

+                           identity_key_pair: IdentityKeyPair

+                           ) -> None:

 

         query = 'SELECT * FROM secret'

         result = self._con.execute(query).fetchone()

         query = '''INSERT INTO secret(device_id, public_key, private_key)

                    VALUES(?, ?, ?)'''

 

-        public_key = identityKeyPair.getPublicKey().getPublicKey().serialize()

-        private_key = identityKeyPair.getPrivateKey().serialize()

+        public_key = identity_key_pair.get_public_key().get_public_key().\

+            serialize()

+        private_key = identity_key_pair.get_private_key().serialize()

         self._con.execute(query, (device_id, public_key, private_key))

         self._con.commit()

 

-    def saveIdentity(self, recipientId: str, identityKey: IdentityKey) -> None:

+    def save_identity(self,

+                      recipient_id: str,

+                      identity_key: IdentityKey

+                      ) -> None:

         query = '''INSERT INTO identities

                    (recipient_id, public_key, trust, shown)

                    VALUES(?, ?, ?, ?)'''

-        if not self.containsIdentity(recipientId, identityKey):

-            trust = self.getDefaultTrust(recipientId)

-            self._con.execute(query, (recipientId,

-                                      identityKey.getPublicKey().serialize(),

+        if not self.contains_identity(recipient_id, identity_key):

+            trust = self.get_default_trust(recipient_id)

+            self._con.execute(query, (recipient_id,

+                                      identity_key.get_public_key().serialize(),

                                       trust,

-                                      1 if trust == Trust.BLIND else 0))

+                                      1 if trust == OMEMOTrust.BLIND else 0))

             self._con.commit()

 

-    def containsIdentity(self,

-                         recipientId: str,

-                         identityKey: IdentityKey

-                         ) -> bool:

+    def contains_identity(self,

+                          recipient_id: str,

+                          identity_key: IdentityKey

+                          ) -> bool:

 

         query = '''SELECT * FROM identities WHERE recipient_id = ?

                    AND public_key = ?'''

 

-        public_key = identityKey.getPublicKey().serialize()

-        result = self._con.execute(query, (recipientId,

+        public_key = identity_key.get_public_key().serialize()

+        result = self._con.execute(query, (recipient_id,

                                            public_key)).fetchone()

 

         return result is not None

 

-    def deleteIdentity(self,

-                       recipientId: str,

-                       identityKey: IdentityKey

-                       ) -> None:

+    def delete_identity(self,

+                        recipient_id: str,

+                        identity_key: IdentityKey

+                        ) -> None:

 

         query = '''DELETE FROM identities

                    WHERE recipient_id = ? AND public_key = ?'''

-        public_key = identityKey.getPublicKey().serialize()

-        self._con.execute(query, (recipientId, public_key))

+        public_key = identity_key.get_public_key().serialize()

+        self._con.execute(query, (recipient_id, public_key))

         self._con.commit()

 

-    def isTrustedIdentity(self,

-                          recipientId: str,

-                          identityKey: IdentityKey

-                          ) -> bool:

+    def is_trusted_identity(self,

+                            recipient_id: str,

+                            identity_key: IdentityKey

+                            ) -> bool:

 

         return True

 

-    def getTrustForIdentity(self,

-                            recipientId: str,

-                            identityKey: IdentityKey

-                            ) -> Optional[Trust]:

+    def get_trust_for_identity(self,

+                               recipient_id: str,

+                               identity_key: IdentityKey

+                               ) -> Optional[OMEMOTrust]:

 

         query = '''SELECT trust FROM identities WHERE recipient_id = ?

                    AND public_key = ?'''

-        public_key = identityKey.getPublicKey().serialize()

-        result = self._con.execute(query, (recipientId, public_key)).fetchone()

+        public_key = identity_key.get_public_key().serialize()

+        result = self._con.execute(query, (recipient_id, public_key)).fetchone()

         return result.trust if result is not None else None

 

-    def getFingerprints(self, jid: str):

+    def get_fingerprints(self, jid: str):

         query = '''SELECT recipient_id,

                           public_key as "public_key [pk]",

                           trust,

                    WHERE recipient_id = ? ORDER BY trust ASC'''

         return self._con.execute(query, (jid,)).fetchall()

 

-    def getMucFingerprints(self, jids: list[str]):

+    def get_muc_fingerprints(self, jids: list[str]):

         query = '''

             SELECT recipient_id,

                    public_key as "public_key [pk]",

 

         return self._con.execute(query, jids).fetchall()

 

-    def hasUndecidedFingerprints(self, jid: str) -> bool:

+    def has_undecided_fingerprints(self, jid: str) -> bool:

         query = '''SELECT public_key as "public_key [pk]" FROM identities

                    WHERE recipient_id = ? AND trust = ?'''

-        result = self._con.execute(query, (jid, Trust.UNDECIDED)).fetchall()

+        result = self._con.execute(query,

+                                   (jid, OMEMOTrust.UNDECIDED)).fetchall()

         undecided = [row.public_key for row in result]

 

-        inactive = self.getInactiveSessionsKeys(jid)

+        inactive = self.get_inactive_sessions_keys(jid)

         undecided = set(undecided) - set(inactive)

         return bool(undecided)

 

-    def getDefaultTrust(self, jid: str) -> Trust:

+    def get_default_trust(self, jid: str) -> OMEMOTrust:

         if not self._is_blind_trust_enabled():

-            return Trust.UNDECIDED

+            return OMEMOTrust.UNDECIDED

 

         query = '''SELECT * FROM identities

                    WHERE recipient_id = ? AND trust IN (0, 1)'''

         result = self._con.execute(query, (jid,)).fetchone()

         if result is None:

-            return Trust.BLIND

-        return Trust.UNDECIDED

+            return OMEMOTrust.BLIND

+        return OMEMOTrust.UNDECIDED

 

-    def getTrustedFingerprints(self, jid: str) -> list[IdentityKeyExtended]:

+    def get_trusted_fingerprints(self, address: str) -> list[IdentityKey]:

         query = '''SELECT public_key as "public_key [pk]" FROM identities

                    WHERE recipient_id = ? AND trust IN(1, 3)'''

-        result = self._con.execute(query, (jid,)).fetchall()

+        result = self._con.execute(query, (address,)).fetchall()

         return [row.public_key for row in result]

 

-    def getNewFingerprints(self, jid: str) -> list[int]:

+    def get_new_fingerprints(self, jid: str) -> list[int]:

         query = '''SELECT _id FROM identities WHERE shown = 0

                    AND recipient_id = ?'''

 

         result = self._con.execute(query, (jid,)).fetchall()

         return [row.id for row in result]

 

-    def setShownFingerprints(self, fingerprints: list[int]) -> None:

+    def set_shown_fingerprints(self, fingerprints: list[int]) -> None:

         query = 'UPDATE identities SET shown = 1 WHERE _id IN ({})'.format(

             ', '.join(['?'] * len(fingerprints)))

         self._con.execute(query, fingerprints)

         self._con.commit()

 

-    def setTrust(self,

-                 recipient_id: str,

-                 identityKey: IdentityKey,

-                 trust: Trust

-                 ) -> None:

+    def set_trust(self,

+                  recipient_id: str,

+                  identity_key: IdentityKey,

+                  trust: OMEMOTrust

+                  ) -> None:

 

         query = '''UPDATE identities SET trust = ? WHERE public_key = ?

                    AND recipient_id = ?'''

-        public_key = identityKey.getPublicKey().serialize()

+        public_key = identity_key.get_public_key().serialize()

         self._con.execute(query, (trust, public_key, recipient_id))

         self._con.commit()

 

-    def isTrusted(self, recipient_id: str, device_id: int) -> bool:

-        record = self.loadSession(recipient_id, device_id)

-        if record.isFresh():

+    def is_trusted(self, recipient_id: str, device_id: int) -> bool:

+        record = self.load_session(recipient_id, device_id)

+        if record.is_fresh():

             return False

-        identity_key = record.getSessionState().getRemoteIdentityKey()

-        return self.getTrustForIdentity(

-            recipient_id, identity_key) in (Trust.VERIFIED, Trust.BLIND)

+        identity_key = record.get_session_state().get_remote_identity_key()

+        return self.get_trust_for_identity(

+            recipient_id, identity_key) in (OMEMOTrust.VERIFIED,

+                                            OMEMOTrust.BLIND)

 

-    def getIdentityLastSeen(self,

-                            recipient_id: str,

-                            identity_key: IdentityKey

-                            ) -> Optional[int]:

+    def get_identity_last_seen(self,

+                               recipient_id: str,

+                               identity_key: IdentityKey

+                               ) -> Optional[int]:

 

-        serialized = identity_key.getPublicKey().serialize()

+        serialized = identity_key.get_public_key().serialize()

         query = '''SELECT timestamp FROM identities

                    WHERE recipient_id = ? AND public_key = ?'''

         result = self._con.execute(query, (recipient_id,

                                            serialized)).fetchone()

         return result.timestamp if result is not None else None

 

-    def setIdentityLastSeen(self,

-                            recipient_id: str,

-                            identity_key: IdentityKey

-                            ) -> None:

+    def set_identity_last_seen(self,

+                               recipient_id: str,

+                               identity_key: IdentityKey

+                               ) -> None:

 

         timestamp = int(time.time())

-        serialized = identity_key.getPublicKey().serialize()

+        serialized = identity_key.get_public_key().serialize()

         self._log.info('Set last seen for %s %s', recipient_id, timestamp)

         query = '''UPDATE identities SET timestamp = ?

                    WHERE recipient_id = ? AND public_key = ?'''

         self._con.execute(query, (timestamp, recipient_id, serialized))

         self._con.commit()

 

-    def getUnacknowledgedCount(self, recipient_id: str, device_id: int) -> int:

-        record = self.loadSession(recipient_id, device_id)

-        if record.isFresh():

+    def get_unacknowledged_count(self,

+                                 recipient_id: str,

+                                 device_id: int

+                                 ) -> int:

+        record = self.load_session(recipient_id, device_id)

+        if record.is_fresh():

             return 0

-        state = record.getSessionState()

-        return state.getSenderChainKey().getIndex()

+        state = record.get_session_state()

+        return state.get_sender_chain_key().get_index()

 

-    def getSubDeviceSessions(self, recipientId: str) -> list[int]:

-        # Not used

-        return []
+    def needs_init(self) -> bool:

+        try:

+            self.get_local_registration_id()

+        except AssertionError:

+            return True

+        return False
 from gi.repository import GdkPixbuf

 from gi.repository import Gtk

 from nbxmpp.protocol import JID

-from omemo_dr.identitykeypair import IdentityKeyPair

+from omemo_dr.const import OMEMOTrust

+from omemo_dr.identitykey import IdentityKey

 

 from gajim.common import app

 from gajim.common import ged

 from gajim.common.helpers import generate_qr_code

 from gajim.common.i18n import _

 from gajim.common.modules.contacts import BareContact

-from gajim.common.omemo.util import get_fingerprint

-from gajim.common.omemo.util import IdentityKeyExtended

-from gajim.common.omemo.util import Trust

 

 from .builder import get_builder

 from .dialogs import ConfirmationDialog

 

 

 TRUST_DATA = {

-    Trust.UNTRUSTED: ('dialog-error-symbolic',

+    OMEMOTrust.UNTRUSTED: ('dialog-error-symbolic',

                       _('Untrusted'),

                       'error-color'),

-    Trust.UNDECIDED: ('security-low-symbolic',

+    OMEMOTrust.UNDECIDED: ('security-low-symbolic',

                       _('Not Decided'),

                       'warning-color'),

-    Trust.VERIFIED: ('security-high-symbolic',

+    OMEMOTrust.VERIFIED: ('security-high-symbolic',

                      _('Verified'),

                      'encrypted-color'),

-    Trust.BLIND: ('security-medium-symbolic',

+    OMEMOTrust.BLIND: ('security-medium-symbolic',

                   _('Blind Trust'),

                   'encrypted-color')

 }

 

         self._omemo = client.get_module('OMEMO')

 

-        self._identity_key = cast(

-            IdentityKeyPair, self._omemo.backend.storage.getIdentityKeyPair())

-        our_fpr_formatted = get_fingerprint(self._identity_key, formatted=True)

+        our_fpr_formatted = self._omemo.backend.get_own_fingerprint(

+            formatted=True)

         self._ui.our_fingerprint_1.set_text(our_fpr_formatted)

         self._ui.our_fingerprint_2.set_text(our_fpr_formatted)

 

         if contact.is_groupchat:

             members = list(self._omemo.backend.get_muc_members(

                 str(contact.jid)))

-            sessions = self._omemo.backend.storage.getSessionsFromJids(members)

-            results = self._omemo.backend.storage.getMucFingerprints(members)

+            sessions = self._omemo.backend.storage.get_sessions_from_jids(

+                members)

+            results = self._omemo.backend.storage.get_muc_fingerprints(members)

         else:

-            sessions = self._omemo.backend.storage.getSessionsFromJid(

+            sessions = self._omemo.backend.storage.get_sessions_from_jid(

                 str(contact.jid))

-            results = self._omemo.backend.storage.getFingerprints(

+            results = self._omemo.backend.storage.get_fingerprints(

                 str(contact.jid))

 

-        rows: dict[IdentityKeyExtended, KeyRow] = {}

+        rows: dict[IdentityKey, KeyRow] = {}

         for result in results:

             rows[result.public_key] = KeyRow(contact,

                                              result.recipient_id,

                                              result.timestamp)

 

         for item in sessions:

-            if item.record.isFresh():

+            if item.record.is_fresh():

                 return

-            identity_key = item.record.getSessionState().getRemoteIdentityKey()

-            identity_key = IdentityKeyExtended(identity_key.getPublicKey())

+            identity_key = item.record.get_session_state().\

+                get_remote_identity_key()

+            identity_key = IdentityKey(identity_key.get_public_key())

             try:

                 key_row = rows[identity_key]

             except KeyError:

                 log.warning('Could not find session identitykey %s',

                             item.device_id)

-                self._omemo.backend.storage.deleteSession(item.recipient_id,

+                self._omemo.backend.storage.delete_session(item.recipient_id,

                                                           item.device_id)

                 continue

 

     @staticmethod

     def _get_qrcode(jid: JID,

                     sid: int,

-                    identity_key: IdentityKeyPair

+                    fingerprint: str

                     ) -> GdkPixbuf.Pixbuf | None:

 

-        fingerprint = get_fingerprint(identity_key)

         qry = (XmppUriQuery.MESSAGE.value, [(f'omemo-sid-{sid}', fingerprint)])

         ver_string = jid.new_as_bare().to_iri(qry)

         log.debug('Verification String: %s', ver_string)

     def _load_qrcode(self) -> None:

         client = app.get_client(self._account)

         pixbuf = self._get_qrcode(client.get_own_jid(),

-                                  self._omemo.backend.own_device,

-                                  self._identity_key)

+                                  self._omemo.backend.get_own_device(),

+                                  self._omemo.backend.get_own_fingerprint())

         self._ui.qr_code_image.set_from_pixbuf(pixbuf)

 

     def _on_show_inactive(self, switch: Gtk.Switch, _param: Any) -> None:

     def __init__(self,

                  contact: types.ChatContactT,

                  jid: JID,

-                 identity_key: IdentityKeyExtended,

-                 trust: Trust,

+                 identity_key: IdentityKey,

+                 trust: OMEMOTrust,

                  last_seen: Optional[float]

                  ) -> None:

 

 

         def _remove():

             self._omemo.backend.remove_device(str(self.jid), self.device_id)

-            self._omemo.backend.storage.deleteSession(

+            self._omemo.backend.storage.delete_session(

                 str(self.jid), self.device_id)

-            self._omemo.backend.storage.deleteIdentity(

+            self._omemo.backend.storage.delete_identity(

                 str(self.jid), self._identity_key)

 

             listbox = cast(Gtk.ListBox, self.get_parent())

         image.get_style_context().add_class(css_class)

         image.set_tooltip_text(tooltip)

 

-        self._omemo.backend.storage.setTrust(

+        self._omemo.backend.storage.set_trust(

             str(self.jid), self._identity_key, self.trust)

 

     @property

 

     def update(self) -> None:

         self._listbox.foreach(self._listbox.remove)

-        if self._row.trust != Trust.VERIFIED:

+        if self._row.trust != OMEMOTrust.VERIFIED:

             self._listbox.add(VerifiedOption())

-        if self._row.trust != Trust.BLIND:

+        if self._row.trust != OMEMOTrust.BLIND:

             self._listbox.add(BlindOption())

-        if self._row.trust != Trust.UNTRUSTED:

+        if self._row.trust != OMEMOTrust.UNTRUSTED:

             self._listbox.add(NotTrustedOption())

         self._listbox.add(DeleteOption())

 

                  icon: str,

                  label_text: str,

                  color: str,

-                 type_: Optional[Trust] = None

+                 type_: Optional[OMEMOTrust] = None

                  ) -> None:

 

         Gtk.ListBoxRow.__init__(self)

                             'security-medium-symbolic',

                             _('Blind Trust'),

                             'encrypted-color',

-                            Trust.BLIND)

+                            OMEMOTrust.BLIND)

 

 

 class VerifiedOption(MenuOption):

                             'security-high-symbolic',

                             _('Verified'),

                             'encrypted-color',

-                            Trust.VERIFIED)

+                            OMEMOTrust.VERIFIED)

 

 

 class NotTrustedOption(MenuOption):

                             'dialog-error-symbolic',

                             _('Untrusted'),

                             'error-color',

-                            Trust.UNTRUSTED)

+                            OMEMOTrust.UNTRUSTED)

 

 

 class DeleteOption(MenuOption):

   "pycairo>=1.16.0",

   "PyGObject>=3.42.0",

   "qrcode>=7.3.1",

-  "omemo-dr>=0.99.0,<2.0.0",

+  "omemo-dr>=0.99.1,<2.0.0",

   "winsdk ; platform_system == 'Windows'",

 ]

 dynamic = ["version"]

Philipp Hörist pushed to branch master at gajim / gajim

Commits:

9 changed files:

Changes: