Coverage for src/backend/InvenTree/users/api.py: 86%

1"""DRF API definition for the 'users' app.""" 

2 

3import datetime 

4 

5from django.contrib.auth import get_user, login 

6from django.contrib.auth.models import Group, User 

7from django.contrib.auth.password_validation import password_changed, validate_password 

8from django.contrib.contenttypes.models import ContentType 

9from django.core.exceptions import ValidationError 

10from django.db.models import Q 

11from django.urls import include, path 

12from django.views.decorators.csrf import ensure_csrf_cookie 

13from django.views.generic.base import RedirectView 

14 

15import django_filters.rest_framework.filters as rest_filters 

16import structlog 

17from django_filters.rest_framework.filterset import FilterSet 

18from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema 

19from rest_framework import exceptions 

20from rest_framework.generics import DestroyAPIView, GenericAPIView 

21from rest_framework.response import Response 

22 

23import InvenTree.permissions 

24from InvenTree.fields import InvenTreeOutputOption, OutputConfiguration 

25from InvenTree.filters import SEARCH_ORDER_FILTER 

26from InvenTree.mixins import ( 

27 ListAPI, 

28 ListCreateAPI, 

29 OutputOptionsMixin, 

30 RetrieveAPI, 

31 RetrieveUpdateAPI, 

32 RetrieveUpdateDestroyAPI, 

33 SerializerContextMixin, 

34 UpdateAPI, 

35) 

36from InvenTree.schema import exclude_from_schema 

37from InvenTree.settings import FRONTEND_URL_BASE 

38from users.models import ApiToken, Owner, RuleSet, UserProfile 

39from users.serializers import ( 

40 ApiTokenSerializer, 

41 ExtendedUserSerializer, 

42 GetAuthTokenSerializer, 

43 GroupSerializer, 

44 MeUserSerializer, 

45 OwnerSerializer, 

46 RoleSerializer, 

47 RuleSetSerializer, 

48 UserCreateSerializer, 

49 UserProfileSerializer, 

50 UserSetPasswordSerializer, 

51) 

52 

53logger = structlog.get_logger('inventree') 

54 

55 

56class OwnerFilter(FilterSet): 

57 """filter set for OwnerList.""" 

58 

59 is_active = rest_filters.BooleanFilter(method='filter_is_active') 

60 

61 class Meta: 

62 """Meta class for owner filter.""" 

63 

64 model = Owner 

65 fields = ['is_active'] 

66 

67 def filter_is_active(self, queryset, name, value): 

68 """Filter by active status.""" 

69 if value is None: 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true

70 return queryset 

71 

72 # Get ContentType for User model 

73 user_content_type = ContentType.objects.get_for_model(User) 

74 

75 active_user_ids = list( 

76 User.objects.filter(is_active=value).values_list('pk', flat=True) 

77 ) 

78 

79 # Filter based on owner type 

80 q_filter = Q() 

81 

82 # If owner_type is not 'user', include all 

83 q_filter |= ~Q(owner_type=user_content_type) 

84 

85 # If owner_type is 'user', only include active/inactive users 

86 if active_user_ids: 

87 q_filter |= Q(owner_type=user_content_type, owner_id__in=active_user_ids) 

88 elif value is False: 88 ↛ 100line 88 didn't jump to line 100 because the condition on line 88 was always true

89 # If value is False and we want inactive users 

90 # Get all user IDs that are NOT in active_user_ids 

91 all_user_ids = list(User.objects.values_list('pk', flat=True)) 

92 inactive_user_ids = [ 

93 uid for uid in all_user_ids if uid not in active_user_ids 

94 ] 

95 if inactive_user_ids: 95 ↛ 100line 95 didn't jump to line 100 because the condition on line 95 was always true

96 q_filter |= Q( 

97 owner_type=user_content_type, owner_id__in=inactive_user_ids 

98 ) 

99 

100 return queryset.filter(q_filter) 

101 

102 

103class OwnerList(ListAPI): 

104 """List API endpoint for Owner model. 

105 

106 Cannot create a new Owner object via the API, but can view existing instances. 

107 """ 

