master
text 555 lines 18.1 KB
Raw
1 #!/usr/bin/env python3
2 # group: rw sudo
3 #
4 # Copyright (C) 2016 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 # Creator/Owner: Daniel P. Berrange <berrange@redhat.com>
20 #
21 # Exercise the QEMU 'luks' block driver to validate interoperability
22 # with the Linux dm-crypt + cryptsetup implementation
23
24 import subprocess
25 import os
26 import os.path
27
28 import base64
29
30 import iotests
31
32
33 class LUKSConfig(object):
34 """Represent configuration parameters for a single LUKS
35 setup to be tested"""
36
37 def __init__(self, name, cipher, keylen, mode, ivgen,
38 ivgen_hash, hash, password=None, passwords=None):
39
40 self.name = name
41 self.cipher = cipher
42 self.keylen = keylen
43 self.mode = mode
44 self.ivgen = ivgen
45 self.ivgen_hash = ivgen_hash
46 self.hash = hash
47
48 if passwords is not None:
49 self.passwords = passwords
50 else:
51 self.passwords = {}
52
53 if password is None:
54 self.passwords["0"] = "123456"
55 else:
56 self.passwords["0"] = password
57
58 def __repr__(self):
59 return self.name
60
61 def image_name(self):
62 return "luks-%s.img" % self.name
63
64 def image_path(self):
65 return os.path.join(iotests.test_dir, self.image_name())
66
67 def device_name(self):
68 return "qiotest-145-%s" % self.name
69
70 def device_path(self):
71 return "/dev/mapper/" + self.device_name()
72
73 def first_password(self):
74 for i in range(8):
75 slot = str(i)
76 if slot in self.passwords:
77 return (self.passwords[slot], slot)
78 raise Exception("No password found")
79
80 def first_password_base64(self):
81 (pw, slot) = self.first_password()
82 return base64.b64encode(pw.encode('ascii')).decode('ascii')
83
84 def active_slots(self):
85 slots = []
86 for i in range(8):
87 slot = str(i)
88 if slot in self.passwords:
89 slots.append(slot)
90 return slots
91
92 def verify_passwordless_sudo():
93 """Check whether sudo is configured to allow
94 password-less access to commands"""
95
96 args = ["sudo", "-n", "/bin/true"]
97
98 try:
99 proc = subprocess.Popen(args,
100 stdin=subprocess.PIPE,
101 stdout=subprocess.PIPE,
102 stderr=subprocess.STDOUT,
103 universal_newlines=True)
104 except FileNotFoundError as e:
105 iotests.notrun('requires sudo binary: %s' % e)
106
107 msg = proc.communicate()[0]
108
109 if proc.returncode != 0:
110 iotests.notrun('requires password-less sudo access: %s' % msg)
111
112
113 def cryptsetup(args, password=None):
114 """Run the cryptsetup command in batch mode"""
115
116 fullargs = ["sudo", "cryptsetup", "-q", "-v"]
117 fullargs.extend(args)
118
119 iotests.log(" ".join(fullargs), filters=[iotests.filter_test_dir])
120 proc = subprocess.Popen(fullargs,
121 stdin=subprocess.PIPE,
122 stdout=subprocess.PIPE,
123 stderr=subprocess.STDOUT,
124 universal_newlines=True)
125
126 msg = proc.communicate(password)[0]
127
128 if proc.returncode != 0:
129 raise Exception(msg)
130
131
132 def cryptsetup_add_password(config, slot):
133 """Add another password to a LUKS key slot"""
134
135 (password, mainslot) = config.first_password()
136
137 pwfile = os.path.join(iotests.test_dir, "passwd.txt")
138 with open(pwfile, "w") as fh:
139 fh.write(config.passwords[slot])
140
141 try:
142 args = ["luksAddKey", config.image_path(),
143 "--key-slot", slot,
144 "--key-file", "-",
145 "--iter-time", "10",
146 pwfile]
147
148 cryptsetup(args, password)
149 finally:
150 os.unlink(pwfile)
151
152
153 def cryptsetup_format(config):
154 """Format a new LUKS volume with cryptsetup, adding the
155 first key slot only"""
156
157 (password, slot) = config.first_password()
158
159 args = ["luksFormat", "--type", "luks1"]
160 cipher = config.cipher + "-" + config.mode + "-" + config.ivgen
161 if config.ivgen_hash is not None:
162 cipher = cipher + ":" + config.ivgen_hash
163 elif config.ivgen == "essiv":
164 cipher = cipher + ":" + "sha256"
165 args.extend(["--cipher", cipher])
166 if config.mode == "xts":
167 args.extend(["--key-size", str(config.keylen * 2)])
168 else:
169 args.extend(["--key-size", str(config.keylen)])
170 if config.hash is not None:
171 args.extend(["--hash", config.hash])
172 args.extend(["--key-slot", slot])
173 args.extend(["--key-file", "-"])
174 args.extend(["--iter-time", "10"])
175 args.append(config.image_path())
176
177 cryptsetup(args, password)
178
179
180 def chown(config):
181 """Set the ownership of a open LUKS device to this user"""
182
183 path = config.device_path()
184
185 args = ["sudo", "chown", "%d:%d" % (os.getuid(), os.getgid()), path]
186 iotests.log(" ".join(args), filters=[iotests.filter_chown])
187 proc = subprocess.Popen(args,
188 stdin=subprocess.PIPE,
189 stdout=subprocess.PIPE,
190 stderr=subprocess.STDOUT)
191
192 msg = proc.communicate()[0]
193
194 if proc.returncode != 0:
195 raise Exception(msg)
196
197
198 def cryptsetup_open(config):
199 """Open an image as a LUKS device"""
200
201 (password, slot) = config.first_password()
202
203 args = ["luksOpen", config.image_path(), config.device_name()]
204
205 cryptsetup(args, password)
206
207
208 def cryptsetup_close(config):
209 """Close an active LUKS device """
210
211 args = ["luksClose", config.device_name()]
212 cryptsetup(args)
213
214
215 def delete_image(config):
216 """Delete a disk image"""
217
218 try:
219 os.unlink(config.image_path())
220 iotests.log("unlink %s" % config.image_path(),
221 filters=[iotests.filter_test_dir])
222 except Exception as e:
223 pass
224
225
226 def create_image(config, size_mb):
227 """Create a bare disk image with requested size"""
228
229 delete_image(config)
230 iotests.log("truncate %s --size %dMB" % (config.image_path(), size_mb),
231 filters=[iotests.filter_test_dir])
232 with open(config.image_path(), "w") as fn:
233 fn.truncate(size_mb * 1024 * 1024)
234
235
236 def check_cipher_support(config, output):
237 """Check the output of qemu-img or qemu-io for mention of the respective
238 cipher algorithm being unsupported, and if so, skip this test.
239 (Returns `output` for convenience.)"""
240
241 if 'Unsupported cipher algorithm' in output:
242 iotests.notrun('Unsupported cipher algorithm '
243 f'{config.cipher}-{config.keylen}-{config.mode}; '
244 'consider configuring qemu with a different crypto '
245 'backend')
246 return output
247
248 def qemu_img_create(config, size_mb):
249 """Create and format a disk image with LUKS using qemu-img"""
250
251 opts = [
252 "key-secret=sec0",
253 "iter-time=10",
254 "cipher-alg=%s-%d" % (config.cipher, config.keylen),
255 "cipher-mode=%s" % config.mode,
256 "ivgen-alg=%s" % config.ivgen,
257 "hash-alg=%s" % config.hash,
258 ]
259 if config.ivgen_hash is not None:
260 opts.append("ivgen-hash-alg=%s" % config.ivgen_hash)
261
262 args = ["create", "-f", "luks",
263 "--object",
264 ("secret,id=sec0,data=%s,format=base64" %
265 config.first_password_base64()),
266 "-o", ",".join(opts),
267 config.image_path(),
268 "%dM" % size_mb]
269
270 iotests.log("qemu-img " + " ".join(args), filters=[iotests.filter_test_dir])
271 try:
272 iotests.qemu_img(*args)
273 except subprocess.CalledProcessError as exc:
274 check_cipher_support(config, exc.output)
275 raise
276
277 def qemu_io_image_args(config, dev=False):
278 """Get the args for access an image or device with qemu-io"""
279
280 if dev:
281 return [
282 "--image-opts",
283 "driver=host_device,filename=%s" % config.device_path()]
284 else:
285 return [
286 "--object",
287 ("secret,id=sec0,data=%s,format=base64" %
288 config.first_password_base64()),
289 "--image-opts",
290 ("driver=luks,key-secret=sec0,file.filename=%s" %
291 config.image_path())]
292
293 def qemu_io_write_pattern(config, pattern, offset_mb, size_mb, dev=False):
294 """Write a pattern of data to a LUKS image or device"""
295
296 if dev:
297 chown(config)
298 args = ["-c", "write -P 0x%x %dM %dM" % (pattern, offset_mb, size_mb)]
299 args.extend(qemu_io_image_args(config, dev))
300 iotests.log("qemu-io " + " ".join(args), filters=[iotests.filter_test_dir])
301 output = iotests.qemu_io(*args, check=False).stdout
302 iotests.log(check_cipher_support(config, output),
303 filters=[iotests.filter_test_dir, iotests.filter_qemu_io])
304
305
306 def qemu_io_read_pattern(config, pattern, offset_mb, size_mb, dev=False):
307 """Read a pattern of data to a LUKS image or device"""
308
309 if dev:
310 chown(config)
311 args = ["-c", "read -P 0x%x %dM %dM" % (pattern, offset_mb, size_mb)]
312 args.extend(qemu_io_image_args(config, dev))
313 iotests.log("qemu-io " + " ".join(args), filters=[iotests.filter_test_dir])
314 output = iotests.qemu_io(*args, check=False).stdout
315 iotests.log(check_cipher_support(config, output),
316 filters=[iotests.filter_test_dir, iotests.filter_qemu_io])
317
318
319 def test_once(config, qemu_img=False):
320 """Run the test with a desired LUKS configuration. Can either
321 use qemu-img for creating the initial volume, or cryptsetup,
322 in order to test interoperability in both directions"""
323
324 iotests.log("# ================= %s %s =================" % (
325 "qemu-img" if qemu_img else "dm-crypt", config))
326
327 oneKB = 1024
328 oneMB = oneKB * 1024
329 oneGB = oneMB * 1024
330 oneTB = oneGB * 1024
331
332 # 4 TB, so that we pass the 32-bit sector number boundary.
333 # Important for testing correctness of some IV generators
334 # The files are sparse, so not actually using this much space
335 image_size = 4 * oneTB
336 if qemu_img:
337 iotests.log("# Create image")
338 qemu_img_create(config, image_size // oneMB)
339 else:
340 iotests.log("# Create image")
341 create_image(config, image_size // oneMB)
342
343 lowOffsetMB = 100
344 highOffsetMB = 3 * oneTB // oneMB
345
346 try:
347 if not qemu_img:
348 iotests.log("# Format image")
349 cryptsetup_format(config)
350
351 for slot in config.active_slots()[1:]:
352 iotests.log("# Add password slot %s" % slot)
353 cryptsetup_add_password(config, slot)
354
355 # First we'll open the image using cryptsetup and write a
356 # known pattern of data that we'll then verify with QEMU
357
358 iotests.log("# Open dev")
359 cryptsetup_open(config)
360
361 try:
362 iotests.log("# Write test pattern 0xa7")
363 qemu_io_write_pattern(config, 0xa7, lowOffsetMB, 10, dev=True)
364 iotests.log("# Write test pattern 0x13")
365 qemu_io_write_pattern(config, 0x13, highOffsetMB, 10, dev=True)
366 finally:
367 iotests.log("# Close dev")
368 cryptsetup_close(config)
369
370 # Ok, now we're using QEMU to verify the pattern just
371 # written via dm-crypt
372
373 iotests.log("# Read test pattern 0xa7")
374 qemu_io_read_pattern(config, 0xa7, lowOffsetMB, 10, dev=False)
375 iotests.log("# Read test pattern 0x13")
376 qemu_io_read_pattern(config, 0x13, highOffsetMB, 10, dev=False)
377
378
379 # Write a new pattern to the image, which we'll later
380 # verify with dm-crypt
381 iotests.log("# Write test pattern 0x91")
382 qemu_io_write_pattern(config, 0x91, lowOffsetMB, 10, dev=False)
383 iotests.log("# Write test pattern 0x5e")
384 qemu_io_write_pattern(config, 0x5e, highOffsetMB, 10, dev=False)
385
386
387 # Now we're opening the image with dm-crypt once more
388 # and verifying what QEMU wrote, completing the circle
389 iotests.log("# Open dev")
390 cryptsetup_open(config)
391
392 try:
393 iotests.log("# Read test pattern 0x91")
394 qemu_io_read_pattern(config, 0x91, lowOffsetMB, 10, dev=True)
395 iotests.log("# Read test pattern 0x5e")
396 qemu_io_read_pattern(config, 0x5e, highOffsetMB, 10, dev=True)
397 finally:
398 iotests.log("# Close dev")
399 cryptsetup_close(config)
400 finally:
401 iotests.log("# Delete image")
402 delete_image(config)
403 print()
404
405
406 # Obviously we only work with the luks image format
407 iotests.script_initialize(supported_fmts=['luks'])
408
409 # We need sudo in order to run cryptsetup to create
410 # dm-crypt devices. This is safe to use on any
411 # machine, since all dm-crypt devices are backed
412 # by newly created plain files, and have a dm-crypt
413 # name prefix of 'qiotest' to avoid clashing with
414 # user LUKS volumes
415 verify_passwordless_sudo()
416
417
418 # If we look at all permutations of cipher, key size,
419 # mode, ivgen, hash, there are ~1000 possible configs.
420 #
421 # We certainly don't want/need to test every permutation
422 # to get good validation of interoperability between QEMU
423 # and dm-crypt/cryptsetup.
424 #
425 # The configs below are a representative set that aim to
426 # exercise each axis of configurability.
427 #
428 configs = [
429 # A common LUKS default
430 LUKSConfig("aes-256-xts-plain64-sha1",
431 "aes", 256, "xts", "plain64", None, "sha1"),
432
433
434 # LUKS default but diff ciphers
435 LUKSConfig("twofish-256-xts-plain64-sha1",
436 "twofish", 256, "xts", "plain64", None, "sha1"),
437 LUKSConfig("serpent-256-xts-plain64-sha1",
438 "serpent", 256, "xts", "plain64", None, "sha1"),
439 # Should really be xts, but kernel doesn't support xts+cast5
440 # nor does it do essiv+cast5
441 LUKSConfig("cast5-128-cbc-plain64-sha1",
442 "cast5", 128, "cbc", "plain64", None, "sha1"),
443 LUKSConfig("cast6-256-xts-plain64-sha1",
444 "cast6", 256, "xts", "plain64", None, "sha1"),
445
446
447 # LUKS default but diff modes / ivgens
448 LUKSConfig("aes-256-cbc-plain-sha1",
449 "aes", 256, "cbc", "plain", None, "sha1"),
450 LUKSConfig("aes-256-cbc-plain64-sha1",
451 "aes", 256, "cbc", "plain64", None, "sha1"),
452 LUKSConfig("aes-256-cbc-essiv-sha256-sha1",
453 "aes", 256, "cbc", "essiv", "sha256", "sha1"),
454 LUKSConfig("aes-256-xts-essiv-sha256-sha1",
455 "aes", 256, "xts", "essiv", "sha256", "sha1"),
456
457
458 # LUKS default but smaller key sizes
459 LUKSConfig("aes-128-xts-plain64-sha256-sha1",
460 "aes", 128, "xts", "plain64", None, "sha1"),
461 LUKSConfig("aes-192-xts-plain64-sha256-sha1",
462 "aes", 192, "xts", "plain64", None, "sha1"),
463
464 LUKSConfig("twofish-128-xts-plain64-sha1",
465 "twofish", 128, "xts", "plain64", None, "sha1"),
466 LUKSConfig("twofish-192-xts-plain64-sha1",
467 "twofish", 192, "xts", "plain64", None, "sha1"),
468
469 LUKSConfig("serpent-128-xts-plain64-sha1",
470 "serpent", 128, "xts", "plain64", None, "sha1"),
471 LUKSConfig("serpent-192-xts-plain64-sha1",
472 "serpent", 192, "xts", "plain64", None, "sha1"),
473
474 LUKSConfig("cast6-128-xts-plain64-sha1",
475 "cast6", 128, "xts", "plain", None, "sha1"),
476 LUKSConfig("cast6-192-xts-plain64-sha1",
477 "cast6", 192, "xts", "plain64", None, "sha1"),
478
479
480 # LUKS default but diff hash
481 LUKSConfig("aes-256-xts-plain64-sha224",
482 "aes", 256, "xts", "plain64", None, "sha224"),
483 LUKSConfig("aes-256-xts-plain64-sha256",
484 "aes", 256, "xts", "plain64", None, "sha256"),
485 LUKSConfig("aes-256-xts-plain64-sha384",
486 "aes", 256, "xts", "plain64", None, "sha384"),
487 LUKSConfig("aes-256-xts-plain64-sha512",
488 "aes", 256, "xts", "plain64", None, "sha512"),
489 LUKSConfig("aes-256-xts-plain64-ripemd160",
490 "aes", 256, "xts", "plain64", None, "ripemd160"),
491
492 # Password in slot 3
493 LUKSConfig("aes-256-xts-plain-sha1-pwslot3",
494 "aes", 256, "xts", "plain", None, "sha1",
495 passwords={
496 "3": "slot3",
497 }),
498
499 # Passwords in every slot
500 LUKSConfig("aes-256-xts-plain-sha1-pwallslots",
501 "aes", 256, "xts", "plain", None, "sha1",
502 passwords={
503 "0": "slot1",
504 "1": "slot1",
505 "2": "slot2",
506 "3": "slot3",
507 "4": "slot4",
508 "5": "slot5",
509 "6": "slot6",
510 "7": "slot7",
511 }),
512
513 # Check handling of default hash alg (sha256) with essiv
514 LUKSConfig("aes-256-cbc-essiv-auto-sha1",
515 "aes", 256, "cbc", "essiv", None, "sha1"),
516
517 # Check that a useless hash provided for 'plain64' iv gen
518 # is ignored and no error raised
519 LUKSConfig("aes-256-cbc-plain64-sha256-sha1",
520 "aes", 256, "cbc", "plain64", "sha256", "sha1"),
521
522 ]
523
524 unsupported_configs = [
525 # We don't have a cast-6 cipher impl for QEMU yet
526 "cast6-256-xts-plain64-sha1",
527 "cast6-128-xts-plain64-sha1",
528 "cast6-192-xts-plain64-sha1",
529
530 # GCrypt doesn't support Twofish with 192 bit key
531 "twofish-192-xts-plain64-sha1",
532 ]
533
534 # Optionally test only the configurations in the LUKS_CONFIG
535 # environment variable
536 tested_configs = None
537 if "LUKS_CONFIG" in os.environ:
538 tested_configs = os.environ["LUKS_CONFIG"].split(",")
539
540 for config in configs:
541 if config.name in unsupported_configs:
542 iotests.log("Skipping %s (config not supported)" % config.name)
543 continue
544
545 if tested_configs is not None and config.name not in tested_configs:
546 iotests.log("Skipping %s (by user request)" % config.name)
547 continue
548
549 test_once(config, qemu_img=False)
550
551 # XXX we should support setting passwords in a non-0
552 # key slot with 'qemu-img create' in future
553 (pw, slot) = config.first_password()
554 if slot == "0":
555 test_once(config, qemu_img=True)