feat: Sync LDAP attributes for SSO/SAML users to fix prefill sources

matteius · matteius · commit fdfcc50b2d10 · 2026-01-21T05:49:01.000-05:00
- Add sync_user_ldap_attributes() function in sso_backends.py that queries
  LDAP directly for user attributes (mail, department, title, phone, etc.)
  and syncs them to UserProfile during SSO login
- Update SAML ACS view to call sync_user_ldap_attributes() after group sync
- Enhance LDAPDataSource.get_value() to:
  - Check user.email first for mail attribute (from Google SSO)
  - Query LDAP directly for SSO users without cached ldap_user object
  - Add _query_ldap_attribute() method for direct LDAP queries

This fixes prefill sources (email, department, etc.) not working for users
who authenticate via Google SAML SSO instead of direct LDAP authentication.

Bump version to 0.6.2
diff --git a/django_forms_workflows/__init__.py b/django_forms_workflows/__init__.py
@@ -3,7 +3,7 @@
 Enterprise-grade, database-driven form builder with approval workflows
 """
 
-__version__ = "0.6.1"
+__version__ = "0.6.2"
 __author__ = "Django Forms Workflows Contributors"
 __license__ = "LGPL-3.0-only"
 
diff --git a/django_forms_workflows/data_sources/ldap_source.py b/django_forms_workflows/data_sources/ldap_source.py
@@ -52,7 +52,7 @@ def get_value(self, user, field_name: str, **kwargs) -> Any | None:
 
         Args:
             user: Django User object
-            field_name: LDAP attribute name (e.g., 'department', 'title')
+            field_name: LDAP attribute name (e.g., 'department', 'title', 'mail')
             **kwargs: Unused
 
         Returns:
@@ -61,13 +61,15 @@ def get_value(self, user, field_name: str, **kwargs) -> Any | None:
         if not user or not user.is_authenticated:
             return None
 
-        if not self.is_available():
-            logger.warning(
-                "LDAP data source is not available (django-auth-ldap not configured)"
-            )
-            return None
+        # Map friendly name to LDAP attribute
+        ldap_attr = self.ATTRIBUTE_MAP.get(field_name, field_name)
 
         try:
+            # For 'mail', try user.email first (usually synced from LDAP/SSO)
+            if field_name == "mail" or ldap_attr == "mail":
+                if user.email:
+                    return user.email
+
             # Try to get from user profile first (cached LDAP data)
             if hasattr(user, "forms_profile"):
                 profile = user.forms_profile
@@ -76,33 +78,107 @@ def get_value(self, user, field_name: str, **kwargs) -> Any | None:
                     if value:
                         return value
 
-            # Try to get from LDAP backend directly
+            # Try to get from LDAP backend directly (for LDAP-authenticated users)
             ldap_user = getattr(user, "ldap_user", None)
-            if not ldap_user:
-                logger.debug(f"User {user.username} has no LDAP user object")
-                return None
+            if ldap_user:
+                # Special handling for manager email
+                if field_name == "manager_email":
+                    return self._get_manager_email(ldap_user)
+
+                # Get attribute from cached LDAP attrs
+                if hasattr(ldap_user, "attrs") and ldap_attr in ldap_user.attrs:
+                    value = ldap_user.attrs[ldap_attr]
+                    # LDAP attributes are often lists
+                    if isinstance(value, list) and value:
+                        return value[0]
+                    return value
+
+            # For SSO users without ldap_user, query LDAP directly
+            return self._query_ldap_attribute(user.username, ldap_attr)
 
-            # Map friendly name to LDAP attribute
-            ldap_attr = self.ATTRIBUTE_MAP.get(field_name, field_name)
+        except Exception as e:
+            logger.error(f"Error getting LDAP attribute {field_name}: {e}")
+            return None
 
-            # Special handling for manager email
-            if field_name == "manager_email":
-                return self._get_manager_email(ldap_user)
+    def _query_ldap_attribute(self, username: str, ldap_attr: str) -> Any | None:
+        """
+        Query LDAP directly for a user attribute.
 
