Fix #949 - Add support for inline PGP signatures in RPSL update form (#…

…976)

Also detects correct fingerprint when using PGP subkeys

docs/releases/4.5.0.rst

@@ -57,6 +57,8 @@ as of 7 October 2024 and therefore no longer supported.
 
 Other changes
 -------------
+* The RPSL update submission page, on `/rpsl/update/`, now also accepts
+  inline PGP signed messages for mntner authentication.
 * The ``sources.{name}.object_class_filter`` setting can now also be used
   to restrict permitted changes to authoritative objects.
 * The setting ``sources.{name}.authoritative_retain_last_modified`` was

irrd/server/http/app.py

@@ -145,7 +145,8 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
 def set_middleware(app):
     testing = os.environ.get("TESTING", False)
-    if testing:
+    csrf_disabled = testing and not getattr(app, "force_csrf_in_testing", False)
+    if csrf_disabled:
         logger.info("Running in testing mode, disabling CSRF.")
     app.user_middleware = [
         # Use asgi-log to work around https://github.com/encode/uvicorn/issues/1384
@@ -157,7 +158,9 @@ def set_middleware(app):
         Middleware(MemoryTrimMiddleware),
         Middleware(SessionMiddleware, secret_key=secret_key_derive("web.session_middleware")),
         Middleware(
-            CSRFProtectMiddleware, csrf_secret=secret_key_derive("web.csrf_middleware"), enabled=not testing
+            CSRFProtectMiddleware,
+            csrf_secret=secret_key_derive("web.csrf_middleware"),
+            enabled=not csrf_disabled,
         ),
         auth_middleware,
     ]

irrd/utils/pgp.py

