European Union Vulnerability Database

import demistomock as demisto
from CommonServerPython import *


class Client(BaseClient):
    def __init__(self, server_url, verify, proxy, headers=None):
        super().__init__(base_url=server_url, verify=verify, proxy=proxy, headers=headers)

    def get_advisory_by_id_request(self, id_):
        params = assign_params(id=id_)

        response = self._http_request("get", "/advisory", params=params)

        return response

    def get_by_enisa_id_request(self, id_):
        params = assign_params(id=id_)

        response = self._http_request("get", "/enisaid", params=params)

        return response

    def get_latest_critical_vulnerabilities_request(self):
        response = self._http_request("get", "/criticalvulnerabilities")

        return response

    def get_latest_exploited_vulnerabilities_request(self):
        response = self._http_request("get", "/exploitedvulnerabilities")

        return response

    def get_latest_vulnerabilities_request(self):
        response = self._http_request("get", "/lastvulnerabilities")

        return response

    def get_vulnerability_by_id_request(self, id_):
        params = assign_params(id=id_)

        response = self._http_request("get", "/vulnerability", params=params)

        return response

    def query_vulnerabilities_request(
        self, from_score, to_score, from_epss, to_epss, from_date, to_date, product, vendor, assigner, exploited, page, text, size
    ):
        params = assign_params(
            from_score=from_score,
            to_score=to_score,
            from_epss=from_epss,
            to_epss=to_epss,
            from_date=from_date,
            to_date=to_date,
            product=product,
            vendor=vendor,
            assigner=assigner,
            exploited=exploited,
            page=page,
            text=text,
            size=size,
        )

        response = self._http_request("get", "/vulnerabilities", params=params)

        return response


def get_advisory_by_id_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    advisory_id = args.get("advisory_id", None)
    if not advisory_id:
        raise ValueError("The 'advisory_id' argument is required.")

    response = client.get_advisory_by_id_request(advisory_id)
    command_results = CommandResults(
        outputs_prefix="EUVD.Advisory",
        outputs_key_field="id",
        outputs=response,
        raw_response=response,
        readable_output=tableToMarkdown(f"Advisory {advisory_id}", response, headerTransform=pascalToSpace),
    )

    return command_results


def get_vulnerability_by_enisa_id_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    enisa_id = args.get("enisa_id", None)
    if not enisa_id:
        raise ValueError("The 'enisa_id' argument is required.")

    response = client.get_by_enisa_id_request(enisa_id)
    command_results = CommandResults(
        outputs_prefix="EUVD.Vulnerability",
        outputs_key_field="id",
        outputs=response,
        raw_response=response,
        readable_output=tableToMarkdown(f"Vulnerability {enisa_id}", response, headerTransform=pascalToSpace),
    )

    return command_results


def get_latest_critical_vulnerabilities_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    response = client.get_latest_critical_vulnerabilities_request()
    command_results = CommandResults(
        outputs_prefix="EUVD.Vulnerability",
        outputs_key_field="id",
        outputs=response,
        raw_response=response,
        readable_output=tableToMarkdown("Latest Critical Vulnerabilities", response, headerTransform=pascalToSpace),
    )

    return command_results


def get_latest_exploited_vulnerabilities_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    response = client.get_latest_exploited_vulnerabilities_request()
    command_results = CommandResults(
        outputs_prefix="EUVD.Vulnerability",
        outputs_key_field="id",
        outputs=response,
        raw_response=response,
        readable_output=tableToMarkdown("Latest Exploited Vulnerabilities", response, headerTransform=pascalToSpace),
    )

    return command_results


def get_latest_vulnerabilities_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    response = client.get_latest_vulnerabilities_request()
    command_results = CommandResults(
        outputs_prefix="EUVD.Vulnerability",
        outputs_key_field="id",
        outputs=response,
        raw_response=response,
        readable_output=tableToMarkdown("Latest Vulnerabilities", response, headerTransform=pascalToSpace),
    )

    return command_results


def get_vulnerability_by_id_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    vulnerability_id = args.get("vulnerability_id", None)
    if not vulnerability_id:
        raise ValueError("The 'vulnerability_id' argument is required.")

    response = client.get_vulnerability_by_id_request(vulnerability_id)
    command_results = CommandResults(
        outputs_prefix="EUVD.Vulnerability",
        outputs_key_field="id",
        outputs=response,
        raw_response=response,
        readable_output=tableToMarkdown(f"Vulnerability {vulnerability_id}", response, headerTransform=pascalToSpace),
    )

    return command_results


def query_vulnerabilities_command(client: Client, args: Dict[str, Any]) -> CommandResults:
    from_score = str(args.get("from_score", ""))
    to_score = str(args.get("to_score", ""))
    from_epss = str(args.get("from_epss", ""))
    to_epss = str(args.get("to_epss", ""))
    from_date = str(args.get("from_date", ""))
    to_date = str(args.get("to_date", ""))
    product = str(args.get("product", ""))
    vendor = str(args.get("vendor", ""))
    assigner = str(args.get("assigner", ""))
    exploited = argToBoolean(args.get("exploited", False))
    page = args.get("page", None)
    text = str(args.get("text", ""))
    size = args.get("size", None)

    response = client.query_vulnerabilities_request(
        from_score, to_score, from_epss, to_epss, from_date, to_date, product, vendor, assigner, exploited, page, text, size
    )
    command_results = CommandResults(
        outputs_prefix="EUVD.Vulnerability",
        outputs_key_field="id",
        outputs=response.get("items"),
        raw_response=response,
        readable_output=tableToMarkdown("Vulnerabilities", response.get("items"), headerTransform=pascalToSpace),
    )

    return command_results


def test_module(client: Client) -> str:
    try:
        client.get_latest_critical_vulnerabilities_request()
    except Exception as e:
        raise e
    # If the request was successful, return ok
    return "ok"


def main() -> None:
    params: Dict[str, Any] = demisto.params()
    args: Dict[str, Any] = demisto.args()
    url = urljoin(params.get("url"), "/api")
    verify_certificate: bool = not params.get("insecure", False)
    proxy = params.get("proxy", False)
    headers = {"Accept": "application/json", "User-Agent": "Cortex XSOAR"}

    command = demisto.command()
    demisto.debug(f"Command being called is {command}")

    try:
        disable_warnings()
        client: Client = Client(url, verify_certificate, proxy, headers=headers)

        commands = {
            "euvd-get-advisory-by-id": get_advisory_by_id_command,
            "euvd-get-vulnerability-by-enisa-id": get_vulnerability_by_enisa_id_command,
            "euvd-get-latest-critical-vulnerabilities": get_latest_critical_vulnerabilities_command,
            "euvd-get-latest-exploited-vulnerabilities": get_latest_exploited_vulnerabilities_command,
            "euvd-get-latest-vulnerabilities": get_latest_vulnerabilities_command,
            "euvd-get-vulnerability-by-id": get_vulnerability_by_id_command,
            "euvd-query-vulnerabilities": query_vulnerabilities_command,
        }

        if command == "test-module":
            return_results(test_module(client))
        elif command in commands:
            return_results(commands[command](client, args))
        else:
            raise NotImplementedError(f"{command} command is not implemented.")

    except Exception as e:
        return_error(str(e))


if __name__ in ["__main__", "builtin", "builtins"]:
    main()