Skip to content

fix: Fix SSC SAST-DAST correlation state handling #1105

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rsenden merged 3 commits into from
Sep 23, 2026
Merged

fix: Fix SSC SAST-DAST correlation state handling #1105

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6501f4b
Fix SSC SAST-DAST correlation state handling
AnkitRathod29 Sep 23, 2026
f681051
Report skipped correlation reasons
AnkitRathod29 Sep 23, 2026
b038fbf
Address SAST-DAST correlation review feedback
AnkitRathod29 Sep 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 23 additions & 5 deletions ...ore/fcli-aviator-common/src/main/java/com/fortify/cli/aviator/grpc/CorrelationResult.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,17 +14,35 @@

import java.util.List;

import lombok.Builder;

/**
* Holds the outcome of a full correlation stream run — both confirmed
* and rejected SAST–DAST pairs, plus the count of correlation requests
* that received a successful response from the server.
* and rejected SAST–DAST pairs, plus request statistics for the correlation phase.
*
* @param confirmedPairs pairs where Phase 2 validation returned confirmed=true
* @param rejectedPairs pairs where Phase 2 validation returned confirmed=false
* @param receivedCorrelationResponses number of Phase 1 correlation requests that received a response
* @param submittedCorrelationRequests number of Phase 1 correlation requests sent to the server
* @param successfulCorrelationResponses number of Phase 1 requests with a successful server response
* @param skippedCorrelationResponses number of Phase 1 requests skipped by the server
* @param failedCorrelationResponses number of Phase 1 requests that failed
*/
@Builder
public record CorrelationResult(
List<CorrelatedPair> confirmedPairs,
List<CorrelatedPair> rejectedPairs,
int receivedCorrelationResponses
) {}
int submittedCorrelationRequests,

@rsenden rsenden Sep 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Consider using Lombok @Builder to avoid potential incorrect order of the multiple int constructor args.

@ankit2995 ankit2995 Sep 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added Lombok @builder to CorrelationResult and replaced positional constructors.

