Merge c73b018f1d into 87f2d783d6

strix/utils/api_spec.py

@@ -123,15 +123,49 @@ def _resolve_server_url(url: str, variables: Any) -> str:
     return _SERVER_VAR_PATTERN.sub(lambda m: defaults.get(m.group(1), m.group(0)), url)
 
 
+_OPENAPI_OPERATION_METHODS = frozenset(
+    {"get", "put", "post", "delete", "options", "head", "patch", "trace"}
+)
+
+
+def _collect_path_item_servers(path_item: Any, server_objs: list[dict[str, Any]]) -> None:
+    if not isinstance(path_item, dict):
+        return
+    path_servers = path_item.get("servers")
+    if isinstance(path_servers, list):
+        server_objs.extend(s for s in path_servers if isinstance(s, dict))
+    # Inspect only standard HTTP method operations; skip metadata, parameters,
+    # and vendor extensions (x-*).
+    for method in _OPENAPI_OPERATION_METHODS:
+        op = path_item.get(method)
+        if isinstance(op, dict):
+            op_servers = op.get("servers")
+            if isinstance(op_servers, list):
+                server_objs.extend(s for s in op_servers if isinstance(s, dict))
+
+
 def _openapi_base_urls(raw: dict[str, Any]) -> list[str]:
-    servers = raw.get("servers")
-    if not isinstance(servers, list):
-        return []
+    server_objs: list[dict[str, Any]] = []
+
+    # 1. Root-level servers
+    root_servers = raw.get("servers")
+    if isinstance(root_servers, list):
+        server_objs.extend(s for s in root_servers if isinstance(s, dict))
+
+    # 2. PathItem- and Operation-level servers (OpenAPI 3.0 paths / 3.1 webhooks)
+    for section_key in ("paths", "webhooks"):
+        section = raw.get(section_key)
+        if isinstance(section, dict):
+            for path_key, path_item in section.items():
+                if str(path_key).startswith("x-"):
+                    continue
+                _collect_path_item_servers(path_item, server_objs)
+
     return _absolute_urls(
         [
             _resolve_server_url(str(server["url"]), server.get("variables"))
-            for server in servers
-            if isinstance(server, dict) and server.get("url")
+            for server in server_objs
+            if server.get("url")
         ],
     )
 

tests/test_api_spec.py

@@ -149,6 +149,92 @@ def test_openapi_drops_unresolved_relative_server() -> None:
     assert spec_base_urls(raw) == []
 
 
+def test_openapi_extracts_path_and_operation_servers() -> None:
+    raw = {
+        "openapi": "3.0.0",
+        "info": {"title": "Microservices"},
+        "servers": [{"url": "https://gateway.test/api"}],
+        "paths": {
+            "/users": {
+                "servers": [{"url": "https://users.service.test"}],
+                "get": {"summary": "List users"},
+            },
+            "/orders": {
+                "post": {
+                    "servers": [{"url": "https://orders.service.test/v1"}],
+                    "summary": "Create order",
+                },
+            },
+            "/relative": {
+                "servers": [{"url": "/relative-path"}],
+                "get": {"summary": "Ignored relative server"},
+            },
+        },
+    }
+    assert spec_base_urls(raw) == [
+        "https://gateway.test/api",
+        "https://users.service.test",
+        "https://orders.service.test/v1",
+    ]
+
+
+def test_openapi_without_root_servers_extracts_nested_servers() -> None:
+    raw = {
+        "openapi": "3.0.0",
+        "info": {"title": "No Root Servers"},
+        "paths": {
+            "/auth": {
+                "servers": [{"url": "https://auth.service.test"}],
+                "post": {"summary": "Login"},
+            },
+        },
+    }
+    assert spec_base_urls(raw) == ["https://auth.service.test"]
+
+
+def test_openapi_extracts_webhook_servers() -> None:
+    raw = {
+        "openapi": "3.1.0",
+        "info": {"title": "Webhooks API"},
+        "webhooks": {
+            "newPet": {
+                "post": {
+                    "servers": [{"url": "https://webhook-receiver.test"}],
+                    "summary": "Webhook notification",
+                },
+            },
+        },
+    }
+    assert spec_base_urls(raw) == ["https://webhook-receiver.test"]
+
+
+def test_openapi_ignores_vendor_extensions_in_paths_and_operations() -> None:
+    raw = {
+        "openapi": "3.0.0",
+        "info": {"title": "Extensions Test"},
+        "paths": {
+            "x-internal-path": {
+                "servers": [{"url": "https://leaked-internal.test"}],
+                "get": {"servers": [{"url": "https://leaked-internal-op.test"}]},
+            },
+            "/api/valid": {
+                "servers": [{"url": "https://valid.test"}],
+                "get": {
+                    "servers": [{"url": "https://valid-get.test"}],
+                    "summary": "Valid GET operation",
+                },
+                "x-deployment": {
+                    "servers": [{"url": "https://admin-extension.test"}],
+                },
+                "parameters": [
+                    {"name": "id", "in": "query", "servers": [{"url": "https://param.test"}]}
+                ],
+            },
+        },
+    }
+    assert spec_base_urls(raw) == ["https://valid.test", "https://valid-get.test"]
+
+
 def test_swagger_base_urls_built_from_host() -> None:
     assert spec_base_urls(SWAGGER_JSON) == ["https://legacy.test/api"]