sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

revamping documentation

Browse Source
This commit is contained in:
Samir Boulahtit
2025-11-17 22:59:42 +01:00
parent bbd64a6f21
commit 807033be16
107 changed files with 11973 additions and 28413 deletions
Download Patch File Download Diff File Expand all files Collapse all files

250
middleware/auth.py
View File

@@ -1,16 +1,24 @@
# middleware/auth.py
"""Summary description ....
"""Authentication and Authorization Module.
This module provides classes and functions for:
- ....
- ....
- ....
This module provides JWT-based authentication and role-based access control (RBAC)
for the application. It handles:
- Password hashing and verification using bcrypt
- JWT token creation and validation
- User authentication against the database
- Role-based access control (admin, vendor, customer)
- Current user extraction from request credentials
The module uses the following technologies:
- Jose for JWT token handling
- Passlib with bcrypt for secure password hashing
- SQLAlchemy for database operations
"""
import logging
import os
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional
from typing import Any, Callable, Dict, Optional
from fastapi import HTTPException
from fastapi.security import HTTPAuthorizationCredentials
@@ -34,132 +42,282 @@ pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class AuthManager:
"""JWT-based authentication manager with bcrypt password hashing."""
"""JWT-based authentication manager with bcrypt password hashing.
This class provides a complete authentication system including:
- User credential verification
- JWT token generation and validation
- Role-based access control
- Password hashing using bcrypt
Attributes:
secret_key (str): Secret key used for JWT encoding/decoding
algorithm (str): Algorithm used for JWT signing (HS256)
token_expire_minutes (int): Token expiration time in minutes
"""
def __init__(self):
"""Class constructor."""
"""Initialize the AuthManager with configuration from environment variables.
Reads the following environment variables:
- JWT_SECRET_KEY: Secret key for JWT signing (defaults to development key)
- JWT_EXPIRE_MINUTES: Token expiration time in minutes (defaults to 30)
"""
# Load JWT secret key from environment, with fallback for development
self.secret_key = os.getenv(
"JWT_SECRET_KEY", "your-secret-key-change-in-production-please"
)
# Use HS256 (HMAC with SHA-256) for token signing
self.algorithm = "HS256"
# Configure token expiration time from environment
self.token_expire_minutes = int(os.getenv("JWT_EXPIRE_MINUTES", "30"))
def hash_password(self, password: str) -> str:
"""Hash password using bcrypt."""
"""Hash a plain text password using bcrypt.
Uses bcrypt hashing algorithm with automatic salt generation.
The resulting hash is safe to store in the database.
Args:
password (str): Plain text password to hash
Returns:
str: Bcrypt hashed password string
"""
return pwd_context.hash(password)
def verify_password(self, plain_password: str, hashed_password: str) -> bool:
"""Verify password against hash."""
"""Verify a plain text password against a bcrypt hash.
Args:
plain_password (str): Plain text password to verify
hashed_password (str): Bcrypt hash to verify against
Returns:
bool: True if password matches hash, False otherwise
"""
return pwd_context.verify(plain_password, hashed_password)
def authenticate_user(self, db: Session, username: str, password: str) -> Optional[User]:
"""Authenticate user and return user object if credentials are valid."""
"""Authenticate user credentials against the database.
Supports authentication using either username or email address.
Verifies the provided password against the stored bcrypt hash.
Args:
db (Session): SQLAlchemy database session
username (str): Username or email address to authenticate
password (str): Plain text password to verify
Returns:
Optional[User]: User object if authentication succeeds, None otherwise
"""
# Query user by username or email (both are unique identifiers)
user = (
db.query(User)
.filter((User.username == username) | (User.email == username))
.first()
)
# User not found in database
if not user:
return None
# Password verification failed
if not self.verify_password(password, user.hashed_password):
return None
return user # Return user if credentials are valid
# Authentication successful, return user object
return user
def create_access_token(self, user: User) -> Dict[str, Any]:
"""Create JWT access token for user."""
"""Create a JWT access token for an authenticated user.
The token includes user identity and role information in the payload.
Token expiration is set based on the configured token_expire_minutes.
Args:
user (User): Authenticated user object
Returns:
Dict[str, Any]: Dictionary containing:
- access_token (str): The JWT token string
- token_type (str): Token type ("bearer")
- expires_in (int): Token lifetime in seconds
"""
# Calculate token expiration time
expires_delta = timedelta(minutes=self.token_expire_minutes)
expire = datetime.now(timezone.utc) + expires_delta
# Build JWT payload with user information
payload = {
"sub": str(user.id),
"username": user.username,
"email": user.email,
"role": user.role,
"exp": expire,
"iat": datetime.now(timezone.utc),
"sub": str(user.id), # Subject: user ID (JWT standard claim)
"username": user.username, # Username for display/logging
"email": user.email, # User email address
"role": user.role, # User role for authorization
"exp": expire, # Expiration time (JWT standard claim)
"iat": datetime.now(timezone.utc), # Issued at time (JWT standard claim)
}
# Encode the payload into a JWT token
token = jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
# Return token with metadata
return {
"access_token": token,
"token_type": "bearer",
"expires_in": self.token_expire_minutes * 60, # Return in seconds
"expires_in": self.token_expire_minutes * 60, # Convert minutes to seconds
}
def verify_token(self, token: str) -> Dict[str, Any]:
"""Verify JWT token and return user data."""
"""Verify and decode a JWT token, returning the user data.
Validates the token signature, expiration, and required claims.
Ensures the token contains all necessary user identification data.
Args:
token (str): JWT token string to verify
Returns:
Dict[str, Any]: Dictionary containing:
- user_id (int): User's database ID
- username (str): User's username
- email (str): User's email address
- role (str): User's role (defaults to "user" if not present)
Raises:
TokenExpiredException: If token has expired
InvalidTokenException: If token is malformed, missing required claims,
or signature verification fails
"""
try:
# Decode and verify the JWT token signature
payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
# Check if token has expired
# Validate token expiration claim exists
exp = payload.get("exp")
if exp is None:
raise InvalidTokenException("Token missing expiration")
# Check if token has expired (additional check beyond jwt.decode)
if datetime.now(timezone.utc) > datetime.fromtimestamp(exp, tz=timezone.utc):
raise TokenExpiredException()
# Extract user data
# Validate user identifier claim exists
user_id = payload.get("sub")
if user_id is None:
raise InvalidTokenException("Token missing user identifier")
# Extract and return user data from token payload
return {
"user_id": int(user_id),
"username": payload.get("username"),
"email": payload.get("email"),
"role": payload.get("role", "user"),
"role": payload.get("role", "user"), # Default to "user" role if not specified
}
except jwt.ExpiredSignatureError:
# Token has expired (caught by jwt.decode)
raise TokenExpiredException()
except jwt.JWTError as e:
# Token signature verification failed or token is malformed
logger.error(f"JWT decode error: {e}")
raise InvalidTokenException("Could not validate credentials")
except Exception as e:
# Catch any other unexpected errors during token verification
logger.error(f"Token verification error: {e}")
raise InvalidTokenException("Authentication failed")
def get_current_user(self, db: Session, credentials: HTTPAuthorizationCredentials) -> User:
"""Get current authenticated user from database."""
"""Extract and validate the current authenticated user from request credentials.
Verifies the JWT token from the Authorization header, looks up the user
in the database, and ensures the user account is active.
Args:
db (Session): SQLAlchemy database session
credentials (HTTPAuthorizationCredentials): Bearer token credentials from request
Returns:
User: The authenticated and active user object
Raises:
InvalidTokenException: If token verification fails
InvalidCredentialsException: If user is not found in database
UserNotActiveException: If user account is inactive
"""
# Verify JWT token and extract user data
user_data = self.verify_token(credentials.credentials)
# Look up user in database by ID from token
user = db.query(User).filter(User.id == user_data["user_id"]).first()
if not user:
raise InvalidCredentialsException("User not found")
# Ensure user account is active
if not user.is_active:
raise UserNotActiveException()
return user
def require_role(self, required_role: str):
"""Require specific role."""
def require_role(self, required_role: str) -> Callable:
"""Create a decorator that enforces a specific role requirement.
This method returns a decorator that can be used to protect functions
requiring a specific user role. The decorator validates that the current
user has the exact required role.
Args:
required_role (str): The role name required (e.g., "admin", "vendor")
Returns:
Callable: Decorator function that enforces role requirement
Raises:
HTTPException: If user doesn't have the required role (403 Forbidden)
Example:
@auth_manager.require_role("admin")
def admin_only_function(current_user: User):
# This will only execute if user has "admin" role
pass
"""
def decorator(func):
"""Decorator that wraps the function with role checking."""
def wrapper(current_user: User, *args, **kwargs):
# Check if current user has the required role
if current_user.role != required_role:
raise HTTPException(
status_code=403,
detail=f"Required role '{required_role}' not found. Current role: '{current_user.role}'",
)
# User has required role, proceed with function execution
return func(current_user, *args, **kwargs)
return wrapper
return decorator
def require_admin(self, current_user: User):
"""Require admin role."""
def require_admin(self, current_user: User) -> User:
"""Enforce that the current user has admin role.
Use this as a dependency in FastAPI routes to restrict access to admins only.
Args:
current_user (User): The authenticated user from get_current_user
Returns:
User: The user object if they have admin role
Raises:
AdminRequiredException: If user does not have admin role
"""
# Verify user has admin role
if current_user.role != "admin":
raise AdminRequiredException()
return current_user
def require_vendor(self, current_user: User):
def require_vendor(self, current_user: User) -> User:
"""
Require vendor role (vendor or admin).
@@ -174,6 +332,7 @@ class AuthManager:
Raises:
InsufficientPermissionsException: If user is not vendor or admin
"""
# Check if user has vendor or admin role (admins have full access)
if current_user.role not in ["vendor", "admin"]:
from app.exceptions import InsufficientPermissionsException
raise InsufficientPermissionsException(
@@ -182,7 +341,7 @@ class AuthManager:
)
return current_user
def require_customer(self, current_user: User):
def require_customer(self, current_user: User) -> User:
"""
Require customer role (customer or admin).
@@ -197,6 +356,7 @@ class AuthManager:
Raises:
InsufficientPermissionsException: If user is not customer or admin
"""
# Check if user has customer or admin role (admins have full access)
if current_user.role not in ["customer", "admin"]:
from app.exceptions import InsufficientPermissionsException
raise InsufficientPermissionsException(
@@ -205,12 +365,32 @@ class AuthManager:
)
return current_user
def create_default_admin_user(self, db: Session):
"""Create default admin user if it doesn't exist."""
def create_default_admin_user(self, db: Session) -> User:
"""Create a default admin user account if one doesn't already exist.
This is typically used during initial application setup or database seeding.
The default admin account should have its password changed immediately in production.
Default credentials:
- Username: admin
- Password: admin123 (CHANGE IN PRODUCTION!)
- Email: admin@example.com
Args:
db (Session): SQLAlchemy database session
Returns:
User: The admin user object (either existing or newly created)
"""
# Check if admin user already exists
admin_user = db.query(User).filter(User.username == "admin").first()
# Create admin user if it doesn't exist
if not admin_user:
# Hash the default password securely
hashed_password = self.hash_password("admin123")
# Create new admin user with default credentials
admin_user = User(
email="admin@example.com",
username="admin",
@@ -218,9 +398,13 @@ class AuthManager:
role="admin",
is_active=True,
)
# Save to database
db.add(admin_user)
db.commit()
db.refresh(admin_user)
# Log creation for audit trail (WARNING: contains default credentials)
logger.info(
"Default admin user created: username='admin', password='admin123'"
)