diff --git a/README.md b/README.md index 4ada66b..31b9515 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,7 @@ Requires Python ≥ 3.9 and NAPALM ≥ 4.0. | `SwitchDriver` | Ethernet switches | Cisco IOS, Arista EOS, Juniper EX | | `FirewallDriver` | Firewalls & UTM appliances | pfSense, Fortinet FortiOS, Cisco ASA | | `HypervisorDriver` | Hypervisors & virtualisation platforms | Proxmox VE, VMware ESXi, KVM/libvirt | +| `OSDriver` | General-purpose operating systems | Linux, BSD, macOS | | `StorageDriver` | Storage appliances & NAS/SAN | TrueNAS, Synology DSM, QNAP QTS | ## Usage @@ -105,6 +106,34 @@ class ProxmoxDriver(HypervisorDriver): ... ``` +### OS / Linux + +```python +from napalm_device_types import OSDriver + +class LinuxDriver(OSDriver): + + def get_packages(self): + # return List[PackageDict] + ... + + def get_services(self): + # return List[ServiceDict] + ... + + def get_users(self): + # return List[UserDict] + ... + + def get_processes(self): + # return List[ProcessDict] + ... + + def get_cron_jobs(self): + # return List[CronJobDict] + ... +``` + ### Storage / NAS ```python diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index 8faa8aa..a006d6a 100644 --- a/napalm_device_types/__init__.py +++ b/napalm_device_types/__init__.py @@ -19,12 +19,14 @@ Available base classes: * :class:`~napalm_device_types.switch.SwitchDriver` * :class:`~napalm_device_types.firewall.FirewallDriver` * :class:`~napalm_device_types.hypervisor.HypervisorDriver` +* :class:`~napalm_device_types.os.OSDriver` * :class:`~napalm_device_types.storage.StorageDriver` """ from napalm_device_types.access_point import AccessPointDriver from napalm_device_types.firewall import FirewallDriver from napalm_device_types.hypervisor import HypervisorDriver +from napalm_device_types.os import OSDriver from napalm_device_types.storage import StorageDriver from napalm_device_types.switch import SwitchDriver @@ -32,6 +34,7 @@ __all__ = [ "AccessPointDriver", "FirewallDriver", "HypervisorDriver", + "OSDriver", "StorageDriver", "SwitchDriver", ] diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index 8005cd5..7e077f8 100644 --- a/napalm_device_types/models.py +++ b/napalm_device_types/models.py @@ -497,3 +497,117 @@ class ReplicationJobDict(TypedDict): last_run: float # Unix epoch; 0.0 if never run last_status: str # "success", "failed", "running", "pending" bytes_sent: int # bytes transferred in the last run + + +# --------------------------------------------------------------------------- +# OS / General-purpose Linux +# --------------------------------------------------------------------------- + + +class UserDict(TypedDict): + """A local OS user account.""" + + username: str + uid: int + gid: int + home: str + shell: str + groups: List[str] # all supplementary group names + + +class ProcessDict(TypedDict): + """A running OS process.""" + + pid: int + ppid: int + user: str + cpu: float # CPU utilisation 0.0–100.0 + memory: float # RSS percentage of total RAM 0.0–100.0 + vsz: int # virtual memory size in KiB + rss: int # resident set size in KiB + tty: str # controlling terminal; empty string if none + state: str # "R" running, "S" sleeping, "D" uninterruptible, "Z" zombie, etc. + started: str # start time string as reported by ``ps`` (e.g. "12:34" or "May28") + command: str # full command line + + +class CronJobDict(TypedDict): + """A scheduled cron task.""" + + user: str # owner of the crontab entry + schedule: str # five-field cron expression, e.g. "0 * * * *" + command: str # shell command to execute + description: NotRequired[str] # optional inline comment + + +class ApplyUpdatesResultDict(TypedDict): + """Result of an ``apply_updates()`` call.""" + + success: bool # True if the package manager exited without error + output: str # raw stdout/stderr captured from the package manager + error: NotRequired[str] # error message if success is False + + +class DockerContainerDict(TypedDict): + """A Docker container entry as returned by ``docker ps -a``.""" + + id: str # short container ID + name: str # container name(s) + image: str # image reference + image_version: str # OCI org.opencontainers.image.version label; empty if absent + command: str # entrypoint / command + created: str # creation timestamp string + status: str # human-readable status (e.g. "Up 3 hours") + ports: str # port mapping string + state: str # "running", "exited", "paused", etc. + + +class DockerImageDict(TypedDict): + """A local Docker image entry as returned by ``docker images``.""" + + id: str # short image ID + repository: str # image repository + tag: str # image tag + size: str # human-readable size string (e.g. "187MB") + created: str # creation timestamp string + version: str # OCI org.opencontainers.image.version label; empty if absent + + +class DockerVolumeDict(TypedDict): + """A Docker volume entry as returned by ``docker volume ls``.""" + + name: str # volume name + driver: str # volume driver (e.g. "local") + mountpoint: str # host filesystem path + scope: str # "local" or "global" + + +class DockerNetworkDict(TypedDict): + """A Docker network entry as returned by ``docker network ls``.""" + + id: str # short network ID + name: str # network name + driver: str # network driver (e.g. "bridge", "host", "overlay") + scope: str # network scope + ipv6: str # "true" if IPv6 is enabled, "false" otherwise + internal: str # "true" if the network is internal, "false" otherwise + + +class DockerInfoDict(TypedDict): + """Return value of ``get_docker_info()``.""" + + available: bool # False if Docker is absent/inaccessible + permission_denied: NotRequired[bool] # True when socket access is denied + version: NotRequired[str] # Docker Engine version string + containers: NotRequired[List[DockerContainerDict]] + images: NotRequired[List[DockerImageDict]] + volumes: NotRequired[List[DockerVolumeDict]] + networks: NotRequired[List[DockerNetworkDict]] + outdated_images: NotRequired[List[str]] # image names with a newer remote digest + + +class DeviceActionResultDict(TypedDict): + """Return value of ``run_device_action()``.""" + + success: bool # True if the action completed without error + output: str # human-readable output or status message diff --git a/napalm_device_types/os.py b/napalm_device_types/os.py new file mode 100644 index 0000000..2514bd5 --- /dev/null +++ b/napalm_device_types/os.py @@ -0,0 +1,375 @@ +""" +Abstract base class for OS / general-purpose server drivers. + +Usage:: + + from napalm_device_types import OSDriver + + class LinuxDriver(OSDriver): + def get_packages(self): + ... +""" + +from typing import List +from napalm.base import NetworkDriver +from napalm_device_types.models import ( + ApplyUpdatesResultDict, + CronJobDict, + DeviceActionResultDict, + DockerInfoDict, + PackageDict, + ProcessDict, + ServiceDict, + UpdateDict, + UserDict, +) + + +class OSDriver(NetworkDriver): + """ + Abstract intermediate driver for general-purpose operating systems + (e.g. Linux, BSD, macOS). + + Inherits all standard NAPALM NetworkDriver methods and adds OS-specific + operations that concrete drivers must implement. + """ + + # ------------------------------------------------------------------ + # Package management + # ------------------------------------------------------------------ + + def get_packages(self) -> List[PackageDict]: + """ + Returns a list of all installed software packages. + + Each entry contains: + + * name (string) - package name + * version (string) - installed version string + * installed (bool) - always ``True`` for this method + * description (string) - short package description + * size (int) - installed size in bytes; ``0`` if unavailable + * source (string) - package source / repository name; empty string if unavailable + + Example:: + + [ + { + "name": "openssh-server", + "version": "1:9.2p1-2+deb12u2", + "installed": True, + "description": "secure shell (SSH) server, for secure access from remote machines", + "size": 524288, + "source": "Debian", + }, + ] + """ + raise NotImplementedError + + def get_pending_updates(self) -> List[UpdateDict]: + """ + Returns a list of packages that have a newer version available. + + Each entry contains: + + * name (string) - package name + * current_version (string) - currently installed version + * new_version (string) - version available in the repository + + Example:: + + [ + { + "name": "openssh-server", + "current_version": "1:9.2p1-2+deb12u1", + "new_version": "1:9.2p1-2+deb12u2", + }, + ] + """ + raise NotImplementedError + + def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict: + """ + Upgrades the given packages to the newest available version. + + Only packages that are already installed may be upgraded; this method + does **not** install new packages. Pass an empty list to upgrade + **all** packages that have pending updates. + + :param packages: List of package names to upgrade. Each name must + match ``^[a-zA-Z0-9_\\-\\+\\.]+$``; a :exc:`ValueError` is raised + for any name that does not conform. + :returns: A dict with: + + * success (bool) – ``True`` if the package manager exited without error + * output (string) – combined stdout / stderr from the package manager + * error (string, optional) – short error message when *success* is ``False`` + + :raises ValueError: If any package name fails the safety check. + + Example:: + + result = driver.apply_updates(["openssh-server", "curl"]) + # → {"success": True, "output": "Reading package lists...\\n..."} + + # Upgrade everything: + result = driver.apply_updates([]) + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Service management + # ------------------------------------------------------------------ + + def get_services(self) -> List[ServiceDict]: + """ + Returns a list of system services and their current state. + + Each entry contains: + + * name (string) - service unit name (without ``.service`` suffix) + * running (bool) - ``True`` if the service is currently active + * enabled (bool) - ``True`` if the service starts automatically on boot + * pid (int) - main process ID; ``0`` if not running + + Example:: + + [ + { + "name": "ssh", + "running": True, + "enabled": True, + "pid": 1234, + }, + { + "name": "cron", + "running": True, + "enabled": True, + "pid": 5678, + }, + ] + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Users + # ------------------------------------------------------------------ + + def get_users(self) -> List[UserDict]: + """ + Returns a list of local OS user accounts. + + Each entry contains: + + * username (string) - login name + * uid (int) - numeric user ID + * gid (int) - primary group ID + * home (string) - home directory path + * shell (string) - login shell path + * groups (list of strings) - all supplementary group names + + Example:: + + [ + { + "username": "admin", + "uid": 1000, + "gid": 1000, + "home": "/home/admin", + "shell": "/bin/bash", + "groups": ["sudo", "docker", "adm"], + }, + ] + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Processes + # ------------------------------------------------------------------ + + def get_processes(self) -> List[ProcessDict]: + """ + Returns a snapshot of currently running processes. + + Each entry contains: + + * pid (int) - process ID + * ppid (int) - parent process ID + * user (string) - effective user name + * cpu (float) - CPU utilisation percentage + * memory (float) - RSS as a percentage of total RAM + * vsz (int) - virtual memory size in KiB + * rss (int) - resident set size in KiB + * tty (string) - controlling terminal; empty string if none + * state (string) - process state: ``"R"`` running, ``"S"`` sleeping, + ``"D"`` uninterruptible, ``"Z"`` zombie, ``"T"`` stopped, etc. + * started (string) - start time as printed by ``ps`` (e.g. ``"12:34"`` or ``"May28"``) + * command (string) - full command line + + Example:: + + [ + { + "pid": 1, + "ppid": 0, + "user": "root", + "cpu": 0.0, + "memory": 0.1, + "vsz": 168576, + "rss": 13312, + "tty": "", + "state": "S", + "started": "May28", + "command": "/sbin/init", + }, + ] + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Cron jobs + # ------------------------------------------------------------------ + + def get_cron_jobs(self) -> List[CronJobDict]: + """ + Returns scheduled cron tasks from all user crontabs and ``/etc/cron.d``. + + Each entry contains: + + * user (string) - owner of the crontab entry + * schedule (string) - five-field cron expression (e.g. ``"0 * * * *"``) + * command (string) - shell command to execute + * description (string, optional) - inline comment text if present + + Example:: + + [ + { + "user": "root", + "schedule": "0 4 * * *", + "command": "/usr/local/bin/backup.sh", + "description": "nightly backup", + }, + ] + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Docker + # ------------------------------------------------------------------ + + def get_docker_info(self) -> DockerInfoDict: + """ + Returns information about the local Docker environment. + + If Docker is not installed or the current user lacks access to the + Docker socket, returns ``{"available": False}``. When the user has + no socket permission, ``permission_denied`` is additionally set to + ``True``. + + When Docker is available the dict contains: + + * available (bool) - always ``True`` + * version (string) - Docker Engine version string + * containers (list) - all containers (running and stopped), each with: + + * id (string) - short container ID + * name (string) - container name(s) + * image (string) - image reference + * image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent + * command (string) - entrypoint / command string + * created (string) - creation timestamp string + * status (string) - human-readable status (e.g. ``"Up 3 hours"``) + * ports (string) - port mapping string + * state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc. + + * images (list) - local images, each with: + + * id (string) - short image ID + * repository (string) - image repository + * tag (string) - image tag + * size (string) - human-readable size string (e.g. ``"187MB"``) + * created (string) - creation timestamp string + * version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent + + * volumes (list) - Docker volumes, each with: + + * name (string) - volume name + * driver (string) - volume driver + * mountpoint (string) - host filesystem path + * scope (string) - ``"local"`` or ``"global"`` + + * networks (list) - Docker networks, each with: + + * id (string) - short network ID + * name (string) - network name + * driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``) + * scope (string) - network scope + * ipv6 (string) - ``"true"`` if IPv6 is enabled + * internal (string) - ``"true"`` if the network is internal + + * outdated_images (list of strings) - image names where the local digest + differs from the latest remote digest; empty list if all images are + current or update checks could not be performed. + + Example:: + + # Docker not installed: + {"available": False} + + # Docker installed, no socket permission: + {"available": False, "permission_denied": True} + + # Docker available: + { + "available": True, + "version": "Docker version 27.3.1, build ce12230", + "containers": [ + { + "id": "a1b2c3d4e5f6", + "name": "my-app", + "image": "nginx:latest", + "image_version": "1.27.0", + "command": "nginx -g 'daemon off;'", + "created": "2026-05-28 10:00:00 +0000 UTC", + "status": "Up 3 days", + "ports": "0.0.0.0:80->80/tcp", + "state": "running", + }, + ], + "images": [...], + "volumes": [], + "networks": [...], + "outdated_images": ["nginx:latest"], + } + """ + raise NotImplementedError + + # ------------------------------------------------------------------ + # Generic device actions + # ------------------------------------------------------------------ + + def run_device_action(self, action: str) -> DeviceActionResultDict: + """ + Executes a named administrative action on the device. + + This method is an extensibility point for driver-specific one-off + operations that do not fit any other NAPALM API method. Each driver + documents the action names it supports. + + :param action: Action identifier string (e.g. ``"fix_docker_permissions"``). + :raises NotImplementedError: If the driver does not implement this method. + :raises ValueError: If ``action`` is not a recognised action name for + this driver. + + :returns: A dict with: + + * success (bool) – ``True`` if the action completed without error + * output (string) – human-readable output or status message + + Example:: + + result = driver.run_device_action("fix_docker_permissions") + # → {"success": True, "output": "Added 'pi' to the docker group. Reconnect..."} + """ + raise NotImplementedError diff --git a/pyproject.toml b/pyproject.toml index 223cf36..b1b4086 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-device-types" -version = "0.2.0" +version = "0.3.0" description = "Abstract device-type base classes for NAPALM drivers" readme = "README.md" requires-python = ">=3.9"