fix: undo no longer fails with a 400 in a signed-in browser

Undo ends every session of the user, including the one that opened the link; saving that deleted session at the end of the request made Django answer 400 although the address had been restored. The view now drops its own session (as sign-out does), also on the unclear-outcome page. Tests for both cases from a signed-in browser. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FocPJ75XPma7TcuYEMy8zm
feat: add e-mail change with confirmation and undo
2026-10-19 10:28:35 +02:00 · 2026-10-19 10:05:41 +02:00 · 2026-10-19 09:03:24 +02:00
42 changed files with 2884 additions and 16 deletions
@@ -50,6 +50,10 @@ SMTP_USERNAME=
 SMTP_PASSWORD=
 MAIL_FROM=
 # Optional. E-mail address of the Admin who is told when a user undoes an e-mail change.
 # Empty: no notice is sent and an error is logged at each undo.
 # ADMIN_NOTIFY_EMAIL=
 # Optional. Directory for the database. Default /data (the container volume).
 # DATA_DIR=/data
@@ -19,3 +19,6 @@
 - source_plan: `_bmad-output/initiative-ping-mee-portal/epic-account-self-service/story-password-change-plan.md`
  summary: Known limitation, accepted in the code review of 2026-10-19: failed binds made elsewhere (authentik, other devices) count toward AD's lockout but not toward the portal's limit.
  evidence: Already stated in the plan's design notes. Fixing it needs a new directory read of the bad-password count (new adapter surface).
 - source_plan: `_bmad-output/initiative-ping-mee-portal/epic-account-self-service/story-profile-edit-plan.md`
  summary: A `Forbidden` from the directory (service account lacks a right) is not mapped to a message in the password change either, so the user gets a raw 500 page.
  evidence: Found reviewing profile edit; `LdapDirectory._service` raises `Forbidden` on INSUFFICIENT_ACCESS and `services.change_password` / `views.password` catch only DirectoryUnavailable, PasswordRejected and OutcomeUnknown. Pre-existing from the password change; fix in one place (AD-17) for every service.
@@ -0,0 +1,131 @@
 ---
 title: 'E-mail change with confirmation and undo'
 type: 'feature'
 ticket: '3'
 epic: 'epic-account-self-service'
 created: '2026-10-19'
 status: 'built'
 baseline_revision: '3fd8f27127e2ef6ba2b800245fdee17901659106'
 route: 'full'
 route_source: 'auto'
 risk: 'high'
 review: 'quick'
 review_source: 'pinned'
 lenses_ran: ['quick']
 review_loop_iteration: 0
 context:
  - '{project-root}/_bmad-output/architecture/architecture-ping-mee-self-service-platform-2026-10-15/ARCHITECTURE-SPINE.md'
  - '{project-root}/_bmad-output/ux-designs/ux-ping-mee-self-service-platform-2026-10-15/EXPERIENCE.md'
 ---
 <frozen-after-approval reason="human-owned intent — do not modify unless human renegotiates">
 ## Intent
 **Problem:** The e-mail address in AD is where announcements and security notices go, yet a user cannot change it, and an address taken over by someone else would go unnoticed. An address change must prove the new mailbox, and the old one must be able to reverse it.
 **Approach:** A page `/email` in the `profile` app asks for the new address and the current password and mails a one-hour confirmation link to the new address; AD is unchanged until that link is confirmed. Confirming writes `mail`, then mails the old address a notice with a 24-hour undo link. Undoing restores the old address, ends every session of the user, tells the user to change their password and notifies the Admin. Both links open a page that names the change; only a POST with a CSRF token applies it. Pending changes are stored in a new table owned by `profile`, with tokens only as hashes.
 ## Boundaries & Constraints
 **Always:** AD-2, AD-3, AD-4, AD-7, AD-8, AD-9, AD-11, AD-17, AD-19 apply. The request form accepts no username; the acting user comes from the session. The order on the request is fixed: validate the form locally, `throttle.check` before any AD or mail call, `verify_password`, then the write of the pending row and the queued mail in one transaction. A wrong current password counts on the same counter as the password page (`password-change`), so both pages together stay below AD's lockout threshold. Queueing a request is also limited to 5 per hour per user. Tokens are `secrets.token_urlsafe(32)`, stored as SHA-256 hashes, single-use, valid 1 hour (confirmation) and 24 hours (undo). A GET never changes anything and never consumes a token. Confirming and undoing each run the AD write, the token use, the audit entry and the queued mail in one transaction; a failed AD call rolls back and writes a failure entry after the rollback. A new request replaces an earlier pending one of the same user. Undo applies only while AD's `mail` still equals the address that was confirmed. Passwords, tokens and addresses are never logged and never audited; the audit holds only what happened. Link pages work whether or not the visitor is signed in. A user with no valid address in AD may still change it: no undo notice is queued and the `email.changed` entry records `notice: skipped`. The Admin notice goes to the optional environment variable `ADMIN_NOTIFY_EMAIL`; a value that is set but not an address stops startup with a clear message (AD-13); when it is empty no mail is queued, an error-level log line is written with the user key only, and the undo still succeeds. Every new string has German formal text, informal where it differs, in `portal/locale`. Mails are German formal, carry a plain-text part and never contain a password.
 **Never:** No change to `mail` before the confirmation POST. No change to another user, no username in a URL or form. No check that the address is unused by another account. No password stored. No JavaScript requirement: every page works as a plain POST.
 ## I/O & Edge-Case Matrix
 | Scenario | Input / State | Expected Output / Behavior | Error Handling |
 |----------|--------------|---------------------------|----------------|
 | Open the page | Signed-in user | Current address shown, fields for the new address and the current password; reached from the profile page | Directory down: "not possible now" (503) |
 | Request | Valid new address, correct current password | Pending row with hashed token; confirmation mail to the new address; page says a link was sent; AD unchanged; `email.change_requested` audited | n/a |
 | Wrong current password | Password does not match | "The current password is not correct." Nothing sent; counts toward the shared limit; failure audited | No AD change, no mail |
 | Limit reached | Failed checks at the effective limit, or 5 requests within an hour | 429 "Too many attempts. Please wait a few minutes."; no AD call, no mail; throttled attempt audited | Recovers after the pause |
 | Bad new address | Empty, not an address, over 254 characters, or equal to the current one (any letter case) | Error under the field; nothing sent; not counted | n/a |
 | Newer request | A second request while one is pending | The first link stops working; only the newest confirms | n/a |
 | Confirmation page | GET with a valid, unused, unexpired token | Page names old and new address and shows a button; nothing changes | n/a |
 | Confirm | POST with CSRF on that page | AD `mail` is the new address; token used; undo token created; notice with undo link mailed to the old address; `email.changed` audited; page confirms | Directory down: 503, token stays unused, failure audited |
 | Dead link | Token unknown, used, expired or replaced | "This link is not valid any more." (410); nothing changes; the page does not say which case | n/a |
 | Undo page | GET with a valid undo token | Page names the change and shows a button; nothing changes | n/a |
 | Undo | POST with CSRF | AD `mail` is the old address again; every session of the user ended; token used; Admin notified; `email.undone` audited; page says to change the password | Directory down: 503, token stays unused |
 | No current address | The user's `mail` in AD is empty or invalid | Request and confirmation work; no undo notice is queued; `email.changed` audited with `notice: skipped`; the page does not mention a notice | n/a |
 | No Admin address | `ADMIN_NOTIFY_EMAIL` empty | Undo succeeds; no Admin mail; an error-level log line names the user key | n/a |
 | Undo after another change | AD `mail` no longer equals the confirmed address | Page says the address was changed again since; nothing changes | n/a |
 | Not signed in, wrong method, no CSRF token | `/email` without session; GET to a POST-only action; POST without token | Redirect to sign-in; 405; 403; nothing changes | n/a |
 </frozen-after-approval>
 ## Code Map
 - `portal/profile/models.py`, `portal/profile/migrations/0001_initial.py` -- new (the app has no tables yet): `PendingEmailChange` (user key, kind `confirm` or `undo`, token hash unique, old and new address, created, expires, used). Owned by `profile` (AD-19).
 - `portal/profile/services.py` -- add `request_email_change`, `confirm_email_change`, `undo_email_change`, token helpers; reuse `ENDPOINT = "password-change"` and `failure_limit` / `pause_seconds` for the shared counter, `_record_failure` style for failure entries, `end_sessions`, `mailer.enqueue`, `mailer.is_valid_address`, `audit.record`, `audit.record_failure`, `directory.update_attributes(sam, {"mail": ...})`.
 - `portal/profile/forms.py`, `views.py`, `urls.py` -- `EmailChangeForm`; views `email` (login required), `email_confirm`, `email_undo` (token in the path, no login, GET shows, POST applies); routes `/email`, `/email/confirm/<token>`, `/email/undo/<token>`.
 - `portal/templates/profile/email*.html`, `portal/templates/mail/email_change_confirm/`, `email_changed_notice/`, `email_undone_admin/` (subject and body) -- pages and mails.
 - `portal/templates/profile/details.html` -- the profile page shows the address read-only today; add the link "Change e-mail address".
 - `portal/config/env.py`, `settings.py`, `.env.example` -- the optional `ADMIN_NOTIFY_EMAIL` (read in `env.py` with the other settings, checked with `mailer.is_valid_address` rules, exposed in `settings`); `portal/tests/test_config.py` gets a case for an invalid value. Link base: `settings.OIDC.public_base_url`.
 - `portal/config/settings.py` -- `SECURE_REFERRER_POLICY = "same-origin"` already keeps the token out of cross-site requests; keep it.
 - `portal/locale/de`, `de_DU`, `portal/tests/test_translations.py` -- German text and its coverage test.
 - Do not change: `accounts.throttle`, the directory adapter interface, the outbox.
 ## Tasks & Acceptance
 **Execution:**
 - [ ] `portal/profile/models.py`, migration -- `PendingEmailChange` -- AD-9, AD-19
 - [ ] `portal/profile/services.py` -- request, confirm, undo with the order and transactions above -- AD-4, AD-7, AD-8, AD-9
 - [ ] forms, views, urls, templates, mail templates, profile page link -- AD-17
 - [ ] `ADMIN_NOTIFY_EMAIL` in config, `.env.example` and its test; German text for every new string
 - [ ] `portal/tests/test_email_change.py` -- every matrix row with the fake directory and fake mailer, plus: only hashes stored, a GET changes nothing and consumes nothing, a replaced or used or expired token is dead, undo ends only that user's sessions, the shared counter with the password page, no password, token or address in logs or audit, `makemigrations --check` clean
 **Acceptance Criteria:**
 - Given a signed-in user, when they request a change with the correct password, then AD `mail` is unchanged and one confirmation mail with a link exists for the new address.
 - Given that link, when it is opened with GET, then the page names the change and nothing changes; when confirmed with POST, then AD holds the new address and the old address is mailed an undo link.
 - Given the undo link, when confirmed with POST within 24 hours, then the old address is restored, every session of the user ends, and the Admin notification is queued.
 - Given any token, then the database holds only its hash, and it works once.
 - Given a failure after the AD call inside the transaction, then the error is logged at error level with the change details (AD-4), without addresses or tokens, and the user is told the outcome is unclear.
 ## Implementation Notes
 - Built by a subagent from this plan, checked against the diff. `pytest portal/tests` 472 passed, 6 skipped; `ruff check` clean; `makemigrations --check` clean. The done checkpoint against the test OU and mailbox is not run.
 - Beyond the plan, kept after reading the diff:
  - `RedactLinkTokens` log filter and `redact()` in the request log path: the request log and Django's 500 log printed the URL, token included.
  - Undo also deletes the user's unused confirmation request, so a hijacker cannot request a change, wait for the undo and confirm it later.
  - A correct password on the request resets the `password-change` counter, as AD forgets failures after a good bind.
  - Undo restores the address AD held at confirmation time; the 400 and 429 pages on `/email` do not read AD (they omit the current-address box); extra pages: "address changed again" (409) and the unclear outcome (500).
  - `is_valid_address` moved to `portal/mailer/address.py` because `env.py` needs it before Django loads; `mailer.is_valid_address` still works. `PendingEmailChange` has a `sam` column for the exact AD name.
 - Not covered: expired and used rows stay in the table until a token-expiry job exists (AD-12).
 ## Plan Change Log
 ## Review Triage Log
 Pass 1 (quick lens): 0 high, 1 medium patched, 8 rejected (7 low, 2 false counted below).
 | Finding | Verdict | Route | Evidence |
 |---|---|---|---|
 | A POST to a link without a CSRF token logs the token on `django.security.csrf`; the redaction filter is only on `django.request` | medium | patch | Django's `CsrfViewMiddleware._reject` logs `request.path`; that logger has no filter and propagates to the root stdout handler. Put the filter on the handler so every logger is covered; test it through the configured handler. |
 | The undo compares AD `mail` before the transaction, not inside it | low | rejected | AD is not transactional, so a re-check inside only narrows the window to a few milliseconds; a concurrent edit by the same user in that window is not met in use. |
 | An earlier undo link works again after a later change is undone (A to B, B to C, undo C) | false | rejected | The earlier link still undoes the A to B change, the address is B again, the link is the user's own, single-use and expires in 24 hours; that is its meaning. |
 | The confirmation page names the address at request time while the notice goes to the address read at confirm time | low | rejected | Differs only if someone edits `mail` between request and confirm; the page still names the change correctly. |
 | The same-address check runs after the password check, so it costs a bind | false | rejected | It cannot be known locally, the plan's order puts the password check before any other directory read, and it is not counted. |
 | The 5-per-hour counter is hit after commit; parallel requests can pass the check first | low | rejected | The same check-then-hit design of the throttle is already in deferred-work.md (AD-7); the user needs a session and their password to send each mail. |
 | `throttle.reset` after a correct password weakens the shared counter | false | rejected | The password page resets on success too and AD forgets failures after a good bind, so the counter follows AD; it is listed in the Implementation Notes. |
 | HEAD on the link pages returns 405 | low | rejected | A scanner's HEAD changes nothing; 405 is harmless. |
 | A sign-in between two `end_sessions` calls in undo is not covered | low | rejected | Needs the password change and a sign-in inside milliseconds; the next undo or the password change ends it. |
 ## Design Notes
 **Why the shared counter.** The password page limits wrong current passwords to AD's lockout threshold minus one. A second page with its own counter would double the wrong binds AD sees. Both pages use the `password-change` counter; the five-per-hour request limit is separate, under its own endpoint name.
 **Confirming does not need a sign-in.** The link proves the mailbox, as the undo link does; the user may read the mail on a phone. A mail scanner opening the link only does a GET, which changes nothing.
 **Undo order.** Undo restores the address first, then ends the sessions, so a rolled-back write ends nothing. The Admin notice carries the username and time, never a password or token.
 ## Verification
 **Commands:**
 - `.venv/Scripts/python -m pytest portal/tests -q` -- expected: all tests pass, including the new ones
 - `.venv/Scripts/python -m ruff check portal manage.py docker` -- expected: no findings
 - `.venv/Scripts/python manage.py makemigrations --check --dry-run` -- expected: no changes detected
 **Manual checks (done checkpoint, on the Linux host against the test OU and the test mailbox):**
 - Request a change as `ptest-user1`, confirm that `mail` in AD is unchanged, open the link and see that it changes nothing, confirm, and see `mail` change and the notice arrive at the old address.
 - Open the undo link, undo, and see `mail` restored, a second browser signed out, and the Admin notice arrive.
