Derive generated signing key IDs from public key content

changelog.d/19967.misc

@@ -0,0 +1 @@
+Generate new server signing keys with URL-safe public-key-derived key ids and warn when existing signing keys use numeric ids.

synapse/_scripts/generate_signing_key.py

@@ -23,9 +23,9 @@
 import os
 import sys
 
-from signedjson.key import generate_signing_key, write_signing_keys
+from signedjson.key import write_signing_keys
 
-from synapse.util.stringutils import random_string
+from synapse.util.signing_key import generate_signing_key
 
 
 def main() -> None:
@@ -40,8 +40,7 @@ def main() -> None:
     )
     args = parser.parse_args()
 
-    key_id = "a_" + random_string(4)
-    key = (generate_signing_key(key_id),)
+    key = (generate_signing_key(),)
     if args.output_file == "-":
         write_signing_keys(sys.stdout, key)
     else:

synapse/config/key.py

@@ -23,6 +23,7 @@
 import hashlib
 import logging
 import os
+import re
 from typing import TYPE_CHECKING, Any, Iterator
 
 import attr
@@ -33,15 +34,19 @@
     VerifyKey,
     decode_signing_key_base64,
     decode_verify_key_bytes,
-    generate_signing_key,
     is_signing_algorithm_supported,
     read_signing_keys,
     write_signing_keys,
 )
 from unpaddedbase64 import decode_base64
 
 from synapse.types import JsonDict
-from synapse.util.stringutils import random_string, random_string_with_symbols
+from synapse.util.signing_key import (
+    PLACEHOLDER_SIGNING_KEY_ID,
+    derive_signing_key_version,
+    generate_signing_key,
+)
+from synapse.util.stringutils import random_string_with_symbols
 
 from ._base import Config, ConfigError, read_file
 
