diff --git a/bin/.local/bin/labfy-removable-media b/bin/.local/bin/labfy-removable-media new file mode 100755 index 000000000..dcb90a1f0 --- /dev/null +++ b/bin/.local/bin/labfy-removable-media @@ -0,0 +1,719 @@ +#!/usr/bin/python3 +"""Daemon d'accès conservateur aux supports amovibles via UDisks2.""" + +from __future__ import annotations + +import json +import logging +import os +import socket +import subprocess +import sys +import tempfile +import threading +import time +from dataclasses import dataclass, replace +from pathlib import Path +from typing import Any, Callable + +SCHEMA = "labfy.removable-media" +VERSION = 1 +MAX_REQUEST = 16 * 1024 +MAX_ERROR = 240 +MAX_CLIENTS = 8 +READ_TIMEOUT = 2.0 +SOCKET_NAME = "labfy-removable-media.sock" +STATE_NAME = "labfy-removable-media.json" +ALLOWED_OPERATIONS = {"list", "mount", "unmount", "open", "safe-remove"} +LOGGER = logging.getLogger(__name__) +UDISKS_DEVICE_BUSY = "org.freedesktop.UDisks2.Error.DeviceBusy" +UDISKS_ERROR_PREFIX = "org.freedesktop.UDisks2.Error." + + +class RequestError(Exception): + """Erreur bornée et présentable au client.""" + + +def bounded_error(error: BaseException | str) -> str: + text = " ".join(str(error).split()) or "Erreur inconnue" + return text[:MAX_ERROR] + + +def public_operation_error(error: BaseException, operation: str) -> str: + """Traduit les identités D-Bus connues sans exposer leur détail technique.""" + if isinstance(error, RequestError): + return bounded_error(error) + detail = str(error) + if UDISKS_DEVICE_BUSY in detail: + return "Le périphérique est utilisé par une application." + if UDISKS_ERROR_PREFIX in detail: + return { + "mount": "Impossible de monter le périphérique.", + "unmount": "Impossible de démonter le périphérique.", + "safe-remove": "Impossible de retirer le périphérique en sécurité.", + }.get(operation, "Impossible d'effectuer l'opération sur le périphérique.") + return bounded_error(error) + + +def runtime_dir() -> Path: + value = os.environ.get("XDG_RUNTIME_DIR") + if not value or not os.path.isabs(value): + raise RuntimeError("XDG_RUNTIME_DIR doit être un chemin absolu") + return Path(value) + + +def decode_udisks_bytes(value: Any) -> str: + """UDisks expose les points de montage en tableaux d'octets terminés par NUL.""" + if hasattr(value, "unpack"): + value = value.unpack() + if isinstance(value, str): + return value.rstrip("\0") + if isinstance(value, (bytes, bytearray)): + return bytes(value).rstrip(b"\0").decode("utf-8", "replace") + if isinstance(value, (list, tuple)): + return bytes(value).rstrip(b"\0").decode("utf-8", "replace") + return "" + + +def unpack(value: Any, default: Any = None) -> Any: + if value is None: + return default + return value.unpack() if hasattr(value, "unpack") else value + + +@dataclass(frozen=True) +class FilesystemRecord: + path: str + drive_path: str + display_name: str + label: str + filesystem: str + size_bytes: int + mount_point: str + removable: bool + ejectable: bool + power_off_capable: bool + + def public(self, busy: bool = False, error: str = "") -> dict[str, Any]: + return { + "runtime_id": self.path, + "drive_runtime_id": self.drive_path, + "display_name": self.display_name, + "label": self.label, + "filesystem": self.filesystem, + "size_bytes": self.size_bytes, + "mount_point": self.mount_point, + "mounted": bool(self.mount_point), + "removable": self.removable, + "ejectable": self.ejectable, + "power_off_capable": self.power_off_capable, + "busy": busy, + "error": error, + } + + +class UDisksBackend: + """Adaptateur Gio minimal ; l'ObjectManager reste l'unique source de vérité.""" + + BUS_NAME = "org.freedesktop.UDisks2" + ROOT = "/org/freedesktop/UDisks2" + IFACE_DRIVE = "org.freedesktop.UDisks2.Drive" + IFACE_BLOCK = "org.freedesktop.UDisks2.Block" + IFACE_FILESYSTEM = "org.freedesktop.UDisks2.Filesystem" + + def __init__(self, Gio: Any, GLib: Any) -> None: + self.Gio = Gio + self.GLib = GLib + self.bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None) + self.manager = Gio.DBusObjectManagerClient.new_sync( + self.bus, + Gio.DBusObjectManagerClientFlags.NONE, + self.BUS_NAME, + self.ROOT, + None, + None, + None, + ) + + @staticmethod + def _prop(interface: Any, name: str, default: Any = None) -> Any: + return unpack(interface.get_cached_property(name), default) + + def _interfaces(self) -> dict[str, dict[str, Any]]: + result: dict[str, dict[str, Any]] = {} + for obj in self.manager.get_objects(): + result[obj.get_object_path()] = { + iface.get_interface_name(): iface for iface in obj.get_interfaces() + } + return result + + def records(self) -> list[FilesystemRecord]: + objects = self._interfaces() + drives: dict[str, dict[str, Any]] = {} + for path, interfaces in objects.items(): + drive = interfaces.get(self.IFACE_DRIVE) + if drive is None: + continue + bus = str(self._prop(drive, "ConnectionBus", "")).lower() + compatibility = { + str(item).lower() + for item in self._prop(drive, "MediaCompatibility", []) + } + is_sd = any("sd" in item or "mmc" in item for item in compatibility) + # CONTRACT: une propriété Removable seule est trop large (disques internes, + # lecteurs optiques). Seuls USB et cartes SD/MMC sont auto-présentés. + if bus != "usb" and not is_sd: + continue + drives[path] = { + "display": str(self._prop(drive, "Model", "")).strip() + or str(self._prop(drive, "Vendor", "")).strip() + or "Support amovible", + "removable": bool(self._prop(drive, "Removable", False)), + "ejectable": bool(self._prop(drive, "Ejectable", False)), + "power": bool(self._prop(drive, "CanPowerOff", False)), + } + + records: list[FilesystemRecord] = [] + for path, interfaces in objects.items(): + block = interfaces.get(self.IFACE_BLOCK) + filesystem = interfaces.get(self.IFACE_FILESYSTEM) + if block is None or filesystem is None: + continue + drive_path = str(self._prop(block, "Drive", "/")) + drive = drives.get(drive_path) + if drive is None: + continue + # CONTRACT: UDisks réserve HintIgnore/HintSystem aux blocs que les + # interfaces utilisateur ne doivent jamais proposer (dont l'EFI). + if bool(self._prop(block, "HintIgnore", False)) or bool( + self._prop(block, "HintSystem", False) + ): + continue + usage = str(self._prop(block, "IdUsage", "")) + fs_type = str(self._prop(block, "IdType", "")) + # INVARIANT: pas d'auto-unlock LUKS, pas de périphérique sans filesystem. + if usage != "filesystem" or fs_type in {"crypto_LUKS", ""}: + continue + points = self._prop(filesystem, "MountPoints", []) or [] + mount_point = decode_udisks_bytes(points[0]) if points else "" + label = str(self._prop(block, "IdLabel", "")).strip() + records.append( + FilesystemRecord( + path=path, + drive_path=drive_path, + display_name=label or drive["display"], + label=label, + filesystem=fs_type, + size_bytes=int(self._prop(block, "Size", 0)), + mount_point=mount_point, + removable=drive["removable"], + ejectable=drive["ejectable"], + power_off_capable=drive["power"], + ) + ) + return sorted(records, key=lambda item: (item.drive_path, item.path)) + + def connect_changed(self, callback: Callable[[], None]) -> None: + self.manager.connect("object-added", lambda *_args: callback()) + self.manager.connect("object-removed", lambda *_args: callback()) + self.manager.connect( + "interface-proxy-properties-changed", lambda *_args: callback() + ) + + def _call(self, path: str, interface: str, method: str) -> Any: + proxy = self.Gio.DBusProxy.new_sync( + self.bus, + self.Gio.DBusProxyFlags.NONE, + None, + self.BUS_NAME, + path, + interface, + None, + ) + parameters = self.GLib.Variant("(a{sv})", ({},)) + return proxy.call_sync( + method, + parameters, + self.Gio.DBusCallFlags.NONE, + 15_000, + None, + ) + + def mount(self, path: str) -> list[Any]: + self._call(path, self.IFACE_FILESYSTEM, "Mount") + points = self.mount_points_sync(path) + if not points: + raise RequestError("Le volume n'est pas monté après la demande") + return points + + def unmount(self, path: str) -> list[Any]: + self._call(path, self.IFACE_FILESYSTEM, "Unmount") + points = self.mount_points_sync(path) + if points: + raise RequestError("Le volume reste monté après la demande") + return points + + def _managed_objects_sync(self) -> dict[str, dict[str, Any]]: + reply = self.bus.call_sync( + self.BUS_NAME, + self.ROOT, + "org.freedesktop.DBus.ObjectManager", + "GetManagedObjects", + None, + None, + self.Gio.DBusCallFlags.NONE, + 15_000, + None, + ) + return unpack(reply)[0] + + def mount_points_sync(self, path: str) -> list[Any]: + interfaces = self._managed_objects_sync().get(path, {}) + filesystem = interfaces.get(self.IFACE_FILESYSTEM) + if filesystem is None: + raise RequestError("Volume introuvable après la demande") + filesystem = unpack(filesystem, {}) + return list(unpack(filesystem.get("MountPoints"), []) or []) + + def safe_remove(self, drive: FilesystemRecord) -> None: + # INVARIANT: la portée de l'éjection est le lecteur entier. Il faut donc + # considérer aussi les filesystems masqués de l'interface utilisateur. + for path, mount_points in reversed(self._drive_filesystems(drive.drive_path)): + if mount_points: + self.unmount(path) + remaining = [ + path + for path, mount_points in self._drive_filesystems_sync(drive.drive_path) + if mount_points + ] + if remaining: + raise RequestError("Éjection refusée : un volume du lecteur reste monté") + method = "PowerOff" if drive.power_off_capable else "Eject" + if method == "Eject" and not drive.ejectable: + raise RequestError("Ce support ne peut être ni éjecté ni mis hors tension") + self._call(drive.drive_path, self.IFACE_DRIVE, method) + + def _drive_filesystems(self, drive_path: str) -> list[tuple[str, list[Any]]]: + result: list[tuple[str, list[Any]]] = [] + for path, interfaces in self._interfaces().items(): + block = interfaces.get(self.IFACE_BLOCK) + filesystem = interfaces.get(self.IFACE_FILESYSTEM) + if block is None or filesystem is None: + continue + if str(self._prop(block, "Drive", "/")) != drive_path: + continue + points = list(self._prop(filesystem, "MountPoints", []) or []) + result.append((path, points)) + return sorted(result) + + def _drive_filesystems_sync(self, drive_path: str) -> list[tuple[str, list[Any]]]: + # WHY: call_sync(Unmount) peut revenir avant que PropertiesChanged ait + # actualisé le cache du client ObjectManager sur ce même thread GLib. + managed = self._managed_objects_sync() + result: list[tuple[str, list[Any]]] = [] + for path, interfaces in managed.items(): + block = interfaces.get(self.IFACE_BLOCK) + filesystem = interfaces.get(self.IFACE_FILESYSTEM) + if block is None or filesystem is None: + continue + block = unpack(block, {}) + filesystem = unpack(filesystem, {}) + if str(unpack(block.get("Drive"), "/")) != drive_path: + continue + points = list(unpack(filesystem.get("MountPoints"), []) or []) + result.append((path, points)) + return sorted(result) + + +class MediaDaemon: + """Modèle métier injectable et indépendant de Gio.""" + + def __init__( + self, + backend: Any, + state_path: Path, + launcher: Callable[[str], None] | None = None, + ) -> None: + self.backend = backend + self.state_path = state_path + self.launcher = launcher or self._launch_yazi + self._records: list[FilesystemRecord] = [] + self._busy_drives: set[str] = set() + self._errors: dict[str, str] = {} + self._automount_seen: set[str] = set() + self._lock = threading.RLock() + self._updated_at = int(time.time()) + self.refresh() + + def state(self) -> dict[str, Any]: + with self._lock: + return { + "schema": SCHEMA, + "version": VERSION, + "devices": [ + record.public( + record.drive_path in self._busy_drives, + self._errors.get(record.path, ""), + ) + for record in self._records + ], + "updated_at": self._updated_at, + } + + def _write_state(self) -> None: + self._updated_at = int(time.time()) + payload = json.dumps( + self.state(), ensure_ascii=False, separators=(",", ":") + ).encode("utf-8") + b"\n" + self.state_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + fd, temporary = tempfile.mkstemp( + prefix=f".{self.state_path.name}.", dir=self.state_path.parent + ) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "wb") as handle: + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, self.state_path) + os.chmod(self.state_path, 0o600) + finally: + try: + os.unlink(temporary) + except FileNotFoundError: + pass + + def refresh(self) -> None: + with self._lock: + self._records = self.backend.records() + live = {item.path for item in self._records} + self._errors = {key: value for key, value in self._errors.items() if key in live} + self._write_state() + + def start(self) -> None: + """Effectue le scan initial après l'abonnement aux événements UDisks.""" + self.devices_changed() + + def devices_changed(self) -> None: + with self._lock: + self.refresh() + live = {item.path for item in self._records} + # INVARIANT: une disparition termine la vie de l'objet ; une même + # object path réinsérée peut alors recevoir une nouvelle tentative. + self._automount_seen.intersection_update(live) + newcomers = [item for item in self._records if item.path not in self._automount_seen] + self._automount_seen.update(item.path for item in newcomers) + + confirmed: dict[str, str] = {} + for item in newcomers: + if item.mount_point: + continue + try: + self.backend.mount(item.path) + points = self.backend.mount_points_sync(item.path) + if not points: + raise RequestError("Le volume n'est pas monté après la demande") + confirmed[item.path] = decode_udisks_bytes(points[0]) + self._errors.pop(item.path, None) + except Exception as error: + # CONTRACT: l'échec d'un volume ne bloque ni les autres + # volumes ni une future réinsertion physique de celui-ci. + LOGGER.warning("Échec UDisks pendant le montage automatique : %s", error) + self._errors[item.path] = public_operation_error(error, "mount") + self.refresh() + if confirmed: + self._records = [ + replace(item, mount_point=confirmed[item.path]) + if item.path in confirmed + else item + for item in self._records + ] + self._write_state() + + def _record(self, runtime_id: str) -> FilesystemRecord: + # CONTRACT: on relit UDisks avant toute mutation ; un identifiant d'objet + # disparu ou réutilisé ne doit jamais viser un ancien support en cache. + self.refresh() + for record in self._records: + if record.path == runtime_id: + return record + raise RequestError("Support introuvable ou retiré") + + def _drive_records(self, drive_path: str) -> list[FilesystemRecord]: + return [item for item in self._records if item.drive_path == drive_path] + + def _set_busy(self, drive_path: str, value: bool) -> None: + if value: + if drive_path in self._busy_drives: + raise RequestError("Une opération est déjà en cours sur ce support") + self._busy_drives.add(drive_path) + else: + self._busy_drives.discard(drive_path) + self._write_state() + + def _run_drive_operation( + self, record: FilesystemRecord, operation: Callable[[list[FilesystemRecord]], None] + ) -> None: + self._set_busy(record.drive_path, True) + try: + operation(self._drive_records(record.drive_path)) + for item in self._drive_records(record.drive_path): + self._errors.pop(item.path, None) + except Exception as error: + LOGGER.warning("Échec UDisks pendant une opération lecteur : %s", error) + message = public_operation_error(error, "safe-remove") + for item in self._drive_records(record.drive_path): + self._errors[item.path] = message + raise RequestError(message) from None + finally: + self._set_busy(record.drive_path, False) + self.refresh() + + def _run_volume_operation( + self, + record: FilesystemRecord, + operation_name: str, + operation: Callable[[], list[Any]], + ) -> list[Any]: + points: list[Any] | None = None + self._set_busy(record.drive_path, True) + try: + points = operation() + self._errors.pop(record.path, None) + except Exception as error: + LOGGER.warning("Échec UDisks pendant une opération volume : %s", error) + message = public_operation_error(error, operation_name) + self._errors[record.path] = message + raise RequestError(message) from None + finally: + self._set_busy(record.drive_path, False) + self.refresh() + if points is not None: + mount_point = decode_udisks_bytes(points[0]) if points else "" + # CONTRACT: publier immédiatement l'état relu auprès d'UDisks, + # même si le cache ObjectManager attend encore PropertiesChanged. + self._records = [ + replace(item, mount_point=mount_point) + if item.path == record.path + else item + for item in self._records + ] + self._write_state() + assert points is not None + return points + + def _mount(self, item: FilesystemRecord) -> list[Any]: + if not item.mount_point: + self.backend.mount(item.path) + points = self.backend.mount_points_sync(item.path) + if not points: + raise RequestError("Le volume n'est pas monté après la demande") + return points + + def _unmount(self, item: FilesystemRecord) -> list[Any]: + if item.mount_point: + self.backend.unmount(item.path) + points = self.backend.mount_points_sync(item.path) + if points: + raise RequestError("Le volume reste monté après la demande") + return points + + def handle(self, request: dict[str, Any]) -> dict[str, Any]: + operation, runtime_id = validate_request(request) + if operation == "list": + self.refresh() + return {"ok": True, "state": self.state()} + assert runtime_id is not None + if operation == "safe-remove": + self.refresh() + records = self._drive_records(runtime_id) + if not records: + raise RequestError("Lecteur introuvable ou retiré") + record = records[0] + + def remove(_all_records: list[FilesystemRecord]) -> None: + self.backend.safe_remove(record) + + self._run_drive_operation(record, remove) + else: + record = self._record(runtime_id) + if operation == "mount": + self._run_volume_operation(record, operation, lambda: self._mount(record)) + elif operation == "unmount": + self._run_volume_operation(record, operation, lambda: self._unmount(record)) + elif operation == "open": + if not record.mount_point: + points = self._run_volume_operation(record, "mount", lambda: self._mount(record)) + mount_point = decode_udisks_bytes(points[0]) if points else "" + else: + mount_point = record.mount_point + if not mount_point: + raise RequestError("Le support n'a pas fourni de point de montage") + self.launcher(mount_point) + return {"ok": True, "state": self.state()} + + @staticmethod + def _launch_yazi(mount_point: str) -> None: + subprocess.Popen( + [ + "/usr/bin/uwsm", + "app", + "--", + "/usr/bin/kitty", + "-e", + "/usr/bin/yazi", + mount_point, + ], + start_new_session=True, + close_fds=True, + ) + + +def validate_request(request: Any) -> tuple[str, str | None]: + if not isinstance(request, dict) or set(request) - {"operation", "runtime_id"}: + raise RequestError("Requête JSON invalide") + operation = request.get("operation") + if operation not in ALLOWED_OPERATIONS: + raise RequestError("Opération non autorisée") + runtime_id = request.get("runtime_id") + if operation == "list": + if runtime_id is not None: + raise RequestError("runtime_id est interdit pour list") + return operation, None + if not isinstance(runtime_id, str) or not runtime_id.startswith( + "/org/freedesktop/UDisks2/" + ) or len(runtime_id) > 512: + raise RequestError("runtime_id invalide") + return operation, runtime_id + + +class JsonSocketServer: + def __init__(self, path: Path, handler: Callable[[dict[str, Any]], dict[str, Any]]) -> None: + self.path = path + self.handler = handler + self.socket: socket.socket | None = None + self._clients = threading.BoundedSemaphore(MAX_CLIENTS) + + def start(self) -> None: + try: + self.path.unlink() + except FileNotFoundError: + pass + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(str(self.path)) + os.chmod(self.path, 0o600) + server.listen(8) + self.socket = server + + def serve_forever(self) -> None: + assert self.socket is not None + while True: + connection, _ = self.socket.accept() + if not self._clients.acquire(blocking=False): + connection.close() + continue + threading.Thread( + target=self._serve_client, args=(connection,), daemon=True + ).start() + + def _serve_client(self, connection: socket.socket) -> None: + try: + self._serve_connection(connection) + finally: + self._clients.release() + + def _serve_connection(self, connection: socket.socket) -> None: + response: dict[str, Any] + try: + # WHY: un client incomplet ne doit pas immobiliser une capacité du + # serveur indéfiniment. + connection.settimeout(READ_TIMEOUT) + data = bytearray() + while b"\n" not in data: + remaining = MAX_REQUEST + 1 - len(data) + if remaining <= 0: + raise RequestError("Requête trop volumineuse") + chunk = connection.recv(min(4096, remaining)) + if not chunk: + break + data.extend(chunk) + if len(data) > MAX_REQUEST: + raise RequestError("Requête trop volumineuse") + if not data or b"\n" not in data or data.partition(b"\n")[2]: + raise RequestError("Une seule requête JSON terminée par LF est requise") + request = json.loads(data[:-1].decode("utf-8")) + response = self.handler(request) + except socket.timeout: + response = {"ok": False, "error": "Délai de lecture dépassé"} + except (RequestError, UnicodeError, json.JSONDecodeError) as error: + response = {"ok": False, "error": bounded_error(error)} + except Exception as error: + response = {"ok": False, "error": bounded_error(error)} + try: + connection.sendall( + json.dumps(response, ensure_ascii=False, separators=(",", ":")).encode() + + b"\n" + ) + except OSError: + pass + finally: + connection.close() + + +class GLibDispatcher: + """Fait exécuter les mutations Gio dans le thread de la boucle GLib.""" + + def __init__(self, GLib: Any, handler: Callable[[dict[str, Any]], dict[str, Any]]) -> None: + self.GLib = GLib + self.handler = handler + + def __call__(self, request: dict[str, Any]) -> dict[str, Any]: + completed = threading.Event() + result: list[dict[str, Any] | BaseException] = [] + + def invoke() -> bool: + try: + result.append(self.handler(request)) + except BaseException as error: + result.append(error) + finally: + completed.set() + return False + + self.GLib.idle_add(invoke) + if not completed.wait(30): + raise RequestError("Délai d'opération dépassé") + value = result[0] + if isinstance(value, BaseException): + raise value + return value + + +def main() -> int: + try: + import gi + + gi.require_version("Gio", "2.0") + from gi.repository import Gio, GLib + + base = runtime_dir() + backend = UDisksBackend(Gio, GLib) + daemon = MediaDaemon(backend, base / STATE_NAME) + # CONTRACT: s'abonner avant le scan ferme la fenêtre où un objet ajouté + # entre ces deux opérations aurait échappé à l'automount. + backend.connect_changed(daemon.devices_changed) + daemon.start() + server = JsonSocketServer(base / SOCKET_NAME, GLibDispatcher(GLib, daemon.handle)) + server.start() + threading.Thread(target=server.serve_forever, daemon=True).start() + GLib.MainLoop().run() + except KeyboardInterrupt: + return 0 + except Exception as error: + print(f"labfy-removable-media: {bounded_error(error)}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/bin/.local/bin/labfy-removable-mediactl b/bin/.local/bin/labfy-removable-mediactl new file mode 100755 index 000000000..f2c6642fe --- /dev/null +++ b/bin/.local/bin/labfy-removable-mediactl @@ -0,0 +1,122 @@ +#!/usr/bin/python3 +"""Client du daemon labfy-removable-media.""" + +from __future__ import annotations + +import json +import os +import socket +import subprocess +import sys +import time +from pathlib import Path + +MAX_RESPONSE = 1024 * 1024 +SOCKET_NAME = "labfy-removable-media.sock" +OPERATIONS = {"list", "mount", "unmount", "open", "safe-remove"} + + +def usage() -> int: + print( + "Usage: labfy-removable-mediactl list\n" + " labfy-removable-mediactl mount \n" + " labfy-removable-mediactl unmount \n" + " labfy-removable-mediactl open \n" + " labfy-removable-mediactl safe-remove ", + file=sys.stderr, + ) + return 2 + + +def socket_path() -> Path: + value = os.environ.get("XDG_RUNTIME_DIR") + if not value or not os.path.isabs(value): + raise RuntimeError("XDG_RUNTIME_DIR doit être un chemin absolu") + return Path(value) / SOCKET_NAME + + +def activate_service() -> None: + """Active l'unité sans shell par l'API D-Bus publique de systemd.""" + subprocess.run( + [ + "/usr/bin/gdbus", + "call", + "--session", + "--dest", + "org.freedesktop.systemd1", + "--object-path", + "/org/freedesktop/systemd1", + "--method", + "org.freedesktop.systemd1.Manager.StartUnit", + "labfy-removable-media.service", + "replace", + ], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=5, + check=False, + ) + + +def request(path: Path, payload: dict[str, str]) -> dict[str, object]: + encoded = json.dumps(payload, separators=(",", ":")).encode("utf-8") + b"\n" + last_error: OSError | None = None + for attempt in range(21): + client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + client.settimeout(35) + try: + client.connect(str(path)) + client.sendall(encoded) + data = bytearray() + while b"\n" not in data: + chunk = client.recv(4096) + if not chunk: + break + data.extend(chunk) + if len(data) > MAX_RESPONSE: + raise RuntimeError("Réponse du daemon trop volumineuse") + if not data or b"\n" not in data: + raise RuntimeError("Réponse incomplète du daemon") + value = json.loads(data.partition(b"\n")[0].decode("utf-8")) + if not isinstance(value, dict): + raise RuntimeError("Réponse JSON invalide du daemon") + return value + except (FileNotFoundError, ConnectionRefusedError) as error: + last_error = error + if attempt == 0: + activate_service() + time.sleep(0.1) + finally: + client.close() + raise RuntimeError(f"Daemon indisponible: {last_error}") + + +def main(argv: list[str]) -> int: + if len(argv) < 2 or argv[1] not in OPERATIONS: + return usage() + operation = argv[1] + expected = 2 if operation == "list" else 3 + if len(argv) != expected: + return usage() + payload = {"operation": operation} + if operation != "list": + payload["runtime_id"] = argv[2] + try: + response = request(socket_path(), payload) + except Exception as error: + print(f"labfy-removable-mediactl: {str(error)[:240]}", file=sys.stderr) + return 1 + if not response.get("ok"): + print( + f"labfy-removable-mediactl: {str(response.get('error', 'Erreur inconnue'))[:240]}", + file=sys.stderr, + ) + return 1 + state = response.get("state") + print(json.dumps(state, ensure_ascii=False, separators=(",", ":"))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv)) diff --git a/docs/systemd.md b/docs/systemd.md index 6de152259..afe4251eb 100644 --- a/docs/systemd.md +++ b/docs/systemd.md @@ -17,7 +17,7 @@ mais il ne l'active pas. Les unités effectivement versionnées sous | Domaine | Unités | Fonction | | --- | --- | --- | | session | `gnome-keyring-daemon.service`, `ssh-agent.service`, `wlsunset.service` | agents et services de session | -| desktop | `quickshell-labfy-sway.service`, `labfy-quickshell-updates.service/.timer` | barre, notifications et vérification des mises à jour | +| desktop | `quickshell-labfy-sway.service`, `labfy-quickshell-updates.service/.timer`, `labfy-removable-media.service` | barre, notifications, vérification des mises à jour et gestion des supports amovibles | | clipboard | `cliphist-text.service`, `cliphist-image.service` | collecte texte et images Wayland | | idle | `swayidle.service` | verrouillage, alimentation des sorties et pause média | | batterie | `battery-low-notify.service/.timer` | contrôle et notification périodiques | @@ -33,6 +33,55 @@ et `timers.target.wants/` expriment les activations utilisateur retenues. `.assets/enabled-user-aux-units.txt` capture en complément les timers et sockets activés observés sur la machine. +## Supports amovibles USB/SD + +`labfy-removable-media.service` appartient au package Stow `systemd` et son +exécutable `labfy-removable-media`, ainsi que le client +`labfy-removable-mediactl`, appartiennent au package `bin`. Après le déploiement +de ces deux packages, le lien versionné dans +`wayland-session@sway.desktop.target.wants/` démarre le service avec la session +Sway. L'unité est rattachée à `wayland-session@sway.desktop.target`, redémarre +sur échec et n'est pas un service système. + +Le daemon observe l'ObjectManager UDisks du bus système et publie son état +runtime JSON V1 dans `$XDG_RUNTIME_DIR/labfy-removable-media.json`. Son socket +Unix `$XDG_RUNTIME_DIR/labfy-removable-media.sock` et le snapshot sont créés +avec le mode `0600`. L'IPC n'accepte qu'une requête JSON +terminée par LF pour `list`, `mount`, `unmount`, `open` ou `safe-remove` ; les +identifiants runtime sont relus dans UDisks avant toute mutation. + +Seuls les systèmes de fichiers des lecteurs USB ou SD/MMC sont présentés. Les +blocs `HintIgnore` et `HintSystem`, ainsi que les volumes `crypto_LUKS`, sont +exclus ; aucun déverrouillage LUKS n'est tenté. MTP ne relève pas de ce contrat +UDisks `Block`/`Filesystem` et n'est pas pris en charge. Un objet UDisks +nouvellement présent est monté au plus une fois : un échec est isolé des autres +volumes, et une nouvelle tentative automatique demande une disparition puis une +réinsertion de l'objet. Un démontage manuel ne provoque donc pas de remontage +sur un événement de propriétés ultérieur. + +La commande `safe-remove` porte sur le lecteur : elle démonte d'abord tous ses +systèmes de fichiers, y compris ceux masqués par l'interface, puis relit leur +état avant d'appeler `PowerOff` ou `Eject`. + +Après Stow et le rechargement de systemd user, vérifier l'unité et exécuter les +tests synthétiques sans périphérique réel : + +```bash +systemctl --user daemon-reload +systemd-analyze --user verify \ + "$HOME"/.config/systemd/user/labfy-removable-media.service +systemctl --user is-enabled labfy-removable-media.service +python3 -B -m unittest discover -s tests/removable_media -p 'test_*.py' +``` + +Les tests utilisent des backends et sockets synthétiques. Ils couvrent le +filtrage UDisks, l'automontage borné, les opérations par volume ou lecteur, le +contrat JSON et les limites de l'IPC, mais ne constituent pas une validation sur +matériel USB/SD réel. Cette validation n'a pas encore été effectuée. + +La gestion des supports amovibles est indépendante de Session Restore : elle ne +participe ni aux checkpoints, ni au choix de restauration au démarrage. + ## Dépendances externes Les unités Trainlog, Lardon et Arch Sentinel supposent que leurs projets ou diff --git a/quickshell/.config/quickshell/labfy-sway/Bar.qml b/quickshell/.config/quickshell/labfy-sway/Bar.qml index 78146f437..44ec39fd9 100644 --- a/quickshell/.config/quickshell/labfy-sway/Bar.qml +++ b/quickshell/.config/quickshell/labfy-sway/Bar.qml @@ -45,6 +45,7 @@ PanelWindow { const opening = !dateCenter.visible && !dateCenter.opening; controlCenter.visible = false; windowStrip.closeMenu(); + rightStatusArea.closeRemovableMedia(); if (opening) dateCenter.requestOpen(); else dateCenter.requestClose(); } @@ -76,6 +77,7 @@ PanelWindow { onMenuOpened: { controlCenter.visible = false; dateCenter.requestClose(); + rightStatusArea.closeRemovableMedia(); } } } @@ -111,6 +113,7 @@ PanelWindow { onOpenPageRequested: page => { dateCenter.requestClose(); windowStrip.closeMenu(); + rightStatusArea.closeRemovableMedia(); controlCenter.currentPage = page; controlCenter.visible = true; } @@ -118,9 +121,17 @@ PanelWindow { const opening = !controlCenter.visible; dateCenter.requestClose(); windowStrip.closeMenu(); + rightStatusArea.closeRemovableMedia(); controlCenter.currentPage = 0; controlCenter.visible = opening; } + onToggleRemovableMediaRequested: { + const opening = !rightStatusArea.removableMediaOpen; + controlCenter.visible = false; + dateCenter.requestClose(); + windowStrip.closeMenu(); + rightStatusArea.setRemovableMediaOpen(opening); + } } } diff --git a/quickshell/.config/quickshell/labfy-sway/README.md b/quickshell/.config/quickshell/labfy-sway/README.md index ecc4a08bf..a9dcafdb6 100644 --- a/quickshell/.config/quickshell/labfy-sway/README.md +++ b/quickshell/.config/quickshell/labfy-sway/README.md @@ -17,6 +17,7 @@ Bar │ ├── Bluetooth │ ├── Battery │ ├── Updates +│ ├── Supports amovibles │ ├── Tray │ └── Control Center └── Control Center @@ -54,6 +55,45 @@ Le gestionnaire de fonds d'écran local est décrit dans [WALLPAPER.md](controlcenter/WALLPAPER.md). Il utilise Sway et `swaybg`, sans service permanent ni changement automatique du thème. +## Supports amovibles + +`RemovableMediaIndicator.qml` affiche dans la zone d'état un indicateur +lorsqu'au moins un système de fichiers éligible est présent. Son `PopupWindow` +natif liste les volumes, leur état et les erreurs publiées, puis offre les +actions **Ouvrir**, **Monter** ou **Démonter**, et **Retirer en sécurité** si +le lecteur le permet. Ouvrir lance Yazi dans Kitty via UWSM. Le popup participe +à la même exclusion mutuelle que le Date Center, le menu des fenêtres et le +Control Center. + +QuickShell lit exclusivement le snapshot JSON V1 +`$XDG_RUNTIME_DIR/labfy-removable-media.json`. Il refuse intégralement un +snapshot dont `schema` n'est pas `labfy.removable-media`, dont `version` n'est +pas `1`, ou dont les appareils ne respectent pas le contrat attendu. Les +commandes passent sans shell à `labfy-removable-mediactl`, qui est l'unique +frontière de commande vers le daemon ; le socket Unix associé est privé (`0600`). + +Le service utilisateur repose sur l'ObjectManager UDisks et ne présente que +les systèmes de fichiers USB ou SD/MMC. Les blocs marqués `HintIgnore` ou +`HintSystem` sont exclus. Un volume `crypto_LUKS` n'est ni présenté ni +déverrouillé automatiquement. Les périphériques MTP ne fournissent pas ce +modèle UDisks de système de fichiers et ne sont donc pas pris en charge. Chaque +objet UDisks présent peut être monté automatiquement une seule fois ; après un +démontage manuel, il reste démonté jusqu'à sa disparition puis sa réinsertion. +Le retrait en sécurité agit au niveau du lecteur et vérifie que ses systèmes de +fichiers sont démontés avant l'éjection ou la mise hors tension. + +Le contrôle visuel synthétique peut être activé avec +`LABFY_REMOVABLE_MEDIA_TEST_STATE` (`0`, `mounted`, `unmounted` ou +`busy-error`) dans le processus QuickShell. Ces états ne lancent aucune +commande. Les tests Python synthétiques du daemon et de son IPC sont décrits +dans [systemd.md](../../../../docs/systemd.md#supports-amovibles-usbsd). + +Cette fonction est indépendante de Session Restore : elle ne lit ni n'écrit de +checkpoint de session et n'intervient pas dans son démarrage. + +La validation sur matériel USB/SD réel n'a pas encore été effectuée ; l'état +actuel est couvert par les tests synthétiques seulement. + ## Parité avec i3status-rs | Ancien bloc | QuickShell | diff --git a/quickshell/.config/quickshell/labfy-sway/status/RemovableMediaIndicator.qml b/quickshell/.config/quickshell/labfy-sway/status/RemovableMediaIndicator.qml new file mode 100644 index 000000000..842649ad2 --- /dev/null +++ b/quickshell/.config/quickshell/labfy-sway/status/RemovableMediaIndicator.qml @@ -0,0 +1,316 @@ +import QtQuick +import Quickshell +import Quickshell.Io +import "../components" +import "../controlcenter" +import "../theme" + +Item { + id: indicator + + signal toggleRequested() + property alias popupOpen: popup.visible + readonly property string statePath: Quickshell.env("XDG_RUNTIME_DIR") + + "/labfy-removable-media.json" + readonly property string clientPath: (Quickshell.env("HOME") || "") + + "/.local/bin/labfy-removable-mediactl" + // TEST_ONLY: états visuels déterministes, activés uniquement par une variable + // explicite dans le processus QuickShell ; le chemin de production reste le FileView. + readonly property string testState: Quickshell.env("LABFY_REMOVABLE_MEDIA_TEST_STATE") || "" + readonly property var devices: testState ? fakeDevices(testState) : readDevices() + readonly property int deviceCount: devices.length + readonly property bool hasBusyDevice: devices.some(device => device.busy) + + visible: deviceCount > 0 + width: visible ? 26 : 0 + height: 26 + + onDeviceCountChanged: if (deviceCount === 0) popup.visible = false + + function validDevice(device) { + return device !== null && typeof device === "object" + && typeof device.runtime_id === "string" && device.runtime_id.length > 0 + && typeof device.drive_runtime_id === "string" && device.drive_runtime_id.length > 0 + && typeof device.display_name === "string" + && (device.label === null || typeof device.label === "string") + && (device.filesystem === null || typeof device.filesystem === "string") + && Number.isSafeInteger(device.size_bytes) && device.size_bytes >= 0 + && (device.mount_point === null || typeof device.mount_point === "string") + && typeof device.mounted === "boolean" + && typeof device.removable === "boolean" + && typeof device.ejectable === "boolean" + && typeof device.power_off_capable === "boolean" + && typeof device.busy === "boolean" + && (device.error === null || typeof device.error === "string"); + } + + function readDevices() { + const raw = snapshot.text(); + if (!raw) return []; + try { + const value = JSON.parse(raw); + // CONTRACT: ignorer intégralement un snapshot d'un autre protocole ou + // partiellement écrit ; aucune commande ne doit viser ses identifiants. + if (value.schema !== "labfy.removable-media" || value.version !== 1 + || !Array.isArray(value.devices) || !Number.isSafeInteger(value.updated_at) + || value.updated_at < 0 + || !value.devices.every(validDevice)) + throw new Error("Snapshot incompatible"); + return value.devices; + } catch (error) { + console.warn("État des supports amovibles illisible :", error); + return []; + } + } + + function fakeDevice(id, mounted, busy, error) { + return { + runtime_id: "test-volume-" + id, + drive_runtime_id: "test-drive-" + id, + display_name: id === 1 ? "Clé USB de test" : "Disque de sauvegarde", + label: id === 1 ? "LABFY" : "ARCHIVES", + filesystem: id === 1 ? "exfat" : "ext4", + size_bytes: id === 1 ? 64000000000 : 1000000000000, + mount_point: mounted ? "/run/media/test/" + (id === 1 ? "LABFY" : "ARCHIVES") : "", + mounted: mounted, + removable: true, + ejectable: id === 1, + power_off_capable: id === 2, + busy: busy, + error: error + }; + } + + function fakeDevices(mode) { + if (mode === "0") return []; + if (mode === "1" || mode === "mounted" || mode === "1-mounted") + return [fakeDevice(1, true, false, "")]; + if (mode === "unmounted" || mode === "1-unmounted") + return [fakeDevice(1, false, false, "")]; + if (mode === "2" || mode === "busy-error" || mode === "2-busy-error") return [ + fakeDevice(1, true, true, ""), + fakeDevice(2, false, false, "Le support nécessite une vérification.") + ]; + console.warn("LABFY_REMOVABLE_MEDIA_TEST_STATE inconnu :", mode); + return []; + } + + function formatSize(bytes) { + if (bytes <= 0) return "Taille inconnue"; + const units = ["o", "Kio", "Mio", "Gio", "Tio"]; + let value = bytes; + let unit = 0; + while (value >= 1024 && unit < units.length - 1) { + value /= 1024; + unit++; + } + return (unit === 0 ? String(value) : value.toFixed(value >= 10 ? 0 : 1)) + + " " + units[unit]; + } + + function runAction(action, device) { + // INVARIANT: les identifiants TEST_ONLY ne quittent jamais le renderer. + if (testState || actionProcess.running || device.busy) return; + let target = device.runtime_id; + if (action === "safe-remove") target = device.drive_runtime_id; + // CONTRACT: le client est l'unique frontière de commande et reçoit un + // argv direct. Les identifiants opaques ne traversent jamais un shell. + actionProcess.command = [clientPath, action, target]; + actionProcess.running = true; + } + + FileView { + id: snapshot + path: indicator.statePath + watchChanges: true + printErrors: false + onFileChanged: reload() + } + + // WHY: le fichier peut être créé après QuickShell et son répertoire parent + // existait déjà ; un watcher de fichier absent ne reçoit pas toujours la création. + Timer { + interval: 5000 + repeat: true + running: !indicator.testState && !snapshot.loaded + onTriggered: snapshot.reload() + } + + Timer { + id: refreshAfterAction + interval: 250 + onTriggered: snapshot.reload() + } + + Process { + id: actionProcess + stderr: StdioCollector { id: actionErrors; waitForEnd: true } + onExited: (exitCode, exitStatus) => { + if (exitCode !== 0) + console.warn("Action sur le support amovible échouée :", + actionErrors.text.trim() || ("code " + exitCode)); + refreshAfterAction.restart(); + } + } + + Rectangle { + anchors.fill: parent + radius: 4 + color: pointer.containsMouse || popup.visible ? Theme.border : "transparent" + } + + NerdIcon { + anchors.centerIn: parent + text: indicator.hasBusyDevice || actionProcess.running ? "󰔟" : "󰕓" + color: indicator.hasBusyDevice || actionProcess.running + ? Theme.warningForeground : Theme.foreground + font.pixelSize: 18 + } + + MouseArea { + id: pointer + anchors.fill: parent + hoverEnabled: true + enabled: indicator.deviceCount > 0 + onClicked: indicator.toggleRequested() + } + + StatusTooltip { + target: indicator + hovered: pointer.containsMouse && !popup.visible + message: indicator.deviceCount + (indicator.deviceCount === 1 + ? " support amovible" : " supports amovibles") + } + + PopupWindow { + id: popup + anchor.item: indicator + anchor.edges: Edges.Bottom | Edges.Left + anchor.gravity: Edges.Bottom | Edges.Right + anchor.margins.bottom: -6 + visible: false + grabFocus: false + color: "transparent" + implicitWidth: 400 + implicitHeight: Math.min(560, popupContent.implicitHeight + 24) + + Rectangle { + anchors.fill: parent + radius: 4 + color: Theme.popupBackground + border.color: Theme.outline + + Flickable { + anchors.fill: parent + anchors.margins: 12 + clip: true + contentWidth: width + contentHeight: popupContent.implicitHeight + boundsBehavior: Flickable.StopAtBounds + + Column { + id: popupContent + width: parent.width + spacing: 10 + + Text { + width: parent.width + text: "Supports amovibles" + color: Theme.accentForeground + font.pixelSize: 16 + font.bold: true + } + + Repeater { + model: indicator.devices + + Rectangle { + required property var modelData + width: popupContent.width + height: cardContent.implicitHeight + 20 + radius: 5 + color: Theme.buttonBackground + border.color: modelData.error ? Theme.danger : Theme.outline + + Column { + id: cardContent + anchors.left: parent.left + anchors.right: parent.right + anchors.top: parent.top + anchors.margins: 10 + spacing: 5 + + Text { + width: parent.width + text: modelData.display_name || modelData.label || "Support amovible" + color: Theme.foreground + font.pixelSize: 14 + font.bold: true + elide: Text.ElideRight + } + + Text { + width: parent.width + text: (modelData.label ? modelData.label + " • " : "") + + (modelData.filesystem || "Système de fichiers inconnu") + + " • " + indicator.formatSize(modelData.size_bytes) + color: Theme.secondaryForeground + font.pixelSize: 11 + elide: Text.ElideRight + } + + Text { + width: parent.width + text: modelData.busy ? "Opération en cours…" + : modelData.mounted ? "Monté sur " + modelData.mount_point + : "Non monté" + color: modelData.busy ? Theme.warningForeground + : modelData.mounted ? Theme.successForeground + : Theme.secondaryForeground + font.pixelSize: 11 + elide: Text.ElideMiddle + } + + Text { + visible: Boolean(modelData.error) + width: parent.width + text: modelData.error + color: Theme.danger + font.pixelSize: 11 + wrapMode: Text.Wrap + } + + Flow { + width: parent.width + spacing: 6 + + ActionButton { + visible: modelData.mounted + label: "Ouvrir" + enabled: !indicator.testState && !modelData.busy + && !actionProcess.running + onClicked: indicator.runAction("open", modelData) + } + ActionButton { + label: modelData.mounted ? "Démonter" : "Monter" + enabled: !indicator.testState && !modelData.busy + && !actionProcess.running + onClicked: indicator.runAction( + modelData.mounted ? "unmount" : "mount", modelData) + } + ActionButton { + visible: modelData.ejectable || modelData.power_off_capable + label: "Retirer en sécurité" + enabled: !indicator.testState && !modelData.busy + && !actionProcess.running + onClicked: indicator.runAction("safe-remove", modelData) + } + } + } + } + } + } + } + } + } +} diff --git a/quickshell/.config/quickshell/labfy-sway/status/RightStatusArea.qml b/quickshell/.config/quickshell/labfy-sway/status/RightStatusArea.qml index 5b44c97ad..08f2717c1 100644 --- a/quickshell/.config/quickshell/labfy-sway/status/RightStatusArea.qml +++ b/quickshell/.config/quickshell/labfy-sway/status/RightStatusArea.qml @@ -9,7 +9,11 @@ Row { required property bool controlCenterOpen signal openPageRequested(int page) signal toggleControlCenterRequested() + signal toggleRemovableMediaRequested() function refreshBatteryThreshold() { batteryIndicator.refreshThreshold() } + function closeRemovableMedia() { removableMediaIndicator.popupOpen = false } + function setRemovableMediaOpen(open) { removableMediaIndicator.popupOpen = open } + readonly property bool removableMediaOpen: removableMediaIndicator.popupOpen spacing: 5 height: 26 @@ -17,6 +21,10 @@ Row { BluetoothIndicator { adapter: area.adapter; onActivated: area.openPageRequested(2) } BatteryIndicator { id: batteryIndicator; onActivated: area.openPageRequested(5) } UpdatesIndicator { } + RemovableMediaIndicator { + id: removableMediaIndicator + onToggleRequested: area.toggleRemovableMediaRequested() + } Tray { barWindow: area.barWindow } ControlCenterButton { open: area.controlCenterOpen; onToggled: area.toggleControlCenterRequested() } } diff --git a/systemd/.config/systemd/user/labfy-removable-media.service b/systemd/.config/systemd/user/labfy-removable-media.service new file mode 100644 index 000000000..2e2e4440b --- /dev/null +++ b/systemd/.config/systemd/user/labfy-removable-media.service @@ -0,0 +1,13 @@ +[Unit] +Description=Gestion des supports amovibles USB et SD +PartOf=wayland-session@sway.desktop.target +After=wayland-session@sway.desktop.target + +[Service] +Type=simple +ExecStart=%h/.local/bin/labfy-removable-media +Restart=on-failure +RestartSec=2s + +[Install] +WantedBy=wayland-session@sway.desktop.target diff --git a/systemd/.config/systemd/user/wayland-session@sway.desktop.target.wants/labfy-removable-media.service b/systemd/.config/systemd/user/wayland-session@sway.desktop.target.wants/labfy-removable-media.service new file mode 120000 index 000000000..d362900ba --- /dev/null +++ b/systemd/.config/systemd/user/wayland-session@sway.desktop.target.wants/labfy-removable-media.service @@ -0,0 +1 @@ +../labfy-removable-media.service \ No newline at end of file diff --git a/tests/removable_media/test_backend.py b/tests/removable_media/test_backend.py new file mode 100644 index 000000000..888876415 --- /dev/null +++ b/tests/removable_media/test_backend.py @@ -0,0 +1,445 @@ +import importlib.util +from importlib.machinery import SourceFileLoader +import json +import stat +import sys +import tempfile +import unittest +from dataclasses import replace +from pathlib import Path + + +SCRIPT = Path(__file__).parents[2] / "bin/.local/bin/labfy-removable-media" +SPEC = importlib.util.spec_from_loader( + "labfy_removable_media", SourceFileLoader("labfy_removable_media", str(SCRIPT)) +) +assert SPEC and SPEC.loader +media = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = media +SPEC.loader.exec_module(media) + + +def record(path, drive="/org/freedesktop/UDisks2/drives/usb", mount=""): + return media.FilesystemRecord( + path=f"/org/freedesktop/UDisks2/block_devices/{path}", + drive_path=drive, + display_name=path, + label=path.upper(), + filesystem="vfat", + size_bytes=1024, + mount_point=mount, + removable=True, + ejectable=True, + power_off_capable=True, + ) + + +class FakeBackend: + def __init__(self, records): + self.items = records + self.calls = [] + + def records(self): + return list(self.items) + + def _change_mount(self, path, value): + self.items = [ + replace(item, mount_point=value if item.path == path else item.mount_point) + for item in self.items + ] + + def mount(self, path): + self.calls.append(("mount", path)) + self._change_mount(path, f"/run/media/test/{Path(path).name}") + return self.mount_points_sync(path) + + def unmount(self, path): + self.calls.append(("unmount", path)) + self._change_mount(path, "") + return [] + + def mount_points_sync(self, path): + for item in self.items: + if item.path == path: + return [item.mount_point.encode() + b"\0"] if item.mount_point else [] + raise media.RequestError("Volume introuvable après la demande") + + def safe_remove(self, drive): + self.calls.append(("safe-remove", drive.drive_path)) + + +class FakeInterface: + def __init__(self, name, **properties): + self.name = name + self.properties = properties + + def get_interface_name(self): + return self.name + + def get_cached_property(self, name): + return self.properties.get(name) + + +class FakeObject: + def __init__(self, path, *interfaces): + self.path = path + self.interfaces = interfaces + + def get_object_path(self): + return self.path + + def get_interfaces(self): + return self.interfaces + + +class MediaDaemonTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.state_path = Path(self.temporary.name) / media.STATE_NAME + + def tearDown(self): + self.temporary.cleanup() + + def test_state_contract_and_private_atomic_file(self): + backend = FakeBackend([record("sdb1", mount="/run/media/test/A")]) + daemon = media.MediaDaemon(backend, self.state_path) + + state = daemon.state() + self.assertEqual(state["schema"], "labfy.removable-media") + self.assertEqual(state["version"], 1) + self.assertEqual(set(state), {"schema", "version", "devices", "updated_at"}) + self.assertEqual( + set(state["devices"][0]), + { + "runtime_id", "drive_runtime_id", "display_name", "label", + "filesystem", "size_bytes", "mount_point", "mounted", + "removable", "ejectable", "power_off_capable", "busy", "error", + }, + ) + self.assertTrue(state["devices"][0]["mounted"]) + self.assertEqual(stat.S_IMODE(self.state_path.stat().st_mode), 0o600) + self.assertEqual(json.loads(self.state_path.read_text()), state) + + def test_mount_and_unmount_only_apply_to_selected_volume(self): + first, second = record("sdb1"), record("sdb2") + other = record("sdc1", "/org/freedesktop/UDisks2/drives/other") + backend = FakeBackend([first, second, other]) + daemon = media.MediaDaemon(backend, self.state_path) + + daemon.handle({"operation": "mount", "runtime_id": first.path}) + self.assertEqual( + backend.calls, + [("mount", first.path)], + ) + daemon.handle({"operation": "unmount", "runtime_id": first.path}) + self.assertEqual( + backend.calls[-1:], + [("unmount", first.path)], + ) + + def test_safe_remove_is_delegated_once_at_drive_scope(self): + first = record("sdb1", mount="/run/media/test/one") + second = record("sdb2", mount="/run/media/test/two") + backend = FakeBackend([first, second]) + daemon = media.MediaDaemon(backend, self.state_path) + + daemon.handle({"operation": "safe-remove", "runtime_id": first.drive_path}) + self.assertEqual( + backend.calls, + [("safe-remove", first.drive_path)], + ) + + def test_stale_runtime_id_is_rejected_after_revalidation(self): + item = record("sdb1") + backend = FakeBackend([item]) + daemon = media.MediaDaemon(backend, self.state_path) + backend.items = [] + + with self.assertRaisesRegex(media.RequestError, "introuvable"): + daemon.handle({"operation": "mount", "runtime_id": item.path}) + self.assertEqual(backend.calls, []) + + def test_open_mounts_then_launches_selected_mountpoint(self): + item = record("sdb1") + backend = FakeBackend([item]) + launched = [] + daemon = media.MediaDaemon(backend, self.state_path, launched.append) + + daemon.handle({"operation": "open", "runtime_id": item.path}) + self.assertEqual(launched, ["/run/media/test/sdb1"]) + + def test_error_is_bounded_and_published(self): + class FailingBackend(FakeBackend): + def mount(self, path): + raise RuntimeError("x" * 500) + + item = record("sdb1") + daemon = media.MediaDaemon(FailingBackend([item]), self.state_path) + with self.assertRaises(media.RequestError): + daemon.handle({"operation": "mount", "runtime_id": item.path}) + self.assertEqual(len(daemon.state()["devices"][0]["error"]), media.MAX_ERROR) + + def test_device_busy_is_translated_and_preserves_mounted_state(self): + item = record("sda1", mount="/run/media/test/Ventoy") + + class BusyUnmountBackend(FakeBackend): + def unmount(self, path): + self.calls.append(("unmount", path)) + raise RuntimeError( + "g-io-error-quark: " + "GDBus.Error:org.freedesktop.UDisks2.Error.DeviceBusy: " + "Error unmounting /dev/sda1: target is busy (36)" + ) + + backend = BusyUnmountBackend([item]) + daemon = media.MediaDaemon(backend, self.state_path) + + with self.assertRaisesRegex( + media.RequestError, "Le périphérique est utilisé par une application\\." + ): + daemon.handle({"operation": "unmount", "runtime_id": item.path}) + + device = daemon.state()["devices"][0] + self.assertTrue(device["mounted"]) + self.assertEqual(device["error"], "Le périphérique est utilisé par une application.") + + def test_start_automounts_each_existing_unmounted_volume_once(self): + first, second = record("sdb1"), record("sdb2", mount="/media/already") + backend = FakeBackend([first, second]) + daemon = media.MediaDaemon(backend, self.state_path) + + daemon.start() + daemon.devices_changed() + + self.assertEqual(backend.calls, [("mount", first.path)]) + self.assertTrue(daemon.state()["devices"][0]["mounted"]) + + def test_added_object_is_mounted_once_despite_duplicate_property_events(self): + backend = FakeBackend([]) + daemon = media.MediaDaemon(backend, self.state_path) + daemon.start() + item = record("sdb1") + backend.items = [item] + + daemon.devices_changed() + daemon.devices_changed() + + self.assertEqual(backend.calls, [("mount", item.path)]) + + def test_manual_unmount_stays_unmounted_on_later_event(self): + item = record("sdb1") + backend = FakeBackend([item]) + daemon = media.MediaDaemon(backend, self.state_path) + daemon.start() + backend.calls.clear() + + daemon.handle({"operation": "unmount", "runtime_id": item.path}) + daemon.devices_changed() + + self.assertEqual(backend.calls, [("unmount", item.path)]) + self.assertFalse(daemon.state()["devices"][0]["mounted"]) + + def test_automount_failure_is_isolated_and_not_retried_until_reinserted(self): + first, second = record("sdb1"), record("sdb2") + + class OneFailureBackend(FakeBackend): + def mount(self, path): + self.calls.append(("mount", path)) + if path == first.path: + raise RuntimeError("échec " + "x" * 500) + self._change_mount(path, "/media/second") + + backend = OneFailureBackend([first, second]) + daemon = media.MediaDaemon(backend, self.state_path) + daemon.start() + daemon.devices_changed() + + state = {item["runtime_id"]: item for item in daemon.state()["devices"]} + self.assertEqual(backend.calls, [("mount", first.path), ("mount", second.path)]) + self.assertEqual(len(state[first.path]["error"]), media.MAX_ERROR) + self.assertTrue(state[second.path]["mounted"]) + self.assertEqual(state[second.path]["error"], "") + + backend.items = [second] + daemon.devices_changed() + backend.items = [first, second] + daemon.devices_changed() + self.assertEqual(backend.calls.count(("mount", first.path)), 2) + + def test_manual_mount_uses_real_mountpoint_and_unmount_verifies_state(self): + item = record("sdb1") + + class StickyUnmountBackend(FakeBackend): + def mount(self, path): + self.calls.append(("mount", path)) + self._change_mount(path, "/media/udisks-result") + + def unmount(self, path): + self.calls.append(("unmount", path)) + + backend = StickyUnmountBackend([item]) + launched = [] + daemon = media.MediaDaemon(backend, self.state_path, launched.append) + daemon.handle({"operation": "open", "runtime_id": item.path}) + self.assertEqual(launched, ["/media/udisks-result"]) + + with self.assertRaisesRegex(media.RequestError, "reste monté"): + daemon.handle({"operation": "unmount", "runtime_id": item.path}) + + +class UDisksBackendTests(unittest.TestCase): + def test_records_keep_usb_filesystems_only_and_exclude_luks(self): + usb_drive = "/org/freedesktop/UDisks2/drives/usb" + sata_drive = "/org/freedesktop/UDisks2/drives/sata" + drive_interface = media.UDisksBackend.IFACE_DRIVE + block_interface = media.UDisksBackend.IFACE_BLOCK + filesystem_interface = media.UDisksBackend.IFACE_FILESYSTEM + objects = [ + FakeObject( + usb_drive, + FakeInterface( + drive_interface, + ConnectionBus="usb", + MediaCompatibility=[], + Model="Clé USB", + Removable=True, + Ejectable=False, + CanPowerOff=True, + ), + ), + FakeObject( + sata_drive, + FakeInterface( + drive_interface, + ConnectionBus="ata", + MediaCompatibility=[], + Model="Disque interne", + Removable=True, + ), + ), + FakeObject( + "/org/freedesktop/UDisks2/block_devices/sdb1", + FakeInterface( + block_interface, + Drive=usb_drive, + IdUsage="filesystem", + IdType="vfat", + IdLabel="DONNEES", + Size=4096, + ), + FakeInterface(filesystem_interface, MountPoints=[b"/run/media/test\0"]), + ), + FakeObject( + "/org/freedesktop/UDisks2/block_devices/sdb2", + FakeInterface( + block_interface, + Drive=usb_drive, + IdUsage="crypto", + IdType="crypto_LUKS", + Size=4096, + ), + FakeInterface(filesystem_interface, MountPoints=[]), + ), + FakeObject( + "/org/freedesktop/UDisks2/block_devices/sda1", + FakeInterface( + block_interface, + Drive=sata_drive, + IdUsage="filesystem", + IdType="ext4", + Size=8192, + ), + FakeInterface(filesystem_interface, MountPoints=[b"/\0"]), + ), + ] + backend = media.UDisksBackend.__new__(media.UDisksBackend) + backend.manager = type("Manager", (), {"get_objects": lambda self: objects})() + + records = backend.records() + + self.assertEqual(len(records), 1) + self.assertEqual(records[0].label, "DONNEES") + self.assertEqual(records[0].mount_point, "/run/media/test") + self.assertTrue(records[0].power_off_capable) + + def test_records_exclude_hint_system_and_hint_ignore(self): + drive_path = "/org/freedesktop/UDisks2/drives/usb" + drive_interface = media.UDisksBackend.IFACE_DRIVE + block_interface = media.UDisksBackend.IFACE_BLOCK + filesystem_interface = media.UDisksBackend.IFACE_FILESYSTEM + objects = [ + FakeObject( + drive_path, + FakeInterface( + drive_interface, + ConnectionBus="usb", + MediaCompatibility=[], + Removable=True, + ), + ) + ] + for name, properties in ( + ("visible", {}), + ("efi", {"HintSystem": True}), + ("ignored", {"HintIgnore": True}), + ): + objects.append( + FakeObject( + f"/org/freedesktop/UDisks2/block_devices/{name}", + FakeInterface( + block_interface, + Drive=drive_path, + IdUsage="filesystem", + IdType="vfat", + **properties, + ), + FakeInterface(filesystem_interface, MountPoints=[]), + ) + ) + backend = media.UDisksBackend.__new__(media.UDisksBackend) + backend.manager = type("Manager", (), {"get_objects": lambda self: objects})() + + self.assertEqual( + [item.path for item in backend.records()], + ["/org/freedesktop/UDisks2/block_devices/visible"], + ) + + def test_safe_remove_rechecks_every_filesystem_child_before_power_off(self): + drive = record("visible", mount="/run/media/test/visible") + + class Backend(media.UDisksBackend): + def __init__(self): + self.calls = [] + self.reads = 0 + + def _drive_filesystems(self, drive_path): + self.reads += 1 + return [ + (drive.path, [b"/run/media/test/visible\0"]), + ("/org/freedesktop/UDisks2/block_devices/hidden", [b"/mnt/hidden\0"]), + ] + + def _drive_filesystems_sync(self, drive_path): + return self._drive_filesystems(drive_path) + + def unmount(self, path): + self.calls.append(("unmount", path)) + + def _call(self, path, interface, method): + self.calls.append((method, path)) + + backend = Backend() + with self.assertRaisesRegex(media.RequestError, "reste monté"): + backend.safe_remove(drive) + self.assertEqual(backend.reads, 2) + self.assertEqual( + backend.calls, + [ + ("unmount", "/org/freedesktop/UDisks2/block_devices/hidden"), + ("unmount", drive.path), + ], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/removable_media/test_ipc.py b/tests/removable_media/test_ipc.py new file mode 100644 index 000000000..72fb54888 --- /dev/null +++ b/tests/removable_media/test_ipc.py @@ -0,0 +1,97 @@ +import importlib.util +from importlib.machinery import SourceFileLoader +import json +import os +import socket +import stat +import sys +import tempfile +import threading +import unittest +from pathlib import Path + + +SCRIPT = Path(__file__).parents[2] / "bin/.local/bin/labfy-removable-media" +SPEC = importlib.util.spec_from_loader( + "labfy_removable_media_ipc", + SourceFileLoader("labfy_removable_media_ipc", str(SCRIPT)), +) +assert SPEC and SPEC.loader +media = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = media +SPEC.loader.exec_module(media) + + +class IpcTests(unittest.TestCase): + def test_request_allowlist_and_shape(self): + self.assertEqual(media.validate_request({"operation": "list"}), ("list", None)) + with self.assertRaises(media.RequestError): + media.validate_request({"operation": "format", "runtime_id": "/tmp/x"}) + with self.assertRaises(media.RequestError): + media.validate_request({"operation": "list", "extra": True}) + with self.assertRaises(media.RequestError): + media.validate_request({"operation": "mount", "runtime_id": "sdb1"}) + + def test_socket_mode_and_json_round_trip(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "media.sock" + seen = [] + server = media.JsonSocketServer( + path, lambda request: seen.append(request) or {"ok": True} + ) + server.start() + self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) + client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + client.connect(str(path)) + thread = threading.Thread( + target=lambda: server._serve_connection(server.socket.accept()[0]) + ) + thread.start() + client.sendall(b'{"operation":"list"}\n') + response = json.loads(client.makefile("rb").readline()) + client.close() + thread.join(2) + server.socket.close() + self.assertFalse(thread.is_alive()) + self.assertEqual(response, {"ok": True}) + self.assertEqual(seen, [{"operation": "list"}]) + + def test_oversized_request_is_rejected_without_calling_handler(self): + called = [] + server = media.JsonSocketServer(Path("unused"), lambda request: called.append(request)) + server_side, client_side = socket.socketpair() + thread = threading.Thread(target=server._serve_connection, args=(server_side,)) + thread.start() + client_side.sendall(b"x" * (media.MAX_REQUEST + 1)) + response = json.loads(client_side.makefile("rb").readline()) + thread.join(2) + client_side.close() + self.assertFalse(response["ok"]) + self.assertIn("volumineuse", response["error"]) + self.assertEqual(called, []) + + def test_slow_incomplete_request_times_out_without_calling_handler(self): + called = [] + server = media.JsonSocketServer(Path("unused"), lambda request: called.append(request)) + server_side, client_side = socket.socketpair() + previous_timeout = media.READ_TIMEOUT + media.READ_TIMEOUT = 0.05 + try: + thread = threading.Thread( + target=server._serve_connection, args=(server_side,) + ) + thread.start() + client_side.sendall(b'{"operation":') + response = json.loads(client_side.makefile("rb").readline()) + thread.join(2) + finally: + media.READ_TIMEOUT = previous_timeout + client_side.close() + self.assertFalse(thread.is_alive()) + self.assertFalse(response["ok"]) + self.assertIn("Délai de lecture", response["error"]) + self.assertEqual(called, []) + + +if __name__ == "__main__": + unittest.main()