Coverage for src/lexigram/admin/controllers/tenancy.py: 0%

1"""Tenant-switching controller for the admin panel.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from starlette.requests import Request 

8from starlette.responses import RedirectResponse, Response 

9from starlette.routing import Route 

10 

11from lexigram.admin.auth.protocols import AdminAuditLogServiceProtocol 

12from lexigram.admin.auth.types import AdminSecurityEventType 

13from lexigram.admin.config import AdminConfig 

14from lexigram.admin.multitenancy.adapter import TenantProviderRegistry 

15from lexigram.admin.rbac.super_admin import is_super_admin 

16from lexigram.di.decorators import inject 

17from lexigram.logging import get_logger 

18 

19logger = get_logger(__name__) 

20 

21 

22@inject 

23class TenancyController: 

24 """Handles superadmin tenant-switching. 

25 

26 CSRF is validated by the global ``AdminCsrfMiddleware`` already applied 

27 to every admin POST route — this controller does not duplicate that 

28 check. 

29 """ 

30 

31 prefix = "" 

32 

33 def __init__( 

34 self, 

35 config: AdminConfig, 

36 registry: TenantProviderRegistry | None = None, 

37 audit_service: AdminAuditLogServiceProtocol | None = None, 

38 ) -> None: 

39 self._config = config 

40 self._registry = registry 

41 self._audit_service = audit_service 

42 

43 def get_routes(self) -> list[Any]: 

44 """Build routes explicitly so ``AdminRouter._build_routes`` mounts them. 

45 

46 Paths are relative to the admin mount (``config.prefix``), matching 

47 the other built-in controllers. 

48 """ 

49 return [ 

50 Route( 

51 "/set-tenant", 

52 endpoint=self.set_tenant, 

53 methods=["POST"], 

54 name="admin_set_tenant", 

55 ), 

56 ] 

57 

58 async def set_tenant(self, request: Request) -> Response: 

59 """Switch the active tenant for a superadmin session.""" 

60 if not self._config.tenancy.enabled or self._registry is None: 

61 return Response(status_code=404) 

62 

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

64 if not user or not is_super_admin(user, self._config.rbac.super_admin_role): 

65 return Response(content="Forbidden", status_code=403) 

66 

67 data = request.scope.get("admin_form_data") 

68 if data is None: 

69 data = await request.form() 

70 tenant_id = str(data.get("tenant_id", "")).strip() 

71 

72 tenant = await self._registry.get(tenant_id) if tenant_id else None 

73 if tenant is None: 

74 return Response(content="Unknown tenant", status_code=400) 

75 

76 previous_tenant_id = getattr(request.state, "tenant_id", None) or "default" 

77 

78 # Open-redirect guard: same-origin check on the Referer target. 

79 from lexigram.admin.controllers.auth.core import AuthCoreMixin 

80 

81 redirect_to = AuthCoreMixin._safe_next_url( 

82 request.headers.get("referer") or "/admin/" 

83 ) 

84 response = RedirectResponse(url=redirect_to, status_code=303) 

85 response.set_cookie( 

86 self._config.tenancy.cookie_name, 

87 tenant.tenant_id, 

88 httponly=True, 

89 samesite="lax", 

90 secure=self._config.environment in {"production", "staging"}, 

91 ) 

92 

93 await self._audit( 

94 request, 

95 from_tenant=previous_tenant_id, 

96 to_tenant=tenant.tenant_id, 

97 ) 

98 

99 return response 

100 

101 async def _audit(self, request: Request, **metadata: Any) -> None: 

102 """Log a TENANT_SWITCHED event, best-effort.""" 

103 if not self._audit_service: 

104 return 

105 try: 

106 client = getattr(request, "client", None) 

107 await self._audit_service.log_event( 

108 event_type=AdminSecurityEventType.TENANT_SWITCHED, 

109 ip_address=getattr(client, "host", "unknown"), 

110 user_agent=request.headers.get("user-agent", "") or "", 

111 success=True, 

112 metadata=metadata, 

113 ) 

114 except Exception: # noqa: BLE001 — audit failures must not break the switch 

115 logger.warning("tenancy.audit_failed", **metadata)