Files
napalm-windows/napalm_windows/windows.py
T
Christian Manivong d43d697700
driver / standard (pull_request) Successful in 15s
driver / lint (pull_request) Successful in 11s
driver / types (pull_request) Successful in 44s
driver / test (3.10) (pull_request) Successful in 46s
driver / test (3.11) (pull_request) Successful in 45s
driver / test (3.12) (pull_request) Successful in 42s
driver / test (3.13) (pull_request) Successful in 38s
driver / lowest (pull_request) Successful in 22s
driver / security (pull_request) Successful in 45s
driver / build (pull_request) Successful in 22s
CI / driver (pull_request) Successful in 5m30s
driver / publish (pull_request) Skipped
chore: adopt the NAPALM driver standard
- CI moves to .gitea/workflows/ci.yml and calls the shared driver workflow
  of NAPALM/driver-standard; .github/ is gone.
- requires-python >=3.10 (napalm-device-types needs it), LICENSE file,
  py.typed, setuptools>=77 with the SPDX license string, Issues and
  Changelog links on git.netork.io, napalm-device-types>=4.21.1 from PyPI,
  floors on the dev extra.
- mypy strict for the whole package, no module exempt: optional_args is
  dict[str, Any] | None.
- tests/quality_baseline.json records get_facts and get_route_to, over
  complexity 10. Coverage floor 95 % (measured 95.67 %).
- CHANGELOG.md in Keep a Changelog form; the README says no host output is
  recorded yet, links napalm-device-types in the NAPALM org and pypsrp on
  PyPI.
2026-10-23 08:49:27 +02:00