@@ -0,0 +1,132 @@
 ---
 title: 'Profile edit'
 type: 'feature'
 ticket: '2'
 epic: 'epic-account-self-service'
 created: '2026-10-19'
 status: 'built'
 baseline_revision: 'ae4b30ec22d927bf1fb9e2a9ba32d6727c6935bb'
 route: 'full'
 route_source: 'auto'
 risk: 'medium'
 review: 'quick'
 review_source: 'pinned'
 lenses_ran: ['quick']
 review_loop_iteration: 0
 context:
  - '{project-root}/_bmad-output/architecture/architecture-ping-mee-self-service-platform-2026-10-15/ARCHITECTURE-SPINE.md'
  - '{project-root}/_bmad-output/ux-designs/ux-ping-mee-self-service-platform-2026-10-15/EXPERIENCE.md'
 ---
 <frozen-after-approval reason="human-owned intent — do not modify unless human renegotiates">
 ## Intent
 **Problem:** A name or phone number still changes only when the admin edits it in AD. A signed-in user cannot see the details the directory holds about them, correct them, or be sure what was saved.
 **Approach:** One page at `/profile` in the `profile` app. It shows the username and e-mail address (read only) and five fields filled from AD: first name, last name, display name, telephone and mobile. A service `update_details` checks the values, sends only the changed ones through the directory adapter's `update_attributes`, and writes the audit entry in the same transaction. The e-mail address is changed by ticket 3, not here.
 ## Boundaries & Constraints
 **Always:** AD-2, AD-3, AD-4, AD-17 apply. The acting user is read from the session only; the view and service accept no username, DN or user id, and a posted `username` field is ignored. Only `givenName`, `sn`, `displayName`, `telephoneNumber` and `mobile` are ever written, and values travel only as attribute values, never inside a filter or DN. Values are trimmed. First name, last name and display name are required and may hold any printable text; control characters (newline, tab, NUL) are refused. Length limits are AD's: 64 for first name and last name, 256 for display name. Telephone and mobile are optional and are stored in one format, `+` and digits only (see Design Notes): the input may be written freely, a leading `0` is read as Germany (`+49`), and anything that is not a number is refused. A field whose text equals the AD value is left alone: it is neither checked nor rewritten. A save with no change makes no AD call and no audit entry. The audit entry holds the names of the changed fields and never their values. A failed AD call rolls back and writes a failure entry after the rollback. Messages are plain words and name the field. Every new string has German formal text, informal where it differs, in `portal/locale`. The form works as a plain POST.
 **Never:** No e-mail edit and no link to one (ticket 3). No password, picture or group change. No attribute outside the five. No raw directory message shown to the user. No password re-check for these fields.
 ## I/O & Edge-Case Matrix
 | Scenario | Input / State | Expected Output / Behavior | Error Handling |
 |----------|--------------|---------------------------|----------------|
 | Open the page | Signed-in user | Five fields with the AD values, username and e-mail as plain text; menu entry "Profile" active | Directory down: "not possible now" page (503), no form |
 | Save changes | Valid values, at least one differs from AD | Only the changed attributes written; `profile.updated` audited with the names of the changed fields only; redirect to the page with a "saved" notice; this session's shown name updated | n/a |
 | Nothing changed | Same values as in AD | No AD call, no audit; notice "Nothing to change" | n/a |
 | Clear a number | Telephone or mobile left empty | The attribute is cleared in AD | n/a |
 | Number written freely | `0171 / 123456`, `0171123456`, `+49 171 123456`, `+49 (0) 171-123456`, `0049 171 123456` | Stored as `+49171123456`; the page shows the stored form after the save | n/a |
 | Number from another country | `+43 660 1234567`, `0043 660 1234567` | Stored as `+436601234567` | n/a |
 | Not a number | Letters, a `+` inside the number, no `0` or `+` at the start (`171123456`), fewer than 7 or more than 15 digits | Error under the field: the number must start with `+` and the country code, or with `0`; nothing written | Not audited |
 | Required name empty | First name, last name or display name empty | Error under the field; nothing written | Not audited |
 | Invalid value | Too long, or a control character | Error under the field and in a summary; nothing written; the other typed values stay in the form | Not audited |
 | Injection strings | `*)(uid=*`, `\00`, `CN=x,DC=y`, a newline | Stored as plain text where allowed or refused by the rules; never alter a query; no other object or attribute changes | n/a |
 | Another user named | A posted `username` or `sam` | Ignored; the session user is edited | n/a |
 | AD refuses a value | Syntax or constraint error from AD | "A value was not accepted. Please check your entries." Nothing changed; failure audited | Transaction rolled back |
 | Directory down on save | AD unavailable | "This is not possible right now. Please try again later." (503); failure audited | Transaction rolled back |
 | Not signed in, wrong method, no CSRF token | Any request | Redirect to sign-in; 405; 403; nothing changes | n/a |
 </frozen-after-approval>
 ## Code Map
 - `portal/profile/forms.py` -- add `DetailsForm` (five `CharField`s, `strip=True`, `max_length` per field, the three names required, a control-character check, `normalize_phone` for the two number fields). `PasswordChangeForm` stays.
 - `portal/profile/services.py` -- add `load_details(user)` (directory read to initial values) and `update_details(user, values, *, ip)` (diff against a fresh read, `transaction.atomic` with `directory.update_attributes` and `audit.record`, failure entry after rollback). Reuse `_record_failure` style and the typed errors `DirectoryUnavailable`, `ValidationFailed`.
 - `portal/profile/views.py`, `urls.py` -- add the `profile` view (GET, POST) at `/profile`, name `profile`; `login_required`; one-time session flag for the notice as the password page does.
 - `portal/templates/profile/details.html`, a field partial for text inputs (`_field.html` is password-only: `type="password"`).
 - `portal/shell/menu.py` -- add `Entry(_("Profile"), "profile")` after Home.
 - `portal/accounts/identity.py` (`remember`, `_display_name`), `session_store.DISPLAY_NAME` -- the session caches the display name for 60 s (`ROLE_CACHE_SECONDS`); refresh it in the saving session, other sessions follow on their own.
 - `portal/adapters/directory/` -- `update_attributes` already exists in the interface, `FakeDirectory` and `LdapDirectory` (`MOD_REPLACE`, empty value clears, allow-list enforced, NUL refused). Reuse unchanged.
 - `portal/locale/de`, `de_DU` -- German text; `portal/tests/test_translations.py` checks coverage.
 - Do not change: the password flow, `accounts.throttle`, the adapter.
 ## Tasks & Acceptance
 **Execution:**
 - [ ] `portal/profile/forms.py` -- `DetailsForm` and `normalize_phone` -- FR-9
 - [ ] `portal/profile/services.py` -- `load_details`, `update_details` -- AD-2, AD-3, AD-4
 - [ ] `portal/profile/views.py`, `urls.py`, `portal/templates/profile/*`, `portal/shell/menu.py` -- page, notice, error mapping to messages and status codes, menu entry -- AD-17
 - [ ] `portal/locale/de`, `de_DU` -- German text for every new string
 - [ ] `portal/tests/test_profile_edit.py` -- every matrix row with the fake directory, plus: a table of phone inputs and their stored form, an unchanged legacy value left alone, injection strings, a posted username ignored, another user's data untouched, a rolled-back write leaves no audit entry, the session name refreshed
 **Acceptance Criteria:**
 - Given a signed-in user, when they open `/profile`, then the five fields hold their AD values and the username and e-mail are shown without an input.
 - Given valid new values, when they save, then the matching AD attributes hold them, one audit entry exists, and nothing else about the user changed.
 - Given invalid values, when they save, then the message names the field and nothing is written (the directory is read once, to tell a changed value from an old one).
 - Given any request, then no username, DN or user id is accepted from the client.
 - Given a failure after the AD call inside the transaction, then the error is logged at error level with the change details (AD-4) and the user is told it is unclear.
 ## Implementation Notes
 - Built by a subagent from this plan, checked against the diff. `pytest portal/tests` 423 passed, 6 skipped; `ruff check` clean. The done checkpoint against the test OU is not run.
 - The checks (required, control characters, length, phone) run in `forms.check_details`, called from `services.update_details`, not in `DetailsForm`: they need the directory value to leave an unchanged old value alone. `DetailsForm` only trims. Consequence: an invalid save reads the directory once; nothing is written or audited.
 - A required name that is empty in AD and left empty is "unchanged" and passes; only a changed value is checked.
 - On a refused or invalid save the page re-reads the directory for the username and e-mail rows; if that read fails the rows are left out of that response.
 - Failure entries: `profile.update_failed` with reason `unavailable`, `refused`, `error` or `unclear`.
 ## Plan Change Log
 ## Review Triage Log
 Pass 1 (quick lens): 0 high, 1 medium, 2 low patched, 3 low rejected.
 | Finding | Verdict | Route | Evidence |
 |---|---|---|---|
 | Other adapter errors (`Forbidden` when the service account lacks the write right) are audited and re-raised, so the user gets a raw 500 page | medium | patch | `LdapDirectory._service` turns `INSUFFICIENT_ACCESS` into `Forbidden`; `update_details` and `profile()` catch neither. Map it to the unavailable page, audit `forbidden`, add a test. |
 | An invalid save reads the directory twice (`update_details`, then `_details_or_none`) | low | patch | Real; the acceptance line says once, and the two reads can disagree. `DetailsInvalid` and `DetailsNotAccepted` carry the details already read. |
 | Step label `audit` is also logged when the commit itself fails | low | patch | `step` is set before `audit.record` and never advanced; set `commit` after the record. |
 | Saving session's name not refreshed after an unclear outcome | low | rejected | Stale for at most 60 s (`ROLE_CACHE_SECONDS`); the user sees the unclear page; the fix adds a branch for no real harm. |
 | Phone value with a newline gets the control-character message; length checked before normalizing | low | rejected | The message is true and the value is refused either way; 64 characters of separators is not met in use. |
 | Missing tests (single read, unmapped errors, newline in a phone, name after unclear) | low | patch for the first two, rejected for the rest | Covered by the two patches above; the others test the rejected findings. |
 ## Design Notes
 **Why compare with a fresh read.** The form is filled from AD when the page opens; the save reads AD again and sends only the fields that differ from that read. A change made elsewhere in between is not overwritten by an unchanged field.
 **Phone numbers.** Remove spaces, `-`, `/`, `.` and brackets; drop a `(0)` that follows a country code. Then: `+` and digits stay as they are; `00` becomes `+`; a single leading `0` becomes `+49` and the `0` is dropped; anything else is refused. A `+` anywhere but the start, a letter, or a result of fewer than 7 or more than 15 digits is refused. No check that the country code exists.
 ```
 0171 / 123456       -> +49171123456
 +49 (0) 171-123456  -> +49171123456
 0049 171 123456     -> +49171123456
 171123456           -> refused (no 0 or + at the start)
 ```
 **Values already in AD.** Numbers entered by the admin may be in another format or not a number at all ("intern 23"). The form shows them as they are; a field left as it is passes unchanged, so one old value never blocks saving another field.
 **No throttle, no password.** The password page needs both because a wrong password reaches AD's lockout. These fields are a plain authenticated write, so only the session and CSRF protect them.
 ## Verification
 **Commands:**
 - `.venv/Scripts/python -m pytest portal/tests -q` -- expected: all tests pass, including the new ones
 - `.venv/Scripts/python -m ruff check portal manage.py docker` -- expected: no findings
 **Manual checks (done checkpoint, on the Linux host against the test OU):**
 - Sign in as `ptest-user1`, change each of the five fields, and see the values in AD (`Get-ADUser -Properties givenName,sn,displayName,telephoneNumber,mobile`).
 - Submit an injection string and confirm only the one attribute holds it.
@@ -10,6 +10,8 @@ from pathlib import Path
 from urllib.parse import urlsplit
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 from portal.mailer.address import is_valid_address
 SMTP_SECURITY_MODES = ("starttls", "ssl")
 SMTP_DEFAULT_PORTS = {"starttls": 587, "ssl": 465}
 LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
@@ -80,6 +82,7 @@ class Config:
    directory: DirectoryConfig
    oidc: OidcConfig
    smtp: SmtpConfig
    admin_notify_email: str = ""  # where the notice of an undone e-mail change goes; may be empty
 class _Reader:
@@ -251,12 +254,25 @@ def _read_smtp(r: _Reader) -> SmtpConfig:
    )
 def _read_admin_notify_email(r: _Reader) -> str:
    value = r.raw("ADMIN_NOTIFY_EMAIL")
    if value is None:
        return ""
    if not is_valid_address(value):
        r.problem("ADMIN_NOTIFY_EMAIL: must be an e-mail address, or left empty")
        return ""
    return value
 def load_config(environ: Mapping[str, str]) -> Config:
    reader = _Reader(environ)
    core = _read_core(reader)
    directory = _read_directory(reader)
    oidc = _read_oidc(reader, debug=core["debug"])
    smtp = _read_smtp(reader)
    admin_notify_email = _read_admin_notify_email(reader)
    if reader.problems:
        raise ConfigError(reader.problems)
-    return Config(**core, directory=directory, oidc=oidc, smtp=smtp)
+    return Config(
        **core, directory=directory, oidc=oidc, smtp=smtp, admin_notify_email=admin_notify_email
    )
@@ -4,6 +4,7 @@ from __future__ import annotations
 import json
 import logging
 import re
 from contextvars import ContextVar
 from datetime import UTC, datetime
