Azure Security Center v2

import ast

import demistomock as demisto
from CommonServerPython import *
from MicrosoftApiModule import *  # noqa: E402

""" GLOBAL VARS """

APP_NAME = "ms-azure-sc"
SUB_ID_REQUIRING_CMD = (
    "azure-sc-get-alert",
    "azure-sc-list-alert",
    "azure-sc-update-alert",
    "azure-sc-list-location",
    "azure-sc-update-atp",
    "azure-sc-get-atp",
    "azure-sc-update-aps",
    "azure-sc-list-aps",
    "azure-sc-get-aps",
    "azure-sc-list-jit",
    "azure-sc-get-jit",
    "azure-sc-initiate-jit",
    "azure-sc-delete-jit",
    "azure-sc-list-storage",
)
RESOURCE_GROUP_REQUIRING_CMD = ("azure-sc-update-atp", "azure-sc-get-atp")
DEFAULT_LIMIT = 50
# API Versions
SUBSCRIPTION_API_VERSION = "2015-01-01"
ALERT_API_VERSION = "2022-01-01"
LOCATION_API_VERSION = "2015-06-01-preview"
ATP_API_VERSION = "2017-08-01-preview"
APS_API_VERSION = "2017-08-01-preview"
IPP_API_VERSION = "2017-08-01-preview"
JIT_API_VERSION = "2015-06-01-preview"
STORAGE_API_VERSION = "2018-07-01"
SECURE_STORES_API_VERSION = "2020-01-01"
LIST_RESOURCE_GROUP_VERSION = "2021-04-01"

""" HELPER FUNCTIONS """


# Format ports in JIT access policy rule to (portNum, protocol, allowedAddress, maxDuration)
def format_jit_port_rule(ports):
    port_array = []
    for port in ports:
        # for each item in unicode, has to use str to decode to ascii
        p_num = str(port.get("number"))
        p_src_addr = str(port.get("allowedSourceAddressPrefix")) if port.get("allowedSourceAddressPrefix") != "*" else "any"
        p_protocol = str(port.get("protocol")) if port.get("protocol") != "*" else "any"
        p_max_duration = str(port.get("maxRequestAccessDuration"))
        port_array.append(str((p_num, p_protocol, p_src_addr, p_max_duration)))
    return ", ".join(port_array)


# Format ports in JIT access request to (portNum, allowedAddress, endTime, status)
def format_jit_port_request(ports):
    port_array = []
    for port in ports:
        # for each item in unicode, has to use str to decode to ascii
        p_num = str(port.get("number"))
        p_src_addr = str(port.get("allowedSourceAddressPrefix")) if port.get("allowedSourceAddressPrefix") != "*" else "any"
        p_status = str(port.get("status"))
        p_end_time = str(port.get("endTimeUtc"))
        port_array.append(str((p_num, p_src_addr, p_end_time, p_status)))
    return ", ".join(port_array)


def normalize_context_key(string):
    """Normalize context keys
    Function will normalize the string (remove white spaces and tailings)
    Args:
        string (str):
    Returns:
        Normalized string
    """
    tmp = string[:1].upper() + string[1:]
    return tmp.replace(" ", "")


