summaryrefslogtreecommitdiff
path: root/sense_data.py
diff options
context:
space:
mode:
Diffstat (limited to 'sense_data.py')
-rw-r--r--sense_data.py267
1 files changed, 267 insertions, 0 deletions
diff --git a/sense_data.py b/sense_data.py
new file mode 100644
index 0000000..043c2db
--- /dev/null
+++ b/sense_data.py
@@ -0,0 +1,267 @@
+#!/usr/bin/env python
+
+import sys
+import os
+import re
+import ctypes
+import fcntl
+import string
+
+MEET_NEWER_SAT = 0
+wwn = ctypes.c_uint64()
+
+BSG_PROTOCOL_SCSI = 0 # <linux/bsg.h>
+BSG_SUB_PROTOCOL_SCSI_CMD = 0 # <linux/bsg.h>
+
+SG_DXFER_FROM_DEV = -3 # SCSI READ command
+
+ASCII_S = 83 # 'S'
+ASCII_Q = 81 # 'Q'
+SG_IO = 0x2285 # <scsi/sg.h>
+
+
+"""
+INQUIRY Command
+https://www.seagate.com/files/staticfiles/support/docs/manual/Interface%20manuals/100293068j.pdf
+3.6.1 Section
+"""
+
+
+class inquiry_cmd(ctypes.Structure):
+ _pack_ = 1
+ _fields_ = [
+ ("opcode", ctypes.c_ubyte),
+ ("reserved", ctypes.c_ubyte),
+ ("pagecode", ctypes.c_ubyte),
+ ("alloc_len_3", ctypes.c_ubyte),
+ ("alloc_len_4", ctypes.c_ubyte),
+ ("control", ctypes.c_ubyte)
+ ]
+
+
+"""
+ATA PASS-THROUGH (12) command
+https://www.t10.org/ftp/t10/document.04/04-262r8.pdf
+13.2.2 Section
+"""
+
+
+class ata_cmd_12(ctypes.Structure):
+ _pack_ = 1
+ _fields_ = [
+ ("opcode", ctypes.c_ubyte),
+ ("protocol", ctypes.c_ubyte),
+ ("flags", ctypes.c_ubyte),
+ ("features", ctypes.c_ubyte),
+ ("sector_count", ctypes.c_ubyte),
+ ("lba_low", ctypes.c_ubyte),
+ ("lba_mid", ctypes.c_ubyte),
+ ("lba_high", ctypes.c_ubyte),
+ ("device", ctypes.c_ubyte),
+ ("command", ctypes.c_ubyte),
+ ("reserved", ctypes.c_ubyte),
+ ("control", ctypes.c_ubyte)
+ ]
+
+
+"""
+ref: include/scsi/sg.h
+"""
+
+
+class sgio_hdr(ctypes.Structure):
+ _pack_ = 1
+ _fields_ = [
+ # [i] 'S' for SCSI generic (required)
+ ("interface_id", ctypes.c_int),
+ ("dxfer_direction", ctypes.c_int), # [i] data transfer direction
+ # [i] SCSI command length ( <= 16 bytes)
+ ("cmd_len", ctypes.c_ubyte),
+ ("mx_sb_len", ctypes.c_ubyte), # [i] max length to write to sbp
+ ("iovec_count", ctypes.c_ushort), # [i] 0 implies no scatter gather
+ ("dxfer_len", ctypes.c_uint), # [i] byte count of data transfer
+ # [i], [*io] points to data transfer memory
+ ("dxferp", ctypes.c_void_p),
+ # [i], [*i] points to command to perform
+ ("cmdp", ctypes.c_void_p),
+ # [i], [*o] points to sense_buffer memory
+ ("sbp", ctypes.c_void_p),
+ # [i] MAX_UINT->no timeout (unit: millisec)
+ ("timeout", ctypes.c_uint),
+ ("flags", ctypes.c_uint), # [i] 0 -> default, see SG_FLAG...
+ # [i->o] unused internally (normally)
+ ("pack_id", ctypes.c_int),
+ ("usr_ptr", ctypes.c_void_p), # [i->o] unused internally
+ ("status", ctypes.c_ubyte), # [o] scsi status
+ ("masked_status", ctypes.c_ubyte), # [o] shifted, masked scsi status
+ # [o] messaging level data (optional)
+ ("msg_status", ctypes.c_ubyte),
+ # [o] byte count actually written to sbp
+ ("sb_len_wr", ctypes.c_ubyte),
+ ("host_status", ctypes.c_ushort), # [o] errors from host adapter
+ ("driver_status", ctypes.c_ushort), # [o] errors from software driver
+ # [o] dxfer_len - actual_transferred
+ ("resid", ctypes.c_int),
+ # [o] time taken by cmd (unit: millisec)
+ ("duration", ctypes.c_uint),
+ ("info", ctypes.c_uint) # [o] auxiliary information
+ ]
+
+
+def from_bytes(bytes_in_array, byteorder="big", signed=False):
+ if byteorder == "little":
+ little_ordered = list(bytes_in_array)
+ elif byteorder == "big":
+ little_ordered = list(reversed(bytes_in_array))
+ else:
+ raise ValueError("byteorder must be either 'little' or 'big'")
+
+ n = sum(b << i*8 for i, b in enumerate(little_ordered))
+ if signed and little_ordered and (little_ordered[-1] & 0x80):
+ n -= 1 << 8*len(little_ordered)
+
+ return n
+
+
+def disk_scsi_inquiry_command(dev, buf):
+ sense = ctypes.c_buffer(32)
+ buf_len = ctypes.sizeof(buf)
+ cdb = inquiry_cmd(opcode=0x12,
+ reserved=0,
+ pagecode=0,
+ alloc_len_3=(buf_len >> 8),
+ alloc_len_4=(buf_len & 0xff),
+ control=0)
+
+ # systemd first tries to identify the disk by version 4, but failed. We directly use version3
+ io_hdr = sgio_hdr(interface_id=ASCII_S, dxfer_direction=SG_DXFER_FROM_DEV,
+ cmd_len=ctypes.sizeof(cdb),
+ mx_sb_len=ctypes.sizeof(sense), iovec_count=0,
+ dxfer_len=buf_len,
+ dxferp=ctypes.cast(buf, ctypes.c_void_p),
+ cmdp=ctypes.addressof(cdb),
+ sbp=ctypes.cast(sense, ctypes.c_void_p), timeout=30 * 1000,
+ flags=0, pack_id=0, usr_ptr=None, status=0, masked_status=0,
+ msg_status=0, sb_len_wr=0, host_status=0, driver_status=0,
+ resid=0, duration=0, info=0)
+
+ try:
+ with open(dev, "r") as fd:
+ ret = fcntl.ioctl(fd.fileno(), SG_IO, io_hdr)
+ if io_hdr.status != 0 or io_hdr.host_status != 0 or io_hdr.driver_status != 0 or ret != 0:
+ return False
+ except OSError as err:
+ return False
+ except IOError as err:
+ return False
+
+ return True
+
+
+def disk_identify_command(dev, buf):
+ global MEET_NEWER_SAT
+ MEET_NEWER_SAT = 0
+ sense = ctypes.c_buffer(32)
+ buf_len = ctypes.sizeof(buf)
+ cdb = ata_cmd_12(opcode=0xa1, protocol=(4 << 1), flags=0x2e,
+ features=0, sector_count=1, lba_low=0, lba_mid=0, lba_high=0,
+ device=0 & 0x4F, command=0xEC, reserved=0, control=0)
+
+ # systemd first tries to identify the disk by version 4, but failed. We directly use version3
+ io_hdr = sgio_hdr(interface_id=ASCII_S, dxfer_direction=SG_DXFER_FROM_DEV,
+ cmd_len=ctypes.sizeof(cdb),
+ mx_sb_len=ctypes.sizeof(sense), iovec_count=0,
+ dxfer_len=buf_len,
+ dxferp=ctypes.cast(buf, ctypes.c_void_p),
+ cmdp=ctypes.addressof(cdb),
+ sbp=ctypes.cast(sense, ctypes.c_void_p), timeout=30 * 1000,
+ flags=0, pack_id=0, usr_ptr=None, status=0, masked_status=0,
+ msg_status=0, sb_len_wr=0, host_status=0, driver_status=0,
+ resid=0, duration=0, info=0)
+
+ try:
+ with open(dev, "r") as fd:
+ ret = fcntl.ioctl(fd.fileno(), SG_IO, io_hdr)
+ if ret != 0:
+ return False
+ except OSError as err:
+ return False
+ except IOError as err:
+ return False
+
+ if sense[0] == b'\x72' and sense[8] == b'\x09' and sense[9] == b'\x0c':
+ return True
+
+ if sense[0] == b'\x70' and sense[12] == b'\x00' and sense[13] == b'\x1d':
+ MEET_NEWER_SAT = 1
+ return True
+
+ return False
+
+
+def disk_identify(dev):
+ identify = ctypes.c_buffer(512)
+ inquiry_buf = ctypes.c_buffer(36)
+ ret = disk_scsi_inquiry_command(dev=dev, buf=inquiry_buf)
+ if not ret:
+ return False
+
+ peripheral_device_type = from_bytes(
+ bytearray(inquiry_buf[0]), byteorder="little") & 0x1f
+ if peripheral_device_type == 0x05:
+ return False
+
+ if not (peripheral_device_type == 0x00 or peripheral_device_type == 0x14):
+ return False
+
+ if not disk_identify_command(dev=dev, buf=identify):
+ return False
+
+ global wwn
+ wwn = ctypes.c_uint64()
+ identify = bytearray(identify)
+ wwn = from_bytes(
+ [identify[108 * 2], identify[108 * 2 + 1]], byteorder="little")
+ wwn = wwn << 16
+ wwn |= from_bytes(
+ [identify[109 * 2], identify[109 * 2 + 1]], byteorder="little")
+ wwn = wwn << 16
+ wwn |= from_bytes(
+ [identify[110 * 2], identify[110 * 2 + 1]], byteorder="little")
+ wwn = wwn << 16
+ wwn |= from_bytes(
+ [identify[111 * 2], identify[111 * 2 + 1]], byteorder="little")
+
+ return True
+
+
+def check_ata_disk():
+ ret = False
+
+ for filename in os.listdir("/dev/"):
+ if not re.match("sd.*[^0-9]$|sr.*", filename):
+ continue
+
+ if not disk_identify("/dev/"+filename):
+ continue
+
+ global MEET_NEWER_SAT
+ if MEET_NEWER_SAT == 0:
+ continue
+
+ for root, dirs, files in os.walk("/dev/disk/by-id/"):
+ global wwn
+ wwn_id = "wwn-0x%x" % wwn
+ if wwn_id not in files:
+ print("The wwn_id of device(%s) will change to 0x%x" %
+ ("/dev/"+filename, wwn))
+ ret = True
+
+ return ret
+
+
+if __name__ == "__main__":
+ # exit with "1" if there is at least one disk's wwn_id will change from scsi_id to ata_id
+ if check_ata_disk():
+ exit(1)
+ exit(0)
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-20 08:30:01 +0000