Google Safe Browsing v2

from typing import Any

import demistomock as demisto
import urllib3
from CommonServerPython import *  # noqa # pylint: disable=unused-wildcard-import

from CommonServerUserPython import *  # noqa

# Disable insecure warnings
urllib3.disable_warnings()

TYPES = {
    "threatTypes": ["MALWARE", "SOCIAL_ENGINEERING", "POTENTIALLY_HARMFUL_APPLICATION", "UNWANTED_SOFTWARE"],
    "platformTypes": ["ANY_PLATFORM", "WINDOWS", "LINUX", "ALL_PLATFORMS", "OSX", "CHROME", "IOS", "ANDROID"],
}

INTEGRATION_NAME = "GoogleSafeBrowsing"
URL_OUTPUT_PREFIX = "GoogleSafeBrowsing.URL"


class Client(BaseClient):
    def __init__(self, proxy: bool, verify: bool, reliability: str, base_url: str, params: dict):
        headers = {"Content-Type": "application/json", "Accept": "application/json"}
        super().__init__(proxy=proxy, verify=verify, base_url=base_url, headers=headers)

        self.base_url = base_url
        self.client_body = {
            "clientId": params.get("api_creds", {}).get("identifier") or params.get("client_id"),
            "clientVersion": params.get("client_version"),
        }

        if DBotScoreReliability.is_valid_type(reliability):
            self.reliability = DBotScoreReliability.get_dbot_score_reliability_from_str(reliability)
        else:
            raise Exception("Google Safe Browsing v2 error: Please provide a valid value for the Source Reliability parameter.")

    def build_request_body(self, client_body: dict, list_url: List) -> dict:
        """build the request body according to the client body and the urls.

        Args:
            client_body: client body to add it in the request body
            list_url: The urls list
        Returns:
            (dict) The request body, in the right format.
        """
        list_urls = []
        for url in list_url:
            list_urls.append({"url": url})

        body: dict = {
            "client": client_body,
            "threatInfo": {
                "threatTypes": TYPES.get("threatTypes"),
                "platformTypes": TYPES.get("platformTypes"),
                "threatEntryTypes": ["URL"],
                "threatEntries": list_urls,
            },
        }
        return body

    def url_request(self, client_body, list_url) -> dict:
        """send the url request.

        Args:
            client_body: client body to add it in the request body
            list_url: The urls list
        Returns:
            (dict) The response from the request.
        """
        body = self.build_request_body(client_body, list_url)
        result = self._http_request(method="POST", json_data=body, full_url=self.base_url)
        return result


def test_module(client: Client) -> str:
    """
    Performs basic get request to get sample URL details.
    """
    try:
        # testing a known malicious URL to check if we get matches
        test_url = "http://testsafebrowsing.appspot.com/apiv4/ANY_PLATFORM/MALWARE/URL/"
        res = client.url_request(client.client_body, [test_url])
        if res.get("matches"):  # matches - There is a match for the URL we were looking for
            message = "ok"
        else:
            message = "Error querying Google Safe Browsing. Expected matching respons, but received none"
    except DemistoException as e:
        if "Forbidden" in str(e) or "Authorization" in str(e):
            message = "Authorization Error: please make sure the API Key is set correctly."
        else:
            raise e
    return message


def handle_errors(result: dict) -> None:
    """
    Handle errors, raise Exception when there is errors in the response.
    """

    status_code = result.get("StatusCode", 0)
    result_body = result.get("Body")

    if result_body == "" and status_code == 204:
        raise Exception("No content received. Possible API rate limit reached.")

    if 200 < status_code < 299:
        raise Exception(f"Failed to perform request, request status code: {status_code}.")

    if result_body == "":
        raise Exception("No content received. Maybe you tried a private API?.")

    if result.get("error"):
        error_massage = result.get("error", {}).get("message")
        error_code = result.get("error", {}).get("code")
        raise Exception(f"Failed accessing Google Safe Browsing APIs. Error: {error_massage}. Error code: {error_code}")