class MsClient:
    """
    Microsoft Client enables authorized access to Azure Security Center.
    """

    def __init__(
        self,
        tenant_id,
        auth_id,
        enc_key,
        app_name,
        server,
        verify,
        proxy,
        self_deployed,
        subscription_id,
        ok_codes,
        certificate_thumbprint,
        private_key,
        resource_group_name=None,
        managed_identities_client_id=None,
    ):
        base_url_with_subscription = f"{server}subscriptions/{subscription_id}/"
        self.ms_client = MicrosoftClient(
            tenant_id=tenant_id,
            auth_id=auth_id,
            enc_key=enc_key,
            app_name=app_name,
            base_url=base_url_with_subscription,
            verify=verify,
            proxy=proxy,
            self_deployed=self_deployed,
            ok_codes=ok_codes,
            scope="https://management.azure.com/.default",
            certificate_thumbprint=certificate_thumbprint,
            private_key=private_key,
            managed_identities_client_id=managed_identities_client_id,
            managed_identities_resource_uri=Resources.management_azure,
            command_prefix="azure-sc",
        )
        self.server = server
        self.subscription_id = subscription_id
        self.resource_group_name = resource_group_name

    def get_alert(self, asc_location, alert_id):
        """
        Args:
            asc_location (str): Azure Security Center location
            alert_id (str): Alert ID

        Returns:
            response body (dict)

        """
        cmd_url = f"/resourceGroups/{self.resource_group_name}" if self.resource_group_name else ""
        cmd_url += f"/providers/Microsoft.Security/locations/{asc_location}/alerts/{alert_id}"
        params = {"api-version": ALERT_API_VERSION}
        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def list_alerts(self, asc_location, filter_query, select_query, expand_query):
        """Listing alerts

        Args:
            asc_location (str): Azure Security Center location
            filter_query (str): what to filter
            select_query (str): what to select
            expand_query (str): what to expand

        Returns:
            dict: contains response body
        """
        if self.resource_group_name:
            cmd_url = f"/resourceGroups/{self.resource_group_name}/providers/Microsoft.Security"
            # ascLocation must be using with specifying resourceGroupName
            if asc_location:
                cmd_url += f"/locations/{asc_location}"
            cmd_url += "/alerts"
        else:
            cmd_url = "/providers/Microsoft.Security/alerts"

        params = {"api-version": ALERT_API_VERSION}
        if filter_query:
            params["$filter"] = filter_query
        if select_query:
            params["$select"] = select_query
        if expand_query:
            params["$expand"] = expand_query

        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def update_alert(self, asc_location, alert_id, alert_update_action_type):
        """
        Args:
            asc_location (str): Azure Security Center Location
            alert_id (str): Alert ID
            alert_update_action_type (str): What update type need to update

        Returns:
            dict: response body
        """
        cmd_url = f"/resourceGroups/{self.resource_group_name}" if self.resource_group_name else ""
        cmd_url += f"/providers/Microsoft.Security/locations/{asc_location}/alerts/{alert_id}/{alert_update_action_type}"
        params = {"api-version": ALERT_API_VERSION}
        #  Using resp_type=response to avoid parsing error.
        self.ms_client.http_request(method="POST", url_suffix=cmd_url, params=params, resp_type="response")

    def list_locations(self):
        """
        Returns:
            dict: response body
        """
        cmd_url = "/providers/Microsoft.Security/locations"
        params = {"api-version": LOCATION_API_VERSION}
        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def update_atp(self, storage_account, setting_name, is_enabled):
        """
        Args:
            storage_account (str): Storange Account
            setting_name (str):  Setting Name
            is_enabled (str): true/false

        Returns:
            dict: respones body
        """
        cmd_url = (
            f"/resourceGroups/{self.resource_group_name}/providers/Microsoft.Storage/storageAccounts/"
            f"{storage_account}/providers/Microsoft.Security/advancedThreatProtectionSettings/{setting_name}"
        )
        params = {"api-version": ATP_API_VERSION}
        data = {
            "id": f"/subscriptions/{self.subscription_id}/resourceGroups/{self.resource_group_name}/providers/"
            f"Microsoft.Storage/storageAccounts/{storage_account}/providers/Microsoft.Security/"
            f"advancedThreatProtectionSettings/{setting_name}",
            "name": setting_name,
            "type": "Microsoft.Security/advancedThreatProtectionSettings",
            "properties": {"isEnabled": is_enabled},
        }

        #  Using resp_type=response to avoid parsing error.
        return self.ms_client.http_request(method="PUT", url_suffix=cmd_url, json_data=data, params=params)

    def get_atp(self, storage_account, setting_name):
        """
        Args:
            storage_account (str): Storange Account
            setting_name (str):  Setting Name

        Returns:

        """
        cmd_url = (
            f"/resourceGroups/{self.resource_group_name}/providers/Microsoft.Storage/storageAccounts"
            f"/{storage_account}/providers/Microsoft.Security/advancedThreatProtectionSettings/{setting_name}"
        )
        params = {"api-version": ATP_API_VERSION}
        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def update_aps(self, setting_name, auto_provision):
        """
        Args:
            setting_name (str): Setting name
            auto_provision (str): Auto provision setting (On/Off)

        Returns:
            dict: response body
        """
        cmd_url = f"/providers/Microsoft.Security/autoProvisioningSettings/{setting_name}"
        params = {"api-version": APS_API_VERSION}

        data = {"properties": {"autoProvision": auto_provision}}

        return self.ms_client.http_request(method="PUT", url_suffix=cmd_url, json_data=data, params=params)

    def list_aps(self):
        """
        Returns:
            dict: response body
        """
        cmd_url = "/providers/Microsoft.Security/autoProvisioningSettings"
        params = {"api-version": APS_API_VERSION}
        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def get_aps(self, setting_name):
        """
        Args:
            setting_name: Setting name

        Returns:
            dict: response body
        """
        cmd_url = f"/providers/Microsoft.Security/autoProvisioningSettings/{setting_name}"
        params = {"api-version": APS_API_VERSION}

        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def list_ipp(self, management_group=None):
        """
        Args:
            management_group: Managment group to pull (if needed)

        Returns:
            dict: response body

        """
        params = {"api-version": IPP_API_VERSION}
        cmd_url = "/providers/Microsoft.Security/informationProtectionPolicies"
        if management_group:
            full_url = f"{self.server}/providers/Microsoft.Management/managementGroups/{management_group}"
            full_url += cmd_url
            return self.ms_client.http_request(method="GET", full_url=full_url, url_suffix="", params=params)
        if not self.subscription_id:
            raise DemistoException("A subscription ID must be provided.")
        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def get_ipp(self, policy_name, management_group):
        """
        Args:
            policy_name (str): Policy name
            management_group (str): Managment group

        Returns:
            dict: respone body
        """
        params = {"api-version": IPP_API_VERSION}

        cmd_url = f"/providers/Microsoft.Security/informationProtectionPolicies/{policy_name}"
        if management_group:
            full_url = f"{self.server}/providers/Microsoft.Management/managementGroups/{management_group}"
            full_url += cmd_url
            return self.ms_client.http_request(method="GET", full_url=full_url, url_suffix="", params=params)
        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def list_jit(self, asc_location):
        """
        Args:
            asc_location: Machine location

        Returns:
            dict: response body
        """
        params = {"api-version": JIT_API_VERSION}
        cmd_url = f"/resourceGroups/{self.resource_group_name}" if self.resource_group_name else ""
        cmd_url += f"/providers/Microsoft.Security/locations/{asc_location}" if asc_location else ""
        cmd_url += "/providers/Microsoft.Security/jitNetworkAccessPolicies"
        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def get_jit(self, policy_name, asc_location, resource_group_name):
        """
        Args:
            policy_name: Policy name
            asc_location: Machine location
            resource_group_name: Resource name group

        Returns:
            dict: response body
        """
        cmd_url = (
            f"/resourceGroups/{resource_group_name}/providers/Microsoft.Security/locations/{asc_location}/"
            f"jitNetworkAccessPolicies/{policy_name}"
        )
        params = {"api-version": JIT_API_VERSION}

        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def initiate_jit(self, resource_group_name, asc_location, policy_name, vm_id, port, source_address, duration):
        """Starting new Just-in-time machine

        Args:
            resource_group_name: Resource group name
            asc_location: Machine location
            policy_name: Policy name
            vm_id: Virtual Machine ID
            port: ports to be used
            source_address: Source address
            duration: Time in

        Returns:
            dict: response body
        """
        cmd_url = (
            f"/resourceGroups/{resource_group_name}/providers/Microsoft.Security/locations/{asc_location}/"
            f"jitNetworkAccessPolicies/{policy_name}/initiate"
        )
        params = {"api-version": JIT_API_VERSION}

        # only supports init access for one vm and one port now
        data = {
            "virtualMachines": [
                {
                    "ID": vm_id,
                    "ports": [
                        {
                            "number": port,
                            "duration": duration,
                            "allowedSourceAddressPrefix": source_address,
                        }
                    ],
                }
            ]
        }
        # response code should be 202 Accepted
        return self.ms_client.http_request(method="POST", url_suffix=cmd_url, json_data=data, params=params, resp_type="response")

    def delete_jit(self, asc_location, resource_group_name, policy_name):
        """
        Args:
            asc_location: Machine location
            resource_group_name: Resource group name
            policy_name: Policy name
        """
        cmd_url = (
            f"/resourceGroups/{resource_group_name}/providers/Microsoft.Security/locations/{asc_location}/"
            f"jitNetworkAccessPolicies/{policy_name}"
        )

        params = {"api-version": JIT_API_VERSION}

        #  Using resp_type=text to avoid parsing error. response should be 204
        self.ms_client.http_request(method="DELETE", url_suffix=cmd_url, params=params, resp_type="text")

    def list_sc_storage(self):
        """
        Returns:
            dict: response body

        """
        cmd_url = "/providers/Microsoft.Storage/storageAccounts"
        params = {"api-version": STORAGE_API_VERSION}
        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def list_sc_subscriptions(self):
        """
        Returns:
            dict: response body

        """
        full_url = f"{self.server}/subscriptions"
        params = {"api-version": SUBSCRIPTION_API_VERSION}
        return self.ms_client.http_request(method="GET", full_url=full_url, url_suffix="", params=params)

    def get_secure_scores(self, secure_score_name):
        """
        Returns:
            dict: response body

        """

        cmd_url = f"/providers/Microsoft.Security/secureScores/{secure_score_name}"
        params = {"api-version": SECURE_STORES_API_VERSION}
        return self.ms_client.http_request(method="GET", url_suffix=cmd_url, params=params)

    def list_resource_groups(self, tag: str, limit: int, full_url: Optional[str] = None) -> dict:
        """
        List all resource groups.

        Args:
            tag str: Tag to filter by.
            limit (int): Maximum number of resource groups to retrieve. Default is 50.

        Returns:
            List[dict]: API response from Azure.
        """
        filter_by_tag = azure_tag_formatter(tag) if tag else None
        params = {"$filter": filter_by_tag, "$top": limit, "api-version": LIST_RESOURCE_GROUP_VERSION} if not full_url else {}
        return self.ms_client.http_request(method="GET", url_suffix="resourcegroups", params=params, full_url=full_url)


