summaryrefslogtreecommitdiff
path: root/plugins/omemo_plugin.py
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/omemo_plugin.py')
-rw-r--r--plugins/omemo_plugin.py366
1 files changed, 164 insertions, 202 deletions
diff --git a/plugins/omemo_plugin.py b/plugins/omemo_plugin.py
index daca9e8d..b396422e 100644
--- a/plugins/omemo_plugin.py
+++ b/plugins/omemo_plugin.py
@@ -10,222 +10,184 @@
"""
import os
-import asyncio
import logging
-from typing import Callable, List, Optional, Set, Tuple, Union
-from poezio.plugin import BasePlugin
-from poezio.tabs import DynamicConversationTab, StaticConversationTab, ConversationTab, MucTab
+from poezio.plugin_e2ee import E2EEPlugin
from poezio.xdg import DATA_HOME
-from slixmpp import JID
-from slixmpp.plugins.xep_0384.plugin import MissingOwnKey
+from slixmpp.stanza import Message
+from slixmpp_omemo import PluginCouldNotLoad, MissingOwnKey, NoAvailableSession
+from slixmpp_omemo import UndecidedException, UntrustedException, EncryptionPrepareException
+import slixmpp_omemo
log = logging.getLogger(__name__)
-class Plugin(BasePlugin):
+class Plugin(E2EEPlugin):
"""OMEMO (XEP-0384) Plugin"""
- _enabled_jids = set() # type: Set[JID]
- def init(self):
- self.info = lambda i: self.api.information(i, 'Info')
- self.xmpp = self.core.xmpp
-
- data_dir = os.path.join(DATA_HOME, 'omemo')
- os.makedirs(data_dir, exist_ok=True)
-
- self.xmpp.register_plugin(
- 'xep_0384', {
- 'data_dir': data_dir,
- })
-
- self.api.add_command(
- 'omemo',
- self.command_status,
- help='Display contextual information',
- )
-
- ConversationTab.add_information_element('omemo', self.display_encryption_status)
- MucTab.add_information_element('omemo', self.display_encryption_status)
-
- self.api.add_command(
- 'omemo_enable',
- self.command_enable,
- help='Enable OMEMO encryption',
- )
-
- self.api.add_command(
- 'omemo_disable',
- self.command_disable,
- help='Disable OMEMO encryption',
- )
-
- self.api.add_command(
- 'encrypted_message',
- self.send_message,
- help='Send OMEMO encrypted message',
- )
-
- self.api.add_event_handler(
- 'conversation_say_after',
- self.on_conversation_say_after,
- )
-
- self.api.add_event_handler(
- 'conversation_msg',
- self.on_conversation_msg,
- )
-
- def cleanup(self) -> None:
- ConversationTab.remove_information_element('omemo')
- MucTab.remove_information_element('omemo')
-
- def display_encryption_status(self, jid: JID) -> str:
- """
- Return information to display in the infobar if OMEMO is enabled
- for the JID.
- """
-
- if jid in self._enabled_jids:
- return " OMEMO"
- return ""
-
- def command_status(self, _args):
- """Display contextual information depending on currenttab."""
- tab = self.api.current_tab()
- self.info('OMEMO!')
- self.info("My device id: %d" % self.xmpp['xep_0384'].my_device_id())
-
- def _jid_from_context(self, jid: Optional[Union[str, JID]]) -> Tuple[Optional[JID], bool]:
- """
- Get bare JID from context if not specified
-
- Return a tuple with the JID and a bool specifying that the JID
- corresponds to the current tab.
- """
-
- tab = self.api.current_tab()
-
- tab_jid = None
- chat_tabs = (DynamicConversationTab, StaticConversationTab, ConversationTab, MucTab)
- if isinstance(tab, chat_tabs):
- tab_jid = JID(tab.name).bare
-
- # If current tab has a JID, use it if none is specified
- if not jid and tab_jid is not None:
- jid = tab_jid
-
- # We might not have found a JID at this stage. No JID provided and not
- # in a tab with a JID (InfoTab etc.).
- # If we do, make we
- if jid:
- # XXX: Ugly. We don't know if 'jid' is a str or a JID. And we want
- # to return a bareJID. We could change the JID API to allow us to
- # clear the resource one way or another.
- jid = JID(JID(jid).bare)
- else:
- jid = None
-
- return (jid, tab_jid is not None and tab_jid == jid)
-
- def command_enable(self, jid: Optional[str]) -> None:
- """
- Enable JID to use OMEMO with.
-
- Use current tab JID is none is specified. Refresh the tab if JID
- corresponds to the one being added.
- """
-
- jid, current_tab = self._jid_from_context(jid)
- if jid is None:
- return None
-
- if jid not in self._enabled_jids:
- self.info('OMEMO enabled for %s' % jid)
- self._enabled_jids.add(jid)
-
- # Refresh tab if JID matches
- if current_tab:
- self.api.current_tab().refresh()
-
- return None
+ encryption_name = 'omemo'
+ eme_ns = slixmpp_omemo.OMEMO_BASE_NS
+ replace_body_with_eme = True
+ stanza_encryption = False
- def command_disable(self, jid: Optional[str]) -> None:
- """
- Enable JID to use OMEMO with.
+ encrypted_tags = [
+ (slixmpp_omemo.OMEMO_BASE_NS, 'payload'),
+ ]
- Use current tab JID is none is specified. Refresh the tab if JID
- corresponds to the one being added.
- """
+ def init(self) -> None:
+ super().init()
- jid, current_tab = self._jid_from_context(jid)
- if jid is None:
- return None
+ self.info = lambda i: self.api.information(i, 'Info')
- if jid in self._enabled_jids:
- self.info('OMEMO disabled for %s' % jid)
- self._enabled_jids.remove(jid)
+ data_dir = os.path.join(DATA_HOME, 'omemo')
+ os.makedirs(data_dir, exist_ok=True)
- # Refresh tab if JID matches
- if current_tab:
- self.api.current_tab().refresh()
+ try:
+ self.core.xmpp.register_plugin(
+ 'xep_0384', {
+ 'data_dir': data_dir,
+ },
+ module=slixmpp_omemo,
+ ) # OMEMO
+ except (PluginCouldNotLoad,):
+ log.exception('And error occured when loading the omemo plugin.')
+
+# def send_message(self, _args):
+# asyncio.ensure_future(
+# self._send_encrypted_message(
+# "Hello Encrypted World!",
+# [JID('pep@bouah.net'), JID('test@bouah.net')],
+# mto=JID('test@bouah.net'),
+# mtype='chat',
+# ),
+# )
+#
+# async def _send_encrypted_message(
+# self,
+# payload: str,
+# recipients: List[JID],
+# mto: Optional[JID] = None,
+# mtype: Optional[str] = 'chat',
+# ) -> None:
+# try:
+# encrypted = await self.xmpp['xep_0384'].encrypt_message(payload, recipients)
+# except EncryptionPrepareException as e:
+# log.debug('Failed to encrypt message: %r', e)
+# return None
+# msg = self.core.xmpp.make_message(mto, mtype=mtype)
+# msg['body'] = 'This message is encrypted with Legacy OMEMO (eu.siacs.conversations.axolotl)'
+# msg['eme']['namespace'] = 'eu.siacs.conversations.axolotl'
+# msg.append(encrypted)
+# log.debug('BAR: message: %r', msg)
+# msg.send()
+# return None
+#
+# def on_conversation_say_after(
+# self,
+# message: Message,
+# tabs: Union[DynamicConversationTab, StaticConversationTab, ConversationTab, MucTab],
+# ) -> None:
+# """
+# Outbound messages
+# """
+#
+# # Check encryption status with the contact, if enabled, add
+# # ['omemo_encrypt'] attribute to message and send. Maybe delete
+# # ['body'] and tab.add_message ourselves to specify typ=0 so messages
+# # are not logged.
+#
+# fromjid = message['from']
+# if fromjid not in self._enabled_jids:
+# return None
+#
+# self.xmpp['xep_0384'].encrypt_message(message)
+# return None
+
+ def display_error(self, txt) -> None:
+ self.api.information(txt, 'Error')
+
+ def decrypt(self, message: Message, tab, allow_untrusted=False) -> None:
+
+ body = None
+ allow_untrusted = False
+ try:
+ body = self.core.xmpp['xep_0384'].decrypt_message(message, allow_untrusted)
+ except (MissingOwnKey,):
+ # The message is missing our own key, it was not encrypted for
+ # us, and we can't decrypt it.
+ self.display_error(
+ 'I can\'t decrypt this message as it is not encrypted for me.'
+ )
+ except (NoAvailableSession,) as exn:
+ # We received a message from that contained a session that we
+ # don't know about (deleted session storage, etc.). We can't
+ # decrypt the message, and it's going to be lost.
+ # Here, as we need to initiate a new encrypted session, it is
+ # best if we send an encrypted message directly. XXX: Is it
+ # where we talk about self-healing messages?
+ self.display_error(
+ 'I can\'t decrypt this message as it uses an encrypted '
+ 'session I don\'t know about.',
+ )
+ except (UndecidedException, UntrustedException) as exn:
+ # We received a message from an untrusted device. We can
+ # choose to decrypt the message nonetheless, with the
+ # `allow_untrusted` flag on the `decrypt_message` call, which
+ # we will do here. This is only possible for decryption,
+ # encryption will require us to decide if we trust the device
+ # or not. Clients _should_ indicate that the message was not
+ # trusted, or in undecided state, if they decide to decrypt it
+ # anyway.
+ self.display_error(
+ "Your device '%s' is not in my trusted devices." % exn.device,
+ )
+ # We resend, setting the `allow_untrusted` parameter to True.
+ self.decrypt(message, tab, allow_untrusted=True)
+ except (EncryptionPrepareException,):
+ # Slixmpp tried its best, but there were errors it couldn't
+ # resolve. At this point you should have seen other exceptions
+ # and given a chance to resolve them already.
+ self.display_error('I was not able to decrypt the message.')
+ except (Exception,) as exn:
+ self.display_error('An error occured while attempting decryption.\n%r' % exn)
+ raise
+
+ if body is not None:
+ message['body'] = body
+
+ async def encrypt(self, message: Message, _tab) -> None:
+ mto = message['from']
+ mtype = message['type']
+ body = message['body']
+
+ while True:
+ try:
+ # `encrypt_message` excepts the plaintext to be sent, a list of
+ # bare JIDs to encrypt to, and optionally a dict of problems to
+ # expect per bare JID.
+ #
+ # Note that this function returns an `<encrypted/>` object,
+ # and not a full Message stanza. This combined with the
+ # `recipients` parameter that requires for a list of JIDs,
+ # allows you to encrypt for 1:1 as well as groupchats (MUC).
+ #
+ # TODO: Document expect_problems
+ # TODO: Handle multiple recipients (MUCs)
+ recipients = [mto]
+ encrypt = await self.core.xmpp['xep_0384'].encrypt_message(body, recipients)
+ message.append(encrypt)
+ except UndecidedException as exn:
+ # The library prevents us from sending a message to an
+ # untrusted/undecided barejid, so we need to make a decision here.
+ # This is where you prompt your user to ask what to do. In
+ # this bot we will automatically trust undecided recipients.
+ self.core.xmpp['xep_0384'].trust(exn.bare_jid, exn.device, exn.ik)
+ # TODO: catch NoEligibleDevicesException and MissingBundleException
+ except Exception as exn:
+ await self.display_error(
+ 'An error occured while attempting to encrypt.\n%r' % exn,
+ )
+ raise
return None
-
- def send_message(self, _args):
- asyncio.ensure_future(
- self._send_message(
- "Hello Encrypted World!",
- [JID('some@jid')],
- mto=JID('some@jid'),
- mtype='chat',
- ),
- )
-
- async def _send_message(
- self,
- payload: str,
- recipients: List[JID],
- mto: Optional[JID] = None,
- mtype: Optional[str] = 'chat',
- ) -> None:
- encrypted = await self.xmpp['xep_0384'].encrypt_message(payload, recipients)
- msg = self.core.xmpp.make_message(mto, mtype=mtype)
- msg['body'] = 'This message is encrypted with Legacy OMEMO (eu.siacs.conversations.axolotl)'
- msg['eme']['namespace'] = 'eu.siacs.conversations.axolotl'
- msg.append(encrypted)
- log.debug('BAR: message: %r', msg)
- msg.send()
-
- def on_conversation_say_after(self, message, tab):
- """
- Outbound messages
- """
-
- # Check encryption status globally and to the contact, if enabled, add
- # ['omemo_encrypt'] attribute to message and send. Maybe delete
- # ['body'] and tab.add_message ourselves to specify typ=0 so messages
- # are not logged.
-
- fromjid = message['from']
- self.xmpp['xep_0384'].encrypt_message(message)
-
- def on_conversation_msg(self, message, _tab):
- """
- Inbound messages
- """
-
- # Check if encrypted, and if so replace message['body'] with
- # plaintext.
-
- self.info('Foo2')
- if self.xmpp['xep_0384'].is_encrypted(message):
- try:
- body = self.xmpp['xep_0384'].decrypt_message(message)
- except (MissingOwnKey,):
- log.debug("The following message is missing our key;"
- "Couldn't decrypt: %r", message)
- return None
- message['body'] = body.decode("utf8")
- return None