futu_opend/startup/

phase1.rs

//! v1.4.110 Layer 3 A: startup Phase 1 — bootstrap前置 (logging / metrics /
//! 守护设施). 抽自原 `mod.rs::run_daemon` 33..219 行段.
//!
//! Phase 1 副作用 (按顺序):
//! 1. keys-file 预 dry-run 验证 (REST/gRPC/WS) → fail-closed 早 abort
//! 2. 初始化日志 (json vs plain + audit guard)
//! 3. `tighten_secret_files_at_startup()` 把 0644 secret 文件收紧到 0600
//! 4. 安装全局 panic hook (tracing + crash log + exit 101)
//! 5. install futu_auth metrics registry
//! 6. 构造 shared `RuntimeCounters`
//! 7. 计算 `listen_addr` 并打印 "starting" 日志
//! 8. WARN: moomoo + 显式 `--login-region`
//! 9. 启动前端口冲突探测
//!
//! `--tz` / TOML `tz` 必须在 Tokio runtime 创建前应用，由
//! [`super::apply_pre_runtime_tz`] 在 sync `main()` 里调用。

use anyhow::Result;
use std::path::PathBuf;
use std::sync::Arc;

use crate::cli::Platform;
use crate::config::RuntimeConfig;
use crate::crash_log::write_crash_log_file;
use futu_core::localization::{
    LanguagePackAutoUpdateOptions, auto_update_language_pack, default_language_pack_cache_root,
};

/// Phase 1 output — 必须由 caller 持有到进程退出, 否则 audit guard drop
/// 会让 tracing-appender 后台线程提早关闭丢事件.
pub(super) struct Phase1Out {
    /// audit 日志 guard, drop = tracing-appender 关闭. caller 必须 outlive.
    pub(super) _audit_guard: Option<tracing_appender::non_blocking::WorkerGuard>,
    /// 共享 RuntimeCounters (REST + gRPC 共用一份保证跨 surface rate window 一致).
    pub(super) shared_counters: Arc<futu_auth::RuntimeCounters>,
    /// "ip:port" 字符串, 后续 server / WS / REST / gRPC / telnet 复用.
    pub(super) listen_addr: String,
    /// 从 config 复制的 keys file 路径 (Phase 4 使用).
    pub(super) rest_keys_file: Option<std::path::PathBuf>,
    pub(super) ws_keys_file: Option<std::path::PathBuf>,
    pub(super) grpc_keys_file: Option<std::path::PathBuf>,
    /// 是否允许无 auth 的 TCP listener (Phase 4 决策).
    pub(super) allow_tcp_unauthenticated: bool,
}

pub(super) fn is_valid_iana_tz_name(tz: &str) -> bool {
    if tz == "UTC" {
        return true;
    }
    // Local shape guard for `--tz` before writing the process-wide TZ env var.
    // We intentionally avoid a chrono_tz parse dependency here (see v1.4.87
    // --tz decision) and accept only IANA-style slash-separated ASCII names.
    // The 128-byte cap is a defensive config-boundary limit, not a protocol
    // value; revisit if IANA tzdb ever grows names beyond this shape.
    if tz.is_empty() || tz.len() > 128 || tz.starts_with('/') || tz.ends_with('/') {
        return false;
    }

    let mut parts = 0usize;
    for part in tz.split('/') {
        parts += 1;
        if part.is_empty() || part == "." || part == ".." {
            return false;
        }
        if part.contains('.') {
            return false;
        }
        if !part
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-' | b'+'))
        {
            return false;
        }
    }

    parts >= 2
}

pub(super) fn apply_pre_runtime_tz(config: &RuntimeConfig) {
    if let Some(tz) = &config.tz {
        // 轻度校验 IANA name 格式 (不跑 chrono_tz parse 避免 dep 膨胀)
        if !is_valid_iana_tz_name(tz) {
            eprintln!(
                "error: --tz 无效 IANA timezone '{tz}'. 示例: Asia/Hong_Kong, America/New_York, UTC"
            );
            std::process::exit(2);
        }
        // SAFETY: production main calls this before constructing the Tokio
        // runtime, so no runtime worker thread can concurrently read the
        // process environment. Tests only cover the shape validator; they do
        // not call this helper concurrently.
        unsafe {
            std::env::set_var("TZ", tz);
        }
        eprintln!("ℹ️  TZ set to '{tz}' via --tz flag / TOML tz (v1.4.87 #3 G1)");
    }
}

