diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/WebhookInputSource.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/WebhookInputSource.java index 73887ad9cc..27a215d3af 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/WebhookInputSource.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/WebhookInputSource.java @@ -28,14 +28,22 @@ import stirling.software.proprietary.policy.webhook.WebhookSpool; /** * A push source: documents arrive by signed HTTP POST to {@code /api/v1/webhooks/{webhookId}} and - * are staged by {@link WebhookSpool}. This source then reads that per-webhook spool exactly as - * {@link FolderInputSource} reads a watched directory - each spooled file is a unit of work claimed - * through the {@link ResolveContext} ledger and tracked in place - so the whole claim/consume model - * is reused rather than reinvented. Options (see {@link WebhookConfig}): "mode" is "consume" - * (default: a spooled file is removed once every policy that claimed it has settled successfully) - * or "snapshot" (stateless, every run re-reads whatever is currently spooled). "webhookId" and - * "signingSecret" are minted server-side on create, never supplied by the client. An empty or - * not-yet-created spool means nothing has been delivered, which reads as "verifiably no files". + * are staged for the referencing policies. Where they are staged depends on the source's config + * (see {@link WebhookConfig}): + * + * + * + *

Either way the {@link ResolveContext} ledger does the claiming, "mode" is "consume" (default: + * a staged document is removed once every policy that claimed it has settled successfully) or + * "snapshot" (stateless), and "webhookId"/"signingSecret" are minted server-side on create. Empty + * staging reads as "verifiably no files". */ @Slf4j @Service @@ -47,6 +55,10 @@ public class WebhookInputSource implements InputSource { private final WebhookSpool spool; private final FileReadinessChecker readinessChecker; + // Deliveries staged to an S3 connection are read/claimed/consumed exactly like an S3 source, so + // that whole path is reused rather than reimplemented; a connection-less webhook uses the + // spool. + private final S3InputSource s3InputSource; @Override public String type() { @@ -60,7 +72,28 @@ public class WebhookInputSource implements InputSource { @Override public void validate(InputSpec spec) { - WebhookConfig.from(spec.options()); + WebhookConfig config = WebhookConfig.from(spec.options()); + if (config.usesConnection()) { + // Resolves the connection with the caller present, so save fails if their team can't + // use it (the ownership check the delivery path then trusts). + s3InputSource.validate(stagingSpec(config)); + } + } + + /** + * The S3 view of a webhook's durable staging: its connection plus the reserved per-webhook + * prefix, as an {@code s3} input spec {@link S3InputSource} can resolve directly. + */ + private static InputSpec stagingSpec(WebhookConfig config) { + return new InputSpec( + "s3", + Map.of( + WebhookConfig.CONNECTION_ID_OPTION, + config.connectionId(), + "prefix", + config.stagingPrefix(), + "mode", + config.mode())); } /** @@ -91,6 +124,12 @@ public class WebhookInputSource implements InputSource { @Override public List resolve(InputSpec spec, ResolveContext ctx) throws IOException { WebhookConfig config = WebhookConfig.from(spec.options()); + if (config.usesConnection()) { + // Durable staging: the delivery objects live in the connection's bucket and are read, + // claimed, and consumed by the S3 source (no principal here - the save-time check the + // resolver trusts already ran). + return s3InputSource.resolve(stagingSpec(config), ctx); + } Path dir = spool.dirFor(config.webhookId()); if (!Files.isDirectory(dir)) { // No deliveries yet (or none since the last consume): a verifiably empty source, so the diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookConfig.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookConfig.java index 26d9521351..5ff2f7b176 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookConfig.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookConfig.java @@ -5,21 +5,30 @@ import java.util.Map; /** * Config for a webhook input source, parsed from a spec's options map. A webhook is a push source: * external systems POST documents to a signed URL keyed by {@code webhookId}, and each delivery is - * verified against {@code signingSecret} before it is spooled for the referencing policies. Both + * verified against {@code signingSecret} before it is staged for the referencing policies. Both * {@code webhookId} (the public URL token) and {@code signingSecret} (the HMAC key) are generated - * server-side on create - a client never supplies them - so the receiver's identity and the - * sender's proof of authenticity are always Stirling's own. {@code mode} is "consume" (default: a - * spooled document is deleted once every policy that claimed it has settled successfully) or - * "snapshot" (stateless, every run re-reads whatever is currently spooled). + * server-side on create - a client never supplies them. + * + *

Where deliveries are staged depends on {@code connectionId}: absent, they go to a node-local + * spool directory (self-hosted); present, they go to the S3 {@code IntegrationConfig} connection it + * references, under a reserved per-webhook prefix, so the staging is durable and reachable from any + * node (the model hosted/SaaS needs). Either way {@code mode} is "consume" (default: a staged + * document is removed once every policy that claimed it has settled successfully) or "snapshot" + * (stateless, every run re-reads whatever is currently staged). */ -public record WebhookConfig(String webhookId, String signingSecret, boolean snapshot) { +public record WebhookConfig( + String webhookId, String signingSecret, Long connectionId, boolean snapshot) { public static final String WEBHOOK_ID_OPTION = "webhookId"; public static final String SIGNING_SECRET_OPTION = "signingSecret"; + public static final String CONNECTION_ID_OPTION = "connectionId"; private static final String MODE_OPTION = "mode"; private static final String MODE_CONSUME = "consume"; private static final String MODE_SNAPSHOT = "snapshot"; + /** Reserved key namespace webhook deliveries are staged under in a connection's bucket. */ + private static final String STAGING_ROOT = "stirling-webhook"; + public static WebhookConfig from(Map options) { String webhookId = trimmed(options.get(WEBHOOK_ID_OPTION)); if (webhookId == null) { @@ -37,7 +46,43 @@ public record WebhookConfig(String webhookId, String signingSecret, boolean snap throw new IllegalArgumentException( "webhook config 'mode' must be 'consume' or 'snapshot'"); } - return new WebhookConfig(webhookId, signingSecret, MODE_SNAPSHOT.equals(mode)); + return new WebhookConfig( + webhookId, signingSecret, connectionId(options), MODE_SNAPSHOT.equals(mode)); + } + + /** Whether deliveries are staged to a durable S3 connection rather than the local spool. */ + public boolean usesConnection() { + return connectionId != null; + } + + /** + * The reserved key prefix this webhook's deliveries are staged under inside its connection's + * bucket. Server-derived from the (unguessable) id, so two webhooks sharing a connection never + * collide and an S3 source pointed at the same bucket won't see another webhook's inbox. + */ + public String stagingPrefix() { + return STAGING_ROOT + "/" + webhookId; + } + + /** The mode string as the S3 source expects it. */ + public String mode() { + return snapshot ? MODE_SNAPSHOT : MODE_CONSUME; + } + + private static Long connectionId(Map options) { + Object reference = options.get(CONNECTION_ID_OPTION); + if (reference == null || (reference instanceof String s && s.isBlank())) { + return null; + } + if (reference instanceof Number number) { + return number.longValue(); + } + try { + return Long.valueOf(reference.toString().trim()); + } catch (NumberFormatException e) { + throw new IllegalArgumentException( + "webhook 'connectionId' is not a valid connection reference: " + reference); + } } private static String trimmed(Object value) { @@ -51,6 +96,12 @@ public record WebhookConfig(String webhookId, String signingSecret, boolean snap /** Never prints the signing secret, so an accidental log line cannot leak it. */ @Override public String toString() { - return "WebhookConfig[webhookId=" + webhookId + ", snapshot=" + snapshot + "]"; + return "WebhookConfig[webhookId=" + + webhookId + + ", connectionId=" + + connectionId + + ", snapshot=" + + snapshot + + "]"; } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookReceiverController.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookReceiverController.java index 55904eba32..647c5e2c95 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookReceiverController.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookReceiverController.java @@ -3,6 +3,7 @@ package stirling.software.proprietary.policy.webhook; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; +import java.util.Map; import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; import org.springframework.http.HttpStatus; @@ -25,10 +26,18 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.policy.s3.S3Config; +import stirling.software.proprietary.policy.s3.S3ConnectionPool; +import stirling.software.proprietary.policy.s3.S3ConnectionResolver; import stirling.software.proprietary.policy.source.Source; import stirling.software.proprietary.policy.source.SourceStore; import stirling.software.proprietary.policy.trigger.WebhookTrigger; +import software.amazon.awssdk.core.exception.SdkException; +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; + /** * Public receiver for webhook input sources. External systems POST a document to {@code * /api/v1/webhooks/{webhookId}}; the request is authenticated not by a login session but by an @@ -62,6 +71,8 @@ public class WebhookReceiverController { private final WebhookSpool spool; private final WebhookTrigger webhookTrigger; private final ApplicationProperties applicationProperties; + private final S3ConnectionResolver connectionResolver; + private final S3ConnectionPool connectionPool; @PostMapping("/{webhookId}") @Operation( @@ -97,16 +108,10 @@ public class WebhookReceiverController { throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Empty request body"); } - String storedName; - try { - storedName = - WebhookSpool.displayName( - spool.store(webhookId, filename, body).getFileName().toString()); - } catch (IOException e) { - log.error("Could not spool webhook delivery for {}: {}", webhookId, e.getMessage()); - throw new ResponseStatusException( - HttpStatus.INTERNAL_SERVER_ERROR, "Could not store delivery"); - } + String storedName = + config.usesConnection() + ? stageToConnection(config, filename, body) + : stageToSpool(webhookId, filename, body); // Fire the referencing policies now; the trigger's reconcile is the safety net. webhookTrigger.fireForWebhook(webhookId); @@ -134,6 +139,58 @@ public class WebhookReceiverController { return null; } + /** Stage a delivery to the node-local spool; returns its display (original) name. */ + private String stageToSpool(String webhookId, String filename, byte[] body) { + try { + return WebhookSpool.displayName( + spool.store(webhookId, filename, body).getFileName().toString()); + } catch (IOException e) { + log.error("Could not spool webhook delivery for {}: {}", webhookId, e.getMessage()); + throw new ResponseStatusException( + HttpStatus.INTERNAL_SERVER_ERROR, "Could not store delivery"); + } + } + + /** + * Stage a delivery to the webhook's S3 connection under its reserved prefix. Resolved with no + * principal - the referencing source's right to use the connection was checked when it was + * saved - so a public delivery never needs the caller to hold connection rights. + */ + private String stageToConnection(WebhookConfig config, String filename, byte[] body) { + S3Config s3 = + connectionResolver.resolve( + Map.of( + WebhookConfig.CONNECTION_ID_OPTION, + config.connectionId(), + "prefix", + config.stagingPrefix())); + S3Client client = connectionPool.clientFor(s3); + String key = keyPrefix(s3.prefix()) + WebhookSpool.objectKeySuffix(filename); + try { + client.putObject( + PutObjectRequest.builder().bucket(s3.bucket()).key(key).build(), + RequestBody.fromBytes(body)); + } catch (SdkException e) { + log.error( + "Could not stage webhook delivery for {} to s3://{}/{}: {}", + config.webhookId(), + s3.bucket(), + key, + e.getMessage()); + throw new ResponseStatusException( + HttpStatus.INTERNAL_SERVER_ERROR, "Could not store delivery"); + } + return WebhookSpool.objectDisplayName(filename); + } + + /** The configured prefix as a key-path prefix ("inbox" and "inbox/" mean the same). */ + private static String keyPrefix(String prefix) { + if (prefix == null || prefix.isEmpty() || prefix.endsWith("/")) { + return prefix == null ? "" : prefix; + } + return prefix + "/"; + } + /** * Read the body into memory, capped at {@code policies.webhookMaxBytes}. Reading one byte past * the limit is enough to reject an over-sized (or unbounded, chunked) delivery with 413 before diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookSpool.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookSpool.java index 182943ee8c..323f496af5 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookSpool.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookSpool.java @@ -90,6 +90,23 @@ public class WebhookSpool { return spoolFileName; } + /** + * The object-key suffix for an S3-staged delivery: a unique segment then the sanitised name + * ({@code /}). A subfolder rather than a name prefix so the object's basename - + * and thus the resolved input's filename - stays the clean original name, matching the local + * spool. + */ + public static String objectKeySuffix(String filename) { + return UUID.randomUUID().toString().replace("-", "") + "/" + sanitize(filename); + } + + /** + * The sanitised display name of a staged delivery (the basename of {@link #objectKeySuffix}). + */ + public static String objectDisplayName(String filename) { + return sanitize(filename); + } + /** Reduce a client-supplied filename to a safe bare basename; fall back to a default. */ private static String sanitize(String filename) { if (filename == null) { diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/WebhookInputSourceTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/WebhookInputSourceTest.java index 8cf2ec7d5c..bc01f50676 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/WebhookInputSourceTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/WebhookInputSourceTest.java @@ -6,7 +6,10 @@ import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.io.IOException; import java.nio.file.Files; @@ -21,6 +24,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; @@ -42,6 +46,7 @@ class WebhookInputSourceTest { private static final String WEBHOOK_ID = "testwebhookid1234"; @Mock private FileReadinessChecker readinessChecker; + @Mock private S3InputSource s3InputSource; @TempDir Path tempDir; @@ -53,7 +58,7 @@ class WebhookInputSourceTest { @BeforeEach void setUp() { spool = new WebhookSpool(tempDir.resolve("spool")); - source = new WebhookInputSource(spool, readinessChecker); + source = new WebhookInputSource(spool, readinessChecker, s3InputSource); ledger = new InProcessProcessedLedger(); ctx = new RecordingContext(); lenient().when(readinessChecker.isReady(any())).thenReturn(true); @@ -146,6 +151,53 @@ class WebhookInputSourceTest { assertEquals("keepme", prepared.get("signingSecret")); } + @Test + void aConnectionBackedWebhookDelegatesToTheS3SourceUnderItsReservedPrefix() throws IOException { + when(s3InputSource.resolve(any(), any())).thenReturn(List.of()); + InputSpec spec = + new InputSpec( + "webhook", + Map.of( + "webhookId", + WEBHOOK_ID, + "signingSecret", + "secret", + "mode", + "consume", + "connectionId", + 7)); + + source.resolve(spec, ctx); + + ArgumentCaptor delegated = ArgumentCaptor.forClass(InputSpec.class); + verify(s3InputSource).resolve(delegated.capture(), eq(ctx)); + InputSpec s3 = delegated.getValue(); + assertEquals("s3", s3.type()); + assertEquals(7L, ((Number) s3.options().get("connectionId")).longValue()); + assertEquals("stirling-webhook/" + WEBHOOK_ID, s3.options().get("prefix")); + assertEquals("consume", s3.options().get("mode")); + } + + @Test + void aConnectionBackedWebhookValidatesTheConnectionThroughTheS3Source() { + InputSpec spec = + new InputSpec( + "webhook", + Map.of( + "webhookId", + WEBHOOK_ID, + "signingSecret", + "secret", + "connectionId", + 7)); + + source.validate(spec); + + // Save-time validation defers to the S3 source, which resolves + ownership-checks the + // connection (the check the delivery path later trusts). + verify(s3InputSource).validate(any()); + } + /** Policy-scoped context backed by the in-process ledger, recording presence reports. */ private class RecordingContext implements ResolveContext { diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java index 136c871af9..325b8e8776 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java @@ -28,6 +28,7 @@ import stirling.software.common.util.FileReadinessChecker; import stirling.software.proprietary.policy.config.PolicyAccessGuard; import stirling.software.proprietary.policy.config.PolicyManagementAuthority; import stirling.software.proprietary.policy.input.InputSource; +import stirling.software.proprietary.policy.input.S3InputSource; import stirling.software.proprietary.policy.input.WebhookInputSource; import stirling.software.proprietary.policy.model.OutputSpec; import stirling.software.proprietary.policy.model.PipelineStep; @@ -88,7 +89,10 @@ class SourceControllerTest { properties, List.of(folderInput)); WebhookInputSource webhookInput = - new WebhookInputSource(new WebhookSpool(tempDir), mock(FileReadinessChecker.class)); + new WebhookInputSource( + new WebhookSpool(tempDir), + mock(FileReadinessChecker.class), + mock(S3InputSource.class)); webhookController = new SourceController( sourceStore, diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/webhook/WebhookReceiverControllerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/webhook/WebhookReceiverControllerTest.java index 64ad8cfad9..d0cf31cb2e 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/webhook/WebhookReceiverControllerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/webhook/WebhookReceiverControllerTest.java @@ -23,6 +23,8 @@ import org.springframework.mock.web.MockHttpServletRequest; import org.springframework.web.server.ResponseStatusException; import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.policy.s3.S3ConnectionPool; +import stirling.software.proprietary.policy.s3.S3ConnectionResolver; import stirling.software.proprietary.policy.source.InProcessSourceStore; import stirling.software.proprietary.policy.source.Source; import stirling.software.proprietary.policy.source.SourceStore; @@ -55,7 +57,15 @@ class WebhookReceiverControllerTest { spool = new WebhookSpool(tempDir.resolve("spool")); trigger = mock(WebhookTrigger.class); properties = new ApplicationProperties(); - controller = new WebhookReceiverController(sourceStore, spool, trigger, properties); + // The local-disk tests never touch a connection; mocks stand in for the S3 collaborators. + controller = + new WebhookReceiverController( + sourceStore, + spool, + trigger, + properties, + mock(S3ConnectionResolver.class), + mock(S3ConnectionPool.class)); } private static Source webhookSource(boolean enabled) { diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/webhook/WebhookS3ConnectionMinioTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/webhook/WebhookS3ConnectionMinioTest.java new file mode 100644 index 0000000000..8a50c4cfcf --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/webhook/WebhookS3ConnectionMinioTest.java @@ -0,0 +1,273 @@ +package stirling.software.proprietary.policy.webhook; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Supplier; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.springframework.mock.web.MockHttpServletRequest; +import org.testcontainers.containers.MinIOContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.common.util.FileReadinessChecker; +import stirling.software.proprietary.access.service.OwnershipService; +import stirling.software.proprietary.integration.model.IntegrationConfig; +import stirling.software.proprietary.integration.model.IntegrationType; +import stirling.software.proprietary.integration.repository.IntegrationConfigRepository; +import stirling.software.proprietary.policy.input.ResolveContext; +import stirling.software.proprietary.policy.input.ResolvedInput; +import stirling.software.proprietary.policy.input.S3InputSource; +import stirling.software.proprietary.policy.input.WebhookInputSource; +import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.s3.S3ConnectionPool; +import stirling.software.proprietary.policy.s3.S3ConnectionResolver; +import stirling.software.proprietary.policy.source.InProcessSourceStore; +import stirling.software.proprietary.policy.source.Source; +import stirling.software.proprietary.policy.trigger.WebhookTrigger; +import stirling.software.proprietary.security.service.UserService; + +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.S3Configuration; +import software.amazon.awssdk.services.s3.model.CreateBucketRequest; +import software.amazon.awssdk.services.s3.model.ListObjectsV2Request; +import software.amazon.awssdk.services.s3.model.S3Object; +import tools.jackson.databind.ObjectMapper; + +/** + * End-to-end test of a connection-backed webhook against a real S3 API (MinIO): a signed delivery + * is staged by the receiver into the S3 connection's bucket under the reserved per-webhook prefix, + * then read and consumed by the source (which delegates to {@link S3InputSource}). Proves the + * durable, multi-node staging model hosted deployments use, through the production S3 client + * factory. + */ +@Testcontainers(disabledWithoutDocker = true) +class WebhookS3ConnectionMinioTest { + + private static final String POLICY = "p1"; + private static final String ACCESS_KEY = "minioadmin"; + private static final String SECRET_KEY = "minioadmin"; + private static final String WEBHOOK_ID = "miniowebhookid12"; + private static final String SECRET = "topsecret"; + private static final long CONNECTION_ID = 7L; + + @Container + static MinIOContainer minio = + new MinIOContainer("minio/minio:latest") + .withUserName(ACCESS_KEY) + .withPassword(SECRET_KEY); + + private static S3Client adminClient; + private static int bucketCounter; + + @TempDir Path tempDir; + + private String bucket; + private String stagingPrefix; + private WebhookReceiverController receiver; + private WebhookInputSource inputSource; + private InProcessProcessedLedger ledger; + private RecordingContext ctx; + + @BeforeEach + void setUp() { + if (adminClient == null) { + adminClient = + S3Client.builder() + .endpointOverride(URI.create(minio.getS3URL())) + .httpClient(UrlConnectionHttpClient.create()) + .region(Region.US_EAST_1) + .credentialsProvider( + StaticCredentialsProvider.create( + AwsBasicCredentials.create(ACCESS_KEY, SECRET_KEY))) + .serviceConfiguration( + S3Configuration.builder().pathStyleAccessEnabled(true).build()) + .build(); + } + bucket = "webhook-inbox-" + ++bucketCounter; + adminClient.createBucket(CreateBucketRequest.builder().bucket(bucket).build()); + stagingPrefix = "stirling-webhook/" + WEBHOOK_ID + "/"; + + // The MinIO endpoint resolves to loopback, so the operator opt-in must be on. + ApplicationProperties properties = new ApplicationProperties(); + properties.getPolicies().setAllowPrivateS3Endpoints(true); + + // A resolver over a stored S3 connection pointing at MinIO. No SecurityContext in the test, + // so the resolver's ownership check is skipped (the delivery path's trust-the-save model). + S3ConnectionResolver resolver = resolverFor(minioConnection()); + S3ConnectionPool pool = new S3ConnectionPool(properties); + S3InputSource s3 = new S3InputSource(pool, resolver); + + inputSource = + new WebhookInputSource( + new WebhookSpool(tempDir.resolve("spool")), + mock(FileReadinessChecker.class), + s3); + + InProcessSourceStore sourceStore = new InProcessSourceStore(); + sourceStore.save( + new Source( + "s1", + "Partner uploads", + "webhook", + Map.of( + "webhookId", WEBHOOK_ID, + "signingSecret", SECRET, + "mode", "consume", + "connectionId", CONNECTION_ID), + true, + "owner", + null)); + receiver = + new WebhookReceiverController( + sourceStore, + new WebhookSpool(tempDir.resolve("spool")), + mock(WebhookTrigger.class), + properties, + resolver, + pool); + ledger = new InProcessProcessedLedger(); + ctx = new RecordingContext(); + } + + @Test + void aDeliveryIsStagedToTheConnectionThenReadAndConsumed() throws IOException { + byte[] body = "a pdf".getBytes(StandardCharsets.UTF_8); + String signature = WebhookSignatures.sign(SECRET, body); + + // Deliver: the receiver stages the object into the connection's bucket under the prefix. + var response = receiver.receive(WEBHOOK_ID, signature, "invoice.pdf", request(body)); + assertThat(response.getStatusCode().value()).isEqualTo(202); + + List staged = listUnder(stagingPrefix); + assertThat(staged).hasSize(1); + // Staged under a unique subfolder so the object basename stays the clean original name. + assertThat(staged.get(0).key()).endsWith("/invoice.pdf"); + + // Read: the webhook source resolves the delivery by delegating to the S3 source. + List work = inputSource.resolve(webhookSpec(), ctx); + assertThat(work).hasSize(1); + assertThat(work.get(0).inputs().primary().get(0).getFilename()).isEqualTo("invoice.pdf"); + assertThat(read(work.get(0))).isEqualTo("a pdf"); + // In flight: a second sweep claims nothing. + assertThat(inputSource.resolve(webhookSpec(), ctx)).isEmpty(); + + // Consume: a successful run removes the staged object from the bucket. + work.get(0).onComplete().accept(true); + assertThat(listUnder(stagingPrefix)).isEmpty(); + assertThat(inputSource.resolve(webhookSpec(), ctx)).isEmpty(); + } + + @Test + void aRejectedDeliveryStagesNothing() { + byte[] body = "a pdf".getBytes(StandardCharsets.UTF_8); + + try { + receiver.receive(WEBHOOK_ID, "sha256=deadbeef", "x.pdf", request(body)); + } catch (RuntimeException expected) { + // 401 invalid signature + } + + assertThat(listUnder(stagingPrefix)).isEmpty(); + } + + private IntegrationConfig minioConnection() { + IntegrationConfig connection = new IntegrationConfig(); + connection.setId(CONNECTION_ID); + connection.setIntegrationType(IntegrationType.S3); + connection.setName("minio"); + connection.setEnabled(true); + connection.setConfig( + new ObjectMapper() + .writeValueAsString( + Map.of( + "bucket", bucket, + "region", "us-east-1", + "endpoint", minio.getS3URL(), + "accessKeyId", ACCESS_KEY, + "secretAccessKey", SECRET_KEY))); + return connection; + } + + private static S3ConnectionResolver resolverFor(IntegrationConfig connection) { + IntegrationConfigRepository connections = mock(IntegrationConfigRepository.class); + when(connections.findById(connection.getId())).thenReturn(Optional.of(connection)); + return new S3ConnectionResolver( + connections, mock(OwnershipService.class), mock(UserService.class)); + } + + private static InputSpec webhookSpec() { + return new InputSpec( + "webhook", + Map.of( + "webhookId", WEBHOOK_ID, + "signingSecret", SECRET, + "mode", "consume", + "connectionId", CONNECTION_ID)); + } + + private static MockHttpServletRequest request(byte[] body) { + MockHttpServletRequest req = + new MockHttpServletRequest("POST", "/api/v1/webhooks/" + WEBHOOK_ID); + req.setContent(body); + return req; + } + + private List listUnder(String prefix) { + return adminClient + .listObjectsV2(ListObjectsV2Request.builder().bucket(bucket).prefix(prefix).build()) + .contents(); + } + + private static String read(ResolvedInput unit) throws IOException { + try (InputStream stream = unit.inputs().primary().get(0).getInputStream()) { + return new String(stream.readAllBytes(), StandardCharsets.UTF_8); + } + } + + private class RecordingContext implements ResolveContext { + + private final List present = new ArrayList<>(); + + @Override + public boolean claim(String identity, String gate, Supplier contentHash) { + return ledger.claim(POLICY, identity, gate, contentHash); + } + + @Override + public void settle( + String identity, String finalGate, String finalContentHash, boolean success) { + ledger.settle(POLICY, identity, finalGate, finalContentHash, success); + } + + @Override + public boolean allSettledDone(String identity) { + return ledger.allSettledDone(identity); + } + + @Override + public void reportPresent(Collection identities) { + present.addAll(identities); + } + } +} diff --git a/frontend/editor/public/locales/en-US/translation.toml b/frontend/editor/public/locales/en-US/translation.toml index 73ce2cec2b..edb5fee042 100644 --- a/frontend/editor/public/locales/en-US/translation.toml +++ b/frontend/editor/public/locales/en-US/translation.toml @@ -8147,6 +8147,10 @@ label = "Webhook" deliveryUrl = "Delivery URL" secretNote = "The signing secret is shown once, when the webhook is created. Recreate the source to roll it." +[portal.sources.types.webhook.fields.connection] +helperText = "Choose an S3 connection to stage deliveries durably (required in hosted). Leave blank on a self-hosted server to stage them on local disk." +label = "Storage connection" + [portal.sources.types.webhook.fields.mode] helperText = "Consume removes each delivered document once every policy has processed it." label = "Read mode" diff --git a/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.test.ts b/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.test.ts index 83354cd44d..b22587fc71 100644 --- a/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.test.ts +++ b/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.test.ts @@ -3,10 +3,17 @@ import { describe, expect, it } from "vitest"; import { creatableSourceTypes } from "@portal/components/sources/creatableSourceTypes"; describe("creatableSourceTypes (SaaS)", () => { - it("never offers server-local-disk sources (folder, webhook) in hosted deployments", () => { - const offered = creatableSourceTypes().map((t) => t.type); - expect(offered).not.toContain("folder"); - expect(offered).not.toContain("webhook"); + it("never offers folder sources: hosted deployments do not read the server filesystem", () => { + expect(creatableSourceTypes().map((t) => t.type)).not.toContain("folder"); + }); + + it("offers webhook with a required S3 connection (durable staging in hosted)", () => { + const webhook = creatableSourceTypes().find((t) => t.type === "webhook"); + expect(webhook).toBeDefined(); + const connection = webhook?.fields.find( + (f) => f.control === "s3Connection", + ); + expect(connection?.required).toBe(true); }); it("still offers the cloud source types", () => { diff --git a/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.ts b/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.ts index 7415b20fa9..276ef09e38 100644 --- a/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.ts +++ b/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.ts @@ -1,19 +1,31 @@ import { CREATABLE_SOURCE_TYPES, + WEBHOOK_SOURCE_TYPE, type CreatableSourceType, } from "@portal/components/sources/sourceTypes"; /** - * Hosted deployments don't rely on the server's local filesystem, so the two - * server-local-disk source types are not offered in the connect wizard: folder - * (denied outright by the backend's FolderAccessGuard) and webhook (whose - * delivery spool is node-local, so it can't be relied on across a multi-node - * fleet). Cloud sources such as S3 remain. + * Hosted deployments don't rely on the server's local filesystem. Folder sources + * are dropped outright (denied by the backend's FolderAccessGuard). Webhook stays + * - it works in hosted by staging deliveries to a durable S3 connection - but its + * connection becomes required, since the node-local fallback can't be relied on + * across a multi-node fleet. Cloud sources such as S3 are unchanged. */ -const SERVER_LOCAL_TYPES = new Set(["folder", "webhook"]); - export function creatableSourceTypes(): CreatableSourceType[] { - return CREATABLE_SOURCE_TYPES.filter( - (type) => !SERVER_LOCAL_TYPES.has(type.type), + return CREATABLE_SOURCE_TYPES.filter((type) => type.type !== "folder").map( + (type) => + type.type === WEBHOOK_SOURCE_TYPE ? withRequiredConnection(type) : type, ); } + +/** The type with its S3-connection field marked required (durable staging is mandatory). */ +function withRequiredConnection( + type: CreatableSourceType, +): CreatableSourceType { + return { + ...type, + fields: type.fields.map((field) => + field.control === "s3Connection" ? { ...field, required: true } : field, + ), + }; +} diff --git a/frontend/editor/src/portal/components/sources/sourceTypes.ts b/frontend/editor/src/portal/components/sources/sourceTypes.ts index c1f6049188..c73ac1078b 100644 --- a/frontend/editor/src/portal/components/sources/sourceTypes.ts +++ b/frontend/editor/src/portal/components/sources/sourceTypes.ts @@ -22,6 +22,9 @@ export interface SourceTypeMeta { */ export const EDITOR_SOURCE_TYPE = "editor"; +/** The webhook source type. Its delivery URL + signing secret are minted server-side on create. */ +export const WEBHOOK_SOURCE_TYPE = "webhook"; + const SOURCE_TYPE_META: Record = { folder: { labelKey: "portal.sources.types.folder.label", @@ -38,6 +41,11 @@ const SOURCE_TYPE_META: Record = { icon: "☁", accent: "brand", }, + webhook: { + labelKey: "portal.sources.types.webhook.label", + icon: "↯", + accent: "warning", + }, }; const UNKNOWN_TYPE_META: SourceTypeMeta = { @@ -180,6 +188,42 @@ export const CREATABLE_SOURCE_TYPES: CreatableSourceType[] = [ }, ], }, + { + // The delivery URL + signing secret are minted server-side on create (revealed once). An S3 + // connection makes staging durable + multi-node (required in hosted); left blank, deliveries + // stage to the server's local disk (self-hosted only). + type: WEBHOOK_SOURCE_TYPE, + labelKey: "portal.sources.types.webhook.label", + descriptionKey: "portal.sources.types.webhook.description", + fields: [ + { + key: "connectionId", + labelKey: "portal.sources.types.webhook.fields.connection.label", + control: "s3Connection", + helperTextKey: + "portal.sources.types.webhook.fields.connection.helperText", + }, + { + key: "mode", + labelKey: "portal.sources.types.webhook.fields.mode.label", + control: "select", + defaultValue: "consume", + helperTextKey: "portal.sources.types.webhook.fields.mode.helperText", + options: [ + { + value: "consume", + labelKey: + "portal.sources.types.webhook.fields.mode.options.consume", + }, + { + value: "snapshot", + labelKey: + "portal.sources.types.webhook.fields.mode.options.snapshot", + }, + ], + }, + ], + }, ]; /** Default option values for a type's create form. */ diff --git a/frontend/editor/src/portal/views/SourceBuilder.css b/frontend/editor/src/portal/views/SourceBuilder.css index 985332046d..4b618618bf 100644 --- a/frontend/editor/src/portal/views/SourceBuilder.css +++ b/frontend/editor/src/portal/views/SourceBuilder.css @@ -85,3 +85,27 @@ justify-content: flex-end; gap: 0.5rem; } + +/* Webhook delivery URL / signing secret: read-only value + copy button. */ +.portal-source-builder__copy-row { + display: flex; + gap: 0.5rem; + align-items: center; +} + +.portal-source-builder__copy-row > :first-child { + flex: 1 1 auto; + min-width: 0; +} + +.portal-source-builder__reveal { + display: flex; + flex-direction: column; + gap: 1rem; +} + +.portal-source-builder__muted { + margin: 0; + font-size: 0.8125rem; + color: var(--color-text-4); +} diff --git a/frontend/editor/src/portal/views/SourceBuilder.test.tsx b/frontend/editor/src/portal/views/SourceBuilder.test.tsx index 1f9b3341f8..29ec08b172 100644 --- a/frontend/editor/src/portal/views/SourceBuilder.test.tsx +++ b/frontend/editor/src/portal/views/SourceBuilder.test.tsx @@ -102,6 +102,40 @@ describe("SourceBuilder", () => { ).toBeDisabled(); }); + it("reveals the delivery URL and signing secret once after creating a webhook", async () => { + createSource.mockResolvedValue({ + id: "wh-1", + options: { webhookId: "whk_abc123", signingSecret: "whsec_topsecret" }, + }); + renderBuilder("/processor/sources/new"); + + fireEvent.change(screen.getByLabelText(/portal\.sources\.wizard\.name/), { + target: { value: "Partner uploads" }, + }); + // Webhook's connection is optional (self-hosted local-disk), so a name is enough to create. + fireEvent.click(screen.getByText("portal.sources.types.webhook.label")); + fireEvent.click(screen.getByText("portal.sources.builder.create")); + + await waitFor(() => expect(createSource).toHaveBeenCalledTimes(1)); + expect(createSource).toHaveBeenCalledWith( + expect.objectContaining({ type: "webhook", name: "Partner uploads" }), + ); + + // The reveal shows the secret + URL and stays put until Done is clicked. + expect( + await screen.findByDisplayValue("whsec_topsecret"), + ).toBeInTheDocument(); + expect( + screen.getByDisplayValue(/\/api\/v1\/webhooks\/whk_abc123$/), + ).toBeInTheDocument(); + expect(screen.queryByText("sources list")).not.toBeInTheDocument(); + + fireEvent.click( + screen.getByText("portal.sources.types.webhook.reveal.done"), + ); + expect(await screen.findByText("sources list")).toBeInTheDocument(); + }); + it("blocks create until required fields are filled", async () => { renderBuilder("/processor/sources/new"); // Name given but directory (required) still blank -> Create disabled. diff --git a/frontend/editor/src/portal/views/SourceBuilder.tsx b/frontend/editor/src/portal/views/SourceBuilder.tsx index 31b8550354..dff0a31184 100644 --- a/frontend/editor/src/portal/views/SourceBuilder.tsx +++ b/frontend/editor/src/portal/views/SourceBuilder.tsx @@ -27,11 +27,17 @@ import { CREATABLE_SOURCE_TYPES, defaultOptions, sourceTypeMeta, + WEBHOOK_SOURCE_TYPE, type CreatableSourceType, } from "@portal/components/sources/sourceTypes"; import { S3ConnectionPicker } from "@portal/components/sources/S3ConnectionPicker"; import "@portal/views/SourceBuilder.css"; +/** Absolute delivery URL a sender POSTs to for a webhook's routing id. */ +function webhookUrl(webhookId: string): string { + return `${window.location.origin}/api/v1/webhooks/${webhookId}`; +} + const OFFERED_TYPES = creatableSourceTypes(); /** A source's stored type resolved to its create-form metadata (edit falls back to any type). */ @@ -84,6 +90,11 @@ export function SourceBuilder() { const [error, setError] = useState(null); const [pendingDelete, setPendingDelete] = useState(false); const [deleting, setDeleting] = useState(false); + // Set after a webhook is created: its one-time delivery id + signing secret, shown before leaving. + const [reveal, setReveal] = useState<{ + webhookId: string; + secret: string; + } | null>(null); // Seed once: immediately for a new source, or after the record loads for edit. useEffect(() => { @@ -112,18 +123,42 @@ export function SourceBuilder() { ); const canSave = name.trim() !== "" && requiredComplete && !submitting; + // An existing webhook's delivery URL, shown read-only in the edit view (the secret is not - it + // is revealed only once, at creation). + const editingWebhookId = + isEdit && sourceState.data?.type === WEBHOOK_SOURCE_TYPE + ? String(sourceState.data.options?.webhookId ?? "") + : ""; + const revealUrl = reveal ? webhookUrl(reveal.webhookId) : ""; + const revealSecret = reveal ? reveal.secret : ""; + + function dismissReveal() { + setReveal(null); + navigate(listPath); + } + async function save() { if (!canSave) return; setSubmitting(true); setError(null); try { - await createSource({ + const saved = await createSource({ id: isEdit ? id : undefined, name: name.trim(), type: type.type, options, enabled, }); + // A new webhook returns its server-minted routing id + signing secret once; reveal them + // (with the delivery URL) before leaving so the operator can copy the secret. + if (!isEdit && type.type === WEBHOOK_SOURCE_TYPE) { + const webhookId = String(saved.options?.webhookId ?? ""); + const secret = String(saved.options?.signingSecret ?? ""); + if (webhookId && secret) { + setReveal({ webhookId, secret }); + return; + } + } navigate(listPath); } catch (e) { setError(errorMessage(e)); @@ -131,6 +166,10 @@ export function SourceBuilder() { } } + function copy(text: string) { + void navigator.clipboard?.writeText(text); + } + async function confirmDelete() { if (!id || deleting) return; setDeleting(true); @@ -290,6 +329,28 @@ export function SourceBuilder() { ))} + {editingWebhookId && ( + +

+ e.currentTarget.select()} + /> + +
+ + )} + {error && } @@ -321,6 +382,69 @@ export function SourceBuilder() { >

{t("portal.sources.delete.body", { name })}

+ + + + + } + > + {reveal && ( +
+ + +
+ e.currentTarget.select()} + /> + +
+ + +
+ e.currentTarget.select()} + /> + +
+ +

+ {t("portal.sources.types.webhook.reveal.usage")} +

+
+ )} + ); }