libervia-backend: sat/memory/encryption.py comparison

comparison sat/memory/encryption.py @ 4037:524856bd7b19

massive refactoring to switch from camelCase to snake_case: historically, Libervia (SàT before) was using camelCase as allowed by PEP8 when using a pre-PEP8 code, to use the same coding style as in Twisted. However, snake_case is more readable and it's better to follow PEP8 best practices, so it has been decided to move on full snake_case. Because Libervia has a huge codebase, this ended with a ugly mix of camelCase and snake_case. To fix that, this patch does a big refactoring by renaming every function and method (including bridge) that are not coming from Twisted or Wokkel, to use fully snake_case. This is a massive change, and may result in some bugs.

author	Goffi <goffi@goffi.org>
date	Sat, 08 Apr 2023 13:54:42 +0200
parents	cc2705225778
children	c23cad65ae99

comparison

equal deleted inserted replaced

-:c4464d7ae97b
+:524856bd7b19
 @property
 def host(self):
 return self.client.host_app
-async def loadSessions(self):
+async def load_sessions(self):
 """Load persistent sessions"""
 await self._stored_session.load()
 start_d_list = []
 for entity_jid_s, namespace in self._stored_session.items():
 entity = jid.JID(entity_jid_s)
 "Could not restart {namespace!r} encryption with {entity}: {err}"
 ).format(namespace=namespace, entity=entity_jid_s, err=err))
 log.info(_("encryption sessions restored"))
 @classmethod
-def registerPlugin(cls, plg_instance, name, namespace, priority=0, directed=False):
+def register_plugin(cls, plg_instance, name, namespace, priority=0, directed=False):
 """Register a plugin handling an encryption algorithm
 @param plg_instance(object): instance of the plugin
 it must have the following methods:
-- getTrustUI(entity): return a XMLUI for trust management
+- get_trust_ui(entity): return a XMLUI for trust management
 entity(jid.JID): entity to manage
 The returned XMLUI must be a form
 if may have the following methods:
-- startEncryption(entity): start encrypted session
+- start_encryption(entity): start encrypted session
 entity(jid.JID): entity to start encrypted session with
-- stopEncryption(entity): start encrypted session
+- stop_encryption(entity): start encrypted session
 entity(jid.JID): entity to stop encrypted session with
 if they don't exists, those 2 methods will be ignored.
 @param name(unicode): human readable name of the encryption algorithm
 @param namespace(unicode): namespace of the encryption algorithm
 @classmethod
 def getPlugins(cls):
 return cls.plugins
 @classmethod
-def getPlugin(cls, namespace):
+def get_plugin(cls, namespace):
 try:
 return next(p for p in cls.plugins if p.namespace == namespace)
 except StopIteration:
 raise exceptions.NotFound(_(
 "Can't find requested encryption plugin: {namespace}").format(
 namespace=namespace))
 @classmethod
-def getNamespaces(cls):
+def get_namespaces(cls):
 """Get available plugin namespaces"""
 return {p.namespace for p in cls.getPlugins()}
 @classmethod
-def getNSFromName(cls, name):
+def get_ns_from_name(cls, name):
 """Retrieve plugin namespace from its name
 @param name(unicode): name of the plugin (case insensitive)
 @return (unicode): namespace of the plugin
 @raise exceptions.NotFound: there is not encryption plugin of this name
 return p.namespace
 raise exceptions.NotFound(_(
 "Can't find a plugin with the name \"{name}\".".format(
 name=name)))
-def getBridgeData(self, session):
+def get_bridge_data(self, session):
 """Retrieve session data serialized for bridge.
 @param session(dict): encryption session
 @return (unicode): serialized data for bridge
 """
 if 'directed_devices' in session:
 bridge_data['directed_devices'] = session['directed_devices']
 return data_format.serialise(bridge_data)
-async def _startEncryption(self, plugin, entity):
+async def _start_encryption(self, plugin, entity):
 """Start encryption with a plugin
 This method must be called just before adding a plugin session.
 StartEncryptionn method of plugin will be called if it exists.
 """
 if not plugin.directed:
 await self._stored_session.aset(entity.userhost(), plugin.namespace)
 try:
-start_encryption = plugin.instance.startEncryption
+start_encryption = plugin.instance.start_encryption
 except AttributeError:
-log.debug(f"No startEncryption method found for {plugin.namespace}")
+log.debug(f"No start_encryption method found for {plugin.namespace}")
 else:
 # we copy entity to avoid having the resource changed by stop_encryption
-await utils.asDeferred(start_encryption, self.client, copy.copy(entity))
+await utils.as_deferred(start_encryption, self.client, copy.copy(entity))
-async def _stopEncryption(self, plugin, entity):
+async def _stop_encryption(self, plugin, entity):
 """Stop encryption with a plugin
 This method must be called just before removing a plugin session.
 StopEncryptionn method of plugin will be called if it exists.
 """
 try:
 await self._stored_session.adel(entity.userhost())
 except KeyError:
 pass
 try:
