# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 """NAPALM driver for OpenMediaVault NAS systems. Extends LinuxDriver with OMV-specific storage management: * Physical disk inventory via lsblk and smartctl * md RAID array status via /proc/mdstat and mdadm * Mounted filesystem info via df * SMB, NFS, and FTP share enumeration via omv-confdbadm Connects via SSH. Sudo access is required for smartctl and mdadm. All optional_args from LinuxDriver are accepted (pkg_manager, sudo_password, …). Tested against OpenMediaVault 6.x and 7.x (Debian Bullseye / Bookworm base). """ from __future__ import annotations import json as _json import re from typing import Any from napalm_device_types import FingerprintRule from napalm_device_types.models import ( DiskPoolDict, LogicalVolumeDict, NASShareDict, PhysicalDiskDict, ReplicationJobDict, StorageQuotaDict, StorageServiceDict, VolumeSnapshotDict, ) from napalm_device_types import StorageDriver from napalm_linux.linux import LinuxDriver class OpenMediaVaultDriver(StorageDriver, LinuxDriver): """NAPALM driver for OpenMediaVault NAS systems. Declares both roles it fills: a NAS built on Debian. Storage comes first, so netOrk shows it as storage rather than as a Linux host. Inherits all Linux OS functionality from LinuxDriver (packages, services, users, processes, cron, Docker) -- OMV really is Debian, so apt applies -- and adds OMV-specific storage management on top. NAS services are ``get_storage_services()``; the systemd service list stays ``get_services()``. Two names for two different things. """ TYPE_LABEL = "Storage" DRIVER_NAME = "openmediavault" driver_name = "openmediavault" SSH_FINGERPRINT = [ FingerprintRule("debian", weight=4.0), ] HTTP_FINGERPRINT = [ FingerprintRule("openmediavault", weight=9.0, mandatory=True), ] # ── Facts override ───────────────────────────────────────────────────────── def get_facts(self) -> dict[str, Any]: """Return device facts, overriding OMV version from dpkg.""" facts = super().get_facts() try: ver = self._send( "dpkg-query -W -f='${Version}' openmediavault 2>/dev/null" ).strip() if ver: facts["os_version"] = f"OMV {ver}" except Exception: pass return facts # ── Warnings ────────────────────────────────────────────────────────────── def get_device_warnings(self) -> list[dict[str, Any]]: """Return warnings, ignoring packages held back by apt-get upgrade. On OMV, kernel and OMV-specific packages are routinely held back (they require dist-upgrade or omv-upgrade). Reporting them as "updates available" after a successful apt-get upgrade is misleading. We simulate the upgrade first and only warn if there are packages that a standard apt-get upgrade would actually install. """ warnings: list[dict[str, Any]] = [] try: sim = self._sudo( "DEBIAN_FRONTEND=noninteractive apt-get upgrade --simulate 2>&1", read_timeout=60, ) m = re.search(r'(\d+) upgraded', sim) upgradable_count = int(m.group(1)) if m else 0 except Exception: upgradable_count = -1 # unknown — fall back to parent behaviour if upgradable_count == 0: return [] # nothing to do with apt-get upgrade; held-back only if upgradable_count < 0: return super().get_device_warnings() updates = self.get_available_updates() if updates: warnings.append({ "code": "updates_available", "meta": { "count": upgradable_count, "packages": [u.get("name", "") for u in updates[:10]], }, }) return warnings # ── Physical disks ───────────────────────────────────────────────────────── def get_disks(self) -> list[PhysicalDiskDict]: """Return physical disk inventory. Uses ``lsblk -J`` for device enumeration and ``smartctl -iA`` (via sudo) for health, temperature, and RPM data. Each entry contains: * slot (string) - device path, e.g. ``"/dev/sda"`` * model (string) - drive model string * serial (string) - drive serial number * vendor (string) - drive manufacturer * type (string) - ``"hdd"``, ``"ssd"``, or ``"nvme"`` * size (int) - raw capacity in bytes * rpm (int) - rotational speed; ``0`` for SSD/NVMe * temperature (int) - temperature in Celsius; ``-1`` if unavailable * health (string) - ``"healthy"``, ``"failed"``, or ``"unknown"`` * pool (string) - md array name if member; empty string otherwise Example:: [ { "slot": "/dev/sda", "model": "HGST HUS726T6TALE6L4", "serial": "K3GXXXXX", "vendor": "HGST", "type": "hdd", "size": 6001175126016, "rpm": 7200, "temperature": 34, "health": "healthy", "pool": "md1", }, ] """ result: list[PhysicalDiskDict] = [] try: raw = self._send( "lsblk -J -b -o NAME,SIZE,TYPE,ROTA,VENDOR,MODEL,SERIAL,TRAN 2>/dev/null" ) lsblk = _json.loads(raw) except Exception: return result for dev in lsblk.get("blockdevices", []): if dev.get("type") != "disk": continue name = dev.get("name", "") if not name: continue size = int(dev.get("size") or 0) rota = str(dev.get("rota", "1")) vendor = (dev.get("vendor") or "").strip() model = (dev.get("model") or "").strip() serial = (dev.get("serial") or "").strip() tran = (dev.get("tran") or "").strip().lower() if tran == "nvme": disk_type = "nvme" elif rota == "0": disk_type = "ssd" else: disk_type = "hdd" temperature = -1 health = "unknown" rpm = 0 try: smart = self._sudo( f"smartctl -iA /dev/{name} 2>/dev/null", read_timeout=15 ) if "PASSED" in smart: health = "healthy" elif "FAILED" in smart: health = "failed" temp_m = re.search( r"(?:Temperature_Celsius|Current Temperature|Temperature:)" r"[^\n]*?(\d+)\s*(?:Celsius|\(|$)", smart, ) if temp_m: temperature = int(temp_m.group(1)) if disk_type == "hdd": rpm_m = re.search(r"Rotation Rate:\s+(\d+)\s+rpm", smart) if rpm_m: rpm = int(rpm_m.group(1)) except Exception: pass result.append(PhysicalDiskDict( slot=f"/dev/{name}", model=model, serial=serial, vendor=vendor, type=disk_type, size=size, rpm=rpm, temperature=temperature, health=health, pool=self._disk_to_pool(name), )) return result def _disk_to_pool(self, disk_name: str) -> str: """Return the md array a disk belongs to, or empty string.""" try: mdstat = self._send("cat /proc/mdstat 2>/dev/null") for line in mdstat.splitlines(): if disk_name in line and re.match(r'^md\d+', line): return line.split(":")[0].strip() except Exception: pass return "" # ── md RAID pools ────────────────────────────────────────────────────────── def get_disk_pools(self) -> dict[str, DiskPoolDict]: """Return md RAID array status from ``/proc/mdstat`` and ``mdadm --detail``. Keys are array names (e.g. ``"md1"``). Each value contains: * name (string) - array name * type (string) - always ``"md"`` * level (string) - RAID level, e.g. ``"raid1"``, ``"raid5"`` * status (string) - ``"online"``, ``"degraded"``, ``"faulted"``, or ``"offline"`` * total (int) - usable array size in bytes (``0`` if mdadm unavailable) * used (int) - always ``0`` (use ``get_volumes()`` for per-filesystem usage) * available (int) - always ``0`` * disks (list of strings) - member device names (e.g. ``["sda1", "sdb1"]``) * auto_expand (bool) - always ``False`` (not applicable for md) * dedup (bool) - always ``False`` * compression (string) - always ``"off"`` Example:: { "md1": { "name": "md1", "type": "md", "level": "raid1", "status": "online", "total": 2000398925824, "used": 0, "available": 0, "disks": ["sda1", "sdb1"], "auto_expand": False, "dedup": False, "compression": "off", }, } """ result: dict[str, DiskPoolDict] = {} try: mdstat = self._send("cat /proc/mdstat 2>/dev/null") except Exception: return result for line in mdstat.splitlines(): m = re.match(r'^(md\d+)\s*:\s+(\S+)\s+(\S+)\s+(.*)', line) if not m: continue md_name, state_word, level_raw, members_raw = m.groups() member_disks = re.findall(r'(\w+)\[\d+\]', members_raw) status = "online" if state_word == "active" else "offline" total = 0 try: detail = self._sudo( f"mdadm --detail /dev/{md_name} 2>/dev/null", read_timeout=10 ) size_m = re.search(r"Array Size\s*:\s*(\d+)\s*\(", detail) if size_m: total = int(size_m.group(1)) * 1024 # kB → bytes if "degraded" in detail.lower(): status = "degraded" elif re.search(r"State\s*:.*(?:faulty|failed)", detail, re.IGNORECASE): status = "faulted" except Exception: pass result[md_name] = DiskPoolDict( name=md_name, type="md", level=level_raw.lower(), status=status, total=total, used=0, available=0, disks=member_disks, auto_expand=False, dedup=False, compression="off", ) return result # ── Logical volumes / filesystems ────────────────────────────────────────── def get_volumes(self) -> dict[str, LogicalVolumeDict]: """Return mounted filesystem info from ``df``. Only real block devices (``/dev/…``) are included; tmpfs, devtmpfs, overlays, and similar pseudo-filesystems are skipped. Keys are derived from the mountpoint (e.g. ``"srv-dev-disk-by-label-data"`` for ``/srv/dev/disk/by-label/data``; ``"root"`` for ``/``). Each value contains: * name (string) - key repeated for convenience * pool (string) - device name without ``/dev/`` prefix * type (string) - always ``"filesystem"`` * total (int) - total size in bytes * used (int) - used space in bytes * available (int) - available space in bytes * mountpoint (string) - absolute mount path * compression (string) - always ``"off"`` * dedup (bool) - always ``False`` * readonly (bool) - ``True`` if mounted ``ro`` * snapshots (int) - always ``0`` Example:: { "srv-data": { "name": "srv-data", "pool": "md1", "type": "filesystem", "total": 2000398925824, "used": 945000000000, "available": 1055398925824, "mountpoint": "/srv/data", "compression": "off", "dedup": False, "readonly": False, "snapshots": 0, }, } """ result: dict[str, LogicalVolumeDict] = {} try: df_out = self._send("df -Pk 2>/dev/null") except Exception: return result try: mounts = self._send("cat /proc/mounts 2>/dev/null") except Exception: mounts = "" for line in df_out.splitlines()[1:]: parts = line.split() if len(parts) < 6: continue device = parts[0] total_k, used_k, avail_k = int(parts[1]), int(parts[2]), int(parts[3]) mountpoint = parts[5] if not device.startswith("/dev/"): continue key = mountpoint.lstrip("/").replace("/", "-") or "root" pool = device.removeprefix("/dev/") readonly = bool(re.search( rf'^{re.escape(device)}\s+\S+\s+\S+\s+(?:[^,\s]*,)*ro(?:,|\s|$)', mounts, re.MULTILINE, )) result[key] = LogicalVolumeDict( name=key, pool=pool, type="filesystem", total=total_k * 1024, used=used_k * 1024, available=avail_k * 1024, mountpoint=mountpoint, compression="off", dedup=False, readonly=readonly, snapshots=0, ) return result # ── NAS shares ──────────────────────────────────────────────────────────── def get_shares(self) -> dict[str, NASShareDict]: """Return all configured NAS shares from the OMV config database. Covers SMB/CIFS, NFS, and FTP shares. Keys are prefixed with the protocol (e.g. ``"smb:media"``, ``"nfs:backups"``, ``"ftp:upload"``). Each value contains: * name (string) - share name without protocol prefix * protocol (string) - ``"smb"``, ``"nfs"``, or ``"ftp"`` * path (string) - absolute filesystem path * volume (string) - last path component (leaf directory) * enabled (bool) - whether the share is currently exported * readonly (bool) - whether the share is exported read-only * description (string) - human-readable comment (SMB only) * clients (list of strings) - allowed client addresses (NFS only) Example:: { "smb:media": { "name": "media", "protocol": "smb", "path": "/srv/dev/disk/by-label/data/media", "volume": "media", "enabled": True, "readonly": False, "description": "Media library", "clients": [], }, "nfs:backups": { "name": "backups", "protocol": "nfs", "path": "/srv/dev/disk/by-label/data/backups", "volume": "backups", "enabled": True, "readonly": False, "description": "", "clients": ["192.168.1.0/24"], }, } """ shares: dict[str, NASShareDict] = {} shares.update(self._smb_shares()) shares.update(self._nfs_shares()) shares.update(self._ftp_shares()) return shares # ── Storage services ────────────────────────────────────────────────────── def get_storage_services(self) -> dict[str, StorageServiceDict]: """Return status of key NAS services (SMB, NFS, FTP, SSH, rsync). This method is separate from ``get_services()`` (inherited from LinuxDriver) which returns the full systemd service list. Keys are service names. Each value contains: * name (string) - service name * enabled (bool) - configured to start on boot * running (bool) - currently active * port (int) - primary listening port * version (string) - protocol version; empty if unknown Example:: { "smb": {"name": "smb", "enabled": True, "running": True, "port": 445, "version": ""}, "nfs": {"name": "nfs", "enabled": True, "running": True, "port": 2049, "version": ""}, "ftp": {"name": "ftp", "enabled": False,"running": False, "port": 21, "version": ""}, "ssh": {"name": "ssh", "enabled": True, "running": True, "port": 22, "version": ""}, "rsync": {"name": "rsync", "enabled": False,"running": False, "port": 873, "version": ""}, } """ _SVC_MAP = [ # (key, systemd unit, default port) ("smb", "smbd", 445), ("nfs", "nfs-server", 2049), ("ftp", "proftpd", 21), ("ssh", "ssh", 22), ("rsync", "rsync", 873), ] result: dict[str, StorageServiceDict] = {} for key, unit, port in _SVC_MAP: try: running = ( self._send(f"systemctl is-active {unit} 2>/dev/null || true").strip() == "active" ) enabled = ( self._send(f"systemctl is-enabled {unit} 2>/dev/null || true").strip() == "enabled" ) except Exception: running = enabled = False result[key] = StorageServiceDict( name=key, enabled=enabled, running=running, port=port, version="", ) return result # ── Not implemented ──────────────────────────────────────────────────────── def get_volume_snapshots(self, volume: str = "") -> list[VolumeSnapshotDict]: """Return volume snapshots (not supported on OpenMediaVault).""" raise NotImplementedError( "OpenMediaVault does not expose volume snapshots via this driver" ) def get_replication_jobs(self) -> list[ReplicationJobDict]: """Return replication jobs (not supported on OpenMediaVault).""" raise NotImplementedError def get_quotas(self) -> list[StorageQuotaDict]: """Return storage quotas (not supported on OpenMediaVault).""" raise NotImplementedError # ── Internal helpers ─────────────────────────────────────────────────────── def _omv_read(self, key: str) -> list: """Query the OMV config database and return a list of objects. Handles both OMV 6+ (returns a JSON array directly) and older versions that wrap the array in a dict under a common key. """ try: raw = self._send(f"omv-confdbadm read {key!r} 2>/dev/null").strip() if not raw: return [] data = _json.loads(raw) if isinstance(data, list): return data # Older OMV wraps the list: {"share": [...]} for wrapper in ("share", "rule", "entry", "mntent", "sharedfolder"): if wrapper in data and isinstance(data[wrapper], list): return data[wrapper] return [] except Exception: return [] def _resolve_shared_folder(self, uuid: str) -> str: """Resolve an OMV shared folder UUID to its absolute filesystem path.""" if not uuid: return "" try: for folder in self._omv_read("conf.system.sharedfolder"): if folder.get("uuid") != uuid: continue rel = folder.get("reldirpath", "").strip("/") mnt_uuid = folder.get("mntentref", "") mnt_dir = "" for mntent in self._omv_read("conf.system.filesystem.mountpoint"): if mntent.get("uuid") == mnt_uuid: mnt_dir = mntent.get("dir", "").rstrip("/") break return f"{mnt_dir}/{rel}".rstrip("/") if rel else mnt_dir except Exception: pass return "" def _smb_shares(self) -> dict[str, NASShareDict]: out: dict[str, NASShareDict] = {} for s in self._omv_read("conf.service.smb.shares"): name = s.get("name", "") if not name: continue path = self._resolve_shared_folder(s.get("sharedfolderref", "")) out[f"smb:{name}"] = NASShareDict( name=name, protocol="smb", path=path, volume=path.split("/")[-1] if path else "", enabled=bool(s.get("enable", True)), readonly=bool(s.get("readonly", False)), description=s.get("comment", ""), clients=[], ) return out def _nfs_shares(self) -> dict[str, NASShareDict]: out: dict[str, NASShareDict] = {} for s in self._omv_read("conf.service.nfs.shares"): path = self._resolve_shared_folder(s.get("sharedfolderref", "")) name = path.split("/")[-1] or s.get("sharedfolderref", "") clients_raw = s.get("client", "") or "" clients = [c.strip() for c in clients_raw.split(",") if c.strip()] opts = s.get("options", "") or "" out[f"nfs:{name}"] = NASShareDict( name=name, protocol="nfs", path=path, volume=path.split("/")[-1] if path else "", enabled=bool(s.get("enable", True)), readonly="ro" in [o.strip() for o in opts.split(",")], description="", clients=clients, ) return out def _ftp_shares(self) -> dict[str, NASShareDict]: out: dict[str, NASShareDict] = {} for s in self._omv_read("conf.service.ftp.shares"): path = self._resolve_shared_folder(s.get("sharedfolderref", "")) name = path.split("/")[-1] or s.get("sharedfolderref", "") out[f"ftp:{name}"] = NASShareDict( name=name, protocol="ftp", path=path, volume=path.split("/")[-1] if path else "", enabled=bool(s.get("enable", True)), readonly=bool(s.get("readonly", False)), description="", clients=[], ) return out