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 # Identity-bound resolution: when auth has already attached an 

71 # authenticated user with a tenant claim, it wins over client hints. 

72 user = getattr(request.state, "user", None) 

73 claim = getattr(user, "tenant_id", None) if user else None 

74 tenant_id = await resolve_tenant_id( 

75 request, 

76 default=self.config.default_tenant_id, 

77 header=self.config.header_name, 

78 cookie=self.config.cookie_name, 

79 claim=claim, 

80 ) 

81 request.state.tenant_id = tenant_id 

82 

83 if self.config.enabled and not tenant_id: 

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

85 response = PlainTextResponse( 

86 "Tenant resolution failed", 

87 status_code=403, 

88 ) 

89 await response(scope, receive, send) 

90 return 

91 

92 # Inject tenant ID into response headers for downstream consumption 

93 if tenant_id: 

94 headers = MutableHeaders(scope=scope) 

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

96 

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