My Profile
Sign Out
Tickets
Knowledge Base
Login
Articles in this section
Category / Section
  2. Public

Password expiration with basic multi auth

Published:
32 mins read

Multi-Auth with Password Expiration Script

This script combines basic_multi_auth with a password expiration policy for local users.

What it does

  1. basic_multi_auth
  • Users can authenticate against multiple LDAP backends (local Gluu directory and remote AD/LDAP servers).
  • The script reads the connection details from:
    /etc/certs/multi_auth_conf.json
  • Authentication attempts each backend in order until a match is found.
  1. Password expiration policy (local users only)
  • If a user has the attribute updatedAt, they are treated as a locally managed user.
  • For these users:
    • The script checks oxPasswordExpirationDate.
    • If missing → automatically sets now + 1 hour.
    • If expired → login is blocked with an error message:
      “Your password expired.”
    • If valid and in the future → login succeeds normally.
  • If a user does not have updatedAt → they are treated as a remote user, and password expiration checks are skipped.

Prerequisite

  1. Load basic_multi_auth script
  2. Load Password expiration policy script
  3. Load Combine script
  4. Add oxPasswordExpirationDate manually to test users. Example value: 20250823160822Z
  5. All scripts are included below

How to test

1. Remote user (no updatedAt)

  • Use a user from a remote AD/LDAP backend.
  • Login should succeed with no expiration check.

2. Local user with updatedAt but no oxPasswordExpirationDate

  • Remove the oxPasswordExpirationDate attribute from a local test user.
  • On first login:
    • The script will set a new expiration = current time + 1 hour.
    • Verify in LDAP that oxPasswordExpirationDate was created.

3. Local user with valid expiration

  • Set oxPasswordExpirationDate to a future time (e.g., tomorrow).
  • Login should succeed.

4. Local user with expired password

  • Set oxPasswordExpirationDate to a past time (e.g., yesterday).
  • Login should fail with:
    “Your password expired.”

Summary

  • Remote users → authenticated via external backends, not affected by password expiration.
  • Local users → must comply with password expiration rules.
  • Default expiration → if missing, script sets to 1 hour from first login.

basic_multi_auth script

  • Custom property (key/value): “auth_configuration_file” --> “/etc/certs/multi_auth_conf.json” [ no quote ]
# oxAuth is available under the MIT License (2008). See http://opensource.org/licenses/MIT for full text.
# Copyright (c) 2016, Gluu
#
# Author: Yuriy Movchan
#

from org.gluu.service.cdi.util import CdiUtil
from org.gluu.oxauth.security import Identity
from org.gluu.model.custom.script.type.auth import PersonAuthenticationType
from org.gluu.oxauth.service import UserService, AuthenticationService, AppInitializer
from org.gluu.oxauth.service import MetricService
from org.gluu.oxauth.service.common import EncryptionService
from org.gluu.model.metric import MetricType
from org.gluu.util import StringHelper
from org.gluu.util import ArrayHelper
from org.gluu.persist.service import PersistanceFactoryService
from org.gluu.persist.ldap.impl import LdapEntryManagerFactory
from org.gluu.model.ldap import GluuLdapConfiguration
from java.util import Arrays, Properties

import java
import json

