Add handling for unexpected successful token response

Author: adamjmcgrath

packages/auth0_api_python/src/auth0_api_python/api_client.py

@@ -465,11 +465,24 @@ async def get_access_token_for_connection(self, options: dict[str, Any]) -> dict
                         response.status_code
                     )
 
-                token_endpoint_response = response.json()
+                try:
+                    token_endpoint_response = response.json()
+                except Exception:
+                    raise ApiError("invalid_json", "Token endpoint returned invalid JSON.")
+
+                access_token = token_endpoint_response.get("access_token")
+                if not isinstance(access_token, str) or not access_token:
+                    raise ApiError("invalid_response", "Missing or invalid access_token in response.", 502)
+
+                expires_in_raw = token_endpoint_response.get("expires_in", 3600)
+                try:
+                    expires_in = int(expires_in_raw)
+                except (TypeError, ValueError):
+                    raise ApiError("invalid_response", "expires_in is not an integer.", 502)
 
                 return {
-                    "access_token": token_endpoint_response.get("access_token"),
-                    "expires_at": int(time.time()) + int(token_endpoint_response.get("expires_in", 3600)),
+                    "access_token": access_token,
+                    "expires_at": int(time.time()) + expires_in,
                     "scope": token_endpoint_response.get("scope", "")
                 }
 

packages/auth0_api_python/tests/test_api_client.py

@@ -1819,4 +1819,91 @@ async def test_get_access_token_for_connection_error_text_json_content_type(http
         })
     assert err.value.code == "invalid_request"
     assert err.value.status_code == 400
-    assert "bad request" in str(err.value).lower()
+    assert "bad request" in str(err.value).lower()
+
+    @pytest.mark.asyncio
+    async def test_get_access_token_for_connection_invalid_json(httpx_mock: HTTPXMock):
+        httpx_mock.add_response(
+            method="GET",
+            url="https://auth0.local/.well-known/openid-configuration",
+            json={"token_endpoint": "https://auth0.local/oauth/token"}
+        )
+        httpx_mock.add_response(
+            method="POST",
+            url="https://auth0.local/oauth/token",
+            status_code=200,
+            content="not a json",  # Invalid JSON
+            headers={"Content-Type": "application/json"}
+        )
+        options = ApiClientOptions(
+            domain="auth0.local",
+            audience="my-audience",
+            client_id="cid",
+            client_secret="csecret",
+        )
+        api_client = ApiClient(options)
+        with pytest.raises(ApiError) as err:
+            await api_client.get_access_token_for_connection({
+                "connection": "test-conn",
+                "access_token": "user-token"
+            })
+        assert err.value.code == "invalid_json"
+        assert "invalid json" in str(err.value).lower()
+
+    @pytest.mark.asyncio
+    async def test_get_access_token_for_connection_invalid_access_token_type(httpx_mock: HTTPXMock):
+        httpx_mock.add_response(
+            method="GET",
+            url="https://auth0.local/.well-known/openid-configuration",
+            json={"token_endpoint": "https://auth0.local/oauth/token"}
+        )
+        httpx_mock.add_response(
+            method="POST",
+            url="https://auth0.local/oauth/token",
+            status_code=200,
+            json={"access_token": 12345, "expires_in": 3600}  # access_token not a string
+        )
+        options = ApiClientOptions(
+            domain="auth0.local",
+            audience="my-audience",
+            client_id="cid",
+            client_secret="csecret",
+        )
+        api_client = ApiClient(options)
+        with pytest.raises(ApiError) as err:
+            await api_client.get_access_token_for_connection({
+                "connection": "test-conn",
+                "access_token": "user-token"
+            })
+        assert err.value.code == "invalid_response"
+        assert "access_token" in str(err.value).lower()
+        assert err.value.status_code == 502
+
+    @pytest.mark.asyncio
+    async def test_get_access_token_for_connection_expires_in_not_integer(httpx_mock: HTTPXMock):
+        httpx_mock.add_response(
+            method="GET",
+            url="https://auth0.local/.well-known/openid-configuration",
+            json={"token_endpoint": "https://auth0.local/oauth/token"}
+        )
+        httpx_mock.add_response(
+            method="POST",
+            url="https://auth0.local/oauth/token",
+            status_code=200,
+            json={"access_token": "abc123", "expires_in": "not-an-int"}
+        )
+        options = ApiClientOptions(
+            domain="auth0.local",
+            audience="my-audience",
+            client_id="cid",
+            client_secret="csecret",
+        )
+        api_client = ApiClient(options)
+        with pytest.raises(ApiError) as err:
+            await api_client.get_access_token_for_connection({
+                "connection": "test-conn",
+                "access_token": "user-token"
+            })
+        assert err.value.code == "invalid_response"
+        assert "expires_in" in str(err.value).lower()
+        assert err.value.status_code == 502