Update client

irrd/mirroring/nrtm4/nrtm4_client.py

@@ -1,12 +1,13 @@
-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 import jws
+from joserfc.errors import BadSignatureError, JoseError
+from joserfc.rfc7515.model import CompactSignature
+from joserfc.rfc7518.ec_key import ECKey
 
 from irrd.conf import get_setting
 from irrd.mirroring.nrtm4.jsonseq import jsonseq_decode
@@ -29,7 +30,6 @@
     NRTM4ClientDatabaseStatus,
 )
 from irrd.storage.queries import DatabaseStatusQuery
-from irrd.utils.crypto import ed25519_public_key_from_str
 from irrd.utils.misc import format_pydantic_errors
 
 logger = logging.getLogger(__name__)
@@ -107,7 +107,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 +116,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,14 +136,14 @@ 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.
         """
+        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:
@@ -162,42 +155,42 @@ def _validate_unf_signature(self, unf_content: str, signature_b64: str) -> str:
             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)
+            if not key:
+                continue
+            try:
+                pubkey = ECKey.import_key(key)
+            except JoseError as error:
+                logger.error(f"{self.source}: Invalid public key, ignoring: {key}", exc_info=error)
+                continue
+            try:
+                compact_signature = jws.deserialize_compact(unf_content_bytes, pubkey)
+                return compact_signature.payload, key
+            except BadSignatureError:
+                continue
+
+        if self.last_status.current_key:
+            try:
+                compact_signature = jws.deserialize_compact(unf_content_bytes, ECKey.import_key(config_key))
+            except JoseError:
+                compact_signature = None
+            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}"
         )
 
-    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_types.py

@@ -4,11 +4,11 @@
 from uuid import UUID
 
 import pydantic
+from joserfc.errors import JoseError
+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,8 +144,8 @@ 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)
-            except ValueError as ve:
+                ECKey.import_key(next_signing_key)
+            except JoseError as ve:
                 raise ValueError(
                     f"Update Notification File has invalid next_signing_key {next_signing_key}: {ve}"
                 )

irrd/mirroring/nrtm4/tests/__init__.py

@@ -1,13 +1,17 @@
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from joserfc.rfc7518.ec_key import ECKey
 
-from irrd.utils.crypto import ed25519_public_key_as_str
-
-MOCK_UNF_PRIVATE_KEY = Ed25519PrivateKey.from_private_bytes(
-    b"\x15\xa9Wr\x1b<\x1c\x856\xd8G\xdc\xde*Ms\x15pc\x00~2\x9d1\xf50\x8c\xf4\x11m\x8a\r"
+MOCK_UNF_PRIVATE_KEY = ECKey.import_key(
+    "-----BEGIN PRIVATE"
+    " KEY-----\nMIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgGQrdALKHTVC4sVav\nmKUjXaPB22CWZP3t5XSkLqKHMO2hRANCAAQ9U/aaZwLV4koey4Jvu9cRaxiXna9k\naQ3YwrPzZlwd5MQSZ59kfT2+LAbQmXbZg0NGzptqHoOK0YD3YVBjv4kc\n-----END"
+    " PRIVATE KEY-----\n"
 )
-MOCK_UNF_PUBLIC_KEY = ed25519_public_key_as_str(MOCK_UNF_PRIVATE_KEY.public_key())
 
-MOCK_UNF_PRIVATE_KEY_OTHER = Ed25519PrivateKey.from_private_bytes(
-    b"\xe1\x80\xe0izQ\x0c\x85<\xbc\x96\xc5a\xe6 =\n\x84k\x86\x00tw\x91\x17[:H\xb7W\n\xc1"
+MOCK_UNF_PUBLIC_KEY = MOCK_UNF_PRIVATE_KEY.as_pem(private=False).decode("ascii")
+
+MOCK_UNF_PRIVATE_KEY_OTHER = ECKey.import_key(
+    "-----BEGIN PRIVATE"
+    " KEY-----\nMIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgmHtbXrQ0uEcrzeZK\niaK8UnpD5c/YAmqdUHqoHLz997ShRANCAAQ18hSL1o3ynp1kLsfXgZBtlWSYwKvc\nLT2qRj7QeJPxHA6X3XMk7eD6xbdeyNFnLXiKwNPFMPcwRLC6oLN81Fvb\n-----END"
+    " PRIVATE KEY-----\n"
 )
-MOCK_UNF_PUBLIC_KEY_OTHER = ed25519_public_key_as_str(MOCK_UNF_PRIVATE_KEY_OTHER.public_key())
+
+MOCK_UNF_PUBLIC_KEY_OTHER = MOCK_UNF_PRIVATE_KEY_OTHER.as_pem(private=False).decode("ascii")

irrd/mirroring/nrtm4/tests/test_nrtm4_client.py

@@ -1,10 +1,9 @@
-import base64
-import hashlib
 import json
 from tempfile import NamedTemporaryFile
 from uuid import UUID, uuid4
 
 import pytest
+from joserfc import jws
 
 from irrd.mirroring.nrtm4.jsonseq import jsonseq_encode
 from irrd.mirroring.nrtm4.nrtm4_client import NRTM4Client, NRTM4ClientError
@@ -88,17 +87,9 @@ def _mock_retrieve_file(tmp_path, mock_responses):
     def mock_retrieve_file(url, expected_hash=None, return_contents=True):
         url = str(url)
         mock_unf_content = json.dumps(mock_responses[MOCK_UNF_URL])
+        mock_unf_serialized = jws.serialize_compact({"alg": "ES256"}, mock_unf_content, MOCK_UNF_PRIVATE_KEY)
         if url == MOCK_UNF_URL and return_contents:
-            return mock_unf_content, False
-        elif "update-notification-file-signature" in url and return_contents:
-            try:
-                return mock_responses[MOCK_UNF_SIG_URL], False
-            except KeyError:
-                unf_bytes = mock_unf_content.encode("utf-8")
-                unf_hash = hashlib.sha256(unf_bytes).hexdigest()
-                if unf_hash not in url:  # pragma: no cover
-                    raise ValueError(f"Signature URL requested {url}, expected hash {unf_hash}")
-                return base64.b64encode(MOCK_UNF_PRIVATE_KEY.sign(unf_bytes)), False
+            return mock_unf_serialized, False
         elif not return_contents:
             assert url == expected_hash
             destination = NamedTemporaryFile(dir=tmp_path, delete=False)
@@ -172,28 +163,24 @@ def test_valid_from_delta(self, prepare_nrtm4_test, caplog):
         self._assert_import_queries(mock_dh, expect_reload=False)
         assert "Updating from deltas, starting from version 3" in caplog.text
 
-    def test_invalid_signature(self, prepare_nrtm4_test, monkeypatch, tmp_path):
-        mock_responses = {
-            MOCK_UNF_URL: MOCK_UNF,
-            MOCK_UNF_SIG_URL: "invalid-base64",
-            MOCK_SNAPSHOT_URL: MOCK_SNAPSHOT,
-            # Shorten delta 3 to header only
-            MOCK_DELTA3_URL: MOCK_DELTA3,
-            MOCK_DELTA4_URL: MOCK_DELTA4,
-        }
-        monkeypatch.setattr(
-            "irrd.mirroring.nrtm4.nrtm4_client.retrieve_file", _mock_retrieve_file(tmp_path, mock_responses)
+    def test_invalid_signature(self, prepare_nrtm4_test, monkeypatch, tmp_path, config_override):
+        config_override(
+            {
+                "sources": {
+                    "TEST": {
+                        "nrtm4_client_notification_file_url": MOCK_UNF_URL,
+                        "nrtm4_client_initial_public_key": MOCK_UNF_PUBLIC_KEY_OTHER,
+                    },
+                },
+                "rpki": {"roa_source": None},
+            }
         )
         mock_dh = MockDatabaseHandler()
         mock_dh.reset_mock()
         mock_dh.query_responses[DatabaseStatusQuery] = iter([])
         with pytest.raises(NRTM4ClientError):
             NRTM4Client("TEST", mock_dh).run_client()
 
-        mock_responses[MOCK_UNF_SIG_URL] = base64.b64encode(b"invalid-key")
-        with pytest.raises(NRTM4ClientError):
-            NRTM4Client("TEST", mock_dh).run_client()
-
     def test_invalid_empty_delta(self, prepare_nrtm4_test, tmp_path, monkeypatch):
         mock_responses = {
             MOCK_UNF_URL: MOCK_UNF,
@@ -434,7 +421,7 @@ def test_invalid_current_db_key_with_valid_config_key(self, prepare_nrtm4_test,
         )
         with pytest.raises(NRTM4ClientError) as exc:
             NRTM4Client("TEST", mock_dh).run_client()
-        assert "use the 'irrdctl" in str(exc)
+        assert "is valid for" in str(exc)
 
     def test_uses_current_db_key(self, prepare_nrtm4_test, config_override):
         config_override(