Unit42v2 Feed Deprecated

import demistomock as demisto
import pytest
from FeedUnit42v2 import (
    Client,
    DemistoException,
    create_attack_pattern_indicator,
    create_course_of_action_indicators,
    create_list_relationships,
    extract_ioc_value,
    fetch_indicators,
    get_indicators_command,
    get_ioc_type,
    get_ioc_value,
    handle_multiple_dates_in_one_field,
    parse_campaigns,
    parse_indicators,
    parse_reports_and_report_relationships,
)
from test_data.feed_data import (
    ATTACK_PATTERN_DATA,
    ATTACK_PATTERN_INDICATOR,
    CAMPAIGN_INDICATOR,
    CAMPAIGN_RESPONSE,
    COURSE_OF_ACTION_DATA,
    COURSE_OF_ACTION_INDICATORS,
    COURSE_OF_ACTION_INDICATORS_WITH_TLP,
    DUMMY_INDICATOR_WITH_RELATIONSHIP_LIST,
    FETCH_MOCK_RESPONSE,
    FETCH_RESULTS,
    ID_TO_OBJECT,
    INDICATORS_DATA,
    INDICATORS_RESULT,
    INTRUSION_SET_DATA,
    INVALID_ATTACK_PATTERN_STRUCTURE,
    MALWARE_DATA,
    RELATIONSHIP_DATA,
    RELATIONSHIP_OBJECTS,
    REPORTS_DATA,
    REPORTS_INDICATORS,
    REPORTS_INDICATORS_WITH_RELATIONSHIPS,
    STIX_ATTACK_PATTERN_INDICATOR,
    SUB_TECHNIQUE_DATA,
    SUB_TECHNIQUE_INDICATOR,
)


@pytest.mark.parametrize(
    "command, args, response, length",
    [
        (get_indicators_command, {"limit": 2, "indicators_type": "indicator"}, INDICATORS_DATA, 2),
        (get_indicators_command, {"limit": 5, "indicators_type": "indicator"}, INDICATORS_DATA, 5),
    ],
)  # noqa: E124
def test_commands(command, args, response, length, mocker):
    """
    Given
    - get_indicators_command func
    - command args
    - command raw response
    When
    - mock the Client's get_stix_objects.
    Then
    - convert the result to human readable table
    - create the context
    validate the raw_response
    """
    client = Client(api_key="1234", verify=False)
    mocker.patch.object(client, "fetch_stix_objects_from_api", return_value=response)
    command_results = command(client, args)
    indicators = command_results.raw_response
    assert len(indicators) == length


TYPE_TO_RESPONSE = {
    "indicator": INDICATORS_DATA,
    "report": REPORTS_DATA,
    "attack-pattern": ATTACK_PATTERN_DATA,
    "malware": MALWARE_DATA,
    "campaign": CAMPAIGN_RESPONSE,
    "relationship": RELATIONSHIP_DATA,
    "course-of-action": COURSE_OF_ACTION_DATA,
    "intrusion-set": INTRUSION_SET_DATA,
}

TYPE_TO_RESPONSE_WIITH_INVALID_ATTACK_PATTERN_DATA = {
    "indicator": INDICATORS_DATA,
    "report": REPORTS_DATA,
    "attack-pattern": INVALID_ATTACK_PATTERN_STRUCTURE,
    "malware": MALWARE_DATA,
    "campaign": CAMPAIGN_RESPONSE,
    "relationship": RELATIONSHIP_DATA,
    "course-of-action": COURSE_OF_ACTION_DATA,
    "intrusion-set": INTRUSION_SET_DATA,
}


TYPE_TO_RESPONSE_FETCH = {
    "indicator": INDICATORS_DATA,
    "report": REPORTS_DATA,
    "attack-pattern": ATTACK_PATTERN_DATA,
    "malware": MALWARE_DATA,
    "campaign": CAMPAIGN_RESPONSE,
    "relationship": RELATIONSHIP_DATA,
    "course-of-action": COURSE_OF_ACTION_DATA,
    "intrusion-set": INTRUSION_SET_DATA,
}


