diff --git a/.claude/commands/review.md b/.claude/commands/review.md index e5f2d9d..71c0a6e 100644 --- a/.claude/commands/review.md +++ b/.claude/commands/review.md @@ -17,6 +17,8 @@ You are orchestrating a multi-agent code review pipeline for a pull request in t Execute the eight steps below in order. Each step that says "spawn an agent" must use the Agent tool. Agent prompts are written in English (more reliable for model instructions); the final user-facing comment is in **Turkish** (the project's working language). +> **No nested agents.** Only the orchestrator (this top-level loop) spawns agents — exactly at Steps 1, 2, 3, 4, and 5. Every spawned agent is a **leaf**: it completes its task entirely within its own context and MUST NOT call the `Agent`/`Task` tool or spawn any sub-agent of its own. If work needs to be split, only the orchestrator splits it. + --- ### Step 1 — Eligibility check (Sonnet) @@ -82,7 +84,7 @@ Pass this object to the five parallel reviewers in Step 4. ### Step 4 — Five parallel Sonnet reviewers -> **Read-only mandate for every spawned agent (Steps 1–5).** Each agent's ONLY output is the JSON described in its prompt. An agent MUST NOT post, edit, or delete anything on the PR: never run `gh pr comment`, `gh issue comment`, `gh pr review`, or any `gh api` call with `-X POST`/`-X PATCH`/`-X PUT`/`-X DELETE`. Only the orchestrator posts, and only once, in Step 8. Any agent that writes to the PR is a pipeline violation and causes duplicate comments. +> **Read-only mandate for every spawned agent (Steps 1–5).** Each agent's ONLY output is the JSON described in its prompt. An agent MUST NOT post, edit, or delete anything on the PR: never run `gh pr comment`, `gh issue comment`, `gh pr review`, or any `gh api` call with `-X POST`/`-X PATCH`/`-X PUT`/`-X DELETE`. Only the orchestrator posts, and only once, in Step 8. Any agent that writes to the PR is a pipeline violation and causes duplicate comments. **Each agent is also a leaf node: it does its own work in its own context and MUST NOT call the `Agent`/`Task` tool or spawn any sub-agent.** A spawned agent that spawns further agents is a pipeline violation. Spawn the following five agents **in parallel** in a single message, all with `subagent_type: "general-purpose"` and `model: "sonnet"`. Each must return strict JSON of the shape: @@ -363,6 +365,7 @@ Use `--body-file` (not `--body`) to preserve markdown formatting and avoid shell - **Never edit files.** This command only reviews. - **Only the orchestrator posts, and only in Step 8.** Spawned agents (Steps 1–5) are read-only: they return JSON and never call `gh pr comment`, `gh issue comment`, `gh pr review`, or a writing `gh api` (`-X POST/PATCH/PUT/DELETE`). If a sub-agent reports it already posted a comment, do not post again — verify the PR state and reconcile to exactly one correct comment. +- **No nested agents.** Spawned agents are leaf nodes — they complete their task in their own context and never call the `Agent`/`Task` tool or spawn sub-agents. Only the orchestrator spawns, exactly at Steps 1, 2, 3, 4, and 5. - **Never post more than one comment per run.** All findings are batched into a single comment. - **Never post without `REVIEW.md`** — abort instead. The whole pipeline relies on it for false-positive control. - **Always use the full head SHA** in permalinks, captured at Step 1, even if the PR receives new commits during the run. The review applies to the SHA you actually read. diff --git a/.env.example b/.env.example index bca9cd3..bc1690a 100644 --- a/.env.example +++ b/.env.example @@ -48,6 +48,13 @@ CLOUDINARY_UPLOAD_FOLDER="uploads" # Max upload size in bytes (default 5 MB). Uploads above this are rejected. MAX_UPLOAD_SIZE_BYTES=5242880 +# Session housekeeping (nightly arq cron). Revoked session rows are kept this +# many days for audit trails before being swept; expired rows go immediately. +# The two cron fields set the daily run time (UTC, 24h). +SESSION_REVOKED_RETENTION_DAYS=30 +SESSION_PURGE_CRON_HOUR=4 +SESSION_PURGE_CRON_MINUTE=0 + # Sentry (only initialized when ENVIRONMENT != "local") SENTRY_DSN= diff --git a/README.md b/README.md index 73537e6..1276c38 100644 --- a/README.md +++ b/README.md @@ -36,13 +36,14 @@ This template integrates the best-in-class Python ecosystem tools to provide a s - **Security & Auth:** JWT access/refresh tokens accepted via either HttpOnly cookies *or* `Authorization: Bearer`, `bcrypt` password hashing, **Redis-backed token blacklist** (logout invalidation), strict origin-check middleware (returns 404 for foreign origins), and [Slowapi](https://slowapi.readthedocs.io/en/latest/) rate limiting for brute-force protection. - **Account Lifecycle:** Email verification, password reset, password change, and **soft-delete with grace period** — accounts marked for deletion can be reactivated until the cron worker purges them. - **File Uploads & Avatars:** [Cloudinary](https://cloudinary.com/)-backed image uploads via `POST /upload` (image-only, 5 MB cap, rate-limited + audited). A generic `File` model lets any feature attach uploads; user avatars enforce **ownership (IDOR) guards** — you may only attach a file you uploaded — and auto-delete the replaced avatar's asset. Admins get full file management (list/filter by uploader, hard-delete with avatar references auto-cleared). See [File Uploads & Avatars](#-file-uploads--avatars) below. -- **Background Jobs:** [arq](https://arq-docs.helpmanual.io/) worker (separate container in compose) runs cron jobs such as `delete_expired_accounts` at the configured time. +- **Background Jobs:** [arq](https://arq-docs.helpmanual.io/) worker (separate container in compose) runs cron jobs such as `delete_expired_accounts` and `purge_stale_sessions` at the configured time. - **Audit Trail:** `user_activity` table records auth events and CRUD actions with IP / user agent, plus the **HTTP status code** of each event (`200` on success, the raised error code on failure) — filterable in the admin panel. The `audit_unexpected_failure` decorator captures unexpected route failures. - **Smart Email Validation & Delivery:** Built-in asynchronous email sending with SMTP, domain MX record checking using `dnspython`, and auto-updating disposable email provider filtering via Redis cache. - **Standardized API Responses:** Global exception handlers standardizing success/error schemas, utilizing a centralized `messages` module (`app/core/messages/`) to prevent hardcoded responses. - **Role-Based Access Control (RBAC):** Three roles — `user` / `admin` / `superadmin`. Superadmins bypass every check and **create/delete admins** and assign their permissions; plain admins are gated per **`resource:action`** permission (`users:read`, `users:write`, `support:update`, …) stored in `admin_permission` and enforced by a single `require_permission(...)` dependency. The **superadmin tier** (promote an admin to superadmin, demote a superadmin, transfer root via email OTP) is reserved for the **root superadmin** (`is_root_superadmin`); the last active superadmin is always protected. See [Admin & RBAC](#-admin--rbac-roles--permissions). - **Support / Ticketing System:** Full ticket lifecycle — open, reply with image attachments, status/priority, admin assignment — fanned out over a **Redis-backed WebSocket bus** so the ticket thread, the admin queue, and per-user feeds update live across workers. See [Support System & Realtime](#-support-ticketing-system--realtime). - **Notification System:** Persistent in-app notifications (`notification` table) layered on the same realtime bus — domain events (a support reply, a ticket status change, an RBAC grant change) call a single `notify()` use case that stores the row **and** pushes it to the recipient's `notifications:{id}` feed, so offline users still find it in their inbox on the next fetch. Notification copy is stored as a stable machine `type` + `data` payload (no human text), so the frontend renders it in any locale. REST covers paginated listing (`unread_only` filter), the unread badge count, and mark-(all-)read. See [Notification System](#-notification-system). +- **Session / Device Management:** Every login opens a `user_session` row whose id rides in both tokens as the JWT `sid` claim, so a user can list their active devices and revoke any one (or all but the current). Refresh **rotates** the session's `jti` and detects **replay** — a stale refresh token revokes the whole session as compromised; logout and password change revoke sessions too. Revocation kills a session's still-valid access tokens instantly via a Redis `sid` flag, and broadcasts `sessions_revoked` so open tabs drop to login live. Admins can list/terminate a user's sessions, gated by the `users:sessions` permission. A nightly arq job purges stale rows. See [Session Management](#-session--device-management). - **First Superadmin Seed:** On startup a **root superadmin** (`is_root_superadmin`) is created from `FIRST_SUPERUSER` / `FIRST_SUPERUSER_PASSWORD` if none exists (a matching existing account is promoted instead; older deployments get their oldest superadmin flagged as root) — guaranteeing a root account. - **Tooling:** [uv](https://docs.astral.sh/uv/) for blazing-fast package management, and [Ruff](https://docs.astral.sh/ruff/) for linting and formatting. - **Testing:** Comprehensive async testing setup with `pytest` and `pytest-asyncio`, in-memory SQLite via `aiosqlite`, `fakeredis`, and autouse SMTP/MX patches — tests never hit Postgres or the network. @@ -380,7 +381,7 @@ A complete support desk built on a generic realtime bus that any feature can reu | `WS /admin/support/ws` · `/support/ws` | admin / owner | Multiplexed ticket + queue feed. | | `WS /users/me/events` | any user | Per-user account feed (e.g. live `permissions_updated`). | -**Realtime bus (`app/core/realtime.py`).** A thin in-process `ConnectionManager` sits behind a Redis pub/sub bridge: services `publish()` to a topic — `ticket:{id}`, `admin`, `user:{id}`, `account:{id}`, or `notifications:{id}` — and a per-worker listener fans the event to the matching local sockets, so an event raised in one worker reaches clients connected to any worker. Publishing is best-effort (`publish_safe`), so a realtime hiccup never breaks the originating request. +**Realtime bus (`app/core/realtime.py`).** A thin in-process `ConnectionManager` sits behind a Redis pub/sub bridge: services `publish()` to a topic — `ticket:{id}`, `admin`, `user:{id}`, `account:{id}`, or `notifications:{id}` — and a per-worker listener fans the event to the matching local sockets, so an event raised in one worker reaches clients connected to any worker. The `account:{id}` topic carries both `permissions_updated` (RBAC grant change) and `sessions_revoked` (a device was logged out). Publishing is best-effort (`publish_safe`), so a realtime hiccup never breaks the originating request. --- @@ -402,6 +403,26 @@ Persistent, per-user in-app notifications built on the same realtime bus as supp --- +## 🔒 Session / Device Management + +Per-device login sessions backed by the `user_session` table, so a user can see where they're signed in and revoke access remotely. Each login mints an access + refresh pair carrying the session id as the JWT `sid` claim; the row records the device user-agent (parsed into browser/OS at the schema layer — the raw IP stays server-side and is never returned). + +| Method & path | Auth | Purpose | +| --- | --- | --- | +| `GET /users/me/sessions` | self | List own active sessions (paginated), current device flagged. | +| `DELETE /users/me/sessions/{id}` | owner | Revoke one of own sessions (foreign / unknown id → uniform 404). | +| `DELETE /users/me/sessions` | self | Revoke every session **except** the current device. | +| `GET /admin/users/{id}/sessions` | `users:sessions` | List a user's active sessions (admin view). | +| `DELETE /admin/users/{id}/sessions` | `users:sessions` | Terminate all of a user's sessions (remote logout). | + +**Rotation & replay (`app/services/auth_service.py`).** Refresh validates the `sid`, rotates the session's stored `refresh_jti`, and pushes the expiry forward. A presented refresh token whose `jti` no longer matches the row is a **replay** — the whole session is revoked as compromised and audited, so a leaked token is single-use even if the Redis blacklist ever loses state. Legacy tokens minted before the feature (no `sid`) are rejected on refresh; the holder simply logs in again. + +**Instant revocation.** Revoking a session blacklists its tokens **and** writes a `revoked:session:{sid}` Redis flag (TTL = access-token lifetime), which `get_current_user` checks on every request — so a still-valid access token dies the moment its session is killed, without a per-request DB hit. The same paths broadcast `sessions_revoked` on the `account:{id}` topic so a kicked device's open tab drops to login live. Logout, password change (revokes **all** of the user's sessions), and admin termination all flow through here. + +**Housekeeping.** A nightly arq cron (`app/worker/jobs/purge_stale_sessions.py`) deletes expired rows immediately and revoked rows after a retention window (`SESSION_REVOKED_RETENTION_DAYS`, default 30) in one set-based `DELETE`. + +--- + ## 📂 Project Structure ```bash @@ -413,7 +434,7 @@ Persistent, per-user in-app notifications built on the same realtime bus as supp │ │ └── admin/ # Admin surface — RBAC-gated via require_permission; │ │ # admins/ (admin management) is superadmin-only │ ├── core/ # config, db, security, redis, rate_limit, email, storage, realtime, bootstrap, messages/ -│ ├── models/ # Domain Layer: SQLAlchemy ORM (User, UserActivity, File, SupportTicket, Notification, AdminPermission, …) +│ ├── models/ # Domain Layer: SQLAlchemy ORM (User, UserSession, UserActivity, File, SupportTicket, Notification, AdminPermission, …) │ ├── repositories/ # Data Layer: async DB queries (no business rules) │ ├── schemas/ # Pydantic v2 DTOs (Create / Update / Response per domain) │ ├── services/ # Business Logic Layer: pure async functions, take AsyncSession diff --git a/app/alembic/versions/19b42a1d4b20_add_user_session_table.py b/app/alembic/versions/19b42a1d4b20_add_user_session_table.py new file mode 100644 index 0000000..a85b684 --- /dev/null +++ b/app/alembic/versions/19b42a1d4b20_add_user_session_table.py @@ -0,0 +1,56 @@ +"""add user_session table + +Revision ID: 19b42a1d4b20 +Revises: 5cd8e80e6bce +Create Date: 2026-06-13 00:23:35.196935 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "19b42a1d4b20" +down_revision: str | Sequence[str] | None = "5cd8e80e6bce" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "user_session", + sa.Column("id", sa.UUID(), nullable=False), + sa.Column("user_id", sa.Uuid(), nullable=False), + sa.Column("refresh_jti", sa.String(length=36), nullable=False), + sa.Column("user_agent", sa.String(length=512), nullable=True), + sa.Column("ip_address", sa.String(length=45), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + "ix_user_session_expires_at", "user_session", ["expires_at"], unique=False + ) + op.create_index( + "ix_user_session_user_last_used", + "user_session", + ["user_id", "last_used_at"], + unique=False, + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index("ix_user_session_user_last_used", table_name="user_session") + op.drop_index("ix_user_session_expires_at", table_name="user_session") + op.drop_table("user_session") + # ### end Alembic commands ### diff --git a/app/api/deps.py b/app/api/deps.py index 56ac1c5..e0e24e6 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -10,11 +10,12 @@ from app.core.config import settings from app.core.db import AsyncSessionLocal from app.core.messages.error_message import ErrorMessages -from app.core.security import verify_token +from app.core.security import decode_token_payload from app.models.user import User from app.repositories.admin.permission import has_permission from app.repositories.token_blacklist import is_token_blacklisted from app.repositories.user import get_user_by_id +from app.repositories.user_session import is_session_revoked from app.schemas.admin_permission import Permission from app.schemas.token import TokenPayload from app.schemas.user import SystemRole @@ -57,15 +58,27 @@ async def get_current_user( headers={"WWW-Authenticate": "Bearer"}, ) - token_subject = verify_token(token) - if token_subject is None: + claims = decode_token_payload(token) + if claims is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail=ErrorMessages.INVALID_TOKEN, headers={"WWW-Authenticate": "Bearer"}, ) - token_data = TokenPayload(sub=token_subject) + # Session guard: a token whose session was revoked (logout elsewhere, + # admin kill, password change) dies here even though its signature and + # expiry are still valid. Tokens without a sid (minted before the + # session feature) pass — they expire within minutes on their own. + sid = claims.get("sid") + if sid and await is_session_revoked(sid): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail=ErrorMessages.INVALID_TOKEN, + headers={"WWW-Authenticate": "Bearer"}, + ) + + token_data = TokenPayload(sub=claims["sub"], sid=sid, jti=claims.get("jti")) except (ValidationError, ValueError): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, @@ -132,10 +145,13 @@ async def get_ws_user(websocket: WebSocket, db: AsyncSession) -> User | None: try: if await is_token_blacklisted(token): return None - token_subject = verify_token(token) - if token_subject is None: + claims = decode_token_payload(token) + if claims is None: + return None + sid = claims.get("sid") + if sid and await is_session_revoked(sid): return None - token_data = TokenPayload(sub=token_subject) + token_data = TokenPayload(sub=claims["sub"], sid=sid, jti=claims.get("jti")) except (ValidationError, ValueError): return None @@ -145,6 +161,29 @@ async def get_ws_user(websocket: WebSocket, db: AsyncSession) -> User | None: return user +async def get_current_session_id( + request: Request, + bearer_token: Annotated[str | None, Depends(reusable_oauth2)] = None, +) -> uuid.UUID | None: + """Resolve the ``sid`` claim of the caller's access token. + + Pure claim extraction — authentication itself is the job of + ``get_current_user``, which the route must also depend on. Returns ``None`` + for legacy tokens minted before the session feature. + """ + token = request.cookies.get("access_token") or bearer_token + if not token: + return None + claims = decode_token_payload(token) + sid = claims.get("sid") if claims else None + if not sid: + return None + try: + return uuid.UUID(sid) + except ValueError: + return None + + # Type aliases for dependency injection SessionDep = Annotated[AsyncSession, Depends(get_db)] # reusable_oauth2 uses auto_error=False, so the dependency may resolve to None. @@ -153,6 +192,7 @@ async def get_ws_user(websocket: WebSocket, db: AsyncSession) -> User | None: CurrentActiveUser = Annotated[User, Depends(get_current_active_user)] CurrentAdminUser = Annotated[User, Depends(get_current_admin_user)] CurrentSuperAdmin = Annotated[User, Depends(get_current_superadmin)] +CurrentSessionId = Annotated[uuid.UUID | None, Depends(get_current_session_id)] async def user_has_permission( diff --git a/app/api/routes/admin/users.py b/app/api/routes/admin/users.py index a6edda8..736dfe4 100644 --- a/app/api/routes/admin/users.py +++ b/app/api/routes/admin/users.py @@ -18,6 +18,11 @@ from app.schemas.msg import Message from app.schemas.user import Language, SystemRole from app.schemas.user_activity import ActivityType, ResourceType +from app.schemas.user_session import SessionListResponse, SessionsRevokedResponse +from app.services.admin.session_service import ( + list_user_sessions_admin_service, + revoke_user_sessions_admin_service, +) from app.services.admin.user_service import ( change_password_admin_service, delete_user_admin_service, @@ -38,6 +43,9 @@ AdminUsersPasswordReset = Annotated[ User, Depends(require_permission(Permission.USERS_PASSWORD_RESET)) ] +AdminUsersSessions = Annotated[ + User, Depends(require_permission(Permission.USERS_SESSIONS)) +] AdminUserUpdateAuth = Annotated[ @@ -180,6 +188,48 @@ async def delete_user( ) +@router.get("/{user_id}/sessions", response_model=SessionListResponse) +@audit_unexpected_failure( + activity_type=ActivityType.READ, + resource_type=ResourceType.USER, + endpoint="/admin/users/{user_id}/sessions", +) +async def list_user_sessions( + _request: Request, + _admin: AdminUsersSessions, + session: SessionDep, + user_id: uuid.UUID, + skip: Annotated[int, Query(ge=0)] = 0, + limit: Annotated[int, Query(ge=1, le=100)] = 50, +) -> SessionListResponse: + """List a user's active sessions for the admin detail view.""" + return await list_user_sessions_admin_service( + session, user_id=user_id, skip=skip, limit=limit + ) + + +@router.delete("/{user_id}/sessions", response_model=SessionsRevokedResponse) +@rate_limit_strict("10/minute") +@audit_unexpected_failure( + activity_type=ActivityType.UPDATE, + resource_type=ResourceType.USER, + endpoint="/admin/users/{user_id}/sessions", +) +async def revoke_user_sessions( + request: Request, + current_user: AdminUsersSessions, + session: SessionDep, + user_id: uuid.UUID, +) -> SessionsRevokedResponse: + """Terminate every session of a user — remote logout on all their devices.""" + return await revoke_user_sessions_admin_service( + request, + session, + current_user=current_user, + user_id=user_id, + ) + + @router.post("/{user_id}/change-password", response_model=Message) @rate_limit_strict("5/minute") @audit_unexpected_failure( diff --git a/app/api/routes/users.py b/app/api/routes/users.py index a4fbe81..ebb7470 100644 --- a/app/api/routes/users.py +++ b/app/api/routes/users.py @@ -1,7 +1,16 @@ -from fastapi import APIRouter, Request, Response, WebSocket +import uuid +from typing import Annotated + +from fastapi import APIRouter, Query, Request, Response, WebSocket from app.api.decorators import audit_unexpected_failure -from app.api.deps import CurrentActiveUser, CurrentUser, SessionDep, get_ws_user +from app.api.deps import ( + CurrentActiveUser, + CurrentSessionId, + CurrentUser, + SessionDep, + get_ws_user, +) from app.core.config import settings from app.core.messages.success_message import SuccessMessages from app.core.rate_limit import rate_limit_strict @@ -14,6 +23,12 @@ UserUpdateResponse, ) from app.schemas.user_activity import ActivityType, ResourceType +from app.schemas.user_session import SessionListResponse, SessionsRevokedResponse +from app.services.session_service import ( + list_my_sessions_service, + revoke_my_other_sessions_service, + revoke_my_session_service, +) from app.services.user_service import ( build_user_me_service, deactivate_own_account_service, @@ -49,6 +64,62 @@ async def account_events_ws(websocket: WebSocket, session: SessionDep) -> None: await serve_account_socket(websocket, topic=account_topic(user.id)) +@router.get("/me/sessions", response_model=SessionListResponse) +async def list_my_sessions( + session: SessionDep, + current_user: CurrentActiveUser, + current_session_id: CurrentSessionId, + skip: Annotated[int, Query(ge=0)] = 0, + limit: Annotated[int, Query(ge=1, le=100)] = 50, +) -> SessionListResponse: + """List the caller's active sessions, flagging the current device.""" + return await list_my_sessions_service( + session, + user=current_user, + current_session_id=current_session_id, + skip=skip, + limit=limit, + ) + + +@router.delete("/me/sessions", response_model=SessionsRevokedResponse) +@rate_limit_strict("5/minute") +@audit_unexpected_failure( + activity_type=ActivityType.UPDATE, + resource_type=ResourceType.AUTH, + endpoint="/users/me/sessions (revoke others)", +) +async def revoke_my_other_sessions( + request: Request, + session: SessionDep, + current_user: CurrentActiveUser, + current_session_id: CurrentSessionId, +) -> SessionsRevokedResponse: + """Log the caller out of every device except the one making this request.""" + return await revoke_my_other_sessions_service( + request, session, user=current_user, current_session_id=current_session_id + ) + + +@router.delete("/me/sessions/{session_id}", response_model=Message) +@rate_limit_strict("10/minute") +@audit_unexpected_failure( + activity_type=ActivityType.UPDATE, + resource_type=ResourceType.AUTH, + endpoint="/users/me/sessions/{id}", +) +async def revoke_my_session( + request: Request, + session: SessionDep, + current_user: CurrentActiveUser, + session_id: uuid.UUID, +) -> Message: + """Revoke a single session of the caller (remote logout for one device).""" + return await revoke_my_session_service( + request, session, user=current_user, session_id=session_id + ) + + @router.patch("/me", response_model=UserUpdateResponse) @audit_unexpected_failure( activity_type=ActivityType.UPDATE, diff --git a/app/core/config.py b/app/core/config.py index 911e770..6845ca0 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -126,6 +126,12 @@ def trusted_hosts(self) -> list[str]: DELETION_JOB_CRON_HOUR: int = 3 DELETION_JOB_CRON_MINUTE: int = 0 + # Session housekeeping: revoked rows are kept this long for audit trails, + # then swept (expired rows go immediately) by the nightly purge job. + SESSION_REVOKED_RETENTION_DAYS: int = 30 + SESSION_PURGE_CRON_HOUR: int = 4 + SESSION_PURGE_CRON_MINUTE: int = 0 + # Database connection pool (tuned per API worker) DB_POOL_SIZE: int = 20 DB_MAX_OVERFLOW: int = 10 diff --git a/app/core/messages/error_message.py b/app/core/messages/error_message.py index ee7e5c7..c45c57e 100644 --- a/app/core/messages/error_message.py +++ b/app/core/messages/error_message.py @@ -65,6 +65,9 @@ class ErrorMessages: NOTIFICATION_NOT_FOUND = "error.notification.not_found" NOTIFICATION_ACCESS_DENIED = "error.notification.access_denied" + # Sessions + SESSION_NOT_FOUND = "error.session.not_found" + # Support / tickets TICKET_NOT_FOUND = "error.support.ticket_not_found" TICKET_ACCESS_DENIED = "error.support.ticket_access_denied" diff --git a/app/core/messages/success_message.py b/app/core/messages/success_message.py index 57ed778..e3a3642 100644 --- a/app/core/messages/success_message.py +++ b/app/core/messages/success_message.py @@ -17,6 +17,11 @@ class SuccessMessages: ACCOUNT_DEACTIVATED = "success.account.deactivated" ACCOUNT_REACTIVATED = "success.account.reactivated" + # Sessions + SESSION_REVOKED = "success.session.revoked" + OTHER_SESSIONS_REVOKED = "success.session.others_revoked" + ADMIN_SESSIONS_REVOKED = "success.session.admin_revoked" + # Admin ADMIN_USER_UPDATED = "success.admin.user_updated" ADMIN_USER_SUSPENDED = "success.admin.user_suspended" diff --git a/app/core/security.py b/app/core/security.py index ab5ff86..cfefb48 100644 --- a/app/core/security.py +++ b/app/core/security.py @@ -15,16 +15,28 @@ # --------------------------------------------------------------------------- -def _base_payload(subject: str, token_type: str, expire: datetime) -> dict: - """Build a standard JWT payload with UTC unix timestamps.""" +def _base_payload( + subject: str, + token_type: str, + expire: datetime, + session_id: str | uuid.UUID | None = None, +) -> dict: + """Build a standard JWT payload with UTC unix timestamps. + + ``session_id`` becomes the ``sid`` claim binding the token to a + ``UserSession`` row, so revoking the session invalidates the token. + """ now = utc_now() - return { + payload = { "sub": subject, "type": token_type, "jti": str(uuid.uuid4()), "iat": now.timestamp(), "exp": expire.timestamp(), } + if session_id is not None: + payload["sid"] = str(session_id) + return payload # --------------------------------------------------------------------------- @@ -35,6 +47,8 @@ def _base_payload(subject: str, token_type: str, expire: datetime) -> dict: def create_access_token( subject: str | uuid.UUID, expires_delta: timedelta | None = None, + *, + session_id: str | uuid.UUID | None = None, ) -> str: """ Create a short-lived JWT access token. @@ -42,6 +56,7 @@ def create_access_token( Args: subject: User ID (str or UUID) expires_delta: Optional custom TTL + session_id: UserSession id embedded as the ``sid`` claim Returns: Encoded JWT string @@ -49,13 +64,15 @@ def create_access_token( expire = utc_now() + ( expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) ) - payload = _base_payload(str(subject), "access", expire) + payload = _base_payload(str(subject), "access", expire, session_id=session_id) return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256") def create_refresh_token( subject: str | uuid.UUID, expires_delta: timedelta | None = None, + *, + session_id: str | uuid.UUID | None = None, ) -> str: """ Create a long-lived JWT refresh token. @@ -65,6 +82,7 @@ def create_refresh_token( Args: subject: User ID (str or UUID) expires_delta: Optional custom TTL + session_id: UserSession id embedded as the ``sid`` claim Returns: Encoded JWT string @@ -72,29 +90,48 @@ def create_refresh_token( expire = utc_now() + ( expires_delta or timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS) ) - payload = _base_payload(str(subject), "refresh", expire) + payload = _base_payload(str(subject), "refresh", expire, session_id=session_id) return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256") -def verify_token(token: str, expected_type: str = "access") -> str | None: +def decode_token_payload(token: str, expected_type: str = "access") -> dict | None: """ - Decode and validate a JWT token. + Decode and validate a JWT, returning its full claims. + + Use this when the caller needs more than the subject — e.g. the ``sid`` + session binding or the ``jti`` for rotation/replay checks. Args: token: Raw JWT string expected_type: ``"access"`` or ``"refresh"`` Returns: - Subject (user ID) on success, ``None`` otherwise + Claims dict on success, ``None`` otherwise """ try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"]) - if payload.get("type") != expected_type: - return None - subject: str | None = payload.get("sub") - return subject if subject else None except jwt.PyJWTError: return None + if payload.get("type") != expected_type: + return None + if not payload.get("sub"): + return None + return payload + + +def verify_token(token: str, expected_type: str = "access") -> str | None: + """ + Decode and validate a JWT token. + + Args: + token: Raw JWT string + expected_type: ``"access"`` or ``"refresh"`` + + Returns: + Subject (user ID) on success, ``None`` otherwise + """ + payload = decode_token_payload(token, expected_type) + return payload["sub"] if payload else None def verify_refresh_token(token: str) -> str | None: diff --git a/app/models/__init__.py b/app/models/__init__.py index b0d17c3..bf988b2 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -8,6 +8,7 @@ ) from app.models.user import User from app.models.user_activity import UserActivity +from app.models.user_session import UserSession __all__ = [ "AdminPermission", @@ -18,4 +19,5 @@ "SupportTicket", "User", "UserActivity", + "UserSession", ] diff --git a/app/models/user_session.py b/app/models/user_session.py new file mode 100644 index 0000000..a3bc944 --- /dev/null +++ b/app/models/user_session.py @@ -0,0 +1,65 @@ +import uuid +from datetime import datetime +from typing import TYPE_CHECKING + +from sqlalchemy import DateTime, ForeignKey, Index, String +from sqlalchemy.dialects.postgresql import UUID as PG_UUID +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.core.db import Base +from app.utils import utc_now + +if TYPE_CHECKING: + from app.models.user import User + + +class UserSession(Base): + """A login session tied to one device/browser. + + The row's ``id`` is the ``sid`` claim embedded in both the access and the + refresh token, so a revoked session can invalidate its access tokens + without a DB lookup (Redis guard keyed by sid). ``refresh_jti`` always + holds the jti of the *latest* refresh token issued for this session; on + rotation it is overwritten, so a replayed (older) refresh token no longer + matches and the whole session is revoked as compromised. + """ + + __tablename__ = "user_session" + __table_args__ = ( + # "My active sessions, most recently used first". + Index("ix_user_session_user_last_used", "user_id", "last_used_at"), + # Purge job scan: expired rows are deleted regardless of revocation. + Index("ix_user_session_expires_at", "expires_at"), + ) + + id: Mapped[uuid.UUID] = mapped_column( + PG_UUID(as_uuid=True), primary_key=True, default=uuid.uuid4 + ) + user_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("user.id", ondelete="CASCADE"), + nullable=False, + ) + # jti of the most recent refresh token issued for this session (rotates). + refresh_jti: Mapped[str] = mapped_column(String(36), nullable=False) + # Raw User-Agent header; parsed into browser/OS at the schema layer so no + # migration is needed when the parsing heuristics improve. + user_agent: Mapped[str | None] = mapped_column(String(512), default=None) + # Text form, sized for IPv6. + ip_address: Mapped[str | None] = mapped_column(String(45), default=None) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=utc_now, nullable=False + ) + # Bumped on every refresh rotation; powers "last active" in the UI. + last_used_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=utc_now, nullable=False + ) + # Mirrors the current refresh token's expiry; pushed forward on rotation. + expires_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False + ) + # Set on logout, manual revoke, or detected refresh-token replay. + revoked_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), default=None + ) + + user: Mapped["User"] = relationship("User") diff --git a/app/repositories/user_session.py b/app/repositories/user_session.py new file mode 100644 index 0000000..482d85f --- /dev/null +++ b/app/repositories/user_session.py @@ -0,0 +1,192 @@ +import uuid +from collections.abc import Iterable, Sequence +from datetime import datetime, timedelta + +from sqlalchemy import delete, func, or_, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.config import settings +from app.core.redis import get_redis +from app.models.user_session import UserSession +from app.utils import utc_now + + +async def create_session( + session: AsyncSession, user_session: UserSession +) -> UserSession: + """Persist a new login session.""" + session.add(user_session) + await session.commit() + await session.refresh(user_session) + return user_session + + +async def get_session_by_id( + session: AsyncSession, session_id: uuid.UUID +) -> UserSession | None: + """Get a single session row by id (the token ``sid`` claim).""" + return await session.get(UserSession, session_id) + + +async def list_active_sessions( + session: AsyncSession, + *, + user_id: uuid.UUID, + skip: int = 0, + limit: int = 50, +) -> tuple[Sequence[UserSession], int]: + """Return a page of a user's live sessions plus the total count. + + A session is live when it has not been revoked and its refresh token has + not expired yet. Most recently used first. + """ + base_stmt = select(UserSession).where( + UserSession.user_id == user_id, + UserSession.revoked_at.is_(None), + UserSession.expires_at > utc_now(), + ) + + count_stmt = base_stmt.with_only_columns( + func.count(), maintain_column_froms=True + ).order_by(None) + total = (await session.execute(count_stmt)).scalar_one() + + list_stmt = ( + base_stmt.order_by(UserSession.last_used_at.desc()).offset(skip).limit(limit) + ) + rows = (await session.execute(list_stmt)).scalars().all() + return rows, total + + +async def rotate_session_jti( + session: AsyncSession, + *, + session_id: uuid.UUID, + refresh_jti: str, + expires_at: datetime, +) -> None: + """Record a refresh rotation: new jti, fresh expiry, bumped last-used. + + Set-based UPDATE so the rotation never races with a concurrent read of a + possibly-expired ORM instance. + """ + statement = ( + update(UserSession) + .where(UserSession.id == session_id) + .values(refresh_jti=refresh_jti, expires_at=expires_at, last_used_at=utc_now()) + ) + await session.execute(statement) + await session.commit() + + +async def revoke_session( + session: AsyncSession, *, session_id: uuid.UUID +) -> UserSession | None: + """Stamp a session as revoked and return the updated row. + + The ``revoked_at IS NULL`` guard makes a double revoke idempotent (the + original revocation time is kept). Returns ``None`` when no such session + exists. + """ + statement = ( + update(UserSession) + .where( + UserSession.id == session_id, + UserSession.revoked_at.is_(None), + ) + .values(revoked_at=utc_now()) + ) + await session.execute(statement) + await session.commit() + return await session.get(UserSession, session_id) + + +async def revoke_all_sessions( + session: AsyncSession, + *, + user_id: uuid.UUID, + except_session_id: uuid.UUID | None = None, +) -> Sequence[UserSession]: + """Revoke every live session of a user and return the rows just revoked. + + ``except_session_id`` keeps the caller's own session alive ("log out other + devices"). The revoked rows are returned so the service layer can also + blacklist their refresh jtis and flag their sids in Redis. + """ + conditions = [ + UserSession.user_id == user_id, + UserSession.revoked_at.is_(None), + UserSession.expires_at > utc_now(), + ] + if except_session_id is not None: + conditions.append(UserSession.id != except_session_id) + + victims = ( + (await session.execute(select(UserSession).where(*conditions))).scalars().all() + ) + if not victims: + return [] + + statement = ( + update(UserSession) + .where(UserSession.id.in_([v.id for v in victims])) + .values(revoked_at=utc_now()) + ) + await session.execute(statement) + await session.commit() + return victims + + +async def purge_stale_sessions( + session: AsyncSession, *, revoked_retention_days: int = 30 +) -> int: + """Delete sessions that are expired or were revoked long ago. + + Recently revoked rows are kept for ``revoked_retention_days`` so the user + can still see "logged out" history if the UI ever wants it; expired rows + carry no value and are dropped immediately. Returns the rows deleted. + """ + cutoff = utc_now() - timedelta(days=revoked_retention_days) + statement = delete(UserSession).where( + or_( + UserSession.expires_at < utc_now(), + UserSession.revoked_at < cutoff, + ) + ) + result = await session.execute(statement) + await session.commit() + return result.rowcount + + +# --------------------------------------------------------------------------- +# Redis revoked-sid flags +# +# The DB row is the source of truth for the *refresh* flow (which loads it +# anyway), but checking the DB on every API request would be too costly. These +# flags let ``get_current_user`` kill the access tokens of a revoked session +# with a single Redis lookup. TTL = access-token lifetime: past that, every +# access token carrying the sid has expired on its own. +# --------------------------------------------------------------------------- + +_REVOKED_SID_PREFIX = "revoked:session:" + + +def _revoked_key(session_id: str | uuid.UUID) -> str: + """Build the Redis key flagging a revoked session id.""" + return f"{_REVOKED_SID_PREFIX}{session_id}" + + +async def flag_sessions_revoked(session_ids: Iterable[str | uuid.UUID]) -> None: + """Flag sids in Redis so their live access tokens die immediately.""" + redis = get_redis() + ttl = settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60 + async with redis.pipeline(transaction=False) as pipe: + for session_id in session_ids: + pipe.set(_revoked_key(session_id), "1", ex=ttl) + await pipe.execute() + + +async def is_session_revoked(session_id: str | uuid.UUID) -> bool: + """Return True if the session id has been flagged as revoked.""" + redis = get_redis() + return bool(await redis.exists(_revoked_key(session_id))) diff --git a/app/schemas/account.py b/app/schemas/account.py index be72bbb..65f3183 100644 --- a/app/schemas/account.py +++ b/app/schemas/account.py @@ -7,6 +7,10 @@ class AccountEventType(StrEnum): """Realtime event types pushed to a single user's account socket.""" PERMISSIONS_UPDATED = "permissions_updated" + # One or more of the user's sessions were revoked (remote logout, admin + # kill, password change). Clients re-validate their own session and drop + # to the login screen if it was theirs. + SESSIONS_REVOKED = "sessions_revoked" class AccountEvent(BaseModel): diff --git a/app/schemas/admin_permission.py b/app/schemas/admin_permission.py index 7ff0647..b5b4bd8 100644 --- a/app/schemas/admin_permission.py +++ b/app/schemas/admin_permission.py @@ -16,6 +16,7 @@ class Permission(StrEnum): USERS_DELETE = "users:delete" USERS_SUSPEND = "users:suspend" USERS_PASSWORD_RESET = "users:password_reset" + USERS_SESSIONS = "users:sessions" FILES_READ = "files:read" FILES_DELETE = "files:delete" SUPPORT_READ = "support:read" diff --git a/app/schemas/token.py b/app/schemas/token.py index 5251ce1..34d0768 100644 --- a/app/schemas/token.py +++ b/app/schemas/token.py @@ -40,3 +40,7 @@ class CookieRefreshResponse(BaseModel): # Contents of JWT token class TokenPayload(BaseModel): sub: str | None = None + # Session binding (UserSession id) — present on tokens minted via login. + sid: str | None = None + # Unique token id; the refresh flow matches it against the session row. + jti: str | None = None diff --git a/app/schemas/user_session.py b/app/schemas/user_session.py new file mode 100644 index 0000000..d9af9f0 --- /dev/null +++ b/app/schemas/user_session.py @@ -0,0 +1,56 @@ +import uuid +from datetime import datetime + +from pydantic import BaseModel + +from app.models.user_session import UserSession +from app.utils.user_agent import parse_user_agent + + +class SessionRead(BaseModel): + """One active login session as shown on the security screen. + + ``browser``/``os`` are parsed server-side from the stored User-Agent so + every client renders the same labels without shipping a UA parser. The + stored IP address is deliberately NOT exposed — it stays in the DB for + audit purposes only. + """ + + id: uuid.UUID + browser: str | None = None + os: str | None = None + created_at: datetime + last_used_at: datetime + # True for the session the caller used to make this request. + is_current: bool = False + + @classmethod + def from_model( + cls, user_session: UserSession, *, current_session_id: uuid.UUID | None + ) -> "SessionRead": + """Map an ORM row to the public shape, marking the caller's own session.""" + parsed = parse_user_agent(user_session.user_agent) + return cls( + id=user_session.id, + browser=parsed.browser, + os=parsed.os, + created_at=user_session.created_at, + last_used_at=user_session.last_used_at, + is_current=user_session.id == current_session_id, + ) + + +class SessionListResponse(BaseModel): + """Paginated listing of active sessions, most recently used first.""" + + data: list[SessionRead] + total: int + skip: int + limit: int + + +class SessionsRevokedResponse(BaseModel): + """Standard response after a bulk session revocation.""" + + revoked: int + message: str diff --git a/app/schemas/worker.py b/app/schemas/worker.py index c27e7e0..3084d02 100644 --- a/app/schemas/worker.py +++ b/app/schemas/worker.py @@ -11,3 +11,10 @@ class DeletionJobResult(BaseModel): processed: int = Field(ge=0, description="Users hard-deleted this run.") failed: int = Field(ge=0, description="Users that errored and will be retried.") duration_ms: int = Field(ge=0, description="Total wall-clock time in ms.") + + +class SessionPurgeJobResult(BaseModel): + """Outcome of a single run of the stale-session purge job.""" + + purged: int = Field(ge=0, description="Session rows deleted this run.") + duration_ms: int = Field(ge=0, description="Total wall-clock time in ms.") diff --git a/app/services/admin/session_service.py b/app/services/admin/session_service.py new file mode 100644 index 0000000..98fd0bc --- /dev/null +++ b/app/services/admin/session_service.py @@ -0,0 +1,106 @@ +import uuid + +from fastapi import HTTPException, Request, status +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.messages.error_message import ErrorMessages +from app.core.messages.success_message import SuccessMessages +from app.core.realtime import account_topic, publish_safe +from app.models.user import User +from app.repositories.user import get_user_by_id +from app.repositories.user_session import ( + flag_sessions_revoked, + list_active_sessions, + revoke_all_sessions, +) +from app.schemas.account import AccountEvent, AccountEventType +from app.schemas.user import SystemRole +from app.schemas.user_activity import ActivityType, ResourceType +from app.schemas.user_session import ( + SessionListResponse, + SessionRead, + SessionsRevokedResponse, +) +from app.use_cases.log_activity import log_activity + + +async def _load_target(session: AsyncSession, user_id: uuid.UUID) -> User: + """Fetch the target user or raise 404.""" + target = await get_user_by_id(session, user_id) + if target is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=ErrorMessages.USER_NOT_FOUND, + ) + return target + + +def _guard_superadmin_target(actor: User, target: User) -> None: + """Only superadmins may touch a superadmin's sessions.""" + if ( + target.role == SystemRole.SUPERADMIN.value + and actor.role != SystemRole.SUPERADMIN.value + ): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=ErrorMessages.INSUFFICIENT_PERMISSIONS, + ) + + +async def list_user_sessions_admin_service( + session: AsyncSession, + *, + user_id: uuid.UUID, + skip: int = 0, + limit: int = 50, +) -> SessionListResponse: + """Return a page of a user's live sessions for the admin detail view. + + ``is_current`` is never set here — "current" is meaningful only to the + session owner, not to the admin looking at the list. + """ + await _load_target(session, user_id) + rows, total = await list_active_sessions( + session, user_id=user_id, skip=skip, limit=limit + ) + data = [SessionRead.from_model(row, current_session_id=None) for row in rows] + return SessionListResponse(data=data, total=total, skip=skip, limit=limit) + + +async def revoke_user_sessions_admin_service( + request: Request, + session: AsyncSession, + *, + current_user: User, + user_id: uuid.UUID, +) -> SessionsRevokedResponse: + """Terminate every live session of a user (admin remote logout). + + All of the target's devices lose their access tokens immediately via the + Redis sid flags, and the account socket broadcasts ``sessions_revoked`` so + open tabs drop to the login screen without waiting for the next request. + """ + target = await _load_target(session, user_id) + _guard_superadmin_target(current_user, target) + + revoked = await revoke_all_sessions(session, user_id=target.id) + if revoked: + await flag_sessions_revoked([row.id for row in revoked]) + await publish_safe( + account_topic(target.id), + AccountEvent(type=AccountEventType.SESSIONS_REVOKED), + ) + + await log_activity( + session=session, + user_id=current_user.id, + activity_type=ActivityType.UPDATE, + resource_type=ResourceType.USER, + resource_id=target.id, + details={"action": "admin_revoked_sessions", "count": len(revoked)}, + request=request, + ) + + return SessionsRevokedResponse( + revoked=len(revoked), message=SuccessMessages.ADMIN_SESSIONS_REVOKED + ) diff --git a/app/services/auth_service.py b/app/services/auth_service.py index f08ebd7..91eadb9 100644 --- a/app/services/auth_service.py +++ b/app/services/auth_service.py @@ -1,4 +1,5 @@ import uuid +from datetime import UTC, datetime from fastapi import HTTPException, Request, status from sqlalchemy.ext.asyncio import AsyncSession @@ -7,19 +8,21 @@ from app.core.email import send_email from app.core.messages.error_message import ErrorMessages from app.core.messages.success_message import SuccessMessages +from app.core.realtime import account_topic, publish_safe from app.core.security import ( aget_password_hash, averify_password, create_access_token, create_password_reset_token, create_refresh_token, + decode_token_payload, generate_new_account_token, get_password_hash, verify_new_account_token, verify_password_reset_token, - verify_refresh_token, ) from app.models.user import User +from app.models.user_session import UserSession from app.repositories.login_attempts import ( clear_login_attempts, is_login_locked, @@ -30,6 +33,15 @@ is_token_blacklisted, ) from app.repositories.user import get_user_by_email, get_user_by_id, update_user +from app.repositories.user_session import ( + create_session, + flag_sessions_revoked, + get_session_by_id, + revoke_all_sessions, + revoke_session, + rotate_session_jti, +) +from app.schemas.account import AccountEvent, AccountEventType from app.schemas.msg import Message from app.schemas.token import AuthTokens, RefreshedTokens from app.schemas.user import ( @@ -233,12 +245,50 @@ async def authenticate( return user +def _refresh_claims(refresh_token: str) -> dict: + """Decode a just-minted refresh token's claims (jti + exp for the session row).""" + claims = decode_token_payload(refresh_token, expected_type="refresh") + if claims is None: # pragma: no cover - we minted the token one line above + raise RuntimeError("freshly minted refresh token failed to decode") + return claims + + +async def _open_session( + request: Request | None, + session: AsyncSession, + user: User, + session_id: uuid.UUID, + refresh_token: str, +) -> None: + """Persist the UserSession row backing a freshly issued token pair. + + Device metadata is truncated to the column sizes so a hostile header can + never fail the insert. + """ + claims = _refresh_claims(refresh_token) + user_agent = request.headers.get("user-agent") if request else None + ip_address = request.client.host if request and request.client else None + await create_session( + session, + UserSession( + id=session_id, + user_id=user.id, + refresh_jti=claims["jti"], + user_agent=user_agent[:512] if user_agent else None, + ip_address=ip_address[:45] if ip_address else None, + expires_at=datetime.fromtimestamp(claims["exp"], tz=UTC), + ), + ) + + async def login_service( request: Request, session: AsyncSession, email: str, password: str ) -> AuthTokens: """ Orchestrate the login process: authenticate user and generate JWT tokens. - Simplified token creation relying on security component defaults. + + Every successful login opens a ``UserSession`` row whose id rides in both + tokens as the ``sid`` claim, enabling per-device session management. """ user = await authenticate( request=request, session=session, email=email, password=password @@ -253,9 +303,14 @@ async def login_service( request=request, ) + session_id = uuid.uuid4() + access_token = create_access_token(user.id, session_id=session_id) + refresh_token = create_refresh_token(user.id, session_id=session_id) + await _open_session(request, session, user, session_id, refresh_token) + return AuthTokens( - access_token=create_access_token(user.id), - refresh_token=create_refresh_token(user.id), + access_token=access_token, + refresh_token=refresh_token, user=UserPublic.model_validate(user), message=SuccessMessages.LOGIN_SUCCESS, ) @@ -269,14 +324,19 @@ async def refresh_token_service( On success the presented refresh token is revoked and a fresh access + refresh pair is issued. Replaying the old (now-blacklisted) refresh token - is rejected by the blacklist guard below, so a leaked token is single-use. + is rejected by the blacklist guard below; should the blacklist ever lose + state (Redis flush), the session-jti match catches the replay instead and + revokes the whole session as compromised. """ - user_id = verify_refresh_token(refresh_token) - if not user_id: + claims = decode_token_payload(refresh_token, expected_type="refresh") + # Tokens without a session binding (pre-session deploys) are rejected — + # the holder simply logs in again and gets a session-bound pair. + if not claims or not claims.get("sid"): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail=ErrorMessages.INVALID_TOKEN, ) + user_id = claims["sub"] # Convert user_id to UUID early to use it in logging try: @@ -343,13 +403,60 @@ async def refresh_token_service( detail=ErrorMessages.ACCOUNT_SUSPENDED, ) - # Rotate: revoke the presented refresh token and mint a fresh pair so the - # old token can never be reused (replay is caught by the guard above). + # Session guard: the binding row must still be live. + session_id = uuid.UUID(claims["sid"]) + user_session = await get_session_by_id(session, session_id) + if ( + user_session is None + or user_session.revoked_at is not None + or user_session.user_id != parsed_user_id + ): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail=ErrorMessages.INVALID_TOKEN, + ) + + # Replay guard: a rotated-away jti means this token was already spent. + # Someone is holding a stale copy — kill the whole session. + if user_session.refresh_jti != claims["jti"]: + await revoke_session(session, session_id=session_id) + await flag_sessions_revoked([session_id]) + if request: + await log_activity( + session=session, + user_id=parsed_user_id, + activity_type=ActivityType.LOGIN, + resource_type=ResourceType.AUTH, + status=ActivityStatus.FAILURE, + status_code=status.HTTP_401_UNAUTHORIZED, + details={ + "reason": "refresh_token_replay", + "session_id": str(session_id), + }, + request=request, + ) + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail=ErrorMessages.INVALID_TOKEN, + ) + + # Rotate: revoke the presented refresh token and mint a fresh pair bound + # to the same session, then record the new jti/expiry on the row. await _revoke_token(refresh_token) + new_access_token = create_access_token(user_id, session_id=session_id) + new_refresh_token = create_refresh_token(user_id, session_id=session_id) + new_claims = _refresh_claims(new_refresh_token) + await rotate_session_jti( + session, + session_id=session_id, + refresh_jti=new_claims["jti"], + expires_at=datetime.fromtimestamp(new_claims["exp"], tz=UTC), + ) + return RefreshedTokens( - access_token=create_access_token(user_id), - refresh_token=create_refresh_token(user_id), + access_token=new_access_token, + refresh_token=new_refresh_token, message=SuccessMessages.LOGIN_SUCCESS, ) @@ -364,6 +471,18 @@ async def _revoke_token(token: str | None) -> None: await add_token_to_blacklist(token) +async def _close_session(session: AsyncSession, sid: str | None) -> None: + """Revoke the session row for ``sid`` and flag it in Redis (best-effort).""" + if not sid: + return + try: + session_id = uuid.UUID(sid) + except ValueError: + return + await revoke_session(session, session_id=session_id) + await flag_sessions_revoked([session_id]) + + async def logout_service( request: Request | None, session: AsyncSession, @@ -374,18 +493,28 @@ async def logout_service( Invalidate the session by blacklisting both the refresh and access tokens. Revoking the access token too closes the window where a stolen access - token would otherwise stay valid until its own expiry after logout. + token would otherwise stay valid until its own expiry after logout. The + backing ``UserSession`` row is revoked as well so the device disappears + from the active-sessions list. """ await _revoke_token(access_token) + # Resolve the session binding from whichever token is available. + claims = None + if refresh_token: + claims = decode_token_payload(refresh_token, expected_type="refresh") + if claims is None and access_token: + claims = decode_token_payload(access_token) + if claims: + await _close_session(session, claims.get("sid")) + if refresh_token: await _revoke_token(refresh_token) # Log success if possible - user_id = verify_refresh_token(refresh_token) - if user_id and request: + if claims and request: try: - parsed_user_id = uuid.UUID(user_id) + parsed_user_id = uuid.UUID(claims["sub"]) await log_activity( session=session, user_id=parsed_user_id, @@ -577,9 +706,10 @@ async def change_password_service( """ Change user password after verifying current password. - On success the current session's access and refresh tokens are revoked so - a password change forces re-authentication and immediately invalidates any - token that may have been stolen before the change. + On success every session of the user is revoked — the current one (its + access and refresh tokens are blacklisted, forcing re-authentication) and + every other device, so a credential thief is logged out everywhere the + moment the owner rotates the password. """ if not await averify_password( update_password.current_password, current_user.hashed_password @@ -607,6 +737,18 @@ async def change_password_service( await _revoke_token(access_token) await _revoke_token(refresh_token) + # Kill every session of the account (this device and all others): a + # password rotation must evict anyone holding stolen credentials. The live + # broadcast drops other devices' open tabs to login at once, mirroring the + # session-service revoke paths, instead of waiting for their next request. + revoked = await revoke_all_sessions(session, user_id=current_user.id) + if revoked: + await flag_sessions_revoked([s.id for s in revoked]) + await publish_safe( + account_topic(current_user.id), + AccountEvent(type=AccountEventType.SESSIONS_REVOKED), + ) + await log_activity( session=session, user_id=current_user.id, diff --git a/app/services/session_service.py b/app/services/session_service.py new file mode 100644 index 0000000..137c082 --- /dev/null +++ b/app/services/session_service.py @@ -0,0 +1,123 @@ +import uuid + +from fastapi import HTTPException, Request, status +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.messages.error_message import ErrorMessages +from app.core.messages.success_message import SuccessMessages +from app.core.realtime import account_topic, publish_safe +from app.models.user import User +from app.repositories.user_session import ( + flag_sessions_revoked, + get_session_by_id, + list_active_sessions, + revoke_all_sessions, + revoke_session, +) +from app.schemas.account import AccountEvent, AccountEventType +from app.schemas.msg import Message +from app.schemas.user_activity import ActivityType, ResourceType +from app.schemas.user_session import ( + SessionListResponse, + SessionRead, + SessionsRevokedResponse, +) +from app.use_cases.log_activity import log_activity +from app.utils import ensure_utc, utc_now + + +async def list_my_sessions_service( + session: AsyncSession, + *, + user: User, + current_session_id: uuid.UUID | None, + skip: int = 0, + limit: int = 50, +) -> SessionListResponse: + """Return a page of the caller's live sessions with their own device flagged.""" + rows, total = await list_active_sessions( + session, user_id=user.id, skip=skip, limit=limit + ) + data = [ + SessionRead.from_model(row, current_session_id=current_session_id) + for row in rows + ] + return SessionListResponse(data=data, total=total, skip=skip, limit=limit) + + +async def revoke_my_session_service( + request: Request, + session: AsyncSession, + *, + user: User, + session_id: uuid.UUID, +) -> Message: + """Revoke one of the caller's sessions (their current one included). + + A session that does not exist, belongs to someone else, or is already + revoked/expired uniformly yields 404 so the endpoint never confirms + foreign session ids. + """ + row = await get_session_by_id(session, session_id) + if ( + row is None + or row.user_id != user.id + or row.revoked_at is not None + or ensure_utc(row.expires_at) <= utc_now() + ): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=ErrorMessages.SESSION_NOT_FOUND, + ) + + await revoke_session(session, session_id=session_id) + await flag_sessions_revoked([session_id]) + # Live signal so an open tab on the kicked device drops to login at once + # instead of dying on its next API call. + await publish_safe( + account_topic(user.id), + AccountEvent(type=AccountEventType.SESSIONS_REVOKED), + ) + + await log_activity( + session=session, + user_id=user.id, + activity_type=ActivityType.UPDATE, + resource_type=ResourceType.AUTH, + details={"action": "session_revoked", "session_id": str(session_id)}, + request=request, + ) + + return Message(success=True, message=SuccessMessages.SESSION_REVOKED) + + +async def revoke_my_other_sessions_service( + request: Request, + session: AsyncSession, + *, + user: User, + current_session_id: uuid.UUID | None, +) -> SessionsRevokedResponse: + """Revoke every session of the caller except the one making this request.""" + revoked = await revoke_all_sessions( + session, user_id=user.id, except_session_id=current_session_id + ) + if revoked: + await flag_sessions_revoked([row.id for row in revoked]) + await publish_safe( + account_topic(user.id), + AccountEvent(type=AccountEventType.SESSIONS_REVOKED), + ) + + await log_activity( + session=session, + user_id=user.id, + activity_type=ActivityType.UPDATE, + resource_type=ResourceType.AUTH, + details={"action": "other_sessions_revoked", "count": len(revoked)}, + request=request, + ) + + return SessionsRevokedResponse( + revoked=len(revoked), message=SuccessMessages.OTHER_SESSIONS_REVOKED + ) diff --git a/app/tests/test_sessions.py b/app/tests/test_sessions.py new file mode 100644 index 0000000..b0d6ed5 --- /dev/null +++ b/app/tests/test_sessions.py @@ -0,0 +1,421 @@ +"""Tests for the session/device management feature. + +Covers the user endpoints (/users/me/sessions), the session binding inside +the auth flows (login/refresh/logout/change-password), the admin endpoints +with RBAC, and the stale-session purge worker job. +""" + +from __future__ import annotations + +from datetime import timedelta +from uuid import UUID, uuid4 + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.core.config import settings +from app.core.messages.error_message import ErrorMessages +from app.core.messages.success_message import SuccessMessages +from app.core.security import create_refresh_token, get_password_hash +from app.main import app +from app.models.user import User +from app.models.user_session import UserSession +from app.schemas.admin_permission import Permission +from app.tests.admin.conftest import ( + grant_permissions, + promote_to_admin, + register_and_verify, +) +from app.tests.conftest import TestingSessionLocal +from app.utils import utc_now + +PASSWORD = "Password123!" + +_UA_WINDOWS = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/125.0 Safari/537.36" +_UA_IPHONE = "Mozilla/5.0 (iPhone; CPU iPhone OS 17_4) CriOS/125.0 Mobile Safari/604.1" + + +def _fresh_client() -> AsyncClient: + """Cookie-isolated client simulating a separate device.""" + return AsyncClient( + transport=ASGITransport(app=app), base_url=f"http://test{settings.API_V1_STR}" + ) + + +async def _login(client: AsyncClient, email: str, *, user_agent: str) -> None: + """Login on the given client with a specific device user-agent.""" + resp = await client.post( + "/auth/login", + data={"username": email, "password": PASSWORD}, + headers={"user-agent": user_agent}, + ) + assert resp.status_code == 200 + + +async def _user_id(email: str) -> UUID: + """Look up a user's id directly in the DB.""" + async with TestingSessionLocal() as session: + user = ( + (await session.execute(select(User).where(User.email == email))) + .scalars() + .one() + ) + return user.id + + +# --------------------------------------------------------------------------- +# /users/me/sessions +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_login_creates_session_with_device_metadata(client: AsyncClient): + await register_and_verify(client, "sess1@test.com") + await _login(client, "sess1@test.com", user_agent=_UA_WINDOWS) + + resp = await client.get("/users/me/sessions") + assert resp.status_code == 200 + body = resp.json() + assert body["total"] == 1 and body["skip"] == 0 and body["limit"] == 50 + row = body["data"][0] + assert row["is_current"] is True + assert row["browser"] == "Chrome" + assert row["os"] == "Windows" + # The stored IP must never leave the backend. + assert "ip_address" not in row + + +@pytest.mark.asyncio +async def test_list_sessions_pagination(client: AsyncClient): + await register_and_verify(client, "sess2@test.com") + await _login(client, "sess2@test.com", user_agent=_UA_WINDOWS) + other = _fresh_client() + await _login(other, "sess2@test.com", user_agent=_UA_IPHONE) + await other.aclose() + + resp = await client.get("/users/me/sessions?skip=1&limit=1") + assert resp.status_code == 200 + body = resp.json() + assert body["total"] == 2 + assert len(body["data"]) == 1 + assert body["skip"] == 1 and body["limit"] == 1 + + +@pytest.mark.asyncio +async def test_revoke_single_session_kills_its_access_token(client: AsyncClient): + await register_and_verify(client, "sess3@test.com") + await _login(client, "sess3@test.com", user_agent=_UA_WINDOWS) + + other = _fresh_client() + await _login(other, "sess3@test.com", user_agent=_UA_IPHONE) + + listing = (await client.get("/users/me/sessions")).json() + target = next(s for s in listing["data"] if not s["is_current"]) + + resp = await client.delete(f"/users/me/sessions/{target['id']}") + assert resp.status_code == 200 + assert resp.json()["message"] == SuccessMessages.SESSION_REVOKED + + # The other device's access token must die immediately. + dead = await other.get("/users/me") + assert dead.status_code == 401 + await other.aclose() + + +@pytest.mark.asyncio +async def test_revoke_foreign_or_unknown_session_returns_404(client: AsyncClient): + await register_and_verify(client, "sess4@test.com") + await register_and_verify(client, "sess4b@test.com") + await _login(client, "sess4@test.com", user_agent=_UA_WINDOWS) + + victim = _fresh_client() + await _login(victim, "sess4b@test.com", user_agent=_UA_IPHONE) + victim_session = (await victim.get("/users/me/sessions")).json()["data"][0] + + # Someone else's session id and a random id both yield the same 404. + for session_id in (victim_session["id"], str(uuid4())): + resp = await client.delete(f"/users/me/sessions/{session_id}") + assert resp.status_code == 404 + assert resp.json()["error"] == ErrorMessages.SESSION_NOT_FOUND + + # The victim is untouched. + assert (await victim.get("/users/me")).status_code == 200 + await victim.aclose() + + +@pytest.mark.asyncio +async def test_revoke_other_sessions_keeps_current(client: AsyncClient): + await register_and_verify(client, "sess5@test.com") + await _login(client, "sess5@test.com", user_agent=_UA_WINDOWS) + + other1, other2 = _fresh_client(), _fresh_client() + await _login(other1, "sess5@test.com", user_agent=_UA_IPHONE) + await _login(other2, "sess5@test.com", user_agent=_UA_IPHONE) + + resp = await client.delete("/users/me/sessions") + assert resp.status_code == 200 + body = resp.json() + assert body["revoked"] == 2 + assert body["message"] == SuccessMessages.OTHER_SESSIONS_REVOKED + + assert (await other1.get("/users/me")).status_code == 401 + assert (await other2.get("/users/me")).status_code == 401 + await other1.aclose() + await other2.aclose() + + # Current device still works and is the only session left. + listing = (await client.get("/users/me/sessions")).json() + assert listing["total"] == 1 + assert listing["data"][0]["is_current"] is True + + +# --------------------------------------------------------------------------- +# Auth-flow integration +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_logout_removes_session_from_list(client: AsyncClient): + await register_and_verify(client, "sess6@test.com") + await _login(client, "sess6@test.com", user_agent=_UA_WINDOWS) + await client.post("/auth/logout") + + await _login(client, "sess6@test.com", user_agent=_UA_WINDOWS) + listing = (await client.get("/users/me/sessions")).json() + assert listing["total"] == 1 # the logged-out session is gone + + +@pytest.mark.asyncio +async def test_refresh_rotates_session_jti(client: AsyncClient): + await register_and_verify(client, "sess7@test.com") + await _login(client, "sess7@test.com", user_agent=_UA_WINDOWS) + user_id = await _user_id("sess7@test.com") + + async with TestingSessionLocal() as session: + row = ( + ( + await session.execute( + select(UserSession).where(UserSession.user_id == user_id) + ) + ) + .scalars() + .one() + ) + jti_before = row.refresh_jti + + resp = await client.post("/auth/refresh") + assert resp.status_code == 200 + + async with TestingSessionLocal() as session: + row = ( + ( + await session.execute( + select(UserSession).where(UserSession.user_id == user_id) + ) + ) + .scalars() + .one() + ) + assert row.refresh_jti != jti_before + assert row.revoked_at is None + + # Still one single session — rotation must not create a second row. + listing = (await client.get("/users/me/sessions")).json() + assert listing["total"] == 1 + + +@pytest.mark.asyncio +async def test_refresh_replay_revokes_whole_session(client: AsyncClient, fake_redis): + """If the blacklist ever loses state, the jti match still kills replays.""" + await register_and_verify(client, "sess8@test.com") + await _login(client, "sess8@test.com", user_agent=_UA_WINDOWS) + old_refresh = client.cookies.get("refresh_token") + + resp = await client.post("/auth/refresh") + assert resp.status_code == 200 + new_refresh = client.cookies.get("refresh_token") + + # Simulate a Redis wipe: the rotated-away token is no longer blacklisted. + await fake_redis.flushall() + + # Replaying the stale token now hits the session-jti guard... + client.cookies.set("refresh_token", old_refresh) + replay = await client.post("/auth/refresh") + assert replay.status_code == 401 + + # ...which revokes the session, so even the *newest* token is dead. + client.cookies.set("refresh_token", new_refresh) + after = await client.post("/auth/refresh") + assert after.status_code == 401 + + +@pytest.mark.asyncio +async def test_refresh_token_without_sid_rejected(client: AsyncClient): + """Legacy refresh tokens (pre-session deploys) cannot be refreshed.""" + await register_and_verify(client, "sess9@test.com") + user_id = await _user_id("sess9@test.com") + + legacy = create_refresh_token(user_id) # no session_id -> no sid claim + client.cookies.set("refresh_token", legacy) + resp = await client.post("/auth/refresh") + assert resp.status_code == 401 + assert resp.json()["error"] == ErrorMessages.INVALID_TOKEN + + +@pytest.mark.asyncio +async def test_change_password_revokes_every_session(client: AsyncClient): + await register_and_verify(client, "sess10@test.com") + await _login(client, "sess10@test.com", user_agent=_UA_WINDOWS) + + other = _fresh_client() + await _login(other, "sess10@test.com", user_agent=_UA_IPHONE) + + resp = await client.patch( + "/auth/change-password", + json={"current_password": PASSWORD, "new_password": "NewPassword123!"}, + ) + assert resp.status_code == 200 + + # Both devices are logged out everywhere. + assert (await other.get("/users/me")).status_code == 401 + await other.aclose() + + async with TestingSessionLocal() as session: + user_id = await _user_id("sess10@test.com") + rows = ( + ( + await session.execute( + select(UserSession).where(UserSession.user_id == user_id) + ) + ) + .scalars() + .all() + ) + assert rows and all(r.revoked_at is not None for r in rows) + + +# --------------------------------------------------------------------------- +# Admin endpoints + RBAC +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_admin_sessions_require_permission(client: AsyncClient): + await register_and_verify(client, "sess11@test.com") + target_id = await _user_id("sess11@test.com") + + await register_and_verify(client, "admin11@test.com") + await promote_to_admin("admin11@test.com") + admin = _fresh_client() + await _login(admin, "admin11@test.com", user_agent=_UA_WINDOWS) + + assert (await admin.get(f"/admin/users/{target_id}/sessions")).status_code == 403 + assert (await admin.delete(f"/admin/users/{target_id}/sessions")).status_code == 403 + await admin.aclose() + + +@pytest.mark.asyncio +async def test_admin_lists_and_revokes_user_sessions(client: AsyncClient): + await register_and_verify(client, "sess12@test.com") + device1, device2 = _fresh_client(), _fresh_client() + await _login(device1, "sess12@test.com", user_agent=_UA_WINDOWS) + await _login(device2, "sess12@test.com", user_agent=_UA_IPHONE) + target_id = await _user_id("sess12@test.com") + + await register_and_verify(client, "admin12@test.com") + await promote_to_admin("admin12@test.com") + await grant_permissions("admin12@test.com", [Permission.USERS_SESSIONS]) + admin = _fresh_client() + await _login(admin, "admin12@test.com", user_agent=_UA_WINDOWS) + + listing = (await admin.get(f"/admin/users/{target_id}/sessions")).json() + assert listing["total"] == 2 + # "Current" is meaningless for the admin view — never flagged. + assert all(s["is_current"] is False for s in listing["data"]) + + resp = await admin.delete(f"/admin/users/{target_id}/sessions") + assert resp.status_code == 200 + body = resp.json() + assert body["revoked"] == 2 + assert body["message"] == SuccessMessages.ADMIN_SESSIONS_REVOKED + + # Every device of the target dies immediately. + assert (await device1.get("/users/me")).status_code == 401 + assert (await device2.get("/users/me")).status_code == 401 + await device1.aclose() + await device2.aclose() + + # Unknown target user -> 404. + resp = await admin.get(f"/admin/users/{uuid4()}/sessions") + assert resp.status_code == 404 + await admin.aclose() + + +# --------------------------------------------------------------------------- +# Purge worker job +# --------------------------------------------------------------------------- + + +async def _insert_session( + user_id: UUID, + *, + expires_in: timedelta, + revoked_ago: timedelta | None = None, +) -> UUID: + """Insert a raw session row for purge-job test setup.""" + async with TestingSessionLocal() as session: + row = UserSession( + user_id=user_id, + refresh_jti=str(uuid4()), + expires_at=utc_now() + expires_in, + revoked_at=utc_now() - revoked_ago if revoked_ago else None, + ) + session.add(row) + await session.commit() + return row.id + + +@pytest.mark.asyncio +async def test_purge_job_sweeps_expired_and_long_revoked(monkeypatch): + import app.worker.jobs.purge_stale_sessions as mod + + monkeypatch.setattr(mod, "AsyncSessionLocal", TestingSessionLocal) + + async with TestingSessionLocal() as session: + user = User( + email="purge@test.com", + hashed_password=get_password_hash(PASSWORD), + is_verified=True, + ) + session.add(user) + await session.commit() + await session.refresh(user) + user_id = user.id + + keep_active = await _insert_session(user_id, expires_in=timedelta(days=7)) + keep_recent_revoke = await _insert_session( + user_id, expires_in=timedelta(days=7), revoked_ago=timedelta(days=1) + ) + await _insert_session(user_id, expires_in=timedelta(days=-1)) # expired + await _insert_session( # revoked past retention + user_id, expires_in=timedelta(days=7), revoked_ago=timedelta(days=40) + ) + + result = await mod.purge_stale_sessions({}) + assert result.purged == 2 + + async with TestingSessionLocal() as session: + remaining = { + row.id + for row in ( + ( + await session.execute( + select(UserSession).where(UserSession.user_id == user_id) + ) + ) + .scalars() + .all() + ) + } + assert remaining == {keep_active, keep_recent_revoke} diff --git a/app/utils/__init__.py b/app/utils/__init__.py index cbc396f..3abd34f 100644 --- a/app/utils/__init__.py +++ b/app/utils/__init__.py @@ -1 +1,2 @@ +from .datetime_utils import ensure_utc as ensure_utc from .datetime_utils import utc_now as utc_now diff --git a/app/utils/datetime_utils.py b/app/utils/datetime_utils.py index b514ca8..ab29e19 100644 --- a/app/utils/datetime_utils.py +++ b/app/utils/datetime_utils.py @@ -4,3 +4,15 @@ def utc_now() -> datetime: """Return the current datetime in UTC timezone.""" return datetime.now(UTC) + + +def ensure_utc(value: datetime) -> datetime: + """Coerce a datetime to UTC-aware. + + DB backends that drop tzinfo (SQLite in tests) hand back naive datetimes + even for ``DateTime(timezone=True)`` columns; those are stored in UTC, so + attaching UTC restores the original instant for safe comparisons. + """ + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) diff --git a/app/utils/user_agent.py b/app/utils/user_agent.py new file mode 100644 index 0000000..17dbd39 --- /dev/null +++ b/app/utils/user_agent.py @@ -0,0 +1,59 @@ +"""Tiny User-Agent parser — browser and OS names for the sessions UI. + +Deliberately heuristic and dependency-free: the sessions screen only needs a +human-recognizable label ("Chrome on Windows"), not full device analytics. +Order matters in both tables — e.g. Edge and Opera embed ``Chrome/`` in their +UA, so they must be tested first. +""" + +from __future__ import annotations + +from typing import NamedTuple + + +class ParsedUserAgent(NamedTuple): + """Browser and OS display names extracted from a raw User-Agent.""" + + browser: str | None + os: str | None + + +# (needle, display name) — first match wins. +_BROWSERS: tuple[tuple[str, str], ...] = ( + ("edg", "Edge"), + ("opr/", "Opera"), + ("opera", "Opera"), + ("samsungbrowser", "Samsung Internet"), + ("firefox", "Firefox"), + ("chrome", "Chrome"), + ("crios", "Chrome"), + ("fxios", "Firefox"), + ("safari", "Safari"), +) + +_OSES: tuple[tuple[str, str], ...] = ( + ("windows", "Windows"), + ("android", "Android"), + ("iphone", "iOS"), + ("ipad", "iPadOS"), + ("mac os x", "macOS"), + ("macintosh", "macOS"), + ("cros", "ChromeOS"), + ("linux", "Linux"), +) + + +def parse_user_agent(user_agent: str | None) -> ParsedUserAgent: + """Extract (browser, os) display names from a raw User-Agent header. + + Unknown or missing values come back as ``None`` so the UI can fall back + to a generic "Unknown device" label. + """ + if not user_agent: + return ParsedUserAgent(browser=None, os=None) + + ua = user_agent.lower() + + browser = next((name for needle, name in _BROWSERS if needle in ua), None) + os = next((name for needle, name in _OSES if needle in ua), None) + return ParsedUserAgent(browser=browser, os=os) diff --git a/app/worker/jobs/purge_stale_sessions.py b/app/worker/jobs/purge_stale_sessions.py new file mode 100644 index 0000000..81938ef --- /dev/null +++ b/app/worker/jobs/purge_stale_sessions.py @@ -0,0 +1,43 @@ +"""Sweep stale user sessions on the arq cron schedule. + +Expired sessions carry no value and are deleted immediately; revoked ones are +kept for ``SESSION_REVOKED_RETENTION_DAYS`` (audit trail), then dropped. The +whole sweep is one set-based DELETE, so no batching loop is needed — Postgres +handles millions of rows in a single statement comfortably. +""" + +from __future__ import annotations + +import logging +import time +from typing import TypedDict + +from app.core.config import settings +from app.core.db import AsyncSessionLocal +from app.repositories.user_session import purge_stale_sessions as purge_repo +from app.schemas.worker import SessionPurgeJobResult + +logger = logging.getLogger(__name__) + + +class JobContext(TypedDict, total=False): + """Subset of the arq context this job relies on (none of it, currently).""" + + job_id: str + + +async def purge_stale_sessions(ctx: JobContext) -> SessionPurgeJobResult: + """Delete expired sessions and long-revoked ones in one sweep.""" + _ = ctx + start = time.monotonic() + + async with AsyncSessionLocal() as session: + purged = await purge_repo( + session, + revoked_retention_days=settings.SESSION_REVOKED_RETENTION_DAYS, + ) + + duration_ms = int((time.monotonic() - start) * 1000) + result = SessionPurgeJobResult(purged=purged, duration_ms=duration_ms) + logger.info("purge_stale_sessions: completed", extra=result.model_dump()) + return result diff --git a/app/worker/settings.py b/app/worker/settings.py index 5bdbccd..eae21e5 100644 --- a/app/worker/settings.py +++ b/app/worker/settings.py @@ -11,6 +11,7 @@ from app.core.config import settings from app.worker.jobs.delete_expired_accounts import delete_expired_accounts +from app.worker.jobs.purge_stale_sessions import purge_stale_sessions def _redis_settings() -> RedisSettings: @@ -22,7 +23,7 @@ class WorkerSettings: """Entry point for ``arq app.worker.settings.WorkerSettings``.""" redis_settings = _redis_settings() - functions: list = [delete_expired_accounts] + functions: list = [delete_expired_accounts, purge_stale_sessions] cron_jobs = [ cron( delete_expired_accounts, @@ -30,6 +31,12 @@ class WorkerSettings: minute={settings.DELETION_JOB_CRON_MINUTE}, run_at_startup=False, ), + cron( + purge_stale_sessions, + hour={settings.SESSION_PURGE_CRON_HOUR}, + minute={settings.SESSION_PURGE_CRON_MINUTE}, + run_at_startup=False, + ), ] max_jobs = 10 job_timeout = 600