diff --git a/README.md b/README.md index b4915f9..79a1e6c 100644 --- a/README.md +++ b/README.md @@ -100,3 +100,47 @@ app = asgi_csrf( skip_if_scope=skip_api_paths ) ``` + +## Customizing error messages + +You can customize the error messages produced by the CSRF protection middleware by providing a custom error message function. This function should accept an error message ID and return the corresponding error message. + +Here is an example of how to use the custom error message function: + +```python +from asgi_csrf import asgi_csrf, ErrorMessageID + +def custom_error_message_function(message_id): + if message_id == ErrorMessageID.FORM_URLENCODED_MISMATCH: + return "Custom error message for form-urlencoded mismatch" + elif message_id == ErrorMessageID.MULTIPART_MISMATCH: + return "Custom error message for multipart mismatch" + elif message_id == ErrorMessageID.FILE_BEFORE_TOKEN: + return "Custom error message for file before token" + elif message_id == ErrorMessageID.UNKNOWN_CONTENT_TYPE: + return "Custom error message for unknown content type" + else: + return "Unknown error" + +app = asgi_csrf( + app, + signing_secret="secret-goes-here", + custom_error_message_function=custom_error_message_function +) +``` + +In this example, the `custom_error_message_function` is defined to return custom error messages based on the error message ID. The `asgi_csrf` middleware is then configured to use this custom error message function by passing it as the `custom_error_message_function` parameter. + +The error message IDs are defined as an enum in the `asgi_csrf` module: + +```python +from enum import Enum + +class ErrorMessageID(Enum): + FORM_URLENCODED_MISMATCH = 1 + MULTIPART_MISMATCH = 2 + FILE_BEFORE_TOKEN = 3 + UNKNOWN_CONTENT_TYPE = 4 +``` + +You can use these error message IDs to customize the error messages produced by the CSRF protection middleware. diff --git a/asgi_csrf.egg-info/PKG-INFO b/asgi_csrf.egg-info/PKG-INFO new file mode 100644 index 0000000..982e567 --- /dev/null +++ b/asgi_csrf.egg-info/PKG-INFO @@ -0,0 +1,121 @@ +Metadata-Version: 2.1 +Name: asgi-csrf +Version: 0.9 +Summary: ASGI middleware for protecting against CSRF attacks +Home-page: https://github.com/simonw/asgi-csrf +Author: Simon Willison +License: Apache License, Version 2.0 +Description-Content-Type: text/markdown +License-File: LICENSE +Requires-Dist: itsdangerous +Requires-Dist: python-multipart +Provides-Extra: test +Requires-Dist: pytest; extra == "test" +Requires-Dist: pytest-asyncio; extra == "test" +Requires-Dist: httpx>=0.16; extra == "test" +Requires-Dist: starlette; extra == "test" +Requires-Dist: pytest-cov; extra == "test" +Requires-Dist: asgi-lifespan; extra == "test" + +# asgi-csrf + +[![PyPI](https://img.shields.io/pypi/v/asgi-csrf.svg)](https://pypi.org/project/asgi-csrf/) +[![Changelog](https://img.shields.io/github/v/release/simonw/asgi-csrf?include_prereleases&label=changelog)](https://github.com/simonw/asgi-csrf/releases) +[![codecov](https://codecov.io/gh/simonw/asgi-csrf/branch/main/graph/badge.svg)](https://codecov.io/gh/simonw/asgi-csrf) +[![License](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](https://github.com/simonw/asgi-csrf/blob/main/LICENSE) + +ASGI middleware for protecting against CSRF attacks + +## Installation + + pip install asgi-csrf + +## Background + +See the [OWASP guide to Cross Site Request Forgery (CSRF)](https://owasp.org/www-community/attacks/csrf) and their [Cross-Site Request Forgery (CSRF) Prevention Cheat Sheet](https://cheatsheetseries.owasp.org/cheatsheets/Cross-Site_Request_Forgery_Prevention_Cheat_Sheet.html). + +This middleware implements the [Double Submit Cookie pattern](https://cheatsheetseries.owasp.org/cheatsheets/Cross-Site_Request_Forgery_Prevention_Cheat_Sheet.html#double-submit-cookie), where a cookie is set that is then compared to a `csrftoken` hidden form field or a `x-csrftoken` HTTP header. + +## Usage + +Decorate your ASGI application like this: + +```python +from asgi_csrf import asgi_csrf +from .my_asgi_app import app + + +app = asgi_csrf(app, signing_secret="secret-goes-here") +``` + +The middleware will set a `csrftoken` cookie, if one is missing. The value of that token will be made available to your ASGI application through the `scope["csrftoken"]` function. + +Your application code should include that value as a hidden form field in any POST forms: + +```html +
+ ... + +
+``` + +Note that `request.scope["csrftoken"]()` is a function that returns a string. Calling that function also lets the middleware know that the cookie should be set by that page, if the user does not already have that cookie. + +If the cookie needs to be set, the middleware will add a `Vary: Cookie` header to the response to ensure it is not incorrectly cached by any CDNs or intermediary proxies. + +The middleware will return a 403 forbidden error for any POST requests that do not include the matching `csrftoken` - either in the POST data or in a `x-csrftoken` HTTP header (useful for JavaScript `fetch()` calls). + +The `signing_secret` is used to sign the tokens, to protect against subdomain vulnerabilities. + +If you do not pass in an explicit `signing_secret` parameter, the middleware will look for a `ASGI_CSRF_SECRET` environment variable. + +If it cannot find that environment variable, it will generate a random secret which will persist for the lifetime of the server. + +This means that if you do not configure a specific secret your user's `csrftoken` cookies will become invalid every time the server restarts! You should configure a secret. + +## Always setting the cookie if it is not already set + +By default this middleware only sets the `csrftoken` cookie if the user encounters a page that needs it - due to that page calling the `request.scope["csrftoken"]()` function, for example to populate a hidden field in a form. + +If you would like the middleware to set that cookie for any incoming request that does not already provide the cookie, you can use the `always_set_cookie=True` argument: + +```python +app = asgi_csrf(app, signing_secret="secret-goes-here", always_set_cookie=True) +``` + +## Other cases that skip CSRF protection + +If the request includes an `Authorization: Bearer ...` header, commonly used by OAuth and JWT authentication, the request will not be required to include a CSRF token. This is because browsers cannot send those headers in a context that can be abused. + +If the request has no cookies at all it will be allowed through, since CSRF protection is only necessary for requests from authenticated users. + +### always_protect + +If you have paths that should always be protected even without cookies - your login form for example (to avoid [login CSRF](https://cheatsheetseries.owasp.org/cheatsheets/Cross-Site_Request_Forgery_Prevention_Cheat_Sheet.html#login-csrf) attacks) you can protect those paths by passing them as the ``always_protect`` parameter: + +```python +app = asgi_csrf( + app, + signing_secret="secret-goes-here", + always_protect={"/login"} +) +``` + +### skip_if_scope + +There may be situations in which you want to opt-out of CSRF protection even for authenticated POST requests - this is often the case for web APIs for example. + +The `skip_if_scope=` parameter can be used to provide a callback function which is passed an ASGI scope and returns `True` if CSRF protection should be skipped for that request. + +This example skips CSRF protection for any incoming request where the request path starts with `/api/`: + +```python +def skip_api_paths(scope) + return scope["path"].startswith("/api/") + +app = asgi_csrf( + app, + signing_secret="secret-goes-here", + skip_if_scope=skip_api_paths +) +``` diff --git a/asgi_csrf.egg-info/SOURCES.txt b/asgi_csrf.egg-info/SOURCES.txt new file mode 100644 index 0000000..39e616d --- /dev/null +++ b/asgi_csrf.egg-info/SOURCES.txt @@ -0,0 +1,9 @@ +LICENSE +README.md +asgi_csrf.py +setup.py +asgi_csrf.egg-info/PKG-INFO +asgi_csrf.egg-info/SOURCES.txt +asgi_csrf.egg-info/dependency_links.txt +asgi_csrf.egg-info/requires.txt +asgi_csrf.egg-info/top_level.txt \ No newline at end of file diff --git a/asgi_csrf.egg-info/dependency_links.txt b/asgi_csrf.egg-info/dependency_links.txt new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/asgi_csrf.egg-info/dependency_links.txt @@ -0,0 +1 @@ + diff --git a/asgi_csrf.egg-info/requires.txt b/asgi_csrf.egg-info/requires.txt new file mode 100644 index 0000000..a99d5f6 --- /dev/null +++ b/asgi_csrf.egg-info/requires.txt @@ -0,0 +1,10 @@ +itsdangerous +python-multipart + +[test] +pytest +pytest-asyncio +httpx>=0.16 +starlette +pytest-cov +asgi-lifespan diff --git a/asgi_csrf.egg-info/top_level.txt b/asgi_csrf.egg-info/top_level.txt new file mode 100644 index 0000000..861d946 --- /dev/null +++ b/asgi_csrf.egg-info/top_level.txt @@ -0,0 +1 @@ +asgi_csrf diff --git a/asgi_csrf.py b/asgi_csrf.py index 5b439e1..bc1a30c 100644 --- a/asgi_csrf.py +++ b/asgi_csrf.py @@ -7,6 +7,7 @@ from itsdangerous.url_safe import URLSafeSerializer from itsdangerous import BadSignature import secrets +from enum import Enum DEFAULT_COOKIE_NAME = "csrftoken" DEFAULT_FORM_INPUT = "csrftoken" @@ -15,6 +16,11 @@ SCOPE_KEY = "csrftoken" ENV_SECRET = "ASGI_CSRF_SECRET" +class ErrorMessageID(Enum): + FORM_URLENCODED_MISMATCH = 1 + MULTIPART_MISMATCH = 2 + FILE_BEFORE_TOKEN = 3 + UNKNOWN_CONTENT_TYPE = 4 def asgi_csrf_decorator( cookie_name=DEFAULT_COOKIE_NAME, @@ -25,6 +31,7 @@ def asgi_csrf_decorator( always_protect=None, always_set_cookie=False, skip_if_scope=None, + custom_error_message_function=None, ): if signing_secret is None: signing_secret = os.environ.get(ENV_SECRET, None) @@ -146,6 +153,8 @@ async def wrapped_send(event): await send_csrf_failed( scope, wrapped_send, + ErrorMessageID.FORM_URLENCODED_MISMATCH, + custom_error_message_function, "form-urlencoded POST field did not match cookie", ) return @@ -168,6 +177,8 @@ async def wrapped_send(event): await send_csrf_failed( scope, wrapped_send, + ErrorMessageID.MULTIPART_MISMATCH, + custom_error_message_function, "multipart/form-data POST field did not match cookie", ) return @@ -175,6 +186,8 @@ async def wrapped_send(event): await send_csrf_failed( scope, wrapped_send, + ErrorMessageID.FILE_BEFORE_TOKEN, + custom_error_message_function, "File encountered before csrftoken - make sure csrftoken is first in the HTML", ) return @@ -183,7 +196,11 @@ async def wrapped_send(event): return else: await send_csrf_failed( - scope, wrapped_send, message="Unknown content-type" + scope, + wrapped_send, + ErrorMessageID.UNKNOWN_CONTENT_TYPE, + custom_error_message_function, + "Unknown content-type", ) return @@ -271,8 +288,9 @@ async def replay_receive(): return None, replay_receive -async def send_csrf_failed(scope, send, message="CSRF check failed"): +async def send_csrf_failed(scope, send, message_id, custom_error_message_function, default_message): assert scope["type"] == "http" + message = custom_error_message_function(message_id) if custom_error_message_function else default_message await send( { "type": "http.response.start", @@ -292,6 +310,7 @@ def asgi_csrf( always_protect=None, always_set_cookie=False, skip_if_scope=None, + custom_error_message_function=None, ): return asgi_csrf_decorator( cookie_name, @@ -301,6 +320,7 @@ def asgi_csrf( always_protect=always_protect, always_set_cookie=always_set_cookie, skip_if_scope=skip_if_scope, + custom_error_message_function=custom_error_message_function, )(app) diff --git a/test_asgi_csrf.py b/test_asgi_csrf.py index 73cf4b5..c5e33b2 100644 --- a/test_asgi_csrf.py +++ b/test_asgi_csrf.py @@ -2,7 +2,7 @@ from starlette.applications import Starlette from starlette.responses import JSONResponse from starlette.routing import Route -from asgi_csrf import asgi_csrf +from asgi_csrf import asgi_csrf, ErrorMessageID from itsdangerous.url_safe import URLSafeSerializer import httpx import json @@ -364,3 +364,103 @@ async def test_asgi_lifespan(): cookies={"foo": "bar"}, ) assert 200 == response.status_code + + +def custom_error_message_function(message_id): + if message_id == ErrorMessageID.FORM_URLENCODED_MISMATCH: + return "Custom error message for form-urlencoded mismatch" + elif message_id == ErrorMessageID.MULTIPART_MISMATCH: + return "Custom error message for multipart mismatch" + elif message_id == ErrorMessageID.FILE_BEFORE_TOKEN: + return "Custom error message for file before token" + elif message_id == ErrorMessageID.UNKNOWN_CONTENT_TYPE: + return "Custom error message for unknown content type" + else: + return "Unknown error" + + +@pytest.mark.asyncio +async def test_custom_error_message_function_form_urlencoded_mismatch(csrftoken): + async with httpx.AsyncClient( + app=asgi_csrf( + hello_world_app, + signing_secret=SECRET, + custom_error_message_function=custom_error_message_function, + ) + ) as client: + response = await client.post( + "http://localhost/", + cookies={"csrftoken": csrftoken}, + data={"csrftoken": csrftoken[-1]}, + ) + assert response.status_code == 403 + assert response.text == "Custom error message for form-urlencoded mismatch" + + +@pytest.mark.asyncio +async def test_custom_error_message_function_multipart_mismatch(csrftoken): + async with httpx.AsyncClient( + app=asgi_csrf( + hello_world_app, + signing_secret=SECRET, + custom_error_message_function=custom_error_message_function, + ) + ) as client: + response = await client.post( + "http://localhost/", + data={"csrftoken": csrftoken}, + files={"csv": ("data.csv", "blah,foo\n1,2", "text/csv")}, + cookies={"csrftoken": csrftoken[:-1]}, + ) + assert response.status_code == 403 + assert response.text == "Custom error message for multipart mismatch" + + +@pytest.mark.asyncio +async def test_custom_error_message_function_file_before_token(csrftoken): + async with httpx.AsyncClient( + app=asgi_csrf( + hello_world_app, + signing_secret=SECRET, + custom_error_message_function=custom_error_message_function, + ) + ) as client: + request = httpx.Request( + url="http://localhost/", + method="POST", + data=( + b"--boo\r\n" + b'Content-Disposition: form-data; name="csv"; filename="data.csv"' + b"\r\nContent-Type: text/csv\r\n\r\n" + b"blah,foo\n1,2" + b"\r\n" + b"--boo\r\n" + b'Content-Disposition: form-data; name="csrftoken"\r\n\r\n' + + csrftoken.encode("utf-8") + + b"\r\n" + b"--boo--\r\n" + ), + headers={"content-type": "multipart/form-data; boundary=boo"}, + cookies={"csrftoken": csrftoken}, + ) + response = await client.send(request) + assert response.status_code == 403 + assert response.text == "Custom error message for file before token" + + +@pytest.mark.asyncio +async def test_custom_error_message_function_unknown_content_type(csrftoken): + async with httpx.AsyncClient( + app=asgi_csrf( + hello_world_app, + signing_secret=SECRET, + custom_error_message_function=custom_error_message_function, + ) + ) as client: + response = await client.post( + "http://localhost/", + headers={"content-type": "application/octet-stream"}, + cookies={"csrftoken": csrftoken}, + ) + assert response.status_code == 403 + assert response.text == "Custom error message for unknown content type"