feat: add OAuth support for cloud APIs (#563)

* feat: add OAuth support for cloud APIs

* test: cover OAuth device error paths

* refactor: simplify encryption key delegation

* Revert "refactor: simplify encryption key delegation"

This reverts commit a62e78cbf0cc305c61c1d488cd1af9b1199199b0.

* fix: address OAuth review feedback

* fix: address OAuth review follow-ups

README.md

@@ -10,6 +10,57 @@ source .venv/bin/activate
 pip install .
 ```
 
+## OAuth account access
+
+pySwitchbot provides helpers for SwitchBot's authorization-code flow. The
+calling application supplies a SwitchBot-issued client ID and its registered
+redirect URI; neither value is tied to Home Assistant or embedded in the
+library.
+
+SwitchBot currently treats these integrations as public clients: the token
+request does not use a client secret, and the authorization server does not
+support PKCE. The caller must generate an unpredictable, single-use `state`,
+store it for the duration of the flow, and reject callbacks whose state does
+not match. State protects the callback from request forgery but does not
+replace PKCE.
+
+```python
+import secrets
+
+from switchbot import (
+    build_oauth_authorize_url,
+    exchange_oauth_code,
+    fetch_cloud_devices_by_token,
+)
+
+state = secrets.token_urlsafe(32)
+authorize_url = build_oauth_authorize_url(client_id, redirect_uri, state)
+
+# Store state before sending the user to authorize_url. On callback:
+if callback_state != state:
+    raise ValueError("OAuth state mismatch")
+
+token = await exchange_oauth_code(
+    session,
+    client_id,
+    redirect_uri,
+    authorization_code,
+)
+devices = await fetch_cloud_devices_by_token(session, token["access_token"])
+```
+
+The client ID and exact HTTPS redirect URI must be registered with SwitchBot;
+arbitrary values and wildcard redirect URIs will not work. `exchange_oauth_code`
+returns the provider's token mapping after validating the access token and
+normalizing `expires_in` to an integer. The access token can then be passed to
+`fetch_cloud_devices_by_token` or
+`SwitchbotEncryptedDevice.async_retrieve_encryption_key_by_token`.
+
+HTTP 401 and 403 responses from the SwitchBot account API raise
+`SwitchbotAuthenticationError`. Other API failures raise `SwitchbotApiError`,
+while transport and availability failures raise
+`SwitchbotAccountConnectionError`.
+
 ## Obtaining encryption key for Switchbot Locks
 
 Using the script `scripts/get_encryption_key.py` you can manually obtain locks encryption key.
@@ -41,11 +92,11 @@ password. The most common failures are account-side, not bugs in this library:
     email/password account.
   - The username is an email but the account is registered to a phone number
     (or vice versa). Use the exact identifier you log in with.
-- **`Failed to retrieve encryption key from SwitchBot Account: ...`** —
-  authentication succeeded but the key could not be read. Usually the account
-  is not the device **owner**: keys are only returned to the owning account,
-  not to shared/family members. Retrieve the key from the owner account, or
-  transfer ownership in the app.
+- **`..., status code: 190`** (`SwitchbotApiError`) — authentication succeeded
+  but the key could not be read. Usually the account is not the device
+  **owner**: keys are only returned to the owning account, not to shared/family
+  members. Retrieve the key from the owner account, or transfer ownership in
+  the app.
 
 The key only needs to be fetched once; store the `key_id` and encryption key
 and reuse them — there is no need to call the script on every connection.

switchbot/__init__.py

@@ -47,6 +47,7 @@ from .devices.device import (
     SwitchbotEncryptedDevice,
     SwitchbotOperationError,
     fetch_cloud_devices,
+    fetch_cloud_devices_by_token,
 )
 from .devices.evaporative_humidifier import SwitchbotEvaporativeHumidifier
 from .devices.fan import (
@@ -79,8 +80,18 @@ from .devices.universal_remote import SwitchbotUniversalRemote
 from .devices.vacuum import SwitchbotVacuum
 from .discovery import GetSwitchbotDevices
 from .models import SwitchBotAdvertisement
+from .oauth import (
+    OAUTH_AUTHORIZE_URL,
+    OAUTH_SCOPE,
+    OAUTH_TOKEN_URL,
+    build_oauth_authorize_url,
+    exchange_oauth_code,
+)
 
 __all__ = [
+    "OAUTH_AUTHORIZE_URL",
+    "OAUTH_SCOPE",
+    "OAUTH_TOKEN_URL",
     "AirPurifierMode",
     "AirQualityLevel",
     "BulbColorMode",
@@ -142,9 +153,12 @@ __all__ = [
     "SwitchbotUniversalRemote",
     "SwitchbotVacuum",
     "VerticalOscillationAngle",
+    "build_oauth_authorize_url",
     "close_stale_connections",
     "close_stale_connections_by_address",
+    "exchange_oauth_code",
     "fetch_cloud_devices",
+    "fetch_cloud_devices_by_token",
     "get_device",
     "parse_advertisement_data",
 ]

switchbot/devices/device.py

@@ -39,15 +39,25 @@ from ..const import (
 from ..discovery import GetSwitchbotDevices
 from ..helpers import create_background_task
 from ..models import SwitchBotAdvertisement
-from ..utils import format_mac_upper
+from ..utils import extract_request_id, format_mac_upper
 
 _LOGGER = logging.getLogger(__name__)
 
 
+def _masked_device_id(device_id: str) -> str:
+    """Mask a device identifier while retaining a useful suffix."""
+    normalized = device_id.replace(":", "").replace("-", "").upper()
+    if not normalized:
+        return "unknown"
+    return f"****{normalized[-4:]}"
+
+
 def _extract_region(userinfo: dict[str, Any]) -> str:
     """Extract region from user info, defaulting to 'us'."""
-    if "botRegion" in userinfo and userinfo["botRegion"] != "":
-        return userinfo["botRegion"]
+    region = userinfo.get("botRegion")
+    if isinstance(region, str) and region:
+        return region
+    _LOGGER.warning("SwitchBot account region missing; defaulting to us")
     return "us"
 
 
@@ -265,6 +275,10 @@ class SwitchbotBaseDevice:
             return await cls.api_request(
                 session, "account", "account/api/v1/user/userinfo", {}, auth_headers
             )
+        except SwitchbotAuthenticationError:
+            raise
+        except SwitchbotApiError:
+            raise
         except Exception as err:
             raise SwitchbotAccountConnectionError(
                 f"Failed to retrieve SwitchBot Account user details: {err}"
@@ -308,8 +322,45 @@ class SwitchbotBaseDevice:
         except Exception as err:
             raise SwitchbotAuthenticationError(f"Authentication failed: {err}") from err
 
+        return await cls._async_get_devices(session, auth_headers)
+
+    @classmethod
+    async def get_devices_by_token(
+        cls,
+        session: aiohttp.ClientSession,
+        access_token: str,
+    ) -> dict[str, SwitchbotModel]:
+        """Get devices from SwitchBot API using an OAuth access token."""
+        started = time.monotonic()
+        _LOGGER.debug("Retrieving SwitchBot cloud devices using an OAuth token")
+        try:
+            devices = await cls._async_get_devices(
+                session, {"authorization": access_token}
+            )
+        except Exception:
+            _LOGGER.debug(
+                "SwitchBot OAuth cloud device retrieval failed; duration_ms=%s",
+                round((time.monotonic() - started) * 1000),
+            )
+            raise
+        _LOGGER.debug(
+            "SwitchBot OAuth cloud device retrieval finished; supported_devices=%s "
+            "duration_ms=%s",
+            len(devices),
+            round((time.monotonic() - started) * 1000),
+        )
+        return devices
+
+    @classmethod
+    async def _async_get_devices(
+        cls,
+        session: aiohttp.ClientSession,
+        auth_headers: dict[str, str],
+    ) -> dict[str, SwitchbotModel]:
+        """Get devices from SwitchBot API using authenticated headers."""
         userinfo = await cls._async_get_user_info(session, auth_headers)
         region = _extract_region(userinfo)
+        _LOGGER.debug("SwitchBot account region resolved to %s", region)
 
         try:
             device_info = await cls.api_request(
@@ -321,12 +372,21 @@ class SwitchbotBaseDevice:
                 },
                 auth_headers,
             )
+        except SwitchbotAuthenticationError:
+            raise
+        except SwitchbotApiError:
+            raise
         except Exception as err:
             raise SwitchbotAccountConnectionError(
                 f"Failed to retrieve devices from SwitchBot Account: {err}"
             ) from err
 
-        items: list[dict[str, Any]] = device_info["Items"]
+        items = device_info.get("Items")
+        if not isinstance(items, list) or not all(
+            isinstance(item, dict) for item in items
+        ):
+            raise SwitchbotApiError("Invalid device response from SwitchBot API")
+        _LOGGER.debug("SwitchBot cloud API returned %s device records", len(items))
         mac_to_model: dict[str, SwitchbotModel] = {}
 
         for item in items:
@@ -360,6 +420,7 @@ class SwitchbotBaseDevice:
                     item,
                 )
 
+        _LOGGER.debug("Mapped %s supported SwitchBot cloud devices", len(mac_to_model))
         return mac_to_model
 
     @classmethod
@@ -372,18 +433,46 @@ class SwitchbotBaseDevice:
         headers: dict | None = None,
     ) -> dict:
         url = f"https://{subdomain}.{SWITCHBOT_APP_API_BASE_URL}/{path}"
+        started = time.monotonic()
+        _LOGGER.debug("Requesting SwitchBot API endpoint %s", url)
         async with session.post(
             url,
             json=data,
             headers=headers,
             timeout=aiohttp.ClientTimeout(total=10),
         ) as result:
+            _LOGGER.debug(
+                "SwitchBot API endpoint %s returned HTTP status %s; duration_ms=%s "
+                "request_id=%s",
+                url,
+                result.status,
+                round((time.monotonic() - started) * 1000),
+                extract_request_id(result.headers) or "unavailable",
+            )
+            if result.status in (401, 403):
+                raise SwitchbotAuthenticationError(
+                    "Authentication rejected by SwitchBot API"
+                )
             if result.status > 299:
                 raise SwitchbotApiError(
                     f"Unexpected status code returned by SwitchBot API: {result.status}"
                 )
 
             response = await result.json()
+            body = response.get("body")
+            body_fields: list[str] | str = (
+                sorted(body) if isinstance(body, dict) else type(body).__name__
+            )
+            _LOGGER.debug(
+                (
+                    "SwitchBot API endpoint %s returned API status %s; "
+                    "response fields=%s; body fields=%s"
+                ),
+                url,
+                response.get("statusCode"),
+                sorted(response),
+                body_fields,
+            )
             if response["statusCode"] != 100:
                 raise SwitchbotApiError(
                     f"{response['message']}, status code: {response['statusCode']}"
@@ -1052,17 +1141,69 @@ class SwitchbotEncryptedDevice(SwitchbotDevice):
         password: str,
     ) -> dict:
         """Retrieve lock key from internal SwitchBot API."""
-        device_mac = device_mac.replace(":", "").replace("-", "").upper()
-
         try:
             auth_result = await cls._get_auth_result(session, username, password)
             auth_headers = {"authorization": auth_result["access_token"]}
         except Exception as err:
             raise SwitchbotAuthenticationError(f"Authentication failed: {err}") from err
 
+        return await cls._async_retrieve_encryption_key(
+            session, device_mac, auth_headers
+        )
+
+    @classmethod
+    async def async_retrieve_encryption_key_by_token(
+        cls,
+        session: aiohttp.ClientSession,
+        device_mac: str,
+        access_token: str,
+    ) -> dict:
+        """Retrieve an encryption key using an OAuth access token."""
+        started = time.monotonic()
+        masked_device = _masked_device_id(device_mac)
+        _LOGGER.debug(
+            "Retrieving a SwitchBot encryption key using an OAuth token; device=%s",
+            masked_device,
+        )
+        try:
+            key_details = await cls._async_retrieve_encryption_key(
+                session, device_mac, {"authorization": access_token}
+            )
+        except Exception:
+            _LOGGER.debug(
+                "SwitchBot OAuth encryption key retrieval failed; device=%s "
+                "duration_ms=%s",
+                masked_device,
+                round((time.monotonic() - started) * 1000),
+            )
+            raise
+        _LOGGER.debug(
+            "SwitchBot OAuth encryption key retrieval finished; device=%s "
+            "duration_ms=%s",
+            masked_device,
+            round((time.monotonic() - started) * 1000),
+        )
+        return key_details
+
+    @classmethod
+    async def _async_retrieve_encryption_key(
+        cls,
+        session: aiohttp.ClientSession,
+        device_mac: str,
+        auth_headers: dict[str, str],
+    ) -> dict:
+        """Retrieve an encryption key using authenticated headers."""
+        device_mac = device_mac.replace(":", "").replace("-", "").upper()
+
         userinfo = await cls._async_get_user_info(session, auth_headers)
         region = _extract_region(userinfo)
+        masked_device = _masked_device_id(device_mac)
 
+        _LOGGER.debug(
+            "SwitchBot encryption key account region resolved; region=%s device=%s",
+            region,
+            masked_device,
+        )
         try:
             device_info = await cls.api_request(
                 session,
@@ -1074,16 +1215,33 @@ class SwitchbotEncryptedDevice(SwitchbotDevice):
                 },
                 auth_headers,
             )
-
-            return {
-                "key_id": device_info["communicationKey"]["keyId"],
-                "encryption_key": device_info["communicationKey"]["key"],
-            }
+        except SwitchbotAuthenticationError:
+            raise
+        except SwitchbotApiError:
+            raise
         except Exception as err:
             raise SwitchbotAccountConnectionError(
                 f"Failed to retrieve encryption key from SwitchBot Account: {err}"
             ) from err
 
+        communication_key = device_info.get("communicationKey")
+        if not isinstance(communication_key, dict):
+            raise SwitchbotApiError(
+                "Invalid encryption key response from SwitchBot API"
+            )
+        key_id = communication_key.get("keyId")
+        encryption_key = communication_key.get("key")
+        if not isinstance(key_id, str) or not isinstance(encryption_key, str):
+            raise SwitchbotApiError(
+                "Invalid encryption key response from SwitchBot API"
+            )
+
+        _LOGGER.debug(
+            "SwitchBot encryption key retrieved successfully; device=%s",
+            masked_device,
+        )
+        return {"key_id": key_id, "encryption_key": encryption_key}
+
     @classmethod
     async def verify_encryption_key(
         cls,
@@ -1336,3 +1494,11 @@ async def fetch_cloud_devices(
     """Fetch devices from SwitchBot API and return MAC to model mapping."""
     # Get devices from the API (which also populates the cache)
     return await SwitchbotBaseDevice.get_devices(session, username, password)
+
+
+async def fetch_cloud_devices_by_token(
+    session: aiohttp.ClientSession,
+    access_token: str,
+) -> dict[str, SwitchbotModel]:
+    """Fetch devices from SwitchBot API using an OAuth access token."""
+    return await SwitchbotBaseDevice.get_devices_by_token(session, access_token)

switchbot/oauth.py

@@ -0,0 +1,183 @@
+"""OAuth helpers for the SwitchBot account API."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Mapping
+from time import monotonic
+from typing import Any
+from urllib.parse import urlencode, urlsplit
+
+import aiohttp
+
+from .const import (
+    SwitchbotAccountConnectionError,
+    SwitchbotApiError,
+    SwitchbotAuthenticationError,
+)
+from .utils import extract_request_id
+
+_LOGGER = logging.getLogger(__name__)
+
+OAUTH_AUTHORIZE_URL = "https://sp.oauth.switchbot.net"
+OAUTH_TOKEN_URL = "https://account.api.switchbot.net/merchant/v1/oauth/token"
+OAUTH_SCOPE = "api_login"
+
+
+def _oauth_error_field(
+    error_data: Any, field: str, authorization_code: str
+) -> str | None:
+    """Return a bounded OAuth error field with the authorization code redacted."""
+    if not isinstance(error_data, Mapping):
+        return None
+    value = error_data.get(field)
+    if not isinstance(value, str):
+        return None
+    value = " ".join(value.split())
+    if authorization_code:
+        value = value.replace(authorization_code, "<redacted>")
+    return value[:256] or None
+
+
+def _raise_for_oauth_error(status: int, error: str | None, error_suffix: str) -> None:
+    """Raise the appropriate exception for an OAuth error response."""
+    if status in (401, 403) or (
+        status == 400 and error in {"invalid_client", "invalid_grant"}
+    ):
+        raise SwitchbotAuthenticationError(
+            f"SwitchBot OAuth token request rejected ({status}){error_suffix}"
+        )
+    if 400 <= status < 500 and status != 429:
+        raise SwitchbotApiError(
+            f"SwitchBot OAuth token request failed ({status}){error_suffix}"
+        )
+    if status == 429 or status >= 500:
+        raise SwitchbotAccountConnectionError(
+            f"SwitchBot OAuth token service unavailable ({status}){error_suffix}"
+        )
+
+
+def build_oauth_authorize_url(
+    client_id: str,
+    redirect_uri: str,
+    state: str,
+) -> str:
+    """Build a SwitchBot OAuth authorization URL."""
+    _LOGGER.debug(
+        "Building SwitchBot OAuth authorization request; authorize_host=%s "
+        "redirect_host=%s",
+        urlsplit(OAUTH_AUTHORIZE_URL).hostname,
+        urlsplit(redirect_uri).hostname,
+    )
+    query = urlencode(
+        {
+            "client_id": client_id,
+            "redirect_uri": redirect_uri,
+            "response_type": "code",
+            "scope": OAUTH_SCOPE,
+            "state": state,
+        }
+    )
+    return f"{OAUTH_AUTHORIZE_URL}?{query}"
+
+
+async def exchange_oauth_code(
+    session: aiohttp.ClientSession,
+    client_id: str,
+    redirect_uri: str,
+    code: str,
+) -> dict[str, Any]:
+    """Exchange an OAuth authorization code for a SwitchBot access token."""
+    started = monotonic()
+    _LOGGER.debug(
+        "Exchanging SwitchBot OAuth authorization code; token_host=%s",
+        urlsplit(OAUTH_TOKEN_URL).hostname,
+    )
+    error: str | None = None
+    error_description: str | None = None
+    token_data: Any = None
+    try:
+        async with session.post(
+            OAUTH_TOKEN_URL,
+            data={
+                "code": code,
+                "client_id": client_id,
+                "grant_type": "authorization_code",
+                "redirect_uri": redirect_uri,
+            },
+            timeout=aiohttp.ClientTimeout(total=10),
+        ) as response:
+            status = response.status
+            _LOGGER.debug(
+                "SwitchBot OAuth token endpoint returned HTTP status %s; "
+                "duration_ms=%s request_id=%s",
+                status,
+                round((monotonic() - started) * 1000),
+                extract_request_id(response.headers) or "unavailable",
+            )
+            if status >= 400:
+                try:
+                    error_data = await response.json()
+                except (aiohttp.ClientError, ValueError, TypeError) as err:
+                    _LOGGER.debug(
+                        "SwitchBot OAuth token error response could not be parsed; "
+                        "error_type=%s",
+                        type(err).__name__,
+                    )
+                    error_data = None
+                error = _oauth_error_field(error_data, "error", code)
+                error_description = _oauth_error_field(
+                    error_data, "error_description", code
+                )
+                _LOGGER.debug(
+                    "SwitchBot OAuth token endpoint returned an error response; "
+                    "error=%s error_description=%s",
+                    error or "unavailable",
+                    error_description or "unavailable",
+                )
+            else:
+                try:
+                    token_data = await response.json()
+                except (aiohttp.ClientError, ValueError, TypeError) as err:
+                    raise SwitchbotApiError(
+                        "Invalid response from SwitchBot OAuth token API"
+                    ) from err
+    except (aiohttp.ClientError, TimeoutError) as err:
+        raise SwitchbotAccountConnectionError(
+            f"Failed to connect to SwitchBot OAuth token API: {err}"
+        ) from err
+
+    error_detail = ": ".join(
+        value for value in (error, error_description) if value is not None
+    )
+    error_suffix = f": {error_detail}" if error_detail else ""
+    _raise_for_oauth_error(status, error, error_suffix)
+    if not isinstance(token_data, dict):
+        raise SwitchbotApiError("Invalid response from SwitchBot OAuth token API")
+
+    token: dict[str, Any] = token_data.copy()
+    _LOGGER.debug("SwitchBot OAuth token response fields: %s", sorted(token))
+    access_token = token.get("access_token")
+    expires_in = token.get("expires_in")
+    if (
+        not isinstance(access_token, str)
+        or not access_token
+        or isinstance(expires_in, bool)
+        or not isinstance(expires_in, int | str)
+    ):
+        raise SwitchbotApiError("Invalid token data from SwitchBot OAuth token API")
+    try:
+        normalized_expires_in = int(expires_in)
+    except ValueError as err:
+        raise SwitchbotApiError(
+            "Invalid token data from SwitchBot OAuth token API"
+        ) from err
+    token["expires_in"] = normalized_expires_in
+    _LOGGER.debug(
+        "SwitchBot OAuth token response validated; expires_in=%s "
+        "refresh_token_present=%s refresh_expires_in_present=%s",
+        normalized_expires_in,
+        bool(token.get("refresh_token")),
+        "refresh_expires_in" in token,
+    )
+    return token

switchbot/utils.py

@@ -1,7 +1,23 @@
 """Utility functions for switchbot."""
 
+from collections.abc import Mapping
 from functools import lru_cache
 
+_REQUEST_ID_HEADERS = ("x-request-id", "x-amzn-requestid", "cf-ray")
+
+
+def extract_request_id(headers: Mapping[str, str]) -> str | None:
+    """Extract a provider request identifier for log correlation."""
+    normalized_headers = {name.lower(): value for name, value in headers.items()}
+    return next(
+        (
+            value
+            for name in _REQUEST_ID_HEADERS
+            if (value := normalized_headers.get(name))
+        ),
+        None,
+    )
+
 
 @lru_cache(maxsize=512)
 def format_mac_upper(mac: str) -> str:

tests/test_device.py

@@ -9,17 +9,20 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import aiohttp
 import pytest
 
-from switchbot import fetch_cloud_devices
+from switchbot import fetch_cloud_devices, fetch_cloud_devices_by_token
 from switchbot.adv_parser import _MODEL_TO_MAC_CACHE, populate_model_to_mac_cache
 from switchbot.const import (
     SwitchbotAccountConnectionError,
+    SwitchbotApiError,
     SwitchbotAuthenticationError,
     SwitchbotModel,
 )
 from switchbot.devices.device import (
     SwitchbotBaseDevice,
     SwitchbotDevice,
+    SwitchbotEncryptedDevice,
     _extract_region,
+    _masked_device_id,
 )
 
 from .test_adv_parser import generate_ble_device
@@ -150,7 +153,8 @@ async def test_get_devices(
 
         # Check that unknown model was logged
         assert "Unknown model WoUnknown for device DD:EE:FF:00:11:22" in caplog.text
-        assert "extra_field" in caplog.text  # Full item should be logged
+        assert "extra_field" in caplog.text
+        assert "extra_value" in caplog.text
 
 
 @pytest.mark.asyncio
@@ -306,6 +310,462 @@ async def test_fetch_cloud_devices(
         assert mock_populate_cache.call_count == 3
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("region", ["us", "eu", "jp"])
+async def test_fetch_cloud_devices_by_token(
+    mock_user_info: dict[str, Any],
+    mock_device_response: dict[str, Any],
+    region: str,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test fetching cloud devices with an OAuth access token."""
+    caplog.set_level(logging.DEBUG, logger="switchbot.devices.device")
+    with (
+        patch.object(SwitchbotBaseDevice, "_get_auth_result") as mock_get_auth_result,
+        patch.object(
+            SwitchbotBaseDevice,
+            "_async_get_user_info",
+            return_value={**mock_user_info, "botRegion": region},
+        ) as mock_get_user_info,
+        patch.object(
+            SwitchbotBaseDevice,
+            "api_request",
+            return_value=mock_device_response,
+        ) as mock_api_request,
+        patch(
+            "switchbot.devices.device.populate_model_to_mac_cache"
+        ) as mock_populate_cache,
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        result = await fetch_cloud_devices_by_token(session, "oauth-access-token")
+
+    mock_get_auth_result.assert_not_called()
+    mock_get_user_info.assert_awaited_once_with(
+        session, {"authorization": "oauth-access-token"}
+    )
+    mock_api_request.assert_awaited_once_with(
+        session,
+        f"wonderlabs.{region}",
+        "wonder/device/v3/getdevice",
+        {"required_type": "All"},
+        {"authorization": "oauth-access-token"},
+    )
+    assert result["AA:BB:CC:DD:EE:FF"] == SwitchbotModel.BOT
+    assert mock_populate_cache.call_count == 3
+    assert "retrieval finished; supported_devices=3 duration_ms=" in caplog.text
+    assert f"region resolved to {region}" in caplog.text
+    assert "oauth-access-token" not in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_fetch_cloud_devices_by_token_connection_error(
+    mock_user_info: dict[str, Any],
+) -> None:
+    """Test an API error while fetching cloud devices with an OAuth token."""
+    with (
+        patch.object(
+            SwitchbotBaseDevice,
+            "_async_get_user_info",
+            return_value=mock_user_info,
+        ),
+        patch.object(
+            SwitchbotBaseDevice,
+            "api_request",
+            side_effect=Exception("Network error"),
+        ),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(
+            SwitchbotAccountConnectionError, match="Failed to retrieve devices"
+        ):
+            await fetch_cloud_devices_by_token(session, "oauth-access-token")
+
+
+@pytest.mark.asyncio
+async def test_fetch_cloud_devices_by_token_authentication_error() -> None:
+    """Test an authentication error while fetching devices with an OAuth token."""
+    with patch.object(
+        SwitchbotBaseDevice,
+        "_async_get_user_info",
+        side_effect=SwitchbotAuthenticationError("invalid token"),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(SwitchbotAuthenticationError, match="invalid token"):
+            await fetch_cloud_devices_by_token(session, "oauth-access-token")
+
+
+@pytest.mark.asyncio
+async def test_get_devices_preserves_authentication_error_after_user_info(
+    mock_user_info: dict[str, Any],
+) -> None:
+    """Test device retrieval preserves authentication errors."""
+    with (
+        patch.object(
+            SwitchbotBaseDevice,
+            "_async_get_user_info",
+            return_value=mock_user_info,
+        ),
+        patch.object(
+            SwitchbotBaseDevice,
+            "api_request",
+            side_effect=SwitchbotAuthenticationError("expired token"),
+        ),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(SwitchbotAuthenticationError, match="expired token"):
+            await fetch_cloud_devices_by_token(session, "oauth-access-token")
+
+
+@pytest.mark.asyncio
+async def test_get_devices_preserves_api_error_after_user_info(
+    mock_user_info: dict[str, Any],
+) -> None:
+    """Test device retrieval preserves API errors."""
+    with (
+        patch.object(
+            SwitchbotBaseDevice,
+            "_async_get_user_info",
+            return_value=mock_user_info,
+        ),
+        patch.object(
+            SwitchbotBaseDevice,
+            "api_request",
+            side_effect=SwitchbotApiError("API error"),
+        ),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(SwitchbotApiError, match="API error"):
+            await fetch_cloud_devices_by_token(session, "oauth-access-token")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "device_info",
+    [
+        pytest.param({}, id="missing-items"),
+        pytest.param({"Items": None}, id="invalid-items"),
+        pytest.param({"Items": {}}, id="items-not-list"),
+        pytest.param({"Items": ["invalid"]}, id="invalid-item"),
+    ],
+)
+async def test_get_devices_rejects_invalid_response(
+    mock_user_info: dict[str, Any], device_info: dict[str, Any]
+) -> None:
+    """Test malformed device responses retain their API error classification."""
+    with (
+        patch.object(
+            SwitchbotBaseDevice,
+            "_async_get_user_info",
+            return_value=mock_user_info,
+        ),
+        patch.object(
+            SwitchbotBaseDevice,
+            "api_request",
+            return_value=device_info,
+        ),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(SwitchbotApiError, match="Invalid device response"):
+            await fetch_cloud_devices_by_token(session, "oauth-access-token")
+
+
+@pytest.mark.asyncio
+async def test_get_user_info_preserves_authentication_error() -> None:
+    """Test user info retrieval preserves authentication errors."""
+    with patch.object(
+        SwitchbotBaseDevice,
+        "api_request",
+        side_effect=SwitchbotAuthenticationError("invalid token"),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(SwitchbotAuthenticationError, match="invalid token"):
+            await SwitchbotBaseDevice._async_get_user_info(
+                session,
+                {"authorization": "invalid-token"},
+            )
+
+
+@pytest.mark.asyncio
+async def test_get_user_info_preserves_api_error() -> None:
+    """Test user info retrieval preserves API errors."""
+    with patch.object(
+        SwitchbotBaseDevice,
+        "api_request",
+        side_effect=SwitchbotApiError("API error"),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(SwitchbotApiError, match="API error"):
+            await SwitchbotBaseDevice._async_get_user_info(
+                session,
+                {"authorization": "invalid-token"},
+            )
+
+
+@pytest.mark.asyncio
+async def test_api_request_debug_logs_response_shape_without_values(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test API debug logs contain response fields but no sensitive values."""
+    response = MagicMock()
+    response.status = 200
+    response.headers = {"x-amzn-requestid": "api-request-id"}
+    response.json = AsyncMock(
+        return_value={
+            "statusCode": 100,
+            "message": "success",
+            "body": {
+                "access_token": "sensitive-access-token",
+                "deviceId": "sensitive-device-id",
+                "encryptionKey": "sensitive-encryption-key",
+            },
+        }
+    )
+    session = MagicMock(spec=aiohttp.ClientSession)
+    session.post.return_value.__aenter__ = AsyncMock(return_value=response)
+    session.post.return_value.__aexit__ = AsyncMock(return_value=None)
+    caplog.set_level(logging.DEBUG, logger="switchbot.devices.device")
+
+    result = await SwitchbotBaseDevice.api_request(
+        session, "account", "account/api/v1/user/userinfo"
+    )
+
+    assert result["deviceId"] == "sensitive-device-id"
+    assert "response fields=['body', 'message', 'statusCode']" in caplog.text
+    assert "body fields=['access_token', 'deviceId', 'encryptionKey']" in caplog.text
+    assert "duration_ms=" in caplog.text
+    assert "request_id=api-request-id" in caplog.text
+    for sensitive_value in (
+        "sensitive-access-token",
+        "sensitive-device-id",
+        "sensitive-encryption-key",
+    ):
+        assert sensitive_value not in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_api_request_authentication_error() -> None:
+    """Test HTTP authentication errors retain their specific error type."""
+    response = MagicMock()
+    response.status = 401
+    session = MagicMock(spec=aiohttp.ClientSession)
+    session.post.return_value.__aenter__.return_value = response
+
+    with pytest.raises(SwitchbotAuthenticationError, match="Authentication rejected"):
+        await SwitchbotBaseDevice.api_request(
+            session,
+            "account",
+            "account/api/v1/user/userinfo",
+            {},
+            {"authorization": "invalid-token"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_retrieve_encryption_key_with_password() -> None:
+    """Test the password flow delegates with its access token."""
+    key_details = {
+        "key_id": "ff",
+        "encryption_key": "ffffffffffffffffffffffffffffffff",
+    }
+    with (
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "_get_auth_result",
+            return_value={"access_token": "password-access-token"},
+        ) as mock_get_auth_result,
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "_async_retrieve_encryption_key",
+            return_value=key_details,
+        ) as mock_retrieve_key,
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        result = await SwitchbotEncryptedDevice.async_retrieve_encryption_key(
+            session,
+            "aa:bb:cc:dd:ee:ff",
+            "test@example.com",
+            "password",
+        )
+
+    mock_get_auth_result.assert_awaited_once_with(
+        session, "test@example.com", "password"
+    )
+    mock_retrieve_key.assert_awaited_once_with(
+        session,
+        "aa:bb:cc:dd:ee:ff",
+        {"authorization": "password-access-token"},
+    )
+    assert result == key_details
+
+
+@pytest.mark.asyncio
+async def test_retrieve_encryption_key_by_token_api_error(
+    mock_user_info: dict[str, Any],
+) -> None:
+    """Test an API error while retrieving a key with an OAuth token."""
+    with (
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "_async_get_user_info",
+            return_value=mock_user_info,
+        ),
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "api_request",
+            side_effect=SwitchbotApiError("API error"),
+        ),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(SwitchbotApiError, match="API error"):
+            await SwitchbotEncryptedDevice.async_retrieve_encryption_key_by_token(
+                session, "aa:bb:cc:dd:ee:ff", "oauth-access-token"
+            )
+
+
+@pytest.mark.asyncio
+async def test_retrieve_encryption_key_by_token_authentication_error(
+    mock_user_info: dict[str, Any],
+) -> None:
+    """Test key retrieval preserves authentication errors."""
+    with (
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "_async_get_user_info",
+            return_value=mock_user_info,
+        ),
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "api_request",
+            side_effect=SwitchbotAuthenticationError("expired token"),
+        ),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(SwitchbotAuthenticationError, match="expired token"):
+            await SwitchbotEncryptedDevice.async_retrieve_encryption_key_by_token(
+                session, "aa:bb:cc:dd:ee:ff", "oauth-access-token"
+            )
+
+
+@pytest.mark.asyncio
+async def test_retrieve_encryption_key_by_token_connection_error(
+    mock_user_info: dict[str, Any],
+) -> None:
+    """Test key retrieval maps unexpected request errors to connection errors."""
+    with (
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "_async_get_user_info",
+            return_value=mock_user_info,
+        ),
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "api_request",
+            side_effect=Exception("network error"),
+        ),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(
+            SwitchbotAccountConnectionError,
+            match="Failed to retrieve encryption key",
+        ):
+            await SwitchbotEncryptedDevice.async_retrieve_encryption_key_by_token(
+                session, "aa:bb:cc:dd:ee:ff", "oauth-access-token"
+            )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "device_info",
+    [
+        pytest.param({}, id="missing-communication-key"),
+        pytest.param({"communicationKey": None}, id="invalid-communication-key"),
+        pytest.param(
+            {"communicationKey": {"key": "encryption-key"}}, id="missing-key-id"
+        ),
+        pytest.param(
+            {"communicationKey": {"keyId": "ff"}}, id="missing-encryption-key"
+        ),
+        pytest.param(
+            {"communicationKey": {"keyId": 1, "key": "encryption-key"}},
+            id="invalid-key-id",
+        ),
+    ],
+)
+async def test_retrieve_encryption_key_by_token_invalid_response(
+    mock_user_info: dict[str, Any], device_info: dict[str, Any]
+) -> None:
+    """Test malformed key responses retain their API error classification."""
+    with (
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "_async_get_user_info",
+            return_value=mock_user_info,
+        ),
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "api_request",
+            return_value=device_info,
+        ),
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        with pytest.raises(SwitchbotApiError, match="Invalid encryption key response"):
+            await SwitchbotEncryptedDevice.async_retrieve_encryption_key_by_token(
+                session, "aa:bb:cc:dd:ee:ff", "oauth-access-token"
+            )
+
+
+@pytest.mark.asyncio
+async def test_retrieve_encryption_key_by_token(
+    mock_user_info: dict[str, Any],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test retrieving an encryption key with an OAuth access token."""
+    caplog.set_level(logging.DEBUG, logger="switchbot.devices.device")
+    with (
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "_async_get_user_info",
+            return_value=mock_user_info,
+        ) as mock_get_user_info,
+        patch.object(
+            SwitchbotEncryptedDevice,
+            "api_request",
+            return_value={
+                "communicationKey": {
+                    "keyId": "ff",
+                    "key": "ffffffffffffffffffffffffffffffff",
+                }
+            },
+        ) as mock_api_request,
+    ):
+        session = MagicMock(spec=aiohttp.ClientSession)
+        result = await SwitchbotEncryptedDevice.async_retrieve_encryption_key_by_token(
+            session, "aa:bb:cc:dd:ee:ff", "oauth-access-token"
+        )
+
+    auth_headers = {"authorization": "oauth-access-token"}
+    mock_get_user_info.assert_awaited_once_with(session, auth_headers)
+    mock_api_request.assert_awaited_once_with(
+        session,
+        "wonderlabs.us",
+        "wonder/keys/v1/communicate",
+        {"device_mac": "AABBCCDDEEFF", "keyType": "user"},
+        auth_headers,
+    )
+    assert result == {
+        "key_id": "ff",
+        "encryption_key": "ffffffffffffffffffffffffffffffff",
+    }
+    assert "device=****EEFF" in caplog.text
+    assert "retrieval finished; device=****EEFF duration_ms=" in caplog.text
+    for sensitive_value in (
+        "aa:bb:cc:dd:ee:ff",
+        "oauth-access-token",
+        "ffffffffffffffffffffffffffffffff",
+    ):
+        assert sensitive_value not in caplog.text
+
+
 @pytest.mark.asyncio
 async def test_get_devices_authentication_error() -> None:
     """Test get_devices with authentication error."""
@@ -368,8 +828,14 @@ async def test_populate_model_to_mac_cache() -> None:
     _MODEL_TO_MAC_CACHE.clear()
 
 
-def test_extract_region() -> None:
+def test_masked_device_id_empty() -> None:
+    """Test an empty device identifier is represented safely."""
+    assert _masked_device_id("") == "unknown"
+
+
+def test_extract_region(caplog: pytest.LogCaptureFixture) -> None:
     """Test the _extract_region helper function."""
+    caplog.set_level(logging.WARNING, logger="switchbot.devices.device")
     # Test with botRegion present and not empty
     assert _extract_region({"botRegion": "eu", "country": "de"}) == "eu"
     assert _extract_region({"botRegion": "us", "country": "us"}) == "us"
@@ -384,6 +850,8 @@ def test_extract_region() -> None:
     # Test with empty dict
     assert _extract_region({}) == "us"
 
+    assert "account region missing; defaulting to us" in caplog.text
+
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(

tests/test_oauth.py

@@ -0,0 +1,363 @@
+"""Tests for SwitchBot OAuth helpers."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+from unittest.mock import ANY, AsyncMock, MagicMock
+from urllib.parse import parse_qs, urlparse
+
+import aiohttp
+import pytest
+
+from switchbot import (
+    OAUTH_AUTHORIZE_URL,
+    OAUTH_SCOPE,
+    OAUTH_TOKEN_URL,
+    SwitchbotAccountConnectionError,
+    SwitchbotApiError,
+    SwitchbotAuthenticationError,
+    build_oauth_authorize_url,
+    exchange_oauth_code,
+)
+
+CLIENT_ID = "client-id"
+REDIRECT_URI = "https://example.com/oauth/callback"
+STATE = "oauth-state"
+
+
+def _mock_session(
+    *,
+    status: int = 200,
+    json_data: Any = None,
+    json_exception: Exception | None = None,
+) -> MagicMock:
+    """Create a mocked client session and response."""
+    response = MagicMock()
+    response.status = status
+    response.headers = {"x-request-id": "oauth-request-id"}
+    response.json = AsyncMock(return_value=json_data)
+    if json_exception is not None:
+        response.json.side_effect = json_exception
+
+    session = MagicMock(spec=aiohttp.ClientSession)
+    session.post.return_value.__aenter__ = AsyncMock(return_value=response)
+    session.post.return_value.__aexit__ = AsyncMock(return_value=None)
+    return session
+
+
+def test_oauth_production_endpoints() -> None:
+    """Test OAuth uses the SwitchBot production endpoints."""
+    assert OAUTH_AUTHORIZE_URL == "https://sp.oauth.switchbot.net"
+    assert (
+        OAUTH_TOKEN_URL == "https://account.api.switchbot.net/merchant/v1/oauth/token"
+    )
+    assert OAUTH_SCOPE == "api_login"
+
+
+def test_build_oauth_authorize_url() -> None:
+    """Test authorization URL generation."""
+    url = build_oauth_authorize_url(CLIENT_ID, REDIRECT_URI, STATE)
+    parsed = urlparse(url)
+
+    assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == OAUTH_AUTHORIZE_URL
+    assert parse_qs(parsed.query) == {
+        "client_id": [CLIENT_ID],
+        "redirect_uri": [REDIRECT_URI],
+        "response_type": ["code"],
+        "scope": [OAUTH_SCOPE],
+        "state": [STATE],
+    }
+    assert "client_secret" not in parsed.query
+    assert "code_challenge" not in parsed.query
+
+
+@pytest.mark.asyncio
+async def test_exchange_oauth_code() -> None:
+    """Test authorization code exchange."""
+    token = {
+        "access_token": "access-token",
+        "refresh_token": "refresh-token",
+        "token_type": "Bearer",
+        "expires_in": 3600,
+        "refresh_expires_in": 2592000,
+    }
+    session = _mock_session(json_data=token)
+
+    result = await exchange_oauth_code(
+        session,
+        CLIENT_ID,
+        REDIRECT_URI,
+        "authorization-code",
+    )
+
+    assert result == token
+    session.post.assert_called_once_with(
+        OAUTH_TOKEN_URL,
+        data={
+            "code": "authorization-code",
+            "client_id": CLIENT_ID,
+            "grant_type": "authorization_code",
+            "redirect_uri": REDIRECT_URI,
+        },
+        timeout=ANY,
+    )
+    assert "client_secret" not in session.post.call_args.kwargs["data"]
+    assert "code_verifier" not in session.post.call_args.kwargs["data"]
+
+
+@pytest.mark.asyncio
+async def test_exchange_oauth_code_accepts_string_expiry() -> None:
+    """Test a numeric string expiry returned by the token endpoint."""
+    token = {
+        "access_token": "access-token",
+        "expires_in": "3600",
+    }
+    session = _mock_session(json_data=token)
+
+    result = await exchange_oauth_code(
+        session,
+        CLIENT_ID,
+        REDIRECT_URI,
+        "authorization-code",
+    )
+
+    assert result == {"access_token": "access-token", "expires_in": 3600}
+    assert token["expires_in"] == "3600"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "token",
+    [
+        pytest.param({"expires_in": 3600}, id="missing-access-token"),
+        pytest.param({"access_token": "access-token"}, id="missing-expires-in"),
+        pytest.param(
+            {"access_token": "access-token", "expires_in": "invalid"},
+            id="invalid-expires-in",
+        ),
+        pytest.param(
+            {"access_token": "access-token", "expires_in": True},
+            id="boolean-expires-in",
+        ),
+    ],
+)
+async def test_exchange_oauth_code_invalid_token(token: dict[str, Any]) -> None:
+    """Test invalid token response data."""
+    session = _mock_session(json_data=token)
+
+    with pytest.raises(SwitchbotApiError, match="Invalid token data"):
+        await exchange_oauth_code(
+            session,
+            CLIENT_ID,
+            REDIRECT_URI,
+            "authorization-code",
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("status", "error"),
+    [
+        pytest.param(400, "invalid_grant", id="invalid-grant"),
+        pytest.param(400, "invalid_client", id="invalid-client"),
+        pytest.param(401, "invalid_request", id="unauthorized"),
+        pytest.param(403, "access_denied", id="forbidden"),
+    ],
+)
+async def test_exchange_oauth_code_authentication_error(
+    status: int, error: str
+) -> None:
+    """Test a rejected authorization code."""
+    session = _mock_session(
+        status=status,
+        json_data={
+            "error": error,
+            "error_description": "Authorization code expired",
+        },
+    )
+
+    with pytest.raises(SwitchbotAuthenticationError, match=error):
+        await exchange_oauth_code(
+            session,
+            CLIENT_ID,
+            REDIRECT_URI,
+            "authorization-code",
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("status", "error"),
+    [
+        pytest.param(400, "invalid_request", id="invalid-request"),
+        pytest.param(404, "not_found", id="not-found"),
+        pytest.param(499, "invalid_grant", id="other-client-error"),
+    ],
+)
+async def test_exchange_oauth_code_api_error(status: int, error: str) -> None:
+    """Test OAuth configuration and endpoint errors."""
+    session = _mock_session(status=status, json_data={"error": error})
+
+    with pytest.raises(SwitchbotApiError, match=error):
+        await exchange_oauth_code(
+            session,
+            CLIENT_ID,
+            REDIRECT_URI,
+            "authorization-code",
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", [429, 500, 503])
+async def test_exchange_oauth_code_transient_error(status: int) -> None:
+    """Test transient token service errors."""
+    session = _mock_session(
+        status=status,
+        json_data={"error": "temporarily_unavailable"},
+    )
+
+    with pytest.raises(
+        SwitchbotAccountConnectionError, match="temporarily_unavailable"
+    ):
+        await exchange_oauth_code(
+            session,
+            CLIENT_ID,
+            REDIRECT_URI,
+            "authorization-code",
+        )
+
+
+@pytest.mark.asyncio
+async def test_exchange_oauth_code_connection_error() -> None:
+    """Test token service connection errors."""
+    session = MagicMock(spec=aiohttp.ClientSession)
+    session.post.side_effect = aiohttp.ClientError("connection failed")
+
+    with pytest.raises(SwitchbotAccountConnectionError, match="connection failed"):
+        await exchange_oauth_code(
+            session,
+            CLIENT_ID,
+            REDIRECT_URI,
+            "authorization-code",
+        )
+
+
+@pytest.mark.asyncio
+async def test_exchange_oauth_code_unparsable_error_response(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test an unparsable OAuth error body is not exposed."""
+    session = _mock_session(
+        status=400,
+        json_exception=ValueError("sensitive-provider-error"),
+    )
+    caplog.set_level(logging.DEBUG, logger="switchbot.oauth")
+
+    with pytest.raises(SwitchbotApiError, match="400"):
+        await exchange_oauth_code(
+            session, CLIENT_ID, REDIRECT_URI, "authorization-code"
+        )
+
+    assert "error=unavailable" in caplog.text
+    assert "error_type=ValueError" in caplog.text
+    assert "sensitive-provider-error" not in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_exchange_oauth_code_invalid_json() -> None:
+    """Test an invalid token service response."""
+    session = _mock_session(json_exception=ValueError("invalid json"))
+
+    with pytest.raises(SwitchbotApiError, match="Invalid response"):
+        await exchange_oauth_code(
+            session,
+            CLIENT_ID,
+            REDIRECT_URI,
+            "authorization-code",
+        )
+
+
+@pytest.mark.asyncio
+async def test_exchange_oauth_code_invalid_json_shape() -> None:
+    """Test a non-object token service response."""
+    session = _mock_session(json_data=[])
+
+    with pytest.raises(SwitchbotApiError, match="Invalid response"):
+        await exchange_oauth_code(
+            session,
+            CLIENT_ID,
+            REDIRECT_URI,
+            "authorization-code",
+        )
+
+
+@pytest.mark.asyncio
+async def test_oauth_debug_logs_exclude_sensitive_values(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test OAuth debug logs contain stages but no credential values."""
+    token = {
+        "access_token": "sensitive-access-token",
+        "refresh_token": "sensitive-refresh-token",
+        "token_type": "Bearer",
+        "expires_in": 3600,
+        "refresh_expires_in": 2592000,
+    }
+    session = _mock_session(json_data=token)
+    caplog.set_level(logging.DEBUG, logger="switchbot.oauth")
+
+    build_oauth_authorize_url(CLIENT_ID, REDIRECT_URI, STATE)
+    await exchange_oauth_code(
+        session,
+        CLIENT_ID,
+        REDIRECT_URI,
+        "sensitive-authorization-code",
+    )
+
+    assert "sp.oauth.switchbot.net" in caplog.text
+    assert "example.com" in caplog.text
+    assert "duration_ms=" in caplog.text
+    assert "request_id=oauth-request-id" in caplog.text
+    assert "access_token" in caplog.text
+    assert "expires_in" in caplog.text
+    for sensitive_value in (
+        "sensitive-authorization-code",
+        "sensitive-access-token",
+        "sensitive-refresh-token",
+        "2592000",
+        STATE,
+    ):
+        assert sensitive_value not in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_oauth_error_logs_include_safe_fields_only(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test only bounded, redacted OAuth error fields are logged."""
+    session = _mock_session(
+        status=400,
+        json_data={
+            "error": "invalid_grant",
+            "error_description": (
+                "Authorization code sensitive-authorization-code expired"
+            ),
+            "ignored": "sensitive-provider-error",
+        },
+    )
+    caplog.set_level(logging.DEBUG, logger="switchbot.oauth")
+
+    with pytest.raises(SwitchbotAuthenticationError, match="invalid_grant"):
+        await exchange_oauth_code(
+            session,
+            CLIENT_ID,
+            REDIRECT_URI,
+            "sensitive-authorization-code",
+        )
+
+    assert "invalid_grant" in caplog.text
+    assert "Authorization code <redacted> expired" in caplog.text
+    assert "<redacted>" in caplog.text
+    assert "sensitive-provider-error" not in caplog.text
+    assert "sensitive-authorization-code" not in caplog.text

tests/test_utils.py

@@ -2,7 +2,28 @@
 
 from __future__ import annotations
 
-from switchbot.utils import format_mac_upper
+import pytest
+
+from switchbot.utils import extract_request_id, format_mac_upper
+
+
+@pytest.mark.parametrize(
+    ("headers", "expected"),
+    [
+        pytest.param({"X-Request-ID": "request-id"}, "request-id", id="request-id"),
+        pytest.param(
+            {"X-Amzn-RequestId": "amazon-request-id"},
+            "amazon-request-id",
+            id="amazon-request-id",
+        ),
+        pytest.param({"CF-Ray": "cloudflare-id"}, "cloudflare-id", id="cf-ray"),
+        pytest.param({"other": "value"}, None, id="no-request-id"),
+        pytest.param({}, None, id="no-headers"),
+    ],
+)
+def test_extract_request_id(headers: dict[str, str], expected: str | None) -> None:
+    """Test provider request IDs are extracted case-insensitively."""
+    assert extract_request_id(headers) == expected
 
 
 def test_format_mac_upper() -> None: