Shodan_v2

import demistomock as demisto
from CommonServerPython import *

from CommonServerUserPython import *

""" IMPORTS """

import json

import requests
import urllib3

# Disable insecure warnings
urllib3.disable_warnings()

""" GLOBALS/PARAMS """
API_KEY = demisto.params().get("credentials", {}).get("password") or demisto.params().get("api_key")

if API_KEY is None:
    raise ValueError("Missing API key.")

# Remove trailing slash to prevent wrong URL path to service
API_URL = demisto.params()["api_url"].rstrip("/")

# Should we use SSL
USE_SSL = not demisto.params().get("insecure", False)

VENDOR = "shodan"
PRODUCT = "banner"
DEFAULT_MAX_EVENTS = 50_000
DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"

handle_proxy()

optional_general_properties_map: dict[str, tuple[str, Any]] = {
    "cpe23": ("CPE23", []),
    "device": ("Device", ""),
    "devicetype": ("DeviceType", ""),
    "info": ("Info", ""),
    "ipv6": ("IPv6", ""),
    "link": ("Link", ""),
    "platform": ("Platform", ""),
    "product": ("Product", ""),
    "tags": ("Tags", []),
    "transport": ("Transport", ""),
    "vendor": ("Vendor", ""),
    "version": ("Version", ""),
}


ntlm_properties_map: dict[str, tuple[str, Any]] = {
    "dns_domain_name": ("DNSDomainName", ""),
    "dns_forest_name": ("DNSForestName", ""),
    "fqdn": ("FQDN", ""),
    "netbios_computer_name": ("NetBIOSComputerName", ""),
    "netbios_domain_name": ("NetBIOSDomainName", ""),
    "os": ("OS", []),
    "os_build": ("OSBuild", ""),
    "target_realm": ("TargetRealm", ""),
    "timestamp": ("Timestamp", 0),
}


ssl_cert_properties_map: dict[str, tuple[str, Any]] = {
    "expired": ("Expired", False),
    "expires": ("Expires", ""),
    "issued": ("Issued", ""),
    "issuer": ("Issuer", {}),
    "serial": ("Serial", 0),
    "sig_alg": ("SigAlg", ""),
    "subject": ("Subject", {}),
    "version": ("Version", 0),
}


""" HELPER FUNCTIONS """


def http_request(method, uri, params=None, data=None, headers=None):
    if params is None:
        params = {}

    params.update({"key": API_KEY})

    url = f"{API_URL}{uri}"
    res = requests.request(method, url, params=params, data=data, headers=headers, verify=USE_SSL)
    if res.status_code == 404:
        return {}
    if res.status_code == 401:
        return_error("Error: the Shodan API key is invalid. Please check your API key.")
    if res.status_code != 200:
        error_msg = f"Error in API call {url} [{res.status_code}] - {res.reason}"
        if "application/json" in res.headers["content-type"] and "error" in res.json():
            error_msg += f': {res.json()["error"]}'

        return_error(error_msg)

    return res.json()


def alert_to_demisto_result(alert):
    ec = {"Shodan": {"Alert": {"ID": alert.get("id", ""), "Expires": alert.get("expires", 0)}}}

    human_readable = tableToMarkdown(
        f'Alert ID {ec["Shodan"]["Alert"]["ID"]}',
        {
            "Name": alert.get("name", ""),
            "IP": alert.get("filters", {"ip": ""})["ip"],
            "Expires": ec["Shodan"]["Alert"]["Expires"],
        },
    )

    demisto.results(
        {
            "Type": entryTypes["note"],
            "Contents": alert,
            "ContentsFormat": formats["json"],
            "HumanReadable": human_readable,
            "HumanReadableFormat": formats["markdown"],
            "EntryContext": ec,
        }
    )


def format_record_keys(dict_list: List[Dict]) -> List[Dict]:
    """
    Formats dictionary keys by replacing underscores with spaces and capitalizing each word.
    """
    new_list = []
    for input_dict in dict_list:
        new_dict = {}
        for key, value in input_dict.items():
            new_key = key.replace("_", " ").title()
            new_dict[new_key] = value
        new_list.append(new_dict)
    return new_list


def add_time_to_events(events: list[dict]):
    """
    Adds the _time key to the events.
    Args:
        events: list[dict] - list of events to add the _time key to.
    Returns:
        list: The events with the _time key.
    """
    if events:
        for event in events:
            create_time = arg_to_datetime(event["created"])
            event["_time"] = create_time.strftime(DATE_FORMAT)  # type: ignore[union-attr]


