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