storage/devicelibs/edd.py

#
# edd.py
# BIOS EDD data parsing functions
#
# Copyright (C) 2010  Red Hat, Inc.  All rights reserved.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# Author(s): Hans de Goede <hdegoede@redhat.com>
#            Ales Kozumplik <akozumpl@redhat.com>
#

import glob
import logging
import os
import re
import struct

log = logging.getLogger("storage")

re_host_bus = re.compile(r'^PCI\s*(\S*)\s*channel: (\S*)\s*$')
re_interface_scsi = re.compile(r'^SCSI\s*id: (\S*)\s*lun: (\S*)\s*$')
re_interface_ata = re.compile(r'^ATA\s*device: (\S*)\s*$')

class EddEntry(object):
    """ This object merely collects what the /sys/firmware/edd/* entries can
        provide.
    """
    def __init__(self, sysfspath):
        self.type = None

        self.ata_device = None
        self.channel = None
        self.mbr_signature = None
        self.pci_dev = None
        self.scsi_id = None
        self.scsi_lun = None
        self.sectors = None

        self.load(sysfspath)

    def __str__(self):
        return \
            "\ttype: %(type)s, ata_device: %(ata_device)s\n" \
            "\tchannel: %(channel)s, mbr_signature: %(mbr_signature)s\n" \
            "\tpci_dev: %(pci_dev)s, scsi_id: %(scsi_id)s\n" \
            "\tscsi_lun: %(scsi_lun)s, sectors: %(sectors)s" % self.__dict__

    def _read_file(self, filename):
        contents = None
        if os.path.exists(filename):
            with open(filename) as f:
                contents = f.read().rstrip()
        return contents

    def load(self, sysfspath):
        interface = self._read_file(os.path.join(sysfspath, "interface"))
        if interface:
            self.type = interface.split()[0]
            if self.type == "SCSI":
                match = re_interface_scsi.match(interface)
                self.scsi_id = int(match.group(1))
                self.scsi_lun = int(match.group(2))
            elif self.type == "ATA":
                match = re_interface_ata.match(interface)
                self.ata_device = int(match.group(1))

        self.mbr_signature = self._read_file(
            os.path.join(sysfspath, "mbr_signature"))
        sectors = self._read_file(os.path.join(sysfspath, "sectors"))
        if sectors:
            self.sectors = int(sectors)
        hbus = self._read_file(os.path.join(sysfspath, "host_bus"))
        if hbus:
            match = re_host_bus.match(hbus)
            if match:
                self.pci_dev = match.group(1)
                self.channel = int(match.group(2))
            else:
                log.warning("edd: can not match host_bus: %s" % hbus)

class EddMatcher(object):
    """ This object tries to match given entry to a disk device name.

        Assuming, heuristic analysis and guessing hapens here.
    """
    def __init__(self, edd_entry):
        self.edd = edd_entry

    def devname_from_pci_dev(self):
        name = None
        if self.edd.type == "ATA" and \
                self.edd.channel is not None and \
                self.edd.ata_device is not None:
            path = "/sys/devices/pci0000:00/0000:%(pci_dev)s/host%(chan)d/"\
                "target%(chan)d:0:%(dev)d/%(chan)d:0:%(dev)d:0/block" % {
                'pci_dev' : self.edd.pci_dev,
                'chan' : self.edd.channel,
                'dev' : self.edd.ata_device
                }
            if os.path.isdir(path):
                block_entries = os.listdir(path)
                if len(block_entries) == 1:
                    name = block_entries[0]
            else:
                log.warning("edd: directory does not exist: %s" % path)
        elif self.edd.type == "SCSI":
            pattern = "/sys/devices/pci0000:00/0000:%(pci_dev)s/virtio*/block" % \
                {'pci_dev' : self.edd.pci_dev}
            matching_paths = glob.glob(pattern)
            if len(matching_paths) != 1 or not os.path.exists(matching_paths[0]):
                return None
            block_entries = os.listdir(matching_paths[0])
            if len(block_entries) == 1:
                name = block_entries[0]
        return name

    def match_via_mbrsigs(self, mbr_dict):
        """ Try to match the edd entry based on its mbr signature.

            This will obviously fail for a fresh drive/image, but in extreme
            cases can also show false positives for randomly matching data.
        """
        for (name, mbr_signature) in mbr_dict.items():
            if mbr_signature == self.edd.mbr_signature:
                return name
        return None

