futu_auth/

store.rs

//! KeyStore: keys.json 加载 + 热替换 + 明文验证

use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use arc_swap::ArcSwap;
use chrono::Utc;
use serde::{Deserialize, Serialize};

use crate::key::{KeyRecord, hash_plaintext};

#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum KeyStoreError {
    #[error("read {path:?}: {source}")]
    Read {
        path: PathBuf,
        source: std::io::Error,
    },
    #[error("parse {path:?}: {source}")]
    Parse {
        path: PathBuf,
        source: serde_json::Error,
    },
    #[error("write {path:?}: {source}")]
    Write {
        path: PathBuf,
        source: std::io::Error,
    },
    #[error("serialize: {0}")]
    Serialize(#[from] serde_json::Error),
    #[error("unsupported keys.json version {0} (supported: 1)")]
    UnsupportedVersion(u32),
    #[error("duplicate key id {0:?}")]
    DuplicateId(String),
}

/// keys.json 顶层文件结构
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KeysFile {
    pub version: u32,
    pub keys: Vec<KeyRecord>,
}

const CURRENT_VERSION: u32 = 1;

/// KeyStore：热可替换的 keys 集合
#[derive(Debug)]
pub struct KeyStore {
    path: Option<PathBuf>,
    current: ArcSwap<KeysFile>,
    hash_index: ArcSwap<HashMap<String, Vec<KeyRecord>>>,
}

impl KeyStore {
    /// 空 store（没有 keys 文件时）
    pub fn empty() -> Self {
        let file = KeysFile {
            version: CURRENT_VERSION,
            keys: vec![],
        };
        let hash_index = Self::build_hash_index(&file);
        Self {
            path: None,
            current: ArcSwap::from_pointee(file),
            hash_index: ArcSwap::from_pointee(hash_index),
        }
    }

    /// 从文件加载
    pub fn load(path: impl Into<PathBuf>) -> Result<Self, KeyStoreError> {
        let path = path.into();
        let file = Self::load_file(&path)?;
        let hash_index = Self::build_hash_index(&file);
        Ok(Self {
            path: Some(path),
            current: ArcSwap::from_pointee(file),
            hash_index: ArcSwap::from_pointee(hash_index),
        })
    }

    fn build_hash_index(file: &KeysFile) -> HashMap<String, Vec<KeyRecord>> {
        let mut index: HashMap<String, Vec<KeyRecord>> = HashMap::with_capacity(file.keys.len());
        for rec in &file.keys {
            index.entry(rec.hash.clone()).or_default().push(rec.clone());
        }
        index
    }

    fn load_file(path: &Path) -> Result<KeysFile, KeyStoreError> {
        if !path.exists() {
            return Self::load_file_unlocked(path);
        }
        let _guard = AdvisoryLockGuard::acquire_shared(path)?;
        Self::load_file_unlocked(path)
    }

