Files
2026-10-03 09:44:11 +03:00

69 lines
2.0 KiB
Python

import bcrypt
import jwt
from jwt.exceptions import PyJWTError
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
import database
import models
SECRET = os.getenv("GYM_JWT_SECRET", "dev-secret-change-me")
ALGORITHM = "HS256"
EXPIRE_DAYS = int(os.getenv("GYM_JWT_EXPIRE_DAYS", "30"))
bearer_scheme = HTTPBearer(auto_error=False)
def hash_password(password: str) -> str:
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
def verify_password(password: str, hashed: str) -> bool:
try:
return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8"))
except ValueError:
return False
def create_access_token(user: models.User) -> str:
now = datetime.now(timezone.utc)
payload = {
"sub": str(user.id),
"username": user.username,
"iat": int(now.timestamp()),
"exp": int((now + timedelta(days=EXPIRE_DAYS)).timestamp()),
}
return jwt.encode(payload, SECRET, algorithm=ALGORITHM)
def get_current_user(
credentials: Optional[HTTPAuthorizationCredentials] = Depends(bearer_scheme),
db: Session = Depends(database.get_db),
) -> models.User:
if credentials is None or not credentials.credentials:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
)
try:
payload = jwt.decode(credentials.credentials, SECRET, algorithms=[ALGORITHM])
user_id = int(payload["sub"])
except (PyJWTError, KeyError, ValueError):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token",
)
user = db.query(models.User).filter(models.User.id == user_id).first()
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found",
)
return user