From 4f07bc73a2673ae0709a9440963dc7728d86b5d4 Mon Sep 17 00:00:00 2001 From: patrick Date: Tue, 27 Oct 2026 14:40:46 +0100 Subject: [PATCH] feat: versioniertes Modul common/ (ohne Verbraucher) Pakete: apierror, auditclient, httpclient, httpharden, loadtest, migrate, policyclient, svcauth, tenantregistryclient, testdb. Code stammt aus den bisher byte-gleichen Modulkopien (ohne Kommentare/Importpfade verglichen). migrate/testdb brauchen pgx v5.9.2. Auslieferung per Git-Tag common/vX.Y.Z (Workflow common-tag.yml, setzt nach Merge in main den naechsten Patch-Tag nach Build/Vet/Test). Kein replace, damit Module einzeln baubar bleiben. Verbraucher werden in Folge-PRs umgestellt. Co-Authored-By: Claude Sonnet 5.5 --- .gitea/workflows/common-tag.yml | 32 +++ PRODUKTARCHITEKTUR.md | 2 + common/README.md | 23 ++ common/apierror/apierror.go | 191 +++++++++++++++ common/apierror/apierror_test.go | 184 +++++++++++++++ common/auditclient/auditclient.go | 71 ++++++ common/auditclient/auditclient_test.go | 57 +++++ common/go.mod | 15 ++ common/go.sum | 26 +++ common/httpclient/httpclient.go | 29 +++ common/httpclient/httpclient_test.go | 105 +++++++++ common/httpharden/harden.go | 119 ++++++++++ common/httpharden/harden_test.go | 105 +++++++++ common/loadtest/loadtest.go | 141 +++++++++++ common/migrate/migrate.go | 221 ++++++++++++++++++ common/policyclient/policyclient.go | 52 +++++ common/policyclient/policyclient_test.go | 80 +++++++ common/svcauth/svcauth.go | 18 ++ common/svcauth/svcauth_test.go | 23 ++ .../tenantregistryclient.go | 47 ++++ .../tenantregistryclient_test.go | 59 +++++ common/testdb/testdb.go | 62 +++++ 22 files changed, 1662 insertions(+) create mode 100644 .gitea/workflows/common-tag.yml create mode 100644 common/README.md create mode 100644 common/apierror/apierror.go create mode 100644 common/apierror/apierror_test.go create mode 100644 common/auditclient/auditclient.go create mode 100644 common/auditclient/auditclient_test.go create mode 100644 common/go.mod create mode 100644 common/go.sum create mode 100644 common/httpclient/httpclient.go create mode 100644 common/httpclient/httpclient_test.go create mode 100644 common/httpharden/harden.go create mode 100644 common/httpharden/harden_test.go create mode 100644 common/loadtest/loadtest.go create mode 100644 common/migrate/migrate.go create mode 100644 common/policyclient/policyclient.go create mode 100644 common/policyclient/policyclient_test.go create mode 100644 common/svcauth/svcauth.go create mode 100644 common/svcauth/svcauth_test.go create mode 100644 common/tenantregistryclient/tenantregistryclient.go create mode 100644 common/tenantregistryclient/tenantregistryclient_test.go create mode 100644 common/testdb/testdb.go diff --git a/.gitea/workflows/common-tag.yml b/.gitea/workflows/common-tag.yml new file mode 100644 index 00000000..5c8a665d --- /dev/null +++ b/.gitea/workflows/common-tag.yml @@ -0,0 +1,32 @@ +name: common Versions-Tag + +on: + push: + branches: [main] + paths: + - "common/**" + +jobs: + tag: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + - uses: actions/setup-go@v5 + with: + go-version: "1.25" + - name: Build, Vet, Test + working-directory: common + run: | + go build ./... + go vet ./... + go test ./... -count=1 + - name: Naechsten Patch-Tag setzen + run: | + last=$(git tag -l 'common/v0.*' --sort=-v:refname | head -n1) + if [ -z "$last" ]; then next="common/v0.1.0"; else + patch=${last##*.}; base=${last%.*}; next="${base}.$((patch + 1))" + fi + git tag "$next" + git push origin "$next" diff --git a/PRODUKTARCHITEKTUR.md b/PRODUKTARCHITEKTUR.md index 631e0fe3..4bb31fa4 100644 --- a/PRODUKTARCHITEKTUR.md +++ b/PRODUKTARCHITEKTUR.md @@ -64,6 +64,8 @@ Kein Modul kopiert die Daten eines anderen — stattdessen ein wiederkehrendes * Gemeinsamer Nenner aller drei: **Referenz statt Kopie, Adapter statt Direktkopplung.** +Go-`internal/`-Pakete sind zwischen den Modulen nicht sichtbar; identische Querschnittspakete liegen deshalb im versionierten Modul `common/` (Git-Tags `common/vX.Y.Z`, siehe `common/README.md`). So bleiben Module einzeln baubar (Docker, eigenes Repo, Partner-Build). Fachlich verschiedene Pakete (`audittrail`, `cutoverstage`, `crypto`) bleiben modul-lokal. + Dasselbe Prinzip gilt auch innerhalb eines Moduls, auf Code-Ebene: Hauptdateien/Entry-Points bleiben klein, Fachlogik wandert in eigene Module. Vor einer neuen Funktions-/Modul-Datei wird geprüft, ob dieselbe Funktionalität nicht schon woanders existiert (auch modulübergreifend, z. B. in Core) — keine doppelten Funktionsdateien. Das eigene Kanban-Planungswerkzeug wurde nach genau diesem Muster korrigiert (sieben `kanban.html` enthielten identisches JS/CSS als Volltext-Duplikat, damals in `_shared/kanban.js`/`kanban.css` ausgelagert) — die Boards selbst sind inzwischen komplett aus dem Repo entfernt (Ticket-Historie liegt in der Gitea-PR-Historie), das Prinzip "keine doppelten Funktionsdateien" gilt unverändert für den eigentlichen NEXARCH-Code. ## 6. Technische Architektur (Kurzfassung, Details in eigenen Dokumenten) diff --git a/common/README.md b/common/README.md new file mode 100644 index 00000000..d0176fb5 --- /dev/null +++ b/common/README.md @@ -0,0 +1,23 @@ +# common + +Geteiltes Go-Modul für Pakete, die in mehreren NEXARCH-Modulen identisch gebraucht werden (`internal/` ist modulübergreifend nicht sichtbar). + +Pakete: `apierror`, `auditclient`, `httpclient`, `httpharden`, `loadtest`, `migrate`, `policyclient`, `svcauth`, `tenantregistryclient`, `testdb`. + +## Einbinden + +``` +go env -w GOPRIVATE=gitea.perlbach24.de +go get gitea.perlbach24.de/scripte/nexarch/common@latest +``` + +Build-Orte brauchen Lesezugriff auf Gitea: Read-only-Token in `~/.netrc` (`machine gitea.perlbach24.de login password `). + +## Versionen + +Tags `common/v0.X.Y` setzt `.gitea/workflows/common-tag.yml` nach jedem Merge in `main`, der `common/` ändert. Module aktualisieren einzeln (`go get ...common@v0.X.Y`). Lokal mit mehreren Modulen gleichzeitig arbeiten: nicht eingechecktes `go.work`. + +## Regeln + +- Nur Code, der in mindestens zwei Modulen byte-gleich (ohne Kommentare) gebraucht wird; fachlich verschiedene Pakete bleiben modul-lokal. +- Änderungen abwärtskompatibel halten, solange Module nicht gleichzeitig umgestellt werden. diff --git a/common/apierror/apierror.go b/common/apierror/apierror.go new file mode 100644 index 00000000..4872e81d --- /dev/null +++ b/common/apierror/apierror.go @@ -0,0 +1,191 @@ +// Package apierror erzwingt das Core-Fehlerformat (API-01) auf allen +// HTTP-Antworten und schliesst QA-15 F-14: 116 Stellen in Core, DMS, Archive +// und Mail betteten err.Error() direkt in Antworten ein, darunter 500er mit +// Datenbank- und Dateipfadtexten. Statt jede einzelne Aufrufstelle +// anzufassen, schreibt Normalize JEDE Fehlerantwort am aeussersten Rand +// (httpharden.ListenAndServe) um: 4xx-Meldungen bleiben inhaltlich +// erhalten (fachlich, keine Interna), 5xx-Antworten werden PRINZIPIELL +// durch eine allgemeine Meldung mit Korrelations-ID ersetzt — der +// urspruengliche Text (der die Interna enthalten haben KANN) geht nur ins +// Log, niemals in die Antwort. Erweitert dasselbe Grundmuster wie +// dms/internal/apierror (INT-01, dort das "Vorbild" laut API-20). +package apierror + +import ( + "bytes" + "crypto/rand" + "encoding/hex" + "encoding/json" + "log" + "net/http" + "strings" +) + +func CodeForStatus(status int) string { + switch status { + case http.StatusBadRequest: + return "invalid_request" + case http.StatusUnauthorized: + return "unauthenticated" + case http.StatusForbidden: + return "forbidden" + case http.StatusNotFound: + return "not_found" + case http.StatusMethodNotAllowed: + return "method_not_allowed" + case http.StatusConflict: + return "conflict" + case http.StatusGone: + return "gone" + case http.StatusRequestEntityTooLarge: + return "payload_too_large" + case http.StatusUnsupportedMediaType: + return "unsupported_media_type" + case http.StatusUnprocessableEntity: + return "unprocessable" + case http.StatusLocked: + return "locked" + case http.StatusTooManyRequests: + return "rate_limited" + case http.StatusServiceUnavailable: + return "unavailable" + } + if status >= 500 { + return "internal_error" + } + return "invalid_request" +} + +type body struct { + Error struct { + Code string `json:"code"` + Message string `json:"message"` + CorrelationID string `json:"correlation_id,omitempty"` + } `json:"error"` +} + +func Write(w http.ResponseWriter, status int, code, message string) { + var b body + b.Error.Code = code + b.Error.Message = message + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(b) +} + +func newCorrelationID() string { + buf := make([]byte, 8) + if _, err := rand.Read(buf); err != nil { + // crypto/rand-Ausfall ist praktisch nie beobachtbar; eine feste + // Ersatz-ID ist besser als ein Absturz in einem Fehlerpfad, der + // selbst schon einen Fehler meldet. + return "korrelations-id-nicht-verfuegbar" + } + return hex.EncodeToString(buf) +} + +// Normalize schreibt jede Fehlerantwort (Status >= 400) im Core-Format um. +// 4xx: der vom Handler geschriebene Text bleibt als "message" erhalten (er +// ist fachlich, kein Interna-Leck). 5xx (Akzeptanzkriterium 1): der Text +// wird verworfen, NIE an den Aufrufer weitergegeben — stattdessen eine +// allgemeine Meldung mit Korrelations-ID; derselbe Text landet mitsamt +// Korrelations-ID im Server-Log (Prozess-stderr, wie ueberall sonst in +// diesem Projekt). +func Normalize(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rec := &recorder{ResponseWriter: w} + next.ServeHTTP(rec, r) + rec.finish(r) + }) +} + +type recorder struct { + http.ResponseWriter + status int + buffered bool + buf bytes.Buffer +} + +func (r *recorder) WriteHeader(status int) { + if r.status != 0 { + return + } + r.status = status + if status >= 400 { + r.buffered = true + return + } + r.ResponseWriter.WriteHeader(status) +} + +func (r *recorder) Write(p []byte) (int, error) { + if r.status == 0 { + r.WriteHeader(http.StatusOK) + } + if r.buffered { + return r.buf.Write(p) + } + return r.ResponseWriter.Write(p) +} + +func (r *recorder) Flush() { + if f, ok := r.ResponseWriter.(http.Flusher); ok && !r.buffered { + f.Flush() + } +} + +func (r *recorder) Unwrap() http.ResponseWriter { return r.ResponseWriter } + +func (r *recorder) finish(req *http.Request) { + if !r.buffered { + return + } + code := CodeForStatus(r.status) + rawMessage := strings.TrimSpace(r.buf.String()) + + if r.status >= 500 { + corrID := newCorrelationID() + log.Printf("apierror: 5xx %s %s status=%d korrelations-id=%s ursprung=%q", req.Method, req.URL.Path, r.status, corrID, rawMessage) + out := body{} + out.Error.Code = code + out.Error.Message = "Interner Fehler" + out.Error.CorrelationID = corrID + encoded, _ := json.Marshal(out) + r.flush(append(encoded, '\n')) + return + } + + out := map[string]any{} + message := rawMessage + var parsed map[string]any + if json.Unmarshal(r.buf.Bytes(), &parsed) == nil && parsed != nil { + switch e := parsed["error"].(type) { + case string: + message = e + delete(parsed, "error") + out = parsed + case map[string]any: + if c, _ := e["code"].(string); c != "" { + if m, _ := e["message"].(string); m != "" { + r.flush(r.buf.Bytes()) + return + } + } + } + } + if message == "" { + message = http.StatusText(r.status) + } + out["error"] = map[string]string{"code": code, "message": message} + encoded, _ := json.Marshal(out) + r.flush(append(encoded, '\n')) +} + +func (r *recorder) flush(payload []byte) { + h := r.ResponseWriter.Header() + h.Set("Content-Type", "application/json") + h.Del("Content-Length") + h.Set("X-Content-Type-Options", "nosniff") + r.ResponseWriter.WriteHeader(r.status) + _, _ = r.ResponseWriter.Write(payload) +} diff --git a/common/apierror/apierror_test.go b/common/apierror/apierror_test.go new file mode 100644 index 00000000..e26d1cdb --- /dev/null +++ b/common/apierror/apierror_test.go @@ -0,0 +1,184 @@ +package apierror + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func decode(t *testing.T, rec *httptest.ResponseRecorder) map[string]any { + t.Helper() + var m map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &m); err != nil { + t.Fatalf("keine json-antwort: %q (%v)", rec.Body.String(), err) + } + return m +} + +func errOf(t *testing.T, m map[string]any) map[string]any { + t.Helper() + e, ok := m["error"].(map[string]any) + if !ok { + t.Fatalf("error ist kein objekt: %v", m) + } + return e +} + +// Akzeptanzkriterium 2: fachliche 4xx-Meldungen bleiben verstaendlich. +func TestNormalize_4xxKeepsHandlerMessage(t *testing.T) { + cases := []struct { + name string + handler http.HandlerFunc + status int + wantCode string + wantMsg string + }{ + {"http.Error klartext", func(w http.ResponseWriter, r *http.Request) { http.Error(w, "dokument nicht gefunden", 404) }, 404, "not_found", "dokument nicht gefunden"}, + {"json string", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(422) + _, _ = w.Write([]byte(`{"error":"ungueltiger wert"}`)) + }, 422, "unprocessable", "ungueltiger wert"}, + {"leerer body", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(403) }, 403, "forbidden", "Forbidden"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + rec := httptest.NewRecorder() + Normalize(c.handler).ServeHTTP(rec, httptest.NewRequest("GET", "/x", nil)) + if rec.Code != c.status { + t.Fatalf("status = %d", rec.Code) + } + if ct := rec.Header().Get("Content-Type"); !strings.HasPrefix(ct, "application/json") { + t.Errorf("content-type = %q", ct) + } + e := errOf(t, decode(t, rec)) + if e["code"] != c.wantCode || e["message"] != c.wantMsg { + t.Errorf("code=%v msg=%v, erwartet %q/%q", e["code"], e["message"], c.wantCode, c.wantMsg) + } + if _, has := e["correlation_id"]; has { + t.Errorf("4xx sollte keine korrelations-id tragen: %v", e) + } + }) + } +} + +// Akzeptanzkriterium 1: 5xx-Antworten enthalten NIE den Handler-Text +// (der die Interna leaken kann), nur eine allgemeine Meldung mit +// Korrelations-ID — egal, was der Handler geschrieben hat. +func TestNormalize_5xxNeverLeaksHandlerText(t *testing.T) { + cases := []struct { + name string + handler http.HandlerFunc + }{ + {"http.Error mit db-fehler", func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `pq: connection to server at "10.0.0.5", port 5432 failed`, 500) + }}, + {"http.Error mit dateipfad", func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "open /var/lib/nexarch/secrets/tenant.key: permission denied", 502) + }}, + {"json mit interna", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(503) + _, _ = w.Write([]byte(`{"error":"dial tcp 127.0.0.1:5432: connect: connection refused"}`)) + }}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + rec := httptest.NewRecorder() + Normalize(c.handler).ServeHTTP(rec, httptest.NewRequest("GET", "/x", nil)) + body := rec.Body.String() + for _, leak := range []string{"pq:", "10.0.0.5", "/var/lib/nexarch", "permission denied", "127.0.0.1", "dial tcp"} { + if strings.Contains(body, leak) { + t.Errorf("5xx-antwort enthaelt interna %q: %s", leak, body) + } + } + e := errOf(t, decode(t, rec)) + if e["message"] != "Interner Fehler" { + t.Errorf("message = %v, erwartet generische meldung", e["message"]) + } + corrID, _ := e["correlation_id"].(string) + if corrID == "" { + t.Fatalf("keine korrelations-id in der antwort: %v", e) + } + if len(corrID) < 8 { + t.Errorf("korrelations-id zu kurz/vorhersagbar: %q", corrID) + } + }) + } +} + +// Zwei Anfragen mit demselben 5xx-Fehler bekommen unterschiedliche +// Korrelations-IDs (sonst waere die ID im Log nicht eindeutig zuordenbar). +func TestNormalize_5xxCorrelationIDsAreUnique(t *testing.T) { + handler := func(w http.ResponseWriter, r *http.Request) { http.Error(w, "boom", 500) } + var ids []string + for i := 0; i < 5; i++ { + rec := httptest.NewRecorder() + Normalize(http.HandlerFunc(handler)).ServeHTTP(rec, httptest.NewRequest("GET", "/x", nil)) + e := errOf(t, decode(t, rec)) + id, _ := e["correlation_id"].(string) + for _, seen := range ids { + if seen == id { + t.Fatalf("korrelations-id doppelt vergeben: %q", id) + } + } + ids = append(ids, id) + } +} + +func TestNormalize_KeepsExtraFieldsOn4xx(t *testing.T) { + rec := httptest.NewRecorder() + Normalize(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(409) + _, _ = w.Write([]byte(`{"error":"duplikat","documentId":"d1"}`)) + })).ServeHTTP(rec, httptest.NewRequest("POST", "/x", nil)) + m := decode(t, rec) + if m["documentId"] != "d1" { + t.Errorf("zusatzfeld verloren: %v", m) + } + if errOf(t, m)["code"] != "conflict" { + t.Errorf("code = %v", errOf(t, m)["code"]) + } + + rec = httptest.NewRecorder() + Normalize(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + Write(w, 401, "unauthenticated", "nicht angemeldet") + })).ServeHTTP(rec, httptest.NewRequest("GET", "/x", nil)) + e := errOf(t, decode(t, rec)) + if e["code"] != "unauthenticated" || e["message"] != "nicht angemeldet" { + t.Errorf("core-format veraendert: %v", e) + } +} + +func TestNormalize_SuccessPassesThroughUnbuffered(t *testing.T) { + rec := httptest.NewRecorder() + Normalize(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/pdf") + _, _ = w.Write([]byte("%PDF-1.7")) + })).ServeHTTP(rec, httptest.NewRequest("GET", "/x", nil)) + if rec.Code != 200 || rec.Body.String() != "%PDF-1.7" || rec.Header().Get("Content-Type") != "application/pdf" { + t.Errorf("erfolgsantwort veraendert: %d %q %q", rec.Code, rec.Body.String(), rec.Header().Get("Content-Type")) + } +} + +func TestNormalize_MuxNotFoundAndMethodNotAllowed(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("GET /a", func(w http.ResponseWriter, r *http.Request) {}) + h := Normalize(mux) + for _, c := range []struct { + method, path string + status int + code string + }{{"GET", "/nope", 404, "not_found"}, {"POST", "/a", 405, "method_not_allowed"}} { + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest(c.method, c.path, nil)) + if rec.Code != c.status { + t.Fatalf("%s %s: status %d", c.method, c.path, rec.Code) + } + if errOf(t, decode(t, rec))["code"] != c.code { + t.Errorf("%s %s: code %v", c.method, c.path, errOf(t, decode(t, rec))["code"]) + } + } +} diff --git a/common/auditclient/auditclient.go b/common/auditclient/auditclient.go new file mode 100644 index 00000000..05fe2e61 --- /dev/null +++ b/common/auditclient/auditclient.go @@ -0,0 +1,71 @@ +// Package auditclient ist ein schlanker HTTP-Client fuer Core AUD-08 +// (internal/audit, POST /audit/ingest). Connect ist ein physisch +// getrenntes Go-Modul und kann Cores internal/audit daher nicht direkt +// importieren — nur ueber dessen HTTP-Schnittstelle. Gleiches Muster wie +// archive/internal/auditclient. +package auditclient + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + + "gitea.perlbach24.de/scripte/nexarch/common/httpclient" +) + +type Client struct { + BaseURL string + ServiceToken string + HTTPClient *http.Client +} + +func New(baseURL, serviceToken string) *Client { + return &Client{BaseURL: baseURL, ServiceToken: serviceToken, HTTPClient: httpclient.Default()} +} + +type ingestRequest struct { + TenantSlug string `json:"tenant_slug"` + Actor string `json:"actor"` + Action string `json:"action"` + Target string `json:"target"` + Metadata map[string]any `json:"metadata"` +} + +// Ingest meldet EIN Audit-Ereignis an Core. Jeder Fehler (Transport, +// Timeout, unerwarteter Status) wird als Fehler zurueckgegeben — der +// Aufrufer entscheidet, ob/wie er das nachbehandelt (siehe +// connect/internal/audittrail fuer den ausfalltoleranten Wrapper). +func (c *Client) Ingest(ctx context.Context, tenantSlug, actor, action, target string, metadata map[string]any) error { + body, err := json.Marshal(ingestRequest{ + TenantSlug: tenantSlug, Actor: actor, Action: action, Target: target, Metadata: metadata, + }) + if err != nil { + return fmt.Errorf("auditclient: request kodieren: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.BaseURL+"/audit/ingest", bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("auditclient: request bauen: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Service-Token", c.ServiceToken) + + resp, err := c.httpClient().Do(req) + if err != nil { + return fmt.Errorf("auditclient: aufruf fehlgeschlagen: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusAccepted { + return fmt.Errorf("auditclient: unerwarteter status %d", resp.StatusCode) + } + return nil +} + +func (c *Client) httpClient() *http.Client { + if c.HTTPClient != nil { + return c.HTTPClient + } + return httpclient.Default() +} diff --git a/common/auditclient/auditclient_test.go b/common/auditclient/auditclient_test.go new file mode 100644 index 00000000..0d37be35 --- /dev/null +++ b/common/auditclient/auditclient_test.go @@ -0,0 +1,57 @@ +package auditclient + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestIngest_SendsTenantSlugAndServiceToken(t *testing.T) { + var gotBody ingestRequest + var gotToken string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotToken = r.Header.Get("X-Service-Token") + if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil { + t.Fatalf("body dekodieren: %v", err) + } + w.WriteHeader(http.StatusAccepted) + })) + defer srv.Close() + + c := New(srv.URL, "svc-token-123") + err := c.Ingest(context.Background(), "acme", "user-1", "connector.sync", "erp:kunde:42", map[string]any{"count": 3}) + if err != nil { + t.Fatalf("Ingest: %v", err) + } + + if gotToken != "svc-token-123" { + t.Fatalf("Service-Token nicht gesendet, erhalten: %q", gotToken) + } + if gotBody.TenantSlug != "acme" { + t.Fatalf("TenantSlug fehlt/falsch im Audit-Ereignis: %q", gotBody.TenantSlug) + } + if gotBody.Action != "connector.sync" { + t.Fatalf("Action falsch: %q", gotBody.Action) + } +} + +func TestIngest_UnexpectedStatus_ReturnsError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + c := New(srv.URL, "svc-token-123") + if err := c.Ingest(context.Background(), "acme", "user-1", "connector.sync", "erp:kunde:42", nil); err == nil { + t.Fatal("erwarteter Fehler bei Status 500 blieb aus") + } +} + +func TestIngest_TransportError_ReturnsError(t *testing.T) { + c := New("http://127.0.0.1:1", "svc-token-123") + if err := c.Ingest(context.Background(), "acme", "user-1", "connector.sync", "erp:kunde:42", nil); err == nil { + t.Fatal("erwarteter Transport-Fehler blieb aus") + } +} diff --git a/common/go.mod b/common/go.mod new file mode 100644 index 00000000..b6244d53 --- /dev/null +++ b/common/go.mod @@ -0,0 +1,15 @@ +module gitea.perlbach24.de/scripte/nexarch/common + +go 1.25.0 + +toolchain go1.25.13 + +require github.com/jackc/pgx/v5 v5.9.2 + +require ( + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + golang.org/x/sync v0.17.0 // indirect + golang.org/x/text v0.29.0 // indirect +) diff --git a/common/go.sum b/common/go.sum new file mode 100644 index 00000000..f5b2410e --- /dev/null +++ b/common/go.sum @@ -0,0 +1,26 @@ +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/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-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw= +github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= +golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= +golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/common/httpclient/httpclient.go b/common/httpclient/httpclient.go new file mode 100644 index 00000000..53c9a9f4 --- /dev/null +++ b/common/httpclient/httpclient.go @@ -0,0 +1,29 @@ +// Package httpclient liefert den HTTP-Client fuer ausgehende Dienstaufrufe +// (QA-15 F-11): Zeitlimit statt endlosem Warten auf einen haengenden +// Nachbardienst, und keine Weiterleitungen (ein Service-Token wuerde sonst an +// das Weiterleitungsziel mitgesendet). +package httpclient + +import ( + "net/http" + "time" +) + +// DefaultTimeout gilt fuer einen kompletten Aufruf (Verbindung bis Antwortende). +const DefaultTimeout = 15 * time.Second + +var shared = New(DefaultTimeout) + +// Default ist der gemeinsame Client mit DefaultTimeout. +func Default() *http.Client { return shared } + +// New liefert einen Client mit dem angegebenen Zeitlimit, der Weiterleitungen +// nicht folgt (die 3xx-Antwort wird zurueckgegeben). +func New(timeout time.Duration) *http.Client { + return &http.Client{ + Timeout: timeout, + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }, + } +} diff --git a/common/httpclient/httpclient_test.go b/common/httpclient/httpclient_test.go new file mode 100644 index 00000000..724c9ea7 --- /dev/null +++ b/common/httpclient/httpclient_test.go @@ -0,0 +1,105 @@ +package httpclient + +import ( + "io/fs" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "regexp" + "strings" + "testing" + "time" +) + +func TestDefaultHasTimeout(t *testing.T) { + if Default().Timeout != DefaultTimeout || DefaultTimeout <= 0 { + t.Fatalf("timeout = %s", Default().Timeout) + } +} + +func TestClientGivesUpOnHangingServer(t *testing.T) { + release := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-release })) + defer srv.Close() + defer close(release) + + start := time.Now() + resp, err := New(200 * time.Millisecond).Get(srv.URL) + if err == nil { + _ = resp.Body.Close() + t.Fatal("kein fehler bei haengendem server") + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("aufruf dauerte %s", elapsed) + } +} + +func TestClientDoesNotFollowRedirects(t *testing.T) { + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Error("weiterleitungsziel wurde aufgerufen (Service-Token-Leck)") + })) + defer target.Close() + redirector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target.URL, http.StatusTemporaryRedirect) + })) + defer redirector.Close() + + req, _ := http.NewRequest(http.MethodPost, redirector.URL, strings.NewReader("x")) + req.Header.Set("X-Service-Token", "geheim") + resp, err := Default().Do(req) + if err != nil { + t.Fatalf("do: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusTemporaryRedirect { + t.Fatalf("status = %d, erwartet 307", resp.StatusCode) + } +} + +// Architekturtest: im Produktivcode von common gibt es keine Aufrufe ueber den +// Standard-Client ohne Zeitlimit. Die Verbraucher-Module pruefen ihren eigenen +// Code selbst. +func TestNoUnboundedDefaultClientInProductionCode(t *testing.T) { + root, err := filepath.Abs(filepath.Join("..")) + if err != nil { + t.Fatal(err) + } + scanNoDefaultClient(t, root, root) +} + +func scanNoDefaultClient(t *testing.T, repoRoot, root string) { + t.Helper() + forbidden := regexp.MustCompile(`http\.DefaultClient|http\.(Get|Post|Head|PostForm)\(`) + err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + if n := d.Name(); n == "node_modules" || n == "testdata" || n == ".git" { + return filepath.SkipDir + } + return nil + } + if !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") { + return nil + } + data, err := os.ReadFile(path) + if err != nil { + return err + } + for i, line := range strings.Split(string(data), "\n") { + if t := strings.TrimSpace(line); strings.HasPrefix(t, "//") { + continue + } + if forbidden.MatchString(line) { + rel, _ := filepath.Rel(repoRoot, path) + t.Errorf("%s:%d nutzt den Standard-Client ohne Zeitlimit: %s", rel, i+1, strings.TrimSpace(line)) + } + } + return nil + }) + if err != nil { + t.Fatal(err) + } +} diff --git a/common/httpharden/harden.go b/common/httpharden/harden.go new file mode 100644 index 00000000..66af8005 --- /dev/null +++ b/common/httpharden/harden.go @@ -0,0 +1,119 @@ +// Package httpharden liefert gehaertete http.Server-Grundeinstellungen +// (Timeouts, Header-Limit) und ein Body-Limit fuer alle Dienste (QA-15 F-05). +// Dieselbe Datei existiert in archive/internal/httpharden und +// dms/internal/httpharden, weil die Module getrennte go.mod-Dateien haben. +package httpharden + +import ( + "log" + "net/http" + "os" + "strconv" + "time" + + "gitea.perlbach24.de/scripte/nexarch/common/apierror" +) + +const ( + DefaultReadHeaderTimeout = 10 * time.Second + DefaultReadTimeout = 30 * time.Second + DefaultWriteTimeout = 60 * time.Second + DefaultIdleTimeout = 120 * time.Second + DefaultMaxHeaderBytes = 1 << 20 + DefaultMaxBodyBytes = 1 << 20 + + EnvReadHeaderTimeout = "NEXARCH_HTTP_READ_HEADER_TIMEOUT_SECONDS" + EnvReadTimeout = "NEXARCH_HTTP_READ_TIMEOUT_SECONDS" + EnvWriteTimeout = "NEXARCH_HTTP_WRITE_TIMEOUT_SECONDS" + EnvIdleTimeout = "NEXARCH_HTTP_IDLE_TIMEOUT_SECONDS" + EnvMaxBodyBytes = "NEXARCH_HTTP_MAX_BODY_BYTES" +) + +// Options weicht vom Standard ab. Nullwerte bedeuten Standard bzw. Env-Wert; +// StreamingBodies/StreamingResponses setzen ReadTimeout/WriteTimeout auf 0 +// (unbegrenzt), damit lange Uploads und Downloads nicht abbrechen. Schutz +// gegen langsame Header bleibt ueber ReadHeaderTimeout und IdleTimeout. +type Options struct { + StreamingBodies bool + StreamingResponses bool +} + +func envSeconds(name string, def time.Duration) time.Duration { + v := os.Getenv(name) + if v == "" { + return def + } + n, err := strconv.Atoi(v) + if err != nil || n < 1 || n > 3600 { + log.Fatalf("%s muss eine ganze Zahl zwischen 1 und 3600 sein, war %q", name, v) + } + return time.Duration(n) * time.Second +} + +// MaxBodyBytes liest die Body-Obergrenze (Standard 1 MiB, Env +// NEXARCH_HTTP_MAX_BODY_BYTES, 1 KiB bis 1 GiB). +func MaxBodyBytes() int64 { + v := os.Getenv(EnvMaxBodyBytes) + if v == "" { + return DefaultMaxBodyBytes + } + n, err := strconv.ParseInt(v, 10, 64) + if err != nil || n < 1<<10 || n > 1<<30 { + log.Fatalf("%s muss zwischen 1024 und 1073741824 liegen, war %q", EnvMaxBodyBytes, v) + } + return n +} + +// NewServer baut einen http.Server mit gehaerteten Timeouts. +func NewServer(addr string, h http.Handler, opt Options) *http.Server { + s := &http.Server{ + Addr: addr, + Handler: h, + ReadHeaderTimeout: envSeconds(EnvReadHeaderTimeout, DefaultReadHeaderTimeout), + ReadTimeout: envSeconds(EnvReadTimeout, DefaultReadTimeout), + WriteTimeout: envSeconds(EnvWriteTimeout, DefaultWriteTimeout), + IdleTimeout: envSeconds(EnvIdleTimeout, DefaultIdleTimeout), + MaxHeaderBytes: DefaultMaxHeaderBytes, + } + if opt.StreamingBodies { + s.ReadTimeout = 0 + } + if opt.StreamingResponses { + s.WriteTimeout = 0 + } + return s +} + +// ListenAndServe ersetzt http.ListenAndServe: JSON-API mit Body-Limit und +// (QA-15 F-14) apierror.Normalize aussen um jeden Handler — jeder der ueber +// diesen einen Einstiegspunkt laufenden Core-Dienste bekommt damit +// automatisch die Korrelations-ID-Vernebelung von 5xx-Antworten, ohne dass +// jede einzelne Aufrufstelle angefasst werden muss. +func ListenAndServe(addr string, h http.Handler) error { + return NewServer(addr, apierror.Normalize(LimitBody(MaxBodyBytes(), h)), Options{}).ListenAndServe() +} + +// ListenAndServeStreaming fuer Dienste mit grossen Uploads/Downloads: keine +// globale Body-Grenze und keine Lese-/Schreib-Gesamtzeit, nur Header-Timeout. +// apierror.Normalize puffert NUR Fehlerantworten (Status >= 400) — der +// Streaming-Erfolgspfad (grosse Downloads) bleibt unangetastet. +func ListenAndServeStreaming(addr string, h http.Handler) error { + return NewServer(addr, apierror.Normalize(h), Options{StreamingBodies: true, StreamingResponses: true}).ListenAndServe() +} + +// LimitBody begrenzt Request-Bodies. Ein Content-Length ueber dem Limit wird +// vor dem Handler mit 413 abgewiesen, chunked Bodies ueber MaxBytesReader. +func LimitBody(max int64, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.ContentLength > max { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusRequestEntityTooLarge) + _, _ = w.Write([]byte(`{"error":{"code":"payload_too_large","message":"Anfrage zu gross"}}`)) + return + } + if r.Body != nil && r.Body != http.NoBody { + r.Body = http.MaxBytesReader(w, r.Body, max) + } + next.ServeHTTP(w, r) + }) +} diff --git a/common/httpharden/harden_test.go b/common/httpharden/harden_test.go new file mode 100644 index 00000000..4edd2481 --- /dev/null +++ b/common/httpharden/harden_test.go @@ -0,0 +1,105 @@ +package httpharden + +import ( + "bufio" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestLimitBody_ContentLengthOverLimitIs413(t *testing.T) { + called := false + h := LimitBody(10, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { called = true })) + req := httptest.NewRequest("POST", "/", strings.NewReader(strings.Repeat("a", 11))) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusRequestEntityTooLarge || called { + t.Fatalf("code=%d called=%v", rec.Code, called) + } +} + +func TestLimitBody_ChunkedOverLimitFailsRead(t *testing.T) { + var readErr error + h := LimitBody(10, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, readErr = io.ReadAll(r.Body) + })) + req := httptest.NewRequest("POST", "/", strings.NewReader(strings.Repeat("a", 50))) + req.ContentLength = -1 + h.ServeHTTP(httptest.NewRecorder(), req) + if readErr == nil { + t.Fatal("lesen ueber dem Limit muss fehlschlagen") + } +} + +func TestLimitBody_UnderLimitPasses(t *testing.T) { + var got string + h := LimitBody(10, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + got = string(b) + })) + h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("POST", "/", strings.NewReader("abc"))) + if got != "abc" { + t.Fatalf("got %q", got) + } +} + +func TestNewServer_DefaultsAndStreaming(t *testing.T) { + s := NewServer(":0", nil, Options{}) + if s.ReadHeaderTimeout == 0 || s.ReadTimeout == 0 || s.WriteTimeout == 0 || s.IdleTimeout == 0 { + t.Fatalf("Timeouts muessen gesetzt sein: %+v", s) + } + st := NewServer(":0", nil, Options{StreamingBodies: true, StreamingResponses: true}) + if st.ReadTimeout != 0 || st.WriteTimeout != 0 || st.ReadHeaderTimeout == 0 || st.IdleTimeout == 0 { + t.Fatalf("Streaming: nur Lese-/Schreib-Gesamtzeit 0: %+v", st) + } +} + +func TestNewServer_EnvOverride(t *testing.T) { + t.Setenv(EnvReadHeaderTimeout, "3") + if got := NewServer(":0", nil, Options{}).ReadHeaderTimeout; got != 3*time.Second { + t.Fatalf("got %v", got) + } +} + +func TestMaxBodyBytes_EnvOverride(t *testing.T) { + if MaxBodyBytes() != DefaultMaxBodyBytes { + t.Fatal("Standard") + } + t.Setenv(EnvMaxBodyBytes, "2048") + if MaxBodyBytes() != 2048 { + t.Fatal("Env") + } +} + +// Slowloris: Verbindung sendet unvollstaendige Header, der Server trennt nach ReadHeaderTimeout. +func TestSlowHeadersAreClosed(t *testing.T) { + t.Setenv(EnvReadHeaderTimeout, "1") + srv := NewServer("127.0.0.1:0", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}), Options{}) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + go srv.Serve(ln) + defer srv.Close() + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if _, err := conn.Write([]byte("GET / HTTP/1.1\r\nHost: x\r\nX-Slow: a")); err != nil { + t.Fatal(err) + } + conn.SetReadDeadline(time.Now().Add(5 * time.Second)) + start := time.Now() + _, err = bufio.NewReader(conn).ReadByte() + if err == nil { + t.Log("Server antwortete (z. B. 408)") + } + if time.Since(start) > 4*time.Second { + t.Fatalf("Verbindung wurde nicht nach dem Header-Timeout beendet (%v)", time.Since(start)) + } +} diff --git a/common/loadtest/loadtest.go b/common/loadtest/loadtest.go new file mode 100644 index 00000000..a6507d1e --- /dev/null +++ b/common/loadtest/loadtest.go @@ -0,0 +1,141 @@ +// Package loadtest implementiert die für QA-07 gemeinsam genutzte +// Last- und Leistungsmessung: parallele Sessions ausführen, Latenzen +// aufzeichnen, Durchsatz und Ressourcenverbrauch bestimmen. k6/vegeta +// sind auf diesem Rechner nicht installierbar (Projektregel: keine +// zusätzlichen Toolchains/Dienste installieren) — echter Ersatz ohne +// externe Abhängigkeit: reale nebenläufige TCP-Sessions über die +// Go-Standardbibliothek, gegen die echten, in dieser Sitzung gebauten +// Protokollserver. +package loadtest + +import ( + "context" + "os" + "runtime" + "sort" + "strconv" + "strings" + "sync" + "time" +) + +// Result fasst Latenz- und Durchsatzwerte eines Lasttest-Laufs +// zusammen (Akzeptanzkriterium 2). +type Result struct { + Protocol string + Concurrency int + Sessions int + Errors int + Duration time.Duration + Latencies []time.Duration // aufsteigend sortiert +} + +// Throughput liefert Sessions pro Sekunde über die Gesamtlaufzeit. +func (r Result) Throughput() float64 { + if r.Duration <= 0 { + return 0 + } + return float64(r.Sessions) / r.Duration.Seconds() +} + +// Percentile liefert die Latenz am p-ten Perzentil (0.0–1.0). +func (r Result) Percentile(p float64) time.Duration { + if len(r.Latencies) == 0 { + return 0 + } + idx := int(p * float64(len(r.Latencies)-1)) + if idx < 0 { + idx = 0 + } + if idx >= len(r.Latencies) { + idx = len(r.Latencies) - 1 + } + return r.Latencies[idx] +} + +// Max liefert die höchste gemessene Latenz. +func (r Result) Max() time.Duration { + if len(r.Latencies) == 0 { + return 0 + } + return r.Latencies[len(r.Latencies)-1] +} + +// Run führt totalSessions Sessions über maximal concurrency parallele +// Worker aus (Akzeptanzkriterium 1: realistische Anzahl paralleler +// Sessions) und misst je Session die Latenz von sessionFunc. +func Run(ctx context.Context, protocol string, concurrency, totalSessions int, sessionFunc func(ctx context.Context) error) Result { + var mu sync.Mutex + latencies := make([]time.Duration, 0, totalSessions) + errCount := 0 + + sem := make(chan struct{}, concurrency) + var wg sync.WaitGroup + start := time.Now() + for i := 0; i < totalSessions; i++ { + sem <- struct{}{} + wg.Add(1) + go func() { + defer wg.Done() + defer func() { <-sem }() + t0 := time.Now() + err := sessionFunc(ctx) + elapsed := time.Since(t0) + mu.Lock() + latencies = append(latencies, elapsed) + if err != nil { + errCount++ + } + mu.Unlock() + }() + } + wg.Wait() + duration := time.Since(start) + + sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] }) + return Result{ + Protocol: protocol, + Concurrency: concurrency, + Sessions: totalSessions, + Errors: errCount, + Duration: duration, + Latencies: latencies, + } +} + +// ResourceSample ist eine Momentaufnahme des Ressourcenverbrauchs +// DIESES Prozesses (Akzeptanzkriterium/Pflichtprüfung 3): Heap über die +// Go-Runtime, kumulierte CPU-Zeit über /proc/self/stat (Linux, kein +// externes Werkzeug nötig). +type ResourceSample struct { + HeapAllocBytes uint64 + CPUTimeSeconds float64 +} + +func SampleResources() ResourceSample { + var m runtime.MemStats + runtime.ReadMemStats(&m) + return ResourceSample{HeapAllocBytes: m.HeapAlloc, CPUTimeSeconds: readSelfCPUTimeSeconds()} +} + +// readSelfCPUTimeSeconds liefert die kumulierte User+System-CPU-Zeit +// dieses Prozesses in Sekunden. Liefert 0, wenn /proc/self/stat nicht +// lesbar ist (z. B. auf Nicht-Linux-Systemen) — Aufrufer behandeln das +// als "keine CPU-Messung verfügbar", kein harter Fehler. +func readSelfCPUTimeSeconds() float64 { + data, err := os.ReadFile("/proc/self/stat") + if err != nil { + return 0 + } + fields := strings.Fields(string(data)) + if len(fields) < 15 { + return 0 + } + utime, err1 := strconv.ParseFloat(fields[13], 64) + stime, err2 := strconv.ParseFloat(fields[14], 64) + if err1 != nil || err2 != nil { + return 0 + } + const clockTicksPerSecond = 100 // Linux-Standard-USER_HZ + return (utime + stime) / clockTicksPerSecond +} diff --git a/common/migrate/migrate.go b/common/migrate/migrate.go new file mode 100644 index 00000000..a4257be2 --- /dev/null +++ b/common/migrate/migrate.go @@ -0,0 +1,221 @@ +// Package migrate implements versioned, revertible SQL migrations (no ORM) — a +// standalone copy of the Core/DMS pattern, since Connect is its own Go module +// and cannot import other modules' internal/ packages. +package migrate + +import ( + "context" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Migration ist eine einzelne versionierte Migrationsdatei. +type Migration struct { + Version string // Dateiname ohne .up.sql/.down.sql, z.B. "0001_documents" + UpSQL string + DownSQL string +} + +// Load liest alle *.up.sql/*.down.sql-Paare aus dir, sortiert nach +// Dateiname (Akzeptanzkriterium: Migrationen sind versioniert). +func Load(dir string) ([]Migration, error) { + entries, err := os.ReadDir(dir) + if err != nil { + return nil, fmt.Errorf("migrationsverzeichnis %q lesen: %w", dir, err) + } + var versions []string + for _, e := range entries { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".up.sql") { + continue + } + versions = append(versions, strings.TrimSuffix(e.Name(), ".up.sql")) + } + sort.Strings(versions) + + migrations := make([]Migration, 0, len(versions)) + for _, v := range versions { + up, err := os.ReadFile(filepath.Join(dir, v+".up.sql")) + if err != nil { + return nil, fmt.Errorf("migration %q: up.sql lesen: %w", v, err) + } + down, err := os.ReadFile(filepath.Join(dir, v+".down.sql")) + if err != nil { + return nil, fmt.Errorf("migration %q: down.sql lesen (jede Migration braucht ein Rollback): %w", v, err) + } + migrations = append(migrations, Migration{Version: v, UpSQL: string(up), DownSQL: string(down)}) + } + return migrations, nil +} + +// TrackingTable ist modulspezifisch benannt, damit sich Module in +// derselben Datenbank/demselben Schema nicht gegenseitig den +// Migrationsfortschritt ueberschreiben (FDN-13). +const TrackingTable = "connect_schema_migrations" + +const legacyTrackingTable = "schema_migrations" + +// ensureTrackingTable legt TrackingTable an. Beim erstmaligen Anlegen wird +// ein vorhandener Bestand aus der frueheren, unqualifizierten +// schema_migrations uebernommen (nur Versionen, die dieses Modul kennt, und +// nur wenn dessen Basisversion dort steht — eine schema_migrations eines +// anderen Moduls oder eines anderen Schemas wird weder gelesen noch +// veraendert). Die alte Tabelle +// bleibt unangetastet. +func ensureTrackingTable(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) error { + tx, err := pool.Begin(ctx) + if err != nil { + return fmt.Errorf("%s anlegen: %w", TrackingTable, err) + } + defer func() { _ = tx.Rollback(ctx) }() + + // Serialisiert gleichzeitige Erstanlage/Uebernahme (mehrere App-Instanzen). + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext('connect_schema_migrations'))`); err != nil { + return fmt.Errorf("%s anlegen: %w", TrackingTable, err) + } + + var existed bool + if err := tx.QueryRow(ctx, `SELECT to_regclass(current_schema() || '.' || $1::text) IS NOT NULL`, TrackingTable).Scan(&existed); err != nil { + return fmt.Errorf("%s anlegen: %w", TrackingTable, err) + } + if existed { + return tx.Commit(ctx) + } + + if _, err := tx.Exec(ctx, ` + CREATE TABLE `+TrackingTable+` ( + version TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + ) + `); err != nil { + return fmt.Errorf("%s anlegen: %w", TrackingTable, err) + } + + if len(migrations) > 0 { + var legacy bool + if err := tx.QueryRow(ctx, `SELECT to_regclass(current_schema() || '.' || $1::text) IS NOT NULL`, legacyTrackingTable).Scan(&legacy); err != nil { + return fmt.Errorf("%s pruefen: %w", legacyTrackingTable, err) + } + if legacy { + versions := make([]string, len(migrations)) + for i, m := range migrations { + versions[i] = m.Version + } + var hasBase bool + if err := tx.QueryRow(ctx, + `SELECT EXISTS (SELECT 1 FROM `+legacyTrackingTable+` WHERE version = $1)`, versions[0], + ).Scan(&hasBase); err != nil { + return fmt.Errorf("%s lesen: %w", legacyTrackingTable, err) + } + if hasBase { + if _, err := tx.Exec(ctx, ` + INSERT INTO `+TrackingTable+` (version, applied_at) + SELECT version, applied_at FROM `+legacyTrackingTable+` WHERE version = ANY($1) + `, versions); err != nil { + return fmt.Errorf("bestand aus %s uebernehmen: %w", legacyTrackingTable, err) + } + } + } + } + return tx.Commit(ctx) +} + +func appliedVersions(ctx context.Context, pool *pgxpool.Pool) (map[string]bool, error) { + rows, err := pool.Query(ctx, `SELECT version FROM `+TrackingTable) + if err != nil { + return nil, fmt.Errorf("angewendete migrationen lesen: %w", err) + } + defer rows.Close() + + applied := map[string]bool{} + for rows.Next() { + var v string + if err := rows.Scan(&v); err != nil { + return nil, fmt.Errorf("migrationsversion lesen: %w", err) + } + applied[v] = true + } + return applied, rows.Err() +} + +// Up wendet alle noch nicht angewendeten Migrationen in Reihenfolge an +// (Akzeptanzkriterium 2: vorwaerts ausfuehrbar) — bereits angewendete +// werden uebersprungen, damit Up auf einer leeren UND auf einer bestehenden +// DB funktioniert (Pruefung 1). +func Up(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (applied []string, err error) { + if err := ensureTrackingTable(ctx, pool, migrations); err != nil { + return nil, err + } + already, err := appliedVersions(ctx, pool) + if err != nil { + return nil, err + } + + for _, m := range migrations { + if already[m.Version] { + continue + } + tx, err := pool.Begin(ctx) + if err != nil { + return applied, fmt.Errorf("transaktion fuer %q starten: %w", m.Version, err) + } + if _, err := tx.Exec(ctx, m.UpSQL); err != nil { + _ = tx.Rollback(ctx) + return applied, fmt.Errorf("migration %q anwenden: %w", m.Version, err) + } + if _, err := tx.Exec(ctx, `INSERT INTO `+TrackingTable+` (version) VALUES ($1)`, m.Version); err != nil { + _ = tx.Rollback(ctx) + return applied, fmt.Errorf("migration %q als angewendet markieren: %w", m.Version, err) + } + if err := tx.Commit(ctx); err != nil { + return applied, fmt.Errorf("migration %q committen: %w", m.Version, err) + } + applied = append(applied, m.Version) + } + return applied, nil +} + +// DownOne macht die zuletzt angewendete Migration rueckgaengig +// (Akzeptanzkriterium 2: rueckwaerts ausfuehrbar) und liefert deren Version, +// oder "" falls keine Migration angewendet war. +func DownOne(ctx context.Context, pool *pgxpool.Pool, migrations []Migration) (version string, err error) { + if err := ensureTrackingTable(ctx, pool, migrations); err != nil { + return "", err + } + already, err := appliedVersions(ctx, pool) + if err != nil { + return "", err + } + + var last *Migration + for i := len(migrations) - 1; i >= 0; i-- { + if already[migrations[i].Version] { + last = &migrations[i] + break + } + } + if last == nil { + return "", nil + } + + tx, err := pool.Begin(ctx) + if err != nil { + return "", fmt.Errorf("transaktion fuer rollback von %q starten: %w", last.Version, err) + } + if _, err := tx.Exec(ctx, last.DownSQL); err != nil { + _ = tx.Rollback(ctx) + return "", fmt.Errorf("migration %q zurueckrollen: %w", last.Version, err) + } + if _, err := tx.Exec(ctx, `DELETE FROM `+TrackingTable+` WHERE version = $1`, last.Version); err != nil { + _ = tx.Rollback(ctx) + return "", fmt.Errorf("migration %q aus %s entfernen: %w", last.Version, TrackingTable, err) + } + if err := tx.Commit(ctx); err != nil { + return "", fmt.Errorf("rollback von %q committen: %w", last.Version, err) + } + return last.Version, nil +} diff --git a/common/policyclient/policyclient.go b/common/policyclient/policyclient.go new file mode 100644 index 00000000..16c67f95 --- /dev/null +++ b/common/policyclient/policyclient.go @@ -0,0 +1,52 @@ +// Package policyclient fragt Core (RBAC-06, POST /authorize) nach der +// Rechteentscheidung; Fehler und Ausfall fuehren immer zur Ablehnung. +package policyclient + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" +) + +type Client struct { + BaseURL string + ServiceToken string + HTTP *http.Client +} + +func New(baseURL, serviceToken string) *Client { + return &Client{BaseURL: strings.TrimRight(baseURL, "/"), ServiceToken: serviceToken, HTTP: &http.Client{Timeout: 5 * time.Second}} +} + +func (c *Client) Allowed(ctx context.Context, tenant, role, permission string) (bool, error) { + body := map[string]string{"role": role, "permission": permission} + if tenant != "" { + body["tenant"] = tenant + } + payload, _ := json.Marshal(body) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.BaseURL+"/authorize", bytes.NewReader(payload)) + if err != nil { + return false, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Service-Token", c.ServiceToken) + resp, err := c.HTTP.Do(req) + if err != nil { + return false, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return false, fmt.Errorf("policy: status %d", resp.StatusCode) + } + var out struct { + Allowed *bool `json:"allowed"` + } + if err := json.NewDecoder(http.MaxBytesReader(nil, resp.Body, 1<<16)).Decode(&out); err != nil || out.Allowed == nil { + return false, fmt.Errorf("policy: antwort ohne entscheidung") + } + return *out.Allowed, nil +} diff --git a/common/policyclient/policyclient_test.go b/common/policyclient/policyclient_test.go new file mode 100644 index 00000000..3ac23bed --- /dev/null +++ b/common/policyclient/policyclient_test.go @@ -0,0 +1,80 @@ +package policyclient + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestAllowed_TenantScopedRequest_TrueOnAllow(t *testing.T) { + var gotBody map[string]string + var gotToken string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotToken = r.Header.Get("X-Service-Token") + if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil { + t.Fatalf("body dekodieren: %v", err) + } + json.NewEncoder(w).Encode(map[string]bool{"allowed": true}) + })) + defer srv.Close() + + c := New(srv.URL, "svc-token-123") + allowed, err := c.Allowed(context.Background(), "acme", "editor", "connector.write") + if err != nil { + t.Fatalf("Allowed: %v", err) + } + if !allowed { + t.Fatal("erwartet allowed=true") + } + if gotToken != "svc-token-123" { + t.Fatalf("Service-Token nicht gesendet, erhalten: %q", gotToken) + } + if gotBody["tenant"] != "acme" { + t.Fatalf("Tenant fehlt/falsch in Anfrage: %q", gotBody["tenant"]) + } +} + +func TestAllowed_DeniedResponse_False(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(map[string]bool{"allowed": false}) + })) + defer srv.Close() + + c := New(srv.URL, "svc-token-123") + allowed, err := c.Allowed(context.Background(), "acme", "viewer", "connector.write") + if err != nil { + t.Fatalf("Allowed: %v", err) + } + if allowed { + t.Fatal("erwartet allowed=false") + } +} + +func TestAllowed_ServiceFailure_DeniesInsteadOfSilentAllow(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + c := New(srv.URL, "svc-token-123") + allowed, err := c.Allowed(context.Background(), "acme", "editor", "connector.write") + if err == nil { + t.Fatal("erwarteter Fehler bei Status 500 blieb aus") + } + if allowed { + t.Fatal("Ausfall darf nie zu allowed=true fuehren") + } +} + +func TestAllowed_TransportFailure_DeniesInsteadOfSilentAllow(t *testing.T) { + c := New("http://127.0.0.1:1", "svc-token-123") + allowed, err := c.Allowed(context.Background(), "acme", "editor", "connector.write") + if err == nil { + t.Fatal("erwarteter Transport-Fehler blieb aus") + } + if allowed { + t.Fatal("Ausfall darf nie zu allowed=true fuehren") + } +} diff --git a/common/svcauth/svcauth.go b/common/svcauth/svcauth.go new file mode 100644 index 00000000..4f803353 --- /dev/null +++ b/common/svcauth/svcauth.go @@ -0,0 +1,18 @@ +// Package svcauth vergleicht Service-Token in konstanter Zeit (QA-15 F-23). +package svcauth + +import ( + "crypto/sha256" + "crypto/subtle" +) + +// Equal vergleicht die SHA-256-Werte beider Eingaben in konstanter Zeit; anders +// als ein direkter ConstantTimeCompare verraet der Vergleich nicht die Laenge +// des erwarteten Tokens. Ein leeres erwartetes Token stimmt nie. +func Equal(got, want string) bool { + if want == "" { + return false + } + g, w := sha256.Sum256([]byte(got)), sha256.Sum256([]byte(want)) + return subtle.ConstantTimeCompare(g[:], w[:]) == 1 +} diff --git a/common/svcauth/svcauth_test.go b/common/svcauth/svcauth_test.go new file mode 100644 index 00000000..c06848c9 --- /dev/null +++ b/common/svcauth/svcauth_test.go @@ -0,0 +1,23 @@ +package svcauth + +import "testing" + +func TestEqual(t *testing.T) { + cases := []struct { + got, want string + ok bool + }{ + {"geheim", "geheim", true}, + {"geheim ", "geheim", false}, + {"", "geheim", false}, + {"geheim", "", false}, + {"", "", false}, + {"g", "geheim", false}, + {"geheimgeheim", "geheim", false}, + } + for _, c := range cases { + if got := Equal(c.got, c.want); got != c.ok { + t.Errorf("Equal(%q,%q) = %v, erwartet %v", c.got, c.want, got, c.ok) + } + } +} diff --git a/common/tenantregistryclient/tenantregistryclient.go b/common/tenantregistryclient/tenantregistryclient.go new file mode 100644 index 00000000..e768c264 --- /dev/null +++ b/common/tenantregistryclient/tenantregistryclient.go @@ -0,0 +1,47 @@ +// Package tenantregistryclient ist ein schlanker HTTP-Client fuer Core +// TEN-01 (Tenant-Registry); AI ist ein eigenes Go-Modul und kann Cores +// Pakete nicht importieren. Gleiches Muster wie archive/internal/ +// tenantregistryclient und dms/internal/tenantregistryclient (RET-12). +package tenantregistryclient + +import ( + "context" + "fmt" + "net/http" + "net/url" + "time" +) + +type Client struct { + BaseURL string + ServiceToken string + HTTPClient *http.Client +} + +func New(baseURL, serviceToken string) *Client { + return &Client{BaseURL: baseURL, ServiceToken: serviceToken, HTTPClient: &http.Client{Timeout: 10 * time.Second}} +} + +// VerifyTenant fragt Core, ob tenantSlug als Tenant gefuehrt wird. Transport- +// und Serverfehler kommen als Fehler zurueck, nie als "gueltig": der Aufrufer +// muss fail-closed reagieren. +func (c *Client) VerifyTenant(ctx context.Context, tenantSlug string) (bool, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.BaseURL+"/tenants/"+url.PathEscape(tenantSlug), nil) + if err != nil { + return false, fmt.Errorf("tenantregistryclient: request bauen: %w", err) + } + req.Header.Set("X-Service-Token", c.ServiceToken) + resp, err := c.HTTPClient.Do(req) + if err != nil { + return false, fmt.Errorf("tenantregistryclient: aufruf fehlgeschlagen: %w", err) + } + defer func() { _ = resp.Body.Close() }() + switch resp.StatusCode { + case http.StatusOK: + return true, nil + case http.StatusNotFound: + return false, nil + default: + return false, fmt.Errorf("tenantregistryclient: unerwarteter status %d", resp.StatusCode) + } +} diff --git a/common/tenantregistryclient/tenantregistryclient_test.go b/common/tenantregistryclient/tenantregistryclient_test.go new file mode 100644 index 00000000..41f9f355 --- /dev/null +++ b/common/tenantregistryclient/tenantregistryclient_test.go @@ -0,0 +1,59 @@ +package tenantregistryclient + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" +) + +func TestVerifyTenant_ExistingTenant(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/tenants/mandant-a" { + t.Errorf("pfad = %q, want /tenants/mandant-a", r.URL.Path) + } + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + c := New(server.URL, "test-token") + ok, err := c.VerifyTenant(context.Background(), "mandant-a") + if err != nil { + t.Fatalf("verifytenant: %v", err) + } + if !ok { + t.Fatal("erwartet gueltigen tenant, habe false") + } +} + +func TestVerifyTenant_UnknownTenant(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNotFound) + })) + defer server.Close() + + c := New(server.URL, "test-token") + ok, err := c.VerifyTenant(context.Background(), "unbekannt") + if err != nil { + t.Fatalf("verifytenant: %v", err) + } + if ok { + t.Fatal("erwartet ungueltigen tenant, habe true") + } +} + +func TestVerifyTenant_ServerErrorIsFailClosed(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + c := New(server.URL, "test-token") + ok, err := c.VerifyTenant(context.Background(), "mandant-a") + if err == nil { + t.Fatal("erwartet fehler bei 500-antwort der tenant-registry") + } + if ok { + t.Fatal("erwartet false bei fehler (fail-closed)") + } +} diff --git a/common/testdb/testdb.go b/common/testdb/testdb.go new file mode 100644 index 00000000..c5db1b1a --- /dev/null +++ b/common/testdb/testdb.go @@ -0,0 +1,62 @@ +// Package testdb liefert Integrationstests einen Pool auf ein eigenes, +// frisch angelegtes Schema (search_path = Schema, public). Tabellen, die ein +// Test per CREATE TABLE anlegt, entstehen damit nur in diesem Schema; Pakete, +// Laeufe und parallele Checkouts teilen sich keine Tabellen mehr. +package testdb + +import ( + "context" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +var seq atomic.Int64 + +// Pool legt ein Schema an und liefert einen darauf ausgerichteten Pool; +// beides wird per t.Cleanup wieder entfernt. +func Pool(t testing.TB, dsn string) *pgxpool.Pool { + t.Helper() + pool, _ := PoolWithSchema(t, dsn, nil) + return pool +} + +// PoolWithSchema wie Pool, zusaetzlich mit dem Schemanamen und der +// Moeglichkeit, die Verbindungskonfiguration anzupassen. +func PoolWithSchema(t testing.TB, dsn string, adjust func(*pgxpool.Config)) (*pgxpool.Pool, string) { + t.Helper() + ctx := context.Background() + admin, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatalf("testdb: verbindung: %v", err) + } + schema := fmt.Sprintf("connecttest_%d_%d", time.Now().UnixNano(), seq.Add(1)) + if _, err := admin.Exec(ctx, "CREATE SCHEMA "+schema); err != nil { + admin.Close() + t.Fatalf("testdb: schema anlegen: %v", err) + } + + cfg, err := pgxpool.ParseConfig(dsn) + if err != nil { + admin.Close() + t.Fatalf("testdb: dsn: %v", err) + } + cfg.ConnConfig.RuntimeParams["search_path"] = schema + ",public" + if adjust != nil { + adjust(cfg) + } + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + admin.Close() + t.Fatalf("testdb: pool: %v", err) + } + t.Cleanup(func() { + pool.Close() + _, _ = admin.Exec(context.Background(), "DROP SCHEMA "+schema+" CASCADE") + admin.Close() + }) + return pool, schema +} -- 2.47.3