Coverage for astrocyte/identity.py: 94%

1"""Identity resolution, OBO permission intersection, and bank resolution (M1 / v0.5.0 with M2). 

2 

3See ``docs/_design/adr/adr-002-identity-model.md``. 

4""" 

5 

6from __future__ import annotations 

7 

8from fnmatch import fnmatchcase 

9 

10from astrocyte.types import AccessGrant, ActorIdentity, AstrocyteContext 

11 

12 

13def format_principal(actor: ActorIdentity) -> str: 

14 """Serialize actor to the grant-matching principal form ``{type}:{id}``.""" 

15 return f"{actor.type}:{actor.id}" 

16 

17 

18def parse_principal(principal: str) -> ActorIdentity: 

19 """Parse ``principal`` string into :class:`ActorIdentity` (ADR-002). 

20 

21 - ``agent:X`` / ``user:X`` / ``service:X`` / ``team:X`` → typed id ``X`` 

22 - no prefix → treated as ``user:{principal}`` 

23 """ 

24 for prefix, typ in (("agent:", "agent"), ("user:", "user"), ("service:", "service"), ("team:", "team")): 

25 if principal.startswith(prefix): 

26 return ActorIdentity(type=typ, id=principal[len(prefix) :]) 

27 return ActorIdentity(type="user", id=principal) 

28 

29 

30def resolve_actor(context: AstrocyteContext) -> ActorIdentity | None: 

31 """Resolved actor: explicit ``context.actor``, else parse ``context.principal``.""" 

32 if context.actor is not None: 

33 return context.actor 

34 if context.principal: 

35 return parse_principal(context.principal) 

36 return None 

37 

38 

39def context_principal_label(context: AstrocyteContext) -> str: 

40 """Short label for logs and :class:`~astrocyte.errors.AccessDenied` messages.""" 

41 actor = resolve_actor(context) 

42 base = format_principal(actor) if actor else "anonymous" 

43 if context.on_behalf_of is not None: 

44 return f"{base} obo {format_principal(context.on_behalf_of)}" 

45 return base 

46 

47 

48_GLOB_CHARS = "*?[" 

49 

50 

51def _is_pattern(value: str) -> bool: 

52 return any(c in value for c in _GLOB_CHARS) 

53 

54 

55def _grant_field_matches(pattern: str, value: str) -> bool: 

56 """Match a grant's ``bank_id`` or ``principal`` against a concrete value. 

57 

58 ``*`` and exact strings behave as they always have. Any other value with a 

59 glob character is an fnmatch pattern (``project:*``, ``user:*``), matched 

60 case-sensitively on every platform. Bank ids cannot contain glob characters 

61 (``validate_bank_id``), so a pattern never collides with a real id. 

62 """ 

63 if pattern == "*" or pattern == value: 

64 return True 

65 return _is_pattern(pattern) and fnmatchcase(value, pattern) 

66 

67 

68def _permissions_for_principal_on_bank( 

69 grants: list[AccessGrant], 

70 bank_id: str, 

71 principal_str: str, 

72 groups: list[str] | None = None, 

73) -> set[str]: 

74 principals = [principal_str, *(groups or ())] 

75 out: set[str] = set() 

76 for g in grants: 

77 if not _grant_field_matches(g.bank_id, bank_id): 

78 continue 

79 if any(_grant_field_matches(g.principal, p) for p in principals): 

80 out.update(g.permissions) 

81 return out 

82 

83 

84def _with_context_grants(context: AstrocyteContext, grants: list[AccessGrant]) -> list[AccessGrant]: 

85 return [*grants, *context.grants] if context.grants else grants 

86 

87 

88def effective_permissions( 

89 context: AstrocyteContext, 

90 grants: list[AccessGrant], 

91 bank_id: str, 

92) -> set[str]: 

93 """Effective permission set for ``bank_id``, including OBO intersection. 

94 

95 Grants carried on the context (``context.grants``, e.g. bound to a gateway 

96 token) are added to ``grants``. ``context.groups`` are extra principals the 

97 actor holds, so a grant to ``team:api`` applies to its members; they do not 

98 extend the on-behalf-of identity. 

99 """ 

100 actor = resolve_actor(context) 

101 if actor is None: 

102 return set() 

103 all_grants = _with_context_grants(context, grants) 

104 actor_perms = _permissions_for_principal_on_bank( 

105 all_grants, bank_id, format_principal(actor), groups=context.groups 

106 ) 

107 if context.on_behalf_of is None: 

108 return actor_perms 

109 obo_perms = _permissions_for_principal_on_bank(all_grants, bank_id, format_principal(context.on_behalf_of)) 

110 return actor_perms & obo_perms 

111 

112 

113def accessible_read_banks( 

114 context: AstrocyteContext, 

115 grants: list[AccessGrant], 

116 *, 

117 known_bank_ids: list[str] | None = None, 

118 resolver: BankResolver | None = None, 

119) -> list[str]: 

120 """Bank IDs where ``context`` has effective ``read`` after OBO intersection. 

121 

122 Candidates are: non-pattern ``bank_id`` values from ``grants`` and 

123 ``context.grants``, optional ``known_bank_ids``, and (when no other 

124 candidates) a convention bank from ``resolver`` for the resolved actor so 

125 wildcard-only ACL rows can still resolve. 

126 """ 

127 candidates: set[str] = {g.bank_id for g in _with_context_grants(context, grants) if not _is_pattern(g.bank_id)} 

128 if known_bank_ids: 

129 candidates.update(known_bank_ids) 

130 

131 actor = resolve_actor(context) 

132 if not candidates and resolver is not None and actor is not None: 

133 suggested = resolver.default_bank_id(actor) 

134 if suggested is not None: 

135 candidates.add(suggested) 

136 

137 out: list[str] = [] 

138 for bid in sorted(candidates): 

139 if "read" in effective_permissions(context, grants, bid): 

140 out.append(bid) 

141 return out 

142 

143 

144class BankResolver: 

145 """Convention-based default bank id from a resolved :class:`ActorIdentity`.""" 

146 

147 def __init__( 

148 self, 

149 *, 

150 user_prefix: str = "user-", 

151 agent_prefix: str = "agent-", 

152 service_prefix: str = "service-", 

153 ) -> None: 

154 self._user_prefix = user_prefix 

155 self._agent_prefix = agent_prefix 

156 self._service_prefix = service_prefix 

157 

158 def default_bank_id(self, actor: ActorIdentity) -> str | None: 

159 if actor.type == "user": 

160 return f"{self._user_prefix}{actor.id}" 

161 if actor.type == "agent": 

162 return f"{self._agent_prefix}{actor.id}" 

163 if actor.type == "service": 

164 return f"{self._service_prefix}{actor.id}" 

165 return None