authentik.providers.oauth2.tests.test_token_cc_user_pw

  1"""Test token view"""
  2
  3from json import loads
  4
  5from django.test import RequestFactory
  6from django.urls import reverse
  7from jwt import decode
  8
  9from authentik.blueprints.tests import apply_blueprint
 10from authentik.common.oauth.constants import (
 11    GRANT_TYPE_CLIENT_CREDENTIALS,
 12    GRANT_TYPE_PASSWORD,
 13    SCOPE_OPENID,
 14    SCOPE_OPENID_EMAIL,
 15    SCOPE_OPENID_PROFILE,
 16    TOKEN_TYPE,
 17)
 18from authentik.core.models import Application, Group, Token, TokenIntents, UserTypes
 19from authentik.core.tests.utils import (
 20    create_test_admin_user,
 21    create_test_cert,
 22    create_test_flow,
 23    create_test_user,
 24)
 25from authentik.policies.models import PolicyBinding
 26from authentik.providers.oauth2.errors import TokenError
 27from authentik.providers.oauth2.models import (
 28    GrantType,
 29    OAuth2Provider,
 30    RedirectURI,
 31    RedirectURIMatchingMode,
 32    ScopeMapping,
 33)
 34from authentik.providers.oauth2.tests.utils import OAuthTestCase
 35
 36
 37class TestTokenClientCredentialsUserNamePassword(OAuthTestCase):
 38    """Test token (client_credentials) view"""
 39
 40    @apply_blueprint("system/providers-oauth2.yaml")
 41    def setUp(self) -> None:
 42        super().setUp()
 43        self.factory = RequestFactory()
 44        self.provider = OAuth2Provider.objects.create(
 45            name="test",
 46            authorization_flow=create_test_flow(),
 47            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
 48            signing_key=create_test_cert(),
 49            grant_types=[GrantType.CLIENT_CREDENTIALS, GrantType.PASSWORD],
 50        )
 51        self.provider.property_mappings.set(ScopeMapping.objects.all())
 52        self.app = Application.objects.create(name="test", slug="test", provider=self.provider)
 53        self.user = create_test_admin_user("sa")
 54        self.user.type = UserTypes.SERVICE_ACCOUNT
 55        self.user.save()
 56        self.token = Token.objects.create(
 57            identifier="sa-token",
 58            user=self.user,
 59            intent=TokenIntents.INTENT_APP_PASSWORD,
 60            expiring=False,
 61        )
 62
 63    def test_wrong_user(self):
 64        """test invalid username"""
 65        response = self.client.post(
 66            reverse("authentik_providers_oauth2:token"),
 67            {
 68                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
 69                "scope": SCOPE_OPENID,
 70                "client_id": self.provider.client_id,
 71                "username": "saa",
 72                "password": self.token.key,
 73            },
 74        )
 75        self.assertEqual(response.status_code, 400)
 76        self.assertJSONEqual(
 77            response.content.decode(),
 78            {
 79                "error": "invalid_grant",
 80                "error_description": TokenError.errors["invalid_grant"],
 81                "request_id": response.headers["X-authentik-id"],
 82            },
 83        )
 84
 85    def test_wrong_token(self):
 86        """test invalid token"""
 87        response = self.client.post(
 88            reverse("authentik_providers_oauth2:token"),
 89            {
 90                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
 91                "scope": SCOPE_OPENID,
 92                "client_id": self.provider.client_id,
 93                "username": "sa",
 94                "password": self.token.key + "foo",
 95            },
 96        )
 97        self.assertEqual(response.status_code, 400)
 98        self.assertJSONEqual(
 99            response.content.decode(),
100            {
101                "error": "invalid_grant",
102                "error_description": TokenError.errors["invalid_grant"],
103                "request_id": response.headers["X-authentik-id"],
104            },
105        )
106
107    def test_no_provider(self):
108        """test no provider"""
109        self.app.provider = None
110        self.app.save()
111        response = self.client.post(
112            reverse("authentik_providers_oauth2:token"),
113            {
114                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
115                "scope": SCOPE_OPENID,
116                "client_id": self.provider.client_id,
117                "username": "sa",
118                "password": self.token.key,
119            },
120        )
121        self.assertEqual(response.status_code, 400)
122        self.assertJSONEqual(
123            response.content.decode(),
124            {
125                "error": "invalid_grant",
126                "error_description": TokenError.errors["invalid_grant"],
127                "request_id": response.headers["X-authentik-id"],
128            },
129        )
130
131    def test_deactivate(self):
132        """test deactivated user"""
133        self.user.is_active = False
134        self.user.save()
135        response = self.client.post(
136            reverse("authentik_providers_oauth2:token"),
137            {
138                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
139                "scope": SCOPE_OPENID,
140                "client_id": self.provider.client_id,
141                "username": "sa",
142                "password": self.token.key,
143            },
144        )
145        self.assertEqual(response.status_code, 400)
146        self.assertJSONEqual(
147            response.content.decode(),
148            {
149                "error": "invalid_grant",
150                "error_description": TokenError.errors["invalid_grant"],
151                "request_id": response.headers["X-authentik-id"],
152            },
153        )
154
155    def test_permission_denied(self):
156        """test permission denied"""
157        group = Group.objects.create(name="foo")
158        PolicyBinding.objects.create(
159            group=group,
160            target=self.app,
161            order=0,
162        )
163        response = self.client.post(
164            reverse("authentik_providers_oauth2:token"),
165            {
166                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
167                "scope": SCOPE_OPENID,
168                "client_id": self.provider.client_id,
169                "username": "sa",
170                "password": self.token.key,
171            },
172        )
173        self.assertEqual(response.status_code, 400)
174        self.assertJSONEqual(
175            response.content.decode(),
176            {
177                "error": "invalid_grant",
178                "error_description": TokenError.errors["invalid_grant"],
179                "request_id": response.headers["X-authentik-id"],
180            },
181        )
182
183    def test_successful(self):
184        """test successful"""
185        response = self.client.post(
186            reverse("authentik_providers_oauth2:token"),
187            {
188                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
189                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
190                "client_id": self.provider.client_id,
191                "username": "sa",
192                "password": self.token.key,
193            },
194        )
195        self.assertEqual(response.status_code, 200)
196        body = loads(response.content.decode())
197        self.assertEqual(body["token_type"], TOKEN_TYPE)
198        _, alg = self.provider.jwt_key
199        jwt = decode(
200            body["access_token"],
201            key=self.provider.signing_key.public_key,
202            algorithms=[alg],
203            audience=self.provider.client_id,
204        )
205        self.assertEqual(jwt["given_name"], self.user.name)
206        self.assertEqual(jwt["preferred_username"], self.user.username)
207        jwt = decode(
208            body["id_token"],
209            key=self.provider.signing_key.public_key,
210            algorithms=[alg],
211            audience=self.provider.client_id,
212        )
213        self.assertEqual(jwt["given_name"], self.user.name)
214        self.assertEqual(jwt["preferred_username"], self.user.username)
215
216    def test_successful_two_tokens(self):
217        """test successful when two app passwords with the same key exist"""
218        Token.objects.create(
219            identifier="sa-token-two",
220            user=create_test_user(),
221            intent=TokenIntents.INTENT_APP_PASSWORD,
222            expiring=False,
223            key=self.token.key,
224        )
225
226        response = self.client.post(
227            reverse("authentik_providers_oauth2:token"),
228            {
229                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
230                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
231                "client_id": self.provider.client_id,
232                "username": "sa",
233                "password": self.token.key,
234            },
235        )
236        self.assertEqual(response.status_code, 200)
237        body = loads(response.content.decode())
238        self.assertEqual(body["token_type"], TOKEN_TYPE)
239        _, alg = self.provider.jwt_key
240        jwt = decode(
241            body["access_token"],
242            key=self.provider.signing_key.public_key,
243            algorithms=[alg],
244            audience=self.provider.client_id,
245        )
246        self.assertEqual(jwt["given_name"], self.user.name)
247        self.assertEqual(jwt["preferred_username"], self.user.username)
248        jwt = decode(
249            body["id_token"],
250            key=self.provider.signing_key.public_key,
251            algorithms=[alg],
252            audience=self.provider.client_id,
253        )
254        self.assertEqual(jwt["given_name"], self.user.name)
255        self.assertEqual(jwt["preferred_username"], self.user.username)
256
257    def test_successful_password(self):
258        """test successful (password grant)"""
259        response = self.client.post(
260            reverse("authentik_providers_oauth2:token"),
261            {
262                "grant_type": GRANT_TYPE_PASSWORD,
263                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
264                "client_id": self.provider.client_id,
265                "username": "sa",
266                "password": self.token.key,
267            },
268        )
269        self.assertEqual(response.status_code, 200)
270        body = loads(response.content.decode())
271        self.assertEqual(body["token_type"], TOKEN_TYPE)
272        _, alg = self.provider.jwt_key
273        jwt = decode(
274            body["access_token"],
275            key=self.provider.signing_key.public_key,
276            algorithms=[alg],
277            audience=self.provider.client_id,
278        )
279        self.assertEqual(jwt["given_name"], self.user.name)
280        self.assertEqual(jwt["preferred_username"], self.user.username)