validate API key

irrd/storage/models.py

@@ -1,6 +1,8 @@
 import enum
+from typing import List, Optional
 
 import sqlalchemy as sa
+from IPy import IP
 from sqlalchemy.dialects import postgresql as pg
 from sqlalchemy.ext.declarative import declarative_base, declared_attr
 from sqlalchemy.orm import relationship
@@ -416,7 +418,27 @@ class AuthApiToken(Base):  # type: ignore
     updated = sa.Column(sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False)
 
     def __repr__(self):
-        return f"<{self.token}/{self.name}/{self.mntner.rpsl_mntner_pk}"
+        return f"<{self.token}/{self.name}/{self.mntner.rpsl_mntner_pk if self.mntner else None}>"
+
+    def ip_restriction_parsed(self) -> Optional[List[IP]]:
+        if not self.ip_restriction:
+            return None
+        return [IP(ip) for ip in self.ip_restriction.split(",")]
+
+    def valid_for(self, origin: AuthoritativeChangeOrigin, remote_ip: IP):
+        if not any(
+            [
+                self.enabled_webapi and origin == AuthoritativeChangeOrigin.webapi,
+                self.enabled_email and origin == AuthoritativeChangeOrigin.email,
+            ]
+        ):
+            return False
+        if self.ip_restriction:
+            for ip in self.ip_restriction.split(","):
+                if remote_ip in IP(ip):
+                    return True
+            return False
+        return True
 
 
 class AuthMntner(Base):  # type: ignore

irrd/updates/tests/test_parser.py

@@ -490,15 +490,13 @@ def test_check_auth_valid_update_mntner(self, prepare_mocks):
         auth_validator = AuthValidator(mock_dh)
 
         result_inetnum = parse_change_requests(
-            SAMPLE_INETNUM + "password: crypt-password\napi-key: key",
+            SAMPLE_INETNUM + "password: crypt-password",
             mock_dh,
             auth_validator,
             reference_validator,
         )[0]
         assert result_inetnum._check_auth()
         assert not result_inetnum.error_messages
-        assert auth_validator.passwords == ["crypt-password"]
-        assert auth_validator.api_keys == ["key"]
 
         assert flatten_mock_calls(mock_dq) == [
             ["sources", (["TEST"],), {}],
@@ -663,11 +661,15 @@ def test_check_auth_invalid_update_mntner_submits_new_object_with_dummy_hash_mul
         )
         data = data.replace("$1$fgW84Y9r$kKEn9MUq8PChNKpQhO6BM.", PASSWORD_HASH_DUMMY_VALUE)
         result_mntner = parse_change_requests(
-            data + "password: md5-password\npassword: other-password",
+            data + "password: md5-password\npassword: other-password\napi-key: key",
             mock_dh,
             auth_validator,
             reference_validator,
         )[0]
+        # This also tests whether API keys are passed to the validator.
+        assert auth_validator.passwords == ["md5-password", "other-password"]
+        assert auth_validator.api_keys == ["key"]
+
         auth_validator.pre_approve([result_mntner.rpsl_obj_new])
         result_mntner._check_auth()
         assert not result_mntner.is_valid()

irrd/updates/tests/test_validators.py

@@ -23,7 +23,13 @@
 from irrd.utils.text import remove_auth_hashes
 from irrd.vendor.mock_alchemy.mocking import UnifiedAlchemyMagicMock
 
-from ...storage.models import AuthMntner, AuthUser
+from ...storage.models import (
+    AuthApiToken,
+    AuthMntner,
+    AuthoritativeChangeOrigin,
+    AuthUser,
+)
+from ...utils.factories import AuthApiTokenFactory
 from ..validators import AuthValidator, RulesValidator
 
 VALID_PW = "override-password"
@@ -46,7 +52,7 @@ def prepare_mocks(self, monkeypatch, config_override):
             }
         )
 
-        validator = AuthValidator(mock_dh, None)
+        validator = AuthValidator(mock_dh, AuthoritativeChangeOrigin.webapi)
         yield validator, mock_dq, mock_dh
 
     def test_override_valid(self, prepare_mocks, config_override):
@@ -142,6 +148,48 @@ def test_valid_new_person(self, prepare_mocks):
             ["rpsl_pks", ({"TEST-MNT"},), {}],
         ]
 
