48427e3bdd
- "start" won't do anything if session is already encrypted with requested algorithm, and will raise a ConflictError if it's already encrypted but with an other algorithm - implemented "stop", with an optional namespace to check we are stopping the expected algorithm - "getSession" retrieve the current encryption session of a jid, if any
175 lines
7.4 KiB
Python
175 lines
7.4 KiB
Python
#!/usr/bin/env python2
|
|
# -*- coding: utf-8 -*-
|
|
|
|
# SAT: a jabber client
|
|
# Copyright (C) 2009-2018 Jérôme Poisson (goffi@goffi.org)
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
from sat.core.i18n import _
|
|
from sat.core.constants import Const as C
|
|
from sat.core import exceptions
|
|
from collections import namedtuple
|
|
from sat.core.log import getLogger
|
|
log = getLogger(__name__)
|
|
|
|
|
|
EncryptionPlugin = namedtuple("EncryptionPlugin", ("instance",
|
|
"name",
|
|
"namespace",
|
|
"priority"))
|
|
|
|
|
|
class EncryptionHandler(object):
|
|
"""Class to handle encryption sessions for a client"""
|
|
plugins = [] # plugin able to encrypt messages
|
|
|
|
def __init__(self, host):
|
|
self._sessions = {} # bare_jid ==> encryption_data
|
|
|
|
@classmethod
|
|
def registerPlugin(cls, plg_instance, name, namespace, priority=0):
|
|
"""Register a plugin handling an encryption algorithm
|
|
|
|
@param plg_instance(object): instance of the plugin
|
|
it must have the following methods:
|
|
- startEncryption(jid.JID): start an encryption session with a bare jid
|
|
- stopEncryption(jid.JID): stop an encryption session with a bare jid
|
|
@param name(unicode): human readable name of the encryption alrgorithm
|
|
@param namespace(unicode): namespace of the encryption algorithm
|
|
@param priority(int): priority of this plugin to encrypt an message when not
|
|
selected manually
|
|
"""
|
|
existing_ns = [p.namespace for p in cls.plugins]
|
|
if namespace in existing_ns:
|
|
raise exceptions.ConflictError("A plugin with this namespace already exists!")
|
|
plg = EncryptionPlugin(
|
|
instance=plg_instance,
|
|
name=name,
|
|
namespace=namespace,
|
|
priority=priority)
|
|
cls.plugins.append(plg)
|
|
cls.plugins.sort(key=lambda p: p.priority)
|
|
log.info(_(u"Encryption plugin registered: {name}").format(name=name))
|
|
|
|
def start(self, entity, namespace=None):
|
|
"""Start an encrypted session with an entity
|
|
|
|
@param entity(jid.JID): entity to start an encrypted session with
|
|
must be bare jid is the algorithm encrypt for all devices
|
|
@param namespace(unicode, None): namespace of the encryption algorithm to use
|
|
None to select automatically an algorithm
|
|
"""
|
|
if not self.plugins:
|
|
raise exceptions.NotFound(_(u"No encryption plugin is registered, "
|
|
u"an encryption session can't be started"))
|
|
|
|
if namespace is None:
|
|
plg = self.plugins[0]
|
|
else:
|
|
try:
|
|
plg = next(p for p in self.plugins if p.namespace == namespace)
|
|
except StopIteration:
|
|
raise exceptions.NotFound(_(
|
|
u"Can't find requested encryption plugin: {namespace}").format(
|
|
namespace=namespace))
|
|
|
|
bare_jid = entity.userhostJID()
|
|
if bare_jid in self._sessions:
|
|
plg = self._sessions[bare_jid]['plugin']
|
|
if plg.namespace == namespace:
|
|
log.info(_(u"Session with {bare_jid} is already encrypted with {name}."
|
|
u"Nothing to do.")
|
|
.format(bare_jid=bare_jid, name=plg.name))
|
|
return
|
|
|
|
msg = (_(u"Session with {bare_jid} is already encrypted with {name}. "
|
|
u"Please stop encryption session before changing algorithm.")
|
|
.format(bare_jid=bare_jid, name=plg.name))
|
|
log.warning(msg)
|
|
raise exceptions.ConflictError(msg)
|
|
|
|
data = {"plugin": plg}
|
|
if entity.resource:
|
|
# indicate that we encrypt only for some devices
|
|
data['directed_devices'] = [entity.resource]
|
|
|
|
self._sessions[entity.userhostJID()] = data
|
|
log.info(_(u"Encryption session has been set for {bare_jid} with "
|
|
u"{encryption_name}").format(
|
|
bare_jid=bare_jid.userhost(), encryption_name=plg.name))
|
|
|
|
def stop(self, entity, namespace=None):
|
|
"""Stop an encrypted session with an entity
|
|
|
|
@param entity(jid.JID): entity with who the encrypted session must be stopped
|
|
must be bare jid is the algorithm encrypt for all devices
|
|
@param namespace(unicode): namespace of the session to stop
|
|
when specified, used to check we stop the right encryption session
|
|
"""
|
|
session = self.getSession(entity.userhostJID())
|
|
if not session:
|
|
raise exceptions.NotFound(_(u"There is no encrypted session with this "
|
|
u"entity."))
|
|
if namespace is not None and session[u'plugin'].namespace != namespace:
|
|
raise exceptions.InternalError(_(
|
|
u"The encrypted session is not run with the expected plugin: encrypted "
|
|
u"with {current_name} and was expecting {expected_name}").format(
|
|
current_name=session[u'plugin'].namespace,
|
|
expected_name=namespace))
|
|
if entity.resource:
|
|
try:
|
|
directed_devices = session[u'directed_devices']
|
|
except KeyError:
|
|
raise exceptions.NotFound(_(
|
|
u"There is a session for the whole entity (i.e. all devices of the "
|
|
u"entity), not a directed one. Please use bare jid if you want to "
|
|
u"stop the whole encryption with this entity."))
|
|
|
|
try:
|
|
directed_devices.remove(entity.resource)
|
|
except ValueError:
|
|
raise exceptions.NotFound(_(u"There is no directed session with this "
|
|
u"entity."))
|
|
else:
|
|
del self._sessions[entity]
|
|
|
|
log.info(_(u"Encrypted session stopped with entity {entity}").format(
|
|
entity=entity.full()))
|
|
|
|
def getSession(self, entity):
|
|
"""Get encryption session for this contact
|
|
|
|
@param entity(jid.JID): get the session for this entity
|
|
must be a bare jid
|
|
@return (dict, None): encrypted session data
|
|
None if there is not encryption for this session with this jid
|
|
"""
|
|
if entity.resource:
|
|
raise exceptions.InternalError(u"Full jid given when expecting bare jid")
|
|
return self._sessions.get(entity)
|
|
|
|
## Triggers ##
|
|
|
|
def setEncryptionFlag(self, mess_data):
|
|
"""Set "encryption" key in mess_data if session with destinee is encrypted"""
|
|
|
|
if mess_data["type"] == "groupchat":
|
|
# FIXME: to change when group chat encryption will be handled
|
|
return
|
|
|
|
to_jid = mess_data['to']
|
|
encryption = self._sessions.get(to_jid.userhostJID())
|
|
if encryption is not None:
|
|
mess_data[C.MESS_KEY_ENCRYPTION] = encryption
|