a-c-m
/
community.crypto
mirror of https://github.com/ansible-collections/community.crypto.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

luks_device: add support for keyslots (#664) 

* luks_device: add support for keyslots

* luks_device: replace python3 format strings with python2 format strings, remove print statements

* luks_device: add missing copyright information in keyslot integration test files

* luks_device: updated failing unit tests for keyslot support

* luks_device: improve detection of luks version

* luks_device: Update documentation on keyslot parameters, minor code improvements

* luks_device: improve validation of keyslot parameters, fix tests for systems that do not support luks2

* luks_device: correct spelling and errors in documentation and output, check all possible locations for LUKS2 header
pull/673/head
Steffen Gufler 2023-10-29 10:53:00 +01:00 committed by GitHub
parent 428550165a
commit 6504e67139
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 407 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

142
plugins/modules/luks_device.py
View File

@ -79,6 +79,17 @@ options:
value is a string with the passphrase."
type: str
version_added: '1.0.0'
keyslot:
description:
- "Adds the O(keyfile) or O(passphrase) to a specific keyslot when
creating a new container on O(device). Parameter value is the
number of the keyslot."
- "B(Note) that a device of O(type=luks1) supports the keyslot numbers
V(0)-V(7) and a device of O(type=luks2) supports the keyslot numbers
V(0)-V(31). In order to use the keyslots V(8)-V(31) when creating a new
container, setting O(type) to V(luks2) is required."
type: int
version_added: '2.16.0'
keysize:
description:
- "Sets the key size only if LUKS container does not exist."
@ -108,6 +119,16 @@ options:
be used even if another keyslot already exists for this passphrase."
type: str
version_added: '1.0.0'
new_keyslot:
description:
- "Adds the additional O(new_keyfile) or O(new_passphrase) to a
specific keyslot on the given O(device). Parameter value is the number
of the keyslot."
- "B(Note) that a device of O(type=luks1) supports the keyslot numbers
V(0)-V(7) and a device of O(type=luks2) supports the keyslot numbers
V(0)-V(31)."
type: int
version_added: '2.16.0'
remove_keyfile:
description:
- "Removes given key from the container on O(device). Does not
@ -133,6 +154,17 @@ options:
to V(true)."
type: str
version_added: '1.0.0'
remove_keyslot:
description:
- "Removes the key in the given slot on O(device). Needs
O(keyfile) or O(passphrase) for authorization."
- "B(Note) that a device of O(type=luks1) supports the keyslot numbers
V(0)-V(7) and a device of O(type=luks2) supports the keyslot numbers
V(0)-V(31)."
- "B(Note) that the given O(keyfile) or O(passphrase) must not be
in the slot to be removed."
type: int
version_added: '2.16.0'
force_remove_last_key:
description:
- "If set to V(true), allows removing the last key from a container."
@ -377,6 +409,26 @@ EXAMPLES = '''
state: "present"
keyfile: "/vault/keyfile"
type: luks2
- name: Create a container with key in slot 4
community.crypto.luks_device:
device: "/dev/loop0"
state: "present"
keyfile: "/vault/keyfile"
keyslot: 4
- name: Add a new key in slot 5
community.crypto.luks_device:
device: "/dev/loop0"
keyfile: "/vault/keyfile"
new_keyfile: "/vault/keyfile"
new_keyslot: 5
- name: Remove the key from slot 4 (given keyfile must not be slot 4)
community.crypto.luks_device:
device: "/dev/loop0"
keyfile: "/vault/keyfile"
remove_keyslot: 4
'''
RETURN = '''
@ -523,6 +575,29 @@ class CryptHandler(Handler):
result = self._run_command([self._cryptsetup_bin, 'isLuks', device])
return result[RETURN_CODE] == 0
def get_luks_type(self, device):
''' get the luks type of a device
'''
if self.is_luks(device):
with open(device, 'rb') as f:
for offset in LUKS2_HEADER_OFFSETS:
f.seek(offset)
data = f.read(LUKS_HEADER_L)
if data == LUKS2_HEADER2:
return 'luks2'
return 'luks1'
return None
def is_luks_slot_set(self, device, keyslot):
''' check if a keyslot is set
'''
result = self._run_command([self._cryptsetup_bin, 'luksDump', device])
if result[RETURN_CODE] != 0:
raise ValueError('Error while dumping LUKS header from %s' % (device, ))
result_luks1 = 'Key Slot %d: ENABLED' % (keyslot) in result[STDOUT]
result_luks2 = ' %d: luks2' % (keyslot) in result[STDOUT]
return result_luks1 or result_luks2
def _add_pbkdf_options(self, options, pbkdf):
if pbkdf['iteration_time'] is not None:
options.extend(['--iter-time', str(int(pbkdf['iteration_time'] * 1000))])
@ -535,7 +610,7 @@ class CryptHandler(Handler):
if pbkdf['parallel'] is not None:
options.extend(['--pbkdf-parallel', str(pbkdf['parallel'])])
def run_luks_create(self, device, keyfile, passphrase, keysize, cipher, hash_, sector_size, pbkdf):
def run_luks_create(self, device, keyfile, passphrase, keyslot, keysize, cipher, hash_, sector_size, pbkdf):
# create a new luks container; use batch mode to auto confirm
luks_type = self._module.params['type']
label = self._module.params['label']
@ -556,6 +631,8 @@ class CryptHandler(Handler):
self._add_pbkdf_options(options, pbkdf)
if sector_size is not None:
options.extend(['--sector-size', str(sector_size)])
if keyslot is not None:
options.extend(['--key-slot', str(keyslot)])
args = [self._cryptsetup_bin, 'luksFormat']
args.extend(options)
@ -615,7 +692,7 @@ class CryptHandler(Handler):
raise ValueError('Error while wiping LUKS container signatures for %s: %s' % (device, exc))
def run_luks_add_key(self, device, keyfile, passphrase, new_keyfile,
new_passphrase, pbkdf):
new_passphrase, new_keyslot, pbkdf):
''' Add new key from a keyfile or passphrase to given 'device';
authentication done using 'keyfile' or 'passphrase'.
Raises ValueError when command fails.
@ -625,6 +702,9 @@ class CryptHandler(Handler):
if pbkdf is not None:
self._add_pbkdf_options(args, pbkdf)
if new_keyslot is not None:
args.extend(['--key-slot', str(new_keyslot)])
if keyfile:
args.extend(['--key-file', keyfile])
else:
@ -640,7 +720,7 @@ class CryptHandler(Handler):
raise ValueError('Error while adding new LUKS keyslot to %s: %s'
% (device, result[STDERR]))
def run_luks_remove_key(self, device, keyfile, passphrase,
def run_luks_remove_key(self, device, keyfile, passphrase, keyslot,
force_remove_last_key=False):
''' Remove key from given device
Raises ValueError when command fails
@ -675,7 +755,10 @@ class CryptHandler(Handler):
"To be able to remove a key, please set "
"`force_remove_last_key` to `true`." % device)
args = [self._cryptsetup_bin, 'luksRemoveKey', device, '-q']
if keyslot is None:
args = [self._cryptsetup_bin, 'luksRemoveKey', device, '-q']
else:
args = [self._cryptsetup_bin, 'luksKillSlot', device, '-q', str(keyslot)]
if keyfile:
args.extend(['--key-file', keyfile])
result = self._run_command(args, data=passphrase)
@ -683,7 +766,7 @@ class CryptHandler(Handler):
raise ValueError('Error while removing LUKS key from %s: %s'
% (device, result[STDERR]))
def luks_test_key(self, device, keyfile, passphrase):
def luks_test_key(self, device, keyfile, passphrase, keyslot=None):
''' Check whether the keyfile or passphrase works.
Raises ValueError when command fails.
'''
@ -695,12 +778,17 @@ class CryptHandler(Handler):
else:
data = passphrase
if keyslot is not None:
args.extend(['--key-slot', str(keyslot)])
result = self._run_command(args, data=data)
if result[RETURN_CODE] == 0:
return True
for output in (STDOUT, STDERR):
if 'No key available with this passphrase' in result[output]:
return False
if 'No usable keyslot is available.' in result[output]:
return False
raise ValueError('Error while testing whether keyslot exists on %s: %s'
% (device, result[STDERR]))
@ -817,7 +905,8 @@ class ConditionsHandler(Handler):
def luks_remove_key(self):
if (self.device is None or
(self._module.params['remove_keyfile'] is None and
self._module.params['remove_passphrase'] is None)):
self._module.params['remove_passphrase'] is None and
self._module.params['remove_keyslot'] is None)):
# conditions for removing a key not fulfilled
return False
@ -825,6 +914,15 @@ class ConditionsHandler(Handler):
self._module.fail_json(msg="Contradiction in setup: Asking to "
"remove a key from absent LUKS.")
if self._module.params['remove_keyslot']:
if not self._crypthandler.is_luks_slot_set(self.device, self._module.params['remove_keyslot']):
return False
result = self._crypthandler.luks_test_key(self.device, self._module.params['keyfile'], self._module.params['passphrase'])
if self._crypthandler.luks_test_key(self.device, self._module.params['keyfile'], self._module.params['passphrase'],
self._module.params['remove_keyslot']):
self._module.fail_json(msg='Cannot remove keyslot with keyfile or passphrase in same slot.')
return result
return self._crypthandler.luks_test_key(self.device, self._module.params['remove_keyfile'], self._module.params['remove_passphrase'])
def luks_remove(self):
@ -832,6 +930,19 @@ class ConditionsHandler(Handler):
self._module.params['state'] == 'absent' and
self._crypthandler.is_luks(self.device))
def validate_keyslot(self, param, luks_type):
if self._module.params[param] is not None:
if luks_type is None and param == 'keyslot':
if 8 <= self._module.params[param] <= 31:
self._module.fail_json(msg="You must specify type=luks2 when creating a new LUKS device to use keyslots 8-31.")
elif not (0 <= self._module.params[param] <= 7):
self._module.fail_json(msg="When not specifying a type, only the keyslots 0-7 are allowed.")
if luks_type == 'luks1' and not 0 <= self._module.params[param] <= 7:
self._module.fail_json(msg="%s must be between 0 and 7 when using LUKS1." % self._module.params[param])
elif luks_type == 'luks2' and not 0 <= self._module.params[param] <= 31:
self._module.fail_json(msg="%s must be between 0 and 31 when using LUKS2." % self._module.params[param])
def run_module():
# available arguments/parameters that a user can pass
@ -845,6 +956,9 @@ def run_module():
passphrase=dict(type='str', no_log=True),
new_passphrase=dict(type='str', no_log=True),
remove_passphrase=dict(type='str', no_log=True),
keyslot=dict(type='int', no_log=False),
new_keyslot=dict(type='int', no_log=False),
remove_keyslot=dict(type='int', no_log=False),
force_remove_last_key=dict(type='bool', default=False),
keysize=dict(type='int'),
label=dict(type='str'),
@ -874,7 +988,7 @@ def run_module():
mutually_exclusive = [
('keyfile', 'passphrase'),
('new_keyfile', 'new_passphrase'),
('remove_keyfile', 'remove_passphrase')
('remove_keyfile', 'remove_passphrase', 'remove_keyslot')
]
# seed the result dict in the object
@ -904,6 +1018,17 @@ def run_module():
if module.params['label'] is not None and module.params['type'] == 'luks1':
module.fail_json(msg='You cannot combine type luks1 with the label option.')
if module.params['keyslot'] is not None or module.params['new_keyslot'] is not None or module.params['remove_keyslot'] is not None:
luks_type = crypt.get_luks_type(conditions.get_device_name())
if luks_type is None and module.params['type'] is not None:
luks_type = module.params['type']
for param in ['keyslot', 'new_keyslot', 'remove_keyslot']:
conditions.validate_keyslot(param, luks_type)
for param in ['new_keyslot', 'remove_keyslot']:
if module.params[param] is not None and module.params['keyfile'] is None and module.params['passphrase'] is None:
module.fail_json(msg="Removing a keyslot requires the passphrase or keyfile of another slot.")
# The conditions are in order to allow more operations in one run.
# (e.g. create luks and add a key to it)
@ -914,6 +1039,7 @@ def run_module():
crypt.run_luks_create(conditions.device,
module.params['keyfile'],
module.params['passphrase'],
module.params['keyslot'],
module.params['keysize'],
module.params['cipher'],
module.params['hash'],
@ -986,6 +1112,7 @@ def run_module():
module.params['passphrase'],
module.params['new_keyfile'],
module.params['new_passphrase'],
module.params['new_keyslot'],
module.params['pbkdf'])
except ValueError as e:
module.fail_json(msg="luks_device error: %s" % e)
@ -1001,6 +1128,7 @@ def run_module():
crypt.run_luks_remove_key(conditions.device,
module.params['remove_keyfile'],
module.params['remove_passphrase'],
module.params['remove_keyslot'],
force_remove_last_key=last_key)
except ValueError as e:
module.fail_json(msg="luks_device error: %s" % e)

178
tests/integration/targets/luks_device/tasks/tests/keyslot-create-destroy.yml Normal file
View File

@ -0,0 +1,178 @@
---
# Copyright (c) Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
- name: Create luks with keyslot 4 (check)
luks_device:
device: "{{ cryptfile_device }}"
state: present
keyfile: "{{ remote_tmp_dir }}/keyfile1"
keyslot: 4
pbkdf:
iteration_time: 0.1
check_mode: true
become: true
register: create_luks_slot4_check
- name: Create luks with keyslot 4
luks_device:
device: "{{ cryptfile_device }}"
state: present
keyfile: "{{ remote_tmp_dir }}/keyfile1"
keyslot: 4
pbkdf:
iteration_time: 0.1
become: true
register: create_luks_slot4
- name: Create luks with keyslot 4 (idempotent)
luks_device:
device: "{{ cryptfile_device }}"
state: present
keyfile: "{{ remote_tmp_dir }}/keyfile1"
keyslot: 4
pbkdf:
iteration_time: 0.1
become: true
register: create_luks_slot4_idem
- name: Create luks with keyslot 4 (idempotent, check)
luks_device:
device: "{{ cryptfile_device }}"
state: present
keyfile: "{{ remote_tmp_dir }}/keyfile1"
keyslot: 4
pbkdf:
iteration_time: 0.1
check_mode: true
become: true
register: create_luks_slot4_idem_check
- name: Dump luks header
command: "cryptsetup luksDump {{ cryptfile_device }}"
become: true
register: luks_header_slot4
- assert:
that:
- create_luks_slot4_check is changed
- create_luks_slot4 is changed
- create_luks_slot4_idem is not changed
- create_luks_slot4_idem_check is not changed
- "'Key Slot 4: ENABLED' in luks_header_slot4.stdout or '4: luks2' in luks_header_slot4.stdout"
- name: Add key in slot 2 (check)
luks_device:
device: "{{ cryptfile_device }}"
state: present
keyfile: "{{ remote_tmp_dir }}/keyfile1"
new_keyfile: "{{ remote_tmp_dir }}/keyfile2"
new_keyslot: 2
pbkdf:
iteration_time: 0.1
check_mode: true
become: true
register: add_luks_slot2_check
- name: Add key in slot 2
luks_device:
device: "{{ cryptfile_device }}"
state: present
keyfile: "{{ remote_tmp_dir }}/keyfile1"
new_keyfile: "{{ remote_tmp_dir }}/keyfile2"
new_keyslot: 2
pbkdf:
iteration_time: 0.1
become: true
register: add_luks_slot2
- name: Add key in slot 2 (idempotent)
luks_device:
device: "{{ cryptfile_device }}"
state: present
keyfile: "{{ remote_tmp_dir }}/keyfile1"
new_keyfile: "{{ remote_tmp_dir }}/keyfile2"
new_keyslot: 2
pbkdf:
iteration_time: 0.1
become: true
register: add_luks_slot2_idem
- name: Add key in slot 2 (idempotent, check)
luks_device:
device: "{{ cryptfile_device }}"
state: present
keyfile: "{{ remote_tmp_dir }}/keyfile1"
new_keyfile: "{{ remote_tmp_dir }}/keyfile2"
new_keyslot: 2
pbkdf:
iteration_time: 0.1
check_mode: true
become: true
register: add_luks_slot2_idem_check
- name: Dump luks header
command: "cryptsetup luksDump {{ cryptfile_device }}"
become: true
register: luks_header_slot2
- assert:
that:
- add_luks_slot2_check is changed
- add_luks_slot2 is changed
- add_luks_slot2_idem is not changed
- add_luks_slot2_idem_check is not changed
- "'Key Slot 2: ENABLED' in luks_header_slot2.stdout or '2: luks2' in luks_header_slot2.stdout"
- name: Check remove slot 4 without key
luks_device:
device: "{{ cryptfile_device }}"
remove_keyslot: 4
ignore_errors: true
become: true
register: kill_slot4_nokey
- name: Check remove slot 4 with slot 4 key
luks_device:
device: "{{ cryptfile_device }}"
remove_keyslot: 4
keyfile: "{{ remote_tmp_dir }}/keyfile1"
ignore_errors: true
become: true
register: kill_slot4_key_slot4
- assert:
that:
- kill_slot4_nokey is failed
- kill_slot4_key_slot4 is failed
- name: Remove key in slot 4 (check)
luks_device:
device: "{{ cryptfile_device }}"
keyfile: "{{ remote_tmp_dir }}/keyfile2"
remove_keyslot: 4
check_mode: true
become: true
register: kill_luks_slot4_check
- name: Remove key in slot 4
luks_device:
device: "{{ cryptfile_device }}"
keyfile: "{{ remote_tmp_dir }}/keyfile2"
remove_keyslot: 4
become: true
register: kill_luks_slot4
- name: Remove key in slot 4 (idempotent)
luks_device:
device: "{{ cryptfile_device }}"
keyfile: "{{ remote_tmp_dir }}/keyfile2"
remove_keyslot: 4
become: true
register: kill_luks_slot4_idem
- name: Remove key in slot 4 (idempotent)
luks_device:
device: "{{ cryptfile_device }}"
keyfile: "{{ remote_tmp_dir }}/keyfile2"
remove_keyslot: 4
check_mode: true
become: true
register: kill_luks_slot4_idem_check
- name: Dump luks header
command: "cryptsetup luksDump {{ cryptfile_device }}"
become: true
register: luks_header_slot4_removed
- assert:
that:
- kill_luks_slot4_check is changed
- kill_luks_slot4 is changed
- kill_luks_slot4_idem is not changed
- kill_luks_slot4_idem_check is not changed
- "'Key Slot 4: DISABLED' in luks_header_slot4_removed.stdout or not '4: luks' in luks_header_slot4_removed.stdout"

79
tests/integration/targets/luks_device/tasks/tests/keyslot-options.yml Normal file
View File

@ -0,0 +1,79 @@
---
# Copyright (c) Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
- name: Check invalid slot (luks1, 8)
luks_device:
device: "{{ cryptfile_device }}"
state: present
type: luks1
keyfile: "{{ remote_tmp_dir }}/keyfile1"
keyslot: 8
pbkdf:
iteration_time: 0.1
ignore_errors: true
become: true
register: create_luks1_slot8
- name: Check invalid slot (luks2, 32)
luks_device:
device: "{{ cryptfile_device }}"
state: present
type: luks2
keyfile: "{{ remote_tmp_dir }}/keyfile1"
keyslot: 32
pbkdf:
iteration_time: 0.1
ignore_errors: true
become: true
register: create_luks2_slot32
- name: Check invalid slot (no luks type, 8)
luks_device:
device: "{{ cryptfile_device }}"
state: present
keyfile: "{{ remote_tmp_dir }}/keyfile1"
keyslot: 8
pbkdf:
iteration_time: 0.1
ignore_errors: true
become: true
register: create_luks_slot8
- assert:
that:
- create_luks1_slot8 is failed
- create_luks2_slot32 is failed
- create_luks_slot8 is failed
- name: Check valid slot (luks2, 8)
luks_device:
device: "{{ cryptfile_device }}"
state: present
type: luks2
keyfile: "{{ remote_tmp_dir }}/keyfile1"
keyslot: 8
pbkdf:
iteration_time: 0.1
become: true
ignore_errors: true
register: create_luks2_slot8
- name: Make sure that the previous task only fails if LUKS2 is not supported
assert:
that:
- "'Unknown option --type' in create_luks2_slot8.msg"
when: create_luks2_slot8 is failed
- name: Check add valid slot (no luks type, 10)
luks_device:
device: "{{ cryptfile_device }}"
state: present
keyfile: "{{ remote_tmp_dir }}/keyfile1"
new_keyfile: "{{ remote_tmp_dir }}/keyfile2"
new_keyslot: 10
pbkdf:
iteration_time: 0.1
become: true
register: create_luks_slot10
when: create_luks2_slot8 is changed
- assert:
that:
- create_luks_slot10 is changed
when: create_luks2_slot8 is changed

29
tests/unit/plugins/modules/test_luks_device.py
View File

@ -148,16 +148,16 @@ LUKS_ADD_KEY_DATA = (
# device, remove_key, remove_passphrase, state, label, expected
LUKS_REMOVE_KEY_DATA = (
("dummy", "key", None, "present", None, True),
(None, "key", None, "present", None, False),
(None, "key", None, "present", "labelName", True),
("dummy", None, None, "present", None, False),
("dummy", "key", None, "absent", None, "exception"),
("dummy", None, "foo", "present", None, True),
(None, None, "foo", "present", None, False),
(None, None, "foo", "present", "labelName", True),
("dummy", None, None, "present", None, False),
("dummy", None, "foo", "absent", None, "exception"))
("dummy", "key", None, None, "present", None, True),
(None, "key", None, None, "present", None, False),
(None, "key", None, None, "present", "labelName", True),
("dummy", None, None, None, "present", None, False),
("dummy", "key", None, None, "absent", None, "exception"),
("dummy", None, "foo", None, "present", None, True),
(None, None, "foo", None, "present", None, False),
(None, None, "foo", None, "present", "labelName", True),
("dummy", None, None, None, "present", None, False),
("dummy", None, "foo", None, "absent", None, "exception"))
@pytest.mark.parametrize("device, keyfile, passphrase, state, is_luks, " +
@ -291,17 +291,18 @@ def test_luks_add_key(device, keyfile, passphrase, new_keyfile, new_passphrase,
assert expected == "exception"
@pytest.mark.parametrize("device, remove_keyfile, remove_passphrase, state, " +
"label, expected",
((d[0], d[1], d[2], d[3], d[4], d[5])
@pytest.mark.parametrize("device, remove_keyfile, remove_passphrase, remove_keyslot, " +
"state, label, expected",
((d[0], d[1], d[2], d[3], d[4], d[5], d[6])
for d in LUKS_REMOVE_KEY_DATA))
def test_luks_remove_key(device, remove_keyfile, remove_passphrase, state,
def test_luks_remove_key(device, remove_keyfile, remove_passphrase, remove_keyslot, state,
label, expected, monkeypatch):
module = DummyModule()
module.params["device"] = device
module.params["remove_keyfile"] = remove_keyfile
module.params["remove_passphrase"] = remove_passphrase
module.params["remove_keyslot"] = remove_keyslot
module.params["state"] = state
module.params["label"] = label