feat: sign in with a code sent by email

A password is a thing to remember and a thing to lose. Somebody who can read
their own mail can now sign in without one: ask, receive six characters, type
them into the page that is already open.

A code rather than a link, and the difference is not cosmetic. The token in a
link was 256 bits, unguessable however long it lived, so its length, its expiry
and its rate limit were three independent decisions. Six characters is 2^30,
and the three stop being independent — so they are argued together:

  * six characters of the invite alphabet, imported rather than copied, because
    there should be one answer to which characters a person may be asked to
    retype and that one already drops O/0 and I/1;
  * a code answers five guesses and is then retired, not slowed — whoever is
    typing has lost the mail or does not own it, and both are one click from a
    new one;
  * one code live per person, since several would mean one guess tested against
    all of them;
  * ten verify attempts per address per fifteen minutes, so nobody buys five
    fresh guesses at a time by asking again.

Tens of guesses an hour against a billion, and the victim gets a mail for every
code burned. Eight characters would buy a thousandfold against an attack the
guess budget has already ended, and cost every person two more characters.

The attempt count lives in the row, not the cache. The Redis limiter fails open
when Redis is down, which is right for what it usually guards and wrong for the
only thing standing between a patient stranger and six characters.

Verifying is scoped to the address. A short code looked up on its own would be
tried against every code live on the site at once — the short code's one real
weakness, closed by knowing whose code it should be before comparing.

Fifteen minutes, because a first mail between strangers is routinely greylisted
five to ten and a code that expires before it arrives is not a sign-in method.
Shortening it buys nothing: one code is live and it answers five guesses
however long it sits there.

Nothing distinguishes an address with an account from one without — same
message, same status, same duration, and both rate limits counted before the
account is looked up, so a 429 cannot become the tell. Redis keys are
fingerprints, and the table holds a fingerprint rather than the code.

SSO stays first where it is configured, and a password is still one click away
for anybody who has one.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
This commit is contained in:
Daniel 2026-09-12 17:29:28 +02:00
parent 1c66eb84b1
commit 25a9a8aca4
19 changed files with 929 additions and 725 deletions

View file

@ -0,0 +1,54 @@
"""A code somebody types, in place of a link somebody clicks.
Replaces login_links rather than reshaping it. Every column changed meaning
the secret is now six characters instead of a 256-bit token, it is compared
against one account rather than looked up on its own, and it carries a count of
the guesses it has answered and the table holds nothing worth carrying
across: rows here are credentials that expire in fifteen minutes, so the worst
a drop costs anybody mid-flight is asking for another code.
Note what is deliberately not carried over: the unique index on the secret. Six
characters collide, rarely but really, and a unique constraint would turn that
into a failed sign-in for the second person. Uniqueness is not needed anyway
a code is only ever looked up against the account it was issued for.
Revision ID: g9a1b2c3d4e5
Revises: f8091a2b3c4d
"""
from alembic import op
revision = "g9a1b2c3d4e5"
down_revision = "f8091a2b3c4d"
branch_labels = None
depends_on = None
def upgrade():
op.execute("DROP TABLE IF EXISTS login_links")
op.execute("""
CREATE TABLE IF NOT EXISTS login_codes (
id SERIAL PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
code_hash VARCHAR(64) NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
expires_at TIMESTAMP NOT NULL,
consumed_at TIMESTAMP,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)""")
# The one lookup this table serves: the live code belonging to one person.
op.execute("CREATE INDEX IF NOT EXISTS ix_login_codes_user_id ON login_codes (user_id)")
def downgrade():
op.execute("DROP TABLE IF EXISTS login_codes")
op.execute("""
CREATE TABLE IF NOT EXISTS login_links (
id SERIAL PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
token_hash VARCHAR(64) NOT NULL UNIQUE,
expires_at TIMESTAMP NOT NULL,
consumed_at TIMESTAMP,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)""")
op.execute("CREATE INDEX IF NOT EXISTS ix_login_links_token_hash ON login_links (token_hash)")
op.execute("CREATE INDEX IF NOT EXISTS ix_login_links_user_id ON login_links (user_id)")

View file

@ -14,7 +14,7 @@ from app.routers import auth, documents, quizzes, attempts, admin, tts, nextclou
from app.routers import access
from app.routers import feedback
from app.routers import study_tools, uploads, articles, share, collections, study_plans, media, search, ai_mode, drafts, public
from app.routers import login_link
from app.routers import login_code
from app.utils.auth import get_password_hash
@ -610,7 +610,7 @@ app.add_middleware(RequestLoggingMiddleware)
app.include_router(uploads.router)
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(login_link.router, prefix="/api/auth", tags=["auth"])
app.include_router(login_code.router, prefix="/api/auth", tags=["auth"])
# Counts the landing page states about itself. No auth: a stranger reads it.
app.include_router(public.router, prefix="/api/public", tags=["public"])
app.include_router(articles.router, prefix="/api/articles", tags=["articles"])

View file

