nessi
14e7710120
Add complete NexaFabric project structure including: - FastAPI backend with SQLAlchemy models, JWT auth, RBAC, audit logging, and provider interfaces - React + TypeScript frontend with Vite, Tailwind CSS, TanStack Query, and Zustand - Docker Compose configuration for PostgreSQL, Redis, API, worker, frontend, and nginx - GitHub Actions and GitLab CI workflows for testing, linting, building, and security scanning - Environment
57 lines
2.6 KiB
Python
57 lines
2.6 KiB
Python
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.services.providers.base import Provider, ProviderConnection
|
|
|
|
|
|
class ProxmoxProvider(Provider):
|
|
name = "proxmox"
|
|
|
|
async def test_connection(self, connection: ProviderConnection) -> dict[str, Any]:
|
|
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={"Authorization": connection.token},
|
|
)
|
|
response.raise_for_status()
|
|
return response.json().get("data", {})
|
|
|
|
async def sync_inventory(self, connection: ProviderConnection) -> dict[str, list[dict[str, Any]]]:
|
|
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={"Authorization": connection.token},
|
|
)
|
|
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"}]
|
|
networks = await self.list_networks(connection)
|
|
return {"nodes": nodes, "workloads": workloads, "networks": networks}
|
|
|
|
async def list_networks(self, connection: ProviderConnection) -> list[dict[str, Any]]:
|
|
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={"Authorization": connection.token},
|
|
)
|
|
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}
|
|
|