-
-
Notifications
You must be signed in to change notification settings - Fork 302
fix: restrict invitation accept/decline actions to invited user only #5893
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: unstable
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -446,3 +446,93 @@ def test_update_invitation_decline(self): | |
| ).exists() | ||
| ) | ||
| self.assertTrue(models.Change.objects.filter(channel=self.channel).exists()) | ||
|
|
||
| def test_accept_invitation_by_channel_editor_is_forbidden(self): | ||
| invitation = models.Invitation.objects.create(**self.invitation_db_metadata) | ||
|
|
||
| self.client.force_authenticate(user=self.user) | ||
| response = self.client.post( | ||
| reverse("invitation-accept", kwargs={"pk": invitation.id}) | ||
| ) | ||
| self.assertEqual(response.status_code, 403, response.content) | ||
| invitation.refresh_from_db() | ||
| self.assertFalse(invitation.accepted) | ||
|
|
||
| def test_decline_invitation_by_channel_editor_is_forbidden(self): | ||
| invitation = models.Invitation.objects.create(**self.invitation_db_metadata) | ||
|
|
||
| self.client.force_authenticate(user=self.user) | ||
| response = self.client.post( | ||
| reverse("invitation-decline", kwargs={"pk": invitation.id}) | ||
| ) | ||
| self.assertEqual(response.status_code, 403, response.content) | ||
| invitation.refresh_from_db() | ||
| self.assertFalse(invitation.declined) | ||
|
|
||
| def test_accept_invitation_by_unrelated_user_is_not_found(self): | ||
| invitation = models.Invitation.objects.create(**self.invitation_db_metadata) | ||
| unrelated_user = testdata.user("unrelated@example.com") | ||
|
|
||
| self.client.force_authenticate(user=unrelated_user) | ||
| response = self.client.post( | ||
| reverse("invitation-accept", kwargs={"pk": invitation.id}) | ||
| ) | ||
| self.assertEqual(response.status_code, 404, response.content) | ||
| invitation.refresh_from_db() | ||
| self.assertFalse(invitation.accepted) | ||
|
|
||
| def test_decline_invitation_by_unrelated_user_is_not_found(self): | ||
| invitation = models.Invitation.objects.create(**self.invitation_db_metadata) | ||
| unrelated_user = testdata.user("unrelated@example.com") | ||
|
|
||
| self.client.force_authenticate(user=unrelated_user) | ||
| response = self.client.post( | ||
| reverse("invitation-decline", kwargs={"pk": invitation.id}) | ||
| ) | ||
| self.assertEqual(response.status_code, 404, response.content) | ||
| invitation.refresh_from_db() | ||
| self.assertFalse(invitation.declined) | ||
|
|
||
| def _make_admin(self, email="admin@example.com"): | ||
| user = testdata.user(email) | ||
| user.is_admin = True | ||
| user.save() | ||
| return user | ||
|
|
||
| def test_accept_invitation_by_admin_succeeds(self): | ||
| invitation = models.Invitation.objects.create(**self.invitation_db_metadata) | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The admin user setup is repeated identically in both admin tests. Extract to a small helper to avoid the duplication: def _make_admin(self, email="admin@example.com"):
user = testdata.user(email)
user.is_admin = True
user.save()
return user
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. done |
||
| admin_user = self._make_admin() | ||
|
|
||
| self.client.force_authenticate(user=admin_user) | ||
| response = self.client.post( | ||
| reverse("invitation-accept", kwargs={"pk": invitation.id}) | ||
| ) | ||
| self.assertEqual(response.status_code, 200, response.content) | ||
| invitation.refresh_from_db() | ||
| self.assertTrue(invitation.accepted) | ||
|
|
||
| def test_decline_invitation_by_admin_succeeds(self): | ||
|
Prashant-thakur77 marked this conversation as resolved.
|
||
| invitation = models.Invitation.objects.create(**self.invitation_db_metadata) | ||
| admin_user = self._make_admin() | ||
|
|
||
| self.client.force_authenticate(user=admin_user) | ||
| response = self.client.post( | ||
| reverse("invitation-decline", kwargs={"pk": invitation.id}) | ||
| ) | ||
| self.assertEqual(response.status_code, 200, response.content) | ||
| invitation.refresh_from_db() | ||
| self.assertTrue(invitation.declined) | ||
|
|
||
| def test_accept_revoked_invitation_returns_400(self): | ||
| invitation = models.Invitation.objects.create(**self.invitation_db_metadata) | ||
| invitation.revoked = True | ||
| invitation.save() | ||
|
|
||
| self.client.force_authenticate(user=self.invited_user) | ||
| response = self.client.post( | ||
| reverse("invitation-accept", kwargs={"pk": invitation.id}) | ||
| ) | ||
| self.assertEqual(response.status_code, 400, response.content) | ||
| invitation.refresh_from_db() | ||
| self.assertFalse(invitation.accepted) | ||
| self.assertFalse(self.channel.editors.filter(pk=self.invited_user.id).exists()) | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,7 +1,9 @@ | ||
| from django_filters.rest_framework import CharFilter | ||
| from django_filters.rest_framework import FilterSet | ||
| from rest_framework import serializers | ||
| from rest_framework import status | ||
| from rest_framework.decorators import action | ||
| from rest_framework.exceptions import PermissionDenied | ||
| from rest_framework.permissions import IsAuthenticated | ||
| from rest_framework.response import Response | ||
|
|
||
|
|
@@ -137,9 +139,21 @@ def perform_update(self, serializer): | |
| instance = serializer.save() | ||
| instance.save() | ||
|
|
||
| def _ensure_invitee(self, request, invitation): | ||
| if request.user.is_admin: | ||
|
Prashant-thakur77 marked this conversation as resolved.
|
||
| return | ||
| if (request.user.email or "").lower() != (invitation.email or "").lower(): | ||
| raise PermissionDenied("Only the invited user may perform this action.") | ||
|
|
||
| @action(detail=True, methods=["post"]) | ||
| def accept(self, request, pk=None): | ||
| invitation = self.get_object() | ||
| invitation = self.get_edit_object() | ||
| self._ensure_invitee(request, invitation) | ||
| if invitation.revoked: | ||
| return Response( | ||
| "Invitation has been revoked", | ||
| status=status.HTTP_400_BAD_REQUEST, | ||
| ) | ||
| invitation.accept() | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The Add a guard before proceeding: if invitation.revoked:
return Response(
"Invitation has been revoked",
status=status.HTTP_400_BAD_REQUEST,
)And a test to match the existing
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is out of your changes, but is a good guard to add here since the update serializer is doing so as well.
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. done:) |
||
| invitation.accepted = True | ||
| invitation.save() | ||
|
|
@@ -157,7 +171,8 @@ def accept(self, request, pk=None): | |
|
|
||
| @action(detail=True, methods=["post"]) | ||
| def decline(self, request, pk=None): | ||
| invitation = self.get_object() | ||
| invitation = self.get_edit_object() | ||
| self._ensure_invitee(request, invitation) | ||
| invitation.declined = True | ||
| invitation.save() | ||
| Change.create_change( | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.