Coverage for src/backend/InvenTree/InvenTree/permissions.py: 70%

1"""Permission set for InvenTree.""" 

2 

3from functools import wraps 

4from typing import Optional 

5 

6from oauth2_provider.contrib.rest_framework import TokenMatchesOASRequirements 

7from oauth2_provider.contrib.rest_framework.authentication import OAuth2Authentication 

8from rest_framework import permissions 

9 

10import users.permissions 

11import users.ruleset 

12from users.oauth2_scopes import ( 

13 DEFAULT_READ, 

14 DEFAULT_STAFF, 

15 DEFAULT_SUPERUSER, 

16 _roles, 

17 get_granular_scope, 

18) 

19 

20ACTION_MAP = { 

21 'GET': 'view', 

22 'POST': 'add', 

23 'PUT': 'change', 

24 'PATCH': 'change', 

25 'DELETE': 'delete', 

26 'OPTIONS': DEFAULT_READ, 

27} 

28 

29 

30def get_model_for_view(view): 

31 """Attempt to introspect the 'model' type for an API view.""" 

32 if hasattr(view, 'get_permission_model'): 

33 return view.get_permission_model() 

34 

35 if hasattr(view, 'serializer_class'): 

36 return view.serializer_class.Meta.model 

37 

38 if hasattr(view, 'get_serializer_class'): 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true

39 return view.get_serializer_class().Meta.model 

40 

41 raise AttributeError(f'Serializer class not specified for {view.__class__}') 

42 

43 

44def map_scope( 

45 roles: Optional[list[str]] = None, 

46 only_read=False, 

47 read_name=DEFAULT_READ, 

48 map_read: Optional[list[str]] = None, 

49 map_read_name=DEFAULT_READ, 

50 override_all_actions: Optional[str] = None, 

51) -> dict: 

52 """Generate the required scopes for OAS permission views. 

53 

54 Args: 

55 roles (Optional[list[str]]): A list of roles or tables to generate granular scopes for. 

56 only_read (bool): If True, only the read scope will be returned for all actions. 

57 read_name (str): The read scope name to use when `only_read` is True. 

58 map_read (Optional[list[str]]): A list of HTTP methods that should map to the default read scope (use if some actions requirea differing role). 

59 map_read_name (str): The read scope name to use for methods specified in `map_read` when `map_read` is specified. 

60 override_all_actions (Optional[str]): If specified, all actions will be overridden to use the provided action name instead of the default action names. 

61 

62 Returns: 

63 dict: A dictionary mapping HTTP methods to their corresponding scopes. 

64 Each scope is represented as a list of lists of strings. 

65 """ 

66 

67 def scope_name(action): 

68 if only_read: 

69 return [[read_name]] 

70 if roles: 70 ↛ 72line 70 didn't jump to line 72 because the condition on line 70 was always true

71 return [[get_granular_scope(action, table) for table in roles]] 

72 return [[action]] 

73 

74 def get_scope(method, action): 

75 if map_read and method in map_read: 

76 return [[map_read_name]] 

77 return scope_name(override_all_actions if override_all_actions else action) 

78 

79 return { 

80 method: get_scope(method, action) if method != 'OPTIONS' else [[DEFAULT_READ]] 

81 for method, action in ACTION_MAP.items() 

82 } 

83 

84 

85# Precalculate the roles mapping 

86roles = users.ruleset.get_ruleset_models() 

87precalculated_roles = {} 

88for role, tables in roles.items(): 

89 for table in tables: 

90 if table not in precalculated_roles: 

91 precalculated_roles[table] = [] 

92 precalculated_roles[table].append(role) 

93 

94 

95class OASTokenMixin: 

96 """Mixin that combines the permissions of normal classes and token classes.""" 

97 

98 ENFORCE_USER_PERMS: bool = False 

99 

100 def has_permission(self, request, view): 

101 """Check if the user has the required scopes or was authenticated another way.""" 

102 if self.ENFORCE_USER_PERMS: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true

103 return super().has_permission(request, view) 

104 return self.check_oauth2_authentication( 

105 request, view 

106 ) or super().has_permission(request, view) 

107 

108 def check_oauth2_authentication(self, request, view): 

109 """Check if the user is authenticated using OAuth2 and has the required scopes.""" 

110 return self.is_oauth2ed( 

111 request 

112 ) and TokenMatchesOASRequirements().has_permission(request, view) 

113 

114 def is_oauth2ed(self, request): 

115 """Check if the user is authenticated using OAuth2.""" 

116 oauth2authenticated = False 

117 if bool(request.user and request.user.is_authenticated): 

