Coverage for postrfp / ref / service / auth_service.py: 100%

1""" 

2CEL-based authorization service for reference content management. 

3 

4This module provides simplified services for managing authorization policies, 

5replacing the complex permission management system. 

6""" 

7 

8from sqlalchemy.orm import Session 

9 

10from postrfp.shared.serial.refmodels import PermissionUpdated 

11from postrfp.model import Subject, Content 

12from postrfp.model.ref import ContentSpec 

13 

14 

15def update_content_policy( 

16 session: Session, content_id: int, policy_expression: str 

17) -> tuple[bool, PermissionUpdated]: 

18 """ 

19 Update the authorization policy for a content item. 

20 

21 Args: 

22 session: Database session 

23 content_id: Content ID 

24 policy_expression: CEL expression for authorization 

25 

26 Returns: 

27 tuple of (success flag, PermissionUpdated object) 

28 """ 

29 content = session.get_one(Content, content_id) 

30 content.auth_policy = policy_expression 

31 

32 # Create and return PermissionUpdated object 

33 return ( 

34 True, 

35 PermissionUpdated( 

36 operation_performed="updated", 

37 permission_name="policy_expression", 

38 entity_type="content", 

39 entity_name=content.title, 

40 target_org="system", # Since this is a system-level operation 

41 ), 

42 ) 

43 

44 

45def update_subject_policy( 

46 session: Session, subject_id: int, policy_expression: str 

47) -> tuple[bool, PermissionUpdated]: 

48 """ 

49 Update the authorization policy for a subject. 

50 

51 Args: 

52 session: Database session 

53 subject_id: Subject ID 

54 policy_expression: CEL expression for authorization 

55 

56 Returns: 

57 tuple of (success flag, PermissionUpdated object) 

58 """ 

59 subject = session.get_one(Subject, subject_id) 

60 subject.auth_policy = policy_expression 

61 

62 return ( 

63 True, 

64 PermissionUpdated( 

65 operation_performed="updated", 

66 permission_name="policy_expression", 

67 entity_type="subject", 

68 entity_name=subject.name, 

69 target_org="system", 

70 ), 

71 ) 

72 

73 

74def update_content_spec_policy( 

75 session: Session, content_spec_id: int, policy_expression: str 

76) -> tuple[bool, PermissionUpdated]: 

77 """ 

78 Update the authorization policy for a content spec. 

79 

80 Args: 

81 session: Database session 

82 content_spec_id: ContentSpec ID 

83 policy_expression: CEL expression for authorization 

84 

85 Returns: 

86 tuple of (success flag, PermissionUpdated object) 

87 """ 

88 content_spec = session.get_one(ContentSpec, content_spec_id) 

89 content_spec.auth_policy = policy_expression 

90 

91 return ( 

92 True, 

93 PermissionUpdated( 

94 operation_performed="updated", 

95 permission_name="policy_expression", 

96 entity_type="content_spec", 

97 entity_name=content_spec.name, 

98 target_org="system", 

99 ), 

100 ) 

101 

102 

103def get_policy_info() -> dict[str, list[str]]: 

104 """ 

105 Get information about available CEL policy features. 

106 

107 Returns: 

108 Dictionary with available policy features and examples 

109 """ 

110 return { 

111 "available_variables": [ 

112 "entity.author_org_id", 

113 "entity.visibility", 

114 "entity.managing_org_id", 

115 "user.org.id", 

116 "user.roles", 

117 "action", 

118 "now", 

119 ], 

120 "example_policies": [ 

121 "entity.author_org_id == user.org.id", 

122 "entity.visibility == 'public'", 

123 "entity.author_org_id == user.org.id || entity.visibility == 'public'", 

124 "'content_admin' in user.roles", 

125 "entity.managing_org_id == user.org.id || entity.subject_type == 'country'", 

126 ], 

127 "supported_actions": [ 

128 "view", 

129 "edit", 

130 "create", 

131 "delete", 

132 "comment", 

133 ], 

134 }