108 

109 queryset = Owner.objects.all() 

110 serializer_class = OwnerSerializer 

111 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope] 

112 filterset_class = OwnerFilter 

113 filter_backends = SEARCH_ORDER_FILTER 

114 

115 def filter_queryset(self, queryset): 

116 """Implement text search for the "owner" model. 

117 

118 Note that an "owner" can be either a group, or a user, 

119 so we cannot do a direct text search. 

120 

121 A "hack" here is to post-process the queryset and simply 

122 remove any values which do not match. 

123 

124 It is not necessarily "efficient" to do it this way, 

125 but until we determine a better way, this is what we have... 

126 """ 

127 search_term = str(self.request.query_params.get('search', '')).lower() 

128 

129 queryset = queryset.select_related('owner_type').prefetch_related('owner') 

130 queryset = super().filter_queryset(queryset) 

131 

132 results = [] 

133 

134 for result in queryset.all(): 

135 name = str(result.name()).lower().strip() 

136 search_match = True 

137 

138 # Extract search term f 

139 if search_term: 

140 for entry in search_term.strip().split(' '): 140 ↛ 145line 140 didn't jump to line 145 because the loop on line 140 didn't complete

141 if entry not in name: 141 ↛ 140line 141 didn't jump to line 140 because the condition on line 141 was always true

142 search_match = False 

143 break 

144 

145 if not search_match: 

146 continue 

147 

148 # If we get here, there is no reason *not* to include this result 

149 results.append(result) 

150 

151 return results 

152 

153 

154class OwnerDetail(RetrieveAPI): 

155 """Detail API endpoint for Owner model. 

156 

157 Cannot edit or delete 

158 """ 

159 

160 queryset = Owner.objects.all() 

161 serializer_class = OwnerSerializer 

162 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope] 

163 

164 

165class RoleDetails(RetrieveAPI): 

166 """API endpoint which lists the available role permissions for the current user.""" 

167 

168 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope] 

169 serializer_class = RoleSerializer 

170 

171 def get_object(self): 

172 """Overwritten to always return current user.""" 

173 return self.request.user 

174 

175 

176class UserDetail(RetrieveUpdateDestroyAPI): 

177 """Detail endpoint for a single user. 

178 

179 Permissions: 

180 - Staff users (who also have the 'admin' role) can perform write operations 

181 - Otherwise authenticated users have read-only access 

182 """ 

183 

184 queryset = User.objects.all() 

185 serializer_class = ExtendedUserSerializer 

186 permission_classes = [InvenTree.permissions.StaffRolePermissionOrReadOnly] 

187 

188 def perform_destroy(self, instance): 

189 """Override destroy method to ensure sessions are deleted first.""" 

190 # Remove all sessions for this user 

191 if sessions := instance.usersession_set.all(): 

192 sessions.delete() 

193 # Normally delete the user 

194 return super().perform_destroy(instance) 

195 

196 

197class UserDetailSetPassword(UpdateAPI): 

198 """Allows superusers to set the password for a user.""" 

199 

200 queryset = User.objects.all() 

201 serializer_class = UserSetPasswordSerializer 

202 permission_classes = [InvenTree.permissions.IsSuperuserOrSuperScope] 

203 

204 def get_object(self): 

205 """Return the user object for this endpoint.""" 

206 return self.get_queryset().get(pk=self.kwargs['pk']) 

207 

208 def perform_update(self, serializer): 

209 """Set the password for the user.""" 

210 user: User = serializer.instance 

211 

212 password: str = serializer.validated_data.get('password', None) 

213 overwrite: bool = serializer.validated_data.get('override_warning', False) 

214 

215 if password: 

216 if not overwrite: 

217 try: 

218 validate_password(password=password, user=user) 

219 except ValidationError as e: 

220 raise exceptions.ValidationError({'password': str(e)}) 

221 

222 user.set_password(password) 

223 password_changed(password=password, user=user) 

224 user.save() 

225 

226 

227class MeUserDetail(RetrieveUpdateAPI, UserDetail): 

