Files
Priceflow/app/services/auth_service.py
T
Claude 1dd6d4dae1 feat: Add authentication system and admin panel
- Add User model with authentication
- Create JWT-based auth service
- Add login page with admin/admin default credentials
- Create admin panel with user management, settings, and site configuration
- Move settings to admin section (admin only)
- Change search page from multi-site checkboxes to single site selector
- Add logout functionality to layout
- Protect routes with authentication
2025-11-23 09:15:56 +00:00

151 lines
4.0 KiB
Python

"""
Authentication service for user management and JWT tokens.
"""
import hashlib
import os
import secrets
from datetime import UTC, datetime, timedelta
import jwt
from sqlalchemy.orm import Session
from app import models
# JWT Configuration
JWT_SECRET = os.getenv("JWT_SECRET", secrets.token_hex(32))
JWT_ALGORITHM = "HS256"
JWT_EXPIRATION_HOURS = 24
# Default admin credentials
DEFAULT_ADMIN_USERNAME = "admin"
DEFAULT_ADMIN_PASSWORD = "admin"
def hash_password(password: str) -> str:
"""Hash password using SHA-256 with salt."""
salt = secrets.token_hex(16)
password_hash = hashlib.sha256(f"{salt}{password}".encode()).hexdigest()
return f"{salt}:{password_hash}"
def verify_password(password: str, password_hash: str) -> bool:
"""Verify password against hash."""
try:
salt, stored_hash = password_hash.split(":")
computed_hash = hashlib.sha256(f"{salt}{password}".encode()).hexdigest()
return computed_hash == stored_hash
except ValueError:
return False
def create_token(user_id: int, username: str, is_admin: bool) -> str:
"""Create JWT token for authenticated user."""
payload = {
"user_id": user_id,
"username": username,
"is_admin": is_admin,
"exp": datetime.now(UTC) + timedelta(hours=JWT_EXPIRATION_HOURS),
"iat": datetime.now(UTC),
}
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
def decode_token(token: str) -> dict | None:
"""Decode and validate JWT token."""
try:
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
return payload
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None
def get_user_by_username(db: Session, username: str) -> models.User | None:
"""Get user by username."""
return db.query(models.User).filter(models.User.username == username).first()
def get_user_by_id(db: Session, user_id: int) -> models.User | None:
"""Get user by ID."""
return db.query(models.User).filter(models.User.id == user_id).first()
def authenticate_user(db: Session, username: str, password: str) -> models.User | None:
"""Authenticate user with username and password."""
user = get_user_by_username(db, username)
if not user:
return None
if not verify_password(password, user.password_hash):
return None
if not user.is_active:
return None
return user
def create_user(
db: Session,
username: str,
password: str,
is_admin: bool = False,
) -> models.User:
"""Create a new user."""
password_hash = hash_password(password)
user = models.User(
username=username,
password_hash=password_hash,
is_admin=is_admin,
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
return user
def update_password(db: Session, user: models.User, new_password: str) -> None:
"""Update user password."""
user.password_hash = hash_password(new_password)
db.commit()
def update_last_login(db: Session, user: models.User) -> None:
"""Update user's last login timestamp."""
user.last_login = datetime.now(UTC)
db.commit()
def get_all_users(db: Session) -> list[models.User]:
"""Get all users."""
return db.query(models.User).all()
def delete_user(db: Session, user_id: int) -> bool:
"""Delete a user by ID."""
user = get_user_by_id(db, user_id)
if not user:
return False
db.delete(user)
db.commit()
return True
def toggle_user_active(db: Session, user_id: int) -> models.User | None:
"""Toggle user active status."""
user = get_user_by_id(db, user_id)
if not user:
return None
user.is_active = not user.is_active
db.commit()
db.refresh(user)
return user
def seed_default_admin(db: Session) -> bool:
"""Create default admin user if no users exist."""
user_count = db.query(models.User).count()
if user_count == 0:
create_user(db, DEFAULT_ADMIN_USERNAME, DEFAULT_ADMIN_PASSWORD, is_admin=True)
return True
return False