diff --git a/app.py b/app.py index 5a99cdc..aa94ed4 100644 --- a/app.py +++ b/app.py @@ -264,6 +264,7 @@ async def root(): "openapi": "/openapi.json", "services": svc_list, "endpoints": { + "capabilities": "GET /capabilities - live machine-readable operation and safety map", "proxy": "GET/POST/PUT/DELETE /{service}/{path} - proxy to backend with auto-auth", "vm_list": "GET /vm/list", "vm_create": "POST /vm/create {node, ip, hostname, cores?, memory?, disk?}", @@ -288,6 +289,80 @@ async def root(): async def health(): return {"status": "ok", "vault_items": len(_vault_cache), "services": len(SERVICES), "version": VERSION} + +def _schema_contains_property(node, property_name: str, components: dict, seen: set[str] | None = None) -> bool: + """Resolve local OpenAPI refs and look for a request property.""" + seen = seen or set() + if isinstance(node, list): + return any(_schema_contains_property(item, property_name, components, seen) for item in node) + if not isinstance(node, dict): + return False + if node.get("name") == property_name or property_name in node.get("properties", {}): + return True + ref = node.get("$ref", "") + if ref.startswith("#/components/schemas/"): + name = ref.rsplit("/", 1)[-1] + if name in seen: + return False + return _schema_contains_property(components.get(name, {}), property_name, components, seen | {name}) + return any( + _schema_contains_property(value, property_name, components, seen) + for key, value in node.items() + if key != "properties" + ) + + +@app.get("/capabilities") +async def capabilities(_=Depends(_verify)): + """Live operation catalog with safety metadata for AI agents.""" + schema = app.openapi() + components = schema.get("components", {}).get("schemas", {}) + operations = [] + for path, methods in schema.get("paths", {}).items(): + if path == "/{service}/{path}" or path in {"/", "/health", "/openapi.json", "/docs", "/redoc"}: + continue + for method, operation in methods.items(): + if method.upper() not in {"GET", "POST", "PUT", "PATCH", "DELETE"}: + continue + mode = "read_only" if method.upper() == "GET" else "mutation" + serialized = {"parameters": operation.get("parameters", []), "requestBody": operation.get("requestBody", {})} + dry_run = _schema_contains_property(serialized, "dry_run", components) + critical = path.startswith(("/network/wireguard", "/caddy/")) + destructive = method.upper() == "DELETE" or any( + marker in path for marker in ("/destroy/", "/cleanup/", "/break-lock/", "/restore/") + ) + operations.append({ + "method": method.upper(), + "path": path, + "summary": operation.get("summary", ""), + "description": operation.get("description", ""), + "mode": mode, + "dry_run": dry_run, + "critical": critical, + "destructive": destructive, + "confirmation_required": mode == "mutation", + }) + operations.sort(key=lambda item: (item["path"], item["method"])) + counts = { + "total": len(operations), + "read_only": sum(item["mode"] == "read_only" for item in operations), + "mutations": sum(item["mode"] == "mutation" for item in operations), + "destructive": sum(item["destructive"] for item in operations), + } + return { + "schema_version": 1, + "service": "homelab-butler", + "version": VERSION, + "generated": datetime.now(timezone.utc).isoformat(), + "counts": counts, + "operations": operations, + "proxy": {"path": "/{service}/{path}", "note": "Generic backend proxy; inspect /info services and OpenAPI before use"}, + "model_contract": { + "instruction": "Prefer read_only operations. Before every mutation inspect its schema, use dry_run when available, and obtain confirmation for critical or destructive actions.", + "source_of_truth": "/openapi.json", + }, + } + def _classify_http_status(status_code: int, expected: set[int]) -> str: """Return a deterministic service state suitable for small models.""" if status_code in expected: @@ -408,6 +483,7 @@ async def info(_=Depends(_verify)): for name, cfg in SERVICES.items() }, "endpoints": { + "capabilities": "/capabilities", "status": "/status", "overview": "/overview?details=false", "audit": "/audit", diff --git a/tests/test_app.py b/tests/test_app.py index 726d11d..a9a0ae1 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -46,6 +46,33 @@ def test_health_exposes_current_version(): assert response.json()["version"] == app.VERSION == "2.3.5" +def test_capabilities_is_live_machine_readable_safety_map(): + with TestClient(app.app) as client: + response = client.get( + "/capabilities", + headers={"Authorization": "Bearer test-token"}, + ) + assert response.status_code == 200 + payload = response.json() + by_operation = {(item["method"], item["path"]): item for item in payload["operations"]} + assert ("GET", "/network/wireguard/{host}") in by_operation + assert by_operation[("GET", "/network/wireguard/{host}")]["mode"] == "read_only" + removal = by_operation[("DELETE", "/network/wireguard/{host}/peer")] + assert removal["mode"] == "mutation" + assert removal["dry_run"] is True + assert removal["critical"] is True + assert by_operation[("DELETE", "/vm/destroy/{vmid}")]["dry_run"] is True + assert all(item["path"] != "/{service}/{path}" for item in payload["operations"]) + assert payload["model_contract"]["instruction"].startswith("Prefer read_only") + + +def test_info_advertises_capabilities_endpoint(): + with TestClient(app.app) as client: + response = client.get("/info", headers={"Authorization": "Bearer test-token"}) + assert response.status_code == 200 + assert response.json()["endpoints"]["capabilities"] == "/capabilities" + + def test_wireguard_status_returns_redacted_live_state(monkeypatch): payload = { "interface": "wg0",