Spaces:
Sleeping
Sleeping
File size: 4,571 Bytes
1ebb69b dc2e1e1 1ebb69b dc2e1e1 | 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 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | """
Nancy — Authentication & Authorization.
Provides FastAPI dependency functions for bearer token validation.
Two separate tokens are used:
- ``NANCY_API_KEY`` → for agent-facing ``/v1/*`` endpoints
- ``NANCY_EXT_SECRET`` → for extension-facing ``/ext/*`` endpoints
"""
from __future__ import annotations
import logging
import hashlib
import time
import asyncio
from typing import Annotated
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from config import settings
from core.redis_client import redis_client
logger = logging.getLogger("nancy.auth")
# Reusable security scheme — auto_error=False lets us return a nicer message
_bearer_scheme = HTTPBearer(auto_error=False)
def _extract_token(
request: Request,
credentials: HTTPAuthorizationCredentials | None,
) -> str:
"""
Extract the bearer token from the Authorization header.
Falls back to the ``authorization`` query parameter for SSE connections
where some clients cannot set custom headers.
Raises:
HTTPException(401): If no token is present.
"""
if credentials and credentials.credentials:
return credentials.credentials
# Fallback: query param (useful for EventSource which can't set headers)
query_token = request.query_params.get("authorization") or request.query_params.get("token")
if query_token:
# Strip "Bearer " prefix if present
if query_token.lower().startswith("bearer "):
return query_token[7:]
return query_token
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Missing authorization header. Provide 'Authorization: Bearer <token>'.",
headers={"WWW-Authenticate": "Bearer"},
)
async def require_api_key(
request: Request,
credentials: Annotated[
HTTPAuthorizationCredentials | None, Depends(_bearer_scheme)
] = None,
) -> str:
"""
FastAPI dependency: validates the bearer token against either NANCY_API_KEY
or a SHA-256 hashed token cached dynamically in Upstash Redis.
Returns the validated token string on success.
"""
token = _extract_token(request, credentials)
# 1. Master Key Local Bypass
if token == settings.nancy_api_key:
return token
# 2. Dynamic Redis Hashed Key validation
hashed = hashlib.sha256(token.encode("utf-8")).hexdigest()
try:
key_meta = await redis_client.get_json(f"nancy:api_keys:{hashed}")
if key_meta:
# Asynchronously update key metadata (fire-and-forget to keep requests fast)
key_meta["last_used"] = int(time.time())
key_meta["request_count"] = key_meta.get("request_count", 0) + 1
asyncio.create_task(redis_client.set_json(f"nancy:api_keys:{hashed}", key_meta))
return token
except Exception as exc:
logger.error("Error validating dynamic hashed token in Redis: %s", exc)
logger.warning("Invalid API key attempt from %s", request.client.host if request.client else "unknown")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API key.",
headers={"WWW-Authenticate": "Bearer"},
)
async def require_ext_secret(
request: Request,
credentials: Annotated[
HTTPAuthorizationCredentials | None, Depends(_bearer_scheme)
] = None,
) -> str:
"""
FastAPI dependency: validates the bearer token against ``NANCY_EXT_SECRET``,
``NANCY_API_KEY``, or a SHA-256 hashed token cached dynamically in Upstash Redis.
Used for all ``/ext/*`` endpoints.
"""
token = _extract_token(request, credentials)
# 1. Master Keys Local Bypass (either extension secret or API key)
if token in (settings.nancy_ext_secret, settings.nancy_api_key):
return token
# 2. Dynamic Redis Hashed Key validation
hashed = hashlib.sha256(token.encode("utf-8")).hexdigest()
try:
key_meta = await redis_client.get_json(f"nancy:api_keys:{hashed}")
if key_meta:
return token
except Exception as exc:
logger.error("Error validating dynamic token for extension: %s", exc)
logger.warning(
"Invalid extension secret/key attempt from %s",
request.client.host if request.client else "unknown",
)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid extension secret or API key.",
headers={"WWW-Authenticate": "Bearer"},
)
|