feat: outbound webhooks — signed event delivery + delivery log

Every audited action now emits a "resource.action" event (bare actions
get resource-qualified; already-qualified actions like service.clone
pass through). Webhooks subscribe per-user with exact names, resource.*
wildcards, or *.

- schema: outbound_webhooks + webhook_deliveries (sqlc + goose)
- dispatcher: 4 workers, 512-queue, 3 attempts with backoff, 10s timeout;
  deliveries record status/http-code/body(4KiB)/duration
- signing: X-Containr-Signature sha256=<hmac> over raw body;
  X-Containr-Event + X-Containr-Delivery headers; custom headers per hook
- api: CRUD + /deliveries + /test (ping) under authed group
- fix: SetAuditDB never called — LogAudit was a silent no-op at every
  call site; now wired at boot
- coverage: service create/delete audited (previously uninstrumented)
- cli: containr webhooks list|create|update|delete|deliveries|test
- mcp: 6 tools, delete requires confirm
- frontend: /settings/webhooks page — list/create/expand-deliveries/
  test/disable/delete, secret-shown-once banner; nav + account menu
- openapi + regenerated types; CLI.md + docs snapshot; roadmap ticks
This commit is contained in:
Tomas Dvorak committed 2026-10-09 09:44:08 +02:00
1 parent 5c2f2861bd
commit f499678964
23 files changed
+2278 -7

No files matched your search

