commit 60525bd35e9013a1cd64e750dcee22f6631ec1e3 Author: Christian Manivong Date: Wed Jun 24 15:17:08 2026 +0200 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 diff --git a/napalm_yealink.egg-info/PKG-INFO b/napalm_yealink.egg-info/PKG-INFO new file mode 100644 index 0000000..f80d5df --- /dev/null +++ b/napalm_yealink.egg-info/PKG-INFO @@ -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 diff --git a/napalm_yealink.egg-info/SOURCES.txt b/napalm_yealink.egg-info/SOURCES.txt new file mode 100644 index 0000000..0377478 --- /dev/null +++ b/napalm_yealink.egg-info/SOURCES.txt @@ -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 \ No newline at end of file diff --git a/napalm_yealink.egg-info/dependency_links.txt b/napalm_yealink.egg-info/dependency_links.txt new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/napalm_yealink.egg-info/dependency_links.txt @@ -0,0 +1 @@ + diff --git a/napalm_yealink.egg-info/entry_points.txt b/napalm_yealink.egg-info/entry_points.txt new file mode 100644 index 0000000..dfce953 --- /dev/null +++ b/napalm_yealink.egg-info/entry_points.txt @@ -0,0 +1,2 @@ +[napalm.drivers] +yealink = napalm_yealink:YealinkDriver diff --git a/napalm_yealink.egg-info/requires.txt b/napalm_yealink.egg-info/requires.txt new file mode 100644 index 0000000..2db9ac5 --- /dev/null +++ b/napalm_yealink.egg-info/requires.txt @@ -0,0 +1,3 @@ +napalm +napalm-device-types +requests diff --git a/napalm_yealink.egg-info/top_level.txt b/napalm_yealink.egg-info/top_level.txt new file mode 100644 index 0000000..f0dc5ee --- /dev/null +++ b/napalm_yealink.egg-info/top_level.txt @@ -0,0 +1 @@ +napalm_yealink diff --git a/napalm_yealink/__init__.py b/napalm_yealink/__init__.py new file mode 100644 index 0000000..3849d8c --- /dev/null +++ b/napalm_yealink/__init__.py @@ -0,0 +1,3 @@ +from napalm_yealink.yealink import YealinkDriver + +__all__ = ["YealinkDriver"] diff --git a/napalm_yealink/__pycache__/__init__.cpython-312.pyc b/napalm_yealink/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..f367807 Binary files /dev/null and b/napalm_yealink/__pycache__/__init__.cpython-312.pyc differ diff --git a/napalm_yealink/__pycache__/yealink.cpython-312.pyc b/napalm_yealink/__pycache__/yealink.cpython-312.pyc new file mode 100644 index 0000000..7803cb4 Binary files /dev/null and b/napalm_yealink/__pycache__/yealink.cpython-312.pyc differ diff --git a/napalm_yealink/yealink.py b/napalm_yealink/yealink.py new file mode 100644 index 0000000..cfa4d22 --- /dev/null +++ b/napalm_yealink/yealink.py @@ -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 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..46c32a4 --- /dev/null +++ b/pyproject.toml @@ -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*"] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/__pycache__/__init__.cpython-312.pyc b/tests/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..1e4997b Binary files /dev/null and b/tests/__pycache__/__init__.cpython-312.pyc differ diff --git a/tests/__pycache__/test_fingerprint.cpython-312-pytest-9.0.3.pyc b/tests/__pycache__/test_fingerprint.cpython-312-pytest-9.0.3.pyc new file mode 100644 index 0000000..5849672 Binary files /dev/null and b/tests/__pycache__/test_fingerprint.cpython-312-pytest-9.0.3.pyc differ diff --git a/tests/test_fingerprint.py b/tests/test_fingerprint.py new file mode 100644 index 0000000..0808f94 --- /dev/null +++ b/tests/test_fingerprint.py @@ -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)