session.py 17.5 KB
Newer Older
1
2
3
4
# -*- coding: utf-8 -*-
# libolm python bindings
# Copyright © 2015-2017 OpenMarket Ltd
# Copyright © 2018 Damir Jelić <poljar@termina.org.uk>
5
6
7
8
9
10
11
12
13
14
15
16
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
"""libolm Session module.

This module contains the Olm Session part of the Olm library.

It is used to establish a peer-to-peer encrypted communication channel between
two Olm accounts.

Examples:
    >>> alice = Account()
    >>> bob = Account()
    >>> bob.generate_one_time_keys(1)
    >>> id_key = bob.identity_keys['curve25519']
    >>> one_time = list(bob.one_time_keys["curve25519"].values())[0]
    >>> session = OutboundSession(alice, id_key, one_time)

"""

# pylint: disable=redefined-builtin,unused-import
from builtins import bytes, super
from typing import AnyStr, Optional, Type

from future.utils import bytes_to_native_str

# pylint: disable=no-name-in-module
from _libolm import ffi, lib  # type: ignore

43
from ._compat import URANDOM, to_bytearray, to_bytes
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
from ._finalize import track_for_finalization

# This is imported only for type checking purposes
if False:
    from .account import Account  # pragma: no cover


class OlmSessionError(Exception):
    """libolm Session exception."""


class _OlmMessage(object):
    def __init__(self, ciphertext, message_type):
        # type: (AnyStr, ffi.cdata) -> None
        if not ciphertext:
            raise ValueError("Ciphertext can't be empty")

        # I don't know why mypy wants a type annotation here nor why AnyStr
        # doesn't work
        self.ciphertext = ciphertext  # type: ignore
        self.message_type = message_type

    def __str__(self):
        # type: () -> str
        type_to_prefix = {
            lib.OLM_MESSAGE_TYPE_PRE_KEY: "PRE_KEY",
            lib.OLM_MESSAGE_TYPE_MESSAGE: "MESSAGE"
        }

        prefix = type_to_prefix[self.message_type]
        return "{} {}".format(prefix, self.ciphertext)


class OlmPreKeyMessage(_OlmMessage):
    """Olm prekey message class

    Prekey messages are used to establish an Olm session. After the first
    message exchange the session switches to normal messages
    """

    def __init__(self, ciphertext):
        # type: (AnyStr) -> None
        """Create a new Olm prekey message with the supplied ciphertext

        Args:
            ciphertext(str): The ciphertext of the prekey message.
        """
        _OlmMessage.__init__(self, ciphertext, lib.OLM_MESSAGE_TYPE_PRE_KEY)

    def __repr__(self):
        # type: () -> str
        return "OlmPreKeyMessage({})".format(self.ciphertext)


class OlmMessage(_OlmMessage):
    """Olm message class"""

    def __init__(self, ciphertext):
        # type: (AnyStr) -> None
        """Create a new Olm message with the supplied ciphertext

        Args:
            ciphertext(str): The ciphertext of the message.
        """
        _OlmMessage.__init__(self, ciphertext, lib.OLM_MESSAGE_TYPE_MESSAGE)

    def __repr__(self):
        # type: () -> str
        return "OlmMessage({})".format(self.ciphertext)


def _clear_session(session):
    # type: (ffi.cdata) -> None
    lib.olm_clear_session(session)


class Session(object):
    """libolm Session class.
    This is an abstract class that can't be instantiated except when unpickling
    a previously pickled InboundSession or OutboundSession object with
    from_pickle.
    """

    def __new__(cls):
        # type: (Type[Session]) -> Session

        obj = super().__new__(cls)
        obj._buf = ffi.new("char[]", lib.olm_session_size())
        obj._session = lib.olm_session(obj._buf)
        track_for_finalization(obj, obj._session, _clear_session)
        return obj

    def __init__(self):
        # type: () -> None
        if type(self) is Session:
            raise TypeError("Session class may not be instantiated.")

        if False:
            self._session = self._session  # type: ffi.cdata

    def _check_error(self, ret):
        # type: (int) -> None
        if ret != lib.olm_error():
            return

        last_error = bytes_to_native_str(
            ffi.string(lib.olm_session_last_error(self._session)))

        raise OlmSessionError(last_error)

    def pickle(self, passphrase=""):
        # type: (Optional[str]) -> bytes
        """Store an Olm session.

        Stores a session as a base64 string. Encrypts the session using the
        supplied passphrase. Returns a byte object containing the base64
        encoded string of the pickled session. Raises OlmSessionError on
        failure.

        Args:
            passphrase(str, optional): The passphrase to be used to encrypt
                the session.
        """
