"""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}