import uuid from collections.abc import Callable from typing import Any import urllib3 from CommonServerPython import * # disable insecure warnings urllib3.disable_warnings() INTEGRATION_NAME = "Office 365" GERMANY = "Germany" ALL_REGIONS_LIST = ["Worldwide", "China", "USGovDoD", "USGovGCCHigh"] ALL_CATEGORY_LIST = ["Optimize", "Allow", "Default"] def build_region_or_category_list(param_list: list, all_config_list: list, allow_germany: bool = False) -> list: """Builds the region or category list for the feed. If the param_list includes 'All', it will add all the items from the 'all_config_list' to the list, and remove the string all. Args: allow_germany: In some cases, Germany endpoints can throw a 400 error, by default we exclude Germany from All param_list: list of regions or categories provided by integration configuration all_config_list: list of all the regions or categories, to be added if All is chosen Returns: list of regions or categories """ if allow_germany: param_list.append(GERMANY) if "All" in param_list: param_list.remove("All") return list(set(param_list + all_config_list)) return param_list def build_urls_dict(regions_list: list, services_list: list, unique_id) -> list[dict[str, Any]]: """Builds a URL dictionary with the relevant data for each service Args: regions_list: list of regions services_list: list of services unique_id: unique uuid Returns: URLs services list """ urls_list = [] for region in regions_list: for service in services_list: if service == "All": url = f"https://endpoints.office.com/endpoints/{region}?ClientRequestId={unique_id}" else: url = f"https://endpoints.office.com/endpoints/{region}?ServiceAreas={service}&ClientRequestId={unique_id}" urls_list.append({"Region": region, "Service": service, "FeedURL": url}) return urls_list class Client: """ Client to use in the Office 365 Feed integration. Overrides BaseClient. Office 365 IP address and URL web service announcement: https://docs.microsoft.com/en-us/office365/enterprise/managing-office-365-endpoints?redirectSourcePath=%252fen-us%252farticle%252fmanaging-office-365-endpoints-99cab9d4-ef59-4207-9f2b-3728eb46bf9a#webservice https://techcommunity.microsoft.com/t5/Office-365-Blog/Announcing-Office-365-endpoint-categories-and-Office-365-IP/ba-p/177638 """ def __init__( self, urls_list: list, category_list: list, insecure: bool = False, tags: list | None = None, tlp_color: str | None = None ): """ Implements class for Office 365 feeds. :param urls_list: List of url, regions and service of each service. :param insecure: boolean, if *false* feed HTTPS server certificate is verified. Default: *false* :param tlp_color: Traffic Light Protocol color. """ self._urls_list: list[dict] = urls_list self._verify: bool = insecure self.tags = [] if tags is None else tags self.tlp_color = tlp_color self._proxies = handle_proxy(proxy_param_name="proxy", checkbox_default_value=False) self.category_list = category_list def build_iterator(self) -> list: """Retrieves all entries from the feed. Returns: A list of objects, containing the indicators. """ result = [] for feed_obj in self._urls_list: feed_url = feed_obj.get("FeedURL", "") region = feed_obj.get("Region") service = feed_obj.get("Service") try: response = requests.get( url=feed_url, verify=self._verify, proxies=self._proxies, ) response.raise_for_status() data = response.json() # filter empty entries and category param, add metadata indicators = [i for i in data if ("ips" in i or "urls" in i) and i.get("category") in self.category_list] for i in indicators: # add relevant fields of services i.update({"Region": region, "Service": service, "FeedURL": feed_url}) result.extend(indicators) except requests.exceptions.SSLError as err: demisto.debug(str(err)) raise Exception( f"Connection error in the API call to {INTEGRATION_NAME}.\nCheck your not secure parameter.\n\n{err}" ) except requests.ConnectionError as err: demisto.debug(str(err)) raise Exception( f"Connection error in the API call to {INTEGRATION_NAME}.\nCheck your Server URL parameter.\n\n{err}" ) except requests.exceptions.HTTPError as err: demisto.debug(f"Got an error from {feed_url} while fetching indicators {err!s} ") if err.response.status_code == 503: raise Exception(f"The service located at {feed_url} is unavailable while fetching indicators {err!s} ") elif err.response.status_code == 400 and region == GERMANY: raise Exception( "The service returned a 400 status code, this could possibly be due to the Germany" " endpoint being unavailable. Please exclude Germany from All using the parameter" " Allow Germany." ) else: raise Exception(f"HTTP error in the API call to {INTEGRATION_NAME}.\n\n{err}") except ValueError as err: demisto.debug(str(err)) raise ValueError(f"Could not parse returned data to Json. \n\nError massage: {err}") return result @staticmethod def check_indicator_type(indicator): """Checks the indicator type. The indicator type can be classified as one of the following values: CIDR, IPv6CIDR, IP, IPv6 or Domain. Args: indicator: indicator value Returns: The type of the indicator """ is_ip_indicator = FeedIndicatorType.ip_to_indicator_type(indicator) if is_ip_indicator: return is_ip_indicator elif "*" in indicator: return FeedIndicatorType.DomainGlob # domain else: return FeedIndicatorType.Domain def test_module(client: Client, *_) -> tuple[str, dict[Any, Any], dict[Any, Any]]: """Builds the iterator to check that the feed is accessible. Args: client: Client object. Returns: Outputs. """ client.build_iterator() return "ok", {}, {} def fetch_indicators(client: Client, indicator_type_lower: str, limit: int = -1, enrichment_excluded: bool = False) -> list[dict]: """Retrieves indicators from the feed Args: client: Client object with request indicator_type_lower: indicator type limit: limit the results Returns: Indicators. """ iterator = client.build_iterator() # filter indicator_type specific entries if indicator_type_lower != "both": iterator = [i for i in iterator if indicator_type_lower in i] indicators = [] if limit > 0: iterator = iterator[:limit] for item in iterator: if indicator_type_lower == "both": values = item.get("ips", []) + item.get("urls", []) else: values = item.get(indicator_type_lower) if values: for value in values: type_ = Client.check_indicator_type(value) raw_data = { "value": value, "type": type_, } for key, val in item.items(): if key not in ["ips", "urls"]: raw_data.update({key: val}) indicator_mapping_fields = {"port": argToList(item.get("tcpPorts", "")), "service": item.get("serviceArea", "")} if item.get("expressRoute"): indicator_mapping_fields["office365expressroute"] = item.get("expressRoute") if item.get("category"): indicator_mapping_fields["office365category"] = item.get("category") if item.get("required"): indicator_mapping_fields["office365required"] = item.get("required") if item.get("notes"): indicator_mapping_fields["description"] = item.get("notes") indicator_mapping_fields["tags"] = client.tags if client.tlp_color: indicator_mapping_fields["trafficlightprotocol"] = client.tlp_color indicator_obj = { "value": value, "type": type_, "rawJSON": raw_data, "fields": indicator_mapping_fields, } if enrichment_excluded: indicator_obj["enrichmentExcluded"] = enrichment_excluded indicators.append(indicator_obj) return indicators def get_indicators_command( client: Client, args: dict[str, str], enrichment_excluded: bool = False ) -> tuple[str, dict[Any, Any], dict[Any, Any]]: """Wrapper for retrieving indicators from the feed to the war-room. Args: client: Client object with request args: demisto.args() Returns: Outputs. """ indicator_type = str(args.get("indicator_type")) indicator_type_lower = indicator_type.lower() limit = int(demisto.args().get("limit")) if "limit" in demisto.args() else 10 indicators = fetch_indicators(client, indicator_type_lower, limit, enrichment_excluded) human_readable = tableToMarkdown("Indicators from Office 365 Feed:", indicators, headers=["value", "type"], removeNull=True) return human_readable, {}, {"raw_response": indicators} def fetch_indicators_command(client: Client, enrichment_excluded: bool = False) -> list[dict]: """Wrapper for fetching indicators from the feed to the Indicators tab. Args: client: Client object with request Returns: Indicators. """ indicators = fetch_indicators(client, "both", enrichment_excluded=enrichment_excluded) return indicators def main(): """ PARSE AND VALIDATE INTEGRATION PARAMS """ params = demisto.params() unique_id = str(uuid.uuid4()) regions_list = build_region_or_category_list( argToList(params.get("regions")), ALL_REGIONS_LIST, allow_germany=params.get("allow_germany") ) services_list = argToList(params.get("services")) category_list = build_region_or_category_list(argToList(params.get("category", ["All"])), ALL_CATEGORY_LIST) urls_list = build_urls_dict(regions_list, services_list, unique_id) use_ssl = not params.get("insecure", False) tags = argToList(params.get("feedTags")) tlp_color = params.get("tlp_color") enrichment_excluded = demisto.params().get("enrichmentExcluded", False) command = demisto.command() demisto.info(f"Command being called is {command}") try: client = Client(urls_list, category_list, use_ssl, tags, tlp_color) commands: dict[str, Callable[[Client, dict[str, str]], tuple[str, dict[Any, Any], dict[Any, Any]]]] = { "test-module": test_module, "office365-get-indicators": get_indicators_command, } if command in commands: return_outputs(*commands[command](client, demisto.args())) elif command == "fetch-indicators": indicators = fetch_indicators_command(client, enrichment_excluded) for iter_ in batch(indicators, batch_size=2000): demisto.createIndicators(iter_) else: raise NotImplementedError(f"Command {command} is not implemented.") except Exception as err: err_msg = f"Error in {INTEGRATION_NAME} Integration. [{err}]" return_error(err_msg) if __name__ in ["__main__", "builtin", "builtins"]: main()