diff options
Diffstat (limited to 'python/olm')
| -rw-r--r-- | python/olm/__init__.py | 4 | ||||
| -rwxr-xr-x | python/olm/__main__.py | 339 | ||||
| -rw-r--r-- | python/olm/_base.py | 19 | ||||
| -rw-r--r-- | python/olm/account.py | 121 | ||||
| -rw-r--r-- | python/olm/inbound_group_session.py | 95 | ||||
| -rw-r--r-- | python/olm/outbound_group_session.py | 107 | ||||
| -rw-r--r-- | python/olm/session.py | 192 |
7 files changed, 877 insertions, 0 deletions
diff --git a/python/olm/__init__.py b/python/olm/__init__.py new file mode 100644 index 0000000..31b29b9 --- /dev/null +++ b/python/olm/__init__.py @@ -0,0 +1,4 @@ +from .account import Account +from .session import Session +from .outbound_group_session import OutboundGroupSession +from .inbound_group_session import InboundGroupSession diff --git a/python/olm/__main__.py b/python/olm/__main__.py new file mode 100755 index 0000000..cf9158d --- /dev/null +++ b/python/olm/__main__.py @@ -0,0 +1,339 @@ +#! /usr/bin/env python + +from __future__ import print_function + +import argparse +import json +import os +import sys +import yaml + +from . import * + +def read_base64_file(filename): + """Read a base64 file, dropping any CR/LF characters""" + with open(filename, "rb") as f: + return f.read().translate(None, "\r\n") + +def build_arg_parser(): + parser = argparse.ArgumentParser() + parser.add_argument("--key", help="Account encryption key", default="") + commands = parser.add_subparsers() + + create_account = commands.add_parser("create_account", help="Create a new account") + create_account.add_argument("account_file", help="Local account file") + + def do_create_account(args): + if os.path.exists(args.account_file): + sys.stderr.write("Account %r file already exists" % ( + args.account_file, + )) + sys.exit(1) + account = Account() + account.create() + with open(args.account_file, "wb") as f: + f.write(account.pickle(args.key)) + + create_account.set_defaults(func=do_create_account) + + keys = commands.add_parser("keys", help="List public keys for an account") + keys.add_argument("account_file", help="Local account file") + keys.add_argument("--json", action="store_true", help="Output as JSON") + + def do_keys(args): + account = Account() + account.unpickle(args.key, read_base64_file(args.account_file)) + result = { + "account_keys": account.identity_keys(), + "one_time_keys": account.one_time_keys(), + } + try: + if args.json: + json.dump(result, sys.stdout, indent=4) + else: + yaml.safe_dump(result, sys.stdout, default_flow_style=False) + except: + pass + + keys.set_defaults(func=do_keys) + + def do_id_key(args): + account = Account() + account.unpickle(args.key, read_base64_file(args.account_file)) + print(account.identity_keys()['curve25519']) + + id_key = commands.add_parser("identity_key", help="Get the identity key for an account") + id_key.add_argument("account_file", help="Local account file") + id_key.set_defaults(func=do_id_key) + + def do_one_time_key(args): + account = Account() + account.unpickle(args.key, read_base64_file(args.account_file)) + keys = account.one_time_keys()['curve25519'].values() + key_num = args.key_num + if key_num < 1 or key_num > len(keys): + print( + "Invalid key number %i: %i keys available" % + (key_num, len(keys)), + file=sys.stderr + ) + sys.exit(1) + print (keys[key_num-1]) + + one_time_key = commands.add_parser("one_time_key", + help="Get a one-time key for the account") + one_time_key.add_argument("account_file", help="Local account file") + one_time_key.add_argument("--key-num", "-n", type=int, default=1, + help="Index of key to retrieve (default: 1)") + one_time_key.set_defaults(func=do_one_time_key) + + + sign = commands.add_parser("sign", help="Sign a message") + sign.add_argument("account_file", help="Local account file") + sign.add_argument("message_file", help="Message to sign") + sign.add_argument("signature_file", help="Signature to output") + + def do_sign(args): + account = Account() + account.unpickle(args.key, read_base64_file(args.account_file)) + with open_in(args.message_file) as f: + message = f.read() + signature = account.sign(message) + with open_out(args.signature_file) as f: + f.write(signature) + + sign.set_defaults(func=do_sign) + + + generate_keys = commands.add_parser("generate_keys", help="Generate one time keys") + generate_keys.add_argument("account_file", help="Local account file") + generate_keys.add_argument("count", type=int, help="Number of keys to generate") + + def do_generate_keys(args): + account = Account() + account.unpickle(args.key, read_base64_file(args.account_file)) + account.generate_one_time_keys(args.count) + with open(args.account_file, "wb") as f: + f.write(account.pickle(args.key)) + + generate_keys.set_defaults(func=do_generate_keys) + + + outbound = commands.add_parser("outbound", help="Create an outbound session") + outbound.add_argument("account_file", help="Local account file") + outbound.add_argument("session_file", help="Local session file") + outbound.add_argument("identity_key", help="Remote identity key") + outbound.add_argument("one_time_key", help="Remote one time key") + + def do_outbound(args): + if os.path.exists(args.session_file): + sys.stderr.write("Session %r file already exists" % ( + args.session_file, + )) + sys.exit(1) + account = Account() + account.unpickle(args.key, read_base64_file(args.account_file)) + session = Session() + session.create_outbound( + account, args.identity_key, args.one_time_key + ) + with open(args.session_file, "wb") as f: + f.write(session.pickle(args.key)) + + outbound.set_defaults(func=do_outbound) + + def open_in(path): + if path == "-": + return sys.stdin + else: + return open(path, "rb") + + def open_out(path): + if path == "-": + return sys.stdout + else: + return open(path, "wb") + + inbound = commands.add_parser("inbound", help="Create an inbound session") + inbound.add_argument("account_file", help="Local account file") + inbound.add_argument("session_file", help="Local session file") + inbound.add_argument("message_file", help="Message", default="-") + inbound.add_argument("plaintext_file", help="Plaintext", default="-") + + def do_inbound(args): + if os.path.exists(args.session_file): + sys.stderr.write("Session %r file already exists" % ( + args.session_file, + )) + sys.exit(1) + account = Account() + account.unpickle(args.key, read_base64_file(args.account_file)) + with open_in(args.message_file) as f: + message_type = f.read(8) + message = f.read() + if message_type != "PRE_KEY ": + sys.stderr.write("Expecting a PRE_KEY message") + sys.exit(1) + session = Session() + session.create_inbound(account, message) + plaintext = session.decrypt(0, message) + with open(args.session_file, "wb") as f: + f.write(session.pickle(args.key)) + with open_out(args.plaintext_file) as f: + f.write(plaintext) + + inbound.set_defaults(func=do_inbound) + + session_id = commands.add_parser("session_id", help="Session ID") + session_id.add_argument("session_file", help="Local session file") + + def do_session_id(args): + session = Session() + session.unpickle(args.key, read_base64_file(args.session_file)) + sys.stdout.write(session.session_id() + "\n") + + session_id.set_defaults(func=do_session_id) + + encrypt = commands.add_parser("encrypt", help="Encrypt a message") + encrypt.add_argument("session_file", help="Local session file") + encrypt.add_argument("plaintext_file", help="Plaintext", default="-") + encrypt.add_argument("message_file", help="Message", default="-") + + def do_encrypt(args): + session = Session() + session.unpickle(args.key, read_base64_file(args.session_file)) + with open_in(args.plaintext_file) as f: + plaintext = f.read() + message_type, message = session.encrypt(plaintext) + with open(args.session_file, "wb") as f: + f.write(session.pickle(args.key)) + with open_out(args.message_file) as f: + f.write(["PRE_KEY ", "MESSAGE "][message_type]) + f.write(message) + + encrypt.set_defaults(func=do_encrypt) + + decrypt = commands.add_parser("decrypt", help="Decrypt a message") + decrypt.add_argument("session_file", help="Local session file") + decrypt.add_argument("message_file", help="Message", default="-") + decrypt.add_argument("plaintext_file", help="Plaintext", default="-") + + def do_decrypt(args): + session = Session() + session.unpickle(args.key, read_base64_file(args.session_file)) + with open_in(args.message_file) as f: + message_type = f.read(8) + message = f.read() + if message_type not in {"PRE_KEY ", "MESSAGE "}: + sys.stderr.write("Expecting a PRE_KEY or MESSAGE message") + sys.exit(1) + message_type = 1 if message_type == "MESSAGE " else 0 + plaintext = session.decrypt(message_type, message) + with open(args.session_file, "wb") as f: + f.write(session.pickle(args.key)) + with open_out(args.plaintext_file) as f: + f.write(plaintext) + + decrypt.set_defaults(func=do_decrypt) + + outbound_group = commands.add_parser("outbound_group", help="Create an outbound group session") + outbound_group.add_argument("session_file", help="Local group session file") + outbound_group.set_defaults(func=do_outbound_group) + + group_credentials = commands.add_parser("group_credentials", help="Export the current outbound group session credentials") + group_credentials.add_argument("session_file", help="Local outbound group session file") + group_credentials.add_argument("credentials_file", help="File to write credentials to (default stdout)", + type=argparse.FileType('w'), nargs='?', + default=sys.stdout) + group_credentials.set_defaults(func=do_group_credentials) + + group_encrypt = commands.add_parser("group_encrypt", help="Encrypt a group message") + group_encrypt.add_argument("session_file", help="Local outbound group session file") + group_encrypt.add_argument("plaintext_file", help="Plaintext file (default stdin)", + type=argparse.FileType('rb'), nargs='?', + default=sys.stdin) + group_encrypt.add_argument("message_file", help="Message file (default stdout)", + type=argparse.FileType('w'), nargs='?', + default=sys.stdout) + group_encrypt.set_defaults(func=do_group_encrypt) + + inbound_group = commands.add_parser( + "inbound_group", + help=("Create an inbound group session based on credentials from an "+ + "outbound group session")) + inbound_group.add_argument("session_file", help="Local inbound group session file") + inbound_group.add_argument("credentials_file", + help="File to read credentials from (default stdin)", + type=argparse.FileType('r'), nargs='?', + default=sys.stdin) + inbound_group.set_defaults(func=do_inbound_group) + + group_decrypt = commands.add_parser("group_decrypt", help="Decrypt a group message") + group_decrypt.add_argument("session_file", help="Local inbound group session file") + group_decrypt.add_argument("message_file", help="Message file (default stdin)", + type=argparse.FileType('r'), nargs='?', + default=sys.stdin) + group_decrypt.add_argument("plaintext_file", help="Plaintext file (default stdout)", + type=argparse.FileType('wb'), nargs='?', + default=sys.stdout) + group_decrypt.set_defaults(func=do_group_decrypt) + return parser + +def do_outbound_group(args): + if os.path.exists(args.session_file): + sys.stderr.write("Session %r file already exists" % ( + args.session_file, + )) + sys.exit(1) + session = OutboundGroupSession() + with open(args.session_file, "wb") as f: + f.write(session.pickle(args.key)) + +def do_group_encrypt(args): + session = OutboundGroupSession() + session.unpickle(args.key, read_base64_file(args.session_file)) + plaintext = args.plaintext_file.read() + message = session.encrypt(plaintext) + with open(args.session_file, "wb") as f: + f.write(session.pickle(args.key)) + args.message_file.write(message) + +def do_group_credentials(args): + session = OutboundGroupSession() + session.unpickle(args.key, read_base64_file(args.session_file)) + result = { + 'message_index': session.message_index(), + 'session_key': session.session_key(), + } + json.dump(result, args.credentials_file, indent=4) + +def do_inbound_group(args): + if os.path.exists(args.session_file): + sys.stderr.write("Session %r file already exists\n" % ( + args.session_file, + )) + sys.exit(1) + credentials = json.load(args.credentials_file) + for k in ('session_key', ): + if not k in credentials: + sys.stderr.write("Credentials file is missing %s\n" % k) + sys.exit(1); + + session = InboundGroupSession() + session.init(credentials['session_key']) + with open(args.session_file, "wb") as f: + f.write(session.pickle(args.key)) + +def do_group_decrypt(args): + session = InboundGroupSession() + session.unpickle(args.key, read_base64_file(args.session_file)) + message = args.message_file.read() + plaintext = session.decrypt(message) + with open(args.session_file, "wb") as f: + f.write(session.pickle(args.key)) + args.plaintext_file.write(plaintext) + +if __name__ == '__main__': + parser = build_arg_parser() + args = parser.parse_args() + args.func(args) diff --git a/python/olm/_base.py b/python/olm/_base.py new file mode 100644 index 0000000..bc5c206 --- /dev/null +++ b/python/olm/_base.py @@ -0,0 +1,19 @@ +import os.path + +from ctypes import * + +def read_random(n): + with open("/dev/urandom", "rb") as f: + return f.read(n) + +lib = cdll.LoadLibrary(os.path.join( + os.path.dirname(__file__), "..", "..", "build", "libolm.so.1") +) + +lib.olm_error.argtypes = [] +lib.olm_error.restypes = c_size_t + +ERR = lib.olm_error() + +class OlmError(Exception): + pass diff --git a/python/olm/account.py b/python/olm/account.py new file mode 100644 index 0000000..7673329 --- /dev/null +++ b/python/olm/account.py @@ -0,0 +1,121 @@ +import json + +from ._base import * + +lib.olm_account_size.argtypes = [] +lib.olm_account_size.restype = c_size_t + +lib.olm_account.argtypes = [c_void_p] +lib.olm_account.restype = c_void_p + +lib.olm_account_last_error.argtypes = [c_void_p] +lib.olm_account_last_error.restype = c_char_p + +def account_errcheck(res, func, args): + if res == ERR: + raise OlmError("%s: %s" % ( + func.__name__, lib.olm_account_last_error(args[0]) + )) + return res + + +def account_function(func, *types): + func.argtypes = (c_void_p,) + types + func.restypes = c_size_t + func.errcheck = account_errcheck + + +account_function( + lib.olm_pickle_account, c_void_p, c_size_t, c_void_p, c_size_t +) +account_function( + lib.olm_unpickle_account, c_void_p, c_size_t, c_void_p, c_size_t +) +account_function(lib.olm_create_account_random_length) +account_function(lib.olm_create_account, c_void_p, c_size_t) +account_function(lib.olm_account_identity_keys_length) +account_function(lib.olm_account_identity_keys, c_void_p, c_size_t) +account_function(lib.olm_account_signature_length) +account_function(lib.olm_account_sign, c_void_p, c_size_t, c_void_p, c_size_t) +account_function(lib.olm_account_one_time_keys_length) +account_function(lib.olm_account_one_time_keys, c_void_p, c_size_t) +account_function(lib.olm_account_mark_keys_as_published) +account_function(lib.olm_account_max_number_of_one_time_keys) +account_function( + lib.olm_account_generate_one_time_keys_random_length, + c_size_t +) +account_function( + lib.olm_account_generate_one_time_keys, + c_size_t, + c_void_p, c_size_t +) +class Account(object): + def __init__(self): + self.buf = create_string_buffer(lib.olm_account_size()) + self.ptr = lib.olm_account(self.buf) + + def create(self): + random_length = lib.olm_create_account_random_length(self.ptr) + random = read_random(random_length) + random_buffer = create_string_buffer(random) + lib.olm_create_account(self.ptr, random_buffer, random_length) + + def pickle(self, key): + key_buffer = create_string_buffer(key) + pickle_length = lib.olm_pickle_account_length(self.ptr) + pickle_buffer = create_string_buffer(pickle_length) + lib.olm_pickle_account( + self.ptr, key_buffer, len(key), pickle_buffer, pickle_length + ) + return pickle_buffer.raw + + def unpickle(self, key, pickle): + key_buffer = create_string_buffer(key) + pickle_buffer = create_string_buffer(pickle) + lib.olm_unpickle_account( + self.ptr, key_buffer, len(key), pickle_buffer, len(pickle) + ) + + def identity_keys(self): + out_length = lib.olm_account_identity_keys_length(self.ptr) + out_buffer = create_string_buffer(out_length) + lib.olm_account_identity_keys( + self.ptr, + out_buffer, out_length + ) + return json.loads(out_buffer.raw) + + def sign(self, message): + out_length = lib.olm_account_signature_length(self.ptr) + message_buffer = create_string_buffer(message) + out_buffer = create_string_buffer(out_length) + lib.olm_account_sign( + self.ptr, message_buffer, len(message), out_buffer, out_length + ) + return out_buffer.raw + + def one_time_keys(self): + out_length = lib.olm_account_one_time_keys_length(self.ptr) + out_buffer = create_string_buffer(out_length) + lib.olm_account_one_time_keys(self.ptr, out_buffer, out_length) + return json.loads(out_buffer.raw) + + def mark_keys_as_published(self): + lib.olm_account_mark_keys_as_published(self.ptr) + + def max_number_of_one_time_keys(self): + return lib.olm_account_max_number_of_one_time_keys(self.ptr) + + def generate_one_time_keys(self, count): + random_length = lib.olm_account_generate_one_time_keys_random_length( + self.ptr, count + ) + random = read_random(random_length) + random_buffer = create_string_buffer(random) + lib.olm_account_generate_one_time_keys( + self.ptr, count, random_buffer, random_length + ) + + def clear(self): + pass diff --git a/python/olm/inbound_group_session.py b/python/olm/inbound_group_session.py new file mode 100644 index 0000000..d5547fd --- /dev/null +++ b/python/olm/inbound_group_session.py @@ -0,0 +1,95 @@ +import json + +from ._base import * + +lib.olm_inbound_group_session_size.argtypes = [] +lib.olm_inbound_group_session_size.restype = c_size_t + +lib.olm_inbound_group_session.argtypes = [c_void_p] +lib.olm_inbound_group_session.restype = c_void_p + +lib.olm_inbound_group_session_last_error.argtypes = [c_void_p] +lib.olm_inbound_group_session_last_error.restype = c_char_p + +def inbound_group_session_errcheck(res, func, args): + if res == ERR: + raise OlmError("%s: %s" % ( + func.__name__, lib.olm_inbound_group_session_last_error(args[0]) + )) + return res + + +def inbound_group_session_function(func, *types): + func.argtypes = (c_void_p,) + types + func.restypes = c_size_t + func.errcheck = inbound_group_session_errcheck + + +inbound_group_session_function( + lib.olm_pickle_inbound_group_session, c_void_p, c_size_t, c_void_p, c_size_t +) +inbound_group_session_function( + lib.olm_unpickle_inbound_group_session, c_void_p, c_size_t, c_void_p, c_size_t +) + +inbound_group_session_function( + lib.olm_init_inbound_group_session, c_void_p, c_size_t +) + +inbound_group_session_function( + lib.olm_group_decrypt_max_plaintext_length, c_void_p, c_size_t +) +inbound_group_session_function( + lib.olm_group_decrypt, + c_void_p, c_size_t, # message + c_void_p, c_size_t, # plaintext +) + +inbound_group_session_function(lib.olm_inbound_group_session_id_length) +inbound_group_session_function(lib.olm_inbound_group_session_id, c_void_p, c_size_t) + +class InboundGroupSession(object): + def __init__(self): + self.buf = create_string_buffer(lib.olm_inbound_group_session_size()) + self.ptr = lib.olm_inbound_group_session(self.buf) + + def pickle(self, key): + key_buffer = create_string_buffer(key) + pickle_length = lib.olm_pickle_inbound_group_session_length(self.ptr) + pickle_buffer = create_string_buffer(pickle_length) + lib.olm_pickle_inbound_group_session( + self.ptr, key_buffer, len(key), pickle_buffer, pickle_length + ) + return pickle_buffer.raw + + def unpickle(self, key, pickle): + key_buffer = create_string_buffer(key) + pickle_buffer = create_string_buffer(pickle) + lib.olm_unpickle_inbound_group_session( + self.ptr, key_buffer, len(key), pickle_buffer, len(pickle) + ) + + def init(self, session_key): + key_buffer = create_string_buffer(session_key) + lib.olm_init_inbound_group_session( + self.ptr, key_buffer, len(session_key) + ) + + def decrypt(self, message): + message_buffer = create_string_buffer(message) + max_plaintext_length = lib.olm_group_decrypt_max_plaintext_length( + self.ptr, message_buffer, len(message) + ) + plaintext_buffer = create_string_buffer(max_plaintext_length) + message_buffer = create_string_buffer(message) + plaintext_length = lib.olm_group_decrypt( + self.ptr, message_buffer, len(message), + plaintext_buffer, max_plaintext_length + ) + return plaintext_buffer.raw[:plaintext_length] + + def session_id(self): + id_length = lib.olm_inbound_group_session_id_length(self.ptr) + id_buffer = create_string_buffer(id_length) + lib.olm_inbound_group_session_id(self.ptr, id_buffer, id_length); + return id_buffer.raw diff --git a/python/olm/outbound_group_session.py b/python/olm/outbound_group_session.py new file mode 100644 index 0000000..56f0962 --- /dev/null +++ b/python/olm/outbound_group_session.py @@ -0,0 +1,107 @@ +import json + +from ._base import * + +lib.olm_outbound_group_session_size.argtypes = [] +lib.olm_outbound_group_session_size.restype = c_size_t + +lib.olm_outbound_group_session.argtypes = [c_void_p] +lib.olm_outbound_group_session.restype = c_void_p + +lib.olm_outbound_group_session_last_error.argtypes = [c_void_p] +lib.olm_outbound_group_session_last_error.restype = c_char_p + +def outbound_group_session_errcheck(res, func, args): + if res == ERR: + raise OlmError("%s: %s" % ( + func.__name__, lib.olm_outbound_group_session_last_error(args[0]) + )) + return res + + +def outbound_group_session_function(func, *types): + func.argtypes = (c_void_p,) + types + func.restypes = c_size_t + func.errcheck = outbound_group_session_errcheck + + +outbound_group_session_function( + lib.olm_pickle_outbound_group_session, c_void_p, c_size_t, c_void_p, c_size_t +) +outbound_group_session_function( + lib.olm_unpickle_outbound_group_session, c_void_p, c_size_t, c_void_p, c_size_t +) + +outbound_group_session_function(lib.olm_init_outbound_group_session_random_length) +outbound_group_session_function(lib.olm_init_outbound_group_session, c_void_p, c_size_t) + +lib.olm_outbound_group_session_message_index.argtypes = [c_void_p] +lib.olm_outbound_group_session_message_index.restype = c_uint32 + +outbound_group_session_function(lib.olm_group_encrypt_message_length, c_size_t) +outbound_group_session_function(lib.olm_group_encrypt, + c_void_p, c_size_t, # Plaintext + c_void_p, c_size_t, # Message +) + +outbound_group_session_function(lib.olm_outbound_group_session_id_length) +outbound_group_session_function(lib.olm_outbound_group_session_id, c_void_p, c_size_t) +outbound_group_session_function(lib.olm_outbound_group_session_key_length) +outbound_group_session_function(lib.olm_outbound_group_session_key, c_void_p, c_size_t) + + +class OutboundGroupSession(object): + def __init__(self): + self.buf = create_string_buffer(lib.olm_outbound_group_session_size()) + self.ptr = lib.olm_outbound_group_session(self.buf) + + random_length = lib.olm_init_outbound_group_session_random_length(self.ptr) + random = read_random(random_length) + random_buffer = create_string_buffer(random) + lib.olm_init_outbound_group_session(self.ptr, random_buffer, random_length) + + def pickle(self, key): + key_buffer = create_string_buffer(key) + pickle_length = lib.olm_pickle_outbound_group_session_length(self.ptr) + pickle_buffer = create_string_buffer(pickle_length) + lib.olm_pickle_outbound_group_session( + self.ptr, key_buffer, len(key), pickle_buffer, pickle_length + ) + return pickle_buffer.raw + + def unpickle(self, key, pickle): + key_buffer = create_string_buffer(key) + pickle_buffer = create_string_buffer(pickle) + lib.olm_unpickle_outbound_group_session( + self.ptr, key_buffer, len(key), pickle_buffer, len(pickle) + ) + + def encrypt(self, plaintext): + message_length = lib.olm_group_encrypt_message_length( + self.ptr, len(plaintext) + ) + message_buffer = create_string_buffer(message_length) + + plaintext_buffer = create_string_buffer(plaintext) + + lib.olm_group_encrypt( + self.ptr, + plaintext_buffer, len(plaintext), + message_buffer, message_length, + ) + return message_buffer.raw + + def session_id(self): + id_length = lib.olm_outbound_group_session_id_length(self.ptr) + id_buffer = create_string_buffer(id_length) + lib.olm_outbound_group_session_id(self.ptr, id_buffer, id_length); + return id_buffer.raw + + def message_index(self): + return lib.olm_outbound_group_session_message_index(self.ptr) + + def session_key(self): + key_length = lib.olm_outbound_group_session_key_length(self.ptr) + key_buffer = create_string_buffer(key_length) + lib.olm_outbound_group_session_key(self.ptr, key_buffer, key_length); + return key_buffer.raw diff --git a/python/olm/session.py b/python/olm/session.py new file mode 100644 index 0000000..308f220 --- /dev/null +++ b/python/olm/session.py @@ -0,0 +1,192 @@ +from ._base import * + + +lib.olm_session_size.argtypes = [] +lib.olm_session_size.restype = c_size_t + +lib.olm_session.argtypes = [c_void_p] +lib.olm_session.restype = c_void_p + +lib.olm_session_last_error.argtypes = [c_void_p] +lib.olm_session_last_error.restype = c_char_p + + +def session_errcheck(res, func, args): + if res == ERR: + raise OlmError("%s: %s" % ( + func.__name__, lib.olm_session_last_error(args[0]) + )) + return res + + +def session_function(func, *types): + func.argtypes = (c_void_p,) + types + func.restypes = c_size_t + func.errcheck = session_errcheck + +session_function(lib.olm_session_last_error) +session_function( + lib.olm_pickle_session, c_void_p, c_size_t, c_void_p, c_size_t +) +session_function( + lib.olm_unpickle_session, c_void_p, c_size_t, c_void_p, c_size_t +) +session_function(lib.olm_create_outbound_session_random_length) +session_function( + lib.olm_create_outbound_session, + c_void_p, # Account + c_void_p, c_size_t, # Identity Key + c_void_p, c_size_t, # One Time Key + c_void_p, c_size_t, # Random +) +session_function( + lib.olm_create_inbound_session, + c_void_p, # Account + c_void_p, c_size_t, # Pre Key Message +) +session_function( + lib.olm_create_inbound_session_from, + c_void_p, # Account + c_void_p, c_size_t, # Identity Key + c_void_p, c_size_t, # Pre Key Message +) +session_function(lib.olm_session_id_length) +session_function(lib.olm_session_id, c_void_p, c_size_t) +session_function(lib.olm_matches_inbound_session, c_void_p, c_size_t) +session_function( + lib.olm_matches_inbound_session_from, + c_void_p, c_size_t, # Identity Key + c_void_p, c_size_t, # Pre Key Message +) +session_function(lib.olm_encrypt_message_type) +session_function(lib.olm_encrypt_random_length) +session_function(lib.olm_encrypt_message_length, c_size_t) +session_function( + lib.olm_encrypt, + c_void_p, c_size_t, # Plaintext + c_void_p, c_size_t, # Random + c_void_p, c_size_t, # Message +); +session_function( + lib.olm_decrypt_max_plaintext_length, + c_size_t, # Message Type + c_void_p, c_size_t, # Message +) +session_function( + lib.olm_decrypt, + c_size_t, # Message Type + c_void_p, c_size_t, # Message + c_void_p, c_size_t, # Plaintext +) + +class Session(object): + def __init__(self): + self.buf = create_string_buffer(lib.olm_session_size()) + self.ptr = lib.olm_session(self.buf) + + def pickle(self, key): + key_buffer = create_string_buffer(key) + pickle_length = lib.olm_pickle_session_length(self.ptr) + pickle_buffer = create_string_buffer(pickle_length) + lib.olm_pickle_session( + self.ptr, key_buffer, len(key), pickle_buffer, pickle_length + ) + return pickle_buffer.raw + + def unpickle(self, key, pickle): + key_buffer = create_string_buffer(key) + pickle_buffer = create_string_buffer(pickle) + lib.olm_unpickle_session( + self.ptr, key_buffer, len(key), pickle_buffer, len(pickle) + ) + + def create_outbound(self, account, identity_key, one_time_key): + r_length = lib.olm_create_outbound_session_random_length(self.ptr) + random = read_random(r_length) + random_buffer = create_string_buffer(random) + identity_key_buffer = create_string_buffer(identity_key) + one_time_key_buffer = create_string_buffer(one_time_key) + lib.olm_create_outbound_session( + self.ptr, + account.ptr, + identity_key_buffer, len(identity_key), + one_time_key_buffer, len(one_time_key), + random_buffer, r_length + ) + + def create_inbound(self, account, one_time_key_message): + one_time_key_message_buffer = create_string_buffer(one_time_key_message) + lib.olm_create_inbound_session( + self.ptr, + account.ptr, + one_time_key_message_buffer, len(one_time_key_message) + ) + + def create_inbound_from(self, account, identity_key, one_time_key_message): + identity_key_buffer = create_string_buffer(identity_key) + one_time_key_message_buffer = create_string_buffer(one_time_key_message) + lib.olm_create_inbound_session_from( + self.ptr, + account.ptr, + identity_key_buffer, len(identity_key), + one_time_key_message_buffer, len(one_time_key_message) + ) + + def session_id(self): + id_length = lib.olm_session_id_length(self.ptr) + id_buffer = create_string_buffer(id_length) + lib.olm_session_id(self.ptr, id_buffer, id_length); + return id_buffer.raw + + def matches_inbound(self, one_time_key_message): + one_time_key_message_buffer = create_string_buffer(one_time_key_message) + return bool(lib.olm_matches_inbound_session( + self.ptr, + one_time_key_message_buffer, len(one_time_key_message) + )) + + def matches_inbound_from(self, identity_key, one_time_key_message): + identity_key_buffer = create_string_buffer(identity_key) + one_time_key_message_buffer = create_string_buffer(one_time_key_message) + return bool(lib.olm_matches_inbound_session( + self.ptr, + identity_key_buffer, len(identity_key), + one_time_key_message_buffer, len(one_time_key_message) + )) + + def encrypt(self, plaintext): + r_length = lib.olm_encrypt_random_length(self.ptr) + random = read_random(r_length) + random_buffer = create_string_buffer(random) + + message_type = lib.olm_encrypt_message_type(self.ptr) + message_length = lib.olm_encrypt_message_length( + self.ptr, len(plaintext) + ) + message_buffer = create_string_buffer(message_length) + + plaintext_buffer = create_string_buffer(plaintext) + + lib.olm_encrypt( + self.ptr, + plaintext_buffer, len(plaintext), + random_buffer, r_length, + message_buffer, message_length, + ) + return message_type, message_buffer.raw + + def decrypt(self, message_type, message): + message_buffer = create_string_buffer(message) + max_plaintext_length = lib.olm_decrypt_max_plaintext_length( + self.ptr, message_type, message_buffer, len(message) + ) + plaintext_buffer = create_string_buffer(max_plaintext_length) + message_buffer = create_string_buffer(message) + plaintext_length = lib.olm_decrypt( + self.ptr, message_type, message_buffer, len(message), + plaintext_buffer, max_plaintext_length + ) + return plaintext_buffer.raw[:plaintext_length] + + def clear(self): + pass |