228 """Detail endpoint for current user. 

229 

230 Permissions: 

231 - User can edit their own details via this endpoint 

232 - Only a subset of fields are available here 

233 """ 

234 

235 serializer_class = MeUserSerializer 

236 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope] 

237 

238 rolemap = {'POST': 'view', 'PUT': 'view', 'PATCH': 'view'} 

239 

240 def get_object(self): 

241 """Always return the current user object.""" 

242 return self.request.user 

243 

244 def get_permission_model(self): 

245 """Return the model for the permission check. 

246 

247 Note that for this endpoint, the current user can *always* edit their own details. 

248 """ 

249 return None 

250 

251 

252class UserList(ListCreateAPI): 

253 """List endpoint for detail on all users. 

254 

255 Permissions: 

256 - Staff users (who also have the 'admin' role) can perform write operations 

257 - Otherwise authenticated users have read-only access 

258 """ 

259 

260 queryset = User.objects.all().prefetch_related('groups') 

261 serializer_class = UserCreateSerializer 

262 

263 # User must have the right role, AND be a staff user, else read-only 

264 permission_classes = [InvenTree.permissions.StaffRolePermissionOrReadOnly] 

265 

266 filter_backends = SEARCH_ORDER_FILTER 

267 

268 search_fields = ['first_name', 'last_name', 'username'] 

269 

270 ordering_fields = [ 

271 'email', 

272 'username', 

273 'first_name', 

274 'last_name', 

275 'is_staff', 

276 'is_superuser', 

277 'is_active', 

278 ] 

279 

280 filterset_fields = ['is_staff', 'is_active', 'is_superuser'] 

281 

282 

283class GroupMixin(SerializerContextMixin): 

284 """Mixin for Group API endpoints to add permissions filter. 

285 

286 Permissions: 

287 - Staff users (who also have the 'admin' role) can perform write operations 

288 - Otherwise authenticated users have read-only access 

289 """ 

290 

291 queryset = Group.objects.all() 

292 serializer_class = GroupSerializer 

293 permission_classes = [InvenTree.permissions.IsStaffOrReadOnlyScope] 

294 

295 

296class GroupOutputOptions(OutputConfiguration): 

297 """Holds all available output options for Group views.""" 

298 

299 OPTIONS = [ 

300 InvenTreeOutputOption('user_detail', description='Include user details'), 

301 InvenTreeOutputOption( 

302 'permission_detail', description='Include permission details' 

303 ), 

304 InvenTreeOutputOption( 

305 'role_detail', description='Include role details', default=True 

306 ), 

307 ] 

308 

309 

310class GroupDetail(GroupMixin, OutputOptionsMixin, RetrieveUpdateDestroyAPI): 

311 """Detail endpoint for a particular auth group.""" 

312 

313 output_options = GroupOutputOptions 

314 

315 

316class GroupList(GroupMixin, OutputOptionsMixin, ListCreateAPI): 

317 """List endpoint for all auth groups.""" 

318 

319 output_options = GroupOutputOptions 

320 filter_backends = SEARCH_ORDER_FILTER 

321 search_fields = ['name'] 

322 ordering_fields = ['name'] 

323 

324 

325class RuleSetMixin: 

326 """Mixin for RuleSet API endpoints.""" 

327 

328 queryset = RuleSet.objects.all() 

329 serializer_class = RuleSetSerializer 

330 permission_classes = [InvenTree.permissions.IsStaffOrReadOnlyScope] 

331 

332 

333class RuleSetList(RuleSetMixin, ListAPI): 

334 """List endpoint for all RuleSet instances.""" 

335 

336 filter_backends = SEARCH_ORDER_FILTER 

337 

338 search_fields = ['name'] 

339 ordering_fields = ['name'] 

340 filterset_fields = ['group', 'name'] 

341 

342 

343class RuleSetDetail(RuleSetMixin, RetrieveUpdateDestroyAPI): 

344 """Detail endpoint for a particular RuleSet instance.""" 

345 

346 

347class GetAuthToken(GenericAPIView): 

348 """Return authentication token for an authenticated user.""" 

349 

