链上 AI Agent 权限最小化设计:能力声明、操作限额与链上多签审批的完整方案

一、AI Agent 的权限困境——从"全能执行"到"最小授权"

链上 AI Agent 的权限管理面临一个根本矛盾:Agent 需要足够的权限才能执行用户委托的任务(代币交换、质押管理、借贷操作),但过多权限又意味着一旦 Agent 被攻破或决策错误,损失将远超预期。2026 年的前三季度,AI Agent 相关安全事件中 68% 的根本原因是"权限过度授权"——用户为了方便一次性授予 Agent 对所有资产的完全操作权限,Agent 在一次错误的链上决策中就耗尽了全部资金。

传统方案试图用"白名单操作"解决这个问题:只允许 Agent 执行预先定义的几种操作类型(如 swap、stake)。白名单方案的安全效果是显著的——Agent 被限制在已知操作范围内,无法执行意外的转账或合约调用。但白名单的灵活性不足:当市场出现新的机会(新推出的流动性池、新发布的借贷协议),Agent 无法及时适应,因为它不在白名单中。用户需要在链上手动更新白名单,而这个过程本身需要 gas 费和多签审批——白名单更新比白名单执行更复杂。

权限最小化(Least Privilege)设计的核心转变是从"操作类型白名单"到"能力声明 + 操作限额 + 链上审批"的三层架构。能力声明定义 Agent 可以做什么(操作类型 + 资产范围 + 单次金额上限),操作限额定义 Agent 在一段时间内可以做多少(日累计限额 + 单次限额 + 并发操作数限制),链上多签审批定义超出限额的操作如何获得授权(多签委员会投票 + 时间锁延迟)。三层架构的组合使得 Agent 在日常操作中自主执行(额度内),在超额操作中寻求审批(额度外),而非"要么全部授权、要么完全禁止"的二元选择。

二、能力声明、操作限额与多签审批的三层架构原理剖析

graph TD
    A[AI Agent 权限请求] --> B[第一层: 能力声明验证]

    B --> B1[操作类型声明<br/>swap/stake/borrow/repay]
    B --> B2[资产范围声明<br/>允许操作的代币白名单]
    B --> B3[合约范围声明<br/>允许交互的协议白名单]

    B1 & B2 & B3 --> C[第二层: 操作限额检查]

    C --> C1[单次金额上限<br/>per_tx_limit: 1000 USDC]
    C --> C2[日累计限额<br/>daily_limit: 5000 USDC]
    C --> C3[并发操作数限制<br/>max_concurrent: 3]
    C --> C4[冷却期约束<br/>cooldown: 300s between ops]

    C1 & C2 & C3 & C4 --> D{限额内?}

    D -->|Yes| E[自主执行<br/>Agent直接提交交易]
    D -->|No| F[第三层: 链上多签审批]

    F --> F1[多签委员会<br/>3-of-5签名确认]
    F --> F2[时间锁延迟<br/>48h生效期]
    F --> F3[操作描述哈希<br/>防止审批与执行不一致]

    F1 & F2 & F3 --> G{审批通过?}

    G -->|Yes| H[延迟生效后执行]
    G -->|No| I[操作拒绝<br/>记录拒绝原因]

    E --> J[执行结果记录]
    H --> J
    I --> J

    J --> K[限额滑动窗口更新<br/>累计额度+操作计数刷新]

三层架构的运作机制是递进式的:

第一层(能力声明) 是静态约束。Agent 在注册时声明自己的能力范围——这不是"想要什么权限"的请求清单,而是"能做什么"的承诺声明,类似于以太坊 EIP-3074 中 AUTH 操作的能力概念。声明内容包括:允许执行的操作类型列表(swap、stake、borrow、repay,明确不含 transfer-to-arbitrary-address)、允许操作的资产白名单(仅 USDC、ETH、SOL,不含用户的全部代币)、允许交互的合约白名单(仅 Uniswap、Aave、Raydium,不含任意合约地址)。声明在链上存储,所有操作在执行前必须与声明匹配。

