Switch to new blinded encryption for test DMs

Author: jagerman

sogs/routes/users.py

@@ -85,9 +85,9 @@ def _serialize_message(msg):
 @users.get("/inbox")
 @auth.user_required
 def get_inbox():
+    """gets all messages"""
     if not g.user.is_blinded:
         abort(http.FORBIDDEN)
-    """gets all messages"""
     limit = utils.get_int_param('limit', 100, min=1, max=256, truncate=True)
     return jsonify([_serialize_message(msg) for msg in Message.to(user=g.user, limit=limit)])
 

tests/test_dm.py

@@ -1,7 +1,11 @@
 from request import sogs_get, sogs_post
-from sogs import crypto
-
-from base64 import b64encode
+from sogs import crypto, config
+from sogs.hashing import blake2b
+from sogs.utils import encode_base64
+from sogs.model.user import SystemUser
+import nacl.bindings as salt
+from nacl.utils import random
+import time
 
 
 def test_dm_default_empty(client, blind_user):
@@ -15,34 +19,52 @@ def test_dm_banned_user(client, banned_user):
     assert r.status_code == 403
 
 
-def make_post(data, user):
-    privkey = crypto.compute_derived_key_bytes(user.privkey.encode())
-    sig = crypto.xed25519_sign(privkey, data)
-    return {'message': b64encode(data).decode('ascii'), 'signature': b64encode(sig).decode('ascii')}
+def make_post(message, sender, to):
+    assert sender.is_blinded
+    assert to.is_blinded
+    a = sender.ed_key.to_curve25519_private_key().encode()
+    kA = bytes.fromhex(sender.session_id[2:])
+    kB = bytes.fromhex(to.session_id[2:])
+    key = blake2b(salt.crypto_scalarmult_ed25519_noclamp(a, kB) + kA + kB, digest_size=32)
+
+    # MESSAGE || UNBLINDED_ED_PUBKEY
+    plaintext = message + sender.ed_key.verify_key.encode()
+    nonce = random(24)
+    ciphertext = salt.crypto_aead_xchacha20poly1305_ietf_encrypt(
+            plaintext, aad=None, nonce=nonce, key=key)
+    data = b'\x00' + ciphertext + nonce
+    return {'message': encode_base64(data)}
 
 
-def test_dm_send_from_bannend_user(client, blind_user, banned_user):
+def test_dm_send_from_banned_user(client, blind_user, blind_user2):
+    blind_user2.ban(banned_by=SystemUser())
     r = sogs_post(
-        client, f'/inbox/{blind_user.session_id}', make_post(b'beep', banned_user), banned_user
+        client, f'/inbox/{blind_user.session_id}', make_post(b'beep', sender=blind_user2, to=blind_user), blind_user2
     )
     assert r.status_code == 403
 
 
-def test_dm_send_to_bannend_user(client, blind_user, banned_user):
+def test_dm_send_to_banned_user(client, blind_user, blind_user2):
+    blind_user2.ban(banned_by=SystemUser())
     r = sogs_post(
-        client, f'/inbox/{banned_user.session_id}', make_post(b'beep', blind_user), blind_user
+        client, f'/inbox/{blind_user2.session_id}', make_post(b'beep', sender=blind_user, to=blind_user2), blind_user
     )
     assert r.status_code == 404
 
 
 def test_dm_send(client, blind_user, blind_user2):
-    post = make_post(b'bep', blind_user)
+    post = make_post(b'bep', sender=blind_user, to=blind_user2)
     r = sogs_post(client, f'/inbox/{blind_user2.session_id}', post, blind_user)
     assert r.status_code == 201
     r = sogs_get(client, '/inbox', blind_user2)
     assert r.status_code == 200
-    for entry in r.json:
-        if entry['message'] == post['message']:
-            break
-    else:
-        assert False
+    assert len(r.json) == 1
+    data = r.json[0]
+    now = time.time()
+    assert -1 < data.pop('posted_at') - time.time() < 1
+    assert -1 < data.pop('expires_at') - config.DM_EXPIRY_DAYS*86400 - time.time() < 1
+    assert data == {
+            'id': 1,
+            'message': post['message'],
+            'sender': blind_user.session_id,
+            }