MalwareBazaar Feed

import demistomock as demisto
import pytest
from pytest_mock import MockerFixture
from requests import HTTPError


def test_fetch_indicators_main(mocker: MockerFixture):
    """
    Given
    - indicators response from malware bazzar feed

    When
    - Running main flow for fetching indicators command

    Then
    - Ensure that all indicators values exist and are not 'None'
    """
    from JSONFeedApiModule import Client
    from MalwareBazaarFeed import main

    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "feed": True,
            "feedBypassExclusionList": False,
            "feedExpirationInterval": "20160",
            "feedExpirationPolicy": "suddenDeath",
            "feedFetchInterval": 1,
            "feedReliability": "A - Completely reliable",
            "feedReputation": "None",
            "feedTags": None,
            "insecure": True,
            "ip_ranges": "All available Google IP ranges",
            "proxy": False,
            "tlp_color": None,
            "url": "https://mb-api.abuse.ch",
            "credentials": {"password": "test"},
        },
    )
    mocker.patch.object(demisto, "command", return_value="fetch-indicators")
    create_indicators_mocker = mocker.patch.object(demisto, "createIndicators")

    mocker.patch.object(
        Client,
        "build_iterator",
        side_effect=[([{"sha256_hash": "1234"}, {"sha256_hash": "12345"}, {"sha256_hash": "123456"}], True), ([], True)],
    )

    main()

    assert create_indicators_mocker.call_args.args[0] == [
        {
            "type": "File",
            "fields": {"tags": [], "downloadurl": "https://bazaar.abuse.ch/sample/1234/"},
            "value": "1234",
            "rawJSON": {"sha256_hash": "1234"},
        },
        {
            "type": "File",
            "fields": {"tags": [], "downloadurl": "https://bazaar.abuse.ch/sample/12345/"},
            "value": "12345",
            "rawJSON": {"sha256_hash": "12345"},
        },
        {
            "type": "File",
            "fields": {"tags": [], "downloadurl": "https://bazaar.abuse.ch/sample/123456/"},
            "value": "123456",
            "rawJSON": {"sha256_hash": "123456"},
        },
    ]


def test_custom_build_relationships():
    """
    Given
    - no feed config

    When
    - Running custom_build_relationships function

    Then
    - Ensure the relationship list is empty
    """
    from MalwareBazaarFeed import custom_build_relationships

    assert custom_build_relationships(feed_config={}, _mapping={}, indicator_data={}) == []


def test_custom_mapping_function():
    """
    Given
    - mapping, attributes and indicators

    When
    - Running custom_mapping_function function

    Then
    - Ensure indicator dict is updated
    """
    from MalwareBazaarFeed import custom_mapping_function

    mapping = {"a": "a.b", "c": "c", "d": "d"}
    attributes = {"a": "a", "b": "b", "c": "c"}
    indicator = {"fields": {"a": [{"a": "a"}], "b": "b"}}

    custom_mapping_function(mapping, indicator, attributes)

    assert indicator == {"fields": {"a": [{"a": "a", "b": "a"}], "b": "b", "c": "c"}}