""" FUNCTIONS """

""" Alert Start """


def get_alert_command(client: MsClient, args: dict):
    """Getting specified alert from API
    Args
        args (dict): dictionary containing commands args
    """
    asc_location = args.get("asc_location")
    alert_id = args.get("alert_id")
    alert = client.get_alert(asc_location, alert_id)
    final_output = []

    # Basic Property Table
    properties = alert.get("properties")
    if properties:
        basic_table_output = [
            {
                "DisplayName": properties.get("alertDisplayName"),
                "CompromisedEntity": properties.get("compromisedEntity"),
                "Description": properties.get("description"),
                "DetectedTime": properties.get("timeGeneratedUtc"),
                "ReportedSeverity": properties.get("severity"),
                "State": properties.get("status"),
                "RemediationSteps": properties.get("remediationSteps"),
                "VendorName": properties.get("vendorName"),
                "ID": alert.get("name"),
                "ExtendedProperties": properties.get("extendedProperties"),
                "Entities": properties.get("entities"),
            }
        ]

        md = tableToMarkdown(
            "Azure Security Center - Get Alert - Basic Property",
            basic_table_output,
            [
                "DisplayName",
                "CompromisedEntity",
                "Description",
                "DetectedTime",
                "ReportedSeverity",
                "State",
                "RemediationSteps",
                "VendorName",
                "AlertName",
                "ID",
            ],
            removeNull=True,
        )

        ec = {"AzureSecurityCenter.Alert(val.ID && val.ID === obj.ID)": basic_table_output}

        basic_table_entry = {
            "Type": entryTypes["note"],
            "Contents": alert,
            "ContentsFormat": formats["json"],
            "ReadableContentsFormat": formats["markdown"],
            "HumanReadable": md,
            "EntryContext": ec,
        }
        final_output.append(basic_table_entry)

        # Extended Properties Table
        if alert.get("properties") and alert.get("properties") and alert.get("properties").get("extendedProperties"):
            extended_properties = {}
            properties = alert.get("properties")
            if isinstance(properties.get("extendedProperties"), dict):
                for key, value in alert["properties"]["extendedProperties"].items():
                    extended_properties[normalize_context_key(key)] = value
                extended_table_entry = {
                    "Type": entryTypes["note"],
                    "Contents": alert["properties"]["extendedProperties"],
                    "ContentsFormat": formats["json"],
                    "ReadableContentsFormat": formats["markdown"],
                    "HumanReadable": tableToMarkdown(
                        "Azure Security Center - Get Alert - Extended Property",
                        extended_properties,
                        removeNull=True,
                    ),
                }
                final_output.append(extended_table_entry)

            # Entities Table
            entities = properties.get("entities")
            if entities and isinstance(entities, dict):
                entities_table_output = []
                for entity in entities:
                    entities_table_output.append(
                        {
                            "Content": ast.literal_eval(str(entity)),
                            "Type": entity["type"],
                        }
                    )

                md = tableToMarkdown(
                    "Azure Security Center - Get Alert - Entity",
                    entities_table_output,
                    removeNull=True,
                )

                entities_table_entry = {
                    "Type": entryTypes["note"],
                    "Contents": alert.get("properties").get("entities"),
                    "ContentsFormat": formats["json"],
                    "ReadableContentsFormat": formats["markdown"],
                    "HumanReadable": md,
                }
                final_output.append(entities_table_entry)
    return final_output


