authentik.providers.oauth2.tests.test_introspect

Test introspect view

View Source

  1"""Test introspect view"""
  2
  3import json
  4from base64 import b64encode
  5from dataclasses import asdict
  6
  7from django.urls import reverse
  8from django.utils import timezone
  9
 10from authentik.common.oauth.constants import ACR_AUTHENTIK_DEFAULT
 11from authentik.core.models import Application
 12from authentik.core.tests.utils import create_test_admin_user, create_test_cert, create_test_flow
 13from authentik.lib.generators import generate_id
 14from authentik.providers.oauth2.id_token import IDToken
 15from authentik.providers.oauth2.models import (
 16    AccessToken,
 17    ClientType,
 18    OAuth2Provider,
 19    RedirectURI,
 20    RedirectURIMatchingMode,
 21    RefreshToken,
 22)
 23from authentik.providers.oauth2.tests.utils import OAuthTestCase
 24
 25
 26class TesOAuth2Introspection(OAuthTestCase):
 27    """Test introspect view"""
 28
 29    def setUp(self) -> None:
 30        super().setUp()
 31        self.provider: OAuth2Provider = OAuth2Provider.objects.create(
 32            name=generate_id(),
 33            authorization_flow=create_test_flow(),
 34            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
 35            signing_key=create_test_cert(),
 36        )
 37        self.app = Application.objects.create(
 38            name=generate_id(), slug=generate_id(), provider=self.provider
 39        )
 40        self.user = create_test_admin_user()
 41        self.auth = b64encode(
 42            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
 43        ).decode()
 44
 45    def test_introspect_refresh(self):
 46        """Test introspect"""
 47        token = RefreshToken.objects.create(
 48            provider=self.provider,
 49            user=self.user,
 50            token=generate_id(),
 51            auth_time=timezone.now(),
 52            _scope="openid user profile",
 53            _id_token=json.dumps(
 54                asdict(
 55                    IDToken("foo", "bar"),
 56                )
 57            ),
 58        )
 59        res = self.client.post(
 60            reverse("authentik_providers_oauth2:token-introspection"),
 61            HTTP_AUTHORIZATION=f"Basic {self.auth}",
 62            data={"token": token.token},
 63        )
 64        self.assertEqual(res.status_code, 200)
 65        self.assertJSONEqual(
 66            res.content.decode(),
 67            {
 68                "acr": ACR_AUTHENTIK_DEFAULT,
 69                "sub": "bar",
 70                "iss": "foo",
 71                "active": True,
 72                "client_id": self.provider.client_id,
 73                "scope": " ".join(token.scope),
 74            },
 75        )
 76
 77    def test_introspect_access(self):
 78        """Test introspect"""
 79        token = AccessToken.objects.create(
 80            provider=self.provider,
 81            user=self.user,
 82            token=generate_id(),
 83            auth_time=timezone.now(),
 84            _scope="openid user profile",
 85            _id_token=json.dumps(
 86                asdict(
 87                    IDToken("foo", "bar"),
 88                )
 89            ),
 90        )
 91        res = self.client.post(
 92            reverse("authentik_providers_oauth2:token-introspection"),
 93            HTTP_AUTHORIZATION=f"Basic {self.auth}",
 94            data={"token": token.token},
 95        )
 96        self.assertEqual(res.status_code, 200)
 97        self.assertJSONEqual(
 98            res.content.decode(),
 99            {
100                "acr": ACR_AUTHENTIK_DEFAULT,
101                "sub": "bar",
102                "iss": "foo",
103                "active": True,
104                "client_id": self.provider.client_id,
105                "scope": " ".join(token.scope),
106            },
107        )
108
109    def test_introspect_invalid_token(self):
110        """Test introspect (invalid token)"""
111        res = self.client.post(
112            reverse("authentik_providers_oauth2:token-introspection"),
113            HTTP_AUTHORIZATION=f"Basic {self.auth}",
114            data={"token": generate_id(), "token_type_hint": "refresh_token"},
115        )
116        self.assertEqual(res.status_code, 200)
117        self.assertJSONEqual(
118            res.content.decode(),
119            {
120                "active": False,
121            },
122        )
123
124    def test_introspect_invalid_provider(self):
125        """Test introspection (mismatched provider and token)"""
126        provider: OAuth2Provider = OAuth2Provider.objects.create(
127            name=generate_id(),
128            authorization_flow=create_test_flow(),
129            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
130            signing_key=create_test_cert(),
131        )
132        auth = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
133
134        token = AccessToken.objects.create(
135            provider=self.provider,
136            user=self.user,
137            token=generate_id(),
138            auth_time=timezone.now(),
139            _scope="openid user profile",
140            _id_token=json.dumps(
141                asdict(
142                    IDToken("foo", "bar"),
143                )
144            ),
145        )
146        res = self.client.post(
147            reverse("authentik_providers_oauth2:token-introspection"),
148            HTTP_AUTHORIZATION=f"Basic {auth}",
149            data={"token": token.token},
150        )
151        self.assertEqual(res.status_code, 200)
152        self.assertJSONEqual(
153            res.content.decode(),
154            {
155                "active": False,
156            },
157        )
158
159    def test_introspect_invalid_auth(self):
160        """Test introspect (invalid auth)"""
161        res = self.client.post(
162            reverse("authentik_providers_oauth2:token-introspection"),
163            HTTP_AUTHORIZATION="Basic qwerqrwe",
164            data={"token": generate_id(), "token_type_hint": "refresh_token"},
165        )
166        self.assertEqual(res.status_code, 200)
167        self.assertJSONEqual(
168            res.content.decode(),
169            {
170                "active": False,
171            },
172        )
173
174    def test_introspect_provider_public(self):
175        """Test introspect"""
176        self.provider.client_type = ClientType.PUBLIC
177        self.provider.save()
178        token = AccessToken.objects.create(
179            provider=self.provider,
180            user=self.user,
181            token=generate_id(),
182            auth_time=timezone.now(),
183            _scope="openid user profile",
184            _id_token=json.dumps(
185                asdict(
186                    IDToken("foo", "bar"),
187                )
188            ),
189        )
190        res = self.client.post(
191            reverse("authentik_providers_oauth2:token-introspection"),
192            HTTP_AUTHORIZATION=f"Basic {self.auth}",
193            data={"token": token.token},
194        )
195        self.assertEqual(res.status_code, 200)
196        self.assertJSONEqual(
197            res.content.decode(),
198            {
199                "active": False,
200            },
201        )
202
203    def test_introspect_provider_fed(self):
204        """Test introspect with federation. self.provider is a confidential
205        client and other_provider is a public client."""
206        other_provider = OAuth2Provider.objects.create(
207            name=generate_id(),
208            authorization_flow=create_test_flow(),
209            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
210            signing_key=create_test_cert(),
211            client_type=ClientType.PUBLIC,
212        )
213        Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider)
214
215        other_provider.jwt_federation_providers.add(self.provider)
216
217        token = AccessToken.objects.create(
218            provider=other_provider,
219            user=self.user,
220            token=generate_id(),
221            auth_time=timezone.now(),
222            _scope="openid user profile",
223            _id_token=json.dumps(
224                asdict(
225                    IDToken("foo", "bar"),
226                )
227            ),
228        )
229        res = self.client.post(
230            reverse("authentik_providers_oauth2:token-introspection"),
231            HTTP_AUTHORIZATION=f"Basic {self.auth}",
232            data={"token": token.token},
233        )
234        self.assertEqual(res.status_code, 200)
235        self.assertJSONEqual(
236            res.content.decode(),
237            {
238                "acr": ACR_AUTHENTIK_DEFAULT,
239                "sub": "bar",
240                "iss": "foo",
241                "active": True,
242                "client_id": other_provider.client_id,
243                "scope": " ".join(token.scope),
244            },
245        )