fn language_pack_cache_root(config: &RuntimeConfig) -> PathBuf {
    config
        .language_pack_cache_dir
        .clone()
        .unwrap_or_else(default_language_pack_cache_root)
}

pub(super) fn language_pack_auto_update_options(
    config: &RuntimeConfig,
) -> LanguagePackAutoUpdateOptions {
    LanguagePackAutoUpdateOptions {
        enabled: config.language_pack_auto_update,
        endpoint: config.language_pack_endpoint.clone(),
        cache_root: language_pack_cache_root(config),
        timeout_ms: config.language_pack_update_timeout_ms,
        lang_filter: None,
    }
}

fn spawn_language_pack_auto_update(config: &RuntimeConfig) {
    let options = language_pack_auto_update_options(config);
    tokio::spawn(async move {
        let outcome = auto_update_language_pack(options).await;
        let status = outcome.state.as_str();
        match outcome.error.as_deref() {
            Some(error) => tracing::warn!(
                target = "language_pack",
                status,
                error,
                pack_version = outcome.pack_version.as_deref().unwrap_or("-"),
                "language_pack_auto_update"
            ),
            None => tracing::info!(
                target = "language_pack",
                status,
                pack_version = outcome.pack_version.as_deref().unwrap_or("-"),
                "language_pack_auto_update"
            ),
        }
    });
}

fn port_probe_addr(bind_ip: &str, port: u16) -> std::net::SocketAddr {
    match bind_ip.parse::<std::net::IpAddr>() {
        Ok(std::net::IpAddr::V4(ip)) if ip.is_unspecified() => {
            std::net::SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port))
        }
        Ok(std::net::IpAddr::V6(ip)) if ip.is_unspecified() => {
            std::net::SocketAddr::from((std::net::Ipv6Addr::LOCALHOST, port))
        }
        Ok(ip) => std::net::SocketAddr::new(ip, port),
        Err(_) => std::net::SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port)),
    }
}

fn startup_port_flag(surface: &str) -> &'static str {
    match surface {
        "FTAPI" => "--port",
        "REST" => "--rest-port",
        "gRPC" => "--grpc-port",
        "WebSocket" => "--websocket-port",
        "Telnet" => "--telnet-port",
        _ => "--port",
    }
}

