from functools import partial import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 from CommonServerUserPython import * """ IMPORTS """ import re import traceback from collections import OrderedDict from collections.abc import Callable from typing import Any import requests import urllib3 from sixgill.sixgill_constants import FeedStream from sixgill.sixgill_feed_client import SixgillFeedClient from sixgill.sixgill_request_classes.sixgill_auth_request import SixgillAuthRequest from sixgill.sixgill_utils import is_indicator # Disable insecure warnings urllib3.disable_warnings() """ GLOBALS/PARAMS """ CHANNEL_CODE = "7457a04d972fceb8e0cc2192ba4abc66" if is_xsiam() else "7698e8287dfde53dcd13082be750a85a" MAX_INDICATORS = 1000 SUSPICIOUS_FEED_IDS = ["darkfeed_003"] DEMISTO_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ" VERIFY = not demisto.params().get("insecure", True) SESSION = requests.Session() DESCRIPTION_FIELD_ORDER = OrderedDict( [ ("Description", "description"), ("Created On", "created"), ("Post Title", "sixgill_posttitle"), ("Threat Actor Name", "sixgill_actor"), ("Source", "sixgill_source"), ("Sixgill Feed ID", "sixgill_feedid"), ("Sixgill Feed Name", "sixgill_feedname"), ("Sixgill Post ID", "sixgill_postid"), ("Sixgill Confidence", "sixgill_confidence"), ("Language", "lang"), ("Indicator ID", "id"), ("External references (e.g. MITRE ATT&CK)", "external_reference"), ] ) HASH_MAPPING = { "hashes.md5": "md5", "hashes.'sha-1'": "sha1", "hashes.'sha-256'": "sha256", "hashes.'sha-512'": "sha512", "hashes.ssdeep": "ssdeep", } """ HELPER FUNCTIONS """ stix_regex_parser = re.compile(r"([\w-]+?):(\w.+?) (?:[!><]?=|IN|MATCHES|LIKE) '(.*?)' *[OR|AND|FOLLOWEDBY]?") class ExternalReferenceSourceTypes: MITRE_ATTACK = "mitre-attack" VIRUS_TOTAL = "VirusTotal" def strip_http(url): return url.split("://")[-1] def url_to_rfc3986(url): if url.startswith(("http://", "https://", "ftp://", "sftp://")): return url else: return f"https://{url}" def clean_url(value): return value.replace("[.]", ".") def run_pipeline(value, pipeline, log): for func in pipeline: log.debug(f"run {func.__name__} function on value: {value}") value = func(value) log.debug(f"post {func.__name__} run value: {value}") log.debug(f"returned value: {value}") return value def to_demisto_score(feed_id: str, revoked: bool): if revoked: return 0 if feed_id in SUSPICIOUS_FEED_IDS: return 2 return 3 def get_description(stix_obj): description_string = "" for name, sixgill_name in DESCRIPTION_FIELD_ORDER.items(): description_string += f"{name}: {stix_obj.get(sixgill_name)}\n" return description_string def extract_external_reference_field(stix2obj, source_name, field_to_extract): for reference in stix2obj.get("external_reference", []): if reference.get("source_name") == source_name: return reference.get(field_to_extract, None) return None def post_id_to_full_url(post_id): return f"https://portal.cybersixgill.com/#/search?q=_id:{post_id}" def to_demisto_indicator(value, indicators_name, stix2obj, tags: list = [], tlp_color: str | None = None): indicator = { "value": value, "type": indicators_name, "rawJSON": stix2obj, "fields": { "actor": stix2obj.get("sixgill_actor"), "tags": list(set(stix2obj.get("labels")).union(set(tags))), "firstseenbysource": stix2obj.get("created"), "description": get_description(stix2obj), "sixgillactor": stix2obj.get("sixgill_actor"), "sixgillfeedname": stix2obj.get("sixgill_feedname"), "sixgillsource": stix2obj.get("sixgill_source"), "sixgilllanguage": stix2obj.get("lang"), "sixgillposttitle": stix2obj.get("sixgill_posttitle"), "sixgillfeedid": stix2obj.get("sixgill_feedid"), "sixgillconfidence": stix2obj.get("sixgill_confidence"), "sixgillpostreference": post_id_to_full_url(stix2obj.get("sixgill_postid", "")), "sixgillindicatorid": stix2obj.get("id"), "sixgilldescription": stix2obj.get("description"), "sixgillvirustotaldetectionrate": extract_external_reference_field( stix2obj, ExternalReferenceSourceTypes.VIRUS_TOTAL, "positive_rate" ), "sixgillvirustotalurl": extract_external_reference_field(stix2obj, ExternalReferenceSourceTypes.VIRUS_TOTAL, "url"), "sixgillmitreattcktactic": extract_external_reference_field( stix2obj, ExternalReferenceSourceTypes.MITRE_ATTACK, "mitre_attack_tactic" ), "sixgillmitreattcktechnique": extract_external_reference_field( stix2obj, ExternalReferenceSourceTypes.MITRE_ATTACK, "mitre_attack_technique" ), }, "score": to_demisto_score(stix2obj.get("sixgill_feedid"), stix2obj.get("revoked", False)), } if stix2obj.get("sixgill_feedname"): indicator["fields"]["tags"].append(stix2obj.get("sixgill_feedname")) if tlp_color: indicator["fields"]["trafficlightprotocol"] = tlp_color mitre_id = extract_external_reference_field(stix2obj, ExternalReferenceSourceTypes.MITRE_ATTACK, "mitre_attack_tactic_id") mitre_url = extract_external_reference_field(stix2obj, ExternalReferenceSourceTypes.MITRE_ATTACK, "mitre_attack_tactic_url") if mitre_id and mitre_url: indicator["fields"]["feedrelatedindicators"] = [{"type": "MITRE ATT&CK", "value": mitre_id, "description": mitre_url}] return indicator def get_limit(str_limit, default_limit): try: return int(str_limit) except Exception: return default_limit def filter_confidence(confidence: int | str, indicator_obj: dict) -> bool: """ Predicate function to filter records based on confidence score. """ if isinstance(confidence, str) and confidence == "all": return True sixgill_confidence = arg_to_number(indicator_obj.get("sixgill_confidence"), "sixgill_confidence", required=False) return sixgill_confidence >= int(confidence) if sixgill_confidence else False def stix2_to_demisto_indicator(stix2obj: dict[str, Any], log, tags: list = [], tlp_color: str | None = None): indicators = [] pattern = stix2obj.get("pattern", "") sixgill_feedid = stix2obj.get("sixgill_feedid", "") hashes: dict[str, Any] = {"md5": None, "sha1": None, "sha256": None, "sha512": None, "ssdeep": None} for match in stix_regex_parser.findall(pattern): try: _, sub_type, value = match demisto_indicator_map = demisto_mapping.get(sixgill_feedid) if demisto_indicator_map: indicators_name = demisto_indicator_map.get("name") value = run_pipeline(value, demisto_indicator_map.get("pipeline", []), log) demisto_indicator = to_demisto_indicator(value, indicators_name, stix2obj, tags, tlp_color) if ( demisto_indicator.get("type") == FeedIndicatorType.File and HASH_MAPPING.get(sub_type.lower()) in hashes and HASH_MAPPING.get(sub_type.lower()) ): hashes[HASH_MAPPING[sub_type.lower()]] = value indicators.append(demisto_indicator) except Exception as e: log.error(f"failed converting STIX object to Demisto indicator: {e}, STIX object: {stix2obj}") continue if len(indicators) > 0 and all(ioc.get("type") == FeedIndicatorType.File for ioc in indicators): temp_indicator = indicators[0].copy() if hashes["sha256"] is not None: temp_indicator["value"] = hashes["sha256"] temp_indicator["fields"].update({hash_k: hash_v for hash_k, hash_v in hashes.items() if hash_v is not None}) indicators = [temp_indicator] return indicators demisto_mapping: dict[str, dict[str, Any]] = { "darkfeed_001": {"name": FeedIndicatorType.Domain, "pipeline": [strip_http, clean_url]}, "darkfeed_002": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_003": {"name": FeedIndicatorType.Domain, "pipeline": [strip_http, clean_url]}, "darkfeed_004": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_005": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_006": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_007": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_008": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_009": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_010": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_011": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_012": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_013": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_014": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_015": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_018": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_019": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_020": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_021": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_022": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_023": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_024": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_025": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_026": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_027": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_028": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_029": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_030": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_031": {"name": FeedIndicatorType.Domain, "pipeline": [strip_http, clean_url]}, "darkfeed_032": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_033": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_034": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_035": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_036": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_037": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_038": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_039": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_040": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_041": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_042": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_043": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_044": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_045": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_046": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_047": {"name": FeedIndicatorType.Domain, "pipeline": [strip_http, clean_url]}, "darkfeed_048": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_049": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_050": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_051": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_052": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_053": {"name": FeedIndicatorType.File, "pipeline": []}, "darkfeed_054": {"name": FeedIndicatorType.URL, "pipeline": [url_to_rfc3986, clean_url]}, "darkfeed_055": {"name": FeedIndicatorType.IP, "pipeline": []}, "darkfeed_056": {"name": FeedIndicatorType.IP, "pipeline": []}, } """ COMMANDS + REQUESTS FUNCTIONS """ def test_module_command(*args): """ Performs basic Auth request """ response = SESSION.send( request=SixgillAuthRequest(demisto.params()["client_id"], demisto.params()["client_secret"], CHANNEL_CODE).prepare(), verify=VERIFY, ) if not response.ok: raise Exception("Auth request failed - please verify client_id, and client_secret.") return "ok", None, "ok" def get_indicators_command(client: SixgillFeedClient, args): limit = int(args.get("limit")) indicators = fetch_indicators_command(client, limit, True) human_readable = tableToMarkdown( "Indicators from Sixgill Dark Feed:", indicators, headers=["value", "type", "rawJSON", "score"] ) return human_readable, {}, indicators def fetch_indicators_command( client: SixgillFeedClient, limit: int = 0, get_indicators_mode: bool = False, tags: list = [], tlp_color: str | None = None, confidence: int | None = None, ): bundle = client.get_bundle() indicators_to_create: list = [] indicator_values_set: set = set() confidence = confidence or "all" for stix_indicator in filter(partial(filter_confidence, confidence), bundle.get("objects")): if is_indicator(stix_indicator): demisto_indicators = stix2_to_demisto_indicator(stix_indicator, demisto, tags, tlp_color) for indicator in demisto_indicators: if indicator.get("value") not in indicator_values_set: indicator_values_set.add(indicator.get("value")) indicators_to_create.append(indicator) if get_indicators_mode and len(indicators_to_create) == limit: break if not get_indicators_mode: client.commit_indicators() return indicators_to_create """ COMMANDS MANAGER / SWITCH PANEL """ def main(): max_indicators = get_limit(demisto.params().get("maxIndicators", MAX_INDICATORS), MAX_INDICATORS) SESSION.proxies = handle_proxy() client = SixgillFeedClient( demisto.params()["client_id"], demisto.params()["client_secret"], CHANNEL_CODE, FeedStream.DARKFEED, demisto, max_indicators, SESSION, VERIFY, ) command = demisto.command() demisto.info(f"Command being called is {command}") tags = argToList(demisto.params().get("feedTags", [])) tlp_color = demisto.params().get("tlp_color") commands: dict[str, Callable] = {"test-module": test_module_command, "sixgill-get-indicators": get_indicators_command} confidence = demisto.params().get("confidence") if confidence is None or confidence in ["", "all"]: confidence = "all" else: confidence = arg_to_number(confidence, "confidence", required=False) try: if demisto.command() == "fetch-indicators": indicators = fetch_indicators_command(client, tags=tags, tlp_color=tlp_color, confidence=confidence) for b in batch(indicators, batch_size=2000): demisto.createIndicators(b) else: readable_output, outputs, raw_response = commands[command](client, demisto.args()) return_outputs(readable_output, outputs, raw_response) except Exception as e: demisto.error(traceback.format_exc()) return_error(f"Error failed to execute {demisto.command()}, error: [{e}]") if __name__ == "__builtin__" or __name__ == "builtins": main()