futu_auth/

metrics.rs

//! 全局 metrics registry —— 供 `/metrics` Prometheus 端点消费
//!
//! ## 设计
//!
//! - 全局 `OnceLock<Arc<Registry>>`，进程初始化时 `Registry::install()` 一次
//! - [`audit::reject`] / [`audit::allow`] / [`audit::trade`] 在写日志的同时
//!   增 counter
//! - [`RuntimeCounters::check_and_commit`] 拒时也会通知 metrics（限额 hit）
//! - `Registry::render_prometheus()` 输出 Prometheus text exposition 格式
//!
//! ## 为什么不用 `prometheus` crate
//!
//! 输出格式简单（几类 counter），手写避免再引一份依赖。真要升级到更复杂的
//! metric（histogram / summary）再换。
//!
//! ## 维度
//!
//! 所有 counter 的 label：`iface`（grpc/rest/ws/mcp）+ `outcome`
//! （allow/reject/success/failure/unknown）+ `key_id`（未配 key 时 `<none>`）。
//! 维度超细会爆 cardinality，但 `key_id` 上限就是 keys.json 里的条数（几十级），
//! 可接受。

use std::sync::{Arc, OnceLock};

use dashmap::DashMap;
use parking_lot::RwLock;
use sha2::{Digest, Sha256};

/// 事件计数：(iface, endpoint_or_tool 这里不放 —— 放会爆维度, outcome, key_id) → count
type EventKey = (String, String, String);

/// 限额拒 counter：(iface, key_id, reason_category) → count
///
/// reason_category 只枚举大类：rate / daily / per_order / market / symbol /
/// side / hours / other，字符串固定不爆维度
type LimitRejectKey = (String, String, String);

/// v1.4.90 P1-B: extension renderer hook —— 让其他 crate（如 futu-server）
/// 把自己持有的 [`GatewayMetrics`]、push_health 等 atomic counter 渲染成
/// Prometheus text 追加到 `/metrics`。
///
/// **架构**：`futu-auth` 不能依赖 `futu-server`（反向 dep cycle），所以反过来
/// 由 futu-server / futu-gateway 在启动时通过 [`Registry::register_renderer`]
/// 注册一个闭包；`Registry::render_prometheus` 末尾调用所有注册的 renderer
/// 并把它们的输出拼接进总 body。
///
/// **要求**：
/// - 闭包必须返回**完整**的 Prometheus text exposition 段（含 `# HELP` /
///   `# TYPE` / 数据行 / 末尾 `\n`），输出会**原样**拼接
/// - 闭包应该 idempotent / O(1)~O(N counters) ——`/metrics` 抓取频率 15s 起
type ExtraRenderer = Arc<dyn Fn() -> String + Send + Sync>;

#[derive(Default)]
pub struct Registry {
    /// auth / trade 事件：(iface, outcome, key_id) → count
    events: DashMap<EventKey, u64>,
    /// 限额拒：(iface, key_id, reason_cat) → count
    limit_rejects: DashMap<LimitRejectKey, u64>,
    /// WS 按 scope 过滤掉的推送：(required_scope, client_key_id) → count
    ///
    /// 例：`(trade, key_bot_a)` 多 = bot_a 只有 qot:read，但服务端在推 trade 事件
    ws_filtered: DashMap<(String, String), u64>,
    /// v1.4.90 P1-B: extension renderers —— 见 [`ExtraRenderer`] 文档
    extra_renderers: RwLock<Vec<ExtraRenderer>>,
}

impl std::fmt::Debug for Registry {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Registry")
            .field("events", &self.events.len())
            .field("limit_rejects", &self.limit_rejects.len())
            .field("ws_filtered", &self.ws_filtered.len())
            .field("extra_renderers", &self.extra_renderers.read().len())
            .finish()
    }
}

impl Registry {
    /// 记录一次 auth/trade 事件
    pub fn record_event(&self, iface: &str, outcome: &str, key_id: &str) {
        let key = (iface.to_string(), outcome.to_string(), key_id.to_string());
        *self.events.entry(key).or_insert(0) += 1;
    }

