MrBlake/NetBox-VM-Import
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove Proxmox sync support

Browse Source
This commit is contained in:
MrBlake
2026-07-20 14:24:28 +02:00
parent 7d8819c374
commit 6ec4093c97
15 changed files with 192 additions and 752 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netbox_vmware_importer/sync.py
+6 -637
View File
@@ -1,9 +1,7 @@
import ipaddress
import re
import ssl
import time
from dataclasses import dataclass, field
from urllib.parse import urlsplit
from django.contrib.contenttypes.models import ContentType
from django.db import transaction
@@ -12,29 +10,7 @@ from dcim.models import MACAddress, Platform
from ipam.models import IPAddress
from virtualization.models import VMInterface, VirtualDisk, VirtualMachine
from .choices import EndpointAuthMethodChoices, EndpointProviderChoices
try:
import requests
from requests.adapters import HTTPAdapter
from urllib3.exceptions import InsecureRequestWarning
except ImportError: # pragma: no cover - handled at runtime inside NetBox
HTTPAdapter = None
requests = None
InsecureRequestWarning = None
try:
from proxmox_sdk import ProxmoxSDK
from proxmox_sdk.sdk.exceptions import (
ProxmoxConnectionError as SDKConnectionError,
ProxmoxTimeoutError as SDKTimeoutError,
ResourceException as SDKResourceException,
)
except ImportError: # pragma: no cover - optional compatibility fallback
ProxmoxSDK = None
SDKConnectionError = None
SDKTimeoutError = None
SDKResourceException = None
from .choices import EndpointProviderChoices
try:
from pyVim.connect import Disconnect, SmartConnect
@@ -96,34 +72,6 @@ class VMwareConnectionError(RuntimeError):
pass
class ProxmoxConnectionError(RuntimeError):
pass
class ProxmoxTLSCompatibilityAdapter(HTTPAdapter):
def __init__(self, validate_ssl=True, *args, **kwargs):
self.validate_ssl = validate_ssl
super().__init__(*args, **kwargs)
def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
pool_kwargs["ssl_context"] = self._build_ssl_context()
return super().init_poolmanager(connections, maxsize, block=block, **pool_kwargs)
def proxy_manager_for(self, *args, **kwargs):
kwargs["ssl_context"] = self._build_ssl_context()
return super().proxy_manager_for(*args, **kwargs)
def _build_ssl_context(self):
context = ssl.create_default_context()
if hasattr(ssl, "TLSVersion"):
context.minimum_version = ssl.TLSVersion.TLSv1_2
context.maximum_version = ssl.TLSVersion.TLSv1_2
if not self.validate_ssl:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
return context
class VMwareClient:
def __init__(self, endpoint):
self.endpoint = endpoint
@@ -264,605 +212,26 @@ class VMwareClient:
return interfaces
class ProxmoxClient:
DISK_KEY_PATTERN = re.compile(r"^(ide|sata|scsi|virtio)\d+$")
LXC_DISK_KEY_PATTERN = re.compile(r"^(rootfs|mp\d+)$")
NET_KEY_PATTERN = re.compile(r"^net\d+$")
FULL_TOKEN_PATTERN = re.compile(r"^(?:PVEAPIToken=)?(?P<user>[^!]+)!(?P<name>[^=]+)=(?P<value>.+)$")
RETRY_STATUSES = {502, 503, 504}
def __init__(self, endpoint):
self.endpoint = endpoint
self.host, self.port = self._parse_endpoint_address(endpoint.host, endpoint.port)
self.url_host = self._format_host_for_url(self.host)
self.session = None
self.sdk = None
self.uses_sdk = False
self.base_url = f"https://{self.url_host}:{self.port}/api2/json"
self.connect_timeout = endpoint.connect_timeout_seconds or 10
self.read_timeout = endpoint.request_timeout_seconds or 120
self.timeout = (self.connect_timeout, self.read_timeout)
self.max_retries = endpoint.max_retries or 0
self.retry_backoff = float(endpoint.retry_backoff_seconds or 0.5)
self.api_probe_succeeded = False
self.api_probe_error = ""
def __enter__(self):
if "@" not in self.endpoint.username:
raise ProxmoxConnectionError("Proxmox username must include a realm, e.g. root@pam or user@pve.")
if self._should_use_sdk():
self._connect_with_sdk()
return self
self._connect_with_requests()
return self
def __exit__(self, exc_type, exc_value, traceback):
if self.sdk is not None:
self.sdk.close()
if self.session is not None:
self.session.close()
@staticmethod
def _parse_endpoint_address(host, port):
raw_host = str(host or "").strip()
resolved_port = int(port or 8006)
if not raw_host:
return raw_host, resolved_port
parsed = urlsplit(raw_host if "://" in raw_host else f"//{raw_host}")
parsed_host = parsed.hostname
if parsed.port:
resolved_port = parsed.port
if parsed_host:
return parsed_host.strip("[]"), resolved_port
return raw_host.strip("[]"), resolved_port
@staticmethod
def _format_host_for_url(host):
if ":" in host and not host.startswith("["):
return f"[{host}]"
return host
def _should_use_sdk(self):
# proxmox-sdk 0.0.12 builds password ticket URLs without /api2/json.
# Keep it for stateless API-token auth, where no ticket endpoint is used.
return (
ProxmoxSDK is not None
and self.endpoint.auth_method == EndpointAuthMethodChoices.METHOD_API_TOKEN
)
def _connect_with_sdk(self):
user = (self.endpoint.username or "").strip()
kwargs = {
"host": self.host,
"backend": "https",
"service": "PVE",
"port": self.port,
"user": user,
"verify_ssl": self.endpoint.validate_ssl,
"timeout": self.read_timeout,
"connect_timeout": self.connect_timeout,
"max_retries": self.max_retries,
"retry_backoff": self.retry_backoff,
}
if self.endpoint.auth_method == EndpointAuthMethodChoices.METHOD_API_TOKEN:
user, token_name, token_value = self._normalized_token_credentials()
kwargs["user"] = user
kwargs.update(
{
"token_name": token_name,
"token_value": token_value,
}
)
else:
kwargs["password"] = self.endpoint.password
try:
self.sdk = ProxmoxSDK.sync(**kwargs)
self.uses_sdk = True
self._get("version")
self.api_probe_succeeded = True
except ProxmoxConnectionError:
if self.sdk is not None:
self.sdk.close()
self.sdk = None
raise
except Exception as exc:
if self.sdk is not None:
self.sdk.close()
self.sdk = None
raise ProxmoxConnectionError(self._describe_sdk_error("GET", "version", exc)) from exc
def _connect_with_requests(self):
if requests is None:
raise ProxmoxConnectionError(
"Neither proxmox-sdk nor requests is installed in the NetBox Python environment."
)
self.session = requests.Session()
self.session.trust_env = False
self.session.verify = self.endpoint.validate_ssl
if HTTPAdapter is not None:
self.session.mount("https://", ProxmoxTLSCompatibilityAdapter(validate_ssl=self.endpoint.validate_ssl))
self.session.headers.update(
{
"Accept": "application/json",
"Connection": "close",
"User-Agent": "netbox-vm-import/0.2",
}
)
if not self.endpoint.validate_ssl and InsecureRequestWarning is not None:
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
self._probe_api_availability()
if self.endpoint.auth_method == EndpointAuthMethodChoices.METHOD_API_TOKEN:
user, token_name, token_value = self._normalized_token_credentials()
self.session.headers.update({"Authorization": f"PVEAPIToken={user}!{token_name}={token_value}"})
else:
self._authenticate_with_password()
def _probe_api_availability(self):
try:
self._request("GET", "version", timeout=(5, 10))
self.api_probe_succeeded = True
self.api_probe_error = ""
except ProxmoxConnectionError as exc:
self.api_probe_succeeded = False
self.api_probe_error = str(exc)
def _normalized_token_credentials(self):
user = (self.endpoint.username or "").strip()
token_name = (self.endpoint.token_name or "").strip()
token_value = (self.endpoint.password or "").strip()
parsed = self._parse_full_token(token_value)
if parsed is not None:
user, token_name, token_value = parsed
else:
parsed = self._parse_full_token(token_name)
if parsed is not None:
parsed_user, parsed_name, parsed_value = parsed
user = parsed_user or user
token_name = parsed_name
if parsed_value and not token_value:
token_value = parsed_value
elif token_name.startswith("PVEAPIToken="):
token_name = token_name.removeprefix("PVEAPIToken=").strip()
if "!" in token_name:
parsed_user, parsed_name = token_name.rsplit("!", 1)
if parsed_user:
user = parsed_user
token_name = parsed_name
if not token_name or not token_value:
raise ProxmoxConnectionError(
"Proxmox API token authentication requires a token name and token secret."
)
return user, token_name, token_value
@classmethod
def _parse_full_token(cls, value):
match = cls.FULL_TOKEN_PATTERN.match(str(value or "").strip())
if not match:
return None
return match.group("user").strip(), match.group("name").strip(), match.group("value").strip()
def _authenticate_with_password(self):
response = self._request(
"POST",
"access/ticket",
data={
"username": self.endpoint.username,
"password": self.endpoint.password,
},
)
data = response.json().get("data") or {}
ticket = data.get("ticket")
csrf_token = data.get("CSRFPreventionToken")
if not ticket:
raise ProxmoxConnectionError("Proxmox authentication did not return a ticket.")
self.session.cookies.set("PVEAuthCookie", ticket)
if csrf_token:
self.session.headers.update({"CSRFPreventionToken": csrf_token})
def _url(self, path):
return f"{self.base_url}/{path.strip('/')}"
def _request(self, method, path, timeout=None, **kwargs):
url = self._url(path)
request_timeout = timeout or self.timeout
attempts = self.max_retries + 1 if method.upper() == "GET" else 1
last_error = None
for attempt in range(attempts):
if attempt:
time.sleep(min(self.retry_backoff * (2 ** (attempt - 1)), 30.0))
try:
response = self.session.request(method, url, timeout=request_timeout, **kwargs)
if response.status_code in self.RETRY_STATUSES and attempt < attempts - 1:
last_error = ProxmoxConnectionError(
f"Proxmox API {method} {path} returned transient HTTP {response.status_code}."
)
continue
response.raise_for_status()
return response
except requests.ConnectTimeout as exc:
last_error = ProxmoxConnectionError(
f"Proxmox API {method} {path} connect timed out after {request_timeout[0]} seconds."
)
if attempt >= attempts - 1:
raise last_error from exc
except requests.ReadTimeout as exc:
detail = f"Proxmox API {method} {path} read timed out after {request_timeout[1]} seconds."
if path == "access/ticket" and self.api_probe_succeeded:
detail += " API version endpoint responded, so the delay is likely in Proxmox authentication. Check username realm (e.g. root@pam), PAM/LDAP auth, two-factor auth, or use an API token."
elif getattr(self, "api_probe_error", "") and path != "version":
detail += f" Earlier unauthenticated version probe also failed: {self.api_probe_error}"
last_error = ProxmoxConnectionError(detail)
if attempt >= attempts - 1:
raise last_error from exc
except requests.exceptions.SSLError as exc:
raise ProxmoxConnectionError(
f"Proxmox API {method} {path} TLS validation failed. Disable Validate SSL for self-signed certificates or install the CA certificate. Details: {exc}"
) from exc
except requests.RequestException as exc:
last_error = ProxmoxConnectionError(f"Proxmox API {method} {path} failed: {exc}")
if attempt >= attempts - 1:
raise last_error from exc
if last_error is not None:
raise last_error
raise ProxmoxConnectionError(f"Proxmox API {method} {path} failed without a response.")
def _sdk_get(self, path, params=None):
try:
return self.sdk(path.strip("/")).get(**(params or {}))
except Exception as exc:
raise ProxmoxConnectionError(self._describe_sdk_error("GET", path, exc)) from exc
def _describe_sdk_error(self, method, path, exc):
if SDKTimeoutError is not None and isinstance(exc, SDKTimeoutError):
return f"Proxmox SDK {method} {path} timed out after {self.read_timeout} seconds."
if SDKConnectionError is not None and isinstance(exc, SDKConnectionError):
detail = str(exc).strip() or "connection failed"
return f"Proxmox SDK {method} {path} connection failed: {detail}"
if SDKResourceException is not None and isinstance(exc, SDKResourceException):
status_code = getattr(exc, "status_code", None)
status_message = getattr(exc, "status_message", "") or ""
content = (getattr(exc, "content", "") or "").strip()
errors = getattr(exc, "errors", None)
parts = []
if status_code:
parts.append(f"HTTP {status_code} {status_message}".strip())
elif status_message:
parts.append(status_message)
if content:
parts.append(content)
if errors:
parts.append(str(errors))
detail = " - ".join(parts) if parts else str(exc)
return f"Proxmox SDK {method} {path} failed: {detail}"
detail = str(exc).strip() or "unknown error"
return f"Proxmox SDK {method} {path} failed: {detail}"
def _get(self, path, params=None):
if self.uses_sdk:
return self._sdk_get(path, params=params)
response = self._request("GET", path, params=params)
return response.json().get("data")
def iter_virtual_machines(self):
resources = self._get("cluster/resources", params={"type": "vm"}) or []
for resource in resources:
vm_type = resource.get("type")
if vm_type == "lxc" and not self.endpoint.sync_lxc_containers:
continue
if vm_type not in ("qemu", "lxc"):
continue
yield self._build_vm_data(resource)
def _build_vm_data(self, resource):
vmid = resource["vmid"]
node = resource["node"]
vm_type = resource["type"]
config = self._get(f"nodes/{node}/{vm_type}/{vmid}/config") or {}
name = resource.get("name") or config.get("name") or config.get("hostname") or f"{vm_type}-{vmid}"
status = "active" if resource.get("status") == "running" else "offline"
memory_mb = self._get_memory_mb(resource, config)
vcpus = self._get_vcpus(resource, config)
if vm_type == "qemu":
disks = self._get_qemu_disks(config)
interfaces = self._get_qemu_interfaces(config)
self._merge_qemu_agent_interfaces(node, vmid, interfaces)
guest_os = self._get_qemu_guest_os(node, vmid)
else:
disks = self._get_lxc_disks(config)
interfaces = self._get_lxc_interfaces(config)
guest_os = "Linux Container"
disk_mb = sum(disk.size_mb for disk in disks) or self._bytes_to_mb(resource.get("maxdisk"))
return VMData(
name=str(name),
status=status,
vcpus=vcpus,
memory_mb=memory_mb,
disk_mb=disk_mb,
guest_os=guest_os,
disks=disks,
interfaces=interfaces,
)
@staticmethod
def _get_memory_mb(resource, config):
memory = config.get("memory")
if memory not in (None, ""):
try:
return int(memory)
except (TypeError, ValueError):
pass
return ProxmoxClient._bytes_to_mb(resource.get("maxmem"))
@staticmethod
def _get_vcpus(resource, config):
try:
sockets = int(config.get("sockets") or 1)
cores = int(config.get("cores") or resource.get("maxcpu") or 0)
return max(sockets * cores, 0)
except (TypeError, ValueError):
return int(resource.get("maxcpu") or 0)
@staticmethod
def _bytes_to_mb(value):
try:
return round(int(value or 0) / 1024 / 1024)
except (TypeError, ValueError):
return 0
def _get_qemu_disks(self, config):
disks = []
for key, value in sorted(config.items()):
if not self.DISK_KEY_PATTERN.match(key):
continue
parsed = self._parse_proxmox_value(value)
if parsed.get("media") == "cdrom":
continue
size_mb = self._parse_size_mb(parsed.get("size"))
if size_mb <= 0:
continue
disks.append(
DiskData(
name=key,
size_mb=size_mb,
description=f"Proxmox: {str(value)[:191]}",
)
)
return disks
def _get_lxc_disks(self, config):
disks = []
for key, value in sorted(config.items()):
if not self.LXC_DISK_KEY_PATTERN.match(key):
continue
parsed = self._parse_proxmox_value(value)
size_mb = self._parse_size_mb(parsed.get("size"))
if size_mb <= 0:
continue
disks.append(
DiskData(
name=key,
size_mb=size_mb,
description=f"Proxmox: {str(value)[:191]}",
)
)
return disks
def _get_qemu_interfaces(self, config):
interfaces = []
for key, value in sorted(config.items()):
if not self.NET_KEY_PATTERN.match(key):
continue
parsed = self._parse_proxmox_value(value)
mac_address = self._extract_proxmox_mac(parsed)
if not mac_address:
continue
bridge = parsed.get("bridge")
name = f"{key} ({bridge})" if bridge else key
interfaces.append(InterfaceData(name=name[:64], mac_address=mac_address))
return interfaces
def _get_lxc_interfaces(self, config):
interfaces = []
for key, value in sorted(config.items()):
if not self.NET_KEY_PATTERN.match(key):
continue
parsed = self._parse_proxmox_value(value)
mac_address = parsed.get("hwaddr")
if not mac_address:
continue
name = parsed.get("name") or key
ip_addresses = []
for ip_key in ("ip", "ip6"):
ip_value = parsed.get(ip_key)
if ip_value and ip_value not in ("dhcp", "manual", "auto"):
ip_addresses.append(ip_value)
interfaces.append(
InterfaceData(
name=str(name)[:64],
mac_address=mac_address,
ip_addresses=ip_addresses,
)
)
return interfaces
def _merge_qemu_agent_interfaces(self, node, vmid, interfaces):
try:
response = self._get(f"nodes/{node}/qemu/{vmid}/agent/network-get-interfaces") or {}
except (requests.RequestException, ProxmoxConnectionError):
return
agent_interfaces = response.get("result") if isinstance(response, dict) else response
if not isinstance(agent_interfaces, list):
return
interfaces_by_mac = {self._normalize_mac(interface.mac_address): interface for interface in interfaces}
for agent_interface in agent_interfaces:
mac_address = agent_interface.get("hardware-address")
normalized_mac = self._normalize_mac(mac_address)
if not normalized_mac:
continue
ip_addresses = []
for ip_data in agent_interface.get("ip-addresses") or []:
ip_address = ip_data.get("ip-address")
prefix = ip_data.get("prefix")
if not ip_address:
continue
if prefix is not None:
ip_addresses.append(f"{ip_address}/{prefix}")
else:
ip_addresses.append(ip_address)
if normalized_mac in interfaces_by_mac:
interfaces_by_mac[normalized_mac].ip_addresses = ip_addresses
else:
interfaces.append(
InterfaceData(
name=str(agent_interface.get("name") or f"NIC-{normalized_mac[-5:]}")[:64],
mac_address=normalized_mac,
ip_addresses=ip_addresses,
)
)
def _get_qemu_guest_os(self, node, vmid):
try:
response = self._get(f"nodes/{node}/qemu/{vmid}/agent/get-osinfo") or {}
except (requests.RequestException, ProxmoxConnectionError):
return ""
result = response.get("result") if isinstance(response, dict) else None
if not isinstance(result, dict):
return ""
return result.get("pretty-name") or result.get("name") or ""
@staticmethod
def _parse_proxmox_value(value):
parsed = {"_raw": str(value)}
parts = str(value).split(",")
if parts:
parsed["_first"] = parts[0]
for part in parts:
if "=" in part:
key, part_value = part.split("=", 1)
parsed[key.strip()] = part_value.strip()
return parsed
@staticmethod
def _extract_proxmox_mac(parsed):
for key in ("virtio", "e1000", "e1000e", "rtl8139", "vmxnet3", "hwaddr"):
if parsed.get(key):
return parsed[key]
return None
@staticmethod
def _parse_size_mb(value):
if value in (None, ""):
return 0
match = re.match(r"^\s*([0-9.]+)\s*([KMGTPE]?)(?:i?B)?\s*$", str(value), re.IGNORECASE)
if not match:
return 0
amount = float(match.group(1))
unit = match.group(2).upper()
factors = {
"": 1,
"K": 1 / 1024,
"M": 1,
"G": 1024,
"T": 1024 * 1024,
"P": 1024 * 1024 * 1024,
"E": 1024 * 1024 * 1024 * 1024,
}
if not unit and amount > 1024 * 1024:
return round(amount / 1024 / 1024)
return round(amount * factors.get(unit, 1))
@staticmethod
def _normalize_mac(mac_address):
if not mac_address:
return None
value = str(mac_address).strip().replace("-", ":").lower()
parts = value.split(":")
if len(parts) != 6:
return None
try:
return ":".join(f"{int(part, 16):02x}" for part in parts)
except ValueError:
return None
class VMwareImporter:
def __init__(self, endpoint, logger):
self.endpoint = endpoint
self.logger = logger
self.include_pattern = re.compile(endpoint.include_name_regex) if endpoint.include_name_regex else None
self.exclude_pattern = re.compile(endpoint.exclude_name_regex) if endpoint.exclude_name_regex else None
self.provider_name = "Proxmox" if endpoint.provider == EndpointProviderChoices.PROVIDER_PROXMOX else "VMware"
self.provider_name = "VMware"
self.interface_content_type = None
self.mac_address_content_type = None
def sync(self):
if self.endpoint.provider != EndpointProviderChoices.PROVIDER_VMWARE:
raise VMwareConnectionError("Legacy Proxmox endpoints are no longer supported. Delete this endpoint.")
result = SyncResult()
platforms = list(Platform.objects.all())
self.interface_content_type = ContentType.objects.get_for_model(VMInterface)
self.mac_address_content_type = ContentType.objects.get_for_model(VMInterface)
client_class = ProxmoxClient if self.endpoint.provider == EndpointProviderChoices.PROVIDER_PROXMOX else VMwareClient
with client_class(self.endpoint) as client:
with VMwareClient(self.endpoint) as client:
for vm_data in client.iter_virtual_machines():
result.seen += 1