Coverage for sm/views_group.py: 79%

1from django.views.generic import ListView, FormView, TemplateView 

2from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin 

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

4from django.db.models import Q 

5from .models import GroupProfile, Invitation 

6from django.urls import reverse_lazy, reverse 

7from django.shortcuts import get_object_or_404, redirect 

8from django.contrib import messages 

9from django.utils.translation import gettext as _ 

10from django import forms 

11from django.views import View 

12from django.core.mail import send_mail 

13from django.contrib.auth.password_validation import validate_password 

14from django.core.exceptions import ValidationError 

15from django.template.loader import render_to_string 

16from typing import Any 

17from .utils_permissions import ( 

18 get_group_permissions_for_model, 

19 APP_MODELS, 

20 MODEL_LABELS, 

21) 

22from django.utils import timezone 

23from django.conf import settings 

24from django.http import JsonResponse 

25 

26 

27class GroupOwnerRequiredMixin(UserPassesTestMixin): 

28 def test_func(self) -> bool: 

29 # Superusers can manage all groups 

30 if self.request.user.is_superuser: 

31 return True 

32 # Check if user is owner of at least one group 

33 return GroupProfile.objects.filter(owner=self.request.user).exists() 

34 

35 

36class GroupMemberListView(LoginRequiredMixin, GroupOwnerRequiredMixin, ListView): 

37 template_name = "group/member_list.html" 

38 context_object_name = "groups" 

39 

40 def get_queryset(self) -> Any: 

41 if self.request.user.is_superuser: 

42 return Group.objects.all().prefetch_related("user_set", "profile") 

43 return ( 

44 Group.objects.filter( 

45 Q(profile__owner=self.request.user) | Q(user=self.request.user) 

46 ) 

47 .prefetch_related("user_set", "profile") 

48 .distinct() 

49 ) 

50 

51 

52class AddGroupMemberForm(forms.Form): 

53 username = forms.CharField(label=_("Username or Email")) 

54 

55 def clean_username(self) -> Any: 

56 username = self.cleaned_data["username"] 

57 try: 

58 if "@" in username: 

59 return User.objects.get(email=username) 

60 return User.objects.get(username=username) 

61 except User.DoesNotExist: 

62 raise forms.ValidationError(_("User does not exist.")) 

63 

64 

65class CreateGroupForm(forms.Form): 

66 name = forms.CharField(label=_("Group Name"), max_length=150) 

67 max_users = forms.IntegerField( 

68 label=_("Max Users"), 

69 initial=10, 

70 min_value=2, 

71 max_value=100, 

72 help_text=_("Maximum number of users in this group"), 

73 ) 

74 

75 

76class AddGroupMemberView(LoginRequiredMixin, GroupOwnerRequiredMixin, FormView): 

77 form_class = AddGroupMemberForm 

78 template_name = "group/add_member.html" 

79 

80 def get_success_url(self) -> str: 

81 return reverse_lazy("group_member_list") 

82 

83 def form_valid(self, form: Any) -> Any: 

84 group_id = self.kwargs.get("group_id") 

85 # Ensure the user is the owner OR a superuser 

86 if self.request.user.is_superuser: 

87 group = get_object_or_404(Group, pk=group_id) 

88 else: 

89 group = get_object_or_404( 

90 Group, pk=group_id, profile__owner=self.request.user 

91 ) 

92 

93 user_to_add = form.cleaned_data["username"] 

94 

95 if group.user_set.count() >= group.profile.max_users: 

96 messages.error( 

97 self.request, 

98 _("User quota exceeded for this group (%d users).") 

99 % group.profile.max_users, 

100 ) 

101 return self.form_invalid(form) 

102 

103 group.user_set.add(user_to_add) 

104 messages.success( 

105 self.request, 

106 _("User %s added to group %s.") % (user_to_add.username, group.name), 

107 ) 

108 return super().form_valid(form) 

109 

110 