    fn load_file_unlocked(path: &Path) -> Result<KeysFile, KeyStoreError> {
        let text = fs::read_to_string(path).map_err(|source| KeyStoreError::Read {
            path: path.to_path_buf(),
            source,
        })?;
        let mut file: KeysFile =
            serde_json::from_str(&text).map_err(|source| KeyStoreError::Parse {
                path: path.to_path_buf(),
                source,
            })?;
        if file.version != CURRENT_VERSION {
            return Err(KeyStoreError::UnsupportedVersion(file.version));
        }
        // 检查重复 id
        let mut seen = std::collections::HashSet::new();
        for k in &file.keys {
            if !seen.insert(k.id.clone()) {
                return Err(KeyStoreError::DuplicateId(k.id.clone()));
            }
        }
        // v1.4.104 external reviewer S-002 (P0) fix: load 时立即注入 fail-closed sentinel —
        //
        // 之前 expand_allowed_card_nums 只在 daemon 启动 + SIGHUP 跑, 但 MCP 等
        // keystore consumer **不调** expand → 受 `allowed_card_nums` 限制的 key
        // 加载后 `allowed_acc_ids = None`, 被限额引擎当作 "无限制" silent allow.
        //
        // **修法**: load_file 时对每条 key, 若 `allowed_card_nums` 非空但
        // `allowed_acc_ids` 是 None / empty → 注入 sentinel `Some({0})`. 真实
        // expansion (e.g. opend daemon GetAccList 之后) 会以 resolved acc_ids 覆盖.
        // MCP 等不跑 expand 的消费方仍受 sentinel 保护 (fail-closed: real acc_id
        // ≠ 0 → 永远 reject).
        //
        // 这是架构层 fix (与 v1.4.103 codex F1 P1 expand-time sentinel 同语义,
        // 但提前到 load 时让所有 consumer 受益, 不依赖每个消费方都调 expand).
        for rec in &mut file.keys {
            // v1.4.106 F-P2-D: snapshot 文件源原始 allowed_acc_ids (sentinel
            // 注入和 card_num expansion 之前). expand_allowed_card_nums 用
            // 此字段作起步, 防止累积 stale resolutions.
            rec.raw_explicit_acc_ids = rec.allowed_acc_ids.clone();

            let has_card_nums = rec
                .allowed_card_nums
                .as_ref()
                .is_some_and(|v| !v.is_empty());
            let has_acc_ids = rec.allowed_acc_ids.as_ref().is_some_and(|s| !s.is_empty());
            if has_card_nums && !has_acc_ids {
                // 写 sentinel acc_id=0; expand_allowed_card_nums 后续会用
                // resolved acc_ids 覆盖. 此期间任何 acc_id ≠ 0 的 query 全 reject.
                let mut sentinel = rec.allowed_acc_ids.clone().unwrap_or_default();
                sentinel.insert(0);
                rec.allowed_acc_ids = Some(sentinel);
                let key_id = crate::metrics::redact_key_id_for_logs(&rec.id);
                let card_num_count = rec.allowed_card_nums.as_ref().map_or(0, Vec::len);
                tracing::warn!(
                    key_id = %key_id,
                    card_num_count,
                    "v1.4.104 external report S-002 (P0): keystore load 注入 fail-closed sentinel \
                     allowed_acc_ids={{0}} (caller 配 allowed_card_nums 但 daemon 还没\
                     expand). expand_allowed_card_nums 跑完后真实 resolved acc_ids 覆盖. \
                     MCP / 不跑 expand 的 consumer 仍按 sentinel 保护."
                );
            }
        }
        Ok(file)
    }

    /// SIGHUP 热重载：用同一路径重新读文件
    pub fn reload(&self) -> Result<(), KeyStoreError> {
        let Some(path) = &self.path else {
            return Ok(());
        };
        let file = Self::load_file(path)?;
        let hash_index = Self::build_hash_index(&file);
        self.current.store(Arc::new(file));
        self.hash_index.store(Arc::new(hash_index));
        Ok(())
    }

