fix(standalone): API-driven mode does not properly handle consumer schema

apisix/admin/standalone.lua

@@ -156,7 +156,12 @@ local function update(ctx)
                     end
                 end
                 if item_checker then
-                    valid, err = item_checker(item_temp)
+                    local item_checker_key
+                    if item.id then
+                        -- credential need to check key
+                        item_checker_key = "/" .. key .. "/" .. item_temp.id
+                    end
+                    valid, err = item_checker(item_temp, item_checker_key)
                     if not valid then
                         core.log.error(err_prefix, err)
                         core.response.exit(400, {error_msg = err_prefix .. err})
@@ -235,6 +240,56 @@ function _M.run()
 end
 
 
+local patch_schema
+do
+    local resource_schema = {
+        "proto",
+        "global_rule",
+        "route",
+        "service",
+        "upstream",
+        "consumer",
+        "consumer_group",
+        "credential",
+        "ssl",
+        "plugin_config",
+    }
+    local function attach_modifiedIndex_schema(name)
+        local schema = core.schema[name]
+        if not schema then
+            core.log.error("schema for ", name, " not found")
+            return
+        end
+        if schema.properties and not schema.properties.modifiedIndex then
+            schema.properties.modifiedIndex = {
+                type = "integer",
+            }
+        end
+    end
+
+    local function patch_credential_schema()
+        local credential_schema = core.schema["credential"]
+        if credential_schema and credential_schema.properties then
+            credential_schema.properties.id = {
+                type = "string",
+                minLength = 15,
+                maxLength = 128,
+                pattern = [[^[a-zA-Z0-9]+/credentials/[a-zA-Z0-9]+$]],
+            }
+        end
+    end
+
+    function patch_schema()
+        -- attach modifiedIndex schema to all resource schemas
+        for _, name in ipairs(resource_schema) do
+            attach_modifiedIndex_schema(name)
+        end
+        -- patch credential schema
+        patch_credential_schema()
+    end
+end
+
+
 function _M.init_worker()
     local function update_config()
         local config, err = shared_dict:get("config")
@@ -251,6 +306,8 @@ function _M.init_worker()
         config_yaml._update_config(config)
     end
     events:register(update_config, EVENT_UPDATE, EVENT_UPDATE)
+
+    patch_schema()
 end
 
 

apisix/consumer.lua

@@ -44,7 +44,10 @@ local consumers_count_for_lrucache = 4096
 local function remove_etcd_prefix(key)
     local prefix = ""
     local local_conf = config_local.local_conf()
