LastInfoSec

from typing import Any
from CommonServerPython import *  # noqa # pylint: disable=unused-wildcard-import
from CommonServerUserPython import *  # noqa
import urllib3

# Disable insecure warnings
urllib3.disable_warnings()

INDICATOR_TYPE_TO_DBOT_SCORE = {
    "FILE": DBotScoreType.FILE,
    "URL": DBotScoreType.URL,
    "DOMAIN": DBotScoreType.DOMAIN,
}

INTEGRATION_NAME = "LastInfoSec"


class GwAPIException(Exception):
    """A base class from which all other exceptions inherit.

    If you want to catch all errors that the gwapi_benedictine package might raise,
    catch this base exception.
    """


class GwRequests:
    """Allows to easily interact with HTTP server.

    Class features:
      - Get requests package wrapper.
      - Put requests package wrapper.
      - Post requests package wrapper.
      - Delete requests package wrapper.
    """

    PROXIES = {"http": "", "https": ""}

    def __init__(self, token: str, headers: dict = {}, check_cert: bool = False, proxy: bool = False) -> None:
        """Init.

        Disable urllib3 warning. Allow unsecure ciphers.

        Args:
            check_cert: True to validate server certificate and False instead.
            proxies: Requests proxies. Default to no proxies.
        """
        self.url = "api.client.lastinfosec.com/v2"
        self.token = token
        self.headers = headers
        self.check_cert = check_cert
        if proxy:
            self.PROXIES["http"] = os.getenv("http_proxy", "")
            self.PROXIES["https"] = os.getenv("https_proxy", "")

    def _gen_request_kwargs(
        self,
        endpoint: str,
        data: dict,
        json_data: dict,
        params: dict,
        headers: dict,
        cookies: dict,
        redirects: bool,
        files: dict = None,
    ) -> dict:
        """Generate requests arguments.

        Args:
            endpoint: URL endpoint in format /XX/YY/ZZ.
            data: request data.
            json_data: Set to True if data is in json_data format and False instead.
            params: Set to True if data need to be send with the url and False instead.
            headers: Set to True if redirection is allowed and False instead.
            cookies: Set to True if redirection is allowed and False instead.
            redirects: Set to True if redirection is allowed and False instead.
            files: files to upload in multipart/form-data

        Returns:
            Return requests arguments in dictionnary format.
        """
        kwargs = {
            "url": f"https://{self.url}{endpoint}",
            "headers": headers if headers else self.headers,
            "cookies": cookies,
            "verify": self.check_cert,
            "proxies": self.PROXIES,
            "allow_redirects": redirects,
            "data": data,
            "json": json_data,
            "params": params,
            "files": files,
        }
        return kwargs

    def _get(
        self,
        endpoint: str,
        data: dict = None,
        json_data: dict = None,
        params: dict = None,
        headers: dict = None,
        cookies: dict = None,
        redirects: bool = True,
    ) -> requests.Response:
        """Wrap the get requests.

        Same arguments as _gen_request_kwargs functions.

        Returns:
            Return a requests object with properties:

            - status_code
            - reason
            - headers
            - text
        """
        kwargs = self._gen_request_kwargs(
            endpoint=endpoint,
            data=data,  # type: ignore
            json_data=json_data,  # type: ignore
            params=params,  # type: ignore
            headers=headers,  # type: ignore
            cookies=cookies,  # type: ignore
            redirects=redirects,
        )
        return requests.get(**kwargs)

    def _post(
        self,
        endpoint: str,
        data: dict = None,
        json_data: dict = None,
        params: dict = None,
        headers: dict = None,
        cookies: dict = None,
        redirects: bool = True,
        files: dict = None,
    ) -> requests.Response:
        """Wrap the post requests.

        Same arguments as _gen_request_kwargs functions.

        Returns:
            Return a requests object with properties:

            - status_code
            - reason
            - headers
            - text
        """
        kwargs = self._gen_request_kwargs(
            endpoint=endpoint,
            data=data,  # type: ignore
            json_data=json_data,  # type: ignore
            params=params,  # type: ignore
            headers=headers,  # type: ignore
            cookies=cookies,  # type: ignore
            redirects=redirects,
            files=files,
        )
        return requests.post(**kwargs)


