feat(webhook): Phase 2 — dispatcher service + HMAC + SSRF guard
WebhookDispatcher (@Scheduled, conditional ordinis.webhook.enabled=true): - matchTick (every 2s): scan unpublished outbox_events × active subs, INSERT pending deliveries idempotently (ON CONFLICT DO NOTHING) - sendTick (every 1s): scan due deliveries (status pending/retrying AND next_attempt_at <= now), POST URL with HMAC-SHA256 + SSRF check, mark delivered (2xx) / retrying (backoff) / dlq (>= 1000 attempts) HmacSigner: - HMAC-SHA256 hex signing для X-Ordinis-Signature header - Format: `sha256=<hex>` — receiver verifies через shared secret - JDK Mac, no external deps SsrfGuard: - Reject URLs резолвящиеся в private CIDR (10/8, 172.16/12, 192.168/16, 127/8, 169.254/16, link-local, multicast) - Через InetAddress.getAllByName() + isXxxAddress() checks - Allowlist через ordinis.webhook.allowed-hosts CSV для testing HttpClient: JDK 21 native, connectTimeout 3s, send timeout configurable ordinis.webhook.send-timeout-ms (default 5000ms) Метрики Micrometer: - nsi_webhook_delivered_total (success counter) - nsi_webhook_failed_total (retry counter) - nsi_webhook_dlq_total (DLQ counter) - nsi_webhook_ssrf_rejected_total (SSRF guard counter) - nsi_webhook_delivery_duration_seconds (timer) Headers отправляемые receiver'у: - Content-Type: application/json - User-Agent: Ordinis-Webhook/1.0 - X-Ordinis-Signature: sha256=<hex> - X-Ordinis-Event-Id: <id> - X-Ordinis-Event-Type: <event_type> - X-Ordinis-Scope: <PUBLIC|INTERNAL|RESTRICTED> Ещё впереди: REST API CRUD subscriptions (admin-only), unit tests для HmacSigner/SsrfGuard, admin-ui /webhooks page.
This commit is contained in:
Zimin A.N.
+54
@@ -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.
|
||||
*
|
||||
* <p>Header format: {@code X-Ordinis-Signature: sha256=<hex>}.
|
||||
*/
|
||||
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=<hex>}
|
||||
*/
|
||||
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();
|
||||
}
|
||||
}
|
||||
+99
@@ -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).
|
||||
*
|
||||
* <p>Без этой защиты пользователь с правом создать subscription мог бы
|
||||
* получить blind-SSRF против:
|
||||
* <ul>
|
||||
* <li>cloud metadata service (169.254.169.254 — AWS, GCP, Yandex Cloud)
|
||||
* <li>internal admin UI / Vault / etcd на 127.0.0.1
|
||||
* <li>kubelet API на 10.x внутри pod network
|
||||
* </ul>
|
||||
*
|
||||
* <p>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<String> allowedHosts;
|
||||
|
||||
public SsrfGuard(List<String> 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;
|
||||
}
|
||||
}
|
||||
+265
@@ -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.
|
||||
*
|
||||
* <p>Два @Scheduled tick'а:
|
||||
* <ol>
|
||||
* <li><b>matchTick</b>: scan recent unpublished outbox_events × active
|
||||
* subscriptions. INSERT webhook_deliveries (idempotent ON CONFLICT) для
|
||||
* каждого match. Пишет PENDING rows.
|
||||
* <li><b>sendTick</b>: 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).
|
||||
* </ol>
|
||||
*
|
||||
* <p>Активируется только при {@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<String> 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.
|
||||
*
|
||||
* <p>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<WebhookSubscription> subs = subscriptionRepository.findActive();
|
||||
if (subs.isEmpty()) return;
|
||||
|
||||
// Берём recent events (последние 1000 самых новых). Старше — уже либо
|
||||
// published в Kafka (значит и сюда match'нулись если был active sub
|
||||
// в момент создания), либо в DLQ. Поллер не нужен для catch-up.
|
||||
List<OutboxEvent> 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<WebhookDelivery> 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<WebhookSubscription> 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<OutboxEvent> 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<String> 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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user