We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
refactor: improve typing in CSRF middleware
https://github.com/litestar-org/litestar/blob/7414f7fd7d4782223502895e6a23b77ed635cd2d/litestar/middleware/csrf.py#L87-L127

At line 105, we use `dict.get()` to set the value of `existing_csrf_token` so it can be `None` if the header doesn't exist.

At line 123, that block is guarded by `self._csrf_tokens_match()` which will return `False` if it is `None`, so actually `existing_csrf_token` cannot be falsy in this block, its just that its value is not narrowed appropriately.

Fixing this would probably be as simple as using `request.cookies.get(..., "")` and `request.headers.get(..., "")` on lines 104 and 105 respectively, and re-type downstream methods to only accept `str` instead of `str | None`.

_Originally posted by @peterschutt in https://github.com/litestar-org/litestar/pull/2751#discussion_r1405515256_
            

<!-- POLAR PLEDGE BADGE START -->
---
> [!NOTE]  
> While we are open for sponsoring on [GitHub Sponsors](https://github.com/sponsors/litestar-org/) and 
> [OpenCollective](https://opencollective.com/litestar), we also utilize [Polar.sh](https://polar.sh/) to engage in pledge-based sponsorship.
>
> Check out all issues funded or available for funding [on our Polar.sh Litestar dashboard](https://polar.sh/litestar-org)
> * If you would like to see an issue prioritized, make a pledge towards it!
> * We receive the pledge once the issue is completed & verified
> * This, along with engagement in the community, helps us know which features are a priority to our users.

<a href="https://polar.sh/litestar-org/litestar/issues/2770">
<picture>
  <source media="(prefers-color-scheme: dark)" srcset="https://polar.sh/api/github/litestar-org/litestar/issues/2770/pledge.svg?darkmode=1">
  <img alt="Fund with Polar" src="https://polar.sh/api/github/litestar-org/litestar/issues/2770/pledge.svg">
</picture>
</a>
<!-- POLAR PLEDGE BADGE END -->


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `litestar/middleware/csrf.py`
Content:
```
1 from __future__ import annotations
2 
3 import hashlib
4 import hmac
5 import secrets
6 from secrets import compare_digest
7 from typing import TYPE_CHECKING, Any
8 
9 from litestar.datastructures import MutableScopeHeaders
10 from litestar.datastructures.cookie import Cookie
11 from litestar.enums import RequestEncodingType, ScopeType
12 from litestar.exceptions import PermissionDeniedException
13 from litestar.middleware._utils import (
14     build_exclude_path_pattern,
15     should_bypass_middleware,
16 )
17 from litestar.middleware.base import MiddlewareProtocol
18 from litestar.utils.scope.state import ScopeState
19 
20 if TYPE_CHECKING:
21     from litestar.config.csrf import CSRFConfig
22     from litestar.connection import Request
23     from litestar.types import (
24         ASGIApp,
25         HTTPSendMessage,
26         Message,
27         Receive,
28         Scope,
29         Scopes,
30         Send,
31     )
32 
33 __all__ = ("CSRFMiddleware",)
34 
35 
36 CSRF_SECRET_BYTES = 32
37 CSRF_SECRET_LENGTH = CSRF_SECRET_BYTES * 2
38 
39 
40 def generate_csrf_hash(token: str, secret: str) -> str:
41     """Generate an HMAC that signs the CSRF token.
42 
43     Args:
44         token: A hashed token.
45         secret: A secret value.
46 
47     Returns:
48         A CSRF hash.
49     """
50     return hmac.new(secret.encode(), token.encode(), hashlib.sha256).hexdigest()
51 
52 
53 def generate_csrf_token(secret: str) -> str:
54     """Generate a CSRF token that includes a randomly generated string signed by an HMAC.
55 
56     Args:
57         secret: A secret string.
58 
59     Returns:
60         A unique CSRF token.
61     """
62     token = secrets.token_hex(CSRF_SECRET_BYTES)
63     token_hash = generate_csrf_hash(token=token, secret=secret)
64     return token + token_hash
65 
66 
67 class CSRFMiddleware(MiddlewareProtocol):
68     """CSRF Middleware class.
69 
70     This Middleware protects against attacks by setting a CSRF cookie with a token and verifying it in request headers.
71     """
72 
73     scopes: Scopes = {ScopeType.HTTP}
74 
75     def __init__(self, app: ASGIApp, config: CSRFConfig) -> None:
76         """Initialize ``CSRFMiddleware``.
77 
78         Args:
79             app: The ``next`` ASGI app to call.
80             config: The CSRFConfig instance.
81         """
82         self.app = app
83         self.config = config
84         self.exclude = build_exclude_path_pattern(exclude=config.exclude)
85 
86     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
87         """ASGI callable.
88 
89         Args:
90             scope: The ASGI connection scope.
91             receive: The ASGI receive function.
92             send: The ASGI send function.
93 
94         Returns:
95             None
96         """
97         if scope["type"] != ScopeType.HTTP:
98             await self.app(scope, receive, send)
99             return
100 
101         request: Request[Any, Any, Any] = scope["app"].request_class(scope=scope, receive=receive)
102         content_type, _ = request.content_type
103         csrf_cookie = request.cookies.get(self.config.cookie_name)
104         existing_csrf_token = request.headers.get(self.config.header_name)
105 
106         if not existing_csrf_token and content_type in {
107             RequestEncodingType.URL_ENCODED,
108             RequestEncodingType.MULTI_PART,
109         }:
110             form = await request.form()
111             existing_csrf_token = form.get("_csrf_token", None)
112 
113         connection_state = ScopeState.from_scope(scope)
114         if request.method in self.config.safe_methods or should_bypass_middleware(
115             scope=scope,
116             scopes=self.scopes,
117             exclude_opt_key=self.config.exclude_from_csrf_key,
118             exclude_path_pattern=self.exclude,
119         ):
120             token = connection_state.csrf_token = csrf_cookie or generate_csrf_token(secret=self.config.secret)
121             await self.app(scope, receive, self.create_send_wrapper(send=send, csrf_cookie=csrf_cookie, token=token))
122         elif self._csrf_tokens_match(existing_csrf_token, csrf_cookie):
123             # we haven't properly narrowed the type of `existing_csrf_token` to be non-None, but we know it is
124             connection_state.csrf_token = existing_csrf_token  # type: ignore[assignment]
125             await self.app(scope, receive, send)
126         else:
127             raise PermissionDeniedException("CSRF token verification failed")
128 
129     def create_send_wrapper(self, send: Send, token: str, csrf_cookie: str | None) -> Send:
130         """Wrap ``send`` to handle CSRF validation.
131 
132         Args:
133             token: The CSRF token.
134             send: The ASGI send function.
135             csrf_cookie: CSRF cookie.
136 
137         Returns:
138             An ASGI send function.
139         """
140 
141         async def send_wrapper(message: Message) -> None:
142             """Send function that wraps the original send to inject a cookie.
143 
144             Args:
145                 message: An ASGI ``Message``
146 
147             Returns:
148                 None
149             """
150             if csrf_cookie is None and message["type"] == "http.response.start":
151                 message.setdefault("headers", [])
152                 self._set_cookie_if_needed(message=message, token=token)
153             await send(message)
154 
155         return send_wrapper
156 
157     def _set_cookie_if_needed(self, message: HTTPSendMessage, token: str) -> None:
158         headers = MutableScopeHeaders.from_message(message)
159         cookie = Cookie(
160             key=self.config.cookie_name,
161             value=token,
162             path=self.config.cookie_path,
163             secure=self.config.cookie_secure,
164             httponly=self.config.cookie_httponly,
165             samesite=self.config.cookie_samesite,
166             domain=self.config.cookie_domain,
167         )
168         headers.add("set-cookie", cookie.to_header(header=""))
169 
170     def _decode_csrf_token(self, token: str) -> str | None:
171         """Decode a CSRF token and validate its HMAC."""
172         if len(token) < CSRF_SECRET_LENGTH + 1:
173             return None
174 
175         token_secret = token[:CSRF_SECRET_LENGTH]
176         existing_hash = token[CSRF_SECRET_LENGTH:]
177         expected_hash = generate_csrf_hash(token=token_secret, secret=self.config.secret)
178         return token_secret if compare_digest(existing_hash, expected_hash) else None
179 
180     def _csrf_tokens_match(self, request_csrf_token: str | None, cookie_csrf_token: str | None) -> bool:
181         """Take the CSRF tokens from the request and the cookie and verify both are valid and identical."""
182         if not (request_csrf_token and cookie_csrf_token):
183             return False
184 
185         decoded_request_token = self._decode_csrf_token(request_csrf_token)
186         decoded_cookie_token = self._decode_csrf_token(cookie_csrf_token)
187         if decoded_request_token is None or decoded_cookie_token is None:
188             return False
189 
190         return compare_digest(decoded_request_token, decoded_cookie_token)
191
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/litestar/middleware/csrf.py b/litestar/middleware/csrf.py
--- a/litestar/middleware/csrf.py
+++ b/litestar/middleware/csrf.py
@@ -119,9 +119,12 @@
         ):
             token = connection_state.csrf_token = csrf_cookie or generate_csrf_token(secret=self.config.secret)
             await self.app(scope, receive, self.create_send_wrapper(send=send, csrf_cookie=csrf_cookie, token=token))
-        elif self._csrf_tokens_match(existing_csrf_token, csrf_cookie):
-            # we haven't properly narrowed the type of `existing_csrf_token` to be non-None, but we know it is
-            connection_state.csrf_token = existing_csrf_token  # type: ignore[assignment]
+        elif (
+            existing_csrf_token is not None
+            and csrf_cookie is not None
+            and self._csrf_tokens_match(existing_csrf_token, csrf_cookie)
+        ):
+            connection_state.csrf_token = existing_csrf_token
             await self.app(scope, receive, send)
         else:
             raise PermissionDeniedException("CSRF token verification failed")
@@ -177,11 +180,8 @@
         expected_hash = generate_csrf_hash(token=token_secret, secret=self.config.secret)
         return token_secret if compare_digest(existing_hash, expected_hash) else None
 
-    def _csrf_tokens_match(self, request_csrf_token: str | None, cookie_csrf_token: str | None) -> bool:
+    def _csrf_tokens_match(self, request_csrf_token: str, cookie_csrf_token: str) -> bool:
         """Take the CSRF tokens from the request and the cookie and verify both are valid and identical."""
-        if not (request_csrf_token and cookie_csrf_token):
-            return False
-
         decoded_request_token = self._decode_csrf_token(request_csrf_token)
         decoded_cookie_token = self._decode_csrf_token(cookie_csrf_token)
         if decoded_request_token is None or decoded_cookie_token is None:

{"golden_diff": "diff --git a/litestar/middleware/csrf.py b/litestar/middleware/csrf.py\n--- a/litestar/middleware/csrf.py\n+++ b/litestar/middleware/csrf.py\n@@ -119,9 +119,12 @@\n         ):\n             token = connection_state.csrf_token = csrf_cookie or generate_csrf_token(secret=self.config.secret)\n             await self.app(scope, receive, self.create_send_wrapper(send=send, csrf_cookie=csrf_cookie, token=token))\n-        elif self._csrf_tokens_match(existing_csrf_token, csrf_cookie):\n-            # we haven't properly narrowed the type of `existing_csrf_token` to be non-None, but we know it is\n-            connection_state.csrf_token = existing_csrf_token  # type: ignore[assignment]\n+        elif (\n+            existing_csrf_token is not None\n+            and csrf_cookie is not None\n+            and self._csrf_tokens_match(existing_csrf_token, csrf_cookie)\n+        ):\n+            connection_state.csrf_token = existing_csrf_token\n             await self.app(scope, receive, send)\n         else:\n             raise PermissionDeniedException(\"CSRF token verification failed\")\n@@ -177,11 +180,8 @@\n         expected_hash = generate_csrf_hash(token=token_secret, secret=self.config.secret)\n         return token_secret if compare_digest(existing_hash, expected_hash) else None\n \n-    def _csrf_tokens_match(self, request_csrf_token: str | None, cookie_csrf_token: str | None) -> bool:\n+    def _csrf_tokens_match(self, request_csrf_token: str, cookie_csrf_token: str) -> bool:\n         \"\"\"Take the CSRF tokens from the request and the cookie and verify both are valid and identical.\"\"\"\n-        if not (request_csrf_token and cookie_csrf_token):\n-            return False\n-\n         decoded_request_token = self._decode_csrf_token(request_csrf_token)\n         decoded_cookie_token = self._decode_csrf_token(cookie_csrf_token)\n         if decoded_request_token is None or decoded_cookie_token is None:\n", "issue": "refactor: improve typing in CSRF middleware\nhttps://github.com/litestar-org/litestar/blob/7414f7fd7d4782223502895e6a23b77ed635cd2d/litestar/middleware/csrf.py#L87-L127\r\n\r\nAt line 105, we use `dict.get()` to set the value of `existing_csrf_token` so it can be `None` if the header doesn't exist.\r\n\r\nAt line 123, that block is guarded by `self._csrf_tokens_match()` which will return `False` if it is `None`, so actually `existing_csrf_token` cannot be falsy in this block, its just that its value is not narrowed appropriately.\r\n\r\nFixing this would probably be as simple as using `request.cookies.get(..., \"\")` and `request.headers.get(..., \"\")` on lines 104 and 105 respectively, and re-type downstream methods to only accept `str` instead of `str | None`.\r\n\r\n_Originally posted by @peterschutt in https://github.com/litestar-org/litestar/pull/2751#discussion_r1405515256_\r\n            \n\n<!-- POLAR PLEDGE BADGE START -->\n---\n> [!NOTE]  \n> While we are open for sponsoring on [GitHub Sponsors](https://github.com/sponsors/litestar-org/) and \n> [OpenCollective](https://opencollective.com/litestar), we also utilize [Polar.sh](https://polar.sh/) to engage in pledge-based sponsorship.\n>\n> Check out all issues funded or available for funding [on our Polar.sh Litestar dashboard](https://polar.sh/litestar-org)\n> * If you would like to see an issue prioritized, make a pledge towards it!\n> * We receive the pledge once the issue is completed & verified\n> * This, along with engagement in the community, helps us know which features are a priority to our users.\n\n<a href=\"https://polar.sh/litestar-org/litestar/issues/2770\">\n<picture>\n  <source media=\"(prefers-color-scheme: dark)\" srcset=\"https://polar.sh/api/github/litestar-org/litestar/issues/2770/pledge.svg?darkmode=1\">\n  <img alt=\"Fund with Polar\" src=\"https://polar.sh/api/github/litestar-org/litestar/issues/2770/pledge.svg\">\n</picture>\n</a>\n<!-- POLAR PLEDGE BADGE END -->\n\n", "before_files": [{"content": "from __future__ import annotations\n\nimport hashlib\nimport hmac\nimport secrets\nfrom secrets import compare_digest\nfrom typing import TYPE_CHECKING, Any\n\nfrom litestar.datastructures import MutableScopeHeaders\nfrom litestar.datastructures.cookie import Cookie\nfrom litestar.enums import RequestEncodingType, ScopeType\nfrom litestar.exceptions import PermissionDeniedException\nfrom litestar.middleware._utils import (\n    build_exclude_path_pattern,\n    should_bypass_middleware,\n)\nfrom litestar.middleware.base import MiddlewareProtocol\nfrom litestar.utils.scope.state import ScopeState\n\nif TYPE_CHECKING:\n    from litestar.config.csrf import CSRFConfig\n    from litestar.connection import Request\n    from litestar.types import (\n        ASGIApp,\n        HTTPSendMessage,\n        Message,\n        Receive,\n        Scope,\n        Scopes,\n        Send,\n    )\n\n__all__ = (\"CSRFMiddleware\",)\n\n\nCSRF_SECRET_BYTES = 32\nCSRF_SECRET_LENGTH = CSRF_SECRET_BYTES * 2\n\n\ndef generate_csrf_hash(token: str, secret: str) -> str:\n    \"\"\"Generate an HMAC that signs the CSRF token.\n\n    Args:\n        token: A hashed token.\n        secret: A secret value.\n\n    Returns:\n        A CSRF hash.\n    \"\"\"\n    return hmac.new(secret.encode(), token.encode(), hashlib.sha256).hexdigest()\n\n\ndef generate_csrf_token(secret: str) -> str:\n    \"\"\"Generate a CSRF token that includes a randomly generated string signed by an HMAC.\n\n    Args:\n        secret: A secret string.\n\n    Returns:\n        A unique CSRF token.\n    \"\"\"\n    token = secrets.token_hex(CSRF_SECRET_BYTES)\n    token_hash = generate_csrf_hash(token=token, secret=secret)\n    return token + token_hash\n\n\nclass CSRFMiddleware(MiddlewareProtocol):\n    \"\"\"CSRF Middleware class.\n\n    This Middleware protects against attacks by setting a CSRF cookie with a token and verifying it in request headers.\n    \"\"\"\n\n    scopes: Scopes = {ScopeType.HTTP}\n\n    def __init__(self, app: ASGIApp, config: CSRFConfig) -> None:\n        \"\"\"Initialize ``CSRFMiddleware``.\n\n        Args:\n            app: The ``next`` ASGI app to call.\n            config: The CSRFConfig instance.\n        \"\"\"\n        self.app = app\n        self.config = config\n        self.exclude = build_exclude_path_pattern(exclude=config.exclude)\n\n    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:\n        \"\"\"ASGI callable.\n\n        Args:\n            scope: The ASGI connection scope.\n            receive: The ASGI receive function.\n            send: The ASGI send function.\n\n        Returns:\n            None\n        \"\"\"\n        if scope[\"type\"] != ScopeType.HTTP:\n            await self.app(scope, receive, send)\n            return\n\n        request: Request[Any, Any, Any] = scope[\"app\"].request_class(scope=scope, receive=receive)\n        content_type, _ = request.content_type\n        csrf_cookie = request.cookies.get(self.config.cookie_name)\n        existing_csrf_token = request.headers.get(self.config.header_name)\n\n        if not existing_csrf_token and content_type in {\n            RequestEncodingType.URL_ENCODED,\n            RequestEncodingType.MULTI_PART,\n        }:\n            form = await request.form()\n            existing_csrf_token = form.get(\"_csrf_token\", None)\n\n        connection_state = ScopeState.from_scope(scope)\n        if request.method in self.config.safe_methods or should_bypass_middleware(\n            scope=scope,\n            scopes=self.scopes,\n            exclude_opt_key=self.config.exclude_from_csrf_key,\n            exclude_path_pattern=self.exclude,\n        ):\n            token = connection_state.csrf_token = csrf_cookie or generate_csrf_token(secret=self.config.secret)\n            await self.app(scope, receive, self.create_send_wrapper(send=send, csrf_cookie=csrf_cookie, token=token))\n        elif self._csrf_tokens_match(existing_csrf_token, csrf_cookie):\n            # we haven't properly narrowed the type of `existing_csrf_token` to be non-None, but we know it is\n            connection_state.csrf_token = existing_csrf_token  # type: ignore[assignment]\n            await self.app(scope, receive, send)\n        else:\n            raise PermissionDeniedException(\"CSRF token verification failed\")\n\n    def create_send_wrapper(self, send: Send, token: str, csrf_cookie: str | None) -> Send:\n        \"\"\"Wrap ``send`` to handle CSRF validation.\n\n        Args:\n            token: The CSRF token.\n            send: The ASGI send function.\n            csrf_cookie: CSRF cookie.\n\n        Returns:\n            An ASGI send function.\n        \"\"\"\n\n        async def send_wrapper(message: Message) -> None:\n            \"\"\"Send function that wraps the original send to inject a cookie.\n\n            Args:\n                message: An ASGI ``Message``\n\n            Returns:\n                None\n            \"\"\"\n            if csrf_cookie is None and message[\"type\"] == \"http.response.start\":\n                message.setdefault(\"headers\", [])\n                self._set_cookie_if_needed(message=message, token=token)\n            await send(message)\n\n        return send_wrapper\n\n    def _set_cookie_if_needed(self, message: HTTPSendMessage, token: str) -> None:\n        headers = MutableScopeHeaders.from_message(message)\n        cookie = Cookie(\n            key=self.config.cookie_name,\n            value=token,\n            path=self.config.cookie_path,\n            secure=self.config.cookie_secure,\n            httponly=self.config.cookie_httponly,\n            samesite=self.config.cookie_samesite,\n            domain=self.config.cookie_domain,\n        )\n        headers.add(\"set-cookie\", cookie.to_header(header=\"\"))\n\n    def _decode_csrf_token(self, token: str) -> str | None:\n        \"\"\"Decode a CSRF token and validate its HMAC.\"\"\"\n        if len(token) < CSRF_SECRET_LENGTH + 1:\n            return None\n\n        token_secret = token[:CSRF_SECRET_LENGTH]\n        existing_hash = token[CSRF_SECRET_LENGTH:]\n        expected_hash = generate_csrf_hash(token=token_secret, secret=self.config.secret)\n        return token_secret if compare_digest(existing_hash, expected_hash) else None\n\n    def _csrf_tokens_match(self, request_csrf_token: str | None, cookie_csrf_token: str | None) -> bool:\n        \"\"\"Take the CSRF tokens from the request and the cookie and verify both are valid and identical.\"\"\"\n        if not (request_csrf_token and cookie_csrf_token):\n            return False\n\n        decoded_request_token = self._decode_csrf_token(request_csrf_token)\n        decoded_cookie_token = self._decode_csrf_token(cookie_csrf_token)\n        if decoded_request_token is None or decoded_cookie_token is None:\n            return False\n\n        return compare_digest(decoded_request_token, decoded_cookie_token)\n", "path": "litestar/middleware/csrf.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport hashlib\nimport hmac\nimport secrets\nfrom secrets import compare_digest\nfrom typing import TYPE_CHECKING, Any\n\nfrom litestar.datastructures import MutableScopeHeaders\nfrom litestar.datastructures.cookie import Cookie\nfrom litestar.enums import RequestEncodingType, ScopeType\nfrom litestar.exceptions import PermissionDeniedException\nfrom litestar.middleware._utils import (\n    build_exclude_path_pattern,\n    should_bypass_middleware,\n)\nfrom litestar.middleware.base import MiddlewareProtocol\nfrom litestar.utils.scope.state import ScopeState\n\nif TYPE_CHECKING:\n    from litestar.config.csrf import CSRFConfig\n    from litestar.connection import Request\n    from litestar.types import (\n        ASGIApp,\n        HTTPSendMessage,\n        Message,\n        Receive,\n        Scope,\n        Scopes,\n        Send,\n    )\n\n__all__ = (\"CSRFMiddleware\",)\n\n\nCSRF_SECRET_BYTES = 32\nCSRF_SECRET_LENGTH = CSRF_SECRET_BYTES * 2\n\n\ndef generate_csrf_hash(token: str, secret: str) -> str:\n    \"\"\"Generate an HMAC that signs the CSRF token.\n\n    Args:\n        token: A hashed token.\n        secret: A secret value.\n\n    Returns:\n        A CSRF hash.\n    \"\"\"\n    return hmac.new(secret.encode(), token.encode(), hashlib.sha256).hexdigest()\n\n\ndef generate_csrf_token(secret: str) -> str:\n    \"\"\"Generate a CSRF token that includes a randomly generated string signed by an HMAC.\n\n    Args:\n        secret: A secret string.\n\n    Returns:\n        A unique CSRF token.\n    \"\"\"\n    token = secrets.token_hex(CSRF_SECRET_BYTES)\n    token_hash = generate_csrf_hash(token=token, secret=secret)\n    return token + token_hash\n\n\nclass CSRFMiddleware(MiddlewareProtocol):\n    \"\"\"CSRF Middleware class.\n\n    This Middleware protects against attacks by setting a CSRF cookie with a token and verifying it in request headers.\n    \"\"\"\n\n    scopes: Scopes = {ScopeType.HTTP}\n\n    def __init__(self, app: ASGIApp, config: CSRFConfig) -> None:\n        \"\"\"Initialize ``CSRFMiddleware``.\n\n        Args:\n            app: The ``next`` ASGI app to call.\n            config: The CSRFConfig instance.\n        \"\"\"\n        self.app = app\n        self.config = config\n        self.exclude = build_exclude_path_pattern(exclude=config.exclude)\n\n    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:\n        \"\"\"ASGI callable.\n\n        Args:\n            scope: The ASGI connection scope.\n            receive: The ASGI receive function.\n            send: The ASGI send function.\n\n        Returns:\n            None\n        \"\"\"\n        if scope[\"type\"] != ScopeType.HTTP:\n            await self.app(scope, receive, send)\n            return\n\n        request: Request[Any, Any, Any] = scope[\"app\"].request_class(scope=scope, receive=receive)\n        content_type, _ = request.content_type\n        csrf_cookie = request.cookies.get(self.config.cookie_name)\n        existing_csrf_token = request.headers.get(self.config.header_name)\n\n        if not existing_csrf_token and content_type in {\n            RequestEncodingType.URL_ENCODED,\n            RequestEncodingType.MULTI_PART,\n        }:\n            form = await request.form()\n            existing_csrf_token = form.get(\"_csrf_token\", None)\n\n        connection_state = ScopeState.from_scope(scope)\n        if request.method in self.config.safe_methods or should_bypass_middleware(\n            scope=scope,\n            scopes=self.scopes,\n            exclude_opt_key=self.config.exclude_from_csrf_key,\n            exclude_path_pattern=self.exclude,\n        ):\n            token = connection_state.csrf_token = csrf_cookie or generate_csrf_token(secret=self.config.secret)\n            await self.app(scope, receive, self.create_send_wrapper(send=send, csrf_cookie=csrf_cookie, token=token))\n        elif (\n            existing_csrf_token is not None\n            and csrf_cookie is not None\n            and self._csrf_tokens_match(existing_csrf_token, csrf_cookie)\n        ):\n            connection_state.csrf_token = existing_csrf_token\n            await self.app(scope, receive, send)\n        else:\n            raise PermissionDeniedException(\"CSRF token verification failed\")\n\n    def create_send_wrapper(self, send: Send, token: str, csrf_cookie: str | None) -> Send:\n        \"\"\"Wrap ``send`` to handle CSRF validation.\n\n        Args:\n            token: The CSRF token.\n            send: The ASGI send function.\n            csrf_cookie: CSRF cookie.\n\n        Returns:\n            An ASGI send function.\n        \"\"\"\n\n        async def send_wrapper(message: Message) -> None:\n            \"\"\"Send function that wraps the original send to inject a cookie.\n\n            Args:\n                message: An ASGI ``Message``\n\n            Returns:\n                None\n            \"\"\"\n            if csrf_cookie is None and message[\"type\"] == \"http.response.start\":\n                message.setdefault(\"headers\", [])\n                self._set_cookie_if_needed(message=message, token=token)\n            await send(message)\n\n        return send_wrapper\n\n    def _set_cookie_if_needed(self, message: HTTPSendMessage, token: str) -> None:\n        headers = MutableScopeHeaders.from_message(message)\n        cookie = Cookie(\n            key=self.config.cookie_name,\n            value=token,\n            path=self.config.cookie_path,\n            secure=self.config.cookie_secure,\n            httponly=self.config.cookie_httponly,\n            samesite=self.config.cookie_samesite,\n            domain=self.config.cookie_domain,\n        )\n        headers.add(\"set-cookie\", cookie.to_header(header=\"\"))\n\n    def _decode_csrf_token(self, token: str) -> str | None:\n        \"\"\"Decode a CSRF token and validate its HMAC.\"\"\"\n        if len(token) < CSRF_SECRET_LENGTH + 1:\n            return None\n\n        token_secret = token[:CSRF_SECRET_LENGTH]\n        existing_hash = token[CSRF_SECRET_LENGTH:]\n        expected_hash = generate_csrf_hash(token=token_secret, secret=self.config.secret)\n        return token_secret if compare_digest(existing_hash, expected_hash) else None\n\n    def _csrf_tokens_match(self, request_csrf_token: str, cookie_csrf_token: str) -> bool:\n        \"\"\"Take the CSRF tokens from the request and the cookie and verify both are valid and identical.\"\"\"\n        decoded_request_token = self._decode_csrf_token(request_csrf_token)\n        decoded_cookie_token = self._decode_csrf_token(cookie_csrf_token)\n        if decoded_request_token is None or decoded_cookie_token is None:\n            return False\n\n        return compare_digest(decoded_request_token, decoded_cookie_token)\n", "path": "litestar/middleware/csrf.py"}]}