smolkik-code
81e0ba3a4a
- fetch_sessions now queries 5 URL variants and merges results by unique session id (no duplicates) - First query: unfiltered with 3x limit (gets recent sessions of all types) - Then: sessionType=BackupJob, FileBackupJob, AgentBackupJob - Ensures all backup job types appear in dashboard
415 lines
15 KiB
Python
415 lines
15 KiB
Python
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():
|
|
for path in (
|
|
"/api/v1/jobs/states",
|
|
"/api/v1/jobs",
|
|
"/api/v1/jobs/states?source=Backup",
|
|
"/api/v1/jobs?source=Backup",
|
|
"/api/v1/jobs/states?typeFilter=Backup",
|
|
"/api/v1/jobs?typeFilter=Backup",
|
|
"/api/v1/agents/policies/states",
|
|
"/api/v1/agents/policies",
|
|
"/api/v1/protectionGroups",
|
|
"/api/v1/agents/protectionGroups",
|
|
):
|
|
try:
|
|
return await self._fetch_json_list(path)
|
|
except Exception:
|
|
continue
|
|
try:
|
|
return await self._fetch_legacy_jobs()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
legacy_url = f"https://{self.host}:9398/api/jobs"
|
|
resp = await self._http.get(legacy_url, headers={"Accept": "application/json"})
|
|
if resp.status_code < 400:
|
|
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]
|
|
if jobs:
|
|
return [self._normalize_legacy_job(j) for j in jobs]
|
|
except Exception:
|
|
pass
|
|
raise ConnectionError("all job/policy endpoints failed (v12 REST + legacy XML)")
|
|
return await self._fetch_legacy_jobs()
|
|
|
|
async def fetch_sessions(self, limit: int = 50) -> list[dict]:
|
|
if self._is_v12():
|
|
seen: set[str] = set()
|
|
all_sessions: list[dict] = []
|
|
for path in (
|
|
f"/api/v1/sessions?limit={limit * 3}",
|
|
f"/api/v1/sessions?limit={limit}&sessionType=BackupJob",
|
|
f"/api/v1/sessions?limit={limit}&sessionType=BackupJob&typeFilter=Backup",
|
|
f"/api/v1/sessions?limit={limit}&sessionType=FileBackupJob",
|
|
f"/api/v1/sessions?limit={limit}&sessionType=AgentBackupJob",
|
|
):
|
|
try:
|
|
items = await self._fetch_json_list(path)
|
|
for s in items:
|
|
sid = s.get("id") or s.get("Id") or ""
|
|
if sid and sid not in seen:
|
|
seen.add(sid)
|
|
all_sessions.append(s)
|
|
except Exception:
|
|
continue
|
|
return all_sessions
|
|
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_agent_policies(self) -> list[dict]:
|
|
if self._is_v12():
|
|
for path in (
|
|
"/api/v1/agents/policies/states",
|
|
"/api/v1/agents/policies",
|
|
):
|
|
try:
|
|
return await self._fetch_json_list(path)
|
|
except Exception:
|
|
continue
|
|
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+)([^>]*)>(.*?)</\1>", 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})"
|