napalm-hpe-officeconnect/napalm_hpe_officeconnect/officeconnect.py

"""NAPALM driver for HPE OfficeConnect 1820 / 1920S web-managed switches.

These are Broadcom FASTPATH-derived "smart managed" switches with **no CLI
at all** — no SSH, no Telnet, and no ArubaOS-Switch REST API. The only
management surface is the web UI, so this driver drives that UI directly.
That is why it shares no code with ``napalm-hpe-aruba-procurve``: a
ProCurve/ArubaOS-Switch speaks ``show``/``configure`` over SSH, which does
not exist here.

The driver is deliberately **read-only**. These switches expose a single
administrator account with no privilege levels, and unless an operator has
uploaded a certificate the session runs over plain HTTP — so the credential
used for polling is necessarily the admin credential. Not implementing any
write path keeps this driver from being the thing that changes a switch
over an unauthenticated channel.

Tested against: HPE OfficeConnect 1820 8G PoE+ (65W), J9982A, firmware PT.02.19.
"""

from __future__ import annotations

import logging
import re
from typing import Any, ClassVar

from napalm_device_types import FingerprintRule, SwitchDriver

from napalm_hpe_officeconnect import parsers
from napalm_hpe_officeconnect.client import OfficeConnectClient

logger = logging.getLogger("napalm_hpe_officeconnect")

_SPEED_RE = re.compile(r"(\d+)\s*Mbps", re.IGNORECASE)

# Participation values that make an interface a member of a VLAN.
_MEMBER_STATES = ("untagged", "tagged")


