atlas-iac/scripts/node_admin_access.py

#!/usr/bin/env python3
"""Audit or restore existing node passwords from JSON on encrypted SSH stdin.

Run as root with {"hostname": "titan-12", "passwords": {"atlas": "...",
"root": "..."}} on stdin. Passwords must come from the node's Vault record.
No passwords or hashes are written to files or output. SSH policy is unchanged.
"""

import argparse
import ctypes
import ctypes.util
import hmac
import json
import os
import pwd
from pathlib import Path
import socket
import subprocess
import sys


def password_status(username, password):
    """Compare an existing shadow hash locally; return non-secret state only."""
    with open("/etc/shadow", encoding="utf-8") as stream:
        row = next((line.rstrip("\n").split(":") for line in stream
                    if line.startswith(username + ":")), None)
    if row is None:
        raise ValueError("account_missing")
    stored = row[1]
    locked = stored.startswith(("!", "*")) or not stored
    matches = False
    if not locked:
        library = ctypes.CDLL(ctypes.util.find_library("crypt"))
        library.crypt.argtypes = [ctypes.c_char_p, ctypes.c_char_p]
        library.crypt.restype = ctypes.c_char_p
        calculated = library.crypt(password.encode(), stored.encode())
        matches = bool(calculated and hmac.compare_digest(calculated, stored.encode()))
    return {"locked": locked, "vault_password_matches": matches}


def run(payload, apply=False):
    """Validate all input before applying only missing/mismatched passwords."""
    if os.geteuid() != 0:
        raise ValueError("root_required")
    hostname = socket.gethostname().split(".")[0]
    if payload.get("hostname") != hostname:
        raise ValueError("hostname_mismatch")
    passwords = payload.get("passwords")
    if not isinstance(passwords, dict) or not passwords:
        raise ValueError("passwords_required")
    for username, password in passwords.items():
        if username not in {"atlas", "root"}:
            raise ValueError("account_not_allowed")
        if not isinstance(password, str) or not password or any(c in password for c in "\r\n\x00"):
            raise ValueError("invalid_password")
        pwd.getpwnam(username)
    before = {user: password_status(user, value) for user, value in passwords.items()}
    changed = []
    if apply:
        for user, value in passwords.items():
            if not before[user]["vault_password_matches"]:
                completed = subprocess.run(["/usr/sbin/chpasswd"], input=user + ":" + value + "\n",
                                           text=True, capture_output=True, timeout=15)
                if completed.returncode:
                    raise ValueError("password_update_failed")
                changed.append(user)
    after = {user: password_status(user, value) for user, value in passwords.items()}
    if apply and not all(state["vault_password_matches"] for state in after.values()):
        raise ValueError("password_verification_failed")
    return {"hostname": hostname, "applied": apply, "changed_accounts": changed,
            "before": before, "after": after}


def retire_legacy_sudo(result):
    """Remove only the known Metis grant after password access is established."""
    if not result["after"].get("atlas", {}).get("vault_password_matches"):
        raise ValueError("atlas_password_not_verified")
    expected = ("atlas ALL=(ALL) NOPASSWD: /usr/bin/systemctl, /usr/sbin/poweroff, "
                "/sbin/poweroff, /usr/local/bin/hecate, /usr/local/bin/k3s, /usr/bin/k3s")
    paths = [Path("/etc/sudoers.d/90-hecate-atlas"), Path("/etc/metis/sudoers-hecate")]
    for path in paths:
        if path.exists() and path.read_text().strip() != expected:
            raise ValueError("legacy_grant_modified_requires_review")
    if subprocess.run(["/usr/sbin/visudo", "-c"], capture_output=True).returncode:
        raise ValueError("sudo_configuration_invalid")
    backup = Path("/var/lib/atlas-maintenance/legacy-sudo-20261004")
    backup.mkdir(parents=True, exist_ok=True, mode=0o700)
    changed = []
    for path in paths:
        if path.exists():
            destination = backup / path.name
            if destination.exists():
                raise ValueError("legacy_backup_already_exists")
            path.rename(destination)
            changed.append(path.name)
    result["retired_legacy_sudo"] = changed


def main():
    """Read one bounded payload and emit only safe operational metadata."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--apply", action="store_true")
    parser.add_argument("--retire-legacy-sudo", action="store_true")
    args = parser.parse_args()
    try:
        content = sys.stdin.read(65537)
        if len(content) > 65536:
            raise ValueError("payload_too_large")
        result = run(json.loads(content), args.apply)
        if args.retire_legacy_sudo:
            retire_legacy_sudo(result)
    except Exception as error:
        # Do not echo exception messages: malformed input can contain credentials.
        print(json.dumps({"error": "node_access_operation_failed", "type": type(error).__name__}))
        return 1
    print(json.dumps(result, sort_keys=True))
    return 0


if __name__ == "__main__":
    sys.exit(main())