fix(csrf): token valido quanto la sessione, errorhandler JSON per fetch (#12)

WTF_CSRF_TIME_LIMIT=None: la pagina di misura resta aperta ore, il token a
1h scadeva a metà turno. Aggiunto errorhandler(CSRFError) in app.py: risposte
JSON/fetch (Accept, Content-Type, X-Requested-With) ricevono 400 JSON con
csrf_expired, le altre un flash + redirect. Lato B1 (task_execute.html) resta
da gestire client-side, fuori da questo commit.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-21 22:30:39 +02:00
parent 48089bcba6
commit 925ed196dd
3 changed files with 154 additions and 4 deletions
+33 -3
View File
@@ -5,9 +5,9 @@ import re
from datetime import date from datetime import date
from urllib.parse import urlparse from urllib.parse import urlparse
from flask import Flask, redirect, url_for, session, request from flask import Flask, flash, jsonify, redirect, url_for, session, request
from flask_babel import Babel from flask_babel import Babel, gettext as _
from flask_wtf.csrf import CSRFProtect from flask_wtf.csrf import CSRFError, CSRFProtect
from markupsafe import Markup, escape from markupsafe import Markup, escape
from werkzeug.middleware.proxy_fix import ProxyFix from werkzeug.middleware.proxy_fix import ProxyFix
@@ -101,6 +101,36 @@ def create_app() -> Flask:
referrer = None referrer = None
return redirect(referrer or url_for("auth.login")) return redirect(referrer or url_for("auth.login"))
@app.errorhandler(CSRFError)
def handle_csrf_error(error):
"""CSRF token expired or missing (#12): a stale page, not a network failure.
WTF_CSRF_TIME_LIMIT is None so a token only goes bad with the session
itself, but a tab left open across a forced logout (or missing token)
still lands here. Fetch/JSON callers get a body they can branch on
instead of an HTML error page that breaks `response.json()`; page
loads get a flash and a redirect back to where they were.
"""
wants_json = (
request.accept_mimetypes.best_match(["application/json", "text/html"])
== "application/json"
or (request.content_type or "").startswith("application/json")
or request.headers.get("X-Requested-With") == "XMLHttpRequest"
)
if wants_json:
return (
jsonify({"error": True, "csrf_expired": True, "detail": error.description}),
400,
)
flash(_("Sessione scaduta: effettua nuovamente l'accesso."), "error")
referrer = request.referrer
if referrer:
parsed = urlparse(referrer)
if parsed.netloc and parsed.netloc != request.host:
referrer = None
return redirect(referrer or url_for("auth.login"))
@app.template_filter("rich_text") @app.template_filter("rich_text")
def rich_text_filter(value): def rich_text_filter(value):
"""Render a task description keeping its line breaks and its bold. """Render a task description keeping its line breaks and its bold.
+4 -1
View File
@@ -20,7 +20,10 @@ class Config:
SESSION_COOKIE_SAMESITE = "Lax" SESSION_COOKIE_SAMESITE = "Lax"
SESSION_COOKIE_SECURE = not DEBUG # Only secure cookies in production (HTTPS) SESSION_COOKIE_SECURE = not DEBUG # Only secure cookies in production (HTTPS)
PERMANENT_SESSION_LIFETIME = 28800 # 8 hours PERMANENT_SESSION_LIFETIME = 28800 # 8 hours
WTF_CSRF_TIME_LIMIT = 3600 # 1 hour # None = token valid for the life of the session (#12): a measurement page
# stays open for hours between saves, and a 1h-expired token turned every
# late save into a silent "Errore di rete" for the operator.
WTF_CSRF_TIME_LIMIT = None
# Station identity: each deployed client container sets this to the station # Station identity: each deployed client container sets this to the station
# code it belongs to. Empty/None means "not configured". # code it belongs to. Empty/None means "not configured".
@@ -0,0 +1,117 @@
"""Tests for the CSRF error handler (app.py) and the session-length token (#12).
Unlike the rest of the suite these tests need CSRF *enabled*, so they build
their own app instead of using the ``flask_app``/``client`` fixtures from
conftest.py (which force ``WTF_CSRF_ENABLED=False``).
"""
import re
import pytest
@pytest.fixture
def csrf_client(monkeypatch):
"""A test client with CSRF protection actually turned on."""
monkeypatch.setenv("STATION_CODE", "ST-TEST")
import config
import importlib
importlib.reload(config)
from app import create_app
app = create_app()
app.config["TESTING"] = True
app.config["SECRET_KEY"] = "test-secret-key"
# WTF_CSRF_TIME_LIMIT is left at its configured value (None, #12) except
# where a test overrides it to force an expiry.
with app.test_client() as c:
with c.session_transaction() as sess:
sess["api_key"] = "test-api-key"
sess["user"] = {
"id": 1,
"username": "op1",
"roles": ["MeasurementTec"],
"is_admin": False,
"language_pref": "it",
"theme_pref": "light",
"active": True,
}
sess["user_id"] = 1
sess["language"] = "it"
sess["theme"] = "light"
yield app, c
def _get_csrf_token(client):
resp = client.get("/auth/profile")
match = re.search(r'csrf-token" content="([^"]+)"', resp.get_data(as_text=True))
assert match, "no CSRF meta tag found on /auth/profile"
return match.group(1)
class TestCsrfTimeLimit:
def test_token_never_expires_within_the_session(self):
"""#12: WTF_CSRF_TIME_LIMIT is None - the token lives as long as the session."""
from config import Config
assert Config.WTF_CSRF_TIME_LIMIT is None
class TestCsrfErrorHandlerJson:
def test_expired_token_on_json_post_returns_400_json(self, csrf_client):
app, client = csrf_client
app.config["WTF_CSRF_TIME_LIMIT"] = 1
token = _get_csrf_token(client)
import time
time.sleep(2)
resp = client.post(
"/measure/save-measurement",
json={"subtask_id": 1, "version_id": 1, "value": 1.0},
headers={"X-CSRFToken": token},
)
assert resp.status_code == 400
assert resp.content_type.startswith("application/json")
body = resp.get_json()
assert body["error"] is True
assert body["csrf_expired"] is True
def test_missing_token_on_fetch_marked_request_returns_400_json(self, csrf_client):
"""A fetch() call without X-Requested-With still gets JSON if it says so via Accept."""
_app, client = csrf_client
resp = client.post(
"/measure/save-measurement",
json={"subtask_id": 1, "version_id": 1, "value": 1.0},
headers={"Accept": "application/json"},
)
assert resp.status_code == 400
assert resp.content_type.startswith("application/json")
body = resp.get_json()
assert body["csrf_expired"] is True
class TestCsrfErrorHandlerHtml:
def test_expired_token_on_html_post_flashes_and_redirects(self, csrf_client, monkeypatch):
"""A plain form POST (no JSON/XHR markers) gets a flash + redirect, not an error page."""
from unittest.mock import MagicMock, patch
app, client = csrf_client
app.config["WTF_CSRF_TIME_LIMIT"] = 1
token = _get_csrf_token(client)
import time
time.sleep(2)
mock = MagicMock()
mock.put.return_value = {"id": 1, "username": "op1"}
with patch("blueprints.auth.api_client", mock):
resp = client.post(
"/auth/profile",
data={"csrf_token": token, "display_name": "New Name"},
headers={"Referer": "http://localhost/auth/profile"},
follow_redirects=False,
)
assert resp.status_code == 302
assert "/auth" in resp.headers["Location"]