[matrix-client] start decrypting the session keys

matrix/matrix-get-messages-from-encrypted-room-via-backup.py

@@ -132,13 +132,72 @@ class UngleichMatrixClient:
         print(f"Public key = {self.security_public_key}")
 
 
+    def decrypt_session_key(self, encrypted_session_key, ephemeral_key, session_mac):
+
+        # Construct the public ephemeral key
+        # use + b'==') to expand padding https://stackoverflow.com/questions/2941995/python-ignore-incorrect-padding-error-when-base64-decoding
+        ephemeral_key_bytes = base64.b64decode(ephemeral_key + '==')
+        ephemeral_public_key = X25519PublicKey.from_public_bytes(ephemeral_key_bytes)
+        shared_key = private_key.exchange(ephemeral_public_key)
+
+        # when we have shared secret, use HDKF to get the AES part
+        # "Using the shared secret,
+        # generate 80 bytes
+        # by performing an HKDF
+        # using SHA-256 as the hash,
+        # with a salt of 32 bytes of 0,
+        # and with the empty string as the info.
+
+        # The first 32 bytes are used as the AES key,
+        # the next 32 bytes are used as the MAC key,
+        # and the last 16 bytes are used as the AES initialization vector."
+        derived_key = HKDF(
+            algorithm=hashes.SHA256(),
+            length=80,
+            salt=bytes(32),
+            info=b'',
+        ).derive(shared_key)
+
+        print(f"Derived key = %s, len=%s" % (derived_key, len(derived_key) ))
+
+        aes_key = derived_key[:32]
+        mac_key = derived_key[32:64]
+        aes_iv = derived_key[64:]
+
+        print("AES key = {0} / len = {1}".format(aes_key, len(aes_key)))
+        print("Mac key = {0} / len = {1}".format(mac_key, len(mac_key)))
+        print("AES IV = {0} / len = {1}".format(aes_iv, len(aes_iv)))
+
+        # Pass an empty string through HMAC-SHA-256 using the MAC key generated above. The first 8 bytes of the resulting MAC are base64-encoded, and become the mac property of the session_data.
+
+        mac = HMAC(mac_key, hashes.SHA256())
+        mac.update(b'')
+
+        # only use first 8 bytes
+        signature = mac.finalize()[:8]
+        print(f"Calculated signature over empty string = {signature}")
+
+        session_signature = base64.b64decode(session_mac + '==')
+        print(f"Session signature  = {session_signature}")
+
+        if signature == session_signature:
+            print("Signature seems to be correct")
+        else:
+            print("Signature likely incorrect")
+            raise Exception("Session key signature broken")
+
+
     def decrypt_message(self, ciphertext, session_id):
         room_key = self.room_keys['rooms'][self.room_id]['sessions']
         print(f"Messages key data: {room_key}")
 
-        session_key_encrypted = room_key['session_data']['ciphertext']
-        ephemeral_key = room_key['session_data']['ephemeral']
-        session_mac = room_key['session_data']['mac']
+        encrypted_session_key = room_key[session_id]['session_data']['ciphertext']
+        ephemeral_key = room_key[session_id]['session_data']['ephemeral']
+        session_mac = room_key[session_id]['session_data']['mac']
+
+        session_key = self.decrypt_session_key(encrypted_session_key,
+                                               ephemeral_key,
+                                               session_mac)
 
     def decrypt_room_messages(self):
         """