53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
import app.auth.jwt as jwt
|
|
from datetime import datetime, timedelta, timezone
|
|
import os
|
|
from fastapi import HTTPException, status
|
|
|
|
JWT_SECRET = os.getenv("JWT_SECRET")
|
|
JWT_ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256")
|
|
|
|
def now():
|
|
return datetime.now(timezone.utc)
|
|
|
|
def create_access_token(data: dict):
|
|
payload = {
|
|
**data,
|
|
"type": "access",
|
|
"exp": now() + timedelta(minutes=60),
|
|
"iat": now(),
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
|
|
def create_refresh_token(data: dict):
|
|
payload = {
|
|
**data,
|
|
"type": "refresh",
|
|
"exp": now() + timedelta(days=7),
|
|
"iat": now(),
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
|
|
def decode_token(token: str):
|
|
try:
|
|
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
|
|
return payload
|
|
except jwt.ExpiredSignatureError:
|
|
raise HTTPException(status_code=401, detail="Token expired")
|
|
except jwt.InvalidTokenError:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
|
|
def verify_access_token(token: str):
|
|
payload = decode_token(token)
|
|
|
|
if payload.get("type") != "access":
|
|
raise HTTPException(status_code=401, detail="Invalid access token")
|
|
|
|
return payload
|
|
|
|
def verify_refresh_token(token: str):
|
|
payload = decode_token(token)
|
|
|
|
if payload.get("type") != "refresh":
|
|
raise HTTPException(status_code=401, detail="Invalid refresh token")
|
|
|
|
return payload |