TYPE_TO_RESPONSE_FETCH = {
    "indicator": INDICATORS_DATA,
    "report": FETCH_MOCK_RESPONSE,
    "attack-pattern": ATTACK_PATTERN_DATA,
    "malware": MALWARE_DATA,
    "campaign": CAMPAIGN_RESPONSE,
    "relationship": RELATIONSHIP_DATA,
    "course-of-action": COURSE_OF_ACTION_DATA,
    "intrusion-set": INTRUSION_SET_DATA,
}


def test_fetch_indicators_command(mocker):
    """
    Given
    - fetch incidents command
    - command args
    - command raw response
    When
    - mock the Client's get_stix_objects.
    Then
    - run the fetch incidents command using the Client
    Validate the amount of indicators fetched
    Validate that the dummy indicator with the relationships list fetched
    """

    def mock_get_stix_objects(test, **kwargs):
        type_ = kwargs.get("type")
        client.objects_data[type_] = TYPE_TO_RESPONSE[type_]

    client = Client(api_key="1234", verify=False)
    mocker.patch.object(client, "fetch_stix_objects_from_api", side_effect=mock_get_stix_objects)

    indicators = fetch_indicators(client, create_relationships=True)
    assert len(indicators) == 24
    assert DUMMY_INDICATOR_WITH_RELATIONSHIP_LIST in indicators
    assert indicators == FETCH_RESULTS


def test_fetch_indicators_fails_on_invalid_attack_pattern_structure(mocker):
    """
    Given
        - Invalid attack pattern indicator structure

    When
        - fetching indicators

    Then
        - DemistoException is raised.
    """

    def mock_get_stix_objects(test, **kwargs):
        type_ = kwargs.get("type")
        client.objects_data[type_] = TYPE_TO_RESPONSE_WIITH_INVALID_ATTACK_PATTERN_DATA[type_]

    client = Client(api_key="1234", verify=False)
    mocker.patch.object(client, "fetch_stix_objects_from_api", side_effect=mock_get_stix_objects)

    with pytest.raises(DemistoException, match=r"Failed parsing attack indicator"):
        fetch_indicators(client, create_relationships=True)


def test_feed_tags_param(mocker):
    """
    Given
    - fetch incidents command
    - command args
    - command raw response
    When
    - mock the feed tags param.
    - mock the Client's get_stix_objects.
    Then
    - run the fetch incidents command using the Client
    Validate The value of the tags field.
    """

    def mock_get_stix_objects(test, **kwargs):
        type_ = kwargs.get("type")
        client.objects_data[type_] = TYPE_TO_RESPONSE[type_]

    client = Client(api_key="1234", verify=False)
    mocker.patch.object(client, "fetch_stix_objects_from_api", side_effect=mock_get_stix_objects)

    indicators = fetch_indicators(client, ["test_tag"])
    assert set(indicators[0].get("fields").get("tags")) == {"malicious-activity", "test_tag"}


@pytest.mark.parametrize(
    "field_name, field_value, expected_result",
    [
        ("created", "2017-05-31T21:31:43.540Z", "2017-05-31T21:31:43.540Z"),
        ("created", "2019-04-25T20:53:07.719Z\n2019-04-25T20:53:07.814Z", "2019-04-25T20:53:07.719Z"),
        ("modified", "2017-05-31T21:31:43.540Z", "2017-05-31T21:31:43.540Z"),
        ("modified", "2020-03-16T15:38:37.650Z\n2020-01-17T16:45:24.252Z", "2020-03-16T15:38:37.650Z"),
    ],
)
def test_handle_multiple_dates_in_one_field(field_name, field_value, expected_result):
    """
    Given
    - created / modified indicator field
    When
    - this field contains two dates
    Then
    - run the handle_multiple_dates_in_one_field
    Validate The field contain one specific date.
    """
    assert handle_multiple_dates_in_one_field(field_name, field_value) == expected_result


