From ea6639ff31c25f068071dc7448930ce639f06ac0 Mon Sep 17 00:00:00 2001 From: clawbot Date: Thu, 20 Aug 2026 05:45:04 +0000 Subject: [PATCH] Add per-delivery replay to the event log (closes #203) A delivery that exhausted max_retries was failed forever. The event body is durably stored, so the only way to get it delivered was to download it and re-POST by hand. The event log now offers a Replay action on any finished delivery. Replay creates a NEW pending delivery for the same event and target and hands it to the delivery engine through the same Notifier the receiver uses, so it is retried, SSRF-guarded and circuit-broken exactly as a first attempt. The original delivery's status, timestamps and recorded attempts are never touched, and what is re-sent is the stored event body, not the response the original attempt received. The target is read as it stands now, including soft-deleted rows so that a deleted target refuses the replay with a message on the page instead of erroring or delivering from stale configuration. A deactivated target and a target id that names nothing refuse the same way, as does a replay of a delivery the engine has not finished. Two bounds on replay storms: the route carries a per-client POST rate limit of 30 per minute, and the handler refuses a replay while an earlier one for the same event and target is still pending or retrying. One new metric, webhooker_delivery_replays_total, on the existing target_type label. A replay is a real delivery and moves the attempt, outcome and duration series like any other; this counter is what separates it from ordinary traffic without adding a dimension to every existing series. The delivery row is written with associations omitted and with neither Event nor Target populated, so no target row reaches the per-webhook event database. --- README.md | 20 + internal/database/model_delivery.go | 17 + internal/handlers/delivery_replay.go | 378 ++++++++++++++++ internal/handlers/delivery_replay_test.go | 526 ++++++++++++++++++++++ internal/handlers/handlers_test.go | 34 +- internal/handlers/source_logs_test.go | 18 +- internal/handlers/source_management.go | 27 +- internal/metrics/metrics.go | 28 ++ internal/middleware/ratelimit.go | 25 + internal/server/routes.go | 12 + internal/server/routes_test.go | 148 ++++++ templates/source_logs.html | 21 + 12 files changed, 1240 insertions(+), 14 deletions(-) create mode 100644 internal/handlers/delivery_replay.go create mode 100644 internal/handlers/delivery_replay_test.go diff --git a/README.md b/README.md index 9a6e0d7..844e410 100644 --- a/README.md +++ b/README.md @@ -1108,6 +1108,23 @@ DeliveryResults. succeeded). - **`failed`** — All retry attempts exhausted without success. +**Replay.** A `delivered` or `failed` delivery is finished as far as +the engine is concerned, but the event is still stored, so the event +log offers a per-delivery **Replay** action for it. Replay creates a +NEW `pending` delivery for the same event and target and hands it to +the engine on the ordinary path — same retries, same SSRF guard, same +circuit breaker as a first attempt. It never touches the delivery it +repeats: that row's status, timestamps and recorded attempts stand as +the record of what happened. + +What is re-sent is the stored event body, against the target's +configuration **as it stands now** — the point of a replay is to +deliver where the destination has since been fixed. A target that has +been deleted or deactivated therefore refuses the replay with a +message on the event log rather than delivering from stale +configuration, and a replay is refused while an earlier one for the +same event and target is still pending or retrying. + #### DeliveryResult The result of a single delivery attempt. Every attempt (including @@ -1439,6 +1456,7 @@ arriving and being stored, they are just not getting anywhere. | `webhooker_deliveries_succeeded_total` | counter | Deliveries that reached `delivered` | | `webhooker_deliveries_failed_total` | counter | Deliveries that failed terminally and will not be retried | | `webhooker_delivery_retries_total` | counter | Deliveries put back into `retrying` | +| `webhooker_delivery_replays_total` | counter | Deliveries an operator replayed from the event log. A replay runs the ordinary engine path, so it also moves the attempt, outcome and duration series; this is the only one that separates it from ordinary traffic | | `webhooker_delivery_duration_seconds` | histogram | Wall time of a single dispatched delivery attempt, the same duration the attempt's `DeliveryResult` records | | `webhooker_deliveries_pending` | gauge | Deliveries currently in `pending` | | `webhooker_deliveries_retrying` | gauge | Deliveries currently in `retrying` | @@ -2036,6 +2054,7 @@ abuse limit later; they are tracked as future work. | `POST` | `/source/{id}/edit` | Edit webhook submission | | `POST` | `/source/{id}/delete` | Delete webhook | | `GET` | `/source/{id}/logs` | Webhook event logs | +| `POST` | `/source/{id}/deliveries/{deliveryID}/replay` | Replay a finished delivery: creates a new delivery for the same event against the target's current configuration (30 per minute per bucket, then `429`) | | `POST` | `/source/{id}/entrypoints` | Add entrypoint to webhook | | `POST` | `/source/{id}/entrypoints/{entrypointID}/delete` | Delete an entrypoint | | `POST` | `/source/{id}/entrypoints/{entrypointID}/toggle` | Enable or disable an entrypoint | @@ -2125,6 +2144,7 @@ webhooker/ │ ├── handlers/ │ │ ├── handlers.go # Base handler struct, JSON helpers, template rendering │ │ ├── auth.go # Login, logout handlers +│ │ ├── delivery_replay.go # Per-delivery replay: new delivery, current target config │ │ ├── entrypoint_view.go # Masked entrypoint view for templates │ │ ├── event_log_view.go # Event log projection, byte-capped in SQL │ │ ├── healthcheck.go # Health check handler diff --git a/internal/database/model_delivery.go b/internal/database/model_delivery.go index a1fdbe9..71f6b6d 100644 --- a/internal/database/model_delivery.go +++ b/internal/database/model_delivery.go @@ -11,6 +11,23 @@ const ( DeliveryStatusRetrying DeliveryStatus = "retrying" ) +// Terminal reports whether a delivery in this status has finished, so +// the delivery engine will make no further attempt of its own. +// +// It is what decides which deliveries the event log offers to replay: +// a pending or retrying delivery is still the engine's, and replaying +// one would race it. +func (s DeliveryStatus) Terminal() bool { + switch s { + case DeliveryStatusDelivered, DeliveryStatusFailed: + return true + case DeliveryStatusPending, DeliveryStatusRetrying: + return false + default: + return false + } +} + // Delivery represents a delivery attempt for an event to a target type Delivery struct { BaseModel diff --git a/internal/handlers/delivery_replay.go b/internal/handlers/delivery_replay.go new file mode 100644 index 0000000..51b7e3b --- /dev/null +++ b/internal/handlers/delivery_replay.go @@ -0,0 +1,378 @@ +package handlers + +import ( + "net/http" + "strconv" + + "github.com/go-chi/chi" + "gorm.io/gorm" + "gorm.io/gorm/clause" + "sneak.berlin/go/webhooker/internal/database" + "sneak.berlin/go/webhooker/internal/delivery" +) + +// replayOutcomeParam is the query parameter the replay POST redirects +// with and the event log page reads its banner from. +const replayOutcomeParam = "replay" + +// replayOutcomeCode is the outcome of a replay POST. The redirect +// carries one of these fixed codes rather than a message, so nothing a +// client submits can reach the rendered page through it. +type replayOutcomeCode string + +const ( + // replayQueued reports that a new delivery was created and handed + // to the delivery engine. + replayQueued replayOutcomeCode = "queued" + + // replayTargetDeleted reports a target that once existed and has + // since been deleted. Deletes are soft and deliveries carry no + // foreign key to the target row, so the history survives its + // target and this is the ordinary case for an old event. + replayTargetDeleted replayOutcomeCode = "target-deleted" + + // replayTargetMissing reports a target id that names no row at + // all, deleted or otherwise. + replayTargetMissing replayOutcomeCode = "target-missing" + + // replayTargetInactive reports a target the operator has + // deactivated. A deactivated target receives no new deliveries, so + // a replay to it would be a delivery they switched off. + replayTargetInactive replayOutcomeCode = "target-inactive" + + // replayNotTerminal reports a delivery the engine has not finished + // with. + replayNotTerminal replayOutcomeCode = "not-terminal" + + // replayInFlight reports that an earlier replay of this event to + // this target is still running. + replayInFlight replayOutcomeCode = "in-flight" +) + +// replayOutcome returns the banner the event log page shows for an +// outcome code, and whether the replay was queued. An unrecognised +// code yields no banner. +func replayOutcome(code string) (string, bool) { + switch replayOutcomeCode(code) { + case replayQueued: + return "Replay queued: a new delivery was created against " + + "the target's current configuration.", true + case replayTargetDeleted: + return "Not replayed: the target this delivery was for has " + + "been deleted. Recreate the target, then replay.", false + case replayTargetMissing: + return "Not replayed: the target this delivery was for no " + + "longer exists.", false + case replayTargetInactive: + return "Not replayed: the target this delivery was for is " + + "deactivated. Activate it, then replay.", false + case replayNotTerminal: + return "Not replayed: this delivery has not finished yet.", + false + case replayInFlight: + return "Not replayed: a delivery of this event to this " + + "target is already in flight.", false + default: + return "", false + } +} + +// HandleDeliveryReplay re-sends a finished delivery's event to its +// target. +// +// A replay never touches the delivery it repeats. It creates a NEW +// pending delivery row for the same event and target and hands it to +// the delivery engine through the same Notifier the receiver uses, so +// the original's status, attempts and timestamps stand as the record +// of what actually happened, and the replay is retried, SSRF-guarded +// and circuit-broken exactly as a first attempt is. +// +// What is re-sent is the stored EVENT body, never the response the +// original delivery received. +// +// The target's configuration is read now rather than as it stood when +// the original ran: a replay exists to deliver where the operator +// currently wants the event to go. That is also why a deleted target +// is refused rather than delivered to from stale configuration. +func (h *Handlers) HandleDeliveryReplay() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + webhook, ok := h.ownedWebhook(w, r) + if !ok { + return + } + + // The body size cap is enforced by the MaxBodySize + // middleware, which runs before CSRF parses the form. + err := r.ParseForm() + if err != nil { + http.Error( + w, "Bad request", http.StatusBadRequest, + ) + + return + } + + h.replayDelivery(w, r, webhook) + } +} + +// replayDelivery performs the replay for a webhook the caller has +// already established the session's user owns. +func (h *Handlers) replayDelivery( + w http.ResponseWriter, + r *http.Request, + webhook database.Webhook, +) { + if !h.dbMgr.DBExists(webhook.ID) { + http.NotFound(w, r) + + return + } + + webhookDB, err := h.dbMgr.GetDB(webhook.ID) + if err != nil { + h.serverError(w, "failed to get webhook database", err) + + return + } + + original, ok := h.loadReplaySource(w, r, webhookDB) + if !ok { + return + } + + if !original.Status.Terminal() { + h.finishReplay(w, r, webhook, replayNotTerminal) + + return + } + + target, code := h.replayTarget(webhook.ID, original.TargetID) + if target == nil { + h.finishReplay(w, r, webhook, code) + + return + } + + h.queueReplay(w, r, webhookDB, webhook, original, target) +} + +// loadReplaySource loads the delivery to be replayed, selecting only +// the columns the replay needs so no association is populated. A +// delivery id that names no row in this webhook's database is a 404. +func (h *Handlers) loadReplaySource( + w http.ResponseWriter, + r *http.Request, + webhookDB *gorm.DB, +) (*database.Delivery, bool) { + var original database.Delivery + + err := webhookDB. + Select("id", "event_id", "target_id", "status"). + First( + &original, "id = ?", chi.URLParam(r, "deliveryID"), + ).Error + if err != nil { + http.NotFound(w, r) + + return nil, false + } + + return &original, true +} + +// queueReplay writes the new delivery and hands it to the engine. +func (h *Handlers) queueReplay( + w http.ResponseWriter, + r *http.Request, + webhookDB *gorm.DB, + webhook database.Webhook, + original *database.Delivery, + target *database.Target, +) { + inFlight, err := countInFlightDeliveries( + webhookDB, original.EventID, target.ID, + ) + if err != nil { + h.serverError( + w, "failed to count in-flight deliveries", err, + ) + + return + } + + if inFlight > 0 { + h.finishReplay(w, r, webhook, replayInFlight) + + return + } + + var event database.Event + + err = webhookDB. + First(&event, "id = ?", original.EventID).Error + if err != nil { + h.serverError(w, "failed to load event for replay", err) + + return + } + + task, err := createReplayDelivery( + webhookDB, webhook.ID, &event, target, + ) + if err != nil { + h.serverError( + w, "failed to create replay delivery", err, + ) + + return + } + + h.mtr.DeliveryReplayed(target.Type) + h.notifier.Notify([]delivery.Task{task}) + + h.log.Info( + "delivery replay queued", + "webhook_id", webhook.ID, + "event_id", event.ID, + "target_id", target.ID, + "replayed_delivery_id", original.ID, + "delivery_id", task.DeliveryID, + ) + + h.finishReplay(w, r, webhook, replayQueued) +} + +// replayTarget loads the delivery's target as it stands now. +// +// The load is Unscoped so that a soft-deleted row is still found: +// deletes are soft and a delivery carries no foreign key to its +// target, so a target's history outlives it, and without the deleted +// row there is no way to tell "you deleted this target" from "this id +// never named anything". A nil target means the replay is refused, +// with the returned code saying why. +func (h *Handlers) replayTarget( + webhookID, targetID string, +) (*database.Target, replayOutcomeCode) { + var target database.Target + + err := h.db.DB().Unscoped().Where( + "id = ? AND webhook_id = ?", targetID, webhookID, + ).First(&target).Error + if err != nil { + return nil, replayTargetMissing + } + + if target.DeletedAt.Valid { + return nil, replayTargetDeleted + } + + if !target.Active { + return nil, replayTargetInactive + } + + return &target, replayQueued +} + +// countInFlightDeliveries reports how many deliveries of this event to +// this target the engine has not finished. +// +// It is the replay-storm guard: a replay is refused while an earlier +// one is still pending or retrying, so a held-down button or a scripted +// loop cannot stack copies of work already queued. It is a check and +// not a lock, so two simultaneous POSTs can still both pass it; the +// per-client rate limit on the route is what bounds that. +func countInFlightDeliveries( + webhookDB *gorm.DB, eventID, targetID string, +) (int64, error) { + var count int64 + + err := webhookDB.Model(&database.Delivery{}).Where( + "event_id = ? AND target_id = ? AND status IN ?", + eventID, targetID, + []database.DeliveryStatus{ + database.DeliveryStatusPending, + database.DeliveryStatusRetrying, + }, + ).Count(&count).Error + + return count, err +} + +// createReplayDelivery writes the new pending delivery row and returns +// the task that carries it to the delivery engine. +// +// The row is written with associations omitted, and neither Event nor +// Target is populated on it: GORM's SaveBeforeAssociations would +// otherwise upsert the whole target row — plaintext config, which for a +// Slack target is the credential — into the per-webhook event database. +// See https://git.eeqj.de/sneak/webhooker/issues/206. +func createReplayDelivery( + webhookDB *gorm.DB, + webhookID string, + event *database.Event, + target *database.Target, +) (delivery.Task, error) { + dlv := &database.Delivery{ + EventID: event.ID, + TargetID: target.ID, + Status: database.DeliveryStatusPending, + } + + err := webhookDB.Omit(clause.Associations).Create(dlv).Error + if err != nil { + return delivery.Task{}, err + } + + return delivery.Task{ + DeliveryID: dlv.ID, + EventID: event.ID, + WebhookID: webhookID, + EntrypointID: event.EntrypointID, + TargetID: target.ID, + TargetName: target.Name, + TargetType: target.Type, + TargetConfig: target.Config, + MaxRetries: target.MaxRetries, + Method: event.Method, + Headers: event.Headers, + ContentType: event.ContentType, + Body: replayBody(event.Body), + AttemptNum: 1, + }, nil +} + +// replayBody returns the stored event body for a replay task to carry +// inline, or nil when it is large enough that the engine should fetch +// it from the per-webhook database instead. +func replayBody(body string) *string { + if len(body) >= delivery.MaxInlineBodySize { + return nil + } + + return &body +} + +// finishReplay redirects back to the event log the replay was +// triggered from, carrying the outcome code the page turns into a +// banner and the page number the form submitted. +func (h *Handlers) finishReplay( + w http.ResponseWriter, + r *http.Request, + webhook database.Webhook, + code replayOutcomeCode, +) { + dest := "/source/" + webhook.ID + "/logs?" + + replayOutcomeParam + "=" + string(code) + + // The page is read from the form rather than the query string: + // this is a POST, and its query string is what logs and Referer + // headers record. + if page := parseNonNegativeInt( + r.PostFormValue("page"), + ); page > 1 { + dest += "&page=" + strconv.Itoa(page) + } + + http.Redirect(w, r, dest, http.StatusSeeOther) +} diff --git a/internal/handlers/delivery_replay_test.go b/internal/handlers/delivery_replay_test.go new file mode 100644 index 0000000..48fc98e --- /dev/null +++ b/internal/handlers/delivery_replay_test.go @@ -0,0 +1,526 @@ +package handlers_test + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/gorm" + "gorm.io/gorm/clause" + "sneak.berlin/go/webhooker/internal/database" + "sneak.berlin/go/webhooker/internal/delivery" + "sneak.berlin/go/webhooker/internal/handlers" + "sneak.berlin/go/webhooker/internal/session" +) + +// paramDeliveryID is the chi URL parameter name the replay handler +// reads. +const paramDeliveryID = "deliveryID" + +// replayTargetURL is a public destination, so a target configured with +// it is one the SSRF guard would accept. Nothing in these tests +// dispatches to it: the notifier is recorded, not run. +const replayTargetURL = "http://93.184.216.34/hook" + +// seedFailedDelivery records an event, a terminally failed delivery of +// it to the given target, and the attempt that failed. +func seedFailedDelivery( + t *testing.T, + dbMgr *database.WebhookDBManager, + webhookID, targetID string, +) (*database.Event, *database.Delivery) { + t.Helper() + + webhookDB, err := dbMgr.GetDB(webhookID) + require.NoError(t, err) + + event := &database.Event{ + WebhookID: webhookID, + EntrypointID: "entrypoint-" + webhookID, + Method: http.MethodPost, + Headers: `{"X-Test":["yes"]}`, + Body: `{"replay":"me"}`, + ContentType: "application/json", + } + + require.NoError(t, webhookDB.Omit( + clause.Associations, + ).Create(event).Error) + + dlv := &database.Delivery{ + EventID: event.ID, + TargetID: targetID, + Status: database.DeliveryStatusFailed, + } + + require.NoError(t, webhookDB.Omit( + clause.Associations, + ).Create(dlv).Error) + + result := &database.DeliveryResult{ + DeliveryID: dlv.ID, + AttemptNum: 1, + Success: false, + StatusCode: http.StatusBadGateway, + Error: "connection refused", + } + + require.NoError(t, webhookDB.Omit( + clause.Associations, + ).Create(result).Error) + + return event, dlv +} + +// loadDelivery reads a delivery back out of a webhook's database. +func loadDelivery( + t *testing.T, webhookDB *gorm.DB, deliveryID string, +) database.Delivery { + t.Helper() + + var dlv database.Delivery + + require.NoError( + t, + webhookDB.First(&dlv, "id = ?", deliveryID).Error, + ) + + return dlv +} + +// listDeliveries reads every delivery of an event. +func listDeliveries( + t *testing.T, webhookDB *gorm.DB, eventID string, +) []database.Delivery { + t.Helper() + + var deliveries []database.Delivery + + require.NoError(t, webhookDB.Where( + "event_id = ?", eventID, + ).Find(&deliveries).Error) + + return deliveries +} + +// theOtherDelivery returns the one delivery in the slice that is not +// excludeID. Identity is used rather than an ordering because the rows +// are minted milliseconds apart and their ids are random. +func theOtherDelivery( + t *testing.T, + deliveries []database.Delivery, + excludeID string, +) database.Delivery { + t.Helper() + + var found []database.Delivery + + for _, d := range deliveries { + if d.ID != excludeID { + found = append(found, d) + } + } + + require.Len(t, found, 1) + + return found[0] +} + +// postReplay runs the real replay handler for one delivery. +func postReplay( + t *testing.T, + h *handlers.Handlers, + sess *session.Session, + webhookID, deliveryID string, +) *httptest.ResponseRecorder { + t.Helper() + + req := postRequest( + "/source/"+webhookID+"/deliveries/"+ + deliveryID+"/replay", + authenticatedCookies( + t, sess, deleteTestUserID, deleteTestUsername, + ), + map[string]string{ + paramSourceID: webhookID, + paramDeliveryID: deliveryID, + }, + ) + + w := httptest.NewRecorder() + h.HandleDeliveryReplay().ServeHTTP(w, req) + + return w +} + +// TestHandleDeliveryReplay_AppendsDeliveryAndLeavesOriginal is the +// core requirement: replaying a failed delivery succeeds, appends a +// new delivery, and leaves the original row and its recorded attempt +// exactly as they were. +// +// It also pins the two things a replay would be wrong to get from the +// original: the task carries the target's CURRENT configuration, which +// this test changes between the failure and the replay, and it carries +// the stored EVENT body rather than anything the failed attempt +// received back. +func TestHandleDeliveryReplay_AppendsDeliveryAndLeavesOriginal( + t *testing.T, +) { + t.Parallel() + + var ( + h *handlers.Handlers + sess *session.Session + db *database.Database + dbMgr *database.WebhookDBManager + notif *recordingNotifier + ) + + app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if) + app.RequireStart() + + t.Cleanup(app.RequireStop) + + wh := seedWebhook(t, db) + tgt := seedConfiguredTarget( + t, db, wh.ID, database.TargetTypeHTTP, + `{"url":"`+replayTargetURL+`"}`, + ) + + event, original := seedFailedDelivery( + t, dbMgr, wh.ID, tgt.ID, + ) + + webhookDB, err := dbMgr.GetDB(wh.ID) + require.NoError(t, err) + + before := loadDelivery(t, webhookDB, original.ID) + + // The operator fixes the destination, which is the whole reason + // to replay. The replay must use this, not the config the + // original delivery ran against. + const fixedConfig = `{"url":"http://93.184.216.34/fixed"}` + + require.NoError(t, db.DB().Model(&database.Target{}). + Where("id = ?", tgt.ID). + Update("config", fixedConfig).Error) + + w := postReplay(t, h, sess, wh.ID, original.ID) + + require.Equal(t, http.StatusSeeOther, w.Code) + assert.Equal( + t, + "/source/"+wh.ID+"/logs?replay=queued", + w.Header().Get("Location"), + ) + + deliveries := listDeliveries(t, webhookDB, event.ID) + require.Len( + t, deliveries, 2, + "replay must append a delivery, not reuse one", + ) + + replayed := theOtherDelivery(t, deliveries, original.ID) + assert.Equal(t, tgt.ID, replayed.TargetID) + assert.Equal(t, event.ID, replayed.EventID) + assert.Equal( + t, database.DeliveryStatusPending, replayed.Status, + ) + + assertDeliveryUntouched(t, webhookDB, before) + + tasks := notif.Tasks() + require.Len(t, tasks, 1) + assertReplayTask( + t, tasks[0], wh.ID, event, tgt, replayed.ID, fixedConfig, + ) + assertNoLeakedTarget(t, webhookDB) +} + +// assertDeliveryUntouched proves a delivery row is exactly as it was +// read before: same terminal status, same timestamps, and the same +// recorded attempts. +func assertDeliveryUntouched( + t *testing.T, + webhookDB *gorm.DB, + before database.Delivery, +) { + t.Helper() + + after := loadDelivery(t, webhookDB, before.ID) + assert.Equal( + t, before.Status, after.Status, + "replay must not resurrect the original delivery", + ) + assert.Equal(t, before.UpdatedAt, after.UpdatedAt) + assert.Equal(t, before.CreatedAt, after.CreatedAt) + + var attempts int64 + + require.NoError(t, webhookDB. + Model(&database.DeliveryResult{}). + Where("delivery_id = ?", before.ID). + Count(&attempts).Error) + assert.Equal( + t, int64(1), attempts, + "the original delivery's attempt history must stand", + ) +} + +// assertReplayTask proves the task handed to the delivery engine is +// the one the receiver would build for this event and this target, and +// that it carries wantConfig — the target's configuration as it stands +// now rather than as the original delivery ran against it. +func assertReplayTask( + t *testing.T, + task delivery.Task, + webhookID string, + event *database.Event, + target *database.Target, + wantDeliveryID, wantConfig string, +) { + t.Helper() + + assert.Equal(t, wantDeliveryID, task.DeliveryID) + assert.Equal(t, event.ID, task.EventID) + assert.Equal(t, webhookID, task.WebhookID) + assert.Equal(t, event.EntrypointID, task.EntrypointID) + assert.Equal(t, target.ID, task.TargetID) + assert.Equal(t, target.Type, task.TargetType) + assert.JSONEq( + t, wantConfig, task.TargetConfig, + "replay must use the target's current configuration", + ) + assert.Equal(t, event.Method, task.Method) + assert.Equal(t, event.Headers, task.Headers) + assert.Equal(t, event.ContentType, task.ContentType) + assert.Equal(t, 1, task.AttemptNum) + require.NotNil(t, task.Body) + assert.Equal( + t, event.Body, *task.Body, + "replay re-sends the stored event body", + ) +} + +// assertNoLeakedTarget proves the per-webhook database holds no target +// rows. AutoMigrate creates the table there because Delivery declares +// the relation, so it is a ROW that signals a leak: an association +// write would have upserted the whole target, plaintext config and +// all, into the event database. See +// https://git.eeqj.de/sneak/webhooker/issues/206. +func assertNoLeakedTarget(t *testing.T, webhookDB *gorm.DB) { + t.Helper() + + var leaked int64 + + require.NoError(t, webhookDB.Unscoped(). + Model(&database.Target{}).Count(&leaked).Error) + assert.Zero( + t, leaked, + "replay must not write the target into the event database", + ) +} + +// TestHandleDeliveryReplay_RefusesDeletedTarget proves the required +// refusal: a target deleted since the delivery ran is reported as +// deleted rather than erroring, and nothing is created or queued. +func TestHandleDeliveryReplay_RefusesDeletedTarget(t *testing.T) { + t.Parallel() + + var ( + h *handlers.Handlers + sess *session.Session + db *database.Database + dbMgr *database.WebhookDBManager + notif *recordingNotifier + ) + + app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if) + app.RequireStart() + + t.Cleanup(app.RequireStop) + + wh := seedWebhook(t, db) + tgt := seedConfiguredTarget( + t, db, wh.ID, database.TargetTypeHTTP, + `{"url":"`+replayTargetURL+`"}`, + ) + + event, original := seedFailedDelivery( + t, dbMgr, wh.ID, tgt.ID, + ) + + // Deletes are soft, so the delivery history outlives the target. + require.NoError(t, db.DB().Where( + "id = ?", tgt.ID, + ).Delete(&database.Target{}).Error) + + w := postReplay(t, h, sess, wh.ID, original.ID) + + require.Equal(t, http.StatusSeeOther, w.Code) + assert.Equal( + t, + "/source/"+wh.ID+"/logs?replay=target-deleted", + w.Header().Get("Location"), + ) + + webhookDB, err := dbMgr.GetDB(wh.ID) + require.NoError(t, err) + + assert.Len( + t, listDeliveries(t, webhookDB, event.ID), 1, + "a refused replay must create no delivery", + ) + assert.Empty( + t, notif.Tasks(), + "a refused replay must queue nothing", + ) + + // The refusal is specific, which is why the target is looked up + // including soft-deleted rows: an id that never named a target + // is a different outcome, and a different message, from one the + // operator deleted. + _, orphan := seedFailedDelivery( + t, dbMgr, wh.ID, "target-that-never-existed", + ) + + missing := postReplay(t, h, sess, wh.ID, orphan.ID) + require.Equal(t, http.StatusSeeOther, missing.Code) + assert.Equal( + t, + "/source/"+wh.ID+"/logs?replay=target-missing", + missing.Header().Get("Location"), + ) +} + +// TestHandleDeliveryReplay_RefusesWhileEarlierReplayInFlight proves +// the replay-storm guard: a second replay of the same event to the +// same target is refused while the first is still queued, so repeated +// submissions cannot stack copies of work the engine has not done. +func TestHandleDeliveryReplay_RefusesWhileEarlierReplayInFlight( + t *testing.T, +) { + t.Parallel() + + var ( + h *handlers.Handlers + sess *session.Session + db *database.Database + dbMgr *database.WebhookDBManager + notif *recordingNotifier + ) + + app := newTestApp(t, &h, &sess, &db, &dbMgr, ¬if) + app.RequireStart() + + t.Cleanup(app.RequireStop) + + wh := seedWebhook(t, db) + tgt := seedConfiguredTarget( + t, db, wh.ID, database.TargetTypeHTTP, + `{"url":"`+replayTargetURL+`"}`, + ) + + event, original := seedFailedDelivery( + t, dbMgr, wh.ID, tgt.ID, + ) + + first := postReplay(t, h, sess, wh.ID, original.ID) + require.Equal(t, http.StatusSeeOther, first.Code) + require.Equal( + t, + "/source/"+wh.ID+"/logs?replay=queued", + first.Header().Get("Location"), + ) + + second := postReplay(t, h, sess, wh.ID, original.ID) + require.Equal(t, http.StatusSeeOther, second.Code) + assert.Equal( + t, + "/source/"+wh.ID+"/logs?replay=in-flight", + second.Header().Get("Location"), + ) + + webhookDB, err := dbMgr.GetDB(wh.ID) + require.NoError(t, err) + + assert.Len( + t, listDeliveries(t, webhookDB, event.ID), 2, + "the refused second replay must add nothing", + ) + assert.Len( + t, notif.Tasks(), 1, + "only the first replay reaches the delivery engine", + ) + + // A delivery the engine has not finished is not replayable + // either, which is the same rule seen from the other side. + queued := theOtherDelivery( + t, listDeliveries(t, webhookDB, event.ID), original.ID, + ) + + pending := postReplay(t, h, sess, wh.ID, queued.ID) + require.Equal(t, http.StatusSeeOther, pending.Code) + assert.Equal( + t, + "/source/"+wh.ID+"/logs?replay=not-terminal", + pending.Header().Get("Location"), + ) +} + +// TestHandleSourceLogs_RendersReplayControlAndBanner proves the action +// reaches the page it belongs on: a finished delivery renders a POST +// form carrying a CSRF token, and the outcome code a refusal redirects +// with becomes a readable message. +func TestHandleSourceLogs_RendersReplayControlAndBanner(t *testing.T) { + t.Parallel() + + var ( + h *handlers.Handlers + sess *session.Session + db *database.Database + dbMgr *database.WebhookDBManager + ) + + app := newTestApp(t, &h, &sess, &db, &dbMgr) + app.RequireStart() + + t.Cleanup(app.RequireStop) + + wh := seedWebhook(t, db) + tgt := seedConfiguredTarget( + t, db, wh.ID, database.TargetTypeHTTP, + `{"url":"`+replayTargetURL+`"}`, + ) + + _, original := seedFailedDelivery(t, dbMgr, wh.ID, tgt.ID) + + body := renderSourceLogsPage(t, h, sess, wh.ID) + + assert.Contains( + t, body, + `action="/source/`+wh.ID+`/deliveries/`+ + original.ID+`/replay"`, + ) + assert.Contains(t, body, `method="POST"`) + assert.Contains(t, body, `name="csrf_token"`) + assert.Contains(t, body, ">Replay<") + + refused := renderSourceLogsPageWithQuery( + t, h, sess, wh.ID, "?replay=target-deleted", + ) + + assert.Contains(t, refused, "alert-error") + assert.Contains(t, refused, "has been deleted") + + // An outcome code nobody issued renders no banner at all. + unknown := renderSourceLogsPageWithQuery( + t, h, sess, wh.ID, "?replay=made-up", + ) + + assert.NotContains(t, unknown, "alert-error") + assert.NotContains(t, unknown, "alert-success") + assert.NotContains(t, unknown, "made-up") +} diff --git a/internal/handlers/handlers_test.go b/internal/handlers/handlers_test.go index 83269d0..2c1ab0b 100644 --- a/internal/handlers/handlers_test.go +++ b/internal/handlers/handlers_test.go @@ -24,9 +24,32 @@ import ( "sneak.berlin/go/webhooker/internal/session" ) -type noopNotifier struct{} +// recordingNotifier is a delivery.Notifier that records the tasks it +// was handed, so a test can prove a handler queued the delivery it +// claims to have queued — and, on the refusal paths, that it queued +// nothing. +type recordingNotifier struct { + mu sync.Mutex + tasks []delivery.Task +} -func (n *noopNotifier) Notify([]delivery.Task) {} +func (n *recordingNotifier) Notify(tasks []delivery.Task) { + n.mu.Lock() + defer n.mu.Unlock() + + n.tasks = append(n.tasks, tasks...) +} + +// Tasks returns a copy of the recorded tasks. +func (n *recordingNotifier) Tasks() []delivery.Task { + n.mu.Lock() + defer n.mu.Unlock() + + out := make([]delivery.Task, len(n.tasks)) + copy(out, n.tasks) + + return out +} // recordingEvictor is a delivery.WebhookEvictor that records // the webhook ids it was asked to evict, so a test can prove @@ -74,8 +97,11 @@ func newTestApp( database.NewWebhookDBManager, healthcheck.New, session.New, - func() delivery.Notifier { - return &noopNotifier{} + func() *recordingNotifier { + return &recordingNotifier{} + }, + func(n *recordingNotifier) delivery.Notifier { + return n }, func() *recordingEvictor { return &recordingEvictor{} diff --git a/internal/handlers/source_logs_test.go b/internal/handlers/source_logs_test.go index e4a2e55..5c2f1a2 100644 --- a/internal/handlers/source_logs_test.go +++ b/internal/handlers/source_logs_test.go @@ -60,10 +60,26 @@ func renderSourceLogsPage( ) string { t.Helper() + return renderSourceLogsPageWithQuery( + t, h, sess, webhookID, "", + ) +} + +// renderSourceLogsPageWithQuery is renderSourceLogsPage over a +// caller-supplied query string, for the page state a redirect back to +// the log carries in one. +func renderSourceLogsPageWithQuery( + t *testing.T, + h *handlers.Handlers, + sess *session.Session, + webhookID, query string, +) string { + t.Helper() + req := httptest.NewRequestWithContext( context.Background(), http.MethodGet, - "/source/"+webhookID+"/logs", + "/source/"+webhookID+"/logs"+query, nil, ) diff --git a/internal/handlers/source_management.go b/internal/handlers/source_management.go index b1ba363..581e218 100644 --- a/internal/handlers/source_management.go +++ b/internal/handlers/source_management.go @@ -781,16 +781,25 @@ func (h *Handlers) HandleSourceLogs() http.HandlerFunc { totalPages++ } + // The banner a replay POST redirected back with. The + // message comes from a fixed set keyed by the outcome + // code, never from the query string itself. + replayMsg, replayOK := replayOutcome( + r.URL.Query().Get(replayOutcomeParam), + ) + data := map[string]any{ - tmplKeyWebhook: &webhook, - "Events": evts, - "Page": page, - "TotalPages": totalPages, - "TotalEvents": total, - "HasPrev": page > 1, - "HasNext": page < totalPages, - "PrevPage": page - 1, - "NextPage": page + 1, + tmplKeyWebhook: &webhook, + "Events": evts, + "ReplayMessage": replayMsg, + "ReplayQueued": replayOK, + "Page": page, + "TotalPages": totalPages, + "TotalEvents": total, + "HasPrev": page > 1, + "HasNext": page < totalPages, + "PrevPage": page - 1, + "NextPage": page + 1, } h.renderTemplate(w, r, "source_logs.html", data) diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index b69538f..5fc5354 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -82,6 +82,7 @@ type Set struct { deliveriesSucceeded *prometheus.CounterVec deliveriesFailed *prometheus.CounterVec deliveryRetries *prometheus.CounterVec + deliveryReplays *prometheus.CounterVec deliveryDuration *prometheus.HistogramVec deliveriesPending *prometheus.GaugeVec deliveriesRetrying *prometheus.GaugeVec @@ -149,6 +150,22 @@ func (s *Set) ObserveDeliveryDuration( Observe(d.Seconds()) } +// DeliveryReplayed counts one delivery an operator replayed from the +// event log. +// +// A replay runs the ordinary engine path, so it already moves the +// attempt, outcome and duration series exactly as a first delivery +// does — deliberately, since a replay is a real delivery and hiding it +// from those would misreport the pipeline. This counter is the one +// place the two are distinguishable, and it carries the existing +// target-type label rather than adding a replay dimension to every +// other series. +func (s *Set) DeliveryReplayed(t database.TargetType) { + s.deliveryReplays. + WithLabelValues(normalizeTargetType(t)). + Inc() +} + // DeliveryStatusChanged counts a delivery's transition into a new // status. The mapping from status to counter lives here, next to the // collectors, so the engine has a single call for every transition it @@ -271,6 +288,16 @@ func (s *Set) registerCounters(factory promauto.Factory) { }, []string{targetTypeLabel}, ) + + s.deliveryReplays = factory.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "delivery_replays_total", + Help: "Deliveries an operator replayed from the " + + "event log, by target type.", + }, + []string{targetTypeLabel}, + ) } func (s *Set) registerGauges(factory promauto.Factory) { @@ -322,6 +349,7 @@ func (s *Set) initSeries() { s.deliveriesSucceeded.WithLabelValues(label) s.deliveriesFailed.WithLabelValues(label) s.deliveryRetries.WithLabelValues(label) + s.deliveryReplays.WithLabelValues(label) s.deliveriesPending.WithLabelValues(label) s.deliveriesRetrying.WithLabelValues(label) s.circuitBreakersOpen.WithLabelValues(label) diff --git a/internal/middleware/ratelimit.go b/internal/middleware/ratelimit.go index a253c69..9b13396 100644 --- a/internal/middleware/ratelimit.go +++ b/internal/middleware/ratelimit.go @@ -34,6 +34,16 @@ const ( // password change rate limit. passwordChangeRateInterval = 1 * time.Minute + // replayRateLimit is the maximum number of delivery replays one + // client may queue per interval. Each replay puts a delivery on + // the engine's queue, so without a ceiling one operator holding + // the button down — or scripting it — queues unbounded outbound + // work. It sits far above any rate a person clicks at. + replayRateLimit = 30 + + // replayRateInterval is the time window for the replay limit. + replayRateInterval = 1 * time.Minute + // receiverRateInterval is the time window for the webhook // receiver rate limit. The configured limit is expressed in // requests per minute. @@ -290,6 +300,21 @@ func (m *Middleware) PasswordChangeRateLimit() func(http.Handler) http.Handler { ) } +// ReplayRateLimit returns middleware that enforces per-IP rate +// limiting on delivery replays. +// +// Like the password-change limit it is spent on arrival, which is safe +// for the same reason: RequireAuth runs ahead of it, so only a request +// already carrying a valid session can reach the bucket. +func (m *Middleware) ReplayRateLimit() func(http.Handler) http.Handler { + return m.postRateLimit( + replayRateLimit, + replayRateInterval, + "delivery replay rate limit exceeded", + "Too many replays. Please try again later.", + ) +} + // postRateLimit builds middleware that enforces a per-IP rate // limit on POST requests only; all other methods pass through // unaffected. Requests over the limit receive a 429 with the diff --git a/internal/server/routes.go b/internal/server/routes.go index a3cc6ff..846d3a3 100644 --- a/internal/server/routes.go +++ b/internal/server/routes.go @@ -201,6 +201,18 @@ func (s *Server) setupSourceRoutes() { "/logs/{eventID}/body", s.h.HandleEventBodyDownload(), ) + // Replay is the one page action that queues outbound work: + // it creates a delivery from a stored event and hands it to + // the delivery engine. The rate limit is what bounds a + // held-down button or a scripted loop; the handler + // separately refuses a replay while an earlier one for the + // same event and target is still in flight. POST only, so + // the action cannot be taken by a link, a prefetch or an + // image tag. + r.With(s.mw.ReplayRateLimit()).Post( + "/deliveries/{deliveryID}/replay", + s.h.HandleDeliveryReplay(), + ) r.Post( "/entrypoints", s.h.HandleEntrypointCreate(), diff --git a/internal/server/routes_test.go b/internal/server/routes_test.go index 087dee2..e8212ec 100644 --- a/internal/server/routes_test.go +++ b/internal/server/routes_test.go @@ -310,6 +310,75 @@ func (e *testEnv) seedEvent( return event } +// seedTarget creates an active HTTP target for a webhook. +func (e *testEnv) seedTarget( + t *testing.T, + webhookID string, +) *database.Target { + t.Helper() + + tgt := &database.Target{ + WebhookID: webhookID, + Name: "routed-target", + Type: database.TargetTypeHTTP, + Active: true, + Config: `{"url":"http://93.184.216.34/hook"}`, + } + + require.NoError( + t, + e.db.DB().Omit(clause.Associations).Create(tgt).Error, + ) + + return tgt +} + +// seedFailedDelivery records a terminally failed delivery of an event +// to a target in the webhook's own database. +func (e *testEnv) seedFailedDelivery( + t *testing.T, + webhookID, eventID, targetID string, +) *database.Delivery { + t.Helper() + + webhookDB, err := e.dbMgr.GetDB(webhookID) + require.NoError(t, err) + + dlv := &database.Delivery{ + EventID: eventID, + TargetID: targetID, + Status: database.DeliveryStatusFailed, + } + + require.NoError( + t, + webhookDB.Omit(clause.Associations).Create(dlv).Error, + ) + + return dlv +} + +// countDeliveries reports how many deliveries a webhook's database +// holds. +func (e *testEnv) countDeliveries( + t *testing.T, webhookID string, +) int64 { + t.Helper() + + webhookDB, err := e.dbMgr.GetDB(webhookID) + require.NoError(t, err) + + var count int64 + + require.NoError( + t, + webhookDB.Model(&database.Delivery{}). + Count(&count).Error, + ) + + return count +} + // storedHash reads the current password hash for a username. func (e *testEnv) storedHash(t *testing.T, username string) string { t.Helper() @@ -674,6 +743,85 @@ func TestSourceLogsBody_OtherUser404s(t *testing.T) { assert.Equal(t, "/pages/login", anon.Header().Get("Location")) } +// TestDeliveryReplay_PostOnlyAndCSRFProtected walks the replay action +// through the production router rather than a forged route context, +// which is the only way to prove what the route group actually gives +// it: a GET cannot trigger a replay, an unauthenticated request never +// reaches the handler, a POST without the token is refused by CSRF, +// and the form the template emits — token and action URL both — works +// as rendered. +func TestDeliveryReplay_PostOnlyAndCSRFProtected(t *testing.T) { + t.Parallel() + + env := newTestEnv(t) + + userID, _ := env.seedUser(t, "replayer", "somepassword") + cookies := env.authCookies(t, userID, "replayer") + + wh := env.seedWebhook(t, userID) + tgt := env.seedTarget(t, wh.ID) + evt := env.seedEvent(t, wh.ID, `{"replay":"me"}`) + dlv := env.seedFailedDelivery(t, wh.ID, evt.ID, tgt.ID) + + path := "/source/" + wh.ID + "/deliveries/" + dlv.ID + + "/replay" + + assert.Equal( + t, http.StatusMethodNotAllowed, + env.get(path, cookies).Code, + "a replay must not be reachable by GET", + ) + + assert.Equal( + t, http.StatusForbidden, + env.post(path, url.Values{}, cookies).Code, + "a replay POST without a CSRF token must be refused", + ) + + anon := env.post(path, url.Values{}, nil) + assert.Equal(t, http.StatusForbidden, anon.Code) + + require.Equal( + t, int64(1), env.countDeliveries(t, wh.ID), + "no refused request may have created a delivery", + ) + + // The token and the action URL both come out of the rendered + // page, so a typo in either the route pattern or the template + // fails here. + logsPath := "/source/" + wh.ID + "/logs" + + token, cookies := env.csrfFrom(t, logsPath, cookies) + + page := env.get(logsPath, cookies) + require.Equal(t, http.StatusOK, page.Code) + + action := regexp.MustCompile( + `action="(/source/[^"]+/replay)"`, + ).FindStringSubmatch(page.Body.String()) + require.Len( + t, action, 2, + "a finished delivery should render a replay form", + ) + + form := url.Values{} + form.Set("csrf_token", token) + + w := env.post( + html.UnescapeString(action[1]), form, cookies, + ) + + require.Equal(t, http.StatusSeeOther, w.Code) + assert.Equal( + t, logsPath+"?replay=queued", + w.Header().Get("Location"), + ) + assert.Equal( + t, int64(2), env.countDeliveries(t, wh.ID), + "the replay appends a delivery", + ) +} + // metricsConfig is a Config differing from the routing default only // in the two /metrics credentials. func metricsConfig( diff --git a/templates/source_logs.html b/templates/source_logs.html index cf48983..46c867e 100644 --- a/templates/source_logs.html +++ b/templates/source_logs.html @@ -12,6 +12,10 @@ + {{if .ReplayMessage}} +
{{.ReplayMessage}}
+ {{end}} +
{{range .Events}} @@ -40,6 +44,23 @@ {{if .BodyTruncated}}

Body truncated for display: showing {{.BodyShownBytes}} of {{.BodyBytes}} bytes. The stored body is unchanged — download the full body.

{{end}} + + {{if .Deliveries}} +
+ {{range .Deliveries}} +
+ {{.Target.Name}}: {{.Status}} + {{if .Status.Terminal}} +
+ + + +
+ {{end}} +
+ {{end}} +
+ {{end}}
{{else}} -- 2.49.1