    /// v1.4.103 (B10): 把每条 key 的 `allowed_card_nums` (string format) 通过
    /// `resolver` 解析成 acc_id, **合并**进 `allowed_acc_ids` (in-memory only,
    /// 不写回 keys.json — 文件源不变, 重载后再 expand).
    ///
    /// `resolver(card_num) -> Vec<u64>` 由 caller 提供 (典型 closure 持
    /// `Arc<TrdCache>` 调 `find_acc_ids_by_card_num`).
    ///
    /// **行为**:
    /// - resolver 返 1 个 acc_id → 加入 allowed_acc_ids (resolved)
    /// - 返 0 个 → 通过 `unresolved_callback` 通知 caller (e.g. log warn)
    /// - 返 ≥ 2 个 → 通过 `ambiguous_callback` 通知 caller (loud, skip 该条)
    ///
    /// 返 `(resolved_count, unresolved_count, ambiguous_count)`.
    ///
    /// **典型调用 (daemon 启动 GetAccList 成功后)**:
    /// ```ignore
    /// let cache_clone = trd_cache.clone();
    /// key_store.expand_allowed_card_nums(
    ///     |cn: &str| cache_clone.find_acc_ids_by_card_num(cn),
    ///     |key_id, cn| tracing::warn!(key_id, card_num=cn, "card_num not found"),
    ///     |key_id, cn, candidates| tracing::warn!(key_id, card_num=cn, ?candidates, "ambiguous card_num"),
    /// );
    /// ```
    pub fn expand_allowed_card_nums<R, FU, FA>(
        &self,
        resolver: R,
        mut unresolved_callback: FU,
        mut ambiguous_callback: FA,
    ) -> (usize, usize, usize)
    where
        R: Fn(&str) -> Vec<u64>,
        FU: FnMut(&str, &str),         // key_id, card_num
        FA: FnMut(&str, &str, &[u64]), // key_id, card_num, candidates
    {
        let current = self.current.load();
        let mut new_keys = current.keys.clone();
        let mut resolved = 0;
        let mut unresolved = 0;
        let mut ambiguous = 0;
        for rec in &mut new_keys {
            let Some(card_nums) = rec.allowed_card_nums.clone() else {
                continue;
            };
            // v1.4.106 F-P2-D: 从 raw_explicit_acc_ids 起步重新 resolve, 不
            // 累积 stale resolutions. 之前 `rec.allowed_acc_ids.clone()` 起
            // 步会让连续 expand 累积 — 若 keys.json 没动但 cache 里某 acc 不
            // 再可见, 旧 resolved acc_id 仍留在 allowed set 中. 现在每次
            // expand 都从 file 源原始集合重新计算. raw 为 None → 空集起步.
            let mut acc_ids = rec.raw_explicit_acc_ids.clone().unwrap_or_default();
            for cn in &card_nums {
                let candidates = resolver(cn);
                match candidates.len() {
                    0 => {
                        unresolved += 1;
                        unresolved_callback(&rec.id, cn);
                    }
                    1 => {
                        acc_ids.insert(candidates[0]);
                        resolved += 1;
                    }
                    _ => {
                        ambiguous += 1;
                        ambiguous_callback(&rec.id, cn, &candidates);
                    }
                }
            }
            // v1.4.103 codex F1 (P1) fail-closed: 无论 acc_ids 是否非空, 都
            // **必须** 写入 Some(...) — 哪怕是空 HashSet (= "denylist 全部",
            // 限额引擎 step 0 acc_id 白名单非空 + 不含 ctx.acc_id → reject).
            //
            // 旧逻辑 (silent unrestricted): `if !acc_ids.is_empty() { rec.allowed_acc_ids = Some(acc_ids); }`
            // 当 caller 配置 allowed_card_nums 但**全部 unresolved/ambiguous** 时,
            // acc_ids 留空, allowed_acc_ids 仍 None → 限额引擎按 "无限制" 处理 →
            // 受限 key silent unrestricted (反模式 D / pitfall #45).
            //
            // 新逻辑: 只要 caller 显式写了 allowed_card_nums (说明 *intent* 是限制),
            // 就强制 Some(acc_ids) — 即便空集. 限额引擎检测到 Some(empty) 时
            // 视为 "全 reject" (限额 step 0 `allowed.is_empty()` 已 short-circuit
            // 不 reject, 但 contains check 永远 false → reject).
            //
            // **wait**: 看 limits.rs:332 `check_full_skip_rate`, step 0 是
            // `if let (Some(allowed), Some(id)) = (&limits.allowed_acc_ids, ctx.acc_id) && !allowed.is_empty() && !allowed.contains(&id)`.
            // 关键: `!allowed.is_empty()` short-circuit empty set, 等于 "无限制".
            // 这就是 silent-unrestricted 的根源. 但 `allowed.is_empty()` 短路是
            // **故意的语义** (允许 None / 空集都视为 "不限制") — 改这个会破坏
            // 现有 user contract.
            //
            // **正确修法**: 既然空集语义 = 无限制不能改, 我们要把 caller intent
            // (allowed_card_nums 非空) → 限额能识别 "想限但无法 resolve" 的状态.
            // 选: 把 sentinel acc_id (e.g. 0) 写入 acc_ids 触发 reject — 因为
            // 没有真账户 acc_id == 0. 限额检查时 acc_ids = {0}, ctx.acc_id =
            // <real id> ≠ 0 → reject. legitimate id 也 reject — 这是
            // fail-closed 保守语义.
            if !card_nums.is_empty() {
                if acc_ids.is_empty() {
                    // 全部 unresolved/ambiguous → 写 sentinel 0 让限额 reject 一切
                    acc_ids.insert(0u64);
                }
                rec.allowed_acc_ids = Some(acc_ids);
            } else if !acc_ids.is_empty() {
                rec.allowed_acc_ids = Some(acc_ids);
            }
        }
        let file = KeysFile {
            version: current.version,
            keys: new_keys,
        };
        let hash_index = Self::build_hash_index(&file);
        self.current.store(Arc::new(file));
        self.hash_index.store(Arc::new(hash_index));
        (resolved, unresolved, ambiguous)
    }

    /// 明文校验：遍历所有未过期 key，匹配则返回 KeyRecord 快照
    ///
    /// 如果 key 设置了 `allowed_machines` 且本机不在白名单，会打 warn 日志并视为未匹配。
    /// 这样做法的代价：攻击者可以通过"能不能过"侧信道区分 key 是否存在 — 我们接受，
    /// 因为 plaintext 空间是 256 bit 随机 hex，侧信道没意义。
    pub fn verify(&self, plaintext: &str) -> Option<Arc<KeyRecord>> {
        if !KeyRecord::is_generated_plaintext_shape(plaintext) {
            return None;
        }
        let computed_hash = hash_plaintext(plaintext);
        let index = self.hash_index.load();
        let now = Utc::now();
        let candidates = index.get(&computed_hash)?;
        for k in candidates {
            if k.is_expired(now) {
                continue;
            }
            if let Err(e) = k.check_machine() {
                let key_id = crate::metrics::redact_key_id_for_logs(&k.id);
                tracing::warn!(
                    key_id = %key_id,
                    error = %e,
                    "api key matched but machine binding failed; rejecting"
                );
                return None;
            }
            return Some(Arc::new(k.clone()));
        }
        None
    }

    /// 是否显式加载了 keys 文件。
    ///
    /// 注意：空 keys 文件也算 configured，用于让 surface 进入 scope mode
    /// 并 fail-closed；只有 [`Self::empty`] 才代表 legacy/no-key 模式。
    #[must_use]
    pub fn is_configured(&self) -> bool {
        self.path.is_some()
    }

    pub fn path(&self) -> Option<&Path> {
        self.path.as_deref()
    }

    #[must_use]
    pub fn len(&self) -> usize {
        self.current.load().keys.len()
    }

    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// v1.4.105 external reviewer #4 fix: 当前 KeyStore 是否有任意 key 配置了
    /// `allowed_card_nums` 限制. 用于 standalone MCP / gRPC / 任何不持
    /// `TrdCache` 的 keystore consumer 在启动时判断:
    /// - `false` → 没有 card_num 限制, 跳过 daemon `GetAccList` + expand 全流程
    ///   (避免无意义的 daemon 请求)
    /// - `true` → 必须连 daemon, 调 `GetAccList`, 通过
    ///   [`Self::expand_allowed_card_nums`] 把 card_num resolve 成 acc_id;
    ///   否则 fail-closed sentinel `{0}` 会让所有真账户 reject (external reviewer BUG
    ///   v1.4.104-002: standalone MCP 漏调 expand 导致 `0757` 配置的 key
    ///   全 reject).
    ///
    /// 注意: 本方法只检查 raw `allowed_card_nums` 是否非空 — load_file 阶段
    /// 注入的 sentinel `allowed_acc_ids = {0}` **不**算"已 expand"; 只有
    /// caller 真跑过 [`Self::expand_allowed_card_nums`] 后才会用 resolved
    /// acc_ids 覆盖 sentinel.
    #[must_use]
    pub fn has_any_card_num_restrictions(&self) -> bool {
        self.current
            .load()
            .keys
            .iter()
            .any(|k| k.allowed_card_nums.as_ref().is_some_and(|v| !v.is_empty()))
    }

    /// 按 id 查询当前快照中的 key（**不做 expiry / machine 校验**，调用方自己做）
    ///
    /// 典型用法：MCP 在启动时 `verify(plaintext)` 拿到 id，后续每个请求用
    /// `get_by_id` 取最新记录，这样 SIGHUP 重载 keys.json 后 scope / 限额 /
    /// expires_at 的变更能立刻生效（不用重启进程）。
    ///
    /// 返回 None 表示 id 在当前文件里不存在（被 remove_key 删掉了），调用方应
    /// 视为"key 已吊销"直接拒绝。
    ///
    /// **注意**：此方法**不做 machine binding 校验**。对于跨 SIGHUP 的 per-msg /
    /// per-tool 复检场景应改用 [`Self::get_by_id_for_current_machine`]，确保
    /// SIGHUP 后新加的 `allowed_machines` 限制立即生效（避免 startup 验过 →
    /// SIGHUP 收紧 → 仍按老 record 放行的语义漂移）。
    pub fn get_by_id(&self, id: &str) -> Option<Arc<KeyRecord>> {
        let snap = self.current.load_full();
        snap.keys
            .iter()
            .find(|k| k.id == id)
            .map(|k| Arc::new(k.clone()))
    }