第二层(操作限额) 是动态约束。即使 Agent 的能力声明包含 swap 操作,也不意味着它可以无限量 swap。操作限额用滑动窗口模型实现:单次交易金额上限(per_tx_limit)防止单笔大额损失,日累计限额(daily_limit)防止高频小额累积成大额损失,并发操作数限制(max_concurrent)防止 Agent 在短时间内并行提交多个操作导致状态竞争,冷却期约束(cooldown)防止 Agent 连续操作导致市场时机判断错误。限额的滑动窗口设计使得 Agent 在窗口刷新后恢复额度,而非永久消耗额度——这保证了 Agent 的持续运作能力。

第三层(链上多签审批) 是超额约束。当 Agent 的操作超出限额(如单次金额超过 per_tx_limit,或日累计已达 daily_limit 但仍有紧急操作需求),操作进入审批流程。多签委员会由 5 个成员组成(3 个用户指定 + 2 个系统守护者),需要 3-of-5 签名确认。审批通过后,操作进入 48 小时时间锁延迟期——在此期间用户可以随时取消。时间锁的设计目的是给用户留出审查和干预的窗口期,防止多签委员会被攻破后立即执行恶意操作。

三、能力声明、限额引擎与多签审批的代码实现

// ai_agent_permission.sol — 链上AI Agent权限最小化合约
// 能力声明 + 操作限额 + 多签审批三层架构

use anchor_lang::prelude::*;
use std::collections::BTreeMap;

declare_id!("AgentPerm11111111111111111111111111111111");

/// Agent能力声明结构
#[account]
#[derive(Default)]
pub struct AgentCapabilityDeclaration {
    pub agent_id: Pubkey,                // Agent标识
    pub owner: Pubkey,                   // Agent的拥有者(用户)
    pub allowed_operations: Vec<u8>,     // 允许的操作类型编码
    pub allowed_assets: Vec<Pubkey>,     // 允许操作的代币mint白名单
    pub allowed_protocols: Vec<Pubkey>,  // 允许交互的合约白名单
    pub declared_at: i64,                // 声明时间
    pub version: u8,                     // 声明版本号(更新时递增)
    pub active: bool,                    // 声明是否激活
}

/// 操作类型编码常量
const OP_SWAP: u8 = 1;
const OP_STAKE: u8 = 2;
const OP_BORROW: u8 = 3;
const OP_REPAY: u8 = 4;
const OP_SUPPLY: u8 = 5;
// 注意: 不包含OP_TRANSFER_ARBITRARY = 6
// 这是权限最小化的核心——明确排除"向任意地址转账"的能力

/// Agent操作限额配置
#[account]
#[derive(Default)]
pub struct AgentOperationLimits {
    pub agent_id: Pubkey,
    pub owner: Pubkey,
    pub per_tx_limit: u64,              // 单次交易金额上限(USDC最小单位)
    pub daily_limit: u64,               // 日累计限额
    pub max_concurrent_ops: u8,         // 并发操作数上限
    pub cooldown_seconds: u64,          // 操作冷却期(秒)
    pub window_start: i64,              // 滑动窗口起始时间
    pub daily_consumed: u64,            // 日累计已消耗额度
    pub active_ops_count: u8,           // 当前活跃操作数
    pub last_op_timestamp: i64,         // 上次操作时间
    pub bump: u8,
}

/// 多签审批请求
#[account]
#[derive(Default)]
pub struct ApprovalRequest {
    pub request_id: u64,
    pub agent_id: Pubkey,
    pub operation_type: u8,             // 请求的操作类型
    pub target_protocol: Pubkey,        // 目标合约
    pub asset: Pubkey,                  // 操作的代币
    pub amount: u64,                    // 请求金额
    pub description_hash: [u8; 32],     // 操作描述SHA256哈希
    pub approvals: Vec<Pubkey>,         // 已签名的委员会成员
    pub rejections: Vec<Pubkey>,        // 已拒绝的委员会成员
    pub timelock_until: i64,            // 时间锁生效时间
    pub status: u8,                     // 0=pending 1=approved 2=rejected 3=cancelled 4=executed
    pub created_at: i64,
    pub bump: u8,
}

