From 9115169c03bf911ca826d3ec7512bba4d232230c Mon Sep 17 00:00:00 2001 From: viettranx Date: Thu, 12 Mar 2026 18:34:56 +0700 Subject: [PATCH] feat: expand audit logging via pub/sub event pattern Replace direct ActivityStore injection with event-driven audit system. Handlers emit audit events via msgBus.Broadcast(), a single subscriber with buffered channel persists to activity_logs table. Coverage expanded from 3 agent CRUD actions to ~65 audit points across all HTTP handlers and WebSocket RPC methods including agents, providers, skills, MCP servers, cron, sessions, teams, pairing, and more. --- cmd/gateway.go | 44 ++++++++++++++++++- cmd/gateway_channels_setup.go | 6 +-- cmd/gateway_http_handlers.go | 4 +- cmd/gateway_methods.go | 8 ++-- internal/bus/types.go | 18 +++++++- internal/gateway/methods/agent_links.go | 8 +++- internal/gateway/methods/agents.go | 6 ++- internal/gateway/methods/agents_create.go | 1 + internal/gateway/methods/agents_delete.go | 1 + internal/gateway/methods/agents_update.go | 1 + internal/gateway/methods/audit.go | 25 +++++++++++ internal/gateway/methods/channel_instances.go | 12 +++-- internal/gateway/methods/config.go | 2 + internal/gateway/methods/cron.go | 13 ++++-- internal/gateway/methods/exec_approval.go | 10 +++-- internal/gateway/methods/pairing.go | 3 ++ internal/gateway/methods/sessions.go | 9 +++- internal/gateway/methods/teams.go | 6 ++- internal/gateway/methods/teams_crud.go | 2 + internal/gateway/methods/teams_members.go | 2 + internal/http/agents.go | 25 ++--------- internal/http/agents_instances.go | 2 + internal/http/agents_sharing.go | 4 ++ internal/http/audit.go | 34 ++++++++++++++ internal/http/builtin_tools.go | 1 + internal/http/channel_instances.go | 3 ++ internal/http/custom_tools.go | 3 ++ internal/http/mcp.go | 3 ++ internal/http/mcp_grants.go | 4 ++ internal/http/mcp_requests.go | 2 + internal/http/oauth.go | 7 ++- internal/http/oauth_test.go | 2 +- internal/http/providers.go | 15 ++++++- internal/http/skills.go | 3 ++ internal/http/skills_grants.go | 4 ++ internal/http/skills_upload.go | 1 + pkg/protocol/events.go | 3 ++ 37 files changed, 241 insertions(+), 56 deletions(-) create mode 100644 internal/gateway/methods/audit.go create mode 100644 internal/http/audit.go diff --git a/cmd/gateway.go b/cmd/gateway.go index f2fd5495..945eaee7 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -655,7 +655,7 @@ func runGateway() { server.SetPolicyEngine(permPE) server.SetPairingService(pgStores.Pairing) server.SetMessageBus(msgBus) - server.SetOAuthHandler(httpapi.NewOAuthHandler(cfg.Gateway.Token, pgStores.Providers, pgStores.ConfigSecrets, providerRegistry)) + server.SetOAuthHandler(httpapi.NewOAuthHandler(cfg.Gateway.Token, pgStores.Providers, pgStores.ConfigSecrets, providerRegistry, msgBus)) // contextFileInterceptor is created inside wireExtras. // Declared here so it can be passed to registerAllMethods → AgentsMethods @@ -821,6 +821,43 @@ func runGateway() { // Wire channel event subscribers (cache invalidation, pairing, cascade disable) wireChannelEventSubscribers(msgBus, server, pgStores, channelMgr, instanceLoader, pairingMethods, cfg) + // Audit log subscriber — persists audit events to activity_logs table. + // Uses a buffered channel with a single worker to avoid unbounded goroutines. + var auditCh chan bus.AuditEventPayload + if pgStores.Activity != nil { + auditCh = make(chan bus.AuditEventPayload, 256) + msgBus.Subscribe(bus.TopicAudit, func(evt bus.Event) { + if evt.Name != protocol.EventAuditLog { + return + } + payload, ok := evt.Payload.(bus.AuditEventPayload) + if !ok { + return + } + select { + case auditCh <- payload: + default: + slog.Warn("audit.queue_full", "action", payload.Action) + } + }) + go func() { + for payload := range auditCh { + if err := pgStores.Activity.Log(context.Background(), &store.ActivityLog{ + ActorType: payload.ActorType, + ActorID: payload.ActorID, + Action: payload.Action, + EntityType: payload.EntityType, + EntityID: payload.EntityID, + IPAddress: payload.IPAddress, + Details: payload.Details, + }); err != nil { + slog.Warn("audit.log_failed", "action", payload.Action, "error", err) + } + } + }() + slog.Info("audit subscriber registered") + } + // Setup graceful shutdown ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -1001,6 +1038,11 @@ func runGateway() { channelMgr.StopAll(context.Background()) pgStores.Cron.Stop() + // Drain audit log queue before closing DB + if auditCh != nil { + close(auditCh) + } + // Close provider resources (e.g. Claude CLI temp files) providerRegistry.Close() diff --git a/cmd/gateway_channels_setup.go b/cmd/gateway_channels_setup.go index 6b2a8f36..d0825ab5 100644 --- a/cmd/gateway_channels_setup.go +++ b/cmd/gateway_channels_setup.go @@ -106,19 +106,19 @@ func wireChannelRPCMethods(server *gateway.Server, pgStores *store.Stores, chann // Register channel instances WS RPC methods if pgStores.ChannelInstances != nil { - methods.NewChannelInstancesMethods(pgStores.ChannelInstances, msgBus).Register(server.Router()) + methods.NewChannelInstancesMethods(pgStores.ChannelInstances, msgBus, msgBus).Register(server.Router()) zalomethods.NewQRMethods(pgStores.ChannelInstances, msgBus).Register(server.Router()) zalomethods.NewContactsMethods(pgStores.ChannelInstances).Register(server.Router()) } // Register agent links WS RPC methods if pgStores.AgentLinks != nil && pgStores.Agents != nil { - methods.NewAgentLinksMethods(pgStores.AgentLinks, pgStores.Agents, agentRouter, msgBus).Register(server.Router()) + methods.NewAgentLinksMethods(pgStores.AgentLinks, pgStores.Agents, agentRouter, msgBus, msgBus).Register(server.Router()) } // Register agent teams WS RPC methods if pgStores.Teams != nil { - methods.NewTeamsMethods(pgStores.Teams, pgStores.Agents, pgStores.AgentLinks, agentRouter, msgBus).Register(server.Router()) + methods.NewTeamsMethods(pgStores.Teams, pgStores.Agents, pgStores.AgentLinks, agentRouter, msgBus, msgBus).Register(server.Router()) } } diff --git a/cmd/gateway_http_handlers.go b/cmd/gateway_http_handlers.go index 9c00bdd6..b73b0baa 100644 --- a/cmd/gateway_http_handlers.go +++ b/cmd/gateway_http_handlers.go @@ -28,9 +28,6 @@ func wireHTTP(stores *store.Stores, token string, msgBus *bus.MessageBus, toolsR summoner = httpapi.NewAgentSummoner(stores.Agents, providerReg, msgBus) } agentsH = httpapi.NewAgentsHandler(stores.Agents, token, msgBus, summoner, isOwner) - if stores.Activity != nil { - agentsH.SetActivityStore(stores.Activity) - } } if stores != nil && stores.Skills != nil { @@ -60,6 +57,7 @@ func wireHTTP(stores *store.Stores, token string, msgBus *bus.MessageBus, toolsR if stores != nil && stores.Providers != nil { providersH = httpapi.NewProvidersHandler(stores.Providers, stores.ConfigSecrets, token, providerReg, gatewayAddr) + providersH.SetMessageBus(msgBus) if stores.MCP != nil { providersH.SetMCPServerLookup(buildMCPServerLookup(stores.MCP)) } diff --git a/cmd/gateway_methods.go b/cmd/gateway_methods.go index 2bf9365f..2883a176 100644 --- a/cmd/gateway_methods.go +++ b/cmd/gateway_methods.go @@ -17,15 +17,15 @@ func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore // Phase 1: Core methods methods.NewChatMethods(agents, sessStore, server.RateLimiter()).Register(router) - methods.NewAgentsMethods(agents, cfg, cfgPath, workspace, agentStore, contextFileInterceptor).Register(router) - methods.NewSessionsMethods(sessStore).Register(router) + methods.NewAgentsMethods(agents, cfg, cfgPath, workspace, agentStore, contextFileInterceptor, msgBus).Register(router) + methods.NewSessionsMethods(sessStore, msgBus).Register(router) methods.NewConfigMethods(cfg, cfgPath, configSecretsStore, msgBus).Register(router) // Phase 2: Skills (uses SkillStore interface — PG or File) methods.NewSkillsMethods(skillStore).Register(router) // Phase 2: Cron (store created externally, shared with gateway) - methods.NewCronMethods(cronStore).Register(router) + methods.NewCronMethods(cronStore, msgBus).Register(router) // Phase 2: Pairing (store created externally, shared with channel manager). // OnApprove callback is set later by the caller after channel manager is created. @@ -36,7 +36,7 @@ func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore methods.NewUsageMethods(sessStore).Register(router) // Phase 2: Exec approval (always registered — returns empty when manager is nil) - methods.NewExecApprovalMethods(execApprovalMgr).Register(router) + methods.NewExecApprovalMethods(execApprovalMgr, msgBus).Register(router) // Phase 2: Send (outbound message routing) methods.NewSendMethods(msgBus).Register(router) diff --git a/internal/bus/types.go b/internal/bus/types.go index 54a8a2e3..981094d3 100644 --- a/internal/bus/types.go +++ b/internal/bus/types.go @@ -1,6 +1,9 @@ package bus -import "context" +import ( + "context" + "encoding/json" +) // MediaFile represents an inbound media file with its MIME type. // Used throughout the media pipeline to preserve content type from channel download to storage. @@ -77,6 +80,7 @@ const ( TopicCacheGroupFileWriters = "cache:group_file_writers" TopicCacheSkillGrants = "cache:skill_grants" TopicCacheMCP = "cache:mcp" + TopicAudit = "audit" TopicChannelStreaming = "channel-streaming" TopicConfigChanged = "config:changed" TopicPairingRevoked = "pairing:revoked" @@ -102,6 +106,18 @@ type AgentStatusChangedPayload struct { NewStatus string `json:"new_status"` } +// AuditEventPayload carries audit log data emitted by handlers. +// A single subscriber persists these to the activity_logs table. +type AuditEventPayload struct { + ActorType string `json:"actor_type"` + ActorID string `json:"actor_id"` + Action string `json:"action"` + EntityType string `json:"entity_type"` + EntityID string `json:"entity_id"` + IPAddress string `json:"ip_address,omitempty"` + Details json.RawMessage `json:"details,omitempty"` +} + // CacheInvalidatePayload signals cache layers to evict stale entries. // Used with protocol.EventCacheInvalidate events. type CacheInvalidatePayload struct { diff --git a/internal/gateway/methods/agent_links.go b/internal/gateway/methods/agent_links.go index be110238..b040e33b 100644 --- a/internal/gateway/methods/agent_links.go +++ b/internal/gateway/methods/agent_links.go @@ -20,10 +20,11 @@ type AgentLinksMethods struct { agentStore store.AgentStore agentRouter *agent.Router // for cache invalidation when links change msgBus *bus.MessageBus // for pub/sub cache invalidation + eventBus bus.EventPublisher } -func NewAgentLinksMethods(linkStore store.AgentLinkStore, agentStore store.AgentStore, agentRouter *agent.Router, msgBus *bus.MessageBus) *AgentLinksMethods { - return &AgentLinksMethods{linkStore: linkStore, agentStore: agentStore, agentRouter: agentRouter, msgBus: msgBus} +func NewAgentLinksMethods(linkStore store.AgentLinkStore, agentStore store.AgentStore, agentRouter *agent.Router, msgBus *bus.MessageBus, eventBus bus.EventPublisher) *AgentLinksMethods { + return &AgentLinksMethods{linkStore: linkStore, agentStore: agentStore, agentRouter: agentRouter, msgBus: msgBus, eventBus: eventBus} } func (m *AgentLinksMethods) Register(router *gateway.MethodRouter) { @@ -181,6 +182,7 @@ func (m *AgentLinksMethods) handleCreate(ctx context.Context, client *gateway.Cl m.agentRouter.InvalidateAgent(targetAgent.AgentKey) } m.emitTeamCacheInvalidate() + emitAudit(m.eventBus, client, "agent_link.created", "agent_link", link.ID.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "link": link, @@ -280,6 +282,7 @@ func (m *AgentLinksMethods) handleUpdate(ctx context.Context, client *gateway.Cl // Invalidate affected agents so AGENTS.md gets regenerated m.invalidateLinkAgents(ctx, linkID) m.emitTeamCacheInvalidate() + emitAudit(m.eventBus, client, "agent_link.updated", "agent_link", linkID.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"ok": true})) @@ -349,6 +352,7 @@ func (m *AgentLinksMethods) handleDelete(ctx context.Context, client *gateway.Cl m.invalidateLinkAgentsByID(ctx, link.SourceAgentID, link.TargetAgentID) } m.emitTeamCacheInvalidate() + emitAudit(m.eventBus, client, "agent_link.deleted", "agent_link", linkID.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"ok": true})) diff --git a/internal/gateway/methods/agents.go b/internal/gateway/methods/agents.go index b9729dfc..dca94bac 100644 --- a/internal/gateway/methods/agents.go +++ b/internal/gateway/methods/agents.go @@ -7,6 +7,7 @@ import ( "slices" "github.com/nextlevelbuilder/goclaw/internal/agent" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/i18n" @@ -24,10 +25,11 @@ type AgentsMethods struct { workspace string agentStore store.AgentStore interceptor *tools.ContextFileInterceptor // invalidated on file writes + eventBus bus.EventPublisher } -func NewAgentsMethods(agents *agent.Router, cfg *config.Config, cfgPath, workspace string, agentStore store.AgentStore, interceptor *tools.ContextFileInterceptor) *AgentsMethods { - return &AgentsMethods{agents: agents, cfg: cfg, cfgPath: cfgPath, workspace: workspace, agentStore: agentStore, interceptor: interceptor} +func NewAgentsMethods(agents *agent.Router, cfg *config.Config, cfgPath, workspace string, agentStore store.AgentStore, interceptor *tools.ContextFileInterceptor, eventBus bus.EventPublisher) *AgentsMethods { + return &AgentsMethods{agents: agents, cfg: cfg, cfgPath: cfgPath, workspace: workspace, agentStore: agentStore, interceptor: interceptor, eventBus: eventBus} } // isOwnerUser checks if the given user ID is in the configured owner IDs. diff --git a/internal/gateway/methods/agents_create.go b/internal/gateway/methods/agents_create.go index cefcaaf5..dd19758f 100644 --- a/internal/gateway/methods/agents_create.go +++ b/internal/gateway/methods/agents_create.go @@ -163,4 +163,5 @@ func (m *AgentsMethods) handleCreate(ctx context.Context, client *gateway.Client "name": params.Name, "workspace": ws, })) + emitAudit(m.eventBus, client, "agent.created", "agent", agentID) } diff --git a/internal/gateway/methods/agents_delete.go b/internal/gateway/methods/agents_delete.go index 4e33c78f..039ed9ef 100644 --- a/internal/gateway/methods/agents_delete.go +++ b/internal/gateway/methods/agents_delete.go @@ -97,4 +97,5 @@ func (m *AgentsMethods) handleDelete(ctx context.Context, client *gateway.Client "agentId": params.AgentID, "removedBindings": removedBindings, })) + emitAudit(m.eventBus, client, "agent.deleted", "agent", params.AgentID) } diff --git a/internal/gateway/methods/agents_update.go b/internal/gateway/methods/agents_update.go index 34a575cf..6b7997a2 100644 --- a/internal/gateway/methods/agents_update.go +++ b/internal/gateway/methods/agents_update.go @@ -159,4 +159,5 @@ func (m *AgentsMethods) handleUpdate(ctx context.Context, client *gateway.Client "ok": true, "agentId": params.AgentID, })) + emitAudit(m.eventBus, client, "agent.updated", "agent", params.AgentID) } diff --git a/internal/gateway/methods/audit.go b/internal/gateway/methods/audit.go new file mode 100644 index 00000000..bdb3434d --- /dev/null +++ b/internal/gateway/methods/audit.go @@ -0,0 +1,25 @@ +package methods + +import ( + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/gateway" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +// emitAudit broadcasts an audit event via eventBus for async persistence. +func emitAudit(pub bus.EventPublisher, client *gateway.Client, action, entityType, entityID string) { + if pub == nil { + return + } + pub.Broadcast(bus.Event{ + Name: protocol.EventAuditLog, + Payload: bus.AuditEventPayload{ + ActorType: "user", + ActorID: client.UserID(), + Action: action, + EntityType: entityType, + EntityID: entityID, + IPAddress: client.RemoteAddr(), + }, + }) +} diff --git a/internal/gateway/methods/channel_instances.go b/internal/gateway/methods/channel_instances.go index 6a54b6ed..7f50ce98 100644 --- a/internal/gateway/methods/channel_instances.go +++ b/internal/gateway/methods/channel_instances.go @@ -16,13 +16,14 @@ import ( // ChannelInstancesMethods handles channel instance CRUD via WebSocket RPC. type ChannelInstancesMethods struct { - store store.ChannelInstanceStore - msgBus *bus.MessageBus + store store.ChannelInstanceStore + msgBus *bus.MessageBus + eventBus bus.EventPublisher } // NewChannelInstancesMethods creates a new handler for channel instance management. -func NewChannelInstancesMethods(s store.ChannelInstanceStore, msgBus *bus.MessageBus) *ChannelInstancesMethods { - return &ChannelInstancesMethods{store: s, msgBus: msgBus} +func NewChannelInstancesMethods(s store.ChannelInstanceStore, msgBus *bus.MessageBus, eventBus bus.EventPublisher) *ChannelInstancesMethods { + return &ChannelInstancesMethods{store: s, msgBus: msgBus, eventBus: eventBus} } // Register registers all channel instance RPC methods. @@ -141,6 +142,7 @@ func (m *ChannelInstancesMethods) handleCreate(ctx context.Context, client *gate } m.emitCacheInvalidate() + emitAudit(m.eventBus, client, "channel_instance.created", "channel_instance", inst.ID.String()) client.SendResponse(protocol.NewOKResponse(req.ID, maskInstance(*inst))) } @@ -173,6 +175,7 @@ func (m *ChannelInstancesMethods) handleUpdate(ctx context.Context, client *gate } m.emitCacheInvalidate() + emitAudit(m.eventBus, client, "channel_instance.updated", "channel_instance", id.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"status": "updated"})) } @@ -209,6 +212,7 @@ func (m *ChannelInstancesMethods) handleDelete(ctx context.Context, client *gate } m.emitCacheInvalidate() + emitAudit(m.eventBus, client, "channel_instance.deleted", "channel_instance", id.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"status": "deleted"})) } diff --git a/internal/gateway/methods/config.go b/internal/gateway/methods/config.go index eb9e38e9..d8dcf46a 100644 --- a/internal/gateway/methods/config.go +++ b/internal/gateway/methods/config.go @@ -92,6 +92,7 @@ func (m *ConfigMethods) handleApply(ctx context.Context, client *gateway.Client, } m.cfg.ApplyEnvOverrides() m.broadcastChanged() + emitAudit(m.eventBus, client, "config.applied", "config", "gateway") client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "ok": true, @@ -164,6 +165,7 @@ func (m *ConfigMethods) handlePatch(ctx context.Context, client *gateway.Client, } m.cfg.ApplyEnvOverrides() m.broadcastChanged() + emitAudit(m.eventBus, client, "config.patched", "config", "gateway") client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "ok": true, diff --git a/internal/gateway/methods/cron.go b/internal/gateway/methods/cron.go index 75aacd69..92d6b7ac 100644 --- a/internal/gateway/methods/cron.go +++ b/internal/gateway/methods/cron.go @@ -6,6 +6,7 @@ import ( "log/slog" "regexp" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -16,11 +17,12 @@ var cronSlugRe = regexp.MustCompile(`^[a-z0-9]([a-z0-9-]*[a-z0-9])?$`) // CronMethods handles cron.list, cron.create, cron.update, cron.delete, cron.toggle. type CronMethods struct { - service store.CronStore + service store.CronStore + eventBus bus.EventPublisher } -func NewCronMethods(service store.CronStore) *CronMethods { - return &CronMethods{service: service} +func NewCronMethods(service store.CronStore, eventBus bus.EventPublisher) *CronMethods { + return &CronMethods{service: service, eventBus: eventBus} } func (m *CronMethods) Register(router *gateway.MethodRouter) { @@ -87,6 +89,7 @@ func (m *CronMethods) handleCreate(ctx context.Context, client *gateway.Client, client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "job": job, })) + emitAudit(m.eventBus, client, "cron.created", "cron", job.ID) } func (m *CronMethods) handleDelete(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { @@ -111,6 +114,7 @@ func (m *CronMethods) handleDelete(ctx context.Context, client *gateway.Client, client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "deleted": true, })) + emitAudit(m.eventBus, client, "cron.deleted", "cron", params.JobID) } func (m *CronMethods) handleToggle(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { @@ -137,6 +141,7 @@ func (m *CronMethods) handleToggle(ctx context.Context, client *gateway.Client, "jobId": params.JobID, "enabled": params.Enabled, })) + emitAudit(m.eventBus, client, "cron.toggled", "cron", params.JobID) } func (m *CronMethods) handleStatus(_ context.Context, client *gateway.Client, req *protocol.RequestFrame) { @@ -172,6 +177,7 @@ func (m *CronMethods) handleUpdate(ctx context.Context, client *gateway.Client, client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "job": job, })) + emitAudit(m.eventBus, client, "cron.updated", "cron", jobID) } func (m *CronMethods) handleRun(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { @@ -208,6 +214,7 @@ func (m *CronMethods) handleRun(ctx context.Context, client *gateway.Client, req "ok": true, "ran": true, })) + emitAudit(m.eventBus, client, "cron.run", "cron", jobID) go func() { if _, _, err := m.service.RunJob(jobID, force); err != nil { diff --git a/internal/gateway/methods/exec_approval.go b/internal/gateway/methods/exec_approval.go index b9658ecc..186f460b 100644 --- a/internal/gateway/methods/exec_approval.go +++ b/internal/gateway/methods/exec_approval.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -13,11 +14,12 @@ import ( // ExecApprovalMethods handles exec.approval.list, exec.approval.approve, exec.approval.deny. type ExecApprovalMethods struct { - manager *tools.ExecApprovalManager + manager *tools.ExecApprovalManager + eventBus bus.EventPublisher } -func NewExecApprovalMethods(manager *tools.ExecApprovalManager) *ExecApprovalMethods { - return &ExecApprovalMethods{manager: manager} +func NewExecApprovalMethods(manager *tools.ExecApprovalManager, eventBus bus.EventPublisher) *ExecApprovalMethods { + return &ExecApprovalMethods{manager: manager, eventBus: eventBus} } func (m *ExecApprovalMethods) Register(router *gateway.MethodRouter) { @@ -91,6 +93,7 @@ func (m *ExecApprovalMethods) handleApprove(ctx context.Context, client *gateway "resolved": true, "decision": string(decision), })) + emitAudit(m.eventBus, client, "exec.approved", "exec", params.ID) } func (m *ExecApprovalMethods) handleDeny(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { @@ -121,4 +124,5 @@ func (m *ExecApprovalMethods) handleDeny(ctx context.Context, client *gateway.Cl "resolved": true, "decision": "deny", })) + emitAudit(m.eventBus, client, "exec.denied", "exec", params.ID) } diff --git a/internal/gateway/methods/pairing.go b/internal/gateway/methods/pairing.go index c39590bc..157921c2 100644 --- a/internal/gateway/methods/pairing.go +++ b/internal/gateway/methods/pairing.go @@ -113,6 +113,7 @@ func (m *PairingMethods) handleApprove(ctx context.Context, client *gateway.Clie m.broadcaster(*protocol.NewEvent(protocol.EventDevicePairRes, map[string]any{"action": "approved"})) } + emitAudit(m.msgBus, client, "pairing.approved", "pairing", params.Code) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "paired": paired, })) @@ -141,6 +142,7 @@ func (m *PairingMethods) handleDeny(ctx context.Context, client *gateway.Client, m.broadcaster(*protocol.NewEvent(protocol.EventDevicePairRes, map[string]any{"action": "denied"})) } + emitAudit(m.msgBus, client, "pairing.denied", "pairing", params.Code) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "denied": true, })) @@ -191,6 +193,7 @@ func (m *PairingMethods) handleRevoke(ctx context.Context, client *gateway.Clien }) } + emitAudit(m.msgBus, client, "pairing.revoked", "pairing", params.SenderID) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "revoked": true, })) diff --git a/internal/gateway/methods/sessions.go b/internal/gateway/methods/sessions.go index 119705eb..3f5f3b96 100644 --- a/internal/gateway/methods/sessions.go +++ b/internal/gateway/methods/sessions.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -13,10 +14,11 @@ import ( // SessionsMethods handles sessions.list, sessions.preview, sessions.patch, sessions.delete, sessions.reset. type SessionsMethods struct { sessions store.SessionStore + eventBus bus.EventPublisher } -func NewSessionsMethods(sess store.SessionStore) *SessionsMethods { - return &SessionsMethods{sessions: sess} +func NewSessionsMethods(sess store.SessionStore, eventBus bus.EventPublisher) *SessionsMethods { + return &SessionsMethods{sessions: sess, eventBus: eventBus} } func (m *SessionsMethods) Register(router *gateway.MethodRouter) { @@ -120,6 +122,7 @@ func (m *SessionsMethods) handlePatch(ctx context.Context, client *gateway.Clien "ok": true, "key": params.Key, })) + emitAudit(m.eventBus, client, "session.patched", "session", params.Key) } func (m *SessionsMethods) handleDelete(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { @@ -138,6 +141,7 @@ func (m *SessionsMethods) handleDelete(ctx context.Context, client *gateway.Clie client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "ok": true, })) + emitAudit(m.eventBus, client, "session.deleted", "session", params.Key) } func (m *SessionsMethods) handleReset(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { @@ -153,4 +157,5 @@ func (m *SessionsMethods) handleReset(ctx context.Context, client *gateway.Clien client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "ok": true, })) + emitAudit(m.eventBus, client, "session.reset", "session", params.Key) } diff --git a/internal/gateway/methods/teams.go b/internal/gateway/methods/teams.go index 5b7557ce..752a071c 100644 --- a/internal/gateway/methods/teams.go +++ b/internal/gateway/methods/teams.go @@ -20,10 +20,11 @@ type TeamsMethods struct { linkStore store.AgentLinkStore // for auto-creating bidirectional links agentRouter *agent.Router // for cache invalidation msgBus *bus.MessageBus // for pub/sub cache invalidation + eventBus bus.EventPublisher } -func NewTeamsMethods(teamStore store.TeamStore, agentStore store.AgentStore, linkStore store.AgentLinkStore, agentRouter *agent.Router, msgBus *bus.MessageBus) *TeamsMethods { - return &TeamsMethods{teamStore: teamStore, agentStore: agentStore, linkStore: linkStore, agentRouter: agentRouter, msgBus: msgBus} +func NewTeamsMethods(teamStore store.TeamStore, agentStore store.AgentStore, linkStore store.AgentLinkStore, agentRouter *agent.Router, msgBus *bus.MessageBus, eventBus bus.EventPublisher) *TeamsMethods { + return &TeamsMethods{teamStore: teamStore, agentStore: agentStore, linkStore: linkStore, agentRouter: agentRouter, msgBus: msgBus, eventBus: eventBus} } // emitTeamCacheInvalidate broadcasts a cache invalidation event for team data. @@ -159,6 +160,7 @@ func (m *TeamsMethods) handleCreate(ctx context.Context, client *gateway.Client, // Invalidate agent + team tool caches so TEAM.md gets injected m.invalidateTeamCaches(ctx, team.ID) + emitAudit(m.eventBus, client, "team.created", "team", team.ID.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "team": team, })) diff --git a/internal/gateway/methods/teams_crud.go b/internal/gateway/methods/teams_crud.go index daedca73..87c3b82f 100644 --- a/internal/gateway/methods/teams_crud.go +++ b/internal/gateway/methods/teams_crud.go @@ -107,6 +107,7 @@ func (m *TeamsMethods) handleDelete(ctx context.Context, client *gateway.Client, } } + emitAudit(m.eventBus, client, "team.deleted", "team", teamID.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"ok": true})) // Emit team.deleted event @@ -224,6 +225,7 @@ func (m *TeamsMethods) handleUpdate(ctx context.Context, client *gateway.Client, } m.invalidateTeamCaches(ctx, teamID) + emitAudit(m.eventBus, client, "team.updated", "team", teamID.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"ok": true})) diff --git a/internal/gateway/methods/teams_members.go b/internal/gateway/methods/teams_members.go index 41b63017..b58e320d 100644 --- a/internal/gateway/methods/teams_members.go +++ b/internal/gateway/methods/teams_members.go @@ -93,6 +93,7 @@ func (m *TeamsMethods) handleAddMember(ctx context.Context, client *gateway.Clie // Invalidate caches for all team members m.invalidateTeamCaches(ctx, teamID) + emitAudit(m.eventBus, client, "team.member_added", "team", teamID.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"ok": true})) @@ -183,6 +184,7 @@ func (m *TeamsMethods) handleRemoveMember(ctx context.Context, client *gateway.C } } + emitAudit(m.eventBus, client, "team.member_removed", "team", teamID.String()) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"ok": true})) // Emit team.member.removed event diff --git a/internal/http/agents.go b/internal/http/agents.go index 235b14a5..245c89a3 100644 --- a/internal/http/agents.go +++ b/internal/http/agents.go @@ -23,12 +23,8 @@ type AgentsHandler struct { msgBus *bus.MessageBus // for cache invalidation events (nil = no events) summoner *AgentSummoner // LLM-based agent setup (nil = disabled) isOwner func(string) bool // checks if user ID is a system owner (nil = no owners configured) - activity store.ActivityStore // optional audit logging (nil = disabled) } -// SetActivityStore sets the optional activity audit store. -func (h *AgentsHandler) SetActivityStore(a store.ActivityStore) { h.activity = a } - // NewAgentsHandler creates a handler for agent management endpoints. // isOwner is a function that checks if a user ID is in GOCLAW_OWNER_IDS (nil = disabled). func NewAgentsHandler(agents store.AgentStore, token string, msgBus *bus.MessageBus, summoner *AgentSummoner, isOwner func(string) bool) *AgentsHandler { @@ -51,21 +47,6 @@ func (h *AgentsHandler) emitCacheInvalidate(kind, key string) { }) } -// logActivity records an audit log entry if the activity store is set. -func (h *AgentsHandler) logActivity(r *http.Request, actorID, action, entityType, entityID string) { - if h.activity == nil { - return - } - h.activity.Log(r.Context(), &store.ActivityLog{ - ActorType: "user", - ActorID: actorID, - Action: action, - EntityType: entityType, - EntityID: entityID, - IPAddress: r.RemoteAddr, - }) -} - // RegisterRoutes registers all agent management routes on the given mux. func (h *AgentsHandler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("GET /v1/agents", h.authMiddleware(h.handleList)) @@ -203,7 +184,7 @@ func (h *AgentsHandler) handleCreate(w http.ResponseWriter, r *http.Request) { go h.summoner.SummonAgent(req.ID, req.Provider, req.Model, description) } - h.logActivity(r, userID, "agent.created", "agent", req.ID.String()) + emitAudit(h.msgBus, r, "agent.created", "agent", req.ID.String()) writeJSON(w, http.StatusCreated, req) } @@ -299,7 +280,7 @@ func (h *AgentsHandler) handleUpdate(w http.ResponseWriter, r *http.Request) { } } - h.logActivity(r, userID, "agent.updated", "agent", id.String()) + emitAudit(h.msgBus, r, "agent.updated", "agent", id.String()) writeJSON(w, http.StatusOK, map[string]string{"ok": "true"}) } @@ -332,6 +313,6 @@ func (h *AgentsHandler) handleDelete(w http.ResponseWriter, r *http.Request) { h.emitCacheInvalidate(bus.CacheKindAgent, ag.AgentKey) h.emitCacheInvalidate(bus.CacheKindBootstrap, id.String()) - h.logActivity(r, userID, "agent.deleted", "agent", id.String()) + emitAudit(h.msgBus, r, "agent.deleted", "agent", id.String()) writeJSON(w, http.StatusOK, map[string]string{"ok": "true"}) } diff --git a/internal/http/agents_instances.go b/internal/http/agents_instances.go index 659893dd..feed0f71 100644 --- a/internal/http/agents_instances.go +++ b/internal/http/agents_instances.go @@ -134,6 +134,7 @@ func (h *AgentsHandler) handleSetInstanceFile(w http.ResponseWriter, r *http.Req // Invalidate caches so the agent picks up the change immediately h.emitCacheInvalidate(bus.CacheKindBootstrap, id.String()) + emitAudit(h.msgBus, r, "agent_instance.file_set", "agent_instance", id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "updated"}) } @@ -184,5 +185,6 @@ func (h *AgentsHandler) handleUpdateInstanceMetadata(w http.ResponseWriter, r *h return } + emitAudit(h.msgBus, r, "agent_instance.metadata_updated", "agent_instance", id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "updated"}) } diff --git a/internal/http/agents_sharing.go b/internal/http/agents_sharing.go index 70b513dd..3c9e1a94 100644 --- a/internal/http/agents_sharing.go +++ b/internal/http/agents_sharing.go @@ -84,6 +84,7 @@ func (h *AgentsHandler) handleShare(w http.ResponseWriter, r *http.Request) { return } + emitAudit(h.msgBus, r, "agent.shared", "agent", id.String()) writeJSON(w, http.StatusCreated, map[string]string{"ok": "true"}) } @@ -117,6 +118,7 @@ func (h *AgentsHandler) handleRevokeShare(w http.ResponseWriter, r *http.Request return } + emitAudit(h.msgBus, r, "agent.share_revoked", "agent", id.String()) writeJSON(w, http.StatusOK, map[string]string{"ok": "true"}) } @@ -168,6 +170,7 @@ func (h *AgentsHandler) handleRegenerate(w http.ResponseWriter, r *http.Request) go h.summoner.RegenerateAgent(id, ag.Provider, ag.Model, req.Prompt) + emitAudit(h.msgBus, r, "agent.regenerated", "agent", id.String()) writeJSON(w, http.StatusAccepted, map[string]string{"ok": "true", "status": store.AgentStatusSummoning}) } @@ -213,6 +216,7 @@ func (h *AgentsHandler) handleResummon(w http.ResponseWriter, r *http.Request) { go h.summoner.SummonAgent(id, ag.Provider, ag.Model, description) + emitAudit(h.msgBus, r, "agent.resummoned", "agent", id.String()) writeJSON(w, http.StatusAccepted, map[string]string{"ok": "true", "status": store.AgentStatusSummoning}) } diff --git a/internal/http/audit.go b/internal/http/audit.go new file mode 100644 index 00000000..9de66e6e --- /dev/null +++ b/internal/http/audit.go @@ -0,0 +1,34 @@ +package http + +import ( + "net/http" + + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +// emitAudit broadcasts an audit event via msgBus for async persistence. +func emitAudit(msgBus *bus.MessageBus, r *http.Request, action, entityType, entityID string) { + if msgBus == nil { + return + } + actorID := store.UserIDFromContext(r.Context()) + if actorID == "" { + actorID = extractUserID(r) + } + if actorID == "" { + actorID = "system" + } + msgBus.Broadcast(bus.Event{ + Name: protocol.EventAuditLog, + Payload: bus.AuditEventPayload{ + ActorType: "user", + ActorID: actorID, + Action: action, + EntityType: entityType, + EntityID: entityID, + IPAddress: r.RemoteAddr, + }, + }) +} diff --git a/internal/http/builtin_tools.go b/internal/http/builtin_tools.go index 0b19fcf1..487ecd97 100644 --- a/internal/http/builtin_tools.go +++ b/internal/http/builtin_tools.go @@ -116,6 +116,7 @@ func (h *BuiltinToolsHandler) handleUpdate(w http.ResponseWriter, r *http.Reques return } + emitAudit(h.msgBus, r, "builtin_tool.updated", "builtin_tool", name) h.emitCacheInvalidate(name) writeJSON(w, http.StatusOK, map[string]string{"status": "updated"}) } diff --git a/internal/http/channel_instances.go b/internal/http/channel_instances.go index 367738a4..ee40fe7c 100644 --- a/internal/http/channel_instances.go +++ b/internal/http/channel_instances.go @@ -181,6 +181,7 @@ func (h *ChannelInstancesHandler) handleCreate(w http.ResponseWriter, r *http.Re } h.emitCacheInvalidate() + emitAudit(h.msgBus, r, "channel_instance.created", "channel_instance", inst.ID.String()) writeJSON(w, http.StatusCreated, maskInstanceHTTP(*inst)) } @@ -222,6 +223,7 @@ func (h *ChannelInstancesHandler) handleUpdate(w http.ResponseWriter, r *http.Re } h.emitCacheInvalidate() + emitAudit(h.msgBus, r, "channel_instance.updated", "channel_instance", id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "updated"}) } @@ -251,6 +253,7 @@ func (h *ChannelInstancesHandler) handleDelete(w http.ResponseWriter, r *http.Re } h.emitCacheInvalidate() + emitAudit(h.msgBus, r, "channel_instance.deleted", "channel_instance", id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"}) } diff --git a/internal/http/custom_tools.go b/internal/http/custom_tools.go index 6a28abae..d3900aaf 100644 --- a/internal/http/custom_tools.go +++ b/internal/http/custom_tools.go @@ -152,6 +152,7 @@ func (h *CustomToolsHandler) handleCreate(w http.ResponseWriter, r *http.Request return } + emitAudit(h.msgBus, r, "custom_tool.created", "custom_tool", def.ID.String()) h.emitCacheInvalidate(def.ID.String()) writeJSON(w, http.StatusCreated, def) } @@ -200,6 +201,7 @@ func (h *CustomToolsHandler) handleUpdate(w http.ResponseWriter, r *http.Request return } + emitAudit(h.msgBus, r, "custom_tool.updated", "custom_tool", id.String()) h.emitCacheInvalidate(id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "updated"}) } @@ -218,6 +220,7 @@ func (h *CustomToolsHandler) handleDelete(w http.ResponseWriter, r *http.Request return } + emitAudit(h.msgBus, r, "custom_tool.deleted", "custom_tool", id.String()) h.emitCacheInvalidate(id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"}) } diff --git a/internal/http/mcp.go b/internal/http/mcp.go index 214491eb..49b548fa 100644 --- a/internal/http/mcp.go +++ b/internal/http/mcp.go @@ -147,6 +147,7 @@ func (h *MCPHandler) handleCreateServer(w http.ResponseWriter, r *http.Request) } h.emitCacheInvalidate() + emitAudit(h.msgBus, r, "mcp_server.created", "mcp_server", srv.ID.String()) writeJSON(w, http.StatusCreated, srv) } @@ -195,6 +196,7 @@ func (h *MCPHandler) handleUpdateServer(w http.ResponseWriter, r *http.Request) } h.emitCacheInvalidate() + emitAudit(h.msgBus, r, "mcp_server.updated", "mcp_server", id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "updated"}) } @@ -213,5 +215,6 @@ func (h *MCPHandler) handleDeleteServer(w http.ResponseWriter, r *http.Request) } h.emitCacheInvalidate() + emitAudit(h.msgBus, r, "mcp_server.deleted", "mcp_server", id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"}) } diff --git a/internal/http/mcp_grants.go b/internal/http/mcp_grants.go index bf492d94..76a15b84 100644 --- a/internal/http/mcp_grants.go +++ b/internal/http/mcp_grants.go @@ -51,6 +51,7 @@ func (h *MCPHandler) handleGrantAgent(w http.ResponseWriter, r *http.Request) { } h.emitCacheInvalidate() + emitAudit(h.msgBus, r, "mcp_server.agent_granted", "mcp_server", serverID.String()) writeJSON(w, http.StatusCreated, map[string]string{"status": "granted"}) } @@ -75,6 +76,7 @@ func (h *MCPHandler) handleRevokeAgent(w http.ResponseWriter, r *http.Request) { } h.emitCacheInvalidate() + emitAudit(h.msgBus, r, "mcp_server.agent_revoked", "mcp_server", serverID.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "revoked"}) } @@ -157,6 +159,7 @@ func (h *MCPHandler) handleGrantUser(w http.ResponseWriter, r *http.Request) { } h.emitCacheInvalidate() + emitAudit(h.msgBus, r, "mcp_server.user_granted", "mcp_server", serverID.String()) writeJSON(w, http.StatusCreated, map[string]string{"status": "granted"}) } @@ -181,5 +184,6 @@ func (h *MCPHandler) handleRevokeUser(w http.ResponseWriter, r *http.Request) { } h.emitCacheInvalidate() + emitAudit(h.msgBus, r, "mcp_server.user_revoked", "mcp_server", serverID.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "revoked"}) } diff --git a/internal/http/mcp_requests.go b/internal/http/mcp_requests.go index e0eb7e29..579e9858 100644 --- a/internal/http/mcp_requests.go +++ b/internal/http/mcp_requests.go @@ -37,6 +37,7 @@ func (h *MCPHandler) handleCreateRequest(w http.ResponseWriter, r *http.Request) return } + emitAudit(h.msgBus, r, "mcp_request.created", "mcp_request", req.ID.String()) writeJSON(w, http.StatusCreated, req) } @@ -80,6 +81,7 @@ func (h *MCPHandler) handleReviewRequest(w http.ResponseWriter, r *http.Request) h.emitCacheInvalidate() } + emitAudit(h.msgBus, r, "mcp_request.reviewed", "mcp_request", requestID.String()) status := "rejected" if req.Approved { status = "approved" diff --git a/internal/http/oauth.go b/internal/http/oauth.go index fa13d5e8..f32b32a3 100644 --- a/internal/http/oauth.go +++ b/internal/http/oauth.go @@ -10,6 +10,7 @@ import ( "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/oauth" "github.com/nextlevelbuilder/goclaw/internal/providers" @@ -22,18 +23,20 @@ type OAuthHandler struct { provStore store.ProviderStore secretStore store.ConfigSecretsStore providerReg *providers.Registry + msgBus *bus.MessageBus mu sync.Mutex pending *oauth.PendingLogin // active OAuth flow (if any) } // NewOAuthHandler creates a handler for OAuth endpoints. -func NewOAuthHandler(token string, provStore store.ProviderStore, secretStore store.ConfigSecretsStore, providerReg *providers.Registry) *OAuthHandler { +func NewOAuthHandler(token string, provStore store.ProviderStore, secretStore store.ConfigSecretsStore, providerReg *providers.Registry, msgBus *bus.MessageBus) *OAuthHandler { return &OAuthHandler{ token: token, provStore: provStore, secretStore: secretStore, providerReg: providerReg, + msgBus: msgBus, } } @@ -115,6 +118,7 @@ func (h *OAuthHandler) handleStart(w http.ResponseWriter, r *http.Request) { // Wait for callback in background, save token when done go h.waitForCallback(pending) + emitAudit(h.msgBus, r, "oauth.login_started", "oauth", "openai") writeJSON(w, http.StatusOK, map[string]any{"auth_url": pending.AuthURL}) } @@ -204,6 +208,7 @@ func (h *OAuthHandler) handleLogout(w http.ResponseWriter, r *http.Request) { h.providerReg.Unregister(oauth.DefaultProviderName) } + emitAudit(h.msgBus, r, "oauth.logout", "oauth", "openai") writeJSON(w, http.StatusOK, map[string]string{"status": "logged out"}) } diff --git a/internal/http/oauth_test.go b/internal/http/oauth_test.go index 16a065b6..11cadd55 100644 --- a/internal/http/oauth_test.go +++ b/internal/http/oauth_test.go @@ -115,7 +115,7 @@ func (m *mockSecretsStore) GetAll(_ context.Context) (map[string]string, error) // --- helper --- func newTestOAuthHandler(token string) *OAuthHandler { - return NewOAuthHandler(token, newMockProviderStore(), newMockSecretsStore(), nil) + return NewOAuthHandler(token, newMockProviderStore(), newMockSecretsStore(), nil, nil) } // --- tests --- diff --git a/internal/http/providers.go b/internal/http/providers.go index 20d7b005..ae7eaa14 100644 --- a/internal/http/providers.go +++ b/internal/http/providers.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/oauth" "github.com/nextlevelbuilder/goclaw/internal/providers" @@ -20,9 +21,10 @@ type ProvidersHandler struct { secretStore store.ConfigSecretsStore token string providerReg *providers.Registry - gatewayAddr string // for injecting MCP bridge into Claude CLI providers + gatewayAddr string // for injecting MCP bridge into Claude CLI providers mcpLookup providers.MCPServerLookup // optional: resolves per-agent MCP servers - cliMu sync.Mutex // serializes Claude CLI provider create to prevent duplicates + cliMu sync.Mutex // serializes Claude CLI provider create to prevent duplicates + msgBus *bus.MessageBus } // NewProvidersHandler creates a handler for provider management endpoints. @@ -30,6 +32,12 @@ func NewProvidersHandler(s store.ProviderStore, secretStore store.ConfigSecretsS return &ProvidersHandler{store: s, secretStore: secretStore, token: token, providerReg: providerReg, gatewayAddr: gatewayAddr} } +// SetMessageBus sets the message bus for audit event broadcasting. +// Must be called before serving requests (not thread-safe). +func (h *ProvidersHandler) SetMessageBus(msgBus *bus.MessageBus) { + h.msgBus = msgBus +} + // SetMCPServerLookup sets the per-agent MCP server lookup for Claude CLI providers. // Must be called before serving requests (not thread-safe). func (h *ProvidersHandler) SetMCPServerLookup(lookup providers.MCPServerLookup) { @@ -189,6 +197,7 @@ func (h *ProvidersHandler) handleCreateProvider(w http.ResponseWriter, r *http.R // Register in-memory so verify/chat work without restart h.registerInMemory(&p) + emitAudit(h.msgBus, r, "provider.created", "provider", p.ID.String()) maskAPIKey(&p) writeJSON(w, http.StatusCreated, p) } @@ -284,6 +293,7 @@ func (h *ProvidersHandler) handleUpdateProvider(w http.ResponseWriter, r *http.R } } + emitAudit(h.msgBus, r, "provider.updated", "provider", id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "updated"}) } @@ -311,5 +321,6 @@ func (h *ProvidersHandler) handleDeleteProvider(w http.ResponseWriter, r *http.R h.providerReg.Unregister(providerName) } + emitAudit(h.msgBus, r, "provider.deleted", "provider", id.String()) writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"}) } diff --git a/internal/http/skills.go b/internal/http/skills.go index 868069f3..eb1cc6e2 100644 --- a/internal/http/skills.go +++ b/internal/http/skills.go @@ -124,6 +124,7 @@ func (h *SkillsHandler) handleUpdate(w http.ResponseWriter, r *http.Request) { } h.skills.BumpVersion() + emitAudit(h.msgBus, r, "skill.updated", "skill", idStr) writeJSON(w, http.StatusOK, map[string]string{"ok": "true"}) } @@ -146,6 +147,7 @@ func (h *SkillsHandler) handleDelete(w http.ResponseWriter, r *http.Request) { } h.skills.BumpVersion() + emitAudit(h.msgBus, r, "skill.deleted", "skill", idStr) writeJSON(w, http.StatusOK, map[string]string{"ok": "true"}) } @@ -371,5 +373,6 @@ func (h *SkillsHandler) handleToggle(w http.ResponseWriter, r *http.Request) { } h.skills.BumpVersion() + emitAudit(h.msgBus, r, "skill.toggled", "skill", idStr) writeJSON(w, http.StatusOK, map[string]any{"ok": true, "enabled": body.Enabled, "status": newStatus}) } diff --git a/internal/http/skills_grants.go b/internal/http/skills_grants.go index 6b5b6418..353bf792 100644 --- a/internal/http/skills_grants.go +++ b/internal/http/skills_grants.go @@ -69,6 +69,7 @@ func (h *SkillsHandler) handleGrantAgent(w http.ResponseWriter, r *http.Request) h.skills.BumpVersion() h.emitCacheInvalidate(bus.CacheKindSkillGrants, "") + emitAudit(h.msgBus, r, "skill.grant_changed", "skill", idStr) writeJSON(w, http.StatusCreated, map[string]string{"ok": "true"}) } @@ -95,6 +96,7 @@ func (h *SkillsHandler) handleRevokeAgent(w http.ResponseWriter, r *http.Request h.skills.BumpVersion() h.emitCacheInvalidate(bus.CacheKindSkillGrants, "") + emitAudit(h.msgBus, r, "skill.grant_changed", "skill", idStr) writeJSON(w, http.StatusOK, map[string]string{"ok": "true"}) } @@ -131,6 +133,7 @@ func (h *SkillsHandler) handleGrantUser(w http.ResponseWriter, r *http.Request) h.skills.BumpVersion() h.emitCacheInvalidate(bus.CacheKindSkillGrants, "") + emitAudit(h.msgBus, r, "skill.grant_changed", "skill", idStr) writeJSON(w, http.StatusCreated, map[string]string{"ok": "true"}) } @@ -155,6 +158,7 @@ func (h *SkillsHandler) handleRevokeUser(w http.ResponseWriter, r *http.Request) h.skills.BumpVersion() h.emitCacheInvalidate(bus.CacheKindSkillGrants, "") + emitAudit(h.msgBus, r, "skill.grant_changed", "skill", idStr) writeJSON(w, http.StatusOK, map[string]string{"ok": "true"}) } diff --git a/internal/http/skills_upload.go b/internal/http/skills_upload.go index 67ccb8a0..3f4b1bb8 100644 --- a/internal/http/skills_upload.go +++ b/internal/http/skills_upload.go @@ -185,6 +185,7 @@ func (h *SkillsHandler) handleUpload(w http.ResponseWriter, r *http.Request) { } h.skills.BumpVersion() + emitAudit(h.msgBus, r, "skill.uploaded", "skill", slug) slog.Info("skill uploaded", "id", id, "slug", slug, "version", version, "size", header.Size) // Scan and check dependencies diff --git a/pkg/protocol/events.go b/pkg/protocol/events.go index abb87f2e..887601d0 100644 --- a/pkg/protocol/events.go +++ b/pkg/protocol/events.go @@ -74,6 +74,9 @@ const ( // Cache invalidation events (internal, not forwarded to WS clients). EventCacheInvalidate = "cache.invalidate" + // Audit log event (internal, not forwarded to WS clients). + EventAuditLog = "audit.log" + // Zalo Personal QR login events (client-scoped, not broadcast). EventZaloPersonalQRCode = "zalo.personal.qr.code" EventZaloPersonalQRDone = "zalo.personal.qr.done"