class GwClient(GwRequests):
    """Client class to interact with the service API."""

    def test_module(self):
        """Return True if status_code == 200 and False instead.

        Raises:
            GwAPIException: If status_code != 200.
        """
        response = self._get(endpoint=f"/lis/getbyminutes/0?api_key={self.token}&headers=false")
        if response.status_code == 200:
            demisto.info("Get healthchecks on LIS API: [OK]")
            return True
        else:
            demisto.error("Get healthchecks on LIS API: [FAILED]", response.text, response.status_code, response.reason)
            return False

    def get_by_minute(self, minute) -> list:
        """Retrieve the data from Gatewatcher CTI feed by minute.
            Max 1440 minutes.

        Args:
            minute: Number of minutes to get. (max 1440 min)

        Returns:
            LIS Json response

        Raises:
            GwAPIException: If status_code != 200.
        """

        response = self._get(endpoint=f"/lis/getbyminutes/{arg_to_number(minute)}?api_key={self.token}&headers=false")
        if response.status_code == 200:
            demisto.info("Get ioc by minute : [OK]")
            return response.json()
        else:
            raise GwAPIException("Get ioc by minute: [FAILED]", response.text, response.status_code, response.reason)

    def get_by_value(self, value: str) -> dict:
        """Allows you to search for an IOC (url, hash, host) or a vulnerability in the Gatewatcher CTI database.
            If the data is known, only the IOC corresponding to the value will be returned.

        Args:
            value: Value to be search

        Returns:
            Value data

        Raises:
            GwAPIException: If status_code != 200.
        """
        response = self._post(endpoint=f"/lis/search?api_key={self.token}&headers=false", json_data=assign_params(value=value))
        if response.status_code == 200:
            demisto.info("Get search ioc: [OK]")
            return response.json()
        else:
            raise GwAPIException(
                "Get ioc by value: [FAILED]",
                response.text,
                response.status_code,
                response.reason,
            )

    def get_leaked_email_by_domain(self, domain: str, after: str) -> dict:
        """Allows you to search for leaked email by domain in Gatewatcher's CTI database.

        Args:
            domain: Domain to be searched
            after: results before this date won't be returned.

        Returns:
            Value data

        Raises:
            GwAPIException: If status_code != 200.
        """
        url = f"/lis/leaked_emails/get_by_domain/{domain}?api_key={self.token}&headers=false"
        url += f"&added_after={after}" if after else ""

        response = self._get(endpoint=url)

        if response.status_code == 200:
            demisto.info("Get search leaked email: [OK]")
            return response.json()
        else:
            raise GwAPIException(
                "Get leaked email: [FAILED]",
                response.text,
                response.status_code,
                response.reason,
            )

    def get_is_email_leaked(self, email: str, after: str) -> dict:
        """Allows you to search if an email has leaked in Gatewatcher's CTI database.

        Args:
            email: Email to be searched.
            after: results before this date won't be returned.

        Returns:
            Value data

        Raises:
            GwAPIException: If status_code != 200.
        """
        url = f"/lis/leaked_emails/get_by_email/{email}?api_key={self.token}&headers=false"
        url += f"&added_after={after}" if after else ""

        response = self._get(
            endpoint=url,
            json_data=assign_params(value=email),
        )
        if response.status_code == 200:
            demisto.info("Get search leaked email: [OK]")
            return response.json()
        else:
            raise GwAPIException(
                "Get is email leaked: [FAILED]",
                response.text,
                response.status_code,
                response.reason,
            )


def get_dbot_score(risk: str) -> int:
    if risk == "Malicious":
        return Common.DBotScore.BAD
    if risk == "Suspicious":
        return Common.DBotScore.SUSPICIOUS
    if risk == "High suspicious":
        return Common.DBotScore.SUSPICIOUS
    return Common.DBotScore.NONE


def file_indicator(ioc: dict):
    md5 = None
    sha1 = None
    sha256 = None
    sha512 = None

    hash = ioc["Value"]
    hash_type = ioc["Type"]

    if hash_type == "MD5":
        md5 = hash
    elif hash_type == "SHA1":
        sha1 = hash
    elif hash_type == "SHA256":
        sha256 = hash
    elif hash_type == "SHA512":
        sha512 = hash

    return Common.File(
        dbot_score=get_dbot_score(ioc.get("Risk", "")),
        description=ioc.get("Description"),
        ssdeep=ioc.get("MetaData", {}).get("ssdeep"),
        file_type=ioc.get("MetaData", {}).get("filetype"),
        traffic_light_protocol=ioc.get("TLP"),
        imphash=ioc.get("MetaData", {}).get("imphash"),
        md5=md5,
        sha1=sha1,
        sha256=sha256,
        sha512=sha512,
    )


def domain_indicator(ioc: dict):
    return Common.Domain(
        domain=ioc.get("Value"),
        dbot_score=get_dbot_score(ioc.get("Risk", "")),
        description=ioc.get("Description"),
        traffic_light_protocol=ioc.get("TLP"),
    )


