| 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() |