mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-06-11 16:54:44 +02:00
Lloyd
45a44eb47b
- Updated byte representations in tests to use lowercase hex format for consistency. - Reformatted code for better readability, including line breaks and indentation adjustments. - Consolidated multiple lines into single lines where appropriate to enhance clarity. - Ensured that all test cases maintain consistent formatting and style across the test suite.
36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
import logging
|
|
import time
|
|
from typing import Dict, Optional
|
|
|
|
import jwt
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class JWTHandler:
|
|
def __init__(self, secret: str, expiry_minutes: int = 15):
|
|
self.secret = secret
|
|
self.expiry_minutes = expiry_minutes
|
|
|
|
def create_jwt(self, username: str, client_id: str) -> str:
|
|
|
|
now = int(time.time())
|
|
expiry = now + (self.expiry_minutes * 60)
|
|
|
|
payload = {"sub": username, "exp": expiry, "iat": now, "client_id": client_id}
|
|
|
|
token = jwt.encode(payload, self.secret, algorithm="HS256")
|
|
logger.info(f"Created JWT for user '{username}' with client_id '{client_id[:8]}...'")
|
|
return token
|
|
|
|
def verify_jwt(self, token: str) -> Optional[Dict]:
|
|
try:
|
|
payload = jwt.decode(token, self.secret, algorithms=["HS256"])
|
|
return payload
|
|
except jwt.ExpiredSignatureError:
|
|
logger.warning("JWT token expired")
|
|
return None
|
|
except jwt.InvalidTokenError as e:
|
|
logger.warning(f"Invalid JWT token: {e}")
|
|
return None
|