def filter_events(events: list[dict], limit: int, last_run: dict = {}) -> list[dict]:
    """
    Filters and sorts events based on the last fetch time, list of excluded IDs, and a limit.

    Args:
        events (list[dict]): List of events where each event is represented as a dictionary.
        limit (int): The maximum number of events to return.
        last_run (dict, optional): Dictionary containing the last fetch time and a list of event IDs to exclude.
                                   Default is an empty dictionary.
    """

    if last_fetch_time := arg_to_datetime(last_run.get("last_fetch_time")):
        events = [event for event in events if parse_event_date(event) >= last_fetch_time]

        if last_ids := last_run.get("last_event_ids"):
            events = [event for event in events if event["id"] not in last_ids]

    return events[:limit]


def parse_event_date(event: Dict) -> datetime:
    """
    Parses the 'created' field from an event dictionary into a datetime object.
    """
    return datetime.strptime(event["created"], DATE_FORMAT)


""" COMMANDS + REQUESTS FUNCTIONS """


def get_scan_status(scan_id):
    res = http_request("GET", f"/shodan/scan/{scan_id}")

    ec = {"Shodan": {"Scan": {"ID": res.get("id", ""), "Status": res.get("status", "")}}}

    human_readable = tableToMarkdown(
        f"Scanning results for scan {scan_id}", {"ID": ec["Shodan"]["Scan"]["ID"], "Status": ec["Shodan"]["Scan"]["Status"]}
    )

    demisto.results(
        {
            "Type": entryTypes["note"],
            "Contents": res,
            "ContentsFormat": formats["json"],
            "HumanReadable": human_readable,
            "HumanReadableFormat": formats["markdown"],
            "EntryContext": ec,
        }
    )


def test_module():
    """
    Sends a basic GET request to verify API connectivity and performs a sample event fetch if event fetching is enabled.
    """
    params = demisto.params()
    is_fetch_events = argToBoolean(params.get("isFetchEvents", False))

    if is_fetch_events and not API_KEY:
        return_error("Missing API key - You must provide API KEY parameter.")

    if API_KEY:
        http_request("GET", "/shodan/alert/info")  # Checking with API key
    else:
        http_request("GET", "/shodan/ports", {"query": "test"})  # Checking without API key


def search_command():
    query = demisto.args()["query"]
    facets = demisto.args().get("facets")
    page = int(demisto.args().get("page", 1))
    return_json = argToBoolean(demisto.args().get("return_json", "No"))

    params = {"query": query}
    if facets:
        params["facets"] = facets
    if page:
        params["page"] = page

    res = http_request("GET", "/shodan/host/search", params)

    matches = res.get("matches", [])

    # Return the full list of match data as a JSON file if requested
    if return_json:
        formatted_json = json.dumps(matches).encode("utf-8")
        file_entry = fileResult(
            filename=f"shodan_search_matches_page{page}.json", data=formatted_json, file_type=EntryType.ENTRY_INFO_FILE
        )
        return_results(file_entry)

    for match in matches:
        location = match.get("location", {"city": "", "country_name": "", "longitude": 0, "latitude": 0})
        # Build context entry with commonly used properties and others listed as Required in Shodan Banner API spec
        ec = {
            "Shodan": {
                "Banner": {
                    "Org": match.get("org", ""),
                    "Isp": match.get("isp", ""),
                    "Transport": match.get("transport", ""),
                    "Asn": match.get("asn", ""),
                    "IP": match.get("ip_str", ""),
                    "Port": match.get("port", 0),
                    "Ssl": {"versions": match.get("ssl", {"versions": []}).get("versions", [])},
                    "Hostnames": match.get("hostnames", []),
                    "Location": {
                        "City": location["city"],
                        "Longitude": location["longitude"],
                        "Latitude": location["latitude"],
                        "Country": location["country_name"],
                    },
                    "Timestamp": match.get("timestamp", ""),
                    "Domains": match.get("domains", []),
                    "OS": match.get("os", ""),
                    "Data": match.get("data", ""),
                    "Hash": match.get("hash", 0),
                }
            }
        }

        # Add any available optional Banner properties to the entry context object
        for optional_property, (ec_key, default_value) in optional_general_properties_map.items():
            if optional_property in match:
                ec["Shodan"]["Banner"][ec_key] = match[optional_property] or default_value

        # Add SSL certificate properties, if present
        if match.get("ssl", {}).get("cert", {}):
            ec["Shodan"]["Banner"]["SslCert"] = {}

            for cert_property, (ec_key, default_value) in ssl_cert_properties_map.items():
                value = match["ssl"]["cert"].get(cert_property, default_value)
                if cert_property in ["issued", "expires"]:
                    cert_datetime = datetime.strptime(value, "%Y%m%d%H%M%SZ")
                    value = cert_datetime.strftime(DATE_FORMAT)

                ec["Shodan"]["Banner"]["SslCert"][ec_key] = value

        # Add NTLM properties, if present
        if match.get("ntlm", {}):
            ec["Shodan"]["Banner"]["Ntlm"] = {}

            for ntlm_property, (ec_key, default_value) in ntlm_properties_map.items():
                ec["Shodan"]["Banner"]["Ntlm"][ec_key] = match["ntlm"].get(ntlm_property, default_value)

        # Format returned data as markdown table
        human_readable = tableToMarkdown(
            f'Search results for query "{query}" - page {page}, facets: {facets}', flattenRow(rowDict=ec["Shodan"]["Banner"])
        )

        demisto.results(
            {
                "Type": entryTypes["note"],
                "Contents": match,
                "ContentsFormat": formats["json"],
                "HumanReadable": human_readable,
                "HumanReadableFormat": formats["markdown"],
                "EntryContext": ec,
            }
        )