@pytest.mark.parametrize(
    "err_msg",
    [
        pytest.param("502 Server Error: Bad Gateway for url: https://test/v1/", id="HTTPError 502"),
        pytest.param("Connection broken: IncompleteRead(3923 bytes read, 4269 more expected", id="Connection broken"),
        pytest.param("503 Server Error: Service Unavailable for url: https://test/v1/", id="HTTPError 503"),
    ],
)
def test_exceptions_handle_known_exception(mocker: MockerFixture, err_msg: str):
    """
    In the main function of the MalwareBazaar Feed we handle exceptions as follows:
        - Catch the exceptions occurred during the run of the 'feed_main' function.
        - Check if the exception is a known error.
        - If yes - wait 10 seconds and start another retry of the 'feed_main' function.
        (number or retries is 3).
    This test tests this logic.

    Given
        - A mock response of the 'feed_main' function that raises a known exception.
        1. The exception is a known 502 server error.
        2. The exception is a known connection error.
        3. The exception is a known 503 server error.

    When
        - Running main flow for the fetching indicators command.

    Then
        - Verify that the exception is caught and the 'feed_main' function is called 4 times in total.
        - Verify that the error logs are as expected.
    """
    from MalwareBazaarFeed import main

    mocker.patch.object(demisto, "params", return_value={"credentials": {"password": "test"}})
    mocker.patch.object(demisto, "command", return_value="fetch-indicators")
    error_log_mock = mocker.patch.object(demisto, "error")

    def mock_feed_main(params, feed_name, prefix):
        raise HTTPError(err_msg)

    feed_main_mock = mocker.patch("MalwareBazaarFeed.feed_main", side_effect=mock_feed_main)

    with pytest.raises(Exception):
        main()
    assert "An Error Occurred during the run of the 'fetch-indicators' command" in error_log_mock.call_args_list[0][0][0]
    assert err_msg in error_log_mock.call_args_list[0][0][0]
    assert "Retrying in 10 seconds..." in error_log_mock.call_args_list[1][0][0]
    assert "This is attempt number: 1" in error_log_mock.call_args_list[1][0][0]
    assert "This is attempt number: 2" in error_log_mock.call_args_list[3][0][0]
    assert "This is attempt number: 3" in error_log_mock.call_args_list[5][0][0]
    assert feed_main_mock.call_count == 4


def test_exceptions_handle_unknown_exception(mocker):
    """
    In the main function of the MalwareBazaar Feed we handle exceptions as follows:
        - Catch the exceptions occurred during the run of the 'feed_main' function.
        - Check if the exception is a known error.
        - If yes - wait 10 seconds and start another retry of the 'feed_main' function.
        (number or retries is 3).
        - If no - raises an error.
    This test tests this logic.

    Given
        - A mock response of the 'feed_main' function that raises an un-known exception.

    When
        - Running main flow for the fetching indicators command.

    Then
        - Verify that the exception is caught but the 'feed_main' function called only once.
        - Verify that the original error is raised.
    """
    from MalwareBazaarFeed import main

    mocker.patch.object(demisto, "params", return_value={"credentials": {"password": "test"}})
    mocker.patch.object(demisto, "command", return_value="fetch-indicators")
    error_log_mock = mocker.patch.object(demisto, "error")

    def mock_feed_main(params, feed_name, prefix):
        raise HTTPError(
            "[429] Failed with error: [parent] Data too large, data for [<http_request>] "
            "would be [1033445328/985.5mb], which is larger than the limit of [1020054732/972.7mb]"
        )

    feed_main_mock = mocker.patch("MalwareBazaarFeed.feed_main", side_effect=mock_feed_main)

    with pytest.raises(Exception) as e:
        main()
    assert "[429] Failed with error: [parent] Data too large, data for [<http_request>]" in str(e)
    assert len(error_log_mock.call_args_list) == 0
    assert feed_main_mock.call_count == 1


def test_exception_thrown_when_no_auth_key_param(mocker):
    from MalwareBazaarFeed import main

    mocker.patch.object(demisto, "params", return_value={})
    mocker.patch.object(demisto, "command", return_value="fetch-indicators")
    error_log_mock = mocker.patch.object(demisto, "error")

    def mock_feed_main(params, feed_name, prefix):
        pass

    feed_main_mock = mocker.patch("MalwareBazaarFeed.feed_main", side_effect=mock_feed_main)

    with pytest.raises(Exception) as e:
        main()

    assert "Missing required parameter Auth Key. Please set this parameter in the instance configuration." in str(e)
    assert len(error_log_mock.call_args_list) == 0
    assert feed_main_mock.call_count == 0