-    if local_conf.etcd and local_conf.etcd.prefix then
+    local role = core.table.try_read_attr(local_conf, "deployment", "role")
+    local provider = core.table.try_read_attr(local_conf, "deployment", "role_" ..
+    role, "config_provider")
+    if provider == "etcd" and local_conf.etcd and local_conf.etcd.prefix then
         prefix = local_conf.etcd.prefix
     end
     return string_sub(key, #prefix + 1)
@@ -285,25 +288,12 @@ local function check_consumer(consumer, key)
 end
 
 
-local function filter(consumer)
-    if not consumer.value then
-        return
-    end
-
-    -- We expect the id is the same as username. Fix up it here if it isn't.
-    consumer.value.id = consumer.value.username
-end
-
-
 function _M.init_worker()
     local err
     local cfg = {
         automatic = true,
         checker = check_consumer,
     }
-    if core.config.type ~= "etcd" then
-        cfg.filter = filter
-    end
 
     consumers, err = core.config.new("/consumers", cfg)
     if not consumers then

apisix/core/config_etcd.lua

@@ -574,6 +574,8 @@ local function load_full_data(self, dir_res, headers)
             end
 
             if data_valid and self.checker then
+                -- TODO: An opts table should be used
+                -- as different checkers may use different parameters
                 data_valid, err = self.checker(item.value, item.key)
                 if not data_valid then
                     log.error("failed to check item data of [", self.key,

apisix/core/config_yaml.lua

@@ -25,7 +25,6 @@ local yaml         = require("lyaml")
 local log          = require("apisix.core.log")
 local json         = require("apisix.core.json")
 local new_tab      = require("table.new")
-local tbl_deepcopy = require("apisix.core.table").deepcopy
 local check_schema = require("apisix.core.schema").check
 local profile      = require("apisix.core.profile")
 local lfs          = require("lfs")
@@ -239,7 +238,9 @@ local function sync_data(self)
             end
 
             if data_valid and self.checker then
-                data_valid, err = self.checker(item)
+                -- TODO: An opts table should be used
+                -- as different checkers may use different parameters
+                data_valid, err = self.checker(item, conf_item.key)
                 if not data_valid then
                     log.error("failed to check item data of [", self.key,
                               "] err:", err, " ,val: ", json.delay_encode(item))
@@ -274,7 +275,7 @@ local function sync_data(self)
                           ", it should be an object")
             end
 
-            local id = item.id or ("arr_" .. idx)
+            local id = item.id or item.username or ("arr_" .. idx)
             local modifiedIndex = item.modifiedIndex or conf_version
             local conf_item = {value = item, modifiedIndex = modifiedIndex,
                             key = "/" .. self.key .. "/" .. id}
@@ -288,7 +289,7 @@ local function sync_data(self)
             end
 
             if data_valid and self.checker then
-                data_valid, err = self.checker(item)
+                data_valid, err = self.checker(item, conf_item.key)
                 if not data_valid then
                     log.error("failed to check item data of [", self.key,
                               "] err:", err, " ,val: ", json.delay_encode(item))
@@ -447,16 +448,6 @@ function _M.new(key, opts)
         key = sub_str(key, 2)
     end
 
-    if is_use_admin_api() then
-        if item_schema and item_schema.properties then
-            local item_schema_cp = tbl_deepcopy(item_schema)
-            -- allow clients to specify modifiedIndex to control resource changes.
-            item_schema_cp.properties.modifiedIndex = {
-                type = "integer",
-            }
-            item_schema = item_schema_cp
-        end
-    end
     local obj = setmetatable({
         automatic = automatic,
         item_schema = item_schema,

apisix/stream/router/ip_port.lua

@@ -233,7 +233,9 @@ function _M.stream_init_worker(filter)
     user_routes, err = core.config.new("/stream_routes", {
             automatic = true,
             item_schema = core.schema.stream_route,
-            checker = stream_route_checker,
+            checker = function(item)
+                return stream_route_checker(item)
+            end,
             filter = filter,
         })
 

t/admin/standalone.spec.ts

@@ -19,6 +19,12 @@ import axios from "axios";
 import YAML from "yaml";
 
 const ENDPOINT = "/apisix/admin/configs";
+const clientConfig = {
+  baseURL: "http://localhost:1984",
+  headers: {
+    "X-API-KEY": "edd1c9f034335f136f87ad84b625c8f1",
+  },
+};
 const config1 = {
   routes: [
     {
@@ -65,12 +71,60 @@ const routeWithModifiedIndex = {
     },
   ],
 };
-const clientConfig = {
-  baseURL: "http://localhost:1984",
-  headers: {
-    "X-API-KEY": "edd1c9f034335f136f87ad84b625c8f1",
-  },
-};
+const routeWithKeyAuth = {
+  routes: [
+    {
+      id: "r1",
+      uri: "/r1",
+      upstream: {
+        nodes: { "127.0.0.1:1980": 1 },
+        type: "roundrobin",
+      },
+      plugins: {
+        "proxy-rewrite": { uri: "/hello" },
+        "key-auth": {},
+      },
+    },
+  ]
+}
+const consumerWithModifiedIndex = {
+  routes: routeWithKeyAuth.routes,
+  consumers: [
+    {
+      modifiedIndex: 10,
+      username: "jack",
+      plugins: {
+        "key-auth": {
+          key: "jack-key",
+        }
+      },
+    },
+  ],
+}
+const credential1 = {
+  routes: routeWithKeyAuth.routes,
+  consumers: [
+    {
+      "username": "john"
+    },
+    {
+      "id": "john/credentials/a",
+      "plugins": {
+        "key-auth": {
+          "key": "auth-a"
+        }
+      }
+    },
+    {
+      "id": "john/credentials/b",
+      "plugins": {
+        "key-auth": {
+          "key": "auth-b"
+        }
+      }
+    }
+  ]
+}
 
 describe("Admin - Standalone", () => {
   const client = axios.create(clientConfig);
@@ -192,14 +246,15 @@ describe("Admin - Standalone", () => {
     it('check route "r2"', () =>
       expect(client.get("/r2")).rejects.toThrow(
         "Request failed with status code 404"
-    ));
+      ));
 
     it("only set routes_conf_version", async () => {
       const resp = await client.put(
         ENDPOINT,
         YAML.stringify({ routes_conf_version: 15 }),
-        {headers: {"Content-Type": "application/yaml"},
-      });
+        {
+          headers: { "Content-Type": "application/yaml" },
+        });
       expect(resp.status).toEqual(202);
 
       const resp_1 = await client.get(ENDPOINT);
@@ -213,8 +268,9 @@ describe("Admin - Standalone", () => {
       const resp2 = await client.put(
         ENDPOINT,
         YAML.stringify({ routes_conf_version: 17 }),
-        {headers: {"Content-Type": "application/yaml"},
-      });
+        {
+          headers: { "Content-Type": "application/yaml" },
+        });
       expect(resp2.status).toEqual(202);
 
       const resp2_1 = await client.get(ENDPOINT);
@@ -269,6 +325,48 @@ describe("Admin - Standalone", () => {
       const resp3_2 = await client.get("/r2");
       expect(resp3_2.status).toEqual(200);
     });
+
+    it("apply consumer with modifiedIndex", async () => {
+      const resp = await client.put(ENDPOINT, consumerWithModifiedIndex);
+      expect(resp.status).toEqual(202);
+
+      const resp_1 = await client.get("/r1", { headers: { "apikey": "invalid-key" } }).catch((err) => err.response);
+      expect(resp_1.status).toEqual(401);
+      const resp_2 = await client.get("/r1", { headers: { "apikey": "jack-key" } });
+      expect(resp_2.status).toEqual(200);
+
+      const updatedConsumer = structuredClone(consumerWithModifiedIndex);
+
+      // update key of key-auth plugin, but modifiedIndex is not changed
+      updatedConsumer.consumers[0].plugins["key-auth"] = { "key": "jack-key-updated" };
+      const resp2 = await client.put(ENDPOINT, updatedConsumer);
+      expect(resp2.status).toEqual(202);
+
+      const resp2_1 = await client.get("/r1", { headers: { "apikey": "jack-key-updated" } }).catch((err) => err.response);
+      expect(resp2_1.status).toEqual(401);
+      const resp2_2 = await client.get("/r1", { headers: { "apikey": "jack-key" } });
+      expect(resp2_2.status).toEqual(200);
+
+      // update key of key-auth plugin, and modifiedIndex is changed
+      updatedConsumer.consumers[0].modifiedIndex++;
+      const resp3 = await client.put(ENDPOINT, updatedConsumer);
+      const resp3_1 = await client.get("/r1", { headers: { "apikey": "jack-key-updated" } });
+      expect(resp3_1.status).toEqual(200);
+      const resp3_2 = await client.get("/r1", { headers: { "apikey": "jack-key" } }).catch((err) => err.response);
+      expect(resp3_2.status).toEqual(401);
+    });
+
+    it("apply consumer with credentials", async () => {
+      const resp = await client.put(ENDPOINT, credential1);
+      expect(resp.status).toEqual(202);
+
+      const resp_1 = await client.get("/r1", { headers: { "apikey": "auth-a" } });
+      expect(resp_1.status).toEqual(200);
+      const resp_2 = await client.get("/r1", { headers: { "apikey": "auth-b" } });
+      expect(resp_2.status).toEqual(200);
+      const resp_3 = await client.get("/r1", { headers: { "apikey": "invalid-key" } }).catch((err) => err.response);
+      expect(resp_3.status).toEqual(401);
+    });
   });
 
   describe("Exceptions", () => {
@@ -279,18 +377,19 @@ describe("Admin - Standalone", () => {
 
     it("update config (lower conf_version)", async () => {
       const resp = await clientException.put(
+        ENDPOINT,
+        { routes_conf_version: 100 },
+        { headers: { "Content-Type": "application/yaml" } }
+      );
+      const resp2 = await clientException.put(
         ENDPOINT,
         YAML.stringify(invalidConfVersionConfig1),
-        {
-          headers: {
-            "Content-Type": "application/yaml",
-          },
-        }
+        { headers: { "Content-Type": "application/yaml" } }
       );
-      expect(resp.status).toEqual(400);
-      expect(resp.data).toEqual({
+      expect(resp2.status).toEqual(400);
+      expect(resp2.data).toEqual({
         error_msg:
-          "routes_conf_version must be greater than or equal to (20)",
+          "routes_conf_version must be greater than or equal to (100)",
       });
     });
 

t/cli/test_status_api.sh

@@ -49,6 +49,8 @@ docker compose -f ./t/cli/docker-compose-etcd-cluster.yaml up -d
 
 make run
 
+sleep 0.5
+
 curl -s -o /dev/null -w "%{http_code}" http://127.0.0.1:7085/status | grep 200 \
 || (echo "failed: status api didn't return 200"; exit 1)