#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# vim:fenc=utf-8 et ts=4 sts=4 sw=4
#
# Copyright © 2019 Maxime “pep” Buquet
#
# Distributed under terms of the zlib license. See COPYING file.
"""
Interface for E2EE (End-to-end Encryption) plugins.
"""
from typing import (
Callable,
Dict,
List,
Optional,
Union,
Tuple,
Set,
Type,
)
from slixmpp import InvalidJID, JID, Message
from slixmpp.xmlstream import StanzaBase
from slixmpp.xmlstream.handler import CoroutineCallback
from slixmpp.xmlstream.matcher import MatchXPath
from poezio.tabs import (
ChatTab,
ConversationTab,
DynamicConversationTab,
MucTab,
PrivateTab,
StaticConversationTab,
)
from poezio.plugin import BasePlugin
from poezio.theming import Theme, get_theme, dump_tuple
from poezio.config import config
from poezio.decorators import command_args_parser
import asyncio
from asyncio import iscoroutinefunction
import logging
log = logging.getLogger(__name__)
ChatTabs = Union[
MucTab,
DynamicConversationTab,
StaticConversationTab,
PrivateTab,
]
EME_NS = 'urn:xmpp:eme:0'
EME_TAG = 'encryption'
JCLIENT_NS = 'jabber:client'
HINTS_NS = 'urn:xmpp:hints'
class NothingToEncrypt(Exception):
"""
Exception to raise inside the _encrypt filter on stanzas that do not need
to be processed.
"""
class E2EEPlugin(BasePlugin):
"""Interface for E2EE plugins.
This is a wrapper built on top of BasePlugin. It provides a base for
End-to-end Encryption mechanisms in poezio.
Plugin developers are excepted to implement the `decrypt` and
`encrypt` function, provide an encryption name (and/or short name),
and an eme namespace.
Once loaded, the plugin will attempt to decrypt any message that
contains an EME message that matches the one set.
The plugin will also register a command (using the short name) to
enable encryption per tab. It is only possible to have one encryption
mechanism per tab, even if multiple e2ee plugins are loaded.
The encryption status will be displayed in the status bar, using the
plugin short name, alongside the JID, nickname etc.
"""
#: Specifies that the encryption mechanism does more than encrypting
#: ``.
stanza_encryption = False
#: Whitelist applied to messages when `stanza_encryption` is `False`.
tag_whitelist = [
(JCLIENT_NS, 'body'),
(EME_NS, EME_TAG),
(HINTS_NS, 'store'),
(HINTS_NS, 'no-copy'),
(HINTS_NS, 'no-store'),
(HINTS_NS, 'no-permanent-store'),
# TODO: Add other encryption mechanisms tags here
]
#: Replaces body with `eme `_
#: if set. Should be suitable for most plugins except those using
#: `` directly as their encryption container, like OTR, or the
#: example base64 plugin in poezio.
replace_body_with_eme = True
#: Encryption name, used in command descriptions, and logs. At least one
#: of `encryption_name` and `encryption_short_name` must be set.
encryption_name: Optional[str] = None
#: Encryption short name, used as command name, and also to display
#: encryption status in a tab. At least one of `encryption_name` and
#: `encryption_short_name` must be set.
encryption_short_name: Optional[str] = None
#: Required. https://xmpp.org/extensions/xep-0380.html.
eme_ns: Optional[str] = None
#: Used to figure out what messages to attempt decryption for. Also used
#: in combination with `tag_whitelist` to avoid removing encrypted tags
#: before sending. If multiple tags are present, a handler will be
#: registered for each invididual tag/ns pair under , as opposed
#: to a single handler for all tags combined.
encrypted_tags: Optional[List[Tuple[str, str]]] = None
# Static map, to be able to limit to one encryption mechanism per tab at a
# time
_enabled_tabs: Dict[JID, Callable] = {}
# Tabs that support this encryption mechanism
supported_tab_types: Tuple[Type[ChatTab], ...] = tuple()
# States for each remote entity
trust_states: Dict[str, Set[str]] = {'accepted': set(), 'rejected': set()}
def init(self):
self._all_trust_states = self.trust_states['accepted'].union(
self.trust_states['rejected']
)
if self.encryption_name is None and self.encryption_short_name is None:
raise NotImplementedError
if self.eme_ns is None:
raise NotImplementedError
if self.encryption_name is None:
self.encryption_name = self.encryption_short_name
if self.encryption_short_name is None:
self.encryption_short_name = self.encryption_name
if not self.supported_tab_types:
raise NotImplementedError
# Ensure decryption is done before everything, so that other handlers
# don't have to know about the encryption mechanism.
self.api.add_event_handler('muc_msg', self._decrypt_wrapper, priority=0)
self.api.add_event_handler('conversation_msg', self._decrypt_wrapper, priority=0)
self.api.add_event_handler('private_msg', self._decrypt_wrapper, priority=0)
# Register a handler for each invididual tag/ns pair in encrypted_tags
# as well. as _msg handlers only include messages with a