    /// 记录一次限额拒（reason_cat 见文档）
    pub fn record_limit_reject(&self, iface: &str, key_id: &str, reason_cat: &str) {
        let key = (
            iface.to_string(),
            key_id.to_string(),
            reason_cat.to_string(),
        );
        *self.limit_rejects.entry(key).or_insert(0) += 1;
    }

    /// 记录 WS 因 scope 不足被过滤掉的一条推送
    pub fn record_ws_filtered(&self, required_scope: &str, client_key_id: &str) {
        let key = (required_scope.to_string(), client_key_id.to_string());
        *self.ws_filtered.entry(key).or_insert(0) += 1;
    }

    /// v1.4.90 P1-B: 注册一个 Prometheus extension renderer —— 见
    /// [`ExtraRenderer`] 文档.
    ///
    /// 调用方：futu-server / futu-gateway 启动时调一次, 把自己持有的
    /// `GatewayMetrics` (per-cmd push counter / 24-hour breakdown / push_health
    /// circuit breaker 等) 渲染成 Prometheus text 追加到 `/metrics`.
    ///
    /// 多次注册会保留所有 renderer; 顺序与注册顺序一致.
    pub fn register_renderer<F>(&self, renderer: F)
    where
        F: Fn() -> String + Send + Sync + 'static,
    {
        self.extra_renderers.write().push(Arc::new(renderer));
    }

    /// Prometheus text exposition 格式输出.
    ///
    /// **v1.4.106 codex 0542 F1 [P2 SECURITY]**: `key_id` label 默认 redact 为
    /// `kh_<8hex>` (短 SHA256). 设 `FUTU_METRICS_PUBLIC=1` env 回退明文.
    /// 见 [`redact_key_id`] / `Scope::MetricsRead`.
    #[must_use]
    pub fn render_prometheus(&self) -> String {
        let mut s = String::with_capacity(4096);

        s.push_str("# HELP futu_auth_events_total Auth / trade events by iface, outcome, key_id (redacted unless FUTU_METRICS_PUBLIC=1)\n");
        s.push_str("# TYPE futu_auth_events_total counter\n");
        for kv in self.events.iter() {
            let (iface, outcome, key_id) = kv.key();
            let v = *kv.value();
            s.push_str(&format!(
                "futu_auth_events_total{{iface={},outcome={},key_id={}}} {}\n",
                prom_label(iface),
                prom_label(outcome),
                prom_label(&redact_key_id(key_id)),
                v
            ));
        }

        s.push_str(
            "# HELP futu_auth_limit_rejects_total Limit-check rejects by iface, key_id (redacted unless FUTU_METRICS_PUBLIC=1), reason\n",
        );
        s.push_str("# TYPE futu_auth_limit_rejects_total counter\n");
        for kv in self.limit_rejects.iter() {
            let (iface, key_id, reason) = kv.key();
            let v = *kv.value();
            s.push_str(&format!(
                "futu_auth_limit_rejects_total{{iface={},key_id={},reason={}}} {}\n",
                prom_label(iface),
                prom_label(&redact_key_id(key_id)),
                prom_label(reason),
                v
            ));
        }

        s.push_str(
            "# HELP futu_ws_filtered_pushes_total Pushes filtered out by reason (PushDropReason 8 variants), client key_id redacted unless FUTU_METRICS_PUBLIC=1\n",
        );
        s.push_str("# TYPE futu_ws_filtered_pushes_total counter\n");
        for kv in self.ws_filtered.iter() {
            let (reason, key_id) = kv.key();
            let v = *kv.value();
            // v1.4.106 codex 0542 F4 [P3]: label name v1.4.105 → reason
            // (was: required_scope). 与 PushDropReason::as_metric_label() 一致.
            s.push_str(&format!(
                "futu_ws_filtered_pushes_total{{reason={},key_id={}}} {}\n",
                prom_label(reason),
                prom_label(&redact_key_id(key_id)),
                v
            ));
        }

        // v1.4.90 P1-B: append extension renderer 输出 (per-cmd / per-hour
        // counter from GatewayMetrics / push_health 等). 见 ExtraRenderer 文档.
        //
        // **v1.4.106 codex 0542 F3 [P3]**: 每个 chunk 过 [`extension_redaction_guard`]
        // 校验 — 防外部 crate 注册的 renderer 不小心 emit 明文 key_id label
        // (e.g. `key_id="bot-prod-1"` 而不是 `kh_<8hex>`). 命中 →
        // all builds: log warning + drop chunk. `/metrics` must remain available
        // even in debug/test daemons; the guard still prevents plaintext leakage.
        let extras: Vec<ExtraRenderer> = self.extra_renderers.read().iter().cloned().collect();
        for renderer in extras {
            let chunk = renderer();
            if chunk.is_empty() {
                continue;
            }
            if let Err(violation) = extension_redaction_guard(&chunk) {
                tracing::warn!(
                    target: "futu_audit",
                    violation = %violation,
                    chunk_preview = %chunk.chars().take(200).collect::<String>(),
                    "v1.4.106 codex 0542 F3 [P3] redaction_guard: extension renderer \
                     emitted plaintext-looking key_id label, dropping chunk. Use \
                     futu_auth::metrics::redact_key_id(raw_key_id) before formatting."
                );
                continue; // drop the offending chunk
            }
            if !s.ends_with('\n') {
                s.push('\n');
            }
            s.push_str(&chunk);
        }

        s
    }
}

/// **v1.4.106 codex 0542 F3 [P3]**: redaction guard for extension renderer chunks.
///
/// 扫描一段 prometheus text exposition, 查找**明文 key_id label** 嫌疑模式:
///
/// `key_id="<value>"` 其中 `<value>` 既不是 [`redact_key_id`] 输出 (`kh_<8hex>`)
/// 也不是合法 sentinel (`<none>` / `<missing>` / `<invalid>`). 命中 → Err.
///
/// **不**扫其他 label (iface / outcome / reason 等), 只针对 `key_id=` 这条专门
/// 的 PII 通道 (= F1 主要修复对象). 其它 label 没有 cardinality enumeration
/// 风险 (是 fixed enum vocabulary).
///
/// 设计参考: regex 简单 byte-level 扫描足够 (extension renderer 输出量典型 <
/// 100 KB / scrape interval, 不是热路径). 不引 regex crate.
///
/// 返 `Err` = 字符串里有疑似明文 key_id, 含 first violation 的 label value 用
/// 作 panic / log message.
pub(crate) fn extension_redaction_guard(chunk: &str) -> Result<(), String> {
    // 扫描每行, 找 key_id="..." pattern. 简单 byte-level state machine.
    for (lineno, line) in chunk.lines().enumerate() {
        let mut search_from = 0;
        while let Some(idx) = line[search_from..].find("key_id=\"") {
            let start = search_from + idx + "key_id=\"".len();
            // 找配对的非转义 "
            let mut end = None;
            let bytes = line.as_bytes();
            let mut j = start;
            while j < bytes.len() {
                let c = bytes[j];
                if c == b'\\' && j + 1 < bytes.len() {
                    j += 2;
                    continue;
                }
                if c == b'"' {
                    end = Some(j);
                    break;
                }
                j += 1;
            }
            let Some(end) = end else {
                // 闭合 quote 找不到, 跳过这一行 (malformed exposition, 但不 trigger
                // F3 — 不是 PII 风险, 是格式问题)
                break;
            };
            let value = &line[start..end];
            if !is_acceptable_key_id_label(value) {
                return Err(format!(
                    "line {}: key_id={value:?} 不是 redact 形式 (kh_<8hex>) \
                     也不是合法 sentinel (<none>/<missing>/<invalid>)",
                    lineno + 1
                ));
            }
            search_from = end + 1;
        }
    }
    Ok(())
}

