Coverage for src/lexigram/graphql/security/extensions.py: 50%

1"""GraphQL security schema extensions. 

2 

3Strawberry ``SchemaExtension`` implementations for security concerns. 

4This approach integrates cleanly with Strawberry's execution pipeline so 

5security checks run inside the schema lifecycle rather than in the executor. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING 

11 

12from strawberry.extensions import SchemaExtension 

13 

14from lexigram.logging import get_logger 

15 

16if TYPE_CHECKING: 

17 from collections.abc import AsyncGenerator 

18 

19 from lexigram.graphql.security.rate_limit import RateLimiter 

20 

21logger = get_logger(__name__) 

22 

23 

24class RateLimitExtension(SchemaExtension): 

25 """Enforce rate limiting on every GraphQL operation. 

26 

27 This extension delegates the actual limit check to the injected 

28 :class:`~lexigram.graphql.security.rate_limit.RateLimiter` (resolved 

29 from the DI container by :class:`~lexigram.graphql.providers.GraphQLProvider`) 

30 and raises :class:`~lexigram.graphql.exceptions.RateLimitError` when the 

31 limit is exceeded. 

32 

33 Register it on the schema builder instead of wiring it into the executor:: 

34 

35 builder.add_extension( 

36 RateLimitExtension(rate_limiter=rate_limiter, max_requests=60) 

37 ) 

38 

39 Args: 

40 rate_limiter: Rate limiter implementation; when ``None`` the extension 

41 is a no-op (useful for local/test environments). 

42 max_requests: Request allowance per window. 

43 window_seconds: Window length in seconds. 

44 """ 

45 

46 def __init__( 

47 self, 

48 *, 

49 rate_limiter: RateLimiter | None = None, 

50 max_requests: int = 60, 

51 window_seconds: int = 60, 

52 ) -> None: 

53 self._rate_limiter = rate_limiter 

54 self._max_requests = max_requests 

55 self._window_seconds = window_seconds 

56 

57 async def on_operation(self) -> AsyncGenerator[None, None]: 

58 """Check the rate limit before the GraphQL operation executes.""" 

59 if self._rate_limiter is not None: 

60 context = self.execution_context.context 

61 

62 from lexigram.graphql.exceptions import RateLimitError 

63 

64 try: 

65 allowed = await self._rate_limiter.is_allowed( 

66 context, 

67 max_requests=self._max_requests, 

68 window_seconds=self._window_seconds, 

69 ) 

70 if not allowed: 

71 raise RateLimitError("Rate limit exceeded") 

72 except RateLimitError: 

73 raise 

74 except (OSError, RuntimeError, LookupError) as e: 

75 # Infrastructure errors (Redis down etc.) must not block requests. 

76 logger.warning("rate_limit_check_error", error=str(e)) 

77 

78 yield 

79 

80 

81__all__ = ["RateLimitExtension"]