    /// 按 id 查询当前快照中的 key + 立即校验本机 machine binding。
    ///
    /// **统一生命周期入口**: 任何 surface (WS / MCP / REST / gRPC) 在
    /// 已 verify-once → 跨 SIGHUP 复检场景下应使用此方法替代裸 [`Self::get_by_id`]，
    /// 避免如下漂移:
    ///
    /// - startup 时 `verify(plaintext)` 检查 machine binding ✅
    /// - SIGHUP reload 把该 key 的 `allowed_machines` 收紧（移除本机指纹）
    /// - 后续 per-msg / per-tool 仅调 `get_by_id` → **绕过 machine binding** →
    ///   silent unrestricted (反模式 D / pitfall #45 silent-success 同模式)
    ///
    /// 行为:
    /// - id 不存在 → `None`（key 已被 remove_key 吊销，caller 视为吊销拒绝）
    /// - id 存在 + machine 校验通过 → `Some(rec)`
    /// - id 存在 + machine 校验失败 → `None` + warn log（与 `verify` 同语义）
    ///
    /// **不做 expiry 校验** —— pipeline.rs Step 1.5 / caller 自己做（与
    /// `get_by_id` 行为对齐，仅差 machine 一层）。
    pub fn get_by_id_for_current_machine(&self, id: &str) -> Option<Arc<KeyRecord>> {
        let rec = self.get_by_id(id)?;
        if let Err(e) = rec.check_machine() {
            let key_id = crate::metrics::redact_key_id_for_logs(&rec.id);
            tracing::warn!(
                key_id = %key_id,
                error = %e,
                "api key get_by_id_for_current_machine: machine binding failed; \
                 treating as revoked (caller should reject as if key not found)"
            );
            return None;
        }
        Some(rec)
    }

    /// 导出当前所有 keys 的 id（用于调试 / 审计）
    #[must_use]
    pub fn ids(&self) -> Vec<String> {
        self.current
            .load()
            .keys
            .iter()
            .map(|k| k.id.clone())
            .collect()
    }
}

/// 追加一条新 key 到 keys.json（atomic rename）
/// v1.4.106 codex 0558 F5 (P2): RMW (read-modify-write) helper that holds the
/// flock for the **entire** sequence — load, mutate, write. Without this,
/// concurrent append_key callers can load → load → write → write and lose
/// the first writer's record.
fn with_keys_lock<F, R>(path: &Path, f: F) -> Result<R, KeyStoreError>
where
    F: FnOnce(&Path) -> Result<R, KeyStoreError>,
{
    if let Some(parent) = path.parent()
        && !parent.as_os_str().is_empty()
    {
        fs::create_dir_all(parent).map_err(|source| KeyStoreError::Write {
            path: parent.to_path_buf(),
            source,
        })?;
    }
    let _guard = AdvisoryLockGuard::acquire_exclusive(path)?;
    f(path)
}

pub fn append_key(path: &Path, new_record: KeyRecord) -> Result<(), KeyStoreError> {
    with_keys_lock(path, |path| {
        let mut file = match fs::metadata(path) {
            Ok(_) => KeyStore::load_file_unlocked(path)?,
            Err(_) => KeysFile {
                version: CURRENT_VERSION,
                keys: vec![],
            },
        };
        if file.keys.iter().any(|k| k.id == new_record.id) {
            return Err(KeyStoreError::DuplicateId(new_record.id));
        }
        file.keys.push(new_record);
        write_atomic_inner(path, &file)
    })
}

/// 读取 keys.json 并返回所有记录快照（展示用；不暴露 hash 以外的敏感位）
pub fn list_keys(path: &Path) -> Result<Vec<KeyRecord>, KeyStoreError> {
    let file = KeyStore::load_file(path)?;
    Ok(file.keys)
}

