diff --git a/.env.demo b/.env.demo index d3dc2a46..1d04bab7 100644 --- a/.env.demo +++ b/.env.demo @@ -164,6 +164,16 @@ AUTH_ENABLED=true # Generate a secure random string, for example: # python -c "import secrets; print(secrets.token_hex(32))" SESSION_SECRET=b39fd43f68d0491ca942f28a16e484b1e763fe9accf4445ca2669a5f3b179eb4 + +# Session lifetime in days (default: 30). Common values: 30, 60, 90. +# Determines how long a user stays logged in before needing to re-authenticate. +# SESSION_LIFETIME_DAYS=30 +# Override with a custom value (takes precedence over SESSION_LIFETIME_DAYS): +# SESSION_LIFETIME_CUSTOM_DAYS= + +# Time-to-live in seconds for QR code login challenges (default: 120 = 2 minutes). +# QR_LOGIN_CHALLENGE_TTL_SECONDS=120 + ADMIN_USERNAME=admin ADMIN_PASSWORD=your_secure_password ADMIN_GROUP_NAME=admin diff --git a/BUILD_DATE b/BUILD_DATE index 98cb717d..81deb728 100644 --- a/BUILD_DATE +++ b/BUILD_DATE @@ -1 +1 @@ -2026-03-17T09:55:43Z +2026-03-17T11:23:05Z diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f53d655..19be28ad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 +## v0.155.0 (2026-03-17) + + ## v0.154.0 (2026-03-17) ### Bug Fixes diff --git a/GIT_SHA b/GIT_SHA index a705a348..78a6cd29 100644 --- a/GIT_SHA +++ b/GIT_SHA @@ -1 +1 @@ -7b7554d +30c2e9a diff --git a/RUNTIME_INFO b/RUNTIME_INFO index 424ae429..b9756ce3 100644 --- a/RUNTIME_INFO +++ b/RUNTIME_INFO @@ -1,10 +1,10 @@ DocuElevate Build Information ============================== -Version: 0.154.0 -Build Date: 2026-03-17T09:55:43Z -Git Commit: 7b7554decf527d2c9f7eee653a1f503249b7d138 -Git Short SHA: 7b7554d +Version: 0.155.0 +Build Date: 2026-03-17T11:23:05Z +Git Commit: 30c2e9afefc57c8d1e19548afec7475f5a838f24 +Git Short SHA: 30c2e9a Git Branch: main -Commit Date: 2026-03-17T10:55:10+01:00 -Build Timestamp: 2026-03-17T09:55:43Z +Commit Date: 2026-03-17T12:22:07+01:00 +Build Timestamp: 2026-03-17T11:23:05Z ============================== diff --git a/VERSION b/VERSION index a1046a0b..03ed6e33 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.154.0 +0.155.0 diff --git a/app/api/__init__.py b/app/api/__init__.py index 4234ad7e..871daa72 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -34,11 +34,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 @@ -98,6 +100,8 @@ 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(system_reset_router) router.include_router(translation_router) diff --git a/app/api/api_tokens.py b/app/api/api_tokens.py index a53ba7f2..1beef61b 100644 --- a/app/api/api_tokens.py +++ b/app/api/api_tokens.py @@ -42,6 +42,9 @@ TOKEN_HASH_ITERATIONS = 100_000 #: PBKDF2 salt for API token hashing (not secret, but fixed for determinism). TOKEN_HASH_SALT = b"api-token-v1" +#: Name prefix used for tokens created by the mobile app flow. +MOBILE_TOKEN_PREFIX = "Mobile App" + # --------------------------------------------------------------------------- # Auth helper @@ -91,6 +94,20 @@ def hash_token(token: str) -> str: return dk.hex() +def _token_to_dict(t: ApiToken) -> dict[str, Any]: + """Convert an ``ApiToken`` ORM instance to a serialisable dict.""" + return { + "id": t.id, + "name": t.name, + "token_prefix": t.token_prefix, + "is_active": t.is_active, + "last_used_at": t.last_used_at, + "last_used_ip": t.last_used_ip, + "created_at": t.created_at, + "revoked_at": t.revoked_at, + } + + # --------------------------------------------------------------------------- # Pydantic schemas # --------------------------------------------------------------------------- @@ -177,21 +194,44 @@ async def list_tokens( owner_id: CurrentOwner, db: DbSession, ) -> list[dict[str, Any]]: - """List all API tokens for the authenticated user.""" - tokens = db.query(ApiToken).filter(ApiToken.owner_id == owner_id).order_by(ApiToken.created_at.desc()).all() - return [ - { - "id": t.id, - "name": t.name, - "token_prefix": t.token_prefix, - "is_active": t.is_active, - "last_used_at": t.last_used_at, - "last_used_ip": t.last_used_ip, - "created_at": t.created_at, - "revoked_at": t.revoked_at, - } - for t in tokens - ] + """List non-mobile API tokens for the authenticated user. + + Mobile tokens (whose names start with ``"Mobile App"``) are excluded + from this list; they are managed on the dedicated Devices page via + ``GET /api/api-tokens/mobile``. + """ + tokens = ( + db.query(ApiToken) + .filter( + ApiToken.owner_id == owner_id, + ~ApiToken.name.startswith(MOBILE_TOKEN_PREFIX), + ) + .order_by(ApiToken.created_at.desc()) + .all() + ) + return [_token_to_dict(t) for t in tokens] + + +@router.get("/mobile", response_model=list[TokenResponse]) +async def list_mobile_tokens( + owner_id: CurrentOwner, + db: DbSession, +) -> list[dict[str, Any]]: + """List mobile API tokens for the authenticated user. + + Returns tokens whose names start with ``"Mobile App"`` — these are + created via the mobile SSO flow or QR code login. + """ + tokens = ( + db.query(ApiToken) + .filter( + ApiToken.owner_id == owner_id, + ApiToken.name.startswith(MOBILE_TOKEN_PREFIX), + ) + .order_by(ApiToken.created_at.desc()) + .all() + ) + return [_token_to_dict(t) for t in tokens] @router.delete("/{token_id}", status_code=status.HTTP_200_OK) 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..c5883dac 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.utils.session_manager import validate_session + + user_session = validate_session(db, session_token) + 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 cad16a06..1f5a25be 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 a3b56ec4..21b50f30 100644 --- a/app/main.py +++ b/app/main.py @@ -324,8 +324,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 86522ac1..e7e1b3a3 100644 --- a/app/models.py +++ b/app/models.py @@ -1014,6 +1014,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..e0de02bc --- /dev/null +++ b/app/utils/session_manager.py @@ -0,0 +1,505 @@ +"""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 _ensure_tz_aware(dt: datetime | None) -> datetime | None: + """Return *dt* with UTC tzinfo if it is naive, or unchanged if already aware. + + SQLite does not persist timezone information, so datetimes read back from + the database are offset-naive. This helper normalises them for safe + comparison with ``datetime.now(timezone.utc)``. + """ + if dt is not None and dt.tzinfo is None: + return dt.replace(tzinfo=timezone.utc) + return dt + + +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: + expires = _ensure_tz_aware(user_session.expires_at) + if expires < now: + logger.debug("[SESSION] Session id=%s has expired", user_session.id) + return None + + # Update last_active_at (throttled to avoid excessive writes) + last_active = _ensure_tz_aware(user_session.last_active_at) + if not last_active or (now - last_active).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) + sessions = ( + db.query(UserSession) + .filter( + UserSession.user_id == user_id, + UserSession.is_revoked.is_(False), + ) + .order_by(UserSession.last_active_at.desc()) + .all() + ) + # Filter expired sessions in Python to handle timezone-naive datetimes (SQLite) + result = [] + for s in sessions: + expires = _ensure_tz_aware(s.expires_at) + if expires and expires > now: + result.append(s) + return result + + +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 + + expires = _ensure_tz_aware(challenge.expires_at) + if expires and expires < 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) + expires = _ensure_tz_aware(challenge.expires_at) + + if challenge.is_claimed: + status = "claimed" + elif challenge.is_cancelled: + status = "cancelled" + elif expires and expires < 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 699badf4..577fed6c 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", diff --git a/app/views/__init__.py b/app/views/__init__.py index 98100b1a..377a0309 100644 --- a/app/views/__init__.py +++ b/app/views/__init__.py @@ -10,6 +10,7 @@ from app.views.audit_logs import router as audit_logs_router from app.views.backup import router as backup_router from app.views.compliance import router as compliance_router from app.views.db_wizard import router as db_wizard_router +from app.views.devices import router as devices_router # Mobile devices dashboard from app.views.dropbox import router as dropbox_router from app.views.filemanager import router as filemanager_router @@ -26,6 +27,7 @@ from app.views.onedrive import router as onedrive_router from app.views.pipelines import router as pipelines_router # Processing pipelines from app.views.plans import router as plans_router # Admin Plan Designer from app.views.profile import router as profile_router # User self-service profile +from app.views.qr_login import router as qr_login_router # QR code mobile login from app.views.queue import router as queue_router from app.views.scheduled_jobs import router as scheduled_jobs_router # Scheduled batch jobs from app.views.search import router as search_router @@ -61,6 +63,7 @@ router.include_router(plans_router) # Admin Plan Designer router.include_router(onboarding_router) # User onboarding wizard router.include_router(pipelines_router) # Processing pipelines router.include_router(profile_router) # User self-service profile settings +router.include_router(qr_login_router) # QR code mobile login page router.include_router(imap_accounts_router) # Per-user IMAP ingestion accounts router.include_router(integrations_router) # Unified integrations dashboard router.include_router(notifications_router) # User notification dashboard @@ -68,4 +71,5 @@ router.include_router(scheduled_jobs_router) # Admin scheduled batch jobs router.include_router(audit_logs_router) # Comprehensive audit log viewer router.include_router(help_router) # Built-in help / How-To docs router.include_router(compliance_router) # Compliance templates dashboard +router.include_router(devices_router) # Mobile devices dashboard router.include_router(system_reset_router) # System reset / factory reset diff --git a/app/views/devices.py b/app/views/devices.py new file mode 100644 index 00000000..e8c346d1 --- /dev/null +++ b/app/views/devices.py @@ -0,0 +1,25 @@ +"""View route for the Devices management page. + +Renders the ``devices.html`` template where users can see their registered +mobile devices, mobile API tokens (created via the mobile SSO flow or QR +code login), and revoke access per-device. +""" + +import logging + +from fastapi import APIRouter, Request + +from app.views.base import require_login, templates + +logger = logging.getLogger(__name__) +router = APIRouter() + + +@router.get("/devices", include_in_schema=False) +@require_login +async def devices_page(request: Request): + """Render the Devices management page.""" + return templates.TemplateResponse( + "devices.html", + {"request": request, "page_title": "Devices"}, + ) diff --git a/app/views/qr_login.py b/app/views/qr_login.py new file mode 100644 index 00000000..aeb33f91 --- /dev/null +++ b/app/views/qr_login.py @@ -0,0 +1,26 @@ +"""View route for the QR code mobile login page. + +Route: + GET /qr-login — renders the QR login page (requires login) +""" + +from __future__ import annotations + +import logging + +from fastapi import Request + +from app.views.base import APIRouter, require_login, templates + +logger = logging.getLogger(__name__) +router = APIRouter() + + +@router.get("/qr-login", include_in_schema=False) +@require_login +async def qr_login_page(request: Request): + """Serve the QR code login page for mobile app authentication.""" + return templates.TemplateResponse( + "qr_login.html", + {"request": request}, + ) diff --git a/docs/AuthenticationSetup.md b/docs/AuthenticationSetup.md index 320e6191..2c9e15a9 100644 --- a/docs/AuthenticationSetup.md +++ b/docs/AuthenticationSetup.md @@ -154,6 +154,59 @@ DocuElevate can work with any OpenID Connect-compliant provider, not just Authen OAUTH_PROVIDER_NAME=Auth0 ``` +## Server-Side Session Management + +DocuElevate supports server-side session tracking. Every login creates a `UserSession` record that can be listed and revoked individually or all at once ("log off everywhere"). + +### Configuration + +| Variable | Description | Default | +|----------|-------------|---------| +| `SESSION_LIFETIME_DAYS` | Number of days before a session expires | `30` | +| `SESSION_LIFETIME_CUSTOM_DAYS` | Override for `SESSION_LIFETIME_DAYS` when set | — | + +### Managing Sessions + +Users can manage their active sessions from the **Profile → Security** section: + +- **View active sessions** — see browser, device, IP address, and last activity for each session. +- **Revoke a single session** — immediately invalidate one session. +- **Log off everywhere** — revoke all sessions (optionally keeping the current one) and all API tokens at once. + +Expired sessions are automatically cleaned up by a periodic background task. + +### API Endpoints + +| Method | Path | Description | +|--------|------|-------------| +| `GET` | `/api/sessions` | List the current user's active sessions | +| `DELETE` | `/api/sessions/{id}` | Revoke a single session | +| `POST` | `/api/sessions/revoke-all` | Revoke all sessions for the current user | + +## QR Code Login + +QR code login allows users to authenticate a mobile device by scanning a QR code displayed in the web UI, without manually entering credentials on the phone. + +### How It Works + +1. The authenticated web user opens the **QR Login** page and a challenge QR code is displayed. +2. The mobile app scans the QR code and calls the claim endpoint. +3. An API token is issued for the mobile device and the web UI is notified via polling. + +### Configuration + +| Variable | Description | Default | +|----------|-------------|---------| +| `QR_LOGIN_CHALLENGE_TTL_SECONDS` | How long a QR challenge is valid | `120` | + +### API Endpoints + +| Method | Path | Description | +|--------|------|-------------| +| `POST` | `/api/qr-auth/challenge` | Create a new QR login challenge | +| `GET` | `/api/qr-auth/challenge/{id}/status` | Poll the status of a challenge | +| `POST` | `/api/qr-auth/claim` | Claim a challenge from a mobile device | + ## Security Considerations 1. **Always use HTTPS** in production to protect authentication tokens and passwords diff --git a/docs/ConfigurationGuide.md b/docs/ConfigurationGuide.md index 595dd203..1874abfb 100644 --- a/docs/ConfigurationGuide.md +++ b/docs/ConfigurationGuide.md @@ -366,6 +366,9 @@ Credentials are encrypted at rest using Fernet encryption. |-------------------------|---------------------------------------------------------------| | `AUTH_ENABLED` | Enable or disable authentication (`true`/`false`). | | `SESSION_SECRET` | Secret key used to encrypt sessions and cookies (at least 32 chars). | +| `SESSION_LIFETIME_DAYS` | Number of days before a server-side session expires. Default: `30`. | +| `SESSION_LIFETIME_CUSTOM_DAYS` | Override for `SESSION_LIFETIME_DAYS` when set. | +| `QR_LOGIN_CHALLENGE_TTL_SECONDS` | How long a QR login challenge is valid (seconds). Default: `120`. | | `ADMIN_USERNAME` | Username for basic authentication (when not using OIDC). | | `ADMIN_PASSWORD` | Password for basic authentication (when not using OIDC). | | `ADMIN_GROUP_NAME` | Group name in OIDC claims that grants admin access. Default: `admin`. | diff --git a/frontend/static/js/common.js b/frontend/static/js/common.js index 67219a11..1ec9ae6b 100644 --- a/frontend/static/js/common.js +++ b/frontend/static/js/common.js @@ -215,6 +215,9 @@ function _makeMenuLink(href, iconClass, label, extraClasses = '') { linksDiv.appendChild( _makeMenuLink('/api-tokens', 'fas fa-key text-yellow-500', window.__i18n.apiTokens || 'API Tokens', 'text-gray-700') ); + linksDiv.appendChild( + _makeMenuLink('/devices', 'fas fa-mobile-alt text-blue-500', window.__i18n.devices || 'Devices', 'text-gray-700') + ); linksDiv.appendChild( _makeMenuLink('/shared-links', 'fas fa-share-alt text-blue-400', window.__i18n.sharedLinks || 'Shared Links', 'text-gray-700') ); @@ -311,6 +314,18 @@ function _makeMenuLink(href, iconClass, label, extraClasses = '') { tokensLink.appendChild(document.createTextNode(window.__i18n.apiTokens || 'API Tokens')); mobileAuthSection.appendChild(tokensLink); + // Devices link + const devicesLink = document.createElement('a'); + devicesLink.href = '/devices'; + devicesLink.className = + 'flex items-center px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50'; + const devicesIcon = document.createElement('i'); + devicesIcon.className = 'fas fa-mobile-alt mr-2 text-blue-500'; + devicesIcon.setAttribute('aria-hidden', 'true'); + devicesLink.appendChild(devicesIcon); + devicesLink.appendChild(document.createTextNode(window.__i18n.devices || 'Devices')); + mobileAuthSection.appendChild(devicesLink); + // Shared Links link const sharedLinksLink = document.createElement('a'); sharedLinksLink.href = '/shared-links'; diff --git a/frontend/templates/base.html b/frontend/templates/base.html index 4cf7c7e2..b7e47e44 100644 --- a/frontend/templates/base.html +++ b/frontend/templates/base.html @@ -570,6 +570,7 @@ profileSettings: {{ _("nav.profile_settings") | tojson }}, mySubscription: {{ _("nav.my_subscription") | tojson }}, apiTokens: {{ _("nav.api_tokens") | tojson }}, + devices: {{ _("nav.devices") | tojson }}, sharedLinks: {{ _("nav.shared_links") | tojson }}, signOut: {{ _("nav.sign_out") | tojson }}, logIn: {{ _("nav.login") | tojson }}, diff --git a/frontend/templates/devices.html b/frontend/templates/devices.html new file mode 100644 index 00000000..40e0a255 --- /dev/null +++ b/frontend/templates/devices.html @@ -0,0 +1,351 @@ +{% extends "base.html" %} + +{% block title %}{{ _("devices.page_title") }}{% endblock %} + +{% block content %} +
+ + +
+

