Coverage for src/semware/api/auth.py: 64%

1"""API Key authentication for SemWare.""" 

2 

3from fastapi import HTTPException, Request, status 

4from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer 

5from loguru import logger 

6 

7from ..config import settings 

8 

9 

10class APIKeyBearer(HTTPBearer): 

11 """Custom HTTPBearer for API key authentication.""" 

12 

13 def __init__(self, auto_error: bool = False): 

14 super().__init__(auto_error=auto_error) 

15 

16 async def __call__(self, request: Request) -> str: 

17 """Authenticate API key from Authorization header or X-API-Key header. 

18 

19 Args: 

20 request: FastAPI request object 

21 

22 Returns: 

23 API key if valid 

24 

25 Raises: 

26 HTTPException: If authentication fails 

27 """ 

28 # Try Authorization header first 

29 credentials: HTTPAuthorizationCredentials = await super().__call__(request) 

30 if credentials and credentials.scheme == "Bearer": 

31 api_key = credentials.credentials 

32 else: 

33 # Try X-API-Key header 

34 api_key = request.headers.get("X-API-Key") 

35 

36 if not api_key: 

37 logger.warning("No API key provided in request") 

38 raise HTTPException( 

39 status_code=status.HTTP_401_UNAUTHORIZED, 

40 detail="API key required. Provide it via 'Authorization: Bearer <key>' or 'X-API-Key: <key>' header", 

41 headers={"WWW-Authenticate": "Bearer"}, 

42 ) 

43 

44 if not self.verify_api_key(api_key): 

45 logger.warning(f"Invalid API key provided: {api_key[:8]}...") 

46 raise HTTPException( 

47 status_code=status.HTTP_401_UNAUTHORIZED, 

48 detail="Invalid API key", 

49 headers={"WWW-Authenticate": "Bearer"}, 

50 ) 

51 

52 logger.debug("API key authentication successful") 

53 return api_key 

54 

55 def verify_api_key(self, api_key: str) -> bool: 

56 """Verify if the provided API key is valid. 

57 

58 Args: 

59 api_key: API key to verify 

60 

61 Returns: 

62 True if valid, False otherwise 

63 """ 

64 return api_key == settings.api_key 

65 

66 

67# Global API key authentication instance 

68api_key_auth = APIKeyBearer() 

69 

70 

71# Alternative header-based authentication for convenience 

72async def get_api_key_from_header(request: Request) -> str: 

73 """Get API key from X-API-Key header. 

74 

75 Args: 

76 request: FastAPI request object 

77 

78 Returns: 

79 API key if valid 

80 

81 Raises: 

82 HTTPException: If authentication fails 

83 """ 

84 api_key = request.headers.get("X-API-Key") 

85 

86 if not api_key: 

87 logger.warning("No X-API-Key header provided") 

88 raise HTTPException( 

89 status_code=status.HTTP_401_UNAUTHORIZED, 

90 detail="API key required in X-API-Key header", 

91 ) 

92 

93 if api_key != settings.api_key: 

94 logger.warning(f"Invalid API key in X-API-Key header: {api_key[:8]}...") 

95 raise HTTPException( 

96 status_code=status.HTTP_401_UNAUTHORIZED, 

97 detail="Invalid API key", 

98 ) 

99 

100 logger.debug("X-API-Key authentication successful") 

101 return api_key