feat(api): 完善动态 Action 与统一过滤调用引擎(by AI)
This commit is contained in:
parent
2370bfa79d
commit
d10e722665
10
CHANGELOG.md
10
CHANGELOG.md
@ -1,5 +1,15 @@
|
||||
# CHANGELOG
|
||||
|
||||
## v1.5.6 (2026-08-17)
|
||||
- **统一调用结果**: `Call` 与 `CallBy` 返回 `ok/statusCode/headers/data/code/error`,支持响应错误字段和成功码规则。
|
||||
- **调用选项**: 新增 Token 验证、超时、配置覆盖与敏感/控制字段覆盖保护。
|
||||
- **流式能力**: 基于 `go/http.ManualDo` 增加同步 Headers/Data/Done 回调,支持 JS 低代码向下游响应透传。
|
||||
- **动态管理**: 增加 Action 与 JavaScript Signer 的注册、替换和移除能力。
|
||||
- **统一 Filter 管线**: Action 仅配置有序 `filters: []`;同一 Filter 实例按 `request/headers/chunk/result/done` 事件处理调用全生命周期,并提供逐调用隔离状态。
|
||||
- **安全配置**: 密文格式收敛为 `**URLBase64(AESGCM)`,新增无状态 `Encrypt`。
|
||||
- **结构化审计**: 增加 API 调用日志;正文默认关闭,开启后仍执行文本、数组和深度裁剪。
|
||||
- **依赖对齐**: 升级 `cast/config/crypto/encoding/http/log/safe` 到当前稳定补丁版本。
|
||||
|
||||
## v1.5.5 (2026-06-21)
|
||||
- **错误堆栈重构**:
|
||||
- `Call` 函数的返回错误使用 `jsmod.MakeError` 动态包裹以保留调用堆栈。
|
||||
|
||||
36
README.md
36
README.md
@ -19,7 +19,7 @@ go get apigo.cc/go/api
|
||||
|
||||
1. **定义 Action**:实现 `Action` 接口(及可选的 `SignerAction`, `ConfigurableAction` 等)。
|
||||
2. **配置授权**:在 `api.yml` 或环境变量中配置密钥。
|
||||
3. **发起调用**:使用 `api.Call[Response](&action)`。
|
||||
3. **发起调用**:使用 `api.Call(action, options)`,统一获得 `Result`。
|
||||
|
||||
## 🛠 接口说明
|
||||
|
||||
@ -28,10 +28,15 @@ go get apigo.cc/go/api
|
||||
* `ConfigurableAction`:提供硬编码的默认参数或元数据。
|
||||
* `URLAction` / `MethodAction`:动态指定 Endpoint 和 HTTP 方法。
|
||||
* `ValidatableAction`:业务参数自校验。
|
||||
* `CallOptions`:提供内部 Token、超时、临时配置覆盖、Logger 与流式回调。
|
||||
* `Result`:统一返回 `ok/statusCode/headers/data/code/error`,业务响应保留在 `data`。
|
||||
* `RegisterAction` / `RemoveAction`:动态注册和热更新数据驱动的 Action。
|
||||
* `RegisterJSSigner` / `RemoveSigner`:管理 JavaScript Signer。
|
||||
* `RegisterJSFilter` / `RemoveFilter`:管理可复用的 JavaScript 请求/响应 Filter。
|
||||
|
||||
## 🔒 安全性 (Ultimate Memory Safety)
|
||||
|
||||
* **内置解密**:支持自动识别并解密配置中的 AES 加密内容。
|
||||
* **内置解密**:只识别 `**URLBase64(AESGCM)` 配置密文。
|
||||
* **内存保护**:敏感配置解密后以 `safe.SafeBuf` 形式存储,防止内存 Dump 泄露。
|
||||
- **防止字符串泄露**:通过 `unsafe.String` 零拷贝技术,确保敏感 Header(如 Authorization)在调用结束后可被物理擦除,彻底解决 Go 字符串不可变性导致的堆泄露问题。
|
||||
- **全生命周期闭环**:`api.Call` 结束后自动触发 `httpReq.Close()`,对所有中间缓冲区进行 `ZeroMemory` 随机覆盖。
|
||||
@ -47,8 +52,33 @@ type MySmsAction struct {
|
||||
func (MySmsAction) ActionName() string { return "tencent.sms.smsPackagesStatistics" }
|
||||
func (MySmsAction) SignerName() string { return "tc3" }
|
||||
|
||||
resp, err := api.Call[MyResponse](&MySmsAction{Limit: 10})
|
||||
result, err := api.Call(&MySmsAction{Limit: 10}, &api.CallOptions{
|
||||
Token: "internal-token",
|
||||
Config: map[string]any{"path": "/v1/messages"},
|
||||
})
|
||||
```
|
||||
|
||||
动态 Action 可以在调用时覆盖非敏感配置:
|
||||
|
||||
```go
|
||||
api.RegisterAction("openai", map[string]any{
|
||||
"url": "http://127.0.0.1:8001/v1/chat/completions",
|
||||
"method": "POST",
|
||||
"signer": "none",
|
||||
"tokens": []string{"internal-token"},
|
||||
})
|
||||
|
||||
result, err := api.CallBy("openai", payload, &api.CallOptions{
|
||||
Token: "internal-token",
|
||||
Config: map[string]any{"path": "/v1/embeddings"},
|
||||
})
|
||||
```
|
||||
|
||||
`tokens` 未配置或为空时无需 Token。`tokens/enabled/test/logging` 与原配置中的密文字段不能通过 `options.config` 覆盖。流式调用使用 `go/http.ManualDo`,`OnHeaders` 与 `OnDone` 接收 lower-camel map,`OnData` 接收原始字节块。
|
||||
|
||||
Action 只使用 `filters: ["name"]` 这一种形式。每个 Filter 都会收到前置 `request` 以及后置 `headers/chunk/result/done` 事件,并自行根据 `event` 决定是否处理。流式 Filter 应通过输入/输出的 `state` 保存单次调用状态,不能假设 HTTP chunk 与 SSE 或 JSONL 消息边界一致。JavaScript Filter 的 `chunk` 是 UTF-8 字符串;Go Filter 仍接收原始 `[]byte`,二进制响应不应挂载文本型 JavaScript Filter。
|
||||
|
||||
动态 Action 可通过 `extends` 继承另一个 Action;父配置先合并,子配置深度覆盖。继承在每次调用时解析,因此父 Action 热更新会立即作用于子 Action。循环继承或不存在的父 Action 会直接返回错误。
|
||||
|
||||
---
|
||||
更多详情请参阅 [TEST.md](./TEST.md) 和 [CHANGELOG.md](./CHANGELOG.md)。
|
||||
|
||||
16
TEST.md
16
TEST.md
@ -25,11 +25,21 @@
|
||||
* **注入保护 (Guard)**:敏感数据禁止自动注入 Action 的 `string` 字段,杜绝无意中的内存留存。
|
||||
* **自动生命周期回收**:`api.Call` 结束时通过 `defer` 机制强制调用 `Close()` 擦除本次请求的所有明文密钥副本及 Header 缓冲区。
|
||||
|
||||
### 5. 动态策略与统一结果 (`TestTokenAndOverridePolicy`, `TestResultRulesAndConfigOverride`)
|
||||
验证 Action 内部 Token、控制字段黑名单、密文字段不可覆盖、URL/Method 临时覆盖,以及 `codeFields/successCodes/errorFields` 对统一结果的判断。
|
||||
|
||||
### 6. 流式调用 (`TestStreamPreservesChunkOrder`)
|
||||
使用 `ManualDo` 验证流式 Body 的分块顺序、同步回调、完成事件和统一结果状态。
|
||||
|
||||
### 7. 动态 Action 继承与统一 Filter 管线
|
||||
|
||||
验证 `extends` 深度继承、父 Action 热解析、循环和缺失父项报错,以及 `filters: []` 同时收到请求与响应事件。JavaScript Filter 的中文流片段按 UTF-8 字符串传递。
|
||||
|
||||
## ⏱ 性能基准测试 (Benchmark)
|
||||
|
||||
使用 `go test -bench=. ./...` 评估框架调用阶段的开销。
|
||||
> **基准**: Intel Core i9-9980HK 2.40GHz
|
||||
* `BenchmarkCallEngineLogic-16`:约 **245 ns/op**, **2 allocs/op**。
|
||||
> **基准**: Darwin / Apple M3 Max
|
||||
* `BenchmarkCallEngineLogic-16`:约 **118.9 ns/op**, **2 allocs/op**。
|
||||
该指标证明引擎的参数合并、注入及校验流程具有极高的运行效率和极小的内存逃逸。
|
||||
|
||||
## 🚀 运行测试
|
||||
@ -43,5 +53,5 @@ go test -bench=. ./...
|
||||
```
|
||||
|
||||
---
|
||||
最后测试日期:2026-06-21
|
||||
最后测试日期:2026-08-17
|
||||
状态:PASS
|
||||
|
||||
65
action.go
65
action.go
@ -1,7 +1,9 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"sync"
|
||||
"unsafe"
|
||||
|
||||
"apigo.cc/go/cast"
|
||||
@ -45,6 +47,7 @@ type FormatAction interface {
|
||||
|
||||
// ActionRegistry 存储已注册的 Action 模板或类型
|
||||
var actionRegistry = make(map[string]any)
|
||||
var actionRegistryMutex sync.RWMutex
|
||||
|
||||
// RegisterAction 注册 API 动作。
|
||||
// definition 可以是一个结构体实例(用作类型模板)或一个 map[string]any(用作动态定义)。
|
||||
@ -59,52 +62,57 @@ func RegisterAction(name string, definition any) {
|
||||
}
|
||||
|
||||
if t.Kind() == reflect.Struct {
|
||||
actionRegistryMutex.Lock()
|
||||
actionRegistry[name] = t
|
||||
actionRegistryMutex.Unlock()
|
||||
} else if m, ok := definition.(map[string]any); ok {
|
||||
config := cloneMap(m)
|
||||
decryptMapWithPrefix(config)
|
||||
ga := &GenericAction{
|
||||
name: name,
|
||||
url: cast.String(m["url"]),
|
||||
method: cast.String(m["method"]),
|
||||
signer: cast.String(m["signer"]),
|
||||
format: cast.String(m["format"]),
|
||||
payload: make(map[string]any),
|
||||
name: name, config: config, payload: make(map[string]any),
|
||||
}
|
||||
if p, ok := m["payload"].(map[string]any); ok {
|
||||
ga.payload = p
|
||||
}
|
||||
actionRegistryMutex.Lock()
|
||||
if previous, ok := actionRegistry[name].(*GenericAction); ok {
|
||||
closeSecrets(previous.config)
|
||||
}
|
||||
actionRegistry[name] = ga
|
||||
actionRegistryMutex.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
// RemoveAction unregisters a dynamic Action.
|
||||
func RemoveAction(name string) {
|
||||
actionRegistryMutex.Lock()
|
||||
if previous, ok := actionRegistry[name].(*GenericAction); ok {
|
||||
closeSecrets(previous.config)
|
||||
}
|
||||
delete(actionRegistry, name)
|
||||
actionRegistryMutex.Unlock()
|
||||
}
|
||||
|
||||
// GenericAction 是一个动态 Action 容器,实现了所有 API 相关接口
|
||||
type GenericAction struct {
|
||||
name string
|
||||
url string
|
||||
method string
|
||||
signer string
|
||||
format string
|
||||
config map[string]any
|
||||
payload map[string]any
|
||||
}
|
||||
|
||||
func (a *GenericAction) ActionName() string { return a.name }
|
||||
func (a *GenericAction) SignerName() string { return a.signer }
|
||||
func (a *GenericAction) GetURL() string { return a.url }
|
||||
func (a *GenericAction) GetMethod() string { return a.method }
|
||||
func (a *GenericAction) GetFormat() string { return a.format }
|
||||
func (a *GenericAction) Config() map[string]any {
|
||||
return map[string]any{
|
||||
"url": a.url,
|
||||
"method": a.method,
|
||||
"signer": a.signer,
|
||||
"format": a.format,
|
||||
}
|
||||
}
|
||||
func (a *GenericAction) ActionName() string { return a.name }
|
||||
func (a *GenericAction) SignerName() string { return cast.String(a.config["signer"]) }
|
||||
func (a *GenericAction) GetURL() string { return cast.String(a.config["url"]) }
|
||||
func (a *GenericAction) GetMethod() string { return cast.String(a.config["method"]) }
|
||||
func (a *GenericAction) GetFormat() string { return cast.String(a.config["format"]) }
|
||||
func (a *GenericAction) Config() map[string]any { return cloneMap(a.config) }
|
||||
func (a *GenericAction) MarshalJSON() ([]byte, error) {
|
||||
return cast.ToJSONBytes(a.payload)
|
||||
}
|
||||
|
||||
// HttpRequest 内部使用的请求描述结构,供 Signer 使用
|
||||
type HttpRequest struct {
|
||||
Context context.Context
|
||||
Url string
|
||||
Method string
|
||||
headers map[string]string
|
||||
@ -210,8 +218,11 @@ func (r *HttpRequest) SetHeader(key string, values ...any) {
|
||||
|
||||
// Result 定义 API 调用的标准返回结果
|
||||
type Result struct {
|
||||
StatusCode int
|
||||
Status string
|
||||
Headers map[string]string
|
||||
Data any
|
||||
Ok bool
|
||||
StatusCode int
|
||||
Headers map[string]string
|
||||
Data any
|
||||
Code string
|
||||
Error string
|
||||
receivedBytes int64
|
||||
}
|
||||
|
||||
11
api.go
11
api.go
@ -11,7 +11,7 @@ import (
|
||||
)
|
||||
|
||||
// Call 是调度引擎的入口
|
||||
func Call[T any](action Action) (*T, error) {
|
||||
func legacyCall[T any](action Action) (*T, error) {
|
||||
// 1. 获取并合并配置
|
||||
actionConfig, safeBufs := GetActionConfig(action.ActionName())
|
||||
defer func() {
|
||||
@ -122,7 +122,7 @@ func Call[T any](action Action) (*T, error) {
|
||||
}
|
||||
|
||||
// CallBy 通过动作名称和动态 Payload 发起调用
|
||||
func CallBy[T any](name string, payload any) (*T, error) {
|
||||
func legacyCallBy[T any](name string, payload any) (*T, error) {
|
||||
tmpl, ok := actionRegistry[name]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("action not found: %s", name)
|
||||
@ -140,10 +140,7 @@ func CallBy[T any](name string, payload any) (*T, error) {
|
||||
// JSON 定义模式 (深拷贝模板以防并发冲突)
|
||||
newGA := &GenericAction{
|
||||
name: ga.name,
|
||||
url: ga.url,
|
||||
method: ga.method,
|
||||
signer: ga.signer,
|
||||
format: ga.format,
|
||||
config: cloneMap(ga.config),
|
||||
payload: make(map[string]any),
|
||||
}
|
||||
// 复制模板 Payload
|
||||
@ -163,7 +160,7 @@ func CallBy[T any](name string, payload any) (*T, error) {
|
||||
action = newGA
|
||||
}
|
||||
|
||||
return Call[T](action)
|
||||
return legacyCall[T](action)
|
||||
}
|
||||
|
||||
func preprocessSecrets(m map[string]any, opened *[]*safe.SecretPlaintext) {
|
||||
|
||||
@ -9,6 +9,7 @@ import (
|
||||
"testing"
|
||||
|
||||
"apigo.cc/go/api"
|
||||
"apigo.cc/go/cast"
|
||||
"apigo.cc/go/encoding"
|
||||
)
|
||||
|
||||
@ -189,7 +190,7 @@ func TestFullCallFlow(t *testing.T) {
|
||||
}
|
||||
|
||||
// 执行调度调用
|
||||
resp, err := api.Call[MockResponse](action)
|
||||
resp, err := api.Call(action)
|
||||
if err != nil {
|
||||
t.Fatalf("Call failed: %v", err)
|
||||
}
|
||||
@ -206,7 +207,9 @@ func TestFullCallFlow(t *testing.T) {
|
||||
}
|
||||
|
||||
// 2. 验证响应绑定
|
||||
if resp.Status != "success" || resp.Echo != "Hello from mock server" {
|
||||
var response MockResponse
|
||||
cast.Convert(&response, resp.Data)
|
||||
if response.Status != "success" || response.Echo != "Hello from mock server" {
|
||||
t.Errorf("Response bind failed, got %+v", resp)
|
||||
}
|
||||
|
||||
|
||||
98
config.go
98
config.go
@ -1,6 +1,7 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync"
|
||||
@ -48,6 +49,31 @@ func SetConfig(name string, conf map[string]any) {
|
||||
}
|
||||
}
|
||||
|
||||
// ReplaceConfig replaces one configuration root. It is intended for dynamic
|
||||
// stores such as DocDB where removed fields must disappear immediately.
|
||||
func ReplaceConfig(name string, conf map[string]any) {
|
||||
configMutex.Lock()
|
||||
defer configMutex.Unlock()
|
||||
if old, ok := GlobalConfigs[name].(map[string]any); ok {
|
||||
closeSecrets(old)
|
||||
}
|
||||
copyConf := cloneMap(conf)
|
||||
decryptMapWithPrefix(copyConf)
|
||||
GlobalConfigs[name] = copyConf
|
||||
}
|
||||
|
||||
// Encrypt encrypts plaintext for storage in API configuration.
|
||||
func Encrypt(plaintext string) (string, error) {
|
||||
if confAES == nil {
|
||||
return "", errors.New("default AES is not initialized")
|
||||
}
|
||||
ciphertext, err := confAES.EncryptBytes([]byte(plaintext))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return "**" + encoding.URLBase64(ciphertext), nil
|
||||
}
|
||||
|
||||
// AddConfig 是 SetConfig 的兼容性别名
|
||||
func AddConfig(name string, conf map[string]any) { SetConfig(name, conf) }
|
||||
|
||||
@ -99,9 +125,7 @@ func GetActionConfig(actionName string) (map[string]any, []*safe.SafeBuf) {
|
||||
}
|
||||
}
|
||||
|
||||
// 这里的解密主要是为了处理文件中加载的 ENC() 格式 (兼容旧版)
|
||||
safeBufs := decryptMap(res)
|
||||
return res, safeBufs
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// fill 注入配置到 Action,非破坏性,仅注入零值
|
||||
@ -211,17 +235,40 @@ func MergeMap(dst, src map[string]any) {
|
||||
}
|
||||
}
|
||||
|
||||
func cloneMap(src map[string]any) map[string]any {
|
||||
dst := make(map[string]any, len(src))
|
||||
for k, v := range src {
|
||||
switch t := v.(type) {
|
||||
case map[string]any:
|
||||
dst[k] = cloneMap(t)
|
||||
case []any:
|
||||
dst[k] = append([]any(nil), t...)
|
||||
case []string:
|
||||
dst[k] = append([]string(nil), t...)
|
||||
default:
|
||||
dst[k] = v
|
||||
}
|
||||
}
|
||||
return dst
|
||||
}
|
||||
|
||||
func closeSecrets(m map[string]any) {
|
||||
for _, v := range m {
|
||||
switch t := v.(type) {
|
||||
case *safe.SafeBuf:
|
||||
t.Close()
|
||||
case map[string]any:
|
||||
closeSecrets(t)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func decryptMapWithPrefix(m map[string]any) []*safe.SafeBuf {
|
||||
var safeBufs []*safe.SafeBuf
|
||||
for k, v := range m {
|
||||
if s, ok := v.(string); ok && strings.HasPrefix(s, "**") {
|
||||
if s, ok := v.(string); ok && strings.HasPrefix(s, "**") && confAES != nil {
|
||||
raw := s[2:]
|
||||
var b64 []byte
|
||||
var err error
|
||||
if b64, err = encoding.UnURLBase64(raw); err != nil {
|
||||
b64, err = encoding.UnBase64(raw)
|
||||
}
|
||||
|
||||
b64, err := encoding.UnURLBase64(raw)
|
||||
if err == nil && len(b64) > 0 {
|
||||
if dec, err := confAES.DecryptBytes(b64); err == nil {
|
||||
sb := safe.NewSafeBufAndErase(dec)
|
||||
@ -236,34 +283,3 @@ func decryptMapWithPrefix(m map[string]any) []*safe.SafeBuf {
|
||||
}
|
||||
return safeBufs
|
||||
}
|
||||
|
||||
func decryptMap(m map[string]any) []*safe.SafeBuf {
|
||||
var safeBufs []*safe.SafeBuf
|
||||
for k, v := range m {
|
||||
if s, ok := v.(string); ok {
|
||||
var b64 []byte
|
||||
var err error
|
||||
// 兼容旧版 ENC(...) 格式扫描
|
||||
inner := s
|
||||
if strings.HasPrefix(s, "ENC(") && strings.HasSuffix(s, ")") {
|
||||
inner = s[4 : len(s)-1]
|
||||
}
|
||||
|
||||
if b64, err = encoding.UnURLBase64(inner); err != nil {
|
||||
b64, err = encoding.UnBase64(inner)
|
||||
}
|
||||
|
||||
if err == nil && len(b64) > 0 {
|
||||
if dec, err := confAES.DecryptBytes(b64); err == nil {
|
||||
sb := safe.NewSafeBufAndErase(dec)
|
||||
m[k] = sb
|
||||
safeBufs = append(safeBufs, sb)
|
||||
continue
|
||||
}
|
||||
}
|
||||
} else if subMap, ok := v.(map[string]any); ok {
|
||||
safeBufs = append(safeBufs, decryptMap(subMap)...)
|
||||
}
|
||||
}
|
||||
return safeBufs
|
||||
}
|
||||
|
||||
388
engine.go
Normal file
388
engine.go
Normal file
@ -0,0 +1,388 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
stdhttp "net/http"
|
||||
"net/url"
|
||||
"reflect"
|
||||
"runtime/debug"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"apigo.cc/go/cast"
|
||||
gohttp "apigo.cc/go/http"
|
||||
"apigo.cc/go/safe"
|
||||
)
|
||||
|
||||
// Call invokes an Action and returns the common API result envelope.
|
||||
func Call(action Action, options ...*CallOptions) (*Result, error) {
|
||||
if action == nil {
|
||||
return nil, errors.New("action is required")
|
||||
}
|
||||
opts := firstOptions(options)
|
||||
started := time.Now()
|
||||
result := &Result{Headers: map[string]string{}}
|
||||
actionConfig, _ := GetActionConfig(action.ActionName())
|
||||
if ca, ok := action.(ConfigurableAction); ok {
|
||||
MergeMap(actionConfig, ca.Config())
|
||||
}
|
||||
applyActionTraits(action, actionConfig)
|
||||
if enabled, exists := actionConfig["enabled"]; exists && !cast.Bool(enabled) {
|
||||
result.Error = "API Action is disabled"
|
||||
return result, errors.New(result.Error)
|
||||
}
|
||||
if cast.Bool(actionConfig["abstract"]) {
|
||||
result.Error = "abstract API Action cannot be called"
|
||||
return result, errors.New(result.Error)
|
||||
}
|
||||
if err := verifyToken(actionConfig, opts.Token); err != nil {
|
||||
result.Error = err.Error()
|
||||
logCall(opts, actionConfig, action.ActionName(), "", "", nil, result, started)
|
||||
return result, err
|
||||
}
|
||||
if err := mergeOverrides(actionConfig, opts.Config); err != nil {
|
||||
result.Error = err.Error()
|
||||
logCall(opts, actionConfig, action.ActionName(), "", "", nil, result, started)
|
||||
return result, err
|
||||
}
|
||||
|
||||
var openedSecrets []*safe.SecretPlaintext
|
||||
preprocessCallSecrets(actionConfig, &openedSecrets)
|
||||
defer func() {
|
||||
for _, secret := range openedSecrets {
|
||||
secret.Close()
|
||||
}
|
||||
}()
|
||||
if va, ok := action.(ValidatableAction); ok {
|
||||
if err := va.Validate(); err != nil {
|
||||
result.Error = fmt.Sprintf("action validation failed: %v", err)
|
||||
return result, errors.New(result.Error)
|
||||
}
|
||||
}
|
||||
fill(action, actionConfig)
|
||||
method := strings.ToUpper(cast.String(actionConfig["method"]))
|
||||
if method == "" {
|
||||
method = "POST"
|
||||
}
|
||||
requestURL, err := buildURL(actionConfig)
|
||||
if err != nil {
|
||||
result.Error = err.Error()
|
||||
return result, err
|
||||
}
|
||||
httpReq := &HttpRequest{Url: requestURL, Method: method, Payload: action, Context: opts.Context}
|
||||
defer httpReq.Close()
|
||||
if ga, ok := action.(*GenericAction); ok {
|
||||
httpReq.Payload = ga.payload
|
||||
}
|
||||
if headers, ok := actionConfig["headers"].(map[string]any); ok {
|
||||
for k, v := range headers {
|
||||
httpReq.SetHeader(k, v)
|
||||
}
|
||||
}
|
||||
filters := newFilterPipeline(actionConfig, "")
|
||||
if filtered, filterErr := filters.apply(opts.Context, "request", map[string]any{"action": action.ActionName(), "request": requestMap(httpReq, actionConfig)}); filterErr != nil {
|
||||
result.Error = filterErr.Error()
|
||||
return result, filterErr
|
||||
} else if request, ok := filtered["request"].(map[string]any); ok {
|
||||
if filteredConfig := applyRequestMap(httpReq, request); filteredConfig != nil {
|
||||
actionConfig = filteredConfig
|
||||
}
|
||||
}
|
||||
if err := sign(cast.String(actionConfig["signer"]), httpReq, actionConfig); err != nil {
|
||||
result.Error = "sign failed: " + err.Error()
|
||||
return result, errors.New(result.Error)
|
||||
}
|
||||
|
||||
payload := formattedPayload(httpReq.Payload, actionConfig)
|
||||
timeout := opts.Timeout
|
||||
if timeout <= 0 {
|
||||
timeout = cast.Duration(actionConfig["timeout"])
|
||||
}
|
||||
client := gohttp.NewClient(timeout)
|
||||
defer client.Destroy()
|
||||
if opts.Stream != nil {
|
||||
err = callStream(client, httpReq, payload, opts.Stream, filters, action.ActionName(), opts.Context, result)
|
||||
} else {
|
||||
err = callBuffered(client, httpReq, payload, result)
|
||||
if err == nil {
|
||||
applyResponseRules(actionConfig, result)
|
||||
if filtered, filterErr := filters.apply(opts.Context, "result", map[string]any{"action": action.ActionName(), "result": resultMap(result)}); filterErr != nil {
|
||||
err = filterErr
|
||||
result.Error = filterErr.Error()
|
||||
} else if value, ok := filtered["result"].(map[string]any); ok {
|
||||
applyResultMap(result, value)
|
||||
}
|
||||
}
|
||||
}
|
||||
if err == nil && opts.Stream != nil {
|
||||
applyResponseRules(actionConfig, result)
|
||||
if filtered, filterErr := filters.apply(opts.Context, "done", map[string]any{"action": action.ActionName(), "result": resultMap(result)}); filterErr != nil {
|
||||
err = filterErr
|
||||
result.Error = filterErr.Error()
|
||||
} else if value, ok := filtered["result"].(map[string]any); ok {
|
||||
applyResultMap(result, value)
|
||||
}
|
||||
}
|
||||
if err == nil && opts.Stream != nil && opts.Stream.OnDone != nil {
|
||||
err = opts.Stream.OnDone(resultMap(result))
|
||||
if err != nil {
|
||||
result.Ok = false
|
||||
result.Error = err.Error()
|
||||
}
|
||||
}
|
||||
logCall(opts, actionConfig, action.ActionName(), httpReq.Method, httpReq.Url, httpReq.Payload, result, started)
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func applyActionTraits(action Action, config map[string]any) {
|
||||
if value, ok := action.(URLAction); ok && value.GetURL() != "" {
|
||||
config["url"] = value.GetURL()
|
||||
}
|
||||
if value, ok := action.(MethodAction); ok && value.GetMethod() != "" {
|
||||
config["method"] = value.GetMethod()
|
||||
}
|
||||
if value, ok := action.(SignerAction); ok && value.SignerName() != "" {
|
||||
config["signer"] = value.SignerName()
|
||||
}
|
||||
if value, ok := action.(FormatAction); ok && value.GetFormat() != "" {
|
||||
config["format"] = value.GetFormat()
|
||||
}
|
||||
}
|
||||
|
||||
// CallBy invokes a registered dynamic or Go Action.
|
||||
func CallBy(name string, payload any, options ...*CallOptions) (*Result, error) {
|
||||
actionRegistryMutex.RLock()
|
||||
tmpl, ok := actionRegistry[name]
|
||||
actionRegistryMutex.RUnlock()
|
||||
if !ok {
|
||||
err := fmt.Errorf("action not found: %s", name)
|
||||
return &Result{Headers: map[string]string{}, Error: err.Error()}, err
|
||||
}
|
||||
var action Action
|
||||
if t, ok := tmpl.(reflect.Type); ok {
|
||||
inst := reflect.New(t).Interface()
|
||||
if payload != nil {
|
||||
cast.Convert(inst, payload)
|
||||
}
|
||||
action = inst.(Action)
|
||||
} else if ga, ok := tmpl.(*GenericAction); ok {
|
||||
config, resolveErr := resolveGenericActionConfig(name, map[string]bool{})
|
||||
if resolveErr != nil {
|
||||
return &Result{Headers: map[string]string{}, Error: resolveErr.Error()}, resolveErr
|
||||
}
|
||||
copyAction := &GenericAction{name: ga.name, config: config, payload: cloneMap(ga.payload)}
|
||||
if payload != nil {
|
||||
if m, ok := payload.(map[string]any); ok {
|
||||
MergeMap(copyAction.payload, m)
|
||||
} else {
|
||||
cast.Convert(©Action.payload, payload)
|
||||
}
|
||||
}
|
||||
action = copyAction
|
||||
}
|
||||
return Call(action, options...)
|
||||
}
|
||||
|
||||
func resolveGenericActionConfig(name string, visiting map[string]bool) (map[string]any, error) {
|
||||
if visiting[name] {
|
||||
return nil, fmt.Errorf("circular Action inheritance: %s", name)
|
||||
}
|
||||
actionRegistryMutex.RLock()
|
||||
template, exists := actionRegistry[name]
|
||||
actionRegistryMutex.RUnlock()
|
||||
generic, ok := template.(*GenericAction)
|
||||
if !exists || !ok {
|
||||
return nil, fmt.Errorf("inherited Action not found: %s", name)
|
||||
}
|
||||
visiting[name] = true
|
||||
defer delete(visiting, name)
|
||||
config := map[string]any{}
|
||||
if parent := cast.String(generic.config["extends"]); parent != "" {
|
||||
parentConfig, err := resolveGenericActionConfig(parent, visiting)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// abstract describes whether this definition itself is callable. Children
|
||||
// inherit its configuration contract, but become concrete by definition.
|
||||
delete(parentConfig, "abstract")
|
||||
MergeMap(config, parentConfig)
|
||||
}
|
||||
MergeMap(config, generic.config)
|
||||
return config, nil
|
||||
}
|
||||
|
||||
func formattedPayload(payload any, config map[string]any) any {
|
||||
switch strings.ToLower(cast.String(config["format"])) {
|
||||
case "form":
|
||||
var form gohttp.Form
|
||||
cast.Convert(&form, payload)
|
||||
return form
|
||||
case "multipart":
|
||||
var multipart gohttp.Multipart
|
||||
cast.Convert(&multipart, payload)
|
||||
return multipart
|
||||
default:
|
||||
return payload
|
||||
}
|
||||
}
|
||||
|
||||
func callBuffered(client *gohttp.Client, req *HttpRequest, payload any, result *Result) error {
|
||||
res := client.Do(req.Method, req.Url, payload, headerSlice(req)...)
|
||||
if res.Error != nil {
|
||||
result.Error = res.Error.Error()
|
||||
return res.Error
|
||||
}
|
||||
copyHTTPMeta(res.Response, result)
|
||||
if res.Response != nil && res.Response.ContentLength > 0 {
|
||||
result.receivedBytes = res.Response.ContentLength
|
||||
}
|
||||
var data any
|
||||
if err := res.To(&data); err != nil {
|
||||
result.Error = err.Error()
|
||||
return err
|
||||
}
|
||||
result.Data = data
|
||||
return nil
|
||||
}
|
||||
|
||||
func callStream(client *gohttp.Client, req *HttpRequest, payload any, stream *StreamOptions, filters *filterPipeline, action string, ctx context.Context, result *Result) error {
|
||||
res := client.ManualDo(req.Method, req.Url, payload, headerSlice(req)...)
|
||||
if res.Error != nil {
|
||||
result.Error = res.Error.Error()
|
||||
return res.Error
|
||||
}
|
||||
if res.Response == nil {
|
||||
result.Error = "empty HTTP response"
|
||||
return errors.New(result.Error)
|
||||
}
|
||||
defer res.Response.Body.Close()
|
||||
copyHTTPMeta(res.Response, result)
|
||||
if filtered, err := filters.apply(ctx, "headers", map[string]any{"action": action, "result": resultMap(result)}); err != nil {
|
||||
result.Error = err.Error()
|
||||
return err
|
||||
} else if value, ok := filtered["result"].(map[string]any); ok {
|
||||
applyResultMap(result, value)
|
||||
}
|
||||
if stream.OnHeaders != nil {
|
||||
if err := invokeStreamCallback(func() error { return stream.OnHeaders(resultMap(result)) }); err != nil {
|
||||
result.Error = err.Error()
|
||||
return err
|
||||
}
|
||||
}
|
||||
buf := make([]byte, 32*1024)
|
||||
for {
|
||||
n, readErr := res.Response.Body.Read(buf)
|
||||
if n > 0 && stream.OnData != nil {
|
||||
result.receivedBytes += int64(n)
|
||||
filtered, filterErr := filters.apply(ctx, "chunk", map[string]any{"action": action, "chunk": append([]byte(nil), buf[:n]...), "drop": false})
|
||||
if filterErr != nil {
|
||||
result.Error = filterErr.Error()
|
||||
return filterErr
|
||||
}
|
||||
chunk := filteredChunk(filtered["chunk"])
|
||||
if !cast.Bool(filtered["drop"]) && len(chunk) > 0 {
|
||||
if err := invokeStreamCallback(func() error { return stream.OnData(chunk) }); err != nil {
|
||||
result.Error = err.Error()
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
if readErr == io.EOF {
|
||||
break
|
||||
}
|
||||
if readErr != nil {
|
||||
result.Error = readErr.Error()
|
||||
return readErr
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func resultMap(result *Result) map[string]any {
|
||||
return map[string]any{
|
||||
"ok": result.Ok, "statusCode": result.StatusCode, "headers": result.Headers,
|
||||
"data": result.Data, "code": result.Code, "error": result.Error,
|
||||
}
|
||||
}
|
||||
|
||||
func invokeStreamCallback(callback func() error) (err error) {
|
||||
defer func() {
|
||||
if recovered := recover(); recovered != nil {
|
||||
err = fmt.Errorf("stream callback panic: %v\n%s", recovered, debug.Stack())
|
||||
}
|
||||
}()
|
||||
return callback()
|
||||
}
|
||||
|
||||
func copyHTTPMeta(response *stdhttp.Response, result *Result) {
|
||||
if response == nil {
|
||||
return
|
||||
}
|
||||
result.StatusCode = response.StatusCode
|
||||
for key, values := range response.Header {
|
||||
result.Headers[key] = strings.Join(values, ", ")
|
||||
}
|
||||
}
|
||||
|
||||
func buildURL(config map[string]any) (string, error) {
|
||||
raw := cast.String(config["url"])
|
||||
path := cast.String(config["path"])
|
||||
if raw == "" {
|
||||
raw = cast.String(config["baseUrl"])
|
||||
if raw == "" {
|
||||
raw = cast.String(config["host"])
|
||||
if raw != "" && !strings.Contains(raw, "://") {
|
||||
raw = "https://" + raw
|
||||
}
|
||||
}
|
||||
if path != "" {
|
||||
raw = strings.TrimRight(raw, "/") + "/" + strings.TrimLeft(path, "/")
|
||||
}
|
||||
}
|
||||
if raw == "" {
|
||||
return "", errors.New("API URL is required")
|
||||
}
|
||||
u, err := url.Parse(raw)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if cast.String(config["url"]) != "" && path != "" {
|
||||
u.Path = "/" + strings.TrimLeft(path, "/")
|
||||
u.RawPath = ""
|
||||
raw = u.String()
|
||||
}
|
||||
if values, ok := config["query"].(map[string]any); ok && len(values) > 0 {
|
||||
query := u.Query()
|
||||
for key, value := range values {
|
||||
query.Set(key, secretString(value))
|
||||
}
|
||||
u.RawQuery = query.Encode()
|
||||
return u.String(), nil
|
||||
}
|
||||
return raw, nil
|
||||
}
|
||||
|
||||
func preprocessCallSecrets(m map[string]any, opened *[]*safe.SecretPlaintext) {
|
||||
for k, v := range m {
|
||||
if sb, ok := v.(*safe.SafeBuf); ok {
|
||||
secret := sb.Open()
|
||||
m[k] = secret
|
||||
*opened = append(*opened, secret)
|
||||
} else if subMap, ok := v.(map[string]any); ok {
|
||||
preprocessCallSecrets(subMap, opened)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func secretString(value any) string {
|
||||
if secret, ok := value.(*safe.SecretPlaintext); ok && secret != nil {
|
||||
return secret.String()
|
||||
}
|
||||
return cast.String(value)
|
||||
}
|
||||
245
engine_test.go
Normal file
245
engine_test.go
Normal file
@ -0,0 +1,245 @@
|
||||
package api_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"apigo.cc/go/api"
|
||||
)
|
||||
|
||||
type testFilter func(context.Context, string, map[string]any) (map[string]any, error)
|
||||
|
||||
func (filter testFilter) Apply(ctx context.Context, event string, input map[string]any) (map[string]any, error) {
|
||||
return filter(ctx, event, input)
|
||||
}
|
||||
|
||||
func registerTestAction(t *testing.T, config map[string]any) string {
|
||||
t.Helper()
|
||||
name := "engine." + t.Name()
|
||||
api.GlobalConfigs = map[string]any{}
|
||||
api.RegisterAction(name, config)
|
||||
return name
|
||||
}
|
||||
|
||||
func TestTokenAndOverridePolicy(t *testing.T) {
|
||||
name := registerTestAction(t, map[string]any{
|
||||
"url": "http://example.invalid", "tokens": []any{"one", "two"}, "secret": "**cipher",
|
||||
})
|
||||
if result, err := api.CallBy(name, nil); err == nil || result.Error == "" {
|
||||
t.Fatal("missing token should be rejected")
|
||||
}
|
||||
if result, err := api.CallBy(name, nil, &api.CallOptions{Token: "one", Config: map[string]any{"tokens": []any{}}}); err == nil || result.Error == "" {
|
||||
t.Fatal("tokens override should be rejected")
|
||||
}
|
||||
if result, err := api.CallBy(name, nil, &api.CallOptions{Token: "one", Config: map[string]any{"secret": "plain"}}); err == nil || result.Error == "" {
|
||||
t.Fatal("encrypted value override should be rejected")
|
||||
}
|
||||
}
|
||||
|
||||
func TestResultRulesAndConfigOverride(t *testing.T) {
|
||||
var receivedPath, receivedMethod string
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
receivedPath, receivedMethod = r.URL.Path, r.Method
|
||||
w.Header().Set("X-Test", "yes")
|
||||
_, _ = w.Write([]byte(`{"code":"OK","payload":{"value":1}}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
name := registerTestAction(t, map[string]any{
|
||||
"url": "http://example.invalid", "method": "POST",
|
||||
"response": map[string]any{"codeFields": []any{"code"}, "successCodes": []any{"OK"}},
|
||||
})
|
||||
result, err := api.CallBy(name, map[string]any{"hello": "world"}, &api.CallOptions{Config: map[string]any{"url": server.URL + "/old", "path": "/v1/new", "method": "PUT"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !result.Ok || result.Code != "OK" || result.StatusCode != 200 || result.Headers["X-Test"] != "yes" {
|
||||
t.Fatalf("unexpected result: %#v", result)
|
||||
}
|
||||
if receivedPath != "/v1/new" || receivedMethod != "PUT" {
|
||||
t.Fatalf("override reached %s %s", receivedMethod, receivedPath)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamPreservesChunkOrder(t *testing.T) {
|
||||
chunks := [][]byte{[]byte("data: first\n\n"), []byte("data: second\n\n")}
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
flusher := w.(http.Flusher)
|
||||
for _, chunk := range chunks {
|
||||
_, _ = w.Write(chunk)
|
||||
flusher.Flush()
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
name := registerTestAction(t, map[string]any{"url": server.URL, "method": "GET"})
|
||||
var got bytes.Buffer
|
||||
var done bool
|
||||
result, err := api.CallBy(name, nil, &api.CallOptions{Stream: &api.StreamOptions{
|
||||
OnData: func(data []byte) error { _, _ = got.Write(data); return nil },
|
||||
OnDone: func(result map[string]any) error { done, _ = result["ok"].(bool); return nil },
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !result.Ok || !done || !reflect.DeepEqual(got.Bytes(), bytes.Join(chunks, nil)) {
|
||||
t.Fatalf("stream mismatch: ok=%v done=%v body=%q", result.Ok, done, got.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnifiedFilterLifecycle(t *testing.T) {
|
||||
var receivedMethod, receivedHeader string
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
receivedMethod, receivedHeader = r.Method, r.Header.Get("X-Filtered")
|
||||
_, _ = w.Write([]byte(`{"message":"raw"}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
beforeName, afterName := t.Name()+".before", t.Name()+".after"
|
||||
api.RegisterFilter(beforeName, testFilter(func(_ context.Context, event string, input map[string]any) (map[string]any, error) {
|
||||
if event != "request" {
|
||||
return nil, nil
|
||||
}
|
||||
request := input["request"].(map[string]any)
|
||||
request["url"] = server.URL
|
||||
request["method"] = "PUT"
|
||||
request["headers"] = map[string]any{"X-Filtered": "yes"}
|
||||
return map[string]any{"request": request}, nil
|
||||
}))
|
||||
api.RegisterFilter(afterName, testFilter(func(_ context.Context, event string, input map[string]any) (map[string]any, error) {
|
||||
if event != "result" {
|
||||
return nil, nil
|
||||
}
|
||||
result := input["result"].(map[string]any)
|
||||
result["data"] = map[string]any{"text": "normalized"}
|
||||
result["code"] = "FILTERED"
|
||||
return map[string]any{"result": result}, nil
|
||||
}))
|
||||
t.Cleanup(func() { api.RemoveFilter(beforeName); api.RemoveFilter(afterName) })
|
||||
name := registerTestAction(t, map[string]any{
|
||||
"url": "http://example.invalid", "filters": []any{beforeName, afterName},
|
||||
})
|
||||
result, err := api.CallBy(name, map[string]any{"prompt": "hello"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if receivedMethod != "PUT" || receivedHeader != "yes" {
|
||||
t.Fatalf("before filter did not update request: %s %q", receivedMethod, receivedHeader)
|
||||
}
|
||||
data, _ := result.Data.(map[string]any)
|
||||
if result.Code != "FILTERED" || data["text"] != "normalized" {
|
||||
t.Fatalf("after filter did not normalize result: %#v", result)
|
||||
}
|
||||
}
|
||||
|
||||
func TestActionInheritanceAndSharedFilters(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte(`{"value":"raw"}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
filterName := t.Name() + ".shared"
|
||||
var events []string
|
||||
api.RegisterFilter(filterName, testFilter(func(_ context.Context, event string, input map[string]any) (map[string]any, error) {
|
||||
events = append(events, event)
|
||||
if event == "request" {
|
||||
request := input["request"].(map[string]any)
|
||||
request["headers"] = map[string]any{"X-Shared": "yes"}
|
||||
return map[string]any{"request": request}, nil
|
||||
}
|
||||
if event == "result" {
|
||||
result := input["result"].(map[string]any)
|
||||
result["code"] = "NORMALIZED"
|
||||
return map[string]any{"result": result}, nil
|
||||
}
|
||||
return nil, nil
|
||||
}))
|
||||
parent, child := t.Name()+"/parent", t.Name()+"/child"
|
||||
api.RegisterAction(parent, map[string]any{"url": server.URL, "method": "POST", "filters": []any{filterName}, "headers": map[string]any{"X-Parent": "yes"}})
|
||||
api.RegisterAction(child, map[string]any{"extends": parent, "timeout": "1m"})
|
||||
t.Cleanup(func() { api.RemoveAction(parent); api.RemoveAction(child); api.RemoveFilter(filterName) })
|
||||
result, err := api.CallBy(child, map[string]any{"hello": "world"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if result.Code != "NORMALIZED" || !reflect.DeepEqual(events, []string{"request", "result"}) {
|
||||
t.Fatalf("inheritance/filter shorthand failed: %#v events=%v", result, events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestActionInheritanceRejectsCyclesAndMissingParents(t *testing.T) {
|
||||
a, b := t.Name()+"/a", t.Name()+"/b"
|
||||
api.RegisterAction(a, map[string]any{"extends": b})
|
||||
api.RegisterAction(b, map[string]any{"extends": a})
|
||||
t.Cleanup(func() { api.RemoveAction(a); api.RemoveAction(b) })
|
||||
if _, err := api.CallBy(a, nil); err == nil || !strings.Contains(err.Error(), "circular") {
|
||||
t.Fatalf("expected circular inheritance error, got %v", err)
|
||||
}
|
||||
api.RegisterAction(a, map[string]any{"extends": t.Name() + "/missing"})
|
||||
if _, err := api.CallBy(a, nil); err == nil || !strings.Contains(err.Error(), "not found") {
|
||||
t.Fatalf("expected missing parent error, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAbstractActionCannotBeCalled(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte(`{"ok":true}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
parent, child := t.Name()+"/parent", t.Name()+"/child"
|
||||
api.RegisterAction(parent, map[string]any{"abstract": true, "method": "POST"})
|
||||
api.RegisterAction(child, map[string]any{"extends": parent, "url": server.URL})
|
||||
t.Cleanup(func() { api.RemoveAction(parent); api.RemoveAction(child) })
|
||||
if _, err := api.CallBy(parent, nil); err == nil || !strings.Contains(err.Error(), "abstract") {
|
||||
t.Fatalf("expected abstract Action error, got %v", err)
|
||||
}
|
||||
if result, err := api.CallBy(child, nil); err != nil || !result.Ok {
|
||||
t.Fatalf("a child of an abstract Action must be callable: result=%#v err=%v", result, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamFilterTransformsAndDropsChunks(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte("keep\ndrop\n"))
|
||||
}))
|
||||
defer server.Close()
|
||||
filterName := t.Name() + ".after"
|
||||
var events []string
|
||||
api.RegisterFilter(filterName, testFilter(func(_ context.Context, event string, input map[string]any) (map[string]any, error) {
|
||||
events = append(events, event)
|
||||
if event == "request" {
|
||||
return map[string]any{"state": map[string]any{"requested": true}}, nil
|
||||
}
|
||||
if event == "chunk" {
|
||||
state, _ := input["state"].(map[string]any)
|
||||
if state["requested"] != true {
|
||||
t.Fatal("request filter state was not preserved for the response stream")
|
||||
}
|
||||
chunk := strings.ReplaceAll(string(input["chunk"].([]byte)), "drop\n", "")
|
||||
return map[string]any{"chunk": strings.ToUpper(chunk), "state": map[string]any{"requested": true, "seen": true}}, nil
|
||||
}
|
||||
if event == "done" {
|
||||
state, _ := input["state"].(map[string]any)
|
||||
if state["seen"] != true {
|
||||
t.Fatal("filter state was not preserved for this call")
|
||||
}
|
||||
}
|
||||
return nil, nil
|
||||
}))
|
||||
t.Cleanup(func() { api.RemoveFilter(filterName) })
|
||||
name := registerTestAction(t, map[string]any{
|
||||
"url": server.URL, "method": "GET",
|
||||
"filters": []any{filterName},
|
||||
})
|
||||
var got bytes.Buffer
|
||||
result, err := api.CallBy(name, nil, &api.CallOptions{Stream: &api.StreamOptions{
|
||||
OnData: func(data []byte) error { _, _ = got.Write(data); return nil },
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !result.Ok || got.String() != "KEEP\n" || !reflect.DeepEqual(events, []string{"request", "headers", "chunk", "done"}) {
|
||||
t.Fatalf("unexpected filtered stream: ok=%v body=%q events=%v", result.Ok, got.String(), events)
|
||||
}
|
||||
}
|
||||
182
filter.go
Normal file
182
filter.go
Normal file
@ -0,0 +1,182 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"apigo.cc/go/cast"
|
||||
)
|
||||
|
||||
// Filter transforms one request or response event. Implementations must be stateless;
|
||||
// per-call state is passed through input["state"] and returned as output["state"].
|
||||
type Filter interface {
|
||||
Apply(ctx context.Context, event string, input map[string]any) (map[string]any, error)
|
||||
}
|
||||
|
||||
var filterRegistry = map[string]Filter{}
|
||||
var filterRegistryMutex sync.RWMutex
|
||||
|
||||
// RegisterFilter registers a reusable request/response filter.
|
||||
func RegisterFilter(name string, filter Filter) {
|
||||
if name == "" || filter == nil {
|
||||
return
|
||||
}
|
||||
filterRegistryMutex.Lock()
|
||||
filterRegistry[name] = filter
|
||||
filterRegistryMutex.Unlock()
|
||||
}
|
||||
|
||||
// RegisterJSFilter registers an administrator-managed JavaScript filter.
|
||||
func RegisterJSFilter(name, code string) { RegisterFilter(name, &jsFilter{code: code}) }
|
||||
|
||||
// RemoveFilter unregisters a custom filter.
|
||||
func RemoveFilter(name string) {
|
||||
filterRegistryMutex.Lock()
|
||||
delete(filterRegistry, name)
|
||||
filterRegistryMutex.Unlock()
|
||||
}
|
||||
|
||||
// GetFilter returns one registered filter.
|
||||
func GetFilter(name string) Filter {
|
||||
filterRegistryMutex.RLock()
|
||||
filter := filterRegistry[name]
|
||||
filterRegistryMutex.RUnlock()
|
||||
return filter
|
||||
}
|
||||
|
||||
type jsFilter struct{ code string }
|
||||
|
||||
func (filter *jsFilter) Apply(ctx context.Context, event string, input map[string]any) (map[string]any, error) {
|
||||
if jsRunner == nil {
|
||||
return nil, errors.New("jsRunner is not set, cannot execute JS filter")
|
||||
}
|
||||
args := cloneMap(input)
|
||||
args["event"] = event
|
||||
if event == "chunk" {
|
||||
if chunk, ok := args["chunk"].([]byte); ok {
|
||||
args["chunk"] = string(chunk)
|
||||
}
|
||||
}
|
||||
return jsRunner(ctx, filter.code, args)
|
||||
}
|
||||
|
||||
type filterPipeline struct {
|
||||
names []string
|
||||
states map[string]map[string]any
|
||||
}
|
||||
|
||||
func newFilterPipeline(config map[string]any, _ string) *filterPipeline {
|
||||
names := stringList(config["filters"])
|
||||
if len(names) == 0 {
|
||||
return nil
|
||||
}
|
||||
return &filterPipeline{names: names, states: map[string]map[string]any{}}
|
||||
}
|
||||
|
||||
func (pipeline *filterPipeline) apply(ctx context.Context, event string, input map[string]any) (map[string]any, error) {
|
||||
if pipeline == nil {
|
||||
return input, nil
|
||||
}
|
||||
current := cloneMap(input)
|
||||
for _, name := range pipeline.names {
|
||||
filter := GetFilter(name)
|
||||
if filter == nil {
|
||||
return nil, fmt.Errorf("filter not found: %s", name)
|
||||
}
|
||||
current["filter"] = name
|
||||
current["state"] = pipeline.states[name]
|
||||
output, err := filter.Apply(ctx, event, current)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("filter %s %s failed: %w", name, event, err)
|
||||
}
|
||||
if output == nil {
|
||||
continue
|
||||
}
|
||||
if state, ok := output["state"].(map[string]any); ok {
|
||||
pipeline.states[name] = state
|
||||
}
|
||||
for key, value := range output {
|
||||
if key != "state" {
|
||||
current[key] = value
|
||||
}
|
||||
}
|
||||
if cast.Bool(current["drop"]) {
|
||||
break
|
||||
}
|
||||
}
|
||||
delete(current, "filter")
|
||||
delete(current, "state")
|
||||
return current, nil
|
||||
}
|
||||
|
||||
func requestMap(request *HttpRequest, config map[string]any) map[string]any {
|
||||
headers := map[string]any{}
|
||||
for key, value := range request.headers {
|
||||
headers[key] = value
|
||||
}
|
||||
return map[string]any{
|
||||
"url": request.Url, "method": request.Method, "headers": headers,
|
||||
"payload": request.Payload, "config": config,
|
||||
}
|
||||
}
|
||||
|
||||
func applyRequestMap(request *HttpRequest, value map[string]any) map[string]any {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
request.Url = cast.String(value["url"])
|
||||
request.Method = cast.String(value["method"])
|
||||
if payload, exists := value["payload"]; exists {
|
||||
request.Payload = payload
|
||||
}
|
||||
if headers, ok := value["headers"].(map[string]any); ok {
|
||||
request.headers = map[string]string{}
|
||||
for key, header := range headers {
|
||||
request.SetHeader(key, header)
|
||||
}
|
||||
}
|
||||
config, _ := value["config"].(map[string]any)
|
||||
return config
|
||||
}
|
||||
|
||||
func applyResultMap(result *Result, value map[string]any) {
|
||||
if result == nil || value == nil {
|
||||
return
|
||||
}
|
||||
if item, exists := value["ok"]; exists {
|
||||
result.Ok = cast.Bool(item)
|
||||
}
|
||||
if item, exists := value["statusCode"]; exists {
|
||||
result.StatusCode = cast.Int(item)
|
||||
}
|
||||
if item, exists := value["data"]; exists {
|
||||
result.Data = item
|
||||
}
|
||||
if item, exists := value["code"]; exists {
|
||||
result.Code = cast.String(item)
|
||||
}
|
||||
if item, exists := value["error"]; exists {
|
||||
result.Error = cast.String(item)
|
||||
}
|
||||
if headers, ok := value["headers"].(map[string]any); ok {
|
||||
result.Headers = map[string]string{}
|
||||
for key, header := range headers {
|
||||
result.Headers[key] = cast.String(header)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func filteredChunk(value any) []byte {
|
||||
switch chunk := value.(type) {
|
||||
case []byte:
|
||||
return chunk
|
||||
case string:
|
||||
return []byte(chunk)
|
||||
case nil:
|
||||
return nil
|
||||
default:
|
||||
return []byte(cast.String(chunk))
|
||||
}
|
||||
}
|
||||
14
go.mod
14
go.mod
@ -3,19 +3,19 @@ module apigo.cc/go/api
|
||||
go 1.25.0
|
||||
|
||||
require (
|
||||
apigo.cc/go/cast v1.5.3
|
||||
apigo.cc/go/config v1.5.3
|
||||
apigo.cc/go/crypto v1.5.3
|
||||
apigo.cc/go/encoding v1.5.4
|
||||
apigo.cc/go/http v1.5.3
|
||||
apigo.cc/go/cast v1.5.5
|
||||
apigo.cc/go/config v1.5.4
|
||||
apigo.cc/go/crypto v1.5.5
|
||||
apigo.cc/go/encoding v1.5.6
|
||||
apigo.cc/go/http v1.5.4
|
||||
apigo.cc/go/jsmod v1.5.3
|
||||
apigo.cc/go/safe v1.5.2
|
||||
apigo.cc/go/safe v1.5.3
|
||||
)
|
||||
|
||||
require (
|
||||
apigo.cc/go/file v1.5.5 // indirect
|
||||
apigo.cc/go/id v1.5.4 // indirect
|
||||
apigo.cc/go/log v1.5.8 // indirect
|
||||
apigo.cc/go/log v1.5.9 // indirect
|
||||
apigo.cc/go/rand v1.5.3 // indirect
|
||||
apigo.cc/go/shell v1.5.3 // indirect
|
||||
golang.org/x/crypto v0.52.0 // indirect
|
||||
|
||||
14
go.sum
14
go.sum
@ -1,25 +1,39 @@
|
||||
apigo.cc/go/cast v1.5.3 h1:jk6VX0rGFhjKtfPhsaV6IKYpiGmORRk9qPTtuNS53tw=
|
||||
apigo.cc/go/cast v1.5.3/go.mod h1:GMjjrYn93tWat1U409G7h1jR3ejfLLI7r0efBo9Sbd4=
|
||||
apigo.cc/go/cast v1.5.5 h1:DMbfK3uPhPjRaXutj3StIZIkfjFIATSXfuAOeNOd4Fw=
|
||||
apigo.cc/go/cast v1.5.5/go.mod h1:GMjjrYn93tWat1U409G7h1jR3ejfLLI7r0efBo9Sbd4=
|
||||
apigo.cc/go/config v1.5.3 h1:peq1FM2xO+vzPHJf8Dwg3DXm8PtFQMfTFKQj6fpoG7A=
|
||||
apigo.cc/go/config v1.5.3/go.mod h1:ZiOAjWa1mQIzszaJZN+kO6YU4GXreng+NxkcK/TAkqQ=
|
||||
apigo.cc/go/config v1.5.4 h1:1c/OarGwbz3+6ikurE+a6LJLjtzXvGvbxw3HV/Nk54M=
|
||||
apigo.cc/go/config v1.5.4/go.mod h1:oN+D2F8ETIyqKp+Yu8R4PRQlUoqR44o35jAHwLrrAq0=
|
||||
apigo.cc/go/crypto v1.5.3 h1:2JUHC2cgR2zrnn36EzwkUAdxmmTXAA/8yTNo+2X1mPE=
|
||||
apigo.cc/go/crypto v1.5.3/go.mod h1:PheYKHEXmoEFI1AK5PpY1borQWcRlkkSaWncT3cWbhE=
|
||||
apigo.cc/go/crypto v1.5.5 h1:YQHumieqviNGMhwoxDtMuUdGWAQiGrkVhz9hZRHOhWs=
|
||||
apigo.cc/go/crypto v1.5.5/go.mod h1:z/FXt0HE7fSMJKF2MF7jY2fSjlHUYYaSffETJ7ZzJmE=
|
||||
apigo.cc/go/encoding v1.5.4 h1:Fk8TrveZATyy8SHukC4ZiqdTSp+QIfsRHtt55xmMK7w=
|
||||
apigo.cc/go/encoding v1.5.4/go.mod h1:dShEsZ3gKqBINz7TSOYf4e7/fBCqCY9VzlenoGUQUFM=
|
||||
apigo.cc/go/encoding v1.5.6 h1:v02swVfbFGidD4QcX2ktuHHbCjdSbOB85fhzAXay+7M=
|
||||
apigo.cc/go/encoding v1.5.6/go.mod h1:Big9q1Zwy4071dXtnrQ3SJDzfa/G7/A60KE/5+M//P8=
|
||||
apigo.cc/go/file v1.5.5 h1:/+HmDumLu6Qk2KuQL63M9lpgzHTDL+QJ8dStOl7e9gs=
|
||||
apigo.cc/go/file v1.5.5/go.mod h1:xRVNhctvqOKeBemmcRW/BQfgkc3B+vT/UZVdSc7duUo=
|
||||
apigo.cc/go/http v1.5.3 h1:nvJh9bqPPcPRv6p8WEw7bJAd0UC+r2zvQA8/QioVLTQ=
|
||||
apigo.cc/go/http v1.5.3/go.mod h1:cFrPK61y9f1PrsNSJscZT/QVOgkT15o9OP7O8cuMb8Q=
|
||||
apigo.cc/go/http v1.5.4 h1:Tm95WMsXyazFnaaPopdMXv1k9TiaPenLxDQ40WCrsEQ=
|
||||
apigo.cc/go/http v1.5.4/go.mod h1:cFrPK61y9f1PrsNSJscZT/QVOgkT15o9OP7O8cuMb8Q=
|
||||
apigo.cc/go/id v1.5.4 h1:D1Zx9gEZhOgdTgZ4SdmPImhpc9xGiOA33Y+j2MkstzQ=
|
||||
apigo.cc/go/id v1.5.4/go.mod h1:hCTQq+KC1ALWe1FpPERf+W4B6FSulg9FAgOUJDDySiY=
|
||||
apigo.cc/go/jsmod v1.5.3 h1:S3W317bH0QV2NMeRO1E0v6ySIBOfMWYv/NuQJbvqKWU=
|
||||
apigo.cc/go/jsmod v1.5.3/go.mod h1:bmyeZtOAP/j5am+YRnaiM89smysK24K7ebk0koFtsSw=
|
||||
apigo.cc/go/log v1.5.8 h1:/IYtGPWhRjT3OayylDIphkWZIQbpLjqVeSnFEiD3Dy0=
|
||||
apigo.cc/go/log v1.5.8/go.mod h1:HfFPANMYxJx197SSTXB21Pgxcz/gGqPP8nlSErgd5WE=
|
||||
apigo.cc/go/log v1.5.9 h1:g8JehZrpVJyesesv+JFig5szEELsO4UjShsjlEihCfc=
|
||||
apigo.cc/go/log v1.5.9/go.mod h1:b4f/UB5Kk7oiFtkXvG4GVilmHGygWgxELRAdEZrIyfs=
|
||||
apigo.cc/go/rand v1.5.3 h1:O4bPIwyaOWEBCr0nL9A4G4qG48AqiGTCzfPeckm3Ius=
|
||||
apigo.cc/go/rand v1.5.3/go.mod h1:q1BTFkY/cXE229dDD5Q22lF7T0DoKPV6xAu+6bCrDH4=
|
||||
apigo.cc/go/safe v1.5.2 h1:EnuEOW/SGwf/5A0nw9LnqfKJE071+TIc6ez8HI9R9Lg=
|
||||
apigo.cc/go/safe v1.5.2/go.mod h1:2GqCCLLGex4OAhdET3iBWm1R+LIYtmTrvHP8W0iESSw=
|
||||
apigo.cc/go/safe v1.5.3 h1:9p/BmdlVWLbekpKByZIFC09Qn8Wdhik2eINiwunBxPs=
|
||||
apigo.cc/go/safe v1.5.3/go.mod h1:Ay8kEPL76DeXH4ifsVTc/3/sfGHlWLQjAp4vi7GA9AI=
|
||||
apigo.cc/go/shell v1.5.3 h1:pI+u12sy6upoygq+1XXqUlvUboBfH4Q52jRpoJFv56A=
|
||||
apigo.cc/go/shell v1.5.3/go.mod h1:FdZWUrcXHGJXo725oSyHqAeFoX0E9yY3PDhrz9hujgY=
|
||||
github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
|
||||
|
||||
10
js_export.go
10
js_export.go
@ -12,14 +12,20 @@ func init() {
|
||||
"SetConfig": SetConfig,
|
||||
"RegisterAction": RegisterAction,
|
||||
"RegisterSigner": registerSigner,
|
||||
"Encrypt": Encrypt,
|
||||
})
|
||||
}
|
||||
|
||||
// call 提供给 JS 的私有入口
|
||||
func call(ctx context.Context, name string, payload any) (any, error) {
|
||||
func call(ctx context.Context, name string, payload any, options ...*CallOptions) (any, error) {
|
||||
// 将 ctx 传入以透传追踪信息给 JS 签名器
|
||||
res, err := CallBy[any](name, payload)
|
||||
opts := firstOptions(options)
|
||||
opts.Context = ctx
|
||||
res, err := CallBy(name, payload, opts)
|
||||
if err != nil {
|
||||
if res != nil {
|
||||
return res, nil
|
||||
}
|
||||
return nil, jsmod.MakeError(err)
|
||||
}
|
||||
return res, nil
|
||||
|
||||
174
log.go
Normal file
174
log.go
Normal file
@ -0,0 +1,174 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"reflect"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"apigo.cc/go/cast"
|
||||
"apigo.cc/go/log"
|
||||
"apigo.cc/go/safe"
|
||||
)
|
||||
|
||||
const logTypeAPICall = "api_call"
|
||||
|
||||
type CallLog struct {
|
||||
log.BaseLog
|
||||
Action string `log:"pos:6,color:cyan"`
|
||||
Method string `log:"pos:7,color:gray"`
|
||||
URL string `log:"pos:8,color:gray"`
|
||||
StatusCode int `log:"pos:9,color:magenta,keyname:Status"`
|
||||
Code string `log:"pos:10,color:magenta"`
|
||||
Error string `log:"pos:11,color:red"`
|
||||
UsedTime float32 `log:"pos:12,color:green,precision:6"`
|
||||
Stream bool `log:"pos:13"`
|
||||
ResponseSize int64 `log:"pos:14,color:magenta,keyname:Size"`
|
||||
RequestData any `log:"pos:15,color:cyan,keyname:Request"`
|
||||
ResponseData any `log:"pos:16,color:magenta,keyname:Response"`
|
||||
}
|
||||
|
||||
func (entry *CallLog) Reset() {
|
||||
entry.BaseLog.Reset()
|
||||
entry.Action = ""
|
||||
entry.Method = ""
|
||||
entry.URL = ""
|
||||
entry.StatusCode = 0
|
||||
entry.Code = ""
|
||||
entry.Error = ""
|
||||
entry.UsedTime = 0
|
||||
entry.Stream = false
|
||||
entry.ResponseSize = 0
|
||||
entry.RequestData = nil
|
||||
entry.ResponseData = nil
|
||||
}
|
||||
|
||||
func init() {
|
||||
log.RegisterType(logTypeAPICall, &CallLog{})
|
||||
}
|
||||
|
||||
func logCall(options *CallOptions, config map[string]any, action, method, url string, request any, result *Result, started time.Time) {
|
||||
logger := options.Logger
|
||||
if logger == nil {
|
||||
logger = log.DefaultLogger
|
||||
}
|
||||
if logger == nil || !logger.CheckLevel(log.INFO) {
|
||||
return
|
||||
}
|
||||
entry := log.GetEntry[CallLog]()
|
||||
logger.FillBase(entry.GetBaseLog(), logTypeAPICall)
|
||||
entry.Action = action
|
||||
entry.Method = method
|
||||
entry.URL = sanitizeLogURL(url, config)
|
||||
entry.StatusCode = result.StatusCode
|
||||
entry.Code = result.Code
|
||||
entry.Error = result.Error
|
||||
entry.UsedTime = float32(time.Since(started).Seconds())
|
||||
entry.Stream = options.Stream != nil
|
||||
entry.ResponseSize = result.receivedBytes
|
||||
logging, _ := config["logging"].(map[string]any)
|
||||
if cast.Bool(logging["request"]) {
|
||||
entry.RequestData = logSnapshot(request, logging)
|
||||
}
|
||||
if cast.Bool(logging["response"]) {
|
||||
entry.ResponseData = logSnapshot(result.Data, logging)
|
||||
}
|
||||
logger.Log(entry)
|
||||
}
|
||||
|
||||
func sanitizeLogURL(raw string, config map[string]any) string {
|
||||
parsed, err := url.Parse(raw)
|
||||
if err != nil {
|
||||
return raw
|
||||
}
|
||||
configured, _ := config["query"].(map[string]any)
|
||||
query := parsed.Query()
|
||||
changed := false
|
||||
for key, value := range configured {
|
||||
if _, ok := value.(*safe.SecretPlaintext); ok {
|
||||
query.Set(key, "***")
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
if changed {
|
||||
parsed.RawQuery = query.Encode()
|
||||
}
|
||||
return parsed.String()
|
||||
}
|
||||
|
||||
func logSnapshot(value any, config map[string]any) any {
|
||||
maxText := cast.Int(config["maxTextLength"])
|
||||
if maxText <= 0 {
|
||||
maxText = 4096
|
||||
}
|
||||
maxItems := cast.Int(config["maxArrayItems"])
|
||||
if maxItems <= 0 {
|
||||
maxItems = 3
|
||||
}
|
||||
maxDepth := cast.Int(config["maxDepth"])
|
||||
if maxDepth <= 0 {
|
||||
maxDepth = 4
|
||||
}
|
||||
return trimLogValue(reflect.ValueOf(value), 0, maxDepth, maxItems, maxText)
|
||||
}
|
||||
|
||||
func trimLogValue(value reflect.Value, depth, maxDepth, maxItems, maxText int) any {
|
||||
if !value.IsValid() {
|
||||
return nil
|
||||
}
|
||||
for value.Kind() == reflect.Interface || value.Kind() == reflect.Ptr {
|
||||
if value.IsNil() {
|
||||
return nil
|
||||
}
|
||||
value = value.Elem()
|
||||
}
|
||||
if depth >= maxDepth {
|
||||
return "[max depth]"
|
||||
}
|
||||
switch value.Kind() {
|
||||
case reflect.String:
|
||||
text := value.String()
|
||||
if len(text) > maxText {
|
||||
return text[:maxText] + fmt.Sprintf("...[truncated %d bytes]", len(text)-maxText)
|
||||
}
|
||||
return text
|
||||
case reflect.Map:
|
||||
out := map[string]any{}
|
||||
iterator := value.MapRange()
|
||||
for iterator.Next() {
|
||||
out[cast.String(iterator.Key().Interface())] = trimLogValue(iterator.Value(), depth+1, maxDepth, maxItems, maxText)
|
||||
}
|
||||
return out
|
||||
case reflect.Slice, reflect.Array:
|
||||
if value.Type().Elem().Kind() == reflect.Uint8 {
|
||||
return fmt.Sprintf("[binary %d bytes]", value.Len())
|
||||
}
|
||||
limit := value.Len()
|
||||
if limit > maxItems {
|
||||
limit = maxItems
|
||||
}
|
||||
out := make([]any, 0, limit+1)
|
||||
for i := 0; i < limit; i++ {
|
||||
out = append(out, trimLogValue(value.Index(i), depth+1, maxDepth, maxItems, maxText))
|
||||
}
|
||||
if value.Len() > limit {
|
||||
out = append(out, fmt.Sprintf("[%d more items]", value.Len()-limit))
|
||||
}
|
||||
return out
|
||||
case reflect.Struct:
|
||||
out := map[string]any{}
|
||||
typeOf := value.Type()
|
||||
for i := 0; i < value.NumField(); i++ {
|
||||
if typeOf.Field(i).PkgPath == "" {
|
||||
out[cast.GetLowerName(typeOf.Field(i).Name)] = trimLogValue(value.Field(i), depth+1, maxDepth, maxItems, maxText)
|
||||
}
|
||||
}
|
||||
return out
|
||||
default:
|
||||
if value.CanInterface() {
|
||||
return value.Interface()
|
||||
}
|
||||
return strings.TrimSpace(fmt.Sprint(value))
|
||||
}
|
||||
}
|
||||
32
options.go
Normal file
32
options.go
Normal file
@ -0,0 +1,32 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"apigo.cc/go/log"
|
||||
)
|
||||
|
||||
// CallOptions controls one API invocation without changing the stored Action.
|
||||
type CallOptions struct {
|
||||
Context context.Context
|
||||
Token string
|
||||
Config map[string]any
|
||||
Timeout time.Duration
|
||||
Logger *log.Logger
|
||||
Stream *StreamOptions
|
||||
}
|
||||
|
||||
// StreamOptions receives the upstream response synchronously on the calling goroutine.
|
||||
type StreamOptions struct {
|
||||
OnHeaders func(result map[string]any) error
|
||||
OnData func(data []byte) error
|
||||
OnDone func(result map[string]any) error
|
||||
}
|
||||
|
||||
func firstOptions(options []*CallOptions) *CallOptions {
|
||||
if len(options) > 0 && options[0] != nil {
|
||||
return options[0]
|
||||
}
|
||||
return &CallOptions{}
|
||||
}
|
||||
174
policy.go
Normal file
174
policy.go
Normal file
@ -0,0 +1,174 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strings"
|
||||
|
||||
"apigo.cc/go/cast"
|
||||
"apigo.cc/go/safe"
|
||||
)
|
||||
|
||||
var overrideBlacklist = map[string]bool{
|
||||
"tokens": true, "enabled": true, "test": true, "logging": true,
|
||||
}
|
||||
|
||||
func verifyToken(config map[string]any, token string) error {
|
||||
raw, exists := config["tokens"]
|
||||
if !exists || raw == nil {
|
||||
return nil
|
||||
}
|
||||
tokens := make([]string, 0)
|
||||
switch values := raw.(type) {
|
||||
case []string:
|
||||
for _, value := range values {
|
||||
if value != "" {
|
||||
tokens = append(tokens, value)
|
||||
}
|
||||
}
|
||||
case []any:
|
||||
for _, value := range values {
|
||||
if token := cast.String(value); token != "" {
|
||||
tokens = append(tokens, token)
|
||||
}
|
||||
}
|
||||
default:
|
||||
v := reflect.ValueOf(raw)
|
||||
if v.IsValid() && (v.Kind() == reflect.Array || v.Kind() == reflect.Slice) {
|
||||
for i := 0; i < v.Len(); i++ {
|
||||
tokens = append(tokens, cast.String(v.Index(i).Interface()))
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(tokens) == 0 {
|
||||
return nil
|
||||
}
|
||||
for _, allowed := range tokens {
|
||||
if allowed == token {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return errors.New("API token is required or invalid")
|
||||
}
|
||||
|
||||
func mergeOverrides(config, overrides map[string]any) error {
|
||||
if len(overrides) == 0 {
|
||||
return nil
|
||||
}
|
||||
return mergeAllowed(config, overrides, "")
|
||||
}
|
||||
|
||||
func mergeAllowed(dst, src map[string]any, prefix string) error {
|
||||
for key, value := range src {
|
||||
path := key
|
||||
if prefix != "" {
|
||||
path = prefix + "." + key
|
||||
}
|
||||
if overrideBlacklist[key] {
|
||||
return fmt.Errorf("configuration field %s cannot be overridden", path)
|
||||
}
|
||||
current, exists := dst[key]
|
||||
if exists && isSecretValue(current) {
|
||||
return fmt.Errorf("encrypted configuration field %s cannot be overridden", path)
|
||||
}
|
||||
if child, ok := value.(map[string]any); ok {
|
||||
currentMap, _ := current.(map[string]any)
|
||||
if currentMap == nil {
|
||||
currentMap = map[string]any{}
|
||||
dst[key] = currentMap
|
||||
}
|
||||
if err := mergeAllowed(currentMap, child, path); err != nil {
|
||||
return err
|
||||
}
|
||||
continue
|
||||
}
|
||||
dst[key] = value
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func isSecretValue(value any) bool {
|
||||
switch t := value.(type) {
|
||||
case string:
|
||||
return strings.HasPrefix(t, "**")
|
||||
case *safe.SafeBuf, *safe.SecretPlaintext:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func applyResponseRules(config map[string]any, result *Result) {
|
||||
response, _ := config["response"].(map[string]any)
|
||||
result.Code = firstPathString(result.Data, response["codeFields"])
|
||||
if result.Error == "" {
|
||||
result.Error = firstPathString(result.Data, response["errorFields"])
|
||||
}
|
||||
result.Ok = result.StatusCode >= 200 && result.StatusCode < 300 && result.Error == ""
|
||||
if (result.StatusCode < 200 || result.StatusCode >= 300) && result.Error == "" {
|
||||
result.Error = fmt.Sprintf("HTTP status %d", result.StatusCode)
|
||||
}
|
||||
if codes := stringList(response["successCodes"]); len(codes) > 0 {
|
||||
if result.Code == "" {
|
||||
result.Ok = false
|
||||
if result.Error == "" {
|
||||
result.Error = "success code field was not found"
|
||||
}
|
||||
return
|
||||
}
|
||||
result.Ok = false
|
||||
for _, code := range codes {
|
||||
if result.Code == code {
|
||||
result.Ok = result.Error == "" && result.StatusCode >= 200 && result.StatusCode < 300
|
||||
break
|
||||
}
|
||||
}
|
||||
if !result.Ok && result.Error == "" {
|
||||
result.Error = "unexpected response code: " + result.Code
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func stringList(value any) []string {
|
||||
var result []string
|
||||
switch values := value.(type) {
|
||||
case []string:
|
||||
return append(result, values...)
|
||||
case []any:
|
||||
for _, item := range values {
|
||||
result = append(result, cast.String(item))
|
||||
}
|
||||
case string:
|
||||
if values != "" {
|
||||
result = append(result, values)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func firstPathString(data, fields any) string {
|
||||
for _, path := range stringList(fields) {
|
||||
if value, ok := pathValue(data, path); ok {
|
||||
if text := cast.String(value); text != "" {
|
||||
return text
|
||||
}
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func pathValue(data any, path string) (any, bool) {
|
||||
current := data
|
||||
for _, part := range strings.Split(path, ".") {
|
||||
m, ok := current.(map[string]any)
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
current, ok = m[part]
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
}
|
||||
return current, true
|
||||
}
|
||||
@ -1,6 +1,7 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"apigo.cc/go/encoding"
|
||||
@ -14,22 +15,25 @@ func TestSafeConfigDecryption(t *testing.T) {
|
||||
SetEncryptKeys(key, iv)
|
||||
|
||||
plaintext := "my-secret-password"
|
||||
ciphertext, _ := confAES.EncryptBytes([]byte(plaintext))
|
||||
b64 := encoding.Base64(ciphertext)
|
||||
|
||||
GlobalConfigs = map[string]any{
|
||||
"api": map[string]any{
|
||||
"testSvc": map[string]any{
|
||||
"password": b64,
|
||||
"username": "admin",
|
||||
},
|
||||
},
|
||||
encoded, err := Encrypt(plaintext)
|
||||
if err != nil || !strings.HasPrefix(encoded, "**") {
|
||||
t.Fatalf("Encrypt() = %q, %v", encoded, err)
|
||||
}
|
||||
ciphertext, _ := confAES.EncryptBytes([]byte(plaintext))
|
||||
b64 := "**" + encoding.URLBase64(ciphertext)
|
||||
|
||||
GlobalConfigs = map[string]any{}
|
||||
SetConfig("api", map[string]any{
|
||||
"testSvc": map[string]any{
|
||||
"password": b64,
|
||||
"username": "admin",
|
||||
},
|
||||
})
|
||||
|
||||
// 2. 测试获取配置
|
||||
cfg, sbs := GetActionConfig("testSvc")
|
||||
if len(sbs) != 1 {
|
||||
t.Fatalf("expected 1 SafeBuf, got %d", len(sbs))
|
||||
if len(sbs) != 0 {
|
||||
t.Fatalf("expected persistent config secrets, got %d temporary buffers", len(sbs))
|
||||
}
|
||||
|
||||
sb, ok := cfg["password"].(*safe.SafeBuf)
|
||||
@ -46,7 +50,7 @@ func TestSafeConfigDecryption(t *testing.T) {
|
||||
// 3. 测试签名器使用 SafeBuf
|
||||
req := &HttpRequest{}
|
||||
signer := GetSigner("basic")
|
||||
err := signer.Sign(req, cfg)
|
||||
err = signer.Sign(req, cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@ -60,20 +64,10 @@ func TestSafeConfigDecryption(t *testing.T) {
|
||||
authStr := req.GetHeader("Authorization")
|
||||
req.Close()
|
||||
|
||||
for _, sb := range sbs {
|
||||
sb.Close()
|
||||
}
|
||||
|
||||
// 验证 Authorization Header 已被擦除 (内容不再是原始数据)
|
||||
// 验证请求 Header 在调用结束后被擦除,常驻配置则仍可用于下一次调用。
|
||||
if authStr == expectedAuth {
|
||||
t.Error("Authorization header should be modified/erased after Close")
|
||||
}
|
||||
|
||||
// 再次尝试 Open 应该失败或得到空 (取决于 SafeBuf 实现,通常 Close 后内容被擦除)
|
||||
p2 := sb.Open()
|
||||
if p2.String() == plaintext && len(plaintext) > 0 {
|
||||
t.Error("SafeBuf should be cleared after Close")
|
||||
}
|
||||
}
|
||||
|
||||
func TestFillSafeGuard(t *testing.T) {
|
||||
|
||||
32
signer.go
32
signer.go
@ -3,6 +3,7 @@ package api
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
|
||||
"apigo.cc/go/safe"
|
||||
)
|
||||
@ -16,6 +17,7 @@ var signers = map[string]Signer{
|
||||
"basic": &basicSigner{},
|
||||
"bearer": &bearerSigner{},
|
||||
}
|
||||
var signersMutex sync.RWMutex
|
||||
|
||||
var jsRunner func(ctx context.Context, code string, args map[string]any) (map[string]any, error)
|
||||
|
||||
@ -27,12 +29,32 @@ func SetJSRunner(runner func(context.Context, string, map[string]any) (map[strin
|
||||
|
||||
// RegisterSigner 注册全局签名器
|
||||
func RegisterSigner(name string, s Signer) {
|
||||
signersMutex.Lock()
|
||||
signers[name] = s
|
||||
signersMutex.Unlock()
|
||||
}
|
||||
|
||||
// RegisterJSSigner registers an administrator-managed JavaScript signer.
|
||||
func RegisterJSSigner(name, code string) {
|
||||
RegisterSigner(name, &jsSigner{code: code})
|
||||
}
|
||||
|
||||
// RemoveSigner unregisters a custom signer. Built-in signers are retained.
|
||||
func RemoveSigner(name string) {
|
||||
if name == "basic" || name == "bearer" || name == "none" {
|
||||
return
|
||||
}
|
||||
signersMutex.Lock()
|
||||
delete(signers, name)
|
||||
signersMutex.Unlock()
|
||||
}
|
||||
|
||||
// GetSigner 获取签名器
|
||||
func GetSigner(name string) Signer {
|
||||
return signers[name]
|
||||
signersMutex.RLock()
|
||||
signer := signers[name]
|
||||
signersMutex.RUnlock()
|
||||
return signer
|
||||
}
|
||||
|
||||
// jsSigner 包装 JS 代码为 Go Signer 接口
|
||||
@ -56,7 +78,11 @@ func (s *jsSigner) Sign(req *HttpRequest, config map[string]any) error {
|
||||
|
||||
// 执行 JS 签名逻辑
|
||||
// 注意:这里默认使用 Background context,除非底层能透传,低代码环境中会由 js_export 注入正确的 ctx
|
||||
res, err := jsRunner(context.Background(), s.code, args)
|
||||
ctx := req.Context
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
res, err := jsRunner(ctx, s.code, args)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -81,7 +107,7 @@ func (s *jsSigner) Sign(req *HttpRequest, config map[string]any) error {
|
||||
|
||||
// 快速应用签名
|
||||
func sign(name string, req *HttpRequest, config map[string]any) error {
|
||||
if name == "" {
|
||||
if name == "" || name == "none" {
|
||||
return nil
|
||||
}
|
||||
s := GetSigner(name)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user