GitLab

at 2023-05-07T10:45:14+02:00

refactor: OMEMO: Change session import

refactor: OMEMO: Simplify deleting a session

refactor: OMEMO: Remove legacy code

Code stems from a time where MAM archives did not communicate the real jid.

refactor: OMEMO: Access set_trust() via SessionManager

refactor: OMEMO: Remove obsolete code

refactor: OMEMO: Use get_identity_infos() in more places

 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.session_manager import OMEMOSessionManager

 from omemo_dr.structs import OMEMOBundle

 from omemo_dr.structs import OMEMOConfig

                                    spk_cycle_seconds=86400,

                                    unacknowledged_count=2000)

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

+        self._backend = OMEMOSessionManager(

+            self._own_jid, storage, omemo_config)

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

                                       self._on_republish_bundle)

             self.get_affiliation_list(jid)

     def _on_republish_bundle(self,

-                             _session: OMEMOSession,

+                             _session: OMEMOSessionManager,

                              _signal_name: str,

                              bundle: OMEMOBundle

                              ) -> None:

         self.set_bundle(bundle=bundle)

     @property

-    def backend(self) -> OMEMOSession:

+    def backend(self) -> OMEMOSessionManager:

         return self._backend

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

                     message=EncryptionInfoMsg.QUERY_DEVICES))

                 return False

-            # check if bundles are missing for some devices

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

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

-                               contact.account, jid)

-                app.ged.raise_event(EncryptionInfo(

-                    account=contact.account,

-                    jid=contact.jid,

-                    message=EncryptionInfoMsg.UNDECIDED_FINGERPRINTS))

-                return False

-

-        if self._new_fingerprints_available(contact):

+        if self.backend.get_identity_infos(jid,

+                                           only_active=True,

+                                           trust=OMEMOTrust.UNDECIDED):

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

+                           contact.account, jid)

+            app.ged.raise_event(EncryptionInfo(

+                account=contact.account,

+                jid=contact.jid,

+                message=EncryptionInfoMsg.UNDECIDED_FINGERPRINTS))

             return False

         self._log.debug('%s => Sending Message to %s',

         return True

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

-        fingerprints: list[int] = []

-        if contact.is_groupchat:

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

-                                                           without_self=False):

-                fingerprints = self.backend.storage.get_new_fingerprints(

-                    member_jid)

-                if fingerprints:

-                    break

-

-        else:

-            fingerprints = self.backend.storage.get_new_fingerprints(

-                str(contact.jid))

-

-        if not fingerprints:

-            return False

-

-        app.ged.raise_event(EncryptionInfo(

-            account=contact.account,

-            jid=contact.jid,

-            message=EncryptionInfoMsg.UNDECIDED_FINGERPRINTS))

-

-        return True

-

     def is_omemo_groupchat(self, room_jid: str) -> bool:

         return room_jid in self._omemo_groupchats

         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.get_jid_from_device(

-            properties.omemo.sid)

-        if not from_jid:

-            self._log.error(

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

-            return None

-        return from_jid

+        self._log.error(

+            'Unable to find JID for group chat message from %s', resource)

+        return None

     def _process_mam_message(self,

                              properties: MessageProperties

                 for device_id in devices_without_session:

                     self.request_bundle(contact_jid, device_id)

-        if self.backend.has_trusted_keys(contact_jid):

+        if self.backend.get_identity_infos(

+                contact_jid,

+                only_active=True,

+                trust=[OMEMOTrust.VERIFIED, OMEMOTrust.BLIND]):

             return False

+

         return True

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

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

-    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, OMEMOTrust.UNDECIDED)).fetchall()

-        undecided = [row.public_key for row in result]

-

-        inactive = self.get_inactive_sessions_keys(jid)

-        undecided = set(undecided) - set(inactive)

-        return bool(undecided)

-

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

         if not self._is_blind_trust_enabled():

             return OMEMOTrust.UNDECIDED

             return OMEMOTrust.BLIND

         return OMEMOTrust.UNDECIDED

-    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, (address,)).fetchall()

-        return [row.public_key for row in result]

-

-    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 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 set_trust(self,

                   recipient_id: str,

                   identity_key: IdentityKey,

     def delete_fingerprint(self, *args: Any) -> None:

         def _remove():

-            device_id = self._identity_info.device_id

-

-            self._omemo.backend.remove_device(self._address, device_id)

-            self._omemo.backend.storage.delete_session(self._address,

-                                                       device_id)

-            self._omemo.backend.storage.delete_identity(

-                self._address, self._identity_info.public_key)

+            self._omemo.backend.delete_session(

+                self._address,

+                self._identity_info.device_id,

+                delete_identity=True)

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

             listbox.remove(self)

         image.get_style_context().add_class(css_class)

         image.set_tooltip_text(tooltip)

-        self._omemo.backend.storage.set_trust(

+        self._omemo.backend.set_trust(

             self._address, self._identity_info.public_key, trust)

     @property

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

Commits:

3 changed files:

Changes: