Display alert if volume key missing or status unknown

src/stratis_cli/_actions/_introspect.py

@@ -266,6 +266,9 @@
     <property name="Uuid" type="s" access="read">
       <annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="const" />
     </property>
+    <property name="VolumeKeyLoaded" type="v" access="read">
+      <annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="false" />
+    </property>
   </interface>
 """,
 }

src/stratis_cli/_actions/_list_pool.py

@@ -19,11 +19,17 @@
 import json
 import os
 from abc import ABC, abstractmethod
+from typing import List, Optional, Union
 
 # isort: THIRDPARTY
 from justbytes import Range
 
-from .._error_codes import PoolAllocSpaceErrorCode, PoolDeviceSizeChangeCode
+from .._error_codes import (
+    PoolAllocSpaceErrorCode,
+    PoolDeviceSizeChangeCode,
+    PoolEncryptionErrorCode,
+    PoolErrorCodeType,
+)
 from .._errors import StratisCliResourceNotFoundError
 from .._stratisd_constants import MetadataVersion, PoolActionAvailability
 from ._connection import get_object
@@ -44,6 +50,24 @@
 )
 
 
+def _metadata_version(mopool) -> Optional[MetadataVersion]:
+    try:
+        return MetadataVersion(int(mopool.MetadataVersion()))
+    except ValueError:  # pragma: no cover
+        return None
+
+
+def _volume_key_loaded(mopool) -> Union[tuple[bool, bool], tuple[bool, str]]:
+    """
+    The string result is an error message indicating that the volume key
+    state is unknown.
+    """
+    result = mopool.VolumeKeyLoaded()
+    if isinstance(result, int):
+        return (True, bool(result))
+    return (False, str(result))  # pragma: no cover
+
+
 # This method is only used with legacy pools
 def _non_existent_or_inconsistent_to_str(
     value,
@@ -108,60 +132,20 @@ def __str__(self):
         )
 
 
-def list_pools(uuid_formatter, *, stopped=False, selection=None):
-    """
-    List the specified information about pools.
-    :param uuid_formatter: how to format UUIDs
-    :type uuid_formatter: (str or UUID) -> str
-    :param bool stopped: True if stopped pools should be listed, else False
-    :param PoolId selection: how to select pools to list
-    """
-    if stopped:
-        if selection is None:
-            klass = StoppedTable(uuid_formatter)
-        else:
-            klass = StoppedDetail(uuid_formatter, selection)
-    else:
-        if selection is None:
-            klass = DefaultTable(uuid_formatter)
-        else:
-            klass = DefaultDetail(uuid_formatter, selection)
-
-    klass.display()
-
-
-def _clevis_to_str(clevis_info):  # pragma: no cover
-    """
-    :param ClevisInfo clevis_info: the Clevis info to stringify
-    :return: a string that represents the clevis info
-    :rtype: str
-    """
-
-    config_string = " ".join(
-        f"{key}: {value}" for key, value in clevis_info.config.items()
-    )
-    return f"{clevis_info.pin}   {config_string}"
-
-
-class List(ABC):  # pylint:disable=too-few-public-methods
+class DefaultAlerts:  # pylint: disable=too-few-public-methods
     """
-    Handle listing a pool or pools.
+    Alerts to display for a started pool.
     """
 
-    @abstractmethod
-    def display(self):
+    def __init__(self, devs):
         """
-        List the pools.
-        """
-
+        The initializer.
 
-class Default(List):
-    """
-    Handle listing the pools that are listed by default.
-    """
+        :param devs: result of GetManagedObjects
+        """
+        (self.increased, self.decreased) = DefaultAlerts._pools_with_changed_devs(devs)
 
-    @staticmethod
-    def alert_codes(mopool):
+    def alert_codes(self, pool_object_path, mopool) -> List[PoolErrorCodeType]:
         """
         Return error code objects for a pool.
 
@@ -176,7 +160,35 @@ def alert_codes(mopool):
             [PoolAllocSpaceErrorCode.NO_ALLOC_SPACE] if mopool.NoAllocSpace() else []
         )
 
-        return availability_error_codes + no_alloc_space_error_codes
+        device_size_changed_codes = DefaultAlerts._from_sets(
+            pool_object_path, self.increased, self.decreased
+        )
+
+        metadata_version = _metadata_version(mopool)
+
+        (vkl_is_bool, volume_key_loaded) = _volume_key_loaded(mopool)
+
+        pool_encryption_error_codes = (
+            [PoolEncryptionErrorCode.VOLUME_KEY_NOT_LOADED]
+            if metadata_version is MetadataVersion.V2
+            and mopool.Encrypted()
+            and vkl_is_bool
+            and not volume_key_loaded
+            else []
+        ) + (
+            [PoolEncryptionErrorCode.VOLUME_KEY_STATUS_UNKNOWN]
+            if metadata_version is MetadataVersion.V2
+            and mopool.Encrypted()
+            and not vkl_is_bool
+            else []
+        )
+
+        return (
+            availability_error_codes
+            + no_alloc_space_error_codes
+            + device_size_changed_codes
+            + pool_encryption_error_codes
+        )
 
     @staticmethod
     def _pools_with_changed_devs(devs_to_search):
@@ -208,7 +220,9 @@ def _pools_with_changed_devs(devs_to_search):
         return (increased, decreased)
 
     @staticmethod
-    def _from_sets(pool_object_path, increased, decreased):
+    def _from_sets(
+        pool_object_path, increased, decreased
+    ) -> List[PoolDeviceSizeChangeCode]:
         """
         Get the code from sets and one pool object path.
 
@@ -234,7 +248,60 @@ def _from_sets(pool_object_path, increased, decreased):
         return []
 
 
-class DefaultDetail(Default):
+def list_pools(uuid_formatter, *, stopped=False, selection=None):
+    """
+    List the specified information about pools.
+    :param uuid_formatter: how to format UUIDs
+    :type uuid_formatter: (str or UUID) -> str
+    :param bool stopped: True if stopped pools should be listed, else False
+    :param PoolId selection: how to select pools to list
+    """
+    if stopped:
+        if selection is None:
+            klass = StoppedTable(uuid_formatter)
+        else:
+            klass = StoppedDetail(uuid_formatter, selection)
+    else:
+        if selection is None:
+            klass = DefaultTable(uuid_formatter)
+        else:
+            klass = DefaultDetail(uuid_formatter, selection)
+
+    klass.display()
+
+
+def _clevis_to_str(clevis_info):  # pragma: no cover
+    """
+    :param ClevisInfo clevis_info: the Clevis info to stringify
+    :return: a string that represents the clevis info
+    :rtype: str
+    """
+
+    config_string = " ".join(
+        f"{key}: {value}" for key, value in clevis_info.config.items()
+    )
+    return f"{clevis_info.pin}   {config_string}"
+
+
+class ListPool(ABC):  # pylint:disable=too-few-public-methods
+    """
+    Handle listing a pool or pools.
+    """
+
+    @abstractmethod
+    def display(self):
+        """
+        List the pools.
+        """
+
+
+class Default(ListPool):  # pylint: disable=too-few-public-methods
+    """
+    Handle listing the pools that are listed by default.
+    """
+
+
+class DefaultDetail(Default):  # pylint: disable=too-few-public-methods
     """
     List one pool with a detail view.
     """
@@ -249,14 +316,16 @@ def __init__(self, uuid_formatter, selection):
         self.uuid_formatter = uuid_formatter
         self.selection = selection
 
-    def _print_detail_view(self, mopool, size_change_codes):
+    def _print_detail_view(
+        self, pool_object_path, mopool, alerts: DefaultAlerts
+    ):  # pylint: disable=too-many-locals
         """
         Print the detailed view for a single pool.
 
         :param UUID uuid: the pool uuid
+        :param pool_object_path: object path of the pool
         :param MOPool mopool: properties of the pool
-        :param size_change_codes: size change codes
-        :type size_change_codes: list of PoolDeviceSizeChangeCode
+        :param DefaultAlerts alerts: pool alerts
         """
         encrypted = mopool.Encrypted()
 
@@ -265,16 +334,13 @@ def _print_detail_view(self, mopool, size_change_codes):
 
         alert_summary = [
             f"{code}: {code.summarize()}"
-            for code in (self.alert_codes(mopool) + size_change_codes)
+            for code in alerts.alert_codes(pool_object_path, mopool)
         ]
         print(f"Alerts: {len(alert_summary)}")
         for line in alert_summary:  # pragma: no cover
             print(f"     {line}")
 
-        try:
-            metadata_version = MetadataVersion(int(mopool.MetadataVersion()))
-        except ValueError:  # pragma: no cover
-            metadata_version = None
+        metadata_version = _metadata_version(mopool)
 
         print(
             f'Metadata Version: {"none" if metadata_version is None else metadata_version}'
@@ -363,16 +429,14 @@ def display(self):
             .search(managed_objects)
         )
 
