Spaces:
Sleeping
Sleeping
File size: 3,066 Bytes
52f5a2a | 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 | from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.config import settings
from app.models.models import Manajemen
from sqlalchemy import select
import logging
logger = logging.getLogger(__name__)
# OAuth2 Scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login/talent")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> dict:
"""
Validasi Token untuk Talent (Mobile App)
"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
user_id: int = payload.get("id")
role: str = payload.get("role")
if user_id is None or role != "talent":
logger.warning(f"Talent Auth Failed: UserID={user_id}, Role={role}")
raise credentials_exception
return {"idtalent": user_id, "role": role}
except JWTError as e:
logger.error(f"JWT Error (Talent): {str(e)}")
raise credentials_exception
async def get_current_admin_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> Manajemen:
"""
Validasi Token untuk Admin (Web Dashboard)
"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Admin credentials invalid",
headers={"WWW-Authenticate": "Bearer"},
)
try:
# 1. Decode Token
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
user_id: int = payload.get("id")
role: str = payload.get("role")
# 2. Validate Role
if user_id is None:
logger.warning("Admin Auth Failed: No User ID in token")
raise credentials_exception
# Allow both 'manajemen' and 'admin' roles for flexibility
if role not in ["manajemen", "admin"]:
logger.warning(f"Admin Auth Failed: Invalid Role '{role}' for user {user_id}")
raise credentials_exception
# 3. Check Database
query = select(Manajemen).where(Manajemen.idmanajemen == user_id)
result = await db.execute(query)
admin = result.scalar_one_or_none()
if admin is None:
logger.warning(f"Admin Auth Failed: User ID {user_id} not found in DB")
raise credentials_exception
return admin
except JWTError as e:
logger.error(f"JWT Error (Admin): {str(e)}")
raise credentials_exception
except Exception as e:
logger.error(f"Auth Unexpected Error: {str(e)}")
raise credentials_exception |