Spaces:
Sleeping
Sleeping
File size: 3,300 Bytes
3907a66 b29925a 740f610 fe02952 740f610 5c9fe8e 740f610 397123d 740f610 397123d 848bdfb 740f610 8776504 740f610 8776504 740f610 9e499eb 740f610 5c9fe8e 740f610 8776504 9e499eb 740f610 3907a66 8795c52 740f610 3907a66 8776504 740f610 5c9fe8e 8776504 740f610 5c9fe8e 8776504 848bdfb 3907a66 8776504 8795c52 740f610 8776504 740f610 3907a66 740f610 | 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 | # core/security.py
from datetime import datetime, timedelta
from passlib.context import CryptContext
from jose import jwt, JWTError
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from core.config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES
from db.mongo import users_collection
import logging
from fastapi import Request
logger = logging.getLogger(__name__)
# OAuth2 setup
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl="/auth/login",
scheme_name="JWT"
)
# Password hashing context
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
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(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode.update({"exp": expire, "iat": datetime.utcnow()})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
logger.debug(f"Created JWT for {data.get('sub')}, expires at {expire}")
return encoded_jwt
async def get_current_user(request: Request, token: str = Depends(oauth2_scheme)):
auth_header = request.headers.get("Authorization", "No Authorization header")
logger.debug(f"Raw Authorization header: {auth_header}")
logger.debug(f"Processed token: {token[:10]}... if present")
if not token:
logger.error(f"No token provided. Full headers: {dict(request.headers)}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="No token provided",
headers={"WWW-Authenticate": "Bearer"}
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
logger.debug(f"Token payload: {payload}")
email = payload.get("sub")
if not email:
logger.error("Invalid token: missing subject")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token: missing subject",
headers={"WWW-Authenticate": "Bearer"}
)
exp = payload.get("exp")
if exp and datetime.utcnow().timestamp() > exp:
logger.error(f"Token expired for {email}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token has expired",
headers={"WWW-Authenticate": "Bearer"}
)
except JWTError as e:
logger.error(f"JWT decode error: {str(e)}. Token: {token[:10]}...")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Could not validate token: {str(e)}",
headers={"WWW-Authenticate": "Bearer"}
)
user = await users_collection.find_one({"email": email})
if not user:
logger.error(f"User not found for email: {email}")
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
logger.info(f"Authenticated user: {user['email']}, role: {user.get('role')}")
return user |