diff --git a/CHANGELOG.md b/CHANGELOG.md index ed9693e..f5146fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 秒）控制执行器观测的有效期； 未设置不等于"永不过期"，过期即"无法确认"。 - 页面/接口不再出现写死的"封存通道未接线""生产恢复未接线"：不能发起时给出真实原因、 diff --git a/cmd/worker/library_capture_identity.go b/cmd/worker/library_capture_identity.go index 7110ecc..11b450f 100644 --- a/cmd/worker/library_capture_identity.go +++ b/cmd/worker/library_capture_identity.go @@ -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) } diff --git a/cmd/worker/library_capture_identity_test.go b/cmd/worker/library_capture_identity_test.go index fcf21bd..f3a9917 100644 --- a/cmd/worker/library_capture_identity_test.go +++ b/cmd/worker/library_capture_identity_test.go @@ -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 } diff --git a/cmd/worker/main.go b/cmd/worker/main.go index d0d6f2c..e72aaf2 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -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 { diff --git a/docs/library-acceptance-matrix.md b/docs/library-acceptance-matrix.md index 38703f9..0bc50ee 100644 --- a/docs/library-acceptance-matrix.md +++ b/docs/library-acceptance-matrix.md @@ -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 差异只在测试与候选构建里验证过 | diff --git a/internal/appconfig/config.go b/internal/appconfig/config.go index 19ef36d..48d0f44 100644 --- a/internal/appconfig/config.go +++ b/internal/appconfig/config.go @@ -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")), diff --git a/internal/libraryagent/agent.go b/internal/libraryagent/agent.go index 65fd3fc..f9a1051 100644 --- a/internal/libraryagent/agent.go +++ b/internal/libraryagent/agent.go @@ -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 diff --git a/internal/libraryagent/agent_test.go b/internal/libraryagent/agent_test.go index 15e62d5..747069a 100644 --- a/internal/libraryagent/agent_test.go +++ b/internal/libraryagent/agent_test.go @@ -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) } diff --git a/internal/libraryagent/client.go b/internal/libraryagent/client.go index 8822718..b956275 100644 --- a/internal/libraryagent/client.go +++ b/internal/libraryagent/client.go @@ -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 } diff --git a/internal/libraryagent/expectation.go b/internal/libraryagent/expectation.go new file mode 100644 index 0000000..dcb546f --- /dev/null +++ b/internal/libraryagent/expectation.go @@ -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 + `"` +} diff --git a/internal/libraryagent/health_test.go b/internal/libraryagent/health_test.go index d682105..9b06abe 100644 --- a/internal/libraryagent/health_test.go +++ b/internal/libraryagent/health_test.go @@ -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) } diff --git a/internal/libraryagent/swap_test.go b/internal/libraryagent/swap_test.go new file mode 100644 index 0000000..070afd7 --- /dev/null +++ b/internal/libraryagent/swap_test.go @@ -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...) + } +}