# Copyright (c) 2016-2017, Neil Booth # # All rights reserved. # # The MIT License (MIT) # # Permission is hereby granted, free of charge, to any person obtaining # a copy of this software and associated documentation files (the # "Software"), to deal in the Software without restriction, including # without limitation the rights to use, copy, modify, merge, publish, # distribute, sublicense, and/or sell copies of the Software, and to # permit persons to whom the Software is furnished to do so, subject to # the following conditions: # # The above copyright notice and this permission notice shall be # included in all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND # NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE # LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION # OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION # WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. '''Classes for acting as a peer over a transport and speaking the JSON RPC versions 1.0 and 2.0. JSONSessionBase can use an arbitrary transport. JSONSession integrates asyncio.Protocol to provide the transport. ''' import asyncio import collections import inspect import json import numbers import time import traceback import lib.util as util class RPCError(Exception): '''RPC handlers raise this error.''' def __init__(self, msg, code=-1, **kw_args): super().__init__(**kw_args) self.msg = msg self.code = code class JSONRPC(object): '''Base class of JSON RPC versions.''' # See http://www.jsonrpc.org/specification PARSE_ERROR = -32700 INVALID_REQUEST = -32600 METHOD_NOT_FOUND = -32601 INVALID_ARGS = -32602 INTERNAL_ERROR = -32603 # Codes for this library INVALID_RESPONSE = -100 ERROR_CODE_UNAVAILABLE = -101 REQUEST_TIMEOUT = -102 ID_TYPES = (type(None), str, numbers.Number) HAS_BATCHES = False @classmethod def canonical_error(cls, error): '''Convert an error to a JSON RPC 2.0 error. Handlers then only have a single form of error to deal with. ''' if isinstance(error, int): error = {'code': error} elif isinstance(error, str): error = {'message': error} elif not isinstance(error, dict): error = {'data': error} error['code'] = error.get('code', JSONRPC.ERROR_CODE_UNAVAILABLE) error['message'] = error.get('message', 'error message unavailable') return error @classmethod def timeout_error(cls): return {'message': 'request timed out', 'code': JSONRPC.REQUEST_TIMEOUT} class JSONRPCv1(JSONRPC): '''JSON RPC version 1.0.''' @classmethod def request_payload(cls, id_, method, params=None): '''JSON v1 request payload. Params is mandatory.''' return {'method': method, 'params': params or [], 'id': id_} @classmethod def notification_payload(cls, method, params=None): '''JSON v1 notification payload. Params and id are mandatory.''' return {'method': method, 'params': params or [], 'id': None} @classmethod def response_payload(cls, result, id_): '''JSON v1 response payload. error is present and None.''' return {'id': id_, 'result': result, 'error': None} @classmethod def error_payload(cls, message, code, id_): '''JSON v1 error payload. result is present and None.''' return {'id': id_, 'result': None, 'error': {'message': message, 'code': code}} @classmethod def handle_response(cls, handler, payload): '''JSON v1 response handler. Both 'error' and 'result' should exist with exactly one being None. Unfortunately many 1.0 clients behave like 2.0, and just send one or the other. ''' error = payload.get('error') if error is None: handler(payload.get('result'), None) else: handler(None, cls.canonical_error(error)) @classmethod def is_request(cls, payload): '''Returns True if the payload (which has a method) is a request. False means it is a notification.''' return payload.get('id') is not None class JSONRPCv2(JSONRPC): '''JSON RPC version 2.0.''' HAS_BATCHES = True @classmethod def request_payload(cls, id_, method, params=None): '''JSON v2 request payload. Params is optional.''' payload = {'jsonrpc': '2.0', 'method': method, 'id': id_} if params: payload['params'] = params return payload @classmethod def notification_payload(cls, method, params=None): '''JSON v2 notification payload. There must be no id.''' payload = {'jsonrpc': '2.0', 'method': method} if params: payload['params'] = params return payload @classmethod def response_payload(cls, result, id_): '''JSON v2 response payload. error is not present.''' return {'jsonrpc': '2.0', 'id': id_, 'result': result} @classmethod def error_payload(cls, message, code, id_): '''JSON v2 error payload. result is not present.''' return {'jsonrpc': '2.0', 'id': id_, 'error': {'message': message, 'code': code}} @classmethod def handle_response(cls, handler, payload): '''JSON v2 response handler. Exactly one of 'error' and 'result' must exist. Errors must have 'code' and 'message' members. ''' if 'error' in payload: handler(None, cls.canonical_error(payload['error'])) elif 'result' in payload: handler(payload['result'], None) else: error = {'message': 'no error or result returned', 'code': JSONRPC.INVALID_RESPONSE} handler(None, cls.canonical_error(error)) @classmethod def batch_size(cls, parts): '''Return the size of a JSON batch from its parts.''' return sum(len(part) for part in parts) + 2 * len(parts) @classmethod def batch_bytes(cls, parts): '''Return the bytes of a JSON batch from its parts.''' if parts: return b'[' + b', '.join(parts) + b']' return b'' @classmethod def is_request(cls, payload): '''Returns True if the payload (which has a method) is a request. False means it is a notification.''' return 'id' in payload class JSONRPCCompat(JSONRPC): '''Intended to be used until receiving a response from the peer, at which point detect_version should be used to choose which version to use. Sends requests compatible with v1 and v2. Errors cannot be compatible so v2 errors are sent. Does not send responses or notifications, nor handle responses. ''' @classmethod def request_payload(cls, id_, method, params=None): '''JSON v2 request payload but with params present.''' return {'jsonrpc': '2.0', 'id': id_, 'method': method, 'params': params or []} @classmethod def error_payload(cls, message, code, id_): '''JSON v2 error payload. result is not present.''' return {'jsonrpc': '2.0', 'id': id_, 'error': {'message': message, 'code': code}} @classmethod def detect_version(cls, payload): '''Return a best guess at a version compatible with the received payload. Return None if one cannot be determined. ''' def item_version(item): if isinstance(item, dict): version = item.get('jsonrpc') if version is None: return JSONRPCv1 if version == '2.0': return JSONRPCv2 return None if isinstance(payload, list) and payload: version = item_version(payload[0]) # If a batch return at least JSONRPCv2 if version in (JSONRPCv1, None): version = JSONRPCv2 else: version = item_version(payload) return version class JSONSessionBase(util.LoggedClass): '''Acts as the application layer session, communicating via JSON RPC over an underlying transport. Processes incoming and sends outgoing requests, notifications and responses. Incoming messages are queued. When the queue goes from empty ''' _next_session_id = 0 _pending_reqs = {} # Outgoing requests waiting for a response @classmethod def next_session_id(cls): '''Return the next unique session ID.''' session_id = cls._next_session_id cls._next_session_id += 1 return session_id def _pending_request_keys(self): '''Return a generator of pending request keys for this session.''' return [key for key in self._pending_reqs if key[0] is self] def has_pending_requests(self): '''Return True if this session has pending requests.''' return bool(self._pending_request_keys()) def pop_response_handler(self, msg_id): '''Return the response handler for the given message ID.''' return self._pending_reqs.pop((self, msg_id), (None, None))[0] def timeout_session(self): '''Trigger timeouts for all of the session's pending requests.''' self._timeout_requests(self._pending_request_keys()) @classmethod def timeout_check(cls): '''Trigger timeouts where necessary for all pending requests.''' now = time.time() keys = [key for key, value in cls._pending_reqs.items() if value[1] < now] cls._timeout_requests(keys) @classmethod def _timeout_requests(cls, keys): '''Trigger timeouts for the given lookup keys.''' values = [cls._pending_reqs.pop(key) for key in keys] handlers = [handler for handler, timeout in values] timeout_error = JSONRPC.timeout_error() for handler in handlers: handler(None, timeout_error) def __init__(self, version=JSONRPCCompat): super().__init__() # Parts of an incomplete JSON line. We buffer them until # getting a newline. self.parts = [] self.version = version self.log_me = False self.session_id = None # Count of incoming complete JSON requests and the time of the # last one. A batch counts as just one here. self.last_recv = time.time() self.send_count = 0 self.send_size = 0 self.recv_size = 0 self.recv_count = 0 self.error_count = 0 self.pause = False # Handling of incoming items self.items = collections.deque() self.items_event = asyncio.Event() self.batch_results = [] # Handling of outgoing requests self.next_request_id = 0 # If buffered incoming data exceeds this the connection is closed self.max_buffer_size = 1000000 self.max_send = 50000 self.close_after_send = False def pause_writing(self): '''Transport calls when the send buffer is full.''' self.log_info('pausing processing whilst socket drains') self.pause = True def resume_writing(self): '''Transport calls when the send buffer has room.''' self.log_info('resuming processing') self.pause = False def is_oversized(self, length, id_): '''Return an error payload if the given outgoing message size is too large, or False if not. ''' if self.max_send and length > max(1000, self.max_send): msg = 'response too large (at least {:d} bytes)'.format(length) return self.error_bytes(msg, JSONRPC.INVALID_REQUEST, id_) return False def send_binary(self, binary): '''Pass the bytes through to the transport. Close the connection if close_after_send is set. ''' if self.is_closing(): return self.using_bandwidth(len(binary)) self.send_count += 1 self.send_size += len(binary) self.send_bytes(binary) if self.close_after_send: self.close_connection() def payload_id(self, payload): '''Extract and return the ID from the payload. Returns None if it is missing or invalid.''' try: return self.check_payload_id(payload) except RPCError: return None def check_payload_id(self, payload): '''Extract and return the ID from the payload. Raises an RPCError if it is missing or invalid.''' if 'id' not in payload: raise RPCError('missing id', JSONRPC.INVALID_REQUEST) id_ = payload['id'] if not isinstance(id_, self.version.ID_TYPES): raise RPCError('invalid id type {}'.format(type(id_)), JSONRPC.INVALID_REQUEST) return id_ def request_bytes(self, id_, method, params=None): '''Return the bytes of a JSON request.''' payload = self.version.request_payload(id_, method, params) return self.encode_payload(payload) def notification_bytes(self, method, params=None): payload = self.version.notification_payload(method, params) return self.encode_payload(payload) def response_bytes(self, result, id_): '''Return the bytes of a JSON response.''' return self.encode_payload(self.version.response_payload(result, id_)) def error_bytes(self, message, code, id_=None): '''Return the bytes of a JSON error. Flag the connection to close on a fatal error or too many errors.''' version = self.version self.error_count += 1 if not self.close_after_send: fatal_log = None if code in (version.PARSE_ERROR, version.INVALID_REQUEST): fatal_log = message elif self.error_count >= 10: fatal_log = 'too many errors, last: {}'.format(message) if fatal_log: self.log_info(fatal_log) self.close_after_send = True return self.encode_payload(self.version.error_payload (message, code, id_)) def encode_payload(self, payload): '''Encode a Python object as binary bytes.''' assert isinstance(payload, dict) id_ = payload.get('id') try: binary = json.dumps(payload).encode() except TypeError: msg = 'JSON encoding failure: {}'.format(payload) self.log_error(msg) binary = self.error_bytes(msg, JSONRPC.INTERNAL_ERROR, id_) error_bytes = self.is_oversized(len(binary), id_) return error_bytes or binary def decode_message(self, payload): '''Decode a binary message and pass it on to process_single_item or process_batch as appropriate. Messages that cannot be decoded are logged and dropped. ''' try: payload = payload.decode() except UnicodeDecodeError as e: msg = 'cannot decode message: {}'.format(e) self.send_error(msg, JSONRPC.PARSE_ERROR) return try: payload = json.loads(payload) except json.JSONDecodeError as e: msg = 'cannot decode JSON: {}'.format(e) self.send_error(msg, JSONRPC.PARSE_ERROR) return if self.version is JSONRPCCompat: # Attempt to detect peer's JSON RPC version version = self.version.detect_version(payload) if not version: version = JSONRPCv2 self.log_info('unable to detect JSON RPC version, using 2.0') self.version = version # Batches must have at least one object. if payload == [] and self.version.HAS_BATCHES: self.send_error('empty batch', JSONRPC.INVALID_REQUEST) return self.items.append(payload) self.items_event.set() async def process_batch(self, batch, count): '''Processes count items from the batch according to the JSON 2.0 spec. If any remain, puts what is left of the batch back in the deque and returns None. Otherwise returns the binary batch result.''' results = self.batch_results self.batch_results = [] for n in range(count): item = batch.pop() result = await self.process_single_item(item) if result: results.append(result) if not batch: return self.version.batch_bytes(results) error_bytes = self.is_oversized(self.batch_size(results), None) if error_bytes: return error_bytes self.items.appendleft(item) self.batch_results = results return None async def process_single_item(self, payload): '''Handle a single JSON request, notification or response. If it is a request, return the binary response, oterhwise None.''' if self.log_me: self.log_info('processing {}'.format(payload)) if not isinstance(payload, dict): return self.error_bytes('request must be a dictionary', JSONRPC.INVALID_REQUEST) try: # Requests and notifications must have a method. if 'method' in payload: if self.version.is_request(payload): return await self.process_single_request(payload) else: await self.process_single_notification(payload) else: self.process_single_response(payload) return None except asyncio.CancelledError: raise except Exception: self.log_error(traceback.format_exc()) return self.error_bytes('internal error processing request', JSONRPC.INTERNAL_ERROR, self.payload_id(payload)) async def process_single_request(self, payload): '''Handle a single JSON request and return the binary response.''' try: result = await self.handle_payload(payload, self.request_handler) return self.response_bytes(result, payload['id']) except RPCError as e: return self.error_bytes(e.msg, e.code, self.payload_id(payload)) except asyncio.CancelledError: raise except Exception: self.log_error(traceback.format_exc()) return self.error_bytes('internal error processing request', JSONRPC.INTERNAL_ERROR, self.payload_id(payload)) async def process_single_notification(self, payload): '''Handle a single JSON notification.''' try: await self.handle_payload(payload, self.notification_handler) except RPCError: pass except Exception: self.log_error(traceback.format_exc()) def process_single_response(self, payload): '''Handle a single JSON response.''' try: id_ = self.check_payload_id(payload) handler = self.pop_response_handler(id_) if handler: self.version.handle_response(handler, payload) else: self.log_info('response for unsent id {}'.format(id_), throttle=True) except RPCError: pass except Exception: self.log_error(traceback.format_exc()) async def handle_payload(self, payload, get_handler): '''Handle a request or notification payload given the handlers.''' # An argument is the value passed to a function parameter... args = payload.get('params', []) method = payload.get('method') if not isinstance(method, str): raise RPCError("invalid method type {}".format(type(method)), JSONRPC.INVALID_REQUEST) handler = get_handler(method) if not handler: raise RPCError("unknown method: '{}'".format(method), JSONRPC.METHOD_NOT_FOUND) if not isinstance(args, (list, dict)): raise RPCError('arguments should be an array or dictionary', JSONRPC.INVALID_REQUEST) params = inspect.signature(handler).parameters names = list(params) min_args = sum(p.default is p.empty for p in params.values()) if len(args) < min_args: raise RPCError('too few arguments to {}: expected {:d} got {:d}' .format(method, min_args, len(args)), JSONRPC.INVALID_ARGS) if len(args) > len(params): raise RPCError('too many arguments to {}: expected {:d} got {:d}' .format(method, len(params), len(args)), JSONRPC.INVALID_ARGS) if isinstance(args, list): kw_args = {name: arg for name, arg in zip(names, args)} else: kw_args = args bad_names = ['<{}>'.format(name) for name in args if name not in names] if bad_names: raise RPCError('invalid parameter names: {}' .format(', '.join(bad_names))) if inspect.iscoroutinefunction(handler): return await handler(**kw_args) else: return handler(**kw_args) # ---- External Interface ---- async def process_pending_items(self, limit=8): '''Processes up to LIMIT pending items asynchronously.''' while limit > 0 and self.items: item = self.items.popleft() if isinstance(item, list) and self.version.HAS_BATCHES: count = min(limit, len(item)) binary = await self.process_batch(item, count) limit -= count else: binary = await self.process_single_item(item) limit -= 1 if binary: self.send_binary(binary) if not self.items: self.items_event.clear() def count_pending_items(self): '''Counts the number of pending items.''' return sum(len(item) if isinstance(item, list) else 1 for item in self.items) def connection_made(self): '''Call when an incoming client connection is established.''' self.session_id = self.next_session_id() self.log_prefix = '[{:d}] '.format(self.session_id) def data_received(self, data): '''Underlying transport calls this when new data comes in. Look for newline separators terminating full requests. ''' if self.is_closing(): return self.using_bandwidth(len(data)) self.recv_size += len(data) # Close abusive connections where buffered data exceeds limit buffer_size = len(data) + sum(len(part) for part in self.parts) if buffer_size > self.max_buffer_size: self.log_error('read buffer of {:,d} bytes over {:,d} byte limit' .format(buffer_size, self.max_buffer_size)) self.close_connection() return while True: npos = data.find(ord('\n')) if npos == -1: self.parts.append(data) break tail, data = data[:npos], data[npos + 1:] parts, self.parts = self.parts, [] parts.append(tail) self.recv_count += 1 self.last_recv = time.time() self.decode_message(b''.join(parts)) def send_error(self, message, code, id_=None): '''Send a JSON error.''' self.send_binary(self.error_bytes(message, code, id_)) def send_request(self, handler, method, params=None, timeout=30): '''Sends a request and arranges for handler to be called with the response when it comes in. A call to request_timeout_check() will result in pending responses that have been waiting more than timeout seconds to call the handler with a REQUEST_TIMEOUT error. ''' id_ = self.next_request_id self.next_request_id += 1 self.send_binary(self.request_bytes(id_, method, params)) self._pending_reqs[(self, id_)] = (handler, time.time() + timeout) def send_notification(self, method, params=None): '''Send a notification.''' self.send_binary(self.notification_bytes(method, params)) def send_notifications(self, mp_iterable): '''Send an iterable of (method, params) notification pairs. A 1-tuple is also valid in which case there are no params.''' if False and self.version.HAS_BATCHES: parts = [self.notification_bytes(*pair) for pair in mp_iterable] self.send_binary(self.version.batch_bytes(parts)) else: for pair in mp_iterable: self.send_notification(*pair) # -- derived classes are intended to override these functions # Transport layer def is_closing(self): '''Return True if the underlying transport is closing.''' raise NotImplementedError def close_connection(self): '''Close the connection.''' raise NotImplementedError def send_bytes(self, binary): '''Pass the bytes through to the underlying transport.''' raise NotImplementedError # App layer def using_bandwidth(self, amount): '''Called as bandwidth is consumed. Override to implement bandwidth management. ''' pass def notification_handler(self, method): '''Return the handler for the given notification. The handler can be synchronous or asynchronous.''' return None def request_handler(self, method): '''Return the handler for the given request method. The handler can be synchronous or asynchronous.''' return None class JSONSession(JSONSessionBase, asyncio.Protocol): '''A JSONSessionBase instance specialized for use with asyncio.protocol to implement the transport layer. The app should await on items_event, which is set when unprocessed incoming items remain and cleared when the queue is empty, and then arrange to call process_pending_items asynchronously. Derived classes may want to override the request and notification handlers. ''' def __init__(self, version=JSONRPCCompat): super().__init__(version=version) self.transport = None self.write_buffer_high = 500000 def peer_info(self): '''Returns information about the peer.''' try: # get_extra_info can throw even if self.transport is not None return self.transport.get_extra_info('peername') except Exception: return None def abort(self): '''Cut the connection abruptly.''' self.transport.abort() def connection_made(self, transport): '''Handle an incoming client connection.''' transport.set_write_buffer_limits(high=self.write_buffer_high) self.transport = transport super().connection_made() def connection_lost(self, exc): '''Trigger timeouts of all pending requests.''' self.timeout_session() def is_closing(self): '''True if the underlying transport is closing.''' return self.transport and self.transport.is_closing() def close_connection(self): '''Close the connection.''' if self.transport: self.transport.close() def send_bytes(self, binary): '''Send JSON text over the transport.''' self.transport.writelines((binary, b'\n')) def peer_addr(self, anon=True): '''Return the peer address and port.''' peer_info = self.peer_info() if not peer_info: return 'unknown' if anon: return 'xx.xx.xx.xx:xx' if ':' in peer_info[0]: return '[{}]:{}'.format(peer_info[0], peer_info[1]) else: return '{}:{}'.format(peer_info[0], peer_info[1])