304 lines
15 KiB
Python
304 lines
15 KiB
Python
from django.test import Client
|
|
|
|
from authentication.tests import SignatureAuthClient, UserTestMixin, GroupTestMixin, ToolshedTestCase, \
|
|
DummyExternalUser
|
|
from authentication.models import Group, GroupInvite, GroupInviteIncoming, GroupMembership, KnownIdentity
|
|
|
|
client = SignatureAuthClient()
|
|
|
|
|
|
class GroupModelTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.prepare_users()
|
|
self.prepare_groups()
|
|
|
|
def test_group_str(self):
|
|
self.assertEqual(str(self.f['group1']), '#group1@' + self.f['example_com'].name)
|
|
|
|
def test_is_member(self):
|
|
self.assertTrue(self.f['group1'].is_member(self.f['local_user1'].public_identity))
|
|
self.assertFalse(self.f['group1'].is_member(self.f['local_user2'].public_identity))
|
|
|
|
|
|
class GroupApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.prepare_users()
|
|
|
|
def test_create_group(self):
|
|
reply = client.post('/api/v1/groups/', self.f['local_user1'], {'name': 'workshop'})
|
|
self.assertEqual(reply.status_code, 201)
|
|
group = Group.objects.get(name='workshop')
|
|
self.assertEqual(group.domain, self.f['local_user1'].domain)
|
|
self.assertTrue(group.is_member(self.f['local_user1'].public_identity))
|
|
|
|
def test_create_group_duplicate_name(self):
|
|
client.post('/api/v1/groups/', self.f['local_user1'], {'name': 'workshop'})
|
|
reply = client.post('/api/v1/groups/', self.f['local_user1'], {'name': 'workshop'})
|
|
self.assertEqual(reply.status_code, 409)
|
|
|
|
def test_create_group_missing_name(self):
|
|
reply = client.post('/api/v1/groups/', self.f['local_user1'], {})
|
|
self.assertEqual(reply.status_code, 400)
|
|
|
|
def test_list_groups(self):
|
|
self.prepare_groups()
|
|
reply = client.get('/api/v1/groups/', self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
self.assertEqual(len(reply.json()), 1)
|
|
reply2 = client.get('/api/v1/groups/', self.f['local_user2'])
|
|
self.assertEqual(reply2.status_code, 200)
|
|
self.assertEqual(len(reply2.json()), 0)
|
|
|
|
def test_group_detail_member(self):
|
|
self.prepare_groups()
|
|
group = self.f['group1']
|
|
reply = client.get('/api/v1/groups/{}@{}/'.format(group.name, group.domain), self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
self.assertEqual(reply.json()['handle'], str(group))
|
|
self.assertEqual(len(reply.json()['members']), 1)
|
|
|
|
def test_group_detail_non_member(self):
|
|
self.prepare_groups()
|
|
group = self.f['group1']
|
|
reply = client.get('/api/v1/groups/{}@{}/'.format(group.name, group.domain), self.f['local_user2'])
|
|
self.assertEqual(reply.status_code, 404)
|
|
|
|
def test_group_detail_no_such_group(self):
|
|
reply = client.get('/api/v1/groups/nonexistent@example.com/', self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 404)
|
|
|
|
def test_remove_member(self):
|
|
self.prepare_groups()
|
|
group = self.f['group1']
|
|
group.members.add(self.f['local_user2'].public_identity)
|
|
reply = client.delete('/api/v1/groups/{}@{}/members/{}/'.format(
|
|
group.name, group.domain, self.f['local_user2'].public_identity.id), self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 204)
|
|
self.assertEqual(group.members.count(), 1)
|
|
|
|
def test_remove_last_member_blocked(self):
|
|
self.prepare_groups()
|
|
group = self.f['group1']
|
|
reply = client.delete('/api/v1/groups/{}@{}/members/{}/'.format(
|
|
group.name, group.domain, self.f['local_user1'].public_identity.id), self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 400)
|
|
self.assertEqual(group.members.count(), 1)
|
|
|
|
def test_remove_member_non_member_denied(self):
|
|
self.prepare_groups()
|
|
group = self.f['group1']
|
|
group.members.add(self.f['local_user2'].public_identity)
|
|
reply = client.delete('/api/v1/groups/{}@{}/members/{}/'.format(
|
|
group.name, group.domain, self.f['local_user1'].public_identity.id), self.f['ext_user1'])
|
|
self.assertEqual(reply.status_code, 404)
|
|
self.assertEqual(group.members.count(), 2)
|
|
|
|
|
|
class GroupInviteApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.prepare_users()
|
|
self.prepare_groups()
|
|
|
|
def test_invite_local_member_full_flow(self):
|
|
group = self.f['group1']
|
|
inviter = self.f['local_user1']
|
|
invitee = self.f['local_user2']
|
|
|
|
reply1 = client.post('/api/v1/groups/{}@{}/invites/'.format(group.name, group.domain), inviter,
|
|
{'invitee': str(invitee)})
|
|
self.assertEqual(reply1.status_code, 201)
|
|
secret = reply1.json()['secret']
|
|
self.assertEqual(GroupInvite.objects.count(), 1)
|
|
|
|
reply2 = client.post('/api/v1/groupinvites/', inviter, {
|
|
'group': str(group), 'inviter': str(inviter), 'inviter_key': inviter.public_key(),
|
|
'invitee': str(invitee), 'secret': secret
|
|
})
|
|
self.assertEqual(reply2.status_code, 201)
|
|
incoming = GroupInviteIncoming.objects.get()
|
|
self.assertEqual(incoming.invitee_user, invitee)
|
|
|
|
reply3 = client.get('/api/v1/groupinvites/', invitee)
|
|
self.assertEqual(reply3.status_code, 200)
|
|
self.assertEqual(len(reply3.json()), 1)
|
|
self.assertEqual(reply3.json()[0]['group'], str(group))
|
|
self.assertEqual(reply3.json()[0]['inviter'], str(inviter))
|
|
|
|
reply4 = client.post('/api/v1/group_invites/accept/', invitee, {
|
|
'group': str(group), 'invitee': str(invitee), 'invitee_key': invitee.public_key(), 'secret': secret
|
|
})
|
|
self.assertEqual(reply4.status_code, 201)
|
|
self.assertTrue(group.is_member(invitee.public_identity))
|
|
self.assertEqual(GroupInvite.objects.count(), 0)
|
|
|
|
reply5 = client.post('/api/v1/groupinvites/{}/accept/'.format(incoming.id), invitee)
|
|
self.assertEqual(reply5.status_code, 201)
|
|
self.assertEqual(reply5.json()['handle'], str(group))
|
|
self.assertEqual(GroupInviteIncoming.objects.count(), 0)
|
|
membership = GroupMembership.objects.get(user=invitee)
|
|
self.assertEqual(membership.group_name, group.name)
|
|
self.assertEqual(membership.group_domain, group.domain)
|
|
|
|
def test_invite_non_member_denied(self):
|
|
group = self.f['group1']
|
|
reply = client.post('/api/v1/groups/{}@{}/invites/'.format(group.name, group.domain), self.f['local_user2'],
|
|
{'invitee': str(self.f['ext_user1'])})
|
|
self.assertEqual(reply.status_code, 404)
|
|
self.assertEqual(GroupInvite.objects.count(), 0)
|
|
|
|
def test_invite_already_member(self):
|
|
group = self.f['group1']
|
|
group.members.add(self.f['local_user2'].public_identity)
|
|
reply = client.post('/api/v1/groups/{}@{}/invites/'.format(group.name, group.domain), self.f['local_user1'],
|
|
{'invitee': str(self.f['local_user2'])})
|
|
self.assertEqual(reply.status_code, 208)
|
|
|
|
def test_decline_invite(self):
|
|
invite = GroupInviteIncoming.objects.create(
|
|
group_name=self.f['group1'].name, group_domain=self.f['group1'].domain,
|
|
inviter_username=self.f['local_user1'].username, inviter_domain=self.f['local_user1'].domain,
|
|
invitee_user=self.f['local_user2'], secret='some-secret')
|
|
reply = client.delete('/api/v1/groupinvites/{}/'.format(invite.id), self.f['local_user2'])
|
|
self.assertEqual(reply.status_code, 204)
|
|
self.assertEqual(GroupInviteIncoming.objects.count(), 0)
|
|
|
|
def test_accept_wrong_secret(self):
|
|
group = self.f['group1']
|
|
inviter = self.f['local_user1']
|
|
invitee = self.f['local_user2']
|
|
client.post('/api/v1/groups/{}@{}/invites/'.format(group.name, group.domain), inviter, {'invitee': str(invitee)})
|
|
reply = client.post('/api/v1/group_invites/accept/', invitee, {
|
|
'group': str(group), 'invitee': str(invitee), 'invitee_key': invitee.public_key(),
|
|
'secret': 'not-the-right-secret'
|
|
})
|
|
self.assertEqual(reply.status_code, 404)
|
|
self.assertFalse(group.is_member(invitee.public_identity))
|
|
|
|
def test_accept_remote_member(self):
|
|
group = self.f['group1']
|
|
inviter = self.f['local_user1']
|
|
invitee = DummyExternalUser('newmember', 'remote.example', known=False)
|
|
reply1 = client.post('/api/v1/groups/{}@{}/invites/'.format(group.name, group.domain), inviter,
|
|
{'invitee': str(invitee)})
|
|
secret = reply1.json()['secret']
|
|
|
|
reply = client.post('/api/v1/group_invites/accept/', invitee, {
|
|
'group': str(group), 'invitee': str(invitee), 'invitee_key': invitee.public_key(), 'secret': secret
|
|
})
|
|
self.assertEqual(reply.status_code, 201)
|
|
identity = KnownIdentity.objects.get(username='newmember', domain='remote.example')
|
|
self.assertTrue(group.is_member(identity))
|
|
# This lands on the group's own home backend, not the invitee's -- the invitee here isn't
|
|
# even a local ToolshedUser on this backend, so there's nothing to point at locally.
|
|
self.assertEqual(GroupMembership.objects.count(), 0)
|
|
|
|
def test_accept_bad_signature(self):
|
|
group = self.f['group1']
|
|
inviter = self.f['local_user1']
|
|
invitee = self.f['local_user2']
|
|
reply1 = client.post('/api/v1/groups/{}@{}/invites/'.format(group.name, group.domain), inviter,
|
|
{'invitee': str(invitee)})
|
|
secret = reply1.json()['secret']
|
|
bad_signature_client = SignatureAuthClient(bad_signature=True)
|
|
reply = bad_signature_client.post('/api/v1/group_invites/accept/', invitee, {
|
|
'group': str(group), 'invitee': str(invitee), 'invitee_key': invitee.public_key(), 'secret': secret
|
|
})
|
|
self.assertEqual(reply.status_code, 401)
|
|
self.assertFalse(group.is_member(invitee.public_identity))
|
|
|
|
|
|
class GroupMembershipApiTestCase(UserTestMixin, GroupTestMixin, ToolshedTestCase):
|
|
"""Covers the pointer index kept on a member's own home backend (GroupMembership), which is
|
|
what lets that backend remember foreign group membership, analogous to how a friendship ends
|
|
up recorded on both sides rather than only on the group's authoritative backend."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.prepare_users()
|
|
self.prepare_groups()
|
|
|
|
def test_record_membership_for_remote_group(self):
|
|
# local_user2's own home backend has no Group row at all for this group -- it's hosted
|
|
# entirely on another domain -- yet it should still remember the membership once the
|
|
# invitee's client confirms the accept against the group's home backend succeeded.
|
|
invitee = self.f['local_user2']
|
|
incoming = GroupInviteIncoming.objects.create(
|
|
group_name='remoteworkshop', group_domain='other.example',
|
|
inviter_username='someone', inviter_domain='other.example',
|
|
invitee_user=invitee, secret='some-secret')
|
|
|
|
reply = client.post('/api/v1/groupinvites/{}/accept/'.format(incoming.id), invitee)
|
|
|
|
self.assertEqual(reply.status_code, 201)
|
|
self.assertEqual(reply.json()['handle'], '#remoteworkshop@other.example')
|
|
self.assertEqual(GroupInviteIncoming.objects.count(), 0)
|
|
membership = GroupMembership.objects.get(user=invitee)
|
|
self.assertEqual(membership.group_name, 'remoteworkshop')
|
|
self.assertEqual(membership.group_domain, 'other.example')
|
|
|
|
def test_record_membership_wrong_user_denied(self):
|
|
incoming = GroupInviteIncoming.objects.create(
|
|
group_name='remoteworkshop', group_domain='other.example',
|
|
inviter_username='someone', inviter_domain='other.example',
|
|
invitee_user=self.f['local_user2'], secret='some-secret')
|
|
|
|
reply = client.post('/api/v1/groupinvites/{}/accept/'.format(incoming.id), self.f['local_user1'])
|
|
|
|
self.assertEqual(reply.status_code, 404)
|
|
self.assertEqual(GroupMembership.objects.count(), 0)
|
|
self.assertEqual(GroupInviteIncoming.objects.count(), 1)
|
|
|
|
def test_decline_does_not_record_membership(self):
|
|
incoming = GroupInviteIncoming.objects.create(
|
|
group_name='remoteworkshop', group_domain='other.example',
|
|
inviter_username='someone', inviter_domain='other.example',
|
|
invitee_user=self.f['local_user2'], secret='some-secret')
|
|
|
|
reply = client.delete('/api/v1/groupinvites/{}/'.format(incoming.id), self.f['local_user2'])
|
|
|
|
self.assertEqual(reply.status_code, 204)
|
|
self.assertEqual(GroupMembership.objects.count(), 0)
|
|
|
|
def test_record_membership_already_a_member_is_idempotent(self):
|
|
# Re-invited (or re-accepting) into a group we already have a pointer for shouldn't blow
|
|
# up on the unique_together constraint, and shouldn't duplicate the pointer either.
|
|
invitee = self.f['local_user2']
|
|
GroupMembership.objects.create(
|
|
user=invitee, group_name='remoteworkshop', group_domain='other.example')
|
|
incoming = GroupInviteIncoming.objects.create(
|
|
group_name='remoteworkshop', group_domain='other.example',
|
|
inviter_username='someone', inviter_domain='other.example',
|
|
invitee_user=invitee, secret='some-secret')
|
|
|
|
reply = client.post('/api/v1/groupinvites/{}/accept/'.format(incoming.id), invitee)
|
|
|
|
self.assertEqual(reply.status_code, 201)
|
|
self.assertEqual(GroupMembership.objects.filter(user=invitee).count(), 1)
|
|
self.assertEqual(GroupInviteIncoming.objects.count(), 0)
|
|
|
|
def test_list_memberships(self):
|
|
GroupMembership.objects.create(
|
|
user=self.f['local_user1'], group_name='remoteworkshop', group_domain='other.example')
|
|
|
|
reply = client.get('/api/v1/groupmemberships/', self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
self.assertEqual(len(reply.json()), 1)
|
|
self.assertEqual(reply.json()[0]['handle'], '#remoteworkshop@other.example')
|
|
|
|
reply2 = client.get('/api/v1/groupmemberships/', self.f['local_user2'])
|
|
self.assertEqual(reply2.status_code, 200)
|
|
self.assertEqual(len(reply2.json()), 0)
|
|
|
|
def test_list_memberships_unauthorized(self):
|
|
GroupMembership.objects.create(
|
|
user=self.f['local_user1'], group_name='remoteworkshop', group_domain='other.example')
|
|
|
|
reply = client.get('/api/v1/groupmemberships/', self.f['ext_user1'])
|
|
|
|
# authenticate() returns bare None (not raise) for a caller with no local ToolshedUser, so
|
|
# DRF falls through to IsAuthenticated denying an anonymous request -- 403, not 401 (same
|
|
# as any other SignatureAuthenticationLocal-only endpoint, see e.g. dropFriend).
|
|
self.assertEqual(reply.status_code, 403)
|