futu_grpc/

auth.rs

//! gRPC 的 Bearer Token 鉴权
//!
//! 两种模式：
//! - **未配置 KeyStore**：完全不鉴权（保持旧行为，启动日志 warn）
//! - **配置了 KeyStore**：所有 RPC 必须带 `authorization: Bearer <plaintext>` metadata，
//!   且按 `EndpointSpec` / proto_id / RPC 类型校验 scope
//!
//! 由于 gRPC 采用通用 `Request(proto_id, body)` RPC 模式，scope 检查优先
//! 使用 `EndpointSpec.runtime.scope`；未进入 surface spec 的 legacy proto
//! 再回退到 capability/低层 proto_id 表。`SubscribePush` 流式 RPC 要求
//! `qot:read`（推送混合了行情与交易，以行情 scope 为最低门槛）。
//!
//! | proto_id 范围 | 所需 scope |
//! |---|---|
//! | 已声明的 1xxx 公开 query（GetGlobalState / GetUserInfo / DelayStatistics） | `EndpointSpec.runtime.scope` |
//! | legacy 1xxx 连接协议（InitConnect / KeepAlive / …） | 无（放行） |
//! | 3xxx 行情 | `qot:read` |
//! | 2001 / 2008 / 2101 / 2102 / 2111 / 2201 / 2211 / 2221 / 2222 / 2223 / 2225 / 2226 账户只读 | `acc:read` |
//! | 2005 UnlockTrade | `trade:unlock`（v1.4.104 codex F1 P1） |
//! | 2202 / 2205 / 2237 下单 / 改单 / 确认 | `trade:real` |
//! | 其他 | 拒绝（保守） |

use futu_auth::Scope;
use tonic::{Request, Status};

/// 按 proto_id 推断所需 scope.
///
/// 返回 `None` 只代表该 proto_id 未进入 `EndpointSpec` 且属于 legacy
/// 免鉴权连接协议（仍会校验 key 本身有效）。已声明的 proto-backed endpoint
/// 优先使用 `EndpointSpec.runtime.scope`.
#[inline]
pub fn scope_for_proto(proto_id: u32) -> Option<Scope> {
    futu_auth_pipeline::capability::scope_for_proto_id(proto_id)
}

// ─────────────────────────────────────────────────────────────────────────────
// v1.4.106 codex 0517 ζ25-redo: gRPC stateful QOT stable identity (F2)
//
// **背景**：v1.4.105 之前 gRPC `request()` 每次 RPC 调 `next_conn_id()`
// 自增分配新 virtual conn_id。`SubscriptionManager` / 各 cache 按 conn_id
// 记账。gRPC 客户端连续两次 RPC（同一 caller、同一 stream）拿到不同 conn_id
// → cache miss、sub 永远无法对齐、`get_sub_info` 看不到自己刚刚 sub 的
// security。等价于 REST v1.4.90 P0-B 之前的 quota 永久泄漏 bug。
//
// **修法**（对齐 REST `REST_SHARED_CONN` 设计哲学）：
//
// 同一 caller（按 Bearer token 派生身份；可选 `grpc-session-id` metadata
// header 进一步隔离 session）的所有 gRPC RPC 共享同一 deterministic stable
// conn_id。不同 caller / session → naturally 隔离。
//
// **取值范围**：`0x4000_0000_0000_0000`（bit 62）以上是本 namespace；
// 高 2 bit 留作语义标识，低 62 bit 由 `bearer + session` hash 派生。
// 与 `next_conn_id` 起点（20M）/ `REST_SHARED_CONN`（0xFFFF_FFFE）/ raw TCP
// `ConnectionRegistry` 分配段（u32 范围）完全不重合。
//
// **legacy mode**（KeyStore 未配置）：bearer = None；输出值仍 deterministic
// 但全 caller 共享一个固定值（namespace base + LEGACY_FALLBACK_HASH）。这跟
// 旧"全 RPC 共享同 ID"实质相同（legacy mode 本来就没鉴权区分），但比旧自增
// 的 conn_id 行为更稳定（cache 不再每次 miss）。
//
// **测试方式**：
// - 同 bearer + 同 session → 同 conn_id（reproducible）
// - 同 bearer + 不同 session → 不同 conn_id
// - 不同 bearer → 不同 conn_id
// - bearer = None → LEGACY 固定值
// - 输出值 bit 62 必置位（namespace 内）
// ─────────────────────────────────────────────────────────────────────────────

