import re from traceback import format_exc import demistomock as demisto # noqa: F401 import urllib3 from bs4 import BeautifulSoup, element from CommonServerPython import * # noqa: F401 # disable insecure warnings urllib3.disable_warnings() CIDR = "CIDR" DOMAIN = "DOMAIN" INTEGRATION_NAME = "Webex" BASE_URL = "https://help.webex.com/en-us/WBX264/How-Do-I-Allow-Webex-Meetings-Traffic-on-My-Network" DOMAIN_REGEX = r"([\^]*[\*\.]*[a-z0-9+\-]+\.+.*)*" def grab_domain_table(html_section: element.Tag) -> List: """Gets the domain table from the given html section""" table = html_section.find("table", attrs={"class": "li"}) table_body = table.find("tbody") # type: ignore[union-attr] rows = table_body.find_all("tr") # type: ignore[union-attr] data = [] for row in rows: cols = row.find_all("td") cols = [ele.text.strip() for ele in cols] data.append([ele for ele in cols if ele]) # Get rid of empty values return data def grab_domains(data: list) -> List: """From a given list of lists, that each contains a row from the webex table, text and domains, return a list of all the domains only """ domainList: List = [] # skip the last line of the table, it's a message and not a domain for lines in data[:-1]: domains = lines[1].split(" ") cleanDomain = " ".join(re.findall(DOMAIN_REGEX, domains[0])) # Strip Whitespace lines to remove blank values cleanDomain = cleanDomain.strip() if "\t\t\t" in cleanDomain: # multiple domains in one line multiple_domains_lst = cleanDomain.split("\t\t\t") domainList += multiple_domains_lst elif cleanDomain: domainList.append(cleanDomain) # remove duplicates and convert it back to a list fot the slicing to work finel_domainList = list(set(domainList)) return finel_domainList def grab_ip_table(html_section: element.Tag) -> List: """Gets the IP table as an html, parses it and returns a list of lists each one contains a row from the table """ rows = html_section.find_all("ul") data = [] for row in rows: cols = row.find_all("li") cols = [ele.text.strip() for ele in cols] data.append([ele for ele in cols if ele]) # Get rid of empty values return data def grab_CIDR_ips(data: list) -> List: """From list of lists that contain all rows of the IP webex table, return a list with only CIDR ip addresses """ CIDR_ip_list: List = [] if not data[0]: raise DemistoException("Did not find any IP indicators in the IP table") for line in data[0]: values = line.split(" (CIDR)") CIDR_ip_list.append(values[0]) # remove duplicates and convert it back to a list fot the slicing to work finel_CIDR_ip_list = list(set(CIDR_ip_list)) return finel_CIDR_ip_list def parse_indicators_from_response(response: requests.Response) -> Dict[str, List[str]]: """Recives an html page from the Webex website that contains a IP table and a DOMAIN table. Parses the page, and returns a dict with two keys: DOMAIN and CIDR(ip ranges), while the value is a list of the related indicators. """ try: soup = BeautifulSoup(response.text, "html.parser") # Get the IP and Domain Sections from the html ipsSection = soup.find("div", {"id": "id_135011"}) domainsSection = soup.find("div", {"id": "id_135010"}) # Get Domains domainTable = grab_domain_table(domainsSection) # type: ignore[arg-type] all_domains_lst = grab_domains(domainTable) # Get IPS ipTable = grab_ip_table(ipsSection) # type: ignore[arg-type] all_IPs_lst = grab_CIDR_ips(ipTable) except Exception as e: raise DemistoException(f"Failed to parse the response from the website. Error: {e!s}") if not all_IPs_lst or not all_domains_lst: raise DemistoException("Did not find the expected indicators in the response from the website") all_info_dict = {CIDR: all_IPs_lst, DOMAIN: all_domains_lst} return all_info_dict def check_indicator_type(indicator: str) -> str: """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 class Client(BaseClient): """A client class that implements connectivity with the website.""" def all_raw_data(self) -> requests.Response: """Gets the entire html page from the website.""" try: return self._http_request(method="GET", url_suffix="", resp_type="response") except DemistoException as e: raise e def test_module(client: Client) -> str: # pragma: no cover """Tests connectivity with the client. Args: client: Client object. Returns: str: ok if test passed else the exception message. """ try: client.all_raw_data() except DemistoException as e: return e.message return "ok" def get_indicators_command(client: Client, **args) -> CommandResults: """Gets indicators from the Webex website and sends them to the war-room.""" client = client limit = arg_to_number(args.get("limit", 20)) requested_indicator_type = args.get("indicator_type", "Both") if requested_indicator_type not in ["Both", "CIDR", "DOMAIN"]: raise DemistoException("The indicator_type argument must be one of the following: Both, CIDR, DOMAIN") res = client.all_raw_data() clean_res = parse_indicators_from_response(res) # parse the data from an html page to a list of dicts with ips and domains if requested_indicator_type != "Both": indicators = clean_res[requested_indicator_type][:limit] else: indicators = clean_res[CIDR][:limit] + clean_res[DOMAIN][:limit] final_indicators_lst = [] for value in indicators: type_ = check_indicator_type(value) indicators_and_type = {"value": value, "type": type_} final_indicators_lst.append(indicators_and_type) md = tableToMarkdown("Indicators from Webex:", final_indicators_lst, headers=["value", "type"], removeNull=True) return CommandResults( readable_output=md, ) def fetch_indicators_command( client: Client, tags: tuple = None, tlp_color: str = None, enrichment_excluded: bool = False ) -> list: """Wrapper for fetching indicators from the feed to the Indicators tab. Args: client: Client object with request Returns: Indicators. """ res = client.all_raw_data() # parse the data from an html page to a list of dicts with ips and domains clean_res = parse_indicators_from_response(res) results = [] indicator_mapping_fields = {"tags": tags, "trafficlightprotocol": tlp_color} for indicator in clean_res[CIDR] + clean_res[DOMAIN]: indicator_obj: dict[str, Any] = { "value": indicator, "type": check_indicator_type(indicator), "fields": indicator_mapping_fields, } if enrichment_excluded: indicator_obj["enrichmentExcluded"] = enrichment_excluded results.append(indicator_obj) return results def main(): params = demisto.params() args = demisto.args() verify_certificate = not params.get("insecure", False) proxy = params.get("proxy", False) tags = (params.get("feedTags"),) tlp_color = params.get("tlp_color") enrichment_excluded = demisto.params().get("enrichmentExcluded", False) command = demisto.command() try: client = Client( base_url=BASE_URL, verify=verify_certificate, proxy=proxy, ) if command == "test-module": return_results(test_module(client=client)) elif command == "fetch-indicators": res = fetch_indicators_command(client=client, tags=tags, tlp_color=tlp_color, enrichment_excluded=enrichment_excluded) for iter_ in batch(res, batch_size=2000): demisto.createIndicators(iter_) elif command == "webex-get-indicators": return_results(get_indicators_command(client=client, **args)) else: raise NotImplementedError(f"command {command} is not implemented.") except DemistoException as e: # For any other integration command exception, return an error demisto.error(format_exc()) return_error(f"Failed to execute {command} command. Error: {e!s}.") if __name__ in ["__main__", "builtin", "builtins"]: main()