authentik.providers.oauth2.tests.test_revoke

  1"""Test revoke 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.core.models import Application, AuthenticatedSession, Session
 11from authentik.core.tests.utils import create_test_admin_user, create_test_cert, create_test_flow
 12from authentik.lib.generators import generate_id
 13from authentik.providers.oauth2.id_token import IDToken
 14from authentik.providers.oauth2.models import (
 15    AccessToken,
 16    ClientTypes,
 17    DeviceToken,
 18    OAuth2Provider,
 19    RedirectURI,
 20    RedirectURIMatchingMode,
 21    RefreshToken,
 22)
 23from authentik.providers.oauth2.tests.utils import OAuthTestCase
 24from authentik.root.middleware import ClientIPMiddleware
 25
 26
 27class TesOAuth2Revoke(OAuthTestCase):
 28    """Test revoke 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.app.save()
 42        self.user = create_test_admin_user()
 43        self.auth = b64encode(
 44            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
 45        ).decode()
 46
 47    def test_revoke_refresh(self):
 48        """Test revoke"""
 49        token: RefreshToken = RefreshToken.objects.create(
 50            provider=self.provider,
 51            user=self.user,
 52            token=generate_id(),
 53            auth_time=timezone.now(),
 54            _scope="openid user profile",
 55            _id_token=json.dumps(
 56                asdict(
 57                    IDToken("foo", "bar"),
 58                )
 59            ),
 60        )
 61        res = self.client.post(
 62            reverse("authentik_providers_oauth2:token-revoke"),
 63            HTTP_AUTHORIZATION=f"Basic {self.auth}",
 64            data={
 65                "token": token.token,
 66            },
 67        )
 68        self.assertEqual(res.status_code, 200)
 69
 70    def test_revoke_access(self):
 71        """Test revoke"""
 72        token: AccessToken = AccessToken.objects.create(
 73            provider=self.provider,
 74            user=self.user,
 75            token=generate_id(),
 76            auth_time=timezone.now(),
 77            _scope="openid user profile",
 78            _id_token=json.dumps(
 79                asdict(
 80                    IDToken("foo", "bar"),
 81                )
 82            ),
 83        )
 84        res = self.client.post(
 85            reverse("authentik_providers_oauth2:token-revoke"),
 86            HTTP_AUTHORIZATION=f"Basic {self.auth}",
 87            data={
 88                "token": token.token,
 89            },
 90        )
 91        self.assertEqual(res.status_code, 200)
 92
 93    def test_revoke_invalid(self):
 94        """Test revoke (invalid token)"""
 95        res = self.client.post(
 96            reverse("authentik_providers_oauth2:token-revoke"),
 97            HTTP_AUTHORIZATION=f"Basic {self.auth}",
 98            data={
 99                "token": generate_id(),
100            },
101        )
102        self.assertEqual(res.status_code, 200)
103
104    def test_revoke_invalid_auth(self):
105        """Test revoke (invalid auth)"""
106        res = self.client.post(
107            reverse("authentik_providers_oauth2:token-revoke"),
108            HTTP_AUTHORIZATION="Basic fqewr",
109            data={
110                "token": generate_id(),
111            },
112        )
113        self.assertEqual(res.status_code, 401)
114
115    def test_revoke_public(self):
116        """Test revoke public client"""
117        self.provider.client_type = ClientTypes.PUBLIC
118        self.provider.save()
119        token: AccessToken = AccessToken.objects.create(
120            provider=self.provider,
121            user=self.user,
122            token=generate_id(),
123            auth_time=timezone.now(),
124            _scope="openid user profile",
125            _id_token=json.dumps(
126                asdict(
127                    IDToken("foo", "bar"),
128                )
129            ),
130        )
131        auth_public = b64encode(f"{self.provider.client_id}:{generate_id()}".encode()).decode()
132        res = self.client.post(
133            reverse("authentik_providers_oauth2:token-revoke"),
134            HTTP_AUTHORIZATION=f"Basic {auth_public}",
135            data={
136                "token": token.token,
137            },
138        )
139        self.assertEqual(res.status_code, 200)
140
141    def test_revoke_logout(self):
142        """Test revoke on logout"""
143        self.client.force_login(self.user)
144        AccessToken.objects.create(
145            provider=self.provider,
146            user=self.user,
147            session=self.client.session["authenticatedsession"],
148            token=generate_id(),
149            auth_time=timezone.now(),
150            _scope="openid user profile",
151            _id_token=json.dumps(
152                asdict(
153                    IDToken("foo", "bar"),
154                )
155            ),
156        )
157        self.client.logout()
158        self.assertEqual(AccessToken.objects.including_expired().all().count(), 0)
159
160    def test_revoke_session_delete(self):
161        """Test revoke on logout"""
162        session = AuthenticatedSession.objects.create(
163            session=Session.objects.create(
164                session_key=generate_id(),
165                last_ip=ClientIPMiddleware.default_ip,
166            ),
167            user=self.user,
168        )
169        AccessToken.objects.create(
170            provider=self.provider,
171            user=self.user,
172            session=session,
173            token=generate_id(),
174            auth_time=timezone.now(),
175            _scope="openid user profile",
176            _id_token=json.dumps(
177                asdict(
178                    IDToken("foo", "bar"),
179                )
180            ),
181        )
182        session.delete()
183        self.assertEqual(AccessToken.objects.including_expired().all().count(), 0)
184
185    def test_revoke_user_deactivated(self):
186        """Test revoke on logout"""
187        AccessToken.objects.create(
188            provider=self.provider,
189            user=self.user,
190            token=generate_id(),
191            auth_time=timezone.now(),
192            _scope="openid user profile",
193            _id_token=json.dumps(
194                asdict(
195                    IDToken("foo", "bar"),
196                )
197            ),
198        )
199        RefreshToken.objects.create(
200            provider=self.provider,
201            user=self.user,
202            token=generate_id(),
203            auth_time=timezone.now(),
204            _scope="openid user profile",
205            _id_token=json.dumps(
206                asdict(
207                    IDToken("foo", "bar"),
208                )
209            ),
210        )
211        DeviceToken.objects.create(
212            provider=self.provider,
213            user=self.user,
214            _scope="openid user profile",
215        )
216
217        self.user.is_active = False
218        self.user.save()
219
220        self.assertEqual(AccessToken.objects.including_expired().all().count(), 0)
221        self.assertEqual(RefreshToken.objects.including_expired().all().count(), 0)
222        self.assertEqual(DeviceToken.objects.including_expired().all().count(), 0)