From ba8a413e85001535f961eef399ff7dd77bbc523f Mon Sep 17 00:00:00 2001 From: smolkik-code Date: Tue, 9 Jun 2026 17:10:35 +0700 Subject: [PATCH] Initial commit: Veeam multi-instance monitoring dashboard - FastAPI backend with CRUD for Veeam instances - Veeam REST API client supporting v9.5-v13 (legacy + OAuth2) - Multi-version detection and auth fallback - SQLite caching layer - Web dashboard with summary cards, instance list, detail tabs - Docker packaging --- .dockerignore | 5 + .gitignore | 8 + Dockerfile | 18 ++ config.py | 20 ++ database.py | 133 ++++++++ docker-compose.yml | 17 + main.py | 445 ++++++++++++++++++++++++++ requirements.txt | 9 + static/index.html | 769 +++++++++++++++++++++++++++++++++++++++++++++ veeam_client.py | 359 +++++++++++++++++++++ 10 files changed, 1783 insertions(+) create mode 100644 .dockerignore create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 config.py create mode 100644 database.py create mode 100644 docker-compose.yml create mode 100644 main.py create mode 100644 requirements.txt create mode 100644 static/index.html create mode 100644 veeam_client.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..e596c45 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,5 @@ +__pycache__ +*.pyc +.env +.git +veeam_dashboard.db diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b00265f --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +__pycache__/ +*.pyc +.venv/ +venv/ +*.db +.env +.DS_Store +*.tar.gz diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..67e27ce --- /dev/null +++ b/Dockerfile @@ -0,0 +1,18 @@ +FROM python:3.13-slim + +WORKDIR /app + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY . . + +RUN python -c "from main import app; print('App loaded')" && \ + rm -rf __pycache__ */__pycache__ + +EXPOSE 8080 + +ENV DB_URL=sqlite+aiosqlite:////data/veeam_dashboard.db +ENV PYTHONUNBUFFERED=1 + +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"] diff --git a/config.py b/config.py new file mode 100644 index 0000000..f7ce234 --- /dev/null +++ b/config.py @@ -0,0 +1,20 @@ +from pydantic_settings import BaseSettings +from pathlib import Path + + +class Settings(BaseSettings): + app_name: str = "Veeam Dashboard" + db_url: str = "sqlite+aiosqlite:///./veeam_dashboard.db" + cache_ttl: int = 60 + request_timeout: int = 30 + host: str = "0.0.0.0" + port: int = 8080 + secret_key: str = "change-me-in-production" + + class Config: + env_file = ".env" + env_file_encoding = "utf-8" + + +settings = Settings() +BASE_DIR = Path(__file__).parent diff --git a/database.py b/database.py new file mode 100644 index 0000000..ced7551 --- /dev/null +++ b/database.py @@ -0,0 +1,133 @@ +from datetime import datetime +from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column +from sqlalchemy import String, Integer, Boolean, DateTime, Text, select, delete as sa_delete +from config import settings +import json + + +engine = create_async_engine(settings.db_url, echo=False) +async_session = async_sessionmaker(engine, expire_on_commit=False) + + +class Base(DeclarativeBase): + pass + + +class Instance(Base): + __tablename__ = "instances" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + name: Mapped[str] = mapped_column(String(255), nullable=False) + host: Mapped[str] = mapped_column(String(255), nullable=False) + port: Mapped[int] = mapped_column(Integer, nullable=False) + username: Mapped[str] = mapped_column(String(255), nullable=False) + password: Mapped[str] = mapped_column(String(255), nullable=False) + version: Mapped[str] = mapped_column(String(32), default="auto") + notes: Mapped[str] = mapped_column(Text, default="") + is_active: Mapped[bool] = mapped_column(Boolean, default=True) + created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + +class CacheEntry(Base): + __tablename__ = "cache" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + instance_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True) + key: Mapped[str] = mapped_column(String(64), nullable=False) + data: Mapped[str] = mapped_column(Text, nullable=False) + fetched_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + + +async def init_db(): + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + +async def get_instances() -> list[Instance]: + async with async_session() as session: + result = await session.execute(select(Instance).order_by(Instance.name)) + return list(result.scalars().all()) + + +async def get_instance(instance_id: int) -> Instance | None: + async with async_session() as session: + result = await session.execute(select(Instance).where(Instance.id == instance_id)) + return result.scalar_one_or_none() + + +async def add_instance(name: str, host: str, port: int, username: str, password: str, + version: str = "auto", notes: str = "") -> Instance: + async with async_session() as session: + inst = Instance( + name=name, host=host, port=port, username=username, + password=password, version=version, notes=notes + ) + session.add(inst) + await session.commit() + await session.refresh(inst) + return inst + + +async def update_instance(instance_id: int, **kwargs) -> Instance | None: + async with async_session() as session: + result = await session.execute(select(Instance).where(Instance.id == instance_id)) + inst = result.scalar_one_or_none() + if not inst: + return None + for key, value in kwargs.items(): + if hasattr(inst, key): + setattr(inst, key, value) + inst.updated_at = datetime.utcnow() + await session.commit() + await session.refresh(inst) + return inst + + +async def delete_instance(instance_id: int) -> bool: + async with async_session() as session: + result = await session.execute(select(Instance).where(Instance.id == instance_id)) + inst = result.scalar_one_or_none() + if not inst: + return False + await session.delete(inst) + await session.execute(sa_delete(CacheEntry).where(CacheEntry.instance_id == instance_id)) + await session.commit() + return True + + +async def set_cache(instance_id: int, key: str, data: dict): + async with async_session() as session: + result = await session.execute( + select(CacheEntry).where( + CacheEntry.instance_id == instance_id, + CacheEntry.key == key + ) + ) + entry = result.scalar_one_or_none() + if entry: + entry.data = json.dumps(data) + entry.fetched_at = datetime.utcnow() + else: + entry = CacheEntry( + instance_id=instance_id, + key=key, + data=json.dumps(data) + ) + session.add(entry) + await session.commit() + + +async def get_cache(instance_id: int, key: str) -> dict | None: + async with async_session() as session: + result = await session.execute( + select(CacheEntry).where( + CacheEntry.instance_id == instance_id, + CacheEntry.key == key + ) + ) + entry = result.scalar_one_or_none() + if entry: + return {"data": json.loads(entry.data), "fetched_at": entry.fetched_at.isoformat()} + return None diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..57abf52 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,17 @@ +version: "3.8" + +services: + veeam-dashboard: + build: . + container_name: veeam-dashboard + restart: unless-stopped + ports: + - "8080:8080" + volumes: + - data:/data + environment: + - DB_URL=sqlite+aiosqlite:////data/veeam_dashboard.db + - CACHE_TTL=60 + +volumes: + data: diff --git a/main.py b/main.py new file mode 100644 index 0000000..ecb6154 --- /dev/null +++ b/main.py @@ -0,0 +1,445 @@ +import asyncio +import json +import time +from contextlib import asynccontextmanager +from datetime import datetime +from pathlib import Path + +from fastapi import FastAPI, HTTPException +from fastapi.responses import FileResponse, JSONResponse +from fastapi.middleware.cors import CORSMiddleware +from pydantic import BaseModel + +from config import settings, BASE_DIR +from database import ( + init_db, get_instances, get_instance, add_instance, + update_instance, delete_instance, set_cache, get_cache +) +from veeam_client import VeeamClient, port_for_version, detect_version_from_host + + +_cache: dict[str, tuple[float, dict]] = {} + + +def _cached(key: str, ttl: int = None) -> dict | None: + ttl = ttl or settings.cache_ttl + entry = _cache.get(key) + if entry and (time.time() - entry[0]) < ttl: + return entry[1] + return None + + +def _set_cache_inmem(key: str, data: dict): + _cache[key] = (time.time(), data) + + +# ─── Pydantic models ─────────────────────────────────────────────── + +class InstanceCreate(BaseModel): + name: str + host: str + port: int | None = None + username: str + password: str + version: str = "auto" + notes: str = "" + + +class InstanceUpdate(BaseModel): + name: str | None = None + host: str | None = None + port: int | None = None + username: str | None = None + password: str | None = None + version: str | None = None + is_active: bool | None = None + notes: str | None = None + + +class InstanceResponse(BaseModel): + id: int + name: str + host: str + port: int + username: str + version: str + notes: str + is_active: bool + created_at: str + updated_at: str + + +class DashboardResponse(BaseModel): + instances: list + summary: dict + updated_at: str + + +# ─── App lifecycle ───────────────────────────────────────────────── + +@asynccontextmanager +async def lifespan(app: FastAPI): + await init_db() + yield + +app = FastAPI(title=settings.app_name, version="1.0", lifespan=lifespan) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +# ─── Helpers ─────────────────────────────────────────────────────── + +def _serialize_instance(inst) -> dict: + return { + "id": inst.id, + "name": inst.name, + "host": inst.host, + "port": inst.port, + "username": inst.username, + "version": inst.version, + "notes": inst.notes, + "is_active": inst.is_active, + "created_at": inst.created_at.isoformat() if inst.created_at else "", + "updated_at": inst.updated_at.isoformat() if inst.updated_at else "", + } + + +def _safe_str(v) -> str: + if v is None: + return "" + if isinstance(v, dict): + return str(v.get("result") or v.get("Result") or "unknown") + return str(v) + + +async def _collect_instance(inst) -> dict | None: + port = inst.port or port_for_version(inst.version) + ver = inst.version + if ver == "auto": + ver = detect_version_from_host(inst.host) + port = port_for_version(ver) + + client = VeeamClient(inst.host, port, inst.username, inst.password, ver) + try: + role = await client.connect() + jobs_data = [] + sessions_data = [] + repos_data = [] + repo_states = [] + alarms_data = [] + agents_data = [] + errors = [] + + try: + jobs_data = await client.fetch_jobs() + except Exception as e: + errors.append(f"jobs: {e}") + + try: + sessions_data = await client.fetch_sessions(100) + except Exception as e: + errors.append(f"sessions: {e}") + + try: + repos_data = await client.fetch_repositories() + except Exception as e: + errors.append(f"repos: {e}") + + try: + repo_states = await client.fetch_repository_states() + except Exception as e: + errors.append(f"repo states: {e}") + + try: + alarms_data = await client.fetch_alarms() + except Exception as e: + pass + + try: + agents_data = await client.fetch_agents() + except Exception as e: + errors.append(f"agents: {e}") + + for s in sessions_data: + if isinstance(s.get("result"), dict): + s["result"] = s["result"].get("result") or s["result"].get("Result") or "unknown" + if "jobName" in s and "job_name" not in s: + s["job_name"] = s["jobName"] + if "creationTime" in s and "creation_time" not in s: + s["creation_time"] = s["creationTime"] + if "endTime" in s and "end_time" not in s: + s["end_time"] = s["endTime"] + + job_last_backup = {} + for s in sessions_data: + s_name = s.get("jobName") or s.get("name") or s.get("Name") or "" + s_end = s.get("endTime") or s.get("end_time") or s.get("creationTime") or "" + if s_name and s_end: + if s_name not in job_last_backup or s_end > job_last_backup[s_name]: + job_last_backup[s_name] = s_end + + for j in jobs_data: + if isinstance(j.get("result"), dict): + j["result"] = j["result"].get("result") or j["result"].get("Result") or "unknown" + if not j.get("result") and j.get("lastResult"): + j["result"] = j["lastResult"] + if not j.get("state") and j.get("status"): + j["state"] = j["status"] + if "scheduleEnabled" in j and "schedule_enabled" not in j: + j["schedule_enabled"] = j["scheduleEnabled"] + if "restorePointsToKeep" in j and "restore_points_to_keep" not in j: + j["restore_points_to_keep"] = j["restorePointsToKeep"] + if "nextRun" in j and "next_run" not in j: + j["next_run"] = j["nextRun"] + j_name = j.get("name") or j.get("Name") or "" + if j_name in job_last_backup: + j["last_backup"] = job_last_backup[j_name] + + job_count = len(jobs_data) + success = sum(1 for j in jobs_data if _safe_str(j.get("result") or "").lower() in ("success", "true")) + failed = sum(1 for j in jobs_data if _safe_str(j.get("result") or "").lower() in ("failed", "error", "false")) + running = sum(1 for j in jobs_data if _safe_str(j.get("state") or "").lower() == "running") + warning = sum(1 for j in jobs_data if "warning" in _safe_str(j.get("result") or "").lower()) + disabled = sum(1 for j in jobs_data if j.get("schedule_enabled") is False or _safe_str(j.get("state") or "").lower() == "disabled") + + last_sessions = sessions_data[:20] if sessions_data else [] + + result = { + "id": inst.id, + "name": inst.name, + "host": inst.host, + "port": port, + "version": client.version, + "is_online": True, + "user_role": role, + "errors": errors, + "jobs": jobs_data, + "jobs_summary": { + "total": job_count, + "success": success, + "failed": failed, + "running": running, + "warning": warning, + "disabled": disabled, + }, + "disabled_policies": disabled, + "recent_sessions": last_sessions, + "repositories": repos_data, + "repository_states": repo_states, + "alarms": alarms_data, + "agents": agents_data, + } + + await set_cache(inst.id, "dashboard", result) + return result + + except Exception as e: + error_result = { + "id": inst.id, + "name": inst.name, + "host": inst.host, + "port": port, + "version": ver, + "is_online": False, + "error": str(e), + "jobs": [], + "jobs_summary": {"total": 0, "success": 0, "failed": 0, "running": 0, "warning": 0, "disabled": 0}, + "disabled_policies": 0, + "recent_sessions": [], + "repositories": [], + "repository_states": [], + "alarms": [], + "agents": [], + } + return error_result + finally: + await client.close() + + +async def _collect_all() -> list[dict]: + instances = await get_instances() + tasks = [_collect_instance(inst) for inst in instances if inst.is_active] + if not tasks: + return [] + results = await asyncio.gather(*tasks, return_exceptions=True) + output = [] + for r in results: + if isinstance(r, Exception): + continue + if r: + output.append(r) + return output + + +def _build_summary(instance_data: list[dict]) -> dict: + total_instances = len(instance_data) + online = sum(1 for i in instance_data if i.get("is_online")) + total_jobs = sum(i.get("jobs_summary", {}).get("total", 0) for i in instance_data) + total_success = sum(i.get("jobs_summary", {}).get("success", 0) for i in instance_data) + total_failed = sum(i.get("jobs_summary", {}).get("failed", 0) for i in instance_data) + total_running = sum(i.get("jobs_summary", {}).get("running", 0) for i in instance_data) + total_warning = sum(i.get("jobs_summary", {}).get("warning", 0) for i in instance_data) + total_alarms = sum(len(i.get("alarms", [])) for i in instance_data) + total_repos = sum(len(i.get("repositories", [])) for i in instance_data) + + return { + "total_instances": total_instances, + "online_instances": online, + "offline_instances": total_instances - online, + "total_jobs": total_jobs, + "successful_jobs": total_success, + "failed_jobs": total_failed, + "running_jobs": total_running, + "warning_jobs": total_warning, + "total_alarms": total_alarms, + "total_repositories": total_repos, + } + + +# ─── API Endpoints ───────────────────────────────────────────────── + +@app.get("/api/health") +async def health(): + return {"status": "ok"} + + +# Instance CRUD + +@app.get("/api/instances", response_model=list[InstanceResponse]) +async def list_instances(): + instances = await get_instances() + return [_serialize_instance(i) for i in instances] + + +@app.post("/api/instances", response_model=InstanceResponse, status_code=201) +async def create_instance(data: InstanceCreate): + port = data.port or port_for_version(data.version) + inst = await add_instance( + name=data.name, + host=data.host, + port=port, + username=data.username, + password=data.password, + version=data.version, + notes=data.notes, + ) + return _serialize_instance(inst) + + +@app.get("/api/instances/{instance_id}", response_model=InstanceResponse) +async def read_instance(instance_id: int): + inst = await get_instance(instance_id) + if not inst: + raise HTTPException(404, "Instance not found") + return _serialize_instance(inst) + + +@app.put("/api/instances/{instance_id}", response_model=InstanceResponse) +async def edit_instance(instance_id: int, data: InstanceUpdate): + updates = data.model_dump(exclude_unset=True) + if not updates: + raise HTTPException(400, "No fields to update") + inst = await update_instance(instance_id, **updates) + if not inst: + raise HTTPException(404, "Instance not found") + return _serialize_instance(inst) + + +@app.delete("/api/instances/{instance_id}") +async def remove_instance(instance_id: int): + ok = await delete_instance(instance_id) + if not ok: + raise HTTPException(404, "Instance not found") + _cache.pop(f"instance_{instance_id}", None) + return {"ok": True} + + +# Test connection — MUST be before {instance_id} routes to avoid path conflicts + +@app.post("/api/test-connection") +async def test_connection_raw(data: InstanceCreate): + port = data.port or port_for_version(data.version) + ver = data.version + if ver == "auto": + ver = detect_version_from_host(data.host) + port = port_for_version(ver) + client = VeeamClient(data.host, port, data.username, data.password, ver) + try: + role = await client.connect() + return {"ok": True, "role": role, "version": client.version} + except Exception as e: + return JSONResponse(status_code=502, content={"ok": False, "error": str(e)}) + finally: + await client.close() + + +@app.post("/api/instances/{instance_id}/test") +async def test_instance(instance_id: int): + inst = await get_instance(instance_id) + if not inst: + raise HTTPException(404, "Instance not found") + result = await _collect_instance(inst) + if result and result.get("is_online"): + return {"ok": True, "role": result.get("user_role"), "version": result.get("version")} + return JSONResponse( + status_code=502, + content={"ok": False, "error": result.get("error", "Unknown error") if result else "No response"} + ) + + +# Dashboard + +@app.get("/api/dashboard") +async def dashboard(force_refresh: bool = False): + if not force_refresh: + cached = _cached("dashboard_aggregate") + if cached: + return cached + + instance_data = await _collect_all() + summary = _build_summary(instance_data) + result = { + "instances": instance_data, + "summary": summary, + "updated_at": datetime.utcnow().isoformat(), + } + _set_cache_inmem("dashboard_aggregate", result) + return result + + +# Per-instance data + +@app.get("/api/instances/{instance_id}/data") +async def instance_data(instance_id: int, force_refresh: bool = False): + inst = await get_instance(instance_id) + if not inst: + raise HTTPException(404, "Instance not found") + + if not force_refresh: + cached = _cached(f"instance_{instance_id}") + if cached: + return cached + + result = await _collect_instance(inst) + _set_cache_inmem(f"instance_{instance_id}", result) + if not result.get("is_online"): + return JSONResponse(status_code=502, content=result) + return result + + +# Serve frontend + +@app.get("/") +async def index(): + return FileResponse(str(BASE_DIR / "static" / "index.html")) + + +if __name__ == "__main__": + import uvicorn + uvicorn.run("main:app", host=settings.host, port=settings.port, reload=True) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..66774e7 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,9 @@ +fastapi>=0.110.0 +uvicorn[standard]>=0.29.0 +httpx>=0.27.0 +sqlalchemy[asyncio]>=2.0.30 +greenlet>=3.0.0 +aiosqlite>=0.20.0 +pydantic>=2.7.0 +pydantic-settings>=2.2.0 +pycryptodome>=3.20.0 diff --git a/static/index.html b/static/index.html new file mode 100644 index 0000000..f3f0d03 --- /dev/null +++ b/static/index.html @@ -0,0 +1,769 @@ + + + + + +Veeam Dashboard + + + + +
+
+