118 oauth2authenticated = isinstance( 

119 request.successful_authenticator, OAuth2Authentication 

120 ) 

121 return oauth2authenticated 

122 

123 

124class InvenTreeRoleScopeMixin(OASTokenMixin): 

125 """Permission that discovers the required scopes from the OpenAPI schema.""" 

126 

127 def get_required_alternate_scopes(self, request, view): 

128 """Return the required scopes for the current request.""" 

129 if hasattr(view, 'required_alternate_scopes'): 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true

130 return view.required_alternate_scopes 

131 try: 

132 # Extract the model name associated with this request 

133 model = get_model_for_view(view) 

134 calc = precalculated_roles.get( 

135 f'{model._meta.app_label}_{model._meta.model_name}', [] 

136 ) 

137 

138 if model is None or not calc: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true

139 return map_scope(only_read=True) 

140 return map_scope(roles=calc) 

141 except AttributeError: 

142 # We will assume that if the serializer class does *not* have a Meta, 

143 # then we don't need a permission 

144 return map_scope(only_read=True) 

145 except Exception: 

146 return map_scope(only_read=True) 

147 

148 

149class InvenTreeTokenMatchesOASRequirements(InvenTreeRoleScopeMixin): 

150 """Combines InvenTree role-based scope handling with OpenAPI schema token requirements. 

151 

152 Used as default permission class. 

153 """ 

154 

155 def has_permission(self, request, view): 

156 """Check if the user has the required scopes or was authenticated another way.""" 

157 if self.is_oauth2ed(request): 157 ↛ 159line 157 didn't jump to line 159 because the condition on line 157 was never true

158 # Check if the user is authenticated using OAuth2 and has the required scopes 

159 return super().has_permission(request, view) 

160 

161 # If the user is authenticated using another method, check if they have the required permissions 

162 return bool(request.user and request.user.is_authenticated) 

163 

164 def has_object_permission(self, request, view, obj): 

165 """Return `True` if permission is granted, `False` otherwise.""" 

166 return True 

167 

168 

169class ModelPermission(permissions.DjangoModelPermissions): 

170 """Custom ModelPermission implementation which provides cached lookup of queryset. 

171 

172 This is entirely for optimization purposes. 

173 """ 

174 

175 def _queryset(self, view): 

176 """Return the queryset associated with this view, with caching. 

177 

178 This is because in a metadata OPTIONS request, the view is copied multiple times. 

179 We can cache the queryset to avoid repeated calculation. 

180 """ 

181 if getattr(view, '_cached_queryset', None) is not None: 

182 return view._cached_queryset 

183 

184 queryset = super()._queryset(view) 

185 

186 if queryset is not None: 186 ↛ 189line 186 didn't jump to line 189 because the condition on line 186 was always true

187 view._cached_queryset = queryset 

188 

189 return queryset 

190 

191 

192class RolePermission(InvenTreeRoleScopeMixin, permissions.BasePermission): 

193 """Role mixin for API endpoints, allowing us to specify the user "role" which is required for certain operations. 

194 

195 Each endpoint can have one or more of the following actions: 

196 - GET 

197 - POST 

198 - PUT 

199 - PATCH 

200 - DELETE 

201 

202 Specify the required "role" using the role_required attribute. 

203 

204 e.g. 

205 

206 role_required = "part" 

207 

208 The RoleMixin class will then determine if the user has the required permission 

209 to perform the specified action. 

210 

211 For example, a DELETE action will be rejected unless the user has the "part.remove" permission 

212 """ 

213 

214 def has_permission(self, request, view): 

215 """Determine if the current user has the specified permissions.""" 

216 user = request.user 

217 

218 # Superuser can do it all 

219 if user.is_superuser: 219 ↛ 223line 219 didn't jump to line 223 because the condition on line 219 was always true

220 return True 

221 

222 # Map the request method to a permission type 

223 rolemap = {**ACTION_MAP, 'OPTIONS': 'view'} 

224 

225 # let the view define a custom rolemap 

226 if hasattr(view, 'rolemap'): 

227 rolemap.update(view.rolemap) 

228 

229 permission = rolemap[request.method] 

230 

231 # The required role may be defined for the view class 

232 if role := getattr(view, 'role_required', None): 

233 # If the role is specified as "role.permission", split it 

234 if '.' in role: 

235 role, permission = role.split('.') 

236 

237 return users.permissions.check_user_role(user, role, permission) 

238 

239 try: 

240 # Extract the model name associated with this request 

241 model = get_model_for_view(view) 

242 

243 if model is None: 

244 return True 

245 

246 except AttributeError: 