111class RemoveGroupMemberView(LoginRequiredMixin, GroupOwnerRequiredMixin, View): 

112 def post(self, request: Any, *args: Any, **kwargs: Any) -> Any: 

113 group_id = self.kwargs.get("group_id") 

114 user_id = self.kwargs.get("user_id") 

115 

116 if self.request.user.is_superuser: 

117 group = get_object_or_404(Group, pk=group_id) 

118 else: 

119 group = get_object_or_404( 

120 Group, pk=group_id, profile__owner=self.request.user 

121 ) 

122 

123 user_to_remove = get_object_or_404(User, pk=user_id) 

124 

125 if user_to_remove == self.request.user and not self.request.user.is_superuser: 

126 messages.error( 

127 request, _("You cannot remove yourself from your own group.") 

128 ) 

129 else: 

130 group.user_set.remove(user_to_remove) 

131 messages.success( 

132 request, 

133 _("User %s removed from group %s.") 

134 % (user_to_remove.username, group.name), 

135 ) 

136 

137 return redirect("group_member_list") 

138 

139 

140class GroupPermissionForm(forms.Form): 

141 def __init__(self, *args: Any, **kwargs: Any) -> None: 

142 self.group = kwargs.pop("group") 

143 super().__init__(*args, **kwargs) 

144 

145 self.models_to_manage = [ 

146 (app_label, MODEL_LABELS.get(app_label, app_label)) 

147 for app_label, _model in APP_MODELS 

148 ] 

149 

150 current_perms = self.group.permissions.all().values_list("codename", flat=True) 

151 

152 for app_label, label in self.models_to_manage: 

153 # Find the actual change permission codename which might be 'change_model' 

154 # because we named our model 'Model' 

155 perms = get_group_permissions_for_model(app_label) 

156 change_perm = next( 

157 (p for p in perms if p.codename.startswith("change_")), None 

158 ) 

159 

160 initial_val = False 

161 if change_perm: 

162 initial_val = change_perm.codename in current_perms 

163 

164 self.fields[f"edit_{app_label}"] = forms.BooleanField( 

165 label=_("Can Edit %s") % label, 

166 required=False, 

167 initial=initial_val, 

168 ) 

169 

170 def save(self) -> None: 

171 for app_label_tuple in self.models_to_manage: 

172 app_label = app_label_tuple[0] 

173 perms = get_group_permissions_for_model(app_label) 

174 # We manage add, change, delete as "Edit" 

175 edit_perms = [ 

176 p 

177 for p in perms 

178 if p.codename.startswith(("add_", "change_", "delete_")) 

179 ] 

180 

181 if self.cleaned_data.get(f"edit_{app_label}"): 

182 self.group.permissions.add(*edit_perms) 

183 else: 

184 self.group.permissions.remove(*edit_perms) 

185 

186 

187class GroupPermissionUpdateView(LoginRequiredMixin, GroupOwnerRequiredMixin, FormView): 

188 template_name = "group/permission_edit.html" 

189 form_class = GroupPermissionForm 

190 

191 def get_form_kwargs(self) -> Any: 

192 kwargs = super().get_form_kwargs() 

193 group_id = self.kwargs.get("group_id") 

194 if self.request.user.is_superuser: 

195 kwargs["group"] = get_object_or_404(Group, pk=group_id) 

196 else: 

197 kwargs["group"] = get_object_or_404( 

198 Group, pk=group_id, profile__owner=self.request.user 

199 ) 

200 return kwargs 

201 

202 def get_success_url(self) -> str: 

203 return reverse_lazy("group_member_list") 

204 

205 def form_valid(self, form: Any) -> Any: 

206 form.save() 

207 messages.success(self.request, _("Group permissions updated successfully.")) 

208 return super().form_valid(form) 

209 

210 

211class UserPermissionForm(forms.Form): 

212 def __init__(self, *args: Any, **kwargs: Any) -> None: 

