rpms/leapp-repository
7
0
Fork 1
Code Issues Pull Requests Releases Activity
c8
leapp-repository/SOURCES/0019-LuksScanner-Add-LUKS-dump-scanner-and-models.patch

1031 lines
40 KiB
Diff
Raw Permalink Normal View History

Import from CS git
2024-11-25 09:10:29 +00:00
From 266c2495b144aa13d96f72c276d7b94638e3a6b7 Mon Sep 17 00:00:00 2001
From: Daniel Zatovic <daniel.zatovic@gmail.com>
Date: Tue, 16 Apr 2024 17:04:41 +0200
Subject: [PATCH 19/40] LuksScanner: Add LUKS dump scanner and models
Add LuksScanner actor that runs 'cryptsetup luksDump' for all 'crypt'
from lsblk output. The output is then parsed and filled into LuksDump
and LuksToken models.
The LuksDump model contains information about LUKS version, device UUID,
corresponding device path, name of the backing device (which contains
the LUKS header) and a list of LuksToken models.
LuksToken model represents a token associated with the given LUKS
device. It contains token ID, IDs of associated keyslot and token type.
If the token type is "clevis", we use "clevis luks list" command to
determine the clevis-specific subtype and append it to the token name.
E.g. if there is a "clevis" token and "clevis luks list" returns "tpm2",
the token type will be "clevis-tpm2".
---
.../common/actors/luksscanner/actor.py | 23 ++
.../luksscanner/libraries/luksdump_parser.py | 199 ++++++++++++++++++
.../luksscanner/libraries/luksscanner.py | 125 +++++++++++
.../tests/files/luksDump_luks1.txt | 27 +++
.../tests/files/luksDump_nvme0n1p3_luks1.txt | 27 +++
.../tests/files/luksDump_nvme0n1p3_luks2.txt | 43 ++++
.../files/luksDump_nvme0n1p3_luks2_tokens.txt | 119 +++++++++++
.../luksscanner/tests/test_luksdump_parser.py | 147 +++++++++++++
.../luksscanner/tests/test_luksscaner.py | 142 +++++++++++++
.../system_upgrade/common/models/luksdump.py | 73 +++++++
10 files changed, 925 insertions(+)
create mode 100644 repos/system_upgrade/common/actors/luksscanner/actor.py
create mode 100755 repos/system_upgrade/common/actors/luksscanner/libraries/luksdump_parser.py
create mode 100644 repos/system_upgrade/common/actors/luksscanner/libraries/luksscanner.py
create mode 100644 repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_luks1.txt
create mode 100644 repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks1.txt
create mode 100644 repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks2.txt
create mode 100644 repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks2_tokens.txt
create mode 100644 repos/system_upgrade/common/actors/luksscanner/tests/test_luksdump_parser.py
create mode 100644 repos/system_upgrade/common/actors/luksscanner/tests/test_luksscaner.py
create mode 100644 repos/system_upgrade/common/models/luksdump.py
diff --git a/repos/system_upgrade/common/actors/luksscanner/actor.py b/repos/system_upgrade/common/actors/luksscanner/actor.py
new file mode 100644
index 00000000..a163374b
--- /dev/null
+++ b/repos/system_upgrade/common/actors/luksscanner/actor.py
@@ -0,0 +1,23 @@
+from leapp.actors import Actor
+from leapp.libraries.actor import luksscanner
+from leapp.models import LuksDumps, StorageInfo
+from leapp.reporting import Report
+from leapp.tags import FactsPhaseTag, IPUWorkflowTag
+
+
+class LuksScanner(Actor):
+ """
+ Provides data about active LUKS devices.
+
+ Scans all block devices of 'crypt' type and attempts to run 'cryptsetup luksDump' on them.
+ For every 'crypt' device a LuksDump model is produced. Furthermore, if there is any LUKS token
+ of type clevis, the concrete subtype is determined using 'clevis luks list'.
+ """
+
+ name = 'luks_scanner'
+ consumes = (StorageInfo,)
+ produces = (Report, LuksDumps)
+ tags = (IPUWorkflowTag, FactsPhaseTag)
+
+ def process(self):
+ self.produce(luksscanner.get_luks_dumps_model())
diff --git a/repos/system_upgrade/common/actors/luksscanner/libraries/luksdump_parser.py b/repos/system_upgrade/common/actors/luksscanner/libraries/luksdump_parser.py
new file mode 100755
index 00000000..44113d0e
--- /dev/null
+++ b/repos/system_upgrade/common/actors/luksscanner/libraries/luksdump_parser.py
@@ -0,0 +1,199 @@
+class LuksDumpParser(object):
+ """
+ Class for parsing "cryptsetup luksDump" output. Given a list of lines, it
+ generates a dictionary representing the dump.
+ """
+
+ class Node(object):
+ """
+ Helper class, every line is represented as a node. The node depth is
+ based on the indentation of the line. A dictionary is produced after
+ all lines are inserted.
+ """
+
+ def __init__(self, indented_line):
+ self.children = []
+ self.level = len(indented_line) - len(indented_line.lstrip())
+ self.text = indented_line.strip()
+
+ def add_children(self, nodes):
+ # NOTE(pstodulk): it's expected that nodes are non-empty list and
+ # having it empty is an error if it happens. So keeping a hard crash
+ # for now as having an empty list it's hypothetical now and I would
+ # probably end with en error anyway if discovered.
+ childlevel = nodes[0].level
+ while nodes:
+ node = nodes.pop(0)
+ if node.level == childlevel: # add node as a child
+ self.children.append(node)
+ elif node.level > childlevel: # add nodes as grandchildren of the last child
+ nodes.insert(0, node)
+ self.children[-1].add_children(nodes)
+ elif node.level <= self.level: # this node is a sibling, no more children
+ nodes.insert(0, node)
+ return
+
+ def as_dict(self):
+ if len(self.children) > 1:
+ children = [node.as_dict() for node in self.children]
+
+ return {self.text: LuksDumpParser._merge_list(children)}
+ if len(self.children) == 1:
+ return {self.text: self.children[0].as_dict()}
+ return self.text
+
+ @staticmethod
+ def _count_type(elem_list, elem_type):
+ """ Count the number of items of elem_type inside the elem_list """
+ return sum(isinstance(x, elem_type) for x in elem_list)
+
+ @staticmethod
+ def _merge_list(elem_list):
+ """
+ Given a list of elements merge them into a single element. If all
+ elements are strings, concatenate them into a single string. When all
+ the elements are dictionaries merge them into a single dictionary
+ containing the keys/values from all of the dictionaries.
+ """
+
+ dict_count = LuksDumpParser._count_type(elem_list, dict)
+ str_count = LuksDumpParser._count_type(elem_list, str)
+
+ result = elem_list
+ if dict_count == len(elem_list):
+ result = {}
+ for element in elem_list:
+ result.update(element)
+ elif str_count == len(elem_list):
+ result = "".join(elem_list)
+
+ return result
+
+ @staticmethod
+ def _find_single_str(elem_list):
+ """ If the list contains exactly one string return it or return None otherwise. """
+
+ result = None
+
+ for elem in elem_list:
+ if isinstance(elem, str):
+ if result is not None:
+ # more than one strings in the list
+ return None
+ result = elem
+
+ return result
+
+ @staticmethod
+ def _fixup_type(elem_list, type_string):
+ single_string = LuksDumpParser._find_single_str(elem_list)
+
+ if single_string is not None:
+ elem_list.remove(single_string)
+ elem_list.append({type_string: single_string})
+
+ @staticmethod
+ def _fixup_section(section, type_string):
+ for key, value in section.items():
+ LuksDumpParser._fixup_type(value, type_string)
+ section[key] = LuksDumpParser._merge_list(section[key])
+
+ @staticmethod
+ def _fixup_dict(parsed_dict):
+ """ Various fixups of the parsed dictionary """
+
+ if "Version" not in parsed_dict:
+ return
+ if parsed_dict["Version"] == "1":
+ for i in range(8):
+ keyslot = "Key Slot {}".format(i)
+
+ if keyslot not in parsed_dict:
+ continue
+
+ if parsed_dict[keyslot] in ["ENABLED", "DISABLED"]:
+ parsed_dict[keyslot] = {"enabled": parsed_dict[keyslot] == "ENABLED"}
+
+ if not isinstance(parsed_dict[keyslot], list):
+ continue
+
+ enabled = None
+ if "ENABLED" in parsed_dict[keyslot]:
+ enabled = True
+ parsed_dict[keyslot].remove("ENABLED")
+ if "DISABLED" in parsed_dict[keyslot]:
+ enabled = False
+ parsed_dict[keyslot].remove("DISABLED")
+ parsed_dict[keyslot] = LuksDumpParser._merge_list(parsed_dict[keyslot])
+ if enabled is not None:
+ parsed_dict[keyslot]["enabled"] = enabled
+ elif parsed_dict["Version"] == "2":
+ for section in ["Keyslots", "Digests", "Data segments", "Tokens"]:
+ if section in parsed_dict:
+ LuksDumpParser._fixup_section(parsed_dict[section], "type")
+
+ @staticmethod
+ def _fixup_dump(dump):
+ """
+ Replace tabs with spaces, for lines with colon a move the text
+ after column on new line with the indent of the following line.
+ """
+
+ dump = [line.replace("\t", " "*8).replace("\n", "") for line in dump]
+ newdump = []
+
+ for i, line in enumerate(dump):
+ if not line.strip():
+ continue
+
+ if ':' in line:
+ first_half = line.split(":")[0]
+ second_half = ":".join(line.split(":")[1:]).lstrip()
+
+ current_level = len(line) - len(line.lstrip())
+ if i+1 < len(dump):
+ next_level = len(dump[i+1]) - len(dump[i+1].lstrip())
+ else:
+ next_level = current_level
+
+ if next_level > current_level:
+ second_half = " " * next_level + second_half
+ else:
+ second_half = " " * (current_level + 8) + second_half
+
+ newdump.append(first_half)
+ if second_half.strip():
+ newdump.append(second_half)
+ else:
+ newdump.append(line)
+
+ return newdump
+
+ @staticmethod
+ def parse(dump):
+ """
+ Parse the output of "cryptsetup luksDump" command into a dictionary.
+
+ :param dump: List of output lines of luksDump
+ :returns: Parsed dictionary
+ """
+
+ root = LuksDumpParser.Node('root')
+
+ nodes = []
+ for line in LuksDumpParser._fixup_dump(dump):
+ nodes.append(LuksDumpParser.Node(line))
+
+ root.add_children(nodes)
+ root = root.as_dict()['root']
+
+ if isinstance(root, list):
+ result = {}
+ for child in root:
+ if isinstance(child, str):
+ child = {child: {}}
+ result.update(child)
+ root = result
+
+ LuksDumpParser._fixup_dict(root)
+ return root
diff --git a/repos/system_upgrade/common/actors/luksscanner/libraries/luksscanner.py b/repos/system_upgrade/common/actors/luksscanner/libraries/luksscanner.py
new file mode 100644
index 00000000..1c7822a5
--- /dev/null
+++ b/repos/system_upgrade/common/actors/luksscanner/libraries/luksscanner.py
@@ -0,0 +1,125 @@
+import functools
+
+from leapp.exceptions import StopActorExecutionError
+from leapp.libraries import stdlib
+from leapp.libraries.actor.luksdump_parser import LuksDumpParser
+from leapp.libraries.stdlib import api
+from leapp.models import LuksDump, LuksDumps, LuksToken, StorageInfo
+
+
+def aslist(f):
+ """ Decorator used to convert generator to list """
+ @functools.wraps(f)
+ def inner(*args, **kwargs):
+ return list(f(*args, **kwargs))
+ return inner
+
+
+def _get_clevis_type(device_path, keyslot):
+ """
+ Assuming the device is initialized using clevis, determine the type of
+ clevis token associated to the specified keyslot.
+ """
+ try:
+ result = stdlib.run(["clevis", "luks", "list", "-d", device_path, "-s", str(keyslot)])
+ except OSError:
+ message = ('A LUKS drive with clevis token was discovered, but there is '
+ 'no clevis package installed. The clevis command is required '
+ 'to determine clevis token type.')
+ details = {'hint': 'Use dnf to install the "clevis-luks" package.'}
+ raise StopActorExecutionError(message=message, details=details)
+ except stdlib.CalledProcessError as e:
+ api.current_logger().debug("clevis list command failed with an error code: {}".format(e.exit_code))
+
+ message = ('The "clevis luks list" command failed. This'
+ 'might be because the clevis-luks package is'
+ 'missing on your system.')
+ details = {'hint': 'Use dnf to install the "clevis-luks" package.'}
+ raise StopActorExecutionError(message=message, details=details)
+
+ line = result["stdout"].split()
+ if len(line) != 3:
+ raise StopActorExecutionError(
+ 'Invalid "clevis list" output detected'
+ )
+
+ return "clevis-{}".format(line[1])
+
+
+@aslist
+def _get_tokens(device_path, luksdump_dict):
+ """ Given a parsed LUKS dump, produce a list of tokens """
+ if "Version" not in luksdump_dict or luksdump_dict["Version"] != '2':
+ return
+ if "Tokens" not in luksdump_dict:
+ raise StopActorExecutionError(
+ 'No tokens in cryptsetup luksDump output'
+ )
+
+ for token_id in luksdump_dict["Tokens"]:
+ token = luksdump_dict["Tokens"][token_id]
+
+ if "Keyslot" not in token or "type" not in token:
+ raise StopActorExecutionError(
+ 'Token specification does not contain keyslot or type',
+ )
+ keyslot = int(token["Keyslot"])
+ token_type = token["type"]
+
+ if token_type == "clevis":
+ token_type = _get_clevis_type(device_path, keyslot)
+
+ yield LuksToken(
+ token_id=int(token_id),
+ keyslot=keyslot,
+ token_type=token_type
+ )
+
+
+def get_luks_dump_by_device(device_path, device_name):
+ """ Determine info about LUKS device using cryptsetup and clevis commands """
+
+ try:
+ result = stdlib.run(['cryptsetup', 'luksDump', device_path])
+ luksdump_dict = LuksDumpParser.parse(result["stdout"].splitlines())
+
+ version = int(luksdump_dict["Version"]) if "Version" in luksdump_dict else None
+ uuid = luksdump_dict["UUID"] if "UUID" in luksdump_dict else None
+ if version is None or uuid is None:
+ api.current_logger().error(
+ 'Failed to detect UUID or version from the output "cryptsetup luksDump {}" command'.format(device_path)
+ )
+ raise StopActorExecutionError(
+ 'Failed to detect UUID or version from the output "cryptsetup luksDump {}" command'.format(device_path)
+ )
+
+ return LuksDump(
+ version=version,
+ uuid=uuid,
+ device_path=device_path,
+ device_name=device_name,
+ tokens=_get_tokens(device_path, luksdump_dict)
+ )
+
+ except (OSError, stdlib.CalledProcessError) as ex:
+ api.current_logger().error(
+ 'Failed to execute "cryptsetup luksDump" command: {}'.format(ex)
+ )
+ raise StopActorExecutionError(
+ 'Failed to execute "cryptsetup luksDump {}" command'.format(device_path),
+ details={'details': str(ex)}
+ )
+
+
+@aslist
+def get_luks_dumps():
+ """ Collect info abaout every active LUKS device """
+
+ for storage_info in api.consume(StorageInfo):
+ for blk in storage_info.lsblk:
+ if blk.tp == 'crypt' and blk.parent_path:
+ yield get_luks_dump_by_device(blk.parent_path, blk.parent_name)
+
+
+def get_luks_dumps_model():
+ return LuksDumps(dumps=get_luks_dumps())
diff --git a/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_luks1.txt b/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_luks1.txt
new file mode 100644
index 00000000..e22cc8ce
--- /dev/null
+++ b/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_luks1.txt
@@ -0,0 +1,27 @@
+LUKS header information for /dev/loop10
+
+Version: 1
+Cipher name: aes
+Cipher mode: xts-plain64
+Hash spec: sha256
+Payload offset: 4096
+MK bits: 512
+MK digest: fb ec 6b 31 ae e4 49 03 3e ad 43 22 02 cf a8 78 ad 3c d2 a8
+MK salt: 17 57 4e 2f ed 0b 5c 62 d5 de 54 f5 7f ab 60 68
+ 71 d8 72 06 64 6c 81 05 39 55 3f 55 32 56 d9 da
+MK iterations: 114573
+UUID: 90242257-d00a-4019-aba6-03083f89404b
+
+Key Slot 0: ENABLED
+ Iterations: 1879168
+ Salt: fc 77 48 72 bd 31 ca 83 23 80 5a 5e b9 5b de bb
+ 55 ac d5 a9 3b 96 ad a5 82 bc 11 68 ba f8 87 56
+ Key material offset: 8
+ AF stripes: 4000
+Key Slot 1: DISABLED
+Key Slot 2: DISABLED
+Key Slot 3: DISABLED
+Key Slot 4: DISABLED
+Key Slot 5: DISABLED
+Key Slot 6: DISABLED
+Key Slot 7: DISABLED
diff --git a/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks1.txt b/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks1.txt
new file mode 100644
index 00000000..e22cc8ce
--- /dev/null
+++ b/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks1.txt
@@ -0,0 +1,27 @@
+LUKS header information for /dev/loop10
+
+Version: 1
+Cipher name: aes
+Cipher mode: xts-plain64
+Hash spec: sha256
+Payload offset: 4096
+MK bits: 512
+MK digest: fb ec 6b 31 ae e4 49 03 3e ad 43 22 02 cf a8 78 ad 3c d2 a8
+MK salt: 17 57 4e 2f ed 0b 5c 62 d5 de 54 f5 7f ab 60 68
+ 71 d8 72 06 64 6c 81 05 39 55 3f 55 32 56 d9 da
+MK iterations: 114573
+UUID: 90242257-d00a-4019-aba6-03083f89404b
+
+Key Slot 0: ENABLED
+ Iterations: 1879168
+ Salt: fc 77 48 72 bd 31 ca 83 23 80 5a 5e b9 5b de bb
+ 55 ac d5 a9 3b 96 ad a5 82 bc 11 68 ba f8 87 56
+ Key material offset: 8
+ AF stripes: 4000
+Key Slot 1: DISABLED
+Key Slot 2: DISABLED
+Key Slot 3: DISABLED
+Key Slot 4: DISABLED
+Key Slot 5: DISABLED
+Key Slot 6: DISABLED
+Key Slot 7: DISABLED
diff --git a/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks2.txt b/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks2.txt
new file mode 100644
index 00000000..407261f4
--- /dev/null
+++ b/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks2.txt
@@ -0,0 +1,43 @@
+LUKS header information
+Version: 2
+Epoch: 3
+Metadata area: 16384 [bytes]
+Keyslots area: 16744448 [bytes]
+UUID: dfd8db30-2b65-4be9-8cae-65f5fac4a06f
+Label: (no label)
+Subsystem: (no subsystem)
+Flags: (no flags)
+
+Data segments:
+ 0: crypt
+ offset: 16777216 [bytes]
+ length: (whole device)
+ cipher: aes-xts-plain64
+ sector: 512 [bytes]
+
+Keyslots:
+ 0: luks2
+ Key: 512 bits
+ Priority: normal
+ Cipher: aes-xts-plain64
+ Cipher key: 512 bits
+ PBKDF: argon2id
+ Time cost: 7
+ Memory: 1048576
+ Threads: 4
+ Salt: 1d d5 97 97 dd 45 e2 d7 2b a7 0b fa c4 7f b3 f4
+ ef 4e 5f 95 e0 ba fd 7a 7e 36 02 69 f8 44 96 d8
+ AF stripes: 4000
+ AF hash: sha256
+ Area offset:32768 [bytes]
+ Area length:258048 [bytes]
+ Digest ID: 0
+Tokens:
+Digests:
+ 0: pbkdf2
+ Hash: sha256
+ Iterations: 99750
+ Salt: 10 1d a1 21 8b 93 dc bb f1 ab 2b 1b 89 8e 3d c4
+ 18 07 51 08 ef f5 95 da 9f 85 fa d7 de c9 c4 96
+ Digest: 4f 27 4c 19 ae 72 b1 75 ef 53 c0 6d ff db 7f fe
+ f1 67 d0 c3 67 03 0c 14 3a 6f 6a 1a 87 a8 6f 32
diff --git a/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks2_tokens.txt b/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks2_tokens.txt
new file mode 100644
index 00000000..c2a7464c
--- /dev/null
+++ b/repos/system_upgrade/common/actors/luksscanner/tests/files/luksDump_nvme0n1p3_luks2_tokens.txt
@@ -0,0 +1,119 @@
+LUKS header information
+Version: 2
+Epoch: 9
+Metadata area: 16384 [bytes]
+Keyslots area: 16744448 [bytes]
+UUID: 6b929b85-b01e-4aa3-8ad2-a05decae6e3d
+Label: (no label)
+Subsystem: (no subsystem)
+Flags: (no flags)
+
+Data segments:
+ 0: crypt
+ offset: 16777216 [bytes]
+ length: (whole device)
+ cipher: aes-xts-plain64
+ sector: 512 [bytes]
+
+Keyslots:
+ 0: luks2
+ Key: 512 bits
+ Priority: normal
+ Cipher: aes-xts-plain64
+ Cipher key: 512 bits
+ PBKDF: argon2id
+ Time cost: 7
+ Memory: 1048576
+ Threads: 4
+ Salt: de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ AF stripes: 4000
+ AF hash: sha256
+ Area offset:32768 [bytes]
+ Area length:258048 [bytes]
+ Digest ID: 0
+ 1: luks2
+ Key: 512 bits
+ Priority: normal
+ Cipher: aes-xts-plain64
+ Cipher key: 512 bits
+ PBKDF: pbkdf2
+ Hash: sha256
+ Iterations: 1000
+ Salt: de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ AF stripes: 4000
+ AF hash: sha256
+ Area offset:290816 [bytes]
+ Area length:258048 [bytes]
+ Digest ID: 0
+ 2: luks2
+ Key: 512 bits
+ Priority: normal
+ Cipher: aes-xts-plain64
+ Cipher key: 512 bits
+ PBKDF: pbkdf2
+ Hash: sha256
+ Iterations: 1000
+ Salt: de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ AF stripes: 4000
+ AF hash: sha256
+ Area offset:548864 [bytes]
+ Area length:258048 [bytes]
+ Digest ID: 0
+ 3: luks2
+ Key: 512 bits
+ Priority: normal
+ Cipher: aes-xts-plain64
+ Cipher key: 512 bits
+ PBKDF: pbkdf2
+ Hash: sha512
+ Iterations: 1000
+ Salt: de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ AF stripes: 4000
+ AF hash: sha512
+ Area offset:806912 [bytes]
+ Area length:258048 [bytes]
+ Digest ID: 0
+Tokens:
+ 0: clevis
+ Keyslot: 1
+ 1: clevis
+ Keyslot: 2
+ 2: systemd-tpm2
+ tpm2-hash-pcrs: 7
+ tpm2-pcr-bank: sha256
+ tpm2-pubkey:
+ (null)
+ tpm2-pubkey-pcrs: n/a
+ tpm2-primary-alg: ecc
+ tpm2-blob: de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ tpm2-policy-hash:
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ tpm2-pin: false
+ tpm2-salt: false
+ Keyslot: 3
+Digests:
+ 0: pbkdf2
+ Hash: sha256
+ Iterations: 117448
+ Salt: de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ Digest: de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
+ de a1 b9 7f 03 cb b4 89 e2 52 20 fc e4 24 65 cd
diff --git a/repos/system_upgrade/common/actors/luksscanner/tests/test_luksdump_parser.py b/repos/system_upgrade/common/actors/luksscanner/tests/test_luksdump_parser.py
new file mode 100644
index 00000000..4b190149
--- /dev/null
+++ b/repos/system_upgrade/common/actors/luksscanner/tests/test_luksdump_parser.py
@@ -0,0 +1,147 @@
+import os
+
+from leapp.libraries.actor.luksdump_parser import LuksDumpParser
+from leapp.snactor.fixture import current_actor_context
+
+CUR_DIR = os.path.dirname(os.path.abspath(__file__))
+
+
+def test_luksdump_parser_luks1(current_actor_context):
+ f = open(os.path.join(CUR_DIR, 'files/luksDump_nvme0n1p3_luks1.txt'))
+ parsed_dict = LuksDumpParser.parse(f.readlines())
+
+ assert parsed_dict["Version"] == "1"
+ assert parsed_dict["Cipher name"] == "aes"
+ assert parsed_dict["Cipher mode"] == "xts-plain64"
+ assert parsed_dict["Hash spec"] == "sha256"
+ assert parsed_dict["Payload offset"] == "4096"
+ assert parsed_dict["MK bits"] == "512"
+ assert parsed_dict["MK digest"].replace(" ", "") == "fbec6b31aee449033ead432202cfa878ad3cd2a8"
+ assert parsed_dict["MK salt"].replace(" ", "") == "17574e2fed0b5c62d5de54f57fab6068"\
+ "71d87206646c810539553f553256d9da"
+ assert parsed_dict["MK iterations"] == "114573"
+ assert parsed_dict["UUID"] == "90242257-d00a-4019-aba6-03083f89404b"
+
+ assert parsed_dict["Key Slot 0"]["enabled"]
+ assert parsed_dict["Key Slot 0"]["Iterations"] == "1879168"
+ assert parsed_dict["Key Slot 0"]["Salt"].replace(" ", "") == "fc774872bd31ca8323805a5eb95bdebb" \
+ "55acd5a93b96ada582bc1168baf88756"
+ assert parsed_dict["Key Slot 0"]["Key material offset"] == "8"
+ assert parsed_dict["Key Slot 0"]["AF stripes"] == "4000"
+
+ assert not parsed_dict["Key Slot 1"]["enabled"]
+ assert not parsed_dict["Key Slot 2"]["enabled"]
+ assert not parsed_dict["Key Slot 3"]["enabled"]
+ assert not parsed_dict["Key Slot 4"]["enabled"]
+ assert not parsed_dict["Key Slot 5"]["enabled"]
+ assert not parsed_dict["Key Slot 6"]["enabled"]
+ assert not parsed_dict["Key Slot 7"]["enabled"]
+
+
+def test_luksdump_parser_luks2_tokens(current_actor_context):
+ f = open(os.path.join(CUR_DIR, 'files/luksDump_nvme0n1p3_luks2_tokens.txt'))
+ parsed_dict = LuksDumpParser.parse(f.readlines())
+
+ assert parsed_dict["Version"] == "2"
+ assert parsed_dict["Epoch"] == "9"
+ assert parsed_dict["Metadata area"] == "16384 [bytes]"
+ assert parsed_dict["Keyslots area"] == "16744448 [bytes]"
+ assert parsed_dict["UUID"] == "6b929b85-b01e-4aa3-8ad2-a05decae6e3d"
+ assert parsed_dict["Label"] == "(no label)"
+ assert parsed_dict["Subsystem"] == "(no subsystem)"
+ assert parsed_dict["Flags"] == "(no flags)"
+
+ assert len(parsed_dict["Data segments"]) == 1
+ assert parsed_dict["Data segments"]["0"]["type"] == "crypt"
+ assert parsed_dict["Data segments"]["0"]["offset"] == "16777216 [bytes]"
+ assert parsed_dict["Data segments"]["0"]["length"] == "(whole device)"
+ assert parsed_dict["Data segments"]["0"]["cipher"] == "aes-xts-plain64"
+ assert parsed_dict["Data segments"]["0"]["sector"] == "512 [bytes]"
+
+ assert len(parsed_dict["Keyslots"]) == 4
+ assert parsed_dict["Keyslots"]["0"]["type"] == "luks2"
+ assert parsed_dict["Keyslots"]["0"]["Key"] == "512 bits"
+ assert parsed_dict["Keyslots"]["0"]["Priority"] == "normal"
+ assert parsed_dict["Keyslots"]["0"]["Cipher"] == "aes-xts-plain64"
+ assert parsed_dict["Keyslots"]["0"]["Cipher key"] == "512 bits"
+ assert parsed_dict["Keyslots"]["0"]["PBKDF"] == "argon2id"
+ assert parsed_dict["Keyslots"]["0"]["Time cost"] == "7"
+ assert parsed_dict["Keyslots"]["0"]["Memory"] == "1048576"
+ assert parsed_dict["Keyslots"]["0"]["Threads"] == "4"
+ assert parsed_dict["Keyslots"]["0"]["Salt"].replace(" ", "") == 2*"dea1b97f03cbb489e25220fce42465cd"
+ assert parsed_dict["Keyslots"]["0"]["AF stripes"] == "4000"
+ assert parsed_dict["Keyslots"]["0"]["AF hash"] == "sha256"
+ assert parsed_dict["Keyslots"]["0"]["Area offset"] == "32768 [bytes]"
+ assert parsed_dict["Keyslots"]["0"]["Area length"] == "258048 [bytes]"
+ assert parsed_dict["Keyslots"]["0"]["Digest ID"] == "0"
+
+ assert parsed_dict["Keyslots"]["1"]["type"] == "luks2"
+ assert parsed_dict["Keyslots"]["1"]["Key"] == "512 bits"
+ assert parsed_dict["Keyslots"]["1"]["Priority"] == "normal"
+ assert parsed_dict["Keyslots"]["1"]["Cipher"] == "aes-xts-plain64"
+ assert parsed_dict["Keyslots"]["1"]["Cipher key"] == "512 bits"
+ assert parsed_dict["Keyslots"]["1"]["PBKDF"] == "pbkdf2"
+ assert parsed_dict["Keyslots"]["1"]["Hash"] == "sha256"
+ assert parsed_dict["Keyslots"]["1"]["Iterations"] == "1000"
+ assert parsed_dict["Keyslots"]["1"]["Salt"].replace(" ", "") == 2*"dea1b97f03cbb489e25220fce42465cd"
+ assert parsed_dict["Keyslots"]["1"]["AF stripes"] == "4000"
+ assert parsed_dict["Keyslots"]["1"]["AF hash"] == "sha256"
+ assert parsed_dict["Keyslots"]["1"]["Area offset"] == "290816 [bytes]"
+ assert parsed_dict["Keyslots"]["1"]["Area length"] == "258048 [bytes]"
+ assert parsed_dict["Keyslots"]["1"]["Digest ID"] == "0"
+
+ assert parsed_dict["Keyslots"]["2"]["type"] == "luks2"
+ assert parsed_dict["Keyslots"]["2"]["Key"] == "512 bits"
+ assert parsed_dict["Keyslots"]["2"]["Priority"] == "normal"
+ assert parsed_dict["Keyslots"]["2"]["Cipher"] == "aes-xts-plain64"
+ assert parsed_dict["Keyslots"]["2"]["Cipher key"] == "512 bits"
+ assert parsed_dict["Keyslots"]["2"]["PBKDF"] == "pbkdf2"
+ assert parsed_dict["Keyslots"]["2"]["Hash"] == "sha256"
+ assert parsed_dict["Keyslots"]["2"]["Iterations"] == "1000"
+ assert parsed_dict["Keyslots"]["2"]["Salt"].replace(" ", "") == 2*"dea1b97f03cbb489e25220fce42465cd"
+ assert parsed_dict["Keyslots"]["2"]["AF stripes"] == "4000"
+ assert parsed_dict["Keyslots"]["2"]["AF hash"] == "sha256"
+ assert parsed_dict["Keyslots"]["2"]["Area offset"] == "548864 [bytes]"
+ assert parsed_dict["Keyslots"]["2"]["Area length"] == "258048 [bytes]"
+ assert parsed_dict["Keyslots"]["2"]["Digest ID"] == "0"
+
+ assert parsed_dict["Keyslots"]["3"]["type"] == "luks2"
+ assert parsed_dict["Keyslots"]["3"]["Key"] == "512 bits"
+ assert parsed_dict["Keyslots"]["3"]["Priority"] == "normal"
+ assert parsed_dict["Keyslots"]["3"]["Cipher"] == "aes-xts-plain64"
+ assert parsed_dict["Keyslots"]["3"]["Cipher key"] == "512 bits"
+ assert parsed_dict["Keyslots"]["3"]["PBKDF"] == "pbkdf2"
+ assert parsed_dict["Keyslots"]["3"]["Hash"] == "sha512"
+ assert parsed_dict["Keyslots"]["3"]["Iterations"] == "1000"
+ assert parsed_dict["Keyslots"]["3"]["Salt"].replace(" ", "") == 2*"dea1b97f03cbb489e25220fce42465cd"
+ assert parsed_dict["Keyslots"]["3"]["AF stripes"] == "4000"
+ assert parsed_dict["Keyslots"]["3"]["AF hash"] == "sha512"
+ assert parsed_dict["Keyslots"]["3"]["Area offset"] == "806912 [bytes]"
+ assert parsed_dict["Keyslots"]["3"]["Area length"] == "258048 [bytes]"
+ assert parsed_dict["Keyslots"]["3"]["Digest ID"] == "0"
+
+ assert len(parsed_dict["Tokens"]) == 3
+ assert parsed_dict["Tokens"]["0"]["type"] == "clevis"
+ assert parsed_dict["Tokens"]["0"]["Keyslot"] == "1"
+
+ assert parsed_dict["Tokens"]["1"]["type"] == "clevis"
+ assert parsed_dict["Tokens"]["1"]["Keyslot"] == "2"
+
+ assert parsed_dict["Tokens"]["2"]["type"] == "systemd-tpm2"
+ assert parsed_dict["Tokens"]["2"]["Keyslot"] == "3"
+ assert parsed_dict["Tokens"]["2"]["tpm2-hash-pcrs"] == "7"
+ assert parsed_dict["Tokens"]["2"]["tpm2-pcr-bank"] == "sha256"
+ assert parsed_dict["Tokens"]["2"]["tpm2-pubkey"] == "(null)"
+ assert parsed_dict["Tokens"]["2"]["tpm2-pubkey-pcrs"] == "n/a"
+ assert parsed_dict["Tokens"]["2"]["tpm2-primary-alg"] == "ecc"
+ assert parsed_dict["Tokens"]["2"]["tpm2-blob"].replace(" ", "") == 14*"dea1b97f03cbb489e25220fce42465cd"
+ assert parsed_dict["Tokens"]["2"]["tpm2-policy-hash"].replace(" ", "") == 2*"dea1b97f03cbb489e25220fce42465cd"
+ assert parsed_dict["Tokens"]["2"]["tpm2-pin"] == "false"
+ assert parsed_dict["Tokens"]["2"]["tpm2-salt"] == "false"
+
+ assert len(parsed_dict["Digests"]) == 1
+ assert parsed_dict["Digests"]["0"]["type"] == "pbkdf2"
+ assert parsed_dict["Digests"]["0"]["Hash"] == "sha256"
+ assert parsed_dict["Digests"]["0"]["Iterations"] == "117448"
+ assert parsed_dict["Digests"]["0"]["Salt"].replace(" ", "") == 2*"dea1b97f03cbb489e25220fce42465cd"
+ assert parsed_dict["Digests"]["0"]["Digest"].replace(" ", "") == 2*"dea1b97f03cbb489e25220fce42465cd"
diff --git a/repos/system_upgrade/common/actors/luksscanner/tests/test_luksscaner.py b/repos/system_upgrade/common/actors/luksscanner/tests/test_luksscaner.py
new file mode 100644
index 00000000..22eb0946
--- /dev/null
+++ b/repos/system_upgrade/common/actors/luksscanner/tests/test_luksscaner.py
@@ -0,0 +1,142 @@
+import os
+
+import pytest
+
+from leapp.libraries.stdlib import api
+from leapp.models import LsblkEntry, LuksDumps, StorageInfo
+from leapp.snactor.fixture import current_actor_context
+
+CUR_DIR = os.path.dirname(os.path.abspath(__file__))
+
+TOKENS_ASSERT = {
+ 0: {
+ "keyslot": 1,
+ "token_type": "clevis-tpm2"
+ },
+ 1: {
+ "keyslot": 2,
+ "token_type": "clevis-tang"
+ },
+ 2: {
+ "keyslot": 3,
+ "token_type": "systemd-tpm2"
+ },
+}
+
+CLEVIS_KEYSLOTS = {
+ 1: 'tpm2 \'{"hash":"sha256","key":"rsa","pcr_bank":"sha256","pcr_ids":"0,1,7"}\'',
+ 2: 'tang \'{"url":"http://localhost"}\''
+}
+
+
+class MockedRun(object):
+ """Simple mock class for leapp.libraries.stdlib.run."""
+
+ def __init__(self, variant, clevis_keyslots):
+ """if exc_type provided, then it will be raised on
+ instance call.
+
+ :type exc_type: None or BaseException
+ """
+ self.logger = api.current_logger()
+
+ self.commands = []
+ self.variant = variant
+ self.clevis_keyslots = clevis_keyslots
+
+ def __call__(self, cmd, *args, **kwargs):
+ self.commands.append(cmd)
+
+ if len(cmd) == 3 and cmd[:2] == ['cryptsetup', 'luksDump']:
+ dev_path = cmd[2]
+
+ # We cannot have the output in a list, since the command is called per device. Therefore, we have to map
+ # each device path to its output.
+ output_files_per_device = {
+ '/dev/nvme0n1p3': 'luksDump_nvme0n1p3{}.txt'.format(("_" + self.variant) if self.variant else "")
+ }
+
+ if dev_path not in output_files_per_device:
+ raise ValueError(
+ 'Attempting to call "cryptsetup luksDump" on an unexpected device: {}'.format(dev_path)
+ )
+ with open(os.path.join(CUR_DIR, 'files/{}'.format(output_files_per_device[dev_path]))) as f:
+ return {"stdout": f.read()}
+ elif len(cmd) >= 3 and cmd[:3] == ['clevis', 'luks', 'list']:
+ dev_path = None
+ keyslot = None
+
+ device_flag = False
+ keyslot_flag = False
+ for element in cmd:
+ if device_flag:
+ dev_path = element
+ elif keyslot_flag:
+ keyslot = element
+
+ device_flag = element == "-d"
+ keyslot_flag = element == "-s"
+
+ if dev_path is None or keyslot is None:
+ raise ValueError('Attempting to call "clevis luks list" without specifying keyslot or device')
+ if dev_path is None or keyslot is None or dev_path != "/dev/nvme0n1p3":
+ raise ValueError('Attempting to call "clevis luks list" on invalid device')
+
+ keyslot = int(keyslot)
+
+ if keyslot in self.clevis_keyslots:
+ return {"stdout": "{}: {}".format(keyslot, self.clevis_keyslots[keyslot])}
+
+ return {}
+
+
+@pytest.mark.parametrize(
+ ("variant", "luks_version", "uuid", "tokens_assert"),
+ [
+ ('luks1', 1, '90242257-d00a-4019-aba6-03083f89404b', {}),
+ ('luks2', 2, 'dfd8db30-2b65-4be9-8cae-65f5fac4a06f', {}),
+ ('luks2_tokens', 2, '6b929b85-b01e-4aa3-8ad2-a05decae6e3d', TOKENS_ASSERT),
+ ]
+)
+def test_actor_with_luks(monkeypatch, current_actor_context, variant, luks_version, uuid, tokens_assert):
+ mocked_run = MockedRun(variant, CLEVIS_KEYSLOTS)
+ monkeypatch.setattr('leapp.libraries.stdlib.run', mocked_run)
+
+ with_luks = [
+ LsblkEntry(
+ name='/dev/nvme0n1', kname='/dev/nvme0n1', maj_min='259:0', rm='0', size='10G', bsize=10*(1 << 39),
+ ro='0', tp='disk', parent_name='', parent_path='', mountpoint=''
+ ),
+ LsblkEntry(
+ name='/dev/nvme0n1p3', kname='/dev/nvme0n1p3', maj_min='259:3', rm='0', size='10G', bsize=10*(1 << 39),
+ ro='0', tp='part', parent_name='nvme0n1', parent_path='/dev/nvme0n1', mountpoint=''
+ ),
+ LsblkEntry(
+ name='/dev/mapper/tst1', kname='/dev/dm-0', maj_min='253:0', rm='0', size='9G', bsize=9*(1 << 39), ro='0',
+ tp='crypt', parent_name='nvme0n1p3', parent_path='/dev/nvme0n1p3', mountpoint=''
+ ),
+ # PKNAME is not set, so this crypt device will be ignored
+ LsblkEntry(
+ name='/dev/mapper/tst2', kname='/dev/dm-1', maj_min='253:0', rm='0', size='9G', bsize=9*(1 << 39), ro='0',
+ tp='crypt', parent_name='', parent_path='', mountpoint=''
+ )
+ ]
+
+ current_actor_context.feed(StorageInfo(lsblk=with_luks))
+ current_actor_context.run()
+
+ luks_dumps = current_actor_context.consume(LuksDumps)
+ assert len(luks_dumps) == 1
+ assert len(luks_dumps[0].dumps) == 1
+ luks_dump = luks_dumps[0].dumps[0]
+
+ assert luks_dump.version == luks_version
+ assert luks_dump.uuid == uuid
+ assert luks_dump.device_name == "nvme0n1p3"
+ assert luks_dump.device_path == "/dev/nvme0n1p3"
+ assert len(luks_dump.tokens) == len(tokens_assert)
+
+ for token in luks_dump.tokens:
+ assert token.token_id in tokens_assert
+ assert token.keyslot == tokens_assert[token.token_id]["keyslot"]
+ assert token.token_type == tokens_assert[token.token_id]["token_type"]
diff --git a/repos/system_upgrade/common/models/luksdump.py b/repos/system_upgrade/common/models/luksdump.py
new file mode 100644
index 00000000..83b56ef8
--- /dev/null
+++ b/repos/system_upgrade/common/models/luksdump.py
@@ -0,0 +1,73 @@
+from leapp.models import fields, Model
+from leapp.topics import SystemInfoTopic
+
+
+class LuksToken(Model):
+ """
+ Represents a single token associated with the LUKS device.
+
+ Note this model is supposed to be used just as part of the LuksDump msg.
+ """
+ topic = SystemInfoTopic
+
+ token_id = fields.Integer()
+ """
+ Token ID (as seen in the luksDump)
+ """
+
+ keyslot = fields.Integer()
+ """
+ ID of the associated keyslot
+ """
+
+ token_type = fields.String()
+ """
+ Type of the token. For "clevis" type the concrete subtype (determined using
+ clevis luks list) is appended e.g. clevis-tpm2. clevis-tang, ...
+ """
+
+
+class LuksDump(Model):
+ """
+ Information about a single LUKS-encrypted device.
+
+ Note this model is supposed to be used as a part of LuksDumps msg.
+ """
+ topic = SystemInfoTopic
+
+ version = fields.Integer()
+ """
+ LUKS version
+ """
+
+ uuid = fields.String()
+ """
+ UUID of the LUKS device
+ """
+
+ device_path = fields.String()
+ """
+ Full path to the backing device
+ """
+
+ device_name = fields.String()
+ """
+ Device name of the backing device
+ """
+
+ tokens = fields.List(fields.Model(LuksToken), default=[])
+ """
+ List of LUKS2 tokens
+ """
+
+
+class LuksDumps(Model):
+ """
+ Information about all LUKS-encrypted devices on the system.
+ """
+ topic = SystemInfoTopic
+
+ dumps = fields.List(fields.Model(LuksDump))
+ """
+ List of LuksDump representing all the encrypted devices on the system.
+ """
--
2.47.0
Reference in New Issue Copy Permalink