flake8/black

Author: jagerman

contrib/auth-example.py

@@ -1,11 +1,9 @@
 # Example script for demonstrating X-SOGS-* authentication calculation.
 
 import nacl.bindings as salt
-from nacl.public import PrivateKey, PublicKey
 from nacl.signing import SigningKey
 from hashlib import blake2b, sha512
 from base64 import b64encode
-from typing import Optional
 
 # import time
 # import nacl.utils
@@ -125,10 +123,10 @@ def get_signing_headers(
 # X-SOGS-Pubkey: 00bac6e71efd7dfa4a83c98ed24f254ab2c267f9ccdb172a5280a0444ad24e89cc
 # X-SOGS-Timestamp: 1642472103
 # X-SOGS-Nonce: CdB5nyKVmQGCw6s0Bvv8Ww==
-# X-SOGS-Signature: xxLpXHbomAJMB9AtGMyqvBsXrdd2040y+Ol/IKzElWfKJa3EYZRv1GLO6CTLhrDFUwVQe8PPltyGs54Kd7O5Cg==
+# X-SOGS-Signature: xxLpXHbomAJMB9AtGMyqvBsXrdd2040y+Ol/IKzElWfKJa3EYZRv1GLO6CTLhrDFUwVQe8PPltyGs54Kd7O5Cg==   # noqa E501
 #
 # Blinded headers:
 # X-SOGS-Pubkey: 1598932d4bccbe595a8789d7eb1629cefc483a0eaddc7e20e8fe5c771efafd9af5
 # X-SOGS-Timestamp: 1642472103
 # X-SOGS-Nonce: CdB5nyKVmQGCw6s0Bvv8Ww==
-# X-SOGS-Signature: n4HK33v7gkcz/3pZuWvzmOlY+AbzbpEN1K12dtCc8Gw0m4iP5gUddGKKLEbmoWNhqJeY2S81Lm9uK2DBBN8aCg==
+# X-SOGS-Signature: n4HK33v7gkcz/3pZuWvzmOlY+AbzbpEN1K12dtCc8Gw0m4iP5gUddGKKLEbmoWNhqJeY2S81Lm9uK2DBBN8aCg==  # noqa E501

contrib/blinding.py

@@ -1,5 +1,4 @@
 from nacl.signing import SigningKey, VerifyKey
-from nacl.public import PublicKey
 import nacl.bindings as salt
 from nacl.utils import random
 import nacl.hash
@@ -36,7 +35,8 @@
     assert salt.crypto_scalarmult_ed25519_base_noclamp(ka) == kA
     assert salt.crypto_core_ed25519_is_valid_point(kA)
 
-    ##### Signing (e.g. for X-SOGS-*) with a blinded keypair ka/kA
+    #############################
+    # Signing (e.g. for X-SOGS-*) with a blinded keypair ka/kA
 
     # This generation is *almost* just bog standard Ed25519 but we have one change: when generating
     # r we add kA into the hash r = H(H_rh || kA || M), rather than r = H(H_rh || M), so that there
@@ -56,7 +56,8 @@
 
     assert VerifyKey(kA).verify(message_to_sign, full_sig)
 
-    ##### Sending a DM
+    #############################
+    # Sending a DM
 
     # Our user A above wants to send a SOGS DM to another user B:
     s2 = SigningKey.generate()
@@ -72,7 +73,9 @@
     assert salt.crypto_scalarmult_ed25519_base_noclamp(kb) == kB
     assert salt.crypto_core_ed25519_is_valid_point(kB)
 
-    ##### Finding friends:
+    #############################
+    # Finding friends:
+
     # For example (in reality this would come directly from the known session id):
     friend_xpk = salt.crypto_sign_ed25519_pk_to_curve25519(A)
 
@@ -112,7 +115,8 @@
         print("failed; got neither ±A")
         continue
 
-    ##### Encrypting a SOGS DM
+    #############################
+    # Encrypting a SOGS DM
     msg = 'hello 🎂'
 
     # Step one: calculate a shared secret, sending from A to B.  We're going to calculate:
@@ -138,7 +142,8 @@
 
     data = b'\x00' + ciphertext + nonce
 
-    ##### Decrypting a SOGS DM
+    #############################
+    # Decrypting a SOGS DM
     # Opening the box on the recipient end.
 
     # I receive alongside the message from sogs (i.e. this is the blinded session id minus the '15')

sogs/crypto.py

@@ -69,12 +69,12 @@ def server_encrypt(pk, data):
 
 @functools.lru_cache(maxsize=1024)
 def compute_derived_key_bytes(pk_bytes):
-    """ compute derived key as bytes with no prefix """
+    """compute derived key as bytes with no prefix"""
     return crypto_scalarmult(server_pubkey_hash_bytes, pk_bytes)
 
 
 def compute_derived_id(session_id, prefix='15'):
-    """ compute derived session """
+    """compute derived session"""
     return prefix + binascii.hexlify(
         compute_derived_key_bytes(decode_hex_or_b64(session_id[2:], 32))
     ).decode('ascii')

sogs/db.py

@@ -200,7 +200,7 @@ def add_new_columns(conn):
             'whisper': 'INTEGER REFERENCES users(id)',
             'whisper_mods': 'BOOLEAN NOT NULL DEFAULT FALSE',
             'filtered': 'BOOLEAN NOT NULL DEFAULT FALSE',
-        },
+        }
     }
 
     added = False