/// gRPC stable conn_id namespace base.
///
/// 任何 `derive_grpc_conn_id` 输出 conn_id 都满足 `out & GRPC_STABLE_CONN_NAMESPACE
/// == GRPC_STABLE_CONN_NAMESPACE`（bit 62 置位）。这与 raw TCP / WS / REST /
/// MCP 各自分配的 conn_id 段完全不重合，让 SubscriptionManager 可以零冲突
/// 共用一个 conn_id 表。
///
/// 取值：`0x4000_0000_0000_0000`（4_611_686_018_427_387_904）。bit 62 单 bit
/// 标识，低 62 bit 留 hash 派生。bit 63 留作未来 surface（避免 sign-bit
/// 与 `i64` 互转踩坑）。
pub const GRPC_STABLE_CONN_NAMESPACE: u64 = 0x4000_0000_0000_0000;

/// legacy fallback magic for `bearer = None`（KeyStore 未配置或 metadata 缺失）。
///
/// 这是一个**任意非零 magic**，喂给 `DefaultHasher` 让 legacy 路径与有 bearer
/// 的 caller 派生不同 conn_id。任何同 daemon 进程内 legacy caller 都拿同值
/// → cache 行为对齐"全 legacy caller 共享一个 conn"。
///
/// 取值 `0x3F3F_3F3F_3F3F_3F3E`：高 2 bit 0、低 62 bit 内交错位模式（任意非零
/// 即可，hasher 内部 mix 让结果分布均匀；本常量本身不直接当 conn_id 用）。
const LEGACY_FALLBACK_HASH: u64 = 0x3F3F_3F3F_3F3F_3F3E;

/// 派生 gRPC stable conn_id。
///
/// **参数**：
/// - `bearer`：Bearer token（已经 `parse_bearer_scheme` 解出的纯 token，不含
///   `Bearer ` 前缀）。`None` 表示 legacy mode（无鉴权配置），此时全部 caller
///   共享同一 fallback id。
/// - `session_id`：可选 `grpc-session-id` metadata header 值；用于同一 caller
///   想运行多个独立 stream / sub-state 时区分。空 `""` 视同 `None`。
///
/// **返回**：u64 stable conn_id，bit 62 必置位（在 `GRPC_STABLE_CONN_NAMESPACE`
/// 内）。同输入 → 同输出（process-stable，单 daemon 进程内幂等；不依赖跨
/// Rust 版本 / 跨 process hash 稳定性，这对 conn_id 用途已足够）。
///
/// **隔离语义**：
/// - 同 bearer + 同 session → 同 conn_id（连续 RPC 命中同一 sub state）
/// - 同 bearer + 不同 session → 不同 conn_id（独立 sub state）
/// - 不同 bearer → 不同 conn_id（caller 之间天然隔离）
/// - bearer=None + session=任意 → 仍 deterministic（legacy 同 caller 行为一致）
///
/// **算法**：
/// 1. 用 `DefaultHasher` 顺序 feed `(domain_tag, bearer_or_legacy, session_or_empty)`
/// 2. 取低 62 bit
/// 3. OR 上 `GRPC_STABLE_CONN_NAMESPACE`（bit 62）
///
/// `domain_tag` 防止和其他 hash 用途碰撞（pitfall #25 idempotency 派生原则）。
///
/// **不依赖项**：不依赖 KeyStore 是否配置（caller 已做完 auth 才到这里）；
/// 不依赖 RPC proto_id（`request()` / `subscribe_push()` 共用同一身份，
/// 这是 stateful sub 的目的）。
#[must_use]
pub fn derive_grpc_conn_id(bearer: Option<&str>, session_id: Option<&str>) -> u64 {
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};

    // domain tag — 让本 hash 与 idempotency / order-id hash 等其他用途分离
    const DOMAIN_TAG: &str = "futu-grpc-stable-conn-id-v1";

    let mut hasher = DefaultHasher::new();
    DOMAIN_TAG.hash(&mut hasher);

    match bearer {
        Some(t) if !t.is_empty() => {
            "bearer".hash(&mut hasher);
            t.hash(&mut hasher);
        }
        _ => {
            // legacy / 无 bearer：feed magic 让 caller 拿稳定值
            "legacy".hash(&mut hasher);
            LEGACY_FALLBACK_HASH.hash(&mut hasher);
        }
    }

    let session = session_id.unwrap_or("");
    if session.is_empty() {
        "no-session".hash(&mut hasher);
    } else {
        "session".hash(&mut hasher);
        session.hash(&mut hasher);
    }

    let raw = hasher.finish();
    // 取低 62 bit + OR namespace base（bit 62）
    (raw & ((1u64 << 62) - 1)) | GRPC_STABLE_CONN_NAMESPACE
}

