fix(library): 期望执行器身份随 capture 请求送达并在执行前核对
build / build (push) Successful in 1m27s

原来 worker 先 health 再 capture,两次拨号之间 socket 可能被换成另一个实例的
执行器:错的 agent 已经把写入者停了、快照也出了,事后 postcheck 只能拒绝登记,
副作用无法撤销。

现在 capture 请求自带期望身份(治理侧配置生成:project 必填,source_app /
archive_root 配了就绑定)与覆盖三项的稳定指纹;接收方在操作互斥内、执行任何源
操作之前比对,缺字段、指纹与字段不一致、project/source_app 不符都直接拒绝,
**错的执行器一次 capture 都不跑**。封存结果回显该指纹,worker 事后复核按它对照,
postcheck 只作为证据防线。恢复路径不要求期望(它只把写入者恢复回来)。

负例是确定性的:两个真实 agent 各占一个 socket,把调用方指向错的那个,断言它的
capture 调用为 0 且仍能回答 health;同一份期望打到对的 socket 则只跑一次。
This commit is contained in:
姚章浩
2026-10-29 13:23:56 +08:00
parent 5d42cba6c6
commit 6ed06357f9
12 changed files with 382 additions and 161 deletions
+7
View File
@@ -24,6 +24,13 @@
### Changed
- **封存身份核对移进请求本身(TOCTOU)**:原来 worker 先 `health` 再 `capture`,两次
拨号之间 socket 可能被换成另一个 agent——错的执行器已经把写入者停了才被拒登记。
现在 `capture` 请求携带治理侧配置生成的期望身份(`project` 必填,`source_app`/
`archive_root` 配了就一起绑定)与稳定指纹(`sha256:`,覆盖三项);接收方在**操作
互斥内、执行任何源操作之前**比对,缺失或被改过一律拒绝,错的执行器一次 `capture`
都不跑。封存结果回显该指纹,worker 事后复核只作为证据防线,不能替代请求内核对。
恢复路径不受期望约束(它只把写入者恢复回来)。
- `LIBRARY_EXECUTOR_STATUS_MAX_AGE`(默认 120 秒)控制执行器观测的有效期;
未设置不等于"永不过期",过期即"无法确认"。
- 页面/接口不再出现写死的"封存通道未接线""生产恢复未接线":不能发起时给出真实原因、
+43 -63
View File
@@ -1,89 +1,69 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"knowledge-governance/internal/appconfig"
"knowledge-governance/internal/libraryagent"
)
// 封存执行的身份核对分两次:执行前一次、执行后一次。
// 封存的身份契约:期望随**同一个 capture 请求**送进 agent,由接收方在操作互斥内、
// 任何源操作之前比对;worker 事后按回显的指纹再复核一次。
//
// 为什么心跳不够:心跳说明"上一次探测时"那个 socket 后面是哪个实例,而从心跳到
// 真正停写入者之间,socket 可能被换成另一个 agent(换挂载、换单元、误启正式单元)。
// 执行前探测把窗口缩到最小;执行后复核 agent 自报的身份,则把"换过 socket"这件事
// 变成一次显式的作业失败,而不是悄悄封错了库还登记成测试封存。
// 为什么不能在 worker 侧先 Health 再 Capture:那是两次拨号,中间 socket 可能被换成
// 另一个 agent,错的执行器已经把写入者停了才被拒登记——副作用已经发生。这里把期望
// 放进请求本身,换过的 agent 会先比对身份、比对失败就一次 run 都不调。
//
// 期望字段只来自治理侧配置(LIBRARY_SOURCE_*),绝不从执行器自报值生成。
// captureExecutorPrecheck 在动源写入者之前确认执行器就是期望的那个实例。
func captureExecutorPrecheck(ctx context.Context, client libraryagent.ClientConfig, expectedProject string, probeTimeout time.Duration) error {
if strings.TrimSpace(expectedProject) == "" {
return errors.New("未配置封存来源实例身份(LIBRARY_SOURCE_PROJECT),拒绝在无法核对执行器的情况下停写入者")
// captureExpectation 把治理侧配置翻译成期望;缺身份就是缺配置,调用方必须失败。
func captureExpectation(config appconfig.Config) (libraryagent.Expectation, error) {
expectation := libraryagent.NewExpectation(config.SourceProject(), config.LibrarySourceApp, "")
if expectation.Project == "" {
return libraryagent.Expectation{}, errors.New("未配置封存来源实例身份(LIBRARY_SOURCE_PROJECT),拒绝在不核对执行器的情况下停写入者")
}
if probeTimeout <= 0 {
probeTimeout = 3 * time.Second
}
bounded, cancel := context.WithTimeout(ctx, probeTimeout)
defer cancel()
result, err := libraryagent.Health(bounded, client)
if err != nil {
return fmt.Errorf("封存执行器不可达,拒绝停写入者:%w", err)
}
identity, err := agentIdentityOf(result.Payload)
if err != nil {
return err
}
if identity.Project != expectedProject {
return fmt.Errorf("封存执行器指向的实例不匹配(agent=%q,期望=%q),拒绝停写入者",
identity.Project, expectedProject)
}
if identity.Degraded != "" {
return fmt.Errorf("封存执行器处于降级状态(%s),拒绝停写入者", identity.Degraded)
}
return nil
return expectation, nil
}
// captureExecutorPostcheck 用封存结果里 agent 自报的身份复核:不一致就当作
// "封存可能来自别的实例",调用方不得把它登记成测试封存。
func captureExecutorPostcheck(payload json.RawMessage, expectedProject string) error {
identity, err := agentIdentityOf(payload)
if err != nil {
return err
// captureExecutorPostcheck 是事后证据防线:结果里回显的期望指纹必须等于我们发出的,
// 自报身份也必须与期望一致。它挡不住副作用,所以不能替代请求内的核对。
func captureExecutorPostcheck(payload json.RawMessage, expectation libraryagent.Expectation) error {
if len(payload) == 0 {
return errors.New("封存结果里没有执行器身份,无法证明这次封存来自期望实例")
}
var document struct {
Identity struct {
Project string `json:"project"`
SourceApp string `json:"source_app"`
ExpectationFingerprint string `json:"expectation_fingerprint"`
} `json:"agent_identity"`
}
if err := json.Unmarshal(payload, &document); err != nil {
return errors.New("封存结果里的执行器身份无法解析")
}
identity := document.Identity
if identity.Project == "" {
return errors.New("封存结果里没有执行器身份,无法证明这次封存来自期望实例")
}
if identity.Project != expectedProject {
if identity.ExpectationFingerprint != expectation.Fingerprint {
return fmt.Errorf("封存结果回显的期望指纹不一致(agent=%q,期望=%q),该快照不能当作本实例的测试封存",
identity.ExpectationFingerprint, expectation.Fingerprint)
}
if identity.Project != expectation.Project {
return fmt.Errorf("封存结果来自另一个实例(agent=%q,期望=%q),该快照不能当作本实例的测试封存",
identity.Project, expectedProject)
identity.Project, expectation.Project)
}
if expectation.SourceApp != "" && identity.SourceApp != expectation.SourceApp {
return fmt.Errorf("封存结果来自另一个源容器(agent=%q,期望=%q),该快照不能当作本实例的测试封存",
identity.SourceApp, expectation.SourceApp)
}
return nil
}
// agentIdentity 是 agent 在 health/封存结果里自报的身份(无凭据)。
type agentIdentity struct {
Project string `json:"project"`
SourceApp string `json:"source_app"`
Degraded string `json:"degraded"`
}
func agentIdentityOf(payload json.RawMessage) (agentIdentity, error) {
if len(payload) == 0 {
return agentIdentity{}, errors.New("执行器应答没有身份信息")
}
var document struct {
Identity agentIdentity `json:"agent_identity"`
agentIdentity
}
if err := json.Unmarshal(payload, &document); err != nil {
return agentIdentity{}, errors.New("执行器身份无法解析")
}
// health 把身份摊平在顶层;封存结果把它放在 agent_identity 下。
if document.Identity.Project != "" {
return document.Identity, nil
}
return document.agentIdentity, nil
// expectationFingerprint 用于日志/证据行的稳定标识,不含路径与凭据。
func expectationFingerprint(expectation libraryagent.Expectation) string {
return strings.TrimSpace(expectation.Fingerprint)
}
+46 -71
View File
@@ -1,98 +1,73 @@
package main
import (
"context"
"encoding/json"
"net"
"os"
"path/filepath"
"strings"
"testing"
"time"
"knowledge-governance/internal/appconfig"
"knowledge-governance/internal/libraryagent"
)
func shortDir(t *testing.T) string {
t.Helper()
// unix socket 路径有长度上限,测试目录要短。
dir, err := os.MkdirTemp("/tmp", "kgcap-")
// 期望只从治理侧配置生成;缺 project 就是缺配置,必须在停写入者之前失败。
func TestCaptureExpectationComesFromGovernanceConfigAndFailsClosed(t *testing.T) {
expect, err := captureExpectation(appconfig.Config{LibrarySourceInstance: "kgmaint-weknora", LibrarySourceApp: "kgmaint-app"})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
return dir
}
func fakeAgent(t *testing.T, payload map[string]any) string {
t.Helper()
socket := filepath.Join(shortDir(t), "agent.sock")
listener, err := net.Listen("unix", socket)
if err != nil {
t.Fatal(err)
if expect.Project != "kgmaint-weknora" || expect.SourceApp != "kgmaint-app" {
t.Fatalf("expectation = %+v", expect)
}
t.Cleanup(func() { listener.Close() })
go func() {
for {
conn, acceptErr := listener.Accept()
if acceptErr != nil {
return
}
var request map[string]any
_ = json.NewDecoder(conn).Decode(&request)
body, _ := json.Marshal(payload)
_ = json.NewEncoder(conn).Encode(map[string]any{"ok": true, "action": "health", "payload": json.RawMessage(body)})
conn.Close()
}
}()
return socket
}
// 心跳说"可用"不等于执行时还是那个实例:执行前的探测必须能挡住换过的 socket。
func TestCaptureExecutorPrecheckRejectsAWrongInstance(t *testing.T) {
socket := fakeAgent(t, map[string]any{"project": "weknora071full", "source_app": "WKNFull-app"})
client := libraryagent.ClientConfig{Socket: socket, Timeout: 2 * time.Second}
err := captureExecutorPrecheck(context.Background(), client, "kgmaint-weknora", 2*time.Second)
if err == nil || !strings.Contains(err.Error(), "指向的实例不匹配") {
t.Fatalf("err = %v", err)
if !expect.Complete() {
t.Fatalf("expectation should be complete: %+v", expect)
}
if !strings.Contains(err.Error(), "weknora071full") {
t.Fatalf("err should name the actual agent project: %v", err)
}
}
func TestCaptureExecutorPrecheckAcceptsTheExpectedInstance(t *testing.T) {
socket := fakeAgent(t, map[string]any{"project": "kgmaint-weknora", "source_app": "kgmaint-app"})
client := libraryagent.ClientConfig{Socket: socket, Timeout: 2 * time.Second}
if err := captureExecutorPrecheck(context.Background(), client, "kgmaint-weknora", 2*time.Second); err != nil {
t.Fatal(err)
}
// 没有期望身份就不许停写入者:那不是"跳过检查",是不能动生产。
if err := captureExecutorPrecheck(context.Background(), client, "", 2*time.Second); err == nil ||
if _, err = captureExpectation(appconfig.Config{}); err == nil ||
!strings.Contains(err.Error(), "LIBRARY_SOURCE_PROJECT") {
t.Fatalf("empty expectation accepted: %v", err)
t.Fatalf("empty identity accepted: %v", err)
}
// 降级同样挡住。
degraded := fakeAgent(t, map[string]any{"project": "kgmaint-weknora", "degraded": "写入者恢复失败"})
if err := captureExecutorPrecheck(context.Background(), libraryagent.ClientConfig{Socket: degraded, Timeout: 2 * time.Second},
"kgmaint-weknora", 2*time.Second); err == nil || !strings.Contains(err.Error(), "降级") {
t.Fatalf("degraded executor accepted: %v", err)
// 期望不取自任何执行器应答:显式给 project 就按它绑定,配置里没有别的来源。
explicit, err := captureExpectation(appconfig.Config{LibrarySourceProject: "kgmaint-weknora"})
if err != nil || explicit.Project != "kgmaint-weknora" || explicit.SourceApp != "" {
t.Fatalf("explicit = %+v err=%v", explicit, err)
}
}
// 执行后复核:结果里自报的身份不符,这次封存就不能被登记成测试封存。
func TestCaptureExecutorPostcheckUsesTheIdentityInTheResult(t *testing.T) {
good := json.RawMessage(`{"snapshot_id":"abc","agent_identity":{"project":"kgmaint-weknora"}}`)
if err := captureExecutorPostcheck(good, "kgmaint-weknora"); err != nil {
// 事后证据防线:回显的期望指纹、project、source_app 任何一项不符都不能登记。
func TestCaptureExecutorPostcheckUsesTheEchoedExpectationFingerprint(t *testing.T) {
expect := libraryagent.NewExpectation("kgmaint-weknora", "kgmaint-app", "")
good := payloadOf(t, "kgmaint-weknora", "kgmaint-app", expect.Fingerprint)
if err := captureExecutorPostcheck(good, expect); err != nil {
t.Fatalf("matching identity rejected: %v", err)
}
swapped := json.RawMessage(`{"snapshot_id":"abc","agent_identity":{"project":"weknora071full"}}`)
err := captureExecutorPostcheck(swapped, "kgmaint-weknora")
if err == nil || !strings.Contains(err.Error(), "另一个实例") {
t.Fatalf("err = %v", err)
swapped := payloadOf(t, "weknora071full", "WKNFull-app", expect.Fingerprint)
if err := captureExecutorPostcheck(swapped, expect); err == nil || !strings.Contains(err.Error(), "另一个实例") {
t.Fatalf("wrong project accepted: %v", err)
}
if err := captureExecutorPostcheck(json.RawMessage(`{"snapshot_id":"abc"}`), "kgmaint-weknora"); err == nil ||
regenerated := payloadOf(t, "kgmaint-weknora", "kgmaint-app",
libraryagent.NewExpectation("kgmaint-weknora", "kgmaint-app", "/other/archive").Fingerprint)
if err := captureExecutorPostcheck(regenerated, expect); err == nil || !strings.Contains(err.Error(), "期望指纹不一致") {
t.Fatalf("regenerated expectation accepted: %v", err)
}
if err := captureExecutorPostcheck(json.RawMessage(`{"snapshot_id":"abc"}`), expect); err == nil ||
!strings.Contains(err.Error(), "没有执行器身份") {
t.Fatalf("missing identity accepted: %v", err)
}
if err := captureExecutorPostcheck(payloadOf(t, "kgmaint-weknora", "kgseal-app", expect.Fingerprint), expect); err == nil ||
!strings.Contains(err.Error(), "另一个源容器") {
t.Fatalf("wrong source app accepted: %v", err)
}
}
func payloadOf(t *testing.T, project, sourceApp, fingerprint string) json.RawMessage {
t.Helper()
encoded, err := json.Marshal(map[string]any{
"snapshot_id": "abc",
"agent_identity": map[string]any{
"project": project, "source_app": sourceApp, "expectation_fingerprint": fingerprint,
},
})
if err != nil {
t.Fatal(err)
}
return encoded
}
+12 -9
View File
@@ -270,14 +270,16 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st
_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", "dead letter release storage unavailable", "", "", "", time.Now().UTC())
return fmt.Errorf("load dead letter releases: %w", releaseErr)
}
// 心跳只能证明"刚才"是哪个实例:动写入者之前再核一次身份。
if identityErr := captureExecutorPrecheck(ctx, agentConfig, config.SourceProject(), 5*time.Second); identityErr != nil {
_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, identityErr.Error(), time.Now().UTC())
_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", identityErr.Error(),
"未停任何写入者;修正执行器身份后重试", "", "", time.Now().UTC())
return identityErr
// 期望执行器身份随请求一起送进 agent,由接收方在操作互斥内、任何源操作之前
// 比对:这样即使 socket 在这之前被换掉,错的执行器也一次 run 都不会跑。
expectation, expectErr := captureExpectation(config)
if expectErr != nil {
_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, expectErr.Error(), time.Now().UTC())
_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", expectErr.Error(),
"未停任何写入者;补齐来源身份配置后重试", "", "", time.Now().UTC())
return expectErr
}
result, err := libraryagent.CaptureReleased(ctx, agentConfig, payload.Scope, payload.CaptureJob, released)
result, err := libraryagent.CaptureReleased(ctx, agentConfig, payload.Scope, payload.CaptureJob, released, expectation)
finishedAt := time.Now().UTC()
if err != nil {
// 导出失败时阶段与作业都要留痕;恢复责任在 agent,不会因为作业失败而假装来源已恢复。
@@ -289,7 +291,7 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st
}
// 执行后复核:结果里 agent 自报的身份必须还是期望实例,否则这次封存
// 可能来自别的栈,绝不能登记成"本实例的测试封存"。
if identityErr := captureExecutorPostcheck(result.Payload, config.SourceProject()); identityErr != nil {
if identityErr := captureExecutorPostcheck(result.Payload, expectation); identityErr != nil {
_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, identityErr.Error(), finishedAt)
_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", identityErr.Error(),
"快照已产生但来源身份不符,未登记为测试封存;核对执行器后重新封存", "", "", finishedAt)
@@ -301,7 +303,8 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st
Scope: payload.Scope,
Kind: governance.LibraryVersionTestSealed, SnapshotID: result.SnapshotID, SourceJobID: libraryJobID,
Evidence: map[string]any{"capture_job": payload.CaptureJob, "replay": result.Replay,
"requested_by": payload.RequestedBy, "agent_result": json.RawMessage(result.Payload)},
"requested_by": payload.RequestedBy, "expectation_fingerprint": expectationFingerprint(expectation),
"agent_result": json.RawMessage(result.Payload)},
CreatedBy: payload.RequestedBy, CreatedAt: finishedAt,
})
if versionErr != nil {
+1 -1
View File
@@ -90,7 +90,7 @@ A06(内容、问答、引用图片效果、性能报告)已由用户取消
| --- | --- | --- |
| S03-b | 死信/孤儿任务阻断的负向演练(本轮只验证了真实死信阻断) | 真实阻断已发生并记录;系统化的负向矩阵未跑 |
| 排空时序(历史结论) | 旧做法"202 之后抢时序上传"在 1faf513 上重跑 6 次全部 `Broken pipe`(`drain-task-final-1faf513*.json` 保留,不记通过) | 已被 S06-a 的确定性做法取代:先证明请求已被闸门接收,再启动维护。旧证据保留,不覆盖 |
| 封存执行期身份复核(真实封存) | 代码与单测在位(agent 附身份、worker 执行前探测+执行后复核),本轮**没有**在候选栈发起真实封存(会产生新快照与停写入者窗口),因此"换 socket 被挡住"只有单测证据 |
| 封存执行期身份复核(真实封存) | 期望身份随 capture 请求送进 agent,由它在操作互斥内、任何源操作之前比对(`swap_test.go` 用两个真实 agent 的 socket 替换做负例,断言错误 agent 的 capture 调用 0 次);真实封存尚未在最终构建上跑,本轮**没有**在候选栈发起真实封存(会产生新快照与停写入者窗口),因此"换 socket 被挡住"只有单测证据 |
| 按版本恢复的真实执行(新构建) | `restore-availability` 真实返回、恢复投递路径有单测;真实恢复作业仍在旧构建上验过(R01),本轮未重跑,不把旧证据算作新构建证据 |
| U05 | 正式 8088 页面验收 | 正式栈仍是 0.1.114,本轮未替换;维护面板/逐页 Wiki 差异只在测试与候选构建里验证过 |
+5 -1
View File
@@ -102,7 +102,10 @@ type Config struct {
LibrarySourceInstance string
// LibrarySourceProject 是封存执行器应该指向的实例名(agent -project)。
// 留空时等于来源实例名:环境里两者本来就同名(kgmaint-weknora)。
LibrarySourceProject string
LibrarySourceProject string
// LibrarySourceApp 是封存源容器名(agent -source-app)。project 可能同名而源接线
// 不同,所以配了就一起绑定进"期望执行器身份"。
LibrarySourceApp string
LibraryTargetProject string
LibraryTargetWriters []string
LibraryTargetNonWriters map[string]string
@@ -210,6 +213,7 @@ func Load(getenv func(string) string) (Config, error) {
LibraryTargetInstance: strings.TrimSpace(getenv("LIBRARY_TARGET_INSTANCE")),
LibrarySourceInstance: strings.TrimSpace(getenv("LIBRARY_SOURCE_INSTANCE")),
LibrarySourceProject: strings.TrimSpace(getenv("LIBRARY_SOURCE_PROJECT")),
LibrarySourceApp: strings.TrimSpace(getenv("LIBRARY_SOURCE_APP")),
LibraryTargetProject: strings.TrimSpace(getenv("LIBRARY_TARGET_PROJECT")),
LibraryTargetWriters: splitList(getenv("LIBRARY_TARGET_WRITERS")),
LibraryTargetNonWriters: splitDeclarations(getenv("LIBRARY_TARGET_NON_WRITERS")),
+21 -4
View File
@@ -66,6 +66,9 @@ type request struct {
// Released 只允许精确死信指纹(sha256:<64 hex>)。它不改变固定命令形态:
// 还能传的只有"这些确切的行已由治理平台审计解除",CLI 会逐字节复核。
Released []string `json:"released,omitempty"`
// Expect 是调用方对执行器身份的期望。capture 必须在操作互斥内、任何源操作
// 之前比对它;缺失即拒绝(fail closed),不靠"事后核对"兜底。
Expect *Expectation `json:"expect,omitempty"`
}
type response struct {
@@ -255,6 +258,16 @@ func (a *Agent) execute(ctx context.Context, req request) response {
out.Error = "invalid job id"
return out
}
if req.Expect == nil {
out.Error = "capture 请求没有携带期望执行器身份,拒绝执行"
return out
}
// 此时已持有操作互斥(a.mu):从"比对身份"到"真正跑子进程"之间没有别的
// 请求能插进来,也不会被另一个 agent 抢走——这是零副作用的前提。
if err := a.config.matchExpectation(*req.Expect); err != nil {
out.Error = err.Error()
return out
}
args = []string{"capture", "-scope", req.Scope}
if req.Job != "" {
args = append(args, "-job", req.Job)
@@ -317,19 +330,23 @@ func (a *Agent) execute(ctx context.Context, req request) response {
}
// 封存结果必须带上"这次是谁执行的":调用方据此复核,防止心跳与执行之间
// socket 被换成另一个 agent(真正的封错库事故就是这样发生的)。
if req.Action == "capture" {
out.Payload = a.attachIdentity(out.Payload)
if req.Action == "capture" && req.Expect != nil {
out.Payload = a.attachIdentity(out.Payload, *req.Expect)
}
return out
}
// attachIdentity 把执行器身份并进封存结果(保留 CLI 原有的字段,如 snapshot_id)。
func (a *Agent) attachIdentity(payload json.RawMessage) json.RawMessage {
// expectation_fingerprint 是"本执行器实际核对过的那份期望"的指纹:调用方据此确认
// 执行时看到的期望与它发出的一致,而不是被替换后重新生成的。
func (a *Agent) attachIdentity(payload json.RawMessage, expectation Expectation) json.RawMessage {
var document map[string]any
if err := json.Unmarshal(payload, &document); err != nil {
return payload
}
document["agent_identity"] = a.config.identity()
identity := a.config.identity()
identity["expectation_fingerprint"] = expectation.Fingerprint
document["agent_identity"] = identity
encoded, err := json.Marshal(document)
if err != nil {
return payload
+20 -7
View File
@@ -56,6 +56,19 @@ func testAgent(t *testing.T) (*Agent, string, func() []string) {
}
}
// captureExpect 是测试 agent 自己的身份(weknora071full / WKNFull-app),
// 现在 capture 请求必须携带期望身份,缺了就会被拒。
func captureExpect(t *testing.T) string {
t.Helper()
// 只绑定 project:各用例的 source_app 不同,绑定字段的专门用例在 expectation_test.go。
expectation := NewExpectation("weknora071full", "", "")
encoded, err := json.Marshal(expectation)
if err != nil {
t.Fatal(err)
}
return string(encoded)
}
func roundTrip(t *testing.T, socket string, payload string) response {
t.Helper()
conn, err := net.Dial("unix", socket)
@@ -105,7 +118,7 @@ func TestAgentSocketIsPrivateAndPinsTheDeclaration(t *testing.T) {
if dir.Mode().Perm() != 0700 {
t.Fatalf("socket directory mode %v", dir.Mode().Perm())
}
out := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-internal-test"}`)
out := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-internal-test","expect":`+captureExpect(t)+`}`)
if !out.OK || out.Job != "lib-internal-test" {
t.Fatalf("capture rejected: %+v", out)
}
@@ -234,7 +247,7 @@ func TestAgentRecoversWritersAfterFailedCapture(t *testing.T) {
defer cancel()
go func() { _ = agent.Serve(ctx) }()
waitSocket(t, socket)
out := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-fail-1"}`)
out := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-fail-1","expect":`+captureExpect(t)+`}`)
if out.OK {
t.Fatalf("failed capture reported success: %+v", out)
}
@@ -255,12 +268,12 @@ func TestAgentBlocksCapturesWhenRecoveryFails(t *testing.T) {
go func() { _ = agent.Serve(ctx) }()
waitSocket(t, socket)
setRecoverErr(errors.New("docker restart failed"))
out := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-fail-2"}`)
out := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-fail-2","expect":`+captureExpect(t)+`}`)
if out.OK || !strings.Contains(out.Error, "写入者恢复失败") || !strings.Contains(out.Error, "已阻断后续封存") {
t.Fatalf("recovery failure not surfaced: %+v", out)
}
before := len(captureCalls(calls()))
blocked := roundTrip(t, socket, `{"action":"capture","scope":"external","job":"lib-fail-3"}`)
blocked := roundTrip(t, socket, `{"action":"capture","scope":"external","job":"lib-fail-3","expect":`+captureExpect(t)+`}`)
if blocked.OK || !strings.Contains(blocked.Error, "降级") {
t.Fatalf("degraded agent accepted a capture: %+v", blocked)
}
@@ -272,7 +285,7 @@ func TestAgentBlocksCapturesWhenRecoveryFails(t *testing.T) {
if recovered := roundTrip(t, socket, `{"action":"recover"}`); !recovered.OK {
t.Fatalf("explicit recovery failed: %+v", recovered)
}
if resumed := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-fail-4"}`); strings.Contains(resumed.Error, "降级") {
if resumed := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-fail-4","expect":`+captureExpect(t)+`}`); strings.Contains(resumed.Error, "降级") {
t.Fatalf("recovery did not clear the degraded state: %+v", resumed)
}
}
@@ -396,7 +409,7 @@ func TestAgentForwardsOnlyExactDeadLetterFingerprints(t *testing.T) {
time.Sleep(10 * time.Millisecond)
}
fingerprint := "sha256:" + strings.Repeat("a", 64)
ok := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"j1","released":["`+fingerprint+`"]}`)
ok := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"j1","released":["`+fingerprint+`"],"expect":`+captureExpect(t)+`}`)
if !ok.OK {
t.Fatalf("exact fingerprint refused: %s", ok.Error)
}
@@ -404,7 +417,7 @@ func TestAgentForwardsOnlyExactDeadLetterFingerprints(t *testing.T) {
if !strings.Contains(last, "-released-dead-letters "+fingerprint) {
t.Fatalf("fingerprint not forwarded: %s", last)
}
bad := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"j2","released":["*"]}`)
bad := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"j2","released":["*"],"expect":`+captureExpect(t)+`}`)
if bad.OK || !strings.Contains(bad.Error, "fingerprints") {
t.Fatalf("wildcard release accepted: %+v", bad)
}
+4 -4
View File
@@ -28,13 +28,13 @@ type Result struct {
// Capture asks the agent to seal one library scope. The job id makes the call
// idempotent: repeating it returns the already sealed candidate.
func Capture(ctx context.Context, config ClientConfig, scope, job string) (Result, error) {
return CaptureReleased(ctx, config, scope, job, nil)
func Capture(ctx context.Context, config ClientConfig, scope, job string, expect Expectation) (Result, error) {
return CaptureReleased(ctx, config, scope, job, nil, expect)
}
// CaptureReleased 在 Capture 之上带上治理平台已审计的精确死信解除指纹。
func CaptureReleased(ctx context.Context, config ClientConfig, scope, job string, released []string) (Result, error) {
request := map[string]any{"action": "capture", "scope": scope, "job": job}
func CaptureReleased(ctx context.Context, config ClientConfig, scope, job string, released []string, expect Expectation) (Result, error) {
request := map[string]any{"action": "capture", "scope": scope, "job": job, "expect": expect}
if len(released) > 0 {
request["released"] = released
}
+72
View File
@@ -0,0 +1,72 @@
package libraryagent
import (
"crypto/sha256"
"encoding/hex"
"errors"
"strings"
)
// Expectation 是调用方对"这一条请求该由哪个执行器执行"的声明。
//
// 为什么是调用方声明而不是执行器自报:自报值只能证明"我是谁",不能证明"你找的
// 就是我"。心跳之后、真正停写入者之前,socket 可能被换成另一个 agent;只有把
// 期望随**同一个请求**送进去、由接收方在操作互斥内、执行任何源操作之前比对,
// 才能让"换过 socket"变成零副作用(错的那个 noche 一次 run 都不调)。
type Expectation struct {
Project string `json:"project"`
SourceApp string `json:"source_app,omitempty"`
ArchiveRoot string `json:"archive_root,omitempty"`
// Fingerprint 覆盖上面三项,供接收方回显、调用方事后核对:
// 证明"执行器看到的就是我发的那份期望",而不是被替换后重新生成的。
Fingerprint string `json:"fingerprint"`
}
// NewExpectation 由治理侧配置构造期望(绝不从执行器应答里取字段)。
func NewExpectation(project, sourceApp, archiveRoot string) Expectation {
expectation := Expectation{
Project: strings.TrimSpace(project),
SourceApp: strings.TrimSpace(sourceApp),
ArchiveRoot: strings.TrimSpace(archiveRoot),
}
expectation.Fingerprint = expectation.fingerprint()
return expectation
}
func (e Expectation) fingerprint() string {
sum := sha256.Sum256([]byte(e.Project + "\x00" + e.SourceApp + "\x00" + e.ArchiveRoot))
return "sha256:" + hex.EncodeToString(sum[:])
}
// Complete 表示期望足够具体:没有 project 的期望无法核对身份,必须 fail closed;
// 指纹与字段不一致说明请求在半路被改过。
func (e Expectation) Complete() bool {
return e.Project != "" && e.Fingerprint == e.fingerprint()
}
// matchExpectation 在操作互斥内比对"本执行器配置"与"调用方期望"。
// 只比较期望里给出的字段:project 必填,source_app/archive_root 给了就必须一致。
func (c Config) matchExpectation(expectation Expectation) error {
if !expectation.Complete() {
return errors.New("capture 请求缺少完整的期望执行器身份,拒绝执行")
}
if c.Project != expectation.Project {
return errors.New("执行器与请求期望的实例不一致:" +
"agent=" + quoteOrEmpty(c.Project) + ",期望=" + quoteOrEmpty(expectation.Project) + ",拒绝执行")
}
if expectation.SourceApp != "" && c.SourceApp != expectation.SourceApp {
return errors.New("执行器与请求期望的源容器不一致:" +
"agent=" + quoteOrEmpty(c.SourceApp) + ",期望=" + quoteOrEmpty(expectation.SourceApp) + ",拒绝执行")
}
if expectation.ArchiveRoot != "" && c.ArchiveRoot != expectation.ArchiveRoot {
return errors.New("执行器与请求期望的归档根不一致,拒绝执行")
}
return nil
}
func quoteOrEmpty(value string) string {
if value == "" {
return `""`
}
return `"` + value + `"`
}
+2 -1
View File
@@ -89,7 +89,8 @@ func TestCaptureResultCarriesTheAgentIdentity(t *testing.T) {
serveAgent(t, agent, socket)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
result, err := Capture(ctx, ClientConfig{Socket: socket, Timeout: 5 * time.Second}, "internal", "lib-internal-identity")
result, err := Capture(ctx, ClientConfig{Socket: socket, Timeout: 5 * time.Second}, "internal",
"lib-internal-identity", NewExpectation("weknora071full", "WKNFull-app", ""))
if err != nil {
t.Fatal(err)
}
+149
View File
@@ -0,0 +1,149 @@
package libraryagent
import (
"context"
"os"
"strings"
"sync"
"testing"
"time"
)
// unix socket 路径有长度上限,测试目录要用短的(TMPDIR 太深)。
func shortSocketDir(t *testing.T) string {
t.Helper()
dir, err := os.MkdirTemp("/tmp", "kgswap-")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
return dir
}
// 两个真实 agent、两个 socket:负例把调用方指到"另一个实例"的 socket 上,
// 断言**错误 agent 一次 capture 都没跑**(不是跑完再拒),正例走对 socket 正常封存。
func TestCaptureRequestIsRefusedBeforeAnySourceOperationWhenSocketIsSwapped(t *testing.T) {
rightAgent, rightSocket, rightCalls := agentWithProject(t, "kgmaint-weknora", "kgmaint-app")
wrongAgent, wrongSocket, wrongCalls := agentWithProject(t, "weknora071full", "WKNFull-app")
serveAgent(t, rightAgent, rightSocket)
serveAgent(t, wrongAgent, wrongSocket)
expectation := NewExpectation("kgmaint-weknora", "kgmaint-app", "")
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
// 负例:调用方手上的 socket 已被换成另一个实例的执行器。
result, err := Capture(ctx, ClientConfig{Socket: wrongSocket, Timeout: 5 * time.Second},
"internal", "lib-swap", expectation)
if err == nil || result.OK {
t.Fatalf("capture against the wrong executor was accepted: %+v", result)
}
if !strings.Contains(result.Error, "期望") {
t.Fatalf("refusal should name the mismatch: %q", result.Error)
}
if calls := captureCalls(wrongCalls()); len(calls) != 0 {
t.Fatalf("wrong executor ran the capture binary %d time(s): %v", len(calls), calls)
}
// 旁证:错误的执行器进程本身仍然可用(health 正常),拒绝来自身份比对而不是不可达。
health, err := Health(ctx, ClientConfig{Socket: wrongSocket, Timeout: 5 * time.Second})
if err != nil || !health.OK {
t.Fatalf("wrong executor should still answer health: %v %+v", err, health)
}
// 正例:同一份期望打到正确的 socket,正常封存一次。
result, err = Capture(ctx, ClientConfig{Socket: rightSocket, Timeout: 5 * time.Second},
"internal", "lib-swap", expectation)
if err != nil || !result.OK {
t.Fatalf("capture against the right executor failed: %v %+v", err, result)
}
if calls := captureCalls(rightCalls()); len(calls) != 1 {
t.Fatalf("right executor ran %d time(s): %v", len(calls), calls)
}
if calls := captureCalls(wrongCalls()); len(calls) != 0 {
t.Fatalf("wrong executor ran after the positive case: %v", calls)
}
}
// 缺期望、期望指纹对不上、期望里 source_app 不同,都必须 fail closed 且不跑子进程。
func TestCaptureFailsClosedWithoutACompleteExpectation(t *testing.T) {
agent, socket, calls := agentWithProject(t, "kgmaint-weknora", "kgmaint-app")
serveAgent(t, agent, socket)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
tampered := NewExpectation("kgmaint-weknora", "kgmaint-app", "")
tampered.Project = "someone-else" // 字段改了、指纹没改:请求在半路被改过
cases := []struct {
name string
expectation Expectation
text string
}{
{name: "missing expectation", expectation: Expectation{}, text: "缺少完整的期望执行器身份"},
{name: "tampered expectation", expectation: tampered, text: "缺少完整的期望执行器身份"},
{name: "wrong source app", expectation: NewExpectation("kgmaint-weknora", "WKNFull-app", ""), text: "源容器不一致"},
}
for _, test := range cases {
result, err := Capture(ctx, ClientConfig{Socket: socket, Timeout: 5 * time.Second},
"internal", "lib-"+strings.ReplaceAll(test.name, " ", "-"), test.expectation)
if err == nil || result.OK {
t.Fatalf("%s: accepted %+v", test.name, result)
}
if !strings.Contains(result.Error, test.text) {
t.Fatalf("%s: error %q", test.name, result.Error)
}
}
if calls := captureCalls(calls()); len(calls) != 0 {
t.Fatalf("capture binary ran despite a refused identity: %v", calls)
}
}
// 恢复路径不受期望约束:它只把写入者恢复回来,任何身份都必须能执行。
func TestRecoverStaysAvailableRegardlessOfExpectation(t *testing.T) {
agent, socket, calls := agentWithProject(t, "kgmaint-weknora", "kgmaint-app")
serveAgent(t, agent, socket)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
result, err := Recover(ctx, ClientConfig{Socket: socket, Timeout: 5 * time.Second})
if err != nil || !result.OK {
t.Fatalf("recover refused: %v %+v", err, result)
}
if len(recoverCalls(calls())) == 0 {
t.Fatal("recover did not run the recovery command")
}
}
// agentWithProject 造一个真实 agent(不跑真命令),返回它自己的调用记录。
func agentWithProject(t *testing.T, project, sourceApp string) (*Agent, string, func() []string) {
t.Helper()
root := shortSocketDir(t)
var mu sync.Mutex
var calls []string
config := Config{
Socket: root + "/agent.sock",
Binary: "/bin/echo",
ArchiveRoot: root + "/library",
JournalRoot: root + "/freeze",
SourceApp: sourceApp,
SourcePG: "pg",
SourceES: "es",
Project: project,
Writers: "app",
NonWriters: "frontend=UI only",
MaxDuration: time.Minute,
Run: func(_ context.Context, name string, args ...string) ([]byte, error) {
mu.Lock()
calls = append(calls, name+" "+strings.Join(args, " "))
mu.Unlock()
return []byte(`{"snapshot_id":"abc","state":"succeeded"}`), nil
},
}
agent, err := New(config)
if err != nil {
t.Fatal(err)
}
return agent, config.Socket, func() []string {
mu.Lock()
defer mu.Unlock()
return append([]string(nil), calls...)
}
}