213 self.user = kwargs.pop("user") 

214 self.group = kwargs.pop("group") 

215 super().__init__(*args, **kwargs) 

216 

217 self.models_to_manage = [ 

218 (app_label, MODEL_LABELS.get(app_label, app_label)) 

219 for app_label, _model in APP_MODELS 

220 ] 

221 

222 group_perms = self.group.permissions.all().values_list("codename", flat=True) 

223 user_perms = self.user.user_permissions.all().values_list("codename", flat=True) 

224 

225 for app_label, label in self.models_to_manage: 

226 perms = get_group_permissions_for_model(app_label) 

227 change_perm = next( 

228 (p for p in perms if p.codename.startswith("change_")), None 

229 ) 

230 

231 initial_val = False 

232 if change_perm: 

233 initial_val = change_perm.codename in user_perms 

234 

235 inherited = change_perm.codename in group_perms if change_perm else False 

236 

237 self.fields[f"edit_{app_label}"] = forms.BooleanField( 

238 label=_("Can Edit %s") % label, 

239 required=False, 

240 initial=initial_val, 

241 ) 

242 if inherited: 

243 self.fields[f"edit_{app_label}"].help_text = _("(Inherited from group)") 

244 

245 def save(self) -> None: 

246 for app_label_tuple in self.models_to_manage: 

247 app_label = app_label_tuple[0] 

248 perms = get_group_permissions_for_model(app_label) 

249 edit_perms = [ 

250 p 

251 for p in perms 

252 if p.codename.startswith(("add_", "change_", "delete_")) 

253 ] 

254 

255 if self.cleaned_data.get(f"edit_{app_label}"): 

256 self.user.user_permissions.add(*edit_perms) 

257 else: 

258 self.user.user_permissions.remove(*edit_perms) 

259 

260 

261class UserPermissionUpdateView(LoginRequiredMixin, GroupOwnerRequiredMixin, FormView): 

262 template_name = "group/user_permission_edit.html" 

263 form_class = UserPermissionForm 

264 

265 def get_form_kwargs(self) -> Any: 

266 kwargs = super().get_form_kwargs() 

267 group_id = self.kwargs.get("group_id") 

268 user_id = self.kwargs.get("user_id") 

269 if self.request.user.is_superuser: 

270 group = get_object_or_404(Group, pk=group_id) 

271 else: 

272 group = get_object_or_404( 

273 Group, pk=group_id, profile__owner=self.request.user 

274 ) 

275 # Only group owners may edit the permissions of their own group's 

276 # members. Without the membership check an owner could grant/revoke 

277 # the (global) user permissions of arbitrary users in other groups. 

278 kwargs["group"] = group 

279 kwargs["user"] = get_object_or_404(User, pk=user_id, groups=group) 

280 return kwargs 

281 

282 def get_success_url(self) -> str: 

283 return reverse_lazy("group_member_list") 

284 

285 def form_valid(self, form: Any) -> Any: 

286 form.save() 

287 messages.success(self.request, _("User permissions updated successfully.")) 

288 return super().form_valid(form) 

289 

290 

291class InviteGroupMemberForm(forms.Form): 

292 email = forms.EmailField(label=_("Email Address")) 

293 

294 def clean_email(self) -> Any: 

295 email = self.cleaned_data["email"] 

296 if User.objects.filter(email__iexact=email).exists(): 

297 raise forms.ValidationError(_("A user with this email already exists.")) 

298 return email.lower() 

299 

300 

301class InviteGroupMemberView(LoginRequiredMixin, GroupOwnerRequiredMixin, FormView): 

302 form_class = InviteGroupMemberForm 

303 template_name = "group/invite_member.html" 

304 

305 def get_success_url(self) -> str: 

306 return reverse_lazy("group_member_list") 

307 

308 def get_group(self) -> Group: 

309 group_id = self.kwargs.get("group_id") 

310 if self.request.user.is_superuser: 

