Ahmed-Raza/phase-3-backend
0
1from fastapi import HTTPException, status, Depends2from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials3import jwt4import os5from datetime import datetime, timedelta6 7# Get secret from environment8SECRET_KEY = os.getenv("JWT_SECRET", "fallback-secret-key-for-development")9ALGORITHM = "HS256"10 11security = HTTPBearer()12 13def verify_token(token: str) -> dict:14 """Verify JWT token and return payload"""15 try:16 payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])17 return payload18 except jwt.ExpiredSignatureError:19 raise HTTPException(20 status_code=status.HTTP_401_UNAUTHORIZED,21 detail="Token has expired",22 headers={"WWW-Authenticate": "Bearer"},23 )24 except jwt.JWTError:25 raise HTTPException(26 status_code=status.HTTP_401_UNAUTHORIZED,27 detail="Could not validate credentials",28 headers={"WWW-Authenticate": "Bearer"},29 )30 31async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> str:32 """Get current user ID from JWT token"""33 token = credentials.credentials34 payload = verify_token(token)35 user_id: str = payload.get("sub")36 37 if user_id is None:38 raise HTTPException(39 status_code=status.HTTP_401_UNAUTHORIZED,40 detail="Could not validate credentials",41 headers={"WWW-Authenticate": "Bearer"},42 )43 44 return user_id45 46def create_access_token(data: dict, expires_delta=None):47 """Create JWT access token"""48 to_encode = data.copy()49 if expires_delta:50 expire = datetime.utcnow() + expires_delta51 else:52 # Default to 7 days expiration53 expire = datetime.utcnow() + timedelta(days=7)54 55 to_encode.update({"exp": expire})56 encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)57 return encoded_jwt