diff --git a/irrd/storage/alembic/versions/a75d0bd6c9f2_add_api_tokens.py b/irrd/storage/alembic/versions/3c93349ab07e_add_api_tokens.py similarity index 91% rename from irrd/storage/alembic/versions/a75d0bd6c9f2_add_api_tokens.py rename to irrd/storage/alembic/versions/3c93349ab07e_add_api_tokens.py index 1a7ef0a21..dc372f640 100644 --- a/irrd/storage/alembic/versions/a75d0bd6c9f2_add_api_tokens.py +++ b/irrd/storage/alembic/versions/3c93349ab07e_add_api_tokens.py @@ -1,8 +1,8 @@ """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 @@ -10,7 +10,7 @@ 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), diff --git a/irrd/storage/models.py b/irrd/storage/models.py index bf8815164..2343ac554 100644 --- a/irrd/storage/models.py +++ b/irrd/storage/models.py @@ -400,7 +400,7 @@ 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) @@ -408,7 +408,10 @@ 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.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 diff --git a/irrd/webui/endpoints_mntners.py b/irrd/webui/endpoints_mntners.py index 45781a38b..bb738ed5c 100644 --- a/irrd/webui/endpoints_mntners.py +++ b/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,23 @@ 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): + def validate_ip_restriction(self, field): if not field.data: - field.data = None + self.ip_restriction_parsed = None + return + try: + self.ip_restriction_parsed = [str(IP(data)) for data in field.data.split(",")] + field.data = ",".join(self.ip_restriction_parsed) + except ValueError: + raise wtforms.ValidationError("Invalid IP or CIDR ranges.") @csrf_protect @@ -73,7 +83,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.ip_restriction_parsed, enabled_webapi=form.data["enabled_webapi"], enabled_email=form.data["enabled_email"], ) @@ -106,7 +116,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( @@ -116,7 +132,7 @@ async def api_token_edit(request: Request, session_provider: ORMSessionProvider) ) api_token.name = form.data["name"] - api_token.ip_restriction = form.data["ip_restriction"] + api_token.ip_restriction = form.ip_restriction_parsed api_token.enabled_webapi = form.data["enabled_webapi"] api_token.enabled_email = form.data["enabled_email"] session_provider.session.add(api_token) diff --git a/irrd/webui/templates/user_permissions.html b/irrd/webui/templates/user_permissions.html index c407b88fd..c65b0bc75 100644 --- a/irrd/webui/templates/user_permissions.html +++ b/irrd/webui/templates/user_permissions.html @@ -71,7 +71,7 @@
API tokens for {{ permission.mntner.rpsl_mntner_pk }}
{{ api_token.name }} {{ api_token.creator.name }} - {{ api_token.ip_restriction }} + {{ api_token.ip_restriction_display() }} {{ "Yes" if api_token.enabled_webapi else "No" }} {{ "Yes" if api_token.enabled_email else "No" }} diff --git a/irrd/webui/tests/test_endpoints_mntners.py b/irrd/webui/tests/test_endpoints_mntners.py index e8addbf71..73e07abd9 100644 --- a/irrd/webui/tests/test_endpoints_mntners.py +++ b/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)