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.6 KiB
Python
92 lines
3.6 KiB
Python
"""Who can reach a route: every view carries require_role or public_route,
|
|
an unmarked one is refused, and a share link opens only its own widgets."""
|
|
import ast
|
|
import pathlib
|
|
|
|
import pytest
|
|
from werkzeug.datastructures import MultiDict
|
|
|
|
import steward
|
|
from steward.dashboard.share import share_allows
|
|
|
|
_REPO = pathlib.Path(steward.__file__).parent.parent
|
|
_ROUTE_METHODS = ("get", "post", "put", "patch", "delete", "route", "websocket")
|
|
|
|
|
|
def route_views(source: str):
|
|
"""(function node, decorator texts) for each route-decorated function."""
|
|
for node in ast.walk(ast.parse(source)):
|
|
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
continue
|
|
decorators = [ast.unparse(d) for d in node.decorator_list]
|
|
if any(f".{m}('/" in d for d in decorators for m in _ROUTE_METHODS):
|
|
yield node, decorators
|
|
|
|
|
|
def unmarked_routes(source: str) -> list[str]:
|
|
return [n.name for n, decs in route_views(source)
|
|
if not any(d.startswith(("require_role(", "public_route(")) for d in decs)]
|
|
|
|
|
|
def _sources():
|
|
for root in ("steward", "plugins"):
|
|
for path in (_REPO / root).rglob("*.py"):
|
|
if "migrations" not in path.parts:
|
|
yield path, path.read_text()
|
|
|
|
|
|
def test_the_route_guard_can_fail():
|
|
source = (
|
|
"@bp.get('/open')\nasync def open_view(): ...\n"
|
|
"@bp.post('/closed')\n@require_role(UserRole.viewer)\nasync def closed(): ...\n"
|
|
"@bp.get('/login')\n@public_route('login')\nasync def login(): ...\n"
|
|
"def helper(): ...\n")
|
|
assert unmarked_routes(source) == ["open_view"]
|
|
|
|
|
|
def test_every_route_says_who_may_reach_it():
|
|
unmarked = [f"{p.relative_to(_REPO)} {name}" for p, text in _sources()
|
|
for name in unmarked_routes(text)]
|
|
assert not unmarked, "routes with neither require_role nor public_route: " + ", ".join(unmarked)
|
|
|
|
|
|
def test_the_scan_sees_routes():
|
|
names = {n.name for _, text in _sources() for n, _ in route_views(text)}
|
|
assert {"login", "health", "ingest", "widget_table", "public_share"} <= names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unmarked_view_is_refused(app, client):
|
|
async def unmarked():
|
|
return "open"
|
|
app.add_url_rule("/__unmarked", "unmarked", unmarked)
|
|
assert (await client.get("/__unmarked")).status_code == 403
|
|
assert (await client.get("/health")).status_code == 200
|
|
|
|
|
|
WIDGETS = ["/plugins/host_agent/widget/history?host_id=h1&hours=6&wid=7",
|
|
"/monitors/widget?wid=8"]
|
|
|
|
|
|
@pytest.mark.parametrize("path,args,allowed", [
|
|
("/plugins/host_agent/widget/history", {"host_id": "h1", "hours": "6", "wid": "7", "s": "t"}, True),
|
|
("/plugins/host_agent/widget/history", {"host_id": "h1", "hours": "24", "wid": "7", "s": "t"}, True),
|
|
("/monitors/widget", {"wid": "8", "s": "t"}, True),
|
|
# the same widget aimed at another host
|
|
("/plugins/host_agent/widget/history", {"host_id": "h2", "hours": "6", "wid": "7", "s": "t"}, False),
|
|
# a parameter the widget was not configured with
|
|
("/monitors/widget", {"wid": "8", "host_id": "h1", "s": "t"}, False),
|
|
# a configured parameter left off
|
|
("/plugins/host_agent/widget/history", {"hours": "6", "wid": "7", "s": "t"}, False),
|
|
# any other page
|
|
("/hosts/", {"s": "t"}, False),
|
|
("/plugins/docker/logs", {"wid": "8", "s": "t"}, False),
|
|
])
|
|
def test_share_opens_only_its_widgets(path, args, allowed):
|
|
assert share_allows(WIDGETS, path, MultiDict(args)) is allowed
|
|
|
|
|
|
def test_share_refuses_a_repeated_parameter():
|
|
args = MultiDict([("wid", "8"), ("wid", "9"), ("s", "t")])
|
|
assert not share_allows(WIDGETS, "/monitors/widget", args)
|