Salesforce Fusion IAM

import traceback

import urllib3
from CommonServerPython import *

# Disable insecure warnings
urllib3.disable_warnings()

ERROR_CODES_TO_SKIP = [404]

""" GLOBAL VARIABLES """
URI_PREFIX = "/services/data/v51.0/"

"""CLIENT CLASS"""


class Client(BaseClient):
    """A client class that implements logic to authenticate with the application."""

    def __init__(
        self,
        base_url,
        verify=True,
        proxy=False,
        ok_codes=(),
        headers=None,
        auth=None,
        client_id=None,
        username=None,
        password=None,
        client_secret=None,
    ):
        super().__init__(base_url, verify, proxy, ok_codes, headers, auth)
        self.client_id = client_id
        self.client_secret = client_secret
        self.username = username
        self.password = password
        self.headers = headers if headers else {}
        self.headers["Authorization"] = f"Bearer {self.create_login()}"

    def create_login(self):
        uri = "/services/oauth2/token"
        params = {
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "username": self.username,
            "password": self.password,
            "grant_type": "password",
        }

        res = self._http_request(method="POST", url_suffix=uri, params=params, headers=self.headers)

        return res.get("access_token")

    def test(self):
        """Tests connectivity with the application."""

        uri = URI_PREFIX + "sobjects/User/testid"
        self._http_request(method="GET", url_suffix=uri, ok_codes=(200, 404), headers=self.headers)

    def get_user_by_id(self, user_id: str) -> Optional["IAMUserAppData"]:
        """Queries the user in the application using REST API by its ID, and returns an IAMUserAppData object
        that holds the user_id, username, is_active and app_data attributes given in the query response.

        :type user_id: ``str``
        :param user_id: ID of the user in the application.

        :return: An IAMUserAppData object if user exists, None otherwise.
        :rtype: ``Optional[IAMUserAppData]``
        """
        uri = f"{URI_PREFIX}sobjects/FF__Key_Contact__c/{user_id}"

        res = self._http_request(method="GET", url_suffix=uri, headers=self.headers)
        if res:
            email = res.get("Work_Email__c")
            username = res.get("Name")
            is_active = True

            return IAMUserAppData(user_id, username, is_active, res, email=email)
        return None

    def get_user(self, filter_name: str, filter_value: str) -> Optional["IAMUserAppData"]:
        if filter_name == "id":
            return self.get_user_by_id(filter_value)
        else:
            uri = f"{URI_PREFIX}parameterizedSearch"
            params = {
                "q": filter_value,
                "sobject": "FF__Key_Contact__c",
                "FF__Key_Contact__c.where": f"{filter_name}='{filter_value}'",
                "FF__Key_Contact__c.fields": "Id, FF__First_Name__c, FF__Last_Name__c, Work_Email__c, Name",
            }

        res = self._http_request(method="GET", url_suffix=uri, params=params, headers=self.headers)

        user_app_data = res.get("searchRecords", [])

        if user_app_data:
            user_id = user_app_data[0].get("Id")
            return self.get_user_by_id(user_id)
        return None

    def create_user(self, user_data: Dict[str, Any]) -> "IAMUserAppData":
        """Creates a user in the application using REST API.

        :type user_data: ``Dict[str, Any]``
        :param user_data: User data in the application format

        :return: An IAMUserAppData object that contains the data of the created user in the application.
        :rtype: ``IAMUserAppData``
        """
        uri = f"{URI_PREFIX}sobjects/FF__Key_Contact__c"
        res = self._http_request(method="POST", url_suffix=uri, json_data=user_data, headers=self.headers)
        user_id = res.get("id")
        email = res.get("Work_Email__c")
        username = res.get("Name")
        is_active = True

        return IAMUserAppData(user_id, username, is_active, res, email=email)

    def update_user(self, user_id: str, user_data: Dict[str, Any]):
        """Updates a user in the application using REST API.

        :type user_id: ``str``
        :param user_id: ID of the user in the application

        :type user_data: ``Dict[str, Any]``
        :param user_data: User data in the application format

        :return: An IAMUserAppData object that contains the data of the updated user in the application.
        :rtype: ``IAMUserAppData``
        """
        uri = f"{URI_PREFIX}sobjects/FF__Key_Contact__c/{user_id}"
        params = {"_HttpMethod": "PATCH"}
        self._http_request(
            method="POST",
            url_suffix=uri,
            params=params,
            json_data=user_data,
            headers=self.headers,
            resp_type="response",
            ok_codes=(
                200,
                204,
            ),
        )

        return self.get_user_by_id(user_id)

    def enable_user(self, user_id: str):
        """Enables a user in the application using REST API.
        There is no action on the user for Salesforce Fusion instance needs to be taken.

        :type user_id: ``str``
        :param user_id: ID of the user in the application

        :return: An IAMUserAppData object that contains the data of the user in the application.
        :rtype: ``IAMUserAppData``
        """

        return self.get_user_by_id(user_id)

    def disable_user(self, user_id: str) -> "IAMUserAppData":
        """Removes a user in the application using REST API.

        :type user_id: ``str``
        :param user_id: ID of the user in the application

        :return: An IAMUserAppData object that contains the data of the user in the application.
        :rtype: ``IAMUserAppData``
        """

        uri = f"{URI_PREFIX}sobjects/FF__Key_Contact__c/{user_id}"

        res = self._http_request(
            method="DELETE",
            url_suffix=uri,
            headers=self.headers,
            ok_codes=(
                200,
                204,
            ),
            resp_type="response",
        )
        if res.status_code == 204:
            return IAMUserAppData(user_id, "", False, {})
        raise DemistoException(f"Could not delete user. Response was: {res}")

    def get_app_fields(self) -> Dict[str, Any]:
        """Gets a dictionary of the user schema fields in the application and their description.

        :return: The user schema fields dictionary
        :rtype: ``Dict[str, str]``
        """

        uri = f"{URI_PREFIX}sobjects/FF__Key_Contact__c/describe/"
        res = self._http_request(method="GET", url_suffix=uri, headers=self.headers)

        fields = res.get("fields", [])
        return {field.get("name"): field.get("label") for field in fields}

    @staticmethod
    def handle_exception(user_profile, e, action):
        """Handles failed responses from the application API by setting the User Profile object with the result.
            The result entity should contain the following data:
            1. action        (``IAMActions``)       The failed action                       Required
            2. success       (``bool``)             The success status                      Optional (by default, True)
            3. skip          (``bool``)             Whether or not the command was skipped  Optional (by default, False)
            3. skip_reason   (``str``)              Skip reason                             Optional (by default, None)
            4. error_code    (``Union[str, int]``)  HTTP error code                         Optional (by default, None)
            5. error_message (``str``)              The error description                   Optional (by default, None)

            Note: This is the place to determine how to handle specific edge cases from the API, e.g.,
            when a DISABLE action was made on a user which is already disabled and therefore we can't
            perform another DISABLE action.

        :type user_profile: ``IAMUserProfile``
        :param user_profile: The user profile object

        :type e: ``Union[DemistoException, Exception]``
        :param e: The exception object - if type is DemistoException, holds the response json object (`res` attribute)

        :type action: ``IAMActions``
        :param action: An enum represents the current action (GET, UPDATE, CREATE, DISABLE or ENABLE)
        """
        if isinstance(e, DemistoException) and e.res is not None:
            error_code = e.res.status_code

            if action == IAMActions.DISABLE_USER and error_code in ERROR_CODES_TO_SKIP:
                skip_message = "Users is already disabled or does not exist in the system."
                user_profile.set_result(action=action, skip=True, skip_reason=skip_message)

            try:
                resp = e.res.json()
                error_message = get_error_details(resp)
            except ValueError:
                error_message = str(e)
        else:
            error_code = ""
            error_message = str(e)

        user_profile.set_result(action=action, success=False, error_code=error_code, error_message=error_message)

        demisto.error(traceback.format_exc())