def ip_command():
    ips = argToList(demisto.args()["ip"])
    results = []
    for ip in ips:
        res = http_request("GET", f"/shodan/host/{ip}")

        if not res:
            results.append(CommandResults(readable_output=f"No information available for the following IP: {ip}"))
        else:
            hostnames = res.get("hostnames")
            # It's a list, only if it exists and not empty we take the first value.
            hostname = hostnames[0] if hostnames else ""

            location = f'{round(res.get("latitude", 0.0), 3)},{round(res.get("longitude", 0.0), 3)}'

            relationships_list: list[EntityRelationship] = []

            vulns_list = res.get("vulns", [])
            for v in vulns_list:
                relationships_list.append(
                    EntityRelationship(
                        entity_a=ip,
                        entity_a_type=FeedIndicatorType.IP,
                        name="related-to",
                        entity_b=v,
                        entity_b_type=FeedIndicatorType.CVE,
                        brand="ShodanV2",
                    )
                )

            dbot_score = Common.DBotScore(
                indicator=ip,
                indicator_type=DBotScoreType.IP,
                reliability=demisto.params().get("integrationReliability"),
                score=0,
                integration_name="Shodan_v2",
            )

            ip_details = Common.IP(
                ip=ip,
                dbot_score=dbot_score,
                asn=res.get("asn", ""),
                hostname=hostname,
                geo_country=res.get("country_name", ""),
                geo_latitude=round(res.get("latitude", 0.0), 3),
                geo_longitude=round(res.get("longitude", 0.0), 3),
                relationships=relationships_list,
            )

            shodan_ip_details = {
                "Tag": res.get("tags", []),
                "Latitude": res.get("latitude", 0.0),
                "Longitude": res.get("longitude", 0.0),
                "Org": res.get("org", ""),
                "ASN": res.get("asn", ""),
                "ISP": res.get("isp", ""),
                "LastUpdate": res.get("last_update", ""),
                "CountryName": res.get("country_name", ""),
                "Address": ip,
                "OS": res.get("os", ""),
                "Port": res.get("ports", []),
                "Vulnerabilities": vulns_list,
            }

            title = f"Shodan details for IP {ip}"

            human_readable = {
                "Country": res.get("country_name", ""),
                "Location": location,
                "ASN": res.get("asn", ""),
                "ISP": res.get("isp", ""),
                "Ports": ", ".join([str(x) for x in res.get("ports", [])]),
                "Hostname": hostname,
            }

            readable_output = tableToMarkdown(name=title, t=human_readable, removeNull=True)

            results.append(
                CommandResults(
                    readable_output=readable_output,
                    raw_response=res,
                    outputs=shodan_ip_details,
                    relationships=relationships_list,
                    outputs_prefix="Shodan.IP",
                    indicator=ip_details,
                )
            )
    return results


def shodan_search_count_command():
    query = demisto.args()["query"]

    res = http_request("GET", "/shodan/host/count", {"query": query})

    ec = {"Shodan": {"Search": {"ResultCount": res.get("total", 0)}}}

    demisto.results(
        {
            "Type": entryTypes["note"],
            "Contents": res,
            "ContentsFormat": formats["json"],
            "HumanReadable": f'## {ec["Shodan"]["Search"]["ResultCount"]} results for query "{query}"',
            "HumanReadableFormat": formats["markdown"],
            "EntryContext": ec,
        }
    )


