feat: email-first login with allow-check and access forward

This commit is contained in:
Jonas Miederer 2026-09-12 11:20:55 +02:00
parent 82f7ff8b02
commit b533549e45
8 changed files with 224 additions and 21 deletions

View file

@ -21,6 +21,8 @@ from .schemas import (
AccessRequestRead,
AdminBootstrapRequest,
MagicConsumeRequest,
MagicRequestRead,
MagicRequestRequest,
MeRead,
MessageRead,
)
@ -120,6 +122,26 @@ def reject_request(
return _read_request(request)
@auth_router.post("/magic/request", response_model=MagicRequestRead)
def request_magic(
data: MagicRequestRequest,
session: Session = Depends(get_db),
):
"""Email-first login: allowed emails get a magic link mailed.
403 means "not allowed" — the frontend forwards those to /access.
The token travels by mail only, never in the response.
"""
try:
user, token = service.request_magic_link(session, data.email)
except service.ForbiddenError as exc:
raise HTTPException(status_code=403, detail=str(exc)) from exc
except service.RateLimitedError as exc:
raise HTTPException(status_code=429, detail=str(exc)) from exc
email_sent = mail.send_magic_link(user.email, mail.magic_link(token))
return MagicRequestRead(message="login link sent", email_sent=email_sent)
@auth_router.post("/magic/consume", response_model=MeRead)
def consume_magic(
data: MagicConsumeRequest,

View file

@ -33,6 +33,15 @@ class MagicConsumeRequest(BaseModel):
token: str = Field(min_length=1, max_length=200)
class MagicRequestRequest(BaseModel):
email: str = Field(min_length=3, max_length=320)
class MagicRequestRead(BaseModel):
message: str
email_sent: bool
class MeRead(BaseModel):
id: UUID
email: str

View file

@ -24,6 +24,8 @@ SESSION_TTL = timedelta(days=30)
# Public request endpoint: at most this many requests per email per hour.
REQUESTS_PER_EMAIL_PER_HOUR = 5
# Same for magic-link requests (email bombing / token spam protection).
MAGIC_LINKS_PER_EMAIL_PER_HOUR = 5
class AuthError(Exception):
@ -164,6 +166,39 @@ def issue_magic_token(session: Session, user: User) -> str:
return token
def recent_token_count(session: Session, user: User, hours: int = 1) -> int:
since = utcnow() - timedelta(hours=hours)
return session.exec(
select(func.count())
.select_from(MagicLinkToken)
.where(
MagicLinkToken.user_id == user.id,
MagicLinkToken.created_at >= since,
)
).one()
def request_magic_link(session: Session, email: str) -> tuple[User, str]:
"""Self-service login link for an allowed email.
Allowed = approved before (has a User row) or listed in ADMIN_EMAILS
(admin user is created on first request). Anything else raises
ForbiddenError — the frontend forwards those to the access form.
Note: 403-vs-200 distinguishes registered emails by design here.
"""
email = normalize_email(email)
user = get_user_by_email(session, email)
if user is None:
if not is_admin_email(email):
raise ForbiddenError("email is not allowed to log in")
user = _get_or_create_user(session, "", email)
session.commit()
session.refresh(user)
if recent_token_count(session, user) >= MAGIC_LINKS_PER_EMAIL_PER_HOUR:
raise RateLimitedError("too many login links, try again later")
return user, issue_magic_token(session, user)
def approve_access_request(
session: Session, request_id: UUID, decided_by: str
) -> tuple[AccessRequest, User, str]:

View file

@ -268,3 +268,51 @@ async def test_approve_survives_delivery_failure(client, session, monkeypatch):
body = approved.json()
assert body["request"]["status"] == "approved"
assert body["email_sent"] is False
async def test_magic_request_for_known_user(client, session):
_get_or_make_user(session, MEMBER)
response = await client.post(
"/api/auth/magic/request", json={"email": MEMBER}
)
assert response.status_code == 200, response.text
body = response.json()
assert body["message"] == "login link sent"
assert body["email_sent"] is False # no SMTP in tests
assert "token" not in response.text
tokens = session.exec(
select(MagicLinkToken).where(
MagicLinkToken.user_id
== service.get_user_by_email(session, MEMBER).id
)
).all()
assert len(tokens) == 1
async def test_magic_request_unknown_email_403(client):
response = await client.post(
"/api/auth/magic/request", json={"email": "stranger@example.com"}
)
assert response.status_code == 403
async def test_magic_request_admin_email_creates_admin(client, session):
response = await client.post(
"/api/auth/magic/request", json={"email": ADMIN}
)
assert response.status_code == 200, response.text
user = service.get_user_by_email(session, ADMIN)
assert user is not None and user.role == "admin"
async def test_magic_request_rate_limit(client, session):
_get_or_make_user(session, MEMBER)
for _ in range(service.MAGIC_LINKS_PER_EMAIL_PER_HOUR):
response = await client.post(
"/api/auth/magic/request", json={"email": MEMBER}
)
assert response.status_code == 200, response.text
limited = await client.post(
"/api/auth/magic/request", json={"email": MEMBER}
)
assert limited.status_code == 429