import uuid import traceback from CommonServerPython import * import demistomock as demisto from urllib import parse WIZ_VERSION = "1.1.0" WIZ_DEFEND = "wiz_defend" WIZ_DEFEND_INCIDENT_TYPE = "WizDefend Detection" USER_AGENT_NAME = "xsoar_defend" INTEGRATION_GUID = "8864e131-72db-4928-1293-e292f0ed699f" WIZ_DOMAIN_URL = "" DEMISTO_OCCURRED_FORMAT = "%Y-%m-%dT%H:%M:%SZ" WIZ_API_LIMIT = 250 API_MIN_FETCH = 10 API_MAX_FETCH = 1000 API_REQUEST_TIMEOUT = 115 # seconds; must be under the Apollo Router timeout (120s) API_END_CURSOR: Optional[str] = "" MAX_DAYS_FIRST_FETCH_DETECTIONS = 2 FETCH_INTERVAL_MINIMUM_MIN = 10 FETCH_INTERVAL_MAXIMUM_MIN = 600 DEFAULT_FETCH_BACK = "12 hours" MAX_FETCH_BUFFER = 15 # Percentage buffer for fetch interval calculations MAX_NOTE_LENGTH = 1400 # Hard limit for issue note text length enforced by the Wiz API # Threats THREATS_DAYS_MIN = 1 THREATS_DAYS_MAX = 30 THREATS_DAYS_DEFAULT = 5 class WizInputParam: DETECTION_ID = "detection_id" ISSUE_ID = "issue_id" TYPE = "type" PLATFORM = "platform" ORIGIN = "origin" CLOUD_ACCOUNT_OR_CLOUD_ORG = "cloud_account_or_cloud_organization" RESOURCE_ID = "resource_id" SEVERITY = "severity" STATUS = "status" CREATION_MINUTES_BACK = "creation_minutes_back" CREATION_DAYS_BACK = "creation_days_back" RULE_MATCH_ID = "rule_match_id" RULE_MATCH_NAME = "rule_match_name" PROJECT_ID = "project" RESOLUTION_REASON = "resolution_reason" RESOLUTION_NOTE = "resolution_note" REOPEN_NOTE = "reopen_note" NOTE = "note" class WizApiResponse: DATA = "data" DETECTIONS = "detections" ISSUES = "issues" UPDATE_ISSUE = "updateIssue" CREATE_ISSUE_NOTE = "createIssueNote" CLOUD_RESOURCES = "cloudResources" PROJECTS = "projects" GRAPH_SEARCH = "graphSearch" NODES = "nodes" PAGE_INFO = "pageInfo" HAS_NEXT_PAGE = "hasNextPage" END_CURSOR = "endCursor" ACCESS_TOKEN = "access_token" FILTER_BY = "filterBy" TYPE = "type" ERRORS = "errors" MESSAGE = "message" NOTES = "notes" class WizApiInputFields: API_ENDPOINT = "api_endpoint" AUTH_ENDPOINT = "auth_endpoint" CLIENT_ID = "client_id" CLIENT_SECRET = "client_secret" class DemistoParams: CREDENTIALS = "credentials" IDENTIFIER = "identifier" PASSWORD = "password" AUTH_ENDPOINT = "auth_endpoint" API_ENDPOINT = "api_endpoint" MAX_FETCH = "max_fetch" FIRST_FETCH = "first_fetch" TIME = "time" NAME = "name" OCCURRED = "occurred" RAW_JSON = "rawJSON" SEVERITY = "severity" MIRROR_ID = "dbotMirrorId" DETAILS = "details" AFTER_TIME = "after_time" URL = "url" IS_FETCH = "isFetch" INCIDENT_FETCH_INTERVAL = "incidentFetchInterval" INCIDENT_TYPE = "incidentType" class WizApiVariables: FIRST = "first" AFTER = "after" BEFORE = "before" FILTER_BY = "filterBy" FILTER_SCOPE = "filterScope" ORDER_BY = "orderBy" STATUS = "status" CREATED_AT = "createdAt" FIELD = "field" DIRECTION = "direction" TYPE = "type" PROVIDER_UNIQUE_ID = "providerUniqueId" RELATED_ENTITY = "relatedEntity" CLOUD_PLATFORM = "cloudPlatform" ID = "id" ISSUE_ID = "issueId" EQUALS = "equals" SEVERITY = "severity" IN_LAST = "inLast" AMOUNT = "amount" UNIT = "unit" RESOURCE = "resource" MATCHED_RULE = "matchedRule" MATCHED_RULE_NAME = "matchedRuleName" PROJECT_ID = "projectId" PROJECT = "project" NAME = "name" RULE = "rule" RULE_MATCH = "ruleMatch" ORIGIN = "origin" EVENT_ORIGIN = "eventOrigin" CLOUD_ACCOUNT_OR_CLOUD_ORGANIZATION_ID = "cloudAccountOrCloudOrganizationId" URL = "url" THREAT_RESOURCE = "threatResource" IDS = "ids" FETCH_CLOUD_ACCOUNTS_AND_CLOUD_ORG = "fetchCloudAccountsAndCloudOrganizations" PATCH = "patch" NOTE = "note" RESOLUTION_REASON = "resolutionReason" DESCRIPTION = "description" class WizThreatVariables: ALL_ISSUE_DETECTIONS = "ALL_ISSUE_DETECTIONS" THREAT_DETECTION = "THREAT_DETECTION" class WizStatus: OPEN = "OPEN" IN_PROGRESS = "IN_PROGRESS" REJECTED = "REJECTED" RESOLVED = "RESOLVED" class WizOrderByFields: SEVERITY = "SEVERITY" CREATED_AT = "CREATED_AT" class WizOrderDirection: DESC = "DESC" ASC = "ASC" class WizDetectionStatus: OPEN = "OPEN" IN_PROGRESS = "IN_PROGRESS" CLOSED = "CLOSED" REJECTED = "REJECTED" class WizIssueType: TOXIC_COMBINATION = "TOXIC_COMBINATION" THREAT_DETECTION = "THREAT_DETECTION" CLOUD_CONFIGURATION = "CLOUD_CONFIGURATION" class WizOperation: REJECT = "reject" RESOLUTION = "resolution" class WizSeverity: CRITICAL = "CRITICAL" HIGH = "HIGH" MEDIUM = "MEDIUM" LOW = "LOW" INFORMATIONAL = "INFORMATIONAL" class DemistoCommands: TEST_MODULE = "test-module" FETCH_INCIDENTS = "fetch-incidents" WIZ_DEFEND_GET_DETECTIONS = "wiz-defend-get-detections" WIZ_DEFEND_GET_DETECTION = "wiz-defend-get-detection" WIZ_DEFEND_GET_THREAT = "wiz-defend-get-threat" WIZ_DEFEND_GET_THREATS = "wiz-defend-get-threats" WIZ_DEFEND_REOPEN_THREAT = "wiz-defend-reopen-threat" WIZ_DEFEND_RESOLVE_THREAT = "wiz-defend-resolve-threat" WIZ_DEFEND_SET_THREAT_IN_PROGRESS = "wiz-defend-set-threat-in-progress" WIZ_DEFEND_SET_THREAT_COMMENT = "wiz-defend-set-threat-comment" WIZ_DEFEND_CLEAR_THREAT_COMMENTS = "wiz-defend-clear-threat-comments" class AuthParams: GRANT_TYPE = "grant_type" AUDIENCE = "audience" CLIENT_ID = "client_id" CLIENT_SECRET = "client_secret" class HeaderFields: CONTENT_TYPE = "Content-Type" USER_AGENT = "User-Agent" AUTHORIZATION = "Authorization" class ContentTypes: JSON = "application/json" FORM_URLENCODED = "application/x-www-form-urlencoded" class OutputPrefix: DETECTIONS = "Wiz.Manager.Detections" DETECTION = "Wiz.Manager.Detection" THREAT = "Wiz.Manager.Threat" THREATS = "Wiz.Manager.Threats" class ValidationType: """Class representing field names for validation results""" IS_VALID = "is_valid" ERROR_MESSAGE = "error_message" VALUE = "value" SEVERITY_LIST = "severity_list" MINUTES_VALUE = "minutes_value" DAYS_VALUE = "days_value" STATUS_LIST = "status_list" class ValidationResponse: """Class for standardized validation responses""" def __init__(self, is_valid=True, error_message=None, value=None): self.is_valid = is_valid self.error_message = error_message self.value = value self.days_value = None self.minutes_value = None self.severity_list = None self.status_list = None def to_dict(self): """Convert the response to a dictionary""" return { ValidationType.IS_VALID: self.is_valid, ValidationType.ERROR_MESSAGE: self.error_message, ValidationType.VALUE: self.value, ValidationType.DAYS_VALUE: self.days_value, ValidationType.MINUTES_VALUE: self.minutes_value, ValidationType.SEVERITY_LIST: self.severity_list, ValidationType.STATUS_LIST: self.status_list, } @classmethod def create_success(cls, value=None): """Create a successful validation response""" return cls(is_valid=True, error_message=None, value=value) @classmethod def create_error(cls, error_message): """Create a failed validation response""" return cls(is_valid=False, error_message=error_message, value=None) class DetectionType: """Detection types supported by the API""" GENERATED_THREAT = "GENERATED THREAT" DID_NOT_GENERATE_THREAT = "DID NOT GENERATE THREAT" api_dict = {GENERATED_THREAT: "GENERATED_THREAT", DID_NOT_GENERATE_THREAT: "MATCH_ONLY"} @classmethod def values(cls): """Get all available detection types with capital letters""" return [key for key in cls.api_dict if any(c.isupper() for c in key)] @classmethod def api_values(cls): """Get all available API values (values in api_dict)""" return list(cls.api_dict.values()) @classmethod def get_api_value(cls, user_input): """Convert user-friendly input to API value using api_dict Args: user_input: String or list of strings to convert Returns: Single API value (if string input) or list of API values (if list input) """ if not user_input: return None # Handle list input if isinstance(user_input, list): api_values = [] for item in user_input: if item: # Skip empty/None items item_lower = item.lower() for friendly_value, api_value in cls.api_dict.items(): if friendly_value.lower() in item_lower: api_values.append(api_value) break return api_values if api_values else None # Handle string input (original logic) user_input_lower = user_input.lower() for friendly_value, api_value in cls.api_dict.items(): if friendly_value.lower() in user_input_lower: return api_value return None class CloudPlatform: """Cloud platforms supported by the API""" AWS = "AWS" GCP = "GCP" AZURE = "Azure" OCI = "OCI" ALIBABA = "Alibaba" VSPHERE = "vSphere" OPENSTACK = "OpenStack" AKS = "AKS" EKS = "EKS" GKE = "GKE" KUBERNETES = "Kubernetes" OPENSHIFT = "OpenShift" OKE = "OKE" LINODE = "Linode" AZURE_DEVOPS = "AzureDevOps" GITHUB = "GitHub" GITLAB = "GitLab" BITBUCKET = "Bitbucket" TERRAFORM = "Terraform" OPENAI = "OpenAI" SNOWFLAKE = "Snowflake" MONGODB_ATLAS = "MongoDBAtlas" DATABRICKS = "Databricks" OKTA = "Okta" CLOUDFLARE = "Cloudflare" MICROSOFT365 = "Microsoft365" WIZ = "Wiz" ACK = "ACK" SELF_HOSTED = "SelfHosted" LKE = "LKE" @classmethod def values(cls): """Get all available cloud platforms""" return [getattr(cls, attr) for attr in dir(cls) if not attr.startswith("_") and not callable(getattr(cls, attr))] class DurationUnit: """Duration units for API filters""" DAYS = "DurationFilterValueUnitDays" HOURS = "DurationFilterValueUnitHours" MINUTES = "DurationFilterValueUnitMinutes" class DetectionOrigin: """Detection origins supported by the API""" WIZ_SENSOR = "WIZ_SENSOR" WIZ_ADMISSION_CONTROLLER = "WIZ_ADMISSION_CONTROLLER" WIZ_FILE_INTEGRITY_MONITORING = "WIZ_FILE_INTEGRITY_MONITORING" AWS_GUARD_DUTY = "AWS_GUARD_DUTY" AWS_CLOUDTRAIL = "AWS_CLOUDTRAIL" AZURE_DEFENDER_FOR_CLOUD = "AZURE_DEFENDER_FOR_CLOUD" AZURE_ACTIVITY_LOGS = "AZURE_ACTIVITY_LOGS" GCP_SECURITY_COMMAND_CENTER = "GCP_SECURITY_COMMAND_CENTER" GCP_AUDIT_LOGS = "GCP_AUDIT_LOGS" WIZ_AGENTLESS_FILE_INTEGRITY_MONITORING = "WIZ_AGENTLESS_FILE_INTEGRITY_MONITORING" AZURE_ACTIVE_DIRECTORY = "AZURE_ACTIVE_DIRECTORY" GOOGLE_WORKSPACE_AUDIT_LOGS = "GOOGLE_WORKSPACE_AUDIT_LOGS" WIN_SENTINEL_ONE = "WIN_SENTINEL_ONE" WIZ_CODE_ANALYZER = "WIZ_CODE_ANALYZER" WIN_SALT = "WIN_SALT" WIN_NONAME = "WIN_NONAME" WIN_CROWD_STRIKE = "WIN_CROWD_STRIKE" WIN_TRACEABLE = "WIN_TRACEABLE" WIZ_CLI = "WIZ_CLI" WIZ_IDE_EXTENSION = "WIZ_IDE_EXTENSION" WIZ_THREAT_DETECTION = "WIZ_THREAT_DETECTION" WIZ_KUBERNETES_AUDIT_LOGS_COLLECTOR = "WIZ_KUBERNETES_AUDIT_LOGS_COLLECTOR" WIZ_CUSTOM_INTEGRATION = "WIZ_CUSTOM_INTEGRATION" WIN_AKAMAI_GUARDICORE = "WIN_AKAMAI_GUARDICORE" OKTA_SYSTEM_LOGS = "OKTA_SYSTEM_LOGS" WIN_SNOWFLAKE = "WIN_SNOWFLAKE" WIN_FALCO = "WIN_FALCO" OCI_AUDIT_LOGS = "OCI_AUDIT_LOGS" WIZ_VCS_FETCHER = "WIZ_VCS_FETCHER" AWS_VPC_FLOW_LOGS = "AWS_VPC_FLOW_LOGS" GITHUB_AUDIT_LOGS = "GITHUB_AUDIT_LOGS" WIN_FIRE_TAIL = "WIN_FIRE_TAIL" AZURE_STORAGE_ACCOUNT = "AZURE_STORAGE_ACCOUNT" AZURE_KEY_VAULT = "AZURE_KEY_VAULT" AWS_RESOLVER_QUERY_LOGS = "AWS_RESOLVER_QUERY_LOGS" AWS_S3_DATA_EVENTS = "AWS_S3_DATA_EVENTS" GCP_STORAGE_DATA_ACCESS_LOGS = "GCP_STORAGE_DATA_ACCESS_LOGS" AWS_CLOUDTRAIL_NETWORK_ACTIVITY = "AWS_CLOUDTRAIL_NETWORK_ACTIVITY" WIZ_BROWSER_EXTENSION = "WIZ_BROWSER_EXTENSION" WIN_SALT_SECURITY = "WIN_SALT_SECURITY" @classmethod def values(cls): """Get all available detection origins""" return [getattr(cls, attr) for attr in dir(cls) if not attr.startswith("_") and not callable(getattr(cls, attr))] def get_integration_user_agent(): integration_user_agent = f"{INTEGRATION_GUID}/{USER_AGENT_NAME}/{WIZ_VERSION}" return integration_user_agent # Standard headers HEADERS_AUTH = {HeaderFields.CONTENT_TYPE: ContentTypes.FORM_URLENCODED, HeaderFields.USER_AGENT: get_integration_user_agent()} HEADERS = {HeaderFields.CONTENT_TYPE: ContentTypes.JSON, HeaderFields.USER_AGENT: get_integration_user_agent()} TOKEN = None URL = "" AUTH_E = "" # Pull Detections PULL_DETECTIONS_QUERY = """ query Detections($filterBy: DetectionFilters, $first: Int, $after: String, $orderBy: DetectionOrder, $includeTriggeringEvents: Boolean = true) { detections( filterBy: $filterBy first: $first after: $after orderBy: $orderBy enforceTimestampContinuity: true ) { nodes { id issue { id url dueAt projects { id name } resolutionReason notes { text } } ruleMatch { rule { id name sourceType } } description severity createdAt cloudAccounts { cloudProvider externalId name linkedProjects { id name } } cloudOrganizations { cloudProvider externalId name } startedAt endedAt actors { id externalId name type nativeType actingAs { id externalId name type nativeType } } primaryActor { id } resources { id externalId name type nativeType region cloudAccount { cloudProvider externalId name } kubernetesNamespace { id providerUniqueId name } kubernetesCluster { id providerUniqueId name } } primaryResource { id } triggeringEvents(first: 10) @include(if: $includeTriggeringEvents) { nodes { ... on CloudEvent { id origin name description cloudProviderUrl cloudPlatform timestamp source category status actor { id actingAs { id } } actorIP actorIPMeta { country autonomousSystemNumber autonomousSystemOrganization reputation reputationDescription reputationSource relatedAttackGroupNames customIPRanges { id name isInternal ipRanges } } resources { id } extraDetails { ... on CloudEventRuntimeDetails { processTree { command container { id externalId name image { id externalId } } path hash size executionTime runtimeProgramId userId userName } } } } } } } pageInfo { hasNextPage endCursor } } } """ PULL_DETECTIONS_VARIABLES = { WizApiVariables.ORDER_BY: { WizApiVariables.FIELD: WizOrderByFields.CREATED_AT, WizApiVariables.DIRECTION: WizOrderDirection.DESC, } } PULL_ISSUE_QUERY = """ query IssuesTable($filterBy: IssueFilters, $filterScope: IssueFiltersScope, $first: Int, $after: String, $orderBy: IssueOrder, $fetchSecurityScoreImpact: Boolean = false, $fetchThreatDetectionDetails: Boolean = false, $securityScoreImpactSelection: SecurityScoreImpactSelection, $fetchTotalCount: Boolean = true, $fetchActorsAndResourcesGraphEntities: Boolean = false, $fetchCloudAccountsAndCloudOrganizations: Boolean = false, $fetchMultipleSourceRules: Boolean = false, $fetchCommentThread: Boolean = false, $fetchThreatCenterActors: Boolean = false, $fetchTdrLogic: Boolean = false, $fetchSecuritySubCategories: Boolean = false) { issues: issuesV2( filterBy: $filterBy first: $first after: $after orderBy: $orderBy filterScope: $filterScope ) { nodes { id type resolutionNote resolvedAt resolutionReason ...ResolvedByUser control { id name description severity type query enabled enabledForLBI enabledForMBI enabledForHBI enabledForUnattributed tagsV2 { key value } risks threats sourceCloudConfigurationRule { id name } serviceTickets { ...ControlServiceTicket } } sourceRules { ...SourceRuleFields securitySubCategories @include(if: $fetchSecuritySubCategories) { id title category { id name framework { id name enabled } } } } sourceRules @include(if: $fetchMultipleSourceRules) { ...SourceRuleFields securitySubCategories @include(if: $fetchSecuritySubCategories) { id title category { id name framework { id name enabled } } } } createdAt updatedAt resolvedAt dueAt rejectionExpiredAt projects { id name slug isFolder businessUnit riskProfile { businessImpact } } status severity resolutionReason entitySnapshot { id type status name cloudPlatform region subscriptionName subscriptionId subscriptionExternalId nativeType kubernetesClusterId kubernetesClusterName kubernetesNamespaceName tags externalId } notes { id text } environments cloudAccounts @include(if: $fetchCloudAccountsAndCloudOrganizations) { id name externalId cloudProvider } cloudOrganizations @include(if: $fetchCloudAccountsAndCloudOrganizations) { id name externalId cloudProvider } threatDetectionDetails @include(if: $fetchThreatDetectionDetails) { ...ThreatDetectionDetailsActorsResources ...ThreatDetectionDetailsMainDetection detections(first: 0) { totalCount } eventOrigin } threatCenterActors @include(if: $fetchThreatCenterActors) { id name type } serviceTickets { id externalId name url } applicationServices { id displayName } commentThread @include(if: $fetchCommentThread) { id hasComments } } pageInfo { hasNextPage endCursor } totalCount @include(if: $fetchTotalCount) } } fragment ResolvedByUser on Issue { resolvedBy { user { id email name } } } fragment ControlServiceTicket on ServiceTicket { id externalId name url project { id name } integration { id type name typeConfiguration { type iconUrl } } } fragment SourceRuleFields on IssueSourceRule { ... on CloudConfigurationRule { id tags { key value } builtin createdBy { name } name description subjectEntityType hasAutoRemediation cloudProvider securityScoreImpact(selection: $securityScoreImpactSelection) @include(if: $fetchSecurityScoreImpact) risks threats control { id resolutionRecommendation } } ... on CloudEventRule { id name cloudEventRuleType: type description ruleSeverity: severity builtin createdBy { name } generateIssues generateFindings enabled sourceType ...CloudEventRuleLogicFields @include(if: $fetchTdrLogic) securityScoreImpact(selection: $securityScoreImpactSelection) @include(if: $fetchSecurityScoreImpact) risks threats } ... on Control { id tagsV2 { key value } name query type enabled enabledForHBI enabledForLBI enabledForMBI enabledForUnattributed builtin createdBy { name } resolutionRecommendation controlDescription: description securityScoreImpact(selection: $securityScoreImpactSelection) @include(if: $fetchSecurityScoreImpact) risks threats } } fragment CloudEventRuleLogicFields on CloudEventRule { params { ...CloudEventRuleParamsLogicFields } } fragment CloudEventRuleParamsLogicFields on CorrelationCloudEventRuleParams { securityGraphContext { description inUse } detectionThresholds { inUse } behavioralBaselines { id builtInId title description } } fragment ThreatDetectionDetailsActorsResources on ThreatDetectionIssueDetails { actorsMaxCountReached actorsTotalCount actors { id name externalId providerUniqueId type nativeType graphEntity @include(if: $fetchActorsAndResourcesGraphEntities) { id deletedAt type name properties } } resourcesTotalCount resourcesMaxCountReached resources { id name externalId providerUniqueId type nativeType graphEntity @include(if: $fetchActorsAndResourcesGraphEntities) { id type deletedAt name properties } } } fragment ThreatDetectionDetailsMainDetection on ThreatDetectionIssueDetails { mainDetection { id startedAt severity description(format: MARKDOWN) ruleMatch { rule { id name origins } } } } """ UPDATE_ISSUE_QUERY = """ mutation UpdateIssue( $issueId: ID! $patch: UpdateIssuePatch $override: UpdateIssuePatch ) { updateIssue(input: { id: $issueId, patch: $patch, override: $override }) { issue { id notes { ...IssueNoteDetails } status dueAt resolutionReason } } } fragment IssueNoteDetails on IssueNote { id text updatedAt createdAt user { id email } serviceAccount { id name } } """ CREATE_COMMENT_QUERY = """ mutation CreateIssueComment($input: CreateIssueNoteInput!) { createIssueNote(input: $input) { issueNote { createdAt id text user { id email } } } } """ DELETE_NOTE_QUERY = """ mutation DeleteIssueNote($input: DeleteIssueNoteInput!) { deleteIssueNote(input: $input) { _stub } } """ PULL_THREAT_ISSUE_VARIABLES = { WizApiVariables.FILTER_BY: {WizApiVariables.TYPE: [WizThreatVariables.THREAT_DETECTION]}, WizApiVariables.FILTER_SCOPE: WizThreatVariables.ALL_ISSUE_DETECTIONS, WizApiVariables.FETCH_CLOUD_ACCOUNTS_AND_CLOUD_ORG: True, WizApiVariables.ORDER_BY: { WizApiVariables.FIELD: WizOrderByFields.CREATED_AT, WizApiVariables.DIRECTION: WizOrderDirection.DESC, }, } class FetchIncident: """ Class to manage fetch incidents functionality with pagination support using last run only """ def __init__(self): """Initialize FetchIncident with last run data""" self.last_run_data = demisto.getLastRun() self.api_start_run_time = datetime.now().strftime(DEMISTO_OCCURRED_FORMAT) # Extract pagination values from last run using enums self.end_cursor = self.last_run_data.get(WizApiResponse.END_CURSOR) self.stored_after = self.last_run_data.get(WizApiVariables.AFTER) self.stored_before = self.last_run_data.get(WizApiVariables.BEFORE) self.last_run_time = self.get_last_run_time() # Read fetch interval for lagged window calculation self.fetch_interval_minutes = self._get_fetch_interval_minutes() self._validate_and_reset_params() def get_last_run_time(self): """ Gets the last run time for fetch incidents. If the last run time is more than MAX_DAYS_FIRST_FETCH_DETECTIONS days ago, it returns MAX_DAYS_FIRST_FETCH_DETECTIONS days ago instead. Returns: str: ISO formatted timestamp string for the last run time """ demisto_params = demisto.params() last_run = demisto.getLastRun().get(DemistoParams.TIME) if not last_run: demisto.info("First Time Fetch") first_fetch_param = demisto_params.get(DemistoParams.FIRST_FETCH, DEFAULT_FETCH_BACK).strip() last_run = get_fetch_timestamp(first_fetch_param) return last_run # Check if last_run is older than MAX_DAYS_FIRST_FETCH_DETECTIONS try: last_run_datetime = datetime.strptime(last_run, DEMISTO_OCCURRED_FORMAT) max_days_ago = datetime.now() - timedelta(days=MAX_DAYS_FIRST_FETCH_DETECTIONS) if last_run_datetime < max_days_ago: demisto.info( f"Last run time ({last_run}) is more than {MAX_DAYS_FIRST_FETCH_DETECTIONS} days ago. " f"Using {MAX_DAYS_FIRST_FETCH_DETECTIONS} days ago as the fetch time." ) last_run = max_days_ago.strftime(DEMISTO_OCCURRED_FORMAT) except Exception as e: demisto.error( f"Error parsing last run time: {str(e)}. Using {MAX_DAYS_FIRST_FETCH_DETECTIONS} days ago as fetch time." ) max_days_ago = datetime.now() - timedelta(days=MAX_DAYS_FIRST_FETCH_DETECTIONS) last_run = max_days_ago.strftime(DEMISTO_OCCURRED_FORMAT) return last_run def _get_fetch_interval_minutes(self): """Read incidentFetchInterval from params, returning validated minutes or the default.""" try: demisto_params = demisto.params() fetch_interval_str = demisto_params.get(DemistoParams.INCIDENT_FETCH_INTERVAL, str(FETCH_INTERVAL_MINIMUM_MIN)) validation_response = validate_fetch_interval(fetch_interval_str) if validation_response.is_valid: return validation_response.minutes_value except Exception: pass return FETCH_INTERVAL_MINIMUM_MIN def reset_params(self, reason="Invalid parameters detected"): """ Reset pagination parameters to safe defaults Args: reason (str): Reason for reset (for logging) """ demisto.info(f"Resetting fetch parameters: {reason}") if self.last_run_time: safe_after_str = self.last_run_time else: # Calculate safe_after_str as api_start_run_time - incidentFetchInterval try: demisto_params = demisto.params() fetch_interval_str = demisto_params.get(DemistoParams.INCIDENT_FETCH_INTERVAL, str(FETCH_INTERVAL_MINIMUM_MIN)) # Validate the fetch interval using existing validation validation_response = validate_fetch_interval(fetch_interval_str) if not validation_response.is_valid: demisto.error(f"Invalid fetch interval, using default: {validation_response.error_message}") fetch_interval_minutes = FETCH_INTERVAL_MINIMUM_MIN else: fetch_interval_minutes = validation_response.minutes_value # Calculate safe_after_str as current time minus fetch interval api_start_datetime = datetime.strptime(self.api_start_run_time, DEMISTO_OCCURRED_FORMAT) safe_after_datetime = api_start_datetime - timedelta(minutes=fetch_interval_minutes) safe_after_str = safe_after_datetime.strftime(DEMISTO_OCCURRED_FORMAT) demisto.debug( f"Calculated safe_after_str using fetch interval of {fetch_interval_minutes} minutes: {safe_after_str}" ) except Exception as e: demisto.error(f"Error calculating safe_after_str with fetch interval: {str(e)}. Using api_start_run_time") safe_after_str = self.api_start_run_time # Reset to safe values self.end_cursor = None self.stored_after = safe_after_str self.stored_before = self.api_start_run_time # Current time as before demisto.info( f"Reset fetch incidents parameter complete - " f"after: {self.stored_after}, before: {self.stored_before}, endCursor: None" ) def _validate_and_reset_params(self): """ Validate stored parameters and reset if invalid """ needs_reset = False reset_reason = [] if self._is_legacy_format(): needs_reset = True reset_reason.append("migrating from legacy format (only 'time' field)") # Check for None values that should have timestamps when pagination is active if self.end_cursor is not None: # If end_cursor exists, both stored_after and stored_before must exist if self.stored_after is None: needs_reset = True reset_reason.append("stored_after is None but endCursor exists") if self.stored_before is None: needs_reset = True reset_reason.append("stored_before is None but endCursor exists") # Validate timestamp formats timestamp_fields = [ ("stored_after", self.stored_after), ("stored_before", self.stored_before), ("last_run_time", self.last_run_time), ] for field_name, timestamp in timestamp_fields: if timestamp and not self._is_valid_timestamp(timestamp): needs_reset = True reset_reason.append(f"invalid {field_name} format: {timestamp}") # Validate time ordering (before >= after) if self.stored_after and self.stored_before and not self._is_valid_time_ordering(self.stored_after, self.stored_before): needs_reset = True reset_reason.append(f"invalid time ordering: before ({self.stored_before}) < after ({self.stored_after})") # Validate after time is not too old if self.stored_after and self._is_after_time_too_old(self.stored_after): needs_reset = True reset_reason.append(f"after time too old: {self.stored_after}") if needs_reset: reason = "; ".join(reset_reason) self.reset_params(reason) else: demisto.info( f"Using fetch incidents parameters: - " f"after: {self.stored_after}, before: {self.stored_before}, endCursor: None" ) def _is_legacy_format(self): """ Check if this is legacy format (existing customer with only 'time' field) Returns: bool: True if legacy format detected """ # Legacy format: has 'time' but missing the new pagination fields has_time = self.last_run_time is not None missing_new_fields = self.stored_after is None and self.stored_before is None and self.end_cursor is None is_legacy = has_time and missing_new_fields if is_legacy: demisto.info( f"Legacy format detected - last_run_time: {self.last_run_time}, " f"missing after/before/endCursor fields" ) return is_legacy def _is_valid_timestamp(self, timestamp): """ Check if timestamp is in valid format Args: timestamp (str): Timestamp to validate Returns: bool: True if valid, False otherwise """ try: datetime.strptime(timestamp, DEMISTO_OCCURRED_FORMAT) return True except Exception: return False def _is_valid_time_ordering(self, after_time, before_time): """ Check if before_time >= after_time Args: after_time (str): After timestamp before_time (str): Before timestamp Returns: bool: True if ordering is valid, False otherwise """ try: after_datetime = datetime.strptime(after_time, DEMISTO_OCCURRED_FORMAT) before_datetime = datetime.strptime(before_time, DEMISTO_OCCURRED_FORMAT) return before_datetime >= after_datetime except Exception: return False def _get_max_fetch_interval_minutes(self): """ Calculate the maximum fetch interval based on first_fetch setting + buffer Returns: int: Maximum allowed fetch interval in minutes """ try: demisto_params = demisto.params() first_fetch_param = demisto_params.get(DemistoParams.FIRST_FETCH, DEFAULT_FETCH_BACK).strip() # Parse first_fetch parameter to get minutes import dateparser first_fetch_time = dateparser.parse(f"{first_fetch_param} ago") if first_fetch_time: current_time = datetime.now() time_delta = current_time - first_fetch_time first_fetch_minutes = int(time_delta.total_seconds() / 60) # Use the global buffer parameter buffer_multiplier = 1 + (MAX_FETCH_BUFFER / 100) # Convert 15 to 1.15 max_minutes = int(first_fetch_minutes * buffer_multiplier) max_minutes = max(max_minutes, FETCH_INTERVAL_MINIMUM_MIN) demisto.debug( f"Calculated max fetch interval: {first_fetch_minutes} minutes " f"+ {MAX_FETCH_BUFFER}% buffer = {max_minutes} minutes (from first_fetch: '{first_fetch_param}')" ) return max_minutes except Exception as e: demisto.debug(f"Error calculating first_fetch interval: {str(e)}. Using default maximum.") # Fallback to original maximum return FETCH_INTERVAL_MAXIMUM_MIN def _is_after_time_too_old(self, after_time): """ Check if after_time exceeds maximum fetch interval Args: after_time (str): After timestamp to check Returns: bool: True if too old, False otherwise """ try: after_datetime = datetime.strptime(after_time, DEMISTO_OCCURRED_FORMAT) current_datetime = datetime.strptime(self.api_start_run_time, DEMISTO_OCCURRED_FORMAT) # Use dynamic maximum based on first_fetch + 15% max_interval_minutes = self._get_max_fetch_interval_minutes() max_interval = timedelta(minutes=max_interval_minutes) time_difference = current_datetime - after_datetime is_too_old = time_difference > max_interval if is_too_old: demisto.info( f"After time {after_time} exceeds maximum interval of {max_interval_minutes} minutes " f"(difference: {int(time_difference.total_seconds() / 60)} minutes)" ) return is_too_old except Exception: return True # If we can't parse, consider it invalid def get_api_after_parameter(self): """ Get the 'after' parameter value for the GraphQL API call. """ if self.should_continue_previous_run(): # Continuing pagination - use stored after time after_time = self.stored_after else: # Fresh fetch - use stored_after (which is set correctly by reset or previous run) after_time = self.stored_after if self.stored_after else self.last_run_time return after_time def get_api_before_parameter(self): """ Get the 'before' parameter value for the GraphQL API call. Fresh runs use a lagged boundary (now - fetch_interval) so consecutive windows don't overlap with near-real-time data that may still be settling. """ if self.should_continue_previous_run(): before_time = self.stored_before else: api_start = datetime.strptime(self.api_start_run_time, DEMISTO_OCCURRED_FORMAT) lagged = api_start - timedelta(minutes=self.fetch_interval_minutes) before_time = lagged.strftime(DEMISTO_OCCURRED_FORMAT) return before_time def should_continue_previous_run(self): """ Determines if this is a continuation of a previous paginated fetch. Returns: bool: True if we should continue previous run, False for fresh run """ return bool(self.end_cursor) def _validate_and_adjust_after_time(self, after_time): """ Validate that after_time is not older than FETCH_INTERVAL_MAXIMUM_MIN minutes and adjust if necessary Args: after_time (str): The after time to validate Returns: str: The validated/adjusted after time """ if not after_time: return self.api_start_run_time try: # Parse the after_time after_datetime = datetime.strptime(after_time, DEMISTO_OCCURRED_FORMAT) current_datetime = datetime.strptime(self.api_start_run_time, DEMISTO_OCCURRED_FORMAT) # Calculate maximum allowed time difference max_interval = timedelta(minutes=FETCH_INTERVAL_MAXIMUM_MIN) time_difference = current_datetime - after_datetime if time_difference > max_interval: # After time is too old, adjust to maximum allowed adjusted_after = current_datetime - max_interval adjusted_after_str = adjusted_after.strftime(DEMISTO_OCCURRED_FORMAT) demisto.info( f"After time {after_time} exceeds maximum fetch interval of {FETCH_INTERVAL_MAXIMUM_MIN} minutes. " f"Adjusting to {adjusted_after_str}" ) return adjusted_after_str return after_time except Exception as e: log_and_return_error(f"Error validating after_time {after_time}: {str(e)}") return None def get_api_cursor_parameter(self): """ Get the cursor parameter value for the GraphQL API call. Returns: str or None: The cursor to use for pagination, None if fresh fetch """ return self.end_cursor def _save_pagination_context(self): last_run_data = { DemistoParams.TIME: self.api_start_run_time, WizApiResponse.END_CURSOR: API_END_CURSOR, WizApiVariables.AFTER: self.stored_after, WizApiVariables.BEFORE: self.stored_before, } # Save using setLastRun demisto.setLastRun(last_run_data) demisto.debug(f"Fetch incidents didn't complete - set last run data to {json.dumps(last_run_data)}") def _clear_pagination_context(self): """ Clear pagination context when no more pages to fetch """ demisto.info("No end cursor found, clearing pagination context") # Create last run data without pagination context using enums last_run_data = { DemistoParams.TIME: self.api_start_run_time, WizApiResponse.END_CURSOR: None, WizApiVariables.AFTER: self.stored_before, WizApiVariables.BEFORE: self.api_start_run_time, } # Save using setLastRun demisto.setLastRun(last_run_data) demisto.info(f"Fetch incidents completed - set last run data to {json.dumps(last_run_data)}") def handle_post_incident_creation(self): """ Handle post-incident creation logic based on global API_END_CURSOR. Decides about pagination context and last run time based on API_END_CURSOR. Returns: None """ if bool(API_END_CURSOR): self._save_pagination_context() else: self._clear_pagination_context() def log_current_state(self): """ Log current state for debugging """ if self.end_cursor: status = ( f"Pagination in progress - {WizApiResponse.END_CURSOR}: {self.end_cursor}, " f"{WizApiVariables.AFTER}: {self.stored_after}, {WizApiVariables.BEFORE}: {self.stored_before}" ) else: status = "No active pagination" demisto.info(f"State: {status} - Last run time: {self.last_run_time}, API start time: {self.api_start_run_time}") def set_authentication_endpoint(auth_endpoint): global AUTH_E AUTH_E = auth_endpoint def set_api_endpoint(api_endpoint): global URL URL = api_endpoint def get_token(): """ Retrieve the token using the credentials """ global TOKEN audience = "wiz-api" demisto_params = demisto.params() said = demisto_params.get(DemistoParams.CREDENTIALS).get(DemistoParams.IDENTIFIER) sasecret = demisto_params.get(DemistoParams.CREDENTIALS).get(DemistoParams.PASSWORD) auth_payload = parse.urlencode( { AuthParams.GRANT_TYPE: "client_credentials", AuthParams.AUDIENCE: audience, AuthParams.CLIENT_ID: said, AuthParams.CLIENT_SECRET: sasecret, } ) response = requests.post(AUTH_E, headers=HEADERS_AUTH, data=auth_payload) if response.status_code != requests.codes.ok: raise Exception(f"Error authenticating to Wiz [{response.status_code}] - {response.text}") try: response_json = response.json() TOKEN = response_json.get(WizApiResponse.ACCESS_TOKEN) if not TOKEN: demisto.debug(json.dumps(response_json)) message = f"Could not retrieve token from Wiz: {response_json.get(WizApiResponse.MESSAGE)}" raise Exception(message) except ValueError as exception: demisto.debug(exception) raise Exception("Could not parse API response") HEADERS[HeaderFields.AUTHORIZATION] = "Bearer " + TOKEN return TOKEN def set_api_end_cursor(page_info): global API_END_CURSOR if page_info and page_info.get(WizApiResponse.HAS_NEXT_PAGE): API_END_CURSOR = page_info.get(WizApiResponse.END_CURSOR, "") else: API_END_CURSOR = None def get_entries(query, variables, wiz_type): if not TOKEN: get_token() data = {"variables": variables, "query": query} demisto.info(f"Invoking Wiz API with variables {json.dumps(variables)}") try: response = requests.post(url=URL, json=data, headers=HEADERS, timeout=API_REQUEST_TIMEOUT) response_json = response.json() demisto.info(f"Wiz API response status code is {response.status_code}") demisto.debug(f"The response is {response_json}") if response.status_code != requests.codes.ok: raise Exception(f"Got an error querying Wiz API [{response.status_code}] - {response.text}") if WizApiResponse.ERRORS in response_json: demisto.error(f"Wiz error content: {response_json[WizApiResponse.ERRORS]}") error_message = f"Wiz API error details: {get_error_output(response_json)}" demisto.error(f"An error has occurred using:\tVariables: {variables} -\t{error_message}") demisto.error(error_message) raise Exception(f"{error_message}\nCheck 'server.log' instance file to get additional information") if WizApiResponse.NODES in response_json[WizApiResponse.DATA][wiz_type]: new_entries = response_json[WizApiResponse.DATA][wiz_type][WizApiResponse.NODES] page_info = response_json[WizApiResponse.DATA][wiz_type][WizApiResponse.PAGE_INFO] else: new_entries = response_json[WizApiResponse.DATA][wiz_type] page_info = None set_api_end_cursor(page_info) return new_entries, page_info except Exception as e: error_message = f"Received an error while performing an API call.\nError info: {str(e)}" demisto.error(error_message) return_error(error_message) def query_detections(variables, paginate=True, max_fetch=API_MAX_FETCH): return query_api(PULL_DETECTIONS_QUERY, variables, WizApiResponse.DETECTIONS, paginate=paginate, max_fetch=max_fetch) def query_issues(variables, paginate=True): return query_api(PULL_ISSUE_QUERY, variables, WizApiResponse.ISSUES, paginate=paginate) def query_single_issue(issue_id): issue_variables = { WizApiVariables.FIRST: 1, WizApiVariables.FILTER_BY: {WizApiVariables.ID: issue_id}, } return query_issues(issue_variables, paginate=False) def query_api(query, variables, wiz_type, paginate=True, max_fetch=API_MAX_FETCH): entries, page_info = get_entries(query, variables, wiz_type) if not entries: demisto.info(f"No {wiz_type}(/s) available to fetch.") entries = [] while page_info[WizApiResponse.HAS_NEXT_PAGE] and paginate: demisto.debug(f"Successfully pulled {len(entries)} {wiz_type}") variables[WizApiVariables.AFTER] = page_info[WizApiResponse.END_CURSOR] new_entries, page_info = get_entries(query, variables, wiz_type) if new_entries is not None: entries += new_entries if len(entries) >= max_fetch: demisto.info( f"Reached the maximum fetch limit of {max_fetch} detections.\n" f"Some detections will not be processed in this fetch cycle.\n" f"Consider adjusting the filters to get relevant logs" ) break if entries: demisto.info(f"Successfully pulled {len(entries)} {wiz_type}") else: demisto.info(f"No {wiz_type}(/s) available to fetch according to this filter.") return entries def translate_severity(detection): """ Translate detection severity to demisto Might take risk grade into account in the future """ severity = demisto.get(detection, WizInputParam.SEVERITY) if severity == WizSeverity.CRITICAL: return 4 if severity == WizSeverity.HIGH: return 3 if severity == WizSeverity.MEDIUM: return 2 if severity == WizSeverity.LOW: return 1 if severity == WizSeverity.INFORMATIONAL: return 0.5 return None def _safe_rule_name(detection): """Return ruleMatch.rule.name from a detection payload, tolerating None at any level. Wiz API has historically returned null at multiple levels of the ruleMatch chain (`ruleMatch=None`, `ruleMatch={"rule": None}`, `ruleMatch={"rule": {}}`). Each needed its own null-safety fix in separate commits. Centralizing the traversal here so future variations only need one update. """ if not detection: return None rule_match = detection.get(WizApiVariables.RULE_MATCH) or {} rule = rule_match.get(WizApiVariables.RULE) or {} return rule.get(WizApiVariables.NAME) def build_fallback_description(detection): rule_name = _safe_rule_name(detection) severity = detection.get(WizApiVariables.SEVERITY, "Unknown") detection_id = detection.get(WizApiVariables.ID, "Unknown") parts = [f"{severity} severity detection"] if rule_name: parts.append(f"triggered by rule '{rule_name}'") parts.append(f"(ID: {detection_id})") return " ".join(parts) def build_incidents(detection): if detection is None: return {} rule_name = _safe_rule_name(detection) incident_name = f"{rule_name or 'Unknown Rule'} - {detection.get(WizApiVariables.ID, '')}" return { DemistoParams.NAME: incident_name, DemistoParams.OCCURRED: detection[WizApiVariables.CREATED_AT], DemistoParams.RAW_JSON: json.dumps(detection), DemistoParams.SEVERITY: translate_severity(detection), DemistoParams.MIRROR_ID: str(detection[WizApiVariables.ID]), DemistoParams.DETAILS: detection.get(WizApiVariables.DESCRIPTION, ""), } def extract_params_from_integration_settings(advanced_params=False): demisto_params = demisto.params() integration_setting_params = { WizInputParam.SEVERITY: demisto_params.get(WizInputParam.SEVERITY), WizInputParam.TYPE: demisto_params.get(WizInputParam.TYPE), WizInputParam.PLATFORM: demisto_params.get(WizInputParam.PLATFORM), WizInputParam.ORIGIN: demisto_params.get(WizInputParam.ORIGIN), WizInputParam.CLOUD_ACCOUNT_OR_CLOUD_ORG: demisto_params.get(WizInputParam.CLOUD_ACCOUNT_OR_CLOUD_ORG), } if advanced_params: for demisto_param in [ DemistoParams.FIRST_FETCH, DemistoParams.INCIDENT_FETCH_INTERVAL, DemistoParams.INCIDENT_TYPE, DemistoParams.IS_FETCH, DemistoParams.MAX_FETCH, ]: integration_setting_params[demisto_param] = demisto_params.get(demisto_param) return integration_setting_params def check_advanced_params(integration_settings_params): error_message = "" are_params_valid = True is_fetch = integration_settings_params.get(DemistoParams.IS_FETCH) first_fetch = integration_settings_params.get(DemistoParams.FIRST_FETCH) incident_fetch_interval = integration_settings_params.get(DemistoParams.INCIDENT_FETCH_INTERVAL) incident_type = integration_settings_params.get(DemistoParams.INCIDENT_TYPE) max_fetch = integration_settings_params.get(DemistoParams.MAX_FETCH) if is_fetch: first_fetch_validation = validate_first_fetch(first_fetch) if not first_fetch_validation.is_valid: are_params_valid = False error_message += f"{first_fetch_validation.error_message}\n" fetch_interval_validation = validate_fetch_interval(incident_fetch_interval) if not fetch_interval_validation.is_valid: are_params_valid = False error_message += f"{fetch_interval_validation.error_message}\n" incident_type_validation = validate_incident_type(incident_type) if not incident_type_validation.is_valid: are_params_valid = False error_message += f"{incident_type_validation.error_message}\n" max_fetch_validation = validate_max_fetch(max_fetch) if not max_fetch_validation.is_valid: are_params_valid = False error_message += f"{max_fetch_validation.error_message}\n" return are_params_valid, error_message def test_module(): """ Test the connection to the Wiz API and validate the params """ integration_settings_params = extract_params_from_integration_settings(advanced_params=True) are_params_valid, error_message = check_advanced_params(integration_settings_params) if not are_params_valid: demisto.results(error_message) return else: demisto.info("Advanced parameters are valid") wiz_detection = get_filtered_detections( detection_type=integration_settings_params[WizInputParam.TYPE], detection_platform=integration_settings_params[WizInputParam.PLATFORM], severity=integration_settings_params[WizInputParam.SEVERITY], detection_origin=integration_settings_params[WizInputParam.ORIGIN], detection_cloud_account_or_cloud_organization=integration_settings_params[WizInputParam.CLOUD_ACCOUNT_OR_CLOUD_ORG], api_limit=1, paginate=False, ) if WizApiResponse.ERRORS in wiz_detection or type(wiz_detection) is not list: demisto.results(wiz_detection) else: demisto.results("ok") def get_fetch_incidents_api_max_fetch(max_fetch): """ Get the API limit for fetching incidents """ max_fetch_validation = validate_max_fetch(max_fetch) api_limit = max_fetch_validation.value if max_fetch_validation.is_valid else API_MAX_FETCH return api_limit def fetch_incidents(): """ Fetch all Detections (OOB XSOAR Fetch) """ global API_MAX_FETCH fetch_manager = FetchIncident() fetch_manager.log_current_state() try: # Get integration settings integration_settings_params = extract_params_from_integration_settings(advanced_params=True) API_MAX_FETCH = get_fetch_incidents_api_max_fetch(integration_settings_params.get(DemistoParams.MAX_FETCH)) wiz_detections = get_filtered_detections( detection_type=integration_settings_params[WizInputParam.TYPE], detection_platform=integration_settings_params[WizInputParam.PLATFORM], severity=integration_settings_params[WizInputParam.SEVERITY], detection_origin=integration_settings_params[WizInputParam.ORIGIN], detection_cloud_account_or_cloud_organization=integration_settings_params[WizInputParam.CLOUD_ACCOUNT_OR_CLOUD_ORG], after_time=fetch_manager.get_api_after_parameter(), before_time=fetch_manager.get_api_before_parameter(), end_cursor=fetch_manager.get_api_cursor_parameter(), max_fetch=API_MAX_FETCH, ) if isinstance(wiz_detections, str): demisto.error(f"Error fetching detections: {wiz_detections}") return None # Build incidents from detections incidents = [] for detection in wiz_detections: incident = build_incidents(detection=detection) incidents.append(incident) demisto.incidents(incidents) fetch_manager.handle_post_incident_creation() if incidents: demisto.info(f"Successfully fetched and created {len(incidents)} incidents") else: demisto.info("No new incidents to fetch") except Exception as e: return log_and_return_error(f"Error fetching incidents: {e}") def get_fetch_timestamp(first_fetch_param): """ Gets the fetch timestamp based on the first fetch parameter Handles validation, error logging, and info messages Args: first_fetch_param (str): The first fetch parameter (e.g., "2 days", "30 days") Returns: str: ISO formatted timestamp for fetching Raises: ValueError: If the first fetch parameter is invalid """ # Validate first fetch timestamp is_valid, error_message, valid_date = validate_first_fetch_timestamp(first_fetch_param) if not is_valid: demisto.error(error_message) raise ValueError(error_message) # Check if we had to adjust the date to MAX_DAYS_FIRST_FETCH_DETECTIONS days max original_date = dateparser.parse(first_fetch_param or DEFAULT_FETCH_BACK) if original_date and valid_date.date() != original_date.date(): demisto.info( f"First fetch timestamp was more than {MAX_DAYS_FIRST_FETCH_DETECTIONS} days " f"({first_fetch_param}), automatically setting to " f"{MAX_DAYS_FIRST_FETCH_DETECTIONS} days back" ) # Return the ISO formatted timestamp return valid_date.isoformat()[:-3] + "Z" def update_wiz_domain_url(): """ Get the Wiz domain URL based on the integration settings """ global WIZ_DOMAIN_URL demisto_params = demisto.params() auth_endpoint = demisto_params.get(DemistoParams.AUTH_ENDPOINT) match = re.search(r"https://auth\.([\w\-]+\.\wiz\.\w+)/oauth/token", auth_endpoint) if match: WIZ_DOMAIN_URL = match.group(1) else: demisto.debug("Could not find the domain in the auth endpoint. Using default domain: app.wiz.io") WIZ_DOMAIN_URL = "app.wiz.io" def get_detection_url(detection): if not WIZ_DOMAIN_URL: update_wiz_domain_url() detection_url = ( f"https://{WIZ_DOMAIN_URL}/findings/detections#~(filters" f"~(updateTime~(dateRange~(past~(amount~5~unit~'day))))~detectionId~'{detection.get('id')}" f"~streamCols~(~'event~'principal~'principalIp~'resource))" ) return detection_url def get_threat_url(threat): if not WIZ_DOMAIN_URL: update_wiz_domain_url() detection_url = ( f"https://{WIZ_DOMAIN_URL}/threats#~(filters~(createdAt~(inTheLast~(amount~90~unit~'days)))~issue~'{threat.get('id')})" ) return detection_url def validate_wiz_enum_parameter(parameter_value, enum_class, parameter_name): """ Generic validation function for Wiz enum parameters Args: parameter_value (str or list): The parameter value(s) to validate enum_class: The enum class that contains valid values (e.g., WizIssueType) parameter_name (str): The human-readable parameter name for error messages (e.g., "issue type") Returns: ValidationResponse: Response with validation results """ if not parameter_value: return ValidationResponse.create_success() values = argToList(parameter_value) valid_values = enum_class.values() invalid_values = [v for v in values if v not in valid_values] if invalid_values: error_msg = ( f"Invalid {parameter_name}(s): {', '.join(invalid_values)}. Valid {parameter_name}s are: " f"{', '.join(valid_values)}" ) demisto.error(error_msg) return ValidationResponse.create_error(error_msg) return ValidationResponse.create_success(values) def validate_first_fetch_timestamp(first_fetch_param): """ Validates if the first fetch timestamp is within the limit Args: first_fetch_param (str): The first fetch parameter (e.g., "2 days", "30 days") Returns: tuple: (is_valid (bool), error_message (str), valid_date (datetime)) """ try: if not first_fetch_param: first_fetch_param = DEFAULT_FETCH_BACK # Parse the first fetch parameter first_fetch_date = dateparser.parse(first_fetch_param) if not first_fetch_date: return False, f"Invalid date format for first fetch: {first_fetch_param}", None # Calculate the maximum allowed date now = datetime.now() max_days_back = now - timedelta(days=MAX_DAYS_FIRST_FETCH_DETECTIONS) # Validate that first fetch is not more than MAX_DAYS_FIRST_FETCH_DETECTIONS if first_fetch_date < max_days_back: # Instead of erroring out, set it to the maximum allowed return True, None, max_days_back return True, None, first_fetch_date except Exception as e: error_msg = f"Error validating first fetch timestamp: {str(e)}" return False, error_msg, None def validate_detection_type(detection_type): """ Validates if the detection type is supported and converts user input to API value Args: detection_type (str): The detection type to validate Returns: ValidationResponse: Response with validation results """ if not detection_type: return ValidationResponse.create_success() # Convert user-friendly input to API value api_value = DetectionType.get_api_value(user_input=detection_type) if api_value: # Handle both single values and lists if isinstance(api_value, list): valid_api_values = set(DetectionType.api_values()) if set(api_value).issubset(valid_api_values): return ValidationResponse.create_success(api_value) else: if api_value in DetectionType.api_values(): return ValidationResponse.create_success(api_value) # If we get here, validation failed error_msg = f"Invalid detection type: {detection_type}. Valid types are: {', '.join(DetectionType.values())}" demisto.error(error_msg) return ValidationResponse.create_error(error_msg) def validate_matched_rule_id(matched_rule_id): """ Validates if the matched rule ID is a valid UUID Args: matched_rule_id (str): The matched rule ID to validate Returns: ValidationResponse: Response with validation results """ if not matched_rule_id: return ValidationResponse.create_success() if is_valid_uuid(matched_rule_id): return ValidationResponse.create_success(matched_rule_id) else: error_msg = f"Invalid matched rule ID: {matched_rule_id}. Must be a valid UUID." demisto.error(error_msg) return ValidationResponse.create_error(error_msg) def validate_detection_platform(platform): return validate_wiz_enum_parameter(platform, CloudPlatform, "platform") def validate_detection_cloud_account_or_cloud_organization(cloud_account_or_cloud_organization): """ Validates the detection cloud_account_or_cloud_organization parameter(s) are valid UUIDs Args: cloud_account_or_cloud_organization (str or list): The cloud_account_or_cloud_organization ID(s) to validate Returns: ValidationResponse: Response with validation results """ if not cloud_account_or_cloud_organization: return ValidationResponse.create_success() # Handle case where cloud_account_or_cloud_organization is a comma-separated string if isinstance(cloud_account_or_cloud_organization, str) and "," in cloud_account_or_cloud_organization: cloud_account_or_cloud_organizations = [s.strip() for s in cloud_account_or_cloud_organization.split(",")] elif isinstance(cloud_account_or_cloud_organization, str): cloud_account_or_cloud_organizations = [cloud_account_or_cloud_organization] elif isinstance(cloud_account_or_cloud_organization, list): cloud_account_or_cloud_organizations = cloud_account_or_cloud_organization else: error_msg = f"{WizInputParam.CLOUD_ACCOUNT_OR_CLOUD_ORG} must be a text value or list of text values" demisto.error(error_msg) return ValidationResponse.create_error(error_msg) # Validate each cloud_account_or_cloud_organization is a UUID invalid_cloud_account_or_cloud_organizations = [s for s in cloud_account_or_cloud_organizations if not is_valid_uuid(s)] if invalid_cloud_account_or_cloud_organizations: error_msg = ( f"Invalid {WizInputParam.CLOUD_ACCOUNT_OR_CLOUD_ORG} ID(s): " f"{', '.join(invalid_cloud_account_or_cloud_organizations)}. " f"All {WizInputParam.CLOUD_ACCOUNT_OR_CLOUD_ORG} must be in valid UUID format." ) demisto.error(error_msg) return ValidationResponse.create_error(error_msg) return ValidationResponse.create_success(cloud_account_or_cloud_organizations) def validate_detection_origin(origin): return validate_wiz_enum_parameter(origin, DetectionOrigin, "origin") def validate_creation_time_back(time_value, time_unit="minutes"): """ Validates if the creation time parameter is valid Args: time_value (str): Number of time units back to retrieve data time_unit (str): The time unit to validate ('minutes' or 'days') Returns: ValidationResponse: Response with validation results and time value """ response = ValidationResponse.create_success() # Set default values and limits based on the time unit if time_unit == "minutes": param_name = WizInputParam.CREATION_MINUTES_BACK min_value = FETCH_INTERVAL_MINIMUM_MIN max_value = FETCH_INTERVAL_MAXIMUM_MIN default_value = FETCH_INTERVAL_MINIMUM_MIN response.minutes_value = default_value elif time_unit == "days": param_name = WizInputParam.CREATION_DAYS_BACK min_value = THREATS_DAYS_MIN max_value = THREATS_DAYS_MAX default_value = THREATS_DAYS_DEFAULT response.days_value = default_value else: error_msg = f"Invalid time unit: {time_unit}. Supported units are 'minutes' and 'days'." return ValidationResponse.create_error(error_msg) if not time_value: return response error_msg = f"{param_name} must be a valid integer between {min_value} and {max_value}." try: time_int_value = int(time_value) if min_value <= time_int_value <= max_value: if time_unit == "minutes": response.minutes_value = time_int_value else: # days response.days_value = time_int_value return response else: return ValidationResponse.create_error(error_msg) except ValueError: demisto.error(error_msg) return ValidationResponse.create_error(error_msg) def validate_fetch_interval(fetch_interval): """ Validates if the creation_minutes_back parameter is valid Args: fetch_interval (int): Number of minutes back to retrieve detections Returns: ValidationResponse: Response with validation results and minutes value """ response = ValidationResponse.create_success() response.minutes_value = FETCH_INTERVAL_MINIMUM_MIN if not fetch_interval: error_msg = "Incidents Fetch Interval is required and cannot be empty." return ValidationResponse.create_error(error_msg) error_msg = ( f"Invalid Incidents Fetch Interval - It must be a valid integer " f"higher or equal than {FETCH_INTERVAL_MINIMUM_MIN}. Received {fetch_interval}." ) try: fetch_interval_int = int(fetch_interval) if fetch_interval_int >= FETCH_INTERVAL_MINIMUM_MIN: response.minutes_value = fetch_interval_int return response else: return ValidationResponse.create_error(error_msg) except (ValueError, TypeError): return ValidationResponse.create_error(error_msg) def validate_incident_type(incident_type): """ Validates if the incident type is set to WizDefend Detection Args: incident_type (str): The incident type to validate Returns: ValidationResponse: Response with validation results """ if incident_type == WIZ_DEFEND_INCIDENT_TYPE: return ValidationResponse.create_success(incident_type) else: error_msg = f"Invalid incident type: {incident_type}. Expected '{WIZ_DEFEND_INCIDENT_TYPE}'." demisto.error(error_msg) return ValidationResponse.create_error(error_msg) def validate_max_fetch(max_fetch): """ Validates if the max fetch parameter is valid Args: max_fetch (str or int): The max fetch value to validate Returns: ValidationResponse: Response with validation results and max fetch value """ response = ValidationResponse.create_success() response.value = API_MAX_FETCH if not max_fetch: return response error_msg = f"{DemistoParams.MAX_FETCH} must be a valid integer between 10 and 1000." try: max_fetch_int = int(max_fetch) if API_MIN_FETCH <= max_fetch_int <= API_MAX_FETCH: response.value = max_fetch_int return response else: return ValidationResponse.create_error(f"{error_msg} - Received {max_fetch}") except ValueError: demisto.error(error_msg) return ValidationResponse.create_error(error_msg) def validate_first_fetch(first_fetch): """ Validates if the first fetch timestamp is in the correct format and within the maximum days limit Args: first_fetch (str): The first fetch parameter (e.g., "2 days", "12 hours") Returns: ValidationResponse: Response with validation results and time value """ response = ValidationResponse.create_success() error_msg = ( f"Invalid first fetch format: {first_fetch}. Expected format is '