Login tokens

commons.users.login_tokens

Passwordless login with a link or code sent to a registered user.

Overview

• LoginTokenService: The flow. request_login(email, ...) creates a request with a random link token and a numeric code and hands both to a sender; redeem_link(token) / redeem_code(request_id, code) return the user to log in. Single use, short-lived, rate limited.
• LoginTokenStore: Where requests are kept (hashes only, never the secrets). InMemoryLoginTokenStore for a single process, SQLLoginTokenStore for several processes or workers.
• SendLoginTokenPort: How link and code reach the user. The default LoggingLoginTokenSender writes them to the log (development only); SMTPLoginTokenSender sends them by email via the SMTP server of app.smtp (see :mod:commons.users.login_tokens.smtp).
• LoginTokenSettings: Validity, code length and the rate limit, read from app.login_tokens.*.

The module is synchronous like :mod:commons.users; async apps call it via asyncio.to_thread. Users are looked up with UsersRepository.find_user_by_email.

Example::

from commons.users.login_tokens import InMemoryLoginTokenStore, LoginTokenService

service = LoginTokenService(users, InMemoryLoginTokenStore(), secret_key=app_secret)

request = service.request_login(
    "jane@example.com",
    link_for=lambda token: f"https://app.example.com/login/token?t={token}",
)
# Show the same message whatever request.status is.

result = service.redeem_code(request.request_id, "123456")
if result.ok:
    ...  # log result.user in

Modules:

Name	Description
memory	In-memory storage of login requests, for a single process.
model	Data types of the login-token flow.
sender	Delivery of login links and codes to the user.
service	The login-token flow: request a link and code, redeem one of them.
smtp	Sending login links and codes by email via SMTP.
sql	SQL storage of login requests, shared by several processes.
store	Abstract storage of login requests.

Classes:

Name	Description
InMemoryLoginTokenStore	Keeps login requests in a dict, guarded by a lock.
LoginRequest	Result of LoginTokenService.request_login.
LoginTokenRequest	One login request of a registered user, as kept by a LoginTokenStore.
RedeemResult	Result of LoginTokenService.redeem_link / redeem_code.
RedeemStatus	Outcome of redeeming a link or a code.
RequestStatus	Outcome of LoginTokenService.request_login.
LoggingLoginTokenSender	Writes login links and codes to the log instead of sending them.
LoginTokenMessage	What to deliver to the user.
SendLoginTokenPort	Delivers a login link and code to a user.
LoginTokenService	Lets registered users log in with a link or code sent to their email
LoginTokenSettings	Settings of the login-token flow.
LoginTokenMail	The login mail to send.
SMTPLoginTokenSender	Sends login links and codes by email via SMTP.
SQLLoginTokenStore	Keeps login requests in the login_token_requests table.
LoginTokenStore	Abstract base class for storing login requests.

Attributes:

Name	Type	Description
ComposeLoginTokenMail		Builds the login mail for a message, e.g. in the user's language.

ComposeLoginTokenMail module-attribute

ComposeLoginTokenMail = Callable[
    [LoginTokenMessage], LoginTokenMail
]

Builds the login mail for a message, e.g. in the user's language.

__all__ module-attribute

__all__ = [
    "ComposeLoginTokenMail",
    "InMemoryLoginTokenStore",
    "LoggingLoginTokenSender",
    "LoginRequest",
    "LoginTokenMail",
    "LoginTokenMessage",
    "LoginTokenRequest",
    "LoginTokenService",
    "LoginTokenSettings",
    "LoginTokenStore",
    "RedeemResult",
    "RedeemStatus",
    "RequestStatus",
    "SMTPLoginTokenSender",
    "SQLLoginTokenStore",
    "SendLoginTokenPort",
]

InMemoryLoginTokenStore

InMemoryLoginTokenStore()

Bases: LoginTokenStore

Keeps login requests in a dict, guarded by a lock.

