From d26f3cf2e864a76ad0b868fd968da3257cc7bb44 Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 13 Oct 2026 11:12:47 +0000 Subject: [PATCH] Protect private project event replay and live delivery Why: Legacy project streams replayed upload filenames and parser metadata to newly authorized members even though the underlying cards remained owner-only. Evidence: SELF real HTTP red regressions exposed legacy replay, spoofed sharing claims, private-history pagination, and live delivery after membership removal. The affected packages, uncached full Backend suite, isolated PostgreSQL Store suite, and agent-system build pass after applying the event boundary. Impact: Registered lifecycle streams resolve trusted resource ownership and require an internal publication proof for cross-owner delivery. Event and proof commit in one transaction; v30 adds an additive SQLite/PostgreSQL sidecar without granting access to historical events. Seen and visible cursors remain separate, revoked members leave live streams, and unregistered base behavior remains compatible. The storage migration tool preserves the new proof rows. Refs: project-member-access-implementation-plan.md Task 1 / R5; authorized legacy SSE privacy follow-up; project-member-composed-legacy-project-history.json --- cmd/storage-migrate/main.go | 4 +- internal/app/app.go | 1 + internal/cardsvc/cardsvc.go | 41 +++++- internal/runmgr/events.go | 28 ++++ internal/runmgr/runmgr.go | 6 + internal/server/project_events.go | 30 ++++ internal/server/project_events_test.go | 163 ++++++++++++++++++++++ internal/server/project_lifecycle_test.go | 3 +- internal/server/server.go | 34 ++++- internal/store/messages.go | 11 ++ internal/store/migrate.go | 7 + internal/store/migrate_compat.go | 1 + internal/store/project_events.go | 147 +++++++++++++++++++ internal/store/project_events_test.go | 140 +++++++++++++++++++ 14 files changed, 605 insertions(+), 11 deletions(-) create mode 100644 internal/server/project_events.go create mode 100644 internal/server/project_events_test.go create mode 100644 internal/store/project_events.go create mode 100644 internal/store/project_events_test.go diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go index bdb1561..7ee9cb2 100644 --- a/cmd/storage-migrate/main.go +++ b/cmd/storage-migrate/main.go @@ -284,7 +284,7 @@ var copyOrder = []string{ "projects", "users", "profiles", "settings", "plugins", "conversations", "runs", "messages", "events", "artifacts", "hitl_cards", "plans", "todos", "knowledge", "knowledge_vectors", - "activity_cards", "command_audit", "prompt_modules", + "activity_cards", "project_event_shares", "command_audit", "prompt_modules", "sessions", "api_keys", "plugin_grants", "plugin_schema", "plugin_kv", "background_tasks", "memory_entries", "audit_events", "pending_confirms", "llm_usage_daily", @@ -784,7 +784,7 @@ func advanceSequences(db *sql.DB, table string) error { // checkOwners：业务行引用的 owner 必须存在于 users（单租户默认 ”/'local' // 除外）。未知 owner 阻塞迁移（U4/R7：不静默带过未知归属）。 func checkOwners(db *sql.DB, src *sql.DB) error { - ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards", "command_audit"} + ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards", "project_event_shares", "command_audit"} for _, t := range ownerTables { if !sqliteHasTable(src, t) { continue diff --git a/internal/app/app.go b/internal/app/app.go index 67587f3..33749b7 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -528,6 +528,7 @@ func Run(opt Options) error { // 收件箱任务卡：事件走会话事件管道（持久化/seq/重放）；关闭即 T0 基线。 cardSvc := cardsvc.New(st, mgr.EmitCardEvent) + cardSvc.SetProjectSharedEmitter(mgr.EmitProjectSharedCardEvent) surfaceSvc := surfacesvc.New(st, mgr.EmitCardEvent) go func() { // 启动恢复：重发「快照已落盘、事件未发出」的待恢复投影（KTD）。 if n, err := surfaceSvc.RecoverPendingEvents(context.Background()); err == nil && n > 0 { diff --git a/internal/cardsvc/cardsvc.go b/internal/cardsvc/cardsvc.go index a17b6dc..6b4d0ab 100644 --- a/internal/cardsvc/cardsvc.go +++ b/internal/cardsvc/cardsvc.go @@ -27,8 +27,9 @@ var ( const cardAppendWindow = 250 * time.Millisecond type Service struct { - st *store.Store - emit Emitter + st *store.Store + emit Emitter + projectSharedEmit ProjectSharedEmitter // MirrorConvCards EXT-12：会话卡是否镜像进项目公共流（缺省 true=现状全量 // 镜像；false 时会话卡只进本人会话流，项目级卡不受影响）。 MirrorConvCards bool @@ -41,6 +42,42 @@ func New(st *store.Store, emit Emitter) *Service { return &Service{st: st, emit: emit, now: time.Now, lastEmit: map[string]time.Time{}, MirrorConvCards: true} } +// ProjectSharedEmitter is an internal trusted publication adapter. It is never +// bound to model arguments or exposed as a mandatory extension capability. +type ProjectSharedEmitter func(owner, producer, channel, runID string, typ core.StreamEventType, payload any) error + +func (s *Service) SetProjectSharedEmitter(emit ProjectSharedEmitter) { s.projectSharedEmit = emit } + +// PublishProjectCard explicitly publishes a project-level structural snapshot. +// Ordinary Create/Update remain private unless this entry issues a new proof. +func (s *Service) PublishProjectCard(owner, producer string, card core.ActivityCard, ref string) (core.ActivityCard, error) { + if s.projectSharedEmit == nil || owner == "" || producer == "" || card.ProjectID == "" || card.ConversationID != "" { + return card, errors.New("card: trusted project publisher required") + } + if card.ID != "" { + current, err := s.st.GetCard(owner, card.ID) + if err != nil { + return card, err + } + if current.ProjectID != card.ProjectID || current.ConversationID != "" { + return card, store.ErrNotFound + } + } else { + card.ID = store.NewID("card") + } + now := s.now().UTC() + card.CreatedAt = now + card.UpdatedAt = now + if card.Status == "" { + card.Status = core.CardActive + } + if err := s.st.SaveCard(owner, &card, ref); err != nil { + return card, err + } + err := s.projectSharedEmit(owner, producer, string(core.ProjectChannelKey(card.ProjectID)), card.SourceRunID, core.EvCardCreated, core.CardPayload{Card: card}) + return card, err +} + // emitCard 按卡归属发布卡片事件（D50）： // - 会话卡 → 会话流；同时镜像项目流（伪会话键 "project:"）—— // 项目收件箱聚合视图（项目级卡 ∪ 各会话未决卡）的实时性依赖它；卡片事件 diff --git a/internal/runmgr/events.go b/internal/runmgr/events.go index 9e41162..971e0c2 100644 --- a/internal/runmgr/events.go +++ b/internal/runmgr/events.go @@ -116,6 +116,34 @@ func (eb *eventBuffer) emit(conversationID, runID string, typ core.StreamEventTy return 0, nil } +// emitProjectShared persists an explicit internal publication and its proof +// before fan-out. A client-visible payload cannot opt itself into this path. +func (eb *eventBuffer) emitProjectShared(owner, producer, channel, runID string, typ core.StreamEventType, payload any) error { + project, ok := core.ProjectFromChannel(channel) + if !ok { + return store.ErrNotFound + } + encoded, err := json.Marshal(payload) + if err != nil { + return err + } + event := core.StreamEvent{ConversationID: channel, ProjectID: project, RunID: runID, Type: typ, Payload: encoded, TS: time.Now().UTC().Format(time.RFC3339Nano), V: core.ProtocolVersion} + eb.flushConversation(channel) + eb.flushing.Lock() + defer eb.flushing.Unlock() + events := []core.StreamEvent{event} + if _, err := eb.store.AppendProjectSharedEvents(owner, producer, channel, events); err != nil { + return err + } + for _, event := range events { + eb.bus.Publish(event) + } + if eb.onPersisted != nil { + eb.onPersisted(channel, events) + } + return nil +} + // append 落盘并按 seq 顺序发布。 func (eb *eventBuffer) append(conversationID string, evs []core.StreamEvent) error { eb.flushing.Lock() diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index e5fe6a3..b3f3d99 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -179,6 +179,12 @@ type IntentGate interface { Match(ctx context.Context, input string) (branch, topID string) } +// EmitProjectSharedCardEvent is an internal producer-authorized structural +// publication entry, separate from the ordinary private card event emitter. +func (m *Manager) EmitProjectSharedCardEvent(owner, producer, channel, runID string, typ core.StreamEventType, payload any) error { + return m.events.emitProjectShared(owner, producer, channel, runID, typ, payload) +} + // SetGate 注入意图闸门（装配期一次；插件意图 overlay 经 pluginhost 回调重建）。 func (m *Manager) SetGate(g IntentGate) { m.gate = g } diff --git a/internal/server/project_events.go b/internal/server/project_events.go new file mode 100644 index 0000000..7058cec --- /dev/null +++ b/internal/server/project_events.go @@ -0,0 +1,30 @@ +package server + +import ( + core "agent-system-core" +) + +// projectStreamMembership consults the registered resource authorizer using the +// physical project's trusted owner, never the event publisher or payload claims. +func (s *Server) projectStreamMembership(actor, project string) bool { + p, err := s.st.ProjectMetadata(project) + if err != nil { + return false + } + return s.authorizeSharedResource(actor, "stream", "project", project, p.Owner, project) +} + +func (s *Server) projectEventVisible(actor string, event core.StreamEvent) bool { + if s.projectLifecycle == nil { + return true + } + access, err := s.st.ResolveProjectEventAccess(event) + if err != nil || access.Owner == "" { + return false + } + if access.Owner == actor { + return true + } + project, ok := core.ProjectFromChannel(event.ConversationID) + return ok && access.Shared && s.projectStreamMembership(actor, project) +} diff --git a/internal/server/project_events_test.go b/internal/server/project_events_test.go new file mode 100644 index 0000000..0360ded --- /dev/null +++ b/internal/server/project_events_test.go @@ -0,0 +1,163 @@ +package server + +import ( + "agent-system-backend/internal/cardsvc" + "agent-system-backend/pkg/extension" + core "agent-system-core" + "bufio" + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +func projectStreamReplay(t *testing.T, s *Server, key, project string) *httptest.ResponseRecorder { + t.Helper() + r := httptest.NewRequest("GET", "/api/events?scope=project&projectId="+project+"&since=0", nil) + r.Header.Set("Authorization", "Bearer "+key) + ctx, cancel := context.WithTimeout(r.Context(), 200*time.Millisecond) + defer cancel() + w := httptest.NewRecorder() + s.Handler().ServeHTTP(w, r.WithContext(ctx)) + return w +} +func projectEventFixture(t *testing.T) (*Server, string, string, string, *atomic.Bool) { + t.Helper() + s, st := newAuthTestServer(t) + alice, _ := st.CreateAPIKey("alice", "test") + bob, _ := st.CreateAPIKey("bob", "test") + p, _ := st.CreateProject("alice", "Project") + s.Cards = cardsvc.New(st, s.mgr.EmitCardEvent) + s.Cards.SetProjectSharedEmitter(s.mgr.EmitProjectSharedCardEvent) + member := &atomic.Bool{} + member.Store(true) + s.SetResourceAuthorizerV2(func(q extension.ResourceAuthzQuery) (bool, error) { + return q.ResourceType == "project" && q.ProjectID == p.ID && (q.Actor == "alice" || (q.Actor == "bob" && member.Load())), nil + }) + return s, alice, bob, p.ID, member +} +func TestProjectEventsLegacyReplayIsPrivate(t *testing.T) { + s, alice, bob, p, _ := projectEventFixture(t) + s.DisableUploadIndexing = true + if w := lifecycleUpload(t, s, alice, p, "legacy-owner-private.txt", "PRIVATE"); w.Code != 200 { + t.Fatal(w.Code) + } + s.SetProjectLifecycle(extension.ProjectLifecycle{Work: func(context.Context, string, core.ID) (bool, error) { return true, nil }}, nil) + member := projectStreamReplay(t, s, bob, p) + if member.Code != 200 || bytes.Contains(member.Body.Bytes(), []byte("legacy-owner-private.txt")) || bytes.Contains(member.Body.Bytes(), []byte("builtin:txt")) { + t.Fatal("legacy private upload leaked to member") + } + owner := projectStreamReplay(t, s, alice, p) + if owner.Code != 200 || !bytes.Contains(owner.Body.Bytes(), []byte("legacy-owner-private.txt")) { + t.Fatal("owner lost private history") + } +} +func TestProjectEventsSharedProofAndPrivatePagination(t *testing.T) { + s, _, bob, p, _ := projectEventFixture(t) + s.SetProjectLifecycle(extension.ProjectLifecycle{}, nil) + card, err := s.Cards.Create("alice", core.ActivityCard{ProjectID: p, Title: "PRIVATE HISTORY"}, "") + if err != nil { + t.Fatal(err) + } + payload, _ := json.Marshal(map[string]any{"card": card, "owner": "bob", "shared": true, "producer": "plugin:rdm:structure"}) + private := make([]core.StreamEvent, 1001) + for i := range private { + private[i] = core.StreamEvent{Type: core.EvCardUpdated, Payload: payload, TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1} + } + if _, err := s.st.AppendEvents(core.ProjectChannelKey(p), private); err != nil { + t.Fatal(err) + } + if _, err := s.Cards.PublishProjectCard("alice", "plugin:rdm:structure", core.ActivityCard{ProjectID: p, Title: "EXPLICIT SHARED"}, ""); err != nil { + t.Fatal(err) + } + w := projectStreamReplay(t, s, bob, p) + if w.Code != 200 || bytes.Contains(w.Body.Bytes(), []byte("PRIVATE HISTORY")) || !bytes.Contains(w.Body.Bytes(), []byte("EXPLICIT SHARED")) { + t.Fatal("private pagination/spoof proof failed") + } +} +func TestProjectEventsLivePrivacyAndRevocation(t *testing.T) { + s, _, bob, p, member := projectEventFixture(t) + s.SetProjectLifecycle(extension.ProjectLifecycle{}, nil) + httpServer := httptest.NewServer(s.Handler()) + defer httpServer.Close() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, "GET", httpServer.URL+"/api/events?scope=project&projectId="+p, nil) + req.Header.Set("Authorization", "Bearer "+bob) + response, err := httpServer.Client().Do(req) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + if _, err := s.Cards.Create("alice", core.ActivityCard{ProjectID: p, Title: "LIVE PRIVATE"}, ""); err != nil { + t.Fatal(err) + } + if _, err := s.Cards.PublishProjectCard("alice", "plugin:rdm:structure", core.ActivityCard{ProjectID: p, Title: "LIVE SHARED"}, ""); err != nil { + t.Fatal(err) + } + reader := bufio.NewReader(response.Body) + var before bytes.Buffer + for { + line, err := reader.ReadString('\n') + if err != nil { + t.Fatalf("live shared event did not arrive: %v", err) + } + before.WriteString(line) + if bytes.Contains([]byte(line), []byte("LIVE SHARED")) { + break + } + } + if bytes.Contains(before.Bytes(), []byte("LIVE PRIVATE")) { + t.Fatal("live private frame leaked before revocation") + } + member.Store(false) + if _, err := s.Cards.PublishProjectCard("alice", "plugin:rdm:structure", core.ActivityCard{ProjectID: p, Title: "AFTER REMOVAL"}, ""); err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("revoked stream did not stop: %v", err) + } + if bytes.Contains(body, []byte("LIVE PRIVATE")) || bytes.Contains(body, []byte("AFTER REMOVAL")) { + t.Fatal("live private/revoked event leaked") + } +} +func TestUnregisteredProjectEventsRetainLegacyReads(t *testing.T) { + s, _, bob, p, _ := projectEventFixture(t) + if _, err := s.Cards.Create("alice", core.ActivityCard{ProjectID: p, Title: "LEGACY BASE"}, ""); err != nil { + t.Fatal(err) + } + w := projectStreamReplay(t, s, bob, p) + if w.Code != 200 || !bytes.Contains(w.Body.Bytes(), []byte("LEGACY BASE")) { + t.Fatal("unregistered project stream changed") + } +} + +func TestProjectEventsIdleStreamStopsAfterMembershipRemoval(t *testing.T) { + previous := sseAuthRecheck + sseAuthRecheck = 20 * time.Millisecond + defer func() { sseAuthRecheck = previous }() + s, _, bob, p, member := projectEventFixture(t) + s.SetProjectLifecycle(extension.ProjectLifecycle{}, nil) + httpServer := httptest.NewServer(s.Handler()) + defer httpServer.Close() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + request, _ := http.NewRequestWithContext(ctx, "GET", httpServer.URL+"/api/events?scope=project&projectId="+p, nil) + request.Header.Set("Authorization", "Bearer "+bob) + response, err := httpServer.Client().Do(request) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + member.Store(false) + body, err := io.ReadAll(response.Body) + if err != nil || !bytes.Contains(body, []byte("project-access-revoked")) { + t.Fatal("idle project stream kept revoked member connected") + } +} diff --git a/internal/server/project_lifecycle_test.go b/internal/server/project_lifecycle_test.go index 47ea686..a58553e 100644 --- a/internal/server/project_lifecycle_test.go +++ b/internal/server/project_lifecycle_test.go @@ -409,8 +409,9 @@ func TestProjectLifecycleUploadSuppressesPrivateProjectCards(t *testing.T) { return q.ResourceType == "project" && q.ProjectID == p.ID && (q.Actor == "alice" || q.Actor == "bob"), nil }) s.Cards = cardsvc.New(st, s.mgr.EmitCardEvent) + s.Cards.SetProjectSharedEmitter(s.mgr.EmitProjectSharedCardEvent) s.DisableUploadIndexing = true - if _, err := s.Cards.Create("alice", core.ActivityCard{ProjectID: p.ID, Title: "Shared structure", Kind: core.CardKindProgress}, ""); err != nil { + if _, err := s.Cards.PublishProjectCard("alice", "plugin:rdm:structure", core.ActivityCard{ProjectID: p.ID, Title: "Shared structure", Kind: core.CardKindProgress}, ""); err != nil { t.Fatal(err) } // The owner path already admitted upload before the member authorization fix. diff --git a/internal/server/server.go b/internal/server/server.go index 426070e..d90b3c3 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -431,6 +431,7 @@ var sseAuthRecheck = 15 * time.Second func (s *Server) events(w http.ResponseWriter, r *http.Request) { conversationID := r.URL.Query().Get("conversationId") isProject := false + streamProject := "" // 项目流（D50）：scope=project&projectId= 订阅项目频道——伪会话键 // "project:" 复用会话流的持久化/seq/重放机制；租户边界按项目校验。 if r.URL.Query().Get("scope") == "project" { @@ -448,6 +449,7 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { } conversationID = string(core.ProjectChannelKey(core.ID(pid))) isProject = true + streamProject = pid } else if conversationID == "" { http.Error(w, "conversationId required", http.StatusBadRequest) return @@ -509,28 +511,38 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { live, stop := s.Subscribe(conversationID) defer stop() + lastVisible := since send := func(e core.StreamEvent) bool { + if e.Seq <= lastVisible { + return true + } + if isProject && !s.projectEventVisible(wantOwner, e) { + return r.Context().Err() == nil + } buf, _ := json.Marshal(e) fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", e.Seq, e.Type, buf) flusher.Flush() + lastVisible = e.Seq return r.Context().Err() == nil } - lastSent := since + // Seen progresses even when a private frame is filtered. Visible only tracks + // frames delivered to this actor, preserving authoritative seq and resumption. + lastSeen := since for { - replay, err := s.st.EventsAfter(conversationID, lastSent, 1000) + replay, err := s.st.EventsAfter(conversationID, lastSeen, 1000) if err != nil || len(replay) == 0 { break } for _, e := range replay { - if e.Seq <= lastSent { + if e.Seq <= lastSeen { continue } if !send(e) { stop() return } - lastSent = e.Seq + lastSeen = e.Seq } if len(replay) < 1000 { break @@ -548,6 +560,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { stop() return case <-recheck.C: + if isProject && s.projectLifecycle != nil && !s.projectStreamMembership(wantOwner, streamProject) { + fmt.Fprint(w, ": project-access-revoked\n\n") + flusher.Flush() + return + } if !stillAuthed() { fmt.Fprint(w, ": session-expired\n\n") flusher.Flush() @@ -562,14 +579,19 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { stop() return } - if e.ConversationID != conversationID || e.Seq <= lastSent { + if e.ConversationID != conversationID || e.Seq <= lastSeen { continue } + if isProject && s.projectLifecycle != nil && !s.projectStreamMembership(wantOwner, streamProject) { + fmt.Fprint(w, ": project-access-revoked\n\n") + flusher.Flush() + return + } if !send(e) { stop() return } - lastSent = e.Seq + lastSeen = e.Seq } } } diff --git a/internal/store/messages.go b/internal/store/messages.go index 36c2042..58724f2 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -266,6 +266,10 @@ func (s *Store) ChildRuns(runID string) ([]core.Run, error) { // 成功后回填 evs[i].Seq。事务内锁定会话序列：PG 用事务级咨询锁（跨实例安全）， // SQLite 单写连接天然串行。空批次为空操作。 func (s *Store) AppendEvents(conversationID string, evs []core.StreamEvent) (int64, error) { + return s.appendEvents(conversationID, evs, "", "") +} + +func (s *Store) appendEvents(conversationID string, evs []core.StreamEvent, owner, producer string) (int64, error) { if len(evs) == 0 { return 0, nil } @@ -298,6 +302,13 @@ func (s *Store) AppendEvents(conversationID string, evs []core.StreamEvent) (int if _, err := tx.Exec(s.q(sb.String()), args...); err != nil { return 0, err } + if producer != "" { + for i, e := range evs { + if err := s.saveProjectEventProof(tx, conversationID, base.Int64+int64(i)+1, e, owner, producer); err != nil { + return 0, err + } + } + } // KTD6：事件提交与 NOTIFY 同事务（outbox 语义的前半——通知不早于事件 // 可见；后半的命令 outbox 走 EnqueueCommandInTx 由批准路径使用）。 if s.dialect.name() == "postgres" { diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 4428563..c31e0e8 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -841,6 +841,13 @@ func init() { `DROP INDEX IF EXISTS idx_redeem_attempts_ip_at`, `DROP TABLE IF EXISTS redeem_attempts`, }}}, + migration{version: 30, name: "project_event_publication_proofs", + up: map[string][]string{ + "sqlite": {`CREATE TABLE project_event_shares(channel TEXT NOT NULL,seq INTEGER NOT NULL,event_type TEXT NOT NULL,payload_sha256 TEXT NOT NULL,owner TEXT NOT NULL,resource_type TEXT NOT NULL,resource_id TEXT NOT NULL,producer TEXT NOT NULL,created_at TEXT NOT NULL,PRIMARY KEY(channel,seq),FOREIGN KEY(channel,seq) REFERENCES events(conversation_id,seq) ON DELETE CASCADE)`}, + "postgres": {`CREATE TABLE project_event_shares(channel TEXT NOT NULL,seq BIGINT NOT NULL,event_type TEXT NOT NULL,payload_sha256 TEXT NOT NULL,owner TEXT NOT NULL,resource_type TEXT NOT NULL,resource_id TEXT NOT NULL,producer TEXT NOT NULL,created_at TEXT NOT NULL,PRIMARY KEY(channel,seq),FOREIGN KEY(channel,seq) REFERENCES events(conversation_id,seq) ON DELETE CASCADE)`}, + }, + down: map[string][]string{"sqlite": {`DROP TABLE IF EXISTS project_event_shares`}, "postgres": {`DROP TABLE IF EXISTS project_event_shares`}}, + }, ) } diff --git a/internal/store/migrate_compat.go b/internal/store/migrate_compat.go index ac0347d..86c692c 100644 --- a/internal/store/migrate_compat.go +++ b/internal/store/migrate_compat.go @@ -34,6 +34,7 @@ var migrationBreaksOldReaders = map[int]bool{ 27: false, 28: false, // auth_version_account_events：仅加列与新表，旧读者不受影响 29: false, // redeem_attempts：新表，旧读者不受影响 + 30: false, // project_event_publication_proofs：新表，旧读者不受影响 } // SchemaTarget 是本二进制会迁移到的 schema 版本。 diff --git a/internal/store/project_events.go b/internal/store/project_events.go new file mode 100644 index 0000000..ea6e7c4 --- /dev/null +++ b/internal/store/project_events.go @@ -0,0 +1,147 @@ +package store + +import ( + core "agent-system-core" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "strings" +) + +// ProjectEventAccess is trusted read metadata; payload claims never establish it. +type ProjectEventAccess struct { + Owner string + Shared bool + Producer string +} +type eventResource struct{ owner, project, conversation, kind, id string } +type eventQuery interface{ QueryRow(string, ...any) *sql.Row } + +func cardEventID(e core.StreamEvent) (string, error) { + switch e.Type { + case core.EvCardCreated, core.EvCardUpdated, core.EvCardStatus, core.EvCardResolved: + default: + return "", nil + } + var payload struct { + Card struct { + ID string `json:"id"` + } `json:"card"` + CardID string `json:"cardId"` + } + if err := json.Unmarshal(e.Payload, &payload); err != nil { + return "", err + } + switch e.Type { + case core.EvCardCreated, core.EvCardUpdated: + return payload.Card.ID, nil + case core.EvCardStatus, core.EvCardResolved: + return payload.CardID, nil + } + return "", nil +} +func eventPayloadHash(payload []byte) string { + hash := sha256.Sum256(payload) + return hex.EncodeToString(hash[:]) +} + +func (s *Store) projectEventResource(q eventQuery, e core.StreamEvent) (eventResource, error) { + var resource eventResource + id, err := cardEventID(e) + if err != nil { + return resource, err + } + switch e.Type { + case core.EvCardCreated, core.EvCardUpdated, core.EvCardStatus, core.EvCardResolved: + if id == "" { + return resource, ErrNotFound + } + } + if id != "" { + resource.kind, resource.id = "card", id + err = q.QueryRow(s.q(`SELECT owner,project_id,conversation_id FROM activity_cards WHERE id=?`), id).Scan(&resource.owner, &resource.project, &resource.conversation) + } else if e.Type == core.EvArtifactCreated { + var payload struct { + Artifact struct { + ID string `json:"id"` + } `json:"artifact"` + } + if err := json.Unmarshal(e.Payload, &payload); err != nil { + return resource, err + } + if payload.Artifact.ID == "" { + return resource, ErrNotFound + } + resource.kind, resource.id = "artifact", payload.Artifact.ID + err = q.QueryRow(s.q(`SELECT owner,project_id FROM artifacts WHERE id=?`), resource.id).Scan(&resource.owner, &resource.project) + } else if e.RunID != "" { + resource.kind, resource.id = "run", e.RunID + err = q.QueryRow(s.q(`SELECT owner,project_id,conversation_id FROM runs WHERE id=?`), e.RunID).Scan(&resource.owner, &resource.project, &resource.conversation) + } else { + return resource, ErrNotFound + } + if errors.Is(err, sql.ErrNoRows) { + return resource, ErrNotFound + } + if err != nil { + return resource, err + } + project, ok := core.ProjectFromChannel(e.ConversationID) + if !ok || resource.owner == "" || resource.project != project || (e.ProjectID != "" && e.ProjectID != project) { + return resource, ErrNotFound + } + return resource, nil +} + +// ResolveProjectEventAccess resolves immutable resource ownership and verifies an +// exact internal publication record. Old events retain their private default. +func (s *Store) ResolveProjectEventAccess(e core.StreamEvent) (ProjectEventAccess, error) { + resource, err := s.projectEventResource(s.db, e) + if err != nil { + return ProjectEventAccess{}, err + } + access := ProjectEventAccess{Owner: resource.owner} + var owner, kind, id, producer, typ, hash string + err = s.db.QueryRow(s.q(`SELECT owner,resource_type,resource_id,producer,event_type,payload_sha256 FROM project_event_shares WHERE channel=? AND seq=?`), e.ConversationID, e.Seq).Scan(&owner, &kind, &id, &producer, &typ, &hash) + if errors.Is(err, sql.ErrNoRows) { + return access, nil + } + if err != nil { + return ProjectEventAccess{}, err + } + if owner != resource.owner || kind != resource.kind || id != resource.id || strings.TrimSpace(producer) == "" || typ != string(e.Type) || hash != eventPayloadHash(e.Payload) || resource.conversation != "" { + return ProjectEventAccess{}, ErrNotFound + } + access.Shared, access.Producer = true, producer + return access, nil +} + +// AppendProjectSharedEvents is an internal publication boundary, not an API or +// model tool. Events and their exact sharing proofs commit together. Caller +// identity must own a project-level card; private conversation cards cannot be +// published through this structural-card path. +func (s *Store) AppendProjectSharedEvents(owner, producer, channel string, events []core.StreamEvent) (int64, error) { + if owner == "" || strings.TrimSpace(producer) == "" { + return 0, fmt.Errorf("store: shared event requires owner and producer") + } + if _, ok := core.ProjectFromChannel(channel); !ok { + return 0, ErrNotFound + } + return s.appendEvents(channel, events, owner, producer) +} + +func (s *Store) saveProjectEventProof(tx *sql.Tx, channel string, seq int64, e core.StreamEvent, owner, producer string) error { + e.ConversationID = channel + resource, err := s.projectEventResource(tx, e) + if err != nil { + return err + } + if resource.kind != "card" || resource.owner != owner || resource.conversation != "" { + return ErrNotFound + } + _, err = tx.Exec(s.q(`INSERT INTO project_event_shares(channel,seq,event_type,payload_sha256,owner,resource_type,resource_id,producer,created_at) VALUES(?,?,?,?,?,?,?,?,?)`), channel, seq, string(e.Type), eventPayloadHash(e.Payload), resource.owner, resource.kind, resource.id, producer, nowISO()) + return err +} diff --git a/internal/store/project_events_test.go b/internal/store/project_events_test.go new file mode 100644 index 0000000..2c69aab --- /dev/null +++ b/internal/store/project_events_test.go @@ -0,0 +1,140 @@ +package store + +import ( + core "agent-system-core" + "encoding/json" + "testing" + "time" +) + +func storedProjectCardEvent(t *testing.T, st *Store, owner, project, id string) core.StreamEvent { + t.Helper() + card := core.ActivityCard{ID: id, ProjectID: project, Title: "PRIVATE", CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC()} + if err := st.SaveCard(owner, &card, ""); err != nil { + t.Fatal(err) + } + payload, _ := json.Marshal(core.CardPayload{Card: card}) + return core.StreamEvent{ConversationID: core.ProjectChannelKey(project), ProjectID: project, Type: core.EvCardCreated, Payload: payload, TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1} +} +func TestProjectEventSharedProofBindsExactEvent(t *testing.T) { + st := openTestStore(t) + p, _ := st.CreateProject("alice", "Project") + e := storedProjectCardEvent(t, st, "alice", p.ID, "card-a") + private := []core.StreamEvent{e} + if _, err := st.AppendEvents(e.ConversationID, private); err != nil { + t.Fatal(err) + } + access, err := st.ResolveProjectEventAccess(private[0]) + if err != nil || access.Owner != "alice" || access.Shared { + t.Fatalf("private access=%+v %v", access, err) + } + shared := []core.StreamEvent{e} + if _, err := st.AppendProjectSharedEvents("alice", "plugin:rdm:structure", e.ConversationID, shared); err != nil { + t.Fatal(err) + } + access, err = st.ResolveProjectEventAccess(shared[0]) + if err != nil || !access.Shared || access.Owner != "alice" { + t.Fatalf("shared proof=%+v %v", access, err) + } + access, err = st.ResolveProjectEventAccess(private[0]) + if err != nil || access.Shared { + t.Fatal("publication retroactively shared private history") + } + for _, mutate := range []func(*core.StreamEvent){func(e *core.StreamEvent) { e.Type = core.EvCardUpdated }, func(e *core.StreamEvent) { e.Payload = append(e.Payload, ' ') }, func(e *core.StreamEvent) { e.Seq += 100 }, func(e *core.StreamEvent) { e.ConversationID = "project:another" }, func(e *core.StreamEvent) { e.ProjectID = "another" }} { + changed := shared[0] + mutate(&changed) + access, err = st.ResolveProjectEventAccess(changed) + if err == nil && access.Shared { + t.Fatal("mutated event retained sharing") + } + } + bad := []core.StreamEvent{e} + before, _ := st.LastEventSeq(e.ConversationID) + if _, err := st.AppendProjectSharedEvents("bob", "plugin:rdm:structure", e.ConversationID, bad); err == nil { + t.Fatal("other owner issued publication") + } + after, _ := st.LastEventSeq(e.ConversationID) + if after != before { + t.Fatal("rejected publication appended event") + } +} +func TestProjectEventMigrationPreservesPrivateHistory(t *testing.T) { + st := openTestStore(t) + if err := st.MigrateDown(29); err != nil { + t.Fatal(err) + } + p, _ := st.CreateProject("alice", "Legacy") + e := storedProjectCardEvent(t, st, "alice", p.ID, "legacy-card") + events := []core.StreamEvent{e} + if _, err := st.AppendEvents(e.ConversationID, events); err != nil { + t.Fatal(err) + } + before, _ := st.EventsAfter(e.ConversationID, 0, 100) + if err := st.migrate(); err != nil { + t.Fatal(err) + } + after, _ := st.EventsAfter(e.ConversationID, 0, 100) + b, _ := json.Marshal(before) + a, _ := json.Marshal(after) + if string(a) != string(b) { + t.Fatal("upgrade changed private event history") + } + access, err := st.ResolveProjectEventAccess(after[0]) + if err != nil || access.Owner != "alice" || access.Shared { + t.Fatalf("legacy privacy=%+v %v", access, err) + } +} + +func TestProjectSharedEventBatchRollsBackProofAndEvents(t *testing.T) { + st := openTestStore(t) + p, _ := st.CreateProject("alice", "Project") + a := storedProjectCardEvent(t, st, "alice", p.ID, "card-a") + b := storedProjectCardEvent(t, st, "bob", p.ID, "card-b") + if _, err := st.AppendProjectSharedEvents("alice", "plugin:rdm:structure", a.ConversationID, []core.StreamEvent{a, b}); err == nil { + t.Fatal("mixed owner batch published") + } + var count int + for _, table := range []string{"events", "project_event_shares"} { + if err := st.db.QueryRow("SELECT COUNT(*) FROM " + table).Scan(&count); err != nil || count != 0 { + t.Fatalf("partial %s publication: count=%d error=%v", table, count, err) + } + } +} + +func TestProjectEventInvalidProofAndMissingOwnerFailClosed(t *testing.T) { + st := openTestStore(t) + p, _ := st.CreateProject("alice", "Project") + e := storedProjectCardEvent(t, st, "alice", p.ID, "card-a") + shared := []core.StreamEvent{e} + if _, err := st.AppendProjectSharedEvents("alice", "plugin:rdm:structure", e.ConversationID, shared); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(`UPDATE project_event_shares SET producer=''`); err != nil { + t.Fatal(err) + } + if access, err := st.ResolveProjectEventAccess(shared[0]); err == nil || access.Shared { + t.Fatal("corrupted publication proof was accepted") + } + if _, err := st.db.Exec(`UPDATE activity_cards SET owner=''`); err != nil { + t.Fatal(err) + } + if _, err := st.ResolveProjectEventAccess(shared[0]); err == nil { + t.Fatal("missing resource owner was accepted") + } +} + +func TestProjectSharingRejectsPrivateConversationCards(t *testing.T) { + st := openTestStore(t) + p, _ := st.CreateProject("alice", "Project") + e := storedProjectCardEvent(t, st, "alice", p.ID, "card-a") + if _, err := st.db.Exec(`UPDATE activity_cards SET conversation_id='private-conversation'`); err != nil { + t.Fatal(err) + } + if _, err := st.AppendProjectSharedEvents("alice", "plugin:rdm:structure", e.ConversationID, []core.StreamEvent{e}); err == nil { + t.Fatal("private conversation card publicly marked") + } + events, err := st.EventsAfter(e.ConversationID, 0, 100) + if err != nil || len(events) != 0 { + t.Fatal("rejected conversation publication appended events") + } +}