AkamaiProlexic

"""Akamai Prolexic Event Collector for Cortex XSIAM.

Fetches DDoS-related events from the Akamai Prolexic Analytics API and forwards
them to Cortex XSIAM as a single ``akamai_prolexic_raw`` dataset
(vendor=``akamai``, product=``prolexic``).

Two event sources are supported:

* **Critical Events** — ``GET /prolexic-analytics/v2/critical-events/contract/{contract}``
* **Events**          — ``GET /prolexic-analytics/v2/events/contract/{contract}?extended=true``

Authentication is Akamai EdgeGrid HMAC-SHA-256, performed by the
``akamai-edgegrid`` library and attached to every HTTP request via the
``requests`` ``auth`` mechanism.
"""

import traceback
from collections.abc import Iterable
from datetime import UTC, datetime
from typing import Any

import demistomock as demisto  # noqa: F401
import urllib3
from akamai.edgegrid import EdgeGridAuth
from CommonServerPython import *  # noqa: F401

urllib3.disable_warnings()

# --------------------------------------------------------------------------- #
# Constants
# --------------------------------------------------------------------------- #

VENDOR = "akamai"
PRODUCT = "prolexic"

CRITICAL_EVENTS = "Critical Events"
EVENTS = "Events"

# Per-source configuration — exact endpoints/timestamp fields come from the
# Akamai Prolexic Analytics design document (CIAC-16080).
#
# ``revision_field`` names a secondary timestamp that changes when the record is
# updated in place. It is folded into the dedup key so that an updated record is
# treated as a distinct entry rather than a duplicate of the original.
SOURCE_CONFIG: dict[str, dict[str, str]] = {
    CRITICAL_EVENTS: {
        "endpoint_template": "/prolexic-analytics/v2/critical-events/contract/{contract}",
        "time_field": "firstOccur",
        "revision_field": "recentOccur",
        "last_run_key": "critical_events",
    },
    EVENTS: {
        "endpoint_template": "/prolexic-analytics/v2/events/contract/{contract}",
        "time_field": "eventStartTime",
        "last_run_key": "events",
    },
}

# Default to "now" so the first fetch does not back-fill historical events
# (matches the XSIAM Event-Collector convention used by OnePassword and others).
DEFAULT_FIRST_FETCH = "now"
DEFAULT_MAX_EVENTS_PER_FETCH = 1000
MAX_EVENTS_PER_FETCH_CEILING = 10000
DEFAULT_GET_EVENTS_LIMIT = 50
# Upper bound on how many dedup ids are carried in ``last_run``. Only ids sharing
# the high-water-mark timestamp are retained, but a source that emits a large
# burst at one timestamp could still grow the set without limit.
MAX_RETAINED_DEDUP_IDS = 5000
ISO_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"


# --------------------------------------------------------------------------- #
# Client
# --------------------------------------------------------------------------- #


