authentik.providers.oauth2.tests.test_introspect

  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    OAuth2Provider,
 18    RedirectURI,
 19    RedirectURIMatchingMode,
 20    RefreshToken,
 21)
 22from authentik.providers.oauth2.tests.utils import OAuthTestCase
 23
 24
 25class TesOAuth2Introspection(OAuthTestCase):
 26    """Test introspect view"""
 27
 28    def setUp(self) -> None:
 29        super().setUp()
 30        self.provider: OAuth2Provider = OAuth2Provider.objects.create(
 31            name=generate_id(),
 32            authorization_flow=create_test_flow(),
 33            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
 34            signing_key=create_test_cert(),
 35        )
 36        self.app = Application.objects.create(
 37            name=generate_id(), slug=generate_id(), provider=self.provider
 38        )
 39        self.user = create_test_admin_user()
 40        self.auth = b64encode(
 41            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
 42        ).decode()
 43
 44    def test_introspect_refresh(self):
 45        """Test introspect"""
 46        token: RefreshToken = RefreshToken.objects.create(
 47            provider=self.provider,
 48            user=self.user,
 49            token=generate_id(),
 50            auth_time=timezone.now(),
 51            _scope="openid user profile",
 52            _id_token=json.dumps(
 53                asdict(
 54                    IDToken("foo", "bar"),
 55                )
 56            ),
 57        )
 58        res = self.client.post(
 59            reverse("authentik_providers_oauth2:token-introspection"),
 60            HTTP_AUTHORIZATION=f"Basic {self.auth}",
 61            data={"token": token.token},
 62        )
 63        self.assertEqual(res.status_code, 200)
 64        self.assertJSONEqual(
 65            res.content.decode(),
 66            {
 67                "acr": ACR_AUTHENTIK_DEFAULT,
 68                "sub": "bar",
 69                "iss": "foo",
 70                "active": True,
 71                "client_id": self.provider.client_id,
 72                "scope": " ".join(token.scope),
 73            },
 74        )
 75
 76    def test_introspect_access(self):
 77        """Test introspect"""
 78        token: AccessToken = AccessToken.objects.create(
 79            provider=self.provider,
 80            user=self.user,
 81            token=generate_id(),
 82            auth_time=timezone.now(),
 83            _scope="openid user profile",
 84            _id_token=json.dumps(
 85                asdict(
 86                    IDToken("foo", "bar"),
 87                )
 88            ),
 89        )
 90        res = self.client.post(
 91            reverse("authentik_providers_oauth2:token-introspection"),
 92            HTTP_AUTHORIZATION=f"Basic {self.auth}",
 93            data={"token": token.token},
 94        )
 95        self.assertEqual(res.status_code, 200)
 96        self.assertJSONEqual(
 97            res.content.decode(),
 98            {
 99                "acr": ACR_AUTHENTIK_DEFAULT,
100                "sub": "bar",
101                "iss": "foo",
102                "active": True,
103                "client_id": self.provider.client_id,
104                "scope": " ".join(token.scope),
105            },
106        )
107
108    def test_introspect_invalid_token(self):
109        """Test introspect (invalid token)"""
110        res = self.client.post(
111            reverse("authentik_providers_oauth2:token-introspection"),
112            HTTP_AUTHORIZATION=f"Basic {self.auth}",
113            data={"token": generate_id(), "token_type_hint": "refresh_token"},
114        )
115        self.assertEqual(res.status_code, 200)
116        self.assertJSONEqual(
117            res.content.decode(),
118            {
119                "active": False,
120            },
121        )
122
123    def test_introspect_invalid_provider(self):
124        """Test introspection (mismatched provider and token)"""
125        provider: OAuth2Provider = OAuth2Provider.objects.create(
126            name=generate_id(),
127            authorization_flow=create_test_flow(),
128            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
129            signing_key=create_test_cert(),
130        )
131        auth = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
132
133        token: AccessToken = AccessToken.objects.create(
134            provider=self.provider,
135            user=self.user,
136            token=generate_id(),
137            auth_time=timezone.now(),
138            _scope="openid user profile",
139            _id_token=json.dumps(
140                asdict(
141                    IDToken("foo", "bar"),
142                )
143            ),
144        )
145        res = self.client.post(
146            reverse("authentik_providers_oauth2:token-introspection"),
147            HTTP_AUTHORIZATION=f"Basic {auth}",
148            data={"token": token.token},
149        )
150        self.assertEqual(res.status_code, 200)
151        self.assertJSONEqual(
152            res.content.decode(),
153            {
154                "active": False,
155            },
156        )
157
158    def test_introspect_invalid_auth(self):
159        """Test introspect (invalid auth)"""
160        res = self.client.post(
161            reverse("authentik_providers_oauth2:token-introspection"),
162            HTTP_AUTHORIZATION="Basic qwerqrwe",
163            data={"token": generate_id(), "token_type_hint": "refresh_token"},
164        )
165        self.assertEqual(res.status_code, 200)
166        self.assertJSONEqual(
167            res.content.decode(),
168            {
169                "active": False,
170            },
171        )