finnandClaude Opus 5.5
4a159006aa
app.logger.info("message", key=value) appends the keyword arguments as
JSON encoded fields, so callers no longer format the values themselves.
All log calls use fixed messages and put variable values into fields.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
from functools import wraps
|
|
|
|
from flask import request, session
|
|
|
|
from max.crypto import constant_time_compare, get_random_string
|
|
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:
|
|
return error
|
|
|
|
return self.handler(*args, **kwargs)
|
|
|
|
def process_request(self):
|
|
"""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
|
|
|
|
# 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",
|
|
method=request.method,
|
|
path=request.path,
|
|
origin=request.headers.get("Origin"),
|
|
sec_fetch_site=request.headers.get("Sec-Fetch-Site"),
|
|
ip=get_real_remote_addr(self.app),
|
|
)
|
|
return "Cross-origin request blocked", 403
|
|
|
|
if no_csrf:
|
|
return
|
|
|
|
if CSRF_TOKEN_KEY not in session:
|
|
return "CSRF token is missing", 400
|
|
|
|
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:
|
|
return "No CSRF token"
|
|
|
|
request_token = formdata[CSRF_TOKEN_KEY]
|
|
if len(request_token) != CSRF_TOKEN_LENGTH:
|
|
return "Wrong CSRF token"
|
|
elif not constant_time_compare(request_token, session_token):
|
|
return "Wrong CSRF token"
|
|
|
|
|
|
def no_csrf(handler):
|
|
@wraps(handler)
|
|
def wrapped_handler(*args, **kwargs):
|
|
return handler(*args, **kwargs)
|
|
|
|
wrapped_handler.no_csrf = True
|
|
return wrapped_handler
|