class PersonAuthentication(PersonAuthenticationType):
   def __init__(self, currentTimeMillis):
       self.currentTimeMillis = currentTimeMillis

   def init(self, customScript, configurationAttributes):
       print "Basic (multi auth conf). Initialization"

       if (not configurationAttributes.containsKey("auth_configuration_file")):
           print "Basic (multi auth conf). The property auth_configuration_file is empty"
           return False
           
       authConfigurationFile = configurationAttributes.get("auth_configuration_file").getValue2()
       authConfiguration = self.loadAuthConfiguration(authConfigurationFile)
       if (authConfiguration == None):
           print "Basic (multi auth conf). File with authentication configuration should be not empty"
           return False
       
       validationResult = self.validateAuthConfiguration(authConfiguration)
       if (not validationResult):
           return False

       ldapExtendedEntryManagers = self.createLdapExtendedEntryManagers(authConfiguration)
       if (ldapExtendedEntryManagers == None):
           return False
       
       self.ldapExtendedEntryManagers = ldapExtendedEntryManagers

       print "Basic (multi auth conf). Initialized successfully"
       return True     

   def destroy(self, authConfiguration):
       print "Basic (multi auth conf). Destroy"
       
       result = True
       for ldapExtendedEntryManager in self.ldapExtendedEntryManagers:
           ldapConfiguration = ldapExtendedEntryManager["ldapConfiguration"]
           ldapEntryManager = ldapExtendedEntryManager["ldapEntryManager"]

           destoryResult = ldapEntryManager.destroy()
           result = result and destoryResult
           print "Basic (multi auth conf). Destroyed: " + ldapConfiguration.getConfigId() + ". Result: " + str(destoryResult) 

       print "Basic (multi auth conf). Destroyed successfully"

       return result

   def getApiVersion(self):
       return 11

   def getAuthenticationMethodClaims(self, requestParameters):
       return None

   def isValidAuthenticationMethod(self, usageType, configurationAttributes):
       return True

   def getAlternativeAuthenticationMethod(self, usageType, configurationAttributes):
       return None

   def authenticate(self, configurationAttributes, requestParameters, step):
       authenticationService = CdiUtil.bean(AuthenticationService)

       if (step == 1):
           print "Basic (multi auth conf). Authenticate for step 1"

           identity = CdiUtil.bean(Identity)
           credentials = identity.getCredentials()

           metricService = CdiUtil.bean(MetricService)
           timerContext = metricService.getTimer(MetricType.OXAUTH_USER_AUTHENTICATION_RATE).time()
           try:
               keyValue = credentials.getUsername()
               userPassword = credentials.getPassword()
   
               if (StringHelper.isNotEmptyString(keyValue) and StringHelper.isNotEmptyString(userPassword)):
                   for ldapExtendedEntryManager in self.ldapExtendedEntryManagers:
                       ldapConfiguration = ldapExtendedEntryManager["ldapConfiguration"]
                       ldapEntryManager = ldapExtendedEntryManager["ldapEntryManager"]
                       loginAttributes = ldapExtendedEntryManager["loginAttributes"]
                       localLoginAttributes = ldapExtendedEntryManager["localLoginAttributes"]
   
                       print "Basic (multi auth conf). Authenticate for step 1. Using configuration: " + ldapConfiguration.getConfigId()
   
                       idx = 0
                       count = len(loginAttributes)
                       while (idx < count):
                           primaryKey = loginAttributes[idx]
                           localPrimaryKey = localLoginAttributes[idx]
   
                           loggedIn = authenticationService.authenticate(ldapConfiguration, ldapEntryManager, keyValue, userPassword, primaryKey, localPrimaryKey)
                           if (loggedIn):
                               metricService.incCounter(MetricType.OXAUTH_USER_AUTHENTICATION_SUCCESS)
                               return True
                           idx += 1
           finally:
               timerContext.stop()
               
           metricService.incCounter(MetricType.OXAUTH_USER_AUTHENTICATION_FAILURES)

           return False
       else:
           return False

   def prepareForStep(self, configurationAttributes, requestParameters, step):
       if (step == 1):
           print "Basic (multi auth conf). Prepare for Step 1"
           return True
       else:
           return False

   def getExtraParametersForStep(self, configurationAttributes, step):
       return None

   def getCountAuthenticationSteps(self, configurationAttributes):
       return 1

   def getPageForStep(self, configurationAttributes, step):
       return ""

   def getNextStep(self, configurationAttributes, requestParameters, step):
       return -1

   def getLogoutExternalUrl(self, configurationAttributes, requestParameters):
       print "Get external logout URL call"
       return None

   def logout(self, configurationAttributes, requestParameters):
       return True

   def loadAuthConfiguration(self, authConfigurationFile):
       authConfiguration = None

       # Load authentication configuration from file
       f = open(authConfigurationFile, 'r')
       try:
           authConfiguration = json.loads(f.read())
       except:
           print "Basic (multi auth conf). Load auth configuration. Failed to load authentication configuration from file:", authConfigurationFile
           return None
       finally:
           f.close()
       
       return authConfiguration

   def validateAuthConfiguration(self, authConfiguration):
       isValid = True

       if (not ("ldap_configuration" in authConfiguration)):
           print "Basic (multi auth conf). Validate auth configuration. There is no ldap_configuration section in configuration"
           return False
       
       idx = 1
       for ldapConfiguration in authConfiguration["ldap_configuration"]:
           if (not self.containsAttributeString(ldapConfiguration, "configId")):
               print "Basic (multi auth conf). Validate auth configuration. There is no 'configId' attribute in ldap_configuration section #" + str(idx)
               return False

           configId = ldapConfiguration["configId"]

           if (not self.containsAttributeArray(ldapConfiguration, "servers")):
               print "Basic (multi auth conf). Validate auth configuration. Property 'servers' in configuration '" + configId + "' is invalid"
               return False

           if (self.containsAttributeString(ldapConfiguration, "bindDN")):
               if (not self.containsAttributeString(ldapConfiguration, "bindPassword")):
                   print "Basic (multi auth conf). Validate auth configuration. Property 'bindPassword' in configuration '" + configId + "' is invalid"
                   return False

           if (not self.containsAttributeString(ldapConfiguration, "useSSL")):
               print "Basic (multi auth conf). Validate auth configuration. Property 'useSSL' in configuration '" + configId + "' is invalid"
               return False

           if (not self.containsAttributeString(ldapConfiguration, "maxConnections")):
               print "Basic (multi auth conf). Validate auth configuration. Property 'maxConnections' in configuration '" + configId + "' is invalid"
               return False
               
           if (not self.containsAttributeArray(ldapConfiguration, "baseDNs")):
               print "Basic (multi auth conf). Validate auth configuration. Property 'baseDNs' in configuration '" + configId + "' is invalid"
               return False

           if (not self.containsAttributeArray(ldapConfiguration, "loginAttributes")):
               print "Basic (multi auth conf). Validate auth configuration. Property 'loginAttributes' in configuration '" + configId + "' is invalid"
               return False

           if (not self.containsAttributeArray(ldapConfiguration, "localLoginAttributes")):
               print "Basic (multi auth conf). Validate auth configuration. Property 'localLoginAttributes' in configuration '" + configId + "' is invalid"
               return False

           if (len(ldapConfiguration["loginAttributes"]) != len(ldapConfiguration["localLoginAttributes"])):
               print "Basic (multi auth conf). Validate auth configuration. The number of attributes in 'loginAttributes' and 'localLoginAttributes' isn't equal in configuration '" + configId + "'"
               return False

           idx += 1

       return True

   def createLdapExtendedEntryManagers(self, authConfiguration):
       ldapExtendedConfigurations = self.createLdapExtendedConfigurations(authConfiguration)
       
       appInitializer = CdiUtil.bean(AppInitializer)
       persistanceFactoryService = CdiUtil.bean(PersistanceFactoryService)
       ldapEntryManagerFactory = persistanceFactoryService.getPersistenceEntryManagerFactory(LdapEntryManagerFactory)
       persistenceType = ldapEntryManagerFactory.getPersistenceType()

       ldapExtendedEntryManagers = []
       for ldapExtendedConfiguration in ldapExtendedConfigurations:
           connectionConfiguration = ldapExtendedConfiguration["connectionConfiguration"]
           ldapConfiguration = ldapExtendedConfiguration["ldapConfiguration"]

           ldapProperties = Properties()
           for key, value in connectionConfiguration.items():
               value_string = value
               if isinstance(value_string, list):
                   value_string = ", ".join(value)
               else:
                   value_string = str(value)

               ldapProperties.setProperty(persistenceType + "#" + key, value_string)

           if StringHelper.isNotEmptyString(ldapConfiguration.getBindPassword()):
               ldapProperties.setProperty(persistenceType + "#bindPassword", ldapConfiguration.getBindPassword())
           
           ldapEntryManager = ldapEntryManagerFactory.createEntryManager(ldapProperties)

           ldapExtendedEntryManagers.append({ "ldapConfiguration" : ldapConfiguration, "ldapProperties" : ldapProperties, "loginAttributes" : ldapExtendedConfiguration["loginAttributes"], "localLoginAttributes" : ldapExtendedConfiguration["localLoginAttributes"], "ldapEntryManager" : ldapEntryManager })
       
       return ldapExtendedEntryManagers

   def createLdapExtendedConfigurations(self, authConfiguration):
       ldapExtendedConfigurations = []

       for connectionConfiguration in authConfiguration["ldap_configuration"]:
           configId = connectionConfiguration["configId"]
           
           servers = connectionConfiguration["servers"]

           bindDN = None
           bindPassword = None
           useAnonymousBind = True
           if (self.containsAttributeString(connectionConfiguration, "bindDN")):
               useAnonymousBind = False
               bindDN = connectionConfiguration["bindDN"]
               bindPassword = CdiUtil.bean(EncryptionService).decrypt(connectionConfiguration["bindPassword"])

           useSSL = connectionConfiguration["useSSL"]
           maxConnections = connectionConfiguration["maxConnections"]
           baseDNs = connectionConfiguration["baseDNs"]
           loginAttributes = connectionConfiguration["loginAttributes"]
           localLoginAttributes = connectionConfiguration["localLoginAttributes"]
           
           ldapConfiguration = GluuLdapConfiguration()
           ldapConfiguration.setConfigId(configId)
           ldapConfiguration.setBindDN(bindDN)
           ldapConfiguration.setBindPassword(bindPassword)
           ldapConfiguration.setServers(Arrays.asList(servers))
           ldapConfiguration.setMaxConnections(maxConnections)
           ldapConfiguration.setUseSSL(useSSL)
           ldapConfiguration.setBaseDNs(Arrays.asList(baseDNs))
           ldapConfiguration.setPrimaryKey(loginAttributes[0])
           ldapConfiguration.setLocalPrimaryKey(localLoginAttributes[0])
           ldapConfiguration.setUseAnonymousBind(useAnonymousBind)

           ldapExtendedConfigurations.append({ "ldapConfiguration" : ldapConfiguration, "connectionConfiguration" : connectionConfiguration, "loginAttributes" : loginAttributes, "localLoginAttributes" : localLoginAttributes })
       
       return ldapExtendedConfigurations

   def containsAttributeString(self, dictionary, attribute):
       return ((attribute in dictionary) and StringHelper.isNotEmptyString(dictionary[attribute]))

   def containsAttributeArray(self, dictionary, attribute):
       return ((attribute in dictionary) and (len(dictionary[attribute]) > 0))

