//! 审计日志(P5):IM 消息 ↔ TunnelCommand 全量记录(SQLite + WAL,与 SessionStore 同机库文件) //! //! 设计口径: //! - **append-only**:只写不改,无 UPDATE/DELETE——审计的公信力来自不可篡改。 //! - **双向打点**:inbound(IM 用户 → 网关)与 outbound(网关/桌面端 → IM)各记一行, //! 同一 request_id 关联请求与响应(排障时按 id 串链)。 //! - **失败不阻断**:审计写入失败仅 tracing::warn,绝不影响主链路(审计是增益非闸门)。 //! - 表:tunnel_audit(id 自增 / ts 毫秒 / direction / platform / chat_id / user_id / //! conv_id / action / request_id / payload 截断 4KB / outcome ok|error|denied)。 //! payload 截断防长消息撑爆库;敏感凭证不经过此层(config 层已隔离)。 //! //! 查询接口(运维用):recent 按时间倒序限量;by_request 串同请求双向链。 use std::path::Path; use std::sync::Arc; use rusqlite::Connection; const MAX_PAYLOAD: usize = 4096; /// 审计方向 #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Direction { /// IM 用户 → 网关(入站消息/审批点击/新会话) Inbound, /// 网关/桌面端 → IM(流式输出/审批卡片/通知) Outbound, /// 网关 → 桌面端(TunnelCommand 下发) Command, } impl Direction { fn as_str(self) -> &'static str { match self { Direction::Inbound => "inbound", Direction::Outbound => "outbound", Direction::Command => "command", } } } #[derive(Clone)] pub struct AuditLog { conn: Arc>, } /// 打点参数(字段可选,None 写空串,保持列 NOT NULL 简化查询) pub struct AuditEntry { pub direction: Direction, pub platform: String, pub chat_id: String, pub user_id: String, pub conv_id: String, /// 动作名:message/new_session/approval_click/approval_card/stream/completed/notice/send_cmd/approve_cmd pub action: String, /// 关联 ID:入站↔出站串链用(消息场景可用 chat+ts 派生,审批场景用 approval_id) pub request_id: String, pub payload: String, /// ok / error / denied(白名单拒/无 channel) pub outcome: String, } impl AuditLog { /// 打开(或创建)审计库。默认与 sessions.db 同目录独立文件(隔离写放大)。 pub fn open(path: &Path) -> anyhow::Result { let conn = Connection::open(path)?; conn.pragma_update(None, "journal_mode", "WAL")?; conn.execute_batch( "CREATE TABLE IF NOT EXISTS tunnel_audit ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, direction TEXT NOT NULL, platform TEXT NOT NULL DEFAULT '', chat_id TEXT NOT NULL DEFAULT '', user_id TEXT NOT NULL DEFAULT '', conv_id TEXT NOT NULL DEFAULT '', action TEXT NOT NULL, request_id TEXT NOT NULL DEFAULT '', payload TEXT NOT NULL DEFAULT '', outcome TEXT NOT NULL DEFAULT 'ok' ); CREATE INDEX IF NOT EXISTS idx_audit_ts ON tunnel_audit(ts); CREATE INDEX IF NOT EXISTS idx_audit_request ON tunnel_audit(request_id);", )?; Ok(Self { conn: Arc::new(std::sync::Mutex::new(conn)) }) } /// 内存库(测试) #[allow(dead_code)] pub fn open_in_memory() -> anyhow::Result { let conn = Connection::open_in_memory()?; conn.execute_batch( "CREATE TABLE IF NOT EXISTS tunnel_audit ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, direction TEXT NOT NULL, platform TEXT NOT NULL DEFAULT '', chat_id TEXT NOT NULL DEFAULT '', user_id TEXT NOT NULL DEFAULT '', conv_id TEXT NOT NULL DEFAULT '', action TEXT NOT NULL, request_id TEXT NOT NULL DEFAULT '', payload TEXT NOT NULL DEFAULT '', outcome TEXT NOT NULL DEFAULT 'ok');", )?; Ok(Self { conn: Arc::new(std::sync::Mutex::new(conn)) }) } fn now_ms() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_millis() as i64) .unwrap_or(0) } /// 追加一条审计(失败仅告警不冒泡) pub fn record(&self, e: AuditEntry) { let payload = if e.payload.len() > MAX_PAYLOAD { format!("{}…[截断]", &e.payload[..MAX_PAYLOAD]) } else { e.payload }; let result = (|| -> anyhow::Result<()> { let conn = self.conn.lock().map_err(|err| anyhow::anyhow!("audit 锁中毒: {err}"))?; conn.execute( "INSERT INTO tunnel_audit (ts, direction, platform, chat_id, user_id, conv_id, action, request_id, payload, outcome) \ VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)", rusqlite::params![ Self::now_ms(), e.direction.as_str(), e.platform, e.chat_id, e.user_id, e.conv_id, e.action, e.request_id, payload, e.outcome, ], )?; Ok(()) })(); if let Err(err) = result { tracing::warn!("[audit] 写入失败(不影响主链路): {err}"); } } /// 最近 N 条(时间倒序,运维排障) #[allow(dead_code)] pub fn recent(&self, limit: usize) -> anyhow::Result> { let conn = self.conn.lock().map_err(|e| anyhow::anyhow!("audit 锁中毒: {e}"))?; let mut stmt = conn.prepare( "SELECT ts, direction, platform, chat_id, user_id, conv_id, action, request_id, outcome \ FROM tunnel_audit ORDER BY id DESC LIMIT ?1", )?; let rows = stmt.query_map([limit as i64], |row| { Ok(format!( "{} [{}] {} plat={} chat={} user={} conv={} req={} outcome={}", row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(6)?, row.get::<_, String>(2)?, row.get::<_, String>(3)?, row.get::<_, String>(4)?, row.get::<_, String>(5)?, row.get::<_, String>(7)?, row.get::<_, String>(8)?, )) })?; Ok(rows.filter_map(|r| r.ok()).collect()) } /// 按 request_id 串双向链(入站→命令→出站全链) #[allow(dead_code)] pub fn by_request(&self, request_id: &str) -> anyhow::Result> { let conn = self.conn.lock().map_err(|e| anyhow::anyhow!("audit 锁中毒: {e}"))?; let mut stmt = conn.prepare( "SELECT ts, direction, action, outcome FROM tunnel_audit WHERE request_id = ?1 ORDER BY id ASC", )?; let rows = stmt.query_map([request_id], |row| { Ok(format!( "{} [{}] {} outcome={}", row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, String>(3)?, )) })?; Ok(rows.filter_map(|r| r.ok()).collect()) } } #[cfg(test)] mod tests { use super::*; fn entry(direction: Direction, action: &str, req: &str, outcome: &str) -> AuditEntry { AuditEntry { direction, platform: "feishu".into(), chat_id: "oc_123".into(), user_id: "u_1".into(), conv_id: "feishu:oc_123".into(), action: action.into(), request_id: req.into(), payload: "hello".into(), outcome: outcome.into(), } } #[test] fn record_and_recent_roundtrip() { let log = AuditLog::open_in_memory().unwrap(); log.record(entry(Direction::Inbound, "message", "req-1", "ok")); log.record(entry(Direction::Command, "send_cmd", "req-1", "ok")); log.record(entry(Direction::Outbound, "stream", "req-1", "ok")); log.record(entry(Direction::Inbound, "message", "req-2", "denied")); let rows = log.recent(10).unwrap(); assert_eq!(rows.len(), 4); // 倒序:最新在前 assert!(rows[0].contains("req-2")); assert!(rows[0].contains("denied")); let chain = log.by_request("req-1").unwrap(); assert_eq!(chain.len(), 3); } #[test] fn payload_truncated() { let log = AuditLog::open_in_memory().unwrap(); let mut e = entry(Direction::Inbound, "message", "req-x", "ok"); e.payload = "x".repeat(10_000); log.record(e); let conn_count = |log: &AuditLog| -> i64 { let conn = log.conn.lock().unwrap(); conn.query_row("SELECT COUNT(*) FROM tunnel_audit", [], |r| r.get(0)).unwrap() }; assert_eq!(conn_count(&log), 1); let conn = log.conn.lock().unwrap(); let stored: String = conn .query_row("SELECT payload FROM tunnel_audit", [], |r| r.get(0)) .unwrap(); assert!(stored.len() < 5_000); assert!(stored.ends_with("…[截断]")); } #[test] fn record_never_panics_on_poisoned_lock() { let log = AuditLog::open_in_memory().unwrap(); // 毒锁:持锁 panic 使 Mutex poisoned,record 应 warn 而非 panic let guard = log.conn.lock().unwrap(); std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { std::panic::panic_any("poison"); })) .err(); drop(guard); log.record(entry(Direction::Inbound, "message", "req-p", "ok")); // 不 panic 即通过 } }