"""HELPER FUNCTIONS"""


def get_error_details(res: Dict[str, Any]) -> str:
    """Parses the error details retrieved from the application and outputs the resulted string.

    :type res: ``Dict[str, Any]``
    :param res: The error data retrieved from the application.

    :return: The parsed error details.
    :rtype: ``str``
    """
    message, details = "Couldn't find details for the error", ""
    if isinstance(res, list):
        res = res[0]
    if isinstance(res, dict):
        message = res.get("message", "No error message was supplied")
        details = res.get("detail", "No details regarding the error were supplied")
    return f"{message}: {details}"


"""COMMAND FUNCTIONS"""


def test_module(client: Client):
    """Tests connectivity with the client."""

    client.test()
    return_results("ok")


def get_mapping_fields(client: Client) -> GetMappingFieldsResponse:
    """Creates and returns a GetMappingFieldsResponse object of the user schema in the application

    :param client: (Client) The integration Client object that implements a get_app_fields() method
    :return: (GetMappingFieldsResponse) An object that represents the user schema
    """
    app_fields = client.get_app_fields()
    incident_type_scheme = SchemeTypeMapping(type_name=IAMUserProfile.DEFAULT_INCIDENT_TYPE)

    for field, description in app_fields.items():
        incident_type_scheme.add_field(field, description)

    return GetMappingFieldsResponse([incident_type_scheme])