/// `value` 是否合法的 key_id label value (redact 形式 or sentinel).
fn is_acceptable_key_id_label(value: &str) -> bool {
    // sentinel
    if value == "<none>" || value == "<missing>" || value == "<invalid>" {
        return true;
    }
    // redact 形式: "kh_" + 8 hex (lower-case)
    if let Some(rest) = value.strip_prefix("kh_")
        && rest.len() == 8
        && rest
            .chars()
            .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase())
    {
        return true;
    }
    // FUTU_METRICS_PUBLIC=1 opt-out 路径: 明文 key_id 也合法 (运维明确选了不
    // redact). 不做 env 检查 (F3 是 dev-time guard, 不应 hot path query env;
    // production 关 redact 时 raw rendered output 已是明文, 不进 extension
    // renderer 路径).
    //
    // **设计取舍**: 这里**仍然**判 false 让 opt-out 用户的 extension renderer
    // 也走 redact_key_id. 这是 secure-by-default 的体现 — 即便用户 opt-out
    // 主 surface, extension renderer 依然走 redact 路径, 减少 PII 多点泄漏面.
    false
}

static GLOBAL: OnceLock<Arc<Registry>> = OnceLock::new();

/// 安装全局 registry。只会生效第一次 install；测试场景可能多次调用，
/// 后续 install 被忽略（保持 first-writer 语义）
pub fn install(reg: Arc<Registry>) {
    let _ = GLOBAL.set(reg);
}

/// 取全局 registry；未 install 时返回 None（调用方 no-op）
pub fn global() -> Option<Arc<Registry>> {
    GLOBAL.get().cloned()
}

/// **v1.4.106 codex 0542 F4 [P3]**: WS push drop 原因 closed-set enum (8 variants).
///
/// 把 4 surface (REST `/ws` / gRPC subscribe / raw TCP WS / MCP push) 散在
/// 各自代码里的 `bump_ws_filtered("foo", ...)` 字符串 label 收口为闭合 enum,
/// 防漂移. 任何**新增 drop 原因必须在 enum 加 variant**, 否则 caller 编译期
/// 不通过 — 强制同步 dashboard label.
///
/// 与 v1.4.105 之前的 `event.event_type.as_str()` 自由字符串方案相比:
///
/// - **Closed set**: enum match 编译期穷举, 不会有 typo `trade_market` →
///   `trade_markete` silent drop dashboard.
/// - **统一命名**: 跨 surface 同 reason 必同一 label (REST `trade_market` ==
///   gRPC `trade_market` == MCP `trade_market`, 不再各自 emit "trade" /
///   "trade_acc_id" 混淆 dashboard aggregation).
/// - **Future-safe**: 加新 reason 时 grep `PushDropReason::` 找全所有 caller,
///   不漏 surface.
///
/// **8 variants 设计依据** (v1.4.106 ship 时):
///
/// | variant | label | 触发场景 |
/// |---|---|---|
/// | `ScopeMissing` | `scope` | client 持 scope 不含 event 需要的 (e.g. event 是 trade push 但 client 仅 qot:read) |
/// | `QuoteScope` | `quote` | quote push scope 不满足 (legacy event_type label 兼容) |
/// | `TradeScope` | `trade` | trade push scope 不满足 (legacy 兼容) |
/// | `BroadcastScope` | `broadcast` | broadcast push scope 不满足 (legacy 兼容) |
/// | `TradeAccIdMismatch` | `trade_acc_id` | trade push event.acc_id ∉ key.allowed_acc_ids |
/// | `TradeMarketMismatch` | `trade_market` | trade push event.trd_market ∉ key.allowed_markets |
/// | `NotifyNotSubscribed` | `notify_unsub` | broadcast/notify push 但 client 未显式 sub (`IsConnSubRecvNotify`) |
/// | `TradeDecodeFailed` | `trade_decode_failed` | trade push body decode 失败, 防御性 drop |
///
/// `as_metric_label()` 返定 `&'static str` 与历史 dashboard label byte-identical
/// (向后兼容).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum PushDropReason {
    /// 通用 scope 不满足 (新加 — 替代各 surface 自定义 event_type label).
    ScopeMissing,
    /// Quote push scope 不满足 (legacy event_type=quote label 兼容).
    QuoteScope,
    /// Trade push scope 不满足 (legacy event_type=trade label 兼容).
    TradeScope,
    /// Broadcast push scope 不满足 (legacy event_type=broadcast label 兼容).
    BroadcastScope,
    /// Layer 1 — trade push event.acc_id ∉ key.allowed_acc_ids.
    TradeAccIdMismatch,
    /// Layer 3 — trade push event.trd_market ∉ key.allowed_markets.
    TradeMarketMismatch,
    /// notify_subscribe gate (`IsConnSubRecvNotify`) 未启用.
    NotifyNotSubscribed,
    /// Trade push body decode 失败 — 防御性 drop, 防 backend wire 错误推不可解的 push.
    TradeDecodeFailed,
}

