import demistomock as demisto from CommonServerPython import * from dataclasses import dataclass import enum OUTPUT_PREFIX = "PANSecurityAdvisory." class Client(BaseClient): """The client that conects to the advisories API""" PRODUCTS_ENDPOINT = "/products" ADVISORIES_ENDPOINT = "/advisories" def __init__(self, base_url, api_timeout=60, verify=True, proxy=False, ok_codes=(), headers=None): super().__init__(base_url, verify=verify, proxy=proxy, ok_codes=ok_codes, headers=headers) self.api_timeout = api_timeout def get_products(self): """ Gets the list of Supported Products by the Advisories API """ return self._http_request(method="GET", url_suffix=Client.PRODUCTS_ENDPOINT, timeout=self.api_timeout) def get_advisories(self, product: str, params: dict): """ Gets the list of advisories :product: Required Product name to list advisories for :params: Optional list of GET parameters to include in the request. """ params = params or {} return self._http_request( method="GET", url_suffix=f"{Client.PRODUCTS_ENDPOINT}/{product}{Client.ADVISORIES_ENDPOINT}", timeout=self.api_timeout, params=params, ) def dataclass_to_command_results(result: Any, raw_response: Union[list, dict]) -> CommandResults: """Takes a list, or a single, dataclass instance and converts it to a CommandResult instance.""" if not result: command_result = CommandResults( readable_output="No results.", ) return command_result outputs: Union[list[dict], dict, Any] if isinstance(result, list): outputs = [vars(x) for x in result] summary_list = [vars(x) for x in result] title = result[0]._title output_prefix = result[0]._output_prefix else: outputs = vars(result) summary_list = [vars(result)] title = result._title output_prefix = result._output_prefix extra_args = {} if hasattr(result, "_outputs_key_field"): extra_args["outputs_key_field"] = result._outputs_key_field readable_output = tableToMarkdown(title, summary_list) command_result = CommandResults( outputs_prefix=output_prefix, outputs=outputs, readable_output=readable_output, raw_response=raw_response, **extra_args ) return command_result @dataclass class Advisory: """ :param data_type: The type of advisory this is :param data_format: The format of the advisory, such as MITRE :param cve_id: The ID of the CVE described by this advisory :param cve_date_public: The date this CVE was released :param cve_title: The name of this CVE :param affects_product_name: The name of the product this affects, such as PAN-OS :param description: Human readable description of Advisory :param affected_version_list: List of affected versions strings """ data_type: str data_format: str cve_id: str cve_date_public: str cve_title: str description: str cvss_score: int cvss_severity: str cvss_vector_string: str affected_version_list: list _output_prefix = OUTPUT_PREFIX + "Advisory" _title = "Palo Alto Networks Security Advisories" class SeverityEnum(enum.Enum): HIGH = "HIGH" CRITICAL = "CRITICAL" MEDIUM = "MEDIUM" LOW = "LOW" NONE = "NONE" def flatten_advisory_dict(advisory_dict: dict) -> Advisory: """ Given a dictionary representing a CVE advisory, return an Advisory object with relevant fields extracted and flattened. :param advisory_dict: Dictionary containing CVE advisory information. :return: Advisory object with fields populated from advisory_dict. """ cna = advisory_dict.get("containers", {}).get("cna", {}) metrics = cna.get("metrics", [{}]) cvss_info = metrics[0].get("cvssV4_0", {}) return Advisory( data_type=advisory_dict.get("dataType", ""), data_format=metrics[0].get("format", ""), cve_id=advisory_dict.get("cveMetadata", {}).get("cveId", ""), cve_title=cna.get("title", ""), cve_date_public=cna.get("datePublic", ""), description=cna.get("descriptions", [{}])[0].get("value", ""), cvss_score=cvss_info.get("baseScore", 0), cvss_severity=cvss_info.get("baseSeverity", ""), cvss_vector_string=cvss_info.get("vectorString", ""), affected_version_list=cna.get("x_affectedList", []), ) def test_module(client: Client): """Test the connectivity to the advisory API by checking for products""" request_result = client.get_products() if request_result.get("success"): return "ok" return None def get_advisories( client: Client, product: str, sort: str = "-date", severity: SeverityEnum = None, q: str = "" ) -> CommandResults: """ Gets all the advisories for the given product. :param client: HTTP Client !no-auto-argument :param product: Product name to search for advisories :param sort: Sort returned advisories by this value, can be date, cvss, etc. Leading hyphpen (-) indicates reverse search. :param severity: Filter advisories to this severity level only. :param q: Text search query """ params_dict = {"severity": severity, "sort": sort, "product": product} if q: params_dict["q"] = f'"{q}"' advisory_data = client.get_advisories(product, params_dict).get("data", {}) advisory_object_list: list[Advisory] = [] advisory_dict: dict for advisory_dict in advisory_data: advisory_object_list.append(flatten_advisory_dict(advisory_dict)) return dataclass_to_command_results(advisory_object_list, raw_response=advisory_data) def advisory_to_indicator(advisory_dict: dict) -> dict: """Convert the advisory dictionary into an indicator dictionary Args: advisory_dict: advisory dictionary Returns: indicator dictionary """ fields: dict = {} containers_cna = advisory_dict.get("containers", {}).get("cna", {}) if problem_types := containers_cna.get("problemTypes"): tags = [] # holds cwe information as tags for problem_type in problem_types: for description in problem_type.get("descriptions"): tags.append(description.get("cweId")) fields["tags"] = tags if references := containers_cna.get("references", {}): fields["publications"] = [{"link": x.get("url")} for x in references] impacts: list = containers_cna.get("impacts", [{}]) cvss: dict = containers_cna.get("metrics", [{}])[0].get("cvssV4_0", {}) # score mirrored to both fields so that default cve layout displays with full data fields["cvss"] = cvss.get("baseScore", "") fields["cvssscore"] = cvss.get("baseScore", "") fields["cvssvector"] = cvss.get("vectorString", "") fields["sourceoriginalseverity"] = cvss.get("baseSeverity", "") # mirror data in these fields so default CVE layout does not need to be changed # cvedescription not in default cve layout advisory_description = containers_cna.get("descriptions", [{}])[0].get("value", "") fields["cvedescription"] = advisory_description # description in default cve layout fields["description"] = advisory_description fields["published"] = containers_cna.get("datePublic", "") fields["name"] = containers_cna.get("title", "") if impacts and cvss.get("version") in ["3.1", "4.0"]: fields["cvssversion"] = cvss.get("version", "") # is this v3/v4 cvss? # fills out the cvsstable in default cve layout - different table column names cvss_data = [] for k, v in cvss.items(): cvss_data.append({"metrics": camel_case_to_underscore(k).replace("_", " ").title(), "value": v}) fields["cvsstable"] = cvss_data return { "value": advisory_dict.get("cveMetadata", {}).get("cveId"), "type": FeedIndicatorType.CVE, "rawJSON": advisory_dict, "fields": fields, } def fetch_indicators(client: Client, fetch_product_name="PAN-OS") -> list[dict]: """ Fetch Advisories as CVE indicators. :param client: Client instance :param fetch_product_name: The name of the product to fetch indicators for. """ advisory_data = client.get_advisories(fetch_product_name, {}).get("data", {}) indicator_objects = [] for advisory_dict in advisory_data: indicator_objects.append(advisory_to_indicator(advisory_dict)) return indicator_objects def main(): """Main entrypoint for script""" client = Client(base_url=demisto.params().get("url")) command_name = demisto.command() demisto.info(f"Command being called is {command_name}") try: if command_name == "test-module": return_results(test_module(client)) elif command_name == "pan-advisories-get-advisories": return_results(get_advisories(client, **demisto.args())) elif command_name == "fetch-indicators": for b in batch(fetch_indicators(client, demisto.params().get("fetch_product_name"))): demisto.createIndicators(b) else: raise NotImplementedError(f"command {command_name} is not implemented.") except Exception as err: return_error(str(err)) if __name__ in ("__main__", "__builtin__", "builtins"): main()