Github Feed
This is the Feed GitHub integration for getting started with your feed integration.
Data Enrichment & Threat Intelligence · GitHub Feed · Feed
Details
| ID | Github Feed |
|---|---|
| Provider | Microsoft |
| Category | Data Enrichment & Threat Intelligence |
| From Version | 6.8.0 |
| Docker Image | demisto/taxii2:1.0.0.11040999 |
| Supported Modules | Agentix XSIAM |
README
This is the Feed GitHub integration for getting started with your feed integration.
This integration was integrated and tested with version 1.0.0 of Github Feed.
Configure Github Feed in Cortex
| Parameter | Description | Required |
|---|---|---|
| Fetch indicators | False | |
| Base URL | The URL to the GitHub API. | True |
| API Token | False | |
| Incremental Feed | When enabled (default), only new/modified files since the last fetch are reported (legacy behavior). Disable to fetch the full current set on every cycle so the “When removed from the feed” (suddenDeath) expiration policy can expire indicators removed from the source. | False |
| Trust any certificate (not secure) | False | |
| Owner | Username of the repository owner | True |
| Repository / Path to fetch | The name of the repository. | True |
| Feed type | Predefined list of indicator types: - YARA: Parses YARA rules from the feed. The “Yara” pack is required for this type. - STIX: Parses STIX data from the feed. - IOCs: Parses Indicators of Compromise (IOCs) using regex patterns. |
True |
| Branch name | The name of the main branch to which to compare. | True |
| Files extensions to fetch | The extension of the file names to target. | True |
| 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 |
| First fetch time | First commit date of first published indicators to bring. e.g., “1 min ago”,”2 weeks ago”,”3 months ago”. | 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 |
| Use system proxy settings | False | |
| Feed Expiration Policy | Controls when indicators expire. Choose “suddenDeath” together with disabling the *Incremental Feed* option to expire indicators that were removed from the source repository. | False |
| Feed Expiration Interval | Only used when *Feed Expiration Policy* is set to “interval”. Ignored for the “suddenDeath”, “never” and “indicatorType” policies. | False |
| Tags | Supports CSV values. | False |
| Enrichment Excluded | Select this option to exclude the fetched indicators from the enrichment process. | False |
| Indicator Reputation | Indicators from this integration instance will be marked with this reputation |
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.
github-get-indicators
Retrieves indicators from the feed within a specified date range and up to a maximum limit.
Base Command
github-get-indicators
Input
| Argument Name | Description | Required |
|---|---|---|
| since | The start date from which to fetch indicators. Accepts date strings like “7 days ago”, “2 weeks ago”, etc. Default is 7 days. | Optional |
| until | The end date until which to fetch indicators. Accepts date strings like “now”, “2023-05-19”, etc. | Optional |
| limit | The maximum number of results to return. Default is 50. | Optional |
Context Output
There is no context output for this command.
Troubleshooting
- HTTP 404 Error:
No common ancestor between {commit} and {branch}- Typically occurs when a “force push” or a “commit squash” overwrites previously pushed commits on the configured branch.
- Resolution: To resolve this issue, reset indicator fetching of the affected integration instance.
- Prevention: To avoid this error, refrain from force pushing or squashing commits on the branch configured in the integration instance.
Configuration parameters
feed— Fetch indicatorsurl— Base URL (required)api_token—feedIncremental— Incremental Feedinsecure— Trust any certificate (not secure)owner— Owner (required)repo— Repository / Path to fetch (required)feedType— Feed type (required)branch_head— Branch name (required)extensions_to_fetch— Files extensions to fetch (required)feedReliability— Source Reliability (required)tlp_color— Traffic Light Protocol Colorfetch_since— First fetch timefeedFetchInterval— Feed Fetch IntervalfeedBypassExclusionList— Bypass exclusion listproxy— Use system proxy settingsfeedExpirationPolicy—feedExpirationInterval— Feed Expiration IntervalfeedTags— TagsenrichmentExcluded— Enrichment ExcludedfeedReputation— Indicator Reputation
Commands (1)
-
github-get-indicatorsRetrieves indicators from the feed within a specified date range and up to a maximum limit.
import base64 import json import demistomock as demisto import pytest from freezegun import freeze_time def util_load_json(path): with open(path, encoding="utf-8") as f: return json.loads(f.read()) def util_load_txt(path): with open(path, encoding="utf-8") as f: return f.read() def mock_client(): """ Create a mock client for testing. """ from FeedGitHub import Client return Client( base_url="example.com", verify=False, proxy=False, owner="", repo="", headers={}, ) def test_get_content_files_from_repo(mocker): """ Given: - A list of relevant files to fetch content from. - Parameters specifying the feed type and extensions to fetch. - A mock response for the content files from the repository. When: - Calling get_content_files_from_repo to fetch the content of the relevant files. Then: - Assert one HTTP request is made to get the contents of the file with the relevant extension. - Assert the returned content of the relevant files matches the expected results. """ from FeedGitHub import get_content_files_from_repo client = mock_client() params = {"feedType": "IOCs", "extensions_to_fetch": ["txt"]} relevant_files = util_load_json("test_data/relevant-files.json") return_data = { "content": base64.b64encode( b"2023-02-08 (WEDNESDAY) - COBALT STRIKE FROM ICEDID (BOKBOT) INFECTION\n\n" b"REFERENCE:\n\n" b"- https://twitter.com/Unit42_Intel/status/1623707361184477185\n\n" b"NOTES:\n\n" b"- IcedID infection generated using a OneNote file reported earlier today\n\n" b"ICEDID TRAFFIC:\n\n" b"- 80.66.88[.]143 port 80 - ehonlionetodo[.]com\n" b"- GET /\n" b"COBALT STRIKE TRAFFIC:\n\n" b"- 167.172.154[.]189 port 80 - GET /36.ps1\n" ).decode("utf-8") } mock_http_request = mocker.patch.object(client, "_http_request", return_value=return_data) content_files = get_content_files_from_repo(client, relevant_files, params) assert mock_http_request.call_count == 1 # One .txt file in all relevant committed files assert mock_http_request.call_args.kwargs == {"method": "GET", "full_url": relevant_files[1]["contents_url"]} assert content_files == util_load_json("test_data/get_content-files-from-repo-result.json") def test_get_commit_files(mocker): """ Given: - A base commit SHA, a head commit SHA, and a flag indicating if it is the first fetch. - A mock response for the list of all commit files between the base and head commits. When: - Calling get_commits_files to retrieve the relevant files and the current repo head SHA. Then: - Returns the list of relevant files and the current repo head SHA matching the expected results. """ from FeedGitHub import get_commits_files client = mock_client() base = "ad3e0503765479e9ee09bac5dee726eb918b9ebd" head = "9a611449423b9992c126c20e47c5de4f58fc1c0e" is_first_fetch = True all_commits_files = util_load_json("test_data/all-commit-files-res.json") current_repo_head_sha = "ad3e0503765479e9ee09bac5dee726eb918b9ebd" mocker.patch.object( client, "get_files_between_commits", return_value=(all_commits_files, current_repo_head_sha), ) relevant_files, current_repo_head_sha = get_commits_files(client, base, head, is_first_fetch) assert relevant_files == util_load_json("test_data/relevant-files.json") def test_filter_out_files_by_status(): """ Given: - A list of dictionaries representing commit files, each containing a status and a filename. When: - Filtering out files by their status using the filter_out_files_by_status function. Then: - Returns a list of URLs for files that are added or modified. """ from FeedGitHub import filter_out_files_by_status commits_files = [ {"status": "added", "filename": "http://example.com/file1"}, {"status": "modified", "filename": "http://example.com/file2"}, {"status": "removed", "filename": "http://example.com/file3"}, {"status": "renamed", "filename": "http://example.com/file4"}, {"status": "added", "filename": "http://example.com/file5"}, ] expected_output = [ {"status": "added", "filename": "http://example.com/file1"}, {"status": "modified", "filename": "http://example.com/file2"}, {"status": "added", "filename": "http://example.com/file5"}, ] actual_output = filter_out_files_by_status(commits_files) assert actual_output == expected_output, f"Expected {expected_output}, but got {actual_output}" @freeze_time("2024-05-12T15:30:49.330015") def test_parse_and_map_yara_content(mocker): """ Given: - YARA rule files as input from different sources. rule-1 = classic yara rule rule-2 = broken yara rule rule-3 = yara rule has a unique structure that contains curly brackets inside the rule strings field list_rules_input = Several different rules from a single file When: - Parsing and mapping YARA content using the parse_and_map_yara_content function. Then: - Returns the parsed YARA rules in JSON format matching the expected results. """ from FeedGitHub import parse_and_map_yara_content mocker.patch.object(demisto, "error") rule_1_input = {"example.com": util_load_txt("test_data/yara-rule-1.yar")} rule_2_input = {"example.com": util_load_txt("test_data/yara-rule-2.yar")} rule_3_input = {"example.com": util_load_txt("test_data/yara-rule-3.yar")} list_rules_input = {"example.com": util_load_txt("test_data/test-split-yara-1.yar")} parsed_rule1 = parse_and_map_yara_content(rule_1_input) parsed_rule2 = parse_and_map_yara_content(rule_2_input) parsed_rule3 = parse_and_map_yara_content(rule_3_input) list_parsed_rules = parse_and_map_yara_content(list_rules_input) assert parsed_rule1 == util_load_json("test_data/yara-rule-1-res.json") assert parsed_rule2 == util_load_json("test_data/yara-rule-2-res.json") assert parsed_rule3 == util_load_json("test_data/yara-rule-3-res.json") assert list_parsed_rules == util_load_json("test_data/list-parsed-rules-res.json") def test_parse_and_map_yara_content_invalid_rule(mocker): """ Given: - An invalid YARA rule file as input. When: - Parsing and mapping the YARA content using the parse_and_map_yara_content function. Then: - Ensure that an empty list is returned and an error is logged. """ from FeedGitHub import parse_and_map_yara_content demisto_error_mock = mocker.patch.object(demisto, "error") yara_rule_file = {"invalid-yara-rule.yar": "invalid yara rule"} parsed_rules = parse_and_map_yara_content(yara_rule_file) assert parsed_rules == [] assert demisto_error_mock.call_args[0][0] == ( "File: 'invalid-yara-rule.yar' cannot be processed. Error Message: Unknown text invalid for token of type ID on line 1" ) @freeze_time("2024-05-12T15:30:49.330015") def test_extract_text_indicators(): """ Given: - A dictionary containing file paths and their respective contents with IOC indicators. - Parameters specifying the repository owner and name. When: - Calling extract_text_indicators to extract IOC indicators from the file contents. Then: - Returns the extracted IOC indicators matching the expected results. """ from FeedGitHub import extract_text_indicators ioc_indicators_input = {"example.com": util_load_txt("test_data/test-ioc-indicators.txt")} params = {"owner": "example.owner", "repo": "example.repo"} res_indicators = extract_text_indicators(ioc_indicators_input, params) assert res_indicators == util_load_json("test_data/iocs-res.json") @pytest.mark.parametrize( "cidr, expected_value", [ # Two-digit prefix lengths must not be truncated to one digit ("1.1.1.1/12", "1.1.1.1/12"), ("192.168.0.0/16", "192.168.0.0/16"), ("10.0.0.0/24", "10.0.0.0/24"), ("172.16.0.0/32", "172.16.0.0/32"), # Single-digit prefix lengths must still match ("10.0.0.0/8", "10.0.0.0/8"), ("1.1.1.1/1", "1.1.1.1/1"), ("1.1.1.1/2", "1.1.1.1/2"), ], ) def test_ipv4_cidr_regex_full_prefix_length(cidr: str, expected_value: str): """ Given: - An IPv4 CIDR string with a multi-digit prefix length (e.g. /12, /16, /24, /32). When: - The ipv4cidrRegex is applied via re.search. Then: - The full CIDR value including the complete prefix length is matched, not a truncated single-digit prefix. """ import re from FeedGitHub import ipv4cidrRegex match = re.search(ipv4cidrRegex, cidr) assert match is not None, f"Expected a match for {cidr!r}" assert match.group(0) == expected_value, f"For input {cidr!r}: expected {expected_value!r} but got {match.group(0)!r}" @pytest.mark.parametrize( "cidr, expected_value", [ # Three-digit prefix lengths must not be truncated ("2001:db8::/128", "2001:db8::/128"), ("2001:db8::/112", "2001:db8::/112"), ("::ffff:192.168.1.0/120", "::ffff:192.168.1.0/120"), # Two-digit prefix lengths must not be truncated to one digit ("2001:db8::/64", "2001:db8::/64"), ("2001:db8::/48", "2001:db8::/48"), ("2001:db8::/32", "2001:db8::/32"), ("fe80::/10", "fe80::/10"), # Single-digit prefix lengths must still match ("2001:db8::/8", "2001:db8::/8"), ("2001:db8::/1", "2001:db8::/1"), ("::/0", "::/0"), # Loopback ("::1/128", "::1/128"), # Full address ("2001:0db8:85a3::8a2e:0370:7334/64", "2001:0db8:85a3::8a2e:0370:7334/64"), # Invalid: literal 'd' instead of digit -- must NOT match ("::ffff:1dd.1dd.1dd.1dd/128", None), # Invalid: non-dot separator -- must NOT match ("::ffff:1.1.1_1/128", None), ], ) def test_ipv6_cidr_regex_full_prefix_length(cidr: str, expected_value: str): """ Given: - An IPv6 CIDR string with various prefix lengths and formats. When: - The ipv6cidrRegex is applied via re.search. Then: - Valid CIDRs are matched with the complete prefix length. - Malformed strings with literal 'd' digits or non-dot separators are not matched. """ import re from FeedGitHub import ipv6cidrRegex match = re.search(ipv6cidrRegex, cidr) got = match.group(0) if match else None assert got == expected_value, f"For input {cidr!r}: expected {expected_value!r} but got {got!r}" def test_get_stix_indicators(): """ Given: - Output of the STIX feed API When: - When calling the 'get_stix_indicators' method Then: - Returns a list of the STIX indicators parsed from "STIX2XSOARParser client" """ from FeedGitHub import get_stix_indicators with open("test_data/taxii_test.json", encoding="utf-8") as f: file_name_contents = [{"taxii_test.json": f.read()}] res_indicators = get_stix_indicators(file_name_contents) assert res_indicators == util_load_json("test_data/taxii_test_res.json") def test_negative_limit(mocker): """ Given: - A negative limit. When: - Calling get_indicators. Then: - Ensure ValueError is raised with the right message. """ mocker.patch.object(demisto, "error") from FeedGitHub import get_indicators_command args = {"limit": "-1"} client = mock_client() with pytest.raises(ValueError) as ve: get_indicators_command(client, {}, args) assert ve.value.args[0] == "get_indicators_command return with error. \n\nError massage: Limit must be a positive number." def test_fetch_indicators(mocker): """ Given: - A mock client and parameters specifying the fetch time frame. - Mocked responses for base and head commit SHAs, and indicators. When: - Calling fetch_indicators to retrieve indicators from the GitHub feed. Then: - Returns the list of indicators matching the expected results. """ import FeedGitHub client = mock_client() mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "setLastRun") params = {"fetch_since": "15 days ago"} mocker.patch.object( client, "get_commits_between_dates", return_value="046a799ebe004e1bff686d6b774387b3bdb3d1ce", ) mocker.patch.object( FeedGitHub, "get_indicators", return_value=( util_load_json("test_data/iterator-test.json"), "9a611449423b9992c126c20e47c5de4f58fc1c0e", ), ) results = FeedGitHub.fetch_indicators(client, None, params) assert results == util_load_json("test_data/fetch-indicators-res.json") def test_fetch_indicators_enrichment_excluded(mocker): """ Given: - A mock client and parameters specifying the fetch time frame. - Mocked responses for base and head commit SHAs, and indicators. - Enrichment excluded marked as true When: - Calling fetch_indicators to retrieve indicators from the GitHub feed. Then: - Returns the list of indicators matching the expected results. - All returned indicators have 'enrichmentExcluded' set to True """ import FeedGitHub client = mock_client() mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "setLastRun") params = {"fetch_since": "15 days ago", "enrichmentExcluded": True} mocker.patch.object( client, "get_commits_between_dates", return_value="046a799ebe004e1bff686d6b774387b3bdb3d1ce", ) mocker.patch.object( FeedGitHub, "get_indicators", return_value=( util_load_json("test_data/iterator-test.json"), "9a611449423b9992c126c20e47c5de4f58fc1c0e", ), ) results = FeedGitHub.fetch_indicators_command(client, params, {}) expected: list = util_load_json("test_data/fetch-indicators-res.json") for ind in expected: ind["enrichmentExcluded"] = True assert results == expected @freeze_time("2024-05-20T11:05:36.984413") def test_get_indicators_command(mocker): """ Given: - A mock client and parameters to retrieve indicators from the GitHub feed. - Mocked responses for base and head commit SHAs, and indicators. When: - Calling get_indicators_command to retrieve and format indicators. Then: - Returns the human-readable output matching the expected results. """ import FeedGitHub from CommonServerPython import tableToMarkdown client = mock_client() mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "error") mocker.patch.object( client, "get_commits_between_dates", return_value=[ "9a611449423b9992c126c20e47c5de4f58fc1c0e", "aabaf42225cb4d18e338bc5c8c934f25be814704", "046a799ebe004e1bff686d6b774387b3bdb3d1ce", ], ) mocker.patch.object( FeedGitHub, "get_indicators", return_value=(util_load_json("test_data/iterator-test.json"), None), ) results = FeedGitHub.get_indicators_command(client, params={}, args={"limit": ""}) hr_indicators = util_load_json("test_data/hr-indicators.json") human_readable = tableToMarkdown( "Indicators from GitHubFeed:", hr_indicators, headers=["Type", "Value"], removeNull=True, ) assert results.readable_output == human_readable def test_extract_commits(mocker): """ Given: - A mock response object with commit data. - Mocked responses for paginated commit data. When: - Calling _extract_commits to retrieve and aggregate commit information. Then: - Returns a list of all commits from the paginated responses. """ client = mock_client() mocker.patch.object(demisto, "debug") mock_response_single = mocker.MagicMock() mock_response_single.links = {} mock_response_single.json.return_value = [{"sha": "commit1"}, {"sha": "commit2"}] mock_response_page1 = mocker.MagicMock() mock_response_page2 = mocker.MagicMock() mock_response_page1.links = {"next": {"url": "http://example.com/page2"}} mock_response_page1.json.return_value = [{"sha": "commit1"}, {"sha": "commit2"}] mock_response_page2.links = {} mock_response_page2.json.return_value = [{"sha": "commit3"}, {"sha": "commit4"}] mocker.patch.object(client, "_http_request", side_effect=[mock_response_page2]) commits_single = client._extract_commits(mock_response_single) assert commits_single == [{"sha": "commit1"}, {"sha": "commit2"}] commits_multiple = client._extract_commits(mock_response_page1) expected_commits = [ {"sha": "commit1"}, {"sha": "commit2"}, {"sha": "commit3"}, {"sha": "commit4"}, ] assert commits_multiple == expected_commits @freeze_time("2024-05-20T11:05:36.984413") def test_arrange_iocs_indicator_to_xsoar(): """ Given: - A file path, a list of parsed indicators, and additional parameters. When: - Calling arrange_iocs_indicator_to_xsoar to format the indicators. Then: - Returns a list of formatted indicators with expected fields and values. """ from FeedGitHub import arrange_iocs_indicator_to_xsoar file_path = "test_file.txt" parsed_indicators = [ {"value": "example.com", "type": "Domain"}, {"value": "123.456.789.0", "type": "IP"}, ] params = {"owner": "example_owner", "repo": "example_repo"} expected_result = [ { "value": "example.com", "type": "Domain", "service": "github", "fields": { "references": "test_file.txt", "tags": {"owner": "example_owner", "repo": "example_repo"}, "firstseenbysource": "2024-05-20T11:05:36.984413", }, "rawJSON": {"value": "example.com", "type": "Domain"}, }, { "value": "123.456.789.0", "type": "IP", "service": "github", "fields": { "references": "test_file.txt", "tags": {"owner": "example_owner", "repo": "example_repo"}, "firstseenbysource": "2024-05-20T11:05:36.984413", }, "rawJSON": {"value": "123.456.789.0", "type": "IP"}, }, ] result = arrange_iocs_indicator_to_xsoar(file_path, parsed_indicators, params) assert result == expected_result def test_identify_json_structure(): """ Given: - A dictionary containing JSON data with different structures. When: - Calling identify_json_structure to identify the structure. Then: - Returns the identified structure based on the provided JSON data. """ from FeedGitHub import identify_json_structure json_data_bundle = {"bundle": {"type": "bundle", "id": "bundle--12345678-1234-5678-1234-567812345678"}} assert identify_json_structure(json_data_bundle) == "Bundle" json_data_envelope = {"objects": [{"type": "indicator", "id": "indicator--12345678-1234-5678-1234-567812345678"}]} assert identify_json_structure(json_data_envelope) == "Envelope" json_data_envelope_alt = {"type": "indicator", "id": "indicator--12345678-1234-5678-1234-567812345678"} assert identify_json_structure(json_data_envelope_alt) == "Envelope" json_data_list = [{"type": "indicator", "id": "indicator--12345678-1234-5678-1234-567812345678"}] assert identify_json_structure(json_data_list) == {"objects": json_data_list} json_data_unknown = {"unknown_key": "unknown_value"} assert identify_json_structure(json_data_unknown) is None def test_filtering_stix_files(): """ Given: - A list of content files containing both STIX and non-STIX files. When: - Calling filtering_stix_files to filter out only the STIX files. Then: - Returns a list containing only the STIX files from the input list. """ from FeedGitHub import filtering_stix_files file_names = ["fileA.json", "fileB.json", "fileC.json"] file_contents = [ '{"type": "indicator", "id": "indicator--12345678-1234-5678-1234-567812345678"}', # STIX format '{"bundle": {"type": "bundle", "id": "bundle--12345678-1234-5678-1234-567812345678"}}', # STIX format '{"type": "non-stix", "id": "non-stix--12345678-1234-5678-1234-567812345678"}', # Non-STIX format ] expected_result = [ {"type": "indicator", "id": "indicator--12345678-1234-5678-1234-567812345678"}, {"bundle": {"type": "bundle", "id": "bundle--12345678-1234-5678-1234-567812345678"}}, {"type": "non-stix", "id": "non-stix--12345678-1234-5678-1234-567812345678"}, ] assert filtering_stix_files(file_names=file_names, file_contents=file_contents) == expected_result def test_fetch_indicators_command_with_tlp_color_red(mocker): """ Given: params with tlp_color set to RED and enrichmentExcluded set to False. When: Calling fetch_indicators_command with the provided parameters. Then: Verify that the fetch_indicators function is called with the expected parameters. """ from FeedGitHub import fetch_indicators_command client_mock = mock_client() params = {"feedTags": "tag1,tag2", "tlp_color": "RED", "enrichmentExcluded": False, "limit": "50"} args = {} mocker.patch("FeedGitHub.is_xsiam_or_xsoar_saas", return_value=True) mocker.patch.object(demisto, "params", return_value=params) fetch_indicators_mock = mocker.patch("FeedGitHub.fetch_indicators") # Call the function under test fetch_indicators_command(client_mock, params, args) # Assertion - verify the output assert fetch_indicators_mock.call_args.kwargs.get("enrichment_excluded") is True def test_get_indicators_command_with_tlp_color_red(mocker): from FeedGitHub import get_indicators_command client_mock = mock_client() params = {"feedTags": "tag1,tag2", "tlp_color": "RED", "enrichmentExcluded": False, "limit": "50"} args = {} mocker.patch("FeedGitHub.Client.get_commits_between_dates", return_value=["test_hash"]) mocker.patch("FeedGitHub.is_xsiam_or_xsoar_saas", return_value=True) mocker.patch.object(demisto, "params", return_value=params) mocker.patch("FeedGitHub.get_indicators", return_value=([{"name": "test_ind"}], None)) command_res = get_indicators_command(client_mock, params, args) assert command_res.outputs[0].get("enrichmentExcluded") is True def test_fetch_indicators_non_incremental_uses_first_fetch_each_cycle(mocker): """ Given: - feedIncremental disabled. - A previously stored last_commit (simulating not the first cycle). When: - Calling fetch_indicators. Then: - The base commit SHA is derived from `fetch_since`, NOT from the stored last_commit. - get_commits_between_dates is called with the configured fetch_since window. """ import FeedGitHub client = mock_client() mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "setLastRun") params = {"fetch_since": "15 days ago", "feedIncremental": "false"} commits_mock = mocker.patch.object( client, "get_commits_between_dates", return_value=[ "head_sha", "mid_sha", "fresh_base_sha_from_fetch_since", ], ) get_indicators_mock = mocker.patch.object( FeedGitHub, "get_indicators", return_value=(util_load_json("test_data/iterator-test.json"), "new_head_sha"), ) FeedGitHub.fetch_indicators(client, "stored_last_commit_must_be_ignored", params) commits_mock.assert_called_once_with("15 days ago", "now") base_sha_arg = get_indicators_mock.call_args.args[2] is_first_fetch_arg = get_indicators_mock.call_args.args[4] assert base_sha_arg == "fresh_base_sha_from_fetch_since" assert is_first_fetch_arg is True def test_fetch_indicators_incremental_uses_last_commit(mocker): """ Given: - feedIncremental enabled. - A previously stored last_commit. When: - Calling fetch_indicators. Then: - The base commit SHA is the stored last_commit. - get_commits_between_dates is NOT called. """ import FeedGitHub client = mock_client() mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "setLastRun") params = {"fetch_since": "15 days ago", "feedIncremental": "true"} commits_mock = mocker.patch.object(client, "get_commits_between_dates") get_indicators_mock = mocker.patch.object( FeedGitHub, "get_indicators", return_value=(util_load_json("test_data/iterator-test.json"), "new_head_sha"), ) FeedGitHub.fetch_indicators(client, "stored_last_commit", params) commits_mock.assert_not_called() base_sha_arg = get_indicators_mock.call_args.args[2] is_first_fetch_arg = get_indicators_mock.call_args.args[4] assert base_sha_arg == "stored_last_commit" assert is_first_fetch_arg is False def test_fetch_indicators_incremental_first_run_falls_back_to_fetch_since(mocker): """ Given: - feedIncremental enabled. - No stored last_commit (first cycle). When: - Calling fetch_indicators. Then: - Base SHA is derived from fetch_since, is_first_fetch=True. """ import FeedGitHub client = mock_client() mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "setLastRun") params = {"fetch_since": "15 days ago", "feedIncremental": "true"} commits_mock = mocker.patch.object( client, "get_commits_between_dates", return_value=["head_sha", "base_sha"], ) get_indicators_mock = mocker.patch.object( FeedGitHub, "get_indicators", return_value=(util_load_json("test_data/iterator-test.json"), "new_head_sha"), ) FeedGitHub.fetch_indicators(client, None, params) commits_mock.assert_called_once_with("15 days ago", "now") base_sha_arg = get_indicators_mock.call_args.args[2] is_first_fetch_arg = get_indicators_mock.call_args.args[4] assert base_sha_arg == "base_sha" assert is_first_fetch_arg is True def test_fetch_indicators_setlastrun_called_in_both_modes(mocker): """ Given: - Either incremental or non-incremental mode. When: - fetch_indicators completes with a valid head commit. Then: - setLastRun is called with the new head commit. """ import FeedGitHub client = mock_client() mocker.patch.object(demisto, "debug") set_last_run_mock = mocker.patch.object(demisto, "setLastRun") mocker.patch.object( client, "get_commits_between_dates", return_value=["head", "base"], ) mocker.patch.object( FeedGitHub, "get_indicators", return_value=(util_load_json("test_data/iterator-test.json"), "new_head_sha"), ) FeedGitHub.fetch_indicators(client, "ignored", {"fetch_since": "15 days ago", "feedIncremental": "false"}) FeedGitHub.fetch_indicators(client, None, {"fetch_since": "15 days ago", "feedIncremental": "true"}) assert set_last_run_mock.call_count == 2 for call in set_last_run_mock.call_args_list: assert call.args[0] == {"last_commit": "new_head_sha"}