feat: add OSDriver base class with Docker and device action APIs
- Introduce OSDriver abstract base class (os.py) for general-purpose OS drivers (Linux, BSD, macOS) — registers package management, service management, users, processes, cron job and the two new OS-specific extension points - Add get_docker_info() contract: returns DockerInfoDict covering containers, images, volumes, networks and outdated image detection - Add run_device_action() contract: generic extensibility point for driver-specific one-off administrative actions - Fix duplicate TypedDicts in models.py: remove early shadow definitions of UserDict, ProcessDict, CronJobDict, ApplyUpdatesResultDict from the Common section; keep the more complete definitions in the OS section - Add Docker TypedDicts: DockerContainerDict, DockerImageDict, DockerVolumeDict, DockerNetworkDict, DockerInfoDict - Add DeviceActionResultDict - Export OSDriver from package __init__; bump version to 0.3.0 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.6
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user