authentik.providers.oauth2.tests.test_token

  1"""Test token view"""
  2
  3from base64 import b64encode
  4from datetime import timedelta
  5from json import dumps
  6from urllib.parse import quote
  7
  8from django.test import RequestFactory
  9from django.urls import reverse
 10from django.utils import timezone
 11from django.utils.timezone import now
 12from freezegun import freeze_time
 13
 14from authentik.blueprints.tests import apply_blueprint
 15from authentik.common.oauth.constants import (
 16    GRANT_TYPE_AUTHORIZATION_CODE,
 17    GRANT_TYPE_REFRESH_TOKEN,
 18    TOKEN_TYPE,
 19)
 20from authentik.core.models import Application
 21from authentik.core.tests.utils import create_test_admin_user, create_test_flow
 22from authentik.events.models import Event, EventAction
 23from authentik.lib.generators import generate_id, generate_key
 24from authentik.providers.oauth2.errors import TokenError
 25from authentik.providers.oauth2.models import (
 26    AccessToken,
 27    AuthorizationCode,
 28    GrantType,
 29    OAuth2Provider,
 30    RedirectURI,
 31    RedirectURIMatchingMode,
 32    RefreshToken,
 33    ScopeMapping,
 34)
 35from authentik.providers.oauth2.tests.utils import OAuthTestCase
 36from authentik.providers.oauth2.utils import extract_client_auth
 37from authentik.providers.oauth2.views.token import TokenParams
 38
 39
 40class TestToken(OAuthTestCase):
 41    """Test token view"""
 42
 43    def setUp(self) -> None:
 44        super().setUp()
 45        self.factory = RequestFactory()
 46        self.app = Application.objects.create(name=generate_id(), slug="test")
 47
 48    def test_invalid_grant_type(self):
 49        """test request param"""
 50        provider = OAuth2Provider.objects.create(
 51            name=generate_id(),
 52            authorization_flow=create_test_flow(),
 53            grant_types=[],
 54            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://TestServer")],
 55            signing_key=self.keypair,
 56        )
 57        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
 58        user = create_test_admin_user()
 59        code = AuthorizationCode.objects.create(
 60            code="foobar", provider=provider, user=user, auth_time=timezone.now()
 61        )
 62        request = self.factory.post(
 63            "/",
 64            data={
 65                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
 66                "code": code.code,
 67                "redirect_uri": "http://TestServer",
 68            },
 69            HTTP_AUTHORIZATION=f"Basic {header}",
 70        )
 71        with self.assertRaises(TokenError) as cm:
 72            TokenParams.parse(request, provider, provider.client_id, provider.client_secret)
 73        self.assertEqual(cm.exception.cause, "grant_type_not_configured")
 74
 75    def test_request_auth_code(self):
 76        """test request param"""
 77        provider = OAuth2Provider.objects.create(
 78            name=generate_id(),
 79            authorization_flow=create_test_flow(),
 80            grant_types=[GrantType.AUTHORIZATION_CODE],
 81            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://TestServer")],
 82            signing_key=self.keypair,
 83        )
 84        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
 85        user = create_test_admin_user()
 86        code = AuthorizationCode.objects.create(
 87            code="foobar", provider=provider, user=user, auth_time=timezone.now()
 88        )
 89        request = self.factory.post(
 90            "/",
 91            data={
 92                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
 93                "code": code.code,
 94                "redirect_uri": "http://TestServer",
 95            },
 96            HTTP_AUTHORIZATION=f"Basic {header}",
 97        )
 98        params = TokenParams.parse(request, provider, provider.client_id, provider.client_secret)
 99        self.assertEqual(params.provider, provider)
