Handle forbidden auth check with token user ID

simplipy/api.py

@@ -4,9 +4,10 @@
 
 import asyncio
 import sys
+from base64 import urlsafe_b64decode
 from collections.abc import Awaitable, Callable
 from datetime import datetime
-from json.decoder import JSONDecodeError
+from json import JSONDecodeError, loads
 from typing import Any, cast
 
 import backoff
@@ -39,6 +40,7 @@
 DEFAULT_MEDIA_RETRIES = 4
 DEFAULT_TIMEOUT = 10
 DEFAULT_TOKEN_EXPIRATION_WINDOW = 5
+USER_ID_CLAIM = "http://simplisafe.com/uid"
 
 
 class API:  # pylint: disable=too-many-instance-attributes
@@ -199,10 +201,41 @@ async def _async_handle_on_backoff(self, _: dict[str, Any]) -> None:
 
     async def _async_post_init(self) -> None:
         """Perform some post-init actions."""
-        auth_check_resp = await self._async_api_request("get", "api/authCheck")
-        self.user_id = auth_check_resp["userId"]
+        try:
+            auth_check_resp = await self._async_api_request("get", "api/authCheck")
+        except ClientResponseError as err:
+            if (
+                err.status != 403
+                or not self.access_token
+                or (user_id := self._user_id_from_access_token(self.access_token))
+                is None
+            ):
+                raise
+
+            LOGGER.debug(
+                "Authorization check returned 403; using user ID from access token"
+            )
+            self.user_id = user_id
+        else:
+            self.user_id = auth_check_resp["userId"]
         self.websocket = WebsocketClient(self)
 
+    @staticmethod
+    def _user_id_from_access_token(access_token: str) -> int | None:
+        """Extract the SimpliSafe user ID from an access token."""
+        try:
+            encoded_payload = access_token.split(".")[1]
+            padding = "=" * (-len(encoded_payload) % 4)
+            payload = loads(urlsafe_b64decode(encoded_payload + padding))
+            return int(payload[USER_ID_CLAIM])
+        except (
+            IndexError,
+            KeyError,
+            TypeError,
+            ValueError,
+        ):
+            return None
+
     async def _async_api_request(
         self, method: str, endpoint: str, url_base: str = API_URL_BASE, **kwargs: Any
     ) -> dict[str, Any]:

tests/test_api.py

@@ -4,7 +4,9 @@
 from __future__ import annotations
 
 import asyncio
+from base64 import urlsafe_b64encode
 from datetime import timedelta
+from json import dumps
 from typing import Any
 from unittest.mock import AsyncMock, Mock, patch
 
@@ -22,9 +24,17 @@
     TEST_CODE_VERIFIER,
     TEST_REFRESH_TOKEN,
     TEST_SUBSCRIPTION_ID,
+    TEST_USER_ID,
 )
 
 
+def create_access_token(user_id: int | None = TEST_USER_ID) -> str:
+    """Create an unsigned access token containing a SimpliSafe user ID."""
+    payload = {} if user_id is None else {"http://simplisafe.com/uid": user_id}
+    encoded_payload = urlsafe_b64encode(dumps(payload).encode()).decode().rstrip("=")
+    return f"header.{encoded_payload}.signature"
+
+
 @pytest.mark.asyncio
 async def test_401_bad_credentials(
     aresponses: ResponsesMockServer,
@@ -229,6 +239,66 @@ async def test_client_async_from_authorization_code(
     aresponses.assert_plan_strictly_followed()
 
 
+@pytest.mark.asyncio
+async def test_client_uses_access_token_user_id_when_auth_check_is_forbidden(
+    api_token_response: dict[str, Any],
+    aresponses: ResponsesMockServer,
+) -> None:
+    """Test extracting the user ID when the authorization check returns 403."""
+    api_token_response["access_token"] = create_access_token()
+    aresponses.add(
+        "auth.simplisafe.com",
+        "/oauth/token",
+        "post",
+        response=aiohttp.web_response.json_response(api_token_response, status=200),
+    )
+    aresponses.add(
+        "api.simplisafe.com",
+        "/v1/api/authCheck",
+        "get",
+        response=aresponses.Response(text="Forbidden", status=403),
+    )
+
+    async with aiohttp.ClientSession() as session:
+        simplisafe = await API.async_from_auth(
+            TEST_AUTHORIZATION_CODE, TEST_CODE_VERIFIER, session=session
+        )
+        assert simplisafe.user_id == TEST_USER_ID
+
+    aresponses.assert_plan_strictly_followed()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("access_token", [TEST_ACCESS_TOKEN, create_access_token(None)])
+async def test_client_rejects_forbidden_auth_check_without_user_id(
+    access_token: str,
+    api_token_response: dict[str, Any],
+    aresponses: ResponsesMockServer,
+) -> None:
+    """Test preserving a 403 when the access token has no usable user ID."""
+    api_token_response["access_token"] = access_token
+    aresponses.add(
+        "auth.simplisafe.com",
+        "/oauth/token",
+        "post",
+        response=aiohttp.web_response.json_response(api_token_response, status=200),
+    )
+    aresponses.add(
+        "api.simplisafe.com",
+        "/v1/api/authCheck",
+        "get",
+        response=aresponses.Response(text="Forbidden", status=403),
+    )
+
+    async with aiohttp.ClientSession() as session:
+        with pytest.raises(aiohttp.ClientResponseError, match="403"):
+            await API.async_from_auth(
+                TEST_AUTHORIZATION_CODE, TEST_CODE_VERIFIER, session=session
+            )
+
+    aresponses.assert_plan_strictly_followed()
+
+
 @pytest.mark.asyncio
 async def test_client_async_from_authorization_code_http_error(
     aresponses: ResponsesMockServer,