inecs
e8b08526d1
Return native JSON with the missing id in the message, clear stale sessions on 401, and document ingress-nginx annotations so branded HTML 404/405 pages do not rewrite simulator bodies.
339 lines
13 KiB
Python
339 lines
13 KiB
Python
"""DB-backed per-path routes for Broadcom Automation API stubs without deep handlers.
|
|
|
|
Each ``(verb, path)`` from the universe registry is registered with
|
|
``APIRouter.add_api_route`` (same idea as Proxmox ``register_contract_routes``),
|
|
instead of a single ``/api/{full_path:path}`` catch-all.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, Request, Response
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from app.db.pool import Database
|
|
from app.dependencies import get_database
|
|
from app.vsphere import inventory
|
|
from app.vsphere.domain import api_state, tagging
|
|
from app.vsphere.domain import content as content_domain
|
|
from app.vsphere.rest.coverage import IMPLEMENTED
|
|
from app.vsphere.security.authz import require_read
|
|
from app.vsphere.security.session import SessionInfo
|
|
|
|
router = APIRouter(tags=["vSphere REST surface"])
|
|
|
|
_PARAM_RE = re.compile(r"\{([A-Za-z0-9_]+)\}")
|
|
|
|
|
|
def _extract_params(template: str, concrete: str) -> dict[str, str]:
|
|
pattern = "^" + _PARAM_RE.sub(r"([^/]+)", template) + "$"
|
|
match = re.match(pattern, concrete)
|
|
if not match:
|
|
return {}
|
|
names = _PARAM_RE.findall(template)
|
|
return {name: match.group(index + 1) for index, name in enumerate(names)}
|
|
|
|
|
|
async def _live_get(database: Database, template: str, concrete: str) -> Any | None:
|
|
"""Return inventory/platform-backed payloads when possible."""
|
|
|
|
params = _extract_params(template, concrete)
|
|
|
|
if template == "/api/cis/tagging/category":
|
|
return await tagging.list_categories(database)
|
|
if template == "/api/cis/tagging/tag":
|
|
return await tagging.list_tags(database)
|
|
if template == "/api/content/library":
|
|
return await content_domain.list_libraries(database)
|
|
if template == "/api/content/library/item":
|
|
libs = await content_domain.list_libraries(database)
|
|
items: list[dict[str, Any]] = []
|
|
for lib in libs:
|
|
items.extend(await content_domain.list_library_items(database, lib["id"]))
|
|
return items
|
|
if template == "/api/content/local-library":
|
|
libs = await content_domain.list_libraries(database)
|
|
return [lib for lib in libs if lib.get("type") == "LOCAL"]
|
|
|
|
vm = params.get("vm")
|
|
if vm and "/hardware/" in template:
|
|
obj = await inventory.require_object(database, vm, type_name="VirtualMachine", resource="vm")
|
|
props = obj.props or {}
|
|
|
|
def _live_list(key: str) -> list[Any] | None:
|
|
value = props.get(key)
|
|
if isinstance(value, list) and value:
|
|
return list(value)
|
|
return None
|
|
|
|
# Prefer non-empty inventory props; otherwise fall through to vsphere_api_state.
|
|
if template.endswith("/hardware/cdrom"):
|
|
return _live_list("cdroms")
|
|
if template.endswith("/hardware/floppy"):
|
|
return _live_list("floppies")
|
|
if template.endswith("/hardware/serial"):
|
|
return _live_list("serials")
|
|
if template.endswith("/hardware/parallel"):
|
|
return _live_list("parallels")
|
|
if template.endswith("/hardware/adapter/scsi"):
|
|
return _live_list("scsi_adapters")
|
|
if template.endswith("/hardware/adapter/sata"):
|
|
return _live_list("sata_adapters")
|
|
if template.endswith("/hardware/adapter/nvme"):
|
|
return _live_list("nvme_adapters")
|
|
if template.endswith("/hardware/boot") and isinstance(props.get("boot"), dict):
|
|
return props["boot"]
|
|
if template.endswith("/hardware/boot/device"):
|
|
return _live_list("boot_devices")
|
|
if template.endswith("/hardware/disk"):
|
|
return _live_list("disks") or list(props.get("disks") or [])
|
|
if template.endswith("/hardware/ethernet"):
|
|
return _live_list("nics") or list(props.get("nics") or [])
|
|
if "/hardware/disk/" in template and template.endswith("}"):
|
|
disk_id = params.get("disk")
|
|
for disk in props.get("disks") or []:
|
|
if str(disk.get("key") or disk.get("disk")) == str(disk_id):
|
|
return disk
|
|
if "/hardware/ethernet/" in template and template.endswith("}"):
|
|
nic_id = params.get("nic")
|
|
for nic in props.get("nics") or []:
|
|
if str(nic.get("key") or nic.get("nic")) == str(nic_id):
|
|
return nic
|
|
|
|
if vm and "/guest/" in template:
|
|
obj = await inventory.require_object(database, vm, type_name="VirtualMachine", resource="vm")
|
|
props = obj.props or {}
|
|
if template.endswith("/guest/local-filesystem"):
|
|
if "guest_filesystems" in props:
|
|
return props["guest_filesystems"]
|
|
return None
|
|
identity = props.get("identity") or {}
|
|
if identity or props.get("guest_OS") or props.get("guest_ip"):
|
|
return {
|
|
"name": identity.get("name") or obj.name,
|
|
"family": "LINUX"
|
|
if "WIN" not in str(props.get("guest_OS", "")).upper()
|
|
else "WINDOWS",
|
|
"full_name": {"name": props.get("guest_OS") or obj.name},
|
|
"host_name": identity.get("name") or obj.name,
|
|
"ip_address": props.get("guest_ip"),
|
|
}
|
|
return None
|
|
|
|
host = params.get("host")
|
|
if host and template.endswith("/networking"):
|
|
obj = await inventory.require_object(database, host, type_name="HostSystem", resource="host")
|
|
networking = (obj.props or {}).get("networking")
|
|
return networking if networking is not None else None
|
|
if host and "storage-device" in template:
|
|
obj = await inventory.require_object(database, host, type_name="HostSystem", resource="host")
|
|
devices = (obj.props or {}).get("storage_devices")
|
|
return devices if devices is not None else None
|
|
|
|
return None
|
|
|
|
|
|
async def _mutate_vm_hardware(
|
|
database: Database,
|
|
template: str,
|
|
concrete: str,
|
|
verb: str,
|
|
body: dict[str, Any],
|
|
) -> Any | None:
|
|
params = _extract_params(template, concrete)
|
|
vm = params.get("vm")
|
|
if not vm or "/hardware/" not in template:
|
|
return False
|
|
obj = await inventory.require_object(database, vm, type_name="VirtualMachine", resource="vm")
|
|
props = dict(obj.props or {})
|
|
|
|
if verb == "POST" and template.endswith("/hardware/cdrom"):
|
|
items = list(props.get("cdroms") or [])
|
|
key = str(3000 + len(items))
|
|
items.append(
|
|
{"cdrom": key, "label": f"CD/DVD drive {len(items) + 1}", "state": "CONNECTED", **body}
|
|
)
|
|
props["cdroms"] = items
|
|
await inventory.upsert_object(
|
|
database,
|
|
moid=vm,
|
|
type_name=obj.type,
|
|
name=obj.name,
|
|
parent_moid=obj.parent_moid,
|
|
props=props,
|
|
)
|
|
return key
|
|
if verb in {"PUT", "PATCH"} and template.endswith("/hardware/boot"):
|
|
props["boot"] = {**(props.get("boot") or {}), **body}
|
|
await inventory.upsert_object(
|
|
database,
|
|
moid=vm,
|
|
type_name=obj.type,
|
|
name=obj.name,
|
|
parent_moid=obj.parent_moid,
|
|
props=props,
|
|
)
|
|
return None
|
|
if verb == "DELETE" and "/hardware/cdrom/" in template:
|
|
cdrom = params.get("cdrom")
|
|
props["cdroms"] = [
|
|
c for c in (props.get("cdroms") or []) if str(c.get("cdrom")) != str(cdrom)
|
|
]
|
|
await inventory.upsert_object(
|
|
database,
|
|
moid=vm,
|
|
type_name=obj.type,
|
|
name=obj.name,
|
|
parent_moid=obj.parent_moid,
|
|
props=props,
|
|
)
|
|
return None
|
|
return False # not handled specialized; fall through to api_state
|
|
|
|
|
|
async def _dispatch(
|
|
template: str,
|
|
verb: str,
|
|
request: Request,
|
|
database: Database,
|
|
) -> Response:
|
|
concrete = request.url.path
|
|
method = verb.upper()
|
|
raw_body: Any = {}
|
|
if method in {"POST", "PUT", "PATCH"}:
|
|
try:
|
|
raw_body = await request.json()
|
|
except Exception:
|
|
raw_body = {}
|
|
if not isinstance(raw_body, dict):
|
|
raw_body = {"value": raw_body}
|
|
|
|
if method == "GET":
|
|
live = await _live_get(database, template, concrete)
|
|
if live is not None and not api_state.is_empty_payload(live):
|
|
return JSONResponse(content=live, status_code=200)
|
|
stored = await api_state.get_payload_or_seed(database, "GET", template)
|
|
if stored is None or api_state.is_empty_payload(stored):
|
|
return JSONResponse(content={"path": template, "status": "NOT_SEEDED"}, status_code=404)
|
|
return JSONResponse(content=stored, status_code=200)
|
|
|
|
hw = await _mutate_vm_hardware(database, template, concrete, method, raw_body)
|
|
if hw is not False:
|
|
if isinstance(hw, str):
|
|
return JSONResponse(content=hw, status_code=201)
|
|
return Response(status_code=204)
|
|
|
|
if method in {"PUT", "PATCH"}:
|
|
existing = await api_state.get_payload(database, "GET", template)
|
|
if isinstance(existing, dict) and isinstance(raw_body, dict):
|
|
merged = {**existing, **raw_body}
|
|
elif isinstance(raw_body, dict) and raw_body:
|
|
merged = raw_body
|
|
elif existing is not None:
|
|
merged = existing
|
|
else:
|
|
merged = {}
|
|
await api_state.put_payload(database, "GET", template, merged)
|
|
return JSONResponse(content=merged, status_code=200)
|
|
|
|
if method == "DELETE":
|
|
# Soft-delete: restore seed_payload from DB so lab GETs never go empty/404.
|
|
await api_state.restore_seed_payload(database, "GET", template)
|
|
if template.endswith("}"):
|
|
parent = template.rsplit("/", 1)[0]
|
|
if parent:
|
|
await api_state.restore_seed_payload(database, "GET", parent)
|
|
return Response(status_code=204)
|
|
|
|
# POST create / action
|
|
action = request.query_params.get("action")
|
|
if action:
|
|
await api_state.put_payload(
|
|
database,
|
|
"GET",
|
|
template,
|
|
{
|
|
"last_action": action,
|
|
"accepted": True,
|
|
"path": template,
|
|
**({} if not raw_body else {"spec": raw_body}),
|
|
},
|
|
)
|
|
if action.endswith("Task") or "task" in action.lower():
|
|
from app.vsphere.domain import tasks as task_store
|
|
|
|
task_id = await task_store.create_task(
|
|
database,
|
|
description=f"{action} {template}",
|
|
service="com.vmware.vapi",
|
|
operation=action,
|
|
status="SUCCEEDED",
|
|
result={"path": template, "action": action},
|
|
)
|
|
return JSONResponse(content=task_id, status_code=200)
|
|
return JSONResponse(content={"status": "SUCCESS", "action": action}, status_code=200)
|
|
|
|
new_id = await api_state.new_id(template.rstrip("/").rsplit("/", 1)[-1].strip("{}") or "id")
|
|
collection_payload = await api_state.get_payload(database, "GET", template)
|
|
created = {"id": new_id, "name": raw_body.get("name") or new_id, **raw_body}
|
|
if isinstance(collection_payload, list):
|
|
collection_payload = [*collection_payload, created]
|
|
await api_state.put_payload(database, "GET", template, collection_payload)
|
|
else:
|
|
await api_state.put_payload(
|
|
database, "GET", f"{template}/{{{template.rsplit('/', 1)[-1]}}}", created
|
|
)
|
|
await api_state.put_payload(database, "GET", template, created)
|
|
return JSONResponse(content=new_id, status_code=201)
|
|
|
|
|
|
def _endpoint(
|
|
template: str,
|
|
verb: str,
|
|
) -> Callable[..., Awaitable[Response]]:
|
|
async def dispatch(
|
|
request: Request,
|
|
database: Database = Depends(get_database),
|
|
_: SessionInfo = Depends(require_read),
|
|
) -> Response:
|
|
return await _dispatch(template, verb, request, database)
|
|
|
|
dispatch.__name__ = f"vsphere_stub_{verb}_{template.replace('/', '_').strip('_')}"
|
|
dispatch.__qualname__ = dispatch.__name__
|
|
return dispatch
|
|
|
|
|
|
def register_stub_routes(
|
|
target: APIRouter | None = None,
|
|
*,
|
|
implemented: dict[tuple[str, str], str] | None = None,
|
|
) -> int:
|
|
"""Register one FastAPI route per stub ``(verb, path)`` from the coverage registry.
|
|
|
|
Deep ``CORE_IMPLEMENTED`` handlers are skipped so they keep winning on their
|
|
dedicated routers. Returns the number of routes added.
|
|
"""
|
|
|
|
api = target if target is not None else router
|
|
registry = implemented if implemented is not None else IMPLEMENTED
|
|
added = 0
|
|
for (verb, path), status in sorted(registry.items(), key=lambda item: (item[0][1], item[0][0])):
|
|
if status != "stub":
|
|
continue
|
|
api.add_api_route(
|
|
path,
|
|
_endpoint(path, verb),
|
|
methods=[verb],
|
|
name=f"vsphere-stub:{verb}:{path}",
|
|
include_in_schema=True,
|
|
openapi_extra={"x-vmware-implementation": "stub"},
|
|
)
|
|
added += 1
|
|
return added
|
|
|
|
|
|
register_stub_routes()
|