-        (increased, decreased) = self._pools_with_changed_devs(
+        alerts = DefaultAlerts(
             devs(props={"Pool": pool_object_path}).search(managed_objects)
         )
 
-        device_change_codes = self._from_sets(pool_object_path, increased, decreased)
+        self._print_detail_view(pool_object_path, MOPool(mopool), alerts)
 
-        self._print_detail_view(MOPool(mopool), device_change_codes)
 
-
-class DefaultTable(Default):
+class DefaultTable(Default):  # pylint: disable=too-few-public-methods
     """
     List several pools with a table view.
     """
@@ -432,10 +496,7 @@ def gen_string(has_property, code):
                 """
                 return (" " if has_property else "~") + code
 
-            try:
-                metadata_version = MetadataVersion(int(mopool.MetadataVersion()))
-            except ValueError:  # pragma: no cover
-                metadata_version = None
+            metadata_version = _metadata_version(mopool)
 
             props_list = [
                 (metadata_version in (MetadataVersion.V1, None), "Le"),
@@ -447,9 +508,7 @@ def gen_string(has_property, code):
 
         managed_objects = ObjectManager.Methods.GetManagedObjects(proxy, {})
 
-        (increased, decreased) = self._pools_with_changed_devs(
-            devs().search(managed_objects)
-        )
+        alerts = DefaultAlerts(devs().search(managed_objects))
 
         pools_with_props = [
             (objpath, MOPool(info)) for objpath, info in pools().search(managed_objects)
@@ -464,10 +523,7 @@ def gen_string(has_property, code):
                 ", ".join(
                     sorted(
                         str(code)
-                        for code in (
-                            self.alert_codes(mopool)
-                            + self._from_sets(pool_object_path, increased, decreased)
-                        )
+                        for code in alerts.alert_codes(pool_object_path, mopool)
                     )
                 ),
             )
@@ -487,7 +543,7 @@ def gen_string(has_property, code):
         )
 
 
-class Stopped(List):  # pylint: disable=too-few-public-methods
+class Stopped(ListPool):  # pylint: disable=too-few-public-methods
     """
     Support for listing stopped pools.
     """

src/stratis_cli/_error_codes.py

@@ -159,14 +159,61 @@ def summarize(self) -> str:
         assert False, "impossible error code reached"  # pragma: no cover
 
 
+class PoolEncryptionErrorCode(IntEnum):
+    """
+    Codes for encryption problems.
+    """
+
+    VOLUME_KEY_NOT_LOADED = 1
+    VOLUME_KEY_STATUS_UNKNOWN = 2
+
+    def __str__(self) -> str:
+        return f"{Level.WARNING}C{str(self.value).zfill(3)}"
+
+    def explain(self) -> str:
+        """
+        Return an explanation of the return code.
+        """
+        if self is PoolEncryptionErrorCode.VOLUME_KEY_NOT_LOADED:
+            return (
+                "The pool's volume key is not loaded. This may result in an "
+                "error if the pool's encryption layer needs to be modified. "
+                "If the pool is encrypted with a key in the kernel keyring "
+                "then setting that key may resolve the problem."
+            )
+        if self is PoolEncryptionErrorCode.VOLUME_KEY_STATUS_UNKNOWN:
+            return (
+                "The pool's volume key may or may not be loaded. If the volume "
+                "key is not loaded, there may an error if the pool's "
+                "encryption layer needs to be modified."
+            )
+
+        assert False, "impossible error code reached"  # pragma: no cover
+
+    def summarize(self) -> str:
+        """
+        Return a short summary of the return code.
+        """
+        if self is PoolEncryptionErrorCode.VOLUME_KEY_NOT_LOADED:
+            return "Volume key not loaded"
+        if self is PoolEncryptionErrorCode.VOLUME_KEY_STATUS_UNKNOWN:
+            return "Volume key status unknown"
+
+        assert False, "impossible error code reached"  # pragma: no cover
+
+
 CLASSES = [
     PoolAllocSpaceErrorCode,
     PoolDeviceSizeChangeCode,
+    PoolEncryptionErrorCode,
     PoolMaintenanceErrorCode,
 ]
 
 type PoolErrorCodeType = Union[
-    PoolAllocSpaceErrorCode, PoolDeviceSizeChangeCode, PoolMaintenanceErrorCode
+    PoolAllocSpaceErrorCode,
+    PoolDeviceSizeChangeCode,
+    PoolEncryptionErrorCode,
+    PoolMaintenanceErrorCode,
 ]