Chapter 2. Using python-otr

Table of Contents

Creating the UserState
Loading the private keys and the fingerprints
Connecting python-otr to your program (MessageAppOps)
Connecting your program to python-otr
Receiving messages
Sending messages
Verifying fingerprints' trusts
The manual way (libotr 3.0 compatible)
The modern way using a shared secret (libotr 3.1 and above)

In this section I'll explain more or less detailed how to build an instant messaging client using python-otr. You should still read the reference to learn what the functions do and how they are used.

Creating the UserState

Simple things first: when you want to use python-otr you need to connect the user to its private keys and contexts. To do this, the user needs a OtrlUserState:

my_userstate = otr.otrl_userstate_create()

Loading the private keys and the fingerprints

The fingerprints and private keys are stored in separate files. You have to load them as early as possible, ie. before any communication with a contact takes places, otherwise a new private key may be generated and would overwrite the existing one.

import os
if os.access("keyfile", os.R_OK):
	otr.otrl_privkey_read(my_userstate, "keyfile")
if os.access("fingerprintfile", os.R_OK):
	otr.otrl_privkey_read_fingerprints(my_userstate, "fingerprintfile")

otrl_privkey_read_fingerprints inserts new ConnContext objects to the chain in my_userstate. If you want to add your own custom app_data to those contexts, you can use the add_app_data_callback:

import time

def add_app_data_cb(data=None, context=None):
	if data == "init":
		context.app_data = time.time()

if os.access("fingerprintfile", os.R_OK):
	otr.otrl_privkey_read_fingerprints(my_userstate, "fingerprintfile", (add_app_data_cb, "init"))
[Note]Note

The data == "init" check is rather meaningless here, but I think it demonstrates the use of the data parameter.

[Caution]Caution

Also note that all the callback MUST have data and context as keyword arguments, otherwise the program will abort non-gracesfully at runtime!

Connecting python-otr to your program (MessageAppOps)

Now OTR has to know how to contact your program on several events. This is done via an object which is used as a callback store: it has several attributes (in the python sense) and the python-otr functions will try to call them. If anything goes wrong (eg. wrong arguments) your program will abort non-gracefully at runtime! An example implementation is as follows:

class OtrOps:
	def policy(self, opdata=None, context=None):
		""" checks for the contacts username in policylist and returns it
		if available, otherwise checks for a default entry and returns it
		if available, otherwise just return python-otr's default """

		if context.username in policylist:
			return policylist[context.username]
		elif "__DEFAULT" in policylist:
			return policylist["__DEFAULT"]
		return otr.OTRL_POLICY_DEFAULT

	def create_privkey(self, opdata=None, accountname=None, protocol=None):
		# should give the user some visual feedback here, generating can take some time!
		# the private key MUST be available when this method returned
		otr.otrl_privkey_generate(my_userstate, "keyfile", accountname, protocol)

	def is_logged_in(self, opdata=None, accountname=None, protocol=None, recipient=None):
		return find_contact(recipient, protocol, our_account=accountname).status != "offline"
	
	def inject_message(self, opdata=None, accountname=None, protocol=None, recipient=None, message=None):
		send_message(from=account, proto=protocol, to=recipient, message=message)

	def notify(sef, opdata=None, level=None, accountname=None, protocol=None, username=None, title=None, primary=None, secondary=None):
		# show a small dialog or something like that
		# level is otr.OTRL_NOTIFY_ERROR, otr.OTRL_NOTIFY_WARNING or otr.OTRL_NOTIFY_INFO
		# primary and secondary are the messages that should be displayed

	def display_otr_message(self, opdata=None, accountname=None, protocol=None, username=None, msg=None):
		# this usually logs to the conversation window
		write_message(our_account=accountname, proto=protocol, contact=username, message=msg)
		# NOTE: this function MUST return 0 if it processed the message
		# OR non-zero, the message will then be passed to notify() by OTR
		return 0

	def update_context_list(self, opdata=None):
		# this method may provide some visual feedback when the context list was updated
		# this may be useful if you have a central way of setting fingerprints' trusts
		# and you want to update the list of contexts to consider in this way

	def protocol_name(self, opdata=None, protocol=None):
		""" returns a "human-readable" version of the given protocol """
		if protocol == "xmpp":
			return "XMPP (eXtensible Messaging and Presence Protocol)"
		if protocol == "oscar":
			return "ICQ"

	def new_fingerprint(self, opdata=None, userstate=None, accountname=None, protocol=None, username=None, fingerprint=None):
		human_fingerprint = otr.otrl_privkey_hash_to_human(fingerprint)
		write_message(our_account=accountname, proto=protocol, contact=username,
			message="New fingerprint: %s"%human_fingerprint)

	def write_fingerprints(self, opdata=None):
		otr.otrl_privkey_write_fingerprints(my_userstate, "fingerprintfile")

	def gone_secure(self, opdata=None, context=None):
		trust = context.active_fingerprint.trust
		if trust:
			trust = "verified"
		else:
			trust = "unverified"
		write_message(our_account=accountname, proto=protocol, contact=username,
			"%s secured OTR connection started"%trust)

	def gone_insecure(self, opdata=None, context=None):
		write_message(our_account=accountname, proto=protocol, contact=username,
			"secured OTR connection stopped")

	def still_secure(self, opdata=None, context=None, is_reply=0):
		# this is called when the OTR session was refreshed
		# (ie. new session keys have been created)
		# is_reply will be 0 when we started we started that refresh, 
		#   1 when the contact started it
		write_message(our_account=accountname, proto=protocol, contact=username,
			"secured OTR connection refreshed")

	def log_message(self, opdata=None, message=None):
		# log message to a logfile or something

	def max_message_size(self, opdata=None, context=None):
		""" looks up the max_message_size for the relevant protocol """
		# return 0 when no limit is defined
		return msg_size[context.protocol]

	def account_name(self, opdata=None, account=None, context=None):
		return find_account(accountname=account, protocol).name

