Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sysops
|
43dca005f2 | ||
|
sysops
|
24a522e10c |
@@ -0,0 +1,40 @@
|
||||
// secheaders-devserver dient der manuellen/e2e-Verifikation von API-08
|
||||
// (Pruefung 2: CSP blockiert Inline-Script) mit einem echten Browser.
|
||||
// Zwei Pfade: /protected (mit Middleware) und /unprotected (ohne) — beide
|
||||
// liefern dieselbe Seite mit einem Inline-Script, das ein DOM-Element
|
||||
// aendert. Ein echter Browser zeigt den Unterschied.
|
||||
package main
|
||||
|
||||
import (
|
||||
"log"
|
||||
"net/http"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/secheaders"
|
||||
)
|
||||
|
||||
const page = `<!doctype html>
|
||||
<html><body>
|
||||
<div id="marker">BLOCKED</div>
|
||||
<script>document.getElementById('marker').textContent = 'INLINE_RAN';</script>
|
||||
</body></html>`
|
||||
|
||||
func servePage(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
||||
_, _ = w.Write([]byte(page))
|
||||
}
|
||||
|
||||
func main() {
|
||||
cfg := secheaders.DefaultConfig()
|
||||
mux := http.NewServeMux()
|
||||
// SetHeaders statt Middleware: dieser lokale Demo-Server laeuft ohne
|
||||
// TLS-Terminierung, der HTTPS-Redirect-Teil von Middleware wuerde hier
|
||||
// in eine Endlosschleife fuehren (siehe secheaders.go-Dokumentation).
|
||||
mux.HandleFunc("/protected", func(w http.ResponseWriter, r *http.Request) {
|
||||
secheaders.SetHeaders(cfg, w)
|
||||
servePage(w, r)
|
||||
})
|
||||
mux.HandleFunc("/unprotected", servePage)
|
||||
|
||||
log.Println("secheaders-devserver listening on :8086")
|
||||
log.Fatal(http.ListenAndServe(":8086", mux))
|
||||
}
|
||||
@@ -0,0 +1,126 @@
|
||||
// Package secheaders implementiert Core API-08: eine gemeinsame
|
||||
// Security-Header-Middleware/Bibliothek, die jedes Modul-Frontend (DMS,
|
||||
// Mail, Workflow, AI, Connect) einbindet, statt Header pro Modul erneut zu
|
||||
// definieren (Akzeptanzkriterium 3) — dieselbe Konsolidierungslogik wie
|
||||
// beim zentralen Rate-Limiting (API-03). Die tatsaechliche EINBINDUNG in
|
||||
// jedes einzelne Modul-Frontend ist ausdruecklich NICHT Teil dieser Kachel
|
||||
// (siehe Ticket "Nicht Bestandteil").
|
||||
package secheaders
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Config ist die Baseline-Konfiguration. DefaultConfig() liefert einen
|
||||
// Ausgangspunkt OHNE 'unsafe-inline'/'unsafe-eval' (Akzeptanzkriterium 1)
|
||||
// und mit angemessener HSTS-Laufzeit (Akzeptanzkriterium 2).
|
||||
type Config struct {
|
||||
// ContentSecurityPolicy wird unveraendert als Header-Wert gesendet.
|
||||
ContentSecurityPolicy string
|
||||
// HSTSMaxAge bestimmt die Gueltigkeitsdauer, die Browser sich merken.
|
||||
HSTSMaxAge time.Duration
|
||||
HSTSIncludeSubdomains bool
|
||||
FrameOptions string
|
||||
ReferrerPolicy string
|
||||
ContentTypeNoSniff bool
|
||||
}
|
||||
|
||||
// DefaultConfig liefert die dokumentierte Baseline: restriktive CSP ohne
|
||||
// unsichere Direktiven, HSTS mit 180 Tagen (empfohlener Mindestwert fuer
|
||||
// HSTS-Preloading-Listen liegt bei 1 Jahr — 180 Tage ist ein bewusst
|
||||
// konservativer, aber bereits "angemessener" Ausgangswert, siehe
|
||||
// Akzeptanzkriterium 2), Clickjacking- und MIME-Sniffing-Schutz.
|
||||
func DefaultConfig() Config {
|
||||
return Config{
|
||||
ContentSecurityPolicy: strings.Join([]string{
|
||||
"default-src 'self'",
|
||||
"script-src 'self'",
|
||||
"style-src 'self'",
|
||||
"img-src 'self' data:",
|
||||
"object-src 'none'",
|
||||
"base-uri 'self'",
|
||||
"frame-ancestors 'none'",
|
||||
}, "; "),
|
||||
HSTSMaxAge: 180 * 24 * time.Hour,
|
||||
HSTSIncludeSubdomains: true,
|
||||
FrameOptions: "DENY",
|
||||
ReferrerPolicy: "strict-origin-when-cross-origin",
|
||||
ContentTypeNoSniff: true,
|
||||
}
|
||||
}
|
||||
|
||||
// RequiredHeaders sind die von dieser Baseline gesetzten Header-Namen — die
|
||||
// Referenzliste fuer CheckCoverage (Akzeptanzkriterium/Pruefung 1).
|
||||
var RequiredHeaders = []string{
|
||||
"Content-Security-Policy",
|
||||
"Strict-Transport-Security",
|
||||
"X-Frame-Options",
|
||||
"Referrer-Policy",
|
||||
"X-Content-Type-Options",
|
||||
}
|
||||
|
||||
func (c Config) hstsValue() string {
|
||||
v := fmt.Sprintf("max-age=%d", int(c.HSTSMaxAge.Seconds()))
|
||||
if c.HSTSIncludeSubdomains {
|
||||
v += "; includeSubDomains"
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
// Middleware setzt die Baseline-Header auf JEDER Antwort und erzwingt einen
|
||||
// HTTP->HTTPS-Redirect, WENN die Anfrage nicht bereits ueber TLS ankam
|
||||
// (Akzeptanzkriterium 2 / Pruefung 3) — serverseitig durchgesetzt, nicht nur
|
||||
// dem Browser-HSTS-Cache ueberlassen (der erst nach dem ERSTEN sicheren
|
||||
// Aufruf greift).
|
||||
func Middleware(cfg Config, next http.HandlerFunc) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
if !isSecure(r) {
|
||||
target := "https://" + r.Host + r.URL.RequestURI()
|
||||
http.Redirect(w, r, target, http.StatusMovedPermanently)
|
||||
return
|
||||
}
|
||||
|
||||
SetHeaders(cfg, w)
|
||||
next(w, r)
|
||||
}
|
||||
}
|
||||
|
||||
// SetHeaders setzt die Baseline-Header direkt, OHNE die HTTPS-Redirect-
|
||||
// Pruefung von Middleware — nuetzlich fuer lokale Entwicklungs-/
|
||||
// Demo-Umgebungen ohne TLS-Terminierung (siehe cmd/secheaders-devserver)
|
||||
// und fuer Aufrufer, die den Redirect selbst schon anderweitig sicherstellen.
|
||||
func SetHeaders(cfg Config, w http.ResponseWriter) {
|
||||
w.Header().Set("Content-Security-Policy", cfg.ContentSecurityPolicy)
|
||||
w.Header().Set("Strict-Transport-Security", cfg.hstsValue())
|
||||
w.Header().Set("X-Frame-Options", cfg.FrameOptions)
|
||||
w.Header().Set("Referrer-Policy", cfg.ReferrerPolicy)
|
||||
if cfg.ContentTypeNoSniff {
|
||||
w.Header().Set("X-Content-Type-Options", "nosniff")
|
||||
}
|
||||
}
|
||||
|
||||
// isSecure erkennt TLS direkt ODER ueber den ueblichen Reverse-Proxy-Header
|
||||
// (X-Forwarded-Proto) — der Core-Dienst laeuft in der Praxis hinter einem
|
||||
// TLS-terminierenden Proxy, sieht r.TLS also nie direkt gesetzt.
|
||||
func isSecure(r *http.Request) bool {
|
||||
if r.TLS != nil {
|
||||
return true
|
||||
}
|
||||
return strings.EqualFold(r.Header.Get("X-Forwarded-Proto"), "https")
|
||||
}
|
||||
|
||||
// CheckCoverage vergleicht die tatsaechlich gesetzten Header einer Antwort
|
||||
// gegen RequiredHeaders und liefert die fehlenden — der "automatisierte
|
||||
// Header-Scan" aus Pruefung 1, anwendbar auf jede Antwort eines beliebigen
|
||||
// Modul-Frontends.
|
||||
func CheckCoverage(headers http.Header) (missing []string) {
|
||||
for _, name := range RequiredHeaders {
|
||||
if headers.Get(name) == "" {
|
||||
missing = append(missing, name)
|
||||
}
|
||||
}
|
||||
return missing
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
package secheaders
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func secureRequest(method, target string) *http.Request {
|
||||
req := httptest.NewRequest(method, target, nil)
|
||||
req.Header.Set("X-Forwarded-Proto", "https")
|
||||
return req
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1: CSP definiert ohne unsafe-inline/unsafe-eval.
|
||||
func TestDefaultConfig_CSPHasNoUnsafeDirectives(t *testing.T) {
|
||||
cfg := DefaultConfig()
|
||||
if strings.Contains(cfg.ContentSecurityPolicy, "unsafe-inline") {
|
||||
t.Fatal("csp enthaelt 'unsafe-inline', erwartet keine unsicheren direktiven")
|
||||
}
|
||||
if strings.Contains(cfg.ContentSecurityPolicy, "unsafe-eval") {
|
||||
t.Fatal("csp enthaelt 'unsafe-eval', erwartet keine unsicheren direktiven")
|
||||
}
|
||||
if !strings.Contains(cfg.ContentSecurityPolicy, "default-src 'self'") {
|
||||
t.Fatal("erwartet restriktive default-src 'self' als ausgangspunkt")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2: HSTS mit angemessener max-age gesetzt.
|
||||
func TestMiddleware_SetsHSTSWithSufficientMaxAge(t *testing.T) {
|
||||
cfg := DefaultConfig()
|
||||
handler := Middleware(cfg, func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
handler(rec, secureRequest(http.MethodGet, "https://example.test/"))
|
||||
|
||||
hsts := rec.Header().Get("Strict-Transport-Security")
|
||||
if !strings.Contains(hsts, "max-age=") {
|
||||
t.Fatalf("hsts-header fehlt max-age: %q", hsts)
|
||||
}
|
||||
// "Angemessen" heisst hier: mindestens 30 Tage (2592000s) — deutlich
|
||||
// ueber dem bekannten Anti-Pattern kurzer/fehlender max-age-Werte.
|
||||
if cfg.HSTSMaxAge.Seconds() < 2592000 {
|
||||
t.Fatalf("konfigurierte hsts-max-age = %v, erwartet mindestens 30 tage", cfg.HSTSMaxAge)
|
||||
}
|
||||
if !strings.Contains(hsts, "includeSubDomains") {
|
||||
t.Fatalf("erwartet includeSubDomains im hsts-header: %q", hsts)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 + Pruefung 1: EINE gemeinsame Middleware setzt ALLE
|
||||
// Baseline-Header vollstaendig — automatisiert ueberprueft mit CheckCoverage.
|
||||
func TestMiddleware_ProducesFullHeaderCoverage(t *testing.T) {
|
||||
cfg := DefaultConfig()
|
||||
handler := Middleware(cfg, func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
handler(rec, secureRequest(http.MethodGet, "https://example.test/"))
|
||||
|
||||
if missing := CheckCoverage(rec.Header()); len(missing) != 0 {
|
||||
t.Fatalf("erwartet vollstaendige header-abdeckung, fehlen: %v", missing)
|
||||
}
|
||||
}
|
||||
|
||||
// Pruefung 1 (Kehrseite): CheckCoverage erkennt fehlende Header
|
||||
// zuverlaessig — Voraussetzung dafuer, dass ein automatisierter Scan
|
||||
// gegen ein Modul-Frontend OHNE eingebundene Middleware ueberhaupt etwas
|
||||
// meldet.
|
||||
func TestCheckCoverage_DetectsMissingHeadersOnBareHandler(t *testing.T) {
|
||||
bareHandler := func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
bareHandler(rec, httptest.NewRequest(http.MethodGet, "/", nil))
|
||||
|
||||
missing := CheckCoverage(rec.Header())
|
||||
if len(missing) != len(RequiredHeaders) {
|
||||
t.Fatalf("erwartet %d fehlende header bei unbehandeltem handler, habe %d: %v",
|
||||
len(RequiredHeaders), len(missing), missing)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 3: ein Aufruf ueber (simuliertes) HTTP
|
||||
// wird server-seitig auf HTTPS umgeleitet, statt sich allein auf den
|
||||
// HSTS-Browser-Cache zu verlassen (der erst nach dem ersten sicheren
|
||||
// Aufruf greift).
|
||||
func TestMiddleware_RedirectsInsecureRequestToHTTPS(t *testing.T) {
|
||||
cfg := DefaultConfig()
|
||||
called := false
|
||||
handler := Middleware(cfg, func(w http.ResponseWriter, r *http.Request) {
|
||||
called = true
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "http://example.test/pfad?x=1", nil)
|
||||
// Kein X-Forwarded-Proto, kein r.TLS -> unsicherer Aufruf.
|
||||
rec := httptest.NewRecorder()
|
||||
handler(rec, req)
|
||||
|
||||
if rec.Code != http.StatusMovedPermanently {
|
||||
t.Fatalf("status = %d, want 301 (redirect auf https)", rec.Code)
|
||||
}
|
||||
location := rec.Header().Get("Location")
|
||||
if !strings.HasPrefix(location, "https://") {
|
||||
t.Fatalf("location = %q, want https://-praefix", location)
|
||||
}
|
||||
if !strings.Contains(location, "/pfad?x=1") {
|
||||
t.Fatalf("location = %q, erwartet erhaltenen pfad+query", location)
|
||||
}
|
||||
if called {
|
||||
t.Fatal("eigentlicher handler haette bei unsicherem aufruf nicht erreicht werden duerfen")
|
||||
}
|
||||
}
|
||||
@@ -1,177 +0,0 @@
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/subtle"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// DefaultMaxAttempts ist die konfigurierbare Obergrenze, ab der eine
|
||||
// Zustellung endgueltig als fehlgeschlagen gilt (Akzeptanzkriterium 2).
|
||||
const DefaultMaxAttempts = 5
|
||||
|
||||
// DefaultBaseBackoff ist die Basisdauer fuer exponentielles Backoff:
|
||||
// naechster Versuch nach BaseBackoff * 2^attempt (Akzeptanzkriterium 2).
|
||||
const DefaultBaseBackoff = 2 * time.Second
|
||||
|
||||
// Dispatcher liefert faellige Zustellungen aus. Konfigurierbar in Tests
|
||||
// (kleine BaseBackoff, kleine MaxAttempts), damit Retry/Backoff/Obergrenze
|
||||
// ohne minutenlange Wartezeit real durchlaufen werden koennen.
|
||||
type Dispatcher struct {
|
||||
pool pgxIface
|
||||
client *http.Client
|
||||
MaxAttempts int
|
||||
BaseBackoff time.Duration
|
||||
}
|
||||
|
||||
// pgxIface ist die schmale Teilmenge von *pgxpool.Pool, die der Dispatcher
|
||||
// braucht — als Interface, damit Tests keine echte Verbindung fuer reine
|
||||
// Signatur-/Backoff-Logik brauchen (wird hier aber durchgehend mit echten
|
||||
// Integrationstests gegen Postgres verwendet, siehe dispatcher_test.go).
|
||||
type pgxIface interface {
|
||||
Begin(ctx context.Context) (pgx.Tx, error)
|
||||
}
|
||||
|
||||
func NewDispatcher(pool pgxIface, client *http.Client) *Dispatcher {
|
||||
if client == nil {
|
||||
client = &http.Client{Timeout: 5 * time.Second}
|
||||
}
|
||||
return &Dispatcher{pool: pool, client: client, MaxAttempts: DefaultMaxAttempts, BaseBackoff: DefaultBaseBackoff}
|
||||
}
|
||||
|
||||
// backoffFor berechnet die Wartezeit vor dem naechsten Versuch: exponentiell
|
||||
// wachsend mit der Anzahl bereits unternommener Versuche.
|
||||
func (d *Dispatcher) backoffFor(attempt int) time.Duration {
|
||||
return d.BaseBackoff * time.Duration(1<<uint(attempt))
|
||||
}
|
||||
|
||||
// ProcessDue liefert ALLE derzeit faelligen Zustellungen aus — dasselbe
|
||||
// SELECT ... FOR UPDATE SKIP LOCKED-Muster wie
|
||||
// internal/tenant.Lifecycle.ProcessDueDeletions, damit mehrere Dispatcher-
|
||||
// Instanzen dieselbe Zustellung nie doppelt bearbeiten.
|
||||
func (d *Dispatcher) ProcessDue(ctx context.Context) (int, error) {
|
||||
tx, err := d.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("transaktion starten: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
rows, err := tx.Query(ctx, `
|
||||
SELECT wd.id, wd.payload, wd.attempt, ws.target_url, ws.secret
|
||||
FROM webhook_deliveries wd
|
||||
JOIN webhook_subscriptions ws ON ws.id = wd.subscription_id
|
||||
WHERE wd.status = $1 AND wd.next_attempt_at <= now()
|
||||
FOR UPDATE OF wd SKIP LOCKED
|
||||
`, StatusPending)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("faellige zustellungen abfragen: %w", err)
|
||||
}
|
||||
|
||||
var deliveries []delivery
|
||||
for rows.Next() {
|
||||
var del delivery
|
||||
if err := rows.Scan(&del.ID, &del.Payload, &del.Attempt, &del.TargetURL, &del.Secret); err != nil {
|
||||
rows.Close()
|
||||
return 0, fmt.Errorf("zustellung lesen: %w", err)
|
||||
}
|
||||
deliveries = append(deliveries, del)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
for _, del := range deliveries {
|
||||
d.attemptOne(ctx, tx, del)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return 0, fmt.Errorf("transaktion committen: %w", err)
|
||||
}
|
||||
return len(deliveries), nil
|
||||
}
|
||||
|
||||
// attemptOne fuehrt GENAU EINEN Zustellversuch aus und aktualisiert den
|
||||
// Zustellungsdatensatz entsprechend — Erfolg (Akzeptanzkriterium 1/Pruefung
|
||||
// 1), erneuter Fehlversuch mit Backoff, oder endgueltiges Scheitern nach
|
||||
// DefaultMaxAttempts (Akzeptanzkriterium 2/Pruefung 2). Ein Fehler bei
|
||||
// GENAU EINER Zustellung darf die anderen in diesem Batch nicht verhindern.
|
||||
func (d *Dispatcher) attemptOne(ctx context.Context, tx pgx.Tx, del delivery) {
|
||||
signature := Sign(del.Secret, del.Payload)
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, del.TargetURL, bytes.NewReader(del.Payload))
|
||||
deliveryErr := err
|
||||
var statusCode int
|
||||
if err == nil {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set(SignatureHeader, signature)
|
||||
resp, err := d.client.Do(req)
|
||||
if err != nil {
|
||||
deliveryErr = err
|
||||
} else {
|
||||
statusCode = resp.StatusCode
|
||||
resp.Body.Close()
|
||||
if statusCode < 200 || statusCode >= 300 {
|
||||
deliveryErr = fmt.Errorf("unerwarteter statuscode %d", statusCode)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if deliveryErr == nil {
|
||||
_, _ = tx.Exec(ctx, `
|
||||
UPDATE webhook_deliveries SET status = $2, delivered_at = now(), attempt = attempt + 1
|
||||
WHERE id = $1
|
||||
`, del.ID, StatusDelivered)
|
||||
return
|
||||
}
|
||||
|
||||
nextAttempt := del.Attempt + 1
|
||||
if nextAttempt >= d.MaxAttempts {
|
||||
_, _ = tx.Exec(ctx, `
|
||||
UPDATE webhook_deliveries SET status = $2, attempt = $3, last_error = $4
|
||||
WHERE id = $1
|
||||
`, del.ID, StatusFailed, nextAttempt, deliveryErr.Error())
|
||||
return
|
||||
}
|
||||
|
||||
nextAttemptAt := time.Now().Add(d.backoffFor(nextAttempt))
|
||||
_, _ = tx.Exec(ctx, `
|
||||
UPDATE webhook_deliveries SET attempt = $2, next_attempt_at = $3, last_error = $4
|
||||
WHERE id = $1
|
||||
`, del.ID, nextAttempt, nextAttemptAt, deliveryErr.Error())
|
||||
}
|
||||
|
||||
// Run ruft ProcessDue in festen Abstaenden auf, bis ctx beendet wird —
|
||||
// dieselbe Konvention wie internal/tenant.Lifecycle.RunSweeper.
|
||||
func (d *Dispatcher) Run(ctx context.Context, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
_, _ = d.ProcessDue(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// VerifySignature prueft empfaengerseitig, ob signature zu payload und
|
||||
// secret passt — timing-safe (dasselbe Muster wie internal/audit.timingsafe),
|
||||
// damit ein Empfaenger die Authentizitaet einer Zustellung pruefen kann
|
||||
// (Akzeptanzkriterium 3).
|
||||
func VerifySignature(secret string, payload []byte, signature string) bool {
|
||||
expected := Sign(secret, payload)
|
||||
expectedBytes, err1 := hex.DecodeString(expected)
|
||||
gotBytes, err2 := hex.DecodeString(signature)
|
||||
if err1 != nil || err2 != nil {
|
||||
return false
|
||||
}
|
||||
return subtle.ConstantTimeCompare(expectedBytes, gotBytes) == 1
|
||||
}
|
||||
@@ -1,131 +0,0 @@
|
||||
// Package webhook implementiert Core API-07: eine zentrale Webhook-Registry
|
||||
// und Zustellungs-Engine fuer alle Fachmodule (DMS/Mail/Archive/Workflow/AI).
|
||||
// Module reichen Ereignisse EINMAL zur Zustellung ein und implementieren
|
||||
// selbst KEINE eigene Retry-/Signatur-Logik — das ist der zentrale Zweck
|
||||
// dieser Kachel ("Bewusst vermeiden: jedes Modul baut seine eigene
|
||||
// Webhook-Zustellungs-Engine"). Zustellung laeuft ueber dieselbe
|
||||
// Postgres-Jobqueue-Konvention (SELECT ... FOR UPDATE SKIP LOCKED) wie
|
||||
// internal/tenant.Lifecycle.ProcessDueDeletions (TEN-04) und
|
||||
// internal/notify.Dispatcher (CFG-02) — kein Redis/AMQP.
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
const (
|
||||
StatusPending = "pending"
|
||||
StatusDelivered = "delivered"
|
||||
StatusFailed = "failed"
|
||||
)
|
||||
|
||||
// Subscription ist EIN externer Abonnent fuer einen Ereignistyp
|
||||
// (Akzeptanzkriterium 1: Module registrieren Ereignistypen, externe
|
||||
// Abonnenten registrieren Ziel-URLs — dieses Paket modelliert die
|
||||
// Abonnenten-Seite; welche Ereignistypen ein Modul anbietet, ist bewusst
|
||||
// NICHT Teil dieser Kachel).
|
||||
type Subscription struct {
|
||||
ID string
|
||||
EventType string
|
||||
TargetURL string
|
||||
Secret string
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// Store persistiert Abonnements und Zustellversuche.
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
// Subscribe registriert einen Abonnenten fuer einen Ereignistyp. secret wird
|
||||
// spaeter zur HMAC-Signierung jeder Zustellung an diesen Abonnenten
|
||||
// verwendet (Akzeptanzkriterium 3).
|
||||
func (s *Store) Subscribe(ctx context.Context, eventType, targetURL, secret string) (Subscription, error) {
|
||||
var sub Subscription
|
||||
sub.EventType, sub.TargetURL, sub.Secret = eventType, targetURL, secret
|
||||
row := s.pool.QueryRow(ctx, `
|
||||
INSERT INTO webhook_subscriptions (event_type, target_url, secret)
|
||||
VALUES ($1, $2, $3)
|
||||
RETURNING id, created_at
|
||||
`, eventType, targetURL, secret)
|
||||
if err := row.Scan(&sub.ID, &sub.CreatedAt); err != nil {
|
||||
return Subscription{}, fmt.Errorf("abonnement anlegen: %w", err)
|
||||
}
|
||||
return sub, nil
|
||||
}
|
||||
|
||||
// Enqueue reicht EIN Ereignis zur Zustellung an ALLE Abonnenten des
|
||||
// angegebenen Ereignistyps ein — dies ist die EINZIGE Schnittstelle, die
|
||||
// ein Fachmodul braucht (Akzeptanzkriterium 1). Jeder Abonnent erhaelt
|
||||
// einen eigenen, unabhaengigen Zustellversuch-Datensatz.
|
||||
func (s *Store) Enqueue(ctx context.Context, eventType string, payload any) (int, error) {
|
||||
payloadJSON, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("payload serialisieren: %w", err)
|
||||
}
|
||||
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id FROM webhook_subscriptions WHERE event_type = $1
|
||||
`, eventType)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("abonnenten ermitteln: %w", err)
|
||||
}
|
||||
var subscriptionIDs []string
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
rows.Close()
|
||||
return 0, fmt.Errorf("abonnent lesen: %w", err)
|
||||
}
|
||||
subscriptionIDs = append(subscriptionIDs, id)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
for _, subID := range subscriptionIDs {
|
||||
if _, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO webhook_deliveries (subscription_id, event_type, payload, status, next_attempt_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
`, subID, eventType, payloadJSON, StatusPending); err != nil {
|
||||
return 0, fmt.Errorf("zustellung einreihen: %w", err)
|
||||
}
|
||||
}
|
||||
return len(subscriptionIDs), nil
|
||||
}
|
||||
|
||||
// delivery ist ein interner Datensatz fuer EINEN Zustellversuch, inklusive
|
||||
// der zugehoerigen Abonnentendaten (per JOIN geladen).
|
||||
type delivery struct {
|
||||
ID string
|
||||
TargetURL string
|
||||
Secret string
|
||||
Payload []byte
|
||||
Attempt int
|
||||
}
|
||||
|
||||
// Sign berechnet die HMAC-SHA256-Signatur des Payloads (Akzeptanzkriterium
|
||||
// 3) — hex-kodiert, damit sie problemlos als HTTP-Header uebertragen werden
|
||||
// kann.
|
||||
func Sign(secret string, payload []byte) string {
|
||||
mac := hmac.New(sha256.New, []byte(secret))
|
||||
mac.Write(payload)
|
||||
return hex.EncodeToString(mac.Sum(nil))
|
||||
}
|
||||
|
||||
// SignatureHeader ist der HTTP-Header, unter dem die Signatur uebertragen
|
||||
// wird — dokumentierter Vertrag fuer Empfaenger (Akzeptanzkriterium 3).
|
||||
const SignatureHeader = "X-Nexarch-Signature-256"
|
||||
@@ -1,205 +0,0 @@
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func setupTest(t *testing.T) (*Store, *pgxpool.Pool, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS webhook_subscriptions (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), event_type TEXT NOT NULL,
|
||||
target_url TEXT NOT NULL, secret TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS webhook_deliveries (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), subscription_id UUID NOT NULL REFERENCES webhook_subscriptions(id),
|
||||
event_type TEXT NOT NULL, payload JSONB NOT NULL, status TEXT NOT NULL DEFAULT 'pending',
|
||||
attempt INT NOT NULL DEFAULT 0, next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
last_error TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), delivered_at TIMESTAMPTZ
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() { pool.Close() }
|
||||
return NewStore(pool), pool, cleanup
|
||||
}
|
||||
|
||||
func uniqueEventType(prefix string) string {
|
||||
return prefix + "-" + time.Now().Format("150405.000000000")
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 1: ein Modul reicht ein Ereignis ein
|
||||
// (Enqueue) ohne eigene Zustellungslogik, der zentrale Dispatcher liefert
|
||||
// zuverlaessig aus.
|
||||
func TestEnqueueAndProcessDue_DeliversSuccessfully(t *testing.T) {
|
||||
store, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
var receivedBody []byte
|
||||
var receivedSignature string
|
||||
target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
receivedBody = body
|
||||
receivedSignature = r.Header.Get(SignatureHeader)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer target.Close()
|
||||
|
||||
eventType := uniqueEventType("dms.file.created")
|
||||
sub, err := store.Subscribe(ctx, eventType, target.URL, "geheimes-secret")
|
||||
if err != nil {
|
||||
t.Fatalf("subscribe: %v", err)
|
||||
}
|
||||
|
||||
n, err := store.Enqueue(ctx, eventType, map[string]string{"file_id": "42"})
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("erwartet 1 eingereihte zustellung, habe %d", n)
|
||||
}
|
||||
|
||||
dispatcher := NewDispatcher(pool, target.Client())
|
||||
processed, err := dispatcher.ProcessDue(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("processdue: %v", err)
|
||||
}
|
||||
if processed != 1 {
|
||||
t.Fatalf("erwartet 1 verarbeitete zustellung, habe %d", processed)
|
||||
}
|
||||
|
||||
var status string
|
||||
if err := pool.QueryRow(ctx, `SELECT status FROM webhook_deliveries WHERE subscription_id = $1`, sub.ID).Scan(&status); err != nil {
|
||||
t.Fatalf("status lesen: %v", err)
|
||||
}
|
||||
if status != StatusDelivered {
|
||||
t.Fatalf("status = %q, want %q", status, StatusDelivered)
|
||||
}
|
||||
|
||||
// Postgres' JSONB-Spalte kann die Byte-Repraesentation des Payloads
|
||||
// gegenueber dem urspruenglichen json.Marshal kanonisieren (z.B.
|
||||
// Leerzeichen) — das ist unschaedlich, denn der Dispatcher signiert
|
||||
// IMMER exakt die Bytes, die er auch sendet. Die Pruefung vergleicht
|
||||
// deshalb Signatur gegen tatsaechlich empfangene Bytes (Selbstkonsistenz),
|
||||
// nicht gegen eine unabhaengig neu marshalte Referenz.
|
||||
if !VerifySignature("geheimes-secret", receivedBody, receivedSignature) {
|
||||
t.Fatalf("empfangene signatur %q passt nicht zum empfangenen payload %q", receivedSignature, receivedBody)
|
||||
}
|
||||
var decoded map[string]string
|
||||
if err := json.Unmarshal(receivedBody, &decoded); err != nil {
|
||||
t.Fatalf("empfangener payload nicht als json lesbar: %v", err)
|
||||
}
|
||||
if decoded["file_id"] != "42" {
|
||||
t.Fatalf("empfangener payload = %v, want file_id=42", decoded)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: fehlschlagendes Ziel loest Retry mit
|
||||
// wachsendem Backoff aus und endet nach der konfigurierten Obergrenze in
|
||||
// "failed".
|
||||
func TestProcessDue_RetriesWithBackoffThenMarksFailed(t *testing.T) {
|
||||
store, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
var callCount int32
|
||||
target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
atomic.AddInt32(&callCount, 1)
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
}))
|
||||
defer target.Close()
|
||||
|
||||
eventType := uniqueEventType("mail.send.failed")
|
||||
sub, err := store.Subscribe(ctx, eventType, target.URL, "secret")
|
||||
if err != nil {
|
||||
t.Fatalf("subscribe: %v", err)
|
||||
}
|
||||
if _, err := store.Enqueue(ctx, eventType, map[string]string{"x": "y"}); err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
|
||||
dispatcher := NewDispatcher(pool, target.Client())
|
||||
dispatcher.MaxAttempts = 2
|
||||
dispatcher.BaseBackoff = 30 * time.Millisecond
|
||||
|
||||
// 1. Versuch: schlaegt fehl, ist aber noch nicht die Obergrenze.
|
||||
if _, err := dispatcher.ProcessDue(ctx); err != nil {
|
||||
t.Fatalf("processdue 1: %v", err)
|
||||
}
|
||||
var status string
|
||||
var attempt int
|
||||
if err := pool.QueryRow(ctx, `SELECT status, attempt FROM webhook_deliveries WHERE subscription_id = $1`, sub.ID).Scan(&status, &attempt); err != nil {
|
||||
t.Fatalf("status lesen 1: %v", err)
|
||||
}
|
||||
if status != StatusPending || attempt != 1 {
|
||||
t.Fatalf("nach 1. fehlschlag: status=%q attempt=%d, want pending/1", status, attempt)
|
||||
}
|
||||
|
||||
// Sofort erneut verarbeiten: Backoff ist noch nicht abgelaufen -> nichts faellig.
|
||||
processedTooEarly, err := dispatcher.ProcessDue(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("processdue (zu frueh): %v", err)
|
||||
}
|
||||
if processedTooEarly != 0 {
|
||||
t.Fatal("erwartet 0 verarbeitete zustellungen, solange backoff nicht abgelaufen ist")
|
||||
}
|
||||
|
||||
time.Sleep(dispatcher.backoffFor(1) + 20*time.Millisecond)
|
||||
|
||||
// 2. Versuch: erreicht MaxAttempts=2 -> endgueltig fehlgeschlagen.
|
||||
if _, err := dispatcher.ProcessDue(ctx); err != nil {
|
||||
t.Fatalf("processdue 2: %v", err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT status, attempt FROM webhook_deliveries WHERE subscription_id = $1`, sub.ID).Scan(&status, &attempt); err != nil {
|
||||
t.Fatalf("status lesen 2: %v", err)
|
||||
}
|
||||
if status != StatusFailed || attempt != 2 {
|
||||
t.Fatalf("nach 2. fehlschlag: status=%q attempt=%d, want failed/2", status, attempt)
|
||||
}
|
||||
if atomic.LoadInt32(&callCount) != 2 {
|
||||
t.Fatalf("erwartet genau 2 tatsaechliche zustellversuche, habe %d", callCount)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 + Pruefung 3: Signaturpruefung erkennt eine
|
||||
// manipulierte Payload zuverlaessig.
|
||||
func TestVerifySignature_DetectsTamperedPayload(t *testing.T) {
|
||||
secret := "geteiltes-geheimnis"
|
||||
payload := []byte(`{"file_id":"42"}`)
|
||||
signature := Sign(secret, payload)
|
||||
|
||||
if !VerifySignature(secret, payload, signature) {
|
||||
t.Fatal("erwartet gueltige signatur fuer unveraenderte payload")
|
||||
}
|
||||
|
||||
tampered := []byte(`{"file_id":"99"}`)
|
||||
if VerifySignature(secret, tampered, signature) {
|
||||
t.Fatal("erwartet ungueltige signatur fuer manipulierte payload")
|
||||
}
|
||||
|
||||
if VerifySignature("falsches-secret", payload, signature) {
|
||||
t.Fatal("erwartet ungueltige signatur bei falschem secret")
|
||||
}
|
||||
}
|
||||
@@ -1,2 +0,0 @@
|
||||
DROP TABLE webhook_deliveries;
|
||||
DROP TABLE webhook_subscriptions;
|
||||
@@ -1,27 +0,0 @@
|
||||
-- Zentrale Webhook-Registry & Zustellung (API-07, siehe
|
||||
-- core-kanban/tickets/API-07.md) — Postgres-basierte Jobqueue, kein
|
||||
-- Redis/AMQP (Ticket-Vorgabe).
|
||||
CREATE TABLE webhook_subscriptions (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
event_type TEXT NOT NULL,
|
||||
target_url TEXT NOT NULL,
|
||||
secret TEXT NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX webhook_subscriptions_event_type_idx ON webhook_subscriptions (event_type);
|
||||
|
||||
CREATE TABLE webhook_deliveries (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
subscription_id UUID NOT NULL REFERENCES webhook_subscriptions(id),
|
||||
event_type TEXT NOT NULL,
|
||||
payload JSONB NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'pending', -- pending|delivered|failed
|
||||
attempt INT NOT NULL DEFAULT 0,
|
||||
next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
last_error TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
delivered_at TIMESTAMPTZ
|
||||
);
|
||||
|
||||
CREATE INDEX webhook_deliveries_due_idx ON webhook_deliveries (status, next_attempt_at);
|
||||
Reference in New Issue
Block a user