class OfficeConnectDriver(SwitchDriver):
    """NAPALM driver for HPE OfficeConnect web-managed switches."""

    VENDOR = "HPE"
    DRIVER_NAME = "hpe_officeconnect"

    # The ProCurve driver claims the bare HPE enterprise arc (1.3.6.1.4.1.11).
    # OfficeConnect models sit under .2.3.7.11 — claiming the longer arc keeps
    # the two drivers from competing for the same device.
    SNMP_OBJECT_ID_PREFIX = "1.3.6.1.4.1.11.2.3.7.11"

    # Verified on a J9982A. Deliberately narrow — a wrong OUI produces false
    # positives, while a missing one only costs a few points of confidence.
    OUI_PREFIXES: ClassVar[list[str]] = ["70:10:6F"]

    HTTP_FINGERPRINT: ClassVar[list[FingerprintRule]] = [
        # The login page title reads e.g.
        # "HPE OfficeConnect Switch 1820 8G PoE+ (65W) J9982A".
        FingerprintRule("officeconnect", weight=9.0, mandatory=True),
        FingerprintRule("1820", weight=6.0),
        FingerprintRule("1920s", weight=6.0),
        FingerprintRule("hewlett packard enterprise", weight=3.0),
    ]

    # sysDescr repeats the model string, e.g. "HPE OfficeConnect Switch 1820
    # 8G PoE+ (65W) J9982A, PT.02.19, Linux 3.6.5-…, U-Boot 2012.10-…".
    # Not mandatory: HTTP already carries the hard requirement, and a device
    # reachable only over SNMP should still be able to score.
    SNMP_FINGERPRINT: ClassVar[list[FingerprintRule]] = [
        FingerprintRule("officeconnect", weight=9.0),
        FingerprintRule("1820", weight=6.0),
        FingerprintRule("1920s", weight=6.0),
    ]

    # --- Web UI endpoints -------------------------------------------------
    DASHBOARD = "/htdocs/pages/base/dashboard.lsp"
    PORT_SUMMARY = "/htdocs/pages/base/port_summary.lsp"
    PORT_STATS = "/htdocs/pages/base/port_summary_stats.lsp"
    MAC_TABLE = "/htdocs/pages/base/mac_address_table.lsp"
    VLAN_STATUS = "/htdocs/pages/switching/vlan_status.lsp"
    VLAN_PER_PORT = "/htdocs/pages/switching/vlan_per_port.lsp?vlan={vlan}"

    def __init__(
        self,
        hostname: str,
        username: str,
        password: str,
        timeout: int = 60,
        optional_args: dict | None = None,
    ) -> None:
        self.hostname = hostname
        self.username = username
        self.password = password
        self.timeout = timeout

        optional_args = optional_args or {}
        self._client = OfficeConnectClient(
            hostname,
            username,
            password,
            # HTTP by default: HTTPS only works once a certificate has been
            # uploaded to the switch, which is not the common deployment.
            scheme=optional_args.get("scheme", "http"),
            port=optional_args.get("port"),
            timeout=timeout,
            verify=optional_args.get("ssl_verify", False),
        )

    # ------------------------------------------------------------------
    # Connection management
    # ------------------------------------------------------------------

    def open(self) -> None:
        self._client.login()

    def close(self) -> None:
        # Always release the session: the switch keeps a small session table
        # and only reclaims entries on idle timeout.
        self._client.logout()

    def is_alive(self) -> dict[str, bool]:
        return {"is_alive": bool(self._client.is_authenticated)}

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _rows(self, path: str) -> list[list[str]]:
        """Fetch a page and return its table rows as plain text cells."""
        page = self._client.fetch(path)
        return [
            [parsers.strip_markup(cell) for cell in row] for row in parsers.extract_data_set(page)
        ]

    @staticmethod
    def _speed_mbit(value: str) -> float:
        """``"100 Mbps Full Duplex"`` → ``100.0``; empty (link down) → ``0.0``."""
        match = _SPEED_RE.search(value or "")
        return float(match.group(1)) if match else 0.0

    # ------------------------------------------------------------------
    # Getters
    # ------------------------------------------------------------------

    def get_facts(self) -> dict[str, Any]:
        facts = parsers.parse_facts(self._client.fetch(self.DASHBOARD))
        hostname = facts["hostname"]
        return {
            "uptime": facts["uptime"],
            "vendor": self.VENDOR,
            "os_version": facts["os_version"],
            "serial_number": facts["serial_number"],
            "model": facts["model"],
            "hostname": hostname,
            "fqdn": hostname,
            "interface_list": [row[1] for row in self._rows(self.PORT_SUMMARY)],
        }

    def get_interfaces(self) -> dict[str, dict[str, Any]]:
        """Physical ports and trunk interfaces.

        Columns: [checkbox, Interface, Port Description, Admin Mode,
        Physical Type, Port Status, Physical Mode, Link Speed, MTU]
        """
        interfaces: dict[str, dict[str, Any]] = {}
        for row in self._rows(self.PORT_SUMMARY):
            interfaces[row[1]] = {
                "is_up": row[5] == "Link Up",
                "is_enabled": row[3] == "Enabled",
                "description": row[2],
                # The web UI reports neither flap time nor a per-port MAC.
                "last_flapped": -1.0,
                "speed": self._speed_mbit(row[7]),
                "mtu": int(row[8]) if row[8].isdigit() else 0,
                "mac_address": "",
            }
        return interfaces

    def get_interfaces_counters(self) -> dict[str, dict[str, int]]:
        """Not available on this platform.

        ``port_summary_stats.lsp`` declares the expected columns but ships an
        empty dataset on PT.02.19, and carries no AJAX endpoint that would
        fill it. Returning zeros would be indistinguishable from a switch
        that has genuinely passed no traffic, so this raises instead.
        """
        raise NotImplementedError(
            "OfficeConnect firmware PT.02.19 does not expose interface counters"
        )

    def _vlan_participation(self) -> dict[str, dict[str, Any]]:
        """VLAN table joined with per-VLAN interface participation.

        The participation page renders one VLAN at a time, so this costs one
        request per VLAN on top of the VLAN list itself.
        """
        result: dict[str, dict[str, Any]] = {}
        for row in self._rows(self.VLAN_STATUS):
            vlan_id, name = row[1], row[2]
            tagged: list[str] = []
            untagged: list[str] = []
            for member in self._rows(self.VLAN_PER_PORT.format(vlan=vlan_id)):
                interface, state = member[1], member[2].lower()
                if state == "tagged":
                    tagged.append(interface)
                elif state == "untagged":
                    untagged.append(interface)
            result[vlan_id] = {"name": name, "tagged": tagged, "untagged": untagged}
        return result

    def get_vlans(self) -> dict[str, dict[str, Any]]:
        return {
            vlan_id: {
                "name": data["name"],
                "interfaces": sorted(
                    data["untagged"] + data["tagged"],
                    key=lambda i: (i.startswith("TRK"), i),
                ),
            }
            for vlan_id, data in self._vlan_participation().items()
        }

    def get_vlans_detail(self) -> dict[str, dict[str, Any]]:
        """Like :meth:`get_vlans` but keeping tagged and untagged apart."""
        return self._vlan_participation()

    def get_mac_address_table(self) -> list[dict[str, Any]]:
        """Columns: [VLAN ID, MAC Address, Interface, Interface Index, Status]."""
        table = []
        for row in self._rows(self.MAC_TABLE):
            status = row[4].lower()
            table.append(
                {
                    "mac": row[1],
                    "interface": row[2],
                    "vlan": int(row[0]) if row[0].isdigit() else 0,
                    "static": status in ("static", "management"),
                    "active": True,
                    "moves": -1,
                    "last_move": -1.0,
                }
            )
        return table