sticknife-home/app/ipa.py

from __future__ import annotations

import os
import subprocess
import tempfile
from dataclasses import dataclass

from .config import Config


@dataclass
class IpaResult:
    ok: bool
    message: str


def kinit(config: Config) -> None:
    if config.ipa_principal and config.ipa_keytab:
        subprocess.run(
            ["kinit", "-kt", config.ipa_keytab, config.ipa_principal],
            check=True,
            text=True,
            capture_output=True,
        )


def create_user(config: Config, user: dict[str, str | None]) -> IpaResult:
    username = user["username"] or ""
    email = user["email"] or ""
    full_name = user["full_name"] or username
    parts = full_name.split(" ", 1)
    first_name = parts[0]
    last_name = parts[1] if len(parts) > 1 else username

    if config.ipa_mode != "cli":
        return IpaResult(True, f"dry-run: would create IPA user {username} <{email}>")

    kinit(config)

    add_cmd = [
        "ipa",
        "user-add",
        username,
        "--first",
        first_name,
        "--last",
        last_name,
        "--cn",
        full_name,
        "--email",
        email,
        "--shell",
        config.ipa_default_shell,
        "--homedir",
        f"{config.ipa_home_root.rstrip('/')}/{username}",
        "--random",
        "--user-auth-type",
        "password",
    ]
    try:
        subprocess.run(add_cmd, check=True, text=True, capture_output=True)
    except subprocess.CalledProcessError as exc:
        return IpaResult(False, exc.stderr.strip() or exc.stdout.strip() or str(exc))
    return IpaResult(True, f"created IPA user {username}")


def add_default_group(config: Config, username: str) -> IpaResult:
    if config.ipa_mode != "cli":
        return IpaResult(True, f"dry-run: would add {username} to {config.default_ipa_group}")
    try:
        kinit(config)
        subprocess.run(
            ["ipa", "group-add-member", config.default_ipa_group, "--users", username],
            check=True,
            text=True,
            capture_output=True,
        )
    except subprocess.CalledProcessError as exc:
        return IpaResult(False, exc.stderr.strip() or exc.stdout.strip() or str(exc))
    return IpaResult(True, f"added {username} to {config.default_ipa_group}")


def set_user_enabled(config: Config, username: str, enabled: bool) -> IpaResult:
    action = "enable" if enabled else "disable"
    if config.ipa_mode != "cli":
        return IpaResult(True, f"dry-run: would {action} IPA user {username}")
    try:
        kinit(config)
        subprocess.run(
            ["ipa", f"user-{action}", username],
            check=True,
            text=True,
            capture_output=True,
        )
    except subprocess.CalledProcessError as exc:
        return IpaResult(False, exc.stderr.strip() or exc.stdout.strip() or str(exc))
    return IpaResult(True, f"{action}d IPA user {username}")


def provision_user(config: Config, user: dict[str, str | None], *, active: bool = True) -> IpaResult:
    username = user["username"] or ""
    created = create_user(config, user)
    if not created.ok:
        return created

    messages = [created.message]
    if active:
        grouped = add_default_group(config, username)
        messages.append(grouped.message)
        if not grouped.ok:
            return IpaResult(False, "; ".join(messages))
    else:
        disabled = set_user_enabled(config, username, False)
        messages.append(disabled.message)
        if not disabled.ok:
            return IpaResult(False, "; ".join(messages))
    return IpaResult(True, "; ".join(messages))


def approve_user(config: Config, username: str) -> IpaResult:
    enabled = set_user_enabled(config, username, True)
    if not enabled.ok:
        return enabled
    grouped = add_default_group(config, username)
    if not grouped.ok:
        return IpaResult(False, f"{enabled.message}; {grouped.message}")
    return IpaResult(True, f"{enabled.message}; {grouped.message}")


def set_password(config: Config, username: str, password: str) -> IpaResult:
    if config.ipa_mode != "cli":
        return IpaResult(True, f"dry-run: would set IPA password for {username}")
    try:
        kinit(config)
        subprocess.run(
            ["ipa", "passwd", username],
            input=f"{password}\n{password}\n",
            check=True,
            text=True,
            capture_output=True,
        )
    except subprocess.CalledProcessError as exc:
        return IpaResult(False, exc.stderr.strip() or exc.stdout.strip() or str(exc))
    return IpaResult(True, f"set IPA password for {username}")


def email_matches(config: Config, username: str, email: str) -> bool:
    try:
        kinit(config)
        result = subprocess.run(
            ["ipa", "user-show", username],
            check=True,
            text=True,
            capture_output=True,
        )
    except subprocess.CalledProcessError:
        return False
    expected = f"email address: {email.lower()}"
    return expected in result.stdout.lower()


def verify_password(config: Config, username: str, password: str) -> IpaResult:
    realm = config.ipa_principal.rsplit("@", 1)[-1] if "@" in config.ipa_principal else "POTTERNET.LAN"
    principal = f"{username}@{realm}"
    with tempfile.NamedTemporaryFile(prefix="charon-krb5cc-") as cache:
        env = os.environ.copy()
        env["KRB5CCNAME"] = cache.name
        try:
            subprocess.run(
                ["kinit", principal],
                input=f"{password}\n",
                check=True,
                text=True,
                capture_output=True,
                env=env,
            )
        except subprocess.CalledProcessError as exc:
            return IpaResult(False, exc.stderr.strip() or exc.stdout.strip() or "old password did not authenticate")
        finally:
            subprocess.run(["kdestroy", "-c", cache.name], text=True, capture_output=True, env=env)
    return IpaResult(True, f"verified password for {username}")