fukuops/scripts/users.py

#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["ldap3", "typer", "httpx"]
# ///

import base64
import os
import re
import secrets
import subprocess
import sys

import httpx
import typer
from ldap3 import ALL, Connection, Server, SUBTREE

LLDAP_NAMESPACE = "apps-fuku"
LLDAP_SECRET = "lldap-secrets"
LLDAP_SERVICE = "lldap.apps-fuku.svc.cluster.local"
LLDAP_LDAP_PORT = 3890
LLDAP_BASE_DN = "dc=fuku,dc=local"
LLDAP_ADMIN_USER = "uid=admin,ou=people,dc=fuku,dc=local"

app = typer.Typer(help="Manage Authelia users in the LLDAP backend.")


def fail(msg: str) -> None:
    print(f"ERROR: {msg}", file=sys.stderr)
    raise typer.Exit(1)


def run_kubectl(args: list[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["kubectl", *args],
        capture_output=True,
        text=True,
        check=True,
    )


def get_secret_value(key: str) -> str:
    result = run_kubectl(
        [
            "get",
            "secret",
            LLDAP_SECRET,
            "-n",
            LLDAP_NAMESPACE,
            "-o",
            f"jsonpath={{.data.{key}}}",
        ]
    )
    return base64.b64decode(result.stdout).decode()


def get_configmap_value(key: str) -> str:
    result = run_kubectl(
        [
            "get",
            "configmap",
            "lldap-config",
            "-n",
            LLDAP_NAMESPACE,
            "-o",
            f"jsonpath={{.data.{key}}}",
        ]
    )
    return result.stdout.strip()


def get_lldap_password() -> str:
    return get_secret_value("LLDAP_LDAP_USER_PASS")


def get_lldap_hosts() -> list[str]:
    result = run_kubectl(
        [
            "get",
            "svc",
            "lldap",
            "-n",
            LLDAP_NAMESPACE,
            "-o",
            "jsonpath={.status.loadBalancer.ingress[*].ip}",
        ]
    )
    ips = [ip.strip() for ip in result.stdout.split() if ip.strip()]
    if not ips:
        fail("could not discover lldap LoadBalancer IPs")
    return ips


def lldap_connection() -> Connection:
    password = get_lldap_password()
    hosts = get_lldap_hosts()
    last_error = None
    for host in hosts:
        try:
            server = Server(host, port=LLDAP_LDAP_PORT, use_ssl=False, get_info=ALL)
            conn = Connection(
                server,
                user=LLDAP_ADMIN_USER,
                password=password,
                auto_bind=True,
                read_only=False,
            )
            return conn
        except Exception as e:
            last_error = e
            continue
    fail(f"could not connect to any lldap endpoint: {last_error}")


def validate_username(username: str) -> None:
    if not re.match(r"^[a-zA-Z0-9_.-]+$", username):
        fail(f"invalid username: {username} (allowed: a-z, 0-9, _, ., -)")


def get_lldap_http_url() -> str:
    # Use the public ingress with a valid certificate. The internal service
    # URL (ldap.fuku) uses a self-signed certificate and is not reachable from
    # this machine anyway.
    return "https://ldap.fukurokuju.dev"


def is_password_reset_enabled() -> bool:
    try:
        return get_secret_value("LLDAP_SMTP_OPTIONS__ENABLE_PASSWORD_RESET").lower() == "true"
    except subprocess.CalledProcessError:
        return False


def trigger_password_reset(username: str) -> None:
    if not is_password_reset_enabled():
        fail("LLDAP password reset is not enabled; cannot send invite email")

    base_url = get_lldap_http_url()
    url = f"{base_url}/auth/reset/step1/{username}"
    try:
        response = httpx.post(url, timeout=30.0)
        response.raise_for_status()
    except httpx.HTTPError as e:
        fail(f"failed to trigger password reset email: {e}")

    print(f"Password reset email triggered for '{username}'.")


def generate_random_password() -> str:
    return secrets.token_urlsafe(32)


def get_group_id(conn: Connection, group_name: str) -> int | None:
    conn.search(
        f"ou=groups,{LLDAP_BASE_DN}",
        f"(cn={group_name})",
        SUBTREE,
        attributes=["uid"],
    )
    if not conn.entries:
        return None
    uid = conn.entries[0].uid.value
    try:
        return int(uid)
    except (TypeError, ValueError):
        return None


def user_exists(conn: Connection, username: str) -> bool:
    conn.search(
        f"ou=people,{LLDAP_BASE_DN}",
        f"(uid={username})",
        SUBTREE,
        attributes=["uid"],
    )
    return bool(conn.entries)


def email_in_use(conn: Connection, email: str) -> bool:
    conn.search(
        f"ou=people,{LLDAP_BASE_DN}",
        f"(mail={email})",
        SUBTREE,
        attributes=["uid"],
    )
    return bool(conn.entries)


@app.command("list", help="List existing users (no passwords shown).")
def cmd_list() -> None:
    with lldap_connection() as conn:
        conn.search(
            f"ou=people,{LLDAP_BASE_DN}",
            "(objectClass=person)",
            SUBTREE,
            attributes=["uid", "cn", "mail", "memberOf"],
        )
        if not conn.entries:
            print("No users found.")
            return
        print(f"{'USERNAME':<20} {'DISPLAY NAME':<30} {'EMAIL':<30} GROUPS")
        for entry in conn.entries:
            uid = entry.uid.value if entry.uid else ""
            cn = entry.cn.value if entry.cn else ""
            mail = entry.mail.value if entry.mail else ""
            groups = ",".join(
                g.split(",")[0].replace("cn=", "") for g in entry.memberOf.values
            ) if entry.memberOf else ""
            print(f"{uid:<20} {cn:<30} {mail:<30} {groups}")


@app.command("add", help="Add a user. Omit --password to send an invite email.")
def cmd_add(
    username: str = typer.Option(..., "--username", "-u", help="Username"),
    displayname: str = typer.Option(..., "--displayname", "-n", help="Display name"),
    password: str | None = typer.Option(None, "--password", "-p", help="Password"),
    email: str = typer.Option("", "--email", "-e", help="Email address"),
    groups: list[str] = typer.Option(
        ["users"], "--group", "-g", help="Group to add the user to (repeatable)"
    ),
) -> None:
    validate_username(username)

    invite_mode = password is None
    if invite_mode and not email:
        fail("--email is required when --password is omitted")

    effective_email = email or f"{username}@roboces.dev"

    with lldap_connection() as conn:
        if user_exists(conn, username):
            fail(f"user '{username}' already exists")
        if email_in_use(conn, effective_email):
            fail(f"email '{effective_email}' is already in use")

        if invite_mode:
            password = generate_random_password()

        user_dn = f"uid={username},ou=people,{LLDAP_BASE_DN}"
        attrs = {
            "objectClass": ["inetOrgPerson", "posixAccount", "mailAccount", "person"],
            "uid": username,
            "cn": displayname,
            "mail": effective_email,
        }
        if not conn.add(user_dn, attributes=attrs):
            fail(f"failed to create user: {conn.result}")

        if not conn.extend.standard.modify_password(user=user_dn, new_password=password):
            fail(f"failed to set password: {conn.result}")

        for group_name in groups:
            group_id = get_group_id(conn, group_name)
            if group_id is None:
                print(f"  WARNING: group '{group_name}' not found, skipping")
                continue
            if not conn.add_user_to_group(user_dn, group_id):
                print(f"  WARNING: could not add to group '{group_name}': {conn.result}")
            else:
                print(f"  Added to group '{group_name}'.")

    if invite_mode:
        trigger_password_reset(username)
        print(f"User '{username}' created and invite email sent to '{effective_email}'.")
    else:
        print(f"User '{username}' created.")


@app.command("remove", help="Remove a user.")
def cmd_remove(
    username: str = typer.Option(..., "--username", "-u", help="Username"),
) -> None:
    validate_username(username)
    with lldap_connection() as conn:
        user_dn = f"uid={username},ou=people,{LLDAP_BASE_DN}"
        conn.search(user_dn, "(objectClass=*)", SUBTREE, attributes=["uid"])
        if not conn.entries:
            fail(f"user '{username}' not found")
        if not conn.delete(user_dn):
            fail(f"failed to delete user: {conn.result}")
        print(f"User '{username}' removed from LLDAP.")


@app.command("reset-password", help="Trigger a password reset email for an existing user.")
def cmd_reset_password(
    username: str = typer.Option(..., "--username", "-u", help="Username"),
    confirm: bool = typer.Option(
        True,
        "--confirm/--no-confirm",
        help="Prompt for confirmation before sending the reset email",
    ),
) -> None:
    validate_username(username)

    with lldap_connection() as conn:
        user_dn = f"uid={username},ou=people,{LLDAP_BASE_DN}"
        conn.search(user_dn, "(objectClass=*)", SUBTREE, attributes=["uid"])
        if not conn.entries:
            fail(f"user '{username}' not found")

    if confirm:
        typer.confirm(
            f"Send password reset email to '{username}'?",
            abort=True,
        )

    trigger_password_reset(username)
    print(f"Password reset email triggered for '{username}'.")


if __name__ == "__main__":
    os.environ.setdefault("PYTHONUNBUFFERED", "1")
    sys.stdout.reconfigure(line_buffering=True)
    app()