"""Internal-user authentication and permission introspection routes. Frontends should call ``GET /api/access/me`` to discover which permission keys the current user has, then use those keys to hide/show navigation items. **Visibility is not security** — every privileged backend route must depend on ``require_permission`` (or one of its siblings) directly. """ from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException, Request, Response, status from pydantic import BaseModel from sqlalchemy import func, select from sqlalchemy.orm import Session, selectinload from app.core.access import ( INTERNAL_USER_SUBJECT, INTERNAL_USER_TENANT_ID, get_current_user, get_user_permissions, permissions_to_module_map, require_permission, ) from app.core.config import settings from app.core.http import CLIENT_AUTH_COOKIE from app.core.rate_limit import SlidingWindowRateLimiter, request_client_key from app.core.security_logging import log_security_event from app.core.security import hash_password, issue_token, verify_password from app.db.session import get_db from app.models.access import Permission, Role, User router = APIRouter(prefix="/api/access", tags=["access"]) login_rate_limiter = SlidingWindowRateLimiter( limit=settings.login_rate_limit_attempts, window_seconds=settings.login_rate_limit_window_seconds, ) class LoginRequest(BaseModel): email: str password: str class UserSession(BaseModel): # Mirrors the existing `LoginResponse` shape so the frontend's `AppSession` # store can consume this response without a separate type. `permissions` # is the new permission-key array; `module_permissions` is the legacy # module→access-level map for nav gating. user_id: int email: str name: str role: str role_name: str | None = None is_active: bool tenant_id: str | None = None client_role: str | None = None client_account_id: int | None = None module_permissions: dict[str, str] = {} permissions: list[str] token: str | None = None class RoleRead(BaseModel): id: int name: str description: str | None permissions: list[str] module_permissions: dict[str, str] is_protected: bool = False user_count: int = 0 class UserRead(BaseModel): id: int email: str name: str is_active: bool role: str | None role_id: int | None # True when this user can never be deleted (lean owner accounts). The UI # uses this to disable the delete control rather than re-deriving the rule. is_protected: bool = False class AssignableRole(BaseModel): id: int name: str description: str | None class RoleModuleDefinition(BaseModel): key: str label: str description: str levels: list[str] class CreateUserRequest(BaseModel): email: str name: str role_id: int | None = None is_active: bool = True password: str | None = None class AdminUpdateUserRequest(BaseModel): name: str | None = None email: str | None = None role_id: int | None = None is_active: bool | None = None class AdminSetPasswordRequest(BaseModel): new_password: str class CreateRoleRequest(BaseModel): name: str description: str | None = None module_permissions: dict[str, str] = {} class UpdateRoleRequest(BaseModel): name: str | None = None description: str | None = None module_permissions: dict[str, str] | None = None # Lean owner accounts are permanent: they may be edited but never deleted, so a # tenant can't accidentally lock itself out of the highest level of access. LEAN_ROLE_NAME = "lean" ADMIN_ROLE_NAME = "admin" ROLE_MANAGEMENT_ALLOWED_ROLES = {LEAN_ROLE_NAME, ADMIN_ROLE_NAME} PROTECTED_ROLE_NAMES = ROLE_MANAGEMENT_ALLOWED_ROLES ROLE_MODULE_DEFINITIONS: tuple[dict[str, object], ...] = ( { "key": "dashboard", "label": "Dashboard", "description": "Home dashboard visibility.", "levels": {"none": (), "view": ("view_dashboard",)}, }, { "key": "mix_calculator", "label": "Mix Calculator", "description": "Open the calculator and save sessions.", "levels": { "none": (), "view": ("view_mix_calculator",), "edit": ("view_mix_calculator", "use_mix_calculator", "save_mix_calculator_session"), }, }, { "key": "raw_materials", "label": "Raw Materials", "description": "View or edit raw materials.", "levels": {"none": (), "view": ("view_raw_materials",), "edit": ("view_raw_materials", "edit_raw_materials")}, }, { "key": "products", "label": "Products", "description": "View or edit finished products.", "levels": {"none": (), "view": ("view_products",), "edit": ("view_products", "edit_products")}, }, { "key": "mix_master", "label": "Mix Master", "description": "View or edit mix recipes.", "levels": {"none": (), "view": ("view_mixes",), "edit": ("view_mixes", "edit_mixes")}, }, { "key": "operations_throughput", "label": "Throughput", "description": "View or edit throughput entries.", "levels": {"none": (), "view": ("view_throughput",), "edit": ("view_throughput", "edit_throughput")}, }, { "key": "ordering", "label": "Ordering", "description": "Access customer ordering and ordering administration.", "levels": { "none": (), "view": ("view_ordering",), "edit": ("view_ordering", "edit_ordering"), "manage": ("view_ordering", "edit_ordering", "manage_ordering"), }, }, { "key": "scenarios", "label": "Scenarios", "description": "View or run scenarios.", "levels": {"none": (), "view": ("view_scenarios",), "edit": ("view_scenarios", "edit_scenarios")}, }, { "key": "client_access", "label": "Client Access", "description": "Manage customer portal accounts and access.", "levels": {"none": (), "manage": ("manage_client_access",)}, }, { "key": "users", "label": "Users", "description": "View or manage internal users.", "levels": {"none": (), "view": ("view_users",), "manage": ("view_users", "manage_users")}, }, { "key": "roles", "label": "Roles", "description": "Manage roles and permission assignments.", "levels": {"none": (), "manage": ("manage_permissions",)}, }, { "key": "settings", "label": "Settings", "description": "Open settings and edit system configuration.", "levels": {"none": (), "view": ("view_settings",), "edit": ("view_settings", "edit_settings")}, }, ) def _serialize_user_read(user: User) -> UserRead: role_name = user.role.name if user.role else None return UserRead( id=user.id, email=user.email, name=user.name, is_active=user.is_active, role=role_name, role_id=user.role_id, is_protected=(role_name or "").lower() == LEAN_ROLE_NAME, ) def _role_name_lower(role: Role | None) -> str: return (role.name if role else "").strip().lower() def _is_protected_role_name(role_name: str | None) -> bool: return (role_name or "").strip().lower() in PROTECTED_ROLE_NAMES def _module_definitions_response() -> list[RoleModuleDefinition]: return [ RoleModuleDefinition( key=definition["key"], label=definition["label"], description=definition["description"], levels=list(definition["levels"].keys()), ) for definition in ROLE_MODULE_DEFINITIONS ] def _permissions_to_role_module_map(permission_keys: set[str]) -> dict[str, str]: result: dict[str, str] = {} for definition in ROLE_MODULE_DEFINITIONS: selected = "none" levels = definition["levels"] for level, required in levels.items(): required_keys = set(required) if not required_keys or required_keys.issubset(permission_keys): selected = level result[definition["key"]] = selected return result def _role_payload_to_permission_keys(module_permissions: dict[str, str]) -> set[str]: known_modules = {definition["key"] for definition in ROLE_MODULE_DEFINITIONS} unknown_modules = sorted(set(module_permissions) - known_modules) if unknown_modules: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Unknown modules: {unknown_modules}", ) granted: set[str] = set() for definition in ROLE_MODULE_DEFINITIONS: key = definition["key"] level = module_permissions.get(key, "none") available_levels: dict[str, tuple[str, ...]] = definition["levels"] # type: ignore[assignment] if level not in available_levels: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Invalid access level '{level}' for module '{key}'", ) granted.update(available_levels[level]) return granted def _serialize_role_read(role: Role, *, user_count: int = 0) -> RoleRead: permission_keys = {permission.key for permission in role.permissions} return RoleRead( id=role.id, name=role.name, description=role.description, permissions=sorted(permission_keys), module_permissions=_permissions_to_role_module_map(permission_keys), is_protected=_is_protected_role_name(role.name), user_count=user_count, ) def _require_role_management_actor(user: User = Depends(get_current_user)) -> User: if _role_name_lower(user.role) not in ROLE_MANAGEMENT_ALLOWED_ROLES: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Only lean and admin accounts can manage roles", ) return user def _load_role(db: Session, role_id: int) -> Role: role = db.scalar( select(Role).where(Role.id == role_id).options(selectinload(Role.permissions)) ) if role is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Role not found") return role def _apply_role_updates( role: Role, *, name: str | None, description: str | None, module_permissions: dict[str, str] | None, permissions_by_key: dict[str, Permission], ) -> None: if name is not None: trimmed_name = name.strip() if not trimmed_name: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Role name cannot be empty") role.name = trimmed_name if description is not None: role.description = description.strip() or None if module_permissions is not None: desired_permission_keys = _role_payload_to_permission_keys(module_permissions) desired = {permissions_by_key[key] for key in desired_permission_keys} current = set(role.permissions) for permission in desired - current: role.permissions.append(permission) for permission in current - desired: role.permissions.remove(permission) def _serialize_session(user: User, *, include_token: bool = False) -> UserSession: permission_set = get_user_permissions(user) permissions = sorted(permission_set) module_permissions = permissions_to_module_map(permission_set) role_name = user.role.name if user.role else None token = None if include_token: token = issue_token( {"sub": INTERNAL_USER_SUBJECT, "user_id": user.id, "email": user.email}, ttl_seconds=settings.session_ttl_seconds, ) # role="internal" is a marker the shared auth deps recognise so internal # users can hit the same routes as client-portal users without being # confused with them. Display name lives in role_name / client_role. return UserSession( user_id=user.id, email=user.email, name=user.name, role="internal", role_name=role_name, is_active=user.is_active, tenant_id=INTERNAL_USER_TENANT_ID, client_role=role_name, client_account_id=None, module_permissions=module_permissions, permissions=permissions, token=token, ) @router.post("/login", response_model=UserSession) def login(payload: LoginRequest, response: Response, request: Request, db: Session = Depends(get_db)): """Internal-user login. Authenticates against the per-user password hash stored on ``users``. Inactive or unknown users are rejected with a generic 401 to avoid leaking which emails are valid. """ login_rate_limiter.hit(request_client_key(request, suffix="internal-login")) email = payload.email.strip().lower() user = db.scalar( select(User) .where(User.email == email) .options(selectinload(User.role).selectinload(Role.permissions)) ) if user is None or not user.is_active: log_security_event("auth.login_failed", audience="internal", ip=request_client_key(request)) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid email or password") if not ( verify_password(payload.password, user.password_hash) or (user.password_hash is None and payload.password == settings.admin_password) ): log_security_event("auth.login_failed", audience="internal", ip=request_client_key(request)) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid email or password") session = _serialize_session(user, include_token=True) if session.token: CLIENT_AUTH_COOKIE.apply(response, session.token) log_security_event("auth.login_succeeded", audience="internal", role=user.role.name if user.role else None, user_id=user.id) return session.model_copy(update={"token": None}) @router.get("/me", response_model=UserSession) def read_me(user: User = Depends(get_current_user)): """Return the current user with permission keys for UI navigation gating.""" return _serialize_session(user).model_copy(update={"token": None}) @router.get("/me/permissions", response_model=list[str]) def read_my_permissions(user: User = Depends(get_current_user)): return sorted(get_user_permissions(user)) class UpdateMeRequest(BaseModel): name: str | None = None email: str | None = None current_password: str | None = None new_password: str | None = None @router.patch("/me", response_model=UserSession) def update_me( payload: UpdateMeRequest, db: Session = Depends(get_db), user: User = Depends(get_current_user), ): """Allow an internal user to update their own name, email, or password.""" if payload.new_password: # Require current password verification before allowing a password # change. Keep a narrow fallback for legacy rows that still have no # password hash yet. current_ok = verify_password(payload.current_password or "", user.password_hash) or ( user.password_hash is None and (payload.current_password or "") == settings.admin_password ) if not current_ok: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Current password is incorrect", ) if len(payload.new_password) < 8: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="New password must be at least 8 characters", ) user.password_hash = hash_password(payload.new_password) if payload.name is not None: name = payload.name.strip() if not name: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Name cannot be empty") user.name = name if payload.email is not None: email = payload.email.strip().lower() if not email or "@" not in email: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid email address") existing = db.scalar(select(User).where(User.email == email, User.id != user.id)) if existing: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email is already in use") user.email = email db.commit() db.refresh(user) return _serialize_session(user, include_token=True).model_copy(update={"token": None}) @router.post("/logout", status_code=status.HTTP_204_NO_CONTENT) def logout(response: Response): CLIENT_AUTH_COOKIE.clear(response) response.status_code = status.HTTP_204_NO_CONTENT return None # Permission-enforced administrative endpoints. Route bodies should not check # role names — every gate is the require_permission(...) dependency. @router.get("/users", response_model=list[UserRead]) def list_users( db: Session = Depends(get_db), _: User = Depends(require_permission("view_users")), # gated by permission key ): users = db.scalars( select(User).options(selectinload(User.role)).order_by(User.name) ).all() return [_serialize_user_read(user) for user in users] @router.get("/assignable-roles", response_model=list[AssignableRole]) def list_assignable_roles( db: Session = Depends(get_db), _: User = Depends(require_permission("manage_users")), # gated by permission key ): """Roles that a user-manager can assign — used to populate the role picker. Separate from ``/roles`` (which exposes full permission sets and is gated by ``manage_permissions``); managing users only needs the role list itself. """ roles = db.scalars(select(Role).order_by(Role.name)).all() return [ AssignableRole(id=role.id, name=role.name, description=role.description) for role in roles ] def _load_managed_user(db: Session, user_id: int) -> User: user = db.scalar( select(User).where(User.id == user_id).options(selectinload(User.role)) ) if user is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") return user def _resolve_role(db: Session, role_id: int | None) -> Role | None: if role_id is None: return None role = db.scalar(select(Role).where(Role.id == role_id)) if role is None: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Unknown role") return role @router.post("/users", response_model=UserRead, status_code=status.HTTP_201_CREATED) def create_user( payload: CreateUserRequest, db: Session = Depends(get_db), actor: User = Depends(require_permission("manage_users")), # gated by permission key ): """Create a new internal user. A user with no password can still sign in with the shared internal password until they set a personal one in their own settings. """ email = payload.email.strip().lower() if not email or "@" not in email: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid email address") name = payload.name.strip() if not name: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Name cannot be empty") if db.scalar(select(User).where(User.email == email)): raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email is already in use") role = _resolve_role(db, payload.role_id) password_hash = None if payload.password: if len(payload.password) < 8: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Password must be at least 8 characters", ) password_hash = hash_password(payload.password) user = User( email=email, name=name, role_id=role.id if role else None, is_active=payload.is_active, password_hash=password_hash, ) db.add(user) db.commit() db.refresh(user) log_security_event("users.created", audience="internal", actor_user_id=actor.id, user_id=user.id) return _serialize_user_read(user) @router.patch("/users/{user_id}", response_model=UserRead) def update_user( user_id: int, payload: AdminUpdateUserRequest, db: Session = Depends(get_db), actor: User = Depends(require_permission("manage_users")), # gated by permission key ): """Update another user's name, email, role, or active status.""" user = _load_managed_user(db, user_id) if payload.is_active is not None: # Guard against locking yourself out of your own management session. if user.id == actor.id and not payload.is_active: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="You cannot deactivate your own account", ) user.is_active = payload.is_active if payload.name is not None: name = payload.name.strip() if not name: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Name cannot be empty") user.name = name if payload.email is not None: email = payload.email.strip().lower() if not email or "@" not in email: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid email address") existing = db.scalar(select(User).where(User.email == email, User.id != user.id)) if existing: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email is already in use") user.email = email if payload.role_id is not None: role = _resolve_role(db, payload.role_id) user.role_id = role.id if role else None db.commit() db.refresh(user) log_security_event("users.updated", audience="internal", actor_user_id=actor.id, user_id=user.id) return _serialize_user_read(user) @router.post("/users/{user_id}/password", response_model=UserRead) def set_user_password( user_id: int, payload: AdminSetPasswordRequest, db: Session = Depends(get_db), actor: User = Depends(require_permission("manage_users")), # gated by permission key ): """Set (reset) another user's password without their current password.""" user = _load_managed_user(db, user_id) if len(payload.new_password) < 8: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="New password must be at least 8 characters", ) user.password_hash = hash_password(payload.new_password) db.commit() db.refresh(user) log_security_event("users.password_reset", audience="internal", actor_user_id=actor.id, user_id=user.id) return _serialize_user_read(user) @router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_user( user_id: int, response: Response, db: Session = Depends(get_db), actor: User = Depends(require_permission("manage_users")), # gated by permission key ): """Delete a user. Lean owner accounts and your own account are protected.""" user = _load_managed_user(db, user_id) if user.id == actor.id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="You cannot delete your own account", ) if (user.role.name if user.role else "").lower() == LEAN_ROLE_NAME: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Lean accounts cannot be deleted", ) db.delete(user) db.commit() log_security_event("users.deleted", audience="internal", actor_user_id=actor.id, user_id=user_id) response.status_code = status.HTTP_204_NO_CONTENT return None @router.get("/roles", response_model=list[RoleRead]) def list_roles( db: Session = Depends(get_db), _: User = Depends(_require_role_management_actor), ): user_counts = dict( db.execute(select(User.role_id, func.count(User.id)).group_by(User.role_id)).all() ) roles = db.scalars( select(Role).options(selectinload(Role.permissions)).order_by(Role.name) ).all() return [_serialize_role_read(role, user_count=user_counts.get(role.id, 0)) for role in roles] @router.get("/role-modules", response_model=list[RoleModuleDefinition]) def list_role_modules(_: User = Depends(_require_role_management_actor)): return _module_definitions_response() @router.post("/roles", response_model=RoleRead, status_code=status.HTTP_201_CREATED) def create_role( payload: CreateRoleRequest, db: Session = Depends(get_db), actor: User = Depends(_require_role_management_actor), ): name = payload.name.strip() if not name: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Role name cannot be empty") existing = db.scalar(select(Role).where(func.lower(Role.name) == name.lower())) if existing: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Role name already exists") permissions_by_key = {permission.key: permission for permission in db.scalars(select(Permission)).all()} role = Role(name=name, description=None) db.add(role) db.flush() _apply_role_updates( role, name=name, description=payload.description, module_permissions=payload.module_permissions, permissions_by_key=permissions_by_key, ) db.commit() db.refresh(role) log_security_event("roles.created", audience="internal", actor_user_id=actor.id, role_id=role.id) return _serialize_role_read(role, user_count=0) @router.patch("/roles/{role_id}", response_model=RoleRead) def update_role( role_id: int, payload: UpdateRoleRequest, db: Session = Depends(get_db), actor: User = Depends(_require_role_management_actor), ): role = _load_role(db, role_id) original_name = role.name protected = _is_protected_role_name(original_name) requested_name = payload.name.strip() if payload.name is not None else role.name if protected and requested_name.lower() != original_name.lower(): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Lean and admin roles cannot be renamed", ) if payload.name is not None: duplicate = db.scalar( select(Role).where(func.lower(Role.name) == requested_name.lower(), Role.id != role.id) ) if duplicate: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Role name already exists") permissions_by_key = {permission.key: permission for permission in db.scalars(select(Permission)).all()} _apply_role_updates( role, name=payload.name, description=payload.description, module_permissions=payload.module_permissions, permissions_by_key=permissions_by_key, ) db.commit() db.refresh(role) user_count = db.scalar(select(func.count(User.id)).where(User.role_id == role.id)) or 0 log_security_event("roles.updated", audience="internal", actor_user_id=actor.id, role_id=role.id) return _serialize_role_read(role, user_count=user_count) @router.delete("/roles/{role_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_role( role_id: int, response: Response, db: Session = Depends(get_db), actor: User = Depends(_require_role_management_actor), ): role = _load_role(db, role_id) if _is_protected_role_name(role.name): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Lean and admin roles cannot be deleted", ) assigned_users = db.scalar(select(func.count(User.id)).where(User.role_id == role.id)) or 0 if assigned_users > 0: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Reassign users before deleting this role", ) db.delete(role) db.commit() log_security_event("roles.deleted", audience="internal", actor_user_id=actor.id, role_id=role_id) response.status_code = status.HTTP_204_NO_CONTENT return None @router.get("/permissions", response_model=list[str]) def list_permissions( db: Session = Depends(get_db), _: User = Depends(require_permission("manage_permissions")), # gated by permission key ): return sorted(p.key for p in db.scalars(select(Permission)).all())