authentik.providers.oauth2.tests.test_token_cc_jwt_source

  1"""Test token view"""
  2
  3from datetime import datetime, timedelta
  4from json import loads
  5
  6from django.test import RequestFactory
  7from django.urls import reverse
  8from jwt import decode
  9
 10from authentik.blueprints.tests import apply_blueprint
 11from authentik.common.oauth.constants import (
 12    GRANT_TYPE_CLIENT_CREDENTIALS,
 13    SCOPE_OPENID,
 14    SCOPE_OPENID_EMAIL,
 15    SCOPE_OPENID_PROFILE,
 16    TOKEN_TYPE,
 17)
 18from authentik.core.models import USERNAME_MAX_LENGTH, Application, Group, User
 19from authentik.core.tests.utils import create_test_cert, create_test_flow
 20from authentik.lib.generators import generate_id
 21from authentik.policies.models import PolicyBinding
 22from authentik.providers.oauth2.models import (
 23    GrantType,
 24    OAuth2Provider,
 25    RedirectURI,
 26    RedirectURIMatchingMode,
 27    ScopeMapping,
 28)
 29from authentik.providers.oauth2.tests.utils import OAuthTestCase
 30from authentik.providers.oauth2.views.jwks import JWKSView
 31from authentik.sources.oauth.models import OAuthSource, OAuthSourcePropertyMapping
 32
 33
 34class TestTokenClientCredentialsJWTSource(OAuthTestCase):
 35    """Test token (client_credentials, with JWT) view"""
 36
 37    @apply_blueprint("system/providers-oauth2.yaml")
 38    def setUp(self) -> None:
 39        super().setUp()
 40        self.factory = RequestFactory()
 41        self.other_cert = create_test_cert()
 42        # Provider used as a helper to sign JWTs with the same key as the OAuth source has
 43        self.helper_provider = OAuth2Provider.objects.create(
 44            name=generate_id(),
 45            authorization_flow=create_test_flow(),
 46            signing_key=self.other_cert,
 47        )
 48        self.cert = create_test_cert()
 49
 50        jwk = JWKSView().get_jwk_for_key(self.other_cert, "sig")
 51        self.source: OAuthSource = OAuthSource.objects.create(
 52            name=generate_id(),
 53            slug=generate_id(),
 54            provider_type="openidconnect",
 55            consumer_key=generate_id(),
 56            consumer_secret=generate_id(),
 57            authorization_url="http://foo",
 58            access_token_url=f"http://{generate_id()}",
 59            profile_url="http://foo",
 60            oidc_well_known_url="",
 61            oidc_jwks_url="",
 62            oidc_jwks={
 63                "keys": [jwk],
 64            },
 65        )
 66
 67        self.provider: OAuth2Provider = OAuth2Provider.objects.create(
 68            name="test",
 69            authorization_flow=create_test_flow(),
 70            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
 71            signing_key=self.cert,
 72            grant_types=[GrantType.CLIENT_CREDENTIALS],
 73        )
 74        self.provider.jwt_federation_sources.add(self.source)
 75        self.provider.property_mappings.set(ScopeMapping.objects.all())
 76        self.app = Application.objects.create(name="test", slug="test", provider=self.provider)
 77
 78    def test_invalid_type(self):
 79        """test invalid type"""
 80        response = self.client.post(
 81            reverse("authentik_providers_oauth2:token"),
 82            {
 83                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
 84                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
 85                "client_id": self.provider.client_id,
 86                "client_assertion_type": "foo",
 87                "client_assertion": "foo.bar",
 88            },
 89        )
 90        self.assertEqual(response.status_code, 400)
 91        body = loads(response.content.decode())
 92        self.assertEqual(body["error"], "invalid_grant")
 93
 94    def test_invalid_jwt(self):
 95        """test invalid JWT"""
 96        response = self.client.post(
 97            reverse("authentik_providers_oauth2:token"),
 98            {
 99                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
100                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
101                "client_id": self.provider.client_id,
102                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
103                "client_assertion": "foo.bar",
104            },
105        )
106        self.assertEqual(response.status_code, 400)
107        body = loads(response.content.decode())
108        self.assertEqual(body["error"], "invalid_grant")
109
110    def test_invalid_signature(self):
111        """test invalid JWT"""
112        token = self.helper_provider.encode(
113            {
114                "sub": "foo",
115                "exp": datetime.now() + timedelta(hours=2),
116            }
117        )
118        response = self.client.post(
119            reverse("authentik_providers_oauth2:token"),
120            {
121                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
122                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
123                "client_id": self.provider.client_id,
124                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
125                "client_assertion": token + "foo",
126            },
127        )
128        self.assertEqual(response.status_code, 400)
129        body = loads(response.content.decode())
130        self.assertEqual(body["error"], "invalid_grant")
131
132    def test_invalid_expired(self):
133        """test invalid JWT"""
134        token = self.helper_provider.encode(
135            {
136                "sub": "foo",
137                "exp": datetime.now() - timedelta(hours=2),
138            }
139        )
140        response = self.client.post(
141            reverse("authentik_providers_oauth2:token"),
142            {
143                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
144                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
145                "client_id": self.provider.client_id,
146                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
147                "client_assertion": token,
148            },
149        )
150        self.assertEqual(response.status_code, 400)
151        body = loads(response.content.decode())
152        self.assertEqual(body["error"], "invalid_grant")
153
154    def test_invalid_no_app(self):
155        """test invalid JWT"""
156        self.app.provider = None
157        self.app.save()
158        token = self.helper_provider.encode(
159            {
160                "sub": "foo",
161                "exp": datetime.now() + timedelta(hours=2),
162            }
163        )
164        response = self.client.post(
165            reverse("authentik_providers_oauth2:token"),
166            {
167                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
168                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
169                "client_id": self.provider.client_id,
170                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
171                "client_assertion": token,
172            },
173        )
174        self.assertEqual(response.status_code, 400)
175        body = loads(response.content.decode())
176        self.assertEqual(body["error"], "invalid_grant")
177
178    def test_invalid_access_denied(self):
179        """test invalid JWT"""
180        group = Group.objects.create(name="foo")
181        PolicyBinding.objects.create(
182            group=group,
183            target=self.app,
184            order=0,
185        )
186        token = self.helper_provider.encode(
187            {
188                "sub": "foo",
189                "exp": datetime.now() + timedelta(hours=2),
190            }
191        )
192        response = self.client.post(
193            reverse("authentik_providers_oauth2:token"),
194            {
195                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
196                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
197                "client_id": self.provider.client_id,
198                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
199                "client_assertion": token,
200            },
201        )
202        self.assertEqual(response.status_code, 400)
203        body = loads(response.content.decode())
204        self.assertEqual(body["error"], "invalid_grant")
205
206    def test_successful(self):
207        """test successful"""
208        token = self.helper_provider.encode(
209            {
210                "sub": "foo",
211                "exp": datetime.now() + timedelta(hours=2),
212            }
213        )
214        response = self.client.post(
215            reverse("authentik_providers_oauth2:token"),
216            {
217                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
218                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
219                "client_id": self.provider.client_id,
220                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
221                "client_assertion": token,
222            },
223        )
224        self.assertEqual(response.status_code, 200)
225
226        user = User.objects.filter(username=f"{self.provider.name}-foo").first()
227        self.assertIsNotNone(user)
228
229        body = loads(response.content.decode())
230        self.assertEqual(body["token_type"], TOKEN_TYPE)
231        _, alg = self.provider.jwt_key
232        jwt = decode(
233            body["access_token"],
234            key=self.provider.signing_key.public_key,
235            algorithms=[alg],
236            audience=self.provider.client_id,
237        )
238        self.assertEqual(
239            jwt["given_name"], "Autogenerated user from application test (client credentials JWT)"
240        )
241        self.assertEqual(jwt["preferred_username"], "test-foo")
242
243    def test_successful_mapping(self):
244        """test successful"""
245        test_username = ("mapped-foo" + ("a" * 150))[:USERNAME_MAX_LENGTH]
246        mapping = OAuthSourcePropertyMapping.objects.create(
247            name="test-mapping",
248            expression="""return {
249                "email": oauth_userinfo.get("email"),
250                "name": oauth_userinfo.get("name"),
251                "username": oauth_userinfo.get("username"),
252            }""",
253        )
254        self.source.user_property_mappings.add(mapping)
255
256        token = self.helper_provider.encode(
257            {
258                "sub": "foo",
259                "email": "test-user@example.com",
260                "name": "Mapped Test User",
261                "username": "mapped-foo" + ("a" * 150),
262                "exp": datetime.now() + timedelta(hours=2),
263            }
264        )
265        response = self.client.post(
266            reverse("authentik_providers_oauth2:token"),
267            {
268                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
269                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
270                "client_id": self.provider.client_id,
271                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
272                "client_assertion": token,
273            },
274        )
275        self.assertEqual(response.status_code, 200)
276
277        user = User.objects.filter(username=test_username).first()
278        self.assertIsNotNone(user)
279
280        body = loads(response.content.decode())
281        self.assertEqual(body["token_type"], TOKEN_TYPE)
282        key_obj, alg = self.provider.jwt_key
283        jwt = decode(
284            body["access_token"],
285            key=key_obj.public_key(),
286            algorithms=[alg],
287            audience=self.provider.client_id,
288        )
289
290        self.assertEqual(jwt["email"], "test-user@example.com")
291        self.assertEqual(jwt["given_name"], "Mapped Test User")
292        self.assertEqual(jwt["preferred_username"], test_username)