CTM360_HackerView
External Attack Surface Management platform, which combines automated asset discovery, issue identification / management, remediation guidelines, security ratings and third party risk management.
Network Security · CTM360
Details
| ID | CTM360_HackerView |
|---|---|
| Provider | CTM360 |
| Category | Network Security |
| From Version | 6.10.0 |
| Docker Image | demisto/python3:3.12.13.10116658 |
| Supported Modules | Agentix XSIAM |
README
External Attack Surface Management platform, which combines automated asset discovery, issue identification / management, remediation guidelines, security ratings and third party risk management.
Use Module To Use to select which HackerView data source this instance fetches:
- Light Scan — external attack surface findings from the standard HackerView light scan module.
- Deep Scan — deeper vulnerability and exposure findings from the HackerView Deep Scan module, including additional fields such as CVSS, evidence, and asset context.
This integration was integrated and tested with version 1.0.0 of CTM360_HackerView.
Configure CTM360 HackerView in Cortex
| Parameter | Description | Required |
|---|---|---|
| Incident Mirroring Direction | Choose the direction to mirror the incident: Incoming (from HackerView to Cortex XSOAR), Outgoing (from Cortex XSOAR to HackerView), or Incoming and Outgoing (from/to Cortex XSOAR and HackerView). | False |
| Module To Use | The module to use: Light Scan or Deep Scan. | False |
| First fetch (<number> <time unit>, e.g., 12 hours) | The time the incidents should be fetched starting from. | False |
| API Key | The CTM360 HackerView API Key to use for fetching data. | True |
| Maximum Number of Incidents per Fetch | The maximum number of incidents to fetch per run. Maximum is 200. | True |
| Fetch incidents | False | |
| Trust any certificate (not secure) | False | |
| Use system proxy settings | False | |
| Incident type | False |
Commands
You can execute these commands from the CLI, as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.
ctm360-hv-incident-list
Get the list of incidents from HV.
Base Command
ctm360-hv-incident-list
Input
| Argument Name | Description | Required |
|---|---|---|
| dateFrom | Select “From” date to fetch incidents starting from it. | Optional |
| dateTo | Select “To” date to fetch incidents up to it. | Optional |
| maxHits | Set number of results to fetch. | Optional |
| order | Set the order of the results. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| HackerView.IncidentList | unknown | List of all HV incidents. |
ctm360-hv-incident-status-change
Change status of a HV incident and optionally add a comment.
Base Command
ctm360-hv-incident-status-change
Input
| Argument Name | Description | Required |
|---|---|---|
| ticketId | “ID” of the incident to change status. | Required |
| ticketStatus | New “Status” of incident. | Required |
| comment | “Comment” to accompany the status change (Optional). | Optional |
Context Output
There is no context output for this command.
get-mapping-fields
Returns the list of fields for an incident type.
Base Command
get-mapping-fields
Input
| Argument Name | Description | Required |
| — | — | — |
Context Output
There is no context output for this command.
ctm360-hv-incident-details
Fetch details of a single incident from the HackerView platform.
Base Command
ctm360-hv-incident-details
Input
| Argument Name | Description | Required |
|---|---|---|
| ticketId | “Ticket ID” of the incident to fetch. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| HackerView.RemoteIncident.id | unknown | The symbolic incident ID. |
| HackerView.RemoteIncident.timestamp | unknown | The database timestamp. |
| HackerView.RemoteIncident.confidence | unknown | The confidence of the report. |
| HackerView.RemoteIncident.cve_id | unknown | The associated CVE identifier(s). |
| HackerView.RemoteIncident.cwe | unknown | The list of associated CWEs. |
| HackerView.RemoteIncident.issue_category | unknown | The category of the incident. |
| HackerView.RemoteIncident.issue_name | unknown | The name of the incident. |
| HackerView.RemoteIncident.potential_attack_type | unknown | The potential attack type that can make use of the incident. |
| HackerView.RemoteIncident.potential_impact | unknown | The potential impact of the incident. |
| HackerView.RemoteIncident.status | unknown | The active status of the incident. |
| HackerView.RemoteIncident.progress_status | unknown | The progress of incident response. |
| HackerView.RemoteIncident.severity | unknown | The severity of the incident. |
| HackerView.RemoteIncident.resolved_ip | unknown | The IP resolved on the affected asset. |
| HackerView.RemoteIncident.first_seen | unknown | The incident creation date. |
| HackerView.RemoteIncident.last_seen | unknown | The last discovery date for the incident. |
| HackerView.RemoteIncident.last_updated | unknown | The last update date for the incident. |
| HackerView.RemoteIncident.environments | unknown | The environments associated with the incident. |
| HackerView.RemoteIncident.ticket_id | unknown | The ticket ID. |
| HackerView.RemoteIncident.technologies | unknown | The technologies on the affected asset. |
| HackerView.RemoteIncident.domain | unknown | The domain of the affected asset. |
| HackerView.RemoteIncident.host | unknown | The host of the affected asset. |
| HackerView.RemoteIncident.asset_type | unknown | The affected asset type. |
| HackerView.RemoteIncident.asset | unknown | The affected asset. |
| HackerView.RemoteIncident.brand | unknown | The organization brand the incident belongs to. |
| HackerView.RemoteIncident.ip | unknown | The IP address associated with the finding. |
| HackerView.RemoteIncident.port | unknown | The network port associated with the finding. |
| HackerView.RemoteIncident.uri | unknown | The URI path or resource related to the finding. |
| HackerView.RemoteIncident.url | unknown | The base URL of the affected asset or service. |
| HackerView.RemoteIncident.issue_type | unknown | The type or classification of the issue. |
| HackerView.RemoteIncident.issue_description | unknown | The detailed description of the issue. |
| HackerView.RemoteIncident.cpe | unknown | The Common Platform Enumeration (CPE) identifier if applicable. |
| HackerView.RemoteIncident.cvss_metrics | unknown | The CVSS vector or metric string for the vulnerability. |
| HackerView.RemoteIncident.cvss_score | unknown | The CVSS base score for the vulnerability. |
| HackerView.RemoteIncident.epss_score | unknown | The Exploit Prediction Scoring System (EPSS) score. |
| HackerView.RemoteIncident.known_exploited | unknown | The indicator of whether the vulnerability is known to be exploited. |
| HackerView.RemoteIncident.hackerview_link | unknown | The link to the issue in the HackerView platform. |
| HackerView.RemoteIncident.evidence | unknown | The request/response evidence for the finding (e.g. request, response, curl_command). |
get-remote-data
Gets remote data from a remote incident. This method does not update the current incident, and should be used for debugging purposes.
Base Command
get-remote-data
Input
| Argument Name | Description | Required |
|---|---|---|
| id | The incident ID. | Required |
| lastUpdate | Retrieves entries that were created after lastUpdate. | Required |
Context Output
There is no context output for this command.
get-modified-remote-data
Gets the list of incidents that were modified since the last update time. Note that this method is here for debugging purposes. The get-modified-remote-data command is used as part of a Mirroring feature, which is available in Cortex XSOAR from version 6.1.
Base Command
get-modified-remote-data
Input
| Argument Name | Description | Required |
|---|---|---|
| lastUpdate | A date string in local time representing the last time the incident was updated. The incident is only returned if it was modified after the last update time. | Required |
Context Output
There is no context output for this command.
update-remote-system
Updates the remote system with local changes.
Base Command
update-remote-system
Input
| Argument Name | Description | Required |
|---|---|---|
| remoteId | Remote ID of incident to update in the remote system. | Required |
Context Output
There is no context output for this command.
Incident Mirroring
You can enable incident mirroring between Cortex XSOAR incidents and CTM360 HackerView corresponding events (available from Cortex XSOAR version 6.0.0).
To set up the mirroring:
- Enable Fetching incidents in your instance configuration.
-
In the Mirroring Direction integration parameter, select in which direction the incidents should be mirrored:
Option Description None Turns off incident mirroring. Incoming Any changes in CTM360 HackerView events (mirroring incoming fields) will be reflected in Cortex XSOAR incidents. Outgoing Any changes in Cortex XSOAR incidents will be reflected in CTM360 HackerView events (outgoing mirrored fields). Incoming And Outgoing Changes in Cortex XSOAR incidents and CTM360 HackerView events will be reflected in both directions.
Newly fetched incidents will be mirrored in the chosen direction. However, this selection does not affect existing incidents.
Important Note: To ensure the mirroring works as expected, mappers are required, both for incoming and outgoing, to map the expected fields in Cortex XSOAR and CTM360 HackerView.
Configuration parameters
mirror_direction— Incident Mirroring Directionmodule_to_use— Module To Usefirst_fetch— First fetch (<number> <time unit>, e.g., 12 hours)api_key— (required)max_fetch— Maximum Number of Incidents per Fetch (required)isFetch— Fetch incidentsinsecure— Trust any certificate (not secure)proxy— Use system proxy settingsincidentType— Incident typeincidentFetchInterval— Incidents Fetch Interval
Commands (7)
-
ctm360-hv-incident-detailsFetch details of a single incident from the HackerView platform.
-
ctm360-hv-incident-listGet the list of incidents from HV.
-
ctm360-hv-incident-status-changeChange status of a HV incident and optionally add a comment.
-
get-mapping-fieldsReturns the list of fields for an incident type.
-
get-modified-remote-dataGets the list of incidents that were modified since the last update time. Note that this method is here for debugging purposes. The get-modified-remote-data command is used as part of a Mirroring feature, which is available in Cortex XSOAR from version 6.1.
-
get-remote-dataGets remote data from a remote incident. This method does not update the current incident, and should be used for debugging purposes.
-
update-remote-systemUpdates the remote system with local changes.
import logging from datetime import datetime from unittest.mock import patch import pytest from CommonServerPython import DemistoException, IncidentSeverity, IncidentStatus from dateparser import parse from HackerView import ( ABSOLUTE_MAX_FETCH, HV_DEEPSCAN_FIELDS, HV_INCOMING_DATE_FORMAT, HV_LIGHTSCAN_FIELDS, HV_OUTGOING_DATE_FORMAT, LOGGING_PREFIX, ) """CONSTANTS""" # pylint: disable="pointless-string-statement” BASE_URL = "https://example.com:443" MODULES = [ ("lightscan", HV_LIGHTSCAN_FIELDS), ("deepscan", HV_DEEPSCAN_FIELDS), ] def load_mock_response(file_name: str) -> dict | list: """ Given: - Name of json file inside `test-data` directory. Load mock file that simulates an API response. When: - load_mock_response is called. Then: - Read the file requested into a dictionary and return it. """ import json import os base = os.path.join(os.path.dirname(os.path.abspath(__file__)), "test_data") with open(os.path.join(base, file_name), encoding="utf-8") as mock_file: return json.loads(mock_file.read()) DEEPSCAN_INCIDENT_DETAILS_EXPECTED = load_mock_response("incident_details_response_deepscan_valid.json")[0] """ MOCK CLIENT""" # pylint: disable="pointless-string-statement” @pytest.fixture() def mock_client(): """ Given: Nothing When: - mock_client is called. Then: - Return a new mock HackerView Client. """ from HackerView import Client return Client( base_url="https://example.com", verify=False, headers={"api-key": "some_mock_api_key"}, ) """ HELPER FUNCTION TESTS""" # pylint: disable="pointless-string-statement” @pytest.mark.parametrize( "hackerview_severity, expected_xsoar_severity", [ ("fyi", 0), ("info", 0.5), ("informational", 0.5), ("low", 1), ("medium", 2), ("high", 3), ("critical", 4), ], ) def test_convert_to_demisto_severity(hackerview_severity, expected_xsoar_severity): """ Given: - A string represents a HackerView severity. When: - convert_to_demisto_severity is called. Then: - Verify that the severity was correctly translated to a Cortex XSOAR severity. """ from HackerView import convert_to_demisto_severity assert convert_to_demisto_severity(hackerview_severity) == expected_xsoar_severity @pytest.mark.parametrize( "mock_log_type,mock_log_function,mock_message,mock_asserts", [ ( [logging.INFO, "INFO"], "log", "This is a log message at debug level", f"{LOGGING_PREFIX} This is a log message at debug level", ), ( [logging.INFO, "INFO"], "log", "This is a log message at info level", f"{LOGGING_PREFIX} This is a log message at info level", ), ( [logging.INFO, "INFO"], "log", "This is a log message at error level", f"{LOGGING_PREFIX} This is a log message at error level", ), ], ) def test_log(mock_log_type, mock_log_function, mock_message, mock_asserts, caplog): """ Given: - Demisto log level. - Message to log. When: - helper function log is called. Then: - Ensure logged message is as expected. """ from HackerView import log logging.getLogger().propagate = True with caplog.at_level(mock_log_type[0]): log(mock_log_type[1], mock_message) assert mock_asserts in caplog.text @pytest.mark.parametrize( "mock_input,mock_args,mock_asserts", [ ( "", {"timestamp": False, "input_format_string": "", "output_format": "", "kwargs": {}}, parse("", [""]), ), ( "", {"timestamp": False, "input_format_string": "NOT_VALID", "output_format": "", "kwargs": {}}, "", ), ( "05-12-2022 23:03:34", {"timestamp": False, "input_format_string": HV_INCOMING_DATE_FORMAT, "output_format": "", "kwargs": {}}, datetime(2022, 12, 5, 23, 3, 34), ), ( "05-12-2022 23:03:34", { "timestamp": True, "input_format_string": HV_INCOMING_DATE_FORMAT, "output_format": "", "kwargs": {"settings": {"TIMEZONE": "UTC+3", "TO_TIMEZONE": "UTC"}}, }, 1670270614000, ), ( "05-12-2022 23:03:34", { "timestamp": False, "input_format_string": HV_INCOMING_DATE_FORMAT, "output_format": "", "kwargs": {"settings": {"TIMEZONE": "UTC", "TO_TIMEZONE": "UTC"}}, }, datetime(2022, 12, 5, 23, 3, 34), ), ( "05-12-2022 23:03:34", { "timestamp": False, "input_format_string": HV_INCOMING_DATE_FORMAT, "output_format": HV_OUTGOING_DATE_FORMAT, "kwargs": {"settings": {"TIMEZONE": "UTC", "TO_TIMEZONE": "UTC"}}, }, "05-12-2022 23:03", ), ], ) def test_convert_time_string(mock_input, mock_args, mock_asserts, capfd, caplog): """ Given: - Input time string. - Input format string. - (Optional) Output format string. - (Optional) Timestamp switch When: - convert_time_string is called. Then: - Return time as either: another time format, [datetime] object, or timestamp in milliseconds """ from HackerView import convert_time_string with capfd.disabled(): result = convert_time_string( mock_input, mock_args["input_format_string"], mock_args["output_format"], mock_args["timestamp"], **mock_args["kwargs"], ) caplog.set_level(logging.INFO) assert result == mock_asserts if caplog.text: assert ( f"{LOGGING_PREFIX} An error was encountered at `convert_time_string()` \ err=ValueError('The passed date string and/or format string is not valid')" in caplog.text ) @pytest.mark.parametrize( "mock_input_file,mock_assert_file,module", [ ("fetch_incidents_response_valid.json", "incident_list_cmd_result_valid.json", "lightscan"), ("fetch_incidents_response_deepscan_valid.json", "map_and_create_deepscan_expected.json", "deepscan"), ], ) def test_map_and_create_incident(mock_input_file, mock_assert_file, module): """ Given: - A dictionary of an unmapped incident. When: - map_and_create_incident is called. Then: - Create a new incident dictionary that is in XSOAR-appropriate structure and return it. """ from HackerView import Instance, map_and_create_incident mock_fetched_incident = dict(load_mock_response(mock_input_file)[0]) expected_raw = load_mock_response(mock_assert_file) mock_assert = dict(expected_raw[0] if isinstance(expected_raw, list) else expected_raw) mock_assert.pop("rawJson", None) with patch("HackerView.INSTANCE", Instance(module=module)): result = map_and_create_incident(mock_fetched_incident) result.pop("rawJson", None) assert result == mock_assert @pytest.mark.parametrize( "input_file_name,last_run_ids,expected_num_ids,expected_num_unique", [ ("", [], 0, 0), ("fetch_incidents_response_valid.json", [], 3, 3), ("fetch_incidents_response_valid.json", ["HVI-98944790"], 3, 2), ("fetch_incidents_response_deepscan_valid.json", [], 3, 3), ("fetch_incidents_response_deepscan_valid.json", ["HVI-23813230"], 3, 2), ("fetch_incidents_response_deepscan_valid.json", ["HVI-99912344"], 3, 2), ], ) def test_deduplicate_and_create_incidents(input_file_name, last_run_ids, expected_num_ids, expected_num_unique, capfd, caplog): """ Given: - List of fetched incidents. - List of last run's calculated ids. When: - deduplicate_and_create_incidents is called. Then: - Calculate ids for the passed list of incidents. - Create a new list of XSOAR-ready incidents only for incidents not found in the last run. """ from HackerView import deduplicate_and_create_incidents with capfd.disabled(): caplog.set_level(logging.DEBUG) fetched = load_mock_response(input_file_name) if input_file_name else [] new_ids, unique_incidents = deduplicate_and_create_incidents(fetched, last_run_ids) assert len(new_ids) == expected_num_ids assert len(unique_incidents) == expected_num_unique @pytest.mark.parametrize( "mock_input,mock_assert", [ ("PascalCaseTest", "pascal_case_test"), ("camelCaseTest", "camel_case_test"), ], ) def test_to_snake_case(mock_input, mock_assert): """ Given: - String in a case other than snake case When: - to_snake_case is called Then: - Convert the input to snake case and return it """ from HackerView import to_snake_case assert to_snake_case(mock_input) == mock_assert """ COMMAND TESTS """ # pylint: disable="pointless-string-statement” @pytest.mark.parametrize( "mock_params,mock_side_effect", [ ( { "mirror_direction": "", }, DemistoException('Invalid "Mirroring Direction" Value'), ), ( { "mirror_direction": "None", "first_fetch": "wrong", }, DemistoException('Invalid "First Fetch" Value'), ), ( { "mirror_direction": "None", "max_fetch": "-1", }, DemistoException(f'Invalid "Max Fetch" Value. Should be between 1 to {ABSOLUTE_MAX_FETCH}'), ), ( { "mirror_direction": "None", "date_from": "wrong", }, DemistoException('Invalid "Date From" Value (Does not match format "%d-%m-%Y %H:%M")'), ), ( { "mirror_direction": "None", "date_to": "wrong", }, DemistoException('Invalid "Date To" Value (Does not match format "%d-%m-%Y %H:%M")'), ), ({"mirror_direction": "None", "api_key": {"password": ""}}, DemistoException('Invalid "API Key" Value')), ], ) def test_test_module(mock_params, mock_side_effect, mock_client, mocker): """ Given: - HackerView Client. - Client arguments. When: - test-module is called. Then: - The key is checked against known valid keys. - If invalid, an exception is raised with a clear message. """ from HackerView import test_module mocker.patch.object( mock_client, "test_configuration", side_effect=mock_side_effect, ) with pytest.raises(DemistoException) as e: test_module(mock_client, mock_params) assert str(e.value) == mock_side_effect.message def test_test_module_ok_when_module_to_use_missing(mock_client, mocker): """Upgraded instances without module_to_use should still test and fetch as Light Scan.""" from HackerView import test_module mock_test = mocker.patch.object(mock_client, "test_configuration", return_value=[]) result = test_module( mock_client, {"mirror_direction": "None", "api_key": {"password": "test"}}, ) assert result == "ok" assert mock_test.call_args[0][0]["module_type"] == "lightscan" @pytest.mark.parametrize("mock_module,module_fields", MODULES) def test_get_mapping_fields_command(mock_module, module_fields): """ Given: Nothing. When: - User schema in the application contains the fields 'field1' and 'field2'. - Calling function get_mapping_fields_command. Then: - Ensure a GetMappingFieldsResponse object that contains the application fields is returned. """ from HackerView import Instance, get_mapping_fields_command mock_instance = Instance(module=mock_module) with patch("HackerView.INSTANCE", new=mock_instance): mappings = get_mapping_fields_command() expected_mappings = {"HackerView Incident": {field["name"]: field["description"] for field in module_fields}} assert mappings.extract_mapping() == expected_mappings # Both modules use three incidents in `fetch_incidents_response_*_valid.json` so fetch/dedup tests stay parallel. FETCH_INCIDENTS_CASES = [ { "module": "lightscan", "last_fetch_ids": ["HVI-75466160", "HVI-11136324", "HVI-98944790"], "dup_prefix": 1, "after_dup_incidents": 2, "after_dup_ids_len": 3, "first_incident_name": "SSL Expiring in 30 days", "expect_issue_category": ["Web Communication"], "mirror_id_if_dup": "HVI-11136324", "nodup_incidents": 3, "second_incident_name": "SSL Expiring in 30 days", }, { "module": "deepscan", "last_fetch_ids": ["HVI-23813230", "HVI-56431595", "HVI-99912344"], "dup_prefix": 1, "after_dup_incidents": 2, "after_dup_ids_len": 3, "first_incident_name": "robots.txt file", "expect_issue_category": None, "mirror_id_if_dup": "HVI-56431595", "nodup_incidents": 3, "second_incident_name": "robots.txt file", "nodup_ids_match_remote": {"HVI-23813230", "HVI-56431595", "HVI-99912344"}, }, ] @pytest.mark.parametrize( "response_files_names,mock_params,fetch_case", [ ( ["fetch_incidents_response_valid.json", "fetch_incidents_response_invalid.json"], { "max_hits": "3", }, FETCH_INCIDENTS_CASES[0], ), ( ["fetch_incidents_response_deepscan_valid.json", "fetch_incidents_response_invalid.json"], { "max_hits": "3", }, FETCH_INCIDENTS_CASES[1], ), ], ) def test_fetch_incidents_command(response_files_names, mock_params, fetch_case, mock_client, mocker): """ Given: - HackerView Client - Client arguments # Case 1: - User has provided valid pagination params. - First run with no new incidents yet, so an empty list will be returned. # Case 2: - User has provided valid pagination params. - Not first run with 1 duplicate incidents, so 3 ids are calculated and 2 unique incidents are returned. # Case 3: - User has provided valid pagination params. - Not first run with no duplicate incidents, so 3 ids are calculated and 3 unique incidents are returned. # Case 4: - User has provided invalid pagination params. When: - fetch-incidents command is called. Then: - Ensure response has correct number of records # Case 1: - An empty list will be returned. # Case 2: - 3 ids are calculated and 2 unique incidents are returned. # Case 3: - 3 ids are calculated and 3 unique incidents are returned. # Case 4: - DemistoException is raised. """ from HackerView import Instance, fetch_incidents last_fetch_ids = fetch_case["last_fetch_ids"] # `main()` merges `module_type` into fetch params before calling this helper. fetch_params = {**mock_params, "module_type": fetch_case["module"]} with patch("HackerView.INSTANCE", Instance(module=fetch_case["module"])): # First run with no incidents returned mocker.patch.object(mock_client, "fetch_incidents", return_value=[]) next_run, incidents = fetch_incidents(mock_client, [], fetch_params, {}) assert next_run == {} assert incidents == [] # Not first run with 1 duplicate in the returned incidents mocker.patch.object(mock_client, "fetch_incidents", return_value=load_mock_response(response_files_names[0])) next_run, incidents = fetch_incidents( mock_client, last_fetch_ids[: fetch_case["dup_prefix"]], fetch_params, {"not_empty": ""} ) assert mock_client.fetch_incidents.call_args[0][0].get("module_type") == fetch_case["module"] assert len(incidents) == fetch_case["after_dup_incidents"] assert len(next_run.get("last_fetch_ids", [])) == fetch_case["after_dup_ids_len"] if incidents and incidents[0].get("xsoar_mirroring", {}).get("mirror_direction"): assert incidents[0].get("xsoar_mirroring", {}).get("mirror_id") == fetch_case["mirror_id_if_dup"] assert incidents[0].get("name") == fetch_case["first_incident_name"] if fetch_case["expect_issue_category"]: assert incidents[0].get("CustomFields", {}).get("issue_category") == fetch_case["expect_issue_category"] # Not first run with no duplicates in the returned incidents mocker.patch.object(mock_client, "fetch_incidents", return_value=load_mock_response(response_files_names[0])) next_run, incidents = fetch_incidents(mock_client, [], fetch_params, {"not_empty": ""}) assert len(incidents) == fetch_case["nodup_incidents"] assert next_run.get("last_fetch_ids") == last_fetch_ids if incidents and incidents[0].get("xsoar_mirroring", {}).get("mirror_direction"): assert incidents[0].get("xsoar_mirroring", {}).get("mirror_id") != "" if fetch_case["module"] == "lightscan": assert incidents[1].get("name") == fetch_case["second_incident_name"] assert incidents[1].get("CustomFields", {}).get("issue_category") == ["Web Communication"] elif fetch_case.get("nodup_ids_match_remote"): assert {inc.get("CustomFields", {}).get("id") for inc in incidents} == fetch_case["nodup_ids_match_remote"] if fetch_case.get("second_incident_name"): assert incidents[1].get("name") == fetch_case["second_incident_name"] # Run with bad params fetch_exception = DemistoException("Error received: Please contact Threat Manager Team") mocker.patch.object(mock_client, "fetch_incidents", side_effect=fetch_exception) bad_mock_params = {**fetch_params, "date_from": "abcdefg123"} with pytest.raises(DemistoException) as e: fetch_incidents(mock_client, [], bad_mock_params, {"not_empty": ""}) assert str(e.value) == "Error received: Please contact Threat Manager Team" @pytest.mark.parametrize( "response_file_name,mock_args,mock_asserts_file,module", [ ( "fetch_incidents_response_valid.json", {"maxHits": "3", "order": "asc", "dateFrom": "23-10-2023 07:00", "dateTo": "23-10-2023 23:00"}, "incident_list_cmd_result_valid.json", "lightscan", ), ( False, {"maxHits": "3", "order": "asc", "dateFrom": "23-10-2023 07:00", "dateTo": "23-10-2023 23:00"}, False, "lightscan", ), ( "fetch_incidents_response_deepscan_valid.json", {"maxHits": "3", "order": "asc", "dateFrom": "23-10-2023 07:00", "dateTo": "23-10-2023 23:00"}, "deepscan_incident_list_cmd_result_valid.json", "deepscan", ), ], ) def test_ctm360_hv_incident_list_command(response_file_name, mock_args, mock_asserts_file, module, mock_client, mocker): """ Given: - HackerView Client. - Client arguments. When: - fetch_incidents is called. Then: - Fetch the list of incidents from the remote server. """ from HackerView import Instance, ctm360_hv_incident_list_command patched_response = load_mock_response(response_file_name) if response_file_name else [] mocker.patch.object(mock_client, "fetch_incidents", return_value=patched_response) with patch("HackerView.INSTANCE", Instance(module=module)): cmd_results = ctm360_hv_incident_list_command(mock_client, mock_args) expected_results = load_mock_response(mock_asserts_file) if mock_asserts_file else [] cmd_results = cmd_results.to_context().get("Contents") called_params = mock_client.fetch_incidents.call_args[0][0] assert called_params.get("module_type") == module if cmd_results and expected_results: cmd_results = [{k: v for k, v in item.items() if k != "rawJson"} for item in cmd_results] expected_results = [{k: v for k, v in item.items() if k != "rawJson"} for item in expected_results] assert cmd_results == expected_results @pytest.mark.parametrize( "response_file_name,mock_args,mock_asserts,module", [ ( "incident_details_response_valid.json", {"ticketId": "HVI-11145070"}, { "id": "HVI-11145070", "timestamp": 1725453245000, "confidence": "confirmed", "cve_id": "CVE-2022-1292", "cwe": ["CWE-78"], "issue_category": ["Common Vulnerabilities"], "issue_id": 20221292, "issue_name": "Openssl 3.0.2 Vulnerability detected", "status": "active", "progress_status": "New", "severity": "critical", "first_seen": "04-09-2024 12:34:03", "last_seen": "26-10-2024 12:30:30", "environments": ["Web Server", "Library"], "ip": "10.161.216.126", "ticket_id": "HVI-11145070", "ip_type": "dedicated", "technologies": ["OpenSSL", "Apache"], "port": 443, "asset_type": "ip", "asset": "10.161.216.126", "brand": "REPERSEA", "last_updated": 1733839847270, }, "lightscan", ), ( False, {"ticketId": "HVI-NOTFOUND"}, {}, "lightscan", ), ( "incident_details_response_deepscan_valid.json", {"ticketId": "HVI-99912344"}, DEEPSCAN_INCIDENT_DETAILS_EXPECTED, "deepscan", ), ], ) def test_ctm360_hv_incident_details_command(response_file_name, mock_args, mock_asserts, module, mock_client, mocker): """ Given: - Ticket ID of incident. When: - ctm360-hv-incident-details command is called. Then: - Ensure result is as expected. """ from HackerView import Instance, ctm360_hv_incident_details_command patched_response = load_mock_response(response_file_name)[0] if response_file_name else {} mocker.patch.object(mock_client, "fetch_incident", return_value=patched_response) with patch("HackerView.INSTANCE", Instance(module=module)): cmd_results = ctm360_hv_incident_details_command(mock_client, mock_args) called_params = mock_client.fetch_incident.call_args[0][0] assert called_params.get("module_type") == module assert cmd_results.to_context().get("Contents") == mock_asserts @pytest.mark.parametrize( "response_file_name,mock_args,mock_asserts", [ ( "incident_status_change_response_valid.json", {"ticketId": "HVI-TEST", "ticketStatus": "investigating", "comment": "test"}, "Status updated successfully", ), ( "incident_status_change_response_invalid_id.json", {"ticketId": "HVI-TEST"}, "Issue not found in the system", ), ( "incident_status_change_response_invalid_status.json", {"ticketId": "HVI-TEST"}, "Issue status not valid. Please provide a value from " "new,investigating,in_progress,fixed,acceptable_risk,false_positive", ), ], ) def test_ctm360_hv_incident_status_change_command(response_file_name, mock_args, mock_asserts, mock_client, mocker): """ Given: - Issue ID of incident. When: - ctm360-hv-incident-change-status command is called. Then: - Ensure result is as expected. """ from HackerView import ctm360_hv_incident_status_change_command mocker.patch.object(mock_client, "change_incident_status", return_value=load_mock_response(response_file_name)) cmd_results = ctm360_hv_incident_status_change_command(mock_client, mock_args) assert cmd_results.to_context().get("HumanReadable") == mock_asserts @pytest.mark.parametrize( "mock_last_seen,mock_last_updated,mock_status", [ ("11-12-2024 11:24:47", 1733905487769, "active"), ("11-12-2024 11:24:47", 1733905487768, "active"), ("11-12-2024 11:24:45", 1733905487769, "active"), ("11-12-2024 11:24:45", 1733905487768, "active"), ("11-12-2024 11:24:47", 1733905487769, "inactive"), ("11-12-2024 11:24:47", 1733905487768, "inactive"), ("11-12-2024 11:24:45", 1733905487769, "inactive"), ("11-12-2024 11:24:45", 1733905487768, "inactive"), ], ) def test_get_remote_data(mock_last_seen, mock_last_updated, mock_status, mock_client, mocker): """ Given: - HackerView Client. - Client arguments. When: - get-remote-data command is called. Then: - Ensure result is as expected. """ from copy import deepcopy from HackerView import ( convert_time_string, get_remote_data_command, map_and_create_incident, ) mock_args = {"id": "HVI-11145070", "lastUpdate": "2024-10-26T12:34:03.172707565Z"} mock_result = load_mock_response("incident_details_response_valid.json")[0] mock_result["status"] = mock_status mock_result["last_updated"] = mock_last_updated progress_statuses = [ ("Investigating", "Your team is actively investigating and examining this particular issue."), ("In Progress", "Your team has concluded its investigation and has shifted its focus to resolution."), ( "Fixed", "Your team has implemented all necessary changes and fixes to address this issue." + " Our system will automatically label issues as resolved once they are no longer present.", ), ("False Positive", "The identified problem was incorrectly tagged and may not be relevant."), ( "Acceptable Risk", "your organization deems any risk(s) linked to a particular issue as acceptable." + " This categorization is intended for internal classification purposes.", ), ] for mock_progress in progress_statuses: mock_result["progress_status"] = mock_progress[0] mocker.patch.object(mock_client, "fetch_incident", return_value=deepcopy(mock_result)) mock_result2 = map_and_create_incident(deepcopy(mock_result)) if mock_status and isinstance(mock_result, dict) else [] if isinstance(mock_result2, dict): del mock_result2["rawJson"] result = get_remote_data_command(mock_client, mock_args) called_params = mock_client.fetch_incident.call_args[0][0] assert called_params.get("module_type") == "lightscan" entry = result.entries[0] if len(result.entries) >= 1 else [] if isinstance(result.mirrored_object, dict): del result.mirrored_object["rawJson"] if ( convert_time_string(mock_last_seen, HV_INCOMING_DATE_FORMAT, in_iso_format=True, is_utc=True, timestamp=True) == mock_last_updated ): if mock_status == "active": assert entry == {"Type": 1, "ContentsFormat": "json", "Contents": {"dbotIncidentReopen": True}} else: close_reason = f'Incident was {"resolved" if mock_progress[0].lower() == "fixed" else "closed"}.' assert entry == { "Type": 1, "ContentsFormat": "json", "Contents": {"dbotIncidentClose": True, "closeReason": close_reason}, } assert result.mirrored_object == mock_result2 elif mock_status != "inactive": assert entry == {"Type": 1, "ContentsFormat": "text", "Contents": mock_progress[1]} assert result.mirrored_object == mock_result2 else: assert result.mirrored_object == [] assert entry == [] @pytest.mark.parametrize( "mock_input_file,module", [ ("fetch_incidents_response_valid.json", "lightscan"), ("fetch_incidents_response_deepscan_valid.json", "deepscan"), ], ) def test_get_modified_remote_data(mock_input_file, module, mock_client, mocker): """ Given: - HackerView Client. - Client arguments. When: - get-modified-remote-data command is called Then: - Ensure result is as expected """ from HackerView import Instance, get_modified_remote_data_command mock_args = { "date_field": "last_updated", "order": "asc", "date_from": "1704183134", "max_hits": 50, "lastUpdate": "2024-01-02T13:30:21.172707565Z", } mock_result = load_mock_response(mock_input_file) mocker.patch.object(mock_client, "fetch_incidents", return_value=mock_result) with patch("HackerView.INSTANCE", Instance(module=module)): result = get_modified_remote_data_command(mock_client, mock_args) called_params = mock_client.fetch_incidents.call_args[0][0] assert called_params.get("module_type") == module mock_assert = [str(item["id"]) for item in mock_result] assert result.modified_incident_ids == mock_assert @pytest.mark.parametrize( "mock_response_file,mock_args,mock_log_asserts", [ ( "", { "remoteId": "1", "data": {}, "entries": [], "status": IncidentStatus.ACTIVE, "incidentChanged": False, "delta": {"not_empty": "value"}, }, f"{LOGGING_PREFIX} Incident 1 was not modified locally..", ), ( "", { "remoteId": "2", "data": {}, "entries": [], "status": IncidentStatus.ARCHIVE, "incidentChanged": True, "delta": {"not_empty": "value"}, }, f"{LOGGING_PREFIX} Modification to 2 is not configured for outgoing mirroring..", ), ( "incident_status_change_response_valid.json", { "remoteId": "3", "data": {}, "entries": [], "status": IncidentStatus.DONE, "incidentChanged": True, "delta": {"not_empty": "value"}, }, f"{LOGGING_PREFIX} Closing incident 3", ), ], ) def test_update_remote_system(mock_response_file, mock_args, mock_log_asserts, mock_client, mocker, caplog): """ Given: - HackerView Client. - Client arguments. When: - update-remote-system command is called. Then: - Ensure result is as expected. """ from HackerView import update_remote_system_command with caplog.at_level(logging.INFO): if mock_response_file: mocker.patch.object(mock_client, "change_incident_status", return_value=load_mock_response(mock_response_file)) result = update_remote_system_command(mock_client, mock_args) assert result == mock_args["remoteId"] assert mock_log_asserts in caplog.text @pytest.mark.parametrize( "module_to_use,expected", [ (None, "lightscan"), ("", "lightscan"), ("Light Scan", "lightscan"), ("Deep Scan", "deepscan"), ("legacy-unknown", "lightscan"), ], ) def test_resolve_hv_module_defaults_to_lightscan(module_to_use, expected): """Missing or blank module_to_use must keep pre-DeepScan behavior (Light Scan).""" from HackerView import resolve_hv_module assert resolve_hv_module(module_to_use) == expected @pytest.mark.parametrize( "is_xsiam_platform, expected_severity", [ (True, IncidentSeverity.MEDIUM), (False, IncidentSeverity.MEDIUM), ], ) def test_map_and_create_incident_severity_by_platform(is_xsiam_platform, expected_severity, mocker): """Incident fetch maps API severity to numeric Demisto severity on all platforms.""" from HackerView import Instance, map_and_create_incident mocker.patch("HackerView.is_xsiam", return_value=is_xsiam_platform) issue = { "issue_name": "Test Issue", "first_seen": "05-07-2024 06:28:28", "last_seen": "26-10-2024 12:15:24", "status": "active", "severity": "medium", "id": "HVI-1", "timestamp": 1720161042000, } with patch("HackerView.INSTANCE", Instance(module="lightscan")): result = map_and_create_incident(issue) assert result["severity"] == expected_severity