napalm-proxmox/napalm_proxmox/config_mixin.py

# Copyright 2025 The NetOrk Project Authors. All rights reserved.
#
# 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.

"""Configuration management NAPALM methods for Proxmox VE nodes."""

from __future__ import annotations

import re
from typing import Any


_JsonDict = dict[str, Any]


class ProxmoxConfigMixin:
    """Mixin providing configuration-management NAPALM methods."""

    def get_config(
        self,
        retrieve: str = "all",
        full: bool = False,
        sanitized: bool = False,
        format: str = "text",
    ) -> _JsonDict:
        """Return the node network configuration.

        ``running`` config is the contents of ``/etc/network/interfaces``
        (and the SDN config directory).  ``startup`` is identical (PVE
        applies on boot).  ``candidate`` is what was loaded via
        ``load_merge_candidate`` / ``load_replace_candidate`` but not yet
        committed.
        """
        configs: _JsonDict = {"running": "", "candidate": "", "startup": ""}

        if retrieve in ("running", "all", "startup"):
            raw = self._exec_ssh_command(
                "cat /etc/network/interfaces 2>/dev/null || true"
            )
            # Append SDN config if available
            sdn_raw = self._exec_ssh_command(
                "cat /etc/pve/sdn/vnets.cfg 2>/dev/null || true"
            )
            running = raw
            if sdn_raw:
                running += "\n# === SDN VNets ===\n" + sdn_raw
            if sanitized:
                running = re.sub(r"password\s+\S+", "password ****", running)
            configs["running"] = running
            self._running_config = running
            if retrieve == "startup":
                configs["startup"] = running
            elif retrieve == "all":
                configs["startup"] = running

        if retrieve in ("candidate", "all"):
            configs["candidate"] = self._candidate_config

        return configs

    def load_merge_candidate(
        self,
        filename: str | None = None,
        config: str | None = None,
    ) -> None:
        """Load a candidate configuration (merge mode)."""
        if filename:
            with open(filename) as fh:
                config = fh.read()
        if config is None:
            raise ValueError("Either filename or config must be provided")
        # In merge mode we append / overlay
        self._candidate_config = config

    def load_replace_candidate(
        self,
        filename: str | None = None,
        config: str | None = None,
    ) -> None:
        """Load a candidate configuration (replace mode)."""
        if filename:
            with open(filename) as fh:
                config = fh.read()
        if config is None:
            raise ValueError("Either filename or config must be provided")
        self._candidate_config = config

    def compare_config(self) -> str:
        """Return a unified diff between running and candidate config."""
        import difflib

        if not self._running_config:
            self.get_config(retrieve="running")
        running_lines = self._running_config.splitlines(keepends=True)
        candidate_lines = self._candidate_config.splitlines(keepends=True)
        diff = difflib.unified_diff(
            running_lines,
            candidate_lines,
            fromfile="running",
            tofile="candidate",
        )
        return "".join(diff)

    def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
        """Commit the candidate configuration to the Proxmox node.

        This writes the candidate config to ``/etc/network/interfaces``
        via the Proxmox node/network PUT API (which applies it live).

        .. note::
            Full programmatic apply requires the Proxmox API to accept raw
            interface configs.  This implementation uses ``pvesh`` via exec
            which requires the node exec endpoint to be available.
        """
        if not self._candidate_config:
            return
        # Write via exec endpoint
        escaped = self._candidate_config.replace("'", "'\\''")
        self._exec_ssh_command(
            f"printf '%s' '{escaped}' > /etc/network/interfaces && "
            "ifreload -a 2>&1 || ifup -a 2>&1 || true"
        )
        self._running_config = self._candidate_config
        self._candidate_config = ""

    def discard_config(self) -> None:
        """Discard the candidate configuration."""
        self._candidate_config = ""

    def rollback(self) -> None:
        """Revert to the stored running configuration."""
        if self._running_config:
            self._candidate_config = self._running_config
            self.commit_config()
        self._candidate_config = ""