Merge branch '2.x' into port-pr-15510

requirements-client.txt

@@ -1,4 +1,4 @@
-anyio >= 4.4.0, < 5.0.0
+anyio >= 4.4.0, < 4.6.1
 asgi-lifespan >= 1.0, < 3.0
 cachetools >= 5.3, < 6.0
 cloudpickle >= 2.0, < 4.0

requirements.txt

@@ -3,7 +3,7 @@
 aiosqlite >= 0.17.0, < 1.0.0
 alembic >= 1.7.5, < 2.0.0
 apprise >= 1.8.0, < 2.0.0
-asyncpg >= 0.23, < 1.0.0
+asyncpg >= 0.23, < 0.30.0
 click >= 8.0, < 8.2
 cryptography >= 36.0.1
 dateparser >= 1.1.1, < 2.0.0
@@ -15,5 +15,5 @@ humanize >= 4.9.0, < 5.0.0
 kubernetes >= 24.2.0, < 32.0.0
 pytz >= 2021.1, < 2025
 readchar >= 4.0.0, < 5.0.0
-sqlalchemy[asyncio] >= 1.4.22, != 1.4.33, < 3.0.0
+sqlalchemy[asyncio] >= 1.4.22, != 1.4.33, < 2.0.36
 typer >= 0.12.0, != 0.12.2, < 0.13.0

setup.cfg

@@ -56,6 +56,8 @@ filterwarnings =
     ignore:The `RayTaskRunner` has moved:DeprecationWarning
     ignore:The `DaskTaskRunner` has moved:DeprecationWarning
     ignore:`DeploymentSpec` has been replaced by `Deployment`:DeprecationWarning
+    # this warning comes from python-multipart
+    ignore:Please use `import python_multipart` instead:PendingDeprecationWarning
     # This warning is raised on Windows by Python internals
     ignore:the imp module is deprecated:DeprecationWarning
     # Dockerpy is behind on this one

src/prefect/blocks/notifications.py

@@ -4,6 +4,7 @@
 
 from prefect._internal.pydantic import HAS_PYDANTIC_V2
 from prefect.logging import LogEavesdropper
+from prefect.utilities.urls import validate_restricted_url
 
 if HAS_PYDANTIC_V2:
     from pydantic.v1 import AnyHttpUrl, Field, SecretStr
@@ -88,6 +89,26 @@ class AppriseNotificationBlock(AbstractAppriseNotificationBlock, ABC):
         description="Incoming webhook URL used to send notifications.",
         examples=["https://hooks.example.com/XXX"],
     )
+    allow_private_urls: bool = Field(
+        default=True,
+        description="Whether to allow notifications to private URLs. Defaults to True.",
+    )
+
+    @sync_compatible
+    async def notify(
+        self,
+        body: str,
+        subject: Optional[str] = None,
+    ):
+        if not self.allow_private_urls:
+            try:
+                validate_restricted_url(self.url.get_secret_value())
+            except ValueError as exc:
+                if self._raise_on_failure:
+                    raise NotificationError(str(exc))
+                raise
+
+        await super().notify(body, subject)
 
 
 # TODO: Move to prefect-slack once collection block auto-registration is

src/prefect/blocks/webhook.py

@@ -3,6 +3,7 @@
 from httpx import AsyncClient, AsyncHTTPTransport, Response
 
 from prefect._internal.pydantic import HAS_PYDANTIC_V2
+from prefect.utilities.urls import validate_restricted_url
 
 if HAS_PYDANTIC_V2:
     from pydantic.v1 import Field, SecretStr
@@ -38,7 +39,10 @@ class Webhook(Block):
         description="The webhook URL.",
         examples=["https://hooks.slack.com/XXX"],
     )