class TesOAuth2Introspection(authentik.providers.oauth2.tests.utils.OAuthTestCase): View Source

 27class TesOAuth2Introspection(OAuthTestCase):
 28    """Test introspect view"""
 29
 30    def setUp(self) -> None:
 31        super().setUp()
 32        self.provider: OAuth2Provider = OAuth2Provider.objects.create(
 33            name=generate_id(),
 34            authorization_flow=create_test_flow(),
 35            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
 36            signing_key=create_test_cert(),
 37        )
 38        self.app = Application.objects.create(
 39            name=generate_id(), slug=generate_id(), provider=self.provider
 40        )
 41        self.user = create_test_admin_user()
 42        self.auth = b64encode(
 43            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
 44        ).decode()
 45
 46    def test_introspect_refresh(self):
 47        """Test introspect"""
 48        token = RefreshToken.objects.create(
 49            provider=self.provider,
 50            user=self.user,
 51            token=generate_id(),
 52            auth_time=timezone.now(),
 53            _scope="openid user profile",
 54            _id_token=json.dumps(
 55                asdict(
 56                    IDToken("foo", "bar"),
 57                )
 58            ),
 59        )
 60        res = self.client.post(
 61            reverse("authentik_providers_oauth2:token-introspection"),
 62            HTTP_AUTHORIZATION=f"Basic {self.auth}",
 63            data={"token": token.token},
 64        )
 65        self.assertEqual(res.status_code, 200)
 66        self.assertJSONEqual(
 67            res.content.decode(),
 68            {
 69                "acr": ACR_AUTHENTIK_DEFAULT,
 70                "sub": "bar",
 71                "iss": "foo",
 72                "active": True,
 73                "client_id": self.provider.client_id,
 74                "scope": " ".join(token.scope),
 75            },
 76        )
 77
 78    def test_introspect_access(self):
 79        """Test introspect"""
 80        token = AccessToken.objects.create(
 81            provider=self.provider,
 82            user=self.user,
 83            token=generate_id(),
 84            auth_time=timezone.now(),
 85            _scope="openid user profile",
 86            _id_token=json.dumps(
 87                asdict(
 88                    IDToken("foo", "bar"),
 89                )
 90            ),
 91        )
 92        res = self.client.post(
 93            reverse("authentik_providers_oauth2:token-introspection"),
 94            HTTP_AUTHORIZATION=f"Basic {self.auth}",
 95            data={"token": token.token},
 96        )
 97        self.assertEqual(res.status_code, 200)
 98        self.assertJSONEqual(
 99            res.content.decode(),
100            {
101                "acr": ACR_AUTHENTIK_DEFAULT,
102                "sub": "bar",
103                "iss": "foo",
104                "active": True,
105                "client_id": self.provider.client_id,
106                "scope": " ".join(token.scope),
107            },
108        )
109
110    def test_introspect_invalid_token(self):
111        """Test introspect (invalid token)"""
112        res = self.client.post(
113            reverse("authentik_providers_oauth2:token-introspection"),
114            HTTP_AUTHORIZATION=f"Basic {self.auth}",
115            data={"token": generate_id(), "token_type_hint": "refresh_token"},
116        )
117        self.assertEqual(res.status_code, 200)
118        self.assertJSONEqual(
119            res.content.decode(),
120            {
121                "active": False,
122            },
123        )
124
125    def test_introspect_invalid_provider(self):
126        """Test introspection (mismatched provider and token)"""
127        provider: OAuth2Provider = OAuth2Provider.objects.create(
128            name=generate_id(),
129            authorization_flow=create_test_flow(),
130            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
131            signing_key=create_test_cert(),
132        )
133        auth = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
134
135        token = AccessToken.objects.create(
136            provider=self.provider,
137            user=self.user,
138            token=generate_id(),
139            auth_time=timezone.now(),
140            _scope="openid user profile",
141            _id_token=json.dumps(
142                asdict(
143                    IDToken("foo", "bar"),
144                )
145            ),
146        )
147        res = self.client.post(
148            reverse("authentik_providers_oauth2:token-introspection"),
149            HTTP_AUTHORIZATION=f"Basic {auth}",
150            data={"token": token.token},
151        )
152        self.assertEqual(res.status_code, 200)
153        self.assertJSONEqual(
154            res.content.decode(),
155            {
156                "active": False,
157            },
158        )
159
160    def test_introspect_invalid_auth(self):
161        """Test introspect (invalid auth)"""
162        res = self.client.post(
163            reverse("authentik_providers_oauth2:token-introspection"),
164            HTTP_AUTHORIZATION="Basic qwerqrwe",
165            data={"token": generate_id(), "token_type_hint": "refresh_token"},
166        )
167        self.assertEqual(res.status_code, 200)
168        self.assertJSONEqual(
169            res.content.decode(),
170            {
171                "active": False,
172            },
173        )
174
175    def test_introspect_provider_public(self):
176        """Test introspect"""
177        self.provider.client_type = ClientType.PUBLIC
178        self.provider.save()
179        token = AccessToken.objects.create(
180            provider=self.provider,
181            user=self.user,
182            token=generate_id(),
183            auth_time=timezone.now(),
184            _scope="openid user profile",
185            _id_token=json.dumps(
186                asdict(
187                    IDToken("foo", "bar"),
188                )
189            ),
190        )
191        res = self.client.post(
192            reverse("authentik_providers_oauth2:token-introspection"),
193            HTTP_AUTHORIZATION=f"Basic {self.auth}",
194            data={"token": token.token},
195        )
196        self.assertEqual(res.status_code, 200)
197        self.assertJSONEqual(
198            res.content.decode(),
199            {
200                "active": False,
201            },
202        )
203
204    def test_introspect_provider_fed(self):
205        """Test introspect with federation. self.provider is a confidential
206        client and other_provider is a public client."""
207        other_provider = OAuth2Provider.objects.create(
208            name=generate_id(),
209            authorization_flow=create_test_flow(),
210            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
211            signing_key=create_test_cert(),
212            client_type=ClientType.PUBLIC,
213        )
214        Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider)
215
216        other_provider.jwt_federation_providers.add(self.provider)
217
218        token = AccessToken.objects.create(
219            provider=other_provider,
220            user=self.user,
221            token=generate_id(),
222            auth_time=timezone.now(),
223            _scope="openid user profile",
224            _id_token=json.dumps(
225                asdict(
226                    IDToken("foo", "bar"),
227                )
228            ),
229        )
230        res = self.client.post(
231            reverse("authentik_providers_oauth2:token-introspection"),
232            HTTP_AUTHORIZATION=f"Basic {self.auth}",
233            data={"token": token.token},
234        )
235        self.assertEqual(res.status_code, 200)
236        self.assertJSONEqual(
237            res.content.decode(),
238            {
239                "acr": ACR_AUTHENTIK_DEFAULT,
240                "sub": "bar",
241                "iss": "foo",
242                "active": True,
243                "client_id": other_provider.client_id,
244                "scope": " ".join(token.scope),
245            },
246        )

