Initial commit. Import from poezio!18

Signed-off-by: Maxime Buquet's avatarMaxime “pep” Buquet <pep@bouah.net>
parents
Contributing to the Poezio-omemo project
========================================
To contribute, the preferred way is to commit your changes on some
publicly-available git repository (on a fork `on gitlab
<https://lab.louiz.org/poezio/poezio-omemo>`_ or on your own
repository) and to notify the developers with either:
- a ticket `on the bug tracker <https://lab.louiz.org/poezio/poezio-omemo/issues/new>`_
- a merge request `on gitlab <https://lab.louiz.org/poezio/poezio-omemo/merge_requests>`_
- a message on `the channel <xmpp:poezio@muc.poez.io?join>`_
This diff is collapsed.
Poezio OMEMO plugin
###################
This is a `Poezio <https://poez.io>`_ plugin providing OMEMO support. It
distributed separately for licensing reasons.
Use in poezio
-------------
Once installed (see the `Installation`_ section below), `/load omemo` in
Poezio.
See the Poezio `documentation
<https://doc.poez.io/plugins/index.html#plugin-autoload>`_ for more
information.
License
-------
This plugin is licensed under GPLv3.
Note on the underlying OMEMO library
------------------------------------
As stated in `python-xeddsa's
README <https://github.com/Syndace/python-xeddsa/blob/136b9f12c8286b9463566308963e70f090b60e50/README.md>`_,
(dependency of python-omemo), this library has not undergone any
security audits. If you have the knowledge, any help is welcome.
Please take this into consideration when using this library.
Installation
------------
- ArchLinux (AUR):
`python-poezio-omemo <https://aur.archlinux.org/packages/python-poezio-omemo>`_, or
`python-poezio-omemo-git <https://aur.archlinux.org/packages/python-poezio-omemo-git>`_
- PIP: `poezio-omemo`
- Manual: `python3 setup.py install`
#! /usr/bin/env python3
# -*- coding: utf-8 -*-
# vim:fenc=utf-8
#
# Copyright © 2019 Maxime “pep” Buquet <pep@bouah.net>
#
# Distributed under terms of the GPLv3 license.
"""
OMEMO Plugin.
"""
import os
import asyncio
import logging
from typing import Dict, List, Optional
from poezio.plugin_e2ee import E2EEPlugin
from poezio.xdg import DATA_HOME
from poezio.tabs import DynamicConversationTab, StaticConversationTab, MucTab
from omemo.exceptions import MissingBundleException
from slixmpp import JID
from slixmpp.stanza import Message
from slixmpp.exceptions import IqError, IqTimeout
from slixmpp_omemo import PluginCouldNotLoad, MissingOwnKey, NoAvailableSession
from slixmpp_omemo import UndecidedException, UntrustedException, EncryptionPrepareException
import slixmpp_omemo
log = logging.getLogger(__name__)
class Plugin(E2EEPlugin):
"""OMEMO (XEP-0384) Plugin"""
encryption_name = 'omemo'
eme_ns = slixmpp_omemo.OMEMO_BASE_NS
replace_body_with_eme = True
stanza_encryption = False
encrypted_tags = [
(slixmpp_omemo.OMEMO_BASE_NS, 'encrypted'),
]
# TODO: Look into blind trust stuff.
# https://gist.github.com/mar-v-in/b683220a55bc65dcdafc809be9c5d0e4
trust_states = {
'accepted': {
'verified',
'accepted',
}, 'rejected': {
'undecided',
'distrusted',
},
}
supported_tab_types = (DynamicConversationTab, StaticConversationTab, MucTab)
def init(self) -> None:
super().init()
self.info = lambda i: self.api.information(i, 'Info')
data_dir = os.path.join(DATA_HOME, 'omemo', self.core.xmpp.boundjid.bare)
os.makedirs(data_dir, exist_ok=True)
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.')
asyncio.ensure_future(
self.core.xmpp['xep_0384'].session_start(self.core.xmpp.boundjid)
)
def display_error(self, txt) -> None:
self.api.information(txt, 'Error')
def get_fingerprints(self, jid: JID) -> List[str]:
devices = self.core.xmpp['xep_0384'].get_trust_for_jid(jid)
# XXX: What to do with did -> None entries?
# XXX: What to do with the active/inactive devices differenciation?
# For now I'll merge both. We should probably display them separately
# later on.
devices['active'].update(devices['inactive'])
return [
slixmpp_omemo.fp_from_ik(trust['key'])
for trust in devices['active'].values()
if trust is not None
]
def decrypt(self, message: Message, tab, allow_untrusted=False) -> None:
body = None
try:
mfrom = message['from']
encrypted = message['omemo_encrypted']
body = self.core.xmpp['xep_0384'].decrypt_message(encrypted, mfrom, allow_untrusted)
body = body.decode('utf-8')
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['to']
body = message['body']
expect_problems = {} # type: Optional[Dict[JID, List[int]]]
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, expect_problems)
message.append(encrypt)
return None
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
except EncryptionPrepareException as exn:
log.debug('FOO: EncryptionPrepareException: %r', exn.errors)
for error in exn.errors:
if isinstance(error, MissingBundleException):
self.display_error(
'Could not find keys for device "%d" of recipient "%s". Skipping.' %
(error.device, error.bare_jid),
)
jid = JID(error.bare_jid)
device_list = expect_problems.setdefault(jid, [])
device_list.append(error.device)
except (IqError, IqTimeout) as exn:
self.display_error(
'An error occured while fetching information on a recipient.\n%r' % exn,
)
return None
return None
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# vim:fenc=utf-8
#
# Copyright © 2019 Maxime “pep” Buquet <pep@bouah.net>
#
# Distributed under terms of the GPLv3+ license.
"""
Slixmpp OMEMO plugin
Copyright © 2019 Maxime “pep” Buquet <pep@bouah.net>
This file is part of poezio-omemo.
See the file LICENSE for copying permission.
"""
__version__ = "0.1.0"
__version_info__ = (0, 1, 0)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Poezio OMEMO plugin
Copyright (C) 2019 Maxime “pep” Buquet <pep@bouah.net>
This file is part of poezio-omemo.
See the file LICENSE for copying permission.
"""
import os
from typing import Any, Dict
from setuptools import setup
MODULE_FILE_PATH = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
'poezio_plugins/omemo', 'version.py'
)
def get_version() -> str:
"""Returns version by looking at poezio_version/version.py"""
version: Dict[str, Any] = {}
with open(MODULE_FILE_PATH) as file:
exec(file.read(), version)
if '__version__' in version:
return version['__version__']
return 'missingno'
DESCRIPTION = ('Poezio OMEMO plugin')
VERSION = get_version()
with open('README.rst', encoding='utf8') as readme:
LONG_DESCRIPTION = readme.read()
CLASSIFIERS = [
'Development Status :: 3 - Alpha',
'Intended Audience :: Developers',
'License :: OSI Approved :: GNU General Public License v3 (GPLv3)',
'Programming Language :: Python',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Topic :: Internet :: XMPP',
'Topic :: Security :: Cryptography',
'Topic :: Software Development :: Libraries :: Python Modules',
]
setup(
name="poezio-omemo",
version=VERSION,
description=DESCRIPTION,
long_description=LONG_DESCRIPTION,
long_description_content_type='text/x-rst',
author='Maxime Buquet',
author_email='pep@bouah.net',
url='https://lab.louiz.org/poezio/poezio-omemo',
license='GPLv3',
platforms=['any'],
packages=['poezio_plugins.omemo'],
install_requires=['poezio', 'slixmpp-omemo'],
classifiers=CLASSIFIERS,
)
[tox]
envlist = py37
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