from datetime import datetime, timezone from typing import Annotated import jwt as _jwt from fastapi import APIRouter, Depends, HTTPException, Query, status from fastapi.security import HTTPAuthorizationCredentials from sqlalchemy import delete, func, select, update as sql_update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from backend.app.api.routes.settings import get_external_login_url from backend.app.core.auth import ( ALGORITHM, SECRET_KEY, RequireAdminIfAuthEnabled, RequireAnyPermissionIfAuthEnabled, RequirePermissionIfAuthEnabled, get_current_user_optional, get_password_hash, revoke_jti, security, verify_password, ) from backend.app.core.database import get_db from backend.app.core.permissions import Permission from backend.app.core.websocket import ws_manager from backend.app.models.api_key import APIKey from backend.app.models.archive import PrintArchive from backend.app.models.auth_ephemeral import AuthEphemeralToken from backend.app.models.connected_app import ConnectedApp, ConnectedAppGrant from backend.app.models.group import Group from backend.app.models.library import LibraryFile from backend.app.models.long_lived_token import LongLivedToken from backend.app.models.oidc_provider import UserOIDCLink from backend.app.models.print_batch import PrintBatch from backend.app.models.print_log import PrintLogEntry from backend.app.models.print_queue import PrintQueueItem from backend.app.models.settings import Settings from backend.app.models.user import User from backend.app.models.user_otp_code import UserOTPCode from backend.app.models.user_totp import UserTOTP from backend.app.schemas.auth import ( ChangePasswordRequest, GroupBrief, UserCreate, UserResponse, UserSlim, UserUpdate, ) from backend.app.services.email_service import ( create_welcome_email_from_template, generate_secure_password, get_smtp_settings, send_email, ) from backend.app.services.finance_defaults import ensure_user_finance_defaults from backend.app.services.printer_manager import printer_manager from backend.app.services.slice_dispatch import slice_dispatch from backend.app.utils.library_paths import remove_library_photos_dir from backend.app.utils.local_time import utcnow_naive router = APIRouter(prefix="/users", tags=["users"]) def _user_to_response(user: User) -> UserResponse: """Convert a User model to UserResponse schema.""" return UserResponse( id=user.id, username=user.username, email=user.email, role=user.role, is_active=user.is_active, is_admin=user.is_admin, auth_source=getattr(user, "auth_source", "local"), groups=[GroupBrief(id=g.id, name=g.name) for g in user.groups], permissions=sorted(user.get_permissions()), created_at=user.created_at.isoformat(), ) @router.get("", response_model=list[UserResponse]) @router.get("/", response_model=list[UserResponse]) async def list_users( _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_READ), db: AsyncSession = Depends(get_db), ): """List all users. Read-only — gated on ``USERS_READ`` only. Operator-visible UIs (Stats filter-by-user, Archives Print Log username column, File Manager username autocomplete) consume this endpoint via custom- group ``users:read`` grants without admin role. The admin-only boundary lives on the write endpoints below.""" result = await db.execute(select(User).options(selectinload(User.groups)).order_by(User.created_at)) users = result.scalars().all() return [_user_to_response(user) for user in users] async def drop_pending_tokens(db: AsyncSession, username: str) -> None: """Delete the one-time tokens pending under a username that is going away. Reset links, 2FA and OIDC steps, connect codes, camera, WebSocket and slicer tokens are looked up by username, so they would work for whoever is given the name next. The logout list (revoked_jti) stays: dropping an entry would make the logged-out token valid again. """ await db.execute( delete(AuthEphemeralToken).where( # Both sides in SQL: SQLite's lower() changes only ASCII letters, so # it never equals Python's lower() of a name with an umlaut. func.lower(AuthEphemeralToken.username) == func.lower(username), AuthEphemeralToken.token_type != "revoked_jti", ) ) @router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED) @router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED) async def create_user( user_data: UserCreate, _admin: User | None = RequireAdminIfAuthEnabled(), _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_CREATE), db: AsyncSession = Depends(get_db), ): """Create a new user. When advanced authentication is enabled: - Email is required - Password is auto-generated and emailed to user - Admin cannot set or see the password """ import logging logger = logging.getLogger(__name__) # Check if advanced auth is enabled result = await db.execute(select(Settings).where(Settings.key == "advanced_auth_enabled")) advanced_auth_setting = result.scalar_one_or_none() advanced_auth_enabled = advanced_auth_setting and advanced_auth_setting.value.lower() == "true" # Check if username already exists (case-insensitive) existing_user = await db.execute(select(User).where(func.lower(User.username) == func.lower(user_data.username))) if existing_user.scalar_one_or_none(): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Username already exists", ) # Validate role if user_data.role not in ["admin", "user"]: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Role must be 'admin' or 'user'", ) # Advanced auth validation if advanced_auth_enabled: if not user_data.email: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email is required when advanced authentication is enabled", ) # Check if email already exists (case-insensitive) existing_email = await db.execute(select(User).where(func.lower(User.email) == func.lower(user_data.email))) if existing_email.scalar_one_or_none(): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email already exists", ) # Generate password if advanced auth enabled, otherwise require password if advanced_auth_enabled: password = generate_secure_password() else: if not user_data.password: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Password is required when advanced authentication is disabled", ) password = user_data.password new_user = User( username=user_data.username, email=user_data.email, password_hash=get_password_hash(password), role=user_data.role, is_active=True, ) # Handle group assignments if user_data.group_ids: groups_result = await db.execute(select(Group).where(Group.id.in_(user_data.group_ids))) groups = groups_result.scalars().all() if len(groups) != len(user_data.group_ids): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="One or more group IDs are invalid", ) new_user.groups = list(groups) db.add(new_user) await db.flush() await ensure_user_finance_defaults(db, new_user) await db.commit() await db.refresh(new_user) # Send welcome email if advanced auth enabled if advanced_auth_enabled and new_user.email: try: smtp_settings = await get_smtp_settings(db) if smtp_settings: login_url = await get_external_login_url(db) subject, text_body, html_body = await create_welcome_email_from_template( db, new_user.username, password, login_url ) send_email(smtp_settings, new_user.email, subject, text_body, html_body) logger.info(f"Welcome email sent to {new_user.email}") else: logger.warning(f"SMTP not configured, could not send welcome email to {new_user.email}") except Exception as e: logger.error(f"Failed to send welcome email: {e}") # Don't fail user creation if email fails return _user_to_response(new_user) @router.get("/slim", response_model=list[UserSlim]) async def list_users_slim( _: User | None = RequireAnyPermissionIfAuthEnabled(Permission.USERS_READ_SLIM, Permission.USERS_READ), db: AsyncSession = Depends(get_db), ): """List users as ``{id, username}`` only (#1894). Exists so an API key -- or a group that should not see emails, roles and permission sets -- can turn the ``created_by_id`` values it already gets back from archives, stats and the queue into names. ``USERS_READ`` is accepted alongside ``USERS_READ_SLIM`` because it is strictly broader; groups that already hold it keep working without a permission backfill. For API keys only the slim permission resolves (the full one is unmapped = administrative), so a key reaches this and not the listing above. Declared before ``/{user_id}`` on purpose: FastAPI matches in declaration order, and the reverse order would parse "slim" as the int path parameter and answer 422. """ result = await db.execute(select(User.id, User.username).order_by(User.username)) return [UserSlim(id=row.id, username=row.username) for row in result.all()] @router.get("/{user_id}", response_model=UserResponse) async def get_user( user_id: int, _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_READ), db: AsyncSession = Depends(get_db), ): """Get a user by ID. Read-only — gated on ``USERS_READ`` only.""" result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups))) user = result.scalar_one_or_none() if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found", ) return _user_to_response(user) @router.patch("/{user_id}", response_model=UserResponse) async def update_user( user_id: int, user_data: UserUpdate, _admin: User | None = RequireAdminIfAuthEnabled(), _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_UPDATE), db: AsyncSession = Depends(get_db), ): """Update a user.""" close_sockets = False deactivated_keys: list[int] = [] result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups))) user = result.scalar_one_or_none() if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found", ) # Prevent deactivating the last admin if user_data.is_active is False and user.is_admin: # Count admins by role or Administrators group membership admin_count_result = await db.execute(select(User).where(User.role == "admin", User.is_active.is_(True))) role_admins = admin_count_result.scalars().all() # Also check for users in Administrators group admin_group_result = await db.execute( select(Group).where(Group.name == "Administrators").options(selectinload(Group.users)) ) admin_group = admin_group_result.scalar_one_or_none() group_admins = [u for u in (admin_group.users if admin_group else []) if u.is_active] # Combine unique admins all_admins = {u.id for u in role_admins} | {u.id for u in group_admins} if len(all_admins) <= 1 and user.id in all_admins: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot deactivate the last admin user", ) # Prevent changing role of last admin if user_data.role and user_data.role != "admin" and user.role == "admin": admin_count_result = await db.execute(select(User).where(User.role == "admin", User.is_active.is_(True))) admin_count = len(admin_count_result.scalars().all()) if admin_count <= 1: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot change role of the last admin user", ) if user_data.username is not None: # Check if new username already exists (case-insensitive) existing_user = await db.execute( select(User).where(func.lower(User.username) == func.lower(user_data.username), User.id != user_id) ) if existing_user.scalar_one_or_none(): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Username already exists", ) if user_data.username != user.username: # The name passes on: a deleted user's still-valid JWTs under the new # name must not sign this user in, and what is pending under the old # one must not reach whoever is given it next. This user's own # sessions carry the old name and end anyway. A change of case counts # too: SQLite compares only ASCII letters without case, so 'Ärger' # and 'ärger' are two names there. await drop_pending_tokens(db, user.username) user.password_changed_at = utcnow_naive() close_sockets = True user.username = user_data.username if user_data.email is not None and not user_data.email.strip(): # Empty removes the email; null leaves it as it is. if user.email: rows = await db.execute( select(Settings.key, Settings.value).where( Settings.key.in_(["advanced_auth_enabled", f"user_{user.id}_email_2fa_enabled"]) ) ) flags = {key: (value or "").lower() == "true" for key, value in rows.all()} if flags.get("advanced_auth_enabled"): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email is required when advanced authentication is enabled", ) # Their sign-in codes go to this address. if flags.get(f"user_{user.id}_email_2fa_enabled"): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email is required while the user has email two-factor authentication on", ) user.email = None elif user_data.email is not None: # Check if new email already exists (case-insensitive) existing_email = await db.execute( select(User).where(func.lower(User.email) == func.lower(user_data.email), User.id != user_id) ) if existing_email.scalar_one_or_none(): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email already exists", ) user.email = user_data.email if user_data.password is not None: if getattr(user, "auth_source", "local") == "ldap": raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot set password for LDAP users", ) user.password_hash = get_password_hash(user_data.password) # As a reset: sessions and pending tokens of the old password end (an # admin setting a password is often taking an account back). await drop_pending_tokens(db, user.username) user.password_changed_at = utcnow_naive() close_sockets = True if user_data.role is not None: if user_data.role not in ["admin", "user"]: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Role must be 'admin' or 'user'", ) user.role = user_data.role if user_data.is_active is not None: if user.is_active and not user_data.is_active: # Deactivated: their printers go at the scope refresh below, but # events of no printer (inventory, SpoolBuddy tags, announcements) # would still reach their open dashboards, and their tokens would # open new ones. The same for their API keys' (a kiosk), which # REST already refuses for a deactivated owner. from backend.app.api.routes.api_keys import drop_api_key_tokens await drop_pending_tokens(db, user.username) deactivated_keys = list( (await db.execute(select(APIKey.id).where(APIKey.user_id == user_id))).scalars().all() ) await drop_api_key_tokens(db, deactivated_keys) close_sockets = True user.is_active = user_data.is_active # Handle group assignments if user_data.group_ids is not None: groups_result = await db.execute(select(Group).where(Group.id.in_(user_data.group_ids))) groups = groups_result.scalars().all() if len(groups) != len(user_data.group_ids): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="One or more group IDs are invalid", ) user.groups = list(groups) await ensure_user_finance_defaults(db, user) await db.commit() if deactivated_keys: await ws_manager.close_api_keys(deactivated_keys) if close_sockets: # Open dashboards are not signed in again. After a rename they keep the # name they connected with and look their printers up by it on every # scope refresh (the next user given it); after a password set they are # whoever held the account before. await ws_manager.close_user(user_id) if user_data.group_ids is not None or user_data.role is not None or user_data.is_active is not None: # Groups, admin role and deactivation all change which printers the # user's open dashboards may hear about (#1727) await ws_manager.refresh_printer_scopes() result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups))) user = result.scalar_one() return _user_to_response(user) @router.get("/{user_id}/items-count") async def get_user_items_count( user_id: int, _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_READ), db: AsyncSession = Depends(get_db), ): """Get count of items created by this user. Read-only — gated on ``USERS_READ`` only.""" # Verify user exists result = await db.execute(select(User).where(User.id == user_id)) if not result.scalar_one_or_none(): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found", ) # Count archives archives_result = await db.execute(select(func.count(PrintArchive.id)).where(PrintArchive.created_by_id == user_id)) archives_count = archives_result.scalar() or 0 # Count queue items queue_result = await db.execute( select(func.count(PrintQueueItem.id)).where(PrintQueueItem.created_by_id == user_id) ) queue_items_count = queue_result.scalar() or 0 # Count library files library_result = await db.execute( select(func.count(LibraryFile.id)).where( LibraryFile.created_by_id == user_id, LibraryFile.deleted_at.is_(None), ) ) library_files_count = library_result.scalar() or 0 return { "archives": archives_count, "queue_items": queue_items_count, "library_files": library_files_count, } @router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_user( user_id: int, delete_items: bool = Query(False, description="Delete all items created by this user"), _admin: User | None = RequireAdminIfAuthEnabled(), current_user: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_DELETE), db: AsyncSession = Depends(get_db), ): """Delete a user. If delete_items=True, all archives, queue items, and library files created by this user will also be deleted. Otherwise, these items will become "ownerless" (created_by_id set to NULL by the foreign key constraint). """ result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups))) user = result.scalar_one_or_none() if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found", ) # Prevent deleting the last admin if user.is_admin: # Count admins by role or Administrators group membership admin_count_result = await db.execute(select(User).where(User.role == "admin", User.id != user_id)) other_role_admins = admin_count_result.scalars().all() # Also check for users in Administrators group admin_group_result = await db.execute( select(Group).where(Group.name == "Administrators").options(selectinload(Group.users)) ) admin_group = admin_group_result.scalar_one_or_none() other_group_admins = [u for u in (admin_group.users if admin_group else []) if u.id != user_id and u.is_active] # Combine unique admins all_other_admins = {u.id for u in other_role_admins} | {u.id for u in other_group_admins} if len(all_other_admins) == 0: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot delete the last admin user", ) # Prevent deleting yourself (only if auth is enabled and we have a current user) if current_user and user.id == current_user.id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot delete your own account", ) # Photo directories of the library rows about to go, resolved while the # rows still exist to say which ids they belong to. Removed after the # commit, so a failed delete leaves the pictures alone (#3077). doomed_library_file_ids: list[int] = [] if delete_items: doomed_library_file_ids = list( (await db.execute(select(LibraryFile.id).where(LibraryFile.created_by_id == user_id))).scalars().all() ) # Delete all items created by this user. The print log stays, without # the archives it pointed at; archives of other users printed from # these library files lose that link (both SET NULL on PostgreSQL). from backend.app.services.library_trash import delete_dependent_variants, release_queue_references await delete_dependent_variants(db, doomed_library_file_ids) await release_queue_references(db, doomed_library_file_ids) # A print still running keeps what it needs to be billed when it # ends: its queue row and the user's archive it prints are detached # (no user), as without delete_items, rather than deleted. Deleted, the # completion found no queue row and charged the archive's creator # instead (for a reprint someone else, who never printed it), or found # no archive and billed nothing, its reservation held for good. from sqlalchemy import and_, or_ from backend.app.models.finance import BudgetReservation running_rows = ( await db.execute( select(PrintQueueItem.id, PrintQueueItem.archive_id).where( PrintQueueItem.created_by_id == user_id, PrintQueueItem.status == "printing" ) ) ).all() running_items = [row.id for row in running_rows] running_archives = [row.archive_id for row in running_rows if row.archive_id is not None] running_archives += ( ( await db.execute( select(PrintArchive.id).where( PrintArchive.created_by_id == user_id, PrintArchive.status == "printing" ) ) ) .scalars() .all() ) # Also one another user is printing (a reprint of this user's archive, # dispatched before the printer's start marks the archive printing): # deleted, it took their queue row with it and their print went unbilled. own_archives = select(PrintArchive.id).where(PrintArchive.created_by_id == user_id) running_archives += ( ( await db.execute( select(PrintQueueItem.archive_id).where( PrintQueueItem.status == "printing", PrintQueueItem.archive_id.in_(own_archives) ) ) ) .scalars() .all() ) if running_items: await db.execute( sql_update(PrintQueueItem).where(PrintQueueItem.id.in_(running_items)).values(created_by_id=None) ) if running_archives: await db.execute( sql_update(PrintArchive) .where(PrintArchive.id.in_(running_archives), PrintArchive.created_by_id == user_id) .values(created_by_id=None) ) # Budget held for the queue items and archives deleted here: no print # completion can settle it any more (it finds no archive), and an # active reservation held the cost centre's budget, and kept the cost # centre from being deleted, for good. Nothing is charged for those: # they go, as the admin chose. (A running print's was detached above.) await db.execute( sql_update(BudgetReservation) .where( BudgetReservation.status == "active", or_( and_( BudgetReservation.source_type == "print_queue", BudgetReservation.source_id.in_( select(PrintQueueItem.id).where(PrintQueueItem.created_by_id == user_id) ), ), BudgetReservation.print_archive_id.in_( select(PrintArchive.id).where(PrintArchive.created_by_id == user_id) ), ), ) .values(status="released", released_at=datetime.now(timezone.utc)) ) await db.execute(delete(PrintArchive).where(PrintArchive.created_by_id == user_id)) await db.execute(delete(PrintQueueItem).where(PrintQueueItem.created_by_id == user_id)) await db.execute(delete(LibraryFile).where(LibraryFile.created_by_id == user_id)) await db.execute(delete(PrintBatch).where(PrintBatch.created_by_id == user_id)) else: # Explicitly set created_by_id to NULL for all items (ensures consistent behavior # across different database backends, including SQLite without foreign key support). # PrintBatch carries the same created_by_id FK with ondelete=SET NULL — admin-deleted # users would otherwise leave dangling created_by_id on SQLite (#1295 review nit). from sqlalchemy import update from backend.app.core.printer_scope import resolve_user_printer_scope # An ownerless "any " job may run on any printer of that model; # the scheduler held it to this user's printers only while it was # theirs (#1727). Stage those jobs instead, so an admin decides where # they run rather than the job quietly spreading across the fleet. if not (await resolve_user_printer_scope(db, user)).is_unrestricted: await db.execute( update(PrintQueueItem) .where( PrintQueueItem.created_by_id == user_id, PrintQueueItem.printer_id.is_(None), PrintQueueItem.status == "pending", ) .values(manual_start=True) ) await db.execute(update(PrintArchive).where(PrintArchive.created_by_id == user_id).values(created_by_id=None)) await db.execute( update(PrintQueueItem).where(PrintQueueItem.created_by_id == user_id).values(created_by_id=None) ) await db.execute(update(LibraryFile).where(LibraryFile.created_by_id == user_id).values(created_by_id=None)) await db.execute(update(PrintBatch).where(PrintBatch.created_by_id == user_id).values(created_by_id=None)) # The print log keeps who started each run as long as the user exists, and # outlives them either way (ON DELETE SET NULL, which SQLite doesn't apply). await db.execute(sql_update(PrintLogEntry).where(PrintLogEntry.created_by_id == user_id).values(created_by_id=None)) # Drop API keys owned by this user. The model declares ON DELETE CASCADE # so Postgres handles this automatically, but SQLite ships with FK # enforcement off (the project's existing pattern — same reason the # blocks above set created_by_id = NULL by hand). Without an explicit # DELETE here, deleting a user on SQLite would leave their API keys # with a dangling user_id and ``_user_from_api_key`` would return None, # silently degrading the keys to anonymous (and locking them out of # /cloud/* — but the rest of the API would still accept them, which is # exactly the orphan-key state the CASCADE was meant to prevent). from backend.app.api.routes.api_keys import drop_api_key_tokens key_ids = (await db.execute(select(APIKey.id).where(APIKey.user_id == user_id))).scalars().all() await drop_api_key_tokens(db, key_ids) await db.execute(delete(APIKey).where(APIKey.user_id == user_id)) # Drop OIDC links, MFA state, and long-lived camera-stream tokens # owned by this user. Same SQLite/FK pattern as APIKey above. Without # these, deleting a user on SQLite leaves: # - UserOIDCLink: the OIDC callback finds the orphan link, fails to # resolve the (now missing) user, and falls through to # "account_inactive" instead of triggering auto_create (#1285). # - UserTOTP: MFA secrets persist in the DB after the owning user. # - UserOTPCode: pending email OTP codes linger. # - LongLivedToken: per-user camera-stream tokens whose secret_hash # is still valid — verify() would happily match them by lookup # prefix even though the user is gone. await db.execute(delete(UserOIDCLink).where(UserOIDCLink.user_id == user_id)) await db.execute(delete(UserTOTP).where(UserTOTP.user_id == user_id)) await db.execute(delete(UserOTPCode).where(UserOTPCode.user_id == user_id)) await db.execute(delete(LongLivedToken).where(LongLivedToken.user_id == user_id)) # Connected apps, same SQLite/FK pattern. A leftover consent row would # skip the consent screen for whoever is next given this user id. from sqlalchemy import update as _update await db.execute(delete(ConnectedAppGrant).where(ConnectedAppGrant.user_id == user_id)) await drop_pending_tokens(db, user.username) await db.execute(_update(ConnectedApp).where(ConnectedApp.created_by_id == user_id).values(created_by_id=None)) # Email 2FA is a settings key named after the id, which no foreign key covers. await db.execute(delete(Settings).where(Settings.key == f"user_{user_id}_email_2fa_enabled")) await db.delete(user) await db.commit() # What still runs for them must not be credited to whoever gets the id next # (SQLite reuses it): a print, a slice, an open dashboard. printer_manager.forget_user(user_id) from backend.app.main import forget_print_creator forget_print_creator(user_id) slice_dispatch.forget_owner(user_id) await ws_manager.close_user(user_id) await ws_manager.close_api_keys(key_ids) from backend.app.api.routes.notifications import forget_app_senders forget_app_senders(key_ids, [user_id]) await ws_manager.refresh_printer_scopes() for file_id in doomed_library_file_ids: remove_library_photos_dir(file_id) @router.post("/me/change-password", response_model=dict) async def change_own_password( password_data: ChangePasswordRequest, credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(security)] = None, current_user: User | None = Depends(get_current_user_optional), db: AsyncSession = Depends(get_db), ): """Change the current user's password. Requires current password verification.""" if not current_user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required to change password", ) # Block password change for LDAP users if getattr(current_user, "auth_source", "local") == "ldap": raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot change password for LDAP users — passwords are managed by the LDAP server", ) # Verify current password if not current_user.password_hash: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Account has no local password set", ) # Rate-limit failed password-change attempts (H-R5-A) from backend.app.api.routes.mfa import MAX_2FA_ATTEMPTS, check_rate_limit, record_failed_attempt await check_rate_limit(db, current_user.username, event_type="password_change", max_attempts=MAX_2FA_ATTEMPTS) if not verify_password(password_data.current_password, current_user.password_hash): await record_failed_attempt(db, current_user.username, event_type="password_change") raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Current password is incorrect", ) # Fetch user from this session to ensure changes are persisted result = await db.execute(select(User).where(User.id == current_user.id)) user = result.scalar_one_or_none() if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found", ) # Update password user.password_hash = get_password_hash(password_data.new_password) user.password_changed_at = datetime.now(timezone.utc) # M-R7-B: invalidate all prior JWTs # The WebSocket and camera tokens too: reusable for an hour, they would # reconnect a closed dashboard straight away. await drop_pending_tokens(db, user.username) await db.commit() # Open dashboards are never signed in again: one someone else had open on # the account would keep receiving its printers. This tab signs in again # (the route revokes its JWT anyway). await ws_manager.close_user(user.id) # L-R6-A: Password verified successfully — reset the failure counter from backend.app.api.routes.mfa import clear_failed_attempts await clear_failed_attempts(db, user.username, event_type="password_change") # Revoke the current session token so the caller must re-authenticate (M-R5-A) if credentials is not None: try: payload = _jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM]) jti = payload.get("jti") exp = payload.get("exp") if jti and exp: try: await revoke_jti(jti, datetime.fromtimestamp(exp, tz=timezone.utc), user.username) except Exception as exc: # B4: log so operators know revocation is broken; password was # already changed so the token will fail freshness checks anyway. import logging logging.getLogger(__name__).error( "Failed to revoke JTI after password change for user %s: %s", user.username, exc ) except Exception: pass # Decode failure is harmless — token is already invalidated by password_changed_at return {"message": "Password changed successfully"}