Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c4840ca55b | ||
|
|
34a705390c | ||
|
|
627961b972 | ||
|
|
27f9866264 |
@@ -0,0 +1,20 @@
|
|||||||
|
package apiserver
|
||||||
|
|
||||||
|
import "context"
|
||||||
|
|
||||||
|
type contextKey int
|
||||||
|
|
||||||
|
const requestContextKey contextKey = iota
|
||||||
|
|
||||||
|
// RequestContext ist der Tenant-/Benutzerkontext, den die Middleware-Kette
|
||||||
|
// fuer nachgelagerte Handler bereitstellt (Akzeptanzkriterium 3).
|
||||||
|
type RequestContext struct {
|
||||||
|
UserID string
|
||||||
|
TenantSlug string
|
||||||
|
}
|
||||||
|
|
||||||
|
// FromContext liest den von der Middleware gesetzten Kontext.
|
||||||
|
func FromContext(ctx context.Context) (RequestContext, bool) {
|
||||||
|
rc, ok := ctx.Value(requestContextKey).(RequestContext)
|
||||||
|
return rc, ok
|
||||||
|
}
|
||||||
@@ -0,0 +1,31 @@
|
|||||||
|
// Package apiserver implementiert Core API-01: das REST-Grundgerüst mit
|
||||||
|
// URL-Versionierung, einheitlichem Fehlerformat und Middleware-Kette
|
||||||
|
// (Auth, Tenant-/Benutzerkontext, Logging).
|
||||||
|
package apiserver
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"net/http"
|
||||||
|
)
|
||||||
|
|
||||||
|
// errorBody ist das EINE Fehlerschema fuer alle Endpunkte unter /api/{version}/
|
||||||
|
// (Akzeptanzkriterium 2).
|
||||||
|
type errorBody struct {
|
||||||
|
Error struct {
|
||||||
|
Code string `json:"code"`
|
||||||
|
Message string `json:"message"`
|
||||||
|
} `json:"error"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// WriteError schreibt einen Fehler im einheitlichen Schema. code ist ein
|
||||||
|
// stabiler, maschinenlesbarer Bezeichner (z.B. "unauthenticated"), message
|
||||||
|
// ein fuer Menschen lesbarer deutscher Text.
|
||||||
|
func WriteError(w http.ResponseWriter, status int, code, message string) {
|
||||||
|
var body errorBody
|
||||||
|
body.Error.Code = code
|
||||||
|
body.Error.Message = message
|
||||||
|
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(status)
|
||||||
|
_ = json.NewEncoder(w).Encode(body)
|
||||||
|
}
|
||||||
@@ -0,0 +1,56 @@
|
|||||||
|
package apiserver
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"log/slog"
|
||||||
|
"net/http"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.perlbach24.de/scripte/nexarch/internal/auth"
|
||||||
|
)
|
||||||
|
|
||||||
|
// authAndTenantContext prueft die Session (wiederverwendet auth.TokenIssuer.Verify
|
||||||
|
// aus IAM-02 — keine zweite JWT-Implementierung) und setzt bei Erfolg
|
||||||
|
// RequestContext fuer nachgelagerte Handler (Akzeptanzkriterium 3). Anders
|
||||||
|
// als auth.RequireAuth (Klartext-Fehler) antwortet diese Middleware im
|
||||||
|
// einheitlichen API-01-Fehlerschema (Akzeptanzkriterium 2), damit ALLE
|
||||||
|
// Endpunkte unter /api/{version}/ dasselbe Format liefern, auch bei
|
||||||
|
// Auth-Fehlern.
|
||||||
|
func authAndTenantContext(issuer *auth.TokenIssuer, next http.HandlerFunc) http.HandlerFunc {
|
||||||
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
cookie, err := r.Cookie(auth.CookieName)
|
||||||
|
if err != nil {
|
||||||
|
WriteError(w, http.StatusUnauthorized, "unauthenticated", "nicht angemeldet")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
claims, err := issuer.Verify(cookie.Value)
|
||||||
|
if err != nil {
|
||||||
|
WriteError(w, http.StatusUnauthorized, "unauthenticated", "nicht angemeldet")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
rc := RequestContext{UserID: claims.UserID, TenantSlug: claims.TenantSlug}
|
||||||
|
next(w, r.WithContext(context.WithValue(r.Context(), requestContextKey, rc)))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type statusRecorder struct {
|
||||||
|
http.ResponseWriter
|
||||||
|
status int
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *statusRecorder) WriteHeader(code int) {
|
||||||
|
s.status = code
|
||||||
|
s.ResponseWriter.WriteHeader(code)
|
||||||
|
}
|
||||||
|
|
||||||
|
// loggingMiddleware protokolliert jede Anfrage strukturiert.
|
||||||
|
func loggingMiddleware(next http.HandlerFunc) http.HandlerFunc {
|
||||||
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
|
||||||
|
start := time.Now()
|
||||||
|
next(rec, r)
|
||||||
|
slog.Info("api-anfrage", "method", r.Method, "path", r.URL.Path, "status", rec.status, "dauer", time.Since(start))
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,37 @@
|
|||||||
|
package apiserver
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/http"
|
||||||
|
|
||||||
|
"gitea.perlbach24.de/scripte/nexarch/internal/auth"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Server registriert versionierte API-Routen (Akzeptanzkriterium 1: unter
|
||||||
|
// /api/{version}/...) und verdrahtet fuer jede Route dieselbe Middleware-
|
||||||
|
// Kette (Logging -> Auth+Tenantkontext -> Handler).
|
||||||
|
type Server struct {
|
||||||
|
mux *http.ServeMux
|
||||||
|
issuer *auth.TokenIssuer
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewServer(issuer *auth.TokenIssuer) *Server {
|
||||||
|
return &Server{mux: http.NewServeMux(), issuer: issuer}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Handle registriert pattern unter der angegebenen Version, z.B.
|
||||||
|
// Handle("v1", "/things", h) -> erreichbar unter /api/v1/things. Verschiedene
|
||||||
|
// Versionen sind unabhaengige Pfade — eine neue Version beeintraechtigt
|
||||||
|
// bestehende nicht (Akzeptanzkriterium 1 / Pruefung 3).
|
||||||
|
func (s *Server) Handle(version, pattern string, h http.HandlerFunc) {
|
||||||
|
full := "/api/" + version + pattern
|
||||||
|
s.mux.HandleFunc(full, loggingMiddleware(authAndTenantContext(s.issuer, h)))
|
||||||
|
}
|
||||||
|
|
||||||
|
// HandleV1 ist die Kurzform fuer die aktuelle Hauptversion.
|
||||||
|
func (s *Server) HandleV1(pattern string, h http.HandlerFunc) {
|
||||||
|
s.Handle("v1", pattern, h)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) Handler() http.Handler {
|
||||||
|
return s.mux
|
||||||
|
}
|
||||||
@@ -0,0 +1,149 @@
|
|||||||
|
package apiserver
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"gitea.perlbach24.de/scripte/nexarch/internal/auth"
|
||||||
|
)
|
||||||
|
|
||||||
|
func newTestServer() (*Server, *auth.TokenIssuer) {
|
||||||
|
issuer := auth.NewTokenIssuer("test-secret-nur-fuer-tests")
|
||||||
|
return NewServer(issuer), issuer
|
||||||
|
}
|
||||||
|
|
||||||
|
func withAuthCookie(req *http.Request, token string) *http.Request {
|
||||||
|
req.AddCookie(&http.Cookie{Name: auth.CookieName, Value: token})
|
||||||
|
return req
|
||||||
|
}
|
||||||
|
|
||||||
|
// Akzeptanzkriterium 1: API unter versioniertem Pfad erreichbar.
|
||||||
|
func TestHandleV1_RegistersUnderVersionedPath(t *testing.T) {
|
||||||
|
srv, issuer := newTestServer()
|
||||||
|
srv.HandleV1("/things", func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
})
|
||||||
|
|
||||||
|
token, err := issuer.Issue("user-1", "acme")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("issue: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
req := withAuthCookie(httptest.NewRequest(http.MethodGet, "/api/v1/things", nil), token)
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
srv.Handler().ServeHTTP(rec, req)
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
t.Fatalf("status = %d, want 200", rec.Code)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Akzeptanzkriterium 2 + Pruefung 1 (Stichprobe): mehrere Endpunkte liefern
|
||||||
|
// bei fehlerhafter Anfrage dasselbe Fehlerschema.
|
||||||
|
func TestErrorFormat_ConsistentAcrossEndpoints(t *testing.T) {
|
||||||
|
srv, _ := newTestServer()
|
||||||
|
srv.HandleV1("/endpunkt-a", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
|
||||||
|
srv.HandleV1("/endpunkt-b", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
|
||||||
|
|
||||||
|
for _, path := range []string{"/api/v1/endpunkt-a", "/api/v1/endpunkt-b"} {
|
||||||
|
req := httptest.NewRequest(http.MethodGet, path, nil) // ohne cookie -> 401
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
srv.Handler().ServeHTTP(rec, req)
|
||||||
|
|
||||||
|
if rec.Code != http.StatusUnauthorized {
|
||||||
|
t.Fatalf("%s: status = %d, want 401", path, rec.Code)
|
||||||
|
}
|
||||||
|
var body errorBody
|
||||||
|
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
|
||||||
|
t.Fatalf("%s: fehlerantwort nicht im erwarteten json-schema: %v (body: %s)", path, err, rec.Body.String())
|
||||||
|
}
|
||||||
|
if body.Error.Code == "" || body.Error.Message == "" {
|
||||||
|
t.Fatalf("%s: erwartet nicht-leeren code/message, habe %+v", path, body)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Akzeptanzkriterium 3 + Pruefung 2: Middleware-Kette setzt Tenant-/
|
||||||
|
// Benutzerkontext zuverlaessig, nachweislich fuer mehrere Endpunkte.
|
||||||
|
func TestMiddleware_SetsRequestContextForEveryEndpoint(t *testing.T) {
|
||||||
|
srv, issuer := newTestServer()
|
||||||
|
|
||||||
|
var gotA, gotB RequestContext
|
||||||
|
srv.HandleV1("/kontext-a", func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
gotA, _ = FromContext(r.Context())
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
})
|
||||||
|
srv.HandleV1("/kontext-b", func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
gotB, _ = FromContext(r.Context())
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
})
|
||||||
|
|
||||||
|
token, err := issuer.Issue("user-42", "tenant-x")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("issue: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for path, got := range map[string]*RequestContext{"/api/v1/kontext-a": &gotA, "/api/v1/kontext-b": &gotB} {
|
||||||
|
req := withAuthCookie(httptest.NewRequest(http.MethodGet, path, nil), token)
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
srv.Handler().ServeHTTP(rec, req)
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
t.Fatalf("%s: status = %d, want 200", path, rec.Code)
|
||||||
|
}
|
||||||
|
if got.UserID != "user-42" || got.TenantSlug != "tenant-x" {
|
||||||
|
t.Fatalf("%s: request-context unerwartet: %+v", path, *got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Akzeptanzkriterium 1 + Pruefung 3: eine neue v2-Route laesst sich anlegen,
|
||||||
|
// ohne v1 zu beeintraechtigen.
|
||||||
|
func TestVersioning_V2DoesNotAffectV1(t *testing.T) {
|
||||||
|
srv, issuer := newTestServer()
|
||||||
|
srv.HandleV1("/things", func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Write([]byte("v1-antwort"))
|
||||||
|
})
|
||||||
|
srv.Handle("v2", "/things", func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Write([]byte("v2-antwort"))
|
||||||
|
})
|
||||||
|
|
||||||
|
token, err := issuer.Issue("user-1", "acme")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("issue: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
reqV1 := withAuthCookie(httptest.NewRequest(http.MethodGet, "/api/v1/things", nil), token)
|
||||||
|
recV1 := httptest.NewRecorder()
|
||||||
|
srv.Handler().ServeHTTP(recV1, reqV1)
|
||||||
|
if recV1.Body.String() != "v1-antwort" {
|
||||||
|
t.Fatalf("v1 antwort = %q, want v1-antwort", recV1.Body.String())
|
||||||
|
}
|
||||||
|
|
||||||
|
reqV2 := withAuthCookie(httptest.NewRequest(http.MethodGet, "/api/v2/things", nil), token)
|
||||||
|
recV2 := httptest.NewRecorder()
|
||||||
|
srv.Handler().ServeHTTP(recV2, reqV2)
|
||||||
|
if recV2.Body.String() != "v2-antwort" {
|
||||||
|
t.Fatalf("v2 antwort = %q, want v2-antwort", recV2.Body.String())
|
||||||
|
}
|
||||||
|
|
||||||
|
// v1 nach dem Anlegen von v2 erneut pruefen — unveraendert.
|
||||||
|
reqV1Again := withAuthCookie(httptest.NewRequest(http.MethodGet, "/api/v1/things", nil), token)
|
||||||
|
recV1Again := httptest.NewRecorder()
|
||||||
|
srv.Handler().ServeHTTP(recV1Again, reqV1Again)
|
||||||
|
if recV1Again.Body.String() != "v1-antwort" {
|
||||||
|
t.Fatalf("v1 antwort nach v2-anlage = %q, want weiterhin v1-antwort", recV1Again.Body.String())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestAuthAndTenantContext_RejectsInvalidToken(t *testing.T) {
|
||||||
|
srv, _ := newTestServer()
|
||||||
|
srv.HandleV1("/geschuetzt", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
|
||||||
|
|
||||||
|
req := withAuthCookie(httptest.NewRequest(http.MethodGet, "/api/v1/geschuetzt", nil), "kaputtes.token.hier")
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
srv.Handler().ServeHTTP(rec, req)
|
||||||
|
if rec.Code != http.StatusUnauthorized {
|
||||||
|
t.Fatalf("status = %d, want 401", rec.Code)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,151 +0,0 @@
|
|||||||
// Package serviceaccount implementiert Core IAM-09: Service-Accounts als
|
|
||||||
// eigenstaendige Identitaeten (getrennt von menschlichen Benutzern, siehe
|
|
||||||
// internal/user) sowie API-Token mit Scopes und optionaler Ablaufzeit zur
|
|
||||||
// programmatischen Authentifizierung.
|
|
||||||
package serviceaccount
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"crypto/rand"
|
|
||||||
"crypto/sha256"
|
|
||||||
"encoding/hex"
|
|
||||||
"errors"
|
|
||||||
"fmt"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/jackc/pgx/v5"
|
|
||||||
"github.com/jackc/pgx/v5/pgxpool"
|
|
||||||
)
|
|
||||||
|
|
||||||
var (
|
|
||||||
ErrTokenInvalid = errors.New("serviceaccount: token ungueltig, widerrufen oder abgelaufen")
|
|
||||||
ErrScopeInsufficient = errors.New("serviceaccount: token hat nicht den erforderlichen scope")
|
|
||||||
ErrServiceAccountNotFound = errors.New("serviceaccount: nicht gefunden")
|
|
||||||
)
|
|
||||||
|
|
||||||
type ServiceAccount struct {
|
|
||||||
ID string
|
|
||||||
Name string
|
|
||||||
}
|
|
||||||
|
|
||||||
type Token struct {
|
|
||||||
ID string
|
|
||||||
ServiceAccountID string
|
|
||||||
Scopes []string
|
|
||||||
ExpiresAt *time.Time
|
|
||||||
}
|
|
||||||
|
|
||||||
type Store struct {
|
|
||||||
pool *pgxpool.Pool
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewStore(pool *pgxpool.Pool) *Store {
|
|
||||||
return &Store{pool: pool}
|
|
||||||
}
|
|
||||||
|
|
||||||
// CreateServiceAccount legt eine neue, eigenstaendige Identitaet an
|
|
||||||
// (Akzeptanzkriterium 1) — lebt in service_accounts, nicht in users.
|
|
||||||
func (s *Store) CreateServiceAccount(ctx context.Context, name string) (ServiceAccount, error) {
|
|
||||||
if name == "" {
|
|
||||||
return ServiceAccount{}, errors.New("serviceaccount: name darf nicht leer sein")
|
|
||||||
}
|
|
||||||
var sa ServiceAccount
|
|
||||||
sa.Name = name
|
|
||||||
if err := s.pool.QueryRow(ctx, `
|
|
||||||
INSERT INTO service_accounts (name) VALUES ($1) RETURNING id
|
|
||||||
`, name).Scan(&sa.ID); err != nil {
|
|
||||||
return ServiceAccount{}, fmt.Errorf("service-account anlegen: %w", err)
|
|
||||||
}
|
|
||||||
return sa, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// IssueToken erzeugt ein neues API-Token fuer einen Service-Account.
|
|
||||||
// scopes beschraenkt, wofuer das Token verwendet werden darf; ttl ist
|
|
||||||
// optional (nil = kein Ablauf) — beides Akzeptanzkriterium 2. Der
|
|
||||||
// Klartext-Token wird NUR hier zurueckgegeben, gespeichert wird ausschliesslich
|
|
||||||
// der Hash (Akzeptanzkriterium 1 / Pruefung 1).
|
|
||||||
func (s *Store) IssueToken(ctx context.Context, serviceAccountID string, scopes []string, ttl *time.Duration) (id, plaintext string, err error) {
|
|
||||||
plaintext, err = randomToken()
|
|
||||||
if err != nil {
|
|
||||||
return "", "", fmt.Errorf("token erzeugen: %w", err)
|
|
||||||
}
|
|
||||||
hash := hashToken(plaintext)
|
|
||||||
if scopes == nil {
|
|
||||||
scopes = []string{}
|
|
||||||
}
|
|
||||||
|
|
||||||
var expiresAt *time.Time
|
|
||||||
if ttl != nil {
|
|
||||||
t := time.Now().Add(*ttl)
|
|
||||||
expiresAt = &t
|
|
||||||
}
|
|
||||||
|
|
||||||
err = s.pool.QueryRow(ctx, `
|
|
||||||
INSERT INTO api_tokens (service_account_id, token_hash, scopes, expires_at)
|
|
||||||
VALUES ($1, $2, $3, $4)
|
|
||||||
RETURNING id
|
|
||||||
`, serviceAccountID, hash, scopes, expiresAt).Scan(&id)
|
|
||||||
if err != nil {
|
|
||||||
return "", "", fmt.Errorf("token speichern: %w", err)
|
|
||||||
}
|
|
||||||
return id, plaintext, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Verify prueft ein Token gegen einen benoetigten Scope. Widerruf und Ablauf
|
|
||||||
// werden bei JEDEM Aufruf direkt gegen die Datenbank geprueft — es gibt
|
|
||||||
// keinen Cache, ein widerrufenes Token wird beim naechsten Request sofort
|
|
||||||
// abgewiesen (Akzeptanzkriterium 3). requiredScope == "" ueberspringt die
|
|
||||||
// Scope-Pruefung.
|
|
||||||
func (s *Store) Verify(ctx context.Context, tokenString, requiredScope string) (Token, error) {
|
|
||||||
hash := hashToken(tokenString)
|
|
||||||
|
|
||||||
var tok Token
|
|
||||||
err := s.pool.QueryRow(ctx, `
|
|
||||||
SELECT id, service_account_id, scopes, expires_at
|
|
||||||
FROM api_tokens
|
|
||||||
WHERE token_hash = $1 AND revoked_at IS NULL AND (expires_at IS NULL OR expires_at > now())
|
|
||||||
`, hash).Scan(&tok.ID, &tok.ServiceAccountID, &tok.Scopes, &tok.ExpiresAt)
|
|
||||||
if err != nil {
|
|
||||||
if errors.Is(err, pgx.ErrNoRows) {
|
|
||||||
return Token{}, ErrTokenInvalid
|
|
||||||
}
|
|
||||||
return Token{}, fmt.Errorf("token pruefen: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if requiredScope != "" && !hasScope(tok.Scopes, requiredScope) {
|
|
||||||
return Token{}, ErrScopeInsufficient
|
|
||||||
}
|
|
||||||
return tok, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Revoke widerruft ein Token sofort — Verify schlaegt ab dem naechsten
|
|
||||||
// Aufruf fehl (Akzeptanzkriterium 3 / Pruefung 2).
|
|
||||||
func (s *Store) Revoke(ctx context.Context, tokenID string) error {
|
|
||||||
_, err := s.pool.Exec(ctx, `UPDATE api_tokens SET revoked_at = now() WHERE id = $1`, tokenID)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("token widerrufen: %w", err)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func hasScope(scopes []string, want string) bool {
|
|
||||||
for _, s := range scopes {
|
|
||||||
if s == want {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
func randomToken() (string, error) {
|
|
||||||
buf := make([]byte, 32)
|
|
||||||
if _, err := rand.Read(buf); err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
return hex.EncodeToString(buf), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func hashToken(token string) []byte {
|
|
||||||
sum := sha256.Sum256([]byte(token))
|
|
||||||
return sum[:]
|
|
||||||
}
|
|
||||||
@@ -1,163 +0,0 @@
|
|||||||
package serviceaccount
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"errors"
|
|
||||||
"os"
|
|
||||||
"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 service_accounts (
|
|
||||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name TEXT NOT NULL,
|
|
||||||
status TEXT NOT NULL DEFAULT 'active', created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
|
||||||
);
|
|
||||||
CREATE TABLE IF NOT EXISTS api_tokens (
|
|
||||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), service_account_id UUID NOT NULL REFERENCES service_accounts(id),
|
|
||||||
token_hash BYTEA NOT NULL UNIQUE, scopes TEXT[] NOT NULL DEFAULT '{}',
|
|
||||||
expires_at TIMESTAMPTZ, revoked_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
|
||||||
);
|
|
||||||
`); err != nil {
|
|
||||||
t.Fatalf("schema: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
cleanup := func() { pool.Close() }
|
|
||||||
return NewStore(pool), pool, cleanup
|
|
||||||
}
|
|
||||||
|
|
||||||
// Akzeptanzkriterium 1: Service-Account anlegen, eigenes gehashtes Token erhalten.
|
|
||||||
// Akzeptanzkriterium 1 + Pruefung 1: Token ausschliesslich gehasht gespeichert.
|
|
||||||
func TestIssueToken_StoresOnlyHash(t *testing.T) {
|
|
||||||
store, pool, cleanup := setupTest(t)
|
|
||||||
defer cleanup()
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
sa, err := store.CreateServiceAccount(ctx, "ci-pipeline")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("create service account: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
tokenID, plaintext, err := store.IssueToken(ctx, sa.ID, []string{"read"}, nil)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("issue token: %v", err)
|
|
||||||
}
|
|
||||||
if plaintext == "" {
|
|
||||||
t.Fatal("erwartet nicht-leeren klartext-token")
|
|
||||||
}
|
|
||||||
|
|
||||||
var storedHash []byte
|
|
||||||
if err := pool.QueryRow(ctx, `SELECT token_hash FROM api_tokens WHERE id = $1`, tokenID).Scan(&storedHash); err != nil {
|
|
||||||
t.Fatalf("hash lesen: %v", err)
|
|
||||||
}
|
|
||||||
if string(storedHash) == plaintext {
|
|
||||||
t.Fatal("token_hash enthaelt den klartext-token — verstoss gegen akzeptanzkriterium 1")
|
|
||||||
}
|
|
||||||
if len(storedHash) != 32 { // sha256
|
|
||||||
t.Fatalf("erwartet 32-byte sha256-hash, habe %d bytes", len(storedHash))
|
|
||||||
}
|
|
||||||
|
|
||||||
tok, err := store.Verify(ctx, plaintext, "")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("verify: %v", err)
|
|
||||||
}
|
|
||||||
if tok.ServiceAccountID != sa.ID {
|
|
||||||
t.Fatalf("service account id = %q, want %q", tok.ServiceAccountID, sa.ID)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Akzeptanzkriterium 2 + Pruefung 3: Scope-Verletzung wird abgewiesen.
|
|
||||||
func TestVerify_RejectsInsufficientScope(t *testing.T) {
|
|
||||||
store, _, cleanup := setupTest(t)
|
|
||||||
defer cleanup()
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
sa, err := store.CreateServiceAccount(ctx, "readonly-bot")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("create service account: %v", err)
|
|
||||||
}
|
|
||||||
_, plaintext, err := store.IssueToken(ctx, sa.ID, []string{"read"}, nil)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("issue token: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if _, err := store.Verify(ctx, plaintext, "read"); err != nil {
|
|
||||||
t.Fatalf("erwartet erfolg fuer passenden scope: %v", err)
|
|
||||||
}
|
|
||||||
if _, err := store.Verify(ctx, plaintext, "write"); !errors.Is(err, ErrScopeInsufficient) {
|
|
||||||
t.Fatalf("erwartet ErrScopeInsufficient fuer fehlenden scope, habe %v", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Akzeptanzkriterium 2: optionale zeitliche Befristung wird durchgesetzt.
|
|
||||||
func TestVerify_RejectsExpiredToken(t *testing.T) {
|
|
||||||
store, _, cleanup := setupTest(t)
|
|
||||||
defer cleanup()
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
sa, err := store.CreateServiceAccount(ctx, "temp-bot")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("create service account: %v", err)
|
|
||||||
}
|
|
||||||
ttl := -1 * time.Second // sofort abgelaufen
|
|
||||||
_, plaintext, err := store.IssueToken(ctx, sa.ID, nil, &ttl)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("issue token: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if _, err := store.Verify(ctx, plaintext, ""); !errors.Is(err, ErrTokenInvalid) {
|
|
||||||
t.Fatalf("erwartet ErrTokenInvalid bei abgelaufenem token, habe %v", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Akzeptanzkriterium 3 + Pruefung 2: widerrufenes Token wird sofort abgewiesen.
|
|
||||||
func TestRevoke_TakesEffectImmediately(t *testing.T) {
|
|
||||||
store, _, cleanup := setupTest(t)
|
|
||||||
defer cleanup()
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
sa, err := store.CreateServiceAccount(ctx, "revoke-bot")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("create service account: %v", err)
|
|
||||||
}
|
|
||||||
tokenID, plaintext, err := store.IssueToken(ctx, sa.ID, nil, nil)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("issue token: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if _, err := store.Verify(ctx, plaintext, ""); err != nil {
|
|
||||||
t.Fatalf("verify vor widerruf: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := store.Revoke(ctx, tokenID); err != nil {
|
|
||||||
t.Fatalf("revoke: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if _, err := store.Verify(ctx, plaintext, ""); !errors.Is(err, ErrTokenInvalid) {
|
|
||||||
t.Fatalf("erwartet ErrTokenInvalid sofort nach widerruf, habe %v", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestVerify_RejectsUnknownToken(t *testing.T) {
|
|
||||||
store, _, cleanup := setupTest(t)
|
|
||||||
defer cleanup()
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
if _, err := store.Verify(ctx, "niemals-ausgestelltes-token", ""); !errors.Is(err, ErrTokenInvalid) {
|
|
||||||
t.Fatalf("erwartet ErrTokenInvalid, habe %v", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -0,0 +1,177 @@
|
|||||||
|
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
|
||||||
|
}
|
||||||
@@ -0,0 +1,131 @@
|
|||||||
|
// 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"
|
||||||
@@ -0,0 +1,205 @@
|
|||||||
|
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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,2 @@
|
|||||||
|
DROP TABLE webhook_deliveries;
|
||||||
|
DROP TABLE webhook_subscriptions;
|
||||||
@@ -0,0 +1,27 @@
|
|||||||
|
-- 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);
|
||||||
@@ -1,2 +0,0 @@
|
|||||||
DROP TABLE IF EXISTS api_tokens;
|
|
||||||
DROP TABLE IF EXISTS service_accounts;
|
|
||||||
@@ -1,20 +0,0 @@
|
|||||||
-- Service-Accounts & API-Token (IAM-09, siehe core-kanban/tickets/IAM-09.md).
|
|
||||||
-- Service-Accounts sind eine eigenstaendige Identitaetsklasse neben
|
|
||||||
-- menschlichen Benutzern (users), nicht dieselbe Tabelle (Zitadel-Vorbild).
|
|
||||||
CREATE TABLE service_accounts (
|
|
||||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
|
||||||
name TEXT NOT NULL,
|
|
||||||
status TEXT NOT NULL DEFAULT 'active',
|
|
||||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
|
||||||
);
|
|
||||||
|
|
||||||
-- token_hash enthaelt NIEMALS den Klartext-Token, nur dessen SHA-256-Hash.
|
|
||||||
CREATE TABLE api_tokens (
|
|
||||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
|
||||||
service_account_id UUID NOT NULL REFERENCES service_accounts(id),
|
|
||||||
token_hash BYTEA NOT NULL UNIQUE,
|
|
||||||
scopes TEXT[] NOT NULL DEFAULT '{}',
|
|
||||||
expires_at TIMESTAMPTZ,
|
|
||||||
revoked_at TIMESTAMPTZ,
|
|
||||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
|
||||||
);
|
|
||||||
@@ -1,11 +1,16 @@
|
|||||||
#!/usr/bin/env bash
|
#!/usr/bin/env bash
|
||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
|
|
||||||
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
|
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
|
||||||
ROLE="nexarch_test"
|
ROLE="nexarch_test"
|
||||||
|
|
||||||
export PGPASSWORD="$PASS"
|
export PGPASSWORD="$PASS"
|
||||||
|
|
||||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS tenants CASCADE;"
|
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS tenants CASCADE;"
|
||||||
|
|
||||||
dbs=$(psql -h localhost -U "$ROLE" -d postgres -tAc "SELECT datname FROM pg_database WHERE datname LIKE 'tenant\_%' ESCAPE '\'")
|
dbs=$(psql -h localhost -U "$ROLE" -d postgres -tAc "SELECT datname FROM pg_database WHERE datname LIKE 'tenant\_%' ESCAPE '\'")
|
||||||
for db in $dbs; do
|
for db in $dbs; do
|
||||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS \"${db}\";"
|
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS \"${db}\";"
|
||||||
done
|
done
|
||||||
|
|
||||||
echo "Testumgebung zurueckgesetzt: registry-tabelle + $(echo "$dbs" | grep -c . || true) tenant-datenbank(en) entfernt."
|
echo "Testumgebung zurueckgesetzt: registry-tabelle + $(echo "$dbs" | grep -c . || true) tenant-datenbank(en) entfernt."
|
||||||
|
|||||||
@@ -1,12 +1,18 @@
|
|||||||
#!/usr/bin/env bash
|
#!/usr/bin/env bash
|
||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
|
|
||||||
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
|
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
|
||||||
cd "$(dirname "$0")/.."
|
cd "$(dirname "$0")/.."
|
||||||
|
|
||||||
NEXARCH_TEST_DB_PASSWORD="$PASS" bash scripts/reset-test-env.sh
|
NEXARCH_TEST_DB_PASSWORD="$PASS" bash scripts/reset-test-env.sh
|
||||||
|
|
||||||
export TEST_ADMIN_DSN="postgresql://nexarch_test:${PASS}@localhost:5432/postgres?sslmode=disable"
|
export TEST_ADMIN_DSN="postgresql://nexarch_test:${PASS}@localhost:5432/postgres?sslmode=disable"
|
||||||
|
|
||||||
echo "== go build =="
|
echo "== go build =="
|
||||||
go build ./...
|
go build ./...
|
||||||
|
|
||||||
echo "== go vet =="
|
echo "== go vet =="
|
||||||
go vet ./...
|
go vet ./...
|
||||||
|
|
||||||
echo "== go test (-p 1) =="
|
echo "== go test (-p 1) =="
|
||||||
go test ./... -p 1 -count=1
|
go test ./... -p 1 -count=1
|
||||||
|
|||||||
Reference in New Issue
Block a user