43 lines
1.6 KiB
Python
43 lines
1.6 KiB
Python
import os
|
|
import jwt
|
|
from datetime import datetime, timedelta, timezone
|
|
from fastapi import Request, HTTPException, Security, Depends
|
|
from fastapi.security.api_key import APIKeyHeader
|
|
from app.models import get_agent_by_key, update_key_last_used
|
|
|
|
JWT_SECRET = os.environ.get("JWT_SECRET", "secret")
|
|
JWT_EXPIRE_HOURS = int(os.environ.get("JWT_EXPIRE_HOURS", "12"))
|
|
|
|
API_KEY_NAME = "X-API-Key"
|
|
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
|
|
|
|
def create_jwt_token(data: dict) -> str:
|
|
to_encode = data.copy()
|
|
expire = datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRE_HOURS)
|
|
to_encode.update({"exp": expire})
|
|
encoded_jwt = jwt.encode(to_encode, JWT_SECRET, algorithm="HS256")
|
|
return encoded_jwt
|
|
|
|
def verify_jwt_token(request: Request) -> dict:
|
|
token = request.cookies.get("access_token")
|
|
if not token:
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
try:
|
|
payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
|
|
return payload
|
|
except jwt.ExpiredSignatureError:
|
|
raise HTTPException(status_code=401, detail="Token expired")
|
|
except jwt.InvalidTokenError:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
|
|
async def get_current_agent(api_key: str = Security(api_key_header)) -> str:
|
|
if not api_key:
|
|
raise HTTPException(status_code=401, detail="API Key header missing")
|
|
|
|
agent_name = await get_agent_by_key(api_key)
|
|
if not agent_name:
|
|
raise HTTPException(status_code=401, detail="Invalid API Key")
|
|
|
|
await update_key_last_used(api_key)
|
|
return agent_name
|