feat(auth): Add optional authentication system (#22)
Some checks failed
CI / style (push) Has been cancelled
CI / build-client (push) Has been cancelled
CI / build-amd64 (push) Has been cancelled
CI / build-tester (push) Has been cancelled
CI / test (cockroachdb) (push) Has been cancelled
CI / test (mariadb) (push) Has been cancelled
CI / test (postgres) (push) Has been cancelled
CI / test (sqlite) (push) Has been cancelled
CI / build-arm64 (push) Has been cancelled
CI / build-armv7 (push) Has been cancelled
CI / publish-images (push) Has been cancelled
CI / publish-release (push) Has been cancelled

Backend:
- Add User model with username, hashed password, role, is_active
- Add passlib[bcrypt] and python-jose[cryptography] dependencies
- Create auth.py with JWT utilities, password hashing, role checking
- Add /auth endpoints: status, setup, login, me, users CRUD
- Add role-based permission system (viewer/operator/editor/admin)
- Environment: SPOOLMAN_AUTH_ENABLED, SPOOLMAN_AUTH_SECRET_KEY
- Migration for user table

Frontend:
- Add AuthProvider context with login/logout/setup
- Add LoginPage component with setup mode for first user
- Add ProtectedRoute wrapper for auth-required pages
- Add axios interceptor to attach JWT token to requests
- Add User Management tab in Settings (admin only)
- Add translation keys for auth UI

When SPOOLMAN_AUTH_ENABLED=true:
- Users must login to access the app
- First visit shows setup page to create admin account
- Role-based access control for future endpoint protection

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-16 00:15:36 -06:00
parent 94612a6dd3
commit 3613e7739a
15 changed files with 1250 additions and 12 deletions

211
spoolman/auth.py Normal file
View File

@@ -0,0 +1,211 @@
"""Authentication utilities and dependencies."""
import logging
import os
import secrets
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Annotated, Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from spoolman.database import models
from spoolman.database.database import get_db_session
logger = logging.getLogger(__name__)
# JWT Configuration
SECRET_KEY = os.environ.get("SPOOLMAN_AUTH_SECRET_KEY", secrets.token_urlsafe(32))
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.environ.get("SPOOLMAN_AUTH_TOKEN_EXPIRE_MINUTES", "1440")) # 24 hours default
# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# OAuth2 scheme - auto_error=False allows unauthenticated access when auth is disabled
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/v1/auth/login", auto_error=False)
def is_auth_enabled() -> bool:
"""Check if authentication is enabled via environment variable."""
return os.environ.get("SPOOLMAN_AUTH_ENABLED", "false").lower() in ("true", "1", "yes")
class Role(str, Enum):
"""User roles with increasing permissions."""
VIEWER = "viewer" # Can only view
OPERATOR = "operator" # Can view + record usage
EDITOR = "editor" # Can view + use + create + edit
ADMIN = "admin" # Full access
# Role hierarchy - each role includes permissions of roles below it
ROLE_HIERARCHY = {
Role.VIEWER: 0,
Role.OPERATOR: 1,
Role.EDITOR: 2,
Role.ADMIN: 3,
}
class Token(BaseModel):
"""JWT token response."""
access_token: str
token_type: str = "bearer"
class TokenData(BaseModel):
"""Data extracted from JWT token."""
username: str
role: Role
class UserResponse(BaseModel):
"""User data returned by API (no password)."""
id: int
username: str
role: Role
is_active: bool
created_at: datetime
@staticmethod
def from_db(user: models.User) -> "UserResponse":
"""Convert database model to response model."""
return UserResponse(
id=user.id,
username=user.username,
role=Role(user.role),
is_active=user.is_active,
created_at=user.created_at,
)
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Verify a plain password against a hashed password."""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""Hash a password."""
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""Create a JWT access token."""
to_encode = data.copy()
expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
async def get_user_by_username(db: AsyncSession, username: str) -> Optional[models.User]:
"""Get a user by username."""
result = await db.execute(select(models.User).where(models.User.username == username))
return result.scalar_one_or_none()
async def authenticate_user(db: AsyncSession, username: str, password: str) -> Optional[models.User]:
"""Authenticate a user by username and password."""
user = await get_user_by_username(db, username)
if not user or not verify_password(password, user.hashed_password):
return None
if not user.is_active:
return None
return user
async def get_current_user(
token: Annotated[Optional[str], Depends(oauth2_scheme)],
db: Annotated[AsyncSession, Depends(get_db_session)],
) -> Optional[models.User]:
"""Get the current authenticated user from JWT token.
Returns None if auth is disabled or no token provided.
Raises HTTPException if token is invalid.
"""
if not is_auth_enabled():
return None
if token is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token",
headers={"WWW-Authenticate": "Bearer"},
)
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token",
headers={"WWW-Authenticate": "Bearer"},
)
user = await get_user_by_username(db, username)
if user is None or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found or inactive",
headers={"WWW-Authenticate": "Bearer"},
)
return user
def require_role(minimum_role: Role):
"""Dependency factory that requires a minimum role level.
Usage:
@router.post("/spool", dependencies=[Depends(require_role(Role.EDITOR))])
async def create_spool(...):
...
"""
async def role_checker(
current_user: Annotated[Optional[models.User], Depends(get_current_user)],
) -> Optional[models.User]:
# If auth is disabled, allow all operations
if not is_auth_enabled():
return None
if current_user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
)
user_role = Role(current_user.role)
if ROLE_HIERARCHY[user_role] < ROLE_HIERARCHY[minimum_role]:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Requires {minimum_role.value} role or higher",
)
return current_user
return role_checker
# Common dependencies for different permission levels
RequireViewer = Depends(require_role(Role.VIEWER))
RequireOperator = Depends(require_role(Role.OPERATOR))
RequireEditor = Depends(require_role(Role.EDITOR))
RequireAdmin = Depends(require_role(Role.ADMIN))