From 5ccb56da2d5b832f85b7e3b0ef1470c9057fa2ea Mon Sep 17 00:00:00 2001 From: James Brunton Date: Fri, 10 Jul 2026 13:19:41 +0100 Subject: [PATCH] Add S3 policy source (#6948) # Description of Changes * Adds an Amazon S3 Source & Output * Removes folder source from SaaS * Some miscellaneous UX fixes around pipelines --- .../common/model/ApplicationProperties.java | 8 + .../proprietary/cluster/s3/S3Clients.java | 26 +- .../LenientEncryptedStringConverter.java | 31 ++ .../policy/controller/PolicyController.java | 75 +++- .../policy/engine/PolicyRunner.java | 10 +- .../policy/engine/PolicySweep.java | 28 ++ .../policy/engine/SweepOutcome.java | 19 + .../policy/input/S3InputSource.java | 286 +++++++++++++++ .../policy/output/FolderOutputSink.java | 20 +- .../policy/output/OutputNames.java | 29 ++ .../policy/output/S3OutputSink.java | 270 +++++++++++++++ .../proprietary/policy/s3/S3Config.java | 103 ++++++ .../policy/s3/S3ConnectionPool.java | 110 ++++++ .../proprietary/policy/s3/S3Identities.java | 28 ++ .../policy/source/SourceController.java | 49 ++- .../policy/source/SourceEntity.java | 6 + .../policy/source/SourceOverviewService.java | 14 +- .../policy/store/PolicyEntity.java | 6 + .../policy/trigger/ScheduleTrigger.java | 14 +- .../proprietary/util/SecretMasker.java | 30 +- .../LenientEncryptedStringConverterTest.java | 45 +++ .../controller/PolicyControllerTest.java | 60 +++- .../policy/engine/PolicyRunnerTest.java | 38 ++ .../policy/input/S3InputSourceMinioTest.java | 231 +++++++++++++ .../policy/input/S3InputSourceTest.java | 327 ++++++++++++++++++ .../policy/output/S3OutputSinkMinioTest.java | 211 +++++++++++ .../policy/output/S3OutputSinkTest.java | 267 ++++++++++++++ .../policy/source/SourceControllerTest.java | 75 ++++ .../policy/trigger/ScheduleTriggerTest.java | 35 ++ .../proprietary/util/SecretMaskerTest.java | 62 +++- .../public/locales/en-US/translation.toml | 49 +++ .../components/pipelines/outputModes.test.ts | 9 + .../components/pipelines/outputModes.ts | 12 + .../sources/creatableSourceTypes.test.ts | 13 + .../sources/creatableSourceTypes.ts | 13 + frontend/editor/src/portal/api/pipelines.ts | 27 +- .../components/pipelines/outputModes.ts | 11 + .../components/sources/ConnectWizard.test.tsx | 47 +++ .../components/sources/ConnectWizard.tsx | 18 +- .../sources/creatableSourceTypes.ts | 14 + .../portal/components/sources/sourceTypes.ts | 71 +++- .../src/portal/mocks/handlers/pipelines.ts | 8 +- .../src/portal/views/PipelineBuilder.css | 24 ++ .../src/portal/views/PipelineBuilder.test.tsx | 115 +++++- .../src/portal/views/PipelineBuilder.tsx | 236 ++++++++++++- 45 files changed, 3091 insertions(+), 89 deletions(-) create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/integration/crypto/LenientEncryptedStringConverter.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/SweepOutcome.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/input/S3InputSource.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/output/OutputNames.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/output/S3OutputSink.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3Config.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3ConnectionPool.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3Identities.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/integration/crypto/LenientEncryptedStringConverterTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/input/S3InputSourceMinioTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/input/S3InputSourceTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/output/S3OutputSinkMinioTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/output/S3OutputSinkTest.java create mode 100644 frontend/editor/src/portal-saas/components/pipelines/outputModes.test.ts create mode 100644 frontend/editor/src/portal-saas/components/pipelines/outputModes.ts create mode 100644 frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.test.ts create mode 100644 frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.ts create mode 100644 frontend/editor/src/portal/components/pipelines/outputModes.ts create mode 100644 frontend/editor/src/portal/components/sources/creatableSourceTypes.ts diff --git a/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java b/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java index 9ed2610f51..d2f0472510 100644 --- a/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java +++ b/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java @@ -246,6 +246,14 @@ public class ApplicationProperties { * and paused runs are kept regardless of age. */ private int runExpiryMinutes = 30; + + /** + * Whether a policy S3 source's custom endpoint may resolve to a loopback, link-local, or + * private address. Off by default so a user-supplied endpoint cannot be pointed at internal + * services (e.g. the cloud metadata address); enable for a self-hosted MinIO or other + * in-network object store. + */ + private boolean allowPrivateS3Endpoints = false; } @Data diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/cluster/s3/S3Clients.java b/app/proprietary/src/main/java/stirling/software/proprietary/cluster/s3/S3Clients.java index eacdaf32c0..d3aab54c76 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/cluster/s3/S3Clients.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/cluster/s3/S3Clients.java @@ -133,29 +133,45 @@ public final class S3Clients { * storage.s3.allow-private-endpoints=true}. */ static void validateEndpointHost(URI endpoint, boolean allowPrivate) { + validateEndpointHost( + endpoint, + allowPrivate, + "storage.s3.endpoint", + "set storage.s3.allow-private-endpoints=true to opt in" + + " (e.g. for MinIO or in-cluster S3)."); + } + + /** + * The same private-address guard for S3 endpoints configured outside the {@code storage.s3.*} + * block (e.g. per-source policy config), with the setting named in messages supplied by the + * caller. + */ + public static void validateEndpointHost( + URI endpoint, boolean allowPrivate, String settingName, String optInHint) { if (allowPrivate) { return; } String host = endpoint.getHost(); if (host == null || host.isBlank()) { - throw new IllegalStateException("storage.s3.endpoint must include a host: " + endpoint); + throw new IllegalStateException(settingName + " must include a host: " + endpoint); } InetAddress[] addresses; try { addresses = InetAddress.getAllByName(host); } catch (UnknownHostException e) { throw new IllegalStateException( - "Unable to resolve storage.s3.endpoint host '" + host + "'", e); + "Unable to resolve " + settingName + " host '" + host + "'", e); } for (InetAddress address : addresses) { if (isPrivateOrLocal(address)) { throw new IllegalStateException( - "storage.s3.endpoint host '" + settingName + + " host '" + host + "' resolves to private/link-local address " + address.getHostAddress() - + "; set storage.s3.allow-private-endpoints=true to opt in" - + " (e.g. for MinIO or in-cluster S3)."); + + "; " + + optInHint); } } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/integration/crypto/LenientEncryptedStringConverter.java b/app/proprietary/src/main/java/stirling/software/proprietary/integration/crypto/LenientEncryptedStringConverter.java new file mode 100644 index 0000000000..7bb6190220 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/integration/crypto/LenientEncryptedStringConverter.java @@ -0,0 +1,31 @@ +package stirling.software.proprietary.integration.crypto; + +import jakarta.persistence.AttributeConverter; +import jakarta.persistence.Converter; + +/** + * {@link EncryptedStringConverter} for columns that held plaintext before encryption shipped: + * writes are always encrypted, but a stored value that is not valid ciphertext is returned as-is, + * so pre-encryption rows keep loading and become encrypted on their next save. The discrimination + * is exact for JSON payloads, which can never be mistaken for ciphertext ('{' is not in the Base64 + * alphabet). The trade-off is that a genuinely corrupted ciphertext surfaces as garbage to the + * caller's parser instead of failing here. + */ +@Converter +public class LenientEncryptedStringConverter implements AttributeConverter { + + @Override + public String convertToDatabaseColumn(String attribute) { + return CredentialEncryption.encrypt(attribute); + } + + @Override + public String convertToEntityAttribute(String dbData) { + try { + return CredentialEncryption.decrypt(dbData); + } catch (IllegalArgumentException | IllegalStateException e) { + // Not ciphertext: legacy plaintext from before encryption shipped. + return dbData; + } + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java index 9494317751..045698c015 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java @@ -48,7 +48,9 @@ import stirling.software.proprietary.policy.engine.PolicyRunHandle; import stirling.software.proprietary.policy.engine.PolicyRunRegistry; import stirling.software.proprietary.policy.engine.PolicyRunner; import stirling.software.proprietary.policy.engine.PolicyValidator; +import stirling.software.proprietary.policy.engine.SweepOutcome; import stirling.software.proprietary.policy.ledger.ProcessedLedger; +import stirling.software.proprietary.policy.model.OutputSpec; import stirling.software.proprietary.policy.model.PipelineDefinition; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.model.PolicyInputs; @@ -64,6 +66,7 @@ import stirling.software.proprietary.policy.store.PolicyStore; import stirling.software.proprietary.policy.trigger.PolicyTrigger; import stirling.software.proprietary.policy.trigger.PolicyTriggerManager; import stirling.software.proprietary.policy.trigger.TriggerInfo; +import stirling.software.proprietary.util.SecretMasker; /** * Policy CRUD plus pipeline runs (stored or ad-hoc). Runs are async: returns a run id, poll {@code @@ -202,7 +205,7 @@ public class PolicyController { + " assigned; returns the stored policy with its id.") public ResponseEntity savePolicy(@RequestBody Policy policy) { requirePolicyEditingAllowed(); - Policy owned = resolveOwnership(policy); + Policy owned = withStoredOutputSecrets(resolveOwnership(policy)); requireAccessibleSources(owned); try { policyValidator.validate(owned); @@ -213,7 +216,7 @@ public class PolicyController { // Re-sync trigger registrations now so a new/changed folder-watch policy starts being // watched immediately instead of after the next reconcile sweep. policyTriggerManager.notifyPoliciesChanged(); - return ResponseEntity.ok(saved); + return ResponseEntity.ok(withMaskedOutputSecrets(saved)); } @PutMapping("/order") @@ -282,6 +285,50 @@ public class PolicyController { teamId); } + /** Output secrets never leave the server: reads return the redaction sentinel instead. */ + private static Policy withMaskedOutputSecrets(Policy policy) { + return withOutput( + policy, + new OutputSpec( + policy.output().type(), SecretMasker.mask(policy.output().options()))); + } + + /** + * An edit that round-trips a masked read sends output secrets back as the sentinel; restore + * them from the stored policy so saving without re-typing keeps them (validation then runs + * against the real values). + */ + private Policy withStoredOutputSecrets(Policy incoming) { + if (incoming.id() == null || incoming.id().isBlank()) { + return incoming; + } + return policyStore + .get(incoming.id()) + .map( + existing -> + withOutput( + incoming, + new OutputSpec( + incoming.output().type(), + SecretMasker.restoreRedacted( + incoming.output().options(), + existing.output().options())))) + .orElse(incoming); + } + + private static Policy withOutput(Policy policy, OutputSpec output) { + return new Policy( + policy.id(), + policy.name(), + policy.owner(), + policy.enabled(), + policy.trigger(), + policy.sourceIds(), + policy.steps(), + output, + policy.teamId()); + } + /** * Creating, editing, pausing/resuming, and deleting policies requires the editor role for the * caller's team — a team leader on SaaS (see {@link PolicyManagementAuthority}); the global @@ -306,9 +353,14 @@ public class PolicyController { @GetMapping @Operation( summary = "List policies", - description = "Lists the policies belonging to the caller's team.") + description = + "Lists the policies belonging to the caller's team. Secret-bearing output" + + " options are returned as a redaction sentinel, never their stored" + + " values.") public List listPolicies() { - return policyAccessGuard.visibleFrom(policyStore); + return policyAccessGuard.visibleFrom(policyStore).stream() + .map(PolicyController::withMaskedOutputSecrets) + .toList(); } @GetMapping("/overview") @@ -337,11 +389,17 @@ public class PolicyController { } @GetMapping("/{policyId}") - @Operation(summary = "Get a policy by id") + @Operation( + summary = "Get a policy by id", + description = + "Secret-bearing output options are returned as a redaction sentinel, never" + + " their stored values; an edit that sends the sentinel back keeps" + + " them.") public ResponseEntity getPolicy(@PathVariable String policyId) { return policyStore .get(policyId) .filter(policyAccessGuard::canAccess) + .map(PolicyController::withMaskedOutputSecrets) .map(ResponseEntity::ok) .orElseGet(() -> ResponseEntity.notFound().build()); } @@ -412,9 +470,10 @@ public class PolicyController { description = "Pulls the policy's configured sources and runs the pipeline now, regardless of" + " the enabled flag (which only gates automatic triggering). Returns" - + " the ids of the runs started; poll the run-status endpoint for each." - + " Empty when the sources yielded no work to do.") - public ResponseEntity> trigger(@PathVariable String policyId) { + + " the ids of the runs started (poll the run-status endpoint for each)" + + " plus what the sweep skipped - already-processed, parked-by-failure," + + " and in-flight counts - so an empty result explains itself.") + public ResponseEntity trigger(@PathVariable String policyId) { Policy policy = policyStore .get(policyId) diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java index 5a35b879c0..7731d936bd 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java @@ -44,7 +44,7 @@ public class PolicyRunner { private final ProcessedLedger processedLedger; /** Full-listing sweep: resolve every source, then reconcile the ledger. */ - public List run(Policy policy) { + public SweepOutcome run(Policy policy) { return run(policy, SweepKind.FULL); } @@ -52,10 +52,10 @@ public class PolicyRunner { * Trigger entry point. Pulls every referenced source; each yielded unit becomes its own run so * one failure does not affect the others. No sources means one run with no input (generator * pipeline). Missing or disabled sources are skipped so one broken reference does not stop the - * rest. Returns the ids of the runs it started (empty when sources yielded no work), so a - * manual trigger can report back which runs to follow. + * rest. Returns the ids of the runs it started plus what the sweep skipped, so a manual trigger + * can report which runs to follow or why nothing ran. */ - public List run(Policy policy, SweepKind sweep) { + public SweepOutcome run(Policy policy, SweepKind sweep) { long sweepStart = System.currentTimeMillis(); PolicySweep context = new PolicySweep(policy.id(), sweep, processedLedger); List runIds = new ArrayList<>(); @@ -95,7 +95,7 @@ public class PolicyRunner { policy.id()); } } - return runIds; + return context.outcome(runIds); } /** Run a stored policy on caller-supplied files (e.g. manual upload), bypassing its sources. */ diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicySweep.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicySweep.java index 5302bacdb3..469b5f769e 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicySweep.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicySweep.java @@ -86,4 +86,32 @@ final class PolicySweep implements ResolveContext { synchronized Set presentIdentities() { return Set.copyOf(present); } + + /** + * Summarise the sweep from state already in hand (no extra ledger reads): the prefetched rows + * were loaded before claiming, and successful claims flipped their entries to PROCESSING, so + * what remains DONE or ERROR is exactly what this sweep skipped. + */ + synchronized SweepOutcome outcome(List runIds) { + int alreadyProcessed = 0; + int parked = 0; + int processing = 0; + for (String identity : present) { + ClaimState state = prefetched.get(identity); + if (state == null) { + continue; + } + switch (state.status()) { + case DONE -> alreadyProcessed++; + case ERROR -> parked++; + case PROCESSING, INTERRUPTED -> processing++; + } + } + return new SweepOutcome( + runIds, + present.size(), + alreadyProcessed, + parked, + Math.max(0, processing - runIds.size())); + } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/SweepOutcome.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/SweepOutcome.java new file mode 100644 index 0000000000..752299976d --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/SweepOutcome.java @@ -0,0 +1,19 @@ +package stirling.software.proprietary.policy.engine; + +import java.util.List; + +/** + * What one policy sweep found and started, so a manual trigger can explain an empty result instead + * of a blanket "nothing to do": how many files the sources listed, how many were skipped because + * they are already processed at their current version, how many are parked by a failed run (not + * retried until they change or history is cleared), and how many are still in flight from an + * earlier sweep. Counts are zero for {@link SweepKind#LIGHT} sweeps, which do not take a full + * listing. + */ +public record SweepOutcome( + List runIds, int filesListed, int alreadyProcessed, int parked, int inFlight) { + + public SweepOutcome { + runIds = runIds == null ? List.of() : List.copyOf(runIds); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/S3InputSource.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/S3InputSource.java new file mode 100644 index 0000000000..99e189f326 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/S3InputSource.java @@ -0,0 +1,286 @@ +package stirling.software.proprietary.policy.input; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.core.io.AbstractResource; +import org.springframework.core.io.Resource; +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.PolicyInputs; +import stirling.software.proprietary.policy.s3.S3Config; +import stirling.software.proprietary.policy.s3.S3ConnectionPool; +import stirling.software.proprietary.policy.s3.S3Identities; + +import software.amazon.awssdk.core.exception.SdkException; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.DeleteObjectRequest; +import software.amazon.awssdk.services.s3.model.GetObjectRequest; +import software.amazon.awssdk.services.s3.model.HeadObjectRequest; +import software.amazon.awssdk.services.s3.model.HeadObjectResponse; +import software.amazon.awssdk.services.s3.model.ListObjectsV2Request; +import software.amazon.awssdk.services.s3.model.ListObjectsV2Response; +import software.amazon.awssdk.services.s3.model.NoSuchKeyException; +import software.amazon.awssdk.services.s3.model.S3Exception; +import software.amazon.awssdk.services.s3.model.S3Object; + +/** + * Reads input files from an Amazon S3 (or S3-compatible) bucket; each listed object is its own unit + * of work, claimed through the {@link ResolveContext} ledger and tracked in place. Identity and + * version gate come from {@link S3Identities}, so the steady-state sweep never downloads content. + * Options (see {@link S3Config}): "bucket" (required), "region" (default us-east-1), "prefix" (only + * keys starting with it are read), "endpoint" (S3-compatible stores such as MinIO; path-style + * addressing is used automatically), "accessKeyId" and "secretAccessKey" (required; requests are + * never signed with the server's own AWS identity), and "mode" which is "consume" (default: a + * processed object is deleted once every policy that claimed it has settled successfully and it is + * still the version that ran; failures stay in place and are not retried until they change) or + * "snapshot" (stateless, every run sees the full set). Keys ending in "/" (folder placeholders) and + * keys with a dot-prefixed path segment are never picked up, mirroring the folder source's + * hidden-file rule. + */ +@Slf4j +@Service +@RequiredArgsConstructor +@ConditionalOnBooleanProperty(name = "policies.enabled") +public class S3InputSource implements InputSource { + + private static final String TYPE = "s3"; + + private final S3ConnectionPool connectionPool; + + @Override + public String type() { + return TYPE; + } + + @Override + public boolean supports(InputSpec spec) { + return spec != null && TYPE.equals(spec.type()); + } + + /** + * Fails fast at save time: bad config shape, a private endpoint without the operator opt-in, or + * a bucket the supplied credentials cannot list. + */ + @Override + public void validate(InputSpec spec) { + S3Config config = S3Config.from(spec.options()); + try { + connectionPool.clientFor(config).listObjectsV2(listRequest(config).maxKeys(1).build()); + } catch (SdkException e) { + throw new IllegalArgumentException( + "cannot access s3://" + + config.bucket() + + "/" + + config.prefix() + + ": " + + e.getMessage(), + e); + } + } + + @Override + public List resolve(InputSpec spec, ResolveContext ctx) throws IOException { + S3Config config = S3Config.from(spec.options()); + S3Client client = connectionPool.clientFor(config); + // A listing failure propagates so the sweep reads it as "could not list" (which vetoes + // presence cleanup), never as "verifiably no objects". + List objects = listObjects(client, config); + + if (config.snapshot()) { + return objects.stream() + .map( + object -> + ResolvedInput.of( + PolicyInputs.of( + List.of( + objectResource( + client, config, object))))) + .toList(); + } + + ctx.reportPresent( + objects.stream() + .map(object -> S3Identities.identity(config.bucket(), object.key())) + .toList()); + + List work = new ArrayList<>(); + for (S3Object object : objects) { + String identity = S3Identities.identity(config.bucket(), object.key()); + String gate = S3Identities.gate(object.eTag(), object.size(), object.lastModified()); + if (!ctx.claim(identity, gate, null)) { + continue; + } + work.add( + new ResolvedInput( + PolicyInputs.of(List.of(objectResource(client, config, object))), + success -> + completeConsumed( + ctx, + client, + config, + object.key(), + identity, + gate, + success))); + } + return work; + } + + /** + * Settle at the version this run claimed, then remove the object only when it still carries + * that version and every policy that claimed it has settled DONE, mirroring the folder source's + * consensus delete. A failed run settles ERROR and never deletes; the DONE row of an object + * that could not be deleted still stops reprocessing. + */ + private void completeConsumed( + ResolveContext ctx, + S3Client client, + S3Config config, + String key, + String identity, + String claimGate, + boolean success) { + ctx.settle(identity, claimGate, null, success); + if (!success) { + return; + } + try { + HeadObjectResponse head = + client.headObject( + HeadObjectRequest.builder().bucket(config.bucket()).key(key).build()); + String currentGate = + S3Identities.gate(head.eTag(), head.contentLength(), head.lastModified()); + if (currentGate.equals(claimGate) && ctx.allSettledDone(identity)) { + client.deleteObject( + DeleteObjectRequest.builder().bucket(config.bucket()).key(key).build()); + } + } catch (NoSuchKeyException alreadyGone) { + // Removed by the user or a co-watching policy's own consensus delete: nothing to do. + } catch (S3Exception e) { + if (e.statusCode() == 404) { + return; + } + log.warn("Could not remove consumed S3 object {}: {}", identity, e.getMessage()); + } catch (SdkException e) { + log.warn("Could not remove consumed S3 object {}: {}", identity, e.getMessage()); + } + } + + /** Every ingestible object under the configured prefix, across all listing pages. */ + private static List listObjects(S3Client client, S3Config config) { + List objects = new ArrayList<>(); + String continuationToken = null; + do { + ListObjectsV2Request.Builder request = listRequest(config); + if (continuationToken != null) { + request.continuationToken(continuationToken); + } + ListObjectsV2Response page = client.listObjectsV2(request.build()); + for (S3Object object : page.contents()) { + if (ingestible(object)) { + objects.add(object); + } + } + continuationToken = page.nextContinuationToken(); + } while (continuationToken != null); + return objects; + } + + private static ListObjectsV2Request.Builder listRequest(S3Config config) { + ListObjectsV2Request.Builder request = + ListObjectsV2Request.builder().bucket(config.bucket()); + if (!config.prefix().isEmpty()) { + request.prefix(config.prefix()); + } + return request; + } + + /** + * Folder-placeholder keys (ending "/") and keys with a dot-prefixed segment are skipped, so a + * hidden convention (e.g. a future output sink's staging prefix) is never re-ingested. + */ + private static boolean ingestible(S3Object object) { + String key = object.key(); + if (key.isEmpty() || key.endsWith("/")) { + return false; + } + for (String segment : key.split("/")) { + if (segment.startsWith(".")) { + return false; + } + } + return true; + } + + private static Resource objectResource(S3Client client, S3Config config, S3Object object) { + return new S3ObjectResource(client, config.bucket(), object); + } + + /** + * Streams the object on demand, pinned to the ETag observed at listing time so a run never + * reads a different version than the sweep claimed (a swapped object fails the read with a + * precondition error and the new version is claimed by a later sweep). + */ + private static final class S3ObjectResource extends AbstractResource { + + private final S3Client client; + private final String bucket; + private final String key; + private final String eTag; + private final Long size; + + private S3ObjectResource(S3Client client, String bucket, S3Object object) { + this.client = client; + this.bucket = bucket; + this.key = object.key(); + this.eTag = object.eTag(); + this.size = object.size(); + } + + @Override + public InputStream getInputStream() throws IOException { + GetObjectRequest.Builder request = GetObjectRequest.builder().bucket(bucket).key(key); + if (eTag != null && !eTag.isBlank()) { + request.ifMatch(eTag); + } + try { + return client.getObject(request.build()); + } catch (NoSuchKeyException e) { + throw new FileNotFoundException(getDescription() + " no longer exists"); + } catch (SdkException e) { + throw new IOException( + "Could not read " + getDescription() + ": " + e.getMessage(), e); + } + } + + /** Listed just now; readers get a precise error from {@link #getInputStream} instead. */ + @Override + public boolean exists() { + return true; + } + + @Override + public long contentLength() { + return size == null ? -1 : size; + } + + @Override + public String getFilename() { + return key.substring(key.lastIndexOf('/') + 1); + } + + @Override + public String getDescription() { + return "S3 object " + S3Identities.identity(bucket, key); + } + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java index ed3cf707ad..52da21208a 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java @@ -15,7 +15,6 @@ import java.util.List; import java.util.UUID; import java.util.stream.Stream; -import org.apache.commons.io.FilenameUtils; import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; import org.springframework.core.io.Resource; import org.springframework.http.MediaType; @@ -82,7 +81,7 @@ public class FolderOutputSink implements PolicyOutputSink { List results = new ArrayList<>(); for (int i = 0; i < outputs.size(); i++) { Resource resource = outputs.get(i); - String name = safeName(resource.getFilename(), i); + String name = OutputNames.safeName(resource.getFilename(), i); Path staged = tmpDir.resolve(UUID.randomUUID().toString()); String contentHash = stage(resource, staged, delivery.policyId() != null); long size = Files.size(staged); @@ -198,29 +197,14 @@ public class FolderOutputSink implements PolicyOutputSink { return Path.of(directory.toString()); } - // Strip any directory component / "../" so a crafted output name cannot escape targetDir. - private static String safeName(String filename, int index) { - if (filename == null || filename.isBlank()) { - return "output-" + index; - } - String name = FilenameUtils.getName(filename); - if (name.isBlank() || ".".equals(name) || "..".equals(name)) { - return "output-" + index; - } - return name; - } - // Non-colliding path, appending " (n)" before the extension. private static Path uniqueTarget(Path dir, String filename) { Path candidate = dir.resolve(filename); if (!Files.exists(candidate)) { return candidate; } - String base = FilenameUtils.getBaseName(filename); - String ext = FilenameUtils.getExtension(filename); - String suffix = ext.isEmpty() ? "" : "." + ext; for (int n = 1; ; n++) { - Path next = dir.resolve(base + " (" + n + ")" + suffix); + Path next = dir.resolve(OutputNames.numbered(filename, n)); if (!Files.exists(next)) { return next; } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/OutputNames.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/OutputNames.java new file mode 100644 index 0000000000..e82cc66b28 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/OutputNames.java @@ -0,0 +1,29 @@ +package stirling.software.proprietary.policy.output; + +import org.apache.commons.io.FilenameUtils; + +/** Output file naming shared by the sinks: sanitised base names and collision suffixes. */ +final class OutputNames { + + private OutputNames() {} + + /** Strip any directory component / "../" so a crafted output name cannot escape the target. */ + static String safeName(String filename, int index) { + if (filename == null || filename.isBlank()) { + return "output-" + index; + } + String name = FilenameUtils.getName(filename); + if (name.isBlank() || ".".equals(name) || "..".equals(name)) { + return "output-" + index; + } + return name; + } + + /** The nth alternative for a taken name, appending " (n)" before the extension. */ + static String numbered(String filename, int n) { + String base = FilenameUtils.getBaseName(filename); + String ext = FilenameUtils.getExtension(filename); + String suffix = ext.isEmpty() ? "" : "." + ext; + return base + " (" + n + ")" + suffix; + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/S3OutputSink.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/S3OutputSink.java new file mode 100644 index 0000000000..c7d740868a --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/S3OutputSink.java @@ -0,0 +1,270 @@ +package stirling.software.proprietary.policy.output; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.security.DigestOutputStream; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.ArrayList; +import java.util.HexFormat; +import java.util.List; +import java.util.UUID; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.core.io.Resource; +import org.springframework.http.MediaType; +import org.springframework.http.MediaTypeFactory; +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import stirling.software.common.model.job.ResultFile; +import stirling.software.proprietary.policy.ledger.ProcessedLedger; +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.s3.S3Config; +import stirling.software.proprietary.policy.s3.S3ConnectionPool; +import stirling.software.proprietary.policy.s3.S3Identities; + +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.HeadObjectRequest; +import software.amazon.awssdk.services.s3.model.NoSuchKeyException; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; +import software.amazon.awssdk.services.s3.model.PutObjectResponse; +import software.amazon.awssdk.services.s3.model.S3Exception; + +/** + * Uploads a run's outputs to the bucket and key prefix given in the {@link OutputSpec} (same + * connection options as the S3 input source; "prefix" is the destination folder). The + * record-before-visible obligation is met without a rename step: a single-part PUT's ETag is the + * MD5 of its content on plain and SSE-S3 buckets, so the ledger row is recorded at that predicted + * gate BEFORE the upload, and the object is claimed under exactly the gate the next listing + * returns. Stores where the returned ETag differs (e.g. SSE-KMS) are re-recorded at the actual gate + * immediately after the PUT - a narrow race those buckets accept rather than a broken loop. Names + * never overwrite: uploads are conditional on the key not existing ({@code If-None-Match: *}), + * re-picking "name (n).ext" on collision exactly like the folder sink; stores without + * conditional-write support fall back to an existence check per candidate. + */ +@Slf4j +@Service +@RequiredArgsConstructor +@ConditionalOnBooleanProperty(name = "policies.enabled") +public class S3OutputSink implements PolicyOutputSink { + + private static final String TYPE = "s3"; + + private final S3ConnectionPool connectionPool; + private final ProcessedLedger processedLedger; + + @Override + public String type() { + return TYPE; + } + + @Override + public boolean supports(OutputSpec spec) { + return spec != null && TYPE.equals(spec.type()); + } + + /** + * Config shape and endpoint guard only - no network probe, since write-only credentials + * (s3:PutObject without s3:ListBucket) are a legitimate setup for an output bucket and a + * listing probe would wrongly reject them. + */ + @Override + public void validate(OutputSpec spec) { + connectionPool.clientFor(S3Config.from(spec.options())); + } + + @Override + public List deliver( + OutputDelivery delivery, List outputs, OutputSpec spec) throws IOException { + S3Config config = S3Config.from(spec.options()); + S3Client client = connectionPool.clientFor(config); + + List results = new ArrayList<>(); + for (int i = 0; i < outputs.size(); i++) { + Resource resource = outputs.get(i); + String name = OutputNames.safeName(resource.getFilename(), i); + Path staged = Files.createTempFile("s3-output-", ".tmp"); + try { + String predictedGate = stage(resource, staged, delivery.policyId() != null); + long size = Files.size(staged); + String key = upload(delivery, client, config, name, staged, predictedGate); + String contentType = + MediaTypeFactory.getMediaType(name) + .orElse(MediaType.APPLICATION_OCTET_STREAM) + .toString(); + results.add( + ResultFile.builder() + .fileId(UUID.randomUUID().toString()) + .fileName(S3Identities.identity(config.bucket(), key)) + .contentType(contentType) + .fileSize(size) + .build()); + log.debug( + "Wrote policy run {} output to {}", + delivery.runId(), + S3Identities.identity(config.bucket(), key)); + } finally { + try { + Files.deleteIfExists(staged); + } catch (IOException e) { + log.warn("Could not remove S3 staging file {}: {}", staged, e.getMessage()); + } + } + } + return results; + } + + /** + * Spool the output to a local staging file (S3 needs a known content length, and the body must + * be re-readable across collision retries). For a recorded delivery the MD5 - the predicted + * single-part ETag - is digested in the same pass; ad-hoc runs record nothing and skip it. + */ + private static String stage(Resource resource, Path staged, boolean recorded) + throws IOException { + if (!recorded) { + try (InputStream is = resource.getInputStream(); + OutputStream out = Files.newOutputStream(staged)) { + is.transferTo(out); + } + return null; + } + MessageDigest digest = newMd5(); + try (InputStream is = resource.getInputStream(); + DigestOutputStream out = + new DigestOutputStream(Files.newOutputStream(staged), digest)) { + is.transferTo(out); + } + return HexFormat.of().formatHex(digest.digest()); + } + + /** + * The S3 shape of the folder sink's record-then-rename loop. The ledger row must exist before + * the object is visible, so it is recorded at the predicted gate before the PUT; losing the + * chosen key to a concurrent writer (the conditional PUT fails) forgets the just-recorded row - + * whatever object actually owns that key must stay claimable at any version - then re-picks. A + * PUT that never made the object visible also forgets its row. + */ + private String upload( + OutputDelivery delivery, + S3Client client, + S3Config config, + String name, + Path staged, + String predictedGate) + throws IOException { + String keyPrefix = keyPrefix(config); + boolean conditionalPuts = true; + for (int attempt = 0; ; attempt++) { + String key = keyPrefix + (attempt == 0 ? name : OutputNames.numbered(name, attempt)); + String identity = S3Identities.identity(config.bucket(), key); + if (!conditionalPuts && exists(client, config.bucket(), key)) { + continue; + } + if (delivery.policyId() != null) { + processedLedger.recordOutput(delivery.policyId(), identity, predictedGate, null); + } + PutObjectRequest.Builder put = + PutObjectRequest.builder().bucket(config.bucket()).key(key); + if (conditionalPuts) { + put.ifNoneMatch("*"); + } + try { + PutObjectResponse response = + client.putObject(put.build(), RequestBody.fromFile(staged)); + reRecordIfGateDiffers(delivery, identity, predictedGate, response); + return key; + } catch (S3Exception e) { + forgetRecorded(delivery, identity, predictedGate); + if (conditionalPuts && e.statusCode() == 412) { + // Known edge: if our own PUT succeeded server-side but the response was lost + // and the SDK retried, that retry 412s here too - we then upload under the + // next name, leaving the first object row-less (claimable, single duplicate). + // Requires a response-lost network flake at exactly this moment; accepted. + log.debug("Output key {} taken concurrently; re-picking", identity); + continue; + } + if (conditionalPuts && e.statusCode() == 501) { + // Store without conditional-write support: retry this candidate with a plain + // existence check instead. + log.debug( + "Conditional PUT unsupported by {}; falling back to existence checks", + config.bucket()); + conditionalPuts = false; + attempt--; + continue; + } + throw new IOException("Could not upload " + identity + ": " + e.getMessage(), e); + } catch (SdkException e) { + forgetRecorded(delivery, identity, predictedGate); + throw new IOException("Could not upload " + identity + ": " + e.getMessage(), e); + } + } + } + + /** + * On buckets where a PUT's ETag is not the content MD5 (e.g. SSE-KMS), re-record at the gate + * listings will actually return. The row is briefly at the wrong gate while the object is + * already visible - the narrow race such stores trade for a working self-output skip. + */ + private void reRecordIfGateDiffers( + OutputDelivery delivery, + String identity, + String predictedGate, + PutObjectResponse response) { + if (delivery.policyId() == null) { + return; + } + String actualGate = S3Identities.gate(response.eTag(), null, null); + if (!actualGate.equals(predictedGate)) { + log.debug( + "PUT ETag for {} differs from content MD5 (encrypted bucket?); re-recording", + identity); + processedLedger.recordOutput(delivery.policyId(), identity, actualGate, null); + } + } + + private void forgetRecorded(OutputDelivery delivery, String identity, String predictedGate) { + if (delivery.policyId() != null) { + processedLedger.forgetOutput(delivery.policyId(), identity, predictedGate); + } + } + + private static boolean exists(S3Client client, String bucket, String key) { + try { + client.headObject(HeadObjectRequest.builder().bucket(bucket).key(key).build()); + return true; + } catch (NoSuchKeyException e) { + return false; + } catch (S3Exception e) { + if (e.statusCode() == 404) { + return false; + } + throw e; + } + } + + /** The configured prefix as a key-path prefix: "processed" and "processed/" mean the same. */ + private static String keyPrefix(S3Config config) { + String prefix = config.prefix(); + if (prefix.isEmpty() || prefix.endsWith("/")) { + return prefix; + } + return prefix + "/"; + } + + private static MessageDigest newMd5() { + try { + return MessageDigest.getInstance("MD5"); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("MD5 unavailable", e); + } + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3Config.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3Config.java new file mode 100644 index 0000000000..152a6de4cf --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3Config.java @@ -0,0 +1,103 @@ +package stirling.software.proprietary.policy.s3; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.Map; + +/** + * Connection settings shared by the S3 input source and output sink, parsed from a spec's options + * map. Credentials are required: there is deliberately no fallback to the server's own AWS + * credential chain, so user-supplied config can never borrow the host's identity. {@code snapshot} + * is input-only and ignored by the sink. + */ +public record S3Config( + String bucket, + String region, + String prefix, + String endpoint, + String accessKeyId, + String secretAccessKey, + boolean snapshot) { + + private static final String BUCKET_OPTION = "bucket"; + private static final String REGION_OPTION = "region"; + private static final String PREFIX_OPTION = "prefix"; + private static final String ENDPOINT_OPTION = "endpoint"; + private static final String ACCESS_KEY_ID_OPTION = "accessKeyId"; + private static final String SECRET_ACCESS_KEY_OPTION = "secretAccessKey"; + private static final String MODE_OPTION = "mode"; + private static final String MODE_CONSUME = "consume"; + private static final String MODE_SNAPSHOT = "snapshot"; + + public static S3Config from(Map options) { + String bucket = trimmed(options.get(BUCKET_OPTION)); + if (bucket == null) { + throw new IllegalArgumentException("s3 config requires a 'bucket' option"); + } + String region = trimmed(options.get(REGION_OPTION)); + String prefix = trimmed(options.get(PREFIX_OPTION)); + if (prefix != null && prefix.startsWith("/")) { + prefix = prefix.substring(1); + } + String endpoint = validEndpoint(trimmed(options.get(ENDPOINT_OPTION))); + String accessKeyId = trimmed(options.get(ACCESS_KEY_ID_OPTION)); + String secretAccessKey = trimmed(options.get(SECRET_ACCESS_KEY_OPTION)); + if (accessKeyId == null || secretAccessKey == null) { + throw new IllegalArgumentException( + "s3 config requires an 'accessKeyId' and 'secretAccessKey'"); + } + String mode = trimmed(options.get(MODE_OPTION)); + if (mode != null && !MODE_CONSUME.equals(mode) && !MODE_SNAPSHOT.equals(mode)) { + throw new IllegalArgumentException("s3 config 'mode' must be 'consume' or 'snapshot'"); + } + return new S3Config( + bucket, + region == null ? "us-east-1" : region, + prefix == null ? "" : prefix, + endpoint, + accessKeyId, + secretAccessKey, + MODE_SNAPSHOT.equals(mode)); + } + + private static String validEndpoint(String endpoint) { + if (endpoint == null) { + return null; + } + URI uri; + try { + uri = new URI(endpoint); + } catch (URISyntaxException e) { + throw new IllegalArgumentException("s3 config 'endpoint' is not a valid URL", e); + } + if (!"http".equals(uri.getScheme()) && !"https".equals(uri.getScheme())) { + throw new IllegalArgumentException( + "s3 config 'endpoint' must be an http(s) URL, e.g. https://s3.example.com"); + } + return endpoint; + } + + private static String trimmed(Object value) { + if (value == null) { + return null; + } + String text = value.toString().trim(); + return text.isEmpty() ? null : text; + } + + /** Never prints the credentials, so an accidental log line cannot leak them. */ + @Override + public String toString() { + return "S3Config[bucket=" + + bucket + + ", region=" + + region + + ", prefix=" + + prefix + + ", endpoint=" + + endpoint + + ", snapshot=" + + snapshot + + "]"; + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3ConnectionPool.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3ConnectionPool.java new file mode 100644 index 0000000000..4118557dbd --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3ConnectionPool.java @@ -0,0 +1,110 @@ +package stirling.software.proprietary.policy.s3; + +import java.net.URI; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Function; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.stereotype.Service; + +import jakarta.annotation.PreDestroy; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.cluster.s3.S3Clients; + +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.S3ClientBuilder; +import software.amazon.awssdk.services.s3.S3Configuration; + +/** + * Long-lived {@link S3Client}s for policy S3 sources and sinks, one per distinct {@link S3Config}, + * closed at shutdown. An edited spec simply maps to a new entry, and a stale entry costs nothing + * (the URL-connection HTTP client holds no pooled sockets or threads). Clients sign exclusively + * with the spec's own credentials - there is deliberately no fallback to the server's AWS + * credential chain, so user-supplied config can never borrow the host's identity. Endpoints are + * guarded against private addresses before a client is ever built, since they come from portal + * users rather than the operator. + */ +@Service +@ConditionalOnBooleanProperty(name = "policies.enabled") +public class S3ConnectionPool { + + private final ApplicationProperties applicationProperties; + private final Function clientFactory; + private final Map clients = new ConcurrentHashMap<>(); + + @Autowired + public S3ConnectionPool(ApplicationProperties applicationProperties) { + this(applicationProperties, S3ConnectionPool::buildClient); + } + + /** Factory-injecting constructor for tests. */ + public S3ConnectionPool( + ApplicationProperties applicationProperties, + Function clientFactory) { + this.applicationProperties = applicationProperties; + this.clientFactory = clientFactory; + } + + public S3Client clientFor(S3Config config) { + return clients.computeIfAbsent( + config, + c -> { + requirePermittedEndpoint(c); + return clientFactory.apply(c); + }); + } + + /** + * A user-supplied endpoint must not reach loopback, link-local, or private addresses unless the + * operator has opted in via {@code policies.allowPrivateS3Endpoints}. + */ + private void requirePermittedEndpoint(S3Config config) { + if (config.endpoint() == null) { + return; + } + try { + S3Clients.validateEndpointHost( + URI.create(config.endpoint()), + applicationProperties.getPolicies().isAllowPrivateS3Endpoints(), + "S3 source endpoint", + "set policies.allowPrivateS3Endpoints=true to opt in (e.g. for a local" + + " MinIO)."); + } catch (IllegalStateException e) { + throw new IllegalArgumentException(e.getMessage(), e); + } + } + + private static S3Client buildClient(S3Config config) { + S3ClientBuilder builder = + S3Client.builder() + .httpClient(UrlConnectionHttpClient.create()) + .region(Region.of(config.region())) + // Path-style addressing whenever a custom endpoint is set: S3-compatible + // stores rarely support virtual-hosted bucket DNS. + .serviceConfiguration( + S3Configuration.builder() + .pathStyleAccessEnabled(config.endpoint() != null) + .build()) + .credentialsProvider( + StaticCredentialsProvider.create( + AwsBasicCredentials.create( + config.accessKeyId(), config.secretAccessKey()))); + if (config.endpoint() != null) { + builder.endpointOverride(URI.create(config.endpoint())); + } + return builder.build(); + } + + @PreDestroy + void closeClients() { + clients.values().forEach(S3Client::close); + clients.clear(); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3Identities.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3Identities.java new file mode 100644 index 0000000000..ae2d279b9f --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3Identities.java @@ -0,0 +1,28 @@ +package stirling.software.proprietary.policy.s3; + +import java.time.Instant; + +/** + * The S3 backend's identity and version scheme, shared by {@code S3InputSource} and {@code + * S3OutputSink} so outputs are recorded under exactly the identity and gate the next listing + * derives. Identity is {@code s3://bucket/key}; the gate is the ETag every listing returns for free + * (multipart ETags are not content hashes, so any ETag change simply reads as a new version). + */ +public final class S3Identities { + + private S3Identities() {} + + public static String identity(String bucket, String key) { + return "s3://" + bucket + "/" + key; + } + + /** ETag stripped of its quotes; falls back to size:lastModified for stores that omit it. */ + public static String gate(String eTag, Long size, Instant lastModified) { + if (eTag != null && !eTag.isBlank()) { + return eTag.replace("\"", ""); + } + return (size == null ? -1 : size) + + ":" + + (lastModified == null ? 0 : lastModified.toEpochMilli()); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java index 27d60703de..508e33a335 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java @@ -1,6 +1,7 @@ package stirling.software.proprietary.policy.source; import java.util.List; +import java.util.Map; import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; import org.springframework.http.HttpStatus; @@ -29,6 +30,7 @@ import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.store.PolicyStore; import stirling.software.proprietary.policy.trigger.PolicyTriggerManager; +import stirling.software.proprietary.util.SecretMasker; /** * CRUD for persisted, reusable input connections plus the Sources overview for the admin portal. A @@ -65,11 +67,16 @@ public class SourceController { } @GetMapping("/{sourceId}") - @Operation(summary = "Get a source by id") + @Operation( + summary = "Get a source by id", + description = + "Secret-bearing options are returned as a redaction sentinel, never their" + + " stored values; an edit that sends the sentinel back keeps them.") public ResponseEntity get(@PathVariable String sourceId) { return sourceStore .get(sourceId) .filter(sourceAccessGuard::canAccess) + .map(SourceController::withMaskedSecrets) .map(ResponseEntity::ok) .orElseGet(() -> ResponseEntity.notFound().build()); } @@ -97,7 +104,7 @@ public class SourceController { + " matching source type.") public ResponseEntity save(@RequestBody Source source) { requireSourceEditingAllowed(); - Source owned = resolveOwnership(source); + Source owned = withStoredSecrets(resolveOwnership(source)); try { validateConfig(owned); } catch (IllegalArgumentException e) { @@ -107,7 +114,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(saved); + return ResponseEntity.ok(withMaskedSecrets(saved)); } @DeleteMapping("/{sourceId}") @@ -169,6 +176,42 @@ public class SourceController { teamId); } + private static Source withOptions(Source source, Map options) { + return new Source( + source.id(), + source.name(), + source.type(), + options, + source.enabled(), + source.owner(), + source.teamId()); + } + + /** Secrets never leave the server: reads return the redaction sentinel in their place. */ + private static Source withMaskedSecrets(Source source) { + return withOptions(source, SecretMasker.mask(source.options())); + } + + /** + * An edit that round-trips a masked read sends secrets back as the sentinel; restore them from + * the stored source so saving without re-typing keeps them (validation then runs against the + * real values). + */ + private Source withStoredSecrets(Source incoming) { + if (incoming.id() == null || incoming.id().isBlank()) { + return incoming; + } + return sourceStore + .get(incoming.id()) + .map( + existing -> + withOptions( + incoming, + SecretMasker.restoreRedacted( + incoming.options(), existing.options()))) + .orElse(incoming); + } + /** 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(); diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceEntity.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceEntity.java index 412f862f89..c94e4cb079 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceEntity.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceEntity.java @@ -3,6 +3,7 @@ package stirling.software.proprietary.policy.source; import java.io.Serializable; import jakarta.persistence.Column; +import jakarta.persistence.Convert; import jakarta.persistence.Entity; import jakarta.persistence.Id; import jakarta.persistence.Table; @@ -11,6 +12,8 @@ import lombok.Getter; import lombok.NoArgsConstructor; import lombok.Setter; +import stirling.software.proprietary.integration.crypto.LenientEncryptedStringConverter; + /** * JPA row for a {@link Source}. The whole source lives as JSON in {@code sourceJson} (authoritative * on read); the scalar columns are denormalized copies for querying. {@code owner} and {@code @@ -45,6 +48,9 @@ public class SourceEntity implements Serializable { @Column(name = "enabled") private boolean enabled; + // Encrypted at rest: source options carry user-supplied credentials (e.g. an S3 secret + // access key). Lenient so rows written before encryption shipped still load. + @Convert(converter = LenientEncryptedStringConverter.class) @Column(name = "source_json", columnDefinition = "text") private String sourceJson; } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java index 6e5f92ce44..af7ec3b11b 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java @@ -14,6 +14,7 @@ import lombok.RequiredArgsConstructor; import stirling.software.proprietary.policy.config.PolicyAccessGuard; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.store.PolicyStore; +import stirling.software.proprietary.util.SecretMasker; /** * Builds the Sources overview: every persisted source the caller's team owns, shown exactly once, @@ -104,13 +105,18 @@ public class SourceOverviewService { return referenceCount == 0 ? "unused" : "active"; } - /** Generic key/value view of the source's config - works for any source type. */ + /** + * Generic key/value view of the source's config - works for any source type. Secret-bearing + * options (e.g. an S3 secret access key) are redacted, not omitted, so the overview still shows + * that a credential is configured. + */ private static List configRows(Source source) { - return source.options().entrySet().stream() + Map masked = SecretMasker.mask(source.options()); + return source.options().keySet().stream() .map( - entry -> + key -> new SourceView.DetailRow( - humanize(entry.getKey()), String.valueOf(entry.getValue()))) + humanize(key), String.valueOf(masked.get(key)))) .toList(); } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyEntity.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyEntity.java index 08e6bd4fbe..e393e47833 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyEntity.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyEntity.java @@ -3,6 +3,7 @@ package stirling.software.proprietary.policy.store; import java.io.Serializable; import jakarta.persistence.Column; +import jakarta.persistence.Convert; import jakarta.persistence.Entity; import jakarta.persistence.Id; import jakarta.persistence.Table; @@ -11,6 +12,8 @@ import lombok.Getter; import lombok.NoArgsConstructor; import lombok.Setter; +import stirling.software.proprietary.integration.crypto.LenientEncryptedStringConverter; + /** * JPA row for a {@link stirling.software.proprietary.policy.model.Policy}. The whole policy lives * as JSON in {@code policyJson} (authoritative on read); the scalar columns are denormalized copies @@ -55,6 +58,9 @@ public class PolicyEntity implements Serializable { @Column(name = "sort_order") private Integer sortOrder; + // Encrypted at rest: output options carry user-supplied credentials (e.g. an S3 secret + // access key). Lenient so rows written before encryption shipped still load. + @Convert(converter = LenientEncryptedStringConverter.class) @Column(name = "policy_json", columnDefinition = "text") private String policyJson; } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ScheduleTrigger.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ScheduleTrigger.java index 3e747d7169..32ba4ec77d 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ScheduleTrigger.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ScheduleTrigger.java @@ -99,11 +99,17 @@ public class ScheduleTrigger implements PolicyTrigger { // Baseline a newly-seen policy to now so it does not fire immediately. Instant last = lastFiredByPolicy.computeIfAbsent(policy.id(), id -> now); ZonedDateTime next = config.schedule().nextAfter(last.atZone(config.zone())); - if (!next.toInstant().isAfter(now)) { - lastFiredByPolicy.put(policy.id(), now); - log.info("Scheduled policy {} ({}) is due", policy.id(), policy.name()); - policyRunner.run(policy); + if (next.toInstant().isAfter(now)) { + continue; } + ZonedDateTime later = config.schedule().nextAfter(next); + while (!later.toInstant().isAfter(now)) { + next = later; + later = config.schedule().nextAfter(later); + } + lastFiredByPolicy.put(policy.id(), next.toInstant()); + log.info("Scheduled policy {} ({}) is due", policy.id(), policy.name()); + policyRunner.run(policy); } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/util/SecretMasker.java b/app/proprietary/src/main/java/stirling/software/proprietary/util/SecretMasker.java index a3975de194..5a9619c2dd 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/util/SecretMasker.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/util/SecretMasker.java @@ -1,5 +1,6 @@ package stirling.software.proprietary.util; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.regex.Pattern; @@ -13,10 +14,15 @@ import stirling.software.common.util.RegexPatternUtils; @Slf4j public final class SecretMasker { + /** The placeholder masked values are replaced with; reads as "a secret is set". */ + public static final String REDACTED = "********"; + private static final Pattern SENSITIVE = RegexPatternUtils.getInstance() .getPattern( - "(?i)\\b(password|token|secret|api[_-]?key|authorization|auth|jwt|cred|cert)\\b"); + // 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"); private SecretMasker() {} @@ -47,8 +53,28 @@ public final class SecretMasker { private static Object deepMaskValue(String key, Object value) { if (key != null && SENSITIVE.matcher(key).find()) { - return "***REDACTED***"; + return REDACTED; } return deepMask(value); } + + /** + * Restore top-level values the caller sent back as the {@link #REDACTED} sentinel from the + * stored map, so a masked read can round-trip through an edit without re-typing secrets. A + * sentinel with no stored counterpart is left as-is (it fails whatever validates it, rather + * than silently passing an unset secret). + */ + public static Map restoreRedacted( + Map incoming, Map stored) { + if (incoming == null || stored == null) { + return incoming; + } + Map merged = new LinkedHashMap<>(incoming); + merged.replaceAll( + (key, value) -> + REDACTED.equals(value) && stored.containsKey(key) + ? stored.get(key) + : value); + return merged; + } } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/integration/crypto/LenientEncryptedStringConverterTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/integration/crypto/LenientEncryptedStringConverterTest.java new file mode 100644 index 0000000000..b61ded9708 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/integration/crypto/LenientEncryptedStringConverterTest.java @@ -0,0 +1,45 @@ +package stirling.software.proprietary.integration.crypto; + +import static org.assertj.core.api.Assertions.assertThat; + +import javax.crypto.KeyGenerator; +import javax.crypto.SecretKey; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +class LenientEncryptedStringConverterTest { + + private final LenientEncryptedStringConverter converter = new LenientEncryptedStringConverter(); + + @BeforeAll + static void initKey() throws Exception { + KeyGenerator generator = KeyGenerator.getInstance("AES"); + generator.init(256); + SecretKey key = generator.generateKey(); + CredentialEncryption.initialiseForTesting(key); + } + + @Test + void roundTripsThroughCiphertext() { + String json = "{\"bucket\":\"inbox\",\"secretAccessKey\":\"shh\"}"; + + String stored = converter.convertToDatabaseColumn(json); + + assertThat(stored).isNotEqualTo(json).doesNotContain("shh"); + assertThat(converter.convertToEntityAttribute(stored)).isEqualTo(json); + } + + @Test + void legacyPlaintextRowsPassThroughOnRead() { + String legacy = "{\"bucket\":\"inbox\",\"mode\":\"consume\"}"; + + assertThat(converter.convertToEntityAttribute(legacy)).isEqualTo(legacy); + } + + @Test + void nullsPassThrough() { + assertThat(converter.convertToDatabaseColumn(null)).isNull(); + assertThat(converter.convertToEntityAttribute(null)).isNull(); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java index f38263ad3e..5d8961a46e 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java @@ -9,6 +9,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; @@ -17,6 +18,7 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.http.HttpStatus; @@ -34,7 +36,9 @@ import stirling.software.proprietary.policy.engine.PolicyRunHandle; import stirling.software.proprietary.policy.engine.PolicyRunRegistry; import stirling.software.proprietary.policy.engine.PolicyRunner; import stirling.software.proprietary.policy.engine.PolicyValidator; +import stirling.software.proprietary.policy.engine.SweepOutcome; import stirling.software.proprietary.policy.ledger.ProcessedLedger; +import stirling.software.proprietary.policy.model.OutputSpec; import stirling.software.proprietary.policy.model.PipelineDefinition; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; @@ -44,6 +48,7 @@ import stirling.software.proprietary.policy.progress.PolicyProgressListener; import stirling.software.proprietary.policy.source.SourceAccessGuard; import stirling.software.proprietary.policy.source.SourceStore; import stirling.software.proprietary.policy.trigger.PolicyTriggerManager; +import stirling.software.proprietary.util.SecretMasker; @ExtendWith(MockitoExtension.class) @DisplayName("PolicyController") @@ -128,6 +133,17 @@ class PolicyControllerTest { return new Policy(id, "name", "owner", true, null, List.of(), List.of(), null, teamId); } + private static Policy s3OutputPolicy(String id, String secret) { + OutputSpec output = + new OutputSpec( + "s3", + Map.of( + "bucket", "outbox", + "accessKeyId", "AKIAEXAMPLE", + "secretAccessKey", secret)); + return new Policy(id, "name", "owner", true, null, List.of(), List.of(), output, 1L); + } + private static PolicyRunHandle handle(String runId) { PolicyRun run = new PolicyRun(runId, null, definitionWithStep()); return new PolicyRunHandle(runId, CompletableFuture.completedFuture(run)); @@ -263,6 +279,27 @@ class PolicyControllerTest { verify(policyTriggerManager).notifyPoliciesChanged(); } + @Test + @DisplayName("saving the sentinel back keeps the stored output secret") + void saveRestoresOutputSecrets() { + applicationProperties.getSecurity().setEnableLogin(false); + Policy existing = s3OutputPolicy("p1", "shh"); + when(policyStore.get("p1")).thenReturn(Optional.of(existing)); + when(policyAccessGuard.canAccess(existing)).thenReturn(true); + when(policyStore.save(any())).thenAnswer(i -> i.getArgument(0)); + + ResponseEntity response = + controller.savePolicy(s3OutputPolicy("p1", SecretMasker.REDACTED)); + + ArgumentCaptor stored = ArgumentCaptor.forClass(Policy.class); + verify(policyStore).save(stored.capture()); + assertThat(stored.getValue().output().options().get("secretAccessKey")) + .isEqualTo("shh"); + // The save response is masked again; only the store sees the real value. + assertThat(response.getBody().output().options().get("secretAccessKey")) + .isEqualTo(SecretMasker.REDACTED); + } + @Test @DisplayName("forbidden when login enabled and caller cannot edit") void forbidden() { @@ -363,6 +400,20 @@ class PolicyControllerTest { assertThat(response.getBody().id()).isEqualTo("a"); } + @Test + @DisplayName("getPolicy returns output secrets as the redaction sentinel") + void getMasksOutputSecrets() { + Policy p = s3OutputPolicy("a", "shh"); + when(policyStore.get("a")).thenReturn(Optional.of(p)); + when(policyAccessGuard.canAccess(p)).thenReturn(true); + + Policy read = controller.getPolicy("a").getBody(); + + assertThat(read.output().options().get("secretAccessKey")) + .isEqualTo(SecretMasker.REDACTED); + assertThat(read.output().options().get("bucket")).isEqualTo("outbox"); + } + @Test @DisplayName("getPolicy returns 404 when not accessible") void getNotAccessible() { @@ -536,17 +587,18 @@ class PolicyControllerTest { } @Test - @DisplayName("trigger runs an accessible policy against its sources and returns run ids") + @DisplayName("trigger runs an accessible policy against its sources and returns the sweep") void triggersRun() { Policy p = policy("a", 1L); when(policyStore.get("a")).thenReturn(Optional.of(p)); when(policyAccessGuard.canAccess(p)).thenReturn(true); - when(policyRunner.run(p)).thenReturn(List.of("run-a", "run-b")); + SweepOutcome outcome = new SweepOutcome(List.of("run-a", "run-b"), 3, 1, 0, 0); + when(policyRunner.run(p)).thenReturn(outcome); - ResponseEntity> response = controller.trigger("a"); + ResponseEntity response = controller.trigger("a"); assertThat(response.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); - assertThat(response.getBody()).containsExactly("run-a", "run-b"); + assertThat(response.getBody()).isEqualTo(outcome); } @Test diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java index 13189b2481..ef4c0300da 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java @@ -1,5 +1,6 @@ package stirling.software.proprietary.policy.engine; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -30,6 +31,7 @@ import org.mockito.junit.jupiter.MockitoExtension; import stirling.software.proprietary.policy.input.InputSource; import stirling.software.proprietary.policy.input.ResolveContext; import stirling.software.proprietary.policy.input.ResolvedInput; +import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger; import stirling.software.proprietary.policy.ledger.ProcessedLedger; import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.model.OutputSpec; @@ -85,6 +87,42 @@ class PolicyRunnerTest { verify(processedLedger).deleteUnseen(eq("p1"), anyLong()); } + @Test + void reportsWhatTheSweepSkippedSoAnEmptyTriggerExplainsItself() throws Exception { + InProcessProcessedLedger ledger = new InProcessProcessedLedger(); + PolicyRunner reporting = + new PolicyRunner( + policyEngine, + List.of(folderSource), + sourceStore, + new InProcessSourceDocCounter(), + ledger); + InputSpec spec = InputSpec.folder("/in"); + Policy policy = policy(List.of(spec)); + // One file already processed at its current version, one parked by a failed run. + ledger.claim("p1", "/in/done.pdf", "g1", null); + ledger.settle("p1", "/in/done.pdf", "g1", null, true); + ledger.claim("p1", "/in/failed.pdf", "g2", null); + ledger.settle("p1", "/in/failed.pdf", "g2", null, false); + when(folderSource.supports(spec)).thenReturn(true); + when(folderSource.resolve(eq(spec), any())) + .thenAnswer( + invocation -> { + ResolveContext ctx = invocation.getArgument(1); + ctx.reportPresent(List.of("/in/done.pdf", "/in/failed.pdf")); + // Both are at their settled versions, so neither claims. + return List.of(); + }); + + SweepOutcome outcome = reporting.run(policy); + + assertTrue(outcome.runIds().isEmpty()); + assertEquals(2, outcome.filesListed()); + assertEquals(1, outcome.alreadyProcessed()); + assertEquals(1, outcome.parked()); + assertEquals(0, outcome.inFlight()); + } + @Test void pullsEverySourceAndRunsOnePerUnitOfWork() throws Exception { InputSpec spec = InputSpec.folder("/in"); diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/S3InputSourceMinioTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/S3InputSourceMinioTest.java new file mode 100644 index 0000000000..4e1e1fc305 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/S3InputSourceMinioTest.java @@ -0,0 +1,231 @@ +package stirling.software.proprietary.policy.input; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +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.testcontainers.containers.MinIOContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.s3.S3ConnectionPool; + +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.core.sync.RequestBody; +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.HeadObjectRequest; +import software.amazon.awssdk.services.s3.model.NoSuchKeyException; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; + +/** + * End-to-end {@link S3InputSource} test against a real S3 API (MinIO), through the production + * client factory: listing, claiming, streaming, consensus delete, and save-time validation. + */ +@Testcontainers(disabledWithoutDocker = true) +class S3InputSourceMinioTest { + + private static final String POLICY = "p1"; + private static final String ACCESS_KEY = "minioadmin"; + private static final String SECRET_KEY = "minioadmin"; + + @Container + static MinIOContainer minio = + new MinIOContainer("minio/minio:latest") + .withUserName(ACCESS_KEY) + .withPassword(SECRET_KEY); + + private static S3Client adminClient; + private static int bucketCounter; + + private String bucket; + private S3InputSource source; + private InProcessProcessedLedger ledger; + private RecordingContext ctx; + + @BeforeEach + void setUp() { + if (adminClient == null) { + adminClient = + S3Client.builder() + .endpointOverride(java.net.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 = "policy-inbox-" + ++bucketCounter; + adminClient.createBucket(CreateBucketRequest.builder().bucket(bucket).build()); + + // The MinIO endpoint resolves to loopback, so the operator opt-in must be on. + ApplicationProperties properties = new ApplicationProperties(); + properties.getPolicies().setAllowPrivateS3Endpoints(true); + source = new S3InputSource(new S3ConnectionPool(properties)); + ledger = new InProcessProcessedLedger(); + ctx = new RecordingContext(); + } + + @Test + void consumeListsStreamsAndDeletesByConsensus() throws IOException { + put("incoming/doc.pdf", "pdf bytes"); + put("incoming/other.txt", "text"); + + List work = source.resolve(spec(Map.of("prefix", "incoming/")), ctx); + + assertThat(work).hasSize(2); + assertThat(ctx.present) + .containsExactlyInAnyOrder( + "s3://" + bucket + "/incoming/doc.pdf", + "s3://" + bucket + "/incoming/other.txt"); + assertThat(read(work.get(0))).isIn("pdf bytes", "text"); + // In flight: nothing to claim on a second sweep. + assertThat(source.resolve(spec(Map.of("prefix", "incoming/")), ctx)).isEmpty(); + + work.forEach(unit -> unit.onComplete().accept(true)); + assertThat(exists("incoming/doc.pdf")).isFalse(); + assertThat(exists("incoming/other.txt")).isFalse(); + } + + @Test + void aFailedObjectStaysInTheBucket() throws IOException { + put("doc.pdf", "data"); + + source.resolve(spec(Map.of()), ctx).get(0).onComplete().accept(false); + + assertThat(exists("doc.pdf")).isTrue(); + assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty(); + } + + @Test + void anObjectOverwrittenMidRunSurvivesTheDeleteAndRunsAgain() throws IOException { + put("doc.pdf", "v1"); + + List work = source.resolve(spec(Map.of()), ctx); + put("doc.pdf", "v2 with a different etag"); + work.get(0).onComplete().accept(true); + + assertThat(exists("doc.pdf")).isTrue(); + assertThat(source.resolve(spec(Map.of()), ctx)).hasSize(1); + } + + @Test + void prefixLimitsWhatIsRead() throws IOException { + put("incoming/doc.pdf", "data"); + put("archive/old.pdf", "data"); + + List work = source.resolve(spec(Map.of("prefix", "incoming/")), ctx); + + assertThat(work).hasSize(1); + assertThat(ctx.present).containsExactly("s3://" + bucket + "/incoming/doc.pdf"); + } + + @Test + void validateAcceptsAReachableBucketAndRejectsBadCredentials() { + source.validate(spec(Map.of())); + + Map wrongSecret = new HashMap<>(baseOptions()); + wrongSecret.put("secretAccessKey", "not-the-secret"); + assertThatThrownBy(() -> source.validate(new InputSpec("s3", wrongSecret))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("cannot access"); + + Map missingBucket = new HashMap<>(baseOptions()); + missingBucket.put("bucket", "no-such-bucket-here"); + assertThatThrownBy(() -> source.validate(new InputSpec("s3", missingBucket))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("cannot access"); + } + + @Test + void aPrivateEndpointIsRejectedWithoutTheOperatorOptIn() { + S3InputSource guarded = + new S3InputSource(new S3ConnectionPool(new ApplicationProperties())); + + assertThatThrownBy(() -> guarded.validate(spec(Map.of()))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("policies.allowPrivateS3Endpoints"); + } + + private Map baseOptions() { + return Map.of( + "bucket", bucket, + "endpoint", minio.getS3URL(), + "accessKeyId", ACCESS_KEY, + "secretAccessKey", SECRET_KEY); + } + + private InputSpec spec(Map extra) { + Map options = new HashMap<>(baseOptions()); + options.putAll(extra); + return new InputSpec("s3", options); + } + + private void put(String key, String content) { + adminClient.putObject( + PutObjectRequest.builder().bucket(bucket).key(key).build(), + RequestBody.fromString(content, StandardCharsets.UTF_8)); + } + + private boolean exists(String key) { + try { + adminClient.headObject(HeadObjectRequest.builder().bucket(bucket).key(key).build()); + return true; + } catch (NoSuchKeyException e) { + return false; + } + } + + private static String read(ResolvedInput unit) throws IOException { + try (InputStream stream = unit.inputs().primary().get(0).getInputStream()) { + return new String(stream.readAllBytes(), StandardCharsets.UTF_8); + } + } + + private class RecordingContext implements ResolveContext { + + private final List present = new ArrayList<>(); + + @Override + public boolean claim(String identity, String gate, Supplier contentHash) { + return ledger.claim(POLICY, identity, gate, contentHash); + } + + @Override + public void settle( + String identity, String finalGate, String finalContentHash, boolean success) { + ledger.settle(POLICY, identity, finalGate, finalContentHash, success); + } + + @Override + public boolean allSettledDone(String identity) { + return ledger.allSettledDone(identity); + } + + @Override + public void reportPresent(Collection identities) { + present.addAll(identities); + } + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/S3InputSourceTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/S3InputSourceTest.java new file mode 100644 index 0000000000..73995248dd --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/S3InputSourceTest.java @@ -0,0 +1,327 @@ +package stirling.software.proprietary.policy.input; + +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.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +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.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.s3.S3ConnectionPool; + +import software.amazon.awssdk.core.ResponseInputStream; +import software.amazon.awssdk.core.exception.SdkClientException; +import software.amazon.awssdk.http.AbortableInputStream; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.DeleteObjectRequest; +import software.amazon.awssdk.services.s3.model.GetObjectRequest; +import software.amazon.awssdk.services.s3.model.GetObjectResponse; +import software.amazon.awssdk.services.s3.model.HeadObjectRequest; +import software.amazon.awssdk.services.s3.model.HeadObjectResponse; +import software.amazon.awssdk.services.s3.model.ListObjectsV2Request; +import software.amazon.awssdk.services.s3.model.ListObjectsV2Response; +import software.amazon.awssdk.services.s3.model.S3Object; + +/** + * Tests for {@link S3InputSource}: consume mode tracks objects in place through the ledger and + * removes them by consensus, snapshot stays stateless, and discovery skips folder placeholders and + * dot-prefixed keys. + */ +@ExtendWith(MockitoExtension.class) +class S3InputSourceTest { + + private static final String POLICY = "p1"; + private static final String BUCKET = "inbox-bucket"; + + @Mock private S3Client s3Client; + + private S3InputSource source; + private InProcessProcessedLedger ledger; + private RecordingContext ctx; + + @BeforeEach + void setUp() { + source = + new S3InputSource( + new S3ConnectionPool(new ApplicationProperties(), config -> s3Client)); + ledger = new InProcessProcessedLedger(); + ctx = new RecordingContext(); + } + + @Test + void consumeRemovesTheObjectOnceProcessed() throws IOException { + listingReturns(object("doc.pdf", "\"etag-1\"")); + headReturns("doc.pdf", "\"etag-1\""); + + List work = source.resolve(spec(), ctx); + + assertEquals(1, work.size()); + assertEquals(1, work.get(0).inputs().primary().size()); + // In flight: a second sweep does not pick it up again. + assertTrue(source.resolve(spec(), ctx).isEmpty()); + + work.get(0).onComplete().accept(true); + verify(s3Client).deleteObject(any(DeleteObjectRequest.class)); + assertTrue(source.resolve(spec(), ctx).isEmpty()); + } + + @Test + void anObjectReplacedMidRunSurvivesTheDelete() throws IOException { + listingReturns(object("doc.pdf", "\"etag-1\"")); + // The object is overwritten while the run is executing. + headReturns("doc.pdf", "\"etag-2\""); + + List work = source.resolve(spec(), ctx); + work.get(0).onComplete().accept(true); + + // The delete is version-guarded: the replacement is not the object that ran, so it stays + // and is claimed as fresh work instead of being marked processed. + verify(s3Client, never()).deleteObject(any(DeleteObjectRequest.class)); + listingReturns(object("doc.pdf", "\"etag-2\"")); + assertEquals(1, source.resolve(spec(), ctx).size()); + } + + @Test + void aSharedObjectIsRemovedOnlyOnceEveryPolicyHasProcessedIt() throws IOException { + listingReturns(object("doc.pdf", "\"etag-1\"")); + headReturns("doc.pdf", "\"etag-1\""); + RecordingContext other = new RecordingContext("p2"); + + List mine = source.resolve(spec(), ctx); + List theirs = source.resolve(spec(), other); + assertEquals(1, mine.size()); + assertEquals(1, theirs.size()); + + mine.get(0).onComplete().accept(true); + // The other policy's claim is still in flight, so the first finisher must not delete. + verify(s3Client, never()).deleteObject(any(DeleteObjectRequest.class)); + + theirs.get(0).onComplete().accept(true); + verify(s3Client).deleteObject(any(DeleteObjectRequest.class)); + } + + @Test + void aFailedObjectStaysAndIsNotRetriedUntilItChanges() throws IOException { + listingReturns(object("doc.pdf", "\"etag-1\"")); + + source.resolve(spec(), ctx).get(0).onComplete().accept(false); + + verify(s3Client, never()).deleteObject(any(DeleteObjectRequest.class)); + assertTrue(source.resolve(spec(), ctx).isEmpty()); + + // A new upload carries a new ETag, which reads as a new version and retries. + listingReturns(object("doc.pdf", "\"etag-2\"")); + assertEquals(1, source.resolve(spec(), ctx).size()); + } + + @Test + void snapshotReadsStatelesslyEverySweep() throws IOException { + listingReturns(object("doc.pdf", "\"etag-1\"")); + InputSpec spec = new InputSpec("s3", options(Map.of("mode", "snapshot"))); + + List first = source.resolve(spec, ctx); + first.get(0).onComplete().accept(true); + List second = source.resolve(spec, ctx); + + assertEquals(1, first.size()); + assertEquals(1, second.size()); + verify(s3Client, never()).deleteObject(any(DeleteObjectRequest.class)); + assertTrue(ctx.present.isEmpty()); + } + + @Test + void folderPlaceholdersAndDotPrefixedKeysAreSkipped() throws IOException { + listingReturns( + object("doc.pdf", "\"etag-1\""), + object("incoming/", "\"etag-2\""), + object(".stirling/tmp/staged.pdf", "\"etag-3\""), + object("incoming/.hidden.pdf", "\"etag-4\"")); + + List work = source.resolve(spec(), ctx); + + assertEquals(1, work.size()); + assertEquals(List.of("s3://" + BUCKET + "/doc.pdf"), ctx.present); + } + + @Test + void listingPagesAreAllRead() throws IOException { + ListObjectsV2Response firstPage = + ListObjectsV2Response.builder() + .contents(object("a.pdf", "\"etag-a\"")) + .nextContinuationToken("next") + .build(); + ListObjectsV2Response secondPage = + ListObjectsV2Response.builder().contents(object("b.pdf", "\"etag-b\"")).build(); + when(s3Client.listObjectsV2(any(ListObjectsV2Request.class))) + .thenReturn(firstPage, secondPage); + + assertEquals(2, source.resolve(spec(), ctx).size()); + } + + @Test + void aListingFailurePropagatesSoTheSweepVetoesCleanup() { + when(s3Client.listObjectsV2(any(ListObjectsV2Request.class))) + .thenThrow(SdkClientException.create("connection refused")); + + assertThrows(SdkClientException.class, () -> source.resolve(spec(), ctx)); + } + + @Test + void resourceStreamsTheObjectAndNamesItByKeyBasename() throws IOException { + listingReturns(object("incoming/doc.pdf", "\"etag-1\"")); + byte[] payload = "data".getBytes(StandardCharsets.UTF_8); + when(s3Client.getObject(any(GetObjectRequest.class))) + .thenReturn( + new ResponseInputStream<>( + GetObjectResponse.builder().build(), + AbortableInputStream.create(new ByteArrayInputStream(payload)))); + + var resource = source.resolve(spec(), ctx).get(0).inputs().primary().get(0); + + assertEquals("doc.pdf", resource.getFilename()); + // Content length comes from the listing, not a download. + assertEquals(4, resource.contentLength()); + try (var stream = resource.getInputStream()) { + assertEquals("data", new String(stream.readAllBytes(), StandardCharsets.UTF_8)); + } + } + + @Test + void aMissingETagFallsBackToSizeAndLastModified() throws IOException { + Instant modified = Instant.parse("2026-01-01T00:00:00Z"); + listingReturns(S3Object.builder().key("doc.pdf").size(4L).lastModified(modified).build()); + + assertEquals(1, source.resolve(spec(), ctx).size()); + // The same gate on the next sweep reads as already claimed. + listingReturns(S3Object.builder().key("doc.pdf").size(4L).lastModified(modified).build()); + assertTrue(source.resolve(spec(), ctx).isEmpty()); + } + + @Test + void validateRejectsBadConfig() { + // No bucket. + assertThrows( + IllegalArgumentException.class, + () -> source.validate(new InputSpec("s3", Map.of()))); + // Credentials are required, never the server's own identity - together and individually. + assertThrows( + IllegalArgumentException.class, + () -> source.validate(new InputSpec("s3", Map.of("bucket", BUCKET)))); + assertThrows( + IllegalArgumentException.class, + () -> + source.validate( + new InputSpec( + "s3", Map.of("bucket", BUCKET, "accessKeyId", "AKIA")))); + assertThrows( + IllegalArgumentException.class, + () -> source.validate(new InputSpec("s3", options(Map.of("mode", "sideways"))))); + assertThrows( + IllegalArgumentException.class, + () -> + source.validate( + new InputSpec( + "s3", options(Map.of("endpoint", "ftp://example.com"))))); + } + + @Test + void validateRejectsAnUnreachableBucket() { + when(s3Client.listObjectsV2(any(ListObjectsV2Request.class))) + .thenThrow(SdkClientException.create("connection refused")); + + assertThrows(IllegalArgumentException.class, () -> source.validate(spec())); + } + + private static InputSpec spec() { + return new InputSpec("s3", options(Map.of())); + } + + /** The required options (bucket + credentials) plus any extras under test. */ + private static Map options(Map extra) { + Map options = new HashMap<>(extra); + options.put("bucket", BUCKET); + options.put("accessKeyId", "AKIAEXAMPLE"); + options.put("secretAccessKey", "shh"); + return options; + } + + private static S3Object object(String key, String eTag) { + return S3Object.builder() + .key(key) + .eTag(eTag) + .size(4L) + .lastModified(Instant.parse("2026-01-01T00:00:00Z")) + .build(); + } + + private void listingReturns(S3Object... objects) { + when(s3Client.listObjectsV2(any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(objects).build()); + } + + private void headReturns(String key, String eTag) { + when(s3Client.headObject(any(HeadObjectRequest.class))) + .thenReturn( + HeadObjectResponse.builder() + .eTag(eTag) + .contentLength(4L) + .lastModified(Instant.parse("2026-01-01T00:00:00Z")) + .build()); + } + + private class RecordingContext implements ResolveContext { + + private final String policyId; + private final List present = new ArrayList<>(); + + private RecordingContext() { + this(POLICY); + } + + private RecordingContext(String policyId) { + this.policyId = policyId; + } + + @Override + public boolean claim(String identity, String gate, Supplier contentHash) { + return ledger.claim(policyId, identity, gate, contentHash); + } + + @Override + public void settle( + String identity, String finalGate, String finalContentHash, boolean success) { + ledger.settle(policyId, identity, finalGate, finalContentHash, success); + } + + @Override + public boolean allSettledDone(String identity) { + return ledger.allSettledDone(identity); + } + + @Override + public void reportPresent(Collection identities) { + present.addAll(identities); + } + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/S3OutputSinkMinioTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/S3OutputSinkMinioTest.java new file mode 100644 index 0000000000..a2a5a41ca0 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/S3OutputSinkMinioTest.java @@ -0,0 +1,211 @@ +package stirling.software.proprietary.policy.output; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.nio.charset.StandardCharsets; +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.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.Resource; +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.model.job.ResultFile; +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.ledger.InProcessProcessedLedger; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.s3.S3ConnectionPool; + +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.core.ResponseInputStream; +import software.amazon.awssdk.core.sync.RequestBody; +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.GetObjectRequest; +import software.amazon.awssdk.services.s3.model.GetObjectResponse; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; + +/** + * End-to-end {@link S3OutputSink} test against a real S3 API (MinIO): uploads, collision renaming, + * and - composed with {@link S3InputSource} - the loop-safety guarantee that a policy writing into + * a bucket it also watches never re-ingests its own outputs, while a second policy still can. + */ +@Testcontainers(disabledWithoutDocker = true) +class S3OutputSinkMinioTest { + + private static final String POLICY = "p1"; + private static final String ACCESS_KEY = "minioadmin"; + private static final String SECRET_KEY = "minioadmin"; + + @Container + static MinIOContainer minio = + new MinIOContainer("minio/minio:latest") + .withUserName(ACCESS_KEY) + .withPassword(SECRET_KEY); + + private static S3Client adminClient; + private static int bucketCounter; + + private String bucket; + private S3OutputSink sink; + private S3InputSource source; + private InProcessProcessedLedger ledger; + + @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 = "policy-outbox-" + ++bucketCounter; + adminClient.createBucket(CreateBucketRequest.builder().bucket(bucket).build()); + + ApplicationProperties properties = new ApplicationProperties(); + properties.getPolicies().setAllowPrivateS3Endpoints(true); + S3ConnectionPool pool = new S3ConnectionPool(properties); + ledger = new InProcessProcessedLedger(); + sink = new S3OutputSink(pool, ledger); + source = new S3InputSource(pool); + } + + @Test + void uploadsOutputsUnderThePrefix() throws IOException { + List results = + sink.deliver( + new OutputDelivery("run-1", POLICY), + List.of(output("doc.pdf", "pdf bytes")), + outputSpec("processed/")); + + assertThat(results).hasSize(1); + assertThat(results.get(0).getFileName()).isEqualTo("s3://" + bucket + "/processed/doc.pdf"); + assertThat(objectContent("processed/doc.pdf")).isEqualTo("pdf bytes"); + } + + @Test + void anExistingKeyIsNeverOverwritten() throws IOException { + adminClient.putObject( + PutObjectRequest.builder().bucket(bucket).key("doc.pdf").build(), + RequestBody.fromString("theirs", StandardCharsets.UTF_8)); + + List results = + sink.deliver( + new OutputDelivery("run-1", POLICY), + List.of(output("doc.pdf", "ours")), + outputSpec("")); + + assertThat(results.get(0).getFileName()).isEqualTo("s3://" + bucket + "/doc (1).pdf"); + assertThat(objectContent("doc.pdf")).isEqualTo("theirs"); + assertThat(objectContent("doc (1).pdf")).isEqualTo("ours"); + } + + @Test + void aPolicyWritingIntoItsWatchedBucketSkipsItsOwnOutputsButAnotherPolicyChains() + throws IOException { + sink.deliver( + new OutputDelivery("run-1", POLICY), + List.of(output("result.pdf", "produced")), + outputSpec("")); + + // The producing policy's sweep sees its own output at the recorded gate and skips it. + assertThat(source.resolve(inputSpec(), new RecordingContext(POLICY))).isEmpty(); + + // A different policy watching the same bucket has no row and processes it - chaining. + List chained = source.resolve(inputSpec(), new RecordingContext("p2")); + assertThat(chained).hasSize(1); + try (InputStream stream = chained.get(0).inputs().primary().get(0).getInputStream()) { + assertThat(new String(stream.readAllBytes(), StandardCharsets.UTF_8)) + .isEqualTo("produced"); + } + } + + private OutputSpec outputSpec(String prefix) { + return new OutputSpec( + "s3", + Map.of( + "bucket", bucket, + "prefix", prefix, + "endpoint", minio.getS3URL(), + "accessKeyId", ACCESS_KEY, + "secretAccessKey", SECRET_KEY)); + } + + private InputSpec inputSpec() { + return new InputSpec( + "s3", + Map.of( + "bucket", bucket, + "endpoint", minio.getS3URL(), + "accessKeyId", ACCESS_KEY, + "secretAccessKey", SECRET_KEY)); + } + + private String objectContent(String key) throws IOException { + try (ResponseInputStream stream = + adminClient.getObject(GetObjectRequest.builder().bucket(bucket).key(key).build())) { + return new String(stream.readAllBytes(), StandardCharsets.UTF_8); + } + } + + private static Resource output(String name, String content) { + return new ByteArrayResource(content.getBytes(StandardCharsets.UTF_8)) { + @Override + public String getFilename() { + return name; + } + }; + } + + private class RecordingContext implements ResolveContext { + + private final String policyId; + + private RecordingContext(String policyId) { + this.policyId = policyId; + } + + @Override + public boolean claim(String identity, String gate, Supplier contentHash) { + return ledger.claim(policyId, identity, gate, contentHash); + } + + @Override + public void settle( + String identity, String finalGate, String finalContentHash, boolean success) { + ledger.settle(policyId, identity, finalGate, finalContentHash, success); + } + + @Override + public boolean allSettledDone(String identity) { + return ledger.allSettledDone(identity); + } + + @Override + public void reportPresent(Collection identities) {} + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/S3OutputSinkTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/S3OutputSinkTest.java new file mode 100644 index 0000000000..8240bcc4e1 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/S3OutputSinkTest.java @@ -0,0 +1,267 @@ +package stirling.software.proprietary.policy.output; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +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.Mockito.when; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.HexFormat; +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 org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.Resource; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.common.model.job.ResultFile; +import stirling.software.proprietary.policy.ledger.ClaimState; +import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger; +import stirling.software.proprietary.policy.ledger.ProcessedFileStatus; +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.s3.S3ConnectionPool; + +import software.amazon.awssdk.awscore.exception.AwsServiceException; +import software.amazon.awssdk.core.exception.SdkClientException; +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.HeadObjectRequest; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; +import software.amazon.awssdk.services.s3.model.PutObjectResponse; +import software.amazon.awssdk.services.s3.model.S3Exception; + +/** + * Tests for {@link S3OutputSink}: the ledger row exists before the object is visible, collisions + * re-pick names, ad-hoc runs record nothing, and encrypted-bucket ETags are re-recorded. + */ +@ExtendWith(MockitoExtension.class) +class S3OutputSinkTest { + + private static final String POLICY = "p1"; + private static final String BUCKET = "outbox-bucket"; + private static final OutputDelivery DELIVERY = new OutputDelivery("run-1", POLICY); + private static final OutputDelivery AD_HOC = new OutputDelivery("run-2", null); + + @Mock private S3Client s3Client; + + private S3OutputSink sink; + private InProcessProcessedLedger ledger; + private final List puts = new ArrayList<>(); + + @BeforeEach + void setUp() { + ledger = new InProcessProcessedLedger(); + sink = + new S3OutputSink( + new S3ConnectionPool(new ApplicationProperties(), config -> s3Client), + ledger); + } + + @Test + void recordsTheRowBeforeTheObjectBecomesVisible() throws IOException { + // The row for the exact key must already be settled DONE at the moment the PUT runs - + // record-before-visible, asserted from inside the upload itself. + List stateAtPutTime = new ArrayList<>(); + when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class))) + .thenAnswer( + invocation -> { + PutObjectRequest request = invocation.getArgument(0); + puts.add(request); + stateAtPutTime.add(stateFor(identity(request.key()))); + return PutObjectResponse.builder().eTag(quotedMd5("data")).build(); + }); + + List results = + sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec()); + + assertEquals(1, results.size()); + assertEquals("s3://" + BUCKET + "/processed/doc.pdf", results.get(0).getFileName()); + assertEquals(4, results.get(0).getFileSize()); + assertNotNull(stateAtPutTime.get(0)); + assertEquals(ProcessedFileStatus.DONE, stateAtPutTime.get(0).status()); + assertEquals(md5("data"), stateAtPutTime.get(0).gate()); + assertTrue(puts.get(0).ifNoneMatch() != null); + } + + @Test + void aTakenKeyIsForgottenAndRePicked() throws IOException { + when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class))) + .thenAnswer( + invocation -> { + PutObjectRequest request = invocation.getArgument(0); + puts.add(request); + if (puts.size() == 1) { + throw s3Error(412, "PreconditionFailed"); + } + return PutObjectResponse.builder().eTag(quotedMd5("data")).build(); + }); + + List results = + sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec()); + + assertEquals("s3://" + BUCKET + "/processed/doc (1).pdf", results.get(0).getFileName()); + // The lost candidate's row is gone; only the delivered key is recorded. + assertNull(stateFor(identity("processed/doc.pdf"))); + assertNotNull(stateFor(identity("processed/doc (1).pdf"))); + } + + @Test + void anEncryptedBucketETagIsReRecordedAtTheActualGate() throws IOException { + when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class))) + .thenReturn(PutObjectResponse.builder().eTag("\"kms-opaque-etag\"").build()); + + sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec()); + + assertEquals("kms-opaque-etag", stateFor(identity("processed/doc.pdf")).gate()); + } + + @Test + void anAdHocDeliveryRecordsNothing() throws IOException { + when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class))) + .thenReturn(PutObjectResponse.builder().eTag(quotedMd5("data")).build()); + + sink.deliver(AD_HOC, List.of(output("doc.pdf", "data")), spec()); + + assertNull(stateFor(identity("processed/doc.pdf"))); + } + + @Test + void aFailedUploadForgetsItsRowAndThrows() { + when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class))) + .thenThrow(SdkClientException.create("connection refused")); + + assertThrows( + IOException.class, + () -> sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec())); + + assertNull(stateFor(identity("processed/doc.pdf"))); + } + + @Test + void aStoreWithoutConditionalPutsFallsBackToExistenceChecks() throws IOException { + when(s3Client.headObject(any(HeadObjectRequest.class))).thenThrow(s3Error(404, "NotFound")); + when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class))) + .thenAnswer( + invocation -> { + PutObjectRequest request = invocation.getArgument(0); + puts.add(request); + if (request.ifNoneMatch() != null) { + throw s3Error(501, "NotImplemented"); + } + return PutObjectResponse.builder().eTag(quotedMd5("data")).build(); + }); + + List results = + sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec()); + + // Same key, second attempt unconditional. + assertEquals("s3://" + BUCKET + "/processed/doc.pdf", results.get(0).getFileName()); + assertEquals(2, puts.size()); + assertNull(puts.get(1).ifNoneMatch()); + assertNotNull(stateFor(identity("processed/doc.pdf"))); + } + + @Test + void aBarePrefixGetsItsSlash() throws IOException { + when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class))) + .thenAnswer( + invocation -> { + puts.add(invocation.getArgument(0)); + return PutObjectResponse.builder().eTag(quotedMd5("data")).build(); + }); + + sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec("processed")); + + assertEquals("processed/doc.pdf", puts.get(0).key()); + } + + @Test + void validateRejectsBadConfigShape() { + assertThrows( + IllegalArgumentException.class, + () -> sink.validate(new OutputSpec("s3", Map.of()))); + // Credentials are required, never the server's own identity. + assertThrows( + IllegalArgumentException.class, + () -> sink.validate(new OutputSpec("s3", Map.of("bucket", BUCKET)))); + assertThrows( + IllegalArgumentException.class, + () -> + sink.validate( + new OutputSpec( + "s3", Map.of("bucket", BUCKET, "accessKeyId", "AKIA")))); + } + + @Test + void supportsOnlyS3Specs() { + assertTrue(sink.supports(spec())); + assertFalse(sink.supports(OutputSpec.inline())); + assertFalse(sink.supports(null)); + } + + private static OutputSpec spec() { + return spec("processed/"); + } + + private static OutputSpec spec(String prefix) { + return new OutputSpec( + "s3", + Map.of( + "bucket", + BUCKET, + "prefix", + prefix, + "accessKeyId", + "AKIAEXAMPLE", + "secretAccessKey", + "shh")); + } + + private static String identity(String key) { + return "s3://" + BUCKET + "/" + key; + } + + private ClaimState stateFor(String identity) { + return ledger.statesFor(POLICY, List.of(identity)).get(identity); + } + + private static Resource output(String name, String content) { + return new ByteArrayResource(content.getBytes(StandardCharsets.UTF_8)) { + @Override + public String getFilename() { + return name; + } + }; + } + + private static String md5(String content) { + try { + return HexFormat.of() + .formatHex( + MessageDigest.getInstance("MD5") + .digest(content.getBytes(StandardCharsets.UTF_8))); + } catch (Exception e) { + throw new IllegalStateException(e); + } + } + + private static String quotedMd5(String content) { + return "\"" + md5(content) + "\""; + } + + private static AwsServiceException s3Error(int status, String code) { + return S3Exception.builder().statusCode(status).message(code).build(); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java index d1d62b7dbe..e71deaf00a 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java @@ -28,6 +28,7 @@ import stirling.software.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.util.SecretMasker; /** * Tests for {@link SourceController}'s delete guard: a source still referenced by a policy is @@ -109,6 +110,80 @@ class SourceControllerTest { assertEquals(404, controller.delete("nope").getStatusCode().value()); } + @Test + void readsReturnSecretsAsTheRedactionSentinel() { + Source saved = sourceStore.save(s3Source("shh")); + + Source read = controller.get(saved.id()).getBody(); + + assertEquals(SecretMasker.REDACTED, read.options().get("secretAccessKey")); + assertEquals("AKIAEXAMPLE", read.options().get("accessKeyId")); + // The store itself keeps the real value. + assertEquals( + "shh", sourceStore.get(saved.id()).orElseThrow().options().get("secretAccessKey")); + } + + @Test + void savingTheSentinelBackKeepsTheStoredSecret() { + Source saved = sourceStore.save(s3Source("shh")); + + Source edited = + new Source( + saved.id(), + "Renamed", + saved.type(), + Map.of( + "bucket", "inbox", + "accessKeyId", "AKIAEXAMPLE", + "secretAccessKey", SecretMasker.REDACTED), + true, + saved.owner(), + saved.teamId()); + Source response = controller.save(edited).getBody(); + + assertEquals( + "shh", sourceStore.get(saved.id()).orElseThrow().options().get("secretAccessKey")); + // The save response is masked too; only the store sees the real value. + assertEquals(SecretMasker.REDACTED, response.options().get("secretAccessKey")); + } + + @Test + void savingANewSecretReplacesTheStoredOne() { + Source saved = sourceStore.save(s3Source("old-secret")); + + Source edited = + new Source( + saved.id(), + saved.name(), + saved.type(), + Map.of( + "bucket", "inbox", + "accessKeyId", "AKIAEXAMPLE", + "secretAccessKey", "new-secret"), + true, + saved.owner(), + saved.teamId()); + controller.save(edited); + + assertEquals( + "new-secret", + sourceStore.get(saved.id()).orElseThrow().options().get("secretAccessKey")); + } + + private static Source s3Source(String secret) { + return new Source( + null, + "Bucket intake", + "s3", + Map.of( + "bucket", "inbox", + "accessKeyId", "AKIAEXAMPLE", + "secretAccessKey", secret), + true, + "owner", + null); + } + private static Source folderSource() { return new Source( null, "Claims intake", "folder", Map.of("directory", "/in"), true, "owner", null); diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/ScheduleTriggerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/ScheduleTriggerTest.java index 889c50ee44..3f81392285 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/ScheduleTriggerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/ScheduleTriggerTest.java @@ -70,6 +70,41 @@ class ScheduleTriggerTest { verify(policyRunner, times(1)).run(eq(policy)); } + @Test + void anIntervalMatchingTheSweepPeriodFiresEverySweepDespiteJitter() { + Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES)); + when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + + Instant t0 = Instant.parse("2026-06-05T10:00:00Z"); + trigger.sweep(t0); // baseline + // The sweep that fires runs a few ms late (scheduler jitter)... + trigger.sweep(t0.plusSeconds(60).plusMillis(5)); + verify(policyRunner, times(1)).run(eq(policy)); + + // ...and the next sweep lands exactly on the 60s grid. Anchoring lastFired to the due + // time (not the jittered observation) means this must still fire, not alias to skip. + trigger.sweep(t0.plusSeconds(120)); + verify(policyRunner, times(2)).run(eq(policy)); + } + + @Test + void aGapFiresOnceNotOncePerMissedInterval() { + Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES)); + when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + + Instant t0 = Instant.parse("2026-06-05T10:00:00Z"); + trigger.sweep(t0); // baseline + // Ten minutes of downtime: nine missed due points collapse into one firing. + trigger.sweep(t0.plusSeconds(600)); + verify(policyRunner, times(1)).run(eq(policy)); + + // Not due again until a full interval after the latest due point. + trigger.sweep(t0.plusSeconds(630)); + verify(policyRunner, times(1)).run(eq(policy)); + trigger.sweep(t0.plusSeconds(660)); + verify(policyRunner, times(2)).run(eq(policy)); + } + @Test void doesNotFireBeforeTheNextScheduledTime() { Policy policy = scheduled("p1", new Schedule.Daily(LocalTime.of(3, 0))); // 03:00 UTC daily diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/util/SecretMaskerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/util/SecretMaskerTest.java index d91aec2029..415117a446 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/util/SecretMaskerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/util/SecretMaskerTest.java @@ -14,8 +14,8 @@ import org.junit.jupiter.api.Test; * Unit tests for {@link SecretMasker}. * *

Assumptions: - Key matching is case-insensitive via the pattern in SENSITIVE. - If the key - * matches a sensitive pattern, the value is replaced with "***REDACTED***". - Nested maps and lists - * are searched recursively. - Null maps and null values are ignored or returned as null. - + * matches a sensitive pattern, the value is replaced with SecretMasker.REDACTED. - Nested maps and + * lists are searched recursively. - Null maps and null values are ignored or returned as null. - * Non-sensitive keys/values remain unchanged. */ class SecretMaskerTest { @@ -40,7 +40,7 @@ class SecretMaskerTest { Map result = SecretMasker.mask(input); - assertEquals("***REDACTED***", result.get("password")); + assertEquals(SecretMasker.REDACTED, result.get("password")); assertEquals("john", result.get("username")); } @@ -55,11 +55,54 @@ class SecretMaskerTest { Map result = SecretMasker.mask(input); - assertEquals("***REDACTED***", result.get("Api-Key")); - assertEquals("***REDACTED***", result.get("TOKEN")); + assertEquals(SecretMasker.REDACTED, result.get("Api-Key")); + assertEquals(SecretMasker.REDACTED, result.get("TOKEN")); assertEquals("keepme", result.get("normal")); } + @Test + @DisplayName("restoreRedacted swaps sentinels for stored values, leaves the rest") + void restoreRedactedRoundTripsAnEdit() { + Map stored = + Map.of("secretAccessKey", "shh", "accessKeyId", "AKIAEXAMPLE"); + Map incoming = + Map.of( + "secretAccessKey", SecretMasker.REDACTED, + "accessKeyId", "AKIA-NEW", + "bucket", "inbox"); + + Map merged = SecretMasker.restoreRedacted(incoming, stored); + + assertEquals("shh", merged.get("secretAccessKey")); + assertEquals("AKIA-NEW", merged.get("accessKeyId")); + assertEquals("inbox", merged.get("bucket")); + } + + @Test + @DisplayName("restoreRedacted leaves a sentinel with no stored counterpart in place") + void restoreRedactedWithoutStoredValueStaysSentinel() { + Map merged = + SecretMasker.restoreRedacted( + Map.of("secretAccessKey", SecretMasker.REDACTED), Map.of()); + + assertEquals(SecretMasker.REDACTED, merged.get("secretAccessKey")); + } + + @Test + @DisplayName("should mask camelCase secretAccessKey despite no word boundary") + void shouldMaskCamelCaseSecretAccessKey() { + Map input = + Map.of( + "secretAccessKey", "shh", + "accessKeyId", "AKIAEXAMPLE"); + + Map result = SecretMasker.mask(input); + + assertEquals(SecretMasker.REDACTED, result.get("secretAccessKey")); + // Access key ids are username-like, not secrets. + assertEquals("AKIAEXAMPLE", result.get("accessKeyId")); + } + @Test @DisplayName("should mask nested map sensitive keys") void shouldMaskNestedMapSensitiveKeys() { @@ -77,9 +120,9 @@ class SecretMaskerTest { Map result = SecretMasker.mask(input); Map outer = (Map) result.get("outer"); - assertEquals("***REDACTED***", outer.get("jwt")); + assertEquals(SecretMasker.REDACTED, outer.get("jwt")); Map inner = (Map) outer.get("inner"); - assertEquals("***REDACTED***", inner.get("secret")); + assertEquals(SecretMasker.REDACTED, inner.get("secret")); assertEquals("ok", inner.get("other")); } @@ -98,7 +141,7 @@ class SecretMaskerTest { List list = (List) result.get("list"); Map first = (Map) list.get(0); - assertEquals("***REDACTED***", first.get("token")); + assertEquals(SecretMasker.REDACTED, first.get("token")); Map second = (Map) list.get(1); assertEquals("john", second.get("username")); assertEquals("stringValue", list.get(2)); @@ -170,7 +213,8 @@ class SecretMaskerTest { Map outer = (Map) result.get("outer"); assertTrue(outer.containsKey(null), "Null key should be preserved"); assertEquals("plainText", outer.get(null), "Value for null key must not be masked"); - assertEquals("***REDACTED***", outer.get("password"), "Sensitive keys must be masked"); + assertEquals( + SecretMasker.REDACTED, outer.get("password"), "Sensitive keys must be masked"); } } } diff --git a/frontend/editor/public/locales/en-US/translation.toml b/frontend/editor/public/locales/en-US/translation.toml index c57174a81e..2cfc42475c 100644 --- a/frontend/editor/public/locales/en-US/translation.toml +++ b/frontend/editor/public/locales/en-US/translation.toml @@ -7246,6 +7246,11 @@ operations_one = "Operation ({{count}})" operations_other = "Operations ({{count}})" output = "Output" removeStep = "Remove operation" +s3Configure = "Configure" +s3Done = "Done" +s3ModalTitle = "Amazon S3 output" +s3NotConfigured = "Not configured" +s3PrefixHelp = "Outputs are uploaded under this key prefix." save = "Save changes" scheduleEvery = "Run every" sources = "Sources" @@ -7265,6 +7270,7 @@ confirm = "Delete" title = "Delete pipeline?" [portal.pipelines.detail] +clearHistory = "Clear history" delete = "Delete pipeline" run = "Run now" @@ -7282,12 +7288,19 @@ total = "Pipelines" [portal.pipelines.output] folder = "Write to folder" inline = "Return files" +s3 = "Write to Amazon S3" [portal.pipelines.run] +allProcessed_one = "Nothing to run: the source's {{count}} document has already been processed." +allProcessed_other = "Nothing to run: all {{count}} documents in the sources have already been processed." completed_one = "Run completed." completed_other = "All {{count}} runs completed." empty = "Nothing to run: the sources had no documents to process." failed = "Run failed: {{error}}" +historyCleared = "History cleared. The next run reprocesses everything currently in the sources." +inFlight = "Nothing new to run: documents are still being processed from an earlier run." +parked_one = "Nothing to run: {{count}} document failed previously and is parked. Fix the cause, then clear history to retry it." +parked_other = "Nothing to run: {{count}} documents failed previously and are parked. Fix the cause, then clear history to retry them." running = "Run started; still in progress." timeout = "Run is taking longer than expected; it may still finish in the background." @@ -7974,6 +7987,42 @@ label = "Folder depth" all = "Include subfolders" top = "Top level only" +[portal.sources.types.s3] +description = "Pull documents from an Amazon S3 or S3-compatible bucket." +label = "Amazon S3" + +[portal.sources.types.s3.fields.accessKeyId] +label = "Access key ID" + +[portal.sources.types.s3.fields.bucket] +label = "Bucket" +placeholder = "my-company-inbox" + +[portal.sources.types.s3.fields.endpoint] +helperText = "Leave blank for Amazon S3. Set to use an S3-compatible service such as MinIO." +label = "Custom endpoint" +placeholder = "https://s3.example.com" + +[portal.sources.types.s3.fields.mode] +helperText = "Consume removes each object from the bucket once every policy has processed it." +label = "Read mode" + +[portal.sources.types.s3.fields.mode.options] +consume = "Consume: process each object once" +snapshot = "Snapshot: re-read the bucket every run" + +[portal.sources.types.s3.fields.prefix] +helperText = "Only objects whose keys start with this prefix are processed." +label = "Key prefix" +placeholder = "incoming/" + +[portal.sources.types.s3.fields.region] +label = "Region" +placeholder = "us-east-1" + +[portal.sources.types.s3.fields.secretAccessKey] +label = "Secret access key" + [portal.sources.types.unknown] label = "Source" diff --git a/frontend/editor/src/portal-saas/components/pipelines/outputModes.test.ts b/frontend/editor/src/portal-saas/components/pipelines/outputModes.test.ts new file mode 100644 index 0000000000..093be2d9da --- /dev/null +++ b/frontend/editor/src/portal-saas/components/pipelines/outputModes.test.ts @@ -0,0 +1,9 @@ +import { describe, expect, it } from "vitest"; +// Resolves to the SaaS override (src/portal-saas) via the @portal cascade. +import { availableOutputModes } from "@portal/components/pipelines/outputModes"; + +describe("availableOutputModes (SaaS)", () => { + it("offers only s3: no server filesystem, and inline results would expire unseen", () => { + expect(availableOutputModes()).toEqual(["s3"]); + }); +}); diff --git a/frontend/editor/src/portal-saas/components/pipelines/outputModes.ts b/frontend/editor/src/portal-saas/components/pipelines/outputModes.ts new file mode 100644 index 0000000000..c55ce3f487 --- /dev/null +++ b/frontend/editor/src/portal-saas/components/pipelines/outputModes.ts @@ -0,0 +1,12 @@ +import type { PipelineOutputMode } from "@portal/api/pipelines"; + +/** + * Hosted deployments never write to the server's filesystem (the backend's + * FolderAccessGuard denies it outright), so folder outputs are not offered in + * the pipeline builder. Inline is not offered either: inline results live in + * transient job storage with no portal download surface, so for an unattended + * pipeline they would simply expire unseen. + */ +export function availableOutputModes(): PipelineOutputMode[] { + return ["s3"]; +} diff --git a/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.test.ts b/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.test.ts new file mode 100644 index 0000000000..1eec2ac959 --- /dev/null +++ b/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.test.ts @@ -0,0 +1,13 @@ +import { describe, expect, it } from "vitest"; +// Resolves to the SaaS override (src/portal-saas) via the @portal cascade. +import { creatableSourceTypes } from "@portal/components/sources/creatableSourceTypes"; + +describe("creatableSourceTypes (SaaS)", () => { + it("never offers folder sources: hosted deployments do not read the server filesystem", () => { + expect(creatableSourceTypes().map((t) => t.type)).not.toContain("folder"); + }); + + it("still offers the cloud source types", () => { + expect(creatableSourceTypes().map((t) => t.type)).toContain("s3"); + }); +}); diff --git a/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.ts b/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.ts new file mode 100644 index 0000000000..9fa73be0ff --- /dev/null +++ b/frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.ts @@ -0,0 +1,13 @@ +import { + CREATABLE_SOURCE_TYPES, + 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. + */ +export function creatableSourceTypes(): CreatableSourceType[] { + return CREATABLE_SOURCE_TYPES.filter((type) => type.type !== "folder"); +} diff --git a/frontend/editor/src/portal/api/pipelines.ts b/frontend/editor/src/portal/api/pipelines.ts index 2fd5b2916e..e75711b912 100644 --- a/frontend/editor/src/portal/api/pipelines.ts +++ b/frontend/editor/src/portal/api/pipelines.ts @@ -29,6 +29,9 @@ export interface OutputSpec { options: Record; } +/** The output destinations the pipeline builder can offer. */ +export type PipelineOutputMode = "inline" | "folder" | "s3"; + /** * The stored policy record: the create/update body (`id` blank on create) and what * the backend returns from GET/POST. Mirrors Policy.java exactly; `owner`/`teamId` @@ -150,12 +153,26 @@ export async function fetchTriggers(): Promise { } /** - * POST /api/v1/policies/{id}/trigger: run the pipeline now against its configured - * sources, regardless of the enabled flag. Returns the ids of the runs started - * (empty when the sources yielded no work); poll {@link fetchRun} for each. + * What a manual trigger found and started. Mirrors the backend `SweepOutcome`: + * when `runIds` is empty, the counts say why - no files listed, all already + * processed at their current version, parked by a failed run, or still in + * flight from an earlier sweep. */ -export async function triggerPipeline(id: string): Promise { - return apiClient.local.json( +export interface TriggerOutcome { + runIds: string[]; + filesListed: number; + alreadyProcessed: number; + parked: number; + inFlight: number; +} + +/** + * POST /api/v1/policies/{id}/trigger: run the pipeline now against its configured + * sources, regardless of the enabled flag. Returns the runs started plus what the + * sweep skipped; poll {@link fetchRun} for each run id. + */ +export async function triggerPipeline(id: string): Promise { + return apiClient.local.json( `/api/v1/policies/${encodeURIComponent(id)}/trigger`, { method: "POST" }, ); diff --git a/frontend/editor/src/portal/components/pipelines/outputModes.ts b/frontend/editor/src/portal/components/pipelines/outputModes.ts new file mode 100644 index 0000000000..a8e698c676 --- /dev/null +++ b/frontend/editor/src/portal/components/pipelines/outputModes.ts @@ -0,0 +1,11 @@ +import type { PipelineOutputMode } from "@portal/api/pipelines"; + +/** + * The output destinations the pipeline builder offers. An extension point: + * deployments where a destination cannot work shadow this module and filter + * the list (e.g. hosted deployments never write to the server's filesystem, + * so folder outputs are not offered there). + */ +export function availableOutputModes(): PipelineOutputMode[] { + return ["inline", "folder", "s3"]; +} diff --git a/frontend/editor/src/portal/components/sources/ConnectWizard.test.tsx b/frontend/editor/src/portal/components/sources/ConnectWizard.test.tsx index 2ae0d9fdf3..7723b447ca 100644 --- a/frontend/editor/src/portal/components/sources/ConnectWizard.test.tsx +++ b/frontend/editor/src/portal/components/sources/ConnectWizard.test.tsx @@ -74,6 +74,53 @@ describe("ConnectWizard", () => { }); }); + it("creates an s3 source with a masked secret in review", async () => { + createSource.mockResolvedValue({ id: "src-2" }); + + renderWithMantine( + , + ); + + // Step 0: pick the S3 type card. + fireEvent.click(screen.getByText("portal.sources.types.s3.label")); + fireEvent.click(screen.getByText("portal.sources.wizard.continue")); + + // Step 1: name, bucket, credentials. The secret renders as a password + // input, so it is not part of the textbox roles. + const inputs = screen.getAllByRole("textbox") as HTMLInputElement[]; + fireEvent.change(inputs[0], { target: { value: "Claims bucket" } }); + fireEvent.change(inputs[1], { target: { value: "claims-inbox" } }); + fireEvent.change(inputs[4], { target: { value: "AKIAEXAMPLE" } }); + const secret = document.querySelector( + 'input[type="password"]', + ) as HTMLInputElement; + fireEvent.change(secret, { target: { value: "shh-secret" } }); + fireEvent.click(screen.getByText("portal.sources.wizard.continue")); + + // Step 2: the secret is masked in review, never echoed. + expect(screen.queryByText("shh-secret")).not.toBeInTheDocument(); + expect(screen.getByText("********")).toBeInTheDocument(); + + fireEvent.click(screen.getByText("portal.sources.actions.connectSource")); + await waitFor(() => { + expect(createSource).toHaveBeenCalledTimes(1); + }); + expect(createSource).toHaveBeenCalledWith({ + name: "Claims bucket", + type: "s3", + options: { + bucket: "claims-inbox", + region: "us-east-1", + prefix: "", + accessKeyId: "AKIAEXAMPLE", + secretAccessKey: "shh-secret", + endpoint: "", + mode: "consume", + }, + enabled: true, + }); + }); + it("edits an existing source: prefilled, skips type, submits with its id", async () => { createSource.mockResolvedValue({ id: "s1" }); const onCreated = vi.fn(); diff --git a/frontend/editor/src/portal/components/sources/ConnectWizard.tsx b/frontend/editor/src/portal/components/sources/ConnectWizard.tsx index afb3982545..c284c24655 100644 --- a/frontend/editor/src/portal/components/sources/ConnectWizard.tsx +++ b/frontend/editor/src/portal/components/sources/ConnectWizard.tsx @@ -11,15 +11,16 @@ import { } from "@app/ui"; import { errorMessage } from "@portal/api/http"; import { createSource, type Source } from "@portal/api/sources"; +import { creatableSourceTypes } from "@portal/components/sources/creatableSourceTypes"; import { - CREATABLE_SOURCE_TYPES, defaultOptions, sourceTypeMeta, type CreatableSourceType, } from "@portal/components/sources/sourceTypes"; import "@portal/views/Sources.css"; -const DEFAULT_TYPE = CREATABLE_SOURCE_TYPES[0]; +const OFFERED_TYPES = creatableSourceTypes(); +const DEFAULT_TYPE = OFFERED_TYPES[0]; /** Wizard steps. Editing skips type selection (the type is fixed once created). */ type StepId = "type" | "configure" | "review"; @@ -35,9 +36,9 @@ interface ConnectWizardProps { source?: Source; } -/** The creatable-type metadata for a source's stored type, falling back to folder. */ +/** The creatable-type metadata for a source's stored type, falling back to the first offered. */ function typeFor(type: string | undefined): CreatableSourceType { - return CREATABLE_SOURCE_TYPES.find((t) => t.type === type) ?? DEFAULT_TYPE; + return OFFERED_TYPES.find((t) => t.type === type) ?? DEFAULT_TYPE; } /** Source options coerced to strings for the form, defaulted from the type's fields. */ @@ -199,7 +200,7 @@ export function ConnectWizard({ {stepId === "type" && (

- {CREATABLE_SOURCE_TYPES.map((ct) => ( + {OFFERED_TYPES.map((ct) => (
diff --git a/frontend/editor/src/portal/components/sources/creatableSourceTypes.ts b/frontend/editor/src/portal/components/sources/creatableSourceTypes.ts new file mode 100644 index 0000000000..5324a90e33 --- /dev/null +++ b/frontend/editor/src/portal/components/sources/creatableSourceTypes.ts @@ -0,0 +1,14 @@ +import { + CREATABLE_SOURCE_TYPES, + type CreatableSourceType, +} from "@portal/components/sources/sourceTypes"; + +/** + * The source types the connect wizard offers. An extension point: deployments + * where a type cannot work shadow this module and filter the list (e.g. hosted + * deployments never read the server's filesystem, so folder sources are not + * offered there). + */ +export function creatableSourceTypes(): CreatableSourceType[] { + return CREATABLE_SOURCE_TYPES; +} diff --git a/frontend/editor/src/portal/components/sources/sourceTypes.ts b/frontend/editor/src/portal/components/sources/sourceTypes.ts index dcae138670..fb6057a183 100644 --- a/frontend/editor/src/portal/components/sources/sourceTypes.ts +++ b/frontend/editor/src/portal/components/sources/sourceTypes.ts @@ -26,6 +26,11 @@ const SOURCE_TYPE_META: Record = { icon: "✏", accent: "success", }, + s3: { + labelKey: "portal.sources.types.s3.label", + icon: "☁", + accent: "brand", + }, }; const UNKNOWN_TYPE_META: SourceTypeMeta = { @@ -42,7 +47,7 @@ export function sourceTypeMeta(type: string): SourceTypeMeta { export interface SourceFieldDef { key: string; labelKey: string; - control: "text" | "select"; + control: "text" | "password" | "select"; required?: boolean; placeholderKey?: string; helperTextKey?: string; @@ -130,6 +135,70 @@ export const CREATABLE_SOURCE_TYPES: CreatableSourceType[] = [ }, ], }, + { + type: "s3", + labelKey: "portal.sources.types.s3.label", + descriptionKey: "portal.sources.types.s3.description", + fields: [ + { + key: "bucket", + labelKey: "portal.sources.types.s3.fields.bucket.label", + control: "text", + required: true, + placeholderKey: "portal.sources.types.s3.fields.bucket.placeholder", + }, + { + key: "region", + labelKey: "portal.sources.types.s3.fields.region.label", + control: "text", + defaultValue: "us-east-1", + placeholderKey: "portal.sources.types.s3.fields.region.placeholder", + }, + { + key: "prefix", + labelKey: "portal.sources.types.s3.fields.prefix.label", + control: "text", + placeholderKey: "portal.sources.types.s3.fields.prefix.placeholder", + helperTextKey: "portal.sources.types.s3.fields.prefix.helperText", + }, + { + key: "accessKeyId", + labelKey: "portal.sources.types.s3.fields.accessKeyId.label", + control: "text", + required: true, + }, + { + key: "secretAccessKey", + labelKey: "portal.sources.types.s3.fields.secretAccessKey.label", + control: "password", + required: true, + }, + { + key: "endpoint", + labelKey: "portal.sources.types.s3.fields.endpoint.label", + control: "text", + placeholderKey: "portal.sources.types.s3.fields.endpoint.placeholder", + helperTextKey: "portal.sources.types.s3.fields.endpoint.helperText", + }, + { + key: "mode", + labelKey: "portal.sources.types.s3.fields.mode.label", + control: "select", + defaultValue: "consume", + helperTextKey: "portal.sources.types.s3.fields.mode.helperText", + options: [ + { + value: "consume", + labelKey: "portal.sources.types.s3.fields.mode.options.consume", + }, + { + value: "snapshot", + labelKey: "portal.sources.types.s3.fields.mode.options.snapshot", + }, + ], + }, + ], + }, ]; /** Default option values for a type's create form. */ diff --git a/frontend/editor/src/portal/mocks/handlers/pipelines.ts b/frontend/editor/src/portal/mocks/handlers/pipelines.ts index 6bce6db31a..4c18da8b6e 100644 --- a/frontend/editor/src/portal/mocks/handlers/pipelines.ts +++ b/frontend/editor/src/portal/mocks/handlers/pipelines.ts @@ -165,7 +165,13 @@ export const pipelinesHandlers = [ http.post("/api/v1/policies/:id/trigger", async ({ params }) => { if (!store.some((p) => p.id === params.id)) return undefined; await delay(120); - return HttpResponse.json([`run_${Date.now().toString(36)}`]); + return HttpResponse.json({ + runIds: [`run_${Date.now().toString(36)}`], + filesListed: 1, + alreadyProcessed: 0, + parked: 0, + inFlight: 0, + }); }), // Raw policy by id. Only our pipeline ids are served here; everything else falls diff --git a/frontend/editor/src/portal/views/PipelineBuilder.css b/frontend/editor/src/portal/views/PipelineBuilder.css index cef65d6e75..8fb8095b32 100644 --- a/frontend/editor/src/portal/views/PipelineBuilder.css +++ b/frontend/editor/src/portal/views/PipelineBuilder.css @@ -392,3 +392,27 @@ min-width: 1.5rem; min-height: 1.5rem; } + +.portal-builder__s3-output { + display: flex; + align-items: center; + gap: 0.625rem; +} + +.portal-builder__s3-summary { + font-size: 0.8125rem; + color: var(--color-text-1); + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.portal-builder__s3-summary.is-unset { + color: var(--color-text-4); +} + +.portal-builder__s3-fields { + display: flex; + flex-direction: column; + gap: 0.75rem; +} diff --git a/frontend/editor/src/portal/views/PipelineBuilder.test.tsx b/frontend/editor/src/portal/views/PipelineBuilder.test.tsx index 3474874dad..f47df5c5ca 100644 --- a/frontend/editor/src/portal/views/PipelineBuilder.test.tsx +++ b/frontend/editor/src/portal/views/PipelineBuilder.test.tsx @@ -7,7 +7,7 @@ import { } from "@testing-library/react"; import { MantineProvider } from "@mantine/core"; import { MemoryRouter, Route, Routes } from "react-router-dom"; -import type { Policy } from "@portal/api/pipelines"; +import type { Policy, TriggerOutcome } from "@portal/api/pipelines"; import type { ToolRegistryCatalog } from "@app/contexts/ToolRegistryContext"; import type { ToolRegistryEntry } from "@app/data/toolsTaxonomy"; import { PipelineBuilder } from "@portal/views/PipelineBuilder"; @@ -45,6 +45,11 @@ vi.mock("@portal/api/sources", () => ({ fetchSources: () => fetchSources(), })); +const clearProcessedHistory = vi.fn(); +vi.mock("@portal/api/policies", () => ({ + clearProcessedHistory: (id: string) => clearProcessedHistory(id), +})); + // One editable tool, Compress, so the picker and step settings have something to render. vi.mock("@app/contexts/ToolRegistryContext", () => { const compress = { @@ -100,6 +105,17 @@ const POLICY: Policy = { output: { type: "inline", options: {} }, }; +function outcome(overrides: Partial): TriggerOutcome { + return { + runIds: [], + filesListed: 0, + alreadyProcessed: 0, + parked: 0, + inFlight: 0, + ...overrides, + }; +} + function renderBuilder(initial: string) { return render( @@ -129,8 +145,10 @@ describe("PipelineBuilder", () => { fetchSources.mockResolvedValue({ kpis: [], sources: [] }); savePipeline.mockResolvedValue({}); deletePipeline.mockResolvedValue(undefined); - triggerPipeline.mockResolvedValue(["run-1"]); + triggerPipeline.mockResolvedValue(outcome({ runIds: ["run-1"] })); fetchRun.mockResolvedValue({ status: "COMPLETED" }); + clearProcessedHistory.mockReset(); + clearProcessedHistory.mockResolvedValue(undefined); }); it("builds a new pipeline: name it, add a tool, and save", async () => { @@ -159,6 +177,60 @@ describe("PipelineBuilder", () => { expect(await screen.findByText("pipelines list")).toBeInTheDocument(); }); + it("saves an s3 output with its connection options", async () => { + renderBuilder("/processor/pipelines/new"); + + fireEvent.change(await screen.findByRole("textbox"), { + target: { value: "Bucket to bucket" }, + }); + fireEvent.click(screen.getByLabelText("portal.pipelines.output.s3")); + + // With s3 selected but no bucket, saving is blocked and the summary reads + // unconfigured; the connection fields live behind the Configure modal. + expect( + screen.getByText("portal.pipelines.composer.create").closest("button"), + ).toBeDisabled(); + expect( + screen.getByText("portal.pipelines.composer.s3NotConfigured"), + ).toBeInTheDocument(); + fireEvent.click(screen.getByText("portal.pipelines.composer.s3Configure")); + + // Textboxes: name, then the modal's bucket, region, prefix, access key id, + // endpoint; the secret renders as a password input outside the textbox role. + const inputs = screen.getAllByRole("textbox") as HTMLInputElement[]; + fireEvent.change(inputs[1], { target: { value: "claims-processed" } }); + fireEvent.change(inputs[3], { target: { value: "processed/" } }); + fireEvent.change(inputs[4], { target: { value: "AKIAEXAMPLE" } }); + const secret = document.querySelector( + 'input[type="password"]', + ) as HTMLInputElement; + fireEvent.change(secret, { target: { value: "shh-secret" } }); + fireEvent.click(screen.getByText("portal.pipelines.composer.s3Done")); + + // The summary now shows the configured destination. + expect( + screen.getByText("s3://claims-processed/processed/"), + ).toBeInTheDocument(); + fireEvent.click(screen.getByText("portal.pipelines.composer.create")); + + await waitFor(() => expect(savePipeline).toHaveBeenCalledTimes(1)); + expect(savePipeline).toHaveBeenCalledWith( + expect.objectContaining({ + output: { + type: "s3", + options: { + bucket: "claims-processed", + region: "us-east-1", + prefix: "processed/", + endpoint: "", + accessKeyId: "AKIAEXAMPLE", + secretAccessKey: "shh-secret", + }, + }, + }), + ); + }); + it("runs an existing pipeline and reports success", async () => { renderBuilder("/processor/pipelines/plc-1"); @@ -170,6 +242,45 @@ describe("PipelineBuilder", () => { ).toBeInTheDocument(); }); + it("explains an empty trigger when files are parked by a failed run", async () => { + triggerPipeline.mockResolvedValue(outcome({ filesListed: 2, parked: 2 })); + renderBuilder("/processor/pipelines/plc-1"); + + fireEvent.click(await screen.findByText("portal.pipelines.detail.run")); + + expect( + await screen.findByText("portal.pipelines.run.parked"), + ).toBeInTheDocument(); + }); + + it("explains an empty trigger when everything is already processed", async () => { + triggerPipeline.mockResolvedValue( + outcome({ filesListed: 3, alreadyProcessed: 3 }), + ); + renderBuilder("/processor/pipelines/plc-1"); + + fireEvent.click(await screen.findByText("portal.pipelines.detail.run")); + + expect( + await screen.findByText("portal.pipelines.run.allProcessed"), + ).toBeInTheDocument(); + }); + + it("clears processed history from the header and confirms", async () => { + renderBuilder("/processor/pipelines/plc-1"); + + fireEvent.click( + await screen.findByText("portal.pipelines.detail.clearHistory"), + ); + + await waitFor(() => + expect(clearProcessedHistory).toHaveBeenCalledWith("plc-1"), + ); + expect( + await screen.findByText("portal.pipelines.run.historyCleared"), + ).toBeInTheDocument(); + }); + it("blocks saving a step that needs an uploaded file", async () => { renderBuilder("/processor/pipelines/new"); diff --git a/frontend/editor/src/portal/views/PipelineBuilder.tsx b/frontend/editor/src/portal/views/PipelineBuilder.tsx index cbd4ba4b45..e82f0b380c 100644 --- a/frontend/editor/src/portal/views/PipelineBuilder.tsx +++ b/frontend/editor/src/portal/views/PipelineBuilder.tsx @@ -5,6 +5,7 @@ import ArrowBackRoundedIcon from "@mui/icons-material/ArrowBackRounded"; import KeyboardArrowUpRoundedIcon from "@mui/icons-material/KeyboardArrowUpRounded"; import KeyboardArrowDownRoundedIcon from "@mui/icons-material/KeyboardArrowDownRounded"; import DeleteOutlineRoundedIcon from "@mui/icons-material/DeleteOutlineRounded"; +import HistoryRoundedIcon from "@mui/icons-material/HistoryRounded"; import AddRoundedIcon from "@mui/icons-material/AddRounded"; import PlayArrowRoundedIcon from "@mui/icons-material/PlayArrowRounded"; import { @@ -42,9 +43,13 @@ import { type OutputSpec, type Policy, type PolicyRunView, + type PipelineOutputMode, type TriggerConfig, type TriggerInfo, + type TriggerOutcome, } from "@portal/api/pipelines"; +import { clearProcessedHistory } from "@portal/api/policies"; +import { availableOutputModes } from "@portal/components/pipelines/outputModes"; import { fetchSources, type SourceView } from "@portal/api/sources"; import { useAsync } from "@portal/hooks/useAsync"; import { VIEW_PATHS, toPortalPath } from "@portal/contexts/ViewContext"; @@ -53,7 +58,29 @@ import { PipelineStepSettings } from "@portal/components/pipelines/PipelineStepS import { ToolPicker } from "@portal/components/pipelines/ToolPicker"; import "@portal/views/PipelineBuilder.css"; -type OutputMode = "inline" | "folder"; +type OutputMode = PipelineOutputMode; + +/** New pipelines (and specs of unoffered types) start on the first offered destination. */ +const DEFAULT_OUTPUT_MODE = availableOutputModes()[0]; + +/** The s3 output's connection fields, mirrored from the OutputSpec options. */ +interface S3OutputOptions { + bucket: string; + region: string; + prefix: string; + endpoint: string; + accessKeyId: string; + secretAccessKey: string; +} + +const EMPTY_S3_OUTPUT: S3OutputOptions = { + bucket: "", + region: "us-east-1", + prefix: "", + endpoint: "", + accessKeyId: "", + secretAccessKey: "", +}; type ScheduleUnit = "MINUTES" | "HOURS" | "DAYS"; const SCHEDULE_UNITS: ScheduleUnit[] = ["MINUTES", "HOURS", "DAYS"]; @@ -95,14 +122,32 @@ function parseTrigger(trigger: TriggerConfig | null): { function parseOutput(output: OutputSpec | undefined): { mode: OutputMode; directory: string; + s3: S3OutputOptions; } { if (output?.type === "folder") { return { mode: "folder", directory: String(output.options?.directory ?? ""), + s3: EMPTY_S3_OUTPUT, }; } - return { mode: "inline", directory: "" }; + if (output?.type === "s3") { + const option = (key: keyof S3OutputOptions, fallback = "") => + String(output.options?.[key] ?? fallback); + return { + mode: "s3", + directory: "", + s3: { + bucket: option("bucket"), + region: option("region", "us-east-1"), + prefix: option("prefix"), + endpoint: option("endpoint"), + accessKeyId: option("accessKeyId"), + secretAccessKey: option("secretAccessKey"), + }, + }; + } + return { mode: DEFAULT_OUTPUT_MODE, directory: "", s3: EMPTY_S3_OUTPUT }; } /** @@ -149,12 +194,15 @@ export function PipelineBuilder() { const [triggerType, setTriggerType] = useState(MANUAL); const [scheduleCount, setScheduleCount] = useState("1"); const [scheduleUnit, setScheduleUnit] = useState("HOURS"); - const [outputMode, setOutputMode] = useState("inline"); + const [outputMode, setOutputMode] = useState(DEFAULT_OUTPUT_MODE); const [outputDirectory, setOutputDirectory] = useState(""); + const [outputS3, setOutputS3] = useState(EMPTY_S3_OUTPUT); + const [s3ConfigOpen, setS3ConfigOpen] = useState(false); const [submitting, setSubmitting] = useState(false); const [error, setError] = useState(null); const [seeded, setSeeded] = useState(false); const [running, setRunning] = useState(false); + const [clearingHistory, setClearingHistory] = useState(false); const [runResult, setRunResult] = useState(null); const [pendingDelete, setPendingDelete] = useState(false); const [deleting, setDeleting] = useState(false); @@ -186,6 +234,7 @@ export function PipelineBuilder() { setScheduleUnit(trigger.unit); setOutputMode(output.mode); setOutputDirectory(output.directory); + setOutputS3(output.s3); setSeeded(true); }, [isEdit, policyState.data, allTools, seeded]); @@ -222,6 +271,10 @@ export function PipelineBuilder() { if (selected && !triggerAvailable(selected)) setTriggerType(MANUAL); }, [triggerType, triggers, triggerAvailable]); + function setS3Field(key: keyof S3OutputOptions, value: string) { + setOutputS3((current) => ({ ...current, [key]: value })); + } + function toggleSource(sourceId: string, checked: boolean) { setSourceIds((ids) => checked @@ -286,6 +339,7 @@ export function PipelineBuilder() { scheduleUnit, outputMode, outputDirectory, + outputS3, }); const baseline = useRef(null); useEffect(() => { @@ -295,7 +349,13 @@ export function PipelineBuilder() { const scheduleCountValid = triggerType !== "schedule" || Number(scheduleCount) > 0; - const outputValid = outputMode !== "folder" || outputDirectory.trim() !== ""; + const s3OutputValid = + outputMode !== "s3" || + (outputS3.bucket.trim() !== "" && + outputS3.accessKeyId.trim() !== "" && + outputS3.secretAccessKey.trim() !== ""); + const outputValid = + (outputMode !== "folder" || outputDirectory.trim() !== "") && s3OutputValid; const canSave = name.trim() !== "" && scheduleCountValid && @@ -357,7 +417,9 @@ export function PipelineBuilder() { const output: OutputSpec = outputMode === "folder" ? { type: "folder", options: { directory: outputDirectory.trim() } } - : { type: "inline", options: {} }; + : outputMode === "s3" + ? { type: "s3", options: { ...outputS3 } } + : { type: "inline", options: {} }; const policy: Policy = { id: policyState.data?.id ?? undefined, name: name.trim(), @@ -387,15 +449,37 @@ export function PipelineBuilder() { return null; } + /** Explain an empty trigger: parked files outrank blander reasons. */ + function emptySweepResult(outcome: TriggerOutcome): RunResult { + if (outcome.parked > 0) { + return { + tone: "warning", + text: t("portal.pipelines.run.parked", { count: outcome.parked }), + }; + } + if (outcome.inFlight > 0) { + return { tone: "info", text: t("portal.pipelines.run.inFlight") }; + } + if (outcome.alreadyProcessed > 0) { + return { + tone: "info", + text: t("portal.pipelines.run.allProcessed", { + count: outcome.alreadyProcessed, + }), + }; + } + return { tone: "info", text: t("portal.pipelines.run.empty") }; + } + async function handleRun() { if (running || !id) return; setRunning(true); setRunResult(null); try { - const runIds = await triggerPipeline(id); + const outcome = await triggerPipeline(id); + const runIds = outcome.runIds; if (runIds.length === 0) { - if (mounted.current) - setRunResult({ tone: "info", text: t("portal.pipelines.run.empty") }); + if (mounted.current) setRunResult(emptySweepResult(outcome)); return; } const finals = await Promise.all(runIds.map((runId) => awaitRun(runId))); @@ -428,6 +512,30 @@ export function PipelineBuilder() { } } + /** + * Forget which source files this pipeline has processed, so the next sweep + * reprocesses everything currently in its sources (the standard retry for a + * parked-by-failure file). Does not touch the files themselves. + */ + async function handleClearHistory() { + if (clearingHistory || !id) return; + setClearingHistory(true); + setRunResult(null); + try { + await clearProcessedHistory(id); + if (mounted.current) + setRunResult({ + tone: "success", + text: t("portal.pipelines.run.historyCleared"), + }); + } catch (e) { + if (mounted.current) + setRunResult({ tone: "danger", text: errorMessage(e) }); + } finally { + if (mounted.current) setClearingHistory(false); + } + } + async function confirmDelete() { if (!id || deleting) return; setDeleting(true); @@ -494,6 +602,17 @@ export function PipelineBuilder() { > {t("portal.pipelines.detail.run")} + + + )} @@ -860,6 +1000,78 @@ export function PipelineBuilder() { >

{t("portal.pipelines.builder.unsavedBody")}

+ + setS3ConfigOpen(false)} + title={t("portal.pipelines.composer.s3ModalTitle")} + footer={ +
+ +
+ } + > +
+ + setS3Field("bucket", e.target.value)} + /> + + + setS3Field("region", e.target.value)} + /> + + + setS3Field("prefix", e.target.value)} + /> + + + setS3Field("accessKeyId", e.target.value)} + /> + + + setS3Field("secretAccessKey", e.target.value)} + /> + + + setS3Field("endpoint", e.target.value)} + /> + +
+ ); }