"""Redis 命令安全验证器 — 区分只读命令和危险命令。 只读命令白名单（查询模式允许）: - 字符串: GET, MGET, STRLEN, GETRANGE, GETBIT, SUBSTR - 哈希: HGET, HGETALL, HKEYS, HVALS, HLEN, HEXISTS, HMGET, HSCAN, HRANDFIELD - 列表: LRANGE, LLEN, LINDEX, LPOS - 集合: SMEMBERS, SCARD, SISMEMBER, SRANDMEMBER, SUNION, SINTER, SDIFF, SSCAN, SMISSING - 有序集合: ZRANGE, ZREVRANGE, ZRANGEBYSCORE, ZREVRANGEBYSCORE, ZCARD, ZSCORE, ZRANK, ZREVRANK, ZCOUNT, ZLEXCOUNT, ZSCAN, ZRANGEBYLEX, ZMPOP - 流: XREAD, XREVRANGE, XRANGE, XLEN, XINFO - 通用: EXISTS, TTL, PTTL, TYPE, DBSIZE, SCAN, INFO, CONFIG GET, COMMAND INFO, COMMAND DOCS, MEMORY USAGE, OBJECT, CLIENT LIST, CLIENT INFO, SLOWLOG GET, TIME, ECHO, PING, LOLWUT - Geo: GEOPOS, GEODIST, GEORADIUS, GEORADIUSBYMEMBER, GEOSEARCH, GEOHASH - Bitmap: BITCOUNT, BITPOS, GETBIT - HyperLogLog: PFCOUNT 危险命令（查询模式禁止）: - 写: SET, MSET, HSET, LPUSH, RPUSH, SADD, ZADD, DEL, FLUSHDB, FLUSHALL - 管理: SHUTDOWN, CONFIG SET, DEBUG, SAVE, BGSAVE, BGREWRITEAOF, ACL, REPLICAOF, SLAVEOF - 执行: EVAL, EVALSHA, FCALL, FCALL_RO """ from typing import Set, Tuple class RedisCommandValidator: """Redis 命令安全验证器。 通过 READONLY_COMMANDS 白名单和 DANGEROUS_COMMANDS 黑名单 对 Redis 命令进行双重校验，确保查询模式下只能执行安全的只读操作。 """ # ── 只读命令白名单 ────────────────────────────────────────── READONLY_COMMANDS: Set[str] = { # 字符串操作 "GET", "MGET", "STRLEN", "GETRANGE", "GETBIT", "SUBSTR", # 哈希操作 "HGET", "HGETALL", "HKEYS", "HVALS", "HLEN", "HEXISTS", "HMGET", "HSCAN", "HRANDFIELD", # 列表操作 "LRANGE", "LLEN", "LINDEX", "LPOS", # 集合操作 "SMEMBERS", "SCARD", "SISMEMBER", "SRANDMEMBER", "SUNION", "SINTER", "SDIFF", "SSCAN", "SMISMEMBER", # 有序集合操作 "ZRANGE", "ZREVRANGE", "ZRANGEBYSCORE", "ZREVRANGEBYSCORE", "ZCARD", "ZSCORE", "ZRANK", "ZREVRANK", "ZCOUNT", "ZLEXCOUNT", "ZSCAN", "ZRANGEBYLEX", # 流操作 "XREAD", "XREVRANGE", "XRANGE", "XLEN", "XINFO", # 通用命令 "EXISTS", "TTL", "PTTL", "TYPE", "DBSIZE", "SCAN", "INFO", "TIME", "ECHO", "PING", "LOLWUT", "COMMAND", "MEMORY", "OBJECT", "CLIENT", "SLOWLOG", # Geo 空间 "GEOPOS", "GEODIST", "GEORADIUS", "GEORADIUSBYMEMBER", "GEOSEARCH", "GEOHASH", "GEOSEARCHSTORE", # Bitmap "BITCOUNT", "BITPOS", # HyperLogLog "PFCOUNT", } # ── 危险命令黑名单 ────────────────────────────────────────── DANGEROUS_COMMANDS: Set[str] = { # 写操作 "SET", "MSET", "MSETNX", "SETEX", "PSETEX", "SETNX", "GETSET", "INCR", "INCRBY", "INCRBYFLOAT", "DECR", "DECRBY", "APPEND", "SETBIT", "SETRANGE", "HSET", "HMSET", "HSETNX", "HDEL", "HINCRBY", "HINCRBYFLOAT", "LPUSH", "RPUSH", "LPUSHX", "RPUSHX", "LPOP", "RPOP", "BLPOP", "BRPOP", "LREM", "LSET", "LTRIM", "RPOPLPUSH", "BRPOPLPUSH", "SADD", "SREM", "SPOP", "SMOVE", "SDIFFSTORE", "SINTERSTORE", "SUNIONSTORE", "ZADD", "ZREM", "ZINCRBY", "ZREMRANGEBYRANK", "ZREMRANGEBYSCORE", "ZREMRANGEBYLEX", "ZUNIONSTORE", "ZINTERSTORE", "ZDIFFSTORE", "ZMPOP", "ZPOPMAX", "ZPOPMIN", "BZPOPMAX", "BZPOPMIN", "DEL", "UNLINK", "FLUSHDB", "FLUSHALL", "RENAME", "RENAMENX", "MOVE", # 管理命令 "SHUTDOWN", "CONFIG", "DEBUG", "SAVE", "BGSAVE", "BGREWRITEAOF", "ACL", "REPLICAOF", "SLAVEOF", "SYNC", "PSYNC", "MONITOR", "KEYS", "MIGRATE", "RESTORE", # 脚本执行 "EVAL", "EVALSHA", "FCALL", "FCALL_RO", "SCRIPT", # 事务 "MULTI", "EXEC", "DISCARD", "WATCH", "UNWATCH", # 发布/订阅 "PUBLISH", "SUBSCRIBE", "UNSUBSCRIBE", "PSUBSCRIBE", "PUNSUBSCRIBE", # 模块 "MODULE", } # ── 验证方法 ──────────────────────────────────────────────── @classmethod def validate(cls, command: str) -> Tuple[bool, str]: """验证 Redis 命令是否只读安全。 验证逻辑: 1. 提取命令关键字（首个单词，转大写）。 2. 若在 DANGEROUS_COMMANDS 中，直接拒绝。 3. 若在 READONLY_COMMANDS 中，允许执行。 4. 否则视为未知命令，拒绝执行。 Args: command: 完整的 Redis 命令字符串，如 "GET mykey" 或 "HGETALL users"。 Returns: (is_safe, error_message): - is_safe: True 表示命令安全，可执行。 - error_message: 空字符串表示无错误，否则包含拒绝原因。 """ if not command or not command.strip(): return False, "命令不能为空" cmd = command.strip().split()[0].upper() if cmd in cls.DANGEROUS_COMMANDS: return False, f"Redis 命令 '{cmd}' 为危险命令，禁止在查询模式下执行" if cmd not in cls.READONLY_COMMANDS: return False, f"Redis 命令 '{cmd}' 不在查询白名单中，禁止执行" return True, "" @classmethod def is_dangerous(cls, command: str) -> bool: """判断命令是否为危险命令。 Args: command: 完整的 Redis 命令字符串。 Returns: True 表示命令属于危险命令。 """ if not command or not command.strip(): return False cmd = command.strip().split()[0].upper() return cmd in cls.DANGEROUS_COMMANDS @classmethod def is_readonly(cls, command: str) -> bool: """判断命令是否为只读命令。 Args: command: 完整的 Redis 命令字符串。 Returns: True 表示命令在白名单中且安全。 """ is_safe, _ = cls.validate(command) return is_safe