www/membership/members/profiles.py

"""Self-service directory operations: pinned identity, user binds, fixed attributes.

Dex authenticates the session; LDAP remains authoritative for editable values.
No service-account writes, password hashes, tokens or profile snapshots are stored.
"""

import hashlib
import json
from contextlib import contextmanager

from django.conf import settings
from django.core import signing
from django.db import transaction
from ldap3 import BASE, MODIFY_REPLACE
from ldap3.core.exceptions import LDAPException
from ldap3.operation.search import compile_filter, parse_filter
from ldap3.protocol.controls import build_control
from ldap3.utils.conv import escape_filter_chars
from ldap3.utils.dn import parse_dn

from . import directory
from .models import Account


# Only attributes the Dex LDAP connector reads back into ID-token claims are
# editable: cn (nameAttr -> "name"). mail ("email") is changed through the
# separate verified flow and uid ("preferred_username") is immutable. Other
# directory attributes are left untouched, neither shown nor cleared.
TEXT_ATTRIBUTES = ("cn",)
READ_ATTRIBUTES = (*TEXT_ATTRIBUTES, "mail", "entryUUID", "uid", "modifyTimestamp")


def _values(entry):
    values = {name: list(entry[name].values) if name in entry else [] for name in READ_ATTRIBUTES}
    # Schema-aware ldap3 may convert this to datetime; assertion filters need
    # the original LDAP GeneralizedTime syntax rather than Python's rendering.
    if "modifyTimestamp" in entry:
        values["modifyTimestamp"] = [value.decode("ascii") for value in entry["modifyTimestamp"].raw_values]
    if len(values["entryUUID"]) != 1 or len(values["uid"]) != 1:
        raise ValueError("A unique directory identity is required.")
    return values


def _assertion(values):
    # Critical RFC 4528 assertion: concurrent edits or a recreated DN fail
    # atomically rather than modifying the wrong entry or overwriting new data.
    if len(values["modifyTimestamp"]) != 1:
        raise ValueError("Read access to modifyTimestamp is required for safe profile edits.")
    query = f"(entryUUID={escape_filter_chars(str(values['entryUUID'][0]))})"
    query = f"(&{query}(modifyTimestamp={escape_filter_chars(str(values['modifyTimestamp'][0]))}))"
    node = parse_filter(query, None, False, False, None, False)
    return build_control("1.3.6.1.1.12", True, compile_filter(node.elements[0]))


def _digest(values):
    def encoded(value):
        return hashlib.sha256(value).hexdigest() if isinstance(value, bytes) else str(value)
    return hashlib.sha256(json.dumps({key: [encoded(v) for v in values[key]] for key in READ_ATTRIBUTES}, sort_keys=True).encode()).hexdigest()


def read_profile(account):
    """Resolve once from the signed uid, then pin UUID; never accept a browser DN."""
    with transaction.atomic():
        member = Account.objects.select_for_update().get(pk=account.pk)
        if not member.oidc_sub or member.oidc_issuer != settings.DEX_ISSUER:
            raise ValueError("A verified Dex identity is required.")
        with directory.connection() as conn:
            query = f"(entryUUID={escape_filter_chars(member.ldap_uuid)})" if member.ldap_uuid else f"(&(objectClass=inetOrgPerson)(uid={escape_filter_chars(member.uid)}))"
            conn.search(settings.LDAP_SEARCH_BASE_DN, query, attributes=list(READ_ATTRIBUTES))
            if len(conn.entries) != 1:
                raise ValueError("Profile is missing or ambiguous.")
            entry = conn.entries[0]
            values = _values(entry)
            entry_uuid = str(values["entryUUID"][0])
            if str(values["uid"][0]) != member.uid or not entry_uuid or entry_uuid == "None" or len(entry_uuid) > 64:
                raise ValueError("Directory identity does not match the account.")
            if Account.objects.filter(ldap_uuid=entry_uuid).exclude(pk=member.pk).exists():
                raise ValueError("Directory identity belongs to another account.")
            if entry.entry_dn.casefold() == settings.LDAP_GROUP_PLACEHOLDER_DN.casefold():
                raise ValueError("System entries cannot be edited.")
            member.ldap_uuid, member.ldap_dn = entry_uuid, entry.entry_dn
            member.save(update_fields=["ldap_uuid", "ldap_dn"])
        account.ldap_uuid, account.ldap_dn = member.ldap_uuid, member.ldap_dn
    return values