@@ -13,6 +14,25 @@ request_id_var: ContextVar[str] = ContextVar("request_id", default="-")
 _STANDARD = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__) | {"message", "asctime"}
 # One-time links carry their token in the path (AD-9); a log line must never hold it.
 _LINK_TOKEN = re.compile(r"(/email/(?:confirm|undo)/)[^/\s\"']+")
 def redact(text: str) -> str:
    return _LINK_TOKEN.sub(r"\1<redacted>", text)
 class RedactLinkTokens(logging.Filter):
    """Blank the token of a one-time link in a record's message."""
    def filter(self, record: logging.LogRecord) -> bool:
        message = record.getMessage()
        redacted = redact(message)
        if redacted != message:
            record.msg, record.args = redacted, ()
        return True
 class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, object] = {
@@ -37,11 +57,13 @@ def build_logging(level: str) -> dict:
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {"json": {"()": "portal.config.logging.JsonFormatter"}},
        "filters": {"redact_link_tokens": {"()": "portal.config.logging.RedactLinkTokens"}},
        "handlers": {
            "stdout": {
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
                "formatter": "json",
                "filters": ["redact_link_tokens"],
            }
        },
        "loggers": {"django.request": {"level": "ERROR"}},
@@ -7,7 +7,7 @@ import re
 import time
 import uuid
-from .logging import request_id_var
+from .logging import redact, request_id_var
 logger = logging.getLogger("portal.request")
@@ -48,7 +48,7 @@ class RequestIdMiddleware:
                extra={
                    "request_id": request_id,
                    "method": request.method,
-                    "path": request.path,
+                    "path": redact(request.path),
                    "status": status,
                    "duration_ms": round((time.monotonic() - started) * 1000, 1),
                },
@@ -26,6 +26,8 @@ DISPLAY_TIME_ZONE = _config.default_timezone
 DIRECTORY = _config.directory
 OIDC = _config.oidc
 SMTP = _config.smtp
 # Gets the notice when an e-mail change is undone; empty means no notice (an error is logged).
 ADMIN_NOTIFY_EMAIL = _config.admin_notify_email
 INSTALLED_APPS = [
    "django.contrib.contenttypes",
@@ -247,3 +247,204 @@ msgstr "Wenn Sie das waren, müssen Sie nichts tun. Wenn nicht, wenden Sie sich
 msgid "This e-mail never contains your password."
 msgstr "Diese E-Mail enthält niemals Ihr Passwort."
 msgid "Profile"
 msgstr "Profil"
 msgid "First name"
 msgstr "Vorname"
 msgid "Last name"
 msgstr "Nachname"
 msgid "Display name"
 msgstr "Anzeigename"
 msgid "Telephone"
 msgstr "Telefon"
 msgid "Mobile"
 msgstr "Mobil"
 #, python-format
 msgid "%(field)s is required."
 msgstr "%(field)s ist erforderlich."
 #, python-format
 msgid "%(field)s must not contain line breaks or control characters."
 msgstr "%(field)s darf keine Zeilenumbrüche oder Steuerzeichen enthalten."
 #, python-format
 msgid "%(field)s must not be longer than %(max)s characters."
 msgstr "%(field)s darf nicht länger als %(max)s Zeichen sein."
 #, python-format
 msgid "%(field)s must be a number that starts with + and the country code, or with 0."
 msgstr "%(field)s muss eine Nummer sein, die mit + und der Landesvorwahl oder mit 0 beginnt."
 msgid "These details are stored in your account. Changes apply to every service that uses it."
 msgstr "Diese Angaben sind in Ihrem Konto gespeichert. Änderungen gelten für jeden Dienst, der es verwendet."
 msgid "Your changes were saved."
 msgstr "Ihre Änderungen wurden gespeichert."
 msgid "Nothing to change."
 msgstr "Es gibt nichts zu ändern."
 msgid "Your changes were not saved."
 msgstr "Ihre Änderungen wurden nicht gespeichert."
 msgid "User name"
 msgstr "Benutzername"
 msgid "E-mail address"
 msgstr "E-Mail-Adresse"
 msgid "For example +49 171 123456. A number that starts with 0 is read as German."
 msgstr "Zum Beispiel +49 171 123456. Eine Nummer, die mit 0 beginnt, gilt als deutsche Nummer."
 msgid "Save changes"
 msgstr "Änderungen speichern"
 msgid "A value was not accepted. Please check your entries."
 msgstr "Ein Wert wurde nicht angenommen. Bitte prüfen Sie Ihre Eingaben."
 msgid "Your details may have been changed, but the portal could not finish. Please open your profile and check them. If they are wrong, contact the administrator."
 msgstr "Ihre Angaben wurden möglicherweise geändert, aber das Portal konnte nicht abschließen. Bitte öffnen Sie Ihr Profil und prüfen Sie sie. Wenn sie falsch sind, wenden Sie sich an die Administration."
 msgid "Back to the profile"
 msgstr "Zurück zum Profil"
 msgid "This is already your e-mail address."
 msgstr "Das ist bereits Ihre E-Mail-Adresse."
 msgid "This e-mail address is too long."
 msgstr "Diese E-Mail-Adresse ist zu lang."
 msgid "Please enter a valid e-mail address."
 msgstr "Bitte geben Sie eine gültige E-Mail-Adresse ein."
 msgid "Change e-mail address"
 msgstr "E-Mail-Adresse ändern"
 msgid "Announcements and security notices are sent to this address. We send a link to the new address. Your address changes only after you open the link and confirm."
 msgstr "Ankündigungen und Sicherheitshinweise werden an diese Adresse gesendet. Wir senden einen Link an die neue Adresse. Ihre Adresse ändert sich erst, wenn Sie den Link öffnen und bestätigen."
 msgid "No link was sent."
 msgstr "Es wurde kein Link gesendet."
 msgid "Current e-mail address"
 msgstr "Aktuelle E-Mail-Adresse"
 msgid "New e-mail address"
 msgstr "Neue E-Mail-Adresse"
 msgid "We check it before anything is sent."
 msgstr "Wir prüfen es, bevor etwas gesendet wird."
 msgid "Send confirmation link"
 msgstr "Bestätigungslink senden"
 msgid "Confirmation link sent"
 msgstr "Bestätigungslink gesendet"
 msgid "We sent a link to your new e-mail address. It is valid for one hour."
 msgstr "Wir haben einen Link an Ihre neue E-Mail-Adresse gesendet. Er ist eine Stunde gültig."
 msgid "Your address has not changed yet. It changes after you open the link and confirm."
 msgstr "Ihre Adresse hat sich noch nicht geändert. Sie ändert sich, sobald Sie den Link öffnen und bestätigen."
 msgid "Confirm new e-mail address"
 msgstr "Neue E-Mail-Adresse bestätigen"
 msgid "The e-mail address of your account will change. Nothing changes until you press the button."
 msgstr "Die E-Mail-Adresse Ihres Kontos wird geändert. Erst wenn Sie die Schaltfläche drücken, ändert sich etwas."
 msgid "A notice with a way to undo the change is sent to the current address."
 msgstr "An die aktuelle Adresse wird ein Hinweis mit der Möglichkeit gesendet, die Änderung rückgängig zu machen."
 msgid "Confirm change"
 msgstr "Änderung bestätigen"
 msgid "E-mail address changed"
 msgstr "E-Mail-Adresse geändert"
 msgid "Your e-mail address was changed."
 msgstr "Ihre E-Mail-Adresse wurde geändert."
 msgid "A notice with a way to undo the change was sent to your previous address."
 msgstr "Ein Hinweis mit der Möglichkeit, die Änderung rückgängig zu machen, wurde an Ihre bisherige Adresse gesendet."
 msgid "Undo e-mail change"
 msgstr "E-Mail-Änderung rückgängig machen"
 msgid "The e-mail address of your account was changed. If you did not do this, undo the change now."
 msgstr "Die E-Mail-Adresse Ihres Kontos wurde geändert. Wenn Sie das nicht waren, machen Sie die Änderung jetzt rückgängig."
 msgid "Previous e-mail address"
 msgstr "Bisherige E-Mail-Adresse"
 msgid "Undoing restores the previous address, signs you out on all devices and informs the administrator."
 msgstr "Beim Rückgängigmachen wird die bisherige Adresse wiederhergestellt, Sie werden auf allen Geräten abgemeldet und die Administration wird informiert."
 msgid "Undo change"
 msgstr "Änderung rückgängig machen"
 msgid "E-mail change undone"
 msgstr "E-Mail-Änderung rückgängig gemacht"
 msgid "Your previous e-mail address is back. You were signed out on all devices."
 msgstr "Ihre bisherige E-Mail-Adresse ist wieder eingetragen. Sie wurden auf allen Geräten abgemeldet."
 msgid "Someone else may have had access to your account. Please change your password now."
 msgstr "Möglicherweise hatte jemand anderes Zugriff auf Ihr Konto. Bitte ändern Sie jetzt Ihr Passwort."
 msgid "Address changed again"
 msgstr "Adresse erneut geändert"
 msgid "The e-mail address was changed again since then. Nothing was changed now. If this was not you, contact the administrator."
 msgstr "Die E-Mail-Adresse wurde seitdem erneut geändert. Jetzt wurde nichts geändert. Wenn Sie das nicht waren, wenden Sie sich an die Administration."
 msgid "Link not valid"
 msgstr "Link nicht gültig"
 msgid "This link is not valid any more."
 msgstr "Dieser Link ist nicht mehr gültig."
 msgid "Your e-mail address may have been changed, but the portal could not finish. Please check your profile or contact the administrator."
 msgstr "Ihre E-Mail-Adresse wurde möglicherweise geändert, aber das Portal konnte den Vorgang nicht abschließen. Bitte prüfen Sie Ihr Profil oder wenden Sie sich an die Administration."
 msgid "Please confirm your new e-mail address"
 msgstr "Bitte bestätigen Sie Ihre neue E-Mail-Adresse"
 msgid "this e-mail address was entered as the new e-mail address of your portal account. To confirm it, open this link and press the button. The link is valid for one hour:"
 msgstr "diese E-Mail-Adresse wurde als neue E-Mail-Adresse Ihres Portal-Kontos angegeben. Um sie zu bestätigen, öffnen Sie diesen Link und drücken Sie die Schaltfläche. Der Link ist eine Stunde gültig:"
 msgid "Your address changes only after you confirm. If you did not ask for this, ignore this e-mail. Nothing will change."
 msgstr "Ihre Adresse ändert sich erst, wenn Sie bestätigen. Wenn Sie das nicht veranlasst haben, ignorieren Sie diese E-Mail. Es ändert sich nichts."
 msgid "Your e-mail address was changed"
 msgstr "Ihre E-Mail-Adresse wurde geändert"
 #, python-format
 msgid "the e-mail address of your portal account was changed on %(when)s (%(zone)s) to %(address)s."
 msgstr "die E-Mail-Adresse Ihres Portal-Kontos wurde am %(when)s (%(zone)s) auf %(address)s geändert."
 msgid "If you did this, you do not need to do anything. If you did not, open this link to restore this address. The link is valid for 24 hours:"
 msgstr "Wenn Sie das waren, müssen Sie nichts tun. Wenn nicht, öffnen Sie diesen Link, um diese Adresse wiederherzustellen. Der Link ist 24 Stunden gültig:"
 msgid "Undoing signs you out on all devices and informs the administrator."
 msgstr "Beim Rückgängigmachen werden Sie auf allen Geräten abgemeldet und die Administration wird informiert."
 msgid "An e-mail change was undone"
 msgstr "Eine E-Mail-Änderung wurde rückgängig gemacht"
 msgid "Hello,"
 msgstr "Guten Tag,"
 #, python-format
 msgid "the user %(username)s undid an e-mail change on %(when)s (%(zone)s). All sessions of this user were ended."
 msgstr "der Benutzer %(username)s hat am %(when)s (%(zone)s) eine E-Mail-Änderung rückgängig gemacht. Alle Sitzungen dieses Benutzers wurden beendet."
 msgid "The user may have lost control of the account. Please check it."
 msgstr "Der Benutzer hat möglicherweise die Kontrolle über das Konto verloren. Bitte prüfen Sie es."
@@ -68,3 +68,60 @@ msgstr "Eine Bestätigung wurde an deine E-Mail-Adresse gesendet."
 msgid "Your password may have been changed, but the portal could not finish. Try to sign in with the new password. If that does not work, contact the administrator."
 msgstr "Dein Passwort wurde möglicherweise geändert, aber das Portal konnte den Vorgang nicht abschließen. Versuche, dich mit dem neuen Passwort anzumelden. Wenn das nicht klappt, wende dich an die Administration."
 msgid "These details are stored in your account. Changes apply to every service that uses it."
 msgstr "Diese Angaben sind in deinem Konto gespeichert. Änderungen gelten für jeden Dienst, der es verwendet."
 msgid "Your changes were saved."
 msgstr "Deine Änderungen wurden gespeichert."
 msgid "Your changes were not saved."
 msgstr "Deine Änderungen wurden nicht gespeichert."
 msgid "A value was not accepted. Please check your entries."
 msgstr "Ein Wert wurde nicht angenommen. Bitte prüfe deine Eingaben."
 msgid "Your details may have been changed, but the portal could not finish. Please open your profile and check them. If they are wrong, contact the administrator."
 msgstr "Deine Angaben wurden möglicherweise geändert, aber das Portal konnte nicht abschließen. Bitte öffne dein Profil und prüfe sie. Wenn sie falsch sind, wende dich an die Administration."
 msgid "This is already your e-mail address."
 msgstr "Das ist bereits deine E-Mail-Adresse."
 msgid "Please enter a valid e-mail address."
 msgstr "Bitte gib eine gültige E-Mail-Adresse ein."
 msgid "Announcements and security notices are sent to this address. We send a link to the new address. Your address changes only after you open the link and confirm."
 msgstr "Ankündigungen und Sicherheitshinweise werden an diese Adresse gesendet. Wir senden einen Link an die neue Adresse. Deine Adresse ändert sich erst, wenn du den Link öffnest und bestätigst."
 msgid "We sent a link to your new e-mail address. It is valid for one hour."
 msgstr "Wir haben einen Link an deine neue E-Mail-Adresse gesendet. Er ist eine Stunde gültig."
 msgid "Your address has not changed yet. It changes after you open the link and confirm."
 msgstr "Deine Adresse hat sich noch nicht geändert. Das passiert erst, wenn du den Link öffnest und bestätigst."
 msgid "The e-mail address of your account will change. Nothing changes until you press the button."
 msgstr "Die E-Mail-Adresse deines Kontos wird geändert. Erst wenn du die Schaltfläche drückst, ändert sich etwas."
 msgid "Your e-mail address was changed."
 msgstr "Deine E-Mail-Adresse wurde geändert."
 msgid "A notice with a way to undo the change was sent to your previous address."
 msgstr "Ein Hinweis mit der Möglichkeit, die Änderung rückgängig zu machen, wurde an deine bisherige Adresse gesendet."
 msgid "The e-mail address of your account was changed. If you did not do this, undo the change now."
 msgstr "Die E-Mail-Adresse deines Kontos wurde geändert. Wenn du das nicht warst, mach die Änderung jetzt rückgängig."
 msgid "Undoing restores the previous address, signs you out on all devices and informs the administrator."
 msgstr "Beim Rückgängigmachen wird die bisherige Adresse wiederhergestellt, du wirst auf allen Geräten abgemeldet und die Administration wird informiert."
 msgid "Your previous e-mail address is back. You were signed out on all devices."
 msgstr "Deine bisherige E-Mail-Adresse ist wieder eingetragen. Du wurdest auf allen Geräten abgemeldet."
 msgid "Someone else may have had access to your account. Please change your password now."
 msgstr "Möglicherweise hatte jemand anderes Zugriff auf dein Konto. Bitte ändere jetzt dein Passwort."
 msgid "The e-mail address was changed again since then. Nothing was changed now. If this was not you, contact the administrator."
 msgstr "Die E-Mail-Adresse wurde seitdem erneut geändert. Jetzt wurde nichts geändert. Wenn du das nicht warst, wende dich an die Administration."
 msgid "Your e-mail address may have been changed, but the portal could not finish. Please check your profile or contact the administrator."
 msgstr "Deine E-Mail-Adresse wurde möglicherweise geändert, aber das Portal konnte den Vorgang nicht abschließen. Bitte prüfe dein Profil oder wende dich an die Administration."
@@ -0,0 +1,9 @@
 """The one rule for a usable e-mail address. No Django imports: the config reads it at startup."""
 import re
 _ADDRESS = re.compile(r"^[^@\s<>,;\"()\[\]\\]+@[^@\s<>,;\"()\[\]\\]+\.[^@\s<>,;\"()\[\]\\]+$")
 def is_valid_address(value: str) -> bool:
    return len(value) <= 254 and bool(_ADDRESS.match(value))
@@ -9,7 +9,6 @@ duplicate one for a password or sign-in notice.
 from __future__ import annotations
 import logging
 import re
 from dataclasses import dataclass
 from datetime import datetime, timedelta
@@ -21,6 +20,7 @@ from django.utils import timezone, translation
 from portal.adapters.mail import MailMessage, MailRejected, MailUnavailable
 from portal.adapters.registry import get_mailer
 from .address import is_valid_address
 from .models import OutboxMail
 logger = logging.getLogger(__name__)
@@ -31,8 +31,6 @@ RATE_PER_MINUTE = 20
 LEASE_SECONDS = 10 * 60  # a row that has been `sending` this long is given up on
 STOP_AFTER_CONSECUTIVE_OUTAGES = 3
 _ADDRESS = re.compile(r"^[^@\s<>,;\"()\[\]\\]+@[^@\s<>,;\"()\[\]\\]+\.[^@\s<>,;\"()\[\]\\]+$")
@dataclass(frozen=True)
 class FlushResult:
@@ -44,10 +42,6 @@ class FlushResult:
        return f"sent={self.sent} retried={self.retried} failed={self.failed}"
 def is_valid_address(value: str) -> bool:
    return len(value) <= 254 and bool(_ADDRESS.match(value))
 def enqueue(template: str, recipient: str, context: dict | None = None) -> OutboxMail:
    """Render the e-mail in the portal-wide language and queue it.
@@ -1,6 +1,12 @@
 import re
 import unicodedata
 from django import forms
 from django.utils.translation import gettext
 from django.utils.translation import gettext_lazy as _
 from portal.mailer.address import is_valid_address
 MAX_PASSWORD_LENGTH = 256  # Active Directory's limit
@@ -33,3 +39,153 @@ class PasswordChangeForm(forms.Form):
        elif new and current and new == current:
            self.add_error("new_password", _("The new password must be different."))
        return data
 # -- profile details --------------------------------------------------------------------------
 # Form field name -> (label, maximum length (AD's limit), required)
 DETAIL_FIELDS = {
    "first_name": (_("First name"), 64, True),
    "last_name": (_("Last name"), 64, True),
    "display_name": (_("Display name"), 256, True),
    "telephone": (_("Telephone"), 64, False),
    "mobile": (_("Mobile"), 64, False),
 }
 PHONE_FIELDS = ("telephone", "mobile")
 PHONE_MIN_DIGITS = 7
 PHONE_MAX_DIGITS = 15
 DEFAULT_COUNTRY_CODE = "+49"  # a leading 0 is read as Germany
 _PHONE_SEPARATORS = re.compile(r"[ \-/.()]")
 _PHONE_SHAPE = re.compile(r"\+?[0-9]+")
 class InvalidNumber(ValueError):
    """The text cannot be read as a telephone number."""
 def normalize_phone(text: str) -> str:
    """The one stored form of a telephone number: `+` and digits only.
    Spaces, `-`, `/`, `.` and brackets are removed and a `(0)` after a country code is dropped.
    `00` becomes `+`, a single leading `0` becomes `+49`. Anything else raises InvalidNumber.
    The result is its own normal form, so normalizing twice changes nothing.
    """
    text = text.strip()
    if text.startswith(("+", "00")):
        text = text.replace("(0)", "")
    text = _PHONE_SEPARATORS.sub("", text)
    if not _PHONE_SHAPE.fullmatch(text):
        raise InvalidNumber(text)
    if text.startswith("00"):
        text = "+" + text[2:]
    elif text.startswith("0"):
        text = DEFAULT_COUNTRY_CODE + text[1:]
    elif not text.startswith("+"):
        raise InvalidNumber(text)
    if not PHONE_MIN_DIGITS <= len(text) - 1 <= PHONE_MAX_DIGITS:
        raise InvalidNumber(text)
    return text
 def _has_control_character(text: str) -> bool:
    return any(unicodedata.category(ch) in ("Cc", "Zl", "Zp") for ch in text)
 def _check_one(name: str, value: str) -> str:
    """The cleaned value of a changed field, or a ValueError carrying the message."""
    label, max_length, required = DETAIL_FIELDS[name]
    if not value:
        if required:
            raise ValueError(gettext("%(field)s is required.") % {"field": label})
        return ""
    if _has_control_character(value):
        raise ValueError(
            gettext("%(field)s must not contain line breaks or control characters.")
            % {"field": label}
        )
    if len(value) > max_length:
        raise ValueError(
            gettext("%(field)s must not be longer than %(max)s characters.")
            % {"field": label, "max": max_length}
        )
    if name in PHONE_FIELDS:
        try:
            return normalize_phone(value)
        except InvalidNumber:
            raise ValueError(
                gettext(
                    "%(field)s must be a number that starts with + and the country code, or with 0."
                )
                % {"field": label}
            ) from None
    return value
 def check_details(
    values: dict[str, str], current: dict[str, str]
 ) -> tuple[dict[str, str], dict[str, str]]:
    """Check submitted values against what the directory holds now.
    Returns (cleaned, errors). A field whose trimmed text equals the directory value is neither
    checked nor changed, so one old value in the directory never blocks saving another field.
    A number written in another way but equal once normalized is unchanged too.
    """
    cleaned: dict[str, str] = {}
    errors: dict[str, str] = {}
    for name in DETAIL_FIELDS:
        value = (values.get(name) or "").strip()
        existing = current.get(name) or ""
        if value == existing.strip():
            cleaned[name] = existing
            continue
        try:
            result = _check_one(name, value)
        except ValueError as exc:
            errors[name] = str(exc)
            continue
        cleaned[name] = existing if result == existing.strip() else result
    return cleaned, errors
 class _TextField(forms.CharField):
    """Trims only: the checks need the directory value and live in `check_details`."""
    def run_validators(self, value):
        pass
 class DetailsForm(forms.Form):
    first_name = _TextField(required=False, strip=True)
    last_name = _TextField(required=False, strip=True)
    display_name = _TextField(required=False, strip=True)
    telephone = _TextField(required=False, strip=True)
    mobile = _TextField(required=False, strip=True)
    def apply_errors(self, errors: dict[str, str]) -> None:
        for name, message in errors.items():
            self.add_error(name, message)
 # -- e-mail change ----------------------------------------------------------------------------
 MAX_ADDRESS_LENGTH = 254
 class EmailChangeForm(forms.Form):
    new_email = forms.CharField(
        strip=True,
        required=False,  # the messages below say it better than the default
        widget=forms.EmailInput(),
    )
    current_password = _password_field()
    def clean_new_email(self):
        value = self.cleaned_data["new_email"]
        if not value:
            raise forms.ValidationError(_("Please fill in this field."))
        if len(value) > MAX_ADDRESS_LENGTH:
            raise forms.ValidationError(_("This e-mail address is too long."))
        if not is_valid_address(value):
            raise forms.ValidationError(_("Please enter a valid e-mail address."))
        return value
@@ -0,0 +1,29 @@
 # Generated by Django 5.2.18 on 2026-10-19 07:52
 from django.db import migrations, models
 class Migration(migrations.Migration):
    initial = True
    dependencies = [
    ]
    operations = [
        migrations.CreateModel(
            name='PendingEmailChange',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('user_key', models.CharField(db_index=True, max_length=256)),
                ('sam', models.CharField(max_length=256)),
                ('kind', models.CharField(choices=[('confirm', 'Confirm'), ('undo', 'Undo')], max_length=10)),
                ('token_hash', models.CharField(max_length=64, unique=True)),
                ('old_address', models.CharField(blank=True, max_length=254)),
                ('new_address', models.CharField(max_length=254)),
                ('created_at', models.DateTimeField()),
                ('expires_at', models.DateTimeField()),
                ('used_at', models.DateTimeField(blank=True, null=True)),
            ],
        ),
    ]
@@ -0,0 +1,25 @@
 from django.db import models
 class PendingEmailChange(models.Model):
    """A one-time link for an e-mail change. Owned by `profile` (AD-19).
    The token itself is never stored, only its SHA-256 hash (AD-9). A `confirm` row holds the
    address the user asked for (`new_address`) and the one they had then (`old_address`). An
    `undo` row holds the address to restore (`old_address`) and the one that was confirmed
    (`new_address`).
    """
    class Kind(models.TextChoices):
        CONFIRM = "confirm"
        UNDO = "undo"
    user_key = models.CharField(max_length=256, db_index=True)  # lower case
    sam = models.CharField(max_length=256)  # exact sAMAccountName, for directory calls
    kind = models.CharField(max_length=10, choices=Kind.choices)
    token_hash = models.CharField(max_length=64, unique=True)
    old_address = models.CharField(max_length=254, blank=True)
    new_address = models.CharField(max_length=254)
    created_at = models.DateTimeField()
    expires_at = models.DateTimeField()
    used_at = models.DateTimeField(null=True, blank=True)
@@ -1,4 +1,4 @@
-"""Password change for the signed-in user (FR-5, FR-6, FR-7; AD-3, AD-4, AD-8).
+"""Password change, profile details and e-mail change (FR-5..FR-11; AD-3, AD-4, AD-8, AD-9).
 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
@@ -8,9 +8,11 @@ stored or put into an exception.
 from __future__ import annotations
 import hashlib
 import logging
 import math
-from dataclasses import dataclass
+import secrets
 from dataclasses import dataclass, replace
 from datetime import timedelta
 from zoneinfo import ZoneInfo
@@ -22,11 +24,21 @@ 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, PasswordPolicy, PasswordRejected
+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
 from .models import PendingEmailChange
 logger = logging.getLogger(__name__)
 ENDPOINT = "password-change"
@@ -309,3 +321,494 @@ def _queue_mail(account) -> bool:
        {"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)
 # -- e-mail change (FR-10, FR-11) -------------------------------------------------------------
 #
 # Request: validate locally (the form), throttle.check, verify_password, then the pending row,
 # the queued mail and the audit entry in one transaction. Confirm and undo: the token claim,
 # the directory write, the other rows, the queued mail and the audit entry in one transaction.
 # Addresses, tokens and passwords never go to a log line or an audit entry.
 REQUEST_ENDPOINT = "email-change-request"  # five requests per hour per user
 REQUEST_LIMIT = 5
 REQUEST_WINDOW_SECONDS = 60 * 60
 CONFIRM_VALID_FOR = timedelta(hours=1)
 UNDO_VALID_FOR = timedelta(hours=24)
 MAX_TOKEN_LENGTH = 128
 class EmailInvalid(Exception):
    """The new address was refused after the directory was read; `reason` is `same`."""
    def __init__(self, reason: str) -> None:
        self.reason = reason
        super().__init__("The new e-mail address is not usable")
 class LinkDead(Exception):
    """The token is unknown, used, expired or replaced. Never says which."""
 class ChangedAgain(Exception):
    """The address in the directory is no longer the one that was confirmed."""
@dataclass(frozen=True)
 class Confirmed:
    notice_queued: bool
@dataclass(frozen=True)
 class Undone:
    admin_notified: bool
 def new_token() -> tuple[str, str]:
    """(token for the link, hash for the database)."""
    raw = secrets.token_urlsafe(32)
    return raw, hash_token(raw)
 def hash_token(raw: str) -> str:
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()
 def find_link(token: str, kind: str) -> PendingEmailChange | None:
    """The unused, unexpired row of this kind for the token, or None. Changes nothing."""
    if not token or len(token) > MAX_TOKEN_LENGTH:
        return None
    return PendingEmailChange.objects.filter(
        token_hash=hash_token(token),
        kind=kind,
        used_at__isnull=True,
        expires_at__gt=timezone.now(),
    ).first()
 def _link(path: str, token: str) -> str:
    return f"{settings.OIDC.public_base_url}/email/{path}/{token}"
 def _display_name(account: DirectoryUser) -> str:
    name = account.display_name or " ".join(p for p in (account.given_name, account.surname) if p)
    return name or account.sam
 def _local_now():
    return timezone.now().astimezone(ZoneInfo(settings.DISPLAY_TIME_ZONE))
 def _email_failure(user: SessionUser, ip: str, action: str, reason: str, **detail) -> None:
    audit.record_failure(user.key, action, ip=ip, detail={"reason": reason, **detail})
 def request_email_change(user: SessionUser, new_address: str, current: str, *, ip: str) -> None:
    """Ask for a confirmation link at `new_address`. The directory stays unchanged.
    `new_address` has passed the form's checks. Raises Throttled, WrongCurrentPassword,
    EmailInvalid or DirectoryUnavailable. Must be called outside an open transaction.
    """
    action = "email.change_failed"
    try:
        throttle.check(ENDPOINT, user=user.key, ip=ip or None)
        throttle.check(REQUEST_ENDPOINT, user=user.key)
    except throttle.Throttled:
        audit.record_failure(user.key, "email.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),
            )
            _email_failure(user, ip, action, "wrong_current", **_threshold_note(policy))
            raise WrongCurrentPassword(policy)
        account = directory.find_user(user.sam)
        if account is None:
            raise DirectoryUnavailable("The account could not be read")
    except DirectoryUnavailable:
        _email_failure(user, ip, action, "unavailable")
        raise
    throttle.reset(ENDPOINT, user=user.key)  # the directory forgets failures after a good bind
    if new_address.casefold() == (account.mail or "").strip().casefold():
        raise EmailInvalid("same")
    raw, digest = new_token()
    now = timezone.now()
    try:
        with transaction.atomic():
            PendingEmailChange.objects.filter(
                user_key=user.key, kind=PendingEmailChange.Kind.CONFIRM, used_at__isnull=True
            ).delete()  # the newest request replaces the earlier one
            PendingEmailChange.objects.create(
                user_key=user.key,
                sam=user.sam,
                kind=PendingEmailChange.Kind.CONFIRM,
                token_hash=digest,
                old_address=(account.mail or "").strip(),
                new_address=new_address,
                created_at=now,
                expires_at=now + CONFIRM_VALID_FOR,
            )
            mailer.enqueue(
                "email_change_confirm",
                new_address,
                {"name": _display_name(account), "link": _link("confirm", raw)},
            )
            audit.record(user.key, "email.change_requested", ip=ip)
    except Exception:
        _email_failure(user, ip, action, "error")
        raise
    throttle.hit(
        REQUEST_ENDPOINT,
        user=user.key,
        max_failures=REQUEST_LIMIT,
        pause_seconds=REQUEST_WINDOW_SECONDS,
    )
 def _unclear(user: SessionUser, ip: str, action: str, what: str, step: str, exc: Exception):
    """The directory changed but the portal's part rolled back (AD-4): log, never throw."""
    logger.error(
        "E-mail %s in the directory but the portal did not record it: "
        "user=%s ip=%s time=%s step=%s error=%s",
        what,
        user.key,
        ip,
        timezone.now().isoformat(),
        step,
        type(exc).__name__,
    )
    try:
        _email_failure(user, ip, action, "unclear", step=step)
    except Exception:
        logger.error("The failure entry for %s could not be written", user.key)
 def confirm_email_change(token: str, *, ip: str) -> Confirmed:
    """Write the new address to the directory and mail the old one an undo link.
    Raises LinkDead, DirectoryUnavailable or OutcomeUnknown. Must be called outside an open
    transaction.
    """
    action = "email.change_failed"
    row = find_link(token, PendingEmailChange.Kind.CONFIRM)
    if row is None:
        raise LinkDead
    user = SessionUser(key=row.user_key, sam=row.sam)
    directory = get_directory()
    try:
        account = directory.find_user(row.sam)
    except DirectoryUnavailable:
        _email_failure(user, ip, action, "unavailable", stage="confirm")
        raise
    if account is None:
        raise LinkDead
    previous = (account.mail or "").strip()
    notice = mailer.is_valid_address(previous)
    changed_in_directory = False
    step = "claim"
    try:
        with transaction.atomic():
            claimed = PendingEmailChange.objects.filter(pk=row.pk, used_at__isnull=True).update(
                used_at=timezone.now()
            )
            if not claimed:  # a parallel request used it first
                raise LinkDead
            step = "directory"
            directory.update_attributes(row.sam, {"mail": row.new_address})
            changed_in_directory = True
            if notice:
                step = "queue_notice"
                raw, digest = new_token()
                now = timezone.now()
                PendingEmailChange.objects.create(
                    user_key=row.user_key,
                    sam=row.sam,
                    kind=PendingEmailChange.Kind.UNDO,
                    token_hash=digest,
                    old_address=previous,
                    new_address=row.new_address,
                    created_at=now,
                    expires_at=now + UNDO_VALID_FOR,
                )
                mailer.enqueue(
                    "email_changed_notice",
                    previous,
                    {
                        "name": _display_name(account),
                        "new_address": row.new_address,
                        "link": _link("undo", raw),
                        "changed_at": _local_now(),
                        "zone_name": settings.DISPLAY_TIME_ZONE,
                    },
                )
            step = "audit"
            audit.record(
                user.key,
                "email.changed",
                ip=ip,
                detail={"notice": "queued" if notice else "skipped"},
            )
    except LinkDead:
        raise
    except DirectoryUnavailable:
        _email_failure(user, ip, action, "unavailable", stage="confirm")
        raise
    except (ValidationFailed, Forbidden) as exc:
        logger.error(
            "The directory refused an e-mail write: user=%s ip=%s error=%s",
            user.key,
            ip,
            type(exc).__name__,
        )
        _email_failure(user, ip, action, "refused", stage="confirm")
        raise DirectoryUnavailable("The directory refused the write") from exc
    except Exception as exc:
        if changed_in_directory:
            _unclear(user, ip, action, "address changed", step, exc)
            raise OutcomeUnknown("The address was changed but not recorded") from exc
        _email_failure(user, ip, action, "error", stage="confirm")
        raise
    return Confirmed(notice_queued=notice)
 def undo_email_change(token: str, *, ip: str) -> Undone:
    """Restore the old address, end every session of the user and tell the Admin.
    Raises LinkDead, ChangedAgain, DirectoryUnavailable or OutcomeUnknown. Must be called
    outside an open transaction.
    """
    action = "email.undo_failed"
    row = find_link(token, PendingEmailChange.Kind.UNDO)
    if row is None:
        raise LinkDead
    user = SessionUser(key=row.user_key, sam=row.sam)
    directory = get_directory()
    try:
        account = directory.find_user(row.sam)
    except DirectoryUnavailable:
        _email_failure(user, ip, action, "unavailable")
        raise
    if account is None:
        raise LinkDead
    if (account.mail or "").strip().casefold() != row.new_address.casefold():
        raise ChangedAgain
    admin = settings.ADMIN_NOTIFY_EMAIL
    changed_in_directory = False
    step = "claim"
    try:
        with transaction.atomic():
            claimed = PendingEmailChange.objects.filter(pk=row.pk, used_at__isnull=True).update(
                used_at=timezone.now()
            )
            if not claimed:
                raise LinkDead
            step = "directory"
            directory.update_attributes(row.sam, {"mail": row.old_address})
            changed_in_directory = True
            step = "end_sessions"  # after the write, so a rolled-back write ends nothing
            ended = end_sessions(user)
            # A request made meanwhile (by whoever had the session) must not outlive the undo.
            PendingEmailChange.objects.filter(
                user_key=row.user_key, kind=PendingEmailChange.Kind.CONFIRM, used_at__isnull=True
            ).delete()
            if admin:
                step = "queue_admin_notice"
                mailer.enqueue(
                    "email_undone_admin",
                    admin,
                    {
                        "username": row.sam,
                        "undone_at": _local_now(),
                        "zone_name": settings.DISPLAY_TIME_ZONE,
                    },
                )
            step = "audit"
            audit.record(
                user.key,
                "email.undone",
                ip=ip,
                detail={
                    "sessions_ended": ended,
                    "admin_notice": "queued" if admin else "skipped",
                },
            )
    except LinkDead:
        raise
    except DirectoryUnavailable:
        _email_failure(user, ip, action, "unavailable")
        raise
    except (ValidationFailed, Forbidden) as exc:
        logger.error(
            "The directory refused an e-mail write: user=%s ip=%s error=%s",
            user.key,
            ip,
            type(exc).__name__,
        )
        _email_failure(user, ip, action, "refused")
        raise DirectoryUnavailable("The directory refused the write") from exc
    except Exception as exc:
        if changed_in_directory:
            try:
                end_sessions(user)
            except Exception:
                logger.error("The sessions of %s could not be ended", user.key)
            _unclear(user, ip, action, "address restored", step, exc)
            raise OutcomeUnknown("The address was restored but not recorded") from exc
        _email_failure(user, ip, action, "error")
        raise
    if not admin:
        logger.error(
            "ADMIN_NOTIFY_EMAIL is not set: the Admin was not told that user=%s undid an "
            "e-mail change",
            user.key,
        )
    return Undone(admin_notified=bool(admin))
@@ -3,6 +3,10 @@ from django.urls import path
 from . import views
 urlpatterns = [
    path("profile", views.profile, name="profile"),
    path("password", views.password, name="password"),
    path("password/done", views.password_done, name="password-done"),
    path("email", views.email, name="email"),
    path("email/confirm/<str:token>", views.email_confirm, name="email-confirm"),
    path("email/undo/<str:token>", views.email_undo, name="email-undo"),
 ]
@@ -1,4 +1,4 @@
-"""The password page (FR-5..FR-7). No business logic here: see services.py (AD-17)."""
+"""The password, profile and e-mail pages (FR-5..FR-11). No business logic here (AD-17)."""
 from __future__ import annotations
@@ -6,16 +6,19 @@ import logging
 from django.shortcuts import redirect, render
 from django.utils.translation import gettext as _
 from django.views.decorators.cache import never_cache
 from django.views.decorators.debug import sensitive_post_parameters
 from django.views.decorators.http import require_http_methods
 from portal.accounts.decorators import login_required
 from portal.accounts.identity import remember
 from portal.accounts.throttle import Throttled
 from portal.adapters.directory import DirectoryUnavailable
 from portal.adapters.registry import get_directory
 from . import services
-from .forms import PasswordChangeForm
+from .forms import DETAIL_FIELDS, DetailsForm, EmailChangeForm, PasswordChangeForm
 from .models import PendingEmailChange
 logger = logging.getLogger(__name__)
@@ -107,3 +110,174 @@ def password_done(request):
    if flag is None:  # opened directly: nothing to report
        return redirect("password")
    return render(request, "profile/password_done.html", {"mail": flag["mail"]})
 # -- profile details (FR-9) -------------------------------------------------------------------
 NOTICE_FLAG = "profile_notice"  # "saved" or "unchanged", shown once after the redirect
 def _details_page(request, form, details, *, status=200, notice=None):
    context = {
        "form": form,
        "details": details,
        "notice": notice,
        "fields": [
            {"field": form[name], "label": label, "max_length": max_length}
            for name, (label, max_length, _required) in DETAIL_FIELDS.items()
        ],
    }
    return render(request, "profile/details.html", context, status=status)
@login_required
@require_http_methods(["GET", "POST"])
 def profile(request):
    user = request.portal_user.user  # the session user only: nothing is read from the request
    if request.method == "GET":
        try:
            details = services.load_details(user)
        except DirectoryUnavailable:
            return _unavailable(request)
        notice = request.session.pop(NOTICE_FLAG, None)
        return _details_page(request, DetailsForm(initial=details.values), details, notice=notice)
    form = DetailsForm(request.POST)
    form.is_valid()  # trims; the checks need the directory value and run in the service
    try:
        result = services.update_details(user, form.cleaned_data, ip=request.client_ip)
    except services.DetailsInvalid as exc:
        form.apply_errors(exc.errors)
        return _details_page(request, form, exc.details, status=400)
    except services.DetailsNotAccepted as exc:
        form.add_error(None, _("A value was not accepted. Please check your entries."))
        return _details_page(request, form, exc.details, status=400)
    except DirectoryUnavailable:
        return _unavailable(request)
    except services.OutcomeUnknown:
        return render(request, "profile/details_unknown.html", status=500)
    if result.changed:
        remember(request, result.account)  # this session shows the new name at once
    request.session[NOTICE_FLAG] = "saved" if result.changed else "unchanged"
    return redirect("profile")
 # -- e-mail change (FR-10, FR-11) -------------------------------------------------------------
 EMAIL_SENT_FLAG = "email_change_sent"
 def _email_page(request, form, current, *, status=200, throttled=False):
    context = {"form": form, "current": current, "throttled": throttled}
    return render(request, "profile/email.html", context, status=status)
@login_required
@sensitive_post_parameters("current_password")
@require_http_methods(["GET", "POST"])
 def email(request):
    user = request.portal_user.user
    if request.method == "GET":
        try:
            details = services.load_details(user)
        except DirectoryUnavailable:
            return _unavailable(request)
        if request.session.pop(EMAIL_SENT_FLAG, None):
            return render(request, "profile/email_sent.html", {"current": details.mail})
        return _email_page(request, EmailChangeForm(), details.mail)
    form = EmailChangeForm(request.POST)
    if not form.is_valid():  # no directory call before the throttle check (AD-7)
        return _email_page(request, form, None, status=400)
    try:
        services.request_email_change(
            user,
            form.cleaned_data["new_email"],
            form.cleaned_data["current_password"],
            ip=request.client_ip,
        )
    except Throttled:
        return _email_page(request, EmailChangeForm(), None, status=429, throttled=True)
    except services.WrongCurrentPassword:
        form.add_error("current_password", _("The current password is not correct."))
        return _email_page(request, form, _current_address(user), status=400)
    except services.EmailInvalid:
        form.add_error("new_email", _("This is already your e-mail address."))
        return _email_page(request, form, _current_address(user), status=400)
    except DirectoryUnavailable:
        return _unavailable(request)
    request.session[EMAIL_SENT_FLAG] = True
    return redirect("email")
 def _current_address(user) -> str | None:
    try:
        return services.load_details(user).mail
    except DirectoryUnavailable:
        return None
 def _dead_link(request):
    return render(request, "profile/email_link_dead.html", status=410)
@never_cache
@sensitive_post_parameters()
@require_http_methods(["GET", "POST"])
 def email_confirm(request, token):
    """No sign-in needed: the link proves the mailbox. GET only shows (AD-9)."""
    if request.method == "GET":
        row = services.find_link(token, PendingEmailChange.Kind.CONFIRM)
        if row is None:
            return _dead_link(request)
        return render(request, "profile/email_confirm.html", {"change": row})
    try:
        result = services.confirm_email_change(token, ip=request.client_ip)
    except services.LinkDead:
        return _dead_link(request)
    except DirectoryUnavailable:
        return _unavailable(request)
    except services.OutcomeUnknown:
        return render(request, "profile/email_unknown.html", status=500)
    return render(request, "profile/email_confirmed.html", {"notice": result.notice_queued})
@never_cache
@sensitive_post_parameters()
@require_http_methods(["GET", "POST"])
 def email_undo(request, token):
    """No sign-in needed: the link proves the mailbox. GET only shows (AD-9)."""
    if request.method == "GET":
        row = services.find_link(token, PendingEmailChange.Kind.UNDO)
        if row is None:
            return _dead_link(request)
        return render(request, "profile/email_undo.html", {"change": row})
    try:
        services.undo_email_change(token, ip=request.client_ip)
    except services.LinkDead:
        return _dead_link(request)
    except services.ChangedAgain:
        return render(request, "profile/email_changed_again.html", status=409)
    except DirectoryUnavailable:
        return _unavailable(request)
    except services.OutcomeUnknown:
        _drop_own_session(request)
        return render(request, "profile/email_unknown.html", status=500)
    _drop_own_session(request)
    return render(request, "profile/email_undone.html")
 def _drop_own_session(request):
    """Undo ends every session of the user, also the one that opened the link.
    Its row is gone, so saving it at the end of the request would fail with a 400; flushing it
    clears the cookie instead.
    """
    session = getattr(request, "session", None)
    if session is not None:
        session.flush()
@@ -22,7 +22,15 @@ class Group:
 MENU = (
-    Group(None, None, (Entry(_("Home"), "home"), Entry(_("Change password"), "password"))),
+    Group(
        None,
        None,
        (
            Entry(_("Home"), "home"),
            Entry(_("Profile"), "profile"),
            Entry(_("Change password"), "password"),
        ),
    ),
    Group(
        _("Publishing"),
        "publisher",
@@ -578,3 +578,17 @@ main button.primary:disabled {
    animation-duration: 3s;
  }
 }
 /* Read-only label/value rows on the profile page; they stack on narrow screens. */
 .details {
  display: grid;
  grid-template-columns: max-content 1fr;
  gap: 0.5rem 1.5rem;
  margin-block: 0 1rem;
 }
 .details dt { font-weight: 600; }
 .details dd { margin: 0; overflow-wrap: anywhere; }
@media (max-width: 720px) {
  .details { grid-template-columns: 1fr; gap: 0; }
  .details dd { margin-bottom: 0.75rem; }
 }
@@ -0,0 +1,9 @@
 {% load i18n %}{% autoescape off %}{% blocktrans with name=name %}Hello {{ name }},{% endblocktrans %}
 {% trans "this e-mail address was entered as the new e-mail address of your portal account. To confirm it, open this link and press the button. The link is valid for one hour:" %}
 {{ link }}
 {% trans "Your address changes only after you confirm. If you did not ask for this, ignore this e-mail. Nothing will change." %}
 {% trans "This e-mail never contains your password." %}{% endautoescape %}
@@ -0,0 +1 @@
 {% load i18n %}{% autoescape off %}{% trans "Please confirm your new e-mail address" %}{% endautoescape %}
@@ -0,0 +1,11 @@
 {% load i18n tz %}{% autoescape off %}{% blocktrans with name=name %}Hello {{ name }},{% endblocktrans %}
 {% timezone zone_name %}{% blocktrans with when=changed_at|date:"j. F Y, H:i" zone=changed_at|date:"T" address=new_address %}the e-mail address of your portal account was changed on {{ when }} ({{ zone }}) to {{ address }}.{% endblocktrans %}{% endtimezone %}
 {% trans "If you did this, you do not need to do anything. If you did not, open this link to restore this address. The link is valid for 24 hours:" %}
 {{ link }}
 {% trans "Undoing signs you out on all devices and informs the administrator." %}
 {% trans "This e-mail never contains your password." %}{% endautoescape %}
@@ -0,0 +1 @@
 {% load i18n %}{% autoescape off %}{% trans "Your e-mail address was changed" %}{% endautoescape %}
@@ -0,0 +1,5 @@
 {% load i18n tz %}{% autoescape off %}{% trans "Hello," %}
 {% timezone zone_name %}{% blocktrans with when=undone_at|date:"j. F Y, H:i" zone=undone_at|date:"T" username=username %}the user {{ username }} undid an e-mail change on {{ when }} ({{ zone }}). All sessions of this user were ended.{% endblocktrans %}{% endtimezone %}
 {% trans "The user may have lost control of the account. Please check it." %}{% endautoescape %}
@@ -0,0 +1 @@
 {% load i18n %}{% autoescape off %}{% trans "An e-mail change was undone" %}{% endautoescape %}
@@ -0,0 +1,7 @@
 {% load i18n %}
 <div class="field{% if field.errors %} has-error{% endif %}">
  <label for="{{ field.id_for_label }}">{{ label }}</label>
  <input type="{{ input_type|default:'text' }}" name="{{ field.html_name }}" id="{{ field.id_for_label }}" value="{{ field.value|default_if_none:'' }}" autocomplete="{{ autocomplete }}" maxlength="{{ max_length }}"{% if required %} required{% endif %}{% if field.errors %} aria-invalid="true"{% endif %} aria-describedby="{{ field.id_for_label }}-hint{% if field.errors %} {{ field.id_for_label }}-error{% endif %}">
  {% if hint %}<p class="hint" id="{{ field.id_for_label }}-hint">{{ hint }}</p>{% else %}<span id="{{ field.id_for_label }}-hint"></span>{% endif %}
  {% if field.errors %}<div id="{{ field.id_for_label }}-error">{% for error in field.errors %}<p class="error">{{ error }}</p>{% endfor %}</div>{% endif %}
 </div>
@@ -0,0 +1,54 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "Profile" %}{% endblock %}
 {% block content %}
 <h1>{% trans "Profile" %}</h1>
 <p class="lead">{% trans "These details are stored in your account. Changes apply to every service that uses it." %}</p>
 {% if notice == "saved" %}
 <div class="notice success" role="status">
  <p>{% trans "Your changes were saved." %}</p>
 </div>
 {% elif notice == "unchanged" %}
 <div class="notice" role="status">
  <p>{% trans "Nothing to change." %}</p>
 </div>
 {% endif %}
 {% if form.errors %}
 <div class="error-summary" role="alert" tabindex="-1" data-focus-on-load>
  <p><strong>{% trans "Your changes were not saved." %}</strong></p>
  <ul>
    {% for error in form.non_field_errors %}<li>{{ error }}</li>{% endfor %}
    {% for field in form %}{% for error in field.errors %}
    <li><a href="#{{ field.id_for_label }}">{{ error }}</a></li>
    {% endfor %}{% endfor %}
  </ul>
 </div>
 {% endif %}
 {% if details %}
 <dl class="card details">
  <dt>{% trans "User name" %}</dt>
  <dd>{{ details.username }}</dd>
  <dt>{% trans "E-mail address" %}</dt>
  <dd>{% if details.mail %}{{ details.mail }}{% else %}&ndash;{% endif %} <a href="{% url 'email' %}">{% trans "Change e-mail address" %}</a></dd>
 </dl>
 {% endif %}
 <form method="post" action="{% url 'profile' %}" class="card" data-busy-on-submit novalidate>
  {% csrf_token %}
  {% trans "For example +49 171 123456. A number that starts with 0 is read as German." as phone_hint %}
  {% for item in fields %}
    {% if item.field.name == "first_name" %}
      {% include "profile/_text_field.html" with field=item.field label=item.label max_length=item.max_length autocomplete="given-name" required=True %}
    {% elif item.field.name == "last_name" %}
      {% include "profile/_text_field.html" with field=item.field label=item.label max_length=item.max_length autocomplete="family-name" required=True %}
    {% elif item.field.name == "display_name" %}
      {% include "profile/_text_field.html" with field=item.field label=item.label max_length=item.max_length autocomplete="name" required=True %}
    {% else %}
      {% include "profile/_text_field.html" with field=item.field label=item.label max_length=item.max_length autocomplete="tel" input_type="tel" hint=phone_hint %}
    {% endif %}
  {% endfor %}
  <p class="actions">
    <button type="submit" class="primary"><span class="spinner" aria-hidden="true"></span>{% trans "Save changes" %}</button>
    <a class="button" href="{% url 'home' %}">{% trans "Cancel" %}</a>
  </p>
 </form>
 {% endblock %}
@@ -0,0 +1,10 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "Result unclear" %}{% endblock %}
 {% block content %}
 <h1>{% trans "Result unclear" %}</h1>
 <div class="notice" role="alert">
  <p>{% trans "Your details may have been changed, but the portal could not finish. Please open your profile and check them. If they are wrong, contact the administrator." %}</p>
 </div>
 <p><a class="button" href="{% url 'profile' %}">{% trans "Back to the profile" %}</a></p>
 {% endblock %}
@@ -0,0 +1,41 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "Change e-mail address" %}{% endblock %}
 {% block content %}
 <h1>{% trans "Change e-mail address" %}</h1>
 <p class="lead">{% trans "Announcements and security notices are sent to this address. We send a link to the new address. Your address changes only after you open the link and confirm." %}</p>
 {% if throttled %}
 <div class="notice" role="alert" tabindex="-1" data-focus-on-load>
  <p>{% trans "Too many attempts. Please wait a few minutes." %}</p>
 </div>
 <p><a class="button" href="{% url 'home' %}">{% trans "Back to the start page" %}</a></p>
 {% else %}
 {% if form.errors %}
 <div class="error-summary" role="alert" tabindex="-1" data-focus-on-load>
  <p><strong>{% trans "No link was sent." %}</strong></p>
  <ul>
    {% for field in form %}{% for error in field.errors %}
    <li><a href="#{{ field.id_for_label }}">{{ error }}</a></li>
    {% endfor %}{% endfor %}
  </ul>
 </div>
 {% endif %}
 {% if current is not None %}
 <dl class="card details">
  <dt>{% trans "Current e-mail address" %}</dt>
  <dd>{% if current %}{{ current }}{% else %}&ndash;{% endif %}</dd>
 </dl>
 {% endif %}
 <form method="post" action="{% url 'email' %}" class="card" data-busy-on-submit novalidate>
  {% csrf_token %}
  {% trans "New e-mail address" as label %}
  {% include "profile/_text_field.html" with field=form.new_email label=label max_length=254 autocomplete="email" input_type="email" required=True %}
  {% trans "Current password" as label %}{% trans "We check it before anything is sent." as hint %}
  {% include "profile/_field.html" with field=form.current_password label=label hint=hint autocomplete="current-password" %}
  <p class="actions">
    <button type="submit" class="primary"><span class="spinner" aria-hidden="true"></span>{% trans "Send confirmation link" %}</button>
    <a class="button" href="{% url 'profile' %}">{% trans "Cancel" %}</a>
  </p>
 </form>
 {% endif %}
 {% endblock %}