Test introspect view

def setUp(self) -> None: View Source

30    def setUp(self) -> None:
31        super().setUp()
32        self.provider: OAuth2Provider = OAuth2Provider.objects.create(
33            name=generate_id(),
34            authorization_flow=create_test_flow(),
35            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
36            signing_key=create_test_cert(),
37        )
38        self.app = Application.objects.create(
39            name=generate_id(), slug=generate_id(), provider=self.provider
40        )
41        self.user = create_test_admin_user()
42        self.auth = b64encode(
43            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
44        ).decode()

Hook method for setting up the test fixture before exercising it.

def test_introspect_refresh(self): View Source

46    def test_introspect_refresh(self):
47        """Test introspect"""
48        token = RefreshToken.objects.create(
49            provider=self.provider,
50            user=self.user,
51            token=generate_id(),
52            auth_time=timezone.now(),
53            _scope="openid user profile",
54            _id_token=json.dumps(
55                asdict(
56                    IDToken("foo", "bar"),
57                )
58            ),
59        )
60        res = self.client.post(
61            reverse("authentik_providers_oauth2:token-introspection"),
62            HTTP_AUTHORIZATION=f"Basic {self.auth}",
63            data={"token": token.token},
64        )
65        self.assertEqual(res.status_code, 200)
66        self.assertJSONEqual(
67            res.content.decode(),
68            {
69                "acr": ACR_AUTHENTIK_DEFAULT,
70                "sub": "bar",
71                "iss": "foo",
72                "active": True,
73                "client_id": self.provider.client_id,
74                "scope": " ".join(token.scope),
75            },
76        )

