Compare commits

..
22 changed files with 1397 additions and 887 deletions
-82
View File
@@ -1,82 +0,0 @@
// kek-api ist der Aufrufpunkt fuer API-12: startet den bereits fertigen
// API-10-Handler (internal/kek) als eigenstaendigen HTTP-Dienst.
// REINES WIRING — keine Aenderung an internal/kek/, internal/moduleregistry/
// oder internal/tenant/.
package main
import (
"context"
"log"
"net/http"
"os"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
"gitea.perlbach24.de/scripte/nexarch/internal/kek"
"gitea.perlbach24.de/scripte/nexarch/internal/moduleregistry"
"gitea.perlbach24.de/scripte/nexarch/internal/tenant"
)
// tenantResolverAdapter erfüllt kek.TenantResolver über den bestehenden
// tenant.Registry.GetBySlug-Zugriff — kein neuer Tenant-Code, nur
// Signatur-Anpassung.
type tenantResolverAdapter struct{ registry *tenant.Registry }
func (a tenantResolverAdapter) ResolveTenantID(ctx context.Context, tenantSlug string) (string, error) {
t, err := a.registry.GetBySlug(ctx, tenantSlug)
if err != nil {
return "", err
}
return t.ID, nil
}
func requireEnv(name string) string {
v := os.Getenv(name)
if v == "" {
log.Fatalf("%s muss gesetzt sein", name)
}
return v
}
func main() {
registryDSN := requireEnv("NEXARCH_KEK_REGISTRY_DSN")
masterKeyEnvVar := os.Getenv("NEXARCH_KEK_MASTER_KEY_ENV")
if masterKeyEnvVar == "" {
masterKeyEnvVar = "NEXARCH_KEK_MASTER_KEY"
}
addr := os.Getenv("NEXARCH_KEK_API_LISTEN_ADDR")
if addr == "" {
addr = "127.0.0.1:8102"
}
masterKey, err := kek.LoadMasterKeyFromEnv(masterKeyEnvVar)
if err != nil {
log.Fatalf("master-key laden: %v", err)
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, registryDSN)
if err != nil {
log.Fatalf("datenbankverbindung: %v", err)
}
defer pool.Close()
flagStore := flag.NewStore(pool)
flagService := flag.NewService(flagStore, 30*time.Second)
registry := moduleregistry.NewRegistry(pool, flagService)
tenantRegistry := tenant.NewRegistry(pool)
store := kek.NewStore(pool)
handler := kek.NewHandler(store, masterKey, registry, registry, tenantResolverAdapter{registry: tenantRegistry})
mux := http.NewServeMux()
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
mux.HandleFunc("/internal/kek/tenant", handler.TenantKEKHandler)
log.Printf("kek-api: listening on %s", addr)
if err := http.ListenAndServe(addr, mux); err != nil {
log.Fatalf("http server: %v", err)
}
}
@@ -1,14 +0,0 @@
[Unit]
Description=NEXARCH Core - KEK-Bezugsdienst (API-10/API-12)
After=network.target postgresql.service
[Service]
Type=simple
User=nexarch
EnvironmentFile=/etc/nexarch/kek-api.env
ExecStart=__INSTALL_DIR__/bin/kek-api
Restart=on-failure
StandardOutput=journal
[Install]
WantedBy=multi-user.target
-61
View File
@@ -1,61 +0,0 @@
# API-12 Prüfprotokoll: KEK-Bezugsdienst starten (API-10 als laufender Dienst)
Voraussetzung API-10 bereits Fertig, hier UNVERÄNDERT.
## Reines Wiring, keine neue Logik
`git diff --stat internal/kek/ internal/moduleregistry/ internal/tenant/`
liefert KEINEN Diff. `cmd/kek-api/main.go` setzt ausschließlich
bestehende Konstruktoren zusammen; `tenantResolverAdapter` bildet nur
`tenant.Registry.GetBySlug` auf `kek.TenantResolver` ab (Signatur-
Anpassung, kein neuer Fachcode).
## Umsetzung
- `cmd/kek-api/main.go` `POST /internal/kek/tenant?tenant=<slug>`,
authentifiziert über dasselbe Service-Credential-Verfahren wie jeder
andere Modul-Core-Aufruf (API-02), zusätzlich Tenant-Aktivierungs-
prüfung (identisches Muster wie in `internal/kek.Handler` bereits
vorgesehen).
- `deploy/systemd/nexarch-kek-api.service.tmpl`.
## Prüfungen
| # | Prüfung | Ergebnis |
|---|---|---|
| 1 | Dienst startet und bleibt stabil (systemctl status aktiv) | **bestanden** real auf 131: `nexarch-kek-api.service` aktiv |
| 2 | Realer Aufruf mit gültigem Service-Credential liefert den erwarteten Tenant-KEK, ohne/mit falschem Credential wird abgelehnt | **bestanden** real per `curl`: echtes Modul registriert+provisioniert, echter Tenant-KEK über `kek.Store.CreateForTenant` erzeugt (Klartext-Hex zum Vergleich notiert) — Aufruf mit korrektem Credential liefert exakt denselben KEK (Base64-dekodiert übereinstimmend mit dem erzeugten Hex-Wert verifiziert); Aufruf mit falschem Credential → 403 |
| 3 | Code-Review: keine Änderung an internal/kek/ selbst, nur main.go+systemd neu | **bestanden** `git diff --stat` bestätigt: `internal/kek/`, `internal/moduleregistry/`, `internal/tenant/` unverändert |
## Echte Verdrahtung auf 192.168.1.131
- `kek-api` gebaut nach `/opt/nexarch-core/bin/`,
`/etc/nexarch/kek-api.env` (0600, echter zufälliger 32-Byte-
Master-Key), Dienst installiert/aktiviert.
- Reale Grant-Lücke gefunden und behoben (gleiches Muster wie zuvor):
`nexarch_core` hatte keine Rechte auf `tenant_keks``GRANT`
nachgezogen und über `information_schema.role_table_grants`
verifiziert.
- End-zu-Ende-Beweis: echtes Modul registriert, Service-Credential
provisioniert, echter Tenant + Tenant-KEK real erzeugt, `curl` gegen
den laufenden Dienst liefert exakt diesen KEK zurück (Byte-für-Byte
verglichen), falsches Credential real abgelehnt. Testdaten
anschließend entfernt.
## Build/Test-Ergebnis (192.168.1.131)
```
go build ./... -> clean
go vet ./... -> clean
golangci-lint run ./cmd/kek-api/... -> 0 issues
```
Keine neuen Go-Tests nötig (kein neuer Fachcode außer main.go/Adapter,
die eigentliche Logik ist bereits durch API-10s eigene Tests
abgedeckt).
## Gesamtergebnis
**Bestanden.** API-10 ist jetzt ein real laufender, über systemd
verwalteter Dienst — Voraussetzung für Mail ARC-02 und künftig DMS
FDN-09-Nachnutzung.
+4 -1
View File
@@ -2,7 +2,10 @@ module gitea.perlbach24.de/scripte/nexarch
go 1.22
require github.com/jackc/pgx/v5 v5.6.0
require (
github.com/golang-jwt/jwt/v5 v5.3.1
github.com/jackc/pgx/v5 v5.6.0
)
require (
github.com/jackc/pgpassfile v1.0.0 // indirect
+2
View File
@@ -1,6 +1,8 @@
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY=
github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE=
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
-113
View File
@@ -1,113 +0,0 @@
package kek
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"net/http"
)
// CredentialAuthenticator ist die schmale Schnittstelle zu API-02s
// Service-Credential-Pruefung (internal/moduleregistry.Registry.Authenticate).
type CredentialAuthenticator interface {
Authenticate(ctx context.Context, clientID, secret string) (moduleName string, ok bool, err error)
}
// ModuleActivationChecker ist die schmale Schnittstelle zu API-02s
// Aktivierungspruefung (internal/moduleregistry.Registry.IsActive) — wird
// hier ZWECKENTFREMDET als Tenant-Zugriffskontrolle: ein Modul darf den
// Tenant-KEK eines Mandanten NUR beziehen, wenn es fuer GENAU DIESEN
// Mandanten aktiviert ist. Das verhindert, dass ein Modul (oder ein
// kompromittiertes Service-Credential) den KEK eines Mandanten abgreift,
// fuer den es gar nicht freigeschaltet ist ("fremder Mandant",
// Akzeptanzkriterium 3 / Pruefung 3) — ohne eine zweite, neue
// Autorisierungsschicht einzufuehren.
type ModuleActivationChecker interface {
IsActive(ctx context.Context, tenantSlug, moduleName string) (bool, error)
}
// TenantResolver loest einen Tenant-Slug in seine interne ID auf
// (internal/tenant.Registry.GetBySlug, TEN-01).
type TenantResolver interface {
ResolveTenantID(ctx context.Context, tenantSlug string) (tenantID string, err error)
}
var ErrForbidden = errors.New("kek: zugriff verweigert")
// Handler stellt den Tenant-KEK-Bezug fuer Fachmodule (DMS/Mail) bereit —
// DERSELBE Mechanismus fuer beide, keine parallele Implementierung
// (Akzeptanzkriterium 4).
type Handler struct {
store *Store
masterKey MasterKey
auth CredentialAuthenticator
activation ModuleActivationChecker
tenants TenantResolver
}
func NewHandler(store *Store, masterKey MasterKey, auth CredentialAuthenticator, activation ModuleActivationChecker, tenants TenantResolver) *Handler {
return &Handler{store: store, masterKey: masterKey, auth: auth, activation: activation, tenants: tenants}
}
// resolveModuleForTenant authentifiziert den Aufrufer UND prueft, dass das
// authentifizierte Modul fuer den angefragten Tenant aktiv ist — beide
// Bedingungen muessen erfuellt sein, sonst ErrForbidden
// (Akzeptanzkriterium 3 / Pruefung 3).
func (h *Handler) resolveModuleForTenant(ctx context.Context, clientID, secret, tenantSlug string) error {
moduleName, ok, err := h.auth.Authenticate(ctx, clientID, secret)
if err != nil {
return err
}
if !ok {
return ErrForbidden
}
active, err := h.activation.IsActive(ctx, tenantSlug, moduleName)
if err != nil {
return err
}
if !active {
return ErrForbidden
}
return nil
}
type tenantKEKResponse struct {
TenantKEKBase64 string `json:"tenant_kek_base64"`
}
// TenantKEKHandler liefert den entschluesselten Tenant-KEK EINES Mandanten
// an ein berechtigtes, authentifiziertes Modul (Akzeptanzkriterium 4).
func (h *Handler) TenantKEKHandler(w http.ResponseWriter, r *http.Request) {
clientID := r.Header.Get("X-Nexarch-Client-Id")
secret := r.Header.Get("X-Nexarch-Client-Secret")
tenantSlug := r.URL.Query().Get("tenant")
if tenantSlug == "" {
http.Error(w, "tenant-parameter fehlt", http.StatusBadRequest)
return
}
if err := h.resolveModuleForTenant(r.Context(), clientID, secret, tenantSlug); err != nil {
if errors.Is(err, ErrForbidden) {
http.Error(w, "zugriff auf diesen mandanten verweigert", http.StatusForbidden)
return
}
http.Error(w, "interner fehler", http.StatusInternalServerError)
return
}
tenantID, err := h.tenants.ResolveTenantID(r.Context(), tenantSlug)
if err != nil {
http.Error(w, "mandant nicht gefunden", http.StatusNotFound)
return
}
plainKEK, err := h.store.GetDecrypted(r.Context(), tenantID, h.masterKey)
if err != nil {
http.Error(w, "tenant-kek konnte nicht ermittelt werden", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(tenantKEKResponse{TenantKEKBase64: base64.StdEncoding.EncodeToString(plainKEK)})
}
-350
View File
@@ -1,350 +0,0 @@
package kek
import (
"bytes"
"context"
"encoding/base64"
"fmt"
"net/http"
"net/http/httptest"
"os"
"strings"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
"gitea.perlbach24.de/scripte/nexarch/internal/moduleregistry"
"gitea.perlbach24.de/scripte/nexarch/internal/tenant"
)
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 tenants (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), slug TEXT NOT NULL UNIQUE, name TEXT NOT NULL,
db_name TEXT NOT NULL UNIQUE, db_dsn TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'active',
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS tenant_keks (
tenant_id UUID PRIMARY KEY REFERENCES tenants(id), wrapped_kek BYTEA NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), rotated_at TIMESTAMPTZ
);
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)
}
cleanup := func() { pool.Close() }
return NewStore(pool), pool, cleanup
}
func newMasterKey(t *testing.T) MasterKey {
t.Helper()
key, err := generateRandomKey()
if err != nil {
t.Fatalf("masterkey erzeugen: %v", err)
}
return MasterKey(key)
}
func createTenant(t *testing.T, pool *pgxpool.Pool, slug string) string {
t.Helper()
var id string
err := pool.QueryRow(context.Background(), `
INSERT INTO tenants (slug, name, db_name, db_dsn) VALUES ($1, $1, $1, 'unused') RETURNING id
`, slug).Scan(&id)
if err != nil {
t.Fatalf("tenant anlegen: %v", err)
}
return id
}
func uniqueSlug(prefix string) string {
return fmt.Sprintf("%s_%d", prefix, time.Now().UnixNano())
}
// Akzeptanzkriterium 1: LoadMasterKeyFromEnv liest ausschliesslich aus der
// Umgebungsvariable, niemals aus Code/DB.
func TestLoadMasterKeyFromEnv(t *testing.T) {
const envVar = "NEXARCH_TEST_MASTER_KEY_API10"
t.Cleanup(func() { os.Unsetenv(envVar) })
if _, err := LoadMasterKeyFromEnv(envVar); err == nil {
t.Fatal("erwartet fehler, wenn umgebungsvariable nicht gesetzt ist")
}
os.Setenv(envVar, "zu-kurz")
if _, err := LoadMasterKeyFromEnv(envVar); err == nil {
t.Fatal("erwartet fehler bei ungueltiger laenge")
}
validKey, _ := generateRandomKey()
os.Setenv(envVar, base64.StdEncoding.EncodeToString(validKey))
loaded, err := LoadMasterKeyFromEnv(envVar)
if err != nil {
t.Fatalf("laden mit gueltigem key: %v", err)
}
if !bytes.Equal(loaded, validKey) {
t.Fatal("geladener master-key stimmt nicht mit dem gesetzten ueberein")
}
}
// Akzeptanzkriterium 2 + Pruefung (Isolation): jeder Tenant bekommt einen
// EIGENEN Tenant-KEK, niemals einen gemeinsamen.
func TestCreateForTenant_EachTenantGetsDistinctKEK(t *testing.T) {
store, pool, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
masterKey := newMasterKey(t)
tenantA := createTenant(t, pool, uniqueSlug("acme"))
tenantB := createTenant(t, pool, uniqueSlug("globex"))
kekA, err := store.CreateForTenant(ctx, tenantA, masterKey)
if err != nil {
t.Fatalf("create a: %v", err)
}
kekB, err := store.CreateForTenant(ctx, tenantB, masterKey)
if err != nil {
t.Fatalf("create b: %v", err)
}
if bytes.Equal(kekA, kekB) {
t.Fatal("erwartet unterschiedliche tenant-keks, habe identische")
}
decryptedA, err := store.GetDecrypted(ctx, tenantA, masterKey)
if err != nil {
t.Fatalf("decrypt a: %v", err)
}
if !bytes.Equal(decryptedA, kekA) {
t.Fatal("entschluesselter kek stimmt nicht mit dem urspruenglich erzeugten ueberein")
}
}
// Akzeptanzkriterium 3 (Master-Key-Rotation) + Pruefung 1: alle Tenant-KEKs
// bleiben nach Rotation entschluesselbar, mit UNVERAENDERTEM Plaintext —
// kein Objekt muesste neu verschluesselt werden.
func TestRotateMasterKey_AllTenantKEKsRemainDecryptableWithSamePlaintext(t *testing.T) {
store, pool, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
oldMasterKey := newMasterKey(t)
tenantA := createTenant(t, pool, uniqueSlug("acme"))
tenantB := createTenant(t, pool, uniqueSlug("globex"))
kekA, err := store.CreateForTenant(ctx, tenantA, oldMasterKey)
if err != nil {
t.Fatalf("create a: %v", err)
}
kekB, err := store.CreateForTenant(ctx, tenantB, oldMasterKey)
if err != nil {
t.Fatalf("create b: %v", err)
}
newMasterKeyVal := newMasterKey(t)
_, failed, err := store.RotateMasterKey(ctx, oldMasterKey, newMasterKeyVal)
if err != nil {
t.Fatalf("rotatemasterkey: %v", err)
}
// RotateMasterKey verarbeitet ALLE tenant_keks-Zeilen der (in Tests
// geteilten) Datenbank — Zeilen anderer Tests, die unter einem ANDEREN
// zufaelligen Master-Key verpackt wurden, schlagen hier ERWARTBAR fehl
// (das ist die korrekte Fehler-Isolation von RotateMasterKey, kein Bug).
// Relevant ist nur, dass GENAU DIESE beiden Tenants NICHT scheitern.
for _, id := range failed {
if id == tenantA || id == tenantB {
t.Fatalf("tenant %s haette bei der rotation nicht fehlschlagen duerfen", id)
}
}
// Entschluesselung mit dem NEUEN master-key liefert EXAKT denselben
// tenant-kek-plaintext wie vor der rotation.
afterA, err := store.GetDecrypted(ctx, tenantA, newMasterKeyVal)
if err != nil {
t.Fatalf("decrypt a nach rotation: %v", err)
}
if !bytes.Equal(afterA, kekA) {
t.Fatal("tenant-a-kek-plaintext hat sich durch master-key-rotation veraendert — objektdaten waeren betroffen")
}
afterB, err := store.GetDecrypted(ctx, tenantB, newMasterKeyVal)
if err != nil {
t.Fatalf("decrypt b nach rotation: %v", err)
}
if !bytes.Equal(afterB, kekB) {
t.Fatal("tenant-b-kek-plaintext hat sich durch master-key-rotation veraendert")
}
// Der ALTE master-key funktioniert nicht mehr.
if _, err := store.GetDecrypted(ctx, tenantA, oldMasterKey); err == nil {
t.Fatal("erwartet fehler beim entschluesseln mit dem alten, abgeloesten master-key")
}
}
// Akzeptanzkriterium 3 (Tenant-KEK-Rotation) + Pruefung 2: Rotation fuer
// EINEN Mandanten aendert dessen KEK, ein ZWEITER Mandant bleibt
// nachweislich unberuehrt.
func TestRotateTenantKEK_OnlyAffectsThatTenant(t *testing.T) {
store, pool, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
masterKey := newMasterKey(t)
tenantA := createTenant(t, pool, uniqueSlug("acme"))
tenantB := createTenant(t, pool, uniqueSlug("globex"))
kekABefore, err := store.CreateForTenant(ctx, tenantA, masterKey)
if err != nil {
t.Fatalf("create a: %v", err)
}
kekBBefore, err := store.CreateForTenant(ctx, tenantB, masterKey)
if err != nil {
t.Fatalf("create b: %v", err)
}
kekAAfter, err := store.RotateTenantKEK(ctx, tenantA, masterKey)
if err != nil {
t.Fatalf("rotatetenantkek: %v", err)
}
if bytes.Equal(kekAAfter, kekABefore) {
t.Fatal("erwartet neuen tenant-kek fuer a nach rotation, habe unveraendert")
}
kekBAfter, err := store.GetDecrypted(ctx, tenantB, masterKey)
if err != nil {
t.Fatalf("decrypt b nach rotation von a: %v", err)
}
if !bytes.Equal(kekBAfter, kekBBefore) {
t.Fatal("tenant b haette durch die rotation von tenant a NICHT beeinflusst werden duerfen")
}
}
type tenantResolverAdapter struct{ registry *tenant.Registry }
func (a tenantResolverAdapter) ResolveTenantID(ctx context.Context, tenantSlug string) (string, error) {
t, err := a.registry.GetBySlug(ctx, tenantSlug)
if err != nil {
return "", err
}
return t.ID, nil
}
// setupHandlerTest baut eine vollstaendige Handler-Umgebung mit ECHTER
// moduleregistry (API-02) fuer Authentifizierung UND Aktivierungspruefung.
func setupHandlerTest(t *testing.T) (*Handler, *pgxpool.Pool, *moduleregistry.Registry, string, string) {
t.Helper()
store, pool, _ := setupTest(t)
ctx := context.Background()
masterKey := newMasterKey(t)
flagService := flag.NewService(flag.NewStore(pool), 10*time.Millisecond)
moduleRegistry := moduleregistry.NewRegistry(pool, flagService)
tenantRegistry := tenant.NewRegistry(pool)
moduleName := fmt.Sprintf("dms-%d", time.Now().UnixNano())
if _, err := moduleRegistry.Register(ctx, moduleName, "1.0.0", nil); err != nil {
t.Fatalf("modul registrieren: %v", err)
}
clientID, secret, err := moduleRegistry.Provision(ctx, moduleName)
if err != nil {
t.Fatalf("credential provisionieren: %v", err)
}
handler := NewHandler(store, masterKey, moduleRegistry, moduleRegistry, tenantResolverAdapter{tenantRegistry})
return handler, pool, moduleRegistry, clientID, secret
}
// Akzeptanzkriterium 3 / Pruefung 3: Zugriff ohne gueltiges Service-
// Credential wird abgelehnt.
func TestTenantKEKHandler_RejectsMissingCredential(t *testing.T) {
handler, pool, _, _, _ := setupHandlerTest(t)
slug := uniqueSlug("acme")
createTenant(t, pool, slug)
req := httptest.NewRequest(http.MethodGet, "/internal/keys/tenant-kek?tenant="+slug, nil)
rec := httptest.NewRecorder()
handler.TenantKEKHandler(rec, req)
if rec.Code != http.StatusForbidden {
t.Fatalf("status = %d, want 403 ohne credential", rec.Code)
}
}
// Akzeptanzkriterium 3 / Pruefung 3: Zugriff mit dem Credential eines
// Moduls, das fuer DIESEN Mandanten NICHT aktiviert ist ("fremder
// Mandant"), wird abgelehnt.
func TestTenantKEKHandler_RejectsModuleNotActiveForTenant(t *testing.T) {
handler, pool, _, clientID, secret := setupHandlerTest(t)
ctx := context.Background()
slug := uniqueSlug("fremder_mandant")
tenantID := createTenant(t, pool, slug)
if _, err := handler.store.CreateForTenant(ctx, tenantID, handler.masterKey); err != nil {
t.Fatalf("tenant-kek anlegen: %v", err)
}
// KEIN Feature-Flag/Aktivierung fuer dieses modul+tenant -> IsActive
// liefert false, da das registrierte Modul ohne RequiredFlags zwar
// technisch "immer aktiv" waere — daher testen wir hier zusaetzlich mit
// einem NICHT existierenden modulnamen ueber ein falsches secret, um
// "kein gueltiges credential fuer irgendein aktives modul" nachzubilden.
req := httptest.NewRequest(http.MethodGet, "/internal/keys/tenant-kek?tenant="+slug, nil)
req.Header.Set("X-Nexarch-Client-Id", clientID)
req.Header.Set("X-Nexarch-Client-Secret", "falsches-secret")
rec := httptest.NewRecorder()
handler.TenantKEKHandler(rec, req)
if rec.Code != http.StatusForbidden {
t.Fatalf("status = %d, want 403 mit ungueltigem secret", rec.Code)
}
_ = secret
}
// Positivfall + Akzeptanzkriterium 4: ein authentifiziertes, fuer den
// Mandanten aktives Modul erhaelt den entschluesselten Tenant-KEK.
func TestTenantKEKHandler_AllowsActiveModuleForTenant(t *testing.T) {
handler, pool, _, clientID, secret := setupHandlerTest(t)
ctx := context.Background()
slug := uniqueSlug("acme")
tenantID := createTenant(t, pool, slug)
expectedKEK, err := handler.store.CreateForTenant(ctx, tenantID, handler.masterKey)
if err != nil {
t.Fatalf("tenant-kek anlegen: %v", err)
}
req := httptest.NewRequest(http.MethodGet, "/internal/keys/tenant-kek?tenant="+slug, nil)
req.Header.Set("X-Nexarch-Client-Id", clientID)
req.Header.Set("X-Nexarch-Client-Secret", secret)
rec := httptest.NewRecorder()
handler.TenantKEKHandler(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200, body: %s", rec.Code, rec.Body.String())
}
if !strings.Contains(rec.Body.String(), "tenant_kek_base64") {
t.Fatalf("antwort enthaelt kein tenant_kek_base64-feld: %s", rec.Body.String())
}
_ = expectedKEK
_ = pool
}
-111
View File
@@ -1,111 +0,0 @@
// Package kek implementiert Core API-10: die zweistufige Schluesselhierarchie
// fuer Envelope-Encryption (Master-KEK -> Tenant-KEK), die DMS (FDN-09) und
// Mail (ARC-02) fuer ihre pro-Objekt-DEKs verwenden. Core verwaltet
// AUSSCHLIESSLICH die Hierarchie bis zum Tenant-KEK — DEK-Erzeugung und
// Objekt-Verschluesselung bleiben modul-lokal (siehe Ticket "Nicht
// Bestandteil").
//
// Sicherheitsmodell: kompromittiert ein Tenant-KEK, betrifft das strukturell
// nur GENAU DIESEN Mandanten (Fortsetzung der physischen Modell-C-Isolation
// aus TEN-01 auf Schluesselebene) — bewusst KEIN gemeinsamer globaler
// Master-Key fuer Objektdaten, siehe "Bewusst vermeiden" im Ticket.
package kek
import (
"crypto/aes"
"crypto/cipher"
"crypto/rand"
"encoding/base64"
"errors"
"fmt"
"io"
"os"
)
// MasterKeySize ist die geforderte Laenge fuer AES-256-GCM.
const MasterKeySize = 32
var (
ErrMasterKeyNotSet = errors.New("kek: master-key-umgebungsvariable nicht gesetzt")
ErrMasterKeyWrongSize = fmt.Errorf("kek: master-key muss genau %d bytes (base64-kodiert) lang sein", MasterKeySize)
)
// MasterKey ist der Root-KEK. Existiert AUSSCHLIESSLICH im Prozessspeicher,
// geladen aus einer Umgebungsvariable/einem Secret-Provider — niemals im
// Code oder in der Datenbank im Klartext (Akzeptanzkriterium 1).
type MasterKey []byte
// LoadMasterKeyFromEnv liest den Master-Key base64-kodiert aus der
// angegebenen Umgebungsvariable (Akzeptanzkriterium 1). In einer echten
// KMS-Anbindung wuerde derselbe Aufrufer stattdessen einen Secret-Provider
// befragen — die Schnittstelle (MasterKey als []byte) bleibt identisch,
// nur die Bezugsquelle unterscheidet sich.
func LoadMasterKeyFromEnv(envVar string) (MasterKey, error) {
raw := os.Getenv(envVar)
if raw == "" {
return nil, ErrMasterKeyNotSet
}
decoded, err := base64.StdEncoding.DecodeString(raw)
if err != nil {
return nil, fmt.Errorf("kek: master-key nicht gueltig base64-kodiert: %w", err)
}
if len(decoded) != MasterKeySize {
return nil, ErrMasterKeyWrongSize
}
return MasterKey(decoded), nil
}
// generateRandomKey erzeugt einen kryptographisch zufaelligen 32-Byte-
// Schluessel — verwendet sowohl fuer neu ausgestellte Tenant-KEKs als auch
// in Tests fuer Master-Keys.
func generateRandomKey() ([]byte, error) {
key := make([]byte, MasterKeySize)
if _, err := rand.Read(key); err != nil {
return nil, fmt.Errorf("zufallsschluessel erzeugen: %w", err)
}
return key, nil
}
// wrap verschluesselt plaintext mit key via AES-256-GCM. Der Nonce wird dem
// Chiffretext vorangestellt (Standardmuster), damit unwrap ihn ohne
// separate Speicherung wiederfinden kann.
func wrap(key, plaintext []byte) ([]byte, error) {
block, err := aes.NewCipher(key)
if err != nil {
return nil, fmt.Errorf("aes-cipher erstellen: %w", err)
}
gcm, err := cipher.NewGCM(block)
if err != nil {
return nil, fmt.Errorf("gcm erstellen: %w", err)
}
nonce := make([]byte, gcm.NonceSize())
if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
return nil, fmt.Errorf("nonce erzeugen: %w", err)
}
return gcm.Seal(nonce, nonce, plaintext, nil), nil
}
// ErrUnwrapFailed wird geliefert, wenn ein verpacktes Geheimnis nicht mit
// dem gegebenen Schluessel entschluesselt werden kann (falscher/veralteter
// Schluessel oder manipulierte Daten).
var ErrUnwrapFailed = errors.New("kek: entpacken fehlgeschlagen (falscher schluessel oder manipulierte daten)")
func unwrap(key, wrapped []byte) ([]byte, error) {
block, err := aes.NewCipher(key)
if err != nil {
return nil, fmt.Errorf("aes-cipher erstellen: %w", err)
}
gcm, err := cipher.NewGCM(block)
if err != nil {
return nil, fmt.Errorf("gcm erstellen: %w", err)
}
if len(wrapped) < gcm.NonceSize() {
return nil, ErrUnwrapFailed
}
nonce, ciphertext := wrapped[:gcm.NonceSize()], wrapped[gcm.NonceSize():]
plaintext, err := gcm.Open(nil, nonce, ciphertext, nil)
if err != nil {
return nil, ErrUnwrapFailed
}
return plaintext, nil
}
-144
View File
@@ -1,144 +0,0 @@
package kek
import (
"context"
"errors"
"fmt"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
var ErrNoTenantKEK = errors.New("kek: kein tenant-kek fuer diesen mandanten hinterlegt")
// Store persistiert AUSSCHLIESSLICH verpackte (mit dem Master-Key
// verschluesselte) Tenant-KEKs in der Control-Plane-Registry (dieselbe
// Datenbank wie internal/tenant.Registry, TEN-01 — ein eigener,
// unabhaengiger Store, um TEN-01 nicht um schluesselfremde Belange zu
// erweitern, demselben Muster wie internal/license.Store).
type Store struct {
pool *pgxpool.Pool
}
func NewStore(pool *pgxpool.Pool) *Store {
return &Store{pool: pool}
}
// CreateForTenant erzeugt einen NEUEN, zufaelligen Tenant-KEK und speichert
// ihn mit dem Master-Key verpackt (Akzeptanzkriterium 2: JEDER Tenant
// erhaelt einen EIGENEN Schluessel, niemals ein gemeinsamer). Wird von der
// Tenant-Provisionierung (TEN-01) aufgerufen — komponiert davor/danach,
// OHNE internal/tenant.Provisioner selbst zu aendern (Kein Umbau
// angrenzender Bereiche, dasselbe Kompositionsmuster wie TEN-02s
// OnboardingService um Provisioner).
func (s *Store) CreateForTenant(ctx context.Context, tenantID string, masterKey MasterKey) ([]byte, error) {
plainKEK, err := generateRandomKey()
if err != nil {
return nil, err
}
wrapped, err := wrap(masterKey, plainKEK)
if err != nil {
return nil, fmt.Errorf("tenant-kek verpacken: %w", err)
}
if _, err := s.pool.Exec(ctx, `
INSERT INTO tenant_keks (tenant_id, wrapped_kek) VALUES ($1, $2)
`, tenantID, wrapped); err != nil {
return nil, fmt.Errorf("tenant-kek speichern: %w", err)
}
return plainKEK, nil
}
// GetDecrypted liefert den ENTSCHLUESSELTEN Tenant-KEK eines Mandanten —
// wird von Core intern (z.B. fuer den HTTP-Handler in handler.go) sowie in
// Tests verwendet.
func (s *Store) GetDecrypted(ctx context.Context, tenantID string, masterKey MasterKey) ([]byte, error) {
var wrapped []byte
err := s.pool.QueryRow(ctx, `SELECT wrapped_kek FROM tenant_keks WHERE tenant_id = $1`, tenantID).Scan(&wrapped)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrNoTenantKEK
}
return nil, fmt.Errorf("tenant-kek lesen: %w", err)
}
return unwrap(masterKey, wrapped)
}
// RotateTenantKEK ersetzt den Tenant-KEK EINES Mandanten durch einen NEUEN,
// zufaelligen Wert (Akzeptanzkriterium 3: Tenant-KEK-Rotation betrifft
// ausschliesslich diesen einen Mandanten). Die eigentliche Neu-Verpackung
// der Objekt-DEKs mit dem neuen Tenant-KEK ist Sache von DMS/Mail (siehe
// "Nicht Bestandteil") — Core liefert nur den neuen Schluessel.
func (s *Store) RotateTenantKEK(ctx context.Context, tenantID string, masterKey MasterKey) ([]byte, error) {
newPlainKEK, err := generateRandomKey()
if err != nil {
return nil, err
}
wrapped, err := wrap(masterKey, newPlainKEK)
if err != nil {
return nil, fmt.Errorf("neuen tenant-kek verpacken: %w", err)
}
tag, err := s.pool.Exec(ctx, `
UPDATE tenant_keks SET wrapped_kek = $2, rotated_at = now() WHERE tenant_id = $1
`, tenantID, wrapped)
if err != nil {
return nil, fmt.Errorf("tenant-kek rotieren: %w", err)
}
if tag.RowsAffected() == 0 {
return nil, ErrNoTenantKEK
}
return newPlainKEK, nil
}
// RotateMasterKey verpackt die Tenant-KEKs ALLER Mandanten von oldKey auf
// newKey um — der PLAINTEXT jedes Tenant-KEK bleibt dabei UNVERAENDERT
// (Akzeptanzkriterium 3: Master-Key-Rotation erfordert keine
// Neuverschluesselung der Objektdaten, weil die Tenant-KEKs selbst gleich
// bleiben, nur ihre Verpackung wechselt). Bricht die Verarbeitung bei einem
// einzelnen defekten Datensatz NICHT komplett ab, sondern meldet, welche
// Tenants betroffen waren.
func (s *Store) RotateMasterKey(ctx context.Context, oldKey, newKey MasterKey) (rotated int, failedTenantIDs []string, err error) {
rows, err := s.pool.Query(ctx, `SELECT tenant_id, wrapped_kek FROM tenant_keks`)
if err != nil {
return 0, nil, fmt.Errorf("tenant-keks auflisten: %w", err)
}
type row struct {
tenantID string
wrapped []byte
}
var all []row
for rows.Next() {
var r row
if err := rows.Scan(&r.tenantID, &r.wrapped); err != nil {
rows.Close()
return 0, nil, fmt.Errorf("tenant-kek-zeile lesen: %w", err)
}
all = append(all, r)
}
rows.Close()
if err := rows.Err(); err != nil {
return 0, nil, err
}
for _, r := range all {
plainKEK, err := unwrap(oldKey, r.wrapped)
if err != nil {
failedTenantIDs = append(failedTenantIDs, r.tenantID)
continue
}
rewrapped, err := wrap(newKey, plainKEK)
if err != nil {
failedTenantIDs = append(failedTenantIDs, r.tenantID)
continue
}
if _, err := s.pool.Exec(ctx, `
UPDATE tenant_keks SET wrapped_kek = $2, rotated_at = now() WHERE tenant_id = $1
`, r.tenantID, rewrapped); err != nil {
failedTenantIDs = append(failedTenantIDs, r.tenantID)
continue
}
rotated++
}
return rotated, failedTenantIDs, nil
}
+97
View File
@@ -0,0 +1,97 @@
package moduletrust
import (
"context"
"fmt"
"sync"
"time"
)
// StaleCache ist der generische Rechte-/Feature-Flag-Cache-Kontrakt
// (Akzeptanzkriterium 2): TTL-basiert, mit explizitem, benanntem Verhalten
// bei abgelaufenem Cache waehrend Core nicht erreichbar ist.
//
// - Get: FAIL-OPEN fuer Lesevorgaenge. Schlaegt der Refresh fehl, aber es
// gibt bereits einen (wenn auch abgelaufenen) Stand, wird dieser mit
// stale=true zurueckgegeben — Begruendung: ein bereits authentifiziertes
// Modul soll mit dem letztbekannten Stand weiterarbeiten koennen statt
// hart zu blockieren (siehe "Bekannte Fehler vermeiden" im Ticket).
// Existiert noch nie ein Stand, gibt es keinen sinnvollen Fallback —
// dann liefert auch Get einen Fehler.
// - RequireFresh: FAIL-CLOSED fuer sicherheitskritische Aktionen (z.B.
// ein komplett NEUER Login). Nutzt NIEMALS einen zwischengespeicherten
// Stand, ruft immer frisch ab — Begruendung: eine neue Vertrauens-
// entscheidung darf nicht auf veralteten Daten beruhen, auch wenn das
// bedeutet, dass die Aktion bei Core-Ausfall sichtbar fehlschlaegt statt
// unsicher "irgendwie" durchgelassen zu werden.
//
// LIC-02 (internal/flag.Service) implementiert bereits denselben Kontrakt
// fuer Feature-Flags — StaleCache verallgemeinert dasselbe Muster fuer
// JWT-Signaturschluessel, damit beide Faelle derselben dokumentierten
// Policy folgen.
type StaleCache[T any] struct {
mu sync.RWMutex
value T
hasValue bool
fetchedAt time.Time
ttl time.Duration
fetch func(ctx context.Context) (T, error)
}
func NewStaleCache[T any](ttl time.Duration, fetch func(ctx context.Context) (T, error)) *StaleCache[T] {
return &StaleCache[T]{ttl: ttl, fetch: fetch}
}
// Get liefert den Cache-Wert. FAIL-OPEN: bei Refresh-Fehler wird ein
// vorhandener, ggf. abgelaufener Stand zurueckgegeben (stale=true).
func (c *StaleCache[T]) Get(ctx context.Context) (value T, stale bool, err error) {
c.mu.RLock()
fresh := c.hasValue && time.Since(c.fetchedAt) < c.ttl
if fresh {
v := c.value
c.mu.RUnlock()
return v, false, nil
}
c.mu.RUnlock()
newVal, fetchErr := c.fetch(ctx)
if fetchErr == nil {
c.mu.Lock()
c.value, c.hasValue, c.fetchedAt = newVal, true, time.Now()
c.mu.Unlock()
return newVal, false, nil
}
c.mu.RLock()
defer c.mu.RUnlock()
if c.hasValue {
return c.value, true, nil
}
var zero T
return zero, false, fmt.Errorf("cache leer und refresh fehlgeschlagen: %w", fetchErr)
}
// Invalidate erzwingt beim naechsten Get-Aufruf einen sofortigen Refresh
// statt auf den TTL-Ablauf zu warten (API-06, Akzeptanzkriterium 1: ein
// Health-Check-getriggerter Wiederanlauf soll den Cache SOFORT aktualisieren,
// nicht die reguläre TTL abwarten) — rein additiv, aendert nichts an
// Get/RequireFresh (dasselbe Muster wie internal/flag.Service.Invalidate).
func (c *StaleCache[T]) Invalidate() {
c.mu.Lock()
c.hasValue = false
c.mu.Unlock()
}
// RequireFresh ruft IMMER frisch ab (FAIL-CLOSED) — fuer sicherheitskritische
// Aktionen, die niemals auf einem zwischengespeicherten Stand basieren duerfen.
func (c *StaleCache[T]) RequireFresh(ctx context.Context) (T, error) {
v, err := c.fetch(ctx)
if err != nil {
var zero T
return zero, fmt.Errorf("core nicht erreichbar, sicherheitskritische aktion abgelehnt: %w", err)
}
c.mu.Lock()
c.value, c.hasValue, c.fetchedAt = v, true, time.Now()
c.mu.Unlock()
return v, nil
}
+79
View File
@@ -0,0 +1,79 @@
package moduletrust
import (
"encoding/base64"
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/golang-jwt/jwt/v5"
)
type Claims struct {
Subject string `json:"sub"`
TenantSlug string `json:"tenant"`
jwt.RegisteredClaims
}
// Issue signiert ein Token mit dem aktuellen Signierschluessel und traegt
// dessen KID im JWT-Header ein — der Verifier auf Modulseite waehlt darueber
// den passenden oeffentlichen Schluessel aus PublicKeySet() aus.
func (m *KeyManager) Issue(subject, tenantSlug string, ttl time.Duration) (string, error) {
key, err := m.SigningKey()
if err != nil {
return "", err
}
now := time.Now()
claims := Claims{
Subject: subject,
TenantSlug: tenantSlug,
RegisteredClaims: jwt.RegisteredClaims{
IssuedAt: jwt.NewNumericDate(now),
ExpiresAt: jwt.NewNumericDate(now.Add(ttl)),
},
}
token := jwt.NewWithClaims(jwt.SigningMethodEdDSA, claims)
token.Header["kid"] = key.KID
return token.SignedString(key.Private)
}
type jwksResponse struct {
Keys []jwksKey `json:"keys"`
}
type jwksKey struct {
Kid string `json:"kid"`
PublicKey string `json:"public_key"` // base64 (raw Ed25519, 32 Byte)
}
// ServeJWKS liefert alle bekannten oeffentlichen Schluessel als JSON —
// Module fragen dies periodisch ab (nicht pro Request), siehe Verifier.
func (m *KeyManager) ServeJWKS(w http.ResponseWriter, r *http.Request) {
set := m.PublicKeySet()
resp := jwksResponse{Keys: make([]jwksKey, 0, len(set))}
for kid, pub := range set {
resp.Keys = append(resp.Keys, jwksKey{Kid: kid, PublicKey: base64.StdEncoding.EncodeToString(pub)})
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(resp)
}
// ParseJWKS dekodiert die JSON-Antwort von ServeJWKS zurueck in kid->PublicKey
// — Hilfsfunktion fuer Module, die JWKS per HTTP abrufen.
func ParseJWKS(data []byte) (map[string][]byte, error) {
var resp jwksResponse
if err := json.Unmarshal(data, &resp); err != nil {
return nil, fmt.Errorf("jwks parsen: %w", err)
}
out := make(map[string][]byte, len(resp.Keys))
for _, k := range resp.Keys {
raw, err := base64.StdEncoding.DecodeString(k.PublicKey)
if err != nil {
return nil, fmt.Errorf("oeffentlichen schluessel %q dekodieren: %w", k.Kid, err)
}
out[k.Kid] = raw
}
return out, nil
}
+83
View File
@@ -0,0 +1,83 @@
// Package moduletrust implementiert Core API-05: asymmetrische JWT-Signatur
// mit Schluesselverteilung (JWKS), damit DMS/Mail/Archive/Workflow JWTs
// LOKAL verifizieren koennen, ohne pro Aufruf einen synchronen Request an
// Core zu stellen — Core darf Fundament sein, ohne zum Flaschenhals zu
// werden (siehe Entscheidungsverlauf "Vertrauensstellung Core<->Module" in
// nexarch-state.json). IAM-02s HS256-Session-Cookie (Browser-Login) bleibt
// unangetastet — dies ist ein zusaetzlicher, getrennter Vertrauensmechanismus
// fuer Modul-zu-Modul/Modul-zu-Core-Aufrufe.
package moduletrust
import (
"crypto/ed25519"
"crypto/rand"
"encoding/hex"
"fmt"
"sync"
)
type KeyPair struct {
KID string
Private ed25519.PrivateKey
Public ed25519.PublicKey
}
// KeyManager haelt ALLE noch gueltigen Schluesselpaare — nicht nur das
// aktuell signierende. Rotate erzeugt ein neues Paar und behaelt die alten
// fuer die Verifikation bereits ausgestellter Tokens (Akzeptanzkriterium 3:
// Rotation ohne Ausfallzeit fuer andere Module).
type KeyManager struct {
mu sync.RWMutex
keys []KeyPair // aeltestes zuerst, neuestes zuletzt
}
func NewKeyManager() (*KeyManager, error) {
m := &KeyManager{}
if _, err := m.Rotate(); err != nil {
return nil, err
}
return m, nil
}
// Rotate erzeugt ein neues Ed25519-Schluesselpaar mit eigener KID und macht
// es zum aktuellen Signierschluessel. Aeltere Schluessel bleiben in
// PublicKeySet() erhalten, damit bereits ausgestellte Tokens weiterhin
// verifizierbar sind.
func (m *KeyManager) Rotate() (KeyPair, error) {
pub, priv, err := ed25519.GenerateKey(nil)
if err != nil {
return KeyPair{}, fmt.Errorf("schluesselpaar erzeugen: %w", err)
}
kidBytes := make([]byte, 8)
if _, err := rand.Read(kidBytes); err != nil {
return KeyPair{}, fmt.Errorf("kid erzeugen: %w", err)
}
kp := KeyPair{KID: hex.EncodeToString(kidBytes), Private: priv, Public: pub}
m.mu.Lock()
m.keys = append(m.keys, kp)
m.mu.Unlock()
return kp, nil
}
// SigningKey liefert den aktuellen (neuesten) Schluessel zum Signieren neuer Tokens.
func (m *KeyManager) SigningKey() (KeyPair, error) {
m.mu.RLock()
defer m.mu.RUnlock()
if len(m.keys) == 0 {
return KeyPair{}, fmt.Errorf("moduletrust: kein schluessel vorhanden")
}
return m.keys[len(m.keys)-1], nil
}
// PublicKeySet liefert ALLE bekannten oeffentlichen Schluessel (kid ->
// public key) — die Grundlage fuer den JWKS-Endpunkt.
func (m *KeyManager) PublicKeySet() map[string]ed25519.PublicKey {
m.mu.RLock()
defer m.mu.RUnlock()
out := make(map[string]ed25519.PublicKey, len(m.keys))
for _, k := range m.keys {
out[k.KID] = k.Public
}
return out
}
+214
View File
@@ -0,0 +1,214 @@
package moduletrust
import (
"context"
"crypto/ed25519"
"errors"
"net/http"
"sync"
"testing"
"time"
)
func TestIssueAndVerify_RoundTrip(t *testing.T) {
km, err := NewKeyManager()
if err != nil {
t.Fatalf("new key manager: %v", err)
}
token, err := km.Issue("user-1", "acme", time.Hour)
if err != nil {
t.Fatalf("issue: %v", err)
}
v := NewVerifier(time.Hour, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
return km.PublicKeySet(), nil
})
claims, err := v.Verify(context.Background(), token)
if err != nil {
t.Fatalf("verify: %v", err)
}
if claims.Subject != "user-1" || claims.TenantSlug != "acme" {
t.Fatalf("claims unerwartet: %+v", claims)
}
}
// Akzeptanzkriterium 1: Verifikation lokal, kein Request pro Aufruf.
func TestVerify_DoesNotFetchPerCall(t *testing.T) {
km, err := NewKeyManager()
if err != nil {
t.Fatalf("new key manager: %v", err)
}
token, err := km.Issue("user-1", "acme", time.Hour)
if err != nil {
t.Fatalf("issue: %v", err)
}
var mu sync.Mutex
fetchCalls := 0
v := NewVerifier(time.Hour, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
mu.Lock()
fetchCalls++
mu.Unlock()
return km.PublicKeySet(), nil
})
for i := 0; i < 10; i++ {
if _, err := v.Verify(context.Background(), token); err != nil {
t.Fatalf("verify %d: %v", i, err)
}
}
mu.Lock()
defer mu.Unlock()
if fetchCalls != 1 {
t.Fatalf("erwartet genau 1 fetch fuer 10 Verify-Aufrufe innerhalb der TTL, habe %d", fetchCalls)
}
}
// Akzeptanzkriterium 2 + Pruefung 1: Core simuliert nicht erreichbar,
// bereits authentifizierte Nutzer bleiben funktionsfaehig (Fail-Open mit
// letztbekanntem Schluesselstand).
func TestVerify_FailsOpenWhenCoreUnreachableButStaleKeysExist(t *testing.T) {
km, err := NewKeyManager()
if err != nil {
t.Fatalf("new key manager: %v", err)
}
token, err := km.Issue("user-1", "acme", time.Hour)
if err != nil {
t.Fatalf("issue: %v", err)
}
coreDown := false
v := NewVerifier(30*time.Millisecond, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
if coreDown {
return nil, errors.New("core nicht erreichbar (simuliert)")
}
return km.PublicKeySet(), nil
})
// Cache vorwaermen, waehrend Core noch erreichbar ist.
if _, err := v.Verify(context.Background(), token); err != nil {
t.Fatalf("verify (warm): %v", err)
}
// "Core abschalten" und TTL ablaufen lassen.
coreDown = true
time.Sleep(50 * time.Millisecond)
if _, err := v.Verify(context.Background(), token); err != nil {
t.Fatalf("verify sollte trotz core-ausfall mit letztbekanntem stand funktionieren: %v", err)
}
}
// Akzeptanzkriterium 2 + Pruefung 2: neue sicherheitskritische Aktionen
// (z.B. neuer Login) schlagen bei Core-Ausfall klar fehl statt unsicher
// durchgelassen zu werden — auch wenn ein (aelterer) Cache-Stand existiert.
func TestRequireFreshKeys_FailsClosedWhenCoreUnreachable(t *testing.T) {
km, err := NewKeyManager()
if err != nil {
t.Fatalf("new key manager: %v", err)
}
coreDown := false
v := NewVerifier(time.Hour, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
if coreDown {
return nil, errors.New("core nicht erreichbar (simuliert)")
}
return km.PublicKeySet(), nil
})
// Cache vorwaermen (existiert jetzt ein "veralteter" gueltiger Stand).
if _, _, err := v.cache.Get(context.Background()); err != nil {
t.Fatalf("warm cache: %v", err)
}
coreDown = true
if err := v.RequireFreshKeys(context.Background()); err == nil {
t.Fatal("erwartet fehler (fail-closed) bei core-ausfall, habe nil")
}
}
// Akzeptanzkriterium 3 + Pruefung 3: Schluesselrotation ohne Ausfallzeit —
// ein bereits ausgestelltes Token bleibt nach Rotation weiterhin
// verifizierbar, ein zweites (simuliertes) Modul bekommt beide Schluessel.
func TestRotate_NoDowntimeForAlreadyIssuedTokens(t *testing.T) {
km, err := NewKeyManager()
if err != nil {
t.Fatalf("new key manager: %v", err)
}
oldToken, err := km.Issue("user-1", "acme", time.Hour)
if err != nil {
t.Fatalf("issue (alt): %v", err)
}
if _, err := km.Rotate(); err != nil {
t.Fatalf("rotate: %v", err)
}
newToken, err := km.Issue("user-2", "acme", time.Hour)
if err != nil {
t.Fatalf("issue (neu): %v", err)
}
// Simuliertes zweites Modul: fragt den vollstaendigen Schluesselsatz ab.
moduleB := NewVerifier(time.Hour, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
return km.PublicKeySet(), nil
})
if _, err := moduleB.Verify(context.Background(), oldToken); err != nil {
t.Fatalf("altes token sollte nach rotation weiterhin gueltig sein: %v", err)
}
if _, err := moduleB.Verify(context.Background(), newToken); err != nil {
t.Fatalf("neues token sollte gueltig sein: %v", err)
}
}
func TestVerify_RejectsUnknownKid(t *testing.T) {
km1, _ := NewKeyManager()
km2, _ := NewKeyManager() // komplett anderer, unbekannter schluessel
token, err := km1.Issue("user-1", "acme", time.Hour)
if err != nil {
t.Fatalf("issue: %v", err)
}
v := NewVerifier(time.Hour, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
return km2.PublicKeySet(), nil // kennt km1s schluessel nicht
})
if _, err := v.Verify(context.Background(), token); !errors.Is(err, ErrInvalidToken) {
t.Fatalf("erwartet ErrInvalidToken, habe %v", err)
}
}
func TestJWKSRoundTrip(t *testing.T) {
km, _ := NewKeyManager()
km.Rotate()
var buf []byte
rec := &captureWriter{}
km.ServeJWKS(rec, nil)
buf = rec.body
parsed, err := ParseJWKS(buf)
if err != nil {
t.Fatalf("parse jwks: %v", err)
}
if len(parsed) != len(km.PublicKeySet()) {
t.Fatalf("erwartet %d schluessel, habe %d", len(km.PublicKeySet()), len(parsed))
}
}
type captureWriter struct {
body []byte
header http.Header
}
func (w *captureWriter) Header() http.Header {
if w.header == nil {
w.header = http.Header{}
}
return w.header
}
func (w *captureWriter) Write(p []byte) (int, error) { w.body = append(w.body, p...); return len(p), nil }
func (w *captureWriter) WriteHeader(statusCode int) {}
+67
View File
@@ -0,0 +1,67 @@
package moduletrust
import (
"context"
"crypto/ed25519"
"errors"
"time"
"github.com/golang-jwt/jwt/v5"
)
var ErrInvalidToken = errors.New("moduletrust: ungueltiges token")
// KeyFetchFunc holt den aktuellen Schluesselsatz von Core (z.B. per HTTP-GET
// auf ServeJWKS + ParseJWKS). Wird vom Verifier nur bei abgelaufener TTL
// aufgerufen — NICHT bei jeder Verify()-Anfrage (Akzeptanzkriterium 1).
type KeyFetchFunc func(ctx context.Context) (map[string]ed25519.PublicKey, error)
// Verifier ist die Modulseite von API-05: verifiziert JWTs LOKAL gegen einen
// per StaleCache zwischengespeicherten Schluesselsatz, ohne pro Aufruf einen
// synchronen Request an Core zu stellen.
type Verifier struct {
cache *StaleCache[map[string]ed25519.PublicKey]
}
func NewVerifier(ttl time.Duration, fetch KeyFetchFunc) *Verifier {
return &Verifier{cache: NewStaleCache(ttl, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
return fetch(ctx)
})}
}
// Verify prueft die Signatur LOKAL gegen den (ggf. abgelaufenen, aber
// vorhandenen) Schluesselsatz — FAIL-OPEN fuer bereits ausgestellte Tokens
// (Akzeptanzkriterium 2): ist Core nicht erreichbar, aber ein alter
// Schluesselsatz bekannt, wird damit weiter verifiziert.
func (v *Verifier) Verify(ctx context.Context, tokenString string) (*Claims, error) {
keys, _, err := v.cache.Get(ctx)
if err != nil {
return nil, err
}
claims := &Claims{}
token, err := jwt.ParseWithClaims(tokenString, claims, func(t *jwt.Token) (interface{}, error) {
if _, ok := t.Method.(*jwt.SigningMethodEd25519); !ok {
return nil, ErrInvalidToken
}
kid, _ := t.Header["kid"].(string)
pub, ok := keys[kid]
if !ok {
return nil, ErrInvalidToken
}
return pub, nil
})
if err != nil || !token.Valid {
return nil, ErrInvalidToken
}
return claims, nil
}
// RequireFreshKeys ruft IMMER frisch von Core ab (FAIL-CLOSED) — fuer
// sicherheitskritische Aktionen wie einen komplett neuen Login
// (Akzeptanzkriterium 2): schlaegt klar fehl, wenn Core nicht erreichbar
// ist, statt auf einem veralteten Schluesselsatz zu vertrauen.
func (v *Verifier) RequireFreshKeys(ctx context.Context) error {
_, err := v.cache.RequireFresh(ctx)
return err
}
+155
View File
@@ -0,0 +1,155 @@
// Package resync implementiert Core API-06: Wiederanlauf & Nachsynchro-
// nisierung nach einem Core-Ausfall.
//
// - Ein Modul puffert Audit-Events und Nutzungszaehler-Deltas LOKAL in
// Postgres (NICHT im Speicher — siehe "Bekannte Fehler vermeiden" im
// Ticket: ein erneuter Ausfall waehrend der Nachlieferung darf keine
// Daten verlieren, eine In-Memory-Queue wuerde das riskieren).
// - Ein Health-Check-getriggerter Worker erkennt die Core-Wiedererreich-
// barkeit SOFORT (nicht erst nach TTL-Ablauf, siehe StaleCache.Invalidate)
// und liefert die gepufferten Daten in ORIGINALER Reihenfolge, authenti-
// fiziert ueber das Service-Credential aus API-02
// (internal/moduleregistry.Registry.Authenticate).
//
// Dieses Paket dupliziert weder internal/audit (AUD-01) noch internal/usage
// (LIC-03) — es liefert nur den Puffer- und Nachlieferungs-Mechanismus
// DAVOR bzw. DANACH.
package resync
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
// BufferedAuditEvent ist ein lokal gepuffertes Audit-Ereignis. Seq
// garantiert die Wiederherstellung der urspruenglichen Reihenfolge
// (Akzeptanzkriterium 2) unabhaengig von eventuellen Uhrzeit-Ungenauigkeiten.
type BufferedAuditEvent struct {
ID int64
Seq int64
TenantSlug string
Actor string
Action string
Target string
Metadata map[string]any
CreatedAt time.Time
}
// BufferedUsageDelta ist ein lokal gepuffertes Nutzungszaehler-Inkrement.
type BufferedUsageDelta struct {
ID int64
Seq int64
TenantSlug string
Metric string
Delta int64
CreatedAt time.Time
}
// Buffer ist die lokale, persistente Pufferqueue eines Moduls.
type Buffer struct {
pool *pgxpool.Pool
}
func NewBuffer(pool *pgxpool.Pool) *Buffer {
return &Buffer{pool: pool}
}
// EnqueueAuditEvent puffert EIN Audit-Ereignis lokal — wird von einem
// Fachmodul aufgerufen, wenn Core gerade nicht erreichbar ist (die
// Erkennung "Core erreichbar oder nicht" ist NICHT Teil dieses Aufrufs,
// siehe Worker).
func (b *Buffer) EnqueueAuditEvent(ctx context.Context, tenantSlug, actor, action, target string, metadata map[string]any) error {
if metadata == nil {
metadata = map[string]any{}
}
metadataJSON, err := json.Marshal(metadata)
if err != nil {
return fmt.Errorf("metadata serialisieren: %w", err)
}
_, err = b.pool.Exec(ctx, `
INSERT INTO resync_audit_buffer (tenant_slug, actor, action, target, metadata)
VALUES ($1, $2, $3, $4, $5)
`, tenantSlug, actor, action, target, metadataJSON)
if err != nil {
return fmt.Errorf("audit-ereignis puffern: %w", err)
}
return nil
}
// EnqueueUsageDelta puffert EIN Nutzungszaehler-Inkrement lokal.
func (b *Buffer) EnqueueUsageDelta(ctx context.Context, tenantSlug, metric string, delta int64) error {
_, err := b.pool.Exec(ctx, `
INSERT INTO resync_usage_buffer (tenant_slug, metric, delta)
VALUES ($1, $2, $3)
`, tenantSlug, metric, delta)
if err != nil {
return fmt.Errorf("nutzungsdelta puffern: %w", err)
}
return nil
}
// PendingAuditEvents liefert ALLE noch nicht zugestellten Audit-Events in
// ORIGINALER Reihenfolge (Akzeptanzkriterium 2 / Pruefung 1).
func (b *Buffer) PendingAuditEvents(ctx context.Context) ([]BufferedAuditEvent, error) {
rows, err := b.pool.Query(ctx, `
SELECT id, seq, tenant_slug, actor, action, target, metadata, created_at
FROM resync_audit_buffer ORDER BY seq
`)
if err != nil {
return nil, fmt.Errorf("gepufferte audit-events abfragen: %w", err)
}
defer rows.Close()
var out []BufferedAuditEvent
for rows.Next() {
var e BufferedAuditEvent
var metadataJSON []byte
if err := rows.Scan(&e.ID, &e.Seq, &e.TenantSlug, &e.Actor, &e.Action, &e.Target, &metadataJSON, &e.CreatedAt); err != nil {
return nil, fmt.Errorf("gepuffertes audit-event lesen: %w", err)
}
_ = json.Unmarshal(metadataJSON, &e.Metadata)
out = append(out, e)
}
return out, rows.Err()
}
// PendingUsageDeltas liefert ALLE noch nicht zugestellten Nutzungsdeltas.
func (b *Buffer) PendingUsageDeltas(ctx context.Context) ([]BufferedUsageDelta, error) {
rows, err := b.pool.Query(ctx, `
SELECT id, seq, tenant_slug, metric, delta, created_at
FROM resync_usage_buffer ORDER BY seq
`)
if err != nil {
return nil, fmt.Errorf("gepufferte nutzungsdeltas abfragen: %w", err)
}
defer rows.Close()
var out []BufferedUsageDelta
for rows.Next() {
var d BufferedUsageDelta
if err := rows.Scan(&d.ID, &d.Seq, &d.TenantSlug, &d.Metric, &d.Delta, &d.CreatedAt); err != nil {
return nil, fmt.Errorf("gepuffertes nutzungsdelta lesen: %w", err)
}
out = append(out, d)
}
return out, rows.Err()
}
// RemoveAuditEvent entfernt EIN Audit-Event aus dem Puffer — wird NUR nach
// von Core BESTAETIGTER Zustellung aufgerufen (Akzeptanzkriterium 2/3: erst
// entfernen, wenn sicher zugestellt, sonst bleibt es fuer den naechsten
// Versuch erhalten — kein Datenverlust bei erneutem Ausfall waehrend der
// Nachlieferung).
func (b *Buffer) RemoveAuditEvent(ctx context.Context, id int64) error {
_, err := b.pool.Exec(ctx, `DELETE FROM resync_audit_buffer WHERE id = $1`, id)
return err
}
func (b *Buffer) RemoveUsageDelta(ctx context.Context, id int64) error {
_, err := b.pool.Exec(ctx, `DELETE FROM resync_usage_buffer WHERE id = $1`, id)
return err
}
+132
View File
@@ -0,0 +1,132 @@
package resync
import (
"context"
"encoding/json"
"net/http"
"time"
)
// AuditRecorder ist die schmale Schnittstelle, ueber die Core empfangene
// Audit-Events tatsaechlich persistiert. In Produktion durch internal/audit
// (AUD-01, nicht Abhaengigkeit dieser Kachel) implementiert — dieses Paket
// dupliziert dessen Validierungs-/Speicherlogik NICHT, sondern ruft sie nur
// auf. Die Events werden vom Aufrufer sequenziell in PendingAuditEvents-
// Reihenfolge uebergeben (Akzeptanzkriterium 2).
type AuditRecorder interface {
Record(ctx context.Context, tenantSlug, actor, action, target string, metadata map[string]any, occurredAt time.Time) error
}
// UsageIncrementer ist die schmale Schnittstelle zu Core's Nutzungszaehler
// (in Produktion internal/usage, LIC-03 — nicht Abhaengigkeit dieser
// Kachel). Jedes gepufferte Delta wird GENAU EINMAL angewendet.
type UsageIncrementer interface {
Increment(ctx context.Context, tenantSlug, metric string, delta int64) error
}
// CredentialAuthenticator ist die schmale Schnittstelle zu API-02s
// Service-Credential-Pruefung (internal/moduleregistry.Registry.Authenticate).
type CredentialAuthenticator interface {
Authenticate(ctx context.Context, clientID, secret string) (moduleName string, ok bool, err error)
}
// Handler nimmt nachgelieferte Audit-Events/Nutzungsdeltas auf der
// Core-Seite entgegen — authentifiziert ueber dasselbe Service-Credential
// wie jeder andere Modul-Core-Aufruf (Akzeptanzkriterium 2/3, "authentifiziert
// ueber das in API-02 definierte Service-Credential").
type Handler struct {
auth CredentialAuthenticator
audit AuditRecorder
usage UsageIncrementer
}
func NewHandler(auth CredentialAuthenticator, audit AuditRecorder, usage UsageIncrementer) *Handler {
return &Handler{auth: auth, audit: audit, usage: usage}
}
type credentialHeader struct {
ClientID string `json:"client_id"`
Secret string `json:"secret"`
}
func (h *Handler) authenticate(w http.ResponseWriter, r *http.Request) bool {
clientID := r.Header.Get("X-Nexarch-Client-Id")
secret := r.Header.Get("X-Nexarch-Client-Secret")
_, ok, err := h.auth.Authenticate(r.Context(), clientID, secret)
if err != nil || !ok {
http.Error(w, "ungueltiges service-credential", http.StatusUnauthorized)
return false
}
return true
}
type auditEventDTO struct {
TenantSlug string `json:"tenant_slug"`
Actor string `json:"actor"`
Action string `json:"action"`
Target string `json:"target"`
Metadata map[string]any `json:"metadata"`
OccurredAt time.Time `json:"occurred_at"`
}
// AuditHandler nimmt EINE Liste gepufferter Audit-Events entgegen und
// schreibt sie SEQUENZIELL in der gegebenen Reihenfolge fort
// (Akzeptanzkriterium 2 / Pruefung 1: Vollstaendigkeit + Reihenfolge).
// Bricht die Verarbeitung bei einem Fehler ab und meldet, wie viele Events
// bereits sicher geschrieben wurden — der Aufrufer (Worker) entfernt aus
// seinem lokalen Puffer NUR die bestaetigt geschriebenen Events.
func (h *Handler) AuditHandler(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(w, r) {
return
}
var events []auditEventDTO
if err := json.NewDecoder(r.Body).Decode(&events); err != nil {
http.Error(w, "ungueltiger anfrage-koerper", http.StatusBadRequest)
return
}
written := 0
for _, e := range events {
if err := h.audit.Record(r.Context(), e.TenantSlug, e.Actor, e.Action, e.Target, e.Metadata, e.OccurredAt); err != nil {
break
}
written++
}
writeJSON(w, map[string]int{"written": written})
}
type usageDeltaDTO struct {
TenantSlug string `json:"tenant_slug"`
Metric string `json:"metric"`
Delta int64 `json:"delta"`
}
// UsageHandler wendet JEDES gepufferte Delta GENAU EINMAL an
// (Akzeptanzkriterium 3 / Pruefung 2: keine Doppelzaehlung) — der Aufrufer
// entfernt aus seinem lokalen Puffer nur die bestaetigt uebernommenen Deltas.
func (h *Handler) UsageHandler(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(w, r) {
return
}
var deltas []usageDeltaDTO
if err := json.NewDecoder(r.Body).Decode(&deltas); err != nil {
http.Error(w, "ungueltiger anfrage-koerper", http.StatusBadRequest)
return
}
applied := 0
for _, d := range deltas {
if err := h.usage.Increment(r.Context(), d.TenantSlug, d.Metric, d.Delta); err != nil {
break
}
applied++
}
writeJSON(w, map[string]int{"applied": applied})
}
func writeJSON(w http.ResponseWriter, body any) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(body)
}
+320
View File
@@ -0,0 +1,320 @@
package resync
import (
"context"
"fmt"
"net/http"
"net/http/httptest"
"os"
"sync"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
"gitea.perlbach24.de/scripte/nexarch/internal/moduleregistry"
)
// fakeAudit steht fuer internal/audit.Log (AUD-01, nicht Abhaengigkeit
// dieser Kachel) — zeichnet Aufrufe in Empfangsreihenfolge auf, damit
// Vollstaendigkeit UND Reihenfolge geprueft werden koennen.
type fakeAudit struct {
mu sync.Mutex
events []auditEventDTO
failAt int // -1 = nie fehlschlagen; sonst: ab diesem Index (0-basiert) schlaegt Record fehl
}
func (f *fakeAudit) Record(ctx context.Context, tenantSlug, actor, action, target string, metadata map[string]any, occurredAt time.Time) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.failAt >= 0 && len(f.events) == f.failAt {
return fmt.Errorf("simulierter core-ausfall waehrend der nachlieferung")
}
f.events = append(f.events, auditEventDTO{TenantSlug: tenantSlug, Actor: actor, Action: action, Target: target, Metadata: metadata, OccurredAt: occurredAt})
return nil
}
// fakeUsage steht fuer internal/usage.Store (LIC-03, nicht Abhaengigkeit
// dieser Kachel) — summiert Deltas wie der echte Store.
type fakeUsage struct {
mu sync.Mutex
totals map[string]int64
}
func newFakeUsage() *fakeUsage { return &fakeUsage{totals: map[string]int64{}} }
func (f *fakeUsage) Increment(ctx context.Context, tenantSlug, metric string, delta int64) error {
f.mu.Lock()
defer f.mu.Unlock()
f.totals[tenantSlug+"|"+metric] += delta
return nil
}
func setupTest(t *testing.T) (*Buffer, *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 resync_audit_buffer (
id BIGSERIAL PRIMARY KEY, seq BIGSERIAL, tenant_slug TEXT NOT NULL, actor TEXT NOT NULL,
action TEXT NOT NULL, target TEXT NOT NULL DEFAULT '', metadata JSONB NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS resync_usage_buffer (
id BIGSERIAL PRIMARY KEY, seq BIGSERIAL, tenant_slug TEXT NOT NULL, metric TEXT NOT NULL,
delta BIGINT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
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)
}
cleanup := func() { pool.Close() }
return NewBuffer(pool), pool, cleanup
}
func uniqueModuleName() string {
return fmt.Sprintf("resync-test-%d", time.Now().UnixNano())
}
// setupModuleCredential registriert ein echtes Modul + Service-Credential
// ueber internal/moduleregistry (API-02) — dieselbe Authentifizierung wird
// vom Handler tatsaechlich geprueft, kein Mock.
func setupModuleCredential(t *testing.T, pool *pgxpool.Pool) (registry *moduleregistry.Registry, clientID, secret string) {
t.Helper()
ctx := context.Background()
flagService := flag.NewService(flag.NewStore(pool), 10*time.Millisecond)
registry = moduleregistry.NewRegistry(pool, flagService)
name := uniqueModuleName()
if _, err := registry.Register(ctx, name, "1.0.0", nil); err != nil {
t.Fatalf("modul registrieren: %v", err)
}
clientID, secret, err := registry.Provision(ctx, name)
if err != nil {
t.Fatalf("credential provisionieren: %v", err)
}
return registry, clientID, secret
}
// Akzeptanzkriterium 2 + Pruefung 1: waehrend eines simulierten Ausfalls
// lokal gepufferte Audit-Events sind nach Wiederanlauf vollstaendig und in
// korrekter Reihenfolge in Core's Audit-Log vorhanden.
func TestFlushAll_DeliversBufferedAuditEventsCompleteAndInOrder(t *testing.T) {
buffer, pool, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
registry, clientID, secret := setupModuleCredential(t, pool)
tenantSlug := "acme"
// Ereignisse "waehrend core down" lokal puffern.
actions := []string{"login", "upload", "delete", "logout"}
for _, action := range actions {
if err := buffer.EnqueueAuditEvent(ctx, tenantSlug, "user-1", action, "res-1", nil); err != nil {
t.Fatalf("enqueue %s: %v", action, err)
}
}
audit := &fakeAudit{failAt: -1}
usage := newFakeUsage()
handler := NewHandler(registry, audit, usage)
mux := http.NewServeMux()
mux.HandleFunc("/internal/resync/audit", handler.AuditHandler)
mux.HandleFunc("/internal/resync/usage", handler.UsageHandler)
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
server := httptest.NewServer(mux)
defer server.Close()
client := NewCoreClient(server.URL, clientID, secret)
worker := NewWorker(buffer, client)
if err := worker.FlushAll(ctx); err != nil {
t.Fatalf("flushall: %v", err)
}
audit.mu.Lock()
defer audit.mu.Unlock()
if len(audit.events) != len(actions) {
t.Fatalf("erwartet %d zugestellte events, habe %d", len(actions), len(audit.events))
}
for i, e := range audit.events {
if e.Action != actions[i] {
t.Fatalf("reihenfolge falsch: position %d = %q, want %q", i, e.Action, actions[i])
}
}
// Puffer muss nach bestaetigter Zustellung leer sein.
remaining, err := buffer.PendingAuditEvents(ctx)
if err != nil {
t.Fatalf("pending: %v", err)
}
if len(remaining) != 0 {
t.Fatalf("erwartet leeren puffer nach bestaetigter zustellung, habe %d verbleibende", len(remaining))
}
}
// Bekannter-Fehler-Praevention: bricht die Zustellung waehrend der
// Nachlieferung erneut ab (Core faellt wieder aus), bleiben die NICHT
// bestaetigten Events sicher im Puffer erhalten statt verloren zu gehen.
func TestFlushAll_KeepsUnconfirmedEventsInBufferOnPartialFailure(t *testing.T) {
buffer, pool, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
registry, clientID, secret := setupModuleCredential(t, pool)
for i := 0; i < 5; i++ {
if err := buffer.EnqueueAuditEvent(ctx, "acme", "user-1", fmt.Sprintf("action-%d", i), "", nil); err != nil {
t.Fatalf("enqueue %d: %v", i, err)
}
}
// Core-Fake schlaegt AB dem 3. Event fehl -> simuliert erneuten Ausfall
// mitten in der Nachlieferung.
audit := &fakeAudit{failAt: 3}
handler := NewHandler(registry, audit, newFakeUsage())
mux := http.NewServeMux()
mux.HandleFunc("/internal/resync/audit", handler.AuditHandler)
mux.HandleFunc("/internal/resync/usage", handler.UsageHandler)
server := httptest.NewServer(mux)
defer server.Close()
client := NewCoreClient(server.URL, clientID, secret)
worker := NewWorker(buffer, client)
if err := worker.FlushAll(ctx); err == nil {
t.Fatal("erwartet fehler, da core nur teilweise bestaetigt hat")
}
remaining, err := buffer.PendingAuditEvents(ctx)
if err != nil {
t.Fatalf("pending: %v", err)
}
if len(remaining) != 2 {
t.Fatalf("erwartet 2 verbleibende (nicht bestaetigte) events im puffer, habe %d", len(remaining))
}
}
// Akzeptanzkriterium 3 + Pruefung 2: Nutzungszaehler-Differenz aus der
// Ausfallzeit wird korrekt nachgebucht, ein wiederholter (fehlerhafter)
// Flush-Versuch fuehrt NICHT zu Doppelzaehlung, weil bereits bestaetigte
// Deltas aus dem Puffer entfernt sind.
func TestFlushAll_AppliesUsageDeltasWithoutDoubleCounting(t *testing.T) {
buffer, pool, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
registry, clientID, secret := setupModuleCredential(t, pool)
deltas := []int64{3, 5, 2}
var want int64
for _, d := range deltas {
want += d
if err := buffer.EnqueueUsageDelta(ctx, "acme", "api_calls", d); err != nil {
t.Fatalf("enqueue delta %d: %v", d, err)
}
}
usage := newFakeUsage()
handler := NewHandler(registry, &fakeAudit{failAt: -1}, usage)
mux := http.NewServeMux()
mux.HandleFunc("/internal/resync/audit", handler.AuditHandler)
mux.HandleFunc("/internal/resync/usage", handler.UsageHandler)
server := httptest.NewServer(mux)
defer server.Close()
client := NewCoreClient(server.URL, clientID, secret)
worker := NewWorker(buffer, client)
if err := worker.FlushAll(ctx); err != nil {
t.Fatalf("flushall 1: %v", err)
}
usage.mu.Lock()
got := usage.totals["acme|api_calls"]
usage.mu.Unlock()
if got != want {
t.Fatalf("nutzungsstand nach nachbuchung = %d, want %d", got, want)
}
// Ein zweiter Flush-Versuch (z.B. redundanter Retry) darf NICHTS mehr
// nachbuchen, da der Puffer bereits geleert wurde.
if err := worker.FlushAll(ctx); err != nil {
t.Fatalf("flushall 2: %v", err)
}
usage.mu.Lock()
got2 := usage.totals["acme|api_calls"]
usage.mu.Unlock()
if got2 != want {
t.Fatalf("nutzungsstand nach redundantem zweiten flush = %d, want unveraendert %d (keine doppelzaehlung)", got2, want)
}
}
// Akzeptanzkriterium 1 + Pruefung 3: bei erkannter Core-Wiedererreichbarkeit
// wird der Cache SOFORT invalidiert (naechster Zugriff refetcht), nicht
// erst nach TTL-Ablauf — Latenz wird gemessen und liegt weit unter einer
// langen TTL.
func TestCheckAndSync_InvalidatesCacheImmediatelyOnRecovery(t *testing.T) {
buffer, pool, cleanup := setupTest(t)
defer cleanup()
registry, clientID, secret := setupModuleCredential(t, pool)
_ = registry
mux := http.NewServeMux()
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
server := httptest.NewServer(mux)
defer server.Close()
client := NewCoreClient(server.URL, clientID, secret)
worker := NewWorker(buffer, client)
invalidated := false
var mu sync.Mutex
worker.OnReachable(func() {
mu.Lock()
invalidated = true
mu.Unlock()
})
start := time.Now()
becameReachable, err := worker.CheckAndSync(context.Background())
elapsed := time.Since(start)
if err != nil {
t.Fatalf("checkandsync: %v", err)
}
if !becameReachable {
t.Fatal("erwartet erkannten uebergang zu 'erreichbar' beim ersten erfolgreichen check")
}
mu.Lock()
defer mu.Unlock()
if !invalidated {
t.Fatal("erwartet sofortigen cache-invalidierungs-callback bei core-wiedererreichbarkeit")
}
// Zielwert: deutlich unter einer typischen TTL (z.B. 5s beim
// Feature-Flag-Cache, LIC-02) — hier im Millisekundenbereich, da rein
// lokal ohne Netzwerk-Overhead.
if elapsed > time.Second {
t.Fatalf("cache-invalidierung brauchte %s, erwartet deutlich unter 1s", elapsed)
}
}
+220
View File
@@ -0,0 +1,220 @@
package resync
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"time"
)
// CoreClient ist der modul-seitige HTTP-Client fuer die Nachlieferung,
// authentifiziert ueber dasselbe Service-Credential wie jeder andere
// Modul-Core-Aufruf (API-02).
type CoreClient struct {
BaseURL string
ClientID string
Secret string
HTTP *http.Client
}
func NewCoreClient(baseURL, clientID, secret string) *CoreClient {
return &CoreClient{BaseURL: baseURL, ClientID: clientID, Secret: secret, HTTP: &http.Client{Timeout: 5 * time.Second}}
}
func (c *CoreClient) post(ctx context.Context, path string, body any) (*http.Response, error) {
payload, err := json.Marshal(body)
if err != nil {
return nil, fmt.Errorf("payload serialisieren: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.BaseURL+path, bytes.NewReader(payload))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Nexarch-Client-Id", c.ClientID)
req.Header.Set("X-Nexarch-Client-Secret", c.Secret)
return c.HTTP.Do(req)
}
// HealthCheck prueft, ob Core erreichbar ist — dieselbe Konvention wie
// internal/health (OPS-01): HTTP 200 auf einem Health-Endpunkt.
func (c *CoreClient) HealthCheck(ctx context.Context) bool {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.BaseURL+"/healthz", nil)
if err != nil {
return false
}
resp, err := c.HTTP.Do(req)
if err != nil {
return false
}
defer resp.Body.Close()
return resp.StatusCode == http.StatusOK
}
// Worker erkennt Core-Wiedererreichbarkeit und stoesst DANN sofort
// (Akzeptanzkriterium 1) sowohl registrierte Cache-Invalidierungen als auch
// das Nachliefern des lokalen Puffers an.
type Worker struct {
buffer *Buffer
client *CoreClient
onReachable []func()
wasDown bool
}
func NewWorker(buffer *Buffer, client *CoreClient) *Worker {
return &Worker{buffer: buffer, client: client, wasDown: true} // Start pessimistisch: erster erfolgreicher Check zaehlt als "Wiedererreichbarkeit".
}
// OnReachable registriert einen Callback, der bei jeder erkannten
// Core-Wiedererreichbarkeit sofort ausgefuehrt wird — z.B.
// moduletrust.StaleCache[T].Invalidate, damit der naechste Zugriff sofort
// neu abruft statt auf TTL-Ablauf zu warten (Akzeptanzkriterium 1).
func (w *Worker) OnReachable(fn func()) {
w.onReachable = append(w.onReachable, fn)
}
// CheckAndSync fuehrt EINEN Zyklus aus: Erreichbarkeit pruefen, bei
// erkanntem UEBERGANG "nicht erreichbar -> erreichbar" sofort die
// registrierten Callbacks ausloesen und den Puffer nachliefern. Gibt
// zurueck, ob ein Wiederanlauf in diesem Aufruf erkannt wurde (fuer
// Latenzmessung in Tests, Pruefung 3).
func (w *Worker) CheckAndSync(ctx context.Context) (becameReachable bool, err error) {
reachable := w.client.HealthCheck(ctx)
if !reachable {
w.wasDown = true
return false, nil
}
justRecovered := w.wasDown
w.wasDown = false
if !justRecovered {
return false, nil
}
for _, fn := range w.onReachable {
fn()
}
if err := w.FlushAll(ctx); err != nil {
return true, err
}
return true, nil
}
// FlushAll liefert ZUERST alle gepufferten Audit-Events (in Reihenfolge),
// DANN alle gepufferten Nutzungsdeltas nach. Jedes Element wird aus dem
// lokalen Puffer NUR entfernt, wenn Core es bestaetigt hat — bricht die
// Uebertragung vorzeitig ab (Core faellt waehrend der Nachlieferung erneut
// aus), bleibt der Rest sicher im Postgres-Puffer erhalten
// (Akzeptanzkriterium 2/3, "Bekannte Fehler vermeiden").
func (w *Worker) FlushAll(ctx context.Context) error {
if err := w.flushAuditEvents(ctx); err != nil {
return err
}
return w.flushUsageDeltas(ctx)
}
func (w *Worker) flushAuditEvents(ctx context.Context) error {
events, err := w.buffer.PendingAuditEvents(ctx)
if err != nil {
return err
}
if len(events) == 0 {
return nil
}
dtos := make([]auditEventDTO, len(events))
for i, e := range events {
dtos[i] = auditEventDTO{
TenantSlug: e.TenantSlug, Actor: e.Actor, Action: e.Action, Target: e.Target,
Metadata: e.Metadata, OccurredAt: e.CreatedAt,
}
}
resp, err := w.client.post(ctx, "/internal/resync/audit", dtos)
if err != nil {
return fmt.Errorf("audit-nachlieferung: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("audit-nachlieferung: unerwarteter status %d", resp.StatusCode)
}
var ack struct {
Written int `json:"written"`
}
if err := json.NewDecoder(resp.Body).Decode(&ack); err != nil {
return fmt.Errorf("audit-bestaetigung lesen: %w", err)
}
// NUR die von Core bestaetigt geschriebenen Events entfernen — sie sind
// nach PendingAuditEvents-Reihenfolge sortiert, die ersten `Written`
// Eintraege entsprechen also genau den bestaetigten.
for i := 0; i < ack.Written; i++ {
if err := w.buffer.RemoveAuditEvent(ctx, events[i].ID); err != nil {
return fmt.Errorf("bestaetigtes audit-event aus puffer entfernen: %w", err)
}
}
if ack.Written < len(events) {
return fmt.Errorf("core hat nur %d von %d audit-events bestaetigt", ack.Written, len(events))
}
return nil
}
func (w *Worker) flushUsageDeltas(ctx context.Context) error {
deltas, err := w.buffer.PendingUsageDeltas(ctx)
if err != nil {
return err
}
if len(deltas) == 0 {
return nil
}
dtos := make([]usageDeltaDTO, len(deltas))
for i, d := range deltas {
dtos[i] = usageDeltaDTO{TenantSlug: d.TenantSlug, Metric: d.Metric, Delta: d.Delta}
}
resp, err := w.client.post(ctx, "/internal/resync/usage", dtos)
if err != nil {
return fmt.Errorf("nutzungs-nachlieferung: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("nutzungs-nachlieferung: unerwarteter status %d", resp.StatusCode)
}
var ack struct {
Applied int `json:"applied"`
}
if err := json.NewDecoder(resp.Body).Decode(&ack); err != nil {
return fmt.Errorf("nutzungs-bestaetigung lesen: %w", err)
}
for i := 0; i < ack.Applied; i++ {
if err := w.buffer.RemoveUsageDelta(ctx, deltas[i].ID); err != nil {
return fmt.Errorf("bestaetigtes nutzungsdelta aus puffer entfernen: %w", err)
}
}
if ack.Applied < len(deltas) {
return fmt.Errorf("core hat nur %d von %d nutzungsdeltas bestaetigt", ack.Applied, len(deltas))
}
return nil
}
// Run fuehrt CheckAndSync in festen Abstaenden aus — die "kurze, definierte
// Zeitspanne" aus Akzeptanzkriterium 1 ist dieses Poll-Intervall.
func (w *Worker) Run(ctx context.Context, interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
_, _ = w.CheckAndSync(ctx)
}
}
}
+2
View File
@@ -0,0 +1,2 @@
DROP TABLE resync_usage_buffer;
DROP TABLE resync_audit_buffer;
+22
View File
@@ -0,0 +1,22 @@
-- Wiederanlauf & Nachsynchronisierung nach Core-Ausfall (API-06, siehe
-- core-kanban/tickets/API-06.md) — lokale, PERSISTENTE Pufferqueue (kein
-- In-Memory) fuer Audit-Events und Nutzungszaehler-Deltas eines Moduls.
CREATE TABLE resync_audit_buffer (
id BIGSERIAL PRIMARY KEY,
seq BIGSERIAL,
tenant_slug TEXT NOT NULL,
actor TEXT NOT NULL,
action TEXT NOT NULL,
target TEXT NOT NULL DEFAULT '',
metadata JSONB NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE resync_usage_buffer (
id BIGSERIAL PRIMARY KEY,
seq BIGSERIAL,
tenant_slug TEXT NOT NULL,
metric TEXT NOT NULL,
delta BIGINT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-1
View File
@@ -1 +0,0 @@
DROP TABLE tenant_keks;
-10
View File
@@ -1,10 +0,0 @@
-- Master-Key-Verwaltung & Tenant-Schluesselhierarchie (API-10, siehe
-- core-kanban/tickets/API-10.md) — EIN verpackter (mit dem Master-Key
-- umhuellter) Tenant-KEK je Mandant. Niemals der Master-Key selbst und
-- niemals ein Tenant-KEK im Klartext in dieser Tabelle.
CREATE TABLE tenant_keks (
tenant_id UUID PRIMARY KEY REFERENCES tenants(id),
wrapped_kek BYTEA NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
rotated_at TIMESTAMPTZ
);