/// 按 id 编辑一条 key（atomic rename）；闭包返回 `false` 代表未改动 → 跳过落盘
///
/// 适用于就地修改 `allowed_machines` / `expires_at` / `note` 等配置，
/// 而不想走 "revoke + regen" 流程（否则 plaintext 会换）。
pub fn update_key<F>(path: &Path, id: &str, mutate: F) -> Result<bool, KeyStoreError>
where
    F: FnOnce(&mut KeyRecord) -> bool,
{
    // v1.4.106 F5: 整个 RMW 在 flock 内, 防 concurrent update_key 互相覆盖
    with_keys_lock(path, |path| {
        let mut file = KeyStore::load_file_unlocked(path)?;
        let Some(rec) = file.keys.iter_mut().find(|k| k.id == id) else {
            return Ok(false);
        };
        let changed = mutate(rec);
        if changed {
            write_atomic_inner(path, &file)?;
        }
        Ok(changed)
    })
}

/// 按 id 删除一条 key（atomic rename）；返回是否真的删掉了一条
pub fn remove_key(path: &Path, id: &str) -> Result<bool, KeyStoreError> {
    // v1.4.106 F5: 整个 RMW 在 flock 内
    with_keys_lock(path, |path| {
        let mut file = KeyStore::load_file_unlocked(path)?;
        let before = file.keys.len();
        file.keys.retain(|k| k.id != id);
        let removed = before != file.keys.len();
        if removed {
            write_atomic_inner(path, &file)?;
        }
        Ok(removed)
    })
}

/// v1.4.106 codex 0558 F5 (P2): atomic write with advisory flock + unique
/// tempfile + fsync.
///
/// **背景**: 之前的 write_atomic 有 3 个问题:
///   1. `tmp = path.with_extension("json.tmp")` — 同 path **每个 process 共
///      享**, 并发写 (daemon + futucli + multiple admin reload) 会互相覆写
///      tempfile, 最后 rename 时数据交错.
///   2. **无 advisory flock** — 没有跨 process coordination, race 可让 reader
///      看到部分写完的 `keys.json` (rename 前如果别的 process 也在 truncate).
///   3. **无 fsync** — 写完立刻 rename 在 ext4 数据=writeback 模式 / SSD power
///      loss 下可能丢内容 (file 在 inode 层面存在但 data block 没 flush).
///
/// **修法**: 1) tempfile 名带 pid + nanos: `keys.json.<pid>.<nanos>.tmp` — race-free.
/// 2) 对 `keys.json.lock` (sibling lock file) 取 LOCK_EX flock; 读路径
/// (load_file) 取 LOCK_SH; RMW 调用方在已经持有 LOCK_EX 时走
/// `load_file_unlocked` 避免同线程嵌套锁. 3) tempfile open 后 write_all →
/// sync_all → close → set_permissions → rename → 持有 lock 期间.
///
/// v1.4.106 codex 0558 F5: 写盘只做 unique tempfile + fsync + rename. flock 由
/// caller (with_keys_lock) 在 RMW 范围统一持有, 这里**不再**单独加锁 (避免
/// 与 with_keys_lock 重入).
fn write_atomic_inner(path: &Path, file: &KeysFile) -> Result<(), KeyStoreError> {
    let text = serde_json::to_string_pretty(file)?;

    // v1.4.106 F5: unique tempfile (pid + nanos), 防 concurrent rename 战 tempfile.
    let nanos = keystore_tempfile_nanos_or_zero();
    let tmp_name = match path.file_name().and_then(|n| n.to_str()) {
        Some(name) => format!(
            "{name}.{pid}.{nanos}.tmp",
            pid = std::process::id(),
            nanos = nanos,
        ),
        None => format!(
            "keys.{pid}.{nanos}.tmp",
            pid = std::process::id(),
            nanos = nanos,
        ),
    };
    let tmp = path
        .parent()
        .map(|p| p.join(&tmp_name))
        .unwrap_or_else(|| Path::new(&tmp_name).to_path_buf());

    // 写入 tempfile + fsync — 0600 mode 通过 OpenOptions (Unix) 创建时即生效.
    use std::io::Write;
    #[cfg(unix)]
    let mut f = {
        use std::os::unix::fs::OpenOptionsExt;
        fs::OpenOptions::new()
            .create_new(true)
            .write(true)
            .mode(0o600)
            .open(&tmp)
            .map_err(|source| KeyStoreError::Write {
                path: tmp.clone(),
                source,
            })?
    };
    #[cfg(not(unix))]
    let mut f = fs::OpenOptions::new()
        .create_new(true)
        .write(true)
        .open(&tmp)
        .map_err(|source| KeyStoreError::Write {
            path: tmp.clone(),
            source,
        })?;

    let write_res = f
        .write_all(text.as_bytes())
        .and_then(|_| f.sync_all())
        .map_err(|source| KeyStoreError::Write {
            path: tmp.clone(),
            source,
        });
    drop(f);

    if let Err(e) = write_res {
        if let Err(cleanup_err) = fs::remove_file(&tmp) {
            tracing::debug!(
                path = %tmp.display(),
                error = %cleanup_err,
                "keystore atomic write failed; tempfile cleanup also failed"
            );
        }
        return Err(e);
    }

    // 防御 chmod (Unix), OpenOptions.mode 已 0600, 但部分 fs / umask 异常时兜底.
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        if let Err(err) = fs::set_permissions(&tmp, fs::Permissions::from_mode(0o600)) {
            tracing::warn!(
                path = %tmp.display(),
                error = %err,
                "keystore tempfile chmod 0600 failed"
            );
        }
    }

    // atomic rename → tempfile 替换 inode, reader 看到的永远是完整 file.
    fs::rename(&tmp, path).map_err(|source| KeyStoreError::Write {
        path: path.to_path_buf(),
        source,
    })?;

    Ok(())
}