247 # We will assume that if the serializer class does *not* have a Meta, 

248 # then we don't need a permission 

249 return True 

250 

251 return users.permissions.check_user_permission(user, model, permission) 

252 

253 

254class RolePermissionOrReadOnly(RolePermission): 

255 """RolePermission which also allows read access for any authenticated user.""" 

256 

257 REQUIRE_STAFF = False 

258 

259 def has_permission(self, request, view): 

260 """Determine if the current user has the specified permissions. 

261 

262 - If the user does have the required role, then allow the request 

263 - If the user does not have the required role, but is authenticated, then allow read-only access 

264 """ 

265 user = getattr(request, 'user', None) 

266 

267 if not user or not user.is_active or not user.is_authenticated: 

268 return False 

269 

270 if user.is_superuser: 270 ↛ 273line 270 didn't jump to line 273 because the condition on line 270 was always true

271 return True 

272 

273 if not self.REQUIRE_STAFF or user.is_staff: 

274 if super().has_permission(request, view): 

275 return True 

276 

277 return request.method in permissions.SAFE_METHODS 

278 

279 def get_required_alternate_scopes(self, request, view): 

280 """Return the required scopes for the current request.""" 

281 scopes = map_scope( 

282 only_read=True, 

283 read_name=DEFAULT_STAFF, 

284 map_read=list(permissions.SAFE_METHODS), 

285 ) 

286 return scopes 

287 

288 

289class StaffRolePermissionOrReadOnly(RolePermissionOrReadOnly): 

290 """RolePermission which requires staff AND role access, or read-only.""" 

291 

292 REQUIRE_STAFF = True 

293 

294 

295class IsSuperuserOrSuperScope(OASTokenMixin, permissions.IsAdminUser): 

296 """Allows access only to superuser users.""" 

297 

298 def has_permission(self, request, view): 

299 """Check if the user is a superuser.""" 

300 return bool(request.user and request.user.is_superuser) 

301 

302 def get_required_alternate_scopes(self, request, view): 

303 """Return the required scopes for the current request.""" 

304 return map_scope(only_read=True, read_name=DEFAULT_SUPERUSER) 

305 

306 

307class IsSuperuserOrReadOnlyOrScope(OASTokenMixin, permissions.IsAdminUser): 

308 """Allow read-only access to any user, but write access is restricted to superuser users.""" 

309 

310 def has_permission(self, request, view): 

311 """Check if the user is a superuser.""" 

312 return bool( 

313 (request.user and request.user.is_superuser) 

314 or request.method in permissions.SAFE_METHODS 

315 ) 

316 

317 def get_required_alternate_scopes(self, request, view): 

318 """Return the required scopes for the current request.""" 

319 return map_scope( 

320 only_read=True, 

321 read_name=DEFAULT_SUPERUSER, 

322 map_read=list(permissions.SAFE_METHODS), 

323 ) 

324 

325 

326class IsAuthenticatedOrReadScope(OASTokenMixin, permissions.IsAuthenticated): 

327 """Allows access only to authenticated users or read scope tokens.""" 

328 

329 def get_required_alternate_scopes(self, request, view): 

330 """Return the required scopes for the current request.""" 

331 return map_scope(only_read=True) 

332 

333 

334class IsStaffOrReadOnlyScope(OASTokenMixin, permissions.IsAuthenticated): 

335 """Allows read-only access to any authenticated user, but write access is restricted to staff users.""" 

336 

337 def has_permission(self, request, view): 

338 """Check if the user is a staff.""" 

339 return bool(permissions.IsAuthenticated().has_permission(request, view)) and ( 

340 (request.user and request.user.is_staff) 

341 or request.method in permissions.SAFE_METHODS 

342 ) 

343 

344 def get_required_alternate_scopes(self, request, view): 

345 """Return the required scopes for the current request.""" 

346 return map_scope( 

347 only_read=True, 

348 read_name=DEFAULT_STAFF, 

349 map_read=list(permissions.SAFE_METHODS), 

350 ) 

351 

352 

353class IsAdminOrAdminScope(OASTokenMixin, permissions.IsAdminUser): 

354 """Allows access only to admin users or admin scope tokens.""" 

355 

356 def get_required_alternate_scopes(self, request, view): 

357 """Return the required scopes for the current request.""" 

358 return map_scope(only_read=True, read_name=DEFAULT_STAFF) 

359 

360 

361class AllowAnyOrReadScope(OASTokenMixin, permissions.AllowAny): 

362 """Allows access to any user or read scope tokens.""" 

363 

364 def has_permission(self, request, view): 