+    def test_valid_new_person_api_key(self, prepare_mocks, monkeypatch):
+        validator, mock_dq, mock_dh = prepare_mocks
+        person = rpsl_object_from_text(SAMPLE_PERSON)
+        mock_sa_session = UnifiedAlchemyMagicMock(
+            data=[
+                (
+                    [
+                        mock.call.query(AuthMntner),
+                        mock.call.filter(
+                            AuthMntner.rpsl_mntner_pk == "TEST-MNT", AuthMntner.rpsl_mntner_source == "TEST"
+                        ),
+                    ],
+                    [AuthMntner(rpsl_mntner_pk="TEST-MNT")],
+                )
+            ]
+        )
+        mock_api_key = AuthApiTokenFactory.build()
+        print(mock_api_key)
+        mock_sa_session.all = lambda: [mock_api_key]
+        monkeypatch.setattr("irrd.updates.validators.saorm.Session", lambda bind: mock_sa_session)
+
+        mock_dh._connection = None
+        mock_dh.execute_query = lambda q: [
+            {"object_class": "mntner", "object_text": SAMPLE_MNTNER},
+        ]
+
+        validator.api_keys = ["key"]
+        result = validator.process_auth(person, None)
+        assert result.is_valid(), result.error_messages
+        assert not result.used_override
+        assert result.mntners_notify[0].pk() == "TEST-MNT"
+
+        mock_sa_session.filter.assert_has_calls(
+            [
+                mock.call(
+                    AuthMntner.rpsl_mntner_pk == "TEST-MNT",
+                    AuthMntner.rpsl_mntner_source == "TEST",
+                    AuthApiToken.token.in_(["key"]),
+                ),
+            ]
+        )
+
     def test_existing_person_mntner_change(self, prepare_mocks):
         validator, mock_dq, mock_dh = prepare_mocks
         # TEST-MNT is in both maintainers

irrd/updates/validators.py

@@ -12,7 +12,12 @@
 from irrd.rpsl.parser import RPSLObject
 from irrd.rpsl.rpsl_objects import RPSLMntner, RPSLSet, rpsl_object_from_text
 from irrd.storage.database_handler import DatabaseHandler
-from irrd.storage.models import AuthMntner, AuthoritativeChangeOrigin, AuthUser
+from irrd.storage.models import (
+    AuthApiToken,
+    AuthMntner,
+    AuthoritativeChangeOrigin,
+    AuthUser,
+)
 from irrd.storage.queries import RPSLDatabaseQuery, RPSLDatabaseSuspendedQuery
 
 from .parser_state import RPSLSetAutnumAuthenticationMode, UpdateRequestType
@@ -280,6 +285,7 @@ def process_auth(
                 [
                     rpsl_obj_new.verify_auth(self.passwords, self.keycert_obj_pk),
                     self._mntner_matches_internal_auth(rpsl_obj_new, rpsl_obj_new.pk(), source),
+                    # API keys are not checked here, as they can never be used on RPSLMntner
                 ]
             ):
                 result.error_messages.add("Authorisation failed for the auth methods on this mntner object.")
@@ -339,7 +345,8 @@ def _check_mntners(
 
         for mntner_name in mntner_pk_list:
             matches_internal_auth = self._mntner_matches_internal_auth(rpsl_obj_new, mntner_name, source)
-            if mntner_name in self._pre_approved or matches_internal_auth:
+            matches_api_key = self._mntner_matches_api_key(rpsl_obj_new, mntner_name, source)
+            if mntner_name in self._pre_approved or matches_internal_auth or matches_api_key:
                 return True, mntner_objs
 
         for mntner_obj in mntner_objs:
@@ -355,12 +362,38 @@ def _mntner_matches_internal_auth(self, rpsl_obj_new: RPSLObject, rpsl_pk: str,
             user_mntner_set = self._internal_authenticated_user.mntners_user_management
         else:
             user_mntner_set = self._internal_authenticated_user.mntners
-        return any(
+        match = any(
             [
                 rpsl_pk == mntner.rpsl_mntner_pk and source == mntner.rpsl_mntner_source
                 for mntner in user_mntner_set
             ]
         )
+        if match:
+            logger.info(
+                f"Authenticated through internally authenticated user {self._internal_authenticated_user}"
+            )
+        return match
+
+    def _mntner_matches_api_key(self, rpsl_obj_new: RPSLObject, rpsl_pk: str, source: str) -> bool:
+        if not self.api_keys or isinstance(rpsl_obj_new, RPSLMntner):
+            return False
+
+        session = saorm.Session(bind=self.database_handler._connection)
+        query = (
+            session.query(AuthApiToken)
+            .join(AuthMntner)
+            .filter(
+                AuthMntner.rpsl_mntner_pk == rpsl_pk,
+                AuthMntner.rpsl_mntner_source == source,
+                AuthApiToken.token.in_(self.api_keys),
+            )
+        )
+        for api_token in query.all():
+            if api_token.valid_for(self.origin, self.remote_ip):
+                logger.info(f"Authenticated through API token {api_token.pk} on mntner {rpsl_pk}")
+                return True
+
+        return False
 
     def _generate_failure_message(
         self,