class Client(BaseClient):
    """Thin wrapper around ``BaseClient`` that issues EdgeGrid-signed requests
    to the Akamai Prolexic Analytics API.

    ``BaseClient`` is intentionally used (instead of ``ContentClient``) because
    the Akamai EdgeGrid library ships a ``requests.auth.AuthBase`` implementation
    that signs each call synchronously via ``self._session.auth`` — this is
    natively supported by ``BaseClient`` but is not directly compatible with
    the ``httpx``-based ``ContentClient``. This mirrors the pattern used by
    the existing ``Akamai_SIEM`` pack.
    """

    def __init__(
        self,
        base_url: str,
        verify: bool,
        proxy: bool,
        client_token: str,
        client_secret: str,
        access_token: str,
        account_switch_key: str | None = None,
    ) -> None:
        super().__init__(base_url=base_url, verify=verify, proxy=proxy)
        # ``EdgeGridAuth`` is a ``requests.auth.AuthBase`` implementation that
        # signs the request with HMAC-SHA-256 immediately before it is sent.
        self._session.auth = EdgeGridAuth(
            client_token=client_token,
            client_secret=client_secret,
            access_token=access_token,
        )
        self._account_switch_key = account_switch_key or None

    def _build_params(self, extra: dict[str, Any] | None = None) -> dict[str, Any]:
        """Always merge in ``accountSwitchKey`` when configured."""
        params: dict[str, Any] = dict(extra or {})
        if self._account_switch_key:
            params["accountSwitchKey"] = self._account_switch_key
        return params

    def get_critical_events(self, contract_id: str) -> dict[str, Any]:
        """Calls ``GET /v2/critical-events/contract/{contract}``."""
        url_suffix = SOURCE_CONFIG[CRITICAL_EVENTS]["endpoint_template"].format(contract=contract_id)
        return self._http_request(
            method="GET",
            url_suffix=url_suffix,
            params=self._build_params(),
            headers={"Accept": "application/json"},
        )

    def get_events(self, contract_id: str) -> dict[str, Any]:
        """Calls ``GET /v2/events/contract/{contract}?extended=true``."""
        url_suffix = SOURCE_CONFIG[EVENTS]["endpoint_template"].format(contract=contract_id)
        return self._http_request(
            method="GET",
            url_suffix=url_suffix,
            params=self._build_params({"extended": "true"}),
            headers={"Accept": "application/json"},
        )


# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #


def _parse_positive_int_arg(raw: Any, display_name: str, default: int, ceiling: int | None = None) -> int:
    """Parse a user-supplied positive integer, rejecting invalid values loudly.

    Distinguishes between "not supplied" (use ``default``) and "explicitly
    0/negative" (reject). The ``arg_to_number(...) or DEFAULT`` idiom silently
    rewrites ``0`` to the default, masking misconfiguration.
    """
    if raw in (None, ""):
        return default
    parsed = arg_to_number(raw)
    if parsed is None:
        raise DemistoException(f"{display_name} must be an integer; got {raw!r}.")
    if parsed <= 0:
        raise DemistoException(f"{display_name} must be greater than 0; got {parsed}.")
    if ceiling is not None and parsed > ceiling:
        raise DemistoException(f"{display_name} must be between 1 and {ceiling}; got {parsed}.")
    return parsed


def parse_first_fetch(first_fetch: str) -> str:
    """Convert the human-friendly ``first_fetch`` parameter into an ISO string.

    Falls back to :data:`DEFAULT_FIRST_FETCH` if the value is empty/invalid.
    """
    parsed: datetime | None = arg_to_datetime(first_fetch or DEFAULT_FIRST_FETCH, required=False)
    if parsed is None:
        parsed = arg_to_datetime(DEFAULT_FIRST_FETCH, required=True)  # type: ignore[assignment]
    assert parsed is not None  # for mypy
    return parsed.astimezone(UTC).strftime(ISO_FORMAT)


def normalize_event_timestamp(value: Any) -> str | None:
    """Normalise a raw timestamp value (string, int, float) to ISO-8601 UTC.

    Returns ``None`` if the value cannot be parsed.
    """
    if value is None or value == "":
        return None
    try:
        parsed = arg_to_datetime(value, required=False)
    except (ValueError, TypeError):
        return None
    if parsed is None:
        return None
    return parsed.astimezone(UTC).strftime(ISO_FORMAT)


def _event_is_at_or_before(event: dict[str, Any], end_dt: datetime) -> bool:
    """Return ``True`` when the enriched event's ``_time`` is at or before ``end_dt``."""
    event_dt = arg_to_datetime(event.get("_time"))
    return event_dt is None or event_dt <= end_dt


