"""FastAPI Router for JMAP (RFC 8620 & RFC 8621) Endpoints.""" from __future__ import annotations import logging from typing import Any, Dict from fastapi import APIRouter, Depends, Header, HTTPException, Request, Response, status from fastapi.responses import JSONResponse, RedirectResponse from ..jmap_engine import get_jmap_session, process_jmap_request from ..mfa import MAX_APP_PASSWORD_VERIFY_CANDIDATES from ..security import ( clear_auth_failures, client_address, ensure_auth_storage, is_rate_limited, load_panel_session, register_auth_failure, verify_admin_secret, ) from ..settings import get_settings from ..utils import verify_password logger = logging.getLogger(__name__) router = APIRouter(tags=["jmap"]) def _verify_jmap_basic(authorization: str) -> str | None: """Verify HTTP Basic credentials (admin or mailbox). Returns the username or None.""" import base64 import binascii import secrets from ..db import SessionLocal from ..models import Account, Domain, MailboxAppPassword settings = get_settings() try: decoded = base64.b64decode(authorization.split(" ", 1)[1], validate=True).decode("utf-8") except (IndexError, ValueError, binascii.Error, UnicodeDecodeError): return None if ":" not in decoded: return None user, password = decoded.split(":", 1) user = user.strip().lower() if not user or not password: return None # Mailbox credentials: primary password, or an active app password. db = SessionLocal() try: account = db.query(Account).join(Domain, Account.domain_id == Domain.id).filter(Account.username == user, Domain.is_active.is_(True)).first() if not account or not account.is_active: return None candidates: list[str] = [] if not account.require_app_password and account.password_hash: candidates.append(account.password_hash) app_passwords = ( db.query(MailboxAppPassword) .filter( MailboxAppPassword.account_id == account.id, MailboxAppPassword.revoked_at.is_(None), ) .order_by(MailboxAppPassword.id.asc()) .limit(MAX_APP_PASSWORD_VERIFY_CANDIDATES) .all() ) candidates.extend(row.password_hash for row in app_passwords if row.password_hash) for password_hash in candidates: try: if verify_password(password, password_hash): return account.username except (ValueError, TypeError): continue return None finally: db.close() def get_jmap_user(request: Request, authorization: str = Header(None)) -> str: """Extract and verify authenticated username from panel session or HTTP Basic auth.""" # JMAP uses mailbox credentials. Panel/API admin credentials do not identify # a mailbox and must not provide a path around admin MFA. # HTTP Basic auth. This is a publicly reachable password oracle, so it is # rate limited exactly like /auth/login and the panel login form. if authorization and authorization.lower().startswith("basic "): address = client_address(request) ensure_auth_storage() if is_rate_limited(address): raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Too many authentication attempts", headers={"Retry-After": str(get_settings().api_auth_block_seconds)}, ) username = _verify_jmap_basic(authorization) if username: clear_auth_failures(address) return username register_auth_failure(address) # 3. If unauthenticated, raise 401 Unauthorized raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required to access JMAP session", headers={"WWW-Authenticate": 'Basic realm="JMAP Authentication"'}, ) @router.get("/.well-known/jmap") def jmap_well_known(request: Request): """RFC 8620 Section 2.1 Well-Known JMAP Location Discovery.""" return RedirectResponse(url="/jmap/session", status_code=status.HTTP_307_TEMPORARY_REDIRECT) @router.get("/jmap") @router.get("/jmap/session") @router.get("/api/jmap/session") def jmap_session_endpoint(request: Request, username: str = Depends(get_jmap_user)): """RFC 8620 Section 2.2 JMAP Session Resource.""" host = get_settings().hostname is_secure = request.url.scheme == "https" session_data = get_jmap_session(username=username, host=host, is_secure=is_secure) return JSONResponse(content=session_data, headers={"Cache-Control": "no-cache, no-store"}) @router.post("/jmap/api") @router.post("/api/jmap/api") async def jmap_api_endpoint(request: Request, username: str = Depends(get_jmap_user)): """RFC 8620 Section 3 JMAP Request Processing.""" try: import json body = bytearray() async for chunk in request.stream(): body.extend(chunk) if len(body) > 1024 * 1024: raise HTTPException(status_code=413, detail="JMAP request exceeds 1 MiB") payload = json.loads(body) except HTTPException: raise except Exception as exc: raise HTTPException( status_code=400, detail={"type": "urn:ietf:params:jmap:error:notJSON", "status": 400, "detail": "Request body must be valid JSON"}, ) from exc if not isinstance(payload, dict): raise HTTPException( status_code=400, detail={"type": "urn:ietf:params:jmap:error:notRequest", "status": 400, "detail": "Request body must be a JSON object"}, ) calls = payload.get("methodCalls") if not isinstance(calls, list) or len(calls) > 50 or any( not isinstance(call, list) or len(call) != 3 or not isinstance(call[0], str) or not isinstance(call[1], dict) or not isinstance(call[2], str) for call in calls ): raise HTTPException(status_code=400, detail="Invalid JMAP methodCalls (maximum 50 calls)") response_data = process_jmap_request(payload, username) return JSONResponse(content=response_data, media_type="application/json") @router.get("/jmap/download/{account_id}/{blob_id}/{name}") @router.get("/api/jmap/download/{account_id}/{blob_id}/{name}") def jmap_download_blob(account_id: str, blob_id: str, name: str, request: Request, username: str = Depends(get_jmap_user)): """RFC 8620 Section 6 JMAP Blob Downloading.""" raise HTTPException(status_code=501, detail="JMAP blob downloads are not implemented") @router.post("/jmap/upload/{account_id}") @router.post("/api/jmap/upload/{account_id}") async def jmap_upload_blob(account_id: str, request: Request, username: str = Depends(get_jmap_user)): """RFC 8620 Section 6 JMAP Blob Uploading.""" raise HTTPException(status_code=501, detail="JMAP blob uploads are not implemented")