@@ -0,0 +1,10 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "Address changed again" %}{% endblock %}
 {% block content %}
 <h1>{% trans "Address changed again" %}</h1>
 <div class="notice" role="alert">
  <p>{% trans "The e-mail address was changed again since then. Nothing was changed now. If this was not you, contact the administrator." %}</p>
 </div>
 <p><a class="button" href="{% url 'home' %}">{% trans "Back to the start page" %}</a></p>
 {% endblock %}
@@ -0,0 +1,20 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "Confirm new e-mail address" %}{% endblock %}
 {% block content %}
 <h1>{% trans "Confirm new e-mail address" %}</h1>
 <p class="lead">{% trans "The e-mail address of your account will change. Nothing changes until you press the button." %}</p>
 <dl class="card details">
  <dt>{% trans "Current e-mail address" %}</dt>
  <dd>{% if change.old_address %}{{ change.old_address }}{% else %}&ndash;{% endif %}</dd>
  <dt>{% trans "New e-mail address" %}</dt>
  <dd>{{ change.new_address }}</dd>
 </dl>
 {% if change.old_address %}<p>{% trans "A notice with a way to undo the change is sent to the current address." %}</p>{% endif %}
 <form method="post" action="{{ request.path }}" class="card" data-busy-on-submit>
  {% csrf_token %}
  <p class="actions">
    <button type="submit" class="primary"><span class="spinner" aria-hidden="true"></span>{% trans "Confirm change" %}</button>
  </p>
 </form>
 {% endblock %}