def make_event_id(event_type: str, raw: dict[str, Any], time_field: str) -> str:
    """Build a stable dedup key.

    The Prolexic API does not return a single canonical ``id`` field, so we
    use the documented timestamp field together with any provided ``id``-like
    field to form a composite key. This is robust against re-emission of the
    same record across overlapping fetch windows.

    For sources that declare a ``revision_field`` (Critical Events use
    ``recentOccur``) that value is appended to the key. Prolexic updates a
    critical event in place while leaving ``firstOccur`` untouched, so without
    the revision component every subsequent update would collide with the
    original key and be discarded as a duplicate.
    """
    candidate_id_fields = ("id", "eventId", "incidentId", "alertId", "uuid")
    raw_id = next((str(raw[k]) for k in candidate_id_fields if raw.get(k) is not None), "")
    raw_time = str(raw.get(time_field, ""))
    key = f"{event_type}:{raw_id}:{raw_time}"

    revision_field = SOURCE_CONFIG.get(event_type, {}).get("revision_field")
    if revision_field:
        key = f"{key}:{raw.get(revision_field, '')!s}"
    return key


def annotate_critical_event(event: dict[str, Any]) -> None:
    """Add the ``_ENTRY_STATUS`` field per the design doc.

    * ``_ENTRY_STATUS = "new"``     when ``recentOccur == firstOccur``
    * ``_ENTRY_STATUS = "updated"`` when ``recentOccur >  firstOccur``
    """
    first_occur = event.get("firstOccur")
    recent_occur = event.get("recentOccur")
    if first_occur is None:
        return
    if recent_occur is None or recent_occur == first_occur:
        event["_ENTRY_STATUS"] = "new"
    else:
        event["_ENTRY_STATUS"] = "updated"


def extract_event_list(response: dict[str, Any]) -> list[dict[str, Any]]:
    """Extract the list of event dicts from a Prolexic API response.

    The Akamai Prolexic Analytics API typically returns a wrapper dict whose
    payload key varies (``events``, ``criticalEvents``, ``data``, ``items``).
    We probe each candidate and fall back to a top-level list if present.
    """
    if isinstance(response, list):
        return [e for e in response if isinstance(e, dict)]
    if not isinstance(response, dict):
        return []
    for key in ("events", "criticalEvents", "data", "items", "results"):
        value = response.get(key)
        if isinstance(value, list):
            return [e for e in value if isinstance(e, dict)]
    return []


