Ebbe BaßandClaude Sonnet 5.5
3fd8f27127
Profile page at /profile with first name, last name, display name, telephone and mobile read from AD and shown with the username and e-mail. Only changed fields are written through the directory adapter in one transaction with the audit entry (field names only); numbers are accepted in any common notation and stored as +country code and digits; names are required. Plain-language errors per field, German text and tests; review findings logged in the plan. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FocPJ75XPma7TcuYEMy8zm
458 lines
16 KiB
Python
458 lines
16 KiB
Python
"""Password change and profile details of the signed-in user (FR-5..FR-9; AD-3, AD-4, AD-8).
|
|
|
|
The order is fixed: throttle check, policy read, verification of the current password, local
|
|
hints, then the directory change, the ended sessions, the audit entry and the queued mail in
|
|
one transaction. Passwords live only in local variables here: they are never logged, audited,
|
|
stored or put into an exception.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import math
|
|
from dataclasses import dataclass, replace
|
|
from datetime import timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from django.conf import settings
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
from django.utils.translation import gettext as _
|
|
from django.utils.translation import ngettext
|
|
|
|
from portal.accounts import throttle
|
|
from portal.accounts.sessions import SessionUser, end_sessions
|
|
from portal.adapters.directory import (
|
|
DirectoryUnavailable,
|
|
DirectoryUser,
|
|
Forbidden,
|
|
PasswordPolicy,
|
|
PasswordRejected,
|
|
ValidationFailed,
|
|
)
|
|
from portal.adapters.registry import get_directory
|
|
from portal.audit import services as audit
|
|
from portal.mailer import services as mailer
|
|
|
|
from .forms import DETAIL_FIELDS, check_details
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ENDPOINT = "password-change"
|
|
CHARACTER_GROUPS_REQUIRED = 3 # AD complexity: three of upper, lower, digit, other
|
|
|
|
|
|
class WrongCurrentPassword(Exception):
|
|
def __init__(self, policy: PasswordPolicy | None = None) -> None:
|
|
self.policy = policy
|
|
super().__init__("The current password is not correct")
|
|
|
|
|
|
class PasswordNotAccepted(Exception):
|
|
"""The new password breaks a rule. `reasons` holds keys for rules the portal could name."""
|
|
|
|
def __init__(self, reasons: list[str], policy: PasswordPolicy | None = None) -> None:
|
|
self.reasons = reasons
|
|
self.policy = policy
|
|
super().__init__("The new password is not accepted")
|
|
|
|
|
|
class OutcomeUnknown(Exception):
|
|
"""The directory accepted the new password but the portal could not finish its own part."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Changed:
|
|
mail_queued: bool
|
|
other_sessions_ended: int
|
|
|
|
|
|
# -- the rules, in words ----------------------------------------------------------------------
|
|
|
|
|
|
def describe_policy(policy: PasswordPolicy) -> list[str]:
|
|
"""The user's password rules as sentences in the active language."""
|
|
rules = []
|
|
if policy.min_length:
|
|
rules.append(
|
|
ngettext(
|
|
"At least %(n)s character",
|
|
"At least %(n)s characters",
|
|
policy.min_length,
|
|
)
|
|
% {"n": policy.min_length}
|
|
)
|
|
if policy.complexity:
|
|
rules.append(
|
|
_(
|
|
"Characters from at least three of these groups: upper case letters, "
|
|
"lower case letters, digits and special characters"
|
|
)
|
|
)
|
|
rules.append(_("No part of your user name or display name"))
|
|
if policy.history_length:
|
|
rules.append(
|
|
ngettext(
|
|
"Not your last %(n)s password",
|
|
"Not one of your last %(n)s passwords",
|
|
policy.history_length,
|
|
)
|
|
% {"n": policy.history_length}
|
|
)
|
|
if policy.min_age:
|
|
rules.append(_min_age_rule(policy.min_age))
|
|
if policy.max_age:
|
|
days = max(1, round(policy.max_age.total_seconds() / 86400))
|
|
rules.append(
|
|
ngettext(
|
|
"A password is valid for %(n)s day",
|
|
"A password is valid for %(n)s days",
|
|
days,
|
|
)
|
|
% {"n": days}
|
|
)
|
|
return rules
|
|
|
|
|
|
def _min_age_rule(age: timedelta) -> str:
|
|
if age >= timedelta(days=1):
|
|
days = math.ceil(age.total_seconds() / 86400) # round up: too early is refused
|
|
return ngettext(
|
|
"A new password can be changed again after %(n)s day",
|
|
"A new password can be changed again after %(n)s days",
|
|
days,
|
|
) % {"n": days}
|
|
hours = max(1, math.ceil(age.total_seconds() / 3600))
|
|
return ngettext(
|
|
"A new password can be changed again after %(n)s hour",
|
|
"A new password can be changed again after %(n)s hours",
|
|
hours,
|
|
) % {"n": hours}
|
|
|
|
|
|
def _character_groups(password: str) -> int:
|
|
groups = set()
|
|
for char in password:
|
|
if char.isupper():
|
|
groups.add("upper")
|
|
elif char.islower():
|
|
groups.add("lower")
|
|
elif char.isdigit():
|
|
groups.add("digit")
|
|
else:
|
|
groups.add("other")
|
|
return len(groups)
|
|
|
|
|
|
def local_hints(new: str, policy: PasswordPolicy) -> list[str]:
|
|
"""Rule keys the portal can check itself; the directory stays the authority.
|
|
|
|
Nothing is claimed when a fine-grained policy applies but could not be read: the domain
|
|
values may be stricter or looser than the rules that really apply.
|
|
"""
|
|
if policy.fine_grained and not policy.fine_grained_readable:
|
|
return []
|
|
reasons = []
|
|
if len(new) < policy.min_length:
|
|
reasons.append("too_short")
|
|
if policy.complexity and _character_groups(new) < CHARACTER_GROUPS_REQUIRED:
|
|
reasons.append("too_few_groups")
|
|
return reasons
|
|
|
|
|
|
# -- the limit -----------------------------------------
|
|
|
|
|
|
def failure_limit(policy: PasswordPolicy) -> int:
|
|
"""Wrong current passwords allowed: below the directory's lockout threshold (FR-6)."""
|
|
threshold = policy.lockout_threshold
|
|
if threshold <= 0:
|
|
return throttle.THROTTLE_MAX_FAILURES
|
|
return max(1, min(throttle.THROTTLE_MAX_FAILURES, threshold - 1))
|
|
|
|
|
|
def pause_seconds(policy: PasswordPolicy) -> int:
|
|
window = int(policy.lockout_window.total_seconds()) if policy.lockout_window else 0
|
|
return max(throttle.THROTTLE_PAUSE_SECONDS, window)
|
|
|
|
|
|
# -- the change -----------------------------------------
|
|
|
|
|
|
def _threshold_note(policy: PasswordPolicy | None) -> dict:
|
|
"""Audit note for a lockout threshold of 1, which the portal cannot protect against."""
|
|
if policy is not None and policy.lockout_threshold == 1:
|
|
return {"lockout_protection": "none"}
|
|
return {}
|
|
|
|
|
|
def _record_failure(user: SessionUser, ip: str, reason: str, **detail) -> None:
|
|
audit.record_failure(
|
|
user.key, "password.change_failed", ip=ip, detail={"reason": reason, **detail}
|
|
)
|
|
|
|
|
|
def change_password(
|
|
user: SessionUser, current: str, new: str, *, ip: str, session_key: str | None
|
|
) -> Changed:
|
|
"""Change the signed-in user's password.
|
|
|
|
Raises Throttled, WrongCurrentPassword, PasswordNotAccepted, DirectoryUnavailable or
|
|
OutcomeUnknown. Must be called outside an open transaction.
|
|
"""
|
|
try:
|
|
throttle.check(ENDPOINT, user=user.key, ip=ip or None)
|
|
except throttle.Throttled:
|
|
audit.record_failure(user.key, "password.throttled", ip=ip)
|
|
raise
|
|
|
|
directory = get_directory()
|
|
try:
|
|
policy = directory.password_policy(user.sam)
|
|
if not directory.verify_password(user.sam, current):
|
|
throttle.hit(
|
|
ENDPOINT,
|
|
user=user.key,
|
|
ip=ip or None,
|
|
max_failures=failure_limit(policy),
|
|
pause_seconds=pause_seconds(policy),
|
|
)
|
|
_record_failure(user, ip, "wrong_current", **_threshold_note(policy))
|
|
raise WrongCurrentPassword(policy)
|
|
account = directory.find_user(user.sam)
|
|
except DirectoryUnavailable:
|
|
_record_failure(user, ip, "unavailable")
|
|
raise
|
|
|
|
reasons = local_hints(new, policy)
|
|
if reasons:
|
|
_record_failure(user, ip, "policy", rules=reasons)
|
|
raise PasswordNotAccepted(reasons, policy)
|
|
|
|
changed_in_directory = False
|
|
step = "directory"
|
|
try:
|
|
with transaction.atomic():
|
|
directory.change_password(user.sam, current, new)
|
|
changed_in_directory = True
|
|
step = "end_sessions"
|
|
ended = end_sessions(user, except_session_key=session_key)
|
|
step = "queue_mail"
|
|
mail_queued = _queue_mail(account)
|
|
step = "audit"
|
|
audit.record(
|
|
user.key,
|
|
"password.changed",
|
|
ip=ip,
|
|
detail={
|
|
"other_sessions_ended": ended,
|
|
"mail": "queued" if mail_queued else "skipped",
|
|
**_threshold_note(policy),
|
|
},
|
|
)
|
|
except PasswordRejected as exc:
|
|
if exc.reason == "wrong_current": # the password changed between the check and the change
|
|
throttle.hit(
|
|
ENDPOINT,
|
|
user=user.key,
|
|
ip=ip or None,
|
|
max_failures=failure_limit(policy),
|
|
pause_seconds=pause_seconds(policy),
|
|
)
|
|
_record_failure(user, ip, "wrong_current", **_threshold_note(policy))
|
|
raise WrongCurrentPassword(policy) from exc
|
|
_record_failure(user, ip, "policy")
|
|
raise PasswordNotAccepted([], policy) from exc
|
|
except DirectoryUnavailable:
|
|
_record_failure(user, ip, "unavailable")
|
|
raise
|
|
except Exception as exc:
|
|
if changed_in_directory:
|
|
# AD-4: the directory changed but the portal's part did not commit.
|
|
_after_unclear_outcome(user, ip, session_key, step, exc)
|
|
raise OutcomeUnknown("The password was changed but not recorded") from exc
|
|
raise
|
|
|
|
throttle.reset(ENDPOINT, user=user.key)
|
|
return Changed(mail_queued=mail_queued, other_sessions_ended=ended)
|
|
|
|
|
|
def _after_unclear_outcome(
|
|
user: SessionUser, ip: str, session_key: str | None, step: str, exc: Exception
|
|
) -> None:
|
|
"""The directory has the new password but the portal's part rolled back.
|
|
|
|
Log it with the change details (AD-4, never the exception text, which could hold input),
|
|
then do what can still be done: end the other sessions and write a failure entry. The
|
|
database may be the very thing that failed, so each attempt stands alone.
|
|
"""
|
|
logger.error(
|
|
"Password changed in the directory but the portal did not record it: "
|
|
"user=%s ip=%s time=%s step=%s error=%s",
|
|
user.key,
|
|
ip,
|
|
timezone.now().isoformat(),
|
|
step,
|
|
type(exc).__name__,
|
|
)
|
|
try:
|
|
end_sessions(user, except_session_key=session_key)
|
|
except Exception:
|
|
logger.error("The other sessions of %s could not be ended", user.key)
|
|
try:
|
|
_record_failure(user, ip, "unclear", step=step)
|
|
except Exception:
|
|
logger.error("The failure entry for %s could not be written", user.key)
|
|
|
|
|
|
def _queue_mail(account) -> bool:
|
|
"""Queue the confirmation e-mail inside the caller's transaction; False if there is none."""
|
|
address = (account.mail if account else "") or ""
|
|
if not mailer.is_valid_address(address):
|
|
return False
|
|
name = account.display_name or " ".join(p for p in (account.given_name, account.surname) if p)
|
|
now = timezone.now().astimezone(ZoneInfo(settings.DISPLAY_TIME_ZONE))
|
|
mailer.enqueue(
|
|
"password_changed",
|
|
address,
|
|
{"name": name or account.sam, "changed_at": now, "zone_name": settings.DISPLAY_TIME_ZONE},
|
|
)
|
|
return True
|
|
|
|
|
|
# -- profile details (FR-9) -------------------------------------------------------------------
|
|
|
|
# Form field -> (directory attribute, DirectoryUser field). Only these five are ever written.
|
|
DETAIL_ATTRIBUTES = {
|
|
"first_name": ("givenName", "given_name"),
|
|
"last_name": ("sn", "surname"),
|
|
"display_name": ("displayName", "display_name"),
|
|
"telephone": ("telephoneNumber", "telephone"),
|
|
"mobile": ("mobile", "mobile"),
|
|
}
|
|
|
|
|
|
class DetailsInvalid(Exception):
|
|
"""Values were refused before anything was written; `errors` maps a field to its message."""
|
|
|
|
def __init__(self, errors: dict[str, str], details: Details) -> None:
|
|
self.errors = errors
|
|
self.details = details # what the one directory read held, for the page
|
|
super().__init__("The details are not valid")
|
|
|
|
|
|
class DetailsNotAccepted(Exception):
|
|
"""The directory refused a value."""
|
|
|
|
def __init__(self, details: Details) -> None:
|
|
self.details = details
|
|
super().__init__("The directory refused a value")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Details:
|
|
username: str
|
|
mail: str
|
|
values: dict[str, str]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DetailsUpdated:
|
|
changed: tuple[str, ...] # form field names, in page order
|
|
account: DirectoryUser # the user as the directory now holds them
|
|
|
|
|
|
def _values_of(account: DirectoryUser) -> dict[str, str]:
|
|
return {name: getattr(account, attr) or "" for name, (_attr, attr) in DETAIL_ATTRIBUTES.items()}
|
|
|
|
|
|
def _read(user: SessionUser) -> DirectoryUser:
|
|
account = get_directory().find_user(user.sam)
|
|
if account is None:
|
|
raise DirectoryUnavailable("The account could not be read")
|
|
return account
|
|
|
|
|
|
def load_details(user: SessionUser) -> Details:
|
|
"""What the directory holds about the signed-in user. Raises DirectoryUnavailable."""
|
|
account = _read(user)
|
|
return Details(username=account.sam, mail=account.mail or "", values=_values_of(account))
|
|
|
|
|
|
def _record_details_failure(user: SessionUser, ip: str, reason: str, **detail) -> None:
|
|
audit.record_failure(
|
|
user.key, "profile.update_failed", ip=ip, detail={"reason": reason, **detail}
|
|
)
|
|
|
|
|
|
def update_details(user: SessionUser, values: dict[str, str], *, ip: str) -> DetailsUpdated:
|
|
"""Save the changed details of the signed-in user.
|
|
|
|
Reads the directory afresh, checks the values that differ and writes only those, with the
|
|
audit entry in the same transaction (AD-4). Raises DetailsInvalid, DetailsNotAccepted,
|
|
DirectoryUnavailable or OutcomeUnknown. Must be called outside an open transaction.
|
|
"""
|
|
directory = get_directory()
|
|
try:
|
|
account = _read(user)
|
|
except DirectoryUnavailable:
|
|
_record_details_failure(user, ip, "unavailable")
|
|
raise
|
|
|
|
current = _values_of(account)
|
|
read = Details(username=account.sam, mail=account.mail or "", values=current)
|
|
cleaned, errors = check_details(values, current)
|
|
if errors:
|
|
raise DetailsInvalid(errors, read)
|
|
changed = tuple(name for name in DETAIL_FIELDS if cleaned[name] != current[name])
|
|
if not changed:
|
|
return DetailsUpdated((), account)
|
|
|
|
changes = {DETAIL_ATTRIBUTES[name][0]: cleaned[name] for name in changed}
|
|
updated = replace(account, **{DETAIL_ATTRIBUTES[name][1]: cleaned[name] for name in changed})
|
|
changed_in_directory = False
|
|
step = "directory"
|
|
try:
|
|
with transaction.atomic():
|
|
directory.update_attributes(user.sam, changes)
|
|
changed_in_directory = True
|
|
step = "audit"
|
|
audit.record(user.key, "profile.updated", ip=ip, detail={"fields": list(changed)})
|
|
step = "commit"
|
|
except ValidationFailed as exc:
|
|
_record_details_failure(user, ip, "refused")
|
|
raise DetailsNotAccepted(read) from exc
|
|
except DirectoryUnavailable:
|
|
_record_details_failure(user, ip, "unavailable")
|
|
raise
|
|
except Forbidden as exc: # the service account lacks the right: the user sees "not possible"
|
|
logger.error(
|
|
"The directory refused a profile write: user=%s ip=%s fields=%s",
|
|
user.key,
|
|
ip,
|
|
",".join(changed),
|
|
)
|
|
_record_details_failure(user, ip, "forbidden")
|
|
raise DirectoryUnavailable("The directory refused the write") from exc
|
|
except Exception as exc:
|
|
if changed_in_directory:
|
|
# AD-4: the directory changed but the portal's part did not commit.
|
|
logger.error(
|
|
"Profile details changed in the directory but the portal did not record them: "
|
|
"user=%s ip=%s time=%s fields=%s step=%s error=%s",
|
|
user.key,
|
|
ip,
|
|
timezone.now().isoformat(),
|
|
",".join(changed),
|
|
step,
|
|
type(exc).__name__,
|
|
)
|
|
try:
|
|
_record_details_failure(user, ip, "unclear", step=step, fields=list(changed))
|
|
except Exception:
|
|
logger.error("The failure entry for %s could not be written", user.key)
|
|
raise OutcomeUnknown("The details were changed but not recorded") from exc
|
|
_record_details_failure(user, ip, "error")
|
|
raise
|
|
return DetailsUpdated(changed, updated)
|