python3-os-win 3.0.0-0ubuntu1 / usr / lib / python3 / dist-packages / os_win / utils / storage / diskutils.py

# Copyright 2016 Cloudbase Solutions Srl
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import collections
import ctypes
import os
import re
import threading

from oslo_log import log as logging

from os_win._i18n import _
from os_win import _utils
from os_win import constants
from os_win import exceptions
from os_win.utils import baseutils
from os_win.utils import win32utils
from os_win.utils.winapi import libs as w_lib

kernel32 = w_lib.get_shared_lib_handle(w_lib.KERNEL32)

LOG = logging.getLogger(__name__)


class DEVICE_ID_VPD_PAGE(ctypes.BigEndianStructure):
    _fields_ = [
        ('DeviceType', ctypes.c_ubyte, 5),
        ('Qualifier', ctypes.c_ubyte, 3),
        ('PageCode', ctypes.c_ubyte),
        ('PageLength', ctypes.c_uint16)
    ]


class IDENTIFICATION_DESCRIPTOR(ctypes.Structure):
    _fields_ = [
        ('CodeSet', ctypes.c_ubyte, 4),
        ('ProtocolIdentifier', ctypes.c_ubyte, 4),
        ('IdentifierType', ctypes.c_ubyte, 4),
        ('Association', ctypes.c_ubyte, 2),
        ('_reserved', ctypes.c_ubyte, 1),
        ('Piv', ctypes.c_ubyte, 1),
        ('_reserved', ctypes.c_ubyte),
        ('IdentifierLength', ctypes.c_ubyte)
    ]


PDEVICE_ID_VPD_PAGE = ctypes.POINTER(DEVICE_ID_VPD_PAGE)
PIDENTIFICATION_DESCRIPTOR = ctypes.POINTER(IDENTIFICATION_DESCRIPTOR)

SCSI_ID_ASSOC_TYPE_DEVICE = 0
SCSI_ID_CODE_SET_BINARY = 1
SCSI_ID_CODE_SET_ASCII = 2

_RESCAN_LOCK = threading.Lock()


