"""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 = ["00:80:F0", "00:1A:8C", "80:5E:C0", "80:5E:0C", "24:9A:D8", "70:32:17", "34:6B:D3"] 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