-            # Get attribute from LDAP
-            if hasattr(ldap_user, "attrs") and ldap_attr in ldap_user.attrs:
-                value = ldap_user.attrs[ldap_attr]
-                # LDAP attributes are often lists
-                if isinstance(value, list) and value:
-                    return value[0]
-                return value
+        This is used for SSO users who don't have a cached ldap_user object.
+        """
+        import os
 
-            logger.debug(f"LDAP attribute not found: {ldap_attr}")
+        try:
+            import ldap
+            from ldap.filter import escape_filter_chars
+        except ImportError:
+            logger.debug("python-ldap not installed")
             return None
 
-        except Exception as e:
-            logger.error(f"Error getting LDAP attribute {field_name}: {e}")
+        ldap_server = getattr(settings, "AUTH_LDAP_SERVER_URI", None)
+        if not ldap_server:
+            return None
+
+        bind_dn = getattr(settings, "AUTH_LDAP_BIND_DN", "")
+        bind_password = getattr(settings, "AUTH_LDAP_BIND_PASSWORD", "")
+
+        # Get search base
+        user_search = getattr(settings, "AUTH_LDAP_USER_SEARCH", None)
+        search_base = None
+        if user_search and hasattr(user_search, "base_dn"):
+            search_base = user_search.base_dn
+        if not search_base:
+            search_base = getattr(settings, "AUTH_LDAP_USER_SEARCH_BASE", None)
+        if not search_base and bind_dn:
+            parts = bind_dn.split(",")
+            dc_parts = [p for p in parts if p.strip().upper().startswith("DC=")]
+            if dc_parts:
+                search_base = ",".join(dc_parts)
+
+        if not search_base:
+            logger.debug("Could not determine LDAP search base")
+            return None
+
+        try:
+            conn = ldap.initialize(ldap_server)
+            conn.set_option(ldap.OPT_REFERRALS, 0)
+            conn.set_option(ldap.OPT_PROTOCOL_VERSION, ldap.VERSION3)
+
+            # Configure TLS
+            tls_require_cert = os.getenv("LDAP_TLS_REQUIRE_CERT", "demand").lower()
+            if tls_require_cert == "never":
+                conn.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
+
+            if bind_dn and bind_password:
+                conn.simple_bind_s(bind_dn, bind_password)
+            else:
+                conn.simple_bind_s("", "")
+
+            search_filter = f"(sAMAccountName={escape_filter_chars(username)})"
+            result = conn.search_s(
+                search_base, ldap.SCOPE_SUBTREE, search_filter, [ldap_attr]
+            )
+
+            if result and result[0][0]:
+                attrs = result[0][1]
+                if ldap_attr in attrs:
+                    value = attrs[ldap_attr]
+                    if isinstance(value, list) and value:
+                        val = value[0]
+                        if isinstance(val, bytes):
+                            return val.decode("utf-8")
+                        return val
+
+            return None
+
+        except ldap.LDAPError as e:
+            logger.debug(f"LDAP query failed for {username}.{ldap_attr}: {e}")
             return None
+        finally:
+            try:
+                conn.unbind_s()
+            except Exception:
+                pass
 
     def _get_manager_email(self, ldap_user) -> str | None:
         """
diff --git a/django_forms_workflows/sso_backends.py b/django_forms_workflows/sso_backends.py
@@ -550,6 +550,166 @@ def sync_user_ldap_groups(user):
         conn.unbind_s()
 
 
