From 5346bc53e449ffef3ca3355f6546914fbf49b12b Mon Sep 17 00:00:00 2001
From: MuchoRio <riogriyaputra@gmail.com>
Date: Sat, 10 Oct 2026 19:22:52 +0700
Subject: [PATCH] fix(api_spec): extract path-level and operation-level servers
 for scope authorization

- Inspect path-level, operation-level, and webhook-level server declarations in OpenAPI specifications.
- Ensure APIs with microservice routing or without root servers authorize all declared base URLs as in-scope targets.
- Add regression tests for nested path/operation servers, root-less specs, and webhook servers.
---
 strix/utils/api_spec.py | 34 ++++++++++++++++++++----
 tests/test_api_spec.py  | 59 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 88 insertions(+), 5 deletions(-)

diff --git a/strix/utils/api_spec.py b/strix/utils/api_spec.py
index db3a7738..87b0569d 100644
--- a/strix/utils/api_spec.py
+++ b/strix/utils/api_spec.py
@@ -123,15 +123,39 @@ 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)
 
 
+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))
+    for op in path_item.values():
+        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_item in section.values():
+                _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")
         ],
     )
 
diff --git a/tests/test_api_spec.py b/tests/test_api_spec.py
index d7979d35..49665d6a 100644
--- a/tests/test_api_spec.py
+++ b/tests/test_api_spec.py
@@ -149,6 +149,65 @@ 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_swagger_base_urls_built_from_host() -> None:
     assert spec_base_urls(SWAGGER_JSON) == ["https://legacy.test/api"]