AWS - Security Hub v2

import demistomock as demisto  # noqa: F401
from datetime import UTC
from CommonServerPython import *  # noqa: F401
from AWSApiModule import *  # noqa: E402
from botocore.client import BaseClient as BotoClient
from dateparser import parse


DEFAULT_RETRIES = 5
DEFAULT_FIRST_FETCH = "3 days"
DEFAULT_MAX_FETCH = 50
MAX_FETCH_LIMIT = 100  # AWS Security Hub V2 caps get_findings_v2 MaxResults at 100.
FETCH_SORT_CRITERIA = [{"Field": "finding_info.created_time_dt", "SortOrder": "asc"}]

# Maps the integration mirror-direction param to the value XSOAR stores on incidents.
MIRROR_DIRECTION_MAPPING = {
    "None": None,
    "Incoming": "In",
    "Outgoing": "Out",
    "Incoming And Outgoing": "Both",
}
# OCSF status_ids (https://schema.ocsf.io) that close the mirrored-in XSOAR incident,
# mapped to the XSOAR close reason. Resolved=4, Suppressed=3.
OCSF_STATUS_ID_TO_CLOSE_REASON = {4: "Resolved", 3: "Other"}
# Outgoing mapping schema surfaced by get-mapping-fields.
# "severity" = built-in XSOAR Severity field (0-4), translated to an OCSF SeverityId.
# "severityid" = raw OCSF severity_id (1-6, incl. Fatal=6); takes precedence over "severity".
OUTGOING_FIELD_DESCRIPTIONS = {
    "severityid": "The OCSF severity_id to set on the finding (1=Informational .. 6=Fatal).",
    "statusid": "The OCSF status_id to set on the finding (1=New, 2=In Progress, 3=Suppressed, 4=Resolved).",
    "comment": "A comment describing the reason for the update.",
    "severity": "The built-in incident severity; mirrored to the finding's OCSF severity in AWS.",
}

# OCSF severity_id -> XSOAR incident severity. Fatal (6) collapses to Critical (XSOAR has no higher).
OCSF_SEVERITY_ID_TO_XSOAR = {
    1: IncidentSeverity.INFO,
    2: IncidentSeverity.LOW,
    3: IncidentSeverity.MEDIUM,
    4: IncidentSeverity.HIGH,
    5: IncidentSeverity.CRITICAL,
    6: IncidentSeverity.CRITICAL,
}
# XSOAR incident severity -> OCSF severity_id. XSOAR Unknown (0) has no OCSF equivalent (skipped by caller).
XSOAR_SEVERITY_TO_OCSF_ID = {
    IncidentSeverity.INFO: 1,
    IncidentSeverity.LOW: 2,
    IncidentSeverity.MEDIUM: 3,
    IncidentSeverity.HIGH: 4,
    IncidentSeverity.CRITICAL: 5,
}


# Minimum severity label -> OCSF severity_id, used to build the fetch severity filter.
SEVERITY_LABEL_TO_OCSF_ID = {
    "Informational": 1,
    "Low": 2,
    "Medium": 3,
    "High": 4,
    "Critical": 5,
    "Fatal": 6,
}


# Drives the generic parse_filters helper. Per category: "fields" maps an entry key to
# (API Filter key, coercion callable); "required" lists mandatory entry keys; "defaults" supplies
# fallbacks; "require_any" (optional) requires at least one of the listed keys.
FILTER_CONFIGS: dict[str, dict] = {
    "string": {
        "fields": {"value": ("Value", str), "comparison": ("Comparison", str)},
        "required": ["value"],
        "defaults": {"comparison": "EQUALS"},
    },
    "number": {
        "fields": {
            "eq": ("Eq", arg_to_number),
            "gt": ("Gt", arg_to_number),
            "gte": ("Gte", arg_to_number),
            "lt": ("Lt", arg_to_number),
            "lte": ("Lte", arg_to_number),
        },
        "required": [],
        "defaults": {},
        "require_any": ["eq", "gt", "gte", "lt", "lte"],
    },
    "boolean": {
        "fields": {"value": ("Value", argToBoolean)},
        "required": ["value"],
        "defaults": {},
    },
    "map": {
        "fields": {"key": ("Key", str), "value": ("Value", str), "comparison": ("Comparison", str)},
        "required": ["key", "value"],
        "defaults": {"comparison": "EQUALS"},
    },
    "ip": {
        "fields": {"cidr": ("Cidr", str)},
        "required": ["cidr"],
        "defaults": {},
    },
}


""" HELPER FUNCTIONS """


def parse_filter_entries(filters_str: str) -> list[dict]:
    """Parse a filter argument (";"-separated entries of ","-separated key=value pairs) into dicts.

    Example:
        ``"field_name=severity,value=High;field_name=status,value=New,comparison=NOT_EQUALS"``
        parses to::

            [
                {"field_name": "severity", "value": "High"},
                {"field_name": "status", "value": "New", "comparison": "NOT_EQUALS"},
            ]

    Args:
        filters_str (str): The raw filter argument string.

    Returns:
        list[dict]: One key/value dict per entry.
    """
    entries = []
    for raw_entry in filters_str.split(";"):
        raw_entry = raw_entry.strip()
        if not raw_entry:
            continue
        entry = {}
        for pair in raw_entry.split(","):
            if "=" not in pair:
                continue
            key, _, value = pair.partition("=")
            entry[key.strip().lower()] = value.strip()
        if entry:
            entries.append(entry)
    return entries


