From 0b036645bb8add36547bee20183d804b295eca73 Mon Sep 17 00:00:00 2001 From: Aska Date: Sun, 6 Sep 2026 06:37:16 +0800 Subject: [PATCH] =?UTF-8?q?feat(channels/webhook):=20=E5=85=A5=E7=AB=99?= =?UTF-8?q?=E5=9B=9E=E8=B0=83=E6=B8=A0=E9=81=93=E6=8F=92=E4=BB=B6=EF=BC=88?= =?UTF-8?q?JSON/Form=20=E8=A7=A3=E6=9E=90=20+=20HMAC=20=E7=AD=BE=E5=90=8D?= =?UTF-8?q?=20+=20InboundHandler=20=E4=BE=9B=20gateway=20=E6=8C=82?= =?UTF-8?q?=E8=BD=BD=EF=BC=8C=E4=B8=BB=E5=8A=A8=E6=8E=A8=E9=80=81=E4=B8=8D?= =?UTF-8?q?=E6=94=AF=E6=8C=81=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugins/channels/webhook/adapter.go | 272 ++++++++++++++++++ plugins/channels/webhook/handler.go | 170 +++++++++++ plugins/channels/webhook/inbound_test.go | 343 +++++++++++++++++++++++ plugins/channels/webhook/parse.go | 227 +++++++++++++++ plugins/channels/webhook/webhook_test.go | 181 ++++++++++++ 5 files changed, 1193 insertions(+) create mode 100644 plugins/channels/webhook/adapter.go create mode 100644 plugins/channels/webhook/handler.go create mode 100644 plugins/channels/webhook/inbound_test.go create mode 100644 plugins/channels/webhook/parse.go create mode 100644 plugins/channels/webhook/webhook_test.go diff --git a/plugins/channels/webhook/adapter.go b/plugins/channels/webhook/adapter.go new file mode 100644 index 0000000..1f0ecba --- /dev/null +++ b/plugins/channels/webhook/adapter.go @@ -0,0 +1,272 @@ +// Package webhook implements the webhook inbound channel adapter (design/08 +// ยง4). A webhook channel receives external HTTP callbacks (JSON or form posts) +// and normalizes them into UnifiedMessage so the runtime can reason about them +// like any other channel. +// +// It is inbound-only: the adapter never pushes messages out, so +// Capabilities.Proactive is false and SendMessage/SendProactive return +// adapter.ErrUnsupported. The gateway mounts the exposed InboundHandler at the +// configured endpoint path (M8). HMAC-SHA256 signature verification is opt-in +// via Config.Secret. +// +// The adapter lives behind adapter.PlatformAdapter so the runtime can treat it +// uniformly with the other channels declared in design/08 (qq-obv11, +// weixin-ilink, telegram). Configuration is sourced externally and injected at +// construction; no values are hard-coded beyond safe fallbacks. +package webhook + +import ( + "context" + "fmt" + "log/slog" + "net/http" + "sync" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +const ( + // PlatformName is the identifier returned by Name and stamped onto every + // inbound UnifiedMessage so consumers can attribute the origin. + PlatformName = "webhook" + + // source is the logger provenance tag required by the logging rule. + source = "plugins/channels/webhook" + + // defaultEndpointPath is used when Config.EndpointPath is empty. + defaultEndpointPath = "/webhook" + // defaultSignatureHeader carries the HMAC signature when Config.Secret is set. + defaultSignatureHeader = "X-Webhook-Signature" + // defaultMaxBodyBytes caps an inbound callback payload at 1 MiB to protect + // against oversized bodies, per the streaming/no-ReadAll-large-files rule. + defaultMaxBodyBytes = 1 << 20 +) + +// Config configures the webhook inbound adapter. It is decoded from the channel +// config map via adapter.DecodeConfig; every field is optional and meaningfully +// defaulted so an empty config yields a usable endpoint. +type Config struct { + // EndpointPath is the route the gateway mounts the inbound handler at. + EndpointPath string `json:"endpointPath"` + // ChannelID identifies this webhook channel; when set it is reported by + // ListChannels. Empty means no channel is pre-registered. + ChannelID string `json:"channelId"` + // ChannelName is the human-readable channel name for ListChannels. + ChannelName string `json:"channelName"` + // Secret enables HMAC-SHA256 signature verification when non-empty. The + // signature is computed over the raw request body and passed in the header + // named by SignatureHeader. Empty disables verification. + Secret string `json:"secret"` + // SignatureHeader names the header carrying the signature. Defaults to + // "X-Webhook-Signature" when empty. + SignatureHeader string `json:"signatureHeader"` + // Methods are the accepted HTTP methods. Defaults to ["POST"]. + Methods []string `json:"methods"` +} + +// Adapter implements adapter.PlatformAdapter for the webhook channel. It is an +// inbound-only adapter: callbacks arrive through the InboundHandler and are +// delivered to the installed message handler (or the emit callback captured by +// ReadMessages). No external session is held, so Connect/Disconnect only track +// lifecycle state. +type Adapter struct { + log *slog.Logger + client *http.Client + cfg Config + + mu sync.RWMutex + connected bool + handler adapter.MessageHandler + emit func(adapter.UnifiedMessage) +} + +// Compile-time assertion that Adapter satisfies the channel contract. +var _ adapter.PlatformAdapter = (*Adapter)(nil) + +// Option adjusts an Adapter at construction time. +type Option func(*Adapter) + +// WithConfig sets the adapter configuration. +func WithConfig(cfg Config) Option { + return func(a *Adapter) { a.cfg = cfg } +} + +// WithClient injects the HTTP client used for any outbound call. The webhook +// adapter is inbound-only and never uses it, but the field is kept so callers +// that later add outbound delivery do not change the constructor. +func WithClient(c *http.Client) Option { + return func(a *Adapter) { a.client = c } +} + +// WithLogger sets the structured logger used by the adapter. A nil logger +// falls back to a default tagged logger so logging never panics. +func WithLogger(l *slog.Logger) Option { + return func(a *Adapter) { a.log = l } +} + +// New builds a webhook adapter. It defaults the HTTP client to +// http.DefaultClient and a source-tagged structured logger, applies opts, then +// normalizes the configuration so a minimal config still yields a working +// endpoint. +func New(opts ...Option) *Adapter { + a := &Adapter{ + client: http.DefaultClient, + log: logger.WithSource(logger.New(logger.Options{Format: string(logger.FormatText), Level: "info"}), source), + } + for _, o := range opts { + o(a) + } + a.normalize() + return a +} + +// normalize fills zero-valued config with safe defaults so an empty Config is +// usable. It is called once at construction; the config is immutable afterward. +func (a *Adapter) normalize() { + if a.cfg.EndpointPath == "" { + a.cfg.EndpointPath = defaultEndpointPath + } + if a.cfg.SignatureHeader == "" { + a.cfg.SignatureHeader = defaultSignatureHeader + } + if len(a.cfg.Methods) == 0 { + a.cfg.Methods = []string{http.MethodPost} + } + if a.client == nil { + a.client = http.DefaultClient + } + if a.log == nil { + a.log = logger.WithSource(logger.New(logger.Options{Format: string(logger.FormatText), Level: "info"}), source) + } +} + +// Name returns the platform identifier "webhook". +func (a *Adapter) Name() string { return PlatformName } + +// Config returns a snapshot of the current configuration as a plain map for +// inspection or persistence. +func (a *Adapter) Config() map[string]any { + return map[string]any{ + "endpointPath": a.cfg.EndpointPath, + "channelId": a.cfg.ChannelID, + "channelName": a.cfg.ChannelName, + "secret": a.cfg.Secret, + "signatureHeader": a.cfg.SignatureHeader, + "methods": a.cfg.Methods, + } +} + +// Connect marks the adapter connected (idempotent). No external session is +// established; the gateway registers the route by mounting InboundHandler, so +// Connect only transitions lifecycle state for the caller's benefit. +func (a *Adapter) Connect(_ context.Context) error { + a.mu.Lock() + defer a.mu.Unlock() + if a.connected { + return nil + } + a.connected = true + a.log.Info("webhook channel connected", "endpointPath", a.cfg.EndpointPath) + return nil +} + +// Disconnect tears down the adapter (idempotent): it clears the connected flag +// and drops any installed handler/emit so no callback is delivered after a +// shutdown. +func (a *Adapter) Disconnect(_ context.Context) error { + a.mu.Lock() + defer a.mu.Unlock() + if !a.connected { + return nil + } + a.connected = false + a.handler = nil + a.emit = nil + a.log.Info("webhook channel disconnected") + return nil +} + +// IsConnected reports whether the adapter is currently connected. +func (a *Adapter) IsConnected() bool { + a.mu.RLock() + defer a.mu.RUnlock() + return a.connected +} + +// HealthCheck verifies the adapter is usable. As an inbound-only adapter that +// holds no session, it only reports the lifecycle state. +func (a *Adapter) HealthCheck(_ context.Context) error { + if !a.IsConnected() { + return fmt.Errorf("webhook: %w", adapter.ErrNotConnected) + } + return nil +} + +// ReadMessages installs emit as a fallback delivery target, then blocks until +// ctx is canceled. Delivery is push-style via the message handler, matching the +// documented behaviour for push adapters; emit is only exercised when no +// handler is installed. +func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMessage)) { + a.mu.Lock() + a.emit = emit + a.mu.Unlock() + + <-ctx.Done() + + a.mu.Lock() + a.emit = nil + a.mu.Unlock() +} + +// SendMessage is unsupported: the webhook channel is inbound-only and has no +// outbound delivery path. It returns adapter.ErrUnsupported. +func (a *Adapter) SendMessage(_ context.Context, _ adapter.UnifiedResponse) error { + return fmt.Errorf("webhook: send message: %w", adapter.ErrUnsupported) +} + +// SendProactive is unsupported for the same reason as SendMessage: +// Capabilities.Proactive is false. It returns adapter.ErrUnsupported. +func (a *Adapter) SendProactive(_ context.Context, _ adapter.UnifiedMessage) error { + return fmt.Errorf("webhook: proactive push: %w", adapter.ErrUnsupported) +} + +// SetMessageHandler installs the callback receiving push-style inbound +// messages. A nil fn clears the handler. +func (a *Adapter) SetMessageHandler(fn adapter.MessageHandler) { + a.mu.Lock() + defer a.mu.Unlock() + a.handler = fn +} + +// Capabilities declares what the webhook channel supports. Because it is +// inbound-only, Proactive is false; image is reported so consumers know media +// URLs may appear in inbound messages. +func (a *Adapter) Capabilities() adapter.Capabilities { + return adapter.Capabilities{ + MaxMessageLength: 8192, + Markdown: true, + Image: true, + Voice: false, + Reaction: false, + TypingHint: false, + RecommendBurstMax: 1, + Proactive: false, + }.Normalize() +} + +// ListChannels returns the channel declared by Config.ChannelID when present, +// otherwise an empty list: the webhook endpoint is a single generic channel +// whose identity only becomes concrete per inbound message. +func (a *Adapter) ListChannels(_ context.Context) ([]adapter.ChannelInfo, error) { + if a.cfg.ChannelID == "" { + return nil, nil + } + return []adapter.ChannelInfo{ + { + ChannelID: a.cfg.ChannelID, + ChannelType: adapter.ChannelDirect, + ChannelName: a.cfg.ChannelName, + }, + }, nil +} diff --git a/plugins/channels/webhook/handler.go b/plugins/channels/webhook/handler.go new file mode 100644 index 0000000..60cbc36 --- /dev/null +++ b/plugins/channels/webhook/handler.go @@ -0,0 +1,170 @@ +package webhook + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// Inbound HTTP callback errors returned by the handler. They are sentinels so +// callers and tests can discriminate failure modes with errors.Is. +var ( + errSignatureMissing = errors.New("webhook: missing signature header") + errSignatureMismatch = errors.New("webhook: signature mismatch") + errNoInboundHandler = errors.New("webhook: no inbound handler installed") + errOversizedBody = errors.New("webhook: request body too large") +) + +// InboundHandler returns the http.Handler the gateway mounts at the configured +// endpoint path (M8). It accepts JSON or form callbacks, verifies the HMAC +// signature when a secret is configured, normalizes the payload to +// UnifiedMessage, and delivers it to the installed message handler. +// +// A handler is constructed from the adapter's immutable config, so it is safe +// to mount once and serve concurrently. +func (a *Adapter) InboundHandler() http.Handler { + return http.HandlerFunc(a.serveInbound) +} + +// serveInbound is the HTTP entry point for an inbound callback. +func (a *Adapter) serveInbound(w http.ResponseWriter, r *http.Request) { + if !a.methodAllowed(r.Method) { + writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "webhook: method not allowed") + return + } + + body, err := readBody(w, r) + if err != nil { + writeError(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + + if err := a.verifySignature(r, body); err != nil { + writeError(w, http.StatusUnauthorized, "unauthorized", err.Error()) + return + } + + msg, err := parseInbound(r.Header.Get("Content-Type"), body) + if err != nil { + writeError(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + if err := msg.Validate(); err != nil { + writeError(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + + if err := a.deliver(r.Context(), msg); err != nil { + writeError(w, http.StatusServiceUnavailable, "unavailable", err.Error()) + return + } + + writeData(w, http.StatusOK, map[string]any{"accepted": true, "messageId": msg.MessageID}) +} + +// methodAllowed reports whether m is one of the adapter's accepted methods. +func (a *Adapter) methodAllowed(m string) bool { + for _, allowed := range a.cfg.Methods { + if allowed == m { + return true + } + } + return false +} + +// readBody reads and size-caps the request body so a malicious or misconfigured +// caller cannot exhaust memory. It returns errOversizedBody when the cap is hit. +func readBody(w http.ResponseWriter, r *http.Request) ([]byte, error) { + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, defaultMaxBodyBytes)) + if err != nil { + if isMaxBytesError(err) { + return nil, fmt.Errorf("%w", errOversizedBody) + } + return nil, fmt.Errorf("webhook: read body: %w", err) + } + return body, nil +} + +// isMaxBytesError reports whether err was produced by a MaxBytesReader overrun. +func isMaxBytesError(err error) bool { + var mbe *http.MaxBytesError + return errors.As(err, &mbe) +} + +// verifySignature checks the HMAC-SHA256 signature over the raw body when a +// secret is configured. An empty secret disables verification entirely. +func (a *Adapter) verifySignature(r *http.Request, body []byte) error { + if a.cfg.Secret == "" { + return nil + } + got := r.Header.Get(a.cfg.SignatureHeader) + if got == "" { + return fmt.Errorf("%w: header %q", errSignatureMissing, a.cfg.SignatureHeader) + } + mac := hmac.New(sha256.New, []byte(a.cfg.Secret)) + _, _ = mac.Write(body) + want := hex.EncodeToString(mac.Sum(nil)) + if !hmac.Equal([]byte(got), []byte(want)) { + return fmt.Errorf("%w", errSignatureMismatch) + } + return nil +} + +// deliver routes a normalized inbound message to the installed message handler, +// falling back to the emit callback captured by ReadMessages when no handler is +// set. It returns errNoInboundHandler if neither is available. +func (a *Adapter) deliver(ctx context.Context, msg adapter.UnifiedMessage) error { + a.mu.RLock() + handler := a.handler + emit := a.emit + a.mu.RUnlock() + + if handler != nil { + handler(ctx, msg) + return nil + } + if emit != nil { + emit(msg) + return nil + } + return fmt.Errorf("%w", errNoInboundHandler) +} + +// dataEnvelope is the success response shape `{"data": ...}`, matching the +// project's versioned API contract. +type dataEnvelope struct { + Data any `json:"data"` +} + +// errorEnvelope is the failure response shape `{"error":{code,message}}`. +type errorEnvelope struct { + Error errorBody `json:"error"` +} + +// errorBody carries a machine code and a human message. +type errorBody struct { + Code string `json:"code"` + Message string `json:"message"` +} + +// writeData writes a JSON success envelope with the given status. +func writeData(w http.ResponseWriter, status int, data any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(dataEnvelope{Data: data}) +} + +// writeError writes a JSON error envelope with the given status. +func writeError(w http.ResponseWriter, status int, code, message string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(errorEnvelope{Error: errorBody{Code: code, Message: message}}) +} diff --git a/plugins/channels/webhook/inbound_test.go b/plugins/channels/webhook/inbound_test.go new file mode 100644 index 0000000..d2ddb02 --- /dev/null +++ b/plugins/channels/webhook/inbound_test.go @@ -0,0 +1,343 @@ +package webhook + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// recorder captures inbound messages delivered to a message handler. +type recorder struct { + mu sync.Mutex + msgs []adapter.UnifiedMessage +} + +func (r *recorder) handle(_ context.Context, m adapter.UnifiedMessage) { + r.mu.Lock() + defer r.mu.Unlock() + r.msgs = append(r.msgs, m) +} + +func (r *recorder) snapshot() []adapter.UnifiedMessage { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]adapter.UnifiedMessage, len(r.msgs)) + copy(out, r.msgs) + return out +} + +func (r *recorder) last() (adapter.UnifiedMessage, bool) { + msgs := r.snapshot() + if len(msgs) == 0 { + return adapter.UnifiedMessage{}, false + } + return msgs[len(msgs)-1], true +} + +// recAdapter returns an adapter with a recording handler installed and a +// connected lifecycle state. +func recAdapter() (*Adapter, *recorder) { + rec := &recorder{} + a := New() + a.SetMessageHandler(rec.handle) + _ = a.Connect(context.Background()) + return a, rec +} + +func TestInboundMethodNotAllowed(t *testing.T) { + t.Parallel() + + a := New() + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/webhook", nil) + a.InboundHandler().ServeHTTP(rec, req) + + if rec.Code != http.StatusMethodNotAllowed { + t.Errorf("status = %d, want 405", rec.Code) + } +} + +func TestInboundJSONDirect(t *testing.T) { + t.Parallel() + + a, rec := recAdapter() + body := `{"messageId":"m-1","senderId":"u-1","senderName":"Alice","content":"hello","contentType":"text","channelType":"direct"}` + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", bytes.NewReader([]byte(body))) + req.Header.Set("Content-Type", "application/json") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec2.Code, rec2.Body.String()) + } + msg, ok := rec.last() + if !ok { + t.Fatal("no message delivered") + } + if msg.Platform != "webhook" { + t.Errorf("Platform = %q, want webhook", msg.Platform) + } + if msg.Content != "hello" { + t.Errorf("Content = %q, want hello", msg.Content) + } + if msg.SenderID != "u-1" || msg.SenderName != "Alice" { + t.Errorf("sender = %q/%q, want u-1/Alice", msg.SenderID, msg.SenderName) + } + if msg.ChannelType != adapter.ChannelDirect { + t.Errorf("ChannelType = %q, want direct", msg.ChannelType) + } + if msg.ContentType != adapter.ContentText { + t.Errorf("ContentType = %q, want text", msg.ContentType) + } + if msg.Sender.Role != adapter.RoleUnknown { + t.Errorf("Sender.Role = %q, want unknown", msg.Sender.Role) + } + if err := msg.Validate(); err != nil { + t.Errorf("Validate() error = %v", err) + } +} + +func TestInboundJSONGroupWithMedia(t *testing.T) { + t.Parallel() + + a, rec := recAdapter() + body := `{ + "messageId":"m-2","senderId":"u-9","senderName":"Bob","content":"pic", + "contentType":"image","channelType":"group","groupId":"g-1","groupName":"Team", + "media":[{"kind":"image","url":"https://cdn.example/a.png","name":"a.png","size":1024,"mimeType":"image/png"}] + }` + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", bytes.NewReader([]byte(body))) + req.Header.Set("Content-Type", "application/json") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec2.Code, rec2.Body.String()) + } + msg, ok := rec.last() + if !ok { + t.Fatal("no message delivered") + } + if msg.ChannelType != adapter.ChannelGroup { + t.Errorf("ChannelType = %q, want group", msg.ChannelType) + } + if msg.GroupID != "g-1" || msg.GroupName != "Team" { + t.Errorf("group = %q/%q, want g-1/Team", msg.GroupID, msg.GroupName) + } + if msg.ChannelID != "g-1" { + t.Errorf("ChannelID = %q, want g-1 (group channel)", msg.ChannelID) + } + if !msg.Sender.IsGroupChat() { + t.Error("Sender.IsGroupChat() = false, want true") + } + if len(msg.Media) != 1 { + t.Fatalf("Media len = %d, want 1", len(msg.Media)) + } + if msg.Media[0].URL != "https://cdn.example/a.png" { + t.Errorf("Media[0].URL = %q, want cdn url", msg.Media[0].URL) + } + if err := msg.Validate(); err != nil { + t.Errorf("Validate() error = %v", err) + } +} + +func TestInboundFormDirect(t *testing.T) { + t.Parallel() + + a, rec := recAdapter() + body := "messageId=m-3&senderId=u-2&senderName=Carol&content=hey&contentType=text&channelType=direct" + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec2.Code, rec2.Body.String()) + } + msg, ok := rec.last() + if !ok { + t.Fatal("no message delivered") + } + if msg.Content != "hey" || msg.SenderID != "u-2" || msg.SenderName != "Carol" { + t.Errorf("msg = %+v, want content=hey sender=u-2/Carol", msg) + } + if msg.ContentType != adapter.ContentText || msg.ChannelType != adapter.ChannelDirect { + t.Errorf("type = %q/%q, want text/direct", msg.ContentType, msg.ChannelType) + } +} + +func TestInboundFormGroupWithMediaURL(t *testing.T) { + t.Parallel() + + a, rec := recAdapter() + body := "channelType=group&groupId=g-7&senderId=u-4&content=clip&contentType=image&mediaUrl=https://cdn.example/b.png&mediaKind=image&mediaName=b.png" + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec2.Code, rec2.Body.String()) + } + msg, ok := rec.last() + if !ok { + t.Fatal("no message delivered") + } + if msg.ChannelType != adapter.ChannelGroup || msg.GroupID != "g-7" { + t.Errorf("channel = %q/%q, want group/g-7", msg.ChannelType, msg.GroupID) + } + if len(msg.Media) != 1 || msg.Media[0].URL != "https://cdn.example/b.png" { + t.Errorf("Media = %+v, want single url media", msg.Media) + } +} + +func TestInboundSignatureValid(t *testing.T) { + t.Parallel() + + secret := "shared-secret" + a := New(WithConfig(Config{Secret: secret})) + rec := &recorder{} + a.SetMessageHandler(rec.handle) + + body := []byte(`{"content":"signed"}`) + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set(defaultSignatureHeader, signBody(secret, body)) + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec2.Code, rec2.Body.String()) + } + if _, ok := rec.last(); !ok { + t.Fatal("no message delivered") + } +} + +func TestInboundSignatureMissing(t *testing.T) { + t.Parallel() + + a := New(WithConfig(Config{Secret: "shared-secret"})) + rec := &recorder{} + a.SetMessageHandler(rec.handle) + + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader(`{"content":"x"}`)) + req.Header.Set("Content-Type", "application/json") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusUnauthorized { + t.Errorf("status = %d, want 401", rec2.Code) + } + if _, ok := rec.last(); ok { + t.Error("message delivered despite missing signature") + } +} + +func TestInboundSignatureMismatch(t *testing.T) { + t.Parallel() + + a := New(WithConfig(Config{Secret: "shared-secret"})) + rec := &recorder{} + a.SetMessageHandler(rec.handle) + + body := []byte(`{"content":"x"}`) + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set(defaultSignatureHeader, "deadbeef") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusUnauthorized { + t.Errorf("status = %d, want 401", rec2.Code) + } +} + +func TestInboundBadJSON(t *testing.T) { + t.Parallel() + + a := New() + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader("{not-json")) + req.Header.Set("Content-Type", "application/json") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400", rec2.Code) + } +} + +func TestInboundUnsupportedContentType(t *testing.T) { + t.Parallel() + + a := New() + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader("hello")) + req.Header.Set("Content-Type", "text/plain") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400", rec2.Code) + } +} + +func TestInboundGroupWithoutGroupID(t *testing.T) { + t.Parallel() + + a := New() + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader(`{"channelType":"group","content":"hi"}`)) + req.Header.Set("Content-Type", "application/json") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400", rec2.Code) + } +} + +func TestInboundNoHandlerDelivers503(t *testing.T) { + t.Parallel() + + a := New() + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader(`{"content":"hi"}`)) + req.Header.Set("Content-Type", "application/json") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusServiceUnavailable { + t.Errorf("status = %d, want 503", rec2.Code) + } +} + +func TestInboundAcceptsCharsetInContentType(t *testing.T) { + t.Parallel() + + a, rec := recAdapter() + rec2 := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader(`{"content":"ok"}`)) + req.Header.Set("Content-Type", "application/json; charset=utf-8") + a.InboundHandler().ServeHTTP(rec2, req) + + if rec2.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec2.Code, rec2.Body.String()) + } + if _, ok := rec.last(); !ok { + t.Fatal("no message delivered") + } +} + +// signBody computes the HMAC-SHA256 hex signature over body with secret. +func signBody(secret string, body []byte) string { + mac := hmac.New(sha256.New, []byte(secret)) + _, _ = mac.Write(body) + return hex.EncodeToString(mac.Sum(nil)) +} diff --git a/plugins/channels/webhook/parse.go b/plugins/channels/webhook/parse.go new file mode 100644 index 0000000..fde9afa --- /dev/null +++ b/plugins/channels/webhook/parse.go @@ -0,0 +1,227 @@ +package webhook + +import ( + "encoding/json" + "fmt" + "mime" + "net/url" + "strings" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// jsonPayload is the normalized wire shape shared by the JSON and form decoders. +// Field names are lowerCamelCase to match the project's API convention; form +// values use the same keys. Missing fields fall back to safe defaults. +type jsonPayload struct { + MessageID string `json:"messageId"` + ChannelType string `json:"channelType"` + SenderID string `json:"senderId"` + SenderName string `json:"senderName"` + Content string `json:"content"` + ContentType string `json:"contentType"` + Media []jsonMedia `json:"media"` + Mentions []jsonMention `json:"mentions"` + ReplyToID string `json:"replyToId"` + ReplyToText string `json:"replyToText"` + GroupID string `json:"groupId"` + GroupName string `json:"groupName"` + ThreadID string `json:"threadId"` + Timestamp string `json:"timestamp"` +} + +// jsonMedia is a single attachment carried by an inbound message. +type jsonMedia struct { + Kind string `json:"kind"` + URL string `json:"url"` + Name string `json:"name"` + Size int64 `json:"size"` + MIMEType string `json:"mimeType"` +} + +// jsonMention is a reference to a user inside the message body. +type jsonMention struct { + UserID string `json:"userId"` + UserName string `json:"userName"` +} + +// parseInbound dispatches on the request Content-Type to select the payload +// decoder. An empty or unparseable content type defaults to JSON, since JSON is +// the common callback encoding. +func parseInbound(contentType string, body []byte) (adapter.UnifiedMessage, error) { + mediaType, _, err := mime.ParseMediaType(contentType) + if err != nil { + if strings.TrimSpace(contentType) == "" { + return parseJSON(body) + } + return adapter.UnifiedMessage{}, fmt.Errorf("webhook: bad content type %q: %w", contentType, err) + } + switch mediaType { + case "application/json": + return parseJSON(body) + case "application/x-www-form-urlencoded": + return parseForm(body) + default: + return adapter.UnifiedMessage{}, fmt.Errorf("webhook: unsupported content type %q", mediaType) + } +} + +// parseJSON decodes a JSON callback body into a UnifiedMessage. +func parseJSON(body []byte) (adapter.UnifiedMessage, error) { + var p jsonPayload + if err := json.Unmarshal(body, &p); err != nil { + return adapter.UnifiedMessage{}, fmt.Errorf("webhook: decode json: %w", err) + } + return p.toUnified(), nil +} + +// parseForm decodes an application/x-www-form-urlencoded callback body into a +// UnifiedMessage. Scalar fields map to form keys; media and mentions are +// carried as JSON-encoded values, and a single mediaUrl/mediaKind pair is +// accepted as a shorthand. +func parseForm(body []byte) (adapter.UnifiedMessage, error) { + values, err := url.ParseQuery(string(body)) + if err != nil { + return adapter.UnifiedMessage{}, fmt.Errorf("webhook: decode form: %w", err) + } + p := jsonPayload{ + MessageID: values.Get("messageId"), + ChannelType: values.Get("channelType"), + SenderID: values.Get("senderId"), + SenderName: values.Get("senderName"), + Content: values.Get("content"), + ContentType: values.Get("contentType"), + ReplyToID: values.Get("replyToId"), + ReplyToText: values.Get("replyToText"), + GroupID: values.Get("groupId"), + GroupName: values.Get("groupName"), + ThreadID: values.Get("threadId"), + Timestamp: values.Get("timestamp"), + } + if err := decodeJSONField(values.Get("media"), &p.Media); err != nil { + return adapter.UnifiedMessage{}, fmt.Errorf("webhook: decode media: %w", err) + } + if err := decodeJSONField(values.Get("mentions"), &p.Mentions); err != nil { + return adapter.UnifiedMessage{}, fmt.Errorf("webhook: decode mentions: %w", err) + } + if len(p.Media) == 0 { + if u := values.Get("mediaUrl"); u != "" { + p.Media = []jsonMedia{{ + Kind: values.Get("mediaKind"), + URL: u, + Name: values.Get("mediaName"), + MIMEType: values.Get("mediaMimeType"), + }} + } + } + return p.toUnified(), nil +} + +// decodeJSONField decodes a JSON-encoded form value into v when present. +func decodeJSONField(s string, v any) error { + if s == "" { + return nil + } + return json.Unmarshal([]byte(s), v) +} + +// toUnified converts the wire payload into the normalized UnifiedMessage, +// applying platform/direct/text defaults and attaching the strongly typed actor +// identity metadata. It does not validate; callers run Validate separately so +// an invalid payload yields a clear HTTP error. +func (p jsonPayload) toUnified() adapter.UnifiedMessage { + ct := adapter.ChannelType(p.ChannelType) + if ct == "" { + ct = adapter.ChannelDirect + } + contentType := adapter.ContentType(p.ContentType) + if contentType == "" { + contentType = adapter.ContentText + } + + channelID := p.SenderID + if ct == adapter.ChannelGroup { + channelID = p.GroupID + } + + return adapter.UnifiedMessage{ + Platform: PlatformName, + ChannelID: channelID, + ChannelType: ct, + SenderID: p.SenderID, + SenderName: p.SenderName, + Sender: buildActor(p), + Content: p.Content, + ContentType: contentType, + Media: toMedia(p.Media), + Mentions: toMentions(p.Mentions), + ReplyToID: p.ReplyToID, + ReplyToText: p.ReplyToText, + GroupID: p.GroupID, + GroupName: p.GroupName, + ThreadID: p.ThreadID, + MessageID: p.MessageID, + Timestamp: parseTimestamp(p.Timestamp), + } +} + +// buildActor derives the strongly typed actor metadata. The adapter fills +// identity fields but leaves Role as RoleUnknown so owner/member resolution +// stays with the strategy engine, per the identity red line. +func buildActor(p jsonPayload) adapter.Actor { + actor := adapter.Actor{ + Role: adapter.RoleUnknown, + UserID: p.SenderID, + DisplayName: p.SenderName, + } + if p.GroupID != "" { + actor.IsGroup = true + actor.GroupID = p.GroupID + actor.GroupName = p.GroupName + } + return actor +} + +// toMedia converts the wire media list into the normalized Media slice. +func toMedia(list []jsonMedia) []adapter.Media { + if len(list) == 0 { + return nil + } + out := make([]adapter.Media, 0, len(list)) + for _, im := range list { + out = append(out, adapter.Media{ + Kind: adapter.MediaKind(im.Kind), + URL: im.URL, + Name: im.Name, + Size: im.Size, + MIMEType: im.MIMEType, + }) + } + return out +} + +// toMentions converts the wire mention list into the normalized Mention slice. +func toMentions(list []jsonMention) []adapter.Mention { + if len(list) == 0 { + return nil + } + out := make([]adapter.Mention, 0, len(list)) + for _, im := range list { + out = append(out, adapter.Mention{UserID: im.UserID, UserName: im.UserName}) + } + return out +} + +// parseTimestamp parses an RFC3339 timestamp, returning the zero time on any +// invalid or empty value so downstream code can decide how to handle it. +func parseTimestamp(s string) time.Time { + if s == "" { + return time.Time{} + } + t, err := time.Parse(time.RFC3339Nano, s) + if err != nil { + return time.Time{} + } + return t +} diff --git a/plugins/channels/webhook/webhook_test.go b/plugins/channels/webhook/webhook_test.go new file mode 100644 index 0000000..e2af46b --- /dev/null +++ b/plugins/channels/webhook/webhook_test.go @@ -0,0 +1,181 @@ +package webhook + +import ( + "bytes" + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestName(t *testing.T) { + t.Parallel() + if got := New().Name(); got != "webhook" { + t.Errorf("Name() = %q, want webhook", got) + } +} + +func TestConfigSnapshot(t *testing.T) { + t.Parallel() + + cfg := Config{ + EndpointPath: "/cb", + ChannelID: "ch-1", + ChannelName: "My Hook", + Secret: "s3cr3t", + Methods: []string{http.MethodPost, http.MethodPut}, + } + got := New(WithConfig(cfg)).Config() + if got["endpointPath"] != "/cb" { + t.Errorf("Config() endpointPath = %v, want /cb", got["endpointPath"]) + } + if got["channelId"] != "ch-1" { + t.Errorf("Config() channelId = %v, want ch-1", got["channelId"]) + } + if got["secret"] != "s3cr3t" { + t.Errorf("Config() secret = %v, want s3cr3t", got["secret"]) + } + if got["signatureHeader"] != defaultSignatureHeader { + t.Errorf("Config() signatureHeader = %v, want %s", got["signatureHeader"], defaultSignatureHeader) + } +} + +func TestConnectDisconnectHealthCheck(t *testing.T) { + t.Parallel() + + a := New() + if a.IsConnected() { + t.Error("IsConnected() = true before Connect, want false") + } + + if err := a.HealthCheck(context.Background()); !errors.Is(err, adapter.ErrNotConnected) { + t.Errorf("HealthCheck() before connect error = %v, want ErrNotConnected", err) + } + + if err := a.Connect(context.Background()); err != nil { + t.Errorf("Connect() error = %v", err) + } + if !a.IsConnected() { + t.Error("IsConnected() = false after Connect, want true") + } + if err := a.HealthCheck(context.Background()); err != nil { + t.Errorf("HealthCheck() after connect error = %v", err) + } + + if err := a.Disconnect(context.Background()); err != nil { + t.Errorf("Disconnect() error = %v", err) + } + if a.IsConnected() { + t.Error("IsConnected() = true after Disconnect, want false") + } + // Idempotent. + if err := a.Disconnect(context.Background()); err != nil { + t.Errorf("Disconnect() second call error = %v", err) + } + if err := a.Connect(context.Background()); err != nil { + t.Errorf("Connect() second call error = %v", err) + } +} + +func TestCapabilitiesProactiveDisabled(t *testing.T) { + t.Parallel() + + caps := New().Capabilities() + if caps.Proactive { + t.Error("Capabilities().Proactive = true, want false") + } + if caps.MaxMessageLength <= 0 { + t.Errorf("Capabilities().MaxMessageLength = %d, want normalized nonzero", caps.MaxMessageLength) + } + if caps.RecommendBurstMax <= 0 { + t.Errorf("Capabilities().RecommendBurstMax = %d, want normalized nonzero", caps.RecommendBurstMax) + } +} + +func TestListChannelsConfigured(t *testing.T) { + t.Parallel() + + a := New(WithConfig(Config{ChannelID: "ch-1", ChannelName: "Hook"})) + chans, err := a.ListChannels(context.Background()) + if err != nil { + t.Fatalf("ListChannels() error = %v", err) + } + if len(chans) != 1 { + t.Fatalf("ListChannels() len = %d, want 1", len(chans)) + } + if chans[0].ChannelID != "ch-1" || chans[0].ChannelName != "Hook" { + t.Errorf("ListChannels()[0] = %+v, want {ch-1 Hook}", chans[0]) + } +} + +func TestListChannelsEmpty(t *testing.T) { + t.Parallel() + + chans, err := New().ListChannels(context.Background()) + if err != nil { + t.Fatalf("ListChannels() error = %v", err) + } + if len(chans) != 0 { + t.Errorf("ListChannels() len = %d, want 0", len(chans)) + } +} + +func TestSendProactiveUnsupported(t *testing.T) { + t.Parallel() + + err := New().SendProactive(context.Background(), adapter.UnifiedMessage{}) + if !errors.Is(err, adapter.ErrUnsupported) { + t.Errorf("SendProactive() error = %v, want ErrUnsupported", err) + } +} + +func TestSendMessageUnsupported(t *testing.T) { + t.Parallel() + + err := New().SendMessage(context.Background(), adapter.UnifiedResponse{}) + if !errors.Is(err, adapter.ErrUnsupported) { + t.Errorf("SendMessage() error = %v, want ErrUnsupported", err) + } +} + +func TestSetMessageHandlerAndClear(t *testing.T) { + t.Parallel() + + a := New() + a.SetMessageHandler(func(context.Context, adapter.UnifiedMessage) {}) + + // A nil handler clears the installed one, so delivery must fail. + a.SetMessageHandler(nil) + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/webhook", + bytes.NewReader([]byte(`{"content":"hi"}`))) + req.Header.Set("Content-Type", "application/json") + a.InboundHandler().ServeHTTP(rec, req) + + if rec.Code != http.StatusServiceUnavailable { + t.Errorf("status = %d, want 503", rec.Code) + } +} + +func TestReadMessagesBlocksUntilCancel(t *testing.T) { + a := New() + ctx, cancel := context.WithCancel(context.Background()) + cancel() // pre-cancel so the loop returns deterministically + + done := make(chan struct{}) + go func() { + a.ReadMessages(ctx, func(adapter.UnifiedMessage) {}) + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("ReadMessages() did not return on canceled context") + } +}