authentik.providers.oauth2.tests.test_token_pkce

  1"""Test token view"""
  2
  3from base64 import b64encode
  4
  5from django.test import RequestFactory
  6from django.urls import reverse
  7
  8from authentik.common.oauth.constants import GRANT_TYPE_AUTHORIZATION_CODE
  9from authentik.core.models import Application
 10from authentik.core.tests.utils import create_test_admin_user, create_test_flow
 11from authentik.lib.generators import generate_id
 12from authentik.providers.oauth2.models import (
 13    AuthorizationCode,
 14    OAuth2Provider,
 15    RedirectURI,
 16    RedirectURIMatchingMode,
 17)
 18from authentik.providers.oauth2.tests.utils import OAuthTestCase
 19from authentik.providers.oauth2.utils import pkce_s256_challenge
 20
 21
 22class TestTokenPKCE(OAuthTestCase):
 23    """Test token view"""
 24
 25    def setUp(self) -> None:
 26        super().setUp()
 27        self.factory = RequestFactory()
 28        self.app = Application.objects.create(name=generate_id(), slug="test")
 29
 30    def test_pkce_missing_in_authorize(self):
 31        """Test PKCE with code_challenge in authorize request
 32        and missing verifier in token request"""
 33        flow = create_test_flow()
 34        provider = OAuth2Provider.objects.create(
 35            name=generate_id(),
 36            client_id="test",
 37            authorization_flow=flow,
 38            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
 39            access_code_validity="seconds=100",
 40        )
 41        Application.objects.create(name="app", slug="app", provider=provider)
 42        state = generate_id()
 43        user = create_test_admin_user()
 44        self.client.force_login(user)
 45        challenge = generate_id()
 46        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
 47        # Step 1, initiate params and get redirect to flow
 48        response = self.client.get(
 49            reverse("authentik_providers_oauth2:authorize"),
 50            data={
 51                "response_type": "code",
 52                "client_id": "test",
 53                "state": state,
 54                "redirect_uri": "foo://localhost",
 55                "code_challenge": challenge,
 56                "code_challenge_method": "S256",
 57            },
 58        )
 59        code: AuthorizationCode = AuthorizationCode.objects.filter(user=user).first()
 60        self.assertEqual(
 61            response.url,
 62            f"foo://localhost?code={code.code}&state={state}",
 63        )
 64        response = self.client.post(
 65            reverse("authentik_providers_oauth2:token"),
 66            data={
 67                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
 68                "code": code.code,
 69                # Missing the code_verifier here
 70                "redirect_uri": "foo://localhost",
 71            },
 72            HTTP_AUTHORIZATION=f"Basic {header}",
 73        )
 74        self.assertJSONEqual(
 75            response.content,
 76            {
 77                "error": "invalid_grant",
 78                "error_description": (
 79                    "The provided authorization grant or refresh token is invalid, expired, "
 80                    "revoked, does not match the redirection URI used in the authorization "
 81                    "request, or was issued to another client"
 82                ),
 83                "request_id": response.headers["X-authentik-id"],
 84            },
 85        )
 86        self.assertEqual(response.status_code, 400)
 87
 88    def test_pkce_missing_in_token(self):
 89        """Test PKCE with missing code_challenge in authorization request but verifier
 90        set in token request"""
 91        flow = create_test_flow()
 92        provider = OAuth2Provider.objects.create(
 93            name=generate_id(),
 94            client_id="test",
 95            authorization_flow=flow,
 96            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
 97            access_code_validity="seconds=100",
 98        )
 99        Application.objects.create(name="app", slug="app", provider=provider)