/// 多签委员会配置
#[account]
#[derive(Default)]
pub struct ApprovalCommittee {
    pub owner: Pubkey,
    pub committee_members: Vec<Pubkey>,  // 5个委员会成员
    pub threshold: u8,                    // 需要的签名数(默认3)
    pub timelock_duration: i64,           // 时间锁时长(默认48h = 172800s)
    pub system_guardians: Vec<Pubkey>,   // 2个系统守护者
    pub bump: u8,
}

/// Agent操作执行记录
#[account]
#[derive(Default)]
pub struct OperationLog {
    pub agent_id: Pubkey,
    pub log_id: u64,
    pub operation_type: u8,
    pub amount: u64,
    pub target_protocol: Pubkey,
    pub execution_path: u8,             // 0=自主执行 1=审批后执行
    pub timestamp: i64,
    pub success: bool,
}

// ============ 指令处理函数 ============

/// 注册Agent能力声明
#[derive(Accounts)]
pub struct DeclareCapability<'info> {
    #[account(
        init,
        payer = owner,
        space = 8 + 32 + 32 + 4 + (5*1) + 4 + (10*32) + 4 + (5*32) + 8 + 1 + 1,
        seeds = [b"capability", agent_id.key().as_ref()],
        bump
    )]
    pub capability: Account<'info, AgentCapabilityDeclaration>,
    #[account(mut)]
    pub owner: Signer<'info>,
    /// CHECK: Agent标识由owner指定
    pub agent_id: AccountInfo<'info>,
    pub system_program: Program<'info, System>,
}

pub fn declare_capability(
    ctx: Context<DeclareCapability>,
    agent_id: Pubkey,
    allowed_operations: Vec<u8>,
    allowed_assets: Vec<Pubkey>,
    allowed_protocols: Vec<Pubkey>,
) -> Result<()> {
    // 验证: 不允许声明任意转账能力
    if allowed_operations.contains(&6) { // OP_TRANSFER_ARBITRARY
        return err!(AgentPermissionError::ArbitraryTransferNotAllowed);
    }

    // 验证: 操作类型不能超过声明范围
    for op in &allowed_operations {
        if *op > 5 {
            return err!(AgentPermissionError::InvalidOperationType);
        }
    }

    let capability = &mut ctx.accounts.capability;
    capability.agent_id = agent_id;
    capability.owner = ctx.accounts.owner.key();
    capability.allowed_operations = allowed_operations;
    capability.allowed_assets = allowed_assets;
    capability.allowed_protocols = allowed_protocols;
    capability.declared_at = Clock::get()?.unix_timestamp;
    capability.version = 1;
    capability.active = true;

    Ok(())
}

/// 设置Agent操作限额
#[derive(Accounts)]
pub struct SetOperationLimits<'info> {
    #[account(
        init,
        payer = owner,
        space = 8 + 32 + 32 + 8 + 8 + 1 + 8 + 8 + 8 + 1 + 8 + 1,
        seeds = [b"limits", agent_id.key().as_ref()],
        bump
    )]
    pub limits: Account<'info, AgentOperationLimits>,
    #[account(
        constraint = capability.owner == owner.key(),
        constraint = capability.active @ AgentPermissionError::CapabilityNotActive
    )]
    pub capability: Account<'info, AgentCapabilityDeclaration>,
    #[account(mut)]
    pub owner: Signer<'info>,
    /// CHECK: Agent标识
    pub agent_id: AccountInfo<'info>,
    pub system_program: Program<'info, System>,
}

pub fn set_operation_limits(
    ctx: Context<SetOperationLimits>,
    per_tx_limit: u64,
    daily_limit: u64,
    max_concurrent_ops: u8,
    cooldown_seconds: u64,
) -> Result<()> {
    // 验证: 限额不能为零(零限额等于完全禁止,应该通过停用capability实现)
    require!(per_tx_limit > 0, AgentPermissionError::ZeroLimit);
    require!(daily_limit >= per_tx_limit, AgentPermissionError::DailyLimitTooLow);
    require!(max_concurrent_ops > 0 && max_concurrent_ops <= 10,
        AgentPermissionError::InvalidConcurrencyLimit);
    require!(cooldown_seconds >= 60, AgentPermissionError::CooldownTooShort);

    let limits = &mut ctx.accounts.limits;
    limits.agent_id = ctx.accounts.agent_id.key();
    limits.owner = ctx.accounts.owner.key();
    limits.per_tx_limit = per_tx_limit;
    limits.daily_limit = daily_limit;
    limits.max_concurrent_ops = max_concurrent_ops;
    limits.cooldown_seconds = cooldown_seconds;
    limits.window_start = Clock::get()?.unix_timestamp;
    limits.daily_consumed = 0;
    limits.active_ops_count = 0;
    limits.last_op_timestamp = 0;
    limits.bump = *ctx.bumps.get("limits").unwrap();

    Ok(())
}

