Coverage for src/lumen/auth/scopes.py: 95%

1""" 

2This module parses and enforces SMART-on-FHIR scopes for requests. 

3It supports both version 1 and version 2 scope syntax to validate access rights. 

4""" 

5 

6from __future__ import annotations 

7 

8from dataclasses import dataclass, field 

9from typing import Literal 

10 

11# ── Interaction sets ─────────────────────────────────────────────────────────── 

12 

13ALL_INTERACTIONS: frozenset[str] = frozenset({"read", "create", "update", "delete", "search"}) 

14 

15_V2_LETTER: dict[str, str] = { 

16 "c": "create", 

17 "r": "read", 

18 "u": "update", 

19 "d": "delete", 

20 "s": "search", 

21} 

22 

23_V1_QUALIFIER: dict[str, frozenset[str]] = { 

24 "read": frozenset({"read", "search"}), 

25 "write": frozenset({"create", "update", "delete"}), 

26 "*": ALL_INTERACTIONS, 

27} 

28 

29ScopeContext = Literal["system", "user", "patient"] 

30 

31 

32# ── Data model ───────────────────────────────────────────────────────────────── 

33 

34 

35@dataclass(frozen=True) 

36class SmartScope: 

37 context: ScopeContext 

38 resource_type: str # "Patient" or "*" 

39 interactions: frozenset[str] 

40 

41 def covers(self, resource_type: str, interaction: str) -> bool: 

42 resource_match = self.resource_type == "*" or self.resource_type == resource_type 

43 return resource_match and interaction in self.interactions 

44 

45 

46@dataclass 

47class GrantedScopes: 

48 scopes: list[SmartScope] = field(default_factory=list[SmartScope]) 

49 

50 def permits(self, resource_type: str, interaction: str) -> bool: 

51 return any(s.covers(resource_type, interaction) for s in self.scopes) 

52 

53 def denial_diagnostic(self, resource_type: str, interaction: str) -> str: 

54 granted = ( 

55 ", ".join( 

56 f"{s.context}/{s.resource_type}.{''.join(sorted(i[0] for i in s.interactions))}" 

57 for s in self.scopes 

58 ) 

59 or "none" 

60 ) 

61 v2 = _interaction_to_v2(interaction) 

62 return ( 

63 f"Token does not permit '{interaction}' on '{resource_type}'. " 

64 f"Granted scopes: [{granted}]. " 

65 f"A scope matching 'system/{resource_type}.{v2}', " 

66 f"'system/{resource_type}.*', 'system/*.{v2}', or 'system/*.*' is required." 

67 ) 

68 

69 

70def _interaction_to_v2(interaction: str) -> str: 

71 return next((k for k, v in _V2_LETTER.items() if v == interaction), interaction[0]) 

72 

73 

74# ── Parser ───────────────────────────────────────────────────────────────────── 

75 

76 

77def parse_scopes(scope_claim: str | list[str]) -> GrantedScopes: 

78 """Parse the token's scope claim into a GrantedScopes object. 

79 

80 Accepts both: 

81 - a space-separated string (standard `scope` claim) 

82 - a list of strings (`scp` claim used by some IdPs) 

83 """ 

84 tokens = scope_claim if isinstance(scope_claim, list) else scope_claim.split() 

85 

86 parsed: list[SmartScope] = [] 

87 for token in tokens: 

88 scope = _parse_single(token) 

89 if scope is not None: 

90 parsed.append(scope) 

91 return GrantedScopes(scopes=parsed) 

92 

93 

94def _parse_single(token: str) -> SmartScope | None: 

95 """Parse one scope token. Returns None for non-SMART scopes (e.g. 'openid').""" 

96 # Expected form: <context>/<ResourceType>.<qualifier> 

97 if "/" not in token or "." not in token: 

98 return None 

99 

100 context_part, rest = token.split("/", 1) 

101 if context_part not in ("system", "user", "patient"): 

102 return None 

103 

104 if "." not in rest: 

105 return None 

106 

107 resource_type, qualifier = rest.rsplit(".", 1) 

108 if not resource_type: 

109 return None 

110 

111 interactions = _parse_qualifier(qualifier) 

112 if not interactions: 

113 return None 

114 

115 return SmartScope( 

116 context=context_part, # type: ignore[arg-type] 

117 resource_type=resource_type, 

118 interactions=interactions, 

119 ) 

120 

121 

122def _parse_qualifier(qualifier: str) -> frozenset[str]: 

123 # v1 keyword 

124 if qualifier in _V1_QUALIFIER: 

125 return _V1_QUALIFIER[qualifier] 

126 

127 # v2 letter string (e.g. "rs", "cruds", "c") 

128 interactions: set[str] = set() 

129 for char in qualifier: 

130 mapped = _V2_LETTER.get(char) 

131 if mapped is None: 

132 return frozenset() # unknown character → reject whole qualifier 

133 interactions.add(mapped) 

134 return frozenset(interactions) if interactions else frozenset()