from http import HTTPStatus import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 from requests import Response JWT_TOKEN_EXPIRATION_PERIOD = 30 V2_PREFIX = "v2.0" V3_PREFIX = "v3.0" HTTPS_PROTOCOLS = ["http", "https", "ftp"] SOCKS_PROTOCOL = ["socks"] ISO8601_CONFIG = "%Y-%m-%dT%H:%M:%SZ" API_DATE_CONFIG = "%m/%d/%Y %H:%M" class Client(BaseClient): """Client class to interact with Cisco WSA API.""" def __init__(self, server_url: str, username: str, password: str, verify: bool, proxy: bool): super().__init__(base_url=server_url, headers={}, verify=verify, proxy=proxy) self.username = username self.password = password self.handle_request_headers() def handle_request_headers(self): """Retrieve and save to integration context JWT token for authorized client class API requests.""" integration_context = get_integration_context() jwt_token = integration_context.get("jwt_token") jwt_token_issued_time = integration_context.get("jwt_token_issued_time") if jwt_token and jwt_token_issued_time >= datetime.timestamp( datetime.now() - timedelta(minutes=JWT_TOKEN_EXPIRATION_PERIOD) ): self._headers["jwtToken"] = jwt_token else: jwt_token = self.retrieve_jwt_token() set_integration_context({"jwt_token": jwt_token, "jwt_token_issued_time": time.time()}) self._headers["jwtToken"] = jwt_token def retrieve_jwt_token(self) -> str: """ Retrieve JWT token from Cisco WSA. Returns: str: JWT token from Cisco WSA. """ data = { "data": { "userName": b64_encode(self.username), "passphrase": b64_encode(self.password), } } try: response = self._http_request("POST", f"{V2_PREFIX}/login", json_data=data) return dict_safe_get(response, ["data", "jwtToken"]) except DemistoException as error: if error.res is not None and error.res.status_code == HTTPStatus.UNAUTHORIZED: raise DemistoException("Authorization Error: make sure username and password are set correctly.") raise error def _http_request(self, *args, **kwargs): """HTTP request handler for Cisco WSA API. In some cases, the API status code is 200 but there are errors. Raises: DemistoException: Error to get to the API. """ res = super()._http_request(*args, **kwargs) if isinstance(res, dict) and all([res.get("res_code"), res.get("res_code") == HTTPStatus.BAD_REQUEST]): raise DemistoException(message=res) return res def access_policy_list(self, policy_names: str | None) -> dict[str, Any]: """ Access Policies list. Args: policy_names (str | None): Policies names to retrieve. Returns: dict[str, Any]: API response from Cisco WSA. """ params = assign_params(policy_names=policy_names) return self._http_request("GET", f"{V3_PREFIX}/web_security/access_policies", params=params, ok_codes=[HTTPStatus.OK]) def access_policy_create( self, policy_name: str, policy_status: str, identification_profiles: str, policy_order: int | None, policy_description: str | None, policy_expiry: str | None, ) -> dict[str, Any]: """ Create an access policy. Args: policy_name (str): Policy name to create. policy_status (str): Policy status. identification_profiles (str): Identification profile name. policy_order (int | None): Policy order. policy_description (str | None): Policy description. policy_expiry (str | None): Policy expiration date. Returns: dict[str, Any]: API response from Cisco WSA. """ data = remove_empty_elements( { "access_policies": [ { "policy_name": policy_name, "policy_order": policy_order, "policy_status": policy_status, "policy_description": policy_description, "policy_expiry": policy_expiry, "membership": { "identification_profiles": [ { "auth": "No Authentication", "profile_name": profile, } for profile in identification_profiles ], }, } ] } ) return self._http_request( "POST", f"{V3_PREFIX}/web_security/access_policies", json_data=data, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT], ) def access_policy_update( self, policy_name: str, new_policy_name: str | None, policy_status: str | None, policy_description: str | None, policy_order: int | None, policy_expiry: str | None, ) -> dict[str, Any]: """ Update an access policy. Args: policy_name (str): Policy name to update. new_policy_name (str | None): Policy status. policy_status (str | None): Policy status. policy_description (str | None): Policy description. policy_order (int | None): Policy order. policy_expiry (str | None): Policy expiry. Returns: dict[str, Any]: API response from Cisco WSA. """ data = remove_empty_elements( { "access_policies": [ { "policy_name": policy_name, "new_policy_name": new_policy_name, "policy_status": policy_status, "policy_description": policy_description, "policy_order": policy_order, "policy_expiry": policy_expiry, } ] } ) return self._http_request( "PUT", f"{V3_PREFIX}/web_security/access_policies", json_data=data, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT], ) def access_policy_protocols_user_agents_update( self, policy_name: str, block_custom_user_agents: List[str] | None, allow_connect_ports: List[str] | None, block_protocols: List[str] | None, settings_status: str, ) -> dict[str, Any]: """ Update access policy's objects settings. Args: policy_name (str): Policy name to update. block_custom_user_agents (List[str] | None): Block custom user agents. allow_connect_ports (List[str] | None): Allow connect ports. block_protocols (List[str] | None): Block protocols. settings_status (str): Settings status for the policy. Returns: dict[str, Any]: API response from Cisco WSA. """ data = remove_empty_elements( { "access_policies": [ { "policy_name": policy_name, "protocols_user_agents": { "block_custom_user_agents": block_custom_user_agents, "allow_connect_ports": allow_connect_ports, "block_protocols": block_protocols, "state": settings_status, }, } ] } ) return self._http_request( "PUT", f"{V3_PREFIX}/web_security/access_policies", json_data=data, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT], ) def access_policy_url_filtering_update( self, policy_name: str, predefined_categories_action: str | None, predefined_categories: List[str] | None, youtube_categories_action: str | None, youtube_categories: List[str] | None, custom_categories_action: str | None, custom_categories: List[str] | None, uncategorized_url: str | None, update_categories_action: str | None, content_rating_action: str | None, content_rating_status: str | None, safe_search_status: str | None, unsupported_safe_search_engine: str | None, ) -> dict[str, Any]: """ Update access policy's URL filtering settings. Args: policy_name (str): Policy name to update. predefined_categories_action (str | None): Predefined categories action. predefined_categories (List[str] | None): Predefined categories. youtube_categories_action (str | None): YouTube categories action. youtube_categories (List[str] | None): YouTube categories. custom_categories_action (str | None): Custom categories action. custom_categories (List[str] | None): Custom categories. uncategorized_url (str | None): Uncategorized URL action. update_categories_action (str | None): Update categories action. content_rating_action (str | None): Content rating action. content_rating_status (str | None): Content rating status. safe_search_status (str | None): Safe search status. unsupported_safe_search_engine (str | None): Unsupported safe search engine. Returns: dict[str, Any]: API response from Cisco WSA. """ data = remove_empty_elements( { "access_policies": [ { "policy_name": policy_name, "url_filtering": { "predefined_cats": {predefined_categories_action: predefined_categories}, "yt_cats": {youtube_categories_action: youtube_categories}, "custom_cats": {custom_categories_action: custom_categories}, "uncategorized_url": uncategorized_url, "update_cats_action": update_categories_action, "content_rating": { "status": content_rating_status, "action": content_rating_action, }, "safe_search": { "status": safe_search_status, "unsupported_safe_search_engine": unsupported_safe_search_engine, }, }, } ] } ) return self._http_request( "PUT", f"{V3_PREFIX}/web_security/access_policies", json_data=data, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT], ) def access_policy_applications_update( self, policy_name: str, application: str, action: str, values: dict[str, Any], settings_status: str, ) -> dict[str, Any]: """ Update access policy's applications settings. Args: policy_name (str): Policy name to update. application (str): Application to update. action (str): Action to perform on values. values (dict[str, Any]): Values to perform action on. settings_status (str): Settings status for the policy. Returns: dict[str, Any]: API response from Cisco WSA. """ data = { "access_policies": [ { "policy_name": policy_name, "avc": { "applications": { application: {action: values} if action == "block" else {action: {value: {} for value in values}}, }, "state": settings_status, }, } ] } return self._http_request( "PUT", f"{V3_PREFIX}/web_security/access_policies", json_data=data, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT], ) def access_policy_objects_update( self, policy_name: str, objects: dict[str, Any], object_type: str | None, object_action: str | None, object_values: List[str] | None, block_custom_mime_types: List[str] | None, http_or_https_max_object_size_mb: int | None, ftp_max_object_size_mb: int | None, ) -> dict[str, Any]: """ Update access policy's objects settings. Args: policy_name (str): Policy name to update. objects (dict[str, Any]): Policies objects. object_type (str | None): Object type. object_action (str | None): Object action. object_values (List[str] | None): Object values. block_custom_mime_types (List[str] | None): Block custom MIME types. http_or_https_max_object_size_mb (int | None): HTTP(S) max object size MB. ftp_max_object_size_mb (int | None): FTP max object size MB. Raises: DemistoException: Policy was not found. DemistoException: Update failed, objects were not found. Returns: dict[str, Any]: API response from Cisco WSA. """ organize_policy_object_data( objects=objects, object_type=object_type, object_action=object_action, object_values=object_values, block_custom_mime_types=block_custom_mime_types, http_or_https_max_object_size_mb=http_or_https_max_object_size_mb, ftp_max_object_size_mb=ftp_max_object_size_mb, ) data = {"access_policies": [{"policy_name": policy_name, "objects": objects}]} return self._http_request( "PUT", f"{V3_PREFIX}/web_security/access_policies", json_data=data, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT], ) def access_policy_anti_malware_update( self, policy_name: str, web_reputation_status: str | None, file_reputation_filtering_status: str | None, file_reputation_action: str | None, anti_malware_scanning_status: str | None, suspect_user_agent_scanning: str | None, block_malware_categories: List[str] | None, block_other_categories: List[str] | None, settings_status: str, ) -> dict[str, Any]: """ Update access policy's applications settings. Args: policy_name (str): Policy name to update. web_reputation_status (str | None): Web reputation status. file_reputation_filtering_status (str | None): File reputation filtering status. file_reputation_action (str | None): File reputation action. anti_malware_scanning_status (str | None): Anti-malware scanning status. suspect_user_agent_scanning (str | None): Suspect user agent scanning. block_malware_categories (List[str] | None): Malware categories to block. block_other_categories (List[str] | None): Other categories to block. settings_status (str): Application settings status. Returns: dict[str, Any]: API response from Cisco WSA. """ data = remove_empty_elements( { "access_policies": [ { "policy_name": policy_name, "amw_reputation": { "web_reputation": {"filtering": web_reputation_status}, "adv_malware_protection": { "file_reputation_filtering": file_reputation_filtering_status, "file_reputation": {file_reputation_action: ["Known Malicious and High-Risk Files"]} if file_reputation_action else {}, }, "cisco_dvs_amw": { "amw_scanning": {"amw_scan_status": anti_malware_scanning_status}, "suspect_user_agent_scanning": suspect_user_agent_scanning, "block_malware_categories": block_malware_categories, "block_other_categories": block_other_categories, }, "state": settings_status, } if settings_status == "custom" else {"state": settings_status}, } ] } ) return self._http_request( "PUT", f"{V3_PREFIX}/web_security/access_policies", json_data=data, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT], ) def access_policy_delete(self, policy_names: str) -> Response: """ Delete access policy. Args: policy_names (str): Comma separated policy names to delete. Returns: Response: API response from Cisco WSA. """ params = assign_params(policy_names=",".join(policy_names)) return self._http_request( "DELETE", f"{V3_PREFIX}/web_security/access_policies", params=params, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT, HTTPStatus.MULTI_STATUS], ) def domain_map_list(self) -> dict[str, Any]: """ List domain mappings. Returns: dict[str, Any]: API response from Cisco WSA. """ return self._http_request("GET", f"{V2_PREFIX}/configure/web_security/domain_map", ok_codes=[HTTPStatus.OK]) def domain_map_create(self, domain_name: str, ip_addresses: List[str], order: int) -> dict[str, Any]: """ Create domain mapping. Args: domain_name (str): Domain name. ip_addresses (List[str]): IP addresses to map to the domain. order (int): Index of domain map in the collection. Returns: dict[str, Any]: API response from Cisco WSA. """ data = [{"IP_addresses": ip_addresses, "domain_name": domain_name, "order": order}] return self._http_request( "POST", f"{V2_PREFIX}/configure/web_security/domain_map", json_data=data, ok_codes=[HTTPStatus.OK] ) def domain_map_update( self, domain_name: str, new_domain_name: str | None, ip_addresses: str | None, order: int | None, ) -> dict[str, Any]: """ Update domain map. Args: domain_name (str): Domain name to update. new_domain_name (str | None): New domain name. ip_addresses (str | None): IP addresses to map. order (int | None): Index of domain map. Returns: dict[str, Any]: API response from Cisco WSA. """ data = remove_empty_elements( [ { "domain_name": domain_name, "new_domain_name": new_domain_name, "IP_addresses": ip_addresses, "order": order, } ] ) return self._http_request( "PUT", f"{V2_PREFIX}/configure/web_security/domain_map", json_data=data, ok_codes=[HTTPStatus.OK] ) def domain_map_delete(self, domain_name: str) -> dict[str, Any]: """ Delete domain map. Args: domain_name (str): Domain name to delete. Returns: dict[str, Any]: API response from Cisco WSA. """ data = {"domain_name": domain_name} return self._http_request( "DELETE", f"{V2_PREFIX}/configure/web_security/domain_map", json_data=data, ok_codes=[HTTPStatus.OK, HTTPStatus.PARTIAL_CONTENT], ) def identification_profiles_list( self, profile_names: List[str] | None, ) -> dict[str, Any]: """ Get identification profiles. Args: profile_names (List[str] | None): Profile names to list. Returns: dict[str, Any]: API response from Cisco WSA. """ params = assign_params(profile_names=",".join(profile_names) if profile_names else None) return self._http_request( "GET", f"{V3_PREFIX}/web_security/identification_profiles", params=params, ok_codes=[HTTPStatus.OK] ) def identification_profiles_create( self, profile_name: str, status: str, description: str, order: int | None, protocols: List[str], proxy_ports: List[str] | None, members_by_subnet: List[str] | None, predefined_url_categories: List[str] | None, custom_url_categories: List[str] | None, ) -> dict[str, Any]: """ Create identification profile. Args: profile_name (str): Identification profile name. status (str): Status - enable/disable. description (str): Description of identification profile. order (int | None): Index of Identification profile in the collection. protocols (List[str]): Protocols - HTTPS/SOCKS. proxy_ports (List[str] | None): Proxy ports. members_by_subnet (List[str] | None): Members by subnet. predefined_url_categories (List[str] | None): Predefined URL categories. custom_url_categories (List[str] | None): Custom URL categories. Returns: dict[str, Any]: API response from Cisco WSA. """ data = remove_empty_elements( { "identification_profiles": [ { "profile_name": profile_name, "description": description, "status": status, "order": order, "members": { "protocols": protocols, "proxy_ports": proxy_ports, "ip": members_by_subnet, "url_categories": { "predefined": predefined_url_categories, "custom": custom_url_categories, }, }, } ] } ) return self._http_request( "POST", f"{V3_PREFIX}/web_security/identification_profiles", json_data=data, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT], ) def identification_profiles_update( self, profile_name: str, new_profile_name: str | None, status: str | None, description: str | None, order: int | None, protocols: List[str] | None, proxy_ports: List[str] | None, members_by_subnet: List[str] | None, predefined_url_categories: List[str] | None, custom_url_categories: List[str] | None, ) -> dict[str, Any]: """ Update identification profile. Args: profile_name (str): Identification profile name. new_profile_name (str | None): Identification profile name to update. status (str | None): Status - enable/disable. description (str | None): Description of identification profile. order (int | None): Index of Identification profile in the collection. protocols (List[str] | None): Protocols - HTTPS/SOCKS. proxy_ports (List[str] | None): Proxy ports. members_by_subnet (List[str] | None): Members by subnet. predefined_url_categories (List[str] | None): Predefined URL categories. custom_url_categories (List[str] | None): Custom URL categories. Returns: dict[str, Any]: API response from Cisco WSA. """ data = remove_empty_elements( { "identification_profiles": [ { "profile_name": profile_name, "new_profile_name": new_profile_name, "description": description, "status": status, "order": order, "members": { "protocols": protocols, "proxy_ports": proxy_ports, "ip": members_by_subnet, "url_categories": { "predefined": predefined_url_categories, "custom": custom_url_categories, }, }, } ] } ) return self._http_request( "PUT", f"{V3_PREFIX}/web_security/identification_profiles", json_data=data, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT], ) def identification_profiles_delete(self, profile_names: List[str]) -> Response: """ Delete identification profiles. Args: profile_names (List[str]): Identification profile names to delete. Returns: dict[str, Any]: API response from Cisco WSA. """ params = assign_params(profile_names=",".join(profile_names)) return self._http_request( "DELETE", f"{V3_PREFIX}/web_security/identification_profiles", params=params, resp_type="response", ok_codes=[HTTPStatus.NO_CONTENT, HTTPStatus.MULTI_STATUS], ) def url_categories_list(self) -> dict[str, Any]: """ List URL categories. Returns: dict[str, Any]: API response from Cisco WSA. """ return self._http_request("GET", f"{V3_PREFIX}/generic_resources/url_categories") def list_access_policy_command(client: Client, args: dict[str, Any]) -> CommandResults: """ List access policies. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ policy_names = args.get("policy_names") response = client.access_policy_list(policy_names=policy_names).get("access_policies", []) paginated_response = pagination(response=response, args=args) outputs = access_policy_output_handler(response=paginated_response) readable_output = tableToMarkdown( name="Access Policies", t=outputs, headers=[ "policy_name", "policy_status", "policy_order", "policy_description", "policy_expiry", ], headerTransform=string_to_table_header, removeNull=True, ) return CommandResults( readable_output=readable_output, outputs_prefix="CiscoWSA.AccessPolicy", outputs_key_field="policy_name", outputs=outputs, raw_response=response, ) def create_access_policy_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Create an access policy. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ policy_name = args["policy_name"] policy_status = args["policy_status"] policy_order = arg_to_number(args["policy_order"]) identification_profiles = argToList(args["identification_profiles"]) policy_description = args.get("policy_description") policy_expiry_date = arg_to_datetime(args.get("policy_expiry")) policy_expiry = policy_expiry_date.strftime(API_DATE_CONFIG) if policy_expiry_date else None client.access_policy_create( policy_name=policy_name, policy_status=policy_status, policy_order=policy_order, identification_profiles=identification_profiles, policy_description=policy_description, policy_expiry=policy_expiry, ) return CommandResults(readable_output=f'Created "{policy_name}" access policy successfully.') def update_access_policy_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Update an access policy. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ policy_name = args["policy_name"] new_policy_name = args.get("new_policy_name") policy_status = args.get("policy_status") policy_description = args.get("policy_description") policy_order = arg_to_number(args.get("policy_order")) policy_expiry_date = arg_to_datetime(args.get("policy_expiry")) policy_expiry = policy_expiry_date.strftime(API_DATE_CONFIG) if policy_expiry_date else None client.access_policy_update( policy_name=policy_name, new_policy_name=new_policy_name, policy_status=policy_status, policy_description=policy_description, policy_order=policy_order, policy_expiry=policy_expiry, ) return CommandResults( readable_output=f'Updated "{policy_name}" access policy successfully.', ) def update_access_policy_protocols_user_agents_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Update access policy's protocols and user agents settings. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ policy_name = args["policy_name"] block_custom_user_agents = argToList(args.get("block_custom_user_agents")) allow_connect_ports = argToList(args.get("allow_connect_ports")) block_protocols = argToList(args.get("block_protocols")) settings_status = args["settings_status"] client.access_policy_protocols_user_agents_update( policy_name=policy_name, block_custom_user_agents=block_custom_user_agents, allow_connect_ports=allow_connect_ports, block_protocols=block_protocols, settings_status=settings_status, ) return CommandResults( readable_output=f'Updated "{policy_name}" access policy successfully.', ) def update_access_policy_url_filtering_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Update access policy's URL filtering settings. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ policy_name = args["policy_name"] predefined_categories_action = args.get("predefined_categories_action") predefined_categories = argToList(args.get("predefined_categories")) youtube_categories_action = args.get("youtube_categories_action") youtube_categories = argToList(args.get("youtube_categories")) custom_categories_action = args.get("custom_categories_action") custom_categories = argToList(args.get("custom_categories")) uncategorized_url = args.get("uncategorized_url") update_categories_action = args.get("update_categories_action") content_rating_action = args.get("content_rating_action") content_rating_status = args.get("content_rating_status") safe_search_status = args.get("safe_search_status") unsupported_safe_search_engine = args.get("unsupported_safe_search_engine") client.access_policy_url_filtering_update( policy_name=policy_name, predefined_categories_action=predefined_categories_action, predefined_categories=predefined_categories, youtube_categories_action=youtube_categories_action, youtube_categories=youtube_categories, custom_categories_action=custom_categories_action, custom_categories=custom_categories, uncategorized_url=uncategorized_url, update_categories_action=update_categories_action, content_rating_action=content_rating_action, content_rating_status=content_rating_status, safe_search_status=safe_search_status, unsupported_safe_search_engine=unsupported_safe_search_engine, ) return CommandResults( readable_output=f'Updated "{policy_name}" access policy successfully.', ) def update_access_policy_applications_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Update access policy's applications settings. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ policy_name = args["policy_name"] application = args["application"] action = args["action"] values = argToList(args["values"]) settings_status = args["settings_status"] client.access_policy_applications_update( policy_name=policy_name, application=application, action=action, values=values, settings_status=settings_status, ) return CommandResults( readable_output=f'Updated "{policy_name}" access policy successfully.', ) def update_access_policy_objects_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Update access policy's objects settings. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ policy_name = args["policy_name"] object_type = args.get("object_type") object_action = args.get("object_action") object_values = argToList(args.get("object_values")) block_custom_mime_types = argToList(args.get("block_custom_mime_types")) http_or_https_max_object_size_mb = arg_to_number(args.get("http_or_https_max_object_size_mb")) ftp_max_object_size_mb = arg_to_number(args.get("ftp_max_object_size_mb")) objects = access_policy_objects_get(client=client, policy_name=policy_name) client.access_policy_objects_update( policy_name=policy_name, objects=objects, object_type=object_type, object_action=object_action, object_values=object_values, block_custom_mime_types=block_custom_mime_types, http_or_https_max_object_size_mb=http_or_https_max_object_size_mb, ftp_max_object_size_mb=ftp_max_object_size_mb, ) return CommandResults( readable_output=f'Updated "{policy_name}" access policy successfully.', ) def update_access_policy_anti_malware_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Update access policy's anti-malware and reputation settings. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ policy_name = args["policy_name"] web_reputation_status = args.get("web_reputation_status") file_reputation_filtering_status = args.get("file_reputation_filtering_status") file_reputation_action = args.get("file_reputation_action") anti_malware_scanning_status = args.get("anti_malware_scanning_status") suspect_user_agent_scanning = args.get("suspect_user_agent_scanning") block_malware_categories = argToList(args.get("block_malware_categories")) block_other_categories = argToList(args.get("block_other_categories")) settings_status = args["settings_status"] client.access_policy_anti_malware_update( policy_name=policy_name, web_reputation_status=web_reputation_status, file_reputation_filtering_status=file_reputation_filtering_status, file_reputation_action=file_reputation_action, anti_malware_scanning_status=anti_malware_scanning_status, suspect_user_agent_scanning=suspect_user_agent_scanning, block_malware_categories=block_malware_categories, block_other_categories=block_other_categories, settings_status=settings_status, ) return CommandResults( readable_output=f'Updated "{policy_name}" access policy successfully.', ) def delete_access_policy_command(client: Client, args: dict[str, Any]) -> Union[List[CommandResults], CommandResults]: """ Delete access policy. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: Union[List[CommandResults], CommandResults]: Readable outputs for XSOAR. """ policy_names = argToList(args["policy_names"]) response = client.access_policy_delete(policy_names) return delete_handler( response=response, obj_key="policy_name", readable_obj_name="Access Policy", success_readable_output="Deleted Access policy profiles successfully.", ) def list_domain_map_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Get domain mappings. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ domain_names = argToList(args.get("domain_names")) ip_addresses = argToList(args.get("ip_addresses")) response = client.domain_map_list().get("res_data", []) if domain_names or ip_addresses: response = [ domain for domain in response if domain.get("domain_name") in domain_names or any(address in domain.get("IP_addresses") for address in ip_addresses) ] paginated_response = pagination(response=response, args=args) readable_output = tableToMarkdown( name="Domain Map", t=paginated_response, headers=["domain_name", "IP_addresses", "order"], headerTransform=string_to_table_header, removeNull=True, ) return CommandResults( readable_output=readable_output, outputs_prefix="CiscoWSA.DomainMap", outputs_key_field="domain_name", outputs=paginated_response, raw_response=paginated_response, ) def create_domain_map_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Create domain mappings. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ domain_name = args["domain_name"] ip_addresses = argToList(args["ip_addresses"]) order = arg_to_number(args["order"]) if not order: raise DemistoException("Please enter correct number to order argument.") response = client.domain_map_create( domain_name=domain_name, ip_addresses=ip_addresses, order=order, ) readable_output = f'Domain "{domain_name}" mapping created successfully.' return CommandResults(readable_output=readable_output, raw_response=response) def update_domain_map_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Update domain mappings. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ domain_name = args["domain_name"] new_domain_name = args.get("new_domain_name") ip_addresses = argToList(args.get("ip_addresses")) order = arg_to_number(args.get("order")) response = client.domain_map_update( domain_name=domain_name, new_domain_name=new_domain_name, ip_addresses=ip_addresses, order=order, ) readable_output = f'Domain "{domain_name}" mapping updated successfully.' return CommandResults(readable_output=readable_output, raw_response=response) def delete_domain_map_command(client: Client, args: dict[str, Any]) -> Union[List[CommandResults], CommandResults]: """ Delete domain mappings. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Raises: DemistoException: In cases that the response code is 200 and the output [res_code] is not 200/206. Returns: CommandResults: readable outputs for XSOAR. """ domain_name = argToList(args["domain_names"]) response = client.domain_map_delete(domain_name=domain_name) if response.get("res_code") == HTTPStatus.OK: readable_output = f'Domain{"s" if len(domain_name) > 1 else ""} "{", ".join(domain_name)}" ' "deleted successfully." # noqa: ISC001 return CommandResults(readable_output=readable_output, raw_response=response) elif response.get("res_code") == HTTPStatus.PARTIAL_CONTENT: command_results_list = [] for domain_map in dict_safe_get(response, ["res_data", "delete_success"]): readable_output = f'Domain "{domain_map}" mapping deleted successfully.' command_results_list.append(CommandResults(readable_output=readable_output, raw_response=response)) readable_output = dict_safe_get(response, ["res_data", "delete_failure", "error_msg"]) if readable_output: command_results_list.append(CommandResults(readable_output=readable_output, raw_response=response)) return command_results_list raise DemistoException(message=response) def list_identification_profiles_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Get identification profiles. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ profile_names = argToList(args.get("profile_names")) response = client.identification_profiles_list(profile_names=profile_names).get("identification_profiles", []) paginated_response = pagination(response, args) readable_output = tableToMarkdown( name="Identification Profiles", t=paginated_response, headers=[ "order", "profile_name", "status", "description", "members", "identification_method", ], headerTransform=string_to_table_header, removeNull=True, ) filtered_data = identification_profile_mapper(paginated_response) return CommandResults( readable_output=readable_output, outputs_prefix="CiscoWSA.IdentificationProfile", outputs_key_field="profile_name", outputs=filtered_data, raw_response=paginated_response, ) def create_identification_profiles_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Create identification profiles. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ profile_name = args["profile_name"] client.identification_profiles_create( status=args["status"], description=args["description"], profile_name=profile_name, order=arg_to_number(args["order"]), protocols=protocols_handler(protocols=argToList(args["protocols"])), proxy_ports=argToList(args.get("proxy_ports")), members_by_subnet=argToList(args.get("members_by_subnet")), predefined_url_categories=argToList(args.get("predefined_url_categories")), custom_url_categories=argToList(args.get("custom_url_categories")), ) return CommandResults(readable_output=f'Created identification profile "{profile_name}" successfully.') def update_identification_profiles_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Update identification profiles. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ profile_name = args["profile_name"] client.identification_profiles_update( profile_name=profile_name, new_profile_name=args.get("new_profile_name"), description=args.get("description"), status=args.get("status"), protocols=protocols_handler(protocols=argToList(args.get("protocols"))), order=arg_to_number(args.get("order")), proxy_ports=argToList(args.get("proxy_ports")), members_by_subnet=argToList(args.get("members_by_subnet")), predefined_url_categories=argToList(args.get("predefined_url_categories")), custom_url_categories=argToList(args.get("custom_url_categories")), ) return CommandResults(readable_output=f'Updated identification profile "{profile_name}" successfully.') def delete_identification_profiles_command(client: Client, args: dict[str, Any]) -> Union[List[CommandResults], CommandResults]: """ Delete identification profiles. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ profile_names = argToList(args.get("profile_names")) response = client.identification_profiles_delete(profile_names) return delete_handler( response=response, obj_key="profile_name", readable_obj_name="Identification profile", success_readable_output="Deleted identification profiles successfully.", ) def list_url_categories_command(client: Client, args: dict[str, Any]) -> CommandResults: """ Get URL categories. Args: client (Client): Cisco WSA API client. args (dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: readable outputs for XSOAR. """ response = client.url_categories_list() outputs = categories_output_filter(response=response, contain=args.get("contain"), type_=args.get("type")) readable_output = tableToMarkdown( name="URL categories", t=outputs, headerTransform=string_to_table_header, removeNull=True, ) return CommandResults( readable_output=readable_output, outputs_prefix="CiscoWSA.UrlCategory", outputs=outputs, raw_response=response, ) def test_module(client: Client) -> str: """ Validates the correctness of the instance parameters and connectivity to Cisco WSA API service. Args: client (Client): Cisco WSA API client. """ client.url_categories_list() return "ok" """ HELPER FUNCTIONS """ def categories_output_filter(response: dict[str, Any], contain: str | None = None, type_: str | None = None) -> dict[str, Any]: """Filter categories response. Args: response (dict[str, Any]): API response. contain (str | None, optional): A string that contains in the categories. Defaults to None. type_ (str | None, optional): The type of the categories. Defaults to None. Returns: dict[str, Any]: Filtered output. """ outputs: dict[str, Any] = {} if all([not contain, not type_]): return response if type_ is not None: response = {type_: response.get(type_)}.copy() if not contain: return response if contain is not None: for key, categories in response.items(): for category in categories: if contain in category: if isinstance(outputs.get(key), list): outputs[key].append(category) else: outputs[key] = [category] return outputs def multi_status_delete_handler(response: Response, obj_key: str, readable_obj_name: str) -> List[CommandResults]: """Handling with 207 delete code response, in cases that some of the objects deleted and some didn't. Args: response (Response): API response from Cisco WSA (with 207 status code). obj_key (str): The key of the argument in the response. readable_obj_name (str): Readable name for the object. Returns: List[CommandResults]: Readable outputs for XSOAR. """ output_data = response.json() command_results_list = [] for profile in output_data.get("success_list"): readable_output = f'{readable_obj_name} "{profile.get(obj_key)}" was successfully deleted.' command_results_list.append(CommandResults(readable_output=readable_output)) for profile in output_data.get("failure_list"): readable_output = f'{readable_obj_name} "{profile.get(obj_key)}" deletion failed, message: "{profile.get("message")}".' command_results_list.append(CommandResults(readable_output=readable_output)) return command_results_list def identification_profile_mapper(data: List[dict[str, Any]]) -> List[dict[str, Any]]: """Map API identification profile response to XSOAR output. Args: data (List[dict[str, Any]]): API response from Cisco WSA. Returns: List[dict[str, Any]]: Identification profile output. """ filtered_data = [] for profile in data: filtered_data.append( { "status": profile["status"], "profile_name": profile["profile_name"], "description": profile["description"], "protocols": dict_safe_get(profile, ["members", "protocols"]), "order": profile["order"], "UrlCategories": { "predefined": dict_safe_get(profile, ["members", "url_categories", "predefined"]), "custom": dict_safe_get(profile, ["members", "url_categories", "custom"]), "uncategorized": dict_safe_get(profile, ["members", "url_categories", "uncategorized"]), }, "ip": dict_safe_get(profile, ["members", "ip"]), "proxy_port": dict_safe_get(profile, ["members", "proxy_port"]), "UserAgents": { "predefined": dict_safe_get(profile, ["members", "user_agents", "predefined"]), "custom": dict_safe_get(profile, ["members", "user_agents", "custom"]), }, } ) return remove_empty_elements(filtered_data) def access_policy_output_handler(response: List[dict[str, Any]]) -> List[dict[str, Any]]: """Handling with access policy handler (ISO 8061), updating to XSOAR standards. Args: response (List[dict[str, Any]]): Access policy response. Returns: List[dict[str, Any]]: Fixed outputs """ outputs = [] for policy in response: if (policy_expiry := policy.get("policy_expiry")) and (policy_datetime := arg_to_datetime(policy_expiry)): policy["policy_expiry"] = policy_datetime.strftime(ISO8601_CONFIG) outputs.append(policy) return outputs def pagination(response: List[dict[str, Any]], args: dict[str, Any]) -> List[dict[str, Any]]: """ Executing Manual paginate_results (using the page and page size arguments) Args: response (List[dict[str, Any]]): API response. args (dict[str, Any]): Command arguments from XSOAR. Returns: List[dict[str, Any]]: Paginated results. """ page = arg_to_number(args.get("page")) page_size = arg_to_number(args.get("page_size")) limit = arg_to_number(args.get("limit")) if limit and limit < 0: raise ValueError("Limit has to be positive number.") if page and page < 0: raise ValueError("page has to be positive number.") if page_size and page_size < 0: raise ValueError("page_size has to be positive number.") if (page and not page_size) or (not page and page_size): raise ValueError("Please insert page and page_size.") if page and page_size: offset = (page - 1) * page_size return response[offset : offset + page_size] else: return response[:limit] def organize_policy_object_data( objects: dict[str, Any], object_type: str | None, object_action: str | None, object_values: List[str] | None, block_custom_mime_types: List[str] | None, http_or_https_max_object_size_mb: int | None, ftp_max_object_size_mb: int | None, ): """ Organize policy object update data. Args: objects (dict[str, Any]): Original objects. object_type (str | None): Object type to update. object_action (str | None): Object action to update. object_values (List[str] | None): Object values to update. block_custom_mime_types (List[str] | None): Block custom MIME types. http_or_https_max_object_size_mb (int | None): HTTP(S) max object size MB. ftp_max_object_size_mb (int | None): FTP max object size MB. """ if object_type and object_action and object_values: original_obj_actions = dict_safe_get(objects, ["object_type", object_type]) if original_obj_actions: for original_obj_action in original_obj_actions: if original_obj_action == object_action: object_values.extend(dict_safe_get(objects, ["object_type", object_type, object_action])) else: original_obj_actions[original_obj_action] = [ value for value in original_obj_actions[original_obj_action] if value not in object_values ] objects["object_type"][object_type].update({object_action: object_values}) elif any([object_type, object_action, object_values]): raise ValueError("object_type, object_action, object_values should be used in conjunction.") if block_custom_mime_types: objects["block_custom_mime_types"] = block_custom_mime_types objects["max_object_size_mb"] = remove_empty_elements( { "http_or_https": http_or_https_max_object_size_mb, "ftp": ftp_max_object_size_mb, } ) def access_policy_objects_get(client: Client, policy_name: str) -> dict[str, Any]: """Get the objects data of access policy. Args: client (Client): Cisco WSA API client. policy_name (str): The access policy name. Raises: DemistoException: Policy was not found. DemistoException: Update failed, objects were not found. Returns: dict[str, Any]: Objects data. """ access_policies = client.access_policy_list(policy_name).get("access_policies", []) if not access_policies: raise DemistoException("Policy was not found.") if not access_policies[0].get("objects"): raise DemistoException("Update failed, objects were not found.") return access_policies[0].get("objects") def protocols_handler(protocols: List[str]) -> List[str]: """Protocols handler. Args: protocols (List[str]): User chosen protocol. Returns: List[str]: Organized protocols list. """ organized_protocols = [] if "HTTPS" in protocols: organized_protocols.extend(HTTPS_PROTOCOLS) if "SOCKS" in protocols: organized_protocols.extend(SOCKS_PROTOCOL) return organized_protocols def delete_handler( response: Response, obj_key: str, readable_obj_name: str, success_readable_output: str ) -> CommandResults | List[CommandResults]: """Handling with delete response. Args: response (Response): API response from Cisco WSA (with 207 status code). obj_key (str): The key of the argument in the response. readable_obj_name (str): Readable name for the object. success_readable_output (str): Readable output text for success. Returns: CommandResults | List[CommandResults]: Readable outputs for XSOAR. """ if response.status_code == HTTPStatus.MULTI_STATUS: return multi_status_delete_handler(response=response, obj_key=obj_key, readable_obj_name=readable_obj_name) return CommandResults(readable_output=success_readable_output) def main() -> None: params: dict[str, Any] = demisto.params() args: dict[str, Any] = demisto.args() base_url = params.get("base_url") username = params.get("credentials", {}).get("identifier") password = params.get("credentials", {}).get("password") verify_certificate: bool = not params.get("insecure", False) proxy = params.get("proxy", False) command = demisto.command() commands = { "cisco-wsa-access-policy-list": list_access_policy_command, "cisco-wsa-access-policy-create": create_access_policy_command, "cisco-wsa-access-policy-update": update_access_policy_command, "cisco-wsa-access-policy-protocols-user-agents-update": update_access_policy_protocols_user_agents_command, "cisco-wsa-access-policy-url-filtering-update": update_access_policy_url_filtering_command, "cisco-wsa-access-policy-applications-update": update_access_policy_applications_command, "cisco-wsa-access-policy-objects-update": update_access_policy_objects_command, "cisco-wsa-access-policy-anti-malware-update": update_access_policy_anti_malware_command, "cisco-wsa-access-policy-delete": delete_access_policy_command, "cisco-wsa-domain-map-list": list_domain_map_command, "cisco-wsa-domain-map-create": create_domain_map_command, "cisco-wsa-domain-map-update": update_domain_map_command, "cisco-wsa-domain-map-delete": delete_domain_map_command, "cisco-wsa-identification-profiles-list": list_identification_profiles_command, "cisco-wsa-identification-profiles-create": create_identification_profiles_command, "cisco-wsa-identification-profiles-update": update_identification_profiles_command, "cisco-wsa-identification-profiles-delete": delete_identification_profiles_command, "cisco-wsa-url-categories-list": list_url_categories_command, } try: client: Client = Client( urljoin(base_url, "/wsa/api"), username, password, verify_certificate, proxy, ) if command == "test-module": return_results(test_module(client)) elif command in commands: return_results(commands[command](client, args)) else: raise NotImplementedError(f"{command} command is not implemented.") except Exception as e: return_error(str(e)) if __name__ in ["__main__", "builtin", "builtins"]: main()