CVE-2022-39254: Other (NVD-CWE-Other)
value: matrix-nio is a Python Matrix client library, designed according to sans I/O principles. Prior to version 0.20, when a users requests a room key from their devices, the software correctly remember the request. Once they receive a forwarded room key, they accept it without checking who the room key came from. This allows homeservers to try to insert room keys of questionable validity, potentially mounting an impersonation attack. Version 0.20 fixes the issue.
CWE description: NVD is only using a subset of CWE for mapping instead of the entire CWE, and the weakness type is not covered by that subset.
CVSS v3 base score: 6.5
Published: 2022-09-29T15:15Z
Repository: https://github.com/poljar/matrix-nio
Language: Python
Files changed: nio/crypto/olm_machine.py
Commit date: 2022-09-28 16:02:39 +0200
Commit message: fix(crypto): Only accept forwarded room keys from our own trusted devices

Diff: diff --git a/nio/crypto/olm_machine.py b/nio/crypto/olm_machine.py
index e274982..0a92999 100644
--- a/nio/crypto/olm_machine.py
+++ b/nio/crypto/olm_machine.py
@@ -1180,34 +1180,26 @@ class Olm:
             content["session_key"],
         )
 
         return event
 
-    # This function is copyrighted under the Apache 2.0 license Zil0
-    def _handle_forwarded_room_key_event(
+    def _should_accept_forward(
         self,
-        sender,  # type: str
-        sender_key,  # type: str
-        payload,  # type: Dict[Any, Any]
-    ):
-        # type: (...) -> Union[ForwardedRoomKeyEvent, BadEventType, None]
-        event = ForwardedRoomKeyEvent.from_dict(payload, sender, sender_key)
-
-        if isinstance(event, (BadEvent, UnknownBadEvent)):
-            return event
-
+        sender: str,
+        sender_key: str,
+        event: ForwardedRoomKeyEvent,
+    ) -> bool:
         if event.algorithm != "m.megolm.v1.aes-sha2":
             logger.error(
                 f"Error: unsupported forwarded room key of type {event.algorithm}"
             )
-            return None
-
-        if event.session_id not in self.outgoing_key_requests:
+            return False
+        elif event.session_id not in self.outgoing_key_requests:
             logger.info(
                 "Ignoring session key we have not requested from device {}.", sender_key
             )
-            return None
+            return False
 
         key_request = self.outgoing_key_requests[event.session_id]
 
         if (
             event.algorithm != key_request.algorithm
@@ -1216,10 +1208,38 @@ class Olm:
         ):
             logger.info(
                 "Ignoring session key with mismatched algorithm, room_id, or "
                 "session id."
             )
+            return False
+
+        device = self.device_store.device_from_sender_key(event.sender, sender_key)
+
+        # Only accept forwarded room keys from our own trusted devices
+        if not device or not device.verified or not device.user_id == self.user_id:
+            logger.warn(
+                "Received a forwarded room key from a untrusted device "
+                f"{event.sender}, {sender_key}"
+            )
+            return False
+
+        return True
+
+    # This function is copyrighted under the Apache 2.0 license Zil0
+    def _handle_forwarded_room_key_event(
+        self,
+        sender,  # type: str
+        sender_key,  # type: str
+        payload,  # type: Dict[Any, Any]
+    ):
+        # type: (...) -> Union[ForwardedRoomKeyEvent, BadEventType, None]
+        event = ForwardedRoomKeyEvent.from_dict(payload, sender, sender_key)
+
+        if isinstance(event, (BadEvent, UnknownBadEvent)):
+            return event
+
+        if not self._should_accept_forward(sender, sender_key, event):
             return None
 
         content = payload["content"]
 
         session_sender_key = content["sender_key"]
@@ -1239,11 +1259,11 @@ class Olm:
             return None
 
         if self.inbound_group_store.add(session):
             self.save_inbound_group_session(session)
 
-        key_request = self.outgoing_key_requests.pop(key_request.request_id)
+        key_request = self.outgoing_key_requests.pop(event.session_id)
         self.store.remove_outgoing_key_request(key_request)
         self.outgoing_to_device_messages.append(
             key_request.as_cancellation(self.user_id, self.device_id)
         )
 

Vulnerable code:     # This function is copyrighted under the Apache 2.0 license Zil0
    def _handle_forwarded_room_key_event(
        sender,  # type: str
        sender_key,  # type: str
        payload,  # type: Dict[Any, Any]
    ):
        # type: (...) -> Union[ForwardedRoomKeyEvent, BadEventType, None]
        event = ForwardedRoomKeyEvent.from_dict(payload, sender, sender_key)

        if isinstance(event, (BadEvent, UnknownBadEvent)):
            return event

            return None

        if event.session_id not in self.outgoing_key_requests:
            return None
        key_request = self.outgoing_key_requests.pop(key_request.request_id)
Fixed code:     def _should_accept_forward(
        sender: str,
        sender_key: str,
        event: ForwardedRoomKeyEvent,
    ) -> bool:
            return False
        elif event.session_id not in self.outgoing_key_requests:
            return False
            return False

        device = self.device_store.device_from_sender_key(event.sender, sender_key)

        # Only accept forwarded room keys from our own trusted devices
        if not device or not device.verified or not device.user_id == self.user_id:
            logger.warn(
                "Received a forwarded room key from a untrusted device "
                f"{event.sender}, {sender_key}"
            )
            return False

        return True

    # This function is copyrighted under the Apache 2.0 license Zil0
    def _handle_forwarded_room_key_event(
        self,
        sender,  # type: str
        sender_key,  # type: str
        payload,  # type: Dict[Any, Any]
    ):
        # type: (...) -> Union[ForwardedRoomKeyEvent, BadEventType, None]
        event = ForwardedRoomKeyEvent.from_dict(payload, sender, sender_key)

        if isinstance(event, (BadEvent, UnknownBadEvent)):
            return event

        if not self._should_accept_forward(sender, sender_key, event):
        key_request = self.outgoing_key_requests.pop(event.session_id)