def test_no_exception_thrown_when_no_auth_key_param(mocker):
    from MalwareBazaarFeed import main

    mocker.patch.object(demisto, "params", return_value={"credentials": {"password": "test"}})
    mocker.patch.object(demisto, "command", return_value="fetch-indicators")
    error_log_mock = mocker.patch.object(demisto, "error")

    def mock_feed_main(params, feed_name, prefix):
        return None

    feed_main_mock = mocker.patch("MalwareBazaarFeed.feed_main", side_effect=mock_feed_main)

    main()
    assert len(error_log_mock.call_args_list) == 0
    assert feed_main_mock.call_count == 1


@pytest.mark.parametrize(
    "feed_tags, api_tags, expected_tags",
    [
        pytest.param(
            ["source_mbazaar"],
            ["SilverFox", "ValleyRAT", "zip"],
            ["source_mbazaar", "SilverFox", "ValleyRAT", "zip"],
            id="merges_feed_tags_with_api_tags",
        ),
        pytest.param(
            ["source_mbazaar"],
            None,
            ["source_mbazaar"],
            id="preserves_feed_tags_when_api_tags_is_none",
        ),
        pytest.param(
            ["source_mbazaar"],
            [],
            ["source_mbazaar"],
            id="preserves_feed_tags_when_api_tags_is_empty",
        ),
        pytest.param(
            ["source_mbazaar", "SilverFox"],
            ["SilverFox", "ValleyRAT"],
            ["source_mbazaar", "SilverFox", "ValleyRAT"],
            id="deduplicates_overlapping_tags",
        ),
    ],
)
def test_custom_mapping_function_tags_merge(feed_tags: list, api_tags: list | None, expected_tags: list):
    """
    Regression test for XSUP-68506:
    The 'tags' mapping entry in MalwareBazaarFeed must MERGE the API response tags
    with the existing feedTags (set by JSONFeedApiModule.handle_indicator) rather
    than overwriting them.

    Given
    - An indicator whose fields["tags"] has already been populated with feedTags
      by handle_indicator.
    - A mapping that maps the API "tags" field to the indicator "tags" field.
    - Attributes from the API response with various tag values (list, None, empty).

    When
    - custom_mapping_function is called.

    Then
    - The resulting indicator["fields"]["tags"] matches the expected merged/deduplicated list.
    - The configured feedTags value is NOT lost.
    """
    from MalwareBazaarFeed import custom_mapping_function

    mapping = {"tags": "tags"}
    attributes = {"tags": api_tags}
    indicator = {"fields": {"tags": feed_tags.copy()}, "value": "abc123"}

    custom_mapping_function(mapping, indicator, attributes)

    assert sorted(indicator["fields"]["tags"]) == sorted(expected_tags)


def test_exceptions_handler_no_exception(mocker):
    """
    In the main function of the MalwareBazaar Feed we handle exceptions as follows:
        - Catch the exceptions occurred during the run of the 'feed_main' function.
        - Check if the exception is a known error.
        - If yes - wait 10 seconds and start another retry of the 'feed_main' function.
        (number or retries is 3).
        - If no - raises an error.
        - If no error occurred the function is ended (without any recursive calls).
    This test tests this logic in case of a success (when no errors were detected in the feed_main function).

    Given
        - A mock response of the 'feed_main' function that returns None.

    When
        - Running main flow for the fetching indicators command.

    Then
        - Verify that the 'feed_main' function called only once.
    """
    from MalwareBazaarFeed import main

    mocker.patch.object(demisto, "params", return_value={"credentials": {"password": "test"}})
    mocker.patch.object(demisto, "command", return_value="fetch-indicators")
    error_log_mock = mocker.patch.object(demisto, "error")

    def mock_feed_main(params, feed_name, prefix):
        return None

    feed_main_mock = mocker.patch("MalwareBazaarFeed.feed_main", side_effect=mock_feed_main)

    main()
    assert len(error_log_mock.call_args_list) == 0
    assert feed_main_mock.call_count == 1