providers/oauth2: fix OAuth form_post response mode for code response_type
Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org> #3113
This commit is contained in:
parent
ece0429ea8
commit
6c9dc7a15b
|
@ -322,18 +322,18 @@ class TestAuthorize(OAuthTestCase):
|
||||||
)
|
)
|
||||||
self.validate_jwt(token, provider)
|
self.validate_jwt(token, provider)
|
||||||
|
|
||||||
def test_full_form_post(self):
|
def test_full_form_post_id_token(self):
|
||||||
"""Test full authorization (form_post response)"""
|
"""Test full authorization (form_post response)"""
|
||||||
flow = create_test_flow()
|
flow = create_test_flow()
|
||||||
provider = OAuth2Provider.objects.create(
|
provider = OAuth2Provider.objects.create(
|
||||||
name=generate_id(),
|
name=generate_id(),
|
||||||
client_id="test",
|
client_id=generate_id(),
|
||||||
client_secret=generate_key(),
|
client_secret=generate_key(),
|
||||||
authorization_flow=flow,
|
authorization_flow=flow,
|
||||||
redirect_uris="http://localhost",
|
redirect_uris="http://localhost",
|
||||||
signing_key=self.keypair,
|
signing_key=self.keypair,
|
||||||
)
|
)
|
||||||
Application.objects.create(name="app", slug="app", provider=provider)
|
app = Application.objects.create(name=generate_id(), slug=generate_id(), provider=provider)
|
||||||
state = generate_id()
|
state = generate_id()
|
||||||
user = create_test_admin_user()
|
user = create_test_admin_user()
|
||||||
self.client.force_login(user)
|
self.client.force_login(user)
|
||||||
|
@ -343,7 +343,7 @@ class TestAuthorize(OAuthTestCase):
|
||||||
data={
|
data={
|
||||||
"response_type": "id_token",
|
"response_type": "id_token",
|
||||||
"response_mode": "form_post",
|
"response_mode": "form_post",
|
||||||
"client_id": "test",
|
"client_id": provider.client_id,
|
||||||
"state": state,
|
"state": state,
|
||||||
"scope": "openid",
|
"scope": "openid",
|
||||||
"redirect_uri": "http://localhost",
|
"redirect_uri": "http://localhost",
|
||||||
|
@ -359,7 +359,7 @@ class TestAuthorize(OAuthTestCase):
|
||||||
"component": "ak-stage-autosubmit",
|
"component": "ak-stage-autosubmit",
|
||||||
"type": ChallengeTypes.NATIVE.value,
|
"type": ChallengeTypes.NATIVE.value,
|
||||||
"url": "http://localhost",
|
"url": "http://localhost",
|
||||||
"title": "Redirecting to app...",
|
"title": f"Redirecting to {app.name}...",
|
||||||
"attrs": {
|
"attrs": {
|
||||||
"access_token": token.access_token,
|
"access_token": token.access_token,
|
||||||
"id_token": provider.encode(token.id_token.to_dict()),
|
"id_token": provider.encode(token.id_token.to_dict()),
|
||||||
|
@ -370,3 +370,48 @@ class TestAuthorize(OAuthTestCase):
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
self.validate_jwt(token, provider)
|
self.validate_jwt(token, provider)
|
||||||
|
|
||||||
|
def test_full_form_post_code(self):
|
||||||
|
"""Test full authorization (form_post response, code type)"""
|
||||||
|
flow = create_test_flow()
|
||||||
|
provider = OAuth2Provider.objects.create(
|
||||||
|
name=generate_id(),
|
||||||
|
client_id=generate_id(),
|
||||||
|
client_secret=generate_key(),
|
||||||
|
authorization_flow=flow,
|
||||||
|
redirect_uris="http://localhost",
|
||||||
|
signing_key=self.keypair,
|
||||||
|
)
|
||||||
|
app = Application.objects.create(name=generate_id(), slug=generate_id(), provider=provider)
|
||||||
|
state = generate_id()
|
||||||
|
user = create_test_admin_user()
|
||||||
|
self.client.force_login(user)
|
||||||
|
# Step 1, initiate params and get redirect to flow
|
||||||
|
self.client.get(
|
||||||
|
reverse("authentik_providers_oauth2:authorize"),
|
||||||
|
data={
|
||||||
|
"response_type": "code",
|
||||||
|
"response_mode": "form_post",
|
||||||
|
"client_id": provider.client_id,
|
||||||
|
"state": state,
|
||||||
|
"scope": "openid",
|
||||||
|
"redirect_uri": "http://localhost",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
response = self.client.get(
|
||||||
|
reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
|
||||||
|
)
|
||||||
|
code: AuthorizationCode = AuthorizationCode.objects.filter(user=user).first()
|
||||||
|
self.assertJSONEqual(
|
||||||
|
response.content.decode(),
|
||||||
|
{
|
||||||
|
"component": "ak-stage-autosubmit",
|
||||||
|
"type": ChallengeTypes.NATIVE.value,
|
||||||
|
"url": "http://localhost",
|
||||||
|
"title": f"Redirecting to {app.name}...",
|
||||||
|
"attrs": {
|
||||||
|
"code": code.code,
|
||||||
|
"state": state,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
|
@ -465,7 +465,6 @@ class OAuthFulfillmentStage(StageView):
|
||||||
def create_response_uri(self) -> str:
|
def create_response_uri(self) -> str:
|
||||||
"""Create a final Response URI the user is redirected to."""
|
"""Create a final Response URI the user is redirected to."""
|
||||||
uri = urlsplit(self.params.redirect_uri)
|
uri = urlsplit(self.params.redirect_uri)
|
||||||
query_params = parse_qs(uri.query)
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
code = None
|
code = None
|
||||||
|
@ -478,6 +477,7 @@ class OAuthFulfillmentStage(StageView):
|
||||||
code.save(force_insert=True)
|
code.save(force_insert=True)
|
||||||
|
|
||||||
if self.params.response_mode == ResponseMode.QUERY:
|
if self.params.response_mode == ResponseMode.QUERY:
|
||||||
|
query_params = parse_qs(uri.query)
|
||||||
query_params["code"] = code.code
|
query_params["code"] = code.code
|
||||||
query_params["state"] = [str(self.params.state) if self.params.state else ""]
|
query_params["state"] = [str(self.params.state) if self.params.state else ""]
|
||||||
|
|
||||||
|
@ -494,6 +494,11 @@ class OAuthFulfillmentStage(StageView):
|
||||||
return urlunsplit(uri)
|
return urlunsplit(uri)
|
||||||
|
|
||||||
if self.params.response_mode == ResponseMode.FORM_POST:
|
if self.params.response_mode == ResponseMode.FORM_POST:
|
||||||
|
post_params = {}
|
||||||
|
if self.params.grant_type in [GrantTypes.AUTHORIZATION_CODE]:
|
||||||
|
post_params["code"] = code.code
|
||||||
|
post_params["state"] = [str(self.params.state) if self.params.state else ""]
|
||||||
|
else:
|
||||||
post_params = self.create_implicit_response(code)
|
post_params = self.create_implicit_response(code)
|
||||||
|
|
||||||
uri = uri._replace(query=urlencode(post_params, doseq=True))
|
uri = uri._replace(query=urlencode(post_params, doseq=True))
|
||||||
|
|
Reference in New Issue