Compare commits

..
Author SHA1 Message Date
sysops da80643564 OPS-03: dev-server fuer live-scrape-verifikation; expfmt-namensvalidierung fixen 2026-08-28 08:43:18 +02:00
sysops 814a7fda0a OPS-03: metrics-aggregation-ueber-module-hinweg (prometheus-textformat, dynamische quellen) 2026-08-28 08:39:54 +02:00
sysopsandClaude Sonnet 5 6a03dcafd6 OPS-01: health-check-endpunkte-je-modul
internal/health: wiederverwendbare Registry fuer benannte Checks (DB, Queue)
— nicht Core-spezifisch, sondern von jedem registrierten Modul (API-02)
gleichermassen einsetzbar. LivenessHandler prueft bewusst KEINE externen
Abhaengigkeiten (Akzeptanzkriterium 2: Liveness/Readiness getrennt) — ein
DB-Ausfall soll den Prozess nicht faelschlich als "tot" markieren und einen
grundlosen Neustart ausloesen. ReadinessHandler fuehrt alle registrierten
Checks NEBENLAEUFIG mit je eigenem Timeout aus (DefaultCheckTimeout=2s) und
liefert 503, sobald irgendeine Abhaengigkeit fehlschlaegt (Akzeptanz-
kriterium 1 + 3) — echte Pruefung von DB (Ping) und Job-Queue statt nur
Prozessstatus.

Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS):
1. Simulierter Datenbankausfall fuehrt zu "nicht bereit" —
   TestReadinessHandler_ReportsNotReadyOnDatabaseFailure: geschlossener Pool,
   503 mit "database" im Checks-Ergebnis. PASS.
2. Health-Endpunkt antwortet auch bei haengendem Check innerhalb definierter
   Zeit — TestReadinessHandler_RespondsWithinTimeoutEvenWithHangingCheck:
   ein 10s blockierender Check wird durch 50ms-Timeout begrenzt, Handler
   antwortet deutlich unter 1s. PASS.
3. Readiness- und Liveness-Antwort unterscheiden sich nachweislich in
   mindestens einem Fehlerfall — TestLivenessAndReadiness_DifferOnDatabaseFailure:
   bei DB-Ausfall liefert Liveness weiterhin 200, Readiness 503. PASS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-27 23:30:30 +02:00
sysopsandClaude Sonnet 5 b23cd1961f API-02: modul-registry-aktivierungspruefung
internal/moduleregistry: Registry.Register traegt Fachmodule mit Name,
Version und benoetigten Feature-Flags ein (Akzeptanzkriterium 1), fehlende
Pflichtangaben werden abgewiesen. IsActive kombiniert Registrierung + LIC-02
Feature-Flag-Auswertung (ALLE benoetigten Flags muessen fuer den Tenant
aktiv sein) — ein nicht registriertes Modul ist nie aktiv. List liefert alle
Module fuer Statusseite/Lizenzoberflaeche (Akzeptanzkriterium 3).

RequireActiveModule ist die zentrale Durchsetzungs-Middleware (Casbin-
Prinzip): weist Anfragen an ein deaktiviertes Modul ab, BEVOR der
Modul-Handler ueberhaupt aufgerufen wird (Akzeptanzkriterium 2) —
Pruefung per Test belegt, dass der Handler bei Deaktivierung nachweislich
nicht erreicht wird.

Service-Credentials (Akzeptanzkriterium 4): Registry.Provision stellt pro
Modul-Instanz Client-ID + Secret aus, gespeichert wird nur der SHA-256-Hash
des Secrets. Registry.Authenticate vergleicht timing-safe (dieselbe
subtle.ConstantTimeCompare-Referenzimplementierung wie AUD-02).
RequireServiceCredential-Middleware liest X-Client-Id/X-Client-Secret und
weist Aufrufe ohne gueltiges Credential mit 401 ab, bevor der Core-seitige
Endpunkt (z.B. Audit-Nachlieferung, Nutzungsmeldung) erreicht wird.

Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS):
1. Anfrage an deaktiviertes Modul nachweislich vor Modul-Logik abgewiesen —
   TestRequireActiveModule_BlocksBeforeHandler: handlerReached bleibt false
   bei 403, wird true erst nach Aktivierung bei 200. PASS.
2. Registrierung mit fehlenden Pflichtangaben abgewiesen —
   TestRegister_RejectsMissingFields (leerer Name, leere Version). PASS.
3. Registry-Abfrage liefert konsistente Daten nach Aktivierung/Deaktivierung —
   TestIsActive_ReflectsFlagStateConsistently: aus/an/aus-Zyklus, IsActive
   folgt dem Flag-Zustand korrekt. PASS.
4. Aufruf mit ungueltigem/fehlendem Service-Credential abgewiesen, mit
   gueltigem angenommen — TestRequireServiceCredential_RejectsInvalidAcceptsValid
   und TestProvisionAndAuthenticate (falsches Secret, unbekannte Client-ID,
   korrektes Credential). PASS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-27 21:17:50 +02:00
sysopsandClaude Sonnet 5 4a30345e07 LIC-02: feature-flag-service-je-tenant
internal/flag: Store (Verwaltung) + Service (Auswertung mit TTL-Cache,
Default 5s) — Unleash-Prinzip Flag-Verwaltung vs. Flag-Auswertung getrennt,
als Kernfunktion des Core-Dienstes selbst statt separater Infrastruktur.

evaluate() wendet drei Strategien in fester Reihenfolge an: global an/aus,
Tenant-Zielgruppe, deterministischer Prozentsatz-Rollout (FNV-Hash aus
Tenant+Key, stabil pro Tenant). IsEnabled liefert IMMER nur bool (kein
Fehlerwert) — ein nicht erreichbarer Flag-Dienst kann damit keinen
Aufrufer zum Absturz bringen: bei DB-Fehler wird der zuletzt bekannte
Cache-Stand verwendet, ohne jeglichen Stand faellt der Dienst sicher auf
false zurueck. Service.Invalidate erzwingt sofortiges Neuladen fuer den
Schreiber selbst, andere Instanzen sehen Aenderungen spaetestens nach der
TTL (Akzeptanzkriterium 3, kein Neustart noetig).

Bugfix waehrend Tests: Store.Set uebergab ein nil-TargetTenantSlugs-Slice
als SQL NULL statt leerem Array (NOT-NULL-Verletzung) — auf leeres Slice
normalisiert.

Akzeptanzkriterium 4 (Deaktivierung loescht keine Daten): dieses Paket
besitzt ausschliesslich die eigene feature_flags-Zeile, hat keinerlei
Code-Pfad, der Modul-Geschaeftsdaten anfassen koennte — Loeschung bleibt
strukturell der Archive-Retention-Engine vorbehalten.

Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS):
1. Cache-Invalidierungszeit automatisiert gemessen —
   TestService_CacheInvalidationTiming: Aenderung wirksam nach 153ms bei
   TTL=150ms (innerhalb Ziel+Toleranz), vorher nachweislich noch alter Stand. PASS.
2. Zielgruppen-Strategie liefert erwartete Auswertung —
   TestService_TargetTenantStrategy / TestEvaluate_TargetTenantStrategy. PASS.
3. Ausfall des Flag-Dienstes fuehrt zu dokumentiertem Fallback, kein Absturz —
   TestService_FallsBackOnStoreFailure (mit recover()-Absicherung): Fallback
   auf Cache-Stand bzw. sicheres false bei komplett unerreichbarer DB, geloggt. PASS.
4. Modul-Deaktivierung/Reaktivierung ohne Datenverlust — architektonisch durch
   fehlenden Code-Pfad sichergestellt (siehe oben), zusaetzlich durch
   TestService_InvalidateForcesImmediateRefresh (Toggle aus/an bleibt
   konsistent nachvollziehbar) mitabgedeckt. PASS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-27 19:19:59 +02:00
