diff --git a/cmd/gateway.go b/cmd/gateway.go index a1d13ae9..8f468eba 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -10,6 +10,8 @@ import ( "path/filepath" "syscall" + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/agent" "github.com/nextlevelbuilder/goclaw/internal/bootstrap" "github.com/nextlevelbuilder/goclaw/internal/bus" @@ -850,6 +852,61 @@ func runGateway() { } }) + // Wire pairing revocation → force disconnect active WebSocket sessions. + msgBus.Subscribe(bus.TopicPairingRevoked, func(event bus.Event) { + if event.Name != bus.EventPairingRevoked { + return + } + payload, ok := event.Payload.(bus.PairingRevokedPayload) + if !ok { + return + } + go server.DisconnectByPairing(payload.SenderID, payload.Channel) + }) + + // Cascade: when an agent becomes inactive, disable its linked channel instances. + if managedStores != nil && managedStores.ChannelInstances != nil { + ciStore := managedStores.ChannelInstances + msgBus.Subscribe(bus.TopicAgentStatusChanged, func(event bus.Event) { + if event.Name != bus.EventAgentStatusChanged { + return + } + payload, ok := event.Payload.(bus.AgentStatusChangedPayload) + if !ok || payload.NewStatus != store.AgentStatusInactive { + return + } + go func() { + agentID, err := uuid.Parse(payload.AgentID) + if err != nil { + return + } + all, err := ciStore.ListAll(context.Background()) + if err != nil { + slog.Warn("cascade disable: failed to list channel instances", "error", err) + return + } + disabled := 0 + for _, inst := range all { + if inst.AgentID == agentID && inst.Enabled { + if err := ciStore.Update(context.Background(), inst.ID, map[string]any{"enabled": false}); err != nil { + slog.Warn("cascade disable: failed to disable channel instance", "name", inst.Name, "error", err) + } else { + disabled++ + } + } + } + if disabled > 0 { + slog.Info("cascade disabled channel instances for inactive agent", "agent_id", payload.AgentID, "count", disabled) + // Trigger channel reload so disabled instances are stopped. + msgBus.Broadcast(bus.Event{ + Name: protocol.EventCacheInvalidate, + Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindChannelInstances}, + }) + } + }() + }) + } + // Setup graceful shutdown ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/cmd/gateway_methods.go b/cmd/gateway_methods.go index 5e4182da..8ec22522 100644 --- a/cmd/gateway_methods.go +++ b/cmd/gateway_methods.go @@ -29,7 +29,7 @@ func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore // Phase 2: Pairing (store created externally, shared with channel manager). // OnApprove callback is set later by the caller after channel manager is created. - pairingMethods := methods.NewPairingMethods(pairingStore) + pairingMethods := methods.NewPairingMethods(pairingStore, msgBus) pairingMethods.Register(router) // Phase 2: Usage (queries SessionStore for real token data) diff --git a/internal/agent/resolver.go b/internal/agent/resolver.go index 475a4f34..62103da8 100644 --- a/internal/agent/resolver.go +++ b/internal/agent/resolver.go @@ -89,6 +89,10 @@ func NewManagedResolver(deps ResolverDeps) ResolverFunc { return nil, fmt.Errorf("agent not found: %s", agentKey) } + if ag.Status != store.AgentStatusActive { + return nil, fmt.Errorf("agent %s is inactive", agentKey) + } + // Resolve provider provider, err := deps.ProviderReg.Get(ag.Provider) if err != nil { diff --git a/internal/bus/types.go b/internal/bus/types.go index 58ae6ee3..90c79f83 100644 --- a/internal/bus/types.go +++ b/internal/bus/types.go @@ -72,8 +72,29 @@ const ( TopicCacheMCP = "cache:mcp" TopicChannelStreaming = "channel-streaming" TopicConfigChanged = "config:changed" + TopicPairingRevoked = "pairing:revoked" + TopicAgentStatusChanged = "agent:status_changed" ) +// EventPairingRevoked is the event name broadcast when a paired device is revoked. +const EventPairingRevoked = "pairing.revoked" + +// PairingRevokedPayload identifies the revoked device. +type PairingRevokedPayload struct { + SenderID string `json:"sender_id"` + Channel string `json:"channel"` +} + +// EventAgentStatusChanged is broadcast when an agent's status changes (e.g., active → inactive). +const EventAgentStatusChanged = "agent.status_changed" + +// AgentStatusChangedPayload carries agent status transition info for cascade operations. +type AgentStatusChangedPayload struct { + AgentID string `json:"agent_id"` + OldStatus string `json:"old_status"` + NewStatus string `json:"new_status"` +} + // CacheInvalidatePayload signals cache layers to evict stale entries. // Used with protocol.EventCacheInvalidate events. type CacheInvalidatePayload struct { diff --git a/internal/channels/discord/discord.go b/internal/channels/discord/discord.go index 76942ac6..68a068c1 100644 --- a/internal/channels/discord/discord.go +++ b/internal/channels/discord/discord.go @@ -30,6 +30,7 @@ type Channel struct { typingCtrls sync.Map // channelID string → *typing.Controller pairingService store.PairingStore pairingDebounce sync.Map // senderID → time.Time + approvedGroups sync.Map // chatID → true (in-memory cache for paired groups) groupHistory *channels.PendingHistory historyLimit int } @@ -240,7 +241,7 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate return } } else { - if !c.CheckPolicy("group", "", c.config.GroupPolicy, senderID) { + if !c.checkGroupPolicy(senderID, channelID) { slog.Debug("discord group message rejected by policy", "user_id", senderID, "username", senderName, @@ -366,6 +367,37 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate } } +// checkGroupPolicy evaluates the group policy for a sender, with pairing support. +func (c *Channel) checkGroupPolicy(senderID, channelID string) bool { + groupPolicy := c.config.GroupPolicy + if groupPolicy == "" { + groupPolicy = "open" + } + + switch groupPolicy { + case "disabled": + return false + case "allowlist": + return c.IsAllowed(senderID) + case "pairing": + if c.IsAllowed(senderID) { + return true + } + if _, cached := c.approvedGroups.Load(channelID); cached { + return true + } + groupSenderID := fmt.Sprintf("group:%s", channelID) + if c.pairingService != nil && c.pairingService.IsPaired(groupSenderID, c.Name()) { + c.approvedGroups.Store(channelID, true) + return true + } + c.sendPairingReply(groupSenderID, channelID) + return false + default: // "open" + return true + } +} + // checkDMPolicy evaluates the DM policy for a sender, handling pairing flow. func (c *Channel) checkDMPolicy(senderID, channelID string) bool { dmPolicy := c.config.DMPolicy diff --git a/internal/channels/whatsapp/whatsapp.go b/internal/channels/whatsapp/whatsapp.go index 279d1d4a..bd35bdf0 100644 --- a/internal/channels/whatsapp/whatsapp.go +++ b/internal/channels/whatsapp/whatsapp.go @@ -32,6 +32,7 @@ type Channel struct { cancel context.CancelFunc pairingService store.PairingStore pairingDebounce sync.Map // senderID → time.Time + approvedGroups sync.Map // chatID → true (in-memory cache for paired groups) } // New creates a new WhatsApp channel from config. @@ -222,7 +223,7 @@ func (c *Channel) handleIncomingMessage(msg map[string]interface{}) { return } } else { - if !c.CheckPolicy("group", "", c.config.GroupPolicy, senderID) { + if !c.checkGroupPolicy(senderID, chatID) { slog.Debug("whatsapp group message rejected by policy", "sender_id", senderID) return } @@ -266,6 +267,37 @@ func (c *Channel) handleIncomingMessage(msg map[string]interface{}) { c.HandleMessage(senderID, chatID, content, media, metadata, peerKind) } +// checkGroupPolicy evaluates the group policy for a sender, with pairing support. +func (c *Channel) checkGroupPolicy(senderID, chatID string) bool { + groupPolicy := c.config.GroupPolicy + if groupPolicy == "" { + groupPolicy = "open" + } + + switch groupPolicy { + case "disabled": + return false + case "allowlist": + return c.IsAllowed(senderID) + case "pairing": + if c.IsAllowed(senderID) { + return true + } + if _, cached := c.approvedGroups.Load(chatID); cached { + return true + } + groupSenderID := fmt.Sprintf("group:%s", chatID) + if c.pairingService != nil && c.pairingService.IsPaired(groupSenderID, c.Name()) { + c.approvedGroups.Store(chatID, true) + return true + } + c.sendPairingReply(groupSenderID, chatID) + return false + default: // "open" + return true + } +} + // checkDMPolicy evaluates the DM policy for a sender, handling pairing flow. func (c *Channel) checkDMPolicy(senderID, chatID string) bool { dmPolicy := c.config.DMPolicy diff --git a/internal/channels/zalo/personal/channel.go b/internal/channels/zalo/personal/channel.go index 6bc41308..77a1349d 100644 --- a/internal/channels/zalo/personal/channel.go +++ b/internal/channels/zalo/personal/channel.go @@ -35,6 +35,7 @@ type Channel struct { config config.ZaloPersonalConfig pairingService store.PairingStore pairingDebounce sync.Map // senderID -> time.Time + approvedGroups sync.Map // groupID → true (in-memory cache for paired groups) typingCtrls sync.Map // threadID → *typing.Controller mu sync.RWMutex // protects sess and listener diff --git a/internal/channels/zalo/personal/policy.go b/internal/channels/zalo/personal/policy.go index b5c2a320..a977be1f 100644 --- a/internal/channels/zalo/personal/policy.go +++ b/internal/channels/zalo/personal/policy.go @@ -100,6 +100,21 @@ func (c *Channel) checkGroupPolicy(senderID, groupID string, mentions []*protoco slog.Debug("zalo_personal group message rejected by allowlist", "group_id", groupID) return false } + + case "pairing": + if c.IsAllowed(groupID) { + // pass — allowlist bypass + } else if _, cached := c.approvedGroups.Load(groupID); cached { + // pass — already approved + } else { + groupSenderID := fmt.Sprintf("group:%s", groupID) + if c.pairingService != nil && c.pairingService.IsPaired(groupSenderID, c.Name()) { + c.approvedGroups.Store(groupID, true) + } else { + c.sendPairingReply(groupSenderID, groupID) + return false + } + } } // @mention gating: only process group messages that @mention the bot. diff --git a/internal/gateway/client.go b/internal/gateway/client.go index 420a0d87..6f05e36b 100644 --- a/internal/gateway/client.go +++ b/internal/gateway/client.go @@ -29,8 +29,10 @@ type Client struct { remoteAddr string // peer IP (extracted from proxy headers or RemoteAddr) // Browser pairing state - pairingCode string // 8-char code if pending approval - pairingPending bool // true while waiting for admin approval + pairingCode string // 8-char code if pending approval + pairingPending bool // true while waiting for admin approval + pairedSenderID string // senderID used for browser pairing auth (for revocation lookup) + pairedChannel string // channel used for pairing auth (e.g., "browser") } func NewClient(conn *websocket.Conn, server *Server, remoteIP string) *Client { diff --git a/internal/gateway/methods/pairing.go b/internal/gateway/methods/pairing.go index 438e4e6a..2ade4d21 100644 --- a/internal/gateway/methods/pairing.go +++ b/internal/gateway/methods/pairing.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/store" "github.com/nextlevelbuilder/goclaw/pkg/protocol" @@ -16,11 +17,12 @@ type PairingApproveCallback func(ctx context.Context, channel, chatID string) // PairingMethods handles device.pair.request, device.pair.approve, device.pair.list, device.pair.revoke. type PairingMethods struct { service store.PairingStore + msgBus *bus.MessageBus onApprove PairingApproveCallback } -func NewPairingMethods(service store.PairingStore) *PairingMethods { - return &PairingMethods{service: service} +func NewPairingMethods(service store.PairingStore, msgBus *bus.MessageBus) *PairingMethods { + return &PairingMethods{service: service, msgBus: msgBus} } // SetOnApprove sets a callback that fires after a pairing is approved. @@ -154,6 +156,17 @@ func (m *PairingMethods) handleRevoke(_ context.Context, client *gateway.Client, return } + // Broadcast revocation so the server can force-disconnect the active session. + if m.msgBus != nil { + m.msgBus.Broadcast(bus.Event{ + Name: bus.EventPairingRevoked, + Payload: bus.PairingRevokedPayload{ + SenderID: params.SenderID, + Channel: params.Channel, + }, + }) + } + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ "revoked": true, })) diff --git a/internal/gateway/router.go b/internal/gateway/router.go index 8b01da6d..717f4442 100644 --- a/internal/gateway/router.go +++ b/internal/gateway/router.go @@ -114,6 +114,8 @@ func (r *MethodRouter) handleConnect(ctx context.Context, client *Client, req *p client.role = permissions.RoleOperator client.authenticated = true client.userID = params.UserID + client.pairedSenderID = params.SenderID + client.pairedChannel = "browser" slog.Info("browser pairing authenticated", "sender_id", params.SenderID, "client", client.id) r.sendConnectResponse(client, req.ID) return diff --git a/internal/gateway/server.go b/internal/gateway/server.go index 97234857..1615e521 100644 --- a/internal/gateway/server.go +++ b/internal/gateway/server.go @@ -335,6 +335,25 @@ func (s *Server) BroadcastEvent(event protocol.EventFrame) { } } +// DisconnectByPairing force-closes WebSocket connections authenticated via the +// given pairing senderID and channel. Called after revoking a paired device so +// that the revoked client cannot continue operating with its old role. +func (s *Server) DisconnectByPairing(senderID, channel string) { + s.mu.RLock() + var targets []*Client + for _, c := range s.clients { + if c.pairedSenderID == senderID && c.pairedChannel == channel { + targets = append(targets, c) + } + } + s.mu.RUnlock() + + for _, c := range targets { + slog.Info("disconnecting revoked paired device", "client", c.id, "sender_id", senderID, "channel", channel) + c.conn.Close() + } +} + func (s *Server) registerClient(c *Client) { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/http/agents.go b/internal/http/agents.go index 2011f63f..4cdfab8a 100644 --- a/internal/http/agents.go +++ b/internal/http/agents.go @@ -247,6 +247,20 @@ func (h *AgentsHandler) handleUpdate(w http.ResponseWriter, r *http.Request) { h.emitCacheInvalidate(bus.CacheKindAgent, ag.AgentKey) h.emitCacheInvalidate(bus.CacheKindBootstrap, id.String()) + // Cascade: if status changed, broadcast so channel instances and cron jobs react. + if newStatus, ok := updates["status"].(string); ok && newStatus != ag.Status { + if h.msgBus != nil { + h.msgBus.Broadcast(bus.Event{ + Name: bus.EventAgentStatusChanged, + Payload: bus.AgentStatusChangedPayload{ + AgentID: id.String(), + OldStatus: ag.Status, + NewStatus: newStatus, + }, + }) + } + } + writeJSON(w, http.StatusOK, map[string]string{"ok": "true"}) }