File Manager - BorneoXploit

Editing: lua_redis.lua

--[[
Copyright (c) 2022, Vsevolod Stakhov <vsevolod@rspamd.com>

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
]]--

local logger = require "rspamd_logger"
local lutil = require "lua_util"
local rspamd_util = require "rspamd_util"
local T = require "lua_shape.core"
local PluginSchema = require "lua_shape.plugin_schema"

local exports = {}

local E = {}
local N = "lua_redis"

local db_schema = T.one_of({
  T.transform(T.number(), tostring),
  T.string()
}):optional():doc({ summary = "Database number" })

local common_schema = T.table({
  timeout = T.one_of({
    T.number({ min = 0 }),
    T.transform(T.string(), lutil.parse_time_interval)
  }):optional():doc({ summary = "Connection timeout (seconds)" }),
  db = db_schema,
  database = db_schema,
  dbname = db_schema,
  prefix = T.string():optional():doc({ summary = "Key prefix" }),
  username = T.string():optional():doc({ summary = "Username" }),
  password = T.string():optional():doc({ summary = "Password" }),
  -- TLS options
  ssl = T.boolean():optional():doc({ summary = "Enable TLS to Redis" }),
  no_ssl_verify = T.boolean():optional():doc({ summary = "Disable TLS certificate verification" }),
  ssl_ca = T.string():optional():doc({ summary = "CA certificate file" }),
  ssl_ca_dir = T.string():optional():doc({ summary = "CA certificates directory" }),
  ssl_cert = T.string():optional():doc({ summary = "Client certificate file (PEM)" }),
  ssl_key = T.string():optional():doc({ summary = "Client private key file (PEM)" }),
  sni = T.string():optional():doc({ summary = "SNI server name override" }),
  expand_keys = T.boolean():optional():doc({ summary = "Expand keys" }),
  sentinels = T.one_of({
    T.string(),
    T.array(T.string())
  }):optional():doc({ summary = "Sentinel servers" }),
  sentinel_watch_time = T.one_of({
    T.number({ min = 0 }),
    T.transform(T.string(), lutil.parse_time_interval)
  }):optional():doc({ summary = "Sentinel watch time" }),
  sentinel_masters_pattern = T.string():optional():doc({ summary = "Sentinel masters pattern" }),
  sentinel_master_maxerrors = T.one_of({
    T.number(),
    T.transform(T.string(), tonumber)
  }):optional():doc({ summary = "Sentinel master max errors" }),
  sentinel_username = T.string():optional():doc({ summary = "Sentinel username" }),
  sentinel_password = T.string():optional():doc({ summary = "Sentinel password" }),
  redis_version = T.one_of({
    T.number(),
    T.transform(T.string(), tonumber)
  }):optional():doc({ summary = "Redis server version (6 or 7)" }),
}, { open = true })

local function extend_fields(...)
  local res = {}
  for i = 1, select('#', ...) do
    local tbl = select(i, ...)
    if tbl then
      for k, v in pairs(tbl) do
        res[k] = v
      end
    end
  end
  return res
end

-- Register common schema as redis_common first
PluginSchema.register("mixins.redis_common", common_schema)

-- Create full redis mixin schema for documentation
local redis_mixin_schema = T.one_of({
  {
    name = "common_only",
    schema = T.table({}, {
      open = true,
      mixins = { T.mixin(common_schema, { as = "redis_common" }) }
    }):doc({ summary = "Redis with no explicit servers (uses localhost)" })
  },
  {
    name = "read_servers",
    schema = T.table({
      read_servers = T.one_of({
        T.string():doc({ summary = "Single Redis server" }),
        T.array(T.string()):doc({ summary = "Multiple Redis servers" })
      }):doc({ summary = "Read servers list" })
    }, {
      open = true,
      mixins = { T.mixin(common_schema, { as = "redis_common" }) }
    }):doc({ summary = "Redis with read-only servers" })
  },
  {
    name = "write_servers",
    schema = T.table({
      write_servers = T.one_of({
        T.string():doc({ summary = "Single Redis server" }),
        T.array(T.string()):doc({ summary = "Multiple Redis servers" })
      }):doc({ summary = "Write servers list" })
    }, {
      open = true,
      mixins = { T.mixin(common_schema, { as = "redis_common" }) }
    }):doc({ summary = "Redis with write-only servers" })
  },
  {
    name = "rw_servers",
    schema = T.table({
      read_servers = T.one_of({
        T.string():doc({ summary = "Single Redis server" }),
        T.array(T.string()):doc({ summary = "Multiple Redis servers" })
      }):doc({ summary = "Read servers list" }),
      write_servers = T.one_of({
        T.string():doc({ summary = "Single Redis server" }),
        T.array(T.string()):doc({ summary = "Multiple Redis servers" })
      }):doc({ summary = "Write servers list" })
    }, {
      open = true,
      mixins = { T.mixin(common_schema, { as = "redis_common" }) }
    }):doc({ summary = "Redis with separate read and write servers" })
  },
  {
    name = "servers",
    schema = T.table({
      servers = T.one_of({
        T.string():doc({ summary = "Single Redis server" }),
        T.array(T.string()):doc({ summary = "Multiple Redis servers" })
      }):doc({ summary = "Redis servers list" })
    }, {
      open = true,
      mixins = { T.mixin(common_schema, { as = "redis_common" }) }
    }):doc({ summary = "Redis with server list" })
  },
  {
    name = "server_legacy",
    schema = T.table({
      server = T.one_of({
        T.string():doc({ summary = "Single Redis server" }),
        T.array(T.string()):doc({ summary = "Multiple Redis servers" })
      }):doc({ summary = "Redis server (legacy)" })
    }, {
      open = true,
      mixins = { T.mixin(common_schema, { as = "redis_common" }) }
    }):doc({ summary = "Redis with legacy server parameter" })
  }
}):doc({ summary = "Redis connection configuration" })

PluginSchema.register("mixins.redis", redis_mixin_schema)

local enrich_schema = function(external)
  local external_fields = external or {}

return T.one_of({
    {
      name = "common_only",
      schema = T.table(extend_fields(external_fields), {
        open = true,
        mixins = {
          T.mixin(common_schema, { as = "redis" })
        }
      })
    },
    {
      name = "read_servers",
      schema = T.table(extend_fields(external_fields, {
        read_servers = T.one_of({
          T.string(),
          T.array(T.string())
        })
      }), {
        open = true,
        mixins = {
          T.mixin(common_schema, { as = "redis" })
        }
      })
    },
    {
      name = "write_servers",
      schema = T.table(extend_fields(external_fields, {
        write_servers = T.one_of({
          T.string(),
          T.array(T.string())
        })
      }), {
        open = true,
        mixins = {
          T.mixin(common_schema, { as = "redis" })
        }
      })
    },
    {
      name = "rw_servers",
      schema = T.table(extend_fields(external_fields, {
        read_servers = T.one_of({
          T.string(),
          T.array(T.string())
        }),
        write_servers = T.one_of({
          T.string(),
          T.array(T.string())
        })
      }), {
        open = true,
        mixins = {
          T.mixin(common_schema, { as = "redis" })
        }
      })
    },
    {
      name = "servers",
      schema = T.table(extend_fields(external_fields, {
        servers = T.one_of({
          T.string(),
          T.array(T.string())
        })
      }), {
        open = true,
        mixins = {
          T.mixin(common_schema, { as = "redis" })
        }
      })
    },
    {
      name = "server_legacy",
      schema = T.table(extend_fields(external_fields, {
        server = T.one_of({
          T.string(),
          T.array(T.string())
        })
      }), {
        open = true,
        mixins = {
          T.mixin(common_schema, { as = "redis" })
        }
      })
    }
  })
end

exports.enrich_schema = enrich_schema

local function flatten_redis_table(tbl)
  local res = {}
  for i = 1, #tbl, 2 do
    res[tbl[i]] = tbl[i + 1]
  end

return res
end

local function redis_query_sentinel(ev_base, params, initialised)
  -- Coroutines syntax
  local rspamd_redis = require "rspamd_redis"
  local sentinels = params.sentinels
  local addr = sentinels:get_upstream_round_robin()

if not addr then
    logger.errx(rspamd_config,
        'no sentinel upstream available (all dead or pending DNS resolution); skipping sentinel watch tick')
    return
  end

local host = addr:get_addr()
  local masters = {}
  local process_masters -- Function that is called to process masters data

local function masters_cb(err, result)
    if not err and result and type(result) == 'table' then

local pending_subrequests = 0

for _, m in ipairs(result) do
        local master = flatten_redis_table(m)

-- Wrap IPv6-addresses in brackets
        if (master.ip:match(":")) then
          master.ip = "[" .. master.ip .. "]"
        end

if params.sentinel_masters_pattern then
          if master.name:match(params.sentinel_masters_pattern) then
            lutil.debugm(N, 'found master %s with ip %s and port %s',
                master.name, master.ip, master.port)
            masters[master.name] = master
          else
            lutil.debugm(N, 'skip master %s with ip %s and port %s, pattern %s',
                master.name, master.ip, master.port, params.sentinel_masters_pattern)
          end
        else
          lutil.debugm(N, 'found master %s with ip %s and port %s',
              master.name, master.ip, master.port)
          masters[master.name] = master
        end
      end

