package http import ( "context" "encoding/json" "errors" "log/slog" "net/http" "time" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) // webhookContentMaxBytes is the maximum allowed content field length (16 KB). const webhookContentMaxBytes = 16 * 1024 // channelDispatcher is the subset of *channels.Manager used by WebhookMessageHandler. // Declared as an interface so tests can substitute a stub without spinning up a full Manager. type channelDispatcher interface { ChannelTenantID(channelName string) (uuid.UUID, bool) ChannelTypeForName(channelName string) string SendToChannel(ctx context.Context, channelName, chatID, content string) error SendMediaToChannel(ctx context.Context, channelName, chatID, content string, media []bus.MediaAttachment) error } // WebhookMessageHandler handles POST /v1/webhooks/message. // Standard edition only — mount via edition.Current().AllowsChannels() gate. // Auth is enforced by WebhookAuthMiddleware (phase 03) with kind="message". type WebhookMessageHandler struct { channelMgr channelDispatcher channelInstances store.ChannelInstanceStore callStore store.WebhookCallStore webhooks store.WebhookStore limiter *webhookLimiter encKey string // AES-256-GCM key for decrypting encrypted_secret at HMAC verify time } // NewWebhookMessageHandler constructs a WebhookMessageHandler. // mgr must be *channels.Manager (satisfies channelDispatcher). func NewWebhookMessageHandler( mgr *channels.Manager, channelInstances store.ChannelInstanceStore, callStore store.WebhookCallStore, webhooks store.WebhookStore, limiter *webhookLimiter, ) *WebhookMessageHandler { return &WebhookMessageHandler{ channelMgr: mgr, channelInstances: channelInstances, callStore: callStore, webhooks: webhooks, limiter: limiter, } } // SetEncKey sets the AES-256-GCM encryption key for decrypting webhook secrets at HMAC verify time. func (h *WebhookMessageHandler) SetEncKey(encKey string) { h.encKey = encKey } // RegisterRoutes mounts POST /v1/webhooks/message wrapped in the auth middleware. // Only call when edition.Current().AllowsChannels() — callers enforce the gate. func (h *WebhookMessageHandler) RegisterRoutes(mux *http.ServeMux) { authMW := WebhookAuthMiddleware( h.webhooks, h.callStore, h.limiter, h.encKey, "message", WebhookMaxBodyMessage, ) mux.Handle("POST /v1/webhooks/message", authMW(http.HandlerFunc(h.handle))) } // webhookMessageReq is the JSON request body for POST /v1/webhooks/message. type webhookMessageReq struct { // ChannelName is the channel instance name to deliver through. // Required when the webhook row has no bound channel_id. ChannelName string `json:"channel_name"` // ChatID is the channel-specific recipient identifier (required). ChatID string `json:"chat_id"` // Content is the text body (required unless media_url is set; max 16 KB). Content string `json:"content"` // MediaURL is an optional HTTPS URL to a media file. MediaURL string `json:"media_url,omitempty"` // MediaCaption is an optional caption attached to the media. MediaCaption string `json:"media_caption,omitempty"` // FallbackToText controls media-unsupported channel behavior: // true → drop media, send text only, 200 + warning // false → 501 (default) FallbackToText bool `json:"fallback_to_text,omitempty"` } // webhookMessageResp is the success response envelope. type webhookMessageResp struct { CallID string `json:"call_id"` Status string `json:"status"` // always "sent" ChannelName string `json:"channel_name"` ChatID string `json:"chat_id"` Warning string `json:"warning,omitempty"` // set when media was dropped on fallback } // handle is the HTTP handler for POST /v1/webhooks/message. func (h *WebhookMessageHandler) handle(w http.ResponseWriter, r *http.Request) { ctx := r.Context() locale := store.LocaleFromContext(ctx) // Webhook row injected by WebhookAuthMiddleware — always present here. webhook := WebhookDataFromContext(ctx) if webhook == nil { writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgInternalError, "webhook context missing")) return } // Decode and validate request body. var req webhookMessageReq if !bindJSON(w, r, locale, &req) { return } // Resolve channel name: webhook-bound channel_id takes precedence. channelName, ok := h.resolveChannelName(ctx, w, webhook, req.ChannelName, locale) if !ok { return } // P0: Cross-tenant isolation — channel must belong to webhook's tenant. if !h.validateChannelTenant(ctx, w, webhook, channelName, locale) { return } // Field validation (after channel resolution so tenant check runs first). if req.ChatID == "" { writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "chat_id")) return } if req.Content == "" && req.MediaURL == "" { writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "content")) return } if len(req.Content) > webhookContentMaxBytes { writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "content exceeds 16 KB limit")) return } // Build the audit call record (written on success or failure below). callID := store.GenNewID() deliveryID := store.GenNewID() now := time.Now() callRecord := h.newCallRecord(r, webhook, callID, deliveryID, now, channelName, req) callReserved, handled := reserveIdempotentCall(w, r, h.callStore, callRecord) if handled { return } // Dispatch — media or text-only path. warning, sendErr := h.dispatch(ctx, w, r, webhook, req, channelName, callRecord, callReserved, locale) if sendErr != nil { return // error response already written by dispatch } // Record successful delivery. completedAt := time.Now() callRecord.Status = "done" callRecord.CompletedAt = &completedAt callRecord.Attempts = 1 respBody := webhookMessageResp{ CallID: callID.String(), Status: "sent", ChannelName: channelName, ChatID: req.ChatID, Warning: warning, } respBytes, _ := json.Marshal(respBody) callRecord.Response = respBytes persistWebhookCall(ctx, h.callStore, callRecord, callReserved, "webhook.message.audit_write_failed") slog.Info("webhook.message.delivered", "tenant_id", webhook.TenantID, "webhook_id", webhook.ID, "channel", channelName, "chat_id", req.ChatID, "has_media", req.MediaURL != "", ) writeJSON(w, http.StatusOK, respBody) } // dispatch sends the message (media or text) to the channel. // Returns (warning string, error). On non-nil error the response was already written. func (h *WebhookMessageHandler) dispatch( ctx context.Context, w http.ResponseWriter, r *http.Request, webhook *store.WebhookData, req webhookMessageReq, channelName string, callRecord *store.WebhookCallData, callReserved bool, locale string, ) (warning string, _ error) { if req.MediaURL == "" { // Text-only path. if err := h.channelMgr.SendToChannel(ctx, channelName, req.ChatID, req.Content); err != nil { h.failCall(ctx, callRecord, callReserved, err.Error()) slog.Error("webhook.message.dispatch_failed", "error", err, "channel_name", channelName, "webhook_id", webhook.ID, ) writeError(w, http.StatusBadGateway, protocol.ErrInternal, i18n.T(locale, i18n.MsgInternalError, "channel send failed")) return "", err } return "", nil } // Media path: SSRF validation + HEAD probe. probe, probeErr := probeMediaURL(req.MediaURL) if probeErr != nil { var mve *mediaValidateError if errors.As(probeErr, &mve) { h.failCall(ctx, callRecord, callReserved, mve.message) switch mve.code { case "ssrf": slog.Warn("security.webhook.ssrf_blocked", "host", redactedHost(req.MediaURL), "webhook_id", webhook.ID, ) writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgWebhookMediaSSRFBlocked)) case "too_large": writeError(w, http.StatusRequestEntityTooLarge, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgWebhookMediaTooLarge)) case "mime_denied": writeError(w, http.StatusUnsupportedMediaType, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgWebhookMediaMIMEDenied)) default: writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgWebhookMediaSSRFBlocked)) } } else { h.failCall(ctx, callRecord, callReserved, probeErr.Error()) writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgWebhookMediaSSRFBlocked)) } return "", probeErr } // Channel media capability gate. channelType := h.channelMgr.ChannelTypeForName(channelName) if channels.IsMediaCapable(channelType) { media := []bus.MediaAttachment{{ URL: req.MediaURL, ContentType: probe.ContentType, Caption: req.MediaCaption, }} if err := h.channelMgr.SendMediaToChannel(ctx, channelName, req.ChatID, req.Content, media); err != nil { h.failCall(ctx, callRecord, callReserved, err.Error()) slog.Error("webhook.message.dispatch_failed", "error", err, "channel_name", channelName, "webhook_id", webhook.ID, ) writeError(w, http.StatusBadGateway, protocol.ErrInternal, i18n.T(locale, i18n.MsgInternalError, "channel send failed")) return "", err } return "", nil } if req.FallbackToText { // Degrade to text-only send. slog.Warn("webhook.media_unsupported_fallback", "channel_name", channelName, "channel_type", channelType, "webhook_id", webhook.ID, ) if err := h.channelMgr.SendToChannel(ctx, channelName, req.ChatID, req.Content); err != nil { h.failCall(ctx, callRecord, callReserved, err.Error()) slog.Error("webhook.message.dispatch_failed", "error", err, "channel_name", channelName, "webhook_id", webhook.ID, ) writeError(w, http.StatusBadGateway, protocol.ErrInternal, i18n.T(locale, i18n.MsgInternalError, "channel send failed")) return "", err } return "media_not_supported_fallback_text", nil } // Media unsupported + no fallback → 501. const reason = "channel does not support media and fallback_to_text is false" h.failCall(ctx, callRecord, callReserved, reason) writeError(w, http.StatusNotImplemented, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgWebhookMediaChannelUnsupported)) return "", errors.New(reason) } // resolveChannelName returns the channel instance name for dispatch. // Preference: webhook-bound channel_id (resolved via ChannelInstanceStore) → req.ChannelName. func (h *WebhookMessageHandler) resolveChannelName( ctx context.Context, w http.ResponseWriter, webhook *store.WebhookData, reqChannelName string, locale string, ) (string, bool) { if webhook.ChannelID != nil { inst, err := h.channelInstances.Get(ctx, *webhook.ChannelID) if err != nil || inst == nil { writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgWebhookChannelNotFound)) return "", false } return inst.Name, true } if reqChannelName == "" { writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "channel_name")) return "", false } return reqChannelName, true } // validateChannelTenant enforces the P0 cross-tenant isolation rule: // the channel must belong to the same tenant as the webhook. // Returns true if the check passes (caller may proceed). func (h *WebhookMessageHandler) validateChannelTenant( ctx context.Context, w http.ResponseWriter, webhook *store.WebhookData, channelName string, locale string, ) bool { channelTenantID, exists := h.channelMgr.ChannelTenantID(channelName) if !exists { writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgWebhookChannelNotFound)) return false } // uuid.Nil means legacy/config-based channel — allow from any tenant (backward compat). if channelTenantID != uuid.Nil && channelTenantID != webhook.TenantID { slog.Warn("security.webhook.tenant_leak_attempt", "webhook_id", webhook.ID, "webhook_tenant", webhook.TenantID, "channel_name", channelName, "channel_tenant", channelTenantID, ) writeError(w, http.StatusForbidden, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgWebhookTenantMismatch)) return false } return true } // newCallRecord builds the initial WebhookCallData for audit logging. func (h *WebhookMessageHandler) newCallRecord( r *http.Request, webhook *store.WebhookData, callID, deliveryID uuid.UUID, now time.Time, channelName string, req webhookMessageReq, ) *store.WebhookCallData { // Encode canonical audit payload: {"body_hash": "", "meta": {...}}. // PG jsonb rejects non-JSON bytes; this shape is valid JSON on both PG and SQLite. bodyBytes := WebhookRawBodyFromContext(r.Context()) if bodyBytes == nil { bodyBytes, _ = json.Marshal(req) } requestPayload, _ := buildAuditPayload(bodyBytes, map[string]any{ "channel_name": channelName, "chat_id": req.ChatID, "has_media": req.MediaURL != "", }) call := &store.WebhookCallData{ ID: callID, TenantID: webhook.TenantID, WebhookID: webhook.ID, AgentID: webhook.AgentID, DeliveryID: deliveryID, Mode: "sync", Status: "running", StartedAt: &now, RequestPayload: requestPayload, CreatedAt: now, } if key := r.Header.Get("Idempotency-Key"); key != "" { call.IdempotencyKey = &key } return call } // failCall mutates call to status=failed and records it in the store. Best-effort. func (h *WebhookMessageHandler) failCall(ctx context.Context, call *store.WebhookCallData, reserved bool, reason string) { now := time.Now() call.Status = "failed" call.CompletedAt = &now call.LastError = &reason call.Attempts = 1 persistWebhookCall(ctx, h.callStore, call, reserved, "webhook.message.audit_write_failed") } // redactedHost extracts the hostname from a URL string for safe (no-path) log output. func redactedHost(rawURL string) string { for _, prefix := range []string{"https://", "http://"} { if len(rawURL) > len(prefix) && rawURL[:len(prefix)] == prefix { rest := rawURL[len(prefix):] for i, c := range rest { if c == '/' || c == '?' || c == '#' { return rest[:i] } } return rest } } return "[unknown]" }