Coverage for src/lumen/auth/enforcement.py: 93%

1""" 

2This file implements the scope enforcement checks as a FastAPI dependency. 

3It compares request credentials against required permissions to block unauthorized access. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Annotated, Any 

9from uuid import UUID 

10 

11from fastapi import Depends, Request, status 

12 

13from ..api.deps import current_tenant 

14from ..config import Settings, get_settings 

15from ..db.pool import get_pool 

16from ..db.repository import ResourceRepository 

17from ..fhir.compartment import compartment_params 

18from .scopes import GrantedScopes, parse_scopes 

19from .token import AuthError 

20 

21# ── Interaction resolver ─────────────────────────────────────────────────────── 

22 

23_METHOD_TO_INTERACTION: dict[str, str] = { 

24 "GET": "read", 

25 "POST": "create", 

26 "PUT": "update", 

27 "DELETE": "delete", 

28 "PATCH": "update", 

29} 

30 

31 

32def resolve_interaction(request: Request) -> str: 

33 """Derive the FHIR interaction from the HTTP method and path params.""" 

34 params = request.path_params 

35 method = request.method.upper() 

36 

37 # FHIR operations ($validate, $everything, …) — require wildcard for Stage 2. 

38 if "operation" in params: 

39 return "*" 

40 

41 # Search: GET /{type} or POST /{type}/_search 

42 if method == "GET" and "resource_id" not in params: 

43 return "search" 

44 if method == "POST" and request.url.path.endswith("/_search"): 

45 return "search" 

46 

47 return _METHOD_TO_INTERACTION.get(method, "read") 

48 

49 

50def resolve_resource_type(request: Request) -> str: 

51 if request.url.path.endswith("/tenant_settings"): 

52 return "Settings" 

53 return request.path_params.get("resource_type", "*") 

54 

55 

56# ── patient/ compartment enforcement ─────────────────────────────────────────── 

57 

58 

59async def _enforce_patient_compartment( 

60 request: Request, 

61 tenant_id: UUID, 

62 resource_type: str, 

63 patient_id: str, 

64) -> None: 

65 """Restrict a request that is *only* covered by a ``patient/`` scope to the 

66 launch patient's compartment. 

67 

68 Direct reads/writes by id (``resource_id`` present in the path) are checked 

69 against the compartment immediately, denying before the handler runs. 

70 Searches (no ``resource_id`` yet) can't be checked here — compartment 

71 membership can be an OR across several reference parameters, which the 

72 generic search query builder doesn't express as a single filter — so an 

73 implicit restriction is stashed on ``request.state.patient_compartment`` 

74 for the search handler (``api/resources.py::_run_search``) to apply as a 

75 post-query filter on the result set. 

76 """ 

77 resource_id = request.path_params.get("resource_id") 

78 

79 if resource_type == "Patient": 

80 if resource_id is not None and resource_id != patient_id: 

81 raise AuthError( 

82 status.HTTP_403_FORBIDDEN, 

83 "forbidden", 

84 "patient/ scope restricts access to the launch patient's own record.", 

85 ) 

86 request.state.patient_compartment = {"patient_id": patient_id, "self_only": True} 

87 return 

88 

89 params = compartment_params("Patient").get(resource_type) 

90 if not params: 

91 raise AuthError( 

92 status.HTTP_403_FORBIDDEN, 

93 "forbidden", 

94 f"'{resource_type}' is not part of the Patient compartment; " 

95 "patient/ scope does not grant access to it.", 

96 ) 

97 

98 if resource_id is None: 

99 request.state.patient_compartment = { 

100 "patient_id": patient_id, 

101 "self_only": False, 

102 "params": params, 

103 } 

104 return 

105 

106 pool = get_pool() 

107 async with pool.acquire() as conn: 

108 is_member = await ResourceRepository(conn).is_compartment_member( 

109 tenant_id, resource_type, resource_id, "Patient", patient_id, params 

110 ) 

111 if not is_member: 

112 raise AuthError( 

113 status.HTTP_403_FORBIDDEN, 

114 "forbidden", 

115 f"{resource_type}/{resource_id} is outside the launch patient's compartment.", 

116 ) 

117 

118 

119# ── Scope enforcer dependency ────────────────────────────────────────────────── 

120 

121 

122async def enforce_scope( 

123 request: Request, 

124 settings: Annotated[Settings, Depends(get_settings)], 

125 # Explicit dep on current_tenant ensures token validation and request.state.smart_claims 

126 # are set before this runs. FastAPI's dep cache means it only executes once per request. 

127 tenant: Annotated[UUID, Depends(current_tenant)], 

128) -> None: 

129 if not settings.smart.enabled: 

130 return 

131 

132 claims: dict[str, Any] = getattr(request.state, "smart_claims", {}) 

133 if not claims: 

134 raise AuthError( 

135 status.HTTP_401_UNAUTHORIZED, 

136 "login", 

137 "Request reached scope enforcement without a validated token.", 

138 ) 

139 

140 raw_scope = claims.get("scope") or claims.get("scp") or "" 

141 granted: GrantedScopes = parse_scopes(raw_scope) 

142 

143 resource_type = resolve_resource_type(request) 

144 interaction = resolve_interaction(request) 

145 

146 # A patient/ scope never covers FHIR operations ($everything, $validate, …) 

147 # in this iteration — per-operation compartment semantics aren't implemented 

148 # yet, so only system/ or user/ scopes can authorize them (safe default: 

149 # deny the complex case rather than silently over-grant). 

150 if interaction == "*": 

151 non_patient = GrantedScopes([s for s in granted.scopes if s.context != "patient"]) 

152 if not non_patient.permits("*", "read"): 

153 raise AuthError( 

154 status.HTTP_403_FORBIDDEN, 

155 "forbidden", 

156 granted.denial_diagnostic(resource_type, "execute"), 

157 ) 

158 return 

159 

160 # A system/ or user/ scope that already covers this request authorizes it 

161 # in full — no compartment restriction applies (matches §6 of 

162 # docs/architecture/smart-on-fhir.md: the restriction only "arises from the 

163 # token claim plus the core's data knowledge" when patient/ is the only 

164 # covering scope). 

165 non_patient = GrantedScopes([s for s in granted.scopes if s.context != "patient"]) 

166 if non_patient.permits(resource_type, interaction): 

167 return 

168 

169 patient_scopes = GrantedScopes([s for s in granted.scopes if s.context == "patient"]) 

170 if not patient_scopes.permits(resource_type, interaction): 

171 raise AuthError( 

172 status.HTTP_403_FORBIDDEN, 

173 "forbidden", 

174 granted.denial_diagnostic(resource_type, interaction), 

175 ) 

176 

177 patient_id = claims.get("patient") 

178 if not patient_id: 

179 raise AuthError( 

180 status.HTTP_403_FORBIDDEN, 

181 "forbidden", 

182 "Token has a patient/ scope but no 'patient' launch-context claim.", 

183 ) 

184 await _enforce_patient_compartment(request, tenant, resource_type, str(patient_id)) 

185 

186 

187ScopeEnforcerDep = Depends(enforce_scope)