OTR plugin for Gajim 1.0+
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

172 lines
7.4 KiB

## Copyright (C) 2008-2012 Kjell Braden <afflux@pentabarf.de>
## Copyright (C) 2019 Pavel R. <pd at narayana dot im>
# This file is part of Gajim OTR Plugin.
# Gajim OTR Plugin is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published
# by the Free Software Foundation; version 3 only.
# This software is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You can always obtain full license text at <http://www.gnu.org/licenses/>.
import os
import time
import logging
from potr import context, crypt
from gajim.common import const, app, helpers, configpaths
from gajim.session import ChatControlSession
from nbxmpp.protocol import Message, JID
from otrplugin.keystore import Keystore
# OTR channel class
class Channel(context.Context):
def __init__(self, account, peer):
super(Channel, self).__init__(account, peer)
self.getPolicy = OTR.DEFAULT_POLICY.get # default OTR flags
self.trustName = peer.getStripped() # peer name
self.stream = account.stream # XMPP stream
self.ctl = account.getControl(peer.getStripped()) # chat window control
self.resend = []
def println(self, line, kind='status', **kwargs):
try: self.ctl.conv_textview.print_conversation_line(line, kind=kind, tim=None, jid=None, name='', **kwargs)
except TypeError: self.ctl.conv_textview.print_conversation_line(line, kind=kind, tim=None, name='', **kwargs) # gajim git fix
return line
def inject(self, msg, appdata={}):
thread = appdata.get('thread', ChatControlSession.generate_thread_id(None))
stanza = Message(to=self.peer, body=msg.decode(), typ='chat')
def setState(self, newstate):
state, self.state = self.state, newstate
if self.getCurrentTrust() is None:
self.println("OTR: new fingerprint received [%s]" % self.getCurrentKey())
if newstate == context.STATE_ENCRYPTED and state != newstate:
self.println("OTR: %s conversation started [%s]" % (self.getCurrentTrust() and 'trusted' or '**untrusted**', self.getCurrentKey()) )
elif newstate == context.STATE_FINISHED and state != newstate: # channel closed
self.println("OTR: conversation closed.")
# OTR class
class OTR(context.Account):
contextclass = Channel
# OTR const
ENCRYPTION_DATA = helpers.AdditionalDataDict({'encrypted':{'name': ENCRYPTION_NAME}})
MMS = 1000
'ALLOW_V1': False,
'ALLOW_V2': True,
'SEND_TAG': True,
SESSION_START = '?OTRv2?\nI would like to start ' \
'an Off-the-Record private conversation. However, you ' \
'do not have a plugin to support that.\nSee '\
'https://otr.cypherpunks.ca/ for more information.'
def __init__(self, account, logger = None):
super(OTR, self).__init__(account, OTR.PROTOCOL, OTR.MMS)
self.log = logger
self.ctxs, self.ctls = {}, {}
self.account = account
self.stream = app.connections[account]
self.jid = self.stream.get_own_jid()
self.keystore = Keystore(os.path.join(configpaths.get('MY_DATA'), 'otr_' + self.jid.getStripped() + '.db'))
# overload some default methods #
def getControl(self, peer):
ctrl = self.ctls.setdefault(peer, app.interface.msg_win_mgr.get_control(peer, self.account))
return ctrl
def getContext(self, peer):
ctx = self.ctxs.setdefault(peer, Channel(self, peer))
ctx = ctx.state == context.STATE_FINISHED and self.ctxs.pop(peer).disconnect() or self.ctxs.setdefault(peer, Channel(self, peer))
return ctx
def loadPrivkey(self):
my = self.keystore.load({'jid': str(self.jid)})
return crypt.PK.parsePrivateKey(bytes.fromhex(my.privatekey))[0] if my and my.privatekey else None
def savePrivkey(self):
return self.keystore.save({'jid': self.jid, 'privatekey': self.getPrivkey().serializePrivateKey().hex()})
def loadTrusts(self):
for c in self.keystore.load(): self.setTrust(c.jid, c.fingerprint, c.trust)
def saveTrusts(self):
for jid, keys in self.trusts.items():
for fingerprint, trust in keys.items(): self.keystore.save({'jid': jid, 'fingerprint': fingerprint, 'trust': trust})
# decrypt & receive
def _decrypt(self, event, callback):
peer = event.stanza.getFrom()
channel = self.getContext(peer)
text, tlvs = channel.receiveMessage(event.msgtxt.encode(), appdata = {'thread': event.stanza.getThread()})
text = text and text.decode() or ""
except context.UnencryptedMessage:
self.log.error('** got plain text over encrypted channel ** %s' % stanza.getBody())
channel.println("OTR: received plain message [%s]" % event.stanza.getBody())
except context.ErrorReceived as e:
self.log.error('** otr error ** %s' % e)
channel.println("OTR: received error [%s]" % e)
except crypt.InvalidParameterError:
self.log.error('** unreadable message **')
channel.println("OTR: received unreadable message (session expired?)")
except context.NotEncryptedError:
self.log.error('** otr session lost **')
channel.println("OTR: session lost.")
# resent messages after channel open
if channel.resend and channel.state == context.STATE_ENCRYPTED:
message = channel.resend.pop()
channel.println(message['msg'].decode(), kind='outgoing', encrypted=self.ENCRYPTION_NAME, additional_data=self.ENCRYPTION_DATA)
event.xhtml = None
event.msgtxt = text
event.encrypted = self.ENCRYPTION_NAME
event.additional_data = self.ENCRYPTION_DATA
# encrypt & send
def _encrypt(self, event, callback):
peer = event.msg_iq.getTo()
channel = self.getContext(peer)
session = event.session or ChatControlSession(self.stream, peer, None, 'chat')
encrypted = channel.sendMessage(sendPolicy = context.FRAGMENT_SEND_ALL_BUT_LAST, msg = event.message.encode(), appdata = {'thread': session.thread_id}) or b''
except context.NotEncryptedError:
self.log.error("** unable to encrypt message **")
channel.println('OTR: unable to start conversation')
# resend lost message after session start
if encrypted == OTR.SESSION_START.encode():
channel.println('OTR: trying to start encrypted conversation')
channel.resend += [{'sendPolicy': context.FRAGMENT_SEND_ALL, 'msg': event.message.encode(), 'appdata': {'thread': session.thread_id}}]
event.message = ''
event.encrypted = 'OTR'
event.additional_data['encrypted'] = {'name': 'OTR'}