@@ -0,0 +1,11 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "E-mail address changed" %}{% endblock %}
 {% block content %}
 <h1>{% trans "E-mail address changed" %}</h1>
 <div class="notice success" role="status">
  <p>{% trans "Your e-mail address was changed." %}</p>
 </div>
 {% if notice %}<p>{% trans "A notice with a way to undo the change was sent to your previous address." %}</p>{% endif %}
 <p><a class="button" href="{% url 'home' %}">{% trans "Back to the start page" %}</a></p>
 {% endblock %}
@@ -0,0 +1,10 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "Link not valid" %}{% endblock %}
 {% block content %}
 <h1>{% trans "Link not valid" %}</h1>
 <div class="notice" role="alert">
  <p>{% trans "This link is not valid any more." %}</p>
 </div>
 <p><a class="button" href="{% url 'home' %}">{% trans "Back to the start page" %}</a></p>
 {% endblock %}
@@ -0,0 +1,11 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "Confirmation link sent" %}{% endblock %}
 {% block content %}
 <h1>{% trans "Confirmation link sent" %}</h1>
 <div class="notice success" role="status">
  <p>{% trans "We sent a link to your new e-mail address. It is valid for one hour." %}</p>
 </div>
 <p>{% trans "Your address has not changed yet. It changes after you open the link and confirm." %}</p>
 <p><a class="button" href="{% url 'profile' %}">{% trans "Back to the profile" %}</a></p>
 {% endblock %}