def url_indicator(ioc: dict):
    return Common.URL(
        url=ioc.get("Value"),
        dbot_score=get_dbot_score(ioc.get("Risk", "")),
        description=ioc.get("Description"),
        traffic_light_protocol=ioc.get("TLP"),
    )


def generic_reputation_command(client: GwClient, args: dict, cmd_type: str, reliability: str) -> List[CommandResults]:
    """Checks the reputation of a file, domain or url

    Args:
        client: Client to interact with the LIS API.
        args: Command arguments.
        cmd_type: Command type ("file", "domain" or "url")
        reliability: Integration reliability

    Return
        List of CommandResults objects.
    """
    output_prefixes = {
        "file": "LIS.File",
        "domain": "LIS.Domain",
        "url": "LIS.URL",
    }
    arg_list = argToList(args[cmd_type])
    results: List[CommandResults] = []
    indicator_type = cmd_type

    # for each IOC in request args
    for arg in arg_list:
        response = client.get_by_value(value=arg)
        ioc = list(filter(lambda x: x["Value"] == arg, response["IOCs"]))[0]

        lis_result = {
            "Value": ioc["Value"],
            "Risk": ioc["Risk"],
            "Categories": ioc["Categories"],
            "Type": ioc["Type"],
            "TLP": ioc["TLP"],
            "UsageMode": ioc["UsageMode"],
            "Vulnerabilities": ioc["Vulnerabilities"],
        }

        readable_result = tableToMarkdown("Get IoC corresponding to the value", lis_result)

        if cmd_type == "file":
            indicator = file_indicator(ioc)
        elif cmd_type == "domain":
            indicator = domain_indicator(ioc)
        else:
            indicator = url_indicator(ioc)

        indicator.dbot_score = Common.DBotScore(
            indicator=arg,
            integration_name=INTEGRATION_NAME,
            indicator_type=indicator_type,
            score=get_dbot_score(ioc["Risk"]),
            reliability=reliability,
            malicious_description="Match found in LastInfoSec",
        )

        results.append(
            CommandResults(
                indicator=indicator,
                readable_output=readable_result,
                outputs_prefix=output_prefixes[cmd_type],
                outputs_key_field="Value",
                outputs=lis_result,
                raw_response=response,
            )
        )
    return results


def test_module(client: GwClient) -> str:  # noqa: E501
    """tests API connectivity.

    Args:
        client: Client to interact with the LIS API.

    Returns:
        'Request successful' when the LIS connection works.
        'Request error' when the LIS connection doesn't works.
    """
    if client.test_module():
        return "ok"
    else:
        return "Request error, please check ip/user/password/token: [ERROR]"


def lis_get_by_minute(client: GwClient, args: dict[Any, Any]) -> CommandResults:  # noqa: E501
    """Retrieve the data from Gatewatcher CTI feed by minute.
        Max 1440 minutes.

    Args:
        client: Client to interact with the LIS API.
        args: Command arguments.

    Returns:
        CommandResults object with the "LIS.IoC.GetByMinute" prefix.
    """
    minute = arg_to_number(args.get("Minute"))
    categories = argToList(args.get("Categories"))
    risk = argToList(args.get("Risk"))
    tlp = argToList(args.get("TLP"))
    ftype = argToList(args.get("Type"))
    mode = args.get("Mode")

    if "Filename" in ftype:
        raise ValueError("Filter filename is not a valid filter. Please use a different filter option.")
    if "Informational" in risk:
        raise ValueError("Filter informational is not a valid filter. Please use a different filter option.")

    response = client.get_by_minute(minute=minute)

    for case in response:
        if categories:
            case["IOCs"] = [ioc for ioc in case["IOCs"] if set(categories) & set(ioc["Categories"])]
        if ftype:
            case["IOCs"] = [ioc for ioc in case["IOCs"] if ioc["Type"] in ftype]
        if mode:
            case["IOCs"] = [ioc for ioc in case["IOCs"] if mode == ioc["UsageMode"]]
        if risk:
            case["IOCs"] = [ioc for ioc in case["IOCs"] if ioc["Risk"] in risk]
        if tlp:
            case["IOCs"] = [ioc for ioc in case["IOCs"] if ioc["TLP"] in tlp]
    result = [ioc["Value"] for case in response for ioc in case["IOCs"]]

    readable_result = tableToMarkdown("Get IoC by minute", result, headers="Value")
    return CommandResults(
        readable_output=readable_result,
        outputs_prefix="LIS.GetByMinute",
        outputs_key_field="Value",
        outputs=result,
        raw_response=result,
    )