impl PushDropReason {
    /// Prometheus `reason=` label 字符串 (8 固定桶).
    #[must_use]
    pub fn as_metric_label(&self) -> &'static str {
        match self {
            PushDropReason::ScopeMissing => "scope",
            PushDropReason::QuoteScope => "quote",
            PushDropReason::TradeScope => "trade",
            PushDropReason::BroadcastScope => "broadcast",
            PushDropReason::TradeAccIdMismatch => "trade_acc_id",
            PushDropReason::TradeMarketMismatch => "trade_market",
            PushDropReason::NotifyNotSubscribed => "notify_unsub",
            PushDropReason::TradeDecodeFailed => "trade_decode_failed",
        }
    }

    /// **v1.4.106 codex 0542 F4 [P3]**: 从 v1.4.105 之前的 `event_type` 字符串
    /// 兼容 helper. 让 surface migration 渐进 (旧 `event.event_type` 字符串
    /// 走这个 fn 转 typed). 未识别字符串 → `ScopeMissing` (向后兼容默认).
    ///
    /// 用法: `PushDropReason::from_legacy_event_type(&event.event_type)`.
    #[must_use]
    pub fn from_legacy_event_type(s: &str) -> Self {
        match s {
            "quote" => PushDropReason::QuoteScope,
            "trade" => PushDropReason::TradeScope,
            "broadcast" => PushDropReason::BroadcastScope,
            "notify" | "notify_unsub" => PushDropReason::NotifyNotSubscribed,
            "trade_acc_id" => PushDropReason::TradeAccIdMismatch,
            "trade_market" => PushDropReason::TradeMarketMismatch,
            "trade_decode_failed" => PushDropReason::TradeDecodeFailed,
            _ => PushDropReason::ScopeMissing,
        }
    }
}

/// 把限额 reject 的 reason 字串分类成固定小集合
///
/// 对应 `limits.rs::check_and_commit` 的拒绝文案前缀。新增类目时要同步这里，
/// 否则会落到 `"other"` 桶，dashboard 看不出来。
#[must_use]
pub fn classify_limit_reason(reason: &str) -> &'static str {
    let r = reason.to_ascii_lowercase();
    if r.starts_with("rate limit") {
        "rate"
    } else if r.starts_with("daily value") {
        "daily"
    } else if r.starts_with("order value") || r.starts_with("per-order") {
        "per_order"
    } else if r.starts_with("market ") {
        "market"
    } else if r.starts_with("symbol ") {
        "symbol"
    } else if r.starts_with("trd_side") {
        "side"
    } else if r.starts_with("outside hours") || r.starts_with("invalid hours_window") {
        "hours"
    } else {
        "other"
    }
}