def list_alerts_command(client: MsClient, args: dict):
    """Getting all alerts

    Args:
        client:
        args (dict): usually demisto.args()
    """
    asc_location = args.get("asc_location")
    filter_query = args.get("filter")
    select_query = args.get("select")
    expand_query = args.get("expand")

    alerts = client.list_alerts(asc_location, filter_query, select_query, expand_query).get("value")
    outputs = []
    for alert in alerts:
        properties = alert.get("properties")
        if properties:
            outputs.append(
                {
                    "DisplayName": properties.get("alertDisplayName"),
                    "CompromisedEntity": properties.get("compromisedEntity"),
                    "Description": properties.get("description"),
                    "ID": alert.get("name"),
                    "DetectedTime": properties.get("timeGeneratedUtc"),
                    "ReportedSeverity": properties.get("severity"),
                    "State": properties.get("status"),
                }
            )

    md = tableToMarkdown(
        "Azure Security Center - List Alerts",
        outputs,
        [
            "DisplayName",
            "CompromisedEntity",
            "DetectedTime",
            "ReportedSeverity",
            "State",
            "Description",
            "ID",
        ],
        removeNull=True,
    )
    ec = {"AzureSecurityCenter.Alert(val.ID && val.ID === obj.ID)": outputs}
    return md, ec, alerts


# There's a Microsoft API bug for reactivate alert -
# https://social.msdn.microsoft.com/Forums/windows/en-US/c2139e1b-b26c-4264-a558-fa4b180b70e7/issue-while-setting-security-alert-state-from-dismiss-to-active?forum=AzureSecurityCenter
def update_alert_command(client: MsClient, args: dict):
    """Update given alert

    Args:
        client: MsClient
        args (dict): usually demisto.args()
    """
    asc_location = args.get("asc_location")
    alert_id = args.get("alert_id")
    alert_update_action_type = args.get("alert_update_action_type")
    if alert_update_action_type == "in_progress":
        alert_update_action_type = "inProgress"
    client.update_alert(asc_location, alert_id, alert_update_action_type)
    outputs = {"ID": alert_id, "ActionTaken": alert_update_action_type}

    ec = {"AzureSecurityCenter.Alert(val.ID && val.ID === obj.ID)": outputs}
    return f"Alert - {alert_id} has been set to {alert_update_action_type}.", ec, None


""" Alert End """

""" Location Start """


def list_locations_command(client: MsClient):
    """Getting all locations"""
    locations = client.list_locations().get("value")
    outputs = []
    if locations:
        for location in locations:
            if location.get("properties") and location.get("properties").get("homeRegionName"):
                home_region_name = location.get("properties").get("homeRegionName")
            else:
                home_region_name = None
            outputs.append(
                {
                    "HomeRegionName": home_region_name,
                    "Name": location.get("name"),
                    "ID": location.get("id"),
                }
            )
            md = tableToMarkdown(
                "Azure Security Center - List Locations",
                outputs,
                ["HomeRegionName", "Name", "ID"],
                removeNull=True,
            )
            ec = {"AzureSecurityCenter.Location(val.ID && val.ID === obj.ID)": outputs}
            return md, ec, locations
        return None
    else:
        return "No locations found", None, None


""" Location End """

""" Advanced Threat Protection Start """


def update_atp_command(client: MsClient, args: dict):
    """Updating given Advanced Threat Protection (enable/disable)

    Args:
        client:
        args (dict): usually demisto.args()
    """
    setting_name = args.get("setting_name")
    is_enabled = args.get("is_enabled")
    storage_account = args.get("storage_account")
    response = client.update_atp(storage_account, setting_name, is_enabled)
    outputs = {
        "ID": response.get("id"),
        "Name": response.get("name"),
        "IsEnabled": response.get("properties").get("is_enabled"),
    }
    md = tableToMarkdown(
        "Azure Security Center - Update Advanced Threat Detection Setting",
        outputs,
        ["ID", "Name", "IsEnabled"],
        removeNull=True,
    )
    ec = {"AzureSecurityCenter.AdvancedThreatProtection(val.ID && val.ID === obj.ID)": outputs}
    return md, ec, response


def get_atp_command(client: MsClient, args: dict):
    """Get given Advanced Threat Protection settings

    Args:
        client:
        args (dict): usually demisto.args()
    """
    setting_name = args.get("setting_name")
    storage_account = args.get("storage_account")
    response = client.get_atp(storage_account, setting_name)
    outputs = {
        "ID": response.get("id"),
        "Name": response.get("name"),
        "IsEnabled": response["properties"]["isEnabled"]
        if response.get("properties") and response.get("properties").get("isEnabled")
        else None,
    }
    md = tableToMarkdown(
        "Azure Security Center - Get Advanced Threat Detection Setting",
        outputs,
        ["ID", "Name", "IsEnabled"],
        removeNull=True,
    )
    ec = {"AzureSecurityCenter.AdvancedThreatProtection(val.ID && val.ID === obj.ID)": outputs}
    return md, ec, response


""" Advanced Threat Protection End """

""" Auto Provisioning Settings Start """