Test introspect

def test_introspect_access(self): View Source

 78    def test_introspect_access(self):
 79        """Test introspect"""
 80        token = AccessToken.objects.create(
 81            provider=self.provider,
 82            user=self.user,
 83            token=generate_id(),
 84            auth_time=timezone.now(),
 85            _scope="openid user profile",
 86            _id_token=json.dumps(
 87                asdict(
 88                    IDToken("foo", "bar"),
 89                )
 90            ),
 91        )
 92        res = self.client.post(
 93            reverse("authentik_providers_oauth2:token-introspection"),
 94            HTTP_AUTHORIZATION=f"Basic {self.auth}",
 95            data={"token": token.token},
 96        )
 97        self.assertEqual(res.status_code, 200)
 98        self.assertJSONEqual(
 99            res.content.decode(),
100            {
101                "acr": ACR_AUTHENTIK_DEFAULT,
102                "sub": "bar",
103                "iss": "foo",
104                "active": True,
105                "client_id": self.provider.client_id,
106                "scope": " ".join(token.scope),
107            },
108        )

Test introspect

def test_introspect_invalid_token(self): View Source

110    def test_introspect_invalid_token(self):
111        """Test introspect (invalid token)"""
112        res = self.client.post(
113            reverse("authentik_providers_oauth2:token-introspection"),
114            HTTP_AUTHORIZATION=f"Basic {self.auth}",
115            data={"token": generate_id(), "token_type_hint": "refresh_token"},
116        )
117        self.assertEqual(res.status_code, 200)
118        self.assertJSONEqual(
119            res.content.decode(),
120            {
121                "active": False,
122            },
123        )