-
+    allow_private_urls: bool = Field(
+        default=True,
+        description="Whether to allow notifications to private URLs. Defaults to True.",
+    )
     headers: SecretDict = Field(
         default_factory=lambda: SecretDict(dict()),
         title="Webhook Headers",
@@ -55,6 +59,9 @@ async def call(self, payload: Optional[dict] = None) -> Response:
         Args:
             payload: an optional payload to send when calling the webhook.
         """
+        if not self.allow_private_urls:
+            validate_restricted_url(self.url.get_secret_value())
+
         async with self._client:
             return await self._client.request(
                 method=self.method,

src/prefect/utilities/urls.py

@@ -0,0 +1,49 @@
+import ipaddress
+import socket
+from urllib.parse import urlparse
+
+
+def validate_restricted_url(url: str):
+    """
+    Validate that the provided URL is safe for outbound requests.  This prevents
+    attacks like SSRF (Server Side Request Forgery), where an attacker can make
+    requests to internal services (like the GCP metadata service, localhost addresses,
+    or in-cluster Kubernetes services)
+    Args:
+        url: The URL to validate.
+    Raises:
+        ValueError: If the URL is a restricted URL.
+    """
+
+    try:
+        parsed_url = urlparse(url)
+    except ValueError:
+        raise ValueError(f"{url!r} is not a valid URL.")
+
+    if parsed_url.scheme not in ("http", "https"):
+        raise ValueError(
+            f"{url!r} is not a valid URL.  Only HTTP and HTTPS URLs are allowed."
+        )
+
+    hostname = parsed_url.hostname or ""
+
+    # Remove IPv6 brackets if present
+    if hostname.startswith("[") and hostname.endswith("]"):
+        hostname = hostname[1:-1]
+
+    if not hostname:
+        raise ValueError(f"{url!r} is not a valid URL.")
+
+    try:
+        ip_address = socket.gethostbyname(hostname)
+        ip = ipaddress.ip_address(ip_address)
+    except socket.gaierror:
+        try:
+            ip = ipaddress.ip_address(hostname)
+        except ValueError:
+            raise ValueError(f"{url!r} is not a valid URL.  It could not be resolved.")
+
+    if ip.is_private:
+        raise ValueError(
+            f"{url!r} is not a valid URL.  It resolves to the private address {ip}."
+        )

src/prefect/workers/base.py

@@ -137,6 +137,12 @@ async def from_template_and_values(
         variables = cls._get_base_config_defaults(
             variables_schema.get("properties", {})
         )
+
+        # copy variable defaults for `env` to job config before they're replaced by
+        # deployment overrides
+        if variables.get("env"):
+            job_config["env"] = variables.get("env")
+
         variables.update(values)
 
         # deep merge `env`

tests/blocks/test_notifications.py

@@ -8,6 +8,7 @@
 import respx
 
 import prefect
+from prefect.blocks.abstract import NotificationError
 from prefect.blocks.notifications import (
     PREFECT_NOTIFY_TYPE_DEFAULT,
     AppriseNotificationBlock,
@@ -22,6 +23,41 @@
 )
 from prefect.testing.utilities import AsyncMock
 
+RESTRICTED_URLS = [
+    ("", ""),
+    (" ", ""),
+    ("[]", ""),
+    ("not a url", ""),
+    ("http://", ""),
+    ("https://", ""),
+    ("ftp://example.com", "HTTP and HTTPS"),
+    ("gopher://example.com", "HTTP and HTTPS"),
+    ("https://localhost", "private address"),
+    ("https://127.0.0.1", "private address"),
+    ("https://[::1]", "private address"),
+    ("https://[fc00:1234:5678:9abc::10]", "private address"),
+    ("https://[fd12:3456:789a:1::1]", "private address"),
+    ("https://[fe80::1234:5678:9abc]", "private address"),
+    ("https://10.0.0.1", "private address"),
+    ("https://10.255.255.255", "private address"),
+    ("https://172.16.0.1", "private address"),
+    ("https://172.31.255.255", "private address"),
+    ("https://192.168.1.1", "private address"),
+    ("https://192.168.1.255", "private address"),
+    ("https://169.254.0.1", "private address"),
+    ("https://169.254.169.254", "private address"),
+    ("https://169.254.254.255", "private address"),
+    # These will resolve to a private address in production, but not in tests,
+    # so we'll use "resolve" as the reason to catch both cases
+    ("https://metadata.google.internal", "resolve"),
+    ("https://anything.privatecloud", "resolve"),
+    ("https://anything.privatecloud.svc", "resolve"),
+    ("https://anything.privatecloud.svc.cluster.local", "resolve"),
+    ("https://cluster-internal", "resolve"),
+    ("https://network-internal.cloud.svc", "resolve"),
+    ("https://private-internal.cloud.svc.cluster.local", "resolve"),
+]
+
 
 def reload_modules():
     """
@@ -100,6 +136,36 @@ def test_is_picklable(self, block_class: Type[AppriseNotificationBlock]):
         unpickled = cloudpickle.loads(pickled)
         assert isinstance(unpickled, block_class)
 
+    @pytest.mark.parametrize("value, reason", RESTRICTED_URLS)
+    async def test_notification_can_prevent_restricted_urls(
+        self, block_class, value: str, reason: str
+    ):
+        notification = block_class(url=value, allow_private_urls=False)
+
+        with pytest.raises(ValueError, match=f"is not a valid URL.*{reason}"):
+            await notification.notify(subject="example", body="example")
+
+    async def test_raises_on_url_validation_failure(self, block_class):
+        """
+        When within a raise_on_failure block, we want URL validation errors to be
+        wrapped and captured as NotificationErrors for reporting back to users.
+        """
+        block = block_class(url="https://127.0.0.1/foo/bar", allow_private_urls=False)
+
+        # outside of a raise_on_failure block, we get a ValueError directly
+        with pytest.raises(ValueError, match="not a valid URL") as captured:
+            await block.notify(subject="Test", body="Test")
+
+        # inside of a raise_on_failure block, we get a NotificationError
+        with block.raise_on_failure():
+            with pytest.raises(NotificationError) as captured:
+                await block.notify(subject="Test", body="Test")
+
+        assert captured.value.log == (
+            "'https://127.0.0.1/foo/bar' is not a valid URL.  It resolves to the "
+            "private address 127.0.0.1."
+        )
+
 
 class TestMattermostWebhook:
     async def test_notify_async(self):

tests/blocks/test_webhook.py

@@ -3,6 +3,42 @@
 from prefect.blocks.webhook import Webhook
 from prefect.testing.utilities import AsyncMock
 
+RESTRICTED_URLS = [
+    ("", ""),
+    (" ", ""),
+    ("[]", ""),
+    ("not a url", ""),
+    ("http://", ""),
+    ("https://", ""),
+    ("http://[]/foo/bar", ""),
+    ("ftp://example.com", "HTTP and HTTPS"),
+    ("gopher://example.com", "HTTP and HTTPS"),
+    ("https://localhost", "private address"),
+    ("https://127.0.0.1", "private address"),
+    ("https://[::1]", "private address"),
+    ("https://[fc00:1234:5678:9abc::10]", "private address"),
+    ("https://[fd12:3456:789a:1::1]", "private address"),
+    ("https://[fe80::1234:5678:9abc]", "private address"),
+    ("https://10.0.0.1", "private address"),
+    ("https://10.255.255.255", "private address"),
+    ("https://172.16.0.1", "private address"),
+    ("https://172.31.255.255", "private address"),
+    ("https://192.168.1.1", "private address"),
+    ("https://192.168.1.255", "private address"),
+    ("https://169.254.0.1", "private address"),
+    ("https://169.254.169.254", "private address"),
+    ("https://169.254.254.255", "private address"),
+    # These will resolve to a private address in production, but not in tests,
+    # so we'll use "resolve" as the reason to catch both cases
+    ("https://metadata.google.internal", "resolve"),
+    ("https://anything.privatecloud", "resolve"),
+    ("https://anything.privatecloud.svc", "resolve"),
+    ("https://anything.privatecloud.svc.cluster.local", "resolve"),
+    ("https://cluster-internal", "resolve"),
+    ("https://network-internal.cloud.svc", "resolve"),
+    ("https://private-internal.cloud.svc.cluster.local", "resolve"),
+]
+
 
 class TestWebhook:
     def test_webhook_raises_error_on_bad_request_method(self):
@@ -13,6 +49,15 @@ def test_webhook_raises_error_on_bad_request_method(self):
             with pytest.raises(ValueError):
                 Webhook(method=bad_method, url="http://google.com")
 
+    @pytest.mark.parametrize("value, reason", RESTRICTED_URLS)
+    async def test_webhook_must_not_point_to_restricted_urls(
+        self, value: str, reason: str
+    ):
+        webhook = Webhook(url=value, allow_private_urls=False)
+
+        with pytest.raises(ValueError, match=f"is not a valid URL.*{reason}"):
+            await webhook.call(payload="some payload")
+
     async def test_webhook_sends(self, monkeypatch):
         send_mock = AsyncMock()
         monkeypatch.setattr("httpx.AsyncClient.request", send_mock)

tests/conftest.py

@@ -594,22 +594,3 @@ def reset_sys_modules():
             del sys.modules[module]
 
     importlib.invalidate_caches()
-
-
-@pytest.fixture(autouse=True, scope="module")
-def leaves_no_extraneous_files():
-    """This fixture will fail a test if it seems to have left new files or directories
-    in the root of the local working tree.  For performance, it only checks for changes
-    at the test module level, but that should generally be enough to narrow down what
-    is happening.  If you're having trouble isolating the problematic test, you can
-    switch it to scope="function" temporarily.  It may also help to run the test suite
-    with one process (-n0) so that unrelated tests won't fail."""
-    before = set(Path(".").iterdir())
-    yield
-    after = set(Path(".").iterdir())
-    new_files = after - before
-    if new_files:
-        raise AssertionError(
-            "One of the tests in this module left new files in the "
-            f"working directory: {new_files}"
-        )

tests/deployment/test_steps.py

@@ -818,8 +818,9 @@ async def test_pip_install_fails_on_error(self):
                     }
                 }
             )