class DiskUtils(baseutils.BaseUtils):

    _wmi_cimv2_namespace = 'root/cimv2'
    _wmi_storage_namespace = 'root/microsoft/windows/storage'

    def __init__(self):
        self._conn_cimv2 = self._get_wmi_conn(self._wmi_cimv2_namespace)
        self._conn_storage = self._get_wmi_conn(self._wmi_storage_namespace)
        self._win32_utils = win32utils.Win32Utils()

        # Physical device names look like \\.\PHYSICALDRIVE1
        self._phys_dev_name_regex = re.compile(r'\\\\.*\\[a-zA-Z]*([\d]+)')

    def _get_disk_by_number(self, disk_number, msft_disk_cls=True):
        if msft_disk_cls:
            disk = self._conn_storage.Msft_Disk(Number=disk_number)
        else:
            disk = self._conn_cimv2.Win32_DiskDrive(Index=disk_number)

        if not disk:
            err_msg = _("Could not find the disk number %s")
            raise exceptions.DiskNotFound(err_msg % disk_number)
        return disk[0]

    def _get_disks_by_unique_id(self, unique_id, unique_id_format):
        # In some cases, multiple disks having the same unique id may be
        # exposed to the OS. This may happen if there are multiple paths
        # to the LUN and MPIO is not properly configured. This can be
        # valuable information to the caller.
        disks = self._conn_storage.Msft_Disk(UniqueId=unique_id,
                                             UniqueIdFormat=unique_id_format)
        if not disks:
            err_msg = _("Could not find any disk having unique id "
                        "'%(unique_id)s' and unique id format "
                        "'%(unique_id_format)s'")
            raise exceptions.DiskNotFound(err_msg % dict(
                unique_id=unique_id,
                unique_id_format=unique_id_format))
        return disks

    def get_disk_numbers_by_unique_id(self, unique_id, unique_id_format):
        disks = self._get_disks_by_unique_id(unique_id, unique_id_format)
        return [disk.Number for disk in disks]

    def get_disk_uid_and_uid_type(self, disk_number):
        disk = self._get_disk_by_number(disk_number)
        return disk.UniqueId, disk.UniqueIdFormat

    def is_mpio_disk(self, disk_number):
        disk = self._get_disk_by_number(disk_number)
        return disk.Path.lower().startswith(r'\\?\mpio')

    def refresh_disk(self, disk_number):
        disk = self._get_disk_by_number(disk_number)
        disk.Refresh()

    def get_device_name_by_device_number(self, device_number):
        disk = self._get_disk_by_number(device_number,
                                        msft_disk_cls=False)
        return disk.Name

    def get_device_number_from_device_name(self, device_name):
        matches = self._phys_dev_name_regex.findall(device_name)
        if matches:
            return matches[0]

        err_msg = _("Could not find device number for device: %s")
        raise exceptions.DiskNotFound(err_msg % device_name)

    def rescan_disks(self, merge_requests=False):
        """Perform a disk rescan.

        :param merge_requests: If this flag is set and a disk rescan is
                               already pending, we'll just wait for it to
                               finish without issuing a new rescan request.
        """
        if merge_requests:
            rescan_pending = _RESCAN_LOCK.locked()
            if rescan_pending:
                LOG.debug("A disk rescan is already pending. "
                          "Waiting for it to complete.")

            with _RESCAN_LOCK:
                if not rescan_pending:
                    self._rescan_disks()
        else:
            self._rescan_disks()

    @_utils.retry_decorator(exceptions=(exceptions.x_wmi,
                                        exceptions.OSWinException))
    def _rescan_disks(self):
        LOG.debug("Rescanning disks.")

        ret = self._conn_storage.Msft_StorageSetting.UpdateHostStorageCache()

        if isinstance(ret, collections.Iterable):
            ret = ret[0]

        if ret:
            err_msg = _("Rescanning disks failed. Error code: %s.")
            raise exceptions.OSWinException(err_msg % ret)

        LOG.debug("Finished rescanning disks.")

    def get_disk_capacity(self, path, ignore_errors=False):
        """Returns total/free space for a given directory."""
        norm_path = os.path.abspath(path)

        total_bytes = ctypes.c_ulonglong(0)
        free_bytes = ctypes.c_ulonglong(0)

        try:
            self._win32_utils.run_and_check_output(
                kernel32.GetDiskFreeSpaceExW,
                ctypes.c_wchar_p(norm_path),
                None,
                ctypes.pointer(total_bytes),
                ctypes.pointer(free_bytes),
                kernel32_lib_func=True)
            return total_bytes.value, free_bytes.value
        except exceptions.Win32Exception as exc:
            LOG.error("Could not get disk %(path)s capacity info. "
                      "Exception: %(exc)s",
                      dict(path=path,
                           exc=exc))
            if ignore_errors:
                return 0, 0
            else:
                raise exc

    def get_disk_size(self, disk_number):
        """Returns the disk size, given a physical disk number."""
        disk = self._get_disk_by_number(disk_number)
        return disk.Size

    def _parse_scsi_page_83(self, buff,
                            select_supported_identifiers=False):
        """Parse SCSI Device Identification VPD (page 0x83 data).

        :param buff: a byte array containing the SCSI page 0x83 data.
        :param select_supported_identifiers: select identifiers supported
            by Windows, in the order of precedence.
        :returns: a list of identifiers represented as dicts, containing
                  SCSI Unique IDs.
        """
        identifiers = []

        buff_sz = len(buff)
        buff = (ctypes.c_ubyte * buff_sz)(*bytearray(buff))

        vpd_pg_struct_sz = ctypes.sizeof(DEVICE_ID_VPD_PAGE)

        if buff_sz < vpd_pg_struct_sz:
            reason = _('Invalid VPD page data.')
            raise exceptions.SCSIPageParsingError(page='0x83',
                                                  reason=reason)

        vpd_page = ctypes.cast(buff, PDEVICE_ID_VPD_PAGE).contents
        vpd_page_addr = ctypes.addressof(vpd_page)
        total_page_sz = vpd_page.PageLength + vpd_pg_struct_sz

        if vpd_page.PageCode != 0x83:
            reason = _('Unexpected page code: %s') % vpd_page.PageCode
            raise exceptions.SCSIPageParsingError(page='0x83',
                                                  reason=reason)
        if total_page_sz > buff_sz:
            reason = _('VPD page overflow.')
            raise exceptions.SCSIPageParsingError(page='0x83',
                                                  reason=reason)
        if not vpd_page.PageLength:
            LOG.info('Page 0x83 data does not contain any '
                     'identification descriptors.')
            return identifiers

        id_desc_offset = vpd_pg_struct_sz
        while id_desc_offset < total_page_sz:
            id_desc_addr = vpd_page_addr + id_desc_offset
            # Remaining buffer size
            id_desc_buff_sz = buff_sz - id_desc_offset

            identifier = self._parse_scsi_id_desc(id_desc_addr,
                                                  id_desc_buff_sz)
            identifiers.append(identifier)

            id_desc_offset += identifier['raw_id_desc_size']

        if select_supported_identifiers:
            identifiers = self._select_supported_scsi_identifiers(identifiers)

        return identifiers

    def _parse_scsi_id_desc(self, id_desc_addr, buff_sz):
        """Parse SCSI VPD identification descriptor."""
        id_desc_struct_sz = ctypes.sizeof(IDENTIFICATION_DESCRIPTOR)

        if buff_sz < id_desc_struct_sz:
            reason = _('Identifier descriptor overflow.')
            raise exceptions.SCSIIdDescriptorParsingError(reason=reason)

        id_desc = IDENTIFICATION_DESCRIPTOR.from_address(id_desc_addr)
        id_desc_sz = id_desc_struct_sz + id_desc.IdentifierLength
        identifier_addr = id_desc_addr + id_desc_struct_sz

        if id_desc_sz > buff_sz:
            reason = _('Identifier overflow.')
            raise exceptions.SCSIIdDescriptorParsingError(reason=reason)

        identifier = (ctypes.c_ubyte *
                      id_desc.IdentifierLength).from_address(
                          identifier_addr)
        raw_id = bytearray(identifier)

        if id_desc.CodeSet == SCSI_ID_CODE_SET_ASCII:
            parsed_id = bytes(
                bytearray(identifier)).decode('ascii').strip('\x00')
        else:
            parsed_id = _utils.byte_array_to_hex_str(raw_id)

        id_dict = {
            'code_set': id_desc.CodeSet,
            'protocol': (id_desc.ProtocolIdentifier
                         if id_desc.Piv else None),
            'type': id_desc.IdentifierType,
            'association': id_desc.Association,
            'raw_id': raw_id,
            'id': parsed_id,
            'raw_id_desc_size': id_desc_sz,
        }
        return id_dict

    def _select_supported_scsi_identifiers(self, identifiers):
        # This method will filter out unsupported SCSI identifiers,
        # also sorting them based on the order of precedence.
        selected_identifiers = []

        for id_type in constants.SUPPORTED_SCSI_UID_FORMATS:
            for identifier in identifiers:
                if identifier['type'] == id_type:
                    selected_identifiers.append(identifier)

        return selected_identifiers

    def get_new_disk_policy(self):
        # This policy is also known as the 'SAN policy', describing
        # how new disks will be handled.
        storsetting = self._conn_storage.MSFT_StorageSetting.Get()[1]
        return storsetting.NewDiskPolicy

    def set_new_disk_policy(self, policy):
        """Sets the new disk policy, also known as SAN policy.

        :param policy: an integer value, one of the DISK_POLICY_*
                       values defined in os_win.constants.
        """
        self._conn_storage.MSFT_StorageSetting.Set(
            NewDiskPolicy=policy)

    def set_disk_online(self, disk_number):
        disk = self._get_disk_by_number(disk_number)
        err_code = disk.Online()[1]
        if err_code:
            err_msg = (_("Failed to bring disk '%(disk_number)s' online. "
                         "Error code: %(err_code)s.") %
                       dict(disk_number=disk_number,
                            err_code=err_code))
            raise exceptions.DiskUpdateError(message=err_msg)

    def set_disk_offline(self, disk_number):
        disk = self._get_disk_by_number(disk_number)
        err_code = disk.Offline()[1]
        if err_code:
            err_msg = (_("Failed to bring disk '%(disk_number)s' offline. "
                         "Error code: %(err_code)s.") %
                       dict(disk_number=disk_number,
                            err_code=err_code))
            raise exceptions.DiskUpdateError(message=err_msg)

    def set_disk_readonly_status(self, disk_number, read_only):
        disk = self._get_disk_by_number(disk_number)
        err_code = disk.SetAttributes(IsReadOnly=bool(read_only))[1]
        if err_code:
            err_msg = (_("Failed to set disk '%(disk_number)s' read-only "
                         "status to '%(read_only)s'. "
                         "Error code: %(err_code)s.") %
                       dict(disk_number=disk_number,
                            err_code=err_code,
                            read_only=bool(read_only)))
            raise exceptions.DiskUpdateError(message=err_msg)