/// 从 gRPC `Request<T>` metadata 提 optional `grpc-session-id` header。
///
/// 同一 Bearer 想分多个独立 stream / sub-state 时可显式带 session-id；
/// 缺失 → `None`，所有 RPC 共享同 caller 的 default session。
///
/// **格式**：任意非空 ASCII string。空字符串 → `None`。
#[must_use]
pub fn extract_grpc_session_id<T>(req: &Request<T>) -> Option<String> {
    let value = req.metadata().get("grpc-session-id")?;
    let s = value.to_str().ok()?.trim();
    if s.is_empty() {
        return None;
    }
    Some(s.to_string())
}

/// 从 gRPC `Request<T>` metadata 提取 trade-write idempotency key。
///
/// REST surface 使用 HTTP `Idempotency-Key` header；gRPC 没有 HTTP body
/// envelope，等价语义放在 metadata。metadata key 在 tonic 中按小写访问，
/// 这里同时接受 canonical `idempotency-key` 与更常见的
/// `x-idempotency-key`，空白值视同未传。
#[must_use]
pub fn extract_grpc_idempotency_key<T>(req: &Request<T>) -> Option<String> {
    for name in ["idempotency-key", "x-idempotency-key"] {
        let Some(value) = req.metadata().get(name) else {
            continue;
        };
        let Ok(s) = value.to_str() else {
            continue;
        };
        let trimmed = s.trim();
        if !trimmed.is_empty() {
            return Some(trimmed.to_string());
        }
    }
    None
}

// ─────────────────────────────────────────────────────────────────────────────
// v1.4.104: 跨 surface auth 中间件 transport adapter helpers
//
// 阶段 2: 把 authenticate / check_scope 等 inline 逻辑替换为
// `futu_auth_pipeline::authenticate_request` 调用. 本节留下 transport-only
// helpers (Bearer extract / RejectKind → Status 翻译), 让 surface adapter 极薄.
// 阶段 7-2: subscribe_push 也走 pipeline, 旧 `authenticate` / `check_scope`
// pub fn 删除 (无调用方).
// ─────────────────────────────────────────────────────────────────────────────

/// 从 gRPC `Request<T>` metadata 提 Bearer token (case-insensitive scheme parse).
/// 返 owned `Option<String>` 让 caller 决定 lifetime (避免 borrow vs move 冲突,
/// gRPC `Request` 主体后续要 `into_inner()` 拿 body).
///
/// metadata 缺失 / scheme 错 / token 空 → `None` (caller 应转成
/// `Credential::None`, pipeline 在 scope mode 下会 reject as Unauthenticated).
///
/// v1.4.104 阶段 7-3: 内部 case-insensitive 解析委托 `futu_auth_pipeline::
/// parse_bearer_scheme`, 4 surface 共用同一逻辑 (gRPC / WS / REST / MCP).
#[must_use]
pub fn extract_grpc_token<T>(req: &Request<T>) -> Option<String> {
    let value = req.metadata().get("authorization")?;
    let s = value.to_str().ok()?;
    let token = futu_auth_pipeline::parse_bearer_scheme(s)?;
    Some(token.to_string())
}

