raise and disconnect if too many parameters are used on search

lbry/error/__init__.py

@@ -234,6 +234,13 @@ class InvalidPasswordError(WalletError):
         super().__init__("Password is invalid.")
 
 
+class TooManyClaimSearchParameters(WalletError):
+
+    def __init__(self, key, limit):
+        self.key = key
+        super().__init__(f"{key} cant be set for more than {limit} items.")
+
+
 class IncompatibleWalletServerError(WalletError):
 
     def __init__(self, server, port):

lbry/wallet/server/db/elasticsearch/search.py

@@ -10,7 +10,7 @@ from elasticsearch import AsyncElasticsearch, NotFoundError, ConnectionError
 from elasticsearch.helpers import async_streaming_bulk
 
 from lbry.crypto.base58 import Base58
-from lbry.error import ResolveCensoredError, claim_id as parse_claim_id
+from lbry.error import ResolveCensoredError, TooManyClaimSearchParameters
 from lbry.schema.result import Outputs, Censor
 from lbry.schema.tags import clean_tags
 from lbry.schema.url import URL, normalize_name
@@ -465,6 +465,8 @@ def expand_query(**kwargs):
     for key, value in kwargs.items():
         key = key.replace('claim.', '')
         many = key.endswith('__in') or isinstance(value, list)
+        if many and len(value) > 2048:
+            raise TooManyClaimSearchParameters(key, 2048)
         if many:
             key = key.replace('__in', '')
             value = list(filter(None, value))

lbry/wallet/server/session.py

@@ -15,14 +15,13 @@ from asyncio import Event, sleep
 from collections import defaultdict
 from functools import partial
 
-from binascii import hexlify
 from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
 
 from elasticsearch import ConnectionTimeout
 from prometheus_client import Counter, Info, Histogram, Gauge
 
 import lbry
-from lbry.utils import LRUCacheWithMetrics
+from lbry.error import TooManyClaimSearchParameters
 from lbry.build_info import BUILD, COMMIT_HASH, DOCKER_TAG
 from lbry.wallet.server.block_processor import LBRYBlockProcessor
 from lbry.wallet.server.db.writer import LBRYLevelDB
@@ -1028,7 +1027,11 @@ class LBRYElectrumX(SessionBase):
 
     async def claimtrie_search(self, **kwargs):
         if kwargs:
-            return await self.run_and_cache_query('search', kwargs)
+            try:
+                return await self.run_and_cache_query('search', kwargs)
+            except TooManyClaimSearchParameters as err:
+                self.loop.call_later(0, self.synchronous_close)
+                return RPCError(1, str(err))
 
     async def claimtrie_resolve(self, *urls):
         if urls:

tests/integration/blockchain/test_wallet_server_sessions.py

@@ -4,6 +4,7 @@ import lbry
 import lbry.wallet
 from lbry.error import ServerPaymentFeeAboveMaxAllowedError
 from lbry.wallet.network import ClientSession
+from lbry.wallet.rpc import RPCError
 from lbry.wallet.server.db.elasticsearch.sync import run as run_sync, make_es_index
 from lbry.wallet.server.session import LBRYElectrumX
 from lbry.testcase import IntegrationTestCase, CommandTestCase
@@ -191,7 +192,7 @@ class TestHubDiscovery(CommandTestCase):
         )
 
 
-class TestStressFlush(CommandTestCase):
+class TestStress(CommandTestCase):
     async def test_flush_over_66_thousand(self):
         history = self.conductor.spv_node.server.db.history
         history.flush_count = 66_000
@@ -199,3 +200,9 @@ class TestStressFlush(CommandTestCase):
         self.assertEqual(history.flush_count, 66_001)
         await self.generate(1)
         self.assertEqual(history.flush_count, 66_002)
+
+    async def test_thousands_claim_ids_on_search(self):
+        await self.stream_create()
+        with self.assertRaises(RPCError) as err:
+            await self.claim_search(not_channel_ids=[("%040x" % i) for i in range(8196)])
+        self.assertEqual(err.exception.message, 'not_channel_ids cant be set for more than 2048 items.')