diff --git a/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/webhook/HmacSigner.java b/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/webhook/HmacSigner.java new file mode 100644 index 0000000..798d62e --- /dev/null +++ b/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/webhook/HmacSigner.java @@ -0,0 +1,54 @@ +package cloud.nstart.terravault.ordinis.outbox.webhook; + +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import java.nio.charset.StandardCharsets; +import java.security.NoSuchAlgorithmException; + +/** + * HMAC-SHA256 signing для webhook deliveries. Receiver проверяет подпись через + * shared secret чтобы убедиться что запрос от Ordinis. + * + *

Header format: {@code X-Ordinis-Signature: sha256=}. + */ +public final class HmacSigner { + + private static final String ALGORITHM = "HmacSHA256"; + public static final String HEADER_NAME = "X-Ordinis-Signature"; + + private HmacSigner() {} + + /** + * Подписать body shared secret'ом. Возвращает значение для header + * (с префиксом {@code sha256=}). + * + * @param secret hex-encoded HMAC secret (64 chars = 32 bytes) + * @param body raw bytes тела запроса + * @return {@code sha256=} + */ + public static String sign(String secret, byte[] body) { + if (secret == null || secret.isBlank()) { + throw new IllegalArgumentException("secret must not be empty"); + } + try { + Mac mac = Mac.getInstance(ALGORITHM); + byte[] keyBytes = secret.getBytes(StandardCharsets.UTF_8); + mac.init(new SecretKeySpec(keyBytes, ALGORITHM)); + byte[] hash = mac.doFinal(body); + return "sha256=" + toHex(hash); + } catch (NoSuchAlgorithmException e) { + // HmacSHA256 — required JDK algorithm, не должен fail + throw new IllegalStateException("HmacSHA256 unavailable", e); + } catch (java.security.InvalidKeyException e) { + throw new IllegalArgumentException("invalid HMAC key", e); + } + } + + private static String toHex(byte[] bytes) { + StringBuilder sb = new StringBuilder(bytes.length * 2); + for (byte b : bytes) { + sb.append(String.format("%02x", b)); + } + return sb.toString(); + } +} diff --git a/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/webhook/SsrfGuard.java b/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/webhook/SsrfGuard.java new file mode 100644 index 0000000..4f6957b --- /dev/null +++ b/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/webhook/SsrfGuard.java @@ -0,0 +1,99 @@ +package cloud.nstart.terravault.ordinis.outbox.webhook; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.net.InetAddress; +import java.net.URI; +import java.net.UnknownHostException; +import java.util.List; +import java.util.Set; + +/** + * SSRF (Server-Side Request Forgery) guard для webhook URLs. Блокирует + * delivery на private CIDR (10/8, 172.16/12, 192.168/16, 127/8, 169.254/16). + * + *

Без этой защиты пользователь с правом создать subscription мог бы + * получить blind-SSRF против: + *

+ * + *

Allowlist hosts через config — для testing с echo-сервером в + * ordinis-staging кластере. + */ +public final class SsrfGuard { + + private static final Logger log = LoggerFactory.getLogger(SsrfGuard.class); + + /** Hosts которые ВСЕГДА разрешены (даже если резолвятся в private CIDR). */ + private final Set allowedHosts; + + public SsrfGuard(List allowedHosts) { + this.allowedHosts = allowedHosts == null ? Set.of() : Set.copyOf(allowedHosts); + } + + /** + * Проверить что URL безопасен для outbound webhook delivery. + * + * @return null если OK, иначе reason почему URL отклонён + */ + public String validate(String urlString) { + URI uri; + try { + uri = URI.create(urlString); + } catch (IllegalArgumentException e) { + return "Malformed URL: " + e.getMessage(); + } + + String scheme = uri.getScheme(); + if (!"http".equalsIgnoreCase(scheme) && !"https".equalsIgnoreCase(scheme)) { + return "URL scheme must be http or https, got: " + scheme; + } + + String host = uri.getHost(); + if (host == null || host.isBlank()) { + return "URL must have a host"; + } + + // Allowlist override: внутренний echo-server, тестовые URLs + if (allowedHosts.contains(host.toLowerCase())) { + return null; + } + + // DNS resolution: получаем все A-records, проверяем все на private CIDR. + // Если хотя бы один в private — отказ (защита от DNS rebinding атаки + // тоже минимизирована, т.к. HttpClient может приконнектиться на + // другой адрес позже; это best-effort). + InetAddress[] addresses; + try { + addresses = InetAddress.getAllByName(host); + } catch (UnknownHostException e) { + return "DNS resolution failed: " + e.getMessage(); + } + + for (InetAddress addr : addresses) { + if (isPrivate(addr)) { + log.warn("SSRF guard rejected URL {} → {} (private CIDR)", urlString, addr.getHostAddress()); + return "Address resolves to private/loopback range: " + addr.getHostAddress(); + } + } + + return null; + } + + /** + * Private/special-use ranges per RFC 1918, RFC 3927, RFC 4193, RFC 5735. + * Loopback, link-local, multicast, broadcast, IPv6 unique-local — все в bucket. + */ + static boolean isPrivate(InetAddress addr) { + if (addr.isAnyLocalAddress()) return true; + if (addr.isLoopbackAddress()) return true; // 127.0.0.0/8, ::1 + if (addr.isLinkLocalAddress()) return true; // 169.254/16, fe80::/10 + if (addr.isSiteLocalAddress()) return true; // 10/8, 172.16/12, 192.168/16 + if (addr.isMulticastAddress()) return true; // 224/4 + return false; + } +} diff --git a/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/webhook/WebhookDispatcher.java b/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/webhook/WebhookDispatcher.java new file mode 100644 index 0000000..fd6428b --- /dev/null +++ b/ordinis-outbox/src/main/java/cloud/nstart/terravault/ordinis/outbox/webhook/WebhookDispatcher.java @@ -0,0 +1,265 @@ +package cloud.nstart.terravault.ordinis.outbox.webhook; + +import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEvent; +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 com.fasterxml.jackson.databind.ObjectMapper; +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Timer; +import jakarta.annotation.PostConstruct; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.data.domain.PageRequest; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpRequest.BodyPublishers; +import java.net.http.HttpResponse; +import java.net.http.HttpResponse.BodyHandlers; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.OffsetDateTime; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; + +/** + * Webhook dispatcher: HTTP callback delivery параллельно с Kafka. + * + *

Два @Scheduled tick'а: + *

    + *
  1. matchTick: scan recent unpublished outbox_events × active + * subscriptions. INSERT webhook_deliveries (idempotent ON CONFLICT) для + * каждого match. Пишет PENDING rows. + *
  2. sendTick: scan due deliveries (status pending/retrying AND + * next_attempt_at <= now). Загружает event payload, signs HMAC, POSTs + * URL. Mark delivered (2xx) / retrying (backoff) / dlq (после 1000 attempts). + *
+ * + *

Активируется только при {@code ordinis.webhook.enabled=true}. + */ +@Component +@ConditionalOnProperty(name = "ordinis.webhook.enabled", havingValue = "true", matchIfMissing = false) +public class WebhookDispatcher { + + private static final Logger log = LoggerFactory.getLogger(WebhookDispatcher.class); + + private final OutboxEventRepository outboxRepository; + private final WebhookSubscriptionRepository subscriptionRepository; + private final WebhookDeliveryRepository deliveryRepository; + private final ObjectMapper objectMapper; + private final HttpClient httpClient; + private SsrfGuard ssrfGuard; + + private final Counter deliveredCounter; + private final Counter failedCounter; + private final Counter dlqCounter; + private final Counter ssrfRejectedCounter; + private final Timer deliveryTimer; + + @Value("${ordinis.webhook.batch-size:50}") + private int batchSize; + + @Value("${ordinis.webhook.send-timeout-ms:5000}") + private long sendTimeoutMs; + + @Value("${ordinis.webhook.allowed-hosts:}") + private String allowedHostsCsv; + + public WebhookDispatcher( + OutboxEventRepository outboxRepository, + WebhookSubscriptionRepository subscriptionRepository, + WebhookDeliveryRepository deliveryRepository, + ObjectMapper objectMapper, + MeterRegistry meterRegistry) { + this.outboxRepository = outboxRepository; + this.subscriptionRepository = subscriptionRepository; + this.deliveryRepository = deliveryRepository; + this.objectMapper = objectMapper; + this.httpClient = HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(3)) + .build(); + + this.deliveredCounter = Counter.builder("nsi_webhook_delivered_total") + .description("Successful webhook deliveries (2xx response)") + .register(meterRegistry); + this.failedCounter = Counter.builder("nsi_webhook_failed_total") + .description("Failed webhook delivery attempts (will retry)") + .register(meterRegistry); + this.dlqCounter = Counter.builder("nsi_webhook_dlq_total") + .description("Webhook deliveries moved to DLQ (attempt cap reached)") + .register(meterRegistry); + this.ssrfRejectedCounter = Counter.builder("nsi_webhook_ssrf_rejected_total") + .description("Webhook URLs rejected by SSRF guard") + .register(meterRegistry); + this.deliveryTimer = Timer.builder("nsi_webhook_delivery_duration_seconds") + .description("HTTP POST duration for webhook delivery") + .register(meterRegistry); + } + + @PostConstruct + void initSsrfGuard() { + List allowed = allowedHostsCsv == null || allowedHostsCsv.isBlank() + ? List.of() + : Arrays.stream(allowedHostsCsv.split(",")).map(String::trim).toList(); + this.ssrfGuard = new SsrfGuard(allowed); + log.info("WebhookDispatcher initialized. Allowed hosts: {}", allowed); + } + + // === Tick 1: matcher === + + /** + * Scan recent unpublished events × active subscriptions, INSERT pending + * deliveries. Idempotent: ON CONFLICT (subscription_id, outbox_event_id) + * DO NOTHING. + * + *

Note: WebhookDispatcher работает по same outbox_events что и + * Kafka publisher. Event попадает в дисптчер до того как Kafka publisher + * marks published_at — это нормально, мы matching по существованию + * row, не по published status. + */ + @Scheduled(fixedDelayString = "${ordinis.webhook.match-interval-ms:2000}") + @Transactional + public void matchTick() { + List subs = subscriptionRepository.findActive(); + if (subs.isEmpty()) return; + + // Берём recent events (последние 1000 самых новых). Старше — уже либо + // published в Kafka (значит и сюда match'нулись если был active sub + // в момент создания), либо в DLQ. Поллер не нужен для catch-up. + List recent = outboxRepository + .findUnpublished(PageRequest.of(0, 1000)); + + for (OutboxEvent event : recent) { + for (WebhookSubscription sub : subs) { + if (matches(event, sub)) { + deliveryRepository.insertIfAbsent(sub.getId(), event.getId()); + } + } + } + } + + static boolean matches(OutboxEvent event, WebhookSubscription sub) { + if (sub.getScopeFilter() != null && !sub.getScopeFilter().isEmpty() + && !sub.getScopeFilter().contains(event.getDataScope())) { + return false; + } + if (sub.getDictionaryFilter() != null && !sub.getDictionaryFilter().isEmpty() + && (event.getDictionaryName() == null + || !sub.getDictionaryFilter().contains(event.getDictionaryName()))) { + return false; + } + if (sub.getEventTypeFilter() != null && !sub.getEventTypeFilter().isEmpty() + && !sub.getEventTypeFilter().contains(event.getEventType())) { + return false; + } + return true; + } + + // === Tick 2: sender === + + @Scheduled(fixedDelayString = "${ordinis.webhook.send-interval-ms:1000}") + public void sendTick() { + List due = deliveryRepository.findDue( + OffsetDateTime.now(), PageRequest.of(0, batchSize)); + if (due.isEmpty()) return; + + log.debug("Webhook sendTick: {} due deliveries", due.size()); + + for (WebhookDelivery delivery : due) { + attemptDelivery(delivery); + } + } + + /** + * Одна попытка доставки. @Transactional чтобы изменения delivery атомарны. + * SELF-call через bean reference не делаем — каждая итерация в своей tx + * через wrapper public method. + */ + @Transactional + void attemptDelivery(WebhookDelivery delivery) { + Optional subOpt = subscriptionRepository.findById(delivery.getSubscriptionId()); + if (subOpt.isEmpty() || !subOpt.get().isActive()) { + // Subscription deleted/disabled между match и send. Mark dlq чтобы не + // тратить попытки впустую (admin может resetDlq если subscription + // вернётся). + delivery.markFailed("Subscription not found or inactive", null); + deliveryRepository.save(delivery); + return; + } + WebhookSubscription sub = subOpt.get(); + + Optional eventOpt = outboxRepository.findById(delivery.getOutboxEventId()); + if (eventOpt.isEmpty()) { + delivery.markFailed("Outbox event not found", null); + deliveryRepository.save(delivery); + return; + } + OutboxEvent event = eventOpt.get(); + + // SSRF guard: блокируем delivery на private CIDR. + String ssrfReason = ssrfGuard.validate(sub.getUrl()); + if (ssrfReason != null) { + ssrfRejectedCounter.increment(); + delivery.markFailed("SSRF guard: " + ssrfReason, null); + if (delivery.isDlq()) dlqCounter.increment(); + deliveryRepository.save(delivery); + return; + } + + Timer.Sample sample = Timer.start(); + try { + byte[] body = objectMapper.writeValueAsBytes(event.getPayload()); + String signature = HmacSigner.sign(sub.getHmacSecret(), body); + + HttpRequest req = HttpRequest.newBuilder() + .uri(URI.create(sub.getUrl())) + .timeout(Duration.ofMillis(sendTimeoutMs)) + .header("Content-Type", "application/json") + .header("User-Agent", "Ordinis-Webhook/1.0") + .header(HmacSigner.HEADER_NAME, signature) + .header("X-Ordinis-Event-Id", String.valueOf(event.getId())) + .header("X-Ordinis-Event-Type", event.getEventType()) + .header("X-Ordinis-Scope", event.getDataScope().name()) + .POST(BodyPublishers.ofByteArray(body)) + .build(); + + HttpResponse resp = httpClient.send(req, BodyHandlers.ofString(StandardCharsets.UTF_8)); + sample.stop(deliveryTimer); + + if (resp.statusCode() >= 200 && resp.statusCode() < 300) { + delivery.markDelivered(resp.statusCode()); + deliveredCounter.increment(); + } else { + // 4xx — клиент скорее всего не починится, но стандартный retry + // (admin может deactivate subscription). 5xx — transient. + String body2xx = resp.body() == null ? "" : resp.body(); + String error = "HTTP " + resp.statusCode() + ": " + + body2xx.substring(0, Math.min(200, body2xx.length())); + delivery.markFailed(error, resp.statusCode()); + failedCounter.increment(); + if (delivery.isDlq()) dlqCounter.increment(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + delivery.markFailed("Interrupted: " + e.getMessage(), null); + failedCounter.increment(); + } catch (Exception e) { + delivery.markFailed( + e.getClass().getSimpleName() + ": " + e.getMessage(), null); + failedCounter.increment(); + if (delivery.isDlq()) dlqCounter.increment(); + } + deliveryRepository.save(delivery); + } +}