sogs/hashing.py

@@ -39,8 +39,8 @@ def blake2b(
     """
 
     return _multipart_hash(
-            nacl.hashlib.blake2b(digest_size=digest_size, key=key, salt=salt, person=person),
-            data)
+        nacl.hashlib.blake2b(digest_size=digest_size, key=key, salt=salt, person=person), data
+    )
 
 
 def sha512(data):
@@ -53,6 +53,4 @@ def sha512(data):
 
     Returns a bytes of length 64.
     """
-    return _multipart_hash(
-            hashlib.sha512(),
-            data)
+    return _multipart_hash(hashlib.sha512(), data)

sogs/model/user.py

@@ -250,7 +250,7 @@ def system_user(self):
 
     @property
     def derived_key(self):
-        """ get the derived key for this user """
+        """get the derived key for this user"""
         if self.session_id[0:2] == '15':
             return self.session_id
         return crypto.compute_derived_id(self.session_id)

sogs/routes/auth.py

@@ -5,7 +5,6 @@
 from ..hashing import blake2b
 
 from flask import request, abort, Response, g
-import string
 import time
 from nacl.signing import VerifyKey
 from nacl.exceptions import BadSignatureError
@@ -219,7 +218,6 @@ def handle_http_auth():
         # TODO: if "blinding required" config option is set then reject the request here
         session_id = '05' + pk.to_curve25519_public_key().encode().hex()
 
-
     try:
         nonce = utils.decode_hex_or_b64(nonce, 16)
     except Exception:
@@ -261,7 +259,6 @@ def handle_http_auth():
     except sqlalchemy.exc.IntegrityError:
         abort_with_reason(http.TOO_EARLY, "Invalid authentication: X-SOGS-Nonce cannot be reused")
 
-
     # Signature validation
 
     # Signature should be on:

tests/auth.py

@@ -1,4 +1,4 @@
-from nacl.signing import VerifyKey, SigningKey
+from nacl.signing import SigningKey
 from nacl.public import PublicKey
 from typing import Optional
 import time
@@ -37,14 +37,15 @@ def x_sogs_raw(
 
     if blinded:
         a = s.to_curve25519_private_key().encode()
-        k = salt.crypto_core_ed25519_scalar_reduce(blake2b(sogs.crypto.server_pubkey_bytes, digest_size=64))
+        k = salt.crypto_core_ed25519_scalar_reduce(
+            blake2b(sogs.crypto.server_pubkey_bytes, digest_size=64)
+        )
         ka = salt.crypto_core_ed25519_scalar_mul(k, a)
         kA = salt.crypto_scalarmult_ed25519_base_noclamp(ka)
         pubkey = '15' + kA.hex()
     else:
         pubkey = '00' + s.verify_key.encode().hex()
 
-
     to_sign = [B.encode(), n, str(ts).encode(), method.encode(), full_path.encode()]
     if body:
         to_sign.append(blake2b(body, digest_size=64))
@@ -54,7 +55,9 @@ def x_sogs_raw(
         r = salt.crypto_core_ed25519_scalar_reduce(sha512([H_rh, kA, *to_sign]))
         sig_R = salt.crypto_scalarmult_ed25519_base_noclamp(r)
         HRAM = salt.crypto_core_ed25519_scalar_reduce(sha512([sig_R, kA, *to_sign]))
-        sig_s = salt.crypto_core_ed25519_scalar_add(r, salt.crypto_core_ed25519_scalar_mul(HRAM, ka))
+        sig_s = salt.crypto_core_ed25519_scalar_add(
+            r, salt.crypto_core_ed25519_scalar_mul(HRAM, ka)
+        )
         sig = sig_R + sig_s
 
     else:
@@ -64,7 +67,7 @@ def x_sogs_raw(
         'X-SOGS-Pubkey': pubkey,
         'X-SOGS-Nonce': sogs.utils.encode_base64(n) if b64_nonce else n.hex(),
         'X-SOGS-Timestamp': str(ts),
-        'X-SOGS-Signature': sogs.utils.encode_base64(sig)
+        'X-SOGS-Signature': sogs.utils.encode_base64(sig),
     }
 
     return h, n, ts, sig

tests/test_auth.py

@@ -75,15 +75,13 @@ def test_auth_basic(client, db):
 
     # Barely good timestamp
     r = client.get(
-        "/auth_test/whoami",
-        headers=x_sogs(a, B, 'GET', '/auth_test/whoami', timestamp_off=86399),
+        "/auth_test/whoami", headers=x_sogs(a, B, 'GET', '/auth_test/whoami', timestamp_off=86399)
     )
     assert r.status_code == 200
     assert r.json == {"user": {"uid": 1, "session_id": session_id}}
 
     r = client.get(
-        "/auth_test/whoami",
-        headers=x_sogs(a, B, 'GET', '/auth_test/whoami', timestamp_off=-86399),
+        "/auth_test/whoami", headers=x_sogs(a, B, 'GET', '/auth_test/whoami', timestamp_off=-86399)
     )
     assert r.status_code == 200
     assert r.json == {"user": {"uid": 1, "session_id": session_id}}
@@ -275,15 +273,13 @@ def test_auth_malformed(client, db):
 
     # Bad timestamps
     r = client.get(
-        "/auth_test/whoami",
-        headers=x_sogs(a, B, 'GET', '/auth_test/whoami', timestamp_off=86401),
+        "/auth_test/whoami", headers=x_sogs(a, B, 'GET', '/auth_test/whoami', timestamp_off=86401)
     )
     assert r.status_code == 425
     assert r.data == b'Invalid authentication: X-SOGS-Timestamp is too far from current time'
 
     r = client.get(
-        "/auth_test/whoami",
-        headers=x_sogs(a, B, 'GET', '/auth_test/whoami', timestamp_off=-86401),
+        "/auth_test/whoami", headers=x_sogs(a, B, 'GET', '/auth_test/whoami', timestamp_off=-86401)
     )
     assert r.status_code == 425
     assert r.data == b'Invalid authentication: X-SOGS-Timestamp is too far from current time'
@@ -332,10 +328,7 @@ def test_auth_batch(client, db):
         {
             'code': 200,
             'headers': {'content-type': 'application/json'},
-            'body': {
-                'user': {'uid': 1, 'session_id': session_id},
-                'body': ['hi', 'world'],
-            },
+            'body': {'user': {'uid': 1, 'session_id': session_id}, 'body': ['hi', 'world']},
         },
         {
             'code': 200,

tests/test_dm.py

@@ -1,5 +1,5 @@
 from request import sogs_get, sogs_post
-from sogs import crypto, config
+from sogs import config
 from sogs.hashing import blake2b
 from sogs.utils import encode_base64
 from sogs.model.user import SystemUser
@@ -31,23 +31,30 @@ def make_post(message, sender, to):
     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)
+        plaintext, aad=None, nonce=nonce, key=key
+    )
     data = b'\x00' + ciphertext + nonce
     return {'message': encode_base64(data)}
 
 
 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', sender=blind_user2, to=blind_user), blind_user2
+        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_banned_user(client, blind_user, blind_user2):
     blind_user2.ban(banned_by=SystemUser())
     r = sogs_post(
-        client, f'/inbox/{blind_user2.session_id}', make_post(b'beep', sender=blind_user, to=blind_user2), 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
 
@@ -60,11 +67,6 @@ def test_dm_send(client, blind_user, blind_user2):
     assert r.status_code == 200
     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,
-            }
+    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}