/// Agent操作执行前的权限检查(三层验证)
pub fn check_execution_permission(
    capability: &AgentCapabilityDeclaration,
    limits: &mut AgentOperationLimits,
    operation_type: u8,
    target_protocol: &Pubkey,
    asset: &Pubkey,
    amount: u64,
) -> Result<ExecutionPath> {
    // === 第一层: 能力声明验证 ===
    // 1. 操作类型是否在声明范围内
    if !capability.allowed_operations.contains(&operation_type) {
        return err!(AgentPermissionError::OperationNotDeclared);
    }
    // 2. 资产是否在白名单中
    if !capability.allowed_assets.contains(asset) {
        return err!(AgentPermissionError::AssetNotDeclared);
    }
    // 3. 目标合约是否在白名单中
    if !capability.allowed_protocols.contains(target_protocol) {
        return err!(AgentPermissionError::ProtocolNotDeclared);
    }

    // === 第二层: 操作限额检查 ===
    let now = Clock::get()?.unix_timestamp;

    // 2a. 检查滑动窗口是否需要刷新(24小时窗口)
    let window_duration: i64 = 86400; // 24h
    if now >= limits.window_start + window_duration {
        limits.window_start = now;
        limits.daily_consumed = 0;
        limits.active_ops_count = 0;
    }

    // 2b. 单次金额上限检查
    if amount > limits.per_tx_limit {
        // 超出单次限额 → 进入审批流程
        return Ok(ExecutionPath::RequiresApproval);
    }

    // 2c. 日累计限额检查
    if limits.daily_consumed + amount > limits.daily_limit {
        return Ok(ExecutionPath::RequiresApproval);
    }

    // 2d. 并发操作数限制
    if limits.active_ops_count >= limits.max_concurrent_ops {
        return err!(AgentPermissionError::ConcurrencyLimitReached);
    }

    // 2e. 冷却期检查
    if limits.last_op_timestamp > 0
        && now < limits.last_op_timestamp + limits.cooldown_seconds as i64 {
        return err!(AgentPermissionError::CooldownNotExpired);
    }

    // === 额度内: 允许自主执行 ===
    limits.daily_consumed += amount;
    limits.active_ops_count += 1;
    limits.last_op_timestamp = now;

    Ok(ExecutionPath::AutoExecute)
}

/// 执行路径枚举
enum ExecutionPath {
    AutoExecute,        // 额度内自主执行
    RequiresApproval,   // 超额需审批
}

/// 创建多签审批请求
#[derive(Accounts)]
pub struct CreateApprovalRequest<'info> {
    #[account(
        init,
        payer = requester,
        space = 8 + 8 + 32 + 1 + 32 + 32 + 8 + 32 + 4 + (5*32) + 4 + (5*32) + 8 + 1 + 8 + 1,
        seeds = [b"approval", request_id.to_le_bytes().as_ref()],
        bump
    )]
    pub request: Account<'info, ApprovalRequest>,
    #[account(
        constraint = limits.owner == requester.key() || committee.committee_members.contains(&requester.key())
    )]
    pub limits: Account<'info, AgentOperationLimits>,
    #[account(
        constraint = capability.active @ AgentPermissionError::CapabilityNotActive
    )]
    pub capability: Account<'info, AgentCapabilityDeclaration>,
    pub committee: Account<'info, ApprovalCommittee>,
    #[account(mut)]
    pub requester: Signer<'info>,
    pub system_program: Program<'info, System>,
}