def parse_filters(filters_str: str, category: str) -> list[dict]:
    """Parse a filter argument into the API ``{FieldName, Filter}`` structure using FILTER_CONFIGS.

    Backs the string/number/boolean/map/ip categories (date is handled by parse_date_filters).

    Example (``category="string"``):
        ``"field_name=severity,value=High,comparison=EQUALS"`` parses to::

            [{"FieldName": "severity", "Filter": {"Value": "High", "Comparison": "EQUALS"}}]

    Args:
        filters_str (str): The raw filter argument string.
        category (str): The FILTER_CONFIGS category key.

    Returns:
        list[dict]: A list of ``{FieldName, Filter}`` dictionaries.

    Raises:
        DemistoException: If an entry is missing ``field_name`` or any category-required field.
    """
    config = FILTER_CONFIGS[category]
    fields, required = config["fields"], config["required"]
    require_any = config.get("require_any")
    filters = []

    for entry in parse_filter_entries(filters_str):
        missing = []
        if not entry.get("field_name"):
            missing.append("field_name")
        missing.extend(key for key in required if not entry.get(key))
        if require_any and not any(entry.get(key) for key in require_any):
            missing.append(f"one of {require_any}")
        if missing:
            error_message = f"Invalid '{category}' filter entry {entry}: missing required field(s): {', '.join(missing)}."
            demisto.error(f"[AWS_Security_Hub_V2] {error_message}")
            raise DemistoException(error_message)

        field_name = entry["field_name"]
        merged = {**config["defaults"], **{key: value for key, value in entry.items() if key in fields}}
        api_filter = {fields[key][0]: fields[key][1](value) for key, value in merged.items()}
        filters.append({"FieldName": field_name, "Filter": api_filter})

    return filters


def parse_date_filters(filters_str: str) -> list[dict]:
    """Parse ``date_filters`` entries into the API ``DateFilters`` structure.

    Each entry uses exactly one form: absolute (``start`` + ``end``) or relative DateRange
    (``value`` with optional ``unit``/``comparison``; ``days`` is an alias for ``value`` + ``unit=DAYS``).

    Examples:
        Absolute range::

            "field_name=finding_info.created_time_dt,start=2024-01-01T00:00:00Z,end=2024-01-31T00:00:00Z"
            -> [{"FieldName": "finding_info.created_time_dt",
                 "Filter": {"Start": "2024-01-01T00:00:00Z", "End": "2024-01-31T00:00:00Z"}}]

        Relative DateRange (``days`` shorthand)::

            "field_name=finding_info.created_time_dt,days=7"
            -> [{"FieldName": "finding_info.created_time_dt",
                 "Filter": {"DateRange": {"Value": 7, "Unit": "DAYS"}}}]

    Args:
        filters_str (str): The raw date filters argument string.

    Returns:
        list[dict]: A list of ``{FieldName, Filter}`` dictionaries.

    Raises:
        DemistoException: If an entry mixes both forms, provides only one of ``start``/``end``, or neither.
    """
    filters = []
    for e in parse_filter_entries(filters_str):
        field_name = e.get("field_name")
        if not field_name:
            demisto.error(f"[AWS_Security_Hub_V2] Skipping date filter entry with no field_name: {e}")
            continue
        start, end = e.get("start"), e.get("end")
        # "days" is our own shorthand (documented in the date_filters arg) for "value" with Unit=DAYS; not an AWS field.
        value = e.get("value") or e.get("days")
        has_range = bool(value)
        has_absolute = bool(start or end)

        if has_range and has_absolute:
            raise DemistoException(
                f"Date filter for '{field_name}': use either the absolute form ('start'+'end') "
                f"or the relative 'DateRange' form ('value'/'days'), not both."
            )
        if has_range:
            date_range = {
                "Value": arg_to_number(value),
                "Unit": e.get("unit", "DAYS"),
                "Comparison": e.get("comparison"),
            }
            date_filter = {"DateRange": remove_empty_elements(date_range)}
        elif start and end:
            date_filter = {"Start": start, "End": end}
        else:
            raise DemistoException(
                f"Date filter for '{field_name}' requires either the relative 'DateRange' form "
                f"('value' with optional 'unit'/'comparison', or 'days'), or both 'start' and 'end'."
            )
        filters.append({"FieldName": field_name, "Filter": date_filter})
    return filters


def generate_filters_for_get_findings(args: dict) -> dict | None:
    """Build the composite ``Filters`` object for get-findings from the per-category filter arguments.

    Args:
        args (dict): Demisto command arguments.

    Returns:
        dict | None: The composite ``Filters`` structure, or ``None`` when no filters were supplied.
    """
    composite_filter: dict = remove_empty_elements(
        {
            "StringFilters": parse_filters(args.get("string_filters", ""), "string"),
            "DateFilters": parse_date_filters(args.get("date_filters", "")),
            "BooleanFilters": parse_filters(args.get("boolean_filters", ""), "boolean"),
            "NumberFilters": parse_filters(args.get("number_filters", ""), "number"),
            "MapFilters": parse_filters(args.get("map_filters", ""), "map"),
            "IpFilters": parse_filters(args.get("ip_filters", ""), "ip"),
        }
    )
    if not composite_filter:
        return None

    composite_filter["Operator"] = args.get("filter_operator", "AND")
    return {"CompositeFilters": [composite_filter]}