def test_parse_indicators():
    """
    Given
    - list of IOCs in STIX format.
    When
    - we extract this IOCs list to Demisto format
    Then
    - run the parse_indicators
    - Validate The IOCs list extracted successfully.

    """
    assert parse_indicators(INDICATORS_DATA, [], "")[0] == INDICATORS_RESULT


def test_parse_indicators_ioc_in_pattern():
    """
    Given
    - IOC in STIX format where the name value is file name and the ioc is in the pattern.
    When
    - we extract this IOCs list to Demisto format.
    Then
    - run the parse_indicators
    - Validate The indicator value is the file hash.
    - Validate The associatedfilenames value is the file name.

    """
    file_indicator = parse_indicators(INDICATORS_DATA, [], "")[10]
    assert file_indicator["value"] == "ca5fb5814ec621f4b79d"
    assert file_indicator["fields"]["associatedfilenames"] == "Jrdhtjydhjf.exe"


def test_parse_reports(mocker):
    """
    Given
    - list of reports in STIX format.
    When
    - we extract this reports list to Demisto format
    Then
    - run the parse_reports
    Validate The reports list extracted successfully.
    """
    client = Client(api_key="1234", verify=False)
    mocker.patch.object(client, "get_report_object", return_value=REPORTS_DATA[1])
    result = parse_reports_and_report_relationships(client, REPORTS_DATA, [], "")
    assert len(result) == 2
    assert result == REPORTS_INDICATORS


def test_parse_campaigns():
    """
    Given
    - list of campaigns in STIX format.
    When
    - we extract this campaigns list to Demisto format
    Then
    - run the parse_campaigns
    Validate The campaigns list extracted successfully.
    """
    client = Client(api_key="1234", verify=False)
    assert parse_campaigns(client, CAMPAIGN_RESPONSE, [], "") == CAMPAIGN_INDICATOR


def test_create_attack_pattern_indicator(mocker):
    """
    Given
    - list of IOCs in STIX format.
    When
    - we extract this attack pattern list to Demisto format
    Then
    - run the attack_pattern_indicator
    Validate The attack pattern list extracted successfully.
    """
    import TAXII2ApiModule

    client = Client(api_key="1234", verify=False)
    mocker.patch.object(TAXII2ApiModule, "is_demisto_version_ge", side_effect=[True, False, True])
    assert create_attack_pattern_indicator(client, ATTACK_PATTERN_DATA, [], "") == ATTACK_PATTERN_INDICATOR
    assert create_attack_pattern_indicator(client, ATTACK_PATTERN_DATA, [], "") == STIX_ATTACK_PATTERN_INDICATOR
    assert create_attack_pattern_indicator(client, SUB_TECHNIQUE_DATA, [], "") == SUB_TECHNIQUE_INDICATOR


def test_create_course_of_action_indicators():
    """
    Given
    - list of course of action in STIX format.
    When
    - we extract this course of action list to Demisto format
    Then
    - run the create_course_of_action_indicators
    Validate The course of action list extracted successfully.
    """
    client = Client(api_key="1234", verify=False)
    assert create_course_of_action_indicators(client, COURSE_OF_ACTION_DATA, [], "") == COURSE_OF_ACTION_INDICATORS


def test_get_ioc_type():
    """
    Given
    - IOC ID to get its type.
    When
    - we extract its type from the pattern field
    Then
    - run the get_ioc_type
    Validate The IOC type extracted successfully.
    """
    assert get_ioc_type("indicator--01a5a209-b94c-450b-b7f9-946497d91055", ID_TO_OBJECT) == "IP"
    assert get_ioc_type("indicator--fd0da09e-a0b2-4018-9476-1a7edd809b59", ID_TO_OBJECT) == "URL"