int successfulCorrelationResponses,
int skippedCorrelationResponses,
int failedCorrelationResponses
) {
public CorrelationResult {
confirmedPairs = confirmedPairs == null ? List.of() : List.copyOf(confirmedPairs);
rejectedPairs = rejectedPairs == null ? List.of() : List.copyOf(rejectedPairs);
}

public static CorrelationResult empty() {
return CorrelationResult.builder()
.build();
}
}
80 changes: 45 additions & 35 deletions ...aviator-common/src/main/java/com/fortify/cli/aviator/grpc/CorrelationStreamProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
Expand Down Expand Up @@ -94,16 +95,15 @@ public class CorrelationStreamProcessor implements AutoCloseable {
// Input data retained for building validation requests
private List<CorrelationWorkItem> correlationWorkItems;
private Map<String, List<DastIssue>> urlToDastIssues;
private final java.util.concurrent.ConcurrentHashMap<String, String> validationRequestToDastId =
new java.util.concurrent.ConcurrentHashMap<>();
private final Set<String> pendingCorrelationRequestIds = ConcurrentHashMap.newKeySet();
private final ConcurrentHashMap<String, String> validationRequestToDastId = new ConcurrentHashMap<>();
private volatile CompletableFuture<CorrelationResult> resultFuture;

/**
* Keys of SAST–DAST pairs that were confirmed in a previous run and should
* be skipped during both Phase 1 (correlation) and Phase 2 (validation).
* Keys of SAST–DAST pairs that were confirmed or rejected in a previous run.
* Each key is formatted as {@code "sastInstanceId::dastIssueId"}.
*/
private Set<String> previouslyCorrelatedPairKeys = Set.of();
private Set<String> previouslyTriedPairKeys = Set.of();

public CorrelationStreamProcessor(
AviatorGrpcClient client,
Expand All @@ -122,23 +122,25 @@ public CorrelationStreamProcessor(

/**
* Entry point: run correlation on the provided mixed-category buckets.
* Previously confirmed pairs (from prior runs) are not re-processed.
* Previously tried pairs (from prior runs) are not re-processed.
*
* @param config stream init configuration (token, app name, etc.)
* @param mixedBuckets category buckets containing both SAST and DAST findings
* @param scanGuid SAST scan UUID for building CorrelatedPair results
* @param previouslyCorrelatedPairKeys keys of already-confirmed pairs to skip, each formatted as
* {@code "sastInstanceId::dastIssueId"}; may be {@code null}
* @param previouslyTriedPairKeys keys of already-tried pairs to skip, each formatted as
* {@code "sastInstanceId::dastIssueId"}; may be {@code null}
* @return future that completes with the list of confirmed correlated pairs
*/
public CompletableFuture<CorrelationResult> processCorrelation(
CorrelationStreamConfig config,
List<? extends Object> mixedBuckets,
String scanGuid,
Set<String> previouslyCorrelatedPairKeys) {
Set<String> previouslyTriedPairKeys) {

this.previouslyCorrelatedPairKeys =
previouslyCorrelatedPairKeys != null ? previouslyCorrelatedPairKeys : Set.of();
this.previouslyTriedPairKeys =
previouslyTriedPairKeys != null ? previouslyTriedPairKeys : Set.of();
pendingCorrelationRequestIds.clear();
validationRequestToDastId.clear();

// Build URL→DAST map first; needed to evaluate Phase 1 skip eligibility
this.urlToDastIssues = buildUrlToDastMap(mixedBuckets);
Expand All @@ -149,7 +151,7 @@ public CompletableFuture<CorrelationResult> processCorrelation(
urlToDastIssues.forEach((k,v)->LOG.debug(" For url {} no. of dast issues {}", k, v.size()));
if (workItems.isEmpty()) {
LOG.info("No SAST findings in mixed buckets; skipping correlation stream.");
return CompletableFuture.completedFuture(new CorrelationResult(List.of(), List.of(), 0));
return CompletableFuture.completedFuture(CorrelationResult.empty());
}

String streamId = UUID.randomUUID().toString();
Expand All @@ -167,6 +169,17 @@ public CompletableFuture<CorrelationResult> processCorrelation(
return resultFuture;
}

private CorrelationResult createResultSnapshot() {
return CorrelationResult.builder()
.confirmedPairs(new ArrayList<>(state.confirmedPairs))
.rejectedPairs(new ArrayList<>(state.rejectedPairs))
.submittedCorrelationRequests(state.totalCorrelationRequests)
.successfulCorrelationResponses(state.successfulCorrelations.get())
.skippedCorrelationResponses(state.skippedCorrelations.get())
.failedCorrelationResponses(state.failedCorrelations.get())
.build();
}

/**
* Convenience overload for callers that have no previously confirmed pairs to skip.
*/
Expand Down Expand Up @@ -220,6 +233,7 @@ private void sendCorrelationRequests() {

for (var item : correlationWorkItems) {
var req = buildCorrelationRequest(state.streamId, item);
pendingCorrelationRequestIds.add(req.getRequestId());
requestHandler.sendRequest(
CorrelationClientMessage.newBuilder().setCorrelation(req).build()
);
Expand All @@ -238,11 +252,7 @@ private void transitionToValidation(String scanGuid) {
if (validationItems.isEmpty()) {
logger.info("No candidates to validate. Completing stream.");
if (!resultFuture.isDone()) {
resultFuture.complete(new CorrelationResult(
new ArrayList<>(state.confirmedPairs),
new ArrayList<>(state.rejectedPairs),
state.successfulCorrelations.get()
));
resultFuture.complete(createResultSnapshot());
}
requestHandler.complete();
streamLatch.countDown();
Expand Down Expand Up @@ -312,11 +322,7 @@ public void onCompleted() {
state.currentPhase = CorrelationStreamState.Phase.COMPLETE;
logger.info("Correlation stream completed — " + state.confirmedPairs.size() + " confirmed pairs");
if (!resultFuture.isDone()) {
resultFuture.complete(new CorrelationResult(
new ArrayList<>(state.confirmedPairs),
new ArrayList<>(state.rejectedPairs),
state.successfulCorrelations.get()
));
resultFuture.complete(createResultSnapshot());
}
streamLatch.countDown();
}
Expand All @@ -340,6 +346,10 @@ private void handleInitResponse(CorrelationInitResponse resp) {
}

private void handleCorrelationResponse(CorrelationResponse resp) {
if (!pendingCorrelationRequestIds.remove(resp.getRequestId())) {
LOG.warn("Ignoring correlation response for unknown or completed request {}", resp.getRequestId());
return;
}
int received = state.receivedCorrelations.incrementAndGet();
LOG.debug("Correlation response {}/{} for SAST {}: status={}",
received, state.totalCorrelationRequests, resp.getSastId(), resp.getStatus());
Expand All @@ -356,7 +366,11 @@ private void handleCorrelationResponse(CorrelationResponse resp) {
match.getRationale()
));
}
} else if ("SKIPPED".equalsIgnoreCase(resp.getStatus())) {
state.skippedCorrelations.incrementAndGet();
LOG.debug("Skipped correlation for SAST {}: {}", resp.getSastId(), resp.getNoCorrelationReason());
} else {
state.failedCorrelations.incrementAndGet();
LOG.debug("Non-OK correlation for SAST {}: {} — {}",
resp.getSastId(), resp.getStatus(), resp.getNoCorrelationReason());
}
Expand Down Expand Up @@ -408,11 +422,7 @@ private void handleValidationResponse(CorrelationValidationResponse resp, String
state.confirmedPairs.size() + " confirmed pairs, " +
state.rejectedPairs.size() + " rejected pairs.");
if (!resultFuture.isDone()) {
resultFuture.complete(new CorrelationResult(
new ArrayList<>(state.confirmedPairs),
new ArrayList<>(state.rejectedPairs),
state.successfulCorrelations.get()
));
resultFuture.complete(createResultSnapshot());
}
requestHandler.complete();
streamLatch.countDown();
Expand Down Expand Up @@ -568,10 +578,10 @@ private List<CorrelationWorkItem> buildCorrelationWorkItems(List<? extends Objec
if (dastUrls.isEmpty()) continue;
List<String> urlList = new ArrayList<>(dastUrls);
for (Vulnerability vuln : data.sastFindings()) {
// Strip URLs where every mapped DAST issue is already confirmed with this SAST finding
// Strip URLs where every mapped DAST issue was already tried with this SAST finding.
List<String> newUrls = filterNewUrls(vuln.getInstanceID(), urlList);
if (newUrls.isEmpty()) {
LOG.debug("Skipping SAST finding {} from Phase 1 — all reachable DAST issues already confirmed",
LOG.debug("Skipping SAST finding {} from Phase 1 — all reachable DAST issues already tried",
vuln.getInstanceID());
continue;
}
Expand All @@ -585,16 +595,16 @@ private List<CorrelationWorkItem> buildCorrelationWorkItems(List<? extends Objec

/**
* Returns the subset of {@code urls} for which at least one mapped DAST issue
* is NOT yet confirmed with {@code sastInstanceId}.
* has NOT yet been tried with {@code sastInstanceId}.
*
* <ul>
* <li>URLs with no mapped DAST issues are kept (the server may resolve them).</li>
* <li>URLs where every mapped DAST issue is already in
* {@link #previouslyCorrelatedPairKeys} are excluded — they add no new work.</li>
* {@link #previouslyTriedPairKeys} are excluded — they add no new work.</li>
* </ul>
*/
private List<String> filterNewUrls(String sastInstanceId, List<String> urls) {
if (previouslyCorrelatedPairKeys.isEmpty()) return urls; // fast path: nothing confirmed yet
if (previouslyTriedPairKeys.isEmpty()) return urls;
List<String> result = new ArrayList<>();
for (String url : urls) {
List<DastIssue> issues = urlToDastIssues.getOrDefault(url, List.of());
Expand All @@ -604,7 +614,7 @@ private List<String> filterNewUrls(String sastInstanceId, List<String> urls) {
}
boolean hasUncorrelated = issues.stream()
.filter(d -> d.getId() != null && !d.getId().isEmpty())
.anyMatch(d -> !previouslyCorrelatedPairKeys.contains(sastInstanceId + "::" + d.getId()));
.anyMatch(d -> !previouslyTriedPairKeys.contains(sastInstanceId + "::" + d.getId()));
if (hasUncorrelated) {
result.add(url);
}
Expand Down Expand Up @@ -645,8 +655,8 @@ private List<ValidationWorkItem> buildValidationWorkItems() {

for (DastIssue dastIssue : issues) {
String pairKey = match.sastInstanceId() + "::" + dastIssue.getId();
if (previouslyCorrelatedPairKeys.contains(pairKey)) {
LOG.debug("Skipping already confirmed pair sast={} dast={} from Phase 2 validation",
if (previouslyTriedPairKeys.contains(pairKey)) {
LOG.debug("Skipping already-tried pair sast={} dast={} from Phase 2 validation",
match.sastInstanceId(), dastIssue.getId());
continue;
}
Expand Down
2 changes: 2 additions & 0 deletions ...cli-aviator-common/src/main/java/com/fortify/cli/aviator/grpc/CorrelationStreamState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@ enum Phase { INIT, CORRELATING, VALIDATING, COMPLETE }
final AtomicInteger sentCorrelations = new AtomicInteger(0);
final AtomicInteger receivedCorrelations = new AtomicInteger(0);
final AtomicInteger successfulCorrelations = new AtomicInteger(0);
final AtomicInteger skippedCorrelations = new AtomicInteger(0);
final AtomicInteger failedCorrelations = new AtomicInteger(0);

// Validation phase counters
int totalValidationRequests;
Expand Down
27 changes: 27 additions & 0 deletions ...tor-common/src/test/java/com/fortify/cli/aviator/grpc/CorrelationStreamProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,11 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertIterableEquals;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.junit.jupiter.api.Test;

Expand Down Expand Up @@ -59,6 +61,25 @@ void buildUrlToDastMapPreservesUrlEncounterOrder() throws Exception {
);
}

@Test
void buildCorrelationWorkItemsSkipsFindingWhenAllPairsWereTried() throws Exception {
CorrelationStreamProcessor processor = new CorrelationStreamProcessor(null, null, null, null, 0, 0);
CorrelationStreamProcessor.CorrelationBucketData bucket = createBucket(
"SQL Injection",
List.of("https://example.com/login", "https://example.com/admin")
);
Map<String, List<DastIssue>> urlMap = invokeBuildUrlToDastMap(processor, List.of(bucket));
setField(processor, "urlToDastIssues", urlMap);
setField(processor, "previouslyTriedPairKeys", Set.of(
"SAST-1::https://example.com/login",
"SAST-1::https://example.com/admin"
));

List<?> items = invokeBuildCorrelationWorkItems(processor, List.of(bucket));

assertEquals(0, items.size());
}

private CorrelationStreamProcessor.CorrelationBucketData createBucket(String category, List<String> sessionUrls) {
List<DastIssue> dastIssues = sessionUrls.stream()
.map(this::createDastIssue)
Expand Down Expand Up @@ -96,4 +117,10 @@ private List<String> getDastUrls(Object workItem) throws Exception {
method.setAccessible(true);
return (List<String>) method.invoke(workItem);
}

private void setField(CorrelationStreamProcessor processor, String fieldName, Object value) throws Exception {
Field field = CorrelationStreamProcessor.class.getDeclaredField(fieldName);
field.setAccessible(true);
field.set(processor, value);
}
}
18 changes: 3 additions & 15 deletions ...src/main/java/com/fortify/cli/aviator/ssc/cli/cmd/AbstractAviatorSSCSastAuditCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import com.fortify.cli.aviator.config.AviatorLoggerImpl;
import com.fortify.cli.aviator.config.TagMappingConfig;
import com.fortify.cli.aviator.ssc.helper.AviatorSSCAuditHelper;
import com.fortify.cli.aviator.ssc.helper.AviatorSSCRefreshHelper;
import com.fortify.cli.aviator.ssc.helper.AviatorSSCTagValidator;
import com.fortify.cli.aviator.util.FprHandle;
import com.fortify.cli.aviator.util.ResourceUtil;
Expand All @@ -48,10 +49,7 @@
import com.fortify.cli.ssc.appversion.cli.mixin.SSCAppVersionRefreshOptions;
import com.fortify.cli.ssc.appversion.cli.mixin.SSCAppVersionResolverMixin;
import com.fortify.cli.ssc.appversion.helper.SSCAppVersionDescriptor;
import com.fortify.cli.ssc.appversion.helper.SSCAppVersionHelper;
import com.fortify.cli.ssc.issue.cli.mixin.SSCIssueFilterSetOptionMixin;
import com.fortify.cli.ssc.system_state.helper.SSCJobDescriptor;
import com.fortify.cli.ssc.system_state.helper.SSCJobHelper;

import kong.unirest.UnirestInstance;
import lombok.Getter;
Expand Down Expand Up @@ -94,7 +92,8 @@ public JsonNode getJsonNode(UnirestInstance unirest) {
AviatorLoggerImpl logger = new AviatorLoggerImpl(progressWriter);
SSCAppVersionDescriptor av = appVersionResolver.getAppVersionDescriptor(unirest);

refreshMetricsIfNeeded(unirest, av, logger);
AviatorSSCRefreshHelper.refreshMetricsIfNeeded(
unirest, av, refreshOptions.isRefresh(), refreshOptions.getRefreshTimeout(), logger);

long auditableIssueCount = AviatorSSCAuditHelper.getAuditableIssueCount(unirest, av, logger, isNoFilterSet(), getFilterSetTitleOrId(), folderNames);
if (auditableIssueCount == 0) {
Expand Down Expand Up @@ -136,17 +135,6 @@ boolean isNoFilterSet() {
return noFilterSet;
}

private void refreshMetricsIfNeeded(UnirestInstance unirest, SSCAppVersionDescriptor av, AviatorLoggerImpl logger) {
if (refreshOptions.isRefresh() && av.isRefreshRequired()) {
logger.progress("Status: Metrics for application version %s:%s are out of date, starting refresh...", av.getApplicationName(), av.getVersionName());
SSCJobDescriptor refreshJobDesc = SSCAppVersionHelper.refreshMetrics(unirest, av);
if (refreshJobDesc != null) {
SSCJobHelper.waitForJob(unirest, refreshJobDesc, refreshOptions.getRefreshTimeout());
logger.progress("Status: Metrics refreshed successfully.");
}
}
}

private boolean isSkipIfExceedingQuota() {
return quotaHandlingArgGroup.skipIfExceedingQuota;
}
Expand Down
Loading