167
        byte_key = bytearray(passphrase, "utf-8") if passphrase else b""
168
169
170
171

        pickle_length = lib.olm_pickle_session_length(self._session)
        pickle_buffer = ffi.new("char[]", pickle_length)

172
173
174
175
176
177
178
179
180
181
182
        try:
            self._check_error(
                lib.olm_pickle_session(self._session,
                                       ffi.from_buffer(byte_key),
                                       len(byte_key),
                                       pickle_buffer, pickle_length))
        finally:
            # clear out copies of the passphrase
            for i in range(0, len(byte_key)):
                    byte_key[i] = 0

183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
        return ffi.unpack(pickle_buffer, pickle_length)

    @classmethod
    def from_pickle(cls, pickle, passphrase=""):
        # type: (bytes, Optional[str]) -> Session
        """Load a previously stored Olm session.

        Loads a session from a pickled base64 string and returns a Session
        object. Decrypts the session using the supplied passphrase. Raises
        OlmSessionError on failure. If the passphrase doesn't match the one
        used to encrypt the session then the error message for the
        exception will be "BAD_ACCOUNT_KEY". If the base64 couldn't be decoded
        then the error message will be "INVALID_BASE64".

        Args:
            pickle(bytes): Base64 encoded byte string containing the pickled
                session
            passphrase(str, optional): The passphrase used to encrypt the
                session.
        """
        if not pickle:
            raise ValueError("Pickle can't be empty")

206
207
        byte_key = bytearray(passphrase, "utf-8") if passphrase else b""
        # copy because unpickle will destroy the buffer
208
209
210
211
        pickle_buffer = ffi.new("char[]", pickle)

        session = cls.__new__(cls)

212
213
214
215
216
217
218
219
220
221
222
        try:
            ret = lib.olm_unpickle_session(session._session,
                                           ffi.from_buffer(byte_key),
                                           len(byte_key),
                                           pickle_buffer,
                                           len(pickle))
            session._check_error(ret)
        finally:
            # clear out copies of the passphrase
            for i in range(0, len(byte_key)):
                    byte_key[i] = 0
223
224
225
226
227
228

        return session

    def encrypt(self, plaintext):
        # type: (AnyStr) -> _OlmMessage
        """Encrypts a message using the session. Returns the ciphertext as a
229
        base64 encoded string on success. Raises OlmSessionError on failure.
230
231
232
233

        Args:
            plaintext(str): The plaintext message that will be encrypted.
        """
234
        byte_plaintext = to_bytearray(plaintext)
235
236
237
238

        r_length = lib.olm_encrypt_random_length(self._session)
        random = URANDOM(r_length)

239
240
        try:
            message_type = lib.olm_encrypt_message_type(self._session)
241

242
            self._check_error(message_type)
243

244
245
246
247
            ciphertext_length = lib.olm_encrypt_message_length(
                self._session, len(byte_plaintext)
            )
            ciphertext_buffer = ffi.new("char[]", ciphertext_length)
248