def test_get_ioc_value():
    """
    Given
    - IOC ID to get its value.
    When
    - we extract its value from the name field
    Then
    - run the get_ioc_value
    Validate The IOC value extracted successfully.
    """
    assert get_ioc_value("indicator--01a5a209-b94c-450b-b7f9-946497d91055", ID_TO_OBJECT) == "T111: Software Discovery"
    assert get_ioc_value("indicator--fd0da09e-a0b2-4018-9476-1a7edd809b59", ID_TO_OBJECT) == "Deploy XSOAR Playbook"
    assert get_ioc_value("report--0f86dccd-29bd-46c6-83fd-e79ba040bf0", ID_TO_OBJECT) == "[Unit42 ATOM] Maze Ransomware"
    assert (
        get_ioc_value("attack-pattern--4bed873f-0b7d-41d4-b93a-b6905d1f90b0", ID_TO_OBJECT)
        == "Virtualization/Sandbox Evasion: Time Based Evasion"
    )


def test_create_list_relationships():
    """
    Given
    - list of relationships in STIX format.
    When
    - we extract this relationships list to Demisto format
    Then
    - run the create_list_relationships
    Validate The relationships list extracted successfully.
    """
    assert create_list_relationships(RELATIONSHIP_DATA, ID_TO_OBJECT) == RELATIONSHIP_OBJECTS


def test_extract_ioc_value():
    """
    Given
    - IOC obj to get its value.
    When
    - we extract its value from the name field
    Then
    - run the get_ioc_value
    Validate The IOC value extracted successfully.
    """
    name = "([file:name = 'blabla' OR file:name = 'blabla'] AND [file:hashes.'SHA-256' = '4f75622c2dd839f'])"
    assert extract_ioc_value(name) == "4f75622c2dd839f"


def test_fetch_indicators_command_with_relationship(mocker):
    """
    Given
    - fetch incidents command
    - command args
    - command raw response
    When
    - mock the Client's get_stix_objects.
    Then
    - run the fetch incidents command using the Client
    Validate the amount of indicators fetched
    Validate that the dummy indicator with the relationships list fetched
    """

    def mock_get_stix_objects(test, **kwargs):
        type_ = kwargs.get("type")
        client.objects_data[type_] = TYPE_TO_RESPONSE_FETCH[type_]

    client = Client(api_key="1234", verify=False)
    mocker.patch.object(client, "fetch_stix_objects_from_api", side_effect=mock_get_stix_objects)

    indicators = fetch_indicators(client, create_relationships=True)
    assert len(indicators) == 23
    assert DUMMY_INDICATOR_WITH_RELATIONSHIP_LIST in indicators
    assert REPORTS_INDICATORS_WITH_RELATIONSHIPS in indicators


def test_create_course_of_action_indicators_with_tlp():
    """
    Given
    - fetch indicator command.
    - mock Client.
    When
    - call the create_course_of_action_indicators method
    Then
    - run the create_course_of_action_indicators method with TLP
    - Validate that the TLP value was set correctly.
    """

    client = Client(api_key="1234", verify=False)
    assert create_course_of_action_indicators(client, COURSE_OF_ACTION_DATA, [], "WHITE") == COURSE_OF_ACTION_INDICATORS_WITH_TLP


def test_fetch_indicators_malware(mocker):
    """
    Given
    - fetch indicator command.
    - mock Client.
    When
    - call the fetch_indicators method
    Then
    - run the fetch_indicators method.
    - Validate that the malware objects created correctly.
    """

    def mock_get_stix_objects(test, **kwargs):
        type_ = kwargs.get("type")
        client.objects_data[type_] = TYPE_TO_RESPONSE_FETCH[type_]

    client = Client(api_key="1234", verify=False)
    mocker.patch.object(client, "fetch_stix_objects_from_api", side_effect=mock_get_stix_objects)
    indicators = fetch_indicators(client, create_relationships=True)
    for i in range(18, 23):
        assert indicators[i]["type"] == "Malware"
    assert len(indicators) == 23


