import logging import httpx import secrets from urllib.parse import quote, urlencode from datetime import datetime, timedelta from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, status, Request from sqlalchemy.orm import Session from app.services import refresh_tokens, site_settings from app.database import get_db from app.models.user import User from app.models.email_verification import EmailVerification from app.models.password_reset import PasswordReset from app.schemas.auth import ( UserResponse, Token, LoginRequest, LogoutRequest, RefreshRequest, UserUpdateMe, SsoExchangeRequest, ) from app.services import email_service from app.utils.auth import ( check_rate_limit, verify_password, create_access_token, get_current_user, ) logger = logging.getLogger(__name__) router = APIRouter() def _login_key(client_ip: str) -> str: return f"login_attempts:{client_ip}" def _check_login_rate_limit(client_ip: str): """How many times an address may guess, before it has to wait. Counted per address and cleared by a success, because what this is for is guessing — and somebody who signs in correctly has not guessed. Counting successes too would lock out a hospital: one public address, a ward full of people, eleven of whom happened to open the app this afternoon. """ try: import redis as redis_lib from app.config import settings r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True, socket_connect_timeout=1) key = _login_key(client_ip) count = r.incr(key) if count == 1: r.expire(key, settings.LOGIN_WINDOW_MINUTES * 60) if count > settings.LOGIN_MAX_ATTEMPTS: raise HTTPException( status_code=429, detail=f"Too many login attempts. Try again in {settings.LOGIN_WINDOW_MINUTES} minutes.") except HTTPException: raise except Exception as e: import logging; logging.getLogger(__name__).warning(f"Redis rate limit unavailable (failing open): {e}") def _clear_login_rate_limit(client_ip: str): """A correct password is the end of the matter.""" try: import redis as redis_lib from app.config import settings redis_lib.from_url(settings.REDIS_URL, decode_responses=True, socket_connect_timeout=1).delete(_login_key(client_ip)) except Exception: pass # Rate limit: max 3 reset requests per email per hour RESET_LIMIT = 3 RESET_WINDOW_HOURS = 1 def _check_reset_rate_limit(db: Session, email: str): """Refuse a fourth request in an hour — for any address, not only a real one. Returning early for an unknown address made this the oracle the careful wording below exists to avoid: ask four times and a registered address answers 429 while an unknown one answers 200 for ever. The limit is now counted against the address as typed, so both answer the same. """ email_normalized = email.lower().strip() user = db.query(User).filter(User.email == email_normalized).first() window_start = datetime.utcnow() - timedelta(hours=RESET_WINDOW_HOURS) if user is None: # No rows to count for an address with no account, so the attempts are # counted in Redis instead — keyed by a fingerprint, because a list of # addresses somebody tried is itself worth not keeping. import hashlib from app.utils.auth import check_rate_limit check_rate_limit( key=f"pwreset:{hashlib.sha256(email_normalized.encode()).hexdigest()}", max_calls=RESET_LIMIT, window_seconds=RESET_WINDOW_HOURS * 3600, detail="Too many reset requests. Please wait before trying again.") return count = db.query(PasswordReset).filter( PasswordReset.user_id == user.id, PasswordReset.created_at >= window_start, ).count() if count >= RESET_LIMIT: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail=f"Too many reset requests. Please wait before trying again.", ) @router.post("/login", response_model=Token) async def login(login_data: LoginRequest, db: Session = Depends(get_db), request: Request = None): # Block password login if SSO-only mode sso_settings = _get_sso_settings() if sso_settings["sso_only"]: raise HTTPException(status_code=403, detail="Password login is disabled. Please use SSO.") client_ip = (request.client.host if request and request.client else "unknown") if request: _check_login_rate_limit(client_ip) email_normalized = login_data.email.lower().strip() user = db.query(User).filter(User.email == email_normalized).first() # An account with no password is not an account with the wrong password, # but it is told the same thing: which accounts have one is not a question # this endpoint answers. if not user or not user.hashed_password or not verify_password( login_data.password, user.hashed_password): raise HTTPException(status_code=401, detail="Invalid email or password") # Check email verification — skip for users without any verification record (legacy/seeded) 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 logging in.", ) _clear_login_rate_limit(client_ip) return _signed_in(db, user, login_data, request) def _signed_in(db: Session, user: User, login_data: LoginRequest, request: Request | None) -> Token: """What a successful sign-in hands back. A browser gets what it always got. A client that says it wants a refresh token also gets one, because it has nowhere safe to keep a password and no person sitting in front of it to ask again. """ # Imported here, as everywhere else in this file: a module-level `settings` # plus these function-level ones would make the name local to each of them # and blow up on first use. from app.config import settings token = Token( access_token=create_access_token(data={"sub": user.email}), expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60, ) if login_data.refresh: token.refresh_token = refresh_tokens.issue( db, user, label=login_data.device, ip=request.client.host if request and request.client else None) return token @router.post("/refresh", response_model=Token) def refresh(data: RefreshRequest, db: Session = Depends(get_db), request: Request = None): """Trade a refresh token for a new pair. The old one is spent by this call. Presenting a spent token ends the whole session it belongs to: either it was copied or a client is replaying, and from here those look the same. """ from app.config import settings client_ip = request.client.host if request and request.client else "unknown" # Flood protection, not a security control: the security here is that a # refresh token is 256 unguessable bits and spending one twice ends the # session. Generous, because an address can be a whole hospital behind one # NAT and every app launch refreshes. check_rate_limit(f"refresh:{client_ip}", settings.REFRESH_MAX_PER_HOUR, 3600, "Too many refresh attempts. Try again later.") spent = refresh_tokens.spend(db, data.refresh_token, ip=client_ip) if spent is None: raise HTTPException(401, "That sign-in has expired. Sign in again.") user, rotated = spent verification = db.query(EmailVerification).filter(EmailVerification.user_id == user.id).first() if verification and verification.verified_at is None: raise HTTPException(403, "Email not verified.") return Token( access_token=create_access_token(data={"sub": user.email}), expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60, refresh_token=rotated, ) @router.post("/logout", status_code=204) def logout(data: LogoutRequest, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): """End this session, or every session. An access token already issued is a signature and cannot be recalled; it dies of old age within the day. What this ends is the ability to get another one, which is what a lost phone actually needs. """ if data.everywhere: refresh_tokens.revoke_all(db, current_user) elif data.refresh_token: refresh_tokens.revoke_one(db, data.refresh_token) @router.get("/sessions") def list_sessions(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): """Where this account is signed in, so a person can see and end them.""" return refresh_tokens.sessions(db, current_user) @router.delete("/sessions/{family}", status_code=204) def end_session(family: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): refresh_tokens.revoke_family(db, current_user, family) # ── No password lifecycle ───────────────────────────────────────────── # # Verifying an address, resending that mail, forgetting a password and # resetting one all lived here. Accounts are at the identity provider now: it # owns the address, the passkey, the second factor and any password there is, # so a reset link minted here would set a credential nothing checks — and an # account with no password cannot forget one. # # /login stays, refused while `sso_only` is set. It is the way back in if the # provider is ever unreachable, together with the DEFAULT_ADMIN_EMAIL seed at # startup, and taking it out would leave no door at all on a bad day. @router.get("/me/settings") def get_user_settings( current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): """Get user settings. Most settings live in Redis (Nextcloud config etc.); reminders_disabled is canonical in Postgres and overrides Redis.""" data = {} try: import redis as redis_lib, json from app.config import settings as cfg r = redis_lib.from_url(cfg.REDIS_URL, decode_responses=True) raw = r.get(f"user_settings:{current_user.id}") if raw: data = json.loads(raw) except Exception as e: import logging; logging.getLogger(__name__).warning(f"Failed to load user settings: {e}") # Canonical preferences come from the DB. Redis holds the rest, and a cache # is not where a choice somebody made once should live. data["reminders_disabled"] = bool(current_user.reminders_disabled) data["tts_voice"] = current_user.tts_voice return data @router.put("/me/settings") def save_user_settings( settings_data: dict, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): """Save user settings. reminders_disabled is persisted to Postgres; other keys go to Redis.""" # Persist opt-out preference to DB (canonical source for the scheduler) changed = False if "reminders_disabled" in settings_data: current_user.reminders_disabled = bool(settings_data.get("reminders_disabled")) changed = True if "tts_voice" in settings_data: voice = (settings_data.get("tts_voice") or "").strip() current_user.tts_voice = voice or None changed = True if changed: db.add(current_user) db.commit() # Keep the full blob in Redis so other fields (Nextcloud config etc.) persist try: import redis as redis_lib, json from app.config import settings as cfg r = redis_lib.from_url(cfg.REDIS_URL, decode_responses=True) r.set(f"user_settings:{current_user.id}", json.dumps(settings_data)) except Exception as e: import logging; logging.getLogger(__name__).warning(f"Failed to save user settings to Redis: {e}") return {"saved": True} @router.get("/me", response_model=UserResponse) def get_me(current_user: User = Depends(get_current_user)): return UserResponse.of(current_user) @router.put("/me") def update_me(data: UserUpdateMe, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): """Your name. Your sign-in belongs to the provider. This used to set a password as well. There is nowhere for one to be used that the provider does not own, so writing one here would store a credential nothing checks — and the settings page that offered it invited somebody to secure their account by a route that secures nothing. """ if data.name: current_user.name = data.name db.commit() db.refresh(current_user) return { "id": current_user.id, "email": current_user.email, "name": current_user.name, "role": current_user.role, } # ── SSO / OIDC ───────────────────────────────────────────────────── def _get_sso_settings(): """Return SSO admin settings from Redis.""" try: import redis as redis_lib from app.config import settings as cfg r = redis_lib.from_url(cfg.REDIS_URL, decode_responses=True, socket_connect_timeout=1) return { "sso_only": r.get("settings:sso_only") == "true", } except Exception: return {"sso_only": False} #: How long a one-time code is worth anything. The browser is mid-redirect and #: spends it immediately; a minute is generous for that and short enough that a #: code seen in a log is already dead. SSO_CODE_TTL = 60 def _sso_store(): import redis as redis_lib from app.config import settings as cfg return redis_lib.from_url(cfg.REDIS_URL, decode_responses=True, socket_connect_timeout=1) def _stash_sso_token(token: str) -> str | None: """Park a freshly minted token behind a random code. None if Redis is away.""" code = secrets.token_urlsafe(32) try: _sso_store().setex(f"sso:exchange:{code}", SSO_CODE_TTL, token) return code except Exception: return None @router.post("/sso/exchange", response_model=Token) def sso_exchange(data: SsoExchangeRequest): """Trade a one-time code for the token it stands for. Once. The key is deleted as it is read, so a code replayed from a log, a history entry or a Referer header buys nothing. """ from app.config import settings as cfg key = f"sso:exchange:{data.code}" try: store = _sso_store() # GETDEL where the server has it, which is atomic and settles the race # between two tabs; the pipeline is the same thing for an older Redis. try: token = store.getdel(key) except Exception: pipe = store.pipeline() pipe.get(key) pipe.delete(key) token = pipe.execute()[0] except Exception: raise HTTPException(status_code=503, detail="Could not complete sign-in. Please try again.") if not token: raise HTTPException(status_code=400, detail="That sign-in link has expired. Please sign in again.") return Token(access_token=token, expires_in=cfg.ACCESS_TOKEN_EXPIRE_MINUTES * 60) def _refuse_when_sso_only(what: str = "Password login is disabled. Please use SSO.") -> None: """No password door while the site is single sign-on. Login checked this; forgot-password and reset-password did not. Accounts are made at the provider now, so what is left to guard is the password somebody already has: resetting it, or setting a new one. """ if _get_sso_settings()["sso_only"]: raise HTTPException(status_code=403, detail=what) @router.get("/sso/config") def sso_config(): """Public endpoint — tells frontend whether SSO is available and login mode.""" from app.config import settings as cfg sso_enabled = bool(cfg.OIDC_PROVIDER_URL and cfg.OIDC_CLIENT_ID) sso_settings = _get_sso_settings() from app.services import sso_roles return { "sso_enabled": sso_enabled, "sso_only": sso_settings["sso_only"], "provider_name": cfg.OIDC_PROVIDER_NAME if sso_enabled else None, # Whether roles come from the provider's groups. Says only that they # do, never which groups — enough for the access page to stop offering # a control that can now only answer 409. "roles_from_provider": sso_roles.is_configured(cfg), } @router.get("/sso/login") async def sso_login(request: Request): """Redirect user to the OIDC provider for login. `async`, and the redirect awaited. Authlib's Starlette client is the async one — `authorize_redirect` hands back a coroutine — so a sync endpoint returned that coroutine to FastAPI, which tried to serialise it as a response body and answered 500: "'coroutine' object is not iterable". The button had never worked; nothing found it because nothing had SSO configured to click it with. """ from authlib.integrations.starlette_client import OAuth from starlette.responses import RedirectResponse from app.config import settings as cfg if not cfg.OIDC_PROVIDER_URL or not cfg.OIDC_CLIENT_ID: raise HTTPException(status_code=400, detail="SSO is not configured") oauth = OAuth() oauth.register( name="oidc", server_metadata_url=f"{cfg.OIDC_PROVIDER_URL.rstrip('/')}/.well-known/openid-configuration", client_id=cfg.OIDC_CLIENT_ID, client_secret=cfg.OIDC_CLIENT_SECRET, client_kwargs={"scope": cfg.OIDC_SCOPES}, ) redirect_uri = f"{cfg.APP_URL}/api/auth/sso/callback" # Where to come back to, and whether this attempt is allowed to show a # login screen. Both live in the session rather than the URL: the callback # needs them and neither is the provider's business. # # `prompt=none` is the whole of "sign me in if you already know me". # Somebody who signed in at the provider for the other app is signed in # here too, without a screen, without a click — and somebody who is not # gets a quiet refusal we can act on rather than a login page they did not # ask for. request.session["sso_next"] = _safe_next(request.query_params.get("next")) silent = request.query_params.get("prompt") == "none" request.session["sso_silent"] = silent extra = {"prompt": "none"} if silent else {} return await oauth.oidc.authorize_redirect(request, redirect_uri, **extra) #: The provider's way of saying "nobody is signed in here" to a silent #: attempt. None of the three is a failure; they are the answer. QUIET_REFUSALS = {"login_required", "interaction_required", "consent_required", "account_selection_required"} def _safe_next(raw: str | None) -> str: """A path within this app, or the front door. Anything else — a scheme, a host, a protocol-relative `//evil` — is an open redirect, which is exactly the thing a sign-in round trip must not become. """ path = (raw or "").strip() if not path.startswith("/") or path.startswith("//"): return "/" return path[:500] @router.get("/sso/logout") async def sso_logout(request: Request): """End the session at the provider, not only here. Signing out used to mean "this app forgets you": the token went, and the provider's session did not — so pressing Sign in put you straight back in with no code. On a shared machine that is the wrong default and the one nobody expects. So the browser is sent to the provider's end-session endpoint, which ends the session behind both apps. That is the cost and it is deliberate: there is one session, and "sign out" should mean the same word in both places. No id_token_hint: neither app keeps the id token, the endpoint is per application, and the redirect is checked against the provider's own list. If the provider will not take our redirect it shows its own "you have logged out" page, which still ends the session — a worse landing, not a failure. """ from starlette.responses import RedirectResponse from app.config import settings as cfg home = f"{cfg.APP_URL}/home" if not (cfg.OIDC_PROVIDER_URL and cfg.OIDC_CLIENT_ID): return RedirectResponse(url=home) try: async with httpx.AsyncClient(timeout=8) as client: found = await client.get( f"{cfg.OIDC_PROVIDER_URL.rstrip('/')}/.well-known/openid-configuration") end_session = (found.json() or {}).get("end_session_endpoint") except Exception: logger.warning("Could not read the provider's logout endpoint", exc_info=True) end_session = None if not end_session: return RedirectResponse(url=home) query = urlencode({"post_logout_redirect_uri": home, "client_id": cfg.OIDC_CLIENT_ID}) return RedirectResponse(url=f"{end_session}?{query}") @router.get("/sso/callback") async def sso_callback(request: Request, db: Session = Depends(get_db)): """Handle OIDC provider callback — create or login user.""" from authlib.integrations.starlette_client import OAuth from starlette.responses import RedirectResponse from app.config import settings as cfg from app.services import sso_roles if not cfg.OIDC_PROVIDER_URL or not cfg.OIDC_CLIENT_ID: raise HTTPException(status_code=400, detail="SSO is not configured") oauth = OAuth() oauth.register( name="oidc", server_metadata_url=f"{cfg.OIDC_PROVIDER_URL.rstrip('/')}/.well-known/openid-configuration", client_id=cfg.OIDC_CLIENT_ID, client_secret=cfg.OIDC_CLIENT_SECRET, client_kwargs={"scope": cfg.OIDC_SCOPES}, ) silent = bool(request.session.pop("sso_silent", False)) landing = _safe_next(request.session.pop("sso_next", None)) # A silent attempt that finds nobody signed in is not a failure. The # provider says so in the query string, and the visitor should land on the # page they asked for with no message and no sign of having been anywhere. refusal = request.query_params.get("error") if refusal: if silent and refusal in QUIET_REFUSALS: return RedirectResponse(url=f"{cfg.APP_URL}{landing}") return RedirectResponse(url=f"{cfg.APP_URL}/login?error=sso_failed") try: token = await oauth.oidc.authorize_access_token(request) except Exception: if silent: # Anything at all going wrong on an attempt nobody asked for is # still not something to interrupt them with. return RedirectResponse(url=f"{cfg.APP_URL}{landing}") return RedirectResponse(url=f"{cfg.APP_URL}/login?error=sso_failed") userinfo = token.get("userinfo") or {} email = userinfo.get("email", "").lower().strip() name = userinfo.get("name") or userinfo.get("preferred_username") or email.split("@")[0] if not email: return RedirectResponse(url=f"{cfg.APP_URL}/login?error=no_email") # An address the provider will not vouch for is not an identity. Matching # on email means whoever proves an address owns the account that already # uses it, so an unverified claim would hand over an existing account to # anybody who typed the address into a provider that does not check. Only # refused when the provider says so explicitly: a provider that omits the # claim is not asserting the address is unverified. if userinfo.get("email_verified") is False: return RedirectResponse(url=f"{cfg.APP_URL}/login?error=email_unverified") # Find or create user user = db.query(User).filter(User.email == email).first() if not user: user = User( email=email, # No password rather than one nobody knows. A random string here # reads as "has a password" everywhere that asks. hashed_password=None, name=name, role="user", ) db.add(user) db.flush() # Auto-verify SSO users verification = EmailVerification( user_id=user.id, token=secrets.token_urlsafe(32), expires_at=datetime.utcnow() + timedelta(hours=1), verified_at=datetime.utcnow(), ) db.add(verification) db.commit() db.refresh(user) # The provider's directory decides what they are here, when it has been # told to. Applied on every sign-in rather than only at creation: a list # that can add somebody to the educators group and never take them out is # not a list anybody can rely on. Off unless configured — see # services/sso_roles. if sso_roles.apply(db, user, userinfo, cfg): db.commit() db.refresh(user) # A code in the address bar, never the token itself. # # This redirect is a page load: the browser asks nginx for it, and nginx # logs `"$request"` — so every sign-in wrote a live bearer token, good for # a day, into the frontend container's access log, and into the browser's # history, and into the Referer of whatever the page loaded next. The code # that goes there instead is worth one exchange, within a minute, and is # gone the moment it is spent. code = _stash_sso_token(create_access_token(data={"sub": user.email})) if code is None: logger.error("SSO succeeded but the exchange store is unreachable") return RedirectResponse(url=f"{cfg.APP_URL}/login?error=sso_failed") # Back to whatever was being opened when the round trip started, so a link # to one article lands on that article rather than the dashboard. where = f"&next={quote(landing, safe='')}" if landing and landing != "/" else "" return RedirectResponse(url=f"{cfg.APP_URL}/sso-callback?code={code}{where}")