Compare commits
3 Commits
641a8ebd66
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 4f5ecb18e5 | |||
| 734606b7af | |||
| ee7c626071 |
31
README.md
31
README.md
@@ -363,10 +363,12 @@ events should be forwarded.
|
|||||||
greater than 0, failed deliveries are retried with exponential backoff
|
greater than 0, failed deliveries are retried with exponential backoff
|
||||||
up to `max_retries` attempts, protected by a per-target circuit
|
up to `max_retries` attempts, protected by a per-target circuit
|
||||||
breaker.
|
breaker.
|
||||||
- **`database`** — Confirm the event is stored in the webhook's
|
- **`database`** — Archive the full event as a row into a separate
|
||||||
per-webhook database (no external delivery). Since events are always
|
per-webhook archive database (`archive-{webhookID}.db`) for long-term
|
||||||
written to the per-webhook DB on ingestion, this target marks delivery
|
retention, with an optional creation-validated expiry (default: keep
|
||||||
as immediately successful. Useful for ensuring durable event archival.
|
forever). No external delivery and no retries; an archive write
|
||||||
|
failure fails the delivery. See the database target section under
|
||||||
|
"Per-Webhook Event Databases" for the full semantics.
|
||||||
- **`log`** — Write the event to the application log (stdout). Useful
|
- **`log`** — Write the event to the application log (stdout). Useful
|
||||||
for debugging.
|
for debugging.
|
||||||
|
|
||||||
@@ -512,11 +514,22 @@ This separation provides:
|
|||||||
page cache, and its own lock, so concurrent event ingestion across
|
page cache, and its own lock, so concurrent event ingestion across
|
||||||
webhooks won't contend.
|
webhooks won't contend.
|
||||||
|
|
||||||
The **database target type** leverages this architecture: since events
|
The **database target type** builds on this architecture to provide
|
||||||
are already stored in the per-webhook database by design, the database
|
long-term archiving, separate from the per-webhook event database (which
|
||||||
target simply marks the delivery as immediately successful. The
|
may prune events under its own retention). Delivering to a database
|
||||||
per-webhook DB IS the dedicated event database — that's the whole point
|
target writes the full event — body, headers, method, content type, and
|
||||||
of the database target type.
|
webhook/entrypoint/event identifiers — as a row into a dedicated archive
|
||||||
|
database, `archive-{webhookID}.db`, stored under the data directory
|
||||||
|
beside the event database. After each write the archive handle is closed
|
||||||
|
and reopened, debounced to at most once per second, so an operator can
|
||||||
|
move the archive file away for offline archiving without stopping the
|
||||||
|
service; a moved or removed archive file is recreated automatically on
|
||||||
|
the next write. An optional `expiry` in the target's config JSON (e.g.
|
||||||
|
`{"expiry":"720h"}`) is validated when the target is created — the
|
||||||
|
default (unset or the literal `never`) keeps rows forever — and rows
|
||||||
|
older than the expiry are pruned each time the archive is (re)opened. An
|
||||||
|
archive write failure is never silent success: the delivery records a
|
||||||
|
failed attempt with the error and is marked failed.
|
||||||
|
|
||||||
The **Slack target type** sends webhook events as formatted messages to
|
The **Slack target type** sends webhook events as formatted messages to
|
||||||
any Slack-compatible incoming webhook URL (works with Slack, Mattermost,
|
any Slack-compatible incoming webhook URL (works with Slack, Mattermost,
|
||||||
|
|||||||
@@ -345,33 +345,29 @@ func TestDeliverDatabase_ImmediateSuccess(
|
|||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
db := testWebhookDB(t)
|
db := testWebhookDB(t)
|
||||||
e := testEngine(t, 1)
|
|
||||||
|
|
||||||
event := seedEvent(t, db, `{"db":"target"}`)
|
// The database target archives for real now, so the engine
|
||||||
|
// needs a webhook DB manager to locate the data directory.
|
||||||
dlv := seedDelivery(
|
e := delivery.NewTestEngineWithDB(
|
||||||
t, db, event.ID, uuid.New().String(),
|
nil,
|
||||||
database.DeliveryStatusPending,
|
database.NewTestWebhookDBManager(t.TempDir()),
|
||||||
|
slog.New(slog.NewTextHandler(
|
||||||
|
os.Stderr,
|
||||||
|
&slog.HandlerOptions{Level: slog.LevelDebug},
|
||||||
|
)),
|
||||||
|
&http.Client{Timeout: 5 * time.Second},
|
||||||
|
1,
|
||||||
)
|
)
|
||||||
|
|
||||||
d := &database.Delivery{
|
event := seedEvent(t, db, `{"db":"target"}`)
|
||||||
EventID: event.ID,
|
d := seedDatabaseTargetDelivery(t, db, event, "")
|
||||||
TargetID: dlv.TargetID,
|
|
||||||
Status: database.DeliveryStatusPending,
|
|
||||||
Event: event,
|
|
||||||
Target: database.Target{
|
|
||||||
Name: "test-db",
|
|
||||||
Type: database.TargetTypeDatabase,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
d.ID = dlv.ID
|
|
||||||
|
|
||||||
e.ExportDeliverDatabase(db, d)
|
e.ExportDeliverDatabase(db, d)
|
||||||
|
|
||||||
var updated database.Delivery
|
var updated database.Delivery
|
||||||
|
|
||||||
require.NoError(t, db.First(
|
require.NoError(t, db.First(
|
||||||
&updated, "id = ?", dlv.ID,
|
&updated, "id = ?", d.ID,
|
||||||
).Error)
|
).Error)
|
||||||
|
|
||||||
assert.Equal(t,
|
assert.Equal(t,
|
||||||
@@ -382,7 +378,7 @@ func TestDeliverDatabase_ImmediateSuccess(
|
|||||||
var result database.DeliveryResult
|
var result database.DeliveryResult
|
||||||
|
|
||||||
require.NoError(t, db.Where(
|
require.NoError(t, db.Where(
|
||||||
"delivery_id = ?", dlv.ID,
|
"delivery_id = ?", d.ID,
|
||||||
).First(&result).Error)
|
).First(&result).Error)
|
||||||
|
|
||||||
assert.True(t, result.Success)
|
assert.True(t, result.Success)
|
||||||
@@ -1161,7 +1157,19 @@ func TestProcessDelivery_RoutesToCorrectHandler(
|
|||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
db := testWebhookDB(t)
|
db := testWebhookDB(t)
|
||||||
e := testEngine(t, 1)
|
|
||||||
|
// The database target archives for real now, so the engine
|
||||||
|
// needs a webhook DB manager to locate the data directory.
|
||||||
|
e := delivery.NewTestEngineWithDB(
|
||||||
|
nil,
|
||||||
|
database.NewTestWebhookDBManager(t.TempDir()),
|
||||||
|
slog.New(slog.NewTextHandler(
|
||||||
|
os.Stderr,
|
||||||
|
&slog.HandlerOptions{Level: slog.LevelDebug},
|
||||||
|
)),
|
||||||
|
&http.Client{Timeout: 5 * time.Second},
|
||||||
|
1,
|
||||||
|
)
|
||||||
|
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
name string
|
name string
|
||||||
|
|||||||
@@ -273,3 +273,64 @@ func NewTestCircuitBreaker(
|
|||||||
cooldown: cooldown,
|
cooldown: cooldown,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ExportArchivedEvent aliases the archive row type so black-box
|
||||||
|
// tests can construct and read archive rows.
|
||||||
|
type ExportArchivedEvent = archivedEvent
|
||||||
|
|
||||||
|
// ExportArchiveWriter wraps an archiveWriter so black-box tests
|
||||||
|
// can exercise the per-webhook archive file mechanics.
|
||||||
|
type ExportArchiveWriter struct {
|
||||||
|
w *archiveWriter
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewExportArchiveWriter builds an archive writer for tests,
|
||||||
|
// optionally overriding the reopen debounce (a non-positive
|
||||||
|
// debounce keeps the production default).
|
||||||
|
func NewExportArchiveWriter(
|
||||||
|
path string, log *slog.Logger, debounce time.Duration,
|
||||||
|
) *ExportArchiveWriter {
|
||||||
|
w := newArchiveWriter(path, log)
|
||||||
|
if debounce > 0 {
|
||||||
|
w.debounce = debounce
|
||||||
|
}
|
||||||
|
|
||||||
|
return &ExportArchiveWriter{w: w}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Write archives a row through the writer.
|
||||||
|
func (e *ExportArchiveWriter) Write(
|
||||||
|
row ExportArchivedEvent, expiry time.Duration,
|
||||||
|
) error {
|
||||||
|
return e.w.write(row, expiry)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Open opens the archive file, pruning when expiry is positive.
|
||||||
|
func (e *ExportArchiveWriter) Open(expiry time.Duration) error {
|
||||||
|
return e.w.open(expiry)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Reopen closes and reopens the archive file.
|
||||||
|
func (e *ExportArchiveWriter) Reopen(
|
||||||
|
expiry time.Duration,
|
||||||
|
) error {
|
||||||
|
return e.w.reopen(expiry)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Reopens reports how many times the file has been opened.
|
||||||
|
func (e *ExportArchiveWriter) Reopens() int {
|
||||||
|
return e.w.reopens
|
||||||
|
}
|
||||||
|
|
||||||
|
// DB returns the writer's current open handle for row
|
||||||
|
// inspection in tests.
|
||||||
|
func (e *ExportArchiveWriter) DB() *gorm.DB {
|
||||||
|
return e.w.db
|
||||||
|
}
|
||||||
|
|
||||||
|
// ExportParseArchiveExpiry exposes parseArchiveExpiry.
|
||||||
|
func ExportParseArchiveExpiry(
|
||||||
|
configJSON string,
|
||||||
|
) (time.Duration, error) {
|
||||||
|
return parseArchiveExpiry(configJSON)
|
||||||
|
}
|
||||||
|
|||||||
@@ -2,21 +2,38 @@ package delivery
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"path/filepath"
|
||||||
|
"sync"
|
||||||
|
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
"sneak.berlin/go/webhooker/internal/database"
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
)
|
)
|
||||||
|
|
||||||
// databaseTarget is a fire-and-forget target: the event is
|
// databaseTarget is a no-retry target that archives the
|
||||||
// already persisted in the per-webhook database by the time
|
// full inbound event into a per-webhook archive SQLite file,
|
||||||
// delivery runs, so the target records a single successful
|
// separate from the per-webhook event database. The event is
|
||||||
// attempt. (Durable archiving to a separate store is tracked
|
// already persisted in the per-webhook event DB by the time
|
||||||
// as its own work.)
|
// delivery runs; the database target additionally writes a
|
||||||
|
// durable long-term copy into archive-{webhookID}.db and then
|
||||||
|
// records a single attempt whose outcome reflects whether the
|
||||||
|
// archive write succeeded. See archiveWriter for the
|
||||||
|
// close/reopen, auto-recreate, and expiry semantics.
|
||||||
type databaseTarget struct {
|
type databaseTarget struct {
|
||||||
eng *Engine
|
eng *Engine
|
||||||
|
|
||||||
|
mu sync.Mutex
|
||||||
|
writers map[string]*archiveWriter
|
||||||
}
|
}
|
||||||
|
|
||||||
// Deliver implements Target.
|
// Deliver implements Target. It archives the event, then
|
||||||
|
// records one successful attempt and marks the delivery
|
||||||
|
// delivered. An archiving error fails the delivery: the
|
||||||
|
// attempt is recorded as failed with the error and the
|
||||||
|
// delivery is marked failed, so a target that could not do
|
||||||
|
// its one job (archiving) never reports success. The target
|
||||||
|
// does not retry; the event remains durably stored in the
|
||||||
|
// per-webhook event database.
|
||||||
func (t *databaseTarget) Deliver(
|
func (t *databaseTarget) Deliver(
|
||||||
_ context.Context,
|
_ context.Context,
|
||||||
webhookDB *gorm.DB,
|
webhookDB *gorm.DB,
|
||||||
@@ -24,6 +41,27 @@ func (t *databaseTarget) Deliver(
|
|||||||
_ *Task,
|
_ *Task,
|
||||||
_ Scheduler,
|
_ Scheduler,
|
||||||
) {
|
) {
|
||||||
|
err := t.archive(d)
|
||||||
|
if err != nil {
|
||||||
|
t.eng.log.Error(
|
||||||
|
"failed to archive event to database target",
|
||||||
|
"delivery_id", d.ID,
|
||||||
|
"event_id", d.EventID,
|
||||||
|
"error", err,
|
||||||
|
)
|
||||||
|
|
||||||
|
t.eng.recordResult(
|
||||||
|
webhookDB, d, 1, false, 0, "",
|
||||||
|
err.Error(), 0,
|
||||||
|
)
|
||||||
|
|
||||||
|
t.eng.updateDeliveryStatus(
|
||||||
|
webhookDB, d, database.DeliveryStatusFailed,
|
||||||
|
)
|
||||||
|
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
t.eng.recordResult(
|
t.eng.recordResult(
|
||||||
webhookDB, d, 1, true, 0, "", "", 0,
|
webhookDB, d, 1, true, 0, "", "", 0,
|
||||||
)
|
)
|
||||||
@@ -32,3 +70,68 @@ func (t *databaseTarget) Deliver(
|
|||||||
webhookDB, d, database.DeliveryStatusDelivered,
|
webhookDB, d, database.DeliveryStatusDelivered,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// archive writes the full event as a row into the webhook's
|
||||||
|
// archive database, honouring the optional per-target expiry
|
||||||
|
// parsed from the target config JSON.
|
||||||
|
func (t *databaseTarget) archive(d *database.Delivery) error {
|
||||||
|
webhookID := d.Event.WebhookID
|
||||||
|
if webhookID == "" {
|
||||||
|
return errArchiveMissingWebhookID
|
||||||
|
}
|
||||||
|
|
||||||
|
expiry, err := parseArchiveExpiry(d.Target.Config)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
w, err := t.writerFor(webhookID)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
row := archivedEvent{
|
||||||
|
EventID: d.Event.ID,
|
||||||
|
WebhookID: webhookID,
|
||||||
|
EntrypointID: d.Event.EntrypointID,
|
||||||
|
Method: d.Event.Method,
|
||||||
|
Headers: d.Event.Headers,
|
||||||
|
Body: d.Event.Body,
|
||||||
|
ContentType: d.Event.ContentType,
|
||||||
|
}
|
||||||
|
|
||||||
|
return w.write(row, expiry)
|
||||||
|
}
|
||||||
|
|
||||||
|
// writerFor returns the archiveWriter for a webhook, creating
|
||||||
|
// and caching it on first use. Each webhook has one writer so
|
||||||
|
// its close/reopen debounce state is shared across concurrent
|
||||||
|
// deliveries. The archive file lives beside the per-webhook
|
||||||
|
// event database in the data directory.
|
||||||
|
func (t *databaseTarget) writerFor(
|
||||||
|
webhookID string,
|
||||||
|
) (*archiveWriter, error) {
|
||||||
|
if t.eng.dbManager == nil {
|
||||||
|
return nil, errArchiveNoDataDir
|
||||||
|
}
|
||||||
|
|
||||||
|
dir := filepath.Dir(t.eng.dbManager.DBPath(webhookID))
|
||||||
|
path := filepath.Join(
|
||||||
|
dir, fmt.Sprintf("archive-%s.db", webhookID),
|
||||||
|
)
|
||||||
|
|
||||||
|
t.mu.Lock()
|
||||||
|
defer t.mu.Unlock()
|
||||||
|
|
||||||
|
if t.writers == nil {
|
||||||
|
t.writers = make(map[string]*archiveWriter)
|
||||||
|
}
|
||||||
|
|
||||||
|
w, ok := t.writers[webhookID]
|
||||||
|
if !ok {
|
||||||
|
w = newArchiveWriter(path, t.eng.log)
|
||||||
|
t.writers[webhookID] = w
|
||||||
|
}
|
||||||
|
|
||||||
|
return w, nil
|
||||||
|
}
|
||||||
|
|||||||
312
internal/delivery/target_database_archive.go
Normal file
312
internal/delivery/target_database_archive.go
Normal file
@@ -0,0 +1,312 @@
|
|||||||
|
package delivery
|
||||||
|
|
||||||
|
import (
|
||||||
|
"database/sql"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"log/slog"
|
||||||
|
"os"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gorm.io/driver/sqlite"
|
||||||
|
"gorm.io/gorm"
|
||||||
|
)
|
||||||
|
|
||||||
|
// archiveExpiryNever is the expiry sentinel (and default) that
|
||||||
|
// disables pruning so archived rows are kept forever.
|
||||||
|
const archiveExpiryNever = "never"
|
||||||
|
|
||||||
|
// archiveReopenDebounce bounds how often an archive file is
|
||||||
|
// closed and reopened. After each write the handle is closed
|
||||||
|
// and reopened so an operator can move the file away for
|
||||||
|
// offline archiving, but never more than once per this window.
|
||||||
|
const archiveReopenDebounce = time.Second
|
||||||
|
|
||||||
|
var (
|
||||||
|
// errArchiveMissingWebhookID is returned when an event to
|
||||||
|
// archive has no webhook id to key its archive file on.
|
||||||
|
errArchiveMissingWebhookID = errors.New(
|
||||||
|
"cannot archive event without a webhook id",
|
||||||
|
)
|
||||||
|
|
||||||
|
// errArchiveNoDataDir is returned when the database target
|
||||||
|
// has no webhook database manager and so cannot locate the
|
||||||
|
// data directory for archive files.
|
||||||
|
errArchiveNoDataDir = errors.New(
|
||||||
|
"database target has no data directory",
|
||||||
|
)
|
||||||
|
|
||||||
|
// errArchiveExpiryNotPositive is returned when a
|
||||||
|
// user-supplied archive expiry parses as a duration but is
|
||||||
|
// zero or negative; "never" is the way to disable pruning.
|
||||||
|
errArchiveExpiryNotPositive = errors.New(
|
||||||
|
"expiry must be a positive duration or \"never\"",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
// databaseTargetConfig is the optional per-target JSON config
|
||||||
|
// for a database (archive) target.
|
||||||
|
type databaseTargetConfig struct {
|
||||||
|
// Expiry is a Go duration (e.g. "720h") after which
|
||||||
|
// archived rows are pruned, or "never" (the default) to
|
||||||
|
// keep them forever.
|
||||||
|
Expiry string `json:"expiry"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// archivedEvent is one fully captured webhook event stored in a
|
||||||
|
// per-webhook archive database for long-term retention. It is a
|
||||||
|
// self-contained copy — independent of the per-webhook event
|
||||||
|
// database, which may prune events under its own retention.
|
||||||
|
type archivedEvent struct {
|
||||||
|
ID uint `gorm:"primaryKey;autoIncrement"`
|
||||||
|
EventID string `gorm:"index"`
|
||||||
|
WebhookID string
|
||||||
|
EntrypointID string
|
||||||
|
Method string
|
||||||
|
Headers string
|
||||||
|
Body string
|
||||||
|
ContentType string
|
||||||
|
|
||||||
|
// ArchivedAt is when the row was archived and is the age
|
||||||
|
// basis for expiry pruning.
|
||||||
|
ArchivedAt time.Time `gorm:"index"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// parseArchiveExpiry reads the optional expiry from a database
|
||||||
|
// target's config JSON. An empty config, an empty expiry, or
|
||||||
|
// the literal "never" all mean keep forever, returned as a zero
|
||||||
|
// duration. Any other value must parse as a positive Go
|
||||||
|
// duration; a set-but-invalid value (unparseable, zero, or
|
||||||
|
// negative) is an error rather than a silent default, matching
|
||||||
|
// ValidateArchiveExpiry at target creation.
|
||||||
|
func parseArchiveExpiry(
|
||||||
|
configJSON string,
|
||||||
|
) (time.Duration, error) {
|
||||||
|
if configJSON == "" {
|
||||||
|
return 0, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
var cfg databaseTargetConfig
|
||||||
|
|
||||||
|
err := json.Unmarshal([]byte(configJSON), &cfg)
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf(
|
||||||
|
"parsing database target config: %w", err,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
if cfg.Expiry == "" || cfg.Expiry == archiveExpiryNever {
|
||||||
|
return 0, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
dur, err := time.ParseDuration(cfg.Expiry)
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf(
|
||||||
|
"parsing archive expiry %q: %w", cfg.Expiry, err,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
if dur <= 0 {
|
||||||
|
return 0, fmt.Errorf(
|
||||||
|
"%w: %q", errArchiveExpiryNotPositive, cfg.Expiry,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
return dur, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// ValidateArchiveExpiry checks a user-supplied archive expiry
|
||||||
|
// for a database target at configuration time. Valid values are
|
||||||
|
// empty, "never" (both meaning keep forever), or a positive Go
|
||||||
|
// duration such as "720h". Anything else is an error, so a bad
|
||||||
|
// expiry is rejected when the target is created rather than
|
||||||
|
// failing every subsequent delivery.
|
||||||
|
func ValidateArchiveExpiry(expiry string) error {
|
||||||
|
if expiry == "" || expiry == archiveExpiryNever {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
dur, err := time.ParseDuration(expiry)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf(
|
||||||
|
"expiry must be %q or a Go duration "+
|
||||||
|
"such as \"720h\": %w",
|
||||||
|
archiveExpiryNever, err,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
if dur <= 0 {
|
||||||
|
return fmt.Errorf(
|
||||||
|
"%w: %q", errArchiveExpiryNotPositive, expiry,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// archiveWriter owns one per-webhook archive SQLite file. It
|
||||||
|
// serialises writes, and after each write closes and reopens
|
||||||
|
// the file (debounced to at most once per debounce window) so
|
||||||
|
// an operator can move the file away for offline archiving. The
|
||||||
|
// next write recreates a moved or removed file, because the
|
||||||
|
// file is opened create-if-missing and its schema is migrated
|
||||||
|
// on every open.
|
||||||
|
type archiveWriter struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
path string
|
||||||
|
log *slog.Logger
|
||||||
|
debounce time.Duration
|
||||||
|
db *gorm.DB
|
||||||
|
lastReopen time.Time
|
||||||
|
reopens int
|
||||||
|
}
|
||||||
|
|
||||||
|
// newArchiveWriter builds an archiveWriter for a file path with
|
||||||
|
// the default reopen debounce.
|
||||||
|
func newArchiveWriter(
|
||||||
|
path string, log *slog.Logger,
|
||||||
|
) *archiveWriter {
|
||||||
|
return &archiveWriter{
|
||||||
|
path: path,
|
||||||
|
log: log,
|
||||||
|
debounce: archiveReopenDebounce,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// write appends the event as a row, then applies the debounced
|
||||||
|
// close/reopen. It recreates the archive file if it was moved
|
||||||
|
// or removed since the last open. A positive expiry prunes rows
|
||||||
|
// older than it on each (re)open.
|
||||||
|
func (w *archiveWriter) write(
|
||||||
|
row archivedEvent, expiry time.Duration,
|
||||||
|
) error {
|
||||||
|
w.mu.Lock()
|
||||||
|
defer w.mu.Unlock()
|
||||||
|
|
||||||
|
if w.db == nil || !fileExists(w.path) {
|
||||||
|
err := w.reopen(expiry)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
row.ArchivedAt = time.Now()
|
||||||
|
|
||||||
|
err := w.db.Create(&row).Error
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf(
|
||||||
|
"archiving event to %s: %w", w.path, err,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
if time.Since(w.lastReopen) >= w.debounce {
|
||||||
|
return w.reopen(expiry)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// open opens (creating if missing) the archive file, migrates
|
||||||
|
// its schema, records the reopen time, and prunes expired rows
|
||||||
|
// when expiry is positive.
|
||||||
|
func (w *archiveWriter) open(expiry time.Duration) error {
|
||||||
|
dbURL := fmt.Sprintf("file:%s?mode=rwc", w.path)
|
||||||
|
|
||||||
|
sqlDB, err := sql.Open("sqlite", dbURL)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf(
|
||||||
|
"opening archive database %s: %w", w.path, err,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
gdb, err := gorm.Open(
|
||||||
|
sqlite.Dialector{Conn: sqlDB}, &gorm.Config{},
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
_ = sqlDB.Close()
|
||||||
|
|
||||||
|
return fmt.Errorf(
|
||||||
|
"connecting to archive database %s: %w",
|
||||||
|
w.path, err,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = gdb.AutoMigrate(&archivedEvent{})
|
||||||
|
if err != nil {
|
||||||
|
_ = sqlDB.Close()
|
||||||
|
|
||||||
|
return fmt.Errorf(
|
||||||
|
"migrating archive database %s: %w", w.path, err,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
w.db = gdb
|
||||||
|
w.lastReopen = time.Now()
|
||||||
|
w.reopens++
|
||||||
|
|
||||||
|
if expiry > 0 {
|
||||||
|
w.prune(expiry)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// reopen closes any open handle and opens the file afresh. The
|
||||||
|
// fresh open recreates the file if it was moved away.
|
||||||
|
func (w *archiveWriter) reopen(expiry time.Duration) error {
|
||||||
|
w.close()
|
||||||
|
|
||||||
|
return w.open(expiry)
|
||||||
|
}
|
||||||
|
|
||||||
|
// close closes the underlying handle, if any.
|
||||||
|
func (w *archiveWriter) close() {
|
||||||
|
if w.db == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
sqlDB, err := w.db.DB()
|
||||||
|
if err == nil {
|
||||||
|
_ = sqlDB.Close()
|
||||||
|
}
|
||||||
|
|
||||||
|
w.db = nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// prune deletes archived rows older than expiry, measured from
|
||||||
|
// each row's archived time. It runs on every (re)open, and
|
||||||
|
// because the file is reopened after writes this keeps the
|
||||||
|
// archive swept without a separate background sweeper. Failures
|
||||||
|
// are logged, not fatal: a prune error must not stop archiving.
|
||||||
|
func (w *archiveWriter) prune(expiry time.Duration) {
|
||||||
|
cutoff := time.Now().Add(-expiry)
|
||||||
|
|
||||||
|
res := w.db.Where("archived_at < ?", cutoff).
|
||||||
|
Delete(&archivedEvent{})
|
||||||
|
if res.Error != nil {
|
||||||
|
w.log.Error(
|
||||||
|
"failed to prune expired archive rows",
|
||||||
|
"path", w.path,
|
||||||
|
"error", res.Error,
|
||||||
|
)
|
||||||
|
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if res.RowsAffected > 0 {
|
||||||
|
w.log.Info(
|
||||||
|
"pruned expired archive rows",
|
||||||
|
"path", w.path,
|
||||||
|
"rows_deleted", res.RowsAffected,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// fileExists reports whether a path currently exists.
|
||||||
|
func fileExists(path string) bool {
|
||||||
|
_, err := os.Stat(path)
|
||||||
|
|
||||||
|
return err == nil
|
||||||
|
}
|
||||||
395
internal/delivery/target_database_test.go
Normal file
395
internal/delivery/target_database_test.go
Normal file
@@ -0,0 +1,395 @@
|
|||||||
|
package delivery_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"database/sql"
|
||||||
|
"fmt"
|
||||||
|
"log/slog"
|
||||||
|
"net/http"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/google/uuid"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"gorm.io/driver/sqlite"
|
||||||
|
"gorm.io/gorm"
|
||||||
|
_ "modernc.org/sqlite" // Pure Go SQLite driver.
|
||||||
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
|
"sneak.berlin/go/webhooker/internal/delivery"
|
||||||
|
)
|
||||||
|
|
||||||
|
func archiveTestLogger() *slog.Logger {
|
||||||
|
return slog.New(slog.NewTextHandler(
|
||||||
|
os.Stderr,
|
||||||
|
&slog.HandlerOptions{Level: slog.LevelDebug},
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
// openArchiveDBForRead opens an archive file read-only so a
|
||||||
|
// test can inspect the rows the writer persisted.
|
||||||
|
func openArchiveDBForRead(
|
||||||
|
t *testing.T, path string,
|
||||||
|
) *gorm.DB {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
sqlDB, err := sql.Open(
|
||||||
|
"sqlite",
|
||||||
|
fmt.Sprintf("file:%s?mode=ro", path),
|
||||||
|
)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
t.Cleanup(func() { _ = sqlDB.Close() })
|
||||||
|
|
||||||
|
gdb, err := gorm.Open(
|
||||||
|
sqlite.Dialector{Conn: sqlDB}, &gorm.Config{},
|
||||||
|
)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
return gdb
|
||||||
|
}
|
||||||
|
|
||||||
|
// removeArchiveFiles simulates an operator moving the archive
|
||||||
|
// away by deleting the SQLite file and its sidecar files.
|
||||||
|
func removeArchiveFiles(t *testing.T, path string) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
for _, suffix := range []string{
|
||||||
|
"", "-wal", "-shm", "-journal",
|
||||||
|
} {
|
||||||
|
err := os.Remove(path + suffix)
|
||||||
|
if err != nil && !os.IsNotExist(err) {
|
||||||
|
t.Fatalf("removing %s%s: %v", path, suffix, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestDeliverDatabase_ArchivesEvent verifies that delivering to
|
||||||
|
// a database target marks the delivery delivered and archives
|
||||||
|
// the full event into a separate per-webhook archive file.
|
||||||
|
func TestDeliverDatabase_ArchivesEvent(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
dataDir := t.TempDir()
|
||||||
|
dbMgr := database.NewTestWebhookDBManager(dataDir)
|
||||||
|
|
||||||
|
e := delivery.NewTestEngineWithDB(
|
||||||
|
nil, dbMgr,
|
||||||
|
archiveTestLogger(),
|
||||||
|
&http.Client{Timeout: 5 * time.Second},
|
||||||
|
1,
|
||||||
|
)
|
||||||
|
|
||||||
|
webhookDB := testWebhookDB(t)
|
||||||
|
event := seedEvent(t, webhookDB, `{"archived":true}`)
|
||||||
|
d := seedDatabaseTargetDelivery(t, webhookDB, event, "")
|
||||||
|
|
||||||
|
e.ExportDeliverDatabase(webhookDB, d)
|
||||||
|
|
||||||
|
var updated database.Delivery
|
||||||
|
|
||||||
|
require.NoError(t, webhookDB.First(
|
||||||
|
&updated, "id = ?", d.ID,
|
||||||
|
).Error)
|
||||||
|
assert.Equal(t,
|
||||||
|
database.DeliveryStatusDelivered, updated.Status,
|
||||||
|
"database target should mark the delivery delivered",
|
||||||
|
)
|
||||||
|
|
||||||
|
archivePath := filepath.Join(
|
||||||
|
dataDir,
|
||||||
|
fmt.Sprintf("archive-%s.db", event.WebhookID),
|
||||||
|
)
|
||||||
|
assert.FileExists(t, archivePath)
|
||||||
|
|
||||||
|
rdb := openArchiveDBForRead(t, archivePath)
|
||||||
|
|
||||||
|
var rows []delivery.ExportArchivedEvent
|
||||||
|
|
||||||
|
require.NoError(t, rdb.Find(&rows).Error)
|
||||||
|
require.Len(t, rows, 1)
|
||||||
|
assert.Equal(t, event.ID, rows[0].EventID)
|
||||||
|
assert.Equal(t, event.WebhookID, rows[0].WebhookID)
|
||||||
|
assert.Equal(t, event.Method, rows[0].Method)
|
||||||
|
assert.JSONEq(t, `{"archived":true}`, rows[0].Body)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArchiveWriter_WritesRow(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
path := filepath.Join(t.TempDir(), "archive-wh.db")
|
||||||
|
w := delivery.NewExportArchiveWriter(
|
||||||
|
path, archiveTestLogger(), 0,
|
||||||
|
)
|
||||||
|
|
||||||
|
row := delivery.ExportArchivedEvent{
|
||||||
|
EventID: "ev-1",
|
||||||
|
WebhookID: "wh-1",
|
||||||
|
EntrypointID: "ep-1",
|
||||||
|
Method: "POST",
|
||||||
|
Headers: `{"X":"Y"}`,
|
||||||
|
Body: `{"hello":"world"}`,
|
||||||
|
ContentType: "application/json",
|
||||||
|
}
|
||||||
|
|
||||||
|
require.NoError(t, w.Write(row, 0))
|
||||||
|
assert.FileExists(t, path)
|
||||||
|
|
||||||
|
var got []delivery.ExportArchivedEvent
|
||||||
|
|
||||||
|
require.NoError(t, w.DB().Find(&got).Error)
|
||||||
|
require.Len(t, got, 1)
|
||||||
|
assert.Equal(t, "ev-1", got[0].EventID)
|
||||||
|
assert.Equal(t, "wh-1", got[0].WebhookID)
|
||||||
|
assert.Equal(t, "ep-1", got[0].EntrypointID)
|
||||||
|
assert.Equal(t, row.Method, got[0].Method)
|
||||||
|
assert.Equal(t, row.ContentType, got[0].ContentType)
|
||||||
|
assert.JSONEq(t, `{"hello":"world"}`, got[0].Body)
|
||||||
|
assert.False(t, got[0].ArchivedAt.IsZero())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArchiveWriter_RecreatesAfterRemoval(
|
||||||
|
t *testing.T,
|
||||||
|
) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
path := filepath.Join(t.TempDir(), "archive-wh.db")
|
||||||
|
w := delivery.NewExportArchiveWriter(
|
||||||
|
path, archiveTestLogger(), 0,
|
||||||
|
)
|
||||||
|
|
||||||
|
require.NoError(t, w.Write(
|
||||||
|
delivery.ExportArchivedEvent{EventID: "a"}, 0,
|
||||||
|
))
|
||||||
|
assert.FileExists(t, path)
|
||||||
|
|
||||||
|
// The operator moves the archive away while the handle is
|
||||||
|
// still open.
|
||||||
|
removeArchiveFiles(t, path)
|
||||||
|
require.NoFileExists(t, path)
|
||||||
|
|
||||||
|
// The next write recreates the file with a fresh schema and
|
||||||
|
// only the new row.
|
||||||
|
require.NoError(t, w.Write(
|
||||||
|
delivery.ExportArchivedEvent{EventID: "b"}, 0,
|
||||||
|
))
|
||||||
|
assert.FileExists(t, path)
|
||||||
|
|
||||||
|
var got []delivery.ExportArchivedEvent
|
||||||
|
|
||||||
|
require.NoError(t, w.DB().Find(&got).Error)
|
||||||
|
require.Len(t, got, 1)
|
||||||
|
assert.Equal(t, "b", got[0].EventID)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArchiveWriter_ReopenDebounce(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
// A generous debounce keeps the two rapid writes inside
|
||||||
|
// the window even on a heavily loaded test machine.
|
||||||
|
path := filepath.Join(t.TempDir(), "archive-wh.db")
|
||||||
|
w := delivery.NewExportArchiveWriter(
|
||||||
|
path, archiveTestLogger(), 2*time.Second,
|
||||||
|
)
|
||||||
|
|
||||||
|
require.NoError(t, w.Write(
|
||||||
|
delivery.ExportArchivedEvent{EventID: "a"}, 0,
|
||||||
|
))
|
||||||
|
require.NoError(t, w.Write(
|
||||||
|
delivery.ExportArchivedEvent{EventID: "b"}, 0,
|
||||||
|
))
|
||||||
|
|
||||||
|
// Two writes inside the debounce window trigger only the
|
||||||
|
// initial open — no extra close/reopen.
|
||||||
|
assert.Equal(t, 1, w.Reopens())
|
||||||
|
|
||||||
|
time.Sleep(2100 * time.Millisecond)
|
||||||
|
|
||||||
|
require.NoError(t, w.Write(
|
||||||
|
delivery.ExportArchivedEvent{EventID: "c"}, 0,
|
||||||
|
))
|
||||||
|
|
||||||
|
// A write after the window elapses closes and reopens once.
|
||||||
|
assert.Equal(t, 2, w.Reopens())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArchiveWriter_ExpiryPrune(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
path := filepath.Join(t.TempDir(), "archive-wh.db")
|
||||||
|
w := delivery.NewExportArchiveWriter(
|
||||||
|
path, archiveTestLogger(), 0,
|
||||||
|
)
|
||||||
|
|
||||||
|
require.NoError(t, w.Open(0))
|
||||||
|
|
||||||
|
old := delivery.ExportArchivedEvent{
|
||||||
|
EventID: "old",
|
||||||
|
ArchivedAt: time.Now().Add(-2 * time.Hour),
|
||||||
|
}
|
||||||
|
fresh := delivery.ExportArchivedEvent{
|
||||||
|
EventID: "fresh",
|
||||||
|
ArchivedAt: time.Now(),
|
||||||
|
}
|
||||||
|
|
||||||
|
require.NoError(t, w.DB().Create(&old).Error)
|
||||||
|
require.NoError(t, w.DB().Create(&fresh).Error)
|
||||||
|
|
||||||
|
// Reopening with a one-hour expiry prunes the old row.
|
||||||
|
require.NoError(t, w.Reopen(time.Hour))
|
||||||
|
|
||||||
|
var got []delivery.ExportArchivedEvent
|
||||||
|
|
||||||
|
require.NoError(t, w.DB().Find(&got).Error)
|
||||||
|
require.Len(t, got, 1)
|
||||||
|
assert.Equal(t, "fresh", got[0].EventID)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestParseArchiveExpiry(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
in string
|
||||||
|
want time.Duration
|
||||||
|
wantErr bool
|
||||||
|
}{
|
||||||
|
{"empty config", "", 0, false},
|
||||||
|
{"explicit never", `{"expiry":"never"}`, 0, false},
|
||||||
|
{"empty expiry", `{"expiry":""}`, 0, false},
|
||||||
|
{"duration", `{"expiry":"1h"}`, time.Hour, false},
|
||||||
|
{"unparseable", `{"expiry":"nonsense"}`, 0, true},
|
||||||
|
{"zero duration", `{"expiry":"0s"}`, 0, true},
|
||||||
|
{"negative duration", `{"expiry":"-5h"}`, 0, true},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
got, err := delivery.ExportParseArchiveExpiry(tc.in)
|
||||||
|
if tc.wantErr {
|
||||||
|
require.Error(t, err)
|
||||||
|
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, tc.want, got)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// seedDatabaseTargetDelivery seeds a pending delivery for a
|
||||||
|
// database target with the given config JSON and returns the
|
||||||
|
// in-memory delivery the target handler is invoked with.
|
||||||
|
func seedDatabaseTargetDelivery(
|
||||||
|
t *testing.T,
|
||||||
|
webhookDB *gorm.DB,
|
||||||
|
event database.Event,
|
||||||
|
config string,
|
||||||
|
) *database.Delivery {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
dlv := seedDelivery(
|
||||||
|
t, webhookDB, event.ID, uuid.New().String(),
|
||||||
|
database.DeliveryStatusPending,
|
||||||
|
)
|
||||||
|
|
||||||
|
d := &database.Delivery{
|
||||||
|
EventID: event.ID,
|
||||||
|
TargetID: dlv.TargetID,
|
||||||
|
Status: database.DeliveryStatusPending,
|
||||||
|
Event: event,
|
||||||
|
Target: database.Target{
|
||||||
|
Name: "test-db",
|
||||||
|
Type: database.TargetTypeDatabase,
|
||||||
|
Config: config,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
d.ID = dlv.ID
|
||||||
|
|
||||||
|
return d
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestDeliverDatabase_ArchiveFailureFailsDelivery verifies that
|
||||||
|
// an archive error (here: an unparseable expiry in the target
|
||||||
|
// config) fails the delivery loudly: the attempt is recorded as
|
||||||
|
// failed with the error and the delivery is marked failed, not
|
||||||
|
// delivered.
|
||||||
|
func TestDeliverDatabase_ArchiveFailureFailsDelivery(
|
||||||
|
t *testing.T,
|
||||||
|
) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
dataDir := t.TempDir()
|
||||||
|
|
||||||
|
e := delivery.NewTestEngineWithDB(
|
||||||
|
nil, database.NewTestWebhookDBManager(dataDir),
|
||||||
|
archiveTestLogger(),
|
||||||
|
&http.Client{Timeout: 5 * time.Second},
|
||||||
|
1,
|
||||||
|
)
|
||||||
|
|
||||||
|
webhookDB := testWebhookDB(t)
|
||||||
|
event := seedEvent(t, webhookDB, `{"archived":false}`)
|
||||||
|
d := seedDatabaseTargetDelivery(
|
||||||
|
t, webhookDB, event, `{"expiry":"nonsense"}`,
|
||||||
|
)
|
||||||
|
|
||||||
|
e.ExportDeliverDatabase(webhookDB, d)
|
||||||
|
|
||||||
|
var updated database.Delivery
|
||||||
|
|
||||||
|
require.NoError(t, webhookDB.First(
|
||||||
|
&updated, "id = ?", d.ID,
|
||||||
|
).Error)
|
||||||
|
assert.Equal(t,
|
||||||
|
database.DeliveryStatusFailed, updated.Status,
|
||||||
|
"archive failure must mark the delivery failed",
|
||||||
|
)
|
||||||
|
|
||||||
|
var results []database.DeliveryResult
|
||||||
|
|
||||||
|
require.NoError(t, webhookDB.Where(
|
||||||
|
"delivery_id = ?", d.ID,
|
||||||
|
).Find(&results).Error)
|
||||||
|
require.Len(t, results, 1)
|
||||||
|
assert.False(t,
|
||||||
|
results[0].Success,
|
||||||
|
"the attempt must be recorded as failed",
|
||||||
|
)
|
||||||
|
assert.Contains(t,
|
||||||
|
results[0].Error, "nonsense",
|
||||||
|
"the archive error must be recorded on the attempt",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert.NoFileExists(t,
|
||||||
|
filepath.Join(
|
||||||
|
dataDir,
|
||||||
|
fmt.Sprintf("archive-%s.db", event.WebhookID),
|
||||||
|
),
|
||||||
|
"no archive file should exist for a failed config",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestValidateArchiveExpiry(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
valid := []string{"", "never", "1h", "720h", "30m"}
|
||||||
|
for _, in := range valid {
|
||||||
|
require.NoError(t,
|
||||||
|
delivery.ValidateArchiveExpiry(in),
|
||||||
|
"expiry %q should be accepted", in,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
invalid := []string{"nonsense", "7d", "-5h", "0s", "0"}
|
||||||
|
for _, in := range invalid {
|
||||||
|
require.Error(t,
|
||||||
|
delivery.ValidateArchiveExpiry(in),
|
||||||
|
"expiry %q should be rejected", in,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -26,3 +26,13 @@ func (s *Handlers) BuildSlackTargetConfigForTest(
|
|||||||
"Webhook URL is required for Slack targets",
|
"Webhook URL is required for Slack targets",
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// BuildDatabaseTargetConfigForTest exposes
|
||||||
|
// buildDatabaseTargetConfig for use in the handlers_test
|
||||||
|
// package.
|
||||||
|
func (s *Handlers) BuildDatabaseTargetConfigForTest(
|
||||||
|
w http.ResponseWriter,
|
||||||
|
expiry string,
|
||||||
|
) (string, error) {
|
||||||
|
return s.buildDatabaseTargetConfig(w, expiry)
|
||||||
|
}
|
||||||
|
|||||||
@@ -186,3 +186,57 @@ func TestRenderTemplate(t *testing.T) {
|
|||||||
t, http.StatusInternalServerError, w.Code,
|
t, http.StatusInternalServerError, w.Code,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestBuildDatabaseTargetConfig_Valid(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
var h *handlers.Handlers
|
||||||
|
|
||||||
|
app := newTestApp(t, &h)
|
||||||
|
app.RequireStart()
|
||||||
|
|
||||||
|
t.Cleanup(app.RequireStop)
|
||||||
|
|
||||||
|
// Empty expiry: the keep-forever default, empty config.
|
||||||
|
w := httptest.NewRecorder()
|
||||||
|
cfg, err := h.BuildDatabaseTargetConfigForTest(w, "")
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Empty(t, cfg)
|
||||||
|
|
||||||
|
// Explicit never is stored as config.
|
||||||
|
w = httptest.NewRecorder()
|
||||||
|
cfg, err = h.BuildDatabaseTargetConfigForTest(w, "never")
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.JSONEq(t, `{"expiry":"never"}`, cfg)
|
||||||
|
|
||||||
|
// A positive duration is stored as config.
|
||||||
|
w = httptest.NewRecorder()
|
||||||
|
cfg, err = h.BuildDatabaseTargetConfigForTest(w, "720h")
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.JSONEq(t, `{"expiry":"720h"}`, cfg)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBuildDatabaseTargetConfig_RejectsBadExpiry(
|
||||||
|
t *testing.T,
|
||||||
|
) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
var h *handlers.Handlers
|
||||||
|
|
||||||
|
app := newTestApp(t, &h)
|
||||||
|
app.RequireStart()
|
||||||
|
|
||||||
|
t.Cleanup(app.RequireStop)
|
||||||
|
|
||||||
|
for _, bad := range []string{"nonsense", "7d", "-5h"} {
|
||||||
|
w := httptest.NewRecorder()
|
||||||
|
cfg, err := h.BuildDatabaseTargetConfigForTest(w, bad)
|
||||||
|
|
||||||
|
require.Error(t, err, "expiry %q", bad)
|
||||||
|
assert.Empty(t, cfg)
|
||||||
|
assert.Equal(
|
||||||
|
t, http.StatusBadRequest, w.Code,
|
||||||
|
"expiry %q should be rejected with 400", bad,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -4,63 +4,202 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
|
|
||||||
"github.com/go-chi/chi"
|
"github.com/go-chi/chi"
|
||||||
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
)
|
)
|
||||||
|
|
||||||
// HandleProfile returns a handler for the user profile page
|
// HandleProfile returns a handler for the user profile page
|
||||||
func (h *Handlers) HandleProfile() http.HandlerFunc {
|
func (h *Handlers) HandleProfile() http.HandlerFunc {
|
||||||
return func(w http.ResponseWriter, r *http.Request) {
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
// Get username from URL
|
sessionUserID, sessionUsername, ok :=
|
||||||
|
h.profileOwnerOrDeny(w, r)
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
h.renderProfile(w, r, sessionUserID, sessionUsername, "", "")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// HandlePasswordChange returns a handler that lets an authenticated
|
||||||
|
// user change their own password. It is served by the CSRF- and
|
||||||
|
// auth-protected POST /password route under /user/{username}.
|
||||||
|
func (h *Handlers) HandlePasswordChange() http.HandlerFunc {
|
||||||
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
sessionUserID, sessionUsername, ok :=
|
||||||
|
h.profileOwnerOrDeny(w, r)
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Limit request body to prevent memory exhaustion.
|
||||||
|
r.Body = http.MaxBytesReader(w, r.Body, 1<<maxBodyShift)
|
||||||
|
|
||||||
|
err := r.ParseForm()
|
||||||
|
if err != nil {
|
||||||
|
h.log.Error("failed to parse form", "error", err)
|
||||||
|
http.Error(w, "Bad request", http.StatusBadRequest)
|
||||||
|
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
successMessage, errorMessage, handled := h.applyPasswordChange(
|
||||||
|
w,
|
||||||
|
sessionUsername,
|
||||||
|
r.FormValue("current_password"),
|
||||||
|
r.FormValue("new_password"),
|
||||||
|
r.FormValue("confirm_password"),
|
||||||
|
)
|
||||||
|
if !handled {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
h.renderProfile(
|
||||||
|
w, r, sessionUserID, sessionUsername,
|
||||||
|
successMessage, errorMessage,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// applyPasswordChange verifies the current password and, on success,
|
||||||
|
// persists a fresh hash for the user, reusing the same helpers that
|
||||||
|
// bootstrap the admin user. It returns the success and error messages
|
||||||
|
// to display on the profile page. On an internal failure it writes a
|
||||||
|
// 500 response itself and returns handled=false, signalling the caller
|
||||||
|
// to stop without re-rendering the page.
|
||||||
|
func (h *Handlers) applyPasswordChange(
|
||||||
|
w http.ResponseWriter,
|
||||||
|
username, currentPassword, newPassword, confirmPassword string,
|
||||||
|
) (string, string, bool) {
|
||||||
|
// Load the user row so we can verify the current password and
|
||||||
|
// persist the new hash.
|
||||||
|
var user database.User
|
||||||
|
|
||||||
|
err := h.db.DB().Where(
|
||||||
|
"username = ?", username,
|
||||||
|
).First(&user).Error
|
||||||
|
if err != nil {
|
||||||
|
h.serverError(
|
||||||
|
w, "failed to load user for password change", err,
|
||||||
|
)
|
||||||
|
|
||||||
|
return "", "", false
|
||||||
|
}
|
||||||
|
|
||||||
|
valid, err := database.VerifyPassword(
|
||||||
|
currentPassword, user.Password,
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
h.serverError(w, "failed to verify password", err)
|
||||||
|
|
||||||
|
return "", "", false
|
||||||
|
}
|
||||||
|
|
||||||
|
if !valid {
|
||||||
|
return "", "Current password is incorrect.", true
|
||||||
|
}
|
||||||
|
|
||||||
|
if newPassword == "" {
|
||||||
|
return "", "New password must not be empty.", true
|
||||||
|
}
|
||||||
|
|
||||||
|
if newPassword != confirmPassword {
|
||||||
|
return "", "New password and confirmation do not match.", true
|
||||||
|
}
|
||||||
|
|
||||||
|
hashedPassword, err := database.HashPassword(newPassword)
|
||||||
|
if err != nil {
|
||||||
|
h.serverError(w, "failed to hash new password", err)
|
||||||
|
|
||||||
|
return "", "", false
|
||||||
|
}
|
||||||
|
|
||||||
|
err = h.db.DB().Model(&user).Update(
|
||||||
|
"password", hashedPassword,
|
||||||
|
).Error
|
||||||
|
if err != nil {
|
||||||
|
h.serverError(w, "failed to update password", err)
|
||||||
|
|
||||||
|
return "", "", false
|
||||||
|
}
|
||||||
|
|
||||||
|
h.log.Info("user changed password", "username", username)
|
||||||
|
|
||||||
|
return "Password changed successfully.", "", true
|
||||||
|
}
|
||||||
|
|
||||||
|
// profileOwnerOrDeny resolves the session identity and enforces that a
|
||||||
|
// user may only act on their own profile (the requested username in the
|
||||||
|
// URL must equal the session username). On any failure it writes the
|
||||||
|
// appropriate HTTP response and returns ok=false; callers must stop
|
||||||
|
// when ok is false.
|
||||||
|
func (h *Handlers) profileOwnerOrDeny(
|
||||||
|
w http.ResponseWriter,
|
||||||
|
r *http.Request,
|
||||||
|
) (string, string, bool) {
|
||||||
requestedUsername := chi.URLParam(r, "username")
|
requestedUsername := chi.URLParam(r, "username")
|
||||||
if requestedUsername == "" {
|
if requestedUsername == "" {
|
||||||
http.NotFound(w, r)
|
http.NotFound(w, r)
|
||||||
|
|
||||||
return
|
return "", "", false
|
||||||
}
|
}
|
||||||
|
|
||||||
// Get session. RequireAuth middleware guarantees an
|
// RequireAuth middleware guarantees an authenticated session
|
||||||
// authenticated session before this handler runs, so we
|
// before this handler runs, so we only need to guard against an
|
||||||
// only need to guard against an unexpected retrieval error.
|
// unexpected retrieval error.
|
||||||
sess, err := h.session.Get(r)
|
sess, err := h.session.Get(r)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
h.log.Error("failed to get session", "error", err)
|
h.serverError(w, "failed to get session", err)
|
||||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
|
||||||
|
|
||||||
return
|
return "", "", false
|
||||||
}
|
}
|
||||||
|
|
||||||
// Get user info from session
|
|
||||||
sessionUsername, ok := h.session.GetUsername(sess)
|
sessionUsername, ok := h.session.GetUsername(sess)
|
||||||
if !ok {
|
if !ok {
|
||||||
h.log.Error("authenticated session missing username")
|
h.log.Error("authenticated session missing username")
|
||||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
http.Error(
|
||||||
|
w, "Internal server error",
|
||||||
|
http.StatusInternalServerError,
|
||||||
|
)
|
||||||
|
|
||||||
return
|
return "", "", false
|
||||||
}
|
}
|
||||||
|
|
||||||
sessionUserID, ok := h.session.GetUserID(sess)
|
sessionUserID, ok := h.session.GetUserID(sess)
|
||||||
if !ok {
|
if !ok {
|
||||||
h.log.Error("authenticated session missing user ID")
|
h.log.Error("authenticated session missing user ID")
|
||||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
http.Error(
|
||||||
|
w, "Internal server error",
|
||||||
|
http.StatusInternalServerError,
|
||||||
|
)
|
||||||
|
|
||||||
return
|
return "", "", false
|
||||||
}
|
}
|
||||||
|
|
||||||
// For now, only allow users to view their own profile
|
// Only allow users to act on their own profile.
|
||||||
if requestedUsername != sessionUsername {
|
if requestedUsername != sessionUsername {
|
||||||
http.Error(w, "Forbidden", http.StatusForbidden)
|
http.Error(w, "Forbidden", http.StatusForbidden)
|
||||||
|
|
||||||
return
|
return "", "", false
|
||||||
}
|
}
|
||||||
|
|
||||||
// Prepare data for template
|
return sessionUserID, sessionUsername, true
|
||||||
|
}
|
||||||
|
|
||||||
|
// renderProfile renders the profile page for the given user,
|
||||||
|
// optionally including a success or error message.
|
||||||
|
func (h *Handlers) renderProfile(
|
||||||
|
w http.ResponseWriter,
|
||||||
|
r *http.Request,
|
||||||
|
userID, username, successMessage, errorMessage string,
|
||||||
|
) {
|
||||||
data := map[string]any{
|
data := map[string]any{
|
||||||
"User": &UserInfo{
|
"User": &UserInfo{
|
||||||
ID: sessionUserID,
|
ID: userID,
|
||||||
Username: sessionUsername,
|
Username: username,
|
||||||
},
|
},
|
||||||
|
"SuccessMessage": successMessage,
|
||||||
|
"ErrorMessage": errorMessage,
|
||||||
}
|
}
|
||||||
|
|
||||||
// Render the profile page
|
|
||||||
h.renderTemplate(w, r, "profile.html", data)
|
h.renderTemplate(w, r, "profile.html", data)
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|||||||
@@ -4,12 +4,15 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
|
"net/url"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/go-chi/chi"
|
"github.com/go-chi/chi"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
"sneak.berlin/go/webhooker/internal/config"
|
"sneak.berlin/go/webhooker/internal/config"
|
||||||
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
"sneak.berlin/go/webhooker/internal/handlers"
|
"sneak.berlin/go/webhooker/internal/handlers"
|
||||||
"sneak.berlin/go/webhooker/internal/logger"
|
"sneak.berlin/go/webhooker/internal/logger"
|
||||||
"sneak.berlin/go/webhooker/internal/middleware"
|
"sneak.berlin/go/webhooker/internal/middleware"
|
||||||
@@ -157,3 +160,134 @@ func TestUserRoute_Unauthenticated_RedirectedByMiddleware(t *testing.T) {
|
|||||||
assert.Equal(t, http.StatusSeeOther, w.Code)
|
assert.Equal(t, http.StatusSeeOther, w.Code)
|
||||||
assert.Equal(t, "/pages/login", w.Header().Get("Location"))
|
assert.Equal(t, "/pages/login", w.Header().Get("Location"))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// passwordChangeRequest builds a POST request to the password-change
|
||||||
|
// endpoint for the given username, attaching the supplied cookies, an
|
||||||
|
// urlencoded form body, and the chi URL parameter the handler reads.
|
||||||
|
func passwordChangeRequest(
|
||||||
|
username string,
|
||||||
|
cookies []*http.Cookie,
|
||||||
|
form url.Values,
|
||||||
|
) *http.Request {
|
||||||
|
req := httptest.NewRequestWithContext(
|
||||||
|
context.Background(),
|
||||||
|
http.MethodPost,
|
||||||
|
"/user/"+username+"/password",
|
||||||
|
strings.NewReader(form.Encode()),
|
||||||
|
)
|
||||||
|
req.Header.Set(
|
||||||
|
"Content-Type", "application/x-www-form-urlencoded",
|
||||||
|
)
|
||||||
|
|
||||||
|
for _, c := range cookies {
|
||||||
|
req.AddCookie(c)
|
||||||
|
}
|
||||||
|
|
||||||
|
rctx := chi.NewRouteContext()
|
||||||
|
rctx.URLParams.Add("username", username)
|
||||||
|
|
||||||
|
return req.WithContext(
|
||||||
|
context.WithValue(req.Context(), chi.RouteCtxKey, rctx),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHandlePasswordChange_Success(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
var h *handlers.Handlers
|
||||||
|
|
||||||
|
var sess *session.Session
|
||||||
|
|
||||||
|
var db *database.Database
|
||||||
|
|
||||||
|
app := newTestApp(t, &h, &sess, &db)
|
||||||
|
app.RequireStart()
|
||||||
|
|
||||||
|
t.Cleanup(app.RequireStop)
|
||||||
|
|
||||||
|
oldHash, err := database.HashPassword("oldpassword")
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
user := &database.User{Username: "pwuser", Password: oldHash}
|
||||||
|
require.NoError(t, db.DB().Create(user).Error)
|
||||||
|
|
||||||
|
cookies := authenticatedCookies(t, sess, user.ID, "pwuser")
|
||||||
|
|
||||||
|
form := url.Values{}
|
||||||
|
form.Set("current_password", "oldpassword")
|
||||||
|
form.Set("new_password", "newpassword")
|
||||||
|
form.Set("confirm_password", "newpassword")
|
||||||
|
|
||||||
|
req := passwordChangeRequest("pwuser", cookies, form)
|
||||||
|
w := httptest.NewRecorder()
|
||||||
|
|
||||||
|
h.HandlePasswordChange().ServeHTTP(w, req)
|
||||||
|
|
||||||
|
assert.Equal(t, http.StatusOK, w.Code)
|
||||||
|
assert.Contains(
|
||||||
|
t, w.Body.String(), "Password changed successfully.",
|
||||||
|
)
|
||||||
|
|
||||||
|
var updated database.User
|
||||||
|
|
||||||
|
require.NoError(t,
|
||||||
|
db.DB().Where("username = ?", "pwuser").First(&updated).Error,
|
||||||
|
)
|
||||||
|
assert.NotEqual(t, oldHash, updated.Password)
|
||||||
|
|
||||||
|
valid, err := database.VerifyPassword(
|
||||||
|
"newpassword", updated.Password,
|
||||||
|
)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.True(t, valid, "new password should verify against new hash")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHandlePasswordChange_WrongCurrentPassword(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
var h *handlers.Handlers
|
||||||
|
|
||||||
|
var sess *session.Session
|
||||||
|
|
||||||
|
var db *database.Database
|
||||||
|
|
||||||
|
app := newTestApp(t, &h, &sess, &db)
|
||||||
|
app.RequireStart()
|
||||||
|
|
||||||
|
t.Cleanup(app.RequireStop)
|
||||||
|
|
||||||
|
oldHash, err := database.HashPassword("oldpassword")
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
user := &database.User{Username: "pwuser2", Password: oldHash}
|
||||||
|
require.NoError(t, db.DB().Create(user).Error)
|
||||||
|
|
||||||
|
cookies := authenticatedCookies(t, sess, user.ID, "pwuser2")
|
||||||
|
|
||||||
|
form := url.Values{}
|
||||||
|
form.Set("current_password", "wrongpassword")
|
||||||
|
form.Set("new_password", "newpassword")
|
||||||
|
form.Set("confirm_password", "newpassword")
|
||||||
|
|
||||||
|
req := passwordChangeRequest("pwuser2", cookies, form)
|
||||||
|
w := httptest.NewRecorder()
|
||||||
|
|
||||||
|
h.HandlePasswordChange().ServeHTTP(w, req)
|
||||||
|
|
||||||
|
assert.Equal(t, http.StatusOK, w.Code)
|
||||||
|
assert.Contains(
|
||||||
|
t, w.Body.String(), "Current password is incorrect.",
|
||||||
|
)
|
||||||
|
|
||||||
|
var unchanged database.User
|
||||||
|
|
||||||
|
require.NoError(t,
|
||||||
|
db.DB().Where(
|
||||||
|
"username = ?", "pwuser2",
|
||||||
|
).First(&unchanged).Error,
|
||||||
|
)
|
||||||
|
assert.Equal(
|
||||||
|
t, oldHash, unchanged.Password,
|
||||||
|
"stored hash must be unchanged after a rejected change",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
|
||||||
"github.com/go-chi/chi"
|
"github.com/go-chi/chi"
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
@@ -815,6 +816,7 @@ func (h *Handlers) processTargetCreate(
|
|||||||
targetType := database.TargetType(r.FormValue("type"))
|
targetType := database.TargetType(r.FormValue("type"))
|
||||||
targetURL := r.FormValue("url")
|
targetURL := r.FormValue("url")
|
||||||
maxRetriesStr := r.FormValue("max_retries")
|
maxRetriesStr := r.FormValue("max_retries")
|
||||||
|
expiry := r.FormValue("expiry")
|
||||||
|
|
||||||
if name == "" {
|
if name == "" {
|
||||||
http.Error(
|
http.Error(
|
||||||
@@ -834,7 +836,7 @@ func (h *Handlers) processTargetCreate(
|
|||||||
}
|
}
|
||||||
|
|
||||||
configJSON, err := h.buildTargetConfig(
|
configJSON, err := h.buildTargetConfig(
|
||||||
w, r, targetType, targetURL,
|
w, r, targetType, targetURL, expiry,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return
|
||||||
@@ -892,11 +894,13 @@ func parseNonNegativeInt(s string) int {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// buildTargetConfig builds the JSON config string for a target.
|
// buildTargetConfig builds the JSON config string for a target.
|
||||||
|
// The expiry form value is read by the caller (which bounds the
|
||||||
|
// request body) and applies to database targets only.
|
||||||
func (h *Handlers) buildTargetConfig(
|
func (h *Handlers) buildTargetConfig(
|
||||||
w http.ResponseWriter,
|
w http.ResponseWriter,
|
||||||
r *http.Request,
|
r *http.Request,
|
||||||
targetType database.TargetType,
|
targetType database.TargetType,
|
||||||
targetURL string,
|
targetURL, expiry string,
|
||||||
) (string, error) {
|
) (string, error) {
|
||||||
switch targetType {
|
switch targetType {
|
||||||
case database.TargetTypeHTTP:
|
case database.TargetTypeHTTP:
|
||||||
@@ -909,7 +913,9 @@ func (h *Handlers) buildTargetConfig(
|
|||||||
w, r, targetURL, "webhookUrl",
|
w, r, targetURL, "webhookUrl",
|
||||||
"Webhook URL is required for Slack targets",
|
"Webhook URL is required for Slack targets",
|
||||||
)
|
)
|
||||||
case database.TargetTypeDatabase, database.TargetTypeLog:
|
case database.TargetTypeDatabase:
|
||||||
|
return h.buildDatabaseTargetConfig(w, expiry)
|
||||||
|
case database.TargetTypeLog:
|
||||||
return "", nil
|
return "", nil
|
||||||
default:
|
default:
|
||||||
http.Error(
|
http.Error(
|
||||||
@@ -972,6 +978,47 @@ func (h *Handlers) buildURLTargetConfig(
|
|||||||
return string(configBytes), nil
|
return string(configBytes), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// buildDatabaseTargetConfig builds config JSON for a database
|
||||||
|
// (archive) target. The optional expiry (a form value read by
|
||||||
|
// the caller, which bounds the request body) is validated here,
|
||||||
|
// at creation time, so an unparseable value is rejected with a
|
||||||
|
// 400 instead of failing every subsequent delivery. An empty
|
||||||
|
// expiry yields an empty config (the keep-forever default).
|
||||||
|
func (h *Handlers) buildDatabaseTargetConfig(
|
||||||
|
w http.ResponseWriter,
|
||||||
|
expiry string,
|
||||||
|
) (string, error) {
|
||||||
|
expiry = strings.TrimSpace(expiry)
|
||||||
|
if expiry == "" {
|
||||||
|
return "", nil
|
||||||
|
}
|
||||||
|
|
||||||
|
err := delivery.ValidateArchiveExpiry(expiry)
|
||||||
|
if err != nil {
|
||||||
|
http.Error(
|
||||||
|
w,
|
||||||
|
"Invalid archive expiry: "+err.Error(),
|
||||||
|
http.StatusBadRequest,
|
||||||
|
)
|
||||||
|
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
|
||||||
|
cfg := map[string]any{"expiry": expiry}
|
||||||
|
|
||||||
|
configBytes, err := json.Marshal(cfg)
|
||||||
|
if err != nil {
|
||||||
|
http.Error(
|
||||||
|
w, "Internal server error",
|
||||||
|
http.StatusInternalServerError,
|
||||||
|
)
|
||||||
|
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
|
||||||
|
return string(configBytes), nil
|
||||||
|
}
|
||||||
|
|
||||||
// HandleEntrypointDelete handles deleting an entrypoint.
|
// HandleEntrypointDelete handles deleting an entrypoint.
|
||||||
func (h *Handlers) HandleEntrypointDelete() http.HandlerFunc {
|
func (h *Handlers) HandleEntrypointDelete() http.HandlerFunc {
|
||||||
return h.deleteChildResource(
|
return h.deleteChildResource(
|
||||||
|
|||||||
@@ -32,3 +32,7 @@ func IsClientTLS(r *http.Request) bool {
|
|||||||
|
|
||||||
// LoginRateLimitConst exposes the loginRateLimit constant.
|
// LoginRateLimitConst exposes the loginRateLimit constant.
|
||||||
const LoginRateLimitConst = loginRateLimit
|
const LoginRateLimitConst = loginRateLimit
|
||||||
|
|
||||||
|
// PasswordChangeRateLimitConst exposes the
|
||||||
|
// passwordChangeRateLimit constant.
|
||||||
|
const PasswordChangeRateLimitConst = passwordChangeRateLimit
|
||||||
|
|||||||
@@ -14,6 +14,16 @@ const (
|
|||||||
|
|
||||||
// loginRateInterval is the time window for the rate limit.
|
// loginRateInterval is the time window for the rate limit.
|
||||||
loginRateInterval = 1 * time.Minute
|
loginRateInterval = 1 * time.Minute
|
||||||
|
|
||||||
|
// passwordChangeRateLimit is the maximum number of password
|
||||||
|
// change attempts per interval. Each attempt verifies the
|
||||||
|
// current password, so the endpoint must be rate-limited
|
||||||
|
// like any other password-based authentication endpoint.
|
||||||
|
passwordChangeRateLimit = 5
|
||||||
|
|
||||||
|
// passwordChangeRateInterval is the time window for the
|
||||||
|
// password change rate limit.
|
||||||
|
passwordChangeRateInterval = 1 * time.Minute
|
||||||
)
|
)
|
||||||
|
|
||||||
// LoginRateLimit returns middleware that enforces per-IP rate
|
// LoginRateLimit returns middleware that enforces per-IP rate
|
||||||
@@ -24,19 +34,53 @@ const (
|
|||||||
// honours X-Forwarded-For, X-Real-IP, and True-Client-IP headers
|
// honours X-Forwarded-For, X-Real-IP, and True-Client-IP headers
|
||||||
// for reverse-proxy setups.
|
// for reverse-proxy setups.
|
||||||
func (m *Middleware) LoginRateLimit() func(http.Handler) http.Handler {
|
func (m *Middleware) LoginRateLimit() func(http.Handler) http.Handler {
|
||||||
limiter := httprate.Limit(
|
return m.postRateLimit(
|
||||||
loginRateLimit,
|
loginRateLimit,
|
||||||
loginRateInterval,
|
loginRateInterval,
|
||||||
|
"login rate limit exceeded",
|
||||||
|
"Too many login attempts. Please try again later.",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
// PasswordChangeRateLimit returns middleware that enforces
|
||||||
|
// per-IP rate limiting on password change attempts. The change
|
||||||
|
// endpoint verifies the current password, so without a limit a
|
||||||
|
// stolen session could be used to brute-force it; the limit
|
||||||
|
// matches the login endpoint's.
|
||||||
|
func (m *Middleware) PasswordChangeRateLimit() func(http.Handler) http.Handler {
|
||||||
|
return m.postRateLimit(
|
||||||
|
passwordChangeRateLimit,
|
||||||
|
passwordChangeRateInterval,
|
||||||
|
"password change rate limit exceeded",
|
||||||
|
"Too many password change attempts. "+
|
||||||
|
"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
|
||||||
|
// given response message, and each rejection is logged with the
|
||||||
|
// given log message. IP extraction honours X-Forwarded-For,
|
||||||
|
// X-Real-IP, and True-Client-IP headers for reverse-proxy
|
||||||
|
// setups.
|
||||||
|
func (m *Middleware) postRateLimit(
|
||||||
|
limit int,
|
||||||
|
interval time.Duration,
|
||||||
|
logMessage, responseMessage string,
|
||||||
|
) func(http.Handler) http.Handler {
|
||||||
|
limiter := httprate.Limit(
|
||||||
|
limit,
|
||||||
|
interval,
|
||||||
httprate.WithKeyFuncs(httprate.KeyByRealIP),
|
httprate.WithKeyFuncs(httprate.KeyByRealIP),
|
||||||
httprate.WithLimitHandler(http.HandlerFunc(
|
httprate.WithLimitHandler(http.HandlerFunc(
|
||||||
func(w http.ResponseWriter, r *http.Request) {
|
func(w http.ResponseWriter, r *http.Request) {
|
||||||
m.log.Warn("login rate limit exceeded",
|
m.log.Warn(logMessage,
|
||||||
"path", r.URL.Path,
|
"path", r.URL.Path,
|
||||||
)
|
)
|
||||||
http.Error(
|
http.Error(
|
||||||
w,
|
w,
|
||||||
"Too many login attempts. "+
|
responseMessage,
|
||||||
"Please try again later.",
|
|
||||||
http.StatusTooManyRequests,
|
http.StatusTooManyRequests,
|
||||||
)
|
)
|
||||||
},
|
},
|
||||||
@@ -50,8 +94,7 @@ func (m *Middleware) LoginRateLimit() func(http.Handler) http.Handler {
|
|||||||
w http.ResponseWriter,
|
w http.ResponseWriter,
|
||||||
r *http.Request,
|
r *http.Request,
|
||||||
) {
|
) {
|
||||||
// Only rate-limit POST requests (actual login
|
// Only rate-limit POST requests.
|
||||||
// attempts)
|
|
||||||
if r.Method != http.MethodPost {
|
if r.Method != http.MethodPost {
|
||||||
next.ServeHTTP(w, r)
|
next.ServeHTTP(w, r)
|
||||||
|
|
||||||
|
|||||||
@@ -46,14 +46,20 @@ func TestLoginRateLimit_AllowsGET(t *testing.T) {
|
|||||||
assert.Equal(t, 20, callCount)
|
assert.Equal(t, 20, callCount)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestLoginRateLimit_LimitsPOST(t *testing.T) {
|
// runPostLimitTest exercises a POST-only rate limit middleware:
|
||||||
t.Parallel()
|
// the first limit POSTs to path from ip must pass, and the next
|
||||||
|
// one must be rejected with 429 without reaching the handler.
|
||||||
m, _ := testMiddleware(t, config.EnvironmentDev)
|
func runPostLimitTest(
|
||||||
|
t *testing.T,
|
||||||
|
mw func(http.Handler) http.Handler,
|
||||||
|
limit int,
|
||||||
|
path, ip string,
|
||||||
|
) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
var callCount int
|
var callCount int
|
||||||
|
|
||||||
handler := m.LoginRateLimit()(http.HandlerFunc(
|
handler := mw(http.HandlerFunc(
|
||||||
func(w http.ResponseWriter, _ *http.Request) {
|
func(w http.ResponseWriter, _ *http.Request) {
|
||||||
callCount++
|
callCount++
|
||||||
|
|
||||||
@@ -61,13 +67,13 @@ func TestLoginRateLimit_LimitsPOST(t *testing.T) {
|
|||||||
},
|
},
|
||||||
))
|
))
|
||||||
|
|
||||||
// First loginRateLimit POST requests should succeed
|
// The first limit POST requests should succeed
|
||||||
for i := range middleware.LoginRateLimitConst {
|
for i := range limit {
|
||||||
req := httptest.NewRequestWithContext(
|
req := httptest.NewRequestWithContext(
|
||||||
context.Background(),
|
context.Background(),
|
||||||
http.MethodPost, "/pages/login", nil,
|
http.MethodPost, path, nil,
|
||||||
)
|
)
|
||||||
req.RemoteAddr = "10.0.0.1:12345"
|
req.RemoteAddr = ip
|
||||||
|
|
||||||
w := httptest.NewRecorder()
|
w := httptest.NewRecorder()
|
||||||
handler.ServeHTTP(w, req)
|
handler.ServeHTTP(w, req)
|
||||||
@@ -81,9 +87,9 @@ func TestLoginRateLimit_LimitsPOST(t *testing.T) {
|
|||||||
// Next POST should be rate-limited
|
// Next POST should be rate-limited
|
||||||
req := httptest.NewRequestWithContext(
|
req := httptest.NewRequestWithContext(
|
||||||
context.Background(),
|
context.Background(),
|
||||||
http.MethodPost, "/pages/login", nil,
|
http.MethodPost, path, nil,
|
||||||
)
|
)
|
||||||
req.RemoteAddr = "10.0.0.1:12345"
|
req.RemoteAddr = ip
|
||||||
|
|
||||||
w := httptest.NewRecorder()
|
w := httptest.NewRecorder()
|
||||||
handler.ServeHTTP(w, req)
|
handler.ServeHTTP(w, req)
|
||||||
@@ -92,7 +98,35 @@ func TestLoginRateLimit_LimitsPOST(t *testing.T) {
|
|||||||
t, http.StatusTooManyRequests, w.Code,
|
t, http.StatusTooManyRequests, w.Code,
|
||||||
"POST after limit should be 429",
|
"POST after limit should be 429",
|
||||||
)
|
)
|
||||||
assert.Equal(t, middleware.LoginRateLimitConst, callCount)
|
assert.Equal(t, limit, callCount)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLoginRateLimit_LimitsPOST(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
m, _ := testMiddleware(t, config.EnvironmentDev)
|
||||||
|
|
||||||
|
runPostLimitTest(
|
||||||
|
t,
|
||||||
|
m.LoginRateLimit(),
|
||||||
|
middleware.LoginRateLimitConst,
|
||||||
|
"/pages/login",
|
||||||
|
"10.0.0.1:12345",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPasswordChangeRateLimit_LimitsPOST(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
m, _ := testMiddleware(t, config.EnvironmentDev)
|
||||||
|
|
||||||
|
runPostLimitTest(
|
||||||
|
t,
|
||||||
|
m.PasswordChangeRateLimit(),
|
||||||
|
middleware.PasswordChangeRateLimitConst,
|
||||||
|
"/user/admin/password",
|
||||||
|
"10.0.0.2:12345",
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestLoginRateLimit_IndependentPerIP(t *testing.T) {
|
func TestLoginRateLimit_IndependentPerIP(t *testing.T) {
|
||||||
|
|||||||
@@ -110,6 +110,9 @@ func (s *Server) setupUserRoutes() {
|
|||||||
r.Use(s.mw.NoCache())
|
r.Use(s.mw.NoCache())
|
||||||
r.Use(s.mw.RequireAuth())
|
r.Use(s.mw.RequireAuth())
|
||||||
r.Get("/", s.h.HandleProfile())
|
r.Get("/", s.h.HandleProfile())
|
||||||
|
r.With(s.mw.PasswordChangeRateLimit()).Post(
|
||||||
|
"/password", s.h.HandlePasswordChange(),
|
||||||
|
)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -6,6 +6,18 @@
|
|||||||
<div class="max-w-4xl mx-auto px-6 py-12">
|
<div class="max-w-4xl mx-auto px-6 py-12">
|
||||||
<h1 class="text-2xl font-medium text-gray-900 mb-6">User Profile</h1>
|
<h1 class="text-2xl font-medium text-gray-900 mb-6">User Profile</h1>
|
||||||
|
|
||||||
|
{{if .SuccessMessage}}
|
||||||
|
<div class="alert-success">
|
||||||
|
<span>{{.SuccessMessage}}</span>
|
||||||
|
</div>
|
||||||
|
{{end}}
|
||||||
|
|
||||||
|
{{if .ErrorMessage}}
|
||||||
|
<div class="alert-error">
|
||||||
|
<span>{{.ErrorMessage}}</span>
|
||||||
|
</div>
|
||||||
|
{{end}}
|
||||||
|
|
||||||
<div class="card p-6">
|
<div class="card p-6">
|
||||||
<div class="flex items-center mb-6">
|
<div class="flex items-center mb-6">
|
||||||
<div class="mr-4">
|
<div class="mr-4">
|
||||||
@@ -43,6 +55,50 @@
|
|||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
<div class="card p-6 mt-6">
|
||||||
|
<h3 class="text-lg font-medium text-gray-900 mb-3">Change Password</h3>
|
||||||
|
<form method="POST" action="/user/{{.User.Username}}/password" class="space-y-6">
|
||||||
|
<input type="hidden" name="csrf_token" value="{{.CSRFToken}}">
|
||||||
|
<div class="form-group">
|
||||||
|
<label for="current_password" class="label">Current Password</label>
|
||||||
|
<input
|
||||||
|
type="password"
|
||||||
|
id="current_password"
|
||||||
|
name="current_password"
|
||||||
|
required
|
||||||
|
autocomplete="current-password"
|
||||||
|
class="input"
|
||||||
|
>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<div class="form-group">
|
||||||
|
<label for="new_password" class="label">New Password</label>
|
||||||
|
<input
|
||||||
|
type="password"
|
||||||
|
id="new_password"
|
||||||
|
name="new_password"
|
||||||
|
required
|
||||||
|
autocomplete="new-password"
|
||||||
|
class="input"
|
||||||
|
>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<div class="form-group">
|
||||||
|
<label for="confirm_password" class="label">Confirm New Password</label>
|
||||||
|
<input
|
||||||
|
type="password"
|
||||||
|
id="confirm_password"
|
||||||
|
name="confirm_password"
|
||||||
|
required
|
||||||
|
autocomplete="new-password"
|
||||||
|
class="input"
|
||||||
|
>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<button type="submit" class="btn-primary">Change Password</button>
|
||||||
|
</form>
|
||||||
|
</div>
|
||||||
|
|
||||||
<div class="mt-6">
|
<div class="mt-6">
|
||||||
<a href="/" class="btn-secondary">Back to Home</a>
|
<a href="/" class="btn-secondary">Back to Home</a>
|
||||||
</div>
|
</div>
|
||||||
|
|||||||
@@ -113,6 +113,10 @@
|
|||||||
<input type="url" name="url" placeholder="https://hooks.slack.com/services/..." :disabled="targetType !== 'slack'" class="input text-sm">
|
<input type="url" name="url" placeholder="https://hooks.slack.com/services/..." :disabled="targetType !== 'slack'" class="input text-sm">
|
||||||
<p class="text-xs text-gray-500 mt-1">Slack or Mattermost incoming webhook URL. Payloads are pretty-printed in code blocks.</p>
|
<p class="text-xs text-gray-500 mt-1">Slack or Mattermost incoming webhook URL. Payloads are pretty-printed in code blocks.</p>
|
||||||
</div>
|
</div>
|
||||||
|
<div x-show="targetType === 'database'">
|
||||||
|
<input type="text" name="expiry" placeholder="never" :disabled="targetType !== 'database'" class="input text-sm">
|
||||||
|
<p class="text-xs text-gray-500 mt-1">Archive expiry: "never" (default) keeps rows forever, or a duration like "720h" prunes older rows.</p>
|
||||||
|
</div>
|
||||||
<button type="submit" class="btn-primary text-sm">Add Target</button>
|
<button type="submit" class="btn-primary text-sm">Add Target</button>
|
||||||
</form>
|
</form>
|
||||||
</div>
|
</div>
|
||||||
|
|||||||
Reference in New Issue
Block a user