Add search to open cti

Packs/OpenCTI/Integrations/OpenCTI/OpenCTI.py

@@ -93,8 +93,7 @@ def reset_last_run():
 
 
 def get_indicators(client: OpenCTIApiClient, indicator_types: List[str], score: List[str] = None,
-                   limit: Optional[int] = 500,
-                   last_run_id: Optional[str] = None) -> dict:
+                   limit: Optional[int] = 500, last_run_id: Optional[str] = None, search: str = "") -> dict:
     """ Retrieving indicators from the API
 
     Args:
@@ -103,6 +102,7 @@ def get_indicators(client: OpenCTIApiClient, indicator_types: List[str], score:
         indicator_types: List of indicators types to return.
         last_run_id: The last id from the previous call to use pagination.
         limit: the max indicators to fetch
+        search: The indicator's value to filter by.
 
     Returns:
         indicators: dict of indicators
@@ -119,7 +119,8 @@ def get_indicators(client: OpenCTIApiClient, indicator_types: List[str], score:
         })
 
     indicators = client.stix_cyber_observable.list(after=last_run_id, first=limit,
-                                                   withPagination=True, filters=filters)
+                                                   withPagination=True, filters=filters,
+                                                   search=search)
     return indicators
 
 
@@ -138,6 +139,7 @@ def get_indicators_command(client: OpenCTIApiClient, args: dict) -> CommandResul
     limit = arg_to_number(args.get('limit', 50))
     start = arg_to_number(args.get('score_start', 1))
     end = arg_to_number(args.get('score_end', 100)) + 1  # type:ignore
+    search = args.get("search", "")
     score = None
     if start or end:
         score = [str(i) for i in range(start, end)]  # type:ignore
@@ -147,7 +149,8 @@ def get_indicators_command(client: OpenCTIApiClient, args: dict) -> CommandResul
         indicator_types=indicator_types,
         limit=limit,
         last_run_id=last_run_id,
-        score=score
+        score=score,
+        search=search
     )
 
     last_run = raw_response.get('pagination', {}).get('endCursor')  # type: ignore

Packs/OpenCTI/Integrations/OpenCTI/OpenCTI.yml

@@ -78,6 +78,12 @@ script:
       name: last_run_id
       required: false
       secret: false
+    - default: false
+      description: The indicator's value to filter by, can be partial value.
+      isArray: false
+      name: search
+      required: false
+      secret: false
     deprecated: false
     description: Gets indicators from OpenCTI.
     execution: false
@@ -450,7 +456,7 @@ script:
     - contextPath: OpenCTI.MarkingDefinitions.markingsLastRun
       description: The last ID of the previous fetch to use for pagination.
       type: String
-  dockerimage: demisto/opencti-v4:1.0.0.43083
+  dockerimage: demisto/opencti-v4:1.0.0.61509
   isfetch: false
   longRunning: false
   longRunningPort: false

Packs/OpenCTI/Integrations/OpenCTI/OpenCTI_test.py

@@ -33,6 +33,27 @@ def test_get_indicators(mocker):
     assert len(indicators) == 2
 
 
+@pytest.mark.parametrize(
+    'response_mock, value, expected_length, expected_value', [
+        ([{"created_at": "2022-10-24T18:16:52.678Z", "entity_type": "IPv4-Addr", "id": "id", "observable_value": "8.8.8.8",
+           "spec_version": "2.1", "standard_id": "standard_id", "updated_at": "2022-10-24T18:16:52.678Z", "value": "8.8.8.8",
+           "x_opencti_score": 50}], "8.8.8.8", 1, "8.8.8.8")])
+def test_get_indicators_value_argument(mocker, response_mock, value, expected_length, expected_value):
+    """Tests get_indicators function
+    Given
+        A value to filter by
+    When
+        - calling get_indicators
+    Then
+        - Ensure that only the result with the same given value is returned.
+    """
+    client = Client
+    mocker.patch.object(client.stix_cyber_observable, 'list', return_value=response_mock)
+    indicators = get_indicators(client, ["ALL"], search=value)
+    assert len(indicators) == expected_length
+    indicators[0].get('value') == expected_value
+
+
 def test_get_indicators_command(mocker):
     """Tests get_indicators_command function
     Given
@@ -214,7 +235,7 @@ def test_organization_list_command(mocker):
     results: CommandResults = organization_list_command(client, {})
     assert "Organizations" in results.readable_output
     assert [{'id': '1', 'name': 'test organization'}] == \
-           results.outputs.get('OpenCTI.Organizations.OrganizationsList(val.id === obj.id)')
+        results.outputs.get('OpenCTI.Organizations.OrganizationsList(val.id === obj.id)')
 
 
 def test_organization_create_command(mocker):
@@ -314,4 +335,4 @@ def test_marking_list_command(mocker):
     results: CommandResults = marking_list_command(client, {})
     assert "Markings" in results.readable_output
     assert [{'id': '1', 'value': 'TLP:RED'}] \
-           == results.outputs.get('OpenCTI.MarkingDefinitions.MarkingDefinitionsList(val.id === obj.id)')
+        == results.outputs.get('OpenCTI.MarkingDefinitions.MarkingDefinitionsList(val.id === obj.id)')

Packs/OpenCTI/ReleaseNotes/1_0_7.md

@@ -0,0 +1,7 @@
+
+#### Integrations
+
+##### OpenCTI
+
+- Updated the Docker image to: *demisto/opencti-v4:1.0.0.61509*.
+- Added the **search** argument to the **opencti-get-indicators** command.

Packs/OpenCTI/pack_metadata.json

@@ -2,7 +2,7 @@
     "name": "OpenCTI",
     "description": "Manages indicators from OpenCTI.",
     "support": "xsoar",
-    "currentVersion": "1.0.6",
+    "currentVersion": "1.0.7",
     "author": "Cortex XSOAR",
     "url": "https://www.paloaltonetworks.com/cortex",
     "email": "",