feat: initial napalm-yealink driver
Discovery-Fingerprinting für Yealink IP-Phones (T-series, W-series, CP-series, VP-series) und Video-Endpoints. - TYPE_LABEL = "Phone" (Override von AccessPointDriver) - SNMP_OBJECT_ID_PREFIX = "1.3.6.1.4.1.37403" (Yealink Enterprise OID) - HTTP_FINGERPRINT: "yealink" (mandatory, weight 10), Modell-Patterns - get_facts(): liest phoneName + firmware aus dem noauth-API-Response (/api/v1/system/info liefert Gerätedaten auch ohne Authentication) - Login via MD5-gehashtem Passwort auf /api/v1/session/login (HTTPS) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,8 @@
|
||||
Metadata-Version: 2.4
|
||||
Name: napalm-yealink
|
||||
Version: 0.1.0
|
||||
Summary: NAPALM driver for Yealink IP phones and video endpoints
|
||||
Requires-Python: >=3.10
|
||||
Requires-Dist: napalm
|
||||
Requires-Dist: napalm-device-types
|
||||
Requires-Dist: requests
|
||||
@@ -0,0 +1,10 @@
|
||||
pyproject.toml
|
||||
napalm_yealink/__init__.py
|
||||
napalm_yealink/yealink.py
|
||||
napalm_yealink.egg-info/PKG-INFO
|
||||
napalm_yealink.egg-info/SOURCES.txt
|
||||
napalm_yealink.egg-info/dependency_links.txt
|
||||
napalm_yealink.egg-info/entry_points.txt
|
||||
napalm_yealink.egg-info/requires.txt
|
||||
napalm_yealink.egg-info/top_level.txt
|
||||
tests/test_fingerprint.py
|
||||
@@ -0,0 +1 @@
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
[napalm.drivers]
|
||||
yealink = napalm_yealink:YealinkDriver
|
||||
@@ -0,0 +1,3 @@
|
||||
napalm
|
||||
napalm-device-types
|
||||
requests
|
||||
@@ -0,0 +1 @@
|
||||
napalm_yealink
|
||||
@@ -0,0 +1,3 @@
|
||||
from napalm_yealink.yealink import YealinkDriver
|
||||
|
||||
__all__ = ["YealinkDriver"]
|
||||
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,224 @@
|
||||
"""NAPALM driver for Yealink IP phones and video endpoints.
|
||||
|
||||
Yealink exposes a REST API at ``/api/v1/``. Authentication sends the
|
||||
password as an MD5 hash in a JSON POST to ``/api/v1/session/login``.
|
||||
The session cookie is then used for subsequent requests.
|
||||
|
||||
Even without authentication, ``/api/v1/system/info`` returns a ``noauth``
|
||||
error that includes ``phoneName`` and ``firmware`` — enough for discovery.
|
||||
|
||||
Pass credentials via *username* / *password* (driver hashes the password
|
||||
automatically) or via ``optional_args={"verify_ssl": False}``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import socket
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
from requests.exceptions import RequestException
|
||||
|
||||
from napalm.base.exceptions import ConnectionClosedException, ConnectionException
|
||||
from napalm_device_types import AccessPointDriver, FingerprintRule
|
||||
from napalm_device_types.models import HealthMetricsDict
|
||||
|
||||
|
||||
class YealinkDriver(AccessPointDriver):
|
||||
"""NAPALM driver for Yealink IP phones (T-series, W-series, CP-series, VP-series)."""
|
||||
|
||||
TYPE_LABEL = "Phone"
|
||||
VENDOR = "Yealink"
|
||||
DRIVER_NAME = "yealink"
|
||||
|
||||
SNMP_OBJECT_ID_PREFIX = "1.3.6.1.4.1.37403"
|
||||
HTTP_FINGERPRINT = [
|
||||
FingerprintRule("yealink", weight=10.0, mandatory=True),
|
||||
FingerprintRule("sip-t", weight=5.0),
|
||||
FingerprintRule("sip-w", weight=5.0),
|
||||
FingerprintRule("sip-cp", weight=5.0),
|
||||
FingerprintRule("vp59", weight=5.0),
|
||||
]
|
||||
|
||||
_API_BASE = "/api/v1"
|
||||
_LOGIN_PATH = "/api/v1/session/login"
|
||||
_INFO_PATH = "/api/v1/system/info"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hostname: str,
|
||||
username: str,
|
||||
password: str,
|
||||
timeout: int = 30,
|
||||
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._verify_ssl = optional_args.get("verify_ssl", False)
|
||||
self._session: requests.Session | None = None
|
||||
self._base_url = f"https://{hostname}"
|
||||
|
||||
# ── Connection lifecycle ──────────────────────────────────────────────────
|
||||
|
||||
def open(self) -> None:
|
||||
self._session = requests.Session()
|
||||
self._session.verify = self._verify_ssl
|
||||
pw_hash = hashlib.md5(self.password.encode()).hexdigest()
|
||||
try:
|
||||
resp = self._session.post(
|
||||
f"{self._base_url}{self._LOGIN_PATH}",
|
||||
json={"username": self.username, "password": pw_hash},
|
||||
timeout=self.timeout,
|
||||
)
|
||||
data = resp.json()
|
||||
if data.get("ret") != "success":
|
||||
# Still usable for read-only discovery data
|
||||
pass
|
||||
except RequestException as exc:
|
||||
raise ConnectionException(f"Cannot connect to {self.hostname}: {exc}") from exc
|
||||
|
||||
def close(self) -> None:
|
||||
if self._session:
|
||||
try:
|
||||
self._session.post(
|
||||
f"{self._base_url}{self._API_BASE}/session/logout",
|
||||
timeout=5,
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
self._session.close()
|
||||
self._session = None
|
||||
|
||||
def is_alive(self) -> dict[str, bool]:
|
||||
try:
|
||||
sock = socket.create_connection((self.hostname, 443), timeout=5)
|
||||
sock.close()
|
||||
return {"is_alive": True}
|
||||
except Exception:
|
||||
return {"is_alive": False}
|
||||
|
||||
# ── Facts ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def get_facts(self) -> dict[str, Any]:
|
||||
"""Return basic device facts.
|
||||
|
||||
Works partially even without authentication — the noauth API response
|
||||
includes phoneName and firmware.
|
||||
"""
|
||||
info: dict = {}
|
||||
try:
|
||||
resp = self._session.get(
|
||||
f"{self._base_url}{self._INFO_PATH}",
|
||||
timeout=self.timeout,
|
||||
)
|
||||
data = resp.json()
|
||||
# Yealink returns facts in 'data' (authed) or 'error' (noauth)
|
||||
payload = data.get("data") or data.get("error") or {}
|
||||
info = payload if isinstance(payload, dict) else {}
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
model = info.get("phoneName") or info.get("model") or "Yealink"
|
||||
firmware = info.get("firmware") or info.get("firmwareVersion") or ""
|
||||
hostname = info.get("hostName") or info.get("deviceName") or self.hostname
|
||||
|
||||
return {
|
||||
"vendor": self.VENDOR,
|
||||
"model": model,
|
||||
"os_version": firmware,
|
||||
"hostname": hostname,
|
||||
"fqdn": hostname,
|
||||
"serial_number": info.get("serialNumber") or "",
|
||||
"uptime": 0,
|
||||
"interface_list": [],
|
||||
}
|
||||
|
||||
# ── NAPALM stubs (not implemented for phones) ─────────────────────────────
|
||||
|
||||
def get_interfaces(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_config(self, retrieve: str = "all", full: bool = False, sanitized: bool = False) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_interfaces_ip(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_interfaces_counters(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_bgp_neighbors(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_environment(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_lldp_neighbors(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_lldp_neighbors_detail(self, interface: str = "") -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_arp_table(self, vrf: str = "") -> list:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_mac_address_table(self) -> list:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_ntp_servers(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_ntp_stats(self) -> list:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_users(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_route_to(self, destination: str = "", protocol: str = "", ip_type: str = "") -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_snmp_information(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def ping(self, destination: str, **kwargs: Any) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def cli(self, commands: list[str], encoding: str = "text") -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def load_replace_candidate(self, filename: str | None = None, config: str | None = None) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
def load_merge_candidate(self, filename: str | None = None, config: str | None = None) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
def compare_config(self) -> str:
|
||||
raise NotImplementedError
|
||||
|
||||
def commit_config(self, message: str = "") -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
def discard_config(self) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
def rollback(self) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
# AccessPointDriver stubs
|
||||
|
||||
def get_wireless_clients(self) -> list:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_ssids(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_radio_info(self) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
|
||||
raise NotImplementedError
|
||||
@@ -0,0 +1,21 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=61"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "napalm-yealink"
|
||||
version = "0.1.0"
|
||||
description = "NAPALM driver for Yealink IP phones and video endpoints"
|
||||
requires-python = ">=3.10"
|
||||
dependencies = [
|
||||
"napalm",
|
||||
"napalm-device-types",
|
||||
"requests",
|
||||
]
|
||||
|
||||
[project.entry-points."napalm.drivers"]
|
||||
yealink = "napalm_yealink:YealinkDriver"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
include = ["napalm_yealink*"]
|
||||
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,43 @@
|
||||
"""Fingerprint tests for YealinkDriver."""
|
||||
|
||||
from napalm_device_types import DeviceTypeDriver, FingerprintRule
|
||||
from napalm_yealink import YealinkDriver
|
||||
|
||||
|
||||
def test_yealink_is_device_type_driver():
|
||||
assert issubclass(YealinkDriver, DeviceTypeDriver)
|
||||
|
||||
|
||||
def test_type_label():
|
||||
assert YealinkDriver.TYPE_LABEL == "Phone"
|
||||
|
||||
|
||||
def test_vendor():
|
||||
assert YealinkDriver.VENDOR == "Yealink"
|
||||
|
||||
|
||||
def test_driver_name():
|
||||
assert YealinkDriver.DRIVER_NAME == "yealink"
|
||||
|
||||
|
||||
def test_snmp_oid_prefix():
|
||||
assert YealinkDriver.SNMP_OBJECT_ID_PREFIX == "1.3.6.1.4.1.37403"
|
||||
|
||||
|
||||
def test_http_fingerprint_mandatory_yealink():
|
||||
mandatory = [r for r in YealinkDriver.HTTP_FINGERPRINT if r.mandatory]
|
||||
assert any(r.pattern == "yealink" for r in mandatory)
|
||||
|
||||
|
||||
def test_fingerprint_matches_t58_title():
|
||||
"""'Yealink T58 Phone' im Titel → mandatory 'yealink' trifft zu."""
|
||||
combined = "yealink t58 phone nginx "
|
||||
mandatory = [r for r in YealinkDriver.HTTP_FINGERPRINT if r.mandatory]
|
||||
assert all(r.pattern in combined for r in mandatory)
|
||||
|
||||
|
||||
def test_fingerprint_does_not_match_cisco():
|
||||
"""Cisco-Seite enthält kein 'yealink' → mandatory Pattern fehlt."""
|
||||
combined = "cisco ip phone cp-8841 "
|
||||
mandatory = [r for r in YealinkDriver.HTTP_FINGERPRINT if r.mandatory]
|
||||
assert not all(r.pattern in combined for r in mandatory)
|
||||
Reference in New Issue
Block a user