fix(apps): include sanitized response details in HTTP errors

runpod/apps/__init__.py

@@ -6,6 +6,7 @@
 from .entrypoint import local_entrypoint
 from .errors import (
     AppError,
+    AppHTTPError,
     EndpointNotFound,
     JobStatusError,
     RemoteExecutionError,
@@ -26,6 +27,7 @@
     "ApiHandle",
     "App",
     "AppError",
+    "AppHTTPError",
     "Context",
     "DataCenter",
     "EndpointNotFound",

runpod/apps/api.py

@@ -17,6 +17,8 @@
 from ..api.rest import run_rest_paginated_request_async, run_rest_request_async
 from ..error import AuthenticationError, QueryError
 from ..user_agent import APPS_USER_AGENT
+from .errors import AppHTTPError
+from .utils.http import raise_for_status
 
 _TRANSPORT_RETRIES = 4
 
@@ -713,7 +715,7 @@ async def _reader():
             try:
                 await self._put_tarball(upload_url, _reader(), total, timeout)
                 return
-            except (aiohttp.ClientError, OSError):
+            except (AppHTTPError, aiohttp.ClientError, OSError):
                 if attempt == attempts - 1:
                     raise
                 await asyncio.sleep(2**attempt)
@@ -729,4 +731,4 @@ async def _put_tarball(self, upload_url, reader, total, timeout) -> None:
                     "Content-Length": str(total),
                 },
             ) as resp:
-                resp.raise_for_status()
+                await raise_for_status(resp)

runpod/apps/errors.py

@@ -7,6 +7,18 @@ class AppError(Exception):
     """base error for the apps sdk."""
 
 
+class AppHTTPError(AppError):
+    """an http failure with a sanitized request url and response excerpt."""
+
+    def __init__(self, method: str, url: str, status: int, body_excerpt: str = ""):
+        self.method = method
+        self.url = str(URL(url).with_user(None).with_query(None).with_fragment(None))
+        self.status = status
+        self.body_excerpt = body_excerpt
+        detail = f": {body_excerpt}" if body_excerpt else ""
+        super().__init__(f"HTTP {status} ({method} {self.url}){detail}")
+
+
 class EndpointNotFound(AppError):
     """a resource was invoked but is not deployed."""
 
@@ -23,18 +35,15 @@ class RemoteExecutionError(AppError):
     """the remote worker reported a failure executing the function."""
 
 
-class JobStatusError(AppError):
+class JobStatusError(AppHTTPError):
     """a job-status request failed with an http error."""
 
-    def __init__(self, job_id: str, status: int, method: str, url: str):
+    def __init__(
+        self, job_id: str, status: int, method: str, url: str, body_excerpt: str = ""
+    ):
         self.job_id = job_id
-        self.status = status
-        self.method = method
-        self.url = str(URL(url).with_user(None).with_query(None).with_fragment(None))
-        super().__init__(
-            f"could not retrieve status for job '{job_id}': "
-            f"HTTP {status} ({method} {self.url})"
-        )
+        super().__init__(method, url, status, body_excerpt)
+        self.args = (f"could not retrieve status for job '{job_id}': {self.args[0]}",)
 
 
 class ScheduleNotSupported(AppError):

runpod/apps/logs.py

@@ -14,6 +14,7 @@
 from ..api.ctl_commands import _path_segment
 from ..api.rest import _build_url
 from ..user_agent import APPS_USER_AGENT
+from .utils.http import raise_for_status
 
 HAPI_BASE = os.environ.get("RUNPOD_HAPI_URL", "https://hapi.runpod.net")
 