def arrange_results_to_urls(results: List, url_list: List) -> dict:
    """Arrange and filter the URLs results according to the URLs list that we asked information on.
    Args:
        results: the API response.
        url_list: The URLs list that we asked information on.
    Returns:
        (dict) The results according the urls.
    """
    urls_results: dict[str, list] = {}
    for url in url_list:
        urls_results[url] = []

    for result in results:
        url = result.get("threat", {}).get("url")
        urls_results[url].append(result)

    return urls_results


def url_command(client: Client, args: dict[str, Any]) -> Union[List[CommandResults], CommandResults]:
    """
    url command: Returns URL details for a list of URL
    """

    url = argToList(args.get("url"))

    result = client.url_request(client.client_body, url)

    if not result:
        dbot_score = Common.DBotScore(
            indicator=url,
            indicator_type=DBotScoreType.URL,
            integration_name=INTEGRATION_NAME,
            score=0,
            reliability=client.reliability,
        )
        url_standard_context = Common.URL(url=url, dbot_score=dbot_score)
        return CommandResults(
            readable_output=f"No information was found for url {url}",
            outputs_prefix=URL_OUTPUT_PREFIX,
            outputs_key_field="IndicatorValue",
            outputs=result,
            indicator=url_standard_context,
        )

    if result.get("StatusCode"):
        handle_errors(result)

    urls_data = arrange_results_to_urls(result.get("matches"), url)  # type: ignore

    url_data_list = []
    for url_key, url_data in urls_data.items():
        if url_data:
            dbot_score = Common.DBotScore(
                indicator=url_key,
                indicator_type=DBotScoreType.URL,
                integration_name=INTEGRATION_NAME,
                score=3,
                reliability=client.reliability,
            )
            url_standard_context = Common.URL(url=url_key, dbot_score=dbot_score)
            url_data_list.append(
                CommandResults(
                    readable_output=tableToMarkdown(f"Google Safe Browsing APIs - URL Query: {url_key}", url_data),
                    outputs_prefix=URL_OUTPUT_PREFIX,
                    outputs_key_field="IndicatorValue",
                    outputs=url_data,
                    indicator=url_standard_context,
                )
            )
        else:
            dbot_score = Common.DBotScore(
                indicator=url_key,
                indicator_type=DBotScoreType.URL,
                integration_name=INTEGRATION_NAME,
                score=0,
                reliability=client.reliability,
            )
            url_standard_context = Common.URL(url=url_key, dbot_score=dbot_score)
            url_data_list.append(
                CommandResults(
                    readable_output=f"No matches for URL {url_key}",
                    outputs_prefix=URL_OUTPUT_PREFIX,
                    outputs_key_field="IndicatorValue",
                    outputs=result,
                    indicator=url_standard_context,
                )
            )

    return url_data_list


def build_base_url(params: dict) -> str:
    api_key = params.get("api_creds", {}).get("password") or params.get("api_key")
    base_url = params.get("url", "")

    if not base_url.endswith("/"):
        base_url += "/"

    return f"{base_url}?key={api_key}"


def main() -> None:
    params = demisto.params()

    verify_certificate = not params.get("insecure", False)
    proxy = params.get("proxy", False)

    base_url = build_base_url(params)

    reliability = params.get("integrationReliability")
    reliability = reliability if reliability else DBotScoreReliability.B

    demisto.debug(f"Command being called is {demisto.command()}")
    try:
        client = Client(params=params, base_url=base_url, verify=verify_certificate, proxy=proxy, reliability=reliability)

        if demisto.command() == "test-module":
            result = test_module(client)
            return_results(result)

        elif demisto.command() == "url":
            return_results(url_command(client, demisto.args()))

    # Log exceptions and return errors
    except Exception as e:
        return_error(f"Failed to execute {demisto.command()} command.\nError:\n{e!s}")


""" ENTRY POINT """

if __name__ in ("__main__", "__builtin__", "builtins"):
    main()