0001-add-get-or-create-users-api-call-22376.patch
| src/authentic2/api_views.py | ||
|---|---|---|
| 571 | 571 |
def put(self, request, *args, **kwargs): |
| 572 | 572 |
return self.patch(request, *args, **kwargs) |
| 573 | 573 | |
| 574 |
def list(self, request, *args, **kwargs): |
|
| 575 |
get_or_create_kwargs = utils.user_get_or_create_kwargs(request) |
|
| 576 |
if get_or_create_kwargs: |
|
| 577 |
payload, status = utils.user_get_or_create(request, get_or_create_kwargs) |
|
| 578 |
return Response(payload, status) |
|
| 579 |
return super(UsersAPI, self).list(request, *args, **kwargs) |
|
| 580 | ||
| 574 | 581 |
def check_perm(self, perm, ou): |
| 575 | 582 |
if ou: |
| 576 | 583 |
if not self.request.user.has_ou_perm(perm, ou): |
| src/authentic2/app_settings.py | ||
|---|---|---|
| 206 | 206 |
A2_ACCOUNTS_URL=Setting(default=None, definition='IdP has no account page, redirect to this one.'), |
| 207 | 207 |
A2_CACHE_ENABLED=Setting(default=True, definition='Disable all cache decorators for testing purpose.'), |
| 208 | 208 |
A2_ACCEPT_EMAIL_AUTHENTICATION=Setting(default=True, definition='Enable authentication by email'), |
| 209 |
A2_USERS_API_GET_OR_CREATE_WHITELIST_FIELDS=Setting(default=['email', 'username', 'ou__slug'], definition='Whitelist fields for "get or create" operations in the users API.'), |
|
| 209 | 210 | |
| 210 | 211 |
) |
| 211 | 212 | |
| src/authentic2/utils.py | ||
|---|---|---|
| 16 | 16 | |
| 17 | 17 |
from django.conf import settings |
| 18 | 18 |
from django.http import HttpResponseRedirect, HttpResponse |
| 19 |
from django.core.exceptions import ImproperlyConfigured, PermissionDenied |
|
| 19 |
from django.core.exceptions import (ImproperlyConfigured, PermissionDenied, |
|
| 20 |
ObjectDoesNotExist, MultipleObjectsReturned) |
|
| 20 | 21 |
from django.http.request import QueryDict |
| 21 | 22 |
from django.contrib.auth import (REDIRECT_FIELD_NAME, login as auth_login, SESSION_KEY, |
| 22 | 23 |
HASH_SESSION_KEY, BACKEND_SESSION_KEY, authenticate, |
| ... | ... | |
| 1073 | 1074 |
if ou_value is not None: |
| 1074 | 1075 |
return ou_value |
| 1075 | 1076 |
return default |
| 1077 | ||
| 1078 | ||
| 1079 |
def user_get_or_create(request, get_or_create_kwargs): |
|
| 1080 |
from django_rbac.utils import get_ou_model |
|
| 1081 | ||
| 1082 |
OrganizationalUnit = get_ou_model() |
|
| 1083 |
ou_slug = get_or_create_kwargs.get('ou__slug')
|
|
| 1084 | ||
| 1085 |
try : |
|
| 1086 |
ou = OrganizationalUnit.objects.get(slug=ou_slug) |
|
| 1087 |
except (ObjectDoesNotExist, MultipleObjectsReturned) as e: |
|
| 1088 |
return ({'reason': 'OrganizationalUnit %s: %s' % (ou_slug, e)}, 400)
|
|
| 1089 | ||
| 1090 |
if not request.user.has_ou_perm('custom_user.add_user', ou):
|
|
| 1091 |
return ({'reason': 'Permission denied'}, 403)
|
|
| 1092 | ||
| 1093 |
get_or_create_kwargs.update({'ou': ou})
|
|
| 1094 |
try: |
|
| 1095 |
user, created = get_user_model().objects.get_or_create(**get_or_create_kwargs) |
|
| 1096 |
except MultipleObjectsReturned as e: |
|
| 1097 |
return ({'reason': 'Multiple objects returned'}, 400)
|
|
| 1098 | ||
| 1099 |
return (user.to_json(), 200+created) |
|
| 1100 | ||
| 1101 | ||
| 1102 |
def user_get_or_create_kwargs(request): |
|
| 1103 |
from authentic2.a2_rbac.utils import get_default_ou |
|
| 1104 | ||
| 1105 |
kwargs_dict = dict() |
|
| 1106 |
for key, value in request.GET.items(): |
|
| 1107 |
if key.startswith('get_or_create_'):
|
|
| 1108 |
fieldkey = key.split('get_or_create_')[-1]
|
|
| 1109 |
if fieldkey in app_settings.A2_USERS_API_GET_OR_CREATE_WHITELIST_FIELDS: |
|
| 1110 |
kwargs_dict.update({fieldkey: value})
|
|
| 1111 |
# provide default values for missing fields when possible: |
|
| 1112 |
if kwargs_dict and not kwargs_dict.get('ou__slug'):
|
|
| 1113 |
kwargs_dict['ou__slug'] = get_default_ou().slug |
|
| 1114 |
return kwargs_dict |
|
| tests/test_api.py | ||
|---|---|---|
| 934 | 934 |
assert response.json['checks'][3]['result'] is True |
| 935 | 935 |
assert response.json['checks'][4]['label'] == 'must contain "ok"' |
| 936 | 936 |
assert response.json['checks'][4]['result'] is True |
| 937 | ||
| 938 | ||
| 939 |
def test_api_users_get_or_create(app, superuser, simple_user, ou1): |
|
| 940 |
User = get_user_model() |
|
| 941 |
app.authorization = ('Basic', (superuser.username, superuser.username))
|
|
| 942 |
url = u'/api/users/?get_or_create_email=john.doe@nowhere.null&' \ |
|
| 943 |
'get_or_create_username=jdoe&get_or_create_ou__slug=%s' % ou1.slug |
|
| 944 | ||
| 945 |
assert not len(User.objects.filter(email='john.doe@nowhere.null', |
|
| 946 |
username='jdoe', ou__slug=ou1.slug)) |
|
| 947 | ||
| 948 |
response = app.get(url, status=201) # created |
|
| 949 |
assert len(User.objects.filter(email='john.doe@nowhere.null', |
|
| 950 |
username='jdoe', ou__slug=ou1.slug)) == 1 |
|
| 951 | ||
| 952 |
john = User.objects.get(email='john.doe@nowhere.null') |
|
| 953 |
uuid = john.uuid |
|
| 954 |
response = app.get(url, status=200) # not created |
|
| 955 |
assert len(User.objects.filter(email='john.doe@nowhere.null', |
|
| 956 |
username='jdoe', ou__slug=ou1.slug)) == 1 |
|
| 957 |
assert uuid == User.objects.get(email='john.doe@nowhere.null').uuid |
|
| 958 | ||
| 959 |
wrong_ou_url = u'/api/users/?get_or_create_email=john.doe@nowhere.null&' \ |
|
| 960 |
'get_or_create_username=jdoe&get_or_create_ou__slug=wrongwrongwrong' |
|
| 961 |
response = app.get(wrong_ou_url, status=400) # bad request |
|
| 962 | ||
| 963 |
app.authorization = ('Basic', (superuser.username, superuser.username+'nope'))
|
|
| 964 |
response = app.get(url, status=401) # unauthorized |
|
| 965 | ||
| 966 |
app.authorization = ('Basic', (simple_user.username, simple_user.username))
|
|
| 967 |
response = app.get(url, status=403) # permission denied |
|
| 937 |
- |
|