def shodan_scan_ip_command():
    ips = demisto.args()["ips"]

    res = http_request("POST", "/shodan/scan", data={"ips": ips})

    if "id" not in res:
        demisto.results(
            {
                "Type": entryTypes["error"],
                "Contents": res,
                "ContentsFormat": formats["json"],
                "HumanReadable": '## Unknown answer format, no "id" field in response',
                "HumanReadableFormat": formats["markdown"],
            }
        )

    get_scan_status(res["id"])


def shodan_scan_internet_command():
    port = demisto.args()["port"]

    try:
        port = int(port)
    except ValueError:
        return_error(f"Port must be number, not {port}")

    protocol = demisto.args()["protocol"]

    res = http_request("POST", "/shodan/scan/internet", data={"port": port, "protocol": protocol})

    ec = {"Shodan": {"Scan": {"ID": res.get("id", "")}}}

    human_readable = tableToMarkdown(
        f"Intenet scanning results for port {port} and protocol {protocol}",
        {
            "ID": ec["Shodan"]["Scan"]["ID"],
        },
    )

    demisto.results(
        {
            "Type": entryTypes["note"],
            "Contents": res,
            "ContentsFormat": formats["json"],
            "HumanReadable": human_readable,
            "HumanReadableFormat": formats["markdown"],
            "EntryContext": ec,
        }
    )


def shodan_scan_status_command():
    scan_id = demisto.args()["scanID"]

    get_scan_status(scan_id)


def shodan_create_network_alert_command():
    alert_name = demisto.args()["alertName"]
    ip = demisto.args()["ip"]
    expires = demisto.args().get("expires", 0)
    try:
        expires = int(expires)
    except ValueError:
        return_error(f"Expires must be a number, not {expires}")

    res = http_request(
        "POST",
        "/shodan/alert",
        data=json.dumps({"name": alert_name, "filters": {"ip": ip}, "expires": expires}),
        headers={"content-type": "application/json"},
    )

    alert_to_demisto_result(res)


def shodan_network_get_alert_by_id_command():
    alert_id = demisto.args()["alertID"]

    res = http_request("GET", f"/shodan/alert/{alert_id}/info")

    alert_to_demisto_result(res)


def shodan_network_get_alerts_command():
    res = http_request("GET", "/shodan/alert/info")

    if len(res) == 0:
        demisto.results("No alerts")
    else:
        for alert in res:
            alert_to_demisto_result(alert)


def shodan_network_delete_alert_command():
    alert_id = demisto.args()["alertID"]

    http_request("DELETE", f"/shodan/alert/{alert_id}")

    demisto.results(f"Deleted alert {alert_id}")


def shodan_network_alert_set_trigger_command():
    alert_id = demisto.args()["alertID"]
    trigger = demisto.args()["Trigger"]

    res = http_request("PUT", f"/shodan/alert/{alert_id}/trigger/{trigger}")

    if not res.get("success", False):
        return_error(f"Failed setting trigger {trigger} for alert {alert_id}")

    demisto.results(f'Set trigger "{trigger}" for alert {alert_id}')


def shodan_network_alert_remove_trigger_command():
    alert_id = demisto.args()["alertID"]
    trigger = demisto.args()["Trigger"]

    res = http_request("DELETE", f"/shodan/alert/{alert_id}/trigger/{trigger}")

    if not res.get("success", False):
        return_error(f"Failed deleting trigger {trigger} for alert {alert_id}")

    demisto.results(f'Deleted trigger "{trigger}" for alert {alert_id}')


def shodan_network_alert_whitelist_service_command():
    alert_id = demisto.args()["alertID"]
    trigger = demisto.args()["trigger"]
    service = demisto.args()["service"]

    res = http_request("PUT", f"/shodan/alert/{alert_id}/trigger/{trigger}/ignore/{service}")

    if not res.get("success", False):
        return_error(f'Failed whitelisting service "{service}" for trigger {trigger} in alert {alert_id}')

    demisto.results(f'Whitelisted service "{service}" for trigger {trigger} in alert {alert_id}')


def shodan_network_alert_remove_service_from_whitelist_command():
    alert_id = demisto.args()["alertID"]
    trigger = demisto.args()["trigger"]
    service = demisto.args()["service"]

    res = http_request("DELETE", f"/shodan/alert/{alert_id}/trigger/{trigger}/ignore/{service}")

    if not res.get("success", False):
        return_error(f'Failed removing service "{service}" for trigger {trigger} in alert {alert_id} from the whitelist')

    demisto.results(f'Removed service "{service}" for trigger {trigger} in alert {alert_id} from the whitelist')


