From 34ebf31fdffc4c65eeec67cbb3e64f6dd79a7fe2 Mon Sep 17 00:00:00 2001 From: xarmian Date: Tue, 7 Apr 2026 01:13:44 +0000 Subject: [PATCH] fix: resolve Codex review findings across SSE, audit, metrics, and deployment - Make SSE limit checks atomic with subscription via SubscribeIfAllowed to prevent TOCTOU races where concurrent requests bypass connection caps - Replace fmt.Sprintf JSON assembly with json.Marshal (auditMeta helper) in all audit log call sites to prevent silent JSONB insert failures on PostgreSQL when metadata contains special characters - Pass database credentials via PGDATABASE env var instead of pg_dump/psql command-line args to avoid leaking passwords in ps/proc output - Fix audit-log query builder to rebind placeholders once after all filters are appended, preventing duplicate $1 placeholders on PostgreSQL - Replace per-workspace SSE GaugeVec with a single Gauge to avoid unbounded Prometheus label cardinality in multi-tenant deployments - Fix prod Docker Compose: override PAD_REDIS_URL with password and add authenticated Redis healthcheck when REDIS_PASSWORD is set --- cmd/pad/main.go | 12 +++++----- docker-compose.prod.yml | 13 ++++++++++ internal/events/bus.go | 34 +++++++++++++++++++++++++++ internal/events/redis_bus.go | 29 +++++++++++++++++++++++ internal/metrics/instrumented_bus.go | 23 +++++++++++++++--- internal/metrics/metrics.go | 12 +++++----- internal/metrics/metrics_test.go | 20 ++++++++-------- internal/server/handlers_admin.go | 5 ++-- internal/server/handlers_auth.go | 10 ++++---- internal/server/handlers_documents.go | 17 +++++++++++--- internal/server/handlers_events.go | 21 +++++++---------- internal/server/handlers_items.go | 5 ++-- internal/server/handlers_members.go | 9 ++++--- internal/server/handlers_tokens.go | 9 ++++--- internal/store/activities.go | 17 +++++++++----- 15 files changed, 171 insertions(+), 65 deletions(-) diff --git a/cmd/pad/main.go b/cmd/pad/main.go index 004b16b9..be61b347 100644 --- a/cmd/pad/main.go +++ b/cmd/pad/main.go @@ -5234,12 +5234,11 @@ For SQLite, simply copy the database file (default: ~/.pad/pad.db).`, output = fmt.Sprintf("pad-backup-%s.sql", time.Now().Format("20060102-150405")) } - // Use --dbname with the full URL so pg_dump inherits all connection - // parameters (sslmode, sslrootcert, timeouts, etc.). + // Pass the connection URL via environment variable instead of + // command-line args, so credentials don't leak in ps/proc output. // --clean emits DROP statements so the dump can be restored into an // existing database, and --if-exists avoids errors on a fresh DB. pgArgs := []string{ - "--dbname", dbURL, "--format", "plain", "--clean", "--if-exists", @@ -5247,6 +5246,7 @@ For SQLite, simply copy the database file (default: ~/.pad/pad.db).`, } pgCmd := exec.Command("pg_dump", pgArgs...) + pgCmd.Env = append(os.Environ(), "PGDATABASE="+dbURL) pgCmd.Stdout = os.Stdout pgCmd.Stderr = os.Stderr @@ -5317,15 +5317,15 @@ WARNING: This will overwrite the current database contents.`, } } - // Use --dbname with the full URL so psql inherits all connection - // parameters (sslmode, sslrootcert, timeouts, etc.). + // Pass the connection URL via environment variable instead of + // command-line args, so credentials don't leak in ps/proc output. psqlArgs := []string{ - "--dbname", dbURL, "--file", inputFile, "--single-transaction", } psqlCmd := exec.Command("psql", psqlArgs...) + psqlCmd.Env = append(os.Environ(), "PGDATABASE="+dbURL) psqlCmd.Stdout = os.Stdout psqlCmd.Stderr = os.Stderr diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index c568d7eb..4afdf8b8 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -12,6 +12,10 @@ services: pad: environment: PAD_SECURE_COOKIES: "true" + # Override Redis URL to include password when REDIS_PASSWORD is set. + # Without this, the pad container inherits the passwordless URL from + # docker-compose.yml and fails to connect when Redis AUTH is enabled. + PAD_REDIS_URL: "redis://:${REDIS_PASSWORD:-}@redis:6379" # Set your public-facing URL for correct invitation links: # PAD_URL: "https://pad.example.com" # CORS origins (comma-separated): @@ -50,6 +54,15 @@ services: redis: command: redis-server --maxmemory 128mb --maxmemory-policy allkeys-lru --requirepass "${REDIS_PASSWORD:-}" + healthcheck: + # Override the base healthcheck to authenticate when REDIS_PASSWORD is set. + # redis-cli reads REDISCLI_AUTH automatically for authentication. + test: ["CMD-SHELL", "REDISCLI_AUTH=$${REDIS_PASSWORD:-} redis-cli ping | grep -q PONG"] + interval: 5s + timeout: 3s + retries: 5 + environment: + REDIS_PASSWORD: "${REDIS_PASSWORD:-}" deploy: resources: limits: diff --git a/internal/events/bus.go b/internal/events/bus.go index a957555c..fe056703 100644 --- a/internal/events/bus.go +++ b/internal/events/bus.go @@ -54,6 +54,12 @@ type EventBus interface { // Returns a buffered channel that will receive events for that workspace. Subscribe(workspaceID string) chan Event + // SubscribeIfAllowed atomically checks the global and per-workspace + // subscriber limits and, only if both are satisfied, subscribes in the + // same critical section. Returns (ch, true) on success or (nil, false) + // when a limit would be exceeded. Pass 0 for either limit to disable it. + SubscribeIfAllowed(workspaceID string, maxGlobal, maxPerWorkspace int) (chan Event, bool) + // Unsubscribe removes a subscriber and closes its channel. Unsubscribe(ch chan Event) @@ -105,6 +111,34 @@ func (b *MemoryBus) Subscribe(workspaceID string) chan Event { return ch } +// SubscribeIfAllowed atomically checks limits and subscribes. +func (b *MemoryBus) SubscribeIfAllowed(workspaceID string, maxGlobal, maxPerWorkspace int) (chan Event, bool) { + b.mu.Lock() + defer b.mu.Unlock() + + if maxGlobal > 0 && len(b.subscribers) >= maxGlobal { + return nil, false + } + if maxPerWorkspace > 0 { + count := 0 + for _, sub := range b.subscribers { + if sub.workspaceID == workspaceID { + count++ + } + } + if count >= maxPerWorkspace { + return nil, false + } + } + + ch := make(chan Event, 64) + b.subscribers[ch] = &subscriber{ + ch: ch, + workspaceID: workspaceID, + } + return ch, true +} + // Unsubscribe removes a subscriber and closes its channel. func (b *MemoryBus) Unsubscribe(ch chan Event) { b.mu.Lock() diff --git a/internal/events/redis_bus.go b/internal/events/redis_bus.go index b6533a7c..09ece4cc 100644 --- a/internal/events/redis_bus.go +++ b/internal/events/redis_bus.go @@ -77,6 +77,35 @@ func (b *RedisBus) Subscribe(workspaceID string) chan Event { return ch } +// SubscribeIfAllowed atomically checks limits and subscribes. +// NOTE: Limits are enforced against local (per-pod) subscriber counts only. +// In multi-replica deployments the effective cap is multiplied by the number +// of replicas. For truly global caps, use a Redis-backed counter. +func (b *RedisBus) SubscribeIfAllowed(workspaceID string, maxGlobal, maxPerWorkspace int) (chan Event, bool) { + b.mu.Lock() + defer b.mu.Unlock() + + if maxGlobal > 0 && len(b.subscribers) >= maxGlobal { + return nil, false + } + if maxPerWorkspace > 0 && b.wsCounts[workspaceID] >= maxPerWorkspace { + return nil, false + } + + ch := make(chan Event, 64) + b.subscribers[ch] = &subscriber{ + ch: ch, + workspaceID: workspaceID, + } + + b.wsCounts[workspaceID]++ + if b.wsCounts[workspaceID] == 1 { + b.startRedisSubscription(workspaceID) + } + + return ch, true +} + // Unsubscribe removes a local subscriber and closes its channel. // Cancels the Redis subscription if this was the last local subscriber for the workspace. func (b *RedisBus) Unsubscribe(ch chan Event) { diff --git a/internal/metrics/instrumented_bus.go b/internal/metrics/instrumented_bus.go index c70e84aa..3c37cc55 100644 --- a/internal/metrics/instrumented_bus.go +++ b/internal/metrics/instrumented_bus.go @@ -35,15 +35,32 @@ func (b *InstrumentedBus) Subscribe(workspaceID string) chan events.Event { b.workspaces[ch] = workspaceID b.mu.Unlock() - b.metrics.SSEConnectionsActive.WithLabelValues(workspaceID).Inc() + (*b.metrics.SSEConnectionsActive).Inc() (*b.metrics.EventBusSubscribers).Set(float64(b.inner.SubscriberCount())) return ch } +// SubscribeIfAllowed delegates the atomic check-and-subscribe to the inner bus +// and updates Prometheus gauges on success. +func (b *InstrumentedBus) SubscribeIfAllowed(workspaceID string, maxGlobal, maxPerWorkspace int) (chan events.Event, bool) { + ch, ok := b.inner.SubscribeIfAllowed(workspaceID, maxGlobal, maxPerWorkspace) + if !ok { + return nil, false + } + + b.mu.Lock() + b.workspaces[ch] = workspaceID + b.mu.Unlock() + + (*b.metrics.SSEConnectionsActive).Inc() + (*b.metrics.EventBusSubscribers).Set(float64(b.inner.SubscriberCount())) + return ch, true +} + // Unsubscribe delegates to the inner bus and decrements the SSE connection gauge. func (b *InstrumentedBus) Unsubscribe(ch chan events.Event) { b.mu.Lock() - workspaceID, ok := b.workspaces[ch] + _, ok := b.workspaces[ch] if ok { delete(b.workspaces, ch) } @@ -52,7 +69,7 @@ func (b *InstrumentedBus) Unsubscribe(ch chan events.Event) { b.inner.Unsubscribe(ch) if ok { - b.metrics.SSEConnectionsActive.WithLabelValues(workspaceID).Dec() + (*b.metrics.SSEConnectionsActive).Dec() } (*b.metrics.EventBusSubscribers).Set(float64(b.inner.SubscriberCount())) } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index f1ac2da4..4065d394 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -19,8 +19,8 @@ type Metrics struct { HTTPRequestDuration *prometheus.HistogramVec HTTPResponseSize *prometheus.HistogramVec - // SSE connection metrics - SSEConnectionsActive *prometheus.GaugeVec + // SSE connection metrics (single gauge to avoid unbounded label cardinality) + SSEConnectionsActive *prometheus.Gauge // EventBus metrics EventBusPublishTotal *prometheus.Counter @@ -53,10 +53,10 @@ func New() *Metrics { Buckets: prometheus.ExponentialBuckets(100, 10, 7), // 100B to 100MB }, []string{"method", "route", "status"}) - sseConnectionsActive := prometheus.NewGaugeVec(prometheus.GaugeOpts{ + sseConnectionsActive := prometheus.NewGauge(prometheus.GaugeOpts{ Name: "pad_sse_connections_active", - Help: "Number of active SSE connections per workspace.", - }, []string{"workspace_id"}) + Help: "Total number of active SSE connections.", + }) eventBusPublishTotal := prometheus.NewCounter(prometheus.CounterOpts{ Name: "pad_eventbus_publish_total", @@ -82,7 +82,7 @@ func New() *Metrics { HTTPRequestsTotal: httpRequestsTotal, HTTPRequestDuration: httpRequestDuration, HTTPResponseSize: httpResponseSize, - SSEConnectionsActive: sseConnectionsActive, + SSEConnectionsActive: &sseConnectionsActive, EventBusPublishTotal: &eventBusPublishTotal, EventBusSubscribers: &eventBusSubscribers, } diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index b8f94165..649c1370 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -103,37 +103,37 @@ func TestInstrumentedBus_SubscribeUnsubscribe(t *testing.T) { t.Fatal("Subscribe should return a channel") } - // Check SSE gauge incremented - gauge := getGaugeValue(t, m.SSEConnectionsActive.WithLabelValues("ws-1")) + // Check SSE gauge incremented (single total gauge, not per-workspace) + gauge := getGaugeValue(t, *m.SSEConnectionsActive) if gauge != 1 { t.Errorf("Expected SSE active connections = 1, got %v", gauge) } // Subscribe a second connection to same workspace ch2 := bus.Subscribe("ws-1") - gauge = getGaugeValue(t, m.SSEConnectionsActive.WithLabelValues("ws-1")) + gauge = getGaugeValue(t, *m.SSEConnectionsActive) if gauge != 2 { t.Errorf("Expected SSE active connections = 2, got %v", gauge) } // Subscribe to a different workspace ch3 := bus.Subscribe("ws-2") - gauge2 := getGaugeValue(t, m.SSEConnectionsActive.WithLabelValues("ws-2")) - if gauge2 != 1 { - t.Errorf("Expected SSE active connections for ws-2 = 1, got %v", gauge2) + gauge = getGaugeValue(t, *m.SSEConnectionsActive) + if gauge != 3 { + t.Errorf("Expected SSE active connections = 3, got %v", gauge) } // Unsubscribe one from ws-1 bus.Unsubscribe(ch) - gauge = getGaugeValue(t, m.SSEConnectionsActive.WithLabelValues("ws-1")) - if gauge != 1 { - t.Errorf("Expected SSE active connections = 1 after unsubscribe, got %v", gauge) + gauge = getGaugeValue(t, *m.SSEConnectionsActive) + if gauge != 2 { + t.Errorf("Expected SSE active connections = 2 after unsubscribe, got %v", gauge) } // Unsubscribe remaining bus.Unsubscribe(ch2) bus.Unsubscribe(ch3) - gauge = getGaugeValue(t, m.SSEConnectionsActive.WithLabelValues("ws-1")) + gauge = getGaugeValue(t, *m.SSEConnectionsActive) if gauge != 0 { t.Errorf("Expected SSE active connections = 0 after all unsubscribed, got %v", gauge) } diff --git a/internal/server/handlers_admin.go b/internal/server/handlers_admin.go index 6728796e..d462b637 100644 --- a/internal/server/handlers_admin.go +++ b/internal/server/handlers_admin.go @@ -1,9 +1,9 @@ package server import ( + "encoding/json" "fmt" "net/http" - "strings" "github.com/xarmian/pad/internal/models" ) @@ -86,7 +86,8 @@ func (s *Server) handleUpdatePlatformSettings(w http.ResponseWriter, r *http.Req keys = append(keys, key) } } - s.logAuditEvent(models.ActionSettingsChanged, r, fmt.Sprintf(`{"keys":["%s"]}`, strings.Join(keys, `","`))) + keysJSON, _ := json.Marshal(keys) + s.logAuditEvent(models.ActionSettingsChanged, r, fmt.Sprintf(`{"keys":%s}`, keysJSON)) writeJSON(w, http.StatusOK, map[string]interface{}{"ok": true}) } diff --git a/internal/server/handlers_auth.go b/internal/server/handlers_auth.go index c369ce1b..fee394c9 100644 --- a/internal/server/handlers_auth.go +++ b/internal/server/handlers_auth.go @@ -162,7 +162,7 @@ func (s *Server) handleBootstrap(w http.ResponseWriter, r *http.Request) { return } - s.logAuditEventForUser(models.ActionBootstrap, r, user.ID, `{"email":"`+user.Email+`"}`) + s.logAuditEventForUser(models.ActionBootstrap, r, user.ID, auditMeta(map[string]string{"email": user.Email})) writeJSON(w, http.StatusCreated, map[string]interface{}{ "user": sessionUserPayload(user), @@ -276,7 +276,7 @@ func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) { return } - s.logAuditEventForUser(models.ActionRegister, r, user.ID, `{"email":"`+user.Email+`"}`) + s.logAuditEventForUser(models.ActionRegister, r, user.ID, auditMeta(map[string]string{"email": user.Email})) writeJSON(w, http.StatusCreated, map[string]interface{}{ "user": sessionUserPayload(user), @@ -314,7 +314,7 @@ func (s *Server) handleLogin(w http.ResponseWriter, r *http.Request) { if user == nil { // Slow down brute force attempts time.Sleep(500 * time.Millisecond) - s.logAuditEvent(models.ActionLoginFailed, r, `{"email":"`+input.Email+`"}`) + s.logAuditEvent(models.ActionLoginFailed, r, auditMeta(map[string]string{"email": input.Email})) writeError(w, http.StatusUnauthorized, "unauthorized", "Invalid email or password") return } @@ -324,7 +324,7 @@ func (s *Server) handleLogin(w http.ResponseWriter, r *http.Request) { return } - s.logAuditEventForUser(models.ActionLogin, r, user.ID, `{"email":"`+user.Email+`"}`) + s.logAuditEventForUser(models.ActionLogin, r, user.ID, auditMeta(map[string]string{"email": user.Email})) writeJSON(w, http.StatusOK, map[string]interface{}{ "user": sessionUserPayload(user), @@ -634,7 +634,7 @@ func (s *Server) handleResetPassword(w http.ResponseWriter, r *http.Request) { // Set CSRF cookie alongside the new session setCSRFCookie(w, int(webSessionTTL.Seconds()), s.secureCookies) - s.logAuditEventForUser(models.ActionPasswordReset, r, user.ID, `{"email":"`+user.Email+`"}`) + s.logAuditEventForUser(models.ActionPasswordReset, r, user.ID, auditMeta(map[string]string{"email": user.Email})) writeJSON(w, http.StatusOK, map[string]interface{}{ "ok": true, diff --git a/internal/server/handlers_documents.go b/internal/server/handlers_documents.go index 0590d202..f2b9052c 100644 --- a/internal/server/handlers_documents.go +++ b/internal/server/handlers_documents.go @@ -414,11 +414,12 @@ func agentMeta(r *http.Request, existingMeta string) string { return existingMeta } if existingMeta == "" || existingMeta == "{}" { - return fmt.Sprintf(`{"agent":"%s"}`, agentName) + return auditMeta(map[string]string{"agent": agentName}) } - // Merge: insert agent field into existing JSON + // Merge: insert agent field into existing JSON object if strings.HasPrefix(existingMeta, "{") { - return fmt.Sprintf(`{"agent":"%s",%s`, agentName, existingMeta[1:]) + agentJSON, _ := json.Marshal(agentName) + return fmt.Sprintf(`{"agent":%s,%s`, agentJSON, existingMeta[1:]) } return existingMeta } @@ -478,6 +479,16 @@ func (s *Server) logAuditEventForUser(action string, r *http.Request, userID str }) } +// auditMeta safely marshals a map to a JSON string for audit log metadata. +// Falls back to "{}" on marshal error so audit calls never break. +func auditMeta(kv map[string]string) string { + data, err := json.Marshal(kv) + if err != nil { + return "{}" + } + return string(data) +} + // logWorkspaceAuditEvent logs a workspace-scoped audit event (e.g. member invited). // Best-effort: errors are silently ignored. func (s *Server) logWorkspaceAuditEvent(workspaceID, action string, r *http.Request, metadata string) { diff --git a/internal/server/handlers_events.go b/internal/server/handlers_events.go index 90ad5308..681ac14c 100644 --- a/internal/server/handlers_events.go +++ b/internal/server/handlers_events.go @@ -47,20 +47,17 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { w.Header().Set("Connection", "keep-alive") w.Header().Set("X-Accel-Buffering", "no") // Disable nginx buffering - // Check SSE connection limits before subscribing - if s.sseMaxConnections > 0 && s.events.SubscriberCount() >= s.sseMaxConnections { - slog.Warn("SSE global connection limit reached", "current", s.events.SubscriberCount(), "max", s.sseMaxConnections) - writeError(w, http.StatusTooManyRequests, "sse_limit_exceeded", "Global SSE connection limit reached") + // Atomically check SSE connection limits and subscribe in one step. + // This prevents TOCTOU races where two concurrent requests both pass the + // limit check before either subscribes. + ch, ok := s.events.SubscribeIfAllowed(ws.ID, s.sseMaxConnections, s.sseMaxPerWorkspace) + if !ok { + slog.Warn("SSE connection limit reached", "workspace", ws.Slug, + "global_current", s.events.SubscriberCount(), "global_max", s.sseMaxConnections, + "ws_current", s.events.WorkspaceSubscriberCount(ws.ID), "ws_max", s.sseMaxPerWorkspace) + writeError(w, http.StatusTooManyRequests, "sse_limit_exceeded", "SSE connection limit reached") return } - if s.sseMaxPerWorkspace > 0 && s.events.WorkspaceSubscriberCount(ws.ID) >= s.sseMaxPerWorkspace { - slog.Warn("SSE per-workspace connection limit reached", "workspace", ws.Slug, "current", s.events.WorkspaceSubscriberCount(ws.ID), "max", s.sseMaxPerWorkspace) - writeError(w, http.StatusTooManyRequests, "sse_workspace_limit_exceeded", "Workspace SSE connection limit reached") - return - } - - // Subscribe to events for this workspace - ch := s.events.Subscribe(ws.ID) defer s.events.Unsubscribe(ch) // Log warning at 80% global capacity diff --git a/internal/server/handlers_items.go b/internal/server/handlers_items.go index 1e5b29f1..e55b355e 100644 --- a/internal/server/handlers_items.go +++ b/internal/server/handlers_items.go @@ -338,7 +338,8 @@ func (s *Server) handleUpdateItem(w http.ResponseWriter, r *http.Request) { } if input.Title != nil && *input.Title != item.Title { if meta == "" { - meta = fmt.Sprintf(`{"changes":"title: %s → %s"}`, item.Title, *input.Title) + titleChange := fmt.Sprintf("title: %s → %s", item.Title, *input.Title) + meta = fmt.Sprintf(`{"changes":%q}`, titleChange) } } // Track role and assignment changes @@ -566,7 +567,7 @@ func (s *Server) handleMoveItem(w http.ResponseWriter, r *http.Request) { // Log activity with metadata about the move actor, source := actorFromRequest(r) - moveMeta := fmt.Sprintf(`{"from_collection":"%s","to_collection":"%s"}`, sourceColl.Slug, targetColl.Slug) + moveMeta := auditMeta(map[string]string{"from_collection": sourceColl.Slug, "to_collection": targetColl.Slug}) s.logActivityWithMeta(workspaceID, moved.ID, "moved", r, moveMeta) // Publish events for both old and new collections diff --git a/internal/server/handlers_members.go b/internal/server/handlers_members.go index 6e42bea8..ee818406 100644 --- a/internal/server/handlers_members.go +++ b/internal/server/handlers_members.go @@ -2,7 +2,6 @@ package server import ( "context" - "fmt" "log/slog" "net/http" @@ -109,7 +108,7 @@ func (s *Server) handleInviteMember(w http.ResponseWriter, r *http.Request) { writeInternalError(w, err) return } - s.logWorkspaceAuditEvent(workspaceID, models.ActionMemberInvited, r, fmt.Sprintf(`{"email":"%s","role":"%s","added_directly":true}`, existingUser.Email, input.Role)) + s.logWorkspaceAuditEvent(workspaceID, models.ActionMemberInvited, r, auditMeta(map[string]string{"email": existingUser.Email, "role": input.Role, "added_directly": "true"})) writeJSON(w, http.StatusCreated, map[string]interface{}{ "added": true, "user_id": existingUser.ID, @@ -139,7 +138,7 @@ func (s *Server) handleInviteMember(w http.ResponseWriter, r *http.Request) { resp["join_url"] = joinURL } - s.logWorkspaceAuditEvent(workspaceID, models.ActionMemberInvited, r, fmt.Sprintf(`{"email":"%s","role":"%s"}`, input.Email, input.Role)) + s.logWorkspaceAuditEvent(workspaceID, models.ActionMemberInvited, r, auditMeta(map[string]string{"email": input.Email, "role": input.Role})) writeJSON(w, http.StatusCreated, resp) @@ -186,7 +185,7 @@ func (s *Server) handleRemoveMember(w http.ResponseWriter, r *http.Request) { return } - s.logWorkspaceAuditEvent(workspaceID, models.ActionMemberRemoved, r, fmt.Sprintf(`{"user_id":"%s"}`, userID)) + s.logWorkspaceAuditEvent(workspaceID, models.ActionMemberRemoved, r, auditMeta(map[string]string{"user_id": userID})) w.WriteHeader(http.StatusNoContent) } @@ -223,7 +222,7 @@ func (s *Server) handleUpdateMemberRole(w http.ResponseWriter, r *http.Request) return } - s.logWorkspaceAuditEvent(workspaceID, models.ActionRoleChanged, r, fmt.Sprintf(`{"user_id":"%s","role":"%s"}`, userID, input.Role)) + s.logWorkspaceAuditEvent(workspaceID, models.ActionRoleChanged, r, auditMeta(map[string]string{"user_id": userID, "role": input.Role})) writeJSON(w, http.StatusOK, map[string]interface{}{ "user_id": userID, diff --git a/internal/server/handlers_tokens.go b/internal/server/handlers_tokens.go index bd05b12f..fc6edfbc 100644 --- a/internal/server/handlers_tokens.go +++ b/internal/server/handlers_tokens.go @@ -2,7 +2,6 @@ package server import ( "database/sql" - "fmt" "net/http" "github.com/go-chi/chi/v5" @@ -38,7 +37,7 @@ func (s *Server) handleCreateToken(w http.ResponseWriter, r *http.Request) { return } - s.logAuditEvent(models.ActionTokenCreated, r, fmt.Sprintf(`{"name":"%s","workspace_id":"%s"}`, input.Name, input.WorkspaceID)) + s.logAuditEvent(models.ActionTokenCreated, r, auditMeta(map[string]string{"name": input.Name, "workspace_id": input.WorkspaceID})) writeJSON(w, http.StatusCreated, token) } @@ -79,7 +78,7 @@ func (s *Server) handleDeleteToken(w http.ResponseWriter, r *http.Request) { return } - s.logAuditEvent(models.ActionTokenRevoked, r, fmt.Sprintf(`{"token_id":"%s"}`, tokenID)) + s.logAuditEvent(models.ActionTokenRevoked, r, auditMeta(map[string]string{"token_id": tokenID})) w.WriteHeader(http.StatusNoContent) } @@ -131,7 +130,7 @@ func (s *Server) handleCreateUserToken(w http.ResponseWriter, r *http.Request) { return } - s.logAuditEvent(models.ActionTokenCreated, r, fmt.Sprintf(`{"name":"%s"}`, input.Name)) + s.logAuditEvent(models.ActionTokenCreated, r, auditMeta(map[string]string{"name": input.Name})) writeJSON(w, http.StatusCreated, token) } @@ -154,7 +153,7 @@ func (s *Server) handleDeleteUserToken(w http.ResponseWriter, r *http.Request) { return } - s.logAuditEvent(models.ActionTokenRevoked, r, fmt.Sprintf(`{"token_id":"%s"}`, tokenID)) + s.logAuditEvent(models.ActionTokenRevoked, r, auditMeta(map[string]string{"token_id": tokenID})) w.WriteHeader(http.StatusNoContent) } diff --git a/internal/store/activities.go b/internal/store/activities.go index 210a191d..eb7bf68a 100644 --- a/internal/store/activities.go +++ b/internal/store/activities.go @@ -244,29 +244,31 @@ func nilIfEmpty(s string) interface{} { // ListAuditLog returns activities matching the given audit log filters. // Supports filtering by action, actor, workspace, and date range. func (s *Store) ListAuditLog(params models.AuditLogParams) ([]models.Activity, error) { - query := s.q(` + // Build the full query with ? placeholders first, then rebind once at + // the end so PostgreSQL $1/$2/... numbering is correct across all filters. + query := ` SELECT a.id, COALESCE(a.workspace_id, ''), COALESCE(a.document_id, ''), a.action, a.actor, a.source, a.metadata, COALESCE(a.user_id, ''), a.created_at, COALESCE(u.name, ''), COALESCE(a.ip_address, ''), COALESCE(a.user_agent, '') FROM activities a LEFT JOIN users u ON a.user_id = u.id WHERE 1=1 - `) + ` args := []interface{}{} if params.WorkspaceID != "" { - query += s.q(` AND a.workspace_id = ?`) + query += ` AND a.workspace_id = ?` args = append(args, params.WorkspaceID) } if params.Action != "" { - query += s.q(` AND a.action = ?`) + query += ` AND a.action = ?` args = append(args, params.Action) } if params.Actor != "" { - query += s.q(` AND a.user_id = ?`) + query += ` AND a.user_id = ?` args = append(args, params.Actor) } if params.Days > 0 { cutoff := time.Now().UTC().AddDate(0, 0, -params.Days).Format(time.RFC3339) - query += s.q(` AND a.created_at >= ?`) + query += ` AND a.created_at >= ?` args = append(args, cutoff) } @@ -281,6 +283,9 @@ func (s *Store) ListAuditLog(params models.AuditLogParams) ([]models.Activity, e query += fmt.Sprintf(` OFFSET %d`, params.Offset) } + // Rebind all ? placeholders in one pass so $1, $2, ... are sequential. + query = s.q(query) + rows, err := s.db.Query(query, args...) if err != nil { return nil, err