+4
View File
@@ -52,6 +52,8 @@ func LogAudit(userID, resource, resourceID, action string, details map[string]in
if err != nil {
}
emitWebhookEvent(userID, resource, resourceID, action, details)
}
func LogAuditWithRequest(c *gin.Context, resource, resourceID, action string, details map[string]interface{}) {
@@ -79,6 +81,8 @@ func LogAuditWithRequest(c *gin.Context, resource, resourceID, action string, de
if err != nil {
}
emitWebhookEvent(userIDStr, resource, resourceID, action, details)
}
var auditDB *database.DB
+13
View File
@@ -23,6 +23,11 @@ import (
)
func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg *config.Config) {
// Audit + outbound webhook delivery need a package-level handle for
// call sites that don't carry a gin context.
SetAuditDB(db)
StartWebhookDispatcher()
// Expose Better Auth through backend so frontend can use a single backend origin.
setupAuthProxyRoutes(router, cfg, db)
@@ -328,6 +333,14 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg
// WebSocket endpoint
authed.GET("/ws", handleWebSocket)
// Outbound webhooks — signed event delivery to user HTTP endpoints
authed.GET("/webhooks", handleListOutboundWebhooks)
authed.POST("/webhooks", handleCreateOutboundWebhook)
authed.PATCH("/webhooks/:id", handleUpdateOutboundWebhook)
authed.DELETE("/webhooks/:id", handleDeleteOutboundWebhook)
authed.GET("/webhooks/:id/deliveries", handleListOutboundWebhookDeliveries)
authed.POST("/webhooks/:id/test", handleTestOutboundWebhook)
// Owned-resource mutations — handlers scope to the caller and
// admit admins. New projects land unapproved until an admin
// publishes them.
+6
View File
@@ -454,6 +454,10 @@ func handleCreateService(c *gin.Context) {
}
service.Domains = loadServiceDomains(db.(*database.DB), service.ID)
LogAuditWithRequest(c, "service", service.ID.String(), "create", map[string]interface{}{
"name": service.Name, "type": service.Type, "project_id": service.ProjectID,
})
c.JSON(http.StatusCreated, gin.H{"service": service})
}
@@ -876,5 +880,7 @@ func handleDeleteService(c *gin.Context) {
return
}
LogAuditWithRequest(c, "service", serviceID.String(), "delete", map[string]interface{}{})
c.JSON(http.StatusOK, gin.H{"message": "Service deleted successfully"})
}
+543
View File
@@ -0,0 +1,543 @@
package api
import (
"bytes"
"context"
"crypto/hmac"
"crypto/rand"
"crypto/sha256"
"database/sql"
"encoding/hex"
"encoding/json"
"errors"
"io"
"net/http"
"net/url"
"strings"
"sync"
"time"
"containr/internal/database"
"containr/internal/database/sqlcdb"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/sqlc-dev/pqtype"
)
// Outbound webhooks: every audited action (LogAudit) emits an event named
// "resource.action" (e.g. "service.deploy"). Webhooks subscribe with an
// event list supporting "service.*" and "*" wildcards. Deliveries are
// HMAC-signed and retried up to 3 times with backoff; each attempt lands
// in webhook_deliveries.
var (
webhookQueue = make(chan webhookJob, 512)
webhookWorkers sync.Once
)
type webhookJob struct {
Webhook sqlcdb.OutboundWebhook
Event string
Payload []byte
}
type webhookEventPayload struct {
Event string `json:"event"`
Resource string `json:"resource"`
ResourceID string `json:"resource_id"`
Action string `json:"action"`
UserID string `json:"user_id,omitempty"`
Details map[string]interface{} `json:"details,omitempty"`
Timestamp time.Time `json:"timestamp"`
}
// emitWebhookEvent fans an audited action out to all enabled webhooks whose
// event list matches. Called from LogAudit — fire-and-forget.
func emitWebhookEvent(userID, resource, resourceID, action string, details map[string]interface{}) {
db := GetAuditDB()
if db == nil {
return
}
// Actions that already carry a resource prefix ("service.clone") pass
// through unchanged; bare actions ("create") get resource-qualified.
event := action
if !strings.Contains(action, ".") {
event = resource + "." + action
}
queries := sqlcdb.New(db.DB)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
hooks, err := queries.ListEnabledOutboundWebhooks(ctx)
if err != nil || len(hooks) == 0 {
return
}
payload, err := json.Marshal(webhookEventPayload{
Event: event, Resource: resource, ResourceID: resourceID,
Action: action, UserID: userID, Details: details, Timestamp: time.Now().UTC(),
})
if err != nil {
return
}
uid, uidErr := uuid.Parse(strings.TrimSpace(userID))
for _, hook := range hooks {
if !webhookEventMatches(hook.Events, event) {
continue
}
// Webhooks only receive events for their owner — unless subscribed to "*"
// is deliberate admin reach; owner scoping keeps per-user isolation.
if uidErr != nil || hook.UserID != uid {
continue
}
select {
case webhookQueue <- webhookJob{Webhook: hook, Event: event, Payload: payload}:
default:
// Queue full: drop rather than block the request path.
}
}
}
func webhookEventMatches(events json.RawMessage, event string) bool {
var subs []string
if err := json.Unmarshal(events, &subs); err != nil || len(subs) == 0 {
return false
}
for _, sub := range subs {
sub = strings.TrimSpace(sub)
if sub == "*" || sub == event {
return true
}
if strings.HasSuffix(sub, ".*") && strings.HasPrefix(event, strings.TrimSuffix(sub, "*")) {
return true
}
}
return false
}
// StartWebhookDispatcher launches delivery workers; call once at boot.
func StartWebhookDispatcher() {
webhookWorkers.Do(func() {
for i := 0; i < 4; i++ {
go webhookDeliveryWorker()
}
})
}
func webhookDeliveryWorker() {
for job := range webhookQueue {
deliverWebhook(job)
}
}
var webhookHTTPClient = &http.Client{Timeout: 10 * time.Second}
func deliverWebhook(job webhookJob) {
db := GetAuditDB()
if db == nil {
return
}
queries := sqlcdb.New(db.DB)
ctx := context.Background()
deliveryID := uuid.New()
_ = queries.CreateWebhookDelivery(ctx, sqlcdb.CreateWebhookDeliveryParams{
ID: deliveryID,
WebhookID: job.Webhook.ID,
Event: job.Event,
Payload: pqtype.NullRawMessage{RawMessage: job.Payload, Valid: true},
Status: "pending",
})
var lastStatus string = "failed"
var respStatus sql.NullInt32
var respBody sql.NullString
var durationMs sql.NullInt32
var deliveredAt sql.NullTime
for attempt := int32(1); attempt <= 3; attempt++ {
start := time.Now()
status, code, body := postWebhook(job, deliveryID, attempt)
durationMs = sql.NullInt32{Int32: int32(time.Since(start).Milliseconds()), Valid: true}
respStatus = code
respBody = body
if status == "success" {
lastStatus = "success"
deliveredAt = sql.NullTime{Time: time.Now().UTC(), Valid: true}
_ = queries.UpdateWebhookDeliveryResult(ctx, sqlcdb.UpdateWebhookDeliveryResultParams{
ID: deliveryID,
Status: lastStatus,
ResponseStatus: respStatus,
ResponseBody: respBody,
Attempts: attempt,
DurationMs: durationMs,
DeliveredAt: deliveredAt,
})
return
}
if attempt < 3 {
time.Sleep(time.Duration(attempt*attempt) * 2 * time.Second)
}
}
_ = queries.UpdateWebhookDeliveryResult(ctx, sqlcdb.UpdateWebhookDeliveryResultParams{
ID: deliveryID,
Status: lastStatus,
ResponseStatus: respStatus,
ResponseBody: respBody,
Attempts: 3,
DurationMs: durationMs,
DeliveredAt: deliveredAt,
})
}
func postWebhook(job webhookJob, deliveryID uuid.UUID, attempt int32) (string, sql.NullInt32, sql.NullString) {
req, err := http.NewRequest(http.MethodPost, job.Webhook.Url, bytes.NewReader(job.Payload))
if err != nil {
return "failed", sql.NullInt32{}, sql.NullString{String: err.Error(), Valid: true}
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("User-Agent", "Containr-Webhook/1.0")
req.Header.Set("X-Containr-Event", job.Event)
req.Header.Set("X-Containr-Delivery", deliveryID.String())
if job.Webhook.Secret != "" {
mac := hmac.New(sha256.New, []byte(job.Webhook.Secret))
mac.Write(job.Payload)
req.Header.Set("X-Containr-Signature", "sha256="+hex.EncodeToString(mac.Sum(nil)))
}
if job.Webhook.Headers.Valid {
var custom map[string]string
if err := json.Unmarshal(job.Webhook.Headers.RawMessage, &custom); err == nil {
for k, v := range custom {
req.Header.Set(k, v)
}
}
}
resp, err := webhookHTTPClient.Do(req)
if err != nil {
return "failed", sql.NullInt32{}, sql.NullString{String: err.Error(), Valid: true}
}
defer resp.Body.Close()
body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
code := sql.NullInt32{Int32: int32(resp.StatusCode), Valid: true}
bodyStr := sql.NullString{String: string(body), Valid: len(body) > 0}
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
return "success", code, bodyStr
}
return "failed", code, bodyStr
}
// ---------- HTTP API ----------
type outboundWebhookResponse struct {
ID string `json:"id"`
Name string `json:"name"`
URL string `json:"url"`
Secret string `json:"secret,omitempty"`
Events []string `json:"events"`
Headers map[string]string `json:"headers,omitempty"`
Enabled bool `json:"enabled"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
func outboundWebhookJSON(row sqlcdb.OutboundWebhook, revealSecret bool) outboundWebhookResponse {
var events []string
_ = json.Unmarshal(row.Events, &events)
if events == nil {
events = []string{}
}
var headers map[string]string
if row.Headers.Valid {
_ = json.Unmarshal(row.Headers.RawMessage, &headers)
}
resp := outboundWebhookResponse{
ID: row.ID.String(),
Name: row.Name,
URL: row.Url,
Events: events,
Headers: headers,
Enabled: row.Enabled,
}
if revealSecret {
resp.Secret = row.Secret
}
if row.CreatedAt.Valid {
resp.CreatedAt = row.CreatedAt.Time
}
if row.UpdatedAt.Valid {
resp.UpdatedAt = row.UpdatedAt.Time
}
return resp
}
func validWebhookURL(raw string) error {
u, err := url.Parse(strings.TrimSpace(raw))
if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
return errors.New("url must be a valid http(s) URL")
}
return nil
}
type webhookMutationRequest struct {
Name string `json:"name"`
URL string `json:"url"`
Secret *string `json:"secret"`
Events []string `json:"events"`
Headers map[string]string `json:"headers"`
Enabled *bool `json:"enabled"`
}
func handleListOutboundWebhooks(c *gin.Context) {
userID, ok := requireAuthenticatedUserUUID(c)
if !ok {
return
}
db := c.MustGet("db").(*database.DB)
rows, err := sqlcdb.New(db.DB).ListOutboundWebhooksByUser(c.Request.Context(), userID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to list webhooks"})
return
}
out := make([]outboundWebhookResponse, 0, len(rows))
for _, row := range rows {
out = append(out, outboundWebhookJSON(row, false))
}
c.JSON(http.StatusOK, gin.H{"webhooks": out})
}
func handleCreateOutboundWebhook(c *gin.Context) {
userID, ok := requireAuthenticatedUserUUID(c)
if !ok {
return
}
var req webhookMutationRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
req.Name = strings.TrimSpace(req.Name)
if req.Name == "" {
respondError(c, http.StatusBadRequest, "VALIDATION", "name is required")
return
}
if err := validWebhookURL(req.URL); err != nil {
respondError(c, http.StatusBadRequest, "VALIDATION", err.Error())
return
}
if len(req.Events) == 0 {
respondError(c, http.StatusBadRequest, "VALIDATION", "at least one event subscription is required (e.g. \"service.*\" or \"*\")")
return
}
secret := ""
if req.Secret != nil {
secret = strings.TrimSpace(*req.Secret)
}
if secret == "" {
buf := make([]byte, 24)
_, _ = rand.Read(buf)
secret = hex.EncodeToString(buf)
}
eventsJSON, _ := json.Marshal(req.Events)
var headersJSON pqtype.NullRawMessage
if len(req.Headers) > 0 {
if raw, err := json.Marshal(req.Headers); err == nil {
headersJSON = pqtype.NullRawMessage{RawMessage: raw, Valid: true}
}
}
enabled := true
if req.Enabled != nil {
enabled = *req.Enabled
}
db := c.MustGet("db").(*database.DB)
id := uuid.New()
err := sqlcdb.New(db.DB).CreateOutboundWebhook(c.Request.Context(), sqlcdb.CreateOutboundWebhookParams{
ID: id,
UserID: userID,
Name: req.Name,
Url: strings.TrimSpace(req.URL),
Secret: secret,
Events: eventsJSON,
Headers: headersJSON,
Enabled: enabled,
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create webhook"})
return
}
LogAuditWithRequest(c, "webhook", id.String(), "create", map[string]interface{}{"name": req.Name})
c.JSON(http.StatusCreated, outboundWebhookJSON(sqlcdb.OutboundWebhook{
ID: id, Name: req.Name, Url: strings.TrimSpace(req.URL), Secret: secret,
Events: eventsJSON, Headers: headersJSON, Enabled: enabled,
CreatedAt: sql.NullTime{Time: time.Now(), Valid: true},
UpdatedAt: sql.NullTime{Time: time.Now(), Valid: true},
}, true))
}
func handleUpdateOutboundWebhook(c *gin.Context) {
userID, ok := requireAuthenticatedUserUUID(c)
if !ok {
return
}
webhookID, err := uuid.Parse(c.Param("id"))
if err != nil {
respondError(c, http.StatusBadRequest, "VALIDATION", "Invalid webhook id")
return
}
db := c.MustGet("db").(*database.DB)
queries := sqlcdb.New(db.DB)
row, err := queries.GetOutboundWebhookByIDAndUser(c.Request.Context(), sqlcdb.GetOutboundWebhookByIDAndUserParams{
ID: webhookID, UserID: userID,
})
if errors.Is(err, sql.ErrNoRows) {
respondError(c, http.StatusNotFound, "NOT_FOUND", "Webhook not found")
return
}
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to load webhook"})
return
}
var req webhookMutationRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
name := row.Name
if req.Name != "" {
name = strings.TrimSpace(req.Name)
}
targetURL := row.Url
if req.URL != "" {
if err := validWebhookURL(req.URL); err != nil {
respondError(c, http.StatusBadRequest, "VALIDATION", err.Error())
return
}
targetURL = strings.TrimSpace(req.URL)
}
secret := row.Secret
if req.Secret != nil {
secret = strings.TrimSpace(*req.Secret)
}
eventsJSON := row.Events
if req.Events != nil {
if len(req.Events) == 0 {
respondError(c, http.StatusBadRequest, "VALIDATION", "at least one event subscription is required")
return
}
eventsJSON, _ = json.Marshal(req.Events)
}
headersJSON := row.Headers
if req.Headers != nil {
if raw, err := json.Marshal(req.Headers); err == nil {
headersJSON = pqtype.NullRawMessage{RawMessage: raw, Valid: true}
}
}
enabled := row.Enabled
if req.Enabled != nil {
enabled = *req.Enabled
}
err = queries.UpdateOutboundWebhook(c.Request.Context(), sqlcdb.UpdateOutboundWebhookParams{
ID: webhookID, UserID: userID,
Name: name, Url: targetURL, Secret: secret,
Events: eventsJSON, Headers: headersJSON, Enabled: enabled,
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update webhook"})
return
}
LogAuditWithRequest(c, "webhook", webhookID.String(), "update", map[string]interface{}{"name": name})
row.Name, row.Url, row.Secret, row.Events, row.Headers, row.Enabled = name, targetURL, secret, eventsJSON, headersJSON, enabled
c.JSON(http.StatusOK, outboundWebhookJSON(row, true))
}
func handleDeleteOutboundWebhook(c *gin.Context) {
userID, ok := requireAuthenticatedUserUUID(c)
if !ok {
return
}
webhookID, err := uuid.Parse(c.Param("id"))
if err != nil {
respondError(c, http.StatusBadRequest, "VALIDATION", "Invalid webhook id")
return
}
db := c.MustGet("db").(*database.DB)
err = sqlcdb.New(db.DB).DeleteOutboundWebhookByIDAndUser(c.Request.Context(), sqlcdb.DeleteOutboundWebhookByIDAndUserParams{
ID: webhookID, UserID: userID,
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to delete webhook"})
return
}
LogAuditWithRequest(c, "webhook", webhookID.String(), "delete", map[string]interface{}{})
c.JSON(http.StatusOK, gin.H{"message": "Webhook deleted"})
}
func handleListOutboundWebhookDeliveries(c *gin.Context) {
userID, ok := requireAuthenticatedUserUUID(c)
if !ok {
return
}
webhookID, err := uuid.Parse(c.Param("id"))
if err != nil {
respondError(c, http.StatusBadRequest, "VALIDATION", "Invalid webhook id")
return
}
db := c.MustGet("db").(*database.DB)
queries := sqlcdb.New(db.DB)
if _, err := queries.GetOutboundWebhookByIDAndUser(c.Request.Context(), sqlcdb.GetOutboundWebhookByIDAndUserParams{
ID: webhookID, UserID: userID,
}); errors.Is(err, sql.ErrNoRows) {
respondError(c, http.StatusNotFound, "NOT_FOUND", "Webhook not found")
return
}
rows, err := queries.ListWebhookDeliveriesByWebhook(c.Request.Context(), webhookID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to list deliveries"})
return
}
c.JSON(http.StatusOK, gin.H{"deliveries": rows})
}
func handleTestOutboundWebhook(c *gin.Context) {
userID, ok := requireAuthenticatedUserUUID(c)
if !ok {
return
}
webhookID, err := uuid.Parse(c.Param("id"))
if err != nil {
respondError(c, http.StatusBadRequest, "VALIDATION", "Invalid webhook id")
return
}
db := c.MustGet("db").(*database.DB)
row, err := sqlcdb.New(db.DB).GetOutboundWebhookByIDAndUser(c.Request.Context(), sqlcdb.GetOutboundWebhookByIDAndUserParams{
ID: webhookID, UserID: userID,
})
if errors.Is(err, sql.ErrNoRows) {
respondError(c, http.StatusNotFound, "NOT_FOUND", "Webhook not found")
return
}
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to load webhook"})
return
}
if !row.Enabled {
respondError(c, http.StatusBadRequest, "VALIDATION", "Webhook is disabled")
return
}
payload, _ := json.Marshal(webhookEventPayload{
Event: "ping", Resource: "webhook", ResourceID: webhookID.String(),
Action: "ping", UserID: userID.String(), Timestamp: time.Now().UTC(),
})
select {
case webhookQueue <- webhookJob{Webhook: row, Event: "ping", Payload: payload}:
c.JSON(http.StatusAccepted, gin.H{"message": "Test delivery queued"})
default:
respondError(c, http.StatusServiceUnavailable, "UNAVAILABLE", "Delivery queue is full")
}
}
+33
View File
@@ -0,0 +1,33 @@
package api
import (
"encoding/json"
"testing"
)
func TestWebhookEventMatches(t *testing.T) {
subs := func(list ...string) json.RawMessage {
raw, _ := json.Marshal(list)
return raw
}
cases := []struct {
events json.RawMessage
event string
want bool
}{
{subs("service.*"), "service.deploy", true},
{subs("service.*"), "database.create", false},
{subs("*"), "service.deploy", true},
{subs("deployment.fail", "service.*"), "deployment.fail", true},
{subs("deployment.fail"), "deployment.success", false},
{subs(" ping "), "ping", true},
{subs(), "service.deploy", false},
{json.RawMessage(`"not-a-list"`), "service.deploy", false},
{json.RawMessage(`invalid`), "service.deploy", false},
}
for i, tc := range cases {
if got := webhookEventMatches(tc.events, tc.event); got != tc.want {
t.Errorf("case %d: matches(%s, %q) = %v, want %v", i, tc.events, tc.event, got, tc.want)
}
}
}