Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_missing_jsonrpc_version_on_reply(prot):
with pytest.raises(InvalidReplyError):
prot.parse_reply('{"result": 7, "id": "1"}')
to the standard.
"""
if isinstance(data, bytes):
data = data.decode()
try:
rep = json.loads(data)
except Exception as e:
raise InvalidReplyError(e)
for k in rep.keys():
if k not in self._ALLOWED_REPLY_KEYS:
raise InvalidReplyError('Key not allowed: %s' % k)
if 'jsonrpc' not in rep:
raise InvalidReplyError('Missing jsonrpc (version) in response.')
if rep['jsonrpc'] != self.JSON_RPC_VERSION:
raise InvalidReplyError('Wrong JSONRPC version')
if 'id' not in rep:
raise InvalidReplyError('Missing id in response')
if ('error' in rep) and ('result' in rep):
raise InvalidReplyError(
'Reply must contain exactly one of result and error.'
)
if 'error' in rep:
response = JSONRPCErrorResponse()
error = rep['error']
response.error = error["message"]
Called by the client to reconstruct the serialized :py:class:`JSONRPCResponse`.
:param bytes data: The data stream received by the transport layer containing the
serialized request.
:return: A reconstructed response.
:rtype: :py:class:`JSONRPCSuccessResponse` or :py:class:`JSONRPCErrorResponse`
:raises InvalidReplyError: if the response is not valid JSON or does not conform
to the standard.
"""
if isinstance(data, bytes):
data = data.decode()
try:
rep = json.loads(data)
except Exception as e:
raise InvalidReplyError(e)
for k in rep.keys():
if k not in self._ALLOWED_REPLY_KEYS:
raise InvalidReplyError('Key not allowed: %s' % k)
if 'jsonrpc' not in rep:
raise InvalidReplyError('Missing jsonrpc (version) in response.')
if rep['jsonrpc'] != self.JSON_RPC_VERSION:
raise InvalidReplyError('Wrong JSONRPC version')
if 'id' not in rep:
raise InvalidReplyError('Missing id in response')
if ('error' in rep) and ('result' in rep):
raise InvalidReplyError(
try:
rep = msgpack.unpackb(data, raw=False)
except Exception as e:
raise InvalidReplyError(e)
if len(rep) != 4:
raise InvalidReplyError("MSGPACKRPC spec requires reply of length 4")
if rep[0] != 1:
raise InvalidReplyError("Invalid MSGPACK message type")
if not isinstance(rep[1], int):
raise InvalidReplyError("Invalid or missing message ID in response")
if rep[2] is not None and rep[3] is not None:
raise InvalidReplyError("Reply must contain only one of result and error.")
if rep[2] is not None:
response = MSGPACKRPCErrorResponse()
if isinstance(rep[2], list) and len(rep[2]) == 2:
code, message = rep[2]
if isinstance(code, int) and isinstance(message, str):
response.error = str(message)
response._msgpackrpc_error_code = int(code)
else:
response.error = rep[2]
response._msgpackrpc_error_code = None
else:
response.error = rep[2]
response._msgpackrpc_error_code = None
else:
response = MSGPACKRPCSuccessResponse()
data = data.decode()
try:
rep = json.loads(data)
except Exception as e:
raise InvalidReplyError(e)
for k in rep.keys():
if k not in self._ALLOWED_REPLY_KEYS:
raise InvalidReplyError('Key not allowed: %s' % k)
if 'jsonrpc' not in rep:
raise InvalidReplyError('Missing jsonrpc (version) in response.')
if rep['jsonrpc'] != self.JSON_RPC_VERSION:
raise InvalidReplyError('Wrong JSONRPC version')
if 'id' not in rep:
raise InvalidReplyError('Missing id in response')
if ('error' in rep) and ('result' in rep):
raise InvalidReplyError(
'Reply must contain exactly one of result and error.'
)
if 'error' in rep:
response = JSONRPCErrorResponse()
error = rep['error']
response.error = error["message"]
response._jsonrpc_error_code = error["code"]
if "data" in error:
response.data = error["data"]
) -> Union["MSGPACKRPCSuccessResponse", "MSGPACKRPCErrorResponse"]:
"""De-serializes and validates a response.
Called by the client to reconstruct the serialized :py:class:`MSGPACKRPCResponse`.
:param bytes data: The data stream received by the transport layer containing the
serialized response.
:return: A reconstructed response.
:rtype: :py:class:`MSGPACKRPCSuccessResponse` or :py:class:`MSGPACKRPCErrorResponse`
:raises InvalidReplyError: if the response is not valid MSGPACK or does not conform
to the standard.
"""
try:
rep = msgpack.unpackb(data, raw=False)
except Exception as e:
raise InvalidReplyError(e)
if len(rep) != 4:
raise InvalidReplyError("MSGPACKRPC spec requires reply of length 4")
if rep[0] != 1:
raise InvalidReplyError("Invalid MSGPACK message type")
if not isinstance(rep[1], int):
raise InvalidReplyError("Invalid or missing message ID in response")
if rep[2] is not None and rep[3] is not None:
raise InvalidReplyError("Reply must contain only one of result and error.")
if rep[2] is not None:
response = MSGPACKRPCErrorResponse()
if isinstance(rep[2], list) and len(rep[2]) == 2:
:raises InvalidReplyError: if the response is not valid MSGPACK or does not conform
to the standard.
"""
try:
rep = msgpack.unpackb(data, raw=False)
except Exception as e:
raise InvalidReplyError(e)
if len(rep) != 4:
raise InvalidReplyError("MSGPACKRPC spec requires reply of length 4")
if rep[0] != 1:
raise InvalidReplyError("Invalid MSGPACK message type")
if not isinstance(rep[1], int):
raise InvalidReplyError("Invalid or missing message ID in response")
if rep[2] is not None and rep[3] is not None:
raise InvalidReplyError("Reply must contain only one of result and error.")
if rep[2] is not None:
response = MSGPACKRPCErrorResponse()
if isinstance(rep[2], list) and len(rep[2]) == 2:
code, message = rep[2]
if isinstance(code, int) and isinstance(message, str):
response.error = str(message)
response._msgpackrpc_error_code = int(code)
else:
response.error = rep[2]
response._msgpackrpc_error_code = None
else:
response.error = rep[2]
for k in rep.keys():
if k not in self._ALLOWED_REPLY_KEYS:
raise InvalidReplyError('Key not allowed: %s' % k)
if 'jsonrpc' not in rep:
raise InvalidReplyError('Missing jsonrpc (version) in response.')
if rep['jsonrpc'] != self.JSON_RPC_VERSION:
raise InvalidReplyError('Wrong JSONRPC version')
if 'id' not in rep:
raise InvalidReplyError('Missing id in response')
if ('error' in rep) and ('result' in rep):
raise InvalidReplyError(
'Reply must contain exactly one of result and error.'
)
if 'error' in rep:
response = JSONRPCErrorResponse()
error = rep['error']
response.error = error["message"]
response._jsonrpc_error_code = error["code"]
if "data" in error:
response.data = error["data"]
else:
response = JSONRPCSuccessResponse()
response.result = rep.get('result', None)
response.unique_id = rep['id']
rep = json.loads(data)
except Exception as e:
raise InvalidReplyError(e)
for k in rep.keys():
if k not in self._ALLOWED_REPLY_KEYS:
raise InvalidReplyError('Key not allowed: %s' % k)
if 'jsonrpc' not in rep:
raise InvalidReplyError('Missing jsonrpc (version) in response.')
if rep['jsonrpc'] != self.JSON_RPC_VERSION:
raise InvalidReplyError('Wrong JSONRPC version')
if 'id' not in rep:
raise InvalidReplyError('Missing id in response')
if ('error' in rep) and ('result' in rep):
raise InvalidReplyError(
'Reply must contain exactly one of result and error.'
)
if 'error' in rep:
response = JSONRPCErrorResponse()
error = rep['error']
response.error = error["message"]
response._jsonrpc_error_code = error["code"]
if "data" in error:
response.data = error["data"]
else:
response = JSONRPCSuccessResponse()
response.result = rep.get('result', None)
Called by the client to reconstruct the serialized :py:class:`MSGPACKRPCResponse`.
:param bytes data: The data stream received by the transport layer containing the
serialized response.
:return: A reconstructed response.
:rtype: :py:class:`MSGPACKRPCSuccessResponse` or :py:class:`MSGPACKRPCErrorResponse`
:raises InvalidReplyError: if the response is not valid MSGPACK or does not conform
to the standard.
"""
try:
rep = msgpack.unpackb(data, raw=False)
except Exception as e:
raise InvalidReplyError(e)
if len(rep) != 4:
raise InvalidReplyError("MSGPACKRPC spec requires reply of length 4")
if rep[0] != 1:
raise InvalidReplyError("Invalid MSGPACK message type")
if not isinstance(rep[1], int):
raise InvalidReplyError("Invalid or missing message ID in response")
if rep[2] is not None and rep[3] is not None:
raise InvalidReplyError("Reply must contain only one of result and error.")
if rep[2] is not None:
response = MSGPACKRPCErrorResponse()
if isinstance(rep[2], list) and len(rep[2]) == 2:
code, message = rep[2]
if isinstance(code, int) and isinstance(message, str):
response.error = str(message)