| import os | |
| import time | |
| import logging | |
| from typing import Any | |
| import jwt | |
| logger = logging.getLogger("qkernel.auth") | |
| SECRET = os.getenv("JWT_SECRET", "qmetaram-dev-secret-change-in-prod") | |
| ALGORITHM = "HS256" | |
| TOKEN_TTL = 86400 | |
| def create_token(user_id: str, extra: dict | None = None) -> str: | |
| payload = { | |
| "user_id": user_id, | |
| "iat": int(time.time()), | |
| "exp": int(time.time()) + TOKEN_TTL, | |
| } | |
| if extra: | |
| payload.update(extra) | |
| return jwt.encode(payload, SECRET, algorithm=ALGORITHM) | |
| def verify_token(token: str) -> dict[str, Any] | None: | |
| try: | |
| payload = jwt.decode(token, SECRET, algorithms=[ALGORITHM]) | |
| return payload | |
| except jwt.ExpiredSignatureError: | |
| logger.warning("Expired token") | |
| return None | |
| except jwt.InvalidTokenError as e: | |
| logger.warning("Invalid token: %s", e) | |
| return None | |