def get_encryption_status()

in VMEncryption/main/DiskUtil.py [0:0]

• 57 lines of code
• 26 McCabe index (conditional complexity)

    def get_encryption_status(self):
        encryption_status = {
            "data": "NotEncrypted",
            "os": "NotEncrypted"
        }

        mount_items = self.get_mount_items()
        device_items = self.get_device_items(None)
        device_items_dict = dict([(device_item.mount_point, device_item) for device_item in device_items])

        os_drive_encrypted = False
        data_drives_found = False
        all_data_drives_encrypted = True

        osmapper_path = os.path.join(CommonVariables.dev_mapper_root, CommonVariables.osmapper_name)

        if self.is_os_disk_lvm():
            grep_result = self.command_executor.ExecuteInBash('pvdisplay | grep {0}'.format(osmapper_path),
                                                              suppress_logging=True)
            if grep_result == 0 and not os.path.exists('/volumes.lvm'):
                self.logger.log("OS PV is encrypted")
                os_drive_encrypted = True

        special_azure_devices_to_skip = self.get_azure_devices()

        for mount_item in mount_items:
            device_item = device_items_dict.get(mount_item["dest"])

            if device_item is not None and \
               mount_item["fs"] in CommonVariables.format_supported_file_systems and \
               self.is_data_disk(device_item, special_azure_devices_to_skip):
                data_drives_found = True

                if not device_item.type == "crypt":
                    self.logger.log("Data volume {0} is mounted from {1}".format(mount_item["dest"], mount_item["src"]))
                    all_data_drives_encrypted = False

            if mount_item["dest"] == "/" and \
               not self.is_os_disk_lvm() and \
               CommonVariables.dev_mapper_root in mount_item["src"] or \
               "/dev/dm" in mount_item["src"]:
                self.logger.log("OS volume {0} is mounted from {1}".format(mount_item["dest"], mount_item["src"]))
                os_drive_encrypted = True
    
        if not data_drives_found:
            encryption_status["data"] = "NotMounted"
        elif all_data_drives_encrypted:
            encryption_status["data"] = "Encrypted"
        if os_drive_encrypted:
            encryption_status["os"] = "Encrypted"

        encryption_marker = EncryptionMarkConfig(self.logger, self.encryption_environment)
        decryption_marker = DecryptionMarkConfig(self.logger, self.encryption_environment)
        if decryption_marker.config_file_exists():
            encryption_status["data"] = "DecryptionInProgress"
        elif encryption_marker.config_file_exists():
            encryption_config = EncryptionConfig(self.encryption_environment, self.logger)
            volume_type = encryption_config.get_volume_type().lower()

            if volume_type == CommonVariables.VolumeTypeData.lower() or \
                volume_type == CommonVariables.VolumeTypeAll.lower():
                encryption_status["data"] = "EncryptionInProgress"

            if volume_type == CommonVariables.VolumeTypeOS.lower() or \
                volume_type == CommonVariables.VolumeTypeAll.lower():
                if not os_drive_encrypted:
                    encryption_status["os"] = "EncryptionInProgress"
        elif os.path.exists(osmapper_path) and not os_drive_encrypted:
            encryption_status["os"] = "VMRestartPending"

        return json.dumps(encryption_status)