vmware-api-simulator/scripts/vsphere_surface_probe.py

#!/usr/bin/env python3
"""Probe every implemented REST path from the coverage registry."""

from __future__ import annotations

import json
import os
import ssl
import sys
import urllib.error
import urllib.request
from base64 import b64encode

from app.vsphere.rest.coverage import catalog_entries

BASE = os.getenv("VSPHERE_BASE", "https://localhost")
USER = "administrator@vsphere.local"
PASSWORD = "VMware1!"


def _ctx() -> ssl.SSLContext | None:
    if not BASE.startswith("https://"):
        return None
    return ssl._create_unverified_context()  # noqa: S323


def _concrete(path: str) -> str:
    import re

    subs = {
        "{vm}": "vm-101",
        "{host}": "host-11",
        "{datastore}": "datastore-31",
        "{task}": "task-missing",
        "{snapshot}": "snapshot-missing",
        "{category_id}": "missing",
        "{tag_id}": "missing",
        "{item_id}": "missing",
        "{library_id}": "lib-missing",
        "{folder}": "group-v23",
        "{datacenter}": "datacenter-21",
        "{cluster}": "domain-c21",
        # Disposable id — seed resgroup-22 is protected from DELETE.
        "{resource_pool}": "resgroup-missing",
        "{permission_id}": "1",
        "{policy}": "policy-default",
        "{disk}": "2000",
        "{nic}": "4000",
        "{cdrom}": "3000",
        "{adapter}": "1000",
        "{network}": "network-41",
    }
    out = path
    for key, value in subs.items():
        out = out.replace(key, value)
    return re.sub(r"\{([A-Za-z0-9_]+)\}", r"probe-\1", out)


def _request(method: str, path: str, *, headers: dict[str, str], data: bytes | None = None) -> int:
    concrete = _concrete(path)
    req = urllib.request.Request(f"{BASE}{concrete}", data=data, method=method, headers=headers)
    try:
        with urllib.request.urlopen(req, context=_ctx()) as resp:  # noqa: S310
            return int(resp.status)
    except urllib.error.HTTPError as error:
        return int(error.code)


def main() -> int:
    basic = b64encode(f"{USER}:{PASSWORD}".encode()).decode()
    status = _request("POST", "/api/session", headers={"Authorization": f"Basic {basic}"})
    if status not in {200, 201}:
        print(f"session failed: {status}", file=sys.stderr)
        return 1
    # Re-login to capture body
    req = urllib.request.Request(
        f"{BASE}/api/session",
        method="POST",
        headers={"Authorization": f"Basic {basic}"},
    )
    with urllib.request.urlopen(req, context=_ctx()) as resp:  # noqa: S310
        session = json.loads(resp.read().decode())
    headers = {"vmware-api-session-id": session, "Content-Type": "application/json"}
    failures: list[str] = []
    probed = 0
    for entry in catalog_entries():
        verb = entry["verb"]
        path = entry["path"]
        if verb == "DELETE" and path in {"/api/session", "/rest/com/vmware/cis/session"}:
            continue
        if (
            "{" in path
            and verb in {"POST", "PATCH", "DELETE"}
            and "missing" in (path.replace("{vm}", "vm-101"))
        ):
            # skip destructive ops on missing ids except GET
            pass
        data = b"{}" if verb in {"POST", "PUT", "PATCH"} else None
        if path.endswith("/power") and verb == "POST":
            code = _request(verb, path + "?action=start", headers=headers)
        elif "tag-association" in path and verb == "POST":
            data = json.dumps(
                {
                    "action": "list-attached-tags",
                    "tag_id": "x",
                    "object_id": {"type": "VirtualMachine", "id": "vm-101"},
                }
            ).encode()
            code = _request(verb, path, headers=headers, data=data)
        else:
            code = _request(verb, path, headers=headers, data=data)
        probed += 1
        # Accept success, not-found for missing substitutions, or validation errors.
        if code >= 500:
            failures.append(f"{verb} {path} -> {code}")
            continue
        if verb == "GET" and 200 <= code < 300:
            # Surface probe reads body via a second request-sized check only for markers.
            # Re-fetch is avoided: empty GET bodies for session are OK.
            pass
    # Re-auth in case any probe request invalidated the session cookie.
    req = urllib.request.Request(
        f"{BASE}/api/session",
        method="POST",
        headers={"Authorization": f"Basic {basic}"},
    )
    with urllib.request.urlopen(req, context=_ctx()) as resp:  # noqa: S310
        session = json.loads(resp.read().decode())
    headers = {"vmware-api-session-id": session, "Content-Type": "application/json"}

    # Spot-check critical inventory payloads are non-empty / non-stub.
    spot = [
        "/api/vcenter/vm",
        "/api/vcenter/host",
        "/api/content/library",
        "/api/cis/tagging/category",
        "/api/esx/settings/clusters/domain-c21/software",
        "/api/vcenter/namespace-management/supervisors/supervisor-1/summary",
        "/api/appliance/access/ssh",
        "/api/appliance/services",
        "/api/vcenter/vm/vm-101/hardware/cdrom",
    ]
    for path in spot:
        url = f"{BASE}{_concrete(path)}"
        req = urllib.request.Request(url, method="GET", headers=headers)
        try:
            with urllib.request.urlopen(req, context=_ctx()) as resp:  # noqa: S310
                body = resp.read().decode("utf-8", errors="replace")
                code = int(resp.status)
        except urllib.error.HTTPError as error:
            body = error.read().decode("utf-8", errors="replace")
            code = int(error.code)
        if code >= 400:
            failures.append(f"GET {path} spot -> {code}")
            continue
        if '"stub": true' in body or '"stub":true' in body:
            failures.append(f"GET {path} spot -> stub marker")
            continue
        try:
            parsed = json.loads(body)
        except json.JSONDecodeError:
            failures.append(f"GET {path} spot -> non-json")
            continue
        if parsed in ([], {}, None):
            failures.append(f"GET {path} spot -> empty")
    print(json.dumps({"probed": probed, "failures": failures}, indent=2))
    return 1 if failures else 0


if __name__ == "__main__":
    raise SystemExit(main())