@@ -78,6 +78,9 @@ def validate_pgp_signature(
     log_message = result.stderr.replace("\n", " -- ").replace("gpg:                ", "")
     logger.info(f"checked PGP signature, response: {log_message}")
     if result.valid and result.key_status is None:
-        logger.info(f"Found valid PGP signature, fingerprint {result.fingerprint}")
-        return new_message, result.fingerprint
+        logger.info(
+            f"Found valid PGP signature, pubkey fingerprint {result.pubkey_fingerprint}, fingerprint"
+            f" {result.fingerprint}"
+        )
+        return new_message, result.pubkey_fingerprint
     return None, None

irrd/webui/endpoints.py

@@ -1,9 +1,10 @@
+import functools
 from collections import defaultdict
 
 from asgiref.sync import sync_to_async
 from starlette.requests import Request
 from starlette.responses import Response
-from starlette_wtf import csrf_protect, csrf_token
+from starlette_wtf import CSRFError, csrf_protect, csrf_token
 
 from irrd import META_KEY_HTTP_CLIENT_IP
 from irrd.conf import get_setting
@@ -18,6 +19,7 @@
 from irrd.storage.orm_provider import ORMSessionProvider, session_provider_manager
 from irrd.storage.queries import RPSLDatabaseQuery
 from irrd.updates.handler import ChangeSubmissionHandler
+from irrd.utils.pgp import validate_pgp_signature
 from irrd.webui.auth.decorators import authentication_required, mark_user_mfa_incomplete
 from irrd.webui.helpers import filter_auth_hash_non_mntner
 from irrd.webui.rendering import template_context_render
@@ -101,19 +103,44 @@ async def rpsl_detail(request: Request, user_mfa_incomplete: bool, session_provi
         )
 
 
-@csrf_protect
+def optional_csrf_protect(func):
+    """
+    The RPSL update endpoint is special re CSRF: while not entirely supported,
+    users sometimes call this through curl, and miss the CSRF token.
+    It does need CSRF protection, because a logged in user gets additional
+    mntner access. Therefore, this decorator tries to validate CSRF,
+    and if not, tells the endpoint, which will then ignore user session
+    info and only look at the post data.
+    """
+
+    @functools.wraps(func)
+    async def wrapper(*args, **kwargs):
+        try:
+            csrf_protected_func = csrf_protect(func)
+            return await csrf_protected_func(*args, csrf_protected=True, **kwargs)
+        except CSRFError:
+            return await func(*args, csrf_protected=False, **kwargs)
+
+    return wrapper
+
+
+@optional_csrf_protect
 @mark_user_mfa_incomplete
 @session_provider_manager
 async def rpsl_update(
-    request: Request, user_mfa_incomplete: bool, session_provider: ORMSessionProvider
+    request: Request, user_mfa_incomplete: bool, session_provider: ORMSessionProvider, csrf_protected: bool
 ) -> Response:
     """
     Web form for submitting RPSL updates.
     Essentially a wrapper around the same submission handlers as emails,
     but with pre-authentication through the logged in user or override.
     Can also be used anonymously.
     """
-    active_user = request.auth.user if request.auth.is_authenticated and not user_mfa_incomplete else None
+    active_user = (
+        request.auth.user
+        if request.auth.is_authenticated and not user_mfa_incomplete and csrf_protected
+        else None
+    )
     mntner_perms = defaultdict(list)
     if active_user:
         for mntner in request.auth.user.mntners_user_management:
@@ -147,6 +174,7 @@ async def rpsl_update(
 
     elif request.method == "POST":
         form_data = await request.form()
+        submission = form_data.get("data", form_data.get("DATA"))
         request_meta = {
             META_KEY_HTTP_CLIENT_IP: request.client.host if request.client else "",
             "HTTP-User-Agent": request.headers.get("User-Agent"),
@@ -160,8 +188,10 @@ async def rpsl_update(
         # and therefore needs wrapping in a thread
         @sync_to_async
         def save():
+            signed_submission, pgp_fingerprint = validate_pgp_signature(submission)
             return ChangeSubmissionHandler().load_text_blob(
-                object_texts_blob=form_data["data"],
+                object_texts_blob=signed_submission if signed_submission else submission,
+                pgp_fingerprint=pgp_fingerprint,
                 origin=AuthoritativeChangeOrigin.webui,
                 request_meta=request_meta,
                 internal_authenticated_user=active_user,
@@ -172,7 +202,7 @@ def save():
             "rpsl_form.html",
             request,
             {
-                "existing_data": form_data["data"],
+                "existing_data": submission,
                 "status": handler.status(),
                 "report": handler.submitter_report_human(),
                 "mntner_perms": mntner_perms,

irrd/webui/templates/rpsl_form.html

@@ -28,8 +28,8 @@ <h2>Change/create/delete object(s){% if status %}: {{ status }}{% endif %}</h2>
                     This form is identical to email submissions, which means you
                     can use the pseudo-attributes <code>delete</code> for deletions
                     or <code>password</code> for password authentication.
-                    PGP is not supported.
-                    See the <a href="TODO">IRRD documentation</a> for more details.
+                    PGP inline signatures are supported.
+                    See the <a href="https://irrd.readthedocs.io/">IRRD documentation</a> for more details.
                 </p>
                 {% if user and user.override %}
                     <p>

irrd/webui/tests/test_endpoints.py

@@ -3,12 +3,14 @@
 from unittest.mock import create_autospec
 
 import pytest
+from starlette.testclient import TestClient
 
 from irrd.updates.handler import ChangeSubmissionHandler
 from irrd.utils.rpsl_samples import SAMPLE_MNTNER
 from irrd.webui import datetime_format
 
 from ...rpsl.rpsl_objects import rpsl_object_from_text
+from ...server.http.app import app
 from ...storage.database_handler import DatabaseHandler
 from ...storage.models import JournalEntryOrigin
 from ...updates.parser_state import UpdateRequestType
@@ -141,7 +143,11 @@ class TestRpslUpdateNoInitial(WebRequestTest):
     requires_mfa = False
 
     def test_valid_mntner_logged_in_mfa_complete_no_user_management(
-        self, irrd_db_session_with_user, test_client, mock_change_submission_handler
+        self,
+        irrd_db_session_with_user,
+        test_client,
+        mock_change_submission_handler,
+        tmp_gpg_dir,
     ):
         session_provider, user = irrd_db_session_with_user
         self._login(test_client, user)
@@ -154,7 +160,11 @@ def test_valid_mntner_logged_in_mfa_complete_no_user_management(
         assert "(you can not update this mntner itself)" in response.text
 
     def test_valid_mntner_logged_in_mfa_complete_user_management(
-        self, irrd_db_session_with_user, test_client, mock_change_submission_handler
+        self,
+        irrd_db_session_with_user,
+        test_client,
+        mock_change_submission_handler,
+        tmp_gpg_dir,
     ):
         session_provider, user = irrd_db_session_with_user
         self._login(test_client, user)
@@ -173,8 +183,35 @@ def test_valid_mntner_logged_in_mfa_complete_user_management(
         assert mock_handler_kwargs["object_texts_blob"] == SAMPLE_MNTNER
         assert mock_handler_kwargs["internal_authenticated_user"].pk == user.pk
 
+    def test_valid_mntner_logged_in_mfa_complete_user_management_no_csrf(
+        self,
+        irrd_db_session_with_user,
+        test_client,
+        mock_change_submission_handler,
+        tmp_gpg_dir,
+    ):
+        session_provider, user = irrd_db_session_with_user
+        self._login(test_client, user)
+        self._verify_mfa(test_client)
+        create_permission(session_provider, user)
+
+        app.force_csrf_in_testing = True
+        with TestClient(app, cookies=test_client.cookies) as client_csrf:
+            app.force_csrf_in_testing = False
+            client_csrf.cookies = test_client.cookies
+            response = client_csrf.post(self.url, data={"data": SAMPLE_MNTNER})
+            assert response.status_code == 200
+            assert mock_change_submission_handler.mock_calls[1][0] == "().load_text_blob"
+            mock_handler_kwargs = mock_change_submission_handler.mock_calls[1][2]
+            assert mock_handler_kwargs["object_texts_blob"] == SAMPLE_MNTNER
+            assert mock_handler_kwargs["internal_authenticated_user"] is None
+
     def test_valid_mntner_logged_in_mfa_incomplete_user_management(
-        self, irrd_db_session_with_user, test_client, mock_change_submission_handler
+        self,
+        irrd_db_session_with_user,
+        test_client,
+        mock_change_submission_handler,
+        tmp_gpg_dir,
     ):
         session_provider, user = irrd_db_session_with_user
         self._login(test_client, user)
@@ -192,7 +229,11 @@ def test_valid_mntner_logged_in_mfa_incomplete_user_management(
         assert mock_handler_kwargs["internal_authenticated_user"] is None
 
     def test_valid_mntner_not_logged_in(
-        self, irrd_db_session_with_user, test_client, mock_change_submission_handler
+        self,
+        irrd_db_session_with_user,
+        test_client,
+        mock_change_submission_handler,
+        tmp_gpg_dir,
     ):
         session_provider, user = irrd_db_session_with_user
         response = test_client.get(self.url)
@@ -213,7 +254,10 @@ class TestRpslUpdateWithInitial(WebRequestTest):
     requires_mfa = False
 
     def test_valid_mntner_logged_in_mfa_complete_no_user_management(
-        self, irrd_db_session_with_user, test_client
+        self,
+        irrd_db_session_with_user,
+        test_client,
+        tmp_gpg_dir,
     ):
         session_provider, user = irrd_db_session_with_user
         self._login(test_client, user)
@@ -227,7 +271,10 @@ def test_valid_mntner_logged_in_mfa_complete_no_user_management(
         assert "DUMMYVALUE" in response.text.upper()
 
     def test_valid_mntner_logged_in_mfa_complete_user_management(
-        self, irrd_db_session_with_user, test_client
+        self,
+        irrd_db_session_with_user,
+        test_client,
+        tmp_gpg_dir,
     ):
         session_provider, user = irrd_db_session_with_user
         self._login(test_client, user)
@@ -241,7 +288,10 @@ def test_valid_mntner_logged_in_mfa_complete_user_management(
         assert "DUMMYVALUE" not in response.text.upper()
 
     def test_valid_mntner_logged_in_mfa_incomplete_user_management(
-        self, irrd_db_session_with_user, test_client
+        self,
+        irrd_db_session_with_user,
+        test_client,
+        tmp_gpg_dir,
     ):
         session_provider, user = irrd_db_session_with_user
         self._login(test_client, user)
@@ -252,7 +302,7 @@ def test_valid_mntner_logged_in_mfa_incomplete_user_management(
         assert "TEST-MNT" in response.text
         assert "DUMMYVALUE" in response.text.upper()
 
-    def test_valid_mntner_not_logged_in(self, irrd_db_session_with_user, test_client):
+    def test_valid_mntner_not_logged_in(self, irrd_db_session_with_user, test_client, tmp_gpg_dir):
         session_provider, user = irrd_db_session_with_user
         response = test_client.get(self.url)
         assert response.status_code == 200