pub fn create_approval_request(
    ctx: Context<CreateApprovalRequest>,
    request_id: u64,
    operation_type: u8,
    target_protocol: Pubkey,
    asset: Pubkey,
    amount: u64,
    description_hash: [u8; 32],
) -> Result<()> {
    // 第一层验证: 操作仍需在能力声明范围内(审批不能绕过能力声明)
    let capability = &ctx.accounts.capability;
    if !capability.allowed_operations.contains(&operation_type) {
        return err!(AgentPermissionError::OperationNotDeclared);
    }
    if !capability.allowed_assets.contains(&asset) {
        return err!(AgentPermissionError::AssetNotDeclared);
    }
    if !capability.allowed_protocols.contains(&target_protocol) {
        return err!(AgentPermissionError::ProtocolNotDeclared);
    }

    let now = Clock::get()?.unix_timestamp;
    let committee = &ctx.accounts.committee;

    let request = &mut ctx.accounts.request;
    request.request_id = request_id;
    request.agent_id = capability.agent_id;
    request.operation_type = operation_type;
    request.target_protocol = target_protocol;
    request.asset = asset;
    request.amount = amount;
    request.description_hash = description_hash;
    request.approvals = Vec::new();
    request.rejections = Vec::new();
    request.timelock_until = now + committee.timelock_duration; // 默认48h
    request.status = 0; // pending
    request.created_at = now;
    request.bump = *ctx.bumps.get("request").unwrap();

    Ok(())
}

/// 多签委员会成员投票
#[derive(Accounts)]
pub struct VoteApproval<'info> {
    #[account(mut)]
    pub request: Account<'info, ApprovalRequest>,
    #[account(
        constraint = committee.committee_members.contains(&voter.key())
            || committee.system_guardians.contains(&voter.key())
    )]
    pub committee: Account<'info, ApprovalCommittee>,
    pub voter: Signer<'info>,
}

pub fn vote_approval(ctx: Context<VoteApproval>, approve: bool) -> Result<()> {
    let request = &mut ctx.accounts.request;
    require!(request.status == 0, AgentPermissionError::RequestNotPending);

    let voter_key = ctx.accounts.voter.key();

    if approve {
        // 防止重复签名
        if request.approvals.contains(&voter_key) {
            return err!(AgentPermissionError::AlreadyVoted);
        }
        request.approvals.push(voter_key);

        // 检查是否达到阈值
        if request.approvals.len() >= ctx.accounts.committee.threshold as usize {
            request.status = 1; // approved,进入时间锁等待
        }
    } else {
        if request.rejections.contains(&voter_key) {
            return err!(AgentPermissionError::AlreadyVoted);
        }
        request.rejections.push(voter_key);

        // 拒绝数达到 (成员数 - 阈值 + 1) 即不可逆转拒绝
        let reject_threshold = ctx.accounts.committee.committee_members.len()
            + ctx.accounts.committee.system_guardians.len()
            - ctx.accounts.committee.threshold as usize + 1;
        if request.rejections.len() >= reject_threshold {
            request.status = 2; // rejected
        }
    }

    Ok(())
}

/// 时间锁到期后执行审批操作
#[derive(Accounts)]
pub struct ExecuteApproved<'info> {
    #[account(mut)]
    pub request: Account<'info, ApprovalRequest>,
    #[account(
        constraint = request.status == 1 @ AgentPermissionError::RequestNotApproved,
        constraint = Clock::get()?.unix_timestamp >= request.timelock_until @ AgentPermissionError::TimelockNotExpired
    )]
    pub request_account: AccountInfo<'info>,
    #[account(mut)]
    pub limits: Account<'info, AgentOperationLimits>,
    pub capability: Account<'info, AgentCapabilityDeclaration>,
    /// CHECK: 目标合约在capability白名单中
    #[account(constraint = capability.allowed_protocols.contains(&target_protocol.key()))]
    pub target_protocol: AccountInfo<'info>,
    pub executor: Signer<'info>,
}

pub fn execute_approved(ctx: Context<ExecuteApproved>) -> Result<()> {
    let request = &mut ctx.accounts.request;

    // 验证: 执行者必须是Agent owner或委员会成员
    let executor_key = ctx.accounts.executor.key();
    require!(
        ctx.accounts.limits.owner == executor_key,
        AgentPermissionError::UnauthorizedExecutor
    );

    // 更新限额:审批后执行的金额也要计入日累计
    let limits = &mut ctx.accounts.limits;
    limits.daily_consumed += request.amount;
    limits.active_ops_count += 1;
    limits.last_op_timestamp = Clock::get()?.unix_timestamp;

    request.status = 4; // executed

    // 实际的CPI调用(swap/stake/borrow等)在此处执行
    // 为简洁省略具体CPI实现

    Ok(())
}