def parse_finding_identifiers(identifiers_str: str) -> list[dict]:
    """Parse ``finding_identifiers`` into the API ``FindingIdentifiers`` structure.

    Required keys per entry: ``cloud_account_uid``, ``finding_info_uid``, ``metadata_product_uid``.

    Args:
        identifiers_str (str): The raw finding identifiers argument string.

    Returns:
        list[dict]: A list of ``{CloudAccountUid, FindingInfoUid, MetadataProductUid}`` dictionaries.

    Raises:
        DemistoException: If an entry is missing any of the three required uid fields.
    """
    required_keys = ("cloud_account_uid", "finding_info_uid", "metadata_product_uid")
    identifiers = []
    for e in parse_filter_entries(identifiers_str):
        missing = [key for key in required_keys if not e.get(key)]
        if missing:
            error_message = f"Invalid finding identifier entry {e}: missing required field(s): {', '.join(missing)}."
            demisto.error(f"[AWS_Security_Hub_V2] {error_message}")
            raise DemistoException(error_message)
        identifiers.append(
            {
                "CloudAccountUid": e["cloud_account_uid"],
                "FindingInfoUid": e["finding_info_uid"],
                "MetadataProductUid": e["metadata_product_uid"],
            }
        )
    return identifiers


def build_fetch_filters(start_time: str, end_time: str, min_severity: str | None, additional_filters: str | None) -> dict:
    """Build the composite ``Filters`` object for the fetch loop.

    Filters on ``finding_info.created_time_dt`` within ``[start_time, end_time]``, AND-ed with an
    optional minimum-severity number filter and optional extra string filters.

    Args:
        start_time (str): ISO8601 inclusive lower bound of the fetch window.
        end_time (str): ISO8601 upper bound of the fetch window.
        min_severity (str | None): Minimum severity label (e.g. ``High``) to include.
        additional_filters (str | None): Extra ``string_filters``-formatted entries to AND into the query.

    Returns:
        dict: The composite ``Filters`` structure for ``get_findings_v2``.
    """
    composite_filter: dict = {
        "DateFilters": [
            {
                "FieldName": "finding_info.created_time_dt",
                "Filter": {"Start": start_time, "End": end_time},
            }
        ],
    }

    if min_severity and (severity_id := SEVERITY_LABEL_TO_OCSF_ID.get(min_severity)):
        composite_filter["NumberFilters"] = [{"FieldName": "severity_id", "Filter": {"Gte": severity_id}}]

    if additional_filters and (string_filters := parse_filters(additional_filters, "string")):
        composite_filter["StringFilters"] = string_filters

    composite_filter["Operator"] = "AND"
    # Only one composite filter is built, so CompositeOperator (which combines multiple composites) is omitted.
    return {"CompositeFilters": [composite_filter]}


def parse_tags(tags_str: str) -> dict:
    """Parse ``key=<key>,value=<value>`` pairs (``;``-separated) into a flat ``{key: value}`` tag mapping.

    Args:
        tags_str (str): The keys and values string.

    Returns:
        dict: A flat mapping suitable for the ``Tags`` API parameter.
    """
    regex = re.compile(r"key=([\w\d_:.-]+),value=([ /\w\d@_,.*-]+)", flags=re.I)
    return dict(regex.findall(tags_str))


def handle_client_error(e: Exception, context: str) -> None:
    """Log a boto3 ``ClientError`` with its code/message and re-raise it as a ``DemistoException``.

    Args:
        e (Exception): The caught ``ClientError`` (carries a ``response`` mapping).
        context (str): Short caller context for the debug log (e.g. ``"Fetch: page query"``).

    Raises:
        DemistoException: Always, wrapping the API error message.
    """
    error = getattr(e, "response", {}).get("Error", {})
    demisto.debug(
        f"[AWS_Security_Hub_V2] {context} raised {type(e).__name__} "
        f"(Code={error.get('Code', '')}, Message={error.get('Message', '')})."
    )
    raise DemistoException(error.get("Message", ""))


def filter_new_findings(findings: list, last_fetch: str, fetched_ids: list) -> list:
    """Drop findings already seen against the previous fetch boundary (inclusive ``Start``).

    A finding is dropped if it was created before ``last_fetch`` (stale) or exactly at ``last_fetch``
    with its uid in ``fetched_ids`` (already ingested).

    Args:
        findings (list): Raw OCSF findings returned by ``get_findings_v2``.
        last_fetch (str): ISO8601 boundary timestamp from the previous run.
        fetched_ids (list): Uids already ingested at the ``last_fetch`` boundary.

    Returns:
        list: The findings that have not been ingested yet.
    """
    new_findings: list = []
    for finding in findings:
        created_time = (finding.get("finding_info") or {}).get("created_time_dt")
        uid = finding.get("metadata", {}).get("uid")

        if created_time and last_fetch and created_time < last_fetch:
            demisto.debug(
                f"[AWS_Security_Hub_V2] Dedup: skipping STALE finding uid={uid} (created={created_time} < Start={last_fetch})."
            )
            continue
        if created_time and last_fetch and created_time == last_fetch and uid in fetched_ids:
            demisto.debug(
                f"[AWS_Security_Hub_V2] Dedup: skipping ALREADY-SEEN boundary finding uid={uid} (created={created_time})."
            )
            continue

        new_findings.append(finding)

    return new_findings


