soleil-kamitto/skinscreen
0
1from datetime import datetime, timedelta, timezone2 3from fastapi import Depends, HTTPException, status4from fastapi.security import OAuth2PasswordBearer5from jose import JWTError, jwt6from passlib.context import CryptContext7from sqlalchemy.orm import Session8 9from app.config import settings10from app.models.db import User, get_db11 12pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")13oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")14 15 16def hash_password(password: str) -> str:17 return pwd_context.hash(password)18 19 20def verify_password(password: str, password_hash: str) -> bool:21 return pwd_context.verify(password, password_hash)22 23 24def create_access_token(subject: str) -> str:25 expire = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)26 payload = {"sub": subject, "exp": expire}27 return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)28 29 30def get_current_user(31 token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)32) -> User:33 credentials_error = HTTPException(34 status_code=status.HTTP_401_UNAUTHORIZED,35 detail="Credenciales inválidas o expiradas",36 headers={"WWW-Authenticate": "Bearer"},37 )38 try:39 payload = jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm])40 email = payload.get("sub")41 if email is None:42 raise credentials_error43 except JWTError:44 raise credentials_error45 46 user = db.query(User).filter(User.email == email).first()47 if user is None:48 raise credentials_error49 return user50 51 52def require_admin(user: User = Depends(get_current_user)) -> User:53 if not user.is_admin:54 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Requiere rol admin")55 return user56 