tests/qemu-iotests/qcow2.py

   1 #!/usr/bin/env python
   2
   3 import sys
   4 import struct
   5 import string
   6
   7 class QcowHeaderExtension:
   8
   9     def __init__(self, magic, length, data):
  10         self.magic  = magic
  11         self.length = length
  12         self.data   = data
  13
  14     @classmethod
  15     def create(cls, magic, data):
  16         return QcowHeaderExtension(magic, len(data), data)
  17
  18 class QcowHeader:
  19
  20     uint32_t = 'I'
  21     uint64_t = 'Q'
  22
  23     fields = [
  24         # Version 2 header fields
  25         [ uint32_t, '%#x',  'magic' ],
  26         [ uint32_t, '%d',   'version' ],
  27         [ uint64_t, '%#x',  'backing_file_offset' ],
  28         [ uint32_t, '%#x',  'backing_file_size' ],
  29         [ uint32_t, '%d',   'cluster_bits' ],
  30         [ uint64_t, '%d',   'size' ],
  31         [ uint32_t, '%d',   'crypt_method' ],
  32         [ uint32_t, '%d',   'l1_size' ],
  33         [ uint64_t, '%#x',  'l1_table_offset' ],
  34         [ uint64_t, '%#x',  'refcount_table_offset' ],
  35         [ uint32_t, '%d',   'refcount_table_clusters' ],
  36         [ uint32_t, '%d',   'nb_snapshots' ],
  37         [ uint64_t, '%#x',  'snapshot_offset' ],
  38
  39         # Version 3 header fields
  40         [ uint64_t, '%#x',  'incompatible_features' ],
  41         [ uint64_t, '%#x',  'compatible_features' ],
  42         [ uint64_t, '%#x',  'autoclear_features' ],
  43         [ uint32_t, '%d',   'refcount_order' ],
  44         [ uint32_t, '%d',   'header_length' ],
  45     ];
  46
  47     fmt = '>' + ''.join(field[0] for field in fields)
  48
  49     def __init__(self, fd):
  50
  51         buf_size = struct.calcsize(QcowHeader.fmt)
  52
  53         fd.seek(0)
  54         buf = fd.read(buf_size)
  55
  56         header = struct.unpack(QcowHeader.fmt, buf)
  57         self.__dict__ = dict((field[2], header[i])
  58             for i, field in enumerate(QcowHeader.fields))
  59
  60         self.set_defaults()
  61         self.cluster_size = 1 << self.cluster_bits
  62
  63         fd.seek(self.header_length)
  64         self.load_extensions(fd)
  65
  66         if self.backing_file_offset:
  67             fd.seek(self.backing_file_offset)
  68             self.backing_file = fd.read(self.backing_file_size)
  69         else:
  70             self.backing_file = None
  71
  72     def set_defaults(self):
  73         if self.version == 2:
  74             self.incompatible_features = 0
  75             self.compatible_features = 0
  76             self.autoclear_features = 0
  77             self.refcount_order = 4
  78             self.header_length = 72
  79
  80     def load_extensions(self, fd):
  81         self.extensions = []
  82
  83         if self.backing_file_offset != 0:
  84             end = min(self.cluster_size, self.backing_file_offset)
  85         else:
  86             end = self.cluster_size
  87
  88         while fd.tell() < end:
  89             (magic, length) = struct.unpack('>II', fd.read(8))
  90             if magic == 0:
  91                 break
  92             else:
  93                 padded = (length + 7) & ~7
  94                 data = fd.read(padded)
  95                 self.extensions.append(QcowHeaderExtension(magic, length, data))
  96
  97     def update_extensions(self, fd):
  98
  99         fd.seek(self.header_length)
 100         extensions = self.extensions
 101         extensions.append(QcowHeaderExtension(0, 0, ""))
 102         for ex in extensions:
 103             buf = struct.pack('>II', ex.magic, ex.length)
 104             fd.write(buf)
 105             fd.write(ex.data)
 106
 107         if self.backing_file != None:
 108             self.backing_file_offset = fd.tell()
 109             fd.write(self.backing_file)
 110
 111         if fd.tell() > self.cluster_size:
 112             raise Exception("I think I just broke the image...")
 113
 114
 115     def update(self, fd):
 116         header_bytes = self.header_length
 117
 118         self.update_extensions(fd)
 119
 120         fd.seek(0)
 121         header = tuple(self.__dict__[f] for t, p, f in QcowHeader.fields)
 122         buf = struct.pack(QcowHeader.fmt, *header)
 123         buf = buf[0:header_bytes-1]
 124         fd.write(buf)
 125
 126     def dump(self):
 127         for f in QcowHeader.fields:
 128             print "%-25s" % f[2], f[1] % self.__dict__[f[2]]
 129         print ""
 130
 131     def dump_extensions(self):
 132         for ex in self.extensions:
 133
 134             data = ex.data[:ex.length]
 135             if all(c in string.printable for c in data):
 136                 data = "'%s'" % data
 137             else:
 138                 data = "<binary>"
 139
 140             print "Header extension:"
 141             print "%-25s %#x" % ("magic", ex.magic)
 142             print "%-25s %d" % ("length", ex.length)
 143             print "%-25s %s" % ("data", data)
 144             print ""
 145
 146
 147 def cmd_dump_header(fd):
 148     h = QcowHeader(fd)
 149     h.dump()
 150     h.dump_extensions()
 151
 152 def cmd_set_header(fd, name, value):
 153     try:
 154         value = int(value, 0)
 155     except:
 156         print "'%s' is not a valid number" % value
 157         sys.exit(1)
 158
 159     fields = (field[2] for field in QcowHeader.fields)
 160     if not name in fields:
 161         print "'%s' is not a known header field" % name
 162         sys.exit(1)
 163
 164     h = QcowHeader(fd)
 165     h.__dict__[name] = value
 166     h.update(fd)
 167
 168 def cmd_add_header_ext(fd, magic, data):
 169     try:
 170         magic = int(magic, 0)
 171     except:
 172         print "'%s' is not a valid magic number" % magic
 173         sys.exit(1)
 174
 175     h = QcowHeader(fd)
 176     h.extensions.append(QcowHeaderExtension.create(magic, data))
 177     h.update(fd)
 178
 179 def cmd_del_header_ext(fd, magic):
 180     try:
 181         magic = int(magic, 0)
 182     except:
 183         print "'%s' is not a valid magic number" % magic
 184         sys.exit(1)
 185
 186     h = QcowHeader(fd)
 187     found = False
 188
 189     for ex in h.extensions:
 190         if ex.magic == magic:
 191             found = True
 192             h.extensions.remove(ex)
 193
 194     if not found:
 195         print "No such header extension"
 196         return
 197
 198     h.update(fd)
 199
 200 def cmd_set_feature_bit(fd, group, bit):
 201     try:
 202         bit = int(bit, 0)
 203         if bit < 0 or bit >= 64:
 204             raise ValueError
 205     except:
 206         print "'%s' is not a valid bit number in range [0, 64)" % bit
 207         sys.exit(1)
 208
 209     h = QcowHeader(fd)
 210     if group == 'incompatible':
 211         h.incompatible_features |= 1 << bit
 212     elif group == 'compatible':
 213         h.compatible_features |= 1 << bit
 214     elif group == 'autoclear':
 215         h.autoclear_features |= 1 << bit
 216     else:
 217         print "'%s' is not a valid group, try 'incompatible', 'compatible', or 'autoclear'" % group
 218         sys.exit(1)
 219
 220     h.update(fd)
 221
 222 cmds = [
 223     [ 'dump-header',    cmd_dump_header,    0, 'Dump image header and header extensions' ],
 224     [ 'set-header',     cmd_set_header,     2, 'Set a field in the header'],
 225     [ 'add-header-ext', cmd_add_header_ext, 2, 'Add a header extension' ],
 226     [ 'del-header-ext', cmd_del_header_ext, 1, 'Delete a header extension' ],
 227     [ 'set-feature-bit', cmd_set_feature_bit, 2, 'Set a feature bit'],
 228 ]
 229
 230 def main(filename, cmd, args):
 231     fd = open(filename, "r+b")
 232     try:
 233         for name, handler, num_args, desc in cmds:
 234             if name != cmd:
 235                 continue
 236             elif len(args) != num_args:
 237                 usage()
 238                 return
 239             else:
 240                 handler(fd, *args)
 241                 return
 242         print "Unknown command '%s'" % cmd
 243     finally:
 244         fd.close()
 245
 246 def usage():
 247     print "Usage: %s <file> <cmd> [<arg>, ...]" % sys.argv[0]
 248     print ""
 249     print "Supported commands:"
 250     for name, handler, num_args, desc in cmds:
 251         print "    %-20s - %s" % (name, desc)
 252
 253 if __name__ == '__main__':
 254     if len(sys.argv) < 3:
 255         usage()
 256         sys.exit(1)
 257
 258     main(sys.argv[1], sys.argv[2], sys.argv[3:])