def findings_to_incidents(findings: list, mirror_direction: str | None = None) -> list:
    """Build XSOAR incidents from OCSF findings, stamping mirroring metadata when enabled.

    Args:
        findings (list): The (already deduped) OCSF findings to convert.
        mirror_direction (str | None): XSOAR mirror direction to stamp on findings, or ``None`` to disable.

    Returns:
        list: XSOAR incident dicts (name/occurred/severity/rawJSON).
    """
    incidents: list = []
    for finding in findings:
        finding_info = finding.get("finding_info") or {}
        uid = finding.get("metadata", {}).get("uid")

        if mirror_direction:
            finding["mirror_direction"] = mirror_direction
            finding["mirror_instance"] = demisto.integrationInstance()

        severity_id = finding.get("severity_id") or 0
        xsoar_severity = OCSF_SEVERITY_ID_TO_XSOAR.get(severity_id, IncidentSeverity.UNKNOWN)
        incidents.append(
            {
                "name": finding_info.get("title") or uid,
                "occurred": finding_info.get("created_time_dt"),
                "severity": xsoar_severity,
                "rawJSON": json.dumps(finding),
            }
        )

    return incidents


def build_close_entries(finding: dict) -> list:
    """Build the incoming-mirror entry that closes the XSOAR incident based on AWS status.

    An OCSF ``status_id`` of Resolved (4) or Suppressed (3) closes the XSOAR incident
    (``dbotIncidentClose``). Any other status yields no entry so the incident is left as-is.
    Reopening a closed incident from AWS is not supported.

    Args:
        finding (dict): The OCSF finding returned by AWS Security Hub V2.

    Returns:
        list: A single-element entry list instructing the server to close, or an empty list.
    """
    status_id = finding.get("status_id")
    if status_id is None:
        return []
    close_reason = OCSF_STATUS_ID_TO_CLOSE_REASON.get(status_id)
    if close_reason:
        finding_status = finding.get("status") or close_reason
        return [
            {
                "Type": EntryType.NOTE,
                "Contents": {
                    "dbotIncidentClose": True,
                    "closeReason": close_reason,
                    "closeNotes": f"Closed by mirroring: AWS Security Hub finding status is '{finding_status}'.",
                },
                "ContentsFormat": EntryFormat.JSON,
            }
        ]
    return []


def build_client(params: dict) -> BotoClient:
    """Build a boto3 ``securityhub`` client via the shared ``AWSClient`` (AWSApiModule).

    ``AWSClient`` handles role assumption, credentials, SSL verification, timeouts, retries and proxy.

    Args:
        params (dict): The integration parameters (``demisto.params()``).

    Returns:
        BotoClient: An initialized boto3 ``securityhub`` client.
    """
    aws_region = params.get("region")
    aws_role_arn = params.get("role_arn")
    aws_role_session_name = params.get("role_session_name")
    aws_role_session_duration = params.get("session_duration")
    aws_role_policy = None
    aws_access_key_id = params.get("credentials", {}).get("identifier")
    aws_secret_access_key = params.get("credentials", {}).get("password")
    verify_certificate = not argToBoolean(params.get("insecure", False))
    timeout = params.get("timeout")
    retries = arg_to_number(params.get("retries")) or DEFAULT_RETRIES
    sts_endpoint_url = params.get("sts_endpoint_url") or None
    endpoint_url = params.get("endpoint_url") or None

    validate_params(aws_region, aws_role_arn, aws_role_session_name, aws_access_key_id, aws_secret_access_key)

    aws_client = AWSClient(
        aws_region,
        aws_role_arn,
        aws_role_session_name,
        aws_role_session_duration,
        aws_role_policy,
        aws_access_key_id,
        aws_secret_access_key,
        verify_certificate,
        timeout,
        retries,
        sts_endpoint_url=sts_endpoint_url,
        endpoint_url=endpoint_url,
    )

    return aws_client.aws_session(
        service="securityhub",
        region=aws_region,
        role_arn=aws_role_arn,
        role_session_name=aws_role_session_name,
        role_session_duration=aws_role_session_duration,
    )


""" COMMAND FUNCTIONS """


def enable_security_hub_command(client: BotoClient, args: dict) -> CommandResults:
    """Enable AWS Security Hub V2 for the configured account and region.

    Args:
        client (BotoClient): The boto3 ``securityhub`` client.
        args (dict): Command arguments. Optional ``tags`` - a string of key/value pairs in the
            format ``key=key1,value=value1;key=key2,value=value2`` to assign to the resource.

    Returns:
        CommandResults: The ARN of the enabled Security Hub V2 resource.
    """
    tags = parse_tags(args.get("tags", ""))
    kwargs = remove_empty_elements({"Tags": tags})

    demisto.debug(f"[AWS_Security_Hub_V2] Enabling Security Hub V2 with tag keys: {list(tags.keys())}")
    response = client.enable_security_hub_v2(**kwargs)

    hub_arn = response.get("HubV2Arn")
    outputs = {"HubV2Arn": hub_arn}
    return CommandResults(
        outputs_prefix="AWS.SecurityHubV2.EnableHubV2",
        outputs_key_field="HubV2Arn",
        outputs=outputs,
        readable_output=tableToMarkdown("AWS Security Hub V2 successfully enabled.", outputs, removeNull=True),
        raw_response=response,
    )


