crud: add execution access to a VShard user

The patch adds execution access on a stroage for 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

@@ -22,6 +22,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
 

README.md

@@ -86,11 +86,14 @@ The CRUD operations should be called from router.
 All VShard storages should call `crud.init_storage()` after
 `vshard.storage.cfg()` (or enable the `crud-storage` role for Cartridge)
 first to initialize storage-side functions that are used to manipulate data
-across the cluster.
+across the cluster. The storage-side functions have the same access
+as a user calling `crud.init_storage()`. Therefore, if `crud` do not have
+enough access to modify some space, then you need to give access to the user.
 
 All VShard routers should call `crud.init_router()` after `vshard.router.cfg()`
 (or enable the `crud-router` role for Cartridge) to make `crud` functions
-callable via `net.box`.
+callable via `net.box`. If a user is allowed to execute `crud` functions on
+the router-side then the user does not need additional access on storages.
 
 You can check out an example of the configuration for local development
 (a single instance that combines router and storage) in

crud.lua

@@ -23,6 +23,9 @@ local stats = require('crud.stats')
 local readview = require('crud.readview')
 local schema = require('crud.schema')
 
+local vshard = require('vshard')
+local luri = require('uri')
+
 local crud = {}
 
 -- @refer crud.version
@@ -165,28 +168,54 @@ crud.schema = schema.call
 -- @function init
 --
 function crud.init_storage()
+    if type(box.cfg) ~= 'table' then
+        error('box.cfg() must be called first')
+    end
+
+    local ok, storage_info = pcall(vshard.storage.info)
+    if not ok then
+        error('vshard.storage.cfg() must be called first')
+    end
+
+    local box_info = box.info()
+    local replicaset_uuid
+    if box_info.replicaset ~= nil then
+        replicaset_uuid = box_info.replicaset.uuid
+    else
+        replicaset_uuid = box_info.cluster.uuid
+    end
+    local replicaset_info = storage_info.replicasets[replicaset_uuid]
+    if replicaset_info == nil or replicaset_info.master == nil then
+        error(string.format('Failed to find a vshard configuration for ' ..
+            ' replicaset with replicaset_uuid %s.',
+            replicaset_uuid))
+    end
+    local user = luri.parse(replicaset_info.master.uri).login or 'guest'
+
     if rawget(_G, '_crud') == nil then
         rawset(_G, '_crud', {})
     end
 
-    insert.init()
-    insert_many.init()
-    get.init()
-    replace.init()
-    replace_many.init()
-    update.init()
-    upsert.init()
-    upsert_many.init()
-    delete.init()
-    select.init()
-    truncate.init()
-    len.init()
-    count.init()
-    borders.init()
-    sharding_metadata.init()
-    readview.init()
-
-    _G._crud.storage_info_on_storage = utils.storage_info_on_storage
+    insert.init(user)
+    insert_many.init(user)
+    get.init(user)
+    replace.init(user)
+    replace_many.init(user)
+    update.init(user)
+    upsert.init(user)
+    upsert_many.init(user)
+    delete.init(user)
+    select.init(user)
+    truncate.init(user)
+    len.init(user)
+    count.init(user)
+    borders.init(user)
+    sharding_metadata.init(user)
+    readview.init(user)
+
+    utils.init_storage_call(user, '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)
@@ -42,8 +43,8 @@ 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
+function borders.init(user)
+    utils.init_storage_call(user, 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
@@ -212,8 +214,8 @@ function sharding_metadata_module.reload_sharding_cache(vshard_router, space_nam
     end
 end
 
-function sharding_metadata_module.init()
-   _G._crud.fetch_on_storage = sharding_metadata_module.fetch_on_storage
+function sharding_metadata_module.init(user)
+    utils.init_storage_call(user, FETCH_FUNC_NAME, sharding_metadata_module.fetch_on_storage)
 end
 
 return sharding_metadata_module

crud/common/utils.lua

@@ -27,6 +27,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
@@ -1034,8 +1047,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 " ..
@@ -1121,7 +1137,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
@@ -1177,6 +1193,31 @@ 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(user, name, func)
+    dev_checks('string', 'string', 'function')
+
+    rawset(_G['_crud'], name, func)
+
+    if box.info().ro then
+        return
+    end
+
+    name = utils.get_storage_call(name)
+    box.schema.func.create(name, {setuid = true, if_not_exists = true})
+    box.schema.user.grant(user, '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 = {}
 
@@ -85,8 +86,8 @@ local function count_on_storage(space_name, index_id, conditions, opts)
     return tuples_count
 end
 
-function count.init()
-    _G._crud.count_on_storage = count_on_storage
+function count.init(user)
+    utils.init_storage_call(user, 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', {
@@ -51,8 +52,8 @@ local function delete_on_storage(space_name, key, field_names, opts)
     })
 end
 
-function delete.init()
-   _G._crud.delete_on_storage = delete_on_storage
+function delete.init(user)
+    utils.init_storage_call(user, 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', {
@@ -49,8 +50,8 @@ local function get_on_storage(space_name, key, field_names, opts)
     })
 end
 
-function get.init()
-   _G._crud.get_on_storage = get_on_storage
+function get.init(user)
+    utils.init_storage_call(user, 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', {
@@ -52,8 +53,8 @@ local function insert_on_storage(space_name, tuple, opts)
     })
 end
 
-function insert.init()
-   _G._crud.insert_on_storage = insert_on_storage
+function insert.init(user)
+    utils.init_storage_call(user, 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
     )