def handle_mcs_config()

in AzureMonitorAgent/agent.py [0:0]

• 55 lines of code
• 30 McCabe index (conditional complexity)

def handle_mcs_config(public_settings, protected_settings, default_configs):
    """
    Populate the defaults for MCS mode
    """
    default_configs["ENABLE_MCS"] = "true"
    default_configs["PA_GIG_BRIDGE_MODE"] = "true"
    # April 2022: PA_FLUENT_SOCKET_PORT setting is being deprecated in place of PA_DATA_PORT. Remove when AMA 1.17 and earlier no longer need servicing.
    default_configs["PA_FLUENT_SOCKET_PORT"] = "13005"
    # this port will be dynamic in future
    default_configs["PA_DATA_PORT"] = "13005"
    proxySet = False

    # fetch proxy settings
    if public_settings is not None and "proxy" in public_settings and "mode" in public_settings.get("proxy") and public_settings.get("proxy").get("mode") == "application":
        default_configs["MDSD_PROXY_MODE"] = "application"

        if "address" in public_settings.get("proxy"):
            default_configs["MDSD_PROXY_ADDRESS"] = public_settings.get("proxy").get("address")
        else:
            log_and_exit("Enable", MissingorInvalidParameterErrorCode, 'Parameter "address" is required in proxy public setting')

        if "auth" in public_settings.get("proxy") and public_settings.get("proxy").get("auth") == True:
            if protected_settings is not None and "proxy" in protected_settings and "username" in protected_settings.get("proxy") and "password" in protected_settings.get("proxy"):
                default_configs["MDSD_PROXY_USERNAME"] = protected_settings.get("proxy").get("username")
                default_configs["MDSD_PROXY_PASSWORD"] = protected_settings.get("proxy").get("password")
                set_proxy(default_configs["MDSD_PROXY_ADDRESS"], default_configs["MDSD_PROXY_USERNAME"], default_configs["MDSD_PROXY_PASSWORD"])
                proxySet = True
            else:
                log_and_exit("Enable", MissingorInvalidParameterErrorCode, 'Parameter "username" and "password" not in proxy protected setting')
        else:
            set_proxy(default_configs["MDSD_PROXY_ADDRESS"], "", "")
            proxySet = True
    
    # is this Arc? If so, check for proxy     
    if os.path.isfile(ArcSettingsFile):
        f = open(ArcSettingsFile, "r")
        data = f.read()

        if (data != ''):
            json_data = json.loads(data)
            BypassProxy = False
            if json_data is not None and "proxy.bypass" in json_data:
                bypass = json_data["proxy.bypass"]
                # proxy.bypass is an array
                if "AMA" in bypass:
                    BypassProxy = True
                    
            if not BypassProxy and json_data is not None and "proxy.url" in json_data:
                url = json_data["proxy.url"]
                # only non-authenticated proxy config is supported
                if url != '':
                    default_configs["MDSD_PROXY_ADDRESS"] = url
                    set_proxy(default_configs["MDSD_PROXY_ADDRESS"], "", "")
                    proxySet = True

    if not proxySet:
        unset_proxy()

    # set arc autonomous endpoints
    az_environment, _ = get_azure_environment_and_region()
    if az_environment == me_handler.ArcACloudName:
        try:
            _, mcs_endpoint = me_handler.get_arca_endpoints_from_himds()
        except Exception as ex:
            log_and_exit("Enable", MissingorInvalidParameterErrorCode, 'Failed to get Arc autonomous endpoints. {0}'.format(ex))

        default_configs["customRegionalEndpoint"] = mcs_endpoint
        default_configs["customGlobalEndpoint"] = mcs_endpoint
        default_configs["customResourceEndpoint"] = "https://monitoring.azs"

    # add managed identity settings if they were provided
    identifier_name, identifier_value, error_msg = get_managed_identity()

    if error_msg:
        log_and_exit("Enable", MissingorInvalidParameterErrorCode, 'Failed to determine managed identity settings. {0}.'.format(error_msg))

    if identifier_name and identifier_value:
        default_configs["MANAGED_IDENTITY"] = "{0}#{1}".format(identifier_name, identifier_value)