@ -42,4 +42,4 @@ __all__ = [
from app.models.feedback import QuestionFeedback # noqa: F401
from app.models.invite import InviteCode # noqa: F401
from app.models.draft_question import DraftBatch, DraftQuestion # noqa: F401
from app.models.login_link import LoginLink # noqa: F401
from app.models.login_code import LoginCode # noqa: F401

View file

@ -0,0 +1,37 @@
from datetime import datetime
from sqlalchemy import Column, DateTime, ForeignKey, Integer, String
from app.database import Base
class LoginCode(Base):
"""One issued sign-in code.
Only the fingerprint of the code is kept. A row here is a session waiting
to happen anybody who could read the table could sign in as its owner
so it holds something to compare against rather than something to type in.
The failed guesses are counted in this row and not in Redis. The rate
limiter fails open when the cache is down, which is the right call for
anything it normally protects; it is the wrong call for the only thing
standing between a patient stranger and six characters.
"""
__tablename__ = "login_codes"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
#: SHA-256 of the normalised code, hex. Not bcrypt: the guess budget is
#: what makes a short code safe, not the cost of testing one, and this is
#: compared on a path a person is waiting on. Not indexed and not unique
#: either — a code is only ever looked up against the one account it was
#: issued for, and six characters do collide.
code_hash = Column(String(64), nullable=False)
attempts = Column(Integer, nullable=False, default=0)
expires_at = Column(DateTime, nullable=False)
#: Set when the code is spent, when its guesses run out, and when a later
#: request retires it. All three mean the same thing to every reader here:
#: no longer usable.
consumed_at = Column(DateTime, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)

View file

@ -1,29 +0,0 @@
from datetime import datetime
from sqlalchemy import Column, DateTime, ForeignKey, Integer, String
from app.database import Base
class LoginLink(Base):
"""One issued sign-in link.
Only the fingerprint of the token is kept. Unlike a verification or reset
token, this one is a session in waiting anybody who can read the table
could sign in as its owner so the row holds something that can be
compared against but not replayed.
"""
__tablename__ = "login_links"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
#: SHA-256 of the token, hex. Not bcrypt: the token is 256 bits of
#: randomness rather than something a person chose, so there is nothing for
#: a slow hash to buy — only the lookup would get slower.
token_hash = Column(String(64), unique=True, nullable=False, index=True)
expires_at = Column(DateTime, nullable=False)
#: Set the moment the link is spent, and also when a later request retires
#: it. Both mean the same thing to everyone who reads this: not usable.
consumed_at = Column(DateTime, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)

View file

@ -0,0 +1,168 @@
"""Ask for a sign-in code, and type one in.
Kept out of the auth router because these two routes answer to a rule the rest
of that file does not: whatever the state of the address given, the reply is
the same reply, and takes the same time to arrive. That is easy to break by
adding an early return next to routes that quite reasonably have several.
"""
import asyncio
import logging
import time
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
from sqlalchemy.orm import Session
from app.database import get_db
from app.models.email_verification import EmailVerification
from app.models.user import User
from app.schemas.auth import LoginCodeRequest, LoginCodeVerify, Token
from app.services import email_service, login_codes, site_settings
from app.utils.auth import check_rate_limit, create_access_token
logger = logging.getLogger(__name__)
router = APIRouter()
#: The only thing the request endpoint ever says. Not "we sent you a code" and
#: not "no such account": either would turn the form into a way to ask the site
#: whether a given person is a member of it.
SENT = "If that address has an account, a sign-in code is on its way."
#: And the only thing the verify endpoint says when it will not let somebody
#: in. Wrong code, expired code, code out of guesses, address with no account —
#: one answer, because the alternative is a way to sort guesses into warm and
#: cold.
REFUSED = "That code is not right, or it has expired. Request a new one."
#: Wording and status are identical whichever way it went — but the work is
#: not, and a stopwatch reads the difference. Every reply from both routes is
#: held to this floor, comfortably above what either costs. Sending the mail is
#: a background task and runs after the response, so it never shows up in the
#: timing at all.
MIN_SECONDS = 0.4
#: Three per quarter hour, per address. This is the limit that matters when
#: issuing: the address is who the mail lands on, and the one thing an attacker
#: cannot vary while still reaching the person they mean to reach.
PER_ADDRESS_CALLS, PER_ADDRESS_WINDOW = 3, 15 * 60
#: Twenty per hour, per source address. Deliberately the looser of the two: a
#: hospital or university puts a whole department behind one address, so a
#: tight limit here locks out real people and barely inconveniences anyone
#: renting a hundred addresses. It protects the site's sending reputation from
#: one machine working through a word list; the per-address limit is what
#: protects any individual mailbox.
PER_IP_CALLS, PER_IP_WINDOW = 20, 60 * 60
#: Ten guesses a quarter hour against one address, on top of the five any one
#: code will answer. The code's own budget is the harder bound and survives the
#: cache being down; this one bounds what somebody gets by burning through
#: codes — three issued in a window, five guesses each, would otherwise be
#: fifteen. It is keyed by address rather than by network because a code only
#: means anything against the address it was issued for: limiting the network
#: would lock out a shared one while an attacker simply moved to the next
#: address.
PER_ADDRESS_TRIES, PER_ADDRESS_TRIES_WINDOW = 10, 15 * 60
def _sso_only() -> bool:
return site_settings.get_flag("sso_only")
async def _hold(started: float):
elapsed = time.perf_counter() - started
if elapsed < MIN_SECONDS:
await asyncio.sleep(MIN_SECONDS - elapsed)
@router.post("/login-code")
async def request_login_code(
data: LoginCodeRequest,
background_tasks: BackgroundTasks,
request: Request = None,
db: Session = Depends(get_db),
):
"""Send a sign-in code to an address, if that address has an account."""
started = time.perf_counter()
if _sso_only():
raise HTTPException(status_code=403, detail="Email sign-in is disabled. Please use SSO.")
address = data.email.lower().strip()
# Both limits are checked before the account is looked up, so that being
# throttled cannot itself become the tell about whether an account exists.
# The address is keyed by its fingerprint: Redis is a cache with a rather
# more relaxed life than the database, and it has no business holding a
# readable list of who has an account here.
check_rate_limit(
key=f"login_code:addr:{login_codes.fingerprint(address)}",
max_calls=PER_ADDRESS_CALLS,
window_seconds=PER_ADDRESS_WINDOW,
detail="Too many sign-in codes requested for that address. Please wait a few minutes.",
)
client_ip = (request.client.host if request and request.client else "unknown")
check_rate_limit(
key=f"login_code:ip:{client_ip}",
max_calls=PER_IP_CALLS,
window_seconds=PER_IP_WINDOW,
detail="Too many sign-in codes requested from this network. Please wait a while.",
)
user = db.query(User).filter(User.email == address).first()
if user is not None:
code = login_codes.issue(db, user)
background_tasks.add_task(email_service.send_login_code_email,
user.email, user.name, login_codes.for_display(code))
await _hold(started)
return {"message": SENT}
@router.post("/login-code/verify", response_model=Token)
async def verify_login_code(data: LoginCodeVerify, db: Session = Depends(get_db)):
"""Spend a code and hand back a session.
The code is typed into the page that asked for it, so nothing has to travel
between devices, no credential ends up in a URL, and there is no history
entry to clear afterwards.
"""
started = time.perf_counter()
if _sso_only():
raise HTTPException(status_code=403, detail="Email sign-in is disabled. Please use SSO.")
address = data.email.lower().strip()
# Before the lookup, and counted for an address with no account exactly as
# for one with — otherwise the 429 says which is which.
check_rate_limit(
key=f"login_code:try:{login_codes.fingerprint(address)}",
max_calls=PER_ADDRESS_TRIES,
window_seconds=PER_ADDRESS_TRIES_WINDOW,
detail="Too many attempts for that address. Please wait a few minutes and request a new code.",
)
user = login_codes.verify(db, address, data.code)
if user is None:
await _hold(started)
raise HTTPException(status_code=400, detail=REFUSED)
# Checked after the code is spent, and deliberately: a refusal must not
# leave a usable credential behind.
#
# Reading a code out of the mailbox proves control of it, which is exactly
# what verification checks — and it is still not allowed to stand in for
# it. Somebody can register an address that is not theirs, choose a
# password and never verify; if the real owner's sign-in code marked that
# account verified, they would be signing in to an account the squatter
# already has the password to. The verification mail is in the same inbox.
verification = db.query(EmailVerification).filter(EmailVerification.user_id == user.id).first()
if verification and verification.verified_at is None:
await _hold(started)
raise HTTPException(
status_code=403,
detail="Email not verified. Please check your inbox and verify your email before signing in.",
)
logger.info("Sign-in code spent by user %s", user.id)
await _hold(started)
return Token(access_token=create_access_token(data={"sub": user.email}))

View file

@ -1,140 +0,0 @@
"""Ask for a sign-in link, and spend one.
Kept out of the auth router because these two routes answer to a rule the rest
of that file does not: whatever the state of the address given, the reply is
the same reply, and takes the same time to arrive. That is easy to break by
adding an early return next to routes that quite reasonably have several.
"""
import asyncio
import logging
import time
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
from sqlalchemy.orm import Session
from app.database import get_db
from app.models.email_verification import EmailVerification
from app.models.user import User
from app.schemas.auth import LoginLinkRequest, LoginLinkConsume, Token
from app.services import email_service, login_links, site_settings
from app.utils.auth import check_rate_limit, create_access_token
logger = logging.getLogger(__name__)
router = APIRouter()
#: The only thing this endpoint ever says. Not "we sent you a link" and not
#: "no such account": either would turn the form into a way to ask the site
#: whether a given person is a member of it.
SENT = "If that address has an account, a sign-in link is on its way."
#: Wording and status are identical for an address that exists and one that
#: does not — but the work is not, and a stopwatch reads the difference. Every
#: reply is held to this floor, which is comfortably above the few milliseconds
#: that issuing a link costs. Sending the mail is a background task and runs
#: after the response, so it never shows up in the timing at all.
MIN_SECONDS = 0.4
#: Three per quarter hour, per address. This is the limit that matters: the
#: address is who the mail lands on, and the one thing an attacker cannot vary
#: while still hitting the person they mean to hit. Three is enough for
#: somebody who fumbles the first attempt and does not trust the second.
PER_ADDRESS_CALLS, PER_ADDRESS_WINDOW = 3, 15 * 60
#: Twenty per hour, per source address. Deliberately the looser of the two: a
#: hospital or university puts a whole department behind one address, so a
#: tight limit here locks out real people, and it barely inconveniences anyone
#: renting a hundred addresses. It is here to protect the site's own sending
#: reputation from one machine working through a word list, not to protect any
#: individual mailbox — the per-address limit does that.
PER_IP_CALLS, PER_IP_WINDOW = 20, 60 * 60
def _sso_only() -> bool:
return site_settings.get_flag("sso_only")
@router.post("/login-link")
async def request_login_link(
data: LoginLinkRequest,
background_tasks: BackgroundTasks,
request: Request = None,
db: Session = Depends(get_db),
):
"""Send a sign-in link to an address, if that address has an account."""
started = time.perf_counter()
if _sso_only():
raise HTTPException(status_code=403, detail="Email sign-in is disabled. Please use SSO.")
address = data.email.lower().strip()
# Both limits are checked before the account is looked up, so that being
# throttled cannot itself become the tell about whether an account exists.
# The address is keyed by its fingerprint: Redis is a cache with a rather
# more relaxed life than the database, and it has no business holding a
# readable list of who has an account here.
check_rate_limit(
key=f"login_link:addr:{login_links.fingerprint(address)}",
max_calls=PER_ADDRESS_CALLS,
window_seconds=PER_ADDRESS_WINDOW,
detail="Too many sign-in links requested for that address. Please wait a few minutes.",
)
client_ip = (request.client.host if request and request.client else "unknown")
check_rate_limit(
key=f"login_link:ip:{client_ip}",
max_calls=PER_IP_CALLS,
window_seconds=PER_IP_WINDOW,
detail="Too many sign-in links requested from this network. Please wait a while.",
)
user = db.query(User).filter(User.email == address).first()
if user is not None:
token = login_links.issue(db, user)
background_tasks.add_task(email_service.send_login_link_email, user.email, user.name, token)
elapsed = time.perf_counter() - started
if elapsed < MIN_SECONDS:
await asyncio.sleep(MIN_SECONDS - elapsed)
return {"message": SENT}
@router.post("/login-link/consume", response_model=Token)
def consume_login_link(data: LoginLinkConsume, db: Session = Depends(get_db)):
"""Spend a link and hand back a session.
A POST rather than the link itself pointing here. Mail gateways, security
scanners and preview panes fetch every URL in a message with GET, and a
single-use token spent by a scanner is a link that is already dead when its
owner clicks it. The mail points at a page in the app, which posts the
token from JavaScript that no scanner runs.
The link is not tied to the browser that asked for it. Binding it a nonce
in a cookie, checked here would break the ordinary case, which is asking
on a laptop and reading mail on a phone, and it defends against almost
nothing: somebody who can read the mailbox can simply ask for a link of
their own. What is left is a link forwarded or leaked to a third party, and
that is what the fifteen minutes and the one use are for.
"""
user = login_links.consume(db, data.token)
if user is None:
raise HTTPException(
status_code=400,
detail="That sign-in link is no longer valid. If you asked more than once, "
"open the most recent email, or request a new link.",
)
# Following the link proves control of the mailbox, which is exactly what
# verification checks — and it is still not allowed to stand in for it.
# Somebody can register an address that is not theirs, choose a password
# and never verify; if the real owner's sign-in link marked that account
# verified, they would be signing in to an account the squatter already has
# the password to. The verification mail is in the same inbox.
verification = db.query(EmailVerification).filter(EmailVerification.user_id == user.id).first()
if verification and verification.verified_at is None:
raise HTTPException(
status_code=403,
detail="Email not verified. Please check your inbox and verify your email before signing in.",
)
logger.info("Sign-in link spent by user %s", user.id)
return Token(access_token=create_access_token(data={"sub": user.email}))

View file

@ -54,9 +54,12 @@ class ResetPasswordRequest(BaseModel):
new_password: str
class LoginLinkRequest(BaseModel):
class LoginCodeRequest(BaseModel):
email: EmailStr
class LoginLinkConsume(BaseModel):
token: str
class LoginCodeVerify(BaseModel):
email: EmailStr
#: Typed by a person, so it arrives in whatever shape they typed it and is
#: normalised before it is compared.
code: str

View file

@ -47,6 +47,12 @@ def _render(md: str) -> str:
if m:
label, url = m.group(1), m.group(2)
out.append(f'<p style="margin:28px 0;"><a href="{url}" style="display:inline-block;background:#09090b;color:#fafafa;font-size:14px;font-weight:500;padding:10px 24px;border-radius:6px;text-decoration:none;letter-spacing:0.01em;">{label} →</a></p>')
elif s.startswith("[code:"):
# A thing to read off the screen and type, so it is set large,
# spaced and monospaced rather than left to look like a heading.
m = re.match(r'\[code:(.+?)\]', s)
if m:
out.append(f'<p style="margin:24px 0;"><span style="display:inline-block;background:#f4f4f5;border:1px solid #e4e4e7;border-radius:8px;padding:14px 20px 14px 24px;font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:28px;font-weight:600;letter-spacing:0.22em;color:#09090b;">{m.group(1)}</span></p>')
elif s.startswith("[link:"):
m = re.match(r'\[link:(.+?)\]\((.+?)\)', s)
if m:
@ -150,23 +156,20 @@ We received a request to reset your password. Click below to choose a new one.
await _send(to_email, subject, _wrap(subject, md))
async def send_login_link_email(to_email: str, name: str, token: str):
url = f"{settings.APP_URL}/signin-link?token={token}"
subject = "Your PedsHub sign-in link"
md = f"""# Sign in to PedsHub
async def send_login_code_email(to_email: str, name: str, code: str):
#: The code arrives here already grouped for reading; any shape it is typed
#: back in is normalised before it is compared.
subject = "Your PedsHub sign-in code"
md = f"""# Your sign-in code
Hi **{name}**,
Here is the sign-in link you asked for. It signs you in on whichever device you open it on.
Enter this code on the sign-in page you just came from.
[button:Sign In]({url})
[code:{code}]
> Expires in **15 minutes** · Works once. If you asked more than once, only the newest link works. If you didn't ask for this, ignore this email — nobody can get in without it.
> Expires in **15 minutes** · Works once. If you asked more than once, only the newest code works.
---
[link:Or copy this link]({url})
Nobody from PedsHub will ever ask you to read this code out, on the phone or anywhere else. If somebody has, they are not us. If you didn't ask for this code, ignore this email — nobody can get in without it.
"""
await _send(to_email, subject, _wrap(subject, md))

View file

@ -0,0 +1,159 @@
"""Signing in with a code sent to your own mailbox.
Six characters a person reads off an email and types into the page they asked
from. Everything here follows from the code being short: it can be guessed, so
what makes it safe is not its length on its own but the arithmetic between its
length, how many guesses it will ever answer, and how long it lives.
"""
import hashlib
import hmac
import re
import secrets
from datetime import datetime, timedelta
from sqlalchemy.orm import Session
from app.models.login_code import LoginCode
from app.models.user import User
#: The house alphabet, and the reason it is the house alphabet: no O/0 and no
#: I/1, because these are read off a screen and typed back. Borrowed rather
#: than copied — there should be one answer to "which characters may a person
#: be asked to retype", not one per feature.
from app.services.invites import ALPHABET
#: Six characters of a 32-letter alphabet: 2^30 codes, where the six digits
#: everyone is used to would be 2^20. The length is only half the argument
#: though, and the smaller half. A code answers at most MAX_ATTEMPTS guesses
#: before it is retired, only one is ever live per person, and the address may
#: only be tried so often — so an attacker gets tens of guesses an hour against
#: a billion, while sending the victim a mail for every code they burn through.
#: Eight characters would buy a thousandfold against an attack the guess budget
#: has already ended, and cost every user two more characters to type.
LENGTH = 6
#: Five. Enough that a mistyped character, a stale code from a first attempt
#: and one more fumble do not lock somebody out of their own account; few
#: enough that the whole budget is nothing against 2^30. Counted in the row
#: rather than the cache — see the model.
MAX_ATTEMPTS = 5
#: Fifteen minutes, unchanged from when this was a link, and for a reason that
#: survived the change: a first mail between a sender and a recipient who have
#: never corresponded is routinely greylisted five to ten minutes, and a code
#: that has expired before it arrives is not a sign-in method. Shortening it
#: would not slow an attacker down — only one code is ever live per person and
#: it answers five guesses however long it sits there — so the clock is a
#: deliverability question, and the one thing it does buy is a shorter window
#: for a mailbox somebody else is reading. Fifteen is where those meet.
CODE_TTL = timedelta(minutes=15)
#: Rows are dead weight once well past expiry and nothing reads them. Kept a
#: day rather than deleted on the spot, so "it did not work this morning" can
#: still be answered from the table.
KEEP_SPENT_FOR = timedelta(days=1)
def normalise(code: str) -> str:
"""What the person meant, from what they typed.
Case, spaces and dashes are not part of the code the mail shows it in two
groups of three, and somebody will type the space. Anything that is not a
letter or a digit goes; the rest is upper-cased. There is no look-alike
folding to do, which is the point of that alphabet: O and 0 are not two
characters one might mistake for each other here, they are both simply not
characters a code can contain.
"""
return re.sub(r"[^A-Z0-9]", "", (code or "").upper())
def fingerprint(value: str) -> str:
return hashlib.sha256(value.strip().encode("utf-8")).hexdigest()
def generate_code() -> str:
return "".join(secrets.choice(ALPHABET) for _ in range(LENGTH))
def for_display(code: str) -> str:
"""Grouped for reading aloud in one's head. Typed back in any shape."""
return f"{code[:3]} {code[3:]}" if len(code) == 6 else code
def issue(db: Session, user: User) -> str:
"""Retire this user's outstanding codes, mint a new one, return it.
Superseding rather than accumulating, and it matters more here than it did
for a link: with several codes live at once, a guess is tested against all
of them, and a six-character code gets easier to hit with every one left
lying around. Exactly one is live per person at any moment.
"""
now = datetime.utcnow()
db.query(LoginCode).filter(
LoginCode.user_id == user.id,
LoginCode.consumed_at.is_(None),
).update({"consumed_at": now}, synchronize_session=False)
db.query(LoginCode).filter(
LoginCode.user_id == user.id,
LoginCode.expires_at < now - KEEP_SPENT_FOR,
).delete(synchronize_session=False)
code = generate_code()
db.add(LoginCode(user_id=user.id, code_hash=fingerprint(code), expires_at=now + CODE_TTL))
db.commit()
return code
def verify(db: Session, email: str, code: str) -> User | None:
"""Spend a code and return whose it is, or None if it is no good.
Scoped to the address, and that is not merely convenient. A code looked up
on its own would be tested against every code live on the site at once, so
each guess would be as many guesses as there are people signing in the
short code's one real weakness, closed by knowing whose code it is meant to
be before comparing.
One answer for every failure: wrong code, expired code, code whose guesses
have run out, address with no account. Telling them apart would say which
of those was true, and none of them leaves the person anything to do but
ask for a new code.
"""
user = db.query(User).filter(User.email == email.lower().strip()).first()
if user is None:
return None
row = db.query(LoginCode).filter(
LoginCode.user_id == user.id,
LoginCode.consumed_at.is_(None),
).order_by(LoginCode.id.desc()).first()
if row is None or row.expires_at <= datetime.utcnow():
return None
if not hmac.compare_digest(row.code_hash, fingerprint(normalise(code))):
# A wrong guess costs one of five, and the fifth ends the code rather
# than leaving it to be guessed at a slower rate: whoever is typing has
# either lost the mail or is not its owner, and both are one click from
# a new code. The count is a SQL expression so that two guesses racing
# each other cost two.
exhausted = row.attempts + 1 >= MAX_ATTEMPTS
spend = {"attempts": LoginCode.attempts + 1}
if exhausted:
spend["consumed_at"] = datetime.utcnow()
db.query(LoginCode).filter(
LoginCode.id == row.id,
LoginCode.consumed_at.is_(None),
).update(spend, synchronize_session=False)
db.commit()
return None
# The claim is the UPDATE, not the read above: two requests carrying the
# same code both pass the comparison, and only the one whose UPDATE matched
# a row still unspent gets to sign in.
claimed = db.query(LoginCode).filter(
LoginCode.id == row.id,
LoginCode.consumed_at.is_(None),
).update({"consumed_at": datetime.utcnow()}, synchronize_session=False)
db.commit()
if not claimed:
return None
return user

View file

@ -1,90 +0,0 @@
"""Signing in by following a link sent to your own mailbox.
The link is a credential that sits in an inbox, so everything here is about
keeping the window in which it is worth stealing as narrow as it can be while
the flow still works: a short life, one use, and a newer request retiring
whatever came before it.
"""
import hashlib
import secrets
from datetime import datetime, timedelta
from sqlalchemy.orm import Session
from app.models.login_link import LoginLink
from app.models.user import User
#: Fifteen minutes. Mail is normally delivered in seconds and whoever asked for
#: the link is at the keyboard now, so five would nearly always do — except
#: that greylisting defers a first delivery between a sender and a recipient
#: who have never spoken, commonly by five to ten minutes, and that is exactly
#: the case here for anybody signing in for the first time. Fifteen covers the
#: deferral and very little else.
#:
#: Deliberately not the hour the password reset link gets: spending a reset
#: link changes the password, which the owner notices the next time they sign
#: in. Spending this one leaves no trace at all in front of the owner, so it
#: does not get the same rope.
LINK_TTL = timedelta(minutes=15)
#: Rows are dead weight once they are well past their expiry, and nothing reads
#: them afterwards. Kept a day rather than deleted on the spot so that a
#: support question about "the link did not work this morning" can still be
#: answered from the table.
KEEP_SPENT_FOR = timedelta(days=1)
def fingerprint(value: str) -> str:
return hashlib.sha256(value.strip().encode("utf-8")).hexdigest()
def issue(db: Session, user: User) -> str:
"""Retire this user's outstanding links, mint a new one, return the token.
Superseding rather than accumulating: with three live at once which is
all the rate limit allows "single use" stops meaning "one way in at a
time", and a mailbox quietly collects credentials. The cost is that somebody who asks twice and then opens the
first mail is refused which is why the failure copy says to use the most
recent email rather than only that the link is dead.
"""
now = datetime.utcnow()
db.query(LoginLink).filter(
LoginLink.user_id == user.id,
LoginLink.consumed_at.is_(None),
).update({"consumed_at": now}, synchronize_session=False)
db.query(LoginLink).filter(
LoginLink.user_id == user.id,
LoginLink.expires_at < now - KEEP_SPENT_FOR,
).delete(synchronize_session=False)
token = secrets.token_urlsafe(32)
db.add(LoginLink(user_id=user.id, token_hash=fingerprint(token), expires_at=now + LINK_TTL))
db.commit()
return token
def consume(db: Session, token: str) -> User | None:
"""Spend a link and return whose it is, or None if it is no good.
One caller, one answer: an expired link, an already-spent link and a token
that was never issued are all None. Telling them apart would tell whoever
is guessing which of their guesses was once real, and none of the three
leaves the person holding it anything to do but ask for another.
"""
row = db.query(LoginLink).filter(LoginLink.token_hash == fingerprint(token)).first()
if row is None or row.expires_at <= datetime.utcnow():
return None
# The claim is the UPDATE, not the read above: two requests carrying the
# same token — a double click, or a mail client racing the person — both
# pass the read, and only the one whose UPDATE matched a row still unspent
# gets to sign in.
claimed = db.query(LoginLink).filter(
LoginLink.id == row.id,
LoginLink.consumed_at.is_(None),
).update({"consumed_at": datetime.utcnow()}, synchronize_session=False)
db.commit()
if not claimed:
return None
return db.query(User).filter(User.id == row.user_id).first()

View file

@ -0,0 +1,290 @@
"""Signing in with a code sent by email.
The properties worth pinning are the ones that are invisible when they break: a
stranger's address gets the same answer as a member's, in the same time; a code
works once and not after that; a wrong guess costs one of five and the fifth
ends the code; and the limiter counts requests for an address that has no
account exactly as it counts the others, or the 429 becomes the answer the
identical message was there to withhold.
Disposable SQLite, a Mock for Redis, and nothing leaves the process.
"""
import hashlib
import sys
import time
import unittest
from datetime import datetime, timedelta
from types import ModuleType
from unittest.mock import AsyncMock, patch
import test_quiz_builder as fixtures
from app.models.email_verification import EmailVerification
from app.models.login_code import LoginCode
from app.routers import login_code
from app.services import email_service, invites, login_codes
class MemoryRedis:
def __init__(self): self.values = {}
def get(self, key): return self.values.get(key)
def set(self, key, value): self.values[key] = value
def incr(self, key):
self.values[key] = int(self.values.get(key, 0)) + 1
return self.values[key]
def expire(self, *args): return True
class LoginCodeTests(unittest.TestCase):
def setUp(self):
self.bank = fixtures.BuilderTests()
self.bank.setUp()
self.bank.owner.email = "owner@example.com"
# The peer too: the address goes through EmailStr on the way in, which
# refuses the reserved .test domain the fixtures otherwise use.
self.bank.peer.email = "peer@example.com"
self.bank.db.commit()
self.client = self.bank.client
self.client.app.include_router(login_code.router, prefix="/auth")
self.redis = MemoryRedis()
module = ModuleType("redis")
module.from_url = lambda *args, **kwargs: self.redis
modules = patch.dict(sys.modules, {"redis": module})
modules.start()
self.addCleanup(modules.stop)
# Every reply is paced to a floor; the pacing itself has its own test,
# and the rest of the suite has no reason to wait for it.
floor = patch.object(login_code, "MIN_SECONDS", 0.01)
floor.start()
self.addCleanup(floor.stop)
sent = patch.object(email_service, "send_login_code_email", new_callable=AsyncMock)
self.sent = sent.start()
self.addCleanup(sent.stop)
def tearDown(self):
self.bank.tearDown()
def ask(self, email="owner@example.com"):
return self.client.post("/auth/login-code", json={"email": email})
def code_for(self, email="owner@example.com"):
"""The code as the mail shows it — spaces and all."""
self.sent.reset_mock()
self.assertEqual(self.ask(email).status_code, 200)
self.sent.assert_awaited_once()
return self.sent.await_args.args[2]
def enter(self, code, email="owner@example.com"):
return self.client.post("/auth/login-code/verify", json={"email": email, "code": code})
def wrong(self, email="owner@example.com"):
return self.enter("ZZZZZZ", email)
# ── the answer gives nothing away ──────────────────────────────────
def test_a_stranger_gets_the_member_s_answer_word_for_word(self):
known = self.ask()
unknown = self.ask("nobody@example.com")
self.assertEqual(known.status_code, unknown.status_code)
self.assertEqual(known.json(), unknown.json())
self.assertEqual(known.json(), {"message": login_code.SENT})
# And the only difference behind it is one that never reaches the caller.
self.assertEqual(self.bank.db.query(LoginCode).count(), 1)
def test_both_answers_are_held_to_the_same_floor(self):
# The real constant, because the point of the test is the value of it.
with patch.object(login_code, "MIN_SECONDS", 0.3):
times = []
for address in ("owner@example.com", "nobody@example.com"):
started = time.perf_counter()
self.ask(address)
times.append(time.perf_counter() - started)
for elapsed in times:
self.assertGreaterEqual(elapsed, 0.3)
self.assertLess(abs(times[0] - times[1]), 0.15)
def test_a_wrong_code_reads_the_same_as_an_address_with_no_account(self):
self.code_for()
mistyped = self.wrong()
stranger = self.enter("ZZZZZZ", "nobody@example.com")
self.assertEqual(mistyped.status_code, stranger.status_code)
self.assertEqual(mistyped.json(), stranger.json())
self.assertEqual(mistyped.json()["detail"], login_code.REFUSED)
def test_the_limiter_counts_a_stranger_s_address_too(self):
# Otherwise the 429 arrives only for addresses that exist, and the
# identical message above has been undone by the thing protecting it.
for _ in range(login_code.PER_ADDRESS_CALLS):
self.assertEqual(self.ask("nobody@example.com").status_code, 200)
self.assertEqual(self.ask("nobody@example.com").status_code, 429)
self.assertEqual(self.ask().status_code, 200)
def test_the_address_limit_bites_on_the_fourth_request(self):
for _ in range(login_code.PER_ADDRESS_CALLS):
self.assertEqual(self.ask().status_code, 200)
refused = self.ask()
self.assertEqual(refused.status_code, 429)
self.assertIn("Too many sign-in codes", refused.json()["detail"])
# Nothing was issued on the refused attempt.
self.assertEqual(self.bank.db.query(LoginCode).count(), login_code.PER_ADDRESS_CALLS)
def test_the_network_limit_is_the_looser_of_the_two(self):
self.assertGreater(login_code.PER_IP_CALLS, login_code.PER_ADDRESS_CALLS)
# A different address each time, so the tighter limit cannot be what
# stops it.
for n in range(login_code.PER_IP_CALLS):
self.assertEqual(self.ask(f"person{n}@example.com").status_code, 200)
refused = self.ask(f"person{login_code.PER_IP_CALLS}@example.com")
self.assertEqual(refused.status_code, 429)
self.assertIn("from this network", refused.json()["detail"])
# ── the code itself ────────────────────────────────────────────────
def test_a_code_signs_you_in_once_and_never_again(self):
code = self.code_for()
first = self.enter(code)
self.assertEqual(first.status_code, 200, first.text)
self.assertTrue(first.json()["access_token"])
second = self.enter(code)
self.assertEqual(second.status_code, 400)
self.assertEqual(second.json()["detail"], login_code.REFUSED)
def test_six_characters_of_the_house_alphabet(self):
code = login_codes.normalise(self.code_for())
self.assertEqual(len(code), login_codes.LENGTH)
self.assertEqual(login_codes.LENGTH, 6)
self.assertTrue(set(code) <= set(invites.ALPHABET))
# The characters that get misread are not in it to be misread.
self.assertFalse(set("O0I1") & set(invites.ALPHABET))
def test_case_spaces_and_dashes_do_not_decide_who_gets_in(self):
code = login_codes.normalise(self.code_for())
typed = f" {code[:3].lower()}-{code[3:]} "
self.assertEqual(self.enter(typed).status_code, 200)
def test_the_code_as_the_mail_shows_it_types_straight_back(self):
shown = self.code_for()
self.assertIn(" ", shown)
self.assertEqual(self.enter(shown).status_code, 200)
def test_an_expired_code_is_refused(self):
code = self.code_for()
row = self.bank.db.query(LoginCode).one()
row.expires_at = datetime.utcnow() - timedelta(seconds=1)
self.bank.db.commit()
self.assertEqual(self.enter(code).status_code, 400)
# Refused without being marked spent: it was never used, and the row
# says so.
self.bank.db.refresh(row)
self.assertIsNone(row.consumed_at)
def test_fifteen_minutes_and_not_an_hour(self):
self.code_for()
row = self.bank.db.query(LoginCode).one()
self.assertAlmostEqual((row.expires_at - row.created_at).total_seconds(), 15 * 60, delta=5)
def test_a_second_request_retires_the_first_code(self):
stale = self.code_for()
fresh = self.code_for()
self.assertEqual(self.enter(stale).status_code, 400)
self.assertEqual(self.enter(fresh).status_code, 200)
def test_the_table_holds_a_fingerprint_rather_than_the_code(self):
code = login_codes.normalise(self.code_for())
row = self.bank.db.query(LoginCode).one()
self.assertNotIn(code, str(row.code_hash))
self.assertEqual(row.code_hash, hashlib.sha256(code.encode()).hexdigest())
def test_a_code_is_no_good_against_somebody_else_s_address(self):
# A short code looked up on its own would be tried against every code
# live on the site at once. It is only ever tried against one account.
code = self.code_for()
self.assertEqual(self.enter(code, "peer@example.com").status_code, 400)
self.assertEqual(self.enter(code).status_code, 200)
# ── guessing costs something ───────────────────────────────────────
def test_a_wrong_code_counts_against_the_limit(self):
self.code_for()
for expected in range(1, 3):
self.assertEqual(self.wrong().status_code, 400)
self.assertEqual(self.bank.db.query(LoginCode).one().attempts, expected)
def test_a_code_out_of_guesses_is_retired_rather_than_left_guessable(self):
code = self.code_for()
for _ in range(login_codes.MAX_ATTEMPTS):
self.assertEqual(self.wrong().status_code, 400)
row = self.bank.db.query(LoginCode).one()
self.assertIsNotNone(row.consumed_at)
# And now not even the right code opens it — the way back is a new one.
self.assertEqual(self.enter(code).status_code, 400)
self.assertEqual(self.enter(self.code_for()).status_code, 200)
def test_the_attempts_limit_keeps_counting_after_the_code_is_dead(self):
# The code's own five run out first. The address limit is what stops
# somebody working through codes to buy five more guesses at a time.
self.code_for()
for _ in range(login_code.PER_ADDRESS_TRIES):
self.assertEqual(self.wrong().status_code, 400)
refused = self.wrong()
self.assertEqual(refused.status_code, 429)
self.assertIn("Too many attempts", refused.json()["detail"])
def test_guessing_a_stranger_s_address_is_counted_the_same_way(self):
for _ in range(login_code.PER_ADDRESS_TRIES):
self.assertEqual(self.wrong("nobody@example.com").status_code, 400)
self.assertEqual(self.wrong("nobody@example.com").status_code, 429)
# ── the rules the password form already keeps ──────────────────────
def test_an_unverified_account_is_turned_away_as_it_is_at_the_password_form(self):
self.bank.db.add(EmailVerification(user_id=self.bank.owner.id, token="verification-token",
expires_at=datetime.utcnow() + timedelta(hours=1), verified_at=None))
self.bank.db.commit()
code = self.code_for()
refused = self.enter(code)
self.assertEqual(refused.status_code, 403)
self.assertIn("Email not verified", refused.json()["detail"])
# Refused, and the code spent all the same: turning somebody away must
# not leave a working credential behind them.
self.assertIsNotNone(self.bank.db.query(LoginCode).one().consumed_at)
def test_sso_only_closes_this_door_as_well_as_the_password_one(self):
code = self.code_for()
self.redis.values["settings:sso_only"] = "true"
refused = self.ask()
self.assertEqual(refused.status_code, 403)
self.assertIn("Please use SSO", refused.json()["detail"])
self.assertEqual(self.enter(code).status_code, 403)
self.assertEqual(self.bank.db.query(LoginCode).count(), 1)
class ServiceTests(unittest.TestCase):
def setUp(self):
self.bank = fixtures.BuilderTests()
self.bank.setUp()
self.addCleanup(self.bank.tearDown)
def test_only_one_of_two_claims_on_one_code_succeeds(self):
code = login_codes.issue(self.bank.db, self.bank.owner)
self.assertIsNotNone(login_codes.verify(self.bank.db, self.bank.owner.email, code))
self.assertIsNone(login_codes.verify(self.bank.db, self.bank.owner.email, code))
def test_rows_long_past_their_expiry_are_swept_on_the_next_request(self):
login_codes.issue(self.bank.db, self.bank.owner)
stale = self.bank.db.query(LoginCode).one()
stale.expires_at = datetime.utcnow() - timedelta(days=3)
self.bank.db.commit()
login_codes.issue(self.bank.db, self.bank.owner)
self.assertEqual(self.bank.db.query(LoginCode).count(), 1)
def test_normalising_keeps_only_what_the_code_is_made_of(self):
self.assertEqual(login_codes.normalise(" ab3-c d9 "), "AB3CD9")
self.assertEqual(login_codes.normalise("abc\u2013def"), "ABCDEF")
self.assertEqual(login_codes.normalise(""), "")
if __name__ == "__main__":
unittest.main()

View file

@ -1,220 +0,0 @@
"""Signing in with a link sent by email.
The properties worth pinning are the ones that are invisible when they break: a
stranger's address gets the same answer as a member's, in the same time; a link
works once and not after that; and the limiter counts requests for an address
that has no account exactly as it counts the others, or the 429 becomes the
answer the identical message was there to withhold.
Disposable SQLite, a Mock for Redis, and nothing leaves the process.
"""
import hashlib
import sys
import time
import unittest
from datetime import datetime, timedelta
from types import ModuleType
from unittest.mock import AsyncMock, patch
import test_quiz_builder as fixtures
from app.models.email_verification import EmailVerification
from app.models.login_link import LoginLink
from app.routers import login_link
from app.services import email_service, login_links
class MemoryRedis:
def __init__(self): self.values = {}
def get(self, key): return self.values.get(key)
def set(self, key, value): self.values[key] = value
def incr(self, key):
self.values[key] = int(self.values.get(key, 0)) + 1
return self.values[key]
def expire(self, *args): return True
class LoginLinkTests(unittest.TestCase):
def setUp(self):
self.bank = fixtures.BuilderTests()
self.bank.setUp()
self.bank.owner.email = "owner@example.com"
self.bank.db.commit()
self.client = self.bank.client
self.client.app.include_router(login_link.router, prefix="/auth")
self.redis = MemoryRedis()
module = ModuleType("redis")
module.from_url = lambda *args, **kwargs: self.redis
modules = patch.dict(sys.modules, {"redis": module})
modules.start()
self.addCleanup(modules.stop)
# Every reply is paced to a floor; the pacing itself has its own test,
# and the rest of the suite has no reason to wait for it.
floor = patch.object(login_link, "MIN_SECONDS", 0.01)
floor.start()
self.addCleanup(floor.stop)
sent = patch.object(email_service, "send_login_link_email", new_callable=AsyncMock)
self.sent = sent.start()
self.addCleanup(sent.stop)
def tearDown(self):
self.bank.tearDown()
def ask(self, email="owner@example.com"):
return self.client.post("/auth/login-link", json={"email": email})
def token_for(self, email="owner@example.com"):
self.sent.reset_mock()
self.assertEqual(self.ask(email).status_code, 200)
self.sent.assert_awaited_once()
return self.sent.await_args.args[2]
def spend(self, token):
return self.client.post("/auth/login-link/consume", json={"token": token})
# ── the answer gives nothing away ──────────────────────────────────
def test_a_stranger_gets_the_member_s_answer_word_for_word(self):
known = self.ask()
unknown = self.ask("nobody@example.com")
self.assertEqual(known.status_code, unknown.status_code)
self.assertEqual(known.json(), unknown.json())
self.assertEqual(known.json(), {"message": login_link.SENT})
# And the only difference behind it is one that never reaches the caller.
self.assertEqual(self.bank.db.query(LoginLink).count(), 1)
def test_both_answers_are_held_to_the_same_floor(self):
# The real constant, because the point of the test is the value of it.
with patch.object(login_link, "MIN_SECONDS", 0.3):
times = []
for address in ("owner@example.com", "nobody@example.com"):
started = time.perf_counter()
self.ask(address)
times.append(time.perf_counter() - started)
for elapsed in times:
self.assertGreaterEqual(elapsed, 0.3)
self.assertLess(abs(times[0] - times[1]), 0.15)
def test_the_limiter_counts_a_stranger_s_address_too(self):
# Otherwise the 429 arrives only for addresses that exist, and the
# identical message above has been undone by the thing protecting it.
for _ in range(login_link.PER_ADDRESS_CALLS):
self.assertEqual(self.ask("nobody@example.com").status_code, 200)
self.assertEqual(self.ask("nobody@example.com").status_code, 429)
self.assertEqual(self.ask().status_code, 200)
def test_the_address_limit_bites_on_the_fourth_request(self):
for _ in range(login_link.PER_ADDRESS_CALLS):
self.assertEqual(self.ask().status_code, 200)
refused = self.ask()
self.assertEqual(refused.status_code, 429)
self.assertIn("Too many sign-in links", refused.json()["detail"])
# Nothing was issued on the refused attempt.
self.assertEqual(self.bank.db.query(LoginLink).count(), login_link.PER_ADDRESS_CALLS)
def test_the_network_limit_is_the_looser_of_the_two(self):
self.assertGreater(login_link.PER_IP_CALLS, login_link.PER_ADDRESS_CALLS)
# Ask from one address until the per-network limit is the one that stops
# it — a different address each time, so the tighter limit cannot.
for n in range(login_link.PER_IP_CALLS):
self.assertEqual(self.ask(f"person{n}@example.com").status_code, 200)
refused = self.ask(f"person{login_link.PER_IP_CALLS}@example.com")
self.assertEqual(refused.status_code, 429)
self.assertIn("from this network", refused.json()["detail"])
# ── the link itself ────────────────────────────────────────────────
def test_a_link_signs_you_in_once_and_never_again(self):
token = self.token_for()
first = self.spend(token)
self.assertEqual(first.status_code, 200, first.text)
self.assertTrue(first.json()["access_token"])
second = self.spend(token)
self.assertEqual(second.status_code, 400)
self.assertIn("no longer valid", second.json()["detail"])
def test_an_expired_link_is_refused(self):
token = self.token_for()
row = self.bank.db.query(LoginLink).one()
row.expires_at = datetime.utcnow() - timedelta(seconds=1)
self.bank.db.commit()
self.assertEqual(self.spend(token).status_code, 400)
# Refused without being marked spent: it was never used, and the row
# says so.
self.bank.db.refresh(row)
self.assertIsNone(row.consumed_at)
def test_fifteen_minutes_and_not_an_hour(self):
self.token_for()
row = self.bank.db.query(LoginLink).one()
self.assertAlmostEqual((row.expires_at - row.created_at).total_seconds(), 15 * 60, delta=5)
def test_a_second_request_retires_the_first_link(self):
stale = self.token_for()
fresh = self.token_for()
self.assertEqual(self.spend(stale).status_code, 400)
self.assertEqual(self.spend(fresh).status_code, 200)
def test_a_token_that_was_never_issued_reads_the_same_as_a_spent_one(self):
token = self.token_for()
self.spend(token)
invented = self.spend("not-a-token-anybody-issued")
spent = self.spend(token)
self.assertEqual(invented.status_code, spent.status_code)
self.assertEqual(invented.json(), spent.json())
def test_the_table_holds_a_fingerprint_rather_than_the_token(self):
token = self.token_for()
row = self.bank.db.query(LoginLink).one()
self.assertNotIn(token, str(row.token_hash))
self.assertEqual(row.token_hash, hashlib.sha256(token.encode()).hexdigest())
def test_the_mail_carries_the_address_the_name_and_the_token(self):
token = self.token_for()
self.assertEqual(self.sent.await_args.args[:2], ("owner@example.com", "Owner"))
self.assertTrue(token)
# ── the rules the password form already keeps ──────────────────────
def test_an_unverified_account_is_turned_away_as_it_is_at_the_password_form(self):
self.bank.db.add(EmailVerification(user_id=self.bank.owner.id, token="verification-token",
expires_at=datetime.utcnow() + timedelta(hours=1), verified_at=None))
self.bank.db.commit()
refused = self.spend(self.token_for())
self.assertEqual(refused.status_code, 403)
self.assertIn("Email not verified", refused.json()["detail"])
def test_sso_only_closes_this_door_as_well_as_the_password_one(self):
self.redis.values["settings:sso_only"] = "true"
refused = self.ask()
self.assertEqual(refused.status_code, 403)
self.assertIn("Please use SSO", refused.json()["detail"])
self.assertEqual(self.bank.db.query(LoginLink).count(), 0)
class ServiceTests(unittest.TestCase):
"""The claim is a single UPDATE, so two holders of one token cannot both win."""
def setUp(self):
self.bank = fixtures.BuilderTests()
self.bank.setUp()
self.addCleanup(self.bank.tearDown)
def test_only_one_of_two_simultaneous_claims_succeeds(self):
token = login_links.issue(self.bank.db, self.bank.owner)
self.assertIsNotNone(login_links.consume(self.bank.db, token))
self.assertIsNone(login_links.consume(self.bank.db, token))
def test_rows_long_past_their_expiry_are_swept_on_the_next_request(self):
login_links.issue(self.bank.db, self.bank.owner)
stale = self.bank.db.query(LoginLink).one()
stale.expires_at = datetime.utcnow() - timedelta(days=3)
self.bank.db.commit()
login_links.issue(self.bank.db, self.bank.owner)
self.assertEqual(self.bank.db.query(LoginLink).count(), 1)
if __name__ == "__main__":
unittest.main()

View file

@ -54,7 +54,6 @@ const CoursesPage = lazyPage(() => import('./pages/CoursesPage'))
const CourseDetailPage = lazyPage(() => import('./pages/CourseDetailPage'))
const CourseEditorPage = lazyPage(() => import('./pages/CourseEditorPage'))
const SsoCallbackPage = lazyPage(() => import('./pages/SsoCallbackPage'))
const SignInLinkPage = lazyPage(() => import('./pages/SignInLinkPage'))
function LoadingFallback() {
return <div className="loading"><div className="spinner" /></div>
@ -202,7 +201,6 @@ function AppRoutes() {
<Route path="/forgot-password" element={<ForgotPasswordPage />} />
<Route path="/reset-password" element={<ResetPasswordPage />} />
<Route path="/sso-callback" element={<SsoCallbackPage />} />
<Route path="/signin-link" element={<SignInLinkPage />} />
<Route path="/share/:token" element={<PublicQuizPage />} />
</Route>

View file

@ -24,7 +24,7 @@ function mount(Component) {
render(<MemoryRouter initialEntries={['/entry']}><AuthProvider><Routes><Route path="/entry" element={<Component />} /><Route path="/" element={<div>Signed in successfully</div>} /></Routes></AuthProvider></MemoryRouter>)
}
async function submitLogin({ revealPassword = false } = {}) {
// The standalone page leads with a link by email; the password is behind a
// The standalone page leads with a code by email; the password is behind a
// second option, and the landing modal still asks for it outright.
if (revealPassword) await userEvent.click(screen.getByRole('button', { name: 'Sign in with a password instead' }))
const form = screen.getByRole('form', { name: 'Sign in' })

View file

@ -0,0 +1,128 @@
import { render, screen, waitFor } from '@testing-library/react'
import userEvent from '@testing-library/user-event'
import { MemoryRouter, Route, Routes } from 'react-router-dom'
import { beforeEach, expect, it, vi } from 'vitest'
vi.mock('../api/client', () => ({ default: { get: vi.fn(), post: vi.fn() } }))
import LoginPage from './LoginPage'
import { AuthProvider } from '../context/AuthContext'
import api from '../api/client'
const SENT = 'If that address has an account, a sign-in code is on its way.'
const REFUSED = 'That code is not right, or it has expired. Request a new one.'
function answers({ sso = { sso_enabled: false }, policy = { registration_open: true } } = {}) {
api.get.mockImplementation(url => Promise.resolve({
data: url === '/auth/sso/config' ? sso
: url === '/auth/signup-policy' ? policy
: { id: 1, name: 'Test', role: 'user' },
}))
}
beforeEach(() => {
vi.resetAllMocks()
localStorage.clear()
answers()
api.post.mockResolvedValue({ data: { message: SENT } })
})
function mount() {
render(<MemoryRouter initialEntries={['/login']}><AuthProvider>
<Routes>
<Route path="/login" element={<LoginPage />} />
<Route path="/" element={<div>Signed in successfully</div>} />
</Routes>
</AuthProvider></MemoryRouter>)
}
async function askForCode(address = 'owner@example.test') {
await userEvent.type(screen.getByLabelText('Email'), address)
await userEvent.click(screen.getByRole('button', { name: 'Continue with email' }))
expect(api.post).toHaveBeenCalledWith('/auth/login-code', { email: address })
return screen.findByLabelText('Sign-in code')
}
it('sends a code and takes it on the same page', async () => {
mount()
expect(screen.queryByLabelText('Password')).not.toBeInTheDocument()
await askForCode()
expect(screen.getByText(SENT)).toBeInTheDocument()
// The address field is gone, so there is nothing left to fill in but the
// code and nowhere for the sign-in to travel to.
expect(screen.queryByLabelText('Email')).not.toBeInTheDocument()
api.post.mockResolvedValue({ data: { access_token: 'synthetic-code-session' } })
await userEvent.type(screen.getByLabelText('Sign-in code'), 'abc def')
await userEvent.click(screen.getByRole('button', { name: 'Sign In', exact: true }))
expect(await screen.findByText('Signed in successfully')).toBeInTheDocument()
// Typed in whatever shape it was read in; the server does the normalising,
// and is sent the address so a code is only ever tried against one account.
expect(api.post).toHaveBeenCalledWith('/auth/login-code/verify',
{ email: 'owner@example.test', code: 'ABC DEF' })
expect(localStorage.getItem('token')).toBe('synthetic-code-session')
})
it('repeats what the server said rather than claiming the mail was sent', async () => {
// The message is the one thing that must read the same for an address with
// an account and one without, so the page is not allowed a copy of its own.
api.post.mockResolvedValue({ data: { message: 'A differently worded but equally uncommitted answer.' } })
mount()
await askForCode('nobody@example.test')
expect(screen.getByText('A differently worded but equally uncommitted answer.')).toBeInTheDocument()
expect(screen.queryByText(/we sent|no account|not registered/i)).not.toBeInTheDocument()
})
it('keeps the code panel up when a code is refused', async () => {
mount()
await askForCode()
api.post.mockRejectedValue({ response: { status: 400, data: { detail: REFUSED } } })
await userEvent.type(screen.getByLabelText('Sign-in code'), 'ZZZZZZ')
await userEvent.click(screen.getByRole('button', { name: 'Sign In', exact: true }))
expect(await screen.findByText(REFUSED)).toBeInTheDocument()
// Still here to try again, and still on the address it was sent to.
expect(screen.getByLabelText('Sign-in code')).toBeInTheDocument()
})
it('shows the reason when the site refuses to send another', async () => {
mount()
await askForCode()
api.post.mockRejectedValue({ response: { status: 429, data: { detail: 'Too many sign-in codes requested for that address. Please wait a few minutes.' } } })
await userEvent.click(screen.getByRole('button', { name: 'Send a new code' }))
expect(await screen.findByText(/Too many sign-in codes/)).toBeInTheDocument()
})
it('lets somebody go back and use a different address', async () => {
mount()
await askForCode()
await userEvent.click(screen.getByRole('button', { name: 'Use a different address' }))
expect(await screen.findByLabelText('Email')).toHaveValue('owner@example.test')
expect(screen.queryByLabelText('Sign-in code')).not.toBeInTheDocument()
})
it('keeps a password one click away for whoever has one', async () => {
api.post.mockResolvedValue({ data: { access_token: 'synthetic-login-token' } })
mount()
await userEvent.click(screen.getByRole('button', { name: 'Sign in with a password instead' }))
await userEvent.type(screen.getByLabelText('Email'), 'owner@example.test')
await userEvent.type(screen.getByLabelText('Password'), 'synthetic-password')
await userEvent.click(screen.getByRole('button', { name: 'Sign In', exact: true }))
expect(await screen.findByText('Signed in successfully')).toBeInTheDocument()
expect(api.post).toHaveBeenCalledWith('/auth/login', { email: 'owner@example.test', password: 'synthetic-password' })
expect(api.post).not.toHaveBeenCalledWith('/auth/login-code', expect.anything())
})
it('leaves single sign-on the first thing on the page', async () => {
answers({ sso: { sso_enabled: true, sso_only: false, provider_name: 'Test SSO' } })
mount()
const sso = await screen.findByRole('link', { name: 'Sign in with Test SSO' })
const byEmail = screen.getByRole('button', { name: 'Continue with email' })
expect(sso.compareDocumentPosition(byEmail) & Node.DOCUMENT_POSITION_FOLLOWING).toBeTruthy()
})
it('offers registration only while the site is open to it', async () => {
answers({ policy: { registration_open: false } })
mount()
expect(await screen.findByRole('button', { name: 'Continue with email' })).toBeInTheDocument()
await waitFor(() => expect(screen.queryByRole('link', { name: 'Sign up' })).not.toBeInTheDocument())
})

View file

@ -1,133 +0,0 @@
import { StrictMode } from 'react'
import { render, screen, waitFor } from '@testing-library/react'
import userEvent from '@testing-library/user-event'
import { MemoryRouter, Route, Routes } from 'react-router-dom'
import { beforeEach, expect, it, vi } from 'vitest'
vi.mock('../api/client', () => ({ default: { get: vi.fn(), post: vi.fn() } }))
import LoginPage from './LoginPage'
import SignInLinkPage from './SignInLinkPage'
import { AuthProvider } from '../context/AuthContext'
import api from '../api/client'
const SENT = 'If that address has an account, a sign-in link is on its way.'
beforeEach(() => {
vi.resetAllMocks()
localStorage.clear()
api.get.mockImplementation(url => Promise.resolve({
data: url === '/auth/sso/config' ? { sso_enabled: false }
: url === '/auth/signup-policy' ? { registration_open: true }
: { id: 1, name: 'Test', role: 'user' },
}))
api.post.mockResolvedValue({ data: { message: SENT } })
})
function mountLogin() {
render(<MemoryRouter initialEntries={['/login']}><AuthProvider>
<Routes>
<Route path="/login" element={<LoginPage />} />
<Route path="/" element={<div>Signed in successfully</div>} />
</Routes>
</AuthProvider></MemoryRouter>)
}
function mountLink(search = '?token=synthetic-link-token', strict = false) {
const tree = (
<MemoryRouter initialEntries={[`/signin-link${search}`]}><AuthProvider>
<Routes>
<Route path="/signin-link" element={<SignInLinkPage />} />
<Route path="/" element={<div>Signed in successfully</div>} />
</Routes>
</AuthProvider></MemoryRouter>
)
render(strict ? <StrictMode>{tree}</StrictMode> : tree)
}
it('offers a link by email first and asks for no password to send one', async () => {
mountLogin()
expect(screen.queryByLabelText('Password')).not.toBeInTheDocument()
await userEvent.type(screen.getByLabelText('Email'), 'owner@example.test')
await userEvent.click(screen.getByRole('button', { name: 'Continue with email' }))
expect(api.post).toHaveBeenCalledWith('/auth/login-link', { email: 'owner@example.test' })
expect(await screen.findByText(SENT)).toBeInTheDocument()
})
it('repeats what the server said rather than claiming the mail was sent', async () => {
// The message is the one thing that must read the same for an address with
// an account and one without, so the page is not allowed a copy of its own.
api.post.mockResolvedValue({ data: { message: 'A differently worded but equally uncommitted answer.' } })
mountLogin()
await userEvent.type(screen.getByLabelText('Email'), 'nobody@example.test')
await userEvent.click(screen.getByRole('button', { name: 'Continue with email' }))
expect(await screen.findByText('A differently worded but equally uncommitted answer.')).toBeInTheDocument()
expect(screen.queryByText(/we sent|no account|not registered/i)).not.toBeInTheDocument()
})
it('shows the reason when a request is refused', async () => {
api.post.mockRejectedValue({ response: { status: 429, data: { detail: 'Too many sign-in links requested for that address. Please wait a few minutes.' } } })
mountLogin()
await userEvent.type(screen.getByLabelText('Email'), 'owner@example.test')
await userEvent.click(screen.getByRole('button', { name: 'Continue with email' }))
expect(await screen.findByText(/Too many sign-in links/)).toBeInTheDocument()
})
it('keeps a password one click away for whoever has one', async () => {
api.post.mockResolvedValue({ data: { access_token: 'synthetic-login-token' } })
mountLogin()
await userEvent.click(screen.getByRole('button', { name: 'Sign in with a password instead' }))
await userEvent.type(screen.getByLabelText('Email'), 'owner@example.test')
await userEvent.type(screen.getByLabelText('Password'), 'synthetic-password')
await userEvent.click(screen.getByRole('button', { name: 'Sign In', exact: true }))
expect(await screen.findByText('Signed in successfully')).toBeInTheDocument()
expect(api.post).toHaveBeenCalledWith('/auth/login', { email: 'owner@example.test', password: 'synthetic-password' })
expect(api.post).not.toHaveBeenCalledWith('/auth/login-link', expect.anything())
})
it('leaves single sign-on the first thing on the page', async () => {
api.get.mockImplementation(url => Promise.resolve({
data: url === '/auth/sso/config' ? { sso_enabled: true, sso_only: false, provider_name: 'Test SSO' }
: url === '/auth/signup-policy' ? { registration_open: true }
: { id: 1, name: 'Test', role: 'user' },
}))
mountLogin()
const sso = await screen.findByRole('link', { name: 'Sign in with Test SSO' })
const email = screen.getByRole('button', { name: 'Continue with email' })
expect(sso.compareDocumentPosition(email) & Node.DOCUMENT_POSITION_FOLLOWING).toBeTruthy()
})
it('offers registration only while the site is open to it', async () => {
api.get.mockImplementation(url => Promise.resolve({
data: url === '/auth/sso/config' ? { sso_enabled: false }
: url === '/auth/signup-policy' ? { registration_open: false }
: { id: 1, name: 'Test', role: 'user' },
}))
mountLogin()
expect(await screen.findByRole('button', { name: 'Continue with email' })).toBeInTheDocument()
await waitFor(() => expect(screen.queryByRole('link', { name: 'Sign up' })).not.toBeInTheDocument())
})
it('spends a followed link once and signs the reader in', async () => {
api.post.mockResolvedValue({ data: { access_token: 'synthetic-link-session' } })
// Rendered strictly, because that is what development does, and a link spent
// twice is a link that fails on arrival.
mountLink('?token=synthetic-link-token', true)
expect(await screen.findByText('Signed in successfully')).toBeInTheDocument()
expect(api.post).toHaveBeenCalledTimes(1)
expect(api.post).toHaveBeenCalledWith('/auth/login-link/consume', { token: 'synthetic-link-token' })
expect(localStorage.getItem('token')).toBe('synthetic-link-session')
})
it('says a spent or expired link is no good and offers another', async () => {
api.post.mockRejectedValue({ response: { status: 400, data: { detail: 'That sign-in link is no longer valid. If you asked more than once, open the most recent email, or request a new link.' } } })
mountLink()
expect(await screen.findByText(/no longer valid/)).toBeInTheDocument()
expect(screen.getByRole('link', { name: 'Back to sign in' })).toHaveAttribute('href', '/login')
})
it('does not call the API for a link with no token at all', async () => {
mountLink('')
expect(await screen.findByText(/incomplete/)).toBeInTheDocument()
expect(api.post).not.toHaveBeenCalled()
})

View file

@ -6,15 +6,19 @@ import api from '../api/client'
// Said by the server, and repeated here only if it says nothing. The wording
// is load-bearing it must not admit whether the address has an account so
// it lives in one place and this is not that place.
const LINK_SENT = 'If that address has an account, a sign-in link is on its way.'
const SENT = 'If that address has an account, a sign-in code is on its way.'
export default function LoginPage() {
const [email, setEmail] = useState('')
const [password, setPassword] = useState('')
const [code, setCode] = useState('')
// A password is the second way in, not the first: most people do not have
// one worth remembering, and the ones who do are one click from it.
const [withPassword, setWithPassword] = useState(false)
const [linkSent, setLinkSent] = useState('')
// The code is typed on this page rather than followed from the mail, so the
// whole sign-in happens where it started nothing to hand between devices,
// and no credential in anybody's address bar or history.
const [sent, setSent] = useState('')
const [error, setError] = useState('')
const [loading, setLoading] = useState(false)
const [unverified, setUnverified] = useState(false)
@ -24,7 +28,7 @@ export default function LoginPage() {
// Null until asked, and treated as open while unknown: a slow answer must
// not hide a way in that exists.
const [policy, setPolicy] = useState(null)
const { login } = useAuth()
const { login, loginWithToken } = useAuth()
const navigate = useNavigate()
const searchParams = new URLSearchParams(window.location.search)
const ssoError = searchParams.get('error')
@ -41,6 +45,12 @@ export default function LoginPage() {
else setError(fallback)
}
const requestCode = async () => {
const res = await api.post('/auth/login-code', { email })
setCode('')
setSent(res.data?.message || SENT)
}
const handleSubmit = async (e) => {
e.preventDefault()
setError('')
@ -50,21 +60,32 @@ export default function LoginPage() {
if (withPassword) {
await login(email, password)
navigate('/')
} else if (sent) {
const res = await api.post('/auth/login-code/verify', { email, code })
await loginWithToken(res.data.access_token)
navigate('/')
} else {
const res = await api.post('/auth/login-link', { email })
setLinkSent(res.data?.message || LINK_SENT)
await requestCode()
}
} catch (err) {
if (withPassword && err.response?.status === 403) {
setUnverified(true)
} else {
failed(err, withPassword ? 'Login failed' : 'Could not send a sign-in link. Please try again.')
failed(err, withPassword ? 'Login failed' : 'Something went wrong. Please try again.')
}
} finally {
setLoading(false)
}
}
const resend = async () => {
setError('')
setResending(true)
try { await requestCode() }
catch (err) { failed(err, 'Could not send another code. Please try again.') }
finally { setResending(false) }
}
const resendVerification = async () => {
setResending(true)
try {
@ -77,27 +98,12 @@ export default function LoginPage() {
}
}
const usePassword = () => { setLinkSent(''); setError(''); setWithPassword(true) }
const startOver = () => { setSent(''); setCode(''); setError('') }
const usePassword = () => { startOver(); setWithPassword(true) }
if (linkSent) {
return (
<div className="auth-page">
<div className="auth-card" style={{ textAlign: 'center' }}>
<div style={{ fontSize: '3rem', marginBottom: 16 }}>📧</div>
<h1 style={{ marginBottom: 12 }}>Check your email</h1>
<p style={{ color: '#64748b', marginBottom: 8 }}>{linkSent}</p>
<p style={{ color: '#64748b', fontSize: '0.875rem', marginBottom: 24 }}>
The link works once and expires in 15 minutes. Open it on any device.
</p>
<button className="btn btn-secondary" onClick={() => { setLinkSent(''); setPassword('') }}>
Use a different address
</button>
<div className="auth-link" style={{ marginTop: 16 }}>
Have a password? <button onClick={usePassword} style={{ background: 'none', border: 'none', color: 'var(--primary)', cursor: 'pointer', padding: 0 }}>Sign in with it instead</button>
</div>
</div>
</div>
)
const linkButton = {
background: 'none', border: 'none', color: 'var(--primary)',
cursor: 'pointer', padding: 0, fontSize: '0.85rem',
}
return (
@ -137,10 +143,29 @@ export default function LoginPage() {
{!ssoConfig?.sso_only && (
<>
<form aria-label="Sign in" onSubmit={handleSubmit}>
<div className="form-group">
<label htmlFor="login-email">Email</label>
<input id="login-email" type="email" autoComplete="username" value={email} onChange={e => setEmail(e.target.value)} required />
</div>
{sent && !withPassword ? (
<>
<p style={{ color: '#64748b', fontSize: '0.9rem', margin: '0 0 4px' }}>{sent}</p>
<p style={{ color: '#64748b', fontSize: '0.82rem', margin: '0 0 16px' }}>
It expires in 15 minutes. Check your spam folder if it isn't there.
</p>
<div className="form-group">
<label htmlFor="login-code">Sign-in code</label>
{/* Upper-cased as it is typed, as the invite field is. What
the server compares is normalised anyway: case, spaces
and dashes are not part of the code. */}
<input id="login-code" value={code} required autoFocus autoComplete="one-time-code"
placeholder="ABC DEF" maxLength={12}
style={{ textTransform: 'uppercase', letterSpacing: '0.18em', fontSize: '1.1rem' }}
onChange={e => setCode(e.target.value.toUpperCase())} />
</div>
</>
) : (
<div className="form-group">
<label htmlFor="login-email">Email</label>
<input id="login-email" type="email" autoComplete="username" value={email} onChange={e => setEmail(e.target.value)} required />
</div>
)}
{withPassword && (
<div className="form-group">
<label htmlFor="login-password">Password</label>
@ -148,26 +173,31 @@ export default function LoginPage() {
</div>
)}
<button className="btn btn-primary" style={{ width: '100%' }} disabled={loading}>
{withPassword
? (loading ? 'Signing in...' : 'Sign In')
{withPassword ? (loading ? 'Signing in...' : 'Sign In')
: sent ? (loading ? 'Signing in...' : 'Sign In')
: (loading ? 'Sending...' : 'Continue with email')}
</button>
</form>
{withPassword ? (
<div className="auth-link" style={{ display: 'flex', justifyContent: 'space-between', gap: 12 }}>
<button onClick={() => setWithPassword(false)} style={{ background: 'none', border: 'none', color: 'var(--primary)', cursor: 'pointer', padding: 0, fontSize: '0.85rem' }}>
Email me a link instead
</button>
<button onClick={() => setWithPassword(false)} style={linkButton}>Email me a code instead</button>
<Link to="/forgot-password" style={{ color: '#64748b', fontSize: '0.85rem' }}>Forgot password?</Link>
</div>
) : sent ? (
<div className="auth-link" style={{ display: 'flex', justifyContent: 'space-between', gap: 12 }}>
<button onClick={resend} disabled={resending} style={linkButton}>
{resending ? 'Sending…' : 'Send a new code'}
</button>
<button onClick={startOver} style={linkButton}>Use a different address</button>
</div>
) : (
<div className="auth-link">
{/* No mail to wait for if you already have a password. */}
<button onClick={usePassword} style={{ background: 'none', border: 'none', color: 'var(--primary)', cursor: 'pointer', padding: 0, fontSize: '0.85rem' }}>
Sign in with a password instead
</button>
<button onClick={usePassword} style={linkButton}>Sign in with a password instead</button>
</div>
)}
{/* Only where there is a door. Registration can be turned off
site-wide, and this offered a locked one the only way to find
out was to fill the form in and be refused. */}

View file

@ -1,54 +0,0 @@
import { useEffect, useRef, useState } from 'react'
import { useSearchParams, useNavigate, Link } from 'react-router-dom'
import { useAuth } from '../context/AuthContext'
import api from '../api/client'
/**
* Where a sign-in link lands.
*
* The link points here rather than straight at the API because spending it is
* a POST made from script: mail gateways and preview panes fetch every URL in
* a message, and a single-use token spent by a scanner is a dead link by the
* time its owner clicks it.
*/
export default function SignInLinkPage() {
const [searchParams] = useSearchParams()
const token = searchParams.get('token')
const navigate = useNavigate()
const { loginWithToken } = useAuth()
const [error, setError] = useState('')
// Asking twice spends the link and then fails on its own second request.
// React deliberately runs an effect twice in development, and state set
// inside the first run is not there yet when the second begins so the
// guard has to be a ref.
const asked = useRef(false)
useEffect(() => {
if (asked.current) return
asked.current = true
if (!token) {
setError('That link is incomplete. Request a new one.')
return
}
api.post('/auth/login-link/consume', { token })
.then(res => loginWithToken(res.data.access_token))
// Replaced rather than pushed: the token is in this URL, and there is no
// reason for the back button to bring it back.
.then(() => navigate('/', { replace: true }))
.catch(err => setError(err.response?.data?.detail
|| 'That sign-in link could not be used. Request a new one.'))
}, [token])
if (!error) return <div className="loading"><div className="spinner"></div> Signing you in...</div>
return (
<div className="auth-page">
<div className="auth-card" style={{ textAlign: 'center' }}>
<div style={{ fontSize: '3rem', marginBottom: 12 }}>🔗</div>
<h1 style={{ marginBottom: 12 }}>This link didn't work</h1>
<p style={{ color: '#64748b', marginBottom: 24 }}>{error}</p>
<Link to="/login" className="btn btn-primary" style={{ display: 'inline-block' }}>Back to sign in</Link>
</div>
</div>
)
}