def update_aps_command(client: MsClient, args: dict):
    """Updating Analytics Platform System

    Args:
        client:
        args (dict): usually demisto.args()
    """
    setting_name = args.get("setting_name")
    auto_provision = args.get("auto_provision")
    setting = client.update_aps(setting_name, auto_provision)
    outputs = [
        {
            "Name": setting.get("name"),
            "AutoProvision": setting["properties"]["auto_provision"]
            if setting.get("properties") and setting.get("properties").get("auto_provision")
            else None,
            "ID": setting.get("id"),
        }
    ]

    md = tableToMarkdown(
        "Azure Security Center - Update Auto Provisioning Setting",
        outputs,
        ["Name", "AutoProvision", "ID"],
        removeNull=True,
    )
    ec = {"AzureSecurityCenter.AutoProvisioningSetting(val.ID && val.ID === obj.ID)": outputs}
    return md, ec, setting


def list_aps_command(client: MsClient):
    """List all Analytics Platform System"""
    settings = client.list_aps().get("value")
    outputs = []
    for setting in settings:
        outputs.append(
            {
                "Name": setting.get("name"),
                "AutoProvision": setting.get("properties").get("autoProvision")
                if setting.get("properties") and setting.get("properties").get("autoProvision")
                else None,
                "ID": setting.get("id"),
            }
        )

    md = tableToMarkdown(
        "Azure Security Center - List Auto Provisioning Settings",
        outputs,
        ["Name", "AutoProvision", "ID"],
        removeNull=True,
    )

    ec = {"AzureSecurityCenter.AutoProvisioningSetting(val.ID && val.ID === obj.ID)": outputs}
    return md, ec, settings


def get_aps_command(client: MsClient, args: dict):
    """Get given Analytics Platform System setting

    Args:
        client:
        args (dict): usually demisto.args()
    """
    setting_name = args.get("setting_name")
    setting = client.get_aps(setting_name)
    outputs = [
        {
            "Name": setting.get("name"),
            "AutoProvision": setting.get("properties").get("autoProvision")
            if setting.get("properties") and setting.get("properties").get("autoProvision")
            else None,
            "ID": setting["id"],
        }
    ]
    md = tableToMarkdown(
        "Azure Security Center - Get Auto Provisioning Setting",
        outputs,
        ["Name", "AutoProvision", "ID"],
        removeNull=True,
    )
    ec = {"AzureSecurityCenter.AutoProvisioningSetting(val.ID && val.ID === obj.ID)": outputs}

    return md, ec, setting


""" Auto Provisioning Settings End """

""" Information Protection Policies Start """


# Unsupported command. issue: issues/24583
def list_ipp_command(client: MsClient, args: dict):
    """Listing all Internet Presence Provider

    Args:
        client:
        args (dict): usually demisto.args()
    """
    management_group = args.get("management_group")
    policies = client.list_ipp(management_group).get("value")
    outputs = []
    if policies:
        for policy in policies:
            if policy.get("properties") and policy.get("properties").get("labels"):
                label_names = ", ".join([label.get("displayName") for label in policy["properties"]["labels"].values()])
                information_type_names = ", ".join(
                    [it["displayName"] for it in policy["properties"]["informationTypes"].values()]
                )
            else:
                label_names, information_type_names = "", ""
            outputs.append(
                {
                    "Name": policy.get("name"),
                    "Labels": label_names,
                    "InformationTypeNames": information_type_names,
                    "InformationTypes": policy.get("properties").get("informationTypes")
                    if policy.get("properties") and policy.get("properties").get("informationTypes")
                    else None,
                    "ID": policy["id"],
                }
            )
        md = tableToMarkdown(
            "Azure Security Center - List Information Protection Policies",
            outputs,
            ["Name", "Labels", "InformationTypeNames", "ID"],
            removeNull=True,
        )

        ec = {"AzureSecurityCenter.InformationProtectionPolicy(val.ID && val.ID === obj.ID)": outputs}

        entry = {
            "Type": entryTypes["note"],
            "Contents": policies,
            "ContentsFormat": formats["json"],
            "ReadableContentsFormat": formats["markdown"],
            "HumanReadable": md,
            "EntryContext": ec,
        }
        demisto.results(entry)
    else:
        demisto.results("No policies found")


# Unsupported command. issue: issues/24583
def get_ipp_command(client: MsClient, args: dict):
    """Getting Internet Presence Provider information
    Args:
        client:
        args (dict): usually demisto.args()
    """
    policy_name = args.get("policy_name")
    management_group = args.get("management_group")
    policy = client.get_ipp(policy_name, management_group)
    properties = policy.get("properties")
    labels = properties.get("labels")
    if properties and isinstance(labels, dict):
        # Basic Property table
        labels = ", ".join([(str(label.get("displayName")) + str(label.get("enabled"))) for label in labels.values()])
        basic_table_output = [{"Name": policy.get("name"), "Labels": labels, "ID": policy.get("id")}]

        md = tableToMarkdown(
            "Azure Security Center - Get Information Protection Policy - Basic Property",
            basic_table_output,
            ["Name", "Labels", "ID"],
            removeNull=True,
        )
        ec = {"AzureSecurityCenter.InformationProtectionPolicy(val.ID && val.ID === obj.ID)": basic_table_output}

        basic_table_entry = {
            "Type": entryTypes["note"],
            "Contents": policy,
            "ContentsFormat": formats["json"],
            "ReadableContentsFormat": formats["markdown"],
            "HumanReadable": md,
            "EntryContext": ec,
        }

        # Information Type table
        info_type_table_output = []
        for information_type_data in properties.get("informationTypes").values():
            keywords = ", ".join(
                [
                    (str(keyword.get("displayName")) + str(keyword.get("custom")) + str(keyword.get("canBeNumeric")))
                    for keyword in information_type_data.get("keywords", [])
                ]
            )
            info_type_table_output.append(
                {
                    "DisplayName": information_type_data.get("displayname"),
                    "Enabled": information_type_data("enabled"),
                    "Custom": information_type_data("custom"),
                    "Keywords": keywords,
                    "RecommendedLabelID": information_type_data("recommendedLabelId"),
                }
            )
        md = tableToMarkdown(
            "Azure Security Center - Get Information Protection Policy - Information Types",
            info_type_table_output,
            ["DisplayName", "Enabled", "Custom", "Keywords", "RecommendedLabelID"],
            removeNull=True,
        )
        info_type_table_entry = {
            "Type": entryTypes["note"],
            "Contents": properties.get("informationTypes"),
            "ContentsFormat": formats["json"],
            "ReadableContentsFormat": formats["markdown"],
            "HumanReadable": md,
        }
        demisto.results([basic_table_entry, info_type_table_entry])
    else:
        demisto.results(f"No properties found in {management_group}")