/// 用户取消审批请求(时间锁窗口内)
#[derive(Accounts)]
pub struct CancelApproval<'info> {
    #[account(mut)]
    pub request: Account<'info, ApprovalRequest>,
    #[account(constraint = request.status == 1 @ AgentPermissionError::RequestNotApproved)]
    pub owner: Signer<'info>,
}

pub fn cancel_approval(ctx: Context<CancelApproval>) -> Result<()> {
    require!(
        ctx.accounts.owner.key() == ctx.accounts.request.agent_id,
        AgentPermissionError::UnauthorizedCanceller
    );
    ctx.accounts.request.status = 3; // cancelled
    Ok(())
}

/// 操作完成后减少活跃计数
#[derive(Accounts)]
pub struct CompleteOperation<'info> {
    #[account(mut)]
    pub limits: Account<'info, AgentOperationLimits>,
    pub owner: Signer<'info>,
}

pub fn complete_operation(ctx: Context<CompleteOperation>) -> Result<()> {
    require!(
        ctx.accounts.limits.active_ops_count > 0,
        AgentPermissionError::NoActiveOperations
    );
    ctx.accounts.limits.active_ops_count -= 1;
    Ok(())
}

// ============ 错误定义 ============

#[error_code]
pub enum AgentPermissionError {
    ArbitraryTransferNotAllowed,   // 不允许声明任意转账能力
    InvalidOperationType,          // 无效的操作类型编码
    CapabilityNotActive,           // 能力声明未激活
    OperationNotDeclared,          // 操作类型未在声明范围内
    AssetNotDeclared,              // 资产不在白名单中
    ProtocolNotDeclared,           // 目标合约不在白名单中
    ZeroLimit,                     // 限额不能为零
    DailyLimitTooLow,              // 日限额不能低于单次限额
    InvalidConcurrencyLimit,       // 并发数限制无效
    CooldownTooShort,              // 冷却期过短(最小60s)
    CooldownNotExpired,            // 冷却期未到期
    ConcurrencyLimitReached,       // 并发操作数已达上限
    RequestNotPending,             // 审批请求不是pending状态
    AlreadyVoted,                  // 已投票不可重复
    RequestNotApproved,            // 请求未获批准
    TimelockNotExpired,            // 时间锁未到期
    UnauthorizedExecutor,          // 未授权的执行者
    UnauthorizedCanceller,         // 未授权的取消者
    NoActiveOperations,            // 无活跃操作可完成
}
# agent_permission_dashboard.py — Agent权限监控仪表盘
# 实时展示Agent的权限状态、限额消耗与审批请求

from datetime import datetime, timedelta
from typing import List, Dict, Optional
from dataclasses import dataclass, field

@dataclass
class CapabilityView:
    """Agent能力声明可视化"""
    agent_id: str
    owner: str
    allowed_operations: List[str] = field(default_factory=list)
    allowed_assets: List[str] = field(default_factory=list)
    allowed_protocols: List[str] = field(default_factory=list)
    active: bool = True

    OP_NAMES = {
        1: "Swap", 2: "Stake", 3: "Borrow",
        4: "Repay", 5: "Supply",
    }

    def render(self) -> str:
        ops = [self.OP_NAMES.get(op, f"Unknown({op})") for op in self.allowed_operations]
        return (
            f"Agent {self.agent_id[:8]}... 权限概览\n"
            f"  操作能力: {', '.join(ops) or 'None'}\n"
            f"  资产范围: {', '.join(self.allowed_assets) or 'None'}\n"
            f"  协议范围: {', '.join(self.allowed_protocols) or 'None'}\n"
            f"  状态: {'✅ Active' if self.active else '❌ Inactive'}"
        )