fn keystore_tempfile_nanos_or_zero() -> u128 {
    match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) {
        Ok(duration) => duration.as_nanos(),
        Err(err) => {
            tracing::warn!(
                error = ?err,
                "keystore wall clock is before UNIX_EPOCH; falling back to zero tempfile timestamp"
            );
            0
        }
    }
}

// ============================================================================
// v1.4.106 codex 0558 F5 (P2): advisory file lock helper (Unix flock-based)
// ============================================================================
//
// 用 sibling `<path>.lock` 文件作锁文件 (空文件 just for inode), LOCK_EX 时
// 阻塞其他 acquire_exclusive caller. drop 时 LOCK_UN 释放.
//
// 选择 sibling lock file (而非 lock keys.json 本身):
//   - 避免 lock + rename 互动 (rename 替换 inode, 老 fd 上的 lock 实际作用于
//     老 inode, 新 reader 拿不到锁信号)
//   - 跨 process 一致 — 任何 process open `<path>.lock` 拿同 inode

fn lock_path_for(path: &Path) -> PathBuf {
    path.parent()
        .map(|p| {
            let mut name = path
                .file_name()
                .and_then(|n| n.to_str())
                .unwrap_or("keys")
                .to_string();
            name.push_str(".lock");
            p.join(name)
        })
        .unwrap_or_else(|| PathBuf::from("keys.lock"))
}

#[cfg(unix)]
struct AdvisoryLockGuard {
    fd: std::os::fd::OwnedFd,
}

#[cfg(unix)]
impl AdvisoryLockGuard {
    fn acquire_exclusive(path: &Path) -> Result<Self, KeyStoreError> {
        Self::acquire(path, libc::LOCK_EX)
    }

    fn acquire_shared(path: &Path) -> Result<Self, KeyStoreError> {
        Self::acquire(path, libc::LOCK_SH)
    }

    fn acquire(path: &Path, operation: i32) -> Result<Self, KeyStoreError> {
        use std::os::fd::AsRawFd;
        use std::os::unix::fs::OpenOptionsExt;

        // sibling lock path: `<dir>/<file_name>.lock` (不复用
        // path.with_extension — 它替换最后一个扩展名, 我们要追加一个新扩展名).
        let lock_path = lock_path_for(path);

        let file = fs::OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .truncate(false)
            .mode(0o600)
            .open(&lock_path)
            .map_err(|source| KeyStoreError::Write {
                path: lock_path.clone(),
                source,
            })?;
        let raw = file.as_raw_fd();
        // SAFETY: flock(2) is async-signal-safe; we hold OwnedFd via File
        // until the guard drops, so fd is valid.
        let rc = unsafe { libc::flock(raw, operation) };
        if rc != 0 {
            return Err(KeyStoreError::Write {
                path: lock_path,
                source: std::io::Error::last_os_error(),
            });
        }
        let owned: std::os::fd::OwnedFd = file.into();
        Ok(Self { fd: owned })
    }
}

