Validate API version families and pin SDK auth flow rules

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Changed
+
+- A `version` of a family that the library cannot authenticate is rejected even when `auth_endpoint` is given, instead of being sent with the Cognito flow and rejected by Amazon without an explanation
+- The error of an unsupported version tells that a newer version of a known family can be used by providing its `auth_endpoint`
+- The auth flow of a version and the `Authorization` header it expects are decided in a single place, and the copies bundled in the SDK are pinned to them by tests, so a bump of the SDK cannot leave both halves disagreeing
+
 ## [7.4.0] - 2026-09-04
 
 ### Added

README.md

@@ -205,7 +205,7 @@ amazon = AmazonCreatorsApi(ID, SECRET, VERSION, TAG, COUNTRY, retries=0)  # Fail
 
 ### Custom Endpoints
 
-The base URL of the API and the one used to get the OAuth2 token can be replaced, which is useful to run the tests of a project against a mock server. Providing `auth_endpoint` also makes any `version` valid, so a new one can be used before the library knows about it:
+The base URL of the API and the one used to get the OAuth2 token can be replaced, which is useful to run the tests of a project against a mock server. Providing `auth_endpoint` also makes valid a `version` that is not in the list yet, so a new one can be used before the library knows about it, as long as it belongs to a family that the library can authenticate: `2.x` with Cognito and `3.x` with Login with Amazon. A version of any other family is rejected, as a new family brings a new authentication flow and not just another endpoint:
 
 ```python
 api = AmazonCreatorsApi(

amazon_creatorsapi/aio/api.py

@@ -25,7 +25,7 @@
     get_unique_items,
     sort_items,
 )
-from amazon_creatorsapi.core.oauth import get_auth_endpoint
+from amazon_creatorsapi.core.oauth import build_authorization_header, get_auth_endpoint
 from amazon_creatorsapi.core.parsers import get_asin, get_items_ids
 from amazon_creatorsapi.core.requests import get_request_body
 from amazon_creatorsapi.core.resources import get_all_resources
@@ -180,8 +180,11 @@ class AsyncAmazonCreatorsApi:
         InvalidArgumentError: If neither country nor marketplace is provided,
             if timeout is not greater than zero, if throttling is negative or
             if retries is negative.
-        ValueError: If the version is not supported and no auth_endpoint is
-            given (valid versions: 2.1, 2.2, 2.3, 3.1, 3.2, 3.3).
+        ValueError: If the version is not one of the supported ones, which
+            the message of the error lists, and no auth_endpoint is given, or
+            if it belongs to a family that the library cannot authenticate
+            (2.x and 3.x are the supported ones), which no auth_endpoint
+            makes valid.
 
     """
 
@@ -778,9 +781,7 @@ def _parse_response(self, response: AsyncHttpResponse) -> dict[str, Any]:
 
     def _build_authorization_header(self, token: str) -> str:
         """Build the version-appropriate Authorization header."""
-        if self._version.startswith("3."):
-            return f"Bearer {token}"
-        return f"Bearer {token}, Version {self._version}"
+        return build_authorization_header(self._version, token)
 
     def _deserialize_errors(self, response: dict[str, Any]) -> list[ErrorData]:
         """Deserialize the partial errors of a response to ErrorData models."""

amazon_creatorsapi/api.py

@@ -121,8 +121,11 @@ class AmazonCreatorsApi:
         InvalidArgumentError: If neither country nor marketplace is provided,
             if timeout is not greater than zero, if throttling is negative or
             if retries is negative.
-        ValueError: If the version is not supported and no auth_endpoint is
-            given (valid versions: 2.1, 2.2, 2.3, 3.1, 3.2, 3.3).
+        ValueError: If the version is not one of the supported ones, which
+            the message of the error lists, and no auth_endpoint is given, or
+            if it belongs to a family that the library cannot authenticate
+            (2.x and 3.x are the supported ones), which no auth_endpoint
+            makes valid.
 
     Example:
         >>> api = AmazonCreatorsApi(

amazon_creatorsapi/core/oauth.py

@@ -14,6 +14,15 @@
 # Lifetime assumed for a token when the auth endpoint does not send one
 DEFAULT_EXPIRATION = 3600
 
+# Auth flow of every family of versions, keyed by the major number of the
+# version. The family decides the scope, how the token request is encoded and
+# whether the version travels in the Authorization header, so a version of an
+# unknown family cannot be used by pointing the library to another endpoint:
+# it needs the flow of its family added here.
+COGNITO_FLOW = "cognito"
+LWA_FLOW = "lwa"
+FAMILY_FLOWS = {"2": COGNITO_FLOW, "3": LWA_FLOW}
+
 # Auth endpoint of every version of the API, Cognito for 2.x and LWA for 3.x
 VERSION_ENDPOINTS = {
     "2.1": "https://creatorsapi.auth.us-east-1.amazoncognito.com/oauth2/token",
@@ -25,6 +34,23 @@
 }
 
 
+def get_flow(version: str) -> str | None:
+    """Return the auth flow that a version authenticates with.
+
+    Args:
+        version: API version in use.
+
+    Returns:
+        The flow of the family of the version, or None when the library does
+        not know how to authenticate that family.
+
+    """
+    # A version given as a number is turned into text instead of failing, as
+    # the value is reported back in the error of an unsupported version
+    major = str(version).partition(".")[0]
+    return FAMILY_FLOWS.get(major)
+
+
 def is_lwa(version: str) -> bool:
     """Return whether a version authenticates with Login with Amazon.
 
@@ -35,7 +61,7 @@ def is_lwa(version: str) -> bool:
         True for the versions using LWA, False for the ones using Cognito.
 
     """
-    return version.startswith("3.")
+    return get_flow(version) == LWA_FLOW
 
 
 def get_scope(version: str) -> str:
@@ -51,27 +77,64 @@ def get_scope(version: str) -> str:
     return LWA_SCOPE if is_lwa(version) else COGNITO_SCOPE
 
 
+def build_authorization_header(version: str, token: str) -> str:
+    """Return the Authorization header that a version expects.
+
+    Args:
+        version: API version in use.
+        token: OAuth2 access token of the request.
+
+    Returns:
+        The value of the Authorization header, which carries the version of
+        the credentials in the Cognito flow and only the token in the LWA one.
+
+    """
+    if is_lwa(version):
+        return f"Bearer {token}"
+    return f"Bearer {token}, Version {version}"
+
+
 def get_auth_endpoint(version: str, auth_endpoint: str | None = None) -> str:
     """Return the auth endpoint to use, validating the version when needed.
 
     Args:
         version: API version in use.
         auth_endpoint: Endpoint provided by the user, which takes precedence
-            over the one of the version and makes any version valid.
+            over the one of the version and makes valid any version of a
+            family that the library knows how to authenticate.
 
     Returns:
         The URL used to get the OAuth2 token.
 
     Raises:
-        ValueError: If the version is not supported and no endpoint is given.
+        ValueError: If the family of the version is unknown, or if the version
+            is not in the list and no endpoint is given.
 
     """
-    if auth_endpoint and auth_endpoint.strip():
-        return auth_endpoint
+    endpoint = auth_endpoint.strip() if auth_endpoint else ""
+
+    if version in VERSION_ENDPOINTS:
+        return endpoint or VERSION_ENDPOINTS[version]
+
+    # A custom endpoint is not enough for an unknown family, as the flow of a
+    # new one is not known: the request would be sent with the encoding, the
+    # scope and the headers of Cognito, which Amazon rejects without saying why
+    if get_flow(version) is None:
+        families = ", ".join(f"{family}.x" for family in FAMILY_FLOWS)
+        msg = (
+            f"Unsupported version: {version}. The library only knows how to "
+            f"authenticate the {families} versions, so a newer one needs "
+            f"support added to the library and not just a custom auth_endpoint"
+        )
+        raise ValueError(msg)
 
-    if version not in VERSION_ENDPOINTS:
+    if not endpoint:
         supported = ", ".join(VERSION_ENDPOINTS)
-        msg = f"Unsupported version: {version}. Supported versions are: {supported}"
+        msg = (
+            f"Unsupported version: {version}. Supported versions are: "
+            f"{supported}. A newer version of a known family can be used by "
+            f"providing its auth_endpoint"
+        )
         raise ValueError(msg)
 
-    return VERSION_ENDPOINTS[version]
+    return endpoint

docs/pages/usage-guide.md

@@ -177,7 +177,7 @@ amazon = AmazonCreatorsApi(ID, SECRET, VERSION, TAG, COUNTRY, retries=0)  # Fail
 
 ## Custom Endpoints
 
-The base URL of the API and the one used to get the OAuth2 token can be replaced, which is useful to run the tests of a project against a mock server:
+The base URL of the API and the one used to get the OAuth2 token can be replaced, which is useful to run the tests of a project against a mock server. Providing `auth_endpoint` also makes valid a `version` that is not in the list yet, as long as it belongs to a family that the library can authenticate: `2.x` with Cognito and `3.x` with Login with Amazon. A version of any other family is rejected, as a new family brings a new authentication flow and not just another endpoint:
 
 ```python
 api = AmazonCreatorsApi(

tests/amazon_creatorsapi/aio/api_test.py

@@ -205,23 +205,41 @@ def test_raises_error_for_invalid_version(
             AsyncAmazonCreatorsApi(
                 credential_id="test_id",
                 credential_secret="test_secret",
-                version="9.9",  # Invalid version
+                version="3.9",  # Version out of the list
                 tag="test-tag",
                 country="ES",
             )
 
-        self.assertIn("Unsupported version: 9.9", str(context.exception))
+        self.assertIn("Unsupported version: 3.9", str(context.exception))
         self.assertIn("Supported versions are:", str(context.exception))
 
     @patch("amazon_creatorsapi.aio.api.AsyncOAuth2TokenManager")
-    def test_custom_endpoint_accepts_any_version(
+    def test_raises_error_for_unknown_family(
+        self, mock_token_manager: MagicMock
+    ) -> None:
+        """Test that a version with an unknown auth flow is always rejected."""
+        with self.assertRaises(ValueError) as context:
+            AsyncAmazonCreatorsApi(
+                credential_id="test_id",
+                credential_secret="test_secret",
+                version="9.9",  # Family with an unknown auth flow
+                tag="test-tag",
+                country="ES",
+                auth_endpoint="https://example.com/token",
+            )
+
+        self.assertIn("Unsupported version: 9.9", str(context.exception))
+        mock_token_manager.assert_not_called()
+
+    @patch("amazon_creatorsapi.aio.api.AsyncOAuth2TokenManager")
+    def test_custom_endpoint_accepts_a_new_version(
         self, mock_token_manager: MagicMock
     ) -> None:
-        """Test that a custom endpoint makes any version valid."""
+        """Test that a custom endpoint makes valid a version out of the list."""
         AsyncAmazonCreatorsApi(
             credential_id="test_id",
             credential_secret="test_secret",
-            version="9.9",
+            version="3.4",
             tag="test-tag",
             country="ES",
             auth_endpoint="https://example.com/token",

tests/amazon_creatorsapi/api_test.py

@@ -1879,14 +1879,24 @@ def build_api_with(self, **options: object) -> AmazonCreatorsApi:
     def test_unsupported_version_is_rejected(self) -> None:
         """Test that a version out of the list needs a custom endpoint."""
         with self.assertRaises(ValueError) as context:
-            self.build_api_with(version="4.0")
+            self.build_api_with(version="3.4")
+
+        self.assertIn("Unsupported version: 3.4", str(context.exception))
+
+    def test_unknown_family_is_rejected_with_a_custom_endpoint(self) -> None:
+        """Test that a version with an unknown auth flow is always rejected."""
+        with self.assertRaises(ValueError) as context:
+            self.build_api_with(
+                version="4.0",
+                auth_endpoint="https://example.com/token",
+            )
 
         self.assertIn("Unsupported version: 4.0", str(context.exception))
 
-    def test_custom_endpoint_accepts_any_version(self) -> None:
-        """Test that a custom endpoint makes any version valid."""
+    def test_custom_endpoint_accepts_a_new_version(self) -> None:
+        """Test that a custom endpoint makes valid a version out of the list."""
         api = self.build_api_with(
-            version="4.0",
+            version="3.4",
             auth_endpoint="https://example.com/token",
         )
 

tests/amazon_creatorsapi/core/oauth_test.py

@@ -5,15 +5,32 @@
 import unittest
 
 from amazon_creatorsapi.core.oauth import (
+    COGNITO_FLOW,
     COGNITO_SCOPE,
+    LWA_FLOW,
     LWA_SCOPE,
     VERSION_ENDPOINTS,
     get_auth_endpoint,
+    get_flow,
     get_scope,
     is_lwa,
 )
 
 
+class TestGetFlow(unittest.TestCase):
+    """Tests for get_flow function."""
+
+    def test_flow_of_the_known_families(self) -> None:
+        """Test that the flow is taken from the major number of a version."""
+        self.assertEqual(get_flow("2.4"), COGNITO_FLOW)
+        self.assertEqual(get_flow("3.4"), LWA_FLOW)
+
+    def test_flow_of_an_unknown_family(self) -> None:
+        """Test that a family out of the known ones has no flow."""
+        self.assertIsNone(get_flow("4.1"))
+        self.assertIsNone(get_flow("not a version"))
+
+
 class TestIsLwa(unittest.TestCase):
     """Tests for is_lwa function."""
 
@@ -52,13 +69,28 @@ def test_custom_endpoint_wins(self) -> None:
             "https://example.test/token",
         )
 
-    def test_custom_endpoint_makes_any_version_valid(self) -> None:
+    def test_custom_endpoint_accepts_a_new_version_of_a_known_family(self) -> None:
         """Test that a custom endpoint accepts a version out of the list."""
         self.assertEqual(
-            get_auth_endpoint("4.0", "https://example.test/token"),
+            get_auth_endpoint("3.4", "https://example.test/token"),
+            "https://example.test/token",
+        )
+
+    def test_custom_endpoint_is_stripped(self) -> None:
+        """Test that the endpoint given by the user is used without spaces."""
+        self.assertEqual(
+            get_auth_endpoint("2.2", "  https://example.test/token\n"),
             "https://example.test/token",
         )
 
+    def test_custom_endpoint_does_not_accept_an_unknown_family(self) -> None:
+        """Test that a family with an unknown auth flow is always rejected."""
+        with self.assertRaises(ValueError) as context:
+            get_auth_endpoint("4.0", "https://example.test/token")
+
+        self.assertIn("Unsupported version: 4.0", str(context.exception))
+        self.assertIn("2.x, 3.x", str(context.exception))
+
     def test_blank_endpoint_is_ignored(self) -> None:
         """Test that a blank endpoint falls back to the one of the version."""
         self.assertEqual(get_auth_endpoint("2.2", "   "), VERSION_ENDPOINTS["2.2"])
@@ -69,3 +101,11 @@ def test_unsupported_version_is_rejected(self) -> None:
             get_auth_endpoint("4.0")
 
         self.assertIn("Unsupported version: 4.0", str(context.exception))
+
+    def test_error_of_a_known_family_points_to_the_custom_endpoint(self) -> None:
+        """Test that the error of a new version explains how to use it."""
+        with self.assertRaises(ValueError) as context:
+            get_auth_endpoint("3.4")
+
+        self.assertIn("Unsupported version: 3.4", str(context.exception))
+        self.assertIn("auth_endpoint", str(context.exception))