100        state = generate_id()
101        user = create_test_admin_user()
102        self.client.force_login(user)
103        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
104        # Step 1, initiate params and get redirect to flow
105        response = self.client.get(
106            reverse("authentik_providers_oauth2:authorize"),
107            data={
108                "response_type": "code",
109                "client_id": "test",
110                "state": state,
111                "redirect_uri": "foo://localhost",
112                # "code_challenge": challenge,
113                # "code_challenge_method": "S256",
114            },
115        )
116        code: AuthorizationCode = AuthorizationCode.objects.filter(user=user).first()
117        self.assertEqual(
118            response.url,
119            f"foo://localhost?code={code.code}&state={state}",
120        )
121        response = self.client.post(
122            reverse("authentik_providers_oauth2:token"),
123            data={
124                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
125                "code": code.code,
126                "code_verifier": generate_id(),
127                "redirect_uri": "foo://localhost",
128            },
129            HTTP_AUTHORIZATION=f"Basic {header}",
130        )
131        self.assertJSONEqual(
132            response.content,
133            {
134                "error": "invalid_grant",
135                "error_description": (
136                    "The provided authorization grant or refresh token is invalid, expired, "
137                    "revoked, does not match the redirection URI used in the authorization "
138                    "request, or was issued to another client"
139                ),
140                "request_id": response.headers["X-authentik-id"],
141            },
142        )
143        self.assertEqual(response.status_code, 400)
144
145    def test_pkce_correct_s256(self):
146        """Test full with pkce"""
147        flow = create_test_flow()
148        provider = OAuth2Provider.objects.create(
149            name=generate_id(),
150            client_id="test",
151            authorization_flow=flow,
152            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
153            access_code_validity="seconds=100",
154        )
155        Application.objects.create(name="app", slug="app", provider=provider)
156        state = generate_id()
157        user = create_test_admin_user()
158        self.client.force_login(user)
159        verifier = generate_id()
160        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
161        # Step 1, initiate params and get redirect to flow
162        response = self.client.get(
163            reverse("authentik_providers_oauth2:authorize"),
164            data={
165                "response_type": "code",
166                "client_id": "test",
167                "state": state,
168                "redirect_uri": "foo://localhost",
169                "code_challenge": pkce_s256_challenge(verifier),
170                "code_challenge_method": "S256",
171            },
172        )
173        code: AuthorizationCode = AuthorizationCode.objects.filter(user=user).first()
174        self.assertEqual(
175            response.url,
176            f"foo://localhost?code={code.code}&state={state}",
177        )
178        response = self.client.post(
179            reverse("authentik_providers_oauth2:token"),
180            data={
181                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
182                "code": code.code,
183                "code_verifier": verifier,
184                "redirect_uri": "foo://localhost",
185            },
186            HTTP_AUTHORIZATION=f"Basic {header}",
187        )
188        self.assertEqual(response.status_code, 200)
189
190    def test_pkce_correct_plain(self):
191        """Test full with pkce"""
192        flow = create_test_flow()
193        provider = OAuth2Provider.objects.create(
194            name=generate_id(),
195            client_id="test",
196            authorization_flow=flow,
197            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
198            access_code_validity="seconds=100",
199        )
200        Application.objects.create(name="app", slug="app", provider=provider)
201        state = generate_id()
202        user = create_test_admin_user()
203        self.client.force_login(user)
204        verifier = generate_id()
205        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
206        # Step 1, initiate params and get redirect to flow
207        response = self.client.get(
208            reverse("authentik_providers_oauth2:authorize"),
209            data={
210                "response_type": "code",
211                "client_id": "test",
212                "state": state,
213                "redirect_uri": "foo://localhost",
214                "code_challenge": verifier,
215            },
216        )
217        code: AuthorizationCode = AuthorizationCode.objects.filter(user=user).first()
218        self.assertEqual(
219            response.url,
220            f"foo://localhost?code={code.code}&state={state}",
221        )
222        response = self.client.post(
223            reverse("authentik_providers_oauth2:token"),
224            data={
225                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
226                "code": code.code,
227                "code_verifier": verifier,
228                "redirect_uri": "foo://localhost",
229            },
230            HTTP_AUTHORIZATION=f"Basic {header}",
231        )
232        self.assertEqual(response.status_code, 200)