Merge pull request #118 from redcanaryco/101-requesting-automatic-ter…

…mination-of-search-after-x-time-or-y-results

101 requesting automatic termination of search after x time or y results

products/cortex_xdr.py

@@ -55,14 +55,16 @@ class CortexXDR(Product):
     _session: requests.Session
     _queries: dict[Tag, list[Query]]
     _last_request: float
+    _limit: int = 1000 # Max is 1000 results otherwise have to get the results via stream
 
     def __init__(self, profile: str, creds_file: str, **kwargs):
         if not os.path.isfile(creds_file):
             raise ValueError(f'Credential file {creds_file} does not exist')
 
         self.creds_file = creds_file
         self._queries = dict()
-
+        if self._limit >= int(kwargs.get('limit',0)) > 0:
+            self._limit = int(kwargs['limit'])
         self._last_request = 0.0
 
         super().__init__(self.product, profile, **kwargs)
@@ -227,13 +229,12 @@ def nested_process_search(self, tag: Tag, criteria: dict, base_query: dict) -> N
         except KeyboardInterrupt:
             self._echo("Caught CTRL-C. Returning what we have...")
 
-    def _get_xql_results(self, query_id: str, limit: int = 1000) -> Tuple[dict, int]:
-        actual_limit = limit if limit < 1000 else 1000  # Max is 1000 results otherwise have to get the results via stream
+    def _get_xql_results(self, query_id: str) -> Tuple[dict, int]:
         params = {
             'request_data': {
                 'query_id': query_id,
                 'pending_flag': True,
-                'limit': actual_limit,
+                'limit': self._limit,
                 'format': 'json'
             }
         }

products/microsoft_defender_for_endpoints.py

@@ -42,13 +42,17 @@ class DefenderForEndpoints(Product):
     product: str = 'dfe'
     creds_file: str  # path to credential configuration file
     _token: str  # AAD access token
+    _limit: int = -1
 
     def __init__(self, profile: str, creds_file: str, **kwargs):
         if not os.path.isfile(creds_file):
             raise ValueError(f'Credential file {creds_file} does not exist')
 
         self.creds_file = creds_file
 
+        if 100000 >= int(kwargs.get('limit', -1)) > self._limit:
+            self._limit = int(kwargs['limit'])
+
         super().__init__(self.product, profile, **kwargs)
 
     def _authenticate(self) -> None:
@@ -139,6 +143,9 @@ def process_search(self, tag: Tag, base_query: dict, query: str) -> None:
 
         query += f" {self.build_query(base_query)}" if base_query != {} else ''
 
+        if self._limit > 0 and 'limit' not in query:
+            query += f" | limit {str(self._limit)}"
+
         self.log.debug(f'Query: {query}')
         full_query = {'Query': query}
 
@@ -158,6 +165,7 @@ def nested_process_search(self, tag: Tag, criteria: dict, base_query: dict) -> N
                     else:
                         query_entry = terms
                         query_entry += f" {query_base}" if query_base != '' else ''
+
                         self.process_search(tag, {}, query_entry)
                 else:
                     all_terms = ', '.join(f"'{term}'" for term in terms)

products/sentinel_one.py

@@ -76,6 +76,7 @@ class SentinelOne(Product):
     """
     product: str = 's1'
     creds_file: str  # path to credential configuration file
+    _limit: int = 1000 # Default limit set to PowerQuery's default of 1000.
     _token: str  # AAD access token
     _url: str  # URL of SentinelOne console
     _site_id: Optional[str]  # Site ID for SentinelOne
@@ -98,6 +99,16 @@ def __init__(self, profile: str, creds_file: str, account_id: Optional[list[str]
         self._query_base = None
         self._pq = pq
 
+        # If no conditions match, the default limit will be set to PowerQuery's default of 1000.
+        if self._pq and self._limit >= int(kwargs.get('limit',0)) > 0:
+            self._limit = int(kwargs['limit'])
+
+        elif not self._pq and 20000 > int(kwargs.get('limit',0)) > 0:
+                self._limit = int(kwargs['limit'])
+
+        elif not self._pq: 
+            self._limit = 20000
+
         self._last_request = 0.0
 
         # Save these values to `self` for reference in _authenticate()
@@ -338,7 +349,7 @@ def build_query(self, filters: dict) -> Tuple[str, datetime, datetime]:
         return query_base, from_date, to_date
 
     def _get_all_paginated_data(self, url: str, params: Optional[dict] = None, headers: Optional[dict] = None,
-                                key: str = 'data', after_request: Optional[Callable] = None, limit: int = 1000,
+                                key: str = 'data', after_request: Optional[Callable] = None,
                                 no_progress: bool = True, progress_desc: str = 'Retrieving data',
                                 add_default_params: bool = True) -> list[dict]:
         """
