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>
76 lines
3.5 KiB
Python
76 lines
3.5 KiB
Python
"""A run is held to what its source lists, and the variables that would run
|
|
commands on the Steward server are admin-only."""
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from steward.ansible import guards, runner
|
|
from steward.models.users import UserRole
|
|
|
|
|
|
@pytest.fixture
|
|
def source(tmp_path):
|
|
(tmp_path / "site.yml").write_text("- hosts: all\n tasks: []\n")
|
|
(tmp_path / "roles").mkdir()
|
|
(tmp_path / "roles" / "deep.yml").write_text("- hosts: all\n")
|
|
(tmp_path / "inventory.yml").write_text("all: {}\n")
|
|
return {"name": "local", "path": str(tmp_path)}
|
|
|
|
|
|
def test_playbook_must_be_listed_by_its_source(source, tmp_path):
|
|
assert guards.check_playbook(source, "site.yml") is None
|
|
assert guards.check_playbook(source, "roles/deep.yml") is None
|
|
(tmp_path.parent / "outside.yml").write_text("- hosts: localhost\n")
|
|
for path in ("../outside.yml", "/etc/passwd", str(tmp_path / "site.yml"), "missing.yml"):
|
|
assert "has no playbook" in guards.check_playbook(source, path)
|
|
|
|
|
|
def test_inventory_scope(source):
|
|
assert guards.check_inventory_scope(source, "steward:all") is None
|
|
assert guards.check_inventory_scope(source, "repo:local:inventory.yml") is None
|
|
assert "no inventory file" in guards.check_inventory_scope(source, "repo:local:../../etc/hosts")
|
|
assert guards.check_inventory_scope(source, "/tmp/inv") == "Invalid inventory_scope"
|
|
|
|
|
|
@pytest.mark.parametrize("variables", [
|
|
{"ansible_host": "10.0.0.5", "ansible_user": "steward", "ansible_become": True},
|
|
{"release": "1.2", "ansible_port": 2222},
|
|
{},
|
|
])
|
|
def test_operator_may_set_target_vars(variables):
|
|
assert guards.check_vars_for_role(variables, UserRole.operator) is None
|
|
|
|
|
|
@pytest.mark.parametrize("variables,word", [
|
|
({"ansible_connection": "local"}, "ansible_connection"),
|
|
({"ansible_ssh_common_args": "-o ProxyCommand=sh -c id"}, "ansible_ssh_common_args"),
|
|
({"ansible_python_interpreter": "/tmp/x"}, "ansible_python_interpreter"),
|
|
({"ansible_user": "{{ lookup('pipe', 'id') }}"}, "ansible_user"),
|
|
({"note": {"nested": ["{% raw %}"]}}, "note"),
|
|
])
|
|
def test_operator_may_not_set_controller_vars(variables, word):
|
|
err = guards.check_vars_for_role(variables, UserRole.operator)
|
|
assert err and word in err and "Only an admin" in err
|
|
assert guards.check_vars_for_role(variables, None) == err
|
|
assert guards.check_vars_for_role(variables, UserRole.admin) is None
|
|
|
|
|
|
def test_only_changes_are_held_to_the_role():
|
|
stored = {"ansible_ssh_common_args": "-J bastion", "ansible_user": "a"}
|
|
edited = {"ansible_ssh_common_args": "-J bastion", "ansible_user": "b"}
|
|
assert guards.changed_vars(edited, stored) == {"ansible_user": "b"}
|
|
assert guards.check_vars_for_role(guards.changed_vars(edited, stored), UserRole.operator) is None
|
|
edited["ansible_ssh_common_args"] = "-o ProxyCommand=x"
|
|
assert guards.check_vars_for_role(guards.changed_vars(edited, stored), UserRole.operator)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_run_refuses_a_path_its_source_does_not_list(source, monkeypatch):
|
|
monkeypatch.setattr(runner.src_module, "get_sources", lambda cfg: [source])
|
|
app = SimpleNamespace(config={})
|
|
run, src, err = await runner.trigger_run(app, source_name="local", playbook_path="../x.yml")
|
|
assert run is None and "has no playbook" in err
|
|
run, src, err = await runner.trigger_run(app, source_name="local", playbook_path="site.yml",
|
|
inventory_scope="repo:local:/etc/hosts")
|
|
assert run is None and "no inventory file" in err
|