Replace a caller supplied Authorization header case-insensitively (#179887)

Co-authored-by: Claude <noreply@anthropic.com>

homeassistant/helpers/config_entry_oauth2_flow.py

@@ -19,9 +19,10 @@ import secrets
 import time
 from typing import Any, cast, override
 
-from aiohttp import ClientError, ClientResponseError, client, web
+from aiohttp import ClientError, ClientResponseError, client, hdrs, web
 from habluetooth import BluetoothServiceInfoBleak
 import jwt
+from multidict import CIMultiDict
 import voluptuous as vol
 from yarl import URL
 
@@ -791,16 +792,9 @@ async def async_oauth2_request(
     This method will not refresh tokens. Use OAuth2 session for that.
     """
     session = async_get_clientsession(hass)
-    headers = kwargs.pop("headers", {})
-    return await session.request(
-        method,
-        url,
-        **kwargs,
-        headers={
-            **headers,
-            "authorization": f"Bearer {token['access_token']}",
-        },
-    )
+    headers = CIMultiDict(kwargs.pop("headers", {}))
+    headers[hdrs.AUTHORIZATION] = f"Bearer {token['access_token']}"
+    return await session.request(method, url, **kwargs, headers=headers)
 
 
 @callback

tests/helpers/test_config_entry_oauth2_flow.py

@@ -8,6 +8,7 @@ from typing import Any
 from unittest.mock import AsyncMock, patch
 
 from aiohttp import ClientError
+from multidict import CIMultiDict
 import pytest
 
 from homeassistant import config_entries, data_entry_flow, setup
@@ -1431,3 +1432,33 @@ async def test_async_get_config_entry_implementation_missing_provider(
         await config_entry_oauth2_flow.async_get_config_entry_implementation(
             hass, config_entry
         )
+
+
+@pytest.mark.parametrize(
+    "header_name",
+    [
+        pytest.param("Authorization", id="canonical_casing"),
+        pytest.param("authorization", id="lowercase"),
+    ],
+)
+async def test_oauth2_request_replaces_caller_authorization_header(
+    hass: HomeAssistant,
+    aioclient_mock: AiohttpClientMocker,
+    header_name: str,
+) -> None:
+    """Test the token replaces a caller supplied Authorization header."""
+    aioclient_mock.post("https://example.com", status=201)
+
+    await config_entry_oauth2_flow.async_oauth2_request(
+        hass,
+        {"access_token": ACCESS_TOKEN_1},
+        "post",
+        "https://example.com",
+        headers={header_name: "Bearer caller supplied"},
+    )
+
+    assert len(aioclient_mock.mock_calls) == 1
+    headers = CIMultiDict(aioclient_mock.mock_calls[0][3])
+
+    # The token must not be sent as a second Authorization header
+    assert headers.getall("Authorization") == [f"Bearer {ACCESS_TOKEN_1}"]