Improve api_token ip_restriction storage

irrd/storage/alembic/versions/a75d0bd6c9f2_add_api_tokens.py → irrd/storage/alembic/versions/3c93349ab07e_add_api_tokens.py

@@ -1,16 +1,16 @@
 """add_api_tokens
 
-Revision ID: a75d0bd6c9f2
+Revision ID: 3c93349ab07e
 Revises: 5bbbc2989aa6
-Create Date: 2023-05-04 21:09:08.591476
+Create Date: 2023-05-05 12:48:50.096432
 
 """
 import sqlalchemy as sa
 from alembic import op
 from sqlalchemy.dialects import postgresql
 
 # revision identifiers, used by Alembic.
-revision = "a75d0bd6c9f2"
+revision = "3c93349ab07e"
 down_revision = "5bbbc2989aa6"
 branch_labels = None
 depends_on = None
@@ -28,7 +28,7 @@ def upgrade():
         sa.Column("name", sa.String(), nullable=False),
         sa.Column("creator_id", postgresql.UUID(), nullable=True),
         sa.Column("mntner_id", postgresql.UUID(), nullable=True),
-        sa.Column("ip_restriction", postgresql.CIDR(), nullable=True),
+        sa.Column("ip_restriction", postgresql.ARRAY(postgresql.CIDR()), nullable=True),
         sa.Column("enabled_webapi", sa.Boolean(), nullable=False),
         sa.Column("enabled_email", sa.Boolean(), nullable=False),
         sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),

irrd/storage/models.py

@@ -400,15 +400,18 @@ class AuthApiToken(Base):  # type: ignore
         backref=sa.orm.backref("api_tokens"),
     )
 
-    ip_restriction = sa.Column(pg.CIDR, nullable=True)
+    ip_restriction = sa.Column(pg.ARRAY(pg.CIDR), nullable=True)
     enabled_webapi = sa.Column(sa.Boolean(), default=True, nullable=False)
     enabled_email = sa.Column(sa.Boolean(), default=True, nullable=False)
 
     created = sa.Column(sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False)
     updated = sa.Column(sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False)
 
     def __repr__(self):
-        return f"<{self.token}/{self.name}/{self.maintainer}"
+        return f"<{self.token}/{self.name}/{self.mntner.rpsl_mntner_pk}"
+
+    def ip_restriction_display(self) -> str:
+        return ", ".join(self.ip_restriction) if self.ip_restriction else ""
 
 
 class AuthMntner(Base):  # type: ignore

irrd/webui/endpoints_mntners.py

@@ -4,6 +4,7 @@
 from typing import Optional
 
 import wtforms
+from IPy import IP
 from starlette.requests import Request
 from starlette.responses import RedirectResponse, Response
 from starlette_wtf import StarletteForm, csrf_protect, csrf_token
@@ -35,14 +36,22 @@ class ApiTokenForm(StarletteForm):
         description="For your own reference.",
     )
     # TODO: cidr validator
-    ip_restriction = wtforms.StringField("Restrict to IP range (CIDR, optional)")
+    ip_restriction = wtforms.StringField(
+        "Restrict to IP/CIDR (optional)",
+        description="You can include multiple IP/CIDRs, separated by commas.",
+    )
     enabled_webapi = wtforms.BooleanField("Enable this token for HTTPS API submissions")
     enabled_email = wtforms.BooleanField("Enable this token for email submissions")
     submit = wtforms.SubmitField("Save API token")
 
     def validate_ip_restriction(form, field):
         if not field.data:
             field.data = None
+            return
+        try:
+            field.data = ",".join([str(IP(data)) for data in field.data.split(",")])
+        except ValueError:
+            raise wtforms.ValidationError("Invalid IP or CIDR ranges.")
 
 
 @csrf_protect
@@ -73,7 +82,7 @@ async def api_token_add(request: Request, session_provider: ORMSessionProvider)
         mntner_id=str(mntner.pk),
         creator_id=str(request.auth.user.pk),
         name=form.data["name"],
-        ip_restriction=form.data["ip_restriction"],
+        ip_restriction=form.data["ip_restriction"].split(","),
         enabled_webapi=form.data["enabled_webapi"],
         enabled_email=form.data["enabled_email"],
     )
@@ -106,7 +115,13 @@ async def api_token_edit(request: Request, session_provider: ORMSessionProvider)
     if not api_token:
         return Response(status_code=404)
 
-    form = await ApiTokenForm.from_formdata(request=request, obj=api_token)
+    form = await ApiTokenForm.from_formdata(
+        request=request,
+        name=api_token.name,
+        enabled_webapi=api_token.enabled_webapi,
+        enabled_email=api_token.enabled_email,
+        ip_restriction=api_token.ip_restriction_display(),
+    )
     if not form.is_submitted() or not await form.validate():
         form_html = render_form(form)
         return template_context_render(

irrd/webui/templates/user_permissions.html

@@ -71,7 +71,7 @@ <h5>API tokens for {{ permission.mntner.rpsl_mntner_pk }}</h5>
                     <tr>
                         <td>{{ api_token.name }}</td>
                         <td>{{ api_token.creator.name }}</td>
-                        <td>{{ api_token.ip_restriction }}</td>
+                        <td>{{ api_token.ip_restriction_display() }}</td>
                         <td>{{ "Yes" if api_token.enabled_webapi else "No" }}</td>
                         <td>{{ "Yes" if api_token.enabled_email else "No" }}</td>
                         <td>

irrd/webui/tests/test_endpoints_mntners.py

@@ -38,7 +38,7 @@ def test_valid_new_token(self, test_client_with_smtp, irrd_db_session_with_user)
         api_token_name = "token name"
         response = test_client.post(
             self.url,
-            data={"name": api_token_name, "enabled_webapi": "1"},
+            data={"name": api_token_name, "enabled_webapi": "1", "ip_restriction": " 192.0.2.1 ,192.0.02.2"},
             follow_redirects=False,
         )
         assert response.status_code == 302
@@ -48,10 +48,28 @@ def test_valid_new_token(self, test_client_with_smtp, irrd_db_session_with_user)
         assert new_api_token.creator == user
         assert new_api_token.name == api_token_name
         assert new_api_token.enabled_webapi
+        assert new_api_token.ip_restriction == ["192.0.2.1/32", "192.0.2.2/32"]
         assert not new_api_token.enabled_email
         assert len(smtpd.messages) == 3
         assert new_api_token.name in smtpd.messages[1].as_string()
 
+    def test_invalid_cidr_range(self, test_client, irrd_db_session_with_user):
+        session_provider, user = irrd_db_session_with_user
+        self.pre_login(session_provider, user)
+        self._login_if_needed(test_client, user)
+
+        api_token_name = "token name"
+        response = test_client.post(
+            self.url,
+            data={"name": api_token_name, "ip_restriction": "192.0.2.1.1"},
+            follow_redirects=False,
+        )
+        assert response.status_code == 200
+
+        new_api_token = session_provider.run_sync(session_provider.session.query(AuthApiToken).one)
+        assert not new_api_token
+        assert "Invalid IP" in response.text
+
     def test_object_not_exists(self, test_client, irrd_db_session_with_user):
         session_provider, user = irrd_db_session_with_user
         self._login_if_needed(test_client, user)