promptgate

    A capability-based security gate for AI agent tool-call interception, with session-scoped policies and default-deny enforcement

    agent
    security
    capability
    gate
    policy
    tool-call
    Download zip
    Author
    Version
    0.1.0
    License
    Apache-2.0
    Last updated
    11 hours ago
    Downloads
    7

    No dependencies

    #PromptGate

    借鉴操作系统 capability-based security 和 seccomp 思想的 LLM Agent 工具调用安全网关库。MoonBit 实现,可嵌入任何 Agent 框架。

    #核心思想

    OS 概念 → PromptGate 映射 capability(能力令牌) → 只有持有能力的 Agent 才能调用对应工具 seccomp(系统调用过滤) → 策略声明"允许哪些工具调用",其余默认拒绝 cgroup(资源配额) → token 消耗预算 + 调用次数限制 /proc(可观测性) → 每次判定的审计日志,可追溯可导出 进程地址空间 → Session 绑定能力,会话结束能力回收

    #与其他项目的差异

    项目定位关系
    MoonPermit计划编译最小权限 + 运行时检查互补:MoonPermit 管"这次任务需要什么权限",PromptGate 管"这个环境允许什么操作"
    MoonPod通用策略执行与审计引擎互补:MoonPod 覆盖文件/命令/网络通用策略,PromptGate 专注于 Agent 工具调用的能力边界

    PromptGate 不复制以上项目的能力,而是填补"Agent 工具调用前的静态安全门禁"这个细分空白。

    #三个完整使用场景

    #场景 1:AI 代码助手安全门禁

    IDE 中的 AI 助理被授予限制能力: allow: (read_file, read, "workspace/**") allow: (run_shell, execute, "moon*") deny: (read_file, read, "*.env") deny: (run_shell, execute, "rm*") 当 LLM 试图读取 .env 时 → Deny(审计记录:匹配 deny 规则) 当 LLM 运行 moon test 时 → Allow 当 LLM 运行 rm -rf 时 → Deny(审计记录:匹配 deny 规则)

    #场景 2:多租户 Agent API 网关

    租户 A:capabilities 绑定 db/tenant_a/* 租户 B:capabilities 绑定 db/tenant_b/* 每个租户的 Agent session 拥有独立的 Gate + Budget + AuditTrail。 租户 A 调用 db/tenant_b/data → Deny(资源不匹配任何 allow 规则,default-deny)

    #场景 3:CI/CD Agent 发布流水线

    allow: (api_call, get, "ci.example.com/status/*") allow: (api_call, post, "ci.example.com/deploy/*") ask: (api_call, post, "ci.example.com/deploy/production/*") deny: (api_call, delete, "ci.example.com/*") 部署到 staging → Allow(自动) 部署到 production → AskUser(需要人工确认) 删除服务 → Deny(审计记录中永久保留)

    #安装与使用

    # 克隆仓库 git clone https://github.com/zyyaa125/PromptGate.git cd PromptGate # 检查 moon check --deny-warn # 测试(48 个测试用例) moon test --deny-warn # 运行示例 moon run examples/code_assistant

    #最简单的 API 调用

    // 创建策略
    let json = "{\"version\":\"1\",\"rules\":[{\"rule_type\":\"allow\",\"tool\":\"read_file\",\"action\":\"read\",\"resource\":\"workspace/**\"}]}"
    let gate = try! @promptgate.load_policy(json)

    // 创建 session
    let budget = @promptgate.Budget::new(max_calls=Some(100L), max_tokens=None, window_seconds=None)
    let session = @promptgate.Session::new(id="agent-1", gate=gate, budget=budget, expires_at=None)

    // 工具调用判定
    let (session, verdict) = session.gate_call("read_file", "read", "workspace/main.mbt")
    // verdict == Verdict::Allow

    // 查看审计日志
    let entries = session.audit_entries()
    // entries[0].verdict == "allow"

    #可复现验收

    在仓库根目录运行:

    moon fmt --check # 格式检查 moon check --deny-warn # 类型检查(0 错误) moon test --deny-warn # 48 个测试全部通过 moon run examples/code_assistant # 输出 5 条判定结果 + 审计日志

    示例预期输出:

    PromptGate Code Assistant Demo ================================ call: read_file(workspace/src/main.mbt) -> Allow call: read_file(.env) -> Deny call: run_shell(moon test) -> Allow call: run_shell(rm -rf /) -> Deny call: api_call(api.github.com/gists) -> AskUser Audit Trail: ------------ #0: read_file workspace/src/main.mbt => allow (matched allow rule) #1: read_file .env => deny (default-deny or matched deny rule) #2: run_shell moon test => allow (matched allow rule) #3: run_shell rm -rf / => deny (default-deny or matched deny rule) #4: api_call api.github.com/gists => ask_user (matched ask-user rule) remaining budget: 95 calls

    #架构

    Policy (JSON) → CapabilitySet → Gate → Session { gate_call(tool, action, resource): 1. Budget.consume() → 超出配额则 Deny 2. Gate.judge() → Allow / Deny / AskUser 3. AuditTrail.record() → 记录决策 } 模块依赖关系: capability.mbt ← 数据模型 capability_set.mbt ← glob 匹配 + deny-first 判定 gate.mbt ← Allow/Deny/AskUser 三层判定 policy.mbt ← JSON 策略加载 + 验证 budget.mbt ← 调用次数/token 配额 audit.mbt ← 环形缓冲审计日志 session.mbt ← 整合以上全部模块

    #技术边界

    • 交付:可嵌入的 MoonBit 库,7 个公开模块,48 个测试,1 个可运行示例
    • 策略格式:JSON(任何语言都能生成)
    • 匹配粒度:(tool, action, resource pattern),支持 * 和 ** 通配符
    • 默认策略:default-deny(安全第一原则)
    • 不交付:真实沙箱隔离、OS 级 enforcement、分布式 session 管理、UI 界面

    #文档

    #许可证

    Apache License 2.0

    PolicyError

    pub(all) suberror PolicyError {
    InvalidJson(String)
    MissingField(String)
    UnknownVersion(String)
    EmptyRules
    }

    Error type for policy loading failures.

    AuditEntry

    pub struct AuditEntry {
    timestamp : Int64
    tool : String
    action : String
    resource : String
    verdict : String
    reason : String
    } derive(Eq,
    Debug
    )

    One recorded decision.

    AuditEntry::equal

    fn AuditEntry::equal(AuditEntry, AuditEntry) -> Bool

    AuditEntry::new

    fn AuditEntry::new(tool : String, action : String, resource : String, verdict : String, reason : String) -> AuditEntry

    Create a new AuditEntry.

    AuditEntry::not_equal

    fn AuditEntry::not_equal(x : AuditEntry, y : AuditEntry) -> Bool

    AuditTrail

    pub struct AuditTrail {
    entries : Array[AuditEntry]
    max_size : Int
    } derive(Eq,
    Debug
    )

    AuditTrail::all

    Return all recorded entries.

    AuditTrail::count

    fn AuditTrail::count(self : AuditTrail) -> Int

    Return the number of recorded entries.

    AuditTrail::equal

    fn AuditTrail::equal(AuditTrail, AuditTrail) -> Bool

    AuditTrail::new

    fn AuditTrail::new(max_size~ : Int) -> AuditTrail

    Create a new AuditTrail with the given maximum size (ring buffer).

    AuditTrail::not_equal

    fn AuditTrail::not_equal(x : AuditTrail, y : AuditTrail) -> Bool

    AuditTrail::record

    fn AuditTrail::record(self : AuditTrail, tool : String, action : String, resource : String, verdict : String, reason : String) -> AuditTrail

    Record a decision into the trail. Oldest entry is evicted when full.

    Budget

    pub struct Budget {
    max_calls : Int64?
    max_tokens : Int64?
    window_seconds : Int64?
    consumed_calls : Int64
    consumed_tokens : Int64
    window_start : Int64
    } derive(Eq,
    Debug
    )

    Budget::consume

    fn Budget::consume(self : Budget, calls : Int64, tokens : Int64) -> (Budget, Bool)

    Attempt to consume the given amount. Returns the updated Budget and whether allowed.

    Budget::equal

    fn Budget::equal(Budget, Budget) -> Bool

    Budget::new

    fn Budget::new(max_calls~ : Int64?, max_tokens~ : Int64?, window_seconds~ : Int64?) -> Budget

    Create a Budget with optional limits. Pass None for any limit to disable it.

    Budget::not_equal

    fn Budget::not_equal(x : Budget, y : Budget) -> Bool

    Budget::remaining_calls

    fn Budget::remaining_calls(self : Budget) -> Int64?

    Check how many calls remain (None means unlimited).

    Budget::remaining_tokens

    fn Budget::remaining_tokens(self : Budget) -> Int64?

    Check how many tokens remain (None means unlimited).

    Budget::reset

    fn Budget::reset(self : Budget) -> Budget

    Reset consumed counters to zero. Keeps the same limits.

    Budget::to_repr

    Capability

    pub struct Capability {
    tool : String
    action : String
    resource : String
    } derive(Eq,
    Debug
    )

    A single capability describing one permitted operation.

    Capability::equal

    fn Capability::equal(Capability, Capability) -> Bool

    Capability::not_equal

    fn Capability::not_equal(x : Capability, y : Capability) -> Bool

    Capability::to_string

    fn Capability::to_string(self : Capability) -> String

    Return a human-readable representation of this capability.

    CapabilitySet

    pub struct CapabilitySet {
    allows : Array[Capability]
    denys : Array[Capability]
    } derive(Eq,
    Debug
    )

    CapabilitySet::add_allow

    fn CapabilitySet::add_allow(self : CapabilitySet, cap : Capability) -> CapabilitySet

    Add an allow rule.

    CapabilitySet::add_deny

    fn CapabilitySet::add_deny(self : CapabilitySet, cap : Capability) -> CapabilitySet

    Add a deny rule.

    CapabilitySet::equal

    CapabilitySet::match_

    fn CapabilitySet::match_(self : CapabilitySet, tool : String, action : String, resource : String) -> MatchResult

    Match a concrete (tool, action, resource) against this set. Deny rules are checked first and take absolute priority.

    CapabilitySet::new

    Create an empty CapabilitySet.

    CapabilitySet::not_equal

    fn CapabilitySet::not_equal(x : CapabilitySet, y : CapabilitySet) -> Bool

    Gate

    pub struct Gate {
    capabilities : CapabilitySet
    ask_rules : Array[Capability]
    } derive(Eq,
    Debug
    )

    Gate::equal

    fn Gate::equal(Gate, Gate) -> Bool

    Gate::judge

    fn Gate::judge(self : Gate, tool : String, action : String, resource : String) -> Verdict

    Judge a tool call and return Allow, Deny, or AskUser.

    Evaluation order:
    1. Ask-user rules match → AskUser
    2. CapabilitySet gives Allow → Allow
    3. CapabilitySet gives Deny or nothing matched → Deny (default-deny)

    Gate::new

    fn Gate::new(capabilities : CapabilitySet, ask_rules : Array[Capability]) -> Gate

    Create a new Gate from a CapabilitySet and optional ask-user rules.

    Gate::not_equal

    fn Gate::not_equal(x : Gate, y : Gate) -> Bool

    Gate::to_repr

    MatchResult

    pub(all) enum MatchResult {
    Allow
    Deny
    } derive(Eq,
    Debug
    )

    Outcome of matching against a CapabilitySet.

    MatchResult::equal

    fn MatchResult::equal(MatchResult, MatchResult) -> Bool

    MatchResult::not_equal

    fn MatchResult::not_equal(x : MatchResult, y : MatchResult) -> Bool

    Session

    pub struct Session {
    id : String
    gate : Gate
    budget : Budget
    audit : AuditTrail
    created_at : Int64
    expires_at : Int64?
    } derive(Eq,
    Debug
    )

    Session::audit_entries

    fn Session::audit_entries(self : Session) -> Array[AuditEntry]

    Return the session's audit trail (read-only snapshot).

    Session::close

    fn Session::close(self : Session) -> Session

    Close the session (clear audit, zero budget). The session struct is consumed; callers should drop the returned value or use it for final audit export before discarding.

    Session::equal

    fn Session::equal(Session, Session) -> Bool

    Session::gate_call

    fn Session::gate_call(self : Session, tool : String, action : String, resource : String) -> (Session, Verdict)

    Gate a single tool call through all layers:
    1. Check budget — deny if exhausted.
    2. Check Gate rules — deny/allow/ask.
    3. Record decision in audit trail.

    Returns (new_session, verdict).

    Session::is_expired

    fn Session::is_expired(self : Session) -> Bool

    Check whether the session has expired.

    Session::new

    fn Session::new(id~ : String, gate~ : Gate, budget~ : Budget, expires_at~ : Int64?) -> Session

    Create a new Session with the given Gate, Budget, and optional expiry. expires_at is a Unix timestamp in milliseconds; None means no expiry.

    Session::not_equal

    fn Session::not_equal(x : Session, y : Session) -> Bool

    Session::remaining_budget

    fn Session::remaining_budget(self : Session) -> Int64?

    Return the session's remaining call budget.

    Session::to_repr

    Verdict

    pub(all) enum Verdict {
    Allow
    Deny
    AskUser
    } derive(Eq,
    Debug
    )

    Verdict for a single tool-call request.

    Verdict::equal

    fn Verdict::equal(Verdict, Verdict) -> Bool

    Verdict::not_equal

    fn Verdict::not_equal(x : Verdict, y : Verdict) -> Bool

    Verdict::to_repr

    cap_matches

    fn cap_matches(rule : Capability, tool : String, action : String, resource : String) -> Bool

    Check if a single Capability rule matches the given request.

    glob_match

    fn glob_match(pattern : String, value : String) -> Bool

    Glob-style pattern matching. * matches any non-/ character sequence within one segment. ** matches any sequence including /.

    load_policy

    fn load_policy(json_str : String) -> Gate raise PolicyError

    Load a Gate from a JSON policy string.

    new

    fn new(tool : String, action : String, resource : String) -> Capability

    Create a new Capability with the given tool, action, and resource pattern.