Compare commits

..
Author SHA1 Message Date
Anthony Stirling 7f6bc5e956 Remove low-value narration comments 2026-07-13 20:33:56 +01:00
Anthony Stirling cb40d7778f Condense comments to single lines 2026-07-13 20:28:07 +01:00
Anthony Stirling 8cb03dba9e Align webhook source with S3 connections 2026-07-13 19:54:15 +01:00
Anthony Stirling 247b783c75 Merge remote-tracking branch 'origin/main' into claude/stirling-sources-setup-fe5517
# Conflicts:
#	frontend/editor/src/portal/components/sources/ConnectWizard.test.tsx
#	frontend/editor/src/portal/components/sources/ConnectWizard.tsx
#	frontend/editor/src/portal/components/sources/SourceDetailPanel.tsx
2026-07-13 16:43:36 +01:00
Anthony Stirling a9f7add87d Add webhook policy source 2026-07-13 16:40:48 +01:00
59 changed files with 2127 additions and 1531 deletions
+1 -1
View File
@@ -1,6 +1,6 @@
# Maintainer: Stirling PDF Inc <contact@stirlingpdf.com>
pkgname=stirling-pdf-desktop
pkgver=2.14.2
pkgver=2.14.1
pkgrel=1
pkgdesc="Locally hosted, web-based PDF manipulation tool (Tauri desktop app, official Stirling PDF Inc build)"
arch=('x86_64')
+1 -1
View File
@@ -1,6 +1,6 @@
# Maintainer: Stirling PDF Inc <contact@stirlingpdf.com>
pkgname=stirling-pdf-server-bin
pkgver=2.14.2
pkgver=2.14.1
pkgrel=1
pkgdesc="Locally hosted, web-based PDF manipulation tool (server JAR, prebuilt)"
arch=('any')
-67
View File
@@ -308,70 +308,3 @@ jobs:
name: playwright-report-enterprise-${{ github.run_id }}
path: frontend/playwright-report/
retention-days: 7
# Multi-node regression: builds + seeds the clustered stack (testing/compose/docker-compose-multinode.yml)
# and runs behave features/multinode. Licence-gated, so it runs after the Playwright job (not in parallel).
multinode-e2e:
needs: [pick, playwright-e2e-enterprise]
# Nightly cron + manual dispatch only (heavy build), fork-gated for the licence secret.
if: >-
always() && needs.pick.outputs.is_fork != 'true'
&& (github.event_name == 'schedule' || github.event_name == 'workflow_dispatch')
runs-on: ${{ needs.pick.outputs.is_fork == 'true' && 'ubuntu-latest' || format('depot-ubuntu-24.04-{0}', inputs.depot_cores || '8') }}
timeout-minutes: 60
env:
PREMIUM_KEY: ${{ secrets.PREMIUM_KEY_ENTERPRISE }}
PREMIUM_ENABLED: "true"
SYSTEM_ENABLEANALYTICS: "false"
DEPOT_TOKEN: ${{ secrets.DEPOT_TOKEN }}
MN_COMPOSE: docker-compose-multinode.yml
steps:
- name: Harden Runner
uses: step-security/harden-runner@ab7a9404c0f3da075243ca237b5fac12c98deaa5 # v2.19.3
with:
egress-policy: audit
- name: Checkout repository
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
- name: Set up Python
uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
with:
python-version: "3.12"
cache: "pip"
cache-dependency-path: ./testing/cucumber/requirements.txt
- name: Install behave test deps
run: |
pip install --require-hashes --only-binary=:all: -r ./testing/cucumber/requirements.txt
- name: Build the multi-node image
working-directory: testing/compose
run: docker compose -f "$MN_COMPOSE" build
- name: Bring up the cluster and wait for both nodes healthy
working-directory: testing/compose
run: |
docker compose -f "$MN_COMPOSE" up -d
for i in $(seq 1 90); do
h1=$(docker inspect -f '{{.State.Health.Status}}' multinode-stirling-1 2>/dev/null || echo starting)
h2=$(docker inspect -f '{{.State.Health.Status}}' multinode-stirling-2 2>/dev/null || echo starting)
if [ "$h1" = healthy ] && [ "$h2" = healthy ]; then echo "both nodes healthy"; exit 0; fi
sleep 5
done
echo "::error::nodes did not become healthy"
docker compose -f "$MN_COMPOSE" logs --tail=200 stirling-1 stirling-2
exit 1
- name: Seed the cluster (teams, users, S3 connection, policy)
working-directory: testing/compose
run: docker compose -f "$MN_COMPOSE" --profile seed run --rm seed
- name: Run multi-node regression (implemented guarantees)
working-directory: testing/cucumber
# -e overrides behave.ini's exclusion of features/multinode; ~@known_gap skips any tracked-gap scenarios.
run: python -m behave features/multinode -e "features/enterprise" --tags="~@known_gap ~@destructive" --no-capture -f plain
- name: Run multi-node failover (destructive)
working-directory: testing/cucumber
run: python -m behave features/multinode -e "features/enterprise" --tags="@destructive ~@known_gap" --no-capture -f plain
- name: Dump node logs on failure
if: failure()
working-directory: testing/compose
run: docker compose -f "$MN_COMPOSE" logs --tail=400 stirling-1 stirling-2
- name: Tear down
if: always()
working-directory: testing/compose
run: docker compose -f "$MN_COMPOSE" --profile seed down -v --remove-orphans
+6 -23
View File
@@ -510,7 +510,6 @@ jobs:
# cargo output unsigned, so checking it produces false negatives.
- name: Verify Windows Code Signature
if: ${{ matrix.platform == 'windows-latest' && env.SM_API_KEY != '' && (github.event_name == 'release' || (github.event_name == 'workflow_dispatch' && github.event.inputs.sign != 'false') || github.ref == 'refs/heads/V2-master') }}
timeout-minutes: 15
shell: pwsh
run: |
$allSigned = $true
@@ -532,26 +531,11 @@ jobs:
# Extract MSI and verify the inner exe (the file that actually gets installed).
# This is the critical check - AV flags the installed exe at runtime.
# Use lessmsi, not `msiexec /a`: msiexec serializes on the global
# _MSIExecute mutex and hangs forever on hosted runners when another
# installer is busy. lessmsi reads MSI tables directly - no mutex, no service.
$msi = $msiFiles[0].FullName
$extractDir = Join-Path $env:RUNNER_TEMP "msi-verify"
if (Test-Path $extractDir) { Remove-Item $extractDir -Recurse -Force }
New-Item -ItemType Directory -Force -Path $extractDir | Out-Null
choco install lessmsi -y --no-progress --limit-output | Out-Null
# Bound the extraction and kill on hang (defence in depth over timeout-minutes).
$proc = Start-Process lessmsi -ArgumentList 'x', "`"$msi`"", "`"$extractDir\`"" -PassThru -NoNewWindow
if (-not $proc.WaitForExit(120000)) {
try { $proc.Kill() } catch {}
Write-Host "[ERROR] MSI extraction timed out after 120s"
$allSigned = $false
} elseif ($proc.ExitCode -ne 0) {
Write-Host "[ERROR] Failed to extract MSI for verification (exit code: $($proc.ExitCode))"
$allSigned = $false
} else {
$proc = Start-Process msiexec.exe -ArgumentList '/a', $msi, '/qn', "TARGETDIR=$extractDir" -Wait -PassThru -NoNewWindow
if ($proc.ExitCode -eq 0) {
$innerExe = Get-ChildItem -Path $extractDir -Filter "stirling-pdf.exe" -Recurse -File | Select-Object -First 1
if ($innerExe) {
$sig = Get-AuthenticodeSignature -FilePath $innerExe.FullName
@@ -564,6 +548,9 @@ jobs:
Write-Host "[ERROR] Could not find stirling-pdf.exe inside MSI"
$allSigned = $false
}
} else {
Write-Host "[ERROR] Failed to extract MSI for verification (exit code: $($proc.ExitCode))"
$allSigned = $false
}
if (-not $allSigned) {
@@ -813,11 +800,7 @@ jobs:
uses: softprops/action-gh-release@b4309332981a82ec1c5618f44dd2e27cc8bfbfda # v3.0.0
with:
tag_name: v${{ needs.determine-matrix.outputs.version }}
# Don't regenerate/append notes on re-runs, and don't force this into the
# "Latest" slot - leave the release body and latest marker as they are.
generate_release_notes: false
append_body: false
make_latest: false
generate_release_notes: true
fail_on_unmatched_files: true
# Installers + updater payloads + manifest. .sig contents are embedded
# in latest.json so the .sig files themselves are not uploaded.
@@ -254,6 +254,9 @@ public class ApplicationProperties {
* in-network object store.
*/
private boolean allowPrivateS3Endpoints = false;
/** Max inbound webhook body (bytes); over it is rejected 413. Default 100 MB. */
private long webhookMaxBytes = 104857600L;
}
@Data
@@ -198,6 +198,8 @@ public class RequestUriUtils {
|| trimmedUri.startsWith("/readiness")
|| trimmedUri.startsWith(
"/api/v1/mobile-scanner/") // Mobile scanner endpoints (no auth)
// Webhook receiver: authenticated per-request by HMAC signature, not a session.
|| trimmedUri.startsWith("/api/v1/webhooks/")
|| trimmedUri.startsWith("/v1/api-docs")
// Workflow participant endpoints - access controlled by share tokens, not login
|| trimmedUri.startsWith("/api/v1/workflow/participant/")
@@ -176,6 +176,13 @@ class RequestUriUtilsTest {
assertFalse(RequestUriUtils.isPublicAuthEndpoint("/api/v1/convert", ""));
}
@Test
void testIsPublicAuthEndpoint_webhookReceiver() {
// The webhook source receiver authenticates each delivery by HMAC signature, not a session.
assertTrue(RequestUriUtils.isPublicAuthEndpoint("/api/v1/webhooks/whk_abc123", ""));
assertTrue(RequestUriUtils.isPublicAuthEndpoint("/app/api/v1/webhooks/whk_abc123", "/app"));
}
@Test
void testIsPublicAuthEndpoint_withContextPath() {
assertTrue(RequestUriUtils.isPublicAuthEndpoint("/app/login", "/app"));
@@ -3,6 +3,7 @@ package stirling.software.proprietary.policy.input;
import java.io.IOException;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import stirling.software.proprietary.policy.model.InputSpec;
@@ -22,6 +23,12 @@ public interface InputSource {
/** Throws {@link IllegalArgumentException} on bad config. Called on save to fail fast. */
default void validate(InputSpec spec) {}
/** Normalise a source's options before persistence (default no-op); must not mutate the arg. */
default Map<String, Object> prepareOptionsForSave(
Map<String, Object> options, boolean isCreate) {
return options;
}
/**
* Resolve the spec into zero or more units of work, each carrying one run's files and a
* completion hook. Empty list means nothing to run right now. Discovery is read-only - files
@@ -0,0 +1,191 @@
package stirling.software.proprietary.policy.input;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.stereotype.Service;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import stirling.software.common.util.FileReadinessChecker;
import stirling.software.proprietary.policy.ledger.FolderIdentities;
import stirling.software.proprietary.policy.model.InputSpec;
import stirling.software.proprietary.policy.model.PolicyInputs;
import stirling.software.proprietary.policy.webhook.WebhookConfig;
import stirling.software.proprietary.policy.webhook.WebhookIds;
import stirling.software.proprietary.policy.webhook.WebhookSpool;
/** Push source: deliveries staged to an S3 connection or the local spool, read via the ledger. */
@Slf4j
@Service
@RequiredArgsConstructor
@ConditionalOnBooleanProperty(name = "policies.enabled")
public class WebhookInputSource implements InputSource {
static final String TYPE = "webhook";
private final WebhookSpool spool;
private final FileReadinessChecker readinessChecker;
// Connection-backed staging delegates to the S3 source; else the local spool.
private final S3InputSource s3InputSource;
@Override
public String type() {
return TYPE;
}
@Override
public boolean supports(InputSpec spec) {
return spec != null && TYPE.equals(spec.type());
}
@Override
public void validate(InputSpec spec) {
WebhookConfig config = WebhookConfig.from(spec.options());
if (config.usesConnection()) {
// Resolve with the caller present so save fails if they can't use the connection.
s3InputSource.validate(stagingSpec(config));
}
}
/** The webhook's durable staging as an {@code s3} input spec for {@link S3InputSource}. */
private static InputSpec stagingSpec(WebhookConfig config) {
return new InputSpec(
"s3",
Map.of(
WebhookConfig.CONNECTION_ID_OPTION,
config.connectionId(),
"prefix",
config.stagingPrefix(),
"mode",
config.mode()));
}
/** Mint the routing id + signing secret on create; an existing webhook is left untouched. */
@Override
public Map<String, Object> prepareOptionsForSave(
Map<String, Object> options, boolean isCreate) {
boolean hasId =
options.get(WebhookConfig.WEBHOOK_ID_OPTION) != null
&& !options.get(WebhookConfig.WEBHOOK_ID_OPTION).toString().isBlank();
if (!isCreate && hasId) {
return options;
}
Map<String, Object> prepared = new LinkedHashMap<>(options);
if (!hasId) {
prepared.put(WebhookConfig.WEBHOOK_ID_OPTION, WebhookIds.newWebhookId());
}
Object secret = prepared.get(WebhookConfig.SIGNING_SECRET_OPTION);
if (secret == null || secret.toString().isBlank()) {
prepared.put(WebhookConfig.SIGNING_SECRET_OPTION, WebhookIds.newSigningSecret());
}
return prepared;
}
@Override
public List<ResolvedInput> resolve(InputSpec spec, ResolveContext ctx) throws IOException {
WebhookConfig config = WebhookConfig.from(spec.options());
if (config.usesConnection()) {
// Durable staging read by the S3 source (no principal; the save-time check is trusted).
return s3InputSource.resolve(stagingSpec(config), ctx);
}
Path dir = spool.dirFor(config.webhookId());
if (!Files.isDirectory(dir)) {
// No deliveries yet: verifiably empty (a missing dir is normal here, not an error).
ctx.reportPresent(List.of());
return List.of();
}
Path canonicalDir = FolderIdentities.canonicalDir(dir);
List<Path> present = listFiles(dir);
if (config.snapshot()) {
List<ResolvedInput> work = new ArrayList<>();
for (Path file : present) {
if (readinessChecker.isReady(file)) {
work.add(ResolvedInput.of(PolicyInputs.of(List.of(fileResource(file)))));
}
}
return work;
}
ctx.reportPresent(
present.stream()
.map(file -> FolderIdentities.identity(canonicalDir, dir, file))
.toList());
List<ResolvedInput> work = new ArrayList<>();
for (Path file : present) {
if (!readinessChecker.isReady(file)) {
continue;
}
String identity = FolderIdentities.identity(canonicalDir, dir, file);
String gate;
boolean claimed;
try {
gate = FolderIdentities.statGate(file);
claimed = ctx.claim(identity, gate, null);
} catch (IOException | UncheckedIOException e) {
log.debug("Could not read {} for its version: {}", file, e.getMessage());
continue; // vanished or unreadable mid-sweep; the next sweep sees the truth
}
if (!claimed) {
continue;
}
work.add(
new ResolvedInput(
PolicyInputs.of(List.of(fileResource(file))),
success -> completeConsumed(ctx, identity, file, gate, success)));
}
return work;
}
/** Settle, then delete when unchanged and every claimant settled DONE (consensus). */
private static void completeConsumed(
ResolveContext ctx, String identity, Path file, String claimGate, boolean success) {
ctx.settle(identity, claimGate, null, success);
if (!success) {
return;
}
try {
if (FolderIdentities.statGate(file).equals(claimGate) && ctx.allSettledDone(identity)) {
Files.deleteIfExists(file);
}
} catch (java.nio.file.NoSuchFileException alreadyGone) {
// Removed by the user or a co-watching policy's own consensus delete: nothing to do.
} catch (IOException e) {
log.warn("Could not remove consumed webhook delivery {}: {}", file, e.getMessage());
}
}
/** Every non-hidden regular file currently spooled for the webhook. */
private static List<Path> listFiles(Path dir) throws IOException {
List<Path> files = new ArrayList<>();
try (Stream<Path> entries = Files.list(dir)) {
entries.filter(Files::isRegularFile)
.filter(file -> !file.getFileName().toString().startsWith("."))
.forEach(files::add);
}
return files;
}
private static Resource fileResource(Path path) {
String name = WebhookSpool.displayName(path.getFileName().toString());
return new FileSystemResource(path.toFile()) {
@Override
public String getFilename() {
return name;
}
};
}
}
@@ -2,6 +2,7 @@ package stirling.software.proprietary.policy.source;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
import org.springframework.http.HttpStatus;
@@ -46,6 +47,8 @@ import stirling.software.proprietary.util.SecretMasker;
@ConditionalOnBooleanProperty(name = "policies.enabled")
public class SourceController {
private static final String WEBHOOK_TYPE = "webhook";
private final SourceStore sourceStore;
private final SourceAccessGuard sourceAccessGuard;
private final SourceOverviewService overviewService;
@@ -109,7 +112,8 @@ public class SourceController {
public ResponseEntity<Source> save(@RequestBody Source source) {
requireSourceEditingAllowed();
requireNotEditor(source.id(), source.type());
Source owned = withStoredSecrets(resolveOwnership(source));
boolean isCreate = source.id() == null || source.id().isBlank();
Source owned = withPreparedOptions(withStoredSecrets(resolveOwnership(source)), isCreate);
try {
validateConfig(owned);
} catch (IllegalArgumentException e) {
@@ -119,7 +123,7 @@ public class SourceController {
// An edited folder source can change which directory needs watching, so re-sync trigger
// registrations now instead of waiting for the next reconcile.
policyTriggerManager.notifyPoliciesChanged();
return ResponseEntity.ok(withMaskedSecrets(saved));
return ResponseEntity.ok(revealOnCreate(saved, isCreate));
}
@DeleteMapping("/{sourceId}")
@@ -221,14 +225,36 @@ public class SourceController {
/** Validate the config against the bean that handles the source's type, as the engine will. */
private void validateConfig(Source source) {
InputSpec spec = source.toInputSpec();
inputSources.stream()
.filter(inputSource -> inputSource.supports(spec))
.findFirst()
inputSourceFor(spec)
.orElseThrow(
() -> new IllegalArgumentException("unknown source type: " + source.type()))
.validate(spec);
}
/** Let the source type populate server-owned config before persistence. */
private Source withPreparedOptions(Source source, boolean isCreate) {
InputSpec spec = source.toInputSpec();
InputSource input = inputSourceFor(spec).orElse(null);
if (input == null) {
return source;
}
Map<String, Object> prepared = input.prepareOptionsForSave(source.options(), isCreate);
// Treat a null return as "unchanged" (else the record wipes config to empty).
return prepared == null ? source : withOptions(source, prepared);
}
/** Reveal a webhook's minted secret once on the create response; other reads mask. */
private static Source revealOnCreate(Source saved, boolean isCreate) {
if (isCreate && WEBHOOK_TYPE.equals(saved.type())) {
return saved;
}
return withMaskedSecrets(saved);
}
private Optional<InputSource> inputSourceFor(InputSpec spec) {
return inputSources.stream().filter(input -> input.supports(spec)).findFirst();
}
/**
* Editing sources requires the editor role for the caller's team (a team leader on SaaS), the
* same rule as policies. Single-user deployments (login disabled) trust the local operator.
@@ -102,7 +102,8 @@ public class SourceOverviewService {
List.of(),
docs.total(),
docs.last24h(),
docs.last30d());
docs.last30d(),
null);
}
/**
@@ -143,7 +144,17 @@ public class SourceOverviewService {
configRows(source),
docs.total(),
docs.last24h(),
docs.last30d());
docs.last30d(),
webhookPath(source));
}
/** The server-relative delivery path for a webhook source, else null. Never a secret. */
private static String webhookPath(Source source) {
if (!"webhook".equals(source.type())) {
return null;
}
Object webhookId = source.options().get("webhookId");
return webhookId == null ? null : "/api/v1/webhooks/" + webhookId;
}
/** A disabled (paused) source reads as "disabled"; an unreferenced one reads as "unused". */
@@ -2,11 +2,7 @@ package stirling.software.proprietary.policy.source;
import java.util.List;
/**
* One row in the Sources overview: a persisted input connection shown exactly once, with how many
* policies reference it (and which) and how many documents it has fed into runs ({@code docsTotal}
* lifetime plus the trailing 24-hour and 30-day windows).
*/
/** One Sources-overview row; {@code webhookPath} is a webhook's delivery path, else null. */
public record SourceView(
String id,
String name,
@@ -17,7 +13,8 @@ public record SourceView(
List<DetailRow> config,
long docsTotal,
long docs24h,
long docs30d) {
long docs30d,
String webhookPath) {
/** A policy that references this source. */
public record PolicyRef(String id, String name) {}
@@ -0,0 +1,137 @@
package stirling.software.proprietary.policy.trigger;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
import org.springframework.stereotype.Service;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import stirling.software.common.model.ApplicationProperties;
import stirling.software.proprietary.policy.engine.PolicyRunner;
import stirling.software.proprietary.policy.engine.SweepKind;
import stirling.software.proprietary.policy.model.Policy;
import stirling.software.proprietary.policy.source.Source;
import stirling.software.proprietary.policy.source.SourceStore;
import stirling.software.proprietary.policy.store.PolicyStore;
import stirling.software.proprietary.policy.webhook.WebhookConfig;
/** Fires webhook policies on delivery plus a periodic reconcile safety net. */
@Slf4j
@Service
@RequiredArgsConstructor
@ConditionalOnBooleanProperty(name = "policies.enabled")
public class WebhookTrigger implements PolicyTrigger {
static final String TYPE = "webhook";
// The webhook source type (equals the trigger type but a distinct concept).
private static final String WEBHOOK_SOURCE_TYPE = "webhook";
private final PolicyStore policyStore;
private final PolicyRunner policyRunner;
private final SourceStore sourceStore;
private final ApplicationProperties applicationProperties;
private volatile ScheduledExecutorService reconciler;
@Override
public String type() {
return TYPE;
}
@Override
public boolean requiresSource() {
return true;
}
@Override
public Set<String> supportedSourceTypes() {
return Set.of(WEBHOOK_SOURCE_TYPE);
}
@Override
public void validate(Policy policy) {
boolean hasWebhookSource =
policy.sourceIds().stream()
.map(sourceStore::get)
.flatMap(java.util.Optional::stream)
.anyMatch(source -> WEBHOOK_SOURCE_TYPE.equals(source.type()));
if (!hasWebhookSource) {
throw new IllegalArgumentException(
"webhook trigger requires at least one webhook input source");
}
}
@Override
public synchronized void start() {
if (reconciler != null) {
return;
}
long reconcileSeconds = applicationProperties.getPolicies().getWatchReconcileSeconds();
reconciler =
Executors.newSingleThreadScheduledExecutor(
Thread.ofVirtual().name("policy-webhook-reconcile-", 0).factory());
// First reconcile runs immediately so deliveries spooled before startup are picked up.
reconciler.scheduleAtFixedRate(this::safeReconcile, 0, reconcileSeconds, TimeUnit.SECONDS);
log.info("Webhook trigger started (reconcile every {}s)", reconcileSeconds);
}
@Override
public synchronized void stop() {
if (reconciler != null) {
reconciler.shutdownNow();
reconciler = null;
}
}
/** Run every webhook policy referencing the delivered-to source (LIGHT, best-effort). */
public void fireForWebhook(String webhookId) {
for (Policy policy : policyStore.findByTriggerType(TYPE)) {
if (!referencesWebhook(policy, webhookId)) {
continue;
}
try {
log.debug("Webhook policy {} ({}) saw a delivery", policy.id(), policy.name());
policyRunner.run(policy, SweepKind.LIGHT);
} catch (RuntimeException e) {
log.warn("Webhook run failed for policy {}: {}", policy.id(), e.getMessage());
}
}
}
private void safeReconcile() {
try {
for (Policy policy : policyStore.findByTriggerType(TYPE)) {
try {
policyRunner.run(policy);
} catch (RuntimeException e) {
log.warn(
"Webhook reconcile run failed for policy {}: {}",
policy.id(),
e.getMessage());
}
}
} catch (RuntimeException e) {
log.error("Webhook reconcile failed: {}", e.getMessage(), e);
}
}
/** Whether the policy references a webhook source whose routing id is {@code webhookId}. */
private boolean referencesWebhook(Policy policy, String webhookId) {
for (String sourceId : policy.sourceIds()) {
Source source = sourceStore.get(sourceId).orElse(null);
if (source == null || !WEBHOOK_SOURCE_TYPE.equals(source.type())) {
continue;
}
Object configured = source.options().get(WebhookConfig.WEBHOOK_ID_OPTION);
if (configured != null && configured.toString().equals(webhookId)) {
return true;
}
}
return false;
}
}
@@ -0,0 +1,89 @@
package stirling.software.proprietary.policy.webhook;
import java.util.Map;
/** Webhook source config: server-minted ids, staging via an S3 connection or the local spool. */
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<String, Object> options) {
String webhookId = trimmed(options.get(WEBHOOK_ID_OPTION));
if (webhookId == null) {
throw new IllegalArgumentException("webhook config requires a 'webhookId' option");
}
if (!WebhookIds.isValidId(webhookId)) {
throw new IllegalArgumentException("webhook config 'webhookId' has an invalid format");
}
String signingSecret = trimmed(options.get(SIGNING_SECRET_OPTION));
if (signingSecret == null) {
throw new IllegalArgumentException("webhook config requires a 'signingSecret' option");
}
String mode = trimmed(options.get(MODE_OPTION));
if (mode != null && !MODE_CONSUME.equals(mode) && !MODE_SNAPSHOT.equals(mode)) {
throw new IllegalArgumentException(
"webhook config 'mode' must be 'consume' or 'snapshot'");
}
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;
}
/** Reserved per-webhook staging prefix inside the connection's bucket. */
public String stagingPrefix() {
return STAGING_ROOT + "/" + webhookId;
}
public String mode() {
return snapshot ? MODE_SNAPSHOT : MODE_CONSUME;
}
private static Long connectionId(Map<String, Object> 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) {
if (value == null) {
return null;
}
String text = value.toString().trim();
return text.isEmpty() ? null : text;
}
/** Never prints the signing secret, so an accidental log line cannot leak it. */
@Override
public String toString() {
return "WebhookConfig[webhookId="
+ webhookId
+ ", connectionId="
+ connectionId
+ ", snapshot="
+ snapshot
+ "]";
}
}
@@ -0,0 +1,38 @@
package stirling.software.proprietary.policy.webhook;
import java.security.SecureRandom;
import java.util.Base64;
import java.util.regex.Pattern;
/** Generates and validates a webhook's routing id and signing secret. */
public final class WebhookIds {
/** URL-safe base64 without padding: the characters a single path segment allows. */
private static final Pattern VALID_ID = Pattern.compile("^[A-Za-z0-9_-]{16,128}$");
private static final SecureRandom RANDOM = new SecureRandom();
private static final Base64.Encoder ENCODER = Base64.getUrlEncoder().withoutPadding();
private WebhookIds() {}
/** A fresh routing token (~24 chars), unguessable so it cannot be enumerated. */
public static String newWebhookId() {
return randomToken(18);
}
/** A fresh HMAC signing key (~43 chars) revealed to the operator once at creation. */
public static String newSigningSecret() {
return randomToken(32);
}
/** Whether {@code id} is well-formed and safe as a path segment / directory name. */
public static boolean isValidId(String id) {
return id != null && VALID_ID.matcher(id).matches();
}
private static String randomToken(int bytes) {
byte[] buffer = new byte[bytes];
RANDOM.nextBytes(buffer);
return ENCODER.encodeToString(buffer);
}
}
@@ -0,0 +1,201 @@
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;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.server.ResponseStatusException;
import io.swagger.v3.oas.annotations.Hidden;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.servlet.http.HttpServletRequest;
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: HMAC-verifies a signed delivery, stages it, and fires the policies. */
@Slf4j
@RestController
@RequestMapping("/api/v1/webhooks")
@Hidden
@RequiredArgsConstructor
@Tag(name = "Webhooks", description = "Inbound webhook source receiver")
@ConditionalOnBooleanProperty(name = "policies.enabled")
public class WebhookReceiverController {
static final String SIGNATURE_HEADER = "X-Stirling-Signature";
static final String FILENAME_HEADER = "X-Stirling-Filename";
private static final String WEBHOOK_TYPE = "webhook";
private final SourceStore sourceStore;
private final WebhookSpool spool;
private final WebhookTrigger webhookTrigger;
private final ApplicationProperties applicationProperties;
private final S3ConnectionResolver connectionResolver;
private final S3ConnectionPool connectionPool;
@PostMapping("/{webhookId}")
@Operation(
summary = "Deliver a document to a webhook source",
description =
"The body is the raw document; sign it with the source's secret and present"
+ " 'sha256=<hex>' in the X-Stirling-Signature header. Returns 202 once"
+ " the document is spooled for the referencing policies.")
public ResponseEntity<WebhookDeliveryResponse> receive(
@PathVariable String webhookId,
@RequestHeader(value = SIGNATURE_HEADER, required = false) String signature,
@RequestHeader(value = FILENAME_HEADER, required = false) String filename,
HttpServletRequest request) {
if (!WebhookIds.isValidId(webhookId)) {
throw new ResponseStatusException(HttpStatus.NOT_FOUND, "No such webhook");
}
Source source = findWebhookSource(webhookId);
if (source == null) {
throw new ResponseStatusException(HttpStatus.NOT_FOUND, "No such webhook");
}
WebhookConfig config = WebhookConfig.from(source.options());
byte[] body = readBoundedBody(request);
if (!WebhookSignatures.verify(config.signingSecret(), body, signature)) {
// Same 401 whether the header was absent or wrong: never confirm a guess.
throw new ResponseStatusException(HttpStatus.UNAUTHORIZED, "Invalid signature");
}
if (!source.enabled()) {
throw new ResponseStatusException(
HttpStatus.FORBIDDEN, "Webhook source is paused; deliveries are not accepted");
}
if (body.length == 0) {
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Empty request body");
}
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);
log.info(
"Accepted webhook delivery '{}' ({} bytes) for {}",
storedName,
body.length,
webhookId);
return ResponseEntity.accepted()
.contentType(MediaType.APPLICATION_JSON)
.body(new WebhookDeliveryResponse(true, storedName, body.length));
}
/** The enabled-or-not webhook source whose routing id matches, or null if there is none. */
private Source findWebhookSource(String webhookId) {
for (Source source : sourceStore.all()) {
if (!WEBHOOK_TYPE.equals(source.type())) {
continue;
}
Object configured = source.options().get(WebhookConfig.WEBHOOK_ID_OPTION);
if (configured != null && configured.toString().equals(webhookId)) {
return source;
}
}
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 S3 connection (save-time access check trusted). */
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}. */
private byte[] readBoundedBody(HttpServletRequest request) {
long maxBytes = applicationProperties.getPolicies().getWebhookMaxBytes();
ByteArrayOutputStream buffer = new ByteArrayOutputStream();
byte[] chunk = new byte[8192];
long total = 0;
try (InputStream in = request.getInputStream()) {
int read;
while ((read = in.read(chunk)) != -1) {
total += read;
if (total > maxBytes) {
throw new ResponseStatusException(
HttpStatus.PAYLOAD_TOO_LARGE,
"Delivery exceeds the " + maxBytes + "-byte limit");
}
buffer.write(chunk, 0, read);
}
} catch (IOException e) {
throw new ResponseStatusException(
HttpStatus.BAD_REQUEST, "Could not read request body");
}
return buffer.toByteArray();
}
/** The 202 body: the stored (display) name and byte count of an accepted delivery. */
public record WebhookDeliveryResponse(boolean accepted, String filename, int bytes) {}
}
@@ -0,0 +1,53 @@
package stirling.software.proprietary.policy.webhook;
import java.nio.charset.StandardCharsets;
import java.security.InvalidKeyException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.HexFormat;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
/** HMAC-SHA256 signing of a webhook delivery's raw body (header sha256=<hex>). */
public final class WebhookSignatures {
private static final String ALGORITHM = "HmacSHA256";
private static final String PREFIX = "sha256=";
private WebhookSignatures() {}
/** The header a sender presents for {@code body}: {@code sha256=<lowercase hex>}. */
public static String sign(String signingSecret, byte[] body) {
return PREFIX + HexFormat.of().formatHex(hmac(signingSecret, body));
}
/** Whether the presented signature is valid for {@code body}; false on bad input. */
public static boolean verify(String signingSecret, byte[] body, String presented) {
if (signingSecret == null || presented == null || body == null) {
return false;
}
String hex = presented.trim();
if (hex.regionMatches(true, 0, PREFIX, 0, PREFIX.length())) {
hex = hex.substring(PREFIX.length());
}
byte[] presentedBytes;
try {
presentedBytes = HexFormat.of().parseHex(hex);
} catch (IllegalArgumentException notHex) {
return false;
}
return MessageDigest.isEqual(hmac(signingSecret, body), presentedBytes);
}
private static byte[] hmac(String signingSecret, byte[] body) {
try {
Mac mac = Mac.getInstance(ALGORITHM);
mac.init(new SecretKeySpec(signingSecret.getBytes(StandardCharsets.UTF_8), ALGORITHM));
return mac.doFinal(body);
} catch (NoSuchAlgorithmException | InvalidKeyException e) {
// HmacSHA256 is a required JCE algorithm and the key is always non-empty here.
throw new IllegalStateException("HMAC-SHA256 unavailable", e);
}
}
}
@@ -0,0 +1,106 @@
package stirling.software.proprietary.policy.webhook;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.util.UUID;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
import org.springframework.stereotype.Component;
import stirling.software.common.configuration.InstallationPathConfig;
/** Server-owned per-webhook staging directory under the install path. */
@Component
@ConditionalOnBooleanProperty(name = "policies.enabled")
public class WebhookSpool {
private static final String SPOOL_DIR = "policy-webhook-spool";
private static final String TEMP_SUFFIX = ".part";
private static final String DEFAULT_NAME = "document.pdf";
private static final int UNIQUE_LEN = 32; // UUID hex without dashes
private final Path spoolRoot;
public WebhookSpool() {
this(Path.of(InstallationPathConfig.getPath(), SPOOL_DIR));
}
// Lets a caller (and tests) root the spool at a chosen directory; Spring uses the no-arg one.
public WebhookSpool(Path spoolRoot) {
this.spoolRoot = spoolRoot.toAbsolutePath().normalize();
}
/** The staging directory for one webhook. Never escapes the spool root; may not yet exist. */
public Path dirFor(String webhookId) {
if (!WebhookIds.isValidId(webhookId)) {
throw new IllegalArgumentException("invalid webhookId");
}
Path dir = spoolRoot.resolve(webhookId).normalize();
if (!dir.getParent().equals(spoolRoot)) {
// A validated id is a single safe segment; this only trips on a bug, never user input.
throw new IllegalArgumentException("invalid webhookId");
}
return dir;
}
/** Write a delivery into the spool atomically (staged .part, then moved into place). */
public Path store(String webhookId, String filename, byte[] content) throws IOException {
Path dir = dirFor(webhookId);
Files.createDirectories(dir);
String finalName = spoolName(filename);
Path target = dir.resolve(finalName);
Path temp = dir.resolve("." + finalName + TEMP_SUFFIX);
Files.write(temp, content);
try {
Files.move(temp, target, StandardCopyOption.ATOMIC_MOVE);
} catch (IOException atomicUnsupported) {
Files.move(temp, target, StandardCopyOption.REPLACE_EXISTING);
}
return target;
}
/** The spool file name for a delivery: a unique prefix plus the sanitised original name. */
static String spoolName(String filename) {
return UUID.randomUUID().toString().replace("-", "") + "-" + sanitize(filename);
}
/** The original name recovered from a spool file name, for the resolved input's filename. */
public static String displayName(String spoolFileName) {
int dash = spoolFileName.indexOf('-');
// The unique prefix is fixed-length hex with no dashes, so the first dash is the separator.
if (dash == UNIQUE_LEN && dash + 1 < spoolFileName.length()) {
return spoolFileName.substring(dash + 1);
}
return spoolFileName;
}
/** S3 object-key suffix {@code <unique>/<name>}; the subfolder keeps the basename clean. */
public static String objectKeySuffix(String filename) {
return UUID.randomUUID().toString().replace("-", "") + "/" + sanitize(filename);
}
/** The staged delivery's display name (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) {
return DEFAULT_NAME;
}
String base = filename.replace('\\', '/');
int slash = base.lastIndexOf('/');
if (slash >= 0) {
base = base.substring(slash + 1);
}
base = base.replaceAll("[^A-Za-z0-9._-]", "_").trim();
// Strip leading dots so the result is never hidden (which resolve would skip) or empty.
while (base.startsWith(".")) {
base = base.substring(1);
}
return base.isEmpty() ? DEFAULT_NAME : base;
}
}
@@ -103,6 +103,7 @@ public class User implements UserDetails, Serializable {
@ElementCollection
@MapKeyColumn(name = "setting_key")
@Lob
@Column(name = "setting_value", columnDefinition = "text")
@CollectionTable(name = "user_settings", joinColumns = @JoinColumn(name = "user_id"))
@JsonIgnore
@@ -20,9 +20,8 @@ public final class SecretMasker {
private static final Pattern SENSITIVE =
RegexPatternUtils.getInstance()
.getPattern(
// secret[_-]?access[_-]?key precedes plain secret so camelCase keys
// like secretAccessKey (no word boundary after "secret") still match.
"(?i)\\b(password|token|secret[_-]?access[_-]?key|secret|api[_-]?key|authorization|auth|jwt|cred|cert)\\b");
// specific keys precede plain "secret" so camelCase still matches.
"(?i)\\b(password|token|secret[_-]?access[_-]?key|signing[_-]?secret|secret|api[_-]?key|authorization|auth|jwt|cred|cert)\\b");
private SecretMasker() {}
@@ -0,0 +1,221 @@
package stirling.software.proprietary.policy.input;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
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;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
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;
import stirling.software.common.util.FileReadinessChecker;
import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
import stirling.software.proprietary.policy.model.InputSpec;
import stirling.software.proprietary.policy.webhook.WebhookConfig;
import stirling.software.proprietary.policy.webhook.WebhookSpool;
/** Tests for {@link WebhookInputSource}: ledger-backed read/consume, and id/secret minting. */
@ExtendWith(MockitoExtension.class)
class WebhookInputSourceTest {
private static final String POLICY = "p1";
private static final String WEBHOOK_ID = "testwebhookid1234";
@Mock private FileReadinessChecker readinessChecker;
@Mock private S3InputSource s3InputSource;
@TempDir Path tempDir;
private WebhookSpool spool;
private WebhookInputSource source;
private InProcessProcessedLedger ledger;
private RecordingContext ctx;
@BeforeEach
void setUp() {
spool = new WebhookSpool(tempDir.resolve("spool"));
source = new WebhookInputSource(spool, readinessChecker, s3InputSource);
ledger = new InProcessProcessedLedger();
ctx = new RecordingContext();
lenient().when(readinessChecker.isReady(any())).thenReturn(true);
}
private static InputSpec spec(String mode) {
return new InputSpec(
"webhook",
Map.of("webhookId", WEBHOOK_ID, "signingSecret", "secret", "mode", mode));
}
@Test
void consumeRemovesTheDeliveryOnceProcessed() throws IOException {
Path delivered = spool.store(WEBHOOK_ID, "doc.pdf", "data".getBytes());
List<ResolvedInput> work = source.resolve(spec("consume"), ctx);
assertEquals(1, work.size());
assertEquals("doc.pdf", work.get(0).inputs().primary().get(0).getFilename());
// In flight: still spooled, but a second sweep does not pick it up again.
assertTrue(Files.exists(delivered));
assertTrue(source.resolve(spec("consume"), ctx).isEmpty());
work.get(0).onComplete().accept(true);
assertTrue(Files.notExists(delivered));
assertTrue(source.resolve(spec("consume"), ctx).isEmpty());
}
@Test
void aFailedRunLeavesTheDeliveryInPlace() throws IOException {
Path delivered = spool.store(WEBHOOK_ID, "doc.pdf", "data".getBytes());
List<ResolvedInput> work = source.resolve(spec("consume"), ctx);
work.get(0).onComplete().accept(false);
assertTrue(Files.exists(delivered));
}
@Test
void snapshotReReadsEveryRunAndNeverDeletes() throws IOException {
Path delivered = spool.store(WEBHOOK_ID, "doc.pdf", "data".getBytes());
assertEquals(1, source.resolve(spec("snapshot"), ctx).size());
List<ResolvedInput> second = source.resolve(spec("snapshot"), ctx);
assertEquals(1, second.size());
second.get(0).onComplete().accept(true);
assertTrue(Files.exists(delivered));
}
@Test
void nothingDeliveredIsAnEmptySourceNotAnError() throws IOException {
List<ResolvedInput> work = source.resolve(spec("consume"), ctx);
assertTrue(work.isEmpty());
assertTrue(ctx.present.isEmpty());
}
@Test
void validateRejectsMissingIdOrSecret() {
assertThrows(
IllegalArgumentException.class,
() -> source.validate(new InputSpec("webhook", Map.of("signingSecret", "s"))));
assertThrows(
IllegalArgumentException.class,
() -> source.validate(new InputSpec("webhook", Map.of("webhookId", WEBHOOK_ID))));
}
@Test
void prepareMintsIdAndSecretOnCreate() {
Map<String, Object> prepared =
source.prepareOptionsForSave(Map.of("mode", "consume"), true);
String id = prepared.get(WebhookConfig.WEBHOOK_ID_OPTION).toString();
String secret = prepared.get(WebhookConfig.SIGNING_SECRET_OPTION).toString();
assertFalse(id.isBlank());
assertFalse(secret.isBlank());
assertEquals("consume", prepared.get("mode"));
Map<String, Object> other = source.prepareOptionsForSave(Map.of(), true);
assertNotEquals(id, other.get(WebhookConfig.WEBHOOK_ID_OPTION).toString());
}
@Test
void prepareLeavesAnExistingWebhookUntouchedOnEdit() {
Map<String, Object> existing =
Map.of("webhookId", WEBHOOK_ID, "signingSecret", "keepme", "mode", "snapshot");
Map<String, Object> prepared = source.prepareOptionsForSave(existing, false);
assertEquals(WEBHOOK_ID, prepared.get("webhookId"));
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<InputSpec> 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 ownership-checks the connection.
verify(s3InputSource).validate(any());
}
/** Policy-scoped context backed by the in-process ledger, recording presence reports. */
private class RecordingContext implements ResolveContext {
private final List<String> present = new ArrayList<>();
@Override
public boolean claim(String identity, String gate, Supplier<String> 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<String> identities) {
present.addAll(identities);
}
}
}
@@ -1,33 +1,42 @@
package stirling.software.proprietary.policy.source;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
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.anyBoolean;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.http.ResponseEntity;
import org.springframework.web.server.ResponseStatusException;
import stirling.software.common.model.ApplicationProperties;
import stirling.software.common.service.UserServiceInterface;
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;
import stirling.software.proprietary.policy.model.Policy;
import stirling.software.proprietary.policy.store.InProcessPolicyStore;
import stirling.software.proprietary.policy.store.PolicyStore;
import stirling.software.proprietary.policy.trigger.PolicyTriggerManager;
import stirling.software.proprietary.policy.webhook.WebhookSpool;
import stirling.software.proprietary.util.SecretMasker;
/**
@@ -41,6 +50,9 @@ class SourceControllerTest {
private final PolicyStore policyStore = new InProcessPolicyStore();
private PolicyTriggerManager triggerManager;
private SourceController controller;
private SourceController webhookController;
@TempDir Path tempDir;
@BeforeEach
void setUp() {
@@ -58,9 +70,11 @@ class SourceControllerTest {
policyGuard,
new InProcessSourceDocCounter());
triggerManager = mock(PolicyTriggerManager.class);
// A permissive input source so config validation passes and save can be exercised.
// A permissive input source; stub prepareOptionsForSave as a pass-through.
InputSource folderInput = mock(InputSource.class);
when(folderInput.supports(any())).thenReturn(true);
when(folderInput.prepareOptionsForSave(any(), anyBoolean()))
.thenAnswer(invocation -> invocation.getArgument(0));
controller =
new SourceController(
sourceStore,
@@ -72,6 +86,48 @@ class SourceControllerTest {
triggerManager,
properties,
List.of(folderInput));
WebhookInputSource webhookInput =
new WebhookInputSource(
new WebhookSpool(tempDir),
mock(FileReadinessChecker.class),
mock(S3InputSource.class));
webhookController =
new SourceController(
sourceStore,
sourceGuard,
overviewService,
policyStore,
policyGuard,
authority,
triggerManager,
properties,
List.of(webhookInput));
}
@Test
void creatingAWebhookRevealsItsSecretOnceThenMasks() {
Source created =
webhookController
.save(
new Source(
null,
"Partner uploads",
"webhook",
Map.of("mode", "consume"),
true,
null,
null))
.getBody();
String secret = String.valueOf(created.options().get("signingSecret"));
String webhookId = String.valueOf(created.options().get("webhookId"));
assertNotEquals(SecretMasker.REDACTED, secret);
assertFalse(secret.isBlank());
assertFalse(webhookId.isBlank());
Source read = webhookController.get(created.id()).getBody();
assertEquals(SecretMasker.REDACTED, read.options().get("signingSecret"));
assertEquals(webhookId, read.options().get("webhookId"));
}
@Test
@@ -0,0 +1,117 @@
package stirling.software.proprietary.policy.trigger;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import stirling.software.common.model.ApplicationProperties;
import stirling.software.proprietary.policy.engine.PolicyRunner;
import stirling.software.proprietary.policy.engine.SweepKind;
import stirling.software.proprietary.policy.model.OutputSpec;
import stirling.software.proprietary.policy.model.PipelineStep;
import stirling.software.proprietary.policy.model.Policy;
import stirling.software.proprietary.policy.model.TriggerConfig;
import stirling.software.proprietary.policy.source.InProcessSourceStore;
import stirling.software.proprietary.policy.source.Source;
import stirling.software.proprietary.policy.source.SourceStore;
import stirling.software.proprietary.policy.store.PolicyStore;
/** Tests for {@link WebhookTrigger}: fires only policies referencing the delivered-to webhook. */
@ExtendWith(MockitoExtension.class)
class WebhookTriggerTest {
private static final String TYPE = "webhook";
@Mock private PolicyStore policyStore;
@Mock private PolicyRunner policyRunner;
private final SourceStore sourceStore = new InProcessSourceStore();
private WebhookTrigger trigger;
@BeforeEach
void setUp() {
trigger =
new WebhookTrigger(
policyStore, policyRunner, sourceStore, new ApplicationProperties());
}
@Test
void firesOnlyPoliciesReferencingTheDeliveredWebhook() {
Policy matching = webhookPolicy("a", "whkA");
Policy other = webhookPolicy("b", "whkB");
when(policyStore.findByTriggerType(TYPE)).thenReturn(List.of(matching, other));
trigger.fireForWebhook("whkA");
verify(policyRunner).run(matching, SweepKind.LIGHT);
verify(policyRunner, never()).run(other, SweepKind.LIGHT);
}
@Test
void ignoresADeliveryForAnUnknownWebhookId() {
Policy policy = webhookPolicy("a", "whkA");
when(policyStore.findByTriggerType(TYPE)).thenReturn(List.of(policy));
trigger.fireForWebhook("whkZ");
verify(policyRunner, never()).run(any(), any(SweepKind.class));
}
@Test
void validateRequiresAWebhookSource() {
assertThrows(
IllegalArgumentException.class,
() -> trigger.validate(policy("p", webhookTriggerConfig(), List.of())));
trigger.validate(webhookPolicy("p", "whkA"));
}
private static TriggerConfig webhookTriggerConfig() {
return new TriggerConfig(TYPE, Map.of());
}
/** Persist a webhook source with the given routing id and return a policy referencing it. */
private Policy webhookPolicy(String id, String webhookId) {
String sourceId =
sourceStore
.save(
new Source(
null,
"hook",
"webhook",
Map.of(
"webhookId",
webhookId,
"signingSecret",
"s",
"mode",
"consume"),
true,
"owner",
null))
.id();
return policy(id, webhookTriggerConfig(), List.of(sourceId));
}
private static Policy policy(String id, TriggerConfig trigger, List<String> sourceIds) {
return new Policy(
id,
"hook",
"owner",
true,
trigger,
sourceIds,
List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
OutputSpec.inline());
}
}
@@ -0,0 +1,186 @@
package stirling.software.proprietary.policy.webhook;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.http.ResponseEntity;
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;
import stirling.software.proprietary.policy.trigger.WebhookTrigger;
import stirling.software.proprietary.policy.webhook.WebhookReceiverController.WebhookDeliveryResponse;
/** Tests for the public webhook receiver: valid delivery spools + fires; bad requests rejected. */
class WebhookReceiverControllerTest {
private static final String WEBHOOK_ID = "receivertestid12";
private static final String SECRET = "topsecret";
private static final byte[] BODY = "a pdf".getBytes(StandardCharsets.UTF_8);
@TempDir Path tempDir;
private SourceStore sourceStore;
private WebhookSpool spool;
private WebhookTrigger trigger;
private ApplicationProperties properties;
private WebhookReceiverController controller;
@BeforeEach
void setUp() {
sourceStore = new InProcessSourceStore();
sourceStore.save(webhookSource(true));
spool = new WebhookSpool(tempDir.resolve("spool"));
trigger = mock(WebhookTrigger.class);
properties = new ApplicationProperties();
// 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) {
return new Source(
"s1",
"Partner uploads",
"webhook",
Map.of("webhookId", WEBHOOK_ID, "signingSecret", SECRET, "mode", "consume"),
enabled,
"owner",
null);
}
private static MockHttpServletRequest request(byte[] body) {
MockHttpServletRequest req =
new MockHttpServletRequest("POST", "/api/v1/webhooks/" + WEBHOOK_ID);
req.setContent(body);
return req;
}
@Test
void aValidDeliveryIsSpooledAndFiresTheTrigger() throws IOException {
String signature = WebhookSignatures.sign(SECRET, BODY);
ResponseEntity<WebhookDeliveryResponse> response =
controller.receive(WEBHOOK_ID, signature, "invoice.pdf", request(BODY));
assertEquals(202, response.getStatusCode().value());
assertTrue(response.getBody().accepted());
assertEquals("invoice.pdf", response.getBody().filename());
assertEquals(1, spooledFiles().size());
verify(trigger).fireForWebhook(WEBHOOK_ID);
}
@Test
void aWrongSignatureIsRejectedAndStoresNothing() {
ResponseStatusException ex =
assertThrows(
ResponseStatusException.class,
() ->
controller.receive(
WEBHOOK_ID, "sha256=deadbeef", "x.pdf", request(BODY)));
assertEquals(401, ex.getStatusCode().value());
assertTrue(spooledFiles().isEmpty());
verify(trigger, never()).fireForWebhook(WEBHOOK_ID);
}
@Test
void anUnknownWebhookIsNotFound() {
ResponseStatusException ex =
assertThrows(
ResponseStatusException.class,
() ->
controller.receive(
"unknownwebhookid",
WebhookSignatures.sign(SECRET, BODY),
"x.pdf",
request(BODY)));
assertEquals(404, ex.getStatusCode().value());
}
@Test
void aPausedSourceRejectsDeliveries() {
sourceStore.save(webhookSource(false));
String signature = WebhookSignatures.sign(SECRET, BODY);
ResponseStatusException ex =
assertThrows(
ResponseStatusException.class,
() -> controller.receive(WEBHOOK_ID, signature, "x.pdf", request(BODY)));
assertEquals(403, ex.getStatusCode().value());
assertTrue(spooledFiles().isEmpty());
}
@Test
void anEmptyBodyIsRejected() {
byte[] empty = new byte[0];
String signature = WebhookSignatures.sign(SECRET, empty);
ResponseStatusException ex =
assertThrows(
ResponseStatusException.class,
() -> controller.receive(WEBHOOK_ID, signature, null, request(empty)));
assertEquals(400, ex.getStatusCode().value());
}
@Test
void anOversizeDeliveryIsRejectedBeforeStoring() {
properties.getPolicies().setWebhookMaxBytes(2);
ResponseStatusException ex =
assertThrows(
ResponseStatusException.class,
() ->
controller.receive(
WEBHOOK_ID,
WebhookSignatures.sign(SECRET, BODY),
"x.pdf",
request(BODY)));
assertEquals(413, ex.getStatusCode().value());
assertTrue(spooledFiles().isEmpty());
}
private List<Path> spooledFiles() {
Path dir = spool.dirFor(WEBHOOK_ID);
if (!Files.isDirectory(dir)) {
return List.of();
}
try (Stream<Path> entries = Files.list(dir)) {
return entries.filter(Files::isRegularFile)
.filter(p -> !p.getFileName().toString().startsWith("."))
.toList();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
@@ -0,0 +1,263 @@
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 connection-backed webhook against MinIO: stage, read via S3 delegation, consume. */
@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 MinIO S3 connection; no principal here, so ownership is skipped.
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);
var response = receiver.receive(WEBHOOK_ID, signature, "invoice.pdf", request(body));
assertThat(response.getStatusCode().value()).isEqualTo(202);
List<S3Object> 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");
List<ResolvedInput> 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();
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<S3Object> 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<String> present = new ArrayList<>();
@Override
public boolean claim(String identity, String gate, Supplier<String> 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<String> identities) {
present.addAll(identities);
}
}
}
@@ -0,0 +1,49 @@
package stirling.software.proprietary.policy.webhook;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.nio.charset.StandardCharsets;
import org.junit.jupiter.api.Test;
/** The HMAC-SHA256 webhook signing scheme: correct signatures verify, tampered ones do not. */
class WebhookSignaturesTest {
private static final String SECRET = "whsec_test_secret";
private static final byte[] BODY = "the document bytes".getBytes(StandardCharsets.UTF_8);
@Test
void aFreshlySignedBodyVerifies() {
String header = WebhookSignatures.sign(SECRET, BODY);
assertTrue(header.startsWith("sha256="));
assertTrue(WebhookSignatures.verify(SECRET, BODY, header));
}
@Test
void abarehexSignatureVerifiesToo() {
String header = WebhookSignatures.sign(SECRET, BODY);
String bareHex = header.substring("sha256=".length());
assertTrue(WebhookSignatures.verify(SECRET, BODY, bareHex));
}
@Test
void aWrongSecretDoesNotVerify() {
String header = WebhookSignatures.sign(SECRET, BODY);
assertFalse(WebhookSignatures.verify("other-secret", BODY, header));
}
@Test
void atamperedBodyDoesNotVerify() {
String header = WebhookSignatures.sign(SECRET, BODY);
byte[] tampered = "the document byteS".getBytes(StandardCharsets.UTF_8);
assertFalse(WebhookSignatures.verify(SECRET, tampered, header));
}
@Test
void aMissingOrMalformedHeaderIsFalseNotAnError() {
assertFalse(WebhookSignatures.verify(SECRET, BODY, null));
assertFalse(WebhookSignatures.verify(SECRET, BODY, "sha256=not-hex"));
assertFalse(WebhookSignatures.verify(SECRET, BODY, ""));
}
}
@@ -103,6 +103,18 @@ class SecretMaskerTest {
assertEquals("AKIAEXAMPLE", result.get("accessKeyId"));
}
@Test
@DisplayName("should mask camelCase signingSecret despite no word boundary")
void shouldMaskCamelCaseSigningSecret() {
Map<String, Object> input = Map.of("signingSecret", "shh", "webhookId", "whk_abc");
Map<String, Object> result = SecretMasker.mask(input);
assertEquals(SecretMasker.REDACTED, result.get("signingSecret"));
// The routing id is a public URL token, not a secret.
assertEquals("whk_abc", result.get("webhookId"));
}
@Test
@DisplayName("should mask nested map sensitive keys")
void shouldMaskNestedMapSensitiveKeys() {
+1 -1
View File
@@ -91,7 +91,7 @@ springBoot {
allprojects {
group = 'stirling.software'
version = '2.14.2'
version = '2.14.1'
configurations.configureEach {
exclude group: "org.springframework.boot", module: "spring-boot-starter-tomcat"
@@ -8139,6 +8139,36 @@ placeholder = "us-east-1"
[portal.sources.types.s3.fields.secretAccessKey]
label = "Secret access key"
[portal.sources.types.webhook]
description = "Receive documents by signed HTTP POST from any external system."
label = "Webhook"
[portal.sources.types.webhook.detail]
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"
[portal.sources.types.webhook.fields.mode.options]
consume = "Consume: process each delivery once"
snapshot = "Snapshot: re-read the spool every run"
[portal.sources.types.webhook.reveal]
copy = "Copy"
done = "Done"
secret = "Signing secret"
secretHelp = "Sign each delivery's raw body with this key (HMAC-SHA256) and send it as the X-Stirling-Signature header."
secretWarning = "Copy the signing secret now. For your security it is shown only once and cannot be retrieved later."
title = "Webhook created"
url = "Delivery URL"
usage = "POST each document as the raw request body to the delivery URL with a binary content type (application/pdf or application/octet-stream). Referencing policies run automatically on arrival."
[portal.sources.types.unknown]
label = "Source"
+1 -1
View File
@@ -2,7 +2,7 @@
"$schema": "../node_modules/@tauri-apps/cli/config.schema.json",
"productName": "Stirling PDF",
"mainBinaryName": "Stirling-PDF",
"version": "2.14.2",
"version": "2.14.1",
"identifier": "stirling.pdf.dev",
"build": {
"frontendDist": "../dist",
@@ -38,7 +38,7 @@ const FREE_LICENSE_INFO: LicenseInfo = {
const BASE_NO_LOGIN_CONFIG: AppConfig = {
enableAnalytics: true,
appVersion: "2.14.2",
appVersion: "2.14.1",
serverCertificateEnabled: false,
enableAlphaFunctionality: false,
serverPort: 8080,
@@ -7,6 +7,15 @@ describe("creatableSourceTypes (SaaS)", () => {
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", () => {
expect(creatableSourceTypes().map((t) => t.type)).toContain("s3");
});
@@ -1,13 +1,25 @@
import {
CREATABLE_SOURCE_TYPES,
WEBHOOK_SOURCE_TYPE,
type CreatableSourceType,
} from "@portal/components/sources/sourceTypes";
/**
* Hosted deployments never read the server's filesystem (the backend's
* FolderAccessGuard denies it outright), so folder connections are not offered
* in the connect wizard.
*/
/** Hosted drops folder (no server FS) and requires webhook's S3 connection (node-local isn't durable). */
export function creatableSourceTypes(): CreatableSourceType[] {
return CREATABLE_SOURCE_TYPES.filter((type) => type.type !== "folder");
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,
),
};
}
@@ -32,6 +32,8 @@ export interface SourceView {
docsTotal: number;
docs24h: number;
docs30d: number;
/** Webhook sources only: the server-relative delivery path senders POST to. Null otherwise. */
webhookPath?: string | null;
}
export interface SourceKpi {
@@ -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<string, SourceTypeMeta> = {
folder: {
labelKey: "portal.sources.types.folder.label",
@@ -38,6 +41,11 @@ const SOURCE_TYPE_META: Record<string, SourceTypeMeta> = {
icon: "☁",
accent: "brand",
},
webhook: {
labelKey: "portal.sources.types.webhook.label",
icon: "↯",
accent: "warning",
},
};
const UNKNOWN_TYPE_META: SourceTypeMeta = {
@@ -180,6 +188,40 @@ export const CREATABLE_SOURCE_TYPES: CreatableSourceType[] = [
},
],
},
{
// URL + secret minted server-side (revealed once); an S3 connection makes staging durable, else local disk.
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. */
@@ -55,6 +55,18 @@ function seedSources(): StoredSource[] {
enabled: false,
owner: "data-eng@acme.com",
},
{
id: "src-webhook",
name: "Partner uploads",
type: "webhook",
options: {
webhookId: "whk_demo_5f3a9c21b7",
signingSecret: "whsec_demo_2b8e1d47a9f60c35",
mode: "consume",
},
enabled: true,
owner: "you@acme.com",
},
];
}
@@ -76,6 +88,7 @@ const docCounts: Record<
"src-contracts": { total: 12840, last24h: 96, last30d: 2310 },
"src-archive": { total: 1180, last24h: 0, last30d: 0 },
"src-legacy": { total: 48600, last24h: 0, last30d: 0 },
"src-webhook": { total: 3120, last24h: 24, last30d: 640 },
};
function docsFor(id: string): {
@@ -96,14 +109,23 @@ function nextId(): string {
return `src_${Date.now().toString(36)}_${idCounter}`;
}
/** A demo token for a newly created webhook's server-generated id/secret (mock only). */
function randomToken(): string {
return (
Math.random().toString(36).slice(2) + Math.random().toString(36).slice(2)
);
}
function refsFor(id: string): SourcePolicyRef[] {
return references[id] ?? [];
}
/** Mirror the backend's secret redaction so config rows never surface a secret in the overview. */
function configRows(options: Record<string, unknown>) {
const secret = /secret|password|token/i;
return Object.entries(options).map(([key, value]) => ({
label: key.charAt(0).toUpperCase() + key.slice(1),
value: String(value),
value: secret.test(key) ? "********" : String(value),
}));
}
@@ -120,6 +142,7 @@ function toSourceView(
refs: SourcePolicyRef[],
): SourceView {
const docs = docsFor(source.id);
const webhookId = source.options.webhookId;
return {
id: source.id,
name: source.name,
@@ -131,6 +154,10 @@ function toSourceView(
docsTotal: docs.total,
docs24h: docs.last24h,
docs30d: docs.last30d,
webhookPath:
source.type === "webhook" && typeof webhookId === "string"
? `/api/v1/webhooks/${webhookId}`
: null,
};
}
@@ -207,8 +234,18 @@ export const sourcesHandlers = [
? store.find((s) => s.id === incoming.id)
: undefined;
const id = existing?.id ?? nextId();
// A new webhook's id + secret are minted server-side and revealed once on this create response.
let options = incoming.options ?? {};
if (incoming.type === "webhook" && !existing && !options.webhookId) {
options = {
...options,
webhookId: `whk_${randomToken()}`,
signingSecret: `whsec_${randomToken()}`,
};
}
const saved: StoredSource = {
...incoming,
options,
id,
owner: existing?.owner ?? "you@acme.com",
};
@@ -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);
}
@@ -102,6 +102,39 @@ 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" }),
);
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.
@@ -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<string | null>(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,40 @@ export function SourceBuilder() {
);
const canSave = name.trim() !== "" && requiredComplete && !submitting;
// An existing webhook's delivery URL, read-only in edit (the secret is revealed only 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 minted id + secret once; reveal them (with the URL) before leaving.
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 +164,10 @@ export function SourceBuilder() {
}
}
function copy(text: string) {
void navigator.clipboard?.writeText(text);
}
async function confirmDelete() {
if (!id || deleting) return;
setDeleting(true);
@@ -290,6 +327,28 @@ export function SourceBuilder() {
</FormField>
))}
{editingWebhookId && (
<FormField
label={t("portal.sources.types.webhook.detail.deliveryUrl")}
helperText={t("portal.sources.types.webhook.detail.secretNote")}
>
<div className="portal-source-builder__copy-row">
<Input
value={webhookUrl(editingWebhookId)}
readOnly
onFocus={(e) => e.currentTarget.select()}
/>
<Button
variant="tertiary"
size="sm"
onClick={() => copy(webhookUrl(editingWebhookId))}
>
{t("portal.sources.types.webhook.reveal.copy")}
</Button>
</div>
</FormField>
)}
{error && <Banner tone="danger" description={error} />}
</div>
@@ -321,6 +380,69 @@ export function SourceBuilder() {
>
<p>{t("portal.sources.delete.body", { name })}</p>
</Modal>
<Modal
open={reveal !== null}
onClose={dismissReveal}
width="md"
title={t("portal.sources.types.webhook.reveal.title")}
footer={
<div className="portal-source-builder__delete-actions">
<Button size="sm" onClick={dismissReveal}>
{t("portal.sources.types.webhook.reveal.done")}
</Button>
</div>
}
>
{reveal && (
<div className="portal-source-builder__reveal">
<Banner
tone="warning"
description={t(
"portal.sources.types.webhook.reveal.secretWarning",
)}
/>
<FormField label={t("portal.sources.types.webhook.reveal.url")}>
<div className="portal-source-builder__copy-row">
<Input
value={revealUrl}
readOnly
onFocus={(e) => e.currentTarget.select()}
/>
<Button
variant="tertiary"
size="sm"
onClick={() => copy(revealUrl)}
>
{t("portal.sources.types.webhook.reveal.copy")}
</Button>
</div>
</FormField>
<FormField
label={t("portal.sources.types.webhook.reveal.secret")}
helperText={t("portal.sources.types.webhook.reveal.secretHelp")}
>
<div className="portal-source-builder__copy-row">
<Input
value={revealSecret}
readOnly
onFocus={(e) => e.currentTarget.select()}
/>
<Button
variant="tertiary"
size="sm"
onClick={() => copy(revealSecret)}
>
{t("portal.sources.types.webhook.reveal.copy")}
</Button>
</div>
</FormField>
<p className="portal-source-builder__muted">
{t("portal.sources.types.webhook.reveal.usage")}
</p>
</div>
)}
</Modal>
</div>
);
}
@@ -48,7 +48,7 @@ const FREE_LICENSE_INFO: LicenseInfo = {
const BASE_NO_LOGIN_CONFIG: AppConfig = {
enableAnalytics: true,
appVersion: "2.14.2",
appVersion: "2.14.1",
serverCertificateEnabled: false,
enableAlphaFunctionality: false,
enableDesktopInstallSlide: true,
@@ -1,225 +0,0 @@
# Multi-node Stirling-PDF processor test stack: shared Postgres/MinIO/Valkey behind an nginx LB fronting N app nodes, with enterprise features unlocked via a local test licence key.
# Bring up: ./start-multinode-test.sh Validate: ./validate-multinode-test.sh Access: http://localhost:8080 (admin / stirling)
x-stirling-node: &stirling-node
build:
context: ../..
dockerfile: docker/embedded/Dockerfile
image: stirling-pdf-multinode:local
restart: unless-stopped
depends_on:
postgres:
condition: service_healthy
valkey:
condition: service_healthy
minio-init:
condition: service_completed_successfully
deploy:
resources:
limits:
memory: 4G
healthcheck:
test: ["CMD-SHELL", "curl -fsS http://localhost:8080/api/v1/info/status | grep -q UP"]
interval: 10s
timeout: 10s
retries: 30
start_period: 90s
environment: &stirling-env
# --- Licensing: unlock pro/enterprise (custom DB + cluster mode) ---
# Cluster mode is licence-gated - set PREMIUM_KEY (locally: export PREMIUM_KEY=...; CI: the PREMIUM_KEY_ENTERPRISE secret); the default below is a non-functional placeholder, never a real key.
PREMIUM_ENABLED: "true"
PREMIUM_KEY: "${PREMIUM_KEY:-00000000-0000-0000-0000-000000000000}"
DISABLE_ADDITIONAL_FEATURES: "false"
# --- Login + teams ---
DOCKER_ENABLE_SECURITY: "true"
SECURITY_ENABLELOGIN: "true"
SECURITY_INITIALLOGIN_USERNAME: "admin"
SECURITY_INITIALLOGIN_PASSWORD: "stirling"
# Global API key: lets the seed/validate scripts call admin APIs without a login flow.
SECURITY_CUSTOMGLOBALAPIKEY: "multinode-test-key"
# --- Shared Postgres (the single DB every node coordinates through) ---
# enableCustomDatabase is an enterprise feature, unlocked by PREMIUM_* above.
SYSTEM_DATASOURCE_ENABLECUSTOMDATABASE: "true"
SYSTEM_DATASOURCE_TYPE: "postgresql"
SYSTEM_DATASOURCE_HOSTNAME: "postgres"
SYSTEM_DATASOURCE_PORT: "5432"
SYSTEM_DATASOURCE_NAME: "stirling"
SYSTEM_DATASOURCE_USERNAME: "stirling"
SYSTEM_DATASOURCE_PASSWORD: "stirling"
# User.settings has no @Lob annotation - with one, Postgres treats it as a large object and every login 500s ("Large Objects may not be used in auto-commit mode"); see multinode/README.md.
# --- Shared object storage (persistent user uploads + share/store feature) ---
STORAGE_ENABLED: "true"
STORAGE_PROVIDER: "s3"
STORAGE_S3_ENDPOINT: "http://minio:9000"
STORAGE_S3_BUCKET: "stirling-storage"
STORAGE_S3_REGION: "us-east-1"
STORAGE_S3_ACCESSKEY: "minioadmin"
STORAGE_S3_SECRETKEY: "minioadmin"
STORAGE_S3_PATHSTYLEACCESS: "true"
STORAGE_S3_ALLOWPRIVATEENDPOINTS: "true"
# --- Cluster mode: Valkey backplane + shared S3 job-artifact store ---
# artifactStore=s3 is REQUIRED for multi-node (transient job artifacts must be shared).
CLUSTER_ENABLED: "true"
CLUSTER_BACKPLANE: "valkey"
CLUSTER_ARTIFACTSTORE: "s3"
CLUSTER_VALKEY_URL: "redis://valkey:6379"
# SPRING_DATA_REDIS_REPOSITORIES_ENABLED is not needed: DataRedisRepositoriesAutoConfiguration is excluded (see application.properties, multinode/README.md).
# --- Shared credential-encryption key (REQUIRED in cluster mode) ---
# AES-256 key that encrypts stored integration/S3 secrets - must match on every node or secrets encrypted on one can't decrypt on another; boot fails if unset with cluster.enabled=true (test-only value; JWT keys persist separately in the shared DB).
STIRLING_CREDENTIAL_ENCRYPTION_KEY: "dMobekyUEnEV7WHBah2FkbboP4Coqifd3JRXB00LiIY="
# --- Policy / processor subsystem (the thing under test) ---
POLICIES_ENABLED: "true"
# Let policy S3 sources/webhook-staging connections point at the in-cluster MinIO.
POLICIES_ALLOWPRIVATES3ENDPOINTS: "true"
# --- Misc features on ---
METRICS_ENABLED: "true"
SYSTEM_DEFAULTLOCALE: "en-US"
SYSTEM_MAXFILESIZE: "100"
UI_APPNAME: "Stirling-PDF Multi-Node"
UI_APPNAMENAVBAR: "Stirling Multi-Node"
networks:
- stirling-multinode
services:
# ---- Shared Postgres: the single DB all nodes coordinate through ----------
postgres:
image: postgres:17-alpine
container_name: multinode-postgres
environment:
POSTGRES_USER: stirling
POSTGRES_PASSWORD: stirling
POSTGRES_DB: stirling
ports:
- "5434:5432" # host access for inspection (psql -h localhost -p 5434 -U stirling)
volumes:
- postgres-data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U stirling -d stirling"]
interval: 3s
timeout: 5s
retries: 30
networks:
- stirling-multinode
# ---- Valkey: cluster backplane (shared job state / rate limiting) ---------
valkey:
image: valkey/valkey:8-alpine
container_name: multinode-valkey
command: ["valkey-server", "--save", "", "--appendonly", "no"]
healthcheck:
test: ["CMD-SHELL", "valkey-cli ping | grep -q PONG"]
interval: 3s
timeout: 5s
retries: 30
networks:
- stirling-multinode
# ---- MinIO: shared S3 (platform storage + policy connections + webhook staging) ----
minio:
image: minio/minio:latest
container_name: multinode-minio
command: ["server", "/data", "--console-address", ":9001"]
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin
ports:
- "9000:9000" # S3 API
- "9001:9001" # web console (minioadmin / minioadmin)
volumes:
- minio-data:/data
healthcheck:
test: ["CMD-SHELL", "mc ready local || exit 1"]
interval: 5s
timeout: 5s
retries: 30
networks:
- stirling-multinode
# ---- One-shot: create the buckets the stack needs -------------------------
minio-init:
image: minio/mc:latest
container_name: multinode-minio-init
depends_on:
minio:
condition: service_healthy
entrypoint: >
/bin/sh -c "
mc alias set local http://minio:9000 minioadmin minioadmin &&
mc mb --ignore-existing local/stirling-storage &&
mc mb --ignore-existing local/policy-data &&
echo 'buckets ready: stirling-storage, policy-data'
"
networks:
- stirling-multinode
# ---- App node 1 -----------------------------------------------------------
stirling-1:
<<: *stirling-node
container_name: multinode-stirling-1
environment:
<<: *stirling-env
SYSTEM_NODEID: "node-1"
# ---- App node 2 -----------------------------------------------------------
# No ordering vs node-1 needed - each node mints its own signing key at boot and publishes the public half to the shared DB, so both verify each other's tokens.
stirling-2:
<<: *stirling-node
container_name: multinode-stirling-2
environment:
<<: *stirling-env
SYSTEM_NODEID: "node-2"
# ---- One-shot seeder: teams, ~40 users, S3 connection, policies -----------
# Profile-gated so plain `docker compose up` skips it; the start script runs it once, using postgres:alpine for psql with curl+jq added for the HTTP calls.
seed:
image: postgres:17-alpine
container_name: multinode-seed
profiles: ["seed"]
depends_on:
stirling-1:
condition: service_healthy
stirling-2:
condition: service_healthy
environment:
BASE_URL: "http://nginx:8080"
PGHOST: "postgres"
PGUSER: "stirling"
PGPASSWORD: "stirling"
PGDATABASE: "stirling"
USER_COUNT: "40"
volumes:
- ./multinode/seed.sh:/seed.sh:ro
entrypoint: ["/bin/sh", "-c", "apk add --no-cache curl jq >/dev/null && sh /seed.sh"]
networks:
- stirling-multinode
# ---- nginx load balancer: single entrypoint round-robining the nodes ------
nginx:
image: nginx:1.27-alpine
container_name: multinode-nginx
depends_on:
stirling-1:
condition: service_healthy
stirling-2:
condition: service_healthy
ports:
- "8080:8080"
volumes:
- ./multinode/nginx.conf:/etc/nginx/nginx.conf:ro
networks:
- stirling-multinode
networks:
stirling-multinode:
driver: bridge
volumes:
postgres-data:
minio-data:
-49
View File
@@ -1,49 +0,0 @@
worker_processes 1;
events { worker_connections 1024; }
http {
# Round-robin across the app nodes. Add a node here to scale out.
# max_fails=1 marks a node down after a single failure so a dead node drains fast.
upstream stirling_nodes {
server stirling-1:8080 max_fails=1 fail_timeout=10s;
server stirling-2:8080 max_fails=1 fail_timeout=10s;
}
# Large uploads (SYSTEM_MAXFILESIZE=100MB) plus headroom.
client_max_body_size 200m;
map $http_upgrade $connection_upgrade {
default upgrade;
'' close;
}
server {
listen 8080;
location / {
proxy_pass http://stirling_nodes;
# Graceful failover: retries the other node only on connection-level failures (unreachable/timeout), never on 5xx, so a POST a node already started is never re-sent.
proxy_next_upstream error timeout;
proxy_next_upstream_tries 2;
proxy_connect_timeout 3s;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# WebSocket / SSE upgrade support (policy run streaming).
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $connection_upgrade;
# Stream server-sent events straight through, don't buffer.
proxy_buffering off;
proxy_read_timeout 3600s;
# Surface which app node served the request, so the validate script can prove the LB is spreading load.
add_header X-Served-By $upstream_addr always;
}
}
}
-140
View File
@@ -1,140 +0,0 @@
#!/bin/sh
# Seeds a running multi-node stack: 4 teams, ~40 users, an S3 connection, a scheduled S3 policy, and a webhook source if the build supports it.
# Auth uses the Bearer JWT from the login response body (not a cookie) since the global API key can't create teams.
# Idempotent-ish: re-running skips existing teams/users; each step is best-effort and logs failures without aborting.
set -u
BASE_URL="${BASE_URL:-http://localhost:8080}"
ADMIN_USER="${ADMIN_USER:-admin}"
ADMIN_PASS="${ADMIN_PASS:-stirling}"
USER_COUNT="${USER_COUNT:-40}"
USER_PASS="${USER_PASS:-Password123!}"
PGHOST="${PGHOST:-postgres}"
PGUSER="${PGUSER:-stirling}"
PGPASSWORD="${PGPASSWORD:-stirling}"
PGDATABASE="${PGDATABASE:-stirling}"
export PGPASSWORD
TEAMS="Engineering Finance Legal Operations"
log() { echo "[seed] $*"; }
psqlq() { psql -h "$PGHOST" -U "$PGUSER" -d "$PGDATABASE" -tAc "$1" 2>/dev/null | tr -d '[:space:]'; }
# --- wait for the load balancer to serve a healthy app -----------------------
log "waiting for $BASE_URL ..."
i=0
until curl -fsS "$BASE_URL/api/v1/info/status" 2>/dev/null | grep -q UP; do
i=$((i+1)); [ "$i" -gt 120 ] && { log "timed out waiting for API"; exit 1; }
sleep 3
done
log "API is up"
# --- admin login -> Bearer token ---------------------------------------------
code=$(curl -sS -o /tmp/login.json -w '%{http_code}' \
-X POST "$BASE_URL/api/v1/auth/login" \
-H 'Content-Type: application/json' \
-d "{\"username\":\"$ADMIN_USER\",\"password\":\"$ADMIN_PASS\"}")
log "admin login: HTTP $code"
[ "$code" = "200" ] || { log "login failed: $(cat /tmp/login.json)"; exit 1; }
TOKEN=$(jq -r '.session.access_token' </tmp/login.json)
[ -n "$TOKEN" ] && [ "$TOKEN" != "null" ] || { log "no access_token in login response"; exit 1; }
auth() { curl -sS -H "Authorization: Bearer $TOKEN" "$@"; }
# --- teams -------------------------------------------------------------------
for t in $TEAMS; do
code=$(auth -o /dev/null -w '%{http_code}' -X POST "$BASE_URL/api/v1/team/create" \
--data-urlencode "name=$t")
log "team '$t': HTTP $code"
done
# Resolve team ids from the DB (no admin list endpoint self-hosted).
seed_team_ids=""
for t in $TEAMS; do
id=$(psqlq "select team_id from teams where name='$t' limit 1")
[ -n "$id" ] && seed_team_ids="$seed_team_ids $id"
done
set -- $seed_team_ids
team_count=$#
log "seedable team ids:$seed_team_ids (count=$team_count)"
# --- users: spread across teams, first two are admins ------------------------
created=0; failed=0
n=1
while [ "$n" -le "$USER_COUNT" ]; do
uname=$(printf "user%02d@stirling.test" "$n")
role="ROLE_USER"; [ "$n" -le 2 ] && role="ROLE_ADMIN"
team_id=""
if [ "$team_count" -gt 0 ]; then
idx=$(( (n % team_count) + 1 )); team_id=$(eval echo "\${$idx}")
fi
code=$(auth -o /tmp/user.json -w '%{http_code}' -X POST "$BASE_URL/api/v1/user/admin/saveUser" \
--data-urlencode "username=$uname" \
--data-urlencode "password=$USER_PASS" \
--data-urlencode "role=$role" \
${team_id:+--data-urlencode "teamId=$team_id"} \
--data-urlencode "authType=WEB" \
--data-urlencode "forceChange=false")
case "$code" in
200|201) created=$((created+1));;
409) log "user $uname already exists";;
*) failed=$((failed+1)); [ "$failed" -le 3 ] && log "user $uname failed HTTP $code: $(cat /tmp/user.json)";;
esac
n=$((n+1))
done
log "users created: $created (failed: $failed, requested: $USER_COUNT)"
# --- S3 connection -> the in-cluster MinIO 'policy-data' bucket ---------------
conn_body=$(cat <<JSON
{"integrationType":"S3","name":"MinIO policy bucket","scope":"SERVER","enabled":true,"locked":false,"defaultAccess":"ORG_ALL",
"config":{"bucket":"policy-data","region":"us-east-1","endpoint":"http://minio:9000","accessKeyId":"minioadmin","secretAccessKey":"minioadmin","pathStyleAccess":true}}
JSON
)
conn_id=$(auth -X POST "$BASE_URL/api/v1/integrations" -H 'Content-Type: application/json' -d "$conn_body" \
| jq -r '.id // empty' 2>/dev/null)
log "S3 connection id: ${conn_id:-<none>}"
# --- a scheduled S3 -> compress -> S3 policy ---------------------------------
if [ -n "${conn_id:-}" ]; then
src_body=$(cat <<JSON
{"name":"Incoming S3","type":"s3","enabled":true,
"options":{"connectionId":$conn_id,"prefix":"incoming/","mode":"consume"}}
JSON
)
src_id=$(auth -X POST "$BASE_URL/api/v1/sources" -H 'Content-Type: application/json' -d "$src_body" \
| jq -r '.id // empty' 2>/dev/null)
log "S3 source id: ${src_id:-<none>}"
if [ -n "${src_id:-}" ]; then
pol_body=$(cat <<JSON
{"name":"Compress incoming PDFs","enabled":true,
"trigger":{"type":"schedule","options":{"schedule":{"type":"every","count":5,"unit":"MINUTES"}}},
"sourceIds":["$src_id"],
"steps":[{"operation":"/api/v1/misc/compress-pdf","parameters":{}}],
"output":{"type":"s3","options":{"connectionId":$conn_id,"prefix":"processed/"}}}
JSON
)
code=$(auth -o /tmp/pol.json -w '%{http_code}' -X POST "$BASE_URL/api/v1/policies" \
-H 'Content-Type: application/json' -d "$pol_body")
log "policy create: HTTP $code $( [ "$code" != 200 ] && head -c 160 /tmp/pol.json )"
fi
# --- webhook source + policy (only if this build has the webhook type) -----
wh_body=$(cat <<JSON
{"name":"Partner webhook","type":"webhook","enabled":true,
"options":{"connectionId":$conn_id,"mode":"consume"}}
JSON
)
wh=$(auth -o /tmp/wh.json -w '%{http_code}' -X POST "$BASE_URL/api/v1/sources" \
-H 'Content-Type: application/json' -d "$wh_body")
if [ "$wh" = "200" ] || [ "$wh" = "201" ]; then
wh_url=$(jq -r '.options.webhookId // empty' </tmp/wh.json 2>/dev/null)
log "webhook source created (deliver to /api/v1/webhooks/$wh_url)"
else
log "webhook source not created (HTTP $wh) - expected on builds without the webhook branch"
fi
fi
log "seed complete."
log " login: $ADMIN_USER / $ADMIN_PASS at $BASE_URL"
log " users: user01..$(printf '%02d' "$USER_COUNT")@stirling.test / $USER_PASS"
@@ -1,69 +0,0 @@
#!/usr/bin/env bash
# Runs the multi-node regression suite (behave features/multinode) against the clustered stack: brings it up if needed, runs non-destructive scenarios then @destructive failover ones, and restores any killed node.
# Usage: ./run-multinode-regression.sh [--no-failover] [--no-seed]
# @known_gap scenarios are expected to fail - they mark work not yet done, so a non-zero exit is fine while those are open.
set -uo pipefail
cd "$(dirname "$0")"
COMPOSE="docker compose -f docker-compose-multinode.yml"
CUKE_DIR="../cucumber"
RUN_FAILOVER=1
SEED=1
for arg in "$@"; do
case "$arg" in
--no-failover) RUN_FAILOVER=0 ;;
--no-seed) SEED=0 ;;
esac
done
echo "==> Ensuring the multi-node stack is up..."
if ! docker inspect -f '{{.State.Health.Status}}' multinode-stirling-1 2>/dev/null | grep -q healthy; then
./start-multinode-test.sh $([ "$SEED" = 0 ] && echo --no-seed) || exit 1
elif [ "$SEED" = 1 ]; then
echo " stack already up; seeding (idempotent)..."
$COMPOSE --profile seed run --rm seed >/dev/null 2>&1 || echo " (seed reported issues, continuing)"
fi
echo "==> Checking Python + behave..."
PY="${PYTHON:-python}"
command -v "$PY" >/dev/null || PY=python3
if ! "$PY" -c "import behave" 2>/dev/null; then
echo " installing test deps..."
"$PY" -m pip install -q -r "$CUKE_DIR/requirements.txt" || {
echo " could not install behave; install $CUKE_DIR/requirements.txt manually"; exit 1; }
fi
REPORT_DIR="$(pwd)/multinode/regression-report"
mkdir -p "$REPORT_DIR"
run_behave() { # $1=tags $2=label
echo "==> behave features/multinode --tags='$1' ($2)"
# behave.ini excludes features/multinode by default; -e here overrides that while still excluding the licence-gated enterprise suite.
( cd "$CUKE_DIR" && "$PY" -m behave features/multinode -e "features/enterprise" \
--tags="$1" --no-capture --format plain --format html --outfile "$REPORT_DIR/$2.html" )
return $?
}
rc=0
run_behave "~@destructive" "core" || rc=1
if [ "$RUN_FAILOVER" = 1 ]; then
run_behave "@destructive" "failover" || rc=1
echo "==> Restoring any killed nodes..."
$COMPOSE up -d >/dev/null 2>&1
for n in multinode-stirling-1 multinode-stirling-2; do
for i in $(seq 1 24); do
[ "$(docker inspect -f '{{.State.Health.Status}}' "$n" 2>/dev/null)" = "healthy" ] && break
sleep 5
done
done
fi
echo
echo "============================================================"
echo " Regression run complete. Reports: $REPORT_DIR"
echo " Exit $rc (non-zero = at least one scenario failed;"
echo " @known_gap scenarios are expected to fail - see the report)."
echo " Stack left running: http://localhost:8080 (admin / stirling)"
echo "============================================================"
exit $rc
-65
View File
@@ -1,65 +0,0 @@
#!/usr/bin/env bash
# Brings up the multi-node stack (Postgres/Valkey/MinIO/2 app nodes/nginx LB), seeds teams/users/an S3 connection/policies, then leaves it running for manual testing.
# Usage: ./start-multinode-test.sh [--no-seed | --down]
set -euo pipefail
cd "$(dirname "$0")"
COMPOSE="docker compose -f docker-compose-multinode.yml"
if [ "${1:-}" = "--down" ]; then
echo "Tearing down multi-node stack + volumes..."
$COMPOSE --profile seed down -v --remove-orphans
exit 0
fi
SEED=1
[ "${1:-}" = "--no-seed" ] && SEED=0
# Cluster mode is licence-gated. Without a valid key the nodes fail the cluster licence gate at boot.
if [ -z "${PREMIUM_KEY:-}" ]; then
echo "WARNING: PREMIUM_KEY is not set - cluster mode needs a valid enterprise/pro licence key."
echo " Run: export PREMIUM_KEY=<your test licence key> before starting."
fi
echo "==> Building the Stirling image (first run compiles the app; be patient)..."
$COMPOSE build
echo "==> Starting Postgres + Valkey + MinIO + 2 app nodes + nginx..."
$COMPOSE up -d
echo "==> Waiting for both app nodes to report healthy..."
for node in multinode-stirling-1 multinode-stirling-2; do
for i in $(seq 1 60); do
status=$(docker inspect -f '{{.State.Health.Status}}' "$node" 2>/dev/null || echo "starting")
[ "$status" = "healthy" ] && { echo " $node: healthy"; break; }
[ "$i" = "60" ] && { echo " $node did not become healthy; see: $COMPOSE logs $node"; exit 1; }
sleep 5
done
done
if [ "$SEED" = "1" ]; then
echo "==> Seeding teams / users / S3 connection / policies..."
$COMPOSE --profile seed run --rm seed || echo " (seed reported issues; check output above)"
fi
cat <<EOF
============================================================================
Multi-node Stirling is UP.
App (via load balancer): http://localhost:8080 (admin / stirling)
MinIO console: http://localhost:9001 (minioadmin / minioadmin)
Postgres: localhost:5434 (stirling / stirling, db 'stirling')
Seeded users: user01..user40@stirling.test / Password123!
Global API key: multinode-test-key (header: X-API-KEY)
Try it:
./validate-multinode-test.sh # multi-node smoke tests (optional)
$COMPOSE logs -f stirling-1 # tail a node
./start-multinode-test.sh --down # stop + wipe
Nodes are reachable directly for cross-node checks:
docker compose -f docker-compose-multinode.yml exec stirling-1 curl -s localhost:8080/api/v1/info/status
============================================================================
EOF
@@ -1,81 +0,0 @@
#!/usr/bin/env bash
# Multi-node smoke tests against a running stack (start-multinode-test.sh): load-balancer spread, cross-node JWT validation (signing keys persist in the shared DB), and processor state visible from every node.
# Auth: extracts the Bearer JWT from the login body via sed (no jq needed host-side) and hits nodes directly with docker exec.
# Non-destructive - safe to re-run against the stack at http://localhost:8080.
set -uo pipefail
cd "$(dirname "$0")"
LB="http://localhost:8080"
ADMIN_USER="admin"; ADMIN_PASS="stirling"
NODES="multinode-stirling-1 multinode-stirling-2"
# An authed, admin-visible endpoint that returns 200 with a valid token, 401 without.
PROBE="/api/v1/sources"
pass=0; fail=0
ok() { echo " PASS - $*"; pass=$((pass+1)); }
bad() { echo " FAIL - $*"; fail=$((fail+1)); }
login() { # -> prints the bearer token
curl -s -X POST "$LB/api/v1/auth/login" -H 'Content-Type: application/json' \
-d "{\"username\":\"$ADMIN_USER\",\"password\":\"$ADMIN_PASS\"}" \
| sed -n 's/.*"access_token":"\([^"]*\)".*/\1/p'
}
echo "== 1. Load balancer spreads requests across nodes =="
served=$(for i in $(seq 1 12); do
curl -s -D - -o /dev/null "$LB/api/v1/info/status" | tr -d '\r' | awk -F': ' '/^X-Served-By/{print $2}'
done | sort -u)
distinct=$(printf '%s\n' "$served" | grep -c .)
echo " upstreams seen: $(printf '%s' "$served" | paste -sd, -)"
[ "$distinct" -ge 2 ] && ok "LB round-robined across $distinct nodes" \
|| bad "only $distinct node(s) served (expected >=2; is X-Served-By enabled?)"
echo "== 2. A JWT from the LB is accepted by BOTH nodes directly (shared signing key) =="
jwt=$(login)
if [ -z "$jwt" ]; then
bad "admin login via LB failed - cannot test cross-node JWT"
else
ok "logged in via LB, got a JWT (${#jwt} chars)"
for n in $NODES; do
hc=$(docker exec "$n" curl -s -o /dev/null -w '%{http_code}' \
-H "Authorization: Bearer $jwt" "http://localhost:8080$PROBE" 2>/dev/null)
[ "$hc" = "200" ] && ok "$n accepted the foreign-minted JWT (HTTP 200)" \
|| bad "$n rejected the JWT (HTTP $hc) - keys not shared across nodes"
done
fi
echo "== 3. Processor state is shared: each node sees the same sources =="
count_of() { # $1=node -> number of sources that node reports
docker exec "$1" curl -s -H "Authorization: Bearer $jwt" "http://localhost:8080$PROBE" 2>/dev/null \
| grep -o '"id"' | grep -c .
}
a=$(count_of multinode-stirling-1); b=$(count_of multinode-stirling-2)
echo " stirling-1 sources: $a stirling-2 sources: $b"
if [ "$a" -gt 0 ] && [ "$a" = "$b" ]; then
ok "both nodes report the same $a sources (shared DB)"
else
bad "source counts differ or zero across nodes ($a vs $b)"
fi
echo "== 4. Seeded org is in the shared DB =="
users=$(docker exec multinode-postgres psql -U stirling -d stirling -tAc "select count(*) from users" 2>/dev/null | tr -d '[:space:]')
teams=$(docker exec multinode-postgres psql -U stirling -d stirling -tAc "select count(*) from teams" 2>/dev/null | tr -d '[:space:]')
conns=$(docker exec multinode-postgres psql -U stirling -d stirling -tAc "select count(*) from integration_configs" 2>/dev/null | tr -d '[:space:]')
echo " users=$users teams=$teams integration_configs=$conns"
[ "${users:-0}" -ge 40 ] && ok "$users users present" || bad "only ${users:-0} users (did the seed run?)"
[ "${conns:-0}" -ge 1 ] && ok "$conns S3/integration connection(s) present" || bad "no integration connections"
echo "== 5. Cross-node encrypted-secret read (shared credential key) =="
# The seed's S3 secret was encrypted by whichever node handled it; fetching it via the LB (either node) and getting a masked, non-error view proves the credential key is shared, not per-node.
lc=$(curl -s -o /tmp/mn_conns.json -w '%{http_code}' -H "Authorization: Bearer $jwt" "$LB/api/v1/integrations")
if [ "$lc" = "200" ] && grep -q '"integrationType"' /tmp/mn_conns.json; then
ok "integration list decrypts through the LB (HTTP 200) - credential key is shared"
else
bad "integration list failed (HTTP $lc) - credential key may not be shared across nodes"
fi
echo
echo "============================================================"
echo " Multi-node validation: $pass passed, $fail failed."
echo " Stack left running: $LB (admin / stirling)"
echo "============================================================"
[ "$fail" -eq 0 ]
+7 -3
View File
@@ -1,5 +1,9 @@
[behave]
# features/enterprise (needs a commercial licence) and features/multinode (needs the clustered stack in testing/compose/docker-compose-multinode.yml) are excluded by default.
# Run either directly, e.g. `python -m behave features/enterprise`.
exclude_re = features/(enterprise|multinode)
# Enterprise and premium-licensed features live in features/enterprise/.
# They are excluded from the default CI run because the test environment
# does not have a commercial licence. To run them explicitly:
#
# python -m behave features/enterprise
#
exclude_re = features/enterprise
tags = ~@manual
@@ -1,18 +0,0 @@
@multinode @health
Feature: Multi-node cluster health
Every node must come up healthy and join the shared Valkey backplane - the precondition for the rest of the suite (stack: testing/compose/docker-compose-multinode.yml).
@smoke
Scenario: Both application nodes are healthy
Given the multi-node stack is running
@smoke
Scenario: Both nodes joined the Valkey backplane in cluster mode
Given the multi-node stack is running
And both nodes are cluster members using the Valkey backplane
Scenario: The load balancer answers the health endpoint
Given the multi-node stack is running
When I request "/api/v1/info/status" 4 times through the load balancer
Then every load-balanced response should be 200
@@ -1,12 +0,0 @@
@multinode @connections
Feature: S3 connections across nodes
Integration connections are team-scoped, secret-encrypted rows in the shared DB - a connection created via the LB must resolve (secret masked) from every node and deletion must remove it cluster-wide.
Scenario: A connection created via the LB resolves everywhere, then deletes cluster-wide
Given the multi-node stack is running
And I am logged in as admin
When I create an S3 connection named "regr_conn_alpha" via the load balancer
Then the connection "regr_conn_alpha" should resolve from every node with its secret masked
When I delete the connection via the load balancer
Then the connection should be gone from every node
@@ -1,14 +0,0 @@
@multinode @auth
Feature: Cross-node authentication
JWTs are signed RS256 and verified by key id, so the signing keys (stored in the shared DB, private half encrypted) must be shared for a token minted on one node to validate on another.
Scenario: A token minted through the LB validates on every node directly
Given the multi-node stack is running
And I am logged in as admin
Then the current token should be accepted by every node
Scenario: Signing keys are persisted in the shared database, encrypted
Given the multi-node stack is running
Then the signing keys should be stored in the shared database
And every stored private key should be encrypted at rest
@@ -1,21 +0,0 @@
@multinode @failover @destructive
Feature: Node failover and recovery
Losing a node must not take the service down: the LB drains it and the survivor keeps serving, then the node rejoins healthy on restart. These scenarios kill/restart containers, so they are @destructive - pass --tags=failover to run them.
@destructive
Scenario: The load balancer keeps serving when a node dies
Given the multi-node stack is running
When I kill node "2"
Then the load balancer should still serve requests
When I restart node "2"
Then node "2" should become healthy again within 120s
@destructive
Scenario: A recovered node validates tokens minted while it was down
Given the multi-node stack is running
And I am logged in as admin
When I kill node "2"
And I restart node "2"
Then node "2" should become healthy again within 120s
And the current token should be accepted by every node
@@ -1,15 +0,0 @@
@multinode @loadbalancing
Feature: Load balancer distributes across nodes
Requests must spread across nodes via round-robin with no session affinity, and a client bounced between nodes must never see a spurious failure since the app is stateless at the HTTP layer.
Scenario: Requests are spread across both nodes
Given the multi-node stack is running
When I request "/api/v1/info/status" 12 times through the load balancer
Then the requests should be served by at least 2 distinct nodes
Scenario: An authenticated client bounced between nodes never gets a spurious 401
Given the multi-node stack is running
And I am logged in as admin
When I request "/api/v1/sources" 12 times through the load balancer
Then every load-balanced response should be 200
@@ -1,19 +0,0 @@
@multinode @policy
Feature: Policy management across nodes
A policy is durable state in the shared DB, so creating, editing, or deleting it on any node must be reflected on every other node, and both nodes must register the same trigger types.
Scenario: A policy created on one node is edited and deleted from the other
Given the multi-node stack is running
And I am logged in as admin
When I create a policy named "regr_pol_alpha" on node "1"
Then the policy "regr_pol_alpha" should be visible from every node
When I rename the policy "regr_pol_alpha" to "regr_pol_beta" via the load balancer
Then the policy "regr_pol_beta" should be visible from every node
When I delete the policy "regr_pol_beta" on node "2"
Then the policy "regr_pol_beta" should be absent from every node
Scenario: Every node registers the same trigger types
Given the multi-node stack is running
And I am logged in as admin
Then the trigger registry should be identical across nodes
@@ -1,10 +0,0 @@
@multinode @processor
Feature: Policy run visibility across nodes
Run state is projected into the shared job store, so the run-view endpoints (/policies/runs, /run/{runId}) show a run from any node, not only the one that executed it - no sticky-session LB required.
Scenario: A run executed on one node is visible from every node
Given the multi-node stack is running
And I am logged in as admin
When I run the policy "Compress incoming PDFs" on node "1"
Then the run should be visible from every node
@@ -1,20 +0,0 @@
@multinode @processor
Feature: Processor pipeline and exactly-once ledger
A shared processed-file ledger with an atomic (identity_hash, policy_id) primary key ensures exactly-once processing when several nodes race the same scheduled policy; depends on the seeded "Compress incoming PDFs" policy.
@slow
Scenario: Files dropped in the source are processed across the cluster when both
nodes trigger the policy at the same instant
Given the multi-node stack is running
And I am logged in as admin
And the processor workspace is clean
When I drop 5 PDF file(s) into the S3 source under "incoming/"
And I trigger the policy "Compress incoming PDFs" on every node simultaneously
Then within 90s every dropped file should be processed across the cluster
Scenario: The ledger's atomic claim enforces exactly-once
Given the multi-node stack is running
Then a duplicate ledger claim for the same file and policy is rejected
# Note: output lands in the source's own bucket so it can be re-ingested; too flaky to assert under consume-mode pruning.
@@ -1,9 +0,0 @@
@multinode @ratelimit
Feature: Shared rate limiting via the backplane
Rate limits must be enforced across all nodes, not per node, or a client's effective limit multiplies by the node count - cluster mode routes counters through Valkey (ValkeyRateLimitStore).
Scenario: Backplane counters are held in Valkey, not per node
Given the multi-node stack is running
And both nodes are cluster members using the Valkey backplane
Then the rate-limit counter should be shared across nodes
@@ -1,20 +0,0 @@
@multinode @state
Feature: Shared state across nodes
All durable state lives in the shared database, so an object created through the LB (landing on whichever node) is immediately visible from every other node - the cluster is a single logical system.
Scenario: A team created through the LB lands in the shared database
Given the multi-node stack is running
And I am logged in as admin
When I create a team named "regr_shared_team" through the load balancer
Then the team "regr_shared_team" should exist in the shared database
Scenario: Every node reports the same sources
Given the multi-node stack is running
And I am logged in as admin
Then every node should report the same number of sources
Scenario: The seeded org is present in the shared database
Given the multi-node stack is running
Then the "users" table should contain at least 40 row(s)
And the "integration_configs" table should contain at least 1 row(s)
@@ -1,25 +0,0 @@
@multinode @sources
Feature: Source management across nodes
Sources are durable state. A source created on one node must be usable from
every node, and the guard that stops a referenced source from being deleted
must hold across nodes - even when the source and the referencing policy were
created on different nodes.
Scenario: A source created on one node is visible and deletable from the other
Given the multi-node stack is running
And I am logged in as admin
When I create an S3 source named "regr_src_alpha" on node "1"
Then the source "regr_src_alpha" should be visible from every node
When I delete the source "regr_src_alpha" on node "2"
Then the source "regr_src_alpha" should be absent from every node
Scenario: A referenced source cannot be deleted from another node
Given the multi-node stack is running
And I am logged in as admin
When I create an S3 source named "regr_src_ref" on node "1"
And I create a policy named "regr_pol_ref" referencing source "regr_src_ref" via the load balancer
Then deleting the source "regr_src_ref" from node "2" is rejected because it is referenced
When I delete the policy "regr_pol_ref" on node "2"
And I delete the source "regr_src_ref" on node "1"
Then the source "regr_src_ref" should be absent from every node
@@ -1,595 +0,0 @@
"""Step definitions for the multi-node regression suite: drive the stack (testing/compose/docker-compose-multinode.yml) via the LB, docker exec curl/psql on individual nodes, and a throwaway minio/mc container."""
import io
import json
import subprocess
import time
import uuid
import requests
from behave import given, then, when
LB_URL = "http://localhost:8080"
NODES = ["multinode-stirling-1", "multinode-stirling-2"]
PG = "multinode-postgres"
MINIO = "multinode-minio"
BUCKET = "policy-data"
SOURCE_PREFIX = "incoming/"
OUTPUT_PREFIX = "processed/"
ADMIN_USER = "admin"
ADMIN_PASS = "stirling"
# --------------------------------------------------------------------------- helpers
def _sh(args, stdin=None, timeout=60):
"""Run a command, return (returncode, stdout, stderr)."""
r = subprocess.run(
args, input=stdin, capture_output=True, timeout=timeout,
text=(stdin is None or isinstance(stdin, str)),
)
out = r.stdout if isinstance(r.stdout, str) else r.stdout.decode("utf-8", "replace")
err = r.stderr if isinstance(r.stderr, str) else r.stderr.decode("utf-8", "replace")
return r.returncode, out, err
def _psql(query):
"""Run a query against the shared Postgres, return the raw tab/newline output (trimmed)."""
rc, out, err = _sh(
["docker", "exec", PG, "psql", "-U", "stirling", "-d", "stirling", "-tAc", query]
)
assert rc == 0, f"psql failed: {err.strip() or out.strip()}"
return out.strip()
def _psql_int(query):
val = _psql(query)
return int(val) if val else 0
def _network():
rc, out, _ = _sh(
["docker", "inspect", "-f",
"{{range $k,$v := .NetworkSettings.Networks}}{{$k}}{{end}}", NODES[0]]
)
return out.strip() or "compose_stirling-multinode"
def _token(context):
tok = getattr(context, "jwt_token", None)
assert tok, "No JWT token in context - use 'Given I am logged in as admin' first."
return tok
def _curl_on_node(node, method, path, token=None, data=None, content_type=None, timeout=30):
"""Hit a node's own :8080 from inside the cluster (bypasses the LB). Returns (status, body)."""
cmd = ["docker", "exec", node, "curl", "-s", "-w", "\n%{http_code}", "-X", method]
if token:
cmd += ["-H", f"Authorization: Bearer {token}"]
if content_type:
cmd += ["-H", f"Content-Type: {content_type}"]
if data is not None:
cmd += ["--data", data]
cmd.append(f"http://localhost:8080{path}")
rc, out, err = _sh(cmd, timeout=timeout)
# Body then a newline then the status code; split the status off the end.
text = out.rstrip("\n")
body, _, status = text.rpartition("\n")
return (int(status.strip()) if status.strip().isdigit() else 0), body
def _node(idx):
return NODES[int(idx) - 1]
def _names_on_node(node, path, token):
"""GET a list endpoint on a node and return the set of resource names it reports."""
status, body = _curl_on_node(node, "GET", path, token=token)
assert status == 200, f"{node} GET {path} returned HTTP {status}: {body[:150]}"
try:
data = json.loads(body)
except ValueError:
return set()
items = data if isinstance(data, list) else data.get(path.rsplit("/", 1)[-1], []) \
or data.get("sources", []) or data.get("policies", [])
return {i.get("name") for i in items if isinstance(i, dict)}
def _policy_body(name, source_ids=None, enabled=True):
return json.dumps({
"name": name, "enabled": enabled, "trigger": None,
"sourceIds": source_ids or [],
"steps": [{"operation": "/api/v1/misc/compress-pdf", "parameters": {}}],
"output": {"type": "inline", "options": {}},
})
def _any_connection_id(context):
"""Id of any usable S3 connection (the seed creates one). Cached on the context."""
cid = getattr(context, "_seed_conn_id", None)
if cid:
return cid
r = requests.get(f"{LB_URL}/api/v1/integrations",
headers={"Authorization": f"Bearer {_token(context)}"}, timeout=15)
assert r.status_code == 200, f"list integrations failed: HTTP {r.status_code}"
s3 = next((c for c in r.json() if c.get("integrationType") == "S3"), None)
assert s3, "no S3 connection available (did the seed run?)"
context._seed_conn_id = s3["id"]
return s3["id"]
def _s3_source_body(name, connection_id):
# Folder sources are config-gated; S3 sources against the seeded connection always work.
return json.dumps({
"name": name, "type": "s3",
"options": {"connectionId": connection_id, "prefix": "regr/", "mode": "snapshot"},
"enabled": True,
})
def _source_id_by_name(context, name):
r = requests.get(f"{LB_URL}/api/v1/sources",
headers={"Authorization": f"Bearer {_token(context)}"}, timeout=15)
assert r.status_code == 200, f"list sources failed: HTTP {r.status_code}"
return next((s["id"] for s in r.json().get("sources", []) if s.get("name") == name), None)
def _s3_connection_body(name):
return json.dumps({
"integrationType": "S3", "name": name, "scope": "SERVER", "enabled": True,
"locked": False, "defaultAccess": "ORG_ALL",
"config": {"bucket": BUCKET, "region": "us-east-1", "endpoint": "http://minio:9000",
"accessKeyId": "minioadmin", "secretAccessKey": "minioadmin",
"pathStyleAccess": True},
})
def _lb_login(context):
r = requests.post(f"{LB_URL}/api/v1/auth/login",
json={"username": ADMIN_USER, "password": ADMIN_PASS}, timeout=15)
assert r.status_code == 200, f"admin login via LB failed: HTTP {r.status_code}"
context.jwt_token = r.json()["session"]["access_token"]
def _pdf_bytes(marker):
"""A minimal valid single-page PDF carrying a unique marker (so outputs are identifiable)."""
try:
from reportlab.pdfgen import canvas
buf = io.BytesIO()
c = canvas.Canvas(buf)
c.drawString(100, 750, f"multinode-regression {marker}")
c.showPage()
c.save()
return buf.getvalue()
except Exception:
# Fallback: a hand-rolled minimal PDF if reportlab is unavailable.
return (b"%PDF-1.4\n1 0 obj<</Type/Catalog/Pages 2 0 R>>endobj\n"
b"2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj\n"
b"3 0 obj<</Type/Page/Parent 2 0 R/MediaBox[0 0 612 792]>>endobj\n"
b"trailer<</Root 1 0 R>>\n%%EOF")
def _mc(context, script, stdin=None):
"""Run an mc script in a throwaway minio/mc container on the cluster network."""
net = getattr(context, "_net", None) or _network()
context._net = net
full = f"mc alias set local http://minio:9000 minioadmin minioadmin >/dev/null 2>&1 && {script}"
args = ["docker", "run", "-i", "--rm", "--network", net, "--entrypoint", "/bin/sh",
"minio/mc", "-c", full]
return _sh(args, stdin=stdin, timeout=90)
def _policy_id_by_name(context, name):
r = requests.get(f"{LB_URL}/api/v1/policies",
headers={"Authorization": f"Bearer {_token(context)}"}, timeout=15)
assert r.status_code == 200, f"list policies failed: HTTP {r.status_code}"
data = r.json()
items = data if isinstance(data, list) else data.get("policies", [])
for p in items:
if p.get("name") == name:
return p.get("id")
return None
# --------------------------------------------------------------------------- preconditions
@given("the multi-node stack is running")
def step_stack_running(context):
for node in NODES:
rc, out, _ = _sh(["docker", "inspect", "-f", "{{.State.Health.Status}}", node])
assert rc == 0 and out.strip() == "healthy", f"{node} is not healthy (got '{out.strip()}')"
context._net = _network()
@given("both nodes are cluster members using the Valkey backplane")
def step_cluster_members(context):
for node in NODES:
rc, out, err = _sh(["docker", "logs", node])
logs = out + err
assert "backplane=valkey" in logs, f"{node} did not join the Valkey backplane"
# --------------------------------------------------------------------------- load balancer
@when('I request "{endpoint}" {count:d} times through the load balancer')
def step_lb_requests(context, endpoint, count):
context._served_by = []
context._lb_statuses = []
headers = {}
if getattr(context, "jwt_token", None):
headers["Authorization"] = f"Bearer {context.jwt_token}"
for _ in range(count):
r = requests.get(f"{LB_URL}{endpoint}", headers=headers, timeout=15)
context._lb_statuses.append(r.status_code)
node = r.headers.get("X-Served-By")
if node:
context._served_by.append(node)
@then("the requests should be served by at least {n:d} distinct nodes")
def step_distinct_nodes(context, n):
distinct = set(context._served_by)
assert len(distinct) >= n, (
f"expected >= {n} distinct upstreams, saw {sorted(distinct)} "
f"(is the X-Served-By header configured on the LB?)")
@then("every load-balanced response should be {code:d}")
def step_all_lb_ok(context, code):
bad = [s for s in context._lb_statuses if s != code]
assert not bad, f"expected all {code}, got failures: {bad}"
# --------------------------------------------------------------------------- cross-node auth
@then("the current token should be accepted by every node")
def step_token_every_node(context):
token = _token(context)
for node in NODES:
status, body = _curl_on_node(node, "GET", "/api/v1/sources", token=token)
assert status == 200, f"{node} rejected the LB-minted token (HTTP {status}): {body[:200]}"
@then("the signing keys should be stored in the shared database")
def step_keys_in_db(context):
assert _psql_int("select count(*) from jwt_signing_keys") >= 1, \
"no rows in jwt_signing_keys - keys are not persisted in the shared DB"
@then("every stored private key should be encrypted at rest")
def step_keys_encrypted(context):
# A plaintext PKCS#8 RSA key base64 begins with 'MII'; an encrypted blob does not.
plaintext = _psql_int("select count(*) from jwt_signing_keys where signing_key like 'MII%'")
assert plaintext == 0, f"{plaintext} signing key(s) look like plaintext PKCS#8 (not encrypted)"
# --------------------------------------------------------------------------- shared state
@when('I create a team named "{name}" through the load balancer')
def step_create_team(context, name):
context._team_name = name
r = requests.post(f"{LB_URL}/api/v1/team/create",
headers={"Authorization": f"Bearer {_token(context)}"},
data={"name": name}, timeout=15)
assert r.status_code in (200, 201, 409), f"create team failed: HTTP {r.status_code}"
@then('the team "{name}" should exist in the shared database')
def step_team_in_db(context, name):
n = _psql_int(f"select count(*) from teams where name = '{name}'")
assert n >= 1, f"team '{name}' not found in the shared DB"
@then("every node should report the same number of sources")
def step_same_sources(context):
token = _token(context)
counts = {}
for node in NODES:
status, body = _curl_on_node(node, "GET", "/api/v1/sources", token=token)
assert status == 200, f"{node} /sources returned HTTP {status}"
counts[node] = body.count('"id"')
values = set(counts.values())
assert len(values) == 1 and values != {0}, f"source counts differ across nodes: {counts}"
@then('the "{table}" table should contain at least {n:d} row(s)')
def step_table_rows(context, table, n):
got = _psql_int(f"select count(*) from {table}")
assert got >= n, f"{table} has {got} rows, expected >= {n}"
# --------------------------------------------------------------------------- processor / ledger
@given("the processor workspace is clean")
def step_clean_workspace(context):
# Start from a known state so file/output/ledger counts reflect only this scenario.
_mc(context, f"mc rm --recursive --force local/{BUCKET}/{SOURCE_PREFIX} || true")
_mc(context, f"mc rm --recursive --force local/{BUCKET}/{OUTPUT_PREFIX} || true")
_psql("delete from policy_processed_files")
@when('I drop {count:d} PDF file(s) into the S3 source under "{prefix}"')
def step_drop_files(context, count, prefix):
context._dropped = []
for _ in range(count):
marker = uuid.uuid4().hex[:12]
key = f"{prefix}regr-{marker}.pdf"
rc, out, err = _mc(context, f"mc pipe local/{BUCKET}/{key}", stdin=_pdf_bytes(marker))
assert rc == 0, f"failed to upload {key}: {err.strip() or out.strip()}"
context._dropped.append(key)
@when('I trigger the policy "{name}" on every node simultaneously')
def step_trigger_all_nodes(context, name):
pid = _policy_id_by_name(context, name)
assert pid, f"policy '{name}' not found"
context._policy_id = pid
token = _token(context)
# Fire the trigger on both nodes as close together as possible to race the ledger claim.
for node in NODES:
_curl_on_node(node, "POST", f"/api/v1/policies/{pid}/trigger", token=token, timeout=15)
@then("within {seconds:d}s every dropped file should be processed across the cluster")
def step_files_processed(context, seconds):
# Consume-mode deletes processed files, so an empty source prefix signals every file was processed (ledger rows are pruned too, so counting them would be racy).
deadline = time.monotonic() + seconds
remaining = None
while time.monotonic() < deadline:
rc, out, _ = _mc(context, f"mc ls --recursive local/{BUCKET}/{SOURCE_PREFIX} | wc -l")
remaining = int(out.strip() or "0") if rc == 0 else -1
if remaining == 0:
break
time.sleep(3)
assert remaining == 0, (
f"{remaining} of {len(context._dropped)} dropped files were still unprocessed after "
f"{seconds}s")
for node in NODES:
rc, out, _ = _sh(["docker", "inspect", "-f", "{{.State.Status}}", node])
assert out.strip() == "running", f"{node} crashed during concurrent processing"
@then("a duplicate ledger claim for the same file and policy is rejected")
def step_ledger_claim_atomic(context):
# Exactly-once relies on the (identity_hash, policy_id) primary key: two nodes claiming the same file both insert it, but only one wins; this proves the constraint rejects the second claim.
ihash = "regr-" + uuid.uuid4().hex
pol = "regr-policy-" + uuid.uuid4().hex[:8]
insert = (f"insert into policy_processed_files (identity_hash, policy_id, status, attempts) "
f"values ('{ihash}', '{pol}', 'PROCESSING', 1)")
_psql(insert) # first claim wins
rc, out, err = _sh(["docker", "exec", PG, "psql", "-U", "stirling", "-d", "stirling",
"-tAc", insert]) # second claim must be rejected
_psql(f"delete from policy_processed_files where identity_hash = '{ihash}'")
assert rc != 0 and "duplicate key" in (out + err).lower(), (
"a second claim for the same file and policy was NOT rejected - the ledger's exactly-once "
"guarantee is not enforced by the primary key")
# --------------------------------------------------------------------------- policy run coordination (gap)
@when('I run the policy "{name}" on node "{idx}"')
def step_run_policy_on_node(context, name, idx):
node = _node(idx)
context._run_node = node
pid = _policy_id_by_name(context, name)
assert pid, f"policy '{name}' not found"
# Drop an input so the trigger actually produces a run (the source is otherwise empty).
marker = uuid.uuid4().hex[:12]
_mc(context, f"mc pipe local/{BUCKET}/{SOURCE_PREFIX}runvis-{marker}.pdf",
stdin=_pdf_bytes(marker))
status, _ = _curl_on_node(node, "POST", f"/api/v1/policies/{pid}/trigger", token=_token(context))
assert status in (200, 202), f"triggering the policy on {node} failed: HTTP {status}"
# Grab the runId that node recorded for the run it just executed.
context._run_id = None
for _ in range(8):
s, body = _curl_on_node(node, "GET", "/api/v1/policies/runs", token=_token(context))
runs = json.loads(body) if s == 200 and body.strip().startswith("[") else []
if runs:
context._run_id = runs[0]["runId"]
break
time.sleep(2)
assert context._run_id, f"{node} recorded no run after triggering '{name}'"
@then("the run should be visible from every node")
def step_run_visible_every(context):
for node in NODES:
status, body = _curl_on_node(node, "GET", "/api/v1/policies/runs", token=_token(context))
assert status == 200, f"{node} /policies/runs returned HTTP {status}"
run_ids = [r.get("runId") for r in json.loads(body)] if body.strip().startswith("[") else []
assert context._run_id in run_ids, (
f"run {context._run_id} (executed on {context._run_node}) is not visible from {node} - "
f"PolicyRunRegistry is a per-node in-JVM map, so run status and cancellation do not "
f"cross nodes")
# --------------------------------------------------------------------------- rate limiting
@then("the rate-limit counter should be shared across nodes")
def step_ratelimit_shared(context):
# In cluster mode the ValkeyRateLimitStore holds counters in Valkey; probe that a key exists.
net = context._net or _network()
rc, out, err = _sh(["docker", "run", "--rm", "--network", net, "--entrypoint", "/bin/sh",
"valkey/valkey:8-alpine", "-c",
"valkey-cli -h valkey keys '*'"], timeout=30)
assert rc == 0, f"valkey probe failed: {err.strip()}"
assert out.strip(), "no keys in Valkey - rate-limit/backplane state is not shared"
# --------------------------------------------------------------------------- failover
@when('I kill node "{idx}"')
def step_kill_node(context, idx):
node = NODES[int(idx) - 1]
context._killed = getattr(context, "_killed", [])
_sh(["docker", "kill", node])
context._killed.append(node)
time.sleep(2)
@then("the load balancer should still serve requests")
def step_lb_survives(context):
ok = 0
for _ in range(6):
try:
r = requests.get(f"{LB_URL}/api/v1/info/status", timeout=10)
if r.status_code == 200:
ok += 1
except requests.RequestException:
pass
assert ok >= 4, f"LB served only {ok}/6 requests with a node down (does it drain dead nodes?)"
@when('I restart node "{idx}"')
def step_restart_node(context, idx):
node = NODES[int(idx) - 1]
_sh(["docker", "start", node])
context._killed = [n for n in getattr(context, "_killed", []) if n != node]
@then('node "{idx}" should become healthy again within {seconds:d}s')
def step_node_recovers(context, idx, seconds):
node = NODES[int(idx) - 1]
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
rc, out, _ = _sh(["docker", "inspect", "-f", "{{.State.Health.Status}}", node])
if out.strip() == "healthy":
return
time.sleep(5)
raise AssertionError(f"{node} did not become healthy within {seconds}s")
# ------------------------------------------------- policy / source / connection management (CRUD)
def _auth(context):
return {"Authorization": f"Bearer {_token(context)}"}
# --- policies ---
@when('I create a policy named "{name}" on node "{idx}"')
def step_create_policy_on_node(context, name, idx):
status, body = _curl_on_node(_node(idx), "POST", "/api/v1/policies", token=_token(context),
data=_policy_body(name), content_type="application/json")
assert status in (200, 201), f"create policy on {_node(idx)} failed: HTTP {status}: {body[:200]}"
@when('I create a policy named "{name}" referencing source "{src}" via the load balancer')
def step_create_policy_ref(context, name, src):
sid = _source_id_by_name(context, src)
assert sid, f"source '{src}' not found"
r = requests.post(f"{LB_URL}/api/v1/policies",
headers={**_auth(context), "Content-Type": "application/json"},
data=_policy_body(name, [sid]), timeout=15)
assert r.status_code in (200, 201), f"create referencing policy failed: HTTP {r.status_code}"
@when('I rename the policy "{old}" to "{new}" via the load balancer')
def step_rename_policy(context, old, new):
pid = _policy_id_by_name(context, old)
assert pid, f"policy '{old}' not found"
pol = requests.get(f"{LB_URL}/api/v1/policies/{pid}", headers=_auth(context), timeout=15).json()
pol["name"] = new
r = requests.post(f"{LB_URL}/api/v1/policies",
headers={**_auth(context), "Content-Type": "application/json"},
json=pol, timeout=15)
assert r.status_code in (200, 201), f"rename policy failed: HTTP {r.status_code}"
@when('I delete the policy "{name}" on node "{idx}"')
def step_delete_policy_on_node(context, name, idx):
pid = _policy_id_by_name(context, name)
assert pid, f"policy '{name}' not found"
status, body = _curl_on_node(_node(idx), "DELETE", f"/api/v1/policies/{pid}", token=_token(context))
assert status in (200, 204), f"delete policy on {_node(idx)} failed: HTTP {status}"
@then('the policy "{name}" should be visible from every node')
def step_policy_visible(context, name):
for node in NODES:
names = _names_on_node(node, "/api/v1/policies", _token(context))
assert name in names, f"{node} does not see policy '{name}' (sees {sorted(names)})"
@then('the policy "{name}" should be absent from every node')
def step_policy_absent(context, name):
for node in NODES:
names = _names_on_node(node, "/api/v1/policies", _token(context))
assert name not in names, f"{node} still sees deleted policy '{name}'"
@then("the trigger registry should be identical across nodes")
def step_triggers_identical(context):
seen = {}
for node in NODES:
status, body = _curl_on_node(node, "GET", "/api/v1/policies/triggers", token=_token(context))
assert status == 200, f"{node} GET /policies/triggers returned HTTP {status}"
items = json.loads(body)
# Trigger descriptors are dicts; normalise to a canonical, order-independent form.
seen[node] = sorted(json.dumps(t, sort_keys=True) for t in items)
a, b = (seen[n] for n in NODES)
assert a == b, f"trigger registry differs across nodes: {seen}"
# --- sources ---
@when('I create an S3 source named "{name}" on node "{idx}"')
def step_create_source_on_node(context, name, idx):
conn = _any_connection_id(context)
status, body = _curl_on_node(_node(idx), "POST", "/api/v1/sources", token=_token(context),
data=_s3_source_body(name, conn), content_type="application/json")
assert status in (200, 201), f"create source on {_node(idx)} failed: HTTP {status}: {body[:200]}"
@when('I delete the source "{name}" on node "{idx}"')
def step_delete_source_on_node(context, name, idx):
sid = _source_id_by_name(context, name)
assert sid, f"source '{name}' not found"
status, body = _curl_on_node(_node(idx), "DELETE", f"/api/v1/sources/{sid}", token=_token(context))
assert status in (200, 204), f"delete source on {_node(idx)} failed: HTTP {status}"
@then('the source "{name}" should be visible from every node')
def step_source_visible(context, name):
for node in NODES:
names = _names_on_node(node, "/api/v1/sources", _token(context))
assert name in names, f"{node} does not see source '{name}' (sees {sorted(names)})"
@then('the source "{name}" should be absent from every node')
def step_source_absent(context, name):
for node in NODES:
names = _names_on_node(node, "/api/v1/sources", _token(context))
assert name not in names, f"{node} still sees deleted source '{name}'"
@then('deleting the source "{name}" from node "{idx}" is rejected because it is referenced')
def step_source_delete_guarded(context, name, idx):
sid = _source_id_by_name(context, name)
assert sid, f"source '{name}' not found"
status, body = _curl_on_node(_node(idx), "DELETE", f"/api/v1/sources/{sid}", token=_token(context))
assert status == 409, (
f"expected 409 (source referenced by a policy created on another node), got HTTP {status}")
# --- connections (integration configs) ---
@when('I create an S3 connection named "{name}" via the load balancer')
def step_create_conn(context, name):
r = requests.post(f"{LB_URL}/api/v1/integrations",
headers={**_auth(context), "Content-Type": "application/json"},
data=_s3_connection_body(name), timeout=15)
assert r.status_code in (200, 201), f"create connection failed: HTTP {r.status_code}: {r.text[:200]}"
context._conn_id = r.json()["id"]
@then('the connection "{name}" should resolve from every node with its secret masked')
def step_conn_resolves(context, name):
cid = context._conn_id
for node in NODES:
status, body = _curl_on_node(node, "GET", f"/api/v1/integrations/{cid}", token=_token(context))
assert status == 200, f"{node} cannot resolve connection {cid}: HTTP {status}"
secret = json.loads(body).get("config", {}).get("secretAccessKey")
assert secret in (None, "", "********"), f"{node} leaked the connection secret on read"
@when('I delete the connection via the load balancer')
def step_delete_conn(context):
r = requests.delete(f"{LB_URL}/api/v1/integrations/{context._conn_id}",
headers=_auth(context), timeout=15)
assert r.status_code in (200, 204), f"delete connection failed: HTTP {r.status_code}"
@then("the connection should be gone from every node")
def step_conn_absent(context):
cid = context._conn_id
for node in NODES:
status, _ = _curl_on_node(node, "GET", f"/api/v1/integrations/{cid}", token=_token(context))
assert status in (403, 404), f"{node} still resolves deleted connection {cid}: HTTP {status}"