-- For each master we need to get a list of slaves
      for k, v in pairs(masters) do
        v.slaves = {}
        local function slaves_cb(slave_err, slave_result)
          if not slave_err and type(slave_result) == 'table' then
            for _, s in ipairs(slave_result) do
              local slave = flatten_redis_table(s)
              lutil.debugm(N, rspamd_config,
                  'found slave for master %s with ip %s and port %s',
                  v.name, slave.ip, slave.port)
              -- Wrap IPv6-addresses in brackets
              if (slave.ip:match(":")) then
                slave.ip = "[" .. slave.ip .. "]"
              end
              v.slaves[#v.slaves + 1] = slave
            end
          else
            logger.errx('cannot get slaves data from Redis Sentinel %s: %s',
                host:to_string(true), slave_err)
            addr:fail()
          end

pending_subrequests = pending_subrequests - 1

if pending_subrequests == 0 then
            -- Finalize masters and slaves
            process_masters()
          end
        end

local ret = rspamd_redis.make_request {
          host = addr:get_addr(),
          timeout = params.timeout,
          username = params.sentinel_username,
          password = params.sentinel_password,
          config = rspamd_config,
          ev_base = ev_base,
          cmd = 'SENTINEL',
          args = { 'slaves', k },
          no_pool = true,
          callback = slaves_cb
        }

if not ret then
          logger.errx(rspamd_config, 'cannot connect sentinel when query slaves at address: %s',
              host:to_string(true))
          addr:fail()
        else
          pending_subrequests = pending_subrequests + 1
        end
      end

addr:ok()
    else
      logger.errx('cannot get masters data from Redis Sentinel %s: %s',
          host:to_string(true), err)
      addr:fail()
    end
  end

local ret = rspamd_redis.make_request {
    host = addr:get_addr(),
    timeout = params.timeout,
    config = rspamd_config,
    ev_base = ev_base,
    username = params.sentinel_username,
    password = params.sentinel_password,
    cmd = 'SENTINEL',
    args = { 'masters' },
    no_pool = true,
    callback = masters_cb,
  }

if not ret then
    logger.errx(rspamd_config, 'cannot connect sentinel at address: %s',
        host:to_string(true))
    addr:fail()
  end

process_masters = function()
    -- We now form new strings for masters and slaves
    local read_servers_tbl, write_servers_tbl = {}, {}

for _, master in pairs(masters) do
      write_servers_tbl[#write_servers_tbl + 1] = string.format(
          '%s:%s', master.ip, master.port
      )
      read_servers_tbl[#read_servers_tbl + 1] = string.format(
          '%s:%s', master.ip, master.port
      )

for _, slave in ipairs(master.slaves) do
        if slave['master-link-status'] == 'ok' then
          read_servers_tbl[#read_servers_tbl + 1] = string.format(
              '%s:%s', slave.ip, slave.port
          )
        end
      end
    end

table.sort(read_servers_tbl)
    table.sort(write_servers_tbl)

local read_servers_str = table.concat(read_servers_tbl, ',')
    local write_servers_str = table.concat(write_servers_tbl, ',')

lutil.debugm(N, rspamd_config,
        'new servers list: %s read; %s write',
        read_servers_str,
        write_servers_str)

if read_servers_str ~= params.read_servers_str then
      local upstream_list = require "rspamd_upstream_list"

local read_upstreams = upstream_list.create(rspamd_config,
          read_servers_str, 6379)

if read_upstreams then
        logger.infox(rspamd_config, 'sentinel %s: replace read servers with new list: %s',
            host:to_string(true), read_servers_str)
        params.read_servers = read_upstreams
        params.read_servers_str = read_servers_str
      end
    end

if write_servers_str ~= params.write_servers_str then
      local upstream_list = require "rspamd_upstream_list"

local write_upstreams = upstream_list.create(rspamd_config,
          write_servers_str, 6379)

if write_upstreams then
        logger.infox(rspamd_config, 'sentinel %s: replace write servers with new list: %s',
            host:to_string(true), write_servers_str)
        params.write_servers = write_upstreams
        params.write_servers_str = write_servers_str

local queried = false

local function monitor_failures(up, _, count)
          if count > params.sentinel_master_maxerrors and not queried then
            logger.infox(rspamd_config, 'sentinel: master with address %s, caused %s failures, try to query sentinel',
                host:to_string(true), count)
            queried = true -- Avoid multiple checks caused by this monitor
            redis_query_sentinel(ev_base, params, true)
          end
        end

write_upstreams:add_watcher('failure', monitor_failures)
      end
    end
  end

end

local function add_redis_sentinels(params)
  local upstream_list = require "rspamd_upstream_list"

local upstreams_sentinels = upstream_list.create(rspamd_config,
      params.sentinels, 5000)

if not upstreams_sentinels then
    logger.errx(rspamd_config, 'cannot load redis sentinels string: %s',
        params.sentinels)

return
  end

params.sentinels = upstreams_sentinels

if not params.sentinel_watch_time then
    params.sentinel_watch_time = 60 -- Each minute
  end

if not params.sentinel_master_maxerrors then
    params.sentinel_master_maxerrors = 2 -- Maximum number of errors before rechecking
  end

rspamd_config:add_on_load(function(_, ev_base, worker)
    local initialised = false
    if worker:is_scanner() or worker:get_type() == 'fuzzy' then
      rspamd_config:add_periodic(ev_base, 0.0, function()
        redis_query_sentinel(ev_base, params, initialised)
        initialised = true

return params.sentinel_watch_time
      end, false)
    end
  end)
end

local cached_results = {}

local function calculate_redis_hash(params)
  local cr = require "rspamd_cryptobox_hash"

local h = cr.create()

local function rec_hash(k, v)
    if type(v) == 'string' then
      h:update(k)
      h:update(v)
    elseif type(v) == 'number' then
      h:update(k)
      h:update(tostring(v))
    elseif type(v) == 'table' then
      for kk, vv in pairs(v) do
        rec_hash(kk, vv)
      end
    end
  end

rec_hash('top', params)

return h:base32()
end

local function process_redis_opts(options, redis_params)
  local default_timeout = 1.0
  local default_expand_keys = false

if not redis_params['timeout'] or redis_params['timeout'] == default_timeout then
    if options['timeout'] then
      redis_params['timeout'] = tonumber(options['timeout'])
    else
      redis_params['timeout'] = default_timeout
    end
  end

if options['prefix'] and not redis_params['prefix'] then
    redis_params['prefix'] = options['prefix']
  end

if options['redis_version'] and not redis_params['redis_version'] then
    redis_params['redis_version'] = tonumber(options['redis_version'])
  end

if type(options['expand_keys']) == 'boolean' then
    redis_params['expand_keys'] = options['expand_keys']
  else
    redis_params['expand_keys'] = default_expand_keys
  end

if not redis_params['db'] then
    if options['db'] then
      redis_params['db'] = tostring(options['db'])
    elseif options['dbname'] then
      redis_params['db'] = tostring(options['dbname'])
    elseif options['database'] then
      redis_params['db'] = tostring(options['database'])
    end
  end
  if options['username'] and not redis_params['username'] then
    redis_params['username'] = options['username']
  end
  if options['password'] and not redis_params['password'] then
    redis_params['password'] = options['password']
  end

-- TLS passthrough
  if options['ssl'] ~= nil and redis_params['ssl'] == nil then
    redis_params['ssl'] = options['ssl'] and true or false
  end
  if options['no_ssl_verify'] ~= nil and redis_params['no_ssl_verify'] == nil then
    redis_params['no_ssl_verify'] = options['no_ssl_verify'] and true or false
  end
  if options['ssl_ca'] and not redis_params['ssl_ca'] then
    redis_params['ssl_ca'] = options['ssl_ca']
  end
  if options['ssl_ca_dir'] and not redis_params['ssl_ca_dir'] then
    redis_params['ssl_ca_dir'] = options['ssl_ca_dir']
  end
  if options['ssl_cert'] and not redis_params['ssl_cert'] then
    redis_params['ssl_cert'] = options['ssl_cert']
  end
  if options['ssl_key'] and not redis_params['ssl_key'] then
    redis_params['ssl_key'] = options['ssl_key']
  end
  if options['sni'] and not redis_params['sni'] then
    redis_params['sni'] = options['sni']
  end

if not redis_params.sentinels and options.sentinels then
    redis_params.sentinels = options.sentinels
  end

if options['sentinel_masters_pattern'] and not redis_params['sentinel_masters_pattern'] then
    redis_params['sentinel_masters_pattern'] = options['sentinel_masters_pattern']
  end

if options['sentinel_watch_time'] and not redis_params['sentinel_watch_time'] then
    redis_params['sentinel_watch_time'] = options['sentinel_watch_time']
  end

if options['sentinel_username'] and not redis_params['sentinel_username'] then
    redis_params['sentinel_username'] = options['sentinel_username']
  end

if options['sentinel_password'] and not redis_params['sentinel_password'] then
    redis_params['sentinel_password'] = options['sentinel_password']
  end

if options['sentinel_master_maxerrors'] and not redis_params['sentinel_master_maxerrors'] then
    redis_params['sentinel_master_maxerrors'] = options['sentinel_master_maxerrors']
  end

end

local function enrich_defaults(rspamd_config, module, redis_params)
  if rspamd_config then
    local opts = rspamd_config:get_all_opt('redis')

if opts then
      if module then
        if opts[module] then
          process_redis_opts(opts[module], redis_params)
        end
      end

process_redis_opts(opts, redis_params)
    end
  end
end

local function maybe_return_cached(redis_params)
  local h = calculate_redis_hash(redis_params)

if cached_results[h] then
    lutil.debugm(N, 'reused redis server: %s', redis_params)
    return cached_results[h]
  end

redis_params.hash = h
  cached_results[h] = redis_params

if not redis_params.read_only and redis_params.sentinels then
    add_redis_sentinels(redis_params)
  end

lutil.debugm(N, 'loaded new redis server: %s', redis_params)
  return redis_params
end

--[[[
-- @module lua_redis
-- This module contains helper functions for working with Redis
--]]
local function process_redis_options(options, rspamd_config, result)
  local default_port = 6379
  local upstream_list = require "rspamd_upstream_list"
  local read_only = true

-- Try to get read servers:
  local upstreams_read, upstreams_write

if options['read_servers'] then
    if rspamd_config then
      upstreams_read = upstream_list.create(rspamd_config,
          options['read_servers'], default_port)
    else
      upstreams_read = upstream_list.create(options['read_servers'],
          default_port)
    end

result.read_servers_str = options['read_servers']
  elseif options['servers'] then
    if rspamd_config then
      upstreams_read = upstream_list.create(rspamd_config,
          options['servers'], default_port)
    else
      upstreams_read = upstream_list.create(options['servers'], default_port)
    end

result.read_servers_str = options['servers']
    read_only = false
  elseif options['server'] then
    if rspamd_config then
      upstreams_read = upstream_list.create(rspamd_config,
          options['server'], default_port)
    else
      upstreams_read = upstream_list.create(options['server'], default_port)
    end

result.read_servers_str = options['server']
    read_only = false
  end

if upstreams_read then
    if options['write_servers'] then
      if rspamd_config then
        upstreams_write = upstream_list.create(rspamd_config,
            options['write_servers'], default_port)
      else
        upstreams_write = upstream_list.create(options['write_servers'],
            default_port)
      end
      result.write_servers_str = options['write_servers']
      read_only = false
    elseif not read_only then
      upstreams_write = upstreams_read
      result.write_servers_str = result.read_servers_str
    end
  end

-- Store options
  process_redis_opts(options, result)

if read_only and not upstreams_write then
    result.read_only = true
  elseif upstreams_write then
    result.read_only = false
  end

if upstreams_read then
    result.read_servers = upstreams_read

if upstreams_write then
      result.write_servers = upstreams_write
    end

return true
  end

lutil.debugm(N, rspamd_config,
      'cannot load redis server from obj: %s, processed to %s',
      options, result)

return false
end

--[[[
@function try_load_redis_servers(options, rspamd_config, no_fallback)
Tries to load redis servers from the specified `options` object.
Returns `redis_params` table or nil in case of failure

--]]
exports.try_load_redis_servers = function(options, rspamd_config, no_fallback, module_name)
  local result = {}

if process_redis_options(options, rspamd_config, result) then
    if not no_fallback then
      enrich_defaults(rspamd_config, module_name, result)
    end
    return maybe_return_cached(result)
  end
end

-- This function parses redis server definition using either
-- specific server string for this module or global
-- redis section
local function rspamd_parse_redis_server(module_name, module_opts, no_fallback)
  local result = {}

-- Try local options
  local opts
  lutil.debugm(N, rspamd_config, 'try load redis config for: %s', module_name)
  if not module_opts then
    opts = rspamd_config:get_all_opt(module_name)
  else
    opts = module_opts
  end

if opts then
    local ret

if opts.redis then
      ret = process_redis_options(opts.redis, rspamd_config, result)

if ret then
        if not no_fallback then
          enrich_defaults(rspamd_config, module_name, result)
        end
        return maybe_return_cached(result)
      end
    end

ret = process_redis_options(opts, rspamd_config, result)

if ret then
      if not no_fallback then
        enrich_defaults(rspamd_config, module_name, result)
      end
      return maybe_return_cached(result)
    end
  end

if no_fallback then
    logger.infox(rspamd_config, "cannot find Redis definitions for %s and fallback is disabled",
        module_name)

return nil
  end

-- Try global options
  opts = rspamd_config:get_all_opt('redis')

if opts then
    local ret

if opts[module_name] then
      ret = process_redis_options(opts[module_name], rspamd_config, result)

if ret then
        return maybe_return_cached(result)
      end
    else
      ret = process_redis_options(opts, rspamd_config, result)

-- Exclude disabled
      if opts['disabled_modules'] then
        for _, v in ipairs(opts['disabled_modules']) do
          if v == module_name then
            logger.infox(rspamd_config, "NOT using default redis server for module %s: it is disabled",
                module_name)

return nil
          end
        end
      end

if ret then
        logger.infox(rspamd_config, "use default Redis settings for %s",
            module_name)
        return maybe_return_cached(result)
      end
    end
  end

if result.read_servers then
    return maybe_return_cached(result)
  end

return nil
end

--[[[
-- @function lua_redis.parse_redis_server(module_name, module_opts, no_fallback)
-- Extracts Redis server settings from configuration
-- @param {string} module_name name of module to get settings for
-- @param {table} module_opts settings for module or `nil` to fetch them from configuration
-- @param {boolean} no_fallback should be `true` if global settings must not be used
-- @return {table} redis server settings
-- @example
-- local rconfig = lua_redis.parse_redis_server('my_module')
-- -- rconfig contains upstream_list objects in ['write_servers'] and ['read_servers']
-- -- ['timeout'] contains timeout in seconds
-- -- ['expand_keys'] if true tells that redis key expansion is enabled
--]]

exports.rspamd_parse_redis_server = rspamd_parse_redis_server
exports.parse_redis_server = rspamd_parse_redis_server

local process_cmd = {
  bitop = function(args)
    local idx_l = {}
    for i = 2, #args do
      table.insert(idx_l, i)
    end
    return idx_l
  end,
  blpop = function(args)
    local idx_l = {}
    for i = 1, #args - 1 do
      table.insert(idx_l, i)
    end
    return idx_l
  end,
  eval = function(args)
    local idx_l = {}
    local numkeys = args[2]
    if numkeys and tonumber(numkeys) >= 1 then
      for i = 3, numkeys + 2 do
        table.insert(idx_l, i)
      end
    end
    return idx_l
  end,
  set = function(args)
    return { 1 }
  end,
  mget = function(args)
    local idx_l = {}
    for i = 1, #args do
      table.insert(idx_l, i)
    end
    return idx_l
  end,
  mset = function(args)
    local idx_l = {}
    for i = 1, #args, 2 do
      table.insert(idx_l, i)
    end
    return idx_l
  end,
  sdiffstore = function(args)
    local idx_l = {}
    for i = 2, #args do
      table.insert(idx_l, i)
    end
    return idx_l
  end,
  smove = function(args)
    return { 1, 2 }
  end,
  script = function()
  end
}
process_cmd.append = process_cmd.set
process_cmd.auth = process_cmd.script
process_cmd.bgrewriteaof = process_cmd.script
process_cmd.bgsave = process_cmd.script
process_cmd.bitcount = process_cmd.set
process_cmd.bitfield = process_cmd.set
process_cmd.bitpos = process_cmd.set
process_cmd.brpop = process_cmd.blpop
process_cmd.brpoplpush = process_cmd.blpop
process_cmd.client = process_cmd.script
process_cmd.cluster = process_cmd.script
process_cmd.command = process_cmd.script
process_cmd.config = process_cmd.script
process_cmd.dbsize = process_cmd.script
process_cmd.debug = process_cmd.script
process_cmd.decr = process_cmd.set
process_cmd.decrby = process_cmd.set
process_cmd.del = process_cmd.mget
process_cmd.discard = process_cmd.script
process_cmd.dump = process_cmd.set
process_cmd.echo = process_cmd.script
process_cmd.evalsha = process_cmd.eval
process_cmd.exec = process_cmd.script
process_cmd.exists = process_cmd.mget
process_cmd.expire = process_cmd.set
process_cmd.expireat = process_cmd.set
process_cmd.flushall = process_cmd.script
process_cmd.flushdb = process_cmd.script
process_cmd.geoadd = process_cmd.set
process_cmd.geohash = process_cmd.set
process_cmd.geopos = process_cmd.set
process_cmd.geodist = process_cmd.set
process_cmd.georadius = process_cmd.set
process_cmd.georadiusbymember = process_cmd.set
process_cmd.get = process_cmd.set
process_cmd.getbit = process_cmd.set
process_cmd.getrange = process_cmd.set
process_cmd.getset = process_cmd.set
process_cmd.hdel = process_cmd.set
process_cmd.hexists = process_cmd.set
process_cmd.hget = process_cmd.set
process_cmd.hgetall = process_cmd.set
process_cmd.hincrby = process_cmd.set
process_cmd.hincrbyfloat = process_cmd.set
process_cmd.hkeys = process_cmd.set
process_cmd.hlen = process_cmd.set
process_cmd.hmget = process_cmd.set
process_cmd.hmset = process_cmd.set
process_cmd.hscan = process_cmd.set
process_cmd.hset = process_cmd.set
process_cmd.hsetnx = process_cmd.set
process_cmd.hstrlen = process_cmd.set
process_cmd.hvals = process_cmd.set
process_cmd.incr = process_cmd.set
process_cmd.incrby = process_cmd.set
process_cmd.incrbyfloat = process_cmd.set
process_cmd.info = process_cmd.script
process_cmd.keys = process_cmd.script
process_cmd.lastsave = process_cmd.script
process_cmd.lindex = process_cmd.set
process_cmd.linsert = process_cmd.set
process_cmd.llen = process_cmd.set
process_cmd.lpop = process_cmd.set
process_cmd.lpush = process_cmd.set
process_cmd.lpushx = process_cmd.set
process_cmd.lrange = process_cmd.set
process_cmd.lrem = process_cmd.set
process_cmd.lset = process_cmd.set
process_cmd.ltrim = process_cmd.set
process_cmd.migrate = process_cmd.script
process_cmd.monitor = process_cmd.script
process_cmd.move = process_cmd.set
process_cmd.msetnx = process_cmd.mset
process_cmd.multi = process_cmd.script
process_cmd.object = process_cmd.script
process_cmd.persist = process_cmd.set
process_cmd.pexpire = process_cmd.set
process_cmd.pexpireat = process_cmd.set
process_cmd.pfadd = process_cmd.set
process_cmd.pfcount = process_cmd.set
process_cmd.pfmerge = process_cmd.mget
process_cmd.ping = process_cmd.script
process_cmd.psetex = process_cmd.set
process_cmd.psubscribe = process_cmd.script
process_cmd.pubsub = process_cmd.script
process_cmd.pttl = process_cmd.set
process_cmd.publish = process_cmd.script
process_cmd.punsubscribe = process_cmd.script
process_cmd.quit = process_cmd.script
process_cmd.randomkey = process_cmd.script
process_cmd.readonly = process_cmd.script
process_cmd.readwrite = process_cmd.script
process_cmd.rename = process_cmd.mget
process_cmd.renamenx = process_cmd.mget
process_cmd.restore = process_cmd.set
process_cmd.role = process_cmd.script
process_cmd.rpop = process_cmd.set
process_cmd.rpoplpush = process_cmd.mget
process_cmd.rpush = process_cmd.set
process_cmd.rpushx = process_cmd.set
process_cmd.sadd = process_cmd.set
process_cmd.save = process_cmd.script
process_cmd.scard = process_cmd.set
process_cmd.sdiff = process_cmd.mget
process_cmd.select = process_cmd.script
process_cmd.setbit = process_cmd.set
process_cmd.setex = process_cmd.set
process_cmd.setnx = process_cmd.set
process_cmd.sinterstore = process_cmd.sdiff
process_cmd.sismember = process_cmd.set
process_cmd.slaveof = process_cmd.script
process_cmd.slowlog = process_cmd.script
process_cmd.smembers = process_cmd.script
process_cmd.sort = process_cmd.set
process_cmd.spop = process_cmd.set
process_cmd.srandmember = process_cmd.set
process_cmd.srem = process_cmd.set
process_cmd.strlen = process_cmd.set
process_cmd.subscribe = process_cmd.script
process_cmd.sunion = process_cmd.mget
process_cmd.sunionstore = process_cmd.mget
process_cmd.swapdb = process_cmd.script
process_cmd.sync = process_cmd.script
process_cmd.time = process_cmd.script
process_cmd.touch = process_cmd.mget
process_cmd.ttl = process_cmd.set
process_cmd.type = process_cmd.set
process_cmd.unsubscribe = process_cmd.script
process_cmd.unlink = process_cmd.mget
process_cmd.unwatch = process_cmd.script
process_cmd.wait = process_cmd.script
process_cmd.watch = process_cmd.mget
process_cmd.zadd = process_cmd.set
process_cmd.zcard = process_cmd.set
process_cmd.zcount = process_cmd.set
process_cmd.zincrby = process_cmd.set
process_cmd.zinterstore = process_cmd.eval
process_cmd.zlexcount = process_cmd.set
process_cmd.zrange = process_cmd.set
process_cmd.zrangebylex = process_cmd.set
process_cmd.zrank = process_cmd.set
process_cmd.zrem = process_cmd.set
process_cmd.zrembylex = process_cmd.set
process_cmd.zrembyrank = process_cmd.set
process_cmd.zrembyscore = process_cmd.set
process_cmd.zrevrange = process_cmd.set
process_cmd.zrevrangebyscore = process_cmd.set
process_cmd.zrevrank = process_cmd.set
process_cmd.zscore = process_cmd.set
process_cmd.zunionstore = process_cmd.eval
process_cmd.scan = process_cmd.script
process_cmd.sscan = process_cmd.set
process_cmd.hscan = process_cmd.set
process_cmd.zscan = process_cmd.set

local function get_key_indexes(cmd, args)
  local idx_l = {}
  cmd = string.lower(cmd)
  if process_cmd[cmd] then
    idx_l = process_cmd[cmd](args)
  else
    logger.warnx(rspamd_config, "Don't know how to extract keys for %s Redis command", cmd)
  end
  return idx_l
end

local gen_meta = {
  principal_recipient = function(task)
    return task:get_principal_recipient()
  end,
  principal_recipient_domain = function(task)
    local p = task:get_principal_recipient()
    if not p then
      return
    end
    return string.match(p, '.*@(.*)')
  end,
  ip = function(task)
    local i = task:get_ip()
    if i and i:is_valid() then
      return i:to_string()
    end
  end,
  from = function(task)
    return ((task:get_from('smtp') or E)[1] or E)['addr']
  end,
  from_domain = function(task)
    return ((task:get_from('smtp') or E)[1] or E)['domain']
  end,
  from_domain_or_helo_domain = function(task)
    local d = ((task:get_from('smtp') or E)[1] or E)['domain']
    if d and #d > 0 then
      return d
    end
    return task:get_helo()
  end,
  mime_from = function(task)
    return ((task:get_from('mime') or E)[1] or E)['addr']
  end,
  mime_from_domain = function(task)
    return ((task:get_from('mime') or E)[1] or E)['domain']
  end,
}

local function gen_get_esld(f)
  return function(task)
    local d = f(task)
    if not d then
      return
    end
    return rspamd_util.get_tld(d)
  end
end

gen_meta.smtp_from = gen_meta.from
gen_meta.smtp_from_domain = gen_meta.from_domain
gen_meta.smtp_from_domain_or_helo_domain = gen_meta.from_domain_or_helo_domain
gen_meta.esld_principal_recipient_domain = gen_get_esld(gen_meta.principal_recipient_domain)
gen_meta.esld_from_domain = gen_get_esld(gen_meta.from_domain)
gen_meta.esld_smtp_from_domain = gen_meta.esld_from_domain
gen_meta.esld_mime_from_domain = gen_get_esld(gen_meta.mime_from_domain)
gen_meta.esld_from_domain_or_helo_domain = gen_get_esld(gen_meta.from_domain_or_helo_domain)
gen_meta.esld_smtp_from_domain_or_helo_domain = gen_meta.esld_from_domain_or_helo_domain

local function get_key_expansion_metadata(task)

local md_mt = {
    __index = function(self, k)
      k = string.lower(k)
      local v = rawget(self, k)
      if v then
        return v
      end
      if gen_meta[k] then
        v = gen_meta[k](task)
        rawset(self, k, v)
      end
      return v
    end,
  }

local lazy_meta = {}
  setmetatable(lazy_meta, md_mt)
  return lazy_meta

end

-- Performs async call to redis hiding all complexity inside function
-- task - rspamd_task
-- redis_params - valid params returned by rspamd_parse_redis_server
-- key - key to select upstream or nil to select round-robin/master-slave
-- is_write - true if need to write to redis server
-- callback - function to be called upon request is completed
-- command - redis command
-- args - table of arguments
-- extra_opts - table of optional request arguments
local function rspamd_redis_make_request(task, redis_params, key, is_write,
                                         callback, command, args, extra_opts)
  local addr
  local function rspamd_redis_make_request_cb(err, data)
    if err then
      addr:fail()
    else
      addr:ok()
    end
    if callback then
      callback(err, data, addr)
    end
  end
  if not task or not redis_params or not command then
    return false, nil, nil
  end

local rspamd_redis = require "rspamd_redis"

if key then
    if is_write then
      addr = redis_params['write_servers']:get_upstream_by_hash(key)
    else
      addr = redis_params['read_servers']:get_upstream_by_hash(key)
    end
  else
    if is_write then
      addr = redis_params['write_servers']:get_upstream_master_slave(key)
    else
      addr = redis_params['read_servers']:get_upstream_round_robin(key)
    end
  end

if not addr then
    logger.errx(task,
        'cannot select redis server (all dead or pending DNS resolution)')
    return false, nil, nil
  end

if redis_params['expand_keys'] then
    local m = get_key_expansion_metadata(task)
    local indexes = get_key_indexes(command, args)
    for _, i in ipairs(indexes) do
      args[i] = lutil.template(args[i], m)
    end
  end

local ip_addr = addr:get_addr()
  local options = {
    task = task,
    callback = rspamd_redis_make_request_cb,
    host = ip_addr,
    timeout = redis_params['timeout'],
    cmd = command,
    args = args
  }

if extra_opts then
    for k, v in pairs(extra_opts) do
      options[k] = v
    end
  end

if redis_params['username'] then
    options['username'] = redis_params['username']
  end

if redis_params['password'] then
    options['password'] = redis_params['password']
  end

if redis_params['db'] then
    options['dbname'] = redis_params['db']
  end

-- TLS options
  if redis_params.ssl ~= nil then options.ssl = redis_params.ssl end
  if redis_params.no_ssl_verify ~= nil then options.no_ssl_verify = redis_params.no_ssl_verify end
  if redis_params.ssl_ca then options.ssl_ca = redis_params.ssl_ca end
  if redis_params.ssl_ca_dir then options.ssl_ca_dir = redis_params.ssl_ca_dir end
  if redis_params.ssl_cert then options.ssl_cert = redis_params.ssl_cert end
  if redis_params.ssl_key then options.ssl_key = redis_params.ssl_key end
  if redis_params.sni then options.sni = redis_params.sni end

lutil.debugm(N, task, 'perform request to redis server' ..
      ' (host=%s, timeout=%s): cmd: %s', ip_addr,
      options.timeout, options.cmd)

local ret, conn = rspamd_redis.make_request(options)

if not ret then
    addr:fail()
    logger.warnx(task, "cannot make redis request to: %s", tostring(ip_addr))
  end

return ret, conn, addr
end

--[[[
-- @function lua_redis.redis_make_request(task, redis_params, key, is_write, callback, command, args)
-- Sends a request to Redis
-- @param {rspamd_task} task task object
-- @param {table} redis_params redis configuration in format returned by lua_redis.parse_redis_server()
-- @param {string} key key to use for sharding
-- @param {boolean} is_write should be `true` if we are performing a write operating
-- @param {function} callback callback function (first parameter is error if applicable, second is a 2D array (table))
-- @param {string} command Redis command to run
-- @param {table} args Numerically indexed table containing arguments for command
--]]

exports.rspamd_redis_make_request = rspamd_redis_make_request
exports.redis_make_request = rspamd_redis_make_request

local function redis_make_request_taskless(ev_base, cfg, redis_params, key,
                                           is_write, callback, command, args, extra_opts)
  if not ev_base or not redis_params or not command then
    return false, nil, nil
  end

local addr
  local function rspamd_redis_make_request_cb(err, data)
    if err then
      addr:fail()
    else
      addr:ok()
    end
    if callback then
      callback(err, data, addr)
    end
  end

local rspamd_redis = require "rspamd_redis"

if not addr then
    logger.errx(cfg,
        'cannot select redis server (all dead or pending DNS resolution)')
    return false, nil, nil
  end

local options = {
    ev_base = ev_base,
    config = cfg,
    callback = rspamd_redis_make_request_cb,
    host = addr:get_addr(),
    timeout = redis_params['timeout'],
    cmd = command,
    args = args
  }
  if extra_opts then
    for k, v in pairs(extra_opts) do
      options[k] = v
    end
  end

if redis_params['username'] then
    options['username'] = redis_params['username']
  end

if redis_params['password'] then
    options['password'] = redis_params['password']
  end

if redis_params['db'] then
    options['dbname'] = redis_params['db']
  end

lutil.debugm(N, cfg, 'perform taskless request to redis server' ..
      ' (host=%s, timeout=%s): cmd: %s', options.host:tostring(true),
      options.timeout, options.cmd)
  local ret, conn = rspamd_redis.make_request(options)
  if not ret then
    logger.errx('cannot execute redis request')
    addr:fail()
  end

return ret, conn, addr
end

--[[[
-- @function lua_redis.redis_make_request_taskless(ev_base, cfg, redis_params, key, is_write, callback, command, args)
-- Sends a request to Redis in context where `task` is not available for some specific use-cases
-- Identical to redis_make_request() except in that first parameter is an `event base` object and the second one is the 'config' object
--]]

exports.rspamd_redis_make_request_taskless = redis_make_request_taskless
exports.redis_make_request_taskless = redis_make_request_taskless

local redis_scripts = {
}

local function script_set_loaded(script)
  if script.sha then
    script.loaded = true
  end

script.pending_upload = false -- Allow further reload

local wait_table = {}
  for _, s in ipairs(script.waitq) do
    table.insert(wait_table, s)
  end

script.waitq = {}

for _, s in ipairs(wait_table) do
    s(script.loaded)
  end
end

local function prepare_redis_call(script)
  local servers = {}
  local options = {}

if script.redis_params.read_servers then
    servers = lutil.table_merge(servers, script.redis_params.read_servers:all_upstreams())
  end
  if script.redis_params.write_servers then
    servers = lutil.table_merge(servers, script.redis_params.write_servers:all_upstreams())
  end

-- Call load script on each server, set loaded flag
  if not script.servers_ready then
    script.servers_ready = { } -- possible values for each server: unsent, tempfail, failed, done
  end

for idx, s in ipairs(servers) do
    local server_addr = s:get_addr()
    if not server_addr then
      -- Pending DNS resolution; mark as tempfail so the next attempt retries.
      -- Insert a placeholder opt (server_idx-aligned) so options/servers_ready
      -- indexing stays consistent for the consumer loop in load_script_task /
      -- load_script_taskless. The placeholder is detected via .skip = true.
      script.servers_ready[idx] = "tempfail"
      logger.infox(rspamd_config,
          'skipping SCRIPT LOAD for upstream %s: address not resolved yet',
          s:get_name())
      table.insert(options, { skip = true, upstream = s, server_idx = idx })
    else
      local cur_opts = {
        host = server_addr,
        timeout = script.redis_params['timeout'],
        cmd = 'SCRIPT',
        args = { 'LOAD', script.script },
        upstream = s,
        server_idx = idx,
      }

-- By default we start from unsent status
      if not script.servers_ready[idx] then
        script.servers_ready[idx] = "unsent"
      end

if script.redis_params['username'] then
        cur_opts['username'] = script.redis_params['username']
      end

if script.redis_params['password'] then
        cur_opts['password'] = script.redis_params['password']
      end

if script.redis_params['db'] then
        cur_opts['dbname'] = script.redis_params['db']
      end

table.insert(options, cur_opts)
    end
  end

return options
end

local function is_any_server_ready(script)
  for _, s in ipairs(script.servers_ready) do
    if s == "done" then
      return true
    end
  end
  return false
end

local function is_all_servers_failed(script)
  for _, s in ipairs(script.servers_ready) do
    if s == "unsent" or s == "tempfail" or s == "done" then
      return false
    end
  end

return true
end

local function script_description(script)
  return script.filename and ("from file: " .. script.filename)
      or ("with id: " .. script.id)
end

local function load_script_task(script, task, is_write)
  local rspamd_redis = require "rspamd_redis"
  local opts = prepare_redis_call(script)

for idx, opt in ipairs(opts) do
    -- opt.skip means the upstream was not resolved when prepare_redis_call ran;
    -- servers_ready is already tempfailed and will be retried on the next pass.
    if not opt.skip and script.servers_ready[idx] ~= 'done' then
      opt.task = task
      opt.is_write = is_write
      opt.callback = function(err, data)
        if err then
          if string.match(err, 'ERR') then
            logger.errx(task, 'cannot upload script %s to %s: %s; registered from: %s:%s',
                script_description(script),
                opt.upstream:get_addr():to_string(true),
                err, script.caller.short_src, script.caller.currentline)
            opt.upstream:fail()
            script.servers_ready[idx] = "failed"
            return
          else
            -- Assume temporary error
            logger.infox(task, 'temporary error uploading script %s to %s: %s; registered from: %s:%s',
                script_description(script),
                opt.upstream:get_addr():to_string(true),
                err, script.caller.short_src, script.caller.currentline)
            script.servers_ready[idx] = "tempfail"
            return
          end
        else
          opt.upstream:ok()
          logger.infox(task,
              "uploaded redis script to %s %s, sha: %s",
              opt.upstream:get_addr():to_string(true),
              script_description(script), data)
          script.sha = data -- We assume that sha is the same on all servers
          script.servers_ready[idx] = "done"
        end
        if is_any_server_ready(script) then
          script_set_loaded(script)
        elseif is_all_servers_failed(script) then
          script.pending_upload = false
          script.fatal_error = "cannot upload script to any server"
        end
      end -- callback

local ret = rspamd_redis.make_request(opt)

if not ret then
        logger.errx('cannot execute redis request to load script on %s',
            opt.upstream:get_addr())
        script.servers_ready[idx] = "failed"
        opt.upstream:fail()
      end
    end

if is_any_server_ready(script) then
      script_set_loaded(script)
    elseif is_all_servers_failed(script) then
      script.pending_upload = false
      script.fatal_error = "cannot upload script to any server"
    end
  end
end

local function load_script_taskless(script, cfg, ev_base, is_write)
  local rspamd_redis = require "rspamd_redis"
  local opts = prepare_redis_call(script)

for idx, opt in ipairs(opts) do
    if not opt.skip and script.servers_ready[idx] ~= 'done' then
      opt.config = cfg
      opt.ev_base = ev_base
      opt.is_write = is_write
      opt.callback = function(err, data)
        if err then
          if string.match(err, 'ERR') then
            logger.errx(cfg, 'cannot upload script %s to %s: %s; registered from: %s:%s',
                script_description(script),
                opt.upstream:get_addr():to_string(true),
                err, script.caller.short_src, script.caller.currentline)
            opt.upstream:fail()
            script.servers_ready[idx] = "failed"
          else
            -- Assume temporary error
            logger.infox(cfg, 'temporary error uploading script %s to %s: %s; registered from: %s:%s',
                script_description(script),
                opt.upstream:get_addr():to_string(true),
                err, script.caller.short_src, script.caller.currentline)
            script.servers_ready[idx] = "tempfail"
          end
        else
          opt.upstream:ok()
          logger.infox(cfg,
              "uploaded redis script %s to %s, sha: %s",
              script_description(script),
              opt.upstream:get_addr():to_string(true),
              data)
          script.sha = data -- We assume that sha is the same on all servers
          script.servers_ready[idx] = "done"
        end

if is_any_server_ready(script) then
          script_set_loaded(script)
        elseif is_all_servers_failed(script) then
          script.pending_upload = false
          script.fatal_error = "cannot upload script to any server"
        end
      end
      local ret = rspamd_redis.make_request(opt)

if not ret then
        logger.errx('cannot execute redis request to load script %s on %s',
            script_description(script),
            opt.upstream:get_addr())
        script.servers_ready[idx] = "failed"
        opt.upstream:fail()
      end
    end

local function load_redis_script(script, cfg, ev_base, _)
  if script.redis_params then
    load_script_taskless(script, cfg, ev_base)
  end
end

local function add_redis_script(script, redis_params, caller_level, maybe_filename)
  if not caller_level then
    caller_level = 2
  end
  local caller = debug.getinfo(caller_level) or debug.getinfo(caller_level - 1) or E

local new_script = {
    caller = caller,
    loaded = false,
    redis_params = redis_params,
    script = script,
    waitq = {}, -- callbacks pending for script being loaded
    id = #redis_scripts + 1,
    filename = maybe_filename,
  }

-- Register on load function
  rspamd_config:add_on_load(function(cfg, ev_base, worker)
    local mult = 0.0
    rspamd_config:add_periodic(ev_base, 0.0, function()
      if not new_script.sha and not new_script.fatal_error then
        load_redis_script(new_script, cfg, ev_base, worker)

-- Do not wait for more than 10 seconds to retry
        if mult < 10 then
          mult = mult + 1
        end
        return 1.0 * mult -- Check one more time in one second
      end

-- All loaded, we are good to go
      return false
    end, false)
  end)

table.insert(redis_scripts, new_script)

return #redis_scripts
end
exports.add_redis_script = add_redis_script

-- Loads a Redis script from a file, strips comments, and passes the content to
-- `add_redis_script` function.
--
-- @param filename The name of the file containing the Redis script.
-- @param redis_params The Redis parameters to use for this script.
-- @return The ID of the newly added Redis script.
--
local function load_redis_script_from_file(filename, redis_params, dir)
  local lua_util = require "lua_util"

if not dir then
    dir = rspamd_paths.LUALIBDIR
  end
  local path = filename
  if filename:sub(1, 1) ~= package.config:sub(1, 1) then
    -- Relative path
    path = lua_util.join_path(dir, "redis_scripts", filename)
  end
  -- Read file contents
  local file = io.open(path, "r")
  if not file then
    return nil, string.format("failed to open Redis script file: %s", path)
  end
  local script = file:read("*all")
  if not script then
    file:close()
    return nil, string.format("failed to read Redis script file: %s", path)
  end
  file:close()
  script = lua_util.strip_lua_comments(script)

return add_redis_script(script, redis_params, 3, filename)
end

exports.load_redis_script_from_file = load_redis_script_from_file

local function exec_redis_script(id, params, callback, keys, args)
  local redis_args = {}

if not redis_scripts[id] then
    logger.errx("cannot find registered script with id %s", id)
    return false
  end

local script = redis_scripts[id]

if script.fatal_error then
    callback(script.fatal_error, nil)
    return true
  end

if not script.redis_params then
    callback('no redis servers defined', nil)
    return true
  end

local function do_call(can_reload)
    local function redis_cb(err, data)
      if not err then
        callback(err, data)
      elseif string.match(err, 'NOSCRIPT') then
        -- Schedule restart if possible
        if can_reload then
          table.insert(script.waitq, do_call)
          if not script.pending_upload then
            logger.infox(params.task or rspamd_config,
                'redis script %s is not loaded (NOSCRIPT returned), scheduling reload',
                script_description(script))
            script.sha = nil
            script.loaded = nil
            script.pending_upload = true
	    -- We must initialize all servers as we don't know here which one failed
	    for i, _ in ipairs(script.servers_ready) do
	      script.servers_ready[i] = "unsent"
	    end
            -- Reload scripts if this has not been initiated yet
            if params.task then
              load_script_task(script, params.task)
            else
              load_script_taskless(script, rspamd_config, params.ev_base)
            end
          else
            logger.infox(params.task or rspamd_config,
                'redis script %s is not ready (NOSCRIPT returned), waiting it to be loaded',
                script_description(script))
          end
        else
          callback(err, data)
        end
      else
        callback(err, data)
      end
    end

if #redis_args == 0 then
      table.insert(redis_args, script.sha)
      table.insert(redis_args, tostring(#keys))
      for _, k in ipairs(keys) do
        table.insert(redis_args, k)
      end

if type(args) == 'table' then
        for _, a in ipairs(args) do
          table.insert(redis_args, a)
        end
      end
    end

local redis_command = 'EVALSHA'
    if not params.is_write and script.redis_params.redis_version and
        script.redis_params.redis_version >= 7 then
      redis_command = 'EVALSHA_RO'
    end
    if params.task then
      if not rspamd_redis_make_request(params.task, script.redis_params,
          params.key, params.is_write, redis_cb, redis_command, redis_args) then
        callback('Cannot make redis request', nil)
      end
    else
      if not redis_make_request_taskless(params.ev_base, rspamd_config,
          script.redis_params,
          params.key, params.is_write, redis_cb, redis_command, redis_args) then
        callback('Cannot make redis request', nil)
      end
    end
  end

if script.loaded and script.sha then
    do_call(true)
  else
    table.insert(script.waitq, do_call)
    if not script.servers_ready then
      -- Delayed until scripts are loaded
      logger.infox(params.task or rspamd_config, 'redis script %s is not loaded, trying to load',
          script_description(script))
      -- Reload scripts if this has not been initiated yet
      if params.task then
        load_script_task(script, params.task)
      else
        load_script_taskless(script, rspamd_config, params.ev_base)
      end
    else
      -- Already loaded, but not yet ready
      logger.infox(params.task or rspamd_config, 'redis script %s is not ready, waiting it to be loaded',
          script_description(script))
    end
  end

return true
end

exports.exec_redis_script = exec_redis_script

local function redis_connect_sync(redis_params, is_write, key, cfg, ev_base)
  if not redis_params then
    return false, nil
  end

local rspamd_redis = require "rspamd_redis"
  local addr

if not addr then
    logger.errx(cfg or rspamd_config,
        'cannot select redis server (all dead or pending DNS resolution)')
    return false, nil
  end

local options = {
    host = addr:get_addr(),
    timeout = redis_params['timeout'],
    config = cfg or rspamd_config,
    ev_base = ev_base or rspamadm_ev_base,
    session = redis_params.session or rspamadm_session
  }

for k, v in pairs(redis_params) do
    options[k] = v
  end

if not options.config then
    logger.errx('config is not set')
    return false, nil, addr
  end

if not options.ev_base then
    logger.errx('ev_base is not set')
    return false, nil, addr
  end

if not options.session then
    logger.errx('session is not set')
    return false, nil, addr
  end

local ret, conn = rspamd_redis.connect_sync(options)
  if not ret then
    logger.errx('cannot create redis connection: %s', conn)
    addr:fail()

return false, nil, addr
  end

if conn then
    local need_exec = false
    if redis_params['username'] then
      if redis_params['password'] then
        conn:add_cmd('AUTH', { redis_params['username'], redis_params['password'] })
        need_exec = true
      else
        logger.warnx('Redis requires a password when username is supplied')
        return false, nil, addr
      end
    elseif redis_params['password'] then
      conn:add_cmd('AUTH', { redis_params['password'] })
      need_exec = true
    end

if redis_params['db'] then
      conn:add_cmd('SELECT', { tostring(redis_params['db']) })
      need_exec = true
    elseif redis_params['dbname'] then
      conn:add_cmd('SELECT', { tostring(redis_params['dbname']) })
      need_exec = true
    end

if need_exec then
      local exec_ret, res = conn:exec()

if not exec_ret then
        logger.errx('cannot prepare redis connection (authentication or db selection failure): %s',
            res)
        addr:fail()
        return false, nil, addr
      end
    end
  end

return ret, conn, addr
end

exports.redis_connect_sync = redis_connect_sync

--[[[
-- @function lua_redis.request(redis_params, attrs, req)
-- Sends a request to Redis synchronously with coroutines or asynchronously using
-- a callback (modern API)
-- @param redis_params a table of redis server parameters
-- @param attrs a table of redis request attributes (e.g. task, or ev_base + cfg + session)
-- @param req a table of request: a command + command options
-- @return {result,data/connection,address} boolean result, connection object in case of async request and results if using coroutines, redis server address
--]]

exports.request = function(redis_params, attrs, req)
  local lua_util = require "lua_util"

if not attrs or not redis_params or not req then
    logger.errx('invalid arguments for redis request')
    return false, nil, nil
  end

if not (attrs.task or (attrs.config and attrs.ev_base)) then
    logger.errx('invalid attributes for redis request')
    return false, nil, nil
  end

local opts = lua_util.shallowcopy(attrs)

local log_obj = opts.task or opts.config

local addr

if opts.callback then
    -- Wrap callback
    local callback = opts.callback
    local function rspamd_redis_make_request_cb(err, data)
      if err then
        addr:fail()
      else
        addr:ok()
      end
      callback(err, data, addr)
    end
    opts.callback = rspamd_redis_make_request_cb
  end

local rspamd_redis = require "rspamd_redis"
  local is_write = opts.is_write

if opts.key then
    if is_write then
      addr = redis_params['write_servers']:get_upstream_by_hash(attrs.key)
    else
      addr = redis_params['read_servers']:get_upstream_by_hash(attrs.key)
    end
  else
    if is_write then
      addr = redis_params['write_servers']:get_upstream_master_slave(attrs.key)
    else
      addr = redis_params['read_servers']:get_upstream_round_robin(attrs.key)
    end
  end

if not addr then
    logger.errx(log_obj,
        'cannot select redis server (all dead or pending DNS resolution)')
    return false, nil, nil
  end

opts.host = addr:get_addr()
  opts.timeout = redis_params.timeout

if type(req) == 'string' then
    opts.cmd = req
  else
    -- XXX: modifies the input table
    opts.cmd = table.remove(req, 1);
    opts.args = req
  end

if redis_params.username then
    opts.username = redis_params.username
  end

if redis_params.password then
    opts.password = redis_params.password
  end

if redis_params.db then
    opts.dbname = redis_params.db
  end

-- TLS options
  if redis_params.ssl ~= nil then opts.ssl = redis_params.ssl end
  if redis_params.no_ssl_verify ~= nil then opts.no_ssl_verify = redis_params.no_ssl_verify end
  if redis_params.ssl_ca then opts.ssl_ca = redis_params.ssl_ca end
  if redis_params.ssl_ca_dir then opts.ssl_ca_dir = redis_params.ssl_ca_dir end
  if redis_params.ssl_cert then opts.ssl_cert = redis_params.ssl_cert end
  if redis_params.ssl_key then opts.ssl_key = redis_params.ssl_key end
  if redis_params.sni then opts.sni = redis_params.sni end

if opts.callback then
    lutil.debugm(N, 'perform generic async request to redis server' ..
        ' (host=%s, timeout=%s): cmd: %s, arguments: %s', addr,
        opts.timeout, opts.cmd, opts.args)
    local ret, conn = rspamd_redis.make_request(opts)
    if not ret then
      logger.errx(log_obj, 'cannot execute redis request')
      addr:fail()
    end

return ret, conn, addr
  else
    -- Coroutines version
    lutil.debugm(N, 'perform generic coroutine request to redis server' ..
        ' (host=%s, timeout=%s): cmd: %s, arguments: %s', addr,
        opts.timeout, opts.cmd, opts.args)
    local ret, conn = rspamd_redis.connect_sync(opts)
    if not ret then
      logger.errx(log_obj, 'cannot execute redis request')
      addr:fail()
    else
      conn:add_cmd(opts.cmd, opts.args)
      return conn:exec()
    end
    return false, nil, addr
  end
end

--[[[
-- @function lua_redis.connect(redis_params, attrs)
-- Connects to Redis synchronously with coroutines or asynchronously using a callback (modern API)
-- @param redis_params a table of redis server parameters
-- @param attrs a table of redis request attributes (e.g. task, or ev_base + cfg + session)
-- @return {result,connection,address} boolean result, connection object, redis server address
--]]

exports.connect = function(redis_params, attrs)
  local lua_util = require "lua_util"

if not attrs or not redis_params then
    logger.errx('invalid arguments for redis connect')
    return false, nil, nil
  end

if not (attrs.task or (attrs.config and attrs.ev_base)) then
    logger.errx('invalid attributes for redis connect')
    return false, nil, nil
  end

local opts = lua_util.shallowcopy(attrs)

local log_obj = opts.task or opts.config

local addr

local rspamd_redis = require "rspamd_redis"
  local is_write = opts.is_write

if not addr then
    logger.errx(log_obj,
        'cannot select redis server (all dead or pending DNS resolution)')
    return false, nil, nil
  end

opts.host = addr:get_addr()
  opts.timeout = redis_params.timeout

if redis_params.username then
    opts.username = redis_params.username
  end

if redis_params.password then
    opts.password = redis_params.password
  end

if redis_params.db then
    opts.dbname = redis_params.db
  end

if opts.callback then
    local ret, conn = rspamd_redis.connect(opts)
    if not ret then
      logger.errx(log_obj, 'cannot execute redis connect')
      addr:fail()
    end

return ret, conn, addr
  else
    -- Coroutines version
    local ret, conn = rspamd_redis.connect_sync(opts)
    if not ret then
      logger.errx(log_obj, 'cannot execute redis connect')
      addr:fail()
    else
      return true, conn, addr
    end

return false, nil, addr
  end
end

local redis_prefixes = {}

--[[[
-- @function lua_redis.register_prefix(prefix, module, description[, optional])
-- Register new redis prefix for documentation purposes
-- @param {string} prefix string prefix
-- @param {string} module module name
-- @param {string} description prefix description
-- @param {table} optional optional kv pairs (e.g. pattern)
--]]
local function register_prefix(prefix, module, description, optional)
  local pr = {
    module = module,
    description = description
  }

if optional and type(optional) == 'table' then
    for k, v in pairs(optional) do
      pr[k] = v
    end
  end

redis_prefixes[prefix] = pr
end

exports.register_prefix = register_prefix

--[[[
-- @function lua_redis.prefixes([mname])
-- Returns prefixes for specific module (or all prefixes). Returns a table prefix -> table
--]]
exports.prefixes = function(mname)
  if not mname then
    return redis_prefixes
  else
    local fun = require "fun"

return fun.totable(fun.filter(function(_, data)
      return data.module == mname
    end,
        redis_prefixes))
  end
end

-- Build new read/write upstream lists from a flattened sentinel topology
-- (table of {[master_name] = { ip, port, slaves = {...} }}) and replace
-- params.read_servers / params.write_servers in-place. Used by the
-- synchronous rspamadm path; the async sentinel watcher in worker context
-- has its own variant that also installs failure-monitor watchers.
local function apply_sentinel_topology(params, masters)
  local read_servers_tbl, write_servers_tbl = {}, {}

for _, master in pairs(masters) do
    write_servers_tbl[#write_servers_tbl + 1] = string.format(
        '%s:%s', master.ip, master.port)
    read_servers_tbl[#read_servers_tbl + 1] = string.format(
        '%s:%s', master.ip, master.port)

for _, slave in ipairs(master.slaves or E) do
      if slave['master-link-status'] == 'ok' then
        read_servers_tbl[#read_servers_tbl + 1] = string.format(
            '%s:%s', slave.ip, slave.port)
      end
    end
  end

if #write_servers_tbl == 0 then
    return false, 'no masters discovered from sentinel'
  end

table.sort(read_servers_tbl)
  table.sort(write_servers_tbl)

local read_servers_str = table.concat(read_servers_tbl, ',')
  local write_servers_str = table.concat(write_servers_tbl, ',')

local upstream_list = require "rspamd_upstream_list"

if read_servers_str ~= params.read_servers_str then
    local read_upstreams = upstream_list.create(rspamd_config,
        read_servers_str, 6379)
    if read_upstreams then
      logger.infox(rspamd_config,
          'sentinel: replace read servers with new list: %s', read_servers_str)
      params.read_servers = read_upstreams
      params.read_servers_str = read_servers_str
    else
      return false, string.format('cannot parse read servers list: %s', read_servers_str)
    end
  end

if write_servers_str ~= params.write_servers_str then
    local write_upstreams = upstream_list.create(rspamd_config,
        write_servers_str, 6379)
    if write_upstreams then
      logger.infox(rspamd_config,
          'sentinel: replace write servers with new list: %s', write_servers_str)
      params.write_servers = write_upstreams
      params.write_servers_str = write_servers_str
    else
      return false, string.format('cannot parse write servers list: %s', write_servers_str)
    end
  end

return true
end

-- Synchronously query Redis Sentinel via rspamd_redis.connect_sync to populate
-- masters and slaves, then rewrite redis_params via apply_sentinel_topology.
-- Walks the configured sentinel upstreams in round-robin order until one
-- answers; only fails after exhausting them all.
local function redis_resolve_sentinels_sync(redis_params, attrs)
  if not redis_params.sentinels then
    return true
  end

local rspamd_redis = require "rspamd_redis"
  local sentinels = redis_params.sentinels
  local last_err = 'no sentinel upstreams to query'

-- Try each sentinel until one succeeds; same upstream may be picked twice
  -- in a row depending on its weight, so we just bound by upstream count.
  local nsentinels = #(sentinels:all_upstreams() or E)
  if nsentinels == 0 then
    return false, 'sentinel upstream list is empty'
  end

for _ = 1, nsentinels do
    local addr = sentinels:get_upstream_round_robin()
    if not addr then
      return false, last_err
    end

local sentinel_addr = addr:get_addr()
    if not sentinel_addr then
      addr:fail()
      last_err = 'sentinel address not resolved'
    else
      local options = {
        host = sentinel_addr,
        timeout = attrs.timeout or redis_params.timeout or 1.0,
        config = attrs.config or rspamd_config,
        ev_base = attrs.ev_base,
        session = attrs.session,
        no_pool = true,
      }
      if redis_params.sentinel_username then
        options.username = redis_params.sentinel_username
      end
      if redis_params.sentinel_password then
        options.password = redis_params.sentinel_password
      end

local masters, err = nil, nil
      local conn_ok, conn = rspamd_redis.connect_sync(options)
      if not conn_ok then
        err = string.format('cannot connect sentinel %s: %s',
            sentinel_addr:to_string(true), tostring(conn))
      else
        conn:add_cmd('SENTINEL', { 'masters' })
        local exec_ok, masters_result = conn:exec()
        if not exec_ok or type(masters_result) ~= 'table' then
          err = string.format('SENTINEL masters failed on %s: %s',
              sentinel_addr:to_string(true), tostring(masters_result))
        else
          masters = {}
          for _, m in ipairs(masters_result) do
            local master = flatten_redis_table(m)
            if master.ip and master.ip:match(':') then
              master.ip = '[' .. master.ip .. ']'
            end
            if redis_params.sentinel_masters_pattern then
              if master.name and master.name:match(redis_params.sentinel_masters_pattern) then
                masters[master.name] = master
              else
                lutil.debugm(N, rspamd_config,
                    'skip master %s with ip %s and port %s, pattern %s',
                    master.name, master.ip, master.port,
                    redis_params.sentinel_masters_pattern)
              end
            else
              masters[master.name] = master
            end
          end
        end
      end

if masters then
        local all_ok = true
        for name, master in pairs(masters) do
          master.slaves = {}
          local sok, sconn = rspamd_redis.connect_sync(options)
          if not sok then
            err = string.format('cannot connect sentinel for slaves on %s: %s',
                sentinel_addr:to_string(true), tostring(sconn))
            all_ok = false
            break
          end
          sconn:add_cmd('SENTINEL', { 'slaves', name })
          local s_eok, s_result = sconn:exec()
          if not s_eok or type(s_result) ~= 'table' then
            err = string.format('SENTINEL slaves %s failed on %s: %s',
                name, sentinel_addr:to_string(true), tostring(s_result))
            all_ok = false
            break
          end
          for _, s in ipairs(s_result) do
            local slave = flatten_redis_table(s)
            if slave.ip and slave.ip:match(':') then
              slave.ip = '[' .. slave.ip .. ']'
            end
            master.slaves[#master.slaves + 1] = slave
          end
        end
        if all_ok then
          addr:ok()
          local ok, apply_err = apply_sentinel_topology(redis_params, masters)
          if ok then
            return true
          end
          return false, apply_err
        end
      end

addr:fail()
      last_err = err or 'unknown sentinel failure'
    end
  end

return false, last_err
end

-- Synchronously upload registered Redis scripts associated with redis_params.
-- 'script_filter' is the value of opts.scripts: true (all matching scripts),
-- false (skip), or an array of script ids returned from add_redis_script.
local function redis_load_scripts_sync(redis_params, attrs, script_filter)
  if script_filter == false then
    return true
  end

local id_filter
  if type(script_filter) == 'table' then
    id_filter = {}
    for _, id in ipairs(script_filter) do
      id_filter[id] = true
    end
  end

local rspamd_redis = require "rspamd_redis"
  local errors = {}
  local matched = 0

for _, script in ipairs(redis_scripts) do
    if script.redis_params == redis_params and not script.fatal_error then
      if not id_filter or id_filter[script.id] then
        matched = matched + 1
        local opts = prepare_redis_call(script)

for idx, opt in ipairs(opts) do
          if not opt.skip and script.servers_ready[idx] ~= 'done' then
            local options = {
              host = opt.host,
              timeout = attrs.timeout or script.redis_params.timeout or 1.0,
              config = attrs.config or rspamd_config,
              ev_base = attrs.ev_base,
              session = attrs.session,
            }
            if opt.username then options.username = opt.username end
            if opt.password then options.password = opt.password end
            if opt.dbname then options.dbname = opt.dbname end

local ok, conn = rspamd_redis.connect_sync(options)
            if not ok then
              opt.upstream:fail()
              script.servers_ready[idx] = 'failed'
              errors[#errors + 1] = string.format(
                  'cannot connect %s for SCRIPT LOAD %s: %s',
                  opt.upstream:get_addr():to_string(true),
                  script_description(script), tostring(conn))
            else
              conn:add_cmd('SCRIPT', { 'LOAD', script.script })
              local eok, sha = conn:exec()
              if not eok then
                opt.upstream:fail()
                script.servers_ready[idx] = 'failed'
                errors[#errors + 1] = string.format(
                    'SCRIPT LOAD %s failed on %s: %s',
                    script_description(script),
                    opt.upstream:get_addr():to_string(true), tostring(sha))
              else
                opt.upstream:ok()
                script.sha = sha
                script.servers_ready[idx] = 'done'
                logger.infox(rspamd_config,
                    'uploaded redis script %s to %s, sha: %s',
                    script_description(script),
                    opt.upstream:get_addr():to_string(true), sha)
              end
            end
          end
        end

if is_any_server_ready(script) then
          script_set_loaded(script)
        elseif is_all_servers_failed(script) then
          script.pending_upload = false
          script.fatal_error = 'cannot upload script to any server'
        end
      end
    end
  end

if id_filter and matched == 0 then
    return false, 'no scripts matched the requested ids'
  end

if #errors > 0 then
    -- Soft-fail: report errors but only fail if no script ended up loaded.
    local any_loaded = false
    for _, script in ipairs(redis_scripts) do
      if script.redis_params == redis_params and script.loaded then
        any_loaded = true
        break
      end
    end
    if not any_loaded and matched > 0 then
      return false, table.concat(errors, '; ')
    end
  end

return true
end

local default_setup_opts = {
  sentinels = true,
  scripts = true,
  timeout = nil,
}

--[[[
-- @function lua_redis.prepare_redis_setup(redis_params[, opts], callback)
-- One-shot synchronous initialization of a Redis connection for contexts
-- where `rspamd_config:add_on_load` callbacks never fire (typically rspamadm
-- tools). Performs, per `opts`, sentinel resolution and script loading,
-- then invokes `callback(err)` once Redis is ready.
--
-- @param {table} redis_params parsed redis params (from `parse_redis_server`).
-- @param {table} opts (optional) merged via `lua_util.override_defaults` over:
--   ```
--   {
--     sentinels = true,    -- resolve Sentinels if redis_params.sentinels is set
--     scripts   = true,    -- true = all scripts bound to redis_params,
--                          -- false = skip script loading,
--                          -- { id1, id2, ... } = only those add_redis_script ids
--     timeout   = nil,     -- override timeout (seconds) for setup IO
--     ev_base   = nil,     -- defaults to rspamadm_ev_base
--     session   = nil,     -- defaults to rspamadm_session
--     config    = nil,     -- defaults to rspamd_config
--   }
--   ```
-- @param {function} callback `callback(err)` — `err` is `nil` on success or
--   a string describing the failure.
--]]
exports.prepare_redis_setup = function(redis_params, opts, callback)
  if type(opts) == 'function' and callback == nil then
    callback = opts
    opts = nil
  end

if type(callback) ~= 'function' then
    error('lua_redis.prepare_redis_setup: callback must be a function')
  end

if not redis_params or type(redis_params) ~= 'table' then
    return callback('redis_params is nil or not a table')
  end

opts = lutil.override_defaults(default_setup_opts, opts or {})

local attrs = {
    config = opts.config or rspamd_config,
    ev_base = opts.ev_base or rawget(_G, 'rspamadm_ev_base'),
    session = opts.session or rawget(_G, 'rspamadm_session'),
    timeout = opts.timeout,
  }

if not attrs.ev_base then
    return callback('ev_base is not available; pass opts.ev_base or run from rspamadm')
  end
  if not attrs.session then
    return callback('async session is not available; pass opts.session or run from rspamadm')
  end

if opts.sentinels and redis_params.sentinels then
    local ok, err = redis_resolve_sentinels_sync(redis_params, attrs)
    if not ok then
      return callback(err)
    end
  end

if opts.scripts ~= false then
    local ok, err = redis_load_scripts_sync(redis_params, attrs, opts.scripts)
    if not ok then
      return callback(err)
    end
  end

return callback(nil)
end

return exports

Cancel