mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
29e6d28205
lldap and OpenLDAP's memberof overlay omit memberOf from "*", so every lldap login fell through to the default group. Request memberOf by name when the schema defines it, and on non-AD directories also search the directory root for groupOfNames/groupOfUniqueNames entries listing the user, since groups often sit outside the user search base and the overlay tracks only one group class. Also: skip ldap3's anonymous schema read after StartTLS, which AD and Samba AD reject, so StartTLS works there; reword a server's StartTLS refusal with an LDAPS hint; stop the bundle sanitizer masking part of an OID as an IP; skip the sync right after auto-provisioning so the default-group warning logs once.
596 lines
23 KiB
Python
596 lines
23 KiB
Python
"""LDAP authentication service for BamBuddy (#794).
|
|
|
|
Supports:
|
|
- LDAP bind authentication (simple bind with user's credentials)
|
|
- StartTLS, LDAPS, and plaintext connections
|
|
- User search with configurable filter
|
|
- Group membership resolution for role mapping
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass
|
|
|
|
from ldap3 import ALL, SUBTREE, Connection, Server, Tls
|
|
from ldap3.core.exceptions import LDAPException, LDAPObjectClassError, LDAPOperationResult
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class LDAPUserInfo:
|
|
"""User information retrieved from LDAP after successful authentication."""
|
|
|
|
username: str
|
|
email: str | None
|
|
display_name: str | None
|
|
groups: list[str] # List of group DNs the user belongs to
|
|
|
|
|
|
@dataclass
|
|
class LDAPSearchResult:
|
|
"""A directory user returned by the admin search endpoint (no auth performed)."""
|
|
|
|
username: str
|
|
email: str | None
|
|
display_name: str | None
|
|
dn: str
|
|
|
|
|
|
@dataclass
|
|
class LDAPConfig:
|
|
"""LDAP configuration parsed from settings."""
|
|
|
|
server_url: str
|
|
bind_dn: str
|
|
bind_password: str
|
|
search_base: str
|
|
user_filter: str # e.g. "(sAMAccountName={username})"
|
|
security: str # "none", "starttls", "ldaps"
|
|
group_mapping: dict[str, str] # LDAP group DN -> BamBuddy group name
|
|
auto_provision: bool
|
|
ca_cert_path: str # Path to CA certificate file (empty = skip verification)
|
|
default_group: str # Fallback BamBuddy group assigned when user has no mapped groups (empty = no fallback)
|
|
|
|
|
|
def parse_ldap_config(settings: dict[str, str]) -> LDAPConfig | None:
|
|
"""Parse LDAP config from settings key-value pairs. Returns None if LDAP not enabled."""
|
|
if settings.get("ldap_enabled", "false").lower() != "true":
|
|
return None
|
|
|
|
server_url = settings.get("ldap_server_url", "").strip()
|
|
if not server_url:
|
|
return None
|
|
|
|
group_mapping_raw = settings.get("ldap_group_mapping", "")
|
|
try:
|
|
group_mapping = json.loads(group_mapping_raw) if group_mapping_raw else {}
|
|
except json.JSONDecodeError:
|
|
group_mapping = {}
|
|
|
|
return LDAPConfig(
|
|
server_url=server_url,
|
|
bind_dn=settings.get("ldap_bind_dn", "").strip(),
|
|
bind_password=settings.get("ldap_bind_password", ""),
|
|
search_base=settings.get("ldap_search_base", "").strip(),
|
|
user_filter=settings.get("ldap_user_filter", "(sAMAccountName={username})").strip(),
|
|
security=settings.get("ldap_security", "starttls").strip(),
|
|
group_mapping=group_mapping if isinstance(group_mapping, dict) else {},
|
|
auto_provision=settings.get("ldap_auto_provision", "false").lower() == "true",
|
|
ca_cert_path=settings.get("ldap_ca_cert_path", "").strip(),
|
|
default_group=settings.get("ldap_default_group", "").strip(),
|
|
)
|
|
|
|
|
|
def _create_server(config: LDAPConfig) -> Server:
|
|
"""Create an ldap3 Server instance from config.
|
|
|
|
Always uses TLS — either LDAPS (TLS from start) or StartTLS (upgrade after connect).
|
|
Plaintext LDAP is not supported.
|
|
"""
|
|
import ssl
|
|
|
|
use_ssl = config.security == "ldaps" or config.server_url.startswith("ldaps://")
|
|
|
|
if config.ca_cert_path:
|
|
tls = Tls(validate=ssl.CERT_REQUIRED, ca_certs_file=config.ca_cert_path)
|
|
else:
|
|
tls = Tls(validate=ssl.CERT_NONE)
|
|
|
|
return Server(config.server_url, use_ssl=use_ssl, tls=tls, get_info=ALL, connect_timeout=10)
|
|
|
|
|
|
class LDAPStartTLSRefusedError(Exception):
|
|
"""The server answered the StartTLS request with an error result."""
|
|
|
|
|
|
def _start_tls_if_configured(conn: Connection, config: LDAPConfig) -> None:
|
|
"""Upgrade `conn` with StartTLS when the config asks for it.
|
|
|
|
A server that doesn't offer StartTLS (lldap, for one, only does LDAPS)
|
|
rejects the request with a bare "Unsupported extended operation" and the
|
|
StartTLS OID, which doesn't tell the admin what to change (#3197). Only a
|
|
rejection by the server is reworded; TLS handshake and certificate failures
|
|
keep their own messages.
|
|
|
|
ldap3 re-reads the server's info and schema straight after StartTLS by
|
|
default, still unauthenticated. A directory that refuses anonymous
|
|
searches (Active Directory, Samba AD) answers that read with
|
|
"operationsError", which failed every StartTLS connection to it and would
|
|
otherwise be reported as a refused StartTLS here. bind() reads the same
|
|
info once authenticated, so the early read is skipped.
|
|
"""
|
|
if config.security != "starttls" or config.server_url.startswith("ldaps://"):
|
|
return
|
|
try:
|
|
conn.start_tls(read_server_info=False)
|
|
except LDAPOperationResult as e:
|
|
raise LDAPStartTLSRefusedError(
|
|
f"the server refused StartTLS ({e.description}). If it only offers LDAPS, "
|
|
"choose LDAPS and use its ldaps:// URL and port"
|
|
) from e
|
|
|
|
|
|
def _open_service_connection(config: LDAPConfig, server: Server, *, check_names: bool = True) -> Connection:
|
|
"""Open and bind a service-account LDAP connection. Raises on failure.
|
|
|
|
`check_names` toggles ldap3's client-side attribute-name validation. The
|
|
default keeps it on so typos in `user_filter` fail loudly. The fuzzy
|
|
directory search disables it because its fixed OR filter spans both AD-only
|
|
(sAMAccountName, displayName) and OpenLDAP-only attribute names — without
|
|
this bypass ldap3 throws `LDAPAttributeError` before any request is sent
|
|
on a directory whose schema doesn't define one of the names.
|
|
"""
|
|
conn = Connection(
|
|
server,
|
|
user=config.bind_dn,
|
|
password=config.bind_password,
|
|
auto_bind=False,
|
|
raise_exceptions=True,
|
|
read_only=True,
|
|
check_names=check_names,
|
|
)
|
|
conn.open()
|
|
_start_tls_if_configured(conn, config)
|
|
conn.bind()
|
|
return conn
|
|
|
|
|
|
# Group classes that list their members by DN, and the attribute each uses.
|
|
_MEMBER_DN_GROUP_CLASSES = (("groupOfNames", "member"), ("groupOfUniqueNames", "uniqueMember"))
|
|
|
|
|
|
def _schema_defines(names, name: str) -> bool | None:
|
|
"""Whether a schema dict (attribute types or object classes) has `name`.
|
|
|
|
None when the server published no schema. ldap3 then skips its client-side
|
|
name checks, so the caller can request the name without it raising.
|
|
"""
|
|
if not names:
|
|
return None
|
|
return name in names
|
|
|
|
|
|
def _user_search_attributes(server: Server) -> list[str]:
|
|
"""Attributes to request for a user entry: all user attributes plus memberOf.
|
|
|
|
`*` alone does not return memberOf on every directory. OpenLDAP's memberof
|
|
overlay makes it operational and lldap only returns it when asked by name
|
|
(#3197), so it has to be listed. But ldap3 rejects a requested name that the
|
|
server's schema doesn't define before anything is sent, even with
|
|
check_names off, so it is only listed when the schema has it (or publishes
|
|
no schema at all).
|
|
"""
|
|
schema = server.schema
|
|
has_member_of = _schema_defines(schema.attribute_types if schema else None, "memberOf")
|
|
return ["*"] if has_member_of is False else ["*", "memberOf"]
|
|
|
|
|
|
def _groups_base(server: Server, search_base: str) -> str:
|
|
"""The naming context that contains `search_base`, else `search_base`.
|
|
|
|
Groups usually live beside the users rather than under them (ou=groups next
|
|
to ou=people), so a group search from the user search base finds nothing.
|
|
"""
|
|
base_lower = search_base.lower()
|
|
contexts = server.info.naming_contexts if server.info and server.info.naming_contexts else []
|
|
for context in contexts:
|
|
context_lower = str(context).lower()
|
|
if base_lower == context_lower or base_lower.endswith("," + context_lower):
|
|
return str(context)
|
|
return search_base
|
|
|
|
|
|
# Root DSE capability that Active Directory (and Samba AD) advertises.
|
|
_ACTIVE_DIRECTORY_CAPABILITY = "1.2.840.113556.1.4.800"
|
|
|
|
|
|
def _is_active_directory(server: Server) -> bool:
|
|
info = server.info
|
|
features = info.supported_features if info and info.supported_features else []
|
|
return any(feature[0] == _ACTIVE_DIRECTORY_CAPABILITY for feature in features)
|
|
|
|
|
|
def _member_dn_groups(service_conn: Connection, config: LDAPConfig, user_dn: str) -> list[str]:
|
|
"""Find groupOfNames / groupOfUniqueNames entries that list `user_dn` as a member.
|
|
|
|
memberOf alone is not enough outside Active Directory. Plain OpenLDAP has
|
|
none without the memberof overlay; with it, the overlay tracks only the
|
|
group class it was configured for (osixia's image: groupOfUniqueNames, so
|
|
groupOfNames groups are missing) and only groups changed after it was
|
|
loaded. The groups themselves are the authority, so they are asked too.
|
|
|
|
Active Directory is skipped: it keeps memberOf complete itself, and its
|
|
groups are objectClass=group, not either class searched here, so the
|
|
search would cost a subtree walk from the domain root and find nothing.
|
|
|
|
Only the classes the schema defines go into the filter, for the same
|
|
client-side check the POSIX lookup trips over.
|
|
"""
|
|
schema = service_conn.server.schema
|
|
object_classes = schema.object_classes if schema else None
|
|
clauses = [
|
|
f"(&(objectClass={object_class})({attribute}={_ldap_escape(user_dn)}))"
|
|
for object_class, attribute in _MEMBER_DN_GROUP_CLASSES
|
|
if _schema_defines(object_classes, object_class) is not False
|
|
]
|
|
if not clauses:
|
|
return []
|
|
search_filter = clauses[0] if len(clauses) == 1 else f"(|{''.join(clauses)})"
|
|
try:
|
|
service_conn.search(
|
|
search_base=_groups_base(service_conn.server, config.search_base),
|
|
search_filter=search_filter,
|
|
search_scope=SUBTREE,
|
|
attributes=["cn"],
|
|
)
|
|
except LDAPException as e:
|
|
# The exception text can carry the user's DN (PII, #2681), so only its
|
|
# type is logged. The user still logs in, with memberOf and POSIX groups.
|
|
logger.warning("LDAP group membership lookup failed (%s); mapping without it", type(e).__name__)
|
|
return []
|
|
return [str(entry.entry_dn) for entry in service_conn.entries]
|
|
|
|
|
|
def _pick_canonical_username(entry, fallback: str) -> str:
|
|
"""Prefer sAMAccountName, then uid, then the supplied fallback."""
|
|
if hasattr(entry, "sAMAccountName") and entry.sAMAccountName:
|
|
return str(entry.sAMAccountName)
|
|
if hasattr(entry, "uid") and entry.uid:
|
|
return str(entry.uid)
|
|
return fallback
|
|
|
|
|
|
def _extract_user_info(
|
|
service_conn: Connection, config: LDAPConfig, user_entry, fallback_username: str
|
|
) -> LDAPUserInfo:
|
|
"""Build an LDAPUserInfo from an already-fetched directory entry.
|
|
|
|
Collects memberOf groups, the groupOfNames / groupOfUniqueNames entries
|
|
that list the user (except on Active Directory), POSIX
|
|
memberUid groups, and the primary gidNumber group; dedups DNs
|
|
case-insensitively. Uses the supplied service-bound connection for the
|
|
group searches. The entry must have been fetched with
|
|
`_user_search_attributes`, or memberOf may be missing from it.
|
|
"""
|
|
email = str(user_entry.mail) if hasattr(user_entry, "mail") and user_entry.mail else None
|
|
display_name = (
|
|
str(user_entry.displayName) if hasattr(user_entry, "displayName") and user_entry.displayName else None
|
|
)
|
|
|
|
# Collect groups from the memberOf attribute (Active Directory, lldap,
|
|
# OpenLDAP with the memberof overlay, 389-DS), then ask the groups too.
|
|
groups = [str(g) for g in user_entry.memberOf] if hasattr(user_entry, "memberOf") and user_entry.memberOf else []
|
|
if not _is_active_directory(service_conn.server):
|
|
groups.extend(_member_dn_groups(service_conn, config, str(user_entry.entry_dn)))
|
|
|
|
canonical_username = _pick_canonical_username(user_entry, fallback_username)
|
|
|
|
# Also search for POSIX groups, both the memberUid kind and the primary
|
|
# gidNumber kind. Both filters name the posixGroup object class, and ldap3
|
|
# validates that name against the schema it fetched at connect time
|
|
# (get_info=ALL) before it builds the request — so on a directory that
|
|
# publishes a schema without posixGroup it raises client-side and nothing is
|
|
# ever sent. A directory with no posixGroup class has no posixGroup entries,
|
|
# which is exactly the answer the searches would have returned, so the
|
|
# correct response is to carry on with the memberOf groups collected above.
|
|
#
|
|
# Left uncaught, that exception escaped authenticate_ldap_user, and the login
|
|
# route reports any LDAP error as "Incorrect username or password" — so an
|
|
# lldap user, whose accounts carry posixAccount but whose directory defines
|
|
# no group classes beyond groupOfNames, could never log in and had nothing
|
|
# but a wrong-password message to go on (#2769). This predates the primary
|
|
# gidNumber lookup: the memberUid filter has named the class since #794.
|
|
try:
|
|
posix_filter = f"(&(objectClass=posixGroup)(memberUid={_ldap_escape(canonical_username)}))"
|
|
service_conn.search(
|
|
search_base=config.search_base,
|
|
search_filter=posix_filter,
|
|
search_scope=SUBTREE,
|
|
attributes=["cn"],
|
|
)
|
|
for entry in service_conn.entries:
|
|
groups.append(str(entry.entry_dn))
|
|
|
|
# POSIX primary group: user's gidNumber matches a posixGroup's gidNumber.
|
|
# Standard Unix semantics treat this as full group membership, so we need
|
|
# to resolve it to a group DN alongside the memberUid results.
|
|
if hasattr(user_entry, "gidNumber") and user_entry.gidNumber:
|
|
primary_gid = str(user_entry.gidNumber)
|
|
primary_filter = f"(&(objectClass=posixGroup)(gidNumber={_ldap_escape(primary_gid)}))"
|
|
service_conn.search(
|
|
search_base=config.search_base,
|
|
search_filter=primary_filter,
|
|
search_scope=SUBTREE,
|
|
attributes=["cn"],
|
|
)
|
|
for entry in service_conn.entries:
|
|
groups.append(str(entry.entry_dn))
|
|
except LDAPObjectClassError:
|
|
# Logged once per authentication, at info: it is the explanation for a
|
|
# user's POSIX groups being absent from their mapping, and it is not an
|
|
# error the operator can or should act on.
|
|
logger.info(
|
|
"Directory publishes no posixGroup object class; skipping POSIX group lookup "
|
|
"(memberOf and groupOfNames groups are unaffected)"
|
|
)
|
|
|
|
# Dedupe group DNs (user may be in a group via both memberUid and primary gidNumber).
|
|
# Case-insensitive comparison — LDAP DNs are case-insensitive by spec.
|
|
seen_lower: set[str] = set()
|
|
deduped_groups: list[str] = []
|
|
for g in groups:
|
|
key = g.lower()
|
|
if key not in seen_lower:
|
|
seen_lower.add(key)
|
|
deduped_groups.append(g)
|
|
|
|
return LDAPUserInfo(
|
|
username=canonical_username,
|
|
email=email,
|
|
display_name=display_name,
|
|
groups=deduped_groups,
|
|
)
|
|
|
|
|
|
def authenticate_ldap_user(config: LDAPConfig, username: str, password: str) -> LDAPUserInfo | None:
|
|
"""Authenticate a user via LDAP bind.
|
|
|
|
1. Bind with service account to search for the user DN
|
|
2. Attempt bind with the user's DN and provided password
|
|
3. On success, retrieve user attributes and group memberships
|
|
|
|
Returns LDAPUserInfo on success, None on failure.
|
|
"""
|
|
if not password:
|
|
return None
|
|
|
|
server = _create_server(config)
|
|
|
|
try:
|
|
service_conn = _open_service_connection(config, server)
|
|
except Exception as e:
|
|
logger.warning("LDAP service account bind failed: %s", e)
|
|
return None
|
|
|
|
try:
|
|
# Search for the user
|
|
search_filter = config.user_filter.replace("{username}", _ldap_escape(username))
|
|
service_conn.search(
|
|
search_base=config.search_base,
|
|
search_filter=search_filter,
|
|
search_scope=SUBTREE,
|
|
attributes=_user_search_attributes(server),
|
|
)
|
|
|
|
if not service_conn.entries:
|
|
logger.info("LDAP user not found: %s", username)
|
|
return None
|
|
|
|
user_entry = service_conn.entries[0]
|
|
user_dn = str(user_entry.entry_dn)
|
|
|
|
# Step 2: Bind as the user to verify password
|
|
try:
|
|
user_conn = Connection(
|
|
server,
|
|
user=user_dn,
|
|
password=password,
|
|
auto_bind=False,
|
|
raise_exceptions=True,
|
|
read_only=True,
|
|
)
|
|
user_conn.open()
|
|
_start_tls_if_configured(user_conn, config)
|
|
user_conn.bind()
|
|
user_conn.unbind()
|
|
except Exception as e:
|
|
logger.info("LDAP bind failed for user %s: %s", username, e)
|
|
return None
|
|
|
|
info = _extract_user_info(service_conn, config, user_entry, username)
|
|
# Don't log the raw DN — its leaf CN is the user's real name (PII, #2681).
|
|
# The username + group count is enough to confirm a successful auth; the
|
|
# support-bundle sanitizer also redacts any DN that slips through (e.g. an
|
|
# ldap3 exception string), but keeping it out of the log at the source is
|
|
# the primary hygiene per the "no private data in logs" rule.
|
|
logger.info(
|
|
"LDAP authentication successful for user: %s (groups: %d)",
|
|
info.username,
|
|
len(info.groups),
|
|
)
|
|
return info
|
|
finally:
|
|
service_conn.unbind()
|
|
|
|
|
|
def lookup_ldap_user(config: LDAPConfig, username: str) -> LDAPUserInfo | None:
|
|
"""Look up a directory user by exact username via the service-account bind.
|
|
|
|
Performs no password verification — intended for the admin manual-provision
|
|
flow, where the caller has already been authenticated as a BamBuddy admin
|
|
and now needs the directory attributes (email, display name, group DNs)
|
|
to create the user.
|
|
|
|
Uses the same `user_filter` template that the login path uses, so anything
|
|
that logs in successfully via auto-provision is also resolvable here.
|
|
"""
|
|
server = _create_server(config)
|
|
|
|
try:
|
|
service_conn = _open_service_connection(config, server)
|
|
except Exception as e:
|
|
logger.warning("LDAP service account bind failed during lookup: %s", e)
|
|
raise
|
|
|
|
try:
|
|
search_filter = config.user_filter.replace("{username}", _ldap_escape(username))
|
|
service_conn.search(
|
|
search_base=config.search_base,
|
|
search_filter=search_filter,
|
|
search_scope=SUBTREE,
|
|
attributes=_user_search_attributes(server),
|
|
)
|
|
if not service_conn.entries:
|
|
logger.info("LDAP lookup: user not found: %s", username)
|
|
return None
|
|
return _extract_user_info(service_conn, config, service_conn.entries[0], username)
|
|
finally:
|
|
service_conn.unbind()
|
|
|
|
|
|
def search_ldap_users(config: LDAPConfig, query: str, limit: int = 25) -> list[LDAPSearchResult]:
|
|
"""Fuzzy search the directory for users matching `query`.
|
|
|
|
Uses a fixed OR filter across sAMAccountName, uid, mail, displayName, and
|
|
cn — covering both Active Directory and OpenLDAP layouts. The query is
|
|
RFC-4515 escaped so a typed `*` doesn't enumerate the whole directory.
|
|
Returns up to `limit` results (default 25). Service-bind failures raise so
|
|
the caller can surface a 503; "no matches" returns an empty list.
|
|
|
|
Callers should enforce a minimum query length (≥2 chars) — short queries
|
|
against a large directory are wasteful and effectively unbounded.
|
|
"""
|
|
query = query.strip()
|
|
if len(query) < 2:
|
|
return []
|
|
|
|
escaped = _ldap_escape(query)
|
|
search_filter = (
|
|
f"(|(sAMAccountName=*{escaped}*)(uid=*{escaped}*)(mail=*{escaped}*)(displayName=*{escaped}*)(cn=*{escaped}*))"
|
|
)
|
|
|
|
server = _create_server(config)
|
|
|
|
try:
|
|
# check_names=False so OpenLDAP directories (no sAMAccountName/displayName
|
|
# in schema) don't reject the cross-schema OR filter — see helper docstring.
|
|
service_conn = _open_service_connection(config, server, check_names=False)
|
|
except Exception as e:
|
|
logger.warning("LDAP service account bind failed during search: %s", e)
|
|
raise
|
|
|
|
try:
|
|
# attributes=["*"] requests all user attributes. We can't enumerate the
|
|
# AD/OpenLDAP-specific names (sAMAccountName, displayName) explicitly
|
|
# because ldap3 validates the attribute list against the server schema
|
|
# even with check_names=False — and OpenLDAP rejects the AD names. The
|
|
# `*` wildcard is hardcoded in ldap3's ATTRIBUTES_EXCLUDED_FROM_CHECK so
|
|
# it bypasses that validation, and the server returns whatever it has.
|
|
service_conn.search(
|
|
search_base=config.search_base,
|
|
search_filter=search_filter,
|
|
search_scope=SUBTREE,
|
|
attributes=["*"],
|
|
size_limit=limit,
|
|
)
|
|
results: list[LDAPSearchResult] = []
|
|
for entry in service_conn.entries:
|
|
username = _pick_canonical_username(entry, "")
|
|
if not username and hasattr(entry, "cn") and entry.cn:
|
|
# Last resort — some OpenLDAP layouts only have cn
|
|
username = str(entry.cn)
|
|
if not username:
|
|
continue
|
|
email = str(entry.mail) if hasattr(entry, "mail") and entry.mail else None
|
|
display_name = str(entry.displayName) if hasattr(entry, "displayName") and entry.displayName else None
|
|
results.append(
|
|
LDAPSearchResult(
|
|
username=username,
|
|
email=email,
|
|
display_name=display_name,
|
|
dn=str(entry.entry_dn),
|
|
)
|
|
)
|
|
logger.info("LDAP directory search for %r returned %d result(s)", query, len(results))
|
|
return results
|
|
finally:
|
|
service_conn.unbind()
|
|
|
|
|
|
def resolve_group_mapping(ldap_groups: list[str], group_mapping: dict[str, str]) -> list[str]:
|
|
"""Map LDAP group DNs to BamBuddy group names.
|
|
|
|
Returns list of BamBuddy group names that the user should be added to.
|
|
Comparison is case-insensitive on the LDAP group DN.
|
|
"""
|
|
if not group_mapping:
|
|
return []
|
|
|
|
# Build case-insensitive lookup
|
|
mapping_lower = {k.lower(): v for k, v in group_mapping.items()}
|
|
result = []
|
|
for ldap_group in ldap_groups:
|
|
bambuddy_group = mapping_lower.get(ldap_group.lower())
|
|
if bambuddy_group:
|
|
result.append(bambuddy_group)
|
|
return result
|
|
|
|
|
|
def test_ldap_connection(config: LDAPConfig) -> tuple[bool, str]:
|
|
"""Test LDAP connection and service account bind.
|
|
|
|
Returns (success, message).
|
|
"""
|
|
try:
|
|
server = _create_server(config)
|
|
conn = Connection(
|
|
server,
|
|
user=config.bind_dn,
|
|
password=config.bind_password,
|
|
auto_bind=False,
|
|
raise_exceptions=True,
|
|
read_only=True,
|
|
)
|
|
conn.open()
|
|
_start_tls_if_configured(conn, config)
|
|
conn.bind()
|
|
|
|
# Try a search to verify search base
|
|
conn.search(
|
|
search_base=config.search_base,
|
|
search_filter="(objectClass=*)",
|
|
search_scope=SUBTREE,
|
|
size_limit=1,
|
|
)
|
|
conn.unbind()
|
|
return True, "LDAP connection successful"
|
|
except Exception as e:
|
|
return False, f"LDAP connection failed: {e}"
|
|
|
|
|
|
def _ldap_escape(value: str) -> str:
|
|
"""Escape special characters in LDAP search filter values (RFC 4515)."""
|
|
replacements = {
|
|
"\\": "\\5c",
|
|
"*": "\\2a",
|
|
"(": "\\28",
|
|
")": "\\29",
|
|
"\x00": "\\00",
|
|
}
|
|
for char, escaped in replacements.items():
|
|
value = value.replace(char, escaped)
|
|
return value
|