40 changed files with 1655 additions and 36 deletions
+2
View File
@@ -1,6 +1,8 @@
# Firmware build and release output.
/target/
/version/
/output/
/.venv-customer-fs/
# Local legacy-board packages retained when switching to the new-board branch.
/artifacts/legacy-no-gnss/
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+18 -1
View File
@@ -1,9 +1,12 @@
@echo off
set CURDIR=%~dp0
for /f %%i in ('powershell -NoProfile -Command "Get-Date -Format yyyyMMddHHmmss"') do set appbuildtime=%%i
for /f %%i in ('powershell -NoProfile -Command "Get-Date -Format yyyyMMddHHmmss"') do set appbuildtime=%%i
set appbuildrev=NULL
set appbuildmodel=EG800AKCN_11LG
set appbuildver=beta1
set otalayoutid=eg800akcn_11lg-hxgnss-fs576-app2e5000
set otabuildprofile=ota_ready_hxgnss_fs576
set pythonexe=ql-cross-tool\win32\owtoolchain\Python\python.exe
if /i not "%1" == "" (
set appbuildver=%1
@@ -18,14 +21,28 @@ echo #define CUSTOM_APP_BUILD_TIME "%appbuildtime%" >> ql-applicati
echo #define CUSTOM_APP_BUILD_REV "%appbuildrev%" >> ql-application\threadx\interface\custom_app\lib\hal\build_macro.h
echo #define CUSTOM_APP_BUILD_MODEL "%appbuildmodel%" >> ql-application\threadx\interface\custom_app\lib\hal\build_macro.h
echo #define CUSTOM_APP_BUILD_VER "%appbuildver%" >> ql-application\threadx\interface\custom_app\lib\hal\build_macro.h
echo #define CUSTOM_APP_OTA_LAYOUT_ID "%otalayoutid%" >> ql-application\threadx\interface\custom_app\lib\hal\build_macro.h
echo #define CUSTOM_APP_OTA_BUILD_PROFILE "%otabuildprofile%" >> ql-application\threadx\interface\custom_app\lib\hal\build_macro.h
echo #endif >> ql-application\threadx\interface\custom_app\lib\hal\build_macro.h
"%pythonexe%" tools\verify_ota_ready_baseline.py
if errorlevel 1 (
echo "OTA-ready baseline source verification failed"
exit /b 1
)
call build.bat project_name -p EG800AKCN_11LG -b NO_EXTFLASH -x XIP -c %appbuildmodel%
call build.bat clean
call build.bat app
call build.bat kernel
call build.bat firmware
"%pythonexe%" tools\verify_ota_ready_baseline.py --app-bin ql-application\threadx\build\app.bin --kernel-bin ql-kernel\threadx\build\release_kernel.bin
if errorlevel 1 (
echo "OTA-ready baseline artifact verification failed"
exit /b 1
)
if exist "target\%appbuildmodel%.zip" (
mkdir target\%appbuildmodel%\fota\
mkdir target\%appbuildmodel%\fota\a
+273
View File
@@ -0,0 +1,273 @@
# Banban OTA-ready 工厂基线
## 1. 目标和边界
本基线用于尚未量产的 `EG800AKCN_11LG` 设备。目标是在不改变现有对话、音频、
MQTT、NFC、GNSS、按键、休眠和 LED 业务流程的前提下,一次性固定后续本地暂存式
OTA 所依赖的底层条件。
本基线完成:
- 将 HX_GNSS 的 `customer_fs` 从 256 KiB 扩大到 576 KiB;
- 将 `customer_app` 从 460 KiB 调整为 384 KiB;
- 将 APP 物理/XIP 地址固定为 `0x002E5000` / `0x802E5000`;
- 保留 SDK 已有的 `QUEC_APP_FULL_FOTA` 与 FULLOTA2 公共 API;
- 将现有 OTA 命令切换为最小本地暂存式 FULLOTA2;
- 为设备和构建产物加入唯一布局身份;
- 在构建前后执行 fail-closed 校验。
本基线暂不完成:
- 不增加签名 manifest、SHA-256 下载校验、断点续传或后台灰度;
- 不提供 A/B APP 分区,也不承诺自动版本回滚;
- 不在线更新 kernel、分区表或 `customer_fs`。
这版保留现有后端的 `version + url` 命令格式,只改变设备端执行方式。网络包会完整写入
`U:/FotaFile.bin`,通过 SDK FBF 校验并设置升级标志后才重启。
> 生产门禁:完成本文的真机 FULLOTA2 正向、失败注入和断电测试前,后端不得向生产设备
> 下发 OTA 命令。验收通过后可继续使用原协议下发;这条门禁针对测试状态,不再是因为
> 设备仍调用 FULLOTA1。
## 2. 固定身份与 Flash 布局
设备布局身份:
```text
eg800akcn_11lg-hxgnss-fs576-app2e5000
```
构建 profile:
```text
ota_ready_hxgnss_fs576
```
4 MiB 内部 Flash 的尾部分区固定如下:
| 分区 | 起始地址 | 结束地址 | 大小 |
| --- | ---: | ---: | ---: |
| reserved(按当前镜像估算) | `0x002D0000` | `0x002E5000` | 84 KiB |
| customer_app | `0x002E5000` | `0x00345000` | 384 KiB |
| customer_fs | `0x00345000` | `0x003D5000` | 576 KiB |
| nvm | `0x003D5000` | `0x003F5000` | 128 KiB |
| erase_rd | `0x003F5000` | `0x003F7000` | 8 KiB |
| fota_param | `0x003F7000` | `0x003FA000` | 12 KiB |
| factory_a | `0x003FA000` | `0x00400000` | 24 KiB |
`customer_app` 是单槽。FULLOTA2 能保护下载阶段:文件未下载完整或 SDK FBF 校验
失败时不会设置升级标志;但 updater 开始擦写 `customer_app` 后,没有旧 APP 槽可供回滚。
## 3. 本次修改
### 3.1 分区和链接地址
- `CRANELRH_QUEC_FLASH_LAYOUT_SS_04M_HX_GNSS.json`:当前 `CC11XXW_GPS` 构建实际选择的
分区表,`customer_app = 384KiB`、`customer_fs = 576KiB`;
- `CRANELRH_QUEC_FLASH_LAYOUT_SS_04M_OPEN.json`:同步保持 `customer_fs = 640KiB`,避免
无 GNSS variant 回退到旧布局;
- `app_linkscript.ld`:ROM 起始地址改为 `0x802E5000`,最大 ROM 大小收紧为 `0x60000`;
- `customer_fs.bin`:从原 256 KiB LittleFS 中提取全部 14 个音频文件,以 4096 字节块、
144 个块重新生成 576 KiB LittleFS;
- 新文件系统镜像 SHA-256:
`45350804d65fc6ded24caf9b215b71e89f9160d10d4998c5d57d4f4496d6000d`。
### 3.2 运行身份
`ota_baseline.h` 固定布局身份、APP/FS 地址和分区大小。应用启动时打印:
```text
[OTA-BASELINE] layout=eg800akcn_11lg-hxgnss-fs576-app2e5000 ... slot=single
[OTA-BASELINE] U: total=... free=... fota2_reserve=65536
```
该日志用于产线和联调确认设备没有混入旧的 fs256 或错误的 fs640 OPEN 基线,不改变任何业务状态。
### 3.3 构建保护
`build_rel.bat` 在构建前运行源码校验,在生成 APP 后运行容量校验。任何一项不满足都会中止:
- HX_GNSS `customer_fs` 必须为 576 KiB;
- APP 必须为 384 KiB,地址必须为 `0x002E5000`;
- LittleFS 必须为 4096 字节块、144 块且镜像哈希一致;
- 当前 kernel、DSP、RF 和 GNSS 镜像到 APP 之间必须至少保留 64 KiB;
- kernel 必须继续定义 `QUEC_APP_FULL_FOTA`;
- FULLOTA2 所需的 SDK API 必须存在;
- `app.bin + 8192` 不得超过 384 KiB。
- MQTT OTA 入口必须调用 `banban_fullfota2_start()`,不得回退到
`ql_fullfota_app_start()`。
可独立执行:
```powershell
ql-cross-tool\win32\owtoolchain\Python\python.exe tools\verify_ota_ready_baseline.py
ql-cross-tool\win32\owtoolchain\Python\python.exe tools\verify_ota_ready_baseline.py `
--app-bin ql-application\threadx\build\app.bin `
--kernel-bin ql-kernel\threadx\build\release_kernel.bin
```
### 3.4 最小 FULLOTA2 执行路径
`ota_fullfota2.c` 使用独立任务完成同步 HTTP 下载。现有 MQTT 工作线程只负责参数校验、
保存目标版本、回复 `accepted` 和启动任务,不会承担长时间网络读取。
执行顺序:
1. 删除上一次残留的 `U:/FotaFile.bin`;
2. HTTP GET 必须返回 200 和非零 `Content-Length`;
3. FBF 必须大于 8 KiB 且不超过 384 KiB;
4. `U:` 空闲空间必须不小于 `Content-Length + 64 KiB`;
5. 设置内部 Flash 固定暂存路径,初始化 FOTA context;
6. 以 4096 字节缓冲顺序读取并调用 `ql_fota_image_write()`;
7. 实际接收字节数必须严格等于 `Content-Length`;
8. 调用 `ql_fota_image_flush()`,再用 `ql_stat()` 复核落盘文件长度;
9. 调用 `ql_fota_image_verify()` 校验 FBF 并设置升级标志;
10. deinit 后重启,由 updater 写入单槽 `customer_app`。
任一步在设置升级标志前失败,都会关闭 HTTP/FOTA context、删除暂存包、撤销固定升级模式、
清除 pending 状态并把 `fota_start` 恢复为 0。旧 APP 未被擦写,可继续运行并接受下一次命令。
旧实现会在 OTA 前删除服务器管理的提示音。fs576 基线有足够空间后,这版不再主动删除
业务资源;若空间不足则明确失败,不通过牺牲业务文件强行升级。
`http_aofei.c` 中未被当前业务调用的旧 HTTP 轮询 OTA 包装函数原本仍可进入 FULLOTA1,
本基线将其显式禁用。生产固件只有 MQTT `version + url` 命令可以启动 OTA,避免两个入口
使用不同的升级语义。
## 4. 构建和生产烧录
在 Windows SDK 构建环境中执行:
```bat
build_rel.bat <生产版本号>
```
量产必须烧录该构建产生的完整 `*-fac.zip`,不能只烧 `customer_app`。完整工厂包需要同时
携带新 `partition.bin`、新 `customer_fs.bin`、11LG kernel/system 和新地址 APP。
量产前冻结并记录:
- 源码 Git commit;
- 工厂 ZIP 文件名、字节数和 SHA-256;
- `customer_fs.bin` SHA-256;
- `customer_app.bin` 字节数和 SHA-256;
- `partition.bin` 字节数和 SHA-256;
- 编译器、Aboot/arelease 和 FBFMake 版本;
- 首台、末台抽检设备串口中的 OTA baseline 日志。
不要把 `EG800AKCN_91LC` 的 `cp.bin`、工厂包或射频/GNSS 制品复制给本项目。两个项目同为
ASR1605/CRANELRH,但量产包必须由 `EG800AKCN_11LG` 配置生成。
## 5. 生产前测试
### 5.1 静态和产物测试
1. 运行 `verify_ota_ready_baseline.py`,结果必须通过。
2. 解包工厂 ZIP,确认其中 `customer_fs.bin` 为 589824 字节且哈希一致。
3. 使用 Aboot 工具检查 `partition.bin`,确认 APP/FS 地址与本文件一致。
4. 确认 `customer_app.bin + 8192 <= 393216`。
5. 确认 FBF 的目标物理地址是 `0x002E5000`,不能是旧地址 `0x00322000`、
也不能是错误基线地址 `0x002C2000`。
### 5.2 全擦工厂烧录测试
至少准备 3 台板执行完整擦除和工厂烧录:
1. 下载器无越界、校验或 partition 错误。
2. USB 重新枚举,`AT`/`ATI` 正常。
3. 串口打印正确 layout、地址和 576 KiB 文件系统容量。
4. 14 个预置音频均可读取和播放。
5. IMEI、SIM、驻网、数据连接和 GNSS 正常。
6. MQTT 注册、命令收发、心跳和状态上报正常。
7. 唤醒、对话、打断、录音、播放、音量和休眠正常。
8. NFC、按键、LED、电池检测和低电关机正常。
9. 连续冷启动、软重启和断电启动各不少于 20 次。
10. 连续运行至少 8 小时,观察 watchdog、内存、文件系统和网络重连。
### 5.3 文件系统压力测试
1. 启动后记录 `ql_fs_size('U')` 和 `ql_fs_free_size('U')`。
2. 写入一个与计划 FBF 等大的测试文件,关闭、重启并逐字节回读。
3. 删除测试文件,确认空间恢复。
4. 在写入 10%、50%、90% 时分别断电,重启后确认业务 APP 仍可运行并可清理残留文件。
5. 验证正常业务文件占用后,仍满足 `FBF size + 64 KiB` 空闲门槛。
### 5.4 最小 FULLOTA2 验收测试
正式量产前至少做一次 A 到 B 的本地暂存升级。B 可以只修改版本号:
1. 后端按现有协议下发 B 版 `version + url`;
2. 串口必须看到 HTTP 长度、下载进度、FBF verify 成功和重启日志;
3. 重启后确认 updater 安装 B,且 `U:/FotaFile.bin` 被正确处理;
4. 确认 NVM、设备身份、业务配置和 14 个预置音频未损坏;
5. 分别测试 404、无 Content-Length、空间不足、传输中断、截断 FBF、损坏 FBF 和旧地址
FBF,确认旧 APP 保持可运行且下一次 OTA 可以重新开始;
6. 在下载 10%、50%、90% 时断电,确认旧 APP 启动后清理残留状态并可再次升级;
7. 在 `ql_fota_image_verify()` 成功后的 updater 擦写阶段断电,记录底层 updater 的实际恢复
行为;当前单槽设计不能承诺该阶段自动回滚。
## 6. 后续 OTA 加固
后续 APP 版本建议按以下顺序加固,不再改变本基线分区:
1. 后端增加 size、SHA-256、layout ID 和单调递增 release sequence;
2. 下载过程中计算 SHA-256,完成后重新打开本地文件复核;
3. manifest 增加数字签名、公钥轮换和允许的 HTTPS host;
4. 将单 pending 文件升级为双时隙、CRC 和递增 generation;
5. 增加下载进度、稳定失败码和升级阶段上报;
6. 增加按设备灰度、暂停和失败率熔断;
7. 若未来硬件 Flash 扩容,再评估真正 A/B APP。
建议使用两个带 CRC 和递增 generation 的状态文件,例如 `U:/ota_state.a`、
`U:/ota_state.b`。这能抵抗状态文件写入中断,但不构成固件 A/B 回滚。
## 7. 后端联调契约
最小 FULLOTA2 继续兼容现有 URL 和版本协议,不要求后端立即修改。完成首轮生产验收后,
建议后端增加:
```json
{
"product": "EG800AKCN_11LG",
"layout_id": "eg800akcn_11lg-hxgnss-fs576-app2e5000",
"version": "1.9.0",
"release_seq": 10900,
"min_release_seq": 10800,
"url": "https://ota.example/releases/1.9.0/customer_app.fbf",
"size": 212992,
"sha256": "64-lowercase-hex",
"signature": "optional-first-stage"
}
```
设备校验顺序:产品、layout、版本序号、URL host、大小、SHA-256,最后才允许设置升级标志。
当前最小版本保留原服务器
`http://101.35.224.118:8080/assets/firmware/`,并允许 `talkingq.com` 根域及任意合法
子域名通过 HTTP/HTTPS 提供 `.bin` 文件,例如 `ota.talkingq.com` 和
`cdn.cn.talkingq.com`。校验基于解析出的完整 host 边界,`eviltalkingq.com`、
`talkingq.com.evil.example`、userinfo 绕过、非法端口和路径穿越均会被拒绝。
设备状态建议至少包含:
- `ACCEPTED`:命令和布局校验通过;
- `DOWNLOADING`:带下载字节数或百分比;
- `VERIFIED`:本地文件 SHA-256 和 FBF 校验通过;
- `REBOOTING`:升级标志已设置;
- `UPDATED`:目标 APP 已启动并通过健康检查;
- `FAILED`:带稳定错误码和失败阶段。
联调测试矩阵必须覆盖:同版本、降级、错误 layout、大小不符、SHA 不符、404、超时、下载
中断、下载阶段断电、verify 前断电、重启后无法确认,以及后台重复下发同一命令。
如果暂时没有动态 OTA 后端,也可以用“不可变 FBF 对象 + 签名 current.json”实现;但灰度、
设备级进度和失败统计需要后台服务。
## 8. 发布和回退策略
- 新生产设备只允许使用 HX_GNSS fs576 工厂包和 `0x002E5000` APP/FBF。
- 禁止向 fs576 设备发布旧地址 `0x00322000` 或错误地址 `0x002C2000` 的 FBF。
- 如生产前验证失败,可回退源码分支和工厂包重新全擦烧录;不得混烧新旧布局制品。
- 量产后若只是业务问题,可通过 APP OTA 修复;若分区或 kernel 基线有问题,需要返厂或有线
完整烧录,不能假定业务 APP OTA 能安全修改分区表。
- 当前为单槽。发布新版本必须先小批量灰度,并保持可用的有线恢复工厂包。
@@ -1,6 +1,6 @@
__ROM_START_ADDR = 0x80322000;
__MAX_ROM_SIZE = 0x100000;
__ROM_START_ADDR = 0x802E5000;
__MAX_ROM_SIZE = 0x60000;
__RAM_START_ADDR = 0x7e040800;
__MAX_RAM_SIZE = 0x100000;
@@ -86,6 +86,8 @@ SRC_FILES+= \
sleep_wake.c \
battery.c \
gnss.c \
ota_fullfota2.c \
ota_url_policy.c \
#-------------------------------------------------------------------------------
@@ -29,6 +29,7 @@
#include "ql_log.h"
#include "hal_api.h"
#include "nfc.h"
#include "ota_baseline.h"
#define APP_ENABLE_SLEEP
@@ -146,6 +147,17 @@ static void custom_app_task(void * argv)
hal_os_get_build_time(tempbuf);
printf("BUILD TIME:%s\r\n",tempbuf);
printf("[OTA-BASELINE] layout=%s profile=%s app_phys=0x%08lX app_xip=0x%08lX app_size=%lu fs_size=%lu slot=single\r\n",
BANBAN_OTA_LAYOUT_ID,
BANBAN_OTA_BUILD_PROFILE,
(unsigned long)BANBAN_OTA_APP_PHYSICAL_ADDRESS,
(unsigned long)BANBAN_OTA_APP_XIP_ADDRESS,
(unsigned long)BANBAN_OTA_APP_PARTITION_SIZE,
(unsigned long)BANBAN_OTA_FS_PARTITION_SIZE);
printf("[OTA-BASELINE] U: total=%d free=%d fota2_reserve=%lu\r\n",
ql_fs_size('U'),
ql_fs_free_size('U'),
(unsigned long)BANBAN_OTA_STAGING_RESERVE_BYTES);
//app_uartsh_init();
@@ -517,7 +517,9 @@ exit:
int http_ota_fw_download(char *url)
{
return ql_fullfota_app_start(url);
(void)url;
RLOGE("legacy HTTP polling FULLOTA1 is disabled; use MQTT FULLOTA2 command\r\n");
return -1;
}
typedef struct recv_buf
@@ -37,7 +37,7 @@ const char *mp3_file[] ={
"barge_in_cute.mp3",//19 对话打断提示音
};
//http://banban.api.talkingq.com:8080/assets/roles/banban/tts_error.mp3
//http://101.35.224.118:8080/assets/roles/banban/tts_error.mp3
//#define PCM_FILE "http://47.113.99.138:7050/output_16k.pcm"
@@ -197,7 +197,7 @@ void app_file_check_download(void)
if(ql_access(filename_buff, 0)!=0){
printf("downing mp3 file %d,%s\r\n", index,filename_buff);
down_loop:
sprintf(url_buff, "http://banban.api.talkingq.com:8080/assets/roles/banban/%s", mp3_file[index]);
sprintf(url_buff, "http://101.35.224.118:8080/assets/roles/banban/%s", mp3_file[index]);
ret = http_file_download(url_buff);
if(ret!=0&&cnt++<3) goto down_loop;
@@ -0,0 +1,21 @@
#ifndef BANBAN_OTA_BASELINE_H
#define BANBAN_OTA_BASELINE_H
#include <stdint.h>
#include "build_macro.h"
#define BANBAN_OTA_EXPECTED_LAYOUT_ID "eg800akcn_11lg-hxgnss-fs576-app2e5000"
#define BANBAN_OTA_LAYOUT_ID CUSTOM_APP_OTA_LAYOUT_ID
#define BANBAN_OTA_BUILD_PROFILE CUSTOM_APP_OTA_BUILD_PROFILE
#define BANBAN_OTA_APP_PHYSICAL_ADDRESS UINT32_C(0x002E5000)
#define BANBAN_OTA_APP_XIP_ADDRESS UINT32_C(0x802E5000)
#define BANBAN_OTA_APP_PARTITION_SIZE UINT32_C(0x00060000)
#define BANBAN_OTA_FS_PHYSICAL_ADDRESS UINT32_C(0x00345000)
#define BANBAN_OTA_FS_PARTITION_SIZE UINT32_C(0x00090000)
#define BANBAN_OTA_STAGING_RESERVE_BYTES UINT32_C(0x00010000)
#if !defined(CUSTOM_APP_OTA_LAYOUT_ID) || !defined(CUSTOM_APP_OTA_BUILD_PROFILE)
#error "Use build_rel.bat to build the OTA-ready production baseline"
#endif
#endif
@@ -0,0 +1,25 @@
#ifndef BANBAN_OTA_FULLOTA2_H
#define BANBAN_OTA_FULLOTA2_H
typedef enum {
BANBAN_FULLOTA2_OK = 0,
BANBAN_FULLOTA2_ERR_ARGUMENT = -1,
BANBAN_FULLOTA2_ERR_BUSY = -2,
BANBAN_FULLOTA2_ERR_TASK = -3,
BANBAN_FULLOTA2_ERR_HTTP = -4,
BANBAN_FULLOTA2_ERR_LENGTH = -5,
BANBAN_FULLOTA2_ERR_SPACE = -6,
BANBAN_FULLOTA2_ERR_INIT = -7,
BANBAN_FULLOTA2_ERR_WRITE = -8,
BANBAN_FULLOTA2_ERR_FLUSH = -9,
BANBAN_FULLOTA2_ERR_VERIFY = -10
} banban_fullfota2_result_t;
typedef void (*banban_fullfota2_finished_cb_t)(int result);
int banban_fullfota2_start(const char *url,
banban_fullfota2_finished_cb_t finished_cb);
int banban_fullfota2_is_running(void);
void banban_fullfota2_discard_staged_package(void);
#endif
@@ -0,0 +1,6 @@
#ifndef BANBAN_OTA_URL_POLICY_H
#define BANBAN_OTA_URL_POLICY_H
int banban_ota_url_is_allowed(const char *url);
#endif
@@ -4,4 +4,6 @@
#define CUSTOM_APP_BUILD_REV "NULL"
#define CUSTOM_APP_BUILD_MODEL "EG800AKCN_11LG"
#define CUSTOM_APP_BUILD_VER "1.8.0"
#define CUSTOM_APP_OTA_LAYOUT_ID "eg800akcn_11lg-hxgnss-fs576-app2e5000"
#define CUSTOM_APP_OTA_BUILD_PROFILE "ota_ready_hxgnss_fs576"
#endif
@@ -21,6 +21,8 @@
#include "cJSON.h"
#include "build_macro.h"
#include "ql_log.h"
#include "ota_fullfota2.h"
#include "ota_url_policy.h"
@@ -45,7 +47,6 @@
#define OTA_STATE_FORMAT_VERSION 1U
#define OTA_PROMPT_PLAY_TIMEOUT_MS 10000
#define OTA_PROMPT_POLL_MS 50
#define OTA_ALLOWED_URL_PREFIX "http://101.35.224.118:8080/assets/firmware/"
static ql_timer_t mqtt_event_task_timer = NULL;
@@ -242,6 +243,17 @@ static void ota_state_clear(void)
}
}
static void ota_fullfota2_finished(int result)
{
if (result == BANBAN_FULLOTA2_OK) {
return;
}
RLOGE("FULLOTA2 failed before activation: %d", result);
ota_state_clear();
fota_start = 0;
}
static int ota_parse_version(const char *version, unsigned int parts[3])
{
char tail;
@@ -285,24 +297,6 @@ static int ota_compare_version(const char *left, const char *right, int *result)
return 0;
}
static int ota_url_is_allowed(const char *url)
{
size_t prefix_len = strlen(OTA_ALLOWED_URL_PREFIX);
if (url == NULL || url[0] == '\0') {
return 0;
}
if (strncmp(url, OTA_ALLOWED_URL_PREFIX, prefix_len) != 0) {
return 0;
}
if (url[prefix_len] == '\0' || strstr(url + prefix_len, "..") != NULL) {
return 0;
}
return strstr(url + prefix_len, ".bin") != NULL;
}
/* ==================== 解析函数 ==================== */
/**
@@ -1246,6 +1240,8 @@ static void ota_report_pending_success(void)
compare_result != 0) {
RLOGE("OTA pending state version mismatch: current=%s target=%s",
current_version, state.target_version);
banban_fullfota2_discard_staged_package();
ota_state_clear();
return;
}
@@ -1304,7 +1300,7 @@ static void handle_ota_upgrade(command_t *cmd)
return;
}
if (!ota_url_is_allowed(cmd->params.ota.url)) {
if (!banban_ota_url_is_allowed(cmd->params.ota.url)) {
RLOGE("OTA URL is not allowed: %s", cmd->params.ota.url);
return;
}
@@ -1369,12 +1365,10 @@ static void handle_ota_upgrade(command_t *cmd)
}
}
/* Product requirement: keep deleting the configured audio files before OTA. */
app_file_check_del();
ret = ql_fullfota_app_start(cmd->params.ota.url);
ret = banban_fullfota2_start(cmd->params.ota.url,
ota_fullfota2_finished);
if (ret != 0) {
RLOGE("start OTA failed: %d", ret);
RLOGE("start FULLOTA2 task failed: %d", ret);
fota_start = 0;
ota_state_clear();
}
@@ -0,0 +1,272 @@
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "ql_fota.h"
#include "ql_fs.h"
#include "ql_power.h"
#include "ql_rtos.h"
#include "webclient.h"
#include "ota_baseline.h"
#include "ota_fullfota2.h"
#define BANBAN_FULLOTA2_PACKAGE_PATH "U:/FotaFile.bin"
#define BANBAN_FULLOTA2_URL_MAX 256U
#define BANBAN_FULLOTA2_READ_BUFFER_SIZE 4096U
#define BANBAN_FULLOTA2_TASK_STACK_SIZE (12U * 1024U)
#define BANBAN_FULLOTA2_TASK_PRIORITY 100U
#define BANBAN_FULLOTA2_HTTP_TIMEOUT_MS 30000
#define BANBAN_FULLOTA2_FBF_OVERHEAD 8192U
typedef struct {
char url[BANBAN_FULLOTA2_URL_MAX];
banban_fullfota2_finished_cb_t finished_cb;
} banban_fullfota2_request_t;
static ql_task_t g_fullfota2_task = NULL;
static volatile int g_fullfota2_running = 0;
static banban_fullfota2_request_t g_fullfota2_request;
static void banban_fullfota2_finish(int result,
struct webclient_session *session,
uint8_t *buffer,
qlFotaImgProcCtxPtr fota)
{
banban_fullfota2_finished_cb_t finished_cb =
g_fullfota2_request.finished_cb;
if (session != NULL) {
webclient_close(session);
}
if (buffer != NULL) {
web_free(buffer);
}
if (fota != NULL) {
ql_fota_deinit(fota);
}
if (result != BANBAN_FULLOTA2_OK) {
ql_fota_fixed_internal_flash_upgrade(0);
if (ql_access(BANBAN_FULLOTA2_PACKAGE_PATH, 0) == 0) {
ql_remove(BANBAN_FULLOTA2_PACKAGE_PATH);
}
}
memset(&g_fullfota2_request, 0, sizeof(g_fullfota2_request));
g_fullfota2_running = 0;
g_fullfota2_task = NULL;
if (finished_cb != NULL) {
finished_cb(result);
}
}
static void banban_fullfota2_task(void *argument)
{
struct webclient_session *session = NULL;
qlFotaImgProcCtxPtr fota = NULL;
uint8_t *buffer = NULL;
int status;
int content_length;
int free_size;
int read_length;
int written = 0;
int next_progress = 10;
int result = BANBAN_FULLOTA2_ERR_HTTP;
struct lfs_info staged_info;
(void)argument;
printf("[FULLOTA2] start url=%s path=%s\r\n",
g_fullfota2_request.url,
BANBAN_FULLOTA2_PACKAGE_PATH);
if (ql_access(BANBAN_FULLOTA2_PACKAGE_PATH, 0) == 0) {
ql_remove(BANBAN_FULLOTA2_PACKAGE_PATH);
}
session = webclient_session_create(2048);
if (session == NULL) {
result = BANBAN_FULLOTA2_ERR_HTTP;
goto failed;
}
webclient_set_timeout(session, BANBAN_FULLOTA2_HTTP_TIMEOUT_MS);
status = webclient_get(session, g_fullfota2_request.url);
if (status != 200) {
printf("[FULLOTA2] HTTP GET failed status=%d\r\n", status);
result = BANBAN_FULLOTA2_ERR_HTTP;
goto failed;
}
content_length = session->content_length;
if (content_length <= (int)BANBAN_FULLOTA2_FBF_OVERHEAD ||
content_length > (int)BANBAN_OTA_APP_PARTITION_SIZE) {
printf("[FULLOTA2] invalid Content-Length=%d\r\n", content_length);
result = BANBAN_FULLOTA2_ERR_LENGTH;
goto failed;
}
free_size = ql_fs_free_size('U');
if (free_size < 0 ||
(uint32_t)free_size <
(uint32_t)content_length + BANBAN_OTA_STAGING_RESERVE_BYTES) {
printf("[FULLOTA2] insufficient space free=%d required=%lu\r\n",
free_size,
(unsigned long)((uint32_t)content_length +
BANBAN_OTA_STAGING_RESERVE_BYTES));
result = BANBAN_FULLOTA2_ERR_SPACE;
goto failed;
}
ql_fota_fixed_internal_flash_upgrade(1);
if (ql_fota_set_package_path(BANBAN_FULLOTA2_PACKAGE_PATH) != 0) {
printf("[FULLOTA2] set package path failed\r\n");
result = BANBAN_FULLOTA2_ERR_INIT;
goto failed;
}
fota = ql_fota_init();
if (fota == NULL) {
printf("[FULLOTA2] ql_fota_init failed\r\n");
result = BANBAN_FULLOTA2_ERR_INIT;
goto failed;
}
buffer = web_malloc(BANBAN_FULLOTA2_READ_BUFFER_SIZE);
if (buffer == NULL) {
printf("[FULLOTA2] allocate read buffer failed\r\n");
result = BANBAN_FULLOTA2_ERR_INIT;
goto failed;
}
while (written < content_length) {
int request_length = content_length - written;
int progress;
if (request_length > (int)BANBAN_FULLOTA2_READ_BUFFER_SIZE) {
request_length = BANBAN_FULLOTA2_READ_BUFFER_SIZE;
}
read_length = webclient_read(session, buffer, request_length);
if (read_length <= 0) {
printf("[FULLOTA2] body read failed ret=%d received=%d expected=%d\r\n",
read_length, written, content_length);
result = BANBAN_FULLOTA2_ERR_HTTP;
goto failed;
}
if (ql_fota_image_write(fota, buffer, read_length) != 0) {
printf("[FULLOTA2] image write failed received=%d chunk=%d\r\n",
written, read_length);
result = BANBAN_FULLOTA2_ERR_WRITE;
goto failed;
}
written += read_length;
progress = (written * 100) / content_length;
if (progress >= next_progress || written == content_length) {
printf("[FULLOTA2] download progress=%d%% bytes=%d/%d\r\n",
progress, written, content_length);
while (next_progress <= progress) {
next_progress += 10;
}
}
}
if (written != content_length) {
result = BANBAN_FULLOTA2_ERR_LENGTH;
goto failed;
}
if (ql_fota_image_flush(fota) != 0) {
printf("[FULLOTA2] image flush failed\r\n");
result = BANBAN_FULLOTA2_ERR_FLUSH;
goto failed;
}
memset(&staged_info, 0, sizeof(staged_info));
if (ql_stat(BANBAN_FULLOTA2_PACKAGE_PATH, &staged_info) != 0 ||
staged_info.size != (uint32_t)content_length) {
printf("[FULLOTA2] staged length mismatch file=%lu expected=%d\r\n",
(unsigned long)staged_info.size, content_length);
result = BANBAN_FULLOTA2_ERR_LENGTH;
goto failed;
}
if (ql_fota_image_verify(fota) != 0) {
printf("[FULLOTA2] FBF verify/setflag failed\r\n");
result = BANBAN_FULLOTA2_ERR_VERIFY;
goto failed;
}
printf("[FULLOTA2] package verified; rebooting into updater\r\n");
banban_fullfota2_finish(BANBAN_FULLOTA2_OK, session, buffer, fota);
ql_rtos_task_sleep_ms(200);
ql_power_reset();
/* ql_power_reset() is not expected to return. Keep the task quiesced if it does. */
while (1) {
ql_rtos_task_sleep_s(60);
}
failed:
printf("[FULLOTA2] failed result=%d\r\n", result);
banban_fullfota2_finish(result, session, buffer, fota);
ql_rtos_task_delete(NULL);
}
int banban_fullfota2_start(const char *url,
banban_fullfota2_finished_cb_t finished_cb)
{
int ret;
size_t url_length;
if (url == NULL || url[0] == '\0') {
return BANBAN_FULLOTA2_ERR_ARGUMENT;
}
url_length = strlen(url);
if (url_length >= sizeof(g_fullfota2_request.url)) {
return BANBAN_FULLOTA2_ERR_ARGUMENT;
}
if (g_fullfota2_running) {
return BANBAN_FULLOTA2_ERR_BUSY;
}
memset(&g_fullfota2_request, 0, sizeof(g_fullfota2_request));
memcpy(g_fullfota2_request.url, url, url_length + 1);
g_fullfota2_request.finished_cb = finished_cb;
g_fullfota2_running = 1;
ret = ql_rtos_task_create(&g_fullfota2_task,
BANBAN_FULLOTA2_TASK_STACK_SIZE,
BANBAN_FULLOTA2_TASK_PRIORITY,
"ota_fullfota2",
banban_fullfota2_task,
NULL);
if (ret != 0) {
memset(&g_fullfota2_request, 0, sizeof(g_fullfota2_request));
g_fullfota2_running = 0;
g_fullfota2_task = NULL;
return BANBAN_FULLOTA2_ERR_TASK;
}
return BANBAN_FULLOTA2_OK;
}
int banban_fullfota2_is_running(void)
{
return g_fullfota2_running ? 1 : 0;
}
void banban_fullfota2_discard_staged_package(void)
{
if (!g_fullfota2_running &&
ql_access(BANBAN_FULLOTA2_PACKAGE_PATH, 0) == 0) {
ql_remove(BANBAN_FULLOTA2_PACKAGE_PATH);
}
}
@@ -0,0 +1,189 @@
#include <stddef.h>
#include <string.h>
#include "ota_url_policy.h"
#define BANBAN_OTA_LEGACY_URL_PREFIX \
"http://101.35.224.118:8080/assets/firmware/"
#define BANBAN_OTA_BASE_DOMAIN "talkingq.com"
static int banban_ascii_equal_nocase(const char *left,
const char *right,
size_t length)
{
size_t i;
for (i = 0; i < length; ++i) {
unsigned char a = (unsigned char)left[i];
unsigned char b = (unsigned char)right[i];
if (a >= 'A' && a <= 'Z') {
a = (unsigned char)(a + ('a' - 'A'));
}
if (b >= 'A' && b <= 'Z') {
b = (unsigned char)(b + ('a' - 'A'));
}
if (a != b) {
return 0;
}
}
return 1;
}
static int banban_hostname_is_valid(const char *host, size_t host_length)
{
size_t i;
int label_has_character = 0;
int previous_was_hyphen = 0;
if (host_length == 0 || host_length > 253) {
return 0;
}
for (i = 0; i < host_length; ++i) {
unsigned char character = (unsigned char)host[i];
if ((character >= 'A' && character <= 'Z') ||
(character >= 'a' && character <= 'z') ||
(character >= '0' && character <= '9')) {
label_has_character = 1;
previous_was_hyphen = 0;
continue;
}
if (character == '-') {
if (!label_has_character) {
return 0;
}
previous_was_hyphen = 1;
continue;
}
if (character == '.') {
if (!label_has_character || previous_was_hyphen) {
return 0;
}
label_has_character = 0;
previous_was_hyphen = 0;
continue;
}
return 0;
}
return label_has_character && !previous_was_hyphen;
}
static int banban_host_is_talkingq(const char *host, size_t host_length)
{
const size_t base_length = sizeof(BANBAN_OTA_BASE_DOMAIN) - 1;
if (!banban_hostname_is_valid(host, host_length)) {
return 0;
}
if (host_length == base_length) {
return banban_ascii_equal_nocase(host,
BANBAN_OTA_BASE_DOMAIN,
base_length);
}
if (host_length <= base_length + 1 ||
host[host_length - base_length - 1] != '.') {
return 0;
}
return banban_ascii_equal_nocase(host + host_length - base_length,
BANBAN_OTA_BASE_DOMAIN,
base_length);
}
static int banban_port_is_valid(const char *port, size_t port_length)
{
size_t i;
unsigned long value = 0;
if (port_length == 0 || port_length > 5) {
return 0;
}
for (i = 0; i < port_length; ++i) {
if (port[i] < '0' || port[i] > '9') {
return 0;
}
value = value * 10 + (unsigned long)(port[i] - '0');
}
return value > 0 && value <= 65535;
}
static int banban_path_is_fbf(const char *path)
{
const char *path_end;
const char *query;
if (path == NULL || path[0] != '/' || path[1] == '\0' ||
strstr(path, "..") != NULL || strchr(path, '\\') != NULL ||
strchr(path, '#') != NULL) {
return 0;
}
query = strchr(path, '?');
path_end = query != NULL ? query : path + strlen(path);
if ((size_t)(path_end - path) < 4) {
return 0;
}
return banban_ascii_equal_nocase(path_end - 4, ".bin", 4);
}
int banban_ota_url_is_allowed(const char *url)
{
const char *authority;
const char *path;
const char *host_end;
const char *colon;
const size_t legacy_prefix_length =
sizeof(BANBAN_OTA_LEGACY_URL_PREFIX) - 1;
if (url == NULL || url[0] == '\0') {
return 0;
}
if (strncmp(url, BANBAN_OTA_LEGACY_URL_PREFIX,
legacy_prefix_length) == 0) {
return banban_path_is_fbf(url + legacy_prefix_length - 1);
}
if (strncmp(url, "http://", 7) == 0) {
authority = url + 7;
} else if (strncmp(url, "https://", 8) == 0) {
authority = url + 8;
} else {
return 0;
}
path = strchr(authority, '/');
if (path == NULL || path == authority) {
return 0;
}
if (memchr(authority, '@', (size_t)(path - authority)) != NULL) {
return 0;
}
colon = memchr(authority, ':', (size_t)(path - authority));
host_end = colon != NULL ? colon : path;
if (!banban_host_is_talkingq(authority,
(size_t)(host_end - authority))) {
return 0;
}
if (colon != NULL &&
!banban_port_is_valid(colon + 1, (size_t)(path - colon - 1))) {
return 0;
}
return banban_path_is_fbf(path);
}
Binary file not shown.
@@ -61,12 +61,12 @@
{
"name": "customer_app",
"format": "raw",
"size": "460KiB"
"size": "384KiB"
},
{
"name": "customer_fs",
"format": "raw",
"size": "256KiB"
"size": "576KiB"
},
{
"name": "nvm",
@@ -56,7 +56,7 @@
{
"name": "customer_fs",
"format": "raw",
"size": "256KiB"
"size": "640KiB"
},
{
"name": "nvm",
+215
View File
@@ -0,0 +1,215 @@
# customer_fs.bin 双系统生成工具
本目录提供 `EG800AKCN_11LG` 项目的 `customer_fs.bin` 自动生成和校验工具,支持
Windows 与 macOS。
当前 `customer_fs` 分区大小为 576 KiB,设备 kernel 只支持 LittleFS 磁盘格式
`0x00020000`(disk 2.0)。所有生成入口都会在输出前执行兼容性检查,避免再次生成
kernel 无法挂载的 disk 2.1 镜像。
## 1. 源文件目录
文件系统的源文件统一存放在:
```text
audio_prompts/customer_fs/
```
当前目录中的 14 个音频文件,全部从已经过实机验证、能够正常挂载的 Windows
`customer_fs.bin` 中提取,并经过重新打包和逐文件回读验证。
设备代码从 `U:/` 根目录访问文件,因此所有文件必须直接放在上述目录中。例如:
```text
正确:audio_prompts/customer_fs/network_connect_zh.mp3
错误:audio_prompts/customer_fs/subdir/network_connect_zh.mp3
```
生成工具会拒绝隐藏文件和子目录,避免把 `.DS_Store` 等无关文件打进正式镜像。
## 2. 如何增删改文件
生成工具采用“根据源目录全量重建”的方式,不是在旧 bin 上追加内容。
- 添加:把新文件直接放进 `audio_prompts/customer_fs/`;
- 替换:用新文件覆盖同名文件;
- 编辑:使用音频工具修改原文件并保存到原位置;
- 删除:从源目录删除文件,下一次生成的 bin 中将不再包含该文件;
- 改名:直接修改源文件名,但必须同步确认业务代码使用的新文件名。
部分音频文件名已经写入业务代码。删除或改名后,如果代码仍然播放原文件,串口会打印:
```text
tips file missing
```
因此,生成工具允许删减文件,但删减前仍需确认业务代码不再使用它。
## 3. Windows 一键生成
在项目根目录执行:
```bat
tools\customer_fs\build_windows.bat
```
Windows 版本使用项目自带并已通过实机验证的工具:
```text
ql-cross-tool\win32\mklfs-v2.exe
```
同时使用 SDK 自带的 Python 执行兼容性校验,不需要另外安装 Python 依赖。
也可以指定自定义源目录和输出文件:
```bat
tools\customer_fs\build_windows.bat 源目录 输出文件
```
例如:
```bat
tools\customer_fs\build_windows.bat ^
audio_prompts\customer_fs ^
output\customer_fs\customer_fs.bin
```
## 4. macOS 一键生成
在项目根目录执行:
```bash
./tools/customer_fs/build_macos.sh
```
首次运行时,脚本会自动创建:
```text
.venv-customer-fs
```
并安装项目固定的依赖版本:
```text
littlefs-python==0.19.0
```
新版 `littlefs-python` 默认会生成 disk 2.1 镜像,不能直接使用它自带的命令行生成。
本项目的 `generate_macos.py` 会显式设置:
```text
disk_version = 0x00020000
```
从而生成当前设备 kernel 可以识别的 disk 2.0 镜像。
也可以指定自定义源目录和输出文件:
```bash
./tools/customer_fs/build_macos.sh 源目录 输出文件
```
例如:
```bash
./tools/customer_fs/build_macos.sh \
audio_prompts/customer_fs \
output/customer_fs/customer_fs.bin
```
## 5. 默认输出位置
Windows 和 macOS 默认都输出到:
```text
output/customer_fs/customer_fs.bin
```
生成时会先写入临时文件:
```text
customer_fs.bin.tmp
```
只有生成和全部校验成功后,临时文件才会替换正式输出。生成或校验失败时,已有正式输出
不会被覆盖。
## 6. 自动门禁检查
每次生成都会检查:
| 检查项 | 要求 |
| --- | ---: |
| 镜像大小 | 589824 字节 |
| LittleFS 磁盘格式 | `0x00020000` |
| Block Size | 4096 字节 |
| Block Count | 144 |
| 第一个 superblock | 必须存在且参数正确 |
| 第二个 superblock | 必须存在且参数正确 |
| 源目录 | 不能为空 |
| 源文件层级 | 只能是根目录文件 |
| 隐藏文件 | 不允许 |
| 文件名 | 必须能在生成镜像中找到 |
也可以单独校验一个镜像:
```bash
python3 tools/customer_fs/verify_customer_fs.py \
--image output/customer_fs/customer_fs.bin \
--source audio_prompts/customer_fs
```
校验成功会输出镜像大小、LittleFS 版本、块参数、源文件数量和 SHA-256。
## 7. 放入正式固件工程
生成并校验成功后,需要显式复制到 SDK 打包位置。
macOS:
```bash
cp output/customer_fs/customer_fs.bin \
ql-config/config/EG800AKCN_11LG/customer_fs.bin
```
Windows:
```bat
copy /Y output\customer_fs\customer_fs.bin ^
ql-config\config\EG800AKCN_11LG\customer_fs.bin
```
替换前建议备份原文件并记录新镜像的 SHA-256。
`customer_fs.bin` 不是当前 APP-only OTA 的升级内容。修改该镜像后,需要重新生成并烧录
包含分区表、kernel、customer_app 和 customer_fs 的完整工厂包,不能只下发 APP OTA。
## 8. 实机验收
Windows 生成工具已经过实机验证。macOS 工具已经通过结构校验、重新挂载和逐文件回读,
首次用于正式版本前仍应至少烧录一台设备验收。
启动日志应直接出现:
```text
[U] mount successfully
```
不能出现:
```text
[U] mount failed (-22), will format fs partition
```
还需要逐一验证镜像中的音频能够通过 `U:/文件名` 正常读取和播放,并记录本次发布使用的
`customer_fs.bin` SHA-256。
## 9. 注意事项
- 不要用十六进制工具把 disk 2.1 镜像的版本字段直接改成 disk 2.0;
- 不要直接使用 `python -m littlefs create`,它当前默认生成 disk 2.1;
- 不要把 `.venv-customer-fs`、`output` 或 `.DS_Store` 提交到仓库;
- Windows 和 macOS 生成镜像的整体 SHA-256 可能不同,这是块分配顺序不同造成的;
- 两个平台产物必须满足相同的 disk 2.0、4096 字节块和 144 块约束;
- 修改、删除或重命名业务音频前,必须同步检查代码中的文件名引用。
+38
View File
@@ -0,0 +1,38 @@
#!/bin/sh
set -eu
SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
ROOT_DIR=$(CDPATH= cd -- "$SCRIPT_DIR/../.." && pwd)
SOURCE_DIR=${1:-"$ROOT_DIR/audio_prompts/customer_fs"}
OUTPUT_IMAGE=${2:-"$ROOT_DIR/output/customer_fs/customer_fs.bin"}
VENV_DIR=${CUSTOMER_FS_VENV:-"$ROOT_DIR/.venv-customer-fs"}
PYTHON_BOOTSTRAP=${PYTHON3:-python3}
TEMP_IMAGE="${OUTPUT_IMAGE}.tmp"
cleanup() {
rm -f -- "$TEMP_IMAGE"
}
trap cleanup EXIT INT TERM
if [ ! -x "$VENV_DIR/bin/python" ]; then
"$PYTHON_BOOTSTRAP" -m venv "$VENV_DIR"
fi
if ! "$VENV_DIR/bin/python" -c 'import littlefs; raise SystemExit(0 if littlefs.__version__ == "0.19.0" else 1)' 2>/dev/null; then
"$VENV_DIR/bin/python" -m pip install --requirement "$SCRIPT_DIR/requirements.txt"
fi
mkdir -p -- "$(dirname -- "$OUTPUT_IMAGE")"
rm -f -- "$TEMP_IMAGE"
"$VENV_DIR/bin/python" "$SCRIPT_DIR/generate_macos.py" \
--source "$SOURCE_DIR" \
--output "$TEMP_IMAGE"
"$VENV_DIR/bin/python" "$SCRIPT_DIR/verify_customer_fs.py" \
--image "$TEMP_IMAGE" \
--source "$SOURCE_DIR"
mv -f -- "$TEMP_IMAGE" "$OUTPUT_IMAGE"
trap - EXIT INT TERM
printf 'customer_fs ready: %s\n' "$OUTPUT_IMAGE"
+56
View File
@@ -0,0 +1,56 @@
@echo off
setlocal EnableExtensions
set "SCRIPT_DIR=%~dp0"
for %%I in ("%SCRIPT_DIR%..\..") do set "ROOT_DIR=%%~fI"
if "%~1"=="" (
set "SOURCE_DIR=%ROOT_DIR%\audio_prompts\customer_fs"
) else (
set "SOURCE_DIR=%~f1"
)
if "%~2"=="" (
set "OUTPUT_IMAGE=%ROOT_DIR%\output\customer_fs\customer_fs.bin"
) else (
set "OUTPUT_IMAGE=%~f2"
)
set "TEMP_IMAGE=%OUTPUT_IMAGE%.tmp"
set "MKLFS=%ROOT_DIR%\ql-cross-tool\win32\mklfs-v2.exe"
set "PYTHON=%ROOT_DIR%\ql-cross-tool\win32\owtoolchain\Python\python.exe"
if not exist "%MKLFS%" (
echo customer_fs generation failed: mklfs-v2.exe not found at "%MKLFS%"
exit /b 1
)
if not exist "%PYTHON%" (
echo customer_fs generation failed: bundled Python not found at "%PYTHON%"
exit /b 1
)
if not exist "%SOURCE_DIR%\" (
echo customer_fs generation failed: source directory not found at "%SOURCE_DIR%"
exit /b 1
)
for %%I in ("%OUTPUT_IMAGE%") do if not exist "%%~dpI" mkdir "%%~dpI"
if exist "%TEMP_IMAGE%" del /q "%TEMP_IMAGE%"
"%MKLFS%" -c "%SOURCE_DIR%" -s 589824 -i "%TEMP_IMAGE%"
if errorlevel 1 goto :failed
"%PYTHON%" "%SCRIPT_DIR%verify_customer_fs.py" ^
--image "%TEMP_IMAGE%" ^
--source "%SOURCE_DIR%"
if errorlevel 1 goto :failed
move /Y "%TEMP_IMAGE%" "%OUTPUT_IMAGE%" >nul
if errorlevel 1 goto :failed
echo customer_fs ready: "%OUTPUT_IMAGE%"
exit /b 0
:failed
if exist "%TEMP_IMAGE%" del /q "%TEMP_IMAGE%"
echo customer_fs generation failed; existing output was not changed.
exit /b 1
+76
View File
@@ -0,0 +1,76 @@
#!/usr/bin/env python3
"""Generate a disk-2.0 LittleFS customer_fs image on macOS."""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
from littlefs import LittleFS, UserContext
IMAGE_SIZE = 589824
BLOCK_SIZE = 4096
BLOCK_COUNT = 144
DISK_VERSION = 0x00020000
def fail(message: str) -> None:
raise SystemExit(f"customer_fs generation failed: {message}")
def collect_source_files(source: Path) -> list[Path]:
if not source.is_dir():
fail(f"source directory does not exist: {source}")
entries = sorted(source.iterdir(), key=lambda path: path.name.encode("utf-8"))
if not entries:
fail(f"source directory is empty: {source}")
invalid = [entry.name for entry in entries if not entry.is_file() or entry.name.startswith(".")]
if invalid:
fail(
"source must contain only visible root-level files; invalid entries: "
+ ", ".join(invalid)
)
return entries
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--source", required=True, type=Path)
parser.add_argument("--output", required=True, type=Path)
args = parser.parse_args()
files = collect_source_files(args.source)
context = UserContext(buffsize=IMAGE_SIZE)
fs = LittleFS(
context=context,
mount=False,
block_size=BLOCK_SIZE,
block_count=BLOCK_COUNT,
read_size=16,
prog_size=16,
cache_size=256,
lookahead_size=32,
block_cycles=500,
disk_version=DISK_VERSION,
)
fs.format()
fs.mount()
try:
for source_file in files:
with source_file.open("rb") as source_handle:
with fs.open("/" + source_file.name, "wb") as image_handle:
while chunk := source_handle.read(64 * 1024):
image_handle.write(chunk)
finally:
fs.unmount()
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_bytes(context.buffer)
print(f"generated {args.output} from {len(files)} files")
return 0
if __name__ == "__main__":
sys.exit(main())
+1
View File
@@ -0,0 +1 @@
littlefs-python==0.19.0
+121
View File
@@ -0,0 +1,121 @@
#!/usr/bin/env python3
"""Validate a customer_fs LittleFS image for the EG800AKCN_11LG kernel."""
from __future__ import annotations
import argparse
import hashlib
import struct
import sys
from pathlib import Path
DEFAULT_SIZE = 589824
DEFAULT_DISK_VERSION = 0x00020000
DEFAULT_BLOCK_SIZE = 4096
DEFAULT_BLOCK_COUNT = 144
SUPERBLOCK_MAGIC_OFFSETS = (8, DEFAULT_BLOCK_SIZE + 8)
def fail(message: str) -> None:
raise SystemExit(f"customer_fs verification failed: {message}")
def source_files(source: Path) -> list[Path]:
if not source.is_dir():
fail(f"source directory does not exist: {source}")
entries = sorted(source.iterdir(), key=lambda path: path.name.encode("utf-8"))
if not entries:
fail(f"source directory is empty: {source}")
invalid = [entry.name for entry in entries if not entry.is_file() or entry.name.startswith(".")]
if invalid:
fail(
"source must contain only visible root-level files; invalid entries: "
+ ", ".join(invalid)
)
return entries
def verify_superblock(
image: bytes,
magic_offset: int,
disk_version: int,
block_size: int,
block_count: int,
) -> None:
if image[magic_offset : magic_offset + 8] != b"littlefs":
fail(f"LittleFS magic is missing at 0x{magic_offset:X}")
actual_version, actual_block_size, actual_block_count = struct.unpack_from(
"<III", image, magic_offset + 12
)
if actual_version != disk_version:
fail(
f"disk version at 0x{magic_offset:X} is 0x{actual_version:08X}; "
f"expected 0x{disk_version:08X}"
)
if actual_block_size != block_size:
fail(
f"block size at 0x{magic_offset:X} is {actual_block_size}; "
f"expected {block_size}"
)
if actual_block_count != block_count:
fail(
f"block count at 0x{magic_offset:X} is {actual_block_count}; "
f"expected {block_count}"
)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--image", required=True, type=Path)
parser.add_argument("--source", type=Path, help="also check root-level source filenames")
parser.add_argument("--size", type=int, default=DEFAULT_SIZE)
parser.add_argument(
"--disk-version", type=lambda value: int(value, 0), default=DEFAULT_DISK_VERSION
)
parser.add_argument("--block-size", type=int, default=DEFAULT_BLOCK_SIZE)
parser.add_argument("--block-count", type=int, default=DEFAULT_BLOCK_COUNT)
args = parser.parse_args()
if not args.image.is_file():
fail(f"image does not exist: {args.image}")
image = args.image.read_bytes()
if len(image) != args.size:
fail(f"image size is {len(image)}; expected {args.size}")
if args.block_size * args.block_count != args.size:
fail("block size multiplied by block count does not equal image size")
for magic_offset in (8, args.block_size + 8):
verify_superblock(
image,
magic_offset,
args.disk_version,
args.block_size,
args.block_count,
)
files: list[Path] = []
if args.source:
files = source_files(args.source)
missing = [path.name for path in files if path.name.encode("utf-8") not in image]
if missing:
fail("source filenames missing from image: " + ", ".join(missing))
digest = hashlib.sha256(image).hexdigest()
print("customer_fs verified")
print(f" image: {args.image}")
print(f" size: {len(image)}")
print(f" disk_version: 0x{args.disk_version:08X}")
print(f" block_size: {args.block_size}")
print(f" block_count: {args.block_count}")
if files:
print(f" source_files: {len(files)}")
print(f" sha256: {digest}")
return 0
if __name__ == "__main__":
sys.exit(main())
+53
View File
@@ -0,0 +1,53 @@
#include <stdio.h>
#include "ota_url_policy.h"
typedef struct {
const char *url;
int expected;
} url_case_t;
int main(void)
{
static const url_case_t cases[] = {
{"http://101.35.224.118:8080/assets/firmware/app.bin", 1},
{"http://101.35.224.118:8080/assets/firmware/release/app.bin?token=x", 1},
{"http://talkingq.com/app.bin", 1},
{"https://talkingq.com/releases/app.bin", 1},
{"https://ota.talkingq.com/releases/app.bin", 1},
{"https://cdn.cn.talkingq.com:8443/a/B.BIN?token=x", 1},
{"https://OTA.TALKINGQ.COM/app.bin", 1},
{"https://eviltalkingq.com/app.bin", 0},
{"https://talkingq.com.evil.example/app.bin", 0},
{"https://talkingq.com@evil.example/app.bin", 0},
{"https://evil.example@talkingq.com/app.bin", 0},
{"https://a..talkingq.com/app.bin", 0},
{"https://-ota.talkingq.com/app.bin", 0},
{"https://ota-.talkingq.com/app.bin", 0},
{"https://ota.talkingq.com:0/app.bin", 0},
{"https://ota.talkingq.com:65536/app.bin", 0},
{"https://ota.talkingq.com:abc/app.bin", 0},
{"ftp://ota.talkingq.com/app.bin", 0},
{"https://ota.talkingq.com/", 0},
{"https://ota.talkingq.com/app.zip", 0},
{"https://ota.talkingq.com/app.bin/extra", 0},
{"https://ota.talkingq.com/a/../app.bin", 0},
{"https://ota.talkingq.com/app.bin#fragment", 0},
{"http://101.35.224.118:8080/other/app.bin", 0},
{"http://192.168.1.1/app.bin", 0},
};
size_t i;
for (i = 0; i < sizeof(cases) / sizeof(cases[0]); ++i) {
int actual = banban_ota_url_is_allowed(cases[i].url);
if (actual != cases[i].expected) {
fprintf(stderr, "case %lu failed: expected=%d actual=%d url=%s\n",
(unsigned long)i, cases[i].expected, actual, cases[i].url);
return 1;
}
}
printf("OTA URL policy tests passed: %lu cases\n",
(unsigned long)(sizeof(cases) / sizeof(cases[0])));
return 0;
}
+242
View File
@@ -0,0 +1,242 @@
#!/usr/bin/env python3
"""Fail-closed checks for the EG800AKCN_11LG OTA-ready factory baseline."""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import struct
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
LAYOUT = ROOT / "ql-config/quec-project/aboot/config/partition/CRANELRH_QUEC_FLASH_LAYOUT_SS_04M_HX_GNSS.json"
FLASH_TYPE = ROOT / "ql-config/config/QuecFLashType.ini"
CHIP_PLATFORM = ROOT / "ql-config/config/CHIP_PLAT.ini"
LINKER = ROOT / "ql-application/threadx/config/common/app_linkscript.ld"
FS_IMAGE = ROOT / "ql-config/config/EG800AKCN_11LG/customer_fs.bin"
DSP_IMAGE = ROOT / "ql-config/soc_platform/CRANELRH/images/DSP_Reduce_SuLog_Buffer/dsp.bin"
RF_IMAGE = ROOT / "ql-config/config/EG800AKCN_11LG/rf.bin"
GNSS_BOOT_IMAGE = ROOT / "ql-config/config/EG800AKCN_11LG/gnss_boot.pkg"
GNSS_FIRM_IMAGE = ROOT / "ql-config/config/EG800AKCN_11LG/gnss_firm.pkg"
KERNEL_FLAGS = ROOT / "ql-kernel/threadx/config/common/default_dflags_1602.mk"
FOTA_HEADER = ROOT / "ql-application/threadx/common/include/ql_fota.h"
BASELINE_HEADER = ROOT / "ql-application/threadx/interface/custom_app/inc/ota_baseline.h"
BUILD_MACRO = ROOT / "ql-application/threadx/interface/custom_app/lib/hal/build_macro.h"
APP_MAKEFILE = ROOT / "ql-application/threadx/interface/custom_app/Makefile"
MQTT_SOURCE = ROOT / "ql-application/threadx/interface/custom_app/mqtt.c"
FULLOTA2_SOURCE = ROOT / "ql-application/threadx/interface/custom_app/ota_fullfota2.c"
URL_POLICY_SOURCE = ROOT / "ql-application/threadx/interface/custom_app/ota_url_policy.c"
LEGACY_HTTP_SOURCE = ROOT / "ql-application/threadx/interface/custom_app/http_aofei.c"
FLASH_SIZE = 0x400000
EXPECTED = {
"app_start": 0x002E5000,
"app_size": 0x00060000,
"fs_start": 0x00345000,
"fs_size": 0x00090000,
"xip_start": 0x802E5000,
}
FBF_HEADER_ALLOWANCE = 8192
MINIMUM_SYSTEM_MARGIN = 64 * 1024
EXPECTED_FS_SHA256 = "45350804d65fc6ded24caf9b215b71e89f9160d10d4998c5d57d4f4496d6000d"
def fail(message: str) -> None:
raise SystemExit(f"OTA-ready baseline verification failed: {message}")
def parse_size(value: str) -> int:
if value == "-1":
return -1
match = re.fullmatch(r"(\d+)(KiB|MiB)", value)
if not match:
fail(f"unsupported partition size {value!r}")
scale = 1024 if match.group(2) == "KiB" else 1024 * 1024
return int(match.group(1)) * scale
def flattened_tail() -> list[tuple[str, int]]:
data = json.loads(LAYOUT.read_text(encoding="utf-8"))
qspi = next(item for item in data["partitions"] if item["name"] == "qspi")
result: list[tuple[str, int]] = []
for item in qspi["partitions"]:
if item["name"] == "factory":
result.extend((child["name"], parse_size(child["size"])) for child in item["partitions"])
elif item["name"] in {"reserved", "customer_app", "customer_fs", "nvm", "erase_rd", "fota_param"}:
result.append((item["name"], parse_size(item["size"])))
return result
def verify_layout() -> None:
if CHIP_PLATFORM.read_text(encoding="utf-8").strip() != "1605":
fail("this baseline requires CHIP_PLAT.ini=1605")
if FLASH_TYPE.read_text(encoding="utf-8").strip() != "CC11XXW_GPS":
fail("this baseline requires QuecFLashType.ini=CC11XXW_GPS")
tail = flattened_tail()
sizes = dict(tail)
if sizes.get("customer_app") != EXPECTED["app_size"]:
fail("customer_app must be 384 KiB")
if sizes.get("customer_fs") != EXPECTED["fs_size"]:
fail("customer_fs must be 576 KiB")
suffix = 0
starts: dict[str, int] = {}
for name, size in reversed(tail):
if size < 0:
continue
suffix += size
starts[name] = FLASH_SIZE - suffix
if starts.get("customer_app") != EXPECTED["app_start"]:
fail(f"customer_app starts at 0x{starts.get('customer_app', 0):08X}, expected 0x{EXPECTED['app_start']:08X}")
if starts.get("customer_fs") != EXPECTED["fs_start"]:
fail(f"customer_fs starts at 0x{starts.get('customer_fs', 0):08X}, expected 0x{EXPECTED['fs_start']:08X}")
def verify_linker() -> None:
text = LINKER.read_text(encoding="utf-8")
rom = re.search(r"__ROM_START_ADDR\s*=\s*(0x[0-9A-Fa-f]+)", text)
maximum = re.search(r"__MAX_ROM_SIZE\s*=\s*(0x[0-9A-Fa-f]+)", text)
if not rom or int(rom.group(1), 16) != EXPECTED["xip_start"]:
fail("linker ROM address is not 0x802E5000")
if not maximum or int(maximum.group(1), 16) != EXPECTED["app_size"]:
fail("linker ROM limit is not the 384 KiB customer_app partition")
def verify_fs_image() -> None:
image = FS_IMAGE.read_bytes()
if len(image) != EXPECTED["fs_size"]:
fail(f"customer_fs.bin is {len(image)} bytes, expected {EXPECTED['fs_size']}")
digest = hashlib.sha256(image).hexdigest()
if digest != EXPECTED_FS_SHA256:
fail(f"customer_fs.bin SHA-256 is {digest}; expected {EXPECTED_FS_SHA256}")
magic_at = image.find(b"littlefs", 0, 64)
if magic_at < 0:
fail("customer_fs.bin has no LittleFS superblock")
# The global-state tag follows the magic; geometry starts 12 bytes later.
disk_version, block_size, block_count = struct.unpack_from("<III", image, magic_at + 12)
if disk_version >> 16 != 2:
fail(f"unsupported LittleFS disk version 0x{disk_version:08X}")
if block_size != 4096 or block_count != 144:
fail(f"LittleFS geometry is block_size={block_size}, block_count={block_count}; expected 4096/144")
def align_up(value: int, alignment: int) -> int:
return (value + alignment - 1) // alignment * alignment
def verify_system_fit(kernel_path: Path) -> None:
images = (kernel_path, DSP_IMAGE, RF_IMAGE, GNSS_BOOT_IMAGE, GNSS_FIRM_IMAGE)
for path in images:
if not path.is_file():
fail(f"system image does not exist: {path}")
# bootloader=88 KiB; system begins with ptable=4 KiB and rd=16 KiB.
system_end = (88 + 4 + 16) * 1024
for path in images:
system_end = align_up(system_end, 4096)
system_end += path.stat().st_size
system_end = align_up(system_end, 4096)
margin = EXPECTED["app_start"] - system_end
if margin < MINIMUM_SYSTEM_MARGIN:
fail(
f"system ends at 0x{system_end:08X}; margin before customer_app is "
f"{margin} bytes, expected at least {MINIMUM_SYSTEM_MARGIN}"
)
def verify_fota_contract() -> None:
flags = KERNEL_FLAGS.read_text(encoding="utf-8", errors="replace")
if "-DQUEC_APP_FULL_FOTA" not in flags:
fail("kernel no longer enables QUEC_APP_FULL_FOTA")
api = FOTA_HEADER.read_text(encoding="utf-8", errors="replace")
for symbol in (
"ql_fota_init",
"ql_fota_image_write",
"ql_fota_image_flush",
"ql_fota_image_verify",
"ql_fota_set_package_path",
"ql_fota_fixed_internal_flash_upgrade",
):
if symbol not in api:
fail(f"SDK FULLOTA2 API {symbol} is missing")
identity = BASELINE_HEADER.read_text(encoding="utf-8")
if "eg800akcn_11lg-hxgnss-fs576-app2e5000" not in identity:
fail("application layout identity is missing")
generated = BUILD_MACRO.read_text(encoding="utf-8", errors="replace")
if '#define CUSTOM_APP_OTA_LAYOUT_ID "eg800akcn_11lg-hxgnss-fs576-app2e5000"' not in generated:
fail("generated build macro has the wrong OTA layout identity")
if '#define CUSTOM_APP_OTA_BUILD_PROFILE "ota_ready_hxgnss_fs576"' not in generated:
fail("generated build macro has the wrong OTA build profile")
makefile = APP_MAKEFILE.read_text(encoding="utf-8", errors="replace")
if "ota_fullfota2.c" not in makefile:
fail("minimal FULLOTA2 implementation is not part of the application build")
if "ota_url_policy.c" not in makefile:
fail("OTA URL policy is not part of the application build")
mqtt = MQTT_SOURCE.read_text(encoding="utf-8", errors="replace")
if "banban_fullfota2_start" not in mqtt:
fail("MQTT OTA command does not start the local-staging FULLOTA2 path")
if "ql_fullfota_app_start(cmd->params.ota.url)" in mqtt:
fail("MQTT OTA command still invokes the legacy FULLOTA1 path")
if "banban_ota_url_is_allowed" not in mqtt:
fail("MQTT OTA command does not enforce the shared URL policy")
url_policy = URL_POLICY_SOURCE.read_text(encoding="utf-8", errors="replace")
if 'BANBAN_OTA_BASE_DOMAIN "talkingq.com"' not in url_policy:
fail("OTA URL policy no longer allows the talkingq.com domain family")
legacy_http = LEGACY_HTTP_SOURCE.read_text(encoding="utf-8", errors="replace")
if "return ql_fullfota_app_start(url)" in legacy_http:
fail("legacy HTTP polling path can still invoke FULLOTA1")
implementation = FULLOTA2_SOURCE.read_text(encoding="utf-8", errors="replace")
required_steps = (
"ql_fs_free_size('U')",
"ql_fota_fixed_internal_flash_upgrade(1)",
"ql_fota_set_package_path",
"ql_fota_init",
"ql_fota_image_write",
"ql_fota_image_flush",
"ql_stat",
"ql_fota_image_verify",
"ql_power_reset",
)
positions = [implementation.find(step) for step in required_steps]
if any(position < 0 for position in positions) or positions != sorted(positions):
fail("minimal FULLOTA2 prepare/verify/activate sequence is incomplete or reordered")
def verify_app(path: Path) -> None:
if not path.is_file():
fail(f"application binary does not exist: {path}")
size = path.stat().st_size
if size + FBF_HEADER_ALLOWANCE > EXPECTED["app_size"]:
fail(
f"app.bin ({size}) plus {FBF_HEADER_ALLOWANCE}-byte FBF allowance exceeds "
f"customer_app ({EXPECTED['app_size']})"
)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--app-bin", type=Path, help="also enforce the application/FBF capacity gate")
parser.add_argument("--kernel-bin", type=Path, help="also enforce the GNSS system/customer_app margin")
args = parser.parse_args()
verify_layout()
verify_linker()
verify_fs_image()
verify_fota_contract()
if args.app_bin:
verify_app(args.app_bin if args.app_bin.is_absolute() else ROOT / args.app_bin)
if args.kernel_bin:
verify_system_fit(args.kernel_bin if args.kernel_bin.is_absolute() else ROOT / args.kernel_bin)
print("OTA-ready baseline verified: hxgnss fs576, app384@0x002E5000, FULLOTA2 available")
return 0
if __name__ == "__main__":
sys.exit(main())