Veeam Dashboard

+
+ + +
+
+ +
+ +
+
—
Инстансы
+
—
Задания
+
—
Успешно
+
—
Ошибки
+
—
Warning
+
—
Running
+
+ +

Нет добавленных инстансов

+ +
+ Загрузка... + +
+
+ + + + + + + diff --git a/veeam_client.py b/veeam_client.py new file mode 100644 index 0000000..6a2e7f1 --- /dev/null +++ b/veeam_client.py @@ -0,0 +1,359 @@ +import base64 +import httpx +import re +from datetime import datetime + + +# Legacy API (v9.5-v11) — used in x-api-version header for session auth +API_VERSIONS_LEGACY = { + "9.5": "1.0-rev1", + "10": "1.0-rev1", + "11": "1.0-rev1", +} + +# v12+ OAuth2 API — used in x-api-version header for /api/oauth2/token and data requests +# 1.0-rev2 is for v11 (legacy). For v12+, use 1.1-rev2 as safe default. +API_VERSIONS_V12 = { + "auto": "1.1-rev2", + "12": "1.1-rev0", + "12.1": "1.1-rev1", + "12.2": "1.1-rev2", + "12.3": "1.2-rev1", + "13": "1.3-rev1", +} + + +def detect_version_from_host(host: str) -> str: + h = host.lower() + for v in ["12.3", "12.2", "12.1", "12", "11", "10", "9.5"]: + if v in h: + return v + return "auto" + + +def port_for_version(version: str) -> int: + try: + ver = float(version) + except ValueError: + return 9419 + return 9419 if ver >= 12 else 9398 + + +class VeeamClient: + def __init__(self, host: str, port: int, username: str, password: str, version: str = "auto"): + self.host = host + self.port = port + self.username = username + self.password = password + self.version = version + self.base_url = f"https://{host}:{port}" + self.token = None + self.session_id = None + self._v12_mode = False + self._legacy_api_version = "1.0-rev1" + self._v12_api_version = "1.1-rev2" + self._http = httpx.AsyncClient(verify=False, timeout=30.0) + + async def close(self): + if self.session_id: + try: + await self._http.delete( + f"{self.base_url}/api/session/{self.session_id}", + headers=self._legacy_headers() + ) + except Exception: + pass + await self._http.aclose() + + def _legacy_headers(self) -> dict: + h = { + "Accept": "application/json", + "x-api-version": self._legacy_api_version, + } + if self.session_id: + h["X-RestSvcSessionId"] = self.session_id + return h + + def _v12_headers(self) -> dict: + h = { + "Accept": "application/json", + "Content-Type": "application/json", + "x-api-version": self._v12_api_version, + } + if self.token: + h["Authorization"] = f"Bearer {self.token}" + return h + + async def connect(self) -> str: + return await self._try_connect() + + def _is_v12(self) -> bool: + return self._v12_mode + + def _should_try_v12_first(self) -> bool: + if self.version == "auto": + return True + try: + return float(self.version) >= 12 + except ValueError: + return True + + def _resolve_v12_revision(self) -> str: + return API_VERSIONS_V12.get(self.version, "1.0-rev2") + + def _resolve_legacy_revision(self) -> str: + return API_VERSIONS_LEGACY.get(self.version, "1.0-rev1") + + async def _try_connect(self) -> str: + combos = [] + default_port = self.port or 9419 + if self.version == "auto": + combos = [ + (9419, "v12"), + (9398, "legacy"), + ] + elif self._should_try_v12_first(): + combos = [ + (default_port, "v12"), + ((9398 if default_port == 9419 else 9419), "legacy"), + ] + else: + combos = [ + (default_port, "legacy"), + ((9419 if default_port == 9398 else 9398), "v12"), + ] + + last_error = "" + for port, method in combos: + self.port = port + self.base_url = f"https://{self.host}:{self.port}" + try: + if method == "v12": + self._v12_api_version = self._resolve_v12_revision() + result = await self._connect_v12() + self._v12_mode = True + else: + self._legacy_api_version = self._resolve_legacy_revision() + result = await self._connect_legacy() + self._v12_mode = False + return result + except Exception as e: + last_error = f"{type(e).__name__}: {e}" + continue + + raise ConnectionError( + f"Cannot connect to {self.host} (tried ports {[c[0] for c in combos]}). " + f"Last error: {last_error}" + ) + + async def _connect_legacy(self) -> str: + auth_str = base64.b64encode(f"{self.username}:{self.password}".encode()).decode() + resp = await self._http.post( + f"{self.base_url}/api/session", + headers={ + "Accept": "application/json", + "x-api-version": self._legacy_api_version, + "Authorization": f"Basic {auth_str}", + }, + ) + data = self._parse(resp) + sid = data.get("sessionId") or data.get("SessionId") + if not sid: + msg = data.get("_raw", resp.text[:300]) if isinstance(data, dict) else str(data)[:300] + raise ConnectionError(f"Legacy auth failed ({resp.status_code}): {msg}") + self.session_id = sid + return data.get("userRole") or data.get("UserRole") or "unknown" + + async def _connect_v12(self) -> str: + resp = await self._http.post( + f"{self.base_url}/api/oauth2/token", + data={ + "grant_type": "password", + "username": self.username, + "password": self.password, + }, + headers={ + "Accept": "application/json", + "Content-Type": "application/x-www-form-urlencoded", + "x-api-version": self._v12_api_version, + }, + ) + if resp.status_code == 404: + raise ConnectionError( + f"v12 auth endpoint not found on {self.base_url} " + f"(404 — REST API may be disabled or wrong port)") + try: + data = resp.json() + except Exception: + raise ConnectionError( + f"v12 auth failed ({resp.status_code}): {resp.text[:200]}") + self.token = data.get("access_token") + if not self.token: + raise ConnectionError( + f"Token auth failed ({resp.status_code}): " + f"{data.get('error_description', resp.text[:200])}") + return data.get("user_role") or data.get("username", "unknown") + + def _resolve_version(self) -> str: + if self.version != "auto": + return self.version + detected = detect_version_from_host(self.host) + if detected != "auto": + return detected + return "12" + + async def fetch_jobs(self) -> list[dict]: + if self._is_v12(): + last_err = None + for url in ("/api/v1/jobs/states", "/api/v1/jobs", "/api/v1/jobs/states?typeFilter=Backup"): + try: + return await self._fetch_json_list(url) + except Exception as e: + last_err = e + continue + raise ConnectionError(f"all job endpoints failed: {last_err}") + return await self._fetch_legacy_jobs() + + async def fetch_sessions(self, limit: int = 50) -> list[dict]: + if self._is_v12(): + return await self._fetch_json_list(f"/api/v1/sessions?limit={limit}") + return await self._fetch_legacy_sessions(limit) + + async def fetch_repositories(self) -> list[dict]: + if self._is_v12(): + return await self._fetch_json_list("/api/v1/backupInfrastructure/repositories") + return await self._fetch_legacy_repos() + + async def fetch_repository_states(self) -> list[dict]: + if self._is_v12(): + return await self._fetch_json_list("/api/v1/backupInfrastructure/repositories/states") + return [] + + async def fetch_alarms(self) -> list[dict]: + if self._is_v12(): + return await self._fetch_json_list("/api/v1/alarms") + return [] + + async def fetch_backup_servers(self) -> list[dict]: + if self._is_v12(): + return await self._fetch_json_list("/api/v1/backupInfrastructure/managedServers") + return await self._fetch_legacy_list("/api/backupServers", "BackupServer") + + async def fetch_agents(self) -> list[dict]: + if self._is_v12(): + return await self._fetch_json_list("/api/v1/agents/protectedComputers") + return [] + + async def _fetch_json_list(self, path: str) -> list[dict]: + resp = await self._http.get( + f"{self.base_url}{path}", + headers=self._v12_headers(), + ) + if resp.status_code >= 400: + raise ConnectionError(f"GET {path}: {resp.status_code} {resp.text[:200]}") + data = resp.json() + return data.get("data") or data.get("items") or data.get("results") or data.get("jobs") or (data if isinstance(data, list) else []) + + async def _fetch_legacy_jobs(self) -> list[dict]: + resp = await self._http.get( + f"{self.base_url}/api/jobs", + headers=self._legacy_headers(), + ) + if resp.status_code >= 400: + raise ConnectionError(f"GET /api/jobs: {resp.status_code} {resp.text[:200]}") + data = self._parse(resp) + jobs = data.get("Jobs") or data.get("jobs") or data.get("Refs", {}).get("Job") or [] + if isinstance(jobs, dict): + jobs = [jobs] + return [self._normalize_legacy_job(j) for j in jobs] + + async def _fetch_legacy_sessions(self, limit: int) -> list[dict]: + resp = await self._http.get( + f"{self.base_url}/api/sessions?limit={limit}", + headers=self._legacy_headers(), + ) + if resp.status_code >= 400: + return [] + data = self._parse(resp) + sessions = data.get("Sessions") or data.get("sessions") or [] + if isinstance(sessions, dict): + sessions = [sessions] + return [self._normalize_legacy_session(s) for s in sessions] + + async def _fetch_legacy_repos(self) -> list[dict]: + return await self._fetch_legacy_list("/api/repositories", "BackupRepository") + + async def _fetch_legacy_list(self, path: str, item_key: str) -> list[dict]: + resp = await self._http.get( + f"{self.base_url}{path}", + headers=self._legacy_headers(), + ) + if resp.status_code >= 400: + return [] + data = self._parse(resp) + items = data.get(f"{item_key}s") or data.get(item_key) or [] + if isinstance(items, dict): + items = [items] + return items + + def _parse(self, resp: httpx.Response) -> dict: + ct = resp.headers.get("content-type", "") + if "json" in ct: + try: + return resp.json() + except Exception: + pass + text = resp.text + if text.strip().startswith("<"): + return self._xml_to_dict_simple(text) + try: + return resp.json() + except Exception: + return {"_raw": text[:500]} + + def _xml_to_dict_simple(self, xml: str) -> dict: + result = {} + tags = re.findall(r"<(\w+)([^>]*)>(.*?)", xml, re.DOTALL) + for tag_name, attrs_str, content in tags: + attrs = dict(re.findall(r'(\w+)="([^"]*)"', attrs_str)) + sub = self._xml_to_dict_simple(content) if "<" in content else content.strip() + if attrs: + entry = {"_text": sub} if sub else {} + entry.update(attrs) + else: + entry = sub + if tag_name not in result: + result[tag_name] = entry + else: + existing = result[tag_name] + if not isinstance(existing, list): + result[tag_name] = [existing] + result[tag_name].append(entry) + return result + + def _normalize_legacy_job(self, job: dict) -> dict: + return { + "id": job.get("@Id") or job.get("id") or job.get("Id", ""), + "name": job.get("@Name") or job.get("name") or job.get("Name", ""), + "type": job.get("@Type") or job.get("type") or job.get("JobType", ""), + "description": job.get("@Description") or job.get("description", ""), + "schedule_enabled": job.get("@ScheduleEnabled") or job.get("scheduleEnabled", ""), + "is_active": job.get("@IsBackupToD2D") or job.get("isActive", ""), + "next_run": job.get("@NextRun") or job.get("nextRun", ""), + } + + def _normalize_legacy_session(self, s: dict) -> dict: + return { + "id": s.get("@Id") or s.get("id", ""), + "name": s.get("@Name") or s.get("name", ""), + "job_name": s.get("@JobName") or s.get("jobName", ""), + "job_id": s.get("@JobId") or s.get("jobId", ""), + "type": s.get("@Type") or s.get("type", ""), + "result": s.get("@Result") or s.get("result", ""), + "state": s.get("@State") or s.get("state", ""), + "creation_time": s.get("@CreationTime") or s.get("creationTime", ""), + "end_time": s.get("@EndTime") or s.get("endTime", ""), + } + + def __repr__(self): + return f"VeeamClient({self.host}:{self.port} v{self.version})"