www/membership/members/directory.py

"""TLS-only LDAP; paid groups are never granted during registration or login."""

import re
import ssl
from contextlib import contextmanager
from urllib.parse import urlsplit

from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from ldap3 import BASE, MODIFY_ADD, MODIFY_DELETE, Connection, Server, Tls
from ldap3.core.exceptions import LDAPException
from ldap3.utils.conv import escape_filter_chars
from ldap3.utils.dn import escape_rdn


@contextmanager
def connection():
    url = urlsplit(settings.LDAP_URL)
    if url.scheme not in {"ldap", "ldaps"} or not url.hostname or not settings.LDAP_BIND_DN or not settings.LDAP_BIND_PASSWORD or not settings.LDAP_ACCOUNT_BASE_DN:
        raise ImproperlyConfigured("Configure LDAP URL, service bind, and account base DN.")
    server = Server(url.hostname, port=url.port or (636 if url.scheme == "ldaps" else 389), use_ssl=url.scheme == "ldaps", tls=Tls(validate=ssl.CERT_REQUIRED, ca_certs_file=settings.LDAP_CA_CERT_FILE, version=ssl.PROTOCOL_TLS_CLIENT), connect_timeout=5)
    conn = Connection(server, user=settings.LDAP_BIND_DN, password=settings.LDAP_BIND_PASSWORD, raise_exceptions=True, receive_timeout=8, auto_referrals=False)
    try:
        conn.open()
        if url.scheme == "ldap" and not conn.start_tls():
            raise LDAPException("StartTLS failed; refusing to bind.")
        if not conn.bind():
            raise LDAPException("LDAP bind failed.")
        yield conn
    finally:
        conn.unbind()


def _find(conn, query):
    conn.search(settings.LDAP_SEARCH_BASE_DN, query, attributes=["uid", "mail", "entryUUID"])
    return list(conn.entries)


def _identity(entry):
    return {"uid": str(entry.uid.value), "email": str(entry.mail.value), "ldap_dn": entry.entry_dn, "ldap_uuid": str(entry.entryUUID.value)}


def identity_for_login(uid):
    if not isinstance(uid, str) or not re.fullmatch(r"[a-z][a-z0-9._-]{2,47}", uid):
        raise ValueError("Dex must return a valid LDAP uid as preferred_username.")
    with connection() as conn:
        entries = _find(conn, f"(&(objectClass=inetOrgPerson)(uid={escape_filter_chars(uid)}))")
        if len(entries) != 1:
            raise ValueError("LDAP identity is missing or ambiguous.")
        return _identity(entries[0])


def create_account(uid, email, password):
    if settings.LDAP_PASSWORD_SCHEME != "{ARGON2}":
        raise ImproperlyConfigured("Verify slapd's Argon2 password policy before enabling registration.")
    with connection() as conn:
        if _find(conn, f"(|(uid={escape_filter_chars(uid)})(mail={escape_filter_chars(email)}))"):
            # Never reset/reuse an existing LDAP password based on mailbox access.
            raise ValueError("Account already exists. Sign in through Dex or contact the administrator.")
        dn = f"uid={escape_rdn(uid)},{settings.LDAP_ACCOUNT_BASE_DN}"
        created = False
        try:
            if not conn.add(dn, attributes={"objectClass": ["top", "inetOrgPerson"], "uid": uid, "cn": uid, "sn": uid, "mail": email}):
                raise LDAPException("LDAP account creation failed.")
            created = True
            if not conn.extend.standard.modify_password(dn, new_password=password):
                raise LDAPException("LDAP password modify failed.")
            conn.search(dn, "(objectClass=inetOrgPerson)", search_scope=BASE, attributes=["uid", "mail", "entryUUID"])
            if len(conn.entries) != 1:
                raise LDAPException("Cannot read created identity.")
            return _identity(conn.entries[0])
        except Exception:
            if created:
                conn.delete(dn)
            raise


def reconcile_groups(account, desired, managed):
    """Idempotent, atomic per-group changes. A real sentinel keeps groups nonempty."""
    placeholder = settings.LDAP_GROUP_PLACEHOLDER_DN
    if not placeholder:
        raise ImproperlyConfigured("LDAP_GROUP_PLACEHOLDER_DN is required to revoke the final member.")
    with connection() as conn:
        entries = _find(conn, f"(entryUUID={escape_filter_chars(account.ldap_uuid)})")
        if len(entries) != 1:
            raise ValueError("Membership identity missing or ambiguous; refusing to modify groups.")
        dn = entries[0].entry_dn
        if placeholder.casefold() == dn.casefold():
            raise ImproperlyConfigured("The LDAP sentinel must not be a login account.")
        if not conn.search(placeholder, "(objectClass=*)", search_scope=BASE, attributes=["objectClass"]):
            raise ImproperlyConfigured("LDAP sentinel does not exist.")
        # Revoke obsolete tiers before granting new tiers on plan changes.
        for group_dn in sorted(managed, key=lambda group: (group in desired, group)):
            if not conn.search(group_dn, "(objectClass=groupOfUniqueNames)", search_scope=BASE, attributes=["uniqueMember"]):
                raise ImproperlyConfigured("Managed group is missing or not groupOfUniqueNames.")
            members = {str(value).casefold() for value in conn.entries[0].uniqueMember.values}
            changes = []
            if placeholder.casefold() not in members:
                changes.append((MODIFY_ADD, [placeholder]))
            # Remove a stale DN too, e.g. after a directory rename.
            candidates = {value.casefold(): value for value in (account.ldap_dn, dn)}
            for candidate in candidates.values():
                if candidate.casefold() in members and (group_dn not in desired or candidate.casefold() != dn.casefold()):
                    changes.append((MODIFY_DELETE, [candidate]))
            if group_dn in desired and dn.casefold() not in members:
                changes.append((MODIFY_ADD, [dn]))
            if changes and not conn.modify(group_dn, {"uniqueMember": changes}):
                raise LDAPException("LDAP group synchronization failed.")
        return dn