fix(security): harden credential handling, concurrency, validation, and retry mechanisms

.github/workflows/ci.yml

@@ -6,6 +6,9 @@ on:
   pull_request:
     branches: [main]
 
+permissions:
+  contents: read
+
 jobs:
   test:
     name: Test (Python ${{ matrix.python-version }} on ${{ matrix.os }})

README.md

@@ -1,7 +1,7 @@
 # AmazonAPIWrapper
 
 [![PyPI version](https://img.shields.io/pypi/v/AmazonAPIWrapper.svg)](https://pypi.org/project/AmazonAPIWrapper/)
-[![Python Versions](https://img.shields.io/pypi/pyversions/AmazonAPIWrapper.svg)](https://pypi.org/project/AmazonAPIWrapper/)
+[![Python Versions](https://img.shields.io/badge/python-3.9%20%7C%203.10%20%7C%203.11%20%7C%203.12%20%7C%203.13-blue.svg)](https://pypi.org/project/AmazonAPIWrapper/)
 [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
 [![CI](https://github.com/lv10/amazonapi/actions/workflows/ci.yml/badge.svg)](https://github.com/lv10/amazonapi/actions/workflows/ci.yml)
 

amazon/auth/oauth.py

@@ -5,7 +5,7 @@
 import asyncio
 import threading
 import time
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 import httpx
 
@@ -16,7 +16,7 @@
 class OAuthToken:
     """OAuth 2.0 Access Token container."""
 
-    access_token: str
+    access_token: str = field(repr=False)
     token_type: str
     expires_at: float  # Epoch timestamp in seconds
     scope: str | None = None
@@ -29,6 +29,16 @@ def is_expired(self, buffer_seconds: float = 300.0) -> bool:
         """
         return time.time() >= (self.expires_at - buffer_seconds)
 
+    def __repr__(self) -> str:
+        if len(self.access_token) > 8:
+            masked = f"{self.access_token[:4]}...{self.access_token[-4:]}"
+        else:
+            masked = "***"
+        return (
+            f"OAuthToken(access_token={masked!r}, token_type={self.token_type!r}, "
+            f"expires_at={self.expires_at}, scope={self.scope!r})"
+        )
+
 
 class OAuthTokenManager:
     """Thread-safe and coroutine-safe manager for OAuth 2.0 access tokens."""
@@ -40,6 +50,7 @@ def __init__(
         token_url: str,
         scope: str = "creatorsapi::default",
         buffer_seconds: float = 300.0,
+        timeout: float = 15.0,
     ) -> None:
         """Initialize OAuthTokenManager.
 
@@ -49,20 +60,31 @@ def __init__(
             token_url: Regional OAuth 2.0 token endpoint (e.g. https://api.amazon.com/auth/o2/token).
             scope: OAuth scope (default: "creatorsapi::default").
             buffer_seconds: Refresh buffer window in seconds before token expires.
+            timeout: HTTP request timeout in seconds when creating standalone clients.
         """
         self.credential_id = credential_id.strip()
         self.credential_secret = credential_secret.strip()
         self.token_url = token_url.strip()
         self.scope = scope.strip()
         self.buffer_seconds = buffer_seconds
+        self.timeout = timeout
 
         self._cached_token: OAuthToken | None = None
         self._sync_lock = threading.Lock()
         self._async_lock: asyncio.Lock | None = None
+        self._async_lock_init_lock = threading.Lock()
+
+    def __repr__(self) -> str:
+        return (
+            f"OAuthTokenManager(credential_id={self.credential_id!r}, credential_secret='***', "
+            f"token_url={self.token_url!r}, scope={self.scope!r})"
+        )
 
     def _get_async_lock(self) -> asyncio.Lock:
         if self._async_lock is None:
-            self._async_lock = asyncio.Lock()
+            with self._async_lock_init_lock:
+                if self._async_lock is None:
+                    self._async_lock = asyncio.Lock()
         return self._async_lock
 
     def _build_token_payload(self) -> dict[str, str]:
@@ -75,7 +97,7 @@ def _build_token_payload(self) -> dict[str, str]:
 
     def _parse_token_response(self, response: httpx.Response) -> OAuthToken:
         if response.status_code != 200:
-            error_details = response.text
+            error_details = response.text[:2048] if len(response.text) > 2048 else response.text
             try:
                 data = response.json()
                 error_msg = data.get("error_description") or data.get("error") or error_details
@@ -124,7 +146,7 @@ def get_token(self, client: httpx.Client | None = None) -> str:
             payload = self._build_token_payload()
             should_close = False
             if client is None:
-                client = httpx.Client(timeout=15.0)
+                client = httpx.Client(timeout=self.timeout)
                 should_close = True
 
             try:
@@ -149,13 +171,14 @@ async def get_token_async(self, client: httpx.AsyncClient | None = None) -> str:
             Bearer access token string.
         """
         async with self._get_async_lock():
-            if self._cached_token and not self._cached_token.is_expired(self.buffer_seconds):
-                return self._cached_token.access_token
+            with self._sync_lock:
+                if self._cached_token and not self._cached_token.is_expired(self.buffer_seconds):
+                    return self._cached_token.access_token
 
             payload = self._build_token_payload()
             should_close = False
             if client is None:
-                client = httpx.AsyncClient(timeout=15.0)
+                client = httpx.AsyncClient(timeout=self.timeout)
                 should_close = True
 
             try:
@@ -164,8 +187,10 @@ async def get_token_async(self, client: httpx.AsyncClient | None = None) -> str:
                     data=payload,
                     headers={"Content-Type": "application/x-www-form-urlencoded"},
                 )
-                self._cached_token = self._parse_token_response(resp)
-                return self._cached_token.access_token
+                token = self._parse_token_response(resp)
+                with self._sync_lock:
+                    self._cached_token = token
+                return token.access_token
             finally:
                 if should_close:
                     await client.aclose()

amazon/auth/sigv4.py

@@ -41,6 +41,12 @@ def __init__(
         self.aws_region = aws_region.strip()
         self.service = service.strip()
 
+    def __repr__(self) -> str:
+        return (
+            f"SigV4Signer(access_key={self.access_key!r}, secret_key='***', "
+            f"aws_region={self.aws_region!r}, service={self.service!r})"
+        )
+
     def sign(
         self,
         host: str,

amazon/clients/base.py

@@ -2,7 +2,10 @@
 
 from __future__ import annotations
 
+import datetime
+import email.utils
 import logging
+import random
 
 import httpx
 
@@ -17,6 +20,8 @@
 
 logger = logging.getLogger("amazonapi")
 
+MAX_ERROR_BODY_LENGTH = 4096
+
 DEFAULT_ITEM_RESOURCES: list[str] = [
     "ItemInfo.Title",
     "ItemInfo.ByLineInfo",
@@ -52,11 +57,59 @@
 ]
 
 
+def parse_retry_after(response: httpx.Response, default: float) -> float:
+    """Parse HTTP Retry-After header if present, returning delay in seconds.
+
+    Supports integer seconds and HTTP-date formats.
+    """
+    retry_header = response.headers.get("retry-after") or response.headers.get("Retry-After")
+    if not retry_header:
+        return default
+
+    retry_header = retry_header.strip()
+    try:
+        # Try integer seconds
+        seconds = float(retry_header)
+        return max(0.0, seconds)
+    except ValueError:
+        pass
+
+    try:
+        # Try HTTP-date format (RFC 7231)
+        target_date = email.utils.parsedate_to_datetime(retry_header)
+        now = datetime.datetime.now(datetime.timezone.utc)
+        delta = (target_date - now).total_seconds()
+        return max(0.0, delta)
+    except Exception:
+        return default
+
+
+def calculate_backoff(retry_count: int, base_delay: float, max_delay: float = 60.0) -> float:
+    """Calculate exponential backoff with full jitter to avoid thundering herds.
+
+    Args:
+        retry_count: Attempt number (1-based index).
+        base_delay: Initial base delay in seconds.
+        max_delay: Maximum delay cap in seconds.
+
+    Returns:
+        Random jittered delay in seconds between 0 and min(max_delay, base_delay * 2^(retry_count-1)).
+    """
+    delay_ceiling = min(max_delay, base_delay * (2 ** max(0, retry_count - 1)))
+    return random.uniform(0.0, delay_ceiling)
+
+
 def map_http_error(response: httpx.Response) -> AmazonAPIError:
-    """Map HTTP response to specific AmazonAPIError subclass."""
+    """Map HTTP response to specific AmazonAPIError subclass with bounded memory footprint."""
     status = response.status_code
     error_code = None
-    message = response.text
+    raw_text = response.text
+    truncated_text = (
+        raw_text[:MAX_ERROR_BODY_LENGTH] + "... [truncated]"
+        if len(raw_text) > MAX_ERROR_BODY_LENGTH
+        else raw_text
+    )
+    message = truncated_text
 
     try:
         data = response.json()
@@ -80,46 +133,46 @@ def map_http_error(response: httpx.Response) -> AmazonAPIError:
             message=f"Rate limit exceeded: {message}",
             status_code=status,
             error_code=error_code or "TooManyRequests",
-            response_body=response.text,
+            response_body=truncated_text,
             headers=headers_dict,
         )
     elif status in (401, 403) or error_code in ("InvalidClientTokenId", "MissingClientTokenId", "AccessDeniedException"):
         return AmazonAuthenticationError(
             message=f"Authentication failed: {message}",
             status_code=status,
             error_code=error_code or "Unauthorized",
-            response_body=response.text,
+            response_body=truncated_text,
             headers=headers_dict,
         )
     elif status == 400 or error_code in ("AWS.MissingParameters", "AWS.InvalidParameterValue", "InvalidParameterValue"):
         return AmazonBadRequestError(
             message=f"Bad request: {message}",
             status_code=status,
             error_code=error_code or "BadRequest",
-            response_body=response.text,
+            response_body=truncated_text,
             headers=headers_dict,
         )
     elif status == 404 or error_code in ("ResourceNotFound", "NoExactMatches"):
         return AmazonNotFoundError(
             message=f"Resource not found: {message}",
             status_code=status,
             error_code=error_code or "NotFound",
-            response_body=response.text,
+            response_body=truncated_text,
             headers=headers_dict,
         )
     elif status >= 500 or error_code == "InternalError":
         return AmazonServerError(
             message=f"Amazon server error (HTTP {status}): {message}",
             status_code=status,
             error_code=error_code or "InternalError",
-            response_body=response.text,
+            response_body=truncated_text,
             headers=headers_dict,
         )
     else:
         return AmazonAPIError(
             message=f"API request failed with status {status}: {message}",
             status_code=status,
             error_code=error_code,
-            response_body=response.text,
+            response_body=truncated_text,
             headers=headers_dict,
         )