diff --git a/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdvice.java b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdvice.java new file mode 100644 index 00000000..ec87e07c --- /dev/null +++ b/etl-service/src/main/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdvice.java @@ -0,0 +1,171 @@ +package com.xtrmetl.etl.controller; + +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import org.springframework.core.MethodParameter; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpInputMessage; +import org.springframework.http.converter.HttpMessageConverter; +import org.springframework.web.bind.annotation.ControllerAdvice; +import org.springframework.web.servlet.mvc.method.annotation.RequestBodyAdviceAdapter; + +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Type; +import java.util.Objects; + +/** + * Enforces the synchronous ETL payload byte limit before Spring MVC materializes a request body. + * + *

Known oversized bodies are rejected from their {@code Content-Length} metadata without reading + * the entity. Unknown-length or understated bodies are wrapped in a byte-counting stream that reads + * at most one byte beyond the configured limit before raising the existing typed payload error. The + * service-level admission check remains in place as defense in depth.

+ */ +@ControllerAdvice(assignableTypes = EtlController.class) +public final class EtlPayloadAdmissionAdvice extends RequestBodyAdviceAdapter { + + private final EtlBatchProperties batchProperties; + + /** + * Creates the MVC transport admission guard. + * + * @param batchProperties bounded ETL request limits shared with the service layer + */ + public EtlPayloadAdmissionAdvice(EtlBatchProperties batchProperties) { + this.batchProperties = Objects.requireNonNull( + batchProperties, + "batchProperties must not be null" + ); + } + + /** + * Applies admission control to string request bodies handled by {@link EtlController}. + * + * @param methodParameter controller method parameter receiving the request body + * @param targetType declared request-body target type + * @param converterType selected HTTP message converter type + * @return {@code true} only for the synchronous ETL string body + */ + @Override + public boolean supports( + MethodParameter methodParameter, + Type targetType, + Class> converterType + ) { + return String.class.equals(methodParameter.getParameterType()); + } + + /** + * Rejects known oversized entities and bounds streaming reads before conversion to a String. + * + * @param inputMessage request headers and body selected by Spring MVC + * @param parameter controller parameter receiving the body + * @param targetType declared request-body target type + * @param converterType selected HTTP message converter type + * @return the original headers with a byte-bounded request stream + * @throws IOException when the underlying request stream cannot be obtained + */ + @Override + public HttpInputMessage beforeBodyRead( + HttpInputMessage inputMessage, + MethodParameter parameter, + Type targetType, + Class> converterType + ) throws IOException { + int maximumBytes = batchProperties.getMaxPayloadBytes(); + long contentLength = inputMessage.getHeaders().getContentLength(); + if (contentLength > maximumBytes) { + throw payloadTooLarge(); + } + return new BoundedHttpInputMessage(inputMessage, maximumBytes); + } + + private static EtlRequestException payloadTooLarge() { + return new EtlRequestException(EtlRequestError.PAYLOAD_TOO_LARGE); + } + + private static final class BoundedHttpInputMessage implements HttpInputMessage { + + private final HttpInputMessage delegate; + private final InputStream body; + + private BoundedHttpInputMessage(HttpInputMessage delegate, int maximumBytes) throws IOException { + this.delegate = Objects.requireNonNull(delegate, "delegate must not be null"); + this.body = new BoundedInputStream(delegate.getBody(), maximumBytes); + } + + @Override + public InputStream getBody() { + return body; + } + + @Override + public HttpHeaders getHeaders() { + return delegate.getHeaders(); + } + } + + private static final class BoundedInputStream extends InputStream { + + private final InputStream delegate; + private long remaining; + + private BoundedInputStream(InputStream delegate, long maximumBytes) { + this.delegate = Objects.requireNonNull(delegate, "delegate must not be null"); + this.remaining = maximumBytes; + } + + @Override + public int read() throws IOException { + if (remaining == 0L) { + int extraByte = delegate.read(); + if (extraByte == -1) { + return -1; + } + throw payloadTooLarge(); + } + + int value = delegate.read(); + if (value != -1) { + remaining--; + } + return value; + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + Objects.checkFromIndexSize(offset, length, bytes.length); + if (length == 0) { + return 0; + } + if (remaining == 0L) { + return rejectExtraByte(); + } + + int boundedLength = (int) Math.min((long) length, remaining + 1L); + int read = delegate.read(bytes, offset, boundedLength); + if (read == -1) { + return -1; + } + if (read > remaining) { + throw payloadTooLarge(); + } + remaining -= read; + return read; + } + + private int rejectExtraByte() throws IOException { + if (delegate.read() == -1) { + return -1; + } + throw payloadTooLarge(); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlHttpPayloadAdmissionTest.java b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlHttpPayloadAdmissionTest.java new file mode 100644 index 00000000..c2ba0589 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlHttpPayloadAdmissionTest.java @@ -0,0 +1,172 @@ +package com.xtrmetl.etl.controller; + +import com.xtrmetl.etl.connector.TargetConnectorDispatcher; +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlService; +import jakarta.servlet.Filter; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletRequestWrapper; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.test.mock.mockito.MockBean; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.security.test.context.support.WithMockUser; +import org.springframework.test.web.servlet.MockMvc; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.not; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; +import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.csrf; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +/** + * Proves synchronous ETL request bytes are bounded before MVC invokes the controller service. + */ +@WebMvcTest(EtlController.class) +@EnableConfigurationProperties(EtlBatchProperties.class) +@Import(EtlHttpPayloadAdmissionTest.UnknownLengthRequestConfig.class) +class EtlHttpPayloadAdmissionTest { + + private static final String PROCESS_PATH = "/api/etl/process"; + private static final String OVERSIZED_MARKER = "oversized-private-marker"; + private static final String UNKNOWN_LENGTH_HEADER = "X-Test-Unknown-Content-Length"; + + @Autowired + private MockMvc mockMvc; + + @MockBean + private EtlService etlService; + + @MockBean + private TargetConnectorDispatcher connectorDispatcher; + + @Test + @WithMockUser + void rejectsKnownOversizedBodyBeforeControllerInvocation() throws Exception { + String request = oversizedJsonRequest(); + when(etlService.processData(anyString())).thenReturn("unexpected controller invocation"); + + mockMvc.perform(post(PROCESS_PATH) + .with(csrf()) + .contentType(MediaType.APPLICATION_JSON) + .content(request)) + .andExpect(status().isPayloadTooLarge()) + .andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_PROBLEM_JSON)) + .andExpect(header().string(HttpHeaders.CACHE_CONTROL, "no-store")) + .andExpect(jsonPath("$.errorCode").value("etl_payload_too_large")) + .andExpect(jsonPath("$.instance").value(PROCESS_PATH)) + .andExpect(content().string(not(containsString(OVERSIZED_MARKER)))); + + verifyNoInteractions(etlService); + } + + @Test + @WithMockUser + void rejectsUnknownLengthOversizedBodyBeforeControllerInvocation() throws Exception { + String request = oversizedJsonRequest(); + when(etlService.processData(anyString())).thenReturn("unexpected controller invocation"); + + mockMvc.perform(post(PROCESS_PATH) + .with(csrf()) + .header(UNKNOWN_LENGTH_HEADER, "true") + .header(HttpHeaders.TRANSFER_ENCODING, "chunked") + .contentType(MediaType.APPLICATION_JSON) + .content(request)) + .andExpect(status().isPayloadTooLarge()) + .andExpect(jsonPath("$.errorCode").value("etl_payload_too_large")); + + verifyNoInteractions(etlService); + } + + @Test + @WithMockUser + void acceptsKnownLengthBodyAtExactByteLimit() throws Exception { + String request = sizedJsonRequest(EtlBatchProperties.DEFAULT_MAX_PAYLOAD_BYTES); + when(etlService.processData(request)).thenReturn("processed"); + + mockMvc.perform(post(PROCESS_PATH) + .with(csrf()) + .contentType(MediaType.APPLICATION_JSON) + .content(request)) + .andExpect(status().isOk()) + .andExpect(content().string("processed")); + + verify(etlService).processData(request); + } + + @Test + @WithMockUser + void acceptsUnknownLengthBodyImmediatelyBelowByteLimit() throws Exception { + String request = sizedJsonRequest(EtlBatchProperties.DEFAULT_MAX_PAYLOAD_BYTES - 1); + when(etlService.processData(request)).thenReturn("processed"); + + mockMvc.perform(post(PROCESS_PATH) + .with(csrf()) + .header(UNKNOWN_LENGTH_HEADER, "true") + .header(HttpHeaders.TRANSFER_ENCODING, "chunked") + .contentType(MediaType.APPLICATION_JSON) + .content(request)) + .andExpect(status().isOk()) + .andExpect(content().string("processed")); + + verify(etlService).processData(request); + } + + private static String oversizedJsonRequest() { + return "[{\"id\":\"" + OVERSIZED_MARKER + "" + + "x".repeat(EtlBatchProperties.DEFAULT_MAX_PAYLOAD_BYTES) + + "\"}]"; + } + + private static String sizedJsonRequest(int totalBytes) { + String prefix = "[{\"id\":\""; + String suffix = "\"}]"; + int fillerLength = totalBytes - prefix.length() - suffix.length(); + return prefix + "x".repeat(fillerLength) + suffix; + } + + /** + * Test-only transport shim that models chunked input whose byte length is not known up front. + */ + @TestConfiguration + static class UnknownLengthRequestConfig { + + @Bean + @Order(Ordered.HIGHEST_PRECEDENCE) + Filter unknownLengthRequestFilter() { + return (request, response, chain) -> { + if (request instanceof HttpServletRequest httpRequest + && httpRequest.getHeader(UNKNOWN_LENGTH_HEADER) != null) { + chain.doFilter(new HttpServletRequestWrapper(httpRequest) { + @Override + public int getContentLength() { + return -1; + } + + @Override + public long getContentLengthLong() { + return -1L; + } + }, response); + return; + } + chain.doFilter(request, response); + }; + } + } +} diff --git a/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdviceTest.java b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdviceTest.java new file mode 100644 index 00000000..7af909a2 --- /dev/null +++ b/etl-service/src/test/java/com/xtrmetl/etl/controller/EtlPayloadAdmissionAdviceTest.java @@ -0,0 +1,285 @@ +package com.xtrmetl.etl.controller; + +import com.xtrmetl.etl.service.EtlBatchProperties; +import com.xtrmetl.etl.service.EtlRequestError; +import com.xtrmetl.etl.service.EtlRequestException; +import org.junit.jupiter.api.Test; +import org.springframework.core.MethodParameter; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpInputMessage; +import org.springframework.http.converter.StringHttpMessageConverter; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.security.Principal; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies the transport-level ETL payload guard without relying on MVC implementation details. + */ +class EtlPayloadAdmissionAdviceTest { + + private static final int MAXIMUM_BYTES = 8; + + @Test + void supportsOnlyStringRequestParameters() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + + assertTrue(advice.supports( + requestBodyParameter(), + String.class, + StringHttpMessageConverter.class + )); + assertFalse(advice.supports( + integerParameter(), + Integer.class, + StringHttpMessageConverter.class + )); + } + + @Test + void requiresBatchProperties() { + assertThrows(NullPointerException.class, () -> new EtlPayloadAdmissionAdvice(null)); + } + + @Test + void rejectsKnownOversizedBodyWithoutReadingAnyEntityByte() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + CountingInputStream body = new CountingInputStream(new byte[MAXIMUM_BYTES + 1]); + TestHttpInputMessage inputMessage = new TestHttpInputMessage(body, MAXIMUM_BYTES + 1L); + + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> advice.beforeBodyRead( + inputMessage, + requestBodyParameter(), + String.class, + StringHttpMessageConverter.class + ) + ); + + assertSame(EtlRequestError.PAYLOAD_TOO_LARGE, exception.error()); + assertEquals(0, body.bytesRead()); + } + + @Test + void preservesHeadersAndRejectsUnknownLengthBodyAfterOnlyLimitPlusOneByte() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + CountingInputStream body = new CountingInputStream(new byte[MAXIMUM_BYTES + 32]); + TestHttpInputMessage inputMessage = new TestHttpInputMessage(body, -1L); + inputMessage.getHeaders().set("X-Test-Header", "preserved"); + HttpInputMessage boundedMessage = bounded(advice, inputMessage); + + assertSame(inputMessage.getHeaders(), boundedMessage.getHeaders()); + EtlRequestException exception = assertThrows( + EtlRequestException.class, + () -> boundedMessage.getBody().readAllBytes() + ); + + assertSame(EtlRequestError.PAYLOAD_TOO_LARGE, exception.error()); + assertEquals(MAXIMUM_BYTES + 1, body.bytesRead()); + } + + @Test + void readsExactLimitToEndWithoutFalsePositive() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + byte[] payload = "12345678".getBytes(StandardCharsets.UTF_8); + CountingInputStream body = new CountingInputStream(payload); + HttpInputMessage boundedMessage = bounded(advice, new TestHttpInputMessage(body, -1L)); + + assertArrayEquals(payload, boundedMessage.getBody().readAllBytes()); + assertEquals(MAXIMUM_BYTES, body.bytesRead()); + } + + @Test + void supportsSingleByteReadsAtAndBeyondTheLimit() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + HttpInputMessage exact = bounded( + advice, + new TestHttpInputMessage(new ByteArrayInputStream("12345678".getBytes(StandardCharsets.UTF_8)), -1L) + ); + InputStream exactBody = exact.getBody(); + for (int index = 0; index < MAXIMUM_BYTES; index++) { + assertEquals('1' + index, exactBody.read()); + } + assertEquals(-1, exactBody.read()); + + HttpInputMessage oversized = bounded( + advice, + new TestHttpInputMessage(new ByteArrayInputStream("123456789".getBytes(StandardCharsets.UTF_8)), -1L) + ); + InputStream oversizedBody = oversized.getBody(); + for (int index = 0; index < MAXIMUM_BYTES; index++) { + oversizedBody.read(); + } + assertThrows(EtlRequestException.class, oversizedBody::read); + } + + @Test + void handlesZeroLengthBulkReadAndEndOfStreamAtTheLimit() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + InputStream body = bounded( + advice, + new TestHttpInputMessage(new ByteArrayInputStream("12345678".getBytes(StandardCharsets.UTF_8)), -1L) + ).getBody(); + byte[] buffer = new byte[MAXIMUM_BYTES]; + + assertEquals(0, body.read(buffer, 0, 0)); + assertEquals(MAXIMUM_BYTES, body.read(buffer, 0, buffer.length)); + assertEquals(-1, body.read(buffer, 0, 1)); + } + + @Test + void propagatesBodyAcquisitionIOExceptionUnchanged() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + IOException expected = new IOException("test stream acquisition failure"); + HttpInputMessage failingMessage = new HttpInputMessage() { + @Override + public InputStream getBody() throws IOException { + throw expected; + } + + @Override + public HttpHeaders getHeaders() { + return new HttpHeaders(); + } + }; + + IOException actual = assertThrows( + IOException.class, + () -> bounded(advice, failingMessage) + ); + assertSame(expected, actual); + } + + @Test + void closesTheUnderlyingRequestBody() throws Exception { + EtlPayloadAdmissionAdvice advice = advice(); + CloseTrackingInputStream delegate = new CloseTrackingInputStream(new byte[0]); + InputStream body = bounded(advice, new TestHttpInputMessage(delegate, -1L)).getBody(); + + body.close(); + + assertTrue(delegate.closed()); + } + + private static HttpInputMessage bounded( + EtlPayloadAdmissionAdvice advice, + HttpInputMessage inputMessage + ) throws Exception { + return advice.beforeBodyRead( + inputMessage, + requestBodyParameter(), + String.class, + StringHttpMessageConverter.class + ); + } + + private static EtlPayloadAdmissionAdvice advice() { + EtlBatchProperties properties = new EtlBatchProperties(); + properties.setMaxPayloadBytes(MAXIMUM_BYTES); + return new EtlPayloadAdmissionAdvice(properties); + } + + private static MethodParameter requestBodyParameter() throws NoSuchMethodException { + Method method = EtlController.class.getMethod( + "processData", + String.class, + String.class, + Principal.class + ); + return new MethodParameter(method, 0); + } + + private static MethodParameter integerParameter() throws NoSuchMethodException { + Method method = EtlPayloadAdmissionAdviceTest.class.getDeclaredMethod("integerBody", Integer.class); + return new MethodParameter(method, 0); + } + + @SuppressWarnings("unused") + private static void integerBody(Integer value) { + // Reflection target used only to prove the RequestBodyAdvice type filter. + } + + private static final class TestHttpInputMessage implements HttpInputMessage { + + private final HttpHeaders headers = new HttpHeaders(); + private final InputStream body; + + private TestHttpInputMessage(InputStream body, long contentLength) { + this.body = body; + if (contentLength >= 0L) { + headers.setContentLength(contentLength); + } + } + + @Override + public InputStream getBody() { + return body; + } + + @Override + public HttpHeaders getHeaders() { + return headers; + } + } + + private static class CountingInputStream extends ByteArrayInputStream { + + private int bytesRead; + + private CountingInputStream(byte[] bytes) { + super(bytes); + } + + @Override + public synchronized int read() { + int value = super.read(); + if (value != -1) { + bytesRead++; + } + return value; + } + + @Override + public synchronized int read(byte[] bytes, int offset, int length) { + int read = super.read(bytes, offset, length); + if (read > 0) { + bytesRead += read; + } + return read; + } + + private int bytesRead() { + return bytesRead; + } + } + + private static final class CloseTrackingInputStream extends CountingInputStream { + + private boolean closed; + + private CloseTrackingInputStream(byte[] bytes) { + super(bytes); + } + + @Override + public void close() throws IOException { + closed = true; + super.close(); + } + + private boolean closed() { + return closed; + } + } +}