futu_auth/

store.rs

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

mod lock;

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};
use lock::AdvisoryLockGuard;

#[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 readers: 1, 2; writer: 2)")]
    UnsupportedVersion(u32),
    #[error("duplicate key id {0:?}")]
    DuplicateId(String),
    #[error(
        "cannot persist a KeyRecord with card-number restrictions and effective allowed_acc_ids but no file-source snapshot"
    )]
    AmbiguousAccountSource,
}

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

const CURRENT_VERSION: u32 = 2;

/// KeyStore：热可替换的 keys 集合
#[derive(Debug)]
pub struct KeyStore {
    path: Option<PathBuf>,
    current: ArcSwap<KeyStoreGeneration>,
    generation_writer: parking_lot::Mutex<()>,
}

#[derive(Debug)]
struct KeyStoreGeneration {
    version: u32,
    records: Vec<Arc<KeyRecord>>,
    by_hash: HashMap<String, Vec<Arc<KeyRecord>>>,
    by_id: HashMap<String, Arc<KeyRecord>>,
}

impl KeyStoreGeneration {
    fn from_file(file: KeysFile) -> Self {
        let mut records = Vec::with_capacity(file.keys.len());
        let mut by_hash: HashMap<String, Vec<Arc<KeyRecord>>> =
            HashMap::with_capacity(file.keys.len());
        let mut by_id = HashMap::with_capacity(file.keys.len());
        for record in file.keys {
            let record = Arc::new(record);
            by_hash
                .entry(record.hash.clone())
                .or_default()
                .push(Arc::clone(&record));
            by_id.insert(record.id.clone(), Arc::clone(&record));
            records.push(record);
        }
        Self {
            version: file.version,
            records,
            by_hash,
            by_id,
        }
    }

    fn to_file(&self) -> KeysFile {
        KeysFile {
            version: self.version,
            keys: self
                .records
                .iter()
                .map(|record| record.as_ref().clone())
                .collect(),
        }
    }
}

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

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

    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_runtime_file_unlocked(path)
    }

    /// Parse and validate the persisted source representation without adding
    /// runtime-only sentinels or card-number resolutions. RMW paths must use
    /// this view so derived authorization state can never leak back to disk.
    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 file: KeysFile =
            serde_json::from_str(&text).map_err(|source| KeyStoreError::Parse {
                path: path.to_path_buf(),
                source,
            })?;
        if !matches!(file.version, 1 | 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()));
            }
        }
        Ok(file)
    }

    fn load_runtime_file_unlocked(path: &Path) -> Result<KeysFile, KeyStoreError> {
        let mut file = Self::load_file_unlocked(path)?;
        // 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 _writer = self.generation_writer.lock();
        let file = Self::load_file(path)?;
        self.current
            .store(Arc::new(KeyStoreGeneration::from_file(file)));
        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 mut resolution_cache: HashMap<String, Vec<u64>> = HashMap::new();
        loop {
            let base = self.current.load_full();
            let mut file = base.to_file();
            let mut resolved = 0;
            let mut unresolved = 0;
            let mut ambiguous = 0;
            let mut unresolved_events = Vec::new();
            let mut ambiguous_events = Vec::new();
            for rec in &mut file.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 = resolution_cache
                        .entry(cn.clone())
                        .or_insert_with(|| resolver(cn));
                    match candidates.len() {
                        0 => {
                            unresolved += 1;
                            unresolved_events.push((rec.id.clone(), cn.clone()));
                        }
                        1 => {
                            acc_ids.insert(candidates[0]);
                            resolved += 1;
                        }
                        _ => {
                            ambiguous += 1;
                            ambiguous_events.push((rec.id.clone(), cn.clone(), candidates.clone()));
                        }
                    }
                }
                // 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 writer = self.generation_writer.lock();
            if !Arc::ptr_eq(&base, &self.current.load_full()) {
                drop(writer);
                continue;
            }
            self.current
                .store(Arc::new(KeyStoreGeneration::from_file(file)));
            drop(writer);

            for (key_id, card_num) in unresolved_events {
                unresolved_callback(&key_id, &card_num);
            }
            for (key_id, card_num, candidates) in ambiguous_events {
                ambiguous_callback(&key_id, &card_num, &candidates);
            }
            return (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 generation = self.current.load();
        let now = Utc::now();
        let candidates = generation.by_hash.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::clone(k));
        }
        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().records.len()
    }

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

    #[must_use]
    pub fn has_usable_scope(&self, scope: crate::Scope) -> bool {
        let now = Utc::now();
        self.current.load().records.iter().any(|record| {
            record.scopes.contains(&scope)
                && !record.is_expired(now)
                && record.check_machine().is_ok()
        })
    }

    /// 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()
            .records
            .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>> {
        self.current.load().by_id.get(id).map(Arc::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()
            .records
            .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, mut new_record: KeyRecord) -> Result<(), KeyStoreError> {
    if new_record.raw_explicit_acc_ids.is_none() {
        let has_card_nums = new_record
            .allowed_card_nums
            .as_ref()
            .is_some_and(|values| !values.is_empty());
        let has_effective_acc_ids = new_record
            .allowed_acc_ids
            .as_ref()
            .is_some_and(|values| !values.is_empty());
        if has_card_nums && has_effective_acc_ids {
            return Err(KeyStoreError::AmbiguousAccountSource);
        }
        // Public callers may construct KeyRecord directly instead of using
        // KeyRecord::generate. Snapshot their explicit file-source intent
        // before serialization. A card-number-constrained record with an
        // effective account set and no snapshot is ambiguous (it may be a
        // runtime sentinel/resolution), so the branch above rejects it.
        new_record.raw_explicit_acc_ids = new_record.allowed_acc_ids.clone();
    }
    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.version = CURRENT_VERSION;
        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 allowed_acc_ids_before = rec.allowed_acc_ids.clone();
        let changed = mutate(rec);
        if changed {
            if rec.allowed_acc_ids != allowed_acc_ids_before {
                // update_key is public and may intentionally edit the explicit
                // account whitelist. Only such an edit replaces the source
                // snapshot; unrelated bind/note updates must not persist the
                // runtime sentinel or card-number expansion result.
                rec.raw_explicit_acc_ids = rec.allowed_acc_ids.clone();
            }
            file.version = CURRENT_VERSION;
            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 {
            file.version = CURRENT_VERSION;
            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
        }
    }
}

#[cfg(test)]
mod tests;