ansible_executor/main.py

#!/usr/bin/env python3
import os
import sys
import tempfile
import subprocess
from pathlib import Path

import hvac
from git import Repo
from dotenv import load_dotenv
from prompt_toolkit.shortcuts import checkboxlist_dialog, radiolist_dialog, message_dialog
# from ansible.parsing.dataloader import DataLoader
# from ansible.inventory.manager import InventoryManager

VERSION = "0.0.14"

# ================= LOAD ENV =================
load_dotenv()

VAULT_ADDR = os.getenv("VAULT_ADDR", "https://vault.sectorq.eu")
VAULT_TOKEN = os.getenv("VAULT_TOKEN") or os.environ["VAULT_TOKEN"]

GITLAB_URL = os.getenv("GITLAB_URL", "https://gitlab.sectorq.eu/jaydee/ansible.git")
GITLAB_BRANCH = os.getenv("GITLAB_BRANCH", "main")

VAULT_GITLAB_PATH = os.getenv("VAULT_GITLAB_PATH", "secret/data/gitlab")
VAULT_GITLAB_FIELD = os.getenv("VAULT_GITLAB_FIELD", "token")

VAULT_ANSIBLE_PATH = os.getenv("VAULT_ANSIBLE_SECRET", "secret/data/ansible")
VAULT_ANSIBLE_VAULT_FIELD = os.getenv("VAULT_ANSIBLE_VAULT_FIELD", "vault")
VAULT_ANSIBLE_SSH_FIELD = os.getenv("VAULT_ANSIBLE_SSH_FIELD", "ssh_key")

CLONE_DIR = Path(os.getenv("CLONE_DIR", "/tmp/ansible-repo"))

# ================= VAULT =================
def get_vault_client():
    if not VAULT_ADDR or not VAULT_TOKEN:
        sys.exit("Vault config missing in .env")
    client = hvac.Client(url=VAULT_ADDR, token=VAULT_TOKEN)
    if not client.is_authenticated():
        sys.exit("Vault authentication failed")
    return client

def read_kv2_secret(client, path, field):
    clean_path = path.replace("secret/data/", "")
    secret = client.secrets.kv.v2.read_secret_version(path=clean_path)
    return secret["data"]["data"][field]

# ================= GIT =================
def clone_or_update_repo(repo_url, branch, token):
    repo_url_auth = repo_url.replace("https://", f"https://oauth2:{token}@")
    if CLONE_DIR.exists():
        print("Updating repository...")
        repo = Repo(CLONE_DIR)
        repo.git.checkout(branch)
        repo.remotes.origin.pull()
    else:
        print("Cloning repository...")
        Repo.clone_from(repo_url_auth, CLONE_DIR, branch=branch)

# ================= MENU =================
def select_inventory():
    inventories = ["hosts_init.yml", "hosts_roles.yml"]
    result = radiolist_dialog(
        title=f"Inventory Selection - Version {VERSION}",
        text="Select inventory file:",
        values=[(inv, inv) for inv in inventories],
    ).run()
    if not result:
        sys.exit(0)
    return result

def select_roles():
    roles_path = CLONE_DIR / "roles"
    if not roles_path.exists():
        return []
    roles = sorted([r.name for r in roles_path.iterdir() if r.is_dir()])
    result = checkboxlist_dialog(
        title="Role Selection",
        text="Select roles to execute (for info/log only):",
        values=[(r, r) for r in roles],
    ).run()
    return result or []

def select_limit():
    # Path to your inventory file
    inventory_path = f"{CLONE_DIR}/hosts_roles.yml"

    # loader = DataLoader()
    # inventory = InventoryManager(loader=loader, sources=[inventory_path])

    # Get all hosts
    # hosts = inventory.get_hosts()

    # for host in hosts:
    #     print(host.name)
    limits = [
        ("all", "All hosts"),
        ("m-server.home.lan", "m-server"),
        ("morefine.home.lan", "morefine"),
        ("asus.home.lan", "asus"),
        ("nas.home.lan", "nas"),
        ("rpi4.home.lan", "rpi4"),
        ("rpi5.home.lan", "rpi5"),
        ("debian13", "Debian 13"),
        ("ubuntu24", "Ubuntu 24"),
        ("ubuntu24s", "Ubuntu 24 Server"),
        ("ubuntu26s", "Ubuntu 26 Server"),
        ("rocky9", "Rocky 9"),
        ("rocky10", "Rocky 10"),
        ("alma10", "AlmaLinux 10"),
        ("custom", "Custom hosts (comma-separated)"),
        ]
    result = checkboxlist_dialog(
        title="Inventory Limit",        
        text="Select host limit:",
        values=limits,
    ).run()
    
    if result and "custom" in result:
        custom_hosts = input("Enter custom hosts (comma-separated): ")
        result = [r for r in result if r != "custom"] + [custom_hosts]
    if not result:
        sys.exit(0)
    return result

# ================= RUN ANSIBLE =================
def run_ansible_playbook(inventory, limit, roles=None, vault_password=None, ssh_key=None):
    # create ssh key file
    content = "Hello world"

    file_path = f"{CLONE_DIR}/ssh_key.pem"

    # create file
    with open(file_path, "w") as f:
        f.write(ssh_key + "\n")
    # set permissions (rw-r--r--)
    os.chmod(file_path, 0o600)
    playbook_file = CLONE_DIR / "all.yml"
    if not playbook_file.exists():
        sys.exit(f"Playbook {playbook_file} not found!")

    # Write vault password to temp file
    with tempfile.NamedTemporaryFile(mode="w", delete=False) as tmp:
        tmp.write(vault_password)
        tmp.flush()
        vault_file = tmp.name

    cmd = [
        "ansible-playbook",
        "-i", inventory,
        str(playbook_file),
        "--vault-password-file", vault_file,
    ]
    if limit != "all":
        cmd.extend(["--limit", ",".join(limit)])
    if roles:
        cmd.extend(["--tags", ",".join(roles)])
    print(cmd)
    try:
        subprocess.run(cmd, cwd=CLONE_DIR, check=True)
    except:
        print("not all jobs finished")
    finally:
        os.unlink(vault_file)


# ================= MAIN =================
def main():
    print("Connecting to Vault...")
    client = get_vault_client()

    print("Retrieving secrets...")
    gitlab_token = read_kv2_secret(client, VAULT_GITLAB_PATH, VAULT_GITLAB_FIELD)
    ansible_vault_pass = read_kv2_secret(client, VAULT_ANSIBLE_PATH, VAULT_ANSIBLE_VAULT_FIELD)
    ansible_ssh_key = read_kv2_secret(client, VAULT_ANSIBLE_PATH, VAULT_ANSIBLE_SSH_FIELD)
    clone_or_update_repo(GITLAB_URL, GITLAB_BRANCH, gitlab_token)
    inventory = select_inventory()
    roles = select_roles()  # purely informational
    limit = select_limit()

    print(f"Selected roles: {roles}")
    print(f"Running playbook all.yml with inventory {inventory} and limit {limit}")

    run_ansible_playbook(inventory, limit, roles, ansible_vault_pass, ansible_ssh_key)
    print("Execution finished successfully.")

if __name__ == "__main__":
    main()