from datetime import datetime, timedelta, timezone from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from jose import JWTError, jwt from passlib.context import CryptContext from sqlalchemy.orm import Session from app.config import settings from app.database import get_db from app.models import AdminRole, AdminUser pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") security = HTTPBearer(auto_error=False) ALGORITHM = "HS256" def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) def create_access_token(subject: str) -> str: expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes) return jwt.encode({"sub": subject, "exp": expire}, settings.secret_key, algorithm=ALGORITHM) def get_current_admin( credentials: HTTPAuthorizationCredentials | None = Depends(security), db: Session = Depends(get_db), ) -> AdminUser: if credentials is None or credentials.scheme.lower() != "bearer": raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated") try: payload = jwt.decode(credentials.credentials, settings.secret_key, algorithms=[ALGORITHM]) email: str | None = payload.get("sub") except JWTError: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") if not email: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") user = db.query(AdminUser).filter(AdminUser.email == email).first() if not user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found") return user def require_super_admin( admin: AdminUser = Depends(get_current_admin), ) -> AdminUser: if admin.role != AdminRole.super_admin: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Super admin required") return admin