277 lines
8.2 KiB
Python
277 lines
8.2 KiB
Python
"""Access requests, magic-link tokens, and sessions (DB via function args).
|
|
|
|
Keep API handlers thin: all persistence lives here. Tokens and session ids
|
|
are bearer secrets — log user/request ids only, never secret values.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
from uuid import UUID
|
|
|
|
from sqlmodel import Session, func, select
|
|
|
|
from app.models import AccessRequest, MagicLinkToken, User, UserSession
|
|
from app.settings import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MAGIC_LINK_TTL = timedelta(minutes=15)
|
|
SESSION_TTL = timedelta(days=30)
|
|
|
|
# Public request endpoint: at most this many requests per email per hour.
|
|
REQUESTS_PER_EMAIL_PER_HOUR = 5
|
|
|
|
|
|
class AuthError(Exception):
|
|
pass
|
|
|
|
|
|
class NotFoundError(AuthError):
|
|
pass
|
|
|
|
|
|
class ConflictError(AuthError):
|
|
pass
|
|
|
|
|
|
class ForbiddenError(AuthError):
|
|
pass
|
|
|
|
|
|
class InvalidTokenError(AuthError):
|
|
pass
|
|
|
|
|
|
class ExpiredTokenError(AuthError):
|
|
pass
|
|
|
|
|
|
class UsedTokenError(AuthError):
|
|
pass
|
|
|
|
|
|
class RateLimitedError(AuthError):
|
|
pass
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def normalize_email(email: str) -> str:
|
|
return email.strip().lower()
|
|
|
|
|
|
def hash_token(token: str) -> str:
|
|
return hashlib.sha256(token.encode()).hexdigest()
|
|
|
|
|
|
def is_admin_email(email: str) -> bool:
|
|
return normalize_email(email) in settings.admin_email_list
|
|
|
|
|
|
def role_for(email: str) -> str:
|
|
return "admin" if is_admin_email(email) else "member"
|
|
|
|
|
|
def get_user(session: Session, user_id: UUID) -> User | None:
|
|
return session.get(User, user_id)
|
|
|
|
|
|
def get_user_by_email(session: Session, email: str) -> User | None:
|
|
return session.exec(
|
|
select(User).where(User.email == normalize_email(email))
|
|
).first()
|
|
|
|
|
|
def user_count(session: Session) -> int:
|
|
return session.exec(select(func.count()).select_from(User)).one()
|
|
|
|
|
|
def recent_request_count(session: Session, email: str, hours: int = 1) -> int:
|
|
since = utcnow() - timedelta(hours=hours)
|
|
return session.exec(
|
|
select(func.count())
|
|
.select_from(AccessRequest)
|
|
.where(
|
|
AccessRequest.email == normalize_email(email),
|
|
AccessRequest.created_at >= since,
|
|
)
|
|
).one()
|
|
|
|
|
|
def create_access_request(
|
|
session: Session, name: str, email: str, reason: str = ""
|
|
) -> AccessRequest:
|
|
email = normalize_email(email)
|
|
if recent_request_count(session, email) >= REQUESTS_PER_EMAIL_PER_HOUR:
|
|
raise RateLimitedError("too many requests from this email, try again later")
|
|
request = AccessRequest(name=name.strip(), email=email, reason=reason.strip())
|
|
session.add(request)
|
|
session.commit()
|
|
session.refresh(request)
|
|
logger.info("access request %s from %s", request.id, email)
|
|
return request
|
|
|
|
|
|
def get_access_request(session: Session, request_id: UUID) -> AccessRequest:
|
|
request = session.get(AccessRequest, request_id)
|
|
if request is None:
|
|
raise NotFoundError(f"access request {request_id} not found")
|
|
return request
|
|
|
|
|
|
def list_access_requests(session: Session, status: str | None = None) -> list[AccessRequest]:
|
|
query = select(AccessRequest).order_by(AccessRequest.created_at.desc())
|
|
if status is not None:
|
|
query = query.where(AccessRequest.status == status)
|
|
return list(session.exec(query).all())
|
|
|
|
|
|
def _get_or_create_user(session: Session, name: str, email: str) -> User:
|
|
user = get_user_by_email(session, email)
|
|
if user is None:
|
|
user = User(name=name.strip(), email=email, role=role_for(email))
|
|
session.add(user)
|
|
session.flush()
|
|
logger.info("user %s created for %s", user.id, email)
|
|
else:
|
|
if not user.name and name.strip():
|
|
user.name = name.strip()
|
|
new_role = role_for(email)
|
|
if user.role != new_role:
|
|
user.role = new_role
|
|
session.add(user)
|
|
session.flush()
|
|
return user
|
|
|
|
|
|
def issue_magic_token(session: Session, user: User) -> str:
|
|
token = secrets.token_urlsafe(32)
|
|
session.add(
|
|
MagicLinkToken(
|
|
user_id=user.id,
|
|
token_hash=hash_token(token),
|
|
expires_at=utcnow() + MAGIC_LINK_TTL,
|
|
)
|
|
)
|
|
session.commit()
|
|
logger.info("magic token issued for user %s", user.id)
|
|
return token
|
|
|
|
|
|
def approve_access_request(
|
|
session: Session, request_id: UUID, decided_by: str
|
|
) -> tuple[AccessRequest, User, str]:
|
|
"""Approve → upsert User + fresh magic-link token (plaintext, mail it)."""
|
|
request = get_access_request(session, request_id)
|
|
if request.status != "pending":
|
|
raise ConflictError(f"access request {request_id} already {request.status}")
|
|
user = _get_or_create_user(session, request.name, request.email)
|
|
request.status = "approved"
|
|
request.decided_at = utcnow()
|
|
request.decided_by = normalize_email(decided_by)
|
|
session.add(request)
|
|
session.commit()
|
|
session.refresh(request)
|
|
token = issue_magic_token(session, user)
|
|
logger.info(
|
|
"access request %s approved by %s; user %s",
|
|
request.id,
|
|
request.decided_by,
|
|
user.id,
|
|
)
|
|
return request, user, token
|
|
|
|
|
|
def reject_access_request(
|
|
session: Session, request_id: UUID, decided_by: str
|
|
) -> AccessRequest:
|
|
request = get_access_request(session, request_id)
|
|
if request.status != "pending":
|
|
raise ConflictError(f"access request {request_id} already {request.status}")
|
|
request.status = "rejected"
|
|
request.decided_at = utcnow()
|
|
request.decided_by = normalize_email(decided_by)
|
|
session.add(request)
|
|
session.commit()
|
|
session.refresh(request)
|
|
logger.info("access request %s rejected by %s", request.id, request.decided_by)
|
|
return request
|
|
|
|
|
|
def _aware(value: datetime) -> datetime:
|
|
# SQLite returns naive datetimes; stored values are UTC.
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value
|
|
|
|
|
|
def consume_magic_token(session: Session, token: str) -> UserSession:
|
|
"""Single-use consume → session row. used_at is set before creation."""
|
|
record = session.exec(
|
|
select(MagicLinkToken).where(
|
|
MagicLinkToken.token_hash == hash_token(token.strip())
|
|
)
|
|
).first()
|
|
if record is None:
|
|
raise InvalidTokenError("unknown token")
|
|
if record.used_at is not None:
|
|
raise UsedTokenError("token already used")
|
|
if _aware(record.expires_at) <= utcnow():
|
|
raise ExpiredTokenError("token expired")
|
|
user = session.get(User, record.user_id)
|
|
if user is None:
|
|
raise InvalidTokenError("unknown token")
|
|
# Mark used first: a reused token can never mint a second session.
|
|
record.used_at = utcnow()
|
|
session.add(record)
|
|
user_session = UserSession(
|
|
id=secrets.token_urlsafe(32),
|
|
user_id=user.id,
|
|
expires_at=utcnow() + SESSION_TTL,
|
|
)
|
|
session.add(user_session)
|
|
session.commit()
|
|
session.refresh(user_session)
|
|
logger.info("magic token consumed for user %s", user.id)
|
|
return user_session
|
|
|
|
|
|
def get_valid_session(session: Session, session_id: str) -> UserSession | None:
|
|
record = session.get(UserSession, session_id)
|
|
if record is None or record.revoked_at is not None:
|
|
return None
|
|
if _aware(record.expires_at) <= utcnow():
|
|
return None
|
|
return record
|
|
|
|
|
|
def revoke_session(session: Session, session_id: str) -> None:
|
|
record = session.get(UserSession, session_id)
|
|
if record is not None and record.revoked_at is None:
|
|
record.revoked_at = utcnow()
|
|
session.add(record)
|
|
session.commit()
|
|
|
|
|
|
def bootstrap_admin(session: Session, email: str) -> tuple[User, str]:
|
|
"""First-admin setup: only when the user table is empty AND the email
|
|
is in ADMIN_EMAILS. Issues a magic-link token like approve does."""
|
|
email = normalize_email(email)
|
|
if user_count(session) > 0:
|
|
raise ConflictError("users already exist, bootstrap is closed")
|
|
if not is_admin_email(email):
|
|
raise ForbiddenError("email is not an admin email")
|
|
user = User(name="", email=email, role="admin")
|
|
session.add(user)
|
|
session.flush()
|
|
session.commit()
|
|
session.refresh(user)
|
|
token = issue_magic_token(session, user)
|
|
logger.info("admin bootstrapped for user %s", user.id)
|
|
return user, token
|