| from fastapi import HTTPException, Header, Depends |
| from fastapi.security import APIKeyHeader |
| from typing import Optional |
| from config import API_KEY, HUGGINGFACE_API_KEY, HUGGINGFACE |
| import os |
| import json |
| import base64 |
|
|
| |
| def validate_api_key(api_key_to_validate: str) -> bool: |
| """ |
| Validate the provided API key against the configured key. |
| """ |
| if not API_KEY: |
| |
| |
| return False |
| return api_key_to_validate == API_KEY |
|
|
| |
| api_key_header = APIKeyHeader(name="Authorization", auto_error=False) |
|
|
| |
| async def get_api_key( |
| authorization: Optional[str] = Header(None), |
| x_ip_token: Optional[str] = Header(None, alias="x-ip-token") |
| ): |
| |
| if HUGGINGFACE: |
| if x_ip_token is None: |
| raise HTTPException( |
| status_code=401, |
| detail="Missing x-ip-token header. This header is required for Hugging Face authentication." |
| ) |
|
|
| try: |
| |
| parts = x_ip_token.split('.') |
| if len(parts) < 2: |
| raise ValueError("Invalid JWT format: Not enough parts to extract payload.") |
| payload_encoded = parts[1] |
| |
| payload_encoded += '=' * (-len(payload_encoded) % 4) |
| decoded_payload_bytes = base64.urlsafe_b64decode(payload_encoded) |
| payload = json.loads(decoded_payload_bytes.decode('utf-8')) |
| except ValueError as ve: |
| |
| print(f"ValueError processing x-ip-token: {ve}") |
| raise HTTPException(status_code=400, detail=f"Invalid JWT format in x-ip-token: {str(ve)}") |
| except (json.JSONDecodeError, base64.binascii.Error, UnicodeDecodeError) as e: |
| print(f"Error decoding/parsing x-ip-token payload: {e}") |
| raise HTTPException(status_code=400, detail=f"Malformed x-ip-token payload: {str(e)}") |
| except Exception as e: |
| print(f"Unexpected error processing x-ip-token: {e}") |
| raise HTTPException(status_code=500, detail="Internal error processing x-ip-token.") |
|
|
| error_in_token = payload.get("error") |
|
|
| if error_in_token == "InvalidAccessToken": |
| raise HTTPException( |
| status_code=403, |
| detail="Access denied: x-ip-token indicates 'InvalidAccessToken'." |
| ) |
| elif error_in_token is None: |
| |
| print(f"HuggingFace authentication successful via x-ip-token (error field was null).") |
| return HUGGINGFACE_API_KEY |
| else: |
| |
| raise HTTPException( |
| status_code=403, |
| detail=f"Access denied: x-ip-token indicates an unhandled error: '{error_in_token}'." |
| ) |
| else: |
| |
| if authorization is None: |
| detail_message = "Missing API key. Please include 'Authorization: Bearer YOUR_API_KEY' header." |
| |
| if os.getenv("HUGGINGFACE") is not None: |
| detail_message += " (Note: HUGGINGFACE mode with x-ip-token is not currently active)." |
| raise HTTPException( |
| status_code=401, |
| detail=detail_message |
| ) |
| |
| |
| if not authorization.startswith("Bearer "): |
| raise HTTPException( |
| status_code=401, |
| detail="Invalid API key format. Use 'Authorization: Bearer YOUR_API_KEY'" |
| ) |
| |
| |
| api_key = authorization.replace("Bearer ", "") |
| |
| |
| if not validate_api_key(api_key): |
| raise HTTPException( |
| status_code=401, |
| detail="Invalid API key" |
| ) |
| |
| return api_key |