Storage mangment API: disks infos

debian/control

@@ -15,6 +15,7 @@ Depends: ${python3:Depends}, ${misc:Depends}
  , python3-miniupnpc, python3-dbus, python3-jinja2
  , python3-toml, python3-packaging, python3-publicsuffix2
  , python3-ldap, python3-zeroconf (>= 0.36), python3-lexicon,
+ , python3-pyudev
  , python-is-python3
  , nginx, nginx-extras (>=1.18)
  , apt, apt-transport-https, apt-utils, aptitude, dirmngr

share/actionsmap.yml

@@ -2082,3 +2082,18 @@ diagnosis:
                     help: Remove a filter (it should be an existing filter as listed with "ignore --list")
                     nargs: "*"
                     metavar: CRITERIA
+
+
+#############################
+#           Storage         #
+#############################
+storage:
+  category_help: Manage hard-drives, filesystem, pools
+  subcategories:
+    disk:
+      subcategory_help: Manage et get infos about hard-drives
+      actions:
+        # storage_disks_list
+        infos:
+          action_help: Gets infos about hard-drives currently attached to this system
+          api: GET /storage/disk/infos

src/disks.py

@@ -0,0 +1,101 @@
+from collections import OrderedDict
+import dataclasses
+from glob import glob
+from typing import Optional
+
+import pyudev
+import psutil
+
+from moulinette.utils.log import getActionLogger
+
+
+from yunohost.utils.disks import filter_device
+
+
+logger = getActionLogger("yunohost.storage")
+
+
+@dataclasses.dataclass
+class DiskParts:
+    devname: str
+    filesystem: str
+    encrypted: bool
+    mountpoint: str
+
+    @staticmethod
+    def from_parent_device(device: pyudev.Device, partitions):
+        result = OrderedDict()
+        for child_dev in sorted(
+            filter(filter_device, device.children), key=lambda it: it.device_node
+        ):
+            encrypted_provider = glob(f"/sys/block/dm-*/slaves/{child_dev.sys_name}")
+            if encrypted_provider:
+                # retrive the dm-x part
+                dm = encrypted_provider[0].split("/")[3]
+                enc_dev = pyudev.Devices.from_name(device.context, "block", dm)
+                # This work for LUKS, what about other partition mecanisms?
+                partname = f"/dev/mapper/{enc_dev.properties['DM_NAME']}"
+                encrypted = True
+            else:
+                partname = child_dev.device_node
+                encrypted = False
+
+            if partname not in partitions:
+                logger.warning(
+                    f"{child_dev.device_node} not found by 'psutil.disk_partitions'"
+                )
+                continue
+
+            result[child_dev.sys_name] = DiskParts(
+                devname=device.device_node,
+                filesystem=partitions[partname].fstype,
+                encrypted=encrypted,
+                mountpoint=partitions[partname].mountpoint,
+            )
+
+        return result
+
+
+@dataclasses.dataclass
+class DiskInfos:
+    devname: str
+    model: str
+    serial: str
+    size: int
+    links: list[str]
+    partitions: Optional[list[DiskParts]]
+
+    @staticmethod
+    def from_device(device, partitions):
+        try:
+            dev_size = device.attributes.asint("size")
+        except (AttributeError, UnicodeError, ValueError):
+            dev_size = None
+
+        dev_links = list(sorted(it for it in device.device_links))
+        child_parts = DiskParts.from_parent_device(device, partitions)
+
+        return DiskInfos(
+            devname=device.device_node,
+            model=device.get("ID_MODEL", None),
+            serial=device.get("ID_SERIAL_SHORT", None),
+            size=dev_size,
+            links=dev_links,
+            partitions=child_parts or None,
+        )
+
+
+def infos():
+    context = pyudev.Context()
+    partitions = {it.device: it for it in psutil.disk_partitions()}
+    result = OrderedDict()
+
+    for it in sorted(
+        filter(filter_device, context.list_devices(subsystem="block", DEVTYPE="disk")),
+        key=lambda it: it.device_node,
+    ):
+        result[it.sys_name] = dataclasses.asdict(
+            DiskInfos.from_device(it, partitions), dict_factory=OrderedDict
+        )
+
+    return result

src/storage.py

@@ -0,0 +1,4 @@
+def storage_disk_infos():
+    from yunohost.disks import infos
+
+    return infos()

src/tests/test_storage.py

@@ -0,0 +1,8 @@
+def setup_function(function):
+    ...
+
+def teardown_function(function):
+    ...
+
+def test_storage_disks_infos():
+    ...

src/utils/disks.py

@@ -0,0 +1,12 @@
+import re
+
+IGNORE_DISKS = "sr", "md", "dm-", "loop", "zd", "pmem"
+# regex: ^((sr)|(md)|...)
+IGNORE_DISK_RE = re.compile(rf"""^({"|".join([f'({it})' for it in IGNORE_DISKS])})""")
+
+
+def filter_device(device):
+    """
+    Returns True if device has parents (e.g. USB device) and its name is not amongst
+    """
+    return device.parent is not None and not IGNORE_DISK_RE.match(device.sys_name)