feat(auth): add server-side session management and QR code login backend

- Add UserSession and QRLoginChallenge models for session tracking
  and mobile QR authentication
- Add session_manager utility with create/validate/revoke/cleanup
  functions and QR challenge helpers
- Add /api/sessions endpoints for listing, revoking, and
  'log off everywhere' functionality
- Add /api/qr-auth endpoints for challenge creation, polling, and
  claiming with API token issuance
- Add session config fields (lifetime, custom override, QR TTL)
- Update get_current_user to validate server-side sessions
- Create server-side sessions on all login paths (local, OAuth,
  social, admin)
- Revoke server-side session on logout
- Configure SessionMiddleware max_age from session lifetime settings
- Graceful degradation: old sessions without _session_token continue
  to work

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

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)

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

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.",
+    }

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)
 

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

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

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).
 

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}"

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",