365 """Anyone is allowed.""" 

366 return True 

367 

368 def get_required_alternate_scopes(self, request, view): 

369 """Return the required scopes for the current request.""" 

370 return map_scope(only_read=True) 

371 

372 

373def auth_exempt(view_func): 

374 """Mark a view function as being exempt from auth requirements.""" 

375 

376 def wrapped_view(*args, **kwargs): 

377 return view_func(*args, **kwargs) 

378 

379 wrapped_view.auth_exempt = True 

380 return wraps(view_func)(wrapped_view) 

381 

382 

383class UserSettingsPermissionsOrScope(OASTokenMixin, permissions.BasePermission): 

384 """Special permission class to determine if the user can view / edit a particular setting.""" 

385 

386 def has_object_permission(self, request, view, obj): 

387 """Check if the user that requested is also the object owner.""" 

388 try: 

389 user = request.user 

390 except AttributeError: # pragma: no cover 

391 return False 

392 

393 if not user.is_authenticated: 393 ↛ 394line 393 didn't jump to line 394 because the condition on line 393 was never true

394 return False 

395 

396 return user == obj.user 

397 

398 def has_permission(self, request, view): 

399 """Check that the requesting user is authenticated.""" 

400 try: 

401 user = request.user 

402 return user.is_authenticated 

403 except AttributeError: 

404 return False 

405 

406 def get_required_alternate_scopes(self, request, view): 

407 """Return the required scopes for the current request.""" 

408 return map_scope(only_read=True) 

409 

410 

411class GlobalSettingsPermissions(OASTokenMixin, permissions.BasePermission): 

412 """Special permission class to determine if the user is "staff".""" 

413 

414 def has_permission(self, request, view): 

415 """Check that the requesting user is 'admin'.""" 

416 try: 

417 user = request.user 

418 if request.method in permissions.SAFE_METHODS: 

419 return True 

420 # Any other methods require staff access permissions 

421 return user.is_staff 

422 

423 except AttributeError: # pragma: no cover 

424 return False 

425 

426 def get_required_alternate_scopes(self, request, view): 

427 """Return the required scopes for the current request.""" 

428 return map_scope( 

429 only_read=True, 

430 read_name=DEFAULT_STAFF, 

431 map_read=list(permissions.SAFE_METHODS), 

432 ) 

433 

434 

435class DataImporterPermission(OASTokenMixin, permissions.BasePermission): 

436 """Mixin class for determining if the user has correct permissions.""" 

437 

438 ENFORCE_USER_PERMS = True 

439 

440 def has_permission(self, request, view): 

441 """Class level permission checks are handled via InvenTree.permissions.IsAuthenticatedOrReadScope.""" 

442 return request.user and request.user.is_authenticated 

443 

444 def get_required_alternate_scopes(self, request, view): 

445 """Return the required scopes for the current request.""" 

446 return map_scope( 

447 roles=_roles, 

448 map_read=permissions.SAFE_METHODS, 

449 override_all_actions='change', # this is done to match the custom has_object_permission method 

450 ) 

451 

452 def has_object_permission(self, request, view, obj): 

453 """Check if the user has permission to access the imported object.""" 

454 import importer.models 

455 

456 # For safe methods (GET, HEAD, OPTIONS), allow access 

457 if request.method in permissions.SAFE_METHODS: 

458 return True 

459 

460 if isinstance(obj, importer.models.DataImportSession): 

461 session = obj 

462 else: 

463 session = getattr(obj, 'session', None) 

464 

465 if session: 

466 if model_class := session.model_class: 

467 return users.permissions.check_user_permission( 

468 request.user, model_class, 'change' 

469 ) 

470 

471 return True 

472 

473 

474class ContentTypePermission(OASTokenMixin, permissions.BasePermission): 

475 """Mixin class for determining if the user has correct permissions.""" 

476 

477 ENFORCE_USER_PERMS = True 

478 

479 def has_permission(self, request, view): 

480 """Class level permission checks are handled via InvenTree.permissions.IsAuthenticatedOrReadScope.""" 

481 return request.user and request.user.is_authenticated 

482 

483 def get_required_alternate_scopes(self, request, view): 

484 """Return the required scopes for the current request.""" 

485 return map_scope(roles=_roles) 

486 

487 def has_object_permission(self, request, view, obj): 

488 """Check if the user has permission to access the object.""" 

489 if model_class := obj.__class__: 489 ↛ 493line 489 didn't jump to line 493 because the condition on line 489 was always true

490 return users.permissions.check_user_permission( 

491 request.user, model_class, 'change' 

492 ) 

493 return False