73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
from flask import current_app
|
|
from ldap3 import Server, Connection, ALL, SUBTREE
|
|
from ldap3.core.exceptions import LDAPException
|
|
|
|
|
|
class LDAPService:
|
|
"""
|
|
Handles authentication against an LDAP / OpenLDAP server using the
|
|
standard "search + bind" pattern:
|
|
1. Bind with a service/admin account just to SEARCH for the user's DN.
|
|
2. Re-bind using that DN + the password the user typed, to verify it.
|
|
The user's typed password is only ever used in step 2, never sent
|
|
anywhere else.
|
|
"""
|
|
|
|
@staticmethod
|
|
def authenticate(username, password):
|
|
if not username or not password:
|
|
return None
|
|
|
|
server = Server(current_app.config["LDAP_SERVER"], get_info=ALL)
|
|
|
|
# --- Step 1: bind as the admin/service account to search the directory ---
|
|
try:
|
|
admin_conn = Connection(
|
|
server,
|
|
user=current_app.config["LDAP_BIND_DN"],
|
|
password=current_app.config["LDAP_BIND_PASSWORD"],
|
|
auto_bind=True,
|
|
)
|
|
except LDAPException as e:
|
|
current_app.logger.error(f"LDAP service account bind failed: {e}")
|
|
return None
|
|
|
|
# --- Step 2: find the user's real DN + profile attributes ---
|
|
try:
|
|
search_filter = current_app.config["LDAP_SEARCH_FILTER"].format(username=username)
|
|
admin_conn.search(
|
|
search_base=current_app.config["LDAP_BASE_DN"],
|
|
search_filter=search_filter,
|
|
search_scope=SUBTREE,
|
|
attributes=["cn", "mail", "uid"],
|
|
)
|
|
except LDAPException as e:
|
|
current_app.logger.error(f"LDAP search failed for '{username}': {e}")
|
|
admin_conn.unbind()
|
|
return None
|
|
|
|
if not admin_conn.entries:
|
|
current_app.logger.warning(f"LDAP user not found: {username}")
|
|
admin_conn.unbind()
|
|
return None
|
|
|
|
entry = admin_conn.entries[0]
|
|
user_dn = entry.entry_dn
|
|
name = str(entry.cn) if "cn" in entry and entry.cn.value else username
|
|
email = (
|
|
str(entry.mail)
|
|
if "mail" in entry and entry.mail.value
|
|
else f"{username}@{current_app.config['LDAP_DOMAIN']}"
|
|
)
|
|
admin_conn.unbind()
|
|
|
|
# --- Step 3: the actual auth check - bind AS the user with their password ---
|
|
try:
|
|
user_conn = Connection(server, user=user_dn, password=password, auto_bind=True)
|
|
user_conn.unbind()
|
|
except LDAPException as e:
|
|
current_app.logger.warning(f"LDAP authentication failed for '{username}': {e}")
|
|
return None
|
|
|
|
return {"username": username, "name": name, "email": email}
|