199 lines
7.9 KiB
Go
199 lines
7.9 KiB
Go
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)
|
|
// Replay can pass more than a thousand private events before the first
|
|
// visible frame; account checks on CI and under -race need headroom.
|
|
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
|
|
defer cancel()
|
|
w := httptest.NewRecorder()
|
|
s.Handler().ServeHTTP(w, r.WithContext(ctx))
|
|
return w
|
|
}
|
|
|
|
// projectStreamReplayUntil reads a replay until marker arrives or the stream
|
|
// ends. Every replayed frame rechecks the API key with a durable write, so a
|
|
// thousand-event replay has taken over 11s on a disk-backed SQLite; the
|
|
// deadline only bounds a hang.
|
|
func projectStreamReplayUntil(t *testing.T, s *Server, key, project, marker string) (int, []byte) {
|
|
t.Helper()
|
|
httpServer := httptest.NewServer(s.Handler())
|
|
defer httpServer.Close()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
r, _ := http.NewRequestWithContext(ctx, "GET", httpServer.URL+"/api/events?scope=project&projectId="+project+"&since=0", nil)
|
|
r.Header.Set("Authorization", "Bearer "+key)
|
|
response, err := httpServer.Client().Do(r)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer response.Body.Close()
|
|
reader := bufio.NewReader(response.Body)
|
|
var body bytes.Buffer
|
|
for !bytes.Contains(body.Bytes(), []byte(marker)) {
|
|
line, err := reader.ReadString('\n')
|
|
body.WriteString(line)
|
|
if err != nil {
|
|
break
|
|
}
|
|
}
|
|
return response.StatusCode, body.Bytes()
|
|
}
|
|
func projectEventFixture(t *testing.T) (*Server, string, string, string, *atomic.Bool) {
|
|
t.Helper()
|
|
s, st := newAuthTestServer(t)
|
|
alice, _ := createLegacyAPIKeyFixture(t, st, "alice", "test")
|
|
bob, _ := createLegacyAPIKeyFixture(t, st, "bob", "test")
|
|
p, _ := st.CreateProject("alice", "Project")
|
|
s.Cards = cardsvc.New(st, s.mgr.EmitCardEvent)
|
|
s.Cards.SetAccountCheckedEmitter(s.mgr.EmitCardEventAccountChecked)
|
|
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)
|
|
}
|
|
// The shared frame follows every private event in seq order, so reading up
|
|
// to it observes the whole private replay.
|
|
code, body := projectStreamReplayUntil(t, s, bob, p, "EXPLICIT SHARED")
|
|
if code != 200 || bytes.Contains(body, []byte("PRIVATE HISTORY")) || !bytes.Contains(body, []byte("EXPLICIT SHARED")) {
|
|
t.Fatalf("private pagination/spoof proof failed: code=%d private=%t shared=%t bytes=%d", code,
|
|
bytes.Contains(body, []byte("PRIVATE HISTORY")), bytes.Contains(body, []byte("EXPLICIT SHARED")), len(body))
|
|
}
|
|
}
|
|
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")
|
|
}
|
|
}
|