DevFlow/crates/df-im-gateway/src/audit.rs

//! 审计日志(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<std::sync::Mutex<Connection>>,
}

/// 打点参数(字段可选,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<Self> {
        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<Self> {
        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<Vec<String>> {
        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<Vec<String>> {
        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 即通过
    }
}