aboutsummaryrefslogtreecommitdiff
path: root/python/olm/outbound_group_session.py
blob: 5032b34c39676db0179aa0adc944038a1f5900dd (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
43
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
import json
from os import urandom

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 = urandom(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