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

1"""Role management orchestrator for the RBAC admin UI. 

2 

3Persists roles through ``AdminRoleStoreProtocol`` and mirrors every 

4mutation into the in-memory ``AuthorizationService`` so runtime access 

5checks reflect edits immediately. System roles may have their 

6permissions changed but can never be renamed or deleted. 

7""" 

8 

9from __future__ import annotations 

10 

11from lexigram.admin.auth.protocols import AdminAuditLogServiceProtocol 

12from lexigram.admin.auth.types import AdminSecurityEventType 

13from lexigram.admin.rbac.errors import ( 

14 AdminRoleError, 

15 RoleDuplicateError, 

16 RoleNotFoundError, 

17 SystemRoleError, 

18) 

19from lexigram.admin.rbac.protocols import AdminRoleStoreProtocol 

20from lexigram.contracts.auth import AuthorizerProtocol, RoleDefinition 

21from lexigram.di.decorators import inject 

22from lexigram.logging import get_logger 

23from lexigram.result import Err, Ok, Result 

24 

25logger = get_logger(__name__) 

26 

27 

28@inject 

29class AdminRoleService: 

30 """Role CRUD with authorization-sync and audit (see module docstring). 

31 

32 Args: 

33 role_store: Role persistence (``admin_roles`` table). 

34 authorization_service: Optional authorizer to mirror role changes 

35 into; ``None`` skips the mirror (fail open). 

36 audit_service: Optional audit logger; ``None`` skips auditing. 

37 """ 

38 

39 def __init__( 

40 self, 

41 role_store: AdminRoleStoreProtocol, 

42 authorization_service: AuthorizerProtocol | None = None, 

43 audit_service: AdminAuditLogServiceProtocol | None = None, 

44 ) -> None: 

45 self._role_store = role_store 

46 self._authorization_service = authorization_service 

47 self._audit_service = audit_service 

48 

49 async def list_roles(self) -> list[RoleDefinition]: 

50 """Return all roles ordered by name (see protocol docs).""" 

51 return await self._role_store.list_roles() 

52 

53 async def create_role( 

54 self, 

55 name: str, 

56 description: str, 

57 permissions: list[str], 

58 inherits: list[str], 

59 ) -> Result[RoleDefinition, RoleDuplicateError | AdminRoleError]: 

60 """Create a role, mirror it, and audit (see protocol docs).""" 

61 name = name.strip() 

62 existing = await self._role_store.get_role(name) 

63 if existing is not None: 

64 return Err(RoleDuplicateError(f"Role '{name}' already exists.")) 

65 if not name: 

66 return Err(AdminRoleError("Role name is required.")) 

67 

68 role = RoleDefinition( 

69 name=name, 

70 description=description.strip(), 

71 permissions=sorted(set(permissions)), 

72 inherits=sorted(set(inherits)), 

73 is_system=False, 

74 ) 

75 await self._role_store.create_role(role) 

76 self._mirror(role) 

77 await self._audit(AdminSecurityEventType.ROLE_CREATED, {"role": name}) 

78 logger.info("admin.role_created", role=name) 

79 return Ok(role) 

80 

81 async def update_role( 

82 self, 

83 name: str, 

84 description: str, 

85 permissions: list[str], 

86 inherits: list[str], 

87 ) -> Result[RoleDefinition, RoleNotFoundError | SystemRoleError | AdminRoleError]: 

88 """Update a role; system roles keep their name (see protocol docs).""" 

89 name = name.strip() 

90 role = await self._role_store.get_role(name) 

91 if role is None: 

92 return Err(RoleNotFoundError(f"Role '{name}' does not exist.")) 

93 if role.is_system and name != role.name: 

94 return Err(SystemRoleError("System role names cannot be changed.")) 

95 

96 updated = RoleDefinition( 

97 name=role.name, 

98 description=description.strip(), 

99 permissions=sorted(set(permissions)), 

100 inherits=sorted(set(inherits)), 

101 is_system=role.is_system, 

102 ) 

103 await self._role_store.update_role(updated) 

104 self._mirror(updated) 

105 await self._audit(AdminSecurityEventType.ROLE_UPDATED, {"role": name}) 

106 logger.info("admin.role_updated", role=name) 

107 return Ok(updated) 

108 

109 async def delete_role( 

110 self, name: str 

111 ) -> Result[None, RoleNotFoundError | SystemRoleError | AdminRoleError]: 

112 """Delete a role; system roles are protected (see protocol docs).""" 

113 name = name.strip() 

114 role = await self._role_store.get_role(name) 

115 if role is None: 

116 return Err(RoleNotFoundError(f"Role '{name}' does not exist.")) 

117 if role.is_system: 

118 return Err(SystemRoleError("System roles cannot be deleted.")) 

119 

120 await self._role_store.delete_role(name) 

121 self._unmirror(name) 

122 await self._audit(AdminSecurityEventType.ROLE_DELETED, {"role": name}) 

123 logger.info("admin.role_deleted", role=name) 

124 return Ok(None) 

125 

126 # ------------------------------------------------------------------ 

127 # Internal helpers 

128 # ------------------------------------------------------------------ 

129 

130 def _mirror(self, role: RoleDefinition) -> None: 

131 """Push a role into the in-memory authorizer if available.""" 

132 if self._authorization_service is None: 

133 return 

134 self._authorization_service.register_role( 

135 role.name, 

136 { 

137 "description": role.description, 

138 "permissions": role.permissions, 

139 "inherits": role.inherits, 

140 }, 

141 ) 

142 

143 def _unmirror(self, name: str) -> None: 

144 """Remove a role from the in-memory authorizer if available.""" 

145 if self._authorization_service is None: 

146 return 

147 self._authorization_service.remove_role(name) 

148 

149 async def _audit(self, event_type: AdminSecurityEventType, metadata: dict) -> None: 

150 """Fire an audit event when an audit service is bound.""" 

151 if self._audit_service is None: 

152 return 

153 await self._audit_service.log_event( 

154 event_type=event_type, 

155 ip_address="", 

156 user_agent="", 

157 success=True, 

158 metadata=metadata, 

159 ) 

160 

161 

162__all__ = ["AdminRoleService"]