Test introspect (invalid token)

def test_introspect_invalid_provider(self): View Source

125    def test_introspect_invalid_provider(self):
126        """Test introspection (mismatched provider and token)"""
127        provider: OAuth2Provider = OAuth2Provider.objects.create(
128            name=generate_id(),
129            authorization_flow=create_test_flow(),
130            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
131            signing_key=create_test_cert(),
132        )
133        auth = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
134
135        token = AccessToken.objects.create(
136            provider=self.provider,
137            user=self.user,
138            token=generate_id(),
139            auth_time=timezone.now(),
140            _scope="openid user profile",
141            _id_token=json.dumps(
142                asdict(
143                    IDToken("foo", "bar"),
144                )
145            ),
146        )
147        res = self.client.post(
148            reverse("authentik_providers_oauth2:token-introspection"),
149            HTTP_AUTHORIZATION=f"Basic {auth}",
150            data={"token": token.token},
151        )
152        self.assertEqual(res.status_code, 200)
153        self.assertJSONEqual(
154            res.content.decode(),
155            {
156                "active": False,
157            },
158        )

Test introspection (mismatched provider and token)

def test_introspect_invalid_auth(self): View Source

160    def test_introspect_invalid_auth(self):
161        """Test introspect (invalid auth)"""
162        res = self.client.post(
163            reverse("authentik_providers_oauth2:token-introspection"),
164            HTTP_AUTHORIZATION="Basic qwerqrwe",
165            data={"token": generate_id(), "token_type_hint": "refresh_token"},
166        )
167        self.assertEqual(res.status_code, 200)
168        self.assertJSONEqual(
169            res.content.decode(),
170            {
171                "active": False,
172            },
173        )

Test introspect (invalid auth)

def test_introspect_provider_public(self): View Source

175    def test_introspect_provider_public(self):
176        """Test introspect"""
177        self.provider.client_type = ClientType.PUBLIC
178        self.provider.save()
179        token = AccessToken.objects.create(
180            provider=self.provider,
181            user=self.user,
182            token=generate_id(),
183            auth_time=timezone.now(),
184            _scope="openid user profile",
185            _id_token=json.dumps(
186                asdict(
187                    IDToken("foo", "bar"),
188                )
189            ),
190        )
191        res = self.client.post(
192            reverse("authentik_providers_oauth2:token-introspection"),
193            HTTP_AUTHORIZATION=f"Basic {self.auth}",
194            data={"token": token.token},
195        )
196        self.assertEqual(res.status_code, 200)
197        self.assertJSONEqual(
198            res.content.decode(),
199            {
200                "active": False,
201            },
202        )

Test introspect

def test_introspect_provider_fed(self): View Source

204    def test_introspect_provider_fed(self):
205        """Test introspect with federation. self.provider is a confidential
206        client and other_provider is a public client."""
207        other_provider = OAuth2Provider.objects.create(
208            name=generate_id(),
209            authorization_flow=create_test_flow(),
210            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
211            signing_key=create_test_cert(),
212            client_type=ClientType.PUBLIC,
213        )
214        Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider)
215
216        other_provider.jwt_federation_providers.add(self.provider)
217
218        token = AccessToken.objects.create(
219            provider=other_provider,
220            user=self.user,
221            token=generate_id(),
222            auth_time=timezone.now(),
223            _scope="openid user profile",
224            _id_token=json.dumps(
225                asdict(
226                    IDToken("foo", "bar"),
227                )
228            ),
229        )
230        res = self.client.post(
231            reverse("authentik_providers_oauth2:token-introspection"),
232            HTTP_AUTHORIZATION=f"Basic {self.auth}",
233            data={"token": token.token},
234        )
235        self.assertEqual(res.status_code, 200)
236        self.assertJSONEqual(
237            res.content.decode(),
238            {
239                "acr": ACR_AUTHENTIK_DEFAULT,
240                "sub": "bar",
241                "iss": "foo",
242                "active": True,
243                "client_id": other_provider.client_id,
244                "scope": " ".join(token.scope),
245            },
246        )

Test introspect with federation. self.provider is a confidential client and other_provider is a public client.

Inherited Members

authentik.providers.oauth2.tests.utils.OAuthTestCase: keypair; required_jwt_keys; setUpClass; assert_non_none_or_unset; validate_jwe; validate_jwt