USTA Stolen Credit Cards

from typing import Any

import demistomock as demisto  # noqa: F401
import urllib3
from CommonServerPython import *  # noqa: F401

from CommonServerUserPython import *  # noqa

# Disable insecure warnings
urllib3.disable_warnings()

USTA_API_PREFIX = "api/threat-stream/v4/"

DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"  # ISO8601 format with UTC, default in XSOAR

# !TODO: Will be changed when API team released new filters.
USTA_TICKET_STATUSES = {
    "all": None,
    "in_progress": "in_progress",
    "open": "open",
    "closed": "closed",
    "out of scope": "out_of_scope",
    "passive": "passive",
}

MAX_ALERTS_TO_FETCH = 100


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

    def check_auth(self):
        self._http_request("GET", "company/me", error_handler=self._http_error_handler)

    def stolen_credit_cards_incidents(self, **kwargs) -> list:
        params = assign_params(**kwargs)
        headers = self._headers

        demisto.debug(f"stolen_credit_cards_incidents: {params}")

        response = self._http_request("GET", "fraud-intelligence/credit-card-tickets", params=params, headers=headers)
        count = response.get("count", 0)
        next_url = response.get("next", None)
        results = response.get("results", [])

        demisto.debug(f"stolen_credit_cards_incidents: Fetched {count} results")

        while next_url:
            demisto.debug(f"stolen_credit_cards_incidents: Fetching next page: {next_url}")
            response_next = self._http_request("GET", full_url=next_url, headers=headers)
            results += response_next.get("results", [])
            next_url = response_next.get("next", None)
        return results

    def stolen_credit_cards_search_api_request(self, **kwargs) -> dict:
        params = assign_params(**kwargs)
        headers = self._headers
        return self._http_request("GET", "fraud-intelligence/credit-card-tickets", params=params, headers=headers)

    @staticmethod
    def _http_error_handler(response):
        # Handle error responses here to proper error messages to the user
        if response.status_code == 401:
            raise DemistoException("Authorization Error: make sure API Key is correctly set")
        if response.status_code == 429:
            raise DemistoException("Rate limit exceeded. Please try again later..!")


def check_module(client: Client):
    try:
        client.check_auth()
    except DemistoException as e:
        if "Connection Timeout Error" in str(e):
            return ValueError("Unable to connect to the USTA API! Make sure that your IP is whitelisted in the USTA.")
        raise e
    return "ok"


def convert_to_demisto_severity(severity: str) -> int:
    return {
        "low": IncidentSeverity.LOW,
        "medium": IncidentSeverity.MEDIUM,
        "high": IncidentSeverity.HIGH,
        "critical": IncidentSeverity.CRITICAL,
        "unknown": IncidentSeverity.UNKNOWN,
    }[severity]


def create_paging_header(results_num: int, page: int, size: int) -> str:
    header = f"Showing {results_num} results"
    if size is not None:
        header += f", Size={size}"
    if page is not None:
        header += f", from Page {page}"
    return header + "\n"


def fetch_incidents(
    client: Client, max_results: int, last_run: dict, first_fetch_time: str, status: Union[str, None] = None
) -> tuple[dict, list[dict]]:
    """Fetches the Stolen Credit Card incidents from the USTA API. If the last run is provided, it will fetch incidents
    from the last fetch time until now. Otherwise, it will fetch incidents from the first_fetch_time until now.

    Args:
        client (Client): USTA Stolen Credit Cards HTTP client
        status (Union[str, None]): The status of the ticket to fetch. If None, fetch all tickets.
        first_fetch_time (str): The first fetch time to fetch incidents from.
    """
    if last_fetch := last_run.get("last_fetch", None):
        first_fetch_time = last_fetch

    assert first_fetch_time

    last_ids: list[int] = last_run.get("last_ids", []) or []

    incidents: list[dict[str, Any]] = []

    alerts = client.stolen_credit_cards_incidents(status=status, start=first_fetch_time, size=max_results)
    demisto.debug(f"Received {len(alerts)} alerts from server.")

    # API returns the newest alerts first so instead of -1 we need to get the first alert's created time
    last_fetched_time = alerts[0]["created"] if alerts else last_fetch

    new_last_ids: list[int] = []

    for alert in alerts:
        # skip the alerts which are already fetched and it is always sorted by created field.
        if alert["created"] == last_fetched_time:
            new_last_ids.append(alert["id"])

        if alert["id"] in last_ids:
            demisto.debug(f"Skipping already fetched alert: {alert['id']}")
            continue

        severity = "medium"
        ticket_id = alert.get("id")

        incident = {
            "name": f"USTA Stolen Credit Card: USTA Ticket ID : {ticket_id}",
            "occurred": alert.get("created"),
            "severity": convert_to_demisto_severity(severity),
            "rawJSON": json.dumps(alert),
        }
        incidents.append(incident)

    demisto.debug(f"setting next run- {last_fetched_time=}")
    next_run = {"last_fetch": last_fetched_time, "last_ids": new_last_ids}

    return next_run, incidents


def stolen_credit_cards_search_command(client: Client, args: dict) -> CommandResults:
    card_number = args.get("card_number", None)
    size = args.get("page_size", None)
    page = args.get("page", None)

    if not card_number:
        raise ValueError("Please provide a credit card number to search for.")

    if results := client.stolen_credit_cards_search_api_request(card_number=card_number, page=page, size=size):
        readable_output = create_paging_header(
            results_num=results.get("count", 0),
            page=page,
            size=size,
        ) + tableToMarkdown("Stolen Credit Cards", results.get("results", []))
        return CommandResults(
            outputs_prefix="USTA.StolenCreditCards", outputs_key_field="id", outputs=results, readable_output=readable_output
        )
    return CommandResults(readable_output="No results found.")


def main() -> None:
    # demisto params and args
    params: dict[str, Any] = demisto.params()
    args: dict[str, Any] = demisto.args()

    # Instance parameters
    verify_certificate: bool = not params.get("insecure", False)
    base_url = urljoin(params["url"], USTA_API_PREFIX)
    proxy = params.get("proxy", False)
    api_key = params.get("api_key")
    cmd = demisto.command()

    # How much time before the first fetch to retrieve alerts
    first_fetch_time = arg_to_datetime(arg=params.get("first_fetch", "1 days"), arg_name="First fetch time", required=True)
    assert first_fetch_time

    demisto.debug(f"Command being called is {demisto.command()}")
    try:
        headers: dict = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}

        client = Client(base_url=base_url, verify=verify_certificate, headers=headers, proxy=proxy)

        commands = {"usta-scc-search": stolen_credit_cards_search_command}

        if cmd == "test-module":
            return_results(check_module(client))
        elif cmd == "fetch-incidents":
            status = USTA_TICKET_STATUSES.get(params.get("status", "Open").lower(), "None")
            max_results = arg_to_number(arg=params.get("max_fetch"), arg_name="max_fetch", required=False)
            if not max_results or max_results > MAX_ALERTS_TO_FETCH:
                max_results = MAX_ALERTS_TO_FETCH
            next_run, incidents = fetch_incidents(
                client=client,
                max_results=100,
                last_run=demisto.getLastRun(),
                first_fetch_time=datetime.strftime(first_fetch_time, DATE_FORMAT),
                status=status,
            )
            demisto.incidents(incidents)
            demisto.setLastRun(next_run)
        elif cmd in commands:
            return_results(commands[cmd](client, args))

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


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