ebbe.bassandClaude Sonnet 5.5
b87ffae94c
The password change must stay below the domain's lockout threshold. The policy now carries lockout_threshold and lockout_window, from the domain or from the fine-grained policy that applies. AD-2's read allow-list names the new attributes. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N3t4WL7bd9bdHHNjEsxzqC
288 lines
12 KiB
Python
288 lines
12 KiB
Python
"""The LDAPS implementation. The only module in the portal that imports `ldap` (AD-1, AD-2)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from contextlib import contextmanager, suppress
|
|
from dataclasses import replace
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import ModuleType
|
|
|
|
from portal.config.env import DirectoryConfig
|
|
|
|
from .base import (
|
|
POLICY_ATTRIBUTES,
|
|
PSO_ATTRIBUTES,
|
|
READABLE_ATTRIBUTES,
|
|
WRITABLE_ATTRIBUTES,
|
|
DirectoryUnavailable,
|
|
DirectoryUser,
|
|
Forbidden,
|
|
PasswordPolicy,
|
|
PasswordRejected,
|
|
ValidationFailed,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TIMEOUT_SECONDS = 10
|
|
UF_ACCOUNTDISABLE = 0x2
|
|
UF_LOCKOUT = 0x10
|
|
COMPLEX_PASSWORDS = 0x1
|
|
# Windows FILETIME counts 100 ns ticks since 1601-01-01.
|
|
_FILETIME_EPOCH = datetime(1601, 1, 1, tzinfo=UTC)
|
|
_NEVER = -(2**63)
|
|
_AD_ERROR_CODE = re.compile(r"\b0000([0-9A-Fa-f]{4}):")
|
|
_WRONG_CURRENT_CODE = "0056" # ERROR_INVALID_PASSWORD; a policy refusal is 052D
|
|
|
|
|
|
def _load_ldap() -> ModuleType:
|
|
# Imported on first use so the rest of the portal (and its tests) load without the C library.
|
|
import ldap
|
|
import ldap.filter
|
|
|
|
return ldap
|
|
|
|
|
|
def _normalize_dn(dn: str) -> str:
|
|
return re.sub(r"\s*([,=])\s*", r"\1", dn).casefold()
|
|
|
|
|
|
def _text(attrs: dict[str, list[bytes]], name: str) -> str:
|
|
values = attrs.get(name)
|
|
return values[0].decode("utf-8") if values else ""
|
|
|
|
|
|
def _filetime(value: str) -> datetime | None:
|
|
ticks = int(value or 0)
|
|
if ticks <= 0:
|
|
return None
|
|
return _FILETIME_EPOCH + timedelta(microseconds=ticks // 10)
|
|
|
|
|
|
def _interval(value: str) -> timedelta | None:
|
|
"""AD stores durations as negative 100 ns ticks; 0 and the minimum value mean 'never'."""
|
|
ticks = int(value or 0)
|
|
if ticks == 0 or ticks == _NEVER:
|
|
return None
|
|
return timedelta(microseconds=abs(ticks) // 10)
|
|
|
|
|
|
class LdapDirectory:
|
|
def __init__(self, config: DirectoryConfig, ldap_module: ModuleType | None = None) -> None:
|
|
self._config = config
|
|
self._ldap = ldap_module or _load_ldap()
|
|
host = config.server
|
|
self._url = f"ldaps://{host}" if ":" in host else f"ldaps://{host}:636"
|
|
|
|
# -- connections -----------------------------------------------------------
|
|
|
|
def _connect(self):
|
|
ldap = self._ldap
|
|
conn = ldap.initialize(self._url)
|
|
conn.set_option(ldap.OPT_PROTOCOL_VERSION, 3)
|
|
conn.set_option(ldap.OPT_REFERRALS, 0)
|
|
conn.set_option(ldap.OPT_NETWORK_TIMEOUT, TIMEOUT_SECONDS)
|
|
conn.set_option(ldap.OPT_TIMEOUT, TIMEOUT_SECONDS)
|
|
conn.set_option(ldap.OPT_X_TLS_CACERTFILE, str(self._config.ca_file))
|
|
conn.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_DEMAND)
|
|
conn.set_option(ldap.OPT_X_TLS_NEWCTX, 0) # apply the options above to this connection
|
|
return conn
|
|
|
|
@contextmanager
|
|
def _service(self):
|
|
"""A bound connection of the service account; library errors become typed errors."""
|
|
ldap = self._ldap
|
|
conn = self._connect()
|
|
try:
|
|
conn.simple_bind_s(self._config.bind_dn, self._config.bind_password)
|
|
yield conn
|
|
except ldap.INSUFFICIENT_ACCESS as exc:
|
|
raise Forbidden("The service account lacks the right for this request") from exc
|
|
except ldap.LDAPError as exc:
|
|
# Name the class only: messages can echo a bind DN or a filter.
|
|
logger.warning("Directory request failed: %s", type(exc).__name__)
|
|
raise DirectoryUnavailable("The directory is not available") from exc
|
|
finally:
|
|
with suppress(self._ldap.LDAPError):
|
|
conn.unbind_s()
|
|
|
|
def _search(self, conn, base: str, scope: int, flt: str, attrs: tuple[str, ...]):
|
|
found = conn.search_s(base, scope, flt, list(attrs))
|
|
return [(dn, a) for dn, a in found if dn is not None] # drop referrals
|
|
|
|
def _find_entry(self, conn, sam: str):
|
|
"""The (dn, attributes) of the user in the configured OU, or None."""
|
|
if not sam:
|
|
return None
|
|
ldap = self._ldap
|
|
escaped = ldap.filter.escape_filter_chars(sam)
|
|
flt = f"(&(objectClass=user)(objectCategory=person)(sAMAccountName={escaped}))"
|
|
entries = self._search(
|
|
conn, self._config.user_ou_dn, ldap.SCOPE_SUBTREE, flt, READABLE_ATTRIBUTES
|
|
)
|
|
if len(entries) != 1:
|
|
if len(entries) > 1:
|
|
logger.warning("Directory lookup was ambiguous; treating the user as not found")
|
|
return None
|
|
return entries[0]
|
|
|
|
def _require_entry(self, conn, sam: str):
|
|
entry = self._find_entry(conn, sam)
|
|
if entry is None:
|
|
raise ValidationFailed("The user does not exist in the configured OU")
|
|
return entry
|
|
|
|
# -- interface ---------------------------------------------------------------
|
|
|
|
def find_user(self, sam: str) -> DirectoryUser | None:
|
|
with self._service() as conn:
|
|
entry = self._find_entry(conn, sam)
|
|
if entry is None:
|
|
return None
|
|
attrs = entry[1]
|
|
computed = _text(attrs, "msDS-User-Account-Control-Computed")
|
|
if computed:
|
|
locked = bool(int(computed) & UF_LOCKOUT)
|
|
else:
|
|
locked = self._locked_by_time(conn, _text(attrs, "lockoutTime"))
|
|
groups = {_normalize_dn(v.decode("utf-8")) for v in attrs.get("memberOf", [])}
|
|
return DirectoryUser(
|
|
sam=_text(attrs, "sAMAccountName"),
|
|
given_name=_text(attrs, "givenName"),
|
|
surname=_text(attrs, "sn"),
|
|
display_name=_text(attrs, "displayName"),
|
|
mail=_text(attrs, "mail"),
|
|
telephone=_text(attrs, "telephoneNumber"),
|
|
mobile=_text(attrs, "mobile"),
|
|
enabled=not int(_text(attrs, "userAccountControl") or 0) & UF_ACCOUNTDISABLE,
|
|
locked=locked,
|
|
password_last_set=_filetime(_text(attrs, "pwdLastSet")),
|
|
is_publisher=_normalize_dn(self._config.publisher_group_dn) in groups,
|
|
is_admin=_normalize_dn(self._config.admin_group_dn) in groups,
|
|
)
|
|
|
|
def _locked_by_time(self, conn, lockout_time: str) -> bool:
|
|
"""Fallback when the computed account-control attribute is not readable."""
|
|
locked_at = _filetime(lockout_time)
|
|
if locked_at is None:
|
|
return False
|
|
domain_dn = self._domain_dn(conn)
|
|
policy = self._search(
|
|
conn, domain_dn, self._ldap.SCOPE_BASE, "(objectClass=*)", POLICY_ATTRIBUTES
|
|
)
|
|
duration = _interval(_text(policy[0][1], "lockoutDuration")) if policy else None
|
|
if duration is None: # 'never unlock': only an administrator clears the lock
|
|
return True
|
|
return datetime.now(UTC) < locked_at + duration
|
|
|
|
def _domain_dn(self, conn) -> str:
|
|
root = self._search(
|
|
conn, "", self._ldap.SCOPE_BASE, "(objectClass=*)", ("defaultNamingContext",)
|
|
)
|
|
return _text(root[0][1], "defaultNamingContext")
|
|
|
|
def verify_password(self, sam: str, password: str) -> bool:
|
|
if not password:
|
|
return False # an empty password would be an anonymous bind, which succeeds
|
|
ldap = self._ldap
|
|
with self._service() as conn:
|
|
entry = self._find_entry(conn, sam)
|
|
if entry is None:
|
|
return False
|
|
user_conn = self._connect()
|
|
try:
|
|
user_conn.simple_bind_s(entry[0], password)
|
|
return True
|
|
except ldap.INVALID_CREDENTIALS:
|
|
return False
|
|
except ldap.LDAPError as exc:
|
|
logger.warning("Directory request failed: %s", type(exc).__name__)
|
|
raise DirectoryUnavailable("The directory is not available") from exc
|
|
finally:
|
|
with suppress(ldap.LDAPError):
|
|
user_conn.unbind_s()
|
|
|
|
def password_policy(self, sam: str) -> PasswordPolicy:
|
|
ldap = self._ldap
|
|
with self._service() as conn:
|
|
entry = self._require_entry(conn, sam)
|
|
domain_dn = self._domain_dn(conn)
|
|
domain = self._search(
|
|
conn, domain_dn, ldap.SCOPE_BASE, "(objectClass=*)", POLICY_ATTRIBUTES
|
|
)[0][1]
|
|
policy = PasswordPolicy(
|
|
min_length=int(_text(domain, "minPwdLength") or 0),
|
|
complexity=bool(int(_text(domain, "pwdProperties") or 0) & COMPLEX_PASSWORDS),
|
|
history_length=int(_text(domain, "pwdHistoryLength") or 0),
|
|
min_age=_interval(_text(domain, "minPwdAge")) or timedelta(0),
|
|
max_age=_interval(_text(domain, "maxPwdAge")),
|
|
lockout_threshold=int(_text(domain, "lockoutThreshold") or 0),
|
|
lockout_window=_interval(_text(domain, "lockoutObservationWindow")),
|
|
)
|
|
pso_dn = _text(entry[1], "msDS-ResultantPSO")
|
|
if not pso_dn:
|
|
return policy
|
|
try:
|
|
pso = self._search(conn, pso_dn, ldap.SCOPE_BASE, "(objectClass=*)", PSO_ATTRIBUTES)
|
|
except ldap.INSUFFICIENT_ACCESS:
|
|
return replace(policy, fine_grained=True, fine_grained_readable=False)
|
|
attrs = pso[0][1]
|
|
return PasswordPolicy(
|
|
min_length=int(_text(attrs, "msDS-MinimumPasswordLength") or 0),
|
|
complexity=_text(attrs, "msDS-PasswordComplexityEnabled").upper() == "TRUE",
|
|
history_length=int(_text(attrs, "msDS-PasswordHistoryLength") or 0),
|
|
min_age=_interval(_text(attrs, "msDS-MinimumPasswordAge")) or timedelta(0),
|
|
max_age=_interval(_text(attrs, "msDS-MaximumPasswordAge")),
|
|
fine_grained=True,
|
|
lockout_threshold=int(_text(attrs, "msDS-LockoutThreshold") or 0),
|
|
lockout_window=_interval(_text(attrs, "msDS-LockoutObservationWindow")),
|
|
)
|
|
|
|
def update_attributes(self, sam: str, changes: dict[str, str | None]) -> None:
|
|
ldap = self._ldap
|
|
refused = [name for name in changes if name not in WRITABLE_ATTRIBUTES]
|
|
if refused:
|
|
raise Forbidden(f"Attributes are not writable: {', '.join(sorted(refused))}")
|
|
mods = []
|
|
for name, value in changes.items():
|
|
if value is not None and (not isinstance(value, str) or "\x00" in value):
|
|
raise ValidationFailed(f"Invalid value for {name}")
|
|
values = [value.encode("utf-8")] if value else []
|
|
mods.append((ldap.MOD_REPLACE, name, values))
|
|
if not mods:
|
|
return
|
|
with self._service() as conn:
|
|
dn = self._require_entry(conn, sam)[0]
|
|
try:
|
|
conn.modify_s(dn, mods)
|
|
except (ldap.CONSTRAINT_VIOLATION, ldap.INVALID_SYNTAX) as exc:
|
|
raise ValidationFailed("The directory refused a value") from exc
|
|
|
|
def change_password(self, sam: str, current: str, new: str) -> None:
|
|
ldap = self._ldap
|
|
if not current or not new:
|
|
raise PasswordRejected("wrong_current" if not current else "policy")
|
|
old_value = f'"{current}"'.encode("utf-16-le")
|
|
new_value = f'"{new}"'.encode("utf-16-le")
|
|
mods = [
|
|
(ldap.MOD_DELETE, "unicodePwd", [old_value]),
|
|
(ldap.MOD_ADD, "unicodePwd", [new_value]),
|
|
]
|
|
with self._service() as conn:
|
|
dn = self._require_entry(conn, sam)[0]
|
|
try:
|
|
conn.modify_s(dn, mods)
|
|
except (ldap.CONSTRAINT_VIOLATION, ldap.UNWILLING_TO_PERFORM) as exc:
|
|
raise PasswordRejected(self._password_reason(exc)) from exc
|
|
|
|
@staticmethod
|
|
def _password_reason(exc: Exception) -> str:
|
|
detail = exc.args[0] if exc.args and isinstance(exc.args[0], dict) else {}
|
|
info = str(detail.get("info", ""))
|
|
match = _AD_ERROR_CODE.search(info)
|
|
if match and match.group(1).upper() == _WRONG_CURRENT_CODE:
|
|
return "wrong_current"
|
|
return "policy"
|