5.4 安全队列——带结构完整性校验的环形缓冲区

场景导入：沉默的数据断裂

你的同事在检查一辆测试车的黑匣子数据时发现了一件怪事：CAN日志显示ABS控制器的四个轮速信号中，左后轮的数据在某一时刻后突然变成全零，持续了整整17秒后恢复正常。车辆在这17秒内经过了一个减速带，正是这个颠簸导致了左后轮速传感器线束的短暂断路。

物理故障并不可怕，ABS软件有专门的诊断逻辑处理传感器开路。真正让同事毛骨悚然的是：负责缓存轮速信号的环形缓冲区，在传感器恢复后吐出的第一条数据不是传感器恢复后的正确值，而是故障前残留的那个旧数据。

根因是软件缺陷：当传感器因断路返回错误代码时，任务直接跳过了入队操作。但缓冲区头指针在前一次操作中已经向前移动，缓冲区中的一个“空位“被错误地标记为“包含有效数据“。当传感器恢复、新数据入队后，这17秒的空白导致消费者和生产者之间的指针同步完全乱套。

内存缓冲区是安全性最脆弱的环节之一，不是因为算法复杂，而是因为它是“状态持有者“。当持有状态的内存被破坏时，没有机制能自检。

今天，你将构建一个能自检结构完整性的安全环形队列，safe_queue。

设计理念：每个条目都带“护照“

普通的环形缓冲区结构如下：

struct ring_buffer {
    uint8_t data[CAPACITY][ELEMENT_SIZE];
    int head, tail, count;  // 三个极易被破坏的整数
};

如果head、tail、count中的任何一个被野指针覆写（这在没有MPU的裸机系统中极为常见），整个队列的逻辑就全毁了：head>tail但count=0、tail>CAPACITY但count=CAPACITY-1……各种不可能的元数据组合不会触发任何检查，数据就被“合法地“送到了错误的地方。

safe_queue的防御策略是：给每个数据条目附加一个魔数和CRC校验，让队列的每个单元能够自我验证身份。

typedef struct {
    uint32_t magic;       /* 0xDEADBEEF — "我是一条合法的队列条目" */
    uint16_t length;      /* 有效数据长度 */
    uint8_t  data[32];     /* 载荷 */
    uint8_t  entry_crc;   /* CRC8(magic || length || data) */
} safe_queue_entry_t;

这就像给每个数据条目签发一本“护照“，magic是水印（证明这是真实的队列条目，不是随机RAM垃圾），entry_crc是防篡改码。当消费者取走数据时，在memcpy之前先验证护照。

注意CRC的覆盖域：magic || length || data。顺序很重要。magic作为CRC输入的第一个字段，意味着如果magic被破坏，CRC必然不匹配（因为CRC是输入的确定性函数，任何输入变化都会改变输出）。这避免了“先检查magic再检查CRC“的两步流程，CRC检查本身就已经包含了magic检查。

数据结构的全部成员

#define SAFE_QUEUE_MAX_ENTRIES  8
#define SAFE_QUEUE_DATA_SIZE    32

typedef struct {
    safe_queue_entry_t entries[SAFE_QUEUE_MAX_ENTRIES];
    uint32_t           head;      /* 生产者写位置 */
    uint32_t           tail;      /* 消费者读位置 */
    uint32_t           count;     /* 当前条目数 */
} safe_queue_t;

为什么要用uint32_t而不是更适合的size_t？ 因为这是一个教学级框架，我刻意使用定长整数（uint32_t）来模拟嵌入式环境中的选择，在Cortex-M上，size_t可能是16位或32位，但环形队列的容量是固定的（最多255个条目），使用uint32_t虽然有浪费但提供了最大的值域安全性。

为什么队列本身没有携带CRC？ 这是一个有意的简化。在生产代码中，队列元数据（head、tail、count）应该被CRC保护，尤其是当队列位于共享内存中被多核访问时。但将一个CRC隐藏在结构体末尾会导致所有元数据的修改都必须先计算CRC再写回，这增加了大量的复杂性。在5.6节你将看到，通过在validate()函数中逐个条目验证CRC，我们间接保护了head指针的有效性。如果head被破坏指向了垃圾，下一个get操作在验证该位置entry的magic时就会失败。

入队操作：封装entry并写入CRC