249
250
251
252
253
254
255
256
257
258
259
            self._check_error(lib.olm_encrypt(
                self._session,
                ffi.from_buffer(byte_plaintext), len(byte_plaintext),
                ffi.from_buffer(random), r_length,
                ciphertext_buffer, ciphertext_length,
            ))
        finally:
            # clear out copies of plaintext
            if byte_plaintext is not plaintext:
                for i in range(0, len(byte_plaintext)):
                    byte_plaintext[i] = 0
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293

        if message_type == lib.OLM_MESSAGE_TYPE_PRE_KEY:
            return OlmPreKeyMessage(
                bytes_to_native_str(ffi.unpack(
                    ciphertext_buffer,
                    ciphertext_length
                )))
        elif message_type == lib.OLM_MESSAGE_TYPE_MESSAGE:
            return OlmMessage(
                bytes_to_native_str(ffi.unpack(
                    ciphertext_buffer,
                    ciphertext_length
                )))
        else:  # pragma: no cover
            raise ValueError("Unknown message type")

    def decrypt(self, message):
        # type: (_OlmMessage) -> str
        """Decrypts a message using the session. Returns the plaintext string
        on success. Raises OlmSessionError on failure. If the base64 couldn't
        be decoded then the error message will be "INVALID_BASE64". If the
        message is for an unsupported version of the protocol the error message
        will be "BAD_MESSAGE_VERSION". If the message couldn't be decoded then
        the error message will be "BAD_MESSAGE_FORMAT". If the MAC on the
        message was invalid then the error message will be "BAD_MESSAGE_MAC".

        Args:
            message(OlmMessage): The Olm message that will be decrypted. It can
            be either an OlmPreKeyMessage or an OlmMessage.
        """
        if not message.ciphertext:
            raise ValueError("Ciphertext can't be empty")

        byte_ciphertext = to_bytes(message.ciphertext)
294
295
        # make a copy the ciphertext buffer, because
        # olm_decrypt_max_plaintext_length wants to destroy something
296
297
298
299
300
301
        ciphertext_buffer = ffi.new("char[]", byte_ciphertext)

        max_plaintext_length = lib.olm_decrypt_max_plaintext_length(
            self._session, message.message_type, ciphertext_buffer,
            len(byte_ciphertext)
        )
302
        self._check_error(max_plaintext_length)
303
        plaintext_buffer = ffi.new("char[]", max_plaintext_length)
304
305
306

        # make a copy the ciphertext buffer, because
        # olm_decrypt_max_plaintext_length wants to destroy something
307
308
        ciphertext_buffer = ffi.new("char[]", byte_ciphertext)
        plaintext_length = lib.olm_decrypt(
309
310
311
            self._session, message.message_type,
            ciphertext_buffer, len(byte_ciphertext),
            plaintext_buffer, max_plaintext_length
312
313
        )
        self._check_error(plaintext_length)
314
        plaintext = bytes_to_native_str(
315
316
            ffi.unpack(plaintext_buffer, plaintext_length))

317
318
319
320
321
        # clear out copies of the plaintext
        lib.memset(plaintext_buffer, 0, max_plaintext_length)

        return plaintext

322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
    @property
    def id(self):
        # type: () -> str
        """str: An identifier for this session. Will be the same for both
        ends of the conversation.
        """
        id_length = lib.olm_session_id_length(self._session)
        id_buffer = ffi.new("char[]", id_length)

        self._check_error(
            lib.olm_session_id(self._session, id_buffer, id_length)
        )
        return bytes_to_native_str(ffi.unpack(id_buffer, id_length))

    def matches(self, message, identity_key=None):
        # type: (OlmPreKeyMessage, Optional[AnyStr]) -> bool
        """Checks if the PRE_KEY message is for this in-bound session.
        This can happen if multiple messages are sent to this session before
        this session sends a message in reply. Returns True if the session
        matches. Returns False if the session does not match. Raises
        OlmSessionError on failure. If the base64 couldn't be decoded then the
        error message will be "INVALID_BASE64". If the message was for an
        unsupported protocol version then the error message will be
        "BAD_MESSAGE_VERSION". If the message couldn't be decoded then then the
        error message will be * "BAD_MESSAGE_FORMAT".

        Args:
            message(OlmPreKeyMessage): The Olm prekey message that will checked
                if it is intended for this session.
            identity_key(str, optional): The identity key of the sender. To
                check if the message was also sent using this identity key.
        """
        if not isinstance(message, OlmPreKeyMessage):
            raise TypeError("Matches can only be called with prekey messages.")

        if not message.ciphertext:
            raise ValueError("Ciphertext can't be empty")

        ret = None

        byte_ciphertext = to_bytes(message.ciphertext)
