Daniel Brötzmann pushed to branch omemo_integration at gajim / gajim

Commits: 3e82098d by wurstsalat at 2022-10-15T13:09:02+02:00 cq: OMEMO: Add type annotations

- - - - - 615f25f7 by wurstsalat at 2022-10-15T13:09:31+02:00 chore: OMEMO: Remove unused method

- - - - -

7 changed files:

- gajim/common/modules/omemo.py - gajim/common/modules/omemobackend/aes.py - gajim/common/modules/omemobackend/devices.py - gajim/common/modules/omemobackend/liteaxolotlstore.py - gajim/common/modules/omemobackend/state.py - gajim/common/modules/omemobackend/util.py - gajim/common/modules/util.py

Changes:

===================================== gajim/common/modules/omemo.py ===================================== @@ -16,25 +16,40 @@

# XEP-0384: OMEMO Encryption

+from __future__ import annotations + +from typing import Any +from typing import Optional + from pathlib import Path

from nbxmpp.namespaces import Namespace +from nbxmpp.protocol import Message from nbxmpp.protocol import NodeProcessed +from nbxmpp.protocol import Presence from nbxmpp.errors import StanzaError from nbxmpp.const import PresenceType from nbxmpp.const import Affiliation +from nbxmpp.structs import MessageProperties +from nbxmpp.structs import OMEMOMessage +from nbxmpp.structs import PresenceProperties from nbxmpp.structs import StanzaHandler from nbxmpp.modules.omemo import create_omemo_message from nbxmpp.modules.omemo import get_key_transport_message from nbxmpp.modules.util import is_error +from nbxmpp.task import Task

from gajim.common import app from gajim.common import configpaths from gajim.common import ged +from gajim.common import types from gajim.common.structs import OutgoingMessage from gajim.common.const import EncryptionData from gajim.common.const import Trust as GajimTrust +from gajim.common.events import MucDiscoUpdate +from gajim.common.events import SignedIn from gajim.common.modules.base import BaseModule +from gajim.common.modules.contacts import GroupchatContact from gajim.common.modules.util import event_node from gajim.common.modules.util import as_task

@@ -85,8 +100,8 @@ class OMEMO(BaseModule): 'request_bundle', ]

- def __init__(self, con): - BaseModule.__init__(self, con) + def __init__(self, client: types.Client) -> None: + BaseModule.__init__(self, client)