@@ -39,7 +40,7 @@ async def pod_logs(
         async with session.get(
             url, params={"type": log_type}, headers=_headers()
         ) as resp:
-            resp.raise_for_status()
+            await raise_for_status(resp)
             data = await resp.json()
     return {k: v or [] for k, v in data.items()}
 
@@ -88,7 +89,7 @@ async def stream_pod_logs(
     client_timeout = aiohttp.ClientTimeout(total=STREAM_TIMEOUT_SECONDS, sock_read=60)
     async with aiohttp.ClientSession(timeout=client_timeout) as session:
         async with session.get(url, params=params, headers=headers) as resp:
-            resp.raise_for_status()
+            await raise_for_status(resp)
             event_id = None
             data = []
             content = resp.content.__aiter__()

runpod/apps/targets.py

@@ -15,11 +15,12 @@
 
 from .._hosts import endpoint_domain
 from ..user_agent import APPS_USER_AGENT
-from .errors import EndpointNotFound, JobStatusError, RemoteExecutionError
+from .errors import AppHTTPError, EndpointNotFound, RemoteExecutionError
 from .protocol import FORMAT_JSON, FunctionRequest
 from .serialization import require_json_value
 from .spec import ResourceSpec
 from .utils.events import emit
+from .utils.http import raise_for_status
 from .utils.network import api_key as _api_key
 from .utils.network import endpoint_url_base as _endpoint_url_base
 
@@ -246,30 +247,20 @@ async def _request_json(
                 async with session.request(
                     method, url, json=payload, headers=headers
                 ) as resp:
-                    if resp.status >= 400 and job_id is not None:
-                        last_exc = JobStatusError(job_id, resp.status, method, url)
-                        if (resp.status == 404 and retry_not_found) or (
-                            resp.status in RETRYABLE_STATUSES and retry_transient
-                        ):
-                            continue
-                        raise last_exc
-                    if resp.status == 404:
+                    if resp.status == 404 and job_id is None:
                         last_exc = EndpointNotFound(app_name, resource_name)
                         if retry_not_found and attempt < attempts - 1:
                             continue
                         raise last_exc
-                    if resp.status in RETRYABLE_STATUSES:
-                        last_exc = aiohttp.ClientResponseError(
-                            resp.request_info,
-                            resp.history,
-                            status=resp.status,
-                            message=await resp.text(),
-                        )
-                        if not retry_transient:
-                            raise last_exc
-                        continue
-                    resp.raise_for_status()
+                    await raise_for_status(resp, job_id=job_id)
                     return await resp.json()
+        except AppHTTPError as exc:
+            if (exc.status == 404 and retry_not_found) or (
+                exc.status in RETRYABLE_STATUSES and retry_transient
+            ):
+                last_exc = exc
+                continue
+            raise
         except (aiohttp.ClientConnectionError, asyncio.TimeoutError) as exc:
             if not retry_transient:
                 raise
@@ -516,7 +507,7 @@ async def http(
             ) as resp:
                 if resp.status == 404 and self._app_name:
                     raise EndpointNotFound(self._app_name, self._resource_name)
-                resp.raise_for_status()
+                await raise_for_status(resp)
                 text = await resp.text()
         try:
             result = json.loads(text)

runpod/apps/tasks.py

@@ -24,6 +24,7 @@
 from .api import AppsApiClient, is_capacity_error
 from .context import CLEANUP_TIMEOUT
 from .spec import ResourceSpec
+from .utils.http import raise_for_status
 
 log = logging.getLogger(__name__)
 
@@ -285,7 +286,7 @@ async def _post(self, path: str, request: Dict[str, Any]) -> None:
                     if resp.status == 404 and attempt < attempts - 1:
                         await asyncio.sleep(2 * (attempt + 1))
                         continue
-                    resp.raise_for_status()
+                    await raise_for_status(resp)
                     if timeout is not None:
                         response = await resp.json()
                         if (
@@ -319,7 +320,7 @@ async def poll_result(self) -> Optional[Dict[str, Any]]:
                     if resp.status == 404 and attempt < attempts - 1:
                         await asyncio.sleep(2 * (attempt + 1))
                         continue
-                    resp.raise_for_status()
+                    await raise_for_status(resp)
                     data = await resp.json()
                     break
         if data.get("status") == "DONE":

runpod/apps/utils/http.py

@@ -0,0 +1,82 @@
+"""sanitized diagnostics for apps http responses."""
+
+import asyncio
+import html
+import json
+from urllib.parse import quote, quote_plus
+
+import aiohttp
+from yarl import URL
+
+from ..errors import AppHTTPError, JobStatusError
+
+BODY_READ_LIMIT = 65536
+BODY_EXCERPT_LIMIT = 1024
+
+
+async def raise_for_status(
+    response: aiohttp.ClientResponse, *, job_id: str | None = None
+) -> None:
+    if response.status < 400:
+        return
+
+    secrets = set()
+    for item in (*response.history, response):
+        url = item.request_info.real_url
+        secrets.update(value for value in url.query.values() if value)
+        secrets.update(value for value in (url.user, url.password) if value)
+        for name, value in item.request_info.headers.items():
+            if any(
+                word in name.lower()
+                for word in ("authorization", "cookie", "key", "token", "secret")
+            ):
+                secrets.add(value)
+                if name.lower() in ("authorization", "proxy-authorization"):
+                    secrets.add(value.partition(" ")[2])
+                if name.lower() == "cookie":
+                    secrets.update(
+                        part.partition("=")[2].strip() for part in value.split(";")
+                    )
+
+    sensitive = set()
+    for value in secrets:
+        if value:
+            sensitive.update(
+                (
+                    value,
+                    quote(value, safe=""),
+                    quote_plus(value),
+                    html.escape(value),
+                    json.dumps(value)[1:-1],
+                )
+            )
+
+    try:
+        try:
+            data = await response.content.readexactly(BODY_READ_LIMIT + 1)
+        except asyncio.IncompleteReadError as exc:
+            data = exc.partial
+        body = data.decode("utf-8", errors="replace")
+    except (aiohttp.ClientError, asyncio.TimeoutError):
+        body = "[response body unavailable]"
+
+    truncated = len(body) > BODY_EXCERPT_LIMIT
+    body = body[:BODY_EXCERPT_LIMIT]
+    url = str(URL(response.url).with_user(None).with_query(None).with_fragment(None))
+    for value in sorted(sensitive, key=len, reverse=True):
+        url = url.replace(value, "[redacted]")
+        body = body.replace(value, "[redacted]")
+        if truncated:
+            for length in range(min(len(value) - 1, len(body)), 0, -1):
+                if body.endswith(value[:length]):
+                    body = body[:-length] + "[redacted]"
+                    break
+    body = body[:BODY_EXCERPT_LIMIT]
+    if truncated:
+        body = body[: BODY_EXCERPT_LIMIT - 3] + "..."
+
+    if job_id is not None:
+        raise JobStatusError(
+            job_id, response.status, response.method, url, body
+        ) from None
+    raise AppHTTPError(response.method, url, response.status, body) from None

runpod/apps/utils/network.py

@@ -6,6 +6,7 @@
 import aiohttp
 
 from ...user_agent import APPS_USER_AGENT
+from .http import raise_for_status
 
 
 def runtime_headers() -> Optional[Dict[str, str]]:
@@ -65,5 +66,5 @@ async def get_json(url: str, headers: Dict[str, str], timeout: float) -> Dict[st
         async with session.get(
             url, headers={**headers, "User-Agent": APPS_USER_AGENT}
         ) as response:
-            response.raise_for_status()
+            await raise_for_status(response)
             return await response.json()