safe_status_t safe_queue_add(safe_queue_t *q, const uint8_t *data, size_t len)
{
    /* 前置检查 */
    if (q->count >= SAFE_QUEUE_MAX_ENTRIES) return SAFE_QUEUE_FULL;
    if (len > SAFE_QUEUE_DATA_SIZE) return SAFE_ERROR;

    /* 定位尾部槽位 */
    safe_queue_entry_t *entry = &q->entries[q->tail];

    /* 写入数据 */
    entry->magic  = SAFE_QUEUE_MAGIC;   /* 0xDEADBEEF */
    entry->length = (uint16_t)len;
    if (len > 0) memcpy(entry->data, data, len);

    /* 计算入队CRC */
    uint8_t crc_buf[sizeof(uint32_t) + sizeof(uint16_t) + SAFE_QUEUE_DATA_SIZE];
    memcpy(&crc_buf[0], &entry->magic,  sizeof(uint32_t));
    memcpy(&crc_buf[4], &entry->length, sizeof(uint16_t));
    if (len > 0) memcpy(&crc_buf[6], entry->data, len);
    entry->entry_crc = safe_crc8_compute(crc_buf, 6 + len);

    /* 更新队列元数据 */
    q->tail = (q->tail + 1) % SAFE_QUEUE_MAX_ENTRIES;
    q->count++;

    return SAFE_OK;
}

注意CRC计算中使用的临时缓冲区crc_buf。为什么不直接在entry上计算CRC？因为entry_crc字段本身也在safe_queue_entry_t中，如果你直接从entry的起始地址计算CRC，CRC计算将包含尚未确定的entry_crc字段，导致循环依赖。临时缓冲区将需要被CRC覆盖的字段（magic、length、data）重新排列并排除entry_crc自身。

PRODUCTION注意：在真实的嵌入式代码中，这个memcpy到临时缓冲区的步骤可以通过“散布-收集“CRC计算器来完成，硬件CRC引擎通常支持非连续的数据块，可以计算(magic, 4) + (length, 2) + (data, len) 三个片段的CRC而无需中间复制。Xilinx ZynqMP的PS端有一个支持DMA scatter-gather的CRC引擎，可以直接完成这个操作。

出队操作：护照检验处

safe_status_t safe_queue_get(safe_queue_t *q, uint8_t *data_out, size_t *len)
{
    if (q->count == 0) return SAFE_QUEUE_EMPTY;

    safe_queue_entry_t *entry = &q->entries[q->head];

    /* 验证护照：魔数验证 */
    if (entry->magic != SAFE_QUEUE_MAGIC) {
        return SAFE_MAGIC_ERROR;
    }

    /* 验证护照：CRC重新计算并对比 */
    uint8_t crc_buf[sizeof(uint32_t) + sizeof(uint16_t) + SAFE_QUEUE_DATA_SIZE];
    memcpy(&crc_buf[0], &entry->magic,  sizeof(uint32_t));
    memcpy(&crc_buf[4], &entry->length, sizeof(uint16_t));
    if (entry->length > 0) memcpy(&crc_buf[6], entry->data, entry->length);
    uint8_t computed = safe_crc8_compute(crc_buf, 6 + entry->length);
    if (computed != entry->entry_crc) {
        return SAFE_CRC_MISMATCH;
    }

    /* 通过验证：拷贝数据、清空条目、更新指针 */
    if (entry->length > 0) memcpy(data_out, entry->data, entry->length);
    *len = entry->length;
    memset(entry, 0, sizeof(*entry));  /* 清零防止残留数据泄露 */
    q->head = (q->head + 1) % SAFE_QUEUE_MAX_ENTRIES;
    q->count--;
    return SAFE_OK;
}

memset(entry, 0, sizeof(*entry))不仅仅是良好的编程习惯。在安全关键系统中，每条数据只应存在于它能被正确使用的上下文中。如果出队后不清零，考虑以下故障场景：

任务A写入敏感数据（如加密密钥）到队列
任务B读出并使用密钥
任务B的栈空间与队列的entry[3]重叠（因为队列在静态全局区）
任务B的栈溢出覆盖了队列的其他条目

如果entry在被覆盖之前没有清零，残留的密钥数据可能通过其他出队操作泄露。虽然教学代码的量级很小，但这是你在生产中应该养成的习惯。

批量验证：safe_queue_validate()

当系统怀疑内存损坏时（例如ECC控制器报告了可纠正错误），需要地毯式搜索整个队列：

safe_status_t safe_queue_validate(const safe_queue_t *q)
{
    uint32_t cnt = q->count;
    uint32_t idx = q->head;
    for (uint32_t i = 0; i < cnt; i++) {
        const safe_queue_entry_t *entry = &q->entries[idx];
        if (entry->magic != SAFE_QUEUE_MAGIC) return SAFE_MAGIC_ERROR;
        uint8_t crc_buf[6 + SAFE_QUEUE_DATA_SIZE];
        memcpy(&crc_buf[0], &entry->magic,  sizeof(uint32_t));
        memcpy(&crc_buf[4], &entry->length, sizeof(uint16_t));
        if (entry->length > 0) memcpy(&crc_buf[6], entry->data, entry->length);
        uint8_t computed = safe_crc8_compute(crc_buf, 6 + entry->length);
        if (computed != entry->entry_crc) return SAFE_CRC_MISMATCH;
        idx = (idx + 1) % SAFE_QUEUE_MAX_ENTRIES;
    }
    return SAFE_OK;
}

