![[BACK]](/cvsweb/icons/back.gif){kind=icon}
Return to qcow2.py CVS log ![[TXT]](/cvsweb/icons/text.gif){kind=icon}
![[DIR]](/cvsweb/icons/dir.gif){kind=icon}
Up to [Qemu by Fabrice Bellard] / qemu / tests / qemu-iotests|
Return to qcow2.py CVS log | Up to [Qemu by Fabrice Bellard] / qemu / tests / qemu-iotests |
1.1 root 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_add_header_ext(fd, magic, data):
153: try:
154: magic = int(magic, 0)
155: except:
156: print "'%s' is not a valid magic number" % magic
157: sys.exit(1)
158:
159: h = QcowHeader(fd)
160: h.extensions.append(QcowHeaderExtension.create(magic, data))
161: h.update(fd)
162:
163: def cmd_del_header_ext(fd, magic):
164: try:
165: magic = int(magic, 0)
166: except:
167: print "'%s' is not a valid magic number" % magic
168: sys.exit(1)
169:
170: h = QcowHeader(fd)
171: found = False
172:
173: for ex in h.extensions:
174: if ex.magic == magic:
175: found = True
176: h.extensions.remove(ex)
177:
178: if not found:
179: print "No such header extension"
180: return
181:
182: h.update(fd)
183:
184: cmds = [
185: [ 'dump-header', cmd_dump_header, 0, 'Dump image header and header extensions' ],
186: [ 'add-header-ext', cmd_add_header_ext, 2, 'Add a header extension' ],
187: [ 'del-header-ext', cmd_del_header_ext, 1, 'Delete a header extension' ],
188: ]
189:
190: def main(filename, cmd, args):
191: fd = open(filename, "r+b")
192: try:
193: for name, handler, num_args, desc in cmds:
194: if name != cmd:
195: continue
196: elif len(args) != num_args:
197: usage()
198: return
199: else:
200: handler(fd, *args)
201: return
202: print "Unknown command '%s'" % cmd
203: finally:
204: fd.close()
205:
206: def usage():
207: print "Usage: %s <file> <cmd> [<arg>, ...]" % sys.argv[0]
208: print ""
209: print "Supported commands:"
210: for name, handler, num_args, desc in cmds:
211: print " %-20s - %s" % (name, desc)
212:
213: if len(sys.argv) < 3:
214: usage()
215: sys.exit(1)
216:
217: main(sys.argv[1], sys.argv[2], sys.argv[3:])
This archive runs on limited infrastructure. Preserving old code on modern bandwidth. Automated agents are requested to crawl responsibly.