Spaces:
Sleeping
Sleeping
File size: 2,509 Bytes
3a19693 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | import os
import secrets
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from database import get_db
SECRET_KEY_FILE = os.path.join(os.path.dirname(__file__), ".secret_key")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_DAYS = 7
def _get_secret_key() -> str:
if os.path.exists(SECRET_KEY_FILE):
with open(SECRET_KEY_FILE) as f:
key = f.read().strip()
if key:
return key
key = secrets.token_hex(32)
with open(SECRET_KEY_FILE, "w") as f:
f.write(key)
return key
SECRET_KEY = _get_secret_key()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
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(user_id: int, username: str) -> str:
expire = datetime.utcnow() + timedelta(days=ACCESS_TOKEN_EXPIRE_DAYS)
payload = {"sub": str(user_id), "username": username, "exp": expire}
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
def decode_token(token: str) -> Optional[dict]:
try:
return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
except JWTError:
return None
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db),
):
# Import here to avoid circular imports
from models import User
credentials_exc = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired session. Please log in again.",
headers={"WWW-Authenticate": "Bearer"},
)
payload = decode_token(token)
if not payload:
raise credentials_exc
user_id = payload.get("sub")
if not user_id:
raise credentials_exc
user = db.query(User).filter(User.id == int(user_id)).first()
if not user or not user.is_active:
raise credentials_exc
return user
def get_current_admin(current_user=Depends(get_current_user)):
if not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Admin access required.",
)
return current_user
|