from xml.sax.saxutils import escape import demistomock as demisto import urllib3 from CommonServerPython import * # noqa # pylint: disable=unused-wildcard-import # Disable insecure warnings urllib3.disable_warnings() VENDOR = "workday" PRODUCT = "signon" REQUEST_DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" # Old format for making requests EVENT_DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%f%z" # New format for processing events TIMEDELTA = 1 VERSION_PATTERN = re.compile(r"^v\d+\.\d+$") # To preserve the XML template, we are ignoring the long line ruff validation # ruff: noqa: E501 def get_api_version(params, default="v40.0"): """ Retrieve and validate the API version from the parameters dictionary. Parameters: params (dict): Dictionary containing parameters, potentially including 'api_version'. default (str): The default API version to use if 'api_version' is not provided or invalid. Returns: str: The validated API version. """ api_version = params.get("api_version", default) if VERSION_PATTERN.match(api_version): return api_version return default def get_from_time(seconds_ago: int) -> str: current_time = datetime.now(tz=timezone.utc) from_time = current_time - timedelta(seconds=seconds_ago) return from_time.strftime(REQUEST_DATE_FORMAT) def fletcher16(data: bytes) -> int: """ Compute the Fletcher-16 checksum for the given data. The Fletcher-16 checksum is a simple and fast checksum algorithm that provides a checksum value based on the input data. It's not as collision-resistant as cryptographic hashes but is faster and can be suitable for non-security-critical applications. Parameters: - data (bytes): The input data for which the checksum is to be computed. Returns: - int: The computed Fletcher-16 checksum value. """ sum1, sum2 = 0, 0 for byte in data: sum1 = (sum1 + byte) % 256 sum2 = (sum2 + sum1) % 256 return (sum2 << 8) | sum1 def generate_pseudo_id(event: dict) -> str: """ Compute a checksum for the given event using the Fletcher-16 algorithm. This function takes the entire event, serializes it to a JSON string, converts that string to bytes, and then computes a Fletcher-16 checksum for the byte data. Parameters: - event (dict): The entire event dictionary. Returns: - str: The unique ID, which is the computed Fletcher-16 checksum value concatenated with the event's Signon_DateTime. """ # Serialize the entire event to a JSON string and encode that to bytes event_str = json.dumps(event, sort_keys=True) data = event_str.encode() # Calculate the checksum checksum = fletcher16(data) # Create a unique ID by concatenating the checksum with the Signon_DateTime try: unique_id = f"{checksum}_{event['Signon_DateTime']}" except KeyError as e: raise DemistoException( f"While calculating the pseudo ID for an event, an event without a Signon_DateTime was found.\nError: {e}" ) return unique_id """ CLIENT CLASS """ class Client(BaseClient): """ Client will implement the service API and should not contain any Demisto logic. Should only do requests and return data. """ def __init__(self, params: dict): base_url = params.get("base_url", "invalid") tenant_name = params.get("tenant_name") api_version = get_api_version(params) username = params.get("credentials", {}).get("identifier") password = params.get("credentials", {}).get("password") verify_certificate = not params.get("insecure", False) proxy = params.get("proxy", False) if not base_url.startswith("https://"): raise ValueError("Invalid base URL. Should begin with https://") base_api_url = f"{base_url}/ccx/service/{tenant_name}/Identity_Management/{api_version}" headers = {"content-type": "text/xml;charset=UTF-8"} super().__init__(base_url=base_api_url, verify=verify_certificate, proxy=proxy, headers=headers) self.tenant_name = tenant_name self.username = escape(username) self.password = escape(password) self.api_version = api_version def generate_workday_account_signons_body( self, page: int, count: int, to_time: Optional[str] = None, from_time: Optional[str] = None, ) -> str: """ Generates XML body for Workday Account Signons Request. :type page: ``int`` :param page: Page number. :type count: ``int`` :param count: Number of results per page. :type to_time: ``Optional[str]`` :param to_time: End time for fetching events. :type from_time: ``Optional[str]`` :param from_time: Start time for fetching events. :return: XML body as string. :rtype: ``str`` """ return f""" {self.username} {self.password} {from_time} {to_time} {page} {count} {from_time} """ # noqa:E501 def generate_test_payload(self, from_time: str, to_time: str) -> str: return f""" {self.username} {self.password} {from_time} {to_time} 1 1 """ # noqa:E501 def retrieve_events( self, page: int, count: int, to_time: Optional[str] = None, from_time: Optional[str] = None, ) -> tuple: """ Retrieves events from Workday. :type page: ``int`` :param page: Page number. :type count: ``int`` :param count: Number of results per page. :type to_time: ``Optional[str]`` :param to_time: End time for fetching events. :type from_time: ``Optional[str]`` :param from_time: Start time for fetching events. :return: Tuple containing raw JSON response and account sign-on data. :rtype: ``Tuple`` """ # Make the HTTP request. raw_response = self._http_request( method="POST", url_suffix="", data=self.generate_workday_account_signons_body(page, count, to_time, from_time), resp_type="text", timeout=120, ) raw_json_response, account_signon_data = convert_to_json(raw_response) total_pages = int( demisto.get( obj=raw_json_response, field="Envelope.Body.Get_Workday_Account_Signons_Response.Response_Results", defaultParam={}, ).get("Total_Pages", "1") ) return account_signon_data, total_pages def test_connectivity(self) -> str: """ Tests API connectivity and authentication. :return: 'ok' if test passed, else exception. :rtype: ``str`` """ seconds_ago = 5 from_time = get_from_time(seconds_ago) to_time = datetime.now(tz=timezone.utc).strftime(REQUEST_DATE_FORMAT) payload = self.generate_test_payload(from_time=from_time, to_time=to_time) self._http_request(method="POST", url_suffix="", data=payload, resp_type="text", timeout=120) return "ok" """ HELPER FUNCTIONS """ def convert_to_json(response: str | dict) -> tuple[Dict[str, Any], Dict[str, Any]]: """ Convert an XML response to a JSON object and extract the 'Workday_Account_Signons' data. :param response: XML response to be converted :return: Tuple containing the full converted response and the extracted 'Workday_Account_Signons' data. :raises ValueError: If the expected data cannot be found in the response. """ if type(response) is dict: raw_json_response = response else: try: raw_json_response = json.loads(xml2json(response)) except Exception as e: raise ValueError(f"Error parsing XML to JSON: {e}") # Get the 'Get_Workday_Account_Signons_Response' dictionary safely response_data = demisto.get(raw_json_response, "Envelope.Body.Get_Workday_Account_Signons_Response") if not response_data: response_data = raw_json_response.get("Get_Workday_Account_Signons_Response", {}) account_signon_data = response_data.get("Response_Data", {}) # Ensure 'Workday_Account_Signon' is a list workday_account_signons = account_signon_data.get("Workday_Account_Signon") if isinstance(workday_account_signons, dict): account_signon_data["Workday_Account_Signon"] = [workday_account_signons] return raw_json_response, account_signon_data def process_and_filter_events(events: list, from_time: str, previous_run_pseudo_ids: set) -> tuple: non_duplicates = [] duplicates = [] pseudo_ids_for_next_iteration = set() try: from_datetime = datetime.strptime(from_time, EVENT_DATE_FORMAT).replace(tzinfo=timezone.utc) except ValueError: # On first run, the from_time is in UTC since that is what's sent in the request, this covers this scenario from_datetime = datetime.strptime(from_time, REQUEST_DATE_FORMAT).replace(tzinfo=timezone.utc) most_recent_event_time = datetime.min.replace(tzinfo=timezone.utc) for event in events: event_datetime = datetime.strptime(event["Signon_DateTime"], EVENT_DATE_FORMAT).replace(tzinfo=timezone.utc) # Add '_time' key to each event event["_time"] = event.get("Signon_DateTime") # Update the most recent event time if event_datetime > most_recent_event_time: most_recent_event_time = event_datetime # Check for duplicates within ±1 second of from_time if abs((event_datetime - from_datetime).total_seconds()) <= 1: event_pseudo_id = generate_pseudo_id(event) if event_pseudo_id not in previous_run_pseudo_ids: non_duplicates.append(event) else: duplicates.append(event_pseudo_id) else: non_duplicates.append(event) # Generate pseudo IDs for events within the last second of the most recent event last_second_start_time = most_recent_event_time - timedelta(seconds=TIMEDELTA) if duplicates: demisto.debug(f"[Dedup] Found {len(duplicates)} duplicate events: {duplicates}") for event in non_duplicates: event_datetime = datetime.strptime(event["_time"], EVENT_DATE_FORMAT).replace(tzinfo=timezone.utc) if event_datetime >= last_second_start_time: event_pseudo_id = generate_pseudo_id(event) pseudo_ids_for_next_iteration.add(event_pseudo_id) return non_duplicates, pseudo_ids_for_next_iteration def fetch_sign_on_logs(client: Client, limit_to_fetch: int, from_date: str, to_date: str): """ Fetches Sign On logs from workday. Args: client: Client object. limit_to_fetch: limit of logs to fetch from Workday. from_date: Events from time. to_date: Events to time. Returns: Sign On Events fetched from Workday. """ sign_on_logs: list = [] page = 1 # We assume that we will need to make one call at least res, total_pages = client.retrieve_events(from_time=from_date, to_time=to_date, page=1, count=999) sign_on_events_from_api = res.get("Workday_Account_Signon", []) sign_on_logs.extend(sign_on_events_from_api) total_fetched = len(sign_on_events_from_api) demisto.debug( f"[Pagination] Request indicates a total of {total_pages} pages to paginate. " f"Fetched {total_fetched} events from page 1." ) pages_remaining = total_pages - 1 while (page <= total_pages and pages_remaining != 0) and res: page += 1 remaining_to_fetch = limit_to_fetch - total_fetched if remaining_to_fetch <= 0: demisto.debug(f"[Pagination] Reached fetch limit of {limit_to_fetch}. Stopping pagination.") break res, _ = client.retrieve_events(from_time=from_date, to_time=to_date, page=page, count=remaining_to_fetch) sign_on_events_from_api = res.get("Workday_Account_Signon", []) fetched_count = len(sign_on_events_from_api) total_fetched += fetched_count demisto.debug(f"[Pagination] Fetched {fetched_count} sign on logs from page {page}.") sign_on_logs.extend(sign_on_events_from_api) pages_remaining -= 1 demisto.debug(f"[Pagination] {pages_remaining} pages left to fetch. Total fetched so far: {total_fetched}.") return sign_on_logs """ COMMAND FUNCTIONS """ def get_sign_on_events_command(client: Client, from_date: str, to_date: str, limit: int) -> tuple[list, CommandResults]: """ Args: limit: The maximum number of logs to return. to_date: date to fetch events from. from_date: date to fetch events to. client: Client object. Returns: Sign on logs from Workday. """ sign_on_events = fetch_sign_on_logs(client=client, limit_to_fetch=limit, from_date=from_date, to_date=to_date) [_event.update({"_time": _event.get("Signon_DateTime")}) for _event in sign_on_events] demisto.info(f"[Get Events] Got a total of {len(sign_on_events)} events between the time {from_date} to {to_date}") readable_output = tableToMarkdown( "Sign On Events List:", sign_on_events, removeNull=True, headerTransform=lambda x: string_to_table_header(camel_case_to_underscore(x)), ) return sign_on_events, CommandResults(readable_output=readable_output) def fetch_sign_on_events_command(client: Client, max_fetch: int, last_run: dict): """ Fetches sign on logs from Workday. Args: client: Client object. max_fetch: max logs to fetch set by customer. last_run: last run object. Returns: Sign on logs from Workday. """ current_time = datetime.utcnow() if "last_fetch_time" not in last_run: first_fetch_time = current_time - timedelta(minutes=1) first_fetch_str = first_fetch_time.strftime(REQUEST_DATE_FORMAT) from_date = last_run.get("last_fetch_time", first_fetch_str) else: from_date = last_run.get("last_fetch_time") # Checksums in this context is used as an ID since none is provided directly from Workday. # This is to prevent duplicates. previous_run_pseudo_ids = set(last_run.get("previous_run_pseudo_ids") or []) to_date = datetime.now(tz=timezone.utc).strftime(REQUEST_DATE_FORMAT) demisto.debug(f"[Fetch] Getting Sign On Events {from_date=}, {to_date=}.") sign_on_events = fetch_sign_on_logs(client=client, limit_to_fetch=max_fetch, from_date=from_date, to_date=to_date) if sign_on_events: demisto.debug(f"[Fetch] Got {len(sign_on_events)} sign_on_events. Begin processing.") non_duplicates, pseudo_ids_for_next_iteration = process_and_filter_events( events=sign_on_events, previous_run_pseudo_ids=previous_run_pseudo_ids, from_time=from_date ) demisto.debug(f"[Fetch] Done processing {len(non_duplicates)} sign_on_events.") last_event = non_duplicates[-1] last_run = { "last_fetch_time": last_event.get("Signon_DateTime"), "previous_run_pseudo_ids": list(pseudo_ids_for_next_iteration), } demisto.debug(f"[Fetch] Saving last run as {last_run}") else: # Handle the case where no events were retrieved last_run["last_fetch_time"] = current_time.strftime(REQUEST_DATE_FORMAT) non_duplicates = [] return non_duplicates, last_run def module_of_testing(client: Client) -> str: # pragma: no cover """Tests API connectivity and authentication Returning 'ok' indicates that the integration works like it is supposed to. Connection to the service is successful. Raises exceptions if something goes wrong. :type client: ``Client`` :param Client: client to use :return: 'ok' if test passed, anything else will fail the test. :rtype: ``str`` """ return client.test_connectivity() """ MAIN FUNCTION """ def main() -> None: # pragma: no cover """main function, parses params and runs command functions""" command = demisto.command() args = demisto.args() params = demisto.params() max_fetch = arg_to_number(params.get("max_fetch")) or 10000 demisto.debug(f"[Main] Command being called is {command}") try: client = Client(params=params) if command == "test-module": return_results(module_of_testing(client)) elif command == "workday-get-sign-on-events": if args.get("relative_from_date", None): from_time = arg_to_datetime( # type:ignore arg=args.get("relative_from_date"), arg_name="Relative datetime", required=False ).strftime(REQUEST_DATE_FORMAT) to_time = datetime.utcnow().strftime(REQUEST_DATE_FORMAT) else: from_time = args.get("from_date") to_time = args.get("to_date") sign_on_events, results = get_sign_on_events_command( client=client, from_date=from_time, to_date=to_time, limit=arg_to_number(args.get("limit", "100"), required=True), # type: ignore ) return_results(results) if argToBoolean(args.get("should_push_events", "true")): send_events_to_xsiam(sign_on_events, vendor=VENDOR, product=PRODUCT) elif command == "fetch-events": last_run = demisto.getLastRun() demisto.debug(f"[Main] Starting new fetch with last_run as {last_run}") sign_on_events, new_last_run = fetch_sign_on_events_command(client=client, max_fetch=max_fetch, last_run=last_run) demisto.debug(f"[Main] Done fetching events, sending to XSIAM {len(sign_on_events)} events.") send_events_to_xsiam(sign_on_events, vendor=VENDOR, product=PRODUCT) if new_last_run: # saves next_run for the time fetch-events is invoked demisto.info(f"[Main] Setting new last_run to {new_last_run}") demisto.setLastRun(new_last_run) else: raise NotImplementedError(f"command {command} is not implemented.") # Log exceptions and return errors except Exception as e: return_error(f"Failed to execute {demisto.command()} command.\nError:\n{str(e)}\nTraceback:\n{traceback.format_exc()}") """ ENTRY POINT """ if __name__ in ("__main__", "__builtin__", "builtins"): main()