authentik.providers.oauth2.tests.test_revoke
Test revoke view
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 ClientType, 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.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.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 aaa", 109 data={ 110 "token": generate_id(), 111 }, 112 ) 113 self.assertEqual(res.status_code, 401) 114 115 def test_revoke_invalid_auth_secret(self): 116 """Test revoke (invalid secret)""" 117 invalid_auth = b64encode(f"{self.provider.client_id}:aaa".encode()).decode() 118 res = self.client.post( 119 reverse("authentik_providers_oauth2:token-revoke"), 120 HTTP_AUTHORIZATION=f"Basic {invalid_auth}", 121 data={ 122 "token": generate_id(), 123 }, 124 ) 125 self.assertEqual(res.status_code, 401) 126 127 def test_revoke_public(self): 128 """Test revoke public client""" 129 self.provider.client_type = ClientType.PUBLIC 130 self.provider.save() 131 token = AccessToken.objects.create( 132 provider=self.provider, 133 user=self.user, 134 token=generate_id(), 135 auth_time=timezone.now(), 136 _scope="openid user profile", 137 _id_token=json.dumps( 138 asdict( 139 IDToken("foo", "bar"), 140 ) 141 ), 142 ) 143 auth_public = b64encode(f"{self.provider.client_id}:{generate_id()}".encode()).decode() 144 res = self.client.post( 145 reverse("authentik_providers_oauth2:token-revoke"), 146 HTTP_AUTHORIZATION=f"Basic {auth_public}", 147 data={ 148 "token": token.token, 149 }, 150 ) 151 self.assertEqual(res.status_code, 200) 152 153 def test_revoke_logout(self): 154 """Test revoke on logout""" 155 self.client.force_login(self.user) 156 AccessToken.objects.create( 157 provider=self.provider, 158 user=self.user, 159 session=self.client.session["authenticatedsession"], 160 token=generate_id(), 161 auth_time=timezone.now(), 162 _scope="openid user profile", 163 _id_token=json.dumps( 164 asdict( 165 IDToken("foo", "bar"), 166 ) 167 ), 168 ) 169 self.client.logout() 170 self.assertEqual(AccessToken.objects.including_expired().all().count(), 0) 171 172 def test_revoke_session_delete(self): 173 """Test revoke on logout""" 174 session = AuthenticatedSession.objects.create( 175 session=Session.objects.create( 176 session_key=generate_id(), 177 last_ip=ClientIPMiddleware.default_ip, 178 ), 179 user=self.user, 180 ) 181 AccessToken.objects.create( 182 provider=self.provider, 183 user=self.user, 184 session=session, 185 token=generate_id(), 186 auth_time=timezone.now(), 187 _scope="openid user profile", 188 _id_token=json.dumps( 189 asdict( 190 IDToken("foo", "bar"), 191 ) 192 ), 193 ) 194 session.delete() 195 self.assertEqual(AccessToken.objects.including_expired().all().count(), 0) 196 197 def test_revoke_user_deactivated(self): 198 """Test revoke on logout""" 199 AccessToken.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 RefreshToken.objects.create( 212 provider=self.provider, 213 user=self.user, 214 token=generate_id(), 215 auth_time=timezone.now(), 216 _scope="openid user profile", 217 _id_token=json.dumps( 218 asdict( 219 IDToken("foo", "bar"), 220 ) 221 ), 222 ) 223 DeviceToken.objects.create( 224 provider=self.provider, 225 user=self.user, 226 _scope="openid user profile", 227 ) 228 229 self.user.is_active = False 230 self.user.save() 231 232 self.assertEqual(AccessToken.objects.including_expired().all().count(), 0) 233 self.assertEqual(RefreshToken.objects.including_expired().all().count(), 0) 234 self.assertEqual(DeviceToken.objects.including_expired().all().count(), 0) 235 236 def test_revoke_provider_fed(self): 237 """Test revoke with federation. self.provider is a confidential 238 client and other_provider is a public client.""" 239 other_provider = OAuth2Provider.objects.create( 240 name=generate_id(), 241 authorization_flow=create_test_flow(), 242 redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")], 243 signing_key=create_test_cert(), 244 client_type=ClientType.PUBLIC, 245 ) 246 Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider) 247 248 other_provider.jwt_federation_providers.add(self.provider) 249 250 token = AccessToken.objects.create( 251 provider=other_provider, 252 user=self.user, 253 token=generate_id(), 254 auth_time=timezone.now(), 255 _scope="openid user profile", 256 _id_token=json.dumps( 257 asdict( 258 IDToken("foo", "bar"), 259 ) 260 ), 261 ) 262 res = self.client.post( 263 reverse("authentik_providers_oauth2:token-revoke"), 264 HTTP_AUTHORIZATION=f"Basic {self.auth}", 265 data={"token": token.token}, 266 ) 267 self.assertEqual(res.status_code, 200) 268 self.assertJSONEqual(res.content.decode(), {}) 269 270 def test_revoke_provider_fed_public(self): 271 """Test revoke with federation. self.provider is a public 272 client and other_provider is a public client.""" 273 self.provider.client_type = ClientType.PUBLIC 274 self.provider.save() 275 other_provider = OAuth2Provider.objects.create( 276 name=generate_id(), 277 authorization_flow=create_test_flow(), 278 redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")], 279 signing_key=create_test_cert(), 280 client_type=ClientType.PUBLIC, 281 ) 282 Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider) 283 284 other_provider.jwt_federation_providers.add(self.provider) 285 286 token = AccessToken.objects.create( 287 provider=other_provider, 288 user=self.user, 289 token=generate_id(), 290 auth_time=timezone.now(), 291 _scope="openid user profile", 292 _id_token=json.dumps( 293 asdict( 294 IDToken("foo", "bar"), 295 ) 296 ), 297 ) 298 auth_public = b64encode(f"{self.provider.client_id}:{generate_id()}".encode()).decode() 299 res = self.client.post( 300 reverse("authentik_providers_oauth2:token-revoke"), 301 HTTP_AUTHORIZATION=f"Basic {auth_public}", 302 data={"token": token.token}, 303 ) 304 self.assertEqual(res.status_code, 200) 305 self.assertTrue(AccessToken.objects.filter(token=token.token).exists())
28class TesOAuth2Revoke(OAuthTestCase): 29 """Test revoke view""" 30 31 def setUp(self) -> None: 32 super().setUp() 33 self.provider: OAuth2Provider = OAuth2Provider.objects.create( 34 name=generate_id(), 35 authorization_flow=create_test_flow(), 36 redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")], 37 signing_key=create_test_cert(), 38 ) 39 self.app = Application.objects.create( 40 name=generate_id(), slug=generate_id(), provider=self.provider 41 ) 42 self.app.save() 43 self.user = create_test_admin_user() 44 self.auth = b64encode( 45 f"{self.provider.client_id}:{self.provider.client_secret}".encode() 46 ).decode() 47 48 def test_revoke_refresh(self): 49 """Test revoke""" 50 token = RefreshToken.objects.create( 51 provider=self.provider, 52 user=self.user, 53 token=generate_id(), 54 auth_time=timezone.now(), 55 _scope="openid user profile", 56 _id_token=json.dumps( 57 asdict( 58 IDToken("foo", "bar"), 59 ) 60 ), 61 ) 62 res = self.client.post( 63 reverse("authentik_providers_oauth2:token-revoke"), 64 HTTP_AUTHORIZATION=f"Basic {self.auth}", 65 data={ 66 "token": token.token, 67 }, 68 ) 69 self.assertEqual(res.status_code, 200) 70 71 def test_revoke_access(self): 72 """Test revoke""" 73 token = AccessToken.objects.create( 74 provider=self.provider, 75 user=self.user, 76 token=generate_id(), 77 auth_time=timezone.now(), 78 _scope="openid user profile", 79 _id_token=json.dumps( 80 asdict( 81 IDToken("foo", "bar"), 82 ) 83 ), 84 ) 85 res = self.client.post( 86 reverse("authentik_providers_oauth2:token-revoke"), 87 HTTP_AUTHORIZATION=f"Basic {self.auth}", 88 data={ 89 "token": token.token, 90 }, 91 ) 92 self.assertEqual(res.status_code, 200) 93 94 def test_revoke_invalid(self): 95 """Test revoke (invalid token)""" 96 res = self.client.post( 97 reverse("authentik_providers_oauth2:token-revoke"), 98 HTTP_AUTHORIZATION=f"Basic {self.auth}", 99 data={ 100 "token": generate_id(), 101 }, 102 ) 103 self.assertEqual(res.status_code, 200) 104 105 def test_revoke_invalid_auth(self): 106 """Test revoke (invalid auth)""" 107 res = self.client.post( 108 reverse("authentik_providers_oauth2:token-revoke"), 109 HTTP_AUTHORIZATION="Basic aaa", 110 data={ 111 "token": generate_id(), 112 }, 113 ) 114 self.assertEqual(res.status_code, 401) 115 116 def test_revoke_invalid_auth_secret(self): 117 """Test revoke (invalid secret)""" 118 invalid_auth = b64encode(f"{self.provider.client_id}:aaa".encode()).decode() 119 res = self.client.post( 120 reverse("authentik_providers_oauth2:token-revoke"), 121 HTTP_AUTHORIZATION=f"Basic {invalid_auth}", 122 data={ 123 "token": generate_id(), 124 }, 125 ) 126 self.assertEqual(res.status_code, 401) 127 128 def test_revoke_public(self): 129 """Test revoke public client""" 130 self.provider.client_type = ClientType.PUBLIC 131 self.provider.save() 132 token = AccessToken.objects.create( 133 provider=self.provider, 134 user=self.user, 135 token=generate_id(), 136 auth_time=timezone.now(), 137 _scope="openid user profile", 138 _id_token=json.dumps( 139 asdict( 140 IDToken("foo", "bar"), 141 ) 142 ), 143 ) 144 auth_public = b64encode(f"{self.provider.client_id}:{generate_id()}".encode()).decode() 145 res = self.client.post( 146 reverse("authentik_providers_oauth2:token-revoke"), 147 HTTP_AUTHORIZATION=f"Basic {auth_public}", 148 data={ 149 "token": token.token, 150 }, 151 ) 152 self.assertEqual(res.status_code, 200) 153 154 def test_revoke_logout(self): 155 """Test revoke on logout""" 156 self.client.force_login(self.user) 157 AccessToken.objects.create( 158 provider=self.provider, 159 user=self.user, 160 session=self.client.session["authenticatedsession"], 161 token=generate_id(), 162 auth_time=timezone.now(), 163 _scope="openid user profile", 164 _id_token=json.dumps( 165 asdict( 166 IDToken("foo", "bar"), 167 ) 168 ), 169 ) 170 self.client.logout() 171 self.assertEqual(AccessToken.objects.including_expired().all().count(), 0) 172 173 def test_revoke_session_delete(self): 174 """Test revoke on logout""" 175 session = AuthenticatedSession.objects.create( 176 session=Session.objects.create( 177 session_key=generate_id(), 178 last_ip=ClientIPMiddleware.default_ip, 179 ), 180 user=self.user, 181 ) 182 AccessToken.objects.create( 183 provider=self.provider, 184 user=self.user, 185 session=session, 186 token=generate_id(), 187 auth_time=timezone.now(), 188 _scope="openid user profile", 189 _id_token=json.dumps( 190 asdict( 191 IDToken("foo", "bar"), 192 ) 193 ), 194 ) 195 session.delete() 196 self.assertEqual(AccessToken.objects.including_expired().all().count(), 0) 197 198 def test_revoke_user_deactivated(self): 199 """Test revoke on logout""" 200 AccessToken.objects.create( 201 provider=self.provider, 202 user=self.user, 203 token=generate_id(), 204 auth_time=timezone.now(), 205 _scope="openid user profile", 206 _id_token=json.dumps( 207 asdict( 208 IDToken("foo", "bar"), 209 ) 210 ), 211 ) 212 RefreshToken.objects.create( 213 provider=self.provider, 214 user=self.user, 215 token=generate_id(), 216 auth_time=timezone.now(), 217 _scope="openid user profile", 218 _id_token=json.dumps( 219 asdict( 220 IDToken("foo", "bar"), 221 ) 222 ), 223 ) 224 DeviceToken.objects.create( 225 provider=self.provider, 226 user=self.user, 227 _scope="openid user profile", 228 ) 229 230 self.user.is_active = False 231 self.user.save() 232 233 self.assertEqual(AccessToken.objects.including_expired().all().count(), 0) 234 self.assertEqual(RefreshToken.objects.including_expired().all().count(), 0) 235 self.assertEqual(DeviceToken.objects.including_expired().all().count(), 0) 236 237 def test_revoke_provider_fed(self): 238 """Test revoke with federation. self.provider is a confidential 239 client and other_provider is a public client.""" 240 other_provider = OAuth2Provider.objects.create( 241 name=generate_id(), 242 authorization_flow=create_test_flow(), 243 redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")], 244 signing_key=create_test_cert(), 245 client_type=ClientType.PUBLIC, 246 ) 247 Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider) 248 249 other_provider.jwt_federation_providers.add(self.provider) 250 251 token = AccessToken.objects.create( 252 provider=other_provider, 253 user=self.user, 254 token=generate_id(), 255 auth_time=timezone.now(), 256 _scope="openid user profile", 257 _id_token=json.dumps( 258 asdict( 259 IDToken("foo", "bar"), 260 ) 261 ), 262 ) 263 res = self.client.post( 264 reverse("authentik_providers_oauth2:token-revoke"), 265 HTTP_AUTHORIZATION=f"Basic {self.auth}", 266 data={"token": token.token}, 267 ) 268 self.assertEqual(res.status_code, 200) 269 self.assertJSONEqual(res.content.decode(), {}) 270 271 def test_revoke_provider_fed_public(self): 272 """Test revoke with federation. self.provider is a public 273 client and other_provider is a public client.""" 274 self.provider.client_type = ClientType.PUBLIC 275 self.provider.save() 276 other_provider = OAuth2Provider.objects.create( 277 name=generate_id(), 278 authorization_flow=create_test_flow(), 279 redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")], 280 signing_key=create_test_cert(), 281 client_type=ClientType.PUBLIC, 282 ) 283 Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider) 284 285 other_provider.jwt_federation_providers.add(self.provider) 286 287 token = AccessToken.objects.create( 288 provider=other_provider, 289 user=self.user, 290 token=generate_id(), 291 auth_time=timezone.now(), 292 _scope="openid user profile", 293 _id_token=json.dumps( 294 asdict( 295 IDToken("foo", "bar"), 296 ) 297 ), 298 ) 299 auth_public = b64encode(f"{self.provider.client_id}:{generate_id()}".encode()).decode() 300 res = self.client.post( 301 reverse("authentik_providers_oauth2:token-revoke"), 302 HTTP_AUTHORIZATION=f"Basic {auth_public}", 303 data={"token": token.token}, 304 ) 305 self.assertEqual(res.status_code, 200) 306 self.assertTrue(AccessToken.objects.filter(token=token.token).exists())
Test revoke view
def
setUp(self) -> None:
31 def setUp(self) -> None: 32 super().setUp() 33 self.provider: OAuth2Provider = OAuth2Provider.objects.create( 34 name=generate_id(), 35 authorization_flow=create_test_flow(), 36 redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")], 37 signing_key=create_test_cert(), 38 ) 39 self.app = Application.objects.create( 40 name=generate_id(), slug=generate_id(), provider=self.provider 41 ) 42 self.app.save() 43 self.user = create_test_admin_user() 44 self.auth = b64encode( 45 f"{self.provider.client_id}:{self.provider.client_secret}".encode() 46 ).decode()
Hook method for setting up the test fixture before exercising it.
def
test_revoke_refresh(self):
48 def test_revoke_refresh(self): 49 """Test revoke""" 50 token = RefreshToken.objects.create( 51 provider=self.provider, 52 user=self.user, 53 token=generate_id(), 54 auth_time=timezone.now(), 55 _scope="openid user profile", 56 _id_token=json.dumps( 57 asdict( 58 IDToken("foo", "bar"), 59 ) 60 ), 61 ) 62 res = self.client.post( 63 reverse("authentik_providers_oauth2:token-revoke"), 64 HTTP_AUTHORIZATION=f"Basic {self.auth}", 65 data={ 66 "token": token.token, 67 }, 68 ) 69 self.assertEqual(res.status_code, 200)
Test revoke
def
test_revoke_access(self):
71 def test_revoke_access(self): 72 """Test revoke""" 73 token = AccessToken.objects.create( 74 provider=self.provider, 75 user=self.user, 76 token=generate_id(), 77 auth_time=timezone.now(), 78 _scope="openid user profile", 79 _id_token=json.dumps( 80 asdict( 81 IDToken("foo", "bar"), 82 ) 83 ), 84 ) 85 res = self.client.post( 86 reverse("authentik_providers_oauth2:token-revoke"), 87 HTTP_AUTHORIZATION=f"Basic {self.auth}", 88 data={ 89 "token": token.token, 90 }, 91 ) 92 self.assertEqual(res.status_code, 200)
Test revoke
def
test_revoke_invalid(self):
94 def test_revoke_invalid(self): 95 """Test revoke (invalid token)""" 96 res = self.client.post( 97 reverse("authentik_providers_oauth2:token-revoke"), 98 HTTP_AUTHORIZATION=f"Basic {self.auth}", 99 data={ 100 "token": generate_id(), 101 }, 102 ) 103 self.assertEqual(res.status_code, 200)
Test revoke (invalid token)
def
test_revoke_invalid_auth(self):
105 def test_revoke_invalid_auth(self): 106 """Test revoke (invalid auth)""" 107 res = self.client.post( 108 reverse("authentik_providers_oauth2:token-revoke"), 109 HTTP_AUTHORIZATION="Basic aaa", 110 data={ 111 "token": generate_id(), 112 }, 113 ) 114 self.assertEqual(res.status_code, 401)
Test revoke (invalid auth)
def
test_revoke_invalid_auth_secret(self):
116 def test_revoke_invalid_auth_secret(self): 117 """Test revoke (invalid secret)""" 118 invalid_auth = b64encode(f"{self.provider.client_id}:aaa".encode()).decode() 119 res = self.client.post( 120 reverse("authentik_providers_oauth2:token-revoke"), 121 HTTP_AUTHORIZATION=f"Basic {invalid_auth}", 122 data={ 123 "token": generate_id(), 124 }, 125 ) 126 self.assertEqual(res.status_code, 401)
Test revoke (invalid secret)
def
test_revoke_public(self):
128 def test_revoke_public(self): 129 """Test revoke public client""" 130 self.provider.client_type = ClientType.PUBLIC 131 self.provider.save() 132 token = AccessToken.objects.create( 133 provider=self.provider, 134 user=self.user, 135 token=generate_id(), 136 auth_time=timezone.now(), 137 _scope="openid user profile", 138 _id_token=json.dumps( 139 asdict( 140 IDToken("foo", "bar"), 141 ) 142 ), 143 ) 144 auth_public = b64encode(f"{self.provider.client_id}:{generate_id()}".encode()).decode() 145 res = self.client.post( 146 reverse("authentik_providers_oauth2:token-revoke"), 147 HTTP_AUTHORIZATION=f"Basic {auth_public}", 148 data={ 149 "token": token.token, 150 }, 151 ) 152 self.assertEqual(res.status_code, 200)
Test revoke public client
def
test_revoke_logout(self):
154 def test_revoke_logout(self): 155 """Test revoke on logout""" 156 self.client.force_login(self.user) 157 AccessToken.objects.create( 158 provider=self.provider, 159 user=self.user, 160 session=self.client.session["authenticatedsession"], 161 token=generate_id(), 162 auth_time=timezone.now(), 163 _scope="openid user profile", 164 _id_token=json.dumps( 165 asdict( 166 IDToken("foo", "bar"), 167 ) 168 ), 169 ) 170 self.client.logout() 171 self.assertEqual(AccessToken.objects.including_expired().all().count(), 0)
Test revoke on logout
def
test_revoke_session_delete(self):
173 def test_revoke_session_delete(self): 174 """Test revoke on logout""" 175 session = AuthenticatedSession.objects.create( 176 session=Session.objects.create( 177 session_key=generate_id(), 178 last_ip=ClientIPMiddleware.default_ip, 179 ), 180 user=self.user, 181 ) 182 AccessToken.objects.create( 183 provider=self.provider, 184 user=self.user, 185 session=session, 186 token=generate_id(), 187 auth_time=timezone.now(), 188 _scope="openid user profile", 189 _id_token=json.dumps( 190 asdict( 191 IDToken("foo", "bar"), 192 ) 193 ), 194 ) 195 session.delete() 196 self.assertEqual(AccessToken.objects.including_expired().all().count(), 0)
Test revoke on logout
def
test_revoke_user_deactivated(self):
198 def test_revoke_user_deactivated(self): 199 """Test revoke on logout""" 200 AccessToken.objects.create( 201 provider=self.provider, 202 user=self.user, 203 token=generate_id(), 204 auth_time=timezone.now(), 205 _scope="openid user profile", 206 _id_token=json.dumps( 207 asdict( 208 IDToken("foo", "bar"), 209 ) 210 ), 211 ) 212 RefreshToken.objects.create( 213 provider=self.provider, 214 user=self.user, 215 token=generate_id(), 216 auth_time=timezone.now(), 217 _scope="openid user profile", 218 _id_token=json.dumps( 219 asdict( 220 IDToken("foo", "bar"), 221 ) 222 ), 223 ) 224 DeviceToken.objects.create( 225 provider=self.provider, 226 user=self.user, 227 _scope="openid user profile", 228 ) 229 230 self.user.is_active = False 231 self.user.save() 232 233 self.assertEqual(AccessToken.objects.including_expired().all().count(), 0) 234 self.assertEqual(RefreshToken.objects.including_expired().all().count(), 0) 235 self.assertEqual(DeviceToken.objects.including_expired().all().count(), 0)
Test revoke on logout
def
test_revoke_provider_fed(self):
237 def test_revoke_provider_fed(self): 238 """Test revoke with federation. self.provider is a confidential 239 client and other_provider is a public client.""" 240 other_provider = OAuth2Provider.objects.create( 241 name=generate_id(), 242 authorization_flow=create_test_flow(), 243 redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")], 244 signing_key=create_test_cert(), 245 client_type=ClientType.PUBLIC, 246 ) 247 Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider) 248 249 other_provider.jwt_federation_providers.add(self.provider) 250 251 token = AccessToken.objects.create( 252 provider=other_provider, 253 user=self.user, 254 token=generate_id(), 255 auth_time=timezone.now(), 256 _scope="openid user profile", 257 _id_token=json.dumps( 258 asdict( 259 IDToken("foo", "bar"), 260 ) 261 ), 262 ) 263 res = self.client.post( 264 reverse("authentik_providers_oauth2:token-revoke"), 265 HTTP_AUTHORIZATION=f"Basic {self.auth}", 266 data={"token": token.token}, 267 ) 268 self.assertEqual(res.status_code, 200) 269 self.assertJSONEqual(res.content.decode(), {})
Test revoke with federation. self.provider is a confidential client and other_provider is a public client.
def
test_revoke_provider_fed_public(self):
271 def test_revoke_provider_fed_public(self): 272 """Test revoke with federation. self.provider is a public 273 client and other_provider is a public client.""" 274 self.provider.client_type = ClientType.PUBLIC 275 self.provider.save() 276 other_provider = OAuth2Provider.objects.create( 277 name=generate_id(), 278 authorization_flow=create_test_flow(), 279 redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")], 280 signing_key=create_test_cert(), 281 client_type=ClientType.PUBLIC, 282 ) 283 Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider) 284 285 other_provider.jwt_federation_providers.add(self.provider) 286 287 token = AccessToken.objects.create( 288 provider=other_provider, 289 user=self.user, 290 token=generate_id(), 291 auth_time=timezone.now(), 292 _scope="openid user profile", 293 _id_token=json.dumps( 294 asdict( 295 IDToken("foo", "bar"), 296 ) 297 ), 298 ) 299 auth_public = b64encode(f"{self.provider.client_id}:{generate_id()}".encode()).decode() 300 res = self.client.post( 301 reverse("authentik_providers_oauth2:token-revoke"), 302 HTTP_AUTHORIZATION=f"Basic {auth_public}", 303 data={"token": token.token}, 304 ) 305 self.assertEqual(res.status_code, 200) 306 self.assertTrue(AccessToken.objects.filter(token=token.token).exists())
Test revoke with federation. self.provider is a public client and other_provider is a public client.