def handle_gcs_config()

in AzureMonitorAgent/agent.py [0:0]

• 85 lines of code
• 39 McCabe index (conditional complexity)

def handle_gcs_config(public_settings, protected_settings, default_configs):
    """
    Populate the defaults for legacy-path GCS mode
    """
    # look for LA protected settings
    for var in list(protected_settings.keys()):
        if "_key" in var or "_id" in var:
            default_configs[var] = protected_settings.get(var)

    # check if required GCS params are available
    MONITORING_GCS_CERT_CERTFILE = None
    if "certificate" in protected_settings:
        MONITORING_GCS_CERT_CERTFILE = base64.standard_b64decode(protected_settings.get("certificate"))

    if "certificatePath" in protected_settings:
        try:
            with open(protected_settings.get("certificatePath"), 'r') as f:
                MONITORING_GCS_CERT_CERTFILE = f.read()
        except Exception as ex:
            log_and_exit('Enable', MissingorInvalidParameterErrorCode, 'Failed to read certificate {0}: {1}'.format(protected_settings.get("certificatePath"), ex))

    MONITORING_GCS_CERT_KEYFILE = None
    if "certificateKey" in protected_settings:
        MONITORING_GCS_CERT_KEYFILE = base64.standard_b64decode(protected_settings.get("certificateKey"))

    if "certificateKeyPath" in protected_settings:
        try:
            with open(protected_settings.get("certificateKeyPath"), 'r') as f:
                MONITORING_GCS_CERT_KEYFILE = f.read()
        except Exception as ex:
            log_and_exit('Enable', MissingorInvalidParameterErrorCode, 'Failed to read certificate key {0}: {1}'.format(protected_settings.get("certificateKeyPath"), ex))

    MONITORING_GCS_ENVIRONMENT = ""
    if "monitoringGCSEnvironment" in protected_settings:
        MONITORING_GCS_ENVIRONMENT = protected_settings.get("monitoringGCSEnvironment")

    MONITORING_GCS_NAMESPACE = ""
    if "namespace" in protected_settings:
        MONITORING_GCS_NAMESPACE = protected_settings.get("namespace")

    MONITORING_GCS_ACCOUNT = ""
    if "monitoringGCSAccount" in protected_settings:
        MONITORING_GCS_ACCOUNT = protected_settings.get("monitoringGCSAccount")

    MONITORING_GCS_REGION = ""
    if "monitoringGCSRegion" in protected_settings:
        MONITORING_GCS_REGION = protected_settings.get("monitoringGCSRegion")

    MONITORING_CONFIG_VERSION = ""
    if "configVersion" in protected_settings:
        MONITORING_CONFIG_VERSION = protected_settings.get("configVersion")

    MONITORING_GCS_AUTH_ID_TYPE = ""
    if "monitoringGCSAuthIdType" in protected_settings:
        MONITORING_GCS_AUTH_ID_TYPE = protected_settings.get("monitoringGCSAuthIdType")

    MONITORING_GCS_AUTH_ID = ""
    if "monitoringGCSAuthId" in protected_settings:
        MONITORING_GCS_AUTH_ID = protected_settings.get("monitoringGCSAuthId")

    MONITORING_TENANT = ""
    if "monitoringTenant" in protected_settings:
        MONITORING_TENANT = protected_settings.get("monitoringTenant")

    MONITORING_ROLE = ""
    if "monitoringRole" in protected_settings:
        MONITORING_ROLE = protected_settings.get("monitoringRole")

    MONITORING_ROLE_INSTANCE = ""
    if "monitoringRoleInstance" in protected_settings:
        MONITORING_ROLE_INSTANCE = protected_settings.get("monitoringRoleInstance")


    if ((MONITORING_GCS_CERT_CERTFILE is None or MONITORING_GCS_CERT_KEYFILE is None) and (MONITORING_GCS_AUTH_ID_TYPE == "")) or MONITORING_GCS_ENVIRONMENT == "" or MONITORING_GCS_NAMESPACE == "" or MONITORING_GCS_ACCOUNT == "" or MONITORING_GCS_REGION == "" or MONITORING_CONFIG_VERSION == "":
        log_and_exit("Enable", MissingorInvalidParameterErrorCode, 'Not all required GCS parameters are provided')
    else:
        # set the values for GCS
        default_configs["MONITORING_USE_GENEVA_CONFIG_SERVICE"] = "true"
        default_configs["MONITORING_GCS_ENVIRONMENT"] = MONITORING_GCS_ENVIRONMENT
        default_configs["MONITORING_GCS_NAMESPACE"] = MONITORING_GCS_NAMESPACE
        default_configs["MONITORING_GCS_ACCOUNT"] = MONITORING_GCS_ACCOUNT
        default_configs["MONITORING_GCS_REGION"] = MONITORING_GCS_REGION
        default_configs["MONITORING_CONFIG_VERSION"] = MONITORING_CONFIG_VERSION

        # write the certificate and key to disk
        uid = pwd.getpwnam("syslog").pw_uid
        gid = grp.getgrnam("syslog").gr_gid

        if MONITORING_GCS_AUTH_ID_TYPE != "":
            default_configs["MONITORING_GCS_AUTH_ID_TYPE"] = MONITORING_GCS_AUTH_ID_TYPE

        if MONITORING_GCS_AUTH_ID != "":
            default_configs["MONITORING_GCS_AUTH_ID"] = MONITORING_GCS_AUTH_ID

        if MONITORING_GCS_CERT_CERTFILE is not None:
            default_configs["MONITORING_GCS_CERT_CERTFILE"] = "/etc/opt/microsoft/azuremonitoragent/gcscert.pem"
            with open("/etc/opt/microsoft/azuremonitoragent/gcscert.pem", "wb") as f:
                f.write(MONITORING_GCS_CERT_CERTFILE)
            os.chown("/etc/opt/microsoft/azuremonitoragent/gcscert.pem", uid, gid)
            os.system('chmod {1} {0}'.format("/etc/opt/microsoft/azuremonitoragent/gcscert.pem", 400))

        if MONITORING_GCS_CERT_KEYFILE is not None:
            default_configs["MONITORING_GCS_CERT_KEYFILE"] = "/etc/opt/microsoft/azuremonitoragent/gcskey.pem"
            with open("/etc/opt/microsoft/azuremonitoragent/gcskey.pem", "wb") as f:
                f.write(MONITORING_GCS_CERT_KEYFILE)
            os.chown("/etc/opt/microsoft/azuremonitoragent/gcskey.pem", uid, gid)
            os.system('chmod {1} {0}'.format("/etc/opt/microsoft/azuremonitoragent/gcskey.pem", 400))

        if MONITORING_TENANT != "":
            default_configs["MONITORING_TENANT"] = MONITORING_TENANT

        if MONITORING_ROLE != "":
            default_configs["MONITORING_ROLE"] = MONITORING_ROLE

        if MONITORING_TENANT != "":
            default_configs["MONITORING_ROLE_INSTANCE"] = MONITORING_ROLE_INSTANCE