@@ -102,6 +107,40 @@
 """
 
 logger = logging.getLogger(__name__)
+_SIGNING_KEY_VERSION_RE = re.compile(r"^[A-Za-z0-9_]+$")
+
+
+def load_signing_keys(lines: list[str]) -> list[SigningKey]:
+    loaded_signing_keys = read_signing_keys(lines)
+    for signing_key in loaded_signing_keys:
+        expected_version = derive_signing_key_version(signing_key)
+        if signing_key.version == expected_version:
+            continue
+        if signing_key.version.isdigit():
+            logger.warning(
+                "Signing key %s:%s uses a numeric key id. Numeric signing key ids "
+                "are deprecated for compatibility reasons; preserving the existing "
+                "key id unchanged.",
+                signing_key.alg,
+                signing_key.version,
+            )
+        elif not _SIGNING_KEY_VERSION_RE.fullmatch(signing_key.version):
+            logger.error(
+                "Signing key %s:%s uses a non-spec-compliant key id. "
+                "Preserving the existing key id unchanged for compatibility.",
+                signing_key.alg,
+                signing_key.version,
+            )
+        else:
+            # TODO: phase out/reject newly introduced non-content-derived signing
+            # key ids loaded via `signing_key` or `signing_key_path`.
+            logger.info(
+                "Signing key %s:%s is not content-derived; expected %s.",
+                signing_key.alg,
+                signing_key.version,
+                expected_version,
+            )
+    return loaded_signing_keys
 
 
 @attr.s(slots=True, auto_attribs=True)
@@ -125,7 +164,7 @@ def read_config(
     ) -> None:
         # the signing key can be specified inline or in a separate file
         if "signing_key" in config:
-            self.signing_key = read_signing_keys([config["signing_key"]])
+            self.signing_key = load_signing_keys([config["signing_key"]])
         else:
             assert config_dir_path is not None
             signing_key_path = config.get("signing_key_path")
@@ -263,7 +302,7 @@ def read_signing_keys(self, signing_key_path: str, name: str) -> list[SigningKey
 
         signing_keys = self.read_file(signing_key_path, name)
         try:
-            loaded_signing_keys = read_signing_keys(
+            loaded_signing_keys = load_signing_keys(
                 [
                     signing_key_line
                     for signing_key_line in signing_keys.splitlines(keepends=False)
@@ -314,16 +353,17 @@ def generate_files(self, config: dict[str, Any], config_dir_path: str) -> None:
             with open(
                 signing_key_path, "w", opener=lambda p, f: os.open(p, f, mode=0o640)
             ) as signing_key_file:
-                key_id = "a_" + random_string(4)
-                write_signing_keys(signing_key_file, (generate_signing_key(key_id),))
+                write_signing_keys(signing_key_file, (generate_signing_key(),))
         else:
             signing_keys = self.read_file(signing_key_path, "signing_key")
             if len(signing_keys.split("\n")[0].split()) == 1:
                 # handle keys in the old format.
-                key_id = "a_" + random_string(4)
                 key = decode_signing_key_base64(
-                    NACL_ED25519, key_id, signing_keys.split("\n")[0]
+                    NACL_ED25519,
+                    PLACEHOLDER_SIGNING_KEY_ID,
+                    signing_keys.split("\n")[0],
                 )
+                key.version = derive_signing_key_version(key)
                 with open(
                     signing_key_path, "w", opener=lambda p, f: os.open(p, f, mode=0o640)
                 ) as signing_key_file:

synapse/util/signing_key.py

@@ -0,0 +1,32 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations 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>.
+#
+import hashlib
+
+from signedjson.key import SigningKey, generate_signing_key as _generate_signing_key
+from unpaddedbase64 import encode_base64
+
+PLACEHOLDER_SIGNING_KEY_ID = "PLACEHOLDER_SIGNING_KEY_ID"
+
+
+def derive_signing_key_version(signing_key: SigningKey) -> str:
+    digest = hashlib.sha256(signing_key.verify_key.encode()).digest()
+    # Matrix key ids do not allow "-" (so, normalize b64url alphabet).
+    # NOTE: "version" is the term used in the codebase, not suffix or ID.
+    return encode_base64(digest[:16], urlsafe=True).replace("-", "_")
+
+
+def generate_signing_key() -> SigningKey:
+    signing_key = _generate_signing_key(PLACEHOLDER_SIGNING_KEY_ID)
+    signing_key.version = derive_signing_key_version(signing_key)
+    return signing_key

tests/config/test_generate.py

@@ -26,9 +26,12 @@
 from contextlib import redirect_stdout
 from io import StringIO
 
+from signedjson.key import generate_signing_key, read_signing_keys, write_signing_keys
+
 from synapse.config.homeserver import HomeServerConfig
 
 from tests import unittest
+from tests.utils import default_config
 
 
 class ConfigGenerationTestCase(unittest.TestCase):
@@ -39,7 +42,7 @@ def setUp(self) -> None:
     def tearDown(self) -> None:
         shutil.rmtree(self.dir)
 
-    def test_generate_config_generates_files(self) -> None:
+    def _generate_config(self) -> None:
         with redirect_stdout(StringIO()):
             HomeServerConfig.load_or_generate_config(
                 "",
@@ -53,6 +56,9 @@ def test_generate_config_generates_files(self) -> None:
                 ],
             )
 
+    def test_generate_config_generates_files(self) -> None:
+        self._generate_config()
+
         self.assertSetEqual(
             {"homeserver.yaml", "lemurs.win.log.config", "lemurs.win.signing.key"},
             set(os.listdir(self.dir)),
@@ -63,6 +69,92 @@ def test_generate_config_generates_files(self) -> None:
             os.path.join(os.getcwd(), "homeserver.log"),
         )
 
+        with open(os.path.join(self.dir, "lemurs.win.signing.key")) as f:
+            keys = read_signing_keys(f)
+
+        self.assertEqual(1, len(keys))
+        self.assertRegex(keys[0].version, r"^[A-Za-z0-9_]{22}$")
+
+    def test_deprecated_one_column_signing_key_is_migrated(self) -> None:
+        self._generate_config()
+
+        signing_key_path = os.path.join(self.dir, "lemurs.win.signing.key")
+        with open(signing_key_path) as f:
+            signing_key = f.read().split()[2]
+
+        with open(signing_key_path, "w") as f:
+            f.write(signing_key + "\n")
+
+        config = HomeServerConfig.load_or_generate_config("", ["-c", self.file])
+        assert config is not None
+
+        with open(signing_key_path) as f:
+            keys = read_signing_keys(f)
+
+        self.assertEqual(1, len(keys))
+        self.assertRegex(keys[0].version, r"^[A-Za-z0-9_]{22}$")
+        self.assertEqual(keys[0].version, config.key.signing_key[0].version)
+
+    def test_numeric_signing_key_version_warns(self) -> None:
+        self._generate_config()
+
+        signing_key = generate_signing_key("1")
+        signing_key_path = os.path.join(self.dir, "lemurs.win.signing.key")
+        with open(signing_key_path, "w") as f:
+            write_signing_keys(f, (signing_key,))
+
+        with self.assertLogs("synapse.config.key", level="WARNING") as logs:
+            config = HomeServerConfig.load_or_generate_config("", ["-c", self.file])
+
+        assert config is not None
+        self.assertEqual("1", config.key.signing_key[0].version)
+        self.assertIn("uses a numeric key id", "\n".join(logs.output))
+
+    def test_inline_numeric_signing_key_version_warns(self) -> None:
+        signing_key = generate_signing_key("1")
+        signing_key_file = StringIO()
+        write_signing_keys(signing_key_file, (signing_key,))
+
+        config_dict = default_config(server_name="test")
+        config_dict["signing_key"] = signing_key_file.getvalue()
+
+        config = HomeServerConfig()
+        with self.assertLogs("synapse.config.key", level="WARNING") as logs:
+            config.parse_config_dict(config_dict, "", "")
+
+        self.assertEqual("1", config.key.signing_key[0].version)
+        self.assertIn("uses a numeric key id", "\n".join(logs.output))
+
+    def test_inline_invalid_signing_key_version_errors(self) -> None:
+        signing_key = generate_signing_key("foo-bar")
+        signing_key_file = StringIO()
+        write_signing_keys(signing_key_file, (signing_key,))
+
+        config_dict = default_config(server_name="test")
+        config_dict["signing_key"] = signing_key_file.getvalue()
+
+        config = HomeServerConfig()
+        with self.assertLogs("synapse.config.key", level="ERROR") as logs:
+            config.parse_config_dict(config_dict, "", "")
+
+        self.assertEqual("foo-bar", config.key.signing_key[0].version)
+        self.assertIn("non-spec-compliant key id", "\n".join(logs.output))
+
+    def test_inline_non_content_derived_signing_key_version_infos(self) -> None:
+        signing_key = generate_signing_key("manual_key_id")
+        signing_key_file = StringIO()
+        write_signing_keys(signing_key_file, (signing_key,))
+
+        config_dict = default_config(server_name="test")
+        config_dict["signing_key"] = signing_key_file.getvalue()
+
+        config = HomeServerConfig()
+        with self.assertLogs("synapse.config.key", level="INFO") as logs:
+            config.parse_config_dict(config_dict, "", "")
+
+        self.assertEqual("manual_key_id", config.key.signing_key[0].version)
+        self.assertIn("not content-derived", "\n".join(logs.output))
+
     def assert_log_filename_is(self, log_config_file: str, expected: str) -> None:
         with open(log_config_file) as f:
             config = f.read()