Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
95275407b1 |
@@ -1,20 +0,0 @@
|
|||||||
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
|
|
||||||
}
|
|
||||||
@@ -1,31 +0,0 @@
|
|||||||
// 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)
|
|
||||||
}
|
|
||||||
@@ -1,56 +0,0 @@
|
|||||||
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))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,37 +0,0 @@
|
|||||||
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
|
|
||||||
}
|
|
||||||
@@ -1,149 +0,0 @@
|
|||||||
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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -0,0 +1,151 @@
|
|||||||
|
// 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[:]
|
||||||
|
}
|
||||||
@@ -0,0 +1,163 @@
|
|||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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);
|
|
||||||
@@ -0,0 +1,2 @@
|
|||||||
|
DROP TABLE IF EXISTS api_tokens;
|
||||||
|
DROP TABLE IF EXISTS service_accounts;
|
||||||
@@ -0,0 +1,20 @@
|
|||||||
|
-- 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,16 +1,11 @@
|
|||||||
#!/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,18 +1,12 @@
|
|||||||
#!/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