canister-security

Canister 安全

这是什么

适用于 Motoko 和 Rust 中 IC canister 的安全模式。异步消息模型会带来 TOCTOU（检查时与使用时）漏洞：在 await 调用之间，状态可能发生变化。canister_inspect_message 不是可靠的安全边界。互联网上的任何人都可以通过发送更新调用来消耗你的 cycles。本技能提供可直接复制使用的正确模式，用于访问控制、防止重入、异步安全和回调陷阱处理。

前提条件

• Motoko：mops 包管理器，mops.toml 中的 core = "2.0.0"
• Rust：ic-cdk = "0.19"、candid = "0.10"

安全隐患

1. 依赖 canister_inspect_message 进行访问控制。 此钩子只在单个副本上运行，不具备完整共识。如果该副本是恶意的，它可以直接跳过检查，在不检查消息的情况下执行更新调用。它也不会用于 canister 间调用、查询调用或管理 canister 调用。务必在每个更新方法中重复进行访问检查。只能将 inspect_message 用作节省 cycles 的优化，绝不能将其视为安全边界。

2. 忘记拒绝匿名主体。 每个需要身份验证的端点都必须检查调用者不是匿名主体（2vxsx-fae）。在 Motoko 中使用 Principal.isAnonymous(caller)，在 Rust 中比较 msg_caller() != Principal::anonymous()。否则，未经身份验证的调用者可以调用受保护的方法；如果 canister 使用调用者主体作为身份键（例如，用于余额），匿名主体就会成为任何人都能使用的共享身份。

3. 在异步调用前读取状态，并假设调用后状态未变（TOCTOU）。 当 canister 对 canister 间调用执行 await 时，其他消息可以交错执行并修改状态。这是 IC 上 DeFi 攻击最关键的来源之一。使用按调用者加锁（CallerGuard 模式）来防止并发操作。对于金融操作，还应考虑 saga 模式（在 await 前扣除，失败时补偿）；但要正确实现并不简单，因为回调陷阱和调用超时等边界情况会导致结果不明确。

4. 在 pre_upgrade 中触发陷阱。 如果 pre_upgrade 触发陷阱（例如，序列化的数据过多，超出指令限制），canister 将永久无法升级。避免在堆中存储必须在升级期间序列化的大型数据结构。在 Rust 中，使用 ic-stable-structures 直接访问稳定内存。在 Motoko 中，persistent actor 声明会自动将所有 let 和 var 变量存储在稳定内存中，无需手动序列化。

5. 不监控 cycles 余额。 每个 canister 的默认 freezing_threshold 为 2,592,000 秒（约 30 天）。当 cycles 低于阈值储备时，canister 会被冻结（拒绝所有更新调用）。当 cycles 归零时，canister 会被卸载，其代码和内存会被移除，但 canister ID 和控制者仍然保留。真正的隐患在于没有主动监控并补充 cycles。对于持有重要状态的生产 canister，应提高冻结阈值并设置自动监控。

   # Check current settings (mainnet)
   icp canister settings show backend -e ic
   # Increase freezing threshold for high-value canisters
   icp canister settings update backend --freezing-threshold 7776000 -e ic  # 90 days
   

6. 只有一个控制者且没有备份。 如果你丢失了控制者身份的私钥，canister 将永远无法升级。没有恢复机制。务必添加备用控制者或治理 canister：

   icp canister settings update backend --add-controller <backup-principal> -e ic
   

   部署时，询问开发者是否有备用控制者主体可供添加。

7. 调用 fetchRootKey() 以支持本地开发。 fetchRootKey() 会在运行时从副本获取根公钥，并信任其返回的任何内容——如果这段代码路径在主网上运行，中间人就可以替换为其他密钥，破坏所有验证。切勿调用它，即使将其放在环境检查之后也不行；环境分支恰恰是这类调用泄漏到生产构建中的原因。请改为从 ic_env Cookie（IC_ROOT_KEY，通过 @icp-sdk/core/agent/canister-env 中的 safeGetCanisterEnv()）读取根密钥，并将其作为 rootKey 选项传给 HttpAgent.create()。静态站点／资产容器前端和 icp-cli Vite 开发服务器都会自动设置此 Cookie，因此本地和生产环境可以使用同一段代码，无需分支——Cookie 始终携带网络的根密钥；在主网上，该密钥与 Agent 中硬编码的密钥相同，因此传入它不会产生任何影响。完整用法请参阅 icp-cli 技能中的 references/binding-generation.md。

