authentik.providers.scim.tests.test_client

  1"""SCIM Client tests"""
  2
  3from django.core.cache import cache
  4from django.test import TestCase
  5from requests_mock import Mocker
  6
  7from authentik.blueprints.tests import apply_blueprint
  8from authentik.core.models import Application
  9from authentik.lib.generators import generate_id
 10from authentik.providers.scim.clients.base import SCIMClient
 11from authentik.providers.scim.models import SCIMMapping, SCIMProvider
 12from authentik.providers.scim.tasks import scim_sync
 13
 14
 15class SCIMClientTests(TestCase):
 16    """SCIM Client tests"""
 17
 18    @apply_blueprint("system/providers-scim.yaml")
 19    def setUp(self) -> None:
 20        # Clear cache before each test
 21        cache.clear()
 22        self.provider: SCIMProvider = SCIMProvider.objects.create(
 23            name=generate_id(),
 24            url="https://localhost",
 25            token=generate_id(),
 26        )
 27        self.app: Application = Application.objects.create(
 28            name=generate_id(),
 29            slug=generate_id(),
 30        )
 31        self.app.backchannel_providers.add(self.provider)
 32        self.provider.property_mappings.add(
 33            SCIMMapping.objects.get(managed="goauthentik.io/providers/scim/user")
 34        )
 35        self.provider.property_mappings_group.add(
 36            SCIMMapping.objects.get(managed="goauthentik.io/providers/scim/group")
 37        )
 38
 39    def test_config(self):
 40        """Test valid config:
 41        https://docs.aws.amazon.com/singlesignon/latest/developerguide/serviceproviderconfig.html"""
 42        with Mocker() as mock:
 43            mock: Mocker
 44            mock.get(
 45                "https://localhost/ServiceProviderConfig",
 46                json={
 47                    "schemas": ["urn:ietf:params:scim:schemas:core:2.0:ServiceProviderConfig"],
 48                    "documentationUri": (
 49                        "https://docs.aws.amazon.com/singlesignon/latest/"
 50                        "userguide/manage-your-identity-source-idp.html"
 51                    ),
 52                    "authenticationSchemes": [
 53                        {
 54                            "type": "oauthbearertoken",
 55                            "name": "OAuth Bearer Token",
 56                            "description": (
 57                                "Authentication scheme using the OAuth Bearer Token Standard"
 58                            ),
 59                            "specUri": "https://www.rfc-editor.org/info/rfc6750",
 60                            "documentationUri": (
 61                                "https://docs.aws.amazon.com/singlesignon/latest/"
 62                                "userguide/provision-automatically.html"
 63                            ),
 64                            "primary": True,
 65                        }
 66                    ],
 67                    "patch": {"supported": True},
 68                    "bulk": {"supported": False, "maxOperations": 1, "maxPayloadSize": 1048576},
 69                    "filter": {"supported": True, "maxResults": 50},
 70                    "changePassword": {"supported": False},
 71                    "sort": {"supported": False},
 72                    "etag": {"supported": False},
 73                },
 74            )
 75            SCIMClient(self.provider)
 76            self.assertEqual(mock.call_count, 1)
 77            self.assertEqual(mock.request_history[0].method, "GET")
 78
 79    def test_config_invalid(self):
 80        """Test invalid config"""
 81        with Mocker() as mock:
 82            mock: Mocker
 83            mock.get(
 84                "https://localhost/ServiceProviderConfig",
 85                json={},
 86            )
 87            SCIMClient(self.provider)
 88            self.assertEqual(mock.call_count, 1)
 89            self.assertEqual(mock.request_history[0].method, "GET")
 90
 91    def test_scim_sync(self):
 92        """test scim_sync task"""
 93        scim_sync.send(self.provider.pk).get_result()
 94
 95    def test_config_caching(self):
 96        """Test that ServiceProviderConfig is cached after first successful fetch"""
 97        config_json = {
 98            "schemas": ["urn:ietf:params:scim:schemas:core:2.0:ServiceProviderConfig"],
 99            "documentationUri": "https://example.com/docs",
100            "authenticationSchemes": [
101                {
102                    "type": "oauthbearertoken",
103                    "name": "OAuth Bearer Token",
104                    "description": "OAuth Bearer Token",
105                    "specUri": "https://www.rfc-editor.org/info/rfc6750",
106                    "documentationUri": "https://example.com/auth",
107                    "primary": True,
108                }
109            ],
110            "patch": {"supported": True},
111            "bulk": {"supported": False, "maxOperations": 1, "maxPayloadSize": 1048576},
112            "filter": {"supported": True, "maxResults": 50},
113            "changePassword": {"supported": False},
114            "sort": {"supported": False},
115            "etag": {"supported": False},
116        }
117
118        with Mocker() as mock:
119            mock.get("https://localhost/ServiceProviderConfig", json=config_json)
120
121            client = SCIMClient(self.provider)
122
123            # First call should hit the API
124            config1 = client.get_service_provider_config()
125            self.assertEqual(mock.call_count, 1)
126
127            # Second call should use cache, no additional API call
128            config2 = client.get_service_provider_config()
129            self.assertEqual(mock.call_count, 1)  # Still 1, not 2
130
131            # Verify both configs are the same
132            self.assertEqual(config1, config2)
133
134    def test_config_caching_invalid(self):
135        """Test that default config is cached when remote config is invalid"""
136        with Mocker() as mock:
137            mock.get("https://localhost/ServiceProviderConfig", json={})
138
139            client = SCIMClient(self.provider)
140
141            # First call should hit the API and get invalid response
142            config1 = client.get_service_provider_config()
143            self.assertEqual(mock.call_count, 1)
144
145            # Second call should use cached default config, no additional API call
146            config2 = client.get_service_provider_config()
147            self.assertEqual(mock.call_count, 1)  # Still 1, not 2
148
149            # Verify both configs are the same default
150            self.assertEqual(config1, config2)
151
152    def test_config_caching_error(self):
153        """Test that default config is cached when remote request fails"""
154        with Mocker() as mock:
155            mock.get("https://localhost/ServiceProviderConfig", status_code=500)
156
157            client = SCIMClient(self.provider)
158
159            # First call should hit the API and fail
160            config1 = client.get_service_provider_config()
161            self.assertEqual(mock.call_count, 1)
162
163            # Second call should use cached default config, no additional API call
164            config2 = client.get_service_provider_config()
165            self.assertEqual(mock.call_count, 1)  # Still 1, not 2
166
167            # Verify both configs are the same default
168            self.assertEqual(config1, config2)
169
170    def test_config_cache_invalidation_on_save(self):
171        """Test that ServiceProviderConfig cache is cleared when provider is saved"""
172        config_json = {
173            "schemas": ["urn:ietf:params:scim:schemas:core:2.0:ServiceProviderConfig"],
174            "documentationUri": "https://example.com/docs",
175            "authenticationSchemes": [
176                {
177                    "type": "oauthbearertoken",
178                    "name": "OAuth Bearer Token",
179                    "description": "OAuth Bearer Token",
180                    "specUri": "https://www.rfc-editor.org/info/rfc6750",
181                    "documentationUri": "https://example.com/auth",
182                    "primary": True,
183                }
184            ],
185            "patch": {"supported": True},
186            "bulk": {"supported": False, "maxOperations": 1, "maxPayloadSize": 1048576},
187            "filter": {"supported": True, "maxResults": 50},
188            "changePassword": {"supported": False},
189            "sort": {"supported": False},
190            "etag": {"supported": False},
191        }
192
193        with Mocker() as mock:
194            mock.get("https://localhost/ServiceProviderConfig", json=config_json)
195
196            # First client instantiation caches the config
197            SCIMClient(self.provider)
198            self.assertEqual(mock.call_count, 1)
199
200            # Creating another client should use cached config
201            SCIMClient(self.provider)
202            self.assertEqual(mock.call_count, 1)  # Still 1, using cache
203
204            # Save the provider (e.g., changing compatibility mode)
205            self.provider.compatibility_mode = "aws"
206            self.provider.save()
207
208            # Creating a new client should now hit the API again since cache was cleared
209            SCIMClient(self.provider)
210            self.assertEqual(mock.call_count, 2)  # New API call after cache invalidation