From 96d3e771619bbf52ff0a021e6e15396bd5fc201f Mon Sep 17 00:00:00 2001 From: Ebbe Date: Sat, 17 Oct 2026 18:28:50 +0200 Subject: [PATCH] feat: add directory adapter LDAPS directory adapter with certificate verification, allow-listed reads and writes, escaped filters, password verification and change, password policy and account state, plus an in-memory fake. Adds the LDAP_* configuration, python-ldap with its image libraries, the CA file mount and an opt-in live test for the test OU. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01N3t4WL7bd9bdHHNjEsxzqC --- .env.example | 23 ++ Dockerfile | 10 +- README.md | 19 +- compose.yaml | 1 + portal/adapters/directory/__init__.py | 31 ++ portal/adapters/directory/base.py | 117 ++++++++ portal/adapters/directory/fake.py | 111 ++++++++ portal/adapters/directory/ldap_directory.py | 287 +++++++++++++++++++ portal/config/env.py | 42 ++- portal/config/settings.py | 1 + portal/tests/conftest.py | 9 + portal/tests/ldap_stub.py | 114 ++++++++ portal/tests/test_config.py | 35 +++ portal/tests/test_directory.py | 110 ++++++++ portal/tests/test_directory_live.py | 78 +++++ portal/tests/test_ldap_directory.py | 298 ++++++++++++++++++++ pyproject.toml | 8 + uv.lock | 33 +++ 18 files changed, 1324 insertions(+), 3 deletions(-) create mode 100644 portal/adapters/directory/__init__.py create mode 100644 portal/adapters/directory/base.py create mode 100644 portal/adapters/directory/fake.py create mode 100644 portal/adapters/directory/ldap_directory.py create mode 100644 portal/tests/ldap_stub.py create mode 100644 portal/tests/test_directory.py create mode 100644 portal/tests/test_directory_live.py create mode 100644 portal/tests/test_ldap_directory.py diff --git a/.env.example b/.env.example index 674adbf..4dbe410 100644 --- a/.env.example +++ b/.env.example @@ -7,6 +7,29 @@ SECRET_KEY= # Required. Comma-separated host names the portal answers to. ALLOWED_HOSTS=portal.example.test +# Required. Host name of the directory server; LDAPS (port 636) is always used. +# Must match a name in the server's certificate. Add :port to use another port. +LDAP_SERVER= + +# Required. Distinguished name of the OU whose users the portal may read and change. +LDAP_USER_OU_DN= + +# Required. The service account, as a distinguished name or user principal name. +LDAP_BIND_DN= +LDAP_BIND_PASSWORD= + +# Required. Path inside the container of the PEM file with the CA chain that signed the +# directory server's certificate. Mount the file read-only, for example at /certs/ca.pem. +LDAP_CA_FILE=/certs/ca.pem + +# Required for docker compose. Path on the Docker host of the CA chain file; compose mounts it +# read-only at /certs/ca.pem. (Only compose reads this; the portal does not.) +CA_FILE_HOST= + +# Required. Distinguished names of the groups that grant the Publisher and Admin roles. +LDAP_PUBLISHER_GROUP_DN= +LDAP_ADMIN_GROUP_DN= + # Optional. Directory for the database. Default /data (the container volume). # DATA_DIR=/data diff --git a/Dockerfile b/Dockerfile index 0f258e5..58c02bf 100644 --- a/Dockerfile +++ b/Dockerfile @@ -4,6 +4,9 @@ FROM python:3.13-slim AS builder COPY --from=ghcr.io/astral-sh/uv:0.13.1 /uv /usr/local/bin/uv ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy UV_PYTHON_DOWNLOADS=never WORKDIR /app +RUN apt-get update \ + && apt-get install -y --no-install-recommends build-essential libldap2-dev libsasl2-dev \ + && rm -rf /var/lib/apt/lists/* COPY pyproject.toml uv.lock .python-version README.md ./ RUN uv sync --frozen --no-dev --no-install-project COPY manage.py ./ @@ -11,11 +14,16 @@ COPY portal ./portal # Build-time only values so collectstatic can load settings; not used at runtime. RUN SECRET_KEY=build-time-only-value-not-used-at-runtime-000000 \ ALLOWED_HOSTS=localhost DATA_DIR=/tmp \ + LDAP_SERVER=build.invalid LDAP_USER_OU_DN=build LDAP_BIND_DN=build LDAP_BIND_PASSWORD=build \ + LDAP_CA_FILE=/app/README.md LDAP_PUBLISHER_GROUP_DN=build LDAP_ADMIN_GROUP_DN=build \ /app/.venv/bin/python manage.py collectstatic --noinput FROM python:3.13-slim ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1 PATH="/app/.venv/bin:$PATH" -RUN useradd --system --uid 10001 --no-create-home --shell /usr/sbin/nologin app \ +RUN apt-get update \ + && apt-get install -y --no-install-recommends libldap2 libsasl2-2 \ + && rm -rf /var/lib/apt/lists/* \ + && useradd --system --uid 10001 --no-create-home --shell /usr/sbin/nologin app \ && mkdir /data /backups \ && chown app:app /data /backups WORKDIR /app diff --git a/README.md b/README.md index a9c76d3..8b05eb1 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,13 @@ All configuration comes from environment variables, validated at startup. If any | `DEBUG` | no | `false` | `true` or `false` | | `TRUSTED_PROXIES` | no | empty | Comma-separated IP addresses or networks. Reserved for the client-IP rule in a later entry; currently has no effect | | `LOG_LEVEL` | no | `INFO` | `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL` | +| `LDAP_SERVER` | yes | | Directory server host name (optionally `:port`); LDAPS on 636 is always used and the name must be in the server's certificate | +| `LDAP_USER_OU_DN` | yes | | Distinguished name of the OU whose users the portal may read and change | +| `LDAP_BIND_DN` | yes | | Service account, as distinguished name or user principal name | +| `LDAP_BIND_PASSWORD` | yes | | Password of the service account | +| `LDAP_CA_FILE` | yes | | Path inside the container of the PEM file with the CA chain; certificate verification is never disabled | +| `LDAP_PUBLISHER_GROUP_DN` | yes | | Group whose direct members are Publishers | +| `LDAP_ADMIN_GROUP_DN` | yes | | Group whose direct members are Admins | | `DEFAULT_TIMEZONE` | no | `Europe/Berlin` | IANA zone used for display; storage is UTC | ## Run in a container @@ -21,7 +28,7 @@ All configuration comes from environment variables, validated at startup. If any The image is built and run on a Linux host from a checkout of this repository. ```sh -cp .env.example .env # then set SECRET_KEY and ALLOWED_HOSTS +cp .env.example .env # then set SECRET_KEY, ALLOWED_HOSTS, the LDAP_* values and CA_FILE_HOST docker compose up --build curl -H "Host: " http://localhost:8000/healthz # {"status":"ok"} ``` @@ -41,3 +48,13 @@ uv run ruff check . ``` `pytest` sets its own test environment (see `pyproject.toml`). + +### Directory adapter against the test OU + +`uv sync` builds `python-ldap`, which needs the OpenLDAP and SASL development libraries (the Docker image installs them). The automated tests use an in-memory fake and a stand-in for the `ldap` module, so they run anywhere. To check the real adapter against the test OU, run this on a Linux host with the libraries installed (never against the production OU): + +```sh +PORTAL_LIVE_LDAP=1 LIVE_LDAP_SERVER=... LIVE_LDAP_USER_OU_DN=... LIVE_LDAP_BIND_DN=... LIVE_LDAP_BIND_PASSWORD=... LIVE_LDAP_CA_FILE=... LIVE_LDAP_PUBLISHER_GROUP_DN=... LIVE_LDAP_ADMIN_GROUP_DN=... LIVE_LDAP_TEST_PASSWORD= uv run pytest portal/tests/test_directory_live.py -v +``` + +It restores what it writes and does not change any password. diff --git a/compose.yaml b/compose.yaml index 9edbee2..31edb71 100644 --- a/compose.yaml +++ b/compose.yaml @@ -7,6 +7,7 @@ services: volumes: - data:/data - backups:/backups + - ${CA_FILE_HOST:?set CA_FILE_HOST in .env}:/certs/ca.pem:ro read_only: true tmpfs: - /tmp diff --git a/portal/adapters/directory/__init__.py b/portal/adapters/directory/__init__.py new file mode 100644 index 0000000..ccc11dc --- /dev/null +++ b/portal/adapters/directory/__init__.py @@ -0,0 +1,31 @@ +"""Directory adapter: the only code that talks to Active Directory (AD-1).""" + +from .base import ( + READABLE_ATTRIBUTES, + WRITABLE_ATTRIBUTES, + Directory, + DirectoryError, + DirectoryUnavailable, + DirectoryUser, + Forbidden, + PasswordPolicy, + PasswordRejected, + ValidationFailed, +) +from .fake import FakeDirectory +from .ldap_directory import LdapDirectory + +__all__ = [ + "READABLE_ATTRIBUTES", + "WRITABLE_ATTRIBUTES", + "Directory", + "DirectoryError", + "DirectoryUnavailable", + "DirectoryUser", + "FakeDirectory", + "Forbidden", + "LdapDirectory", + "PasswordPolicy", + "PasswordRejected", + "ValidationFailed", +] diff --git a/portal/adapters/directory/base.py b/portal/adapters/directory/base.py new file mode 100644 index 0000000..14f2b95 --- /dev/null +++ b/portal/adapters/directory/base.py @@ -0,0 +1,117 @@ +"""The directory interface, its data types and its typed errors (AD-1, AD-2, AD-17).""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import Protocol + +# Fixed allow-lists (AD-2). unicodePwd is writable only through change_password. +WRITABLE_ATTRIBUTES = ( + "givenName", + "sn", + "displayName", + "mail", + "telephoneNumber", + "mobile", +) +READABLE_ATTRIBUTES = ( + *WRITABLE_ATTRIBUTES, + "sAMAccountName", + "memberOf", + "thumbnailPhoto", + "pwdLastSet", + "userAccountControl", + "msDS-User-Account-Control-Computed", + "lockoutTime", + "msDS-ResultantPSO", +) +POLICY_ATTRIBUTES = ( + "minPwdLength", + "pwdProperties", + "pwdHistoryLength", + "minPwdAge", + "maxPwdAge", + "lockoutDuration", +) +PSO_ATTRIBUTES = ( + "msDS-MinimumPasswordLength", + "msDS-PasswordComplexityEnabled", + "msDS-PasswordHistoryLength", + "msDS-MinimumPasswordAge", + "msDS-MaximumPasswordAge", +) + + +class DirectoryError(Exception): + """Base class of every error the directory adapter raises.""" + + +class DirectoryUnavailable(DirectoryError): + """The directory cannot be reached or refused the service account.""" + + +class Forbidden(DirectoryError): + """The request is outside the allow-list or the service account lacks the right.""" + + +class ValidationFailed(DirectoryError): + """The directory refused a value.""" + + +class PasswordRejected(ValidationFailed): + """A password change was refused; `reason` is `wrong_current` or `policy`.""" + + def __init__(self, reason: str) -> None: + self.reason = reason + super().__init__(f"Password change refused: {reason}") + + +@dataclass(frozen=True) +class DirectoryUser: + sam: str # exact sAMAccountName + given_name: str = "" + surname: str = "" + display_name: str = "" + mail: str = "" + telephone: str = "" + mobile: str = "" + enabled: bool = True + locked: bool = False + password_last_set: datetime | None = None # UTC; None when the user must change it at sign-in + is_publisher: bool = False + is_admin: bool = False + + @property + def must_change_password(self) -> bool: + return self.password_last_set is None + + +@dataclass(frozen=True) +class PasswordPolicy: + min_length: int + complexity: bool + history_length: int + min_age: timedelta + max_age: timedelta | None # None: passwords do not expire + fine_grained: bool = False # a fine-grained policy applies to the user + fine_grained_readable: bool = True # False: the values above are the domain's, not the user's + + +class Directory(Protocol): + """The only way the rest of the portal talks to the directory.""" + + def find_user(self, sam: str) -> DirectoryUser | None: + """Look the user up by sAMAccountName inside the configured OU.""" + + def verify_password(self, sam: str, password: str) -> bool: + """True when the user can bind with this password. Empty passwords are never valid.""" + + def password_policy(self, sam: str) -> PasswordPolicy: + """The policy that applies to this user (the domain's, or a fine-grained one).""" + + def update_attributes(self, sam: str, changes: dict[str, str | None]) -> None: + """Replace allow-listed attributes; None or an empty string clears one.""" + + def change_password(self, sam: str, current: str, new: str) -> None: + """Change the password as the user does, which needs the current password.""" diff --git a/portal/adapters/directory/fake.py b/portal/adapters/directory/fake.py new file mode 100644 index 0000000..acdd089 --- /dev/null +++ b/portal/adapters/directory/fake.py @@ -0,0 +1,111 @@ +"""In-memory directory with the same rules as the real one, for tests (AD-1).""" + +from __future__ import annotations + +from dataclasses import replace +from datetime import UTC, datetime, timedelta + +from .base import ( + WRITABLE_ATTRIBUTES, + DirectoryUnavailable, + DirectoryUser, + Forbidden, + PasswordPolicy, + PasswordRejected, + ValidationFailed, +) + +_FIELD = { + "givenName": "given_name", + "sn": "surname", + "displayName": "display_name", + "mail": "mail", + "telephoneNumber": "telephone", + "mobile": "mobile", +} +DEFAULT_POLICY = PasswordPolicy( + min_length=8, + complexity=True, + history_length=24, + min_age=timedelta(0), + max_age=timedelta(days=90), +) + + +def _character_classes(password: str) -> int: + tests = (str.isupper, str.islower, str.isdigit, lambda c: not c.isalnum()) + return sum(any(test(c) for c in password) for test in tests) + + +class FakeDirectory: + def __init__(self, policy: PasswordPolicy = DEFAULT_POLICY) -> None: + self._users: dict[str, DirectoryUser] = {} + self._passwords: dict[str, str] = {} + self._history: dict[str, list[str]] = {} + self.policy = policy + self.unavailable = False # set True to simulate an outage + + def add_user(self, sam: str, password: str = "Correct-Horse-9", **fields) -> DirectoryUser: # noqa: S107 + fields.setdefault("password_last_set", datetime.now(UTC)) + user = DirectoryUser(sam=sam, **fields) + self._users[sam.casefold()] = user + self._passwords[sam.casefold()] = password + self._history[sam.casefold()] = [password] + return user + + def _check_up(self) -> None: + if self.unavailable: + raise DirectoryUnavailable("The directory is not available") + + def _get(self, sam: str) -> DirectoryUser: + user = self._users.get(sam.casefold()) if sam else None + if user is None: + raise ValidationFailed("The user does not exist in the configured OU") + return user + + def find_user(self, sam: str) -> DirectoryUser | None: + self._check_up() + return self._users.get(sam.casefold()) if sam else None + + def verify_password(self, sam: str, password: str) -> bool: + self._check_up() + if not password or not sam: + return False + user = self._users.get(sam.casefold()) + return user is not None and self._passwords[sam.casefold()] == password + + def password_policy(self, sam: str) -> PasswordPolicy: + self._check_up() + self._get(sam) + return self.policy + + def update_attributes(self, sam: str, changes: dict[str, str | None]) -> None: + self._check_up() + refused = [name for name in changes if name not in WRITABLE_ATTRIBUTES] + if refused: + raise Forbidden(f"Attributes are not writable: {', '.join(sorted(refused))}") + user = self._get(sam) + values = {} + 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[_FIELD[name]] = value or "" + self._users[sam.casefold()] = replace(user, **values) + + def change_password(self, sam: str, current: str, new: str) -> None: + self._check_up() + user = self._get(sam) + key = sam.casefold() + if not current or self._passwords[key] != current: + raise PasswordRejected("wrong_current") + age = datetime.now(UTC) - user.password_last_set if user.password_last_set else None + too_young = age is not None and age < self.policy.min_age + weak = len(new) < self.policy.min_length or ( + self.policy.complexity and _character_classes(new) < 3 + ) + reused = new in self._history[key][-self.policy.history_length :] + if not new or too_young or weak or reused: + raise PasswordRejected("policy") + self._passwords[key] = new + self._history[key].append(new) + self._users[key] = replace(user, password_last_set=datetime.now(UTC)) diff --git a/portal/adapters/directory/ldap_directory.py b/portal/adapters/directory/ldap_directory.py new file mode 100644 index 0000000..8826cdc --- /dev/null +++ b/portal/adapters/directory/ldap_directory.py @@ -0,0 +1,287 @@ +"""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 +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: + try: + conn.unbind_s() + except self._ldap.LDAPError: + pass + + 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: + try: + user_conn.unbind_s() + except ldap.LDAPError: + pass + + 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")), + ) + 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, + ) + + 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" diff --git a/portal/config/env.py b/portal/config/env.py index d9de5d9..a4bdb83 100644 --- a/portal/config/env.py +++ b/portal/config/env.py @@ -4,7 +4,7 @@ from __future__ import annotations import ipaddress from collections.abc import Mapping -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from zoneinfo import ZoneInfo, ZoneInfoNotFoundError @@ -31,6 +31,17 @@ class ConfigError(Exception): return "\n".join(lines) +@dataclass(frozen=True) +class DirectoryConfig: + server: str # host name, optionally host:port; always reached over LDAPS + user_ou_dn: str + bind_dn: str # distinguished name or user principal name of the service account + bind_password: str = field(repr=False) + ca_file: Path + publisher_group_dn: str + admin_group_dn: str + + @dataclass(frozen=True) class Config: secret_key: str @@ -40,6 +51,7 @@ class Config: trusted_proxies: list[str] log_level: str default_timezone: str + directory: DirectoryConfig def load_config(environ: Mapping[str, str]) -> Config: @@ -104,6 +116,25 @@ def load_config(environ: Mapping[str, str]) -> Config: except (ZoneInfoNotFoundError, ValueError, OSError): problems.append("DEFAULT_TIMEZONE: must be a valid IANA time zone name") + def required(name: str, rule: str) -> str: + value = raw(name) + if value is None: + problems.append(f"{name}: required, {rule}") + return "" + return value + + server = required("LDAP_SERVER", "host name of the directory server, with optional :port") + if "://" in server or "/" in server: + problems.append("LDAP_SERVER: must be a host name, not a URL (LDAPS is always used)") + user_ou_dn = required("LDAP_USER_OU_DN", "distinguished name of the OU that holds the users") + bind_dn = required("LDAP_BIND_DN", "service account distinguished name or user principal name") + bind_password = required("LDAP_BIND_PASSWORD", "password of the service account") + publisher_group_dn = required("LDAP_PUBLISHER_GROUP_DN", "distinguished name of the group") + admin_group_dn = required("LDAP_ADMIN_GROUP_DN", "distinguished name of the group") + ca_file = Path(required("LDAP_CA_FILE", "path of the PEM file with the CA chain")) + if raw("LDAP_CA_FILE") is not None and not ca_file.is_file(): + problems.append("LDAP_CA_FILE: must be an existing file") + if problems: raise ConfigError(problems) @@ -115,4 +146,13 @@ def load_config(environ: Mapping[str, str]) -> Config: trusted_proxies=proxies, log_level=log_level, default_timezone=timezone, + directory=DirectoryConfig( + server=server, + user_ou_dn=user_ou_dn, + bind_dn=bind_dn, + bind_password=bind_password, + ca_file=ca_file, + publisher_group_dn=publisher_group_dn, + admin_group_dn=admin_group_dn, + ), ) diff --git a/portal/config/settings.py b/portal/config/settings.py index 8e552ca..a692535 100644 --- a/portal/config/settings.py +++ b/portal/config/settings.py @@ -23,6 +23,7 @@ ALLOWED_HOSTS = _config.allowed_hosts TRUSTED_PROXIES = _config.trusted_proxies DATA_DIR = _config.data_dir DISPLAY_TIME_ZONE = _config.default_timezone +DIRECTORY = _config.directory INSTALLED_APPS = [ "django.contrib.contenttypes", diff --git a/portal/tests/conftest.py b/portal/tests/conftest.py index c6d666c..d529507 100644 --- a/portal/tests/conftest.py +++ b/portal/tests/conftest.py @@ -3,8 +3,17 @@ import pytest @pytest.fixture def valid_env(tmp_path): + ca_file = tmp_path / "ca.pem" + ca_file.write_text("test ca") return { "SECRET_KEY": "x" * 48, "ALLOWED_HOSTS": "testserver", "DATA_DIR": str(tmp_path), + "LDAP_SERVER": "dc.example.test", + "LDAP_USER_OU_DN": "OU=Users,DC=example,DC=test", + "LDAP_BIND_DN": "svc@example.test", + "LDAP_BIND_PASSWORD": "service-password", + "LDAP_CA_FILE": str(ca_file), + "LDAP_PUBLISHER_GROUP_DN": "CN=Publishers,OU=Users,DC=example,DC=test", + "LDAP_ADMIN_GROUP_DN": "CN=Admins,OU=Users,DC=example,DC=test", } diff --git a/portal/tests/ldap_stub.py b/portal/tests/ldap_stub.py new file mode 100644 index 0000000..f029e54 --- /dev/null +++ b/portal/tests/ldap_stub.py @@ -0,0 +1,114 @@ +"""A stand-in for the `ldap` module that records calls. Real python-ldap needs a C library.""" + +from types import SimpleNamespace + + +class LDAPError(Exception): + pass + + +class SERVER_DOWN(LDAPError): # noqa: N801 + pass + + +class INVALID_CREDENTIALS(LDAPError): # noqa: N801 + pass + + +class INSUFFICIENT_ACCESS(LDAPError): # noqa: N801 + pass + + +class CONSTRAINT_VIOLATION(LDAPError): # noqa: N801 + pass + + +class UNWILLING_TO_PERFORM(LDAPError): # noqa: N801 + pass + + +class INVALID_SYNTAX(LDAPError): # noqa: N801 + pass + + +def escape_filter_chars(value): + table = {"\\": r"\5c", "*": r"\2a", "(": r"\28", ")": r"\29", "\x00": r"\00"} + return "".join(table.get(c, c) for c in value) + + +class Connection: + def __init__(self, world, url): + self.world = world + self.url = url + self.options = {} + self.bound = None + + def set_option(self, name, value): + self.options[name] = value + + def simple_bind_s(self, dn, password): + if self.world.down: + raise SERVER_DOWN() + if self.world.passwords.get(dn) != password: + raise INVALID_CREDENTIALS() + self.bound = dn + + def unbind_s(self): + pass + + def search_s(self, base, scope, flt, attrs): + self.world.searches.append((base, scope, flt, tuple(attrs))) + return self.world.respond(base, scope, flt) + + def modify_s(self, dn, mods): + self.world.modifications.append((dn, mods)) + if self.world.modify_error: + raise self.world.modify_error + + +class World: + """The server side: canned search answers and the connections the adapter opened.""" + + def __init__(self): + self.down = False + self.passwords = {"svc@example.test": "service-password"} + self.searches = [] + self.modifications = [] + self.modify_error = None + self.connections = [] + self.entries = {} # (base, scope) -> list of (dn, attrs) + + def respond(self, base, scope, flt): + return self.entries.get((base, scope), []) + + +def build(world): + def initialize(url): + conn = Connection(world, url) + world.connections.append(conn) + return conn + + return SimpleNamespace( + initialize=initialize, + LDAPError=LDAPError, + SERVER_DOWN=SERVER_DOWN, + INVALID_CREDENTIALS=INVALID_CREDENTIALS, + INSUFFICIENT_ACCESS=INSUFFICIENT_ACCESS, + CONSTRAINT_VIOLATION=CONSTRAINT_VIOLATION, + UNWILLING_TO_PERFORM=UNWILLING_TO_PERFORM, + INVALID_SYNTAX=INVALID_SYNTAX, + filter=SimpleNamespace(escape_filter_chars=escape_filter_chars), + SCOPE_BASE=0, + SCOPE_SUBTREE=2, + MOD_ADD=0, + MOD_DELETE=1, + MOD_REPLACE=2, + OPT_PROTOCOL_VERSION="protocol", + OPT_REFERRALS="referrals", + OPT_NETWORK_TIMEOUT="network_timeout", + OPT_TIMEOUT="timeout", + OPT_X_TLS_CACERTFILE="cacertfile", + OPT_X_TLS_REQUIRE_CERT="require_cert", + OPT_X_TLS_DEMAND="demand", + OPT_X_TLS_NEWCTX="newctx", + ) diff --git a/portal/tests/test_config.py b/portal/tests/test_config.py index e0b5f07..5dc85dd 100644 --- a/portal/tests/test_config.py +++ b/portal/tests/test_config.py @@ -70,3 +70,38 @@ def test_process_exits_nonzero_naming_both_variables(tmp_path): assert "SECRET_KEY" in result.stderr assert "LOG_LEVEL" in result.stderr assert "Traceback" not in result.stderr + + +@pytest.mark.parametrize( + "name", + [ + "LDAP_SERVER", + "LDAP_USER_OU_DN", + "LDAP_BIND_DN", + "LDAP_BIND_PASSWORD", + "LDAP_CA_FILE", + "LDAP_PUBLISHER_GROUP_DN", + "LDAP_ADMIN_GROUP_DN", + ], +) +def test_each_directory_variable_is_required(valid_env, name): + del valid_env[name] + with pytest.raises(ConfigError, match=name): + load_config(valid_env) + + +def test_directory_settings_are_loaded_and_the_password_is_not_printed(valid_env): + directory = load_config(valid_env).directory + assert directory.server == "dc.example.test" + assert directory.user_ou_dn == "OU=Users,DC=example,DC=test" + assert "service-password" not in repr(directory) + + +def test_ldap_server_must_not_be_a_url(valid_env): + with pytest.raises(ConfigError, match="LDAP_SERVER"): + load_config({**valid_env, "LDAP_SERVER": "ldap://dc.example.test"}) + + +def test_a_missing_ca_file_is_reported(valid_env, tmp_path): + with pytest.raises(ConfigError, match="LDAP_CA_FILE: must be an existing file"): + load_config({**valid_env, "LDAP_CA_FILE": str(tmp_path / "missing.pem")}) diff --git a/portal/tests/test_directory.py b/portal/tests/test_directory.py new file mode 100644 index 0000000..7da9d68 --- /dev/null +++ b/portal/tests/test_directory.py @@ -0,0 +1,110 @@ +"""Contract tests for the in-memory directory that the other tickets build against.""" + +from datetime import UTC, datetime, timedelta + +import pytest + +from portal.adapters.directory import ( + WRITABLE_ATTRIBUTES, + DirectoryUnavailable, + FakeDirectory, + Forbidden, + PasswordRejected, + ValidationFailed, +) + + +@pytest.fixture +def directory(): + d = FakeDirectory() + d.add_user("Alice", given_name="Alice", mail="alice@example.test", is_publisher=True) + d.add_user("gone", enabled=False) + d.add_user("locked", locked=True) + d.add_user("expired", password_last_set=None) + return d + + +def test_find_user_is_case_insensitive_and_keeps_the_exact_name(directory): + assert directory.find_user("alice").sam == "Alice" + assert directory.find_user("nobody") is None + assert directory.find_user("") is None + + +def test_account_state_is_reported(directory): + assert directory.find_user("gone").enabled is False + assert directory.find_user("locked").locked is True + assert directory.find_user("expired").must_change_password is True + assert directory.find_user("Alice").is_publisher is True + assert directory.find_user("Alice").is_admin is False + + +def test_verify_password(directory): + assert directory.verify_password("alice", "Correct-Horse-9") is True + assert directory.verify_password("alice", "wrong") is False + assert directory.verify_password("nobody", "Correct-Horse-9") is False + + +def test_an_empty_password_is_never_valid(directory): + directory.add_user("blank", password="") + assert directory.verify_password("blank", "") is False + + +def test_allowed_attributes_can_be_written_and_cleared(directory): + directory.update_attributes("alice", {"mobile": "+49 1", "givenName": "Al"}) + user = directory.find_user("alice") + assert (user.mobile, user.given_name) == ("+49 1", "Al") + directory.update_attributes("alice", {"mobile": None}) + assert directory.find_user("alice").mobile == "" + + +@pytest.mark.parametrize( + "name", ["description", "memberOf", "unicodePwd", "sAMAccountName", "userAccountControl"] +) +def test_other_attributes_are_refused_before_anything_is_written(directory, name): + with pytest.raises(Forbidden): + directory.update_attributes("alice", {"givenName": "Changed", name: "x"}) + assert directory.find_user("alice").given_name == "Alice" + + +def test_writing_an_unknown_user_fails(directory): + with pytest.raises(ValidationFailed): + directory.update_attributes("nobody", {"mobile": "1"}) + + +def test_writable_list_is_exactly_the_agreed_six(): + assert set(WRITABLE_ATTRIBUTES) == { + "givenName", + "sn", + "displayName", + "mail", + "telephoneNumber", + "mobile", + } + + +def test_password_policy_is_returned(directory): + policy = directory.password_policy("alice") + assert policy.min_length == 8 + assert policy.min_age == timedelta(0) + + +def test_password_change_needs_the_current_password(directory): + with pytest.raises(PasswordRejected) as exc: + directory.change_password("alice", "wrong", "Another-Pass-7!") + assert exc.value.reason == "wrong_current" + directory.change_password("alice", "Correct-Horse-9", "Another-Pass-7!") + assert directory.verify_password("alice", "Another-Pass-7!") is True + assert directory.find_user("alice").password_last_set > datetime.now(UTC) - timedelta(minutes=1) + + +@pytest.mark.parametrize("new", ["short1A", "alllowercase-long", "Correct-Horse-9"]) +def test_policy_violations_are_reported_as_policy(directory, new): + with pytest.raises(PasswordRejected) as exc: + directory.change_password("alice", "Correct-Horse-9", new) + assert exc.value.reason == "policy" + + +def test_an_outage_is_a_typed_error(directory): + directory.unavailable = True + with pytest.raises(DirectoryUnavailable): + directory.find_user("alice") diff --git a/portal/tests/test_directory_live.py b/portal/tests/test_directory_live.py new file mode 100644 index 0000000..833ddbd --- /dev/null +++ b/portal/tests/test_directory_live.py @@ -0,0 +1,78 @@ +"""Opt-in check against the real test OU. Skipped unless PORTAL_LIVE_LDAP=1. + +Set LIVE_LDAP_SERVER, LIVE_LDAP_USER_OU_DN, LIVE_LDAP_BIND_DN, LIVE_LDAP_BIND_PASSWORD, +LIVE_LDAP_CA_FILE, LIVE_LDAP_PUBLISHER_GROUP_DN, LIVE_LDAP_ADMIN_GROUP_DN and +LIVE_LDAP_TEST_PASSWORD (the current password of ptest-user1). Needs python-ldap (Linux). +Only ever point this at the test OU. It restores what it writes and never changes a password. +""" + +import os +from pathlib import Path + +import pytest + +from portal.adapters.directory import Forbidden, LdapDirectory +from portal.config.env import DirectoryConfig + +pytestmark = pytest.mark.skipif( + os.environ.get("PORTAL_LIVE_LDAP") != "1", reason="set PORTAL_LIVE_LDAP=1 to run" +) + + +@pytest.fixture(scope="module") +def live(): + pytest.importorskip("ldap") + env = os.environ + config = DirectoryConfig( + server=env["LIVE_LDAP_SERVER"], + user_ou_dn=env["LIVE_LDAP_USER_OU_DN"], + bind_dn=env["LIVE_LDAP_BIND_DN"], + bind_password=env["LIVE_LDAP_BIND_PASSWORD"], + ca_file=Path(env["LIVE_LDAP_CA_FILE"]), + publisher_group_dn=env["LIVE_LDAP_PUBLISHER_GROUP_DN"], + admin_group_dn=env["LIVE_LDAP_ADMIN_GROUP_DN"], + ) + return LdapDirectory(config) + + +def test_reads_state_and_roles(live): + user = live.find_user("ptest-user1") + assert user is not None and user.enabled and not user.is_publisher + assert live.find_user("ptest-publisher").is_publisher + assert live.find_user("ptest-admin").is_admin + assert live.find_user("ptest-disabled").enabled is False + assert live.find_user("ptest-expired").must_change_password is True + assert live.find_user("does-not-exist") is None + + +def test_users_outside_the_ou_are_not_found(live): + assert live.find_user("Administrator") is None + + +def test_allowed_attributes_are_written_and_restored(live): + before = live.find_user("ptest-user1") + try: + live.update_attributes("ptest-user1", {"mobile": "+49 000 0000", "displayName": "Test X"}) + after = live.find_user("ptest-user1") + assert (after.mobile, after.display_name) == ("+49 000 0000", "Test X") + finally: + live.update_attributes( + "ptest-user1", {"mobile": before.mobile, "displayName": before.display_name} + ) + assert live.find_user("ptest-user1").mobile == before.mobile + + +def test_other_attributes_are_refused(live): + with pytest.raises(Forbidden): + live.update_attributes("ptest-user1", {"description": "x"}) + + +def test_password_policy_is_returned(live): + policy = live.password_policy("ptest-user1") + assert policy.min_length >= 0 + + +def test_verify_password(live): + assert live.verify_password("ptest-user1", os.environ["LIVE_LDAP_TEST_PASSWORD"]) is True + assert live.verify_password("ptest-user1", "definitely-not-the-password") is False + assert live.verify_password("ptest-user1", "") is False diff --git a/portal/tests/test_ldap_directory.py b/portal/tests/test_ldap_directory.py new file mode 100644 index 0000000..e185bd4 --- /dev/null +++ b/portal/tests/test_ldap_directory.py @@ -0,0 +1,298 @@ +"""The LDAPS adapter against a stand-in `ldap` module: what it sends and how it maps errors.""" + +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace + +import pytest + +from portal.adapters.directory import ( + DirectoryUnavailable, + Forbidden, + LdapDirectory, + PasswordRejected, + ValidationFailed, +) +from portal.tests import ldap_stub + +OU = "OU=Users,DC=example,DC=test" +DOMAIN = "DC=example,DC=test" +USER_DN = "CN=Alice A,OU=Users,DC=example,DC=test" +SVC = "svc@example.test" +EPOCH = datetime(1601, 1, 1, tzinfo=UTC) + + +def filetime(moment): + return (moment - EPOCH) // timedelta(microseconds=1) * 10 + + +def b(**attrs): + return {k: [v.encode()] for k, v in attrs.items()} + + +def config(tmp_path, server="dc.example.test"): + ca = tmp_path / "ca.pem" + ca.write_text("ca") + return SimpleNamespace( + server=server, + user_ou_dn=OU, + bind_dn=SVC, + bind_password="service-password", + ca_file=ca, + publisher_group_dn="CN=Publishers,OU=Users,DC=example,DC=test", + admin_group_dn="CN=Admins,OU=Users,DC=example,DC=test", + ) + + +@pytest.fixture +def world(): + w = ldap_stub.World() + w.passwords[USER_DN] = "Alice-Pass-1!" + w.entries[(OU, 2)] = [ + ( + USER_DN, + { + **b( + sAMAccountName="Alice", + givenName="Alice", + sn="A", + mail="a@example.test", + userAccountControl="512", + pwdLastSet=str(filetime(datetime(2026, 1, 1, tzinfo=UTC))), + ), + "memberOf": [ + b"cn=publishers, OU=Users,DC=example,DC=test", + b"CN=Other,DC=example,DC=test", + ], + }, + ) + ] + w.entries[("", 0)] = [("", b(defaultNamingContext=DOMAIN))] + w.entries[(DOMAIN, 0)] = [ + ( + DOMAIN, + b( + minPwdLength="12", + pwdProperties="1", + pwdHistoryLength="24", + minPwdAge=str(-864000000000), + maxPwdAge=str(-77760000000000), + lockoutDuration=str(-18000000000), + ), + ) + ] + return w + + +@pytest.fixture +def directory(world, tmp_path): + return LdapDirectory(config(tmp_path), ldap_module=ldap_stub.build(world)) + + +def user_attrs(world): + return world.entries[(OU, 2)][0][1] + + +def test_connection_is_ldaps_with_certificate_verification(directory, world, tmp_path): + directory.find_user("alice") + conn = world.connections[0] + assert conn.url == "ldaps://dc.example.test:636" + assert conn.options["require_cert"] == "demand" + assert conn.options["cacertfile"] == str(tmp_path / "ca.pem") + assert conn.options["newctx"] == 0 + + +def test_an_explicit_port_is_kept(world, tmp_path): + d = LdapDirectory(config(tmp_path, "dc.example.test:3269"), ldap_module=ldap_stub.build(world)) + d.find_user("alice") + assert world.connections[0].url == "ldaps://dc.example.test:3269" + + +def test_find_user_reads_state_and_roles(directory): + user = directory.find_user("alice") + assert (user.sam, user.given_name, user.surname, user.mail) == ( + "Alice", + "Alice", + "A", + "a@example.test", + ) + assert user.enabled and not user.locked + assert user.is_publisher and not user.is_admin + assert user.password_last_set == datetime(2026, 1, 1, tzinfo=UTC) + + +def test_the_lookup_is_confined_to_the_ou_and_asks_only_for_listed_attributes(directory, world): + directory.find_user("alice") + base, scope, _flt, attrs = world.searches[0] + assert (base, scope) == (OU, 2) + assert "unicodePwd" not in attrs and "description" not in attrs + + +@pytest.mark.parametrize( + ("evil", "escaped"), + [ + ("*", r"\2a"), + ("a)(objectClass=*", r"a\29\28objectClass=\2a"), + ("a*)(|(sAMAccountName=*", r"a\2a\29\28|\28sAMAccountName=\2a"), + ("back\\slash", r"back\5cslash"), + ], +) +def test_filter_input_is_escaped(directory, world, evil, escaped): + directory.find_user(evil) + flt = world.searches[0][2] + assert flt == f"(&(objectClass=user)(objectCategory=person)(sAMAccountName={escaped}))" + + +def test_a_disabled_account_and_a_lock_are_reported(directory, world): + user_attrs(world)["userAccountControl"] = [b"514"] + user_attrs(world)["msDS-User-Account-Control-Computed"] = [b"16"] + user = directory.find_user("alice") + assert not user.enabled and user.locked + + +def test_lockout_falls_back_to_the_time_when_the_computed_attribute_is_missing(directory, world): + user_attrs(world)["lockoutTime"] = [str(filetime(datetime.now(UTC))).encode()] + assert directory.find_user("alice").locked is True # the lockout duration is 30 minutes + user_attrs(world)["lockoutTime"] = [str(filetime(datetime(2026, 1, 1, tzinfo=UTC))).encode()] + assert directory.find_user("alice").locked is False + + +def test_pwd_last_set_zero_means_the_user_must_change_it(directory, world): + user_attrs(world)["pwdLastSet"] = [b"0"] + assert directory.find_user("alice").must_change_password is True + + +def test_an_unknown_or_ambiguous_user_is_not_found(directory, world): + world.entries[(OU, 2)] = world.entries[(OU, 2)] * 2 + assert directory.find_user("alice") is None + world.entries[(OU, 2)] = [] + assert directory.find_user("alice") is None + + +def test_a_refused_service_account_is_an_outage(world, tmp_path): + world.passwords[SVC] = "changed" + d = LdapDirectory(config(tmp_path), ldap_module=ldap_stub.build(world)) + with pytest.raises(DirectoryUnavailable): + d.find_user("alice") + + +def test_a_server_that_is_down_is_an_outage(directory, world): + world.down = True + with pytest.raises(DirectoryUnavailable): + directory.find_user("alice") + + +def test_verify_password_binds_as_the_found_user(directory, world): + assert directory.verify_password("alice", "Alice-Pass-1!") is True + assert world.connections[-1].bound == USER_DN + assert directory.verify_password("alice", "nope") is False + assert directory.verify_password("nobody", "x") is False + + +def test_an_empty_password_is_refused_without_contacting_the_server(directory, world): + assert directory.verify_password("alice", "") is False + assert world.connections == [] + + +def test_password_policy_converts_the_directory_units(directory): + policy = directory.password_policy("alice") + assert policy.min_length == 12 + assert policy.complexity is True + assert policy.history_length == 24 + assert policy.min_age == timedelta(days=1) + assert policy.max_age == timedelta(days=90) + assert policy.fine_grained is False + + +def test_a_fine_grained_policy_is_used_when_readable(directory, world): + pso = "CN=PSO,CN=Password Settings Container,CN=System,DC=example,DC=test" + user_attrs(world)["msDS-ResultantPSO"] = [pso.encode()] + world.entries[(pso, 0)] = [ + ( + pso, + b( + **{ + "msDS-MinimumPasswordLength": "20", + "msDS-PasswordComplexityEnabled": "FALSE", + "msDS-PasswordHistoryLength": "5", + "msDS-MinimumPasswordAge": "0", + "msDS-MaximumPasswordAge": str(-(2**63)), + } + ), + ) + ] + policy = directory.password_policy("alice") + assert (policy.min_length, policy.complexity, policy.history_length) == (20, False, 5) + assert policy.max_age is None and policy.fine_grained is True + + +def test_an_unreadable_fine_grained_policy_is_flagged_not_hidden(directory, world, monkeypatch): + pso = "CN=PSO,CN=System,DC=example,DC=test" + user_attrs(world)["msDS-ResultantPSO"] = [pso.encode()] + original = world.respond + + def respond(base, scope, flt): + if base == pso: + raise ldap_stub.INSUFFICIENT_ACCESS() + return original(base, scope, flt) + + monkeypatch.setattr(world, "respond", respond) + policy = directory.password_policy("alice") + assert policy.fine_grained is True and policy.fine_grained_readable is False + assert policy.min_length == 12 + + +def test_update_sends_a_replace_for_allowed_attributes(directory, world): + directory.update_attributes("alice", {"mobile": "+49 1", "sn": None}) + dn, mods = world.modifications[0] + assert dn == USER_DN + assert mods == [(2, "mobile", [b"+49 1"]), (2, "sn", [])] + + +def test_update_refuses_other_attributes_without_connecting(directory, world): + with pytest.raises(Forbidden): + directory.update_attributes("alice", {"description": "x"}) + assert world.connections == [] and world.modifications == [] + + +def test_update_maps_a_server_refusal_and_missing_rights(directory, world): + world.modify_error = ldap_stub.CONSTRAINT_VIOLATION() + with pytest.raises(ValidationFailed): + directory.update_attributes("alice", {"mail": "x"}) + world.modify_error = ldap_stub.INSUFFICIENT_ACCESS() + with pytest.raises(Forbidden): + directory.update_attributes("alice", {"mail": "x"}) + + +def test_change_password_sends_delete_old_and_add_new_as_utf16(directory, world): + directory.change_password("alice", "Old-Pass-1!", "New-Pass-2@") + dn, mods = world.modifications[0] + assert dn == USER_DN + assert mods == [ + (1, "unicodePwd", ['"Old-Pass-1!"'.encode("utf-16-le")]), + (0, "unicodePwd", ['"New-Pass-2@"'.encode("utf-16-le")]), + ] + + +@pytest.mark.parametrize( + ("info", "reason"), + [ + ("0000052D: AtrErr: DSID-03191077, problem 1005 (CONSTRAINT_ATT_TYPE)", "policy"), + ("00000056: AtrErr: DSID-03190F80, problem 1005 (CONSTRAINT_ATT_TYPE)", "wrong_current"), + ("something unexpected", "policy"), + ], +) +def test_password_refusals_are_told_apart_by_the_directory_error_code( + directory, world, info, reason +): + world.modify_error = ldap_stub.CONSTRAINT_VIOLATION({"info": info}) + with pytest.raises(PasswordRejected) as exc: + directory.change_password("alice", "Old-Pass-1!", "New-Pass-2@") + assert exc.value.reason == reason + + +def test_a_refused_password_change_does_not_carry_the_passwords(directory, world): + world.modify_error = ldap_stub.CONSTRAINT_VIOLATION({"info": "0000052D: x"}) + with pytest.raises(PasswordRejected) as exc: + directory.change_password("alice", "Old-Pass-1!", "New-Pass-2@") + text = repr(exc.value) + str(exc.value) + assert "Old-Pass-1!" not in text and "New-Pass-2@" not in text diff --git a/pyproject.toml b/pyproject.toml index 3daffb7..932b699 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,6 +9,7 @@ dependencies = [ "gunicorn>=23.0", "whitenoise>=6.9", "tzdata", + "python-ldap>=3.4.4", ] [dependency-groups] @@ -29,6 +30,13 @@ env = [ "DATA_DIR=.", "DEBUG=false", "LOG_LEVEL=INFO", + "LDAP_SERVER=dc.example.test", + "LDAP_USER_OU_DN=OU=Users,DC=example,DC=test", + "LDAP_BIND_DN=svc@example.test", + "LDAP_BIND_PASSWORD=test-only-service-password", + "LDAP_CA_FILE=pyproject.toml", + "LDAP_PUBLISHER_GROUP_DN=CN=Publishers,OU=Users,DC=example,DC=test", + "LDAP_ADMIN_GROUP_DN=CN=Admins,OU=Users,DC=example,DC=test", ] [tool.ruff] diff --git a/uv.lock b/uv.lock index 57bcb16..015f1bc 100644 --- a/uv.lock +++ b/uv.lock @@ -70,6 +70,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/40/9e/2b38731e0fc536806f16490e1a12d7f0dc2a1235aa8cc07bcc75416a7daa/pluggy-1.7.0-py3-none-any.whl", hash = "sha256:7dd7b0d8832ba3cb632c306926ded123429211b83641b35dc5c41ad2d34f9bec", size = 27082, upload-time = "2026-10-15T09:50:56.808Z" }, ] +[[package]] +name = "pyasn1" +version = "0.6.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a4/9a/23310166d960def5897e91fe20e5b724601b02a22e84ba1f94232c0b7f67/pyasn1-0.6.4.tar.gz", hash = "sha256:9c447d8431c947fe4c8febc4ed9e760bc29011a5b01e5c74b67025bd9fb8ce81", size = 151262, upload-time = "2026-07-09T01:12:33.988Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9a/3b/6163796d69c3977d1e4287bea4a6979161cbbdd170ebb430511e8e1999ce/pyasn1-0.6.4-py3-none-any.whl", hash = "sha256:deda9277cfd454080ec40b207fb6df82206a3a2688735233cdcd8d3d565f088b", size = 84410, upload-time = "2026-07-09T01:12:32.92Z" }, +] + +[[package]] +name = "pyasn1-modules" +version = "0.4.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pyasn1" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e9/e6/78ebbb10a8c8e4b61a59249394a4a594c1a7af95593dc933a349c8d00964/pyasn1_modules-0.4.2.tar.gz", hash = "sha256:677091de870a80aae844b1ca6134f54652fa2c8c5a52aa396440ac3106e941e6", size = 307892, upload-time = "2025-03-28T02:41:22.17Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/47/8d/d529b5d697919ba8c11ad626e835d4039be708a35b0d22de83a269a6682c/pyasn1_modules-0.4.2-py3-none-any.whl", hash = "sha256:29253a9207ce32b64c3ac6600edc75368f98473906e8fd1043bd6b5b1de2c14a", size = 181259, upload-time = "2025-03-28T02:41:19.028Z" }, +] + [[package]] name = "pygments" version = "2.21.0" @@ -129,6 +150,16 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/60/d1/38f3a3405989a89ac18390803e70c6ad7c7760da4f9b83cbeca0c44a0c72/python_dotenv-1.2.4-py3-none-any.whl", hash = "sha256:42269a8a5b3fd54ffa6f3d84b18abed50064717576b4ecf03dc4a55d8aa04fdc", size = 23266, upload-time = "2026-10-01T05:36:08.633Z" }, ] +[[package]] +name = "python-ldap" +version = "3.4.8" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pyasn1" }, + { name = "pyasn1-modules" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/1c/45/489fcf46984f916ef165cdc6a2a1f1ca01ab6f486d47169256ebe44bdd62/python_ldap-3.4.8.tar.gz", hash = "sha256:18dc7460470c6ff64ed5c04ee21c56dbfee7ab433a53213ba91e407eea44c34c", size = 387710, upload-time = "2026-09-16T12:41:20.15Z" } + [[package]] name = "ruff" version = "0.17.0" @@ -161,6 +192,7 @@ source = { virtual = "." } dependencies = [ { name = "django" }, { name = "gunicorn" }, + { name = "python-ldap" }, { name = "tzdata" }, { name = "whitenoise" }, ] @@ -177,6 +209,7 @@ dev = [ requires-dist = [ { name = "django", specifier = ">=5.2,<5.3" }, { name = "gunicorn", specifier = ">=23.0" }, + { name = "python-ldap", specifier = ">=3.4.4" }, { name = "tzdata" }, { name = "whitenoise", specifier = ">=6.9" }, ]