@@ -0,0 +1,20 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "Undo e-mail change" %}{% endblock %}
 {% block content %}
 <h1>{% trans "Undo e-mail change" %}</h1>
 <p class="lead">{% trans "The e-mail address of your account was changed. If you did not do this, undo the change now." %}</p>
 <dl class="card details">
  <dt>{% trans "Previous e-mail address" %}</dt>
  <dd>{{ change.old_address }}</dd>
  <dt>{% trans "Current e-mail address" %}</dt>
  <dd>{{ change.new_address }}</dd>
 </dl>
 <p>{% trans "Undoing restores the previous address, signs you out on all devices and informs the administrator." %}</p>
 <form method="post" action="{{ request.path }}" class="card" data-busy-on-submit>
  {% csrf_token %}
  <p class="actions">
    <button type="submit" class="primary"><span class="spinner" aria-hidden="true"></span>{% trans "Undo change" %}</button>
  </p>
 </form>
 {% endblock %}
@@ -0,0 +1,11 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "E-mail change undone" %}{% endblock %}
 {% block content %}
 <h1>{% trans "E-mail change undone" %}</h1>
 <div class="notice success" role="status">
  <p>{% trans "Your previous e-mail address is back. You were signed out on all devices." %}</p>
 </div>
 <p><strong>{% trans "Someone else may have had access to your account. Please change your password now." %}</strong></p>
 <p><a class="button primary" href="{% url 'password' %}">{% trans "Change password" %}</a></p>
 {% endblock %}
@@ -0,0 +1,10 @@
 {% extends "base.html" %}
 {% load i18n %}
 {% block title %}{% trans "Result unclear" %}{% endblock %}
 {% block content %}
 <h1>{% trans "Result unclear" %}</h1>
 <div class="notice" role="alert">
  <p>{% trans "Your e-mail address may have been changed, but the portal could not finish. Please check your profile or contact the administrator." %}</p>
 </div>
 <p><a class="button" href="{% url 'home' %}">{% trans "Back to the start page" %}</a></p>
 {% endblock %}
@@ -178,3 +178,13 @@ def test_the_mail_settings_are_required(valid_env, name):
    del valid_env[name]
    with pytest.raises(ConfigError, match=name):
        load_config(valid_env)
 def test_the_admin_notice_address_is_optional_and_checked(valid_env):
    assert load_config(valid_env).admin_notify_email == ""
    assert load_config({**valid_env, "ADMIN_NOTIFY_EMAIL": "  "}).admin_notify_email == ""
    good = {**valid_env, "ADMIN_NOTIFY_EMAIL": "admin@example.test"}
    assert load_config(good).admin_notify_email == "admin@example.test"
    with pytest.raises(ConfigError, match="ADMIN_NOTIFY_EMAIL") as info:
        load_config({**valid_env, "ADMIN_NOTIFY_EMAIL": "not an address"})
    assert "not an address" not in str(info.value)
@@ -0,0 +1,720 @@
 """E-mail change: request, confirmation link, undo link, limits, sessions, audit and logs."""
 import logging
 import re
 from datetime import timedelta
 import pytest
 from django.core.management import call_command
 from django.test import Client
 from django.utils import timezone
 from portal.accounts.models import ThrottleCounter, UserSession
 from portal.adapters.directory import DirectoryUnavailable, ValidationFailed
 from portal.audit.models import AuditEntry
 from portal.mailer.models import OutboxMail
 from portal.profile import services
 from portal.profile.models import PendingEmailChange
 from portal.tests.helpers import sign_in
 pytestmark = pytest.mark.django_db(transaction=True)
 CURRENT = "Correct-Horse-9"
 WRONG = "Not-The-Password-1"
 OLD = "alice@example.test"
 NEW = "alice.new@example.test"
 ADMIN = "admin@example.test"
 LINK = re.compile(r"https://portal\.example\.test(/email/(?:confirm|undo)/[\w-]+)")
 DEAD = "This link is not valid any more."
@pytest.fixture(autouse=True)
 def admin_address(settings):
    settings.ADMIN_NOTIFY_EMAIL = ADMIN
@pytest.fixture
 def people(adapters):
    d = adapters.directory
    d.add_user("alice", password=CURRENT, display_name="Alice Anders", mail=OLD)
    d.add_user("bob", display_name="Bob Brown", mail="bob@example.test")
    d.add_user("nomail", password=CURRENT, display_name="No Mail")
    return d
 class Spy:
    def __init__(self, directory):
        self.calls = []
        for name in ("password_policy", "verify_password", "find_user", "update_attributes"):
            self._wrap(directory, name)
    def _wrap(self, directory, name):
        original = getattr(directory, name)
        def wrapper(*args, **kwargs):
            self.calls.append(name)
            return original(*args, **kwargs)
        setattr(directory, name, wrapper)
 def signed_in(adapters, name="alice"):
    client, _ = sign_in(adapters, name)
    return client
 def ask(client, new=NEW, password=CURRENT):
    return client.post("/email", {"new_email": new, "current_password": password})
 def mails(to=None):
    rows = OutboxMail.objects.all()
    return list(rows.filter(recipient=to) if to else rows)
 def link_in(mail):
    return LINK.search(mail.body_text).group(1)
 def request_link(adapters, name="alice", new=NEW):
    """Sign in, ask for a change and return the confirmation path."""
    client = signed_in(adapters, name)
    assert ask(client, new).status_code == 302
    return link_in(mails(new)[-1])
 def confirmed_undo_path(adapters):
    path = request_link(adapters)
    Client().post(path)
    return link_in(mails(OLD)[-1])
 def actions():
    names = AuditEntry.objects.values_list("action", flat=True)
    return [name for name in names if name.startswith("email.")]
 def mail_of(adapters, name="alice"):
    return adapters.directory.find_user(name).mail
 # -- the page -----------------------------------------
 def test_the_page_shows_the_current_address_and_is_reached_from_the_profile(people, adapters):
    client = signed_in(adapters)
    profile = client.get("/profile").content.decode()
    assert 'href="/email"' in profile and "Change e-mail address" in profile
    html = client.get("/email").content.decode()
    assert OLD in html and 'name="new_email"' in html and 'name="current_password"' in html
    assert 'autocomplete="current-password"' in html and "<form" in html
 def test_the_page_is_unavailable_when_the_directory_is_down(people, adapters):
    client = signed_in(adapters)
    adapters.directory.unavailable = True
    assert client.get("/email").status_code == 503
 def test_the_page_and_the_mail_speak_german(people, adapters):
    client = signed_in(adapters)
    client.cookies["pm_lang"] = "de"
    assert "E-Mail-Adresse ändern" in client.get("/email").content.decode()
    assert ask(client).status_code == 302
    mail = mails(NEW)[0]
    assert mail.subject == "Bitte bestätigen Sie Ihre neue E-Mail-Adresse"
    assert "Guten Tag Alice Anders," in mail.body_text and "eine Stunde" in mail.body_text
 # -- the request -----------------------------------------
 def test_a_request_mails_a_link_to_the_new_address_and_leaves_the_directory_alone(people, adapters):
    client = signed_in(adapters)
    response = ask(client)
    assert response.status_code == 302 and response["Location"] == "/email"
    assert mail_of(adapters) == OLD
    [mail] = mails()
    assert mail.recipient == NEW and mail.template == "email_change_confirm"
    assert LINK.search(mail.body_text) and "/email/confirm/" in mail.body_text
    assert "password" not in mail.body_text.lower().replace("this e-mail never", "")
    sent = client.get("/email").content.decode()
    assert "We sent a link" in sent and 'name="new_email"' not in sent
    assert 'name="new_email"' in client.get("/email").content.decode()  # shown once
    assert actions() == ["email.change_requested"]
 def test_only_the_hash_of_a_token_is_stored(people, adapters):
    path = request_link(adapters)
    token = path.rsplit("/", 1)[1]
    assert len(token) == 43  # secrets.token_urlsafe(32)
    row = PendingEmailChange.objects.get()
    assert row.token_hash == services.hash_token(token) and len(row.token_hash) == 64
    assert token not in " ".join(str(v) for v in row.__dict__.values())
    assert row.kind == "confirm" and row.user_key == "alice" and row.sam == "alice"
    assert row.expires_at - row.created_at == timedelta(hours=1)
    assert row.old_address == OLD and row.new_address == NEW and row.used_at is None
 def test_the_order_is_form_then_throttle_then_password_then_the_write(people, adapters):
    spy = Spy(adapters.directory)
    client = signed_in(adapters)
    spy.calls.clear()
    assert ask(client, new="nonsense").status_code == 400
    assert spy.calls == []  # the form is checked before any directory call
    ThrottleCounter.objects.create(
        endpoint="password-change",
        subject="user:alice",
        count=4,
        blocked_until=timezone.now() + timedelta(minutes=5),
    )
    assert ask(client).status_code == 429
    assert spy.calls == [] and mails() == [] and PendingEmailChange.objects.count() == 0
    assert "email.throttled" in actions()
 def test_a_wrong_password_is_refused_counted_and_audited(people, adapters):
    client = signed_in(adapters)
    response = ask(client, password=WRONG)
    assert response.status_code == 400
    assert "The current password is not correct." in response.content.decode()
    assert mails() == [] and PendingEmailChange.objects.count() == 0
    assert ThrottleCounter.objects.get(endpoint="password-change", subject="user:alice").count == 1
    assert AuditEntry.objects.get(action="email.change_failed").detail == {
        "reason": "wrong_current"
    }
    assert mail_of(adapters) == OLD
 def test_both_pages_share_one_counter_below_the_directory_lockout(people, adapters):
    client = signed_in(adapters)
    # The fake directory locks at 5, so the portal allows 4 wrong passwords in all.
    for _ in range(2):
        client.post(
            "/password",
            {
                "current_password": WRONG,
                "new_password": "Xx1-new-pass",
                "confirm_password": "Xx1-new-pass",
            },
        )
    assert ask(client, password=WRONG).status_code == 400
    assert ask(client, password=WRONG).status_code == 400
    spy = Spy(adapters.directory)
    response = ask(client)  # even the right password: the limit is reached
    assert response.status_code == 429
    assert "Too many attempts. Please wait a few minutes." in response.content.decode()
    assert spy.calls == [] and mails() == []
    assert ThrottleCounter.objects.get(endpoint="password-change", subject="user:alice").count == 4
    # and the password page is paused as well
    blocked = client.post(
        "/password",
        {
            "current_password": CURRENT,
            "new_password": "Xx1-new-pass",
            "confirm_password": "Xx1-new-pass",
        },
    )
    assert blocked.status_code == 429
 def test_five_requests_per_hour_are_allowed_and_the_sixth_is_not(people, adapters):
    client = signed_in(adapters)
    for number in range(5):
        assert ask(client, new=f"new{number}@example.test").status_code == 302
    spy = Spy(adapters.directory)
    assert ask(client, new="sixth@example.test").status_code == 429
    assert spy.calls == [] and len(mails()) == 5
    assert actions().count("email.throttled") == 1
    ThrottleCounter.objects.filter(endpoint="email-change-request").update(
        blocked_until=timezone.now() - timedelta(seconds=1)
    )
    assert ask(client, new="after@example.test").status_code == 302  # recovers after the pause