#[cfg(unix)]
impl Drop for AdvisoryLockGuard {
    fn drop(&mut self) {
        use std::os::fd::AsRawFd;
        // best-effort unlock on close; closing fd also implicitly releases lock.
        // SAFETY: fd remains valid while owned by this guard. `flock(LOCK_UN)`
        // only touches the live fd; if it fails, dropping the fd still releases
        // the advisory lock as the final fallback.
        let rc = unsafe { libc::flock(self.fd.as_raw_fd(), libc::LOCK_UN) };
        if rc != 0 {
            eprintln!(
                "futu-auth warning: keystore advisory lock unlock failed: {}",
                std::io::Error::last_os_error()
            );
        }
    }
}

#[cfg(windows)]
struct AdvisoryLockGuard {
    file: fs::File,
    overlapped: windows_sys::Win32::System::IO::OVERLAPPED,
    lock_path: PathBuf,
}

#[cfg(windows)]
impl AdvisoryLockGuard {
    fn acquire_exclusive(path: &Path) -> Result<Self, KeyStoreError> {
        use windows_sys::Win32::Storage::FileSystem::LOCKFILE_EXCLUSIVE_LOCK;

        Self::acquire(path, LOCKFILE_EXCLUSIVE_LOCK)
    }

    fn acquire_shared(path: &Path) -> Result<Self, KeyStoreError> {
        Self::acquire(path, 0)
    }

    fn acquire(path: &Path, flags: u32) -> Result<Self, KeyStoreError> {
        use std::os::windows::io::AsRawHandle;
        use windows_sys::Win32::Storage::FileSystem::LockFileEx;
        use windows_sys::Win32::System::IO::OVERLAPPED;

        let lock_path = lock_path_for(path);
        let file = fs::OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .truncate(false)
            .open(&lock_path)
            .map_err(|source| KeyStoreError::Write {
                path: lock_path.clone(),
                source,
            })?;

        let mut overlapped = OVERLAPPED::default();
        let raw = file.as_raw_handle() as windows_sys::Win32::Foundation::HANDLE;
        // SAFETY: `raw` is the live handle owned by `file`; `overlapped` is kept
        // inside the guard and reused for UnlockFileEx on drop. Lock length
        // covers the full file range from offset 0, matching Unix sibling-lock
        // semantics for the whole RMW critical section.
        let ok = unsafe { LockFileEx(raw, flags, 0, u32::MAX, u32::MAX, &mut overlapped) };
        if ok == 0 {
            return Err(KeyStoreError::Write {
                path: lock_path,
                source: std::io::Error::last_os_error(),
            });
        }

        Ok(Self {
            file,
            overlapped,
            lock_path,
        })
    }
}

#[cfg(windows)]
impl Drop for AdvisoryLockGuard {
    fn drop(&mut self) {
        use std::os::windows::io::AsRawHandle;
        use windows_sys::Win32::Storage::FileSystem::UnlockFileEx;

        let raw = self.file.as_raw_handle() as windows_sys::Win32::Foundation::HANDLE;
        // SAFETY: `raw` remains valid while `file` is held by this guard, and
        // `overlapped` is the same offset structure used for LockFileEx.
        let ok = unsafe { UnlockFileEx(raw, 0, u32::MAX, u32::MAX, &mut self.overlapped) };
        if ok == 0 {
            eprintln!(
                "futu-auth warning: keystore advisory lock unlock failed for {}: {}",
                self.lock_path.display(),
                std::io::Error::last_os_error()
            );
        }
    }
}

#[cfg(all(not(unix), not(windows)))]
struct AdvisoryLockGuard;

#[cfg(all(not(unix), not(windows)))]
impl AdvisoryLockGuard {
    fn acquire_exclusive(_path: &Path) -> Result<Self, KeyStoreError> {
        // Unknown platform fallback. Unix and Windows have real cross-process
        // advisory locks; other targets keep legacy single-process behavior.
        Ok(Self)
    }

    fn acquire_shared(_path: &Path) -> Result<Self, KeyStoreError> {
        // Unknown platform fallback. Unix and Windows have real cross-process
        // advisory locks; other targets keep legacy single-process behavior.
        Ok(Self)
    }
}

#[cfg(test)]
mod tests;