3 from __future__
import print_function
8 class QcowHeaderExtension
:
10 def __init__(self
, magic
, length
, data
):
12 padding
= 8 - (length
% 8)
13 data
+= b
"\0" * padding
20 def create(cls
, magic
, data
):
21 return QcowHeaderExtension(magic
, len(data
), data
)
29 # Version 2 header fields
30 [ uint32_t
, '%#x', 'magic' ],
31 [ uint32_t
, '%d', 'version' ],
32 [ uint64_t
, '%#x', 'backing_file_offset' ],
33 [ uint32_t
, '%#x', 'backing_file_size' ],
34 [ uint32_t
, '%d', 'cluster_bits' ],
35 [ uint64_t
, '%d', 'size' ],
36 [ uint32_t
, '%d', 'crypt_method' ],
37 [ uint32_t
, '%d', 'l1_size' ],
38 [ uint64_t
, '%#x', 'l1_table_offset' ],
39 [ uint64_t
, '%#x', 'refcount_table_offset' ],
40 [ uint32_t
, '%d', 'refcount_table_clusters' ],
41 [ uint32_t
, '%d', 'nb_snapshots' ],
42 [ uint64_t
, '%#x', 'snapshot_offset' ],
44 # Version 3 header fields
45 [ uint64_t
, 'mask', 'incompatible_features' ],
46 [ uint64_t
, 'mask', 'compatible_features' ],
47 [ uint64_t
, 'mask', 'autoclear_features' ],
48 [ uint32_t
, '%d', 'refcount_order' ],
49 [ uint32_t
, '%d', 'header_length' ],
52 fmt
= '>' + ''.join(field
[0] for field
in fields
)
54 def __init__(self
, fd
):
56 buf_size
= struct
.calcsize(QcowHeader
.fmt
)
59 buf
= fd
.read(buf_size
)
61 header
= struct
.unpack(QcowHeader
.fmt
, buf
)
62 self
.__dict
__ = dict((field
[2], header
[i
])
63 for i
, field
in enumerate(QcowHeader
.fields
))
66 self
.cluster_size
= 1 << self
.cluster_bits
68 fd
.seek(self
.header_length
)
69 self
.load_extensions(fd
)
71 if self
.backing_file_offset
:
72 fd
.seek(self
.backing_file_offset
)
73 self
.backing_file
= fd
.read(self
.backing_file_size
)
75 self
.backing_file
= None
77 def set_defaults(self
):
79 self
.incompatible_features
= 0
80 self
.compatible_features
= 0
81 self
.autoclear_features
= 0
82 self
.refcount_order
= 4
83 self
.header_length
= 72
85 def load_extensions(self
, fd
):
88 if self
.backing_file_offset
!= 0:
89 end
= min(self
.cluster_size
, self
.backing_file_offset
)
91 end
= self
.cluster_size
93 while fd
.tell() < end
:
94 (magic
, length
) = struct
.unpack('>II', fd
.read(8))
98 padded
= (length
+ 7) & ~
7
99 data
= fd
.read(padded
)
100 self
.extensions
.append(QcowHeaderExtension(magic
, length
, data
))
102 def update_extensions(self
, fd
):
104 fd
.seek(self
.header_length
)
105 extensions
= self
.extensions
106 extensions
.append(QcowHeaderExtension(0, 0, b
""))
107 for ex
in extensions
:
108 buf
= struct
.pack('>II', ex
.magic
, ex
.length
)
112 if self
.backing_file
!= None:
113 self
.backing_file_offset
= fd
.tell()
114 fd
.write(self
.backing_file
)
116 if fd
.tell() > self
.cluster_size
:
117 raise Exception("I think I just broke the image...")
120 def update(self
, fd
):
121 header_bytes
= self
.header_length
123 self
.update_extensions(fd
)
126 header
= tuple(self
.__dict
__[f
] for t
, p
, f
in QcowHeader
.fields
)
127 buf
= struct
.pack(QcowHeader
.fmt
, *header
)
128 buf
= buf
[0:header_bytes
-1]
132 for f
in QcowHeader
.fields
:
133 value
= self
.__dict
__[f
[2]]
136 for bit
in range(64):
137 if value
& (1 << bit
):
139 value_str
= str(bits
)
141 value_str
= f
[1] % value
143 print("%-25s" % f
[2], value_str
)
146 def dump_extensions(self
):
147 for ex
in self
.extensions
:
149 data
= ex
.data
[:ex
.length
]
150 if all(c
in string
.printable
.encode('ascii') for c
in data
):
151 data
= "'%s'" % data
.decode('ascii')
155 print("Header extension:")
156 print("%-25s %#x" % ("magic", ex
.magic
))
157 print("%-25s %d" % ("length", ex
.length
))
158 print("%-25s %s" % ("data", data
))
162 def cmd_dump_header(fd
):
167 def cmd_dump_header_exts(fd
):
171 def cmd_set_header(fd
, name
, value
):
173 value
= int(value
, 0)
175 print("'%s' is not a valid number" % value
)
178 fields
= (field
[2] for field
in QcowHeader
.fields
)
179 if not name
in fields
:
180 print("'%s' is not a known header field" % name
)
184 h
.__dict
__[name
] = value
187 def cmd_add_header_ext(fd
, magic
, data
):
189 magic
= int(magic
, 0)
191 print("'%s' is not a valid magic number" % magic
)
195 h
.extensions
.append(QcowHeaderExtension
.create(magic
, data
.encode('ascii')))
198 def cmd_add_header_ext_stdio(fd
, magic
):
199 data
= sys
.stdin
.read()
200 cmd_add_header_ext(fd
, magic
, data
)
202 def cmd_del_header_ext(fd
, magic
):
204 magic
= int(magic
, 0)
206 print("'%s' is not a valid magic number" % magic
)
212 for ex
in h
.extensions
:
213 if ex
.magic
== magic
:
215 h
.extensions
.remove(ex
)
218 print("No such header extension")
223 def cmd_set_feature_bit(fd
, group
, bit
):
226 if bit
< 0 or bit
>= 64:
229 print("'%s' is not a valid bit number in range [0, 64)" % bit
)
233 if group
== 'incompatible':
234 h
.incompatible_features |
= 1 << bit
235 elif group
== 'compatible':
236 h
.compatible_features |
= 1 << bit
237 elif group
== 'autoclear':
238 h
.autoclear_features |
= 1 << bit
240 print("'%s' is not a valid group, try 'incompatible', 'compatible', or 'autoclear'" % group
)
246 [ 'dump-header', cmd_dump_header
, 0, 'Dump image header and header extensions' ],
247 [ 'dump-header-exts', cmd_dump_header_exts
, 0, 'Dump image header extensions' ],
248 [ 'set-header', cmd_set_header
, 2, 'Set a field in the header'],
249 [ 'add-header-ext', cmd_add_header_ext
, 2, 'Add a header extension' ],
250 [ 'add-header-ext-stdio', cmd_add_header_ext_stdio
, 1, 'Add a header extension, data from stdin' ],
251 [ 'del-header-ext', cmd_del_header_ext
, 1, 'Delete a header extension' ],
252 [ 'set-feature-bit', cmd_set_feature_bit
, 2, 'Set a feature bit'],
255 def main(filename
, cmd
, args
):
256 fd
= open(filename
, "r+b")
258 for name
, handler
, num_args
, desc
in cmds
:
261 elif len(args
) != num_args
:
267 print("Unknown command '%s'" % cmd
)
272 print("Usage: %s <file> <cmd> [<arg>, ...]" % sys
.argv
[0])
274 print("Supported commands:")
275 for name
, handler
, num_args
, desc
in cmds
:
276 print(" %-20s - %s" % (name
, desc
))
278 if __name__
== '__main__':
279 if len(sys
.argv
) < 3:
283 main(sys
.argv
[1], sys
.argv
[2], sys
.argv
[3:])