Update client

irrd/conf/__init__.py

@@ -244,10 +244,7 @@ def _check_staging_config(self) -> List[str]:
         Validate the current staging configuration.
         Returns a list of any errors, or an empty list for a valid config.
         """
-        from irrd.utils.crypto import (
-            ed25519_private_key_from_str,
-            ed25519_public_key_from_str,
-        )
+        from irrd.utils.crypto import eckey_from_str
 
         errors = []
         config = self.user_config_staging
@@ -433,7 +430,7 @@ def _validate_subconfig(key, value):
 
             if details.get("nrtm4_client_initial_public_key"):
                 try:
-                    ed25519_public_key_from_str(details["nrtm4_client_initial_public_key"])
+                    eckey_from_str(details["nrtm4_client_initial_public_key"])
                 except ValueError as ve:
                     errors.append(
                         f"Invalid value for setting nrtm4_client_initial_public_key for source {name}: {ve}"
@@ -469,15 +466,15 @@ def _validate_subconfig(key, value):
 
             if details.get("nrtm4_server_private_key"):
                 try:
-                    ed25519_private_key_from_str(details["nrtm4_server_private_key"])
+                    eckey_from_str(details["nrtm4_server_private_key"], require_private=True)
                 except ValueError as ve:
                     errors.append(
                         f"Invalid value for setting nrtm4_server_private_key for source {name}: {ve}"
                     )
 
             if details.get("nrtm4_server_private_key_next"):
                 try:
-                    ed25519_private_key_from_str(details["nrtm4_server_private_key_next"])
+                    eckey_from_str(details["nrtm4_server_private_key_next"], require_private=True)
                 except ValueError as ve:
                     errors.append(
                         f"Invalid value for setting nrtm4_server_private_key_next for source {name}: {ve}"

irrd/conf/test_conf.py

@@ -6,6 +6,12 @@
 import pytest
 import yaml
 
+from ..mirroring.nrtm4.tests import (
+    MOCK_UNF_PRIVATE_KEY,
+    MOCK_UNF_PRIVATE_KEY_OTHER_STR,
+    MOCK_UNF_PRIVATE_KEY_STR,
+)
+from ..utils.crypto import eckey_public_key_as_str
 from . import (
     ConfigurationError,
     config_init,
@@ -119,10 +125,10 @@ def test_load_valid_reload_valid_config(self, monkeypatch, save_yaml_config, tmp
                     "TESTDB-NRTM4": {
                         "keep_journal": True,
                         "nrtm4_client_notification_file_url": "https://testhost/",
-                        "nrtm4_client_initial_public_key": "kL7kSk56ASeaHl6Nj0eXC3XCHkCzktoPA3ceKz/cjOo=",
+                        "nrtm4_client_initial_public_key": eckey_public_key_as_str(MOCK_UNF_PRIVATE_KEY),
                         "nrtm4_server_base_url": "https://example.com",
-                        "nrtm4_server_private_key": "FalXchs8HIU22Efc3ipNcxVwYwB+Mp0x9TCM9BFtig0=",
-                        "nrtm4_server_private_key_next": "4YDgaXpRDIU8vJbFYeYgPQqEa4YAdHeRF1s6SLdXCsE=",
+                        "nrtm4_server_private_key": MOCK_UNF_PRIVATE_KEY_STR,
+                        "nrtm4_server_private_key_next": MOCK_UNF_PRIVATE_KEY_OTHER_STR,
                         "nrtm4_server_local_path": str(tmpdir),
                         "nrtm4_server_snapshot_frequency": 3600 * 2,
                     },
@@ -457,14 +463,8 @@ def test_load_invalid_config(self, save_yaml_config, tmpdir):
         assert "Unknown setting key: log.unknown" in str(ce.value)
         assert "Unknown key(s) under source TESTDB: unknown" in str(ce.value)
 
-        assert (
-            "Invalid value for setting nrtm4_server_private_key for source TESTDB: Incorrect padding"
-            in str(ce.value)
-        )
-        assert (
-            "Invalid value for setting nrtm4_server_private_key_next for source TESTDB: Incorrect padding"
-            in str(ce.value)
-        )
+        assert "Invalid value for setting nrtm4_server_private_key for source TESTDB:" in str(ce.value)
+        assert "Invalid value for setting nrtm4_server_private_key_next for source TESTDB:" in str(ce.value)
         assert "Setting nrtm4_server_base_url for source TESTDB is not a valid https or file URL." in str(
             ce.value
         )

irrd/integration_tests/run.py

@@ -15,7 +15,7 @@
 import ujson
 import yaml
 from alembic import command, config
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from joserfc.rfc7518.ec_key import ECKey
 from python_graphql_client import GraphqlClient
 
 from irrd.conf import PASSWORD_HASH_DUMMY_VALUE, config_init
@@ -41,7 +41,7 @@
 )
 from irrd.utils.whois_client import whois_query, whois_query_irrd
 
-from ..utils.crypto import ed25519_private_key_as_str, ed25519_public_key_as_str
+from ..utils.crypto import eckey_private_key_as_str, eckey_public_key_as_str
 from .constants import (
     EMAIL_DISCARD_MSGS_COMMAND,
     EMAIL_END,
@@ -924,7 +924,7 @@ def _start_irrds(self):
         with open(self.config_path1, "w") as yaml_file:
             yaml.safe_dump(config1, yaml_file)
 
-        self.nrtm4_private_key = Ed25519PrivateKey.generate()
+        self.nrtm4_private_key = ECKey.generate_key()
         config2 = base_config.copy()
         config2["irrd"]["piddir"] = self.piddir2
         config2["irrd"]["database_url"] = self.database_url2
@@ -944,7 +944,7 @@ def _start_irrds(self):
             "nrtm_host": "127.0.0.1",
             "nrtm_port": str(self.port_whois1),
             "nrtm_access_list": "localhost",
-            "nrtm4_server_private_key": ed25519_private_key_as_str(self.nrtm4_private_key),
+            "nrtm4_server_private_key": eckey_private_key_as_str(self.nrtm4_private_key),
             "nrtm4_server_local_path": self.nrtm4_dir2,
             "nrtm4_server_base_url": f"file://{self.nrtm4_dir2}",
             "nrtm4_server_snapshot_frequency": 3600,
@@ -964,7 +964,7 @@ def _start_irrds(self):
         config3["irrd"]["sources"]["TEST"] = {
             "keep_journal": True,
             "nrtm4_client_notification_file_url": f"file://{self.nrtm4_dir2}update-notification-file.json",
-            "nrtm4_client_initial_public_key": ed25519_public_key_as_str(self.nrtm4_private_key.public_key()),
+            "nrtm4_client_initial_public_key": eckey_public_key_as_str(self.nrtm4_private_key),
         }
         with open(self.config_path3, "w") as yaml_file:
             yaml.safe_dump(config3, yaml_file)

irrd/mirroring/nrtm4/nrtm4_client.py

@@ -1,12 +1,10 @@
-import hashlib
 import logging
 import os
-from base64 import b64decode
 from typing import Any, Dict, List, Optional, Tuple
 from urllib.parse import urlparse
 
 import pydantic
-from cryptography.exceptions import InvalidSignature
+from joserfc.rfc7515.model import CompactSignature
 
 from irrd.conf import get_setting
 from irrd.mirroring.nrtm4.jsonseq import jsonseq_decode
@@ -29,7 +27,7 @@
     NRTM4ClientDatabaseStatus,
 )
 from irrd.storage.queries import DatabaseStatusQuery
-from irrd.utils.crypto import ed25519_public_key_from_str
+from irrd.utils.crypto import eckey_from_config, eckey_from_str, jws_deserialize
 from irrd.utils.misc import format_pydantic_errors
 
 logger = logging.getLogger(__name__)
@@ -107,7 +105,7 @@ def _run_client(self) -> bool:
         )
         return has_loaded_snapshot
 
-    def _retrieve_unf(self) -> Tuple[NRTM4UpdateNotificationFile, str]:
+    def _retrieve_unf(self) -> Tuple[NRTM4UpdateNotificationFile, Optional[str]]:
         """
         Retrieve, verify and parse the Update Notification File.
         Returns the UNF object and the used key in base64 string.
