Commit 3061aa33 authored by Maxime Buquet's avatar Maxime Buquet

First attempt at an example

Signed-off-by: Maxime Buquet's avatarMaxime “pep” Buquet <>
parent a72e81c3
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
Slixmpp OMEMO plugin
Copyright (C) 2010 Nathanael C. Fritz
Copyright (C) 2019 Maxime “pep” Buquet <>
This file is part of slixmpp-omemo.
See the file LICENSE for copying permission.
import os
import sys
import logging
from getpass import getpass
from argparse import ArgumentParser
from slixmpp import ClientXMPP, JID
from slixmpp.stanza import Message
from slixmpp_omemo import PluginCouldNotLoad, MissingOwnKey, EncryptionPrepareException
from slixmpp_omemo import UndecidedException, UntrustedException, NoAvailableSession
import slixmpp_omemo
log = logging.getLogger(__name__)
class EchoBot(ClientXMPP):
A simple Slixmpp bot that will echo encrypted messages it receives, along
with a short thank you message.
For details on how to build a client with slixmpp, look at exemples in the
slixmpp repository.
def __init__(self, jid, password):
ClientXMPP.__init__(self, jid, password)
self.add_event_handler("session_start", self.start)
self.add_event_handler("message", self.message)
self.add_event_handler("message_encryption", self.message)
def start(self, _event) -> None:
Process the session_start event.
Typical actions for the session_start event are
requesting the roster and broadcasting an initial
presence stanza.
event -- An empty dictionary. The session_start
event does not provide any additional
def message(self, msg: Message) -> None:
Process incoming message stanzas. Be aware that this also
includes MUC messages and error messages. It is usually
a good idea to check the messages's type before processing
or sending replies.
msg -- The received message stanza. See the documentation
for stanza objects and the Message stanza to see
how it may be used.
if msg['type'] not in ('chat', 'normal'):
return None
if not self['xep_0384'].is_encrypted(msg):
self.plain_reply(msg, 'This message was not encrypted.\n%(body)s' % msg)
return None
while True:
body = self['xep_0384'].decrypt_message(msg)
self.encrypted_reply(msg, 'Thanks for sending\n%s' % body.decode("utf8"))
except (MissingOwnKey,):
# The message is missing our own key, it was not encrypted for
# us, and we can't decrypt it.
'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.
'I can\'t decrypt this message as it uses an encrypted '
'session I don\'t know about.',
except (UndecidedException,) as exn:
# I don't think the comment below is correct.
# We should be able to read the message whatever the trust
# state. I think we want to force a decision only when
# sending. I think other clients also do this. Conversations,
# dino, etc. Same for UntrustedException, we can just let the
# user know that the sender is untrusted.
# We have not decided yet wether to trust the person sending
# us the message. We must explicitely tell slixmpp what to do.
# In this case, we will automatically trust. In a real
# application, this is where you would prompt the user to
# decide.
self['xep_0384'].trust(JID(exn.bare_jid), exn.device, exn.ik)
'Adding %(device) of %(bare_jid)s in trusted devices.' % exn,
# Now that we added the device in the trust manager, we need
# to try and decrypt it again, (we let it loop).
except (UntrustedException,) as exn:
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.plain_reply(msg, 'I was not able to decrypt the message.')
return None
def plain_reply(self, original_msg, body) -> None:
Helper to reply to messages
mto = original_msg['from']
mtype = original_msg['type']
msg = self.make_message(mto=mto, mtype=mtype)
msg['body'] = body
def encrypted_reply(self, msg, body) -> None:
if __name__ == '__main__':
# Setup the command line arguments.
parser = ArgumentParser(description=EchoBot.__doc__)
# Output verbosity options.
parser.add_argument("-q", "--quiet", help="set logging to ERROR",
action="store_const", dest="loglevel",
const=logging.ERROR, default=logging.INFO)
parser.add_argument("-d", "--debug", help="set logging to DEBUG",
action="store_const", dest="loglevel",
const=logging.DEBUG, default=logging.INFO)
# JID and password options.
parser.add_argument("-j", "--jid", dest="jid",
help="JID to use")
parser.add_argument("-p", "--password", dest="password",
help="password to use")
# Data dir for omemo plugin
DATA_DIR = os.path.join(
parser.add_argument("--data-dir", dest="data_dir",
help="data directory", default=DATA_DIR)
args = parser.parse_args()
# Setup logging.
format='%(levelname)-8s %(message)s')
if args.jid is None:
args.jid = input("Username: ")
if args.password is None:
args.password = getpass("Password: ")
# Setup the EchoBot and register plugins. Note that while plugins may
# have interdependencies, the order in which you register them does
# not matter.
# Ensure OMEMO data dir is created
os.makedirs(args.data_dir, exist_ok=True)
xmpp = EchoBot(args.jid, args.password)
xmpp.register_plugin('xep_0030') # Service Discovery
xmpp.register_plugin('xep_0199') # XMPP Ping
xmpp.register_plugin('xep_0380') # Explicit Message Encryption
'data_dir': args.data_dir,
except (PluginCouldNotLoad,):
log.exception('And error occured when loading the omemo plugin.')
# Connect to the XMPP server and start processing XMPP stanzas.
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