350 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope] 

351 serializer_class = GetAuthTokenSerializer 

352 

353 @extend_schema( 

354 parameters=[ 

355 OpenApiParameter( 

356 name='name', 

357 type=str, 

358 location=OpenApiParameter.QUERY, 

359 description='Name of the token', 

360 ) 

361 ], 

362 responses={200: OpenApiResponse(response=GetAuthTokenSerializer())}, 

363 ) 

364 def get(self, request, *args, **kwargs): 

365 """Return an API token if the user is authenticated. 

366 

367 - If the user already has a matching token, delete it and create a new one 

368 - Existing tokens are *never* exposed again via the API 

369 - Once the token is provided, it can be used for auth until it expires 

370 """ 

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

372 raise exceptions.NotAuthenticated() # pragma: no cover 

373 

374 user = request.user 

375 name = request.query_params.get('name', '') 

376 

377 name = ApiToken.sanitize_name(name) 

378 

379 today = datetime.date.today() 

380 

381 # Find existing token, which has not expired 

382 token = ApiToken.objects.filter( 

383 user=user, name=name, revoked=False, expiry__gte=today 

384 ).first() 

385 

386 if not token: 

387 # User is authenticated, and requesting a token against the provided name. 

388 token = ApiToken.objects.create(user=request.user, name=name) 

389 

390 logger.info( 

391 "Created new API token for user '%s' (name='%s')", user.username, name 

392 ) 

393 

394 # Add some metadata about the request 

395 token.set_metadata('user_agent', request.headers.get('user-agent', '')) 

396 token.set_metadata('remote_addr', request.META.get('REMOTE_ADDR', '')) 

397 token.set_metadata('remote_host', request.META.get('REMOTE_HOST', '')) 

398 token.set_metadata('remote_user', request.META.get('REMOTE_USER', '')) 

399 token.set_metadata('server_name', request.META.get('SERVER_NAME', '')) 

400 token.set_metadata('server_port', request.META.get('SERVER_PORT', '')) 

401 

402 data = {'token': token.key, 'name': token.name, 'expiry': token.expiry} 

403 

404 # Ensure that the users session is logged in 

405 if not get_user(request).is_authenticated: 405 ↛ 408line 405 didn't jump to line 408 because the condition on line 405 was always true

406 login(request, user, backend='django.contrib.auth.backends.ModelBackend') 

407 

408 return Response(data) 

409 

410 

411class TokenMixin: 

412 """Mixin for API token endpoints.""" 

413 

414 permission_classes = (InvenTree.permissions.IsAuthenticatedOrReadScope,) 

415 serializer_class = ApiTokenSerializer 

416 

417 def get_queryset(self): 

418 """Only return data for current user.""" 

419 if self.request.user.is_superuser and self.request.query_params.get( 

420 'all_users', False 

421 ): 

422 return ApiToken.objects.all() 

423 return ApiToken.objects.filter(user=self.request.user) 

424 

425 @extend_schema( 

426 parameters=[ 

427 OpenApiParameter( 

428 name='all_users', 

429 type=bool, 

430 location=OpenApiParameter.QUERY, 

431 description='Display tokens for all users (superuser only)', 

432 ) 

433 ] 

434 ) 

435 def get(self, request, *args, **kwargs): 

436 """Details for a user token.""" 

437 return super().get(request, *args, **kwargs) 

438 

439 

440class TokenListView(TokenMixin, ListCreateAPI): 

441 """List of user tokens for current user.""" 

442 

443 filter_backends = SEARCH_ORDER_FILTER 

444 search_fields = ['name', 'key'] 

445 ordering_fields = [ 

446 'created', 

447 'expiry', 

448 'last_seen', 

449 'user', 

450 'name', 

451 'revoked', 

452 'revoked', 

453 ] 

454 filterset_fields = ['revoked', 'user'] 

455 queryset = ApiToken.objects.none() 

456 

457 def create(self, request, *args, **kwargs): 

458 """Create token and show key to user.""" 

459 resp = super().create(request, *args, **kwargs) 

