From 1971accbc32269392bf65b8d9dc0c61f1e59a519 Mon Sep 17 00:00:00 2001 From: Tamar Weisskopf Date: Tue, 18 Aug 2026 01:50:55 +0300 Subject: [PATCH 1/2] fix: SBOM analysis request should be blocked when user limit exceeded Add upfront queue capacity check to SPDX upload before creating product, matching RPM/CycloneDX behavior. When user limit exceeded, return 429 immediately instead of accepting request then failing all components. Changes: - SbomReportService: Add RequestQueueService injection and capacity check - Add SbomReportServiceQueueAdmissionTest to verify exception handling --- .../exploitiq/service/SbomReportService.java | 71 ++++++--- .../SbomReportServiceQueueAdmissionTest.java | 148 ++++++++++++++++++ 2 files changed, 194 insertions(+), 25 deletions(-) create mode 100644 src/test/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportServiceQueueAdmissionTest.java diff --git a/src/main/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportService.java b/src/main/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportService.java index a7ef4492..712d77b1 100644 --- a/src/main/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportService.java +++ b/src/main/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportService.java @@ -53,6 +53,7 @@ public class SbomReportService { private ComponentProcessingService componentProcessingService; private CredentialProcessingService credentialProcessingService; private ObjectMapper objectMapper; + private RequestQueueService queueService; @Inject public void setCycloneDxParsingService(CycloneDxParsingService cycloneDxParsingService) { @@ -94,6 +95,11 @@ public void setObjectMapper(ObjectMapper objectMapper) { this.objectMapper = objectMapper; } + @Inject + public void setRequestQueueService(RequestQueueService queueService) { + this.queueService = queueService; + } + /** * Generates a product ID from SBOM name and version. @@ -221,35 +227,50 @@ public String submitSpdx(InputStream fileInputStream, String cveId, String crede throw new ValidationException(errors); } LOGGER.info("Processing SPDX file upload for CVE: " + cveId); - - SpdxParsingService.ProductInfo productInfo = parsed.productInfo(); - Map metadata = new HashMap<>(); - // Add CPE to metadata if present - if (productInfo.cpe() != null && !productInfo.cpe().trim().isEmpty()) { - metadata.put("cpe", productInfo.cpe()); - } - - if (Objects.nonNull(productInfo.spdxId())) { - metadata.put(RepositoryConstants.SPDX_ID_METADATA_KEY, productInfo.spdxId()); - } - int totalComponentCount = parsed.components().size() + parsed.unsupportedComponents().size(); - Product product = this.createProduct(cveId, productInfo.name(), productInfo.version(), totalComponentCount, metadata); + final SpdxParsingService.ProductInfo productInfo = parsed.productInfo(); - for (SpdxParsingService.UnsupportedComponentInfo unsupported : parsed.unsupportedComponents()) { - String errorMessage = - "Expects a container image purl with format pkg:oci/name@sha256:hash or pkg:oci/name@sha256%3Ahash?repository_url=...&tag=..."; - String imageForDisplay = unsupported.purl() != null ? unsupported.purl() : ""; - productRepository.addSubmissionFailure(product.id(), new FailedComponent( - unsupported.name(), unsupported.version(), imageForDisplay, errorMessage)); - } + // Generate productId before capacity check (needed for product slot optimization) + final String productId = generateProductId(productInfo.name(), productInfo.version()); + + // Get current user + final String user = userService.getUserName(); + + // Make variables final for lambda capture + final SpdxParsingService.ParsedSpdx finalParsed = parsed; + final String finalCveId = cveId; + final String finalCredentialId = credentialId; + + // Check user capacity before creating product (matches RPM/CycloneDX pattern) + return queueService.runIfHasCapacity(user, productId, () -> { + Map metadata = new HashMap<>(); + // Add CPE to metadata if present + if (productInfo.cpe() != null && !productInfo.cpe().trim().isEmpty()) { + metadata.put("cpe", productInfo.cpe()); + } + + if (Objects.nonNull(productInfo.spdxId())) { + metadata.put(RepositoryConstants.SPDX_ID_METADATA_KEY, productInfo.spdxId()); + } + + int totalComponentCount = finalParsed.components().size() + finalParsed.unsupportedComponents().size(); + Product product = this.createProduct(finalCveId, productInfo.name(), productInfo.version(), totalComponentCount, metadata); + + for (SpdxParsingService.UnsupportedComponentInfo unsupported : finalParsed.unsupportedComponents()) { + String errorMessage = + "Expects a container image purl with format pkg:oci/name@sha256:hash or pkg:oci/name@sha256%3Ahash?repository_url=...&tag=..."; + String imageForDisplay = unsupported.purl() != null ? unsupported.purl() : ""; + productRepository.addSubmissionFailure(product.id(), new FailedComponent( + unsupported.name(), unsupported.version(), imageForDisplay, errorMessage)); + } + + // Start component processing (chunks run in parallel on executor) + processSpdxComponents(product.id(), finalParsed, finalCveId, finalCredentialId); - // Start component processing (chunks run in parallel on executor) - processSpdxComponents(product.id(), parsed, cveId, credentialId); + LOGGER.infof("Created product %s, started component processing", product.id()); - LOGGER.infof("Created product %s, started component processing", product.id()); - - return product.id(); + return product.id(); + }); } private void processSpdxComponents(String productId, SpdxParsingService.ParsedSpdx parsed, String vulnerabilityId, String credentialId) { diff --git a/src/test/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportServiceQueueAdmissionTest.java b/src/test/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportServiceQueueAdmissionTest.java new file mode 100644 index 00000000..3507c76d --- /dev/null +++ b/src/test/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportServiceQueueAdmissionTest.java @@ -0,0 +1,148 @@ +package com.redhat.ecosystemappeng.exploitiq.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.redhat.ecosystemappeng.exploitiq.repository.ProductRepositoryService; +import io.quarkus.test.InjectMock; +import io.quarkus.test.component.QuarkusComponentTest; +import jakarta.inject.Inject; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.nullable; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Tests that SPDX upload checks queue capacity before creating product, + * matching the pattern used by RPM and CycloneDX flows. + */ +@QuarkusComponentTest +class SbomReportServiceQueueAdmissionTest { + + @Inject + SbomReportService sbomReportService; + + @InjectMock + RequestQueueService queueService; + @InjectMock + UserService userService; + @InjectMock + ProductRepositoryService productRepository; + @InjectMock + SpdxParsingService spdxParsingService; + @InjectMock + ComponentProcessingService componentProcessingService; + @InjectMock + CycloneDxParsingService cycloneDxParsingService; + @InjectMock + ReportService reportService; + @InjectMock + CredentialProcessingService credentialProcessingService; + @InjectMock + ObjectMapper objectMapper; + + /** + * Test that SPDX upload throws UserQueueExceededException when user limit is exceeded, + * and does NOT create a product document. This matches the behavior of RPM analysis. + */ + @Test + void submitSpdx_UserQueueExceeded_DoesNotCreateProduct() throws Exception { + // Prepare minimal valid SPDX JSON + String spdxJson = """ + { + "spdxVersion": "SPDX-2.3", + "name": "test-product", + "documentNamespace": "https://example.com/test", + "packages": [ + { + "name": "test-component", + "versionInfo": "1.0.0", + "externalRefs": [ + { + "referenceCategory": "PACKAGE-MANAGER", + "referenceType": "purl", + "referenceLocator": "pkg:oci/test@sha256:abcd1234" + } + ] + } + ] + } + """; + + InputStream spdxStream = new ByteArrayInputStream(spdxJson.getBytes(StandardCharsets.UTF_8)); + + // Mock parsing to return valid parsed data + SpdxParsingService.ProductInfo productInfo = new SpdxParsingService.ProductInfo( + "test-product", "1.0.0", null, null + ); + SpdxParsingService.ParsedSpdx parsedSpdx = new SpdxParsingService.ParsedSpdx( + productInfo, + java.util.List.of(), // components + java.util.List.of() // unsupported components + ); + when(spdxParsingService.parse(any())).thenReturn(parsedSpdx); + when(userService.getUserName()).thenReturn("alice"); + + // Mock queueService to throw UserQueueExceededException (user limit exceeded) + doThrow(new UserQueueExceededException(5)) + .when(queueService).runIfHasCapacity(any(), nullable(String.class), any()); + + // Assert that exception is thrown + assertThrows(UserQueueExceededException.class, + () -> sbomReportService.submitSpdx(spdxStream, "CVE-2024-1234", null)); + + // Verify: No product created (save never called on productRepository) + verify(productRepository, never()).save(any(), any()); + + // Verify: No component processing started + verify(componentProcessingService, never()).processComponents(any(), any(), any(), any(), any()); + } + + /** + * Test that SPDX upload throws RequestQueueExceededException when global queue is full, + * and does NOT create a product document. + */ + @Test + void submitSpdx_GlobalQueueExceeded_DoesNotCreateProduct() throws Exception { + String spdxJson = """ + { + "spdxVersion": "SPDX-2.3", + "name": "test-product", + "versionInfo": "2.0.0", + "documentNamespace": "https://example.com/test2", + "packages": [] + } + """; + + InputStream spdxStream = new ByteArrayInputStream(spdxJson.getBytes(StandardCharsets.UTF_8)); + + SpdxParsingService.ProductInfo productInfo = new SpdxParsingService.ProductInfo( + "test-product", "2.0.0", null, null + ); + SpdxParsingService.ParsedSpdx parsedSpdx = new SpdxParsingService.ParsedSpdx( + productInfo, + java.util.List.of(), + java.util.List.of() + ); + when(spdxParsingService.parse(any())).thenReturn(parsedSpdx); + when(userService.getUserName()).thenReturn("bob"); + + // Mock queueService to throw RequestQueueExceededException (global queue full) + doThrow(new RequestQueueExceededException(500)) + .when(queueService).runIfHasCapacity(any(), nullable(String.class), any()); + + // Assert that exception is thrown + assertThrows(RequestQueueExceededException.class, + () -> sbomReportService.submitSpdx(spdxStream, "CVE-2024-5678", null)); + + // Verify: No product created + verify(productRepository, never()).save(any(), any()); + } +} From 7802955a4a9ed5333850abc043657dd3f1b98153 Mon Sep 17 00:00:00 2001 From: Tamar Weisskopf Date: Tue, 18 Aug 2026 10:54:52 +0300 Subject: [PATCH 2/2] fix: address PR review feedback - product ID correlation and test improvements Critical fixes: - Fix product ID mismatch: pass productId to createProduct() instead of regenerating with different timestamp, ensuring queue slot and DB product have matching IDs - Add 429 response documentation to OpenAPI for both SPDX and CycloneDX upload endpoints Test improvements: - Add happy path test verifying lambda execution and product creation - Use eq() for user and anyString() for productId in test assertions - Add error logging in lambda to avoid silent 500 errors - Add componentProcessingService verification to second test Addresses review feedback from zvigrinberg and tmihalac Co-Authored-By: Claude Sonnet 4.5 --- .../exploitiq/rest/ProductEndpoint.java | 28 ++++++++- .../exploitiq/service/SbomReportService.java | 60 +++++++++--------- .../SbomReportServiceQueueAdmissionTest.java | 61 ++++++++++++++++++- 3 files changed, 115 insertions(+), 34 deletions(-) diff --git a/src/main/java/com/redhat/ecosystemappeng/exploitiq/rest/ProductEndpoint.java b/src/main/java/com/redhat/ecosystemappeng/exploitiq/rest/ProductEndpoint.java index eda87dc9..87d6f84e 100644 --- a/src/main/java/com/redhat/ecosystemappeng/exploitiq/rest/ProductEndpoint.java +++ b/src/main/java/com/redhat/ecosystemappeng/exploitiq/rest/ProductEndpoint.java @@ -226,6 +226,17 @@ public Response remove( responseCode = "400", description = "Validation error with field-specific error messages" ), + @APIResponse( + responseCode = "429", + description = "Per-user concurrent request limit exceeded or global queue is full", + content = @Content( + mediaType = MediaType.APPLICATION_JSON, + schema = @Schema( + type = SchemaType.OBJECT, + example = "{\"error\": \"Per-user concurrent request limit exceeded\"}" + ) + ) + ), @APIResponse( responseCode = "500", description = "Internal server error" @@ -356,7 +367,7 @@ public Response mapSbomValidationException(SbomValidationException e) { description = "Uploads an SPDX SBOM file, parses it, creates a product, and starts async processing. Requires a vulnerability ID to include in all component reports. Accepts optional credentials for private repository access.") @APIResponses({ @APIResponse( - responseCode = "202", + responseCode = "202", description = "Product creation request accepted", content = @Content( mediaType = MediaType.APPLICATION_JSON, @@ -366,11 +377,22 @@ public Response mapSbomValidationException(SbomValidationException e) { ) ), @APIResponse( - responseCode = "400", + responseCode = "400", description = "Invalid SPDX file, missing required data, missing CVE ID, or credential validation error" ), @APIResponse( - responseCode = "500", + responseCode = "429", + description = "Per-user concurrent request limit exceeded or global queue is full", + content = @Content( + mediaType = MediaType.APPLICATION_JSON, + schema = @Schema( + type = SchemaType.OBJECT, + example = "{\"error\": \"Per-user concurrent request limit exceeded\"}" + ) + ) + ), + @APIResponse( + responseCode = "500", description = "Internal server error" ) }) diff --git a/src/main/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportService.java b/src/main/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportService.java index 712d77b1..747c2123 100644 --- a/src/main/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportService.java +++ b/src/main/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportService.java @@ -243,33 +243,38 @@ public String submitSpdx(InputStream fileInputStream, String cveId, String crede // Check user capacity before creating product (matches RPM/CycloneDX pattern) return queueService.runIfHasCapacity(user, productId, () -> { - Map metadata = new HashMap<>(); - // Add CPE to metadata if present - if (productInfo.cpe() != null && !productInfo.cpe().trim().isEmpty()) { - metadata.put("cpe", productInfo.cpe()); - } - - if (Objects.nonNull(productInfo.spdxId())) { - metadata.put(RepositoryConstants.SPDX_ID_METADATA_KEY, productInfo.spdxId()); - } - - int totalComponentCount = finalParsed.components().size() + finalParsed.unsupportedComponents().size(); - Product product = this.createProduct(finalCveId, productInfo.name(), productInfo.version(), totalComponentCount, metadata); - - for (SpdxParsingService.UnsupportedComponentInfo unsupported : finalParsed.unsupportedComponents()) { - String errorMessage = - "Expects a container image purl with format pkg:oci/name@sha256:hash or pkg:oci/name@sha256%3Ahash?repository_url=...&tag=..."; - String imageForDisplay = unsupported.purl() != null ? unsupported.purl() : ""; - productRepository.addSubmissionFailure(product.id(), new FailedComponent( - unsupported.name(), unsupported.version(), imageForDisplay, errorMessage)); + try { + Map metadata = new HashMap<>(); + // Add CPE to metadata if present + if (productInfo.cpe() != null && !productInfo.cpe().trim().isEmpty()) { + metadata.put("cpe", productInfo.cpe()); + } + + if (Objects.nonNull(productInfo.spdxId())) { + metadata.put(RepositoryConstants.SPDX_ID_METADATA_KEY, productInfo.spdxId()); + } + + int totalComponentCount = finalParsed.components().size() + finalParsed.unsupportedComponents().size(); + Product product = this.createProduct(productId, finalCveId, productInfo.name(), productInfo.version(), totalComponentCount, metadata); + + for (SpdxParsingService.UnsupportedComponentInfo unsupported : finalParsed.unsupportedComponents()) { + String errorMessage = + "Expects a container image purl with format pkg:oci/name@sha256:hash or pkg:oci/name@sha256%3Ahash?repository_url=...&tag=..."; + String imageForDisplay = unsupported.purl() != null ? unsupported.purl() : ""; + productRepository.addSubmissionFailure(product.id(), new FailedComponent( + unsupported.name(), unsupported.version(), imageForDisplay, errorMessage)); + } + + // Start component processing (chunks run in parallel on executor) + processSpdxComponents(product.id(), finalParsed, finalCveId, finalCredentialId); + + LOGGER.infof("Created product %s, started component processing", product.id()); + + return product.id(); + } catch (Exception e) { + LOGGER.errorf(e, "Failed to create product or start component processing for CVE %s", finalCveId); + throw e; } - - // Start component processing (chunks run in parallel on executor) - processSpdxComponents(product.id(), finalParsed, finalCveId, finalCredentialId); - - LOGGER.infof("Created product %s, started component processing", product.id()); - - return product.id(); }); } @@ -294,8 +299,7 @@ private void processSpdxComponents(String productId, SpdxParsingService.ParsedSp } } - private Product createProduct(String cveId, String sbomName, String sbomVersion, int componentCount, Map metadata) { - String productId = generateProductId(sbomName, sbomVersion); + private Product createProduct(String productId, String cveId, String sbomName, String sbomVersion, int componentCount, Map metadata) { Product product = newProductDocument(cveId, productId, sbomName, sbomVersion, componentCount, metadata); productRepository.save(product, userService.getUserName()); return product; diff --git a/src/test/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportServiceQueueAdmissionTest.java b/src/test/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportServiceQueueAdmissionTest.java index 3507c76d..da53834c 100644 --- a/src/test/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportServiceQueueAdmissionTest.java +++ b/src/test/java/com/redhat/ecosystemappeng/exploitiq/service/SbomReportServiceQueueAdmissionTest.java @@ -12,8 +12,11 @@ import java.nio.charset.StandardCharsets; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.nullable; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -48,6 +51,55 @@ class SbomReportServiceQueueAdmissionTest { @InjectMock ObjectMapper objectMapper; + /** + * Test happy path: when capacity is available, the lambda executes, + * product is created, and component processing starts. + */ + @Test + void submitSpdx_HappyPath_CreatesProductAndProcessesComponents() throws Exception { + String spdxJson = """ + { + "spdxVersion": "SPDX-2.3", + "name": "happy-product", + "versionInfo": "3.0.0", + "documentNamespace": "https://example.com/happy", + "packages": [] + } + """; + + InputStream spdxStream = new ByteArrayInputStream(spdxJson.getBytes(StandardCharsets.UTF_8)); + + SpdxParsingService.ProductInfo productInfo = new SpdxParsingService.ProductInfo( + "happy-product", "3.0.0", null, null + ); + SpdxParsingService.ParsedSpdx parsedSpdx = new SpdxParsingService.ParsedSpdx( + productInfo, + java.util.List.of(), + java.util.List.of() + ); + when(spdxParsingService.parse(any())).thenReturn(parsedSpdx); + when(userService.getUserName()).thenReturn("charlie"); + + // Mock queueService to execute the lambda when capacity is available + doAnswer(invocation -> { + @SuppressWarnings("unchecked") + java.util.function.Supplier action = invocation.getArgument(2); + return action.get(); // Execute the lambda + }).when(queueService).runIfHasCapacity(eq("charlie"), anyString(), any()); + + // Execute + String result = sbomReportService.submitSpdx(spdxStream, "CVE-2024-9999", null); + + // Verify: Product created + verify(productRepository).save(any(), eq("charlie")); + + // Verify: Component processing started + verify(componentProcessingService).processComponents(any(), anyString(), any(), eq("CVE-2024-9999"), eq(null)); + + // Verify: Returns product ID (non-null) + assertEquals(true, result != null && !result.isEmpty()); + } + /** * Test that SPDX upload throws UserQueueExceededException when user limit is exceeded, * and does NOT create a product document. This matches the behavior of RPM analysis. @@ -92,7 +144,7 @@ void submitSpdx_UserQueueExceeded_DoesNotCreateProduct() throws Exception { // Mock queueService to throw UserQueueExceededException (user limit exceeded) doThrow(new UserQueueExceededException(5)) - .when(queueService).runIfHasCapacity(any(), nullable(String.class), any()); + .when(queueService).runIfHasCapacity(eq("alice"), anyString(), any()); // Assert that exception is thrown assertThrows(UserQueueExceededException.class, @@ -136,7 +188,7 @@ void submitSpdx_GlobalQueueExceeded_DoesNotCreateProduct() throws Exception { // Mock queueService to throw RequestQueueExceededException (global queue full) doThrow(new RequestQueueExceededException(500)) - .when(queueService).runIfHasCapacity(any(), nullable(String.class), any()); + .when(queueService).runIfHasCapacity(eq("bob"), anyString(), any()); // Assert that exception is thrown assertThrows(RequestQueueExceededException.class, @@ -144,5 +196,8 @@ void submitSpdx_GlobalQueueExceeded_DoesNotCreateProduct() throws Exception { // Verify: No product created verify(productRepository, never()).save(any(), any()); + + // Verify: No component processing started + verify(componentProcessingService, never()).processComponents(any(), any(), any(), any(), any()); } }