fix(form): harden AirForm CSRF validation

CHANGELOG/latest.md

@@ -1 +1,3 @@
 ### Latest Changes
+
+- Harden AirForm CSRF validation by rejecting cross-origin submissions and weak signing secrets.

docs/deployment/webassembly.md

@@ -37,17 +37,18 @@ to incoming requests and for packaging application dependencies.
 ## Runtime configuration
 
 `AIRFORM_SECRET` configures form signing when the runtime exposes environment
-variables normally. Runtimes that provide secrets through another API can
-configure the same value directly:
+variables normally. It must contain at least 32 unpredictable bytes. Runtimes
+that provide secrets through another API can configure the same value directly:
 
 ```python
 import air
 
 air.configure_csrf_secret(runtime_secret)
 ```
 
-Call this before rendering or validating forms. Every process or isolate that
-can serve the application must use the same secret.
+Call this before rendering or validating forms. The setting is process-global;
+every process or isolate that can serve the application must use the same
+secret. Changing it while serving traffic invalidates outstanding forms.
 
 Storage, static assets, database bindings, application lifecycle behavior, and
 deployment configuration are responsibilities of the hosting integration.

docs/learn/forms.md

@@ -117,18 +117,40 @@ form.render()
 </div>
 ```
 
-Each field is wrapped in a `<div class="air-field">` with a label, input, and (after validation) error messages. HTML5 validation attributes (`required`, `minlength`, `maxlength`) are derived from Pydantic constraints. CSRF protection is automatic.
+Each field is wrapped in a `<div class="air-field">` with a label, input, and (after validation) error messages. HTML5 validation attributes (`required`, `minlength`, `maxlength`) are derived from Pydantic constraints. CSRF protection is automatic. Browser submissions validated with `from_request()` must include a same-origin `Origin` header, or a same-origin `Referer` header when `Origin` is unavailable. AirForm rejects submissions without either header.
 
-For multi-process production, set `AIRFORM_SECRET` so every process signs
-forms with the same key. If the runtime exposes secrets through another API,
-configure the key before rendering or validating forms:
+If Air runs behind a reverse proxy, configure the proxy integration so the ASGI
+request URL retains the browser-facing scheme and host; AirForm compares the
+source header against that request URL. For example, Caddy's `reverse_proxy`
+preserves `Host` and sets `X-Forwarded-Proto` by default. A Uvicorn backend must
+then trust proxy headers only from the proxy's address:
+
+```console
+uvicorn app:app --proxy-headers --forwarded-allow-ips="127.0.0.1"
+```
+
+Use the proxy's actual IP or trusted CIDR in production. Do not use `*` unless
+the backend is unreachable except through a trusted proxy. Other server/proxy
+combinations must provide the equivalent behavior: preserve the external
+`Host`, communicate the external scheme, and accept those forwarding headers
+only from trusted proxies.
+
+For multi-process production, set `AIRFORM_SECRET` to at least 32 unpredictable
+bytes so every process signs forms with the same key. If the runtime exposes
+secrets through another API, configure the key before rendering or validating
+forms. The configuration is process-global and must not change while serving
+traffic; rotating it invalidates outstanding forms:
 
 ```python
 import air
 
 air.configure_csrf_secret(runtime_secret)
 ```
 
+Rotate the secret atomically across all workers when possible. During a rolling
+rotation, old-key and new-key workers coexist temporarily, so a form rendered by
+one group can be rejected by the other until the rollout finishes.
+
 ## Validating a form
 
 ### From a request

src/air/form/csrf.py

@@ -1,8 +1,8 @@
 """CSRF protection for AirForm.
 
-Tokens are HMAC-signed with a per-process secret. No configuration is
+Tokens are HMAC-signed with a process-global secret. No configuration is
 needed for single-process deployments. For multi-process production, set
-the AIRFORM_SECRET environment variable or call
+the AIRFORM_SECRET environment variable to at least 32 unpredictable bytes or call
 ``configure_csrf_secret()`` so every process uses the same secret.
 
 Token format: timestamp:nonce:signature
@@ -17,18 +17,41 @@
 import sys
 import time
 from typing import TYPE_CHECKING, Any
+from urllib.parse import urlsplit
 
 from pydantic_core import core_schema
 
 if TYPE_CHECKING:
     from pydantic import GetCoreSchemaHandler
     from pydantic_core import CoreSchema
 
+    from air.requests import Request
+
+
+_MIN_SECRET_BYTES = 32
+_MAX_TOKEN_LENGTH = 256
+
+
+def _validate_secret(secret: bytes) -> bytes:
+    """Require enough key material for HMAC-SHA-256.
+
+    Raises:
+        ValueError: If the secret contains fewer than 32 bytes.
+    """
+    if not secret:
+        msg = "CSRF secret must not be empty."
+        raise ValueError(msg)
+    if len(secret) < _MIN_SECRET_BYTES:
+        msg = f"CSRF secret must be at least {_MIN_SECRET_BYTES} bytes."
+        raise ValueError(msg)
+    return secret
+
 
 def _initial_secret() -> bytes | None:
     """Load the configured secret or safely initialize a process-local one."""
-    if configured_secret := os.environ.get("AIRFORM_SECRET", "").encode():
-        return configured_secret
+    configured_secret = os.environ.get("AIRFORM_SECRET")
+    if configured_secret is not None:
+        return _validate_secret(configured_secret.encode())
     if sys.platform == "emscripten":
         return None
     return secrets.token_bytes(32)
@@ -54,22 +77,20 @@ def configure_csrf_secret(secret: str | bytes) -> None:
     signed with the previous secret.
 
     Args:
-        secret: A non-empty text or byte secret.
+        secret: A text or byte secret containing at least 32 bytes.
 
     Raises:
         TypeError: If ``secret`` is not text or bytes.
-        ValueError: If ``secret`` is empty.
-    """
+        ValueError: If ``secret`` contains fewer than 32 bytes.
+    """  # noqa: DOC502
     if isinstance(secret, str):
         normalized_secret = secret.encode()
     elif isinstance(secret, bytes):
         normalized_secret = secret
     else:
         msg = "CSRF secret must be str or bytes."
         raise TypeError(msg)
-    if not normalized_secret:
-        msg = "CSRF secret must not be empty."
-        raise ValueError(msg)
+    _validate_secret(normalized_secret)
 
     global _SECRET
     _SECRET = normalized_secret
@@ -90,6 +111,10 @@ def _check_csrf_token(token: str, max_age: int = CSRF_MAX_AGE) -> str:
     Raises:
         ValueError: If the token is missing, tampered, or expired.
     """
+    if not isinstance(token, str) or len(token) > _MAX_TOKEN_LENGTH:
+        msg = "Invalid CSRF token."
+        raise ValueError(msg)
+
     parts = token.split(":")
     if len(parts) != 3:
         msg = "Invalid CSRF token."
@@ -116,6 +141,49 @@ def _check_csrf_token(token: str, max_age: int = CSRF_MAX_AGE) -> str:
     return token
 
 
+def _url_origin(url: str, *, allow_path: bool) -> tuple[str, str, int] | None:
+    """Return a normalized HTTP origin, or None for malformed input."""
+    if any(char.isspace() for char in url):
+        return None
+    try:
+        parsed = urlsplit(url)
+        hostname = parsed.hostname
+        port = parsed.port
+    except ValueError:
+        return None
+    if (
+        parsed.scheme not in {"http", "https"}
+        or hostname is None
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.fragment
+        or (not allow_path and (parsed.path or parsed.query))
+    ):
+        return None
+    if port is None:
+        port = 443 if parsed.scheme == "https" else 80
+    return parsed.scheme, hostname, port
+
+
+def _check_csrf_origin(request: Request) -> None:
+    """Require a browser submission to originate from the request target.
+
+    Raises:
+        ValueError: If the source origin is missing, malformed, or different.
+    """
+    target_origin = _url_origin(str(request.url), allow_path=True)
+    origin_headers = request.headers.getlist("origin")
+    if origin_headers:
+        source_origin = _url_origin(origin_headers[0], allow_path=False) if len(origin_headers) == 1 else None
+    else:
+        referer_headers = request.headers.getlist("referer")
+        source_origin = _url_origin(referer_headers[0], allow_path=True) if len(referer_headers) == 1 else None
+
+    if source_origin is None or target_origin is None or source_origin != target_origin:
+        msg = "CSRF source origin does not match the request origin."
+        raise ValueError(msg)
+
+
 def _get_secret() -> bytes:
     """Return the configured secret, generating a process-local one on first use."""
     global _SECRET

src/air/form/main.py

@@ -469,6 +469,7 @@ def __init__(self, initial_data: dict | None = None) -> None:
         self.initial_data = initial_data
         self.submitted_data: dict | None = None
         self._csrf_token: str | None = None
+        self._csrf_request: Request | None = None
 
     @property
     def data(self) -> M:
@@ -490,7 +491,8 @@ async def __call__(self, form_data: Mapping[str, Any]) -> Self:
     async def from_request(cls, request: Request) -> Self:
         """Create and validate an AirForm instance from a request.
 
-        CSRF is always enforced for browser submissions.
+        CSRF is always enforced for browser submissions. The request's Origin,
+        or Referer when Origin is absent, must match the request target origin.
 
         Args:
             request: An object with an async ``form()`` method.
@@ -499,6 +501,7 @@ async def from_request(cls, request: Request) -> Self:
         self = cls()
         # A browser submission came from a rendered form, enforce CSRF
         self._csrf_token = "from_request"
+        self._csrf_request = request
         self.validate(dict(form_data))
         return self
 
@@ -515,7 +518,7 @@ def validate(self, form_data: Mapping[str, Any]) -> bool:
         Raises:
             ValueError: If the CSRF token is missing, tampered, or expired.
         """
-        from air.form.csrf import CSRF_FIELD_NAME, _check_csrf_token  # noqa: PLC0415
+        from air.form.csrf import CSRF_FIELD_NAME, _check_csrf_origin, _check_csrf_token  # noqa: PLC0415
 
         self._data = None
         self.is_valid = False
@@ -526,6 +529,8 @@ def validate(self, form_data: Mapping[str, Any]) -> bool:
         if self._csrf_token is not None:
             raw_token = self.submitted_data.pop(CSRF_FIELD_NAME, None)
             try:
+                if self._csrf_request is not None:
+                    _check_csrf_origin(self._csrf_request)
                 if raw_token is None:
                     msg = "CSRF token is missing."
                     raise ValueError(msg)  # noqa: TRY301

tests/test_csrf.py

@@ -1,10 +1,26 @@
 import pytest
+from starlette.requests import Request
 
 import air
 import air.form.csrf as csrf
 from air.form import configure_csrf_secret
 
 
+def _request_with_headers(*headers: tuple[str, str]) -> Request:
+    raw_headers = [(b"host", b"example.test")]
+    raw_headers.extend((name.lower().encode(), value.encode()) for name, value in headers)
+    return Request({
+        "type": "http",
+        "method": "POST",
+        "scheme": "https",
+        "server": ("example.test", 443),
+        "path": "/submit",
+        "raw_path": b"/submit",
+        "query_string": b"",
+        "headers": raw_headers,
+    })
+
+
 def test_initial_secret_is_eager_on_threaded_runtimes(monkeypatch: pytest.MonkeyPatch) -> None:
     generated_secret = b"n" * 32
     calls: list[int] = []
@@ -35,17 +51,25 @@ def unexpected_entropy(_: int) -> bytes:
 
 
 def test_initial_secret_prefers_configuration_on_emscripten(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setenv("AIRFORM_SECRET", "configured")
+    monkeypatch.setenv("AIRFORM_SECRET", "c" * 32)
     monkeypatch.setattr(csrf.sys, "platform", "emscripten")
 
-    assert csrf._initial_secret() == b"configured"
+    assert csrf._initial_secret() == b"c" * 32
+
+
+@pytest.mark.parametrize("secret", ["", "short"])
+def test_initial_secret_rejects_weak_configuration(monkeypatch: pytest.MonkeyPatch, secret: str) -> None:
+    monkeypatch.setenv("AIRFORM_SECRET", secret)
+
+    with pytest.raises(ValueError, match=r"must not be empty|at least 32 bytes"):
+        csrf._initial_secret()
 
 
 @pytest.mark.parametrize(
     ("secret", "expected"),
     [
-        pytest.param("configured", b"configured", id="text-secret"),
-        pytest.param(b"configured", b"configured", id="byte-secret"),
+        pytest.param("t" * 32, b"t" * 32, id="text-secret"),
+        pytest.param(b"b" * 32, b"b" * 32, id="byte-secret"),
     ],
 )
 def test_configure_csrf_secret_accepts_text_or_bytes(
@@ -77,5 +101,64 @@ def test_configure_csrf_secret_rejects_other_types() -> None:
         configure_csrf_secret(123)
 
 
+def test_configure_csrf_secret_rejects_short_values() -> None:
+    with pytest.raises(ValueError, match="at least 32 bytes"):
+        configure_csrf_secret("predictable")
+
+
+def test_configured_secret_round_trip_and_rotation(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(csrf, "_SECRET", b"a" * 32)
+    token = csrf.generate_csrf_token()
+
+    assert csrf._check_csrf_token(token) == token
+
+    configure_csrf_secret(b"b" * 32)
+    with pytest.raises(ValueError, match="Invalid CSRF token"):
+        csrf._check_csrf_token(token)
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [
+        pytest.param((("Origin", "https://example.test"),), id="origin"),
+        pytest.param((("Origin", "https://example.test:443"),), id="explicit-default-port"),
+        pytest.param((("Referer", "https://example.test/form?step=2"),), id="referer-fallback"),
+    ],
+)
+def test_check_csrf_origin_accepts_exact_origin(headers: tuple[tuple[str, str], ...]) -> None:
+    csrf._check_csrf_origin(_request_with_headers(*headers))
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [
+        pytest.param((), id="missing-source"),
+        pytest.param((("Origin", "null"),), id="null-origin"),
+        pytest.param((("Origin", "http://example.test"),), id="different-scheme"),
+        pytest.param((("Origin", "https://attacker.example"),), id="cross-origin"),
+        pytest.param((("Origin", "https://sibling.example.test"),), id="sibling-subdomain"),
+        pytest.param((("Origin", "https://example.test.attacker.example"),), id="hostname-suffix"),
+        pytest.param((("Origin", "https://example.test:444"),), id="different-port"),
+        pytest.param((("Origin", "https://example.test:0"),), id="explicit-zero-port"),
+        pytest.param((("Origin", "https://user@example.test"),), id="userinfo"),
+        pytest.param((("Origin", "https://example.test/path"),), id="origin-with-path"),
+        pytest.param(
+            (("Origin", "https://example.test"), ("Origin", "https://attacker.example")),
+            id="multiple-origins",
+        ),
+        pytest.param((("Referer", "https://attacker.example/form"),), id="cross-origin-referer"),
+    ],
+)
+def test_check_csrf_origin_rejects_nonmatching_source(headers: tuple[tuple[str, str], ...]) -> None:
+    with pytest.raises(ValueError, match="source origin"):
+        csrf._check_csrf_origin(_request_with_headers(*headers))
+
+
+@pytest.mark.parametrize("token", [b"not-text", "x" * 257])
+def test_check_csrf_token_rejects_invalid_types_and_oversized_values(token: object) -> None:
+    with pytest.raises(ValueError, match="Invalid CSRF token"):
+        csrf._check_csrf_token(token)
+
+
 def test_configure_csrf_secret_is_exported_from_air() -> None:
     assert air.configure_csrf_secret is configure_csrf_secret