Christian Manivong
aba36b3b11
driver / standard (pull_request) Successful in 15s
driver / lint (pull_request) Successful in 12s
driver / types (pull_request) Successful in 55s
driver / test (3.10) (pull_request) Successful in 58s
driver / test (3.11) (pull_request) Successful in 56s
driver / lowest (pull_request) Successful in 26s
driver / test (3.12) (pull_request) Successful in 55s
driver / test (3.13) (pull_request) Successful in 48s
driver / build (pull_request) Successful in 24s
driver / security (pull_request) Successful in 58s
CI / driver (pull_request) Successful in 3m0s
driver / publish (pull_request) Skipped
- CI moves to .gitea/workflows/ci.yml and calls the shared driver workflow of NAPALM/driver-standard; napalm-device-types and napalm-linux come from git until napalm-linux 0.15.0 is on PyPI. - requires-python >=3.10 (napalm-device-types needs it), SPDX license string, setuptools>=77, LICENSE file, py.typed, links to git.netork.io. - Dependencies: napalm-device-types>=4.22.0, napalm-linux>=0.15.0, napalm>=4.0 (tested at 4.0.0); dev extra with pytest, pytest-cov, mypy and ruff. - mypy strict for the whole package, no exempt module: _omv_read is annotated as returning list[dict[str, Any]] (types only). - Coverage floor 15 % (measured 15.89 %); the four long getters and the complexity of get_disks are recorded in the quality baseline. - README (what it is, how it connects, the getters, what it does not do) and CHANGELOG are new. - The committed egg-info and __pycache__ files are untracked; .gitignore from the template keeps them, .venv and build output out.
609 lines
23 KiB
Python
609 lines
23 KiB
Python
# 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 logging
|
|
import re
|
|
from typing import Any
|
|
|
|
from napalm_device_types import FingerprintRule, StorageDriver
|
|
from napalm_device_types.models import (
|
|
DiskPoolDict,
|
|
LogicalVolumeDict,
|
|
NASShareDict,
|
|
PhysicalDiskDict,
|
|
ReplicationJobDict,
|
|
StorageQuotaDict,
|
|
StorageServiceDict,
|
|
VolumeSnapshotDict,
|
|
)
|
|
from napalm_linux.linux import LinuxDriver
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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"
|
|
#: "Upgrade everything" must be a full upgrade: OMV's own packages and the
|
|
#: kernel are held back by a plain ``apt-get upgrade`` (``omv-upgrade`` is a
|
|
#: dist-upgrade). netOrk reads this when it upgrades the host.
|
|
FULL_UPGRADE = True
|
|
driver_name = "openmediavault"
|
|
SSH_FINGERPRINT = [
|
|
FingerprintRule("debian", weight=4.0),
|
|
]
|
|
HTTP_FINGERPRINT = [
|
|
FingerprintRule("openmediavault", weight=9.0, mandatory=True),
|
|
]
|
|
|
|
#: OMV generates sshd_config from its own database; a drop-in netOrk wrote
|
|
#: would be dropped on the next apply of the SSH settings. Inherited from
|
|
#: napalm-linux, so opted out here (NetOrk/netork#1027).
|
|
harden_ssh_server = None # type: ignore[assignment]
|
|
|
|
# ── 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:
|
|
logger.debug("%s: OMV version not read: ignored", self.hostname, exc_info=True)
|
|
return facts
|
|
|
|
# ── 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:
|
|
logger.debug("%s: smartctl for %s: ignored", self.hostname, name, exc_info=True)
|
|
|
|
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:
|
|
logger.debug("%s: /proc/mdstat not read: ignored", self.hostname, exc_info=True)
|
|
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:
|
|
logger.debug(
|
|
"%s: mdadm --detail for %s: ignored", self.hostname, md_name, exc_info=True
|
|
)
|
|
|
|
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[dict[str, Any]]:
|
|
"""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):
|
|
wrapped: list[dict[str, Any]] = data[wrapper]
|
|
return wrapped
|
|
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:
|
|
logger.debug("%s: shared folder %s: ignored", self.hostname, uuid, exc_info=True)
|
|
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
|