#!/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" LLDAP_URL = "https://ldap.fukurokuju.dev" 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 trigger_password_reset(username: str) -> None: url = f"{LLDAP_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 ") 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: if not user_exists(conn=conn, username=username): 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()