@pytest.mark.parametrize(
    "value",
    ["", "   ", "not-an-address", "a b@example.test", "x" * 250 + "@example.test", OLD.upper()],
 )
 def test_a_bad_new_address_is_refused_and_not_counted(people, adapters, value):
    client = signed_in(adapters)
    response = ask(client, new=value)
    assert response.status_code == 400 and "No link was sent." in response.content.decode()
    assert mails() == [] and PendingEmailChange.objects.count() == 0
    assert not ThrottleCounter.objects.exists()
 def test_the_same_address_in_another_letter_case_is_named(people, adapters):
    response = ask(signed_in(adapters), new="Alice@Example.Test")
    assert "This is already your e-mail address." in response.content.decode()
 def test_a_newer_request_replaces_the_pending_one(people, adapters):
    client = signed_in(adapters)
    ask(client, new="first@example.test")
    ask(client, new="second@example.test")
    first = link_in(mails("first@example.test")[0])
    second = link_in(mails("second@example.test")[0])
    other = Client()
    assert other.get(first).status_code == 410 and other.post(first).status_code == 410
    assert other.get(second).status_code == 200
    assert other.post(second).status_code == 200
    assert mail_of(adapters) == "second@example.test"
 def test_the_request_works_when_the_user_has_no_address_yet(people, adapters):
    client = signed_in(adapters, "nomail")
    assert (
        "&ndash;" in client.get("/email").content.decode()
        or "–" in client.get("/email").content.decode()
    )
    assert ask(client).status_code == 302
    assert len(mails(NEW)) == 1
 def test_a_request_is_unavailable_when_the_directory_is_down(people, adapters):
    client = signed_in(adapters)
    adapters.directory.unavailable = True
    assert ask(client).status_code == 503
    assert mails() == [] and PendingEmailChange.objects.count() == 0
    assert AuditEntry.objects.get(action="email.change_failed").detail["reason"] == "unavailable"
 # -- the confirmation link -----------------------------------------
 def test_opening_the_confirmation_link_changes_and_consumes_nothing(people, adapters):
    path = request_link(adapters)
    before = list(AuditEntry.objects.values_list("id", flat=True))
    visitor = Client()  # not signed in
    for _ in range(3):
        response = visitor.get(path)
        assert response.status_code == 200
        html = response.content.decode()
        assert OLD in html and NEW in html and "<form" in html and 'method="post"' in html
        assert "csrfmiddlewaretoken" in html
        assert (
            response["Cache-Control"].startswith("max-age=0")
            or "no-cache" in response["Cache-Control"]
        )
    assert visitor.head(path).status_code in (200, 405)
    assert mail_of(adapters) == OLD
    assert PendingEmailChange.objects.get().used_at is None
    assert len(mails()) == 1 and list(AuditEntry.objects.values_list("id", flat=True)) == before
 def test_confirming_writes_the_address_and_mails_the_old_one_an_undo_link(people, adapters):
    path = request_link(adapters)
    visitor = Client()
    response = visitor.post(path)
    assert (
        response.status_code == 200
        and "Your e-mail address was changed." in response.content.decode()
    )
    assert "A notice with a way to undo" in response.content.decode()
    assert mail_of(adapters) == NEW
    confirm = PendingEmailChange.objects.get(kind="confirm")
    assert confirm.used_at is not None
    undo = PendingEmailChange.objects.get(kind="undo")
    assert undo.old_address == OLD and undo.new_address == NEW
    assert undo.expires_at - undo.created_at == timedelta(hours=24)
    [notice] = mails(OLD)
    assert notice.template == "email_changed_notice"
    token = link_in(notice).rsplit("/", 1)[1]
    assert undo.token_hash == services.hash_token(token) and token != undo.token_hash
    assert NEW in notice.body_text and "24" in notice.body_text
    entry = AuditEntry.objects.get(action="email.changed")
    assert entry.actor == "alice" and entry.detail == {"notice": "queued"}
 def test_a_confirmation_works_once(people, adapters):
    path = request_link(adapters)
    visitor = Client()
    assert visitor.post(path).status_code == 200
    assert visitor.post(path).status_code == 410
    assert visitor.get(path).status_code == 410
    assert len(mails(OLD)) == 1
 def test_dead_links_all_look_alike(people, adapters):
    unknown = Client().get("/email/confirm/" + "a" * 43)
    assert unknown.status_code == 410 and DEAD in unknown.content.decode()
    expired = request_link(adapters)
    PendingEmailChange.objects.update(expires_at=timezone.now() - timedelta(seconds=1))
    used = confirmed_undo_path(adapters)  # a fresh pair of rows
    PendingEmailChange.objects.filter(kind="undo").update(used_at=timezone.now())
    pages = {
        Client().get(expired).content,
        Client().get(used).content,
        Client().post(expired).content,
        Client().post(used).content,
        unknown.content,
    }
    assert len(pages) == 1  # the page does not say which case it is
    assert Client().post(expired).status_code == 410
 def test_a_confirmation_fails_cleanly_when_the_directory_is_down(people, adapters):
    path = request_link(adapters)
    adapters.directory.unavailable = True
    response = Client().post(path)
    assert response.status_code == 503
    adapters.directory.unavailable = False
    assert PendingEmailChange.objects.get().used_at is None
    assert mails(OLD) == [] and mail_of(adapters) == OLD
    assert AuditEntry.objects.get(action="email.change_failed").detail == {
        "reason": "unavailable",
        "stage": "confirm",
    }
    assert Client().post(path).status_code == 200  # and it works once the directory is back
 def test_a_refused_write_rolls_everything_back(people, adapters, monkeypatch):
    path = request_link(adapters)
    def refuse(*args, **kwargs):
        raise ValidationFailed("no")
    monkeypatch.setattr(adapters.directory, "update_attributes", refuse)
    assert Client().post(path).status_code == 503
    assert PendingEmailChange.objects.get().used_at is None and mails(OLD) == []
    assert "email.changed" not in actions()
 def test_a_user_without_an_address_can_change_it_without_a_notice(people, adapters):
    client = signed_in(adapters, "nomail")
    ask(client)
    path = link_in(mails(NEW)[0])
    response = Client().post(path)
    assert response.status_code == 200 and "A notice with a way" not in response.content.decode()
    assert Client().get(path).status_code == 410
    assert mail_of(adapters, "nomail") == NEW
    assert PendingEmailChange.objects.filter(kind="undo").count() == 0
    assert [m.recipient for m in mails()] == [NEW]
    assert AuditEntry.objects.get(action="email.changed").detail == {"notice": "skipped"}
    page = Client().get(link_in(mails(NEW)[0]))  # nothing to show any more
    assert page.status_code == 410
 def test_the_confirmation_page_does_not_mention_a_notice_without_an_old_address(people, adapters):
    client = signed_in(adapters, "nomail")
    ask(client)
    html = Client().get(link_in(mails(NEW)[0])).content.decode()
    assert "A notice with a way" not in html and NEW in html
 # -- the undo link -----------------------------------------
 def test_opening_the_undo_link_changes_and_consumes_nothing(people, adapters):
    path = confirmed_undo_path(adapters)
    count = AuditEntry.objects.count()
    for _ in range(2):
        response = Client().get(path)
        assert response.status_code == 200
        html = response.content.decode()
        assert OLD in html and NEW in html and 'method="post"' in html
    assert mail_of(adapters) == NEW and AuditEntry.objects.count() == count
    assert PendingEmailChange.objects.get(kind="undo").used_at is None
 def test_undoing_restores_the_address_ends_the_sessions_and_tells_the_admin(people, adapters):
    path = confirmed_undo_path(adapters)
    alice_one, alice_two = signed_in(adapters), signed_in(adapters)
    bob = signed_in(adapters, "bob")
    response = Client().post(path)
    html = response.content.decode()
    assert response.status_code == 200 and "Please change your password now." in html
    assert mail_of(adapters) == OLD
    assert alice_one.get("/").status_code == 302 and alice_two.get("/").status_code == 302
    assert bob.get("/").status_code == 200  # only this user's sessions end
    assert UserSession.objects.filter(username="alice").count() == 0
    assert PendingEmailChange.objects.get(kind="undo").used_at is not None
    [admin] = mails(ADMIN)
    assert admin.template == "email_undone_admin" and "alice" in admin.body_text
    assert path.rsplit("/", 1)[1] not in admin.body_text and OLD not in admin.body_text
    entry = AuditEntry.objects.get(action="email.undone")
    assert entry.detail == {"sessions_ended": 3, "admin_notice": "queued"}
    assert Client().post(path).status_code == 410 and Client().get(path).status_code == 410
 def test_undoing_from_a_browser_that_is_still_signed_in_works(people, adapters):
    path = confirmed_undo_path(adapters)
    browser = signed_in(adapters)  # the link is opened where the user is signed in
    response = browser.post(path)
    assert (
        response.status_code == 200
        and "Please change your password now." in response.content.decode()
    )
    assert mail_of(adapters) == OLD
    assert browser.get("/").status_code == 302  # its own session ended too
    assert UserSession.objects.filter(username="alice").count() == 0
 def test_undoing_also_drops_a_request_made_in_the_meantime(people, adapters):
    path = confirmed_undo_path(adapters)
    signed_in(adapters)
    pending_client = signed_in(adapters)
    ask(pending_client, new="thief@example.test")
    stolen = link_in(mails("thief@example.test")[0])
    assert Client().post(path).status_code == 200
    assert Client().post(stolen).status_code == 410
    assert mail_of(adapters) == OLD
 def test_the_undo_is_unavailable_when_the_directory_is_down(people, adapters):
    path = confirmed_undo_path(adapters)
    other = signed_in(adapters)
    adapters.directory.unavailable = True
    assert Client().post(path).status_code == 503
    adapters.directory.unavailable = False
    assert PendingEmailChange.objects.get(kind="undo").used_at is None
    assert other.get("/").status_code == 200 and mails(ADMIN) == []
    assert "email.undone" not in actions()
 def test_an_undo_after_another_change_changes_nothing(people, adapters):
    path = confirmed_undo_path(adapters)
    adapters.directory.update_attributes("alice", {"mail": "third@example.test"})
    other = signed_in(adapters)
    response = Client().post(path)
    assert response.status_code == 409 and "changed again" in response.content.decode()
    assert mail_of(adapters) == "third@example.test"
    assert other.get("/").status_code == 200 and mails(ADMIN) == []
    assert PendingEmailChange.objects.get(kind="undo").used_at is None
 def test_an_undo_without_an_admin_address_still_succeeds_and_logs_an_error(
    people, adapters, settings, caplog
 ):
    settings.ADMIN_NOTIFY_EMAIL = ""
    path = confirmed_undo_path(adapters)
    with caplog.at_level(logging.ERROR):
        response = Client().post(path)
    assert response.status_code == 200 and mail_of(adapters) == OLD
    assert mails(ADMIN) == [] and len(mails()) == 2  # the two earlier mails only
    [record] = [r for r in caplog.records if "ADMIN_NOTIFY_EMAIL" in r.getMessage()]
    assert record.levelno == logging.ERROR and "user=alice" in record.getMessage()
    assert OLD not in caplog.text and NEW not in caplog.text
    assert AuditEntry.objects.get(action="email.undone").detail["admin_notice"] == "skipped"
 def test_an_expired_undo_link_is_dead(people, adapters):
    path = confirmed_undo_path(adapters)
    PendingEmailChange.objects.filter(kind="undo").update(
        expires_at=timezone.now() - timedelta(seconds=1)
    )
    assert Client().get(path).status_code == 410
    assert Client().post(path).status_code == 410
    assert mail_of(adapters) == NEW
 # -- guards -----------------------------------------
 def test_the_page_needs_a_sign_in(people):
    client = Client()
    assert client.get("/email").status_code == 302
    assert "/auth/login" in client.get("/email")["Location"]
    assert client.post("/email", {"new_email": NEW, "current_password": CURRENT}).status_code == 302
    assert mails() == []
 def test_wrong_methods_are_refused(people, adapters):
    path = request_link(adapters)
    client = signed_in(adapters)
    for target in ("/email", path, "/email/undo/abc"):
        assert client.put(target).status_code == 405
        assert client.delete(target).status_code == 405
    assert mail_of(adapters) == OLD
 def test_a_post_without_a_csrf_token_changes_nothing(people, adapters):
    path = request_link(adapters)
    guarded = Client(enforce_csrf_checks=True)
    assert guarded.post(path).status_code == 403
    assert mail_of(adapters) == OLD and PendingEmailChange.objects.get().used_at is None
    undo = confirmed_undo_path(adapters)
    assert guarded.post(undo).status_code == 403
    assert mail_of(adapters) == NEW
    signed = Client(enforce_csrf_checks=True)
    signed.cookies.update(signed_in(adapters).cookies)
    before = len(mails())
    assert (
        signed.post(
            "/email", {"new_email": "x@example.test", "current_password": CURRENT}
        ).status_code
        == 403
    )
    assert len(mails()) == before
 def test_the_form_has_no_username_and_ignores_one(people, adapters):
    client = signed_in(adapters)
    html = client.get("/email").content.decode()
    assert "username" not in html.lower().replace("user name", "")
    client.post(
        "/email", {"new_email": NEW, "current_password": CURRENT, "username": "bob", "sam": "bob"}
    )
    assert PendingEmailChange.objects.get().user_key == "alice"
    assert mail_of(adapters, "bob") == "bob@example.test"
 # -- failures after the directory changed -----------------------------------------
 def test_a_failure_after_the_confirming_write_is_reported_as_unclear(
    people, adapters, monkeypatch, caplog
 ):
    path = request_link(adapters)
    def broken(*args, **kwargs):
        raise RuntimeError("database trouble " + OLD)
    monkeypatch.setattr(services.mailer, "enqueue", broken)
    with caplog.at_level(logging.ERROR):
        response = Client().post(path)
    assert response.status_code == 500 and "Result unclear" in response.content.decode()
    assert mail_of(adapters) == NEW  # the directory did change
    assert "email.changed" not in actions()
    entry = AuditEntry.objects.get(action="email.change_failed")
    assert entry.detail == {"reason": "unclear", "step": "queue_notice"}
    record = next(r for r in caplog.records if "did not record it" in r.getMessage())
    assert record.levelno == logging.ERROR and "user=alice" in record.getMessage()
    assert "RuntimeError" in record.getMessage()
    for secret in (OLD, NEW, path.rsplit("/", 1)[1], "database trouble"):
        assert secret not in caplog.text
 def test_a_failure_after_the_undoing_write_still_ends_the_sessions(
    people, adapters, monkeypatch, caplog
 ):
    path = confirmed_undo_path(adapters)
    other = signed_in(adapters)
    def broken(*args, **kwargs):
        raise RuntimeError("database trouble")
    monkeypatch.setattr(services.mailer, "enqueue", broken)
    with caplog.at_level(logging.ERROR):
        response = Client().post(path)
    assert response.status_code == 500
    assert mail_of(adapters) == OLD and other.get("/").status_code == 302
    assert AuditEntry.objects.get(action="email.undo_failed").detail["reason"] == "unclear"
    assert "user=alice" in caplog.text and OLD not in caplog.text
 def test_an_unclear_undo_from_a_signed_in_browser_shows_the_unclear_page(
    people, adapters, monkeypatch
 ):
    path = confirmed_undo_path(adapters)
    browser = signed_in(adapters)
    def broken(*args, **kwargs):
        raise RuntimeError("database trouble")
    monkeypatch.setattr(services.mailer, "enqueue", broken)
    assert browser.post(path).status_code == 500  # not a 400 from its own ended session
    assert browser.get("/").status_code == 302
 def test_a_failure_before_any_write_leaves_the_request_clean(people, adapters, monkeypatch):
    client = signed_in(adapters)
    def broken(*args, **kwargs):
        raise RuntimeError("boom")
    monkeypatch.setattr(services.mailer, "enqueue", broken)
    with pytest.raises(RuntimeError):
        ask(client)
    assert PendingEmailChange.objects.count() == 0
    assert AuditEntry.objects.get(action="email.change_failed").detail == {"reason": "error"}
 # -- no secrets anywhere -----------------------------------------
 def test_no_password_token_or_address_reaches_logs_or_the_audit(people, adapters, caplog):
    with caplog.at_level(logging.DEBUG):
        client = signed_in(adapters)
        ask(client, password=WRONG)
        ask(client)
        confirm = link_in(mails(NEW)[0])
        Client().get(confirm)
        Client().post(confirm)
        undo = link_in(mails(OLD)[0])
        Client().post(undo)
        Client().post(undo)
    tokens = [confirm.rsplit("/", 1)[1], undo.rsplit("/", 1)[1]]
    stored = " ".join(str(e.detail) + e.action + e.target + e.ip for e in AuditEntry.objects.all())
    stored += " ".join(str(dict(s.get_decoded())) for s in UserSession.objects.all())
    for secret in (CURRENT, WRONG, OLD, NEW, ADMIN, *tokens):
        assert secret not in caplog.text, secret
        assert secret not in stored, secret
    hashes = " ".join(PendingEmailChange.objects.values_list("token_hash", flat=True))
    assert all(token not in hashes for token in tokens)
 def test_the_password_field_is_masked_in_error_reports(people, adapters, monkeypatch):
    from django.views.debug import SafeExceptionReporterFilter
    def crash(*args, **kwargs):
        raise RuntimeError("crash")
    monkeypatch.setattr(services, "request_email_change", crash)
    client = signed_in(adapters)
    client.raise_request_exception = False
    response = ask(client)
    assert response.status_code == 500
    request = response.wsgi_request
    shown = SafeExceptionReporterFilter().get_post_parameters(request)
    assert shown["current_password"] == SafeExceptionReporterFilter.cleansed_substitute
 def test_the_migrations_are_complete(db):
    call_command("makemigrations", check=True, dry_run=True, verbosity=0)
 def test_an_unavailable_directory_error_type_is_what_the_service_raises(people, adapters):
    adapters.directory.unavailable = True
    from portal.accounts.sessions import SessionUser
    with pytest.raises(DirectoryUnavailable):
        services.request_email_change(SessionUser.from_sam("alice"), NEW, CURRENT, ip="127.0.0.1")
 def test_a_link_token_is_blanked_in_the_request_log_path():
    from portal.config.logging import redact
    assert redact("/email/confirm/abc-DEF_123") == "/email/confirm/<redacted>"
    assert redact("Internal Server Error: /email/undo/abc") == (
        "Internal Server Error: /email/undo/<redacted>"
    )
    assert redact("/password") == "/password"
 def test_the_configured_handler_blanks_tokens_from_every_logger():
    import io
    import logging.config
    from portal.config.logging import build_logging
    root = logging.getLogger()
    saved = (root.handlers[:], root.level)
    request_logger = logging.getLogger("django.request")
    saved_request = (request_logger.handlers[:], request_logger.level)
    try:
        logging.config.dictConfig(build_logging("INFO"))
        handler = root.handlers[0]
        stream = io.StringIO()
        handler.setStream(stream)
        logging.getLogger("django.security.csrf").warning(
            "Forbidden (CSRF token missing): /email/confirm/SECRET-token_123"
        )
        output = stream.getvalue()
    finally:
        root.handlers[:], root.level = saved
        request_logger.handlers[:], request_logger.level = saved_request
    assert "SECRET-token_123" not in output and "/email/confirm/<redacted>" in output