460 resp.data['token'] = self.serializer_class.Meta.model.objects.get( 

461 id=resp.data['id'] 

462 ).key 

463 return resp 

464 

465 def get(self, request, *args, **kwargs): 

466 """List of user tokens for current user.""" 

467 return super().get(request, *args, **kwargs) 

468 

469 

470class TokenDetailView(TokenMixin, DestroyAPIView, RetrieveAPI): 

471 """Details for a user token.""" 

472 

473 def perform_destroy(self, instance): 

474 """Revoke token.""" 

475 instance.revoked = True 

476 instance.save() 

477 

478 

479class LoginRedirect(RedirectView): 

480 """Redirect to the correct starting page after backend login.""" 

481 

482 def get_redirect_url(self, *args, **kwargs): 

483 """Return the URL to redirect to.""" 

484 return f'/{FRONTEND_URL_BASE}/logged-in/' 

485 

486 

487class UserProfileDetail(RetrieveUpdateAPI): 

488 """Detail endpoint for the user profile. 

489 

490 Permissions: 

491 - Any authenticated user has write access against this endpoint 

492 - The endpoint always returns the profile associated with the current user 

493 """ 

494 

495 queryset = UserProfile.objects.all() 

496 serializer_class = UserProfileSerializer 

497 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope] 

498 

499 def get_object(self): 

500 """Return the profile of the current user.""" 

501 return self.request.user.profile 

502 

503 

504user_urls = [ 

505 # Legacy endpoints (to avoid breaking existing API clients) 

506 # TODO @matmair - remove these legacy endpoints in the next breaking release 

507 path( 

508 'roles/', 

509 exclude_from_schema(RoleDetails, '/api/user/me/roles/').as_view(), 

510 name='api-user-roles_legacy', 

511 ), 

512 path( 

513 'token/', 

514 ensure_csrf_cookie( 

515 exclude_from_schema(GetAuthToken, '/api/user/me/token/').as_view() 

516 ), 

517 name='api-token_legacy', 

518 ), 

519 path( 

520 'profile/', 

521 exclude_from_schema(UserProfileDetail, '/api/user/me/profile/').as_view(), 

522 name='api-user-profile_legacy', 

523 ), 

524 # Individual user endpoints 

525 path( 

526 'me/', 

527 include([ 

528 path('profile/', UserProfileDetail.as_view(), name='api-user-profile'), 

529 path('roles/', RoleDetails.as_view(), name='api-user-roles'), 

530 path( 

531 'token/', ensure_csrf_cookie(GetAuthToken.as_view()), name='api-token' 

532 ), 

533 path('', MeUserDetail.as_view(), name='api-user-me'), 

534 ]), 

535 ), 

536 # User related endpoints 

537 path( 

538 'tokens/', 

539 include([ 

540 path('<int:pk>/', TokenDetailView.as_view(), name='api-token-detail'), 

541 path('', TokenListView.as_view(), name='api-token-list'), 

542 ]), 

543 ), 

544 path( 

545 'owner/', 

546 include([ 

547 path('<int:pk>/', OwnerDetail.as_view(), name='api-owner-detail'), 

548 path('', OwnerList.as_view(), name='api-owner-list'), 

549 ]), 

550 ), 

551 path( 

552 'group/', 

553 include([ 

554 path('<int:pk>/', GroupDetail.as_view(), name='api-group-detail'), 

555 path('', GroupList.as_view(), name='api-group-list'), 

556 ]), 

557 ), 

558 path( 

559 'ruleset/', 

560 include([ 

561 path('<int:pk>/', RuleSetDetail.as_view(), name='api-ruleset-detail'), 

562 path('', RuleSetList.as_view(), name='api-ruleset-list'), 

563 ]), 

564 ), 

565 path( 

566 '<int:pk>/', 

567 include([ 

568 path( 

569 'set-password/', 

570 UserDetailSetPassword.as_view(), 

571 name='api-user-set-password', 

572 ), 

573 path('', UserDetail.as_view(), name='api-user-detail'), 

574 ]), 

575 ), 

576 path('', UserList.as_view(), name='api-user-list'), 

577]