import copy
import json
import logging
import math
import random
import string
from collections.abc import (
Iterable,
Mapping,
Sequence,
)
from decimal import Decimal
from . import unicodify
__all__ = ("safe_dumps", "validate_jsonrpc_request", "validate_jsonrpc_response", "jsonrpc_request", "jsonrpc_response")
log = logging.getLogger(__name__)
to_json_string = json.dumps
from_json_string = json.loads
def swap_inf_nan(val):
"""
This takes an arbitrary object and preps it for jsonifying safely, templating Inf/NaN and
casting Decimal instances as strings.
"""
if isinstance(val, str):
# basestring first, because it's a sequence and would otherwise get caught below.
return val
elif isinstance(val, Sequence):
return [swap_inf_nan(v) for v in val]
elif isinstance(val, Mapping):
return {swap_inf_nan(k): swap_inf_nan(v) for (k, v) in val.items()}
elif isinstance(val, float):
if math.isnan(val):
return "__NaN__"
elif val == math.inf:
return "__Infinity__"
elif val == -math.inf:
return "__-Infinity__"
else:
return val
elif isinstance(val, Decimal):
return str(val)
else:
return val
def safe_loads(arg):
"""
This is a wrapper around loads that returns the parsed value instead of
raising a value error. It also avoids autoconversion of non-iterables
i.e numeric and boolean values.
"""
try:
loaded = json.loads(arg)
if loaded is not None and not isinstance(loaded, Iterable):
loaded = arg
except (TypeError, ValueError):
loaded = copy.deepcopy(arg)
return loaded
[docs]def safe_dumps(obj, **kwargs):
"""
This is a wrapper around dumps that encodes Infinity and NaN values. It's a
fairly rare case (which will be low in request volume). Basically, we tell
json.dumps to blow up if it encounters Infinity/NaN, or Decimal values
and we 'fix' it before re-encoding.
"""
try:
dumped = json.dumps(obj, allow_nan=False, **kwargs)
except (ValueError, TypeError):
obj = swap_inf_nan(obj)
dumped = json.dumps(obj, allow_nan=False, **kwargs)
if kwargs.get("escape_closing_tags", True):
return dumped.replace("</", "<\\/")
return dumped
def safe_dumps_formatted(obj):
"""Attempt to format an object for display.
If serialization fails, the object's string representation will be returned instead.
"""
try:
return safe_dumps(obj, sort_keys=True, indent=4, separators=(",", ": "))
except TypeError:
log.warning("JSON serialization failed for object: %s", str(obj))
return str(obj)
# Methods for handling JSON-RPC
[docs]def validate_jsonrpc_request(request, regular_methods, notification_methods):
try:
request = json.loads(request)
except Exception as e:
return (
False,
request,
jsonrpc_response(id=None, error=dict(code=-32700, message="Parse error", data=unicodify(e))),
)
try:
assert (
"jsonrpc" in request
), 'This server requires JSON-RPC 2.0 and no "jsonrpc" member was sent with the Request object as per the JSON-RPC 2.0 Specification.'
assert (
request["jsonrpc"] == "2.0"
), f"Requested JSON-RPC version \"{request['jsonrpc']}\" != required version \"2.0\"."
assert "method" in request, 'No "method" member was sent with the Request object'
except AssertionError as e:
return (
False,
request,
jsonrpc_response(request=request, error=dict(code=-32600, message="Invalid Request", data=unicodify(e))),
)
try:
assert request["method"] in (regular_methods + notification_methods)
except AssertionError:
return (
False,
request,
jsonrpc_response(
request=request,
error=dict(
code=-32601,
message="Method not found",
data=f"Valid methods are: {', '.join(regular_methods + notification_methods)}",
),
),
)
try:
if request["method"] in regular_methods:
assert (
"id" in request
), f"No \"id\" member was sent with the Request object and the requested method \"{request['method']}\" is not a notification method"
except AssertionError as e:
return (
False,
request,
jsonrpc_response(request=request, error=dict(code=-32600, message="Invalid Request", data=unicodify(e))),
)
return True, request, None
[docs]def validate_jsonrpc_response(response, id=None):
try:
response = json.loads(response)
except Exception as e:
log.error("Response was not valid JSON: %s", unicodify(e))
log.debug("Response was: %s", response)
return False, response
try:
assert (
"jsonrpc" in response
), 'This server requires JSON-RPC 2.0 and no "jsonrpc" member was sent with the Response object as per the JSON-RPC 2.0 Specification.'
assert (
"result" in response or "error" in response
), 'Neither of "result" or "error" members were sent with the Response object.'
if "error" in response:
assert int(
response["error"]["code"]
), 'The "code" member of the "error" object in the Response is missing or not an integer.'
assert "message" in response, 'The "message" member of the "error" object in the Response is missing.'
except Exception:
log.exception("Response was not valid JSON-RPC")
log.debug(f"Response was: {response}")
return False, response
if id is not None:
try:
assert "id" in response and response["id"] == id
except Exception:
log.error(f"The response id \"{response['id']}\" does not match the request id \"{id}\"")
return False, response
return True, response
[docs]def jsonrpc_request(method, params=None, id=None, jsonrpc="2.0"):
if method is None:
log.error('jsonrpc_request(): "method" parameter cannot be None')
return None
request = dict(jsonrpc=jsonrpc, method=method)
if params:
request["params"] = params
if id is not None and id is True:
request["id"] = "".join(random.choice(string.hexdigits) for i in range(16))
elif id is not None:
request["id"] = id
return request
[docs]def jsonrpc_response(request=None, id=None, result=None, error=None, jsonrpc="2.0"):
if result:
rval = dict(jsonrpc=jsonrpc, result=result)
elif error:
rval = dict(jsonrpc=jsonrpc, error=error)
else:
msg = 'jsonrpc_response() called with out a "result" or "error" parameter'
log.error(msg)
rval = dict(jsonrpc=jsonrpc, error=msg)
if id is not None:
rval["id"] = id
elif request is not None and "id" in request:
rval["id"] = request["id"]
return rval