feat(ansible): runs stay inside their source; controller-side vars are admin-only (#5034)
#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>
This commit is contained in:
+32
-15
@@ -13,28 +13,45 @@ _REPO = pathlib.Path(steward.__file__).parent.parent
|
||||
_ROUTE_METHODS = ("get", "post", "put", "patch", "delete", "route", "websocket")
|
||||
|
||||
|
||||
def _route_decorated_views():
|
||||
files = [p for root in ("steward", "plugins") for p in (_REPO / root).rglob("*.py")
|
||||
if "migrations" not in p.parts]
|
||||
for path in files:
|
||||
tree = ast.parse(path.read_text())
|
||||
for node in ast.walk(tree):
|
||||
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 or f'.{m}("/' in d for d in decorators for m in _ROUTE_METHODS):
|
||||
yield path, node, decorators
|
||||
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)}:{n.lineno} {n.name}"
|
||||
for p, n, decs in _route_decorated_views()
|
||||
if not any(d.startswith(("require_role(", "public_route(")) for d in decs)]
|
||||
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 _, n, _ in _route_decorated_views()}
|
||||
names = {n.name for _, text in _sources() for n, _ in route_views(text)}
|
||||
assert {"login", "health", "ingest", "widget_table", "public_share"} <= names
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user