def name_editable(dn):
    """False when cn names the entry: slapd refuses to remove an RDN value.

    Legacy accounts are cn=<name>,ou=People; renaming them would change DNs
    that other services store, so their name is shown read-only instead.
    """
    try:
        components = parse_dn(dn)
    except Exception:
        return False
    for attribute, _value, separator in components:
        if attribute.casefold() in {"cn", "2.5.4.3"}:
            return False
        if separator != "+":
            return True
    return False


def revision(account, values):
    return signing.dumps({"account": str(account.pk), "uuid": account.ldap_uuid, "digest": _digest(values)}, salt="ldap-profile")


def initial(values):
    # Dex publishes the first cn value; saving collapses any extras to one.
    return {name: str(values[name][0]) if values[name] else "" for name in TEXT_ATTRIBUTES}


@contextmanager
def _authorized(account, password):
    if not password or not account.ldap_uuid or not account.ldap_dn:
        raise ValueError("A pinned directory entry and current password are required.")
    # Authenticate as this user, not as the service identity. slapd self ACLs
    # authorize each modification; a compromised service cannot reset passwords.
    with directory._connection(account.ldap_dn, password) as conn:
        conn.search(account.ldap_dn, f"(entryUUID={escape_filter_chars(account.ldap_uuid)})", search_scope=BASE, attributes=list(READ_ATTRIBUTES))
        if len(conn.entries) != 1:
            raise ValueError("Directory identity changed; refusing to edit.")
        values = _values(conn.entries[0])
        if str(values["entryUUID"][0]) != account.ldap_uuid or str(values["uid"][0]) != account.uid:
            raise ValueError("Directory identity changed; refusing to edit.")
        yield conn, values


def verify_password(account, password):
    with _authorized(account, password):
        pass


def update_profile(account, cleaned, expected_revision):
    proof = signing.loads(expected_revision, salt="ldap-profile", max_age=3600)
    with _authorized(account, cleaned["current_password"]) as (conn, values):
        if proof != {"account": str(account.pk), "uuid": account.ldap_uuid, "digest": _digest(values)}:
            raise ValueError("Your profile changed. Reload it before saving.")
        if not name_editable(account.ldap_dn):
            raise ValueError("This account's name is part of its directory entry name.")
        changes = {}
        for name in TEXT_ATTRIBUTES:
            new_values = cleaned[name]
            if new_values != values[name]:
                changes[name] = [(MODIFY_REPLACE, new_values)]
        if changes and not conn.modify(account.ldap_dn, changes, controls=[_assertion(values)]):
            raise LDAPException("Profile modification rejected.")


def change_password(account, current_password, new_password):
    with _authorized(account, current_password) as (conn, _values):
        # RFC 3062 delegates hashing to slapd (including existing server policies).
        # No client-side MD5-Crypt, plaintext attribute write or admin reset.
        # No userIdentity targets: operate on the authenticated user. RFC 4528
        # assertions are not defined for this extended operation; supplying a
        # critical assertion here can make slapd reject an otherwise valid change.
        if not conn.extend.standard.modify_password(old_password=current_password, new_password=new_password):
            raise LDAPException("Password modification rejected.")


def change_email(account, current_password, email):
    with _authorized(account, current_password) as (conn, values):
        # Enforce directory-side mail uniqueness too: only an LDAP uniqueness
        # overlay can prevent another registrar racing this preflight search.
        conn.search(settings.LDAP_SEARCH_BASE_DN, f"(mail={escape_filter_chars(email)})", attributes=["entryUUID"])
        if any(str(entry.entryUUID.value) != account.ldap_uuid for entry in conn.entries):
            raise ValueError("Email change cannot be completed.")
        if not conn.modify(account.ldap_dn, {"mail": [(MODIFY_REPLACE, [email])]}, controls=[_assertion(values)]):
            raise LDAPException("Email modification rejected.")