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

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 self.authorization_service.set_roles(roles) # type: ignore[union-attr] 

35 

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

37 """Delegate to unified service.""" 

38 await self.authorization_service.sync_from_db(container) # type: ignore[union-attr] 

39 

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

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

42 self._schemas[resource_name] = schema 

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

44 

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

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

47 return self._schemas.get(resource_name) 

48 

49 # --- CRUD Operations --- 

50 

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

52 schema = self.get_schema(resource_name) 

53 return ( 

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

55 if schema 

56 else True 

57 ) 

58 

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

60 schema = self.get_schema(resource_name) 

61 return ( 

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

63 if schema 

64 else True 

65 ) 

66 

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

68 schema = self.get_schema(resource_name) 

69 return ( 

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

71 if schema 

72 else True 

73 ) 

74 

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

76 schema = self.get_schema(resource_name) 

77 if not schema: 

78 return True 

79 

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

81 return False 

82 

83 # Apply RLS policy if record is provided 

84 if schema.rls_policy and record: 

85 policy = get_policy(schema.rls_policy) 

86 if policy: 

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

88 return policy(context) 

89 

90 return True 

91 

92 def can_delete(self, user: Any, resource_name: str, record: Any = None) -> bool: 

93 schema = self.get_schema(resource_name) 

94 if not schema: 

95 return True 

96 

97 if not self._check_access(user, schema.can_delete, resource_name, "delete"): 

98 return False 

99 

100 if schema.rls_policy and record: 

101 policy = get_policy(schema.rls_policy) 

102 if policy: 

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

104 return policy(context) 

105 

106 return True 

107 

108 # --- Field Level --- 

109 

110 def can_view_field(self, user: Any, resource_name: str, field_name: str) -> bool: 

111 schema = self.get_schema(resource_name) 

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

113 return True 

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

115 

116 def can_edit_field(self, user: Any, resource_name: str, field_name: str) -> bool: 

117 schema = self.get_schema(resource_name) 

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

119 return True 

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

121 

122 def should_mask_field(self, user: Any, resource_name: str, field_name: str) -> bool: 

123 schema = self.get_schema(resource_name) 

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

125 return False 

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

127 

128 # --- Actions --- 

129 

130 def can_perform_action( 

131 self, 

132 user: Any, 

133 resource_name: str, 

134 action_name: str, 

135 ) -> bool: 

136 schema = self.get_schema(resource_name) 

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

138 return True 

139 return self._check_access( 

140 user, 

141 schema.actions[action_name].allowed_roles, 

142 resource_name, 

143 action_name, 

144 ) 

145 

146 # --- Internal Helpers --- 

147 

148 def _check_access( 

149 self, 

150 user: Any, 

151 allowed_roles: set[str], 

152 resource: str | None = None, 

153 action: str | None = None, 

154 ) -> bool: 

155 """Delegate to AuthorizationService.""" 

156 if self.authorization_service: 

157 return self.authorization_service.check_access( # type: ignore[return-value] 

158 user, 

159 allowed_roles, 

160 resource, 

161 action, 

162 ) 

163 # Fallback for legacy usage without a configured authorizer 

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

165 return False