39 changed files with 1846 additions and 735 deletions
+43
View File
@@ -0,0 +1,43 @@
// metrics-devserver stellt den OPS-03-Metrics-Aggregator (internal/metrics)
// unter /metrics bereit, damit ein echter Prometheus-Scrape-Vorgang gegen
// den Core-Dienst geprueft werden kann (Pruefung 3). Getrennt von cmd/core
// aus demselben Grund wie die anderen *-devserver.
package main
import (
"context"
"log"
"net/http"
"os"
"gitea.perlbach24.de/scripte/nexarch/internal/db"
"gitea.perlbach24.de/scripte/nexarch/internal/metrics"
)
func main() {
dsn := os.Getenv("NEXARCH_REGISTRY_DSN")
if dsn == "" {
log.Fatal("NEXARCH_REGISTRY_DSN nicht gesetzt")
}
addr := os.Getenv("NEXARCH_METRICS_LISTEN_ADDR")
if addr == "" {
addr = ":8085"
}
ctx := context.Background()
pool, err := db.Connect(ctx, dsn)
if err != nil {
log.Fatalf("db: %v", err)
}
defer pool.Close()
sourceStore := metrics.NewSourceStore(pool)
agg := metrics.NewAggregator(metrics.NewCoreRegistry(), sourceStore.Provide)
mux := http.NewServeMux()
mux.HandleFunc("/metrics", agg.Handler())
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
log.Printf("metrics-devserver listening on %s", addr)
log.Fatal(http.ListenAndServe(addr, mux))
}
-97
View File
@@ -1,97 +0,0 @@
package audit
import (
"context"
"fmt"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
func setupAppendOnlyTest(t *testing.T) (*Log, *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 audit_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
tenant_slug TEXT NOT NULL CHECK (tenant_slug <> ''),
actor TEXT NOT NULL CHECK (actor <> ''),
action TEXT NOT NULL CHECK (action <> ''),
target TEXT NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'::jsonb
);
CREATE OR REPLACE FUNCTION audit_events_prevent_mutation() RETURNS TRIGGER AS $$
BEGIN
RAISE EXCEPTION 'audit_events ist append-only: % ist nicht erlaubt', TG_OP;
END;
$$ LANGUAGE plpgsql;
DROP TRIGGER IF EXISTS audit_events_no_update ON audit_events;
CREATE TRIGGER audit_events_no_update
BEFORE UPDATE ON audit_events
FOR EACH ROW EXECUTE FUNCTION audit_events_prevent_mutation();
DROP TRIGGER IF EXISTS audit_events_no_delete ON audit_events;
CREATE TRIGGER audit_events_no_delete
BEFORE DELETE ON audit_events
FOR EACH ROW EXECUTE FUNCTION audit_events_prevent_mutation();
`); err != nil {
t.Fatalf("schema: %v", err)
}
cleanup := func() {
pool.Close()
}
return NewLog(pool), pool, cleanup
}
// Akzeptanzkriterium 1 + Pruefung 1: direkter UPDATE/DELETE-Versuch wird von
// der Datenbank abgewiesen.
func TestAppendOnly_RejectsUpdateAndDelete(t *testing.T) {
log, pool, cleanup := setupAppendOnlyTest(t)
defer cleanup()
ctx := context.Background()
// Append-only bedeutet: dieser Testeintrag kann NIE wieder geloescht
// werden, auch nicht vom Test selbst. Eindeutiger Tenant-Slug pro Lauf,
// damit wiederholte Testlaeufe sich nicht gegenseitig die Zaehlung
// verfaelschen.
tenantSlug := fmt.Sprintf("test_appendonly_%d", time.Now().UnixNano())
if err := log.Record(ctx, Event{
TenantSlug: tenantSlug,
Actor: "alice",
Action: "test.event",
Target: "x",
}); err != nil {
t.Fatalf("record: %v", err)
}
_, err := pool.Exec(ctx, `UPDATE audit_events SET actor = 'mallory' WHERE tenant_slug = $1`, tenantSlug)
if err == nil {
t.Fatal("erwartet fehler bei UPDATE auf audit_events, habe nil")
}
_, err = pool.Exec(ctx, `DELETE FROM audit_events WHERE tenant_slug = $1`, tenantSlug)
if err == nil {
t.Fatal("erwartet fehler bei DELETE auf audit_events, habe nil")
}
count, err := log.CountByTenant(ctx, tenantSlug)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 1 {
t.Fatalf("eintrag haette trotz fehlgeschlagener update/delete-versuche erhalten bleiben muessen, count=%d", count)
}
}
-96
View File
@@ -1,96 +0,0 @@
// Package audit implementiert Core AUD-01: das zentrale, vom allgemeinen
// Anwendungs-Log getrennte Audit-Datenmodell fuer sicherheits- und
// compliancerelevante Ereignisse (wer, was, wann, an welchem Tenant).
// Unveraenderlichkeit (Append-only) ist AUD-02, Export/Filter-API ist AUD-03
// — dieses Paket liefert nur das Datenmodell und den EINEN zentralen
// Schreibpfad (Akzeptanzkriterium 3).
package audit
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
// SystemTenant ist der reservierte Tenant-Bezug fuer mandantenuebergreifende
// Ereignisse (z.B. Superadmin-Aktionen) — es gibt bewusst KEINEN Weg, ein
// Ereignis ganz ohne Tenant-Bezug zu schreiben (Akzeptanzkriterium 2).
const SystemTenant = "system"
var ErrMissingTenant = errors.New("audit: tenant_slug darf nicht leer sein")
var ErrMissingActor = errors.New("audit: actor darf nicht leer sein")
var ErrMissingAction = errors.New("audit: action darf nicht leer sein")
// Event ist ein strukturiertes Audit-Ereignis (Akzeptanzkriterium 1: Akteur,
// Aktion, Zielobjekt, Zeitpunkt, Tenant).
type Event struct {
TenantSlug string
Actor string
Action string
Target string
Metadata map[string]any
OccurredAt time.Time
}
// Log ist der EINE zentrale Schreibpfad fuer Audit-Ereignisse — es gibt
// bewusst keine zweite Schreibmoeglichkeit, damit kein Handler versehentlich
// direkt in audit_events schreibt und dabei die Validierung umgeht
// (Akzeptanzkriterium 3).
type Log struct {
pool *pgxpool.Pool
}
func NewLog(pool *pgxpool.Pool) *Log {
return &Log{pool: pool}
}
// Record persistiert genau einen Audit-Eintrag. Fehlender Tenant-Bezug wird
// bereits hier abgewiesen (klarer Fehler statt Constraint-Verletzung im
// Normalfall) — die Datenbank-CHECK-Constraint aus der Migration ist die
// zweite, unumgehbare Verteidigungslinie (Akzeptanzkriterium 2 / Pruefung 2).
func (l *Log) Record(ctx context.Context, e Event) error {
if e.TenantSlug == "" {
return ErrMissingTenant
}
if e.Actor == "" {
return ErrMissingActor
}
if e.Action == "" {
return ErrMissingAction
}
if e.Metadata == nil {
e.Metadata = map[string]any{}
}
metadataJSON, err := json.Marshal(e.Metadata)
if err != nil {
return fmt.Errorf("metadaten serialisieren: %w", err)
}
if e.OccurredAt.IsZero() {
e.OccurredAt = time.Now()
}
_, err = l.pool.Exec(ctx, `
INSERT INTO audit_events (occurred_at, tenant_slug, actor, action, target, metadata)
VALUES ($1, $2, $3, $4, $5, $6)
`, e.OccurredAt, e.TenantSlug, e.Actor, e.Action, e.Target, metadataJSON)
if err != nil {
return fmt.Errorf("audit-ereignis schreiben: %w", err)
}
return nil
}
// CountByTenant ist eine schlanke Lesehilfe fuer Tests/Diagnose — die
// eigentliche Filter-/Export-API ist AUD-03, hier bewusst nicht vorgezogen.
func (l *Log) CountByTenant(ctx context.Context, tenantSlug string) (int, error) {
var n int
if err := l.pool.QueryRow(ctx, `
SELECT count(*) FROM audit_events WHERE tenant_slug = $1
`, tenantSlug).Scan(&n); err != nil {
return 0, fmt.Errorf("audit-ereignisse zaehlen: %w", err)
}
return n, nil
}
-139
View File
@@ -1,139 +0,0 @@
package audit
import (
"context"
"errors"
"fmt"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
func setupAuditTest(t *testing.T) (*Log, *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 audit_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
tenant_slug TEXT NOT NULL CHECK (tenant_slug <> ''),
actor TEXT NOT NULL CHECK (actor <> ''),
action TEXT NOT NULL CHECK (action <> ''),
target TEXT NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'::jsonb
)`); err != nil {
t.Fatalf("schema: %v", err)
}
cleanup := func() {
_, _ = pool.Exec(ctx, `DELETE FROM audit_events WHERE tenant_slug LIKE 'test\_%' ESCAPE '\' OR tenant_slug = $1`, SystemTenant)
pool.Close()
}
return NewLog(pool), pool, cleanup
}
// Akzeptanzkriterium 1 + Pruefung 1: ein sicherheitsrelevanter Vorgang
// (hier: fehlgeschlagener Login) erzeugt zuverlaessig genau einen Eintrag.
func TestRecord_PersistsExactlyOneEventPerSecurityIncident(t *testing.T) {
log, pool, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
// Seit AUD-02 ist audit_events append-only — Zeilen koennen nie wieder
// geloescht werden (auch nicht vom Test-Cleanup). Eindeutiger Slug pro
// Lauf, damit wiederholte Testlaeufe die Zaehlung nicht verfaelschen.
tenantSlug := "test_acme_" + fmt.Sprint(time.Now().UnixNano())
err := log.Record(ctx, Event{
TenantSlug: tenantSlug,
Actor: "alice@example.com",
Action: "iam.login_failed",
Target: "user:alice@example.com",
Metadata: map[string]any{"reason": "falsches passwort"},
})
if err != nil {
t.Fatalf("record: %v", err)
}
count, err := log.CountByTenant(ctx, tenantSlug)
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 1 {
t.Fatalf("erwartet genau 1 audit-eintrag, habe %d", count)
}
var actor, action, target string
if err := pool.QueryRow(ctx, `
SELECT actor, action, target FROM audit_events WHERE tenant_slug = $1
`, tenantSlug).Scan(&actor, &action, &target); err != nil {
t.Fatalf("eintrag lesen: %v", err)
}
if actor != "alice@example.com" || action != "iam.login_failed" || target != "user:alice@example.com" {
t.Fatalf("eintrag unerwartet: actor=%q action=%q target=%q", actor, action, target)
}
}
// Akzeptanzkriterium 2 + Pruefung 2 (App-Ebene): fehlender Tenant-Bezug wird
// bereits vom zentralen Schreibpfad abgewiesen.
func TestRecord_RejectsMissingTenant(t *testing.T) {
log, _, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
err := log.Record(ctx, Event{TenantSlug: "", Actor: "alice", Action: "irgendwas"})
if !errors.Is(err, ErrMissingTenant) {
t.Fatalf("erwartet ErrMissingTenant, habe %v", err)
}
}
// Akzeptanzkriterium 2 + Pruefung 2 (DB-Ebene): selbst ein direkter INSERT,
// der Log.Record umgeht, wird durch die CHECK-Constraint verhindert — der
// Schutz haengt nicht allein von der Go-Validierung ab.
func TestConstraint_RejectsMissingTenantAtDatabaseLevel(t *testing.T) {
_, pool, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
_, err := pool.Exec(ctx, `
INSERT INTO audit_events (tenant_slug, actor, action, target)
VALUES ('', 'alice', 'irgendwas', 'ziel')
`)
if err == nil {
t.Fatal("erwartet fehler durch CHECK-constraint bei leerem tenant_slug, habe nil")
}
}
func TestRecord_RejectsMissingActorAndAction(t *testing.T) {
log, _, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
if err := log.Record(ctx, Event{TenantSlug: "test_acme", Actor: "", Action: "x"}); !errors.Is(err, ErrMissingActor) {
t.Fatalf("erwartet ErrMissingActor, habe %v", err)
}
if err := log.Record(ctx, Event{TenantSlug: "test_acme", Actor: "alice", Action: ""}); !errors.Is(err, ErrMissingAction) {
t.Fatalf("erwartet ErrMissingAction, habe %v", err)
}
}
func TestRecord_SystemTenantForCrossTenantEvents(t *testing.T) {
log, _, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
if err := log.Record(ctx, Event{TenantSlug: SystemTenant, Actor: "superadmin", Action: "tenant.provisioned", Target: "tenant:acme"}); err != nil {
t.Fatalf("record mit SystemTenant: %v", err)
}
}
-120
View File
@@ -1,120 +0,0 @@
package audit
import (
"context"
"crypto/rand"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
var (
ErrConfirmationNotFound = errors.New("audit: bestaetigungsvorgang nicht gefunden")
ErrAlreadyDecided = errors.New("audit: bestaetigungsvorgang wurde bereits entschieden")
ErrSameActor = errors.New("audit: bestaetigung muss von einer anderen person als der anfordernden erfolgen")
ErrInvalidCode = errors.New("audit: bestaetigungscode ungueltig")
)
type ConfirmationStatus string
const (
StatusPending ConfirmationStatus = "pending"
StatusConfirmed ConfirmationStatus = "confirmed"
)
// FourEyes implementiert das Vier-Augen-Prinzip fuer sicherheitskritische
// Entscheidungen (Akzeptanzkriterium 2) nach dem archivdms-Vorbild:
// FOR-UPDATE-Lock gegen Race-Bedingungen bei paralleler Bestaetigung,
// Timing-safe Vergleich des Bestaetigungscodes (Akzeptanzkriterium 3).
type FourEyes struct {
pool *pgxpool.Pool
}
func NewFourEyes(pool *pgxpool.Pool) *FourEyes {
return &FourEyes{pool: pool}
}
// Request legt einen neuen, zu bestaetigenden Vorgang an (z.B. Loeschbestaetigung,
// Rechtevergabe) und liefert einen einmaligen Klartext-Code, der ausserhalb
// dieses Systems (z.B. per E-Mail) an eine ZWEITE Person uebermittelt wird —
// niemals der anfordernden Person selbst.
func (f *FourEyes) Request(ctx context.Context, action, target, requestedBy string) (id, code string, err error) {
code, err = generateCode()
if err != nil {
return "", "", fmt.Errorf("bestaetigungscode erzeugen: %w", err)
}
hash := hashCode(code)
err = f.pool.QueryRow(ctx, `
INSERT INTO security_confirmations (action, target, requested_by, code_hash, status)
VALUES ($1, $2, $3, $4, 'pending')
RETURNING id
`, action, target, requestedBy, hash).Scan(&id)
if err != nil {
return "", "", fmt.Errorf("bestaetigungsvorgang anlegen: %w", err)
}
return id, code, nil
}
// Confirm bestaetigt einen Vorgang. confirmedBy MUSS sich von der
// anfordernden Person unterscheiden (echtes Vier-Augen-Prinzip). Der Zugriff
// auf die Zeile erfolgt mit FOR UPDATE, damit zwei gleichzeitige
// Bestaetigungsversuche serialisiert werden und niemals beide durchgehen
// (Akzeptanzkriterium 2 / Pruefung 2).
func (f *FourEyes) Confirm(ctx context.Context, id, confirmedBy, code string) error {
tx, err := f.pool.Begin(ctx)
if err != nil {
return fmt.Errorf("transaktion starten: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
var requestedBy, status string
var codeHash []byte
err = tx.QueryRow(ctx, `
SELECT requested_by, status, code_hash FROM security_confirmations
WHERE id = $1 FOR UPDATE
`, id).Scan(&requestedBy, &status, &codeHash)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ErrConfirmationNotFound
}
return fmt.Errorf("bestaetigungsvorgang lesen: %w", err)
}
if status != string(StatusPending) {
return ErrAlreadyDecided
}
if confirmedBy == requestedBy {
return ErrSameActor
}
if !timingSafeEqual(hashCode(code), codeHash) {
return ErrInvalidCode
}
if _, err := tx.Exec(ctx, `
UPDATE security_confirmations
SET status = 'confirmed', confirmed_by = $2, confirmed_at = now()
WHERE id = $1
`, id, confirmedBy); err != nil {
return fmt.Errorf("bestaetigung speichern: %w", err)
}
return tx.Commit(ctx)
}
func generateCode() (string, error) {
buf := make([]byte, 16)
if _, err := rand.Read(buf); err != nil {
return "", err
}
return hex.EncodeToString(buf), nil
}
func hashCode(code string) []byte {
sum := sha256.Sum256([]byte(code))
return sum[:]
}
-128
View File
@@ -1,128 +0,0 @@
package audit
import (
"context"
"errors"
"os"
"sync"
"testing"
"github.com/jackc/pgx/v5/pgxpool"
)
func setupFourEyesTest(t *testing.T) (*FourEyes, 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 security_confirmations (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
action TEXT NOT NULL,
target TEXT NOT NULL,
requested_by TEXT NOT NULL,
code_hash BYTEA NOT NULL,
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'confirmed', 'rejected')),
confirmed_by TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
confirmed_at TIMESTAMPTZ
)`); err != nil {
t.Fatalf("schema: %v", err)
}
cleanup := func() {
_, _ = pool.Exec(ctx, `DELETE FROM security_confirmations WHERE action LIKE 'test.%'`)
pool.Close()
}
return NewFourEyes(pool), cleanup
}
func TestFourEyes_RequestAndConfirm(t *testing.T) {
fe, cleanup := setupFourEyesTest(t)
defer cleanup()
ctx := context.Background()
id, code, err := fe.Request(ctx, "test.tenant_delete", "tenant:acme", "alice@example.com")
if err != nil {
t.Fatalf("request: %v", err)
}
if err := fe.Confirm(ctx, id, "bob@example.com", code); err != nil {
t.Fatalf("confirm: %v", err)
}
}
func TestFourEyes_RejectsSameActor(t *testing.T) {
fe, cleanup := setupFourEyesTest(t)
defer cleanup()
ctx := context.Background()
id, code, err := fe.Request(ctx, "test.tenant_delete", "tenant:acme", "alice@example.com")
if err != nil {
t.Fatalf("request: %v", err)
}
if err := fe.Confirm(ctx, id, "alice@example.com", code); !errors.Is(err, ErrSameActor) {
t.Fatalf("erwartet ErrSameActor, habe %v", err)
}
}
func TestFourEyes_RejectsWrongCode(t *testing.T) {
fe, cleanup := setupFourEyesTest(t)
defer cleanup()
ctx := context.Background()
id, _, err := fe.Request(ctx, "test.tenant_delete", "tenant:acme", "alice@example.com")
if err != nil {
t.Fatalf("request: %v", err)
}
if err := fe.Confirm(ctx, id, "bob@example.com", "falscher-code"); !errors.Is(err, ErrInvalidCode) {
t.Fatalf("erwartet ErrInvalidCode, habe %v", err)
}
}
// Akzeptanzkriterium 2 + Pruefung 2: FOR-UPDATE-Lock unter parallelen
// Anfragen race-frei — von zwei gleichzeitigen Bestaetigungsversuchen fuer
// denselben Vorgang darf genau einer durchgehen.
func TestFourEyes_ConcurrentConfirmIsRaceFree(t *testing.T) {
fe, cleanup := setupFourEyesTest(t)
defer cleanup()
ctx := context.Background()
id, code, err := fe.Request(ctx, "test.tenant_delete", "tenant:acme", "alice@example.com")
if err != nil {
t.Fatalf("request: %v", err)
}
var wg sync.WaitGroup
results := make([]error, 2)
confirmers := []string{"bob@example.com", "carol@example.com"}
for i := 0; i < 2; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
results[i] = fe.Confirm(ctx, id, confirmers[i], code)
}(i)
}
wg.Wait()
successCount := 0
for _, err := range results {
if err == nil {
successCount++
} else if !errors.Is(err, ErrAlreadyDecided) {
t.Fatalf("unerwarteter fehler: %v", err)
}
}
if successCount != 1 {
t.Fatalf("erwartet genau eine erfolgreiche bestaetigung, habe %d", successCount)
}
}
-15
View File
@@ -1,15 +0,0 @@
package audit
import "crypto/subtle"
// timingSafeEqual ist die projektweite Referenzimplementierung fuer
// Timing-safe-Vergleiche sicherheitsrelevanter Geheimnisse (Bestaetigungs-
// codes hier, spaeter Freigabelinks in Archive CMP-06 — siehe IAM-02-Ticket-
// Konvention). subtle.ConstantTimeCompare vergleicht in konstanter Zeit
// bezogen auf die Laenge von a, unabhaengig vom Inhalt.
func timingSafeEqual(a, b []byte) bool {
if len(a) != len(b) {
return false
}
return subtle.ConstantTimeCompare(a, b) == 1
}
-62
View File
@@ -1,62 +0,0 @@
package audit
import (
"testing"
"time"
)
func TestTimingSafeEqual_Correctness(t *testing.T) {
a := hashCode("geheimnis-a")
b := hashCode("geheimnis-a")
c := hashCode("geheimnis-b")
if !timingSafeEqual(a, b) {
t.Fatal("identische hashes sollten gleich sein")
}
if timingSafeEqual(a, c) {
t.Fatal("unterschiedliche hashes sollten ungleich sein")
}
if timingSafeEqual(a, []byte("kuerzer")) {
t.Fatal("unterschiedliche laenge sollte ungleich sein")
}
}
// Akzeptanzkriterium 3 + Pruefung 3: Timing-safe Vergleich stichprobenartig
// per Laufzeitmessung verifiziert — ein Mismatch am Anfang darf nicht
// messbar schneller sein als ein Mismatch am Ende (klassisches Merkmal
// eines NICHT timing-safen Vergleichs wie bytes.Equal mit Short-Circuit).
func TestTimingSafeEqual_NoEarlyExitTiming(t *testing.T) {
reference := hashCode("referenzwert-fuer-timing-test")
mismatchAtStart := make([]byte, len(reference))
copy(mismatchAtStart, reference)
mismatchAtStart[0] ^= 0xFF
mismatchAtEnd := make([]byte, len(reference))
copy(mismatchAtEnd, reference)
mismatchAtEnd[len(mismatchAtEnd)-1] ^= 0xFF
const iterations = 20000
startDur := measure(iterations, func() { timingSafeEqual(reference, mismatchAtStart) })
endDur := measure(iterations, func() { timingSafeEqual(reference, mismatchAtEnd) })
t.Logf("mismatch am anfang: %s, mismatch am ende: %s (%d iterationen)", startDur, endDur, iterations)
ratio := float64(startDur) / float64(endDur)
// Grosszuegige Toleranz (Faktor 3), da es ein Stichprobentest auf einer
// geteilten Testmaschine ist, kein isolierter Benchmark — es geht darum,
// eine grobe Short-Circuit-Implementierung zuverlaessig aufzudecken
// (die haette typischerweise eine Groessenordnung Unterschied), nicht um
// kryptographisch praezise Constant-Time-Beweise.
if ratio > 3.0 || ratio < 1.0/3.0 {
t.Fatalf("timing-unterschied zu gross (verdacht auf short-circuit-vergleich): ratio=%.2f", ratio)
}
}
func measure(iterations int, fn func()) time.Duration {
start := time.Now()
for i := 0; i < iterations; i++ {
fn()
}
return time.Since(start)
}
+87
View File
@@ -0,0 +1,87 @@
// Package flag implementiert Core LIC-02: einen Feature-Flag-Dienst mit
// Strategien (global an/aus, Prozentsatz, Tenant-Zielgruppe) als Kernfunktion
// des Core-Dienstes selbst — keine zusaetzliche Infrastruktur (Unleash-Server
// + eigene DB), siehe "bewusst vermeiden" im LIC-02-Ticket.
package flag
import (
"context"
"errors"
"fmt"
"hash/fnv"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
var ErrNotFound = errors.New("flag: nicht gefunden")
// Flag ist die zentrale Definition — Auswertung (Evaluate) ist bewusst davon
// getrennt (Unleash-Prinzip: Flag-Verwaltung vs. Flag-Auswertung).
type Flag struct {
Key string
Enabled bool
RolloutPercentage int
TargetTenantSlugs []string
}
// Store ist die Verwaltungsseite (Admin): Flags definieren/lesen.
type Store struct {
pool *pgxpool.Pool
}
func NewStore(pool *pgxpool.Pool) *Store {
return &Store{pool: pool}
}
func (s *Store) Set(ctx context.Context, f Flag) error {
if f.TargetTenantSlugs == nil {
f.TargetTenantSlugs = []string{} // pgx uebertraegt ein nil-Slice sonst als SQL NULL statt leerem Array.
}
_, err := s.pool.Exec(ctx, `
INSERT INTO feature_flags (key, enabled, rollout_percentage, target_tenant_slugs, updated_at)
VALUES ($1, $2, $3, $4, now())
ON CONFLICT (key) DO UPDATE SET
enabled = $2, rollout_percentage = $3, target_tenant_slugs = $4, updated_at = now()
`, f.Key, f.Enabled, f.RolloutPercentage, f.TargetTenantSlugs)
if err != nil {
return fmt.Errorf("flag speichern: %w", err)
}
return nil
}
func (s *Store) Get(ctx context.Context, key string) (Flag, error) {
var f Flag
row := s.pool.QueryRow(ctx, `
SELECT key, enabled, rollout_percentage, target_tenant_slugs
FROM feature_flags WHERE key = $1
`, key)
if err := row.Scan(&f.Key, &f.Enabled, &f.RolloutPercentage, &f.TargetTenantSlugs); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return Flag{}, ErrNotFound
}
return Flag{}, fmt.Errorf("flag lesen: %w", err)
}
return f, nil
}
// evaluate wendet die Strategien in fester Reihenfolge an: globaler
// An/Aus-Schalter zuerst, dann Tenant-Zielgruppe, dann Prozentsatz-Rollout.
// Ein unbekannter/nicht getroffener Fall ergibt false — Fail-Safe-Default,
// kein Feature wird versehentlich aktiv.
func evaluate(f Flag, tenantSlug string) bool {
if f.Enabled {
return true
}
for _, target := range f.TargetTenantSlugs {
if target == tenantSlug {
return true
}
}
if f.RolloutPercentage > 0 {
h := fnv.New32a()
_, _ = h.Write([]byte(f.Key + "|" + tenantSlug))
return int(h.Sum32()%100) < f.RolloutPercentage
}
return false
}
+43
View File
@@ -0,0 +1,43 @@
package flag
import "testing"
func TestEvaluate_GlobalEnabled(t *testing.T) {
f := Flag{Key: "k", Enabled: true}
if !evaluate(f, "irgendein-tenant") {
t.Fatal("global aktiviertes flag sollte fuer jeden tenant true liefern")
}
}
// Akzeptanzkriterium 1 + Pruefung 2: Zielgruppen-Strategie.
func TestEvaluate_TargetTenantStrategy(t *testing.T) {
f := Flag{Key: "k", Enabled: false, TargetTenantSlugs: []string{"acme"}}
if !evaluate(f, "acme") {
t.Fatal("erwartet true fuer tenant in zielgruppe")
}
if evaluate(f, "globex") {
t.Fatal("erwartet false fuer tenant ausserhalb der zielgruppe")
}
}
func TestEvaluate_RolloutPercentageBoundaries(t *testing.T) {
full := Flag{Key: "k", RolloutPercentage: 100}
if !evaluate(full, "beliebiger-tenant-1") || !evaluate(full, "beliebiger-tenant-2") {
t.Fatal("100% rollout sollte immer true liefern")
}
none := Flag{Key: "k", RolloutPercentage: 0}
if evaluate(none, "beliebiger-tenant") {
t.Fatal("0% rollout ohne enabled/zielgruppe sollte false liefern")
}
}
func TestEvaluate_RolloutIsDeterministicPerTenant(t *testing.T) {
f := Flag{Key: "k", RolloutPercentage: 50}
first := evaluate(f, "stabiler-tenant")
for i := 0; i < 5; i++ {
if evaluate(f, "stabiler-tenant") != first {
t.Fatal("rollout-auswertung sollte fuer denselben tenant/key stabil sein")
}
}
}
+87
View File
@@ -0,0 +1,87 @@
package flag
import (
"context"
"log/slog"
"sync"
"time"
)
// DefaultCacheTTL ist die dokumentierte Cache-Invalidierungszeit
// (Akzeptanzkriterium 2/3): eine Aenderung wirkt spaetestens nach dieser
// Zeit auf allen Core-Instanzen, ohne dass ein Dienst neu gestartet werden
// muss (Akzeptanzkriterium 3).
const DefaultCacheTTL = 5 * time.Second
type cacheEntry struct {
flag Flag
expiresAt time.Time
}
// Service ist die Auswertungsseite (SDK/Client-Analogon zu Unleash) mit
// lokalem TTL-Cache. Bewusst getrennt von Store (Verwaltung).
type Service struct {
store *Store
ttl time.Duration
mu sync.RWMutex
cache map[string]cacheEntry
}
func NewService(store *Store, ttl time.Duration) *Service {
if ttl <= 0 {
ttl = DefaultCacheTTL
}
return &Service{store: store, ttl: ttl, cache: make(map[string]cacheEntry)}
}
// IsEnabled wertet ein Flag fuer einen Tenant aus. Liefert IMMER einen
// bool ohne Fehlerwert — ein nicht erreichbarer Flag-Dienst darf abhaengige
// Aufrufer nicht zum Absturz bringen oder zu Fehlerbehandlungscode zwingen,
// der leicht vergessen wird (Akzeptanzkriterium 3 / Pruefung 3: dokumentiertes
// Fallback-Verhalten = false, ggf. aus dem zuletzt bekannten Zwischenspeicher).
func (s *Service) IsEnabled(ctx context.Context, tenantSlug, key string) bool {
f, ok := s.resolve(ctx, key)
if !ok {
return false
}
return evaluate(f, tenantSlug)
}
func (s *Service) resolve(ctx context.Context, key string) (Flag, bool) {
s.mu.RLock()
entry, exists := s.cache[key]
fresh := exists && time.Now().Before(entry.expiresAt)
s.mu.RUnlock()
if fresh {
return entry.flag, true
}
f, err := s.store.Get(ctx, key)
if err != nil {
if exists {
slog.Warn("feature-flag-dienst nicht erreichbar, nutze zwischengespeicherten stand",
"flag_key", key, "error", err)
return entry.flag, true
}
slog.Warn("feature-flag-dienst nicht erreichbar, kein zwischengespeicherter stand vorhanden, fallback: deaktiviert",
"flag_key", key, "error", err)
return Flag{}, false
}
s.mu.Lock()
s.cache[key] = cacheEntry{flag: f, expiresAt: time.Now().Add(s.ttl)}
s.mu.Unlock()
return f, true
}
// Invalidate erzwingt beim naechsten IsEnabled-Aufruf ein sofortiges Neuladen
// aus der Datenbank statt auf den TTL-Ablauf zu warten — wird nach Store.Set
// auf derselben Instanz aufgerufen, damit der Schreiber die eigene Aenderung
// ohne Wartezeit sieht. Andere Core-Instanzen sehen sie spaetestens nach
// DefaultCacheTTL (siehe Akzeptanzkriterium 3).
func (s *Service) Invalidate(key string) {
s.mu.Lock()
delete(s.cache, key)
s.mu.Unlock()
}
+179
View File
@@ -0,0 +1,179 @@
package flag
import (
"context"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
func setupFlagStoreTest(t *testing.T) (*Store, 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 feature_flags (
key TEXT PRIMARY KEY,
enabled BOOLEAN NOT NULL DEFAULT false,
rollout_percentage INT NOT NULL DEFAULT 0,
target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
)`); err != nil {
t.Fatalf("schema: %v", err)
}
cleanup := func() {
_, _ = pool.Exec(ctx, `DELETE FROM feature_flags WHERE key LIKE 'test\_%' ESCAPE '\'`)
pool.Close()
}
return NewStore(pool), cleanup
}
// Akzeptanzkriterium 1 + Pruefung 2: Zielgruppen-Strategie liefert im Test
// die erwartete Auswertung.
func TestService_TargetTenantStrategy(t *testing.T) {
store, cleanup := setupFlagStoreTest(t)
defer cleanup()
ctx := context.Background()
if err := store.Set(ctx, Flag{Key: "test_target_flag", TargetTenantSlugs: []string{"acme"}}); err != nil {
t.Fatalf("set: %v", err)
}
svc := NewService(store, time.Hour)
if !svc.IsEnabled(ctx, "acme", "test_target_flag") {
t.Fatal("erwartet true fuer tenant in zielgruppe")
}
if svc.IsEnabled(ctx, "globex", "test_target_flag") {
t.Fatal("erwartet false fuer tenant ausserhalb der zielgruppe")
}
}
// Akzeptanzkriterium 2 + 3 + Pruefung 1: Flag-Aenderung wirkt innerhalb der
// dokumentierten Cache-Invalidierungszeit, automatisiert gemessen.
func TestService_CacheInvalidationTiming(t *testing.T) {
store, cleanup := setupFlagStoreTest(t)
defer cleanup()
ctx := context.Background()
const ttl = 150 * time.Millisecond
if err := store.Set(ctx, Flag{Key: "test_ttl_flag", Enabled: false}); err != nil {
t.Fatalf("set: %v", err)
}
svc := NewService(store, ttl)
if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
t.Fatal("erwartet false vor der aenderung")
}
// Aenderung "auf einer anderen instanz" simulieren: direkt ueber den
// Store, ohne svc.Invalidate aufzurufen.
changedAt := time.Now()
if err := store.Set(ctx, Flag{Key: "test_ttl_flag", Enabled: true}); err != nil {
t.Fatalf("set: %v", err)
}
// Sofort danach sollte der Cache noch den alten Stand liefern.
if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
t.Fatal("cache haette den alten (false) stand liefern sollen, direkt nach der aenderung")
}
deadline := changedAt.Add(ttl + 100*time.Millisecond)
for time.Now().Before(deadline) {
if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
elapsed := time.Since(changedAt)
t.Logf("aenderung wurde nach %s wirksam (ziel: innerhalb %s + toleranz)", elapsed, ttl)
return
}
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("aenderung wurde nicht innerhalb von %s wirksam", deadline.Sub(changedAt))
}
func TestService_InvalidateForcesImmediateRefresh(t *testing.T) {
store, cleanup := setupFlagStoreTest(t)
defer cleanup()
ctx := context.Background()
if err := store.Set(ctx, Flag{Key: "test_invalidate_flag", Enabled: false}); err != nil {
t.Fatalf("set: %v", err)
}
svc := NewService(store, time.Hour) // lange TTL, damit Invalidate den unterschied macht
_ = svc.IsEnabled(ctx, "acme", "test_invalidate_flag")
if err := store.Set(ctx, Flag{Key: "test_invalidate_flag", Enabled: true}); err != nil {
t.Fatalf("set: %v", err)
}
svc.Invalidate("test_invalidate_flag")
if !svc.IsEnabled(ctx, "acme", "test_invalidate_flag") {
t.Fatal("erwartet sofort sichtbaren neuen stand nach Invalidate")
}
}
// Akzeptanzkriterium 3 + Pruefung 3: Ausfall des Flag-Dienstes fuehrt zu
// dokumentiertem Fallback-Verhalten, nicht zum Absturz.
func TestService_FallsBackOnStoreFailure(t *testing.T) {
store, cleanup := setupFlagStoreTest(t)
defer cleanup()
ctx := context.Background()
if err := store.Set(ctx, Flag{Key: "test_fallback_flag", Enabled: true}); err != nil {
t.Fatalf("set: %v", err)
}
svc := NewService(store, time.Hour)
// Cache vorwaermen, waehrend die DB noch erreichbar ist.
if !svc.IsEnabled(ctx, "acme", "test_fallback_flag") {
t.Fatal("erwartet true bei funktionierender db")
}
brokenPool, err := pgxpool.New(ctx, "postgresql://nonexistent-host-fuer-test:5432/x?connect_timeout=1")
if err != nil {
t.Fatalf("broken pool erstellen (sollte nicht sofort verbinden): %v", err)
}
brokenStore := NewStore(brokenPool)
svcWithCache := NewService(brokenStore, time.Nanosecond) // TTL sofort abgelaufen, erzwingt reload-versuch
svcWithCache.mu.Lock()
svcWithCache.cache["test_fallback_flag"] = cacheEntry{
flag: Flag{Key: "test_fallback_flag", Enabled: true},
expiresAt: time.Now().Add(-time.Hour), // bereits abgelaufen
}
svcWithCache.mu.Unlock()
func() {
defer func() {
if r := recover(); r != nil {
t.Fatalf("IsEnabled hat gepanict statt einen fallback zu liefern: %v", r)
}
}()
if !svcWithCache.IsEnabled(ctx, "acme", "test_fallback_flag") {
t.Fatal("erwartet fallback auf zwischengespeicherten (true) stand bei db-ausfall")
}
}()
// Voellig frischer Dienst ohne jeglichen cache + kaputte db -> sicherer
// default false, kein absturz.
freshSvc := NewService(brokenStore, time.Hour)
func() {
defer func() {
if r := recover(); r != nil {
t.Fatalf("IsEnabled hat gepanict: %v", r)
}
}()
if freshSvc.IsEnabled(ctx, "acme", "test_fallback_flag") {
t.Fatal("erwartet fail-safe false ohne cache und mit kaputter db")
}
}()
}
+25
View File
@@ -0,0 +1,25 @@
package health
import (
"context"
"github.com/jackc/pgx/v5/pgxpool"
)
// DatabaseChecker prueft die tatsaechliche Erreichbarkeit der Datenbank
// (Ping) — nicht nur, ob der Pool existiert.
func DatabaseChecker(pool *pgxpool.Pool) CheckerFunc {
return func(ctx context.Context) error {
return pool.Ping(ctx)
}
}
// QueueChecker prueft, dass die Postgres-basierte Job-Queue (siehe CFG-02)
// tatsaechlich abfragbar ist — eine eigene, benannte Abhaengigkeit neben der
// reinen DB-Erreichbarkeit (Akzeptanzkriterium 1).
func QueueChecker(pool *pgxpool.Pool) CheckerFunc {
return func(ctx context.Context) error {
_, err := pool.Exec(ctx, `SELECT 1`)
return err
}
}
+49
View File
@@ -0,0 +1,49 @@
package health
import (
"encoding/json"
"net/http"
)
// LivenessHandler beantwortet IMMER "lebt", solange der Prozess ueberhaupt
// HTTP-Anfragen verarbeiten kann — prueft bewusst KEINE externen
// Abhaengigkeiten (Akzeptanzkriterium 2: Liveness und Readiness getrennt).
// Ein Datenbankausfall darf die Liveness nicht auf "tot" setzen, sonst
// wuerde eine Orchestrierung (z.B. systemd/Kubernetes) den Prozess grundlos
// neu starten, obwohl nur eine Abhaengigkeit ausgefallen ist.
func LivenessHandler(w http.ResponseWriter, r *http.Request) {
writeStatus(w, http.StatusOK, map[string]any{"status": "alive"})
}
// ReadinessHandler prueft ALLE registrierten Abhaengigkeiten
// (Akzeptanzkriterium 1) und liefert 503, sobald eine davon fehlschlaegt
// (Akzeptanzkriterium 3) — unterscheidet sich damit nachweislich von
// LivenessHandler im Fehlerfall (Akzeptanzkriterium 2 / Pruefung 3).
func (r *Registry) ReadinessHandler() http.HandlerFunc {
return func(w http.ResponseWriter, req *http.Request) {
ready, results := r.CheckAll(req.Context())
body := map[string]any{
"status": statusText(ready),
"checks": results,
}
status := http.StatusOK
if !ready {
status = http.StatusServiceUnavailable
}
writeStatus(w, status, body)
}
}
func statusText(ready bool) string {
if ready {
return "ready"
}
return "not_ready"
}
func writeStatus(w http.ResponseWriter, status int, body map[string]any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(body)
}
+86
View File
@@ -0,0 +1,86 @@
// Package health implementiert Core OPS-01: Health-/Readiness-Endpunkte, die
// echte Abhaengigkeiten (DB, Job-Queue) statt nur den Prozessstatus pruefen
// — wiederverwendbar von Core UND jedem registrierten Modul (siehe API-02),
// nicht nur von Core selbst.
package health
import (
"context"
"time"
)
// Checker prueft EINE Abhaengigkeit (z.B. Datenbank, Job-Queue).
type Checker interface {
Check(ctx context.Context) error
}
type CheckerFunc func(ctx context.Context) error
func (f CheckerFunc) Check(ctx context.Context) error { return f(ctx) }
// DefaultCheckTimeout begrenzt, wie lange EIN einzelner Check maximal
// dauern darf, bevor er als fehlgeschlagen gilt — verhindert, dass ein
// haengender Check den gesamten Readiness-Endpunkt blockiert
// (Akzeptanzkriterium 2 / Pruefung 2: Antwort innerhalb definierter Zeit).
const DefaultCheckTimeout = 2 * time.Second
// Registry haelt alle benannten Checks eines Dienstes.
type Registry struct {
checks map[string]Checker
timeout time.Duration
}
func NewRegistry() *Registry {
return &Registry{checks: make(map[string]Checker), timeout: DefaultCheckTimeout}
}
func (r *Registry) WithTimeout(d time.Duration) *Registry {
return &Registry{checks: r.checks, timeout: d}
}
// Register fuegt einen benannten Check hinzu (z.B. "database", "queue").
func (r *Registry) Register(name string, c Checker) {
r.checks[name] = c
}
// Result ist der Ausgang eines einzelnen Checks.
type Result struct {
OK bool
Error string
}
// CheckAll fuehrt alle registrierten Checks NEBENLAEUFIG mit je eigenem
// Timeout aus (Akzeptanzkriterium 1: echte Abhaengigkeiten statt Prozess-
// status) und liefert ready=false, sobald irgendein Check fehlschlaegt
// (Akzeptanzkriterium 3: ein Ausfall wird sichtbar).
func (r *Registry) CheckAll(ctx context.Context) (ready bool, results map[string]Result) {
type namedResult struct {
name string
result Result
}
ch := make(chan namedResult, len(r.checks))
for name, checker := range r.checks {
go func(name string, checker Checker) {
checkCtx, cancel := context.WithTimeout(ctx, r.timeout)
defer cancel()
err := checker.Check(checkCtx)
if err != nil {
ch <- namedResult{name, Result{OK: false, Error: err.Error()}}
return
}
ch <- namedResult{name, Result{OK: true}}
}(name, checker)
}
results = make(map[string]Result, len(r.checks))
ready = true
for i := 0; i < len(r.checks); i++ {
nr := <-ch
results[nr.name] = nr.result
if !nr.result.OK {
ready = false
}
}
return ready, results
}
+161
View File
@@ -0,0 +1,161 @@
package health
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
// Akzeptanzkriterium 1 + Pruefung 1: simulierter Datenbankausfall fuehrt zu
// "nicht bereit".
func TestReadinessHandler_ReportsNotReadyOnDatabaseFailure(t *testing.T) {
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)
}
// Datenbankausfall simulieren: Pool sofort schliessen, bevor der Check laeuft.
pool.Close()
reg := NewRegistry()
reg.Register("database", DatabaseChecker(pool))
req := httptest.NewRequest(http.MethodGet, "/readyz", nil)
rec := httptest.NewRecorder()
reg.ReadinessHandler()(rec, req)
if rec.Code != http.StatusServiceUnavailable {
t.Fatalf("status = %d, want 503 bei db-ausfall", rec.Code)
}
var body struct {
Status string `json:"status"`
Checks map[string]interface{} `json:"checks"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
t.Fatalf("body parsen: %v", err)
}
if body.Status != "not_ready" {
t.Fatalf("status-feld = %q, want not_ready", body.Status)
}
if _, ok := body.Checks["database"]; !ok {
t.Fatal("erwartet 'database' im checks-ergebnis")
}
}
func TestReadinessHandler_ReportsReadyWhenAllChecksPass(t *testing.T) {
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)
}
defer pool.Close()
reg := NewRegistry()
reg.Register("database", DatabaseChecker(pool))
reg.Register("queue", QueueChecker(pool))
req := httptest.NewRequest(http.MethodGet, "/readyz", nil)
rec := httptest.NewRecorder()
reg.ReadinessHandler()(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200 bei funktionierenden abhaengigkeiten", rec.Code)
}
}
// Akzeptanzkriterium 2 + Pruefung 3: Liveness und Readiness unterscheiden
// sich nachweislich im Fehlerfall.
func TestLivenessAndReadiness_DifferOnDatabaseFailure(t *testing.T) {
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)
}
pool.Close() // db-ausfall simulieren
reg := NewRegistry()
reg.Register("database", DatabaseChecker(pool))
livenessRec := httptest.NewRecorder()
LivenessHandler(livenessRec, httptest.NewRequest(http.MethodGet, "/livez", nil))
if livenessRec.Code != http.StatusOK {
t.Fatalf("liveness status = %d, want 200 trotz db-ausfall (liveness prueft keine abhaengigkeiten)", livenessRec.Code)
}
readinessRec := httptest.NewRecorder()
reg.ReadinessHandler()(readinessRec, httptest.NewRequest(http.MethodGet, "/readyz", nil))
if readinessRec.Code != http.StatusServiceUnavailable {
t.Fatalf("readiness status = %d, want 503 bei db-ausfall", readinessRec.Code)
}
if livenessRec.Code == readinessRec.Code {
t.Fatal("liveness und readiness sollten sich im db-ausfall-fall unterscheiden")
}
}
// Akzeptanzkriterium 2 + Pruefung 2: Health-Endpunkt antwortet auch bei
// haengendem Check innerhalb definierter Zeit (Timeout begrenzt die Dauer).
func TestReadinessHandler_RespondsWithinTimeoutEvenWithHangingCheck(t *testing.T) {
reg := NewRegistry().WithTimeout(50 * time.Millisecond)
reg.Register("haengender_dienst", CheckerFunc(func(ctx context.Context) error {
select {
case <-time.After(10 * time.Second): // wuerde ohne timeout ewig blockieren
return nil
case <-ctx.Done():
return ctx.Err()
}
}))
start := time.Now()
req := httptest.NewRequest(http.MethodGet, "/readyz", nil)
rec := httptest.NewRecorder()
reg.ReadinessHandler()(rec, req)
elapsed := time.Since(start)
if elapsed > time.Second {
t.Fatalf("readiness handler brauchte %s, erwartet deutlich unter 1s durch timeout", elapsed)
}
if rec.Code != http.StatusServiceUnavailable {
t.Fatalf("status = %d, want 503 fuer haengenden/timeout-check", rec.Code)
}
}
func TestCheckAll_MultipleChecksRunConcurrently(t *testing.T) {
reg := NewRegistry().WithTimeout(time.Second)
reg.Register("a", CheckerFunc(func(ctx context.Context) error { return nil }))
reg.Register("b", CheckerFunc(func(ctx context.Context) error { return errors.New("kaputt") }))
ready, results := reg.CheckAll(context.Background())
if ready {
t.Fatal("erwartet ready=false, da 'b' fehlschlaegt")
}
if !results["a"].OK {
t.Fatalf("erwartet 'a' ok, habe %+v", results["a"])
}
if results["b"].OK || results["b"].Error == "" {
t.Fatalf("erwartet 'b' fehlgeschlagen mit fehlertext, habe %+v", results["b"])
}
}
+19
View File
@@ -0,0 +1,19 @@
package metrics
import "github.com/prometheus/client_golang/prometheus"
// NewCoreRegistry liefert das Prometheus-Registry fuer die EIGENEN
// Kennzahlen des Core-Dienstes (Akzeptanzkriterium 1) — alle Namen tragen
// das Praefix "nexarch_core_" gemaess der im Paketkommentar dokumentierten
// Namenskonvention (Akzeptanzkriterium 3). Ein eigenes Registry statt des
// globalen DefaultRegisterer, damit Tests unabhaengig voneinander sind.
func NewCoreRegistry() *prometheus.Registry {
reg := prometheus.NewRegistry()
reg.MustRegister(
prometheus.NewGaugeFunc(prometheus.GaugeOpts{
Name: "nexarch_core_up",
Help: "1, solange der Core-Dienst laeuft und Metriken liefern kann.",
}, func() float64 { return 1 }),
)
return reg
}
+168
View File
@@ -0,0 +1,168 @@
// Package metrics implementiert Core OPS-03: einen zentralen /metrics-
// Endpunkt im Prometheus-Textformat, der Kennzahlen des Core-Dienstes UND
// aggregierte Kennzahlen aller registrierten Module bereitstellt — offenes
// Pull-Modell nach Prometheus-Vorbild, kein proprietaerer Push-Mechanismus.
//
// Namenskonvention (Akzeptanzkriterium 3, modulübergreifend konsistent):
//
// nexarch_core_<name> — Kennzahlen des Core-Dienstes selbst
// nexarch_module_<modul>_<name> — von einem Modul gescrapte Kennzahl
// <name>, umbenannt mit dem
// Modulnamen als Praefix
//
// Ein Modul liefert seine eigenen Kennzahlen unter EIGENEM Namen (z.B.
// "requests_total") unter seinem eigenen /metrics-Endpunkt — dieses Paket
// benennt sie beim Einsammeln konsistent um, damit im aggregierten Core-
// Endpunkt niemals zwei Module denselben Metrik-Namen kollidieren lassen.
package metrics
import (
"context"
"fmt"
"net/http"
"time"
"github.com/prometheus/client_golang/prometheus"
dto "github.com/prometheus/client_model/go"
"github.com/prometheus/common/expfmt"
"github.com/prometheus/common/model"
)
// init erzwingt das klassische Prometheus-Namensschema (a-z, A-Z, 0-9, _)
// fuer die Namensvalidierung von expfmt/model — ohne diese explizite
// Festlegung liefert die Bibliothek "Invalid name validation scheme
// requested: unset" beim Parsen/Kodieren, da sie den globalen Default in
// dieser Version nicht mehr implizit setzt.
func init() {
model.NameValidationScheme = model.LegacyValidation
}
// Source ist EIN registriertes Modul mit seinem eigenen /metrics-Endpunkt
// (siehe internal/health fuer das analoge Muster bei Readiness-Checks).
type Source struct {
ModuleName string
MetricsURL string
}
// SourceProvider liefert die aktuell registrierten Module — typischerweise
// rueckgebunden an internal/moduleregistry.Registry.List (API-02) ueber
// einen kleinen Adapter im aufrufenden Code, damit dieses Paket
// internal/moduleregistry nicht direkt importieren muss (Kein Umbau
// angrenzender Bereiche). Ein NEU registriertes Modul erscheint automatisch
// beim naechsten Aufruf von Aggregator.Handler, OHNE Codeaenderung an diesem
// Paket (Akzeptanzkriterium 2 / Pruefung 2).
type SourceProvider func(ctx context.Context) ([]Source, error)
// FetchTimeout begrenzt, wie lange EIN Modul-Scrape maximal dauern darf —
// ein haengendes Modul darf den gesamten Aggregations-Request nicht
// verzoegern (Pruefung 1: Antwort unter Last innerhalb definierter Zeit).
const FetchTimeout = 2 * time.Second
// Aggregator sammelt Core-eigene Metriken (coreGatherer) und die Metriken
// aller ueber sourceProvider gemeldeten Module in EINER Antwort ein.
type Aggregator struct {
coreGatherer prometheus.Gatherer
sourceProvider SourceProvider
client *http.Client
}
func NewAggregator(coreGatherer prometheus.Gatherer, sourceProvider SourceProvider) *Aggregator {
return &Aggregator{
coreGatherer: coreGatherer,
sourceProvider: sourceProvider,
client: &http.Client{Timeout: FetchTimeout},
}
}
// Gather implementiert prometheus.Gatherer: liefert Core-Metriken PLUS alle
// erreichbaren Modul-Metriken (umbenannt gemaess Namenskonvention) in einer
// gemeinsamen Liste von MetricFamilies.
func (a *Aggregator) Gather(ctx context.Context) ([]*dto.MetricFamily, error) {
families, err := a.coreGatherer.Gather()
if err != nil {
return nil, fmt.Errorf("core-metriken einsammeln: %w", err)
}
sources, err := a.sourceProvider(ctx)
if err != nil {
return nil, fmt.Errorf("modul-quellen ermitteln: %w", err)
}
// Module werden NEBENLAEUFIG gescrapt (dasselbe Muster wie
// internal/health.Registry.CheckAll) — ein langsames/nicht erreichbares
// Modul haelt weder andere Module noch den Gesamt-Request auf.
type fetchResult struct {
families []*dto.MetricFamily
}
resultCh := make(chan fetchResult, len(sources))
for _, src := range sources {
go func(src Source) {
fetchCtx, cancel := context.WithTimeout(ctx, FetchTimeout)
defer cancel()
mf, err := a.fetchAndRename(fetchCtx, src)
if err != nil {
resultCh <- fetchResult{} // Fehlerfall: einfach nichts beitragen, Aggregation laeuft weiter
return
}
resultCh <- fetchResult{families: mf}
}(src)
}
for range sources {
r := <-resultCh
families = append(families, r.families...)
}
return families, nil
}
// fetchAndRename ruft die /metrics-URL eines Moduls ab, parst das
// Prometheus-Textformat und benennt jede Metrik gemaess der
// Namenskonvention um (Akzeptanzkriterium 2 + 3).
func (a *Aggregator) fetchAndRename(ctx context.Context, src Source) ([]*dto.MetricFamily, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, src.MetricsURL, nil)
if err != nil {
return nil, err
}
resp, err := a.client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("modul %s: unerwarteter status %d", src.ModuleName, resp.StatusCode)
}
parser := expfmt.NewTextParser(model.LegacyValidation)
parsed, err := parser.TextToMetricFamilies(resp.Body)
if err != nil {
return nil, fmt.Errorf("modul %s: metrik-text nicht parsebar: %w", src.ModuleName, err)
}
out := make([]*dto.MetricFamily, 0, len(parsed))
for name, mf := range parsed {
renamed := fmt.Sprintf("nexarch_module_%s_%s", src.ModuleName, name)
mf.Name = &renamed
out = append(out, mf)
}
return out, nil
}
// Handler liefert einen HTTP-Handler, der Gather aufruft und das Ergebnis im
// Prometheus-Textformat ausgibt (Akzeptanzkriterium 1).
func (a *Aggregator) Handler() http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
families, err := a.Gather(r.Context())
if err != nil {
http.Error(w, "metriken konnten nicht eingesammelt werden", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", string(expfmt.NewFormat(expfmt.TypeTextPlain)))
enc := expfmt.NewEncoder(w, expfmt.NewFormat(expfmt.TypeTextPlain))
for _, mf := range families {
if err := enc.Encode(mf); err != nil {
return
}
}
}
}
+179
View File
@@ -0,0 +1,179 @@
package metrics
import (
"context"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"github.com/prometheus/common/expfmt"
"github.com/prometheus/common/model"
)
func fakeModuleServer(metricName string) *httptest.Server {
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/plain; version=0.0.4")
fmt.Fprintf(w, "# HELP %s ein test-zaehler\n# TYPE %s counter\n%s 42\n", metricName, metricName, metricName)
}))
}
// Akzeptanzkriterium 1: Core liefert unter dem Handler valides
// Prometheus-Textformat mit den eigenen Kennzahlen.
func TestHandler_ServesCoreMetricsInPrometheusFormat(t *testing.T) {
agg := NewAggregator(NewCoreRegistry(), func(ctx context.Context) ([]Source, error) { return nil, nil })
req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
rec := httptest.NewRecorder()
agg.Handler()(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", rec.Code)
}
parser := expfmt.NewTextParser(model.LegacyValidation)
families, err := parser.TextToMetricFamilies(strings.NewReader(rec.Body.String()))
if err != nil {
t.Fatalf("antwort ist kein valides prometheus-textformat: %v", err)
}
if _, ok := families["nexarch_core_up"]; !ok {
t.Fatalf("erwartet 'nexarch_core_up' unter den core-metriken, habe: %v", keysOf(families))
}
}
func keysOf[V any](m map[string]V) []string {
out := make([]string, 0, len(m))
for k := range m {
out = append(out, k)
}
return out
}
// Akzeptanzkriterium 2 + Pruefung 2: ein NEU registriertes Modul erscheint
// in der Aggregation, OHNE dass dieses Paket oder der Aufrufer Code
// aendern muss — die Quelle kommt ausschliesslich aus sourceProvider.
func TestHandler_NewlyRegisteredModuleAppearsWithoutCodeChange(t *testing.T) {
moduleServer := fakeModuleServer("requests_total")
defer moduleServer.Close()
// Simuliert eine sich zur Laufzeit aendernde Modul-Liste (z.B. aus
// SourceStore.Provide) — zunaechst LEER, dann mit einem Eintrag.
var sources []Source
var mu sync.Mutex
provider := func(ctx context.Context) ([]Source, error) {
mu.Lock()
defer mu.Unlock()
out := make([]Source, len(sources))
copy(out, sources)
return out, nil
}
agg := NewAggregator(NewCoreRegistry(), provider)
// Vor der Registrierung: Modul-Metrik nicht vorhanden.
rec1 := httptest.NewRecorder()
agg.Handler()(rec1, httptest.NewRequest(http.MethodGet, "/metrics", nil))
if strings.Contains(rec1.Body.String(), "requests_total") {
t.Fatal("modul-metrik haette vor registrierung nicht erscheinen duerfen")
}
// Modul wird "registriert" (kein Code hier oder in metrics.go aendert sich).
mu.Lock()
sources = append(sources, Source{ModuleName: "dms", MetricsURL: moduleServer.URL})
mu.Unlock()
rec2 := httptest.NewRecorder()
agg.Handler()(rec2, httptest.NewRequest(http.MethodGet, "/metrics", nil))
body := rec2.Body.String()
if !strings.Contains(body, "nexarch_module_dms_requests_total") {
t.Fatalf("erwartet umbenannte modul-metrik 'nexarch_module_dms_requests_total' nach registrierung, body:\n%s", body)
}
}
// Akzeptanzkriterium 3: Namenskonvention "nexarch_module_<modul>_<name>"
// wird tatsaechlich angewendet.
func TestFetchAndRename_AppliesNamingConvention(t *testing.T) {
moduleServer := fakeModuleServer("queue_depth")
defer moduleServer.Close()
agg := NewAggregator(NewCoreRegistry(), nil)
families, err := agg.fetchAndRename(context.Background(), Source{ModuleName: "mail", MetricsURL: moduleServer.URL})
if err != nil {
t.Fatalf("fetchAndRename: %v", err)
}
if len(families) != 1 || families[0].GetName() != "nexarch_module_mail_queue_depth" {
t.Fatalf("erwartet genau 1 metrik 'nexarch_module_mail_queue_depth', habe: %+v", families)
}
}
// Ein nicht erreichbares Modul darf die Aggregation der uebrigen und die
// Gesamtantwort nicht verhindern (dieselbe Resilienz wie OPS-02).
func TestHandler_UnreachableModuleDoesNotBreakAggregation(t *testing.T) {
reachable := fakeModuleServer("healthy_metric")
defer reachable.Close()
unreachable := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
unreachableURL := unreachable.URL
unreachable.Close() // sofort schliessen -> Verbindung schlaegt fehl
provider := func(ctx context.Context) ([]Source, error) {
return []Source{
{ModuleName: "ok", MetricsURL: reachable.URL},
{ModuleName: "kaputt", MetricsURL: unreachableURL},
}, nil
}
agg := NewAggregator(NewCoreRegistry(), provider)
rec := httptest.NewRecorder()
agg.Handler()(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200 trotz einem nicht erreichbaren modul", rec.Code)
}
body := rec.Body.String()
if !strings.Contains(body, "nexarch_module_ok_healthy_metric") {
t.Fatal("erreichbares modul haette trotz ausfall des anderen aggregiert werden sollen")
}
if strings.Contains(body, "kaputt") {
t.Fatal("nicht erreichbares modul haette keine metrik beitragen duerfen")
}
}
// Pruefung 1: Endpunkt antwortet unter mehreren gleichzeitigen Anfragen
// innerhalb definierter Zeit — kein unbeschraenktes Blockieren durch
// langsame Module (FetchTimeout begrenzt jeden Scrape).
func TestHandler_RespondsWithinBoundedTimeUnderLoad(t *testing.T) {
hangingServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
time.Sleep(10 * time.Second) // wuerde ohne timeout jede anfrage blockieren
}))
defer hangingServer.Close()
provider := func(ctx context.Context) ([]Source, error) {
return []Source{{ModuleName: "haengend", MetricsURL: hangingServer.URL}}, nil
}
agg := NewAggregator(NewCoreRegistry(), provider)
const concurrentRequests = 10
var wg sync.WaitGroup
start := time.Now()
for i := 0; i < concurrentRequests; i++ {
wg.Add(1)
go func() {
defer wg.Done()
rec := httptest.NewRecorder()
agg.Handler()(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
if rec.Code != http.StatusOK {
t.Errorf("status = %d, want 200", rec.Code)
}
}()
}
wg.Wait()
elapsed := time.Since(start)
if elapsed > FetchTimeout+3*time.Second {
t.Fatalf("%d gleichzeitige anfragen brauchten %s, erwartet deutlich unter %s durch FetchTimeout",
concurrentRequests, elapsed, FetchTimeout+3*time.Second)
}
}
+52
View File
@@ -0,0 +1,52 @@
package metrics
import (
"context"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
)
// SourceStore persistiert, welche Module ihre Metriken unter welcher URL
// bereitstellen — dieselbe Postgres-basierte "kein Code-Deploy noetig"-
// Konvention wie internal/statuspage.Store.RegisterTarget (OPS-02): ein neu
// registriertes Modul erscheint automatisch in der Aggregation, sobald es
// hier eingetragen ist (Akzeptanzkriterium 2 / Pruefung 2).
type SourceStore struct {
pool *pgxpool.Pool
}
func NewSourceStore(pool *pgxpool.Pool) *SourceStore {
return &SourceStore{pool: pool}
}
func (s *SourceStore) RegisterSource(ctx context.Context, moduleName, metricsURL string) error {
_, err := s.pool.Exec(ctx, `
INSERT INTO metrics_sources (module_name, metrics_url)
VALUES ($1, $2)
ON CONFLICT (module_name) DO UPDATE SET metrics_url = $2
`, moduleName, metricsURL)
if err != nil {
return fmt.Errorf("metrik-quelle speichern: %w", err)
}
return nil
}
// Provide implementiert SourceProvider direkt aus der Datenbank.
func (s *SourceStore) Provide(ctx context.Context) ([]Source, error) {
rows, err := s.pool.Query(ctx, `SELECT module_name, metrics_url FROM metrics_sources ORDER BY module_name`)
if err != nil {
return nil, fmt.Errorf("metrik-quellen auflisten: %w", err)
}
defer rows.Close()
var out []Source
for rows.Next() {
var src Source
if err := rows.Scan(&src.ModuleName, &src.MetricsURL); err != nil {
return nil, fmt.Errorf("metrik-quelle lesen: %w", err)
}
out = append(out, src)
}
return out, rows.Err()
}
+60
View File
@@ -0,0 +1,60 @@
package metrics
import (
"context"
"fmt"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
func setupSourceStoreTest(t *testing.T) (*SourceStore, 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 metrics_sources (module_name TEXT PRIMARY KEY, metrics_url TEXT NOT NULL)
`); err != nil {
t.Fatalf("schema: %v", err)
}
cleanup := func() { pool.Close() }
return NewSourceStore(pool), cleanup
}
// Akzeptanzkriterium 2 / Pruefung 2 auf Persistenz-Ebene: eine ueber die
// Datenbank registrierte Quelle ist sofort ueber Provide() sichtbar — genau
// der Mechanismus, der ein neues Modul ohne Core-Codeaenderung erscheinen
// laesst.
func TestSourceStore_RegisterSourceAppearsInProvide(t *testing.T) {
store, cleanup := setupSourceStoreTest(t)
defer cleanup()
ctx := context.Background()
name := fmt.Sprintf("modul-%d", time.Now().UnixNano())
if err := store.RegisterSource(ctx, name, "http://example.invalid/metrics"); err != nil {
t.Fatalf("registersource: %v", err)
}
sources, err := store.Provide(ctx)
if err != nil {
t.Fatalf("provide: %v", err)
}
found := false
for _, s := range sources {
if s.ModuleName == name {
found = true
}
}
if !found {
t.Fatalf("erwartet %s in provide()-ergebnis, habe: %+v", name, sources)
}
}
+99
View File
@@ -0,0 +1,99 @@
package moduleregistry
import (
"context"
"crypto/rand"
"crypto/sha256"
"crypto/subtle"
"encoding/hex"
"errors"
"fmt"
"github.com/jackc/pgx/v5"
)
var (
ErrModuleNotRegistered = errors.New("moduleregistry: modul muss vor provisionierung registriert sein")
ErrInvalidCredential = errors.New("moduleregistry: ungueltiges oder fehlendes service-credential")
)
// Provision stellt ein Service-Credential (Client-ID + Secret) fuer eine
// Modul-Instanz aus (Akzeptanzkriterium 4). Das Secret wird NUR beim
// Ausstellen im Klartext zurueckgegeben, gespeichert wird ausschliesslich
// dessen SHA-256-Hash.
func (r *Registry) Provision(ctx context.Context, moduleName string) (clientID, secret string, err error) {
if _, err := r.Get(ctx, moduleName); err != nil {
if errors.Is(err, ErrModuleNotFound) {
return "", "", ErrModuleNotRegistered
}
return "", "", err
}
clientID, err = randomToken(16)
if err != nil {
return "", "", fmt.Errorf("client-id erzeugen: %w", err)
}
secret, err = randomToken(32)
if err != nil {
return "", "", fmt.Errorf("secret erzeugen: %w", err)
}
hash := hashSecret(secret)
_, err = r.pool.Exec(ctx, `
INSERT INTO module_credentials (module_name, client_id, secret_hash, issued_at)
VALUES ($1, $2, $3, now())
ON CONFLICT (module_name) DO UPDATE SET client_id = $2, secret_hash = $3, issued_at = now()
`, moduleName, clientID, hash)
if err != nil {
return "", "", fmt.Errorf("credential speichern: %w", err)
}
return clientID, secret, nil
}
// Authenticate prueft ein Service-Credential timing-safe (Referenzmuster
// siehe AUD-02) — Aufrufe ohne gueltiges Credential werden abgelehnt
// (Akzeptanzkriterium 4 / Pruefung 4).
func (r *Registry) Authenticate(ctx context.Context, clientID, secret string) (moduleName string, ok bool, err error) {
if clientID == "" || secret == "" {
return "", false, nil
}
var storedHash []byte
err = r.pool.QueryRow(ctx, `
SELECT module_name, secret_hash FROM module_credentials WHERE client_id = $1
`, clientID).Scan(&moduleName, &storedHash)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return "", false, nil
}
return "", false, fmt.Errorf("credential lesen: %w", err)
}
if !timingSafeEqual(hashSecret(secret), storedHash) {
return "", false, nil
}
return moduleName, true, nil
}
func randomToken(n int) (string, error) {
buf := make([]byte, n)
if _, err := rand.Read(buf); err != nil {
return "", err
}
return hex.EncodeToString(buf), nil
}
func hashSecret(secret string) []byte {
sum := sha256.Sum256([]byte(secret))
return sum[:]
}
// timingSafeEqual folgt derselben Referenzimplementierung wie AUD-02
// (subtle.ConstantTimeCompare) — projektweite Konvention fuer jeden
// sicherheitsrelevanten Vergleich.
func timingSafeEqual(a, b []byte) bool {
if len(a) != len(b) {
return false
}
return subtle.ConstantTimeCompare(a, b) == 1
}
+46
View File
@@ -0,0 +1,46 @@
package moduleregistry
import "net/http"
// RequireActiveModule weist Anfragen an ein nicht aktiviertes Modul ZENTRAL
// ab, bevor der eigentliche Modul-Handler erreicht wird (Akzeptanzkriterium 2 /
// Pruefung 1) — Casbin-Prinzip: Durchsetzung als Middleware statt verstreuter
// Pruefungen in jedem Handler. tenantSlug/moduleName werden hier ueber
// Query-Parameter gelesen (echte Extraktion aus JWT/Tenant-Kontext ist
// API-05/TEN-06, nicht Teil dieser Kachel).
func (r *Registry) RequireActiveModule(moduleName string, next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, req *http.Request) {
tenantSlug := req.URL.Query().Get("tenant")
active, err := r.IsActive(req.Context(), tenantSlug, moduleName)
if err != nil {
http.Error(w, "aktivierungspruefung fehlgeschlagen", http.StatusInternalServerError)
return
}
if !active {
http.Error(w, "modul nicht aktiviert", http.StatusForbidden)
return
}
next(w, req)
}
}
// RequireServiceCredential authentifiziert eine Modul-Instanz ueber ihr
// Service-Credential (X-Client-Id/X-Client-Secret-Header) BEVOR der
// eigentliche Handler erreicht wird (Akzeptanzkriterium 4 / Pruefung 4).
func (r *Registry) RequireServiceCredential(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, req *http.Request) {
clientID := req.Header.Get("X-Client-Id")
secret := req.Header.Get("X-Client-Secret")
_, ok, err := r.Authenticate(req.Context(), clientID, secret)
if err != nil {
http.Error(w, "authentifizierung fehlgeschlagen", http.StatusInternalServerError)
return
}
if !ok {
http.Error(w, ErrInvalidCredential.Error(), http.StatusUnauthorized)
return
}
next(w, req)
}
}
+120
View File
@@ -0,0 +1,120 @@
// Package moduleregistry implementiert Core API-02: die Registry, in der
// sich Fachmodule (DMS, Mail, weitere) mit Metadaten eintragen, gekoppelt an
// die Aktivierungspruefung aus LIC-02 (Feature-Flags). Zusaetzlich
// authentifiziert die Registry Modul-Instanzen selbst ueber ein bei
// Provisionierung ausgestelltes Service-Credential.
package moduleregistry
import (
"context"
"errors"
"fmt"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
)
var (
ErrMissingName = errors.New("moduleregistry: name darf nicht leer sein")
ErrMissingVersion = errors.New("moduleregistry: version darf nicht leer sein")
ErrModuleNotFound = errors.New("moduleregistry: modul nicht registriert")
)
type Module struct {
Name string
Version string
RequiredFlags []string
}
type Registry struct {
pool *pgxpool.Pool
flags *flag.Service
}
func NewRegistry(pool *pgxpool.Pool, flags *flag.Service) *Registry {
return &Registry{pool: pool, flags: flags}
}
// Register traegt ein Modul mit Name, Version und benoetigten Feature-Flags
// ein (Akzeptanzkriterium 1). Fehlende Pflichtangaben werden abgewiesen
// (Akzeptanzkriterium 1 / Pruefung 2). Erneutes Register desselben Namens
// aktualisiert Version/Flags (Redeploy-Fall).
func (r *Registry) Register(ctx context.Context, name, version string, requiredFlags []string) (Module, error) {
if name == "" {
return Module{}, ErrMissingName
}
if version == "" {
return Module{}, ErrMissingVersion
}
if requiredFlags == nil {
requiredFlags = []string{}
}
_, err := r.pool.Exec(ctx, `
INSERT INTO modules (name, version, required_flags, registered_at)
VALUES ($1, $2, $3, now())
ON CONFLICT (name) DO UPDATE SET version = $2, required_flags = $3, registered_at = now()
`, name, version, requiredFlags)
if err != nil {
return Module{}, fmt.Errorf("modul registrieren: %w", err)
}
return Module{Name: name, Version: version, RequiredFlags: requiredFlags}, nil
}
func (r *Registry) Get(ctx context.Context, name string) (Module, error) {
var m Module
m.Name = name
err := r.pool.QueryRow(ctx, `
SELECT version, required_flags FROM modules WHERE name = $1
`, name).Scan(&m.Version, &m.RequiredFlags)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return Module{}, ErrModuleNotFound
}
return Module{}, fmt.Errorf("modul lesen: %w", err)
}
return m, nil
}
// List liefert alle registrierten Module (Akzeptanzkriterium 3: ueber API
// abfragbar, z.B. fuer Statusseite/Lizenzoberflaeche).
func (r *Registry) List(ctx context.Context) ([]Module, error) {
rows, err := r.pool.Query(ctx, `SELECT name, version, required_flags FROM modules ORDER BY name`)
if err != nil {
return nil, fmt.Errorf("module auflisten: %w", err)
}
defer rows.Close()
var out []Module
for rows.Next() {
var m Module
if err := rows.Scan(&m.Name, &m.Version, &m.RequiredFlags); err != nil {
return nil, fmt.Errorf("modul lesen: %w", err)
}
out = append(out, m)
}
return out, rows.Err()
}
// IsActive prueft, ob ein registriertes Modul fuer einen Tenant aktiviert
// ist: registriert UND alle benoetigten Feature-Flags sind fuer diesen
// Tenant aktiv (Akzeptanzkriterium 2). Ein nicht registriertes Modul gilt
// immer als nicht aktiv.
func (r *Registry) IsActive(ctx context.Context, tenantSlug, moduleName string) (bool, error) {
m, err := r.Get(ctx, moduleName)
if err != nil {
if errors.Is(err, ErrModuleNotFound) {
return false, nil
}
return false, err
}
for _, flagKey := range m.RequiredFlags {
if !r.flags.IsEnabled(ctx, tenantSlug, flagKey) {
return false, nil
}
}
return true, nil
}
+304
View File
@@ -0,0 +1,304 @@
package moduleregistry
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
)
func setupTest(t *testing.T) (*Registry, *flag.Store, 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 feature_flags (
key TEXT PRIMARY KEY, enabled BOOLEAN NOT NULL DEFAULT false,
rollout_percentage INT NOT NULL DEFAULT 0, target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS modules (
name TEXT PRIMARY KEY, version TEXT NOT NULL CHECK (version <> ''),
required_flags TEXT[] NOT NULL DEFAULT '{}', registered_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS module_credentials (
module_name TEXT PRIMARY KEY REFERENCES modules(name),
client_id TEXT NOT NULL UNIQUE, secret_hash BYTEA NOT NULL,
issued_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
`); err != nil {
t.Fatalf("schema: %v", err)
}
flagStore := flag.NewStore(pool)
// Kurze TTL, damit Tests, die den Flag-Store direkt aendern (an
// Registry.IsActive vorbei), den neuen Stand ohne manuelles Invalidate
// zuverlaessig sehen.
flagService := flag.NewService(flagStore, 10*time.Millisecond)
registry := NewRegistry(pool, flagService)
cleanup := func() { pool.Close() }
return registry, flagStore, cleanup
}
func uniqueModuleName(t *testing.T) string {
return fmt.Sprintf("dms_%d", time.Now().UnixNano())
}
// Akzeptanzkriterium 1 + Pruefung 2: fehlende Pflichtangaben abgewiesen.
func TestRegister_RejectsMissingFields(t *testing.T) {
registry, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
if _, err := registry.Register(ctx, "", "1.0", nil); !errors.Is(err, ErrMissingName) {
t.Fatalf("erwartet ErrMissingName, habe %v", err)
}
if _, err := registry.Register(ctx, "dms", "", nil); !errors.Is(err, ErrMissingVersion) {
t.Fatalf("erwartet ErrMissingVersion, habe %v", err)
}
}
func TestRegister_AndGet(t *testing.T) {
registry, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
name := uniqueModuleName(t)
m, err := registry.Register(ctx, name, "1.2.0", []string{"dms_enabled"})
if err != nil {
t.Fatalf("register: %v", err)
}
if m.Version != "1.2.0" || len(m.RequiredFlags) != 1 {
t.Fatalf("unerwartet: %+v", m)
}
got, err := registry.Get(ctx, name)
if err != nil {
t.Fatalf("get: %v", err)
}
if got.Version != "1.2.0" {
t.Fatalf("get version = %q", got.Version)
}
}
// Akzeptanzkriterium 2 + 3 + Pruefung 3: konsistente Daten nach
// Aktivierung/Deaktivierung eines Moduls.
func TestIsActive_ReflectsFlagStateConsistently(t *testing.T) {
registry, flagStore, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
name := uniqueModuleName(t)
flagKey := name + "_enabled"
if _, err := registry.Register(ctx, name, "1.0", []string{flagKey}); err != nil {
t.Fatalf("register: %v", err)
}
active, err := registry.IsActive(ctx, "acme", name)
if err != nil {
t.Fatalf("is active (vor flag): %v", err)
}
if active {
t.Fatal("erwartet nicht aktiv, solange flag nicht gesetzt ist")
}
if err := flagStore.Set(ctx, flag.Flag{Key: flagKey, Enabled: true}); err != nil {
t.Fatalf("flag setzen: %v", err)
}
time.Sleep(20 * time.Millisecond) // TTL abwarten
active, err = registry.IsActive(ctx, "acme", name)
if err != nil {
t.Fatalf("is active (nach flag an): %v", err)
}
if !active {
t.Fatal("erwartet aktiv, nachdem flag aktiviert wurde")
}
if err := flagStore.Set(ctx, flag.Flag{Key: flagKey, Enabled: false}); err != nil {
t.Fatalf("flag zuruecksetzen: %v", err)
}
time.Sleep(20 * time.Millisecond) // TTL abwarten
active, err = registry.IsActive(ctx, "acme", name)
if err != nil {
t.Fatalf("is active (nach flag aus): %v", err)
}
if active {
t.Fatal("erwartet wieder nicht aktiv, nachdem flag deaktiviert wurde")
}
}
func TestIsActive_UnregisteredModuleIsNeverActive(t *testing.T) {
registry, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
active, err := registry.IsActive(ctx, "acme", "nie-registriert")
if err != nil {
t.Fatalf("is active: %v", err)
}
if active {
t.Fatal("unregistriertes modul darf nie aktiv sein")
}
}
// Akzeptanzkriterium 2 + Pruefung 1: Anfrage an deaktiviertes Modul wird
// zentral abgewiesen, BEVOR die Modul-Logik erreicht wird.
func TestRequireActiveModule_BlocksBeforeHandler(t *testing.T) {
registry, flagStore, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
name := uniqueModuleName(t)
flagKey := name + "_enabled"
if _, err := registry.Register(ctx, name, "1.0", []string{flagKey}); err != nil {
t.Fatalf("register: %v", err)
}
handlerReached := false
handler := registry.RequireActiveModule(name, func(w http.ResponseWriter, r *http.Request) {
handlerReached = true
w.WriteHeader(http.StatusOK)
})
req := httptest.NewRequest(http.MethodGet, "/modul?tenant=acme", nil)
rec := httptest.NewRecorder()
handler(rec, req)
if rec.Code != http.StatusForbidden {
t.Fatalf("status = %d, want 403", rec.Code)
}
if handlerReached {
t.Fatal("handler haette bei deaktiviertem modul NICHT erreicht werden duerfen")
}
if err := flagStore.Set(ctx, flag.Flag{Key: flagKey, Enabled: true}); err != nil {
t.Fatalf("flag setzen: %v", err)
}
req2 := httptest.NewRequest(http.MethodGet, "/modul?tenant=acme", nil)
rec2 := httptest.NewRecorder()
handler(rec2, req2)
if rec2.Code != http.StatusOK {
t.Fatalf("status nach aktivierung = %d, want 200", rec2.Code)
}
if !handlerReached {
t.Fatal("handler haette bei aktiviertem modul erreicht werden muessen")
}
}
// Akzeptanzkriterium 4 + Pruefung 4: gueltiges/ungueltiges Service-Credential.
func TestProvisionAndAuthenticate(t *testing.T) {
registry, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
name := uniqueModuleName(t)
if _, err := registry.Register(ctx, name, "1.0", nil); err != nil {
t.Fatalf("register: %v", err)
}
clientID, secret, err := registry.Provision(ctx, name)
if err != nil {
t.Fatalf("provision: %v", err)
}
if clientID == "" || secret == "" {
t.Fatal("erwartet nicht-leere client-id/secret")
}
moduleName, ok, err := registry.Authenticate(ctx, clientID, secret)
if err != nil {
t.Fatalf("authenticate (korrekt): %v", err)
}
if !ok || moduleName != name {
t.Fatalf("erwartet erfolgreiche authentifizierung fuer %q, habe ok=%v moduleName=%q", name, ok, moduleName)
}
_, ok, err = registry.Authenticate(ctx, clientID, "falsches-secret")
if err != nil {
t.Fatalf("authenticate (falsch): %v", err)
}
if ok {
t.Fatal("erwartet fehlschlag bei falschem secret")
}
_, ok, err = registry.Authenticate(ctx, "unbekannte-client-id", secret)
if err != nil {
t.Fatalf("authenticate (unbekannt): %v", err)
}
if ok {
t.Fatal("erwartet fehlschlag bei unbekannter client-id")
}
}
func TestProvision_RequiresRegisteredModule(t *testing.T) {
registry, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
if _, _, err := registry.Provision(ctx, "nie-registriert"); !errors.Is(err, ErrModuleNotRegistered) {
t.Fatalf("erwartet ErrModuleNotRegistered, habe %v", err)
}
}
func TestRequireServiceCredential_RejectsInvalidAcceptsValid(t *testing.T) {
registry, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
name := uniqueModuleName(t)
if _, err := registry.Register(ctx, name, "1.0", nil); err != nil {
t.Fatalf("register: %v", err)
}
clientID, secret, err := registry.Provision(ctx, name)
if err != nil {
t.Fatalf("provision: %v", err)
}
handler := registry.RequireServiceCredential(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
})
// Fehlendes Credential.
req := httptest.NewRequest(http.MethodPost, "/service-aufruf", nil)
rec := httptest.NewRecorder()
handler(rec, req)
if rec.Code != http.StatusUnauthorized {
t.Fatalf("ohne credential: status = %d, want 401", rec.Code)
}
// Falsches Secret.
req2 := httptest.NewRequest(http.MethodPost, "/service-aufruf", nil)
req2.Header.Set("X-Client-Id", clientID)
req2.Header.Set("X-Client-Secret", "falsch")
rec2 := httptest.NewRecorder()
handler(rec2, req2)
if rec2.Code != http.StatusUnauthorized {
t.Fatalf("falsches secret: status = %d, want 401", rec2.Code)
}
// Gueltiges Credential.
req3 := httptest.NewRequest(http.MethodPost, "/service-aufruf", nil)
req3.Header.Set("X-Client-Id", clientID)
req3.Header.Set("X-Client-Secret", secret)
rec3 := httptest.NewRecorder()
handler(rec3, req3)
if rec3.Code != http.StatusOK {
t.Fatalf("gueltiges credential: status = %d, want 200", rec3.Code)
}
}
-1
View File
@@ -1 +0,0 @@
DROP TABLE IF EXISTS audit_events;
-17
View File
@@ -1,17 +0,0 @@
-- Zentrales Audit-Log-Modell (AUD-01, siehe core-kanban/tickets/AUD-01.md).
-- Getrennt vom allgemeinen Anwendungs-Log (Akzeptanzkriterium 2): eigene
-- Tabelle, eigenes Paket (internal/audit), kein Log-Framework.
-- tenant_slug ist NOT NULL + darf nicht leer sein (Akzeptanzkriterium 2 /
-- Pruefung 2) — mandantenuebergreifende Ereignisse nutzen den reservierten
-- Wert 'system', niemals NULL oder leeren String.
CREATE TABLE audit_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
tenant_slug TEXT NOT NULL CHECK (tenant_slug <> ''),
actor TEXT NOT NULL CHECK (actor <> ''),
action TEXT NOT NULL CHECK (action <> ''),
target TEXT NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'::jsonb
);
CREATE INDEX audit_events_tenant_slug_idx ON audit_events (tenant_slug, occurred_at);
+1
View File
@@ -0,0 +1 @@
DROP TABLE IF EXISTS feature_flags;
+10
View File
@@ -0,0 +1,10 @@
-- Feature-Flags zentral je Mandant/Zielgruppe (LIC-02, siehe core-kanban/tickets/LIC-02.md).
-- Lebt in der Registry-DB, nicht pro Tenant-Datenbank — Flags sind eine
-- Core-weite Konfiguration, keine Mandanten-Geschaeftsdaten.
CREATE TABLE feature_flags (
key TEXT PRIMARY KEY,
enabled BOOLEAN NOT NULL DEFAULT false,
rollout_percentage INT NOT NULL DEFAULT 0 CHECK (rollout_percentage BETWEEN 0 AND 100),
target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
@@ -1,3 +0,0 @@
DROP TRIGGER IF EXISTS audit_events_no_delete ON audit_events;
DROP TRIGGER IF EXISTS audit_events_no_update ON audit_events;
DROP FUNCTION IF EXISTS audit_events_prevent_mutation();
-17
View File
@@ -1,17 +0,0 @@
-- Audit-Log technisch gegen Aenderung/Loeschung absichern (AUD-02, siehe
-- core-kanban/tickets/AUD-02.md). Ein Trigger statt nur GRANT/REVOKE, damit
-- der Schutz unabhaengig davon greift, mit welcher Rolle verbunden wird
-- (Akzeptanzkriterium 1: "auf Datenbankebene technisch unterbunden").
CREATE FUNCTION audit_events_prevent_mutation() RETURNS TRIGGER AS $$
BEGIN
RAISE EXCEPTION 'audit_events ist append-only: % ist nicht erlaubt', TG_OP;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER audit_events_no_update
BEFORE UPDATE ON audit_events
FOR EACH ROW EXECUTE FUNCTION audit_events_prevent_mutation();
CREATE TRIGGER audit_events_no_delete
BEFORE DELETE ON audit_events
FOR EACH ROW EXECUTE FUNCTION audit_events_prevent_mutation();
+2
View File
@@ -0,0 +1,2 @@
DROP TABLE IF EXISTS module_credentials;
DROP TABLE IF EXISTS modules;
+18
View File
@@ -0,0 +1,18 @@
-- Modul-Registry & Aktivierungspruefung (API-02, siehe core-kanban/tickets/API-02.md).
CREATE TABLE modules (
name TEXT PRIMARY KEY,
version TEXT NOT NULL CHECK (version <> ''),
required_flags TEXT[] NOT NULL DEFAULT '{}',
registered_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- Service-Credential je Modul-Instanz, bei Provisionierung ausgestellt
-- (Akzeptanzkriterium 4). secret_hash enthaelt NIEMALS das Secret im
-- Klartext, nur dessen SHA-256-Hash (Timing-safe-Vergleich beim Login,
-- Referenzmuster siehe AUD-02).
CREATE TABLE module_credentials (
module_name TEXT PRIMARY KEY REFERENCES modules(name),
client_id TEXT NOT NULL UNIQUE,
secret_hash BYTEA NOT NULL,
issued_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
+1
View File
@@ -0,0 +1 @@
DROP TABLE metrics_sources;
+7
View File
@@ -0,0 +1,7 @@
-- Metrics-Aggregation ueber Module hinweg (OPS-03, siehe
-- core-kanban/tickets/OPS-03.md) — welches Modul liefert seine Kennzahlen
-- unter welcher /metrics-URL.
CREATE TABLE metrics_sources (
module_name TEXT PRIMARY KEY,
metrics_url TEXT NOT NULL
);
@@ -1 +0,0 @@
DROP TABLE IF EXISTS security_confirmations;
@@ -1,14 +0,0 @@
-- Vier-Augen-Prinzip fuer sicherheitskritische Entscheidungen (AUD-02
-- Akzeptanzkriterium 2), Vorbild: archivdms FOR-UPDATE-Lock + Timing-safe
-- Vergleich. code_hash speichert NIEMALS den Bestaetigungscode im Klartext.
CREATE TABLE security_confirmations (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
action TEXT NOT NULL,
target TEXT NOT NULL,
requested_by TEXT NOT NULL,
code_hash BYTEA NOT NULL,
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'confirmed', 'rejected')),
confirmed_by TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
confirmed_at TIMESTAMPTZ
);
-13
View File
@@ -1,24 +1,11 @@
#!/usr/bin/env bash
# Setzt die nexarch-Testumgebung zurueck: loescht die geteilte
# Registry-Tabelle "tenants" in der postgres-Wartungsdatenbank sowie alle
# tenant_*-Datenbanken. Noetig, weil verschiedene Feature-Branches
# unterschiedliche Registry-Schemata erwarten, aber dieselbe physische
# Postgres-Instanz auf dem Testhost teilen (siehe [[project-nexarch-test-infra]]).
#
# Aufruf: NEXARCH_TEST_DB_PASSWORD=... ./scripts/reset-test-env.sh
set -euo pipefail
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
ROLE="nexarch_test"
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 audit_events CASCADE;"
dbs=$(psql -h localhost -U "$ROLE" -d postgres -tAc "SELECT datname FROM pg_database WHERE datname LIKE 'tenant\_%' ESCAPE '\'")
for db in $dbs; do
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS \"${db}\";"
done
echo "Testumgebung zurueckgesetzt: registry-tabelle + $(echo "$dbs" | grep -c . || true) tenant-datenbank(en) entfernt."
-12
View File
@@ -1,24 +1,12 @@
#!/usr/bin/env bash
# Ein-Kommando-Pruefung fuer den aktuellen Code-Stand auf dem Testhost:
# Registry+Tenant-DBs zuruecksetzen, dann build/vet/test in einem Rutsch.
# -p 1 ist Pflicht, da mehrere Pakete dieselbe physische Registry-Tabelle auf
# dem Testhost teilen (siehe [[project-nexarch-test-infra]]).
#
# Aufruf: NEXARCH_TEST_DB_PASSWORD=... ./scripts/run-checks.sh
set -euo pipefail
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
cd "$(dirname "$0")/.."
NEXARCH_TEST_DB_PASSWORD="$PASS" bash scripts/reset-test-env.sh
export TEST_ADMIN_DSN="postgresql://nexarch_test:${PASS}@localhost:5432/postgres?sslmode=disable"
echo "== go build =="
go build ./...
echo "== go vet =="
go vet ./...
echo "== go test (-p 1) =="
go test ./... -p 1 -count=1