napalm-yealink/napalm_yealink/yealink.py

"""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 logging
import socket
from typing import Any

import requests
from napalm.base.exceptions import ConnectionException
from napalm_device_types import FingerprintRule, PhoneDriver
from requests.exceptions import RequestException

logger = logging.getLogger(__name__)


class YealinkDriver(PhoneDriver):
    """NAPALM driver for Yealink IP phones (T-series, W-series, CP-series, VP-series)."""

    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 = [
        # Not mandatory: the redirect of some models to their API page has no "yealink" in it.
        FingerprintRule("yealink", weight=10.0),
        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
        # The login API takes the password as an MD5 hex digest: the wire format, not a
        # protection (the session runs over HTTPS).
        pw_hash = hashlib.md5(self.password.encode(), usedforsecurity=False).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:
                logger.debug("Logging out of the phone failed: ignored", exc_info=True)
            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:
            logger.debug("Reading the system info failed: ignored", exc_info=True)

        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