diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 15f8c6037..b69bfb22f 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -176,6 +176,19 @@ func (m *Manager) initChannels() error { } } + if m.config.Channels.Messenger.Enabled && m.config.Channels.Messenger.PageAccessToken != "" { + logger.DebugC("channels", "Attempting to initialize Messenger channel") + messenger, err := NewMessengerChannel(m.config.Channels.Messenger, m.bus) + if err != nil { + logger.ErrorCF("channels", "Failed to initialize Messenger channel", map[string]interface{}{ + "error": err.Error(), + }) + } else { + m.channels["messenger"] = messenger + logger.InfoC("channels", "Messenger channel enabled successfully") + } + } + logger.InfoCF("channels", "Channel initialization completed", map[string]interface{}{ "enabled_channels": len(m.channels), }) diff --git a/pkg/channels/messenger.go b/pkg/channels/messenger.go new file mode 100644 index 000000000..8b384da6a --- /dev/null +++ b/pkg/channels/messenger.go @@ -0,0 +1,331 @@ +package channels + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/logger" + "github.com/sipeed/picoclaw/pkg/utils" +) + +const ( + messengerGraphAPIBase = "https://graph.facebook.com/v21.0" + messengerMessagesAPI = messengerGraphAPIBase + "/me/messages" +) + +// MessengerChannel implements the Channel interface for Facebook Messenger +// using the Meta Webhook API for receiving messages and the Graph API for +// sending replies. +type MessengerChannel struct { + *BaseChannel + config config.MessengerConfig + httpServer *http.Server + ctx context.Context + cancel context.CancelFunc +} + +// NewMessengerChannel creates a new Messenger channel instance. +func NewMessengerChannel(cfg config.MessengerConfig, messageBus *bus.MessageBus) (*MessengerChannel, error) { + if cfg.PageAccessToken == "" || cfg.AppSecret == "" { + return nil, fmt.Errorf("messenger page_access_token and app_secret are required") + } + + base := NewBaseChannel("messenger", cfg, messageBus, cfg.AllowFrom) + + return &MessengerChannel{ + BaseChannel: base, + config: cfg, + }, nil +} + +// Start launches the HTTP webhook server for Messenger. +func (c *MessengerChannel) Start(ctx context.Context) error { + logger.InfoC("messenger", "Starting Messenger channel (Webhook Mode)") + + c.ctx, c.cancel = context.WithCancel(ctx) + + mux := http.NewServeMux() + path := c.config.WebhookPath + if path == "" { + path = "/messenger/webhook" + } + mux.HandleFunc(path, c.webhookHandler) + + port := c.config.WebhookPort + if port == 0 { + port = 18791 + } + addr := fmt.Sprintf("0.0.0.0:%d", port) + c.httpServer = &http.Server{ + Addr: addr, + Handler: mux, + } + + go func() { + logger.InfoCF("messenger", "Messenger webhook server listening", map[string]interface{}{ + "addr": addr, + "path": path, + }) + if err := c.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + logger.ErrorCF("messenger", "Webhook server error", map[string]interface{}{ + "error": err.Error(), + }) + } + }() + + c.setRunning(true) + logger.InfoC("messenger", "Messenger channel started (Webhook Mode)") + return nil +} + +// Stop gracefully shuts down the HTTP server. +func (c *MessengerChannel) Stop(ctx context.Context) error { + logger.InfoC("messenger", "Stopping Messenger channel") + + if c.cancel != nil { + c.cancel() + } + + if c.httpServer != nil { + shutdownCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + if err := c.httpServer.Shutdown(shutdownCtx); err != nil { + logger.ErrorCF("messenger", "Webhook server shutdown error", map[string]interface{}{ + "error": err.Error(), + }) + } + } + + c.setRunning(false) + logger.InfoC("messenger", "Messenger channel stopped") + return nil +} + +// webhookHandler handles incoming Messenger webhook requests. +// GET requests handle webhook verification; POST requests handle incoming messages. +func (c *MessengerChannel) webhookHandler(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + c.handleVerification(w, r) + case http.MethodPost: + c.handleIncoming(w, r) + default: + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + } +} + +// handleVerification responds to the Meta webhook verification challenge. +func (c *MessengerChannel) handleVerification(w http.ResponseWriter, r *http.Request) { + mode := r.URL.Query().Get("hub.mode") + token := r.URL.Query().Get("hub.verify_token") + challenge := r.URL.Query().Get("hub.challenge") + + if mode == "subscribe" && token == c.config.VerifyToken { + logger.InfoC("messenger", "Webhook verification successful") + w.WriteHeader(http.StatusOK) + w.Write([]byte(challenge)) + return + } + + logger.WarnC("messenger", "Webhook verification failed") + http.Error(w, "Forbidden", http.StatusForbidden) +} + +// handleIncoming processes incoming webhook POST requests. +func (c *MessengerChannel) handleIncoming(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + logger.ErrorCF("messenger", "Failed to read request body", map[string]interface{}{ + "error": err.Error(), + }) + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + + // Verify signature + signature := r.Header.Get("X-Hub-Signature-256") + if !c.verifySignature(body, signature) { + logger.WarnC("messenger", "Invalid webhook signature") + http.Error(w, "Forbidden", http.StatusForbidden) + return + } + + var payload messengerWebhookPayload + if err := json.Unmarshal(body, &payload); err != nil { + logger.ErrorCF("messenger", "Failed to parse webhook payload", map[string]interface{}{ + "error": err.Error(), + }) + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + + // Return 200 immediately, process events asynchronously + w.WriteHeader(http.StatusOK) + w.Write([]byte("EVENT_RECEIVED")) + + for _, entry := range payload.Entry { + for _, event := range entry.Messaging { + go c.processEvent(event) + } + } +} + +// verifySignature validates the X-Hub-Signature-256 using HMAC-SHA256. +func (c *MessengerChannel) verifySignature(body []byte, signature string) bool { + if signature == "" { + return false + } + + // Signature format: "sha256=" + prefix := "sha256=" + if !strings.HasPrefix(signature, prefix) { + return false + } + sigHex := signature[len(prefix):] + + mac := hmac.New(sha256.New, []byte(c.config.AppSecret)) + mac.Write(body) + expected := hex.EncodeToString(mac.Sum(nil)) + + return hmac.Equal([]byte(expected), []byte(sigHex)) +} + +// Messenger webhook payload types +type messengerWebhookPayload struct { + Object string `json:"object"` + Entry []messengerWebhookEntry `json:"entry"` +} + +type messengerWebhookEntry struct { + ID string `json:"id"` + Time int64 `json:"time"` + Messaging []messengerMessagingEvent `json:"messaging"` +} + +type messengerMessagingEvent struct { + Sender messengerUser `json:"sender"` + Recipient messengerUser `json:"recipient"` + Timestamp int64 `json:"timestamp"` + Message *messengerMsg `json:"message,omitempty"` +} + +type messengerUser struct { + ID string `json:"id"` +} + +type messengerMsg struct { + MID string `json:"mid"` + Text string `json:"text"` + Attachments []messengerAttachment `json:"attachments,omitempty"` +} + +type messengerAttachment struct { + Type string `json:"type"` + Payload messengerAttachmentPayload `json:"payload"` +} + +type messengerAttachmentPayload struct { + URL string `json:"url"` +} + +func (c *MessengerChannel) processEvent(event messengerMessagingEvent) { + // Only process message events + if event.Message == nil { + logger.DebugCF("messenger", "Ignoring non-message event", map[string]interface{}{ + "sender": event.Sender.ID, + }) + return + } + + senderID := event.Sender.ID + chatID := senderID // Messenger conversations are 1:1 with sender + + var content string + + if event.Message.Text != "" { + content = event.Message.Text + } else if len(event.Message.Attachments) > 0 { + // Log attachments but skip for now + for _, att := range event.Message.Attachments { + logger.InfoCF("messenger", "Received attachment (skipping)", map[string]interface{}{ + "type": att.Type, + "url": att.Payload.URL, + }) + } + content = fmt.Sprintf("[%s attachment]", event.Message.Attachments[0].Type) + } + + if strings.TrimSpace(content) == "" { + return + } + + metadata := map[string]string{ + "platform": "messenger", + "message_id": event.Message.MID, + } + + logger.DebugCF("messenger", "Received message", map[string]interface{}{ + "sender_id": senderID, + "chat_id": chatID, + "preview": utils.Truncate(content, 50), + }) + + c.HandleMessage(senderID, chatID, content, nil, metadata) +} + +// Send sends a message to a Messenger user via the Graph API. +func (c *MessengerChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + if !c.IsRunning() { + return fmt.Errorf("messenger channel not running") + } + + payload := map[string]interface{}{ + "recipient": map[string]string{ + "id": msg.ChatID, + }, + "message": map[string]string{ + "text": msg.Content, + }, + } + + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("failed to marshal payload: %w", err) + } + + url := messengerMessagesAPI + "?access_token=" + c.config.PageAccessToken + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + client := &http.Client{Timeout: 30 * time.Second} + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("Graph API request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(resp.Body) + return fmt.Errorf("Messenger Graph API error (status %d): %s", resp.StatusCode, string(respBody)) + } + + logger.DebugCF("messenger", "Message sent", map[string]interface{}{ + "chat_id": msg.ChatID, + }) + + return nil +} diff --git a/pkg/channels/messenger_test.go b/pkg/channels/messenger_test.go new file mode 100644 index 000000000..c41c82f09 --- /dev/null +++ b/pkg/channels/messenger_test.go @@ -0,0 +1,216 @@ +package channels + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/config" +) + +func newTestMessengerChannel(t *testing.T) *MessengerChannel { + t.Helper() + cfg := config.MessengerConfig{ + Enabled: true, + PageAccessToken: "test-page-token", + VerifyToken: "test-verify-token", + AppSecret: "test-app-secret", + WebhookPort: 0, + WebhookPath: "/messenger/webhook", + } + msgBus := bus.NewMessageBus() + ch, err := NewMessengerChannel(cfg, msgBus) + if err != nil { + t.Fatalf("NewMessengerChannel: %v", err) + } + return ch +} + +func computeSignature(secret, body string) string { + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write([]byte(body)) + return "sha256=" + hex.EncodeToString(mac.Sum(nil)) +} + +func TestMessengerVerifySignature(t *testing.T) { + ch := newTestMessengerChannel(t) + + tests := []struct { + name string + body string + signature string + want bool + }{ + { + name: "valid signature", + body: `{"object":"page"}`, + signature: computeSignature("test-app-secret", `{"object":"page"}`), + want: true, + }, + { + name: "invalid signature", + body: `{"object":"page"}`, + signature: "sha256=0000000000000000000000000000000000000000000000000000000000000000", + want: false, + }, + { + name: "empty signature", + body: `{"object":"page"}`, + signature: "", + want: false, + }, + { + name: "missing sha256 prefix", + body: `{"object":"page"}`, + signature: "deadbeef", + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := ch.verifySignature([]byte(tt.body), tt.signature) + if got != tt.want { + t.Errorf("verifySignature() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestMessengerWebhookVerification(t *testing.T) { + ch := newTestMessengerChannel(t) + + tests := []struct { + name string + query string + wantStatus int + wantBody string + }{ + { + name: "valid verification", + query: "hub.mode=subscribe&hub.verify_token=test-verify-token&hub.challenge=challenge123", + wantStatus: http.StatusOK, + wantBody: "challenge123", + }, + { + name: "wrong verify token", + query: "hub.mode=subscribe&hub.verify_token=wrong-token&hub.challenge=challenge123", + wantStatus: http.StatusForbidden, + }, + { + name: "wrong mode", + query: "hub.mode=unsubscribe&hub.verify_token=test-verify-token&hub.challenge=challenge123", + wantStatus: http.StatusForbidden, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/messenger/webhook?"+tt.query, nil) + w := httptest.NewRecorder() + ch.webhookHandler(w, req) + + resp := w.Result() + if resp.StatusCode != tt.wantStatus { + t.Errorf("status = %d, want %d", resp.StatusCode, tt.wantStatus) + } + if tt.wantBody != "" { + body, _ := io.ReadAll(resp.Body) + if strings.TrimSpace(string(body)) != tt.wantBody { + t.Errorf("body = %q, want %q", string(body), tt.wantBody) + } + } + }) + } +} + +func TestMessengerMessageParsing(t *testing.T) { + ch := newTestMessengerChannel(t) + + payload := messengerWebhookPayload{ + Object: "page", + Entry: []messengerWebhookEntry{ + { + ID: "page123", + Time: 1234567890, + Messaging: []messengerMessagingEvent{ + { + Sender: messengerUser{ID: "user456"}, + Recipient: messengerUser{ID: "page123"}, + Timestamp: 1234567890, + Message: &messengerMsg{ + MID: "mid.123", + Text: "Hello, bot!", + }, + }, + }, + }, + }, + } + + body, err := json.Marshal(payload) + if err != nil { + t.Fatalf("marshal: %v", err) + } + + signature := computeSignature("test-app-secret", string(body)) + + req := httptest.NewRequest(http.MethodPost, "/messenger/webhook", strings.NewReader(string(body))) + req.Header.Set("X-Hub-Signature-256", signature) + w := httptest.NewRecorder() + + ch.webhookHandler(w, req) + + resp := w.Result() + if resp.StatusCode != http.StatusOK { + t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusOK) + } + + respBody, _ := io.ReadAll(resp.Body) + if strings.TrimSpace(string(respBody)) != "EVENT_RECEIVED" { + t.Errorf("body = %q, want %q", string(respBody), "EVENT_RECEIVED") + } +} + +func TestMessengerRejectsInvalidSignature(t *testing.T) { + ch := newTestMessengerChannel(t) + + body := `{"object":"page","entry":[]}` + req := httptest.NewRequest(http.MethodPost, "/messenger/webhook", strings.NewReader(body)) + req.Header.Set("X-Hub-Signature-256", "sha256=invalid") + w := httptest.NewRecorder() + + ch.webhookHandler(w, req) + + resp := w.Result() + if resp.StatusCode != http.StatusForbidden { + t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusForbidden) + } +} + +func TestNewMessengerChannelRequiresCredentials(t *testing.T) { + msgBus := bus.NewMessageBus() + + _, err := NewMessengerChannel(config.MessengerConfig{ + Enabled: true, + }, msgBus) + if err == nil { + t.Error("expected error for missing credentials") + } + + _, err = NewMessengerChannel(config.MessengerConfig{ + Enabled: true, + PageAccessToken: "token", + AppSecret: "secret", + }, msgBus) + if err != nil { + t.Errorf("unexpected error: %v", err) + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 932c53cb1..b1d1573a6 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -75,16 +75,17 @@ type AgentDefaults struct { } type ChannelsConfig struct { - WhatsApp WhatsAppConfig `json:"whatsapp"` - Telegram TelegramConfig `json:"telegram"` - Feishu FeishuConfig `json:"feishu"` - Discord DiscordConfig `json:"discord"` - MaixCam MaixCamConfig `json:"maixcam"` - QQ QQConfig `json:"qq"` - DingTalk DingTalkConfig `json:"dingtalk"` - Slack SlackConfig `json:"slack"` - LINE LINEConfig `json:"line"` - OneBot OneBotConfig `json:"onebot"` + WhatsApp WhatsAppConfig `json:"whatsapp"` + Telegram TelegramConfig `json:"telegram"` + Feishu FeishuConfig `json:"feishu"` + Discord DiscordConfig `json:"discord"` + MaixCam MaixCamConfig `json:"maixcam"` + QQ QQConfig `json:"qq"` + DingTalk DingTalkConfig `json:"dingtalk"` + Slack SlackConfig `json:"slack"` + LINE LINEConfig `json:"line"` + OneBot OneBotConfig `json:"onebot"` + Messenger MessengerConfig `json:"messenger"` } type WhatsAppConfig struct { @@ -162,6 +163,16 @@ type OneBotConfig struct { AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_ONEBOT_ALLOW_FROM"` } +type MessengerConfig struct { + Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_MESSENGER_ENABLED"` + PageAccessToken string `json:"page_access_token" env:"PICOCLAW_CHANNELS_MESSENGER_PAGE_ACCESS_TOKEN"` + VerifyToken string `json:"verify_token" env:"PICOCLAW_CHANNELS_MESSENGER_VERIFY_TOKEN"` + AppSecret string `json:"app_secret" env:"PICOCLAW_CHANNELS_MESSENGER_APP_SECRET"` + WebhookPort int `json:"webhook_port" env:"PICOCLAW_CHANNELS_MESSENGER_WEBHOOK_PORT"` + WebhookPath string `json:"webhook_path" env:"PICOCLAW_CHANNELS_MESSENGER_WEBHOOK_PATH"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_MESSENGER_ALLOW_FROM"` +} + type HeartbeatConfig struct { Enabled bool `json:"enabled" env:"PICOCLAW_HEARTBEAT_ENABLED"` Interval int `json:"interval" env:"PICOCLAW_HEARTBEAT_INTERVAL"` // minutes, min 5 @@ -299,6 +310,15 @@ func DefaultConfig() *Config { GroupTriggerPrefix: []string{}, AllowFrom: FlexibleStringSlice{}, }, + Messenger: MessengerConfig{ + Enabled: false, + PageAccessToken: "", + VerifyToken: "", + AppSecret: "", + WebhookPort: 18791, + WebhookPath: "/messenger/webhook", + AllowFrom: FlexibleStringSlice{}, + }, }, Providers: ProvidersConfig{ Anthropic: ProviderConfig{},