feat: replace admin auth with cookie-based profile picker
Remove all authentication (login, sessions, bcrypt, itsdangerous) since the app runs on a private homelab LAN. Replace with a profile picker landing page and cookie-based profile selection (1-year expiry). - Add Alembic migration to drop password_hash/is_admin columns - Delete auth service, auth routes, login template, and auth tests - Rewrite app/utils/auth.py with NoProfileSelectedError and require_active_profile dependency - Add profile creation flow (GET/POST /profiles/create) - Rewrite home page as profile picker with card layout - Update all route files to use profile dependency instead of admin auth - Remove bcrypt and itsdangerous from requirements - Remove admin_username/admin_password from config - Update all tests for new profile-based access model Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,100 +0,0 @@
|
||||
"""Service layer for admin authentication and session management.
|
||||
|
||||
Handles password verification, session token creation/validation.
|
||||
Uses bcrypt for password hashing and itsdangerous for session signing.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import bcrypt
|
||||
import structlog
|
||||
from itsdangerous import BadSignature, URLSafeTimedSerializer
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.models.user import User
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
# Session token max age: 24 hours
|
||||
SESSION_MAX_AGE_SECONDS = 86400
|
||||
|
||||
|
||||
class AuthService:
|
||||
"""Handles admin authentication and session token management.
|
||||
|
||||
Args:
|
||||
session: An active SQLModel Session.
|
||||
secret_key: Secret key for signing session tokens.
|
||||
"""
|
||||
|
||||
def __init__(self, session: Session, secret_key: str) -> None:
|
||||
self._session = session
|
||||
self._serializer = URLSafeTimedSerializer(secret_key)
|
||||
|
||||
def authenticate(self, username: str, password: str) -> Optional[User]:
|
||||
"""Verify admin credentials and return the User if valid.
|
||||
|
||||
Only admin users can authenticate. Non-admin users are rejected.
|
||||
|
||||
Args:
|
||||
username: The username to check.
|
||||
password: The plaintext password to verify.
|
||||
|
||||
Returns:
|
||||
The authenticated User, or None if credentials are invalid.
|
||||
"""
|
||||
statement = select(User).where(User.username == username)
|
||||
user = self._session.exec(statement).first()
|
||||
|
||||
if user is None:
|
||||
logger.warning("auth_failed", reason="user_not_found", username=username)
|
||||
return None
|
||||
|
||||
if not user.is_admin:
|
||||
logger.warning("auth_failed", reason="not_admin", username=username)
|
||||
return None
|
||||
|
||||
if not user.password_hash:
|
||||
logger.warning("auth_failed", reason="no_password_hash", username=username)
|
||||
return None
|
||||
|
||||
# Verify password against bcrypt hash
|
||||
if not bcrypt.checkpw(
|
||||
password.encode("utf-8"),
|
||||
user.password_hash.encode("utf-8"),
|
||||
):
|
||||
logger.warning("auth_failed", reason="wrong_password", username=username)
|
||||
return None
|
||||
|
||||
logger.info("auth_success", username=username)
|
||||
return user
|
||||
|
||||
def create_session_token(self, user_id: int) -> str:
|
||||
"""Create a signed session token for the given user.
|
||||
|
||||
Args:
|
||||
user_id: The authenticated user's ID.
|
||||
|
||||
Returns:
|
||||
A signed, URL-safe token string.
|
||||
"""
|
||||
return self._serializer.dumps({"user_id": user_id})
|
||||
|
||||
def validate_session_token(self, token: str) -> Optional[int]:
|
||||
"""Validate a session token and extract the user_id.
|
||||
|
||||
Args:
|
||||
token: The session token to validate.
|
||||
|
||||
Returns:
|
||||
The user_id if the token is valid, or None.
|
||||
"""
|
||||
try:
|
||||
data = self._serializer.loads(token, max_age=SESSION_MAX_AGE_SECONDS)
|
||||
return data.get("user_id")
|
||||
except BadSignature:
|
||||
logger.warning("session_invalid", reason="bad_signature")
|
||||
return None
|
||||
except Exception:
|
||||
logger.warning("session_invalid", reason="unknown_error")
|
||||
return None
|
||||
@@ -4,11 +4,9 @@ Reads config/exercises.yaml and config/user_programs.yaml to populate
|
||||
the exercise library, warmups, workout days, and user programs.
|
||||
"""
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import bcrypt
|
||||
import structlog
|
||||
import yaml
|
||||
from sqlmodel import Session, select
|
||||
@@ -43,14 +41,7 @@ class SeedService:
|
||||
self._config_dir = config_dir or Path(__file__).resolve().parent.parent.parent / "config"
|
||||
|
||||
def _load_yaml(self, filename: str) -> dict:
|
||||
"""Load and parse a YAML file from the config directory.
|
||||
|
||||
Args:
|
||||
filename: Name of the YAML file to load.
|
||||
|
||||
Returns:
|
||||
Parsed YAML content as a dictionary.
|
||||
"""
|
||||
"""Load and parse a YAML file from the config directory."""
|
||||
filepath = self._config_dir / filename
|
||||
with open(filepath, "r") as f:
|
||||
return yaml.safe_load(f)
|
||||
@@ -111,48 +102,8 @@ class SeedService:
|
||||
self._session.commit()
|
||||
logger.info("seed_complete", table="warmups", count=len(warmups))
|
||||
|
||||
def seed_admin(self) -> None:
|
||||
"""Create admin user from environment variables if not exists.
|
||||
|
||||
Reads ADMIN_USERNAME and ADMIN_PASSWORD from env,
|
||||
hashes the password with bcrypt, and creates the admin user.
|
||||
"""
|
||||
admin_username = os.environ.get("ADMIN_USERNAME", "admin")
|
||||
admin_password = os.environ.get("ADMIN_PASSWORD", "")
|
||||
|
||||
existing = self._session.exec(
|
||||
select(User).where(User.username == admin_username)
|
||||
).first()
|
||||
if existing:
|
||||
logger.info("seed_skipped", table="users", reason="admin already exists")
|
||||
return
|
||||
|
||||
if not admin_password:
|
||||
logger.warning("seed_skipped", table="users", reason="ADMIN_PASSWORD not set")
|
||||
return
|
||||
|
||||
# Hash password with bcrypt
|
||||
password_hash = bcrypt.hashpw(
|
||||
admin_password.encode("utf-8"),
|
||||
bcrypt.gensalt(),
|
||||
).decode("utf-8")
|
||||
|
||||
admin = User(
|
||||
username=admin_username,
|
||||
password_hash=password_hash,
|
||||
display_name="Admin",
|
||||
is_admin=True,
|
||||
)
|
||||
self._session.add(admin)
|
||||
self._session.commit()
|
||||
logger.info("seed_complete", table="users", user="admin")
|
||||
|
||||
def seed_user_programs(self) -> None:
|
||||
"""Seed user profiles and their exercise programs from user_programs.yaml.
|
||||
|
||||
Creates non-admin user profiles and links them to exercises
|
||||
with week 1/4 rep and weight targets.
|
||||
"""
|
||||
"""Seed user profiles and their exercise programs from user_programs.yaml."""
|
||||
data = self._load_yaml("user_programs.yaml")
|
||||
programs = data.get("programs", [])
|
||||
|
||||
@@ -172,12 +123,10 @@ class SeedService:
|
||||
else:
|
||||
user = User(
|
||||
username=username,
|
||||
password_hash="", # Non-admin users don't log in initially
|
||||
display_name=display_name,
|
||||
height=profile.get("height", ""),
|
||||
weight=profile.get("weight", ""),
|
||||
goals=profile.get("goals", ""),
|
||||
is_admin=False,
|
||||
)
|
||||
self._session.add(user)
|
||||
self._session.commit()
|
||||
@@ -219,15 +168,10 @@ class SeedService:
|
||||
logger.info("seed_complete", table="user_exercise_programs", user=username)
|
||||
|
||||
def seed_all(self) -> None:
|
||||
"""Run all seed operations in the correct order.
|
||||
|
||||
Order matters: workout_days and exercises must exist before
|
||||
user_programs can reference them.
|
||||
"""
|
||||
"""Run all seed operations in the correct order."""
|
||||
logger.info("seed_all_started")
|
||||
self.seed_workout_days()
|
||||
self.seed_exercises()
|
||||
self.seed_warmups()
|
||||
self.seed_admin()
|
||||
self.seed_user_programs()
|
||||
logger.info("seed_all_complete")
|
||||
|
||||
@@ -27,77 +27,48 @@ class UserService:
|
||||
def create_user(
|
||||
self,
|
||||
username: str,
|
||||
password_hash: str,
|
||||
display_name: str,
|
||||
height: Optional[str] = None,
|
||||
weight: Optional[str] = None,
|
||||
goals: Optional[str] = None,
|
||||
is_admin: bool = False,
|
||||
) -> User:
|
||||
"""Create a new user profile.
|
||||
|
||||
Args:
|
||||
username: Unique login identifier.
|
||||
password_hash: Pre-hashed password string.
|
||||
username: Unique identifier.
|
||||
display_name: Human-readable name.
|
||||
height: User height as string.
|
||||
weight: User weight as string.
|
||||
goals: Free-text goals.
|
||||
is_admin: Whether user has admin privileges.
|
||||
|
||||
Returns:
|
||||
The newly created User record.
|
||||
"""
|
||||
user = User(
|
||||
username=username,
|
||||
password_hash=password_hash,
|
||||
display_name=display_name,
|
||||
height=height,
|
||||
weight=weight,
|
||||
goals=goals,
|
||||
is_admin=is_admin,
|
||||
)
|
||||
self._session.add(user)
|
||||
self._session.commit()
|
||||
self._session.refresh(user)
|
||||
logger.info("user_created", username=username, is_admin=is_admin)
|
||||
logger.info("user_created", username=username)
|
||||
return user
|
||||
|
||||
def get_user_by_id(self, user_id: int) -> Optional[User]:
|
||||
"""Retrieve a user by primary key.
|
||||
|
||||
Args:
|
||||
user_id: The user's ID.
|
||||
|
||||
Returns:
|
||||
The User record, or None if not found.
|
||||
"""
|
||||
"""Retrieve a user by primary key."""
|
||||
return self._session.get(User, user_id)
|
||||
|
||||
def get_user_by_username(self, username: str) -> Optional[User]:
|
||||
"""Retrieve a user by username.
|
||||
|
||||
Args:
|
||||
username: The username to look up.
|
||||
|
||||
Returns:
|
||||
The User record, or None if not found.
|
||||
"""
|
||||
"""Retrieve a user by username."""
|
||||
statement = select(User).where(User.username == username)
|
||||
return self._session.exec(statement).first()
|
||||
|
||||
def list_users(self, exclude_admin: bool = False) -> list[User]:
|
||||
"""List all user profiles.
|
||||
|
||||
Args:
|
||||
exclude_admin: If True, omit admin users from the result.
|
||||
|
||||
Returns:
|
||||
List of User records.
|
||||
"""
|
||||
def list_users(self) -> list[User]:
|
||||
"""List all user profiles."""
|
||||
statement = select(User)
|
||||
if exclude_admin:
|
||||
statement = statement.where(User.is_admin == False) # noqa: E712
|
||||
return list(self._session.exec(statement).all())
|
||||
|
||||
def update_user(self, user_id: int, **kwargs) -> User:
|
||||
|
||||
Reference in New Issue
Block a user