Suited for development, tests and apps running a single process. With several processes or workers, each would have its own requests (a link sent by one worker wouldn't work on another) and its own rate limits; use SQLLoginTokenStore there.

Methods:

Name	Description
add
get
find_by_link_hash
consume
record_failed_attempt
count_requests
delete_expired

Source code in src/commons/users/login_tokens/memory.py

def __init__(self) -> None:
    self._requests: dict[str, LoginTokenRequest] = {}
    self._lock = threading.Lock()

add

add(request: LoginTokenRequest) -> None

Source code in src/commons/users/login_tokens/memory.py

def add(self, request: LoginTokenRequest) -> None:
    with self._lock:
        self._requests[request.id] = replace(request)

get

get(request_id: str) -> LoginTokenRequest | None

Source code in src/commons/users/login_tokens/memory.py

def get(self, request_id: str) -> LoginTokenRequest | None:
    with self._lock:
        request = self._requests.get(request_id)
        return replace(request) if request else None

find_by_link_hash

find_by_link_hash(
    link_hash: str,
) -> LoginTokenRequest | None

Source code in src/commons/users/login_tokens/memory.py

def find_by_link_hash(self, link_hash: str) -> LoginTokenRequest | None:
    with self._lock:
        for request in self._requests.values():
            if request.link_hash == link_hash:
                return replace(request)
        return None

consume

consume(request_id: str, now: datetime) -> bool

Source code in src/commons/users/login_tokens/memory.py

def consume(self, request_id: str, now: datetime) -> bool:
    with self._lock:
        request = self._requests.get(request_id)
        if request is None or request.consumed_at is not None or request.expires_at <= now:
            return False
        request.consumed_at = now
        return True

record_failed_attempt

record_failed_attempt(request_id: str) -> int

Source code in src/commons/users/login_tokens/memory.py

def record_failed_attempt(self, request_id: str) -> int:
    with self._lock:
        request = self._requests.get(request_id)
        if request is None:
            return 0
        request.failed_attempts += 1
        return request.failed_attempts

count_requests

count_requests(*, email_hash: str, since: datetime) -> int

Source code in src/commons/users/login_tokens/memory.py

def count_requests(
    self,
    *,
    email_hash: str,
    since: datetime,
) -> int:
    with self._lock:
        return sum(
            1
            for request in self._requests.values()
            if request.email_hash == email_hash and request.created_at >= since
        )

delete_expired

delete_expired(before: datetime) -> int

Source code in src/commons/users/login_tokens/memory.py

def delete_expired(self, before: datetime) -> int:
    with self._lock:
        expired = [request_id for request_id, r in self._requests.items() if r.expires_at < before]
        for request_id in expired:
            del self._requests[request_id]
        return len(expired)

LoginRequest dataclass

LoginRequest(
    request_id: str,
    status: RequestStatus,
    expires_at: datetime,
)

Result of LoginTokenService.request_login.

Attributes:

Name	Type	Description
request_id	str	Identifier to redeem the code with (keep it, e.g. in the session). Present for every status, so callers can treat all outcomes alike; for an unknown address or a rate-limited request it's random and stored nowhere, so a code for it fails.
status	RequestStatus	What happened; don't show it to the user (see RequestStatus).
expires_at	datetime	When the link and the code expire.

request_id instance-attribute

request_id: str

status instance-attribute

status: RequestStatus

expires_at instance-attribute

expires_at: datetime

LoginTokenRequest dataclass

LoginTokenRequest(
    id: str,
    user_id: str,
    email_hash: str,
    link_hash: str,
    code_hash: str,
    created_at: datetime,
    expires_at: datetime,
    redirect_to: str | None = None,
    failed_attempts: int = 0,
    consumed_at: datetime | None = None,
)

One login request of a registered user, as kept by a LoginTokenStore.

Requests for unknown email addresses are not stored: nothing is kept about people who aren't users.

Attributes:

Name	Type	Description
id	str	Random identifier of the request. Callers keep it (e.g. in the session) to redeem the code in the same browser.
email_hash	str	HMAC-SHA256 of the lower-cased email address, for the per-address rate limit.
created_at	datetime	When the request was made (timezone-aware, UTC).
expires_at	datetime	When the link and the code stop working (UTC).
user_id	str	The user to log in.
link_hash	str	HMAC-SHA256 of the link token.
code_hash	str	HMAC-SHA256 of "{id}:{code}".
redirect_to	str | None	Where the caller wants to go after logging in.
failed_attempts	int	Number of wrong codes entered so far.
consumed_at	datetime | None	When the request was redeemed (or invalidated); None while it's still usable.

id instance-attribute

id: str

user_id instance-attribute

user_id: str

email_hash instance-attribute

email_hash: str

link_hash instance-attribute

link_hash: str

code_hash instance-attribute

code_hash: str

created_at instance-attribute

created_at: datetime

expires_at instance-attribute

expires_at: datetime

redirect_to class-attribute instance-attribute

redirect_to: str | None = None

failed_attempts class-attribute instance-attribute

failed_attempts: int = 0

consumed_at class-attribute instance-attribute

consumed_at: datetime | None = None

RedeemResult dataclass

RedeemResult(
    status: RedeemStatus,
    user: RegisteredUser | None = None,
    redirect_to: str | None = None,
)

Result of LoginTokenService.redeem_link / redeem_code.

Attributes:

Name	Type	Description
status	RedeemStatus	Whether the redemption succeeded and, if not, why.
user	RegisteredUser | None	The user to log in; only set when status is OK.
redirect_to	str | None	The redirect_to given when the login was requested; only set when status is OK.

status instance-attribute

status: RedeemStatus

user class-attribute instance-attribute

user: RegisteredUser | None = None

redirect_to class-attribute instance-attribute

redirect_to: str | None = None

ok property

ok: bool

True if the user may be logged in.

RedeemStatus

Bases: StrEnum

Outcome of redeeming a link or a code.

Attributes:

Name	Type	Description
OK		The user is logged in.
INVALID		Unknown link or request, or a wrong code.
EXPIRED		The link or code is too old.
USED		The request was already redeemed (or invalidated).
TOO_MANY_ATTEMPTS		Too many wrong codes; the request is invalidated.

OK class-attribute instance-attribute

OK = 'ok'

The user is logged in.

INVALID class-attribute instance-attribute

INVALID = 'invalid'

Unknown link or request, or a wrong code.

EXPIRED class-attribute instance-attribute

EXPIRED = 'expired'

The link or code is too old.

USED class-attribute instance-attribute

USED = 'used'

The request was already redeemed (or invalidated).

TOO_MANY_ATTEMPTS class-attribute instance-attribute

TOO_MANY_ATTEMPTS = 'too_many_attempts'

Too many wrong codes; the request is invalidated.

RequestStatus

Bases: StrEnum

Outcome of LoginTokenService.request_login.

Callers should show the same message for every status ("If an account exists, we've sent you a link"), so the response doesn't reveal which addresses are registered or rate limited.

Attributes:

Name	Type	Description
SENT		A link and a code were sent to the user.
UNKNOWN_EMAIL		No user has this address; nothing was sent or stored.
RATE_LIMITED		Too many requests for this address; nothing was sent.

SENT class-attribute instance-attribute

SENT = 'sent'

A link and a code were sent to the user.

UNKNOWN_EMAIL class-attribute instance-attribute

UNKNOWN_EMAIL = 'unknown_email'

No user has this address; nothing was sent or stored.

RATE_LIMITED class-attribute instance-attribute

RATE_LIMITED = 'rate_limited'

Too many requests for this address; nothing was sent.

LoggingLoginTokenSender

LoggingLoginTokenSender(
    logger: Logger | None = None, level: int = WARNING
)

Writes login links and codes to the log instead of sending them.

The default sender: lets an app use login tokens in development without a mail server. Never use it in production — anyone with access to the logs could log in as any user.

Parameters:

Name	Type	Description	Default
logger	Logger | None	The logger to write to; defaults to this module's logger.	None
level	int	The log level; defaults to WARNING, so the line shows up with default logging configuration.	WARNING

Methods:

Name	Description
send_login_token

Source code in src/commons/users/login_tokens/sender.py

def __init__(self, logger: logging.Logger | None = None, level: int = logging.WARNING) -> None:
    self._logger = logger or LOG
    self._level = level

send_login_token

send_login_token(message: LoginTokenMessage) -> None

Source code in src/commons/users/login_tokens/sender.py

def send_login_token(self, message: LoginTokenMessage) -> None:
    self._logger.log(
        self._level,
        "Login token for %s (%s): code %s, link %s (valid until %s)",
        message.user.username,
        message.email,
        message.code,
        message.link,
        message.expires_at.isoformat(timespec="seconds"),
    )

LoginTokenMessage dataclass

LoginTokenMessage(
    user: RegisteredUser,
    email: str,
    link: str,
    code: str,
    expires_at: datetime,
    locale: str | None = None,
)

What to deliver to the user.

Attributes:

Name	Type	Description
user	RegisteredUser	The user who requested the login.
email	str	The address the user entered (and the message goes to).
link	str	Absolute URL that logs the user in when opened.
code	str	Code the user can type in instead, in the browser that requested the login.
expires_at	datetime	When the link and the code stop working (UTC).
locale	str | None	The language to write the message in (e.g. "de"), as the caller resolved it for the user; None lets the sender choose.

user instance-attribute

user: RegisteredUser

email instance-attribute

email: str

link instance-attribute

link: str

code instance-attribute

code: str

expires_at instance-attribute

expires_at: datetime

locale class-attribute instance-attribute

locale: str | None = None

SendLoginTokenPort

Bases: Protocol

Delivers a login link and code to a user.

Implementations must not swallow errors: if the message can't be delivered, raise, so the caller can tell the user to try again.

Example::

class TeamsLoginTokenSender:
    def send_login_token(self, message: LoginTokenMessage) -> None:
        post_to_teams(message.user, f"Your code: {message.code}")

Methods:

Name	Description
send_login_token	Deliver the message to the user.

send_login_token

send_login_token(message: LoginTokenMessage) -> None

Deliver the message to the user.

Parameters:

Name	Type	Description	Default
message	LoginTokenMessage	The link, code and expiry to deliver.	required

Source code in src/commons/users/login_tokens/sender.py

def send_login_token(self, message: LoginTokenMessage) -> None:
    """
    Deliver the message to the user.

    Args:
        message: The link, code and expiry to deliver.
    """
    ...

LoginTokenService

LoginTokenService(
    users: UsersRepository,
    store: LoginTokenStore,
    *,
    secret_key: str,
    sender: SendLoginTokenPort | None = None,
    settings: LoginTokenSettings | None = None,
    clock: Callable[[], datetime] | None = None
)

Lets registered users log in with a link or code sent to their email address.

request_login creates a request with a random link token and a numeric code, stores only their hashes, and hands both to the sender. The user then either opens the link (redeem_link) or types the code into the browser that requested it (redeem_code). Redeeming one invalidates the other; both expire after token_ttl.

The rate limit counts the stored requests per address (against flooding one person's inbox). No client data such as IP addresses is stored. Wrong codes count per request; after max_code_attempts the request is invalidated, which makes guessing a 6-digit code hopeless.

All methods are synchronous; call them via asyncio.to_thread from async code.

Everything stored is an HMAC-SHA256 keyed with secret_key: without the key, neither an address can be confirmed from email_hash (as a plain hash of known addresses would allow) nor a 6-digit code be brute-forced from code_hash.

Parameters:

Name	Type	Description	Default
users	UsersRepository	Repository to look users up by email and id.	required
store	LoginTokenStore	Where requests are kept.	required
secret_key	str	Key for hashing what is stored (e.g. the app's session secret); at least 16 characters, and the same for every process sharing the store.	required
sender	SendLoginTokenPort | None	Delivers link and code; defaults to LoggingLoginTokenSender (development only).	None
settings	LoginTokenSettings | None	Defaults to LoginTokenSettings.read().	None
clock	Callable[[], datetime] | None	Returns the current time (UTC); for tests.	None

Example::

service = LoginTokenService(users, InMemoryLoginTokenStore(), secret_key=app_secret)

request = service.request_login(
    "jane@example.com",
    link_for=lambda token: f"https://app.example.com/login/token?t={token}",
)
# keep request.request_id, e.g. in the session

result = service.redeem_code(request.request_id, "123456")
if result.ok:
    log_in(result.user)

Methods:

Name	Description
request_login	Request a login link and code for an email address.
request_id_for_link	Find out which request a login link belongs to, without redeeming it,
redeem_link	Redeem the token from a login link.
redeem_code	Redeem a code typed in by the user.
purge_expired	Delete requests no longer needed, neither to log in nor for the

Attributes:

Name	Type	Description
settings	LoginTokenSettings

Source code in src/commons/users/login_tokens/service.py

def __init__(
    self,
    users: UsersRepository,
    store: LoginTokenStore,
    *,
    secret_key: str,
    sender: SendLoginTokenPort | None = None,
    settings: LoginTokenSettings | None = None,
    clock: Callable[[], datetime] | None = None,
) -> None:
    if len(secret_key) < 16:
        raise ValueError("`secret_key` must be at least 16 characters long.")

    self._key = secret_key.encode("utf-8")
    self._users = users
    self._store = store
    self._sender = sender or LoggingLoginTokenSender()
    self._settings = settings or LoginTokenSettings.read()
    self._clock = clock or (lambda: datetime.now(UTC))

settings property

settings: LoginTokenSettings

request_login

request_login(
    email: str,
    *,
    link_for: Callable[[str], str],
    redirect_to: str | None = None,
    locale: str | None = None
) -> LoginRequest

Request a login link and code for an email address.

Returns a request_id for every outcome, and callers should show the same message for every status, so the response doesn't reveal whether the address is registered.

Parameters:

Name	Type	Description	Default
email	str	The address the user entered.	required
link_for	Callable[[str], str]	Builds the absolute login URL from the link token.	required
redirect_to	str | None	Where to go after logging in; returned by the redeem methods.	None
locale	str | None	The language of the message (LoginTokenMessage.locale), e.g. as pyhx resolves it for the user; None lets the sender choose.	None

Returns:

Type	Description
LoginRequest	The request id, what happened, and when the link expires.

Raises:

Type	Description
Exception	Whatever the sender raises if delivery fails.

Source code in src/commons/users/login_tokens/service.py

def request_login(
    self,
    email: str,
    *,
    link_for: Callable[[str], str],
    redirect_to: str | None = None,
    locale: str | None = None,
) -> LoginRequest:
    """
    Request a login link and code for an email address.

    Returns a ``request_id`` for every outcome, and callers should show
    the same message for every status, so the response doesn't reveal
    whether the address is registered.

    Args:
        email: The address the user entered.
        link_for: Builds the absolute login URL from the link token.
        redirect_to: Where to go after logging in; returned by the
            redeem methods.
        locale: The language of the message (``LoginTokenMessage.locale``),
            e.g. as pyhx resolves it for the user; ``None`` lets the sender
            choose.

    Returns:
        The request id, what happened, and when the link expires.

    Raises:
        Exception: Whatever the sender raises if delivery fails.
    """
    now = self._clock()
    expires_at = now + timedelta(seconds=self._settings.token_ttl)
    normalized = _normalize_email(email)
    email_hash = self._hash(normalized)

    # Unknown addresses and rate-limited requests are not stored; their
    # random request id exists nowhere, so a code entered for it fails.
    user = self._users.find_user_by_email(normalized)
    if user is None:
        return LoginRequest(uuid4().hex, RequestStatus.UNKNOWN_EMAIL, expires_at)

    if self._is_rate_limited(now, email_hash):
        return LoginRequest(uuid4().hex, RequestStatus.RATE_LIMITED, expires_at)

    request_id = uuid4().hex
    link_token = secrets.token_urlsafe(32)
    code = str(secrets.randbelow(10**self._settings.code_length)).zfill(self._settings.code_length)

    request = LoginTokenRequest(
        id=request_id,
        user_id=user.id,
        email_hash=email_hash,
        link_hash=self._hash(link_token),
        code_hash=self._hash(f"{request_id}:{code}"),
        created_at=now,
        expires_at=expires_at,
        redirect_to=redirect_to,
    )
    self._store.add(request)

    self._sender.send_login_token(
        LoginTokenMessage(
            user=user,
            email=email.strip(),
            link=link_for(link_token),
            code=code,
            expires_at=expires_at,
            locale=locale,
        )
    )
    return LoginRequest(request.id, RequestStatus.SENT, expires_at)

request_id_for_link

request_id_for_link(link_token: str) -> str | None

Find out which request a login link belongs to, without redeeming it, as long as the link can still be redeemed.

Lets a web app log a user in right away when the link is opened in the browser that requested it (its session holds the request id), and ask for a confirmation click otherwise — mail scanners open links in advance, but never with the requesting browser's session.

Parameters:

Name	Type	Description	Default
link_token	str	The token from the link.	required

Returns:

Type	Description
str | None	The request id, or None if no request has this link or it has
str | None	been redeemed or has expired.

Source code in src/commons/users/login_tokens/service.py

def request_id_for_link(self, link_token: str) -> str | None:
    """
    Find out which request a login link belongs to, without redeeming it,
    as long as the link can still be redeemed.

    Lets a web app log a user in right away when the link is opened in
    the browser that requested it (its session holds the request id),
    and ask for a confirmation click otherwise — mail scanners open links
    in advance, but never with the requesting browser's session.

    Args:
        link_token: The token from the link.

    Returns:
        The request id, or None if no request has this link or it has
        been redeemed or has expired.
    """
    request = self._store.find_by_link_hash(self._hash(link_token))
    if request is None or request.consumed_at is not None or request.expires_at <= self._clock():
        return None
    return request.id

redeem_link

redeem_link(link_token: str) -> RedeemResult

Redeem the token from a login link.

Parameters:

Name	Type	Description	Default
link_token	str	The token from the link.	required

Returns:

Type	Description
RedeemResult	OK with the user, or why the link doesn't work.

Source code in src/commons/users/login_tokens/service.py

def redeem_link(self, link_token: str) -> RedeemResult:
    """
    Redeem the token from a login link.

    Args:
        link_token: The token from the link.

    Returns:
        ``OK`` with the user, or why the link doesn't work.
    """
    request = self._store.find_by_link_hash(self._hash(link_token))
    if request is None:
        return RedeemResult(RedeemStatus.INVALID)
    return self._redeem(request)

redeem_code

redeem_code(request_id: str, code: str) -> RedeemResult

Redeem a code typed in by the user.

Parameters:

Name	Type	Description	Default
request_id	str	The id returned by request_login.	required
code	str	The code the user entered.	required

Returns:

Type	Description
RedeemResult	OK with the user, or why the code doesn't work.

Source code in src/commons/users/login_tokens/service.py

def redeem_code(self, request_id: str, code: str) -> RedeemResult:
    """
    Redeem a code typed in by the user.

    Args:
        request_id: The id returned by ``request_login``.
        code: The code the user entered.

    Returns:
        ``OK`` with the user, or why the code doesn't work.
    """
    request = self._store.get(request_id)
    if request is None:
        return RedeemResult(RedeemStatus.INVALID)
    if request.consumed_at is not None:
        return RedeemResult(self._consumed_status(request))
    if request.expires_at <= self._clock():
        return RedeemResult(RedeemStatus.EXPIRED)

    entered = self._hash(f"{request.id}:{code.strip()}")
    if not hmac.compare_digest(request.code_hash, entered):
        attempts = self._store.record_failed_attempt(request.id)
        if attempts >= self._settings.max_code_attempts:
            self._store.consume(request.id, self._clock())
            return RedeemResult(RedeemStatus.TOO_MANY_ATTEMPTS)
        return RedeemResult(RedeemStatus.INVALID)

    return self._redeem(request)

purge_expired

purge_expired() -> int

Delete requests no longer needed, neither to log in nor for the rate limit.

Returns:

Type	Description
int	The number of deleted requests.

Source code in src/commons/users/login_tokens/service.py

def purge_expired(self) -> int:
    """
    Delete requests no longer needed, neither to log in nor for the
    rate limit.

    Returns:
        The number of deleted requests.
    """
    keep = max(self._settings.token_ttl, self._settings.email_window)
    return self._store.delete_expired(self._clock() - timedelta(seconds=keep))

LoginTokenSettings

LoginTokenSettings(
    token_ttl: int = 15 * 60,
    code_length: int = 6,
    max_code_attempts: int = 5,
    max_requests_per_email: int = 3,
    email_window: int = 15 * 60,
    mail_subject: str | None = None,
    mail_text: str | None = None,
    mail_html: str | None = None,
    **_kwargs: Any
)

Settings of the login-token flow.

Read from app.login_tokens.* (see :mod:commons.settings); every field has a default, so no configuration is needed.

TOML::

[app.login_tokens]
token_ttl = 900  # link and code validity, seconds
code_length = 6  # digits of the code
max_code_attempts = 5  # wrong codes before the request is invalidated
max_requests_per_email = 3  # requests per address ...
email_window = 900  # ... within this many seconds
mail_subject = "..."  # optional: the login mail of SMTPLoginTokenSender,
mail_text = "..."  # see commons.users.login_tokens.smtp
mail_html = "..."

Env var overrides follow the standard pattern, e.g. APP__LOGIN_TOKENS__TOKEN_TTL.

Methods:

Name	Description
read	Read the settings; defaults if app.login_tokens is not configured.

Attributes:

Name	Type	Description
token_ttl
code_length
max_code_attempts
max_requests_per_email
email_window
mail_subject
mail_text
mail_html

Source code in src/commons/users/login_tokens/service.py

def __init__(
    self,
    token_ttl: int = 15 * 60,
    code_length: int = 6,
    max_code_attempts: int = 5,
    max_requests_per_email: int = 3,
    email_window: int = 15 * 60,
    mail_subject: str | None = None,
    mail_text: str | None = None,
    mail_html: str | None = None,
    **_kwargs: Any,
) -> None:
    self.token_ttl = int(token_ttl)
    self.code_length = int(code_length)
    self.max_code_attempts = int(max_code_attempts)
    self.max_requests_per_email = int(max_requests_per_email)
    self.email_window = int(email_window)
    self.mail_subject = mail_subject
    self.mail_text = mail_text
    self.mail_html = mail_html

token_ttl instance-attribute

token_ttl = int(token_ttl)

code_length instance-attribute

code_length = int(code_length)

max_code_attempts instance-attribute

max_code_attempts = int(max_code_attempts)

max_requests_per_email instance-attribute

max_requests_per_email = int(max_requests_per_email)

email_window instance-attribute

email_window = int(email_window)

mail_subject instance-attribute

mail_subject = mail_subject

mail_text instance-attribute

mail_text = mail_text

mail_html instance-attribute

mail_html = mail_html

read staticmethod

read() -> LoginTokenSettings

Read the settings; defaults if app.login_tokens is not configured.

Source code in src/commons/users/login_tokens/service.py

@staticmethod
def read() -> "LoginTokenSettings":
    """Read the settings; defaults if ``app.login_tokens`` is not configured."""
    try:
        return read_settings_object("login_tokens", LoginTokenSettings)
    except KeyError:
        return LoginTokenSettings()

LoginTokenMail dataclass

LoginTokenMail(
    subject: str, text: str, html: str | None = None
)

The login mail to send.

Attributes:

Name	Type	Description
subject	str	Subject line.
text	str	Plain-text body.
html	str | None	Optional HTML body; the mail then has both versions.

subject instance-attribute

subject: str

text instance-attribute

text: str

html class-attribute instance-attribute

html: str | None = None

SMTPLoginTokenSender

SMTPLoginTokenSender(
    smtp: SMTPSettings | None = None,
    settings: LoginTokenSettings | None = None,
    compose: ComposeLoginTokenMail | None = None,
    language: Language = "en",
)

Sends login links and codes by email via SMTP.

Reads and checks its settings when it is created, so a misconfigured app fails at startup. Every delivery error is raised (connection, timeout, TLS, sign-in, refused sender or recipient); nothing is logged.

Parameters:

Name	Type	Description	Default
smtp	SMTPSettings | None	The SMTP connection; defaults to SMTPSettings.read().	None
settings	LoginTokenSettings | None	The templates (mail_subject, mail_text, mail_html) and the validity; defaults to LoginTokenSettings.read(). Not read if compose is given.	None
compose	ComposeLoginTokenMail | None	Builds the mail per message; defaults to the templates.	None
language	Language	Language of the built-in templates for messages without a built-in locale: "en", "de", "it" or "fr".	'en'

Raises:

Type	Description
ValueError	If language is unknown or a template has an unknown placeholder or an invalid $.
ValidationError	If app.smtp is missing or invalid.

Methods:

Name	Description
send_login_token

Source code in src/commons/users/login_tokens/smtp.py

def __init__(
    self,
    smtp: SMTPSettings | None = None,
    settings: LoginTokenSettings | None = None,
    compose: ComposeLoginTokenMail | None = None,
    language: Language = "en",
) -> None:
    if language not in _BUILT_IN:
        raise ValueError(f"Unknown language `{language}` of the login mail; use one of {sorted(_BUILT_IN)}.")
    self._smtp = smtp or SMTPSettings.read()
    self._compose = compose or _TemplateMail(settings or LoginTokenSettings.read(), language)

send_login_token

send_login_token(message: LoginTokenMessage) -> None

Source code in src/commons/users/login_tokens/smtp.py

def send_login_token(self, message: LoginTokenMessage) -> None:
    smtp = self._smtp
    mail = self._compose(message)

    email = EmailMessage()
    email["Subject"] = mail.subject
    email["From"] = smtp.from_address
    email["To"] = message.email
    email["Date"] = formatdate()
    email["Message-ID"] = make_msgid(domain=parseaddr(smtp.from_address)[1].rpartition("@")[2] or None)
    email.set_content(mail.text)
    if mail.html is not None:
        email.add_alternative(mail.html, subtype="html")

    context = ssl.create_default_context()
    connection: smtplib.SMTP
    if smtp.security == "ssl":
        connection = smtplib.SMTP_SSL(smtp.host, smtp.port or 465, timeout=smtp.timeout, context=context)
    else:
        connection = smtplib.SMTP(smtp.host, smtp.port or 587, timeout=smtp.timeout)

    with connection as server:
        if smtp.security == "starttls":
            server.starttls(context=context)  # raises SMTPNotSupportedError if not offered
        if smtp.username:
            server.login(smtp.username, smtp.password.get_secret_value())
        server.send_message(email)

SQLLoginTokenStore

SQLLoginTokenStore(engine: Engine)

Bases: LoginTokenStore

Keeps login requests in the login_token_requests table.

Use it when the app runs several processes or workers: they all see the same requests (a link sent by one worker works on every other) and share the rate limits. Creates its table on initialization if it doesn't exist.

Parameters:

Name	Type	Description	Default
engine	Engine	SQLAlchemy Engine connected to the database.	required

Example::

from sqlalchemy import create_engine
from commons.users.login_tokens import SQLLoginTokenStore

store = SQLLoginTokenStore(create_engine("sqlite:///app.db"))

Methods:

Name	Description
add
get
find_by_link_hash
consume
record_failed_attempt
count_requests
delete_expired

Source code in src/commons/users/login_tokens/sql.py

def __init__(self, engine: Engine) -> None:
    self._engine = engine
    self._create_session = sessionmaker(bind=self._engine)

    SQLModel.metadata.create_all(
        self._engine,
        tables=[LoginTokenRequestRow.__table__],  # type: ignore[attr-defined]
        checkfirst=True,
    )

add

add(request: LoginTokenRequest) -> None

Source code in src/commons/users/login_tokens/sql.py

def add(self, request: LoginTokenRequest) -> None:
    with self._create_session() as session:
        session.add(_to_row(request))
        session.commit()

get

get(request_id: str) -> LoginTokenRequest | None

Source code in src/commons/users/login_tokens/sql.py

def get(self, request_id: str) -> LoginTokenRequest | None:
    with self._create_session() as session:
        row = session.get(LoginTokenRequestRow, request_id)
        return _from_row(row) if row else None

find_by_link_hash

find_by_link_hash(
    link_hash: str,
) -> LoginTokenRequest | None

Source code in src/commons/users/login_tokens/sql.py

def find_by_link_hash(self, link_hash: str) -> LoginTokenRequest | None:
    with self._create_session() as session:
        row = session.execute(
            select(LoginTokenRequestRow).where(LoginTokenRequestRow.link_hash == link_hash)  # type: ignore[arg-type]
        ).scalar_one_or_none()
        return _from_row(row) if row else None

consume

consume(request_id: str, now: datetime) -> bool

Source code in src/commons/users/login_tokens/sql.py

def consume(self, request_id: str, now: datetime) -> bool:
    # A single conditional UPDATE: the database guarantees that only one
    # of two concurrent calls changes the row.
    db_now = _to_db(now)
    with self._create_session() as session:
        result = session.execute(
            update(LoginTokenRequestRow)
            .where(
                LoginTokenRequestRow.id == request_id,  # type: ignore[arg-type]
                LoginTokenRequestRow.consumed_at.is_(None),  # type: ignore[union-attr]
                LoginTokenRequestRow.expires_at > db_now,  # type: ignore[arg-type]
            )
            .values(consumed_at=db_now)
        )
        session.commit()
        return bool(result.rowcount)  # type: ignore[attr-defined]

record_failed_attempt

record_failed_attempt(request_id: str) -> int

Source code in src/commons/users/login_tokens/sql.py

def record_failed_attempt(self, request_id: str) -> int:
    with self._create_session() as session:
        session.execute(
            update(LoginTokenRequestRow)
            .where(LoginTokenRequestRow.id == request_id)  # type: ignore[arg-type]
            .values(failed_attempts=LoginTokenRequestRow.failed_attempts + 1)
        )
        session.commit()
        attempts = session.execute(
            select(col(LoginTokenRequestRow.failed_attempts)).where(
                LoginTokenRequestRow.id == request_id  # type: ignore[arg-type]
            )
        ).scalar_one_or_none()
        return attempts or 0

count_requests

count_requests(*, email_hash: str, since: datetime) -> int

Source code in src/commons/users/login_tokens/sql.py

def count_requests(self, *, email_hash: str, since: datetime) -> int:
    query = (
        select(func.count())
        .select_from(LoginTokenRequestRow)
        .where(
            LoginTokenRequestRow.email_hash == email_hash,  # type: ignore[arg-type]
            LoginTokenRequestRow.created_at >= _to_db(since),  # type: ignore[arg-type]
        )
    )

    with self._create_session() as session:
        return session.execute(query).scalar_one()

delete_expired

delete_expired(before: datetime) -> int

Source code in src/commons/users/login_tokens/sql.py

def delete_expired(self, before: datetime) -> int:
    with self._create_session() as session:
        rows = (
            session.execute(
                select(LoginTokenRequestRow).where(LoginTokenRequestRow.expires_at < _to_db(before))  # type: ignore[arg-type]
            )
            .scalars()
            .all()
        )
        for row in rows:
            session.delete(row)
        session.commit()
        return len(rows)

LoginTokenStore

Bases: ABC

Abstract base class for storing login requests.

Implementations must be safe to call from several threads (callers in async web apps run them via asyncio.to_thread). consume must be atomic: of two concurrent calls for the same request, only one may succeed, so a link or code can't be redeemed twice.

The rate limit of LoginTokenService is a count over the stored requests (count_requests), so a store shared by several processes (e.g. SQLLoginTokenStore) also shares the limits.

See Also

InMemoryLoginTokenStore: For a single process (default, tests). SQLLoginTokenStore: For several processes or workers.

Methods:

Name	Description
add	Store a new request.
get	Find a request by its id.
find_by_link_hash	Find a request by the hash of its link token.
consume	Mark a request as redeemed, if it's still usable.
record_failed_attempt	Count one more wrong code for a request.
count_requests	Count the requests for an address since a point in time, for rate
delete_expired	Delete requests that expired before a point in time.

add abstractmethod

add(request: LoginTokenRequest) -> None

Store a new request.

Parameters:

Name	Type	Description	Default
request	LoginTokenRequest	The request to store.	required

Source code in src/commons/users/login_tokens/store.py

@abc.abstractmethod
def add(self, request: LoginTokenRequest) -> None:
    """
    Store a new request.

    Args:
        request: The request to store.
    """

get abstractmethod

get(request_id: str) -> LoginTokenRequest | None

Find a request by its id.

Parameters:

Name	Type	Description	Default
request_id	str	The request's id.	required

Returns:

Type	Description
LoginTokenRequest | None	The request, or None if there is none with this id.

Source code in src/commons/users/login_tokens/store.py

@abc.abstractmethod
def get(self, request_id: str) -> LoginTokenRequest | None:
    """
    Find a request by its id.

    Args:
        request_id: The request's id.

    Returns:
        The request, or None if there is none with this id.
    """

find_by_link_hash abstractmethod

find_by_link_hash(
    link_hash: str,
) -> LoginTokenRequest | None

Find a request by the hash of its link token.

Parameters:

Name	Type	Description	Default
link_hash	str	HMAC-SHA256 of the link token (see LoginTokenService).	required

Returns:

Type	Description
LoginTokenRequest | None	The request, or None if no request has this link.

Source code in src/commons/users/login_tokens/store.py

@abc.abstractmethod
def find_by_link_hash(self, link_hash: str) -> LoginTokenRequest | None:
    """
    Find a request by the hash of its link token.

    Args:
        link_hash: HMAC-SHA256 of the link token (see ``LoginTokenService``).

    Returns:
        The request, or None if no request has this link.
    """

consume abstractmethod

consume(request_id: str, now: datetime) -> bool

Mark a request as redeemed, if it's still usable.

Atomic: succeeds only if the request exists, is not consumed yet and has not expired at now.

Parameters:

Name	Type	Description	Default
request_id	str	The request's id.	required
now	datetime	The current time (UTC).	required

Returns:

Type	Description
bool	True if this call consumed the request, False otherwise.

Source code in src/commons/users/login_tokens/store.py

@abc.abstractmethod
def consume(self, request_id: str, now: datetime) -> bool:
    """
    Mark a request as redeemed, if it's still usable.

    Atomic: succeeds only if the request exists, is not consumed yet and
    has not expired at ``now``.

    Args:
        request_id: The request's id.
        now: The current time (UTC).

    Returns:
        True if this call consumed the request, False otherwise.
    """

record_failed_attempt abstractmethod

record_failed_attempt(request_id: str) -> int

Count one more wrong code for a request.

Parameters:

Name	Type	Description	Default
request_id	str	The request's id.	required

Returns:

Type	Description
int	The number of failed attempts after this one; 0 if there is no
int	request with this id.

Source code in src/commons/users/login_tokens/store.py

@abc.abstractmethod
def record_failed_attempt(self, request_id: str) -> int:
    """
    Count one more wrong code for a request.

    Args:
        request_id: The request's id.

    Returns:
        The number of failed attempts after this one; 0 if there is no
        request with this id.
    """

count_requests abstractmethod

count_requests(*, email_hash: str, since: datetime) -> int

Count the requests for an address since a point in time, for rate limiting.

Parameters:

Name	Type	Description	Default
email_hash	str	Count requests for this address hash.	required
since	datetime	Count requests created at or after this time (UTC).	required

Returns:

Type	Description
int	The number of matching requests.

Source code in src/commons/users/login_tokens/store.py

@abc.abstractmethod
def count_requests(self, *, email_hash: str, since: datetime) -> int:
    """
    Count the requests for an address since a point in time, for rate
    limiting.

    Args:
        email_hash: Count requests for this address hash.
        since: Count requests created at or after this time (UTC).

    Returns:
        The number of matching requests.
    """

delete_expired abstractmethod

delete_expired(before: datetime) -> int

Delete requests that expired before a point in time.

Keep requests at least as long as the rate-limit window, or the limit counts too few.

Parameters:

Name	Type	Description	Default
before	datetime	Delete requests whose expires_at is before this time.	required

Returns:

Type	Description
int	The number of deleted requests.

Source code in src/commons/users/login_tokens/store.py

@abc.abstractmethod
def delete_expired(self, before: datetime) -> int:
    """
    Delete requests that expired before a point in time.

    Keep requests at least as long as the rate-limit window, or the limit
    counts too few.

    Args:
        before: Delete requests whose ``expires_at`` is before this time.

    Returns:
        The number of deleted requests.
    """

Sending by email

Sending login links and codes by email via SMTP.

SMTPLoginTokenSender implements SendLoginTokenPort: it connects to the server of app.smtp (:class:commons.settings.SMTPSettings), always encrypted, and sends one mail per login request.

The mail is built per message by a mail function (ComposeLoginTokenMail). The default one fills string.Template templates with these placeholders:

• $name: display name of the user, else the username
• $link: the login link
• $code: the login code
• $valid_minutes: how long link and code are valid
• $expires_at: when they expire, YYYY-MM-DD HH:MM UTC

Write $$ for a literal $; braces (e.g. CSS) need no escaping. Subject and text are built in for English ("en"), German ("de"), Italian ("it") and French ("fr"); there is no built-in HTML. Each mail is written in the language of its message (LoginTokenMessage.locale, e.g. the user's language as pyhx resolves it; "de-CH" counts as "de") if it is built in, else in the language of the sender. An app overrides the texts in app.login_tokens; its templates then apply to every language::

[app.login_tokens]
mail_subject = "Your code for CaseSight: $code"
mail_html = """
<p>Hello $name,</p>
<p>your code is <strong>$code</strong>, or <a href="$link">sign in</a>.</p>
"""

In the HTML, the values are HTML-escaped. For anything else (texts of your own per language, a template engine, a template file) pass your own mail function.

Wiring it into pyhx::

sender = SMTPLoginTokenSender(language="de")  # fallback language; reads app.smtp and app.login_tokens
EmailTokenLogin(store=SQLLoginTokenStore(engine), sender=sender)