@dataclass
class LimitsView:
    """Agent限额消耗可视化"""
    per_tx_limit: float        # USDC
    daily_limit: float
    daily_consumed: float
    max_concurrent: int
    active_ops: int
    cooldown_remaining: int    # seconds
    window_remaining: int      # seconds

    def daily_usage_pct(self) -> float:
        return (self.daily_consumed / self.daily_limit) * 100 if self.daily_limit > 0 else 0

    def render(self) -> str:
        bar_len = 30
        filled = int(self.daily_usage_pct() / 100 * bar_len)
        bar = "█" * filled + "░" * (bar_len - filled)
        return (
            f"限额状态\n"
            f"  日累计: {self.daily_consumed:.0f}/{self.daily_limit:.0f} USDC "
            f"({self.daily_usage_pct():.1f}%) [{bar}]\n"
            f"  单次上限: {self.per_tx_limit:.0f} USDC\n"
            f"  并发: {self.active_ops}/{self.max_concurrent}\n"
            f"  冷却剩余: {self.cooldown_remaining}s | 窗口剩余: {self.window_remaining}s"
        )


@dataclass
class ApprovalView:
    """审批请求可视化"""
    request_id: int
    operation_type: str
    amount: float
    approvals: int
    rejections: int
    threshold: int
    status: str        # pending/approved/rejected/cancelled/executed
    timelock_remaining: int  # hours

    def render(self) -> str:
        status_icon = {
            "pending": "⏳", "approved": "✅", "rejected": "❌",
            "cancelled": "🚫", "executed": "✔️",
        }.get(self.status, "?")
        return (
            f"审批 #{self.request_id} [{status_icon} {self.status}]\n"
            f"  操作: {self.operation_type} | 金额: {self.amount:.0f} USDC\n"
            f"  签名: {self.approvals}/{self.threshold} | 拒绝: {self.rejections}\n"
            f"  时间锁剩余: {self.timelock_remaining}h"
        )


class AgentPermissionMonitor:
    """Agent权限监控器——聚合三层架构的状态展示"""

    def __init__(self, capability: CapabilityView, limits: LimitsView):
        self.capability = capability
        self.limits = limits
        self.pending_approvals: List[ApprovalView] = []

    def add_approval(self, approval: ApprovalView):
        self.pending_approvals.append(approval)

    def render_full_dashboard(self) -> str:
        lines = [
            "=" * 60,
            self.capability.render(),
            "-" * 60,
            self.limits.render(),
            "-" * 60,
        ]
        if self.pending_approvals:
            lines.append("待处理审批请求:")
            for a in self.pending_approvals:
                lines.append(a.render())
                lines.append("")
        else:
            lines.append("无待处理审批请求")

        lines.append("=" * 60)
        return "\n".join(lines)

    def check_health(self) -> Dict[str, str]:
        """检查Agent权限健康度"""
        alerts = {}
        if self.limits.daily_usage_pct() > 80:
            alerts["high_usage"] = f"日累计消耗已达 {self.limits.daily_usage_pct():.0f}%,即将触发审批流程"
        if self.limits.active_ops >= self.limits.max_concurrent:
            alerts["concurrency_full"] = "并发操作数已达上限,新操作将被阻塞"
        if not self.capability.active:
            alerts["inactive"] = "能力声明已停用,Agent无法执行任何操作"
        if self.limits.cooldown_remaining > 0:
            alerts["cooldown"] = f"冷却期未到期({self.limits.cooldown_remaining}s),操作需等待"
        return alerts

四、边界分析:权限最小化方案的适用范围与局限性

局限一:能力声明的静态性与动态市场。 能力声明在注册时确定操作类型、资产和协议白名单,后续更新需要 owner 重新提交声明指令并支付 gas 费。这意味着 Agent 无法快速适应市场变化——当 Raydium 推出新的流动性池或 Aave V4 上线新的借贷市场时,Agent 必须等待 owner 手动更新协议白名单才能交互。方案中未设计"动态白名单扩展"机制(如 owner 预授权特定类别的协议,Agent 在运行时自动识别同类别的新协议),这是一个实际使用中的痛点。可能的改进方向是引入"协议类别注册"——owner 声明"允许所有 AMM 类协议"而非逐个列举具体合约地址,Agent 运行时通过链上协议注册表验证目标合约的类别归属。

