Merge pull request #68 from matrix-org/poljar-python

Poljar's improved python bindings

3 files changed, 357 insertions, 0 deletions

diff --git a/python/tests/account_test.py b/python/tests/account_test.py
new file mode 100644
index 0000000..4fef72c
--- /dev/null
+++ b/python/tests/account_test.py
@@ -0,0 +1,100 @@
+from builtins import int
+
+import pytest
+from hypothesis import given
+from hypothesis.strategies import text
+
+from olm import Account, OlmAccountError, OlmVerifyError, ed25519_verify
+from olm._compat import to_bytes
+
+
+class TestClass(object):
+    def test_to_bytes(self):
+        assert isinstance(to_bytes("a"), bytes)
+        assert isinstance(to_bytes(u"a"), bytes)
+        assert isinstance(to_bytes(b"a"), bytes)
+        assert isinstance(to_bytes(r"a"), bytes)
+        with pytest.raises(TypeError):
+            to_bytes(0)
+
+    def test_account_creation(self):
+        alice = Account()
+        assert alice.identity_keys
+        assert len(alice.identity_keys) == 2
+
+    def test_account_pickle(self):
+        alice = Account()
+        pickle = alice.pickle()
+        assert (alice.identity_keys == Account.from_pickle(pickle)
+                .identity_keys)
+
+    def test_invalid_unpickle(self):
+        with pytest.raises(ValueError):
+            Account.from_pickle(b"")
+
+    def test_passphrase_pickle(self):
+        alice = Account()
+        passphrase = "It's a secret to everybody"
+        pickle = alice.pickle(passphrase)
+        assert (alice.identity_keys == Account.from_pickle(
+            pickle, passphrase).identity_keys)
+
+    def test_wrong_passphrase_pickle(self):
+        alice = Account()
+        passphrase = "It's a secret to everybody"
+        pickle = alice.pickle(passphrase)
+
+        with pytest.raises(OlmAccountError):
+            Account.from_pickle(pickle, "")
+
+    def test_one_time_keys(self):
+        alice = Account()
+        alice.generate_one_time_keys(10)
+        one_time_keys = alice.one_time_keys
+        assert one_time_keys
+        assert len(one_time_keys["curve25519"]) == 10
+
+    def test_max_one_time_keys(self):
+        alice = Account()
+        assert isinstance(alice.max_one_time_keys, int)
+
+    def test_publish_one_time_keys(self):
+        alice = Account()
+        alice.generate_one_time_keys(10)
+        one_time_keys = alice.one_time_keys
+
+        assert one_time_keys
+        assert len(one_time_keys["curve25519"]) == 10
+
+        alice.mark_keys_as_published()
+        assert not alice.one_time_keys["curve25519"]
+
+    def test_clear(self):
+        alice = Account()
+        del alice
+
+    @given(text())
+    def test_valid_signature(self, message):
+        alice = Account()
+
+        signature = alice.sign(message)
+        signing_key = alice.identity_keys["ed25519"]
+
+        assert signature
+        assert signing_key
+
+        ed25519_verify(signing_key, message, signature)
+
+    @given(text())
+    def test_invalid_signature(self, message):
+        alice = Account()
+        bob = Account()
+
+        signature = alice.sign(message)
+        signing_key = bob.identity_keys["ed25519"]
+
+        assert signature
+        assert signing_key
+
+        with pytest.raises(OlmVerifyError):
+            ed25519_verify(signing_key, message, signature)
diff --git a/python/tests/group_session_test.py b/python/tests/group_session_test.py
new file mode 100644
index 0000000..c17e84f
--- /dev/null
+++ b/python/tests/group_session_test.py
@@ -0,0 +1,114 @@
+import pytest
+
+from olm import InboundGroupSession, OlmGroupSessionError, OutboundGroupSession
+
+
+class TestClass(object):
+    def test_session_create(self):
+        OutboundGroupSession()
+
+    def test_session_id(self):
+        session = OutboundGroupSession()
+        assert isinstance(session.id, str)
+
+    def test_session_index(self):
+        session = OutboundGroupSession()
+        assert isinstance(session.message_index, int)
+        assert session.message_index == 0
+
+    def test_outbound_pickle(self):
+        session = OutboundGroupSession()
+        pickle = session.pickle()
+
+        assert (session.id == OutboundGroupSession.from_pickle(
+            pickle).id)
+
+    def test_invalid_unpickle(self):
+        with pytest.raises(ValueError):
+            OutboundGroupSession.from_pickle(b"")
+
+        with pytest.raises(ValueError):
+            InboundGroupSession.from_pickle(b"")
+
+    def test_inbound_create(self):
+        outbound = OutboundGroupSession()
+        InboundGroupSession(outbound.session_key)
+
+    def test_invalid_decrypt(self):
+        outbound = OutboundGroupSession()
+        inbound = InboundGroupSession(outbound.session_key)
+
+        with pytest.raises(ValueError):
+            inbound.decrypt("")
+
+    def test_inbound_pickle(self):
+        outbound = OutboundGroupSession()
+        inbound = InboundGroupSession(outbound.session_key)
+        pickle = inbound.pickle()
+        InboundGroupSession.from_pickle(pickle)
+
+    def test_inbound_export(self):
+        outbound = OutboundGroupSession()
+        inbound = InboundGroupSession(outbound.session_key)
+        imported = InboundGroupSession.import_session(
+            inbound.export_session(inbound.first_known_index)
+        )
+        assert "Test", 0 == imported.decrypt(outbound.encrypt("Test"))
+
+    def test_first_index(self):
+        outbound = OutboundGroupSession()
+        inbound = InboundGroupSession(outbound.session_key)
+        index = inbound.first_known_index
+        assert isinstance(index, int)
+
+    def test_encrypt(self, benchmark):
+        benchmark.weave(OutboundGroupSession.encrypt, lazy=True)
+        outbound = OutboundGroupSession()
+        inbound = InboundGroupSession(outbound.session_key)
+        assert "Test", 0 == inbound.decrypt(outbound.encrypt("Test"))
+
+    def test_decrypt(self, benchmark):
+        benchmark.weave(InboundGroupSession.decrypt, lazy=True)
+        outbound = OutboundGroupSession()
+        inbound = InboundGroupSession(outbound.session_key)
+        assert "Test", 0 == inbound.decrypt(outbound.encrypt("Test"))
+
+    def test_decrypt_twice(self):
+        outbound = OutboundGroupSession()
+        inbound = InboundGroupSession(outbound.session_key)
+        outbound.encrypt("Test 1")
+        message, index = inbound.decrypt(outbound.encrypt("Test 2"))
+        assert isinstance(index, int)
+        assert ("Test 2", 1) == (message, index)
+
+    def test_decrypt_failure(self):
+        outbound = OutboundGroupSession()
+        inbound = InboundGroupSession(outbound.session_key)
+        eve_outbound = OutboundGroupSession()
+        with pytest.raises(OlmGroupSessionError):
+            inbound.decrypt(eve_outbound.encrypt("Test"))
+
+    def test_id(self):
+        outbound = OutboundGroupSession()
+        inbound = InboundGroupSession(outbound.session_key)
+        assert outbound.id == inbound.id
+
+    def test_inbound_fail(self):
+        with pytest.raises(TypeError):
+            InboundGroupSession()
+
+    def test_oubtound_pickle_fail(self):
+        outbound = OutboundGroupSession()
+        pickle = outbound.pickle("Test")
+
+        with pytest.raises(OlmGroupSessionError):
+            OutboundGroupSession.from_pickle(pickle)
+
+    def test_outbound_clear(self):
+        session = OutboundGroupSession()
+        del session
+
+    def test_inbound_clear(self):
+        outbound = OutboundGroupSession()
+        inbound = InboundGroupSession(outbound.session_key)
+        del inbound
diff --git a/python/tests/session_test.py b/python/tests/session_test.py
new file mode 100644
index 0000000..ab1c38b
--- /dev/null
+++ b/python/tests/session_test.py
@@ -0,0 +1,143 @@
+import pytest
+
+from olm import (Account, InboundSession, OlmMessage, OlmPreKeyMessage,
+                 OlmSessionError, OutboundSession, Session)
+
+
+class TestClass(object):
+    def _create_session(self):
+        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)
+        return alice, bob, session
+
+    def test_session_create(self):
+        _, _, session_1 = self._create_session()
+        _, _, session_2 = self._create_session()
+        assert session_1
+        assert session_2
+        assert session_1.id != session_2.id
+        assert isinstance(session_1.id, str)
+
+    def test_session_clear(self):
+        _, _, session = self._create_session()
+        del session
+
+    def test_invalid_session_create(self):
+        with pytest.raises(TypeError):
+            Session()
+
+    def test_session_pickle(self):
+        alice, bob, session = self._create_session()
+        Session.from_pickle(session.pickle()).id == session.id
+
+    def test_session_invalid_pickle(self):
+        with pytest.raises(ValueError):
+            Session.from_pickle(b"")
+
+    def test_wrong_passphrase_pickle(self):
+        alice, bob, session = self._create_session()
+        passphrase = "It's a secret to everybody"
+        pickle = alice.pickle(passphrase)
+
+        with pytest.raises(OlmSessionError):
+            Session.from_pickle(pickle, "")
+
+    def test_encrypt(self):
+        plaintext = "It's a secret to everybody"
+        alice, bob, session = self._create_session()
+        message = session.encrypt(plaintext)
+
+        assert (repr(message)
+                == "OlmPreKeyMessage({})".format(message.ciphertext))
+
+        assert (str(message)
+                == "PRE_KEY {}".format(message.ciphertext))
+
+        bob_session = InboundSession(bob, message)
+        assert plaintext == bob_session.decrypt(message)
+
+    def test_empty_message(self):
+        with pytest.raises(ValueError):
+            OlmPreKeyMessage("")
+        empty = OlmPreKeyMessage("x")
+        empty.ciphertext = ""
+        alice, bob, session = self._create_session()
+
+        with pytest.raises(ValueError):
+            session.decrypt(empty)
+
+    def test_inbound_with_id(self):
+        plaintext = "It's a secret to everybody"
+        alice, bob, session = self._create_session()
+        message = session.encrypt(plaintext)
+        alice_id = alice.identity_keys["curve25519"]
+        bob_session = InboundSession(bob, message, alice_id)
+        assert plaintext == bob_session.decrypt(message)
+
+    def test_two_messages(self):
+        plaintext = "It's a secret to everybody"
+        alice, bob, session = self._create_session()
+        message = session.encrypt(plaintext)
+        alice_id = alice.identity_keys["curve25519"]
+        bob_session = InboundSession(bob, message, alice_id)
+        bob.remove_one_time_keys(bob_session)
+        assert plaintext == bob_session.decrypt(message)
+
+        bob_plaintext = "Grumble, Grumble"
+        bob_message = bob_session.encrypt(bob_plaintext)
+
+        assert (repr(bob_message)
+                == "OlmMessage({})".format(bob_message.ciphertext))
+
+        assert bob_plaintext == session.decrypt(bob_message)
+
+    def test_matches(self):
+        plaintext = "It's a secret to everybody"
+        alice, bob, session = self._create_session()
+        message = session.encrypt(plaintext)
+        alice_id = alice.identity_keys["curve25519"]
+        bob_session = InboundSession(bob, message, alice_id)
+        assert plaintext == bob_session.decrypt(message)
+
+        message_2nd = session.encrypt("Hey! Listen!")
+
+        assert bob_session.matches(message_2nd) is True
+        assert bob_session.matches(message_2nd, alice_id) is True
+
+    def test_invalid(self):
+        alice, bob, session = self._create_session()
+        message = OlmMessage("x")
+
+        with pytest.raises(TypeError):
+            session.matches(message)
+
+        message = OlmPreKeyMessage("x")
+        message.ciphertext = ""
+
+        with pytest.raises(ValueError):
+            session.matches(message)
+
+        with pytest.raises(ValueError):
+            InboundSession(bob, message)
+
+        with pytest.raises(ValueError):
+            OutboundSession(alice, "", "x")
+
+        with pytest.raises(ValueError):
+            OutboundSession(alice, "x", "")
+
+    def test_doesnt_match(self):
+        plaintext = "It's a secret to everybody"
+        alice, bob, session = self._create_session()
+        message = session.encrypt(plaintext)
+        alice_id = alice.identity_keys["curve25519"]
+        bob_session = InboundSession(bob, message, alice_id)
+
+        _, _, new_session = self._create_session()
+
+        new_message = new_session.encrypt(plaintext)
+        assert bob_session.matches(new_message) is False