Compare commits
1 Commits
ea6639ff31
...
c1fce4326c
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c1fce4326c |
20
README.md
20
README.md
@@ -904,6 +904,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
|
||||
@@ -1231,6 +1248,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` |
|
||||
@@ -1828,6 +1846,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 |
|
||||
@@ -1912,6 +1931,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
|
||||
│ │ ├── event_log_view.go # Event log projection, byte-capped in SQL
|
||||
│ │ ├── healthcheck.go # Health check handler
|
||||
│ │ ├── index.go # Index page handler
|
||||
|
||||
@@ -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
|
||||
|
||||
378
internal/handlers/delivery_replay.go
Normal file
378
internal/handlers/delivery_replay.go
Normal file
@@ -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)
|
||||
}
|
||||
526
internal/handlers/delivery_replay_test.go
Normal file
526
internal/handlers/delivery_replay_test.go
Normal file
@@ -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")
|
||||
}
|
||||
@@ -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{}
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
|
||||
@@ -777,9 +777,18 @@ 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,
|
||||
"ReplayMessage": replayMsg,
|
||||
"ReplayQueued": replayOK,
|
||||
"Page": page,
|
||||
"TotalPages": totalPages,
|
||||
"TotalEvents": total,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -12,6 +12,10 @@
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{{if .ReplayMessage}}
|
||||
<div class="{{if .ReplayQueued}}alert-success{{else}}alert-error{{end}}">{{.ReplayMessage}}</div>
|
||||
{{end}}
|
||||
|
||||
<div class="card">
|
||||
<div class="divide-y divide-gray-100">
|
||||
{{range .Events}}
|
||||
@@ -40,6 +44,23 @@
|
||||
{{if .BodyTruncated}}
|
||||
<p class="mt-2 text-xs text-gray-500">Body truncated for display: showing {{.BodyShownBytes}} of {{.BodyBytes}} bytes. The stored body is unchanged — <a href="/source/{{$.Webhook.ID}}/logs/{{.ID}}/body" class="text-primary-600 hover:text-primary-700 underline">download the full body</a>.</p>
|
||||
{{end}}
|
||||
|
||||
{{if .Deliveries}}
|
||||
<div class="mt-3 pt-4 border-t border-gray-200">
|
||||
{{range .Deliveries}}
|
||||
<div class="flex items-center justify-between py-2 text-xs text-gray-700">
|
||||
<span><span class="font-medium">{{.Target.Name}}</span>: {{.Status}}</span>
|
||||
{{if .Status.Terminal}}
|
||||
<form method="POST" action="/source/{{$.Webhook.ID}}/deliveries/{{.ID}}/replay" class="inline">
|
||||
<input type="hidden" name="csrf_token" value="{{$.CSRFToken}}">
|
||||
<input type="hidden" name="page" value="{{$.Page}}">
|
||||
<button type="submit" class="text-xs text-primary-600 hover:text-primary-700" title="Send this event to the target again">Replay</button>
|
||||
</form>
|
||||
{{end}}
|
||||
</div>
|
||||
{{end}}
|
||||
</div>
|
||||
{{end}}
|
||||
</div>
|
||||
</div>
|
||||
{{else}}
|
||||
|
||||
Reference in New Issue
Block a user