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. REST TLS 配置矩阵与证书/私钥一次加载 → fail-closed 早 abort
//! 2. keys-file 预 dry-run 验证 (REST/gRPC/WS) → fail-closed 早 abort
//! 3. 初始化日志 (json vs plain + audit guard)
//! 4. `tighten_secret_files_at_startup()` 把 0644 secret 文件收紧到 0600
//! 5. 安装全局 panic hook (tracing + crash log + exit 101)
//! 6. install futu_auth metrics registry
//! 7. 构造 shared `RuntimeCounters`
//! 8. 计算 `listen_addr` 并打印 "starting" 日志
//! 9. WARN: moomoo + 显式 `--login-region`
//! 10. 启动前端口冲突探测
//!
//! `--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,
    bind_runtime_language_pack_cache_root, 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) rest_key_store: Option<Arc<futu_auth::KeyStore>>,
    pub(super) rest_transport: futu_rest::server::RestTransport,
    pub(super) ws_keys_file: Option<std::path::PathBuf>,
    pub(super) ws_key_store: Option<Arc<futu_auth::KeyStore>>,
    pub(super) grpc_keys_file: Option<std::path::PathBuf>,
    pub(super) grpc_key_store: Option<Arc<futu_auth::KeyStore>>,
    /// 是否允许无 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) -> Result<()> {
    let options = language_pack_auto_update_options(config);
    bind_runtime_language_pack_cache_root(options.cache_root.clone())
        .map_err(anyhow::Error::msg)?;
    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"
            ),
        }
    });
    Ok(())
}

fn runtime_language_log_facts(config: &RuntimeConfig) -> (&str, &'static str) {
    (
        config.lang.as_str(),
        config.runtime_language_source.as_str(),
    )
}

fn log_runtime_language(config: &RuntimeConfig) {
    let (lang, source) = runtime_language_log_facts(config);
    tracing::info!(
        target: "startup",
        lang = %lang,
        source = %source,
        "runtime_language_resolved"
    );
}

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",
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct StartupPortProbe<'a> {
    surface: &'static str,
    bind_ip: &'a str,
    port: u16,
}

pub(super) fn ftapi_listener_enabled(config: &RuntimeConfig) -> bool {
    if config.setup_only {
        return false;
    }
    let any_keys_configured = config.rest_keys_file.is_some()
        || config.grpc_keys_file.is_some()
        || config.ws_keys_file.is_some();
    !any_keys_configured || config.allow_tcp_unauthenticated
}

fn startup_port_probes(config: &RuntimeConfig) -> Vec<StartupPortProbe<'_>> {
    if config.setup_only {
        return Vec::new();
    }

    let mut probes = Vec::new();
    if ftapi_listener_enabled(config) {
        probes.push(StartupPortProbe {
            surface: "FTAPI",
            bind_ip: &config.ip,
            port: config.port,
        });
    }
    for (surface, port) in [
        ("REST", config.rest_port),
        ("gRPC", config.grpc_port),
        ("WebSocket", config.websocket_port),
    ] {
        if let Some(port) = port {
            probes.push(StartupPortProbe {
                surface,
                bind_ip: &config.ip,
                port,
            });
        }
    }
    if let Some(port) = config.telnet_port {
        probes.push(StartupPortProbe {
            surface: "Telnet",
            bind_ip: &config.telnet_ip,
            port,
        });
    }
    probes
}

fn startup_port_conflict_hint(surface: &str, addr: &str) -> String {
    futu_server::bind_hint::port_conflict_message(surface, startup_port_flag(surface), addr)
}

