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:
@@ -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"]
|
||||
Reference in New Issue
Block a user