+def sync_user_ldap_attributes(user):
+    """
+    Sync LDAP attributes to UserProfile for a specific user.
+
+    This fetches key attributes from LDAP (mail, department, title, etc.)
+    and stores them in the user's UserProfile. This is especially useful
+    for SSO users who authenticate via Google/SAML but need LDAP attributes
+    for form prefilling.
+
+    Args:
+        user: Django User object
+
+    Returns:
+        dict: Dictionary of synced attributes, or None if LDAP unavailable
+    """
+    from django_forms_workflows.models import UserProfile
+
+    # Check if LDAP is configured
+    ldap_server = getattr(settings, "AUTH_LDAP_SERVER_URI", None)
+    if not ldap_server:
+        logger.debug("LDAP not configured, skipping attribute sync")
+        return None
+
+    try:
+        import ldap
+        import os
+        from ldap.filter import escape_filter_chars
+    except ImportError:
+        logger.debug("python-ldap not installed, skipping attribute sync")
+        return None
+
+    # Get LDAP connection settings
+    bind_dn = getattr(settings, "AUTH_LDAP_BIND_DN", "")
+    bind_password = getattr(settings, "AUTH_LDAP_BIND_PASSWORD", "")
+    user_search_base = getattr(settings, "AUTH_LDAP_USER_SEARCH", None)
+
+    # Try to get search base from user search configuration
+    search_base = None
+    if user_search_base:
+        if hasattr(user_search_base, "base_dn"):
+            search_base = user_search_base.base_dn
+        elif isinstance(user_search_base, tuple) and len(user_search_base) >= 1:
+            search_base = user_search_base[0]
+
+    if not search_base:
+        search_base = getattr(settings, "AUTH_LDAP_USER_SEARCH_BASE", None)
+        if not search_base and bind_dn:
+            parts = bind_dn.split(",")
+            dc_parts = [p for p in parts if p.strip().upper().startswith("DC=")]
+            if dc_parts:
+                search_base = ",".join(dc_parts)
+
+    if not search_base:
+        logger.warning("Could not determine LDAP search base, skipping attribute sync")
+        return None
+
+    # Connect to LDAP
+    try:
+        conn = ldap.initialize(ldap_server)
+        conn.set_option(ldap.OPT_REFERRALS, 0)
+        conn.set_option(ldap.OPT_PROTOCOL_VERSION, ldap.VERSION3)
+
+        # Configure TLS
+        tls_require_cert = os.getenv("LDAP_TLS_REQUIRE_CERT", "demand").lower()
+        if tls_require_cert == "never":
+            conn.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
+
+        if bind_dn and bind_password:
+            conn.simple_bind_s(bind_dn, bind_password)
+        else:
+            conn.simple_bind_s("", "")
+
+    except ldap.LDAPError as e:
+        logger.error(f"LDAP connection failed for attribute sync: {e}")
+        raise
+
+    try:
+        # Attributes to fetch from LDAP
+        ldap_attrs = [
+            "mail",
+            "department",
+            "title",
+            "telephoneNumber",
+            "mobile",
+            "employeeID",
+            "physicalDeliveryOfficeName",
+            "company",
+            "givenName",
+            "sn",
+        ]
+
+        # Search for the user
+        search_filter = f"(sAMAccountName={escape_filter_chars(user.username)})"
+        result = conn.search_s(
+            search_base,
+            ldap.SCOPE_SUBTREE,
+            search_filter,
+            ldap_attrs,
+        )
+
+        if not result:
+            logger.debug(f"User '{user.username}' not found in LDAP for attribute sync")
+            return {}
+
+        # Get user's attributes
+        user_dn, user_attrs = result[0]
+        synced_attrs = {}
+
+        def get_attr(attrs, name):
+            """Helper to get first value of an LDAP attribute."""
+            val = attrs.get(name, [])
+            if val:
+                if isinstance(val[0], bytes):
+                    return val[0].decode("utf-8")
+                return val[0]
+            return None
+
+        # Map LDAP attributes
+        synced_attrs["mail"] = get_attr(user_attrs, "mail")
+        synced_attrs["department"] = get_attr(user_attrs, "department")
+        synced_attrs["title"] = get_attr(user_attrs, "title")
+        synced_attrs["phone"] = get_attr(user_attrs, "telephoneNumber")
+        synced_attrs["mobile"] = get_attr(user_attrs, "mobile")
+        synced_attrs["employee_id"] = get_attr(user_attrs, "employeeID")
+        synced_attrs["office_location"] = get_attr(user_attrs, "physicalDeliveryOfficeName")
+        synced_attrs["company"] = get_attr(user_attrs, "company")
+        synced_attrs["first_name"] = get_attr(user_attrs, "givenName")
+        synced_attrs["last_name"] = get_attr(user_attrs, "sn")
+
+        # Update or create UserProfile
+        profile, created = UserProfile.objects.get_or_create(user=user)
+
+        # Sync attributes to profile
+        profile_fields = {
+            "department": synced_attrs.get("department"),
+            "title": synced_attrs.get("title"),
+            "phone": synced_attrs.get("phone"),
+            "employee_id": synced_attrs.get("employee_id"),
+            "office_location": synced_attrs.get("office_location"),
+        }
+
+        updated = False
+        for field, value in profile_fields.items():
+            if value and hasattr(profile, field):
+                setattr(profile, field, value)
+                updated = True
+
+        if updated or created:
+            profile.save()
+            logger.info(
+                f"Synced LDAP attributes for user '{user.username}': "
+                f"{[k for k, v in synced_attrs.items() if v]}"
+            )
+
+        return synced_attrs
+
+    finally:
+        conn.unbind_s()
+
+
 # Pipeline for social-auth-app-django
 # Recommended pipeline configuration:
 #
diff --git a/django_forms_workflows/sso_views.py b/django_forms_workflows/sso_views.py
@@ -254,6 +254,20 @@ def post(self, request):
                     f"Failed to sync LDAP groups for SAML user '{user.username}': {e}"
                 )
 
+            # Sync LDAP attributes (mail, department, etc.) to UserProfile
+            try:
+                from .sso_backends import sync_user_ldap_attributes
+
+                ldap_attrs = sync_user_ldap_attributes(user)
+                if ldap_attrs:
+                    logger.info(
+                        f"Synced LDAP attributes for SAML user '{user.username}'"
+                    )
+            except Exception as e:
+                logger.warning(
+                    f"Failed to sync LDAP attributes for SAML user '{user.username}': {e}"
+                )
+
             # Log user in
             login(request, user, backend="django.contrib.auth.backends.ModelBackend")
             logger.info(f"SAML login successful for: {user.username}")
diff --git a/pyproject.toml b/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "django-forms-workflows"
-version = "0.6.1"
+version = "0.6.2"
 description = "Enterprise-grade, database-driven form builder with approval workflows and external data integration"
 license = "LGPL-3.0-only"
 readme = "README.md"
