Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions src/falconpy/_error/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@
InvalidCredentialFormat,
InvalidRoute,
InvalidServiceCollection,
InvalidOperationSearch
InvalidOperationSearch,
ContentDecodingError
)
from ._warnings import (
SDKWarning,
Expand All @@ -74,5 +75,6 @@
"NoAuthenticationMechanism", "FeatureNotSupportedByPythonVersion",
"InvalidIndex", "InvalidCredentialFormat", "UnnecessaryEncodingUsed",
"SDKDeprecationWarning", "DeprecatedOperation", "DeprecatedClass",
"InvalidRoute", "InvalidServiceCollection", "InvalidOperationSearch"
"InvalidRoute", "InvalidServiceCollection", "InvalidOperationSearch",
"ContentDecodingError"
]
45 changes: 45 additions & 0 deletions src/falconpy/_error/_exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -201,3 +201,48 @@ class InvalidCredentialFormat(SDKError):
"""Credentials dictionary was provided as a datatype that will not convert to a dictionary."""

_message = "Invalid credential format. This keyword must be provided as a dictionary."


class ContentDecodingError(SDKError):
"""The API response body could not be decoded with the expected encoding.
This typically occurs when the API returns a JSON response containing
non-ASCII characters (e.g. accented names) and the SDK attempts to
decode the raw bytes with the wrong codec (Issue #1298).
"""

_code = 500
_message = "Unable to decode API response content."

def __init__(self,
code: int = None,
message: str = None,
headers: dict = None,
encoding: str = None,
position: int = None
):
"""Construct an instance of the exception.
Keyword arguments:
encoding -- The codec that failed (e.g. 'ascii', 'utf-8').
position -- The byte position where decoding failed.
"""
self._encoding = encoding
self._position = position
if not message and encoding:
message = (
f"Unable to decode API response using '{encoding}' codec"
f"{f' at byte position {position}' if position is not None else ''}"
". The response likely contains non-ASCII characters."
)
super().__init__(code=code, message=message, headers=headers)

@property
def encoding(self) -> str:
"""Return the codec that failed."""
return self._encoding

@property
def position(self) -> int:
"""Return the byte position where decoding failed."""
return self._position
30 changes: 29 additions & 1 deletion src/falconpy/_error/_warnings.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,11 +89,39 @@ class SSLDisabledWarning(SDKWarning):


class NoContentWarning(SDKWarning):
"""No content was received."""
"""No content was received or the content could not be decoded.
When content decoding fails, the encoding that was attempted is
preserved in this warning for diagnostic purposes (Issue #1298).
"""

_code = 204
_message = "No content was received for this request."

def __init__(self,
code: int = None,
message: str = None,
headers: dict = None,
encoding: str = None
):
"""Construct an instance of the warning.
Keyword arguments:
encoding -- The encoding that was attempted when decoding failed.
"""
self._encoding = encoding
if encoding and not message:
message = (
f"No content could be decoded for this request "
f"(attempted encoding: {encoding})."
)
super().__init__(code=code, message=message, headers=headers)

@property
def encoding(self) -> str:
"""Return the encoding that was attempted, if available."""
return self._encoding