def disable_security_hub_command(client: BotoClient, args: dict) -> CommandResults:
    """Disable AWS Security Hub V2 for the configured account and region.

    Args:
        client (BotoClient): The boto3 ``securityhub`` client.
        args (dict): Command arguments. No arguments are required.

    Returns:
        CommandResults: A confirmation message that Security Hub V2 was disabled.
    """
    demisto.debug("[AWS_Security_Hub_V2] Disabling Security Hub V2")
    response = client.disable_security_hub_v2()

    return CommandResults(
        readable_output="AWS Security Hub V2 was successfully disabled.",
        raw_response=response,
    )


def findings_get_command(client: BotoClient, args: dict) -> CommandResults:
    """Retrieve a list of OCSF-formatted findings from AWS Security Hub V2.

    Args:
        client (BotoClient): The boto3 ``securityhub`` client.
        args (dict): Command arguments (per-category filters, sort_field/sort_order, limit, next_token).

    Returns:
        CommandResults: The retrieved findings and the pagination token, if any.
    """
    sort_field = args.get("sort_field")
    sort_criteria = [{"Field": sort_field, "SortOrder": args.get("sort_order")}] if sort_field else None

    kwargs = remove_empty_elements(
        {
            "Filters": generate_filters_for_get_findings(args),
            "SortCriteria": sort_criteria,
            "MaxResults": min(arg_to_number(args.get("limit")) or DEFAULT_MAX_FETCH, MAX_FETCH_LIMIT),
            "NextToken": args.get("next_token"),
        }
    )

    demisto.debug(f"[AWS_Security_Hub_V2] Getting findings {kwargs.keys()=}")
    response = client.get_findings_v2(**kwargs)

    findings = response.get("Findings", [])
    if not findings:
        return CommandResults(readable_output="No findings were found.")

    next_token = response.get("NextToken")
    outputs = {
        "AWS.SecurityHubV2.Findings(val.metadata.uid && val.metadata.uid == obj.metadata.uid)": findings,
        "AWS.SecurityHubV2(true)": {"FindingsNextToken": next_token},
    }
    findings_table = [
        {
            "uid": finding.get("metadata", {}).get("uid"),
            "severity": finding.get("severity"),
            "status": finding.get("status"),
            "class_name": finding.get("class_name"),
            "resource_uid": ", ".join(
                resource["uid"]
                for resource in (finding.get("resources") or [])
                if isinstance(resource, dict) and resource.get("uid")
            ),
        }
        for finding in findings
    ]
    return CommandResults(
        outputs=remove_empty_elements(outputs),
        readable_output=tableToMarkdown(
            "AWS Security Hub V2 Findings",
            findings_table,
            headers=["uid", "severity", "status", "class_name", "resource_uid"],
            removeNull=True,
        ),
        raw_response=response,
    )


def findings_batch_update_command(client: BotoClient, args: dict) -> CommandResults:
    """Update one or more findings in a single batch request.

    Args:
        client (BotoClient): The boto3 ``securityhub`` client.
        args (dict): Targeting (``metadata_uids`` and/or ``finding_identifiers``) and updates
            (``comment``, ``severity_id``, ``status_id``).

    Returns:
        CommandResults: The processed and unprocessed findings returned by the API.

    Raises:
        DemistoException: If not exactly one of ``metadata_uids`` / ``finding_identifiers`` is provided.
    """
    metadata_uids = argToList(args.get("metadata_uids"))
    finding_identifiers = parse_finding_identifiers(args.get("finding_identifiers", ""))

    if bool(metadata_uids) == bool(finding_identifiers):
        raise DemistoException(
            "You must provide exactly one of 'metadata_uids' or 'finding_identifiers' to target findings, not both or neither."
        )

    kwargs = remove_empty_elements(
        {
            "MetadataUids": metadata_uids,
            "FindingIdentifiers": finding_identifiers,
            "Comment": args.get("comment"),
            "SeverityId": arg_to_number(args.get("severity_id")),
            "StatusId": arg_to_number(args.get("status_id")),
        }
    )

    demisto.debug(f"[AWS_Security_Hub_V2] Batch updating findings {kwargs.keys()=}")
    response = client.batch_update_findings_v2(**kwargs)

    processed = response.get("ProcessedFindings", [])
    unprocessed = response.get("UnprocessedFindings", [])
    outputs = {
        "ProcessedFindings": processed,
        "UnprocessedFindings": unprocessed,
    }
    readable_output = tableToMarkdown(
        "AWS Security Hub V2 Batch Update Findings",
        {
            "Processed": [finding.get("MetadataUid") for finding in processed],
            "Unprocessed": [finding.get("MetadataUid") for finding in unprocessed],
        },
        removeNull=True,
    )
    return CommandResults(
        outputs_prefix="AWS.SecurityHubV2.BatchUpdateFindings",
        outputs=remove_empty_elements(outputs),
        readable_output=readable_output,
        raw_response=response,
    )


