import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 """Cortex XSOAR Integration for Expanse Expander and Behavior """ from CommonServerUserPython import * # noqa import urllib3 import base64 from datetime import datetime, timedelta from ipaddress import IPv4Address, AddressValueError, summarize_address_range from typing import ( Any, Dict, Optional, Iterable, List ) # Disable insecure warnings urllib3.disable_warnings() # pylint: disable=no-member """ CONSTANTS """ TOKEN_DURATION = 7200 DEFAULT_MAX_INDICATORS = 10 # used in expanse-get-indicators DEFAULT_FETCH_MAX_INDICATORS = 1000 # used in fetch-indicators DEFAULT_FETCH_MIN_LAST_OBSERVED = 7 # used in fetch-indicators """ CLIENT CLASS """ class Client(BaseClient): """Client class to interact with the Expanse API""" def __init__( self, base_url: str, api_key: str, verify: bool, proxy: bool, **kwargs ): self.api_key = api_key hdr = { "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "Expanse_XSOAR/0.0.1", } super().__init__(base_url, verify=verify, proxy=proxy, headers=hdr, **kwargs) def authenticate(self) -> None: """ Perform authentication using API_KEY, stores token and stored timestamp in integration context, retrieves new token when expired """ current_utc_timestamp = int(datetime.utcnow().timestamp()) stored_token = demisto.getIntegrationContext() if ( isinstance(stored_token, dict) and "token" in stored_token and "expires" in stored_token and current_utc_timestamp < int(stored_token["expires"]) ): self._headers['Authorization'] = f'JWT {stored_token["token"]}' else: # fetch new token hdr = self._headers.copy() hdr["Authorization"] = f"Bearer {self.api_key}" r = self._http_request("GET", "/v1/IdToken", headers=hdr) if isinstance(r, dict) and r.get("token", None) is None: raise ValueError("Authorization failed") token_expiration = current_utc_timestamp + TOKEN_DURATION self._headers['Authorization'] = f'JWT {r["token"]}' demisto.setIntegrationContext( {"token": r["token"], "expires": token_expiration} ) def _paginate(self, method: str, url_suffix: str, params: Optional[Dict[str, Any]]) -> Iterable[Any]: next_url: Optional[str] = None while True: result = self._http_request( method=method, url_suffix=url_suffix, full_url=next_url, params=params, raise_on_status=True, timeout=30 ) data = result.get('data', []) if data is not None: yield from data pagination = result.get('pagination', None) if pagination is None: break next_url = pagination.get('next', None) if next_url is None: break params = None def get_iprange_by_id(self, iprange_id: str) -> Dict[str, Any]: result = self._http_request( method="GET", url_suffix=f"/v2/ip-range/{iprange_id}", raise_on_status=True, params={ 'include': "severityCounts,annotations,attributionReasons,relatedRegistrationInformation,locationInformation" } ) return result def get_domain_by_domain(self, domain: str, last_observed_date: Optional[str]) -> Dict[str, Any]: params = {} if last_observed_date is not None: params['minRecentIpLastObservedDate'] = last_observed_date result = self._http_request( method="GET", url_suffix=f"/v2/assets/domains/{domain}", raise_on_status=True, params=params ) return result def get_certificate_by_pem_md5_hash(self, pem_md5_hash: str, last_observed_date: Optional[str]) -> Dict[str, Any]: params = {} if last_observed_date is not None: params['minRecentIpLastObservedDate'] = last_observed_date result = self._http_request( method="GET", url_suffix=f"/v2/assets/certificates/{pem_md5_hash}", raise_on_status=True, params=params ) return result def get_ipranges(self, params: Dict[str, Any]) -> Iterable[Any]: return self._paginate( method="GET", url_suffix="/v2/ip-range", params=params ) def get_domains(self, params: Dict[str, Any]) -> Iterable[Any]: return self._paginate( method="GET", url_suffix="/v2/assets/domains", params=params ) def get_certificates(self, params: Dict[str, Any]) -> Iterable[Any]: return self._paginate( method="GET", url_suffix="/v2/assets/certificates", params=params ) def get_ips(self, params: Dict[str, Any]) -> Iterable[Any]: return self._paginate( method="GET", url_suffix="/v2/assets/ips", params=params ) """ HELPER FUNCTIONS """ def check_int(arg: Any, arg_name: str, min_val: int = None, max_val: int = None, required: bool = False) -> Optional[int]: """Converts a string argument to a Python int This function is used to quickly validate an argument provided and convert it into an ``int`` type. It will throw a ValueError if the input is invalid or outside the optional range. If the input is None, it will throw a ValueError if required is ``True``, or ``None`` if required is ``False. """ # check if argument is mandatory if arg is None: if required is True: raise ValueError(f'Missing argument "{arg_name}"') return None i: Optional[int] = None if isinstance(arg, str): if not arg.isdigit(): raise ValueError(f'Integer invalid: "{arg_name}"="{arg}"') try: i = int(arg) except ValueError: raise ValueError(f'Integer invalid: "{arg_name}"="{arg}"') elif isinstance(arg, int): i = arg else: raise ValueError(f'Invalid number: "{arg_name}"') # range check if min_val and i < min_val: raise ValueError(f'Integer outside minimum range: "{arg_name}"="{arg}" ("min={min_val}")') if max_val and i > max_val: raise ValueError(f'Integer outside maximum range: "{arg_name}"="{arg}" ("max={max_val}")') return i def validate_max_indicators(max_indicators_param: Optional[str]) -> Optional[int]: try: max_indicators = check_int( max_indicators_param, 'maxIndicators', None, None, False ) except ValueError: return None if max_indicators is None: max_indicators = DEFAULT_FETCH_MAX_INDICATORS return max_indicators def validate_min_last_observed(min_last_observed_param: Optional[str]) -> Optional[str]: try: min_last_observed = check_int( min_last_observed_param, 'minLastObserved', None, None, False ) except ValueError: return None if min_last_observed is None: min_last_observed = DEFAULT_FETCH_MIN_LAST_OBSERVED start_time = datetime.utcnow() - timedelta(days=min_last_observed) return start_time.strftime("%Y-%m-%d") def safe_b64_to_hex(i: str) -> Optional[str]: if not i: return None try: return base64.urlsafe_b64decode(i).hex() except Exception: return None """ INDICATORS HANDLING FUNCTIONS """ def ip_to_demisto_indicator(ip_indicator: Dict[str, Any]) -> Optional[Dict[str, Any]]: value = ip_indicator.get('ip', None) if value is None: return None provider_name = ip_indicator.get('provider', {}).get('name') tenant_name = ip_indicator.get('tenant', {}).get('name') business_units = ip_indicator.get("businessUnits", []) business_unit_names = [bu['name'] for bu in business_units if bu.get('name')] # to faciliate classifiers ip_indicator['expanseType'] = 'ip' return { 'type': FeedIndicatorType.IP, 'value': value, 'rawJSON': ip_indicator, 'score': Common.DBotScore.NONE, 'fields': { 'expansetype': ip_indicator.get('type', None), 'expanseassetype': ip_indicator.get('assetType', None), 'expansedomain': ip_indicator.get('domain', None), 'expansecommonname': ip_indicator.get('commonName', None), 'expanselastobserved': ip_indicator.get('lastObserved', None), 'lastseenbysource': ip_indicator.get('lastObserved', None), 'expanseprovidername': provider_name, 'expansetenantname': tenant_name, 'expansebusinessunits': business_unit_names } } def certificate_to_demisto_indicator(certificate: Dict[str, Any]) -> Optional[Dict[str, Any]]: certificate_details = certificate.get('certificate') if certificate_details is None: return None ec_sha256 = certificate_details.get('pemSha256') if ec_sha256 is None: return None indicator_value = safe_b64_to_hex(ec_sha256) if indicator_value is None: return None ec_md5 = certificate_details.get('md5Hash') ec_sha1 = certificate_details.get('pemSha1') ec_spki = certificate_details.get('publicKeySpki') ec_modulus = certificate_details.get('publicKeyModulus') ec_publickey = None if (pktemp := certificate_details.get('publicKey')) is not None: ec_publickey = safe_b64_to_hex(pktemp) ec_san = certificate_details.get('subjectAlternativeNames') ec_names = set() if isinstance(ec_san, str) and len(ec_san) > 0: ec_names.update([san for san in ec_san.split() if len(san) != 0]) if (ec_subject_name := certificate_details.get('subjectName')) is not None: ec_names.add(ec_subject_name) annotations = certificate.get('annotations', {}) tags = [tag['name'] for tag in annotations.get('tags', [])] providers = certificate.get('providers', []) provider_name = None if len(providers) == 0 else providers[0].get('name') tenant_name = certificate.get('tenant', {}).get('name') business_units = certificate.get("businessUnits", []) business_unit_names = [bu['name'] for bu in business_units if bu.get('name')] return { 'type': 'Certificate', 'value': indicator_value, 'rawJSON': certificate, 'score': Common.DBotScore.NONE, 'fields': { # standard context "serialnumber": certificate_details.get('serialNumber'), "certificatenames": '\n'.join(sorted(list(ec_names))) if ec_names else None, "subjectdn": certificate_details.get('subject'), "certificatesignature": {'algorithm': certificate_details.get('signatureAlgorithm')}, "subjectalternativenames": [{'value': san} for san in ec_san.split() if len(san) != 0] if ec_san else None, "validitynotafter": certificate_details.get('validNotAfter'), "spkisha256": safe_b64_to_hex(ec_spki) if ec_spki else None, "validitynotbefore": certificate_details.get('validNotBefore'), "publickey": { "algorithm": certificate_details.get('publicKeyAlgorithm'), "length": certificate_details.get('publicKeyBits'), "modulus": ':'.join([ec_modulus[i:i + 2] for i in range(0, len(ec_modulus), 2)]) if ec_modulus else None, "exponent": certificate_details.get('publicKeyRsaExponent'), "publickey": ':'.join([ec_publickey[i:i + 2] for i in range(0, len(ec_publickey), 2)]) if ec_publickey else None }, "issuerdn": certificate_details.get('issuer'), "md5": safe_b64_to_hex(ec_md5) if ec_md5 else None, "sha1": safe_b64_to_hex(ec_sha1) if ec_sha1 else None, "sha256": indicator_value, # expanse specific 'expansetags': tags, 'expansecertificateadvertisementstatus': certificate.get('certificateAdvertisementStatus', None), 'expansedateadded': certificate.get('dateAdded', None), 'expansefirstobserved': certificate.get('firstObserved', None), 'firstseenbysource': certificate.get('firstObserved', None), 'expanselastobserved': certificate.get('lastObserved', None), 'lastseenbysource': certificate.get('lastObserved', None), 'expanseproperties': '\n'.join(certificate.get('properties', [])), 'expanseservicestatus': certificate.get('serviceStatus', None), 'expanseprovidername': provider_name, 'expansetenantname': tenant_name, 'expansebusinessunits': business_unit_names } } def domain_to_demisto_indicator(domain_indicator: Dict[str, Any]) -> Optional[Dict[str, Any]]: domain = domain_indicator.get('domain', None) if domain is None: return None annotations = domain_indicator.get('annotations', {}) tags = [tag['name'] for tag in annotations.get('tags', [])] providers = domain_indicator.get('providers', []) provider_name = None if len(providers) == 0 else providers[0].get('name') tenant_name = domain_indicator.get('tenant', {}).get('name') business_units = domain_indicator.get("businessUnits", []) business_unit_names = [bu['name'] for bu in business_units if bu.get('name')] # to faciliate classifiers domain_indicator['expanseType'] = 'domain' return { 'type': 'DomainGlob' if '*' in domain else 'Domain', 'value': domain, 'rawJSON': domain_indicator, 'score': Common.DBotScore.NONE, 'fields': { 'expansetags': tags, 'expansednsresolutionstatus': domain_indicator.get('dnsResolutionStatus', None), 'expansedateadded': domain_indicator.get('dateAdded', None), 'expansefirstobserved': domain_indicator.get('firstObserved', None), 'firstseenbysource': domain_indicator.get('firstObserved', None), 'expanselastobserved': domain_indicator.get('lastObserved', None), 'lastseenbysource': domain_indicator.get('lastObserved', None), 'expanselastsampledip': domain_indicator.get('lastSampledIp', None), 'expanseservicestatus': domain_indicator.get('serviceStatus', None), 'expansesourcedomain': domain_indicator.get('sourceDomain', None), 'expanseprovidername': provider_name, 'expansetenantname': tenant_name, 'expansebusinessunits': business_unit_names } } def iprange_to_demisto_indicator(iprange_indicator: Dict[str, Any]) -> Iterable[Dict[str, Any]]: demisto.debug(f"{iprange_indicator!r}") ipVersion = iprange_indicator.get('ipVersion', "4") if ipVersion != "4": return [] start_address = iprange_indicator.get('startAddress', None) if start_address is None: return [] try: start_address = IPv4Address(start_address) except AddressValueError: return [] end_address = iprange_indicator.get('endAddress', None) if end_address is None: return [] try: end_address = IPv4Address(end_address) except AddressValueError: return [] if end_address < start_address: return [] annotations = iprange_indicator.get('annotations', {}) tags = [tag['name'] for tag in annotations.get('tags', []) if tag.get('name')] business_units = iprange_indicator.get("businessUnits", []) business_unit_names = [bu['name'] for bu in business_units if bu.get('name')] attribution_reasons: List[str] = [ar['reason'] for ar in iprange_indicator.get('attributionReason', []) if ar.get('reason')] # to faciliate classifiers iprange_indicator['expanseType'] = 'iprange' for cidr in summarize_address_range(start_address, end_address): yield { 'type': 'CIDR', 'value': str(cidr), 'rawJSON': iprange_indicator, 'score': Common.DBotScore.NONE, 'fields': { 'expansetags': tags, 'expansebusinessunits': business_unit_names, 'expanseseveritycount': iprange_indicator.get('severityCounts', {}), 'expanseattributionreason': attribution_reasons, 'expanseresponsiveipcount': iprange_indicator.get('responsiveIpCount', None) } } """ COMMAND FUNCTIONS """ def test_module( client: Client, max_indicators_param: Optional[str], min_last_observed_param: Optional[str], tlp_color: Optional[str]) -> str: max_indicators = validate_max_indicators(max_indicators_param) if max_indicators is None: return "Invalid value for max indicators" min_last_observed = validate_min_last_observed(min_last_observed_param) if min_last_observed is None: return "Invalide value for last observed range" try: client.get_ips(params={ 'limit': 1, 'minLastObservedDate': min_last_observed }) except DemistoException as e: if "Forbidden" in str(e) or "Authorization failed" in str(e): return "Authorization Error: make sure API Key is correctly set" else: raise e return "ok" def indicator_generator(client: Client, max_indicators: Optional[int] = None, min_last_observed: Optional[str] = None, retrieve_ip: Optional[bool] = True, retrieve_domain: Optional[bool] = True, retrieve_certificate: Optional[bool] = True, retrieve_iprange: Optional[bool] = True, tlp_color: Optional[str] = '', feed_tags: str = '') -> Iterable[Dict[str, Any]]: if retrieve_ip: get_ips_params = {} if min_last_observed is not None: get_ips_params['minLastObservedDate'] = min_last_observed num_indicators = 0 for ip_indicator in client.get_ips(params=get_ips_params): if max_indicators is not None and num_indicators >= max_indicators: break demisto_indicator = ip_to_demisto_indicator(ip_indicator) if demisto_indicator is None: continue if tlp_color is not None: demisto_indicator['fields']['trafficlightprotocol'] = tlp_color if feed_tags: demisto_indicator['fields']['tags'] = feed_tags yield demisto_indicator num_indicators += 1 if retrieve_certificate: get_certificates_params = {'sort': '-dateAdded'} if min_last_observed is not None: get_certificates_params['minLastObservedDate'] = min_last_observed num_indicators = 0 for cert_indicator in client.get_certificates(params=get_certificates_params): if max_indicators is not None and num_indicators >= max_indicators: break demisto_indicator = certificate_to_demisto_indicator(cert_indicator) if demisto_indicator is None: continue if tlp_color is not None: demisto_indicator['fields']['trafficlightprotocol'] = tlp_color if feed_tags: demisto_indicator['fields']['tags'] = feed_tags yield demisto_indicator num_indicators += 1 if retrieve_domain: get_domains_params = {'sort': '-dateAdded'} if min_last_observed is not None: get_domains_params['minLastObservedDate'] = min_last_observed num_indicators = 0 for domain_indicator in client.get_domains(params=get_domains_params): if max_indicators is not None and num_indicators >= max_indicators: break demisto_indicator = domain_to_demisto_indicator(domain_indicator) if demisto_indicator is None: continue if tlp_color is not None: demisto_indicator['fields']['trafficlightprotocol'] = tlp_color if feed_tags: demisto_indicator['fields']['tags'] = feed_tags yield demisto_indicator num_indicators += 1 if retrieve_iprange: num_indicators = 0 iprange_params = {'include': 'severityCounts,annotations,attributionReasons'} for iprange_indicator in client.get_ipranges(params=iprange_params): if max_indicators is not None and num_indicators >= max_indicators: break for demisto_indicator in iprange_to_demisto_indicator(iprange_indicator): if tlp_color is not None: demisto_indicator['fields']['trafficlightprotocol'] = tlp_color if feed_tags: demisto_indicator['fields']['tags'] = feed_tags yield demisto_indicator num_indicators += 1 def get_indicators_command(client: Client, args: Dict[str, Any], tlp_color: Optional[str] = '', feed_tags: str = '') -> CommandResults: retrieve_ip = argToBoolean(args.get('ip', 'yes')) retrieve_domain = argToBoolean(args.get('domain', 'yes')) retrieve_certificate = argToBoolean(args.get('certificate', 'yes')) retrieve_iprange = argToBoolean(args.get('iprange', 'yes')) max_indicators = check_int(args.get('max_indicators'), 'max_indicators', None, None, False) if not max_indicators: max_indicators = DEFAULT_MAX_INDICATORS indicators: List[Dict[str, Any]] = [] for indicator in indicator_generator( client, max_indicators=max_indicators, retrieve_ip=retrieve_ip, retrieve_domain=retrieve_domain, retrieve_certificate=retrieve_certificate, retrieve_iprange=retrieve_iprange, tlp_color=tlp_color, feed_tags=feed_tags): indicators.append(indicator) readable_output = tableToMarkdown( f'Expanse Indicators (capped at {max_indicators})', indicators, headers=['value', 'type'] ) return CommandResults( readable_output=readable_output, outputs_prefix=None, outputs_key_field=None, outputs=None, raw_response=indicators ) """ MAIN FUNCTION """ def main() -> None: """main function, parses params and runs command functions :return: :rtype: """ params = demisto.params() command = demisto.command() api_key = params.get('credentials', {}).get('password', '') or params.get("apikey", '') base_url = urljoin(params["url"], "/api") verify_certificate = not params.get("insecure", False) proxy = params.get("proxy", False) max_indicators_param = params.get('max_fetch') min_last_observed_param = params.get('min_last_observed') tlp_color = params.get('tlp_color') feed_tags = params.get('feedTags', '') try: client = Client( api_key=api_key, base_url=base_url, verify=verify_certificate, proxy=proxy ) client.authenticate() if command == "test-module": result = test_module(client, max_indicators_param, min_last_observed_param, tlp_color) return_results(result) elif command == "fetch-indicators": max_indicators = validate_max_indicators(max_indicators_param) if validate_max_indicators is None: raise ValueError("Invalid value for max indicators") min_last_observed = validate_min_last_observed(min_last_observed_param) if min_last_observed is None: raise ValueError("Invalid value for last observed day range") indicator_batch = [] for indicator in indicator_generator(client, tlp_color=tlp_color, feed_tags=feed_tags, max_indicators=max_indicators): indicator_batch.append(indicator) if len(indicator_batch) == 2000: demisto.createIndicators(indicator_batch) indicator_batch = [] if len(indicator_batch) != 0: demisto.createIndicators(indicator_batch) elif command == "feedexpanse-get-indicators": return_results(get_indicators_command(client, demisto.args(), tlp_color=tlp_color, feed_tags=feed_tags)) else: raise NotImplementedError(f'Command {command} is not implemented.') # Log exceptions and return errors except Exception as e: return_error( f"Failed to execute {command} command.\nError:\n{str(e)}" ) """ ENTRY POINT """ if __name__ in ("__main__", "__builtin__", "builtins"): main()