diff --git a/changelog.d/19967.misc b/changelog.d/19967.misc new file mode 100644 index 00000000000..f37ad4a2046 --- /dev/null +++ b/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. diff --git a/synapse/_scripts/generate_signing_key.py b/synapse/_scripts/generate_signing_key.py index ba8aba6ee47..9d1f9d91221 100755 --- a/synapse/_scripts/generate_signing_key.py +++ b/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: diff --git a/synapse/config/key.py b/synapse/config/key.py index bfeeac5e307..c58a123e8a2 100644 --- a/synapse/config/key.py +++ b/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,7 +34,6 @@ VerifyKey, decode_signing_key_base64, decode_verify_key_bytes, - generate_signing_key, is_signing_algorithm_supported, read_signing_keys, write_signing_keys, @@ -41,7 +41,12 @@ 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: diff --git a/synapse/util/signing_key.py b/synapse/util/signing_key.py new file mode 100644 index 00000000000..f11afb94abb --- /dev/null +++ b/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: +# . +# +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 diff --git a/tests/config/test_generate.py b/tests/config/test_generate.py index 5ab96e16e1b..ecb19ea79d6 100644 --- a/tests/config/test_generate.py +++ b/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()