From 64a1d82db428519f71284cef4c3c04c46cfb2010 Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Wed, 12 Aug 2026 17:48:48 +0800 Subject: [PATCH] feat: persist immutable Bell events and receipts (#17) --- Bell/server/app/event/http.go | 52 +++++++ Bell/server/app/event/service.go | 134 ++++++++++++++++++ Bell/server/app/event/service_test.go | 99 +++++++++++++ Bell/server/app/event/types.go | 38 +++++ Bell/server/app/receipt/types.go | 9 ++ Bell/server/cmd/bell/root.go | 5 +- .../migrations/2026081202_event_receipt.sql | 29 ++++ Bell/web/src/api/event/index.js | 3 + Bell/web/src/bootstrap/router.js | 2 + Bell/web/src/layout/AppLayout.vue | 1 + Bell/web/src/views/event/Events.vue | 5 + 11 files changed, 376 insertions(+), 1 deletion(-) create mode 100644 Bell/server/app/event/http.go create mode 100644 Bell/server/app/event/service.go create mode 100644 Bell/server/app/event/service_test.go create mode 100644 Bell/server/app/event/types.go create mode 100644 Bell/server/app/receipt/types.go create mode 100644 Bell/server/migrations/2026081202_event_receipt.sql create mode 100644 Bell/web/src/api/event/index.js create mode 100644 Bell/web/src/views/event/Events.vue diff --git a/Bell/server/app/event/http.go b/Bell/server/app/event/http.go new file mode 100644 index 0000000..feeadc9 --- /dev/null +++ b/Bell/server/app/event/http.go @@ -0,0 +1,52 @@ +package event + +import ( + "encoding/json" + "errors" + "net/http" + "strconv" + + "git.ilapage.cn/ila/yovision/Bell/server/app/auth" + "git.ilapage.cn/ila/yovision/Bell/server/app/rbac" + "github.com/jackc/pgx/v5" +) + +type HTTP struct { + Service Service + Auth auth.HTTP +} + +func (h HTTP) Register(mux *http.ServeMux) { + mux.Handle("GET /api/v1/events", h.Auth.Require(rbac.EventsRead, http.HandlerFunc(h.list))) + mux.Handle("GET /api/v1/events/{id}", h.Auth.Require(rbac.EventsRead, http.HandlerFunc(h.get))) +} +func (h HTTP) list(w http.ResponseWriter, r *http.Request) { + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) + items, err := h.Service.List(r.Context(), limit, r.URL.Query().Get("before")) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "读取事件失败"}) + return + } + next := "" + if len(items) > 0 { + next = items[len(items)-1].ID + } + writeJSON(w, http.StatusOK, map[string]any{"items": items, "next": next}) +} +func (h HTTP) get(w http.ResponseWriter, r *http.Request) { + item, err := h.Service.Get(r.Context(), r.PathValue("id")) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + writeJSON(w, http.StatusNotFound, map[string]string{"error": "事件不存在"}) + return + } + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "读取事件失败"}) + return + } + writeJSON(w, http.StatusOK, item) +} +func writeJSON(w http.ResponseWriter, status int, value any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(value) +} diff --git a/Bell/server/app/event/service.go b/Bell/server/app/event/service.go new file mode 100644 index 0000000..75abe21 --- /dev/null +++ b/Bell/server/app/event/service.go @@ -0,0 +1,134 @@ +package event + +import ( + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "strings" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +var ErrConflict = errors.New("幂等键已用于不同事件载荷") +var ErrInvalid = errors.New("事件字段无效") + +type Service struct{ DB *pgxpool.Pool } + +func normalize(command Command) ([]byte, [32]byte, error) { + command.ProducerID = strings.TrimSpace(command.ProducerID) + command.SourceEventID = strings.TrimSpace(command.SourceEventID) + command.EventType = strings.TrimSpace(command.EventType) + command.Location = strings.TrimSpace(command.Location) + command.Severity = strings.ToLower(strings.TrimSpace(command.Severity)) + command.OccurredAt = command.OccurredAt.UTC() + if command.ProducerID == "" || command.SourceEventID == "" || command.EventType == "" || command.Location == "" || command.OccurredAt.IsZero() { + return nil, [32]byte{}, ErrInvalid + } + if command.Severity != "low" && command.Severity != "medium" && command.Severity != "high" && command.Severity != "critical" { + return nil, [32]byte{}, ErrInvalid + } + if command.EvidenceRef != nil { + value := strings.TrimSpace(*command.EvidenceRef) + if value == "" || strings.Contains(value, "@") || strings.Contains(value, "\\") || strings.HasPrefix(strings.ToLower(value), "file:") { + return nil, [32]byte{}, fmt.Errorf("%w: evidence_ref must be a safe logical reference", ErrInvalid) + } + command.EvidenceRef = &value + } + if command.Attributes == nil { + command.Attributes = map[string]any{} + } + data, err := json.Marshal(command) + if err != nil { + return nil, [32]byte{}, fmt.Errorf("normalize event: %w", err) + } + return data, sha256.Sum256(data), nil +} + +func (s Service) Ingest(ctx context.Context, command Command) (Result, error) { + payload, digest, err := normalize(command) + if err != nil { + return Result{}, err + } + var normalized Command + if err := json.Unmarshal(payload, &normalized); err != nil { + return Result{}, err + } + tx, err := s.DB.Begin(ctx) + if err != nil { + return Result{}, err + } + defer func() { _ = tx.Rollback(ctx) }() + var eventID string + err = tx.QueryRow(ctx, `INSERT INTO bell_events(producer_id,source_event_id,event_type,occurred_at,location,severity,evidence_ref,normalized_payload,payload_sha256) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9) ON CONFLICT(producer_id,source_event_id) DO NOTHING RETURNING id::text`, normalized.ProducerID, normalized.SourceEventID, normalized.EventType, normalized.OccurredAt, normalized.Location, normalized.Severity, normalized.EvidenceRef, payload, digest[:]).Scan(&eventID) + created := err == nil + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + return Result{}, err + } + if created { + if _, err = tx.Exec(ctx, `INSERT INTO bell_event_receipts(event_id,producer_id,source_event_id,payload_sha256) VALUES($1,$2,$3,$4)`, eventID, normalized.ProducerID, normalized.SourceEventID, digest[:]); err != nil { + return Result{}, err + } + } + result, storedDigest, err := loadResult(ctx, tx, normalized.ProducerID, normalized.SourceEventID) + if err != nil { + return Result{}, err + } + if !created && !equalDigest(storedDigest, digest[:]) { + return Result{}, ErrConflict + } + result.Duplicate = !created + if err = tx.Commit(ctx); err != nil { + return Result{}, err + } + return result, nil +} + +func loadResult(ctx context.Context, q interface { + QueryRow(context.Context, string, ...any) pgx.Row +}, producerID, sourceEventID string) (Result, []byte, error) { + var result Result + var digest []byte + err := q.QueryRow(ctx, `SELECT e.id::text,e.producer_id,e.source_event_id,e.event_type,e.occurred_at::text,e.location,e.severity,e.evidence_ref,e.normalized_payload,e.created_at::text,e.payload_sha256,r.id::text,r.accepted_at::text FROM bell_events e JOIN bell_event_receipts r ON r.event_id=e.id WHERE e.producer_id=$1 AND e.source_event_id=$2`, producerID, sourceEventID).Scan(&result.Event.ID, &result.Event.ProducerID, &result.Event.SourceEventID, &result.Event.EventType, &result.Event.OccurredAt, &result.Event.Location, &result.Event.Severity, &result.Event.EvidenceRef, &result.Event.NormalizedPayload, &result.Event.CreatedAt, &digest, &result.Receipt.ID, &result.Receipt.AcceptedAt) + result.Receipt.EventID = result.Event.ID + result.Receipt.ProducerID = producerID + result.Receipt.SourceEventID = sourceEventID + return result, digest, err +} +func equalDigest(a, b []byte) bool { + if len(a) != len(b) { + return false + } + var different byte + for i := range a { + different |= a[i] ^ b[i] + } + return different == 0 +} + +func (s Service) List(ctx context.Context, limit int, before string) ([]Event, error) { + if limit < 1 || limit > 100 { + limit = 25 + } + rows, err := s.DB.Query(ctx, `SELECT id::text,producer_id,source_event_id,event_type,occurred_at::text,location,severity,evidence_ref,normalized_payload,created_at::text FROM bell_events WHERE ($2='' OR (created_at,id)<((SELECT created_at FROM bell_events WHERE id=nullif($2,'')::uuid),nullif($2,'')::uuid)) ORDER BY created_at DESC,id DESC LIMIT $1`, limit, before) + if err != nil { + return nil, err + } + defer rows.Close() + items := []Event{} + for rows.Next() { + var item Event + if err := rows.Scan(&item.ID, &item.ProducerID, &item.SourceEventID, &item.EventType, &item.OccurredAt, &item.Location, &item.Severity, &item.EvidenceRef, &item.NormalizedPayload, &item.CreatedAt); err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} +func (s Service) Get(ctx context.Context, id string) (Event, error) { + var item Event + err := s.DB.QueryRow(ctx, `SELECT id::text,producer_id,source_event_id,event_type,occurred_at::text,location,severity,evidence_ref,normalized_payload,created_at::text FROM bell_events WHERE id=$1`, id).Scan(&item.ID, &item.ProducerID, &item.SourceEventID, &item.EventType, &item.OccurredAt, &item.Location, &item.Severity, &item.EvidenceRef, &item.NormalizedPayload, &item.CreatedAt) + return item, err +} diff --git a/Bell/server/app/event/service_test.go b/Bell/server/app/event/service_test.go new file mode 100644 index 0000000..e0c6fb0 --- /dev/null +++ b/Bell/server/app/event/service_test.go @@ -0,0 +1,99 @@ +package event + +import ( + "context" + "errors" + "fmt" + "os" + "sync" + "testing" + "time" + + "git.ilapage.cn/ila/yovision/Bell/server/internal/platform" + "git.ilapage.cn/ila/yovision/Bell/server/migrations" +) + +func TestNormalizeRejectsUnsafeEvidence(t *testing.T) { + value := `file:C:\camera\secret.mp4` + _, _, err := normalize(Command{ProducerID: "test", SourceEventID: "1", EventType: "intrusion", OccurredAt: time.Now(), Location: "gate", Severity: "high", EvidenceRef: &value}) + if !errors.Is(err, ErrInvalid) { + t.Fatalf("expected invalid evidence reference, got %v", err) + } +} + +func TestPostgresIdempotencyAndImmutability(t *testing.T) { + url := os.Getenv("BELL_TEST_DATABASE_URL") + if url == "" { + t.Skip("BELL_TEST_DATABASE_URL is not set") + } + ctx := context.Background() + db, err := platform.OpenDatabase(ctx, url) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := migrations.Apply(ctx, db); err != nil { + t.Fatal(err) + } + service := Service{DB: db} + key := fmt.Sprintf("event-test-%d", time.Now().UnixNano()) + command := Command{ProducerID: "integration-test", SourceEventID: key, EventType: "danger-zone", OccurredAt: time.Now().UTC().Truncate(time.Millisecond), Location: "test-zone", Severity: "high", Attributes: map[string]any{"track_id": "anonymous-1"}} + const count = 16 + results := make(chan Result, count) + errorsFound := make(chan error, count) + var wg sync.WaitGroup + for range count { + wg.Add(1) + go func() { + defer wg.Done() + result, err := service.Ingest(ctx, command) + if err != nil { + errorsFound <- err + return + } + results <- result + }() + } + wg.Wait() + close(results) + close(errorsFound) + for err := range errorsFound { + t.Errorf("concurrent ingest: %v", err) + } + var eventID, receiptID string + seen := 0 + for result := range results { + seen++ + if eventID == "" { + eventID = result.Event.ID + receiptID = result.Receipt.ID + } + if result.Event.ID != eventID || result.Receipt.ID != receiptID { + t.Errorf("unstable identities: %#v", result) + } + } + if seen != count { + t.Fatalf("received %d results, want %d", seen, count) + } + conflict := command + conflict.Location = "different" + if _, err := service.Ingest(ctx, conflict); !errors.Is(err, ErrConflict) { + t.Fatalf("expected conflict, got %v", err) + } + if _, err := db.Exec(ctx, `UPDATE bell_events SET location='tampered' WHERE id=$1`, eventID); err == nil { + t.Fatal("immutable Event accepted update") + } + db.Close() + reopened, err := platform.OpenDatabase(ctx, url) + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + afterRestart, err := (Service{DB: reopened}).Ingest(ctx, command) + if err != nil { + t.Fatal(err) + } + if afterRestart.Event.ID != eventID || afterRestart.Receipt.ID != receiptID || !afterRestart.Duplicate { + t.Fatalf("receipt changed after restart: %#v", afterRestart) + } +} diff --git a/Bell/server/app/event/types.go b/Bell/server/app/event/types.go new file mode 100644 index 0000000..d3a2295 --- /dev/null +++ b/Bell/server/app/event/types.go @@ -0,0 +1,38 @@ +package event + +import ( + "encoding/json" + "time" + + "git.ilapage.cn/ila/yovision/Bell/server/app/receipt" +) + +type Command struct { + ProducerID string `json:"producer_id"` + SourceEventID string `json:"source_event_id"` + EventType string `json:"event_type"` + OccurredAt time.Time `json:"occurred_at"` + Location string `json:"location"` + Severity string `json:"severity"` + EvidenceRef *string `json:"evidence_ref,omitempty"` + Attributes map[string]any `json:"attributes"` +} + +type Event struct { + ID string `json:"id"` + ProducerID string `json:"producer_id"` + SourceEventID string `json:"source_event_id"` + EventType string `json:"event_type"` + OccurredAt string `json:"occurred_at"` + Location string `json:"location"` + Severity string `json:"severity"` + EvidenceRef *string `json:"evidence_ref,omitempty"` + NormalizedPayload json.RawMessage `json:"normalized_payload"` + CreatedAt string `json:"created_at"` +} + +type Result struct { + Event Event `json:"event"` + Receipt receipt.Receipt `json:"receipt"` + Duplicate bool `json:"duplicate"` +} diff --git a/Bell/server/app/receipt/types.go b/Bell/server/app/receipt/types.go new file mode 100644 index 0000000..4013159 --- /dev/null +++ b/Bell/server/app/receipt/types.go @@ -0,0 +1,9 @@ +package receipt + +type Receipt struct { + ID string `json:"id"` + EventID string `json:"event_id"` + ProducerID string `json:"producer_id"` + SourceEventID string `json:"source_event_id"` + AcceptedAt string `json:"accepted_at"` +} diff --git a/Bell/server/cmd/bell/root.go b/Bell/server/cmd/bell/root.go index c70038a..7233569 100644 --- a/Bell/server/cmd/bell/root.go +++ b/Bell/server/cmd/bell/root.go @@ -14,6 +14,7 @@ import ( "git.ilapage.cn/ila/yovision/Bell/server/app/audit" "git.ilapage.cn/ila/yovision/Bell/server/app/auth" + "git.ilapage.cn/ila/yovision/Bell/server/app/event" "git.ilapage.cn/ila/yovision/Bell/server/config" "git.ilapage.cn/ila/yovision/Bell/server/internal/platform" "git.ilapage.cn/ila/yovision/Bell/server/migrations" @@ -63,7 +64,9 @@ func Run(ctx context.Context, args []string) error { return authService.BootstrapAdministrator(ctx, username, displayName, password) } app := platform.NewHTTPApp(db) - auth.HTTP{Service: authService, Store: authStore, Audit: auditStore, SecureCookie: cfg.CookieSecure}.Register(app.Router()) + authHTTP := auth.HTTP{Service: authService, Store: authStore, Audit: auditStore, SecureCookie: cfg.CookieSecure} + authHTTP.Register(app.Router()) + event.HTTP{Service: event.Service{DB: db}, Auth: authHTTP}.Register(app.Router()) server := &http.Server{Addr: cfg.HTTPAddress, Handler: app.Handler()} serverCtx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) defer stop() diff --git a/Bell/server/migrations/2026081202_event_receipt.sql b/Bell/server/migrations/2026081202_event_receipt.sql new file mode 100644 index 0000000..802a8ae --- /dev/null +++ b/Bell/server/migrations/2026081202_event_receipt.sql @@ -0,0 +1,29 @@ +CREATE TABLE bell_events ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + producer_id text NOT NULL, + source_event_id text NOT NULL, + event_type text NOT NULL, + occurred_at timestamptz NOT NULL, + location text NOT NULL, + severity text NOT NULL CHECK (severity IN ('low','medium','high','critical')), + evidence_ref text, + normalized_payload jsonb NOT NULL, + payload_sha256 bytea NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (producer_id, source_event_id) +); +CREATE INDEX bell_events_created_page_idx ON bell_events(created_at DESC,id DESC); +CREATE TABLE bell_event_receipts ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + event_id uuid NOT NULL UNIQUE REFERENCES bell_events(id), + producer_id text NOT NULL, + source_event_id text NOT NULL, + payload_sha256 bytea NOT NULL, + accepted_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (producer_id, source_event_id) +); + +CREATE OR REPLACE FUNCTION bell_reject_event_mutation() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN RAISE EXCEPTION 'Bell Event and Receipt facts are immutable'; END $$; +CREATE TRIGGER bell_event_no_update BEFORE UPDATE OR DELETE ON bell_events FOR EACH ROW EXECUTE FUNCTION bell_reject_event_mutation(); +CREATE TRIGGER bell_receipt_no_update BEFORE UPDATE OR DELETE ON bell_event_receipts FOR EACH ROW EXECUTE FUNCTION bell_reject_event_mutation(); diff --git a/Bell/web/src/api/event/index.js b/Bell/web/src/api/event/index.js new file mode 100644 index 0000000..209e6f0 --- /dev/null +++ b/Bell/web/src/api/event/index.js @@ -0,0 +1,3 @@ +import request from '../../bootstrap/request' +export const listEvents = params => request.get('/api/v1/events', { params }) +export const getEvent = id => request.get(`/api/v1/events/${id}`) diff --git a/Bell/web/src/bootstrap/router.js b/Bell/web/src/bootstrap/router.js index 492ce28..0deb729 100644 --- a/Bell/web/src/bootstrap/router.js +++ b/Bell/web/src/bootstrap/router.js @@ -3,6 +3,7 @@ import Dashboard from './Dashboard.vue' import Login from '../views/login/Login.vue' import Users from '../views/system/Users.vue' import Audit from '../views/system/Audit.vue' +import Events from '../views/event/Events.vue' import store from './store' const router = createRouter({ @@ -10,6 +11,7 @@ const router = createRouter({ routes: [ { path: '/login', name: 'login', component: Login, meta: { title: '登录', public: true } }, { path: '/', name: 'dashboard', component: Dashboard, meta: { title: '工作台', permission: 'dashboard:read' } }, + { path: '/events', name: 'events', component: Events, meta: { title: '事件查询', permission: 'events:read' } }, { path: '/system/users', name: 'users', component: Users, meta: { title: '用户与角色', permission: 'users:read' } }, { path: '/system/audit', name: 'audit', component: Audit, meta: { title: '认证审计', permission: 'audit:read' } } ] diff --git a/Bell/web/src/layout/AppLayout.vue b/Bell/web/src/layout/AppLayout.vue index 9c57d47..18498db 100644 --- a/Bell/web/src/layout/AppLayout.vue +++ b/Bell/web/src/layout/AppLayout.vue @@ -4,6 +4,7 @@
B
Bell预警中心
diff --git a/Bell/web/src/views/event/Events.vue b/Bell/web/src/views/event/Events.vue new file mode 100644 index 0000000..804c03e --- /dev/null +++ b/Bell/web/src/views/event/Events.vue @@ -0,0 +1,5 @@ + + +