diff --git a/deploy/postgres/goose/00034_2026-09-14-workspace-lease-fencing.sql b/deploy/postgres/goose/00034_2026-09-14-workspace-lease-fencing.sql new file mode 100644 index 00000000..d52f3ecf --- /dev/null +++ b/deploy/postgres/goose/00034_2026-09-14-workspace-lease-fencing.sql @@ -0,0 +1,8 @@ +-- +goose Up +ALTER TABLE document_records + ADD COLUMN IF NOT EXISTS lease_fencing_token TEXT NOT NULL DEFAULT '', + ADD COLUMN IF NOT EXISTS workspace_schema_version INT NOT NULL DEFAULT 1; + +-- +goose Down +ALTER TABLE document_records DROP COLUMN IF EXISTS workspace_schema_version; +ALTER TABLE document_records DROP COLUMN IF EXISTS lease_fencing_token; diff --git a/internal/docgov/lease.go b/internal/docgov/lease.go index 1dd0c1e4..08f3d691 100644 --- a/internal/docgov/lease.go +++ b/internal/docgov/lease.go @@ -2,8 +2,11 @@ package docgov import ( "context" + "crypto/rand" + "encoding/hex" "encoding/json" "fmt" + "io" "path/filepath" "strings" "time" @@ -29,6 +32,7 @@ type ContentSaveRequest struct { ActorName string MarkdownBody string ExpectedSHA256 string // 乐观并发：非空时必须匹配当前 ContentSHA256 + FencingToken string } // ContentSaveResult 正文保存结果 @@ -101,6 +105,7 @@ func (e *GovernanceEngine) AcquireLease(ctx context.Context, req LeaseAcquireReq } doc.LeaseExpiresAt = &expires doc.LeaseHeartbeatAt = &now + doc.LeaseFencingToken = newLeaseFencingToken() doc.UpdatedAt = now e.persistLeaseLocked(ctx, doc) @@ -344,6 +349,11 @@ func (e *GovernanceEngine) SaveDocumentContent(ctx context.Context, req ContentS if req.ActorID <= 0 { holderID = "human_anon" } + if req.FencingToken != "" { + if err := e.AssertLeaseFencing(ctx, req.DocID, req.FencingToken); err != nil { + return nil, err + } + } doc, err := e.AcquireLease(ctx, LeaseAcquireRequest{ DocID: req.DocID, @@ -409,6 +419,28 @@ func (e *GovernanceEngine) SaveDocumentContent(ctx context.Context, req ContentS return &ContentSaveResult{Doc: doc, ContentSHA256: shaStr, WorkDir: workDir}, nil } +// AssertLeaseFencing rejects writers holding a superseded token. +func (e *GovernanceEngine) AssertLeaseFencing(ctx context.Context, docID, token string) error { + e.mu.Lock() + defer e.mu.Unlock() + doc, err := e.getDocLocked(ctx, docID) + if err != nil { + return err + } + if token == "" || doc.LeaseFencingToken == "" || token != doc.LeaseFencingToken { + return ErrStaleFencing + } + return nil +} + +func newLeaseFencingToken() string { + var b [16]byte + if _, err := io.ReadFull(rand.Reader, b[:]); err != nil { + return fmt.Sprintf("%d", time.Now().UnixNano()) + } + return hex.EncodeToString(b[:]) +} + func (e *GovernanceEngine) getDocLocked(ctx context.Context, docID string) (*DocumentRecord, error) { doc, exists := e.memDocs[docID] if !exists { @@ -428,10 +460,10 @@ func (e *GovernanceEngine) persistLeaseLocked(ctx context.Context, doc *Document _, _ = e.pool.Exec(ctx, ` UPDATE document_records SET lease_holder_type = $1, lease_holder_id = $2, lease_work_dir = $3, - lease_expires_at = $4, lease_heartbeat_at = $5, updated_at = $6 - WHERE doc_id = $7 + lease_expires_at = $4, lease_heartbeat_at = $5, lease_fencing_token = $6, updated_at = $7 + WHERE doc_id = $8 `, string(doc.LeaseHolderType), doc.LeaseHolderID, doc.LeaseWorkDir, - doc.LeaseExpiresAt, doc.LeaseHeartbeatAt, doc.UpdatedAt, doc.DocID) + doc.LeaseExpiresAt, doc.LeaseHeartbeatAt, doc.LeaseFencingToken, doc.UpdatedAt, doc.DocID) } func (e *GovernanceEngine) appendAuditLocked(doc *DocumentRecord, action string, actorID int64, actorName string, actorRole UserRole, details map[string]any) { diff --git a/internal/docgov/lease_test.go b/internal/docgov/lease_test.go index 6e1d4deb..4e53e92f 100644 --- a/internal/docgov/lease_test.go +++ b/internal/docgov/lease_test.go @@ -86,6 +86,38 @@ func TestSameHolderRenewLease(t *testing.T) { } } +func TestStaleFencingRejectedAfterRenew(t *testing.T) { + engine := docgov.NewGovernanceEngine(nil) + ctx := context.Background() + docID := "doc_fence_001" + _, _ = engine.CreateOrGetDocument(ctx, docID, 1, "intake", "摸底", "V0.1", "", 1, "a", 2, "b", 3, "c") + first, err := engine.AcquireLease(ctx, docgov.LeaseAcquireRequest{ + DocID: docID, HolderType: docgov.LeaseHolderHuman, HolderID: "1", TTL: time.Minute, + }) + if err != nil { + t.Fatal(err) + } + old := first.LeaseFencingToken + if old == "" { + t.Fatal("expected fencing token") + } + second, err := engine.AcquireLease(ctx, docgov.LeaseAcquireRequest{ + DocID: docID, HolderType: docgov.LeaseHolderHuman, HolderID: "1", TTL: time.Minute, + }) + if err != nil { + t.Fatal(err) + } + if second.LeaseFencingToken == old { + t.Fatal("renew must rotate fencing token") + } + if err := engine.AssertLeaseFencing(ctx, docID, old); err != docgov.ErrStaleFencing { + t.Fatalf("old writer: %v", err) + } + if err := engine.AssertLeaseFencing(ctx, docID, second.LeaseFencingToken); err != nil { + t.Fatal(err) + } +} + func TestAssertHumanCanEdit_ChecksDutyBeforeLease(t *testing.T) { engine := docgov.NewGovernanceEngine(nil) ctx := context.Background() diff --git a/internal/docgov/types.go b/internal/docgov/types.go index b06f9791..01ea18aa 100644 --- a/internal/docgov/types.go +++ b/internal/docgov/types.go @@ -18,6 +18,7 @@ var ( ErrProposalAlreadyDecided = errors.New("docgov: change proposal has already been decided") ErrChangeProposalPending = errors.New("docgov: document is claimed and has pending change proposal waiting for author approval") ErrDocumentBusy = errors.New("docgov: document edit lease is held by another writer") + ErrStaleFencing = errors.New("docgov: stale lease fencing token") ErrLeaseExpired = errors.New("docgov: document edit lease has expired") ErrLeaseNotHeld = errors.New("docgov: caller does not hold the document edit lease") ErrForceEditNotAllowed = errors.New("docgov: force_direct_edit requires an approved_by_author proposal and unlocked drafting document") @@ -183,11 +184,12 @@ type DocumentRecord struct { ApprovedAt *time.Time `json:"approved_at"` // 编辑租约 (运行时互斥，正交于 IsLocked) - LeaseHolderType LeaseHolderType `json:"lease_holder_type,omitempty"` - LeaseHolderID string `json:"lease_holder_id,omitempty"` - LeaseWorkDir string `json:"lease_work_dir,omitempty"` - LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"` - LeaseHeartbeatAt *time.Time `json:"lease_heartbeat_at,omitempty"` + LeaseHolderType LeaseHolderType `json:"lease_holder_type,omitempty"` + LeaseHolderID string `json:"lease_holder_id,omitempty"` + LeaseWorkDir string `json:"lease_work_dir,omitempty"` + LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"` + LeaseHeartbeatAt *time.Time `json:"lease_heartbeat_at,omitempty"` + LeaseFencingToken string `json:"lease_fencing_token,omitempty"` Config DocumentConfig `json:"config"` CreatedAt time.Time `json:"created_at"` @@ -212,6 +214,7 @@ func (d *DocumentRecord) ClearLease() { d.LeaseWorkDir = "" d.LeaseExpiresAt = nil d.LeaseHeartbeatAt = nil + d.LeaseFencingToken = "" } // CommentType 批注类型 diff --git a/internal/squad/workspace_adapter.go b/internal/squad/workspace_adapter.go new file mode 100644 index 00000000..5120bf35 --- /dev/null +++ b/internal/squad/workspace_adapter.go @@ -0,0 +1,15 @@ +package squad + +import "chat-agent-system/internal/workspace" + +// WorkspaceRefForRun is the only squad→workspace DTO. Physical paths stay inside workspace. +func WorkspaceRefForRun(runID string, projectID int64, env, token string) workspace.WorkspaceRef { + return workspace.WorkspaceRef{ + ID: "run-" + runID, + SchemaVersion: workspace.SchemaVersion, + FencingToken: token, + ProjectScope: projectID, + RunID: runID, + Environment: env, + } +} diff --git a/internal/workspace/capability.go b/internal/workspace/capability.go new file mode 100644 index 00000000..ad2750c5 --- /dev/null +++ b/internal/workspace/capability.go @@ -0,0 +1,219 @@ +package workspace + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "syscall" + + "chat-agent-system/internal/app/uow" +) + +// Capability is the only module allowed to turn a WorkspaceRef into bytes on disk. +type Capability struct { + dataRoot string + leases *LeaseTable +} + +func NewCapability(dataRoot string, leases *LeaseTable) *Capability { + if leases == nil { + leases = NewLeaseTable() + } + return &Capability{dataRoot: dataRoot, leases: leases} +} + +func (c *Capability) Issue(ref WorkspaceRef) (WorkspaceRef, error) { + if err := c.validateRefMeta(ref); err != nil { + return WorkspaceRef{}, err + } + token, err := newFencingToken() + if err != nil { + return WorkspaceRef{}, err + } + issued := ref + issued.SchemaVersion = SchemaVersion + issued.FencingToken = token + c.leases.Issue(issued) + return issued, nil +} + +func (c *Capability) ReadArtifact(ctx context.Context, ref WorkspaceRef, key string) ([]byte, error) { + if err := c.guard(ctx, ref, false); err != nil { + return nil, err + } + path, err := c.resolveKey(ref, key) + if err != nil { + return nil, err + } + resolved, err := c.enforceRealpath(path) + if err != nil { + return nil, err + } + return os.ReadFile(resolved) +} + +func (c *Capability) WriteArtifact(ctx context.Context, ref WorkspaceRef, key string, body []byte) error { + if err := c.guard(ctx, ref, true); err != nil { + return err + } + path, err := c.resolveKey(ref, key) + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + tmp := path + ".tmp-" + ref.FencingToken[:8] + if err := os.WriteFile(tmp, body, 0o644); err != nil { + return err + } + if _, err := c.enforceRealpath(tmp); err != nil { + _ = os.Remove(tmp) + return err + } + if err := os.Rename(tmp, path); err != nil { + _ = os.Remove(tmp) + return err + } + return nil +} + +func (c *Capability) CommitProjection(ctx context.Context, ref WorkspaceRef, key, digest string) error { + if err := c.guard(ctx, ref, true); err != nil { + return err + } + body, err := c.ReadArtifact(ctx, ref, key) + if err != nil { + return err + } + sum := sha256.Sum256(body) + got := hex.EncodeToString(sum[:]) + if digest != "" && digest != got { + return fmt.Errorf("workspace: digest mismatch") + } + marker := []byte(got + "\ncomplete\n") + return c.WriteArtifact(ctx, ref, key+".complete", marker) +} + +func (c *Capability) guard(ctx context.Context, ref WorkspaceRef, write bool) error { + if ctx != nil { + if _, ok := uow.ScopeFrom(ctx); ok { + return ErrIOInTransaction + } + } + if err := c.validateRefMeta(ref); err != nil { + return err + } + if write { + return c.leases.Assert(ref) + } + return nil +} + +func (c *Capability) validateRefMeta(ref WorkspaceRef) error { + if strings.TrimSpace(ref.ID) == "" { + return ErrInvalidKey + } + if ref.SchemaVersion != 0 && ref.SchemaVersion != SchemaVersion { + return ErrUnknownSchema + } + if ref.ProjectScope < 0 { + return ErrCrossProject + } + return nil +} + +func (c *Capability) resolveKey(ref WorkspaceRef, key string) (string, error) { + if err := validateArtifactKey(key); err != nil { + return "", err + } + root := filepath.Join(c.dataRoot, "workspaces", ref.ID) + joined := filepath.Join(root, filepath.FromSlash(key)) + absRoot, err := filepath.Abs(root) + if err != nil { + return "", err + } + absJoin, err := filepath.Abs(joined) + if err != nil { + return "", err + } + rel, err := filepath.Rel(absRoot, absJoin) + if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) { + return "", ErrPathEscape + } + return absJoin, nil +} + +func (c *Capability) enforceRealpath(path string) (string, error) { + info, err := os.Lstat(path) + if err != nil { + if os.IsNotExist(err) { + return path, nil + } + return "", err + } + if info.Mode()&os.ModeSymlink != 0 { + return "", ErrPathEscape + } + if info.Mode()&os.ModeNamedPipe != 0 { + return "", ErrPathEscape + } + resolved, err := filepath.EvalSymlinks(filepath.Dir(path)) + if err != nil { + return "", err + } + full := filepath.Join(resolved, filepath.Base(path)) + root, err := filepath.Abs(c.dataRoot) + if err != nil { + return "", err + } + rel, err := filepath.Rel(root, full) + if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) { + return "", ErrPathEscape + } + if st, ok := info.Sys().(*syscall.Stat_t); ok && st.Nlink > 1 && !info.IsDir() { + // Hardlink to an inode that may live outside: still require the named path stay in root. + if !strings.HasPrefix(full, root) { + return "", ErrPathEscape + } + } + return full, nil +} + +func validateArtifactKey(key string) error { + key = strings.TrimSpace(key) + if key == "" { + return ErrInvalidKey + } + if strings.ContainsRune(key, 0) { + return ErrInvalidKey + } + if filepath.IsAbs(key) || strings.HasPrefix(key, "/") || strings.HasPrefix(key, `\`) { + return ErrInvalidKey + } + lower := strings.ToLower(key) + if strings.Contains(lower, "%2e%2e") || strings.Contains(lower, "%2f") || strings.Contains(lower, "%5c") { + return ErrInvalidKey + } + cleaned := filepath.ToSlash(key) + for _, part := range strings.Split(cleaned, "/") { + if part == "" || part == "." || part == ".." { + return ErrInvalidKey + } + } + return nil +} + +func newFencingToken() (string, error) { + var b [16]byte + if _, err := io.ReadFull(rand.Reader, b[:]); err != nil { + return "", err + } + return hex.EncodeToString(b[:]), nil +} diff --git a/internal/workspace/capability_test.go b/internal/workspace/capability_test.go new file mode 100644 index 00000000..803a6d52 --- /dev/null +++ b/internal/workspace/capability_test.go @@ -0,0 +1,119 @@ +package workspace + +import ( + "context" + "os" + "path/filepath" + "testing" + + "chat-agent-system/internal/app/uow" +) + +func issued(t *testing.T, cap *Capability, id string, project int64) WorkspaceRef { + t.Helper() + ref, err := cap.Issue(WorkspaceRef{ID: id, ProjectScope: project, SchemaVersion: SchemaVersion}) + if err != nil { + t.Fatal(err) + } + return ref +} + +func TestWorkspaceRefHasNoPhysicalPathFields(t *testing.T) { + var ref WorkspaceRef + _ = ref.ID + _ = ref.SchemaVersion + _ = ref.FencingToken + _ = ref.ProjectScope + _ = ref.RunID + _ = ref.Environment +} + +func TestCapabilityRejectsBadKeys(t *testing.T) { + cap := NewCapability(t.TempDir(), nil) + ref := issued(t, cap, "ws1", 7) + ctx := context.Background() + for _, key := range []string{"", "../etc/passwd", "/etc/passwd", "..", "a/../b", "a//b", "%2e%2e/secret", "ok/%2fetc"} { + if _, err := cap.ReadArtifact(ctx, ref, key); err == nil { + t.Fatalf("key %q should reject", key) + } + } +} + +func TestCapabilityRejectsSymlinkEscape(t *testing.T) { + root := t.TempDir() + cap := NewCapability(root, nil) + ref := issued(t, cap, "ws1", 7) + outside := filepath.Join(t.TempDir(), "secret") + if err := os.WriteFile(outside, []byte("nope"), 0o644); err != nil { + t.Fatal(err) + } + wsDir := filepath.Join(root, "workspaces", "ws1") + if err := os.MkdirAll(wsDir, 0o755); err != nil { + t.Fatal(err) + } + link := filepath.Join(wsDir, "leak") + if err := os.Symlink(outside, link); err != nil { + t.Fatal(err) + } + if _, err := cap.ReadArtifact(context.Background(), ref, "leak"); err != ErrPathEscape { + t.Fatalf("symlink: %v", err) + } +} + +func TestStaleFencingRejected(t *testing.T) { + cap := NewCapability(t.TempDir(), nil) + old := issued(t, cap, "ws1", 3) + _ = issued(t, cap, "ws1", 3) + ctx := context.Background() + if err := cap.WriteArtifact(ctx, old, "output.md", []byte("stale")); err != ErrStaleFencing { + t.Fatalf("old writer: %v", err) + } + fresh := issued(t, cap, "ws1", 3) + if err := cap.WriteArtifact(ctx, fresh, "output.md", []byte("ok")); err != nil { + t.Fatal(err) + } +} + +func TestUnknownSchemaRejected(t *testing.T) { + cap := NewCapability(t.TempDir(), nil) + _, err := cap.Issue(WorkspaceRef{ID: "ws1", SchemaVersion: 99}) + if err != ErrUnknownSchema { + t.Fatalf("got %v", err) + } +} + +func TestNoFileIOInsideUnitOfWork(t *testing.T) { + cap := NewCapability(t.TempDir(), nil) + ref := issued(t, cap, "ws1", 1) + ctx := uow.WithScope(context.Background(), stubScope{}) + if err := cap.WriteArtifact(ctx, ref, "output.md", []byte("x")); err != ErrIOInTransaction { + t.Fatalf("got %v", err) + } +} + +func TestCommitProjectionCompleteMarker(t *testing.T) { + cap := NewCapability(t.TempDir(), nil) + ref := issued(t, cap, "ws1", 1) + ctx := context.Background() + if err := cap.WriteArtifact(ctx, ref, "output.md", []byte("hello")); err != nil { + t.Fatal(err) + } + if err := cap.CommitProjection(ctx, ref, "output.md", ""); err != nil { + t.Fatal(err) + } + got, err := cap.ReadArtifact(ctx, ref, "output.md.complete") + if err != nil { + t.Fatal(err) + } + if !containsComplete(got) { + t.Fatalf("marker %q", got) + } +} + +func containsComplete(b []byte) bool { + return string(b) != "" && (string(b)[len(b)-1] == '\n') +} + +type stubScope struct{} + +func (stubScope) Context() context.Context { return context.Background() } diff --git a/internal/workspace/lease.go b/internal/workspace/lease.go new file mode 100644 index 00000000..23cee70e --- /dev/null +++ b/internal/workspace/lease.go @@ -0,0 +1,35 @@ +package workspace + +import "sync" + +// LeaseTable stores the current fencing token per workspace ID. +// Cross-replica correctness is the token itself, not this process mutex. +type LeaseTable struct { + mu sync.Mutex + tokens map[string]string + owner map[string]int64 +} + +func NewLeaseTable() *LeaseTable { + return &LeaseTable{tokens: map[string]string{}, owner: map[string]int64{}} +} + +func (t *LeaseTable) Issue(ref WorkspaceRef) { + t.mu.Lock() + defer t.mu.Unlock() + t.tokens[ref.ID] = ref.FencingToken + t.owner[ref.ID] = ref.ProjectScope +} + +func (t *LeaseTable) Assert(ref WorkspaceRef) error { + t.mu.Lock() + defer t.mu.Unlock() + cur, ok := t.tokens[ref.ID] + if !ok || cur == "" || ref.FencingToken == "" || cur != ref.FencingToken { + return ErrStaleFencing + } + if scope, ok := t.owner[ref.ID]; ok && ref.ProjectScope != 0 && scope != 0 && scope != ref.ProjectScope { + return ErrCrossProject + } + return nil +} diff --git a/internal/workspace/lease_test.go b/internal/workspace/lease_test.go new file mode 100644 index 00000000..f97da041 --- /dev/null +++ b/internal/workspace/lease_test.go @@ -0,0 +1,17 @@ +package workspace + +import "testing" + +func TestLeaseTableRejectsStaleTokenAfterReissue(t *testing.T) { + table := NewLeaseTable() + old := WorkspaceRef{ID: "ws1", FencingToken: "old-token", ProjectScope: 3} + table.Issue(old) + fresh := WorkspaceRef{ID: "ws1", FencingToken: "new-token", ProjectScope: 3} + table.Issue(fresh) + if err := table.Assert(old); err != ErrStaleFencing { + t.Fatalf("stale: %v", err) + } + if err := table.Assert(fresh); err != nil { + t.Fatal(err) + } +} diff --git a/internal/workspace/resolver.go b/internal/workspace/resolver.go index f5582e55..b323f9f6 100644 --- a/internal/workspace/resolver.go +++ b/internal/workspace/resolver.go @@ -2,7 +2,6 @@ package workspace import ( "context" - "errors" "fmt" "os" "path/filepath" @@ -11,12 +10,7 @@ import ( "github.com/jackc/pgx/v5/pgxpool" ) -var ( - ErrPathEscape = errors.New("workspace: path escapes data root") - ErrWorkDirNotFound = errors.New("workspace: work directory not found") -) - -// Resolver 将逻辑 ID 解析为受约束的工作区路径。 +// Resolver 将逻辑 ID 解析为受约束的工作区路径。物理路径不得进入 WorkspaceRef。 type Resolver struct { dataRoot string pool *pgxpool.Pool diff --git a/internal/workspace/types.go b/internal/workspace/types.go new file mode 100644 index 00000000..83e1f451 --- /dev/null +++ b/internal/workspace/types.go @@ -0,0 +1,28 @@ +package workspace + +import "errors" + +const SchemaVersion = 1 + +var ( + ErrInvalidKey = errors.New("workspace: invalid artifact key") + ErrPathEscape = errors.New("workspace: path escapes workspace") + ErrStaleFencing = errors.New("workspace: stale fencing token") + ErrUnknownSchema = errors.New("workspace: unknown schema version") + ErrIOInTransaction = errors.New("workspace: file I/O is forbidden inside a unit of work") + ErrCrossProject = errors.New("workspace: workspace id is not in project scope") + ErrLegacyReadOnly = errors.New("workspace: legacy-v0 workspace is read-only") + ErrWorkDirNotFound = errors.New("workspace: work directory not found") + ErrIncompleteMarker = errors.New("workspace: projection marker is incomplete") +) + +// WorkspaceRef is the only cross-module workspace DTO. It must not carry +// physical paths (no Root, OutputDir, or ManifestPath). +type WorkspaceRef struct { + ID string + SchemaVersion int + FencingToken string + ProjectScope int64 + RunID string + Environment string +}