pub(super) async fn run_phase1(config: &RuntimeConfig) -> Result<Phase1Out> {
    // codex 0547 F6 (P3): 安全字段从 config 读 (而非 args.*) — TOML 也能
    // override 这些 (与 docs "字段与 CLI 一致" 契约对齐).
    let rest_keys_file = config.rest_keys_file.clone();
    let ws_keys_file = config.ws_keys_file.clone();
    let grpc_keys_file = config.grpc_keys_file.clone();
    let audit_log = config.audit_log.clone();
    let allow_tcp_unauthenticated = config.allow_tcp_unauthenticated;

    // v1.4.104 external reviewer P1-003 (P1) fix: keys-file 解析时序前移到 broker auth /
    // SMS 之前. 之前 schema 错的 keys-file 要等到 surface server 启动时才报错
    // (broker auth + SMS 之后 7+ 秒), 配置错应该启动就发现.
    //
    // 这里只**预解析 + dry-run 验证**, 不持久化结果 (实际 Arc 在 surface server
    // 启动时由 KeyStore::load 重新读+解析, 因为 SIGHUP reload 也走那条路径).
    // dry-run 失败 → 立即 abort, 不进 broker auth.
    for (label, path_opt) in [
        ("REST", &rest_keys_file),
        ("gRPC", &grpc_keys_file),
        ("WS", &ws_keys_file),
    ] {
        if let Some(path) = path_opt {
            match futu_auth::KeyStore::load(path) {
                Ok(ks) => {
                    tracing::info!(
                        surface = label,
                        path = %path.display(),
                        keys_loaded = ks.len(),
                        "v1.4.104 external report P1-003 (P1): {} keys file pre-validated OK \
                         (broker auth not yet started)",
                        label
                    );
                }
                Err(e) => {
                    tracing::error!(
                        surface = label,
                        error = %e,
                        path = %path.display(),
                        "v1.4.104 external report P1-003 (P1): {} keys file pre-validation FAILED — \
                         abort before broker auth / SMS to fail-closed early",
                        label
                    );
                    return Err(anyhow::anyhow!(
                        "v1.4.104 external report P1-003 (P1) fix: {} keys file at {} failed schema \
                         validation: {e}. abort before broker auth / SMS. fix the keys \
                         file then restart.",
                        label,
                        path.display()
                    ));
                }
            }
        }
    }
    // codex 0547 F6 (P3): merge_config 已提前到 args 仍可用阶段 (~line 1006);
    // 此处不再重复 capture inject_auth_failure_every / merge_config.
    // dev-flags 在 cfg(feature = "dev-flags") 路径上, capture 已在 args 解析后立即做.

    // 1. 初始化日志（--log-level 参数生效，RUST_LOG 环境变量优先）
    // audit 日志 guard 必须活到进程退出，否则 tracing-appender 后台线程可能丢事件。
    let _audit_guard = if config.json_log {
        // v1.4.27（BUG-7，加拿大同事 v1.4.26 回归测试发现）：`--audit-log` 和
        // `--json-log` 一起用时，之前是**静默忽略** `--audit-log`（只打 warn
        // 到 stderr）、创建空文件；用户会误以为"没有审计事件发生"。现在改
        // 硬失败 → 用户必须显式二选一，避免审计文件空导致的合规事故。
        if audit_log.is_some() {
            eprintln!(
                "error: --audit-log and --json-log are mutually exclusive.\n\
                 - --json-log: entire stderr as JSONL (full event stream)\n\
                 - --audit-log: only target=futu_audit events as JSONL to a file\n\
                 choose one. If you need both machine-readable stderr AND a separate audit \
                 file, open an issue — today's layer composition doesn't support it."
            );
            std::process::exit(2);
        }
        futu_core::log::init_json_logging_with_level(&config.log_level);
        None
    } else {
        match futu_core::log::init_logging_with_audit(&config.log_level, audit_log.as_deref()) {
            Ok(guard) => {
                if let (Some(path), Some(_)) = (audit_log.as_ref(), guard.as_ref()) {
                    tracing::info!(
                        path = %path.display(),
                        "audit JSONL logger enabled (target=futu_audit → file)"
                    );
                }
                guard
            }
            Err(e) => {
                eprintln!("warning: failed to init audit log: {e}");
                futu_core::log::init_logging_with_level(&config.log_level);
                None
            }
        }
    };

    spawn_language_pack_auto_update(config);

    // v1.4.102 codex 27 F11 (P2) fix: startup chmod migration 移到无条件 path,
    // 不再放在 login 分支里. 升级 (v1.4.101 及以前) 用户的
    // `~/.futu-opend-rs/credentials-*.json` / `device-*.dat` 默认是 0644,
    // 多用户机其他本地用户能读 tgtgt / web_sig. 此 fn 扫 ~/.futu-opend-rs/ 把
    // secret 文件统一收紧到 0600.
    //
    // **历史**: BUG-012 修法 v1.4.102 ship 时把此 call 放在 `if let
    // (Some(account), Some(password))` 分支内 — 无登录凭据 / 只跑 admin shell
    // / 凭据解析失败的 daemon 都不执行 migration. codex 27 F11 audit 抓到.
    //
    // best-effort: chmod 失败 warn but don't fail (失败 != 凭据本身失效).
    futu_backend::auth::tighten_secret_files_at_startup();

    // v1.4.41 (P3.1 第二阶段): tracing subscriber 已装，重装 panic hook
    // 让 panic 走 tracing::error!（audit log / JSON log / stderr 三出）。
    std::panic::set_hook(Box::new(|info| {
        let location = info
            .location()
            .map(|l| format!("{}:{}", l.file(), l.line()))
            .unwrap_or_else(|| "<unknown>".to_string());
        let payload = info
            .payload()
            .downcast_ref::<&str>()
            .copied()
            .or_else(|| info.payload().downcast_ref::<String>().map(|s| s.as_str()))
            .unwrap_or("<non-string panic payload>");
        let thread = std::thread::current()
            .name()
            .unwrap_or("<unnamed>")
            .to_string();
        tracing::error!(
            target: "panic",
            location = %location,
            payload = %payload,
            thread = %thread,
            "PANIC caught by global hook"
        );
        eprintln!("PANIC at {location}: {payload} (thread={thread})");
        // v1.4.97 P1-D-D: also write dated crash log to disk for forensics.
        // Same as pre-tracing hook — covers panics that occur after tracing
        // subscriber is up.
        write_crash_log_file(info);
        // v1.4.97 P1-D-E: propagate any panic → process exit so systemd
        // Restart=on-failure can restart the daemon. Without this, tokio
        // task panic silently kills only the task, leaving daemon zombie
        // (REST/gRPC/WS/telnet/push tasks die one-by-one with main alive).
        //
        // Aligned with C++ NNCrashCenter `exit(NN_ExitCode_Crash)` pattern
        // (NNCrashCenter_Mac.cpp:99; per agent 9 finding).
        // `exit(101)` matches Rust panic default exit code; not used in
        // test build (cfg(not(test))) to avoid disrupting unit tests.
        #[cfg(not(test))]
        std::process::exit(101);
    }));

    // 2. install 全局 metrics registry（让 audit::* 的 counter hook 和
    //    REST `/metrics` 端点能对齐同一套计数器）
    futu_auth::metrics::install(std::sync::Arc::new(futu_auth::MetricsRegistry::default()));

    // 2.1 共享 RuntimeCounters：REST / gRPC 共用一个，这样 rate limit 和日累计
    //     跨接口一致（同一把 key 通过 REST 下 3 单、gRPC 下 3 单，rate 窗口
    //     看到 6 单，不是各看 3 单）
    let shared_counters = std::sync::Arc::new(futu_auth::RuntimeCounters::new());

    let listen_addr = format!("{}:{}", config.ip, config.port);
    tracing::info!(addr = %listen_addr, "starting FutuOpenD Rust Gateway");

    // v1.4.42 (external reviewer v1.4.40 报告 P3.5 澄清): moomoo 账户 + 显式 --login-region
    // → WARN 提示此 flag 对 moomoo 账户 noop（不影响 platform IP 选择）。
    //
    // 原因：login_region 只用在 CN 手机号账户的 salt URL `region_no` 参数，
    // platform IP 池按 user_attribution（CN/HK/US/SG/AU/JP）从 conn_points
    // 选，不按 login_region 切（3 个 region 代号 gz/sh/hk 是 CN 大陆分区，
    // 和 platform IP 池没对应关系）。external reviewer v1.4.40 报告观察 "三种 region 下
    // platform IP 相同" 是预期行为，不是 bug。
    //
    // v1.4.40 计划中"加 WARN"实际没加（CHANGELOG 声称但代码漏），v1.4.42 补上。
    if config.login_region_explicit && matches!(config.platform, Platform::Moomoo) {
        tracing::warn!(
            login_region = %config.login_region,
            platform = "moomoo",
            "--login-region={region} is a NO-OP for moomoo accounts — flag only \
             applies to --platform futunn + CN phone-number login. moomoo accounts \
             route via user_attribution automatically. Observed \"same platform IP \
             across gz/sh/hk\" is expected behavior. Remove --login-region to silence.",
            region = config.login_region
        );
    }

    // v1.4.16：端口冲突检测——在登录之前先检查核心端口是否已被占用。
    // 多实例并行（如同时跑 futunn + moomoo）是预期场景，但用户容易忘记改端口，
    // 导致 futucli 连到了旧实例看到错账号的数据（同事 bug report #6）。
    {
        let ports_to_check: Vec<(&str, &str, u16)> =
            std::iter::once(("FTAPI", config.ip.as_str(), config.port))
                .chain(config.rest_port.map(|p| ("REST", config.ip.as_str(), p)))
                .chain(config.grpc_port.map(|p| ("gRPC", config.ip.as_str(), p)))
                .chain(
                    config
                        .websocket_port
                        .map(|p| ("WebSocket", config.ip.as_str(), p)),
                )
                .chain(
                    config
                        .telnet_port
                        .map(|p| ("Telnet", config.telnet_ip.as_str(), p)),
                )
                .collect();
        for (name, bind_ip, port) in &ports_to_check {
            let addr = port_probe_addr(bind_ip, *port);
            if std::net::TcpStream::connect_timeout(&addr, std::time::Duration::from_millis(200))
                .is_ok()
            {
                let hint = futu_server::bind_hint::port_conflict_message(
                    name,
                    startup_port_flag(name),
                    &addr.to_string(),
                );
                tracing::warn!(
                    name,
                    port,
                    hint = %hint,
                    "startup port conflict probe detected an existing listener"
                );
                eprintln!(
                    "warning: startup port conflict probe detected an existing listener: {hint}"
                );
            }
        }
    }

    Ok(Phase1Out {
        _audit_guard,
        shared_counters,
        listen_addr,
        rest_keys_file,
        ws_keys_file,
        grpc_keys_file,
        allow_tcp_unauthenticated,
    })
}

#[cfg(test)]
mod tests;