@@ -0,0 +1,411 @@
 """Profile edit: the page, the phone format, the diff against AD, audit and errors (FR-9)."""
 import logging
 import re
 import pytest
 from django.test import Client
 from portal.adapters.directory import DirectoryUnavailable, Forbidden, ValidationFailed
 from portal.audit import services as audit_services
 from portal.audit.models import AuditEntry
 from portal.profile.forms import InvalidNumber, normalize_phone
 from portal.tests.helpers import sign_in
 pytestmark = pytest.mark.django_db(transaction=True)
 VALID = {
    "first_name": "Alice",
    "last_name": "Anders",
    "display_name": "Alice Anders",
    "telephone": "+49301234567",
    "mobile": "intern 23",
 }
@pytest.fixture
 def people(adapters):
    d = adapters.directory
    d.add_user(
        "alice",
        given_name="Alice",
        surname="Anders",
        display_name="Alice Anders",
        mail="alice@example.test",
        telephone="+49301234567",
        mobile="intern 23",  # an old value that is not a number
    )
    d.add_user("bob", given_name="Bob", surname="Brown", display_name="Bob Brown")
    return d
 def signed_in(adapters, name="alice"):
    client, _ = sign_in(adapters, name)
    return client
 def save(client, **changes):
    return client.post("/profile", {**VALID, **changes})
 def profile_entries():
    return AuditEntry.objects.filter(action__startswith="profile.")
 def audit_actions():
    return list(profile_entries().values_list("action", flat=True))
 def stored(adapters, sam="alice"):
    return adapters.directory.find_user(sam)
 class Spy:
    def __init__(self, directory):
        self.writes = []
        original = directory.update_attributes
        def wrapper(sam, changes):
            self.writes.append((sam, dict(changes)))
            return original(sam, changes)
        directory.update_attributes = wrapper
 # -- the page ------------------------------------------------------------------------------
 def test_the_page_shows_the_ad_values_and_plain_text_for_username_and_mail(people, adapters):
    html = signed_in(adapters).get("/profile").content.decode()
    for value in ("Alice", "Anders", "Alice Anders", "+49301234567", "intern 23"):
        assert f'value="{value}"' in html
    assert "alice@example.test" in html
    assert 'name="username"' not in html and 'name="mail"' not in html
    assert re.search(r'<a href="/profile" class="current" aria-current="page">', html)
 def test_the_page_speaks_german_in_both_registers(people, adapters):
    client = signed_in(adapters)
    client.cookies["pm_lang"] = "de"
    assert "Diese Angaben sind in Ihrem Konto" in client.get("/profile").content.decode()
    client.cookies["pm_form"] = "du"
    assert "in deinem Konto" in client.get("/profile").content.decode()
 def test_the_page_is_unavailable_when_the_directory_is_down(people, adapters):
    client = signed_in(adapters)
    adapters.directory.unavailable = True
    response = client.get("/profile")
    assert response.status_code == 503 and b'name="first_name"' not in response.content
 def test_access_rules(people, adapters):
    assert Client().get("/profile").status_code == 302
    assert Client().post("/profile", VALID).status_code == 302
    assert signed_in(adapters).put("/profile").status_code == 405
    client = signed_in(adapters)
    strict = Client(enforce_csrf_checks=True)
    strict.cookies = client.cookies
    assert strict.post("/profile", {**VALID, "first_name": "Eve"}).status_code == 403
    assert stored(adapters).given_name == "Alice"
 # -- saving ----------------------------------------------------------------------------------
 def test_only_the_changed_fields_are_written_and_audited_without_values(people, adapters):
    spy = Spy(adapters.directory)
    client = signed_in(adapters)
    response = save(client, first_name="Alicia", mobile="0171 / 123456")
    assert response.status_code == 302 and response["Location"] == "/profile"
    assert spy.writes == [("alice", {"givenName": "Alicia", "mobile": "+49171123456"})]
    account = stored(adapters)
    assert (account.given_name, account.mobile) == ("Alicia", "+49171123456")
    assert (account.surname, account.telephone, account.mail) == (
        "Anders",
        "+49301234567",
        "alice@example.test",
    )
    entry = AuditEntry.objects.get(action="profile.updated")
    assert entry.actor == "alice" and entry.ip == "127.0.0.1"
    assert entry.detail == {"fields": ["first_name", "mobile"]}
    assert profile_entries().count() == 1
 def test_the_saved_notice_shows_once_with_the_stored_number_form(people, adapters):
    client = signed_in(adapters)
    response = save(client, telephone="0049 (0) 30 1234568")
    page = client.get(response["Location"]).content.decode()
    assert "Your changes were saved." in page and 'value="+49301234568"' in page
    assert "Your changes were saved." not in client.get("/profile").content.decode()
 def test_nothing_changed_makes_no_write_and_no_audit(people, adapters):
    spy = Spy(adapters.directory)
    client = signed_in(adapters)
    response = save(client)
    assert response.status_code == 302 and spy.writes == [] and profile_entries().count() == 0
    assert "Nothing to change." in client.get("/profile").content.decode()
 def test_a_number_equal_once_normalized_is_not_a_change(people, adapters):
    spy = Spy(adapters.directory)
    save(signed_in(adapters), telephone="+49 30 1234567")
    assert spy.writes == [] and profile_entries().count() == 0
 def test_an_empty_number_clears_the_attribute(people, adapters):
    spy = Spy(adapters.directory)
    save(signed_in(adapters), telephone="")
    assert spy.writes == [("alice", {"telephoneNumber": ""})]
    assert stored(adapters).telephone == ""
 def test_an_old_value_that_is_not_a_number_never_blocks_another_field(people, adapters):
    save(signed_in(adapters), last_name="Ander")
    account = stored(adapters)
    assert account.surname == "Ander" and account.mobile == "intern 23"
 def test_old_values_that_break_the_rules_are_left_alone_but_changed_ones_are_checked(
    people, adapters
 ):
    long_name = "x" * 100
    adapters.directory.add_user("carol", given_name=long_name, surname="", display_name="Carol")
    client = signed_in(adapters, "carol")
    spy = Spy(adapters.directory)
    old = {
        "first_name": long_name,
        "last_name": "",
        "display_name": "Carol",
        "telephone": "",
        "mobile": "",
    }
    assert client.post("/profile", {**old, "mobile": "5"}).status_code == 400
    assert client.post("/profile", {**old, "display_name": "Caro"}).status_code == 302
    assert spy.writes == [("carol", {"displayName": "Caro"})]
 def test_the_session_name_follows_the_saved_display_name(people, adapters):
    client = signed_in(adapters)
    assert "Alice Anders" in client.get("/").content.decode()
    save(client, display_name="Ally")
    assert "Ally" in client.get("/profile").content.decode()
    assert "Alice Anders" not in client.get("/").content.decode()
 def test_the_save_compares_with_a_fresh_read_not_with_the_page(people, adapters):
    spy = Spy(adapters.directory)
    client = signed_in(adapters)
    client.get("/profile")
    adapters.directory.update_attributes("alice", {"sn": "Elsewhere"})
    save(client, display_name="Ally", last_name="Elsewhere")  # same as AD now: not written
    assert spy.writes[1:] == [("alice", {"displayName": "Ally"})]
 # -- phone numbers ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    ("text", "result"),
    [
        ("0171 / 123456", "+49171123456"),
        ("0171123456", "+49171123456"),
        ("+49 171 123456", "+49171123456"),
        ("+49 (0) 171-123456", "+49171123456"),
        ("0049 171 123456", "+49171123456"),
        ("0049 (0) 171 123456", "+49171123456"),
        ("+43 660 1234567", "+436601234567"),
        ("0043 660 1234567", "+436601234567"),
        ("(0171) 123.456", "+49171123456"),
        ("+49171123456", "+49171123456"),
    ],
 )
 def test_phone_inputs_and_their_stored_form(text, result):
    assert normalize_phone(text) == result
    assert normalize_phone(result) == result
@pytest.mark.parametrize(
    "text",
    [
        "171123456",
        "abc",
        "0171 abc",
        "+49 171 12+3456",
        "+49 1711 23456 7890 1234",
        "0123",
        "+49 12",
        "++49171123456",
        "intern 23",
        "",
        "*)(uid=*",
    ],
 )
 def test_phone_inputs_that_are_refused(text):
    with pytest.raises(InvalidNumber):
        normalize_phone(text)
@pytest.mark.parametrize("field", ["telephone", "mobile"])
@pytest.mark.parametrize("text", ["171123456", "0171 abc", "+49 12", "+49 171 12+3456"])
 def test_a_bad_number_is_refused_under_its_field(people, adapters, field, text):
    spy = Spy(adapters.directory)
    response = save(signed_in(adapters), **{field: text})
    html = response.content.decode()
    assert response.status_code == 400 and spy.writes == []
    assert "must be a number that starts with + and the country code, or with 0" in html
    assert 'class="error"' in html and profile_entries().count() == 0
 # -- invalid values ---------------------------------------------------------------------------
@pytest.mark.parametrize("field", ["first_name", "last_name", "display_name"])
 def test_a_required_name_cannot_be_empty(people, adapters, field):
    spy = Spy(adapters.directory)
    response = save(signed_in(adapters), **{field: "   "})
    assert response.status_code == 400 and spy.writes == [] and profile_entries().count() == 0
    assert "is required." in response.content.decode()
@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("first_name", "x" * 65),
        ("last_name", "x" * 65),
        ("display_name", "x" * 257),
        ("first_name", "A\nB"),
        ("last_name", "A\tB"),
        ("display_name", "A\x00B"),
    ],
 )
 def test_too_long_or_control_characters_are_refused_and_other_input_stays(
    people, adapters, field, value
 ):
    spy = Spy(adapters.directory)
    response = save(signed_in(adapters), mobile="0171 5", **{field: value})
    html = response.content.decode()
    assert response.status_code == 400 and spy.writes == [] and profile_entries().count() == 0
    assert "Your changes were not saved." in html
    assert 'value="0171 5"' in html  # what else was typed stays in the form
    assert f'<a href="#id_{field}">' in html
 def test_the_limits_are_ads_limits(people, adapters):
    save(signed_in(adapters), first_name="x" * 64, display_name="y" * 256)
    account = stored(adapters)
    assert len(account.given_name) == 64 and len(account.display_name) == 256
 def test_values_are_trimmed(people, adapters):
    save(signed_in(adapters), first_name="  Alicia  ")
    assert stored(adapters).given_name == "Alicia"
@pytest.mark.parametrize("text", ["*)(uid=*", "\\00", "CN=x,DC=y", "a)(|(cn=*"])
 def test_injection_strings_are_plain_text_and_touch_nothing_else(people, adapters, text):
    spy = Spy(adapters.directory)
    before_bob = stored(adapters, "bob")
    assert save(signed_in(adapters), display_name=text).status_code == 302
    assert spy.writes == [("alice", {"displayName": text})]
    assert stored(adapters).display_name == text
    assert stored(adapters, "bob") == before_bob
 # -- who is edited -----------------------------------------------------------------------------
 def test_a_posted_username_is_ignored_and_another_user_is_untouched(people, adapters):
    before_bob = stored(adapters, "bob")
    response = save(signed_in(adapters), first_name="Alicia", username="bob", sam="bob", dn="x")
    assert response.status_code == 302
    assert stored(adapters).given_name == "Alicia" and stored(adapters, "bob") == before_bob
    assert AuditEntry.objects.get(action="profile.updated").actor == "alice"
 # -- failures ------------------------------------------------------------------------------------
 def test_ad_refusing_a_value_changes_nothing_and_is_audited_as_failure(people, adapters):
    def refuse(sam, changes):
        raise ValidationFailed("constraint")
    adapters.directory.update_attributes = refuse
    response = save(signed_in(adapters), first_name="Alicia")
    html = response.content.decode()
    assert response.status_code == 400
    assert "A value was not accepted. Please check your entries." in html
    assert "constraint" not in html and 'value="Alicia"' in html
    assert stored(adapters).given_name == "Alice"
    assert audit_actions() == ["profile.update_failed"]
    assert profile_entries().get().detail == {"reason": "refused"}
 def test_directory_down_on_save_is_503_and_audited(people, adapters):
    client = signed_in(adapters)
    adapters.directory.unavailable = True
    response = save(client, first_name="Alicia")
    assert response.status_code == 503
    assert "This is not possible right now. Please try again later." in response.content.decode()
    adapters.directory.unavailable = False
    assert stored(adapters).given_name == "Alice"
    assert audit_actions() == ["profile.update_failed"]
    assert profile_entries().get().detail == {"reason": "unavailable"}
 def test_directory_failing_during_the_write_rolls_back_and_audits(people, adapters):
    def down(sam, changes):
        raise DirectoryUnavailable("gone")
    adapters.directory.update_attributes = down
    assert save(signed_in(adapters), first_name="Alicia").status_code == 503
    assert audit_actions() == ["profile.update_failed"]
 def test_a_failure_after_the_ad_call_is_logged_and_the_user_told_it_is_unclear(
    people, adapters, monkeypatch, caplog
 ):
    real_record = audit_services.record
    def broken(*args, **kwargs):
        raise RuntimeError("database gone: Alicia")
    client = signed_in(adapters)
    monkeypatch.setattr(audit_services, "record", broken)
    with caplog.at_level(logging.ERROR):
        response = save(client, first_name="Alicia", mobile="0171 5555555")
    monkeypatch.setattr(audit_services, "record", real_record)
    assert response.status_code == 500 and "may have been changed" in response.content.decode()
    assert stored(adapters).given_name == "Alicia"  # AD did change
    assert audit_actions() == ["profile.update_failed"]  # no "updated" entry survived
    log = next(r.getMessage() for r in caplog.records if r.levelno == logging.ERROR)
    assert "user=alice" in log and "fields=first_name,mobile" in log and "step=audit" in log
    assert "Alicia" not in log and "5555555" not in log
 def test_a_missing_write_right_is_the_not_possible_page_and_a_failure_entry(
    people, adapters, caplog
 ):
    def forbidden(sam, changes):
        raise Forbidden("insufficient access: Alicia")
    client = signed_in(adapters)
    adapters.directory.update_attributes = forbidden
    with caplog.at_level(logging.ERROR):
        response = save(client, first_name="Alicia")
    assert response.status_code == 503
    assert "This is not possible right now." in response.content.decode()
    assert audit_actions() == ["profile.update_failed"]
    assert profile_entries().get().detail == {"reason": "forbidden"}
    assert not any("Alicia" in r.getMessage() for r in caplog.records)
 def test_an_invalid_save_reads_the_directory_once(people, adapters):
    client = signed_in(adapters)
    reads = []
    original = adapters.directory.find_user
    adapters.directory.find_user = lambda sam: reads.append(sam) or original(sam)
    response = save(client, first_name="  ")
    assert response.status_code == 400 and len(reads) == 1
    assert "alice@example.test" in response.content.decode()
		`@@ -0,0 +1 @@`
							`{% load i18n %}{% autoescape off %}{% trans "Please confirm your new e-mail address" %}{% endautoescape %}`