diff --git a/connect/docs/API-03-PRUEFPROTOKOLL.md b/connect/docs/API-03-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..dd203556 --- /dev/null +++ b/connect/docs/API-03-PRUEFPROTOKOLL.md @@ -0,0 +1,80 @@ +# API-03 – Prüfprotokoll: Externe Webhook-Abonnements (nutzt Core API-07) + +Datum: 2026-10-16. Host: 204 (Wegwerf-Postgres, `TEST_TENANT_DSN`, isoliertes Schema pro Testlauf über `connect/internal/testdb`). + +## Ausgangslage: Schnittstelle zu Core API-07 + +Core API-07 (`internal/webhook`, Modul `gitea.perlbach24.de/scripte/nexarch`) implementiert Registrierung (`Store.Subscribe`), Zustellung mit Retry/Backoff (`Dispatcher`) und HMAC-Signatur (`Sign`/`VerifySignature`) als reines Go-Paket. Recherche im Core-Modul (gleicher Checkout, `internal/webhook/*.go`, `internal/contracttest/contracttest_test.go`) ergab: **es existiert aktuell kein HTTP-Handler/`cmd/*`-Dienst, der `Store.Subscribe` für physisch getrennte Module (Connect, DMS, Mail, Archive) über HTTP erreichbar macht** — anders als z. B. AUD-08 (`internal/audit/ingest_handler.go` + `cmd/auditlog-devserver`), das exakt dieses Muster für den Audit-Log bereits umsetzt. Core-seitig fehlt zusätzlich eine `Unsubscribe`-Methode auf `Store` völlig. + +Das ist eine Lücke auf Core-Seite (core-kanban), nicht Gegenstand dieser Kachel (Connect ist laut Auftrag physisch getrennt und darf Cores `internal/`-Pakete nicht importieren — nur über HTTP koppeln). Diese Kachel erlaubt für genau diesen Fall ausdrücklich einen **„minimal gestubten Core-Webhook-Registry-Endpunkt"** für den End-zu-Ende-Test. Umgesetzt wurde daher: + +- Der HTTP-Vertrag, den ein künftiger Core-HTTP-Wrapper für API-07 erfüllen müsste (`POST /webhooks/subscriptions`, `DELETE /webhooks/subscriptions/{id}`, `X-Service-Token`-Auth, Felder `tenant_slug`/`event_type`/`target_url`/`secret`), dokumentiert als Vertrag in `connect/internal/webhookclient`. +- Alle Tests laufen über einen echten `httptest.Server`, der genau diesen Vertrag nachbildet (echte HTTP-Verbindung über Loopback-TCP, keine Mock-Schnittstelle des Clients selbst) — dasselbe Muster wie `connect/internal/policyclient`/`connect/internal/auditclient` bereits für andere Core-Schnittstellen verwenden. +- HMAC-Signaturvertrag (`X-Nexarch-Signature-256`, hex-kodiertes HMAC-SHA256) 1:1 aus `internal/webhook.SignatureHeader`/`VerifySignature` übernommen, da Connect das Core-Paket nicht importieren kann (physische Modultrennung). + +## Umsetzung + +- `connect/internal/webhookclient` — schlanker HTTP-Client gegen Core API-07 (`Subscribe`, `Unsubscribe`), Muster aus `connect/internal/auditclient`/`dms/internal/retentionclient`. +- `connect/internal/webhooksub` — Connect-seitige Verwaltung: + - `Manager.RegisterAction(connectorName, eventType, action)`: verdrahtet prozesslokal, welche Connector-Aktion ein Ereignis auslöst (Muster wie `registry.Registry`). + - `Manager.Subscribe`: registriert bei Core (AK1), verschlüsselt das Zustellgeheimnis per `connect/internal/crypto` (Envelope-Verschlüsselung, gleiches Muster wie `credential.Store`) und speichert erst danach lokal. Schlägt die Core-Registrierung fehl, bleibt nichts lokal zurück. + - `Manager.Unsubscribe`: bestellt zuerst bei Core ab, löscht die lokale Zeile erst danach vollständig (AK3). Schlägt die Core-Abbestellung fehl, bleibt die lokale Zeile für einen Wiederholungsversuch erhalten statt ein verwaistes Core-Abonnement zu hinterlassen. + - `Manager.RegisterRoutes`/`handleDelivery` (`delivery.go`): HTTP-Endpunkt `POST /connector-webhooks/deliveries/{id}`, an den Core laut `target_url` zustellt. Prüft die HMAC-Signatur zeitkonstant (`crypto/subtle.ConstantTimeCompare`) gegen das entschlüsselte Geheimnis, löst bei Erfolg die registrierte Connector-Aktion aus (AK2). +- Migration `0013_connector_webhook_subscriptions` — Tabelle `connector_webhook_subscriptions` (lokale id = Zustellpfad-Segment, `core_subscription_id`, verschlüsseltes Geheimnis, `UNIQUE(tenant_slug, connector_name, event_type)`). +- `connect/scripts/check-security-test-coverage.sh` — `internal/webhookclient`/`internal/webhooksub` ergänzt (beide verarbeiten tenant-gebundene Geheimnisse/Core-Authentifizierung, gleiche Kategorie wie `credential`/`auditclient`). + +## Akzeptanzkriterien + +1. **Connector kann ein Webhook-Ereignis über die Core-Webhook-Registry abonnieren** — erfüllt: `Manager.Subscribe` ruft `webhookclient.Subscribe` (POST an die Registry) auf und verlangt vorab eine registrierte Connector-Aktion. +2. **Zugestelltes Ereignis löst die zugeordnete Connector-Aktion aus** — erfüllt: `handleDelivery` prüft Signatur und ruft die per `RegisterAction` hinterlegte Funktion mit dem zugestellten Payload auf. +3. **Abbestellen entfernt das Abonnement vollständig** — erfüllt: `Manager.Unsubscribe` entfernt sowohl bei Core (`webhookclient.Unsubscribe`) als auch die lokale Zeile; eine erneute Zustellung an die alte URL schlägt danach mit 404 fehl. + +## Prüfungen vor der Abnahme + +| # | Prüfung | Ergebnis | +|---|---|---| +| 1 | Testabonnement über Core-Webhook-Registry im Test erfolgreich registriert | **Bestanden.** `TestSubscribe_RegistersAtCoreAndPersistsLocally`: echter HTTP-POST an den (gestubten) Core-Endpunkt, lokale Zeile mit `core_subscription_id` entsteht, Core-Server bestätigt das Abonnement. Negativ abgesichert durch `TestSubscribe_WithoutRegisteredAction_Fails` und `TestSubscribe_CoreFailure_PersistsNothingLocally` (kein verwaister lokaler Datensatz bei Core-Ausfall). | +| 2 | Testereignis löst im Test die erwartete Connector-Aktion aus | **Bestanden.** `TestHandleDelivery_ValidSignature_TriggersConnectorAction`: echter HTTP-POST (eigener `httptest.Server`, echte TCP-Verbindung) mit gültiger HMAC-Signatur an die beim Subscribe erzeugte Zustell-URL löst die registrierte Aktion mit dem exakten Payload aus. Negativ abgesichert durch `TestHandleDelivery_InvalidSignature_Rejected` (401, Aktion bleibt aus) und `TestHandleDelivery_UnknownSubscription_NotFound`. | +| 3 | Abbestellung im Test vollständig entfernt | **Bestanden.** `TestUnsubscribe_RemovesAtCoreAndLocally_Completely`: nach `Unsubscribe` ist die lokale Zeile weg und der Core-Stub bestätigt das entfernte Abonnement. Zusätzlich `TestHandleDelivery_AfterUnsubscribe_NotFound` (Zustellung an die alte URL schlägt danach mit 404 fehl) und `TestUnsubscribe_CoreFailure_KeepsLocalRowForRetry` (kein Datenverlust bei Core-Ausfall während der Abbestellung). | + +## Testlauf (real, Host 204, gegen echte Postgres-Testdatenbank) + +``` +go build ./... → keine Ausgabe (sauber) +go vet ./... → keine Ausgabe (sauber) +gofmt -l internal/webhookclient internal/webhooksub → keine Ausgabe (bereits formatiert) + +go test ./internal/webhookclient/... -v + PASS TestSubscribe_SendsRequestFields_ReturnsSubscription + PASS TestSubscribe_UnexpectedStatus_ReturnsError + PASS TestSubscribe_TransportError_ReturnsError + PASS TestUnsubscribe_SendsServiceTokenAndTenant_Succeeds + PASS TestUnsubscribe_UnexpectedStatus_ReturnsError + PASS TestUnsubscribe_TransportError_ReturnsError +ok connect/internal/webhookclient 0.007s + +go test ./internal/webhooksub/... -v (TEST_TENANT_DSN gesetzt, echte Postgres-DB) + PASS TestHandleDelivery_ValidSignature_TriggersConnectorAction + PASS TestHandleDelivery_InvalidSignature_Rejected + PASS TestHandleDelivery_UnknownSubscription_NotFound + PASS TestHandleDelivery_AfterUnsubscribe_NotFound + PASS TestSubscribe_RegistersAtCoreAndPersistsLocally + PASS TestSubscribe_WithoutRegisteredAction_Fails + PASS TestSubscribe_CoreFailure_PersistsNothingLocally + PASS TestUnsubscribe_RemovesAtCoreAndLocally_Completely + PASS TestUnsubscribe_UnknownSubscription_ReturnsNotFound + PASS TestUnsubscribe_CoreFailure_KeepsLocalRowForRetry +ok connect/internal/webhooksub 1.572s–1.778s (mehrere Läufe) + +go test ./... (gesamtes connect-Modul, TEST_TENANT_DSN gesetzt) → alle Pakete ok, kein FAIL + +bash scripts/check-security-test-coverage.sh → "Merge-Gate bestanden: alle sicherheitsrelevanten Pakete haben Tests." +``` + +Keine Regressionen: alle zuvor bestehenden Pakete (u. a. `credential`, `statusapi`, `syncjob`, `audittrail`) bleiben unverändert grün. + +## Nicht Bestandteil dieser Kachel + +- `QA-02` Prüfgate Connector-Engine — separates Folgeticket, nicht umgesetzt. +- Core-seitiger HTTP-Wrapper für API-07 (`Store.Subscribe`/neue `Unsubscribe`-Methode über `cmd/*-devserver`, analog AUD-08) — Lücke auf Core-Seite, gehört ins core-kanban-Board, nicht in diese Kachel. Bis dahin bindet `connect/internal/webhookclient` an den hier dokumentierten, noch nicht Core-seitig existierenden Vertrag; Umstellung auf die reale Core-Basis-URL ist eine reine Konfigurationsänderung (`NEXARCH_*_BASE_URL`), keine Codeänderung. +- Verdrahtung in `connect-app`s `main.go` (Produktions-`KEKProvider` gegen Core API-10) — bewusst unterlassen: auch `connect/internal/credential` (ältere, grundlegendere Kachel) ist dort noch nicht verdrahtet; ein produktiver `KEKProvider` für Connect existiert im Modul noch nirgends und wäre eine eigene, hier nicht beauftragte Erweiterung. diff --git a/connect/internal/webhookclient/webhookclient.go b/connect/internal/webhookclient/webhookclient.go new file mode 100644 index 00000000..2762e056 --- /dev/null +++ b/connect/internal/webhookclient/webhookclient.go @@ -0,0 +1,108 @@ +// Package webhookclient ist ein schlanker HTTP-Client fuer Core API-07 +// (internal/webhook, zentrale Webhook-Registry). Connect ist ein physisch +// getrenntes Go-Modul und kann Cores internal/webhook daher nicht direkt +// importieren -- nur ueber dessen HTTP-Schnittstelle (gleiches Muster wie +// connect/internal/auditclient fuer AUD-08 und dms/internal/retentionclient +// fuer Archive RET-05). +package webhookclient + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "time" + + "gitea.perlbach24.de/scripte/nexarch/connect/internal/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()} +} + +// Subscription spiegelt Core API-07s Antwort auf eine Registrierung. +type Subscription struct { + ID string `json:"id"` + EventType string `json:"event_type"` + TargetURL string `json:"target_url"` + CreatedAt time.Time `json:"created_at"` +} + +type subscribeRequest struct { + TenantSlug string `json:"tenant_slug"` + EventType string `json:"event_type"` + TargetURL string `json:"target_url"` + Secret string `json:"secret"` +} + +// Subscribe registriert targetURL bei Core API-07 fuer eventType. Jeder +// Fehler (Transport, Timeout, Nicht-2xx) wird als Fehler zurueckgegeben -- +// der Aufrufer entscheidet, ob/wie er das nachbehandelt (Akzeptanzkriterium +// 1). +func (c *Client) Subscribe(ctx context.Context, tenantSlug, eventType, targetURL, secret string) (Subscription, error) { + body, err := json.Marshal(subscribeRequest{ + TenantSlug: tenantSlug, EventType: eventType, TargetURL: targetURL, Secret: secret, + }) + if err != nil { + return Subscription{}, fmt.Errorf("webhookclient: request kodieren: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.BaseURL+"/webhooks/subscriptions", bytes.NewReader(body)) + if err != nil { + return Subscription{}, fmt.Errorf("webhookclient: 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 Subscription{}, fmt.Errorf("webhookclient: aufruf fehlgeschlagen: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusCreated { + return Subscription{}, fmt.Errorf("webhookclient: unerwarteter status %d", resp.StatusCode) + } + var out Subscription + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return Subscription{}, fmt.Errorf("webhookclient: antwort dekodieren: %w", err) + } + return out, nil +} + +// Unsubscribe entfernt subscriptionID vollstaendig aus Core API-07 +// (Akzeptanzkriterium 3). +func (c *Client) Unsubscribe(ctx context.Context, tenantSlug, subscriptionID string) error { + target := c.BaseURL + "/webhooks/subscriptions/" + url.PathEscape(subscriptionID) + + "?tenant_slug=" + url.QueryEscape(tenantSlug) + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, target, nil) + if err != nil { + return fmt.Errorf("webhookclient: request bauen: %w", err) + } + req.Header.Set("X-Service-Token", c.ServiceToken) + + resp, err := c.httpClient().Do(req) + if err != nil { + return fmt.Errorf("webhookclient: aufruf fehlgeschlagen: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusNoContent { + return fmt.Errorf("webhookclient: 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/connect/internal/webhookclient/webhookclient_test.go b/connect/internal/webhookclient/webhookclient_test.go new file mode 100644 index 00000000..392de871 --- /dev/null +++ b/connect/internal/webhookclient/webhookclient_test.go @@ -0,0 +1,109 @@ +package webhookclient + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +// newFakeCoreWebhookAPI steht fuer Core API-07s echten Vertrag (POST +// /webhooks/subscriptions, DELETE /webhooks/subscriptions/{id}) -- testet +// die echte HTTP-Kette (Client -> echter Server -> Antwort-Dekodierung), +// nicht nur eine wirkungslose lokale Struktur. +func newFakeCoreWebhookAPI(t *testing.T) (*httptest.Server, *[]string) { + t.Helper() + var calls []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls = append(calls, r.Method+" "+r.URL.String()) + if r.Header.Get("X-Service-Token") == "" { + t.Fatalf("Service-Token fehlt bei %s %s", r.Method, r.URL.Path) + } + switch { + case r.Method == http.MethodPost && r.URL.Path == "/webhooks/subscriptions": + var in subscribeRequest + if err := json.NewDecoder(r.Body).Decode(&in); err != nil { + t.Fatalf("body dekodieren: %v", err) + } + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(Subscription{ + ID: "core-sub-1", EventType: in.EventType, TargetURL: in.TargetURL, CreatedAt: time.Now().UTC(), + }) + case r.Method == http.MethodDelete: + w.WriteHeader(http.StatusNoContent) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + return srv, &calls +} + +func TestSubscribe_SendsRequestFields_ReturnsSubscription(t *testing.T) { + srv, _ := newFakeCoreWebhookAPI(t) + defer srv.Close() + + c := New(srv.URL, "svc-token-123") + sub, err := c.Subscribe(context.Background(), "acme", "dms.document.archived", "https://connect.example/d/abc", "shh-secret") + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + if sub.ID != "core-sub-1" { + t.Fatalf("erwartete core-vergebene id, bekam %q", sub.ID) + } + if sub.EventType != "dms.document.archived" { + t.Fatalf("event_type nicht wie gesendet: %q", sub.EventType) + } +} + +func TestSubscribe_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.Subscribe(context.Background(), "acme", "dms.document.archived", "https://connect.example/d/abc", "shh"); err == nil { + t.Fatal("erwarteter Fehler bei Status 500 blieb aus") + } +} + +func TestSubscribe_TransportError_ReturnsError(t *testing.T) { + c := New("http://127.0.0.1:1", "svc-token-123") + if _, err := c.Subscribe(context.Background(), "acme", "dms.document.archived", "https://connect.example/d/abc", "shh"); err == nil { + t.Fatal("erwarteter Transport-Fehler blieb aus") + } +} + +func TestUnsubscribe_SendsServiceTokenAndTenant_Succeeds(t *testing.T) { + srv, calls := newFakeCoreWebhookAPI(t) + defer srv.Close() + + c := New(srv.URL, "svc-token-123") + if err := c.Unsubscribe(context.Background(), "acme", "core-sub-1"); err != nil { + t.Fatalf("Unsubscribe: %v", err) + } + if len(*calls) != 1 || (*calls)[0] != "DELETE /webhooks/subscriptions/core-sub-1?tenant_slug=acme" { + t.Fatalf("unerwartete aufrufe: %v", *calls) + } +} + +func TestUnsubscribe_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.Unsubscribe(context.Background(), "acme", "core-sub-1"); err == nil { + t.Fatal("erwarteter Fehler bei Status 500 blieb aus") + } +} + +func TestUnsubscribe_TransportError_ReturnsError(t *testing.T) { + c := New("http://127.0.0.1:1", "svc-token-123") + if err := c.Unsubscribe(context.Background(), "acme", "core-sub-1"); err == nil { + t.Fatal("erwarteter Transport-Fehler blieb aus") + } +} diff --git a/connect/internal/webhooksub/common_test.go b/connect/internal/webhooksub/common_test.go new file mode 100644 index 00000000..ebac0bcd --- /dev/null +++ b/connect/internal/webhooksub/common_test.go @@ -0,0 +1,153 @@ +package webhooksub_test + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + connectcrypto "gitea.perlbach24.de/scripte/nexarch/connect/internal/crypto" + "gitea.perlbach24.de/scripte/nexarch/connect/internal/migrate" + "gitea.perlbach24.de/scripte/nexarch/connect/internal/testdb" + "gitea.perlbach24.de/scripte/nexarch/connect/internal/webhookclient" + "gitea.perlbach24.de/scripte/nexarch/connect/internal/webhooksub" +) + +type fakeKEK struct{ kek []byte } + +func (f *fakeKEK) TenantKEK(ctx context.Context, tenantSlug string) ([]byte, error) { + return f.kek, nil +} + +// fakeCoreWebhookAPI reproduziert Core API-07s Subscribe/Unsubscribe-Vertrag +// -- der von dieser Kachel ausdruecklich erlaubte "minimal gestubte +// Core-Webhook-Registry-Endpunkt" fuer den echten End-zu-Ende-Test (echte +// HTTP-Verbindung, echte Datenbank, kein Mock der Client-Logik selbst). +type fakeCoreWebhookAPI struct { + mu sync.Mutex + subscriptions map[string]bool + secrets map[string]string + nextID int +} + +func newFakeCoreWebhookAPI() *fakeCoreWebhookAPI { + return &fakeCoreWebhookAPI{subscriptions: make(map[string]bool), secrets: make(map[string]string)} +} + +func (f *fakeCoreWebhookAPI) hasSubscription(id string) bool { + f.mu.Lock() + defer f.mu.Unlock() + return f.subscriptions[id] +} + +// secretFor liefert das Geheimnis, das beim Subscribe fuer coreSubID +// uebergeben wurde -- in echt kennt nur Core dieses Geheimnis und signiert +// damit jede Zustellung; der Test braucht es, um eine echte, gueltige +// Core-Zustellung nachzubilden. +func (f *fakeCoreWebhookAPI) secretFor(coreSubID string) string { + f.mu.Lock() + defer f.mu.Unlock() + return f.secrets[coreSubID] +} + +func (f *fakeCoreWebhookAPI) handle(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("X-Service-Token") == "" { + http.Error(w, "kein service-token", http.StatusUnauthorized) + return + } + switch { + case r.Method == http.MethodPost && r.URL.Path == "/webhooks/subscriptions": + var in struct { + TenantSlug string `json:"tenant_slug"` + EventType string `json:"event_type"` + TargetURL string `json:"target_url"` + Secret string `json:"secret"` + } + if err := json.NewDecoder(r.Body).Decode(&in); err != nil { + http.Error(w, "ungueltiger body", http.StatusBadRequest) + return + } + f.mu.Lock() + f.nextID++ + id := fmt.Sprintf("core-sub-%d", f.nextID) + f.subscriptions[id] = true + f.secrets[id] = in.Secret + f.mu.Unlock() + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]any{ + "id": id, "event_type": in.EventType, "target_url": in.TargetURL, "created_at": time.Now().UTC(), + }) + case r.Method == http.MethodDelete: + id := strings.TrimPrefix(r.URL.Path, "/webhooks/subscriptions/") + f.mu.Lock() + _, existed := f.subscriptions[id] + delete(f.subscriptions, id) + f.mu.Unlock() + if !existed { + http.Error(w, "unbekanntes abonnement", http.StatusNotFound) + return + } + w.WriteHeader(http.StatusNoContent) + default: + http.Error(w, "nicht unterstuetzt", http.StatusNotFound) + } +} + +// testEnv buendelt einen Manager gegen eine echte, isolierte Postgres- +// Testdatenbank und einen echten (gestubten) Core-Webhook-Server. +type testEnv struct { + Manager *webhooksub.Manager + Pool *pgxpool.Pool + Core *fakeCoreWebhookAPI +} + +func setup(t *testing.T) testEnv { + t.Helper() + dsn := os.Getenv("TEST_TENANT_DSN") + if dsn == "" { + t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen") + } + ctx := context.Background() + pool := testdb.Pool(t, dsn) + + migrations, err := migrate.Load(findMigrationsDir(t)) + if err != nil { + t.Fatalf("migrationen laden: %v", err) + } + if _, err := migrate.Up(ctx, pool, migrations); err != nil { + t.Fatalf("migrationen anwenden: %v", err) + } + + core := newFakeCoreWebhookAPI() + coreSrv := httptest.NewServer(http.HandlerFunc(core.handle)) + t.Cleanup(coreSrv.Close) + + cryptoSvc := connectcrypto.NewService(&fakeKEK{kek: make([]byte, connectcrypto.KEKSize)}) + client := webhookclient.New(coreSrv.URL, "svc-token-test") + mgr := webhooksub.NewManager(pool, cryptoSvc, client, "https://connect.example.test/connector-webhooks/deliveries") + + return testEnv{Manager: mgr, Pool: pool, Core: core} +} + +func findMigrationsDir(t *testing.T) string { + t.Helper() + dir, _ := os.Getwd() + for i := 0; i < 6; i++ { + p := dir + "/migrations" + if fi, err := os.Stat(p); err == nil && fi.IsDir() { + return p + } + dir += "/.." + } + t.Fatal("connect/migrations nicht gefunden") + return "" +} diff --git a/connect/internal/webhooksub/delivery.go b/connect/internal/webhooksub/delivery.go new file mode 100644 index 00000000..c310089c --- /dev/null +++ b/connect/internal/webhooksub/delivery.go @@ -0,0 +1,92 @@ +package webhooksub + +import ( + "crypto/hmac" + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "errors" + "io" + "net/http" + + "github.com/jackc/pgx/v5" + + "gitea.perlbach24.de/scripte/nexarch/connect/internal/crypto" +) + +// SignatureHeader ist Core API-07s dokumentierter Vertrag +// (internal/webhook.SignatureHeader) -- hier erneut definiert, weil Connect +// als physisch getrenntes Go-Modul Cores internal-Paket nicht importieren +// kann. +const SignatureHeader = "X-Nexarch-Signature-256" + +func verifySignature(secret string, payload []byte, signature string) bool { + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(payload) + expected := mac.Sum(nil) + got, err := hex.DecodeString(signature) + if err != nil { + return false + } + return subtle.ConstantTimeCompare(expected, got) == 1 +} + +const maxDeliveryBody = 1 << 20 + +// RegisterRoutes verdrahtet Core API-07s Zustellungs-Callback +// (Akzeptanzkriterium 2): Core ruft diesen Pfad unter der bei Subscribe +// uebergebenen target_url auf. +func (m *Manager) RegisterRoutes(mux *http.ServeMux) { + mux.HandleFunc("POST /connector-webhooks/deliveries/{id}", m.handleDelivery) +} + +func (m *Manager) handleDelivery(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + + var tenantSlug, connectorName, eventType string + var ciphertext, wrappedDEK []byte + err := m.pool.QueryRow(r.Context(), ` + SELECT tenant_slug, connector_name, event_type, secret_ciphertext, secret_wrapped_dek + FROM connector_webhook_subscriptions WHERE id = $1 + `, id).Scan(&tenantSlug, &connectorName, &eventType, &ciphertext, &wrappedDEK) + if errors.Is(err, pgx.ErrNoRows) { + http.Error(w, "unbekanntes abonnement", http.StatusNotFound) + return + } + if err != nil { + http.Error(w, "abonnement konnte nicht gelesen werden", http.StatusInternalServerError) + return + } + + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxDeliveryBody)) + if err != nil { + http.Error(w, "payload zu gross oder nicht lesbar", http.StatusBadRequest) + return + } + + secretPlain, err := m.crypto.Open(r.Context(), tenantSlug, &crypto.Sealed{Ciphertext: ciphertext, WrappedDEK: wrappedDEK}) + if err != nil { + http.Error(w, "geheimnis nicht entschluesselbar", http.StatusInternalServerError) + return + } + + if !verifySignature(string(secretPlain), body, r.Header.Get(SignatureHeader)) { + http.Error(w, "signatur ungueltig", http.StatusUnauthorized) + return + } + + action, ok := m.action(connectorName, eventType) + if !ok { + // Abonnement existiert, aber im aktuellen Prozess (noch) keine + // Aktion registriert -- 503 statt 404, damit Cores Zustellungs- + // Retry (API-07 AC2) es erneut versucht statt endgueltig + // aufzugeben. + http.Error(w, "connector-aktion derzeit nicht verfuegbar", http.StatusServiceUnavailable) + return + } + if err := action(r.Context(), body); err != nil { + http.Error(w, "connector-aktion fehlgeschlagen", http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusOK) +} diff --git a/connect/internal/webhooksub/delivery_test.go b/connect/internal/webhooksub/delivery_test.go new file mode 100644 index 00000000..ade4ede4 --- /dev/null +++ b/connect/internal/webhooksub/delivery_test.go @@ -0,0 +1,160 @@ +package webhooksub_test + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "gitea.perlbach24.de/scripte/nexarch/connect/internal/webhooksub" +) + +func sign(secret string, payload []byte) string { + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(payload) + return hex.EncodeToString(mac.Sum(nil)) +} + +// subscribeAndGetDeliveryURL abonniert ueber env.Manager und liefert die +// echte Zustell-URL samt Core-Geheimnis zurueck, wie sie ein echter +// Core-Zustellversuch verwenden wuerde. +func subscribeAndGetDeliveryURL(t *testing.T, env testEnv, srv *httptest.Server, connectorName, eventType string) (deliveryURL, secret string) { + t.Helper() + reg, err := env.Manager.Subscribe(context.Background(), "acme", connectorName, eventType) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + var coreSubID string + if err := env.Pool.QueryRow(context.Background(), ` + SELECT core_subscription_id FROM connector_webhook_subscriptions WHERE id = $1 + `, reg.ID).Scan(&coreSubID); err != nil { + t.Fatalf("core_subscription_id lesen: %v", err) + } + return srv.URL + "/connector-webhooks/deliveries/" + reg.ID, env.Core.secretFor(coreSubID) +} + +// Pruefung 2 / Akzeptanzkriterium 2: ein zugestelltes Ereignis loest die +// zugeordnete Connector-Aktion aus -- echter HTTP-Roundtrip gegen den +// echten Handler, echte Signaturpruefung. +func TestHandleDelivery_ValidSignature_TriggersConnectorAction(t *testing.T) { + env := setup(t) + + var mu sync.Mutex + var gotPayload []byte + env.Manager.RegisterAction("testcrm", "dms.document.archived", func(_ context.Context, payload []byte) error { + mu.Lock() + defer mu.Unlock() + gotPayload = append([]byte(nil), payload...) + return nil + }) + + mux := http.NewServeMux() + env.Manager.RegisterRoutes(mux) + srv := httptest.NewServer(mux) + defer srv.Close() + + deliveryURL, secret := subscribeAndGetDeliveryURL(t, env, srv, "testcrm", "dms.document.archived") + + payload := []byte(`{"document_id":"doc-42","event":"archived"}`) + req, _ := http.NewRequest(http.MethodPost, deliveryURL, bytes.NewReader(payload)) + req.Header.Set(webhooksub.SignatureHeader, sign(secret, payload)) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("zustellung: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("erwartet 200, bekam %d", resp.StatusCode) + } + + mu.Lock() + defer mu.Unlock() + if string(gotPayload) != string(payload) { + t.Fatalf("connector-aktion erhielt falschen payload: %q", gotPayload) + } +} + +func TestHandleDelivery_InvalidSignature_Rejected(t *testing.T) { + env := setup(t) + + triggered := false + env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { + triggered = true + return nil + }) + + mux := http.NewServeMux() + env.Manager.RegisterRoutes(mux) + srv := httptest.NewServer(mux) + defer srv.Close() + + deliveryURL, _ := subscribeAndGetDeliveryURL(t, env, srv, "testcrm", "dms.document.archived") + + payload := []byte(`{"document_id":"doc-42"}`) + req, _ := http.NewRequest(http.MethodPost, deliveryURL, bytes.NewReader(payload)) + req.Header.Set(webhooksub.SignatureHeader, sign("falsches-geheimnis", payload)) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("zustellung: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("erwartet 401 bei ungueltiger signatur, bekam %d", resp.StatusCode) + } + if triggered { + t.Fatal("connector-aktion darf bei ungueltiger signatur nicht ausgeloest werden") + } +} + +func TestHandleDelivery_UnknownSubscription_NotFound(t *testing.T) { + env := setup(t) + mux := http.NewServeMux() + env.Manager.RegisterRoutes(mux) + srv := httptest.NewServer(mux) + defer srv.Close() + + resp, err := http.Post(srv.URL+"/connector-webhooks/deliveries/unbekannt", "application/json", bytes.NewReader([]byte("{}"))) + if err != nil { + t.Fatalf("zustellung: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("erwartet 404, bekam %d", resp.StatusCode) + } +} + +// Nach erfolgreichem Abbestellen darf eine (verspaetete) Zustellung an die +// alte URL nicht mehr ankommen -- bestaetigt die Vollstaendigkeit aus +// Akzeptanzkriterium 3 auch aus Empfaengersicht. +func TestHandleDelivery_AfterUnsubscribe_NotFound(t *testing.T) { + env := setup(t) + env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil }) + + mux := http.NewServeMux() + env.Manager.RegisterRoutes(mux) + srv := httptest.NewServer(mux) + defer srv.Close() + + deliveryURL, secret := subscribeAndGetDeliveryURL(t, env, srv, "testcrm", "dms.document.archived") + + if err := env.Manager.Unsubscribe(context.Background(), "acme", "testcrm", "dms.document.archived"); err != nil { + t.Fatalf("Unsubscribe: %v", err) + } + + payload := []byte(`{"document_id":"doc-42"}`) + req, _ := http.NewRequest(http.MethodPost, deliveryURL, bytes.NewReader(payload)) + req.Header.Set(webhooksub.SignatureHeader, sign(secret, payload)) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("zustellung: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("erwartet 404 nach abbestellung, bekam %d", resp.StatusCode) + } +} diff --git a/connect/internal/webhooksub/webhooksub.go b/connect/internal/webhooksub/webhooksub.go new file mode 100644 index 00000000..c21b609b --- /dev/null +++ b/connect/internal/webhooksub/webhooksub.go @@ -0,0 +1,157 @@ +// Package webhooksub laesst Connect-Connectoren Ereignisse ueber Core +// API-07 (zentrale Webhook-Registry) abonnieren (API-03). Connect +// registriert lediglich Abonnements und empfaengt Zustellungen -- es +// betreibt selbst keine Zustellungs-Engine (Produkt-DNA dieser Kachel). +package webhooksub + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/connect/internal/crypto" + "gitea.perlbach24.de/scripte/nexarch/connect/internal/webhookclient" +) + +var ( + ErrActionNotRegistered = errors.New("webhooksub: keine connector-aktion fuer dieses ereignis registriert") + ErrNotFound = errors.New("webhooksub: abonnement nicht gefunden") +) + +// Action verarbeitet ein zugestelltes Ereignis fuer genau einen Connector +// und Ereignistyp (Akzeptanzkriterium 2). +type Action func(ctx context.Context, payload []byte) error + +// Registration ist Connects lokale Sicht auf ein bei Core registriertes +// Abonnement. +type Registration struct { + ID string + TenantSlug string + ConnectorName string + EventType string + CreatedAt time.Time +} + +// Manager verwaltet Connect-seitige Webhook-Abonnements: er registriert sie +// bei Core API-07, haelt die Zuordnung Ereignistyp -> Connector-Aktion im +// Prozessspeicher (wie registry.Registry) und verifiziert/verteilt +// zugestellte Ereignisse. +type Manager struct { + pool *pgxpool.Pool + crypto *crypto.Service + core *webhookclient.Client + callbackBaseURL string + + mu sync.RWMutex + actions map[string]Action +} + +func NewManager(pool *pgxpool.Pool, cryptoSvc *crypto.Service, core *webhookclient.Client, callbackBaseURL string) *Manager { + return &Manager{ + pool: pool, crypto: cryptoSvc, core: core, + callbackBaseURL: strings.TrimRight(callbackBaseURL, "/"), + actions: make(map[string]Action), + } +} + +func actionKey(connectorName, eventType string) string { return connectorName + "\x00" + eventType } + +// RegisterAction verdrahtet die Aktion eines Connectors fuer eventType, vor +// jedem Subscribe/jeder Zustellung dafuer -- prozesslokal, wie +// registry.Registry, bei jedem Prozessstart neu aufgebaut. +func (m *Manager) RegisterAction(connectorName, eventType string, action Action) { + m.mu.Lock() + defer m.mu.Unlock() + m.actions[actionKey(connectorName, eventType)] = action +} + +func (m *Manager) action(connectorName, eventType string) (Action, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + a, ok := m.actions[actionKey(connectorName, eventType)] + return a, ok +} + +// Subscribe registriert connectorName fuer eventType bei Core API-07 +// (Akzeptanzkriterium 1). Es verlangt eine vorab per RegisterAction +// hinterlegte Aktion -- ein Abonnement ohne Ziel waere nie zustellbar. +// Schlaegt die Core-Registrierung fehl, wird nichts lokal gespeichert. +func (m *Manager) Subscribe(ctx context.Context, tenantSlug, connectorName, eventType string) (Registration, error) { + if _, ok := m.action(connectorName, eventType); !ok { + return Registration{}, fmt.Errorf("%w: %s/%s", ErrActionNotRegistered, connectorName, eventType) + } + id, err := newToken(16) + if err != nil { + return Registration{}, fmt.Errorf("webhooksub: id generieren: %w", err) + } + secret, err := newToken(32) + if err != nil { + return Registration{}, fmt.Errorf("webhooksub: geheimnis generieren: %w", err) + } + targetURL := m.callbackBaseURL + "/" + id + + coreSub, err := m.core.Subscribe(ctx, tenantSlug, eventType, targetURL, secret) + if err != nil { + return Registration{}, fmt.Errorf("webhooksub: core-registrierung: %w", err) + } + + sealed, err := m.crypto.Seal(ctx, tenantSlug, []byte(secret)) + if err != nil { + return Registration{}, fmt.Errorf("webhooksub: geheimnis versiegeln: %w", err) + } + + var createdAt time.Time + err = m.pool.QueryRow(ctx, ` + INSERT INTO connector_webhook_subscriptions + (id, tenant_slug, connector_name, event_type, core_subscription_id, secret_ciphertext, secret_wrapped_dek) + VALUES ($1, $2, $3, $4, $5, $6, $7) + RETURNING created_at + `, id, tenantSlug, connectorName, eventType, coreSub.ID, sealed.Ciphertext, sealed.WrappedDEK).Scan(&createdAt) + if err != nil { + return Registration{}, fmt.Errorf("webhooksub: abonnement speichern: %w", err) + } + return Registration{ID: id, TenantSlug: tenantSlug, ConnectorName: connectorName, EventType: eventType, CreatedAt: createdAt}, nil +} + +// Unsubscribe entfernt connectorNames eventType-Abonnement vollstaendig +// (Akzeptanzkriterium 3): erst bei Core, dann lokal. Schlaegt die +// Core-Abbestellung fehl, bleibt die lokale Zeile fuer einen Wiederholungs- +// versuch bestehen statt ein verwaistes Core-Abonnement zu hinterlassen. +func (m *Manager) Unsubscribe(ctx context.Context, tenantSlug, connectorName, eventType string) error { + var localID, coreSubscriptionID string + err := m.pool.QueryRow(ctx, ` + SELECT id, core_subscription_id FROM connector_webhook_subscriptions + WHERE tenant_slug = $1 AND connector_name = $2 AND event_type = $3 + `, tenantSlug, connectorName, eventType).Scan(&localID, &coreSubscriptionID) + if errors.Is(err, pgx.ErrNoRows) { + return ErrNotFound + } + if err != nil { + return fmt.Errorf("webhooksub: abonnement lesen: %w", err) + } + + if err := m.core.Unsubscribe(ctx, tenantSlug, coreSubscriptionID); err != nil { + return fmt.Errorf("webhooksub: core-abbestellung: %w", err) + } + + if _, err := m.pool.Exec(ctx, `DELETE FROM connector_webhook_subscriptions WHERE id = $1`, localID); err != nil { + return fmt.Errorf("webhooksub: abonnement loeschen: %w", err) + } + return nil +} + +func newToken(numBytes int) (string, error) { + b := make([]byte, numBytes) + if _, err := rand.Read(b); err != nil { + return "", err + } + return hex.EncodeToString(b), nil +} diff --git a/connect/internal/webhooksub/webhooksub_test.go b/connect/internal/webhooksub/webhooksub_test.go new file mode 100644 index 00000000..91d90c7c --- /dev/null +++ b/connect/internal/webhooksub/webhooksub_test.go @@ -0,0 +1,157 @@ +package webhooksub_test + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + + connectcrypto "gitea.perlbach24.de/scripte/nexarch/connect/internal/crypto" + "gitea.perlbach24.de/scripte/nexarch/connect/internal/webhookclient" + "gitea.perlbach24.de/scripte/nexarch/connect/internal/webhooksub" +) + +// Pruefung 1 / Akzeptanzkriterium 1: Testabonnement wird ueber die +// (gestubte) Core-Webhook-Registry registriert. +func TestSubscribe_RegistersAtCoreAndPersistsLocally(t *testing.T) { + env := setup(t) + env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil }) + + reg, err := env.Manager.Subscribe(context.Background(), "acme", "testcrm", "dms.document.archived") + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + if reg.ID == "" { + t.Fatal("erwartete lokale id") + } + + var count int + var coreSubID string + if err := env.Pool.QueryRow(context.Background(), ` + SELECT count(*), max(core_subscription_id) FROM connector_webhook_subscriptions WHERE id = $1 + `, reg.ID).Scan(&count, &coreSubID); err != nil { + t.Fatalf("row lesen: %v", err) + } + if count != 1 { + t.Fatalf("erwartete genau eine zeile, habe %d", count) + } + if !env.Core.hasSubscription(coreSubID) { + t.Fatalf("core-abonnement %q nicht bei core registriert", coreSubID) + } +} + +func TestSubscribe_WithoutRegisteredAction_Fails(t *testing.T) { + env := setup(t) + + _, err := env.Manager.Subscribe(context.Background(), "acme", "testcrm", "dms.document.archived") + if !errors.Is(err, webhooksub.ErrActionNotRegistered) { + t.Fatalf("erwartete ErrActionNotRegistered, bekam %v", err) + } +} + +// Negativtest: schlaegt die Core-Registrierung fehl, bleibt lokal nichts +// zurueck (kein verwaistes Abonnement ohne Core-Gegenstueck). +func TestSubscribe_CoreFailure_PersistsNothingLocally(t *testing.T) { + env := setup(t) + env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil }) + + failingCore := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer failingCore.Close() + + cryptoSvc := connectcrypto.NewService(&fakeKEK{kek: make([]byte, connectcrypto.KEKSize)}) + mgr := webhooksub.NewManager(env.Pool, cryptoSvc, webhookclient.New(failingCore.URL, "svc-token-test"), "https://connect.example.test/connector-webhooks/deliveries") + mgr.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil }) + + if _, err := mgr.Subscribe(context.Background(), "acme", "testcrm", "dms.document.archived"); err == nil { + t.Fatal("erwarteter fehler bei core-ausfall blieb aus") + } + + var count int + if err := env.Pool.QueryRow(context.Background(), ` + SELECT count(*) FROM connector_webhook_subscriptions WHERE tenant_slug = 'acme' AND connector_name = 'testcrm' + `).Scan(&count); err != nil { + t.Fatalf("zeilen zaehlen: %v", err) + } + if count != 0 { + t.Fatalf("erwartete keine lokale zeile nach core-ausfall, habe %d", count) + } +} + +// Pruefung 3 / Akzeptanzkriterium 3: Abbestellen entfernt das Abonnement +// vollstaendig -- bei Core und lokal. +func TestUnsubscribe_RemovesAtCoreAndLocally_Completely(t *testing.T) { + env := setup(t) + env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil }) + reg, err := env.Manager.Subscribe(context.Background(), "acme", "testcrm", "dms.document.archived") + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + + var coreSubID string + if err := env.Pool.QueryRow(context.Background(), ` + SELECT core_subscription_id FROM connector_webhook_subscriptions WHERE id = $1 + `, reg.ID).Scan(&coreSubID); err != nil { + t.Fatalf("core_subscription_id lesen: %v", err) + } + + if err := env.Manager.Unsubscribe(context.Background(), "acme", "testcrm", "dms.document.archived"); err != nil { + t.Fatalf("Unsubscribe: %v", err) + } + + var count int + if err := env.Pool.QueryRow(context.Background(), ` + SELECT count(*) FROM connector_webhook_subscriptions WHERE id = $1 + `, reg.ID).Scan(&count); err != nil { + t.Fatalf("zeilen zaehlen: %v", err) + } + if count != 0 { + t.Fatalf("erwartete vollstaendig entfernte lokale zeile, habe noch %d", count) + } + if env.Core.hasSubscription(coreSubID) { + t.Fatal("erwartete bei core entferntes abonnement, ist noch vorhanden") + } +} + +func TestUnsubscribe_UnknownSubscription_ReturnsNotFound(t *testing.T) { + env := setup(t) + err := env.Manager.Unsubscribe(context.Background(), "acme", "testcrm", "dms.document.archived") + if !errors.Is(err, webhooksub.ErrNotFound) { + t.Fatalf("erwartete ErrNotFound, bekam %v", err) + } +} + +// Negativtest: schlaegt die Core-Abbestellung fehl, bleibt die lokale Zeile +// fuer einen Wiederholungsversuch bestehen statt ein verwaistes +// Core-Abonnement zu hinterlassen. +func TestUnsubscribe_CoreFailure_KeepsLocalRowForRetry(t *testing.T) { + env := setup(t) + env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil }) + reg, err := env.Manager.Subscribe(context.Background(), "acme", "testcrm", "dms.document.archived") + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + + brokenCore := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer brokenCore.Close() + cryptoSvc := connectcrypto.NewService(&fakeKEK{kek: make([]byte, connectcrypto.KEKSize)}) + mgrWithBrokenCore := webhooksub.NewManager(env.Pool, cryptoSvc, webhookclient.New(brokenCore.URL, "svc-token-test"), "https://connect.example.test/connector-webhooks/deliveries") + + if err := mgrWithBrokenCore.Unsubscribe(context.Background(), "acme", "testcrm", "dms.document.archived"); err == nil { + t.Fatal("erwarteter fehler bei core-ausfall blieb aus") + } + + var count int + if err := env.Pool.QueryRow(context.Background(), ` + SELECT count(*) FROM connector_webhook_subscriptions WHERE id = $1 + `, reg.ID).Scan(&count); err != nil { + t.Fatalf("zeilen zaehlen: %v", err) + } + if count != 1 { + t.Fatalf("erwartete erhaltene lokale zeile nach core-ausfall, habe %d", count) + } +} diff --git a/connect/migrations/0013_connector_webhook_subscriptions.down.sql b/connect/migrations/0013_connector_webhook_subscriptions.down.sql new file mode 100644 index 00000000..c91747c0 --- /dev/null +++ b/connect/migrations/0013_connector_webhook_subscriptions.down.sql @@ -0,0 +1 @@ +DROP TABLE connector_webhook_subscriptions; diff --git a/connect/migrations/0013_connector_webhook_subscriptions.up.sql b/connect/migrations/0013_connector_webhook_subscriptions.up.sql new file mode 100644 index 00000000..4ffebc44 --- /dev/null +++ b/connect/migrations/0013_connector_webhook_subscriptions.up.sql @@ -0,0 +1,16 @@ +-- API-03: lokale Abbildung eines Connect-Connector-Abonnements bei Core +-- API-07 (zentrale Webhook-Registry). Der lokale id-Wert ist zugleich das +-- Pfadsegment der an Core uebergebenen target_url, damit eine zugestellte +-- Anfrage ohne Suche nach event_type+tenant dem richtigen Connector +-- zugeordnet werden kann. +CREATE TABLE connector_webhook_subscriptions ( + id TEXT PRIMARY KEY, + tenant_slug TEXT NOT NULL, + connector_name TEXT NOT NULL, + event_type TEXT NOT NULL, + core_subscription_id TEXT NOT NULL, + secret_ciphertext BYTEA NOT NULL, + secret_wrapped_dek BYTEA NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (tenant_slug, connector_name, event_type) +); diff --git a/connect/scripts/check-security-test-coverage.sh b/connect/scripts/check-security-test-coverage.sh index ad0efe71..41abb252 100755 --- a/connect/scripts/check-security-test-coverage.sh +++ b/connect/scripts/check-security-test-coverage.sh @@ -25,6 +25,8 @@ SECURITY_PACKAGES=( "internal/m365calsync" "internal/syncjob" "internal/statusapi" + "internal/webhookclient" + "internal/webhooksub" ) cd "$(dirname "$0")/.."