feat(discovery): check node format (#4610)

To prevent 500 error or infinite loop because of a malformed nodes
format.
Signed-off-by: spacewander <[email protected]>

apisix/discovery/consul_kv.lua

@@ -458,11 +458,12 @@ function _M.init_worker()
     end
 
     log.notice("consul_conf: ", core.json.encode(consul_conf))
+    default_weight = consul_conf.weight
     -- set default service, used when the server node cannot be found
     if consul_conf.default_service then
         default_service = consul_conf.default_service
+        default_service.weight = default_weight
     end
-    default_weight = consul_conf.weight
     default_prefix_rule = "(" .. consul_conf.prefix .. "/.*/)([a-zA-Z0-9.]+):([0-9]+)"
     log.info("default params, default_weight: ", default_weight,
             ", default_prefix_rule: ", default_prefix_rule)

apisix/schema_def.lua

@@ -331,6 +331,37 @@ local nodes_schema = {
         }
     }
 }
+_M.discovery_nodes = {
+    type = "array",
+    items = {
+        type = "object",
+        properties = {
+            host = {
+                description = "domain or ip",
+            },
+            port = {
+                description = "port of node",
+                type = "integer",
+                minimum = 1,
+            },
+            weight = {
+                description = "weight of node",
+                type = "integer",
+                minimum = 0,
+            },
+            priority = {
+                description = "priority of node",
+                type = "integer",
+            },
+            metadata = {
+                description = "metadata of node",
+                type = "object",
+            }
+        },
+        -- nodes from DNS discovery may not contain port
+        required = {"host", "weight"},
+    },
+}
 
 
 local certificate_scheme = {

apisix/upstream.lua

@@ -235,7 +235,8 @@ function _M.set_by_route(route, api_ctx)
 
         local dis = discovery[up_conf.discovery_type]
         if not dis then
-            return 500, "discovery " .. up_conf.discovery_type .. " is uninitialized"
+            local err = "discovery " .. up_conf.discovery_type .. " is uninitialized"
+            return 500, err
         end
 
         local new_nodes, err = dis.nodes(up_conf.service_name)
@@ -245,6 +246,11 @@ function _M.set_by_route(route, api_ctx)
 
         local same = upstream_util.compare_upstream_node(up_conf, new_nodes)
         if not same then
+            local pass, err = core.schema.check(core.schema.discovery_nodes, new_nodes)
+            if not pass then
+                return HTTP_CODE_UPSTREAM_UNAVAILABLE, "invalid nodes format: " .. err
+            end
+
             up_conf.nodes = new_nodes
             local new_up_conf = core.table.clone(up_conf)
             core.log.info("discover new upstream from ", up_conf.service_name, ", type ",

t/node/upstream-discovery.t

@@ -331,3 +331,35 @@ qr/proxy request to \S+/
 --- grep_error_log_out
 proxy request to 127.0.0.1:1979
 proxy request to 0.0.0.0:1980
+
+
+
+=== TEST 7: bad nodes return by the discovery
+--- apisix_yaml
+routes:
+  -
+    uris:
+        - /hello
+    upstream_id: 1
+--- config
+    location /t {
+        content_by_lua_block {
+            local discovery = require("apisix.discovery.init").discovery
+            discovery.mock = {
+                nodes = function()
+                    return {
+                        {host = "127.0.0.1", port = 1980, weight = "0"},
+                    }
+                end
+            }
+            local http = require "resty.http"
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+            local httpc = http.new()
+            local res, err = httpc:request_uri(uri, {method = "GET", keepalive = false})
+            ngx.say(res.status)
+        }
+    }
+--- response_body
+503
+--- error_log
+invalid nodes format: failed to validate item 1: property "weight" validation failed: wrong type: expected integer, got string