311 return get_object_or_404(Group, pk=group_id) 

312 return get_object_or_404(Group, pk=group_id, profile__owner=self.request.user) 

313 

314 def form_valid(self, form: Any) -> Any: 

315 group = self.get_group() 

316 email = form.cleaned_data["email"] 

317 

318 existing_invitation = Invitation.objects.filter( 

319 email__iexact=email, group=group 

320 ).first() 

321 if existing_invitation and not existing_invitation.is_expired(): 

322 messages.error( 

323 self.request, 

324 _("An invitation has already been sent to this email address."), 

325 ) 

326 return self.form_invalid(form) 

327 

328 # Replace any expired/used invitation for the same email+group so we 

329 # don't trip the (email, group) unique constraint. 

330 Invitation.objects.filter(email__iexact=email, group=group).delete() 

331 

332 invitation = Invitation.objects.create( 

333 email=email, 

334 group=group, 

335 created_by=self.request.user, 

336 ) 

337 

338 self.send_invitation_email(invitation) 

339 

340 messages.success( 

341 self.request, 

342 _("Invitation sent to %(email)s.") % {"email": email}, 

343 ) 

344 return super().form_valid(form) 

345 

346 def send_invitation_email(self, invitation: Invitation) -> None: 

347 invite_url = self.request.build_absolute_uri( 

348 reverse("accept_invitation", kwargs={"token": invitation.token}) 

349 ) 

350 subject = _("Invitation to join %(group_name)s on ServerManager") % { 

351 "group_name": invitation.group.name 

352 } 

353 message = render_to_string( 

354 "group/email/invite.txt", 

355 { 

356 "invitation": invitation, 

357 "invite_url": invite_url, 

358 "owner_name": ( 

359 invitation.created_by.username 

360 if invitation.created_by 

361 else "An administrator" 

362 ), 

363 }, 

364 ) 

365 send_mail( 

366 subject, 

367 message, 

368 getattr(settings, "DEFAULT_FROM_EMAIL", "noreply@servermanager"), 

369 [invitation.email], 

370 fail_silently=False, 

371 ) 

372 

373 

374class AcceptInvitationView(TemplateView): 

375 template_name = "group/accept_invitation.html" 

376 

377 def get_context_data(self, **kwargs: Any) -> Any: 

378 context = super().get_context_data(**kwargs) 

379 token = self.kwargs.get("token") 

380 invitation = get_object_or_404(Invitation, token=token) 

381 

382 if invitation.is_expired(): 

383 context["error"] = _("This invitation has expired.") 

384 return context 

385 

386 if invitation.accepted_at: 

387 context["error"] = _("This invitation has already been used.") 

388 return context 

389 

390 context["invitation"] = invitation 

391 return context 

392 

393 def post(self, request: Any, *args: Any, **kwargs: Any) -> Any: 

394 token = self.kwargs.get("token") 

395 invitation = get_object_or_404(Invitation, token=token) 

396 

397 if invitation.is_expired() or invitation.accepted_at: 

398 messages.error(request, _("This invitation is no longer valid.")) 

399 return redirect("account_login") 

400 

401 username = request.POST.get("username") 

402 email = request.POST.get("email") 

403 password = request.POST.get("password") 

404 password_confirm = request.POST.get("password_confirm") 

405 

406 if not username or not email or not password: 

407 messages.error(request, _("All fields are required.")) 

408 return self.get(request, *args, **kwargs) 

409 

410 if password != password_confirm: 

411 messages.error(request, _("Passwords do not match.")) 

412 return self.get(request, *args, **kwargs) 

413 

414 # The invitee must claim the invited account under the invited email, 

415 # otherwise anyone with the link could register a new identity. 

416 if email.lower() != invitation.email.lower(): 

417 messages.error(request, _("The email does not match the invitation.")) 

418 return self.get(request, *args, **kwargs) 

419 

420 try: 

421 validate_password(password) 

