TAXII 2 Feed

Ingests indicator feeds from TAXII 2.0 and 2.1 servers.

Data Enrichment & Threat Intelligence · TAXII Feed · Feed

Details

ID	`TAXII 2 Feed`
Provider	Open Source
Category	Data Enrichment & Threat Intelligence
From Version	5.5.0
Docker Image	`demisto/taxii2:1.0.0.11040999`
Supported Modules	Agentix XSIAM

README

Ingest indicator feeds from TAXII 2.0 and 2.1 servers.

Configure TAXII 2 Feed in Cortex

Parameter	Description	Required
Name		True
Discovery Service URL (e.g. https://example.net/taxii)		True
Username / API Key		False
Password		False
API Root to Use	The API root to use (for example default or public). If left empty, the server default API root is used. If the server has no default root, the first available API root is used instead.	False
Collection Name To Fetch Indicators From	Indicators will be fetched from this collection. Run “taxii2-get-collections” command to get a valid value. If left empty, the instance will try to fetch from all the collections in the given discovery service.	False
Certificate File as Text	Add a certificate file as text to connect to the TAXII server.	False
Key File as Text	Add a key file as text to connect to the TAXII server	False
Run on Single engine		False
Trust any certificate (not secure)	Located under Advanced Settings.	False
Use system proxy settings	Located under Advanced Settings.	False
Log Level	Debug/Verbose logging is recommended only during troubleshooting. Logging can affect integration performance. Recommended usage is to turn logging on during setup and troubleshooting, and then turn it off in production. These settings only affect the integration log. The server log is not affected.	False
Do not use in CLI by default		False
Fetch indicators		False
Classifier	Determines the type of incident that is created for events ingested from this integration instance.	False
Mapper (incoming)	Determines how event fields are mapped to the Cortex XSOAR incident fields.	False
Indicator Reputation	Indicators from this integration instance will be marked with this reputation.	False
Source Reliability	Reliability of the source providing the intelligence data.	True
Traffic Light Protocol Color	The Traffic Light Protocol (TLP) designation to apply to indicators fetched from the feed.	False
Max Indicators Per Fetch (disabled for Full Feed Fetch)	The maximum number of indicators that can be fetched per fetch. If this field is left empty, there will be no limit on the number of indicators fetched.	False
First Fetch Time	The time interval for the first fetch (retroactive). <number> <time unit> of type minute/hour/day/year. For example, 1 minute, 12 hour	False
STIX Objects To Fetch	Which STIX objects to fetch from the TAXII server. If left empty, all available object types will be fetched.	False
Max STIX Objects Per Poll	Set the number of stix object that will be requested with each TAXII poll (http request). A single fetch is made of several taxii polls. Changing this setting can help speed up fetches, or fix issues on slower networks. Please note server restrictions may apply, overriding and limiting the “requested limit”.	False
Enrichment Excluded	Select this option to exclude the fetched indicators from the enrichment process.	False
Indicator Expiration Method		False
Feed Fetch Interval		False
Bypass exclusion list	When selected, the exclusion list is ignored for indicators from this feed. This means that if an indicator from this feed is on the exclusion list, the indicator might still be added to the system.	False
Incremental Feed	Incremental feeds pull only new or modified indicators that have been sent from the integration. As the determination if the indicator is new or modified happens on the 3rd-party vendor’s side, and only indicators that are new or modified are sent to Cortex XSOAR, all indicators coming from these feeds are labeled new or modified.	False
Full Feed Fetch	When enabled, fetch-indicators will try to fetch the entire feed for every fetch. When disabled, fetch-indicators will try to fetch just the latest entries (since the last fetch).	False
Complex Observation Mode	Choose how to handle complex observations. Two or more observation expressions can be combined using a complex observation operator such as “AND”, “OR”. e.g. `[ IP = ‘b’ ] AND [ URL = ‘d’ ]`	False
Update custom fields	Choose whether to import the XSOAR custom fields. Note: this might overwrite the data pulled from other sources.	False
Tags	Supports CSV values.	False

Using API token authentication

To use the integration with an API token you first need to change the Username / API Key (see '?') field to _api_token_key. You can then enter the API Token into the Password field - this value will be used as an API key.

Using a custom authentication header

If the TAXII 2 server you are trying to connect to requires a custom authentication header, you first need to change the Username / API Key (see '?') field to _header: and the custom header name, e.g. _header:custom_auth. You can then enter the custom auth header value into the Password field - this value will be used as a custom auth header.

Complex Observation Mode consideration

You can combine two or more observation expressions using a complex observation operator such as “AND”, “OR”, and “FOLLOWEDBY”, for example [ IP = 'b' ] AND [ URL = 'd' ]. These relationships are not represented in CORTEX XSOAR TIM indicators. You can create them while ignoring these relations, or you can ignore these expressions. If you choose to ignore these expressions, then no indicator will be created for complex observations.

Enrichment Excluded consideration

Setting the Traffic Light Protocol Color to red automatically excludes enrichment, even if Enrichment Excluded is unchecked.

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.

taxii2-get-indicators

Allows you to test your feed and to make sure you can fetch indicators successfuly.

Base Command

taxii2-get-indicators

Input

Argument Name	Description	Required
raw	Will return only the rawJSON of the indicator object.	Optional
limit	Maximum number of indicators to fetch.	Optional
added_after	Fetch only indicators that were added to the server after the given time. Please provide a <number> and <time unit> of type minute/hour/day. For example, 1 minute, 12 hour, 24 days.	Optional

Context Output

Path	Type	Description
TAXII2.Indicators.type	String	Indicator type.
TAXII2.Indicators.value	String	Indicator value.
TAXII2.Indicators.rawJSON	String	Indicator rawJSON.

Command Example

!taxii2-get-indicators limit=3

Human Readable Output

value	type
coronashop.jp	Domain
e6ecb146f469d243945ad8a5451ba1129c5b190f7d50c64580dbad4b8246f88e	File
2014[.]zzux[.]com	Domain

taxii2-get-collections

Gets the list of collections from the discovery service.

Base Command

taxii2-get-collections

Input

There are no input arguments for this command.

Context Output

Path	Type	Description
TAXII2.Collections.ID	String	Collection ID.
TAXII2.Collections.Name	String	Collection Name.

Command Example

!taxii2-get-collections

Human Readable Output

Name	ID
Phish Tank	107
Abuse.ch Ransomware IPs	135
Abuse.ch Ransomware Domains	136
DShield Scanning IPs	150
Malware Domain List - Hotlist	200
Blutmagie TOR Nodes	209
Emerging Threats C&C Server	31
DT COVID-19	313
Lehigh Malwaredomains	33
CyberCrime	41
Emerging Threats - Compromised	68

taxii2-reset-fetch-indicators

WARNING: This command will reset your fetch history.

Base Command

taxii2-reset-fetch-indicators

Input

There are no input arguments for this command.

Context Output

There is no context output for this command.

Command Example

!taxii2-reset-fetch-indicators

Human Readable Output

Fetch was reset successfully. Your next indicator fetch will collect indicators from the configured “First Fetch Time”

Troubleshooting

When the feed is set to “Incremental Feed”, we recommend specifying a value for the Max Indicators Per Fetch parameter to prevent potential timeout issues.

Configuration parameters

feed — Fetch indicators
feedReputation — Indicator Reputation
feedReliability — Source Reliability (required)
tlp_color — Traffic Light Protocol Color
feedExpirationPolicy —
feedExpirationInterval —
feedFetchInterval — Feed Fetch Interval
feedBypassExclusionList — Bypass exclusion list
url — Discovery Service URL (e.g. https://example.net/taxii2) (required)
credentials — Username / API Key
default_api_root — API Root to Use
collection_to_fetch — Collection Name To Fetch Indicators From
feedIncremental — Incremental Feed
fetch_full_feed — Full Feed Fetch
limit — Max Indicators Per Fetch (disabled for Full Feed Fetch)
initial_interval — First Fetch Time
objects_to_fetch — STIX Objects To Fetch
creds_certificate — Certificate File as Text
certificate — Certificate File as Text
key — Key File as Text
limit_per_request — Max STIX Objects Per Poll
observation_operator_mode — Complex Observation Mode
update_custom_fields — Update custom fields
insecure — Trust any certificate (not secure)
proxy — Use system proxy settings
feedTags — Tags
enrichmentExcluded — Enrichment Excluded

Commands (3)

taxii2-get-collections
Gets the list of collections from the discovery service.
taxii2-get-indicators
Allows you to test your feed and to make sure you can fetch indicators successfuly.
taxii2-reset-fetch-indicators Deprecated
WARNING: This command will reset your fetch history.

Source

FeedTAXII2.py FeedTAXII2_test.py FeedTAXII2.yml FeedTAXII2_description.md

Download History Raw

import json

import pytest
import demistomock as demisto
from FeedTAXII2 import *

with open("test_data/results.json") as f:
    RESULTS_JSON = json.load(f)
with open("test_data/cortex_indicators_1.json") as f:
    CORTEX_IOCS_1 = json.load(f)
with open("test_data/cortex_indicators_1.json") as f:
    CORTEX_IOCS_2 = json.load(f)


class MockCollection:
    def __init__(self, id_, title):
        self.id = id_
        self.title = title


class TestFetchIndicators:
    """
    Scenario: Test fetch_indicators_command
    """

    def test_single_no_context(self, mocker):
        """
        Scenario: Test single collection fetch with no last run

        Given:
        - collection to fetch is available and set to 'default'
        - there is no integration context
        - limit is -1
        - initial interval is `1 day`

        When:
        - fetch_indicators_command is called

        Then:
        - update last run with latest collection fetch time
        """
        mock_client = Taxii2FeedClient(url="", collection_to_fetch="default", proxies=[], verify=False, objects_to_fetch=[])
        default_id = 1
        nondefault_id = 2
        mock_client.collections = [MockCollection(default_id, "default"), MockCollection(nondefault_id, "not_default")]

        mock_client.collection_to_fetch = mock_client.collections[0]
        result = RESULTS_JSON.get("Contents")
        mocker.patch.object(mock_client, "build_iterator", return_value=result)
        indicators, last_run = fetch_indicators_command(mock_client, "1 day", -1, {})
        assert indicators == result
        assert mock_client.collection_to_fetch.id in last_run

    def test_single_with_context(self, mocker):
        """
        Scenario: Test single collection fetch with no last run context

        Given:
        - collection to fetch is available and set to 'default'
        - there is an integration context, with 2 collections
        - limit is -1
        - initial interval is `1 day`

        When:
        - fetch_indicators_command is called

        Then:
        - update last run with latest collection fetch time
        - don't update collection that wasn't fetched from
        """
        mock_client = Taxii2FeedClient(url="", collection_to_fetch="default", proxies=[], verify=False, objects_to_fetch=[])
        default_id = 1
        nondefault_id = 2
        mock_client.collections = [MockCollection(default_id, "default"), MockCollection(nondefault_id, "not_default")]

        mock_client.collection_to_fetch = mock_client.collections[0]
        last_run = {mock_client.collections[1]: "test"}
        result = RESULTS_JSON.get("Contents")
        mocker.patch.object(mock_client, "build_iterator", return_value=result)
        indicators, last_run = fetch_indicators_command(mock_client, "1 day", -1, last_run)
        assert indicators == result
        assert mock_client.collection_to_fetch.id in last_run
        assert last_run.get(mock_client.collections[1]) == "test"

    def test_multi_no_context(self, mocker):
        """
        Scenario: Test multi collection fetch with no last run

        Given:
        - collection to fetch is set to None
        - there is no integration context
        - limit is -1
        - initial interval is `1 day`

        When:
        - fetch_indicators_command is called

        Then:
        - fetch 14 indicators
        - update last run with latest collection fetch time
        """
        mock_client = Taxii2FeedClient(url="", collection_to_fetch=None, proxies=[], verify=False, objects_to_fetch=[])
        default_id = 1
        nondefault_id = 2
        mock_client.collections = [MockCollection(default_id, "default"), MockCollection(nondefault_id, "not_default")]

        mocker.patch.object(mock_client, "build_iterator", side_effect=[CORTEX_IOCS_1, CORTEX_IOCS_2])
        indicators, last_run = fetch_indicators_command(mock_client, "1 day", -1, {})
        assert len(indicators) == 14
        assert mock_client.collection_to_fetch.id in last_run

    @pytest.mark.parametrize("empty_collection_type", [None, ""])
    def test_multi_with_context(self, mocker, empty_collection_type):
        """
        Scenario: Test multi collection fetch with no last run, testing both types of empty collection

        Given:
        - collection to fetch is set to None
        - there is no integration context
        - limit is len(CORTEX_IOCS_1)
        - initial interval is `1 day`

        When:
        - fetch_indicators_command is called

        Then:
        - fetch 7 indicators
        - update last run with latest collection fetch time
        """
        mock_client = Taxii2FeedClient(
            url="", collection_to_fetch=empty_collection_type, proxies=[], verify=False, objects_to_fetch=[]
        )
        id_1 = 1
        id_2 = 2
        mock_client.collections = [MockCollection(id_1, "a"), MockCollection(id_2, "b")]

        last_run = {mock_client.collections[1]: "test"}
        mocker.patch.object(mock_client, "build_iterator", side_effect=[CORTEX_IOCS_1, CORTEX_IOCS_2])
        indicators, last_run = fetch_indicators_command(mock_client, "1 day", len(CORTEX_IOCS_1), last_run)
        assert len(indicators) == len(CORTEX_IOCS_1)
        assert last_run.get(mock_client.collections[1]) == "test"


def test_get_collections_function():
    mock_client = Taxii2FeedClient(url="", collection_to_fetch=None, proxies=[], verify=False, objects_to_fetch=[])
    mock_client.collections = [MockCollection("first id", "first name"), MockCollection("second id", "second name")]

    result = get_collections_command(mock_client)

    assert len(result.outputs) == 2
    assert result.outputs[0] == {"Name": "first name", "ID": "first id"}
    assert result.outputs[1] == {"Name": "second name", "ID": "second id"}


@pytest.mark.parametrize(
    "response, expected_md_results",
    [
        (
            [{"value": "1.1.1.1", "type": "IP"}, {"value": "google.com", "type": "Domain"}],
            "Found 2 results:\n|value|type|\n|---|---|\n| 1.1.1.1 | IP |\n| google.com | Domain |\n",
        ),
        (
            [
                {"value": "1.1.1.1", "type": "IP"},
                {"value": "google.com", "type": "Domain"},
                {
                    "value": "$$DummyIndicator$$",
                    "relationships": [
                        {
                            "name": "related-to",
                            "reverseName": "related-to",
                            "type": "IndicatorToIndicator",
                            "entityA": "1.1.1.1",
                            "entityAFamily": "Indicator",
                            "entityAType": "IP",
                            "entityB": "google.com",
                            "entityBFamily": "Indicator",
                            "entityBType": "Domain",
                            "fields": {
                                "lastseenbysource": "2023-03-26T12:45:55.068670Z",
                                "firstseenbysource": "2023-03-26T12:45:55.068662Z",
                            },
                        }
                    ],
                },
            ],
            "Found 2 results:\n|value|type|\n|---|---|\n| 1.1.1.1 | IP |\n| google.com | Domain |\n\n\n\nRelations"
            " ships:\n|entityA|entityAFamily|entityAType|entityB|entityBFamily|entityBType|fields|name|"
            "reverseName|type|\n|---|---|---|---|---|---|---|---|---|---|\n| 1.1.1.1 | Indicator | IP | "
            "google.com | Indicator | Domain | lastseenbysource: 2023-03-26T12:45:55.068670Z<br>"
            "firstseenbysource: 2023-03-26T12:45:55.068662Z | related-to | related-to | IndicatorToIndicator |\n",
        ),
    ],
)
def test_get_indicators_command(mocker, response, expected_md_results):
    """
    Given:
    - A mock response
    - Case 1: response with 2 indicators and no relationship between them.
    - Case 2: response with 2 indicators and a relationship between them.

    When:
    - calling test_get_indicators_command

    Then:
    - Ensure the information was parsed correctly.
    - Case 1: No relationships section is mentioned in the readable output.
    - Case 2: Relationships section is mentioned in the readable output,
    and both the indicators and the relationship are in the outputs section.
    """
    mock_client = Taxii2FeedClient(url="", collection_to_fetch=None, proxies=[], verify=False, objects_to_fetch=[])
    mock_client.collection_to_fetch = [1]
    mocker.patch.object(Taxii2FeedClient, "build_iterator", return_value=response)
    results = get_indicators_command(mock_client)
    md = results.readable_output
    outputs = results.outputs
    assert md == expected_md_results
    assert outputs == response


class TestHelperFunctions:
    def test_try_parse_integer(self):
        assert try_parse_integer(None, "") is None
        assert try_parse_integer("8", "") == 8
        assert try_parse_integer(8, "") == 8
        with pytest.raises(DemistoException, match="parse failure"):
            try_parse_integer("a", "parse failure")

    class TestAssertIncrementalFeedParams:
        """Scenario: Test assert_incremental_feed_params raises appropriate errors"""

        def test_both_params_are_false(self):
            """
            Scenario: Both params are False

            Given:
            - fetch_full_feed is false
            - feedIncremental is false

            When:
            - calling assert_incremental_feed_params

            Then:
            - raise appropriate error
            """
            fetch_full_feed = is_incremental_feed = False
            with pytest.raises(DemistoException) as e:
                assert_incremental_feed_params(fetch_full_feed, is_incremental_feed)
                assert "'Full Feed Fetch' cannot be disabled when 'Incremental Feed' is disabled." in str(e)

        def test_both_params_are_true(self):
            """
            Scenario: Both params are True

            Given:
            - fetch_full_feed is true
            - feedIncremental is true

            When:
            - calling assert_incremental_feed_params

            Then:
            - raise appropriate error
            """
            fetch_full_feed = is_incremental_feed = True
            with pytest.raises(DemistoException) as e:
                assert_incremental_feed_params(fetch_full_feed, is_incremental_feed)
                assert "'Full Feed Fetch' cannot be enabled when 'Incremental Feed' is enabled." in str(e)

        def test_params_have_different_values(self):
            """
            Scenario: Both params are False

            Given:
            - fetch_full_feed is false / true
            - feedIncremental is true / false

            When:
            - calling assert_incremental_feed_params

            Then:
            - don't raise any error
            """
            fetch_full_feed = False
            is_incremental_feed = True
            assert_incremental_feed_params(fetch_full_feed, is_incremental_feed)

            fetch_full_feed = True
            is_incremental_feed = False
            assert_incremental_feed_params(fetch_full_feed, is_incremental_feed)

    class TestGetAddedAfter:
        """Scenario: Test get_added_after"""

        def test_get_last_fetch_time(self):
            """
            Scenario: fetch_full_feed and last fetch is set

            Given:
            - fetch_full_feed is false
            - last fetch time is set

            When:
            - calling get_added_after

            Then:
            - return last fetch time
            """
            fetch_full_feed = False
            last_fetch_time = "last_fetch_mock"
            initial_interval = "initial_mock"

            assert get_added_after(fetch_full_feed, initial_interval, last_fetch_time) == last_fetch_time

        def test_get_initial_interval__fetch_full_feed_true(self):
            """
            Scenario: Full feed and last fetch is set

            Given:
            - fetch_full_feed is true
            - initial interval is set
            - last fetch time is set

            When:
            - calling get_added_after

            Then:
            - return initial interval
            """
            fetch_full_feed = True
            last_fetch_time = "last_fetch_mock"
            initial_interval = "initial_mock"

            assert get_added_after(fetch_full_feed, initial_interval, last_fetch_time) == initial_interval

        def test_get_initial_interval__fetch_full_feed_false(self):
            """
            Scenario: Incremental feed and last fetch is not set

            Given:
            - fetch_full_feed is true
            - initial interval is set
            - last fetch time is not set

            When:
            - calling get_added_after

            Then:
            - return initial interval
            """
            fetch_full_feed = False
            last_fetch_time = None
            initial_interval = "initial_mock"

            assert get_added_after(fetch_full_feed, initial_interval, last_fetch_time) == initial_interval

    @pytest.mark.parametrize(
        "indicators, last_run, new_indicators",
        [
            (
                [
                    {"value": "one", "type": "IP", "rawJSON": {"id": "test_one", "modified": "2023-06-14T13:18:21.598591Z"}},
                    {"value": "two", "type": "IP", "rawJSON": {"id": "test_two", "modified": "2023-07-06T08:59:57.339606Z"}},
                    {
                        "value": "three",
                        "type": "Domain",
                        "rawJSON": {"id": "test_three", "modified": "2023-07-06T08:59:57.339606Z"},
                    },
                    {"value": "$$DummyIndicator$$", "relationships": [{}, {}, {}]},
                ],
                {},
                [
                    {"value": "one", "type": "IP", "rawJSON": {"id": "test_one", "modified": "2023-06-14T13:18:21.598591Z"}},
                    {"value": "two", "type": "IP", "rawJSON": {"id": "test_two", "modified": "2023-07-06T08:59:57.339606Z"}},
                    {
                        "value": "three",
                        "type": "Domain",
                        "rawJSON": {"id": "test_three", "modified": "2023-07-06T08:59:57.339606Z"},
                    },
                    {"value": "$$DummyIndicator$$", "relationships": [{}, {}, {}]},
                ],
            ),
            (
                [
                    {"value": "two", "type": "IP", "rawJSON": {"id": "test_two", "modified": "2023-07-06T08:59:57.339606Z"}},
                    {
                        "value": "three",
                        "type": "Domain",
                        "rawJSON": {"id": "test_three", "modified": "2023-07-06T08:59:57.339606Z"},
                    },
                ],
                {
                    "latest_indicators": [
                        {"test_one": "2023-06-14T13:18:21.598591Z"},
                        {"test_two": "2023-07-06T08:59:57.339606Z"},
                        {"test_three": "2023-07-06T08:59:57.339606Z"},
                    ]
                },
                [],
            ),
            (
                [
                    {"value": "two", "type": "IP", "rawJSON": {"id": "test_two", "modified": "2023-10-02T05:34:45.339145Z"}},
                    {
                        "value": "three",
                        "type": "Domain",
                        "rawJSON": {"id": "test_three", "modified": "2023-07-06T08:59:57.339606Z"},
                    },
                    {
                        "value": "four",
                        "type": "Domain",
                        "rawJSON": {"id": "test_four", "modified": "2023-10-02T05:34:28.339145Z"},
                    },
                    {"value": "$$DummyIndicator$$", "relationships": [{}, {}, {}]},
                ],
                {
                    "latest_indicators": [
                        {"test_two": "2023-07-06T08:59:57.339606Z"},
                        {"test_three": "2023-07-06T08:59:57.339606Z"},
                        {"value": "$$DummyIndicator$$", "relationships": [{}, {}, {}]},
                    ]
                },
                [
                    {"value": "two", "type": "IP", "rawJSON": {"id": "test_two", "modified": "2023-10-02T05:34:45.339145Z"}},
                    {
                        "value": "four",
                        "type": "Domain",
                        "rawJSON": {"id": "test_four", "modified": "2023-10-02T05:34:28.339145Z"},
                    },
                    {"value": "$$DummyIndicator$$", "relationships": [{}, {}, {}]},
                ],
            ),
        ],
    )
    def test_filter_previously_fetched_indicators(self, indicators, last_run, new_indicators):
        """
        Scenario: Test filtering indicators received from the fetch call before sending indicators to server

        Given:
        - list of indicators returned from the fetch call with empty lastrun object.
        - list of indicators returned from the fetch call with lastrun containing the same indicators (they were fetched
         in the previous fetch call).
        - list of indicators returned from the fetch call, and lastrun containing an indicator that was
         modified in the fetch call.

        When:
        - running filter_indicators command.

        Then:
        - all fetched indicators are returned, lastrun is updated with all fetched indicators ids and modified date.
        - non of the fetched indicators returned, lastrun is updated with all fetched indicators ids and modified date.
        - only the new and modified indicators are returned, lastrun is updated with all
         fetched indicators ids and modified dates.
        """

        from FeedTAXII2 import filter_previously_fetched_indicators

        next_latest_indicators = [
            {obj.get("rawJSON", {}).get("id"): obj.get("rawJSON", {}).get("modified")}
            if obj.get("value") != "$$DummyIndicator$$"
            else obj
            for obj in indicators
        ]

        result = filter_previously_fetched_indicators(indicators, last_run)

        assert result == new_indicators
        assert last_run.get("latest_indicators") == next_latest_indicators


@pytest.mark.parametrize(
    "url, expected_result",
    [
        ("https://test.com/taxii/", True),
        ("https://test.com/taxii2/", True),
        ("https://test.com/test/", False),
        ("domain.com/taxii2/bad_path/", False),
    ],
)
def test_is_valid_url(url, expected_result):
    """
    Scenario: Test validation of the is_valid_url method.

    Given:
    - A url

    When:
    - running is_valid_url command.

    Then:
    - Ensure the function evaluates correctly the correctness of the URL.
    """
    from FeedTAXII2 import is_valid_taxii_url

    assert is_valid_taxii_url(url) == expected_result


def test_feed_main_enrichment_excluded(mocker):
    """
    Given: params with tlp_color set to RED and enrichmentExcluded set to False
    When: Calling feed_main
    Then: validate enrichment_excluded is set to True
    """
    from FeedTAXII2 import main

    params = {"tlp_color": "RED", "enrichmentExcluded": False, "server_url": "test.test.com"}

    client_mocker = mocker.patch("FeedTAXII2.Taxii2FeedClient")

    mocker.patch("FeedTAXII2.is_xsiam_or_xsoar_saas", return_value=True)
    mocker.patch("FeedTAXII2.assert_incremental_feed_params")
    mocker.patch("FeedTAXII2.fetch_indicators_command", return_value=([], []))
    mocker.patch.object(demisto, "params", return_value=params)
    mocker.patch.object(demisto, "command", return_value="fetch-indicators")

    # Call the function under test
    main()

    # Assertion - verify that enrichment_excluded is set to True
    assert client_mocker.call_args.kwargs.get("enrichment_excluded") is True


def test_fetch_indicators_error_handling(mocker):
    """
    Test the error handling of the fetch_indicators_command function.

    Given:
    - A Taxii2FeedClient instance configured with two collections: one that throws an exception and another that passes.
    - A mocked version of the Taxii2FeedClient.build_iterator method to simulate an error when accessing the first collection and to return dummy indicators for any other cases.

    When:
    - The fetch_indicators_command function is called with the mocked client, and an exception is simulated for the first collection.

    Then:
    - Ensure that the demisto.updateModuleHealth method is called with the correct parameters indicating an error.
    - Ensure that the demisto.error method is called with the correct error message.
    - Assert that the number of indicators returned is as expected (2 indicators from the scenario that does not throw an exception).
    """  # noqa: E501
    mock_client = Taxii2FeedClient(url="", collection_to_fetch="default", proxies=[], verify=False, objects_to_fetch=[])
    mock_client.collections = [MockCollection("1", "trow exception"), MockCollection("2", "pass")]

    mock_update_health = mocker.patch.object(demisto, "updateModuleHealth")
    mock_error = mocker.patch("FeedTAXII2.demisto.error")

    # Error simulation on the first collection
    def side_effect(limit, added_after):
        if mock_client.collection_to_fetch.id == "1":
            raise Exception("Simulated Error")
        return [{"value": "dummy_indicator_1"}, {"value": "dummy_indicator_2"}]

    mock_client.collection_to_fetch = None
    mocker.patch("FeedTAXII2.Taxii2FeedClient.build_iterator", side_effect=side_effect)
    try:
        indicators, last_run = fetch_indicators_command(mock_client, "1 day", -1, {})
    except Exception:
        pass

    mock_update_health.assert_called_once_with({"message": "Error fetching collection 1: Simulated Error"}, is_error=True)
    mock_error.assert_called_once_with("Failed to fetch IOCs from collection 1: Simulated Error")
    assert len(indicators) == 2