""" Information Protection Policies End """

""" Jit Network Access Policies Start """


def list_jit_command(client: MsClient, args: dict):
    """Lists all Just-in-time Virtual Machines

    Args:
        client:
        args (dict): usually demisto.args()
    """
    asc_location = args.get("asc_location")
    policies = client.list_jit(asc_location)["value"]
    outputs = []
    for policy in policies:
        # summarize rules in (VMName: allowPort,...) format
        if policy.get("properties") and policy.get("properties").get("virtualMachines"):
            rules_data = policy["properties"]["virtualMachines"]
            rules_summary_array = []
            for rule in rules_data:
                ID = rule.get("id")
                vm_name = ID.split("/")[-1] if isinstance(ID, str) else None
                vm_ports = [str(port.get("number")) for port in rule.get("ports")]
                rules_summary_array.append("({}: {})".format(vm_name, ", ".join(vm_ports)))
            rules = ", ".join(rules_summary_array)

            outputs.append(
                {
                    "Name": policy.get("name"),
                    "Rules": rules,
                    "Location": policy.get("location"),
                    "Kind": policy.get("kind"),
                    "ID": policy.get("id"),
                }
            )
    md = tableToMarkdown(
        "Azure Security Center - List JIT Access Policies",
        outputs,
        ["Name", "Rules", "Location", "Kind"],
        removeNull=True,
    )
    ec = {"AzureSecurityCenter.JITPolicy(val.ID && val.ID === obj.ID)": outputs}
    return md, ec, policies


# Unsupported command. issue: issues/24583
def get_jit_command(client: MsClient, args: dict):
    """Getting given Just-in-time machine

    Args:
        client:
        args (dict): usually demisto.args()
    """
    policy_name = args.get("policy_name")
    asc_location = args.get("asc_location")
    resource_group_name = args.get("resource_group_name")
    policy = client.get_jit(policy_name, asc_location, resource_group_name)

    # Property table
    property_table_output = [
        {
            "Name": policy.get("name"),
            "Kind": policy.get("kind"),
            "ProvisioningState": policy.get("properties").get("provisioningState")
            if policy.get("properties") and policy.get("properties", {}).get("provisioningState")
            else None,
            "Location": policy.get("location"),
            "Rules": policy.get("properties").get("virtualMachines")
            if policy.get("properties") and policy.get("properties", {}).get("virtualMachines")
            else None,
            "Requests": policy.get("properties").get("requests")
            if policy.get("properties") and policy.get("properties", {}).get("requests")
            else None,
            "ID": policy.get("id"),
        }
    ]
    md = tableToMarkdown(
        "Azure Security Center - Get JIT Access Policy - Properties",
        property_table_output,
        ["Name", "Kind", "ProvisioningState", "Location", "ID"],
        removeNull=True,
    )

    ec = {"AzureSecurityCenter.JITPolicy(val.ID && val.ID === obj.ID)": property_table_output}

    property_table_entry = {
        "Type": entryTypes["note"],
        "Contents": policy,
        "ContentsFormat": formats["json"],
        "ReadableContentsFormat": formats["markdown"],
        "HumanReadable": md,
        "EntryContext": ec,
    }

    # Rules table
    rules_table_output = []
    properties = policy.get("properties")
    virtual_machines = properties.get("virtualMachines")
    if isinstance(properties, dict) and virtual_machines:
        for rule in virtual_machines:
            rules_table_output.append(
                {
                    "VmID": rule.get("id"),
                    "Ports": format_jit_port_rule(rule.get("ports")),
                }
            )
        md = tableToMarkdown(
            "Azure Security Center - Get JIT Access Policy - Rules",
            rules_table_output,
            ["VmID", "Ports"],
            removeNull=True,
        )
        rules_table_entry = {
            "Type": entryTypes["note"],
            "Contents": properties.get("virtualMachines"),
            "ContentsFormat": formats["json"],
            "ReadableContentsFormat": formats["markdown"],
            "HumanReadable": md,
        }

        # Requests table
        requests_table_output = []

        for requestData in properties.get("requests", []):
            vms = []
            for vm in requestData.get("virtualMachines"):
                vm_name = vm["id"].split("/")[-1]
                vm_ports = format_jit_port_request(vm.get("ports"))
                vms.append(f"[{vm_name}: {vm_ports}]")
            requests_table_output.append(
                {
                    "VirtualMachines": ", ".join(vms),
                    "Requestor": requestData.get("requestor") if requestData.get("requestor") else "service-account",
                    "StartTimeUtc": requestData.get("startTimeUtc"),
                }
            )
        md = tableToMarkdown(
            "Azure Security Center - Get JIT Access Policy - Requests",
            requests_table_output,
            ["VirtualMachines", "Requestor", "StartTimeUtc"],
            removeNull=True,
        )

        requests_table_entry = {
            "Type": entryTypes["note"],
            "Contents": properties.get("requests"),
            "ContentsFormat": formats["json"],
            "ReadableContentsFormat": formats["markdown"],
            "HumanReadable": md,
        }
        demisto.results([property_table_entry, rules_table_entry, requests_table_entry])


