bvandeusen
f77b029943
- Registration auto-closes after first user; admin can toggle from /admin/users - Admin user management view with user list and delete - Password confirmation on registration form - Password change in Settings (PUT /api/auth/password) - Session cookie hardening: HttpOnly, SameSite=Lax, optional Secure flag - Startup warning when SECRET_KEY is default - Production deployment docs: reverse proxy, rate limiting, CSP headers - Fix assist prompt to preserve markdown headings in target sections - Simplify prod compose networking Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
145 lines
4.9 KiB
Python
145 lines
4.9 KiB
Python
import logging
|
|
|
|
import bcrypt
|
|
from sqlalchemy import func, select, update
|
|
|
|
from fabledassistant.models import async_session
|
|
from fabledassistant.models.conversation import Conversation
|
|
from fabledassistant.models.note import Note
|
|
from fabledassistant.models.setting import Setting
|
|
from fabledassistant.models.user import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
|
|
|
|
|
|
def verify_password(password: str, password_hash: str) -> bool:
|
|
return bcrypt.checkpw(password.encode(), password_hash.encode())
|
|
|
|
|
|
async def get_user_count() -> int:
|
|
async with async_session() as session:
|
|
return await session.scalar(select(func.count(User.id))) or 0
|
|
|
|
|
|
async def create_user(
|
|
username: str, password: str, email: str | None = None
|
|
) -> User:
|
|
user_count = await get_user_count()
|
|
role = "admin" if user_count == 0 else "user"
|
|
|
|
async with async_session() as session:
|
|
user = User(
|
|
username=username,
|
|
email=email,
|
|
password_hash=hash_password(password),
|
|
role=role,
|
|
)
|
|
session.add(user)
|
|
await session.commit()
|
|
await session.refresh(user)
|
|
|
|
# First user claims all pre-existing data (migration assigns user_id=1,
|
|
# which matches SERIAL first insert; also handles any NULLs)
|
|
if role == "admin":
|
|
await session.execute(
|
|
update(Note)
|
|
.where((Note.user_id.is_(None)) | (Note.user_id == user.id))
|
|
.values(user_id=user.id)
|
|
)
|
|
await session.execute(
|
|
update(Conversation)
|
|
.where(
|
|
(Conversation.user_id.is_(None))
|
|
| (Conversation.user_id == user.id)
|
|
)
|
|
.values(user_id=user.id)
|
|
)
|
|
await session.execute(
|
|
update(Setting)
|
|
.where(Setting.user_id == user.id)
|
|
.values(user_id=user.id)
|
|
)
|
|
# Auto-close registration after first user setup
|
|
session.add(Setting(user_id=user.id, key="registration_open", value="false"))
|
|
await session.commit()
|
|
logger.info("First user '%s' created as admin, claimed orphaned data", username)
|
|
|
|
return user
|
|
|
|
|
|
async def authenticate(username: str, password: str) -> User | None:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(User).where(User.username == username)
|
|
)
|
|
user = result.scalars().first()
|
|
if user and verify_password(password, user.password_hash):
|
|
return user
|
|
return None
|
|
|
|
|
|
async def get_user_by_id(user_id: int) -> User | None:
|
|
async with async_session() as session:
|
|
return await session.get(User, user_id)
|
|
|
|
|
|
async def change_password(user_id: int, current_password: str, new_password: str) -> bool:
|
|
"""Change a user's password. Returns True on success, False if current password is wrong."""
|
|
async with async_session() as session:
|
|
user = await session.get(User, user_id)
|
|
if not user:
|
|
return False
|
|
if not verify_password(current_password, user.password_hash):
|
|
return False
|
|
user.password_hash = hash_password(new_password)
|
|
await session.commit()
|
|
logger.info("Password changed for user %d (%s)", user_id, user.username)
|
|
return True
|
|
|
|
|
|
async def is_registration_open() -> bool:
|
|
"""Check if new user registration is allowed.
|
|
|
|
Always open when no users exist (first-user setup).
|
|
Otherwise reads the admin's 'registration_open' setting (default closed).
|
|
"""
|
|
user_count = await get_user_count()
|
|
if user_count == 0:
|
|
return True
|
|
|
|
async with async_session() as session:
|
|
# Find the admin user's registration_open setting
|
|
result = await session.execute(
|
|
select(Setting)
|
|
.join(User, Setting.user_id == User.id)
|
|
.where(User.role == "admin", Setting.key == "registration_open")
|
|
)
|
|
setting = result.scalar_one_or_none()
|
|
return setting.value == "true" if setting else False
|
|
|
|
|
|
async def list_users() -> list[User]:
|
|
async with async_session() as session:
|
|
result = await session.execute(select(User).order_by(User.created_at))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def delete_user(user_id: int) -> bool:
|
|
async with async_session() as session:
|
|
user = await session.get(User, user_id)
|
|
if not user:
|
|
return False
|
|
await session.delete(user)
|
|
await session.commit()
|
|
logger.info("Deleted user %d (%s)", user_id, user.username)
|
|
return True
|
|
|
|
|
|
async def set_registration_open(admin_user_id: int, open: bool) -> None:
|
|
from fabledassistant.services.settings import set_setting
|
|
await set_setting(admin_user_id, "registration_open", "true" if open else "false")
|