PapaTramp
56c469ddbd
Add sac_users table, JWT roles, settings/users API guarded for admin, Users page in UI, and sac_manage_user.py CLI. Co-authored-by: Cursor <cursoragent@cursor.com>
97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
"""SAC UI users: password hashing, authentication, bootstrap."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from passlib.context import CryptContext
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import get_settings
|
|
from app.models.user import USER_ROLE_ADMIN, USER_ROLE_MONITOR, USER_ROLES, User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
|
|
|
|
|
def hash_password(plain: str) -> str:
|
|
return _pwd_context.hash(plain)
|
|
|
|
|
|
def verify_password(plain: str, password_hash: str) -> bool:
|
|
try:
|
|
return _pwd_context.verify(plain, password_hash)
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def get_user_by_username(db: Session, username: str) -> User | None:
|
|
normalized = username.strip()
|
|
if not normalized:
|
|
return None
|
|
return db.scalar(
|
|
select(User).where(
|
|
func.lower(User.username) == normalized.lower(),
|
|
User.is_active.is_(True),
|
|
)
|
|
)
|
|
|
|
|
|
def authenticate_user(db: Session, username: str, password: str) -> User | None:
|
|
user = get_user_by_username(db, username)
|
|
if user is None:
|
|
return None
|
|
if not verify_password(password, user.password_hash):
|
|
return None
|
|
return user
|
|
|
|
|
|
def create_user(
|
|
db: Session,
|
|
*,
|
|
username: str,
|
|
password: str,
|
|
role: str = USER_ROLE_MONITOR,
|
|
) -> User:
|
|
normalized = username.strip()
|
|
if not normalized:
|
|
raise ValueError("username is required")
|
|
if role not in USER_ROLES:
|
|
raise ValueError(f"invalid role: {role}")
|
|
if len(password) < 8:
|
|
raise ValueError("password must be at least 8 characters")
|
|
existing = db.scalar(select(User).where(func.lower(User.username) == normalized.lower()))
|
|
if existing is not None:
|
|
raise ValueError("username already exists")
|
|
user = User(
|
|
username=normalized,
|
|
password_hash=hash_password(password),
|
|
role=role,
|
|
is_active=True,
|
|
)
|
|
db.add(user)
|
|
db.flush()
|
|
return user
|
|
|
|
|
|
def bootstrap_admin_user(db: Session) -> None:
|
|
"""Create first admin from SAC_ADMIN_* when sac_users table is empty."""
|
|
try:
|
|
count = db.scalar(select(func.count()).select_from(User)) or 0
|
|
except Exception as exc:
|
|
logger.warning("SAC users bootstrap skipped (run alembic upgrade head): %s", exc)
|
|
db.rollback()
|
|
return
|
|
if count > 0:
|
|
return
|
|
settings = get_settings()
|
|
if not settings.sac_admin_password:
|
|
logger.warning("SAC users table empty and SAC_ADMIN_PASSWORD not set — UI login disabled until a user is created")
|
|
return
|
|
username = (settings.sac_admin_username or "admin").strip() or "admin"
|
|
create_user(db, username=username, password=settings.sac_admin_password, role=USER_ROLE_ADMIN)
|
|
db.commit()
|
|
logger.info("SAC bootstrap: created admin user %r from SAC_ADMIN_* env", username)
|