bvandeusenandClaude Opus 5.5
8dad79f17a
#5034 "Playbook runs: path and inventory checks in trigger_run, connection vars admin-only, confirm enforced server-side, web actions audited". Ansible runs inside the Steward container, so an operator who could name any playbook path, set ansible_connection=local or ProxyCommand ssh args, or put {{ lookup('pipe', …) }} in a value could run commands on Steward. - steward/ansible/guards.py: check_playbook (the source must list it), check_inventory_scope (steward:* or an inventory file the source has), check_vars_for_role (ansible_* beyond host/port/user/password/become, and any Jinja in a value, are admin-only), changed_vars (an edit is held to the role only for what it changes). - trigger_run applies the path checks, so every run path — web, MCP, alerts, schedules, plugins — gets them. - Variables are checked by role on the web run form, host-page runs, schedule create/update, inventory group/target create/update, and MCP run_playbook / create_schedule / update_schedule. - A steward:confirm playbook needs the confirmation ticked on the server side for a real (non-check) web run, as MCP already required. - Audited: runs, cancel, schedule create/update/enable/disable/delete/ run-now, playbook create/update/delete, inventory group and target create/update/delete (variable names only, never values). One audit_request helper serves the web routes. - The CSRF form guard and the route guard now each test that they fail on a synthetic broken input (rule: a guard must be able to fail). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
92 lines
3.7 KiB
Python
92 lines
3.7 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 forms_missing_token(text: str) -> list[int]:
|
|
"""Line numbers of POST forms whose first child is not {{ csrf_field() }}."""
|
|
return [text.count("\n", 0, m.start()) + 1 for m in _FORM_TAG.finditer(text)
|
|
if _POST.search(m.group(0))
|
|
and not text[m.end():].lstrip().startswith("{{ csrf_field() }}")]
|
|
|
|
|
|
def test_the_form_guard_can_fail():
|
|
broken = '<form\n method="post" action="/x">\n <input name="a">\n</form>'
|
|
fixed = '<form method="POST" action="/x">{{ csrf_field() }}<input name="a"></form>'
|
|
assert forms_missing_token(broken) == [1]
|
|
assert forms_missing_token(fixed) == []
|
|
assert forms_missing_token('<form method="get">') == []
|
|
|
|
|
|
def test_every_post_form_carries_the_token():
|
|
files = sorted((_REPO / "steward" / "templates").rglob("*.html")) + \
|
|
sorted((_REPO / "plugins").rglob("templates/**/*.html"))
|
|
assert files
|
|
missing = [f"{f.relative_to(_REPO)}:{line}" for f in files
|
|
for line in forms_missing_token(f.read_text())]
|
|
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
|