Connecting your program to python-otr

Now to get things going you need to tell your program to actually use python-otr for it's messages.

Receiving messages

For receiving messages, they should be processed as early as possible. Something like the following code has to be called before any further processing of the message's contents happen.

is_internal, message, tlvs = gajim.otr_module.otrl_message_receiving(my_userstate, (OtrOps(), None), my_accountname, protocol, contact, message)

if otr.otrl_tlv_find(tlvs, otr.OTRL_TLV_DISCONNECTED) is not None:
	write_message(our_account=my_accountname, proto=protocol, contact=contact,
		"The contact has ended his/her private conversation with you; You should do the same")
	# you MUST NOST set the state here, otrl_message_receiving sets it to FINISHED anyway
	# and setting it to PLAINTEXT could lead to confidential messages being sent
	# unintentionally on an unsecure channel!
	# ONLY inform the user!

# also you should not process the message now if it's empty since it was an internal message then and should be discarded

Sending messages

Sending messages should be done immediatly before transforming the message text to a low-level protocol message (eg. before filling the body-tag of a XMPP message).

A very simpliefied example: The following code was used without python-otr.

def on_send_message(self, recipient, message):
	metadata = self.generate_metadata(recipient, message)

	send_message(from=self.acccount, to=recipient, body=message, meta=metadata)
[...]
def send_message(from="me", to="you", body="Hi", meta=None):
	xml_meta = ""
	if meta:
		xml_meta = gen_xml_from_metadata(meta)

	xml = """<message>
<to>%s</to>
<from>%s</from>
<body>%s</body>

%s
</message>"""%(to, from, body, xml_meta)

	socket.write(xml)

I hope that's not too complicated. Here's what we would use with python-otr.

[...]
class OtrOps:
[...]
	def inject_message(self, opdata=None, accountname=None, protocol=None, recipient=None, message=None):
		send_message(from=acccount, to=recipient, body=message, meta=opdata)
[...]

def on_send_message(self, recipient, message):
	metadata = self.generate_metadata(recipient, message)

	new_message = otrl_message_sending(my_userstate, (OtrOps(), metadata),
		self.account, "some-protocol", recipient, message, None)

	context = otr.otrl_context_find(my_userstate, self.account, recipient, "some-protocol", 1)[0]

	otr.otrl_message_fragment_and_send((OtrOps(), metadata), context, new_message, OTRL_FRAGMENT_SEND_ALL)
[...]
def send_message(from="me", to="you", body="Hi", meta=None):
	xml_meta = ""
	if meta:
		xml_meta = gen_xml_from_metadata(meta)

	xml = """<message>
<to>%s</to>
<from>%s</from>
<body>%s</body>

%s
</message>"""%(to, from, body, xml_meta)

	socket.write(xml)

As you can see, the call to send_message() in on_send_message() disappeared. It actually moved to OtrOps()->inject_message() which will be called by otrl_message_fragment_and_send.

[Note]Note

You would still need the call in on_send_message() when you'd use another fragment policy.

Verifying fingerprints' trusts

The manual way (libotr 3.0 compatible)

You can obtain the human-readable fingerprints for you using otrl_privkey_fingerprint which is quite straightforward.

The contacts fingerprint can be retrieved using the following (or similar).

context = otr.otrl_context_find(my_userstate, my_account, contact, protocol, 1)[0]
human_fingerprint = "unkown"
if context.active_fingerprint is not None:
	human_fingerprint = otr.otrl_privkey_hash_to_human(context.active_fingerprint.fingerprint)