def lis_get_by_value(client: GwClient, args: dict[Any, Any]) -> CommandResults:  # noqa: E501
    """Allows you to search for an IOC (url, hash, host) or a vulnerability in the Gatewatcher CTI database.
        If the data is known, only the IOC corresponding to the value will be returned.

    Args:
        client: Client to interact with the LIS API.
        args: Command arguments.

    Returns:
        CommandResults object with the "LIS.IoC.GetByValue" prefix.
    """
    value = args["Value"]
    response = client.get_by_value(value=value)

    ioc = list(filter(lambda x: x["Value"] == value, response["IOCs"]))[0]
    result = {
        "Value": ioc["Value"],
        "Risk": ioc["Risk"],
        "Categories": ioc["Categories"],
        "Type": ioc["Type"],
        "TLP": ioc["TLP"],
        "UsageMode": ioc["UsageMode"],
        "Vulnerabilities": ioc["Vulnerabilities"],
    }

    readable_result = tableToMarkdown("Get IoC corresponding to the value", result)
    return CommandResults(
        readable_output=readable_result,
        outputs_prefix="LIS.GetByValue",
        outputs_key_field="Value",
        outputs=result,
        raw_response=result,
    )


def lis_get_leaked_email_by_domain(client: GwClient, args: dict[Any, Any]) -> CommandResults:
    """Allows you to search for leaked email by domain in Gatewatcher's CTI database.

    Args:
        client: Client to interact with the LIS API.
        args: Command arguments,
            Domain: domain to search for.
            After: date to do not return results before this date.

    Returns:
        CommandResults object with the "LIS.leakedEmail.GetByDomain" prefix.
    """
    domain = args.get("Domain", None)
    after = args.get("After", None)
    response = client.get_leaked_email_by_domain(domain, after)

    emails = [res["Value"] for res in response]
    result = emails if len(emails) > 0 else None
    readable_result = tableToMarkdown("Leaked email", result, headers="Emails")

    return CommandResults(
        readable_output=readable_result,
        outputs_prefix="LIS.LeakedEmail.GetByDomain",
        outputs=result,
        raw_response=result,
    )


def lis_is_email_leaked(client: GwClient, args: dict[Any, Any]) -> CommandResults:
    """Allows you to search if an email has leaked in Gatewatcher's CTI database.

    Args:
        client: Client to interact with the LIS API.
        args: Command arguments,
            Email: email to search for.
            After: date to do not return results before this date.

    Returns:
        CommandResults object with the "LIS.leakedEmail.getByEmail" prefix.
    """
    email = args.get("Email", None)
    after = args.get("After", None)
    response = client.get_is_email_leaked(email, after)

    result = email if len(response) > 0 else None
    readable_result = tableToMarkdown("Is email leaked", result, headers="Value")

    return CommandResults(
        readable_output=readable_result,
        outputs_prefix="LIS.LeakedEmail.GetByEmail",
        outputs=result,
        raw_response=result,
    )


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

    token = params.get("token")
    check_cert = params.get("check_cert", False)
    proxy = params.get("proxy", False)
    reliability = params.get("integrationReliability", "C - Fairly reliable")
    reliability = reliability if reliability else DBotScoreReliability.B

    if DBotScoreReliability.is_valid_type(reliability):
        reliability = DBotScoreReliability.get_dbot_score_reliability_from_str(reliability)
    else:
        return_error("Please provide a valid value for the Source Reliability parameter")

    demisto.debug(f"Command being called is {command}")
    try:
        client = GwClient(token=token, proxy=proxy, check_cert=check_cert)
        if command == "test-module":
            return_results(test_module(client=client))
        elif command == "url":
            return_results(generic_reputation_command(client=client, args=args, cmd_type="url", reliability=reliability))
        elif command == "file":
            return_results(generic_reputation_command(client=client, args=args, cmd_type="file", reliability=reliability))
        elif command == "domain":
            return_results(generic_reputation_command(client=client, args=args, cmd_type="domain", reliability=reliability))
        elif command == "gw-lis-get-by-minute":
            return_results(lis_get_by_minute(client=client, args=args))
        elif command == "gw-lis-get-by-value":
            return_results(lis_get_by_value(client=client, args=args))
        elif command == "gw-lis-leaked-email-by-domain":
            return_results(lis_get_leaked_email_by_domain(client=client, args=args))
        elif command == "gw-lis-is-email-leaked":
            return_results(lis_is_email_leaked(client=client, args=args))
        else:
            raise NotImplementedError(f"{command} command is not implemented.")
    except Exception as e:
        return_error(f"Failed to execute {command} command.\nError:\n{str(e)}")


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