Protect against login CSRF without cookies for anonymous users
Anonymous users must not get cookies or session store entries, so a CSRF token for /login is not possible. Instead, every unsafe request is checked statelessly via Sec-Fetch-Site, falling back to Origin. Requests without both headers are not from a browser and are accepted. The expected origin comes from the new MAX_ORIGIN setting, which is required outside of debug mode. Further CSRF changes: - Do not delete the CSRF token on a failed check. It added no security, but invalidated forms in open tabs and could be triggered cross-site. - Treat HEAD and OPTIONS as safe methods. Keep the login page free of sessions: - Show login messages via ?msg=login_required|logout (fixed keys only) and the login error via the template context instead of flash(). - Only send the session deletion cookie if the client sent a cookie. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
91e95b1947
commit
dc85c578a2
10 files changed
+89
-32
No files matched your search
+46
-15
@@ -3,39 +3,70 @@ from functools import wraps
|
||||
from flask import request, session
|
||||
|
||||
from max.crypto import constant_time_compare, get_random_string
|
||||
from max.middlewares import BaseMiddleware
|
||||
from max.middlewares import BaseMiddleware, get_real_remote_addr
|
||||
|
||||
CSRF_TOKEN_KEY = "csrf_token"
|
||||
CSRF_TOKEN_LENGTH = 32
|
||||
SAFE_METHODS = ("GET", "HEAD", "OPTIONS")
|
||||
# "none": navigation started by the user (e.g. reload). "same-site" is rejected,
|
||||
# since other services on a subdomain must not trigger requests.
|
||||
ALLOWED_FETCH_SITES = ("same-origin", "none")
|
||||
|
||||
|
||||
class CsrfMiddleware(BaseMiddleware):
|
||||
def __call__(self, *args, **kwargs):
|
||||
error = self.process_request()
|
||||
if error is not None:
|
||||
# clear csrf from session
|
||||
if CSRF_TOKEN_KEY in session:
|
||||
del session[CSRF_TOKEN_KEY]
|
||||
return error, 400
|
||||
return error
|
||||
|
||||
return self.handler(*args, **kwargs)
|
||||
|
||||
def process_request(self):
|
||||
"""If a string is returned, it mist be used as a 400er error"""
|
||||
if getattr(self.handler, "no_csrf", False):
|
||||
"""If a (message, status) tuple is returned, it must be used as the response"""
|
||||
no_csrf = getattr(self.handler, "no_csrf", False)
|
||||
|
||||
if request.method in SAFE_METHODS:
|
||||
if not no_csrf and CSRF_TOKEN_KEY not in session:
|
||||
session[CSRF_TOKEN_KEY] = get_random_string(CSRF_TOKEN_LENGTH)
|
||||
return
|
||||
|
||||
session_has_csrf_token = CSRF_TOKEN_KEY in session
|
||||
method_is_unsafe = request.method.lower() != "get"
|
||||
# Stateless check, so it also protects routes without a session (login)
|
||||
if not self.is_origin_allowed():
|
||||
self.app.logger.debug(
|
||||
'Blocked cross-origin request %s %s origin="%s" sec_fetch_site="%s" from %s',
|
||||
request.method,
|
||||
request.path,
|
||||
request.headers.get("Origin"),
|
||||
request.headers.get("Sec-Fetch-Site"),
|
||||
get_real_remote_addr(self.app),
|
||||
)
|
||||
return "Cross-origin request blocked", 403
|
||||
|
||||
if not session_has_csrf_token and method_is_unsafe:
|
||||
return "CSRF token is missing"
|
||||
if no_csrf:
|
||||
return
|
||||
|
||||
elif not session_has_csrf_token and not method_is_unsafe:
|
||||
session[CSRF_TOKEN_KEY] = get_random_string(CSRF_TOKEN_LENGTH)
|
||||
if CSRF_TOKEN_KEY not in session:
|
||||
return "CSRF token is missing", 400
|
||||
|
||||
elif session_has_csrf_token and method_is_unsafe:
|
||||
return self.check_token(session[CSRF_TOKEN_KEY], request.form)
|
||||
error = self.check_token(session[CSRF_TOKEN_KEY], request.form)
|
||||
if error is not None:
|
||||
return error, 400
|
||||
|
||||
def is_origin_allowed(self):
|
||||
fetch_site = request.headers.get("Sec-Fetch-Site")
|
||||
if fetch_site is not None:
|
||||
return fetch_site in ALLOWED_FETCH_SITES
|
||||
|
||||
origin = request.headers.get("Origin")
|
||||
if origin is None:
|
||||
return True # not a browser, so no CSRF possible
|
||||
return origin == self.get_expected_origin()
|
||||
|
||||
def get_expected_origin(self):
|
||||
origin = self.app.config["MAX_ORIGIN"]
|
||||
if origin is None:
|
||||
origin = request.host_url.rstrip("/")
|
||||
return origin
|
||||
|
||||
def check_token(self, session_token, formdata):
|
||||
if CSRF_TOKEN_KEY not in formdata:
|
||||
|
||||
Reference in new issue
Block a user