Coverage for src/lexigram/admin/rbac/service.py: 0%

1"""Central permission service for enforcing RBAC rules.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.admin.rbac.policies import get_policy 

8from lexigram.admin.rbac.schema import ResourcePermissions 

9from lexigram.admin.rbac.types import PolicyContext 

10from lexigram.contracts.core.di import ContainerResolverProtocol 

11from lexigram.di.decorators import inject 

12from lexigram.logging import get_logger 

13 

14logger = get_logger(__name__) 

15 

16 

17from lexigram.contracts.auth import AuthorizerProtocol 

18 

19 

20@inject 

21class PermissionService: 

22 """Central service for all permission checks in Lexigram Admin. 

23 

24 Delegates core RBAC logic to an implementation of AuthorizerProtocol. 

25 Maintains ResourcePermissions registry for admin UI. 

26 """ 

27 

28 def __init__(self, authorization_service: AuthorizerProtocol | None = None): 

29 self.authorization_service = authorization_service 

30 self._schemas: dict[str, ResourcePermissions] = {} 

31 

32 def set_roles(self, roles: dict[str, Any]) -> None: 

33 """Delegate to unified service.""" 

34 if self.authorization_service is None: 

35 logger.warning( 

36 "PermissionService: No authorizer configured; skipping set_roles" 

37 ) 

38 return 

39 self.authorization_service.set_roles(roles) 

40 

41 async def sync_from_db(self, container: ContainerResolverProtocol) -> None: 

42 """Delegate to unified service.""" 

43 if self.authorization_service is None: 

44 logger.warning( 

45 "PermissionService: No authorizer configured; skipping sync_from_db" 

46 ) 

47 return 

48 await self.authorization_service.sync_from_db(container) 

49 

50 def register(self, resource_name: str, schema: ResourcePermissions) -> None: 

51 """Register permission schema for a resource.""" 

52 self._schemas[resource_name] = schema 

53 logger.debug("Registered permissions for resource: %s", resource_name) 

54 

55 def get_schema(self, resource_name: str) -> ResourcePermissions | None: 

56 """Get the permission schema for a resource.""" 

57 return self._schemas.get(resource_name) 

58 

59 # --- CRUD Operations --- 

60 

61 async def can_list(self, user: Any, resource_name: str) -> bool: 

62 schema = self.get_schema(resource_name) 

63 return ( 

64 await self._check_access(user, schema.can_list, resource_name, "list") 

65 if schema 

66 else True 

67 ) 

68 

69 async def can_view(self, user: Any, resource_name: str) -> bool: 

70 schema = self.get_schema(resource_name) 

71 return ( 

72 await self._check_access(user, schema.can_view, resource_name, "view") 

73 if schema 

74 else True 

75 ) 

76 

77 async def can_create(self, user: Any, resource_name: str) -> bool: 

78 schema = self.get_schema(resource_name) 

79 return ( 

80 await self._check_access(user, schema.can_create, resource_name, "create") 

81 if schema 

82 else True 

83 ) 

84 

85 async def can_edit(self, user: Any, resource_name: str, record: Any = None) -> bool: 

86 schema = self.get_schema(resource_name) 

87 if not schema: 

88 return True 

89 

90 if not await self._check_access(user, schema.can_edit, resource_name, "edit"): 

91 return False 

92 

93 # Apply RLS policy if record is provided 

94 if schema.rls_policy and record: 

95 policy = get_policy(schema.rls_policy) 

96 if policy: 

97 context = PolicyContext(user, resource_name, "edit", record) 

98 return policy(context) 

99 

100 return True 

101 

102 async def can_delete( 

103 self, user: Any, resource_name: str, record: Any = None 

104 ) -> bool: 

105 schema = self.get_schema(resource_name) 

106 if not schema: 

107 return True 

108 

109 if not await self._check_access( 

110 user, schema.can_delete, resource_name, "delete" 

111 ): 

112 return False 

113 

114 if schema.rls_policy and record: 

115 policy = get_policy(schema.rls_policy) 

116 if policy: 

117 context = PolicyContext(user, resource_name, "delete", record) 

118 return policy(context) 

119 

120 return True 

121 

122 # --- Field Level --- 

123 

124 async def can_view_field( 

125 self, user: Any, resource_name: str, field_name: str 

126 ) -> bool: 

127 schema = self.get_schema(resource_name) 

128 if not schema or field_name not in schema.fields: 

129 return True 

130 return await self._check_access(user, schema.fields[field_name].view_roles) 

131 

132 async def can_edit_field( 

133 self, user: Any, resource_name: str, field_name: str 

134 ) -> bool: 

135 schema = self.get_schema(resource_name) 

136 if not schema or field_name not in schema.fields: 

137 return True 

138 return await self._check_access(user, schema.fields[field_name].edit_roles) 

139 

140 async def should_mask_field( 

141 self, user: Any, resource_name: str, field_name: str 

142 ) -> bool: 

143 schema = self.get_schema(resource_name) 

144 if not schema or field_name not in schema.fields: 

145 return False 

146 return await self._check_access(user, schema.fields[field_name].mask_for) 

147 

148 # --- Actions --- 

149 

150 async def can_perform_action( 

151 self, 

152 user: Any, 

153 resource_name: str, 

154 action_name: str, 

155 ) -> bool: 

156 schema = self.get_schema(resource_name) 

157 if not schema or action_name not in schema.actions: 

158 return True 

159 return await self._check_access( 

160 user, 

161 schema.actions[action_name].allowed_roles, 

162 resource_name, 

163 action_name, 

164 ) 

165 

166 # --- Internal Helpers --- 

167 

168 async def _check_access( 

169 self, 

170 user: Any, 

171 allowed_roles: set[str], 

172 resource: str | None = None, 

173 action: str | None = None, 

174 ) -> bool: 

175 """Delegate to the unified authorizer.""" 

176 if self.authorization_service: 

177 return await self.authorization_service.check_access( 

178 user, 

179 allowed_roles, 

180 resource, 

181 action, 

182 ) 

183 # Fallback for legacy usage without a configured authorizer 

184 logger.warning("PermissionService: No authorizer configured; denying access") 

185 return False