Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ed81db137e | ||
|
|
752d6beead | ||
|
|
b1f43c9520 |
@@ -0,0 +1,112 @@
|
|||||||
|
package delivery_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"log/slog"
|
||||||
|
"net/http"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/webhooker/internal/delivery"
|
||||||
|
)
|
||||||
|
|
||||||
|
// newSSRFTestEngine builds an Engine whose shared client
|
||||||
|
// carries the SSRF-safe transport, mirroring production.
|
||||||
|
func newSSRFTestEngine() *delivery.Engine {
|
||||||
|
log := slog.New(slog.DiscardHandler)
|
||||||
|
|
||||||
|
client := &http.Client{
|
||||||
|
Timeout: 30 * time.Second,
|
||||||
|
Transport: delivery.NewSSRFSafeTransport(),
|
||||||
|
}
|
||||||
|
|
||||||
|
return delivery.NewTestEngine(log, client, 1)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestClientForConfig_TimeoutKeepsSSRFGuard asserts that a
|
||||||
|
// client returned by clientForConfig for a config with a
|
||||||
|
// per-target timeout still refuses connections to
|
||||||
|
// private/reserved addresses (the timeout must not drop the
|
||||||
|
// SSRF-safe transport).
|
||||||
|
func TestClientForConfig_TimeoutKeepsSSRFGuard(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
engine := newSSRFTestEngine()
|
||||||
|
|
||||||
|
blocked := []string{
|
||||||
|
"http://127.0.0.1/hook",
|
||||||
|
"http://169.254.169.254/latest/meta-data/",
|
||||||
|
"http://[fe80::1]/hook",
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, target := range blocked {
|
||||||
|
t.Run(target, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
cfg := &delivery.HTTPTargetConfig{
|
||||||
|
URL: target,
|
||||||
|
Timeout: 5,
|
||||||
|
}
|
||||||
|
|
||||||
|
client := engine.ExportClientForConfig(cfg)
|
||||||
|
|
||||||
|
require.NotSame(t, engine.ExportClient(), client,
|
||||||
|
"a per-target timeout must yield a "+
|
||||||
|
"distinct client",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert.Equal(t,
|
||||||
|
5*time.Second, client.Timeout,
|
||||||
|
"the per-target timeout must be applied",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert.Same(t,
|
||||||
|
engine.ExportClient().Transport,
|
||||||
|
client.Transport,
|
||||||
|
"the SSRF-safe transport must be reused, "+
|
||||||
|
"not dropped",
|
||||||
|
)
|
||||||
|
|
||||||
|
req, err := http.NewRequestWithContext(
|
||||||
|
context.Background(),
|
||||||
|
http.MethodPost, target, nil,
|
||||||
|
)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
resp, doErr := client.Do(req)
|
||||||
|
if resp != nil {
|
||||||
|
_ = resp.Body.Close()
|
||||||
|
}
|
||||||
|
|
||||||
|
require.Error(t, doErr,
|
||||||
|
"request to %s must be blocked", target,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert.Contains(t, doErr.Error(), "blocked",
|
||||||
|
"error must come from the SSRF guard",
|
||||||
|
)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestClientForConfig_NoTimeoutUnchanged asserts that with
|
||||||
|
// no per-target timeout the shared SSRF-safe client is
|
||||||
|
// returned unchanged.
|
||||||
|
func TestClientForConfig_NoTimeoutUnchanged(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
engine := newSSRFTestEngine()
|
||||||
|
|
||||||
|
cfg := &delivery.HTTPTargetConfig{
|
||||||
|
URL: "https://example.com/hook",
|
||||||
|
}
|
||||||
|
|
||||||
|
client := engine.ExportClientForConfig(cfg)
|
||||||
|
|
||||||
|
assert.Same(t, engine.ExportClient(), client,
|
||||||
|
"without a per-target timeout the shared client "+
|
||||||
|
"must be returned unchanged",
|
||||||
|
)
|
||||||
|
}
|
||||||
@@ -1006,8 +1006,11 @@ func (e *Engine) deliverLog(
|
|||||||
"webhook event delivered to log target",
|
"webhook event delivered to log target",
|
||||||
"delivery_id", d.ID,
|
"delivery_id", d.ID,
|
||||||
"event_id", d.EventID,
|
"event_id", d.EventID,
|
||||||
|
"webhook_id", d.Event.WebhookID,
|
||||||
|
"entrypoint_id", d.Event.EntrypointID,
|
||||||
"target_id", d.TargetID,
|
"target_id", d.TargetID,
|
||||||
"target_name", d.Target.Name,
|
"target_name", d.Target.Name,
|
||||||
|
"outcome", database.DeliveryStatusDelivered,
|
||||||
"method", d.Event.Method,
|
"method", d.Event.Method,
|
||||||
"content_type", d.Event.ContentType,
|
"content_type", d.Event.ContentType,
|
||||||
"body_length", len(d.Event.Body),
|
"body_length", len(d.Event.Body),
|
||||||
@@ -1713,10 +1716,15 @@ func (e *Engine) clientForConfig(
|
|||||||
cfg *HTTPTargetConfig,
|
cfg *HTTPTargetConfig,
|
||||||
) *http.Client {
|
) *http.Client {
|
||||||
if cfg.Timeout > 0 {
|
if cfg.Timeout > 0 {
|
||||||
|
// Reuse the shared client's SSRF-safe transport so
|
||||||
|
// a per-target timeout does not drop the
|
||||||
|
// request-time private-IP guard. Only the timeout
|
||||||
|
// is overridden.
|
||||||
return &http.Client{
|
return &http.Client{
|
||||||
Timeout: time.Duration(
|
Timeout: time.Duration(
|
||||||
cfg.Timeout,
|
cfg.Timeout,
|
||||||
) * time.Second,
|
) * time.Second,
|
||||||
|
Transport: e.client.Transport,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package delivery_test
|
package delivery_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
@@ -435,6 +436,91 @@ func TestDeliverLog_ImmediateSuccess(t *testing.T) {
|
|||||||
assert.True(t, result.Success)
|
assert.True(t, result.Success)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestDeliverLog_StructuredLogFields(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db := testWebhookDB(t)
|
||||||
|
|
||||||
|
var logBuf bytes.Buffer
|
||||||
|
|
||||||
|
e := delivery.NewTestEngine(
|
||||||
|
slog.New(slog.NewTextHandler(
|
||||||
|
&logBuf,
|
||||||
|
&slog.HandlerOptions{Level: slog.LevelDebug},
|
||||||
|
)),
|
||||||
|
&http.Client{Timeout: 5 * time.Second},
|
||||||
|
1,
|
||||||
|
)
|
||||||
|
|
||||||
|
event := seedEvent(t, db, `{"log":"structured"}`)
|
||||||
|
|
||||||
|
dlv := seedDelivery(
|
||||||
|
t, db, 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: "structured-log",
|
||||||
|
Type: database.TargetTypeLog,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
d.ID = dlv.ID
|
||||||
|
|
||||||
|
e.ExportDeliverLog(db, d)
|
||||||
|
|
||||||
|
// The delivery is marked delivered and a success
|
||||||
|
// DeliveryResult with no HTTP status is recorded,
|
||||||
|
// mirroring the other target types' bookkeeping.
|
||||||
|
var updated database.Delivery
|
||||||
|
|
||||||
|
require.NoError(t, db.First(
|
||||||
|
&updated, "id = ?", dlv.ID,
|
||||||
|
).Error)
|
||||||
|
|
||||||
|
assert.Equal(t,
|
||||||
|
database.DeliveryStatusDelivered, updated.Status,
|
||||||
|
"log target should immediately succeed",
|
||||||
|
)
|
||||||
|
|
||||||
|
var result database.DeliveryResult
|
||||||
|
|
||||||
|
require.NoError(t, db.Where(
|
||||||
|
"delivery_id = ?", dlv.ID,
|
||||||
|
).First(&result).Error)
|
||||||
|
|
||||||
|
assert.True(t, result.Success)
|
||||||
|
assert.Equal(t, 0, result.StatusCode,
|
||||||
|
"log target should not have an HTTP status",
|
||||||
|
)
|
||||||
|
|
||||||
|
assertLogFields(t, logBuf.String(), event, "structured-log")
|
||||||
|
}
|
||||||
|
|
||||||
|
// assertLogFields checks that a log target's structured
|
||||||
|
// log line carries the required fields: event id,
|
||||||
|
// webhook/entrypoint, target name, and outcome.
|
||||||
|
func assertLogFields(
|
||||||
|
t *testing.T,
|
||||||
|
logged string,
|
||||||
|
event database.Event,
|
||||||
|
targetName string,
|
||||||
|
) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
assert.Contains(t, logged, "event_id="+event.ID)
|
||||||
|
assert.Contains(t, logged, "webhook_id="+event.WebhookID)
|
||||||
|
assert.Contains(t,
|
||||||
|
logged, "entrypoint_id="+event.EntrypointID,
|
||||||
|
)
|
||||||
|
assert.Contains(t, logged, "target_name="+targetName)
|
||||||
|
assert.Contains(t, logged, "outcome=delivered")
|
||||||
|
}
|
||||||
|
|
||||||
func TestDeliverHTTP_WithRetries_Success(t *testing.T) {
|
func TestDeliverHTTP_WithRetries_Success(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
@@ -126,6 +126,18 @@ func (e *Engine) ExportDoHTTPRequest(
|
|||||||
return e.doHTTPRequest(ctx, cfg, event)
|
return e.doHTTPRequest(ctx, cfg, event)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ExportClientForConfig exposes clientForConfig.
|
||||||
|
func (e *Engine) ExportClientForConfig(
|
||||||
|
cfg *HTTPTargetConfig,
|
||||||
|
) *http.Client {
|
||||||
|
return e.clientForConfig(cfg)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ExportClient returns the engine's shared HTTP client.
|
||||||
|
func (e *Engine) ExportClient() *http.Client {
|
||||||
|
return e.client
|
||||||
|
}
|
||||||
|
|
||||||
// ExportScheduleRetry exposes scheduleRetry.
|
// ExportScheduleRetry exposes scheduleRetry.
|
||||||
func (e *Engine) ExportScheduleRetry(
|
func (e *Engine) ExportScheduleRetry(
|
||||||
task Task, delay time.Duration,
|
task Task, delay time.Duration,
|
||||||
|
|||||||
Reference in New Issue
Block a user