/// v1.4.106 D1 5b: gRPC surface adapter — 把 pipeline `AuthDecision::Reject`
/// 翻成 tonic `Status`.
///
/// **历史**: v1.4.104 阶段 5 把"翻 reject 为 Status"作 free fn `grpc_status_for`
/// 写在本文件; v1.4.106 D1 把 4 surface 的同类 translate fn 收敛到
/// [`futu_auth_pipeline::SurfaceAdapter`] trait, 让 4 surface 一致, 防
/// sibling-route 不一致 regression (codex round 3 F1 教训).
///
/// **gRPC Status 映射** (v1.4.105 #3 P2 sealed):
/// - `Unauthenticated` → `Status::unauthenticated(reason)` (保留 reason 让
///   client 知道 missing token / invalid bearer)
/// - `Forbidden` → `Status::permission_denied("forbidden")` (generic 文案 **不泄**
///   required scope name. raw reason 仍由 pipeline 写入 audit log, 运维能查;
///   客户端只看通用 forbidden, 不能从 message 反推 daemon 内部 scope 名)
/// - `RateLimited` → `Status::resource_exhausted(reason)` (rate 信息客户端需知道
///   backoff 策略)
/// - `NotFound` → `Status::not_found(reason)`
/// - `InternalError` → `Status::internal("internal error")` (类似 generic)
///
/// **历史上下文** (eli FINAL-BUG-REPORT-v5 #3 P2):
/// REST 已统一 generic `{"error":"forbidden"}`, gRPC 之前透传 `reason` 暴露了
/// `missing scope acc:read` / `missing scope trade:real` 等 daemon 内部 scope
/// 名称, 让 qot-only key 的攻击者能从拒绝消息探测 daemon scope 命名空间.
/// v1.4.105 对齐 REST 的 generic 策略.
pub struct GrpcAdapter;

impl futu_auth_pipeline::SurfaceAdapter for GrpcAdapter {
    type WireResponse = Status;

    fn surface_id() -> futu_auth_pipeline::SurfaceId {
        futu_auth_pipeline::SurfaceId::Grpc
    }

    fn translate_reject(
        kind: futu_auth_pipeline::RejectKind,
        reason: String,
    ) -> Self::WireResponse {
        use futu_auth_pipeline::RejectKind::*;
        match kind {
            Unauthenticated => Status::unauthenticated(reason),
            // v1.4.105 #3 P2: 对齐 REST `{"error":"forbidden"}` generic 策略,
            // 不泄 scope name. raw `reason` 已由 pipeline 写 audit log
            // (futu_auth::audit::reject), 运维查 daemon log 仍能拿到完整 scope
            // 信息.
            Forbidden => {
                let _ = reason; // explicit drop, 不透给客户端
                Status::permission_denied("forbidden")
            }
            RateLimited => Status::resource_exhausted(reason),
            NotFound => Status::not_found(reason),
            InternalError => {
                let _ = reason;
                Status::internal("internal error")
            }
        }
    }
}

/// gRPC surface 的 rejection 状态码翻译入口.
///
/// 内部委托 [`GrpcAdapter::translate_reject`]. `server.rs` 与测试继续走这个
/// free fn，避免 surface adapter 细节散到 request handler.
#[must_use]
pub fn grpc_status_for(kind: futu_auth_pipeline::RejectKind, reason: String) -> Status {
    use futu_auth_pipeline::SurfaceAdapter;
    GrpcAdapter::translate_reject(kind, reason)
}

#[cfg(test)]
mod tests;