def filter_and_dedup(
    raw_events: list[dict[str, Any]],
    event_type: str,
    last_fetch_iso: str,
    fetched_ids: set[str],
    max_events: int,
) -> tuple[list[dict[str, Any]], str, set[str]]:
    """Filter out previously-seen events and cap by ``max_events``.

    Returns the selected events, the new high-water-mark timestamp, and the
    set of dedup ids that share that high-water-mark (to be persisted in
    ``last_run`` so the next fetch can skip them even if the API re-emits them
    at exactly the same timestamp).

    Note: The returned id-set is computed defensively. When the high-water
    mark does NOT advance (i.e. ``new_high_water == last_fetch_iso``) we
    UNION with the input ``fetched_ids`` so that previously-seen ids at the
    same boundary timestamp survive into the next ``last_run``. Replacing
    them would risk re-ingestion if the API re-emitted the same event.
    The union is capped at :data:`MAX_RETAINED_DEDUP_IDS` so a source that
    keeps emitting new records at a single timestamp cannot grow ``last_run``
    without bound.
    """
    time_field = SOURCE_CONFIG[event_type]["time_field"]
    selected: list[dict[str, Any]] = []
    last_fetch_dt = arg_to_datetime(last_fetch_iso)
    new_high_water = last_fetch_iso
    new_high_water_dt = last_fetch_dt
    skipped_invalid_ts = 0
    skipped_old = 0
    skipped_seen = 0

    # Normalise once up front: the raw timestamp may be an ISO string or epoch
    # millis, and sorting those by ``str()`` is not chronological (e.g.
    # "999999999999" sorts after "1767225600000"). Because the loop below stops
    # at ``max_events`` and only ever moves the cursor forward, a mis-ordered
    # walk would skip past events that are then never fetched again.
    normalized_pairs: list[tuple[datetime, str, dict[str, Any]]] = []
    for raw in raw_events:
        normalized_ts = normalize_event_timestamp(raw.get(time_field))
        event_dt = arg_to_datetime(normalized_ts) if normalized_ts is not None else None
        if normalized_ts is None or event_dt is None:
            skipped_invalid_ts += 1
            continue
        normalized_pairs.append((event_dt, normalized_ts, raw))

    # Sort ascending by the parsed datetime so we walk the time window forward.
    for event_dt, normalized_ts, raw in sorted(normalized_pairs, key=lambda pair: pair[0]):
        if last_fetch_dt is not None and event_dt < last_fetch_dt:
            skipped_old += 1
            continue

        dedup_id = make_event_id(event_type, raw, time_field)
        if dedup_id in fetched_ids:
            skipped_seen += 1
            continue

        enriched: dict[str, Any] = dict(raw)
        enriched["_time"] = normalized_ts
        enriched["event_type"] = event_type
        enriched["source_log_type"] = event_type.upper().replace(" ", "_")
        if event_type == CRITICAL_EVENTS:
            annotate_critical_event(enriched)

        selected.append(enriched)
        fetched_ids.add(dedup_id)

        if new_high_water_dt is None or event_dt > new_high_water_dt:
            new_high_water_dt = event_dt
            new_high_water = normalized_ts

        if len(selected) >= max_events:
            break

    if skipped_invalid_ts or skipped_old or skipped_seen:
        demisto.debug(
            f"{event_type}: skipped {skipped_invalid_ts} invalid-timestamp, "
            f"{skipped_old} pre-cursor and {skipped_seen} previously-seen events."
        )
    if len(selected) >= max_events:
        demisto.debug(f"Reached max_events={max_events} for {event_type}; truncating.")

    # Compute retained ids for next run.
    new_run_ids = {make_event_id(event_type, ev, time_field) for ev in selected if ev.get("_time") == new_high_water}
    if new_high_water == last_fetch_iso:
        # Cursor did not advance — keep prior boundary ids so they cannot
        # be re-ingested next run.
        retained_ids = fetched_ids | new_run_ids
    else:
        retained_ids = new_run_ids

    if len(retained_ids) > MAX_RETAINED_DEDUP_IDS:
        # Keep the newest ids (deterministically ordered) and drop the oldest.
        # Trimming can only ever cause a re-fetch of an already-delivered event,
        # never a permanent loss, so bounding state is the safer trade-off.
        demisto.debug(
            f"{event_type}: retained dedup ids ({len(retained_ids)}) exceeded "
            f"{MAX_RETAINED_DEDUP_IDS}; trimming the oldest entries."
        )
        retained_ids = set(sorted(retained_ids)[-MAX_RETAINED_DEDUP_IDS:])
    return selected, new_high_water, retained_ids


# --------------------------------------------------------------------------- #
# Source dispatch
# --------------------------------------------------------------------------- #


