crud: add execution rights to a VShard user

The patch adds execution access for storage crud functions to a
VShard storage user in the VShard manner [1].

1. https://github.com/tarantool/vshard/blob/b3c27b32637863e9a03503e641bb7c8c69779a00/vshard/storage/init.lua#L777-L780

Closes #366

CHANGELOG.md

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 * `crud.readview` resource cleanup on garbage collect (#379).
 * `doc/playground.lua` does not work with Tarantool 3 (#371).
 * Tests with Tarantool 3 (#364).
+* VShard storage user have not execution rights for internal functions (#366).
 
 ## [1.3.0] - 27-09-23
 

crud.lua

@@ -181,7 +181,9 @@ function crud.init_storage()
     sharding_metadata.init()
     readview.init()
 
-    _G._crud.storage_info_on_storage = utils.storage_info_on_storage
+    utils.init_storage_call('storage_info_on_storage',
+        utils.storage_info_on_storage
+    )
 end
 
 function crud.init_router()

crud/borders.lua

@@ -13,7 +13,8 @@ local BorderError = errors.new_class('BorderError', {capture_stack = false})
 
 local borders = {}
 
-local STAT_FUNC_NAME = '_crud.get_border_on_storage'
+local STAT_FUNC_NAME = 'get_border_on_storage'
+local CRUD_STAT_FUNC_NAME = utils.get_storage_call(STAT_FUNC_NAME)
 
 
 local function get_border_on_storage(border_name, space_name, index_id, field_names, fetch_latest_metadata)
@@ -43,7 +44,7 @@ local function get_border_on_storage(border_name, space_name, index_id, field_na
 end
 
 function borders.init()
-   _G._crud.get_border_on_storage = get_border_on_storage
+    utils.init_storage_call(STAT_FUNC_NAME, get_border_on_storage)
 end
 
 local is_closer
@@ -111,7 +112,7 @@ local function call_get_border_on_router(vshard_router, border_name, space_name,
         timeout = opts.timeout,
     }
     local results, err, storages_info = call.map(vshard_router,
-        STAT_FUNC_NAME,
+        CRUD_STAT_FUNC_NAME,
         {border_name, space_name, index.id, field_names, opts.fetch_latest_metadata},
         call_opts
     )

crud/common/sharding/sharding_metadata.lua

@@ -4,6 +4,7 @@ local log = require('log')
 
 local call = require('crud.common.call')
 local const = require('crud.common.const')
+local utils = require('crud.common.utils')
 local dev_checks = require('crud.common.dev_checks')
 local router_cache = require('crud.common.sharding.router_metadata_cache')
 local storage_cache = require('crud.common.sharding.storage_metadata_cache')
@@ -13,7 +14,8 @@ local sharding_utils = require('crud.common.sharding.utils')
 
 local FetchShardingMetadataError = errors.new_class('FetchShardingMetadataError', {capture_stack = false})
 
-local FETCH_FUNC_NAME = '_crud.fetch_on_storage'
+local FETCH_FUNC_NAME = 'fetch_on_storage'
+local CRUD_FETCH_FUNC_NAME = utils.get_storage_call(FETCH_FUNC_NAME)
 
 local sharding_metadata_module = {}
 
@@ -107,7 +109,7 @@ local _fetch_on_router = locked(function(vshard_router, space_name, metadata_map
         return
     end
 
-    local metadata_map, err = call.any(vshard_router, FETCH_FUNC_NAME, {}, {
+    local metadata_map, err = call.any(vshard_router, CRUD_FETCH_FUNC_NAME, {}, {
         timeout = timeout
     })
     if err ~= nil then
@@ -213,7 +215,7 @@ function sharding_metadata_module.reload_sharding_cache(vshard_router, space_nam
 end
 
 function sharding_metadata_module.init()
-   _G._crud.fetch_on_storage = sharding_metadata_module.fetch_on_storage
+    utils.init_storage_call(FETCH_FUNC_NAME, sharding_metadata_module.fetch_on_storage)
 end
 
 return sharding_metadata_module

crud/common/utils.lua

@@ -1,9 +1,11 @@
 local errors = require('errors')
 local ffi = require('ffi')
 local vshard = require('vshard')
+local vshard_util = require('vshard.util')
 local fun = require('fun')
 local bit = require('bit')
 local log = require('log')
+local luri = require('uri')
 
 local is_cartridge, cartridge = pcall(require, 'cartridge')
 local is_cartridge_hotreload, cartridge_hotreload = pcall(require, 'cartridge.hotreload')
@@ -27,6 +29,19 @@ local fiber = require('fiber')
 
 local utils = {}
 
+--- Returns a full call string for a storage function name.
+--
+--  @param string name a base name of the storage function.
+--
+--  @return a full string for the call.
+function utils.get_storage_call(name)
+    dev_checks('string')
+
+    return '_crud.' .. name
+end
+
+local CRUD_STORAGE_INFO_FUNC_NAME = utils.get_storage_call('storage_info_on_storage')
+
 local space_format_cache = setmetatable({}, {__mode = 'k'})
 
 -- copy from LuaJIT lj_char.c
@@ -1026,8 +1041,11 @@ function utils.update_storage_call_error_description(err, func_name, replicaset_
         return nil
     end
 
-    if err.type == 'ClientError' and type(err.message) == 'string' then
-        if err.message == string.format("Procedure '%s' is not defined", func_name) then
+    if (err.type == 'ClientError' or err.type == 'AccessDeniedError')
+        and type(err.message) == 'string' then
+        local not_defined_str = string.format("Procedure '%s' is not defined", func_name)
+        local access_denied_str = string.format("Execute access to function '%s' is denied", func_name)
+        if err.message == not_defined_str or err.message:startswith(access_denied_str) then
             if func_name:startswith('_crud.') then
                 err = NotInitializedError:new("Function %s is not registered: " ..
                     "crud isn't initialized on replicaset %q or crud module versions mismatch " ..
@@ -1113,7 +1131,7 @@ function utils.storage_info(opts)
                 status = "error",
                 is_master = replicaset.master == replica
             }
-            local ok, res = pcall(replica.conn.call, replica.conn, "_crud.storage_info_on_storage",
+            local ok, res = pcall(replica.conn.call, replica.conn, CRUD_STORAGE_INFO_FUNC_NAME,
                                   {}, async_opts)
             if ok then
                 futures_by_replicas[replica_uuid] = res
@@ -1169,6 +1187,55 @@ function utils.storage_info_on_storage()
     return {status = "running"}
 end
 
+--- Initializes a storage function by its name.
+--
+--  It adds the function into the global scope by its name and required
+--  rights to a vshard storage user.
+--
+--  @function init_storage_call
+--
+--  @param string name a name of the function.
+--  @param function func the function.
+--
+--  @return nil
+function utils.init_storage_call(name, func)
+    dev_checks('string', 'function')
+
+    rawset(_G['_crud'], name, func)
+    name = utils.get_storage_call(name)
+
+    if type(box.cfg) ~= 'table' or box.cfg.read_only then
+        -- If the storage is not configured yet then it could be
+        -- a unit-test or something else.
+        --
+        -- Anyway, it is out of scope of the call to detect and to raise an
+        -- error for the case.
+        return
+    end
+
+    local ok, storage_info = pcall(vshard.storage.info)
+    if not ok then
+        -- The storage is not configured. This is also none of our business.
+        return
+    end
+
+    local replicaset_uuid = vshard_util.replicaset_uuid()
+    local replicaset_info = storage_info.replicasets[replicaset_uuid]
+    if replicaset_info == nil or replicaset_info.master == nil then
+        -- It should not happen, but the crud could still work fine in some
+        -- configurations (Cartridge role, as example). So we just print a
+        -- warning here.
+        log.warn('Failed to find a vshard configuration for replicaset with ' ..
+            'replicaset_uuid %s. Rights to execute functions may be missed.',
+            replicaset_uuid)
+        return
+    end
+
+    local login = luri.parse(replicaset_info.master.uri).login or 'guest'
+    box.schema.func.create(name, {if_not_exists = true})
+    box.schema.user.grant(login, 'execute', 'function', name, {if_not_exists=true})
+end
+
 local expected_vshard_api = {
     'routeall', 'route', 'bucket_id_strcrc32',
     'callrw', 'callro', 'callbro', 'callre',

crud/count.lua

@@ -17,7 +17,8 @@ local compare_conditions = require('crud.compare.conditions')
 
 local CountError = errors.new_class('CountError', {capture_stack = false})
 
-local COUNT_FUNC_NAME = '_crud.count_on_storage'
+local COUNT_FUNC_NAME = 'count_on_storage'
+local CRUD_COUNT_FUNC_NAME = utils.get_storage_call(COUNT_FUNC_NAME)
 
 local count = {}
 
@@ -86,7 +87,7 @@ local function count_on_storage(space_name, index_id, conditions, opts)
 end
 
 function count.init()
-    _G._crud.count_on_storage = count_on_storage
+    utils.init_storage_call(COUNT_FUNC_NAME, count_on_storage)
 end
 
 local check_count_safety_rl = ratelimit.new()
@@ -240,7 +241,7 @@ local function call_count_on_router(vshard_router, space_name, user_conditions,
         skip_sharding_hash_check = skip_sharding_hash_check,
     }
 
-    local results, err = call.map(vshard_router, COUNT_FUNC_NAME, {
+    local results, err = call.map(vshard_router, CRUD_COUNT_FUNC_NAME, {
         space_name, plan.index_id, plan.conditions, count_opts
     }, call_opts)
 

crud/delete.lua

@@ -14,7 +14,8 @@ local DeleteError = errors.new_class('DeleteError', {capture_stack = false})
 
 local delete = {}
 
-local DELETE_FUNC_NAME = '_crud.delete_on_storage'
+local DELETE_FUNC_NAME = 'delete_on_storage'
+local CRUD_DELETE_FUNC_NAME = utils.get_storage_call(DELETE_FUNC_NAME
 
 local function delete_on_storage(space_name, key, field_names, opts)
     dev_checks('string', '?', '?table', {
@@ -52,7 +53,7 @@ local function delete_on_storage(space_name, key, field_names, opts)
 end
 
 function delete.init()
-   _G._crud.delete_on_storage = delete_on_storage
+    utils.init_storage_call(DELETE_FUNC_NAME, delete_on_storage)
 end
 
 -- returns result, err, need_reload
@@ -127,7 +128,7 @@ local function call_delete_on_router(vshard_router, space_name, key, opts)
     }
 
     local storage_result, err = call.single(vshard_router,
-        bucket_id_data.bucket_id, DELETE_FUNC_NAME,
+        bucket_id_data.bucket_id, CRUD_DELETE_FUNC_NAME,
         {space_name, key, opts.fields, delete_on_storage_opts},
         call_opts
     )

crud/get.lua

@@ -14,7 +14,8 @@ local GetError = errors.new_class('GetError', {capture_stack = false})
 
 local get = {}
 
-local GET_FUNC_NAME = '_crud.get_on_storage'
+local GET_FUNC_NAME = 'get_on_storage'
+local CRUD_GET_FUNC_NAME = utils.get_storage_call(GET_FUNC_NAME)
 
 local function get_on_storage(space_name, key, field_names, opts)
     dev_checks('string', '?', '?table', {
@@ -50,7 +51,7 @@ local function get_on_storage(space_name, key, field_names, opts)
 end
 
 function get.init()
-   _G._crud.get_on_storage = get_on_storage
+    utils.init_storage_call(GET_FUNC_NAME, get_on_storage)
 end
 
 -- returns result, err, need_reload
@@ -125,7 +126,7 @@ local function call_get_on_router(vshard_router, space_name, key, opts)
     }
 
     local storage_result, err = call.single(vshard_router,
-        bucket_id_data.bucket_id, GET_FUNC_NAME,
+        bucket_id_data.bucket_id, CRUD_GET_FUNC_NAME,
         {space_name, key, opts.fields, get_on_storage_opts},
         call_opts
     )

crud/insert.lua

@@ -12,7 +12,8 @@ local InsertError = errors.new_class('InsertError', {capture_stack = false})
 
 local insert = {}
 
-local INSERT_FUNC_NAME = '_crud.insert_on_storage'
+local INSERT_FUNC_NAME = 'insert_on_storage'
+local CRUD_INSERT_FUNC_NAME = utils.get_storage_call(INSERT_FUNC_NAME)
 
 local function insert_on_storage(space_name, tuple, opts)
     dev_checks('string', 'table', {
@@ -53,7 +54,7 @@ local function insert_on_storage(space_name, tuple, opts)
 end
 
 function insert.init()
-   _G._crud.insert_on_storage = insert_on_storage
+    utils.init_storage_call(INSERT_FUNC_NAME, insert_on_storage)
 end
 
 -- returns result, err, need_reload
@@ -102,7 +103,7 @@ local function call_insert_on_router(vshard_router, space_name, original_tuple,
     }
 
     local storage_result, err = call.single(vshard_router,
-        sharding_data.bucket_id, INSERT_FUNC_NAME,
+        sharding_data.bucket_id, CRUD_INSERT_FUNC_NAME,
         {space_name, tuple, insert_on_storage_opts},
         call_opts
     )

crud/insert_many.lua

@@ -16,7 +16,8 @@ local InsertManyError = errors.new_class('InsertManyError', {capture_stack = fal
 
 local insert_many = {}
 
-local INSERT_MANY_FUNC_NAME = '_crud.insert_many_on_storage'
+local INSERT_MANY_FUNC_NAME = 'insert_many_on_storage'
+local CRUD_INSERT_MANY_FUNC_NAME = utils.get_storage_call(INSERT_MANY_FUNC_NAME)
 
 local function insert_many_on_storage(space_name, tuples, opts)
     dev_checks('string', 'table', {
@@ -123,7 +124,7 @@ local function insert_many_on_storage(space_name, tuples, opts)
 end
 
 function insert_many.init()
-    _G._crud.insert_many_on_storage = insert_many_on_storage
+    utils.init_storage_call(INSERT_MANY_FUNC_NAME, insert_many_on_storage)
 end
 
 -- returns result, err, need_reload
@@ -175,7 +176,7 @@ local function call_insert_many_on_router(vshard_router, space_name, original_tu
 
     local postprocessor = BatchPostprocessor:new(vshard_router)
 
-    local rows, errs, storages_info = call.map(vshard_router, INSERT_MANY_FUNC_NAME, nil, {
+    local rows, errs, storages_info = call.map(vshard_router, CRUD_INSERT_MANY_FUNC_NAME, nil, {
         timeout = opts.timeout,
         mode = 'write',
         iter = iter,

crud/len.lua

@@ -8,7 +8,8 @@ local LenError = errors.new_class('LenError', {capture_stack = false})
 
 local len = {}
 
-local LEN_FUNC_NAME = '_crud.len_on_storage'
+local LEN_FUNC_NAME = 'len_on_storage'
+local CRUD_LEN_FUNC_NAME = utils.get_storage_call(LEN_FUNC_NAME)
 
 local function len_on_storage(space_name)
     dev_checks('string|number')
@@ -17,7 +18,7 @@ local function len_on_storage(space_name)
 end
 
 function len.init()
-    _G._crud.len_on_storage = len_on_storage
+    utils.init_storage_call(LEN_FUNC_NAME, len_on_storage)
 end
 
 --- Calculates the number of tuples in the space for memtx engine
@@ -61,7 +62,7 @@ function len.call(space_name, opts)
         return nil, LenError:new("Space %q doesn't exist", space_name)
     end
 
-    local results, err = vshard_router:map_callrw(LEN_FUNC_NAME, {space_name}, opts)
+    local results, err = vshard_router:map_callrw(CRUD_LEN_FUNC_NAME, {space_name}, opts)
 
     if err ~= nil then
         return nil, LenError:new("Failed to call len on storage-side: %s", err)