def _query_findings_page(client: BotoClient, filters: dict, max_results: int, next_token: str | None) -> tuple[list, str | None]:
    """Run a single ``get_findings_v2`` page query and return its findings and next token.

    Args:
        client (BotoClient): The boto3 ``securityhub`` client.
        filters (dict): The composite ``Filters`` object to query with (reused across pages).
        max_results (int): ``MaxResults`` for this page (the still-needed count).
        next_token (str | None): Pagination token to continue a previous page, or ``None`` for a fresh query.

    Returns:
        tuple[list, str | None]: The page's findings and the ``NextToken`` for the next page (or ``None``).

    Raises:
        DemistoException: If the API raises a ``ClientError``.
    """
    kwargs: dict = {"MaxResults": max_results, "Filters": filters, "SortCriteria": FETCH_SORT_CRITERIA}
    if next_token:
        kwargs["NextToken"] = next_token
    demisto.debug(
        f"[AWS_Security_Hub_V2] Fetch: get_findings_v2 page. NextToken={'<set>' if next_token else None}, "
        f"MaxResults={max_results}, Filters={json.dumps(filters)}, SortCriteria={FETCH_SORT_CRITERIA}"
    )
    try:
        response = client.get_findings_v2(**kwargs)
    except client.exceptions.ClientError as e:
        handle_client_error(e, "Fetch: page query")
    return response.get("Findings", []), response.get("NextToken")


def _compute_fetch_boundary(new_findings: list, last_fetch: str, fetched_ids: list) -> tuple[str, list]:
    """Compute the next boundary state (``last_fetch`` + ``fetched_ids``) from the accumulated new findings.

    Findings are returned already sorted ascending by ``created_time_dt`` (see ``FETCH_SORT_CRITERIA``),
    so the last element carries the latest creation time. The boundary advances to that time and
    ``fetched_ids`` becomes every uid sharing it (so the next inclusive-``Start`` query can dedup them).

    Args:
        new_findings (list): All deduped findings collected across the pages fetched this cycle.
        last_fetch (str): The current boundary (returned unchanged when there are no new findings).
        fetched_ids (list): The current boundary uids (returned unchanged when there are no new findings).

    Returns:
        tuple[str, list]: The new ``last_fetch`` and ``fetched_ids``.
    """
    if not new_findings:
        return last_fetch, fetched_ids
    latest_finding_creation_time = (new_findings[-1].get("finding_info") or {}).get("created_time_dt") or ""
    boundary_ids = [
        f.get("metadata", {}).get("uid")
        for f in new_findings
        if (f.get("finding_info") or {}).get("created_time_dt") == latest_finding_creation_time
    ]
    return latest_finding_creation_time, boundary_ids


def fetch_incidents(client: BotoClient, params: dict) -> None:
    """Fetch AWS Security Hub V2 findings as XSOAR incidents.

    Queries findings by ascending created time over an inclusive [last_fetch, now] window and pages
    through results, deduping each page against boundary state saved in the last run. Pages are pulled
    until ``max_fetch`` new findings are collected or the API returns no next token, so a single fetch
    cycle still fills up to ``max_fetch`` incidents even when earlier pages are fully deduped away.

    Args:
        client (BotoClient): The boto3 ``securityhub`` client.
        params (dict): The integration parameters (first_fetch, max_fetch, min_severity,
            fetch_filters, mirror_direction).
    """
    demisto.debug("[AWS_Security_Hub_V2] Fetch: ===== fetch-incidents START =====")
    max_fetch = min(arg_to_number(params.get("max_fetch")) or DEFAULT_MAX_FETCH, MAX_FETCH_LIMIT)
    last_run = demisto.getLastRun()
    demisto.debug(
        f"[AWS_Security_Hub_V2] Fetch: raw lastRun from server: {last_run}, min_severity={params.get('min_severity')},"
        f" fetch_filters={params.get('fetch_filters')}, {max_fetch=}"
    )
    last_fetch = last_run.get("last_fetch")
    if not last_fetch:
        # No saved boundary yet (first run): derive the window start from the first_fetch param.
        first_fetch = (params.get("first_fetch") or DEFAULT_FIRST_FETCH).strip()
        format_first_fetch = parse(f"{first_fetch} UTC")
        if not format_first_fetch:
            raise DemistoException(f"Invalid 'First fetch time' value: {first_fetch!r}.")
        last_fetch = format_first_fetch.isoformat()
    next_token = last_run.get("next_token")
    fetched_ids: list = list(last_run.get("fetched_ids") or [])

    # Reuse the persisted filters when resuming a paginated window; otherwise build a fresh window query.
    if next_token:
        raw_filters = last_run.get("filters") or {}
        filters = json.loads(raw_filters) if isinstance(raw_filters, str) else raw_filters
    else:
        filters = build_fetch_filters(
            start_time=last_fetch,
            end_time=datetime.now(UTC).isoformat(),
            min_severity=params.get("min_severity"),
            additional_filters=params.get("fetch_filters"),
        )

    # Page-fill loop: keep pulling pages (requesting only the still-needed count) until we have max_fetch
    # new findings or there is no next token. This backfills incidents dropped by per-page dedup.
    # Bounded by max_fetch iterations to guard against the API returning a next token indefinitely
    # while every page is fully deduped away.
    new_findings: list = []
    for _ in range(max_fetch):
        remaining = max_fetch - len(new_findings)
        page_findings, next_token = _query_findings_page(client, filters, remaining, next_token)
        new_findings.extend(filter_new_findings(page_findings, last_fetch, fetched_ids))
        if len(new_findings) >= max_fetch or not next_token:
            break

    mirror_direction = MIRROR_DIRECTION_MAPPING.get(params.get("mirror_direction", "None"))
    incidents = findings_to_incidents(new_findings, mirror_direction)
    last_fetch, fetched_ids = _compute_fetch_boundary(new_findings, last_fetch, fetched_ids)

    # Persist the token (and the filters it is valid against) only when we stopped mid-window on max_fetch.
    next_token_to_persist = next_token if next_token else None
    new_last_run = {
        "last_fetch": last_fetch,
        "next_token": next_token_to_persist,
        "fetched_ids": fetched_ids,
        "filters": json.dumps(filters) if next_token_to_persist else {},
    }

    demisto.info(f"[AWS_Security_Hub_V2] Fetch: summary -> created {len(incidents)} incidents; new lastRun -> {new_last_run=}")
    demisto.setLastRun(new_last_run)
    demisto.incidents(incidents)
    demisto.debug("[AWS_Security_Hub_V2] Fetch: ===== fetch-incidents END =====")


