futu_mcp/state/

push_filter.rs

/// v1.4.105 T-C2: 从 daemon push 的 raw body 里解 `(acc_id, trd_market)`，
/// 用于按 caller key `allowed_markets` 白名单过滤 trade push event。
///
/// 只处理 trade push（TRD_UPDATE_ORDER / TRD_UPDATE_ORDER_FILL），proto Header
/// 同时含 `acc_id` (u64) 与 `trd_market` (i32 enum)。返 `Some((acc_id, trd_market))`
/// 仅当 decode 成功且 s2c.header 存在.
///
/// 行情 push (QOT_UPDATE_*) 不含 trd_market 概念 → 返 None → 调用方按
/// `event_acc_id=None` 路径不做 acc/market gate, 直接 broadcast.
pub(crate) fn extract_acc_id_and_market_from_push(
    proto_id: u32,
    body: &[u8],
) -> Option<(u64, i32)> {
    use prost::Message;
    match proto_id {
        TRD_UPDATE_ORDER_PROTO_ID => {
            let rsp = match futu_proto::trd_update_order::Response::decode(body) {
                Ok(rsp) => rsp,
                Err(e) => {
                    tracing::debug!(
                        proto_id,
                        body_len = body.len(),
                        error = %e,
                        "MCP trade push body decode failed while extracting acc_id"
                    );
                    return None;
                }
            };
            let h = rsp.s2c?.header;
            Some((h.acc_id, h.trd_market))
        }
        TRD_UPDATE_ORDER_FILL_PROTO_ID => {
            let rsp = match futu_proto::trd_update_order_fill::Response::decode(body) {
                Ok(rsp) => rsp,
                Err(e) => {
                    tracing::debug!(
                        proto_id,
                        body_len = body.len(),
                        error = %e,
                        "MCP trade push body decode failed while extracting acc_id"
                    );
                    return None;
                }
            };
            let h = rsp.s2c?.header;
            Some((h.acc_id, h.trd_market))
        }
        _ => None,
    }
}

/// v1.4.106 codex 0932 F6/F7: trade push proto_ids (set membership 测试).
const TRD_UPDATE_ORDER_PROTO_ID: u32 = 2208;
const TRD_UPDATE_ORDER_FILL_PROTO_ID: u32 = 2218;

/// v1.4.106 codex 0932 F6 [P2]: 仅按 proto_id 判断是否 trade push 类.
///
/// 修 F6 silent-misclassify bug: 旧实装靠 `extract_acc_id_and_market_from_push`
/// 返 None 推断 "非 trade" — 但 trade push (proto_id 2208/2218) 若 body decode
/// 失败也返 None → 误归 quote → restricted key (有 allowed_acc_ids 限额) 看到
/// 该 push 时按 quote 路径直接 broadcast (绕过 trade-market filter 的 ACL).
///
/// **正确语义**: event_type 由 proto_id 决定 (set membership), 与 body 完整性
/// 无关. body 解码状态另用 `decode_status` 字段表达 (F7).
pub(crate) fn is_trade_push_proto_id(proto_id: u32) -> bool {
    matches!(
        proto_id,
        TRD_UPDATE_ORDER_PROTO_ID | TRD_UPDATE_ORDER_FILL_PROTO_ID
    )
}

/// v1.4.106 codex 0932 F6 [P2]: trade push 的解码结果 + 分发决策.
#[derive(Debug, Clone)]
pub(crate) enum TradePushDecode {
    /// proto_id 不在 trade set — 非 trade push (quote / notify / 其他).
    NotTrade,
    /// trade push body decode 成功 — 含 (acc_id, trd_market).
    Decoded { acc_id: u64, trd_market: i32 },
    /// trade push body decode 失败 — caller 必须按 trade 语义处理 (event_type="trade")
    /// 但无 acc_id / market gate 信息. restricted key 应 drop, unrestricted
    /// 应透传带 `decode_status="failed"`.
    DecodeFailed,
}