422 except ValidationError as exc: 

423 messages.error( 

424 request, _("Password is not strong enough: %s") % " ".join(exc.messages) 

425 ) 

426 return self.get(request, *args, **kwargs) 

427 

428 if User.objects.filter(username=username).exists(): 

429 messages.error(request, _("Username already exists.")) 

430 return self.get(request, *args, **kwargs) 

431 

432 if User.objects.filter(email__iexact=email).exists(): 

433 messages.error(request, _("Email already registered.")) 

434 return self.get(request, *args, **kwargs) 

435 

436 user = User.objects.create_user( 

437 username=username, 

438 email=email, 

439 password=password, 

440 ) 

441 

442 invitation.group.user_set.add(user) 

443 invitation.accepted_at = timezone.now() 

444 invitation.save() 

445 

446 messages.success( 

447 request, 

448 _("You have joined %(group_name)s!") 

449 % {"group_name": invitation.group.name}, 

450 ) 

451 

452 from django.contrib.auth import login 

453 

454 login(request, user, backend="django.contrib.auth.backends.ModelBackend") 

455 

456 return redirect("dashboard") 

457 

458 

459class GroupFilterView(View): 

460 def post(self, request: Any, *args: Any, **kwargs: Any) -> Any: 

461 if not request.user.is_authenticated: 

462 return JsonResponse({"error": "Unauthorized"}, status=401) 

463 

464 # Support both the checkbox format ("1,2,3" in `groups`) and the 

465 # single-select format (a `group` field) used by the search filter. 

466 groups_param = request.POST.get("groups", "") 

467 single = request.POST.get("group", "") 

468 if groups_param: 

469 selected_groups = [g for g in groups_param.split(",") if g] 

470 elif single: 

471 selected_groups = [single] 

472 else: 

473 selected_groups = [] 

474 

475 if selected_groups: 

476 request.session["selected_groups"] = selected_groups 

477 else: 

478 request.session["selected_groups"] = [] 

479 

480 request.session.modified = True 

481 

482 # If a `next` URL was provided, redirect there so the filter applies 

483 # to the referring page (e.g. search results). 

484 next_url = request.POST.get("next") or request.GET.get("next") 

485 if next_url: 

486 from django.shortcuts import redirect 

487 

488 return redirect(next_url) 

489 

490 return JsonResponse({"status": "success"}) 

491 

492 

493class GroupCreateView(LoginRequiredMixin, FormView): 

494 form_class = CreateGroupForm 

495 template_name = "group/member_list.html" 

496 success_url = reverse_lazy("group_member_list") 

497 

498 def form_valid(self, form: Any) -> Any: 

499 name = form.cleaned_data["name"] 

500 max_users = form.cleaned_data.get("max_users", 10) 

501 

502 max_groups = 5 

503 owned_groups_count = self.request.user.owned_groups.count() 

504 if owned_groups_count >= max_groups: 

505 messages.error( 

506 self.request, 

507 _("You can only create up to %d groups.") % max_groups, 

508 ) 

509 return redirect("group_member_list") 

510 

511 existing = Group.objects.filter(name=name).first() 

512 if existing: 

513 messages.error(self.request, _("A group with this name already exists.")) 

514 return redirect("group_member_list") 

515 

516 group = Group.objects.create(name=name) 

517 # Profile is already created by create_group_profile signal 

518 profile = group.profile 

519 profile.owner = self.request.user 

520 profile.max_users = max_users 

521 profile.save() 

522 

523 self.request.user.groups.add(group) 

524 

525 from .utils_permissions import sync_group_permissions 

526 

527 sync_group_permissions(group, grant_all=True) 

528 

529 messages.success(self.request, _("Group '%s' created successfully.") % name) 

530 return super().form_valid(form) 

531 

532 def form_invalid(self, form: Any) -> Any: 

533 messages.error(self.request, _("Failed to create group.")) 

534 return redirect("group_member_list")