@pytest.mark.parametrize(
    "report_obj, expected_result",
    [
        ({"object_refs": ["indicator--a", "indicator--b", "indicator--c", "indicator--d"]}, True),
        ({"object_refs": ["intrusion-set--a", "indicator--ab"]}, True),
        ({"object_refs": ["intrusion-set--a", "indicator--ab"], "description": "description"}, False),
        ({"object_refs": ["intrusion-set--a"], "description": "description"}, False),
        ({"object_refs": ["intrusion-set--a"]}, False),
        ({"object_refs": ["indicator--a", "indicator--ab"]}, True),
        ({"object_refs": ["indicator--a", "indicator--ab"], "description": "description"}, False),
        ({"object_refs": ["intrusion-set--a", "report--ab"], "description": "description"}, False),
        ({"object_refs": ["intrusion-set--a", "report--ab"]}, False),
    ],
)
def test_is_atom42_sub_report(report_obj, expected_result):
    """
    Given
    - report object.
    When
    - call the is_atom42_sub_report method
    Then
    - run the is_atom42_sub_report method with a report.
    - Validate that the function output is correct.
    """
    from FeedUnit42v2 import is_atom42_sub_report

    assert is_atom42_sub_report(report_obj) is expected_result


@pytest.mark.parametrize(
    "report_obj, expected_result",
    [
        ({"object_refs": ["indicator--a", "indicator--b", "indicator--c", "indicator--d"]}, False),
        ({"object_refs": ["indicator--a", "indicator--b", "indicator--c", "indicator--d"], "description": "description"}, False),
        ({"object_refs": ["intrusion-set--a", "indicator--ab"]}, False),
        ({"object_refs": ["intrusion-set--a", "indicator--ab"], "description": "description"}, False),
        ({"object_refs": ["intrusion-set--a"], "description": "description"}, False),
        ({"object_refs": ["intrusion-set--a"]}, False),
        ({"object_refs": ["indicator--a", "indicator--ab"]}, False),
        ({"object_refs": ["indicator--a", "indicator--ab"], "description": "description"}, False),
        ({"object_refs": ["intrusion-set--a", "report--ab"], "description": "description"}, True),
        ({"object_refs": ["intrusion-set--a", "report--ab"]}, True),
    ],
)
def test_is_atom42_main_report(report_obj, expected_result):
    """
    Given
    - report object.
    When
    - call the is_atom42_main_report method
    Then
    - run the is_atom42_main_report method with report
    - Validate that the function output is correct.
    """

    from FeedUnit42v2 import is_atom42_main_report

    assert is_atom42_main_report(report_obj) is expected_result


def test_test_module(mocker):
    """
    Given
    - A response from the API.
    When
    - call the test_module method
    Then
    - run the test_module method
    - Validate that the response is correct.
    """

    def mock_get_stix_objects(test, **kwargs):
        type_ = kwargs.get("type")
        client.objects_data[type_] = TYPE_TO_RESPONSE_FETCH[type_]

    client = Client(api_key="1234", verify=False)
    mocker.patch.object(client, "fetch_stix_objects_from_api", side_effect=mock_get_stix_objects)
    from FeedUnit42v2 import test_module

    assert test_module(client) == "ok"


def test_get_report_object(mocker):
    """
    Given
    - A Client.
    When
    - call the fetch_stix_objects_from_api method
    Then
    - run the fetch_stix_objects_from_api method
    - Validate the debug logs.
    """

    def mock_get_stix_objects(**kwargs):
        type_ = kwargs.get("type")
        client.objects_data[type_] = TYPE_TO_RESPONSE_FETCH[type_]
        return TYPE_TO_RESPONSE_FETCH[type_]

    client = Client(api_key="1234", verify=False)
    mocker.patch.object(client, "fetch_stix_objects_from_api", side_effect=mock_get_stix_objects)
    debug_logs_mock = mocker.patch.object(demisto, "debug")
    client.get_report_object("object_id")
    assert (
        "Unit42v2 Feed: Found more then one object for report object object_id skipping" in debug_logs_mock.call_args_list[0][0]
    )


def test_parse_indicators_no_name():
    """
    Given
    - An indicator without a name.
    When
    - Calling the fetch-indicators command.
    Then
    - Use value in the pattern.
    """
    indicators = [{"pattern": "[domain-name:value = 'www.example.com']"}]

    res = parse_indicators(indicators)

    assert res[0]["value"] == "www.example.com"