From fcfbb1c42023c8160b4799d562b74c0181c87a1a Mon Sep 17 00:00:00 2001 From: "Zimin A.N." Date: Wed, 6 May 2026 15:49:41 +0300 Subject: [PATCH] =?UTF-8?q?test(webhook):=20e2e=20dispatcher=20=E2=86=92?= =?UTF-8?q?=20embedded=20HttpServer=20+=20HMAC=20verify?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Закрывает Phase 5 deferred item. Завершает webhooks v2 покрытие. WebhookE2ETest (2 tests): Test 1: success path - Стартует JDK HttpServer на random localhost port (no extra deps) - Создаёт WebhookSubscription указывающий на /hook - POST record в dictionary → outbox event - Ждёт что receiver получит POST в течение 20s - Verify body содержит business_key - Verify HMAC signature header валиден (HmacSigner.sign(secret, body) matches X-Ordinis-Signature) - Verify Ordinis headers (X-Ordinis-Event-Type, X-Ordinis-Scope=PUBLIC) - Verify DB delivery row.status = delivered + lastStatusCode = 200 Test 2: retry path - Receiver всегда 500 - POST record → outbox → dispatcher tries deliver - Verify delivery transitions to status=retrying + lastStatusCode=500 - Не ждём DLQ (1000 attempts × 1m backoff = недели) SSRF guard через ordinis.webhook.allowed-hosts=localhost,127.0.0.1 test override (production deny private CIDR остаётся включённым). Test config: - ordinis.outbox.enabled=true (для outbox poller) - ordinis.webhook.enabled=true (для dispatcher) - match-interval-ms=200, send-interval-ms=200 (faster tests) Webhooks v2 — все 5 phases закрыты. --- .../ordinis/app/e2e/WebhookE2ETest.java | 239 ++++++++++++++++++ 1 file changed, 239 insertions(+) create mode 100644 ordinis-app/src/test/java/cloud/nstart/terravault/ordinis/app/e2e/WebhookE2ETest.java diff --git a/ordinis-app/src/test/java/cloud/nstart/terravault/ordinis/app/e2e/WebhookE2ETest.java b/ordinis-app/src/test/java/cloud/nstart/terravault/ordinis/app/e2e/WebhookE2ETest.java new file mode 100644 index 0000000..73149b7 --- /dev/null +++ b/ordinis-app/src/test/java/cloud/nstart/terravault/ordinis/app/e2e/WebhookE2ETest.java @@ -0,0 +1,239 @@ +package cloud.nstart.terravault.ordinis.app.e2e; + +import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository; +import cloud.nstart.terravault.ordinis.domain.webhook.WebhookDelivery; +import cloud.nstart.terravault.ordinis.domain.webhook.WebhookDeliveryRepository; +import cloud.nstart.terravault.ordinis.domain.webhook.WebhookSubscription; +import cloud.nstart.terravault.ordinis.domain.webhook.WebhookSubscriptionRepository; +import cloud.nstart.terravault.ordinis.outbox.webhook.HmacSigner; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sun.net.httpserver.HttpServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.http.MediaType; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; +import org.springframework.test.web.servlet.MockMvc; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +/** + * End-to-end webhook delivery test: + *
    + *
  1. Стартуем embedded {@link HttpServer} на random localhost port — + * imitate consumer's webhook receiver.
    2. + *
  2. Создаём WebhookSubscription указывающий на этот endpoint.
    3. + *
  3. POST a record — попадает в outbox_events.
    4. + *
  4. WebhookDispatcher должен matchTick → INSERT delivery → + * sendTick → POST на наш HttpServer → mark delivered.
    5. + *
  5. Проверяем: receiver получил body + valid HMAC signature + + * Ordinis headers; delivery row mark'нут как delivered.
    6. + *
+ * + *

SSRF guard разрешает localhost через {@code allowed-hosts} config. + */ +@SpringBootTest( + webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = { + "ordinis.outbox.enabled=true", + "ordinis.outbox.poll-interval-ms=200", + "ordinis.outbox.batch-size=20", + "ordinis.webhook.enabled=true", + "ordinis.webhook.match-interval-ms=200", + "ordinis.webhook.send-interval-ms=200", + "ordinis.webhook.allowed-hosts=localhost,127.0.0.1", + }) +@AutoConfigureMockMvc +@ActiveProfiles("test") +class WebhookE2ETest { + + @DynamicPropertySource + static void props(DynamicPropertyRegistry registry) { + E2ESupport.registerProperties(registry); + } + + @Autowired MockMvc mvc; + @Autowired ObjectMapper om; + @Autowired OutboxEventRepository outboxRepo; + @Autowired WebhookSubscriptionRepository subRepo; + @Autowired WebhookDeliveryRepository deliveryRepo; + + private HttpServer receiver; + private final ConcurrentLinkedQueue received = new ConcurrentLinkedQueue<>(); + + record ReceivedRequest(String path, String body, String signatureHeader, String eventTypeHeader, String scopeHeader) {} + + @BeforeEach + void startReceiver() throws IOException { + receiver = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + receiver.createContext("/hook", exchange -> { + byte[] body = exchange.getRequestBody().readAllBytes(); + received.add(new ReceivedRequest( + exchange.getRequestURI().getPath(), + new String(body, StandardCharsets.UTF_8), + exchange.getRequestHeaders().getFirst(HmacSigner.HEADER_NAME), + exchange.getRequestHeaders().getFirst("X-Ordinis-Event-Type"), + exchange.getRequestHeaders().getFirst("X-Ordinis-Scope"))); + exchange.sendResponseHeaders(200, -1); + exchange.close(); + }); + receiver.start(); + } + + @AfterEach + void stopReceiver() { + if (receiver != null) receiver.stop(0); + } + + @Test + void recordCreate_dispatchedToWebhookReceiver_withValidHmac() throws Exception { + int port = receiver.getAddress().getPort(); + String hookUrl = "http://127.0.0.1:" + port + "/hook"; + + // 1. Создаём dict + String dictName = "wh_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8); + JsonNode schema = om.readTree(""" + {"$schema":"http://json-schema.org/draft-07/schema#","type":"object", + "additionalProperties":false,"x-id-source":"code","required":["code"], + "properties":{"code":{"type":"string","x-unique":true}}}"""); + var dictBody = om.createObjectNode() + .put("name", dictName) + .put("scope", "PUBLIC") + .put("schemaVersion", "1.0.0") + .put("bundle", "test"); + dictBody.set("schemaJson", schema); + mvc.perform(post("/api/v1/dictionaries") + .contentType(MediaType.APPLICATION_JSON) + .content(om.writeValueAsBytes(dictBody))) + .andExpect(status().is2xxSuccessful()); + + // 2. Создаём webhook subscription напрямую через repository (CRUD endpoint + // требует RESTRICTED scope от JWT, в тесте без auth — проще через repo) + String secret = "test-secret-" + UUID.randomUUID(); + WebhookSubscription sub = new WebhookSubscription( + "e2e-test-receiver", hookUrl, secret, "test-user"); + sub = subRepo.save(sub); + UUID subId = sub.getId(); + + long deliveriesBefore = deliveryRepo.count(); + + // 3. POST record → outbox_events row created + String needle = "EVT_" + UUID.randomUUID().toString().replace("-", "").substring(0, 12).toUpperCase(); + var recBody = om.createObjectNode().put("businessKey", needle); + recBody.set("data", om.createObjectNode().put("code", needle)); + mvc.perform(post("/api/v1/dictionaries/{n}/records", dictName) + .contentType(MediaType.APPLICATION_JSON) + .content(om.writeValueAsBytes(recBody))) + .andExpect(status().is2xxSuccessful()); + + // 4. Wait for delivery: receiver получил POST request + await().atMost(20, TimeUnit.SECONDS) + .pollInterval(Duration.ofMillis(300)) + .until(() -> received.stream().anyMatch(r -> r.body().contains(needle))); + + // 5. Verify request properties + ReceivedRequest req = received.stream() + .filter(r -> r.body().contains(needle)) + .findFirst() + .orElseThrow(); + + assertThat(req.path()).isEqualTo("/hook"); + assertThat(req.scopeHeader()).isEqualTo("PUBLIC"); + assertThat(req.eventTypeHeader()).isNotBlank(); + + // HMAC signature валиден: receiver мог бы verify через свой secret + String expectedSig = HmacSigner.sign(secret, req.body().getBytes(StandardCharsets.UTF_8)); + assertThat(req.signatureHeader()).isEqualTo(expectedSig); + + // 6. Verify DB delivery row marked delivered + await().atMost(10, TimeUnit.SECONDS) + .pollInterval(Duration.ofMillis(300)) + .until(() -> deliveryRepo.count() > deliveriesBefore); + + List all = deliveryRepo.findBySubscriptionIdOrderByCreatedAtDesc( + subId, org.springframework.data.domain.PageRequest.of(0, 10)).getContent(); + assertThat(all).isNotEmpty(); + WebhookDelivery latest = all.get(0); + await().atMost(10, TimeUnit.SECONDS) + .pollInterval(Duration.ofMillis(300)) + .until(() -> { + WebhookDelivery d = deliveryRepo.findById(latest.getId()).orElseThrow(); + return d.getStatus() == WebhookDelivery.Status.delivered; + }); + + WebhookDelivery delivered = deliveryRepo.findById(latest.getId()).orElseThrow(); + assertThat(delivered.getStatus()).isEqualTo(WebhookDelivery.Status.delivered); + assertThat(delivered.getLastStatusCode()).isEqualTo(200); + assertThat(delivered.getDeliveredAt()).isNotNull(); + assertThat(delivered.getLastError()).isNull(); + } + + @Test + void receiverReturning500_deliveryRetried_notDelivered() throws Exception { + // Стартуем ALT receiver который всегда 500 + HttpServer failing = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + failing.createContext("/fail", exchange -> { + byte[] body = exchange.getRequestBody().readAllBytes(); // drain + assertThat(body.length).isGreaterThan(0); + exchange.sendResponseHeaders(500, -1); + exchange.close(); + }); + failing.start(); + try { + int port = failing.getAddress().getPort(); + String hookUrl = "http://127.0.0.1:" + port + "/fail"; + + // dict + String dictName = "whf_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8); + JsonNode schema = om.readTree(""" + {"$schema":"http://json-schema.org/draft-07/schema#","type":"object", + "additionalProperties":false,"x-id-source":"code","required":["code"], + "properties":{"code":{"type":"string","x-unique":true}}}"""); + var dictBody = om.createObjectNode() + .put("name", dictName).put("scope", "PUBLIC") + .put("schemaVersion", "1.0.0").put("bundle", "test"); + dictBody.set("schemaJson", schema); + mvc.perform(post("/api/v1/dictionaries").contentType(MediaType.APPLICATION_JSON) + .content(om.writeValueAsBytes(dictBody))).andExpect(status().is2xxSuccessful()); + + WebhookSubscription sub = subRepo.save(new WebhookSubscription( + "e2e-failing", hookUrl, "secret-fail", "test-user")); + + String needle = "EVT_" + UUID.randomUUID().toString().replace("-", "").substring(0, 12); + var recBody = om.createObjectNode().put("businessKey", needle); + recBody.set("data", om.createObjectNode().put("code", needle)); + mvc.perform(post("/api/v1/dictionaries/{n}/records", dictName) + .contentType(MediaType.APPLICATION_JSON) + .content(om.writeValueAsBytes(recBody))).andExpect(status().is2xxSuccessful()); + + // Ждём что delivery с status=retrying появится (HTTP 500 → markFailed) + await().atMost(20, TimeUnit.SECONDS) + .pollInterval(Duration.ofMillis(300)) + .until(() -> deliveryRepo.findBySubscriptionIdOrderByCreatedAtDesc( + sub.getId(), org.springframework.data.domain.PageRequest.of(0, 10)) + .getContent().stream() + .anyMatch(d -> d.getStatus() == WebhookDelivery.Status.retrying + && d.getLastStatusCode() != null && d.getLastStatusCode() == 500)); + } finally { + failing.stop(0); + } + } +}