8. 公开未设防护的管理员方法。 任何人都可以调用每个更新方法。管理员方法（迁移、配置、铸造）必须明确检查调用者是否在允许名单中。系统没有内置的角色机制——你必须自行实现。务必支持撤销管理员权限——缺少撤销机制是常见的 bug 来源。

9. 在容器状态中存储机密。 标准应用子网上的容器内存可由节点运营者读取。切勿在容器状态中存储私钥、API 密钥或密码。对于链上机密管理，请使用 vetKD（阈值密钥派生）。

10. 允许用户不受限制地存储数据。 如果用户可以无限制地存储数据，攻击者就能填满 4 GiB Wasm 堆或稳定内存，导致容器无法运行。务必实施每用户存储配额，并验证输入大小。

11. 状态变更后在回调中触发陷阱。 如果容器在进行跨容器调用前修改了状态，而回调触发陷阱，则调用前的修改会保留，但回调中的修改会回滚。恶意被调用方可以利用这一点，跳过扣款等关键安全操作。应合理组织代码，确保关键状态变更发生在异步边界之前，并且在失败或触发陷阱时能正确回滚。使用 try/finally（Motoko）或 Drop 守卫（Rust）确保始终执行清理。清理代码应尽可能精简——清理过程中触发陷阱会再次造成同样的问题。可考虑使用 call_on_cleanup 处理回滚逻辑，并使用日志记录实现崩溃安全的状态转换。

12. 无界等待调用会阻止升级。 如果容器向不可信或存在 bug 的被调用方发起调用，而对方始终不响应，那么在等待未完成的响应期间，容器无法停止（因此也无法升级）。请使用有界等待调用（超时），确保无论被调用方的行为如何，调用都会在有限时间内完成。

工作原理

IC 安全模型

1. 更新调用通过共识执行——子网上的所有节点都会执行代码，并且必须对结果达成一致。标准应用子网有 13 个节点；系统子网和受托人子网的节点更多（28+）。这使更新调用能够防篡改，但速度较慢（约 2 秒）。
2. 查询调用在单个副本上运行——速度快（约 200 毫秒），但副本可能返回错误或恶意结果。副本签名查询可提供一定程度的缓解（响应副本会对响应签名），但要获得完全信任，请使用经过认证的数据，或对安全关键读取使用更新调用。
3. Canister 间调用是异步消息。从发送请求到收到响应期间，你的 canister 可以处理其他消息。状态可能在此期间发生变化（见上文的 TOCTOU 陷阱）。
4. **发生 trap 时回滚状态。**如果消息执行发生 trap，其所有状态更改都会回滚。对于 canister 间调用，首次执行（await 之前）和回调（await 之后）是两条独立消息——回调中的 trap 只会回滚回调所做的更改，而首次执行所做的更改会保留。因此，清理逻辑（例如释放锁）必须放在清理上下文中（finally/Drop），而不是常规回调代码中。

实现

Motoko

访问控制

使用 shared(msg) 模式以原子方式捕获部署者——无需单独调用 init()，也没有抢先调用的风险。

import Principal "mo:core/Principal";
import Set "mo:core/pure/Set";
import Runtime "mo:core/Runtime";

