bvandeusenandClaude Opus 5.5
96126b1c83
#5031 "Forged-request protection: SameSite cookie, per-session CSRF token on every non-GET, HTMX header". A page on another site could make a logged-in browser POST to Steward with the session cookie attached. - steward/auth/csrf.py: a per-session token, checked in one before_request for every POST/PUT/PATCH/DELETE (header X-CSRF-Token or form field csrf_token, constant-time compare). The bearer-authenticated host-agent ingest is exempt; /mcp never reaches Quart. - Every POST form renders {{ csrf_field() }}; base.html carries the token in a meta tag and adds it to every HTMX request; the dashboard's fetch() calls send it. - The session cookie is SameSite=Lax; login issues a fresh token. - PUBLIC_BASE_URL is now loaded from Settings at startup (it was only set when Settings → General was saved). - tests/test_csrf.py: refusal, acceptance, rotation, the ingest exemption, and a guard that every POST form in the templates carries the field. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
"""Forged-request protection: every unsafe browser request carries the
|
|
session's token, and every POST form in the templates sends it."""
|
|
import pathlib
|
|
import re
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
import steward
|
|
from steward.auth import csrf
|
|
|
|
_REPO = pathlib.Path(steward.__file__).parent.parent
|
|
_FORM_TAG = re.compile(r"<form\b[^>]*>", re.S | re.I)
|
|
_POST = re.compile(r"""\bmethod\s*=\s*["']?post\b""", re.I)
|
|
|
|
|
|
def test_every_post_form_carries_the_token():
|
|
files = sorted((_REPO / "steward" / "templates").rglob("*.html")) + \
|
|
sorted((_REPO / "plugins").rglob("templates/**/*.html"))
|
|
missing = []
|
|
for f in files:
|
|
text = f.read_text()
|
|
for m in _FORM_TAG.finditer(text):
|
|
if _POST.search(m.group(0)) and \
|
|
not text[m.end():].lstrip().startswith("{{ csrf_field() }}"):
|
|
line = text.count("\n", 0, m.start()) + 1
|
|
missing.append(f"{f.relative_to(_REPO)}:{line}")
|
|
assert not missing, "POST forms without {{ csrf_field() }}: " + ", ".join(missing)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_without_token_is_refused(client):
|
|
response = await client.post("/login", form={"username": "a", "password": "b"})
|
|
assert response.status_code == 400
|
|
assert "did not come from Steward" in await response.get_data(as_text=True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_with_wrong_token_is_refused(client):
|
|
async with client.session_transaction() as sess:
|
|
sess["_csrf"] = "right"
|
|
response = await client.post("/login", form={"username": "a", "password": "b"},
|
|
headers={"X-CSRF-Token": "wrong"})
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_form_field_token_is_accepted_and_rotated_at_login(client):
|
|
from tests.test_auth import make_user
|
|
async with client.session_transaction() as sess:
|
|
sess["_csrf"] = "before-login"
|
|
with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
|
|
patch("steward.auth.routes.get_user_by_username", new=AsyncMock(return_value=make_user())):
|
|
response = await client.post("/login", form={
|
|
"username": "admin", "password": "password", "csrf_token": "before-login"})
|
|
assert response.status_code == 302
|
|
async with client.session_transaction() as sess:
|
|
assert sess.get("_csrf") != "before-login"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_page_renders_token_into_form_and_meta(client):
|
|
page = await (await client.get("/login")).get_data(as_text=True)
|
|
async with client.session_transaction() as sess:
|
|
token = sess["_csrf"]
|
|
assert f'name="csrf_token" value="{token}"' in page
|
|
assert f'<meta name="csrf-token" content="{token}">' in page
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bearer_ingest_is_exempt(app):
|
|
async with app.test_request_context("/plugins/host_agent/ingest", method="POST"):
|
|
assert await csrf.check_request() is None
|
|
async with app.test_request_context("/hosts/", method="POST"):
|
|
assert (await csrf.check_request()).status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_needs_no_token(app):
|
|
async with app.test_request_context("/hosts/", method="GET"):
|
|
assert await csrf.check_request() is None
|