habsi/authentik
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

providers/oauth2: fix validation ordering (cherry-pick #8793) (#8795)

Browse Source 
providers/oauth2: fix validation ordering (#8793)

Signed-off-by: Jens Langhammer <jens@goauthentik.io>
Co-authored-by: Jens L <jens@goauthentik.io>
This commit is contained in:
gcp-cherry-pick-bot[bot]
2024-03-04 18:56:43 +01:00
committed by GitHub
parent 9b84bf7174
commit eda38a30b1
2 changed files with 44 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
authentik/providers/oauth2/tests/test_authorize.py
View File

@ -36,8 +36,21 @@ class TestAuthorize(OAuthTestCase):
def test_invalid_grant_type(self):
"""Test with invalid grant type"""
OAuth2Provider.objects.create(
name=generate_id(),
client_id="test",
authorization_flow=create_test_flow(),
redirect_uris="http://local.invalid/Foo",
)
with self.assertRaises(AuthorizeError):
request = self.factory.get("/", data={"response_type": "invalid"})
request = self.factory.get(
"/",
data={
"response_type": "invalid",
"client_id": "test",
"redirect_uri": "http://local.invalid/Foo",
},
)
OAuthAuthorizationParams.from_request(request)
def test_invalid_client_id(self):

57
authentik/providers/oauth2/views/authorize.py
View File

@ -121,44 +121,18 @@ class OAuthAuthorizationParams:
redirect_uri = query_dict.get("redirect_uri", "")
response_type = query_dict.get("response_type", "")
grant_type = None
# Determine which flow to use.
if response_type in [ResponseTypes.CODE]:
grant_type = GrantTypes.AUTHORIZATION_CODE
elif response_type in [
ResponseTypes.ID_TOKEN,
ResponseTypes.ID_TOKEN_TOKEN,
]:
grant_type = GrantTypes.IMPLICIT
elif response_type in [
ResponseTypes.CODE_TOKEN,
ResponseTypes.CODE_ID_TOKEN,
ResponseTypes.CODE_ID_TOKEN_TOKEN,
]:
grant_type = GrantTypes.HYBRID
# Grant type validation.
if not grant_type:
LOGGER.warning("Invalid response type", type=response_type)
raise AuthorizeError(redirect_uri, "unsupported_response_type", "", state)
# Validate and check the response_mode against the predefined dict
# Set to Query or Fragment if not defined in request
response_mode = query_dict.get("response_mode", False)
if response_mode not in ResponseMode.values:
response_mode = ResponseMode.QUERY
if grant_type in [GrantTypes.IMPLICIT, GrantTypes.HYBRID]:
response_mode = ResponseMode.FRAGMENT
max_age = query_dict.get("max_age")
return OAuthAuthorizationParams(
client_id=query_dict.get("client_id", ""),
redirect_uri=redirect_uri,
response_type=response_type,
response_mode=response_mode,
grant_type=grant_type,
grant_type="",
scope=set(query_dict.get("scope", "").split()),
state=state,
nonce=query_dict.get("nonce"),
@ -178,6 +152,7 @@ class OAuthAuthorizationParams:
LOGGER.warning("Invalid client identifier", client_id=self.client_id)
raise ClientIdError(client_id=self.client_id)
self.check_redirect_uri()
self.check_grant()
self.check_scope(github_compat)
self.check_nonce()
self.check_code_challenge()
@ -186,6 +161,34 @@ class OAuthAuthorizationParams:
self.redirect_uri, "request_not_supported", self.grant_type, self.state
)
def check_grant(self):
"""Check grant"""
# Determine which flow to use.
if self.response_type in [ResponseTypes.CODE]:
self.grant_type = GrantTypes.AUTHORIZATION_CODE
elif self.response_type in [
ResponseTypes.ID_TOKEN,
ResponseTypes.ID_TOKEN_TOKEN,
]:
self.grant_type = GrantTypes.IMPLICIT
elif self.response_type in [
ResponseTypes.CODE_TOKEN,
ResponseTypes.CODE_ID_TOKEN,
ResponseTypes.CODE_ID_TOKEN_TOKEN,
]:
self.grant_type = GrantTypes.HYBRID
# Grant type validation.
if not self.grant_type:
LOGGER.warning("Invalid response type", type=self.response_type)
raise AuthorizeError(self.redirect_uri, "unsupported_response_type", "", self.state)
if self.response_mode not in ResponseMode.values:
self.response_mode = ResponseMode.QUERY
if self.grant_type in [GrantTypes.IMPLICIT, GrantTypes.HYBRID]:
self.response_mode = ResponseMode.FRAGMENT
def check_redirect_uri(self):
"""Redirect URI validation."""
allowed_redirect_urls = self.provider.redirect_uris.split()