Add subscription limits

docs/ENV-NOTES

@@ -32,8 +32,20 @@ RPC_PORT      - Listen on this port for local RPC connections, defaults to
                 8000.
 BANNER_FILE   - a path to a banner file to serve to clients.  The banner file
                 is re-read for each new client.
-DONATION_ADDRESS  - server donation address.  Defaults to none.
-ANON_LOGS - set to remove IP addresses from logs. Default: disabled
+ANON_LOGS     - set to anything non-empty to remove IP addresses from
+                logs. By default IP addresses will be logged.
+DONATION_ADDRESS - server donation address.  Defaults to none.
+
+These following environment variables are to help limit server
+resource consumption and to prevent simple DoS.  Address subscriptions
+in ElectrumX are very cheap - they consume about 100 bytes of memory
+each and are processed efficiently.  I feel the defaults are low and
+encourage you to raise them.
+
+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.
 
 If you want IRC connectivity to advertise your node:
 

docs/RELEASE-NOTES

@@ -1,3 +1,9 @@
+version 0.6.3
+-------------
+
+- new environment variables MAX_SUBS and MAX_SESSION_SUBS.  Please read
+  docs/ENV-NOTES - I encourage you to raise the default values.
+
 version 0.6.2
 -------------
 

server/env.py

@@ -46,6 +46,9 @@ class Env(LoggedClass):
         self.db_engine = self.default('DB_ENGINE', 'leveldb')
         self.debug = self.default('DEBUG', '')
         self.debug = [item.lower() for item in self.debug.split()]
+        # Subscription limits
+        self.max_subs = self.integer('MAX_SUBS', 250000)
+        self.max_session_subs = self.integer('MAX_SESSION_SUBS', 50000)
         # 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

@@ -246,7 +246,13 @@ class ServerManager(LoggedClass):
         self.servers = []
         self.irc = IRC(env)
         self.sessions = {}
+        self.max_subs = env.max_subs
+        self.subscription_count = 0
         self.futures = []  # At present just the IRC future, if any
+        self.logger.info('max subscriptions across all sessions: {:,d}'
+                         .format(self.max_subs))
+        self.logger.info('max subscriptions per session: {:,d}'
+                         .format(env.max_session_subs))
 
     async def start_server(self, kind, *args, **kw_args):
         loop = asyncio.get_event_loop()
@@ -318,9 +324,17 @@ class ServerManager(LoggedClass):
         self.sessions[session] = asyncio.ensure_future(coro)
 
     def remove_session(self, session):
+        if isinstance(session, ElectrumX):
+            self.subscription_count -= len(session.hash168s)
         future = self.sessions.pop(session)
         future.cancel()
 
+    def new_subscription(self):
+        if self.subscription_count >= self.max_subs:
+            raise JSONRPC.RPCError('server subscription limit {:,d} reached'
+                                   .format(self.max_subs))
+        self.subscription_count += 1
+
     def irc_peers(self):
         return self.irc.peers
 
@@ -330,18 +344,19 @@ class ServerManager(LoggedClass):
         total = len(self.sessions)
         return {'active': active, 'inert': total - active, 'total': total}
 
-    def address_count(self):
-        return sum(len(session.hash168s) for session in self.sessions
-                   if isinstance(session, ElectrumX))
-
     async def rpc_getinfo(self, params):
         '''The RPC 'getinfo' call.'''
+        # FIXME: remove later
+        indep_count = sum(len(session.hash168s) for session in self.sessions
+                          if isinstance(session, ElectrumX))
+        if indep_count != self.subscription_count:
+            self.logger.error('sub count {:,d} but session total {:,d}'
+                              .format(self.subscription_count, indep_count))
         return {
             'blocks': self.bp.height,
             'peers': len(self.irc.peers),
             'sessions': self.session_count(),
-            'watched': self.address_count(),
-            'cached': 0,
+            'watched': self.subscription_count,
         }
 
     async def rpc_sessions(self, params):
@@ -503,6 +518,7 @@ class ElectrumX(Session):
         self.subscribe_headers = False
         self.subscribe_height = False
         self.notified_height = None
+        self.max_subs = self.env.max_session_subs
         self.hash168s = set()
         rpcs = [
             ('blockchain',
@@ -689,8 +705,14 @@ class ElectrumX(Session):
 
     async def address_subscribe(self, params):
         hash168 = self.extract_hash168(params)
+        if len(self.hash168s) >= self.max_subs:
+            raise self.RPCError('your address subscription limit {:,d} reached'
+                                .format(self.max_subs))
+        result = await self.address_status(hash168)
+        # add_subscription can raise so call it before adding
+        self.manager.new_subscription()
         self.hash168s.add(hash168)
-        return await self.address_status(hash168)
+        return result
 
     async def block_get_chunk(self, params):
         index = self.extract_non_negative_integer(params)