Commit a8762b47 authored by mathieui's avatar mathieui

Rewrite the OTR plugin to use pure-python-otr

parent feb101f1
...@@ -5,4 +5,5 @@ class MucTab(ChatTab): pass ...@@ -5,4 +5,5 @@ class MucTab(ChatTab): pass
class ConversationTab(ChatTab): pass class ConversationTab(ChatTab): pass
class RosterInfoTab(Tab): pass class RosterInfoTab(Tab): pass
class XMLTab(Tab): pass class XMLTab(Tab): pass
class DynamicConversationTab(Tab): pass
""" """
This plugin implements `Off The Record messaging`_. This plugin implements `Off The Record messaging`_.
This is a plugin used to encrypt one-to-one conversation using the OTR This is a plugin used to encrypt one-to-one conversation using the OTR
...@@ -22,53 +19,47 @@ Note that if you are having an encrypted conversation with a contact, you can ...@@ -22,53 +19,47 @@ Note that if you are having an encrypted conversation with a contact, you can
**not** send XHTML-IM messages to him. They will be removed and be replaced by **not** send XHTML-IM messages to him. They will be removed and be replaced by
plain text messages. plain text messages.
Installation and configuration Installation
------------------------------ ------------
To use the OTR plugin, you must first install libopenotr.
If you use Archlinux, there is a `libopenotr-git`_ package on the AUR. To use the OTR plugin, you must first install pure-python-otr.
If not, then you will have to install it by hand. You have to install it from the git because a few issues were
found with the python3 compatibility while writing this plugin,
and the fixes did not make it into a stable release yet.
First, clone the repo and go inside the created directory: Install the python module:
.. code-block:: bash .. code-block:: bash
git clone git clone
cd libopenotr cd pure-python-otr
python3 install --user
Then run and configure You can also use pip with the requirements.txt at the root of
the poezio directory.
.. code-block:: bash
sh Usage
./configure --enable-gaping-security-hole -----
(as of now, you *should* have been warned enough that the library is not finished) Command added to Conversation Tabs and Private Tabs:
Then compile & install the lib: .. glossary::
.. code-block:: bash /otr
**Usage:** ``/otr [start|refresh|end|fpr|ourfpr]``
make This command is used to start (or refresh), or end an OTR private session.
sudo make install
Finally, install the python module: The ``fpr`` command gives you the fingerprint of the key of the remove entity, and
the ``ourfpr`` command gives you the fingerprint of your own key.
.. code-block:: bash
python3 build
sudo python3 install
To use OTR, make sure the plugin is loaded (if not, then do ``/load otr``). To use OTR, make sure the plugin is loaded (if not, then do ``/load otr``).
Once you are in a private conversation, you have to do a: Once you are in a private conversation, you have to do a:
.. code-block:: none .. code-block:: none
/otr start /otr start
...@@ -76,160 +67,281 @@ Once you are in a private conversation, you have to do a: ...@@ -76,160 +67,281 @@ Once you are in a private conversation, you have to do a:
The status of the OTR encryption should appear in the bar between the chat and The status of the OTR encryption should appear in the bar between the chat and
the input as ``OTR: encrypted``. the input as ``OTR: encrypted``.
Once you’re done, end the OTR session with Once you’re done, end the OTR session with
.. code-block:: none .. code-block:: none
/otr end /otr end
Known problems
Empty messages send when changing status. Important details
.. _Off The Record messaging: The OTR session is considered for a full jid.
.. _libopenotr-git:
import pyotr .. _Off The Record messaging:
from sleekxmpp.xmlstream.stanzabase import JID
import potr
import theming
import logging import logging
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
import os
from potr.context import NotEncryptedError, UnencryptedMessage, ErrorReceived, NotOTRMessage, STATE_ENCRYPTED, STATE_PLAINTEXT, STATE_FINISHED, Context, Account
from plugin import BasePlugin from plugin import BasePlugin
from tabs import ConversationTab, DynamicConversationTab, PrivateTab
from common import safeJID
OTR_DIR = os.path.join(os.getenv('XDG_DATA_HOME') or
'~/.local/share', 'poezio', 'otr')
import tabs POLICY_FLAGS = {
from tabs import ConversationTab 'ALLOW_V1':False,
'SEND_TAG': True,
log = logging.getLogger(__name__)
class PoezioContext(Context):
def __init__(self, account, peer, xmpp, core):
super(PoezioContext, self).__init__(account, peer)
self.xmpp = xmpp
self.core = core
self.flags = {}
def getPolicy(self, key):
if key in self.flags:
return self.flags[key]
return False
def inject(self, msg, appdata=None):
self.xmpp.send_message(mto=self.peer, mbody=msg.decode('ascii'), mtype='chat')
def setState(self, newstate):
tab = self.core.get_tab_by_name(self.peer)
if not tab:
tab = self.core.get_tab_by_name(safeJID(self.peer).bare, DynamicConversationTab)
if not tab.locked_resource == safeJID(self.peer).resource:
tab = None
if self.state == STATE_ENCRYPTED:
if newstate == STATE_ENCRYPTED:
log.debug('OTR conversation with %s refreshed', self.peer)
if tab:
tab.add_message('Refreshed OTR conversation with %s' % self.peer)
elif newstate == STATE_FINISHED or newstate == STATE_PLAINTEXT:
log.debug('OTR conversation with %s finished', self.peer)
if tab:
tab.add_message('Ended OTR conversation with %s' % self.peer)
if newstate == STATE_ENCRYPTED:
if tab:
tab.add_message('Started OTR conversation with %s' % self.peer)
log.debug('Set encryption state of %s to %s', self.peer, states[newstate])
super(PoezioContext, self).setState(newstate)
if tab:
class PoezioAccount(Account):
def __init__(self, jid, key_dir):
super(PoezioAccount, self).__init__(jid, 'xmpp', 1024)
self.key_dir = os.path.join(key_dir, jid)
def load_privkey(self):
with open(self.key_dir + '.key3', 'rb') as keyfile:
return potr.crypt.PK.parsePrivateKey([0]
log.error('Error in load_privkey', exc_info=True)
def save_privkey(self):
with open(self.key_dir + '.key3', 'xb') as keyfile:
log.error('Error in save_privkey', exc_info=True)
def save_trusts(self):
saveTrusts = save_trusts
loadPrivkey = load_privkey
savePrivkey = save_privkey
states = {
STATE_PLAINTEXT: 'plaintext',
STATE_ENCRYPTED: 'encrypted',
STATE_FINISHED: 'finished',
class Plugin(BasePlugin): class Plugin(BasePlugin):
def init(self): def init(self):
self.contacts = {} # set the default values from the config
# a dict of {full-JID: OTR object} policy = self.config.get('encryption_policy', 'ondemand').lower()
self.api.add_event_handler('conversation_say_after', self.on_conversation_say) POLICY_FLAGS['REQUIRE_ENCRYPTION'] = (policy == 'always')
self.api.add_event_handler('conversation_msg', self.on_conversation_msg) allow_v2 = self.config.get('allow_v2', 'true').lower()
POLICY_FLAGS['ALLOW_V2'] = (allow_v2 != 'false')
allow_v1 = self.config.get('allow_v1', 'false').lower()
POLICY_FLAGS['ALLOW_v1'] = (allow_v1 == 'true')
global OTR_DIR
OTR_DIR = os.path.expanduser(self.config.get('keys_dir', '') or OTR_DIR)
except FileExistsError:
self.api.information('The OTR-specific folder could not be created'
' poezio will be unable to save keys and trusts', 'OTR')
self.api.add_tab_command(ConversationTab, 'otr', self.command_otr, self.api.add_event_handler('conversation_msg', self.on_conversation_msg)
usage='<start|end|fpr>', self.api.add_event_handler('private_msg', self.on_conversation_msg)
help='Start or stop OTR for the current conversation.', self.api.add_event_handler('conversation_say_after', self.on_conversation_say)
short='Manage OTR status', self.api.add_event_handler('private_say_after', self.on_conversation_say)
ConversationTab.add_information_element('otr', self.display_encryption_status) ConversationTab.add_information_element('otr', self.display_encryption_status)
PrivateTab.add_information_element('otr', self.display_encryption_status)
self.account = PoezioAccount(self.core.xmpp.boundjid.bare, OTR_DIR)
self.contexts = {}
self.api.add_tab_command(ConversationTab, 'otr', self.command_otr,
self.api.add_tab_command(PrivateTab, 'otr', self.command_otr,
def cleanup(self): def cleanup(self):
ConversationTab.remove_information_element('otr') ConversationTab.remove_information_element('otr')
self.del_tab_command(ConversationTab, 'otr') PrivateTab.remove_information_element('otr')
def otr_special(self, tab, typ): def get_context(self, jid):
def helper(msg): jid = safeJID(jid).full
tab.add_message('%s: %s' % (typ, msg.decode())) if not jid in self.contexts:
return helper flags = POLICY_FLAGS.copy()
policy = self.config.get_by_tabname('encryption_policy', 'ondemand', jid).lower()
def otr_on_state_change(self, tab): flags['REQUIRE_ENCRYPTION'] = (policy == 'always')
def helper(old, new): allow_v2 = self.config.get_by_tabname('allow_v2', 'true', jid).lower()
old = self.otr_state(old) flags['ALLOW_V2'] = (allow_v2 != 'false')
new = self.otr_state(new) allow_v1 = self.config.get_by_tabname('allow_v1', 'false', jid).lower()
tab.add_message('OTR state has changed from %s to %s' % (old, new)) flags['ALLOW_V1'] = (allow_v1 == 'true')
return helper self.contexts[jid] = PoezioContext(self.account, jid, self.core.xmpp, self.core)
self.contexts[jid].flags = flags
def get_otr(self, tab): return self.contexts[jid]
if tab not in self.contacts:
self.contacts[tab] = pyotr.OTR(on_error=self.otr_special(tab, 'Error'), on_warn=self.otr_special(tab, 'Warn'), on_state_change=self.otr_on_state_change(tab)) def on_conversation_msg(self, msg, tab):
return self.contacts[tab] try:
ctx = self.get_context(msg['from'])
def on_conversation_say(self, message, tab): txt, tlvs = ctx.receiveMessage(msg["body"].encode('utf-8'))
""" except UnencryptedMessage as err:
Feed the message through the OTR filter # received an unencrypted message inside an OTR session
""" tab.add_message('The following message from %s was not encrypted:\n%s' % (msg['from'], err.args[0].decode('utf-8')),
to = message['to'] jid=msg['from'], nick_color=theming.get_theme().COLOR_REMOTE_USER,
if not message['body']: typ=0)
# there’s nothing to encrypt if this is a chatstate, for example del msg['body']
except ErrorReceived as err:
# Received an OTR error
tab.add_message('Received the following error from %s: %s' % (msg['from'], err.args[0]))
del msg['body']
except NotOTRMessage as err:
# ignore non-otr messages
# if we expected an OTR message, we would have
# got an UnencryptedMesssage
except NotEncryptedError as err:
tab.add_message('An encrypted message from %s was received but is unreadable, as you are not'
' currently communicating privately.' % msg['from'],
jid=msg['from'], nick_color=theming.get_theme().COLOR_REMOTE_USER,
del msg['body']
return return
otr_state = self.get_otr(tab) except:
# Not sure what to do with xhtml bodies, and I don't like them anyway ;) return
del message['xhtml_im']
say = otr_state.transform_msg(message['body'].encode()) # remove xhtml
if say is not None: del msg['html']
message['body'] = say.decode() del msg['body']
if not txt:
if isinstance(tab, PrivateTab):
user = tab.parent_muc.get_user_by_name(msg['from'].resource)
else: else:
del message['body'] user = None
body = txt.decode()
tab.add_message(body, nickname=tab.nick, jid=msg['from'],
forced_user=user, typ=0)
del msg['body']
def on_conversation_msg(self, message, tab): def on_conversation_say(self, msg, tab):
""" """
Feed the message through the OTR filter On message sent
""" """
fro = message['from'] if isinstance(tab, DynamicConversationTab) and tab.locked_resource:
if not message['body']: name = safeJID(
# there’s nothing to decrypt if this is a chatstate, for example name.resource = tab.locked_resource
return name = name.full
otr_state = self.get_otr(tab)
# Not sure what to do with xhtml bodies, and I don't like them anyway ;)
del message['xhtml_im']
display, reply = otr_state.handle_msg(message['body'].encode())
#self.core.information('D: {!r}, R: {!r}'.format(display, reply))
if display is not None:
message['body'] = display.decode()
else: else:
del message['body'] name =
if reply is not None: ctx = self.contexts.get(name)
self.otr_say(tab, reply.decode()) if ctx and ctx.state == STATE_ENCRYPTED:
ctx.sendMessage(0, msg['body'].encode('utf-8'))
@staticmethod # remove everything from the message so that it doesn’t get sent
def otr_state(state): del msg['body']
if state == pyotr.MSG_STATE_PLAINTEXT: del msg['replace']
return 'plaintext' del msg['html']
elif state == pyotr.MSG_STATE_ENCRYPTED:
return 'encrypted'
elif state == pyotr.MSG_STATE_FINISHED:
return 'finished'
def display_encryption_status(self, jid): def display_encryption_status(self, jid):
context = self.get_context(jid)
state = states[context.state]
return ' OTR: %s' % state
def command_otr(self, arg):
""" """
Returns the status of encryption for the associated jid. This is to be used /otr [start|refresh|end|fpr|ourfpr]
in the ConversationTab’s InfoWin.
tab = self.core.get_tab_by_name(jid, tabs.ConversationTab)
if tab not in self.contacts:
return ''
state = self.otr_state(self.contacts[tab].state)
return ' OTR: %s' % (state,)
def otr_say(self, tab, line):
msg = self.core.xmpp.make_message(tab.get_name())
msg['type'] = 'chat'
msg['body'] = line
def command_otr(self, args):
A command to start or end OTR encryption
""" """
args = args.split() arg = arg.strip()
if not args: tab = self.api.current_tab()
return self.api.run_command("/help otr") name =
if isinstance(self.api.current_tab(), ConversationTab): if isinstance(tab, DynamicConversationTab) and tab.locked_resource:
jid = JID(self.api.current_tab().get_name()) name = safeJID(
command = args[0] name.resource = tab.locked_resource
if command == 'start': name = name.full
otr_state = self.get_otr(self.api.current_tab()) if arg == 'end': # close the session
self.otr_say(self.api.current_tab(), otr_state.start().decode()) context = self.get_context(name)
elif command == 'end': context.disconnect()
otr_state = self.get_otr(self.api.current_tab()) elif arg == 'start' or arg == 'refresh':
msg = otr_state.end() otr = self.get_context(name)
if msg is not None: self.core.xmpp.send_message(mto=name, mtype='chat',
self.otr_say(self.api.current_tab(), msg.decode()) mbody=self.contexts[name].sendMessage(0, b'?OTRv?').decode())
elif command == 'fpr': elif arg == 'ourfpr':
otr_state = self.get_otr(self.api.current_tab()) fpr = self.account.getPrivkey()
our = otr_state.our_fpr self.api.information('Your OTR key fingerprint is %s' % fpr, 'OTR')
if our: elif arg == 'fpr':
our = hex(int.from_bytes(our, 'big'))[2:].ljust(40).upper() tab = self.api.current_tab()
their = otr_state.their_fpr if tab.get_name() in self.contexts:
if their: ctx = self.contexts[tab.get_name()]
their = hex(int.from_bytes(their, 'big'))[2:].ljust(40).upper() self.api.information('The key fingerprint for %s is %s' % (name, ctx.getCurrentKey()) , 'OTR')
self.api.current_tab().add_message('Your: %s Their: %s' % (our, their))
self.core.refresh_window() def completion_otr(self, the_input):
return the_input.new_completion(['start', 'fpr', 'ourfpr', 'refresh', 'end'], 1, quotify=True)
def otr_completion(self, the_input):
return the_input.auto_completion(['start', 'fpr', 'end'], '', quotify=False)
-e git:// -e git://
-e git://
dnspython3==1.10.0 dnspython3==1.10.0
sphinx==1.2b1 sphinx==1.2b1
argparse argparse
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment