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.
This commit is contained in:
MuchoRio 2026-10-10 19:22:52 +07:00
parent 62b496430d
commit 5346bc53e4
2 changed files with 88 additions and 5 deletions

View file

@ -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")
],
)

View file

@ -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"]