authentik.enterprise.providers.scim.tests

  1"""SCIM OAuth tests"""
  2
  3from base64 import b64encode
  4from datetime import timedelta
  5from unittest.mock import MagicMock, PropertyMock, patch
  6
  7from django.urls import reverse
  8from django.utils.timezone import now
  9from requests_mock import Mocker
 10from rest_framework.test import APITestCase
 11
 12from authentik.blueprints.tests import apply_blueprint
 13from authentik.core.models import Application, Group, User
 14from authentik.core.tests.utils import create_test_admin_user
 15from authentik.enterprise.license import LicenseKey
 16from authentik.enterprise.models import License
 17from authentik.enterprise.tests.test_license import expiry_valid
 18from authentik.lib.generators import generate_id
 19from authentik.providers.scim.models import SCIMAuthenticationMode, SCIMMapping, SCIMProvider
 20from authentik.sources.oauth.models import OAuthSource, UserOAuthSourceConnection
 21from authentik.tenants.models import Tenant
 22
 23
 24class SCIMOAuthTests(APITestCase):
 25    """SCIM User tests"""
 26
 27    @apply_blueprint("system/providers-scim.yaml")
 28    def setUp(self) -> None:
 29        # Delete all users and groups as the mocked HTTP responses only return one ID
 30        # which will cause errors with multiple users
 31        Tenant.objects.update(avatars="none")
 32        User.objects.all().exclude_anonymous().delete()
 33        Group.objects.all().delete()
 34        self.source = OAuthSource.objects.create(
 35            name=generate_id(),
 36            slug=generate_id(),
 37            access_token_url="http://localhost/token",  # nosec
 38            consumer_key=generate_id(),
 39            consumer_secret=generate_id(),
 40            provider_type="openidconnect",
 41        )
 42        self.provider = SCIMProvider.objects.create(
 43            name=generate_id(),
 44            url="https://localhost",
 45            auth_mode=SCIMAuthenticationMode.OAUTH,
 46            auth_oauth=self.source,
 47            auth_oauth_params={
 48                "foo": "bar",
 49            },
 50            exclude_users_service_account=True,
 51        )
 52        self.app: Application = Application.objects.create(
 53            name=generate_id(),
 54            slug=generate_id(),
 55        )
 56        self.app.backchannel_providers.add(self.provider)
 57        self.provider.property_mappings.add(
 58            SCIMMapping.objects.get(managed="goauthentik.io/providers/scim/user")
 59        )
 60        self.provider.property_mappings_group.add(
 61            SCIMMapping.objects.get(managed="goauthentik.io/providers/scim/group")
 62        )
 63
 64    def test_retrieve_token(self):
 65        """Test token retrieval"""
 66        with Mocker() as mocker:
 67            token = generate_id()
 68            mocker.post("http://localhost/token", json={"access_token": token, "expires_in": 3600})
 69            self.provider.scim_auth()
 70        conn = UserOAuthSourceConnection.objects.filter(
 71            source=self.source,
 72            user=self.provider.auth_oauth_user,
 73        ).first()
 74        self.assertIsNotNone(conn)
 75        self.assertTrue(conn.is_valid)
 76        auth = (
 77            b64encode(
 78                b":".join((self.source.consumer_key.encode(), self.source.consumer_secret.encode()))
 79            )
 80            .strip()
 81            .decode()
 82        )
 83        self.assertEqual(
 84            mocker.request_history[0].headers["Authorization"],
 85            f"Basic {auth}",
 86        )
 87        self.assertEqual(mocker.request_history[0].body, "grant_type=password&foo=bar")
 88
 89    def test_existing_token(self):
 90        """Test existing token"""
 91        UserOAuthSourceConnection.objects.create(
 92            source=self.source,
 93            user=self.provider.auth_oauth_user,
 94            access_token=generate_id(),
 95            expires=now() + timedelta(hours=3),
 96        )
 97        with Mocker() as mocker:
 98            self.provider.scim_auth()
 99            self.assertEqual(len(mocker.request_history), 0)
100
101    @Mocker()
102    def test_user_create(self, mock: Mocker):
103        """Test user creation"""
104        scim_id = generate_id()
105        token = generate_id()
106        mock.post("http://localhost/token", json={"access_token": token, "expires_in": 3600})
107        mock.get(
108            "https://localhost/ServiceProviderConfig",
109            json={},
110        )
111        mock.post(
112            "https://localhost/Users",
113            json={
114                "id": scim_id,
115            },
116        )
117        uid = generate_id()
118        user = User.objects.create(
119            username=uid,
120            name=f"{uid} {uid}",
121            email=f"{uid}@goauthentik.io",
122        )
123        self.assertEqual(mock.call_count, 3)
124        self.assertEqual(mock.request_history[1].method, "GET")
125        self.assertEqual(mock.request_history[2].method, "POST")
126        self.assertJSONEqual(
127            mock.request_history[2].body,
128            {
129                "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
130                "active": True,
131                "emails": [
132                    {
133                        "primary": True,
134                        "type": "other",
135                        "value": f"{uid}@goauthentik.io",
136                    }
137                ],
138                "externalId": user.uid,
139                "name": {
140                    "familyName": uid,
141                    "formatted": f"{uid} {uid}",
142                    "givenName": uid,
143                },
144                "displayName": f"{uid} {uid}",
145                "userName": uid,
146            },
147        )
148
149    @patch(
150        "authentik.enterprise.license.LicenseKey.validate",
151        MagicMock(
152            return_value=LicenseKey(
153                aud="",
154                exp=expiry_valid,
155                name=generate_id(),
156                internal_users=100,
157                external_users=100,
158            )
159        ),
160    )
161    def test_api_create(self):
162        License.objects.create(key=generate_id())
163        self.client.force_login(create_test_admin_user())
164        res = self.client.post(
165            reverse("authentik_api:scimprovider-list"),
166            {
167                "name": generate_id(),
168                "url": "http://localhost",
169                "auth_mode": "oauth",
170                "auth_oauth": str(self.source.pk),
171            },
172        )
173        self.assertEqual(res.status_code, 201)
174
175    @patch(
176        "authentik.enterprise.models.LicenseUsageStatus.is_valid",
177        PropertyMock(return_value=False),
178    )
179    def test_api_create_no_license(self):
180        self.client.force_login(create_test_admin_user())
181        res = self.client.post(
182            reverse("authentik_api:scimprovider-list"),
183            {
184                "name": generate_id(),
185                "url": "http://localhost",
186                "auth_mode": "oauth",
187                "auth_oauth": str(self.source.pk),
188            },
189        )
190        self.assertEqual(res.status_code, 400)
191        self.assertJSONEqual(
192            res.content, {"auth_mode": ["Enterprise is required to use the OAuth mode."]}
193        )