Refuse requests and events signed by banned signing keys

changelog.d/19459.misc

@@ -0,0 +1 @@
+Block federation requests and events authenticated using a known insecure signing key. See [CVE-2026-24044](https://www.cve.org/CVERecord?id=CVE-2026-24044) / [ELEMENTSEC-2025-1670](https://github.com/element-hq/ess-helm/security/advisories/GHSA-qwcj-h6m8-vp6q).

synapse/crypto/keyring.py

@@ -22,6 +22,7 @@
 import abc
 import logging
 from contextlib import ExitStack
+from http import HTTPStatus
 from typing import TYPE_CHECKING, Callable, Iterable
 
 import attr
@@ -60,6 +61,15 @@
 logger = logging.getLogger(__name__)
 
 
+# List of Unpadded Base64 server signing keys that are known to be vulnerable to attack.
+# Incoming requests from homeservers using any of these keys should be refused.
+# Events containing signatures using any of these keys should be refused.
+BANNED_SERVER_SIGNING_KEYS = (
+    # ELEMENTSEC-2025-1670
+    "l/O9hxMVKB6Lg+3Hqf0FQQZhVESQcMzbPN1Cz2nM3og=",
+)
+
+
 @attr.s(slots=True, frozen=True, cmp=False, auto_attribs=True)
 class VerifyJsonRequest:
     """
@@ -349,6 +359,19 @@ async def process_request(self, verify_request: VerifyJsonRequest) -> None:
             if key_result.valid_until_ts < verify_request.minimum_valid_until_ts:
                 continue
 
+            key = encode_verify_key_base64(key_result.verify_key)
+            if key in BANNED_SERVER_SIGNING_KEYS:
+                raise SynapseError(
+                    HTTPStatus.UNAUTHORIZED,
+                    "Server signing key %s:%s for server %s has been banned by this server"
+                    % (
+                        key_result.verify_key.alg,
+                        key_result.verify_key.version,
+                        verify_request.server_name,
+                    ),
+                    Codes.UNAUTHORIZED,
+                )
+
             await self.process_json(key_result.verify_key, verify_request)
             verified = True
 

tests/crypto/test_keyring.py

@@ -20,7 +20,7 @@
 #
 import time
 from typing import Any, cast
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import attr
 import canonicaljson
@@ -238,6 +238,51 @@ def test_verify_json_for_server(self) -> None:
         # self.assertFalse(d.called)
         self.get_success(d)
 
+    def test_verify_json_for_server_using_banned_key(self) -> None:
+        """Ensure that JSON signed using a banned server_signing_key fails verification."""
+        kr = keyring.Keyring(self.hs)
+
+        banned_signing_key = signedjson.key.generate_signing_key("1")
+        r = self.hs.get_datastores().main.store_server_keys_response(
+            "server9",
+            from_server="test",
+            ts_added_ms=int(time.time() * 1000),
+            verify_keys={
+                get_key_id(banned_signing_key): FetchKeyResult(
+                    verify_key=get_verify_key(banned_signing_key), valid_until_ts=1000
+                )
+            },
+            # The entire response gets signed & stored, just include the bits we
+            # care about.
+            response_json={
+                "verify_keys": {
+                    get_key_id(banned_signing_key): {
+                        "key": encode_verify_key_base64(
+                            get_verify_key(banned_signing_key)
+                        )
+                    }
+                }
+            },
+        )
+        self.get_success(r)
+
+        json1: JsonDict = {}
+        signedjson.sign.sign_json(json1, "server9", banned_signing_key)
+
+        # Ensure the signatures check out normally
+        d = kr.verify_json_for_server("server9", json1, 500)
+        self.get_success(d)
+
+        # Patch the list of banned signing keys and ensure the signature check fails
+        with patch.object(
+            keyring,
+            "BANNED_SERVER_SIGNING_KEYS",
+            (encode_verify_key_base64(get_verify_key(banned_signing_key))),
+        ):
+            # should fail on a signed object signed by the banned key
+            d = kr.verify_json_for_server("server9", json1, 500)
+            self.get_failure(d, SynapseError)
+
     def test_verify_for_local_server(self) -> None:
         """Ensure that locally signed JSON can be verified without fetching keys
         over federation

tests/federation/test_federation_base.py

@@ -0,0 +1,68 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl_3.0.html>.
+#
+#
+
+
+from unittest.mock import patch
+
+from signedjson.key import encode_verify_key_base64, get_verify_key
+
+from synapse.crypto import keyring
+from synapse.crypto.event_signing import add_hashes_and_signatures
+from synapse.events import make_event_from_dict
+from synapse.federation.federation_base import InvalidEventSignatureError
+
+from tests import unittest
+
+
+class FederationBaseTestCase(unittest.HomeserverTestCase):
+    def test_events_signed_by_banned_key_are_refused(self) -> None:
+        """Ensure that event JSON signed using a banned server_signing_key fails verification."""
+        event_dict = {
+            "content": {"body": "Here is the message content"},
+            "event_id": "$0:domain",
+            "origin_server_ts": 1000000,
+            "type": "m.room.message",
+            "room_id": "!r:domain",
+            "sender": f"@u:{self.hs.config.server.server_name}",
+            "signatures": {},
+            "unsigned": {"age_ts": 1000000},
+        }
+
+        add_hashes_and_signatures(
+            self.hs.config.server.default_room_version,
+            event_dict,
+            self.hs.config.server.server_name,
+            self.hs.signing_key,
+        )
+        event = make_event_from_dict(event_dict)
+        fs = self.hs.get_federation_server()
+
+        # Ensure the signatures check out normally
+        self.get_success(
+            fs._check_sigs_and_hash(self.hs.config.server.default_room_version, event)
+        )
+
+        # Patch the list of banned signing keys and ensure the signature check fails
+        with patch.object(
+            keyring,
+            "BANNED_SERVER_SIGNING_KEYS",
+            (encode_verify_key_base64(get_verify_key(self.hs.signing_key))),
+        ):
+            self.get_failure(
+                fs._check_sigs_and_hash(
+                    self.hs.config.server.default_room_version, event
+                ),
+                InvalidEventSignatureError,
+            )