扩展 Open Policy Agent：自定义 Built-in 函数、插件与存储后端开发指南

扩展 Open Policy Agent自定义 Built-in 函数、插件与存储后端开发指南【免费下载链接】opaOpen Policy Agent (OPA) is an open source, general-purpose policy engine.项目地址: https://gitcode.com/gh_mirrors/op/opa导读OPAOpen Policy Agent是一个开源的通用策略引擎内置了大量开箱即用的 built-in 函数字符串处理、算术、JWT 校验、HTTP 请求等。但现实中的策略场景往往超出内置能力你可能需要把内部系统的数据暴露给 Rego 策略、实现自定义的决策日志通道或接入自己的存储。本文以官方文档 docs/docs/extensions.md 为主线结合仓库源码系统讲解三种扩展 OPA 的方式——自定义 built-in 函数、运行时插件Plugin与自定义存储后端并附带运行时版本注入的构建技巧。读完本文你将能够在嵌入式场景下用rego包注册自定义函数为 OPA 可执行文件全局注册函数通过实现Factory/Plugin接口接入决策日志等新行为以及替换默认的内存存储。一、扩展 OPA 的总体思路OPA 的扩展能力覆盖三个层面扩展层面适用场景主要 API自定义 built-in 函数把未内置的能力如读取私有数据源暴露给 Regorego/rego.go 中的rego.Function、rego.Function1~rego.Function4、rego.RegisterBuiltinN运行时插件定制决策日志、加新的查询 API、扩展服务端行为plugins/plugins.go 中的Factory、Plugin接口配合runtime.RegisterPlugin自定义存储后端用外部存储替换默认的内存存储storage.Store接口 v1/runtime.RegisterStorageBackend三个层面互不排斥插件运行在 OPA 进程内、可访问 Manager 提供的存储、编译器等全局组件自定义 built-in 函数既可以在嵌入式查询中局部注册也可以全局注册进运行时。二、在 Go 中自定义 Built-in 函数嵌入式场景如果你把 OPA 作为库嵌入自己的程序并通过github.com/open-policy-agent/opa/rego包执行策略可以用自定义 built-in 函数为策略补充能力。注册一个函数需要两部分声明Declaration描述函数的类型签名实现Implementation提供求值期回调。2.1 最小示例首先导入三个包import github.com/open-policy-agent/opa/ast import github.com/open-policy-agent/opa/types import github.com/open-policy-agent/opa/regoast包提供 Rego 的抽象语法树类型函数接收/返回的*ast.Termtypes包提供构建函数类型签名所需的类型构造器rego包则是嵌入式求值的入口。下面是一个简单的hello函数r : rego.New( rego.Query(x hello(bob)), rego.Function1( rego.Function{ Name: hello, Decl: types.NewFunction(types.Args(types.S), types.S), }, func(_ rego.BuiltinContext, a *ast.Term) (*ast.Term, error) { if str, ok : a.Value.(ast.String); ok { return ast.StringTerm(hello, string(str)), nil } return nil, nil }), ) query, err : r.PrepareForEval(ctx) if err ! nil { // handle error. }准备完成后即可执行查询rs, err : query.Eval(ctx) if err ! nil { // handle error. } // Do something with result. fmt.Println(rs[0].Bindings[x])输出结果为hello, bob从源码看rego.Function1返回的是一个func(*Rego)选项它把声明与实现绑定到当前 Rego 对象rego/rego.go。这个例子揭示了几个关键点操作数数量rego包提供了rego.Function1、rego.Function2、rego.Function3、rego.Function4以及接受参数列表的rego.FunctionDyn等变体见 rego/rego.gorego.Function#Name指定策略查询中可以引用的操作符名rego.Function#Decl指定函数类型签名上例接受一个字符串并返回一个字符串未定义语义当第一个返回值是nil时函数整体视为 undefined——这与 Rego 的未定义即安全失败哲学一致。2.2 函数声明结构详解rego.Function结构体定义在 v1/rego/rego.gorego包是v1包的兼容别名转发层type Function struct { Name string Description string Decl *types.Function Memoize bool Nondeterministic bool }字段含义如下字段作用Name函数在 Rego 中的调用名可含.用于命名空间隔离避免与内置函数或其他自定义函数冲突Description函数说明用于生成文档/元数据Decltypes.NewFunction(types.Args(...), 返回值)构造的类型签名Memoize同一查询内多次调用时是否记忆化结果Nondeterministic标记函数结果可能不确定如依赖网络运行时会据此阻止其在 bundle 构建或部分求值等场景中被意外调用types.Sstring和types.Aany是构造 Rego 类型的便捷别名如果需要用例特定类型例如字段为foo、bar、baz的对象列表需要使用types包的完整 API如types.NewObject、types.NewArray构造参考 types/types.go。2.3 复杂示例实现github.repo函数假如你想把 GitHub 仓库元数据暴露给策略可以自定义 built-in 函数在求值时按需拉取数据r : rego.New( rego.Query(github.repo(open-policy-agent, opa)), rego.Function2( rego.Function{ Name: github.repo, Decl: types.NewFunction(types.Args(types.S, types.S), types.A), Memoize: true, Nondeterministic: true, }, func(bctx rego.BuiltinContext, a, b *ast.Term) (*ast.Term, error) { // see implementation below. }, ), )这里Decl表明函数接收两个字符串、返回anyRego 中所有类型的并集。注意Name中包含.字符这是允许的也推荐用来为自定义函数做命名空间。声明中两个标志位很重要Memoize: true该函数执行 I/O开启记忆化后同一查询内多次调用只执行一次保证求值确定性Nondeterministic: true函数结果依赖网络条件、可能不确定标记后 OPA 会提供基本的安全信息避免它在 bundle 构建或部分求值partial evaluation时被误执行。实现部分包装 Go 标准库发起对 GitHub API 的 HTTP 请求func(bctx rego.BuiltinContext, a, b *ast.Term) (*ast.Term, error) { var org, repo string if err : ast.As(a.Value, org); err ! nil { return nil, err } else if err : ast.As(b.Value, repo); err ! nil { return nil, err } req, err : http.NewRequest(GET, fmt.Sprintf(https://api.github.com/repos/%v/%v, org, repo), nil) if err ! nil { return nil, err } resp, err : http.DefaultClient.Do(req.WithContext(bctx.Context)) if err ! nil { return nil, err } defer resp.Body.Close() if resp.StatusCode ! http.StatusOK { return nil, fmt.Errorf(resp.Status) } v, err : ast.ValueFromReader(resp.Body) if err ! nil { return nil, err } return ast.NewTerm(v), nil }实现要点用ast.As把*ast.Term解包为 Go 原生类型如string务必使用bctx.Context——rego.BuiltinContext携带求值期的context.Context、缓存等求值器属性见 v1/rego/rego.goHTTP 请求绑定该上下文可随查询取消而中止用ast.ValueFromReader把响应体解析为 Rego 值再经ast.NewTerm包装返回。完整的可运行示例含main()、r.Eval、JSON 输出见本文附录 A。2.4 重要安全边界:::danger 自定义 built-in 函数不得用于对外部系统产生副作用如写库、发消息。由于策略求值阶段会自动应用性能优化如规则索引、短路求值、部分求值等OPA不保证语句一定会被执行。若确需副作用应放在求值完成后的应用层如把决策结果传给下游系统而不是放进 built-in。 :::三、把自定义 Built-in 函数注册进 OPA 运行时如果你不想嵌入 OPA而是希望定制 OPA 可执行文件本身可以在main函数中用rego.RegisterBuiltin2及RegisterBuiltin1/RegisterBuiltin3/RegisterBuiltin4/RegisterBuiltinDyn见 rego/rego.go全局注册函数然后启动 CLIfunc main() { rego.RegisterBuiltin2( rego.Function{ Name: github.repo, Decl: types.NewFunction(types.Args(types.S, types.S), types.A), Memoize: true, Nondeterministic: true, }, func(bctx rego.BuiltinContext, a, b *ast.Term) (*ast.Term, error) { // 实现与上文 github.repo 相同 }, ) if err : cmd.RootCommand.Execute(); err ! nil { fmt.Println(err) os.Exit(1) } }从源码看RegisterBuiltin2内部做了两件事v1/rego/rego.go先调用ast.RegisterBuiltin把声明注册到 AST 层再用topdown.RegisterBuiltinFunc注册求值期实现同时通过memoize封装处理Memoize标志、通过finishFunction统一处理 undefined/错误迭代语义。这意味着全局注册的函数对所有通过该二进制执行的查询含run、eval、test等子命令都可用。完整示例见附录 B。四、为 OPA 运行时编写自定义插件如果要定制行为而非函数——例如实现新的决策日志通道、新的查询 API——需要实现 OPA 的插件接口。插件体系由 plugins/plugins.go 定义plugins包本身已标记为 v0.x 兼容包新项目建议使用对应的 v1/plugins/plugins.go 组件两者 API 一致核心是两个接口Factory负责实例化插件。OPA 处理配置时会查找通过runtime.RegisterPlugin注册到某名字的工厂工厂只有在配置里出现对应名字的配置块时才会被调用见 plugins/plugins.goPlugin提供插件行为。OPA 启动时会调用Start启动所有已配置的插件通过 discovery 收到新配置时会先经工厂Validate校验再调用Reconfigure见 plugins/plugins.go。在main函数中通过runtime.RegisterPlugin注册工厂runtime.RegisterPlugin(PluginName, Factory{})4.1 插件状态上报插件可以通过plugins.Manager#UpdatePluginStatusAPI 可选地向 Manager 上报自身状态Status结构体包含State和可选Message见 v1/plugins/plugins.go。如果不上报插件默认被视为工作正常。预定义状态在 plugins/plugins.go 中声明状态含义StateNotReady未进入错误状态但尚未就绪通常只在初始化时出现StateOK正常运行StateErr处于错误状态不应视为可用StateWarn运行中但处于潜在危险或降级状态可能提示需要人工干预惯例是创建时上报StatusNotReadyStart成功后更新为StatusOK出错则StatusErr停止时再回到StatusNotReady。4.2 实战实现一个把决策日志写到 stdout/stderr 的插件下面实现一个自定义Decision Logger把决策事件流式输出到 stdout 或 stderr事件字段与 docs/docs/management-decision-logs.md 中描述的 1:1 对应。第一步定义插件本体实现Start、Stop、Reconfigure与日志回调import ( encoding/json github.com/open-policy-agent/opa/plugins/logs ) const PluginName println_decision_logger type Config struct { Stderr bool json:stderr // false stdout, true stderr } type PrintlnLogger struct { manager *plugins.Manager mtx sync.Mutex config Config } func (p *PrintlnLogger) Start(ctx context.Context) error { p.manager.UpdatePluginStatus(PluginName, plugins.Status{State: plugins.StateOK}) return nil } func (p *PrintlnLogger) Stop(ctx context.Context) { p.manager.UpdatePluginStatus(PluginName, plugins.Status{State: plugins.StateNotReady}) } func (p *PrintlnLogger) Reconfigure(ctx context.Context, config any) { p.mtx.Lock() defer p.mtx.Unlock() p.config config.(Config) } // Log is called by the decision logger when a record (event) should be emitted. func (p *PrintlnLogger) Log(ctx context.Context, event logs.EventV1) error { p.mtx.Lock() defer p.mtx.Unlock() w : os.Stdout if p.config.Stderr { w os.Stderr } bs, err : json.Marshal(event) if err ! nil { p.manager.UpdatePluginStatus(PluginName, plugins.Status{State: plugins.StateErr}) return nil } _, err fmt.Fprintln(w, string(bs)) if err ! nil { p.manager.UpdatePluginStatus(PluginName, plugins.Status{State: plugins.StateErr}) } return nil }第二步实现工厂。Validate接收插件配置的原始字节反序列化并做语义校验后返回配置值New接收校验过的配置并实例化插件、上报初始状态import ( github.com/open-policy-agent/opa/plugins github.com/open-policy-agent/opa/util ) type Factory struct{} func (Factory) New(m *plugins.Manager, config any) plugins.Plugin { m.UpdatePluginStatus(PluginName, plugins.Status{State: plugins.StateNotReady}) return PrintlnLogger{ manager: m, config: config.(Config), } } func (Factory) Validate(_ *plugins.Manager, config []byte) (any, error) { parsedConfig : Config{} return parsedConfig, util.Unmarshal(config, parsedConfig) }第三步注册工厂并启动 OPA。cmd.RootCommand.Execute会启动 OPA 且不返回import ( github.com/open-policy-agent/opa/cmd github.com/open-policy-agent/opa/runtime ) func main() { runtime.RegisterPlugin(PluginName, Factory{}) if err : cmd.RootCommand.Execute(); err ! nil { fmt.Println(err) os.Exit(1) } }第四步构建包含插件的 OPA 可执行文件go build -o opa第五步编写配置并启用插件opa-config.yamldecision_logs: plugin: println_decision_logger plugins: println_decision_logger: stderr: false第六步启动并验证./opa run --server --config-file opa-config.yaml然后在另一个终端通过 OPA API 触发一次决策curl localhost:8181/v1/data一切正常的话你会看到决策日志事件的 Go 结构体表示被 JSON 序列化后写到 stdout。注意事项如果配置了 mask 策略详见 docs/docs/management-decision-logs.md插件收到的Event可能与上面示例文档化的字段有差异插件命名空间应与配置键保持一致若某个注册过的插件未出现在配置中其工厂不会被调用见 plugins/plugins.goOPA 目前不会在停止时主动调用Stop见 plugins/plugins.go状态管理应以此为设计前提。五、自定义存储后端OPA 默认使用内存存储可通过实现storage.Store接口替换为自定义实现如接入外部数据库。要点实现 storage/storage.go 中的storage.Store接口在 OPA 运行时初始化前调用v1/runtime.RegisterStorageBackend注册后端见 v1/runtime/runtime.go如果存储需要资源清理关闭连接、刷写缓冲等实现storage.Closer接口其Close()方法会在优雅停机时被调用。注册示例package main import ( github.com/open-policy-agent/opa/cmd github.com/open-policy-agent/opa/v1/runtime ) func init() { runtime.RegisterStorageBackend(func( ctx context.Context, logger logging.Logger, registerer prometheus.Registerer, config []byte, id string, ) (storage.Store, error) { return myCustomStore, nil }) } func main() { if err : cmd.RootCommand.Execute(); err ! nil { os.Exit(1) } }注意注册回调会收到prometheus.Registerer用于指标注册与id实例标识方便与运行时监控体系集成。六、设置 OPA 运行时版本信息OPA 运行时版本在构建期静态注入。github.com/open-policy-agent/opa/version包导出了四个全局变量见 v1/version/version.goversion包同样转发到v1/version变量名说明VersionOPA 运行时的可读/语义化版本号VcsOPA 运行时构建所基于的 Git SHATimestampOPA 运行时的构建日期/时间Hostname构建 OPA 运行时所在系统的主机名从源码看Vcs与Timestamp在init()中会从 Go 构建信息debug.ReadBuildInfo的vcs.revision、vcs.time、vcs.modified设置自动回填vcs.modified为 true 时会在Vcs后追加-dirty后缀v1/version/version.go。自定义这些值只需在构建时通过-ldflags -X注入go build \ -ldflags \ -X github.com/open-policy-agent/opa/v1/version.VersionMY_VERSION\ -X github.com/open-policy-agent/opa/v1/version.VcsMY_COMMIT_HASH \ -X github.com/open-policy-agent/opa/v1/version.HostnameMY_HOSTNAME \ -X github.com/open-policy-agent/opa/v1/version.TimestampMY_TIMESTAMP \ -o opa这样产出的二进制在opa version输出中会显示自定义的版本标识便于在分布式环境中识别构建来源。附录 A自定义 built-in 函数完整示例嵌入式package main import ( context encoding/json fmt log net/http github.com/open-policy-agent/opa/ast github.com/open-policy-agent/opa/rego github.com/open-policy-agent/opa/types ) func main() { r : rego.New( rego.Query(github.repo(open-policy-agent, opa)), rego.Function2( rego.Function{ Name: github.repo, Decl: types.NewFunction(types.Args(types.S, types.S), types.A), Memoize: true, Nondeterministic: true, }, func(bctx rego.BuiltinContext, a, b *ast.Term) (*ast.Term, error) { var org, repo string if err : ast.As(a.Value, org); err ! nil { return nil, err } else if err : ast.As(b.Value, repo); err ! nil { return nil, err } req, err : http.NewRequest(GET, fmt.Sprintf(https://api.github.com/repos/%v/%v, org, repo), nil) if err ! nil { return nil, err } resp, err : http.DefaultClient.Do(req.WithContext(bctx.Context)) if err ! nil { return nil, err } defer resp.Body.Close() if resp.StatusCode ! http.StatusOK { return nil, fmt.Errorf(resp.Status) } v, err : ast.ValueFromReader(resp.Body) if err ! nil { return nil, err } return ast.NewTerm(v), nil }, ), ) rs, err : r.Eval(context.Background()) if err ! nil { log.Fatal(err) } else if len(rs) 0 { fmt.Println(undefined) } else { bs, _ : json.MarshalIndent(rs[0].Expressions[0].Value, , ) fmt.Println(string(bs)) } }附录 B向 OPA 运行时添加 built-in 函数完整示例package main import ( fmt net/http os github.com/open-policy-agent/opa/ast github.com/open-policy-agent/opa/cmd github.com/open-policy-agent/opa/rego github.com/open-policy-agent/opa/types ) func main() { rego.RegisterBuiltin2( rego.Function{ Name: github.repo, Decl: types.NewFunction(types.Args(types.S, types.S), types.A), Memoize: true, Nondeterministic: true, }, func(bctx rego.BuiltinContext, a, b *ast.Term) (*ast.Term, error) { var org, repo string if err : ast.As(a.Value, org); err ! nil { return nil, err } else if err : ast.As(b.Value, repo); err ! nil { return nil, err } req, err : http.NewRequest(GET, fmt.Sprintf(https://api.github.com/repos/%v/%v, org, repo), nil) if err ! nil { return nil, err } resp, err : http.DefaultClient.Do(req.WithContext(bctx.Context)) if err ! nil { return nil, err } defer resp.Body.Close() if resp.StatusCode ! http.StatusOK { return nil, fmt.Errorf(resp.Status) } v, err : ast.ValueFromReader(resp.Body) if err ! nil { return nil, err } return ast.NewTerm(v), nil }, ) if err : cmd.RootCommand.Execute(); err ! nil { fmt.Println(err) os.Exit(1) } }延伸阅读docs/docs/management-decision-logs.md决策日志插件所对接的事件模型与 mask 策略plugins/plugins.goFactory/Plugin接口与插件生命周期、状态常量的权威定义rego/rego.go嵌入式rego.New的全部选项Query、Function1~Function4、RegisterBuiltinN等v1/rego/rego.gorego.Function结构体与全局注册的内部实现ast.RegisterBuiltintopdown.RegisterBuiltinFunc 记忆化封装v1/version/version.go构建期版本变量的定义与自动回填逻辑types/types.gotypes.NewFunction、types.Args、types.S/types.A等类型构造 API【免费下载链接】opaOpen Policy Agent (OPA) is an open source, general-purpose policy engine.项目地址: https://gitcode.com/gh_mirrors/op/opa创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考