Coverage for src / lexigram / admin / middleware / tenant.py: 0%

1"""Tenant resolution middleware for admin multi-tenancy support.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7from lexigram.admin.multitenancy.adapter import resolve_tenant_id 

8from lexigram.logging import get_logger 

9 

10if TYPE_CHECKING: 

11 from starlette.types import ASGIApp, Receive, Scope, Send 

12 

13 from lexigram.admin.config import TenancyConfig 

14 

15logger = get_logger(__name__) 

16 

17# Public paths that bypass tenant resolution 

18_TENANT_BYPASS_PATHS: frozenset[str] = frozenset( 

19 { 

20 "/login", 

21 "/setup", 

22 "/health", 

23 "/static", 

24 } 

25) 

26 

27 

28class AdminTenantMiddleware: 

29 """Tenant resolution middleware. 

30 

31 Resolves the current tenant ID from the request (header, cookie, or 

32 subdomain) and stores it in ``request.state.tenant_id``. When tenancy 

33 is enabled and no tenant can be resolved for a non-public path, a 

34 403 response is returned. 

35 

36 The middleware must be placed **before** ``AdminAuthGuardMiddleware`` 

37 in the stack so ``request.state.tenant_id`` is available to auth and 

38 data layers. 

39 """ 

40 

41 def __init__( 

42 self, 

43 app: ASGIApp, 

44 config: TenancyConfig, 

45 ) -> None: 

46 self.app = app 

47 self.config = config 

48 

49 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

50 if scope["type"] != "http": 

51 await self.app(scope, receive, send) 

52 return 

53 

54 from starlette.datastructures import MutableHeaders 

55 from starlette.requests import Request 

56 from starlette.responses import PlainTextResponse 

57 

58 request = Request(scope, receive=receive) 

59 

60 # Extract path relative to admin mount 

61 path = request.url.path 

62 

63 # Bypass for public paths 

64 if any(path.startswith(p) or path == p for p in _TENANT_BYPASS_PATHS): 

65 request.state.tenant_id = "" 

66 await self.app(scope, receive, send) 

67 return 

68 

69 # Resolve tenant ID (delegates to lexigram-tenancy when available) 

70 tenant_id = await resolve_tenant_id( 

71 request, 

72 default=self.config.default_tenant_id, 

73 header=self.config.header_name, 

74 cookie=self.config.cookie_name, 

75 ) 

76 request.state.tenant_id = tenant_id 

77 

78 if self.config.enabled and not tenant_id: 

79 logger.warning("tenant_resolution_failed", path=path) 

80 response = PlainTextResponse( 

81 "Tenant resolution failed", 

82 status_code=403, 

83 ) 

84 await response(scope, receive, send) 

85 return 

86 

87 # Inject tenant ID into response headers for downstream consumption 

88 if tenant_id: 

89 headers = MutableHeaders(scope=scope) 

90 headers.append("X-Tenant-Id", tenant_id) 

91 

92 await self.app(scope, receive, send)