100        with self.assertRaises(TokenError):
101            TokenParams.parse(request, provider, provider.client_id, generate_key())
102
103    def test_request_auth_code_invalid(self):
104        """test request param"""
105        provider = OAuth2Provider.objects.create(
106            name=generate_id(),
107            authorization_flow=create_test_flow(),
108            grant_types=[GrantType.REFRESH_TOKEN],
109            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
110            signing_key=self.keypair,
111        )
112        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
113        request = self.factory.post(
114            "/",
115            data={
116                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
117                "code": "foo",
118                "redirect_uri": "http://testserver",
119            },
120            HTTP_AUTHORIZATION=f"Basic {header}",
121        )
122        with self.assertRaises(TokenError):
123            TokenParams.parse(request, provider, provider.client_id, provider.client_secret)
124
125    def test_request_refresh_token(self):
126        """test request param"""
127        provider = OAuth2Provider.objects.create(
128            name=generate_id(),
129            authorization_flow=create_test_flow(),
130            grant_types=[GrantType.REFRESH_TOKEN],
131            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
132            signing_key=self.keypair,
133        )
134        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
135        user = create_test_admin_user()
136        token = RefreshToken.objects.create(
137            provider=provider,
138            user=user,
139            token=generate_id(),
140            auth_time=timezone.now(),
141        )
142        request = self.factory.post(
143            "/",
144            data={
145                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
146                "refresh_token": token.token,
147                "redirect_uri": "http://local.invalid",
148            },
149            HTTP_AUTHORIZATION=f"Basic {header}",
150        )
151        params = TokenParams.parse(request, provider, provider.client_id, provider.client_secret)
152        self.assertEqual(params.provider, provider)
153
154    def test_extract_client_auth_basic_auth_percent_decodes(self):
155        """test percent-decoding of client credentials in Basic auth"""
156        header = b64encode(
157            f"{quote('client/id', safe='')}:{quote('secret+/==', safe='')}".encode()
158        ).decode()
159        request = self.factory.post("/", HTTP_AUTHORIZATION=f"Basic {header}")
160        self.assertEqual(extract_client_auth(request), ("client/id", "secret+/=="))
161
162    def test_extract_client_auth_basic_auth_preserves_raw_plus(self):
163        """test compatibility with clients that still send raw plus characters"""
164        header = b64encode(b"client:secret+plus").decode()
165        request = self.factory.post("/", HTTP_AUTHORIZATION=f"Basic {header}")
166        self.assertEqual(extract_client_auth(request), ("client", "secret+plus"))
167
168    def test_auth_code_view(self):
169        """test request param"""
170        provider = OAuth2Provider.objects.create(
171            name=generate_id(),
172            authorization_flow=create_test_flow(),
173            grant_types=[GrantType.AUTHORIZATION_CODE],
174            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
175            signing_key=self.keypair,
176        )
177        # Needs to be assigned to an application for iss to be set
178        self.app.provider = provider
179        self.app.save()
180        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
181        user = create_test_admin_user()
182        code = AuthorizationCode.objects.create(
183            code="foobar", provider=provider, user=user, auth_time=timezone.now()
184        )
185        response = self.client.post(
186            reverse("authentik_providers_oauth2:token"),
187            data={
188                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
189                "code": code.code,
190                "redirect_uri": "http://local.invalid",
191            },
192            HTTP_AUTHORIZATION=f"Basic {header}",
193        )
194        access = AccessToken.objects.filter(user=user, provider=provider).first()
195        self.assertJSONEqual(
196            response.content.decode(),
197            {
198                "access_token": access.token,
199                "token_type": TOKEN_TYPE,
200                "expires_in": 3600,
201                "id_token": provider.encode(
202                    access.id_token.to_dict(),
203                ),
204                "scope": "",
205            },
206        )
207        self.validate_jwt(access, provider)
208
209    def test_auth_code_enc(self):
210        """test request param"""
211        provider = OAuth2Provider.objects.create(
212            name=generate_id(),
213            authorization_flow=create_test_flow(),
214            grant_types=[GrantType.AUTHORIZATION_CODE],
215            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
216            signing_key=self.keypair,
217            encryption_key=self.keypair,
218        )
219        # Needs to be assigned to an application for iss to be set
220        self.app.provider = provider
221        self.app.save()
222        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
223        user = create_test_admin_user()
224        code = AuthorizationCode.objects.create(
225            code="foobar", provider=provider, user=user, auth_time=timezone.now()
226        )
227        response = self.client.post(
228            reverse("authentik_providers_oauth2:token"),
229            data={
230                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
231                "code": code.code,
232                "redirect_uri": "http://local.invalid",
233            },
234            HTTP_AUTHORIZATION=f"Basic {header}",
235        )
236        self.assertEqual(response.status_code, 200)
237        access = AccessToken.objects.filter(user=user, provider=provider).first()
238        self.validate_jwe(access, provider)
239
240    @apply_blueprint("system/providers-oauth2.yaml")
241    def test_refresh_token_view(self):
242        """test request param"""
243        provider = OAuth2Provider.objects.create(
244            name=generate_id(),
245            authorization_flow=create_test_flow(),
246            grant_types=[GrantType.REFRESH_TOKEN],
247            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
248            signing_key=self.keypair,
249        )
250        provider.property_mappings.set(
251            ScopeMapping.objects.filter(
252                managed__in=[
253                    "goauthentik.io/providers/oauth2/scope-openid",
254                    "goauthentik.io/providers/oauth2/scope-email",
255                    "goauthentik.io/providers/oauth2/scope-profile",
256                    "goauthentik.io/providers/oauth2/scope-offline_access",
257                ]
258            )
259        )
260        # Needs to be assigned to an application for iss to be set
261        self.app.provider = provider
262        self.app.save()
263        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
264        user = create_test_admin_user()
265        token = RefreshToken.objects.create(
266            provider=provider,
267            user=user,
268            token=generate_id(),
269            _id_token=dumps({}),
270            auth_time=timezone.now(),
271            _scope="offline_access",
272        )
273        response = self.client.post(
274            reverse("authentik_providers_oauth2:token"),
275            data={
276                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
277                "refresh_token": token.token,
278                "redirect_uri": "http://local.invalid",
279            },
280            HTTP_AUTHORIZATION=f"Basic {header}",
281            HTTP_ORIGIN="http://local.invalid",
282        )
283        self.assertEqual(response["Access-Control-Allow-Credentials"], "true")
284        self.assertEqual(response["Access-Control-Allow-Origin"], "http://local.invalid")
285        access = AccessToken.objects.filter(user=user, provider=provider).first()
286        refresh = RefreshToken.objects.filter(user=user, provider=provider, revoked=False).first()
287        self.assertJSONEqual(
288            response.content.decode(),
289            {
290                "access_token": access.token,
291                "refresh_token": refresh.token,
292                "token_type": TOKEN_TYPE,
293                "expires_in": 3600,
294                "id_token": provider.encode(
295                    access.id_token.to_dict(),
296                ),
297                "scope": "offline_access",
298            },
299        )
300        self.validate_jwt(access, provider)
301
302    @apply_blueprint("system/providers-oauth2.yaml")
303    def test_refresh_token_view_invalid_origin(self):
304        """test request param"""
305        provider = OAuth2Provider.objects.create(
306            name=generate_id(),
307            authorization_flow=create_test_flow(),
308            grant_types=[GrantType.REFRESH_TOKEN],
309            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
310            signing_key=self.keypair,
311        )
312        provider.property_mappings.set(
313            ScopeMapping.objects.filter(
314                managed__in=[
315                    "goauthentik.io/providers/oauth2/scope-openid",
316                    "goauthentik.io/providers/oauth2/scope-email",
317                    "goauthentik.io/providers/oauth2/scope-profile",
318                    "goauthentik.io/providers/oauth2/scope-offline_access",
319                ]
320            )
321        )
322        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
323        user = create_test_admin_user()
324        token = RefreshToken.objects.create(
325            provider=provider,
326            user=user,
327            token=generate_id(),
328            _id_token=dumps({}),
329            auth_time=timezone.now(),
330            _scope="offline_access",
331        )
332        response = self.client.post(
333            reverse("authentik_providers_oauth2:token"),
334            data={
335                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
336                "refresh_token": token.token,
337                "redirect_uri": "http://local.invalid",
338            },
339            HTTP_AUTHORIZATION=f"Basic {header}",
340            HTTP_ORIGIN="http://another.invalid",
341        )
342        access = AccessToken.objects.filter(user=user, provider=provider).first()
343        refresh = RefreshToken.objects.filter(user=user, provider=provider, revoked=False).first()
344        self.assertNotIn("Access-Control-Allow-Credentials", response)
345        self.assertNotIn("Access-Control-Allow-Origin", response)
346        self.assertJSONEqual(
347            response.content.decode(),
348            {
349                "access_token": access.token,
350                "refresh_token": refresh.token,
351                "token_type": TOKEN_TYPE,
352                "expires_in": 3600,
353                "id_token": provider.encode(
354                    access.id_token.to_dict(),
355                ),
356                "scope": "offline_access",
357            },
358        )
359
360    @apply_blueprint("system/providers-oauth2.yaml")
361    def test_refresh_token_revoke(self):
362        """test request param"""
363        provider = OAuth2Provider.objects.create(
364            name=generate_id(),
365            authorization_flow=create_test_flow(),
366            grant_types=[GrantType.REFRESH_TOKEN],
367            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
368            signing_key=self.keypair,
369        )
370        provider.property_mappings.set(
371            ScopeMapping.objects.filter(
372                managed__in=[
373                    "goauthentik.io/providers/oauth2/scope-openid",
374                    "goauthentik.io/providers/oauth2/scope-email",
375                    "goauthentik.io/providers/oauth2/scope-profile",
376                    "goauthentik.io/providers/oauth2/scope-offline_access",
377                ]
378            )
379        )
380        # Needs to be assigned to an application for iss to be set
381        self.app.provider = provider
382        self.app.save()
383        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
384        user = create_test_admin_user()
385        token = RefreshToken.objects.create(
386            provider=provider,
387            user=user,
388            token=generate_id(),
389            _id_token=dumps({}),
390            auth_time=timezone.now(),
391            _scope="offline_access",
392        )
393        # Create initial refresh token
394        response = self.client.post(
395            reverse("authentik_providers_oauth2:token"),
396            data={
397                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
398                "refresh_token": token.token,
399                "redirect_uri": "http://testserver",
400            },
401            HTTP_AUTHORIZATION=f"Basic {header}",
402        )
403        new_token = RefreshToken.objects.filter(user=user).exclude(pk=token.pk).first()
404        # Post again with initial token -> get new refresh token
405        # and revoke old one
406        response = self.client.post(
407            reverse("authentik_providers_oauth2:token"),
408            data={
409                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
410                "refresh_token": new_token.token,
411                "redirect_uri": "http://local.invalid",
412            },
413            HTTP_AUTHORIZATION=f"Basic {header}",
414        )
415        self.assertEqual(response.status_code, 200)
416        # Post again with old token, is now revoked and should error
417        response = self.client.post(
418            reverse("authentik_providers_oauth2:token"),
419            data={
420                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
421                "refresh_token": new_token.token,
422                "redirect_uri": "http://local.invalid",
423            },
424            HTTP_AUTHORIZATION=f"Basic {header}",
425        )
426        self.assertEqual(response.status_code, 400)
427        self.assertTrue(Event.objects.filter(action=EventAction.SUSPICIOUS_REQUEST).exists())
428
429    @apply_blueprint("system/providers-oauth2.yaml")
430    def test_refresh_token_view_threshold(self):
431        """refresh token threshold
432
433        threshold set to 1 hour, refresh token expires in 2 hours.
434        First request should not return a new refresh token, second request
435        has a fake time 1 hours in the future which should return a new access token"""
436        provider = OAuth2Provider.objects.create(
437            name=generate_id(),
438            authorization_flow=create_test_flow(),
439            grant_types=[GrantType.REFRESH_TOKEN],
440            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
441            signing_key=self.keypair,
442            refresh_token_threshold="hours=1",  # nosec
443        )
444        provider.property_mappings.set(
445            ScopeMapping.objects.filter(
446                managed__in=[
447                    "goauthentik.io/providers/oauth2/scope-openid",
448                    "goauthentik.io/providers/oauth2/scope-email",
449                    "goauthentik.io/providers/oauth2/scope-profile",
450                    "goauthentik.io/providers/oauth2/scope-offline_access",
451                ]
452            )
453        )
454        # Needs to be assigned to an application for iss to be set
455        self.app.provider = provider
456        self.app.save()
457        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
458        user = create_test_admin_user()
459        token = RefreshToken.objects.create(
460            provider=provider,
461            user=user,
462            token=generate_id(),
463            _id_token=dumps({}),
464            auth_time=timezone.now(),
465            _scope="offline_access",
466            expires=now() + timedelta(hours=2),
467        )
468        response = self.client.post(
469            reverse("authentik_providers_oauth2:token"),
470            data={
471                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
472                "refresh_token": token.token,
473                "redirect_uri": "http://local.invalid",
474            },
475            HTTP_AUTHORIZATION=f"Basic {header}",
476            HTTP_ORIGIN="http://local.invalid",
477        )
478        access = AccessToken.objects.filter(user=user, provider=provider).first()
479        self.assertJSONEqual(
480            response.content.decode(),
481            {
482                "access_token": access.token,
483                "token_type": TOKEN_TYPE,
484                "expires_in": 3600,
485                "id_token": provider.encode(
486                    access.id_token.to_dict(),
487                ),
488                "scope": "offline_access",
489            },
490        )
491        self.validate_jwt(access, provider)
492
493        with freeze_time(now() + timedelta(hours=1, minutes=10)):
494            response = self.client.post(
495                reverse("authentik_providers_oauth2:token"),
496                data={
497                    "grant_type": GRANT_TYPE_REFRESH_TOKEN,
498                    "refresh_token": token.token,
499                    "redirect_uri": "http://local.invalid",
500                },
501                HTTP_AUTHORIZATION=f"Basic {header}",
502                HTTP_ORIGIN="http://local.invalid",
503            )
504            access = AccessToken.objects.filter(user=user, provider=provider).first()
505            refresh = RefreshToken.objects.filter(user=user, provider=provider).last()
506            self.assertJSONEqual(
507                response.content.decode(),
508                {
509                    "access_token": access.token,
510                    "token_type": TOKEN_TYPE,
511                    "expires_in": 3600,
512                    "id_token": provider.encode(
513                        access.id_token.to_dict(),
514                    ),
515                    "scope": "offline_access",
516                    "refresh_token": refresh.token,
517                },
518            )
519            self.validate_jwt(access, provider)
520
521    @apply_blueprint("system/providers-oauth2.yaml")
522    def test_scope_claim_override_via_property_mapping(self):
523        """Test that property mappings can override the scope claim in access tokens.
524
525        See: https://github.com/goauthentik/authentik/issues/19224
526        """
527        # Create a custom scope mapping that returns a custom scope claim
528        custom_scope_mapping = ScopeMapping.objects.create(
529            name="custom-scope-override",
530            scope_name="custom",
531            expression='return {"scope": "custom-scope-value additional-scope"}',
532        )
533
534        provider = OAuth2Provider.objects.create(
535            name=generate_id(),
536            authorization_flow=create_test_flow(),
537            grant_types=[GrantType.AUTHORIZATION_CODE],
538            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
539            signing_key=self.keypair,
540            include_claims_in_id_token=True,
541        )
542        provider.property_mappings.add(custom_scope_mapping)
543
544        # Needs to be assigned to an application for iss to be set
545        self.app.provider = provider
546        self.app.save()
547
548        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
549        user = create_test_admin_user()
550        code = AuthorizationCode.objects.create(
551            code="foobar",
552            provider=provider,
553            user=user,
554            auth_time=timezone.now(),
555            _scope="openid custom",  # Request the custom scope
556        )
557
558        response = self.client.post(
559            reverse("authentik_providers_oauth2:token"),
560            data={
561                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
562                "code": code.code,
563                "redirect_uri": "http://local.invalid",
564            },
565            HTTP_AUTHORIZATION=f"Basic {header}",
566        )
567        self.assertEqual(response.status_code, 200)
568
569        access = AccessToken.objects.filter(user=user, provider=provider).first()
570        jwt_data = self.validate_jwt(access, provider)
571
572        # The scope should be the custom value from the property mapping,
573        # not the default "openid custom"
574        self.assertEqual(jwt_data["scope"], "custom-scope-value additional-scope")