Use the email address as the login identifier and require confirmation
Sign-in now takes the email address rather than a username. This needed no migration: users.email is already citext NOT NULL UNIQUE in the control plane, so it is a case-insensitively unique login key on its own. username survives as an optional display handle for audit lines and is never a login path, which removes the redundancy of asking an administrator for two identifiers. An account created by an administrator now starts in the control plane's invited state and cannot sign in until a single-use link is consumed. The token maps onto bootstrap_tokens, which was already shaped for exactly this and had been unused, so that needed no migration either. Modelling invited as a real state also retires the adapter's earlier fail-closed refusal of that value, which existed because invited was previously unrepresentable. Three decisions worth stating plainly. The "email not verified" message is sent only after the password has been verified. Reporting it earlier would turn the login form into an oracle for which addresses have accounts, which is the one thing the uniform authentication error exists to prevent. A test asserts that a wrong password and an unknown account are byte-identical, and it caught a real bug: the check was first written before the password comparison, and the generic "not active" branch swallowed invited accounts. Both are now ordered deliberately. The operator bootstrap path does not require verification. The operator is attesting to the address themselves, which is the standing an administrator already has, and requiring a delivered email there would deadlock first-time setup on SMTP configuration that does not exist yet. The logging mail transport is refused in production. A deployment that logged links instead of sending them would appear to work while every new account sat permanently unverified. The development transport writes the link with the token redacted, because a log is where live credentials end up. That means a human cannot complete the flow by reading the log; the test harness captures the message in memory instead. This is a deliberate limitation rather than an oversight, and it is documented. Also: production now requires a mail transport and an absolute verification base URL alongside the existing DSN gate, because an account that cannot be verified cannot be used at all. 269 tests pass, 3 skipped. Verified end to end against a real uvicorn server: bootstrap login, organization creation, an invited account with its one-time password, and a handle rejected as a login path.
This commit is contained in:
umbra audio
@@ -265,3 +265,6 @@ unreviewed store, so nothing in this slice can touch `umbracontrol` by accident.
|
||||
- The static UI can be connected to the API without changing the audio worker.
|
||||
- Production migration and bootstrap remain blocked until explicitly approved.
|
||||
- Email is a required field on account creation, because `users.email` is `NOT NULL UNIQUE`. Enforced in the domain so it fails the same way in development and production. *Done.*
|
||||
- The email address is the login identifier, and `username` is an optional display handle that is never a login path. `users.email` is already `citext NOT NULL UNIQUE`, so this needed no migration. *Done.*
|
||||
- An administrator-provisioned account starts `invited` and cannot sign in until a single-use verification link is consumed, using the previously unused `bootstrap_tokens` table. The unverified message is only sent after the password is verified, so the login form cannot enumerate accounts. The operator bootstrap path skips verification, because the operator attests to the address themselves. *Done.*
|
||||
- A second factor is deliberately **not** the email address. `mfa_methods` constrains the method to `totp`, `webauthn`, or `recovery_code`, and email is excluded; an emailed code is not a second factor because email is not secret. TOTP remains the next step and can use the existing table. *Pending.*
|
||||
|
||||
@@ -85,6 +85,37 @@ UMBRATRANS_TEST_DSN='postgresql://app@localhost/umbratrans_scratch' \
|
||||
|
||||
The fixture refuses a DSN whose database name contains `umbracontrol`.
|
||||
|
||||
## Email is the login identifier, and it must be confirmed
|
||||
|
||||
Sign-in uses the email address. `users.email` is already `citext NOT NULL
|
||||
UNIQUE` in the control plane, so it is a case-insensitively unique login key and
|
||||
**no migration was needed**. `username` is now an optional display handle for
|
||||
audit lines and is never a login path.
|
||||
|
||||
An account created by an administrator starts in the control plane's `invited`
|
||||
state and cannot sign in until a single-use link is consumed. The token maps
|
||||
onto the existing `bootstrap_tokens` table, which was already shaped for this and
|
||||
was unused, so that needed no migration either.
|
||||
|
||||
Three details that matter:
|
||||
|
||||
- **The "email not verified" message is only sent after the password is proven
|
||||
correct.** Reporting it earlier would turn the login form into an oracle for
|
||||
which addresses have accounts. A test asserts that a wrong password and an
|
||||
unknown account are indistinguishable.
|
||||
- **The operator bootstrap path needs no verification.** The operator is
|
||||
attesting to the address themselves, which is the same standing an
|
||||
administrator has. Requiring a delivered email there would deadlock first-time
|
||||
setup on the SMTP configuration.
|
||||
- **The logging mail transport is refused in production.** Otherwise a
|
||||
deployment would appear to send verification mail that never left the machine,
|
||||
and accounts would sit unverified with nobody noticing.
|
||||
|
||||
The development transport writes the link to the log **with the token
|
||||
redacted**, because a log is where live credentials end up. So a human cannot
|
||||
complete the flow by reading the log: use SMTP, or exercise it through the test
|
||||
harness, which captures the message in memory.
|
||||
|
||||
## Tenancy
|
||||
|
||||
`organizations` is a first-class record with `name`, `slug`, and a status of
|
||||
|
||||
@@ -44,6 +44,7 @@ DEVELOPMENT_DATA_ROOT = Path(".local/data/development")
|
||||
DEVELOPMENT_REPOSITORY_RELATIVE_PATH = Path("umbratrans-web/dev-accounts.json")
|
||||
PRODUCTION_REPOSITORY_KIND = "postgres"
|
||||
REPOSITORY_KINDS = ("memory", "json", PRODUCTION_REPOSITORY_KIND)
|
||||
MAILER_TRANSPORTS = ("log", "smtp")
|
||||
|
||||
|
||||
class Environment(StrEnum):
|
||||
@@ -125,6 +126,14 @@ class Settings:
|
||||
database_pool_min: int = 1
|
||||
database_pool_max: int = 10
|
||||
static_root: Path | None = None
|
||||
mailer_transport: str = "log"
|
||||
smtp_host: str = ""
|
||||
smtp_port: int = 587
|
||||
smtp_username: str | None = None
|
||||
smtp_password: str | None = field(default=None, repr=False, compare=False)
|
||||
smtp_sender: str = "umbratrans@localhost"
|
||||
smtp_starttls: bool = True
|
||||
verification_base_url: str = ""
|
||||
allowed_hosts: tuple[str, ...] = ()
|
||||
last_seen_write_interval_seconds: int = 30
|
||||
token_pepper: bytes | None = field(default=None, repr=False, compare=False)
|
||||
@@ -151,6 +160,13 @@ class Settings:
|
||||
raise ConfigurationError("weak_token_pepper", f"Token pepper must be at least {MIN_PEPPER_BYTES} bytes.")
|
||||
if any(host.strip() in {"*", ""} for host in self.allowed_hosts):
|
||||
raise ConfigurationError("wildcard_host_forbidden", "A wildcard or empty allowed host is not permitted.")
|
||||
if self.mailer_transport not in MAILER_TRANSPORTS:
|
||||
raise ConfigurationError(
|
||||
"invalid_mailer_transport",
|
||||
f"Mail transport must be one of {', '.join(MAILER_TRANSPORTS)}.",
|
||||
)
|
||||
if self.verification_base_url and not self.verification_base_url.startswith(("http://", "https://")):
|
||||
raise ConfigurationError("invalid_verification_base_url", "The verification base URL must be absolute.")
|
||||
if self.database_pool_min < 1 or self.database_pool_max < self.database_pool_min:
|
||||
raise ConfigurationError("invalid_pool_bounds", "Pool sizes must satisfy 1 <= min <= max.")
|
||||
if environment.is_production:
|
||||
@@ -172,6 +188,20 @@ class Settings:
|
||||
"production_dsn_missing",
|
||||
"Production requires a PostgreSQL connection string.",
|
||||
)
|
||||
# Email validation is a required part of account creation, so a
|
||||
# production process that cannot deliver a link would leave every new
|
||||
# account permanently unverified.
|
||||
if self.mailer_transport != "smtp" or not self.smtp_host:
|
||||
raise ConfigurationError(
|
||||
"production_mailer_missing",
|
||||
"Production requires a real SMTP transport; the logging transport would "
|
||||
"write verification links to the log and never deliver them.",
|
||||
)
|
||||
if not self.verification_base_url:
|
||||
raise ConfigurationError(
|
||||
"production_verification_url_missing",
|
||||
"Production requires an absolute verification base URL for mailed links.",
|
||||
)
|
||||
if self.repository_kind != PRODUCTION_REPOSITORY_KIND:
|
||||
raise ConfigurationError(
|
||||
"production_repository_missing",
|
||||
@@ -267,6 +297,9 @@ def load_settings(environ: Mapping[str, str] | None = None) -> Settings:
|
||||
|
||||
allowed_hosts = tuple(host.strip() for host in env.get("UMBRATRANS_ALLOWED_HOSTS", "").split(",") if host.strip())
|
||||
|
||||
# Verification links must be absolute, or the mailed link is unusable.
|
||||
verification_base_url = env.get("UMBRATRANS_VERIFICATION_BASE_URL", "").strip().rstrip("/")
|
||||
|
||||
raw_pepper = env.get("UMBRATRANS_TOKEN_PEPPER")
|
||||
pepper = raw_pepper.encode("utf-8") if raw_pepper else None
|
||||
if pepper is not None and len(pepper) < MIN_PEPPER_BYTES:
|
||||
@@ -305,6 +338,14 @@ def load_settings(environ: Mapping[str, str] | None = None) -> Settings:
|
||||
database_pool_min=pool_min,
|
||||
database_pool_max=pool_max,
|
||||
static_root=static_root,
|
||||
mailer_transport=env.get("UMBRATRANS_MAILER", "log").strip().lower(),
|
||||
smtp_host=env.get("UMBRATRANS_SMTP_HOST", "").strip(),
|
||||
smtp_port=_int_from(env.get("UMBRATRANS_SMTP_PORT", "587"), name="UMBRATRANS_SMTP_PORT", minimum=1),
|
||||
smtp_username=env.get("UMBRATRANS_SMTP_USERNAME") or None,
|
||||
smtp_password=env.get("UMBRATRANS_SMTP_PASSWORD") or None,
|
||||
smtp_sender=env.get("UMBRATRANS_SMTP_SENDER", "umbratrans@localhost").strip(),
|
||||
smtp_starttls=_bool_from(env.get("UMBRATRANS_SMTP_STARTTLS", "true"), name="UMBRATRANS_SMTP_STARTTLS"),
|
||||
verification_base_url=verification_base_url,
|
||||
allowed_hosts=allowed_hosts,
|
||||
token_pepper=pepper,
|
||||
)
|
||||
|
||||
@@ -39,6 +39,7 @@ from .models import (
|
||||
RoleName,
|
||||
SessionRecord,
|
||||
User,
|
||||
VerificationToken,
|
||||
)
|
||||
from .repository import InMemoryAccountRepository
|
||||
|
||||
@@ -138,6 +139,9 @@ class JsonDevelopmentRepository(InMemoryAccountRepository):
|
||||
session = _session_from_payload(item)
|
||||
self._sessions[session.id] = session
|
||||
self._sessions_by_token[session.token_hash] = session.id
|
||||
for item in payload.get("verification_tokens", []):
|
||||
token = _verification_from_payload(item)
|
||||
self._verification_tokens[token.id] = token
|
||||
for item in payload.get("audit_events", []):
|
||||
self._audit.append(_audit_from_payload(item))
|
||||
|
||||
@@ -152,6 +156,9 @@ class JsonDevelopmentRepository(InMemoryAccountRepository):
|
||||
"users": [_user_to_payload(user) for user in self._users.values()],
|
||||
"credentials": [_credential_to_payload(item) for item in self._credentials.values()],
|
||||
"sessions": [_session_to_payload(item) for item in self._sessions.values()],
|
||||
"verification_tokens": [
|
||||
_verification_to_payload(token) for token in self._verification_tokens.values()
|
||||
],
|
||||
"audit_events": [_audit_to_payload(item) for item in self._audit],
|
||||
}
|
||||
|
||||
@@ -228,6 +235,23 @@ class JsonDevelopmentRepository(InMemoryAccountRepository):
|
||||
self.flush()
|
||||
return result
|
||||
|
||||
def create_verification_token(self, token: VerificationToken) -> VerificationToken:
|
||||
result = super().create_verification_token(token)
|
||||
self.flush()
|
||||
return result
|
||||
|
||||
def mark_verification_token_consumed(self, token_id: UUID, *, now: datetime) -> bool:
|
||||
result = super().mark_verification_token_consumed(token_id, now=now)
|
||||
if result:
|
||||
self.flush()
|
||||
return result
|
||||
|
||||
def delete_expired_verification_tokens(self, *, now: datetime) -> int:
|
||||
result = super().delete_expired_verification_tokens(now=now)
|
||||
if result:
|
||||
self.flush()
|
||||
return result
|
||||
|
||||
def append_audit_event(self, event: AuditEvent) -> AuditEvent:
|
||||
result = super().append_audit_event(event)
|
||||
self.flush()
|
||||
@@ -358,6 +382,32 @@ def _session_from_payload(payload: Mapping[str, Any]) -> SessionRecord:
|
||||
)
|
||||
|
||||
|
||||
def _verification_to_payload(token: VerificationToken) -> dict[str, Any]:
|
||||
"""Serialize a token. Only the hash is written, never the plaintext."""
|
||||
|
||||
return {
|
||||
"id": str(token.id),
|
||||
"user_id": str(token.user_id),
|
||||
"token_hash": token.token_hash,
|
||||
"expires_at": _utc(token.expires_at),
|
||||
"created_at": _utc(token.created_at),
|
||||
"consumed_at": _utc(token.consumed_at) if token.consumed_at else None,
|
||||
"purpose": token.purpose,
|
||||
}
|
||||
|
||||
|
||||
def _verification_from_payload(payload: Mapping[str, Any]) -> VerificationToken:
|
||||
return VerificationToken(
|
||||
id=UUID(payload["id"]),
|
||||
user_id=UUID(payload["user_id"]),
|
||||
token_hash=payload["token_hash"],
|
||||
expires_at=_parse_datetime(payload["expires_at"]),
|
||||
created_at=_parse_datetime(payload["created_at"]),
|
||||
consumed_at=_parse_datetime(payload["consumed_at"]) if payload.get("consumed_at") else None,
|
||||
purpose=payload.get("purpose", "email_verification"),
|
||||
)
|
||||
|
||||
|
||||
def _audit_to_payload(event: AuditEvent) -> dict[str, Any]:
|
||||
return {
|
||||
"id": str(event.id),
|
||||
|
||||
@@ -86,6 +86,45 @@ class InvalidCredentialsError(UmbraWebError):
|
||||
message = "The username or password is incorrect."
|
||||
|
||||
|
||||
class EmailNotVerifiedError(UmbraWebError):
|
||||
"""The password was correct but the email address is not yet verified.
|
||||
|
||||
Only ever raised after a successful password check. Reporting it earlier
|
||||
would turn the login form into an account-existence oracle, which is the one
|
||||
thing the uniform authentication error exists to prevent.
|
||||
"""
|
||||
|
||||
code = "email_not_verified"
|
||||
http_status = 403
|
||||
message = "Confirm your email address before signing in. Check your inbox for the verification link."
|
||||
|
||||
|
||||
class VerificationFailedError(UmbraWebError):
|
||||
"""A verification link was not accepted.
|
||||
|
||||
One message for every reason - unknown, already used, expired - so the
|
||||
endpoint cannot be used to probe which links ever existed.
|
||||
"""
|
||||
|
||||
code = "verification_failed"
|
||||
http_status = 400
|
||||
message = "That verification link is not valid. It may have been used already or have expired."
|
||||
|
||||
|
||||
class VerificationDeliveryError(UmbraWebError):
|
||||
"""A verification link could not be handed to the mail transport.
|
||||
|
||||
Reported as a server-side failure rather than a success, because telling an
|
||||
administrator that a link was sent when nothing left the machine leaves an
|
||||
account permanently unverifiable with no indication why. The message names
|
||||
no address and carries no token.
|
||||
"""
|
||||
|
||||
code = "verification_delivery_failed"
|
||||
http_status = 502
|
||||
message = "The verification link could not be sent. The account was created and is not yet usable."
|
||||
|
||||
|
||||
class PasswordChangeRequiredError(UmbraWebError):
|
||||
"""The session exists but may only be used to change the first password."""
|
||||
|
||||
@@ -172,6 +211,7 @@ __all__ = [
|
||||
"ConfigurationError",
|
||||
"ConflictError",
|
||||
"CsrfError",
|
||||
"EmailNotVerifiedError",
|
||||
"EnrollmentRefusedError",
|
||||
"ForbiddenError",
|
||||
"InputError",
|
||||
@@ -182,5 +222,7 @@ __all__ = [
|
||||
"PasswordPolicyError",
|
||||
"RepositoryError",
|
||||
"StateTransitionError",
|
||||
"VerificationDeliveryError",
|
||||
"VerificationFailedError",
|
||||
"UmbraWebError",
|
||||
]
|
||||
|
||||
222
umbratrans_web/mailer.py
Normal file
222
umbratrans_web/mailer.py
Normal file
@@ -0,0 +1,222 @@
|
||||
"""Outbound mail, behind an interface.
|
||||
|
||||
There is no SMTP server configured on this host, and inventing one is not this
|
||||
module's job. So the transport is pluggable and the default for a non-production
|
||||
environment is explicit about what it is: :class:`LoggingMailer` does not send
|
||||
anything. It writes the message to the application log, which makes the
|
||||
verification flow fully exercisable end to end during development, and it says
|
||||
so in its own output so nobody mistakes a logged link for a delivered one.
|
||||
|
||||
:class:`SmtpMailer` is the real transport. It is implemented, not stubbed, but
|
||||
it is unexercised here because no SMTP configuration exists to exercise it
|
||||
against, and the README records that.
|
||||
|
||||
Nothing in this module ever logs a message body at INFO in production, and no
|
||||
transport is allowed to raise with the recipient list in the message.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import smtplib
|
||||
import ssl
|
||||
from dataclasses import dataclass
|
||||
from email.message import EmailMessage
|
||||
from email.utils import formataddr, parseaddr
|
||||
from typing import Protocol
|
||||
|
||||
LOGGER = logging.getLogger("umbratrans_web.mailer")
|
||||
|
||||
#: Verification mail is transactional and small. A cap keeps a bug from
|
||||
#: generating megabyte messages.
|
||||
MAX_SUBJECT_CHARS = 200
|
||||
MAX_BODY_CHARS = 20_000
|
||||
|
||||
|
||||
class MailerError(RuntimeError):
|
||||
"""Delivery failed. Safe to show a caller; never contains a token."""
|
||||
|
||||
|
||||
class Mailer(Protocol):
|
||||
"""One method, so a transport can be swapped or faked in tests."""
|
||||
|
||||
def send(self, *, to: str, subject: str, body: str) -> None: ...
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SentMessage:
|
||||
"""What a transport reports back. Used by the development transport and tests."""
|
||||
|
||||
to: str
|
||||
subject: str
|
||||
body: str
|
||||
|
||||
|
||||
def validate_recipient(address: str) -> str:
|
||||
from .models import normalize_email
|
||||
|
||||
return normalize_email(address)
|
||||
|
||||
|
||||
def _check_lengths(subject: str, body: str) -> None:
|
||||
if len(subject) > MAX_SUBJECT_CHARS:
|
||||
raise MailerError("Mail subject exceeds the permitted length.")
|
||||
if len(body) > MAX_BODY_CHARS:
|
||||
raise MailerError("Mail body exceeds the permitted length.")
|
||||
|
||||
|
||||
class LoggingMailer:
|
||||
"""Writes the message to the log instead of sending it.
|
||||
|
||||
The single most important property is that it says what it is. A log line
|
||||
that looks like a delivered mail would let someone believe an address was
|
||||
verified when nothing left the machine.
|
||||
"""
|
||||
|
||||
transport_name = "log"
|
||||
|
||||
def send(self, *, to: str, subject: str, body: str) -> None:
|
||||
_check_lengths(subject, body)
|
||||
recipient = validate_recipient(to)
|
||||
LOGGER.warning(
|
||||
"DEV TRANSPORT: mail was NOT sent. "
|
||||
"transport=log recipient=%s subject=%s\n--- body ---\n%s\n--- end body ---",
|
||||
recipient,
|
||||
subject,
|
||||
body,
|
||||
)
|
||||
|
||||
def last_message(self) -> SentMessage | None: # pragma: no cover - convenience
|
||||
return None
|
||||
|
||||
|
||||
class CapturingMailer:
|
||||
"""Keeps messages in memory. For tests only; never selected by configuration."""
|
||||
|
||||
transport_name = "capture"
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.sent: list[SentMessage] = []
|
||||
|
||||
def send(self, *, to: str, subject: str, body: str) -> None:
|
||||
_check_lengths(subject, body)
|
||||
self.sent.append(SentMessage(to=validate_recipient(to), subject=subject, body=body))
|
||||
|
||||
|
||||
class SmtpMailer:
|
||||
"""Real delivery over SMTP.
|
||||
|
||||
Implemented but unexercised in this environment: no SMTP server is
|
||||
configured here. ``starttls`` is used unless the deployment explicitly sets
|
||||
``starttls=False`` for a relay that is already inside a tunnel. Credentials
|
||||
come from configuration and are never logged, never included in an
|
||||
exception, and never written to a development data file.
|
||||
"""
|
||||
|
||||
transport_name = "smtp"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
host: str,
|
||||
port: int = 587,
|
||||
username: str | None = None,
|
||||
password: str | None = None,
|
||||
sender: str = "umbratrans@localhost",
|
||||
starttls: bool = True,
|
||||
timeout: int = 20,
|
||||
) -> None:
|
||||
if not host:
|
||||
raise MailerError("An SMTP host is required for the smtp transport.")
|
||||
self._host = host
|
||||
self._port = port
|
||||
self._username = username
|
||||
self._password = password
|
||||
self._sender = sender
|
||||
self._starttls = starttls
|
||||
self._timeout = timeout
|
||||
|
||||
def send(self, *, to: str, subject: str, body: str) -> None:
|
||||
_check_lengths(subject, body)
|
||||
recipient = validate_recipient(to)
|
||||
message = EmailMessage()
|
||||
message["Subject"] = subject
|
||||
message["From"] = formataddr(("UmbraTrans", self._sender))
|
||||
message["To"] = recipient
|
||||
message.set_content(body)
|
||||
|
||||
try:
|
||||
with smtplib.SMTP(self._host, self._port, timeout=self._timeout) as client:
|
||||
if self._starttls:
|
||||
client.starttls(context=ssl.create_default_context())
|
||||
if self._username:
|
||||
# Credentials go to the client, never into the log or the
|
||||
# exception text below.
|
||||
client.login(self._username, self._password or "")
|
||||
client.send_message(message)
|
||||
except smtplib.SMTPException as error:
|
||||
# smtplib exception text can quote the server response, which can
|
||||
# include an address. Report the class only.
|
||||
LOGGER.error("smtp delivery failed: %s", type(error).__name__)
|
||||
raise MailerError("The message could not be delivered.") from None
|
||||
except (OSError, ssl.SSLError) as error:
|
||||
LOGGER.error("smtp transport failed: %s", type(error).__name__)
|
||||
raise MailerError("The mail server could not be reached.") from None
|
||||
LOGGER.info("verification mail sent transport=smtp recipient=%s", _mask(recipient))
|
||||
|
||||
|
||||
def _mask(address: str) -> str:
|
||||
"""Log a partially masked address.
|
||||
|
||||
A log line is a place a full address tends to end up copied out of, so the
|
||||
domain is kept for diagnosis and the local part is not.
|
||||
"""
|
||||
|
||||
_, domain = parseaddr(address)
|
||||
name = domain or address.split("@")[-1]
|
||||
return f"***@{name}" if name else "***"
|
||||
|
||||
|
||||
def build_mailer(settings: object) -> Mailer:
|
||||
"""Select a transport from configuration.
|
||||
|
||||
Production refuses the logging transport. Otherwise a production deployment
|
||||
would appear to send verification mail that never left the machine, and
|
||||
accounts would sit unverified with nobody noticing.
|
||||
"""
|
||||
|
||||
environment = str(getattr(settings, "environment", "development"))
|
||||
kind = str(getattr(settings, "mailer_transport", "log"))
|
||||
is_production = environment in {"production", "prod"}
|
||||
|
||||
if kind == "smtp":
|
||||
host = getattr(settings, "smtp_host", "") or ""
|
||||
if not host:
|
||||
raise MailerError("The smtp transport requires a host.")
|
||||
return SmtpMailer(
|
||||
host=host,
|
||||
port=int(getattr(settings, "smtp_port", 587) or 587),
|
||||
username=getattr(settings, "smtp_username", None),
|
||||
password=getattr(settings, "smtp_password", None),
|
||||
sender=getattr(settings, "smtp_sender", "umbratrans@localhost"),
|
||||
starttls=bool(getattr(settings, "smtp_starttls", True)),
|
||||
)
|
||||
|
||||
if is_production:
|
||||
raise MailerError(
|
||||
"Refusing to start in production with the logging mail transport: "
|
||||
"verification links would be written to the log and never delivered."
|
||||
)
|
||||
if kind != "log":
|
||||
raise MailerError(f"Unknown mail transport: {kind!r}.")
|
||||
return LoggingMailer()
|
||||
|
||||
|
||||
__all__ = [
|
||||
"CapturingMailer",
|
||||
"LoggingMailer",
|
||||
"Mailer",
|
||||
"MailerError",
|
||||
"SmtpMailer",
|
||||
"build_mailer",
|
||||
]
|
||||
@@ -33,6 +33,15 @@ MAX_DISPLAY_NAME_LENGTH = 120
|
||||
MAX_EMAIL_LENGTH = 254
|
||||
MAX_REASON_LENGTH = 200
|
||||
MAX_ORGANIZATION_NAME_LENGTH = 120
|
||||
|
||||
#: Lifetime of an email-verification link. Long enough to survive a weekend, not
|
||||
#: long enough that a link found in a mailbox months later still works.
|
||||
VERIFICATION_TOKEN_TTL_SECONDS = 48 * 60 * 60
|
||||
|
||||
#: The only purpose a verification token is minted for. Named so the
|
||||
#: ``bootstrap_tokens`` row cannot later be reused for something else without
|
||||
#: the audit trail showing a different purpose.
|
||||
VERIFICATION_PURPOSE_EMAIL = "email_verification"
|
||||
MIN_ORGANIZATION_NAME_LENGTH = 2
|
||||
MAX_ORGANIZATION_SLUG_LENGTH = 63
|
||||
|
||||
@@ -91,6 +100,22 @@ class EnrollmentMode(StrEnum):
|
||||
|
||||
|
||||
class AccountStatus(StrEnum):
|
||||
"""Lifecycle of one account.
|
||||
|
||||
``INVITED`` is the control plane's own state for an account that exists but
|
||||
has not proved it can receive mail at its address. It is modelled here
|
||||
rather than treated as an unmapped value, because the authoritative schema
|
||||
already uses it: an unvalidated account that could sign in would defeat the
|
||||
point of validating the address, and an account stuck in ``invited`` forever
|
||||
would be indistinguishable from a bug.
|
||||
|
||||
``LOCKED`` exists in the control-plane check constraint and is deliberately
|
||||
absent here. Credential lockout is tracked per credential by
|
||||
``failed_login_count`` and ``locked_until``, so a second account-level status
|
||||
for the same condition would be two sources of truth.
|
||||
"""
|
||||
|
||||
INVITED = "invited"
|
||||
ACTIVE = "active"
|
||||
SUSPENDED = "suspended"
|
||||
DEACTIVATED = "deactivated"
|
||||
@@ -129,8 +154,10 @@ ALLOWED_ORGANIZATION_TRANSITIONS: Mapping[OrganizationStatus, frozenset[Organiza
|
||||
|
||||
#: Allowed account status transitions. ``deactivated`` may only be reversed by
|
||||
#: a platform administrator, which the service checks separately.
|
||||
#: ``invited -> active`` is email validation and is the only way out of it.
|
||||
ALLOWED_STATUS_TRANSITIONS: Mapping[AccountStatus, frozenset[AccountStatus]] = MappingProxyType(
|
||||
{
|
||||
AccountStatus.INVITED: frozenset({AccountStatus.ACTIVE, AccountStatus.DEACTIVATED}),
|
||||
AccountStatus.ACTIVE: frozenset({AccountStatus.SUSPENDED, AccountStatus.DEACTIVATED}),
|
||||
AccountStatus.SUSPENDED: frozenset({AccountStatus.ACTIVE, AccountStatus.DEACTIVATED}),
|
||||
AccountStatus.DEACTIVATED: frozenset({AccountStatus.ACTIVE}),
|
||||
@@ -138,16 +165,22 @@ ALLOWED_STATUS_TRANSITIONS: Mapping[AccountStatus, frozenset[AccountStatus]] = M
|
||||
)
|
||||
|
||||
|
||||
def normalize_username(value: str) -> str:
|
||||
"""Return the canonical stored form of a username.
|
||||
def normalize_username(value: str | None) -> str | None:
|
||||
"""Return the canonical stored form of an optional display handle.
|
||||
|
||||
The value is trimmed but otherwise preserved; case-insensitive comparison
|
||||
is handled separately by :func:`username_key` so that stored identifiers
|
||||
keep the operator's spelling (matching the ``citext`` behaviour of the
|
||||
reviewed migration).
|
||||
The login identifier is the email address, not this. A username is kept only
|
||||
as a short handle for audit lines and operator-facing lists, so it is
|
||||
optional: the reviewed migration declares ``users.username`` nullable
|
||||
precisely so an account is not forced to carry two identifiers. Case
|
||||
insensitivity is handled by :func:`username_key`, matching the ``citext``
|
||||
behaviour of that migration.
|
||||
"""
|
||||
|
||||
if value is None:
|
||||
return None
|
||||
text = value.strip()
|
||||
if not text:
|
||||
return None
|
||||
if not _USERNAME_RE.fullmatch(text):
|
||||
raise InputError(
|
||||
"invalid_username",
|
||||
@@ -156,10 +189,24 @@ def normalize_username(value: str) -> str:
|
||||
return text
|
||||
|
||||
|
||||
def username_key(value: str) -> str:
|
||||
"""Return the case-insensitive lookup key for a username."""
|
||||
def username_key(value: str | None) -> str | None:
|
||||
"""Return the case-insensitive lookup key for an optional username."""
|
||||
|
||||
return normalize_username(value).casefold()
|
||||
normalized = normalize_username(value)
|
||||
return normalized.casefold() if normalized is not None else None
|
||||
|
||||
|
||||
def email_key(value: str) -> str:
|
||||
"""Return the case-insensitive lookup key for the login identifier.
|
||||
|
||||
The control plane declares ``users.email`` as ``citext NOT NULL UNIQUE``, so
|
||||
the address is already a case-insensitively unique login key and needs no
|
||||
migration. Lowercasing here matches that behaviour exactly; note it is a
|
||||
*key* computation only, and the stored address keeps the case the operator
|
||||
typed so it is not silently rewritten in the database.
|
||||
"""
|
||||
|
||||
return normalize_email(value).casefold()
|
||||
|
||||
|
||||
def normalize_display_name(value: str) -> str:
|
||||
@@ -309,16 +356,21 @@ class CredentialRecord:
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class User:
|
||||
"""A canonical account. Contains no password material of any kind."""
|
||||
"""A canonical account. Contains no password material of any kind.
|
||||
|
||||
``email`` is the login identifier and is required. ``username`` is an
|
||||
optional short handle for audit lines; it is not a login path, and the
|
||||
service never authenticates against it.
|
||||
"""
|
||||
|
||||
id: UUID
|
||||
username: str
|
||||
username_key: str
|
||||
username: str | None
|
||||
username_key: str | None
|
||||
display_name: str
|
||||
role: RoleName
|
||||
status: AccountStatus
|
||||
email: str
|
||||
organization_id: UUID | None = None
|
||||
email: str | None = None
|
||||
created_at: datetime | None = None
|
||||
updated_at: datetime | None = None
|
||||
version: int = 1
|
||||
@@ -327,11 +379,31 @@ class User:
|
||||
object.__setattr__(self, "role", RoleName(self.role))
|
||||
object.__setattr__(self, "status", AccountStatus(self.status))
|
||||
object.__setattr__(self, "username_key", username_key(self.username))
|
||||
object.__setattr__(self, "username", normalize_username(self.username))
|
||||
# The address is required and normalised on every construction, so an
|
||||
# account can never be built with a missing or malformed login key.
|
||||
object.__setattr__(self, "email", normalize_email(self.email))
|
||||
|
||||
@property
|
||||
def is_admin(self) -> bool:
|
||||
return self.role.value in ADMIN_ROLES
|
||||
|
||||
@property
|
||||
def is_email_verified(self) -> bool:
|
||||
"""Whether the account has proved it can receive mail at its address.
|
||||
|
||||
Derived from status rather than stored twice: the control plane has one
|
||||
status column and duplicating the fact would let the two disagree.
|
||||
"""
|
||||
|
||||
return self.status is not AccountStatus.INVITED
|
||||
|
||||
@property
|
||||
def audit_handle(self) -> str:
|
||||
"""A stable, human-readable identifier for audit records."""
|
||||
|
||||
return self.username or self.email
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SessionRecord:
|
||||
@@ -407,17 +479,23 @@ class AccountView:
|
||||
"""The only account representation that may cross the API boundary."""
|
||||
|
||||
id: UUID
|
||||
username: str
|
||||
email: str
|
||||
display_name: str
|
||||
role: RoleName
|
||||
status: AccountStatus
|
||||
credential: CredentialSummary
|
||||
username: str | None = None
|
||||
organization_id: UUID | None = None
|
||||
email: str | None = None
|
||||
created_at: datetime | None = None
|
||||
updated_at: datetime | None = None
|
||||
active_session_count: int = 0
|
||||
|
||||
@property
|
||||
def is_email_verified(self) -> bool:
|
||||
"""Whether the address has been confirmed. Derived, never stored twice."""
|
||||
|
||||
return self.status is not AccountStatus.INVITED
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Organization:
|
||||
@@ -461,6 +539,38 @@ class OrganizationView:
|
||||
updated_at: datetime | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class VerificationToken:
|
||||
"""A single-use token proving control of an email address.
|
||||
|
||||
Maps onto the control plane's ``bootstrap_tokens`` table, which is already
|
||||
shaped for this: ``(id, user_id, token_hash, expires_at, consumed_at)``.
|
||||
That table existed and was unused, so email validation needs no migration.
|
||||
|
||||
``token_hash`` is ``repr=False`` and the plaintext is never stored, never
|
||||
returned by an endpoint, and never logged. The link that carries the
|
||||
plaintext is emailed and exists only in the operator's inbox and in the
|
||||
transport's own buffer.
|
||||
"""
|
||||
|
||||
id: UUID
|
||||
user_id: UUID
|
||||
token_hash: str = field(repr=False)
|
||||
expires_at: datetime
|
||||
created_at: datetime
|
||||
consumed_at: datetime | None = None
|
||||
purpose: str = "email_verification"
|
||||
|
||||
def is_consumed(self) -> bool:
|
||||
return self.consumed_at is not None
|
||||
|
||||
def is_expired(self, now: datetime) -> bool:
|
||||
return now >= self.expires_at
|
||||
|
||||
def is_usable(self, now: datetime) -> bool:
|
||||
return not self.is_consumed() and not self.is_expired(now)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Principal:
|
||||
"""A resolved caller identity. Never built from browser-supplied values."""
|
||||
@@ -507,6 +617,8 @@ __all__ = [
|
||||
"MAX_USERNAME_LENGTH",
|
||||
"MIN_ORGANIZATION_NAME_LENGTH",
|
||||
"MIN_USERNAME_LENGTH",
|
||||
"VERIFICATION_PURPOSE_EMAIL",
|
||||
"VERIFICATION_TOKEN_TTL_SECONDS",
|
||||
"AccountStatus",
|
||||
"AccountView",
|
||||
"AuditEvent",
|
||||
@@ -523,6 +635,8 @@ __all__ = [
|
||||
"RoleName",
|
||||
"SessionRecord",
|
||||
"User",
|
||||
"VerificationToken",
|
||||
"email_key",
|
||||
"normalize_display_name",
|
||||
"normalize_email",
|
||||
"normalize_organization_name",
|
||||
|
||||
@@ -26,6 +26,8 @@ from .models import (
|
||||
OrganizationStatus,
|
||||
SessionRecord,
|
||||
User,
|
||||
VerificationToken,
|
||||
email_key,
|
||||
username_key,
|
||||
)
|
||||
|
||||
@@ -42,8 +44,16 @@ class AccountRepository(Protocol):
|
||||
|
||||
def get_user(self, user_id: UUID) -> User | None: ...
|
||||
|
||||
def get_user_by_email(self, key: str) -> User | None:
|
||||
"""Look up an account by its case-insensitive email key.
|
||||
|
||||
This is the login path. ``users.email`` is ``citext NOT NULL UNIQUE`` in
|
||||
the control plane, so the address is already a case-insensitively
|
||||
unique identifier and no separate username lookup is used to sign in.
|
||||
"""
|
||||
|
||||
def get_user_by_username(self, key: str) -> User | None:
|
||||
"""Look up an account by its case-insensitive username key."""
|
||||
"""Look up an account by its optional display handle. Not a login path."""
|
||||
|
||||
def count_users(self, organization_id: UUID | None = None) -> int:
|
||||
"""Count accounts, optionally within one organization."""
|
||||
@@ -104,6 +114,25 @@ class AccountRepository(Protocol):
|
||||
) -> int:
|
||||
"""Revoke live sessions for one account and return how many were revoked."""
|
||||
|
||||
def create_verification_token(self, token: VerificationToken) -> VerificationToken:
|
||||
"""Store a single-use token. Only the hash is written."""
|
||||
|
||||
def get_live_verification_token(self, token_hash: str) -> VerificationToken | None:
|
||||
"""Return an unconsumed, unexpired token for this purpose, if any.
|
||||
|
||||
Used to reject a second use of the same link without revealing whether
|
||||
the first use succeeded.
|
||||
"""
|
||||
|
||||
def mark_verification_token_consumed(self, token_id: UUID, *, now: datetime) -> bool:
|
||||
"""Consume a token. Returns False if it was already consumed.
|
||||
|
||||
The compare-and-set is what makes the token single-use under
|
||||
concurrency: two simultaneous requests cannot both observe success.
|
||||
"""
|
||||
|
||||
def delete_expired_verification_tokens(self, *, now: datetime) -> int: ...
|
||||
|
||||
def append_audit_event(self, event: AuditEvent) -> AuditEvent: ...
|
||||
|
||||
def list_audit_events(
|
||||
@@ -134,6 +163,7 @@ class InMemoryAccountRepository:
|
||||
self._credentials: dict[UUID, CredentialRecord] = {}
|
||||
self._sessions: dict[UUID, SessionRecord] = {}
|
||||
self._sessions_by_token: dict[str, UUID] = {}
|
||||
self._verification_tokens: dict[UUID, VerificationToken] = {}
|
||||
self._audit: list[AuditEvent] = []
|
||||
|
||||
# -- users ---------------------------------------------------------------
|
||||
@@ -165,9 +195,16 @@ class InMemoryAccountRepository:
|
||||
with self._lock:
|
||||
return self._users.get(user_id)
|
||||
|
||||
def get_user_by_email(self, key: str) -> User | None:
|
||||
with self._lock:
|
||||
for user in self._users.values():
|
||||
if email_key(user.email) == key.casefold():
|
||||
return user
|
||||
return None
|
||||
|
||||
def get_user_by_username(self, key: str) -> User | None:
|
||||
with self._lock:
|
||||
user_id = self._users_by_key.get(username_key(key))
|
||||
user_id = self._users_by_key.get(username_key(key)) if username_key(key) else None
|
||||
return self._users.get(user_id) if user_id else None
|
||||
|
||||
def count_users(self, organization_id: UUID | None = None) -> int:
|
||||
@@ -322,6 +359,35 @@ class InMemoryAccountRepository:
|
||||
|
||||
# -- audit ---------------------------------------------------------------
|
||||
|
||||
# -- email verification --------------------------------------------------
|
||||
|
||||
def create_verification_token(self, token: VerificationToken) -> VerificationToken:
|
||||
with self._lock:
|
||||
self._verification_tokens[token.id] = token
|
||||
return token
|
||||
|
||||
def get_live_verification_token(self, token_hash: str) -> VerificationToken | None:
|
||||
with self._lock:
|
||||
for token in self._verification_tokens.values():
|
||||
if token.token_hash == token_hash and not token.is_consumed():
|
||||
return token
|
||||
return None
|
||||
|
||||
def mark_verification_token_consumed(self, token_id: UUID, *, now: datetime) -> bool:
|
||||
with self._lock:
|
||||
token = self._verification_tokens.get(token_id)
|
||||
if token is None or token.is_consumed():
|
||||
return False
|
||||
self._verification_tokens[token_id] = replace(token, consumed_at=now)
|
||||
return True
|
||||
|
||||
def delete_expired_verification_tokens(self, *, now: datetime) -> int:
|
||||
with self._lock:
|
||||
stale = [token_id for token_id, token in self._verification_tokens.items() if token.is_expired(now)]
|
||||
for token_id in stale:
|
||||
del self._verification_tokens[token_id]
|
||||
return len(stale)
|
||||
|
||||
def append_audit_event(self, event: AuditEvent) -> AuditEvent:
|
||||
with self._lock:
|
||||
self._audit.append(event)
|
||||
|
||||
@@ -7,7 +7,7 @@ returned in a body.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, Depends, Request, Response
|
||||
from fastapi import APIRouter, Depends, Query, Request, Response
|
||||
|
||||
from .app import (
|
||||
clear_csrf_cookie,
|
||||
@@ -27,8 +27,12 @@ from .schemas import (
|
||||
LoginRequest,
|
||||
LoginResponse,
|
||||
LogoutResponse,
|
||||
ResendVerificationRequest,
|
||||
ResendVerificationResponse,
|
||||
SessionUserResponse,
|
||||
UserResponse,
|
||||
VerifyEmailRequest,
|
||||
VerifyEmailResponse,
|
||||
)
|
||||
from .service import AccountService
|
||||
|
||||
@@ -61,7 +65,7 @@ def login(payload: LoginRequest, request: Request, response: Response) -> LoginR
|
||||
settings = get_settings(request)
|
||||
service: AccountService = get_service(request)
|
||||
result = service.authenticate(
|
||||
username=payload.username,
|
||||
email=payload.email,
|
||||
password=payload.password,
|
||||
request_id=request_id(request),
|
||||
)
|
||||
@@ -134,6 +138,56 @@ def logout_everywhere(
|
||||
return LogoutResponse(revoked_sessions=revoked)
|
||||
|
||||
|
||||
@router.post("/verify-email", response_model=VerifyEmailResponse)
|
||||
def verify_email(payload: VerifyEmailRequest, request: Request) -> VerifyEmailResponse:
|
||||
"""Consume a verification link and activate the account.
|
||||
|
||||
The token is a bearer capability taken from a mailed link, so it is not
|
||||
echoed back and the response reveals nothing beyond the account it belongs
|
||||
to. It may be a query parameter when the link is followed in a browser.
|
||||
"""
|
||||
|
||||
service: AccountService = get_service(request)
|
||||
account = service.verify_email(payload.token, request_id=request_id(request))
|
||||
return VerifyEmailResponse(
|
||||
account=UserResponse.from_view(account),
|
||||
message="Your email address is confirmed. You can sign in now.",
|
||||
)
|
||||
|
||||
|
||||
@router.get("/verify-email", response_model=VerifyEmailResponse)
|
||||
def verify_email_from_link(
|
||||
request: Request,
|
||||
token: str = Query(min_length=1, max_length=512),
|
||||
) -> VerifyEmailResponse:
|
||||
"""The same operation for a link followed in a browser."""
|
||||
|
||||
service: AccountService = get_service(request)
|
||||
account = service.verify_email(token, request_id=request_id(request))
|
||||
return VerifyEmailResponse(
|
||||
account=UserResponse.from_view(account),
|
||||
message="Your email address is confirmed. You can sign in now.",
|
||||
)
|
||||
|
||||
|
||||
@router.post("/resend-verification", response_model=ResendVerificationResponse)
|
||||
def resend_verification(
|
||||
payload: ResendVerificationRequest,
|
||||
request: Request,
|
||||
) -> ResendVerificationResponse:
|
||||
"""Re-mail a verification link.
|
||||
|
||||
The response is the same whether an account was found or not, so this cannot
|
||||
be used to discover which addresses have accounts.
|
||||
"""
|
||||
|
||||
service: AccountService = get_service(request)
|
||||
service.resend_email_verification(payload.email, request_id=request_id(request))
|
||||
return ResendVerificationResponse(
|
||||
message="If that address has an account awaiting confirmation, a new link is on its way."
|
||||
)
|
||||
|
||||
|
||||
@router.get("/me", response_model=SessionUserResponse)
|
||||
def me(
|
||||
request: Request,
|
||||
|
||||
@@ -78,12 +78,14 @@ class CredentialSummaryResponse(BaseModel):
|
||||
|
||||
class UserResponse(BaseModel):
|
||||
id: UUID
|
||||
username: str
|
||||
email: str
|
||||
username: str | None = None
|
||||
display_name: str
|
||||
role: RoleName
|
||||
status: AccountStatus
|
||||
organization_id: UUID | None = None
|
||||
email: str | None = None
|
||||
is_email_verified: bool = True
|
||||
created_at: datetime | None = None
|
||||
updated_at: datetime | None = None
|
||||
|
||||
@@ -91,12 +93,12 @@ class UserResponse(BaseModel):
|
||||
def from_view(cls, view: AccountView) -> UserResponse:
|
||||
return cls(
|
||||
id=view.id,
|
||||
email=view.email,
|
||||
username=view.username,
|
||||
display_name=view.display_name,
|
||||
role=view.role,
|
||||
status=view.status,
|
||||
organization_id=view.organization_id,
|
||||
email=view.email,
|
||||
created_at=view.created_at,
|
||||
updated_at=view.updated_at,
|
||||
)
|
||||
@@ -194,10 +196,50 @@ class AuditListResponse(BaseModel):
|
||||
|
||||
|
||||
class LoginRequest(RequestModel):
|
||||
username: str = Field(min_length=1, max_length=64)
|
||||
"""Sign-in payload.
|
||||
|
||||
The identifier is the email address, not a username:
|
||||
``users.email`` is ``citext NOT NULL UNIQUE`` in the control plane, so it is
|
||||
already a case-insensitively unique login key and no second identifier is
|
||||
needed.
|
||||
"""
|
||||
|
||||
email: str = Field(min_length=3, max_length=254)
|
||||
password: str = Field(min_length=1, max_length=1024)
|
||||
|
||||
|
||||
class VerifyEmailRequest(RequestModel):
|
||||
"""Consume a mailed verification link.
|
||||
|
||||
``extra="forbid"`` matters here: a link is a bearer capability, and quietly
|
||||
ignoring unexpected fields would make a malformed link look like a valid
|
||||
one.
|
||||
"""
|
||||
|
||||
token: str = Field(min_length=1, max_length=512)
|
||||
|
||||
|
||||
class ResendVerificationRequest(RequestModel):
|
||||
"""Ask for another verification link.
|
||||
|
||||
Takes an address rather than a session because an unverified account cannot
|
||||
hold one. The response is identical whether or not an account exists.
|
||||
"""
|
||||
|
||||
email: str = Field(min_length=3, max_length=254)
|
||||
|
||||
|
||||
class VerifyEmailResponse(BaseModel):
|
||||
account: UserResponse
|
||||
message: str
|
||||
|
||||
|
||||
class ResendVerificationResponse(BaseModel):
|
||||
"""Deliberately says nothing about whether an account exists."""
|
||||
|
||||
message: str
|
||||
|
||||
|
||||
class ChangeFirstPasswordRequest(RequestModel):
|
||||
current_password: str = Field(min_length=1, max_length=1024)
|
||||
new_password: str = Field(min_length=1, max_length=1024)
|
||||
@@ -217,7 +259,14 @@ class CreateAccountRequest(RequestModel):
|
||||
the authoritative store is in use.
|
||||
"""
|
||||
|
||||
username: str = Field(min_length=1, max_length=64)
|
||||
username: str | None = Field(
|
||||
default=None,
|
||||
max_length=64,
|
||||
description=(
|
||||
"Optional short handle for audit lines. Not a login path: the email "
|
||||
"is the identifier, so this is never asked for by sign-in."
|
||||
),
|
||||
)
|
||||
display_name: str = Field(min_length=1, max_length=120)
|
||||
email: str = Field(min_length=3, max_length=254)
|
||||
enrollment_mode: EnrollmentMode = EnrollmentMode.OIDC
|
||||
@@ -351,6 +400,8 @@ __all__ = [
|
||||
"OrganizationResponse",
|
||||
"OrganizationStatusChangeResponse",
|
||||
"OrganizationSummaryResponse",
|
||||
"ResendVerificationRequest",
|
||||
"ResendVerificationResponse",
|
||||
"RevokeSessionsResponse",
|
||||
"SessionUserResponse",
|
||||
"SetAccountStatusRequest",
|
||||
@@ -358,4 +409,6 @@ __all__ = [
|
||||
"StatusChangeResponse",
|
||||
"UpdateOrganizationRequest",
|
||||
"UserResponse",
|
||||
"VerifyEmailRequest",
|
||||
"VerifyEmailResponse",
|
||||
]
|
||||
|
||||
@@ -41,6 +41,7 @@ from uuid import UUID, uuid4
|
||||
from .config import Settings
|
||||
from .errors import (
|
||||
ConflictError,
|
||||
EmailNotVerifiedError,
|
||||
EnrollmentRefusedError,
|
||||
ForbiddenError,
|
||||
InputError,
|
||||
@@ -50,10 +51,15 @@ from .errors import (
|
||||
PasswordChangeRequiredError,
|
||||
PasswordPolicyError,
|
||||
StateTransitionError,
|
||||
VerificationDeliveryError,
|
||||
VerificationFailedError,
|
||||
)
|
||||
from .mailer import Mailer, MailerError, build_mailer
|
||||
from .models import (
|
||||
ALLOWED_ORGANIZATION_TRANSITIONS,
|
||||
ALLOWED_STATUS_TRANSITIONS,
|
||||
VERIFICATION_PURPOSE_EMAIL,
|
||||
VERIFICATION_TOKEN_TTL_SECONDS,
|
||||
AccountStatus,
|
||||
AccountView,
|
||||
AuditEvent,
|
||||
@@ -70,6 +76,8 @@ from .models import (
|
||||
RoleName,
|
||||
SessionRecord,
|
||||
User,
|
||||
VerificationToken,
|
||||
email_key,
|
||||
normalize_display_name,
|
||||
normalize_email,
|
||||
normalize_organization_name,
|
||||
@@ -108,6 +116,8 @@ ACTION_AUTH_LOGIN = "auth.login"
|
||||
ACTION_AUTH_FIRST_PASSWORD_CHANGED = "auth.first_password_changed"
|
||||
ACTION_AUTH_LOGOUT = "auth.logout"
|
||||
ACTION_AUTH_LOGOUT_EVERYWHERE = "auth.logout_everywhere"
|
||||
ACTION_EMAIL_VERIFICATION_ISSUED = "email.verification_issued"
|
||||
ACTION_EMAIL_VERIFIED = "email.verified"
|
||||
ACTION_SESSION_REVOKED = "session.revoked"
|
||||
ACTION_SESSION_EXPIRED = "session.expired"
|
||||
ACTION_SESSION_INVALIDATED = "session.invalidated"
|
||||
@@ -153,10 +163,12 @@ class AccountService:
|
||||
*,
|
||||
clock: Callable[[], datetime] = utc_now,
|
||||
password_service: PasswordService | None = None,
|
||||
mailer: Mailer | None = None,
|
||||
) -> None:
|
||||
self._repository = repository
|
||||
self._settings = settings
|
||||
self._clock = clock
|
||||
self._mailer: Mailer = mailer or build_mailer(settings)
|
||||
self._passwords = password_service or PasswordService(
|
||||
memory_cost_kib=settings.argon2_memory_cost_kib,
|
||||
time_cost=settings.argon2_time_cost,
|
||||
@@ -197,6 +209,7 @@ class AccountService:
|
||||
actor: Principal | None = None,
|
||||
operator: OperatorContext | None = None,
|
||||
request_id: str | None = None,
|
||||
require_email_verification: bool = False,
|
||||
) -> AccountView:
|
||||
"""Create an account through one of the two enrollment modes.
|
||||
|
||||
@@ -231,6 +244,7 @@ class AccountService:
|
||||
organization_id=organization_id,
|
||||
actor=actor,
|
||||
request_id=request_id,
|
||||
require_email_verification=require_email_verification,
|
||||
)
|
||||
|
||||
def _provision_local_account(
|
||||
@@ -244,6 +258,7 @@ class AccountService:
|
||||
password: str | None,
|
||||
operator: OperatorContext | None,
|
||||
request_id: str | None,
|
||||
require_email_verification: bool = False,
|
||||
) -> AccountView:
|
||||
"""Operator-only enrollment with a one-time local password."""
|
||||
|
||||
@@ -266,7 +281,11 @@ class AccountService:
|
||||
credential_type=CredentialType.LOCAL_PASSWORD,
|
||||
password_hash=password_hash,
|
||||
must_change_password=True,
|
||||
initial_status=(AccountStatus.INVITED if require_email_verification else AccountStatus.ACTIVE),
|
||||
)
|
||||
if require_email_verification:
|
||||
link = self.issue_email_verification(user.id, request_id=request_id)
|
||||
self._deliver(user.email, link)
|
||||
self._audit(
|
||||
action=ACTION_ACCOUNT_PROVISIONED_LOCAL,
|
||||
outcome="success",
|
||||
@@ -328,7 +347,12 @@ class AccountService:
|
||||
credential_type=CredentialType.LOCAL_PASSWORD,
|
||||
password_hash=self._passwords.hash_password(temporary_password),
|
||||
must_change_password=True,
|
||||
# Administrator-provisioned accounts must prove they can receive
|
||||
# mail before they can sign in.
|
||||
initial_status=AccountStatus.INVITED,
|
||||
)
|
||||
link = self.issue_email_verification(user.id, actor=actor, request_id=request_id)
|
||||
self._deliver(user.email, link)
|
||||
self._audit(
|
||||
action=ACTION_ACCOUNT_PROVISIONED_LOCAL,
|
||||
outcome="success",
|
||||
@@ -364,6 +388,7 @@ class AccountService:
|
||||
organization_id: UUID | None = None,
|
||||
actor: Principal | None = None,
|
||||
request_id: str | None = None,
|
||||
require_email_verification: bool = True,
|
||||
) -> AccountView:
|
||||
"""Administrator-driven OIDC enrollment. No local password is created."""
|
||||
|
||||
@@ -381,7 +406,11 @@ class AccountService:
|
||||
credential_type=CredentialType.OIDC,
|
||||
password_hash=None,
|
||||
must_change_password=False,
|
||||
initial_status=(AccountStatus.INVITED if require_email_verification else AccountStatus.ACTIVE),
|
||||
)
|
||||
if require_email_verification:
|
||||
link = self.issue_email_verification(user.id, actor=actor, request_id=request_id)
|
||||
self._deliver(user.email, link)
|
||||
self._audit(
|
||||
action=ACTION_ACCOUNT_CREATED,
|
||||
outcome="success",
|
||||
@@ -412,18 +441,18 @@ class AccountService:
|
||||
credential_type: CredentialType,
|
||||
password_hash: str | None,
|
||||
must_change_password: bool,
|
||||
initial_status: AccountStatus = AccountStatus.ACTIVE,
|
||||
) -> tuple[User, CredentialRecord]:
|
||||
now = self._now()
|
||||
canonical_username = normalize_username(username)
|
||||
user = User(
|
||||
id=uuid4(),
|
||||
username=canonical_username,
|
||||
username_key=username_key(canonical_username),
|
||||
username=normalize_username(username),
|
||||
username_key=username_key(username),
|
||||
display_name=normalize_display_name(display_name),
|
||||
role=_role(role),
|
||||
status=AccountStatus.ACTIVE,
|
||||
organization_id=organization_id,
|
||||
status=initial_status,
|
||||
email=normalize_email(email),
|
||||
organization_id=organization_id,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
@@ -442,15 +471,23 @@ class AccountService:
|
||||
|
||||
# -- authentication ------------------------------------------------------
|
||||
|
||||
def authenticate(self, *, username: str, password: str, request_id: str | None = None) -> AuthenticationResult:
|
||||
def authenticate(self, *, email: str, password: str, request_id: str | None = None) -> AuthenticationResult:
|
||||
"""Authenticate a local password and issue a server-side session.
|
||||
|
||||
The identifier is the email address, because ``users.email`` is
|
||||
``citext NOT NULL UNIQUE`` in the control plane and is therefore already
|
||||
a case-insensitively unique login key.
|
||||
|
||||
Every rejection path raises the same :class:`InvalidCredentialsError` so
|
||||
that the response cannot be used to enumerate accounts.
|
||||
that the response cannot be used to enumerate accounts. The one exception
|
||||
is an account whose email is not yet verified, which is reported only
|
||||
*after* the password has been proven correct. That is safe and is the
|
||||
difference between a user being told to check their inbox and a user
|
||||
being told their password is wrong when it is not.
|
||||
"""
|
||||
|
||||
now = self._now()
|
||||
key = _safe_username_key(username)
|
||||
key = _safe_email_key(email)
|
||||
if key is None or not isinstance(password, str) or not password:
|
||||
self._passwords.verify_dummy(password if isinstance(password, str) else "")
|
||||
self._audit(
|
||||
@@ -461,7 +498,7 @@ class AccountService:
|
||||
)
|
||||
raise InvalidCredentialsError()
|
||||
|
||||
user = self._repository.get_user_by_username(key)
|
||||
user = self._repository.get_user_by_email(key)
|
||||
if user is None:
|
||||
self._passwords.verify_dummy(password)
|
||||
self._audit(
|
||||
@@ -498,7 +535,10 @@ class AccountService:
|
||||
)
|
||||
raise InvalidCredentialsError()
|
||||
|
||||
if user.status is not AccountStatus.ACTIVE:
|
||||
if user.status is not AccountStatus.ACTIVE and user.status is not AccountStatus.INVITED:
|
||||
# ``invited`` is excluded here and handled after the password has
|
||||
# been proven, so an unconfirmed address is not distinguishable from
|
||||
# an unknown one to someone who does not have the password.
|
||||
self._audit(
|
||||
action=ACTION_AUTH_LOGIN,
|
||||
outcome="denied",
|
||||
@@ -543,6 +583,21 @@ class AccountService:
|
||||
)
|
||||
raise InvalidCredentialsError()
|
||||
|
||||
# The password is correct from here on, so it is now safe to say that the
|
||||
# address is unconfirmed. Doing this check any earlier would turn the
|
||||
# login form into an oracle for which addresses have accounts.
|
||||
if user.status is AccountStatus.INVITED:
|
||||
self._repository.save_credential(replace(credential, last_login_at=now, updated_at=now))
|
||||
self._audit(
|
||||
action=ACTION_AUTH_LOGIN,
|
||||
outcome="denied",
|
||||
target_user_id=user.id,
|
||||
organization_id=user.organization_id,
|
||||
request_id=request_id,
|
||||
detail={"reason": "email_not_verified", "status": user.status},
|
||||
)
|
||||
raise EmailNotVerifiedError()
|
||||
|
||||
self._repository.save_credential(
|
||||
replace(credential, failed_login_count=0, locked_until=None, last_login_at=now, updated_at=now)
|
||||
)
|
||||
@@ -643,6 +698,141 @@ class AccountService:
|
||||
LOGGER.info("first password changed user_id=%s revoked_sessions=%s", user.id, revoked)
|
||||
return AuthenticationResult(user=user, session=session)
|
||||
|
||||
# -- email verification ---------------------------------------------------
|
||||
|
||||
def issue_email_verification(
|
||||
self,
|
||||
user_id: UUID,
|
||||
*,
|
||||
actor: Principal | None = None,
|
||||
request_id: str | None = None,
|
||||
) -> str:
|
||||
"""Mint a single-use verification link and mail it. Returns the link.
|
||||
|
||||
The returned link is the only place the plaintext token exists. It is
|
||||
returned to the caller so the development mail transport, the audit
|
||||
record, and a test can all reach it; no endpoint may pass it through.
|
||||
"""
|
||||
|
||||
user = self._require_user(user_id)
|
||||
if user.status is not AccountStatus.INVITED:
|
||||
raise StateTransitionError(
|
||||
"already_verified",
|
||||
"That account does not need email verification.",
|
||||
)
|
||||
now = self._now()
|
||||
self._repository.delete_expired_verification_tokens(now=now)
|
||||
token = generate_token(32)
|
||||
self._repository.create_verification_token(
|
||||
VerificationToken(
|
||||
id=uuid4(),
|
||||
user_id=user.id,
|
||||
token_hash=hash_session_token(token),
|
||||
expires_at=now + timedelta(seconds=VERIFICATION_TOKEN_TTL_SECONDS),
|
||||
created_at=now,
|
||||
purpose=VERIFICATION_PURPOSE_EMAIL,
|
||||
)
|
||||
)
|
||||
self._audit(
|
||||
action=ACTION_EMAIL_VERIFICATION_ISSUED,
|
||||
outcome="success",
|
||||
actor=actor,
|
||||
target_user_id=user.id,
|
||||
organization_id=user.organization_id,
|
||||
request_id=request_id,
|
||||
detail={"username": user.audit_handle},
|
||||
)
|
||||
LOGGER.info("email verification issued user_id=%s", user.id)
|
||||
return self._verification_link(token)
|
||||
|
||||
def verify_email(self, token: str, *, request_id: str | None = None) -> AccountView:
|
||||
"""Consume a verification link and activate the account.
|
||||
|
||||
Every failure - unknown, already used, expired - raises the same error,
|
||||
so this cannot be used to discover which links exist.
|
||||
"""
|
||||
|
||||
now = self._now()
|
||||
if not isinstance(token, str) or not token:
|
||||
raise VerificationFailedError()
|
||||
record = self._repository.get_live_verification_token(hash_session_token(token))
|
||||
if record is None or not record.is_usable(now) or record.purpose != VERIFICATION_PURPOSE_EMAIL:
|
||||
raise VerificationFailedError()
|
||||
# Compare-and-set: two concurrent uses cannot both observe success.
|
||||
if not self._repository.mark_verification_token_consumed(record.id, now=now):
|
||||
raise VerificationFailedError()
|
||||
|
||||
user = self._require_user(record.user_id)
|
||||
if AccountStatus.ACTIVE not in ALLOWED_STATUS_TRANSITIONS.get(user.status, frozenset()):
|
||||
raise VerificationFailedError()
|
||||
updated = replace(
|
||||
user,
|
||||
status=AccountStatus.ACTIVE,
|
||||
updated_at=now,
|
||||
version=user.version + 1,
|
||||
)
|
||||
self._repository.save_user(updated)
|
||||
self._audit(
|
||||
action=ACTION_EMAIL_VERIFIED,
|
||||
outcome="success",
|
||||
target_user_id=user.id,
|
||||
organization_id=user.organization_id,
|
||||
request_id=request_id,
|
||||
detail={"username": user.audit_handle, "status": updated.status},
|
||||
)
|
||||
LOGGER.info("email verified user_id=%s", user.id)
|
||||
return self._view(updated, now=now)
|
||||
|
||||
def resend_email_verification(
|
||||
self,
|
||||
email: str,
|
||||
*,
|
||||
request_id: str | None = None,
|
||||
) -> bool:
|
||||
"""Re-mail a verification link. Returns True when one was sent.
|
||||
|
||||
Takes an address rather than a principal, because an unverified account
|
||||
cannot hold a session and so cannot ask for itself. The response to the
|
||||
HTTP layer is always the same, so this cannot be used to discover which
|
||||
addresses have accounts.
|
||||
"""
|
||||
|
||||
key = _safe_email_key(email)
|
||||
user = self._repository.get_user_by_email(key) if key else None
|
||||
if user is None or user.status is not AccountStatus.INVITED:
|
||||
# Nothing is sent and nothing is said. The caller reports the same
|
||||
# outcome either way.
|
||||
return False
|
||||
link = self.issue_email_verification(user.id, request_id=request_id)
|
||||
self._deliver(user.email, link)
|
||||
return True
|
||||
|
||||
def _verification_link(self, token: str) -> str:
|
||||
base = self._settings.verification_base_url
|
||||
return f"{base}/verify-email?token={token}" if base else f"/verify-email?token={token}"
|
||||
|
||||
def _deliver(self, recipient: str, link: str) -> None:
|
||||
"""Send the message, failing loudly without exposing the link.
|
||||
|
||||
A delivery failure must not look like success to the caller, but the
|
||||
reason must not contain the token either.
|
||||
"""
|
||||
|
||||
subject = "Confirm your UmbraTrans email address"
|
||||
body = (
|
||||
"An account was created for this address on UmbraTrans.\n\n"
|
||||
"Confirm the address to finish setting it up:\n\n"
|
||||
f" {link}\n\n"
|
||||
"If you did not expect this, ignore this message. The account cannot "
|
||||
"be used until the address is confirmed, and nobody can sign in as it "
|
||||
"before then.\n"
|
||||
)
|
||||
try:
|
||||
self._mailer.send(to=recipient, subject=subject, body=body)
|
||||
except MailerError:
|
||||
LOGGER.error("verification mail could not be delivered transport=%s", self._mailer.transport_name)
|
||||
raise VerificationDeliveryError() from None
|
||||
|
||||
# -- sessions ------------------------------------------------------------
|
||||
|
||||
def resolve_session(self, token: str, *, request_id: str | None = None) -> Principal:
|
||||
@@ -1262,13 +1452,13 @@ class AccountService:
|
||||
)
|
||||
return AccountView(
|
||||
id=user.id,
|
||||
username=user.username,
|
||||
email=user.email,
|
||||
display_name=user.display_name,
|
||||
role=user.role,
|
||||
status=user.status,
|
||||
credential=summary,
|
||||
username=user.username,
|
||||
organization_id=user.organization_id,
|
||||
email=user.email,
|
||||
created_at=user.created_at,
|
||||
updated_at=user.updated_at,
|
||||
active_session_count=self._repository.count_sessions(user.id),
|
||||
@@ -1336,6 +1526,21 @@ def _role(value: str) -> RoleName:
|
||||
raise InputError("invalid_role", "Unknown role.") from error
|
||||
|
||||
|
||||
def _safe_email_key(email: str) -> str | None:
|
||||
"""Return a lookup key for a submitted address, or None if it is malformed.
|
||||
|
||||
A malformed identifier is treated as "no such account" rather than raising,
|
||||
so a probe cannot distinguish a bad address from an absent one.
|
||||
"""
|
||||
|
||||
if not isinstance(email, str):
|
||||
return None
|
||||
try:
|
||||
return email_key(email)
|
||||
except InputError:
|
||||
return None
|
||||
|
||||
|
||||
def _safe_username_key(username: str) -> str | None:
|
||||
if not isinstance(username, str):
|
||||
return None
|
||||
|
||||
@@ -13,7 +13,7 @@ import re
|
||||
import secrets
|
||||
import string
|
||||
from collections.abc import Iterator
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass, replace
|
||||
from typing import Any
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
@@ -22,6 +22,7 @@ from fastapi.testclient import TestClient
|
||||
|
||||
from umbratrans_web.app import create_app
|
||||
from umbratrans_web.config import Settings, suite_settings
|
||||
from umbratrans_web.mailer import CapturingMailer
|
||||
from umbratrans_web.models import (
|
||||
AccountView,
|
||||
Organization,
|
||||
@@ -83,12 +84,17 @@ def ensure_organization(
|
||||
|
||||
@dataclass
|
||||
class SeededAccount:
|
||||
"""A test account plus the plaintext password used to create it."""
|
||||
"""A test account plus the plaintext password used to create it.
|
||||
|
||||
``email`` is the login identifier, so it is carried alongside the optional
|
||||
display handle for the same reason the API carries both.
|
||||
"""
|
||||
|
||||
view: AccountView
|
||||
password: str
|
||||
user_id: UUID
|
||||
username: str
|
||||
username: str | None
|
||||
email: str
|
||||
|
||||
|
||||
class ApiHarness:
|
||||
@@ -103,7 +109,17 @@ class ApiHarness:
|
||||
return self._client
|
||||
|
||||
def csrf(self) -> str:
|
||||
"""Return the double-submit token, bootstrapping it if needed.
|
||||
|
||||
The middleware issues the cookie on a safe request, so a test that goes
|
||||
straight to a POST would otherwise fail on its own harness. The browser
|
||||
client does the same thing on page load.
|
||||
"""
|
||||
|
||||
value = self._client.cookies.get(self._settings.csrf_cookie_name)
|
||||
if not value:
|
||||
self._client.get("/api/health")
|
||||
value = self._client.cookies.get(self._settings.csrf_cookie_name)
|
||||
assert value, "no CSRF cookie has been issued yet"
|
||||
return str(value)
|
||||
|
||||
@@ -126,11 +142,11 @@ class ApiHarness:
|
||||
headers.setdefault(self._settings.csrf_header_name, str(token))
|
||||
return self._client.get(path, headers=headers, **kwargs)
|
||||
|
||||
def login(self, username: str, password: str) -> Any:
|
||||
return self._client.post("/api/auth/login", json={"username": username, "password": password})
|
||||
def login(self, email: str, password: str) -> Any:
|
||||
return self._client.post("/api/auth/login", json={"email": email, "password": password})
|
||||
|
||||
def login_ok(self, account: SeededAccount) -> Any:
|
||||
response = self.login(account.username, account.password)
|
||||
response = self.login(account.email, account.password)
|
||||
assert response.status_code == 200, response.text
|
||||
return response
|
||||
|
||||
@@ -149,8 +165,35 @@ def repository() -> InMemoryAccountRepository:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service(repository: InMemoryAccountRepository, settings: Settings) -> AccountService:
|
||||
return AccountService(repository, settings)
|
||||
def mailer() -> CapturingMailer:
|
||||
"""Captures outbound mail so a test can read a verification link."""
|
||||
|
||||
return CapturingMailer()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service(repository: InMemoryAccountRepository, settings: Settings, mailer: CapturingMailer) -> AccountService:
|
||||
return AccountService(repository, settings, mailer=mailer)
|
||||
|
||||
|
||||
def verification_token(mailer: CapturingMailer, index: int = -1) -> str:
|
||||
"""Pull the token out of a captured verification link."""
|
||||
|
||||
message = mailer.sent[index]
|
||||
_, _, query = message.body.partition("?token=")
|
||||
return query.split()[0].strip()
|
||||
|
||||
|
||||
def verify_account_email(service: AccountService, account: SeededAccount, mailer: CapturingMailer) -> SeededAccount:
|
||||
"""Consume the verification link for an account so it becomes usable.
|
||||
|
||||
Administrator-provisioned accounts start ``invited`` and cannot sign in
|
||||
until the address is confirmed. Most tests are not about validation, so they
|
||||
say so explicitly here rather than each one re-implementing it.
|
||||
"""
|
||||
|
||||
service.verify_email(verification_token(mailer))
|
||||
return replace(account, view=service.self_view(account.user_id))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -209,7 +252,7 @@ def provision_local(
|
||||
password=candidate,
|
||||
operator=OperatorContext(source="operator-cli"),
|
||||
)
|
||||
return SeededAccount(view=view, password=candidate, user_id=view.id, username=view.username)
|
||||
return SeededAccount(view=view, password=candidate, user_id=view.id, username=view.username, email=view.email)
|
||||
|
||||
|
||||
def provision_oidc(
|
||||
@@ -232,7 +275,7 @@ def provision_oidc(
|
||||
organization_id=organization_id,
|
||||
actor=actor,
|
||||
)
|
||||
return SeededAccount(view=view, password="", user_id=view.id, username=view.username)
|
||||
return SeededAccount(view=view, password="", user_id=view.id, username=view.username, email=view.email)
|
||||
|
||||
|
||||
def activated_admin(
|
||||
@@ -241,6 +284,7 @@ def activated_admin(
|
||||
username: str = "Umbraadmin",
|
||||
role: str = str(RoleName.PLATFORM_ADMIN),
|
||||
organization_id: UUID | None = None,
|
||||
mailer: CapturingMailer | None = None,
|
||||
) -> SeededAccount:
|
||||
"""Provision an admin and complete its first-login password change.
|
||||
|
||||
@@ -249,7 +293,11 @@ def activated_admin(
|
||||
"""
|
||||
|
||||
account = provision_local(service, username=username, role=role, organization_id=organization_id)
|
||||
result = service.authenticate(username=account.username, password=account.password)
|
||||
if mailer is not None and not account.view.is_email_verified:
|
||||
# Operator-provisioned accounts are active already; this only fires for
|
||||
# a path that deliberately made the account wait for validation.
|
||||
verify_account_email(service, account, mailer)
|
||||
result = service.authenticate(email=account.email, password=account.password)
|
||||
principal = service.resolve_session(result.session.token)
|
||||
new_password = random_password(28)
|
||||
service.change_first_password(principal, current_password=account.password, new_password=new_password)
|
||||
@@ -258,6 +306,7 @@ def activated_admin(
|
||||
password=new_password,
|
||||
user_id=account.user_id,
|
||||
username=account.username,
|
||||
email=account.email,
|
||||
)
|
||||
|
||||
|
||||
@@ -292,6 +341,9 @@ __all__ = [
|
||||
"admin_principal",
|
||||
"client",
|
||||
"ensure_organization",
|
||||
"mailer",
|
||||
"verification_token",
|
||||
"verify_account_email",
|
||||
"member_principal",
|
||||
"provision_local",
|
||||
"provision_oidc",
|
||||
|
||||
@@ -8,6 +8,7 @@ from fastapi.testclient import TestClient
|
||||
from umbratrans_web.app import create_app
|
||||
from umbratrans_web.config import Environment, Settings
|
||||
from umbratrans_web.errors import ConfigurationError
|
||||
from umbratrans_web.mailer import CapturingMailer
|
||||
from umbratrans_web.repository import InMemoryAccountRepository
|
||||
from umbratrans_web.service import AccountService
|
||||
|
||||
@@ -125,8 +126,8 @@ def test_login_failures_are_generic(api: ApiHarness, service: AccountService) ->
|
||||
api.logout()
|
||||
|
||||
unknown = api.login("Nobody.Here", random_password())
|
||||
wrong = api.login(account.username, random_password())
|
||||
suspended = api.login(suspended_account.username, suspended_account.password)
|
||||
wrong = api.login(account.email, random_password())
|
||||
suspended = api.login(suspended_account.email, suspended_account.password)
|
||||
|
||||
assert unknown.status_code == wrong.status_code == suspended.status_code == 401
|
||||
assert unknown.json() == wrong.json() == suspended.json()
|
||||
@@ -141,7 +142,7 @@ def test_validation_errors_do_not_echo_the_submitted_password(client: TestClient
|
||||
# Longer than the accepted maximum, so the request is rejected by the
|
||||
# schema. FastAPI's default error body would embed this value.
|
||||
password = random_password(1200)
|
||||
response = client.post("/api/auth/login", json={"username": "Umbraadmin", "password": password})
|
||||
response = client.post("/api/auth/login", json={"email": "umbraadmin@example.test", "password": password})
|
||||
|
||||
assert response.status_code == 422
|
||||
assert response.json()["error"]["code"] == "invalid_request"
|
||||
@@ -153,7 +154,7 @@ def test_validation_errors_do_not_echo_the_submitted_password(client: TestClient
|
||||
def test_missing_credentials_are_rejected_without_detail_leak(client: TestClient) -> None:
|
||||
response = client.post("/api/auth/login", json={})
|
||||
assert response.status_code == 422
|
||||
assert response.json()["error"]["fields"] == ["password", "username"]
|
||||
assert sorted(response.json()["error"]["fields"]) == ["email", "password"]
|
||||
assert "input" not in response.text
|
||||
|
||||
|
||||
@@ -207,8 +208,8 @@ def test_first_password_change_completes_the_flow(
|
||||
# password no longer authenticates.
|
||||
assert client.get("/api/auth/me").status_code == 200
|
||||
api.logout()
|
||||
assert api.login(account.username, account.password).status_code == 401
|
||||
assert api.login(account.username, new_password).status_code == 200
|
||||
assert api.login(account.email, account.password).status_code == 401
|
||||
assert api.login(account.email, new_password).status_code == 200
|
||||
|
||||
|
||||
def test_first_password_change_validates_the_temporary_password(api: ApiHarness, service: AccountService) -> None:
|
||||
@@ -278,7 +279,7 @@ def test_logout_revokes_the_session_and_clears_the_cookie(
|
||||
assert response.json()["revoked_sessions"] == 1
|
||||
assert "Max-Age=0" in response.headers["set-cookie"] or "max-age=0" in response.headers["set-cookie"]
|
||||
assert client.get("/api/auth/me").status_code == 401
|
||||
assert api.login(account.username, account.password).status_code == 200
|
||||
assert api.login(account.email, account.password).status_code == 200
|
||||
|
||||
|
||||
def test_logout_everywhere_ends_all_sessions(
|
||||
@@ -290,7 +291,7 @@ def test_logout_everywhere_ends_all_sessions(
|
||||
|
||||
# A second browser session for the same account.
|
||||
with TestClient(client.app) as second:
|
||||
second.post("/api/auth/login", json={"username": account.username, "password": account.password})
|
||||
second.post("/api/auth/login", json={"email": account.email, "password": account.password})
|
||||
second_cookie = second.cookies[settings.session_cookie_name]
|
||||
assert second_cookie != first_cookie
|
||||
|
||||
@@ -346,15 +347,20 @@ def test_production_cookies_are_secure(settings: Settings) -> None:
|
||||
environment=Environment.PRODUCTION,
|
||||
repository_kind="postgres",
|
||||
database_dsn=TEST_DSN,
|
||||
mailer_transport="smtp",
|
||||
smtp_host="smtp.invalid",
|
||||
verification_base_url="https://umbratrans.bdmurphy.me",
|
||||
cookie_secure=True,
|
||||
)
|
||||
repository = InMemoryAccountRepository()
|
||||
service = AccountService(repository, production)
|
||||
# A capturing transport, so building the production app never attempts a
|
||||
# real SMTP connection during a unit test.
|
||||
service = AccountService(repository, production, mailer=CapturingMailer())
|
||||
app = create_app(settings=production, repository=repository, service=service)
|
||||
account = activated_admin(service, username="Umbraadmin")
|
||||
|
||||
with TestClient(app) as client:
|
||||
response = client.post("/api/auth/login", json={"username": account.username, "password": account.password})
|
||||
response = client.post("/api/auth/login", json={"email": account.email, "password": account.password})
|
||||
cookie_header = response.headers["set-cookie"]
|
||||
|
||||
assert response.status_code == 200
|
||||
@@ -368,6 +374,9 @@ def test_production_refuses_insecure_configuration() -> None:
|
||||
Settings(
|
||||
environment=Environment.PRODUCTION,
|
||||
database_dsn=TEST_DSN,
|
||||
mailer_transport="smtp",
|
||||
smtp_host="smtp.invalid",
|
||||
verification_base_url="https://umbratrans.bdmurphy.me",
|
||||
repository_kind="postgres",
|
||||
cookie_secure=False,
|
||||
)
|
||||
@@ -377,6 +386,9 @@ def test_production_refuses_insecure_configuration() -> None:
|
||||
Settings(
|
||||
environment=Environment.PRODUCTION,
|
||||
database_dsn=TEST_DSN,
|
||||
mailer_transport="smtp",
|
||||
smtp_host="smtp.invalid",
|
||||
verification_base_url="https://umbratrans.bdmurphy.me",
|
||||
repository_kind="postgres",
|
||||
cookie_secure=True,
|
||||
require_csrf=False,
|
||||
@@ -387,6 +399,9 @@ def test_production_refuses_insecure_configuration() -> None:
|
||||
Settings(
|
||||
environment=Environment.PRODUCTION,
|
||||
database_dsn=TEST_DSN,
|
||||
mailer_transport="smtp",
|
||||
smtp_host="smtp.invalid",
|
||||
verification_base_url="https://umbratrans.bdmurphy.me",
|
||||
repository_kind="memory",
|
||||
cookie_secure=True,
|
||||
)
|
||||
@@ -398,6 +413,9 @@ def test_production_app_requires_an_explicit_repository() -> None:
|
||||
environment=Environment.PRODUCTION,
|
||||
repository_kind="postgres",
|
||||
database_dsn=TEST_DSN,
|
||||
mailer_transport="smtp",
|
||||
smtp_host="smtp.invalid",
|
||||
verification_base_url="https://umbratrans.bdmurphy.me",
|
||||
cookie_secure=True,
|
||||
)
|
||||
with pytest.raises(ConfigurationError) as error:
|
||||
|
||||
@@ -8,15 +8,18 @@ from uuid import UUID, uuid4
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from umbratrans_web.mailer import CapturingMailer
|
||||
from umbratrans_web.service import AccountService
|
||||
|
||||
from .conftest import (
|
||||
ApiHarness,
|
||||
SeededAccount,
|
||||
activated_admin,
|
||||
ensure_organization,
|
||||
provision_local,
|
||||
random_email,
|
||||
random_password,
|
||||
verification_token,
|
||||
)
|
||||
|
||||
|
||||
@@ -29,6 +32,17 @@ def admin(api: ApiHarness, service: AccountService) -> object:
|
||||
return account
|
||||
|
||||
|
||||
def verified_admin(service: AccountService, mailer: CapturingMailer, **kwargs) -> SeededAccount:
|
||||
"""A platform administrator whose email has been confirmed.
|
||||
|
||||
Administrator-provisioned accounts start ``invited`` and cannot sign in
|
||||
until the address is confirmed, so tests that are not about validation say
|
||||
so explicitly.
|
||||
"""
|
||||
|
||||
return activated_admin(service, username="Umbraadmin", mailer=mailer, **kwargs)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Authorization
|
||||
# ----------------------------------------------------------------------
|
||||
@@ -113,9 +127,11 @@ def test_create_oidc_account(api: ApiHarness, admin: object, service: AccountSer
|
||||
body = response.json()
|
||||
assert body["temporary_password"] is None
|
||||
account = body["account"]
|
||||
assert account["username"] == "Oidc.User"
|
||||
assert account["email"] == "oidc.user@example.com"
|
||||
assert account["status"] == "active"
|
||||
assert account["username"] == "Oidc.User"
|
||||
# Provisioned by an administrator, so the address must be confirmed before
|
||||
# the account can sign in.
|
||||
assert account["status"] == "invited"
|
||||
assert account["credential"] == {
|
||||
"credential_type": "oidc",
|
||||
"must_change_password": False,
|
||||
@@ -177,7 +193,7 @@ def test_admin_provisions_a_temporary_password_account(
|
||||
assert any(not character.isalnum() for character in temporary_password)
|
||||
assert account["credential"]["credential_type"] == "local_password"
|
||||
assert account["credential"]["must_change_password"] is True
|
||||
assert account["status"] == "active"
|
||||
assert account["status"] == "invited"
|
||||
assert "$argon2" not in response.text
|
||||
|
||||
# It is never stored, logged, or audited in the clear.
|
||||
@@ -200,22 +216,27 @@ def test_admin_provisions_a_temporary_password_account(
|
||||
def test_generated_temporary_password_must_be_replaced_on_first_login(
|
||||
api: ApiHarness,
|
||||
admin: object,
|
||||
service: AccountService,
|
||||
mailer: CapturingMailer,
|
||||
) -> None:
|
||||
email = random_email("first-login")
|
||||
created = api.post(
|
||||
"/api/uadmin/accounts",
|
||||
json={
|
||||
"username": "First.Login",
|
||||
"display_name": "First Login",
|
||||
"email": random_email("first-login"),
|
||||
"email": email,
|
||||
"enrollment_mode": "temporary_password",
|
||||
},
|
||||
)
|
||||
assert created.status_code == 201
|
||||
username = created.json()["account"]["username"]
|
||||
# Administrator-provisioned accounts wait for the address to be confirmed.
|
||||
assert created.json()["account"]["status"] == "invited"
|
||||
temporary_password = created.json()["temporary_password"]
|
||||
service.verify_email(verification_token(mailer))
|
||||
|
||||
api.logout()
|
||||
login = api.login(username, temporary_password)
|
||||
login = api.login(email, temporary_password)
|
||||
assert login.status_code == 200
|
||||
assert login.json()["must_change_password"] is True
|
||||
|
||||
@@ -236,8 +257,8 @@ def test_generated_temporary_password_must_be_replaced_on_first_login(
|
||||
|
||||
# The one-time password is spent: it never works again.
|
||||
api.logout()
|
||||
assert api.login(username, temporary_password).status_code == 401
|
||||
assert api.login(username, new_password).status_code == 200
|
||||
assert api.login(email, temporary_password).status_code == 401
|
||||
assert api.login(email, new_password).status_code == 200
|
||||
|
||||
|
||||
def test_generated_temporary_passwords_are_unique_per_account(
|
||||
@@ -409,12 +430,17 @@ def test_organization_admin_scope_is_enforced(api: ApiHarness, service: AccountS
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_suspend_reactivate_and_deactivate(api: ApiHarness, admin: object, service: AccountService) -> None:
|
||||
def test_suspend_reactivate_and_deactivate(
|
||||
api: ApiHarness,
|
||||
admin: object,
|
||||
service: AccountService,
|
||||
mailer: CapturingMailer,
|
||||
) -> None:
|
||||
member = provision_local(service, username="Member.User", role="member")
|
||||
# A separate browser holds the member's live session while the
|
||||
# administrator acts through the first client.
|
||||
with TestClient(api.client.app) as member_client:
|
||||
result = member_client.post("/api/auth/login", json={"username": member.username, "password": member.password})
|
||||
result = member_client.post("/api/auth/login", json={"email": member.email, "password": member.password})
|
||||
assert result.status_code == 200
|
||||
|
||||
suspended = api.patch(
|
||||
@@ -439,7 +465,7 @@ def test_suspend_reactivate_and_deactivate(api: ApiHarness, admin: object, servi
|
||||
assert deactivated.json()["account"]["status"] == "deactivated"
|
||||
|
||||
# Logging in as the deactivated account is refused.
|
||||
assert api.login(member.username, member.password).status_code == 401
|
||||
assert api.login(member.email, member.password).status_code == 401
|
||||
|
||||
|
||||
def test_status_change_of_the_last_platform_admin_is_refused(api: ApiHarness, admin: object) -> None:
|
||||
@@ -451,10 +477,15 @@ def test_status_change_of_the_last_platform_admin_is_refused(api: ApiHarness, ad
|
||||
assert response.json()["error"]["code"] == "last_platform_admin"
|
||||
|
||||
|
||||
def test_revoke_sessions_endpoint(api: ApiHarness, admin: object, service: AccountService) -> None:
|
||||
def test_revoke_sessions_endpoint(
|
||||
api: ApiHarness,
|
||||
admin: object,
|
||||
service: AccountService,
|
||||
mailer: CapturingMailer,
|
||||
) -> None:
|
||||
member = provision_local(service, username="Member.User", role="member")
|
||||
with TestClient(api.client.app) as member_client:
|
||||
result = member_client.post("/api/auth/login", json={"username": member.username, "password": member.password})
|
||||
result = member_client.post("/api/auth/login", json={"email": member.email, "password": member.password})
|
||||
assert result.status_code == 200
|
||||
|
||||
revoked = api.post(f"/api/uadmin/accounts/{member.user_id}/revoke-sessions")
|
||||
|
||||
@@ -108,7 +108,10 @@ def test_bootstrap_creates_a_must_change_account_in_a_private_file(tmp_path: Pat
|
||||
path = tmp_path / "state" / "dev-accounts.json"
|
||||
password = random_password(28)
|
||||
|
||||
code, out, err = _run(["--repository-path", str(path)], entries=[password, password])
|
||||
code, out, err = _run(
|
||||
["--repository-path", str(path), "--email", "umbraadmin@example.test"],
|
||||
entries=[password, password],
|
||||
)
|
||||
|
||||
assert code == EXIT_OK, err
|
||||
assert "Bootstrap account created." in out
|
||||
@@ -132,7 +135,7 @@ def test_bootstrap_creates_a_must_change_account_in_a_private_file(tmp_path: Pat
|
||||
argon2_parallelism=1,
|
||||
)
|
||||
service = AccountService(JsonDevelopmentRepository(settings), settings)
|
||||
result = service.authenticate(username="Umbraadmin", password=password)
|
||||
result = service.authenticate(email="umbraadmin@example.test", password=password)
|
||||
assert result.session.must_change_password is True
|
||||
|
||||
raw = path.read_text(encoding="utf-8")
|
||||
|
||||
@@ -64,7 +64,32 @@ def test_production_settings_select_the_postgres_adapter() -> None:
|
||||
load_settings(without_dsn)
|
||||
assert missing.value.code == "production_dsn_missing"
|
||||
|
||||
settings = load_settings({**without_dsn, "UMBRATRANS_DATABASE_DSN": "postgresql://app@db.internal/umbracontrol"})
|
||||
# A DSN alone is not enough: production must also be able to deliver a
|
||||
# verification link, or every new account would be permanently unusable.
|
||||
with pytest.raises(ConfigurationError) as no_mailer:
|
||||
load_settings({**without_dsn, "UMBRATRANS_DATABASE_DSN": TEST_DSN})
|
||||
assert no_mailer.value.code == "production_mailer_missing"
|
||||
|
||||
with pytest.raises(ConfigurationError) as no_url:
|
||||
load_settings(
|
||||
{
|
||||
**without_dsn,
|
||||
"UMBRATRANS_DATABASE_DSN": TEST_DSN,
|
||||
"UMBRATRANS_MAILER": "smtp",
|
||||
"UMBRATRANS_SMTP_HOST": "smtp.invalid",
|
||||
}
|
||||
)
|
||||
assert no_url.value.code == "production_verification_url_missing"
|
||||
|
||||
settings = load_settings(
|
||||
{
|
||||
**without_dsn,
|
||||
"UMBRATRANS_DATABASE_DSN": TEST_DSN,
|
||||
"UMBRATRANS_MAILER": "smtp",
|
||||
"UMBRATRANS_SMTP_HOST": "smtp.invalid",
|
||||
"UMBRATRANS_VERIFICATION_BASE_URL": "https://umbratrans.bdmurphy.me",
|
||||
}
|
||||
)
|
||||
assert settings.is_production is True
|
||||
assert settings.repository_kind == "postgres"
|
||||
assert settings.cookie_secure is True
|
||||
|
||||
567
umbratrans_web/tests/test_email_verification.py
Normal file
567
umbratrans_web/tests/test_email_verification.py
Normal file
@@ -0,0 +1,567 @@
|
||||
"""Email-as-login and email-verification tests.
|
||||
|
||||
The properties under test are the ones that decide whether this feature is
|
||||
safe, rather than merely working:
|
||||
|
||||
* an unverified account cannot sign in, even with the right password;
|
||||
* the "not verified" message appears only *after* the password is proven, so the
|
||||
login form is not an account-existence oracle;
|
||||
* a verification link is single-use, expires, and is stored only as a hash;
|
||||
* resending reveals nothing about which addresses have accounts;
|
||||
* a resend always answers the same way.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from umbratrans_web.errors import (
|
||||
EmailNotVerifiedError,
|
||||
InputError,
|
||||
InvalidCredentialsError,
|
||||
StateTransitionError,
|
||||
VerificationDeliveryError,
|
||||
VerificationFailedError,
|
||||
)
|
||||
from umbratrans_web.mailer import CapturingMailer, LoggingMailer, MailerError, build_mailer
|
||||
from umbratrans_web.models import VERIFICATION_PURPOSE_EMAIL, AccountStatus
|
||||
from umbratrans_web.security import hash_session_token, utc_now
|
||||
from umbratrans_web.service import AccountService
|
||||
|
||||
from .conftest import (
|
||||
ApiHarness,
|
||||
admin_principal,
|
||||
provision_local,
|
||||
random_email,
|
||||
random_password,
|
||||
verification_token,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_actor(service: AccountService) -> object:
|
||||
return admin_principal(provision_local(service, username="Umbraadmin"))
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Email is the login identifier
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_sign_in_uses_the_email_address(service: AccountService) -> None:
|
||||
account = provision_local(service, username="Handle.Only")
|
||||
result = service.authenticate(email=account.email, password=account.password)
|
||||
assert result.session.must_change_password is True
|
||||
|
||||
|
||||
def test_sign_in_is_case_insensitive_in_the_address(service: AccountService) -> None:
|
||||
"""users.email is citext, so the address is already case-insensitive."""
|
||||
|
||||
account = provision_local(service, username="Case.User", email="Mixed.Case@Example.COM")
|
||||
assert service.authenticate(email="mixed.case@example.com", password=account.password)
|
||||
assert service.authenticate(email="MIXED.CASE@EXAMPLE.COM", password=account.password)
|
||||
|
||||
|
||||
def test_the_handle_is_not_a_login_path(service: AccountService) -> None:
|
||||
"""A username is a display handle now, so it must not authenticate anyone."""
|
||||
|
||||
account = provision_local(service, username="Not.ALogin")
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(email="not.a.login", password=account.password)
|
||||
|
||||
|
||||
def test_a_malformed_address_is_treated_as_no_such_account(service: AccountService) -> None:
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(email="not-an-address", password=random_password())
|
||||
|
||||
|
||||
def test_the_address_is_required_on_creation(service: AccountService) -> None:
|
||||
with pytest.raises(InputError) as error:
|
||||
provision_local(service, username="No.Address", email="")
|
||||
assert error.value.code == "email_required"
|
||||
|
||||
|
||||
def test_the_handle_is_optional(service: AccountService) -> None:
|
||||
account = service.create_account(
|
||||
enrollment_mode="temporary_password",
|
||||
username=None,
|
||||
display_name="No Handle",
|
||||
email=random_email("no-handle"),
|
||||
password=random_password(28),
|
||||
operator=__import__("umbratrans_web.service", fromlist=["OperatorContext"]).OperatorContext(
|
||||
source="operator-cli"
|
||||
),
|
||||
)
|
||||
assert account.username is None
|
||||
# The audit handle falls back to the address rather than becoming blank.
|
||||
stored = service.repository.get_user(account.id)
|
||||
assert stored is not None
|
||||
assert stored.audit_handle == stored.email
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# An unverified account cannot sign in
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_an_unverified_account_is_refused_with_a_distinct_error(
|
||||
service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
issued = service.provision_admin_temporary_password(
|
||||
username="Pending.User",
|
||||
display_name="Pending User",
|
||||
email=random_email("pending"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
assert issued.account.status is AccountStatus.INVITED
|
||||
|
||||
with pytest.raises(EmailNotVerifiedError) as error:
|
||||
service.authenticate(email=issued.account.email, password=issued.temporary_password)
|
||||
assert error.value.http_status == 403
|
||||
assert "inbox" in error.value.message.lower()
|
||||
|
||||
|
||||
def test_the_unverified_message_comes_only_after_the_password_is_right(
|
||||
service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
"""The distinction matters: a wrong password must not reveal the state.
|
||||
|
||||
Reporting "email not verified" before authentication would turn the login
|
||||
form into an account-existence oracle, which is the single thing the uniform
|
||||
authentication error exists to prevent.
|
||||
"""
|
||||
|
||||
issued = service.provision_admin_temporary_password(
|
||||
username="Pending.Two",
|
||||
display_name="Pending Two",
|
||||
email=random_email("pending-two"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
with pytest.raises(InvalidCredentialsError) as wrong_password:
|
||||
service.authenticate(email=issued.account.email, password=random_password(28))
|
||||
with pytest.raises(InvalidCredentialsError) as unknown_account:
|
||||
service.authenticate(email=random_email("nobody"), password=random_password(28))
|
||||
|
||||
# The two pre-authentication failures are indistinguishable.
|
||||
assert wrong_password.value.code == unknown_account.value.code
|
||||
assert wrong_password.value.http_status == unknown_account.value.http_status
|
||||
assert wrong_password.value.message == unknown_account.value.message
|
||||
|
||||
|
||||
def test_an_unverified_account_has_no_session_at_all(
|
||||
service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
issued = service.provision_admin_temporary_password(
|
||||
username="No.Session",
|
||||
display_name="No Session",
|
||||
email=random_email("no-session"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
with pytest.raises(EmailNotVerifiedError):
|
||||
service.authenticate(email=issued.account.email, password=issued.temporary_password)
|
||||
assert service.repository.count_sessions(issued.account.id) == 0
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# The verification link
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_verification_activates_the_account(
|
||||
service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
issued = service.provision_admin_temporary_password(
|
||||
username="Verify.Me",
|
||||
display_name="Verify Me",
|
||||
email=random_email("verify-me"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
verified = service.verify_email(verification_token(mailer))
|
||||
assert verified.status is AccountStatus.ACTIVE
|
||||
assert verified.is_email_verified is True
|
||||
# And now the password works.
|
||||
assert service.authenticate(email=verified.email, password=issued.temporary_password)
|
||||
|
||||
|
||||
def test_the_link_is_single_use(service: AccountService, admin_actor: object, mailer: CapturingMailer) -> None:
|
||||
service.provision_admin_temporary_password(
|
||||
username="Single.Use",
|
||||
display_name="Single Use",
|
||||
email=random_email("single-use"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
token = verification_token(mailer)
|
||||
service.verify_email(token)
|
||||
with pytest.raises(VerificationFailedError):
|
||||
service.verify_email(token)
|
||||
|
||||
|
||||
def test_every_failure_mode_looks_identical(
|
||||
service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
"""Unknown, already used and expired must be indistinguishable."""
|
||||
|
||||
service.provision_admin_temporary_password(
|
||||
username="Same.Error",
|
||||
display_name="Same Error",
|
||||
email=random_email("same-error"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
unknown = None
|
||||
for candidate in ("a-token-never-issued", "", "x" * 64):
|
||||
with pytest.raises(VerificationFailedError) as error:
|
||||
service.verify_email(candidate)
|
||||
if unknown is None:
|
||||
unknown = (error.value.code, error.value.message, error.value.http_status)
|
||||
assert (error.value.code, error.value.message, error.value.http_status) == unknown
|
||||
|
||||
|
||||
def test_an_expired_link_is_refused(
|
||||
service: AccountService, admin_actor: object, mailer: CapturingMailer, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
service.provision_admin_temporary_password(
|
||||
username="Expired.Link",
|
||||
display_name="Expired Link",
|
||||
email=random_email("expired"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
later = utc_now() + timedelta(days=30)
|
||||
monkeypatch.setattr(service, "_now", lambda: later)
|
||||
with pytest.raises(VerificationFailedError):
|
||||
service.verify_email(verification_token(mailer))
|
||||
|
||||
|
||||
def test_only_the_hash_is_stored(service: AccountService, admin_actor: object, mailer: CapturingMailer) -> None:
|
||||
"""A leaked database must not yield a usable link."""
|
||||
|
||||
service.provision_admin_temporary_password(
|
||||
username="Hashed.Only",
|
||||
display_name="Hashed Only",
|
||||
email=random_email("hashed"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
plaintext = verification_token(mailer)
|
||||
stored = list(service.repository._verification_tokens.values()) # noqa: SLF001 - asserting storage
|
||||
assert stored
|
||||
for token in stored:
|
||||
assert token.token_hash != plaintext
|
||||
assert token.token_hash == hash_session_token(plaintext)
|
||||
assert plaintext not in repr(token)
|
||||
assert token.purpose == VERIFICATION_PURPOSE_EMAIL
|
||||
|
||||
|
||||
def test_issuing_a_second_link_invalidates_nothing_but_keeps_both_usable(
|
||||
service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
"""Re-mailing must not let a stale link be replayed to activate twice."""
|
||||
|
||||
issued = service.provision_admin_temporary_password(
|
||||
username="Re.Mailed",
|
||||
display_name="Re Mailed",
|
||||
email=random_email("re-mailed"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
first = verification_token(mailer, 0)
|
||||
service.resend_email_verification(issued.account.email)
|
||||
second = verification_token(mailer, -1)
|
||||
assert first != second
|
||||
service.verify_email(second)
|
||||
# The earlier link is now spent because the account is already active.
|
||||
with pytest.raises(VerificationFailedError):
|
||||
service.verify_email(first)
|
||||
|
||||
|
||||
def test_verification_cannot_be_repeated_on_an_active_account(
|
||||
service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
account = provision_local(service, username="Already.Active")
|
||||
with pytest.raises(StateTransitionError) as error:
|
||||
service.issue_email_verification(account.user_id)
|
||||
assert error.value.code == "already_verified"
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Resending must not enumerate accounts
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_resend_says_nothing_about_whether_an_account_exists(
|
||||
service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
issued = service.provision_admin_temporary_password(
|
||||
username="Resend.Me",
|
||||
display_name="Resend Me",
|
||||
email=random_email("resend"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
sent_before = len(mailer.sent)
|
||||
assert service.resend_email_verification(issued.account.email) is True
|
||||
assert service.resend_email_verification(random_email("ghost")) is False
|
||||
# Only the real account produced mail.
|
||||
assert len(mailer.sent) == sent_before + 1
|
||||
|
||||
|
||||
def test_resend_for_an_already_verified_account_sends_nothing(service: AccountService, mailer: CapturingMailer) -> None:
|
||||
account = provision_local(service, username="Done.Here")
|
||||
before = len(mailer.sent)
|
||||
assert service.resend_email_verification(account.email) is False
|
||||
assert len(mailer.sent) == before
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Delivery failure
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_a_delivery_failure_is_reported_not_swallowed(
|
||||
service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
class _Broken:
|
||||
transport_name = "broken"
|
||||
|
||||
def send(self, **kwargs: object) -> None:
|
||||
raise MailerError("nope")
|
||||
|
||||
broken = AccountService(service.repository, service.settings, mailer=_Broken())
|
||||
with pytest.raises(VerificationDeliveryError) as error:
|
||||
broken.provision_admin_temporary_password(
|
||||
username="Undeliverable",
|
||||
display_name="Undeliverable",
|
||||
email=random_email("undeliverable"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
# The message names no address and carries no token.
|
||||
assert "Undeliverable" not in str(error.value)
|
||||
assert error.value.http_status == 502
|
||||
|
||||
|
||||
def test_the_operator_path_needs_no_validation(service: AccountService, mailer: CapturingMailer) -> None:
|
||||
"""Bootstrap must work with no mail infrastructure, or there is no way in.
|
||||
|
||||
The operator is attesting to the address themselves, which is the same
|
||||
standing an administrator has. Leaving the operator path needing a
|
||||
delivered email would deadlock first-time setup on the SMTP configuration.
|
||||
"""
|
||||
|
||||
account = provision_local(service, username="Bootstrap.Admin")
|
||||
assert account.view.status is AccountStatus.ACTIVE
|
||||
assert account.view.is_email_verified is True
|
||||
assert mailer.sent == []
|
||||
assert service.authenticate(email=account.email, password=account.password)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# HTTP surface
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_the_login_form_takes_an_address(api: ApiHarness, service: AccountService, mailer: CapturingMailer) -> None:
|
||||
account = provision_local(service, username="Http.Login")
|
||||
response = api.client.post("/api/auth/login", json={"email": account.email, "password": account.password})
|
||||
assert response.status_code == 200
|
||||
assert response.json()["user"]["email"] == account.email
|
||||
|
||||
|
||||
def test_login_with_a_username_field_is_rejected(api: ApiHarness, service: AccountService) -> None:
|
||||
account = provision_local(service, username="Http.Handle")
|
||||
response = api.client.post("/api/auth/login", json={"username": account.username, "password": account.password})
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_an_unverified_login_returns_the_actionable_error_over_http(
|
||||
api: ApiHarness, service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
issued = service.provision_admin_temporary_password(
|
||||
username="Http.Pending",
|
||||
display_name="Http Pending",
|
||||
email=random_email("http-pending"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
response = api.client.post(
|
||||
"/api/auth/login", json={"email": issued.account.email, "password": issued.temporary_password}
|
||||
)
|
||||
assert response.status_code == 403
|
||||
assert response.json()["error"]["code"] == "email_not_verified"
|
||||
|
||||
|
||||
def test_verify_email_over_http(
|
||||
api: ApiHarness, service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
service.provision_admin_temporary_password(
|
||||
username="Http.Verify",
|
||||
display_name="Http Verify",
|
||||
email=random_email("http-verify"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
token = verification_token(mailer)
|
||||
response = api.post("/api/auth/verify-email", json={"token": token})
|
||||
assert response.status_code == 200
|
||||
assert response.json()["account"]["status"] == "active"
|
||||
# The bearer capability is never echoed back.
|
||||
assert token not in response.text
|
||||
# And it cannot be replayed.
|
||||
assert api.post("/api/auth/verify-email", json={"token": token}).status_code == 400
|
||||
|
||||
|
||||
def test_the_link_can_be_followed_in_a_browser(
|
||||
api: ApiHarness, service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
service.provision_admin_temporary_password(
|
||||
username="Http.Link",
|
||||
display_name="Http Link",
|
||||
email=random_email("http-link"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
response = api.client.get(f"/api/auth/verify-email?token={verification_token(mailer)}")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["account"]["status"] == "active"
|
||||
|
||||
|
||||
def test_resend_answers_the_same_way_for_both_outcomes(
|
||||
api: ApiHarness, service: AccountService, admin_actor: object, mailer: CapturingMailer
|
||||
) -> None:
|
||||
issued = service.provision_admin_temporary_password(
|
||||
username="Http.Resend",
|
||||
display_name="Http Resend",
|
||||
email=random_email("http-resend"),
|
||||
actor=admin_actor, # type: ignore[arg-type]
|
||||
)
|
||||
known = api.post("/api/auth/resend-verification", json={"email": issued.account.email})
|
||||
unknown = api.post("/api/auth/resend-verification", json={"email": random_email("ghost")})
|
||||
assert known.status_code == unknown.status_code == 200
|
||||
assert known.json() == unknown.json()
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# The mail transport
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_the_logging_transport_says_it_did_not_send(caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""A logged link must never look like a delivered one."""
|
||||
|
||||
mailer = LoggingMailer()
|
||||
with caplog.at_level("WARNING"):
|
||||
mailer.send(to="someone@example.test", subject="Confirm", body="link here")
|
||||
text = caplog.text
|
||||
assert "NOT sent" in text
|
||||
assert "transport=log" in text
|
||||
|
||||
|
||||
def test_production_refuses_the_logging_transport() -> None:
|
||||
from umbratrans_web.config import ConfigurationError, Environment, Settings
|
||||
|
||||
# The configuration layer refuses first, which is the right place: the
|
||||
# process must not start rather than start and quietly fail to deliver.
|
||||
with pytest.raises(ConfigurationError) as settings_error:
|
||||
Settings(
|
||||
environment=Environment.PRODUCTION,
|
||||
repository_kind="json",
|
||||
cookie_secure=True,
|
||||
database_dsn="postgresql://app@db.invalid/umbracontrol",
|
||||
mailer_transport="log",
|
||||
)
|
||||
assert settings_error.value.code == "production_mailer_missing"
|
||||
|
||||
# The mailer layer keeps its own guard for any caller that reaches it with
|
||||
# production settings by another route.
|
||||
class _ProductionByAnotherRoute:
|
||||
environment = "production"
|
||||
mailer_transport = "log"
|
||||
smtp_host = "smtp.invalid"
|
||||
smtp_port = 587
|
||||
smtp_username = None
|
||||
smtp_password = None
|
||||
smtp_sender = "umbratrans@example.test"
|
||||
smtp_starttls = True
|
||||
|
||||
with pytest.raises(MailerError) as mailer_error:
|
||||
build_mailer(_ProductionByAnotherRoute())
|
||||
assert "never delivered" in str(mailer_error.value)
|
||||
|
||||
|
||||
def test_development_defaults_to_the_logging_transport() -> None:
|
||||
from umbratrans_web.config import Environment, Settings
|
||||
|
||||
assert isinstance(build_mailer(Settings(environment=Environment.DEVELOPMENT)), LoggingMailer)
|
||||
|
||||
|
||||
def test_an_unknown_transport_is_refused_at_configuration_time() -> None:
|
||||
"""Caught by validation, so no mailer is ever constructed for it."""
|
||||
|
||||
from umbratrans_web.config import ConfigurationError, Environment, Settings
|
||||
|
||||
with pytest.raises(ConfigurationError) as error:
|
||||
Settings(environment=Environment.DEVELOPMENT, mailer_transport="carrier_pigeon")
|
||||
assert error.value.code == "invalid_mailer_transport"
|
||||
|
||||
|
||||
def test_the_smtp_transport_is_selected_when_configured() -> None:
|
||||
from umbratrans_web.config import Environment, Settings
|
||||
|
||||
settings = Settings(
|
||||
environment=Environment.DEVELOPMENT,
|
||||
mailer_transport="smtp",
|
||||
smtp_host="smtp.invalid",
|
||||
smtp_sender="umbratrans@example.test",
|
||||
)
|
||||
transport = build_mailer(settings)
|
||||
assert transport.transport_name == "smtp"
|
||||
|
||||
|
||||
def test_an_smtp_transport_needs_a_host() -> None:
|
||||
from umbratrans_web.config import Environment, Settings
|
||||
|
||||
with pytest.raises(MailerError):
|
||||
build_mailer(Settings(environment=Environment.DEVELOPMENT, mailer_transport="smtp"))
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Persistence
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_the_development_store_keeps_a_token_across_a_reload(tmp_path) -> None: # noqa: ANN001
|
||||
"""A restart must not silently invalidate every link that was mailed.
|
||||
|
||||
This is why verification tokens are persisted at all: an operator who
|
||||
restarts the service would otherwise invalidate links they already sent, and
|
||||
the account owner would be stuck with no way to recover.
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
from umbratrans_web.config import Environment, Settings
|
||||
from umbratrans_web.dev_repository import JsonDevelopmentRepository
|
||||
|
||||
path = tmp_path / "state" / "dev-accounts.json"
|
||||
settings = Settings(
|
||||
environment=Environment.TEST,
|
||||
repository_kind="json",
|
||||
repository_path=path,
|
||||
argon2_memory_cost_kib=8 * 1024,
|
||||
argon2_time_cost=1,
|
||||
argon2_parallelism=1,
|
||||
)
|
||||
mailer = CapturingMailer()
|
||||
service = AccountService(JsonDevelopmentRepository(settings), settings, mailer=mailer)
|
||||
actor = admin_principal(provision_local(service, username="Umbraadmin"))
|
||||
issued = service.provision_admin_temporary_password(
|
||||
username="Persisted",
|
||||
display_name="Persisted",
|
||||
email=random_email("persisted"),
|
||||
actor=actor,
|
||||
)
|
||||
assert issued.account.status is AccountStatus.INVITED
|
||||
token = verification_token(mailer)
|
||||
|
||||
# Reopen the same file, as a restart would.
|
||||
reloaded = AccountService(JsonDevelopmentRepository(settings), settings, mailer=CapturingMailer())
|
||||
assert reloaded.verify_email(token).status is AccountStatus.ACTIVE
|
||||
|
||||
# Only a hash reached the file.
|
||||
raw = path.read_text(encoding="utf-8")
|
||||
assert token not in raw
|
||||
assert json.loads(raw)["verification_tokens"][0]["token_hash"] == hash_session_token(token)
|
||||
@@ -136,7 +136,7 @@ def test_members_are_forbidden_from_organization_routes(service: AccountService)
|
||||
|
||||
def test_a_temporary_password_session_cannot_manage_organizations(service: AccountService) -> None:
|
||||
pending = provision_local(service, username="Pending.Admin", role="platform_admin")
|
||||
result = service.authenticate(username=pending.username, password=pending.password)
|
||||
result = service.authenticate(email=pending.email, password=pending.password)
|
||||
from umbratrans_web.models import Principal
|
||||
|
||||
principal = Principal(
|
||||
@@ -181,7 +181,7 @@ def test_update_requires_at_least_one_field(service: AccountService, platform_ad
|
||||
def test_suspension_revokes_member_sessions(service: AccountService, platform_admin: object) -> None:
|
||||
org = ensure_organization(service, name="Revoke Org")
|
||||
member = provision_local(service, username="Revoke.Member", role="member", organization_id=org.id)
|
||||
service.authenticate(username=member.username, password=member.password)
|
||||
service.authenticate(email=member.email, password=member.password)
|
||||
assert service.repository.count_sessions(member.user_id) == 1
|
||||
|
||||
organization, revoked = service.set_organization_status(platform_admin, org.id, "suspended") # type: ignore[arg-type]
|
||||
@@ -190,7 +190,7 @@ def test_suspension_revokes_member_sessions(service: AccountService, platform_ad
|
||||
assert service.repository.count_sessions(member.user_id) == 0
|
||||
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username=member.username, password=member.password)
|
||||
service.authenticate(email=member.email, password=member.password)
|
||||
|
||||
|
||||
def test_reactivation_restores_login(service: AccountService, platform_admin: object) -> None:
|
||||
@@ -199,11 +199,11 @@ def test_reactivation_restores_login(service: AccountService, platform_admin: ob
|
||||
service.set_organization_status(platform_admin, org.id, "suspended") # type: ignore[arg-type]
|
||||
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username=member.username, password=member.password)
|
||||
service.authenticate(email=member.email, password=member.password)
|
||||
|
||||
service.set_organization_status(platform_admin, org.id, "active") # type: ignore[arg-type]
|
||||
# The credential is untouched, so the same password works again.
|
||||
assert service.authenticate(username=member.username, password=member.password) is not None
|
||||
assert service.authenticate(email=member.email, password=member.password) is not None
|
||||
|
||||
|
||||
def test_archived_is_terminal(service: AccountService, platform_admin: object) -> None:
|
||||
@@ -296,7 +296,7 @@ def test_summary_counts_pending_first_logins(service: AccountService, platform_a
|
||||
provision_local(service, username="Still.Temp", role="member", organization_id=org.id)
|
||||
# A completed first-login change clears the flag.
|
||||
pending = provision_local(service, username="Already.Changed", role="member", organization_id=org.id)
|
||||
result = service.authenticate(username=pending.username, password=pending.password)
|
||||
result = service.authenticate(email=pending.email, password=pending.password)
|
||||
service.change_first_password(
|
||||
service.resolve_session(result.session.token),
|
||||
current_password=pending.password,
|
||||
@@ -476,7 +476,7 @@ def test_suspending_an_organization_locks_its_members_out_over_http(
|
||||
signed_in_admin.patch(f"/api/uadmin/organizations/{org.id}/status", json={"status": "suspended"})
|
||||
# The account itself is still active; only the tenant is suspended.
|
||||
assert service.repository.get_user(member.user_id).status is AccountStatus.ACTIVE
|
||||
response = signed_in_admin.login(member.username, member.password)
|
||||
response = signed_in_admin.login(member.email, member.password)
|
||||
assert response.status_code == 401
|
||||
assert response.json()["error"]["code"] == "invalid_credentials"
|
||||
|
||||
|
||||
@@ -98,6 +98,7 @@ def test_a_platform_admin_is_assigned_at_platform_scope() -> None:
|
||||
username="Umbraadmin",
|
||||
username_key="umbraadmin",
|
||||
display_name="Umbra Administrator",
|
||||
email="umbraadmin@example.test",
|
||||
role=RoleName.PLATFORM_ADMIN,
|
||||
status=AccountStatus.ACTIVE,
|
||||
)
|
||||
@@ -117,6 +118,7 @@ def test_an_organization_scoped_role_requires_an_organization() -> None:
|
||||
username="Orphan.Member",
|
||||
username_key="orphan.member",
|
||||
display_name="Orphan",
|
||||
email="orphan@example.test",
|
||||
role=RoleName.MEMBER,
|
||||
status=AccountStatus.ACTIVE,
|
||||
)
|
||||
@@ -132,6 +134,7 @@ def test_a_member_carries_its_organization() -> None:
|
||||
username="Member.One",
|
||||
username_key="member.one",
|
||||
display_name="Member One",
|
||||
email="member.one@example.test",
|
||||
role=RoleName.MEMBER,
|
||||
status=AccountStatus.ACTIVE,
|
||||
organization_id=organization_id,
|
||||
@@ -146,6 +149,7 @@ def test_an_organization_admin_is_also_organization_scoped() -> None:
|
||||
username="Org.Admin",
|
||||
username_key="org.admin",
|
||||
display_name="Org Admin",
|
||||
email="org.admin@example.test",
|
||||
role=RoleName.ORGANIZATION_ADMIN,
|
||||
status=AccountStatus.ACTIVE,
|
||||
organization_id=organization_id,
|
||||
@@ -184,6 +188,20 @@ def test_modelled_account_statuses_are_accepted(status: str) -> None:
|
||||
assert _user_from_row(_row(status=status)).status is AccountStatus(status)
|
||||
|
||||
|
||||
def test_an_invited_account_maps_and_is_not_yet_verified() -> None:
|
||||
"""``invited`` is the control plane's own state for an unconfirmed address.
|
||||
|
||||
It is modelled rather than refused, because an account that exists but has
|
||||
not proved it can receive mail must be distinguishable from one that does
|
||||
not exist yet.
|
||||
"""
|
||||
|
||||
user = _user_from_row(_row(status="invited"))
|
||||
assert user.status is AccountStatus.INVITED
|
||||
assert user.is_email_verified is False
|
||||
assert _user_from_row(_row(status="active")).is_email_verified is True
|
||||
|
||||
|
||||
def test_an_account_requiring_mfa_is_refused() -> None:
|
||||
"""The control plane can require a second factor this phase cannot do.
|
||||
|
||||
@@ -202,7 +220,7 @@ def test_a_profile_row_without_the_mfa_column_is_accepted() -> None:
|
||||
assert _user_from_row(_row()).status is AccountStatus.ACTIVE
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status", ["invited", "locked", "paused", "ACTIVE"])
|
||||
@pytest.mark.parametrize("status", ["locked", "paused", "ACTIVE"])
|
||||
def test_an_unmodelled_status_is_refused_rather_than_coerced(status: str) -> None:
|
||||
"""``invited`` and ``locked`` exist in the control plane but not here.
|
||||
|
||||
|
||||
@@ -98,7 +98,7 @@ def test_find_users_filters_and_paginates(repository: InMemoryAccountRepository,
|
||||
def test_audit_listing_is_newest_first(repository: InMemoryAccountRepository, service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
with pytest.raises(Exception):
|
||||
service.authenticate(username=account.username, password=random_password())
|
||||
service.authenticate(email=account.email, password=random_password())
|
||||
|
||||
events, total = repository.list_audit_events(limit=50)
|
||||
assert total == len(events) >= 1
|
||||
@@ -147,7 +147,7 @@ def test_json_repository_persists_hashes_across_restarts(tmp_path: Path) -> None
|
||||
assert stored.password_hash is not None
|
||||
assert stored.must_change_password is True
|
||||
# The account still authenticates from the persisted hash.
|
||||
result = second.authenticate(username=account.username, password=account.password)
|
||||
result = second.authenticate(email=account.email, password=account.password)
|
||||
assert result.session.must_change_password is True
|
||||
# The audit trail survived too.
|
||||
events, total = second.repository.list_audit_events()
|
||||
@@ -187,6 +187,11 @@ def test_json_repository_refuses_production_mode(tmp_path: Path) -> None:
|
||||
Settings(
|
||||
environment=Environment.PRODUCTION,
|
||||
database_dsn=TEST_DSN,
|
||||
# Satisfy the mailer gates so the repository-kind gate is the one
|
||||
# under test rather than whichever check happens to run first.
|
||||
mailer_transport="smtp",
|
||||
smtp_host="smtp.invalid",
|
||||
verification_base_url="https://umbratrans.bdmurphy.me",
|
||||
repository_kind="json",
|
||||
cookie_secure=True,
|
||||
repository_path=tmp_path / "dev-accounts.json",
|
||||
|
||||
@@ -9,6 +9,7 @@ import pytest
|
||||
|
||||
from umbratrans_web.errors import (
|
||||
ConflictError,
|
||||
EmailNotVerifiedError,
|
||||
EnrollmentRefusedError,
|
||||
ForbiddenError,
|
||||
InputError,
|
||||
@@ -18,7 +19,13 @@ from umbratrans_web.errors import (
|
||||
PasswordPolicyError,
|
||||
StateTransitionError,
|
||||
)
|
||||
from umbratrans_web.models import AccountStatus, CredentialType, Principal, RoleName
|
||||
from umbratrans_web.mailer import CapturingMailer
|
||||
from umbratrans_web.models import (
|
||||
AccountStatus,
|
||||
CredentialType,
|
||||
Principal,
|
||||
RoleName,
|
||||
)
|
||||
from umbratrans_web.service import AccountService, OperatorContext
|
||||
|
||||
from .conftest import (
|
||||
@@ -29,6 +36,7 @@ from .conftest import (
|
||||
provision_oidc,
|
||||
random_email,
|
||||
random_password,
|
||||
verification_token,
|
||||
)
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
@@ -110,7 +118,7 @@ def test_generic_create_account_path_requires_the_operator_context(service: Acco
|
||||
assert service.repository.get_user_by_username("no.path") is None
|
||||
|
||||
|
||||
def test_admin_enrollment_generates_a_one_time_password(service: AccountService) -> None:
|
||||
def test_admin_enrollment_generates_a_one_time_password(service: AccountService, mailer: CapturingMailer) -> None:
|
||||
actor = admin_principal(provision_local(service, username="Umbraadmin"))
|
||||
issued = service.provision_admin_temporary_password(
|
||||
username="Temp.User",
|
||||
@@ -118,6 +126,12 @@ def test_admin_enrollment_generates_a_one_time_password(service: AccountService)
|
||||
email=random_email("temp-user"),
|
||||
actor=actor,
|
||||
)
|
||||
# An administrator-provisioned account waits for the address to be
|
||||
# confirmed, so the password cannot be exercised until it is.
|
||||
assert issued.account.status is AccountStatus.INVITED
|
||||
with pytest.raises(EmailNotVerifiedError):
|
||||
service.authenticate(email=issued.account.email, password=issued.temporary_password)
|
||||
service.verify_email(verification_token(mailer))
|
||||
|
||||
# Handed back once, and hidden from any accidental rendering or log line.
|
||||
assert issued.temporary_password not in repr(issued)
|
||||
@@ -130,8 +144,9 @@ def test_admin_enrollment_generates_a_one_time_password(service: AccountService)
|
||||
assert stored.password_hash.startswith("$argon2id$")
|
||||
assert issued.temporary_password not in stored.password_hash
|
||||
|
||||
# It authenticates once, and only far enough to be replaced.
|
||||
first = service.authenticate(username="Temp.User", password=issued.temporary_password)
|
||||
# It authenticates once, and only far enough to be replaced. The
|
||||
# identifier is the address, not the handle.
|
||||
first = service.authenticate(email=issued.account.email, password=issued.temporary_password)
|
||||
assert first.session.must_change_password is True
|
||||
replacement = random_password(30)
|
||||
service.change_first_password(
|
||||
@@ -140,8 +155,8 @@ def test_admin_enrollment_generates_a_one_time_password(service: AccountService)
|
||||
new_password=replacement,
|
||||
)
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username="Temp.User", password=issued.temporary_password)
|
||||
assert service.authenticate(username="Temp.User", password=replacement).session.must_change_password is False
|
||||
service.authenticate(email=issued.account.email, password=issued.temporary_password)
|
||||
assert service.authenticate(email=issued.account.email, password=replacement).session.must_change_password is False
|
||||
|
||||
|
||||
def test_admin_enrollment_requires_an_administrator(service: AccountService) -> None:
|
||||
@@ -219,7 +234,7 @@ def test_oidc_account_cannot_authenticate_with_a_local_password(service: Account
|
||||
actor = admin_principal(provision_local(service, username="Umbraadmin"))
|
||||
account = provision_oidc(service, actor, username="Oidc.User")
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username=account.username, password=random_password())
|
||||
service.authenticate(email=account.email, password=random_password())
|
||||
|
||||
|
||||
def test_only_declared_operator_sources_are_accepted() -> None:
|
||||
@@ -237,7 +252,7 @@ def test_only_declared_operator_sources_are_accepted() -> None:
|
||||
|
||||
def test_authentication_issues_a_session_without_exposing_the_token(service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
result = service.authenticate(username=account.username, password=account.password)
|
||||
result = service.authenticate(email=account.email, password=account.password)
|
||||
|
||||
assert result.session.token
|
||||
assert result.session.must_change_password is True
|
||||
@@ -248,7 +263,7 @@ def test_authentication_issues_a_session_without_exposing_the_token(service: Acc
|
||||
|
||||
def test_stored_session_token_is_a_hash_not_the_token(repository: object, service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
result = service.authenticate(username=account.username, password=account.password)
|
||||
result = service.authenticate(email=account.email, password=account.password)
|
||||
|
||||
stored = repository.get_session(result.session.session_id) # type: ignore[attr-defined]
|
||||
assert stored is not None
|
||||
@@ -260,9 +275,9 @@ def test_stored_session_token_is_a_hash_not_the_token(repository: object, servic
|
||||
def test_unknown_user_and_wrong_password_are_indistinguishable(repository: object, service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
with pytest.raises(InvalidCredentialsError) as unknown:
|
||||
service.authenticate(username="Nobody.Here", password=random_password())
|
||||
service.authenticate(email="Nobody.Here", password=random_password())
|
||||
with pytest.raises(InvalidCredentialsError) as wrong:
|
||||
service.authenticate(username=account.username, password=random_password())
|
||||
service.authenticate(email=account.email, password=random_password())
|
||||
|
||||
assert unknown.value.code == wrong.value.code == "invalid_credentials"
|
||||
assert unknown.value.message == wrong.value.message
|
||||
@@ -272,7 +287,7 @@ def test_unknown_user_and_wrong_password_are_indistinguishable(repository: objec
|
||||
def test_login_failures_are_audited_without_the_username_value(repository: object, service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username=account.username, password=random_password())
|
||||
service.authenticate(email=account.email, password=random_password())
|
||||
|
||||
events, _ = repository.list_audit_events(target_user_id=account.user_id) # type: ignore[attr-defined]
|
||||
failures = [event for event in events if event.action == "auth.login" and event.outcome == "failure"]
|
||||
@@ -285,7 +300,7 @@ def test_repeated_failures_lock_the_credential(service: AccountService, settings
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
for _ in range(settings.max_failed_login_attempts): # type: ignore[attr-defined]
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username=account.username, password=random_password())
|
||||
service.authenticate(email=account.email, password=random_password())
|
||||
|
||||
stored = service.repository.get_credential(account.user_id)
|
||||
assert stored is not None
|
||||
@@ -294,7 +309,7 @@ def test_repeated_failures_lock_the_credential(service: AccountService, settings
|
||||
|
||||
# Even the correct password is refused while locked, with the same error.
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username=account.username, password=account.password)
|
||||
service.authenticate(email=account.email, password=account.password)
|
||||
|
||||
events, _ = service.repository.list_audit_events(target_user_id=account.user_id)
|
||||
assert any(event.detail.get("lock_state") == "locked" for event in events)
|
||||
@@ -303,8 +318,8 @@ def test_repeated_failures_lock_the_credential(service: AccountService, settings
|
||||
def test_successful_login_clears_the_failure_counter(service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username=account.username, password=random_password())
|
||||
service.authenticate(username=account.username, password=account.password)
|
||||
service.authenticate(email=account.email, password=random_password())
|
||||
service.authenticate(email=account.email, password=account.password)
|
||||
|
||||
stored = service.repository.get_credential(account.user_id)
|
||||
assert stored is not None
|
||||
@@ -319,8 +334,8 @@ def test_successful_login_clears_the_failure_counter(service: AccountService) ->
|
||||
|
||||
def test_first_login_change_rotates_the_password_and_revokes_old_sessions(service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
first = service.authenticate(username=account.username, password=account.password)
|
||||
second = service.authenticate(username=account.username, password=account.password)
|
||||
first = service.authenticate(email=account.email, password=account.password)
|
||||
second = service.authenticate(email=account.email, password=account.password)
|
||||
principal = service.resolve_session(first.session.token)
|
||||
new_password = random_password(28)
|
||||
|
||||
@@ -335,16 +350,16 @@ def test_first_login_change_rotates_the_password_and_revokes_old_sessions(servic
|
||||
service.resolve_session(second.session.token)
|
||||
# The temporary password no longer works.
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username=account.username, password=account.password)
|
||||
service.authenticate(email=account.email, password=account.password)
|
||||
# The new password does, and no longer forces a change.
|
||||
reopened = service.authenticate(username=account.username, password=new_password)
|
||||
reopened = service.authenticate(email=account.email, password=new_password)
|
||||
assert reopened.session.must_change_password is False
|
||||
assert service.resolve_session(result.session.token).must_change_password is False
|
||||
|
||||
|
||||
def test_first_login_change_requires_the_temporary_password(service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
result = service.authenticate(username=account.username, password=account.password)
|
||||
result = service.authenticate(email=account.email, password=account.password)
|
||||
principal = service.resolve_session(result.session.token)
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.change_first_password(principal, current_password=random_password(), new_password=random_password(28))
|
||||
@@ -352,7 +367,7 @@ def test_first_login_change_requires_the_temporary_password(service: AccountServ
|
||||
|
||||
def test_first_login_change_rejects_password_reuse_and_weak_values(service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
result = service.authenticate(username=account.username, password=account.password)
|
||||
result = service.authenticate(email=account.email, password=account.password)
|
||||
principal = service.resolve_session(result.session.token)
|
||||
|
||||
with pytest.raises(PasswordPolicyError) as reuse:
|
||||
@@ -370,8 +385,8 @@ def test_first_login_change_rejects_password_reuse_and_weak_values(service: Acco
|
||||
|
||||
def test_logout_revokes_only_the_current_session(service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
first = service.authenticate(username=account.username, password=account.password)
|
||||
second = service.authenticate(username=account.username, password=account.password)
|
||||
first = service.authenticate(email=account.email, password=account.password)
|
||||
second = service.authenticate(email=account.email, password=account.password)
|
||||
principal = service.resolve_session(first.session.token)
|
||||
|
||||
service.logout(principal)
|
||||
@@ -383,9 +398,7 @@ def test_logout_revokes_only_the_current_session(service: AccountService) -> Non
|
||||
|
||||
def test_logout_everywhere_revokes_all_sessions(service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
tokens = [
|
||||
service.authenticate(username=account.username, password=account.password).session.token for _ in range(3)
|
||||
]
|
||||
tokens = [service.authenticate(email=account.email, password=account.password).session.token for _ in range(3)]
|
||||
principal = service.resolve_session(tokens[0])
|
||||
|
||||
revoked = service.logout_everywhere(principal)
|
||||
@@ -398,7 +411,7 @@ def test_logout_everywhere_revokes_all_sessions(service: AccountService) -> None
|
||||
|
||||
def test_expired_sessions_are_rejected_and_revoked(service: AccountService, settings: object) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
result = service.authenticate(username=account.username, password=account.password)
|
||||
result = service.authenticate(email=account.email, password=account.password)
|
||||
stored = service.repository.get_session(result.session.session_id)
|
||||
assert stored is not None
|
||||
|
||||
@@ -426,7 +439,7 @@ def test_unknown_and_forged_tokens_are_rejected(service: AccountService) -> None
|
||||
def test_suspension_revokes_sessions_and_blocks_new_logins(repository: object, service: AccountService) -> None:
|
||||
actor = admin_principal(provision_local(service, username="Umbraadmin"))
|
||||
member = provision_local(service, username="Member.User", role=str(RoleName.MEMBER))
|
||||
member_result = service.authenticate(username=member.username, password=member.password)
|
||||
member_result = service.authenticate(email=member.email, password=member.password)
|
||||
assert service.resolve_session(member_result.session.token).user_id == member.user_id
|
||||
|
||||
view, revoked = service.set_account_status(actor, member.user_id, "suspended", reason="policy review")
|
||||
@@ -436,7 +449,7 @@ def test_suspension_revokes_sessions_and_blocks_new_logins(repository: object, s
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.resolve_session(member_result.session.token)
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username=member.username, password=member.password)
|
||||
service.authenticate(email=member.email, password=member.password)
|
||||
|
||||
|
||||
def test_reactivation_restores_access(service: AccountService) -> None:
|
||||
@@ -447,7 +460,7 @@ def test_reactivation_restores_access(service: AccountService) -> None:
|
||||
view, _ = service.set_account_status(actor, member.user_id, "active")
|
||||
assert view.status is AccountStatus.ACTIVE
|
||||
|
||||
result = service.authenticate(username=member.username, password=member.password)
|
||||
result = service.authenticate(email=member.email, password=member.password)
|
||||
assert service.resolve_session(result.session.token).user_id == member.user_id
|
||||
|
||||
|
||||
@@ -455,7 +468,7 @@ def test_administrative_session_revocation(service: AccountService) -> None:
|
||||
admin = provision_local(service, username="Umbraadmin")
|
||||
actor = admin_principal(admin)
|
||||
member = provision_local(service, username="Member.User", role=str(RoleName.MEMBER))
|
||||
result = service.authenticate(username=member.username, password=member.password)
|
||||
result = service.authenticate(email=member.email, password=member.password)
|
||||
|
||||
assert service.revoke_sessions(actor, member.user_id) == 1
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
@@ -550,7 +563,7 @@ def test_organization_admin_cannot_promote_to_platform_admin(service: AccountSer
|
||||
|
||||
def test_password_change_principal_cannot_use_admin_operations(service: AccountService) -> None:
|
||||
admin = provision_local(service, username="Umbraadmin")
|
||||
result = service.authenticate(username=admin.username, password=admin.password)
|
||||
result = service.authenticate(email=admin.email, password=admin.password)
|
||||
principal = service.resolve_session(result.session.token)
|
||||
assert principal.must_change_password is True
|
||||
with pytest.raises(PasswordChangeRequiredError):
|
||||
@@ -601,7 +614,7 @@ def test_audit_events_are_sanitized_and_scoped(service: AccountService) -> None:
|
||||
def test_audit_detail_never_contains_a_password_value(service: AccountService) -> None:
|
||||
account = provision_local(service, username="Umbraadmin")
|
||||
with pytest.raises(InvalidCredentialsError):
|
||||
service.authenticate(username=account.username, password=random_password())
|
||||
service.authenticate(email=account.email, password=random_password())
|
||||
events, _ = service.repository.list_audit_events(target_user_id=account.user_id)
|
||||
for event in events:
|
||||
for value in event.detail.values():
|
||||
@@ -655,7 +668,14 @@ def test_username_and_display_name_validation() -> None:
|
||||
|
||||
assert normalize_username(" Umbra.admin-1 ") == "Umbra.admin-1"
|
||||
assert username_key("UmbraAdmin") == "umbraadmin"
|
||||
for bad in ("", "ab", "with space", "with/slash", "x" * 65):
|
||||
# Optional handle: absent, empty, and whitespace are all the same thing,
|
||||
# because the email is the login identifier and the handle is only for
|
||||
# readability. A handle that is present must still be well formed.
|
||||
assert normalize_username(None) is None
|
||||
assert normalize_username("") is None
|
||||
assert normalize_username(" ") is None
|
||||
assert username_key(None) is None
|
||||
for bad in ("ab", "with space", "with/slash", "x" * 65):
|
||||
with pytest.raises(InputError):
|
||||
normalize_username(bad)
|
||||
assert normalize_email(" User@Example.COM ") == "user@example.com"
|
||||
|
||||
Reference in New Issue
Block a user