diff --git a/pkg/channels/README.md b/pkg/channels/README.md index 7f238ece5..3b07ad2ae 100644 --- a/pkg/channels/README.md +++ b/pkg/channels/README.md @@ -1372,7 +1372,7 @@ agentLoop.Stop() // Stop Agent 3. **WeCom is now a single channel**: `"wecom"` is implemented as a WebSocket-based AI Bot channel with route persistence. Access control uses the shared channel allowlist mechanism. It no longer exposes the legacy webhook/app split. -4. **Pico Protocol**: `pkg/channels/pico/` implements a custom PicoClaw native protocol channel that receives messages via WebSocket webhook (`/pico/ws`). +4. **Pico Protocol**: `pkg/channels/pico/` implements a custom PicoClaw native protocol channel that receives messages via WebSocket (`/pico/ws`) and HTTP message ingress (`POST /pico/message`). The HTTP endpoint accepts a JSON body with `content` and optional `session_id`, authenticates with the same Pico token as the WebSocket (Bearer header, or query parameter when `allow_token_query` is enabled), and publishes the message to the bus asynchronously (returns `202 Accepted`). 5. **WhatsApp has two modes**: `"whatsapp"` (Bridge mode, communicates via external bridge URL) and `"whatsapp_native"` (native whatsmeow mode, connects directly to WhatsApp). Manager selects which to initialize based on `WhatsAppConfig.UseNative`. diff --git a/pkg/channels/README.zh.md b/pkg/channels/README.zh.md index 8bc8c8dbc..4163b8fe1 100644 --- a/pkg/channels/README.zh.md +++ b/pkg/channels/README.zh.md @@ -1371,7 +1371,7 @@ agentLoop.Stop() // 停止 Agent 3. **WeCom 现在只有一个 channel**：`"wecom"` 采用 WebSocket AI Bot 实现，带路由持久化；访问控制走统一的 channel 白名单机制，不再保留旧的 webhook/app 双分支。 -4. **Pico Protocol**：`pkg/channels/pico/` 实现了一个自定义的 PicoClaw 原生协议 channel，通过 WebSocket webhook (`/pico/ws`) 接收消息。 +4. **Pico Protocol**：`pkg/channels/pico/` 实现了一个自定义的 PicoClaw 原生协议 channel，通过 WebSocket (`/pico/ws`) 和 HTTP 消息入口 (`POST /pico/message`) 接收消息。HTTP 端点接受包含 `content` 和可选 `session_id` 的 JSON 请求体，使用与 WebSocket 相同的 Pico token 进行认证（Bearer header，或在启用 `allow_token_query` 时支持查询参数），并将消息异步发布到总线（返回 `202 Accepted`）。 5. **WhatsApp 有两种模式**：`"whatsapp"`（Bridge 模式，通过外部 bridge URL 通信）和 `"whatsapp_native"`（原生 whatsmeow 模式，直接连接 WhatsApp）。Manager 根据 `WhatsAppConfig.UseNative` 决定初始化哪个。 diff --git a/pkg/channels/pico/pico.go b/pkg/channels/pico/pico.go index f3ba55a92..e9e5a1ca7 100644 --- a/pkg/channels/pico/pico.go +++ b/pkg/channels/pico/pico.go @@ -2,8 +2,10 @@ package pico import ( "context" + "crypto/subtle" "encoding/json" "fmt" + "io" "net/http" "strings" "sync" @@ -20,6 +22,23 @@ import ( "github.com/sipeed/picoclaw/pkg/logger" ) +// maxHTTPBodySize is the maximum request body for HTTP message ingress. +// Text prompts are small; 64 KiB is generous while keeping the attack surface low. +const maxHTTPBodySize = 64 << 10 // 64 KiB + +// httpMessageRequest is the JSON body for POST /pico/message. +type httpMessageRequest struct { + Content string `json:"content"` + SessionID string `json:"session_id,omitempty"` +} + +// httpMessageResponse is the JSON response for POST /pico/message. +type httpMessageResponse struct { + OK bool `json:"ok"` + Status string `json:"status"` + SessionID string `json:"session_id"` +} + // picoConn represents a single WebSocket connection. type picoConn struct { id string @@ -228,6 +247,8 @@ func (c *PicoChannel) ServeHTTP(w http.ResponseWriter, r *http.Request) { switch path { case "/ws", "/ws/": c.handleWebSocket(w, r) + case "/message", "/message/": + c.handleHTTPMessage(w, r) default: http.NotFound(w, r) } @@ -383,6 +404,8 @@ func (c *PicoChannel) handleWebSocket(w http.ResponseWriter, r *http.Request) { // 1. Authorization: Bearer header // 2. Sec-WebSocket-Protocol "token." (for browsers that can't set headers) // 3. Query parameter "token" (only when AllowTokenQuery is on) +// +// Token comparison uses subtle.ConstantTimeCompare to reduce timing side-channels. func (c *PicoChannel) authenticate(r *http.Request) bool { token := c.config.Token() if token == "" { @@ -392,7 +415,7 @@ func (c *PicoChannel) authenticate(r *http.Request) bool { // Check Authorization header auth := r.Header.Get("Authorization") if after, ok := strings.CutPrefix(auth, "Bearer "); ok { - if after == token { + if subtle.ConstantTimeCompare([]byte(after), []byte(token)) == 1 { return true } } @@ -404,7 +427,7 @@ func (c *PicoChannel) authenticate(r *http.Request) bool { // Check query parameter only when explicitly allowed if c.config.AllowTokenQuery { - if r.URL.Query().Get("token") == token { + if subtle.ConstantTimeCompare([]byte(r.URL.Query().Get("token")), []byte(token)) == 1 { return true } } @@ -424,6 +447,89 @@ func (c *PicoChannel) matchedSubprotocol(r *http.Request) string { return "" } +// handleHTTPMessage processes an inbound HTTP message trigger. +// This is the HTTP equivalent of a Pico WebSocket message.send — it accepts +// a JSON body, publishes the message to the bus, and returns 202 Accepted. +func (c *PicoChannel) handleHTTPMessage(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + if !c.IsRunning() { + http.Error(w, "channel not running", http.StatusServiceUnavailable) + return + } + + if !c.authenticate(r) { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + + body, err := io.ReadAll(io.LimitReader(r.Body, maxHTTPBodySize+1)) + if err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + if int64(len(body)) > maxHTTPBodySize { + http.Error(w, "request entity too large", http.StatusRequestEntityTooLarge) + return + } + + var req httpMessageRequest + if err := json.Unmarshal(body, &req); err != nil { + http.Error(w, "bad request: invalid JSON", http.StatusBadRequest) + return + } + + // Validate non-empty, but preserve original content including whitespace. + // This matches WebSocket message.send semantics in handleMessageSend. + content := req.Content + if strings.TrimSpace(content) == "" { + http.Error(w, "bad request: empty content", http.StatusBadRequest) + return + } + + sessionID := req.SessionID + if sessionID == "" { + sessionID = uuid.New().String() + } + + chatID := "pico:" + sessionID + senderID := "pico-user" + peer := bus.Peer{Kind: "direct", ID: "pico:" + sessionID} + + metadata := map[string]string{ + "platform": "pico", + "session_id": sessionID, + "transport": "http", + } + + logger.DebugCF("pico", "Received HTTP message", map[string]any{ + "session_id": sessionID, + "preview": truncate(content, 50), + }) + + sender := bus.SenderInfo{ + Platform: "pico", + PlatformID: senderID, + CanonicalID: identity.BuildCanonicalID("pico", senderID), + } + + // Allow-list enforcement is handled internally by BaseChannel.HandleMessage, + // matching WebSocket message.send semantics (silent drop, not HTTP error). + msgID := uuid.New().String() + c.HandleMessage(c.ctx, peer, msgID, senderID, chatID, content, nil, metadata, sender) + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(httpMessageResponse{ + OK: true, + Status: "accepted", + SessionID: sessionID, + }) +} + // readLoop reads messages from a WebSocket connection. func (c *PicoChannel) readLoop(pc *picoConn) { defer func() { diff --git a/pkg/channels/pico/pico_test.go b/pkg/channels/pico/pico_test.go index e712767ad..b92c66def 100644 --- a/pkg/channels/pico/pico_test.go +++ b/pkg/channels/pico/pico_test.go @@ -1,17 +1,25 @@ package pico import ( + "bytes" "context" + "encoding/json" "errors" "fmt" + "net/http" + "net/http/httptest" + "strings" "sync" "testing" + "time" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" ) +const testToken = "test-pico-token" + func newTestPicoChannel(t *testing.T) *PicoChannel { t.Helper() @@ -142,3 +150,380 @@ func (c *PicoChannel) addConnForTest(pc *picoConn) { } bySession[pc.id] = pc } + +// --- HTTP message endpoint helpers --- + +// newTestChannelWithConfig creates a PicoChannel and returns it alongside the bus. +// The channel is NOT started — call Start to mark it running. +func newTestChannelWithConfig(t *testing.T, token string, opts ...func(*config.PicoConfig)) (*PicoChannel, *bus.MessageBus) { + t.Helper() + b := bus.NewMessageBus() + cfg := config.PicoConfig{Enabled: true} + cfg.SetToken(token) + for _, opt := range opts { + opt(&cfg) + } + ch, err := NewPicoChannel(cfg, b) + if err != nil { + t.Fatalf("NewPicoChannel: %v", err) + } + return ch, b +} + +// newTestHTTPChannel creates a PicoChannel with default config for HTTP tests. +func newTestHTTPChannel(t *testing.T) *PicoChannel { + t.Helper() + ch, _ := newTestChannelWithConfig(t, testToken) + return ch +} + +// startTestChannel creates, starts, and registers cleanup for a PicoChannel. +func startTestChannel(t *testing.T) *PicoChannel { + t.Helper() + ch := newTestHTTPChannel(t) + if err := ch.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + t.Cleanup(func() { ch.Stop(context.Background()) }) + return ch +} + +// startTestChannelWithBus creates and starts a PicoChannel, returning the bus +// for message verification. +func startTestChannelWithBus(t *testing.T) (*PicoChannel, *bus.MessageBus) { + t.Helper() + ch, b := newTestChannelWithConfig(t, testToken) + if err := ch.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + t.Cleanup(func() { ch.Stop(context.Background()) }) + return ch, b +} + +func authRequest(method, url, body, token string) *http.Request { + req := httptest.NewRequest(method, url, strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+token) + return req +} + +// consumeInbound reads one message from the bus with a short timeout. +func consumeInbound(t *testing.T, b *bus.MessageBus) bus.InboundMessage { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + select { + case msg := <-b.InboundChan(): + return msg + case <-ctx.Done(): + t.Fatal("expected inbound message on bus, got none") + return bus.InboundMessage{} + } +} + +// --- HTTP message rejection tests --- + +func TestHTTPMessage_MethodNotAllowed(t *testing.T) { + ch := startTestChannel(t) + + for _, method := range []string{http.MethodGet, http.MethodPut, http.MethodDelete, http.MethodPatch} { + rec := httptest.NewRecorder() + req := authRequest(method, "/pico/message", `{"content":"hello"}`, testToken) + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusMethodNotAllowed { + t.Errorf("%s: expected %d, got %d", method, http.StatusMethodNotAllowed, rec.Code) + } + } +} + +func TestHTTPMessage_NotRunning(t *testing.T) { + ch := newTestHTTPChannel(t) // not started + + rec := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/pico/message", `{"content":"hello"}`, testToken) + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusServiceUnavailable { + t.Errorf("expected %d, got %d", http.StatusServiceUnavailable, rec.Code) + } +} + +func TestHTTPMessage_MissingAuth(t *testing.T) { + ch := startTestChannel(t) + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/pico/message", strings.NewReader(`{"content":"hello"}`)) + // No Authorization header. + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusUnauthorized { + t.Errorf("expected %d, got %d", http.StatusUnauthorized, rec.Code) + } +} + +func TestHTTPMessage_InvalidAuth(t *testing.T) { + ch := startTestChannel(t) + + rec := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/pico/message", `{"content":"hello"}`, "wrong-token") + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusUnauthorized { + t.Errorf("expected %d, got %d", http.StatusUnauthorized, rec.Code) + } +} + +func TestHTTPMessage_OversizedBody(t *testing.T) { + ch := startTestChannel(t) + + oversized := bytes.Repeat([]byte("A"), maxHTTPBodySize+1) + req := httptest.NewRequest(http.MethodPost, "/pico/message", bytes.NewReader(oversized)) + req.Header.Set("Authorization", "Bearer "+testToken) + rec := httptest.NewRecorder() + + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusRequestEntityTooLarge { + t.Errorf("expected %d, got %d", http.StatusRequestEntityTooLarge, rec.Code) + } +} + +func TestHTTPMessage_AcceptsMaxBodySize(t *testing.T) { + ch := startTestChannel(t) + + // Exactly at the limit — should not trigger 413. + // Body is not valid JSON, so expect 400 (not 413). + body := bytes.Repeat([]byte("A"), maxHTTPBodySize) + req := httptest.NewRequest(http.MethodPost, "/pico/message", bytes.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+testToken) + rec := httptest.NewRecorder() + + ch.handleHTTPMessage(rec, req) + + if rec.Code == http.StatusRequestEntityTooLarge { + t.Error("body at exactly max size should not be rejected as too large") + } +} + +func TestHTTPMessage_MalformedJSON(t *testing.T) { + ch := startTestChannel(t) + + rec := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/pico/message", "not json at all", testToken) + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Errorf("expected %d, got %d", http.StatusBadRequest, rec.Code) + } +} + +func TestHTTPMessage_EmptyContent(t *testing.T) { + ch := startTestChannel(t) + + rec := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/pico/message", `{"content":""}`, testToken) + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Errorf("expected %d, got %d", http.StatusBadRequest, rec.Code) + } +} + +func TestHTTPMessage_WhitespaceContent(t *testing.T) { + ch := startTestChannel(t) + + rec := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/pico/message", `{"content":" "}`, testToken) + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Errorf("expected %d, got %d", http.StatusBadRequest, rec.Code) + } +} + +// --- HTTP message acceptance tests --- + +func TestHTTPMessage_Accepted(t *testing.T) { + ch := startTestChannel(t) + + body := `{"content":"hello","session_id":"test-session"}` + rec := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/pico/message", body, testToken) + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusAccepted { + t.Fatalf("expected %d, got %d", http.StatusAccepted, rec.Code) + } + + var resp httpMessageResponse + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("decode response: %v", err) + } + if !resp.OK { + t.Error("expected ok=true") + } + if resp.Status != "accepted" { + t.Errorf("expected status=accepted, got %s", resp.Status) + } + if resp.SessionID != "test-session" { + t.Errorf("expected session_id=test-session, got %s", resp.SessionID) + } +} + +func TestHTTPMessage_GeneratesSessionID(t *testing.T) { + ch := startTestChannel(t) + + rec := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/pico/message", `{"content":"hello"}`, testToken) + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusAccepted { + t.Fatalf("expected %d, got %d", http.StatusAccepted, rec.Code) + } + + var resp httpMessageResponse + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("decode response: %v", err) + } + if resp.SessionID == "" { + t.Error("expected non-empty generated session_id") + } +} + +// --- HTTP message bus semantics tests --- + +func TestHTTPMessage_PublishesBusMessage(t *testing.T) { + ch, b := startTestChannelWithBus(t) + + body := `{"content":"hello world","session_id":"sess-42"}` + rec := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/pico/message", body, testToken) + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusAccepted { + t.Fatalf("expected %d, got %d", http.StatusAccepted, rec.Code) + } + + msg := consumeInbound(t, b) + + if msg.Channel != "pico" { + t.Errorf("Channel: expected pico, got %s", msg.Channel) + } + if msg.Content != "hello world" { + t.Errorf("Content: expected hello world, got %s", msg.Content) + } + if msg.ChatID != "pico:sess-42" { + t.Errorf("ChatID: expected pico:sess-42, got %s", msg.ChatID) + } + if msg.Peer.Kind != "direct" { + t.Errorf("Peer.Kind: expected direct, got %s", msg.Peer.Kind) + } + if msg.Peer.ID != "pico:sess-42" { + t.Errorf("Peer.ID: expected pico:sess-42, got %s", msg.Peer.ID) + } + if msg.MessageID == "" { + t.Error("MessageID: expected non-empty generated ID") + } + if msg.Metadata["platform"] != "pico" { + t.Errorf("Metadata[platform]: expected pico, got %s", msg.Metadata["platform"]) + } + if msg.Metadata["session_id"] != "sess-42" { + t.Errorf("Metadata[session_id]: expected sess-42, got %s", msg.Metadata["session_id"]) + } + if msg.Metadata["transport"] != "http" { + t.Errorf("Metadata[transport]: expected http, got %s", msg.Metadata["transport"]) + } + // SenderID is resolved to canonical form by BaseChannel.HandleMessage. + if msg.SenderID != "pico:pico-user" { + t.Errorf("SenderID: expected pico:pico-user, got %s", msg.SenderID) + } + if msg.Sender.Platform != "pico" { + t.Errorf("Sender.Platform: expected pico, got %s", msg.Sender.Platform) + } +} + +func TestHTTPMessage_PreservesContentWhitespace(t *testing.T) { + ch, b := startTestChannelWithBus(t) + + // Content has leading/trailing whitespace — should be preserved on the bus + // (matching WebSocket message.send semantics). + body := `{"content":" hello world ","session_id":"ws-compat"}` + rec := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/pico/message", body, testToken) + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusAccepted { + t.Fatalf("expected %d, got %d", http.StatusAccepted, rec.Code) + } + + msg := consumeInbound(t, b) + if msg.Content != " hello world " { + t.Errorf("Content: expected ' hello world ', got %q", msg.Content) + } +} + +// --- HTTP message auth variant tests --- + +func TestHTTPMessage_QueryTokenAccepted(t *testing.T) { + ch, _ := newTestChannelWithConfig(t, testToken, func(cfg *config.PicoConfig) { + cfg.AllowTokenQuery = true + }) + if err := ch.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + t.Cleanup(func() { ch.Stop(context.Background()) }) + + req := httptest.NewRequest( + http.MethodPost, + "/pico/message?token="+testToken, + strings.NewReader(`{"content":"hello"}`), + ) + rec := httptest.NewRecorder() + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusAccepted { + t.Errorf("expected %d, got %d", http.StatusAccepted, rec.Code) + } +} + +func TestHTTPMessage_QueryTokenRejectedWhenDisabled(t *testing.T) { + ch := startTestChannel(t) // AllowTokenQuery defaults to false + + req := httptest.NewRequest( + http.MethodPost, + "/pico/message?token="+testToken, + strings.NewReader(`{"content":"hello"}`), + ) + rec := httptest.NewRecorder() + ch.handleHTTPMessage(rec, req) + + if rec.Code != http.StatusUnauthorized { + t.Errorf("expected %d, got %d", http.StatusUnauthorized, rec.Code) + } +} + +// --- HTTP message routing tests --- + +func TestServeHTTP_RoutesMessage(t *testing.T) { + ch := startTestChannel(t) + + rec := httptest.NewRecorder() + req := authRequest(http.MethodPost, "/pico/message", `{"content":"routed"}`, testToken) + ch.ServeHTTP(rec, req) + + if rec.Code != http.StatusAccepted { + t.Errorf("expected %d, got %d", http.StatusAccepted, rec.Code) + } +} + +func TestServeHTTP_UnknownPathReturns404(t *testing.T) { + ch := startTestChannel(t) + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/pico/unknown", nil) + ch.ServeHTTP(rec, req) + + if rec.Code != http.StatusNotFound { + t.Errorf("expected %d, got %d", http.StatusNotFound, rec.Code) + } +}