-stop_encryption = plugin.instance.stopEncryption
+stop_encryption = plugin.instance.stop_encryption
 except AttributeError:
-log.debug(f"No stopEncryption method found for {plugin.namespace}")
+log.debug(f"No stop_encryption method found for {plugin.namespace}")
 else:
 # we copy entity to avoid having the resource changed by stop_encryption
-return utils.asDeferred(stop_encryption, self.client, copy.copy(entity))
+return utils.as_deferred(stop_encryption, self.client, copy.copy(entity))
 async def start(self, entity, namespace=None, replace=False):
 """Start an encryption session with an entity
 @param entity(jid.JID): entity to start an encryption session with
 "an encryption session can't be started"))
 if namespace is None:
 plugin = self.plugins[0]
 else:
-plugin = self.getPlugin(namespace)
+plugin = self.get_plugin(namespace)
 bare_jid = entity.userhostJID()
 if bare_jid in self._sessions:
 # we have already an encryption session with this contact
 former_plugin = self._sessions[bare_jid]["plugin"]
 if replace:
 # there is a conflict, but replacement is requested
 # so we stop previous encryption to use new one
 del self._sessions[bare_jid]
-await self._stopEncryption(former_plugin, entity)
+await self._stop_encryption(former_plugin, entity)
 else:
 msg = (_("Session with {bare_jid} is already encrypted with {name}. "
 "Please stop encryption session before changing algorithm.")
 .format(bare_jid=bare_jid, name=plugin.name))
 log.warning(msg)
 raise exceptions.ConflictError(msg)
 data = {"plugin": plugin}
 if plugin.directed:
 if not entity.resource:
-entity.resource = self.host.memory.getMainResource(self.client, entity)
+entity.resource = self.host.memory.main_resource_get(self.client, entity)
 if not entity.resource:
 raise exceptions.NotFound(
 _("No resource found for {destinee}, can't encrypt with {name}")
 .format(destinee=entity.full(), name=plugin.name))
 log.info(_("No resource specified to encrypt with {name}, using "
 # indicate that we encrypt only for some devices
 directed_devices = data['directed_devices'] = [entity.resource]
 elif entity.resource:
 raise ValueError(_("{name} encryption must be used with bare jids."))
-await self._startEncryption(plugin, entity)
+await self._start_encryption(plugin, entity)
 self._sessions[entity.userhostJID()] = data
 log.info(_("Encryption session has been set for {entity_jid} with "
 "{encryption_name}").format(
 entity_jid=entity.full(), encryption_name=plugin.name))
-self.host.bridge.messageEncryptionStarted(
+self.host.bridge.message_encryption_started(
 entity.full(),
-self.getBridgeData(data),
+self.get_bridge_data(data),
 self.client.profile)
 msg = D_("Encryption session started: your messages with {destinee} are "
 "now end to end encrypted using {name} algorithm.").format(
 destinee=entity.full(), name=plugin.name)
 directed_devices = data.get('directed_devices')
 if not directed_devices:
 # if we have no more directed device sessions,
 # we stop the whole session
 # see comment below for deleting session before stopping encryption
 del self._sessions[entity.userhostJID()]
-await self._stopEncryption(plugin, entity)
+await self._stop_encryption(plugin, entity)
 else:
-# plugin's stopEncryption may call stop again (that's the case with OTR)
+# plugin's stop_encryption may call stop again (that's the case with OTR)
-# so we need to remove plugin from session before calling self._stopEncryption
+# so we need to remove plugin from session before calling self._stop_encryption
 del self._sessions[entity.userhostJID()]
-await self._stopEncryption(plugin, entity)
+await self._stop_encryption(plugin, entity)
 log.info(_("encryption session stopped with entity {entity}").format(
 entity=entity.full()))
-self.host.bridge.messageEncryptionStopped(
+self.host.bridge.message_encryption_stopped(
 entity.full(),
 {'name': plugin.name,
 'namespace': plugin.namespace,
 },
 self.client.profile)
 session = self.getSession(entity)
 if session is None:
 return None
 return session["plugin"].namespace
-def getTrustUI(self, entity_jid, namespace=None):
+def get_trust_ui(self, entity_jid, namespace=None):
 """Retrieve encryption UI
 @param entity_jid(jid.JID): get the UI for this entity
 must be a bare jid
 @param namespace(unicode): namespace of the algorithm to manage
 raise exceptions.NotFound(
 "No encryption session currently active for {entity_jid}"
 .format(entity_jid=entity_jid.full()))
 plugin = session['plugin']
 else:
-plugin = self.getPlugin(namespace)
+plugin = self.get_plugin(namespace)
 try:
-get_trust_ui = plugin.instance.getTrustUI
+get_trust_ui = plugin.instance.get_trust_ui
 except AttributeError:
 raise NotImplementedError(
 "Encryption plugin doesn't handle trust management UI")
 else:
-return utils.asDeferred(get_trust_ui, self.client, entity_jid)
+return utils.as_deferred(get_trust_ui, self.client, entity_jid)
 ## Menus ##
 @classmethod