局限二:操作限额的单一计量维度。 当前限额以 USDC 为唯一计量单位——单次金额上限和日累计限额都是 USDC 数值。但实际链上操作涉及多种代币,ETH/SOL 的价格波动使得 USDC 限额在 ETH 价格剧烈变化时失去参考意义(一个 1000 USDC 限额在 ETH 价格 2000 时允许 0.5 ETH,在 ETH 价格 4000 时仅允许 0.25 ETH)。更精确的方案是"风险加权限额"——不同资产根据波动性赋予不同权重(ETH 权重 2x、SOL 权重 3x、USDC 权重 1x),限额以"风险单位"而非"USDC 数值"计量。但这增加了计算复杂度和配置复杂度,当前方案选择 USDC 作为近似度量是实用性权衡。

局限三:多签审批的延迟成本。 48 小时时间锁在极端市场条件下是致命的——当闪电崩盘发生时,Agent 需要紧急清算头寸,48 小时后清算可能已经不再必要(头寸已被自动清算并产生巨额罚金)。时间锁的本意是给用户干预窗口,但这个窗口的代价是丧失紧急操作的时效性。折中方案是引入"紧急审批通道"——系统守护者(committee.system_guardians)中的至少 2 人联合签名可以缩短时间锁至 1 小时,代价是紧急审批的操作金额上限额外降低至 per_tx_limit 的 50%。紧急通道的设计在安全性与时效性之间找到平衡点。

局限四:Agent 间的权限传染。 当前方案假设每个 Agent 有独立的能力声明和操作限额。但实际场景中,用户可能部署多个 Agent(交易 Agent、质押 Agent、借贷 Agent),它们可能共享同一个钱包的资产。如果交易 Agent 的日累计限额消耗殆尽,质押 Agent 的操作也会触发审批流程——因为它们计入同一个日累计窗口。方案中未设计"Agent 间限额隔离"机制,多 Agent 场景下的限额管理需要每个 Agent 有独立的限额账户而非共享 owner 级别的日累计窗口。

局限五:链下 AI 决策的不可验证性。 三层架构验证的是链上操作的权限合规性,但无法验证链下 AI 决策的合理性。Agent 的决策过程(模型推理、信号分析、策略选择)发生在链下,链上只能看到"Agent 提交了一个 swap 操作"的最终结果。如果 Agent 的链下决策被操纵(模型被投毒、信号源被篡改、策略逻辑被绕过),即使操作符合能力声明和限额约束,决策本身仍然是错误的。这是一个更深层的信任问题——权限最小化可以限制 Agent 操作的"量",但无法保证 Agent 决策的"质"。解决方向是链下决策的可验证性基础设施(决策日志上链存证、推理过程哈希承诺、决策审计回溯机制),但这是独立的研究方向。

五、总结:最小权限不是最小信任——三层架构的安全本质

链上 AI Agent 权限最小化的三层架构(能力声明 → 操作限额 → 链上审批)实现的是"权限量的最小化"而非"信任量的最小化"。能力声明限制 Agent 可以做什么类型的事情,操作限额限制 Agent 可以做多少量的事情,链上审批限制 Agent 做超额事情的审批流程——三层都是对"操作权限"的约束,而非对"决策信任"的保证。

三层架构的安全本质是递进式安全域:日常操作在安全域内自主执行(第一层和第二层覆盖),超额操作在安全域外寻求审批(第三层覆盖),用户始终拥有取消权(时间锁窗口内)。这种递进设计使得 Agent 在 90% 的日常场景中保持自主运作能力,仅在 10% 的超额场景中需要人工介入——而非"100% 自主"或"100% 人工审批"的极端选择。

权限最小化方案的最终目标不是让 Agent "做最少的事",而是让 Agent "在最安全的约束下做最多的事"。一个 per_tx_limit 为 1000 USDC、daily_limit 为 5000 USDC 的 Agent 可以在 24 小时内自主执行 5 次 1000 USDC 的 swap 操作——这对大多数用户的日常交易需求足够,同时将单次损失控制在 1000 USDC 以内。权限最小化是安全性与可用性的帕累托最优——在最小权限约束下最大化 Agent 的自主操作空间。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