pub(super) async fn prepare_rest_transport(
    rest_port: Option<u16>,
    rest_keys_file: Option<&std::path::Path>,
    cert_path: Option<&std::path::Path>,
    key_path: Option<&std::path::Path>,
) -> Result<futu_rest::server::RestTransport> {
    use futu_rest::server::{RestTlsConfig, RestTransport};

    match (rest_port, cert_path, key_path) {
        (None, None, None) => Ok(RestTransport::Plaintext),
        (None, _, _) => Err(anyhow::anyhow!(
            "--rest-tls-cert/--rest-tls-key require --rest-port"
        )),
        (Some(_), Some(cert), Some(key)) => RestTlsConfig::from_pem_files(cert, key)
            .await
            .map(RestTransport::Tls)
            .map_err(|error| {
                anyhow::Error::new(error)
                    .context("invalid REST TLS certificate/private-key configuration")
            }),
        (Some(_), None, None) if rest_keys_file.is_some() => Err(anyhow::anyhow!(
            "--rest-keys-file requires native REST TLS: configure both \
             --rest-tls-cert and --rest-tls-key"
        )),
        (Some(_), None, None) => Ok(RestTransport::Plaintext),
        (Some(_), _, _) => Err(anyhow::anyhow!(
            "--rest-tls-cert and --rest-tls-key must be configured together"
        )),
    }
}

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;
    let rest_transport = prepare_rest_transport(
        config.rest_port,
        rest_keys_file.as_deref(),
        config.rest_tls_cert.as_deref(),
        config.rest_tls_key.as_deref(),
    )
    .await?;

    // v1.4.104 external reviewer P1-003 (P1) fix: keys-file 解析时序前移到 broker auth /
    // SMS 之前. 之前 schema 错的 keys-file 要等到 surface server 启动时才报错
    // (broker auth + SMS 之后 7+ 秒), 配置错应该启动就发现.
    //
    // Parse once and retain the exact generation that Phase 4 will install.
    // Re-reading after one listener has bound creates a TOCTOU window where a
    // mutated sibling file can produce a partially open public surface set.
    let preload = |label: &'static str,
                   path: Option<&std::path::Path>|
     -> Result<Option<Arc<futu_auth::KeyStore>>> {
        let Some(path) = path else {
            return Ok(None);
        };
        let store = futu_auth::KeyStore::load(path).map_err(|error| {
            anyhow::anyhow!(
                "{label} keys file at {} failed schema validation before broker auth / SMS: \
                 {error}. fix the keys file then restart.",
                path.display()
            )
        })?;
        tracing::info!(
            surface = label,
            path = %path.display(),
            keys_loaded = store.len(),
            "keys file parsed and retained before listener bind"
        );
        Ok(Some(Arc::new(store)))
    };
    let rest_key_store = preload("REST", rest_keys_file.as_deref())?;
    let grpc_key_store = preload("gRPC", grpc_keys_file.as_deref())?;
    let ws_key_store = preload("WS", ws_keys_file.as_deref())?;
    // 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(error) => {
                let context = audit_log.as_ref().map_or_else(
                    || "failed to initialize daemon logging".to_owned(),
                    |path| {
                        format!(
                            "failed to initialize explicitly configured audit log at {}",
                            path.display()
                        )
                    },
                );
                return Err(anyhow::Error::new(error).context(context));
            }
        }
    };

    log_runtime_language(config);
    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）。
    // C++ Ref: FTGateway.cpp:1643-1667. Port-lock or listener startup failure
    // aborts RunAPISvr instead of allowing a partial API surface set.
    {
        for probe in startup_port_probes(config) {
            let addr = port_probe_addr(probe.bind_ip, probe.port);
            if std::net::TcpStream::connect_timeout(&addr, std::time::Duration::from_millis(200))
                .is_ok()
            {
                let hint = startup_port_conflict_hint(probe.surface, &addr.to_string());
                return Err(anyhow::anyhow!(
                    "startup port conflict probe detected an existing listener: {hint}"
                ));
            }
        }
    }

    Ok(Phase1Out {
        _audit_guard,
        shared_counters,
        listen_addr,
        rest_keys_file,
        rest_key_store,
        rest_transport,
        ws_keys_file,
        ws_key_store,
        grpc_keys_file,
        grpc_key_store,
        allow_tcp_unauthenticated,
    })
}

#[cfg(test)]
mod tests;