Support Client ID Metadata Documents when verifying redirect_uri (#176286)

Co-authored-by: kingpanther13 <kingpanther13@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
kingpanther13
2026-08-19 17:10:06 -04:00
committed by GitHub
co-authored by kingpanther13 Claude Fable 5 Paulus Schoutsen
parent fc946f747e
commit f4303d3636
4 changed files with 611 additions and 18 deletions
+166 -12
View File
@@ -1,7 +1,9 @@
"""Helpers to resolve client ID/secret."""
from html.parser import HTMLParser
from http import HTTPStatus
from ipaddress import ip_address
import json
import logging
from typing import override
from urllib.parse import ParseResult, urljoin, urlparse
@@ -14,6 +16,9 @@ from homeassistant.util.network import is_local
_LOGGER = logging.getLogger(__name__)
# We limit reads of a client_id page to the first 10kB.
MAX_FETCH_BYTES = 10240
async def verify_redirect_uri(
hass: HomeAssistant, client_id: str, redirect_uri: str
@@ -24,7 +29,10 @@ async def verify_redirect_uri(
except ValueError:
return False
redirect_parts = _parse_url(redirect_uri)
try:
redirect_parts = _parse_url(redirect_uri)
except ValueError:
return False
# Verify redirect url and client url have same scheme and domain.
is_valid = (
@@ -53,7 +61,15 @@ async def verify_redirect_uri(
# IndieAuth 4.2.2 allows for redirect_uri to be on different domain
# but needs to be specified in link tag when fetching `client_id`.
redirect_uris = await fetch_redirect_uris(hass, client_id)
return redirect_uri in redirect_uris
if redirect_uri in redirect_uris:
return True
_LOGGER.debug(
"redirect_uri %s is not among the advertised redirect uris %s for client_id %s",
redirect_uri,
redirect_uris,
client_id,
)
return False
class LinkTagParser(HTMLParser):
@@ -63,7 +79,7 @@ class LinkTagParser(HTMLParser):
"""Initialize a link tag parser."""
super().__init__()
self.rel = rel
self.found: list[str | None] = []
self.found: list[str] = []
@override
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
@@ -73,48 +89,115 @@ class LinkTagParser(HTMLParser):
attributes: dict[str, str | None] = dict(attrs)
if attributes.get("rel") == self.rel:
self.found.append(attributes.get("href"))
# Skip tags with a missing or empty href: urljoin resolves those to
# the client_id URL itself instead of naming a redirect target.
if attributes.get("rel") == self.rel and (href := attributes.get("href")):
self.found.append(href)
def _reject_json_constant(constant: str) -> None:
"""Reject NaN/Infinity/-Infinity, which RFC 8259 does not allow."""
raise ValueError(f"Invalid JSON constant: {constant}")
def _is_valid_metadata_client_id(url: str) -> bool:
"""Validate a client_id URL for the metadata-document fallback.
The client identifier URL must be https with a path component and no
fragment (a bare trailing # counts as a fragment component). The remaining
client identifier rules are enforced upstream by _parse_client_id.
"""
try:
parts = urlparse(url)
# urlparse defers port validation until the attribute is accessed.
_ = parts.port
except ValueError:
return False
return parts.scheme == "https" and bool(parts.path) and "#" not in url
def _is_valid_metadata_redirect_uri(redirect_uri: str) -> bool:
"""Validate a client ID metadata document redirect_uris entry.
Entries must be absolute, fragment-free URIs: a non-empty scheme (so
private-use schemes like app:/callback stay valid) and no fragment per
RFC 6749 3.1.2 (a bare trailing # counts as a fragment component).
"""
try:
parts = urlparse(redirect_uri)
# urlparse defers port validation until the attribute is accessed.
_ = parts.port
except ValueError:
return False
return bool(parts.scheme) and "#" not in redirect_uri
async def fetch_redirect_uris(hass: HomeAssistant, url: str) -> list[str]:
"""Find link tag with redirect_uri values.
"""Find the redirect_uri values that a client_id advertises.
We support two formats, checked in this order:
IndieAuth 4.2.2
The client SHOULD publish one or more <link> tags or Link HTTP headers with
a rel attribute of redirect_uri at the client_id URL.
We limit to the first 10kB of the page.
OAuth Client ID Metadata Document
(draft-ietf-oauth-client-id-metadata-document)
The client_id URL returns a JSON document with a redirect_uris array. As we
advertise client_id_metadata_document_supported in the authorization server
metadata, we fall back to this format when no link tags are found.
We read roughly the first 10kB of the page and a fetch error yields no
redirect uris.
We do not implement extracting redirect uris from headers.
"""
parser = LinkTagParser("redirect_uri")
chunks = 0
body: bytes = b""
status: int | None = None
redirected = False
try:
async with (
aiohttp.ClientSession() as session,
session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp,
):
status = resp.status
redirected = bool(resp.history)
async for data in resp.content.iter_chunked(1024):
parser.feed(data.decode())
chunks += 1
body += data
if chunks == 10:
if len(body) >= MAX_FETCH_BYTES:
break
except TimeoutError:
_LOGGER.error("Timeout while looking up redirect_uri %s", url)
return []
except aiohttp.client_exceptions.ClientSSLError:
_LOGGER.error("SSL error while looking up redirect_uri %s", url)
return []
except aiohttp.client_exceptions.ClientOSError as ex:
_LOGGER.error("OS error while looking up redirect_uri %s: %s", url, ex.strerror)
return []
except aiohttp.client_exceptions.ClientConnectionError:
_LOGGER.error(
"Low level connection error while looking up redirect_uri %s", url
)
return []
except aiohttp.client_exceptions.ClientError:
_LOGGER.error("Unknown error while looking up redirect_uri %s", url)
return []
if redirect_uris := _parse_link_tag_redirect_uris(url, body):
return redirect_uris
return _parse_metadata_document_redirect_uris(url, body, status, redirected)
def _parse_link_tag_redirect_uris(url: str, body: bytes) -> list[str]:
"""Find <link rel="redirect_uri"> values in the client_id page body."""
parser = LinkTagParser("redirect_uri")
parser.feed(body.decode(errors="replace"))
# Authorization endpoints verifying that a redirect_uri is allowed for use
# by a client MUST look for an exact match of the given redirect_uri in the
@@ -123,6 +206,77 @@ async def fetch_redirect_uris(hass: HomeAssistant, url: str) -> list[str]:
return [urljoin(url, found) for found in parser.found]
def _parse_metadata_document_redirect_uris(
url: str, body: bytes, status: int | None, redirected: bool
) -> list[str]:
"""Parse the client_id page body as an OAuth Client ID Metadata Document.
Per draft-ietf-oauth-client-id-metadata-document the document only counts
when the client_id URL is https with a path and no fragment, the response
was a direct 200 (not redirected), the document's client_id round-trips,
and every redirect_uris entry is an absolute, fragment-free URI matched
exactly. The url and its document are client-controlled and fetched
unauthenticated, so rejections log at DEBUG (higher levels would be a
log-flood vector).
"""
# A body at the read cap may be truncated; a truncated prefix must not be
# trusted even if it happens to be parseable.
if (
len(body) >= MAX_FETCH_BYTES
or status != HTTPStatus.OK
or redirected
or not _is_valid_metadata_client_id(url)
):
_LOGGER.debug(
"Not treating %s as a client ID metadata document: body length %s,"
" status %s, redirected %s (client_id must be a fragment-free https"
" URL with a path)",
url,
len(body),
status,
redirected,
)
return []
try:
# Strict decode (RFC 8259 requires UTF-8): the link tag parser's
# lenient replacement decode would mask invalid bytes as U+FFFD.
document = json.loads(body.decode(), parse_constant=_reject_json_constant)
except UnicodeDecodeError:
_LOGGER.debug("Client ID metadata document at %s is not valid UTF-8", url)
return []
except ValueError:
_LOGGER.debug("Client ID metadata document at %s is not valid JSON", url)
return []
if not isinstance(document, dict):
_LOGGER.debug("Client ID metadata document at %s is not a JSON object", url)
return []
if document.get("client_id") != url:
_LOGGER.debug(
"Client ID metadata document at %s client_id does not match the"
" document URL",
url,
)
return []
# redirect_uris entries are returned unmodified for RFC 6749 exact matching
# rather than resolving relative references.
redirect_uris = document.get("redirect_uris")
if not isinstance(redirect_uris, list) or not all(
isinstance(redirect_uri, str) and _is_valid_metadata_redirect_uri(redirect_uri)
for redirect_uri in redirect_uris
):
_LOGGER.debug(
"Client ID metadata document at %s has missing or invalid redirect_uris",
url,
)
return []
return redirect_uris
def verify_client_id(client_id: str) -> bool:
"""Verify that the client id is valid."""
try:
+5 -6
View File
@@ -137,12 +137,11 @@ class WellKnownOAuthInfoView(HomeAssistantView):
"authorization_endpoint": f"{url_prefix}/auth/authorize",
"token_endpoint": f"{url_prefix}/auth/token",
"revocation_endpoint": f"{url_prefix}/auth/revoke",
# Home Assistant already accepts URL-based client_ids via
# IndieAuth without prior registration, which is compatible with
# draft-ietf-oauth-client-id-metadata-document. This flag
# advertises that support to encourage clients to use it. The
# metadata document is not actually fetched as IndieAuth doesn't
# require it.
# Home Assistant accepts URL-based client_ids via IndieAuth without
# prior registration, and discovers allowed redirect URIs from link
# tags or a Client ID Metadata Document served at the client_id URL.
# This flag advertises that support
# (draft-ietf-oauth-client-id-metadata-document).
"client_id_metadata_document_supported": True,
"response_types_supported": ["code"],
"service_documentation": (
+436
View File
@@ -1,8 +1,10 @@
"""Tests for the client validator."""
import asyncio
import json
from unittest.mock import patch
import aiohttp
import pytest
from homeassistant.components.auth import indieauth
@@ -167,6 +169,440 @@ async def test_find_link_tag_max_size(hass: HomeAssistant, mock_session) -> None
assert redirect_uris == ["http://127.0.0.1:8000/wine"]
async def test_find_link_tag_without_href(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test a redirect_uri link tag without a usable href is skipped."""
mock_session.get(
"http://127.0.0.1:8000",
text="""
<!doctype html>
<html>
<head>
<link rel="redirect_uri">
<link rel="redirect_uri" href="">
<link rel="redirect_uri" href="https://example.com/cb">
</head>
</html>
""",
)
redirect_uris = await indieauth.fetch_redirect_uris(hass, "http://127.0.0.1:8000")
assert redirect_uris == ["https://example.com/cb"]
async def test_fetch_redirect_uris_metadata_document(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test fetching redirect uris from a client id metadata document."""
mock_session.get(
"https://example.com/client",
text=json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": [
"https://example.com/callback",
"https://other.com/callback",
],
}
),
headers={"Content-Type": "application/json"},
)
redirect_uris = await indieauth.fetch_redirect_uris(
hass, "https://example.com/client"
)
assert redirect_uris == [
"https://example.com/callback",
"https://other.com/callback",
]
async def test_fetch_redirect_uris_metadata_document_text_plain(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test the metadata document is parsed regardless of content type."""
mock_session.get(
"https://example.com/client",
text=json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["https://example.com/callback"],
}
),
headers={"Content-Type": "text/plain"},
)
redirect_uris = await indieauth.fetch_redirect_uris(
hass, "https://example.com/client"
)
assert redirect_uris == ["https://example.com/callback"]
async def test_fetch_redirect_uris_link_tag_precedence(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test link tags take precedence over metadata document parsing."""
mock_session.get(
"http://127.0.0.1:8000",
text="""
<!doctype html>
<html>
<head>
<link rel="redirect_uri" href="hass://oauth2_redirect">
</head>
<body>
{"redirect_uris": ["https://example.com/should-be-ignored"]}
</body>
</html>
""",
)
redirect_uris = await indieauth.fetch_redirect_uris(hass, "http://127.0.0.1:8000")
assert redirect_uris == ["hass://oauth2_redirect"]
@pytest.mark.parametrize(
"text",
[
pytest.param("this is neither json nor html", id="not-json-not-html"),
pytest.param('["https://example.com/callback"]', id="json-array"),
pytest.param("42", id="json-scalar"),
pytest.param(
json.dumps({"redirect_uris": ["https://example.com/callback"]}),
id="missing-client-id",
),
pytest.param(
json.dumps({"client_id": "https://example.com/client"}),
id="missing-redirect-uris",
),
pytest.param(
json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": [],
}
),
id="empty-redirect-uris",
),
pytest.param(
json.dumps(
{
"client_id": "https://other.example/client",
"redirect_uris": ["https://example.com/callback"],
}
),
id="client-id-mismatch",
),
pytest.param(
json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": "https://example.com/callback",
}
),
id="redirect-uris-not-list",
),
pytest.param(
json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["https://example.com/callback", 123],
}
),
id="redirect-uris-non-string-entry",
),
pytest.param(
json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["/callback"],
}
),
id="redirect-uris-relative-entry",
),
pytest.param(
json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["https://example.com/callback#fragment"],
}
),
id="redirect-uris-fragment-entry",
),
pytest.param(
json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["https://["],
}
),
id="redirect-uris-unparsable-entry",
),
pytest.param(
json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["https://example.com/callback#"],
}
),
id="redirect-uris-empty-fragment-entry",
),
pytest.param(
json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["https://example.com:not-a-port/callback"],
}
),
id="redirect-uris-invalid-port-entry",
),
pytest.param(
'{"client_id": "https://example.com/client",'
' "redirect_uris": ["https://example.com/callback"], "x": NaN}',
id="json-nan-constant",
),
],
)
async def test_fetch_redirect_uris_metadata_document_invalid(
hass: HomeAssistant, mock_session: AiohttpClientMocker, text: str
) -> None:
"""Test that invalid metadata documents yield no redirect uris."""
mock_session.get(
"https://example.com/client",
text=text,
headers={"Content-Type": "application/json"},
)
assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
assert not await indieauth.verify_redirect_uri(
hass, "https://example.com/client", "https://other.com/callback"
)
async def test_verify_redirect_uri_metadata_document(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test verifying a cross-origin redirect uri from a metadata document."""
client_id = "https://example.com/client"
mock_session.get(
client_id,
text=json.dumps(
{
"client_id": client_id,
"redirect_uris": ["https://other.com/callback"],
}
),
headers={"Content-Type": "application/json"},
)
assert await indieauth.verify_redirect_uri(
hass, client_id, "https://other.com/callback"
)
assert not await indieauth.verify_redirect_uri(
hass, client_id, "https://other.com/not-listed"
)
async def test_verify_redirect_uri_unparsable(hass: HomeAssistant) -> None:
"""Test an unparsable requested redirect uri is rejected without raising."""
assert not await indieauth.verify_redirect_uri(
hass, "https://example.com/client", "https://["
)
async def test_fetch_redirect_uris_metadata_document_invalid_utf8(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test a metadata document with invalid UTF-8 is rejected."""
mock_session.get(
"https://example.com/client",
content=(
b'{"client_id": "https://example.com/client",'
b' "redirect_uris": ["https://other.com/callback"], "note": "\xff"}'
),
headers={"Content-Type": "application/json"},
)
assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
@pytest.mark.parametrize(
"client_id",
[
pytest.param("https://example.com", id="no-path"),
pytest.param("https://example.com/client#", id="empty-fragment"),
],
)
async def test_fetch_redirect_uris_metadata_document_invalid_client_id(
hass: HomeAssistant, mock_session: AiohttpClientMocker, client_id: str
) -> None:
"""Test client ids violating the metadata document URL rules are ignored."""
mock_session.get(
client_id,
text=json.dumps(
{
"client_id": client_id,
"redirect_uris": ["https://other.com/callback"],
}
),
headers={"Content-Type": "application/json"},
)
assert await indieauth.fetch_redirect_uris(hass, client_id) == []
async def test_fetch_redirect_uris_metadata_document_not_ok(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test a metadata document not served with 200 OK is ignored."""
mock_session.get(
"https://example.com/client",
text=json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["https://example.com/callback"],
}
),
status=404,
headers={"Content-Type": "application/json"},
)
assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
async def test_fetch_redirect_uris_metadata_document_http_scheme(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test a metadata document served over http is ignored."""
client_id = "http://example.com/client"
mock_session.get(
client_id,
text=json.dumps(
{
"client_id": client_id,
"redirect_uris": ["https://other.com/callback"],
}
),
headers={"Content-Type": "application/json"},
)
assert await indieauth.fetch_redirect_uris(hass, client_id) == []
assert not await indieauth.verify_redirect_uri(
hass, client_id, "https://other.com/callback"
)
async def test_fetch_redirect_uris_metadata_document_redirected(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test a metadata document reached via a redirect is ignored."""
mock_session.get(
"https://example.com/client",
text=json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["https://example.com/callback"],
}
),
headers={"Content-Type": "application/json"},
history=(object(),),
)
assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
async def test_fetch_redirect_uris_metadata_document_private_use_scheme(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test a private-use scheme redirect uri is accepted as an absolute URI."""
mock_session.get(
"https://example.com/client",
text=json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["app:/oauth-callback"],
}
),
headers={"Content-Type": "application/json"},
)
assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == [
"app:/oauth-callback"
]
async def test_fetch_redirect_uris_metadata_document_oversized(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test a document past the 10kB cap is rejected as an incomplete read."""
mock_session.get(
"https://example.com/client",
text=json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": ["https://example.com/callback"],
"padding": "x" * 11000,
}
),
headers={"Content-Type": "application/json"},
)
assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
async def test_fetch_redirect_uris_metadata_document_exactly_at_cap(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test a document of exactly the read cap is rejected as possibly truncated."""
document = {
"client_id": "https://example.com/client",
"redirect_uris": ["https://other.com/callback"],
"padding": "",
}
document["padding"] = "x" * (10240 - len(json.dumps(document)))
text = json.dumps(document)
assert len(text) == 10240
mock_session.get(
"https://example.com/client",
text=text,
headers={"Content-Type": "application/json"},
)
assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
async def test_fetch_redirect_uris_metadata_document_at_cap_ineligible(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test a valid document that reaches the 10kB cap is ineligible."""
mock_session.get(
"https://example.com/client",
text=json.dumps(
{
"client_id": "https://example.com/client",
"redirect_uris": [
f"https://example.com/callback/{index}" for index in range(400)
],
}
),
headers={"Content-Type": "application/json"},
)
assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
async def test_fetch_redirect_uris_network_error(
hass: HomeAssistant, mock_session: AiohttpClientMocker
) -> None:
"""Test a network error yields no redirect uris without raising."""
mock_session.get("https://example.com/client", exc=aiohttp.ClientError())
assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
@pytest.mark.parametrize(
"client_id",
["https://home-assistant.io/android", "https://home-assistant.io/iOS"],
+4
View File
@@ -64,6 +64,7 @@ class AiohttpClientMocker:
side_effect=None,
closing=None,
timeout=None,
history=(),
):
"""Mock a request."""
if not isinstance(url, RETYPE):
@@ -83,6 +84,7 @@ class AiohttpClientMocker:
headers=headers,
side_effect=side_effect,
closing=closing,
history=history,
)
self._mocks.append(resp)
return resp
@@ -185,6 +187,7 @@ class AiohttpClientMockResponse:
headers=None,
side_effect=None,
closing=None,
history=(),
) -> None:
"""Initialize a fake response."""
if json is not None:
@@ -197,6 +200,7 @@ class AiohttpClientMockResponse:
self.method = method
self._url = url
self.status = status
self.history = history
self._response = response
self.exc = exc
self.side_effect = side_effect