diff --git a/src/frontend/flask_app/app.py b/src/frontend/flask_app/app.py index 6d78580..1bf972d 100644 --- a/src/frontend/flask_app/app.py +++ b/src/frontend/flask_app/app.py @@ -5,9 +5,9 @@ import re from datetime import date from urllib.parse import urlparse -from flask import Flask, redirect, url_for, session, request -from flask_babel import Babel -from flask_wtf.csrf import CSRFProtect +from flask import Flask, flash, jsonify, redirect, url_for, session, request +from flask_babel import Babel, gettext as _ +from flask_wtf.csrf import CSRFError, CSRFProtect from markupsafe import Markup, escape from werkzeug.middleware.proxy_fix import ProxyFix @@ -101,6 +101,36 @@ def create_app() -> Flask: referrer = None 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") def rich_text_filter(value): """Render a task description keeping its line breaks and its bold. diff --git a/src/frontend/flask_app/config.py b/src/frontend/flask_app/config.py index 0e09b8d..3cb396c 100644 --- a/src/frontend/flask_app/config.py +++ b/src/frontend/flask_app/config.py @@ -20,7 +20,10 @@ class Config: SESSION_COOKIE_SAMESITE = "Lax" SESSION_COOKIE_SECURE = not DEBUG # Only secure cookies in production (HTTPS) 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 # code it belongs to. Empty/None means "not configured". diff --git a/src/frontend/flask_app/tests/test_csrf_errorhandler.py b/src/frontend/flask_app/tests/test_csrf_errorhandler.py new file mode 100644 index 0000000..c4c642e --- /dev/null +++ b/src/frontend/flask_app/tests/test_csrf_errorhandler.py @@ -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"]