127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
from jose import JWTError, jwt
|
|
from passlib.context import CryptContext
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from sqlalchemy.orm import Session
|
|
import os
|
|
import secrets
|
|
from database import get_db
|
|
from models import User, UserRole
|
|
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
security = HTTPBearer()
|
|
|
|
JWT_SECRET = os.environ.get('JWT_SECRET', 'your-secret-key')
|
|
JWT_ALGORITHM = os.environ.get('JWT_ALGORITHM', 'HS256')
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.environ.get('ACCESS_TOKEN_EXPIRE_MINUTES', 30))
|
|
|
|
def verify_password(plain_password, hashed_password):
|
|
return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
def get_password_hash(password):
|
|
return pwd_context.hash(password)
|
|
|
|
def generate_reset_token():
|
|
"""Generate secure random token for password reset"""
|
|
return secrets.token_urlsafe(32)
|
|
|
|
def create_password_reset_token(user, db):
|
|
"""Create reset token with 1-hour expiration"""
|
|
token = generate_reset_token()
|
|
expires = datetime.now(timezone.utc) + timedelta(hours=1)
|
|
|
|
user.password_reset_token = token
|
|
user.password_reset_expires = expires
|
|
db.commit()
|
|
|
|
return token
|
|
|
|
def verify_reset_token(token, db):
|
|
"""Verify token is valid and not expired"""
|
|
user = db.query(User).filter(User.password_reset_token == token).first()
|
|
|
|
if not user:
|
|
return None
|
|
|
|
if user.password_reset_expires < datetime.now(timezone.utc):
|
|
return None # Token expired
|
|
|
|
return user
|
|
|
|
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
|
|
to_encode = data.copy()
|
|
if expires_delta:
|
|
expire = datetime.now(timezone.utc) + expires_delta
|
|
else:
|
|
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
to_encode.update({"exp": expire})
|
|
encoded_jwt = jwt.encode(to_encode, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
return encoded_jwt
|
|
|
|
def decode_token(token: str):
|
|
try:
|
|
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
|
|
return payload
|
|
except JWTError:
|
|
return None
|
|
|
|
async def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials = Depends(security),
|
|
db: Session = Depends(get_db)
|
|
) -> User:
|
|
token = credentials.credentials
|
|
payload = decode_token(token)
|
|
|
|
if payload is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid authentication credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
user_id: str = payload.get("sub")
|
|
if user_id is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid authentication credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="User not found",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
return user
|
|
|
|
async def get_current_admin_user(current_user: User = Depends(get_current_user)) -> User:
|
|
if current_user.role != UserRole.admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Not enough permissions"
|
|
)
|
|
return current_user
|
|
|
|
async def get_active_member(current_user: User = Depends(get_current_user)) -> User:
|
|
"""Require user to be active member with valid payment"""
|
|
from models import UserStatus
|
|
|
|
if current_user.status != UserStatus.active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Active membership required. Please complete payment."
|
|
)
|
|
|
|
if current_user.role not in [UserRole.member, UserRole.admin]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Member access only"
|
|
)
|
|
|
|
return current_user
|