229 lines
7.7 KiB
Python
229 lines
7.7 KiB
Python
"""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"
|
|
OUI_PREFIXES = [
|
|
"80:5E:C0", # YEALINK(XIAMEN) NETWORK TECHNOLOGY — IEEE, verifiziert an T58V
|
|
"80:5E:0C", # YEALINK(XIAMEN) NETWORK TECHNOLOGY — IEEE, verifiziert an T58V
|
|
]
|
|
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
|