-def _importMenus(cls, host):
+def _import_menus(cls, host):
-host.importMenu(
+host.import_menu(
 (D_("Encryption"), D_("unencrypted (plain text)")),
-partial(cls._onMenuUnencrypted, host=host),
+partial(cls._on_menu_unencrypted, host=host),
 security_limit=0,
 help_string=D_("End encrypted session"),
 type_=C.MENU_SINGLE,
 )
 for plg in cls.getPlugins():
-host.importMenu(
+host.import_menu(
 (D_("Encryption"), plg.name),
-partial(cls._onMenuName, host=host, plg=plg),
+partial(cls._on_menu_name, host=host, plg=plg),
 security_limit=0,
 help_string=D_("Start {name} session").format(name=plg.name),
 type_=C.MENU_SINGLE,
 )
-host.importMenu(
+host.import_menu(
 (D_("Encryption"), D_("⛨ {name} trust").format(name=plg.name)),
-partial(cls._onMenuTrust, host=host, plg=plg),
+partial(cls._on_menu_trust, host=host, plg=plg),
 security_limit=0,
 help_string=D_("Manage {name} trust").format(name=plg.name),
 type_=C.MENU_SINGLE,
 )
 @classmethod
-def _onMenuUnencrypted(cls, data, host, profile):
+def _on_menu_unencrypted(cls, data, host, profile):
-client = host.getClient(profile)
+client = host.get_client(profile)
 peer_jid = jid.JID(data['jid']).userhostJID()
 d = defer.ensureDeferred(client.encryption.stop(peer_jid))
 d.addCallback(lambda __: {})
 return d
 @classmethod
-def _onMenuName(cls, data, host, plg, profile):
+def _on_menu_name(cls, data, host, plg, profile):
-client = host.getClient(profile)
+client = host.get_client(profile)
 peer_jid = jid.JID(data['jid'])
 if not plg.directed:
 peer_jid = peer_jid.userhostJID()
 d = defer.ensureDeferred(
 client.encryption.start(peer_jid, plg.namespace, replace=True))
 d.addCallback(lambda __: {})
 return d
 @classmethod
 @defer.inlineCallbacks
-def _onMenuTrust(cls, data, host, plg, profile):
+def _on_menu_trust(cls, data, host, plg, profile):
-client = host.getClient(profile)
+client = host.get_client(profile)
 peer_jid = jid.JID(data['jid']).userhostJID()
-ui = yield client.encryption.getTrustUI(peer_jid, plg.namespace)
+ui = yield client.encryption.get_trust_ui(peer_jid, plg.namespace)
 defer.returnValue({'xmlui': ui.toXml()})
 ## Triggers ##
-def setEncryptionFlag(self, mess_data):
+def set_encryption_flag(self, mess_data):
 """Set "encryption" key in mess_data if session with destinee is encrypted"""
 to_jid = mess_data['to']
 encryption = self._sessions.get(to_jid.userhostJID())
 if encryption is not None:
 plugin = encryption['plugin']
 if mess_data["type"] == "groupchat" and plugin.directed:
 raise exceptions.InternalError(
 f"encryption flag must not be set for groupchat if encryption algorithm "
 f"({encryption['plugin'].name}) is directed!")
 mess_data[C.MESS_KEY_ENCRYPTION] = encryption
-self.markAsEncrypted(mess_data, plugin.namespace)
+self.mark_as_encrypted(mess_data, plugin.namespace)
 ## Misc ##
-def markAsEncrypted(self, mess_data, namespace):
+def mark_as_encrypted(self, mess_data, namespace):
 """Helper method to mark a message as having been e2e encrypted.
 This should be used in the post_treat workflow of messageReceived trigger of
 the plugin
 @param mess_data(dict): message data as used in post treat workflow
 )
 defer.ensureDeferred(self.start(from_bare_jid, namespace))
 return mess_data
-def isEncryptionRequested(
+def is_encryption_requested(
 self,
 mess_data: MessageData,
 namespace: Optional[str] = None
 ) -> bool:
 """Helper method to check if encryption is requested in an outgoind message
 @return (bool): True if the encrypted flag is present
 """
 return mess_data['extra'].get(C.MESS_KEY_ENCRYPTED, False)
-def markAsTrusted(self, mess_data):
+def mark_as_trusted(self, mess_data):
 """Helper methor to mark a message as sent from a trusted entity.
 This should be used in the post_treat workflow of messageReceived trigger of
 the plugin
 @param mess_data(dict): message data as used in post treat workflow
 """
 mess_data[C.MESS_KEY_TRUSTED] = True
 return mess_data
-def markAsUntrusted(self, mess_data):
+def mark_as_untrusted(self, mess_data):
 """Helper methor to mark a message as sent from an untrusted entity.
 This should be used in the post_treat workflow of messageReceived trigger of
 the plugin
 @param mess_data(dict): message data as used in post treat workflow

Mercurial > libervia-backend

comparison sat/memory/encryption.py @ 4037:524856bd7b19