def biosdev_to_edd_dir(biosdev):
    return "/sys/firmware/edd/int13_dev%x" % biosdev

def collect_edd_data():
    edd_data_dict = {}
    # the hard drive numbering starts at 0x80 (128 decimal):
    for biosdev in range(0x80, 0x80+16):
        sysfspath = biosdev_to_edd_dir(biosdev)
        if not os.path.exists(sysfspath):
            break
        edd_data_dict[biosdev] = EddEntry(sysfspath)
    return edd_data_dict

def collect_mbrs(devices):
    """ Read MBR signatures from devices.

        Returns a dict mapping device names to their MBR signatures. It is not
        guaranteed this will succeed, with a new disk for instance.
    """
    mbr_dict = {}
    for dev in devices:
        try:
            fd = os.open(dev.path, os.O_RDONLY)
            # The signature is the unsigned integer at byte 440:
            os.lseek(fd, 440, 0)
            mbrsig = struct.unpack('I', os.read(fd, 4))
            os.close(fd)
        except OSError as e:
            log.warning("edd: error reading mbrsig from disk %s: %s" %
                        (dev.name, str(e)))
            continue

        mbrsig_str = "0x%08x" % mbrsig
        # sanity check
        if mbrsig_str == '0x00000000':
            log.info("edd: MBR signature on %s is zero. new disk image?" % dev.name)
            continue
        else:
            for (dev_name, mbrsig_str_old) in mbr_dict.items():
                if mbrsig_str_old == mbrsig_str:
                    log.error("edd: dupicite MBR signature %s for %s and %s" %
                              (mbrsig_str, dev_name, dev.name))
                    # this actually makes all the other data useless
                    return {}
        # update the dictionary
        mbr_dict[dev.name] = mbrsig_str
    log.info("edd: collected mbr signatures: %s" % mbr_dict)
    return mbr_dict

def get_edd_dict(devices):
    """ Generates the 'device name' -> 'edd number' mapping.

        The EDD kernel module that exposes /sys/firmware/edd is thoroughly
        broken, the information there is incomplete and sometimes downright
        wrong. So after we mine out all useful information that the files under
        /sys/firmware/edd/int13_*/ can provide, we resort to heuristics and
        guessing. Our first attempt is, by looking at the device type int
        'interface', attempting to map pci device number, channel number etc. to
        a sysfs path, check that the path really exists, then read the device
        name (e.g 'sda') from there. Should this fail we try to match contents
        of 'mbr_signature' to a real MBR signature found on the existing block
        devices.
    """
    mbr_dict = collect_mbrs(devices)
    edd_entries_dict = collect_edd_data()
    edd_dict = {}
    for (edd_number, edd_entry) in edd_entries_dict.items():
        log.debug("edd: data extracted from 0x%x:\n%s" % (edd_number, edd_entry))
        matcher = EddMatcher(edd_entry)
        # first try to match through the pci dev etc.
        name = matcher.devname_from_pci_dev()
        # next try to compare mbr signatures
        if name:
            log.debug("edd: matched 0x%x to %s using pci_dev" % (edd_number, name))
        else:
            name = matcher.match_via_mbrsigs(mbr_dict)
            if name:
                log.info("edd: matched 0x%x to %s using MBR sig" % (edd_number, name))

        if name:
            old_edd_number = edd_dict.get(name)
            if old_edd_number:
                log.info("edd: both edd entries 0x%x and 0x%x seem to map to %s" %
                          (old_edd_number, edd_number, name))
                # this means all the other data can be confused and useless
                return {}
            edd_dict[name] = edd_number
            continue
        log.error("edd: unable to match edd entry 0x%x" % edd_number)
    return edd_dict
