nessi
150a69b60b
Add /ipam/discover endpoint to automatically import IP addresses from Proxmox clusters with error tracking and audit logging, implement ensure_discovered_network helper to create "discovered-ipam" network for auto-discovered IPs, add import_discovered_ips function to parse IP interfaces and create subnet/address records with assignment tracking, enhance ProxmoxProvider.enrich_work
120 lines
5.2 KiB
Python
120 lines
5.2 KiB
Python
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.services.providers.base import Provider, ProviderConnection
|
|
|
|
|
|
class ProxmoxProvider(Provider):
|
|
name = "proxmox"
|
|
|
|
def auth_header(self, token: str) -> str:
|
|
token = token.strip()
|
|
if token.startswith("PVEAPIToken="):
|
|
return token
|
|
return f"PVEAPIToken={token}"
|
|
|
|
async def test_connection(self, connection: ProviderConnection) -> dict[str, Any]:
|
|
headers = {"Authorization": self.auth_header(connection.token)}
|
|
async with httpx.AsyncClient(verify=connection.verify_tls, timeout=10) as client:
|
|
response = await client.get(
|
|
f"{connection.api_url.rstrip('/')}/api2/json/version",
|
|
headers=headers,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json().get("data", {})
|
|
|
|
async def sync_inventory(self, connection: ProviderConnection) -> dict[str, list[dict[str, Any]]]:
|
|
base_url = connection.api_url.rstrip("/")
|
|
headers = {"Authorization": self.auth_header(connection.token)}
|
|
async with httpx.AsyncClient(verify=connection.verify_tls, timeout=20) as client:
|
|
resources = await client.get(
|
|
f"{base_url}/api2/json/cluster/resources",
|
|
headers=headers,
|
|
)
|
|
resources.raise_for_status()
|
|
data = resources.json().get("data", [])
|
|
|
|
nodes = [item for item in data if item.get("type") == "node"]
|
|
workloads = [item for item in data if item.get("type") in {"qemu", "lxc"}]
|
|
async with httpx.AsyncClient(verify=connection.verify_tls, timeout=8) as client:
|
|
for workload in workloads:
|
|
await self.enrich_workload_ips(client, base_url, headers, workload)
|
|
networks = await self.list_networks(connection)
|
|
return {"nodes": nodes, "workloads": workloads, "networks": networks}
|
|
|
|
async def enrich_workload_ips(
|
|
self,
|
|
client: httpx.AsyncClient,
|
|
base_url: str,
|
|
headers: dict[str, str],
|
|
workload: dict[str, Any],
|
|
) -> None:
|
|
node = workload.get("node")
|
|
vmid = workload.get("vmid")
|
|
kind = workload.get("type")
|
|
workload["ip_addresses"] = []
|
|
if not node or not vmid:
|
|
return
|
|
|
|
if kind == "qemu":
|
|
try:
|
|
response = await client.get(
|
|
f"{base_url}/api2/json/nodes/{node}/qemu/{vmid}/agent/network-get-interfaces",
|
|
headers=headers,
|
|
)
|
|
if response.status_code >= 400:
|
|
return
|
|
interfaces = response.json().get("data", {}).get("result", [])
|
|
for interface in interfaces:
|
|
for address in interface.get("ip-addresses", []):
|
|
ip_address = address.get("ip-address")
|
|
prefix = address.get("prefix")
|
|
if ip_address and ":" not in ip_address and prefix is not None:
|
|
workload["ip_addresses"].append(f"{ip_address}/{prefix}")
|
|
except httpx.HTTPError:
|
|
return
|
|
|
|
if kind == "lxc":
|
|
try:
|
|
response = await client.get(
|
|
f"{base_url}/api2/json/nodes/{node}/lxc/{vmid}/config",
|
|
headers=headers,
|
|
)
|
|
if response.status_code >= 400:
|
|
return
|
|
config = response.json().get("data", {})
|
|
for key, value in config.items():
|
|
if key.startswith("net") and isinstance(value, str):
|
|
for part in value.split(","):
|
|
if part.startswith("ip="):
|
|
ip_address = part.removeprefix("ip=")
|
|
if ip_address != "dhcp" and ":" not in ip_address:
|
|
workload["ip_addresses"].append(ip_address)
|
|
except httpx.HTTPError:
|
|
return
|
|
|
|
async def list_networks(self, connection: ProviderConnection) -> list[dict[str, Any]]:
|
|
headers = {"Authorization": self.auth_header(connection.token)}
|
|
async with httpx.AsyncClient(verify=connection.verify_tls, timeout=20) as client:
|
|
resources = await client.get(
|
|
f"{connection.api_url.rstrip('/')}/api2/json/cluster/resources",
|
|
headers=headers,
|
|
)
|
|
resources.raise_for_status()
|
|
data = resources.json().get("data", [])
|
|
return [item for item in data if item.get("type") in {"network", "sdn"}]
|
|
|
|
async def preview_rules(self, connection: ProviderConnection, rules: list[dict[str, Any]]) -> dict[str, Any]:
|
|
return {
|
|
"provider": self.name,
|
|
"read_only": connection.read_only,
|
|
"generated": rules,
|
|
"warnings": ["Preview only. No Proxmox firewall changes were sent."],
|
|
}
|
|
|
|
async def apply_rules(self, connection: ProviderConnection, rules: list[dict[str, Any]]) -> dict[str, Any]:
|
|
if connection.read_only:
|
|
return {"applied": False, "reason": "Cluster is read-only", "rules": rules}
|
|
return {"applied": False, "reason": "Apply adapter intentionally requires explicit implementation", "rules": rules}
|