diff --git a/CHANGELOG.md b/CHANGELOG.md index 35e7283..bab942f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## Unreleased + +### Fixed + +- `GuardFilter` no longer buffers every response. Responses stream unbuffered (SSE, `StreamingResponseBody`, `ResponseBodyEmitter`, `DeferredResult`, file downloads) and the security headers, CORS headers, metrics and status based behavior rules are applied just before the response commits, the way fastapi-guard's `PureASGISecurityMiddleware` does on `http.response.start`. The response is buffered only on requests where a `custom_response_modifier` is configured or a route or global return-pattern rule inspects the body. Guard-generated rejections are unchanged. The buffered response now implements `BoundedResponseBodyReader`, so body based return patterns (with `guard.behavior-scan-response-body=true`) evaluate; before, they never could. + +- Body based `return_pattern` behavior rules now evaluate: `ServletGuardResponse` implements `BoundedResponseBodyReader`. Before, the engine could never read the response body in this adapter. + +### Changed + +- Streamed responses no longer get a `Content-Length` set by the guard, since the guard no longer buffers them. The container decides, usually chunked encoding. +- Documented how to align Kotlin and coroutines under Spring Boot 3.5 dependency management, which otherwise downgrades both and fails the application at startup (README and docs/installation.md). + ## [1.1.0] - 2026-10-16 ### Added diff --git a/README.md b/README.md index b4f8c0e..d316d41 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,37 @@ dependencies { It brings in the engine facade `com.guard-core:guard-core:4.3.3`. To build from source, see "Engine dependency" below, or consume the composite from a sibling checkout. +## Kotlin and coroutines versions under Spring Boot dependency management + +Spring Boot 3.5 manages `kotlin.version` 1.9.25, `kotlin-coroutines.version` 1.8.1 and `kotlin-serialization.version` 1.6.3, and dependency management overrides the versions this library was built with (Kotlin 2.4.21, coroutines 1.11.0). A Java application that imports the `spring-boot-dependencies` BOM, or uses the Spring Boot Gradle plugin with `io.spring.dependency-management`, therefore fails at startup with `NoSuchMethodError: kotlinx.coroutines.BuildersKt.runBlockingK$default` (coroutines 1.11.0 added `runBlockingK`), and with coroutines 1.11.0 on a Kotlin 1.9 stdlib every request fails with `ClassNotFoundException: kotlin.coroutines.jvm.internal.SpillingKt`. Align both versions. Serialization can stay on the managed 1.6.3. + +Gradle with the Spring Boot plugin and `io.spring.dependency-management`: + +```groovy +ext['kotlin.version'] = '2.4.21' +ext['kotlin-coroutines.version'] = '1.11.0' +``` + +Maven with the BOM imported (declare these imports before `spring-boot-dependencies`, the first declaration wins): + +```xml + + + + org.jetbrains.kotlinkotlin-bom + 2.4.21pomimport + + + org.jetbrains.kotlinxkotlinx-coroutines-bom + 1.11.0pomimport + + + + +``` + +With `spring-boot-starter-parent` as the parent, set `2.4.21` and `1.11.0` as properties instead. + ## Quick start ```kotlin @@ -145,8 +176,7 @@ workflow proves the published artifact graph (docs/release.md). Adapter parity with the reference middleware releases is assessed per upstream feature release: [docs/adapter-parity-8.1.0.md](docs/adapter-parity-8.1.0.md) maps fastapi-guard 8.1.0's -`PureASGISecurityMiddleware` onto this adapter's `OncePerRequestFilter` idiom (the streaming -trade is documented there honestly: the response wrapper buffers by design). +`PureASGISecurityMiddleware` onto this adapter's `OncePerRequestFilter` idiom. Responses stream unbuffered (SSE, `StreamingResponseBody`, file downloads): security headers, CORS, metrics and status based return rules apply just before the response commits. The response is buffered only on requests where a `custom_response_modifier` is configured or a route or global return-pattern rule inspects the body (`guard.behavior-scan-response-body=true` with a pattern that is not `status:`). Local checkout: diff --git a/docs/adapter-parity-8.1.0.md b/docs/adapter-parity-8.1.0.md index b3b1c13..1eea12d 100644 --- a/docs/adapter-parity-8.1.0.md +++ b/docs/adapter-parity-8.1.0.md @@ -60,7 +60,9 @@ are the bytes the handler sees, with bounded prefix reads stopping the wire pull at the cap. This is the exact contract of the reference's caching receive (and Starlette's own `_body` cache it mirrors). -### 3. Unbuffered response streaming: DOES NOT HOLD (documented finding) +### 3. Unbuffered response streaming: HOLDS since the Unreleased change (history below) + +Update: `GuardFilter` now streams by default. It wraps the response in `CommitHookResponse`, which runs the facade's `process_response` exit path and the CORS injection once, just before the first byte can leave (first write, flush, close, `sendError`, `sendRedirect`, or the completion of an async handler), the servlet analog of `http.response.start`. Only requests where a `custom_response_modifier` is configured or a route or global return-pattern rule inspects the body keep the buffering wrapper. The finding below is the original W10 assessment, kept for history. `GuardFilter` wraps the response in Spring's `ContentCachingResponseWrapper` and calls `copyBodyToResponse()` only after the facade's `process_response` diff --git a/docs/installation.md b/docs/installation.md index e2967e1..4ef11c9 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -20,6 +20,37 @@ dependencies { The group ID is `com.guard-core` (the Central Portal namespace); Kotlin packages stay `com.guardcore.*`. +## Kotlin and coroutines versions under Spring Boot dependency management + +Spring Boot 3.5 manages `kotlin.version` 1.9.25, `kotlin-coroutines.version` 1.8.1 and `kotlin-serialization.version` 1.6.3, and dependency management overrides the versions this library was built with (Kotlin 2.4.21, coroutines 1.11.0). A Java application that imports the `spring-boot-dependencies` BOM, or uses the Spring Boot Gradle plugin with `io.spring.dependency-management`, therefore fails at startup with `NoSuchMethodError: kotlinx.coroutines.BuildersKt.runBlockingK$default` (coroutines 1.11.0 added `runBlockingK`), and with coroutines 1.11.0 on a Kotlin 1.9 stdlib every request fails with `ClassNotFoundException: kotlin.coroutines.jvm.internal.SpillingKt`. Align both versions. Serialization can stay on the managed 1.6.3. + +Gradle with the Spring Boot plugin and `io.spring.dependency-management`: + +```groovy +ext['kotlin.version'] = '2.4.21' +ext['kotlin-coroutines.version'] = '1.11.0' +``` + +Maven with the BOM imported (declare these imports before `spring-boot-dependencies`, the first declaration wins): + +```xml + + + + org.jetbrains.kotlinkotlin-bom + 2.4.21pomimport + + + org.jetbrains.kotlinxkotlinx-coroutines-bom + 1.11.0pomimport + + + + +``` + +With `spring-boot-starter-parent` as the parent, set `2.4.21` and `1.11.0` as properties instead. + ## Engine dependency (includeBuild + dependency substitution) Consumers resolve the engine artifact `com.guard-core:guard-core` from Maven Central. This repository's own build pins the engine from source instead: CI and diff --git a/src/main/kotlin/com/guardcore/spring/CommitHookResponse.kt b/src/main/kotlin/com/guardcore/spring/CommitHookResponse.kt new file mode 100644 index 0000000..836f861 --- /dev/null +++ b/src/main/kotlin/com/guardcore/spring/CommitHookResponse.kt @@ -0,0 +1,117 @@ +package com.guardcore.spring + +import jakarta.servlet.ServletOutputStream +import jakarta.servlet.WriteListener +import jakarta.servlet.http.HttpServletResponse +import jakarta.servlet.http.HttpServletResponseWrapper +import java.io.PrintWriter + +/** + * The unbuffered response flavor (the `PureASGISecurityMiddleware` response path of fastapi-guard, + * guard/asgi.py): the body streams straight to the container and [beforeCommit] runs exactly once, + * just before the first byte can leave (first write, flush, close, sendError, sendRedirect), the + * servlet analog of the `http.response.start` message. Headers set by the hook therefore still + * reach the client, and nothing is held in memory. + */ +internal class CommitHookResponse(response: HttpServletResponse) : + HttpServletResponseWrapper(response) { + var beforeCommit: (() -> Unit)? = null + private var fired = false + private var stream: ServletOutputStream? = null + private var writer: PrintWriter? = null + + /** Runs the hook once; safe to call from any commit point and again after the chain returns. */ + fun fireBeforeCommit() { + if (fired) { + return + } + fired = true + beforeCommit?.invoke() + } + + override fun getOutputStream(): ServletOutputStream = + stream ?: HookedStream(super.getOutputStream()).also { stream = it } + + override fun getWriter(): PrintWriter = + writer ?: HookedWriter(super.getWriter()).also { writer = it } + + override fun flushBuffer() { + fireBeforeCommit() + super.flushBuffer() + } + + // The status is stamped first so the hook (metrics, status behavior rules) sees the real one. + override fun sendError(sc: Int) { + super.setStatus(sc) + fireBeforeCommit() + super.sendError(sc) + } + + override fun sendError(sc: Int, msg: String?) { + super.setStatus(sc) + fireBeforeCommit() + super.sendError(sc, msg) + } + + override fun sendRedirect(location: String?) { + super.setStatus(HttpServletResponse.SC_FOUND) + fireBeforeCommit() + super.sendRedirect(location) + } + + private inner class HookedStream(private val delegate: ServletOutputStream) : + ServletOutputStream() { + override fun write(b: Int) { + fireBeforeCommit() + delegate.write(b) + } + + override fun write(b: ByteArray, off: Int, len: Int) { + fireBeforeCommit() + delegate.write(b, off, len) + } + + override fun flush() { + fireBeforeCommit() + delegate.flush() + } + + override fun close() { + fireBeforeCommit() + delegate.close() + } + + override fun isReady(): Boolean = delegate.isReady + + override fun setWriteListener(listener: WriteListener) { + delegate.setWriteListener(listener) + } + } + + private inner class HookedWriter(delegate: PrintWriter) : PrintWriter(delegate) { + override fun write(c: Int) { + fireBeforeCommit() + super.write(c) + } + + override fun write(buf: CharArray, off: Int, len: Int) { + fireBeforeCommit() + super.write(buf, off, len) + } + + override fun write(s: String, off: Int, len: Int) { + fireBeforeCommit() + super.write(s, off, len) + } + + override fun flush() { + fireBeforeCommit() + super.flush() + } + + override fun close() { + fireBeforeCommit() + super.close() + } + } +} diff --git a/src/main/kotlin/com/guardcore/spring/GuardFilter.kt b/src/main/kotlin/com/guardcore/spring/GuardFilter.kt index 0aa3aba..a55377b 100644 --- a/src/main/kotlin/com/guardcore/spring/GuardFilter.kt +++ b/src/main/kotlin/com/guardcore/spring/GuardFilter.kt @@ -45,14 +45,13 @@ public class GuardFilter( filterChain: FilterChain, ) { val guardRequestWrapper = GuardHttpServletRequest(request) - val responseWrapper = ContentCachingResponseWrapper(response) val guardRequest = ServletGuardRequest(guardRequestWrapper) // Python `_ensure_initialized` + the GuardRedisError 503 path (middleware.py:582-587). val initFailure = kotlinx.coroutines.runBlocking { middleware.ensureInitialized() } if (initFailure != null) { injectCorsHeaders(guardRequest.headers, initFailure) - finish(initFailure, responseWrapper, response) + finish(initFailure, response) return } @@ -63,7 +62,7 @@ public class GuardFilter( handlePreflight(guardRequest) } if (preflightResponse != null) { - finish(preflightResponse, responseWrapper, response) + finish(preflightResponse, response) return } @@ -74,33 +73,39 @@ public class GuardFilter( } if (passthroughResponse != null) { injectCorsHeaders(guardRequest.headers, passthroughResponse) - finish(passthroughResponse, responseWrapper, response) + finish(passthroughResponse, response) return } // Python dispatch: client IP -> route config -> security bypass (middleware.py:603-612). - val bypassStage: Pair? = + val bypassStage: Triple? = kotlinx.coroutines.runBlocking { val ip = middleware.extractClientIp(guardRequest) val routeConfig = middleware.routeResolver.getRouteConfig(guardRequest) + val downstream = downstreamResponseFor(response, routeConfig) + if (downstream is CommitHookResponse) { + downstream.beforeCommit = { + injectCorsHeadersIntoWrapper(guardRequest.headers, downstream) + } + } val bypass = middleware.bypassHandler.handleSecurityBypass( guardRequest, - { callNext(guardRequestWrapper, responseWrapper, filterChain) }, + { callNext(guardRequestWrapper, downstream, filterChain) }, routeConfig, ) if (bypass != null) { injectCorsHeaders(guardRequest.headers, bypass) - finish(bypass, responseWrapper, response) + finish(bypass, response, downstream) null } else { - ip to routeConfig + Triple(ip, routeConfig, downstream) } } if (bypassStage == null) { return } - val (clientIp, routeConfig) = bypassStage + val (clientIp, routeConfig, responseWrapper) = bypassStage // Python `_handle_pipeline_block` (middleware.py:560): the 17 fixed-order checks, through the // facade's blocking bridge (JavaBridges.executePipelineBlocking). @@ -110,7 +115,7 @@ public class GuardFilter( } if (blocked != null) { injectCorsHeaders(guardRequest.headers, blocked) - finish(blocked, responseWrapper, response) + finish(blocked, response) return } @@ -124,22 +129,93 @@ public class GuardFilter( // Python: `call_next` (middleware.py:624). val startTime = System.nanoTime() - try { + // Python `_process_response` + CORS injection (middleware.py:627-631). Buffered, it runs once + // the chain returns; streaming, just before the response commits (asgi.py http.response.start), + // so the body is never held and the status and headers are the ones that reach the client. + val processResponse = { + val responseTime = (System.nanoTime() - startTime) / 1_000_000_000.0 + val guardResponse = ServletGuardResponse.wrapping(responseWrapper) + runGuardStage { + middleware.processResponse(guardRequest, guardResponse, responseTime, routeConfig) + null + } + injectCorsHeadersIntoWrapper(guardRequest.headers, responseWrapper) + } + if (responseWrapper is CommitHookResponse) { + responseWrapper.beforeCommit = processResponse filterChain.doFilter(guardRequestWrapper, responseWrapper) + completeStreamingResponse(guardRequestWrapper, responseWrapper) + return + } + val buffered = responseWrapper as ContentCachingResponseWrapper + try { + filterChain.doFilter(guardRequestWrapper, buffered) } catch (e: Exception) { - runCatching { responseWrapper.copyBodyToResponse() } + runCatching { buffered.copyBodyToResponse() } throw e } - val responseTime = (System.nanoTime() - startTime) / 1_000_000_000.0 + processResponse() + buffered.copyBodyToResponse() + } + + /** + * Buffers only when a feature reads the response body: a `custom_response_modifier` (it receives + * the body) or a return-pattern rule that is not status based (processor + BehaviorTracker read + * the body prefix). Everything else (security headers, CORS, metrics, status rules) needs only + * the status and headers, which the streaming flavor supplies without holding the body. + */ + private fun downstreamResponseFor( + response: HttpServletResponse, + routeConfig: RouteConfig?, + ): HttpServletResponse = + if (needsResponseBody(routeConfig)) ContentCachingResponseWrapper(response) + else CommitHookResponse(response) + + private fun needsResponseBody(routeConfig: RouteConfig?): Boolean { + val config = middleware.config + if (config.customResponseModifier != null) { + return true + } + if (!config.behaviorScanResponseBody) { + return false + } + val routePatterns = + routeConfig + ?.behaviorRules + .orEmpty() + .filter { it.ruleType == RETURN_PATTERN } + .map { it.pattern } + val globalPatterns = + config.globalBehaviorRules + .filter { it.ruleType.pythonName == RETURN_PATTERN } + .map { it.pattern } + return (routePatterns + globalPatterns).any { it != null && !it.startsWith(STATUS_PATTERN) } + } - // Python `_process_response` + CORS injection (middleware.py:627-631). - val guardResponse = ServletGuardResponse.wrapping(responseWrapper) - runGuardStage { - middleware.processResponse(guardRequest, guardResponse, responseTime, routeConfig) - null + /** + * Fires the commit hook for responses that never wrote or flushed, and for async handlers + * (StreamingResponseBody, SseEmitter, DeferredResult) arms it on completion; their writes arrive + * later through the same wrapper and fire it at the first byte. + */ + private fun completeStreamingResponse( + request: HttpServletRequest, + response: CommitHookResponse, + ) { + if (!request.isAsyncStarted) { + response.fireBeforeCommit() + return } - injectCorsHeadersIntoWrapper(guardRequest.headers, responseWrapper) - responseWrapper.copyBodyToResponse() + request.asyncContext.addListener( + object : jakarta.servlet.AsyncListener { + override fun onComplete(event: jakarta.servlet.AsyncEvent) = response.fireBeforeCommit() + + override fun onTimeout(event: jakarta.servlet.AsyncEvent) = response.fireBeforeCommit() + + override fun onError(event: jakarta.servlet.AsyncEvent) = response.fireBeforeCommit() + + override fun onStartAsync(event: jakarta.servlet.AsyncEvent) = Unit + }, + ) } /** Python `_handle_preflight` (middleware.py:514-527). */ @@ -163,7 +239,7 @@ public class GuardFilter( /** The wrapped call_next (middleware.py:590 `wrap_call_next`): runs the downstream chain. */ private fun callNext( request: GuardHttpServletRequest, - responseWrapper: ContentCachingResponseWrapper, + responseWrapper: HttpServletResponse, filterChain: FilterChain, ): GuardResponse { filterChain.doFilter(request, responseWrapper) @@ -229,7 +305,7 @@ public class GuardFilter( /** Python `_inject_cors_headers` over the buffered downstream response. */ private fun injectCorsHeadersIntoWrapper( requestHeaders: Map, - wrapper: ContentCachingResponseWrapper, + wrapper: HttpServletResponse, ) { val corsHandler = middleware.corsHandler ?: return val corsHeaders = corsHandler.buildResponseHeaders(requestHeaders) @@ -240,18 +316,26 @@ public class GuardFilter( /** * The `unwrap_response` analog (adapters.py:189): a facade-created response is written to the raw - * servlet response; a wrapper-backed response (security-bypass path) is already buffered in the - * wrapper and only needs its body flushed. + * servlet response; a wrapper-backed response (security-bypass path) was produced by the chain + * and only needs its buffered body copied, or its commit hook fired when it streamed. */ private fun finish( response: GuardResponse, - responseWrapper: ContentCachingResponseWrapper, servletResponse: HttpServletResponse, + downstream: HttpServletResponse? = null, ) { if (response is ServletGuardResponse && response.wrapped != null) { - responseWrapper.copyBodyToResponse() + when (downstream) { + is ContentCachingResponseWrapper -> downstream.copyBodyToResponse() + is CommitHookResponse -> downstream.fireBeforeCommit() + } return } writeGuardResponse(response, servletResponse) } + + private companion object { + const val RETURN_PATTERN = "return_pattern" + const val STATUS_PATTERN = "status:" + } } diff --git a/src/main/kotlin/com/guardcore/spring/ServletGuardResponse.kt b/src/main/kotlin/com/guardcore/spring/ServletGuardResponse.kt index 0683b49..861ad9a 100644 --- a/src/main/kotlin/com/guardcore/spring/ServletGuardResponse.kt +++ b/src/main/kotlin/com/guardcore/spring/ServletGuardResponse.kt @@ -1,5 +1,6 @@ package com.guardcore.spring +import com.guardcore.protocols.BoundedResponseBodyReader import com.guardcore.protocols.GuardResponse import com.guardcore.protocols.GuardResponseFactory import com.guardcore.protocols.MiddlewareResponseFactoryProtocol @@ -12,39 +13,50 @@ import org.springframework.web.util.ContentCachingResponseWrapper * the facade's [GuardResponse] protocol, plus the plain factory the facade's response factory and * checks drive. * - * The wrapper-backed flavor writes header mutations straight onto the buffered servlet response - * (ContentCachingResponseWrapper keeps the real response uncommitted until copyBodyToResponse, so - * the post-handler processResponse pass can still apply security headers and CORS, exactly the - * envelope Starlette's BaseHTTPMiddleware gives fastapi-guard). + * The wrapper-backed flavor writes header mutations straight onto the downstream servlet response. + * Buffered (ContentCachingResponseWrapper) it stays uncommitted until copyBodyToResponse, so the + * post-handler processResponse pass can still apply security headers and CORS (the envelope + * Starlette's BaseHTTPMiddleware gives fastapi-guard). Streaming (CommitHookResponse) the same pass + * runs just before the response commits and [body] is null (the PureASGI envelope). */ public class ServletGuardResponse private constructor( override val statusCode: Int, override var body: ByteArray?, override val headers: MutableMap, - /** Non-null when this response wraps the buffered downstream servlet response. */ - internal val wrapped: ContentCachingResponseWrapper?, -) : GuardResponse { + /** Non-null when this response wraps the downstream servlet response. */ + internal val wrapped: HttpServletResponse?, +) : GuardResponse, BoundedResponseBodyReader { public constructor( statusCode: Int, body: ByteArray?, headers: Map = linkedMapOf(), ) : this(statusCode, body, linkedMapOf().apply { putAll(headers) }, null) + /** The bounded prefix the return-pattern rules inspect; empty when the body is not buffered. */ + override suspend fun readBodyPrefix(maxBytes: Int): ByteArray { + val buffered = body ?: return ByteArray(0) + return buffered.copyOf(minOf(maxBytes, buffered.size)) + } + public companion object { - /** Wraps a buffered downstream response (the StarletteGuardResponse analog). */ + /** Kept for callers linked against 1.1.0 (the JVM signature took the caching wrapper). */ public fun wrapping(wrapper: ContentCachingResponseWrapper): ServletGuardResponse = + wrapping(wrapper as HttpServletResponse) + + /** Wraps the downstream response (the StarletteGuardResponse analog). */ + public fun wrapping(wrapper: HttpServletResponse): ServletGuardResponse = ServletGuardResponse( wrapper.status, - wrapper.contentAsByteArray, + (wrapper as? ContentCachingResponseWrapper)?.contentAsByteArray, WrapperHeaders(wrapper), wrapper, ) } } -/** Mutable header view over the buffered servlet response (case-insensitive reads). */ -private class WrapperHeaders(private val wrapper: ContentCachingResponseWrapper) : +/** Mutable header view over the downstream servlet response (case-insensitive reads). */ +private class WrapperHeaders(private val wrapper: HttpServletResponse) : MutableMap { override val size: Int get() = wrapper.headerNames.size diff --git a/src/test/kotlin/com/guardcore/spring/GuardStreamingResponseTest.kt b/src/test/kotlin/com/guardcore/spring/GuardStreamingResponseTest.kt new file mode 100644 index 0000000..e22f925 --- /dev/null +++ b/src/test/kotlin/com/guardcore/spring/GuardStreamingResponseTest.kt @@ -0,0 +1,235 @@ +package com.guardcore.spring + +import jakarta.servlet.Filter +import jakarta.servlet.FilterChain +import jakarta.servlet.ServletRequest +import jakarta.servlet.ServletResponse +import jakarta.servlet.http.HttpServletRequest +import java.io.BufferedReader +import java.io.InputStreamReader +import java.net.URI +import java.net.http.HttpClient +import java.net.http.HttpRequest +import java.net.http.HttpResponse +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import org.assertj.core.api.Assertions.assertThat +import org.junit.jupiter.api.Test +import org.springframework.beans.factory.annotation.Value +import org.springframework.boot.SpringBootConfiguration +import org.springframework.boot.autoconfigure.EnableAutoConfiguration +import org.springframework.boot.test.context.SpringBootTest +import org.springframework.boot.web.servlet.FilterRegistrationBean +import org.springframework.context.annotation.Bean +import org.springframework.core.Ordered +import org.springframework.http.HttpStatus +import org.springframework.http.ResponseEntity +import org.springframework.test.context.TestPropertySource +import org.springframework.web.bind.annotation.GetMapping +import org.springframework.web.bind.annotation.RestController +import org.springframework.web.context.request.async.DeferredResult +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter +import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody +import org.springframework.web.util.ContentCachingResponseWrapper + +private const val CHUNK_BYTES = 1024 * 1024 +private const val CHUNKS = 64 + +/** Records which response flavor the downstream chain received, per request path. */ +public object ResponseKindProbe { + public val kinds: ConcurrentHashMap = ConcurrentHashMap() + public val release: CountDownLatch = CountDownLatch(1) + public val sseRelease: CountDownLatch = CountDownLatch(1) +} + +@RestController +public class StreamingTestController { + @GetMapping("/s/big") + public fun big(): StreamingResponseBody = StreamingResponseBody { out -> + val chunk = ByteArray(CHUNK_BYTES) { 'x'.code.toByte() } + out.write(chunk) + out.flush() + ResponseKindProbe.release.await(60, TimeUnit.SECONDS) + repeat(CHUNKS - 1) { out.write(chunk) } + } + + @GetMapping("/s/sse") + public fun sse(): SseEmitter { + val emitter = SseEmitter(0L) + Thread { + emitter.send(SseEmitter.event().name("tick").data("one")) + ResponseKindProbe.sseRelease.await(60, TimeUnit.SECONDS) + emitter.send(SseEmitter.event().name("tick").data("two")) + emitter.complete() + } + .start() + return emitter + } + + @GetMapping("/s/deferred") + public fun deferred(): DeferredResult> { + val result = DeferredResult>() + Thread { result.setResult(mapOf("deferred" to "true")) }.start() + return result + } + + @GetMapping("/s/json") public fun json(): Map = mapOf("ok" to "true") + + @GetMapping("/s/teapot") + @GuardReturnMonitor(pattern = "status:418", maxOccurrences = 1, window = 3600, action = "ban") + public fun teapot(): ResponseEntity = + ResponseEntity.status(HttpStatus.I_AM_A_TEAPOT).body("short and stout") + + @GetMapping("/s/leaky") + @GuardReturnMonitor(pattern = "leaked-token", maxOccurrences = 1, window = 3600, action = "ban") + public fun leaky(): String = "here is a " + listOf("leaked", "token").joinToString("-") + + @GetMapping("/s/redirect") + public fun redirect(response: jakarta.servlet.http.HttpServletResponse) { + response.sendRedirect("/s/json") + } +} + +@SpringBootConfiguration +@EnableAutoConfiguration +@org.springframework.context.annotation.Import(StreamingTestController::class) +public open class StreamingGuardTestApp { + @Bean + public open fun responseKindProbe(): FilterRegistrationBean = + FilterRegistrationBean( + object : Filter { + override fun doFilter( + req: ServletRequest, + res: ServletResponse, + chain: FilterChain, + ) { + val path = (req as HttpServletRequest).requestURI + ResponseKindProbe.kinds[path] = + if (res is ContentCachingResponseWrapper) "buffered" + else res.javaClass.simpleName + chain.doFilter(req, res) + } + }, + ) + .apply { order = Ordered.LOWEST_PRECEDENCE } +} + +@SpringBootTest( + classes = [StreamingGuardTestApp::class], + webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, +) +@TestPropertySource( + properties = + [ + "guard.enable-redis=false", + "guard.behavior-scan-response-body=true", + "guard.enable-cors=true", + "guard.trusted-proxies[0]=127.0.0.1", + "guard.cors-allow-origins[0]=https://app.example", + ], +) +public class GuardStreamingResponseTest { + @Value("\${local.server.port}") public var port: Int = 0 + + private val client = HttpClient.newHttpClient() + + private fun request(path: String, ip: String = "198.51.100.7"): HttpRequest = + HttpRequest.newBuilder(URI("http://127.0.0.1:$port$path")) + .header("Origin", "https://app.example") + .header("X-Forwarded-For", ip) + .GET() + .build() + + private fun HttpResponse<*>.assertGuardHeaders() { + assertThat(headers().firstValue("X-Content-Type-Options")).hasValue("nosniff") + assertThat(headers().firstValue("Access-Control-Allow-Origin")).hasValue("https://app.example") + } + + @Test + public fun `a 64 MB streamed body is never buffered and the first chunk arrives before the handler finishes`() { + val response = client.send(request("/s/big"), HttpResponse.BodyHandlers.ofInputStream()) + assertThat(ResponseKindProbe.kinds["/s/big"]).isEqualTo("CommitHookResponse") + response.assertGuardHeaders() + response.body().use { body -> + val first = body.readNBytes(CHUNK_BYTES) + assertThat(first).hasSize(CHUNK_BYTES) + // The handler is still parked on the latch: the first chunk reached the client before it + // finished, which a buffering wrapper cannot do. + assertThat(ResponseKindProbe.release.count).isEqualTo(1) + ResponseKindProbe.release.countDown() + var total = first.size.toLong() + val sink = ByteArray(CHUNK_BYTES) + while (true) { + val n = body.read(sink) + if (n < 0) break + total += n + } + assertThat(total).isEqualTo(CHUNK_BYTES.toLong() * CHUNKS) + } + } + + @Test + public fun `server-sent events arrive incrementally with the guard headers`() { + val response = client.send(request("/s/sse"), HttpResponse.BodyHandlers.ofInputStream()) + assertThat(ResponseKindProbe.kinds["/s/sse"]).isEqualTo("CommitHookResponse") + response.assertGuardHeaders() + BufferedReader(InputStreamReader(response.body())).use { reader -> + assertThat(reader.readLine()).isEqualTo("event:tick") + assertThat(reader.readLine()).isEqualTo("data:one") + ResponseKindProbe.sseRelease.countDown() + reader.readLine() + assertThat(reader.readLine()).isEqualTo("event:tick") + assertThat(reader.readLine()).isEqualTo("data:two") + } + } + + @Test + public fun `a deferred result completes through the same unbuffered wrapper with guard headers`() { + val response = client.send(request("/s/deferred"), HttpResponse.BodyHandlers.ofString()) + assertThat(response.statusCode()).isEqualTo(200) + assertThat(response.body()).contains("deferred") + response.assertGuardHeaders() + } + + @Test + public fun `a plain json response streams unbuffered with security headers`() { + val response = client.send(request("/s/json"), HttpResponse.BodyHandlers.ofString()) + assertThat(ResponseKindProbe.kinds["/s/json"]).isEqualTo("CommitHookResponse") + assertThat(response.body()).contains("ok") + response.assertGuardHeaders() + } + + @Test + public fun `a redirect carries the guard headers`() { + val noFollow = HttpClient.newBuilder().followRedirects(HttpClient.Redirect.NEVER).build() + val response = noFollow.send(request("/s/redirect"), HttpResponse.BodyHandlers.ofString()) + assertThat(response.statusCode()).isEqualTo(302) + response.assertGuardHeaders() + } + + @Test + public fun `a status return pattern bans without buffering the response`() { + val ip = "198.51.100.21" + val first = client.send(request("/s/teapot", ip), HttpResponse.BodyHandlers.ofString()) + assertThat(first.statusCode()).isEqualTo(418) + assertThat(ResponseKindProbe.kinds["/s/teapot"]).isEqualTo("CommitHookResponse") + // The tracker trips when the count exceeds the threshold, so the second hit bans. + client.send(request("/s/teapot", ip), HttpResponse.BodyHandlers.ofString()) + val banned = client.send(request("/s/teapot", ip), HttpResponse.BodyHandlers.ofString()) + assertThat(banned.statusCode()).isEqualTo(403) + } + + @Test + public fun `a body return pattern buffers only that route and still bans`() { + val ip = "198.51.100.22" + val first = client.send(request("/s/leaky", ip), HttpResponse.BodyHandlers.ofString()) + assertThat(first.statusCode()).isEqualTo(200) + assertThat(first.body()).isEqualTo("here is a leaked-token") + first.assertGuardHeaders() + assertThat(ResponseKindProbe.kinds["/s/leaky"]).isEqualTo("buffered") + client.send(request("/s/leaky", ip), HttpResponse.BodyHandlers.ofString()) + val banned = client.send(request("/s/leaky", ip), HttpResponse.BodyHandlers.ofString()) + assertThat(banned.statusCode()).isEqualTo(403) + } +}