diff --git a/steward/ansible/routes.py b/steward/ansible/routes.py index 61ae3aa..c71e05d 100644 --- a/steward/ansible/routes.py +++ b/steward/ansible/routes.py @@ -56,6 +56,7 @@ async def browse(): "source": source, "playbooks": playbooks, "inventories": inventories, + "editable": src_module.is_editable_source(source), }) async with current_app.db_sessionmaker() as db: @@ -90,6 +91,7 @@ async def view_playbook(source_name: str, playbook_path: str): view_source=source_name, view_path=playbook_path, view_contents=contents, + view_editable=src_module.is_editable_source(source), ) @@ -354,3 +356,102 @@ async def run_schedule_now(schedule_id: str): if run: return redirect(url_for("ansible.run_detail", run_id=run.id)) return redirect(url_for("ansible.schedules")) + + +# ── In-app playbook authoring/editor (admin only) ───────────────────────────── + +def _editable_sources() -> list[dict]: + return [s for s in _get_sources() if src_module.is_editable_source(s)] + + +def _editable_source(name: str) -> dict | None: + return next((s for s in _editable_sources() if s["name"] == name), None) + + +@ansible_bp.get("/playbooks/new") +@require_role(UserRole.admin) +async def new_playbook(): + sources = _editable_sources() + return await render_template( + "ansible/playbook_editor.html", + editable_sources=sources, editing=False, + source_name=(sources[0]["name"] if sources else ""), + playbook_path="", content=_NEW_PLAYBOOK_TEMPLATE, error=None, + ) + + +@ansible_bp.get("/playbooks/edit//") +@require_role(UserRole.admin) +async def edit_playbook(source_name: str, playbook_path: str): + source = _editable_source(source_name) + if source is None: + return "Source is not editable", 404 + content = src_module.read_playbook(source["path"], playbook_path) + if content is None: + return "Playbook not found", 404 + return await render_template( + "ansible/playbook_editor.html", + editable_sources=_editable_sources(), editing=True, + source_name=source_name, playbook_path=playbook_path, + content=content, error=None, + ) + + +@ansible_bp.post("/playbooks/save") +@require_role(UserRole.admin) +async def save_playbook(): + form = await request.form + source_name = (form.get("source_name", "") or "").strip() + playbook_path = (form.get("playbook_path", "") or "").strip() + content = form.get("content", "") or "" + editing = (form.get("editing", "") == "1") + + source = _editable_source(source_name) + err = None + if source is None: + err = "Source is not editable" + elif not playbook_path: + err = "Filename is required" + else: + ok, verr = src_module.validate_playbook_yaml(content) + if not ok: + err = verr + if err is None: + ok, werr = src_module.write_playbook(source["path"], playbook_path, content) + if not ok: + err = werr + + if err is not None: + return await render_template( + "ansible/playbook_editor.html", + editable_sources=_editable_sources(), editing=editing, + source_name=source_name, playbook_path=playbook_path, + content=content, error=err, + ), 400 + + return redirect(url_for( + "ansible.view_playbook", source_name=source_name, playbook_path=playbook_path)) + + +@ansible_bp.post("/playbooks/delete") +@require_role(UserRole.admin) +async def delete_playbook(): + form = await request.form + source = _editable_source((form.get("source_name", "") or "").strip()) + if source is None: + return "Source is not editable", 400 + ok, err = src_module.delete_playbook(source["path"], (form.get("playbook_path", "") or "").strip()) + if not ok: + return err, 400 + return redirect(url_for("ansible.browse")) + + +_NEW_PLAYBOOK_TEMPLATE = """\ +--- +- name: My playbook + hosts: all + gather_facts: false + tasks: + - name: Ping + ansible.builtin.ping: +""" diff --git a/steward/ansible/sources.py b/steward/ansible/sources.py index 04c0c7a..a74cdf9 100644 --- a/steward/ansible/sources.py +++ b/steward/ansible/sources.py @@ -14,6 +14,12 @@ INVENTORY_NAMES = {"hosts", "inventory", "inventory.yml", "inventory.ini"} # inside the app (maintenance tasks, host-agent install). Not operator-editable. BUILTIN_SOURCE_NAME = "steward-builtin" +# Always-present, WRITABLE local source where the in-app editor saves playbooks, +# so a homelab user with no git workflow can author automation. Lives on the +# persistent /data volume; created lazily on first save. +LOCAL_SOURCE_NAME = "steward-local" +USER_PLAYBOOK_DIR = os.environ.get("STEWARD_PLAYBOOK_DIR", "/data/ansible/playbooks") + def _builtin_source() -> dict: return { @@ -27,6 +33,24 @@ def _builtin_source() -> dict: } +def _local_source() -> dict: + return { + "name": LOCAL_SOURCE_NAME, + "type": "local", + "path": USER_PLAYBOOK_DIR, + "url": None, + "branch": "main", + "pull_interval_seconds": 0, + "http_token": "", + } + + +def is_editable_source(source: dict) -> bool: + """Editable in the in-app editor: local dir sources except the read-only + bundled one. Git sources are GitOps (clobbered on pull) so not editable.""" + return source.get("type") == "local" and source.get("name") != BUILTIN_SOURCE_NAME + + def get_sources(ansible_cfg: dict) -> list[dict]: """Return resolved source list from ansible config section. @@ -49,8 +73,8 @@ def get_sources(ansible_cfg: dict) -> list[dict]: """ sources = ansible_cfg.get("sources", []) cache_dir = ansible_cfg.get("cache_dir", "/var/cache/steward/ansible") - # Always expose the bundled first-party playbooks as a source. - result = [_builtin_source()] + # Always expose the bundled (read-only) + local (writable) first-party sources. + result = [_builtin_source(), _local_source()] for src in sources: src_type = src.get("type", "local") if src_type == "git": @@ -119,6 +143,65 @@ def read_playbook(source_path: str, relative_path: str) -> str | None: return target.read_text(errors="replace") +def _resolve_writable(source_path: str, relative_path: str) -> tuple[Path | None, str | None]: + """Resolve relative_path under source_path, guarding traversal + extension. + + Returns (target_path, None) on success or (None, error). The root need not + exist yet (steward-local is created on first save).""" + rel = relative_path.strip().lstrip("/") + if not rel: + return None, "Filename is required" + if not rel.endswith((".yml", ".yaml")): + return None, "Playbook filename must end in .yml or .yaml" + root = Path(source_path).resolve() + target = (root / rel).resolve() + try: + target.relative_to(root) + except ValueError: + return None, "Invalid path" + return target, None + + +def write_playbook(source_path: str, relative_path: str, content: str) -> tuple[bool, str | None]: + """Write a playbook into a source, creating parent dirs. Traversal-guarded.""" + target, err = _resolve_writable(source_path, relative_path) + if err: + return False, err + try: + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + except OSError as exc: + return False, f"Could not write file: {exc}" + return True, None + + +def delete_playbook(source_path: str, relative_path: str) -> tuple[bool, str | None]: + """Delete a playbook from a source. Traversal-guarded; no-op if absent.""" + target, err = _resolve_writable(source_path, relative_path) + if err: + return False, err + try: + if target.exists(): + target.unlink() + except OSError as exc: + return False, f"Could not delete file: {exc}" + return True, None + + +def validate_playbook_yaml(content: str) -> tuple[bool, str | None]: + """Cheap save-time validation: parses as YAML and is a non-empty play list.""" + import yaml + try: + data = yaml.safe_load(content) + except yaml.YAMLError as exc: + return False, f"YAML error: {exc}" + if data is None: + return False, "Playbook is empty" + if not isinstance(data, list): + return False, "A playbook must be a list of plays (top-level YAML list)" + return True, None + + async def git_pull(source: dict) -> None: """Clone the git repo if absent; pull if already present. diff --git a/steward/templates/ansible/browse.html b/steward/templates/ansible/browse.html index 6591d79..bf04039 100644 --- a/steward/templates/ansible/browse.html +++ b/steward/templates/ansible/browse.html @@ -3,14 +3,30 @@ {% block content %}

