Validate Prefect pagination URLs (#23354)

* Fix Prefect event pagination host validation

* Add changelog

* Rename changelog

* Change changelog

Author: lucia-sb

prefect/changelog.d/23354.fixed

@@ -0,0 +1 @@
+Adds validation for event pagination URLs.

prefect/datadog_checks/prefect/check.py

@@ -8,6 +8,7 @@
 from functools import cached_property
 from json import JSONDecodeError
 from typing import TYPE_CHECKING, Any, Literal
+from urllib.parse import urlparse
 
 from requests.exceptions import ConnectionError, HTTPError, InvalidURL, Timeout
 
@@ -667,6 +668,9 @@ class PrefectClient:
     def __init__(self, url: str, http: RequestsWrapper, log: CheckLoggingAdapter):
         self.http_exceptions = (HTTPError, InvalidURL, ConnectionError, Timeout, JSONDecodeError)
         self.url = url
+        parsed_url = urlparse(url)
+        self.base_url_scheme = parsed_url.scheme
+        self.base_url_netloc = parsed_url.netloc
         self.http = http
         self.log = log
 
@@ -689,6 +693,11 @@ def post(self, endpoint: str, payload: dict | None = None) -> Any:
         except self.http_exceptions:
             raise
 
+    def check_pagination_url(self, next_page: str):
+        parsed_next_page = urlparse(next_page)
+        if parsed_next_page.scheme != self.base_url_scheme or parsed_next_page.netloc != self.base_url_netloc:
+            raise InvalidURL(f'Invalid next_page URL with unexpected host: {next_page}')
+
     def paginate_filter(self, endpoint: str, payload: dict | None = None) -> list[dict]:
         """Implements pagination for /filter endpoints using limit/offset loop."""
         payload = dict(payload) if payload else {}
@@ -728,9 +737,10 @@ def paginate_events(self, endpoint: str, payload: dict | None = None) -> list[di
             events.extend(response.get("events", []))
             if not response.get("next_page"):
                 break
-
             try:
-                response = self.get(response.get("next_page"), pagination=True)
+                next_page = response.get("next_page")
+                self.check_pagination_url(next_page)
+                response = self.get(next_page, pagination=True)
             except self.http_exceptions as e:
                 self.log.error("Could not collect next page of events: %s, data is incomplete", e)
                 return events

prefect/tests/test_unit.py

@@ -1,11 +1,14 @@
 from datetime import datetime, timezone
 from typing import Callable, NamedTuple
+from unittest.mock import Mock
 
 import pytest
+from requests.exceptions import InvalidURL
 
 from datadog_checks.base.stubs.aggregator import AggregatorStub
 from datadog_checks.dev.utils import get_metadata_metrics
 from datadog_checks.prefect import PrefectCheck
+from datadog_checks.prefect.check import PrefectClient
 
 WP1_TAGS = ["work_pool_id:wp-1", "work_pool_name:default-pool", "work_pool_type:process"]
 WP2_TAGS = ["work_pool_id:wp-2", "work_pool_name:paused-pool", "work_pool_type:docker"]
@@ -600,6 +603,62 @@ class EventCase(NamedTuple):
 pytestmark = [pytest.mark.usefixtures("mock_prefect_client"), pytest.mark.unit]
 
 
+class MockResponse:
+    def __init__(self, payload: dict):
+        self.payload = payload
+
+    def raise_for_status(self):
+        return None
+
+    def json(self) -> dict:
+        return self.payload
+
+
+def test_paginate_events_rejects_external_next_page():
+    http = Mock()
+    http.post.return_value = MockResponse(
+        {
+            "events": [{"id": "event-1"}],
+            "next_page": "http://attacker.example/evil",
+        }
+    )
+    log = Mock()
+    client = PrefectClient("http://prefect.local/api", http, log)
+
+    events = client.paginate_events("/events/filter", {})
+
+    assert events == [{"id": "event-1"}]
+    http.get.assert_not_called()
+    log.error.assert_called_once()
+    args = log.error.call_args[0]
+    assert args[0] == "Could not collect next page of events: %s, data is incomplete"
+    assert isinstance(args[1], InvalidURL)
+    assert str(args[1]) == "Invalid next_page URL with unexpected host: http://attacker.example/evil"
+
+
+def test_paginate_events_allows_same_host_absolute_next_page():
+    http = Mock()
+    http.post.return_value = MockResponse(
+        {
+            "events": [{"id": "event-1"}],
+            "next_page": "http://prefect.local/api/events/filter?page=2",
+        }
+    )
+    http.get.return_value = MockResponse(
+        {
+            "events": [{"id": "event-2"}],
+            "next_page": None,
+        }
+    )
+    log = Mock()
+    client = PrefectClient("http://prefect.local/api", http, log)
+
+    events = client.paginate_events("/events/filter", {})
+
+    assert events == [{"id": "event-1"}, {"id": "event-2"}]
+    http.get.assert_called_once_with("http://prefect.local/api/events/filter?page=2")
+
+
 @pytest.fixture()
 def ready_check(check: PrefectCheck, dd_run_check: Callable, aggregator, mocker) -> PrefectCheck:
     mocker.patch(