FabledSteward/steward/ansible/guards.py

"""What a playbook run may be pointed at, and which variables only an admin
may set.

Ansible runs inside the Steward container, so a run is code on the Steward
server unless it is held to playbooks its source lists and to variables that
only shape what happens on the target. An operator may run playbooks and set
their variables. Three things would let one run commands on Steward itself,
and are refused for anyone but an admin:

- a playbook or inventory path outside what the source lists;
- an `ansible_*` variable that changes how or where Ansible connects
  (`ansible_connection: local`, `ansible_ssh_common_args` with a
  ProxyCommand, an interpreter or executable path);
- a Jinja expression in any value (`{{ lookup('pipe', …) }}` is evaluated
  on the controller when the value is used).

Playbook files themselves are admin-only to write (ansible/routes.py).
"""
from __future__ import annotations

from steward.ansible import sources as src_module
from steward.models.users import UserRole

# The ansible_* variables an operator may set: which address, port and user
# to reach a target as, and whether to become another user there.
OPERATOR_ANSIBLE_VARS = frozenset({
    "ansible_host", "ansible_port", "ansible_user", "ansible_password",
    "ansible_become", "ansible_become_user", "ansible_become_password",
})
_JINJA = ("{{", "{%", "{#")


def check_playbook(source: dict, playbook_path: str) -> str | None:
    """None when the source lists this playbook, else why it is refused."""
    if playbook_path not in src_module.discover_playbooks(source["path"]):
        return f"Source {source['name']!r} has no playbook {playbook_path!r}"
    return None


def check_inventory_scope(source: dict, scope: str) -> str | None:
    """None for a Steward scope or an inventory file the source has, else why not."""
    if scope.startswith("steward:"):
        return None
    if scope.startswith("repo:"):
        parts = scope.split(":", 2)
        path = parts[2] if len(parts) == 3 else ""
        if path in src_module.discover_inventories(source["path"]):
            return None
        return f"Source {source['name']!r} has no inventory file {path!r}"
    return "Invalid inventory_scope"


def _has_jinja(value) -> bool:
    if isinstance(value, str):
        return any(mark in value for mark in _JINJA)
    if isinstance(value, dict):
        return any(_has_jinja(k) or _has_jinja(v) for k, v in value.items())
    if isinstance(value, (list, tuple)):
        return any(_has_jinja(v) for v in value)
    return False


def check_vars_for_role(variables: dict | None, role: UserRole | None) -> str | None:
    """None when `role` may set these variables, else a message naming what
    only an admin may set. role=None is treated as the least privileged."""
    if role == UserRole.admin or not variables:
        return None
    names = sorted(n for n in variables
                   if str(n).startswith("ansible_") and n not in OPERATOR_ANSIBLE_VARS)
    if names:
        return (f"Only an admin can set {', '.join(names)}: these change how or where "
                "Ansible connects, and can run commands on the Steward server itself.")
    templated = sorted(str(k) for k, v in variables.items() if _has_jinja(k) or _has_jinja(v))
    if templated:
        return (f"Only an admin can set a value containing {{{{ }}}}, {{% %}} or {{# #}} "
                f"({', '.join(templated)}): Ansible evaluates it on the Steward server.")
    return None



def changed_vars(new: dict | None, old: dict | None) -> dict:
    """The entries of `new` that are added or differ from `old`. An edit is
    held to the editor's role only for what it changes, so an operator can
    still edit a group whose connection vars an admin set."""
    old = old or {}
    return {k: v for k, v in (new or {}).items() if k not in old or old[k] != v}