CVE-2023-6022 (Python, published 2023-11-16T17:15Z)
Description: value: Cross-Site Request Forgery (CSRF) in GitHub repository prefecthq/prefect prior to 2.16.5.
Severity: nan (CVSS v3 base score: 8.8)
Weakness: CWE-352 \u2014 Cross-Site Request Forgery (CSRF)
The web application does not, or can not, sufficiently verify whether a well-formed, valid, consistent request was intentionally provided by the user who submitted the request.
Repository: https://github.com/prefecthq/prefect
Files changed:
src/prefect/client/base.py
src/prefect/client/cloud.py
src/prefect/client/orchestration.py
src/prefect/client/schemas/objects.py
src/prefect/settings.py
tests/cli/test_worker.py
tests/client/test_base_client.py
tests/client/test_prefect_client.py
tests/test_deployments.py
Commit message: Add CSRF support to orchestration client (#12314)



Vulnerable code:
from functools import partial
from httpx import HTTPStatusError, Response
        request: Callable,
                response = await request()
                # We got a response; return immediately if it is not retryable
                if response.status_code not in retry_codes:
    async def send(self, *args, **kwargs) -> Response:
        api_request = partial(super().send, *args, **kwargs)

            request=api_request,
        self._client = PrefectHttpxClient(**httpx_settings)
        self._client = PrefectHttpxClient(**httpx_settings)
"""Whether or not to protect the API from CSRF attacks. Experimental and
currently defaults to `False`."""
"""How long a CSRF token is valid for. Defaults to 1 hour."""
from unittest.mock import call
        mock_anyio_sleep.assert_has_awaits([call(2), call(4), call(8)])
        mock_anyio_sleep.assert_has_awaits([call(5), call(0), call(10), call(2.0)])
            [call(pytest.approx(n, rel=0.2)) for n in [2, 4, 8]]

Fixed code:
import uuid
from datetime import datetime, timezone
    Optional,
from httpx import HTTPStatusError, Request, Response
from prefect.client.schemas.objects import CsrfToken
    def __init__(self, *args, enable_csrf_support: bool = False, **kwargs):
        self.enable_csrf_support: bool = enable_csrf_support
        self.csrf_token: Optional[str] = None
        self.csrf_token_expiration: Optional[datetime] = None
        self.csrf_client_id: uuid.UUID = uuid.uuid4()

        super().__init__(*args, **kwargs)

        request: Request,
        send: Callable[[Request], Awaitable[Response]],
        send_args: Tuple,
        send_kwargs: Dict,
        is_change_request = request.method.lower() in {"post", "put", "patch", "delete"}

        if self.enable_csrf_support and is_change_request:
            await self._add_csrf_headers(request=request)

                response = await send(request, *send_args, **send_kwargs)
                # We got a response; check if it's a CSRF error, otherwise
                # return immediately if it is not retryable
                if (
                    response.status_code == status.HTTP_403_FORBIDDEN
                    and "Invalid CSRF token" in response.text
                ):
                    # We got a CSRF error, clear the token and try again
                    self.csrf_token = None
                    await self._add_csrf_headers(request)
                elif response.status_code not in retry_codes:
    async def send(self, request: Request, *args, **kwargs) -> Response:
        - 403 Forbidden, if the request failed due to CSRF protection
        - 408 Request Timeout
        - Any additional status codes provided in `PREFECT_CLIENT_RETRY_EXTRA_CODES`
        super_send = super().send
            request=request,
            send=super_send,
            send_args=args,
            send_kwargs=kwargs,

    async def _add_csrf_headers(self, request: Request):
        now = datetime.now(timezone.utc)

        if not self.enable_csrf_support:
            return

        if not self.csrf_token or (
            self.csrf_token_expiration and now > self.csrf_token_expiration
        ):
            token_request = self.build_request(
                "GET", f"/csrf-token?client={self.csrf_client_id}"
            )

            try:
                token_response = await self.send(token_request)
            except PrefectHTTPStatusError as exc:
                if exc.response.status_code == status.HTTP_404_NOT_FOUND:
                    # The token endpoint is not available, likely an older
                    # server version; disable CSRF support
                    self.enable_csrf_support = False
                    return

                raise

            token: CsrfToken = CsrfToken.parse_obj(token_response.json())
            self.csrf_token = token.token
            self.csrf_token_expiration = token.expiration

        request.headers["Prefect-Csrf-Token"] = self.csrf_token
        request.headers["Prefect-Csrf-Client"] = str(self.csrf_client_id)
        self._client = PrefectHttpxClient(**httpx_settings, enable_csrf_support=False)
    PREFECT_CLIENT_CSRF_SUPPORT_ENABLED,

        enable_csrf_support = (
            self.server_type != ServerType.CLOUD
            and PREFECT_CLIENT_CSRF_SUPPORT_ENABLED.value()
        )

        self._client = PrefectHttpxClient(
            **httpx_settings, enable_csrf_support=enable_csrf_support
        )


class CsrfToken(ObjectBaseModel):
    token: str = Field(
        default=...,
        description="The CSRF token",
    )
    client: str = Field(
        default=..., description="The client id associated with the CSRF token"
    )
    expiration: DateTimeTZ = Field(
        default=..., description="The expiration time of the CSRF token"
    )
PREFECT_CLIENT_CSRF_SUPPORT_ENABLED = Setting(bool, default=True)
"""
Determines if CSRF token handling is active in the Prefect client for API
requests.

When enabled (`True`), the client automatically manages CSRF tokens by
retrieving, storing, and including them in applicable state-changing requests
(POST, PUT, PATCH, DELETE) to the API.

Disabling this setting (`False`) means the client will not handle CSRF tokens,
which might be suitable for environments where CSRF protection is disabled.

Defaults to `True`, ensuring CSRF protection is enabled by default.
"""

"""
Controls the activation of CSRF protection for the Prefect server API.

When enabled (`True`), the server enforces CSRF validation checks on incoming
state-changing requests (POST, PUT, PATCH, DELETE), requiring a valid CSRF
token to be included in the request headers or body. This adds a layer of
security by preventing unauthorized or malicious sites from making requests on
behalf of authenticated users.

It is recommended to enable this setting in production environments where the
API is exposed to web clients to safeguard against CSRF attacks.

Note: Enabling this setting requires corresponding support in the client for
CSRF token management. See PREFECT_CLIENT_CSRF_SUPPORT_ENABLED for more.
"""
"""
Specifies the duration for which a CSRF token remains valid after being issued
by the server.

The default expiration time is set to 1 hour, which offers a reasonable
compromise. Adjust this setting based on your specific security requirements
and usage patterns.
"""
        respx_mock.get("/csrf-token", params={"client": ANY}).pass_through()
from contextlib import asynccontextmanager
from datetime import datetime, timedelta, timezone
from typing import AsyncGenerator, List, Tuple
from unittest import mock
from prefect.client.schemas.objects import CsrfToken
now = datetime.now(timezone.utc)

RESPONSE_CSRF = Response(
    status.HTTP_200_OK,
    json=CsrfToken(
        client="test_client", token="test_token", expiration=now + timedelta(days=1)
    ).dict(json_compatible=True, exclude_unset=True),
    request=Request("a test request", "fake.url/fake/route"),
)

RESPONSE_400 = Response(
    status.HTTP_400_BAD_REQUEST,
    json={"detail": "You done bad things"},
    request=Request("a test request", "fake.url/fake/route"),
)

RESPONSE_404 = Response(
    status.HTTP_404_NOT_FOUND,
    request=Request("a test request", "fake.url/fake/route"),
)

RESPONSE_INVALID_TOKEN = Response(
    status_code=status.HTTP_403_FORBIDDEN,
    json={"detail": "Invalid CSRF token or client identifier."},
    request=Request("a test request", "fake.url/fake/route"),
)

        mock_anyio_sleep.assert_has_awaits([mock.call(2), mock.call(4), mock.call(8)])
        mock_anyio_sleep.assert_has_awaits(
            [mock.call(5), mock.call(0), mock.call(10), mock.call(2.0)]
        )
            [mock.call(pytest.approx(n, rel=0.2)) for n in [2, 4, 8]]


@asynccontextmanager
async def mocked_client(
    responses: List[Response],
) -> AsyncGenerator[Tuple[PrefectHttpxClient, mock.AsyncMock], None]:
    with mock.patch("httpx.AsyncClient.send", autospec=True) as send:
        send.side_effect = responses
        client = PrefectHttpxClient(enable_csrf_support=True)
        async with client:
            try:
                yield client, send
            finally:
                pass


class TestCsrfSupport:
    async def test_no_csrf_headers_not_change_request(self):
        async with mocked_client(responses=[RESPONSE_200]) as (client, send):
            await client.get(url="fake.url/fake/route")

        request = send.call_args[0][1]
        assert isinstance(request, httpx.Request)

        assert "Prefect-Csrf-Token" not in request.headers
        assert "Prefect-Csrf-Client" not in request.headers

    @pytest.mark.parametrize("method", ["post", "put", "patch", "delete"])
    async def test_csrf_headers_on_change_request(self, method: str):
        async with mocked_client(responses=[RESPONSE_CSRF, RESPONSE_200]) as (
            client,
            send,
        ):
            await getattr(client, method)(url="fake.url/fake/route")

        assert send.await_count == 2

        # The first call should be for the CSRF token
        request = send.call_args_list[0][0][1]
        assert isinstance(request, httpx.Request)
        assert request.method == "GET"
        assert request.url == httpx.URL(
            f"/csrf-token?client={str(client.csrf_client_id)}"
        )

        # The second call should be for the actual request
        request = send.call_args_list[1][0][1]
        assert isinstance(request, httpx.Request)
        assert request.method == method.upper()
        assert request.url == httpx.URL("/fake.url/fake/route")
        assert request.headers["Prefect-Csrf-Token"] == "test_token"
        assert request.headers["Prefect-Csrf-Client"] == str(client.csrf_client_id)

    async def test_refreshes_token_on_csrf_403(self):
        async with mocked_client(
            responses=[
                RESPONSE_CSRF,
                RESPONSE_INVALID_TOKEN,
                RESPONSE_CSRF,
                RESPONSE_200,
            ]
        ) as (
            client,
            send,
        ):
            await client.post(url="fake.url/fake/route")

        assert send.await_count == 4

        # The first call should be for the CSRF token
        request = send.call_args_list[0][0][1]
        assert isinstance(request, httpx.Request)
        assert request.method == "GET"
        assert request.url == httpx.URL(
            f"/csrf-token?client={str(client.csrf_client_id)}"
        )

        # The second call should be for the actual request
        request = send.call_args_list[1][0][1]
        assert isinstance(request, httpx.Request)
        assert request.url == httpx.URL("/fake.url/fake/route")
        assert request.headers["Prefect-Csrf-Token"] == "test_token"
        assert request.headers["Prefect-Csrf-Client"] == str(client.csrf_client_id)

        # The third call should be a refresh of the CSRF token
        request = send.call_args_list[0][0][1]
        assert isinstance(request, httpx.Request)
        assert request.method == "GET"
        assert request.url == httpx.URL(
            f"/csrf-token?client={str(client.csrf_client_id)}"
        )

        # The fourth call should be for the actual request
        request = send.call_args_list[1][0][1]
        assert isinstance(request, httpx.Request)
        assert request.url == httpx.URL("/fake.url/fake/route")
        assert request.headers["Prefect-Csrf-Token"] == "test_token"
        assert request.headers["Prefect-Csrf-Client"] == str(client.csrf_client_id)

    async def test_does_not_refresh_csrf_token_not_expired(self):
        async with mocked_client(responses=[RESPONSE_200]) as (
            client,
            send,
        ):
            client.csrf_token = "fresh_token"
            client.csrf_token_expiration = now + timedelta(days=1)
            await client.post(url="fake.url/fake/route")

        assert send.await_count == 1

        request = send.call_args_list[0][0][1]
        assert isinstance(request, httpx.Request)
        assert request.url == httpx.URL("/fake.url/fake/route")
        assert request.headers["Prefect-Csrf-Token"] == "fresh_token"
        assert request.headers["Prefect-Csrf-Client"] == str(client.csrf_client_id)

    async def test_does_refresh_csrf_token_when_expired(self):
        async with mocked_client(responses=[RESPONSE_CSRF, RESPONSE_200]) as (
            client,
            send,
        ):
            client.csrf_token = "old_token"
            client.csrf_token_expiration = now - timedelta(days=1)
            await client.post(url="fake.url/fake/route")

        assert send.await_count == 2

        # The first call should be for the CSRF token
        request = send.call_args_list[0][0][1]
        assert isinstance(request, httpx.Request)
        assert request.method == "GET"
        assert request.url == httpx.URL(
            f"/csrf-token?client={str(client.csrf_client_id)}"
        )

        # The second call should be for the actual request
        request = send.call_args_list[1][0][1]
        assert isinstance(request, httpx.Request)
        assert request.url == httpx.URL("/fake.url/fake/route")
        assert request.headers["Prefect-Csrf-Token"] == "test_token"
        assert request.headers["Prefect-Csrf-Client"] == str(client.csrf_client_id)

    async def test_raises_exception_bad_csrf_token_response(self):
        async with mocked_client(responses=[RESPONSE_400]) as (
            client,
            _,
        ):
            with pytest.raises(PrefectHTTPStatusError):
                await client.post(url="fake.url/fake/route")

    async def test_disables_csrf_support_404_token_endpoint(self):
        async with mocked_client(responses=[RESPONSE_404, RESPONSE_200]) as (
            client,
            send,
        ):
            assert client.enable_csrf_support is True
            await client.post(url="fake.url/fake/route")
            assert client.enable_csrf_support is False
    PREFECT_CLIENT_CSRF_SUPPORT_ENABLED,
        enable_csrf_support=ANY,
        enable_csrf_support=ANY,
        enable_csrf_support=ANY,


class TestPrefectClientCsrfSupport:
    def test_enabled_ephemeral(self, prefect_client: PrefectClient):
        assert prefect_client.server_type == ServerType.EPHEMERAL
        assert prefect_client._client.enable_csrf_support

    async def test_enabled_server_type(self, hosted_api_server):
        async with PrefectClient(hosted_api_server) as prefect_client:
            assert prefect_client.server_type == ServerType.SERVER
            assert prefect_client._client.enable_csrf_support

    async def test_not_enabled_server_type_cloud(self):
        async with PrefectClient(PREFECT_CLOUD_API_URL.value()) as prefect_client:
            assert prefect_client.server_type == ServerType.CLOUD
            assert not prefect_client._client.enable_csrf_support

    async def test_disabled_setting_disabled(self, hosted_api_server):
        with temporary_settings({PREFECT_CLIENT_CSRF_SUPPORT_ENABLED: False}):
            async with PrefectClient(hosted_api_server) as prefect_client:
                assert prefect_client.server_type == ServerType.SERVER
                assert not prefect_client._client.enable_csrf_support
            router.get("/csrf-token", params={"client": mock.ANY}).pass_through()
            router.get("/csrf-token", params={"client": mock.ANY}).pass_through()
            router.get("/csrf-token", params={"client": mock.ANY}).pass_through()
            router.get("/csrf-token", params={"client": mock.ANY}).pass_through()
            router.get("/csrf-token", params={"client": mock.ANY}).pass_through()