0001-api-add-timestamp-parameter-to-users-synchronization.patch
| src/authentic2/api_views.py | ||
|---|---|---|
| 23 | 23 |
from django.conf import settings |
| 24 | 24 |
from django.contrib.auth import get_user_model |
| 25 | 25 |
from django.contrib.auth.hashers import identify_hasher |
| 26 |
from django.contrib.contenttypes.models import ContentType |
|
| 26 | 27 |
from django.core.exceptions import MultipleObjectsReturned |
| 27 | 28 |
from django.db import models |
| 28 | 29 |
from django.shortcuts import get_object_or_404 |
| ... | ... | |
| 58 | 59 | |
| 59 | 60 |
from . import api_mixins, app_settings, decorators, hooks |
| 60 | 61 |
from .a2_rbac.utils import get_default_ou |
| 61 |
from .custom_user.models import User |
|
| 62 |
from .apps.journal.models import Event |
|
| 63 |
from .custom_user.models import DeletedUser, User |
|
| 62 | 64 |
from .journal_event_types import UserLogin, UserRegistration |
| 63 | 65 |
from .models import Attribute, PasswordReset, Service |
| 64 | 66 |
from .passwords import get_password_checker |
| ... | ... | |
| 795 | 797 | |
| 796 | 798 |
class SynchronizationSerializer(serializers.Serializer): |
| 797 | 799 |
known_uuids = serializers.ListField(child=serializers.CharField()) |
| 800 |
timestamp = serializers.DateTimeField(required=False) |
|
| 798 | 801 | |
| 799 |
def check_uuids(self, uuids): |
|
| 802 |
def check_uuids(self, uuids, timestamp=None):
|
|
| 800 | 803 |
User = get_user_model() |
| 801 | 804 |
known_uuids = User.objects.filter(uuid__in=uuids).values_list('uuid', flat=True)
|
| 802 |
return set(uuids) - set(known_uuids) |
|
| 805 |
unknown_uuids = set(uuids) - set(known_uuids) |
|
| 806 |
modified_users_uuids = [] |
|
| 807 |
if timestamp: |
|
| 808 |
unknown_uuids = DeletedUser.objects.filter( |
|
| 809 |
old_uuid__in=unknown_uuids, deleted__gt=timestamp |
|
| 810 |
).values_list('old_uuid', flat=True)
|
|
| 811 |
user_ct = ContentType.objects.get_for_model(get_user_model()) |
|
| 812 |
user_events = Event.objects.filter( |
|
| 813 |
reference_ct_ids__contains=[user_ct.id], timestamp__gt=timestamp |
|
| 814 |
) |
|
| 815 |
for user_event in user_events: |
|
| 816 |
for reference in user_event.references: |
|
| 817 |
if ( |
|
| 818 |
reference is not None # xxx None references in journal?! |
|
| 819 |
and ContentType.objects.get_for_model(reference) == user_ct |
|
| 820 |
and reference.uuid not in modified_users_uuids |
|
| 821 |
and reference.uuid not in unknown_uuids |
|
| 822 |
): |
|
| 823 |
modified_users_uuids.append(reference.uuid) |
|
| 824 |
return (unknown_uuids, modified_users_uuids) |
|
| 803 | 825 | |
| 804 | 826 |
@action(detail=False, methods=['post'], permission_classes=(DjangoPermission('custom_user.search_user'),))
|
| 805 | 827 |
def synchronization(self, request): |
| ... | ... | |
| 808 | 830 |
response = {'result': 0, 'errors': serializer.errors}
|
| 809 | 831 |
return Response(response, status.HTTP_400_BAD_REQUEST) |
| 810 | 832 |
hooks.call_hooks('api_modify_serializer_after_validation', self, serializer)
|
| 811 |
unknown_uuids = self.check_uuids(serializer.validated_data.get('known_uuids', []))
|
|
| 833 |
known_uuids = serializer.validated_data.get('known_uuids', [])
|
|
| 834 |
timestamp = serializer.validated_data.get('timestamp', None)
|
|
| 835 |
unknown_uuids, modified_users_uuids = self.check_uuids(known_uuids, timestamp) |
|
| 812 | 836 |
data = {
|
| 813 | 837 |
'result': 1, |
| 814 | 838 |
'unknown_uuids': unknown_uuids, |
| 839 |
'modified_users_uuids': modified_users_uuids, |
|
| 815 | 840 |
} |
| 816 | 841 |
hooks.call_hooks('api_modify_response', self, 'synchronization', data)
|
| 817 | 842 |
return Response(data) |
| tests/test_api.py | ||
|---|---|---|
| 1150 | 1150 |
assert set(response.json['unknown_uuids']) == set(unknown_uuids) |
| 1151 | 1151 | |
| 1152 | 1152 | |
| 1153 |
def test_user_synchronization_timestamp(app, admin): |
|
| 1154 |
headers = basic_authorization_header(admin) |
|
| 1155 |
url = reverse('a2-api-users-synchronization')
|
|
| 1156 |
now = datetime.datetime.now() |
|
| 1157 | ||
| 1158 |
ou = get_default_ou() |
|
| 1159 |
User = get_user_model() |
|
| 1160 |
users = [] |
|
| 1161 | ||
| 1162 |
for i in range(6): |
|
| 1163 |
users.append( |
|
| 1164 |
User.objects.create( |
|
| 1165 |
first_name='john%s' % i, |
|
| 1166 |
last_name='doe', |
|
| 1167 |
username='user%s' % i, |
|
| 1168 |
email='user%s' % i, |
|
| 1169 |
ou=ou, |
|
| 1170 |
) |
|
| 1171 |
) |
|
| 1172 | ||
| 1173 |
for i, event_name in enumerate( |
|
| 1174 |
[ |
|
| 1175 |
'manager.user.creation', |
|
| 1176 |
'manager.user.profile.edit', |
|
| 1177 |
'manager.user.activation', |
|
| 1178 |
'manager.user.deactivation', |
|
| 1179 |
'manager.user.password.change.force', |
|
| 1180 |
'manager.user.password.change.unforce', |
|
| 1181 |
] |
|
| 1182 |
): |
|
| 1183 |
event_type = EventType.objects.get_for_name(event_name) |
|
| 1184 |
Event.objects.create( |
|
| 1185 |
type=event_type, |
|
| 1186 |
timestamp=now - datetime.timedelta(days=i, hours=1), |
|
| 1187 |
references=[users[i]], |
|
| 1188 |
) |
|
| 1189 | ||
| 1190 |
content = {
|
|
| 1191 |
'known_uuids': [user.uuid for user in users], |
|
| 1192 |
'timestamp': (now - datetime.timedelta(days=3)).isoformat(), |
|
| 1193 |
} |
|
| 1194 | ||
| 1195 |
response = app.post(url, params=content, headers=headers) |
|
| 1196 | ||
| 1197 |
for user in users[:3]: |
|
| 1198 |
assert user.uuid in response.json['modified_users_uuids'] |
|
| 1199 |
assert user.uuid not in response.json['unknown_uuids'] |
|
| 1200 |
for user in users[3:]: |
|
| 1201 |
assert user.uuid not in response.json['modified_users_uuids'] |
|
| 1202 |
assert user.uuid not in response.json['unknown_uuids'] |
|
| 1203 | ||
| 1204 |
for user in users[:3]: |
|
| 1205 |
user.delete() |
|
| 1206 | ||
| 1207 |
content['timestamp'] = (now - datetime.timedelta(days=7)).isoformat() |
|
| 1208 | ||
| 1209 |
response = app.post(url, params=content, headers=headers) |
|
| 1210 | ||
| 1211 |
for user in users[:3]: |
|
| 1212 |
assert user.uuid not in response.json['modified_users_uuids'] |
|
| 1213 |
assert user.uuid in response.json['unknown_uuids'] |
|
| 1214 |
for user in users[3:]: |
|
| 1215 |
assert user.uuid in response.json['modified_users_uuids'] |
|
| 1216 |
assert user.uuid not in response.json['unknown_uuids'] |
|
| 1217 | ||
| 1218 |
for user in users[3:]: |
|
| 1219 |
user.delete() |
|
| 1220 | ||
| 1221 |
response = app.post(url, params=content, headers=headers) |
|
| 1222 | ||
| 1223 |
assert not response.json['modified_users_uuids'] |
|
| 1224 |
for user in users: |
|
| 1225 |
assert user.uuid in response.json['unknown_uuids'] |
|
| 1226 | ||
| 1227 |
for user in users[:3]: |
|
| 1228 |
content['known_uuids'].remove(user.uuid) |
|
| 1229 | ||
| 1230 |
response = app.post(url, params=content, headers=headers) |
|
| 1231 | ||
| 1232 |
assert not response.json['modified_users_uuids'] |
|
| 1233 |
assert len(response.json['unknown_uuids']) == 3 |
|
| 1234 |
for user in users[3:]: |
|
| 1235 |
assert user.uuid in response.json['unknown_uuids'] |
|
| 1236 | ||
| 1237 | ||
| 1153 | 1238 |
def test_api_drf_authentication_class(app, admin, user_ou1, oidc_client): |
| 1154 | 1239 |
url = '/api/users/%s/' % user_ou1.uuid |
| 1155 | 1240 |
# test invalid client |
| 1156 |
- |
|