def get_remote_data_command(client: BotoClient, args: dict) -> GetRemoteDataResponse:
    """Re-fetch a single mirrored finding by its uid and return its current state for incoming mirroring.

    ``get-modified-remote-data`` is intentionally not implemented: Security Hub V2 does not advance
    ``modified_time_dt`` on manual edits, so the server instead calls this per mirror-enrolled incident
    each cycle and diffs the returned finding against the incident (catching untimestamped console edits).

    Args:
        client (BotoClient): The boto3 ``securityhub`` client.
        args (dict): Command arguments. ``id`` - the finding ``metadata.uid`` to retrieve.

    Returns:
        GetRemoteDataResponse: The updated finding object and any close entries.
    """
    demisto.debug("[AWS_Security_Hub_V2] Mirror-in: ===== get-remote-data START =====")
    remote_args = GetRemoteDataArgs(args)
    finding_uid = remote_args.remote_incident_id
    demisto.debug(f"[AWS_Security_Hub_V2] Mirror-in: fetching current state of finding uid={finding_uid}")

    filters = {
        "CompositeFilters": [
            {
                "StringFilters": [{"FieldName": "metadata.uid", "Filter": {"Value": finding_uid, "Comparison": "EQUALS"}}],
            }
        ],
    }

    try:
        response = client.get_findings_v2(Filters=filters, MaxResults=1)
        demisto.debug("[AWS_Security_Hub_V2] Mirror-in: get_findings_v2 query succeeded.")
    except client.exceptions.ClientError as e:
        handle_client_error(e, "Mirror-in: get-remote-data query")

    findings = response.get("Findings", [])
    if not findings:
        demisto.debug(
            f"[AWS_Security_Hub_V2] Mirror-in: no finding found for uid={finding_uid}; nothing to mirror. "
            "===== get-remote-data END ====="
        )
        return GetRemoteDataResponse(mirrored_object={}, entries=[])

    finding = findings[0]
    # Attach the XSOAR severity so the incoming mapper can map it 1:1 (no transformer).
    severity_id = finding.get("severity_id") or 0
    finding["xsoar_severity"] = OCSF_SEVERITY_ID_TO_XSOAR.get(severity_id, IncidentSeverity.UNKNOWN)

    # Lifecycle sync: close the XSOAR incident to match the AWS finding status.
    entries = build_close_entries(finding)

    demisto.debug(
        f"[AWS_Security_Hub_V2] Mirror-in: returning current finding uid={finding_uid} "
        f"(severity_id={severity_id} -> xsoar_severity={finding['xsoar_severity']}, "
        f"status_id={finding.get('status_id')}, close entries={len(entries)}). "
        "===== get-remote-data END ====="
    )
    return GetRemoteDataResponse(mirrored_object=finding, entries=entries)


def get_mapping_fields_command() -> GetMappingFieldsResponse:
    """Return the schema of fields available for outgoing mirroring (from OUTGOING_FIELD_DESCRIPTIONS).

    Returns:
        GetMappingFieldsResponse: The outgoing mapping schema for the Security Hub finding incident type.
    """
    demisto.debug("[AWS_Security_Hub_V2] Mirror-out: get-mapping-fields")
    finding_scheme = SchemeTypeMapping(type_name="AWS Security Hub v2 Finding")
    for name, description in OUTGOING_FIELD_DESCRIPTIONS.items():
        finding_scheme.add_field(name=name, description=description)

    mapping_response = GetMappingFieldsResponse()
    mapping_response.add_scheme_type(finding_scheme)
    return mapping_response


