如何动态存储并执行公式与函数以处理多类型传感器的字节数据

方案1：声明式配置规则引擎（首推） #

这是兼顾安全性、扩展性、性能的最优方案，完全不需要硬编码单类传感器逻辑，也不需要任何动态代码执行。

实现思路 #

你可以基于Django ORM直接存储解析规则，预先定义所有业务需要的原子操作枚举，新增传感器只需要在后台配置规则无需改代码：

1. 建两张表存储规则：

• SensorType表：存储deviceID、struct解析格式、关联字段规则
• SensorField表：关联SensorType，每个记录对应一个输出字段，包含字段名、原子操作类型、操作参数

2. 预先定义原子操作枚举（可随时扩展）：

• RAW：直接取values指定下标的原始值，参数为下标
• MATH_CALC：数学运算，参数为(系数, 偏移量) 对应公式 a * values[n] + b，可扩展支持多值运算
• BOOL_CHECK：布尔判断，参数为(下标, 比较运算符, 对比值)，例 (7, "==", 0x80)
• STR_JOIN：字符串拼接，参数为(下标列表, 分隔符)，例 ([2], ".")

3. 写通用规则执行函数，按照配置的原子操作依次处理解析出来的values列表即可。

适配你的最小代码示例 #

import struct
from enum import Enum

# 预先定义的原子操作类型
class OpType(Enum):
    RAW = 1
    MATH_CALC = 2
    BOOL_CHECK = 3
    STR_JOIN =4

# 模拟从数据库查询到的deviceID对应规则，实际场景存在Django表中
DEVICE_RULES = {
    "421056758725": { # 示例deviceID，对应你测试用例的前6字节hex
        "struct_format": ">BBHBBhBHhHL",
        "fields": [
            {"name": "field1", "op": OpType.RAW, "params": 0},
            {"name": "version", "op": OpType.STR_JOIN, "params": ([2], ".")},
            {"name": "temp", "op": OpType.MATH_CALC, "params": (4, 2, -113)}, # 下标、系数、偏移：2*values[4]-113
            {"name": "is_alarm", "op": OpType.BOOL_CHECK, "params": (7, "==", 0x80)}
        ]
    }
}

def decode(input):
    val = bytearray(input)
    deviceID = val[0:6].hex()
    del(val[0:6])
    # 查规则
    rule = DEVICE_RULES.get(deviceID)
    if not rule:
        return {}
    values = list(struct.unpack(rule["struct_format"], val))
    # 按规则处理字段
    result = {"deviceID": deviceID}
    for field in rule["fields"]:
        op = field["op"]
        params = field["params"]
        if op == OpType.RAW:
            result[field["name"]] = values[params]
        elif op == OpType.MATH_CALC:
            idx, a, b = params
            result[field["name"]] = a * values[idx] + b
        elif op == OpType.BOOL_CHECK:
            idx, opt, target = params
            if opt == "==":
                result[field["name"]] = values[idx] == target
            # 可扩展其他比较符> < !=等
        elif op == OpType.STR_JOIN:
            idxs, sep = params
            result[field["name"]] = sep.join(str(values[i]) for i in idxs)
    return result

print(decode(b'B\x10Vu\x87%\x00x\r\x0f\x04\x01\x00\x00\x00\x00\x1e\x00\x00\x00\x00ad;l'))

方案2：受限AST表达式解析（适合有复杂灵活规则的场景） #

如果你的业务确实有原子操作覆盖不到的复杂逻辑，可以自己实现基于AST的安全表达式解析，完全规避eval的安全风险：

1. 只允许白名单内的语法节点：比如仅允许访问values下标、数字运算、比较运算、字符串拼接、常量，禁止所有函数调用、赋值、导入等危险操作
2. 规则直接存表达式字符串，执行前先过AST校验，只有符合白名单的表达式才会执行
3. 性能接近原生代码，安全性完全可控

示例安全表达式执行函数：

import ast

ALLOWED_NODE_TYPES = {
    ast.Module, ast.Expression, ast.Constant, ast.BinOp, ast.Compare,
    ast.Subscript, ast.Name, ast.Load, ast.List, ast.Str, ast.Num,
    ast.Eq, ast.NotEq, ast.Gt, ast.Lt, ast.Add, ast.Sub, ast.Mult, ast.Div
}

def safe_eval(expr, variables):
    # 解析AST
    tree = ast.parse(expr, mode='eval')
    # 校验所有节点都在白名单内
    for node in ast.walk(tree):
        if type(node) not in ALLOWED_NODE_TYPES:
            raise ValueError(f"非法语法: {type(node).__name__}")
    # 执行表达式
    return eval(compile(tree, filename='', mode='eval'), {"__builtins__": {}}, variables)

# 用法示例，表达式直接存在数据库中
expr = '".".join([str(values[2])])'
# 执行前先校验，不会有注入风险
result = safe_eval(expr, {"values": values})

若非要用原生eval的安全限制 #

如果一定要用原生eval，必须严格限制执行上下文：

• globals参数设为{"__builtins__": {}}，完全禁用内置函数
• locals参数只传入你允许访问的变量，比如只有values和你开放的少量工具函数
• 所有传入的表达式必须是内部管理员配置，不接受任何用户提交的表达式
  该方案仍存在潜在的逃逸漏洞，仅建议内网封闭场景使用。

内容的提问来源于stack exchange，提问作者Zaid Al Shattle