123 lines
5.2 KiB
Python
Executable File
123 lines
5.2 KiB
Python
Executable File
#!/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")
|
|
known_grants = {expected, expected.split(", /usr/local/bin/k3s")[0]}
|
|
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() not in known_grants:
|
|
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())
|