Password Expiration script

# -*- coding: utf-8 -*-

from org.gluu.service.cdi.util import CdiUtil
from org.gluu.oxauth.security import Identity
from org.gluu.model.custom.script.type.auth import PersonAuthenticationType
from org.gluu.oxauth.service import AuthenticationService
from org.gluu.util import StringHelper
from java.util import GregorianCalendar, TimeZone, Arrays, Date
from org.gluu.jsf2.message import FacesMessages
from javax.faces.application import FacesMessage
from java.text import SimpleDateFormat
import re

class PersonAuthentication(PersonAuthenticationType):
   def __init__(self, currentTimeMillis):
       self.currentTimeMillis = currentTimeMillis

   def init(self, customScript, configurationAttributes):
       print "Password-expiration check (dual-format) initialized"
       return True

   def destroy(self, configurationAttributes):
       print "Password-expiration check (dual-format) destroyed"
       return True

   def getApiVersion(self):
       return 11

   def isValidAuthenticationMethod(self, usageType, configurationAttributes):
       return True

   def getAuthenticationMethodClaims(self, requestParameters):
       return None

   def getAlternativeAuthenticationMethod(self, usageType, configurationAttributes):
       return None


   def _utc_now(self):
       cal = GregorianCalendar(TimeZone.getTimeZone("UTC"))
       return cal.getTime()

   def _add_error(self, msg):
       try:
           faces = CdiUtil.bean(FacesMessages)
           faces.add(FacesMessage.SEVERITY_ERROR, msg)
       except:
           print "AUTH ERROR: %s" % msg

   def _parse_expiration_as_date(self, raw):
       """
       Accepts:
         - java.util.Date (returns as-is)
         - 'YYYYMMDDHHMMSSZ' (GeneralizedTime)
         - 'YYYYMMDDHHMMSS.fffZ' (fraction ignored)
         - 'YYYY-MM-DD' (ISO date -> 23:59:59 UTC that day)
         - 'YYYY-MM-DDTHH:MM:SSZ' or '...SS.SSSZ' (ISO instant in UTC)
       Returns java.util.Date or None if unparseable.
       """
       if raw is None:
           return None

       if isinstance(raw, Date):
           return raw

       try:
           if hasattr(raw, "__len__") and not isinstance(raw, basestring):
               raw = raw[0] if len(raw) > 0 else None
               if raw is None:
                   return None
       except:
           pass

       s = (str(raw)).strip()
       if len(s) == 0:
           return None

       s_norm = re.sub(r"\.\d+Z$", "Z", s)

       def _fmt(pattern):
           f = SimpleDateFormat(pattern)
           f.setTimeZone(TimeZone.getTimeZone("UTC"))
           return f

       parsers = [
           ("gen", _fmt("yyyyMMddHHmmss'Z'")),                 # 20250823150956Z
           ("iso_dt_ms", _fmt("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")),# 2025-08-23T15:09:56.123Z
           ("iso_dt", _fmt("yyyy-MM-dd'T'HH:mm:ss'Z'")),       # 2025-08-23T15:09:56Z
       ]

       try:
           if re.match(r"^\d{14}Z$", s_norm):
               return parsers[0][1].parse(s_norm)
       except:
           pass

       try:
           if re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$", s):
               return parsers[1][1].parse(s)
       except:
           pass

       try:
           if re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$", s):
               return parsers[2][1].parse(s)
       except:
           pass

       try:
           if re.match(r"^\d{4}-\d{2}-\d{2}$", s):
               d = _fmt("yyyy-MM-dd").parse(s)
               cal = GregorianCalendar(TimeZone.getTimeZone("UTC"))
               cal.setTime(d)
               cal.set(GregorianCalendar.HOUR_OF_DAY, 23)
               cal.set(GregorianCalendar.MINUTE, 59)
               cal.set(GregorianCalendar.SECOND, 59)
               cal.set(GregorianCalendar.MILLISECOND, 0)
               return cal.getTime()
       except:
           pass

       return None


   def authenticate(self, configurationAttributes, requestParameters, step):
       if step != 1:
           return False

       identity = CdiUtil.bean(Identity)
       authn = CdiUtil.bean(AuthenticationService)

       creds = identity.getCredentials()
       user_name = creds.getUsername()
       user_password = creds.getPassword()

       if not (StringHelper.isNotEmptyString(user_name) and StringHelper.isNotEmptyString(user_password)):
           self._add_error("Username or password is empty.")
           return False

       if not authn.authenticate(user_name, user_password):
           return False

       user = authn.getAuthenticatedUser()
       if user is None:
           self._add_error("Authentication context missing user.")
           return False

       exp_raw = user.getAttributeObject("oxPasswordExpirationDate")
       exp = self._parse_expiration_as_date(exp_raw)

       if exp is None:
           try:
               s = user.getAttribute("oxPasswordExpirationDate")
               exp = self._parse_expiration_as_date(s)
           except:
               pass

       if exp is None:
           self._add_error("You don't have any expiration date or the format is invalid. Use YYYYMMDDHHMMSSZ or YYYY-MM-DD.")
           return False

       now = self._utc_now()
       if now.after(exp):
           self._add_error("Your password expired.")
           return False

       return True

   def prepareForStep(self, configurationAttributes, requestParameters, step):
       return step == 1

   def getExtraParametersForStep(self, configurationAttributes, step):
       return Arrays.asList()

   def getCountAuthenticationSteps(self, configurationAttributes):
       return 1

   def getPageForStep(self, configurationAttributes, step):
       return ""

   def getNextStep(self, configurationAttributes, requestParameters, step):
       return -1

   def getLogoutExternalUrl(self, configurationAttributes, requestParameters):
       return None

   def logout(self, configurationAttributes, requestParameters):
       return True