# Unsupported command. issue: issues/24583
def initiate_jit_command(client: MsClient, args: dict):
    resource_group_name = args.get("resource_group_name")
    asc_location = args.get("asc_location")
    policy_name = args.get("policy_name")
    vm_id = args.get("vmID")
    port = args.get("port")
    source_address = args.get("source_address")
    duration = args.get("duration")
    response = client.initiate_jit(
        resource_group_name,
        asc_location,
        policy_name,
        vm_id,
        port,
        source_address,
        duration,
    )
    policy_id = (
        f"/subscriptions/{client.subscription_id}/resourceGroups/{resource_group_name}/providers/"
        f"Microsoft.Security/locations/{asc_location}/jitNetworkAccessPolicies/{policy_name}"
    )
    virtual_machines = response.get("virtualMachines")
    if virtual_machines and len(virtual_machines) > 0:
        machine = virtual_machines[0]
        port = machine.get("ports")[0]

        outputs = {
            "VmID": machine.get("id"),
            "PortNum": port.get("number"),
            "AllowedSourceAddress": port.get("allowedSourceAddressPrefix"),
            "EndTimeUtc": port.get("endTimeUtc"),
            "Status": port.get("status"),
            "Requestor": response.get("requestor"),
            "PolicyID": policy_id,
        }

        md = tableToMarkdown(
            "Azure Security Center - Initiate JIT Access Request",
            outputs,
            [
                "VmID",
                "PortNum",
                "AllowedSourceAddress",
                "EndTimeUtc",
                "Status",
                "Requestor",
            ],
            removeNull=True,
        )

        ec = {
            "AzureSecurityCenter.JITPolicy(val.ID && val.ID ="
            f"== obj.{policy_id}).Initiate(val.endTimeUtc === obj.EndTimeUtc)": outputs
        }

        demisto.results(
            {
                "Type": entryTypes["note"],
                "Contents": response,
                "ContentsFormat": formats["json"],
                "ReadableContentsFormat": formats["markdown"],
                "HumanReadable": md,
                "EntryContext": ec,
            }
        )


# Unsupported command. issue: issues/24583
def delete_jit_command(client: MsClient, args: dict):
    """Deletes a Just-in-time machine

    Args:
        client:
        args (dict): usually demisto.args()
    """
    asc_location = args.get("asc_location")
    resource_group_name = args.get("resource_group_name")
    policy_name = args.get("policy_name")
    client.delete_jit(asc_location, resource_group_name, policy_name)

    policy_id = (
        f"/subscriptions/{client.subscription_id}/resourceGroups/{resource_group_name}/providers/"
        f"Microsoft.Security/locations/{asc_location}/jitNetworkAccessPolicies/{policy_name}"
    )

    outputs = {"ID": policy_id, "Action": "deleted"}

    ec = {"AzureSecurityCenter.JITPolicy(val.ID && val.ID === obj.ID)": outputs}
    demisto.results(
        {
            "Type": entryTypes["note"],
            "Contents": f"Policy - {policy_name} has been deleted sucessfully.",
            "ContentsFormat": formats["text"],
            "EntryContext": ec,
        }
    )


""" Jit Network Access Policies End """

""" Storage Start """


# Add this command to security center integration because ATP-related command requires storage account info
def list_sc_storage_command(client: MsClient):
    """Listing all Security Center Storages"""
    accounts = client.list_sc_storage().get("value")
    outputs = []
    for account in accounts:
        account_id_array = account.get("id", "").split("/")
        resource_group_name = account_id_array[account_id_array.index("resourceGroups") + 1]
        outputs.append(
            {
                "Name": account.get("name"),
                "ResourceGroupName": resource_group_name,
                "Location": account.get("location"),
                "ID": account.get("id"),
            }
        )
    md = tableToMarkdown(
        "Azure Security Center - List Storage Accounts",
        outputs,
        ["Name", "ResourceGroupName", "Location"],
        removeNull=True,
    )
    ec = {"AzureSecurityCenter.Storage(val.ID && val.ID === obj.ID)": outputs}

    return md, ec, accounts


""" Storage End """

""" Subscriptions Start """


def list_sc_subscriptions_command(client: MsClient):
    """Listing Subscriptions for this application"""
    subscriptions = client.list_sc_subscriptions().get("value")
    outputs = []
    for sub in subscriptions:
        outputs.append(
            {
                "Name": sub.get("displayName"),
                "State": sub.get("state"),
                "ID": sub.get("id"),
            }
        )
    md = tableToMarkdown(
        "Azure Security Center - Subscriptions",
        outputs,
        ["ID", "Name", "State"],
        removeNull=True,
    )
    ec = {"Azure.Subscription(val.ID && val.ID === obj.ID)": outputs}

    return md, ec, subscriptions


""" Subscriptions end """

""" Secure Score Start"""


def get_secure_scores_command(client: MsClient, args: dict):
    secure_score_name = args.get("secure_score_name", "ascScore")

    securescore = client.get_secure_scores(secure_score_name)

    md = tableToMarkdown("Azure Security Center - Secure Score", securescore["properties"])

    ec = {"Azure.Securescore(val.ID && val.ID === obj.ID)": securescore["properties"]}

    return md, ec, securescore


""" Secure Scores End"""


def list_resource_groups_command(client: MsClient, args: dict[str, Any]) -> CommandResults:
    """
    List all resource groups in the subscription.

    Args:
        client (KeyVaultClient):  Azure Key Vault API client.
        args (Dict[str, Any]): command arguments.

    Returns:
        CommandResults: Command results with raw response, outputs and readable outputs.

    """
    tag = args.get("tag", "")
    limit = arg_to_number(args.get("limit")) or DEFAULT_LIMIT

    raw_responses = []
    resource_groups: List[dict] = []

    next_link: bool | str = True
    while next_link and len(resource_groups) < limit:
        full_url = next_link if isinstance(next_link, str) else None
        response = client.list_resource_groups(tag=tag, limit=limit, full_url=full_url)
        value = response.get("value", [])
        next_link = response.get("nextLink", "")

        raw_responses.extend(value)
        for resource_group in value:
            resource_group_context = {
                "Name": resource_group.get("name"),
                "Location": resource_group.get("location"),
                "Tags": resource_group.get("tags"),
                "Provisioning State": resource_group.get("properties", {}).get("provisioningState"),
            }
            resource_groups.append(resource_group_context)

    raw_responses = raw_responses[:limit]
    resource_groups = resource_groups[:limit]
    readable_output = tableToMarkdown("Resource Groups List", resource_groups, removeNull=True)

    return CommandResults(
        outputs_prefix="Azure.ResourceGroupName",
        outputs_key_field="id",
        outputs=raw_responses,
        raw_response=raw_responses,
        readable_output=readable_output,
    )