+        assert "pip_install_requirements failed with error code 1" in str(exc.value)
         assert (
-            "pip_install_requirements failed with error code 1: ERROR: Could not open "
+            "ERROR: Could not open "
             "requirements file: [Errno 2] No such file or directory: "
             "'doesnt-exist.txt'" in str(exc.value)
         )

tests/utilities/test_urls.py

@@ -0,0 +1,45 @@
+import pytest
+
+from prefect.utilities.urls import validate_restricted_url
+
+RESTRICTED_URLS = [
+    ("", ""),
+    (" ", ""),
+    ("[]", ""),
+    ("not a url", ""),
+    ("http://", ""),
+    ("https://", ""),
+    ("http://[]/foo/bar", ""),
+    ("ftp://example.com", "HTTP and HTTPS"),
+    ("gopher://example.com", "HTTP and HTTPS"),
+    ("https://localhost", "private address"),
+    ("https://127.0.0.1", "private address"),
+    ("https://[::1]", "private address"),
+    ("https://[fc00:1234:5678:9abc::10]", "private address"),
+    ("https://[fd12:3456:789a:1::1]", "private address"),
+    ("https://[fe80::1234:5678:9abc]", "private address"),
+    ("https://10.0.0.1", "private address"),
+    ("https://10.255.255.255", "private address"),
+    ("https://172.16.0.1", "private address"),
+    ("https://172.31.255.255", "private address"),
+    ("https://192.168.1.1", "private address"),
+    ("https://192.168.1.255", "private address"),
+    ("https://169.254.0.1", "private address"),
+    ("https://169.254.169.254", "private address"),
+    ("https://169.254.254.255", "private address"),
+    # These will resolve to a private address in production, but not in tests,
+    # so we'll use "resolve" as the reason to catch both cases
+    ("https://metadata.google.internal", "resolve"),
+    ("https://anything.privatecloud", "resolve"),
+    ("https://anything.privatecloud.svc", "resolve"),
+    ("https://anything.privatecloud.svc.cluster.local", "resolve"),
+    ("https://cluster-internal", "resolve"),
+    ("https://network-internal.cloud.svc", "resolve"),
+    ("https://private-internal.cloud.svc.cluster.local", "resolve"),
+]
+
+
+@pytest.mark.parametrize("value, reason", RESTRICTED_URLS)
+def test_validate_restricted_url_validates(value: str, reason: str):
+    with pytest.raises(ValueError, match=f"is not a valid URL.*{reason}"):
+        validate_restricted_url(url=value)