1	#
2	# edd.py
3	# BIOS EDD data parsing functions
4	#
5	# Copyright (C) 2010 Red Hat, Inc. All rights reserved.
6	#
7	# This program is free software; you can redistribute it and/or modify
8	# it under the terms of the GNU General Public License as published by
9	# the Free Software Foundation; either version 2 of the License, or
10	# (at your option) any later version.
11	#
12	# This program is distributed in the hope that it will be useful,
13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15	# GNU General Public License for more details.
16	#
17	# You should have received a copy of the GNU General Public License
18	# along with this program. If not, see <http://www.gnu.org/licenses/>.
19	#
20	# Author(s): Hans de Goede <hdegoede@redhat.com>
21	# Ales Kozumplik <akozumpl@redhat.com>
22	#
23
24	import glob
25	import logging
26	import os
27	import re
28	import struct
29
30	log = logging.getLogger("storage")
31
32	re_host_bus = re.compile(r'^PCI\s(\S)\schannel: (\S)\s*$')
33	re_interface_scsi = re.compile(r'^SCSI\sid: (\S)\slun: (\S)\s*$')
34	re_interface_ata = re.compile(r'^ATA\sdevice: (\S)\s*$')
35
36	class EddEntry(object):
37	""" This object merely collects what the /sys/firmware/edd/* entries can
38	provide.
39	"""
40	def __init__(self, sysfspath):
41	self.type = None
42
43	self.ata_device = None
44	self.channel = None
45	self.mbr_signature = None
46	self.pci_dev = None
47	self.scsi_id = None
48	self.scsi_lun = None
49	self.sectors = None
50
51	self.load(sysfspath)
52
53	def __str__(self):
54	return \
55	"\ttype: %(type)s, ata_device: %(ata_device)s\n" \
56	"\tchannel: %(channel)s, mbr_signature: %(mbr_signature)s\n" \
57	"\tpci_dev: %(pci_dev)s, scsi_id: %(scsi_id)s\n" \
58	"\tscsi_lun: %(scsi_lun)s, sectors: %(sectors)s" % self.__dict__
59
60	def _read_file(self, filename):
61	contents = None
62	if os.path.exists(filename):
63	with open(filename) as f:
64	contents = f.read().rstrip()
65	return contents
66
67	def load(self, sysfspath):
68	interface = self._read_file(os.path.join(sysfspath, "interface"))
69	if interface:
70	self.type = interface.split()[0]
71	if self.type == "SCSI":
72	match = re_interface_scsi.match(interface)
73	self.scsi_id = int(match.group(1))
74	self.scsi_lun = int(match.group(2))
75	elif self.type == "ATA":
76	match = re_interface_ata.match(interface)
77	self.ata_device = int(match.group(1))
78
79	self.mbr_signature = self._read_file(
80	os.path.join(sysfspath, "mbr_signature"))
81	sectors = self._read_file(os.path.join(sysfspath, "sectors"))
82	if sectors:
83	self.sectors = int(sectors)
84	hbus = self._read_file(os.path.join(sysfspath, "host_bus"))
85	if hbus:
86	match = re_host_bus.match(hbus)
87	if match:
88	self.pci_dev = match.group(1)
89	self.channel = int(match.group(2))
90	else:
91	log.warning("edd: can not match host_bus: %s" % hbus)
92
93	class EddMatcher(object):
94	""" This object tries to match given entry to a disk device name.
95
96	Assuming, heuristic analysis and guessing hapens here.
97	"""
98	def __init__(self, edd_entry):
99	self.edd = edd_entry
100
101	def devname_from_pci_dev(self):
102	name = None
103	if self.edd.type == "ATA" and \
104	self.edd.channel is not None and \
105	self.edd.ata_device is not None:
106	path = "/sys/devices/pci0000:00/0000:%(pci_dev)s/host%(chan)d/"\
107	"target%(chan)d:0:%(dev)d/%(chan)d:0:%(dev)d:0/block" % {
108	'pci_dev' : self.edd.pci_dev,
109	'chan' : self.edd.channel,
110	'dev' : self.edd.ata_device
111	}
112	if os.path.isdir(path):
113	block_entries = os.listdir(path)
114	if len(block_entries) == 1:
115	name = block_entries[0]
116	else:
117	log.warning("edd: directory does not exist: %s" % path)
118	elif self.edd.type == "SCSI":
119	pattern = "/sys/devices/pci0000:00/0000:%(pci_dev)s/virtio*/block" % \
120	{'pci_dev' : self.edd.pci_dev}
121	matching_paths = glob.glob(pattern)
122	if len(matching_paths) != 1 or not os.path.exists(matching_paths[0]):
123	return None
124	block_entries = os.listdir(matching_paths[0])
125	if len(block_entries) == 1:
126	name = block_entries[0]
127	return name
128
129	def match_via_mbrsigs(self, mbr_dict):
130	""" Try to match the edd entry based on its mbr signature.
131
132	This will obviously fail for a fresh drive/image, but in extreme
133	cases can also show false positives for randomly matching data.
134	"""
135	for (name, mbr_signature) in mbr_dict.items():
136	if mbr_signature == self.edd.mbr_signature:
137	return name
138	return None
139
140	def biosdev_to_edd_dir(biosdev):
141	return "/sys/firmware/edd/int13_dev%x" % biosdev
142
143	def collect_edd_data():
144	edd_data_dict = {}
145	# the hard drive numbering starts at 0x80 (128 decimal):
146	for biosdev in range(0x80, 0x80+16):
147	sysfspath = biosdev_to_edd_dir(biosdev)
148	if not os.path.exists(sysfspath):
149	break
150	edd_data_dict[biosdev] = EddEntry(sysfspath)
151	return edd_data_dict
152
153	def collect_mbrs(devices):
154	""" Read MBR signatures from devices.
155
156	Returns a dict mapping device names to their MBR signatures. It is not
157	guaranteed this will succeed, with a new disk for instance.
158	"""
159	mbr_dict = {}
160	for dev in devices:
161	try:
162	fd = os.open(dev.path, os.O_RDONLY)
163	# The signature is the unsigned integer at byte 440:
164	os.lseek(fd, 440, 0)
165	mbrsig = struct.unpack('I', os.read(fd, 4))
166	os.close(fd)
167	except OSError as e:
168	log.warning("edd: error reading mbrsig from disk %s: %s" %
169	(dev.name, str(e)))
170	continue
171
172	mbrsig_str = "0x%08x" % mbrsig
173	# sanity check
174	if mbrsig_str == '0x00000000':
175	log.info("edd: MBR signature on %s is zero. new disk image?" % dev.name)
176	continue
177	else:
178	for (dev_name, mbrsig_str_old) in mbr_dict.items():
179	if mbrsig_str_old == mbrsig_str:
180	log.error("edd: dupicite MBR signature %s for %s and %s" %
181	(mbrsig_str, dev_name, dev.name))
182	# this actually makes all the other data useless
183	return {}
184	# update the dictionary
185	mbr_dict[dev.name] = mbrsig_str
186	log.info("edd: collected mbr signatures: %s" % mbr_dict)
187	return mbr_dict
188
189	def get_edd_dict(devices):
190	""" Generates the 'device name' -> 'edd number' mapping.
191
192	The EDD kernel module that exposes /sys/firmware/edd is thoroughly
193	broken, the information there is incomplete and sometimes downright
194	wrong. So after we mine out all useful information that the files under
195	/sys/firmware/edd/int13_*/ can provide, we resort to heuristics and
196	guessing. Our first attempt is, by looking at the device type int
197	'interface', attempting to map pci device number, channel number etc. to
198	a sysfs path, check that the path really exists, then read the device
199	name (e.g 'sda') from there. Should this fail we try to match contents
200	of 'mbr_signature' to a real MBR signature found on the existing block
201	devices.
202	"""
203	mbr_dict = collect_mbrs(devices)
204	edd_entries_dict = collect_edd_data()
205	edd_dict = {}
206	for (edd_number, edd_entry) in edd_entries_dict.items():
207	log.debug("edd: data extracted from 0x%x:\n%s" % (edd_number, edd_entry))
208	matcher = EddMatcher(edd_entry)
209	# first try to match through the pci dev etc.
210	name = matcher.devname_from_pci_dev()
211	# next try to compare mbr signatures
212	if name:
213	log.debug("edd: matched 0x%x to %s using pci_dev" % (edd_number, name))
214	else:
215	name = matcher.match_via_mbrsigs(mbr_dict)
216	if name:
217	log.info("edd: matched 0x%x to %s using MBR sig" % (edd_number, name))
218
219	if name:
220	old_edd_number = edd_dict.get(name)
221	if old_edd_number:
222	log.info("edd: both edd entries 0x%x and 0x%x seem to map to %s" %
223	(old_edd_number, edd_number, name))
224	# this means all the other data can be confused and useless
225	return {}
226	edd_dict[name] = edd_number
227	continue
228	log.error("edd: unable to match edd entry 0x%x" % edd_number)
229	return edd_dict