shared(msg) persistent actor class MyCanister() {

  // --- Authorization state ---
  // transient: recomputed on each install/upgrade from msg.caller (the controller)
  transient let owner = msg.caller;
  var admins : Set.Set<Principal> = Set.empty();

  // --- Guards ---

  func requireAuthenticated(caller : Principal) {
    if (Principal.isAnonymous(caller)) {
      Runtime.trap("anonymous caller not allowed");
    };
  };

  func requireOwner(caller : Principal) {
    requireAuthenticated(caller);
    if (caller != owner) {
      Runtime.trap("caller is not the owner");
    };
  };

  func requireAdmin(caller : Principal) {
    requireAuthenticated(caller);
    if (caller != owner and not Set.contains(admins, Principal.compare, caller)) {
      Runtime.trap("caller is not an admin");
    };
  };

  // --- Admin management ---

  public shared ({ caller }) func addAdmin(newAdmin : Principal) : async () {
    requireOwner(caller);
    admins := Set.add(admins, Principal.compare, newAdmin);
  };

  public shared ({ caller }) func removeAdmin(admin : Principal) : async () {
    requireOwner(caller);
    admins := Set.remove(admins, Principal.compare, admin);
  };

  // --- Endpoints ---

  public shared ({ caller }) func publicAction() : async Text {
    requireAuthenticated(caller);
    "ok";
  };

  public shared ({ caller }) func adminAction() : async () {
    requireAdmin(caller);
    // ... protected logic
  };
};

防止重入（CallerGuard 模式）

按调用方进行锁定，防止同一调用方在首次调用等待响应期间再次调用。必须在 finally 块中释放守卫——如果回调发生 trap，catch 中的状态更改会回滚，但 finally 会在清理上下文中运行，此时状态更改会保留。

import Map "mo:core/Map";
import Principal "mo:core/Principal";
import Error "mo:core/Error";
import Result "mo:core/Result";