+ + {{ _("devices.heading") }} +

+

+ {{ _("devices.intro") }} +

+
+ + +
+
+

+ {{ _("devices.mobile_tokens_heading") }} +

+

{{ _("devices.mobile_tokens_description") }}

+
+ + + + + + + + + + + + +
+ + +
+
+

+ {{ _("devices.registered_devices_heading") }} +

+

{{ _("devices.registered_devices_description") }}

+
+ + + + + + + + + + + + +
+ + +
+ + + {{ _("devices.qr_login_cta") }} + +
+ + + +
+ + +{% endblock %} diff --git a/frontend/templates/profile.html b/frontend/templates/profile.html index b3c47219..cc6144d7 100644 --- a/frontend/templates/profile.html +++ b/frontend/templates/profile.html @@ -343,9 +343,203 @@ + +
+

+ {{ _("sessions.security_heading") }} +

+

+ {{ _("sessions.security_subtitle") }} +

+ + +
+ + +
+ + +
+ +

{{ _("sessions.no_other_sessions") }}

+
+ + +
+ + + + {{ _("sessions.qr_login_link") }} + +
+ + + +
+ + + +{% endblock %} diff --git a/frontend/translations/en.json b/frontend/translations/en.json index 27d117d9..ebb67b2e 100644 --- a/frontend/translations/en.json +++ b/frontend/translations/en.json @@ -608,6 +608,37 @@ "dashboard.title": "Dashboard", "dashboard.total_files": "Total Files", "dashboard.welcome": "Welcome to DocuElevate", + "devices.col_created": "Connected", + "devices.col_device": "Device", + "devices.col_last_ip": "Last IP", + "devices.col_last_seen": "Last Seen", + "devices.col_last_used": "Last Used", + "devices.col_platform": "Platform", + "devices.col_push_token": "Push Token", + "devices.col_status": "Status", + "devices.col_token_prefix": "Token Prefix", + "devices.confirm_deactivate_device": "Remove this device? It will stop receiving push notifications.", + "devices.confirm_revoke_token": "Revoke access for this device? It will need to log in again.", + "devices.deactivate_device": "Remove", + "devices.device_removed_success": "Device removed successfully.", + "devices.heading": "Mobile Devices", + "devices.intro": "Manage your mobile app connections and registered devices. You can revoke access for individual devices here.", + "devices.loading": "Loading devices…", + "devices.mobile_tokens_description": "These tokens were created when you logged in via the mobile app or scanned a QR code. Revoking a token will sign the device out.", + "devices.mobile_tokens_heading": "Mobile App Tokens", + "devices.no_devices": "No registered devices", + "devices.no_devices_help": "Install the DocuElevate mobile app and log in to register a device for push notifications.", + "devices.no_mobile_tokens": "No mobile app tokens", + "devices.no_mobile_tokens_help": "Log in via the mobile app or scan a QR code to create a mobile token.", + "devices.page_title": "Devices – DocuElevate", + "devices.qr_login_cta": "Connect a new device via QR code", + "devices.registered_devices_description": "Devices registered for push notifications from the DocuElevate mobile app.", + "devices.registered_devices_heading": "Registered Devices", + "devices.revoke_token": "Revoke", + "devices.status_active": "Active", + "devices.status_inactive": "Inactive", + "devices.status_revoked": "Revoked", + "devices.token_revoked_success": "Device token revoked successfully.", "duplicates.file_id_label": "File ID", "duplicates.file_id_placeholder": "e.g. 42", "duplicates.find_btn": "Find", @@ -1152,6 +1183,7 @@ "nav.dark_mode": "Dark Mode", "nav.dashboard": "Dashboard", "nav.developer_docs": "Developer Docs", + "nav.devices": "Devices", "nav.duplicates": "Duplicates", "nav.file_manager": "File Manager", "nav.files": "Files", @@ -1452,6 +1484,20 @@ "profile.theme_system": "System Default", "profile.update_password": "Update Password", "profile.updating": "Updating…", + "qr_login.claimed_device": "Device: {device_name}", + "qr_login.claimed_message": "QR code login successful! Your mobile device is now connected.", + "qr_login.description": "Scan this QR code with the DocuElevate mobile app to log in instantly.", + "qr_login.expired_message": "This QR code has expired. Please generate a new one.", + "qr_login.generate_new": "Generate New QR Code", + "qr_login.heading": "Mobile App QR Login", + "qr_login.how_it_works": "How it works", + "qr_login.page_title": "QR Code Login – DocuElevate", + "qr_login.pending_message": "Waiting for mobile app to scan…", + "qr_login.step_1": "Open the DocuElevate app on your phone", + "qr_login.step_2": "Tap \"Scan QR Code\" on the login screen", + "qr_login.step_3": "Point your camera at this QR code", + "qr_login.subtitle": "Log in to the mobile app by scanning a QR code from this page.", + "qr_login.time_remaining": "Expires in {seconds} seconds", "queue.active_tasks": "Active Tasks", "queue.auto_refresh_1": "Auto-refreshes every", "queue.auto_refresh_2": "seconds", @@ -1513,6 +1559,25 @@ "search.saved_label": "Saved Searches", "search.saved_loading": "Loading...", "search.title": "Search Documents", + "sessions.active_sessions": "Active Sessions", + "sessions.confirm_revoke_all": "This will log you out of all other devices and browsers, and revoke all API tokens. Continue?", + "sessions.confirm_revoke_one": "Are you sure you want to end this session?", + "sessions.current_session": "This device", + "sessions.device_info": "Device", + "sessions.expires": "Expires", + "sessions.ip_address": "IP Address", + "sessions.last_active": "Last active", + "sessions.log_off_everywhere": "Log Off All Other Sessions", + "sessions.log_off_everywhere_desc": "End all other browser sessions and revoke all API tokens. Your current session will remain active.", + "sessions.no_other_sessions": "No other active sessions found.", + "sessions.qr_login_link": "Log in on mobile via QR code", + "sessions.revoke": "End Session", + "sessions.revoked_all_success": "All other sessions have been ended.", + "sessions.revoked_success": "Session ended successfully.", + "sessions.security_heading": "Security & Sessions", + "sessions.security_subtitle": "Manage your active sessions across devices and browsers.", + "sessions.session_lifetime": "Session lifetime: {days} days", + "sessions.started": "Started", "settings.audit_log_btn": "Audit Log", "settings.autocomplete_hint": "Type to search known values, or enter any custom value.", "settings.autocomplete_no_matches": "No matches — you can still type a custom value", diff --git a/migrations/versions/037_add_user_sessions_and_qr_challenges.py b/migrations/versions/037_add_user_sessions_and_qr_challenges.py new file mode 100644 index 00000000..9610f56d --- /dev/null +++ b/migrations/versions/037_add_user_sessions_and_qr_challenges.py @@ -0,0 +1,72 @@ +"""Add user_sessions and qr_login_challenges tables. + +Adds server-side session tracking (user_sessions) for the "log off +everywhere" feature and per-session revocation, and QR login challenges +(qr_login_challenges) for secure mobile app authentication via QR code. + +Revision ID: 037_add_user_sessions_and_qr_challenges +Revises: 036_add_document_translation_fields +Create Date: 2026-03-16 +""" + +from typing import Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "037_add_user_sessions_and_qr_challenges" +down_revision: Union[str, None] = "036_add_document_translation_fields" +depends_on: Union[str, None] = None + + +def upgrade() -> None: + """Create user_sessions and qr_login_challenges tables.""" + conn = op.get_bind() + inspector = sa.inspect(conn) + existing_tables = set(inspector.get_table_names()) + + if "user_sessions" not in existing_tables: + op.create_table( + "user_sessions", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("session_token", sa.String(128), nullable=False, unique=True, index=True), + sa.Column("user_id", sa.String(), nullable=False, index=True), + sa.Column("ip_address", sa.String(45), nullable=True), + sa.Column("user_agent", sa.String(512), nullable=True), + sa.Column("device_info", sa.String(255), nullable=True), + sa.Column("is_revoked", sa.Boolean(), nullable=False, server_default="0"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("last_active_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True), + ) + + if "qr_login_challenges" not in existing_tables: + op.create_table( + "qr_login_challenges", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("challenge_token", sa.String(128), nullable=False, unique=True, index=True), + sa.Column("user_id", sa.String(), nullable=False, index=True), + sa.Column("is_claimed", sa.Boolean(), nullable=False, server_default="0"), + sa.Column("is_cancelled", sa.Boolean(), nullable=False, server_default="0"), + sa.Column("created_by_ip", sa.String(45), nullable=True), + sa.Column("claimed_by_ip", sa.String(45), nullable=True), + sa.Column("device_name", sa.String(255), nullable=True), + sa.Column("issued_token_id", sa.Integer(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("claimed_at", sa.DateTime(timezone=True), nullable=True), + ) + + +def downgrade() -> None: + """Drop user_sessions and qr_login_challenges tables.""" + conn = op.get_bind() + inspector = sa.inspect(conn) + existing_tables = set(inspector.get_table_names()) + + if "qr_login_challenges" in existing_tables: + op.drop_table("qr_login_challenges") + + if "user_sessions" in existing_tables: + op.drop_table("user_sessions") diff --git a/migrations/versions/037_add_classification_rules.py b/migrations/versions/038_add_classification_rules.py similarity index 88% rename from migrations/versions/037_add_classification_rules.py rename to migrations/versions/038_add_classification_rules.py index 08cade2c..f5eb3888 100644 --- a/migrations/versions/037_add_classification_rules.py +++ b/migrations/versions/038_add_classification_rules.py @@ -1,8 +1,8 @@ """Add classification_rules table for custom document classification rules. -Revision ID: 037_add_classification_rules -Revises: 036_add_document_translation_fields -Create Date: 2026-03-09 +Revision ID: 038_add_classification_rules +Revises: 037_add_user_sessions_and_qr_challenges +Create Date: 2026-03-17 """ from typing import Union @@ -10,8 +10,8 @@ from typing import Union import sqlalchemy as sa from alembic import op -revision: str = "037_add_classification_rules" -down_revision: Union[str, None] = "036_add_document_translation_fields" +revision: str = "038_add_classification_rules" +down_revision: Union[str, None] = "037_add_user_sessions_and_qr_challenges" depends_on: Union[str, None] = None diff --git a/mobile/src/context/AuthContext.tsx b/mobile/src/context/AuthContext.tsx index 4bbb66e7..e4e1ab84 100644 --- a/mobile/src/context/AuthContext.tsx +++ b/mobile/src/context/AuthContext.tsx @@ -38,6 +38,7 @@ export interface AuthState { user: WhoAmIResponse | null; baseUrl: string; signIn: (serverUrl: string) => Promise; + signInWithQR: (serverUrl: string, challengeToken: string) => Promise; signOut: () => Promise; setToken: (token: string) => Promise; } @@ -52,6 +53,7 @@ const AuthContext = createContext({ user: null, baseUrl: "", signIn: async () => {}, + signInWithQR: async () => {}, signOut: async () => {}, setToken: async () => {}, }); @@ -143,6 +145,19 @@ export function AuthProvider({ children }: { children: React.ReactNode }) { [setToken] ); + const signInWithQR = useCallback( + async (serverUrl: string, challengeToken: string) => { + const cleanUrl = serverUrl.replace(/\/$/, ""); + await api.init(cleanUrl); + setBaseUrl(cleanUrl); + + const deviceInfo = await _getDeviceName(); + const resp = await api.claimQRChallenge(challengeToken, deviceInfo); + await setToken(resp.token); + }, + [setToken] + ); + const signOut = useCallback(async () => { await SecureStore.deleteItemAsync(SECURE_STORE_API_TOKEN_KEY); await SecureStore.deleteItemAsync(SECURE_STORE_OWNER_ID_KEY); @@ -158,6 +173,7 @@ export function AuthProvider({ children }: { children: React.ReactNode }) { user, baseUrl, signIn, + signInWithQR, signOut, setToken, }} diff --git a/mobile/src/screens/LoginScreen.tsx b/mobile/src/screens/LoginScreen.tsx index f278b46e..cc2b3133 100644 --- a/mobile/src/screens/LoginScreen.tsx +++ b/mobile/src/screens/LoginScreen.tsx @@ -1,13 +1,16 @@ /** - * LoginScreen – server URL entry and SSO sign-in. + * LoginScreen – server URL entry, SSO sign-in, and QR code login. * - * Renders a server URL input and a "Sign in with SSO" button that opens the - * DocuElevate web login page in the system browser. On success the - * AuthContext stores the API token and navigates to the main app. + * Renders a server URL input, a "Sign in with SSO" button that opens the + * DocuElevate web login page in the system browser, and a "Scan QR Code" + * button that opens the device camera to scan a QR code generated from the + * web interface. On success the AuthContext stores the API token and + * navigates to the main app. */ +import * as Linking from "expo-linking"; import { useRouter } from "expo-router"; -import React, { useState } from "react"; +import React, { useCallback, useEffect, useState } from "react"; import { ActivityIndicator, Alert, @@ -23,10 +26,46 @@ import { import { useAuth } from "../context/AuthContext"; export default function LoginScreen() { - const { signIn } = useAuth(); + const { signIn, signInWithQR } = useAuth(); const router = useRouter(); const [serverUrl, setServerUrl] = useState(""); const [loading, setLoading] = useState(false); + const [qrLoading, setQrLoading] = useState(false); + + // Handle incoming deep links for QR login (docuelevate://qr-login?token=...&server=...) + const handleDeepLink = useCallback( + async (event: { url: string }) => { + try { + const url = new URL(event.url); + if (url.hostname === "qr-login" || url.pathname === "/qr-login") { + const token = url.searchParams.get("token"); + const server = url.searchParams.get("server"); + if (token && server) { + setQrLoading(true); + await signInWithQR(server, token); + } + } + } catch (err: unknown) { + const message = err instanceof Error ? err.message : "QR login failed"; + Alert.alert("QR Login Failed", message); + } finally { + setQrLoading(false); + } + }, + [signInWithQR] + ); + + useEffect(() => { + // Listen for incoming deep links + const subscription = Linking.addEventListener("url", handleDeepLink); + + // Check if the app was opened via a deep link + Linking.getInitialURL().then((url) => { + if (url) handleDeepLink({ url }); + }); + + return () => subscription.remove(); + }, [handleDeepLink]); async function handleSignIn() { const url = serverUrl.trim(); @@ -85,7 +124,7 @@ export default function LoginScreen() { @@ -96,8 +135,33 @@ export default function LoginScreen() { )} + + + or + + + + { + Alert.alert( + "Scan QR Code", + "Open the DocuElevate web app on your computer, go to Profile → Security & Sessions → \"Log in on mobile via QR code\", and scan the QR code shown there.\n\nThe app will automatically detect the QR code when scanned with your device camera." + ); + }} + disabled={loading || qrLoading} + accessibilityRole="button" + accessibilityLabel="Sign in with QR code" + > + {qrLoading ? ( + + ) : ( + 📱 Scan QR Code to Login + )} + + - You will be redirected to your organisation's sign-in page. + Sign in via SSO or scan a QR code from the web app. { + return this.request("POST", "/api/qr-auth/claim", { + body: { challenge_token: challengeToken, device_name: deviceName }, + }); + } + /** Return profile information for the authenticated user. */ async whoAmI(): Promise { return this.request("GET", "/api/mobile/whoami"); diff --git a/tests/test_devices_page.py b/tests/test_devices_page.py new file mode 100644 index 00000000..106cba0c --- /dev/null +++ b/tests/test_devices_page.py @@ -0,0 +1,201 @@ +"""Tests for the Devices page and mobile token filtering (app/api/api_tokens.py mobile endpoint). + +These tests validate: +- ``GET /api/api-tokens/mobile`` returns only mobile tokens +- ``GET /api/api-tokens/`` excludes mobile tokens +- ``GET /devices`` renders the devices page +""" + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from app.database import Base, get_db +from app.models import ApiToken + +# --------------------------------------------------------------------------- +# Test data +# --------------------------------------------------------------------------- + +_OWNER = "devices_user@example.com" +_OTHER_OWNER = "other_devices@example.com" + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def dev_engine(): + """In-memory SQLite engine.""" + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(bind=engine) + yield engine + Base.metadata.drop_all(bind=engine) + + +@pytest.fixture() +def dev_session(dev_engine): + """DB session scoped to one test.""" + Session = sessionmaker(bind=dev_engine) + session = Session() + yield session + session.close() + + +def _make_client(dev_engine, owner_id: str = _OWNER) -> TestClient: + """Return a TestClient with *owner_id* injected as the authenticated user.""" + from app.api.api_tokens import _get_owner_id + from app.main import app + + Session = sessionmaker(bind=dev_engine) + + def _override_get_db(): + session = Session() + try: + yield session + finally: + session.close() + + def _override_owner(): + return owner_id + + app.dependency_overrides[get_db] = _override_get_db + app.dependency_overrides[_get_owner_id] = _override_owner + + client = TestClient(app, base_url="http://localhost", raise_server_exceptions=False) + return client + + +def _cleanup(app): + """Remove dependency overrides after test.""" + app.dependency_overrides.clear() + + +def _seed_tokens(session, owner_id: str = _OWNER): + """Create a mix of regular and mobile tokens for testing.""" + from app.api.api_tokens import generate_api_token, hash_token + + tokens = [] + # Regular API tokens + for name in ["CI Pipeline", "Webhook Upload"]: + pt = generate_api_token() + t = ApiToken(owner_id=owner_id, name=name, token_hash=hash_token(pt), token_prefix=pt[:12]) + session.add(t) + tokens.append(t) + + # Mobile tokens (various naming patterns) + for name in [ + "Mobile App – iPhone 15 Pro", + "Mobile App (QR) – Christian's iPad", + "Mobile App", + ]: + pt = generate_api_token() + t = ApiToken(owner_id=owner_id, name=name, token_hash=hash_token(pt), token_prefix=pt[:12]) + session.add(t) + tokens.append(t) + + session.commit() + return tokens + + +# --------------------------------------------------------------------------- +# Tests – Mobile Token Filtering +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestMobileTokenFiltering: + """Tests for GET /api/api-tokens/mobile and filtering from GET /api/api-tokens/.""" + + def test_list_mobile_tokens_returns_only_mobile(self, dev_engine, dev_session): + """GET /api/api-tokens/mobile should only return tokens starting with 'Mobile App'.""" + _seed_tokens(dev_session) + client = _make_client(dev_engine) + try: + res = client.get("/api/api-tokens/mobile") + assert res.status_code == 200 + data = res.json() + assert len(data) == 3 + for t in data: + assert t["name"].startswith("Mobile App") + finally: + _cleanup(client.app) + + def test_list_regular_tokens_excludes_mobile(self, dev_engine, dev_session): + """GET /api/api-tokens/ should NOT return tokens starting with 'Mobile App'.""" + _seed_tokens(dev_session) + client = _make_client(dev_engine) + try: + res = client.get("/api/api-tokens/") + assert res.status_code == 200 + data = res.json() + assert len(data) == 2 + for t in data: + assert not t["name"].startswith("Mobile App") + finally: + _cleanup(client.app) + + def test_list_mobile_tokens_empty(self, dev_engine): + """GET /api/api-tokens/mobile returns [] when no mobile tokens exist.""" + client = _make_client(dev_engine) + try: + res = client.get("/api/api-tokens/mobile") + assert res.status_code == 200 + assert res.json() == [] + finally: + _cleanup(client.app) + + def test_list_mobile_tokens_isolation(self, dev_engine, dev_session): + """Mobile tokens for other users should not appear.""" + _seed_tokens(dev_session, owner_id=_OTHER_OWNER) + client = _make_client(dev_engine, owner_id=_OWNER) + try: + res = client.get("/api/api-tokens/mobile") + assert res.status_code == 200 + assert res.json() == [] + finally: + _cleanup(client.app) + + def test_mobile_token_revoke_via_api_tokens_endpoint(self, dev_engine, dev_session): + """Mobile tokens can still be revoked via DELETE /api/api-tokens/{id}.""" + tokens = _seed_tokens(dev_session) + mobile_token = next(t for t in tokens if t.name.startswith("Mobile App")) + client = _make_client(dev_engine) + try: + res = client.delete(f"/api/api-tokens/{mobile_token.id}") + assert res.status_code == 200 + # Verify it's gone from mobile list + res2 = client.get("/api/api-tokens/mobile") + active_names = [t["name"] for t in res2.json() if t["is_active"]] + assert mobile_token.name not in active_names + finally: + _cleanup(client.app) + + +# --------------------------------------------------------------------------- +# Tests – Devices Page View +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestDevicesPageView: + """Tests for GET /devices page rendering.""" + + def test_devices_page_renders(self, dev_engine): + """GET /devices should return 200 with the devices template.""" + from app.views.devices import router as _ # noqa: F401 – ensures route is registered + + client = _make_client(dev_engine) + try: + res = client.get("/devices") + assert res.status_code == 200 + assert "devices.heading" in res.text or "Mobile Devices" in res.text + finally: + _cleanup(client.app) diff --git a/tests/test_session_management.py b/tests/test_session_management.py new file mode 100644 index 00000000..039aca46 --- /dev/null +++ b/tests/test_session_management.py @@ -0,0 +1,670 @@ +"""Tests for server-side session management and QR code login. + +Covers: +* Session creation, validation, revocation, and cleanup +* "Log off everywhere" (revoke all sessions) +* QR login challenge creation, validation, claiming, and status polling +* Session management API endpoints (list, revoke, revoke-all) +* QR auth API endpoints (challenge, status, claim) +* Device info parsing from User-Agent strings +""" + +from __future__ import annotations + +import secrets +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker + +from app.database import Base +from app.models import ApiToken, QRLoginChallenge, UserSession + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def db_session(): + """Provide an in-memory SQLite session with all tables created.""" + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + TestSession = sessionmaker(bind=engine) + session = TestSession() + yield session + session.close() + Base.metadata.drop_all(engine) + + +@pytest.fixture() +def sample_user_id(): + return "user@example.com" + + +# --------------------------------------------------------------------------- +# Model Tests +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestUserSessionModel: + """Tests for the UserSession ORM model.""" + + def test_create_user_session(self, db_session: Session, sample_user_id: str): + """Test creating a UserSession record.""" + now = datetime.now(timezone.utc) + session = UserSession( + session_token=secrets.token_urlsafe(64), + user_id=sample_user_id, + ip_address="192.168.1.1", + user_agent="Mozilla/5.0", + device_info="Chrome on macOS", + expires_at=now + timedelta(days=30), + ) + db_session.add(session) + db_session.commit() + + assert session.id is not None + assert session.user_id == sample_user_id + assert session.is_revoked is False + assert session.device_info == "Chrome on macOS" + + def test_session_default_values(self, db_session: Session, sample_user_id: str): + """Test that default values are set correctly.""" + session = UserSession( + session_token="test_token_123", + user_id=sample_user_id, + expires_at=datetime.now(timezone.utc) + timedelta(days=30), + ) + db_session.add(session) + db_session.commit() + + assert session.is_revoked is False + assert session.revoked_at is None + + +@pytest.mark.unit +class TestQRLoginChallengeModel: + """Tests for the QRLoginChallenge ORM model.""" + + def test_create_challenge(self, db_session: Session, sample_user_id: str): + """Test creating a QRLoginChallenge record.""" + challenge = QRLoginChallenge( + challenge_token=secrets.token_urlsafe(64), + user_id=sample_user_id, + created_by_ip="10.0.0.1", + expires_at=datetime.now(timezone.utc) + timedelta(seconds=120), + ) + db_session.add(challenge) + db_session.commit() + + assert challenge.id is not None + assert challenge.is_claimed is False + assert challenge.is_cancelled is False + + def test_challenge_default_values(self, db_session: Session, sample_user_id: str): + """Test that QRLoginChallenge defaults are correct.""" + challenge = QRLoginChallenge( + challenge_token="challenge_test_123", + user_id=sample_user_id, + expires_at=datetime.now(timezone.utc) + timedelta(seconds=120), + ) + db_session.add(challenge) + db_session.commit() + + assert challenge.is_claimed is False + assert challenge.is_cancelled is False + assert challenge.claimed_at is None + assert challenge.device_name is None + + +# --------------------------------------------------------------------------- +# Session Manager Tests +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestSessionManager: + """Tests for app/utils/session_manager.py functions.""" + + @patch("app.utils.session_manager.settings") + def test_get_session_lifetime_days_default(self, mock_settings): + """Test default session lifetime.""" + from app.utils.session_manager import get_session_lifetime_days + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + assert get_session_lifetime_days() == 30 + + @patch("app.utils.session_manager.settings") + def test_get_session_lifetime_days_custom(self, mock_settings): + """Test custom session lifetime overrides default.""" + from app.utils.session_manager import get_session_lifetime_days + + mock_settings.session_lifetime_custom_days = 90 + mock_settings.session_lifetime_days = 30 + assert get_session_lifetime_days() == 90 + + @patch("app.utils.session_manager.settings") + def test_get_session_lifetime_days_minimum(self, mock_settings): + """Test session lifetime has a minimum of 1 day.""" + from app.utils.session_manager import get_session_lifetime_days + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 0 + assert get_session_lifetime_days() == 1 + + @patch("app.utils.session_manager.settings") + def test_get_session_max_age_seconds(self, mock_settings): + """Test session max age in seconds.""" + from app.utils.session_manager import get_session_max_age_seconds + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + assert get_session_max_age_seconds() == 30 * 86400 + + @patch("app.utils.session_manager.settings") + def test_create_session(self, mock_settings, db_session: Session, sample_user_id: str): + """Test creating a server-side session.""" + from app.utils.session_manager import create_session + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + mock_settings.qr_login_challenge_ttl_seconds = 120 + + user_session = create_session( + db_session, + user_id=sample_user_id, + ip_address="10.0.0.1", + user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Chrome/120.0", + ) + + assert user_session.id is not None + assert user_session.user_id == sample_user_id + assert user_session.ip_address == "10.0.0.1" + assert user_session.session_token is not None + assert len(user_session.session_token) > 32 + assert user_session.is_revoked is False + assert user_session.device_info is not None + + @patch("app.utils.session_manager.settings") + def test_validate_session_valid(self, mock_settings, db_session: Session, sample_user_id: str): + """Test validating a valid session.""" + from app.utils.session_manager import create_session, validate_session + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + + user_session = create_session(db_session, user_id=sample_user_id) + result = validate_session(db_session, user_session.session_token) + assert result is not None + assert result.id == user_session.id + + @patch("app.utils.session_manager.settings") + def test_validate_session_revoked(self, mock_settings, db_session: Session, sample_user_id: str): + """Test that revoked sessions are rejected.""" + from app.utils.session_manager import create_session, validate_session + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + + user_session = create_session(db_session, user_id=sample_user_id) + user_session.is_revoked = True + db_session.commit() + + result = validate_session(db_session, user_session.session_token) + assert result is None + + @patch("app.utils.session_manager.settings") + def test_validate_session_expired(self, mock_settings, db_session: Session, sample_user_id: str): + """Test that expired sessions are rejected.""" + from app.utils.session_manager import create_session, validate_session + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + + user_session = create_session(db_session, user_id=sample_user_id) + user_session.expires_at = datetime.now(timezone.utc) - timedelta(hours=1) + db_session.commit() + + result = validate_session(db_session, user_session.session_token) + assert result is None + + def test_validate_session_empty_token(self, db_session: Session): + """Test that empty token returns None.""" + from app.utils.session_manager import validate_session + + assert validate_session(db_session, "") is None + assert validate_session(db_session, None) is None + + def test_validate_session_nonexistent_token(self, db_session: Session): + """Test that nonexistent token returns None.""" + from app.utils.session_manager import validate_session + + assert validate_session(db_session, "nonexistent_token_xyz") is None + + @patch("app.utils.session_manager.settings") + def test_revoke_session(self, mock_settings, db_session: Session, sample_user_id: str): + """Test revoking a single session.""" + from app.utils.session_manager import create_session, revoke_session, validate_session + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + + user_session = create_session(db_session, user_id=sample_user_id) + assert revoke_session(db_session, user_session.id, sample_user_id) is True + + # Session should now be invalid + assert validate_session(db_session, user_session.session_token) is None + + @patch("app.utils.session_manager.settings") + def test_revoke_session_wrong_user(self, mock_settings, db_session: Session, sample_user_id: str): + """Test that a user cannot revoke another user's session.""" + from app.utils.session_manager import create_session, revoke_session + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + + user_session = create_session(db_session, user_id=sample_user_id) + assert revoke_session(db_session, user_session.id, "other_user@example.com") is False + + @patch("app.utils.session_manager.settings") + def test_revoke_all_sessions(self, mock_settings, db_session: Session, sample_user_id: str): + """Test revoking all sessions for a user.""" + from app.utils.session_manager import create_session, list_user_sessions, revoke_all_sessions + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + + s1 = create_session(db_session, user_id=sample_user_id) + s2 = create_session(db_session, user_id=sample_user_id) + s3 = create_session(db_session, user_id=sample_user_id) + + count = revoke_all_sessions(db_session, sample_user_id, revoke_api_tokens=False) + assert count == 3 + + # All sessions should be revoked + active = list_user_sessions(db_session, sample_user_id) + assert len(active) == 0 + + @patch("app.utils.session_manager.settings") + def test_revoke_all_except_current(self, mock_settings, db_session: Session, sample_user_id: str): + """Test revoking all sessions except the current one.""" + from app.utils.session_manager import create_session, list_user_sessions, revoke_all_sessions + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + + s1 = create_session(db_session, user_id=sample_user_id) + s2 = create_session(db_session, user_id=sample_user_id) + s3 = create_session(db_session, user_id=sample_user_id) + + count = revoke_all_sessions( + db_session, + sample_user_id, + except_session_id=s1.id, + revoke_api_tokens=False, + ) + assert count == 2 + + active = list_user_sessions(db_session, sample_user_id) + assert len(active) == 1 + assert active[0].id == s1.id + + @patch("app.utils.session_manager.settings") + def test_revoke_all_includes_api_tokens(self, mock_settings, db_session: Session, sample_user_id: str): + """Test that revoke-all also revokes API tokens.""" + from app.utils.session_manager import create_session, revoke_all_sessions + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + + create_session(db_session, user_id=sample_user_id) + + # Create an API token + token = ApiToken( + owner_id=sample_user_id, + name="Test Token", + token_hash="abc123hash", + token_prefix="de_abc12345", + ) + db_session.add(token) + db_session.commit() + + revoke_all_sessions(db_session, sample_user_id, revoke_api_tokens=True) + + db_session.refresh(token) + assert token.is_active is False + + @patch("app.utils.session_manager.settings") + def test_list_user_sessions(self, mock_settings, db_session: Session, sample_user_id: str): + """Test listing active sessions for a user.""" + from app.utils.session_manager import create_session, list_user_sessions + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + + create_session(db_session, user_id=sample_user_id) + create_session(db_session, user_id=sample_user_id) + create_session(db_session, user_id="other@example.com") + + sessions = list_user_sessions(db_session, sample_user_id) + assert len(sessions) == 2 + + @patch("app.utils.session_manager.settings") + def test_cleanup_expired_sessions(self, mock_settings, db_session: Session, sample_user_id: str): + """Test cleaning up expired sessions.""" + from app.utils.session_manager import cleanup_expired_sessions, create_session + + mock_settings.session_lifetime_custom_days = None + mock_settings.session_lifetime_days = 30 + + # Create a session that expired 10 days ago + session = create_session(db_session, user_id=sample_user_id) + session.expires_at = datetime.now(timezone.utc) - timedelta(days=10) + db_session.commit() + + count = cleanup_expired_sessions(db_session) + assert count == 1 + + +# --------------------------------------------------------------------------- +# QR Login Tests +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestQRLogin: + """Tests for QR login challenge/claim flow.""" + + @patch("app.utils.session_manager.settings") + def test_create_qr_challenge(self, mock_settings, db_session: Session, sample_user_id: str): + """Test creating a QR login challenge.""" + from app.utils.session_manager import create_qr_challenge + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id, ip_address="10.0.0.1") + + assert challenge.id is not None + assert challenge.user_id == sample_user_id + assert challenge.challenge_token is not None + assert len(challenge.challenge_token) > 32 + assert challenge.is_claimed is False + assert challenge.created_by_ip == "10.0.0.1" + # SQLite returns naive datetimes; normalise before comparison + expires = challenge.expires_at + if expires.tzinfo is None: + expires = expires.replace(tzinfo=timezone.utc) + assert expires > datetime.now(timezone.utc) + + @patch("app.utils.session_manager.settings") + def test_validate_qr_challenge_valid(self, mock_settings, db_session: Session, sample_user_id: str): + """Test validating a valid QR challenge.""" + from app.utils.session_manager import create_qr_challenge, validate_qr_challenge + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + result = validate_qr_challenge(db_session, challenge.challenge_token) + assert result is not None + assert result.id == challenge.id + + @patch("app.utils.session_manager.settings") + def test_validate_qr_challenge_expired(self, mock_settings, db_session: Session, sample_user_id: str): + """Test that expired challenges are rejected.""" + from app.utils.session_manager import create_qr_challenge, validate_qr_challenge + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + challenge.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1) + db_session.commit() + + result = validate_qr_challenge(db_session, challenge.challenge_token) + assert result is None + + @patch("app.utils.session_manager.settings") + def test_validate_qr_challenge_claimed(self, mock_settings, db_session: Session, sample_user_id: str): + """Test that claimed challenges are rejected (replay protection).""" + from app.utils.session_manager import create_qr_challenge, validate_qr_challenge + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + challenge.is_claimed = True + db_session.commit() + + result = validate_qr_challenge(db_session, challenge.challenge_token) + assert result is None + + @patch("app.utils.session_manager.settings") + def test_validate_qr_challenge_cancelled(self, mock_settings, db_session: Session, sample_user_id: str): + """Test that cancelled challenges are rejected.""" + from app.utils.session_manager import create_qr_challenge, validate_qr_challenge + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + challenge.is_cancelled = True + db_session.commit() + + result = validate_qr_challenge(db_session, challenge.challenge_token) + assert result is None + + def test_validate_qr_challenge_empty(self, db_session: Session): + """Test that empty challenge token returns None.""" + from app.utils.session_manager import validate_qr_challenge + + assert validate_qr_challenge(db_session, "") is None + assert validate_qr_challenge(db_session, None) is None + + @patch("app.utils.session_manager.settings") + def test_claim_qr_challenge_success(self, mock_settings, db_session: Session, sample_user_id: str): + """Test successfully claiming a QR challenge.""" + from app.utils.session_manager import claim_qr_challenge, create_qr_challenge + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + result = claim_qr_challenge( + db_session, + challenge.challenge_token, + device_name="Christian's iPhone 15 Pro", + ip_address="192.168.1.100", + ) + + assert result is not None + assert result["token"].startswith("de_") + assert result["token_id"] is not None + assert result["owner_id"] == sample_user_id + assert "QR" in result["name"] + + # Challenge should now be claimed + db_session.refresh(challenge) + assert challenge.is_claimed is True + assert challenge.claimed_by_ip == "192.168.1.100" + assert challenge.device_name == "Christian's iPhone 15 Pro" + + @patch("app.utils.session_manager.settings") + def test_claim_qr_challenge_replay_protection(self, mock_settings, db_session: Session, sample_user_id: str): + """Test that a claimed challenge cannot be claimed again.""" + from app.utils.session_manager import claim_qr_challenge, create_qr_challenge + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + + # First claim succeeds + result1 = claim_qr_challenge(db_session, challenge.challenge_token) + assert result1 is not None + + # Second claim fails (replay protection) + result2 = claim_qr_challenge(db_session, challenge.challenge_token) + assert result2 is None + + @patch("app.utils.session_manager.settings") + def test_claim_qr_challenge_expired(self, mock_settings, db_session: Session, sample_user_id: str): + """Test that expired challenges cannot be claimed.""" + from app.utils.session_manager import claim_qr_challenge, create_qr_challenge + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + challenge.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1) + db_session.commit() + + result = claim_qr_challenge(db_session, challenge.challenge_token) + assert result is None + + def test_claim_qr_challenge_invalid_token(self, db_session: Session): + """Test claiming with an invalid token.""" + from app.utils.session_manager import claim_qr_challenge + + result = claim_qr_challenge(db_session, "nonexistent_token_xyz") + assert result is None + + @patch("app.utils.session_manager.settings") + def test_get_challenge_status_pending(self, mock_settings, db_session: Session, sample_user_id: str): + """Test getting status of a pending challenge.""" + from app.utils.session_manager import create_qr_challenge, get_challenge_status + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + status = get_challenge_status(db_session, challenge.id, sample_user_id) + + assert status is not None + assert status["status"] == "pending" + + @patch("app.utils.session_manager.settings") + def test_get_challenge_status_claimed(self, mock_settings, db_session: Session, sample_user_id: str): + """Test getting status of a claimed challenge.""" + from app.utils.session_manager import claim_qr_challenge, create_qr_challenge, get_challenge_status + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + claim_qr_challenge(db_session, challenge.challenge_token, device_name="Test Device") + + status = get_challenge_status(db_session, challenge.id, sample_user_id) + assert status is not None + assert status["status"] == "claimed" + assert status["device_name"] == "Test Device" + + @patch("app.utils.session_manager.settings") + def test_get_challenge_status_expired(self, mock_settings, db_session: Session, sample_user_id: str): + """Test getting status of an expired challenge.""" + from app.utils.session_manager import create_qr_challenge, get_challenge_status + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + challenge.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1) + db_session.commit() + + status = get_challenge_status(db_session, challenge.id, sample_user_id) + assert status["status"] == "expired" + + @patch("app.utils.session_manager.settings") + def test_get_challenge_status_wrong_user(self, mock_settings, db_session: Session, sample_user_id: str): + """Test that a user cannot see another user's challenge status.""" + from app.utils.session_manager import create_qr_challenge, get_challenge_status + + mock_settings.qr_login_challenge_ttl_seconds = 120 + + challenge = create_qr_challenge(db_session, sample_user_id) + status = get_challenge_status(db_session, challenge.id, "other@example.com") + assert status is None + + +# --------------------------------------------------------------------------- +# Device Info Parsing Tests +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestDeviceInfoParsing: + """Tests for User-Agent parsing.""" + + def test_chrome_macos(self): + from app.utils.session_manager import _parse_device_info + + ua = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + result = _parse_device_info(ua) + assert "Chrome" in result + assert "macOS" in result + + def test_safari_iphone(self): + from app.utils.session_manager import _parse_device_info + + ua = "Mozilla/5.0 (iPhone; CPU iPhone OS 17_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Mobile/15E148 Safari/604.1" + result = _parse_device_info(ua) + assert "Safari" in result + assert "iPhone" in result + + def test_firefox_windows(self): + from app.utils.session_manager import _parse_device_info + + ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0" + result = _parse_device_info(ua) + assert "Firefox" in result + assert "Windows" in result + + def test_edge_windows(self): + from app.utils.session_manager import _parse_device_info + + ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0" + result = _parse_device_info(ua) + assert "Edge" in result + assert "Windows" in result + + def test_android_chrome(self): + from app.utils.session_manager import _parse_device_info + + ua = "Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.6099.210 Mobile Safari/537.36" + result = _parse_device_info(ua) + assert "Chrome" in result + assert "Android" in result + + def test_none_user_agent(self): + from app.utils.session_manager import _parse_device_info + + assert _parse_device_info(None) is None + + def test_empty_user_agent(self): + from app.utils.session_manager import _parse_device_info + + assert _parse_device_info("") is None + + +# --------------------------------------------------------------------------- +# Config Tests +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestSessionConfig: + """Tests for session-related configuration fields.""" + + def test_session_lifetime_days_field_exists(self): + """Verify session_lifetime_days field is defined in Settings.""" + from app.config import Settings + + # Check the field exists in the model + assert "session_lifetime_days" in Settings.model_fields + + def test_session_lifetime_custom_days_field_exists(self): + """Verify session_lifetime_custom_days field is defined in Settings.""" + from app.config import Settings + + assert "session_lifetime_custom_days" in Settings.model_fields + + def test_qr_login_challenge_ttl_field_exists(self): + """Verify qr_login_challenge_ttl_seconds field is defined in Settings.""" + from app.config import Settings + + assert "qr_login_challenge_ttl_seconds" in Settings.model_fields