napalm-hpe-officeconnect/napalm_hpe_officeconnect/client.py

"""HTTP session handling for HPE OfficeConnect switches.

The switch authenticates with a form POST that answers JSON and hands out a
``SID`` cookie::

    GET  /htdocs/login/login.lsp     → sets the short-lived RID cookie
    POST /htdocs/login/login.lua     → {"redirect": "...", "error": ""}
    GET  /htdocs/pages/main/logout.lsp

An empty ``error`` means success. Firmware generations differ in what else
they put in that object — older builds omit ``redirect``, newer ones omit
``username`` — so ``error`` is the only field worth branching on.

Session hygiene matters here: the switch keeps a small table of concurrent
sessions and only reclaims them on idle timeout. A driver that logs in on
every poll and never logs out will eventually lock the administrator out of
the web UI, so :meth:`logout` is best-effort and always safe to call.
"""

from __future__ import annotations

import logging

import requests
from napalm.base.exceptions import (
    CommandErrorException,
    ConnectionClosedException,
    ConnectionException,
)

logger = logging.getLogger("napalm_hpe_officeconnect")

LOGIN_PAGE = "/htdocs/login/login.lsp"
LOGIN_ENDPOINT = "/htdocs/login/login.lua"
LOGOUT_ENDPOINT = "/htdocs/pages/main/logout.lsp"


class OfficeConnectClient:
    """Authenticated HTTP session against an OfficeConnect switch."""

    def __init__(
        self,
        host: str,
        username: str,
        password: str,
        scheme: str = "http",
        port: int | None = None,
        timeout: int = 30,
        verify: bool = False,
    ) -> None:
        self.host = host
        self.username = username
        self.password = password
        # HTTP by default: the switch only serves HTTPS once an operator has
        # uploaded a certificate, which is not the common case.
        self.scheme = scheme
        self.port = port
        self.timeout = timeout
        self.verify = verify
        self._session = requests.Session()
        self._authenticated = False

    @property
    def base_url(self) -> str:
        if self.port:
            return f"{self.scheme}://{self.host}:{self.port}"
        return f"{self.scheme}://{self.host}"

    @property
    def is_authenticated(self) -> bool:
        return self._authenticated

    def login(self) -> None:
        """Establish a session. Raises ConnectionException on any failure."""
        try:
            # The login page issues the RID cookie the POST is validated against.
            self._session.get(self.base_url + LOGIN_PAGE, timeout=self.timeout, verify=self.verify)
            response = self._session.post(
                self.base_url + LOGIN_ENDPOINT,
                data={"username": self.username, "password": self.password},
                timeout=self.timeout,
                verify=self.verify,
            )
        except requests.RequestException as exc:
            raise ConnectionException(f"Cannot reach {self.host}: {exc}") from exc

        try:
            payload = response.json()
        except ValueError as exc:
            raise ConnectionException(
                f"Login to {self.host} returned no JSON "
                f"(HTTP {response.status_code}) — is this an OfficeConnect switch?"
            ) from exc

        error = (payload.get("error") or "").strip()
        if error:
            raise ConnectionException(f"Login to {self.host} failed: {error}")

        self._authenticated = True

    def fetch(self, path: str) -> str:
        """Return the body of an authenticated page."""
        if not self._authenticated:
            raise ConnectionException(f"Not logged in to {self.host} — call login() first")

        try:
            response = self._session.get(
                self.base_url + path,
                timeout=self.timeout,
                verify=self.verify,
                # Do not follow the redirect: a 3xx here *is* the error signal.
                allow_redirects=False,
            )
        except requests.RequestException as exc:
            raise ConnectionClosedException(f"Request to {self.host}{path} failed: {exc}") from exc

        if 300 <= response.status_code < 400:
            # The switch bounces expired sessions to the login page instead of
            # answering 401.
            self._authenticated = False
            raise ConnectionClosedException(f"Session to {self.host} expired while fetching {path}")
        if response.status_code != 200:
            raise CommandErrorException(f"{self.host}{path} returned HTTP {response.status_code}")

        return response.text

    def logout(self) -> None:
        """Release the session. Best-effort: never raises.

        Called from ``close()``, including on the failure path, where raising
        would mask the exception that actually caused the disconnect.
        """
        if not self._authenticated:
            return
        try:
            self._session.get(
                self.base_url + LOGOUT_ENDPOINT,
                timeout=self.timeout,
                verify=self.verify,
                allow_redirects=False,
            )
        except Exception as exc:  # noqa: BLE001 — deliberate: see docstring
            logger.debug("Logout from %s failed, session will idle out: %s", self.host, exc)
        finally:
            self._authenticated = False