Complete architecture simplification: 54+ files → 37 files
- Removed 11 duplicate router files (auth.py, users.py, decks.py, card_router.py, interactions.py, user_data.py, rooms.py, games/router.py, refresh.py, card_import.py, ws.py) - Removed duplicate core files (logging_config.py, settings.py) - Removed unused services (card_mirror_service.py, mtgjson_loader.py, mtgjson_uploader.py) - Removed unused schemas (proto_messages.py, protocol_constants.py) - Created run.py startup script for simple execution - Created app/core/config.py for unified configuration management - Created app/core/logging.py for logging configuration - Consolidated routers into 6 focused files (auth_router.py, users_router.py, decks_router.py, cards_router.py, user_data_router.py, games_router.py) - Updated all imports to use consolidated model paths - Added backward compatibility aliases in database.py - Created SIMPLIFICATION_SUMMARY.md documentation Architecture improvements: - Single unified database engine with auto-detection (SQLite/PostgreSQL) - Consolidated models into 2 files (models.py, mtg_models.py) - Clean router structure with one file per domain - Simplified configuration with sensible defaults - SQLite by default for development, PostgreSQL for production Testing: - All imports verified working - App loads successfully - Ready for development and testing
This commit is contained in:
+152
-31
@@ -2,28 +2,91 @@
|
||||
Security utilities for authentication and password hashing.
|
||||
|
||||
Implements JWT token management and bcrypt password hashing with salt.
|
||||
Provides functions for creating, validating, and decoding JWT tokens,
|
||||
as well as password verification and hashing.
|
||||
|
||||
## Features
|
||||
|
||||
- Bcrypt password hashing with configurable rounds
|
||||
- JWT access and refresh token creation
|
||||
- Token decoding and validation
|
||||
- FastAPI dependency for extracting current user from Authorization header
|
||||
|
||||
## Usage
|
||||
|
||||
from app.core.security import (
|
||||
verify_password,
|
||||
hash_password,
|
||||
create_access_token,
|
||||
create_refresh_token,
|
||||
decode_token,
|
||||
get_current_user,
|
||||
)
|
||||
|
||||
# Hash and verify passwords
|
||||
hashed = hash_password("secure_password")
|
||||
is_valid = verify_password("secure_password", hashed)
|
||||
|
||||
# Create and decode tokens
|
||||
token = create_access_token(subject="user_id", privlevel="Admin")
|
||||
payload = decode_token(token)
|
||||
|
||||
# FastAPI dependency
|
||||
@app.get("/protected")
|
||||
async def protected_route(user: dict = Depends(get_current_user)):
|
||||
...
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import Header, HTTPException, status
|
||||
from jose import JWTError, jwt
|
||||
from passlib.context import CryptContext
|
||||
from app.core.settings import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
from app.core.config import get_config
|
||||
|
||||
config = get_config()
|
||||
|
||||
# Password hashing context with bcrypt
|
||||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||
|
||||
|
||||
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
||||
"""Verify a plain password against a bcrypt hash."""
|
||||
"""Verify a plain password against a bcrypt hash.
|
||||
|
||||
Args:
|
||||
plain_password: The plain text password to verify.
|
||||
hashed_password: The bcrypt hash to verify against.
|
||||
|
||||
Returns:
|
||||
True if the password matches the hash, False otherwise.
|
||||
|
||||
Example:
|
||||
>>> hashed = hash_password("my_password")
|
||||
>>> verify_password("my_password", hashed)
|
||||
True
|
||||
>>> verify_password("wrong_password", hashed)
|
||||
False
|
||||
"""
|
||||
return pwd_context.verify(plain_password, hashed_password)
|
||||
|
||||
|
||||
def hash_password(password: str) -> str:
|
||||
"""Hash a password using bcrypt with configurable rounds."""
|
||||
return pwd_context.hash(password, rounds=settings.BCRYPT_ROUNDS)
|
||||
"""Hash a password using bcrypt with configurable rounds.
|
||||
|
||||
Args:
|
||||
password: The plain text password to hash.
|
||||
|
||||
Returns:
|
||||
The bcrypt hash string.
|
||||
|
||||
Example:
|
||||
>>> hashed = hash_password("secure_password")
|
||||
>>> len(hashed) > 0
|
||||
True
|
||||
"""
|
||||
return pwd_context.hash(password, rounds=config.BCRYPT_ROUNDS)
|
||||
|
||||
|
||||
def create_access_token(
|
||||
@@ -31,17 +94,31 @@ def create_access_token(
|
||||
privlevel: str = "User",
|
||||
expires_delta: Optional[timedelta] = None,
|
||||
) -> str:
|
||||
"""Create a JWT access token."""
|
||||
"""Create a JWT access token.
|
||||
|
||||
Args:
|
||||
subject: The user identifier (typically user ID or username).
|
||||
privlevel: User privilege level (default "User").
|
||||
expires_delta: Optional custom expiration time delta.
|
||||
If not provided, uses JWT_ACCESS_TOKEN_EXPIRE_MINUTES from config.
|
||||
|
||||
Returns:
|
||||
Encoded JWT access token string.
|
||||
|
||||
Example:
|
||||
>>> token = create_access_token(subject="user123", privlevel="Admin")
|
||||
>>> isinstance(token, str)
|
||||
True
|
||||
"""
|
||||
if expires_delta:
|
||||
expire = datetime.now(timezone.utc) + expires_delta
|
||||
else:
|
||||
expire = datetime.now(timezone.utc) + timedelta(
|
||||
minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES
|
||||
minutes=config.JWT_ACCESS_TOKEN_EXPIRE_MINUTES
|
||||
)
|
||||
|
||||
# Use JWT_SECRET_KEY if available, fall back to SECRET_KEY
|
||||
secret = getattr(settings, 'JWT_SECRET_KEY', None) or settings.SECRET_KEY
|
||||
|
||||
|
||||
secret = config.jwt_secret
|
||||
|
||||
payload = {
|
||||
"sub": subject,
|
||||
"exp": expire,
|
||||
@@ -49,18 +126,30 @@ def create_access_token(
|
||||
"type": "access",
|
||||
"privlevel": privlevel,
|
||||
}
|
||||
return jwt.encode(payload, secret, algorithm=settings.JWT_ALGORITHM)
|
||||
return jwt.encode(payload, secret, algorithm=config.JWT_ALGORITHM)
|
||||
|
||||
|
||||
def create_refresh_token(subject: str, privlevel: str = "User") -> str:
|
||||
"""Create a JWT refresh token with longer expiry."""
|
||||
"""Create a JWT refresh token with longer expiry.
|
||||
|
||||
Args:
|
||||
subject: The user identifier (typically user ID or username).
|
||||
privlevel: User privilege level (default "User").
|
||||
|
||||
Returns:
|
||||
Encoded JWT refresh token string.
|
||||
|
||||
Example:
|
||||
>>> token = create_refresh_token(subject="user123")
|
||||
>>> isinstance(token, str)
|
||||
True
|
||||
"""
|
||||
expire = datetime.now(timezone.utc) + timedelta(
|
||||
days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS
|
||||
days=config.JWT_REFRESH_TOKEN_EXPIRE_DAYS
|
||||
)
|
||||
|
||||
# Use JWT_SECRET_KEY if available, fall back to SECRET_KEY
|
||||
secret = getattr(settings, 'JWT_SECRET_KEY', None) or settings.SECRET_KEY
|
||||
|
||||
|
||||
secret = config.jwt_secret
|
||||
|
||||
payload = {
|
||||
"sub": subject,
|
||||
"exp": expire,
|
||||
@@ -68,19 +157,31 @@ def create_refresh_token(subject: str, privlevel: str = "User") -> str:
|
||||
"type": "refresh",
|
||||
"privlevel": privlevel,
|
||||
}
|
||||
return jwt.encode(payload, secret, algorithm=settings.JWT_ALGORITHM)
|
||||
return jwt.encode(payload, secret, algorithm=config.JWT_ALGORITHM)
|
||||
|
||||
|
||||
def decode_token(token: str) -> Optional[dict]:
|
||||
"""Decode and validate a JWT token."""
|
||||
"""Decode and validate a JWT token.
|
||||
|
||||
Args:
|
||||
token: The JWT token string to decode.
|
||||
|
||||
Returns:
|
||||
The token payload dictionary if valid, None if invalid or expired.
|
||||
|
||||
Example:
|
||||
>>> token = create_access_token(subject="user123")
|
||||
>>> payload = decode_token(token)
|
||||
>>> payload["sub"]
|
||||
'user123'
|
||||
"""
|
||||
try:
|
||||
# Use JWT_SECRET_KEY if available, fall back to SECRET_KEY
|
||||
secret = getattr(settings, 'JWT_SECRET_KEY', None) or settings.SECRET_KEY
|
||||
|
||||
secret = config.jwt_secret
|
||||
|
||||
payload = jwt.decode(
|
||||
token,
|
||||
secret,
|
||||
algorithms=[settings.JWT_ALGORITHM],
|
||||
algorithms=[config.JWT_ALGORITHM],
|
||||
)
|
||||
return payload
|
||||
except JWTError:
|
||||
@@ -88,28 +189,48 @@ def decode_token(token: str) -> Optional[dict]:
|
||||
|
||||
|
||||
def get_current_user(authorization: str = Header(...)) -> dict:
|
||||
"""Extract user info from JWT token in Authorization header."""
|
||||
"""Extract user info from JWT token in Authorization header.
|
||||
|
||||
FastAPI dependency that validates the Bearer token and extracts
|
||||
user information from the decoded payload.
|
||||
|
||||
Args:
|
||||
authorization: The Authorization header value (Bearer <token>).
|
||||
|
||||
Returns:
|
||||
Dictionary containing user_id, privlevel, and token_type.
|
||||
|
||||
Raises:
|
||||
HTTPException: 401 if token is invalid, missing, or wrong type.
|
||||
|
||||
Example:
|
||||
>>> @app.get("/protected")
|
||||
... async def protected(user: dict = Depends(get_current_user)):
|
||||
... return {"user_id": user["user_id"]}
|
||||
"""
|
||||
# Parse Bearer token
|
||||
parts = authorization.split()
|
||||
if len(parts) != 2 or parts[0].lower() != 'bearer':
|
||||
if len(parts) != 2 or parts[0].lower() != "bearer":
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid authentication scheme"
|
||||
detail="Invalid authentication scheme",
|
||||
)
|
||||
|
||||
|
||||
token = parts[1]
|
||||
payload = decode_token(token)
|
||||
|
||||
if not payload:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid token"
|
||||
detail="Invalid token",
|
||||
)
|
||||
|
||||
if payload.get("type") != "access":
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid token type"
|
||||
detail="Invalid token type",
|
||||
)
|
||||
|
||||
|
||||
return {
|
||||
"user_id": payload.get("sub"),
|
||||
"privlevel": payload.get("privlevel", "User"),
|
||||
|
||||
Reference in New Issue
Block a user