Validate OAuth request parameters and S256 challenges (#184255)

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Jason Hunter
2026-10-05 11:43:09 -04:00
committed by GitHub
co-authored by Paulus Schoutsen Claude Opus 5.5
parent af8049febc
commit 6bd23385f8
4 changed files with 88 additions and 1 deletions
@@ -291,6 +291,14 @@ class TokenView(HomeAssistantView):
hass = request.app[KEY_HASS]
data = cast(MultiDictProxy[str], await request.post())
# RFC 6749 3.2: parameters must not be included more than once.
if len(data) != len(set(data)) or any(
not isinstance(value, str) for value in data.values()
):
return self.json(
{"error": "invalid_request"}, status_code=HTTPStatus.BAD_REQUEST
)
grant_type = data.get("grant_type")
# IndieAuth 6.3.5
@@ -365,6 +365,7 @@ class LoginFlowIndexView(LoginFlowBaseView):
r"^[A-Za-z0-9_-]{43}\Z"
),
probatio.Optional("code_challenge_method"): str,
probatio.Optional("response_type"): str,
probatio.Optional(
"type", default="authorize"
): str, # not used, kept for backwards compatibility
@@ -380,6 +381,11 @@ class LoginFlowIndexView(LoginFlowBaseView):
if not indieauth.verify_client_id(client_id):
return self.json_message("Invalid client id", HTTPStatus.BAD_REQUEST)
if data.get("response_type", "code") != "code":
return self.json_message(
"Response type not supported", HTTPStatus.BAD_REQUEST
)
code_challenge = data.get("code_challenge")
code_challenge_method = data.get("code_challenge_method")
if code_challenge_method is not None and not code_challenge:
+59
View File
@@ -6,8 +6,10 @@ import logging
from typing import Any
from unittest.mock import patch
from aiohttp import FormData
from aiohttp.test_utils import TestClient
from freezegun.api import FrozenDateTimeFactory
from multidict import MultiDict
import pytest
from homeassistant.auth import InvalidAuthError
@@ -800,6 +802,63 @@ async def _async_login_for_code(
return step["result"]
@pytest.mark.parametrize(
"parameter", ["client_id", "grant_type", "code", "redirect_uri", "code_verifier"]
)
async def test_pkce_token_request_rejects_duplicate_parameters(
hass: HomeAssistant,
aiohttp_client: ClientSessionGenerator,
parameter: str,
) -> None:
"""Test ambiguous token parameters are rejected without consuming the code."""
client = await async_setup_auth(hass, aiohttp_client, setup_api=True)
code = await _async_login_for_code(client, RFC7636_CHALLENGE)
token_data = MultiDict(
{
"client_id": CLIENT_ID,
"grant_type": "authorization_code",
"code": code,
"redirect_uri": CLIENT_REDIRECT_URI,
"code_verifier": RFC7636_VERIFIER,
}
)
token_data.add(parameter, "other")
resp = await client.post("/auth/token", data=token_data)
assert resp.status == HTTPStatus.BAD_REQUEST
assert (await resp.json())["error"] == "invalid_request"
resp = await client.post("/auth/token", data=dict(token_data))
assert resp.status == HTTPStatus.OK
async def test_pkce_token_request_rejects_file_verifier(
hass: HomeAssistant, aiohttp_client: ClientSessionGenerator
) -> None:
"""Test a multipart file cannot be used as a code verifier."""
client = await async_setup_auth(hass, aiohttp_client, setup_api=True)
code = await _async_login_for_code(client, RFC7636_CHALLENGE)
token_data = {
"client_id": CLIENT_ID,
"grant_type": "authorization_code",
"code": code,
"redirect_uri": CLIENT_REDIRECT_URI,
}
form_data = FormData(token_data)
form_data.add_field("code_verifier", RFC7636_VERIFIER.encode(), filename="verifier")
resp = await client.post("/auth/token", data=form_data)
assert resp.status == HTTPStatus.BAD_REQUEST
assert (await resp.json())["error"] == "invalid_request"
resp = await client.post(
"/auth/token", data={**token_data, "code_verifier": RFC7636_VERIFIER}
)
assert resp.status == HTTPStatus.OK
async def test_auth_code_pkce_success(
hass: HomeAssistant, aiohttp_client: ClientSessionGenerator
) -> None:
+15 -1
View File
@@ -245,8 +245,14 @@ async def test_invalid_redirect_uri(
assert data["message"] == "Invalid redirect URI"
@pytest.mark.parametrize(
"authorization_data",
[{}, {"response_type": "code"}],
)
async def test_login_exist_user(
hass: HomeAssistant, aiohttp_client: ClientSessionGenerator
hass: HomeAssistant,
aiohttp_client: ClientSessionGenerator,
authorization_data: dict[str, str],
) -> None:
"""Test logging in with exist user."""
client = await async_setup_auth(hass, aiohttp_client, setup_api=True)
@@ -261,6 +267,7 @@ async def test_login_exist_user(
"client_id": CLIENT_ID,
"handler": ["insecure_example", None],
"redirect_uri": CLIENT_REDIRECT_URI,
**authorization_data,
},
)
assert resp.status == HTTPStatus.OK
@@ -538,6 +545,12 @@ async def test_well_known_protected_resource_no_url(
},
"Message format incorrect",
),
(
{
"response_type": "token",
},
"Response type not supported",
),
],
ids=[
"method_without_challenge",
@@ -545,6 +558,7 @@ async def test_well_known_protected_resource_no_url(
"unsupported_plain_method",
"challenge_too_short",
"challenge_padded",
"unsupported_response_type",
],
)
async def test_login_flow_pkce_validation(