Storage mangment API: disks infos

debian/control

@@ -17,6 +17,7 @@ Depends: python3-all (>= 3.11),
  , python3-ldap, python3-zeroconf (>= 0.47), python3-lexicon,
  , python3-cryptography, python3-jwt, python3-passlib, python3-magic
  , python-is-python3, python3-pydantic, python3-email-validator
+ , udisks2,
  , nginx, nginx-extras (>=1.22)
  , apt, apt-transport-https, apt-utils, aptitude, dirmngr
  , openssh-server, iptables, fail2ban, bind9-dnsutils

share/actionsmap.yml

@@ -2000,3 +2000,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,54 @@
+import dataclasses
+from typing import Optional
+
+import dbus
+from moulinette.utils.log import getActionLogger
+
+logger = getActionLogger("yunohost.storage")
+
+
+UDISK_DRIVE_PATH = "/org/freedesktop/UDisks2/drives/"
+UDISK_BLOCK_PATH = "/org/freedesktop/UDisks2/block_devices/"
+UDISK_PART_TABLE_IFC = "org.freedesktop.UDisks2.PartitionTable"
+UDISK_BLOCK_IFC = "org.freedesktop.UDisks2.Block"
+UDISK_DRIVE_IFC = "org.freedesktop.UDisks2.Drive"
+UDISK_ENCRYPTED_IFC = "org.freedesktop.UDisks2.Encrypted"
+UDISK_FILESYSTEM_IFC = "org.freedesktop.UDisks2.Filesystem"
+
+
+@dataclasses.dataclass
+class DiskInfos:
+    name: str
+    model: str
+    serial: str
+    size: int
+    type: str
+    rpm: Optional[int] = None
+
+
+def infos():
+    result = {}
+
+    bus = dbus.SystemBus()
+    manager = bus.get_object("org.freedesktop.UDisks2", "/org/freedesktop/UDisks2")
+
+    for k, v in manager.get_dbus_method(
+        "GetManagedObjects", "org.freedesktop.DBus.ObjectManager"
+    )().items():
+        if k.startswith(UDISK_DRIVE_PATH):
+            # These are hard drives
+            drive = v[UDISK_DRIVE_IFC]
+            name = k.removeprefix(UDISK_DRIVE_PATH)
+            rotation_rate = drive["RotationRate"]
+            result[name] = dataclasses.asdict(
+                DiskInfos(
+                    name=name,
+                    model=drive["Model"],
+                    serial=drive["Serial"],
+                    size=drive["Size"],
+                    type="HDD" if rotation_rate != 0 else "SSD",
+                    rpm=rotation_rate if rotation_rate != 0 else None,
+                )
+            )
+
+    return result

src/storage.py

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