fix(csrf): clarify rejection diagnostics

This commit is contained in:
Frank Song
2026-05-24 19:48:42 +08:00
committed by nesquena-hermes
parent b6f7412b53
commit 67a204773e
2 changed files with 111 additions and 6 deletions

View File

@@ -1260,8 +1260,37 @@ def _csrf_exempt_path(path: str) -> bool:
return path in {"/api/auth/login", "/api/csp-report"}
_CSRF_FAILURE_ATTR = "_hermes_csrf_failure_reason"
def _set_csrf_failure_reason(handler, reason: str) -> bool:
try:
setattr(handler, _CSRF_FAILURE_ATTR, reason)
except Exception:
pass
return False
def _clear_csrf_failure_reason(handler) -> None:
try:
if hasattr(handler, _CSRF_FAILURE_ATTR):
delattr(handler, _CSRF_FAILURE_ATTR)
except Exception:
pass
def _csrf_rejection_error(handler) -> str:
reason = getattr(handler, _CSRF_FAILURE_ATTR, "")
if reason == "origin_mismatch":
return "Cross-origin mismatch - check reverse proxy headers"
if reason == "token_mismatch":
return "Session expired - reload the page"
return "Cross-origin request rejected"
def _check_csrf(handler) -> bool:
"""Reject cross-origin or tokenless authenticated browser unsafe requests."""
_clear_csrf_failure_reason(handler)
origin = handler.headers.get("Origin", "")
referer = handler.headers.get("Referer", "")
host = handler.headers.get("Host", "")
@@ -1271,7 +1300,7 @@ def _check_csrf(handler) -> bool:
# Extract host:port from origin/referer
m = _re.match(r"^https?://([^/]+)", target)
if not m:
return False
return _set_csrf_failure_reason(handler, "origin_mismatch")
origin_host = m.group(1)
origin_scheme = m.group(0).split('://')[0].lower() # 'http' or 'https'
origin_name, origin_port = _normalize_host_port(origin_host)
@@ -1299,7 +1328,7 @@ def _check_csrf(handler) -> bool:
origin_allowed = True
break
if not origin_allowed:
return False
return _set_csrf_failure_reason(handler, "origin_mismatch")
from api.auth import CSRF_HEADER_NAME, is_auth_enabled, parse_cookie, verify_csrf_token
@@ -1307,7 +1336,9 @@ def _check_csrf(handler) -> bool:
return True
cookie_val = parse_cookie(handler)
submitted = handler.headers.get(CSRF_HEADER_NAME) or handler.headers.get("X-CSRF-Token")
return verify_csrf_token(cookie_val or "", submitted or "")
if verify_csrf_token(cookie_val or "", submitted or ""):
return True
return _set_csrf_failure_reason(handler, "token_mismatch")
def _client_ip_for_rate_limit(handler) -> str:
@@ -4616,7 +4647,7 @@ def handle_post(handler, parsed) -> bool:
diag.stage("csrf")
if not _csrf_exempt_path(parsed.path) and not _check_csrf(handler):
try:
return j(handler, {"error": "Cross-origin request rejected"}, status=403)
return j(handler, {"error": _csrf_rejection_error(handler)}, status=403)
finally:
if diag:
diag.finish()
@@ -6196,7 +6227,7 @@ def handle_post(handler, parsed) -> bool:
def handle_patch(handler, parsed) -> bool:
"""Handle all PATCH routes. Returns True if handled, False for 404."""
if not _check_csrf(handler):
return j(handler, {"error": "Cross-origin request rejected"}, status=403)
return j(handler, {"error": _csrf_rejection_error(handler)}, status=403)
body = read_body(handler)
if parsed.path.startswith("/api/kanban/"):
from api.kanban_bridge import handle_kanban_patch
@@ -6211,7 +6242,7 @@ def handle_patch(handler, parsed) -> bool:
def handle_delete(handler, parsed) -> bool:
"""Handle all DELETE routes. Returns True if handled, False for 404."""
if not _check_csrf(handler):
return j(handler, {"error": "Cross-origin request rejected"}, status=403)
return j(handler, {"error": _csrf_rejection_error(handler)}, status=403)
body = read_body(handler)
if parsed.path.startswith("/api/kanban/"):
from api.kanban_bridge import handle_kanban_delete

View File

@@ -0,0 +1,74 @@
"""Regression tests for issue #2572 CSRF rejection diagnostics."""
import hmac
import io
import json
import time
from types import SimpleNamespace
import api.auth as auth
import api.routes as routes
class _FakeHandler:
def __init__(self, headers=None, body=b"{}"):
self.headers = headers or {}
self.client_address = ("127.0.0.1", 12345)
self.rfile = io.BytesIO(body)
self.wfile = io.BytesIO()
self.status = None
self.sent_headers = {}
def send_response(self, status):
self.status = status
def send_header(self, key, value):
self.sent_headers[key] = value
def end_headers(self):
pass
def _signed_cookie(raw_token: str) -> str:
sig = hmac.new(auth._signing_key(), raw_token.encode(), "sha256").hexdigest()
auth._sessions[raw_token] = time.time() + 60
return f"{raw_token}.{sig}"
def _json_body(handler: _FakeHandler) -> dict:
return json.loads(handler.wfile.getvalue().decode("utf-8"))
def test_origin_mismatch_csrf_rejection_has_diagnostic_error(monkeypatch):
monkeypatch.setattr(auth, "is_auth_enabled", lambda: False)
handler = _FakeHandler(
{
"Origin": "https://evil.example",
"Host": "127.0.0.1:8787",
}
)
routes.handle_post(handler, SimpleNamespace(path="/api/providers/delete"))
assert handler.status == 403
assert _json_body(handler)["error"] == "Cross-origin mismatch - check reverse proxy headers"
def test_token_mismatch_csrf_rejection_has_reload_error(monkeypatch):
cookie = _signed_cookie("z" * 64)
monkeypatch.setattr(auth, "is_auth_enabled", lambda: True)
try:
handler = _FakeHandler(
{
"Origin": "http://127.0.0.1:8787",
"Host": "127.0.0.1:8787",
"Cookie": f"{auth.COOKIE_NAME}={cookie}",
}
)
routes.handle_post(handler, SimpleNamespace(path="/api/providers/delete"))
assert handler.status == 403
assert _json_body(handler)["error"] == "Session expired - reload the page"
finally:
auth._sessions.pop("z" * 64, None)