self.handlers = [ StanzaHandler(name='message', @@ -101,50 +116,46 @@ def __init__(self, con):

self._register_pubsub_handler(self._devicelist_notification_received) self.register_event('signed-in', ged.CORE, self._on_signed_in) - self.register_event('muc-disco-update', ged.GUI1, self._on_muc_disco_update) + self.register_event( + 'muc-disco-update', ged.GUI1, self._on_muc_disco_update)

self.encryption_name = 'OMEMO' self.allow_groupchat = True

- self._own_jid = self._con.get_own_jid().bare + self._own_jid = self._client.get_own_jid().bare self._backend = self._get_backend()

- self._omemo_groupchats = set() - self._muc_temp_store = {} - self._query_for_bundles = [] - self._device_bundle_querys = [] - self._query_for_devicelists = [] + self._omemo_groupchats: set[str] = set() + self._muc_temp_store: dict[bytes, str] = {} + self._query_for_bundles: list[str] = [] + self._device_bundle_querys: list[int] = [] + self._query_for_devicelists: list[str] = []

- def _on_signed_in(self, event) -> None: + def _on_signed_in(self, _event: SignedIn) -> None: self._log.info('Announce Support after Sign In') self._query_for_bundles = [] self.set_bundle() self.request_devicelist()

- def on_muc_disco_update(self, event): - self._check_if_omemo_capable(event.jid) - - def get_own_jid(self, stripped=False): - if stripped: - return self._con.get_own_jid().bare - return self._con.get_own_jid() + def _on_muc_disco_update(self, event: MucDiscoUpdate) -> None: + self._check_if_omemo_capable(str(event.jid))

@property - def backend(self): + def backend(self) -> OmemoState: return self._backend

- def _get_backend(self): + def _get_backend(self) -> OmemoState: data_dir = Path(configpaths.get('MY_DATA')) db_path = data_dir / f'omemo_{self._own_jid}.db' return OmemoState(self._own_jid, db_path, self._account, self)

- def is_omemo_groupchat(self, room_jid): + def is_omemo_groupchat(self, room_jid: str) -> bool: return room_jid in self._omemo_groupchats

def activate(self): """ Method called when the Plugin is activated in the PluginManager """ - self._con.get_module('Caps').update_caps() + self._client.get_module('Caps').update_caps()

if app.account_is_connected(self._account): self._log.info('Announce Support after Plugin Activation') @@ -161,7 +172,7 @@ def encrypt_message(self, event: OutgoingMessage) -> None: if not event.message: return

- omemo_message = self.backend.encrypt(event.jid, event.message) + omemo_message = self.backend.encrypt(str(event.jid), event.message) if omemo_message is None: raise Exception('Encryption error')

@@ -179,7 +190,12 @@ def encrypt_message(self, event: OutgoingMessage) -> None:

self._debug_print_stanza(event.stanza)

- def _send_key_transport_message(self, typ, jid, devices): + def _send_key_transport_message(self, + typ: str, + jid: str, + devices: list[int] + ) -> None: + omemo_message = self.backend.encrypt_key_transport(jid, devices) if omemo_message is None: self._log.warning('Key transport message to %s (%s) failed', @@ -188,9 +204,14 @@ def _send_key_transport_message(self, typ, jid, devices):

transport_message = get_key_transport_message(typ, jid, omemo_message) self._log.info('Send key transport message %s (%s)', jid, devices) - self._con.connection.send(transport_message) + self._client.connection.send(transport_message) + + def _message_received(self, + _client: types.xmppClient, + stanza: Message, + properties: MessageProperties + ) -> None:

- def _message_received(self, _con, stanza, properties): if not properties.is_omemo: return

@@ -211,6 +232,7 @@ def _message_received(self, _con, stanza, properties):

self._log.info('Message received from: %s', from_jid)

+ assert isinstance(properties.omemo, OMEMOMessage) try: plaintext, fingerprint, trust = self.backend.decrypt_message( properties.omemo, from_jid) @@ -247,28 +269,35 @@ def _message_received(self, _con, stanza, properties):

prepare_stanza(stanza, plaintext) self._debug_print_stanza(stanza) - properties.encrypted = EncryptionData({'name': ENCRYPTION_NAME, - 'fingerprint': fingerprint, - 'trust': GajimTrust[trust.name]}) + properties.encrypted = EncryptionData({ + 'name': ENCRYPTION_NAME, + 'fingerprint': fingerprint, + 'trust': GajimTrust[trust.name]}) + + def _process_muc_message(self, + properties: MessageProperties + ) -> Optional[str]:

- def _process_muc_message(self, properties): resource = properties.jid.resource if properties.muc_ofrom is not None: # History Message from MUC return properties.muc_ofrom.bare

- contact = self._con.get_module('Contacts').get_contact(properties.jid) + contact = self._client.get_module('Contacts').get_contact( + properties.jid) if contact.real_jid is not None: return contact.real_jid.bare

+ 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) if not from_jid: - self._log.error("Can't decrypt GroupChat Message from %s", resource) + self._log.error( + "Can't decrypt GroupChat Message from %s", resource) return return from_jid

- def _process_mam_message(self, properties): + def _process_mam_message(self, properties: MessageProperties) -> str: self._log.info('Message received, archive: %s', properties.mam.archive) if properties.from_muc: self._log.info('MUC MAM Message received') @@ -279,7 +308,12 @@ def _process_mam_message(self, properties): return properties.muc_user.jid.bare return properties.from_.bare

- def _on_muc_user_presence(self, _con, _stanza, properties): + def _on_muc_user_presence(self, + _client: types.xmppClient, + _stanza: Presence, + properties: PresenceProperties + ) -> None: + if properties.type == PresenceType.ERROR: return

@@ -305,7 +339,7 @@ def _on_muc_user_presence(self, _con, _stanza, properties): self._log.info('%s not in Roster, query devicelist...', jid) self.request_devicelist(jid)

- def get_affiliation_list(self, room_jid): + def get_affiliation_list(self, room_jid: str) -> None: for affiliation in ('owner', 'admin', 'member'): self._nbxmpp('MUC').get_affiliation( room_jid, @@ -313,7 +347,7 @@ def get_affiliation_list(self, room_jid): callback=self._on_affiliations_received, user_data=room_jid)

- def _on_affiliations_received(self, task): + def _on_affiliations_received(self, task: Task) -> None: room_jid = task.get_user_data() try: result = task.finish() @@ -330,24 +364,24 @@ def _on_affiliations_received(self, task): self._log.info('%s not in Roster, query devicelist...', jid) self.request_devicelist(jid)

- def is_contact_in_roster(self, jid): + def is_contact_in_roster(self, jid: str) -> bool: if jid == self._own_jid: return True

- roster_item = self._con.get_module('Roster').get_item(jid) + roster_item = self._client.get_module('Roster').get_item(jid) if roster_item is None: return False

- contact = self._con.get_module('Contacts').get_contact(jid) + contact = self._client.get_module('Contacts').get_contact(jid) return contact.subscription == 'both'

- def on_room_joined(self, contact): + def on_room_joined(self, contact: GroupchatContact) -> None: jid = str(contact.jid) self._check_if_omemo_capable(jid) if self.is_omemo_groupchat(jid): self.get_affiliation_list(jid)

- def _check_if_omemo_capable(self, jid): + def _check_if_omemo_capable(self, jid: str) -> None: disco_info = app.storage.cache.get_last_disco_info(jid) if disco_info.muc_is_members_only and disco_info.muc_is_nonanonymous: self._log.info('OMEMO room discovered: %s', jid) @@ -356,7 +390,7 @@ def _check_if_omemo_capable(self, jid): self._log.info('OMEMO room removed due to config change: %s', jid) self._omemo_groupchats.discard(jid)

- def _check_for_missing_sessions(self, jid): + def _check_for_missing_sessions(self, jid: str) -> None: devices_without_session = self.backend.devices_without_sessions(jid) for device_id in devices_without_session: if device_id in self._device_bundle_querys: @@ -364,7 +398,7 @@ def _check_for_missing_sessions(self, jid): self._device_bundle_querys.append(device_id) self.request_bundle(jid, device_id)

- def are_keys_missing(self, contact_jid): + def are_keys_missing(self, contact_jid: str) -> bool: """ Checks if devicekeys are missing and queries the bundles

@@ -407,12 +441,12 @@ def are_keys_missing(self, contact_jid): return False return True

- def set_bundle(self): + def set_bundle(self) -> None: self._nbxmpp('OMEMO').set_bundle(self.backend.bundle, self.backend.own_device)

@as_task - def request_bundle(self, jid, device_id): + def request_bundle(self, jid: str, device_id: int): _task = yield

self._log.info('Fetch device bundle %s %s', device_id, jid) @@ -437,19 +471,20 @@ def request_bundle(self, jid, device_id): if ctrl: app.ged.raise_event(OMEMONewFingerprint(chat_control=ctrl))

- def set_devicelist(self, devicelist=None): - devicelist_ = set([self.backend.own_device]) + def set_devicelist(self, devicelist: Optional[list[int]] = None) -> None: + devicelist_: set[int] = set([self.backend.own_device]) if devicelist is not None: devicelist_.update(devicelist) self._log.info('Publishing own devicelist: %s', devicelist_) self._nbxmpp('OMEMO').set_devicelist(devicelist_)

- def clear_devicelist(self): - self.backend.update_devicelist(self._own_jid, [self.backend.own_device]) + def clear_devicelist(self) -> None: + self.backend.update_devicelist( + self._own_jid, [self.backend.own_device]) self.set_devicelist()

@as_task - def request_devicelist(self, jid=None): + def request_devicelist(self, jid: Optional[str] = None): _task = yield

if jid is None: @@ -468,7 +503,12 @@ def request_devicelist(self, jid=None): self._process_devicelist_update(jid, devicelist)

@event_node(Namespace.OMEMO_TEMP_DL) - def _devicelist_notification_received(self, _con, _stanza, properties): + def _devicelist_notification_received(self, + _client: types.xmppClient, + _stanza: Message, + properties: MessageProperties + ) -> None: + if properties.pubsub_event.retracted: return

@@ -476,8 +516,12 @@ def _devicelist_notification_received(self, _con, _stanza, properties):

self._process_devicelist_update(str(properties.jid), devicelist)

- def _process_devicelist_update(self, jid, devicelist): - own_devices = jid is None or self._con.get_own_jid().bare_match(jid) + def _process_devicelist_update(self, + jid: str, + devicelist: list[int] + ) -> None: + + own_devices = jid is None or self._client.get_own_jid().bare_match(jid) if own_devices: jid = self._own_jid

@@ -496,7 +540,7 @@ def _process_devicelist_update(self, jid, devicelist):

self._check_for_missing_sessions(jid)

- def _debug_print_stanza(self, stanza): + def _debug_print_stanza(self, stanza: Any) -> None: stanzastr = '\n' + stanza.__str__(fancy=True) stanzastr = stanzastr[0:-1] self._log.debug(stanzastr)

===================================== gajim/common/modules/omemobackend/aes.py ===================================== @@ -14,10 +14,13 @@ # 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 NamedTuple +from typing import Union

import os import logging -from collections import namedtuple

from cryptography.hazmat.primitives.ciphers import Cipher from cryptography.hazmat.primitives.ciphers import algorithms @@ -26,12 +29,16 @@

log = logging.getLogger('gajim.p.omemo')

-EncryptionResult = namedtuple('EncryptionResult', 'payload key iv') - IV_SIZE = 12

-def _decrypt(key, iv, tag, data): +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), @@ -39,7 +46,7 @@ def _decrypt(key, iv, tag, data): return decryptor.update(data) + decryptor.finalize()

-def aes_decrypt(_key, iv, payload): +def aes_decrypt(_key: bytes, iv: bytes, payload: bytes) -> str: if len(_key) >= 32: # XEP-0384 log.debug('XEP Compliant Key/Tag') @@ -56,13 +63,17 @@ def aes_decrypt(_key, iv, payload): return _decrypt(key, iv, tag, data).decode()

-def aes_decrypt_file(key, iv, payload): +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, key_size, iv_size=IV_SIZE): +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) @@ -76,21 +87,21 @@ def _encrypt(data, key_size, iv_size=IV_SIZE): return key, iv, encryptor.tag, payload

-def aes_encrypt(plaintext): +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): +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(): +def get_new_key() -> bytes: return os.urandom(16)

-def get_new_iv(): +def get_new_iv() -> bytes: return os.urandom(IV_SIZE)

===================================== gajim/common/modules/omemobackend/devices.py ===================================== @@ -14,17 +14,19 @@ # 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 collections import defaultdict

from gajim.common import app

-from gajim.common.modules.util import UNACKNOWLEDGED_COUNT +from gajim.common.modules.omemobackend.util import UNACKNOWLEDGED_COUNT

class DeviceManager: - def __init__(self): - self.__device_store = defaultdict(set) - self.__muc_member_store = defaultdict(set) + def __init__(self) -> None: + 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: @@ -38,7 +40,7 @@ def __init__(self): self._log.info('Load device from storage: %s - %s', jid, device) self.add_device(jid, device)

- def update_devicelist(self, jid, devicelist): + def update_devicelist(self, jid: str, devicelist: list[int]) -> None: for device in list(devicelist): if device == self.own_device: continue @@ -52,35 +54,39 @@ def update_devicelist(self, jid, devicelist): self._log.info('Saved devices for %s', jid) self._storage.setActiveState(jid, devicelist)

- def add_muc_member(self, room_jid, jid): + 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, 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, without_self=True): + 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, device): + def add_device(self, jid: str, device: int) -> None: self.__device_store[jid].add(device)

- def remove_device(self, jid, 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, without_self=False): + 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): - devices_for_encryption = [] + 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) @@ -95,22 +101,25 @@ def get_devices_for_encryption(self, jid): devices_for_encryption += self._get_own_devices_for_encryption() return set(devices_for_encryption)

- def _get_devices_for_muc_encryption(self, jid): - 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): - 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): - 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: @@ -119,18 +128,18 @@ def _get_devices_for_encryption(self, jid): return devices_for_encryption

@property - def own_device(self): + def own_device(self) -> int: return self.__own_device

@property - def devices_for_publish(self): + 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): + def is_own_device_published(self) -> bool: return self.own_device in self.get_devices(self._own_jid)

===================================== gajim/common/modules/omemobackend/liteaxolotlstore.py ===================================== @@ -15,9 +15,16 @@ # 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 NamedTuple +from typing import Optional + import time import sqlite3 from collections import namedtuple +from pathlib import Path

from axolotl.state.axolotlstore import AxolotlStore from axolotl.state.signedprekeyrecord import SignedPreKeyRecord @@ -26,28 +33,29 @@ from axolotl.invalidkeyidexception import InvalidKeyIdException from axolotl.ecc.djbec import DjbECPrivateKey from axolotl.ecc.djbec import DjbECPublicKey +from axolotl.identitykey import IdentityKey from axolotl.identitykeypair import IdentityKeyPair from axolotl.util.medium import Medium from axolotl.util.keyhelper import KeyHelper

from gajim.common import app - -from gajim.common.modules.util import Trust -from gajim.common.modules.util import IdentityKeyExtended -from gajim.common.modules.util import DEFAULT_PREKEY_AMOUNT +from gajim.common.modules.omemobackend.util import Trust +from gajim.common.modules.omemobackend.util import IdentityKeyExtended +from gajim.common.modules.omemobackend.util import DEFAULT_PREKEY_AMOUNT +from gajim.common.modules.util import LogAdapter

def _convert_to_string(text): return text.decode()

-def _convert_identity_key(key): +def _convert_identity_key(key: bytes) -> Optional[IdentityKeyExtended]: if not key: return return IdentityKeyExtended(DjbECPublicKey(key[1:]))

-def _convert_record(record): +def _convert_record(record: bytes) -> SessionRecord: return SessionRecord(serialized=record)

@@ -56,7 +64,7 @@ def _convert_record(record):

class LiteAxolotlStore(AxolotlStore): - def __init__(self, account: str, db_path, log): + def __init__(self, account: str, db_path: Path, log: LogAdapter) -> None: self._log = log self._account = account self._con = sqlite3.connect(db_path, @@ -78,13 +86,16 @@ def __init__(self, account: str, db_path, log): self._log.info("Generating OMEMO keys") self._generate_axolotl_keys()

- def _is_blind_trust_enabled(self): + def _is_blind_trust_enabled(self) -> bool: return app.settings.get_account_setting(self._account, 'omemo_blind_trust')

@staticmethod - def _namedtuple_factory(cursor, row): - fields = [] + def _namedtuple_factory(cursor: sqlite3.Cursor, + row: tuple[Any, ...] + ) -> NamedTuple: + + fields: list[str] = [] for col in cursor.description: if col[0] == '_id': fields.append('id') @@ -98,7 +109,7 @@ def _namedtuple_factory(cursor, row): fields.append(col[0]) return namedtuple("Row", fields)(*row)

- def _generate_axolotl_keys(self): + def _generate_axolotl_keys(self) -> None: identity_key_pair = KeyHelper.generateIdentityKeyPair() registration_id = KeyHelper.getRandomSequence(2147483647) pre_keys = KeyHelper.generatePreKeys( @@ -114,10 +125,10 @@ def _generate_axolotl_keys(self): for pre_key in pre_keys: self.storePreKey(pre_key.getId(), pre_key)

- def user_version(self): + def user_version(self) -> int: return self._con.execute('PRAGMA user_version').fetchone()[0]

- def createDb(self): + def createDb(self) -> None: if self.user_version() == 0:

create_tables = ''' @@ -159,7 +170,7 @@ def createDb(self): """ % (create_tables) self._con.executescript(create_db_sql)

- def migrateDb(self): + def migrateDb(self) -> None: """ Migrates the DB """

@@ -377,7 +388,7 @@ def migrateDb(self): self._con.execute('PRAGMA user_version=12') self._con.commit()

- def loadSignedPreKey(self, signedPreKeyId): + def loadSignedPreKey(self, signedPreKeyId: int) -> SignedPreKeyRecord: query = 'SELECT record FROM signed_prekeys WHERE prekey_id = ?' result = self._con.execute(query, (signedPreKeyId, )).fetchone() if result is None: @@ -385,39 +396,43 @@ def loadSignedPreKey(self, signedPreKeyId): signedPreKeyId) return SignedPreKeyRecord(serialized=result.record)

- def loadSignedPreKeys(self): + def loadSignedPreKeys(self) -> list[SignedPreKeyRecord]: query = 'SELECT record FROM signed_prekeys' results = self._con.execute(query).fetchall() return [SignedPreKeyRecord(serialized=row.record) for row in results]

- def storeSignedPreKey(self, signedPreKeyId, signedPreKeyRecord): + def storeSignedPreKey(self, + signedPreKeyId: int, + signedPreKeyRecord: SignedPreKeyRecord + ) -> None: + query = 'INSERT INTO signed_prekeys (prekey_id, record) VALUES(?,?)' self._con.execute(query, (signedPreKeyId, signedPreKeyRecord.serialize())) self._con.commit()

- def containsSignedPreKey(self, signedPreKeyId): + def containsSignedPreKey(self, signedPreKeyId: int) -> bool: query = 'SELECT record FROM signed_prekeys WHERE prekey_id = ?' result = self._con.execute(query, (signedPreKeyId,)).fetchone() return result is not None

- def removeSignedPreKey(self, signedPreKeyId): + def removeSignedPreKey(self, signedPreKeyId: int) -> None: query = 'DELETE FROM signed_prekeys WHERE prekey_id = ?' self._con.execute(query, (signedPreKeyId,)) self._con.commit()

- def getNextSignedPreKeyId(self): + def getNextSignedPreKeyId(self) -> int: result = self.getCurrentSignedPreKeyId() if result is None: return 1 # StartId if no SignedPreKeys exist return (result % (Medium.MAX_VALUE - 1)) + 1

- def getCurrentSignedPreKeyId(self): + def getCurrentSignedPreKeyId(self) -> Optional[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

- def getSignedPreKeyTimestamp(self, signedPreKeyId): + def getSignedPreKeyTimestamp(self, signedPreKeyId: int) -> int: query = '''SELECT strftime('%s', timestamp) FROM signed_prekeys WHERE prekey_id = ?'''

@@ -428,19 +443,19 @@ def getSignedPreKeyTimestamp(self, signedPreKeyId):

return result.formated_time

- def removeOldSignedPreKeys(self, timestamp): + def removeOldSignedPreKeys(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, deviceId): + def loadSession(self, recipientId: str, deviceId: 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() return result.record if result is not None else SessionRecord()

- def getJidFromDevice(self, device_id): + def getJidFromDevice(self, device_id: int) -> Optional[str]: query = '''SELECT recipient_id FROM sessions WHERE device_id = ?''' result = self._con.execute(query, (device_id, )).fetchone() @@ -451,7 +466,12 @@ def getActiveDeviceTuples(self): FROM sessions WHERE active = 1''' return self._con.execute(query).fetchall()

- def storeSession(self, recipientId, deviceId, sessionRecord): + def storeSession(self, + recipientId: str, + deviceId: int, + sessionRecord: SessionRecord + ) -> None: + query = '''INSERT INTO sessions(recipient_id, device_id, record) VALUES(?,?,?)''' try: @@ -467,24 +487,24 @@ def storeSession(self, recipientId, deviceId, sessionRecord):

self._con.commit()

- def containsSession(self, recipientId, deviceId): + def containsSession(self, recipientId: str, deviceId: int) -> bool: query = '''SELECT record FROM sessions WHERE recipient_id = ? AND device_id = ?''' result = self._con.execute(query, (recipientId, deviceId)).fetchone() return result is not None

- def deleteSession(self, recipientId, deviceId): + def deleteSession(self, recipientId: str, deviceId: int) -> None: self._log.info('Delete session for %s %s', recipientId, deviceId) query = "DELETE FROM sessions WHERE recipient_id = ? AND device_id = ?" self._con.execute(query, (recipientId, deviceId)) self._con.commit()

- def deleteAllSessions(self, recipientId): + def deleteAllSessions(self, recipientId: str) -> None: query = 'DELETE FROM sessions WHERE recipient_id = ?' self._con.execute(query, (recipientId,)) self._con.commit()

- def getSessionsFromJid(self, recipientId): + def getSessionsFromJid(self, recipientId: str): query = '''SELECT recipient_id, device_id, record as "record [session_record]", @@ -492,7 +512,7 @@ def getSessionsFromJid(self, recipientId): FROM sessions WHERE recipient_id = ?''' return self._con.execute(query, (recipientId,)).fetchall()

- def getSessionsFromJids(self, recipientIds): + def getSessionsFromJids(self, recipientIds: list[str]): query = '''SELECT recipient_id, device_id, record as "record [session_record]", @@ -502,7 +522,7 @@ def getSessionsFromJids(self, recipientIds): ', '.join(['?'] * len(recipientIds))) return self._con.execute(query, recipientIds).fetchall()

- def setActiveState(self, jid, devicelist): + def setActiveState(self, jid: str, devicelist: list[int]) -> None: query = '''UPDATE sessions SET active = 1 WHERE recipient_id = ? AND device_id IN ({})'''.format( ', '.join(['?'] * len(devicelist))) @@ -514,24 +534,27 @@ def setActiveState(self, jid, devicelist): self._con.execute(query, (jid,) + tuple(devicelist)) self._con.commit()

- def setInactive(self, jid, device_id): + def setInactive(self, jid: 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.commit()

- def getInactiveSessionsKeys(self, recipientId): + def getInactiveSessionsKeys(self, + recipientId: str + ) -> list[IdentityKeyExtended]: + query = '''SELECT record as "record [session_record]" FROM sessions WHERE active = 0 AND recipient_id = ?''' results = self._con.execute(query, (recipientId,)).fetchall()

- keys = [] + keys: list[IdentityKeyExtended] = [] for result in results: key = result.record.getSessionState().getRemoteIdentityKey() keys.append(IdentityKeyExtended(key.getPublicKey())) return keys

- def loadPreKey(self, preKeyId): + def loadPreKey(self, preKeyId: int) -> PreKeyRecord: query = '''SELECT record FROM prekeys WHERE prekey_id = ?'''

result = self._con.execute(query, (preKeyId,)).fetchone() @@ -539,41 +562,41 @@ def loadPreKey(self, preKeyId): raise Exception("No such prekeyRecord!") return PreKeyRecord(serialized=result.record)

- def loadPendingPreKeys(self): + def loadPendingPreKeys(self) -> list[PreKeyRecord]: query = '''SELECT record FROM prekeys''' result = self._con.execute(query).fetchall() return [PreKeyRecord(serialized=row.record) for row in result]

- def storePreKey(self, preKeyId, preKeyRecord): + def storePreKey(self, preKeyId: int, preKeyRecord: PreKeyRecord) -> None: query = 'INSERT INTO prekeys (prekey_id, record) VALUES(?,?)' self._con.execute(query, (preKeyId, preKeyRecord.serialize())) self._con.commit()

- def containsPreKey(self, preKeyId): + def containsPreKey(self, preKeyId: int) -> bool: query = 'SELECT record FROM prekeys WHERE prekey_id = ?' result = self._con.execute(query, (preKeyId,)).fetchone() return result is not None

- def removePreKey(self, preKeyId): + def removePreKey(self, preKeyId: int) -> None: query = 'DELETE FROM prekeys WHERE prekey_id = ?' self._con.execute(query, (preKeyId,)) self._con.commit()

- def getCurrentPreKeyId(self): + def getCurrentPreKeyId(self) -> int: query = 'SELECT MAX(prekey_id) FROM prekeys' return self._con.execute(query).fetchone().max_prekey_id

- def getPreKeyCount(self): + def getPreKeyCount(self) -> int: query = 'SELECT COUNT(prekey_id) FROM prekeys' return self._con.execute(query).fetchone().count_prekey_id

- def generateNewPreKeys(self, count): + 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): + def getIdentityKeyPair(self) -> IdentityKeyPair: query = '''SELECT public_key as "public_key [pk]", private_key FROM secret LIMIT 1''' result = self._con.execute(query).fetchone() @@ -581,12 +604,16 @@ def getIdentityKeyPair(self): return IdentityKeyPair(result.public_key, DjbECPrivateKey(result.private_key))

- def getLocalRegistrationId(self): + def getLocalRegistrationId(self) -> Optional[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

- def storeLocalData(self, device_id, identityKeyPair): + def storeLocalData(self, + device_id: int, + identityKeyPair: IdentityKeyPair + ) -> None: + query = 'SELECT * FROM secret' result = self._con.execute(query).fetchone() if result is not None: @@ -602,7 +629,7 @@ def storeLocalData(self, device_id, identityKeyPair): self._con.execute(query, (device_id, public_key, private_key)) self._con.commit()

- def saveIdentity(self, recipientId, identityKey): + def saveIdentity(self, recipientId: int, identityKey: IdentityKey) -> None: query = '''INSERT INTO identities (recipient_id, public_key, trust, shown) VALUES(?, ?, ?, ?)''' if not self.containsIdentity(recipientId, identityKey): @@ -613,7 +640,11 @@ def saveIdentity(self, recipientId, identityKey): 1 if trust == Trust.BLIND else 0)) self._con.commit()

- def containsIdentity(self, recipientId, identityKey): + def containsIdentity(self, + recipientId: str, + identityKey: IdentityKey + ) -> bool: + query = '''SELECT * FROM identities WHERE recipient_id = ? AND public_key = ?'''

@@ -623,24 +654,36 @@ def containsIdentity(self, recipientId, identityKey):

return result is not None

- def deleteIdentity(self, recipientId, identityKey): + def deleteIdentity(self, + recipientId: str, + identityKey: IdentityKey + ) -> None: + query = '''DELETE FROM identities WHERE recipient_id = ? AND public_key = ?''' public_key = identityKey.getPublicKey().serialize() self._con.execute(query, (recipientId, public_key)) self._con.commit()

- def isTrustedIdentity(self, recipientId, identityKey): + def isTrustedIdentity(self, + recipientId: str, + identityKey: IdentityKey + ) -> bool: + return True

- def getTrustForIdentity(self, recipientId, identityKey): + def getTrustForIdentity(self, + recipientId: str, + identityKey: IdentityKey + ) -> Optional[Trust]: + 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() return result.trust if result is not None else None

- def getFingerprints(self, jid): + def getFingerprints(self, jid: str): query = '''SELECT recipient_id, public_key as "public_key [pk]", trust, @@ -649,7 +692,7 @@ def getFingerprints(self, jid): WHERE recipient_id = ? ORDER BY trust ASC''' return self._con.execute(query, (jid,)).fetchall()

- def getMucFingerprints(self, jids): + def getMucFingerprints(self, jids: list[str]): query = ''' SELECT recipient_id, public_key as "public_key [pk]", @@ -661,7 +704,7 @@ def getMucFingerprints(self, jids):

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

- def hasUndecidedFingerprints(self, jid): + def hasUndecidedFingerprints(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() @@ -671,7 +714,7 @@ def hasUndecidedFingerprints(self, jid): undecided = set(undecided) - set(inactive) return bool(undecided)

- def getDefaultTrust(self, jid): + def getDefaultTrust(self, jid: str) -> Trust: if not self._is_blind_trust_enabled(): return Trust.UNDECIDED

@@ -682,33 +725,38 @@ def getDefaultTrust(self, jid): return Trust.BLIND return Trust.UNDECIDED

- def getTrustedFingerprints(self, jid): + def getTrustedFingerprints(self, jid: str) -> list[IdentityKeyExtended]: 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() return [row.public_key for row in result]

- def getNewFingerprints(self, jid): + def getNewFingerprints(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): + def setShownFingerprints(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, identityKey, trust): + def setTrust(self, + recipient_id: str, + identityKey: IdentityKey, + trust: Trust + ) -> None: + query = '''UPDATE identities SET trust = ? WHERE public_key = ? AND recipient_id = ?''' public_key = identityKey.getPublicKey().serialize() self._con.execute(query, (trust, public_key, recipient_id)) self._con.commit()

- def isTrusted(self, recipient_id, device_id): + def isTrusted(self, recipient_id: str, device_id: int) -> bool: record = self.loadSession(recipient_id, device_id) if record.isFresh(): return False @@ -716,7 +764,11 @@ def isTrusted(self, recipient_id, device_id): return self.getTrustForIdentity( recipient_id, identity_key) in (Trust.VERIFIED, Trust.BLIND)

- def getIdentityLastSeen(self, recipient_id, identity_key): + def getIdentityLastSeen(self, + recipient_id: str, + identity_key: IdentityKey + ) -> Optional[int]: + identity_key = identity_key.getPublicKey().serialize() query = '''SELECT timestamp FROM identities WHERE recipient_id = ? AND public_key = ?''' @@ -724,7 +776,11 @@ def getIdentityLastSeen(self, recipient_id, identity_key): identity_key)).fetchone() return result.timestamp if result is not None else None

- def setIdentityLastSeen(self, recipient_id, identity_key): + def setIdentityLastSeen(self, + recipient_id: str, + identity_key: IdentityKey + ) -> None: + timestamp = int(time.time()) identity_key = identity_key.getPublicKey().serialize() self._log.info('Set last seen for %s %s', recipient_id, timestamp) @@ -733,7 +789,7 @@ def setIdentityLastSeen(self, recipient_id, identity_key): self._con.execute(query, (timestamp, recipient_id, identity_key)) self._con.commit()

- def getUnacknowledgedCount(self, recipient_id, device_id): + def getUnacknowledgedCount(self, recipient_id: str, device_id: int) -> int: record = self.loadSession(recipient_id, device_id) if record.isFresh(): return 0

===================================== gajim/common/modules/omemobackend/state.py ===================================== @@ -15,8 +15,14 @@ # 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 @@ -24,6 +30,7 @@ from axolotl.ecc.djbec import DjbECPublicKey from axolotl.identitykey import IdentityKey

+from axolotl.identitykeypair import IdentityKeyPair from axolotl.protocol.prekeywhispermessage import PreKeyWhisperMessage from axolotl.protocol.whispermessage import WhisperMessage from axolotl.sessionbuilder import SessionBuilder @@ -32,6 +39,7 @@ from axolotl.util.keyhelper import KeyHelper from axolotl.duplicatemessagexception import DuplicateMessageException

+from gajim.common import types from gajim.common.modules.omemobackend.aes import aes_decrypt from gajim.common.modules.omemobackend.aes import aes_encrypt from gajim.common.modules.omemobackend.aes import get_new_key @@ -39,21 +47,28 @@ from gajim.common.modules.omemobackend.devices import DeviceManager from gajim.common.modules.omemobackend.devices import NoDevicesFound from gajim.common.modules.omemobackend.liteaxolotlstore import LiteAxolotlStore -from gajim.common.modules.util import get_fingerprint -from gajim.common.modules.util import Trust -from gajim.common.modules.util import DEFAULT_PREKEY_AMOUNT -from gajim.common.modules.util import MIN_PREKEY_AMOUNT -from gajim.common.modules.util import SPK_CYCLE_TIME -from gajim.common.modules.util import SPK_ARCHIVE_TIME -from gajim.common.modules.util import UNACKNOWLEDGED_COUNT +from gajim.common.modules.omemobackend.util import get_fingerprint +from gajim.common.modules.omemobackend.util import Trust +from gajim.common.modules.omemobackend.util import DEFAULT_PREKEY_AMOUNT +from gajim.common.modules.omemobackend.util import MIN_PREKEY_AMOUNT +from gajim.common.modules.omemobackend.util import SPK_CYCLE_TIME +from gajim.common.modules.omemobackend.util import SPK_ARCHIVE_TIME +from gajim.common.modules.omemobackend.util import UNACKNOWLEDGED_COUNT

class OmemoState(DeviceManager): - def __init__(self, own_jid, db_path, account, xmpp_con): + def __init__(self, + own_jid: str, + db_path: Path, + account: str, + xmpp_con: types.xmppClient + ) -> None: + self._account = account self._own_jid = own_jid self._log = xmpp_con._log - self._session_ciphers = defaultdict(dict) + self._session_ciphers: dict[ + str, dict[int, SessionCipher]] = defaultdict(dict) self._storage = LiteAxolotlStore(account, db_path, self._log)

DeviceManager.__init__(self) @@ -63,7 +78,12 @@ def __init__(self, own_jid, db_path, account, xmpp_con): self._log.info('%s PreKeys available', self._storage.getPreKeyCount())

- def build_session(self, jid, device_id, bundle): + 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)

@@ -88,18 +108,18 @@ def build_session(self, jid, device_id, bundle): self._get_session_cipher(jid, device_id)

@property - def storage(self): + def storage(self) -> LiteAxolotlStore: return self._storage

@property - def own_fingerprint(self): + def own_fingerprint(self) -> str: return get_fingerprint(self._storage.getIdentityKeyPair())

@property - def bundle(self): + def bundle(self) -> OMEMOBundle: self._check_pre_key_count()

- bundle = {'otpks': []} + bundle: dict[str, Any] = {'otpks': []} for k in self._storage.loadPendingPreKeys(): key = k.getKeyPair().getPublicKey().serialize() bundle['otpks'].append({'key': key, 'id': k.getId()}) @@ -117,7 +137,11 @@ def bundle(self):

return OMEMOBundle(**bundle)

- def decrypt_message(self, omemo_message, jid): + 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 @@ -157,13 +181,18 @@ def decrypt_message(self, omemo_message, jid): self._log.debug("Decrypted Message => %s", result) return result, fingerprint, trust

- def _get_whisper_message(self, jid, device, key): + 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, plaintext): + def encrypt(self, jid: str, plaintext: str) -> Optional[OMEMOMessage]: try: devices_for_encryption = self.get_devices_for_encryption(jid) except NoDevicesFound: @@ -171,7 +200,8 @@ def encrypt(self, jid, plaintext): return

result = aes_encrypt(plaintext) - whisper_messages = defaultdict(dict) + whisper_messages: dict[ + str, dict[int, tuple[bytes, bool]]] = defaultdict(dict)

for jid_, device in devices_for_encryption: count = self._storage.getUnacknowledgedCount(jid_, device) @@ -194,7 +224,7 @@ def encrypt(self, jid, plaintext): self._log.error('Encrypted keys empty') return

- encrypted_keys = {} + encrypted_keys: dict[int, tuple[bytes, bool]] = {} for jid_ in whisper_messages: encrypted_keys.update(whisper_messages[jid_])

@@ -204,8 +234,13 @@ def encrypt(self, jid, plaintext): iv=result.iv, payload=result.payload)

- def encrypt_key_transport(self, jid, devices): - whisper_messages = defaultdict(dict) + 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( @@ -224,12 +259,12 @@ def encrypt_key_transport(self, jid, devices): iv=get_new_iv(), payload=None)

- def has_trusted_keys(self, jid): + 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): + 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 @@ -239,7 +274,7 @@ def devices_without_sessions(self, jid): jid, missing_devices) return missing_devices

- def _get_session_cipher(self, jid, device_id): + def _get_session_cipher(self, jid: str, device_id: int) -> SessionCipher: try: return self._session_ciphers[jid][device_id] except KeyError: @@ -248,7 +283,12 @@ def _get_session_cipher(self, jid, device_id): self._session_ciphers[jid][device_id] = cipher return cipher

- def _process_pre_key_message(self, jid, device, key): + 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(serialized=key) if not pre_key_message.getPreKeyId(): @@ -268,7 +308,12 @@ def _process_pre_key_message(self, jid, device, key): self.add_device(jid, device) return key, fingerprint, trust

- def _process_message(self, jid, device, key): + def _process_message(self, + jid: str, + device: int, + key: bytes + ) -> tuple[bytes, str, Trust]: + self._log.info('Process message from %s', jid) message = WhisperMessage(serialized=key)

@@ -290,15 +335,23 @@ def _get_identity_key_from_pk_message(key): pre_key_message = PreKeyWhisperMessage(serialized=key) return pre_key_message.getIdentityKey()

- def _get_identity_key_from_device(self, jid, device): + 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, identity_key): + 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): + 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: @@ -306,7 +359,7 @@ def _check_pre_key_count(self): self._storage.generateNewPreKeys(missing_count) self._log.info('%s PreKeys created', missing_count)

- def _cycle_signed_pre_key(self, ik_pair): + def _cycle_signed_pre_key(self, ik_pair: IdentityKeyPair) -> None: # Publish every SPK_CYCLE_TIME a new SignedPreKey # Delete all exsiting SignedPreKeys that are older # then SPK_ARCHIVE_TIME

===================================== gajim/common/modules/omemobackend/util.py ===================================== @@ -14,11 +14,15 @@ # 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 axolotl.identitykey import IdentityKey +from axolotl.identitykeypair import IdentityKeyPair +

DEFAULT_PREKEY_AMOUNT = 100 MIN_PREKEY_AMOUNT = 80 @@ -34,7 +38,10 @@ class Trust(IntEnum): BLIND = 3

-def get_fingerprint(identity_key, formatted=False): +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: @@ -49,8 +56,8 @@ def get_fingerprint(identity_key, formatted=False):

class IdentityKeyExtended(IdentityKey): - def __hash__(self): + def __hash__(self) -> int: return hash(self.publicKey.serialize())

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

===================================== gajim/common/modules/util.py ===================================== @@ -17,6 +17,7 @@ from __future__ import annotations

from typing import Any +from typing import Optional from typing import Union

import logging @@ -187,13 +188,17 @@ def check_if_message_correction(properties: MessageProperties, return True

-def prepare_stanza(stanza, plaintext): +def prepare_stanza(stanza: Message, plaintext: str) -> None: delete_nodes(stanza, 'encrypted', Namespace.OMEMO_TEMP) delete_nodes(stanza, 'body') stanza.setBody(plaintext)

-def delete_nodes(stanza, name, namespace=None): +def delete_nodes(stanza: Message, + name: str, + namespace: Optional[str] = None + ) -> None: + nodes = stanza.getTags(name, namespace=namespace) for node in nodes: stanza.delChild(node)

View it on GitLab: https://dev.gajim.org/gajim/gajim/-/compare/41a6ca8a6fec1085cad7fe21fb0c710e...