def test_module(client: MsClient):
    """
    Performs basic GET request to check if the API is reachable and authentication is successful.
    Returns ok if successful.
    """
    if client.subscription_id:
        client.list_locations()
    else:
        client.list_sc_subscriptions()
    demisto.results("ok")


def main():
    params: dict = demisto.params()
    args = demisto.args()
    command = demisto.command()

    try:
        server = params.get("server_url", "").rstrip("/") + "/"
        tenant = params.get("credentials_tenant_id", {}).get("password") or params.get("tenant_id")
        auth_and_token_url = params.get("credentials_auth_id", {}).get("password") or params.get("auth_id", "")
        if not auth_and_token_url:
            raise DemistoException("ID must be provided.")
        enc_key = params.get("credentials_enc_key", {}).get("password") or params.get("enc_key")
        use_ssl = not params.get("unsecure", False)
        proxy = params.get("proxy", False)
        subscription_id = (
            args.get("subscription_id")
            or params.get("credentials_default_sub_id", {}).get("password")
            or params.get("default_sub_id")
        )
        resource_group_name = args.get("resource_group_name") or params.get("resource_group_name")
        ok_codes = (200, 201, 202, 204)
        certificate_thumbprint = params.get("credentials_certificate_thumbprint", {}).get("password") or params.get(
            "certificate_thumbprint"
        )
        private_key = params.get("private_key")
        managed_identities_client_id = get_azure_managed_identities_client_id(params)
        self_deployed: bool = params.get("self_deployed", False) or managed_identities_client_id is not None

        if not managed_identities_client_id:
            if not (tenant and auth_and_token_url):
                raise DemistoException(
                    "Token and ID must be provided. For further information see "
                    "https://xsoar.pan.dev/docs/reference/articles/microsoft-integrations---authentication"
                )

            if not self_deployed and not enc_key:
                raise DemistoException(
                    "Key must be provided. For further information see "
                    "https://xsoar.pan.dev/docs/reference/articles/microsoft-integrations---authentication"
                )
            elif not enc_key and not (certificate_thumbprint and private_key):
                raise DemistoException(
                    "Key or Certificate Thumbprint and Private Key must be provided."
                    "For further information see "
                    "https://xsoar.pan.dev/docs/reference/articles/microsoft-integrations---authentication"
                )

        if command in SUB_ID_REQUIRING_CMD and not subscription_id:
            raise DemistoException("A subscription ID must be provided.")
        if command in RESOURCE_GROUP_REQUIRING_CMD and not resource_group_name:
            raise DemistoException("A Resource Group Name must be provided.")
        client = MsClient(
            tenant_id=tenant,
            auth_id=auth_and_token_url,
            enc_key=enc_key,
            app_name=APP_NAME,
            proxy=proxy,
            server=server,
            verify=use_ssl,
            self_deployed=self_deployed,
            subscription_id=subscription_id,
            ok_codes=ok_codes,
            certificate_thumbprint=certificate_thumbprint,
            private_key=private_key,
            resource_group_name=resource_group_name,
            managed_identities_client_id=managed_identities_client_id,
        )

        if command == "test-module":
            # If the command will fail, error will be thrown from the request itself
            test_module(client)
        elif command == "azure-sc-get-alert":
            return_results(get_alert_command(client, args))
        elif command == "azure-sc-list-alert":
            return_outputs(*list_alerts_command(client, args))
        elif command == "azure-sc-update-alert":
            return_outputs(*update_alert_command(client, args))
        elif command == "azure-sc-list-location":
            return_outputs(*list_locations_command(client))
        elif command == "azure-sc-update-atp":
            return_outputs(*update_atp_command(client, args))
        elif command == "azure-sc-get-atp":
            return_outputs(*get_atp_command(client, args))
        elif command == "azure-sc-update-aps":
            return_outputs(*update_aps_command(client, args))
        elif command == "azure-sc-list-aps":
            return_outputs(*list_aps_command(client))
        elif command == "azure-sc-get-aps":
            return_outputs(*get_aps_command(client, args))
        elif command == "azure-sc-list-ipp":
            list_ipp_command(client, args)
        elif command == "azure-sc-get-ipp":
            get_ipp_command(client, args)
        elif command == "azure-sc-list-jit":
            return_outputs(*list_jit_command(client, args))
        elif command == "azure-sc-get-jit":
            get_jit_command(client, args)
        elif command == "azure-sc-initiate-jit":
            initiate_jit_command(client, args)
        elif command == "azure-sc-delete-jit":
            delete_jit_command(client, args)
        elif command == "azure-sc-list-storage":
            return_outputs(*list_sc_storage_command(client))
        elif command == "azure-list-subscriptions":
            return_outputs(*list_sc_subscriptions_command(client))
        elif command == "azure-get-secure-score":
            return_outputs(*get_secure_scores_command(client, args))
        elif command == "azure-resource-group-list":
            return_results(list_resource_groups_command(client, args))
        elif command == "azure-sc-auth-reset":
            return_results(reset_auth())
    except Exception as err:
        return_error(f"Failed to execute {command} command. Error: {err!s}")


if __name__ in ["__main__", "builtin", "builtins"]:
    main()