def update_remote_system_command(client: BotoClient, args: dict, resolve_finding: bool) -> str:
    """Push local (XSOAR) incident changes to the corresponding finding via batch_update_findings_v2.

    Mirrors out only the whitelisted delta fields (severityid, statusid, comment) plus the built-in severity.
    When ``resolve_finding`` is enabled and the incident is closed, the finding is set to Resolved (status_id 4).
    Reopening an incident is not mirrored out.

    Args:
        client (BotoClient): The boto3 ``securityhub`` client.
        args (dict): The ``update-remote-system`` arguments (delta, incident status, remote id, etc.).
        resolve_finding (bool): Whether closing the incident in XSOAR should resolve the finding in AWS.

    Returns:
        str: The remote finding uid that was updated.
    """
    demisto.debug("[AWS_Security_Hub_V2] Mirror-out: ===== update-remote-system START =====")
    parsed_args = UpdateRemoteSystemArgs(args)
    remote_finding_uid = parsed_args.remote_incident_id
    delta = parsed_args.delta or {}
    demisto.debug(
        f"[AWS_Security_Hub_V2] Mirror-out: uid={remote_finding_uid}, incident_changed={parsed_args.incident_changed}, "
        f"inc_status={parsed_args.inc_status}, delta_keys={list(delta.keys())}, resolve_finding={resolve_finding}"
    )

    kwargs: dict = {}
    if parsed_args.incident_changed and delta:
        for delta_key, api_key in {"severityid": "SeverityId", "statusid": "StatusId", "comment": "Comment"}.items():
            if delta_key in delta and delta[delta_key] not in (None, ""):
                value = delta[delta_key]
                # severity_id and status_id are numeric in the API.
                kwargs[api_key] = arg_to_number(value) if api_key in ("SeverityId", "StatusId") else value

        # Changing the built-in XSOAR "severity" field surfaces a "severity" delta key (an XSOAR
        # severity number). Translate it to the OCSF SeverityId so editing the incident severity
        # mirrors out. An explicit "severityid" delta (handled above) takes precedence if both exist.
        if "SeverityId" not in kwargs and delta.get("severity") not in (None, ""):
            xsoar_severity = float(delta["severity"])
            ocsf_severity_id = XSOAR_SEVERITY_TO_OCSF_ID.get(xsoar_severity) if xsoar_severity is not None else None
            if ocsf_severity_id:
                kwargs["SeverityId"] = ocsf_severity_id
            else:
                demisto.debug(
                    f"[AWS_Security_Hub_V2] Mirror-out: XSOAR severity={delta['severity']} has no OCSF "
                    "equivalent (e.g. Unknown); not mirroring severity."
                )

    # If configured, closing the incident in XSOAR resolves the finding in AWS (overrides any delta status).
    if resolve_finding and parsed_args.inc_status == IncidentStatus.DONE:
        kwargs["StatusId"] = 4  # OCSF status_id 4 = Resolved.
        demisto.debug(
            "[AWS_Security_Hub_V2] Mirror-out: incident closed and resolve_finding enabled; " "forcing StatusId=4 (Resolved)."
        )

    if not kwargs:
        demisto.debug(
            f"[AWS_Security_Hub_V2] Mirror-out: no mirrorable changes for uid={remote_finding_uid}; skipping. "
            "===== update-remote-system END ====="
        )
        return remote_finding_uid

    kwargs["MetadataUids"] = [remote_finding_uid]
    demisto.debug(f"[AWS_Security_Hub_V2] Mirror-out: calling batch_update_findings_v2 with kwargs={kwargs}")
    try:
        response = client.batch_update_findings_v2(**kwargs)
        demisto.debug("[AWS_Security_Hub_V2] Mirror-out: batch_update_findings_v2 succeeded.")
    except client.exceptions.ClientError as e:
        handle_client_error(e, "Mirror-out: batch_update_findings_v2")

    unprocessed = response.get("UnprocessedFindings", [])
    if unprocessed:
        demisto.error(f"[AWS_Security_Hub_V2] Mirror-out: {len(unprocessed)} finding(s) were not updated: {unprocessed}")
    demisto.debug(f"[AWS_Security_Hub_V2] Mirror-out: updated uid={remote_finding_uid}. ===== update-remote-system END =====")
    return remote_finding_uid


def test_module(client: BotoClient) -> str:
    """Test connectivity and authentication against the AWS Security Hub V2 API.

    Args:
        client (BotoClient): An initialized boto3 ``securityhub`` client.

    Returns:
        str: ``"ok"`` if the call succeeds.

    Raises:
        DemistoException: When Security Hub V2 is not enabled or permissions are insufficient.
    """
    demisto.debug("[AWS_Security_Hub_V2] Test Connectivity and Authentication")
    try:
        client.describe_security_hub_v2()
    except client.exceptions.ResourceNotFoundException:
        raise DemistoException(
            "Security Hub V2 is not enabled in the configured account/region. "
            "Enable Security Hub V2 or verify the configured region."
        )
    except client.exceptions.AccessDeniedException:
        raise DemistoException(
            "Access denied. Verify the configured role/credentials have the 'securityhub:DescribeSecurityHubV2' permission."
        )
    return "ok"


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

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

    try:
        client = build_client(params)

        if command == "test-module":
            return_results(test_module(client))
        elif command == "aws-securityhub-v2-security-hub-enable":
            return_results(enable_security_hub_command(client, args))
        elif command == "aws-securityhub-v2-security-hub-disable":
            return_results(disable_security_hub_command(client, args))
        elif command == "aws-securityhub-v2-findings-get":
            return_results(findings_get_command(client, args))
        elif command == "aws-securityhub-v2-findings-batch-update":
            return_results(findings_batch_update_command(client, args))
        elif command == "fetch-incidents":
            fetch_incidents(client, params)
        elif command == "get-remote-data":
            return_results(get_remote_data_command(client, args))
        elif command == "get-mapping-fields":
            return_results(get_mapping_fields_command())
        elif command == "update-remote-system":
            resolve_finding = argToBoolean(params.get("resolve_finding", False))
            return_results(update_remote_system_command(client, args, resolve_finding))
        else:
            raise NotImplementedError(f"{command} command is not implemented.")

    except Exception as e:
        demisto.error(traceback.format_exc())
        return_error(f"Error has occurred in the AWS Security Hub V2 Integration: {type(e)} {e}", error=e)


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