Spaces:
Running
Running
File size: 1,081 Bytes
6ccc851 |
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 |
import jwt
from aiohttp import web
SECRET_KEY = "your_secret_key" # Use a strong, randomly generated secret key
def create_jwt(user_id: str):
"""Create a JWT token."""
payload = {"sub": user_id, "role": "api_user"}
return jwt.encode(payload, SECRET_KEY, algorithm="HS256")
@web.middleware
async def jwt_middleware(request, handler):
"""JWT Middleware for /api endpoints."""
if request.path.startswith("/api"):
token = request.headers.get("Authorization")
if not token or not token.startswith("Bearer "):
raise web.HTTPUnauthorized(reason="Authorization header missing or malformed")
token = token.split("Bearer ")[1]
try:
decoded = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
request["user"] = decoded # Attach user info to the request
except jwt.ExpiredSignatureError:
raise web.HTTPUnauthorized(reason="Token has expired")
except jwt.InvalidTokenError:
raise web.HTTPUnauthorized(reason="Invalid token")
return await handler(request)
|