qcow2_format.py (14526B)
1 # Library for manipulations with qcow2 image 2 # 3 # Copyright (c) 2020 Virtuozzo International GmbH. 4 # Copyright (C) 2012 Red Hat, Inc. 5 # 6 # This program is free software; you can redistribute it and/or modify 7 # it under the terms of the GNU General Public License as published by 8 # the Free Software Foundation; either version 2 of the License, or 9 # (at your option) any later version. 10 # 11 # This program is distributed in the hope that it will be useful, 12 # but WITHOUT ANY WARRANTY; without even the implied warranty of 13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 14 # GNU General Public License for more details. 15 # 16 # You should have received a copy of the GNU General Public License 17 # along with this program. If not, see <http://www.gnu.org/licenses/>. 18 # 19 20 import struct 21 import string 22 import json 23 24 25 class ComplexEncoder(json.JSONEncoder): 26 def default(self, obj): 27 if hasattr(obj, 'to_json'): 28 return obj.to_json() 29 else: 30 return json.JSONEncoder.default(self, obj) 31 32 33 class Qcow2Field: 34 35 def __init__(self, value): 36 self.value = value 37 38 def __str__(self): 39 return str(self.value) 40 41 42 class Flags64(Qcow2Field): 43 44 def __str__(self): 45 bits = [] 46 for bit in range(64): 47 if self.value & (1 << bit): 48 bits.append(bit) 49 return str(bits) 50 51 52 class BitmapFlags(Qcow2Field): 53 54 flags = { 55 0x1: 'in-use', 56 0x2: 'auto' 57 } 58 59 def __str__(self): 60 bits = [] 61 for bit in range(64): 62 flag = self.value & (1 << bit) 63 if flag: 64 bits.append(self.flags.get(flag, f'bit-{bit}')) 65 return f'{self.value:#x} ({bits})' 66 67 68 class Enum(Qcow2Field): 69 70 def __str__(self): 71 return f'{self.value:#x} ({self.mapping.get(self.value, "<unknown>")})' 72 73 74 class Qcow2StructMeta(type): 75 76 # Mapping from c types to python struct format 77 ctypes = { 78 'u8': 'B', 79 'u16': 'H', 80 'u32': 'I', 81 'u64': 'Q' 82 } 83 84 def __init__(self, name, bases, attrs): 85 if 'fields' in attrs: 86 self.fmt = '>' + ''.join(self.ctypes[f[0]] for f in self.fields) 87 88 89 class Qcow2Struct(metaclass=Qcow2StructMeta): 90 91 """Qcow2Struct: base class for qcow2 data structures 92 93 Successors should define fields class variable, which is: list of tuples, 94 each of three elements: 95 - c-type (one of 'u8', 'u16', 'u32', 'u64') 96 - format (format_spec to use with .format() when dump or 'mask' to dump 97 bitmasks) 98 - field name 99 """ 100 101 def __init__(self, fd=None, offset=None, data=None): 102 """ 103 Two variants: 104 1. Specify data. fd and offset must be None. 105 2. Specify fd and offset, data must be None. offset may be omitted 106 in this case, than current position of fd is used. 107 """ 108 if data is None: 109 assert fd is not None 110 buf_size = struct.calcsize(self.fmt) 111 if offset is not None: 112 fd.seek(offset) 113 data = fd.read(buf_size) 114 else: 115 assert fd is None and offset is None 116 117 values = struct.unpack(self.fmt, data) 118 self.__dict__ = dict((field[2], values[i]) 119 for i, field in enumerate(self.fields)) 120 121 def dump(self, is_json=False): 122 if is_json: 123 print(json.dumps(self.to_json(), indent=4, cls=ComplexEncoder)) 124 return 125 126 for f in self.fields: 127 value = self.__dict__[f[2]] 128 if isinstance(f[1], str): 129 value_str = f[1].format(value) 130 else: 131 value_str = str(f[1](value)) 132 133 print('{:<25} {}'.format(f[2], value_str)) 134 135 def to_json(self): 136 return dict((f[2], self.__dict__[f[2]]) for f in self.fields) 137 138 139 class Qcow2BitmapExt(Qcow2Struct): 140 141 fields = ( 142 ('u32', '{}', 'nb_bitmaps'), 143 ('u32', '{}', 'reserved32'), 144 ('u64', '{:#x}', 'bitmap_directory_size'), 145 ('u64', '{:#x}', 'bitmap_directory_offset') 146 ) 147 148 def __init__(self, fd, cluster_size): 149 super().__init__(fd=fd) 150 tail = struct.calcsize(self.fmt) % 8 151 if tail: 152 fd.seek(8 - tail, 1) 153 position = fd.tell() 154 self.cluster_size = cluster_size 155 self.read_bitmap_directory(fd) 156 fd.seek(position) 157 158 def read_bitmap_directory(self, fd): 159 fd.seek(self.bitmap_directory_offset) 160 self.bitmap_directory = \ 161 [Qcow2BitmapDirEntry(fd, cluster_size=self.cluster_size) 162 for _ in range(self.nb_bitmaps)] 163 164 def dump(self): 165 super().dump() 166 for entry in self.bitmap_directory: 167 print() 168 entry.dump() 169 170 def to_json(self): 171 fields_dict = super().to_json() 172 fields_dict['bitmap_directory'] = self.bitmap_directory 173 return fields_dict 174 175 176 class Qcow2BitmapDirEntry(Qcow2Struct): 177 178 fields = ( 179 ('u64', '{:#x}', 'bitmap_table_offset'), 180 ('u32', '{}', 'bitmap_table_size'), 181 ('u32', BitmapFlags, 'flags'), 182 ('u8', '{}', 'type'), 183 ('u8', '{}', 'granularity_bits'), 184 ('u16', '{}', 'name_size'), 185 ('u32', '{}', 'extra_data_size') 186 ) 187 188 def __init__(self, fd, cluster_size): 189 super().__init__(fd=fd) 190 self.cluster_size = cluster_size 191 # Seek relative to the current position in the file 192 fd.seek(self.extra_data_size, 1) 193 bitmap_name = fd.read(self.name_size) 194 self.name = bitmap_name.decode('ascii') 195 # Move position to the end of the entry in the directory 196 entry_raw_size = self.bitmap_dir_entry_raw_size() 197 padding = ((entry_raw_size + 7) & ~7) - entry_raw_size 198 fd.seek(padding, 1) 199 self.bitmap_table = Qcow2BitmapTable(fd=fd, 200 offset=self.bitmap_table_offset, 201 nb_entries=self.bitmap_table_size, 202 cluster_size=self.cluster_size) 203 204 def bitmap_dir_entry_raw_size(self): 205 return struct.calcsize(self.fmt) + self.name_size + \ 206 self.extra_data_size 207 208 def dump(self): 209 print(f'{"Bitmap name":<25} {self.name}') 210 super(Qcow2BitmapDirEntry, self).dump() 211 self.bitmap_table.dump() 212 213 def to_json(self): 214 # Put the name ahead of the dict 215 return { 216 'name': self.name, 217 **super().to_json(), 218 'bitmap_table': self.bitmap_table 219 } 220 221 222 class Qcow2BitmapTableEntry(Qcow2Struct): 223 224 fields = ( 225 ('u64', '{}', 'entry'), 226 ) 227 228 BME_TABLE_ENTRY_RESERVED_MASK = 0xff000000000001fe 229 BME_TABLE_ENTRY_OFFSET_MASK = 0x00fffffffffffe00 230 BME_TABLE_ENTRY_FLAG_ALL_ONES = 1 231 232 def __init__(self, fd): 233 super().__init__(fd=fd) 234 self.reserved = self.entry & self.BME_TABLE_ENTRY_RESERVED_MASK 235 self.offset = self.entry & self.BME_TABLE_ENTRY_OFFSET_MASK 236 if self.offset: 237 if self.entry & self.BME_TABLE_ENTRY_FLAG_ALL_ONES: 238 self.type = 'invalid' 239 else: 240 self.type = 'serialized' 241 elif self.entry & self.BME_TABLE_ENTRY_FLAG_ALL_ONES: 242 self.type = 'all-ones' 243 else: 244 self.type = 'all-zeroes' 245 246 def to_json(self): 247 return {'type': self.type, 'offset': self.offset, 248 'reserved': self.reserved} 249 250 251 class Qcow2BitmapTable: 252 253 def __init__(self, fd, offset, nb_entries, cluster_size): 254 self.cluster_size = cluster_size 255 position = fd.tell() 256 fd.seek(offset) 257 self.entries = [Qcow2BitmapTableEntry(fd) for _ in range(nb_entries)] 258 fd.seek(position) 259 260 def dump(self): 261 bitmap_table = enumerate(self.entries) 262 print(f'{"Bitmap table":<14} {"type":<15} {"size":<12} {"offset"}') 263 for i, entry in bitmap_table: 264 if entry.type == 'serialized': 265 size = self.cluster_size 266 else: 267 size = 0 268 print(f'{i:<14} {entry.type:<15} {size:<12} {entry.offset}') 269 270 def to_json(self): 271 return self.entries 272 273 274 QCOW2_EXT_MAGIC_BITMAPS = 0x23852875 275 276 277 class QcowHeaderExtension(Qcow2Struct): 278 279 class Magic(Enum): 280 mapping = { 281 0xe2792aca: 'Backing format', 282 0x6803f857: 'Feature table', 283 0x0537be77: 'Crypto header', 284 QCOW2_EXT_MAGIC_BITMAPS: 'Bitmaps', 285 0x44415441: 'Data file' 286 } 287 288 def to_json(self): 289 return self.mapping.get(self.value, "<unknown>") 290 291 fields = ( 292 ('u32', Magic, 'magic'), 293 ('u32', '{}', 'length') 294 # length bytes of data follows 295 # then padding to next multiply of 8 296 ) 297 298 def __init__(self, magic=None, length=None, data=None, fd=None, 299 cluster_size=None): 300 """ 301 Support both loading from fd and creation from user data. 302 For fd-based creation current position in a file will be used to read 303 the data. 304 The cluster_size value may be obtained by dependent structures. 305 306 This should be somehow refactored and functionality should be moved to 307 superclass (to allow creation of any qcow2 struct), but then, fields 308 of variable length (data here) should be supported in base class 309 somehow. Note also, that we probably want to parse different 310 extensions. Should they be subclasses of this class, or how to do it 311 better? Should it be something like QAPI union with discriminator field 312 (magic here). So, it's a TODO. We'll see how to properly refactor this 313 when we have more qcow2 structures. 314 """ 315 if fd is None: 316 assert all(v is not None for v in (magic, length, data)) 317 self.magic = magic 318 self.length = length 319 if length % 8 != 0: 320 padding = 8 - (length % 8) 321 data += b'\0' * padding 322 self.data = data 323 else: 324 assert all(v is None for v in (magic, length, data)) 325 super().__init__(fd=fd) 326 if self.magic == QCOW2_EXT_MAGIC_BITMAPS: 327 self.obj = Qcow2BitmapExt(fd=fd, cluster_size=cluster_size) 328 self.data = None 329 else: 330 padded = (self.length + 7) & ~7 331 self.data = fd.read(padded) 332 assert self.data is not None 333 self.obj = None 334 335 if self.data is not None: 336 data_str = self.data[:self.length] 337 if all(c in string.printable.encode( 338 'ascii') for c in data_str): 339 data_str = f"'{ data_str.decode('ascii') }'" 340 else: 341 data_str = '<binary>' 342 self.data_str = data_str 343 344 345 def dump(self): 346 super().dump() 347 348 if self.obj is None: 349 print(f'{"data":<25} {self.data_str}') 350 else: 351 self.obj.dump() 352 353 def to_json(self): 354 # Put the name ahead of the dict 355 res = {'name': self.Magic(self.magic), **super().to_json()} 356 if self.obj is not None: 357 res['data'] = self.obj 358 else: 359 res['data_str'] = self.data_str 360 361 return res 362 363 @classmethod 364 def create(cls, magic, data): 365 return QcowHeaderExtension(magic, len(data), data) 366 367 368 class QcowHeader(Qcow2Struct): 369 370 fields = ( 371 # Version 2 header fields 372 ('u32', '{:#x}', 'magic'), 373 ('u32', '{}', 'version'), 374 ('u64', '{:#x}', 'backing_file_offset'), 375 ('u32', '{:#x}', 'backing_file_size'), 376 ('u32', '{}', 'cluster_bits'), 377 ('u64', '{}', 'size'), 378 ('u32', '{}', 'crypt_method'), 379 ('u32', '{}', 'l1_size'), 380 ('u64', '{:#x}', 'l1_table_offset'), 381 ('u64', '{:#x}', 'refcount_table_offset'), 382 ('u32', '{}', 'refcount_table_clusters'), 383 ('u32', '{}', 'nb_snapshots'), 384 ('u64', '{:#x}', 'snapshot_offset'), 385 386 # Version 3 header fields 387 ('u64', Flags64, 'incompatible_features'), 388 ('u64', Flags64, 'compatible_features'), 389 ('u64', Flags64, 'autoclear_features'), 390 ('u32', '{}', 'refcount_order'), 391 ('u32', '{}', 'header_length'), 392 ) 393 394 def __init__(self, fd): 395 super().__init__(fd=fd, offset=0) 396 397 self.set_defaults() 398 self.cluster_size = 1 << self.cluster_bits 399 400 fd.seek(self.header_length) 401 self.load_extensions(fd) 402 403 if self.backing_file_offset: 404 fd.seek(self.backing_file_offset) 405 self.backing_file = fd.read(self.backing_file_size) 406 else: 407 self.backing_file = None 408 409 def set_defaults(self): 410 if self.version == 2: 411 self.incompatible_features = 0 412 self.compatible_features = 0 413 self.autoclear_features = 0 414 self.refcount_order = 4 415 self.header_length = 72 416 417 def load_extensions(self, fd): 418 self.extensions = [] 419 420 if self.backing_file_offset != 0: 421 end = min(self.cluster_size, self.backing_file_offset) 422 else: 423 end = self.cluster_size 424 425 while fd.tell() < end: 426 ext = QcowHeaderExtension(fd=fd, cluster_size=self.cluster_size) 427 if ext.magic == 0: 428 break 429 else: 430 self.extensions.append(ext) 431 432 def update_extensions(self, fd): 433 434 fd.seek(self.header_length) 435 extensions = self.extensions 436 extensions.append(QcowHeaderExtension(0, 0, b'')) 437 for ex in extensions: 438 buf = struct.pack('>II', ex.magic, ex.length) 439 fd.write(buf) 440 fd.write(ex.data) 441 442 if self.backing_file is not None: 443 self.backing_file_offset = fd.tell() 444 fd.write(self.backing_file) 445 446 if fd.tell() > self.cluster_size: 447 raise Exception('I think I just broke the image...') 448 449 def update(self, fd): 450 header_bytes = self.header_length 451 452 self.update_extensions(fd) 453 454 fd.seek(0) 455 header = tuple(self.__dict__[f] for t, p, f in QcowHeader.fields) 456 buf = struct.pack(QcowHeader.fmt, *header) 457 buf = buf[0:header_bytes-1] 458 fd.write(buf) 459 460 def dump_extensions(self, is_json=False): 461 if is_json: 462 print(json.dumps(self.extensions, indent=4, cls=ComplexEncoder)) 463 return 464 465 for ex in self.extensions: 466 print('Header extension:') 467 ex.dump() 468 print()