@@ -116,24 +114,17 @@ def _retrieve_unf(self) -> Tuple[NRTM4UpdateNotificationFile, str]:
         if not notification_file_url:  # pragma: no cover
             raise RuntimeError("NRTM4 client called for a source without a Update Notification File URL")
 
-        unf_content, _ = retrieve_file(notification_file_url, return_contents=True)
-        unf_hash = hashlib.sha256(unf_content.encode("ascii")).hexdigest()
-        sig_url = notification_file_url.replace(
-            "update-notification-file.json", f"update-notification-file-signature-{unf_hash}.sig"
-        )
-        legacy_sig_url = notification_file_url + ".sig"
+        unf_signed, _ = retrieve_file(notification_file_url, return_contents=True)
         if "nrtm.db.ripe.net" in notification_file_url:  # pragma: no cover
-            logger.warning(
-                f"Downloading signature from legacy url {legacy_sig_url} instead of expected {sig_url}"
-            )
-            signature, _ = retrieve_file(legacy_sig_url, return_contents=True)
+            # When removing this, also remove Optional[] from return type
+            logger.warning("Expecting raw UNF as source is RIPE*, signature not checked")
+            unf_payload = unf_signed.encode("ascii")
+            used_key = None
         else:
-            signature, _ = retrieve_file(sig_url, return_contents=True)
-
-        used_key = self._validate_unf_signature(unf_content, signature)
+            unf_payload, used_key = self._deserialize_unf(unf_signed)
 
         unf = NRTM4UpdateNotificationFile.model_validate_json(
-            unf_content,
+            unf_payload,
             context={
                 "update_notification_file_scheme": urlparse(notification_file_url).scheme,
                 "expected_values": {
@@ -143,61 +134,64 @@ def _retrieve_unf(self) -> Tuple[NRTM4UpdateNotificationFile, str]:
         )
         return unf, used_key
 
-    def _validate_unf_signature(self, unf_content: str, signature_b64: str) -> str:
+    def _deserialize_unf(self, unf_content: str) -> Tuple[bytes, str]:
         """
         Verify the Update Notification File signature,
-        given the content (before JSON parsing) and a base64 signature.
-        Returns the used key in base64 string.
+        given the content (before JWS deserialize).
+        Returns the deserialized payload and used key in PEM string.
         """
+        compact_signature: Optional[CompactSignature]
         unf_content_bytes = unf_content.encode("utf-8")
-        signature = b64decode(signature_b64, validate=True)
         config_key = get_setting(f"sources.{self.source}.nrtm4_client_initial_public_key")
 
         if self.last_status.current_key:
-            keys = [
+            keys_pem = [
                 self.last_status.current_key,
                 self.last_status.next_key,
             ]
         else:
-            keys = [get_setting(f"sources.{self.source}.nrtm4_client_initial_public_key")]
-
-        for key in keys:
-            if key and self._validate_ed25519_signature(key, unf_content_bytes, signature):
-                return key
-
-        if self.last_status.current_key and self._validate_ed25519_signature(
-            config_key, unf_content_bytes, signature
-        ):
-            # While technically just a "signature not valid case", it is a rather
-            # confusing situation for the user, so gets a special message.
-            msg = (
-                f"{self.source}: No valid signature found for the Update Notification File for signature"
-                f" {signature_b64}. The signature is valid for public key {config_key} set in the"
-                " nrtm4_client_initial_public_key setting, but that is only used for initial validation."
-                f" IRRD is currently expecting the public key {self.last_status.current_key}. If you want to"
-                " clear IRRDs key information and revert to nrtm4_client_initial_public_key, use the"
-                " 'irrdctl nrtmv4 client-clear-known-keys' command."
-            )
-            if self.last_status.next_key:
-                msg += f" or {self.last_status.next_key}"
-            raise NRTM4ClientError(msg)
+            keys_pem = [get_setting(f"sources.{self.source}.nrtm4_client_initial_public_key")]
+
+        for key_pem in keys_pem:
+            if not key_pem:  # pragma: no cover
+                continue
+            pubkey = eckey_from_str(key_pem)
+            try:
+                compact_signature = jws_deserialize(unf_content_bytes, pubkey)
+                return compact_signature.payload, key_pem
+            except ValueError:
+                continue
+
+        if self.last_status.current_key:
+            compact_signature = None
+
+            try:
+                ec_key = eckey_from_config(f"sources.{self.source}.nrtm4_client_initial_public_key")
+                if ec_key:
+                    compact_signature = jws_deserialize(
+                        unf_content_bytes,
+                        ec_key,
+                    )
+            except ValueError:  # pragma: no cover
+                pass
+            if compact_signature:
+                # While technically just a "signature not valid case", it is a rather
+                # confusing situation for the user, so gets a special message.
+                msg = (
+                    f"{self.source}: No valid signature found for the Update Notification File. The signature"
+                    f" is valid for public key {config_key} set in the nrtm4_client_initial_public_key"
+                    " setting, but that is only used for initial validation. IRRD is currently expecting the"
+                    f" public key {self.last_status.current_key}. If you want to clear IRRDs key information"
+                    " and revert to nrtm4_client_initial_public_key, use the 'irrdctl nrtmv4"
+                    " client-clear-known-keys' command."
+                )
+                if self.last_status.next_key:
+                    msg += f" or {self.last_status.next_key}"
+                raise NRTM4ClientError(msg)
         raise NRTM4ClientError(
-            f"{self.source}: No valid signature found for any known keys, signature {signature_b64},"
-            f" considered public keys: {keys}"
+            f"{self.source}: No valid signature found for any known keys, considered public keys: {keys_pem}"
         )
 
-    def _validate_ed25519_signature(self, key_b64: str, content: bytes, signature: bytes) -> bool:
-        """
-        Verify an Ed25519 signature, given the key in base64, and the content
-        and signature in bytes. Returns True or False for validity, raises other
-        exceptions for things like an invalid key format.
-        """
-        try:
-            ed25519_public_key_from_str(key_b64).verify(signature, content)
-            return True
-        except InvalidSignature:
-            return False
-
     def _current_db_status(self) -> Tuple[bool, NRTM4ClientDatabaseStatus]:
         """Look up the current status of self.source in the database."""
         query = DatabaseStatusQuery().source(self.source)

irrd/mirroring/nrtm4/nrtm4_server.py

@@ -1,7 +1,5 @@
-import base64
 import datetime
 import gzip
-import hashlib
 import logging
 import os
 import secrets
@@ -24,7 +22,7 @@
     RPSLDatabaseJournalStatisticsQuery,
     RPSLDatabaseQuery,
 )
-from irrd.utils.crypto import ed25519_private_key_from_config, ed25519_public_key_as_str
+from irrd.utils.crypto import eckey_from_config, eckey_public_key_as_str, jws_serialize
 from irrd.utils.text import remove_auth_hashes
 
 from ...utils.process_support import get_lockfile
@@ -222,13 +220,11 @@ def _write_unf(self) -> None:
         This is based on settings and self.status.
         """
         assert self.status
-        next_signing_private_key = ed25519_private_key_from_config(
+        next_signing_private_key = eckey_from_config(
             f"sources.{self.source}.nrtm4_server_private_key_next", permit_empty=True
         )
         next_signing_public_key = (
-            ed25519_public_key_as_str(next_signing_private_key.public_key())
-            if next_signing_private_key
-            else None
+            eckey_public_key_as_str(next_signing_private_key) if next_signing_private_key else None
         )
         unf = NRTM4UpdateNotificationFile(
             nrtm_version=4,
@@ -251,14 +247,11 @@ def _write_unf(self) -> None:
             ],
         )
         unf_content = unf.model_dump_json(exclude_none=True, include=unf.model_fields_set).encode("ascii")
-        private_key = ed25519_private_key_from_config(f"sources.{self.source}.nrtm4_server_private_key")
+        private_key = eckey_from_config(f"sources.{self.source}.nrtm4_server_private_key")
         assert private_key
-        signature = private_key.sign(unf_content)
-        unf_hash = hashlib.sha256(unf_content).hexdigest()
-        with open(self.path / f"update-notification-file-signature-{unf_hash}.sig", "wb") as sig_file:
-            sig_file.write(base64.b64encode(signature))
-        with open(self.path / "update-notification-file.json", "wb") as unf_file:
-            unf_file.write(unf_content)
+        unf_serialized = jws_serialize(unf_content, private_key)
+        with open(self.path / "update-notification-file.json", "w") as unf_file:
+            unf_file.write(unf_serialized)
         self.status.last_update_notification_file_update = unf.timestamp
 
     def _expire_deltas(self) -> None:

irrd/mirroring/nrtm4/nrtm4_types.py

@@ -4,11 +4,10 @@
 from uuid import UUID
 
 import pydantic
+from joserfc.rfc7518.ec_key import ECKey
 from pytz import UTC
 from typing_extensions import Self
 
-from irrd.utils.crypto import ed25519_public_key_from_str
-
 
 def get_from_pydantic_context(info: pydantic.ValidationInfo, key: str) -> Optional[Any]:
     """
@@ -144,7 +143,7 @@ def validate_timestamp(cls, timestamp: datetime.datetime):
     def validate_next_signing_key(cls, next_signing_key: Optional[str]):
         if next_signing_key:
             try:
-                ed25519_public_key_from_str(next_signing_key)
+                ECKey.import_key(next_signing_key)
             except ValueError as ve:
                 raise ValueError(
                     f"Update Notification File has invalid next_signing_key {next_signing_key}: {ve}"