Merge pull request #749 from christianlouis/copilot/implement-log-off-everywhere-functionality

Add dedicated Devices page, separate mobile tokens from API Tokens
This commit is contained in:
Christian Krakau-Louis
2026-03-17 12:22:07 +01:00
committed by GitHub
28 changed files with 3248 additions and 24 deletions
+10
View File
@@ -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
+4
View File
@@ -33,11 +33,13 @@ from app.api.pipelines import router as pipelines_router
from app.api.plans import router as plans_router
from app.api.process import router as process_router
from app.api.profile import router as profile_router
from app.api.qr_auth import router as qr_auth_router
from app.api.queue import router as queue_router
from app.api.routing_rules import router as routing_rules_router
from app.api.saved_searches import router as saved_searches_router
from app.api.scheduled_jobs import router as scheduled_jobs_router
from app.api.search import router as search_router
from app.api.sessions import router as sessions_router
from app.api.settings import router as settings_router
from app.api.shared_links import public_router as shared_links_public_router
from app.api.shared_links import router as shared_links_router
@@ -97,6 +99,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)
+55 -15
View File
@@ -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)
+198
View File
@@ -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
+196
View File
@@ -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.",
}
+102
View File
@@ -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)
+20
View File
@@ -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
+12 -1
View File
@@ -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
+80
View File
@@ -969,6 +969,86 @@ class MobileDevice(Base):
__table_args__ = (UniqueConstraint("owner_id", "push_token", name="uq_mobile_device_owner_token"),)
class UserSession(Base):
"""Server-side session tracking for invalidation and device management.
Each row represents an active browser or app session. The ``session_token``
is stored in the user's cookie and validated on every authenticated request.
Revoking a row (``is_revoked=True``) immediately terminates that session
on the next request.
"""
__tablename__ = "user_sessions"
id = Column(Integer, primary_key=True, index=True)
# Cryptographically random token stored in the session cookie.
session_token = Column(String(128), unique=True, nullable=False, index=True)
# Stable owner identifier — matches FileRecord.owner_id.
user_id = Column(String, nullable=False, index=True)
# Client metadata for display in the session management UI.
ip_address = Column(String(45), nullable=True)
user_agent = Column(String(512), nullable=True)
device_info = Column(String(255), nullable=True)
is_revoked = Column(Boolean, nullable=False, default=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
last_active_at = Column(DateTime(timezone=True), server_default=func.now())
expires_at = Column(DateTime(timezone=True), nullable=False)
revoked_at = Column(DateTime(timezone=True), nullable=True)
class QRLoginChallenge(Base):
"""Time-limited QR code login challenge for mobile app authentication.
A logged-in web user generates a challenge that produces a QR code. The
mobile app scans the QR code and calls the claim endpoint with the
``challenge_token``. The server verifies the challenge is still valid,
unclaimed, and unexpired, then issues an API token for the mobile app.
Security properties:
* Time-bound (default 2 minutes).
* Single-use (``is_claimed`` prevents replay).
* Cryptographically random 64-byte token.
* Bound to the creating user — only that user's mobile device receives a
token.
"""
__tablename__ = "qr_login_challenges"
id = Column(Integer, primary_key=True, index=True)
# Cryptographically random token encoded in the QR code.
challenge_token = Column(String(128), unique=True, nullable=False, index=True)
# The user who created this challenge (from the web session).
user_id = Column(String, nullable=False, index=True)
# Whether the challenge has been successfully claimed by a mobile app.
is_claimed = Column(Boolean, nullable=False, default=False)
# Whether the challenge has been explicitly cancelled or expired.
is_cancelled = Column(Boolean, nullable=False, default=False)
# IP address of the web client that created the challenge.
created_by_ip = Column(String(45), nullable=True)
# IP address of the mobile client that claimed the challenge.
claimed_by_ip = Column(String(45), nullable=True)
# Device name provided by the mobile app when claiming.
device_name = Column(String(255), nullable=True)
# The API token ID that was issued to the mobile app (for audit trail).
issued_token_id = Column(Integer, nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
expires_at = Column(DateTime(timezone=True), nullable=False)
claimed_at = Column(DateTime(timezone=True), nullable=True)
class ComplianceTemplate(Base):
"""Pre-built compliance configuration templates (GDPR, HIPAA, SOC2).
+505
View File
@@ -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}"
+24
View File
@@ -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",
+4
View File
@@ -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
+25
View File
@@ -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"},
)
+26
View File
@@ -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},
)
+53
View File
@@ -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
+3
View File
@@ -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`. |
+15
View File
@@ -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';
+1
View File
@@ -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 }},
+351
View File
@@ -0,0 +1,351 @@
{% extends "base.html" %}
{% block title %}{{ _("devices.page_title") }}{% endblock %}
{% block content %}
<div x-data="devicesPage()" x-init="init()" class="container mx-auto px-4 py-8 max-w-4xl">
<!-- ── Header ─────────────────────────────────────────────────────────── -->
<header class="mb-8">
<h1 class="text-2xl font-bold text-gray-900 dark:text-white flex items-center gap-2">
<i class="fas fa-mobile-alt text-blue-500" aria-hidden="true"></i>
{{ _("devices.heading") }}
</h1>
<p class="mt-2 text-gray-600 dark:text-gray-400 text-sm leading-relaxed max-w-2xl">
{{ _("devices.intro") }}
</p>
</header>
<!-- ── Mobile App Tokens ──────────────────────────────────────────────── -->
<section class="bg-white dark:bg-gray-800 shadow rounded-lg overflow-hidden mb-6" aria-labelledby="mobile-tokens-heading">
<div class="px-6 py-4 border-b border-gray-200 dark:border-gray-700">
<h2 id="mobile-tokens-heading" class="text-lg font-semibold text-gray-900 dark:text-white">
<i class="fas fa-key text-yellow-500 mr-2" aria-hidden="true"></i>{{ _("devices.mobile_tokens_heading") }}
</h2>
<p class="text-sm text-gray-500 dark:text-gray-400 mt-1">{{ _("devices.mobile_tokens_description") }}</p>
</div>
<!-- Loading -->
<template x-if="loadingTokens">
<div class="p-8 text-center text-gray-500 dark:text-gray-400">
<i class="fas fa-spinner fa-spin text-2xl mb-2" aria-hidden="true"></i>
<p class="text-sm">{{ _("devices.loading") }}</p>
</div>
</template>
<!-- Empty state -->
<template x-if="!loadingTokens && mobileTokens.length === 0">
<div class="p-8 text-center text-gray-500 dark:text-gray-400">
<i class="fas fa-mobile-alt text-4xl mb-3 text-gray-300 dark:text-gray-600" aria-hidden="true"></i>
<p class="font-medium">{{ _("devices.no_mobile_tokens") }}</p>
<p class="text-sm mt-1">{{ _("devices.no_mobile_tokens_help") }}</p>
</div>
</template>
<!-- Tokens table -->
<template x-if="!loadingTokens && mobileTokens.length > 0">
<div class="overflow-x-auto">
<table class="w-full text-sm" aria-label="{{ _('devices.mobile_tokens_heading') }}">
<thead>
<tr class="bg-gray-50 dark:bg-gray-750 text-left">
<th scope="col" class="px-6 py-3 font-medium text-gray-500 dark:text-gray-400 uppercase text-xs tracking-wider">{{ _("devices.col_device") }}</th>
<th scope="col" class="px-6 py-3 font-medium text-gray-500 dark:text-gray-400 uppercase text-xs tracking-wider">{{ _("devices.col_token_prefix") }}</th>
<th scope="col" class="px-6 py-3 font-medium text-gray-500 dark:text-gray-400 uppercase text-xs tracking-wider">{{ _("devices.col_created") }}</th>
<th scope="col" class="px-6 py-3 font-medium text-gray-500 dark:text-gray-400 uppercase text-xs tracking-wider">{{ _("devices.col_last_used") }}</th>
<th scope="col" class="px-6 py-3 font-medium text-gray-500 dark:text-gray-400 uppercase text-xs tracking-wider">{{ _("devices.col_status") }}</th>
<th scope="col" class="px-6 py-3 font-medium text-gray-500 dark:text-gray-400 uppercase text-xs tracking-wider sr-only">{{ _("common.actions") }}</th>
</tr>
</thead>
<tbody class="divide-y divide-gray-200 dark:divide-gray-700">
<template x-for="token in mobileTokens" :key="token.id">
<tr class="hover:bg-gray-50 dark:hover:bg-gray-750 transition-colors">
<td class="px-6 py-4 whitespace-nowrap">
<div class="flex items-center gap-2">
<i class="fas fa-mobile-alt text-gray-400" aria-hidden="true"></i>
<span class="font-medium text-gray-900 dark:text-white" x-text="formatDeviceName(token.name)"></span>
</div>
</td>
<td class="px-6 py-4 whitespace-nowrap">
<code class="bg-gray-100 dark:bg-gray-700 px-2 py-1 rounded text-xs font-mono" x-text="token.token_prefix + '…'"></code>
</td>
<td class="px-6 py-4 whitespace-nowrap text-gray-500 dark:text-gray-400" x-text="formatDate(token.created_at)"></td>
<td class="px-6 py-4 whitespace-nowrap text-gray-500 dark:text-gray-400">
<span x-text="token.last_used_at ? formatDate(token.last_used_at) : '—'"></span>
<span x-show="token.last_used_ip" class="block text-xs text-gray-400 mt-0.5">
<i class="fas fa-globe mr-1" aria-hidden="true"></i><span x-text="token.last_used_ip"></span>
</span>
</td>
<td class="px-6 py-4 whitespace-nowrap">
<span
class="inline-flex items-center px-2.5 py-0.5 rounded-full text-xs font-medium"
:class="token.is_active ? 'bg-green-100 text-green-800 dark:bg-green-900/30 dark:text-green-400' : 'bg-red-100 text-red-800 dark:bg-red-900/30 dark:text-red-400'"
x-text="token.is_active ? '{{ _('devices.status_active') }}' : '{{ _('devices.status_revoked') }}'"
></span>
</td>
<td class="px-6 py-4 whitespace-nowrap text-right">
<button
x-show="token.is_active"
type="button"
@click="revokeToken(token)"
:disabled="revokingToken === token.id"
class="inline-flex items-center px-3 py-1.5 text-sm font-medium text-red-600 hover:text-red-800
dark:text-red-400 dark:hover:text-red-300 hover:bg-red-50 dark:hover:bg-red-900/20 rounded-md
focus:outline-none focus:ring-2 focus:ring-red-500 disabled:opacity-50 transition-colors"
style="min-height:36px; min-width:44px;"
:aria-label="'{{ _('devices.revoke_token') }} ' + token.name"
>
<i :class="revokingToken === token.id ? 'fas fa-spinner fa-spin' : 'fas fa-sign-out-alt'" class="mr-1" aria-hidden="true"></i>
{{ _("devices.revoke_token") }}
</button>
</td>
</tr>
</template>
</tbody>
</table>
</div>
</template>
<!-- Error -->
<template x-if="tokenError">
<div class="m-4 bg-red-50 dark:bg-red-900/20 border border-red-200 dark:border-red-800 text-red-700 dark:text-red-400 p-3 rounded text-sm" role="alert">
<i class="fas fa-exclamation-triangle mr-1" aria-hidden="true"></i>
<span x-text="tokenError"></span>
</div>
</template>
</section>
<!-- ── Registered Devices (Push Notifications) ────────────────────────── -->
<section class="bg-white dark:bg-gray-800 shadow rounded-lg overflow-hidden mb-6" aria-labelledby="devices-heading">
<div class="px-6 py-4 border-b border-gray-200 dark:border-gray-700">
<h2 id="devices-heading" class="text-lg font-semibold text-gray-900 dark:text-white">
<i class="fas fa-bell text-purple-500 mr-2" aria-hidden="true"></i>{{ _("devices.registered_devices_heading") }}
</h2>
<p class="text-sm text-gray-500 dark:text-gray-400 mt-1">{{ _("devices.registered_devices_description") }}</p>
</div>
<!-- Loading -->
<template x-if="loadingDevices">
<div class="p-8 text-center text-gray-500 dark:text-gray-400">
<i class="fas fa-spinner fa-spin text-2xl mb-2" aria-hidden="true"></i>
<p class="text-sm">{{ _("devices.loading") }}</p>
</div>
</template>
<!-- Empty state -->
<template x-if="!loadingDevices && devices.length === 0">
<div class="p-8 text-center text-gray-500 dark:text-gray-400">
<i class="fas fa-bell-slash text-4xl mb-3 text-gray-300 dark:text-gray-600" aria-hidden="true"></i>
<p class="font-medium">{{ _("devices.no_devices") }}</p>
<p class="text-sm mt-1">{{ _("devices.no_devices_help") }}</p>
</div>
</template>
<!-- Devices list -->
<template x-if="!loadingDevices && devices.length > 0">
<div class="divide-y divide-gray-200 dark:divide-gray-700">
<template x-for="device in devices" :key="device.id">
<div class="flex items-center justify-between px-6 py-4 hover:bg-gray-50 dark:hover:bg-gray-750 transition-colors">
<div class="flex items-center gap-3 min-w-0">
<i
:class="device.platform === 'ios' ? 'fab fa-apple' :
device.platform === 'android' ? 'fab fa-android text-green-500' :
'fas fa-globe'"
class="text-lg text-gray-400 flex-shrink-0"
aria-hidden="true"
></i>
<div class="min-w-0">
<div class="text-sm font-medium text-gray-900 dark:text-white truncate">
<span x-text="device.device_name || 'Unknown Device'"></span>
<span
class="ml-2 inline-flex items-center px-2 py-0.5 rounded text-xs font-medium"
:class="device.is_active ? 'bg-green-100 text-green-800 dark:bg-green-900/30 dark:text-green-400' : 'bg-gray-100 text-gray-600 dark:bg-gray-700 dark:text-gray-400'"
x-text="device.is_active ? '{{ _('devices.status_active') }}' : '{{ _('devices.status_inactive') }}'"
></span>
</div>
<div class="text-xs text-gray-500 dark:text-gray-400 space-x-3 mt-0.5">
<span>
<i class="fas fa-microchip mr-1" aria-hidden="true"></i>
<span x-text="device.platform.charAt(0).toUpperCase() + device.platform.slice(1)"></span>
</span>
<span x-show="device.last_seen_at">
<i class="fas fa-clock mr-1" aria-hidden="true"></i>{{ _("devices.col_last_seen") }}:
<span x-text="formatDate(device.last_seen_at)"></span>
</span>
<span>
<i class="fas fa-calendar mr-1" aria-hidden="true"></i>
<span x-text="formatDate(device.created_at)"></span>
</span>
</div>
</div>
</div>
<button
x-show="device.is_active"
type="button"
@click="deactivateDevice(device)"
:disabled="deactivatingDevice === device.id"
class="flex-shrink-0 inline-flex items-center px-3 py-1.5 text-sm font-medium text-red-600 hover:text-red-800
dark:text-red-400 dark:hover:text-red-300 hover:bg-red-50 dark:hover:bg-red-900/20 rounded-md
focus:outline-none focus:ring-2 focus:ring-red-500 disabled:opacity-50 transition-colors"
style="min-height:36px; min-width:44px;"
:aria-label="'{{ _('devices.deactivate_device') }} ' + (device.device_name || 'device')"
>
<i :class="deactivatingDevice === device.id ? 'fas fa-spinner fa-spin' : 'fas fa-trash-alt'" class="mr-1" aria-hidden="true"></i>
{{ _("devices.deactivate_device") }}
</button>
</div>
</template>
</div>
</template>
<!-- Error -->
<template x-if="deviceError">
<div class="m-4 bg-red-50 dark:bg-red-900/20 border border-red-200 dark:border-red-800 text-red-700 dark:text-red-400 p-3 rounded text-sm" role="alert">
<i class="fas fa-exclamation-triangle mr-1" aria-hidden="true"></i>
<span x-text="deviceError"></span>
</div>
</template>
</section>
<!-- ── QR Login CTA ───────────────────────────────────────────────────── -->
<div class="text-center">
<a
href="/qr-login"
class="inline-flex items-center px-5 py-2.5 bg-blue-600 hover:bg-blue-700 text-white text-sm font-medium
rounded-lg focus:outline-none focus:ring-2 focus:ring-blue-500 transition-colors"
style="min-height:44px;"
>
<i class="fas fa-qrcode mr-2" aria-hidden="true"></i>
{{ _("devices.qr_login_cta") }}
</a>
</div>
<!-- ── Status banner ──────────────────────────────────────────────────── -->
<div
x-show="banner.visible"
x-transition
class="mt-6 rounded-lg p-3 text-sm"
:class="banner.error
? 'bg-red-50 dark:bg-red-900/30 text-red-800 dark:text-red-200 border border-red-300 dark:border-red-700'
: 'bg-green-50 dark:bg-green-900/30 text-green-800 dark:text-green-200 border border-green-300 dark:border-green-700'"
role="alert"
aria-live="polite"
>
<span x-text="banner.message"></span>
</div>
</div>
<script>
function devicesPage() {
const csrfToken = '{{ csrf_token | default("") }}';
return {
mobileTokens: [],
devices: [],
loadingTokens: true,
loadingDevices: true,
revokingToken: null,
deactivatingDevice: null,
tokenError: null,
deviceError: null,
banner: { visible: false, error: false, message: '' },
async init() {
await Promise.all([this.loadMobileTokens(), this.loadDevices()]);
},
async loadMobileTokens() {
this.loadingTokens = true;
this.tokenError = null;
try {
const res = await fetch('/api/api-tokens/mobile', {
headers: { 'X-CSRF-Token': csrfToken },
});
if (!res.ok) throw new Error('Failed to load mobile tokens');
this.mobileTokens = await res.json();
} catch (e) {
this.tokenError = e.message;
} finally {
this.loadingTokens = false;
}
},
async loadDevices() {
this.loadingDevices = true;
this.deviceError = null;
try {
const res = await fetch('/api/mobile/devices', {
headers: { 'X-CSRF-Token': csrfToken },
});
if (!res.ok) throw new Error('Failed to load devices');
this.devices = await res.json();
} catch (e) {
this.deviceError = e.message;
} finally {
this.loadingDevices = false;
}
},
async revokeToken(token) {
if (!confirm({{ _("devices.confirm_revoke_token") | tojson }})) return;
this.revokingToken = token.id;
try {
const res = await fetch(`/api/api-tokens/${token.id}`, {
method: 'DELETE',
headers: { 'X-CSRF-Token': csrfToken },
});
if (!res.ok) {
const data = await res.json().catch(() => ({}));
throw new Error(data.detail || 'Failed to revoke token');
}
await this.loadMobileTokens();
this._showBanner({{ _("devices.token_revoked_success") | tojson }}, false);
} catch (e) {
this._showBanner(e.message, true);
} finally {
this.revokingToken = null;
}
},
async deactivateDevice(device) {
if (!confirm({{ _("devices.confirm_deactivate_device") | tojson }})) return;
this.deactivatingDevice = device.id;
try {
const res = await fetch(`/api/mobile/devices/${device.id}`, {
method: 'DELETE',
headers: { 'X-CSRF-Token': csrfToken },
});
if (!res.ok && res.status !== 204) {
const data = await res.json().catch(() => ({}));
throw new Error(data.detail || 'Failed to remove device');
}
await this.loadDevices();
this._showBanner({{ _("devices.device_removed_success") | tojson }}, false);
} catch (e) {
this._showBanner(e.message, true);
} finally {
this.deactivatingDevice = null;
}
},
/** Extract the device name from the full token name (e.g. "Mobile App iPhone 15 Pro" → "iPhone 15 Pro"). */
formatDeviceName(name) {
if (!name) return 'Unknown Device';
// Match either em dash () or hyphen (-) separators used by the mobile flows.
const match = name.match(/[\-]\s*(.+)$/);
return match ? match[1].trim() : name;
},
formatDate(d) {
if (!d) return '—';
const dt = new Date(d);
return dt.toLocaleDateString(undefined, { year: 'numeric', month: 'short', day: 'numeric' }) +
' ' + dt.toLocaleTimeString(undefined, { hour: '2-digit', minute: '2-digit' });
},
_showBanner(msg, isError) {
this.banner = { visible: true, error: isError, message: msg };
setTimeout(() => { this.banner.visible = false; }, 5000);
},
};
}
</script>
{% endblock %}
+194
View File
@@ -343,9 +343,203 @@
</section>
</template>
<!-- ── Security & Sessions card ──────────────────────────────────────── -->
<section
class="bg-white dark:bg-gray-800 shadow rounded-lg p-6 mb-6"
aria-labelledby="security-heading"
x-data="sessionManager()"
x-init="loadSessions()"
>
<h2 id="security-heading" class="text-base font-semibold text-gray-900 dark:text-white mb-1">
<i class="fas fa-shield-alt text-gray-400 mr-2" aria-hidden="true"></i>{{ _("sessions.security_heading") }}
</h2>
<p class="text-sm text-gray-500 dark:text-gray-400 mb-4">
{{ _("sessions.security_subtitle") }}
</p>
<!-- Session lifetime info -->
<div class="text-xs text-gray-400 dark:text-gray-500 mb-4" x-show="lifetimeDays > 0">
<i class="fas fa-clock mr-1" aria-hidden="true"></i>
<span x-text="'{{ _("sessions.session_lifetime") }}'.replace('{days}', lifetimeDays)"></span>
</div>
<!-- Active sessions list -->
<div class="space-y-3 mb-5">
<template x-for="session in sessions" :key="session.id">
<div
class="flex items-center justify-between border border-gray-200 dark:border-gray-700 rounded-lg p-3"
:class="session.is_current ? 'bg-blue-50 dark:bg-blue-900/20 border-blue-300 dark:border-blue-700' : ''"
>
<div class="flex items-center gap-3 min-w-0">
<i
:class="session.device_info && session.device_info.includes('iPhone') ? 'fas fa-mobile-alt' :
session.device_info && session.device_info.includes('iPad') ? 'fas fa-tablet-alt' :
session.device_info && session.device_info.includes('Android') ? 'fas fa-mobile-alt' :
session.device_info && session.device_info.includes('App') ? 'fas fa-mobile-alt' :
'fas fa-desktop'"
class="text-gray-400 text-lg flex-shrink-0"
aria-hidden="true"
></i>
<div class="min-w-0">
<div class="text-sm font-medium text-gray-900 dark:text-white truncate">
<span x-text="session.device_info || 'Unknown Device'"></span>
<span
x-show="session.is_current"
class="ml-2 inline-flex items-center px-2 py-0.5 rounded text-xs font-medium bg-blue-100 text-blue-800 dark:bg-blue-900 dark:text-blue-200"
>{{ _("sessions.current_session") }}</span>
</div>
<div class="text-xs text-gray-500 dark:text-gray-400 space-x-3">
<span x-show="session.ip_address">
<i class="fas fa-globe mr-1" aria-hidden="true"></i><span x-text="session.ip_address"></span>
</span>
<span>
<i class="fas fa-clock mr-1" aria-hidden="true"></i>{{ _("sessions.last_active") }}
<span x-text="timeAgo(session.last_active_at)"></span>
</span>
</div>
</div>
</div>
<button
x-show="!session.is_current"
@click="revokeSession(session.id)"
class="flex-shrink-0 text-red-600 hover:text-red-800 dark:text-red-400 dark:hover:text-red-300 text-sm font-medium px-3 py-1 rounded hover:bg-red-50 dark:hover:bg-red-900/20 transition"
style="min-height:44px; min-width:44px;"
:aria-label="'{{ _("sessions.revoke") }}'"
>
<i class="fas fa-sign-out-alt mr-1" aria-hidden="true"></i>{{ _("sessions.revoke") }}
</button>
</div>
</template>
<p
x-show="sessions.length <= 1"
class="text-sm text-gray-500 dark:text-gray-400 italic"
>{{ _("sessions.no_other_sessions") }}</p>
</div>
<!-- Log off everywhere + QR login row -->
<div class="flex flex-col sm:flex-row gap-3">
<button
@click="revokeAllSessions()"
class="inline-flex items-center justify-center px-4 py-2 border border-red-300 dark:border-red-700 rounded-lg text-sm font-medium text-red-700 dark:text-red-300 bg-white dark:bg-gray-800 hover:bg-red-50 dark:hover:bg-red-900/20 transition"
style="min-height:44px;"
:disabled="revoking"
>
<i class="fas fa-power-off mr-2" aria-hidden="true"></i>
<span x-text="revoking ? '{{ _("profile.saving") }}' : '{{ _("sessions.log_off_everywhere") }}'"></span>
</button>
<a
href="/qr-login"
class="inline-flex items-center justify-center px-4 py-2 border border-gray-300 dark:border-gray-600 rounded-lg text-sm font-medium text-gray-700 dark:text-gray-300 bg-white dark:bg-gray-800 hover:bg-gray-50 dark:hover:bg-gray-700 transition"
style="min-height:44px;"
>
<i class="fas fa-qrcode mr-2" aria-hidden="true"></i>
{{ _("sessions.qr_login_link") }}
</a>
</div>
<!-- Status banner for session actions -->
<div
x-show="sessionBanner.visible"
x-transition
class="mt-4 rounded-lg p-3 text-sm"
:class="sessionBanner.error
? 'bg-red-50 dark:bg-red-900/30 text-red-800 dark:text-red-200 border border-red-300 dark:border-red-700'
: 'bg-green-50 dark:bg-green-900/30 text-green-800 dark:text-green-200 border border-green-300 dark:border-green-700'"
role="alert"
aria-live="polite"
>
<span x-text="sessionBanner.message"></span>
</div>
</section>
</div><!-- /container -->
<script>
/* ── Session Manager Alpine component ──────────────────────────────────── */
function sessionManager() {
return {
sessions: [],
lifetimeDays: 0,
revoking: false,
sessionBanner: { visible: false, error: false, message: '' },
_csrfToken() {
return document.cookie
.split('; ')
.find(row => row.startsWith('csrf_token='))
?.split('=')[1];
},
async loadSessions() {
try {
const res = await fetch('/api/sessions/');
if (res.ok) {
const data = await res.json();
this.sessions = data.sessions || [];
this.lifetimeDays = data.session_lifetime_days || 30;
}
} catch (_e) { /* silently ignore */ }
},
async revokeSession(sessionId) {
if (!confirm({{ _("sessions.confirm_revoke_one") | tojson }})) return;
const csrf = this._csrfToken();
try {
const res = await fetch(`/api/sessions/${sessionId}`, {
method: 'DELETE',
headers: csrf ? { 'X-CSRF-Token': csrf } : {},
});
if (res.ok || res.status === 204) {
this.sessions = this.sessions.filter(s => s.id !== sessionId);
this._showSessionBanner({{ _("sessions.revoked_success") | tojson }}, false);
}
} catch (_e) {
this._showSessionBanner('Network error — please try again.', true);
}
},
async revokeAllSessions() {
if (!confirm({{ _("sessions.confirm_revoke_all") | tojson }})) return;
this.revoking = true;
const csrf = this._csrfToken();
try {
const res = await fetch('/api/sessions/revoke-all', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
...(csrf ? { 'X-CSRF-Token': csrf } : {}),
},
});
if (res.ok) {
await this.loadSessions();
this._showSessionBanner({{ _("sessions.revoked_all_success") | tojson }}, false);
}
} catch (_e) {
this._showSessionBanner('Network error — please try again.', true);
} finally {
this.revoking = false;
}
},
timeAgo(dateStr) {
if (!dateStr) return 'unknown';
const now = new Date();
const then = new Date(dateStr);
const diff = Math.floor((now - then) / 1000);
if (diff < 60) return 'just now';
if (diff < 3600) return Math.floor(diff / 60) + 'm ago';
if (diff < 86400) return Math.floor(diff / 3600) + 'h ago';
return Math.floor(diff / 86400) + 'd ago';
},
_showSessionBanner(msg, err) {
this.sessionBanner = { visible: true, error: err, message: msg };
if (!err) setTimeout(() => { this.sessionBanner.visible = false; }, 4000);
},
};
}
/* ── Profile Settings Alpine component ─────────────────────────────────── */
function profileSettings() {
return {
// ── State ──────────────────────────────────────────────────────────────
+229
View File
@@ -0,0 +1,229 @@
{% extends "base.html" %}
{% block title %}{{ _("qr_login.page_title") }}{% endblock %}
{% block content %}
<div
x-data="qrLoginPage()"
x-init="generateChallenge()"
class="container mx-auto px-4 py-8 max-w-xl"
>
<!-- ── Header ─────────────────────────────────────────────────────────── -->
<header class="mb-8 text-center">
<h1 class="text-2xl font-bold text-gray-900 dark:text-white flex items-center justify-center gap-2">
<i class="fas fa-qrcode text-blue-500" aria-hidden="true"></i>
{{ _("qr_login.heading") }}
</h1>
<p class="mt-2 text-sm text-gray-500 dark:text-gray-400">
{{ _("qr_login.subtitle") }}
</p>
</header>
<!-- ── QR Code Card ───────────────────────────────────────────────────── -->
<section
class="bg-white dark:bg-gray-800 shadow rounded-lg p-8 mb-6 text-center"
aria-labelledby="qr-heading"
>
<!-- Pending state: show QR code -->
<template x-if="status === 'pending'">
<div>
<div
class="mx-auto mb-4 bg-white p-4 inline-block rounded-lg shadow-inner"
id="qr-container"
aria-label="{{ _('qr_login.description') }}"
>
<canvas id="qr-canvas" width="256" height="256"></canvas>
</div>
<p class="text-sm text-gray-500 dark:text-gray-400 mb-2">
{{ _("qr_login.description") }}
</p>
<div class="flex items-center justify-center gap-2 text-xs text-gray-400 dark:text-gray-500">
<i class="fas fa-hourglass-half animate-pulse" aria-hidden="true"></i>
<span x-text="'{{ _("qr_login.time_remaining") }}'.replace('{seconds}', countdown)"></span>
</div>
<p class="mt-3 text-sm text-blue-600 dark:text-blue-400">
<i class="fas fa-spinner fa-spin mr-1" aria-hidden="true"></i>
{{ _("qr_login.pending_message") }}
</p>
</div>
</template>
<!-- Claimed state: success -->
<template x-if="status === 'claimed'">
<div class="py-8">
<i class="fas fa-check-circle text-green-500 text-5xl mb-4" aria-hidden="true"></i>
<p class="text-lg font-semibold text-green-700 dark:text-green-400 mb-2">
{{ _("qr_login.claimed_message") }}
</p>
<p x-show="deviceName" class="text-sm text-gray-500 dark:text-gray-400"
x-text="'{{ _("qr_login.claimed_device") }}'.replace('{device_name}', deviceName)">
</p>
</div>
</template>
<!-- Expired state -->
<template x-if="status === 'expired'">
<div class="py-8">
<i class="fas fa-clock text-yellow-500 text-5xl mb-4" aria-hidden="true"></i>
<p class="text-base text-gray-700 dark:text-gray-300 mb-4">
{{ _("qr_login.expired_message") }}
</p>
<button
@click="generateChallenge()"
class="inline-flex items-center px-4 py-2 bg-blue-600 hover:bg-blue-700 text-white font-medium rounded-lg transition"
style="min-height:44px;"
>
<i class="fas fa-redo mr-2" aria-hidden="true"></i>
{{ _("qr_login.generate_new") }}
</button>
</div>
</template>
<!-- Error state -->
<template x-if="status === 'error'">
<div class="py-8">
<i class="fas fa-exclamation-triangle text-red-500 text-5xl mb-4" aria-hidden="true"></i>
<p class="text-base text-gray-700 dark:text-gray-300 mb-4" x-text="errorMsg"></p>
<button
@click="generateChallenge()"
class="inline-flex items-center px-4 py-2 bg-blue-600 hover:bg-blue-700 text-white font-medium rounded-lg transition"
style="min-height:44px;"
>
<i class="fas fa-redo mr-2" aria-hidden="true"></i>
{{ _("qr_login.generate_new") }}
</button>
</div>
</template>
</section>
<!-- ── How it works ───────────────────────────────────────────────────── -->
<section class="bg-white dark:bg-gray-800 shadow rounded-lg p-6">
<h2 class="text-base font-semibold text-gray-900 dark:text-white mb-3">
<i class="fas fa-info-circle text-gray-400 mr-2" aria-hidden="true"></i>
{{ _("qr_login.how_it_works") }}
</h2>
<ol class="list-decimal list-inside space-y-2 text-sm text-gray-600 dark:text-gray-400">
<li>{{ _("qr_login.step_1") }}</li>
<li>{{ _("qr_login.step_2") }}</li>
<li>{{ _("qr_login.step_3") }}</li>
</ol>
</section>
</div>
<!-- QR Code library (lightweight, no external deps) -->
<script src="https://cdn.jsdelivr.net/npm/qrcode@1.5.4/build/qrcode.min.js"></script>
<script>
function qrLoginPage() {
return {
status: 'loading', // loading | pending | claimed | expired | error
challengeId: null,
challengeToken: '',
qrPayload: '',
expiresAt: null,
countdown: 0,
deviceName: '',
errorMsg: '',
_pollTimer: null,
_countdownTimer: null,
_csrfToken() {
return document.cookie
.split('; ')
.find(row => row.startsWith('csrf_token='))
?.split('=')[1];
},
async generateChallenge() {
this.status = 'loading';
this._stopTimers();
const csrf = this._csrfToken();
try {
const res = await fetch('/api/qr-auth/challenge', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
...(csrf ? { 'X-CSRF-Token': csrf } : {}),
},
});
if (!res.ok) {
this.status = 'error';
this.errorMsg = 'Failed to generate QR code. Please try again.';
return;
}
const data = await res.json();
this.challengeId = data.challenge_id;
this.challengeToken = data.challenge_token;
this.qrPayload = data.qr_payload;
this.expiresAt = new Date(data.expires_at);
this.status = 'pending';
this.deviceName = '';
// Render QR code
this.$nextTick(() => {
const canvas = document.getElementById('qr-canvas');
if (canvas && typeof QRCode !== 'undefined') {
QRCode.toCanvas(canvas, this.qrPayload, {
width: 256,
margin: 2,
color: { dark: '#000000', light: '#ffffff' },
});
}
});
// Start polling and countdown
this._startPolling();
this._startCountdown();
} catch (_e) {
this.status = 'error';
this.errorMsg = 'Network error — please check your connection and try again.';
}
},
_startPolling() {
this._pollTimer = setInterval(async () => {
if (this.status !== 'pending') { this._stopTimers(); return; }
try {
const res = await fetch(`/api/qr-auth/challenge/${this.challengeId}/status`);
if (!res.ok) return;
const data = await res.json();
if (data.status === 'claimed') {
this.status = 'claimed';
this.deviceName = data.device_name || '';
this._stopTimers();
} else if (data.status === 'expired') {
this.status = 'expired';
this._stopTimers();
} else if (data.status === 'cancelled') {
this.status = 'expired';
this._stopTimers();
}
} catch (_e) { /* ignore transient errors */ }
}, 2000);
},
_startCountdown() {
this._updateCountdown();
this._countdownTimer = setInterval(() => {
this._updateCountdown();
if (this.countdown <= 0 && this.status === 'pending') {
this.status = 'expired';
this._stopTimers();
}
}, 1000);
},
_updateCountdown() {
if (!this.expiresAt) { this.countdown = 0; return; }
const remaining = Math.max(0, Math.floor((this.expiresAt - new Date()) / 1000));
this.countdown = remaining;
},
_stopTimers() {
if (this._pollTimer) { clearInterval(this._pollTimer); this._pollTimer = null; }
if (this._countdownTimer) { clearInterval(this._countdownTimer); this._countdownTimer = null; }
},
};
}
</script>
{% endblock %}
+65
View File
@@ -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",
@@ -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")
+16
View File
@@ -38,6 +38,7 @@ export interface AuthState {
user: WhoAmIResponse | null;
baseUrl: string;
signIn: (serverUrl: string) => Promise<void>;
signInWithQR: (serverUrl: string, challengeToken: string) => Promise<void>;
signOut: () => Promise<void>;
setToken: (token: string) => Promise<void>;
}
@@ -52,6 +53,7 @@ const AuthContext = createContext<AuthState>({
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,
}}
+102 -8
View File
@@ -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() {
<Pressable
style={[styles.button, loading && styles.buttonDisabled]}
onPress={handleSignIn}
disabled={loading}
disabled={loading || qrLoading}
accessibilityRole="button"
accessibilityLabel="Sign in with SSO"
>
@@ -96,8 +135,33 @@ export default function LoginScreen() {
)}
</Pressable>
<View style={styles.dividerRow}>
<View style={styles.dividerLine} />
<Text style={styles.dividerText}>or</Text>
<View style={styles.dividerLine} />
</View>
<Pressable
style={[styles.qrButton, qrLoading && styles.buttonDisabled]}
onPress={() => {
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 ? (
<ActivityIndicator color="#1e40af" />
) : (
<Text style={styles.qrButtonText}>📱 Scan QR Code to Login</Text>
)}
</Pressable>
<Text style={styles.hint}>
You will be redirected to your organisation's sign-in page.
Sign in via SSO or scan a QR code from the web app.
</Text>
<Pressable
@@ -184,6 +248,36 @@ const styles = StyleSheet.create({
fontSize: 16,
fontWeight: "600",
},
dividerRow: {
flexDirection: "row",
alignItems: "center",
marginVertical: 16,
},
dividerLine: {
flex: 1,
height: 1,
backgroundColor: "#e5e7eb",
},
dividerText: {
marginHorizontal: 12,
fontSize: 12,
color: "#9ca3af",
},
qrButton: {
borderWidth: 1,
borderColor: "#1e40af",
borderRadius: 8,
paddingVertical: 14,
alignItems: "center",
justifyContent: "center",
minHeight: 48,
backgroundColor: "#eff6ff",
},
qrButtonText: {
color: "#1e40af",
fontSize: 15,
fontWeight: "600",
},
hint: {
marginTop: 16,
fontSize: 12,
+15
View File
@@ -35,6 +35,14 @@ export interface GenerateTokenResponse {
created_at: string;
}
export interface QRClaimResponse {
token: string;
token_id: number;
name: string;
owner_id: string;
created_at: string;
}
export interface DeviceRegistration {
push_token: string;
device_name?: string;
@@ -157,6 +165,13 @@ class DocuElevateAPI {
});
}
/** Claim a QR login challenge and receive an API token. */
async claimQRChallenge(challengeToken: string, deviceName: string): Promise<QRClaimResponse> {
return this.request<QRClaimResponse>("POST", "/api/qr-auth/claim", {
body: { challenge_token: challengeToken, device_name: deviceName },
});
}
/** Return profile information for the authenticated user. */
async whoAmI(): Promise<WhoAmIResponse> {
return this.request<WhoAmIResponse>("GET", "/api/mobile/whoami");
+201
View File
@@ -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)
+670
View File
@@ -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