35 lines
1.3 KiB
Python
35 lines
1.3 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from jose import jwt, JWTError, ExpiredSignatureError
|
|
from app.core.config.Config import settings, load_private_key_bytes, load_public_key_bytes
|
|
from fastapi import HTTPException, status
|
|
|
|
def create_access_token(user_id: str, email: str, role: str = "") -> str:
|
|
now = datetime.now(timezone.utc)
|
|
expire = now + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MIN)
|
|
payload = {
|
|
"sub": user_id,
|
|
"email": email,
|
|
"role": role,
|
|
"iat": now,
|
|
"exp": expire
|
|
}
|
|
private_key = load_private_key_bytes()
|
|
return jwt.encode(payload, private_key, algorithm=settings.ALGORITHM)
|
|
|
|
def verify_access_token(token: str) -> dict:
|
|
try:
|
|
public_key = load_public_key_bytes()
|
|
payload = jwt.decode(token, public_key, algorithms=[settings.ALGORITHM])
|
|
return payload
|
|
except ExpiredSignatureError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Token has expired",
|
|
headers={"WWW-Authenticate": "Bearer"}
|
|
)
|
|
except JWTError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid signature or payload",
|
|
headers={"WWW-Authenticate": "Bearer"}
|
|
)
|