class NoAuthenticationMechanism(SDKWarning):
"""No authentication mechanism was specified when creating this class."""
Expand Down
27 changes: 26 additions & 1 deletion src/falconpy/_result/_result.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,14 +78,39 @@ def __init__(self,
self.headers = Headers(_headers)
self._parse_body(body_rcv=body)

@staticmethod
def _safe_decode_body(body: bytes) -> str:
"""Safely decode response body bytes for Result construction.

Tries UTF-8 first (RFC 8259 standard for JSON), then falls
back to latin-1 which maps every byte 0x00-0xFF and never
raises UnicodeDecodeError. This prevents crashes when the
API returns content with non-ASCII characters (Issue #1298).
"""
try:
return body.decode("utf-8")
except (UnicodeDecodeError, ValueError):
return body.decode("latin-1")

def _parse_body(self, body_rcv: Dict[str, Union[str, dict, list, int, float, bytes]]):
if isinstance(body_rcv, list):
# Specific to report_executions_download_get returning raw
# JSON payloads as a list. There will be no Meta or Errors
# branch in this response.
self.resources = Resources(body_rcv) # pragma: no cover
elif isinstance(body_rcv, bytes):
# Binary response
# Binary response — attempt safe text decoding first so that
# non-ASCII JSON payloads can still be parsed (Issue #1298).
decoded = self._safe_decode_body(body_rcv)
try:
from json import loads as _loads
parsed = _loads(decoded)
if isinstance(parsed, dict):
self._parse_body(body_rcv=parsed)
return
except (ValueError, TypeError):
pass
# Genuine binary data (e.g. file downloads)
self.resources = BinaryFile(body_rcv)
self.meta = Meta()
self.errors = Errors()
Expand Down
38 changes: 37 additions & 1 deletion src/falconpy/_util/_functions.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@
from .._error import (
RegionSelectError,
SDKError,
ContentDecodingError,
InvalidMethod,
KeywordsOnly,
APIError,
Expand Down Expand Up @@ -258,6 +259,23 @@ def service_request(caller: ServiceClass = None, **kwargs) -> Union[Dict[str, Un
)


def _safe_decode_content(content: bytes) -> str:
"""Safely decode response content bytes to a string.

Tries UTF-8 first (the standard encoding for JSON per RFC 8259),
then falls back to latin-1 which never raises UnicodeDecodeError
because every byte 0x00-0xFF is a valid latin-1 code point.

This replaces the previous ``content.decode("ascii")`` call that
crashed on any response containing non-ASCII characters such as
accented names (Issue #1298).
"""
try:
return content.decode("utf-8")
except (UnicodeDecodeError, ValueError):
return content.decode("latin-1")


# pylint: disable=R0912 # I don't disagree, but this will work for now.
def calc_content_return(resp: requests.Response,
contain: bool,
Expand All @@ -280,7 +298,7 @@ def calc_content_return(resp: requests.Response,
except JSONDecodeError:
if api_method != "HEAD":
# It says JSON in the headers but it came back to us as a binary string.
json_resp = loads(resp.content.decode("ascii"))
json_resp = loads(_safe_decode_content(resp.content))
finally:
# Default behavior is to return results as a standardized dictionary.
returned = Result(status_code=resp.status_code,
Expand Down Expand Up @@ -473,6 +491,24 @@ def perform_request(endpoint: str = "", # noqa: C901
code=response.status_code
) from json_decode_error

except UnicodeDecodeError as decode_error:
# The response contains bytes that could not be decoded.
# This commonly happens when the API returns JSON with
# non-ASCII characters (e.g. accented names) and a
# fallback codec is too restrictive (Issue #1298).
_enc = getattr(decode_error, 'encoding', 'unknown')
_pos = getattr(decode_error, 'start', None)
api.log_warning(
f"WARNING: Content decoding failed "
f"(encoding={_enc}, position={_pos})"
)
raise ContentDecodingError(
code=500,
headers=api.debug_headers,
encoding=_enc,
position=_pos
) from decode_error

except Exception as havoc: # pylint: disable=W0703
# General catch-all for anything coming ____ ____ _ _ \\ o o
# out of requests or the library itself. |___ |--< Y || |\O/|
Expand Down
9 changes: 8 additions & 1 deletion src/falconpy/api_complete/_legacy.py
Original file line number Diff line number Diff line change
Expand Up @@ -220,7 +220,14 @@ def authenticate(self: object) -> bool:
self._token_fail_headers = result["headers"]
if "errors" in result["body"]:
if result["body"]["errors"]:
self.token_fail_reason = result["body"]["errors"][0]["message"]
err = result["body"]["errors"][0]
# Handle both standard API error dicts and
# non-standard entries (e.g. from decode failures).
if isinstance(err, dict):
self.token_fail_reason = err.get(
"message", str(err))
else:
self.token_fail_reason = str(err)
else: # pragma: no cover
self.authenticated = False
self.token_fail_reason = TokenFailReason["UNEXPECTED"].value
Expand Down
Loading