from __future__ import annotations import asyncio import logging import os import stat import tempfile from pathlib import Path logger = logging.getLogger(__name__) INVENTORY_NAMES = {"hosts", "inventory", "inventory.yml", "inventory.ini"} def get_sources(ansible_cfg: dict) -> list[dict]: """Return resolved source list from ansible config section. Each source dict has: name, type, path (resolved local path), url (git only), branch (git only), pull_interval_seconds (git only). Config structure in config.yaml:: ansible: cache_dir: /var/cache/steward/ansible sources: - name: my-playbooks type: local path: /opt/playbooks - name: infra-repo type: git url: https://github.com/user/infra.git branch: main pull_interval_seconds: 3600 """ sources = ansible_cfg.get("sources", []) cache_dir = ansible_cfg.get("cache_dir", "/var/cache/steward/ansible") result = [] for src in sources: src_type = src.get("type", "local") if src_type == "git": if not src.get("url"): raise ValueError(f"Ansible git source {src['name']!r} is missing required 'url' field") path = str(Path(cache_dir) / src["name"]) else: if not src.get("path"): raise ValueError(f"Ansible local source {src['name']!r} is missing required 'path' field") path = src.get("path", "") result.append({ "name": src["name"], "type": src_type, "path": path, "url": src.get("url"), "branch": src.get("branch", "main"), "pull_interval_seconds": int(src.get("pull_interval_seconds", 3600)), "http_token": src.get("http_token", "") if src_type == "git" else "", }) return result def discover_playbooks(source_path: str) -> list[str]: """Recursively find .yml and .yaml files in source_path. Returns relative paths.""" root = Path(source_path) if not root.exists(): return [] playbooks = set() for ext in ("*.yml", "*.yaml"): for p in root.rglob(ext): playbooks.add(str(p.relative_to(root))) return sorted(playbooks) def discover_inventories(source_path: str) -> list[str]: """Non-recursive: return inventory filenames present in root of source_path.""" root = Path(source_path) if not root.exists(): return [] return sorted(name for name in INVENTORY_NAMES if (root / name).exists()) def host_inventory_content(host) -> str: """An ephemeral single-host inventory line for a Steward Host. Targets the host's address when set (`ansible_host=`), else just the name (resolved via DNS). Playbooks run against this should use `hosts: all` (or the host's name). Pure — `host` only needs `.name` and `.address`. """ if getattr(host, "address", ""): return f"{host.name} ansible_host={host.address}\n" return f"{host.name}\n" def read_playbook(source_path: str, relative_path: str) -> str | None: """Return contents of a playbook file, or None if not found / path escape.""" root = Path(source_path).resolve() target = (root / relative_path).resolve() # Guard against path traversal try: target.relative_to(root) except ValueError: return None if not target.exists() or not target.is_file(): return None return target.read_text(errors="replace") async def git_pull(source: dict) -> None: """Clone the git repo if absent; pull if already present. When source['http_token'] is set, injects credentials via a temporary GIT_ASKPASS script so they never touch .git/config or process args. """ path = Path(source["path"]) token = source.get("http_token", "") env = None askpass_path = None if token: fd, askpass_path = tempfile.mkstemp(prefix="steward-askpass-", suffix=".sh") try: script = ( "#!/bin/sh\n" 'case "$1" in\n' ' Username*) echo "oauth2" ;;\n' f' Password*) echo "{token}" ;;\n' "esac\n" ) os.write(fd, script.encode()) finally: os.close(fd) os.chmod(askpass_path, stat.S_IRWXU) env = {**os.environ, "GIT_ASKPASS": askpass_path} try: if not (path / ".git").exists(): path.mkdir(parents=True, exist_ok=True) proc = await asyncio.create_subprocess_exec( "git", "clone", "--branch", source["branch"], "--single-branch", source["url"], str(path), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, ) _, stderr = await proc.communicate() if proc.returncode != 0: logger.error( "git clone failed for %r: %s", source["name"], stderr.decode(errors="replace"), ) else: proc = await asyncio.create_subprocess_exec( "git", "-C", str(path), "pull", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, ) _, stderr = await proc.communicate() if proc.returncode != 0: logger.error( "git pull failed for %r: %s", source["name"], stderr.decode(errors="replace"), ) finally: if askpass_path and os.path.exists(askpass_path): os.unlink(askpass_path)