Browse Playbooks

- ← Run History +
+ {% if session.user_role == 'admin' %} + New playbook + {% endif %} + ← Run History +
{% if view_contents is defined %}

{{ view_path }}

- ← Back to browse +
+ {% if view_editable and session.user_role == 'admin' %} + Edit +
+ + + +
+ {% endif %} + ← Back to browse + 
{{ view_contents }}
@@ -47,6 +63,9 @@ {{ pb }} View + {% if sd.editable and session.user_role == 'admin' %} + Edit + {% endif %} +

{{ "Edit playbook" if editing else "New playbook" }}

+ ← Browse + + +{% if error %}
{{ error }}
{% endif %} + +{% if not editable_sources %} +
+ No writable playbook source is available. The built-in steward-local source should appear + automatically; you can also configure a local-dir source under Settings → Ansible. +
+{% else %} +
+ +
+
+ + {% if editing %} + + + {% else %} + + {% endif %} +
+
+ + +
+
+
+ + +

+ Validated as YAML on save. Runs as the configured Ansible user — admin only. +

+
+
+ + Cancel +
+
+ +{% if editing %} +
+ + + +
+{% endif %} +{% endif %} +{% endblock %} diff --git a/tests/core/test_playbook_editor.py b/tests/core/test_playbook_editor.py new file mode 100644 index 0000000..627b826 --- /dev/null +++ b/tests/core/test_playbook_editor.py @@ -0,0 +1,53 @@ +"""Unit tests for in-app playbook authoring helpers (write/delete/validate/guard).""" +from steward.ansible.sources import ( + BUILTIN_SOURCE_NAME, + LOCAL_SOURCE_NAME, + delete_playbook, + get_sources, + is_editable_source, + read_playbook, + validate_playbook_yaml, + write_playbook, +) + + +def test_write_then_read_roundtrip(tmp_path): + ok, err = write_playbook(str(tmp_path), "maint/x.yml", "- name: p\n hosts: all\n") + assert ok and err is None + assert read_playbook(str(tmp_path), "maint/x.yml").startswith("- name: p") + + +def test_write_rejects_non_yaml_extension(tmp_path): + ok, err = write_playbook(str(tmp_path), "evil.sh", "rm -rf /") + assert not ok and "yml" in err + + +def test_write_blocks_path_traversal(tmp_path): + ok, err = write_playbook(str(tmp_path), "../escape.yml", "- {}") + assert not ok and err == "Invalid path" + + +def test_delete_guarded_and_idempotent(tmp_path): + write_playbook(str(tmp_path), "a.yml", "- {}") + assert delete_playbook(str(tmp_path), "a.yml")[0] is True + assert delete_playbook(str(tmp_path), "a.yml")[0] is True # already gone, no error + ok, err = delete_playbook(str(tmp_path), "../x.yml") + assert not ok and err == "Invalid path" + + +def test_validate_playbook_yaml(): + assert validate_playbook_yaml("- name: p\n hosts: all\n")[0] is True + assert validate_playbook_yaml("foo: [unclosed")[0] is False # malformed YAML + assert validate_playbook_yaml("")[0] is False # empty + assert validate_playbook_yaml("key: value")[0] is False # not a play list + + +def test_is_editable_source(): + assert is_editable_source({"type": "local", "name": LOCAL_SOURCE_NAME}) is True + assert is_editable_source({"type": "local", "name": BUILTIN_SOURCE_NAME}) is False + assert is_editable_source({"type": "git", "name": "repo"}) is False + + +def test_get_sources_includes_writable_local(): + names = [s["name"] for s in get_sources({"sources": []})] + assert LOCAL_SOURCE_NAME in names