核心逻辑：从head开始，遍历count个槽位（而不是遍历整个entries数组）。这确保了只检查“队列认为已使用的“条目。如果head被损坏指向了垃圾，validate会立刻在第一个槽位就检测到magic不匹配。如果count被损坏为一个不合理的值（如count=255，但实际只有3个条目），validate会遍历到未初始化的条目，其magic不会是0xDEADBEEF，也会被检测到。

故障模型与检测

让我们通过测试代码来验证每种故障场景：

正常操作测试：

safe_queue_add(&q, "alpha", 6);
safe_queue_add(&q, "beta", 5);
safe_queue_add(&q, "gamma", 6);
// 按FIFO顺序取出
safe_queue_get(&q, buf, &len);  // buf = "alpha" ✓
safe_queue_get(&q, buf, &len);  // buf = "beta"  ✓
safe_queue_get(&q, buf, &len);  // buf = "gamma" ✓

魔数损坏测试：

safe_queue_add(&q, "sensitive_data", 15);
fault_queue_magic_corrupt(&q, 0);   // → entry[0].magic = 0xBAADF00D
safe_queue_get(&q, buf, &len);      // → SAFE_MAGIC_ERROR ✓

CRC损坏测试：

safe_queue_add(&q, "Q_FAULT", 8);
fault_queue_crc_corrupt(&q, last_entry_index);  // → entry_crc ^= 0x01
safe_queue_get(&q, buf, &len);                   // → SAFE_CRC_MISMATCH ✓

元数据损坏测试：

safe_queue_add(&q, "AAAA", 5);
safe_queue_add(&q, "BBBB", 5);   // head=0, tail=2, count=2
fault_queue_metadata_corrupt(&q);  // → head+=3 (head=3)
safe_queue_validate(&q);           // → SAFE_MAGIC_ERROR ✓
// head=3指向了未初始化的条目，magic != 0xDEADBEEF

满队列拒绝测试：

for (int i = 0; i < SAFE_QUEUE_MAX_ENTRIES + 2; i++) {
    st = safe_queue_add(&q, &i, sizeof(i));
}
// 前SAFE_QUEUE_MAX_ENTRIES次成功，后2次返回SAFE_QUEUE_FULL ✓

与E2E保护的配合

这里有一个容易忽略的设计要点：**E2E保护的是消息内容，安全队列保护的是存储结构。**两者并不冗余。

考虑以下数据流：

传感器 → E2E Protect() → [CRC|Ctr|DID|payload] → 安全队列 Add() → [magic|len|data|entry_crc]

数据经历了两层保护：

E2E层验证“这条消息是不是来自正确的传感器，内容是否完整“
队列层验证“这个存储槽位本身有没有被RAM损坏影响“

如果只依赖E2E保护而不保护队列结构，以下故障将无法被检测：

队列的head指针被破坏 → 消费者从错误的槽位读取数据 → 该槽位恰好也包含一条旧E2E消息 → E2E校验通过（消息本身没坏，但它不是新数据）
此时，安全队列的magic检查会先于E2E检查执行，正确拦截该错误

检查顺序的深度原理：get()函数先检查magic，再检查CRC。这两个检查的顺序不同于E2E的CRC→DataID→Counter的顺序。因为这里的magic检查是一个“类型检查“。如果数据连队列条目都不是（magic不匹配），就没有必要计算CRC了。这是一种优化，但更重要的是它在语义上是正确的：先验证身份，再验证内容。

本篇小结

安全队列的每个条目携带魔数（0xDEADBEEF）和独立 entry_crc（覆盖 magic + length + data），实现单条目级别的自检能力。
safe_queue_get() 先验证 magic（身份检查）再验证 CRC（内容检查），顺序与 E2E 相反，因为条目级 magic 校验是更轻量的预检。
出队后 memset 清零防止残留数据泄露，这是安全关键系统中“每条数据只存在于正确使用上下文“原则的体现。
safe_queue_validate() 从 head 开始遍历 count 个条目，间接保护了 head/tail/count 元数据的完整性——被破坏的指针会立即触发 magic 错误。
E2E 保护消息内容，安全队列保护存储结构，两者互补不可替代；队列本身不携带 CRC 是有意简化，生产系统中应加入元数据 CRC 保护。

【下集预告】：三道防线都建好了，但你凭什么相信它们真的能挡住攻击？下一节扮演“破坏者“角色，用四种故障注入函数分别模拟 SEU 比特翻转、EMI 钉死字节、ECU 意外复位和野指针破坏——然后观察每条防线是否如期响应。功能安全领域有一个残酷的原则：没有被测试过的安全机制等同于不存在。你会亲眼看到 safe_e2e_check() 在 CRC 损坏后返回 SAFE_CRC_MISMATCH 的那一刻，理解为什么疫苗逻辑同样适用于系统安全验证。

Keyboard shortcuts

汽车电子七部曲