130 lines
3.1 KiB
Python
130 lines
3.1 KiB
Python
|
|
from ldap3 import (
|
||
|
|
Server,
|
||
|
|
Connection,
|
||
|
|
ALL,
|
||
|
|
NTLM,
|
||
|
|
SIMPLE,
|
||
|
|
SUBTREE
|
||
|
|
)
|
||
|
|
|
||
|
|
from flask import current_app
|
||
|
|
from app.config import Config
|
||
|
|
|
||
|
|
|
||
|
|
class LDAPService:
|
||
|
|
"""
|
||
|
|
LDAP / Active Directory Authentication Service
|
||
|
|
"""
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def authenticate(username, password):
|
||
|
|
"""
|
||
|
|
Authenticate LDAP User
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
{
|
||
|
|
"success": True,
|
||
|
|
"user": {
|
||
|
|
"username": "...",
|
||
|
|
"name": "...",
|
||
|
|
"email": "..."
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
OR
|
||
|
|
|
||
|
|
{
|
||
|
|
"success": False,
|
||
|
|
"message": "Invalid username or password"
|
||
|
|
}
|
||
|
|
"""
|
||
|
|
|
||
|
|
if not username or not password:
|
||
|
|
return {
|
||
|
|
"success": False,
|
||
|
|
"message": "Username and Password are required."
|
||
|
|
}
|
||
|
|
|
||
|
|
try:
|
||
|
|
|
||
|
|
# -----------------------------------
|
||
|
|
# LDAP SERVER
|
||
|
|
# -----------------------------------
|
||
|
|
server = Server(
|
||
|
|
Config.LDAP_SERVER,
|
||
|
|
port=Config.LDAP_PORT,
|
||
|
|
use_ssl=Config.LDAP_USE_SSL,
|
||
|
|
get_info=ALL
|
||
|
|
)
|
||
|
|
|
||
|
|
# -----------------------------------
|
||
|
|
# Login Format
|
||
|
|
#
|
||
|
|
# username@domain.com
|
||
|
|
# -----------------------------------
|
||
|
|
user_dn = f"{username}@{Config.LDAP_DOMAIN}"
|
||
|
|
|
||
|
|
conn = Connection(
|
||
|
|
server,
|
||
|
|
user=user_dn,
|
||
|
|
password=password,
|
||
|
|
authentication=SIMPLE,
|
||
|
|
auto_bind=True
|
||
|
|
)
|
||
|
|
|
||
|
|
# -----------------------------------
|
||
|
|
# Search User
|
||
|
|
# -----------------------------------
|
||
|
|
search_filter = f"(sAMAccountName={username})"
|
||
|
|
|
||
|
|
conn.search(
|
||
|
|
search_base=Config.LDAP_SEARCH_BASE,
|
||
|
|
search_filter=search_filter,
|
||
|
|
search_scope=SUBTREE,
|
||
|
|
attributes=[
|
||
|
|
"displayName",
|
||
|
|
"mail",
|
||
|
|
"givenName",
|
||
|
|
"sn",
|
||
|
|
"cn"
|
||
|
|
]
|
||
|
|
)
|
||
|
|
|
||
|
|
display_name = username
|
||
|
|
email = ""
|
||
|
|
|
||
|
|
if conn.entries:
|
||
|
|
|
||
|
|
entry = conn.entries[0]
|
||
|
|
|
||
|
|
if "displayName" in entry:
|
||
|
|
display_name = str(entry.displayName)
|
||
|
|
|
||
|
|
if "mail" in entry:
|
||
|
|
email = str(entry.mail)
|
||
|
|
|
||
|
|
conn.unbind()
|
||
|
|
|
||
|
|
current_app.logger.info(
|
||
|
|
f"LDAP Login Success : {username}"
|
||
|
|
)
|
||
|
|
|
||
|
|
return {
|
||
|
|
"success": True,
|
||
|
|
"user": {
|
||
|
|
"username": username,
|
||
|
|
"name": display_name,
|
||
|
|
"email": email
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
except Exception as ex:
|
||
|
|
|
||
|
|
current_app.logger.warning(
|
||
|
|
f"LDAP Login Failed : {username} : {str(ex)}"
|
||
|
|
)
|
||
|
|
|
||
|
|
return {
|
||
|
|
"success": False,
|
||
|
|
"message": "Invalid Username or Password."
|
||
|
|
}
|