diff --git a/app/api/__init__.py b/app/api/__init__.py index 246e47f3..94e5eaae 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -33,11 +33,13 @@ from app.api.pipelines import router as pipelines_router from app.api.plans import router as plans_router from app.api.process import router as process_router from app.api.profile import router as profile_router +from app.api.qr_auth import router as qr_auth_router from app.api.queue import router as queue_router from app.api.routing_rules import router as routing_rules_router from app.api.saved_searches import router as saved_searches_router from app.api.scheduled_jobs import router as scheduled_jobs_router from app.api.search import router as search_router +from app.api.sessions import router as sessions_router from app.api.settings import router as settings_router from app.api.shared_links import public_router as shared_links_public_router from app.api.shared_links import router as shared_links_router @@ -96,5 +98,7 @@ router.include_router(scheduled_jobs_router) router.include_router(audit_logs_router) router.include_router(i18n_router) router.include_router(mobile_router) +router.include_router(sessions_router) +router.include_router(qr_auth_router) router.include_router(compliance_router) router.include_router(translation_router) diff --git a/app/api/qr_auth.py b/app/api/qr_auth.py new file mode 100644 index 00000000..b7e1d439 --- /dev/null +++ b/app/api/qr_auth.py @@ -0,0 +1,198 @@ +"""QR code login API endpoints for mobile app authentication. + +Provides a secure challenge-response flow for logging into the mobile app +by scanning a QR code displayed in the web interface: + +1. **Web user** calls ``POST /qr-auth/challenge`` → receives a time-limited + challenge token (encoded in the QR code). +2. **Web UI** polls ``GET /qr-auth/challenge/{id}/status`` to detect when + the mobile app has claimed the challenge. +3. **Mobile app** scans the QR code and calls ``POST /qr-auth/claim`` with + the challenge token + device name → receives an API token. + +Security properties: +* Challenges expire after a configurable TTL (default 2 minutes). +* Single-use: once claimed, a challenge cannot be reused (replay-safe). +* Cryptographically random 64-byte tokens. +* IP addresses are logged for audit. +""" + +from __future__ import annotations + +import logging +from datetime import datetime +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +from app.auth import require_login +from app.database import get_db +from app.middleware.audit_log import get_client_ip +from app.utils.session_manager import ( + claim_qr_challenge, + create_qr_challenge, + get_challenge_status, +) +from app.utils.user_scope import get_current_owner_id + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/qr-auth", tags=["qr-auth"]) + +DbSession = Annotated[Session, Depends(get_db)] + + +# --------------------------------------------------------------------------- +# Auth helper +# --------------------------------------------------------------------------- + + +def _get_owner_id(request: Request) -> str: + """Return the current user's owner ID, raising 401 if unauthenticated.""" + owner_id = get_current_owner_id(request) + if not owner_id: + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated") + return owner_id + + +CurrentOwner = Annotated[str, Depends(_get_owner_id)] + + +# --------------------------------------------------------------------------- +# Request / Response schemas +# --------------------------------------------------------------------------- + + +class CreateChallengeResponse(BaseModel): + """Response after creating a QR login challenge.""" + + challenge_id: int + challenge_token: str + expires_at: datetime + qr_payload: str = Field(description="The string to encode in the QR code.") + + +class ChallengeStatusResponse(BaseModel): + """Response for polling the status of a QR challenge.""" + + id: int + status: str # "pending", "claimed", "expired", "cancelled" + device_name: str | None = None + claimed_at: datetime | None = None + expires_at: datetime + + +class ClaimChallengeRequest(BaseModel): + """Request body for claiming a QR login challenge.""" + + challenge_token: str = Field(min_length=1, max_length=256) + device_name: str = Field( + default="Mobile App", + min_length=1, + max_length=120, + description="Human-readable device name.", + ) + + +class ClaimChallengeResponse(BaseModel): + """Response after successfully claiming a QR challenge.""" + + token: str + token_id: int + name: str + owner_id: str + created_at: datetime + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +@router.post("/challenge", status_code=status.HTTP_201_CREATED, response_model=CreateChallengeResponse) +@require_login +async def create_challenge( + request: Request, + owner_id: CurrentOwner, + db: DbSession, +) -> dict[str, Any]: + """Create a new QR login challenge. + + The returned ``qr_payload`` should be encoded into a QR code and + displayed to the user. The mobile app scans this QR code and + calls the ``/claim`` endpoint. + """ + ip = get_client_ip(request) + challenge = create_qr_challenge(db, owner_id, ip_address=ip) + + # The QR payload is a JSON-like string with enough info for the mobile + # app to know the server URL and challenge token. + base_url = str(request.base_url).rstrip("/") + qr_payload = f"docuelevate://qr-login?token={challenge.challenge_token}&server={base_url}" + + return { + "challenge_id": challenge.id, + "challenge_token": challenge.challenge_token, + "expires_at": challenge.expires_at, + "qr_payload": qr_payload, + } + + +@router.get("/challenge/{challenge_id}/status", response_model=ChallengeStatusResponse) +@require_login +async def poll_challenge_status( + request: Request, + challenge_id: int, + owner_id: CurrentOwner, + db: DbSession, +) -> dict[str, Any]: + """Poll the status of a QR login challenge. + + The web UI calls this endpoint every few seconds to check if the + mobile app has scanned the QR code and claimed the challenge. + """ + result = get_challenge_status(db, challenge_id, owner_id) + if not result: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Challenge not found") + return result + + +@router.post("/claim", response_model=ClaimChallengeResponse) +async def claim_challenge( + request: Request, + body: ClaimChallengeRequest, + db: DbSession, +) -> dict[str, Any]: + """Claim a QR login challenge and receive an API token. + + This endpoint is called by the mobile app after scanning a QR code. + It does **not** require authentication — the challenge token itself + serves as proof that the user authorized this login from their web + session. + """ + ip = get_client_ip(request) + result = claim_qr_challenge(db, body.challenge_token, device_name=body.device_name, ip_address=ip) + + if not result: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Invalid, expired, or already claimed challenge.", + ) + + try: + from app.utils.audit_service import record_event + + record_event( + db, + action="qr_login_claimed", + user=result["owner_id"], + resource_type="session", + ip_address=ip, + details={"device_name": body.device_name, "token_id": result["token_id"]}, + severity="info", + ) + except Exception: + logger.debug("Failed to write QR login audit event", exc_info=True) + + return result diff --git a/app/api/sessions.py b/app/api/sessions.py new file mode 100644 index 00000000..5772016c --- /dev/null +++ b/app/api/sessions.py @@ -0,0 +1,196 @@ +"""API endpoints for managing user sessions. + +Provides endpoints for listing active sessions, revoking individual sessions, +and the "log off everywhere" feature that invalidates all sessions and API +tokens across all devices. +""" + +from __future__ import annotations + +import logging +from datetime import datetime +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import BaseModel +from sqlalchemy.orm import Session + +from app.auth import require_login +from app.database import get_db +from app.middleware.audit_log import get_client_ip +from app.utils.session_manager import ( + get_session_lifetime_days, + list_user_sessions, + revoke_all_sessions, + revoke_session, +) +from app.utils.user_scope import get_current_owner_id + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/sessions", tags=["sessions"]) + +DbSession = Annotated[Session, Depends(get_db)] + + +# --------------------------------------------------------------------------- +# Auth helper +# --------------------------------------------------------------------------- + + +def _get_owner_id(request: Request) -> str: + """Return the current user's owner ID, raising 401 if unauthenticated.""" + owner_id = get_current_owner_id(request) + if not owner_id: + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated") + return owner_id + + +CurrentOwner = Annotated[str, Depends(_get_owner_id)] + + +# --------------------------------------------------------------------------- +# Response schemas +# --------------------------------------------------------------------------- + + +class SessionResponse(BaseModel): + """Serialised user session for the management UI.""" + + id: int + device_info: str | None + ip_address: str | None + created_at: datetime + last_active_at: datetime + expires_at: datetime + is_current: bool = False + + +class SessionListResponse(BaseModel): + """Response for listing active sessions.""" + + sessions: list[SessionResponse] + session_lifetime_days: int + + +class RevokeAllResponse(BaseModel): + """Response after revoking all sessions.""" + + revoked_count: int + message: str + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +@router.get("/", response_model=SessionListResponse) +@require_login +async def list_sessions( + request: Request, + owner_id: CurrentOwner, + db: DbSession, +) -> dict[str, Any]: + """List all active sessions for the current user.""" + sessions = list_user_sessions(db, owner_id) + + # Determine which session is the current one + current_token = request.session.get("_session_token") + + session_list = [] + for s in sessions: + session_list.append( + { + "id": s.id, + "device_info": s.device_info, + "ip_address": s.ip_address, + "created_at": s.created_at, + "last_active_at": s.last_active_at, + "expires_at": s.expires_at, + "is_current": s.session_token == current_token if current_token else False, + } + ) + + return { + "sessions": session_list, + "session_lifetime_days": get_session_lifetime_days(), + } + + +@router.delete("/{session_id}", status_code=status.HTTP_204_NO_CONTENT) +@require_login +async def revoke_single_session( + request: Request, + session_id: int, + owner_id: CurrentOwner, + db: DbSession, +) -> None: + """Revoke a specific session by ID.""" + success = revoke_session(db, session_id, owner_id) + if not success: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found") + + try: + from app.utils.audit_service import record_event + + record_event( + db, + action="session_revoked", + user=owner_id, + resource_type="session", + resource_id=str(session_id), + ip_address=get_client_ip(request), + severity="info", + ) + except Exception: + logger.debug("Failed to write session revocation audit event", exc_info=True) + + +@router.post("/revoke-all", response_model=RevokeAllResponse) +@require_login +async def revoke_all( + request: Request, + owner_id: CurrentOwner, + db: DbSession, +) -> dict[str, Any]: + """Revoke all sessions except the current one ("log off everywhere"). + + Also revokes all active API tokens for the user, which invalidates + mobile app sessions and any programmatic access. + """ + # Find current session to preserve it + current_token = request.session.get("_session_token") + current_session_id = None + if current_token: + from app.models import UserSession + + current = db.query(UserSession).filter(UserSession.session_token == current_token).first() + if current: + current_session_id = current.id + + count = revoke_all_sessions( + db, + owner_id, + except_session_id=current_session_id, + revoke_api_tokens=True, + ) + + try: + from app.utils.audit_service import record_event + + record_event( + db, + action="revoke_all_sessions", + user=owner_id, + resource_type="session", + ip_address=get_client_ip(request), + details={"revoked_count": count}, + severity="warning", + ) + except Exception: + logger.debug("Failed to write revoke-all audit event", exc_info=True) + + return { + "revoked_count": count, + "message": f"Successfully revoked {count} session(s) and all API tokens.", + } diff --git a/app/auth.py b/app/auth.py index a1a40702..be1ad999 100644 --- a/app/auth.py +++ b/app/auth.py @@ -129,6 +129,25 @@ def get_current_user(request: Request): return api_user session_user = request.session.get("user") if session_user: + # Validate server-side session if a session token is present + session_token = request.session.get("_session_token") + if session_token: + try: + from app.database import SessionLocal + from app.utils.session_manager import validate_session + + db = SessionLocal() + try: + valid = validate_session(db, session_token) + if not valid: + logger.debug("[AUTH] get_current_user: server-side session invalid — clearing") + request.session.pop("user", None) + request.session.pop("_session_token", None) + return None + finally: + db.close() + except Exception: + logger.debug("[AUTH] get_current_user: session validation error", exc_info=True) logger.debug( "[AUTH] get_current_user: resolved from session (user=%s)", session_user.get("preferred_username") or session_user.get("email") or session_user.get("id"), @@ -481,6 +500,27 @@ async def social_callback(request: Request, provider: str, db: Session = Depends request.session["user"] = user_data + # Create server-side session for tracking and revocation + try: + from app.utils.session_manager import create_session + + _session_user_id = ( + user_data.get("sub") + or user_data.get("preferred_username") + or user_data.get("email") + or user_data.get("id") + ) + if _session_user_id: + user_session = create_session( + db, + user_id=_session_user_id, + ip_address=get_client_ip(request), + user_agent=request.headers.get("user-agent"), + ) + request.session["_session_token"] = user_session.session_token + except Exception: + logger.debug("[AUTH] Failed to create server-side session for social user", exc_info=True) + # Auto-create or update UserProfile _ensure_user_profile(db, user_data, is_admin=False) @@ -665,6 +705,27 @@ async def oauth_callback(request: Request, db: Session = Depends(get_db)): request.session["user"] = user_data + # Create server-side session for tracking and revocation + try: + from app.utils.session_manager import create_session + + _session_user_id = ( + user_data.get("sub") + or user_data.get("preferred_username") + or user_data.get("email") + or user_data.get("id") + ) + if _session_user_id: + user_session = create_session( + db, + user_id=_session_user_id, + ip_address=get_client_ip(request), + user_agent=request.headers.get("user-agent"), + ) + request.session["_session_token"] = user_session.session_token + except Exception: + logger.debug("[AUTH] Failed to create server-side session for OAuth user", exc_info=True) + # Auto-create or update UserProfile so the user appears in admin user management _ensure_user_profile(db, user_data, is_admin=is_admin) @@ -918,6 +979,19 @@ async def auth(request: Request, db: Session = Depends(get_db)): return RedirectResponse(url="/login?error=Invalid+username+or+password", status_code=302) user_data = _build_session_user(local_user) request.session["user"] = user_data + # Create server-side session for tracking and revocation + try: + from app.utils.session_manager import create_session + + user_session = create_session( + db, + user_id=local_user.email, + ip_address=get_client_ip(request), + user_agent=request.headers.get("user-agent"), + ) + request.session["_session_token"] = user_session.session_token + except Exception: + logger.debug("[AUTH] Failed to create server-side session", exc_info=True) logger.info("[SECURITY] LOCAL_LOGIN_SUCCESS user=%s", local_user.email) _record_login_event(db, request, local_user.email, success=True) _ensure_user_profile(db, user_data, is_admin=bool(local_user.is_admin)) @@ -974,6 +1048,20 @@ async def auth(request: Request, db: Session = Depends(get_db)): "is_admin": True, } request.session["user"] = admin_user_data + # Create server-side session for tracking and revocation + try: + from app.utils.session_manager import create_session + + admin_user_id = settings.admin_username or "admin" + user_session = create_session( + db, + user_id=admin_user_id, + ip_address=get_client_ip(request), + user_agent=request.headers.get("user-agent"), + ) + request.session["_session_token"] = user_session.session_token + except Exception: + logger.debug("[AUTH] Failed to create server-side session for admin", exc_info=True) logger.info("[SECURITY] LOCAL_LOGIN_SUCCESS user=%s", username) _record_login_event(db, request, username, success=True) _ensure_user_profile(db, admin_user_data, is_admin=True) @@ -1024,6 +1112,20 @@ async def logout(request: Request, db: Session = Depends(get_db)): ) except Exception: logger.debug("Failed to write logout audit event for user=%s", username, exc_info=True) + # Revoke server-side session + session_token = request.session.get("_session_token") + if session_token: + try: + from app.models import UserSession + + user_session = db.query(UserSession).filter(UserSession.session_token == session_token).first() + if user_session: + user_session.is_revoked = True + user_session.revoked_at = datetime.now(timezone.utc) + db.commit() + except Exception: + logger.debug("[AUTH] Failed to revoke server-side session", exc_info=True) + request.session.pop("_session_token", None) request.session.pop("user", None) return RedirectResponse(url="/login?message=You+have+been+logged+out+successfully", status_code=302) diff --git a/app/config.py b/app/config.py index ac9ab7bb..49e696b8 100644 --- a/app/config.py +++ b/app/config.py @@ -190,6 +190,26 @@ class Settings(BaseSettings): admin_username: Optional[str] = None admin_password: Optional[str] = None session_secret: Optional[str] = None + session_lifetime_days: int = Field( + default=30, + description=( + "Session lifetime in days. Common values: 30, 60, 90. " + "Determines how long a user stays logged in before being required to re-authenticate. " + "Applies to both browser sessions and the session cookie max_age." + ), + ) + session_lifetime_custom_days: int | None = Field( + default=None, + description=( + "Override session_lifetime_days with a custom value. " + "When set, this takes precedence over session_lifetime_days. " + "Useful for admin-configured non-standard durations." + ), + ) + qr_login_challenge_ttl_seconds: int = Field( + default=120, + description="Time-to-live in seconds for QR login challenges (default: 2 minutes).", + ) admin_group_name: str = "admin" # Multi-user settings diff --git a/app/main.py b/app/main.py index 97bb1d57..c77e5701 100644 --- a/app/main.py +++ b/app/main.py @@ -318,8 +318,19 @@ app.add_middleware(CSRFMiddleware, config=settings) # See SECURITY_AUDIT.md – Infrastructure Security section app.add_middleware(AuditLogMiddleware, config=settings) + # 3) Session Middleware (for request.session to work) -app.add_middleware(SessionMiddleware, secret_key=SESSION_SECRET) +def _get_session_max_age() -> int: + """Compute session max-age at startup time.""" + try: + from app.utils.session_manager import get_session_max_age_seconds + + return get_session_max_age_seconds() + except Exception: + return 30 * 86400 # 30 days default fallback + + +app.add_middleware(SessionMiddleware, secret_key=SESSION_SECRET, max_age=_get_session_max_age()) # 3a) CORS Middleware - handles cross-origin requests and preflight (OPTIONS) responses. # Disabled by default: set CORS_ENABLED=True only when NOT using a reverse proxy diff --git a/app/models.py b/app/models.py index 54cef37d..fe5b9c41 100644 --- a/app/models.py +++ b/app/models.py @@ -969,6 +969,86 @@ class MobileDevice(Base): __table_args__ = (UniqueConstraint("owner_id", "push_token", name="uq_mobile_device_owner_token"),) +class UserSession(Base): + """Server-side session tracking for invalidation and device management. + + Each row represents an active browser or app session. The ``session_token`` + is stored in the user's cookie and validated on every authenticated request. + Revoking a row (``is_revoked=True``) immediately terminates that session + on the next request. + """ + + __tablename__ = "user_sessions" + + id = Column(Integer, primary_key=True, index=True) + + # Cryptographically random token stored in the session cookie. + session_token = Column(String(128), unique=True, nullable=False, index=True) + + # Stable owner identifier — matches FileRecord.owner_id. + user_id = Column(String, nullable=False, index=True) + + # Client metadata for display in the session management UI. + ip_address = Column(String(45), nullable=True) + user_agent = Column(String(512), nullable=True) + device_info = Column(String(255), nullable=True) + + is_revoked = Column(Boolean, nullable=False, default=False) + created_at = Column(DateTime(timezone=True), server_default=func.now()) + last_active_at = Column(DateTime(timezone=True), server_default=func.now()) + expires_at = Column(DateTime(timezone=True), nullable=False) + revoked_at = Column(DateTime(timezone=True), nullable=True) + + +class QRLoginChallenge(Base): + """Time-limited QR code login challenge for mobile app authentication. + + A logged-in web user generates a challenge that produces a QR code. The + mobile app scans the QR code and calls the claim endpoint with the + ``challenge_token``. The server verifies the challenge is still valid, + unclaimed, and unexpired, then issues an API token for the mobile app. + + Security properties: + * Time-bound (default 2 minutes). + * Single-use (``is_claimed`` prevents replay). + * Cryptographically random 64-byte token. + * Bound to the creating user — only that user's mobile device receives a + token. + """ + + __tablename__ = "qr_login_challenges" + + id = Column(Integer, primary_key=True, index=True) + + # Cryptographically random token encoded in the QR code. + challenge_token = Column(String(128), unique=True, nullable=False, index=True) + + # The user who created this challenge (from the web session). + user_id = Column(String, nullable=False, index=True) + + # Whether the challenge has been successfully claimed by a mobile app. + is_claimed = Column(Boolean, nullable=False, default=False) + + # Whether the challenge has been explicitly cancelled or expired. + is_cancelled = Column(Boolean, nullable=False, default=False) + + # IP address of the web client that created the challenge. + created_by_ip = Column(String(45), nullable=True) + + # IP address of the mobile client that claimed the challenge. + claimed_by_ip = Column(String(45), nullable=True) + + # Device name provided by the mobile app when claiming. + device_name = Column(String(255), nullable=True) + + # The API token ID that was issued to the mobile app (for audit trail). + issued_token_id = Column(Integer, nullable=True) + + created_at = Column(DateTime(timezone=True), server_default=func.now()) + expires_at = Column(DateTime(timezone=True), nullable=False) + claimed_at = Column(DateTime(timezone=True), nullable=True) + + class ComplianceTemplate(Base): """Pre-built compliance configuration templates (GDPR, HIPAA, SOC2). diff --git a/app/utils/session_manager.py b/app/utils/session_manager.py new file mode 100644 index 00000000..aaeda266 --- /dev/null +++ b/app/utils/session_manager.py @@ -0,0 +1,480 @@ +"""Server-side session management utilities. + +Provides helpers for creating, validating, and revoking user sessions. +Sessions are tracked in the ``user_sessions`` table and referenced by a +cryptographically random token stored in the browser cookie. This enables +the "log off everywhere" feature and per-session revocation. +""" + +from __future__ import annotations + +import logging +import secrets +from datetime import datetime, timedelta, timezone + +from sqlalchemy.orm import Session + +from app.config import settings +from app.models import ApiToken, QRLoginChallenge, UserSession + +logger = logging.getLogger(__name__) + + +def get_session_lifetime_days() -> int: + """Return the effective session lifetime in days. + + If ``session_lifetime_custom_days`` is set it takes precedence over + ``session_lifetime_days``. + """ + custom = getattr(settings, "session_lifetime_custom_days", None) + if custom is not None and isinstance(custom, int) and custom > 0: + return custom + return max(1, getattr(settings, "session_lifetime_days", 30)) + + +def get_session_max_age_seconds() -> int: + """Return the session max-age in seconds for the cookie.""" + return get_session_lifetime_days() * 86400 + + +def create_session( + db: Session, + user_id: str, + ip_address: str | None = None, + user_agent: str | None = None, +) -> UserSession: + """Create a new server-side session record. + + Args: + db: Database session. + user_id: Stable owner identifier. + ip_address: Client IP address. + user_agent: Client User-Agent header. + + Returns: + The newly created ``UserSession`` instance. + """ + session_token = secrets.token_urlsafe(64) + now = datetime.now(timezone.utc) + lifetime_days = get_session_lifetime_days() + expires_at = now + timedelta(days=lifetime_days) + + device_info = _parse_device_info(user_agent) + + user_session = UserSession( + session_token=session_token, + user_id=user_id, + ip_address=ip_address, + user_agent=(user_agent or "")[:512], + device_info=device_info, + created_at=now, + last_active_at=now, + expires_at=expires_at, + ) + try: + db.add(user_session) + db.commit() + db.refresh(user_session) + except Exception: + db.rollback() + logger.exception("Failed to create session for user_id=%s", user_id) + raise + + logger.info( + "[SESSION] Created session id=%s user=%s device=%r expires=%s", + user_session.id, + user_id, + device_info, + expires_at.isoformat(), + ) + return user_session + + +def validate_session(db: Session, session_token: str) -> UserSession | None: + """Validate a session token and return the session if valid. + + A session is valid when: + * It exists in the database. + * ``is_revoked`` is ``False``. + * ``expires_at`` is in the future. + + Side-effect: updates ``last_active_at`` on valid sessions. + + Returns: + The ``UserSession`` if valid, else ``None``. + """ + if not session_token: + return None + + now = datetime.now(timezone.utc) + user_session = db.query(UserSession).filter(UserSession.session_token == session_token).first() + + if not user_session: + logger.debug("[SESSION] Token not found in database") + return None + + if user_session.is_revoked: + logger.debug("[SESSION] Session id=%s is revoked", user_session.id) + return None + + if user_session.expires_at and user_session.expires_at < now: + logger.debug("[SESSION] Session id=%s has expired", user_session.id) + return None + + # Update last_active_at (throttled to avoid excessive writes) + if not user_session.last_active_at or (now - user_session.last_active_at).total_seconds() > 60: + try: + user_session.last_active_at = now + db.commit() + except Exception: + db.rollback() + logger.debug("[SESSION] Failed to update last_active_at for session id=%s", user_session.id) + + return user_session + + +def revoke_session(db: Session, session_id: int, user_id: str) -> bool: + """Revoke a single session by ID. + + Args: + db: Database session. + session_id: The session record ID to revoke. + user_id: The owner — ensures a user can only revoke their own sessions. + + Returns: + ``True`` if the session was found and revoked, ``False`` otherwise. + """ + user_session = db.get(UserSession, session_id) + if not user_session or user_session.user_id != user_id: + return False + + now = datetime.now(timezone.utc) + user_session.is_revoked = True + user_session.revoked_at = now + try: + db.commit() + except Exception: + db.rollback() + raise + + logger.info("[SESSION] Revoked session id=%s user=%s", session_id, user_id) + return True + + +def revoke_all_sessions( + db: Session, + user_id: str, + *, + except_session_id: int | None = None, + revoke_api_tokens: bool = True, +) -> int: + """Revoke all active sessions for a user ("log off everywhere"). + + Args: + db: Database session. + user_id: The owner whose sessions should be revoked. + except_session_id: If provided, keep this session active (the + current browser session). + revoke_api_tokens: If ``True``, also revoke all active API tokens. + + Returns: + Number of sessions revoked. + """ + now = datetime.now(timezone.utc) + query = db.query(UserSession).filter( + UserSession.user_id == user_id, + UserSession.is_revoked.is_(False), + ) + if except_session_id is not None: + query = query.filter(UserSession.id != except_session_id) + + sessions = query.all() + count = 0 + for s in sessions: + s.is_revoked = True + s.revoked_at = now + count += 1 + + if revoke_api_tokens: + tokens = ( + db.query(ApiToken) + .filter( + ApiToken.owner_id == user_id, + ApiToken.is_active.is_(True), + ) + .all() + ) + for t in tokens: + t.is_active = False + t.revoked_at = now + + try: + db.commit() + except Exception: + db.rollback() + raise + + logger.info( + "[SESSION] Revoked all sessions for user=%s (count=%d, except_session_id=%s, tokens_revoked=%s)", + user_id, + count, + except_session_id, + revoke_api_tokens, + ) + return count + + +def list_user_sessions(db: Session, user_id: str) -> list[UserSession]: + """Return all non-revoked, non-expired sessions for a user. + + Results are ordered by most recently active first. + """ + now = datetime.now(timezone.utc) + return ( + db.query(UserSession) + .filter( + UserSession.user_id == user_id, + UserSession.is_revoked.is_(False), + UserSession.expires_at > now, + ) + .order_by(UserSession.last_active_at.desc()) + .all() + ) + + +def cleanup_expired_sessions(db: Session) -> int: + """Delete sessions that expired more than 7 days ago. + + Intended to be called periodically (e.g. via Celery beat) to keep the + table from growing unbounded. + + Returns: + Number of rows deleted. + """ + cutoff = datetime.now(timezone.utc) - timedelta(days=7) + count = db.query(UserSession).filter(UserSession.expires_at < cutoff).delete(synchronize_session=False) + try: + db.commit() + except Exception: + db.rollback() + raise + if count: + logger.info("[SESSION] Cleaned up %d expired sessions", count) + return count + + +# --------------------------------------------------------------------------- +# QR login helpers +# --------------------------------------------------------------------------- + + +def create_qr_challenge(db: Session, user_id: str, ip_address: str | None = None) -> QRLoginChallenge: + """Create a new QR login challenge. + + Args: + db: Database session. + user_id: The authenticated web user creating the challenge. + ip_address: IP address of the web client. + + Returns: + The newly created ``QRLoginChallenge``. + """ + token = secrets.token_urlsafe(64) + ttl = getattr(settings, "qr_login_challenge_ttl_seconds", 120) + now = datetime.now(timezone.utc) + expires_at = now + timedelta(seconds=ttl) + + challenge = QRLoginChallenge( + challenge_token=token, + user_id=user_id, + created_by_ip=ip_address, + created_at=now, + expires_at=expires_at, + ) + try: + db.add(challenge) + db.commit() + db.refresh(challenge) + except Exception: + db.rollback() + logger.exception("Failed to create QR login challenge for user_id=%s", user_id) + raise + + logger.info("[QR_AUTH] Challenge created: id=%s user=%s expires=%s", challenge.id, user_id, expires_at.isoformat()) + return challenge + + +def validate_qr_challenge(db: Session, challenge_token: str) -> QRLoginChallenge | None: + """Validate a QR challenge token without claiming it. + + Returns the challenge if it exists, is not expired, not claimed, + and not cancelled. Returns ``None`` otherwise. + """ + if not challenge_token: + return None + + now = datetime.now(timezone.utc) + challenge = db.query(QRLoginChallenge).filter(QRLoginChallenge.challenge_token == challenge_token).first() + + if not challenge: + return None + if challenge.is_claimed or challenge.is_cancelled: + return None + if challenge.expires_at < now: + return None + + return challenge + + +def claim_qr_challenge( + db: Session, + challenge_token: str, + device_name: str = "Mobile App", + ip_address: str | None = None, +) -> dict | None: + """Claim a QR challenge and issue an API token. + + This is the critical security path. The challenge is validated, + marked as claimed atomically, and an API token is issued for the + user who created the challenge. + + Args: + db: Database session. + challenge_token: The token from the QR code. + device_name: Name provided by the mobile app. + ip_address: IP address of the claiming mobile device. + + Returns: + Dict with ``token`` (plaintext), ``token_id``, ``name``, ``owner_id`` + and ``created_at`` on success, or ``None`` if the challenge is invalid. + """ + from app.api.api_tokens import generate_api_token, hash_token + + challenge = validate_qr_challenge(db, challenge_token) + if not challenge: + logger.warning("[QR_AUTH] Invalid or expired challenge token attempted") + return None + + now = datetime.now(timezone.utc) + + # Mark as claimed first to prevent race conditions + challenge.is_claimed = True + challenge.claimed_at = now + challenge.claimed_by_ip = ip_address + challenge.device_name = device_name + + # Generate API token for the mobile app + token_name = f"Mobile App (QR) – {device_name}" + plaintext = generate_api_token() + token_hash_value = hash_token(plaintext) + prefix = plaintext[:12] + + db_token = ApiToken( + owner_id=challenge.user_id, + name=token_name, + token_hash=token_hash_value, + token_prefix=prefix, + ) + + try: + db.add(db_token) + db.flush() + challenge.issued_token_id = db_token.id + db.commit() + db.refresh(db_token) + except Exception: + db.rollback() + logger.exception("[QR_AUTH] Failed to issue token for challenge id=%s", challenge.id) + raise + + logger.info( + "[QR_AUTH] Challenge claimed: id=%s user=%s device=%r token_id=%s", + challenge.id, + challenge.user_id, + device_name, + db_token.id, + ) + + return { + "token": plaintext, + "token_id": db_token.id, + "name": token_name, + "owner_id": challenge.user_id, + "created_at": db_token.created_at, + } + + +def get_challenge_status(db: Session, challenge_id: int, user_id: str) -> dict | None: + """Get the current status of a QR challenge (for polling from the web UI). + + Returns: + Dict with ``status`` ("pending", "claimed", "expired", "cancelled") + and metadata, or ``None`` if the challenge doesn't belong to the user. + """ + challenge = db.get(QRLoginChallenge, challenge_id) + if not challenge or challenge.user_id != user_id: + return None + + now = datetime.now(timezone.utc) + if challenge.is_claimed: + status = "claimed" + elif challenge.is_cancelled: + status = "cancelled" + elif challenge.expires_at < now: + status = "expired" + else: + status = "pending" + + return { + "id": challenge.id, + "status": status, + "device_name": challenge.device_name, + "claimed_at": challenge.claimed_at, + "expires_at": challenge.expires_at, + } + + +def _parse_device_info(user_agent: str | None) -> str | None: + """Extract a human-readable device description from User-Agent. + + This is a lightweight parser — not a full UA library — that covers + the most common browsers and platforms. + """ + if not user_agent: + return None + + ua = user_agent.lower() + + # Platform detection + platform = "Unknown" + if "iphone" in ua: + platform = "iPhone" + elif "ipad" in ua: + platform = "iPad" + elif "android" in ua: + platform = "Android" + elif "macintosh" in ua or "mac os" in ua: + platform = "macOS" + elif "windows" in ua: + platform = "Windows" + elif "linux" in ua: + platform = "Linux" + elif "cros" in ua: + platform = "ChromeOS" + + # Browser detection + browser = "Unknown Browser" + if "edg/" in ua or "edge/" in ua: + browser = "Edge" + elif "opr/" in ua or "opera" in ua: + browser = "Opera" + elif "chrome/" in ua and "safari/" in ua: + browser = "Chrome" + elif "safari/" in ua and "chrome/" not in ua: + browser = "Safari" + elif "firefox/" in ua: + browser = "Firefox" + elif "docuelevate" in ua: + browser = "DocuElevate App" + + return f"{browser} on {platform}" diff --git a/app/utils/settings_service.py b/app/utils/settings_service.py index e130fbd9..ae1f52d9 100644 --- a/app/utils/settings_service.py +++ b/app/utils/settings_service.py @@ -134,6 +134,30 @@ SETTING_METADATA = { "required": True, # Required when auth_enabled=True (validated in config.py) "restart_required": True, }, + "session_lifetime_days": { + "category": "Authentication", + "description": "Session lifetime in days (default 30). Determines how long a user stays logged in.", + "type": "integer", + "sensitive": False, + "required": False, + "restart_required": True, + }, + "session_lifetime_custom_days": { + "category": "Authentication", + "description": "Override session_lifetime_days with a custom value. Takes precedence when set.", + "type": "integer", + "sensitive": False, + "required": False, + "restart_required": True, + }, + "qr_login_challenge_ttl_seconds": { + "category": "Authentication", + "description": "Time-to-live in seconds for QR login challenges (default 120).", + "type": "integer", + "sensitive": False, + "required": False, + "restart_required": False, + }, "admin_username": { "category": "Authentication", "description": "Admin username for local authentication",