pub(crate) fn classify_trade_push(proto_id: u32, body: &[u8]) -> TradePushDecode {
    if !is_trade_push_proto_id(proto_id) {
        return TradePushDecode::NotTrade;
    }
    match extract_acc_id_and_market_from_push(proto_id, body) {
        Some((acc_id, trd_market)) => TradePushDecode::Decoded { acc_id, trd_market },
        None => TradePushDecode::DecodeFailed,
    }
}

/// v1.4.105 T-C2: `Trd_Common.TrdMarket` enum int → 字符串 (与 `keys.json`
/// 配 `allowed_markets` 中字符串一致). 实际映射由 `futu_trd::market`
/// 统一维护，避免 MCP push 过滤、CLI 展示和 trade read projection 漂移.
///
/// 来源: `Trd_Common.proto::TrdMarket` enum.
pub(crate) fn trd_market_int_to_str(i: i32) -> &'static str {
    futu_trd::market::trd_market_label(i).unwrap_or("")
}

/// v1.4.39 Phase 5 filter 核心决策（pure function，便于单测）：
///
/// 判断 push 是否应推给某个订阅者。2 层 gate：
/// 1. **subscriber acc_ids**：`futu_sub_acc_push` 参数里用户指定的 acc_id 列表
///    （空集 = subscribe-all，即不做 subscriber 级过滤）
/// 2. **key allowed_acc_ids** (v1.4.39)：注册时快照的 per-key 白名单。defense-in-depth
///    层，防止 agent 订阅了 key 无权限的 acc_id（主 auth 在 guard.rs tool 调用
///    时 enforce，但 push 是服务端主动发起绕过 tool 调用）
///
/// 两层都 pass 才推。行情类 push (`push_acc_id=None`) 跳过所有 acc_id gate。
/// v1.4.58 MED-NEW-3（2nd code review）: summary 过滤决策 pure function。
///
/// 用于 `push_subscribers_summary` 的 cross-tenant filter —— scope mode 下 caller
/// 只能看到**自己 allowed_acc_ids 有交集的订阅**（防止跨租户泄漏其他 agent 的
/// acc_ids）。
///
/// 规则：
/// - `caller_allowed = None` 或空集（legacy / unrestricted）→ 全可见
/// - 被查订阅的 `sub_acc_ids` 空集（subscribe-all）→ 全可见（没有 specific 账户可泄漏）
/// - 否则 → 只要 `sub_acc_ids` 与 `caller_allowed` 有**任一交集** → 可见
pub(crate) fn subscriber_visible_to_caller(
    sub_acc_ids: &std::collections::HashSet<u64>,
    caller_allowed: Option<&std::collections::HashSet<u64>>,
) -> bool {
    match caller_allowed {
        Some(allowed) if !allowed.is_empty() => {
            sub_acc_ids.is_empty() || sub_acc_ids.iter().any(|a| allowed.contains(a))
        }
        _ => true,
    }
}

/// v1.4.105 T-C2: 生产路径已 migrate 到 `subscriber_should_receive_with_market`
/// (含 market gate Layer 3). 本 helper 保留作向后兼容入口 (老 6 个 filter_*
/// pure-fn 测仍引用), 不入生产路径.
///
/// **v1.4.105 F5 fix**: production migrated to `FilterRegistry::should_drop_event`,
/// 本 fn 跟 `subscriber_should_receive_with_market` 同 `#[cfg(test)]` gate 防
/// dead_code 警告.
#[cfg(test)]
pub(crate) fn subscriber_should_receive(
    push_acc_id: Option<u64>,
    sub_acc_ids: &std::collections::HashSet<u64>,
    key_allowed_acc_ids: Option<&std::collections::HashSet<u64>>,
) -> bool {
    // v1.4.105 T-C2: 旧 API 委托新 API (无 market 信息 → 不做 market gate).
    // 保留为向后兼容入口 (单测仍以 sub_acc_ids + allowed_acc_ids 双 gate 模式验证).
    subscriber_should_receive_with_market(
        push_acc_id,
        None, // 无 market context → 不做 market gate
        sub_acc_ids,
        key_allowed_acc_ids,
        None, // 无 allowed_markets snapshot → 不做 market gate
    )
}

