Merge branch 'release-0.7.14'

RELEASE-NOTES

@@ -1,3 +1,19 @@
+version 0.7.14
+--------------
+
+Improved DoS protection:
+
+- incoming network request buffers - which hold incomplete requests
+  are limited to 150,000 bytes, which I believe is large for genuine
+  clients.  I don't foresee a need to change this so it is hard-coded.
+  If an incoming request (for example, text without a newline) exceeds
+  this limit the connection is dropped and the event logged.
+- RPC connections have high MAX_SEND and incoming buffer limits as these
+  connections are assumed to be trusted.
+- new environment variable BANDWIDTH_LIMIT.  See docs/ENV-NOTES.
+- fixes: LOG_SESSIONS of 0.7.13 wasn't being properly interpreted.
+  Tweak to rocksdb close() that should permit db reopening to work.
+
 version 0.7.13
 --------------
 

docs/ENV-NOTES

@@ -70,6 +70,19 @@ MAX_SUBS         - maximum number of address subscriptions across all
                    sessions.  Defaults to 250,000.
 MAX_SESSION_SUBS - maximum number of address subscriptions permitted to a
                    single session.  Defaults to 50,000.
+BANDWIDTH_LIMIT  - per-session periodic bandwith usage limit in bytes.
+                   Bandwidth usage over each period is totalled, and
+                   when this limit is exceeded each subsequent request
+                   is stalled by sleeping before handling it,
+                   effectively yielding processing resources to other
+                   sessions.  Each time this happens the event is
+                   logged.  The more bandwidth usage exceeds the limit
+                   the longer the next request will sleep.  Each sleep
+                   is a round number of seconds with a minimum of one.
+                   The bandwith usage counter is reset to zero at the
+                   end of each period.  Currently the period is
+                   hard-coded to be one hour.  The default limit value
+                   is 2 million bytes.
 
 If you want IRC connectivity to advertise your node:
 

electrumx_rpc.py

@@ -23,6 +23,8 @@ from server.protocol import ServerManager
 class RPCClient(JSONRPC):
 
     async def send_and_wait(self, method, params, timeout=None):
+        # Raise incoming buffer size - presumably connection is trusted
+        self.max_buffer_size = 5000000
         self.send_json_request(method, id_=method, params=params)
 
         future = asyncio.ensure_future(self.messages.get())

lib/jsonrpc.py

@@ -79,6 +79,10 @@ class JSONRPC(asyncio.Protocol, LoggedClass):
     def __init__(self):
         super().__init__()
         self.start = time.time()
+        self.bandwidth_start = self.start
+        self.bandwidth_interval = 3600
+        self.bandwidth_used = 0
+        self.bandwidth_limit = 5000000
         self.transport = None
         # Parts of an incomplete JSON line.  We buffer them until
         # getting a newline.
@@ -96,6 +100,8 @@ class JSONRPC(asyncio.Protocol, LoggedClass):
         # connection.  The request causing it is logged.  Values under
         # 1000 are treated as 1000.
         self.max_send = 0
+        # If buffered incoming data exceeds this the connection is closed
+        self.max_buffer_size = 150000
         self.anon_logs = False
 
     def peername(self, *, for_log=True):
@@ -115,6 +121,13 @@ class JSONRPC(asyncio.Protocol, LoggedClass):
         '''Handle client disconnection.'''
         pass
 
+    def using_bandwidth(self, amount):
+        now = time.time()
+        if now >= self.bandwidth_start + self.bandwidth_interval:
+            self.bandwidth_start = now
+            self.bandwidth_used = 0
+        self.bandwidth_used += amount
+
     def data_received(self, data):
         '''Handle incoming data (synchronously).
 
@@ -122,6 +135,18 @@ class JSONRPC(asyncio.Protocol, LoggedClass):
         decode_message for handling.
         '''
         self.recv_size += len(data)