// Inside the persistent actor class { ... }
// otherCanister is application-specific — replace with your canister reference.

  let pendingRequests = Map.empty<Principal, Bool>();

  func acquireGuard(principal : Principal) : Result.Result<(), Text> {
    if (Map.get(pendingRequests, Principal.compare, principal) != null) {
      return #err("already processing a request for this caller");
    };
    Map.add(pendingRequests, Principal.compare, principal, true);
    #ok;
  };

  func releaseGuard(principal : Principal) {
    ignore Map.delete(pendingRequests, Principal.compare, principal);
  };

  public shared ({ caller }) func doSomethingAsync() : async Result.Result<Text, Text> {
    requireAuthenticated(caller);

    // 1. Acquire per-caller lock — rejects concurrent calls from same principal
    switch (acquireGuard(caller)) {
      case (#err(msg)) { return #err(msg) };
      case (#ok) {};
    };

    // 2. Make inter-canister call
    try {
      let result = await otherCanister.someMethod();
      #ok(result)
    } catch (e) {
      #err("call failed: " # Error.message(e))
    } finally {
      // Runs in cleanup context even if the callback traps — changes here persist.
      releaseGuard(caller);
    };
  };

inspect_message（仅用于优化 cycle）

// Inside persistent actor { ... }
// Method variants must match your public methods

system func inspect(
  {
    caller : Principal;
    msg : {
      #adminAction : () -> ();
      #addAdmin : () -> Principal;
      #removeAdmin : () -> Principal;
      #publicAction : () -> ();
      #doSomethingAsync : () -> ();
    }
  }
) : Bool {
  switch (msg) {
    // Admin methods: reject anonymous to save cycles on Candid decoding
    case (#adminAction _) { not Principal.isAnonymous(caller) };
    case (#addAdmin _) { not Principal.isAnonymous(caller) };
    case (#removeAdmin _) { not Principal.isAnonymous(caller) };
    case (#doSomethingAsync _) { not Principal.isAnonymous(caller) };
    // Public methods: accept all
    case (_) { true };
  };
};

Rust

访问控制（使用 CDK 守卫模式）

guard 属性会在方法体执行前运行检查。如果守卫返回 Err，调用会在任何方法代码执行前被拒绝。这比在方法内部调用守卫函数更可靠——你不会忘记添加它。

use ic_cdk::{init, update};
use ic_cdk::api::msg_caller;
use candid::Principal;
use std::cell::RefCell;

thread_local! {
    static OWNER: RefCell<Principal> = RefCell::new(Principal::anonymous());
    static ADMINS: RefCell<Vec<Principal>> = RefCell::new(vec![]);
}

// --- Guards (for #[update(guard = "...")] attribute) ---
// Must return Result<(), String>. Err rejects the call.

fn require_authenticated() -> Result<(), String> {
    if msg_caller() == Principal::anonymous() {
        return Err("anonymous caller not allowed".to_string());
    }
    Ok(())
}

fn require_owner() -> Result<(), String> {
    require_authenticated()?;
    OWNER.with(|o| {
        if msg_caller() != *o.borrow() {
            return Err("caller is not the owner".to_string());
        }
        Ok(())
    })
}

fn require_admin() -> Result<(), String> {
    require_authenticated()?;
    let caller = msg_caller();
    let is_authorized = OWNER.with(|o| caller == *o.borrow())
        || ADMINS.with(|a| a.borrow().contains(&caller));
    if !is_authorized {
        return Err("caller is not an admin".to_string());
    }
    Ok(())
}

// --- Init ---

#[init]
fn init(owner: Principal) {
    OWNER.with(|o| *o.borrow_mut() = owner);
}

// --- Endpoints ---

#[update(guard = "require_authenticated")]
fn public_action() -> String {
    "ok".to_string()
}

#[update(guard = "require_admin")]
fn admin_action() {
    // ... protected logic — guard already validated caller
}

#[update(guard = "require_owner")]
fn add_admin(new_admin: Principal) {
    ADMINS.with(|a| a.borrow_mut().push(new_admin));
}

#[update(guard = "require_owner")]
fn remove_admin(admin: Principal) {
    ADMINS.with(|a| a.borrow_mut().retain(|p| p != &admin));
}

ic_cdk::export_candid!();

防止重入（CallerGuard 模式）

CallerGuard 使用 Drop trait，在守卫离开作用域时释放锁——包括回调发生 trap 时（从 ic-cdk 0.5.1 起，局部变量会在清理期间离开作用域）。切勿使用 let _ = CallerGuard::new(caller)?——这会立即丢弃守卫，使锁定失效。

use std::cell::RefCell;
use std::collections::BTreeSet;
use candid::Principal;
use ic_cdk::update;
use ic_cdk::api::msg_caller;
use ic_cdk::call::Call;

// other_canister_id is application-specific — replace with your canister reference.

thread_local! {
    static PENDING: RefCell<BTreeSet<Principal>> = RefCell::new(BTreeSet::new());
}

struct CallerGuard {
    principal: Principal,
}

impl CallerGuard {
    fn new(principal: Principal) -> Result<Self, String> {
        PENDING.with(|p| {
            if !p.borrow_mut().insert(principal) {
                return Err("already processing a request for this caller".to_string());
            }
            Ok(Self { principal })
        })
    }
}

impl Drop for CallerGuard {
    fn drop(&mut self) {
        PENDING.with(|p| {
            p.borrow_mut().remove(&self.principal);
        });
    }
}

#[update]
async fn do_something_async() -> Result<String, String> {
    let caller = msg_caller();
    if caller == Principal::anonymous() {
        return Err("anonymous caller not allowed".to_string());
    }

    // Acquire per-caller lock — rejects concurrent calls from same principal.
    // Drop releases lock even if callback traps.
    let _guard = CallerGuard::new(caller)?;

    // Make inter-canister call
    let response = Call::bounded_wait(other_canister_id(), "some_method")
        .await
        .map_err(|e| format!("call failed: {:?}", e))?;
    let result: String = response.candid()
        .map_err(|e| format!("decode failed: {:?}", e))?;

    Ok(result)
    // _guard dropped here → lock released
}

inspect_message（仅用于优化 cycle）

use ic_cdk::api::{accept_message, msg_caller, msg_method_name};
use candid::Principal;

/// Pre-filter to reduce cycle waste from spam.
/// Runs on ONE node. Can be bypassed. NOT a security check.
/// Always duplicate real access control inside each method or via guard attribute.
#[ic_cdk::inspect_message]
fn inspect_message() {
    let method = msg_method_name();
    match method.as_str() {
        // Admin methods: only accept from non-anonymous callers
        "admin_action" | "add_admin" | "remove_admin" | "do_something_async" => {
            if msg_caller() != Principal::anonymous() {
                accept_message();
            }
            // Silently reject anonymous — saves cycles on Candid decoding
        }
        // Public methods: accept all
        _ => accept_message(),
    }
}