/// v1.4.106 codex 0542 F1 [P2 SECURITY]: redact key_id label 为 8-hex 短 SHA256.
///
/// 输入 key id 明文 (如 "bot_a" / "trader-prod-1") → 输出 `kh_<8hex>`
/// (如 `kh_3a4f5b6c`). dashboard 看到 `kh_3a4f5b6c` 仍能区分 (counter 维度
/// 差异化), 但**不暴露 key id 明文** — 反查需要离线 dictionary 攻击 (尝试
/// SHA256 大量候选 id 直到 hash prefix 匹配).
///
/// **三个 sentinel 不 hash**, 直接透传 — 这些是 audit 内部 sentinel, 无 key id
/// 可泄漏:
/// - `<none>` (no api key configured / legacy mode)
/// - `<missing>` (key 缺失)
/// - `<invalid>` (key verify 失败 / unknown bearer)
///
/// **opt-out**: 设 `FUTU_METRICS_PUBLIC=1` env 时回退 v1.4.105 行为 (明文
/// key_id) — 老用户 dashboard 依赖 key_id 明文时用. 默认 secure (redact).
///
/// 8 hex (32 bit) 在 ~10 keys 量级内 collision 极低 (<10^-7), dashboard 维度
/// 差异化够用. 若 keys.json 上 100+ keys 仍想 redact 时单 metric 多个 hash
/// 是可接受 cardinality (类似 user-agent label 多 variant).
#[must_use]
pub fn redact_key_id(raw: &str) -> String {
    // sentinel 透传
    if raw == "<none>" || raw == "<missing>" || raw == "<invalid>" {
        return raw.to_string();
    }
    // opt-out env: 公开 dashboard / 老 prom 抓取保留明文行为
    if std::env::var_os("FUTU_METRICS_PUBLIC").is_some() {
        return raw.to_string();
    }
    let mut hasher = Sha256::new();
    hasher.update(raw.as_bytes());
    let digest = hasher.finalize();
    let mut out = String::with_capacity(11); // "kh_" + 8 hex
    out.push_str("kh_");
    for byte in &digest[..4] {
        let _ = std::fmt::Write::write_fmt(&mut out, format_args!("{byte:02x}"));
    }
    out
}

/// Prometheus label 值转义：按 exposition 格式要求，label 值必须在双引号里且
/// 转义 `\"`、反斜杠、换行。暴露给 [`Registry::render_prometheus`] 用
fn prom_label(v: &str) -> String {
    let mut out = String::with_capacity(v.len() + 2);
    out.push('"');
    for c in v.chars() {
        match c {
            '\\' => out.push_str("\\\\"),
            '"' => out.push_str("\\\""),
            '\n' => out.push_str("\\n"),
            _ => out.push(c),
        }
    }
    out.push('"');
    out
}

/// 在 [`audit::reject`] / [`audit::allow`] 里同步调用，保证日志和 metrics 一致
///
/// 单独写在这里（而不是 audit.rs）是为了让 `audit` 只关心 tracing，
/// 这个模块专管数值聚合
pub(crate) fn bump_auth_event(iface: &str, outcome: &str, key_id: &str) {
    if let Some(r) = global() {
        r.record_event(iface, outcome, key_id);
    }
}

pub(crate) fn bump_limit_reject(iface: &str, key_id: &str, reason: &str) {
    if let Some(r) = global() {
        r.record_limit_reject(iface, key_id, classify_limit_reason(reason));
    }
}

/// 便捷：记录 ws 因 scope 过滤掉的事件.
///
/// **v1.4.106 codex 0542 F4 [P3]**: 第一参数语义实际是 push **drop reason**
/// (不是 required_scope). label 名 v1.4.106 起渲染时改为 `reason=` 但函数签名
/// 保持兼容 (现存 caller 传 `event_type` / `"trade_market"` 等不变). 新代码
/// 推荐走 [`bump_ws_filtered_typed`] 拿 closed-set [`PushDropReason`] 防漂移.
pub fn bump_ws_filtered(reason: &str, client_key_id: &str) {
    if let Some(r) = global() {
        r.record_ws_filtered(reason, client_key_id);
    }
}

/// **v1.4.106 codex 0542 F4 [P3]**: typed 版 [`bump_ws_filtered`] —— 推荐新代码用.
///
/// caller 传 `PushDropReason::*` enum, 编译期穷举确保新加 reason 时 grep 找到
/// 所有 surface caller. dashboard label 跟 [`bump_ws_filtered`] 字符串路径
/// byte-identical (走同 [`Registry::record_ws_filtered`]).
pub fn bump_ws_filtered_typed(reason: PushDropReason, client_key_id: &str) {
    bump_ws_filtered(reason.as_metric_label(), client_key_id);
}

/// v1.4.90 P1-B: 便捷把 renderer 注册到全局 registry (若已 install).
///
/// 未 install 时 no-op (测试 / 无 metrics 模式)。
pub fn register_global_renderer<F>(renderer: F)
where
    F: Fn() -> String + Send + Sync + 'static,
{
    if let Some(r) = global() {
        r.register_renderer(renderer);
    }
}

#[cfg(test)]
mod tests;