import math import demistomock as demisto import urllib3 from CommonServerPython import * # noqa # pylint: disable=unused-wildcard-import from CommonServerUserPython import * # noqa # Disable insecure warnings urllib3.disable_warnings() """ CONSTANTS """ DEFAULT_MAX_FETCH = 3000 MAX_PAGE_SIZE = 1000 VENDOR = "Workday" PRODUCT = "Activity" DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" # ISO8601 format with UTC, default in XSOAR """ CLIENT CLASS """ class Client(BaseClient): """Client class to interact with the service API This Client implements API calls to the Saas Security platform, and does not contain any XSOAR logic. Handles the token retrieval. :param base_url (str): Workday server url. :param client_id (str): Workday client id. :param client_secret (str): Workday client_secret. :param token_url (str): Workday token url. :param refresh_token (str): Workday refresh token. :param verify (bool): specifies whether to verify the SSL certificate or not. :param proxy (bool): specifies if to use XSOAR proxy settings. """ def __init__(self, base_url, token_url, verify, proxy, headers, client_id, client_secret, refresh_token, max_fetch): super().__init__(base_url, verify=verify, proxy=proxy, headers=headers) self.client_id = client_id self.client_secret = client_secret self.refresh_token = refresh_token self.token_url = token_url self.max_fetch = max_fetch self.access_token = self.get_access_token() def get_access_token(self): # pragma: no cover """ Getting access token from Workday API. """ demisto.debug("Fetching access token from Workday API.") headers = {"Content-Type": "application/x-www-form-urlencoded"} data = {"grant_type": "refresh_token", "refresh_token": self.refresh_token} workday_resp_token = self._http_request( method="POST", full_url=self.token_url, headers=headers, data=data, auth=(self.client_id, self.client_secret) ) if workday_resp_token: return workday_resp_token.get("access_token") return None def http_request( self, method: str, url_suffix: str = "", params: dict = None, json_data: dict = None, retries: int = 0 ) -> dict: # pragma: no cover """ Overriding BaseClient http request in order to use the access token. """ headers = self._headers headers["Authorization"] = f"Bearer {self.access_token}" return self._http_request( method=method, url_suffix=url_suffix, params=params, json_data=json_data, headers=headers, retries=retries ) def get_activity_logging_request( self, from_date: str, to_date: str, offset: Optional[int] = 0, user_activity_entry_count: bool = False, limit: Optional[int] = 1000, ) -> list: """Returns a simple python dict with the information provided Args: offset: The zero-based index of the first object in a response collection. limit: The maximum number of loggings to return. to_date: date to fetch events from. from_date: date to fetch events to. user_activity_entry_count: If true, returns only the total count of user activity instances for the params. Returns: activity loggings returned from Workday API. """ instance_returned = math.ceil(self.max_fetch / 10000) params = { "from": from_date, "to": to_date, "limit": limit, "instancesReturned": instance_returned, "offset": offset, "returnUserActivityEntryCount": user_activity_entry_count, "type": "userActivity", } demisto.debug(f"params sent to Workday API are {params!s}") res = self.http_request(method="GET", url_suffix="/activityLogging", params=params, retries=3) return res.get("data", []) """ HELPER FUNCTIONS """ def resolve_max_fetch(params: dict) -> int: """ Resolves the max_fetch value from the integration params, falling back to DEFAULT_MAX_FETCH when the parameter is missing, empty, or evaluates to a falsy value. Args: params: The integration parameters. Returns: The resolved max_fetch value. """ return arg_to_number(params.get("max_fetch")) or DEFAULT_MAX_FETCH def get_max_fetch_activity_logging(client: Client, logging_to_fetch: int, from_date: str, to_date: str): """ Fetches up to logging_to_fetch activity logging avaiable from Workday. Args: client: Client object. logging_to_fetch: limit of logging to fetch from Workday. from_date: loggings from time. to_date: loggings to time. Returns: Activity loggings fetched from Workday. """ activity_loggings: list = [] offset = 0 while logging_to_fetch > 0: limit = min(MAX_PAGE_SIZE, logging_to_fetch) res = client.get_activity_logging_request(from_date=from_date, to_date=to_date, offset=offset, limit=limit) demisto.debug(f"Fetched {len(res)} activity loggings.") activity_loggings.extend(res) offset += len(res) logging_to_fetch -= len(res) if not res: break demisto.debug(f"{logging_to_fetch} loggings left to fetch.") demisto.debug(f"Found {len(activity_loggings)} activity loggings.") return activity_loggings def remove_duplications(activity_loggings: list, last_run: dict): """ Removes potential duplicated activity loggings. Args: activity_loggings: activity loggings fetched from Workday. last_run: Last run object. """ demisto.debug("Started removing duplications") last_log_stored = last_run.get("last_log") log_found = False final_count = 0 if last_log_stored: for count, log in enumerate(activity_loggings): if log == last_log_stored: log_found = True final_count = count break if log_found: demisto.debug(f"Found duplicated with {last_log_stored}, returning from {final_count}") return activity_loggings[final_count + 1 :] demisto.debug("Didn't find duplications, returning everything") return activity_loggings def remove_milliseconds_from_time_of_logging(activity_logging: dict): """ Workday API receive from_date only without milliseconds, therefor need to be removed. Args: activity_logging: activity logging Returns: The logging with the string in the correct format. """ demisto.debug("Changing timestamp of loggings to match date format.") date_format_with_milliseconds = "%Y-%m-%dT%H:%M:%S.%fZ" request_time_date_obj = datetime.strptime(activity_logging.get("requestTime"), date_format_with_milliseconds) # type: ignore request_time_date_obj.replace(microsecond=0) return datetime.strftime(request_time_date_obj, DATE_FORMAT) """ COMMAND FUNCTIONS """ def get_activity_logging_command( client: Client, from_date: str, to_date: str, limit: Optional[int], offset: Optional[int] ) -> tuple[list, CommandResults]: """ Args: offset: The zero-based index of the first object in a response collection. limit: The maximum number of loggings to return. to_date: date to fetch events from. from_date: date to fetch events to. client: Client object. Returns: Activity loggings from Workday. """ activity_loggings = client.get_activity_logging_request(to_date=to_date, from_date=from_date, limit=limit, offset=offset) readable_output = tableToMarkdown( "Activity Logging List:", activity_loggings, removeNull=True, headerTransform=lambda x: string_to_table_header(camel_case_to_underscore(x)), ) return activity_loggings, CommandResults(readable_output=readable_output) def fetch_activity_logging(client: Client, max_fetch: int, first_fetch: datetime, last_run: dict): """ Fetches activity loggings from Workday. Args: first_fetch: first fetch date. client: Client object. max_fetch: max loggings to fetch set by customer. last_run: last run object. Returns: Activity loggings from Workday. """ from_date = last_run.get("last_fetch_time", first_fetch.strftime(DATE_FORMAT)) to_date = datetime.now(tz=timezone.utc).strftime(DATE_FORMAT) demisto.debug(f"Getting activity loggings {from_date=}, {to_date=}.") activity_loggings = get_max_fetch_activity_logging( client=client, logging_to_fetch=max_fetch, from_date=from_date, to_date=to_date ) activity_loggings = remove_duplications(activity_loggings=activity_loggings, last_run=last_run) if activity_loggings: last_log = activity_loggings[-1] last_log_time = remove_milliseconds_from_time_of_logging(last_log) last_run = {"last_fetch_time": last_log_time, "last_log": last_log} return activity_loggings, last_run def test_module(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`` """ client.get_access_token() return "ok" """ 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() base_url = params.get("base_url") token_url = params.get("token_url") client_id = params.get("credentials", {}).get("identifier") client_secret = params.get("credentials", {}).get("password") token = params.get("token", {}).get("password") verify_certificate = not params.get("insecure", False) proxy = params.get("proxy", False) max_fetch = resolve_max_fetch(params) first_fetch = arg_to_datetime(arg=params.get("first_fetch", "3 days"), arg_name="First fetch time", required=True) demisto.debug(f"Command being called is {command}") try: client = Client( base_url=base_url, token_url=token_url, client_id=client_id, client_secret=client_secret, refresh_token=token, verify=verify_certificate, proxy=proxy, headers={"Accept": "application/json", "Content-Type": "application/json"}, max_fetch=max_fetch, ) if command == "test-module": return_results(test_module(client)) elif command == "workday-get-activity-logging": should_push_events = argToBoolean(args.get("should_push_events", "false")) activity_loggings, results = get_activity_logging_command( client=client, from_date=args.get("from_date"), to_date=args.get("to_date"), limit=arg_to_number(args.get("limit")), offset=arg_to_number(args.get("offset")), ) return_results(results) if should_push_events: send_events_to_xsiam(activity_loggings, vendor=VENDOR, product=PRODUCT) elif command == "fetch-events": last_run = demisto.getLastRun() activity_loggings, new_last_run = fetch_activity_logging( client=client, max_fetch=max_fetch, first_fetch=first_fetch, # type: ignore last_run=last_run, ) send_events_to_xsiam(activity_loggings, vendor=VENDOR, product=PRODUCT) if new_last_run: # saves next_run for the time fetch-events is invoked demisto.info(f"Setting new last_run to {new_last_run}") demisto.setLastRun(new_last_run) # Log exceptions and return errors except Exception as e: return_error(f"Failed to execute {demisto.command()} command.\nError:\n{e!s}") """ ENTRY POINT """ if __name__ in ("__main__", "__builtin__", "builtins"): main()