1127 lines
42 KiB
Python
Executable file
1127 lines
42 KiB
Python
Executable file
#!/usr/bin/python3
|
|
"""Daemon d'accès aux supports bloc UDisks2 et aux appareils MTP Gio."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass, replace
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
SCHEMA = "labfy.removable-media"
|
|
VERSION = 2
|
|
MAX_REQUEST = 16 * 1024
|
|
MAX_ERROR = 240
|
|
MAX_CLIENTS = 8
|
|
READ_TIMEOUT = 2.0
|
|
OPERATION_TIMEOUT_SECONDS = 30
|
|
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 public_mtp_operation_error(error: BaseException, operation: str) -> str:
|
|
"""Ne publie jamais le détail Gio, qui peut contenir l'URI MTP."""
|
|
if isinstance(error, RequestError):
|
|
return bounded_error(error)
|
|
return {
|
|
"mount": "Impossible de monter l'appareil MTP.",
|
|
"unmount": "Impossible de démonter l'appareil MTP.",
|
|
"safe-remove": "Impossible de fermer la session MTP.",
|
|
}.get(operation, "Impossible d'effectuer l'opération sur l'appareil MTP.")
|
|
|
|
|
|
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
|
|
kind: str = "block"
|
|
mounted_state: bool | None = None
|
|
|
|
@property
|
|
def mounted(self) -> bool:
|
|
return bool(self.mount_point) if self.mounted_state is None else self.mounted_state
|
|
|
|
@property
|
|
def safe_remove_runtime_id(self) -> str:
|
|
return self.drive_path
|
|
|
|
def actions(self) -> list[str]:
|
|
actions = ["unmount", "open"] if self.mount_point else ["mount", "open"]
|
|
if self.power_off_capable or self.ejectable:
|
|
actions.append("safe-remove")
|
|
return actions
|
|
|
|
def public(self, busy: bool = False, error: str = "") -> dict[str, Any]:
|
|
return {
|
|
"runtime_id": self.path,
|
|
"safe_remove_runtime_id": self.safe_remove_runtime_id,
|
|
"kind": self.kind,
|
|
"actions": self.actions(),
|
|
"display_name": self.display_name,
|
|
"label": self.label,
|
|
"filesystem": self.filesystem,
|
|
"size_bytes": self.size_bytes,
|
|
"mount_point": self.mount_point,
|
|
"mounted": self.mounted,
|
|
"removable": self.removable,
|
|
"ejectable": self.ejectable,
|
|
"power_off_capable": self.power_off_capable,
|
|
"busy": busy,
|
|
"error": error,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MTPRecord(FilesystemRecord):
|
|
"""Vue publique d'un volume MTP, sans URI ni identité matérielle."""
|
|
|
|
kind: str = "mtp"
|
|
|
|
def actions(self) -> list[str]:
|
|
if not self.mounted:
|
|
return ["mount"]
|
|
actions = ["unmount"]
|
|
if self.mount_point:
|
|
actions.append("open")
|
|
actions.append("safe-remove")
|
|
return actions
|
|
|
|
|
|
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 MTPBackend:
|
|
"""Adaptateur des volumes MTP exposés par GVFS via Gio.VolumeMonitor."""
|
|
|
|
def __init__(self, Gio: Any, GLib: Any, monitor: Any | None = None) -> None:
|
|
self.Gio = Gio
|
|
self.GLib = GLib
|
|
self.monitor = monitor or Gio.VolumeMonitor.get()
|
|
self._nonce = uuid.uuid4().hex
|
|
self._counter = 0
|
|
self._ids_by_identity: dict[str, str] = {}
|
|
self._identity_by_id: dict[str, str] = {}
|
|
self._volumes_by_id: dict[str, Any] = {}
|
|
|
|
@staticmethod
|
|
def _is_mtp(volume: Any) -> bool:
|
|
try:
|
|
root = volume.get_activation_root()
|
|
return root is not None and str(root.get_uri_scheme()).lower() == "mtp"
|
|
except (AttributeError, TypeError):
|
|
return False
|
|
|
|
def _live_volumes(self) -> list[Any]:
|
|
return [volume for volume in self.monitor.get_volumes() if self._is_mtp(volume)]
|
|
|
|
@staticmethod
|
|
def _identity(volume: Any) -> str:
|
|
"""Identifie une présence sans conserver ni publier son URI MTP."""
|
|
root = volume.get_activation_root()
|
|
uri = str(root.get_uri())
|
|
return hashlib.sha256(uri.encode("utf-8")).hexdigest()
|
|
|
|
def _runtime_id(self, volume: Any) -> str:
|
|
identity = self._identity(volume)
|
|
runtime_id = self._ids_by_identity.get(identity)
|
|
if runtime_id is None:
|
|
self._counter += 1
|
|
runtime_id = f"mtp:{self._nonce}:{self._counter}"
|
|
self._ids_by_identity[identity] = runtime_id
|
|
self._identity_by_id[runtime_id] = identity
|
|
self._volumes_by_id[runtime_id] = volume
|
|
return runtime_id
|
|
|
|
def _forget_volume(self, volume: Any) -> None:
|
|
if not self._is_mtp(volume):
|
|
return
|
|
identity = self._identity(volume)
|
|
runtime_id = self._ids_by_identity.pop(identity, None)
|
|
if runtime_id is not None:
|
|
self._identity_by_id.pop(runtime_id, None)
|
|
self._volumes_by_id.pop(runtime_id, None)
|
|
|
|
@staticmethod
|
|
def _mount_path(volume: Any) -> str:
|
|
mount = volume.get_mount()
|
|
if mount is None:
|
|
return ""
|
|
root = mount.get_root()
|
|
if root is None:
|
|
return ""
|
|
# CONTRACT: ce chemin local est fourni par GVFS. Une URI MTP ne doit
|
|
# jamais servir à reconstruire à la main un chemin sous /run/user.
|
|
return str(root.get_path() or "")
|
|
|
|
@staticmethod
|
|
def _size(volume: Any) -> int:
|
|
mount = volume.get_mount()
|
|
if mount is None:
|
|
return 0
|
|
try:
|
|
info = mount.get_root().query_filesystem_info("filesystem::size", None)
|
|
return int(info.get_attribute_uint64("filesystem::size"))
|
|
except Exception:
|
|
return 0
|
|
|
|
def records(self) -> list[FilesystemRecord]:
|
|
volumes = self._live_volumes()
|
|
live_identities = {self._identity(volume) for volume in volumes}
|
|
for identity, runtime_id in list(self._ids_by_identity.items()):
|
|
if identity not in live_identities:
|
|
self._ids_by_identity.pop(identity, None)
|
|
self._identity_by_id.pop(runtime_id, None)
|
|
self._volumes_by_id.pop(runtime_id, None)
|
|
records: list[FilesystemRecord] = []
|
|
for volume in volumes:
|
|
runtime_id = self._runtime_id(volume)
|
|
name = str(volume.get_name() or "Appareil Android")
|
|
mount = volume.get_mount()
|
|
display_name = name
|
|
if mount is not None:
|
|
try:
|
|
display_name = str(mount.get_name() or name)
|
|
except AttributeError:
|
|
pass
|
|
records.append(
|
|
MTPRecord(
|
|
path=runtime_id,
|
|
drive_path=runtime_id,
|
|
display_name=display_name,
|
|
label=name,
|
|
filesystem="mtp",
|
|
size_bytes=self._size(volume),
|
|
mount_point=self._mount_path(volume),
|
|
removable=True,
|
|
ejectable=False,
|
|
power_off_capable=False,
|
|
mounted_state=mount is not None,
|
|
)
|
|
)
|
|
return sorted(records, key=lambda item: item.path)
|
|
|
|
def connect_changed(self, callback: Callable[[], None]) -> None:
|
|
def volume_removed(_monitor: Any, volume: Any, *_args: Any) -> None:
|
|
self._forget_volume(volume)
|
|
callback()
|
|
|
|
self.monitor.connect("volume-removed", volume_removed)
|
|
for signal in (
|
|
"volume-added",
|
|
"volume-changed",
|
|
"mount-added",
|
|
"mount-changed",
|
|
"mount-removed",
|
|
):
|
|
self.monitor.connect(signal, lambda *_args: callback())
|
|
|
|
def _volume(self, runtime_id: str) -> Any:
|
|
# INVARIANT: relire le moniteur avant toute action et utiliser le wrapper
|
|
# vivant correspondant à la présence ayant reçu cet identifiant opaque.
|
|
identity = self._identity_by_id.get(runtime_id)
|
|
volume = next(
|
|
(
|
|
item
|
|
for item in self._live_volumes()
|
|
if identity is not None and self._identity(item) == identity
|
|
),
|
|
None,
|
|
)
|
|
if volume is None:
|
|
raise RequestError("Appareil MTP introuvable ou retiré")
|
|
self._volumes_by_id[runtime_id] = volume
|
|
return volume
|
|
|
|
def operate_async(
|
|
self,
|
|
runtime_id: str,
|
|
operation: str,
|
|
callback: Callable[[str | None, BaseException | None], None],
|
|
) -> Callable[[], None]:
|
|
volume = self._volume(runtime_id)
|
|
cancellable = self.Gio.Cancellable.new()
|
|
finished = False
|
|
timeout_id: int | None = None
|
|
settle_id: int | None = None
|
|
|
|
def complete(path: str | None = None, error: BaseException | None = None) -> None:
|
|
nonlocal finished
|
|
if finished:
|
|
return
|
|
finished = True
|
|
if timeout_id is not None:
|
|
self.GLib.source_remove(timeout_id)
|
|
if settle_id is not None:
|
|
self.GLib.source_remove(settle_id)
|
|
callback(path, error)
|
|
|
|
def timed_out() -> bool:
|
|
cancellable.cancel()
|
|
complete(error=RequestError("Délai d'opération MTP dépassé"))
|
|
return False
|
|
|
|
timeout_id = self.GLib.timeout_add_seconds(OPERATION_TIMEOUT_SECONDS, timed_out)
|
|
|
|
def verify_postcondition(expect_mounted: bool) -> bool:
|
|
try:
|
|
current = self._volume(runtime_id)
|
|
mounted = current.get_mount() is not None
|
|
if mounted != expect_mounted:
|
|
return True
|
|
complete(self._mount_path(current) if mounted else "")
|
|
except BaseException as error:
|
|
complete(error=error)
|
|
return False
|
|
|
|
def settle(expect_mounted: bool) -> None:
|
|
nonlocal settle_id
|
|
if not verify_postcondition(expect_mounted):
|
|
return
|
|
# WHY: GVFS peut publier GMount quelques instants après mount_finish.
|
|
# Le polling rend la main entre chaque lecture et reste couvert par
|
|
# le timeout/cancellable global de l'opération.
|
|
settle_id = self.GLib.timeout_add(100, verify_postcondition, expect_mounted)
|
|
|
|
def mount_done(source: Any, result: Any, *_args: Any) -> None:
|
|
try:
|
|
source.mount_finish(result)
|
|
settle(True)
|
|
except BaseException as error:
|
|
complete(error=error)
|
|
|
|
def unmount_done(source: Any, result: Any, *_args: Any) -> None:
|
|
try:
|
|
source.unmount_with_operation_finish(result)
|
|
settle(False)
|
|
except BaseException as error:
|
|
complete(error=error)
|
|
|
|
try:
|
|
if operation == "mount":
|
|
volume.mount(
|
|
self.Gio.MountMountFlags.NONE, None, cancellable, mount_done
|
|
)
|
|
elif operation in {"unmount", "safe-remove"}:
|
|
mount = volume.get_mount()
|
|
if mount is None:
|
|
if operation == "safe-remove":
|
|
raise RequestError("L'appareil MTP n'a pas de session à fermer")
|
|
complete("")
|
|
else:
|
|
# CONTRACT: le retrait MTP ferme uniquement la session GVFS ;
|
|
# il ne déclenche jamais UDisks PowerOff/Eject.
|
|
mount.unmount_with_operation(
|
|
self.Gio.MountUnmountFlags.NONE,
|
|
None,
|
|
cancellable,
|
|
unmount_done,
|
|
)
|
|
else:
|
|
raise RequestError("Opération MTP non autorisée")
|
|
except BaseException as error:
|
|
complete(error=error)
|
|
|
|
return cancellable.cancel
|
|
|
|
|
|
class RemovableMediaBackend:
|
|
"""Agrège les deux sources sans modifier la politique UDisks existante."""
|
|
|
|
def __init__(self, block: UDisksBackend, mtp: MTPBackend) -> None:
|
|
self.block = block
|
|
self.mtp = mtp
|
|
|
|
def records(self) -> list[FilesystemRecord]:
|
|
return self.block.records() + self.mtp.records()
|
|
|
|
def connect_changed(self, callback: Callable[[], None]) -> None:
|
|
self.block.connect_changed(callback)
|
|
self.mtp.connect_changed(callback)
|
|
|
|
def __getattr__(self, name: str) -> Any:
|
|
return getattr(self.block, name)
|
|
|
|
def operate_mtp_async(
|
|
self,
|
|
runtime_id: str,
|
|
operation: str,
|
|
callback: Callable[[str | None, BaseException | None], None],
|
|
) -> Callable[[], None]:
|
|
return self.mtp.operate_async(runtime_id, operation, callback)
|
|
|
|
|
|
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 deux sources."""
|
|
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.mounted:
|
|
continue
|
|
if item.kind == "mtp":
|
|
self._start_mtp_operation(item, "mount", None)
|
|
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 _start_mtp_operation(
|
|
self,
|
|
record: FilesystemRecord,
|
|
operation: str,
|
|
done: Callable[[dict[str, Any] | BaseException], None] | None,
|
|
) -> Callable[[], None]:
|
|
self._set_busy(record.safe_remove_runtime_id, True)
|
|
|
|
def completed(path: str | None, error: BaseException | None) -> None:
|
|
with self._lock:
|
|
result_error: BaseException | None = None
|
|
if error is None:
|
|
self._errors.pop(record.path, None)
|
|
else:
|
|
message = public_mtp_operation_error(error, operation)
|
|
self._errors[record.path] = message
|
|
result_error = RequestError(message)
|
|
LOGGER.warning("Échec Gio pendant une opération MTP : %s", error)
|
|
self._busy_drives.discard(record.safe_remove_runtime_id)
|
|
self.refresh()
|
|
if done is not None:
|
|
done(result_error or {"ok": True, "state": self.state()})
|
|
|
|
try:
|
|
return self.backend.operate_mtp_async(record.path, operation, completed)
|
|
except BaseException as error:
|
|
completed(None, error)
|
|
return lambda: None
|
|
|
|
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
|
|
self.refresh()
|
|
mtp_record = next(
|
|
(item for item in self._records if item.path == runtime_id and item.kind == "mtp"),
|
|
None,
|
|
)
|
|
if mtp_record is not None:
|
|
if operation == "open":
|
|
if "open" not in mtp_record.actions() or not mtp_record.mount_point:
|
|
raise RequestError("L'appareil MTP n'a pas fourni de chemin local")
|
|
self.launcher(mtp_record.mount_point)
|
|
return {"ok": True, "state": self.state()}
|
|
raise RequestError("Cette opération MTP exige une complétion asynchrone")
|
|
if operation == "safe-remove":
|
|
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()}
|
|
|
|
def handle_async(
|
|
self,
|
|
request: dict[str, Any],
|
|
done: Callable[[dict[str, Any] | BaseException], None],
|
|
) -> Callable[[], None]:
|
|
"""Lance une requête depuis GLib et remet sa réponse au callback."""
|
|
try:
|
|
operation, runtime_id = validate_request(request)
|
|
if operation == "list":
|
|
done(self.handle(request))
|
|
return lambda: None
|
|
assert runtime_id is not None
|
|
self.refresh()
|
|
record = next(
|
|
(
|
|
item
|
|
for item in self._records
|
|
if item.kind == "mtp"
|
|
and runtime_id in {item.path, item.safe_remove_runtime_id}
|
|
),
|
|
None,
|
|
)
|
|
if record is None:
|
|
done(self.handle(request))
|
|
return lambda: None
|
|
if operation not in record.actions():
|
|
raise RequestError("Opération indisponible pour cet appareil MTP")
|
|
if operation == "open":
|
|
if not record.mount_point:
|
|
raise RequestError("L'appareil MTP n'a pas fourni de chemin local")
|
|
self.launcher(record.mount_point)
|
|
done({"ok": True, "state": self.state()})
|
|
return lambda: None
|
|
return self._start_mtp_operation(record, operation, done)
|
|
except BaseException as error:
|
|
done(error)
|
|
return lambda: None
|
|
|
|
@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
|
|
valid_prefix = isinstance(runtime_id, str) and (
|
|
runtime_id.startswith("/org/freedesktop/UDisks2/")
|
|
or runtime_id.startswith("mtp:")
|
|
)
|
|
if not valid_prefix 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], Callable[[dict[str, Any] | BaseException], None]],
|
|
Callable[[], None],
|
|
],
|
|
) -> 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] = []
|
|
cancellation: list[Callable[[], None]] = []
|
|
|
|
def complete(value: dict[str, Any] | BaseException) -> None:
|
|
if not completed.is_set():
|
|
result.append(value)
|
|
completed.set()
|
|
|
|
def invoke() -> bool:
|
|
try:
|
|
cancellation.append(self.handler(request, complete))
|
|
except BaseException as error:
|
|
complete(error)
|
|
return False
|
|
|
|
self.GLib.idle_add(invoke)
|
|
if not completed.wait(OPERATION_TIMEOUT_SECONDS + 2):
|
|
if cancellation:
|
|
self.GLib.idle_add(cancellation[0])
|
|
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 = RemovableMediaBackend(
|
|
UDisksBackend(Gio, GLib), MTPBackend(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_async)
|
|
)
|
|
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())
|