/// v1.4.105 T-C2: subscriber_should_receive 的扩展版 — 加 market gate (Layer 3).
///
/// 完整 3-layer trade push filter:
/// 1. **subscriber acc_ids** (`futu_sub_acc_push` 参数): 空 = subscribe-all
/// 2. **key allowed_acc_ids** (caller key 注册时快照): 硬白名单
/// 3. **key allowed_markets** (caller key 注册时快照, v1.4.105 加): 硬白名单
///
/// 三 gate 都 pass 才推. `event_trd_market=None` (decode 失败 / 行情 push) 跳过
/// market gate. `key_allowed_markets=None` / 空 = 不做 market gate (stdio /
/// legacy / 未配 allowed_markets 的 key).
///
/// 行情类 push (`push_acc_id=None`) 仍按 v1.4.39 行为全推 — 行情无 acc / market
/// 概念, 不参与 trade event filter.
///
/// **v1.4.105 F5 fix (codex review C4)**: production 路径已 migrate 到
/// `FilterRegistry::should_drop_event` (跟 4 surface 一致). 本 fn 现仅作
/// `#[cfg(test)]` unit test target — 直接验证 logic, 不再 production 调用.
/// 保留 + cfg-gate 让 12 既有 test 继续 lock 行为契约 (logic 等价两套断言:
/// 这里 pure-fn assert + cross_surface_invariants FilterRegistry 路径 assert).
#[cfg(test)]
pub(crate) fn subscriber_should_receive_with_market(
    push_acc_id: Option<u64>,
    event_trd_market: Option<i32>,
    sub_acc_ids: &std::collections::HashSet<u64>,
    key_allowed_acc_ids: Option<&std::collections::HashSet<u64>>,
    key_allowed_markets: Option<&std::collections::HashSet<String>>,
) -> bool {
    let Some(aid) = push_acc_id else {
        return true; // 非 trade push，全推（行情类无 key 级 gate 概念）
    };
    // Layer 1: 订阅者 acc_ids 过滤
    let sub_gate = sub_acc_ids.is_empty() || sub_acc_ids.contains(&aid);
    // Layer 2: caller key allowed_acc_ids 硬白名单
    let key_acc_gate = match key_allowed_acc_ids {
        Some(allowed) if !allowed.is_empty() => allowed.contains(&aid),
        _ => true, // 无 key 级约束（allowed_acc_ids=None 或空）/ stdio 模式
    };
    // Layer 3 (v1.4.105 T-C2): caller key allowed_markets 硬白名单
    //
    // 行为契约:
    // - `key_allowed_markets=None` 或空集 → 不做 market gate (向后兼容 + stdio)
    // - `event_trd_market=None` → 不做 market gate (decode 失败保守 — 不 silent
    //   drop, 让 caller 看到原始 push, downstream 单测覆盖 decode 失败路径)
    // - `event_trd_market=Some(int)` → 转字符串与 allowed 集合比对
    //   (注: int=0/未知 → trd_market_int_to_str 返 "" → 不在任何非空 allowed
    //    集合内 → drop. 这是 fail-closed 语义: 未知 market 配 restricted key
    //    应 drop 而非 silently leak)
    let key_market_gate = match key_allowed_markets {
        Some(allowed) if !allowed.is_empty() => match event_trd_market {
            Some(int) => allowed.contains(trd_market_int_to_str(int)),
            None => true, // decode 失败 → 不 drop (向后兼容 — 单独 metric 提示)
        },
        _ => true, // 无 market 限制
    };
    sub_gate && key_acc_gate && key_market_gate
}