//! 设备指令下发 — 通过 Redis 队列向设备发送指令 //! //! API 服务不直接持有 TCP 连接,而是通过 Redis LIST `device:cmd:{imei}` //! 推送指令,由设备服务消费并发送到设备。指令响应通过 `device:reply:{imei}` //! 队列返回,API 服务通过 `PendingCommands` 匹配 msg_id 完成异步等待。 use std::collections::HashMap; use std::sync::Arc; use axum::extract::{Path, State}; use axum::Json; use serde::Deserialize; use serde_json::{json, Value}; use tokio::sync::{oneshot, Mutex}; use crate::error::AppError; use crate::middleware::auth::{self, CurrentUser}; use crate::protocol::ServerResponse; /// msg_id → oneshot Sender 映射表 #[derive(Clone)] pub struct PendingCommands { inner: Arc>>>, } impl PendingCommands { pub fn new() -> Self { Self { inner: Arc::new(Mutex::new(HashMap::new())) } } /// 注册等待中的指令 #[allow(dead_code)] pub async fn insert(&self, msg_id: u32, tx: oneshot::Sender) { self.inner.lock().await.insert(msg_id, tx); } /// 移除并返回指定 msg_id 的 sender(消费一次) pub async fn remove(&self, msg_id: u32) -> Option> { self.inner.lock().await.remove(&msg_id) } } /// 设备服务节点信息 #[derive(Debug, Clone)] pub struct NodeInfo { pub node_id: String, pub ip: String, pub tcp_port: u16, /// RFC3339 格式的最后心跳时间,从 Redis 读取 pub last_heartbeat: Option, } /// 节点注册表(Redis 存储,API 服务从 Redis 读取节点信息) /// /// 设备服务通过 HSET 写入 `node:{node_id}` hash, /// API 服务通过此模块读取。Redis 是节点信息的单一数据源。 #[derive(Clone)] pub struct NodeRegistry { redis: redis::aio::ConnectionManager, } impl NodeRegistry { pub fn new(redis: redis::aio::ConnectionManager) -> Self { Self { redis } } fn node_key(node_id: &str) -> String { format!("node:{}", node_id) } /// 注册节点(写入 Redis HSET) pub async fn register(&self, node_id: &str, ip: &str, tcp_port: u16) { let key = Self::node_key(node_id); let now = chrono::Utc::now().to_rfc3339(); let mut conn = self.redis.clone(); let _: Result<(), _> = redis::cmd("HSET") .arg(&key) .arg("node_id").arg(node_id) .arg("ip").arg(ip) .arg("tcp_port").arg(tcp_port.to_string()) .arg("status").arg("online") .arg("last_heartbeat").arg(&now) .query_async(&mut conn) .await; // ponytail: HSET 不支持 EX,用 EXPIRE 设置 TTL let _: Result<(), _> = redis::cmd("EXPIRE") .arg(&key) .arg(300_u64) .query_async(&mut conn) .await; } /// 移除节点(从 Redis 删除) pub async fn remove(&self, node_id: &str) { let key = Self::node_key(node_id); let mut conn = self.redis.clone(); let _: Result<(), _> = redis::cmd("DEL") .arg(&key) .query_async(&mut conn) .await; } /// 更新心跳时间戳 pub async fn heartbeat(&self, node_id: &str) -> bool { let key = Self::node_key(node_id); let mut conn = self.redis.clone(); let result: Result<(), _> = redis::cmd("HSET") .arg(&key) .arg("last_heartbeat") .arg(chrono::Utc::now().to_rfc3339()) .query_async(&mut conn) .await; // 心跳同时刷新 TTL let _: Result<(), _> = redis::cmd("EXPIRE") .arg(&key) .arg(300_u64) .query_async(&mut conn) .await; result.is_ok() } /// 列出所有节点(从 Redis 读取,使用 SCAN 替代 KEYS 避免阻塞) pub async fn list(&self) -> Vec { let mut conn = self.redis.clone(); let keys = self.scan_keys(&mut conn, "node:*").await; let mut nodes = Vec::with_capacity(keys.len()); for key in keys { let mut conn = self.redis.clone(); let info: HashMap = redis::cmd("HGETALL") .arg(&key) .query_async(&mut conn) .await .unwrap_or_default(); let node_id = info.get("node_id").cloned().unwrap_or_default(); if node_id.is_empty() { continue; // ponytail: 跳过空记录 } nodes.push(NodeInfo { node_id, ip: info.get("ip").cloned().unwrap_or_default(), tcp_port: info.get("tcp_port").and_then(|p| p.parse().ok()).unwrap_or(0), last_heartbeat: info.get("last_heartbeat").cloned(), }); } nodes } /// 使用 SCAN 命令安全地获取所有匹配模式的键(避免 KEYS 阻塞) async fn scan_keys(&self, conn: &mut redis::aio::ConnectionManager, pattern: &str) -> Vec { let mut keys = Vec::new(); let mut cursor = 0i64; loop { let (next_cursor, batch): (i64, Vec) = redis::cmd("SCAN") .arg(cursor) .arg("MATCH") .arg(pattern) .arg("COUNT") .arg(100) .query_async(conn) .await .unwrap_or_default(); keys.extend(batch); if next_cursor == 0 { break; } cursor = next_cursor; } keys } /// 检查设备是否**可下发**(即:是否存在一条**已 `login` 成功**的连接)。 /// /// # 为什么不是查 `device:{imei}` /// 该键在**验签之前**就会被写(设备一发 `auth_str`/`login` 就写),断连后还能存活 300 秒 ⇒ /// 用它判在线会得到"管理端显示已下发成功、设备实际收不到"这种最难排查的故障。 /// /// 改用 **连接级认证标记** `device:authed:{imei}`(与 `device-server` 的**跨服务契约**, /// 见 `device-server/src/tcp/connection.rs::auth_marker_key`): /// - **写入**:向该连接投递 `login` 成功应答(含 `spec` + `suc=1`)时; /// - **刷新**:该连接每次上行 / 每次下发; /// - **删除**:连接断开、或被同一 IMEI 的新连接接管时(新连接须重新 `login`)。 /// /// 它是"**可下发**"的必要条件,不是充分条件:`device-server` 侧还有第二道门控 /// (未认证连接一律不投递,见 `PRE_LOGIN_ALLOWED_ACTS`)。 pub async fn is_device_online(&self, imei: &str) -> bool { let key = auth_marker_key(imei); let mut conn = self.redis.clone(); let exists: Result = redis::cmd("EXISTS") .arg(&key) .query_async(&mut conn) .await; exists.unwrap_or(false) } } /// 连接级认证标记的键前缀 —— **必须与 `device-server` 的 `auth_marker_key` 完全一致** /// (跨服务契约;两侧各有常量/键格式断言守住,见 `commands.rs` 与 `connection.rs` 的单测)。 const AUTH_MARKER_PREFIX: &str = "device:authed:"; /// 连接级认证标记的键。 fn auth_marker_key(imei: &str) -> String { format!("{}{}", AUTH_MARKER_PREFIX, imei) } /// 应用共享状态 #[derive(Clone)] #[allow(dead_code)] pub struct AppState { pub redis: redis::aio::ConnectionManager, pub mysql: sqlx::MySqlPool, pub pending_commands: PendingCommands, pub nodes: NodeRegistry, pub redis_url: String, /// JWT 签名密钥 — 启动阶段从环境变量读取,缺失则 panic(禁止硬编码回退) pub jwt_secret: Arc, } /// 指令下发请求体 #[derive(Debug, Deserialize)] pub struct CommandRequest { pub act: String, #[serde(default)] pub msg_id: u32, #[serde(flatten)] pub params: Option, } /// 内部路由字段名:推入 `device:commands` 时附加,值 = 目标设备 IMEI。 /// /// 用途:device-server 据此决定把指令发给哪条连接;**下发前摘除,设备侧不可见**。 /// 为什么需要它:协议定稿后 `id` 只表示**动作目标**(柜级省略 / 一段=单板 / 两段=仓体级), /// **不再含 IMEI**,所以设备身份必须另用内部字段传递。 pub const ROUTE_DEV_ID_FIELD: &str = "_dev_id"; /// 设备指令结构(序列化后推入 Redis) /// /// 协议格式(v1.1):`id` 只表示**动作目标**,**不含 IMEI**(设备身份由 TCP 连接确定) /// - **柜级**指令(open/pow_on/pow_off/reset 整柜):**不带 `id`**(字段省略) /// - **单仓控板**指令(reset 单板):`仓控板`(如 `"01"`) /// - **仓体级**指令(on/off):`仓控板-通道`(如 `"01-3"`) #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct DeviceCommand { pub act: String, pub msg_id: u32, /// 动作目标(柜级为 `None`),序列化为 "id" #[serde(skip_serializing_if = "Option::is_none")] pub id: Option, /// 扩展参数(充电参数等) #[serde(flatten)] pub params: Option, } /// 向设备发送指令(通过 Redis 队列) pub async fn send_device_command( redis: &redis::aio::ConnectionManager, dev_id: &str, cmd: &DeviceCommand, ) -> Result<(), String> { let mut json_val = serde_json::to_value(cmd).map_err(|e| format!("指令序列化失败: {}", e))?; // 内部路由字段:告知 device-server 这条指令发给哪台设备(下发前摘除) json_val[ROUTE_DEV_ID_FIELD] = serde_json::json!(dev_id); let json_str = json_val.to_string(); let mut conn = redis.clone(); redis::cmd("LPUSH") .arg("device:commands") .arg(&json_str) .query_async::<()>(&mut conn) .await .map_err(|e| format!("Redis 推入失败: {}", e))?; Ok(()) } /// 向设备发送响应(通过 Redis 队列) #[allow(dead_code)] pub async fn send_device_response( redis: &redis::aio::ConnectionManager, _dev_id: &str, resp: &ServerResponse, ) -> Result<(), String> { let json_str = resp.to_json(); let mut conn = redis.clone(); redis::cmd("LPUSH") .arg("device:commands") .arg(&json_str) .query_async::<()>(&mut conn) .await .map_err(|e| format!("Redis 推入失败: {}", e))?; Ok(()) } /// 本接口放行的**柜级/仓体级指令白名单**。 /// /// 为什么必须有白名单:`act` 是请求体里的自由字符串。原先不加限制 + 不做归属校验, /// 任何持有 `device:operate` 的低权限用户都能对**全平台任意柜机**下发协议里的 /// **任何**动作(开门 / 断电 / 重启),属于越权高危操作(C2)。 /// /// 只放行管理端确有需要的动作: /// - `open` / `pow_on` / `pow_off`:**柜级**(协议不带 `id`) /// - `reset`:柜级;单仓控板时由调用方显式传 `id = "01"` /// - `on` / `off`:**仓体级**,调用方必须显式传 `id = "01-3"`(协议要求) const CABINET_COMMAND_ACTS: &[&str] = &["on", "off", "open", "pow_on", "pow_off", "reset"]; /// 协议 `id` 的合法形态校验:`id` 只表示**动作目标** —— /// 一段 = 单仓控板(`"01"`)/ 两段 = 仓体级(`"01-3"`),每段 1~2 位数字。 /// /// 协议定稿后 `id` **不含 IMEI**(设备身份由 TCP 连接确定),因此 15 位 IMEI 一律非法。 fn is_valid_command_target(id: &str) -> bool { /// 单段:1~2 位数字(仓控板 `"1"`/`"01"`,通道 `"3"`) fn segment_ok(seg: &str) -> bool { !seg.is_empty() && seg.len() <= 2 && seg.bytes().all(|b| b.is_ascii_digit()) } let mut parts = id.split('-'); match (parts.next(), parts.next(), parts.next()) { (Some(a), None, _) => segment_ok(a), (Some(a), Some(b), None) => segment_ok(a) && segment_ok(b), // 空串、三段及以上(含 IMEI 这类无分隔的长数字)一律拒绝 _ => false, } } /// 从请求体参数里取出协议 `id`(动作目标),并从 `params` 中**摘除**。 /// /// - 取不到(柜级指令本就不带 `id`)→ `None` /// - 由调用方显式给出:单板 `reset` → `"01"`;仓体级 `on`/`off` → `"01-3"` /// - **绝不允许再用 IMEI 当 `id`**:下行路由靠内部字段 /// [`ROUTE_DEV_ID_FIELD`](`_dev_id`),协议 `id` 与设备身份无关 /// /// 为什么必须摘除:`DeviceCommand::params` 是 `#[serde(flatten)]`, /// 若 `params` 里仍带 `id`,序列化时会与结构体显式 `id` 字段互相覆盖, /// 可能把未校验的值(甚至非字符串)写进下行报文。 fn take_command_target(params: &mut Option) -> Result, AppError> { let raw = params .as_mut() .and_then(|p| p.as_object_mut()) .and_then(|obj| obj.remove("id")); match raw { None | Some(Value::Null) => Ok(None), Some(Value::String(s)) => { if !is_valid_command_target(&s) { return Err(AppError::BadRequest(format!( "指令目标 id 非法: {}(应形如 \"01\" 或 \"01-3\",且不含 IMEI)", s ))); } Ok(Some(s)) } Some(_) => Err(AppError::BadRequest("指令目标 id 必须是字符串".into())), } } /// POST /api/devices/:dev_id/command — 向指定设备下发指令 /// /// 安全链路(C2 加固,缺一不可): /// 1. `dev_id`(IMEI) 反查 `cabinets.id`,查不到 → `NotFound` /// 2. **组织归属校验** `verify_cabinet_ownership`(与 `routes/cabinets.rs` 同一套隔离规则; /// 总管理员 role_level=2 豁免) /// 3. 设备禁用校验 /// 4. `act` 白名单(见 [`CABINET_COMMAND_ACTS`]) /// 5. 设备必须在线(与 `/api/h5/charge/start` 一致,复用 `NodeRegistry::is_device_online`) /// 6. 协议 `id` 只表示动作目标、由调用方显式给出,**绝不再填 IMEI** pub async fn send_command( user: CurrentUser, State(state): State, Path(dev_id): Path, Json(mut req): Json, ) -> Result, AppError> { auth::check_permission(&user, "device:operate")?; // 1. IMEI → 柜子ID(归属校验必须落在 cabinets.id 上,故先反查) let cabinet_id: Option = sqlx::query_scalar("SELECT id FROM cabinets WHERE imei = ?") .bind(&dev_id) .fetch_optional(&state.mysql) .await .map_err(AppError::Database)?; let cabinet_id = cabinet_id.ok_or_else(|| AppError::NotFound(format!("设备 {} 不存在", dev_id)))?; // 2. 组织数据隔离:非总管理员只能操作本组织项目下的柜子 crate::routes::scope_guard::verify_cabinet_ownership(&state.mysql, &user, cabinet_id).await?; // 3. 设备禁用校验 let disabled: bool = sqlx::query_scalar( r#"SELECT COUNT(*) > 0 FROM cabinets WHERE id = ? AND status = 2"#, ) .bind(cabinet_id) .fetch_one(&state.mysql) .await .map_err(AppError::Database)?; if disabled { return Err(AppError::BadRequest("设备已被禁用,无法下发指令".into())); } // 4. `act` 白名单 if !CABINET_COMMAND_ACTS.contains(&req.act.as_str()) { return Err(AppError::BadRequest(format!( "不支持的指令 act={},本接口仅允许: {}", req.act, CABINET_COMMAND_ACTS.join("/") ))); } // 5. 设备必须在线(平台对不在线的设备一律不投递) if !state.nodes.is_device_online(&dev_id).await { return Err(AppError::BadRequest("设备不在线".into())); } // 6. 协议 `id`(动作目标):只从请求体显式取,**不再把 IMEI 塞进去** let id = take_command_target(&mut req.params)?; // 仓体级动作(on/off)必须带目标,否则设备侧无法定位仓体(协议要求 `id` = 仓控板-通道) if matches!(req.act.as_str(), "on" | "off") && id.is_none() { return Err(AppError::BadRequest(format!( "指令 act={} 必须显式给出目标 id(仓控板-通道,如 \"01-3\")", req.act ))); } let cmd = DeviceCommand { act: req.act, msg_id: req.msg_id, id, params: req.params }; send_device_command(&state.redis, &dev_id, &cmd) .await .map_err(AppError::Internal)?; tracing::info!( "[command] 已推入队列 dev_id={} cabinet_id={} act={} id={:?}", dev_id, cabinet_id, cmd.act, cmd.id ); Ok(Json(json!({ "suc": 1 }))) } /// GET /api/devices/stats — 在线设备统计(**仅总管理员**) /// /// 返回的是**全平台**设备服务节点数(`node_count`),属内部设施信息: /// 原先只校验 `device:view`,任何客户账号登录后都能读到"我们有多少台设备服务节点"。 /// 与 `/api/nodes/*` 同一口径,收紧为 `role_level >= 2`(内部管理面,不对客户开放)。 pub async fn device_stats( user: CurrentUser, State(state): State, ) -> Result, AppError> { auth::require_super_admin(&user)?; let nodes = state.nodes.list().await; Ok(Json(json!({ "node_count": nodes.len(), }))) } // --------------------------------------------------------------------------- // 单元测试:C2 协议 `id`(动作目标)的校验与摘除 // --------------------------------------------------------------------------- #[cfg(test)] mod tests { use super::*; use serde_json::json; /// 合法形态:一段 = 单仓控板(`"01"`)、两段 = 仓体级(`"01-3"`),每段 1~2 位数字 #[test] fn test_is_valid_command_target_accepts_board_and_compartment() { for ok in ["01", "1", "6", "01-3", "1-3", "12-34"] { assert!(is_valid_command_target(ok), "{} 应为合法目标", ok); } } /// IMEI(15 位纯数字)与其它畸形值一律非法 —— 协议 `id` **不含 IMEI** #[test] fn test_is_valid_command_target_rejects_imei_and_malformed() { for bad in ["864720562557021", "", "-", "01-", "-3", "01-3-2", "0a", "123", "01 -3"] { assert!(!is_valid_command_target(bad), "{} 应为非法目标", bad); } } /// 显式给出 `id` 时取出,并把它从 `params` 中摘除 /// (否则 `#[serde(flatten)]` 会在序列化时覆盖结构体的显式 `id` 字段) #[test] fn test_take_command_target_extracts_and_removes_id() { let mut params = Some(json!({ "id": "01-3", "duration": 60 })); let id = take_command_target(&mut params).unwrap(); assert_eq!(id.as_deref(), Some("01-3")); let obj = params.as_ref().and_then(|p| p.as_object()).unwrap(); assert!(!obj.contains_key("id"), "id 必须从 params 摘除"); assert_eq!(obj["duration"], 60, "其它参数必须原样保留"); } /// 柜级指令不带 `id` → `None`(**绝不回填 IMEI**) #[test] fn test_take_command_target_absent_returns_none() { assert_eq!(take_command_target(&mut None).unwrap(), None); assert_eq!(take_command_target(&mut Some(json!({ "id": null }))).unwrap(), None); assert_eq!(take_command_target(&mut Some(json!({ "board": 1 }))).unwrap(), None); } /// 非法目标(IMEI / 非字符串)必须报错,不得下发 #[test] fn test_take_command_target_rejects_invalid_id() { let mut imei = Some(json!({ "id": "864720562557021" })); assert!(matches!(take_command_target(&mut imei), Err(AppError::BadRequest(_)))); let mut numeric = Some(json!({ "id": 3 })); assert!(matches!(take_command_target(&mut numeric), Err(AppError::BadRequest(_)))); } /// **跨服务契约**:连接级认证标记的键必须与 `device-server` 的 `auth_marker_key` /// 逐字一致(那边同样有 `test_auth_marker_key_format_contract` 守门)—— /// 不一致的后果是"所有设备都被判为不在线、指令全下发不出去"。 #[test] fn test_auth_marker_key_matches_device_server_contract() { assert_eq!(AUTH_MARKER_PREFIX, "device:authed:"); assert_eq!(auth_marker_key("864720562557021"), "device:authed:864720562557021"); // 不能退化成旧的在线键(那正是"假在线"的根因) assert_ne!(auth_marker_key("864720562557021"), "device:864720562557021"); } /// `/api/devices/stats`(全平台节点数)属内部设施面:只有总管理员可读 #[test] fn test_device_stats_requires_super_admin() { let make = |role_level: i32| CurrentUser { user_id: 1, phone: "13800138000".into(), permissions: std::collections::HashSet::new(), role_level, organization_id: Some(7), scope: crate::middleware::auth::UserScope::default(), }; assert!(auth::require_super_admin(&make(2)).is_ok(), "总管理员放行"); assert!(auth::require_super_admin(&make(1)).is_err(), "企业管理员必须被拒"); assert!(auth::require_super_admin(&make(0)).is_err(), "操作员必须被拒"); } }