+        self.using_bandwidth(len(data))
+
+        # Close abuvsive 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.logger.error('read buffer of {:,d} bytes exceeds {:,d} '
+                              'byte limit, closing {}'
+                              .format(buffer_size, self.max_buffer_size,
+                                      self.peername()))
+            self.transport.close()
+
+        # Do nothing if this connection is closing
         if self.transport.is_closing():
             return
 
@@ -200,6 +225,7 @@ class JSONRPC(asyncio.Protocol, LoggedClass):
             else:
                 self.send_count += 1
                 self.send_size += len(data)
+                self.using_bandwidth(len(data))
                 self.transport.write(data)
 
     async def handle_message(self, message):
@@ -207,6 +233,17 @@ class JSONRPC(asyncio.Protocol, LoggedClass):
 
         Handles batches according to the JSON 2.0 spec.
         '''
+        # Throttle high-bandwidth connections by delaying processing
+        # their requests.  Delay more the higher the excessive usage.
+        excess = self.bandwidth_used - self.bandwidth_limit
+        if excess > 0:
+            secs = 1 + excess // self.bandwidth_limit
+            self.logger.warning('{} has high bandwidth use of {:,d} bytes, '
+                                'sleeping {:d}s'
+                                .format(self.peername(), self.bandwidth_used,
+                                        secs))
+            await asyncio.sleep(secs)
+
         if isinstance(message, list):
             payload = await self.batch_payload(message)
         else:

server/env.py

@@ -41,7 +41,7 @@ class Env(LoggedClass):
         self.max_subscriptions = self.integer('MAX_SUBSCRIPTIONS', 10000)
         self.banner_file = self.default('BANNER_FILE', None)
         self.anon_logs = self.default('ANON_LOGS', False)
-        self.log_sessions = self.default('LOG_SESSIONS', 3600)
+        self.log_sessions = self.integer('LOG_SESSIONS', 3600)
         # The electrum client takes the empty string as unspecified
         self.donation_address = self.default('DONATION_ADDRESS', '')
         self.db_engine = self.default('DB_ENGINE', 'leveldb')
@@ -49,6 +49,7 @@ class Env(LoggedClass):
         self.max_send = self.integer('MAX_SEND', 1000000)
         self.max_subs = self.integer('MAX_SUBS', 250000)
         self.max_session_subs = self.integer('MAX_SESSION_SUBS', 50000)
+        self.bandwidth_limit = self.integer('BANDWIDTH_LIMIT', 2000000)
         # IRC
         self.report_tcp_port = self.integer('REPORT_TCP_PORT', self.tcp_port)
         self.report_ssl_port = self.integer('REPORT_SSL_PORT', self.ssl_port)

server/protocol.py

@@ -230,6 +230,8 @@ class ServerManager(util.LoggedClass):
         self.subscription_count = 0
         self.futures = []
         env.max_send = max(350000, env.max_send)
+        self.logger.info('session bandwidth limit {:,d} bytes'
+                         .format(env.bandwidth_limit))
         self.logger.info('max response size {:,d} bytes'.format(env.max_send))
         self.logger.info('max subscriptions across all sessions: {:,d}'
                          .format(self.max_subs))
@@ -471,6 +473,7 @@ class Session(JSONRPC):
         self.client = 'unknown'
         self.anon_logs = env.anon_logs
         self.max_send = env.max_send
+        self.bandwidth_limit = env.bandwidth_limit
         self.txs_sent = 0
 
     def connection_made(self, transport):
@@ -923,3 +926,4 @@ class LocalRPC(Session):
         self.handlers = {cmd: getattr(self.manager, 'rpc_{}'.format(cmd))
                          for cmd in cmds}
         self.client = 'RPC'
+        self.max_send = 5000000

server/storage.py

@@ -113,7 +113,7 @@ class RocksDB(Storage):
 
     def close(self):
         # PyRocksDB doesn't provide a close method; hopefully this is enough
-        self.db = None
+        self.db = self.get = self.put = None
         import gc
         gc.collect()