Don't CSRF protect Authorization: Bearer ..., refs #11

README.md

@@ -53,6 +53,10 @@ If it cannot find that environment variable, it will generate a random secret wh
 
 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.
 
+## 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.
+
 ## Limitations
 
 * Currently only works for `application/x-www-form-urlencoded` forms, not `multipart/form-data` forms (with file uploads)

asgi_csrf.py

@@ -105,6 +105,14 @@ async def wrapped_send(event):
                     # x-csrftoken header matches
                     await app(scope, receive, wrapped_send)
                     return
+                # Authorization: Bearer skips CSRF check
+                if (
+                    headers.get(b"authorization", b"")
+                    .decode("latin-1")
+                    .startswith("Bearer ")
+                ):
+                    await app(scope, receive, wrapped_send)
+                    return
                 # We need to look for it in the POST body
                 content_type = headers.get(b"content-type", b"").split(b";", 1)[0]
                 if content_type == b"application/x-www-form-urlencoded":

test_asgi_csrf.py

@@ -167,3 +167,17 @@ async def test_multipart_not_supported(csrftoken):
                 files={"csv": ("data.csv", "blah,foo\n1,2", "text/csv")},
                 cookies={"csrftoken": csrftoken},
             )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "authorization,expected_status", [("Bearer xxx", 200), ("Basic xxx", 403)]
+)
+async def test_post_with_authorization(authorization, expected_status):
+    async with httpx.AsyncClient(
+        app=asgi_csrf(hello_world_app, signing_secret=SECRET)
+    ) as client:
+        response = await client.post(
+            "http://localhost/", headers={"Authorization": authorization}
+        )
+        assert expected_status == response.status_code