Combined script

  • Custom property (key/value): “auth_configuration_file” --> “/etc/certs/multi_auth_conf.json” [ no quote ]
# oxAuth is available under the MIT License (2008). See http://opensource.org/licenses/MIT for full text.
# Copyright (c) 2016, Gluu
# Author: Yuriy Movchan
# -*- coding: utf-8 -*-

from org.gluu.service.cdi.util import CdiUtil
from org.gluu.oxauth.security import Identity
from org.gluu.model.custom.script.type.auth import PersonAuthenticationType
from org.gluu.oxauth.service import UserService, AuthenticationService, AppInitializer
from org.gluu.oxauth.service.common import EncryptionService
from org.gluu.persist.service import PersistanceFactoryService
from org.gluu.persist.ldap.impl import LdapEntryManagerFactory
from org.gluu.model.ldap import GluuLdapConfiguration
from org.gluu.util import StringHelper
from java.util import Arrays, Properties, GregorianCalendar, TimeZone, Date
from java.text import SimpleDateFormat
from org.gluu.jsf2.message import FacesMessages
from javax.faces.application import FacesMessage

import json
import re

class PersonAuthentication(PersonAuthenticationType):

   def __init__(self, currentTimeMillis):
       self.currentTimeMillis = currentTimeMillis

   def init(self, customScript, configurationAttributes):
       print "basic_multi_auth(pass-exp). Initialization"

       if not configurationAttributes.containsKey("auth_configuration_file"):
           print "basic_multi_auth(pass-exp). The property auth_configuration_file is empty"
           return False

       authConfigurationFile = configurationAttributes.get("auth_configuration_file").getValue2()
       authConfiguration = self.loadAuthConfiguration(authConfigurationFile)
       if authConfiguration is None:
           print "basic_multi_auth(pass-exp). File with authentication configuration should be not empty"
           return False

       if not self.validateAuthConfiguration(authConfiguration):
           return False

       ldapExtendedEntryManagers = self.createLdapExtendedEntryManagers(authConfiguration)
       if ldapExtendedEntryManagers is None:
           return False

       self.ldapExtendedEntryManagers = ldapExtendedEntryManagers
       print "basic_multi_auth(pass-exp). Initialized successfully"
       return True

   def destroy(self, authConfiguration):
       print "basic_multi_auth(pass-exp). Destroy"
       result = True
       for ldapExtendedEntryManager in self.ldapExtendedEntryManagers:
           ldapConfiguration = ldapExtendedEntryManager["ldapConfiguration"]
           ldapEntryManager = ldapExtendedEntryManager["ldapEntryManager"]
           destoryResult = ldapEntryManager.destroy()
           result = result and destoryResult
           print "basic_multi_auth(pass-exp). Destroyed: %s. Result: %s" % (ldapConfiguration.getConfigId(), str(destoryResult))
       return result

   def getApiVersion(self):
       return 11

   def getAuthenticationMethodClaims(self, requestParameters):
       return None

   def isValidAuthenticationMethod(self, usageType, configurationAttributes):
       return True

   def getAlternativeAuthenticationMethod(self, usageType, configurationAttributes):
       return None

   def authenticate(self, configurationAttributes, requestParameters, step):
       authenticationService = CdiUtil.bean(AuthenticationService)

       if step != 1:
           return False

       identity = CdiUtil.bean(Identity)
       credentials = identity.getCredentials()
       keyValue = credentials.getUsername()
       userPassword = credentials.getPassword()

       if StringHelper.isNotEmptyString(keyValue) and StringHelper.isNotEmptyString(userPassword):
           for ldapExtendedEntryManager in self.ldapExtendedEntryManagers:
               ldapConfiguration = ldapExtendedEntryManager["ldapConfiguration"]
               ldapEntryManager = ldapExtendedEntryManager["ldapEntryManager"]
               loginAttributes = ldapExtendedEntryManager["loginAttributes"]
               localLoginAttributes = ldapExtendedEntryManager["localLoginAttributes"]

               print "basic_multi_auth(pass-exp). Using configuration: %s" % ldapConfiguration.getConfigId()

               for idx in range(len(loginAttributes)):
                   primaryKey = loginAttributes[idx]
                   localPrimaryKey = localLoginAttributes[idx]

                   loggedIn = authenticationService.authenticate(
                       ldapConfiguration, ldapEntryManager, keyValue, userPassword, primaryKey, localPrimaryKey
                   )
                   if loggedIn:
                       # Enforce post-auth password policy (conditional on updatedAt)
                       return self._post_auth_policy(authenticationService)

       return False

   def prepareForStep(self, configurationAttributes, requestParameters, step):
       return step == 1

   def getExtraParametersForStep(self, configurationAttributes, step):
       return None

   def getCountAuthenticationSteps(self, configurationAttributes):
       return 1

   def getPageForStep(self, configurationAttributes, step):
       return ""

   def getNextStep(self, configurationAttributes, requestParameters, step):
       return -1

   def getLogoutExternalUrl(self, configurationAttributes, requestParameters):
       return None

   def logout(self, configurationAttributes, requestParameters):
       return True

   def _post_auth_policy(self, authenticationService):
       ctx_user = authenticationService.getAuthenticatedUser()
       if ctx_user is None:
           self._add_error("Authentication context missing user.")
           return False

       userService = CdiUtil.bean(UserService)
       user = userService.getUser(ctx_user.getUserId())
       if user is None:
           self._add_error("Failed to reload user from LDAP.")
           return False

       updatedAt = user.getUpdatedAt()  # java.util.Date or None

       if updatedAt is None:
           print "basic_multi_auth(pass-exp). No 'updatedAt' -> skipping expiration policy"
           return True

       exp = None
       try:
           exp = self._parse_expiration_as_date(user.getAttributeObject("oxPasswordExpirationDate"))
       except:
           exp = None
       if exp is None:
           try:
               exp = self._parse_expiration_as_date(user.getAttribute("oxPasswordExpirationDate"))
           except:
               exp = None

       if exp is None:
           try:
               cal = GregorianCalendar(TimeZone.getTimeZone("UTC"))
               cal.add(GregorianCalendar.HOUR_OF_DAY, 1)
               defaultExp = cal.getTime()
               fmt = SimpleDateFormat("yyyyMMddHHmmss'Z'")
               fmt.setTimeZone(TimeZone.getTimeZone("UTC"))
               genTime = fmt.format(defaultExp)

               user.setAttribute("oxPasswordExpirationDate", genTime)
               userService.updateUser(user)

               print "basic_multi_auth(pass-exp). Missing expiration -> set default +1h: %s" % genTime
               return True
           except Exception as e:
               self._add_error("Failed to set default expiration.")
               print "basic_multi_auth(pass-exp). ERROR setting default expiration: %s" % e
               return False

       now = self._utc_now()
       if now.after(exp):
           self._add_error("Your password expired.")
           return False

       print "basic_multi_auth(pass-exp). Expiration OK"
       return True

   def _add_error(self, msg):
       try:
           faces = CdiUtil.bean(FacesMessages)
           faces.add(FacesMessage.SEVERITY_ERROR, msg)
       except:
           pass
       print "basic_multi_auth(pass-exp). ERROR: %s" % msg

   def _utc_now(self):
       cal = GregorianCalendar(TimeZone.getTimeZone("UTC"))
       return cal.getTime()

   def _parse_expiration_as_date(self, raw):
       if raw is None:
           return None
       if isinstance(raw, Date):
           return raw

       s = str(raw).strip()
       if len(s) == 0:
           return None

       s_norm = re.sub(r"\.\d+Z$", "Z", s)

       try:
           if re.match(r"^\d{14}Z$", s_norm):
               f = SimpleDateFormat("yyyyMMddHHmmss'Z'")
               f.setTimeZone(TimeZone.getTimeZone("UTC"))
               return f.parse(s_norm)
       except:
           pass

       try:
           if re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$", s):
               f = SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'")
               f.setTimeZone(TimeZone.getTimeZone("UTC"))
               return f.parse(s)
       except:
           pass

       try:
           if re.match(r"^\d{4}-\d{2}-\d{2}$", s):
               # If you prefer end-of-day, adjust here
               f = SimpleDateFormat("yyyy-MM-dd")
               f.setTimeZone(TimeZone.getTimeZone("UTC"))
               d = f.parse(s)
               return d
       except:
           pass

       return None

   def loadAuthConfiguration(self, authConfigurationFile):
       try:
           f = open(authConfigurationFile, 'r')
           try:
               return json.loads(f.read())
           finally:
               f.close()
       except:
           print "basic_multi_auth(pass-exp). Failed to load config: %s" % authConfigurationFile
           return None

   def validateAuthConfiguration(self, authConfiguration):
       if "ldap_configuration" not in authConfiguration:
           print "basic_multi_auth(pass-exp). Invalid config (no ldap_configuration)"
           return False

       for ldapConfiguration in authConfiguration["ldap_configuration"]:
           if not self.containsAttributeString(ldapConfiguration, "configId"): return False
           if not self.containsAttributeArray(ldapConfiguration, "servers"): return False
           if ("bindDN" in ldapConfiguration) and (not self.containsAttributeString(ldapConfiguration, "bindPassword")): return False
           if not self.containsAttributeString(ldapConfiguration, "useSSL"): return False
           if not self.containsAttributeString(ldapConfiguration, "maxConnections"): return False
           if not self.containsAttributeArray(ldapConfiguration, "baseDNs"): return False
           if not self.containsAttributeArray(ldapConfiguration, "loginAttributes"): return False
           if not self.containsAttributeArray(ldapConfiguration, "localLoginAttributes"): return False
           if len(ldapConfiguration["loginAttributes"]) != len(ldapConfiguration["localLoginAttributes"]): return False
       return True

   def createLdapExtendedEntryManagers(self, authConfiguration):
       ldapExtendedConfigurations = self.createLdapExtendedConfigurations(authConfiguration)

       persistanceFactoryService = CdiUtil.bean(PersistanceFactoryService)
       ldapEntryManagerFactory = persistanceFactoryService.getPersistenceEntryManagerFactory(LdapEntryManagerFactory)
       persistenceType = ldapEntryManagerFactory.getPersistenceType()

       ldapExtendedEntryManagers = []
       for ldapExtendedConfiguration in ldapExtendedConfigurations:
           connectionConfiguration = ldapExtendedConfiguration["connectionConfiguration"]
           ldapConfiguration = ldapExtendedConfiguration["ldapConfiguration"]

           ldapProperties = Properties()
           for key, value in connectionConfiguration.items():
               value_string = ", ".join(value) if isinstance(value, list) else str(value)
               ldapProperties.setProperty(persistenceType + "#" + key, value_string)

           if StringHelper.isNotEmptyString(ldapConfiguration.getBindPassword()):
               ldapProperties.setProperty(persistenceType + "#bindPassword", ldapConfiguration.getBindPassword())

           ldapEntryManager = ldapEntryManagerFactory.createEntryManager(ldapProperties)
           ldapExtendedEntryManagers.append({
               "ldapConfiguration": ldapConfiguration,
               "ldapProperties": ldapProperties,
               "loginAttributes": ldapExtendedConfiguration["loginAttributes"],
               "localLoginAttributes": ldapExtendedConfiguration["localLoginAttributes"],
               "ldapEntryManager": ldapEntryManager
           })
       return ldapExtendedEntryManagers

   def createLdapExtendedConfigurations(self, authConfiguration):
       ldapExtendedConfigurations = []
       for connectionConfiguration in authConfiguration["ldap_configuration"]:
           configId = connectionConfiguration["configId"]
           servers = connectionConfiguration["servers"]

           bindDN = None
           bindPassword = None
           useAnonymousBind = True
           if self.containsAttributeString(connectionConfiguration, "bindDN"):
               useAnonymousBind = False
               bindDN = connectionConfiguration["bindDN"]
               bindPassword = CdiUtil.bean(EncryptionService).decrypt(connectionConfiguration["bindPassword"])

           useSSL = connectionConfiguration["useSSL"]
           maxConnections = connectionConfiguration["maxConnections"]
           baseDNs = connectionConfiguration["baseDNs"]
           loginAttributes = connectionConfiguration["loginAttributes"]
           localLoginAttributes = connectionConfiguration["localLoginAttributes"]

           ldapConfiguration = GluuLdapConfiguration()
           ldapConfiguration.setConfigId(configId)
           ldapConfiguration.setBindDN(bindDN)
           ldapConfiguration.setBindPassword(bindPassword)
           ldapConfiguration.setServers(Arrays.asList(servers))
           ldapConfiguration.setMaxConnections(maxConnections)
           ldapConfiguration.setUseSSL(useSSL)
           ldapConfiguration.setBaseDNs(Arrays.asList(baseDNs))
           ldapConfiguration.setPrimaryKey(loginAttributes[0])
           ldapConfiguration.setLocalPrimaryKey(localLoginAttributes[0])
           ldapConfiguration.setUseAnonymousBind(useAnonymousBind)

           ldapExtendedConfigurations.append({
               "ldapConfiguration": ldapConfiguration,
               "connectionConfiguration": connectionConfiguration,
               "loginAttributes": loginAttributes,
               "localLoginAttributes": localLoginAttributes
           })
       return ldapExtendedConfigurations

   def containsAttributeString(self, dictionary, attribute):
       return ((attribute in dictionary) and StringHelper.isNotEmptyString(dictionary[attribute]))

   def containsAttributeArray(self, dictionary, attribute):
       return ((attribute in dictionary) and (len(dictionary[attribute]) > 0))
Was this article useful?
Like
Dislike
Help us improve this page
Please provide feedback or comments
Access denied
Access denied