napalm-openwrt/napalm_openwrt/packages_mixin.py

# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import re
from typing import Any


class OpenWrtPackageMixin:
    """Mixin providing package-management NAPALM methods."""

    # Package management  (opkg <= OpenWrt 23 / apk >= OpenWrt 24)
    # ------------------------------------------------------------------

    def get_packages(self) -> list[dict[str, Any]]:
        """Return installed packages from the device's package manager.

        Automatically detects whether to use ``apk`` (OpenWrt 24+, Alpine
        APK) or ``opkg`` (older OpenWrt releases).  Returns one entry per
        installed package.
        """
        pm = self._send_command("command -v apk 2>/dev/null || echo __no_apk__").strip()
        if "__no_apk__" not in pm and pm:
            return self._get_packages_apk()
        return self._get_packages_opkg()

    def _get_packages_opkg(self) -> list[dict[str, Any]]:
        """Parse ``opkg status`` (dpkg-style stanzas)."""
        out = self._send_command("opkg status")
        packages: list[dict[str, Any]] = []
        stanza: dict[str, str] = {}
        for raw in out.splitlines():
            line = raw.rstrip()
            if line == "":
                if stanza.get("Package"):
                    packages.append(self._opkg_stanza_to_dict(stanza))
                stanza = {}
            elif line[:1] in (" ", "\t"):
                # Continuation of previous field (e.g. multi-line Description)
                last_key = list(stanza)[-1] if stanza else None
                if last_key:
                    stanza[last_key] += " " + line.strip()
            elif ":" in line:
                key, _, val = line.partition(":")
                stanza[key.strip()] = val.strip()
        if stanza.get("Package"):
            packages.append(self._opkg_stanza_to_dict(stanza))
        return sorted(packages, key=lambda p: p["name"].lower())

    @staticmethod
    def _opkg_stanza_to_dict(stanza: dict[str, str]) -> dict[str, Any]:
        status = stanza.get("Status", "")
        try:
            size = int(stanza.get("Installed-Size", 0) or 0)
        except ValueError:
            size = 0
        return {
            "name": stanza["Package"],
            "version": stanza.get("Version", ""),
            "installed": "installed" in status.lower(),
            "description": stanza.get("Description", ""),
            "size": size,
            "source": stanza.get("Section", ""),
        }

    def _get_packages_apk(self) -> list[dict[str, Any]]:
        """Parse ``apk list --installed`` output.

        Line format::

            busybox-1.37.0-r0 x86_64 {busybox} (GPL-2.0-only) [installed]
            kmod-nft-bridge-6.6.75-r0 mips_24kc {kmod-nft-bridge} (GPL-2.0-only) [installed]
        """
        out = self._send_command("apk list --installed 2>/dev/null")
        packages: list[dict[str, Any]] = []
        for line in out.splitlines():
            line = line.strip()
            if not line or "[installed]" not in line:
                continue
            # Split name from version: version always starts with a digit after '-'
            m = re.match(r"^(.*?)-(\d\S*)\s+\S+\s+\{(\S+)\}", line)
            if m:
                name, version, origin = m.group(1), m.group(2), m.group(3)
            else:
                # Minimal fallback: first token only
                token = line.split()[0]
                vm = re.search(r"-(\d\S*)$", token)
                name = token[: vm.start()] if vm else token
                version = vm.group(1) if vm else ""
                origin = ""
            packages.append({
                "name": name,
                "version": version,
                "installed": True,
                "description": "",
                "size": 0,
                "source": origin,
            })
        return sorted(packages, key=lambda p: p["name"].lower())

    def _pm_type(self) -> str:
        """Return ``'apk'`` if device has apk (OpenWrt 24+), otherwise ``'opkg'``."""
        out = self._send_command("command -v apk 2>/dev/null || echo __no_apk__").strip()
        return "apk" if ("__no_apk__" not in out and out) else "opkg"

    def search_packages(self, query: str) -> list[dict[str, Any]]:
        """Search available packages matching *query* (name or description).

        Runs ``opkg update`` / ``apk update`` first to ensure the package
        index is populated (OpenWrt stores it in RAM and loses it on reboot).
        """
        import shlex
        safe_q = shlex.quote(query)
        if self._pm_type() == "apk":
            # Refresh index (no-ops if already current, safe to run every time)
            self._send_command("apk update 2>/dev/null || true")
            out = self._send_command(f"apk search {safe_q} 2>/dev/null")
            installed = {p["name"] for p in self._get_packages_apk()}
            packages: list[dict[str, Any]] = []
            for line in out.splitlines():
                line = line.strip()
                if not line:
                    continue
                m = re.match(r"^(.*?)-(\d\S*)(?:\s+(.*))?$", line)
                if m:
                    name, version, description = m.group(1), m.group(2), (m.group(3) or "")
                else:
                    name, version, description = line, "", ""
                packages.append({
                    "name": name,
                    "version": version,
                    "installed": name in installed,
                    "description": description,
                    "size": 0,
                    "source": "",
                })
        else:
            # opkg lists live in /var/opkg-lists/ (RAM) — cleared on reboot
            self._send_command("opkg update 2>/dev/null || true")
            out = self._send_command(f"opkg list 2>/dev/null | grep -i {safe_q}")
            installed = {p["name"] for p in self._get_packages_opkg()}
            packages = []
            for line in out.splitlines():
                line = line.strip()
                if not line:
                    continue
                parts = line.split(" - ", 2)
                name = parts[0].strip()
                version = parts[1].strip() if len(parts) > 1 else ""
                description = parts[2].strip() if len(parts) > 2 else ""
                packages.append({
                    "name": name,
                    "version": version,
                    "installed": name in installed,
                    "description": description,
                    "size": 0,
                    "source": "",
                })
        return packages

    @staticmethod
    def _clean_pkg_output(raw: str) -> str:
        """Strip ANSI/VT100 escape sequences and progress-bar lines."""
        # Strip CSI sequences (\x1b[...X), OSC, charset designations, and
        # 2-byte DEC private sequences like ESC 7 (cursor save) / ESC 8 (restore)
        cleaned = re.sub(
            r'\x1b(?:\[[0-9;?]*[a-zA-Z]|\][^\x07]*\x07|[()][0-9A-Za-z]|[\x30-\x7e])',
            '', raw,
        )
        # After stripping cursor-save/restore sequences, apk progress updates
        # end up concatenated on a single line.  Strip those inline patterns.
        cleaned = re.sub(r'\s*\d{1,3}%\s*#*', ' ', cleaned)
        lines = []
        for segment in cleaned.split('\n'):
            # \r overwrites the line; keep only the portion after the last \r
            part = segment.split('\r')[-1].strip()
            if not part:
                continue
            # Drop pure progress-bar lines (only #, spaces, digits, %)
            if re.match(r'^[#\s\d%]*$', part):
                continue
            lines.append(part)
        return '\n'.join(lines)

    def install_package(self, name: str) -> dict[str, Any]:
        """Install a package by name. Returns ``{"success": bool, "output": str}``."""
        import shlex
        safe_name = shlex.quote(name)
        if self._pm_type() == "apk":
            raw = self._send_command(f"apk add {safe_name} 2>&1")
        else:
            raw = self._send_command(f"opkg install {safe_name} 2>&1")
        out = self._clean_pkg_output(raw)
        low = out.lower()
        success = not any(kw in low for kw in ("error:", "failed", "not found", "unknown package"))
        return {"success": success, "output": out}

    def uninstall_package(self, name: str) -> dict[str, Any]:
        """Remove a package by name. Returns ``{"success": bool, "output": str}``."""
        import shlex
        safe_name = shlex.quote(name)
        if self._pm_type() == "apk":
            raw = self._send_command(f"apk del {safe_name} 2>&1")
        else:
            raw = self._send_command(f"opkg remove {safe_name} 2>&1")
        out = self._clean_pkg_output(raw)
        low = out.lower()
        success = not any(kw in low for kw in ("error:", "failed", "not found", "unknown package"))
        return {"success": success, "output": out}