def get_events_command(args: dict) -> tuple[str, list[dict]]:
    """
    Get events command, used mainly for debugging
    """
    events = http_request("GET", "/shodan/alert/info")
    if not isinstance(events, list):
        events = [events]

    limit = arg_to_number(args.get("max_fetch")) or DEFAULT_MAX_EVENTS
    events = filter_events(events, limit)

    hr = tableToMarkdown(f"{VENDOR.title()} - {PRODUCT.title()} Events:", format_record_keys(events))
    return hr, events


def fetch_events(last_run: dict, params: dict[str, str]) -> tuple[Dict, List[Dict]]:
    """
    Fetches events from an API, filters them, and updates the last_run data with the latest event's date.

    Args:
        last_run (dict): A dictionary containing data from the last run. It should include 'last_fetch_time'
                         and 'last_event_ids', which represent the last fetch time and IDs of the last events processed.
        params (dict[str, str]): Dictionary of parameters. It should include 'max_fetch' to define the maximum number
                                 of events to fetch.

    Returns:
        tuple[Dict, List[Dict]]: A tuple where the first item is the updated last_run data, including the latest fetch
                                 time and event IDs, and the second item is a list of filtered events.
    """
    if not last_run.get("last_fetch_time"):  # If this is a first run
        last_run = {"last_fetch_time": datetime.now().strftime(DATE_FORMAT)}
        demisto.debug("First run detected. Setting last_fetch_time to now.")
        return last_run, []

    events = http_request("GET", "/shodan/alert/info")
    if not isinstance(events, list):
        events = [events]
    demisto.debug(f"Fetched {len(events)} events before filtering")

    limit = arg_to_number(params.get("max_fetch")) or DEFAULT_MAX_EVENTS
    events_filtered = filter_events(events, limit, last_run)
    demisto.debug(f"After filtering, {len(events_filtered)} events remain")

    if events_filtered:
        latest_fetch_time = max(parse_event_date(event) for event in events_filtered)
        latest_event_ids = [event.get("id") for event in events_filtered if parse_event_date(event) == latest_fetch_time]

        last_run["last_fetch_time"] = latest_fetch_time.strftime(DATE_FORMAT)
        last_run["last_event_ids"] = latest_event_ids
    else:
        demisto.debug("No new events found after filtering")

    return last_run, events_filtered


""" COMMANDS MANAGER / SWITCH PANEL """

if demisto.command() == "test-module":
    # This is the call made when pressing the integration test button.
    test_module()
    demisto.results("ok")
elif demisto.command() == "search":
    search_command()
elif demisto.command() == "ip":
    return_results(ip_command())
elif demisto.command() == "shodan-search-count":
    shodan_search_count_command()
elif demisto.command() == "shodan-scan-ip":
    shodan_scan_ip_command()
elif demisto.command() == "shodan-scan-internet":
    shodan_scan_internet_command()
elif demisto.command() == "shodan-scan-status":
    shodan_scan_status_command()
elif demisto.command() == "shodan-create-network-alert":
    shodan_create_network_alert_command()
elif demisto.command() == "shodan-network-get-alert-by-id":
    shodan_network_get_alert_by_id_command()
elif demisto.command() == "shodan-network-get-alerts":
    shodan_network_get_alerts_command()
elif demisto.command() == "shodan-network-delete-alert":
    shodan_network_delete_alert_command()
elif demisto.command() == "shodan-network-alert-set-trigger":
    shodan_network_alert_set_trigger_command()
elif demisto.command() == "shodan-network-alert-remove-trigger":
    shodan_network_alert_remove_trigger_command()
elif demisto.command() == "shodan-network-alert-whitelist-service":
    shodan_network_alert_whitelist_service_command()
elif demisto.command() == "shodan-network-alert-remove-service-from-whitelist":
    shodan_network_alert_remove_service_from_whitelist_command()
elif demisto.command() == "shodan-get-events":
    args = demisto.args()
    hr, events = get_events_command(args)
    return_results(CommandResults(readable_output=hr))
    should_push_events = argToBoolean(args.get("should_push_events"))
    if should_push_events:
        add_time_to_events(events)
        send_events_to_xsiam(events, vendor=VENDOR, product=PRODUCT)
elif demisto.command() == "fetch-events":
    params = demisto.params()
    last_run = demisto.getLastRun()
    demisto.debug(f"Last_run before the fetch: {last_run}")
    next_run, events = fetch_events(last_run, params)

    add_time_to_events(events)
    send_events_to_xsiam(events=events, vendor=VENDOR, product=PRODUCT)
    demisto.debug(f"last_run after the fetch {last_run}")
    demisto.setLastRun(next_run)