537 lines
19 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# Licensed under the Apache License, Version 2.0
"""NAPALM driver for Windows hosts.
Talks PowerShell Remoting over WinRM (see :mod:`napalm_windows.transport`).
Each getter sends one script that projects its cmdlet results onto flat,
primitive fields and ends in ``ConvertTo-Json``; the Python side only
normalises formats. Two Windows PowerShell 5.1 quirks shape every script:
* ``ConvertTo-Json`` writes enums as integers and ``DateTime`` as
``"\\/Date(...)\\/"`` — so enums are cast with ``[string]`` and times are
converted to numbers before they leave the host.
* A one-element array arrives as a bare object — so list results go through
``-InputObject @(...)``, and :func:`_as_list` guards the rest.
"""
from __future__ import annotations
import ipaddress
import json
import logging
import re
from typing import Any
from napalm.base.exceptions import ConnectionClosedException
from napalm_device_types import FingerprintRule, OSDriver, PortSpec
from napalm_device_types.models import ListeningSocketDict, ListeningSocketsDict, ServiceDict
from napalm_windows.transport import PowerShellError, PsrpTransport
logger = logging.getLogger(__name__)
WINRM_HTTP_PORT = 5985
WINRM_HTTPS_PORT = 5986
# ---------------------------------------------------------------------------
# PowerShell scripts — one per getter
# ---------------------------------------------------------------------------
_PS_FACTS = r"""
$ErrorActionPreference = 'Stop'
$cs = Get-CimInstance -ClassName Win32_ComputerSystem
$os = Get-CimInstance -ClassName Win32_OperatingSystem
$bios = Get-CimInstance -ClassName Win32_BIOS
$cv = Get-ItemProperty -ErrorAction SilentlyContinue `
-Path 'HKLM:\SOFTWARE\Microsoft\Windows NT\CurrentVersion'
[pscustomobject]@{
hostname = $env:COMPUTERNAME
dns_hostname = $cs.DNSHostName
domain = $cs.Domain
part_of_domain = [bool]$cs.PartOfDomain
manufacturer = $cs.Manufacturer
model = $cs.Model
serial = $bios.SerialNumber
caption = $os.Caption
version = $os.Version
display_version = $cv.DisplayVersion
ubr = $cv.UBR
uptime = [int64]((Get-Date) - $os.LastBootUpTime).TotalSeconds
interfaces = @(Get-NetAdapter | ForEach-Object { $_.Name })
} | ConvertTo-Json -Compress -Depth 3
"""
_PS_INTERFACES = r"""
$ErrorActionPreference = 'Stop'
ConvertTo-Json -Compress -Depth 3 -InputObject @(
Get-NetAdapter | ForEach-Object {
[pscustomobject]@{
name = $_.Name
description = $_.InterfaceDescription
status = [string]$_.Status
mtu = $_.MtuSize
speed = $_.Speed
mac = $_.MacAddress
}
}
)
"""
_PS_INTERFACES_IP = r"""
$ErrorActionPreference = 'Stop'
ConvertTo-Json -Compress -Depth 3 -InputObject @(
Get-NetIPAddress | ForEach-Object {
[pscustomobject]@{
interface = $_.InterfaceAlias
address = $_.IPAddress
prefix = [int]$_.PrefixLength
family = [string]$_.AddressFamily
}
}
)
"""
_PS_ARP = r"""
$ErrorActionPreference = 'Stop'
ConvertTo-Json -Compress -Depth 3 -InputObject @(
Get-NetNeighbor -AddressFamily IPv4 | ForEach-Object {
[pscustomobject]@{
interface = $_.InterfaceAlias
ip = $_.IPAddress
mac = $_.LinkLayerAddress
state = [string]$_.State
}
}
)
"""
_PS_ROUTES = r"""
$ErrorActionPreference = 'Stop'
ConvertTo-Json -Compress -Depth 3 -InputObject @(
Get-NetRoute | ForEach-Object {
[pscustomobject]@{
destination = $_.DestinationPrefix
next_hop = $_.NextHop
interface = $_.InterfaceAlias
metric = [int]$_.RouteMetric
protocol = [string]$_.Protocol
}
}
)
"""
_PS_SERVICES = r"""
$ErrorActionPreference = 'Stop'
ConvertTo-Json -Compress -Depth 3 -InputObject @(
Get-CimInstance -ClassName Win32_Service | ForEach-Object {
[pscustomobject]@{
name = $_.Name
state = [string]$_.State
start_mode = [string]$_.StartMode
pid = [int]$_.ProcessId
}
}
)
"""
# Every listening TCP and bound UDP socket, the processes holding them and the
# services those processes run. Sockets owned by http.sys (WinRM, IIS) belong to
# the System process; a shared svchost runs several services in one process.
_PS_LISTENING = r"""
$ErrorActionPreference = 'Stop'
$tcp = @(Get-NetTCPConnection -State Listen -ErrorAction SilentlyContinue)
$udp = @(Get-NetUDPEndpoint -ErrorAction SilentlyContinue)
$ids = @(@($tcp) + @($udp) | ForEach-Object { [int]$_.OwningProcess } | Sort-Object -Unique)
ConvertTo-Json -Compress -Depth 3 -InputObject ([pscustomobject]@{
tcp = @($tcp | ForEach-Object {
[pscustomobject]@{
address = [string]$_.LocalAddress
port = [int]$_.LocalPort
pid = [int]$_.OwningProcess
}
})
udp = @($udp | ForEach-Object {
[pscustomobject]@{
address = [string]$_.LocalAddress
port = [int]$_.LocalPort
pid = [int]$_.OwningProcess
}
})
processes = @(if ($ids.Count) {
Get-Process -Id $ids -ErrorAction SilentlyContinue | ForEach-Object {
[pscustomobject]@{ pid = [int]$_.Id; name = $_.ProcessName }
}
})
services = @(
Get-CimInstance -ClassName Win32_Service -Filter 'ProcessId <> 0' | ForEach-Object {
[pscustomobject]@{ pid = [int]$_.ProcessId; name = $_.Name }
}
)
})
"""
_SERVICE_COMMANDS = {
"start": "Start-Service -Name {name}",
"stop": "Stop-Service -Name {name}",
"restart": "Restart-Service -Name {name}",
"enable": "Set-Service -Name {name} -StartupType Automatic",
"disable": "Set-Service -Name {name} -StartupType Disabled",
}
# Service key names, not display names: letters, digits and the few symbols
# real services use (MSSQL$SQLEXPRESS, CDPUserSvc_4a5b6, {GUID}-style names).
_SERVICE_NAME_RE = re.compile(r"^[\w.$@{}\-]+$", re.ASCII)
# PowerShell accepts these as single-quote delimiters alongside the ASCII one.
_PS_SINGLE_QUOTES = ("'", "‘", "’", "‚", "‛")
_ROUTE_PROTOCOLS = {
"local": "connected",
"routeradvertisement": "connected",
"netmgmt": "static",
"dhcp": "dhcp",
"ospf": "ospf",
"bgp": "bgp",
"rip": "rip",
}
_ENABLED_START_MODES = {"auto", "boot", "system"}
_DEAD_NEIGHBOUR_STATES = {"unreachable", "incomplete"}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _as_list(value: Any) -> list[Any]:
"""Undo ConvertTo-Json unwrapping a one-element array into a bare value."""
if value is None:
return []
if isinstance(value, list):
return value
return [value]
def _ps_quote(value: str) -> str:
"""Quote ``value`` as a PowerShell single-quoted (verbatim) string."""
for q in _PS_SINGLE_QUOTES:
value = value.replace(q, q + q)
return f"'{value}'"
def _mac(value: str | None) -> str:
"""``00-15-5d-01-02-03`` → ``00:15:5D:01:02:03``."""
if not value:
return ""
return value.replace("-", ":").upper()
def _is_real_neighbour_mac(mac: str) -> bool:
if not mac or mac == "00:00:00:00:00:00":
return False
# Broadcast and multicast share the group bit in the first octet.
return not int(mac.split(":", 1)[0], 16) & 1
def _listening_socket(
proto: str, entry: dict[str, Any], names: dict[int, str], owners: dict[int, list[str]]
) -> ListeningSocketDict:
"""One socket of ``_PS_LISTENING``. A process running one service names it; a
shared svchost runs several, and which of them holds the socket Windows does
not say. PID 0 is no process."""
pid = int(entry.get("pid") or 0)
address, _, zone = str(entry.get("address") or "").partition("%")
services = owners.get(pid, [])
return {
"proto": proto,
"address": address,
"port": int(entry.get("port") or 0),
"interface": zone or None,
"process": names.get(pid) if pid else None,
"pid": pid or None,
"unit": services[0] if len(services) == 1 else None,
"container_id": None,
}
def _is_noise_route(destination: str) -> bool:
"""Multicast, limited broadcast and loopback — present on every host."""
try:
net = ipaddress.ip_network(destination, strict=False)
except ValueError:
return False
if net.is_multicast or net.is_loopback:
return True
return str(net) == "255.255.255.255/32"
# ---------------------------------------------------------------------------
# Driver
# ---------------------------------------------------------------------------
class WindowsDriver(OSDriver):
"""NAPALM driver for Windows Server 2016+ and Windows 10/11.
``optional_args``:
* ``port`` — WinRM port, default 5986 (HTTPS).
* ``winrm_ssl`` — force HTTPS on/off; default is HTTPS unless port is 5985.
* ``winrm_auth`` — pypsrp auth protocol, default ``negotiate``.
* ``ssl_verify`` — validate the WinRM certificate, default ``True``.
"""
TYPE_LABEL = "Windows"
VENDOR = "Microsoft"
DRIVER_NAME = "windows"
USES_SSH = False
# Updates install during shutdown and boot; a Windows reboot routinely
# takes several minutes rather than a Linux host's ninety seconds.
REBOOT_SETTLE_SECONDS = 180
#: Read by netOrk when the device has no port of its own.
default_port = WINRM_HTTPS_PORT
PORT_SPECS = [
PortSpec("http", WINRM_HTTP_PORT, ("/wsman",), weight=6.0),
PortSpec("https", WINRM_HTTPS_PORT, ("/wsman",), weight=6.0),
]
# WinRM answers through http.sys: "Server: Microsoft-HTTPAPI/2.0". Discovery
# only reaches 5985 when port 80 is silent, so an IIS on 80 is the header
# it sees instead.
HTTP_FINGERPRINT = [
FingerprintRule("microsoft-httpapi", weight=5.0),
FingerprintRule("microsoft-iis", weight=3.0),
]
# sysDescr: "Hardware: ... Software: Windows Version 6.3 (Build 17763 ...)"
SNMP_FINGERPRINT = [
FingerprintRule("software: windows", weight=8.0),
]
# Banner: "SSH-2.0-OpenSSH_for_Windows_8.1"
SSH_FINGERPRINT = [
FingerprintRule("openssh_for_windows", weight=8.0),
]
def __init__(
self,
hostname: str,
username: str,
password: str,
timeout: int = 60,
optional_args: dict[str, Any] | None = None,
) -> None:
self.hostname = hostname
self.username = username
self.password = password
self.timeout = timeout
optional_args = optional_args or {}
self.port: int = int(optional_args.get("port") or WINRM_HTTPS_PORT)
self.ssl: bool = bool(optional_args.get("winrm_ssl", self.port != WINRM_HTTP_PORT))
self.cert_validation: bool = bool(optional_args.get("ssl_verify", True))
self.auth: str = optional_args.get("winrm_auth", "negotiate")
self._transport: PsrpTransport | None = None
# ------------------------------------------------------------------
# Connection management
# ------------------------------------------------------------------
def open(self) -> None:
transport = PsrpTransport(
self.hostname,
self.username,
self.password,
port=self.port,
ssl=self.ssl,
cert_validation=self.cert_validation,
auth=self.auth,
timeout=self.timeout,
)
transport.open()
self._transport = transport
def close(self) -> None:
if self._transport is not None:
self._transport.close()
self._transport = None
def is_alive(self) -> dict[str, bool]:
return {"is_alive": bool(self._transport and self._transport.is_open)}
def _run_ps(self, script: str) -> Any:
"""Run ``script`` and return its JSON output parsed, or ``None`` if empty."""
out = self._run_raw(script).strip()
return json.loads(out) if out else None
def _run_raw(self, script: str) -> str:
if self._transport is None:
raise ConnectionClosedException("WinRM connection is not open")
return self._transport.run(script)
# ------------------------------------------------------------------
# NAPALM getters
# ------------------------------------------------------------------
def get_facts(self) -> dict[str, Any]:
raw = self._run_ps(_PS_FACTS) or {}
hostname = raw.get("dns_hostname") or raw.get("hostname") or ""
domain = raw.get("domain") or ""
fqdn = f"{hostname}.{domain}" if raw.get("part_of_domain") and domain else hostname
version = raw.get("version") or ""
ubr = raw.get("ubr")
build = version.rsplit(".", 1)[-1] if version else ""
if ubr is not None:
build = f"{build}.{ubr}"
os_version = " ".join(
p for p in ((raw.get("caption") or "").strip(), raw.get("display_version")) if p
)
if build:
os_version = f"{os_version} (build {build})"
return {
"hostname": hostname,
"fqdn": fqdn,
"vendor": raw.get("manufacturer") or self.VENDOR,
"model": raw.get("model") or "",
"serial_number": raw.get("serial") or "",
"os_version": os_version,
"uptime": int(raw.get("uptime") or 0),
"interface_list": [str(i) for i in _as_list(raw.get("interfaces"))],
# Full build including the update revision — what CVE matching
# needs, since Windows fixes ship as UBR bumps, not version bumps.
"running_kernel": f"{version}.{ubr}" if ubr is not None else version,
}
def get_interfaces(self) -> dict[str, dict[str, Any]]:
result: dict[str, dict[str, Any]] = {}
for a in _as_list(self._run_ps(_PS_INTERFACES)):
status = (a.get("status") or "").lower()
result[a["name"]] = {
"is_up": status == "up",
"is_enabled": status != "disabled",
"description": a.get("description") or "",
"last_flapped": -1.0,
"speed": float(a.get("speed") or 0) / 1_000_000,
"mtu": int(a.get("mtu") or 0),
"mac_address": _mac(a.get("mac")),
}
return result
def get_interfaces_ip(self) -> dict[str, dict[str, dict[str, dict[str, int]]]]:
result: dict[str, dict[str, dict[str, dict[str, int]]]] = {}
for entry in _as_list(self._run_ps(_PS_INTERFACES_IP)):
# Link-local IPv6 carries its zone index ("fe80::1%6").
address = (entry.get("address") or "").split("%", 1)[0]
try:
if ipaddress.ip_address(address).is_loopback:
continue
except ValueError:
continue
family = "ipv6" if (entry.get("family") or "").lower() == "ipv6" else "ipv4"
iface = result.setdefault(entry["interface"], {})
iface.setdefault(family, {})[address] = {"prefix_length": int(entry["prefix"])}
return result
def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
entries = []
for n in _as_list(self._run_ps(_PS_ARP)):
if (n.get("state") or "").lower() in _DEAD_NEIGHBOUR_STATES:
continue
mac = _mac(n.get("mac"))
if not _is_real_neighbour_mac(mac):
continue
entries.append({"interface": n["interface"], "mac": mac, "ip": n["ip"], "age": 0.0})
return entries
def get_route_to(
self,
destination: str = "",
protocol: str = "",
longer: bool = False,
) -> dict[str, list[dict[str, Any]]]:
routes: dict[str, list[dict[str, Any]]] = {}
for r in _as_list(self._run_ps(_PS_ROUTES)):
network = r["destination"]
if _is_noise_route(network):
continue
if destination and network != destination:
continue
raw_proto = (r.get("protocol") or "").lower()
proto = _ROUTE_PROTOCOLS.get(raw_proto, raw_proto)
if protocol and proto != protocol.lower():
continue
next_hop = r.get("next_hop") or ""
# The unspecified next hop of an on-link route, read from the host; nothing binds.
if next_hop in ("0.0.0.0", "::"): # nosec B104
next_hop = ""
routes.setdefault(network, []).append(
{
"protocol": proto,
"family": "ipv6" if ":" in network else "ipv4",
"current_active": True,
"last_active": False,
"age": -1,
"next_hop": next_hop,
"outgoing_interface": r.get("interface") or "",
"selected_next_hop": True,
"preference": int(r.get("metric") or 0),
"routing_table": "global",
"protocol_attributes": {},
}
)
return routes
# ------------------------------------------------------------------
# Services
# ------------------------------------------------------------------
def get_services(self) -> list[ServiceDict]:
services: list[ServiceDict] = []
for s in _as_list(self._run_ps(_PS_SERVICES)):
running = (s.get("state") or "").lower() == "running"
services.append(
{
"name": s["name"],
"running": running,
"enabled": (s.get("start_mode") or "").lower() in _ENABLED_START_MODES,
"pid": int(s.get("pid") or 0) if running else 0,
}
)
return services
def get_listening_sockets(self) -> ListeningSocketsDict:
"""Every listening TCP and bound UDP socket, with the process and the
Windows service behind it (``Get-NetTCPConnection``, ``Get-NetUDPEndpoint``,
``Win32_Service``). WinRM needs an administrator, who sees every
process, so the reading is ``attributed``."""
raw = self._run_ps(_PS_LISTENING) or {}
names = {int(p["pid"]): str(p["name"]) for p in _as_list(raw.get("processes"))}
owners: dict[int, list[str]] = {}
for service in _as_list(raw.get("services")):
owners.setdefault(int(service["pid"]), []).append(str(service["name"]))
sockets = [
_listening_socket(proto, entry, names, owners)
for proto in ("tcp", "udp")
for entry in _as_list(raw.get(proto))
]
sockets.sort(key=lambda s: (s["proto"], s["port"], s["address"], s["interface"] or ""))
return {"attributed": True, "sockets": sockets}
def manage_service(self, name: str, action: str) -> dict[str, Any]:
template = _SERVICE_COMMANDS.get(action)
if template is None:
raise ValueError(
f"Unsupported action {action!r}; use one of {sorted(_SERVICE_COMMANDS)}"
)
if not _SERVICE_NAME_RE.match(name or ""):
raise ValueError(f"Invalid service name {name!r}")
command = template.format(name=_ps_quote(name))
try:
output = self._run_raw(f"{command} -ErrorAction Stop | Out-Null")
except PowerShellError as exc:
return {"success": False, "output": str(exc)}
return {"success": True, "output": output}