363
364
        # make a copy, because olm_matches_inbound_session(_from) will distroy
        # it
365
366
367
368
369
370
371
        message_buffer = ffi.new("char[]", byte_ciphertext)

        if identity_key:
            byte_id_key = to_bytes(identity_key)

            ret = lib.olm_matches_inbound_session_from(
                self._session,
372
                ffi.from_buffer(byte_id_key), len(byte_id_key),
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
                message_buffer, len(byte_ciphertext)
            )

        else:
            ret = lib.olm_matches_inbound_session(
                self._session,
                message_buffer, len(byte_ciphertext))

        self._check_error(ret)

        return bool(ret)


class InboundSession(Session):
    """Inbound Olm session for p2p encrypted communication.
    """

    def __new__(cls, account, message, identity_key=None):
        # type: (Account, OlmPreKeyMessage, Optional[AnyStr]) -> Session
        return super().__new__(cls)

    def __init__(self, account, message, identity_key=None):
        # type: (Account, OlmPreKeyMessage, Optional[AnyStr]) -> None
        """Create a new inbound Olm session.

        Create a new in-bound session for sending/receiving messages from an
        incoming prekey message. Raises OlmSessionError on failure. If the
        base64 couldn't be decoded then error message will be "INVALID_BASE64".
        If the message was for an unsupported protocol version then
        the errror message will be "BAD_MESSAGE_VERSION". If the message
        couldn't be decoded then then the error message will be
        "BAD_MESSAGE_FORMAT". If the message refers to an unknown one-time
        key then the error message will be "BAD_MESSAGE_KEY_ID".

        Args:
            account(Account): The Olm Account that will be used to create this
                session.
            message(OlmPreKeyMessage): The Olm prekey message that will checked
                that will be used to create this session.
            identity_key(str, optional): The identity key of the sender. To
                check if the message was also sent using this identity key.
        """
        if not message.ciphertext:
            raise ValueError("Ciphertext can't be empty")

        super().__init__()
        byte_ciphertext = to_bytes(message.ciphertext)
        message_buffer = ffi.new("char[]", byte_ciphertext)

        if identity_key:
            byte_id_key = to_bytes(identity_key)
            identity_key_buffer = ffi.new("char[]", byte_id_key)
            self._check_error(lib.olm_create_inbound_session_from(
                self._session,
                account._account,
                identity_key_buffer, len(byte_id_key),
                message_buffer, len(byte_ciphertext)
            ))
        else:
            self._check_error(lib.olm_create_inbound_session(
                self._session,
                account._account,
                message_buffer, len(byte_ciphertext)
            ))


class OutboundSession(Session):
    """Outbound Olm session for p2p encrypted communication."""

    def __new__(cls, account, identity_key, one_time_key):
        # type: (Account, AnyStr, AnyStr) -> Session
        return super().__new__(cls)

    def __init__(self, account, identity_key, one_time_key):
        # type: (Account, AnyStr, AnyStr) -> None
        """Create a new outbound Olm session.

        Creates a new outbound session for sending messages to a given
        identity key and one-time key.

        Raises OlmSessionError on failure. If the keys couldn't be decoded as
454
        base64 then the error message will be "INVALID_BASE64".
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482

        Args:
            account(Account): The Olm Account that will be used to create this
                session.
            identity_key(str): The identity key of the person with whom we want
                to start the session.
            one_time_key(str): A one-time key from the person with whom we want
                to start the session.
        """
        if not identity_key:
            raise ValueError("Identity key can't be empty")

        if not one_time_key:
            raise ValueError("One-time key can't be empty")

        super().__init__()

        byte_id_key = to_bytes(identity_key)
        byte_one_time = to_bytes(one_time_key)

        session_random_length = lib.olm_create_outbound_session_random_length(
            self._session)

        random = URANDOM(session_random_length)

        self._check_error(lib.olm_create_outbound_session(
            self._session,
            account._account,
483
484
485
            ffi.from_buffer(byte_id_key), len(byte_id_key),
            ffi.from_buffer(byte_one_time), len(byte_one_time),
            ffi.from_buffer(random), session_random_length
486
        ))