def main():
    user_profile = None
    params = demisto.params()
    base_url = urljoin(params["url"].strip("/"))
    client_id = params.get("client_id_creds", {}).get("identifier") or params.get("client_id")
    client_secret = params.get("client_id_creds", {}).get("password") or params.get("client_secret")
    username = params.get("credentials", {}).get("identifier")
    password = params.get("credentials", {}).get("password")
    secret_token = params.get("secret_token_creds", {}).get("password") or params.get("secret_token")
    mapper_in = params.get("mapper_in")
    mapper_out = params.get("mapper_out")
    verify_certificate = not params.get("insecure", False)
    proxy = params.get("proxy", False)
    command = demisto.command()
    args = demisto.args()

    is_create_enabled = params.get("create_user_enabled")
    is_enable_enabled = params.get("enable_user_enabled")
    is_disable_enabled = params.get("disable_user_enabled")
    is_update_enabled = params.get("update_user_enabled")
    create_if_not_exists = params.get("create_if_not_exists")

    iam_command = IAMCommand(
        is_create_enabled,
        is_enable_enabled,
        is_disable_enabled,
        is_update_enabled,
        create_if_not_exists,
        mapper_in,
        mapper_out,
        get_user_iam_attrs=["id", "Name", "Work_Email__c"],
    )

    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
    }

    client = Client(
        base_url=base_url,
        verify=verify_certificate,
        proxy=proxy,
        headers=headers,
        ok_codes=(200, 201),
        client_id=client_id,
        client_secret=client_secret,
        username=username,
        password=password + secret_token if password and secret_token else None,
    )

    demisto.debug(f"Command being called is {command}")

    """CRUD commands"""

    if command == "iam-get-user":
        user_profile = iam_command.get_user(client, args)

    elif command == "iam-create-user":
        user_profile = iam_command.create_user(client, args)

    elif command == "iam-update-user":
        user_profile = iam_command.update_user(client, args)

    elif command == "iam-disable-user":
        user_profile = iam_command.disable_user(client, args)

    if user_profile:
        return_results(user_profile)

    """non-CRUD commands"""

    try:
        if command == "test-module":
            test_module(client)

        elif command == "get-mapping-fields":
            return_results(get_mapping_fields(client))

    except Exception as exc:
        # For any other integration command exception, return an error
        return_error(f"Failed to execute the {command} command.\nError: {exc}")


from IAMApiModule import *  # noqa E402

if __name__ in ("__main__", "__builtin__", "builtins"):  # pragma: no cover
    main()