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=,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={'' 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()