idp_oidc: add ou selection on ropc grant (#39383)

src/authentic2_idp_oidc/views.py

@@ -42,6 +42,7 @@ from authentic2.utils import (login_require, redirect, timestamp_from_datetime,
                               last_authentication_event, make_url)
 from authentic2.views import logout as a2_logout
 from authentic2 import hooks
+from django_rbac.utils import get_ou_model
 
 from . import app_settings, models, utils
 
@@ -427,6 +428,7 @@ def idtoken_from_user_credential(request):
             'wrong content type. request content type must be \'application/x-www-form-urlencoded\'')
     username = request.POST.get('username')
     scope = request.POST.get('scope')
+    OrganizationalUnit = get_ou_model()
 
     # scope is ignored, we used the configured scope
 
@@ -467,7 +469,15 @@ def idtoken_from_user_credential(request):
             'too many attempts with erroneous RO password, you must wait '
             '%s seconds to try again.' % int(math.ceil(seconds_to_wait)))
 
-    user = authenticate(request, username=username, password=request.POST.get('password'))
+    ou = None
+    if 'ou_slug' in request.POST:
+        try:
+            ou = OrganizationalUnit.objects.get(slug=request.POST.get('ou_slug'))
+        except OrganizationalUnit.DoesNotExist:
+            return invalid_request_response(
+                'ou_slug parameter does not match a valid organization unit')
+
+    user = authenticate(request, username=username, password=request.POST.get('password'), ou=ou)
     if not user:
         exponential_backoff.failure(*backoff_keys)
         return access_denied_response('invalid resource owner credentials')

tests/test_idp_oidc.py

@@ -1362,3 +1362,90 @@ def test_credentials_grant_invalid_content_type(
         status=400)
     assert response.json['error'] == 'invalid_request'
     assert 'wrong content type' in response.json['error_description']
+
+
+def test_credentials_grant_ou_selection_simple(
+        app, oidc_client, admin, user_ou1, user_ou2, ou1, ou2, settings):
+    cache.clear()
+    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
+    oidc_client.save()
+    params = {
+        'client_id': oidc_client.client_id,
+        'client_secret': oidc_client.client_secret,
+        'grant_type': 'password',
+        'ou_slug': ou1.slug,
+        'username': user_ou1.username,
+        'password': user_ou1.username,
+    }
+    token_url = make_url('oidc-token')
+    response = app.post(token_url, params=params)
+
+    params['username'] = user_ou2.username
+    params['password'] = user_ou2.password
+    response = app.post(token_url, params=params, status=400)
+
+
+def test_credentials_grant_ou_selection_username_not_unique(
+        app, oidc_client, admin, user_ou1, admin_ou2, ou1, ou2, settings):
+    cache.clear()
+    settings.A2_USERNAME_IS_UNIQUE = False
+    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
+    oidc_client.save()
+    admin_ou2.username = user_ou1.username
+    admin_ou2.set_password(user_ou1.username)
+    admin_ou2.save()
+    params = {
+        'client_id': oidc_client.client_id,
+        'client_secret': oidc_client.client_secret,
+        'grant_type': 'password',
+        'ou_slug': ou1.slug,
+        'username': user_ou1.username,
+        'password': user_ou1.username,
+    }
+    token_url = make_url('oidc-token')
+    response = app.post(token_url, params=params)
+    assert OIDCAccessToken.objects.get(
+            uuid=response.json['access_token']).user == user_ou1
+
+    params['ou_slug'] = ou2.slug
+    response = app.post(token_url, params=params)
+    assert OIDCAccessToken.objects.get(
+            uuid=response.json['access_token']).user == admin_ou2
+
+
+def test_credentials_grant_ou_selection_username_not_unique_wrong_ou(
+        app, oidc_client, admin, user_ou1, admin_ou2, ou1, ou2, settings):
+    cache.clear()
+    settings.A2_USERNAME_IS_UNIQUE = False
+    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
+    oidc_client.save()
+    params = {
+        'client_id': oidc_client.client_id,
+        'client_secret': oidc_client.client_secret,
+        'grant_type': 'password',
+        'ou_slug': ou2.slug,
+        'username': user_ou1.username,
+        'password': user_ou1.username,
+    }
+    token_url = make_url('oidc-token')
+    response = app.post(token_url, params=params, status=400)
+
+    params['ou_slug'] = ou1.slug
+    params['username'] = admin_ou2.username
+    params['password'] = admin_ou2.password
+    response = app.post(token_url, params=params, status=400)
+
+
+def test_credentials_grant_ou_selection_invalid_ou(
+        app, oidc_client, admin, user_ou1, settings):
+    cache.clear()
+    params = {
+        'client_id': oidc_client.client_id,
+        'client_secret': oidc_client.client_secret,
+        'grant_type': 'password',
+        'ou_slug': 'invalidslug',
+        'username': user_ou1.username,
+        'password': user_ou1.username,
+    }
+    token_url = make_url('oidc-token')
+    response = app.post(token_url, params=params, status=400)