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
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
"""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),
|
|
lockout_threshold=5,
|
|
lockout_window=timedelta(minutes=30),
|
|
)
|
|
|
|
|
|
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))
|