def fetch_source_events(
    client: Client,
    event_type: str,
    contract_id: str,
    source_last_run: dict[str, Any],
    max_events: int,
    first_fetch_iso: str,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Fetch + dedup events for a single source. Returns ``(events, next_run)``."""
    last_fetch_iso = source_last_run.get("last_fetch_ts") or first_fetch_iso
    fetched_ids = set(source_last_run.get("fetched_ids") or [])

    if event_type == CRITICAL_EVENTS:
        response = client.get_critical_events(contract_id)
    elif event_type == EVENTS:
        response = client.get_events(contract_id)
    else:
        raise DemistoException(f"Unsupported event type: {event_type}")

    raw_events = extract_event_list(response)
    demisto.debug(f"Fetched {len(raw_events)} raw {event_type} from Prolexic API.")

    events, new_high_water, retained_ids = filter_and_dedup(
        raw_events=raw_events,
        event_type=event_type,
        last_fetch_iso=last_fetch_iso,
        fetched_ids=fetched_ids,
        max_events=max_events,
    )
    next_run = {"last_fetch_ts": new_high_water, "fetched_ids": sorted(retained_ids)}
    return events, next_run


def push_events(events: list[dict[str, Any]]) -> None:
    """Push events to XSIAM with the documented vendor/product."""
    if not events:
        demisto.debug("No events to push to XSIAM.")
        return
    demisto.debug(f"Pushing {len(events)} events to XSIAM as {VENDOR}/{PRODUCT}.")
    send_events_to_xsiam(events=events, vendor=VENDOR, product=PRODUCT)


# --------------------------------------------------------------------------- #
# Commands
# --------------------------------------------------------------------------- #


def run_test_module(client: Client, contract_id: str, event_types: Iterable[str]) -> str:
    """Run a lightweight call against each configured source to confirm
    credentials and contract id.

    Named ``run_test_module`` (not ``test_module``) so that pytest does not
    collect it as a test case.
    """
    selected_types = list(event_types) or [CRITICAL_EVENTS, EVENTS]
    for event_type in selected_types:
        if event_type not in SOURCE_CONFIG:
            return f"Unknown event type configured: {event_type}"
        try:
            if event_type == CRITICAL_EVENTS:
                client.get_critical_events(contract_id)
            else:
                client.get_events(contract_id)
        except DemistoException as exc:
            # Use duck-typing on ``exc.res`` so that any object exposing
            # ``status_code`` (real ``requests.Response`` or stub) works.
            status = getattr(getattr(exc, "res", None), "status_code", None)
            if status in (401, 403):
                return (
                    "Authorization Error: verify the EdgeGrid Client Token, Client Secret, "
                    "Access Token and that the API client has access to the configured Contract ID."
                )
            if status == 404:
                return "Endpoint not found: verify the Server URL and Contract ID."
            raise
    return "ok"


def get_events_command(
    client: Client,
    args: dict[str, Any],
    contract_id: str,
    configured_types: list[str],
    first_fetch_iso: str,
) -> tuple[list[dict[str, Any]], CommandResults]:
    """Manual fetch (``akamai-prolexic-get-events``) used for development."""
    limit = _parse_positive_int_arg(args.get("limit"), "limit", DEFAULT_GET_EVENTS_LIMIT)
    requested_types = argToList(args.get("event_type")) or configured_types or [CRITICAL_EVENTS, EVENTS]

    # ``start_time`` lets the caller override the lower-bound timestamp used
    # for client-side filtering. If omitted we fall back to ``first_fetch``.
    start_time_arg = args.get("start_time")
    if start_time_arg:
        start_iso = parse_first_fetch(str(start_time_arg))
    else:
        start_iso = first_fetch_iso

    # ``end_time`` is the standard upper bound for the manual command. The
    # Prolexic endpoints do not accept a time window, so it is applied
    # client-side after normalisation.
    end_time_arg = args.get("end_time")
    end_dt = arg_to_datetime(end_time_arg) if end_time_arg else None
    start_dt = arg_to_datetime(start_iso)
    if end_dt is not None and start_dt is not None and end_dt < start_dt:
        raise DemistoException("end_time must be later than start_time.")

    all_events: list[dict[str, Any]] = []
    for event_type in requested_types:
        events, _ = fetch_source_events(
            client=client,
            event_type=event_type,
            contract_id=contract_id,
            source_last_run={"last_fetch_ts": start_iso, "fetched_ids": []},
            max_events=limit,
            first_fetch_iso=start_iso,
        )
        if end_dt is not None:
            events = [ev for ev in events if _event_is_at_or_before(ev, end_dt)]
        all_events.extend(events)

    human_readable = tableToMarkdown(
        name=f"{VENDOR.title()} {PRODUCT.title()} Events",
        t=all_events,
        removeNull=True,
    )
    return all_events, CommandResults(readable_output=human_readable)


def fetch_events(
    client: Client,
    contract_id: str,
    event_types: list[str],
    max_events_per_fetch: int,
    first_fetch_iso: str,
    last_run: dict[str, Any],
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """Top-level fetch. Returns ``(events, next_run)``."""
    next_run: dict[str, Any] = dict(last_run or {})
    all_events: list[dict[str, Any]] = []
    for event_type in event_types:
        source_key = SOURCE_CONFIG[event_type]["last_run_key"]
        source_last_run = (last_run or {}).get(source_key, {})
        events, source_next_run = fetch_source_events(
            client=client,
            event_type=event_type,
            contract_id=contract_id,
            source_last_run=source_last_run,
            max_events=max_events_per_fetch,
            first_fetch_iso=first_fetch_iso,
        )
        all_events.extend(events)
        next_run[source_key] = source_next_run
    return all_events, next_run


# --------------------------------------------------------------------------- #
# Entry point
# --------------------------------------------------------------------------- #


def _parse_max_events_per_fetch(raw: Any) -> int:
    """Parse and validate the ``max_events_per_fetch`` parameter."""
    return _parse_positive_int_arg(
        raw,
        "Maximum events per fetch",
        DEFAULT_MAX_EVENTS_PER_FETCH,
        ceiling=MAX_EVENTS_PER_FETCH_CEILING,
    )


def main() -> None:  # pragma: no cover
    params = demisto.params()
    args = demisto.args()
    command = demisto.command()

    base_url: str = (params.get("url") or "").rstrip("/")
    contract_id: str = (params.get("contract_id") or "").strip()
    client_token: str = (params.get("client_token_creds") or {}).get("password", "") or params.get("client_token", "")
    client_secret: str = (params.get("client_secret_creds") or {}).get("password", "") or params.get("client_secret", "")
    access_token: str = (params.get("access_token_creds") or {}).get("password", "") or params.get("access_token", "")
    account_switch_key: str = (params.get("account_switch_key") or "").strip() or ""

    event_types: list[str] = argToList(params.get("event_types_to_fetch")) or [CRITICAL_EVENTS, EVENTS]

    # Collect every configuration problem up front so the user sees all of them
    # at once, and report through a single ``return_error`` call (W9011).
    config_errors: list[str] = []
    unsupported = [event_type for event_type in event_types if event_type not in SOURCE_CONFIG]
    if unsupported:
        config_errors.append(f"Unsupported event type(s) configured: {', '.join(unsupported)}.")
    if not base_url:
        config_errors.append("Server URL is required.")
    if not contract_id:
        config_errors.append("Contract ID is required.")
    if not (client_token and client_secret and access_token):
        config_errors.append("Client Token, Client Secret and Access Token are all required.")
    if config_errors:
        return_error(" ".join(config_errors))
        return

    first_fetch_iso = parse_first_fetch(params.get("first_fetch") or DEFAULT_FIRST_FETCH)
    verify_certificate = not params.get("insecure", False)
    proxy = params.get("proxy", False)

    demisto.debug(f"Command being called is {command!r}")
    try:
        max_events_per_fetch = _parse_max_events_per_fetch(params.get("max_events_per_fetch"))

        client = Client(
            base_url=base_url,
            verify=verify_certificate,
            proxy=proxy,
            client_token=client_token,
            client_secret=client_secret,
            access_token=access_token,
            account_switch_key=account_switch_key,
        )

        if command == "test-module":
            return_results(run_test_module(client, contract_id, event_types))
            return

        if command == "akamai-prolexic-get-events":
            should_push = argToBoolean(args.get("should_push_events", "false"))
            events, results = get_events_command(
                client=client,
                args=args,
                contract_id=contract_id,
                configured_types=event_types,
                first_fetch_iso=first_fetch_iso,
            )
            return_results(results)
            if should_push:
                push_events(events)
            return

        if command == "fetch-events":
            last_run = demisto.getLastRun() or {}
            events, next_run = fetch_events(
                client=client,
                contract_id=contract_id,
                event_types=event_types,
                max_events_per_fetch=max_events_per_fetch,
                first_fetch_iso=first_fetch_iso,
                last_run=last_run,
            )
            push_events(events)
            demisto.setLastRun(next_run)
            return

        raise NotImplementedError(f"Command {command!r} is not implemented.")

    except Exception as exc:  # noqa: BLE001
        demisto.error(traceback.format_exc())
        return_error(f"Failed to execute {command!r} command. Error: {exc!s}")


if __name__ in ("__main__", "__builtin__", "builtins"):  # pragma: no cover
    main()