@@ -371,7 +382,7 @@ def _get_all_paginated_data(self, url: str, params: Optional[dict] = None, heade
         if add_default_params:
             params.update(self._get_default_body())
 
-        params['limit'] = limit
+        params['limit'] = self._limit
 
         if headers is None:
             headers = dict()
@@ -603,7 +614,7 @@ def _run_query(self, merged_query: str, start_date: datetime, end_date: datetime
             params.update({
                 "fromDate": datetime_to_epoch_millis(start_date),
                 "toDate": datetime_to_epoch_millis(end_date),
-                "limit": 20000,
+                "limit": self._limit,
                 "query": merged_query
             })
 

products/vmware_cb_enterprise_edr.py

@@ -38,10 +38,12 @@ def _convert_relative_time(relative_time) -> str:
 class CbEnterpriseEdr(Product):
     product: str = 'cbc'
     _conn: CBCloudAPI  # CB Cloud API
+    _limit: int = -1
 
     def __init__(self, profile: str, **kwargs):
         self._device_group = kwargs['device_group'] if 'device_group' in kwargs else None
         self._device_policy = kwargs['device_policy'] if 'device_group' in kwargs else None
+        self._limit = int(kwargs['limit']) if 'limit' in kwargs else self._limit
 
         super().__init__(self.product, profile, **kwargs)
 
@@ -118,6 +120,9 @@ def perform_query(self, tag: Tag, base_query: dict, query: str) -> set[Result]:
                 result = Result(hostname, user, proc_name, cmdline, (ts, proc_guid,))
 
                 results.add(result)
+                if self._limit > 0 and len(results)+1 > self._limit:
+                    break
+
         except cbc_sdk.errors.ApiError as e:
             self._echo(f'CbC SDK Error (see log for details): {e}', logging.ERROR)
             self.log.exception(e)

products/vmware_cb_response.py

@@ -9,9 +9,11 @@
 class CbResponse(Product):
     product: str = 'cbr'
     _conn: CbEnterpriseResponseAPI  # CB Response API
+    _limit: int = -1
 
     def __init__(self, profile: str, **kwargs):
         self._sensor_group = kwargs['sensor_group'] if 'sensor_group' in kwargs else None
+        self._limit = int(kwargs['limit']) if 'limit' in kwargs else self._limit
 
         super().__init__(self.product, profile, **kwargs)
 
@@ -58,6 +60,10 @@ def process_search(self, tag: Tag, base_query: dict, query: str) -> None:
                 result = Result(proc.hostname.lower(), proc.username.lower(), proc.path, proc.cmdline,
                                 (proc.start, proc.id))
                 results.add(result)
+
+                if self._limit > 0 and len(results)+1 > self._limit:
+                        break
+
         except KeyboardInterrupt:
             self._echo("Caught CTRL-C. Returning what we have . . .")
 
@@ -89,6 +95,9 @@ def nested_process_search(self, tag: Tag, criteria: dict, base_query: dict) -> N
                     result = Result(proc.hostname.lower(), proc.username.lower(), proc.path, proc.cmdline,
                                     (proc.start,))
                     results.add(result)
+                    if self._limit > 0 and len(results)+1 > self._limit:
+                        break
+
         except Exception as e:
             self._echo(f'Error (see log for details): {e}', logging.ERROR)
             self.log.exception(e)

surveyor.py

@@ -85,6 +85,7 @@ class ExecutionOptions:
     days: Optional[int]
     minutes: Optional[int]
     username: Optional[str]
+    limit: Optional[int]
     ioc_file: Optional[str]
     ioc_type: Optional[str]
     query: Optional[str]
@@ -106,6 +107,17 @@ class ExecutionOptions:
 @click.option("--profile", help="The credentials profile to use.", type=click.STRING)
 @click.option("--days", help="Number of days to search.", type=click.INT)
 @click.option("--minutes", help="Number of minutes to search.", type=click.INT)
+@click.option("--limit",help="""
+              Number of results to return. Cortex XDR: Default: 1000, Max: Default
+              Microsoft Defender for Endpoint: Default/Max: 100000
+              SentinelOne (PowerQuery): Default/Max: 1000
+              SentinelOne (Deep Visibility): Default/Max: 20000
+              VMware Carbon Black EDR: Default/Max: None
+              VMware Carbon Black Cloud Enterprise EDR: Default/Max: None
+              
+              Note: Exceeding the maximum limits will automatically set the limit to its maximum value, where applicable.
+              """
+              , type=click.INT)
 @click.option("--hostname", help="Target specific host by name.", type=click.STRING)
 @click.option("--username", help="Target specific username.")
 # different ways you can survey the EDR
@@ -127,14 +139,14 @@ class ExecutionOptions:
 @click.option("--log-dir", 'log_dir', help="Specify the logging directory.", type=click.STRING, default='logs')
 @click.pass_context
 def cli(ctx, prefix: Optional[str], hostname: Optional[str], profile: str, days: Optional[int], minutes: Optional[int],
-        username: Optional[str],
+        username: Optional[str], limit: Optional[int],
         ioc_file: Optional[str], ioc_type: Optional[str], query: Optional[str], output: Optional[str],
         def_dir: Optional[str], def_file: Optional[str], no_file: bool, no_progress: bool,
         sigma_rule: Optional[str], sigma_dir: Optional[str],
         log_dir: str) -> None:
 
     ctx.ensure_object(dict)
-    ctx.obj = ExecutionOptions(prefix, hostname, profile, days, minutes, username, ioc_file, ioc_type, query, output,
+    ctx.obj = ExecutionOptions(prefix, hostname, profile, days, minutes, username, limit, ioc_file, ioc_type, query, output,
                                def_dir, def_file, sigma_rule, sigma_dir, no_file, no_progress, log_dir, dict())
 
     if ctx.invoked_subcommand is None:
@@ -265,6 +277,10 @@ def survey(ctx, product_str: str = 'cbr') -> None:
     if len(opt.product_args) > 0:
         kwargs.update(opt.product_args)
 
+    if opt.limit:
+        kwargs['limit'] = str(opt.limit)
+
+
     kwargs['tqdm_echo'] = str(not opt.no_progress)
 
     # instantiate a product class instance based on the product string

tests/test_microsoft_defender_for_endpoints.py

@@ -8,6 +8,31 @@
 from products.microsoft_defender_for_endpoints import DefenderForEndpoints
 from common import Tag
 
+def test_init_lower_limit_option(tmpdir, mocker):
+    mocker.patch.object(DefenderForEndpoints, '_authenticate')
+    cred_file_path = tmpdir.mkdir('test_dir').join('test_creds.ini')
+    cred_file_path.write("asdfasdfasdf")
+    dfe_product = DefenderForEndpoints(profile='default',creds_file=cred_file_path, limit=-2)
+    assert dfe_product._limit == -1
+
+
+def test_init_upper_limit_option(tmpdir, mocker):
+    mocker.patch.object(DefenderForEndpoints, '_authenticate')
+    cred_file_path = tmpdir.mkdir('test_dir').join('test_creds.ini')
+    cred_file_path.write("asdfasdfasdf")
+    dfe_product = DefenderForEndpoints(profile='default',creds_file=cred_file_path, limit=100001)
+    assert dfe_product._limit == -1
+
+
+def test_init_lower_limit_option(tmpdir, mocker):
+    mocker.patch.object(DefenderForEndpoints, '_authenticate')
+    cred_file_path = tmpdir.mkdir('test_dir').join('test_creds.ini')
+    cred_file_path.write("asdfasdfasdf")
+    dfe_product = DefenderForEndpoints(profile='default',creds_file=cred_file_path, limit=10)
+    assert dfe_product._limit == 10
+
+
+
 @pytest.fixture
 def dfe_product():
     with patch.object(DefenderForEndpoints, "__init__", lambda x, y: None):
@@ -40,15 +65,29 @@ def test_build_query_with_unsupported_field(dfe_product: DefenderForEndpoints, m
 
     assert dfe_product.build_query(filters) == ''
 
+def test_process_search_limit_option(dfe_product: DefenderForEndpoints, mocker):
+    query = 'DeviceFileEvents | where FileName = "foo bar"'
+    full_query = 'DeviceFileEvents | where FileName = "foo bar" | limit 5'
+    dfe_product._limit = 5
+
+    mocked_post_advanced_query = mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._post_advanced_query')
+    mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._add_results')
+    mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._get_default_header', return_value=None)
+
+    dfe_product.log = logging.getLogger('pytest_surveyor')
+    dfe_product._token = 'test_token_value'
+    dfe_product.process_search(Tag('test123'), {}, query)
+    mocked_post_advanced_query.assert_called_once_with(data={'Query': full_query}, headers=None)
+
 def test_process_search(dfe_product : DefenderForEndpoints, mocker):
     """
     Verify process_search() does not alter a given query
     """
     query = 'DeviceFileEvents | where FileName="foo bar"'
 
     mocked_post_advanced_query = mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._post_advanced_query')
-    mocked_add_results = mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._add_results')
-    mocked_get_default_headers = mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._get_default_header', return_value=None)
+    mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._add_results')
+    mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._get_default_header', return_value=None)
 
     dfe_product.log = logging.getLogger('pytest_surveyor')
     dfe_product._token = 'test_token_value'
@@ -87,6 +126,20 @@ def test_nested_process_search(dfe_product : DefenderForEndpoints, mocker):
         ]
     )
 
+def test_nested_process_search_limit_option(dfe_product: DefenderForEndpoints, mocker):
+    query = 'DeviceImageLoadEvents | where FileName = "foo bar"'
+    full_query = 'DeviceImageLoadEvents | where FileName = "foo bar" | limit 5'
+    dfe_product._limit = 5
+
+    mocked_post_advanced_query = mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._post_advanced_query')
+    mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._add_results')
+    mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._get_default_header', return_value=None)
+
+    dfe_product.log = logging.getLogger('pytest_surveyor')
+    dfe_product._token = 'test_token_value'
+    dfe_product.nested_process_search(Tag('test123'), {'query': query}, {})
+    mocked_post_advanced_query.assert_called_once_with(data={'Query': full_query}, headers=None)
+
 def test_nested_process_search_unsupported_field(dfe_product : DefenderForEndpoints, mocker):
     """
     Verify nested_process_search() gracefully handles an unsupported field in a definition file
@@ -113,8 +166,8 @@ def test_process_search_build_query(dfe_product : DefenderForEndpoints, mocker):
     }
 
     mocked_post_advanced_query = mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._post_advanced_query')
-    mocked_add_results = mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._add_results')
-    mocked_get_default_headers = mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._get_default_header', return_value=None)
+    mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._add_results')
+    mocker.patch('products.microsoft_defender_for_endpoints.DefenderForEndpoints._get_default_header', return_value=None)
 
     dfe_product.log = logging.getLogger('pytest_surveyor')
     dfe_product._token = 'test_token_value'

tests/test_sentinel_one.py

@@ -9,6 +9,50 @@
 from products.sentinel_one import SentinelOne, Query
 from common import Tag
 
+def test_init_dv_lower_limit_option(tmpdir, mocker):
+    mocker.patch.object(SentinelOne, '_authenticate')
+    cred_file_path = tmpdir.mkdir('test_dir').join('test_creds.ini')
+    cred_file_path.write("asdfasdfasdf")
+    s1_product = SentinelOne(profile='default',creds_file=cred_file_path, account_id=None, site_id=None, account_name=None, pq=False, limit = -1)
+    assert s1_product._limit == 20000
+
+def test_init_dv_upper_limit_option(tmpdir, mocker):
+    mocker.patch.object(SentinelOne, '_authenticate')
+    cred_file_path = tmpdir.mkdir('test_dir').join('test_creds.ini')
+    cred_file_path.write("asdfasdfasdf")
+
+    s1_product = SentinelOne(profile='default',creds_file=cred_file_path, account_id=None, site_id=None, account_name=None, pq=False, limit = 30000)
+    assert s1_product._limit == 20000
+
+def test_init_dv_limit_option(tmpdir, mocker):
+    mocker.patch.object(SentinelOne, '_authenticate')
+    cred_file_path = tmpdir.mkdir('test_dir').join('test_creds.ini')
+    cred_file_path.write("asdfasdfasdf")
+    s1_product = SentinelOne(profile='default',creds_file=cred_file_path, account_id=None, site_id=None, account_name=None, pq=False, limit = 5)
+    assert s1_product._limit == 5
+
+def test_init_pq_lower_limit_option(tmpdir, mocker):
+    mocker.patch.object(SentinelOne, '_authenticate')
+    cred_file_path = tmpdir.mkdir('test_dir').join('test_creds.ini')
+    cred_file_path.write("asdfasdfasdf")
+
+    s1_product = SentinelOne(profile='default',creds_file=cred_file_path, account_id=None, site_id=None, account_name=None, pq=True, limit = -1)
+    assert s1_product._limit == 1000
+
+def test_init_pq_upper_limit_option(tmpdir, mocker):
+    mocker.patch.object(SentinelOne, '_authenticate')
+    cred_file_path = tmpdir.mkdir('test_dir').join('test_creds.ini')
+    cred_file_path.write("asdfasdfasdf")
+    s1_product = SentinelOne(profile='default',creds_file=cred_file_path, account_id=None, site_id=None, account_name=None, pq=True, limit = 30000)
+    assert s1_product._limit == 1000
+
+def test_init_pq_limit_option(tmpdir, mocker):
+    mocker.patch.object(SentinelOne, '_authenticate')
+    cred_file_path = tmpdir.mkdir('test_dir').join('test_creds.ini')
+    cred_file_path.write("asdfasdfasdf")
+    s1_product = SentinelOne(profile='default',creds_file=cred_file_path, account_id=None, site_id=None, account_name=None, pq=True, limit = 6)
+    assert s1_product._limit == 6
+
 @pytest.fixture
 def s1_product():
     with patch.object(SentinelOne, "__init__", lambda x, y: None):

tests/test_surveyor.py

@@ -215,7 +215,7 @@ def test_ioc_file_with_base_query(runner, mocker):
 
 
 def test_no_argument_provided(runner):
-    arguments = ["--deffile", "--profile", "--prefix", "--output", "--defdir", "--iocfile", "--ioctype", "--query", "--hostname", "--days", "--minutes", "--username"]
+    arguments = ["--deffile", "--profile", "--prefix", "--output", "--defdir", "--iocfile", "--ioctype", "--query", "--hostname", "--days", "--minutes", "--username", "--limit"]
 
     for arg in arguments:
         result = runner.invoke(cli, [arg])

tests/test_vmware_cb_enterprise_edr.py

@@ -4,11 +4,17 @@
 import logging
 import json
 from datetime import datetime, timedelta
+import random
 from unittest.mock import patch
 sys.path.append(os.getcwd())
 from products.vmware_cb_enterprise_edr import CbEnterpriseEdr
 from common import Tag
 
+class MockProcResult():
+    def get_details(self):
+        random_num = random.randint(1, 1000)
+        return {'device_name': f'workstation{random_num}', 'process_username':[f'username{random_num}'], 'process_name':f'proc{random_num}', 'process_cmdline':[f'cmdline{random_num}'], 'device_timestamp': f'ts{random_num}', 'process_guid': f'guid{random_num}'}
+
 
 @pytest.fixture
 def cbc_product():
@@ -128,3 +134,34 @@ def test_nested_process_search(cbc_product : CbEnterpriseEdr, mocker):
     for program, criteria in programs.items():
         cbc_product.nested_process_search(Tag(program), criteria, {})
     cbc_product.perform_query.assert_has_calls(expected_calls, any_order=True)
+
+
+def test_perform_query_with_limit_option(cbc_product : CbEnterpriseEdr, mocker):
+    cbc_product.log = logging.getLogger('pytest_surveyor')
+    cbc_product._device_policy = None
+    cbc_product._device_group = None
+    cbc_product._limit = 2
+
+    cbc_product._conn = mocker.Mock()
+    cbc_product._conn.select = mocker.Mock()
+    cbc_product._conn.select.return_value = mocker.Mock(where = mocked_query_return)
+
+    assert len(cbc_product.perform_query(Tag('test_tag'), {}, 'process_name:pwsh.exe')) == 2
+
+def test_perform_query_without_limit_option(cbc_product : CbEnterpriseEdr, mocker):
+    cbc_product.log = logging.getLogger('pytest_surveyor')
+    cbc_product._device_policy = None
+    cbc_product._device_group = None
+
+    cbc_product._conn = mocker.Mock()
+    cbc_product._conn.select = mocker.Mock()
+    cbc_product._conn.select.return_value = mocker.Mock(where = mocked_query_return)
+
+    assert len(cbc_product.perform_query(Tag('test_tag'), {}, 'process_name:pwsh.exe')) == 3
+
+def mocked_query_return(full_query: str):
+    return [
+        MockProcResult(),
+        MockProcResult(),
+        MockProcResult()
+    ]

tests/test_vmware_cb_response.py

@@ -3,13 +3,24 @@
 import os
 import logging
 import json
+from dataclasses import dataclass
 from unittest.mock import patch
 from cbapi.response.models import Process
 sys.path.append(os.getcwd())
 from products.vmware_cb_response import CbResponse
 from common import Tag
 
 
+@dataclass(eq=True, frozen=True)
+class MockProcResult:
+    hostname: str
+    username: str
+    path: str
+    cmdline: str
+    start: str
+    id: str
+
+
 @pytest.fixture
 def cbr_product():
     with patch.object(CbResponse, "__init__", lambda x, y: None):
@@ -55,6 +66,21 @@ def test_process_search(cbr_product : CbResponse, mocker):
     cbr_product._conn.select.assert_called_once_with(Process)
     cbr_product._conn.select.return_value.where.assert_called_once_with('process_name:cmd.exe')
 
+
+def test_process_search_limit_option(cbr_product : CbResponse, mocker):
+    cbr_product.log = logging.getLogger('pytest_surveyor')
+    cbr_product._sensor_group = None
+    cbr_product._results = {}
+    cbr_product._limit = 1
+    cbr_product._conn = mocker.Mock()
+
+    cbr_product._conn.select = mocker.Mock()
+    cbr_product._conn.select.return_value = mocker.Mock(where = mocked_query_return)
+    cbr_product.process_search(Tag('test_tag'), {}, 'process_name:cmd.exe')
+
+    assert len(cbr_product._results[Tag('test_tag')]) == 1
+
+
 def test_nested_process_search(cbr_product : CbResponse, mocker):
     with open(os.path.join(os.getcwd(), 'tests', 'data', 'cbr_surveyor_testing.json')) as f:
         programs = json.load(f)
@@ -85,4 +111,24 @@ def test_nested_process_search(cbr_product : CbResponse, mocker):
 
     for program, criteria in programs.items():
         cbr_product.nested_process_search(Tag(program), criteria, {})
-    cbr_product._conn.select.return_value.where.assert_has_calls(expected_calls, any_order=True)
+    cbr_product._conn.select.return_value.where.assert_has_calls(expected_calls, any_order=True)
+
+
+def test_nested_process_search_limit_option(cbr_product : CbResponse, mocker):
+    cbr_product.log = logging.getLogger('pytest_surveyor')
+    cbr_product._sensor_group = None
+    cbr_product._results = {}
+    cbr_product._limit = 1
+    cbr_product._conn = mocker.Mock()
+
+    cbr_product._conn.select = mocker.Mock()
+    cbr_product._conn.select.return_value = mocker.Mock(where = mocked_query_return)
+    cbr_product.nested_process_search(Tag('test_tag'), {'query':'process_name:cmd.exe'}, {})
+
+    assert len(cbr_product._results[Tag('test_tag')]) == 1
+
+
+def mocked_query_return(query: str):
+    return [MockProcResult('workstation1','username1','path1','cmdline1','start1','id1'),
+            MockProcResult('workstation2','username2','path2','cmdline2','start2','id2'),
+            MockProcResult('workstation3','username3','path3','cmdline3','start3','id3')]