""" Event Collector Source file for AdminByRequest API. """ from typing import Any import re import demistomock as demisto import urllib3 from CommonServerPython import * from CommonServerUserPython import * # Disable insecure warnings urllib3.disable_warnings() """ CONSTANTS """ VENDOR = "Admin" PRODUCT = "By_Request" MAX_FETCH_AUDIT_LIMIT = 50000 MAX_FETCH_EVENT_LIMIT = 50000 MAX_FETCH_REQUEST_LIMIT = 5000 EVENTS_LIMIT = 100_000 DATE_FORMAT_CALLS = "%Y-%m-%d" class EventType: """ This class defines an AdminByRequest API Event - used to dynamically store different types of events data. """ def __init__(self, suffix: str, take: int, source_log_type: str, time_field: str, default_params: dict): """ Prepare constructor for EventType class. Args: suffix: The url suffix of AdminByRequest API endpoint. take: Maximum events to fetch per API call. source_log_type: Key name for "source_log_type" field mapping inside XSIAM time_field: Key name for "_TIME" field mapping inside XSIAM """ self.suffix = suffix self.take = take self.source_log_type = source_log_type self.time_field = time_field self.last_run_key = "start_id_" + suffix self.max_fetch = 1 self.default_params = default_params EVENT_TYPES: dict[str, EventType] = { "Auditlog": EventType( suffix="auditlog", take=10000, source_log_type="auditlog", time_field="startTimeUTC", default_params={} ), "Events": EventType(suffix="events", take=10000, source_log_type="events", time_field="eventTimeUTC", default_params={}), "Requests": EventType( suffix="requests", take=1000, source_log_type="request", time_field="requestTime", default_params={"wantscandetails": 1} ), } """ CLIENT CLASS """ class Client(BaseClient): """Client class to interact with the service API""" def __init__(self, base_url: str, api_key: str, verify: bool, use_proxy: bool) -> None: """ Prepare constructor for Client class. Calls the constructor of BaseClient class and updates the header with the authentication token. Args: base_url: The url of AdminByRequest instance. api_key: The Api key for AdminByRequest API - specific for every licensing. verify: True if verify SSL certificate is checked in integration configuration, False otherwise. use_proxy: True if the proxy server needs to be used, False otherwise. """ super().__init__(base_url=base_url, verify=verify, proxy=use_proxy) self._api_key = api_key self._headers: dict[str, Any] = {"apiKey": self._api_key} def approve_request(self, url_suffix: str, headers: dict) -> requests.Response: """Approve a request.""" request_headers = self._headers.copy() request_headers.update(headers) return self._http_request("PUT", url_suffix=url_suffix, headers=request_headers, resp_type="response", ok_codes=(204,)) def deny_request(self, url_suffix: str, headers: dict) -> requests.Response: """Deny a request.""" request_headers = self._headers.copy() request_headers.update(headers) return self._http_request("DELETE", url_suffix=url_suffix, headers=request_headers, resp_type="response", ok_codes=(204,)) def get_events_request(self, url_suffix: str, params: dict) -> dict: """Retrieve the detections from AdminByRequest API.""" return self._http_request("GET", url_suffix=url_suffix, params=params, resp_type="json") """ HELPER FUNCTIONS """ def validate_email_address(email: str) -> bool: """ Validate an email address. Args: email (str): The email address to validate. Returns: bool: True if the email is valid, False otherwise. """ return re.fullmatch(emailRegex, email) is not None def remove_first_run_params(params: dict[str, Any]) -> None: """ Remove the "First Run" items from the param dictionary. Args: params (Dict[str, Any]): Integration parameters. """ if "startdate" in params: params.pop("startdate") if "enddate" in params: params.pop("enddate") def validate_fetch_events_params(last_run: dict, event_type: EventType, use_last_run_as_params: bool) -> tuple[dict, str, str]: """ Validate and update the params needed for the api call Args: last_run (dict): The last_run dictionary having the state of previous cycle. event_type (EventType): Event Type to fetch from API use_last_run_as_params (boolean): Flag that sign do we use the last-run as params for the API call Returns: Tuple[dict, str, str]: A tuple containing: - API call parameters. - URL suffix for the API endpoint. - Key used to update the `last_run` dictionary. """ suffix = event_type.suffix key = event_type.last_run_key if use_last_run_as_params: params = last_run elif key in last_run: # Not First fetch: Use last run's tracking ID as startid. params = {**event_type.default_params, "startid": last_run[key]} else: # First-time fetch: use today's date for time-based fetch (except for 'requests') today = get_current_time().strftime(DATE_FORMAT_CALLS) date_params = {} if suffix == "requests" else {"startdate": today, "enddate": today} params = {**event_type.default_params, **date_params} # Limit the number of records per fetch params["take"] = min(event_type.take, event_type.max_fetch) return params, suffix, key def fetch_events_list(client: Client, last_run: dict, event_type: EventType, use_last_run_as_params) -> list[dict[str, Any]]: """ Main Function that Handles the Fetch action to the API service of AdminByRequest. Args: client (Client): The client object used to interact with the AdminByRequest service. last_run (dict): The last_run dictionary having the state of previous cycle. event_type (EventType): Event Type to fetch from API use_last_run_as_params (bool): Flag that sign do we use the last-run as params for the API call Returns: list[dict[str, Any]]: List of records retrieved from the api call. """ params, suffix, last_run_key = validate_fetch_events_params(last_run, event_type, use_last_run_as_params) time_field, source_log_type = event_type.time_field, event_type.source_log_type fetch_limit = event_type.max_fetch last_id: int = 0 output: list[dict[str, Any]] = [] while True: try: # API call events = list(client.get_events_request(url_suffix=suffix, params=params)) except DemistoException as error: err_type = getattr(error, "exception", None) # If we have a Connection error with the server - return clean error message if isinstance(err_type, requests.exceptions.ConnectionError): clean_msg = str(error).split("\nError Type")[0] raise DemistoException(f"AdminByRequest: During fetch, exception occurred {clean_msg}") else: raise DemistoException(f"AdminByRequest: During fetch, exception occurred {str(error)}") if not events: break for event in events: # Updates each records in the list with _TIME and source_log_type fields # based on specific fields for each EventType. last_id = event["id"] event["_TIME"] = time_field event["source_log_type"] = source_log_type output.append(event) if len(output) >= fetch_limit: # update last run and return because we reach limit last_run.update({last_run_key: int(last_id + 1)}) return output # If it was the first run, we have a first run "params values" remove_first_run_params(params) params["startid"] = last_id + 1 # If we got at list one entity to add to output - update last ID if last_id: last_run.update({last_run_key: int(last_id + 1)}) return output def set_event_type_fetch_limit(params: dict[str, Any]) -> list[EventType]: """ Parses the event types to fetch from parameters and returns a dictionary mapping each selected event type's suffix to its corresponding max fetch limit. Args: params (Dict[str, Any]): Integration parameters. Returns: list[EventType]: List of event type to fetch from the api call. """ event_types_to_fetch = argToList(params.get("event_types_to_fetch", [])) event_types_to_fetch = [event_type.strip() for event_type in event_types_to_fetch] max_auditlog_per_fetch = arg_to_number(params.get("max_auditlog_per_fetch")) or MAX_FETCH_AUDIT_LIMIT max_events_per_fetch = arg_to_number(params.get("max_events_per_fetch")) or MAX_FETCH_EVENT_LIMIT max_requests_per_fetch = arg_to_number(params.get("max_requests_per_fetch")) or MAX_FETCH_REQUEST_LIMIT event_types = [] if "Auditlog" in event_types_to_fetch: EVENT_TYPES["Auditlog"].max_fetch = max_auditlog_per_fetch event_types.append(EVENT_TYPES["Auditlog"]) if "Events" in event_types_to_fetch: EVENT_TYPES["Events"].max_fetch = max_events_per_fetch event_types.append(EVENT_TYPES["Events"]) if "Requests" in event_types_to_fetch: EVENT_TYPES["Requests"].max_fetch = max_requests_per_fetch event_types.append(EVENT_TYPES["Requests"]) return event_types def prepare_list_output(records: List[dict[str, Any]]) -> str: """Prepare human-readable output. Args: records: List of entities response from the API. Returns: markdown string to be displayed in the war room. """ hr_outputs = [] for rec in records: hr_output = { "ID": rec.get("id"), "Type": rec.get("type"), "Settings Name": rec.get("settingsName"), "Application Name": demisto.get(rec, "application.name"), "Application Scan Result": demisto.get(rec, "application.scanResult"), "User": rec.get("user", {}), "Computer name": demisto.get(rec, "computer.name"), "Status": rec.get("status"), "Reason": rec.get("reason"), "Approved By": rec.get("approvedBy"), "Approved By Email": rec.get("approvedByEmail"), "Denied Reason": rec.get("deniedReason"), "Denied By": rec.get("deniedBy"), "Denied By Email": rec.get("deniedByEmail"), "Request Time": rec.get("requestTime"), "Start Time": rec.get("startTime"), "Event Text": rec.get("eventText"), "Event Time": rec.get("eventTime"), } hr_outputs.append(hr_output) return tableToMarkdown(name="AdminByRequests Record(s)", t=hr_outputs, removeNull=True) """ COMMAND FUNCTIONS """ def test_module(client: Client) -> str: # pragma: no cover """ Tests the connection to the service by calling each one of the api endpoints. Args: client (Client): The client object used to interact with the service. Returns: str: 'ok' if the connection is successful. If an authorization error occurs, an appropriate error message is returned. """ event_types = list(EVENT_TYPES.values()) for e in event_types: e.max_fetch = 1 last_run: dict[str, Any] = {} fetch_events(client, last_run, event_types) return "ok" def fetch_events( client: Client, last_run: dict, fetch_events_types: list[EventType], use_last_run_as_params: bool = False ) -> tuple[list[dict[str, Any]], dict]: """Fetch the specified AdminByRequest entity records. Args: client (Client): The client object used to interact with the AdminByRequest service. last_run (dict): The last_run dictionary having the state of previous cycle. fetch_events_types (list[EventType]) : list of Event Types to fetch from API use_last_run_as_params (bool): Flag that sign do we use the last-run as params for the API call Returns: - List of new records to be pushed into XSIAM. - Updated last_run dictionary. """ demisto.debug("AdminByRequest fetch_events invoked") events = [] for event_type in fetch_events_types: output = fetch_events_list(client, last_run, event_type, use_last_run_as_params) events.extend(output) demisto.debug(f"AdminByRequest next_run is {last_run}") return events, last_run def get_events(client: Client, args: dict) -> CommandResults: """ Inner Test Function to make sure the integration works Args: client: AdminByRequest client to be used. args: command arguments. Returns: Command results object that contain the results. """ max_events = arg_to_number(args.get("limit")) or None # User start date in the get events arguments, else get from today first_fetch = arg_to_datetime(args.get("first_fetch")) or get_current_time() first_fetch_date = first_fetch.strftime(DATE_FORMAT_CALLS) call_type: str = args.get("event_type", "") if not max_events: if call_type == "Auditlog": max_events = MAX_FETCH_AUDIT_LIMIT elif call_type == "Events": max_events = MAX_FETCH_EVENT_LIMIT else: max_events = MAX_FETCH_REQUEST_LIMIT event_type = EVENT_TYPES[call_type] event_type.max_fetch = max_events first_parm = {"startdate": first_fetch_date} last_run_to_use_as_params = {**event_type.default_params, **first_parm} output, _ = fetch_events(client, last_run_to_use_as_params, [event_type], use_last_run_as_params=True) human_readable = prepare_list_output(output) command_results = CommandResults( readable_output=human_readable, outputs=output, outputs_prefix="AdminByRequest." + call_type, ) return command_results def list_requests_command(client: Client, args: dict) -> CommandResults: """ Lists requests from AdminByRequest. Args: client: AdminByRequest client to be used. args: command arguments. Returns: CommandResults: Command results object that contains the results. """ request_id = args.get("request_id") status = args.get("status") want_scan_details = argToBoolean(args.get("want_scan_details", False)) limit = arg_to_number(args.get("limit", 50)) all_results = argToBoolean(args.get("all_results", False)) params: dict[str, Any] = {} if status: params["status"] = status if want_scan_details: params["wantscandetails"] = 1 results: list[dict[str, Any]] = [] if request_id: url_suffix = f"requests/{request_id}" result = client.get_events_request(url_suffix, params) if result: results.append(result) else: url_suffix = "requests" if all_results: params["take"] = 1000 while len(results) < EVENTS_LIMIT: response = client.get_events_request(url_suffix, params) if not response: break results.extend(response) if len(response) < 1000: break params["startid"] = response[-1]["id"] + 1 else: params["take"] = limit response = client.get_events_request(url_suffix, params) if response: results.extend(response) human_readable = prepare_list_output(results) command_results = CommandResults( readable_output=human_readable, outputs=results, outputs_prefix="AdminByRequest.Request", ) return command_results def approve_request_command(client: Client, args: dict) -> CommandResults: """ Approves a request in AdminByRequest. Use adminbyrequest-list-requests command to list all available requests. Args: client: AdminByRequest client to be used. args: command arguments. Returns: CommandResults: Command results object that contains the results. """ request_id = args.get("request_id") if not request_id: raise ValueError("request_id is required.") approved_by = args.get("approved_by") headers = {} if approved_by: if not validate_email_address(approved_by): raise DemistoException("approved_by must be a valid email address.") headers["approvedby"] = approved_by url_suffix = f"requests/{request_id}" response = client.approve_request(url_suffix, headers) if response.status_code == 204: readable_output = f"Request with {request_id} id was successfully approved." else: raise DemistoException(f"Failed to approve request {request_id}. Status code: {response.status_code}") return CommandResults(readable_output=readable_output) def deny_request_command(client: Client, args: dict) -> CommandResults: """ Denies a request in AdminByRequest. Use adminbyrequest-list-requests command to list all available requests. Args: client: AdminByRequest client to be used. args: command arguments. Returns: CommandResults: Command results object that contains the results. """ request_id = args.get("request_id") if not request_id: raise ValueError("request_id is required.") denied_by = args.get("denied_by") if denied_by and not validate_email_address(denied_by): raise ValueError("denied_by must be a valid email address.") reason = args.get("reason") headers = {} if denied_by: headers["deniedby"] = denied_by if reason: headers["reason"] = reason url_suffix = f"requests/{request_id}" response = client.deny_request(url_suffix, headers) if response.status_code == 204: readable_output = f"Request with {request_id} id was successfully denied." else: raise DemistoException(f"Failed to deny request {request_id}. Status code: {response.status_code}") return CommandResults(readable_output=readable_output) def main(): # pragma: no cover """main function, parses params and runs command functions""" params = demisto.params() args = demisto.args() command = demisto.command() # get the service API url base_url = params.get("url") verify_certificate = not argToBoolean(params.get("insecure", False)) proxy = argToBoolean(params.get("proxy", False)) api_key = params.get("credentials", {}).get("password") demisto.debug(f"Command being called is {command}") try: client = Client(base_url=base_url, api_key=api_key, verify=verify_certificate, use_proxy=proxy) events: List[dict[str, Any]] if command == "test-module": # Command made to test the integration result = test_module(client) return_results(result) elif command == "fetch-events": fetch_events_types = set_event_type_fetch_limit(params) last_run = demisto.getLastRun() events, next_run = fetch_events(client, last_run, fetch_events_types) if len(events): demisto.debug(f"Sending {len(events)} events to XSIAM AdminByRequest, before server call.") send_events_to_xsiam(events=events, vendor=VENDOR, product=PRODUCT) demisto.setLastRun(next_run) demisto.debug(f"Successfully saved last_run= {demisto.getLastRun()}") elif command == "adminbyrequest-get-events": command_results = get_events(client, args) events = cast(List[dict[str, Any]], command_results.outputs) if events and argToBoolean(args.get("should_push_events", False)): demisto.debug(f"Sending {len(events)} events.") send_events_to_xsiam(events=events, vendor=VENDOR, product=PRODUCT) return_results(command_results) elif command == "adminbyrequest-list-requests": return_results(list_requests_command(client, args)) elif command == "adminbyrequest-request-approve": return_results(approve_request_command(client, args)) elif command == "adminbyrequest-request-deny": return_results(deny_request_command(client, args)) else: raise NotImplementedError(f"Command {command} is not implemented") # Log exceptions and return errors except Exception as e: return_error(f"Failed to execute {command} command.\nError:\n{str(e)}") if __name__ in ("__main__", "__builtin__", "builtins"): main()