The fingerprint in human_fingerprint has to be checked against the fingerprint the contact sees for himself using a secure channel. If it was successful, the trust for that fingerprint can be set and written.

if context.active_fingerprint is not None:
	context.active_fingerprint.trust = "verified"
	otr.otrl_privkey_write_fingerprints(my_userstate, "fingerprintfile")

The modern way using a shared secret (libotr 3.1 and above)

also known as Socialist Millionaires' Protocol or just SMP

This method obsoletes the way mentioned above, as it avoids the fingerprints (which are hard to remember). However, It depends on a secret which is knewn to both contacts as well they must know that nobody else knows it.

For making this work you need the following post-processing of the messages received (so directly after decrypting of the messages).

def _abort(ctx):
	# notify user about failed verifying here
	# always abort the verifiying using:
	otr.otrl_message_abort_smp(my_userstate, (OtrOps(), None), ctx)

context = otr.otrl_context_find(my_userstate, my_account, contact, protocol, 1)[0]

if otr.otrl_tlv_find(tlvs, otr.OTRL_TLV_SMP_ABORT) is not None:
	# inform the user that SMP was aborted

## BELOW IS NEW SINCE libotr 3.2.0
elif context.smstate.sm_prog_state = otr.OTRL_SMP_PROG_CHEATED:
	# some verification failed, so let's bail out...
	_abort(context)
## ABOVE IS NEW SINCE libotr 3.2.0

elif otr.otrl_tlv_find(tlvs, otr.OTRL_TLV_SMP1) is not None:
	if context.smstate.nextExpected != otr.OTRL_SMP_EXPECT1:
		# unexpected SMP step!
		_abort(context)
	else:
		# remote contact asked for a SMP verifying, ask the local user for his secret here
		# either a respond or an abort TLV should be sent

## BELOW IS NEW SINCE libotr 3.2.0
elif otr.otrl_tlv_find(tlvs, otr.OTRL_TLV_SMP1Q) is not None:
	if context.smstate.nextExpected != otr.OTRL_SMP_EXPECT1: # yes, EXPECT1 works for SMP1Q too
		# unexpected SMP step!
		_abort(context)
	else:
		# ask the local user for his secret using the question specified in tlv.data
		# either a respond or an abort TLV should be sent
## ABOVE IS NEW SINCE libotr 3.2.0

elif otr.otrl_tlv_find(tlvs, otr.OTRL_TLV_SMP2) is not None:
	if context.smstate.nextExpected != otr.OTRL_SMP_EXPECT2:
		# unexpected SMP step!
		_abort(context)
	else:
		context.smstate.nextExpected = otr.OTRL_SMP_EXPECT4
		# we're nearly finished now. you should inform the user about the progress

elif otr.otrl_tlv_find(tlvs, otr.OTRL_TLV_SMP3) is not None:
	if context.smstate.nextExpected != otr.OTRL_SMP_EXPECT3:
		# unexpected SMP step!
		_abort(context)
	else:
		context.smstate.nextExpected = otr.OTRL_SMP_EXPECT1
		# finished, now check the results:
		## NEW SINCE libotr 3.2.0
		if context.smstate.sm_prog_state == otr.OTRL_SMP_PROG_SUCCESS:
			# success!
		else:
			# fail

elif otr.otrl_tlv_find(tlvs, otr.OTRL_TLV_SMP4) is not None:
	if context.smstate.nextExpected != otr.OTRL_SMP_EXPECT4:
		# unexpected SMP step!
		_abort(context)
	else:
		context.smstate.nextExpected = otr.OTRL_SMP_EXPECT1
		# finished, now check the results:
		## NEW SINCE libotr 3.2.0
		if context.smstate.sm_prog_state == otr.OTRL_SMP_PROG_SUCCESS:
			# success!
		else:
			# fail

It gets started by the following functions.

## question IS NEW SINCE libotr 3.2.0
def start_smp(my_account, contact, secret, question=None):
	context = otr.otrl_context_find(my_userstate, my_account, contact, protocol, 1)[0]
	if question is None:
		otr.otrl_message_initiate_smp(my_userstate, (OtrOps(), None), context, secret)
	else:
		otr.otrl_message_initiate_smp_q(my_userstate, (OtrOps(), None), context, question, secret)

def respond_smp(my_account, contact, secret):
	context = otr.otrl_context_find(my_userstate, my_account, contact, protocol, 1)[0]
	otr.otrl_message_respond_smp(my_userstate, (OtrOps(), None), context, secret)

And that's it. You won't need to otrl_privkey_write_fingerprints, SMP will do that for you using the Ops.