fukuops/scripts/authelia-users.sh

#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["ldap3"]
# ///
"""
Manage Authelia users in the LLDAP backend.

Usage:
  scripts/authelia-users.sh list
  scripts/authelia-users.sh add -u USERNAME -n "Display Name" -p PASSWORD [-e EMAIL] [-g group1,group2,...]
  scripts/authelia-users.sh remove -u USERNAME
  scripts/authelia-users.sh apply

Examples:
  scripts/authelia-users.sh add -u jdoe -n "John Doe" -p changeme -e jdoe@example.com -g users,admins
  scripts/authelia-users.sh remove -u jdoe

Note:
  With the LDAP backend, users are managed directly in LLDAP. The 'apply'
  command is kept for convenience but only validates the LLDAP connection.
"""

import argparse
import os
import re
import subprocess
import sys

from ldap3 import ALL, Connection, Server, SUBTREE

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


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


def get_lldap_password() -> str:
    result = subprocess.run(
        [
            "kubectl",
            "get",
            "secret",
            LLDAP_SECRET,
            "-n",
            LLDAP_NAMESPACE,
            "-o",
            "jsonpath={.data.LLDAP_LDAP_USER_PASS}",
        ],
        capture_output=True,
        text=True,
        check=True,
    )
    return __import__("base64").b64decode(result.stdout).decode()


def get_lldap_hosts() -> list[str]:
    result = subprocess.run(
        [
            "kubectl",
            "get",
            "svc",
            "lldap",
            "-n",
            LLDAP_NAMESPACE,
            "-o",
            "jsonpath={.status.loadBalancer.ingress[*].ip}",
        ],
        capture_output=True,
        text=True,
        check=True,
    )
    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_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 list_users() -> 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}")


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 add_user(
    username: str,
    displayname: str,
    password: str,
    email: str,
    groups: list[str],
) -> None:
    validate_username(username)

    with lldap_connection() as conn:
        user_dn = f"uid={username},ou=people,{LLDAP_BASE_DN}"

        # Check if user already exists.
        conn.search(user_dn, "(objectClass=*)", SUBTREE, attributes=["uid"])
        if conn.entries:
            print(f"User '{username}' already exists, updating password and groups.")
        else:
            attrs = {
                "objectClass": ["inetOrgPerson", "posixAccount", "mailAccount", "person"],
                "uid": username,
                "cn": displayname,
                "mail": email or f"{username}@roboces.dev",
            }
            if not conn.add(user_dn, attributes=attrs):
                fail(f"failed to create user: {conn.result}")
            print(f"User '{username}' created in LLDAP.")

        # Set password.
        if not conn.extend.standard.modify_password(user=user_dn, new_password=password):
            fail(f"failed to set password: {conn.result}")
        print(f"Password set for '{username}'.")

        # Add to groups.
        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}'.")


def remove_user(username: str) -> 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.")


def apply_users() -> None:
    # With the LDAP backend there is no local file to apply; users live in LLDAP.
    with lldap_connection():
        print("LLDAP connection OK. Users are managed directly; no apply step needed.")


def main() -> None:
    os.environ.setdefault("PYTHONUNBUFFERED", "1")
    sys.stdout.reconfigure(line_buffering=True)

    parser = argparse.ArgumentParser(
        description="Manage Authelia users in the LLDAP backend."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    subparsers.add_parser("list", help="List existing users (no passwords shown).")

    add_parser = subparsers.add_parser("add", help="Add or update a user.")
    add_parser.add_argument("-u", "--username", required=True, help="Username")
    add_parser.add_argument("-n", "--displayname", required=True, help="Display name")
    add_parser.add_argument("-p", "--password", required=True, help="Password")
    add_parser.add_argument("-e", "--email", default="", help="Email address")
    add_parser.add_argument(
        "-g", "--groups", default="users", help="Comma-separated groups"
    )

    remove_parser = subparsers.add_parser("remove", help="Remove a user.")
    remove_parser.add_argument("-u", "--username", required=True, help="Username")

    subparsers.add_parser(
        "apply",
        help="Validate LLDAP connection (no local state to apply with LDAP backend).",
    )

    args = parser.parse_args()
    if args.command == "list":
        list_users()
    elif args.command == "add":
        groups = [g.strip() for g in args.groups.split(",") if g.strip()]
        add_user(args.username, args.displayname, args.password, args.email, groups)
    elif args.command == "remove":
        remove_user(args.username)
    elif args.command == "apply":
        apply_users()


if __name__ == "__main__":
    main()