From d63343ac0ee95126502d3052dc53a4b776cebd34 Mon Sep 17 00:00:00 2001 From: Chandrahas Reddy Pola Date: Mon, 20 Jul 2026 15:57:36 -0700 Subject: [PATCH 1/8] Implemented the poll chain to await the download future Signed-off-by: Chandrahas Reddy Pola --- .../opa/plugins/BundleDownloader.java | 186 ++++++++++-------- .../plugins/BundleDownloaderSecurityTest.java | 105 ++++++++++ 2 files changed, 211 insertions(+), 80 deletions(-) diff --git a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java index 99af2444..a84ea67a 100644 --- a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java +++ b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java @@ -189,48 +189,60 @@ public CompletableFuture startPolling(ScheduledExecutorService scheduler) ? polling.getMaxDelaySeconds() : 120; - scheduler.schedule(this::downloadBundle, 0, TimeUnit.SECONDS); - scheduleNextPoll(scheduler, minDelay, maxDelay); + // Run the first download immediately, then chain each subsequent poll only after the previous + // download AND its (possibly async) activation have fully completed. Chaining off completion + // instead of a fixed timer keeps polling strictly serial, so overlapping downloads can never + // run activateBundle() concurrently against a shared Store (issue #113). + scheduleDownload(scheduler, minDelay, maxDelay, 0); return initialActivation; } - // Re-schedules the next download with a uniformly random delay in [minDelay, maxDelay], - // matching Go-OPA's jittered polling. ScheduledExecutorService has no built-in jitter, so the - // task chains itself. RejectedExecutionException after a shutdown breaks the chain cleanly. - private void scheduleNextPoll(ScheduledExecutorService scheduler, int minDelay, int maxDelay) { - long delay = - minDelay >= maxDelay - ? minDelay - : ThreadLocalRandom.current().nextLong(minDelay, (long) maxDelay + 1); + // Schedules one download after delaySeconds; once that download (including any async activation) + // completes, schedules the next poll with a fresh jittered delay in [minDelay, maxDelay]. Because + // the next poll is chained off the download's completion — not fired on an independent timer — + // at most one download is ever in flight, mirroring Go-OPA's strictly serial poll loop. + // ScheduledExecutorService has no built-in jitter, so the task chains itself; a + // RejectedExecutionException after shutdown ends the chain cleanly. + private void scheduleDownload( + ScheduledExecutorService scheduler, int minDelay, int maxDelay, long delaySeconds) { try { scheduler.schedule( - () -> { - try { - downloadBundle(); - } catch (Exception e) { - // downloadBundle() handles its own logging; swallow so the chain keeps polling. - // Only Exception is caught here — Errors (OOM, etc.) propagate and let the - // executor's uncaught-exception handler tear down the pool, which is the right - // outcome for unrecoverable conditions. - } finally { - scheduleNextPoll(scheduler, minDelay, maxDelay); - } - }, - delay, + () -> + downloadBundle() + .whenComplete( + (result, throwable) -> + scheduleDownload( + scheduler, + minDelay, + maxDelay, + nextDelaySeconds(minDelay, maxDelay))), + delaySeconds, TimeUnit.SECONDS); } catch (RejectedExecutionException stopped) { // Scheduler was shut down; let the chain end. } } + private static long nextDelaySeconds(int minDelay, int maxDelay) { + return minDelay >= maxDelay + ? minDelay + : ThreadLocalRandom.current().nextLong(minDelay, (long) maxDelay + 1); + } + /** * Download the bundle from the configured service and resource. * *

Handles HTTP/HTTPS downloads with ETag caching, file:// URIs, and filesystem paths. Calls * {@link #activateBundle(byte[])} when new bundle data is available. + * + * @return a future that completes when this download attempt — including any asynchronous + * activation — has finished. The poll scheduler awaits it before scheduling the next poll, so + * downloads (and therefore {@code activateBundle} calls) never overlap. The future completes + * normally on both success and handled failure so that polling continues; the actual outcome + * is recorded on {@code initialActivation}. */ - protected void downloadBundle() { + protected CompletableFuture downloadBundle() { try { Config.ServiceConfig serviceConfig = manager.getConfig().getService(service); if (serviceConfig == null) { @@ -239,7 +251,7 @@ protected void downloadBundle() { initialActivation.completeExceptionally( new RuntimeException("Service '" + service + "' not found")); } - return; + return CompletableFuture.completedFuture(null); } String baseUrl = serviceConfig.getUrl(); @@ -257,16 +269,17 @@ protected void downloadBundle() { // Handle file:// URIs if ("file".equalsIgnoreCase(uri.getScheme())) { handleFileDownload(Paths.get(uri)); - return; + return CompletableFuture.completedFuture(null); } // Handle HTTP/HTTPS URIs - handleHttpDownload(uri); + return handleHttpDownload(uri); } else { // It's a file path (relative or absolute) Path basePath = Paths.get(baseUrl); Path filePath = basePath.resolve(resource); handleFileDownload(filePath); + return CompletableFuture.completedFuture(null); } } catch (Exception e) { @@ -274,6 +287,7 @@ protected void downloadBundle() { if (!initialActivation.isDone()) { initialActivation.completeExceptionally(e); } + return CompletableFuture.completedFuture(null); } } @@ -307,8 +321,12 @@ private void handleFileDownload(Path filePath) throws IOException { * Handle downloading from an HTTP/HTTPS URI. * * @param uri the URI to download from + * @return a future that completes when the response has been handled (including activation). It + * completes normally regardless of the outcome — {@code handle} turns both success and + * failure into a completed stage — so the poll chain always resumes; the actual outcome is + * recorded on {@code initialActivation}. */ - private void handleHttpDownload(URI uri) { + private CompletableFuture handleHttpDownload(URI uri) { HttpRequest.Builder requestBuilder = HttpRequest.newBuilder() .uri(uri) @@ -325,63 +343,71 @@ private void handleHttpDownload(URI uri) { } HttpRequest request = requestBuilder.build(); - httpClient + return httpClient .sendAsync(request, sizeLimitedBodyHandler(maxSizeBytes)) - .whenComplete( + .handle( (response, throwable) -> { - if (throwable != null) { - Throwable cause = - throwable instanceof java.util.concurrent.CompletionException - ? throwable.getCause() - : throwable; - manager.getLogger().error("Bundle '%s': Download error: %s", name, cause.getMessage()); - if (!initialActivation.isDone()) { - initialActivation.completeExceptionally(cause); - } - return; - } - - if (response.statusCode() == 304) { - manager.getLogger().debug("Bundle '%s': Not modified (ETag match)", name); - if (!initialActivation.isDone()) { - initialActivation.complete(null); - } - return; - } - - if (response.statusCode() == 200) { - String contentType = response.headers().firstValue("Content-Type").orElse(""); - if (!isAcceptableContentType(contentType)) { - String errorMsg = "Unexpected Content-Type: '" + contentType + "'"; - manager.getLogger().error("Bundle '%s': %s", name, errorMsg); - if (!initialActivation.isDone()) { - initialActivation.completeExceptionally(new RuntimeException(errorMsg)); - } - return; - } - response.headers().firstValue("ETag").ifPresent(newEtag -> this.etag = newEtag); - try { - activateBundle(response.body()); - } catch (Exception e) { - manager.getLogger().error("Bundle '%s': Activation failed: %s", name, e.getMessage()); - if (!initialActivation.isDone()) { - initialActivation.completeExceptionally(e); - } - return; - } - if (!initialActivation.isDone()) { - initialActivation.complete(null); - } - } else { - String errorMsg = "Download failed with status " + response.statusCode(); - manager.getLogger().error("Bundle '%s': %s", name, errorMsg); - if (!initialActivation.isDone()) { - initialActivation.completeExceptionally(new RuntimeException(errorMsg)); - } - } + handleHttpResponse(response, throwable); + return null; }); } + // Handles a completed HTTP exchange: records success or failure on initialActivation. Runs on the + // HTTP client's executor thread, but the poll chain never starts the next download until the + // future returned by handleHttpDownload completes, so activateBundle is never called concurrently. + private void handleHttpResponse(HttpResponse response, Throwable throwable) { + if (throwable != null) { + Throwable cause = + throwable instanceof java.util.concurrent.CompletionException + ? throwable.getCause() + : throwable; + manager.getLogger().error("Bundle '%s': Download error: %s", name, cause.getMessage()); + if (!initialActivation.isDone()) { + initialActivation.completeExceptionally(cause); + } + return; + } + + if (response.statusCode() == 304) { + manager.getLogger().debug("Bundle '%s': Not modified (ETag match)", name); + if (!initialActivation.isDone()) { + initialActivation.complete(null); + } + return; + } + + if (response.statusCode() == 200) { + String contentType = response.headers().firstValue("Content-Type").orElse(""); + if (!isAcceptableContentType(contentType)) { + String errorMsg = "Unexpected Content-Type: '" + contentType + "'"; + manager.getLogger().error("Bundle '%s': %s", name, errorMsg); + if (!initialActivation.isDone()) { + initialActivation.completeExceptionally(new RuntimeException(errorMsg)); + } + return; + } + response.headers().firstValue("ETag").ifPresent(newEtag -> this.etag = newEtag); + try { + activateBundle(response.body()); + } catch (Exception e) { + manager.getLogger().error("Bundle '%s': Activation failed: %s", name, e.getMessage()); + if (!initialActivation.isDone()) { + initialActivation.completeExceptionally(e); + } + return; + } + if (!initialActivation.isDone()) { + initialActivation.complete(null); + } + } else { + String errorMsg = "Download failed with status " + response.statusCode(); + manager.getLogger().error("Bundle '%s': %s", name, errorMsg); + if (!initialActivation.isDone()) { + initialActivation.completeExceptionally(new RuntimeException(errorMsg)); + } + } + } + /** * Returns true if the response Content-Type is acceptable for a bundle. * A blank/absent value is tolerated. diff --git a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java index 7c0c852e..c1603319 100644 --- a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java +++ b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java @@ -1,5 +1,6 @@ package io.github.open_policy_agent.opa.plugins; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -17,7 +18,11 @@ import java.util.Collections; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.zip.GZIPOutputStream; import org.apache.commons.compress.archivers.tar.TarArchiveEntry; import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream; @@ -32,12 +37,20 @@ class BundleDownloaderSecurityTest { private HttpServer server; private Store store; + private ExecutorService serverExecutor; + private ScheduledExecutorService scheduler; @AfterEach void tearDown() { + if (scheduler != null) { + scheduler.shutdownNow(); + } if (server != null) { server.stop(0); } + if (serverExecutor != null) { + serverExecutor.shutdownNow(); + } } @Test @@ -293,6 +306,98 @@ private CompletableFuture runBundleDownload(int port, long maxSizeBytes) { .whenComplete((v, t) -> bundlePlugin.stop()); } + @Test + void polling_slowActivationExceedingInterval_neverRunsConcurrently() throws Exception { + // Regression test for issue #113: the async download refactor let overlapping polls run + // activateBundle() concurrently against a shared Store. Poll with a zero-second interval while + // activation is deliberately slow, so polls would pile up if they were not serialized. The fix + // chains each poll off the previous download's completion, so only one activation runs at a + // time. (With the old fixed-timer scheduling, maxInFlight would exceed 1.) + byte[] bundleData = createValidBundle(); + + // Multi-threaded server executor so several downloads can be in flight at once — this is what + // would let activations overlap if polling were not serialized. + serverExecutor = Executors.newFixedThreadPool(8); + server = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + server.setExecutor(serverExecutor); + server.createContext("/bundle.tar.gz", exchange -> { + exchange.getResponseHeaders().add("Content-Type", "application/vnd.openpolicyagent.bundles"); + exchange.sendResponseHeaders(200, bundleData.length); + exchange.getResponseBody().write(bundleData); + exchange.close(); + }); + server.start(); + + Config config = new Config(); + config.setServices(Collections.singletonMap("test-service", + new Config.ServiceConfig() + .setName("test-service") + .setUrl("http://localhost:" + server.getAddress().getPort()))); + + Logger logger = new Logger.StandardLogger(); + PluginManager manager = + new PluginManager.Builder() + .withId("test") + .withStore(new InMem()) + .withConfig(config) + .withLogger(logger) + .build(); + + ConcurrencyTrackingDownloader downloader = + new ConcurrencyTrackingDownloader("authz", manager); + downloader + .setService("test-service") + .setResource("/bundle.tar.gz") + .setPolling(new Config.PollingConfig().setMinDelaySeconds(0).setMaxDelaySeconds(0)); + + scheduler = BundleDownloader.newPollScheduler("concurrency-test"); + downloader.startPolling(scheduler); + + // Let many poll cycles run back-to-back. + Thread.sleep(800); + scheduler.shutdownNow(); + scheduler.awaitTermination(2, TimeUnit.SECONDS); + + assertTrue( + downloader.activations.get() > 1, + "expected multiple poll cycles to have run, but only " + downloader.activations.get() + + " did"); + assertEquals( + 1, + downloader.maxInFlight.get(), + "activateBundle ran concurrently — max simultaneous activations was " + + downloader.maxInFlight.get()); + } + + /** + * A BundleDownloader whose activation is deliberately slow and records whether any two + * activations ever overlap, so a test can assert that polling stays serialized. + */ + private static final class ConcurrencyTrackingDownloader extends BundleDownloader { + final AtomicInteger inFlight = new AtomicInteger(); + final AtomicInteger maxInFlight = new AtomicInteger(); + final AtomicInteger activations = new AtomicInteger(); + + ConcurrencyTrackingDownloader(String name, PluginManager manager) { + super(name, manager, null); + } + + @Override + protected void activateBundle(byte[] bundleData) { + int now = inFlight.incrementAndGet(); + maxInFlight.accumulateAndGet(now, Math::max); + try { + // Slower than the (zero-second) poll interval, so unserialized polls would overlap here. + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + inFlight.decrementAndGet(); + activations.incrementAndGet(); + } + } + } + @Test void download_chunkedBodyExceedsLimit_failsMidStream() throws Exception { int chunkSize = 64 * 1024; // 64 KB per chunk From 406aa0183bd8d5bb362b04ad6b24114b84ba4d71 Mon Sep 17 00:00:00 2001 From: Chandrahas Reddy Pola Date: Tue, 21 Jul 2026 18:26:47 -0700 Subject: [PATCH 2/8] Removed Code Smells. Signed-off-by: Chandrahas Reddy Pola --- .../opa/plugins/BundleDownloader.java | 20 ++++++------------- .../plugins/BundleDownloaderSecurityTest.java | 9 +-------- 2 files changed, 7 insertions(+), 22 deletions(-) diff --git a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java index a84ea67a..be5a6210 100644 --- a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java +++ b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java @@ -189,19 +189,15 @@ public CompletableFuture startPolling(ScheduledExecutorService scheduler) ? polling.getMaxDelaySeconds() : 120; - // Run the first download immediately, then chain each subsequent poll only after the previous - // download AND its (possibly async) activation have fully completed. Chaining off completion - // instead of a fixed timer keeps polling strictly serial, so overlapping downloads can never - // run activateBundle() concurrently against a shared Store (issue #113). scheduleDownload(scheduler, minDelay, maxDelay, 0); return initialActivation; } - // Schedules one download after delaySeconds; once that download (including any async activation) - // completes, schedules the next poll with a fresh jittered delay in [minDelay, maxDelay]. Because - // the next poll is chained off the download's completion — not fired on an independent timer — - // at most one download is ever in flight, mirroring Go-OPA's strictly serial poll loop. + // Schedules one download after delaySeconds; + // once that download completes, schedules the next poll with a fresh jittered delay in [minDelay, maxDelay]. + // Because the next poll is chained off the download's completion — not fired on an independent timer — + // at most one download is ever in flight, matching Go-OPA's jittered polling. // ScheduledExecutorService has no built-in jitter, so the task chains itself; a // RejectedExecutionException after shutdown ends the chain cleanly. private void scheduleDownload( @@ -238,9 +234,7 @@ private static long nextDelaySeconds(int minDelay, int maxDelay) { * * @return a future that completes when this download attempt — including any asynchronous * activation — has finished. The poll scheduler awaits it before scheduling the next poll, so - * downloads (and therefore {@code activateBundle} calls) never overlap. The future completes - * normally on both success and handled failure so that polling continues; the actual outcome - * is recorded on {@code initialActivation}. + * downloads never overlap. */ protected CompletableFuture downloadBundle() { try { @@ -321,9 +315,7 @@ private void handleFileDownload(Path filePath) throws IOException { * Handle downloading from an HTTP/HTTPS URI. * * @param uri the URI to download from - * @return a future that completes when the response has been handled (including activation). It - * completes normally regardless of the outcome — {@code handle} turns both success and - * failure into a completed stage — so the poll chain always resumes; the actual outcome is + * @return a future that completes when the response has been handled. The actual outcome is * recorded on {@code initialActivation}. */ private CompletableFuture handleHttpDownload(URI uri) { diff --git a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java index c1603319..60a1edbe 100644 --- a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java +++ b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java @@ -308,15 +308,9 @@ private CompletableFuture runBundleDownload(int port, long maxSizeBytes) { @Test void polling_slowActivationExceedingInterval_neverRunsConcurrently() throws Exception { - // Regression test for issue #113: the async download refactor let overlapping polls run - // activateBundle() concurrently against a shared Store. Poll with a zero-second interval while - // activation is deliberately slow, so polls would pile up if they were not serialized. The fix - // chains each poll off the previous download's completion, so only one activation runs at a - // time. (With the old fixed-timer scheduling, maxInFlight would exceed 1.) + byte[] bundleData = createValidBundle(); - // Multi-threaded server executor so several downloads can be in flight at once — this is what - // would let activations overlap if polling were not serialized. serverExecutor = Executors.newFixedThreadPool(8); server = HttpServer.create(new InetSocketAddress("localhost", 0), 0); server.setExecutor(serverExecutor); @@ -387,7 +381,6 @@ protected void activateBundle(byte[] bundleData) { int now = inFlight.incrementAndGet(); maxInFlight.accumulateAndGet(now, Math::max); try { - // Slower than the (zero-second) poll interval, so unserialized polls would overlap here. Thread.sleep(50); } catch (InterruptedException e) { Thread.currentThread().interrupt(); From 98ede2f42d788675b86f4d52619731877b3709e3 Mon Sep 17 00:00:00 2001 From: Chandrahas Reddy Pola Date: Thu, 23 Jul 2026 23:27:13 -0700 Subject: [PATCH 3/8] Minor placement changes Signed-off-by: Chandrahas Reddy Pola --- .../plugins/BundleDownloaderSecurityTest.java | 31 +++++++++---------- 1 file changed, 14 insertions(+), 17 deletions(-) diff --git a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java index 60a1edbe..52787b2f 100644 --- a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java +++ b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java @@ -307,8 +307,20 @@ private CompletableFuture runBundleDownload(int port, long maxSizeBytes) { } @Test - void polling_slowActivationExceedingInterval_neverRunsConcurrently() throws Exception { + void download_chunkedBodyExceedsLimit_failsMidStream() throws Exception { + int chunkSize = 64 * 1024; // 64 KB per chunk + int totalChunks = 20; // 1.28 MB total + long limit = 10 * 1024; // 10 KB limit + + int port = startRawChunkedServer(chunkSize, totalChunks); + ExecutionException ex = assertThrows(ExecutionException.class, + () -> runBundleDownload(port, limit).get(10, TimeUnit.SECONDS)); + assertTrue(ex.getCause().getMessage().contains("exceeds limit")); + } + + @Test + void polling_slowActivationExceedingInterval_neverRunsConcurrently() throws Exception { byte[] bundleData = createValidBundle(); serverExecutor = Executors.newFixedThreadPool(8); @@ -337,9 +349,7 @@ void polling_slowActivationExceedingInterval_neverRunsConcurrently() throws Exce .withLogger(logger) .build(); - ConcurrencyTrackingDownloader downloader = - new ConcurrencyTrackingDownloader("authz", manager); - downloader + ConcurrencyTrackingDownloader downloader = (ConcurrencyTrackingDownloader) new ConcurrencyTrackingDownloader("authz", manager) .setService("test-service") .setResource("/bundle.tar.gz") .setPolling(new Config.PollingConfig().setMinDelaySeconds(0).setMaxDelaySeconds(0)); @@ -391,19 +401,6 @@ protected void activateBundle(byte[] bundleData) { } } - @Test - void download_chunkedBodyExceedsLimit_failsMidStream() throws Exception { - int chunkSize = 64 * 1024; // 64 KB per chunk - int totalChunks = 20; // 1.28 MB total - long limit = 10 * 1024; // 10 KB limit - - int port = startRawChunkedServer(chunkSize, totalChunks); - - ExecutionException ex = assertThrows(ExecutionException.class, - () -> runBundleDownload(port, limit).get(10, TimeUnit.SECONDS)); - assertTrue(ex.getCause().getMessage().contains("exceeds limit")); - } - private int startRawChunkedServer(int chunkSize, int totalChunks) throws IOException { ServerSocket serverSocket = new ServerSocket(0); int port = serverSocket.getLocalPort(); From e4f86d6c554e55b8f27620ed5ec4dc92904ed681 Mon Sep 17 00:00:00 2001 From: Chandrahas Reddy Pola Date: Mon, 3 Aug 2026 11:57:00 -0700 Subject: [PATCH 4/8] Incorporated PR Comments. Signed-off-by: Chandrahas Reddy Pola --- .../opa/plugins/BundleDownloader.java | 14 ++++ .../plugins/BundleDownloaderSecurityTest.java | 64 +++++++++++++++++++ 2 files changed, 78 insertions(+) diff --git a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java index be5a6210..0c98bac5 100644 --- a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java +++ b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java @@ -57,6 +57,10 @@ public abstract class BundleDownloader { protected long lastModifiedTime = 0; protected long maxSizeBytes = Config.BundleConfig.DEFAULT_MAX_SIZE_BYTES; + // Fallback response timeout when no service config is available; matches the + // response_header_timeout_seconds default in Config.ServiceConfig. + private static final int DEFAULT_RESPONSE_TIMEOUT_SECONDS = 10; + private static final Set ALLOWED_CONTENT_TYPES = Set.of( "application/vnd.openpolicyagent.bundles", @@ -226,6 +230,15 @@ private static long nextDelaySeconds(int minDelay, int maxDelay) { : ThreadLocalRandom.current().nextLong(minDelay, (long) maxDelay + 1); } + // Per-request response timeout + private Duration requestTimeout() { + int seconds = + authService != null + ? authService.getResponseHeaderTimeoutSeconds() + : DEFAULT_RESPONSE_TIMEOUT_SECONDS; + return Duration.ofSeconds(seconds > 0 ? seconds : DEFAULT_RESPONSE_TIMEOUT_SECONDS); + } + /** * Download the bundle from the configured service and resource. * @@ -322,6 +335,7 @@ private CompletableFuture handleHttpDownload(URI uri) { HttpRequest.Builder requestBuilder = HttpRequest.newBuilder() .uri(uri) + .timeout(requestTimeout()) .header("Accept", "application/vnd.openpolicyagent.bundles") .GET(); diff --git a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java index 52787b2f..276acf11 100644 --- a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java +++ b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java @@ -17,6 +17,7 @@ import java.net.Socket; import java.util.Collections; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -319,6 +320,69 @@ void download_chunkedBodyExceedsLimit_failsMidStream() throws Exception { assertTrue(ex.getCause().getMessage().contains("exceeds limit")); } + @Test + void polling_serverAcceptsButNeverReplies_timesOutAndKeepsPolling() throws Exception { + CountDownLatch connections = new CountDownLatch(2); + ServerSocket silentServer = new ServerSocket(0); + int port = silentServer.getLocalPort(); + + Thread accepter = new Thread(() -> { + try { + while (!Thread.currentThread().isInterrupted()) { + silentServer.accept(); + connections.countDown(); + } + } catch (IOException ignored) { + } + }); + accepter.setDaemon(true); + accepter.start(); + + try { + Config config = new Config(); + config.setServices(Collections.singletonMap("test-service", + new Config.ServiceConfig() + .setName("test-service") + .setUrl("http://localhost:" + port) + .setResponseHeaderTimeoutSeconds(1))); + config.setBundles(Collections.singletonMap("authz", + new Config.BundleConfig() + .setService("test-service") + .setResource("/bundle.tar.gz") + .setPolling( + new Config.PollingConfig().setMinDelaySeconds(0).setMaxDelaySeconds(0)))); + + Logger logger = new Logger.StandardLogger(); + PluginManager manager = + new PluginManager.Builder() + .withId("test") + .withStore(new InMem()) + .withConfig(config) + .withLogger(logger) + .build(); + + ServicePlugin servicePlugin = (ServicePlugin) new ServicePlugin().initialize(manager); + manager.registerPlugin("services", servicePlugin); + servicePlugin.start(); + + BundlePlugin bundlePlugin = (BundlePlugin) new BundlePlugin().initialize(manager); + manager.registerPlugin("bundles", bundlePlugin); + try { + bundlePlugin.start(); + + assertTrue( + connections.await(10, TimeUnit.SECONDS), + "polling stopped after a stalled request — expected the request timeout to fail the " + + "exchange so the poll chain retries"); + } finally { + bundlePlugin.stop(); + } + } finally { + silentServer.close(); + accepter.interrupt(); + } + } + @Test void polling_slowActivationExceedingInterval_neverRunsConcurrently() throws Exception { byte[] bundleData = createValidBundle(); From 82589cd104c293afe1b6ae86686e863f7f6fbfc6 Mon Sep 17 00:00:00 2001 From: Chandrahas Reddy Pola Date: Mon, 3 Aug 2026 14:57:33 -0700 Subject: [PATCH 5/8] Added a configurable download timeout Signed-off-by: Chandrahas Reddy Pola --- .../open_policy_agent/opa/config/Config.java | 28 +++++++++++++++++++ .../opa/plugins/BundleDownloader.java | 18 ++++++------ .../opa/plugins/BundlePlugin.java | 6 ++++ .../plugins/BundleDownloaderSecurityTest.java | 5 ++-- 4 files changed, 45 insertions(+), 12 deletions(-) diff --git a/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java b/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java index ed96d4ec..be03f160 100644 --- a/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java +++ b/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java @@ -330,10 +330,26 @@ public static class BundleConfig { /** Default maximum bundle size (compressed download and decompressed contents): 512 MB. */ public static final long DEFAULT_MAX_SIZE_BYTES = 512L * 1024 * 1024; + /** + * Default bundle download timeout: 5 minutes. This bounds the entire HTTP exchange, + * so it is sized to let a bundle at the default {@link #DEFAULT_MAX_SIZE_BYTES} (512 MB) limit + * download over a slow link, while still guaranteeing a stalled server cannot block polling + * indefinitely. + */ + public static final int DEFAULT_DOWNLOAD_TIMEOUT_SECONDS = 300; + private PollingConfig polling; private String service; private String resource; + /** + * Maximum time allowed for a single bundle download, in seconds. Defaults to {@link + * #DEFAULT_DOWNLOAD_TIMEOUT_SECONDS} (5 minutes). + * + */ + @JsonProperty("download_timeout_seconds") + private int downloadTimeoutSeconds = DEFAULT_DOWNLOAD_TIMEOUT_SECONDS; + /** * Maximum bundle size in bytes, applied both to the compressed HTTP download and to the * decompressed tarball contents. Defaults to {@link #DEFAULT_MAX_SIZE_BYTES} (512 MB). @@ -381,6 +397,18 @@ public BundleConfig setResource(String resource) { return this; } + public int getDownloadTimeoutSeconds() { + return downloadTimeoutSeconds; + } + + /** + * Set the maximum time allowed for a single bundle download, in seconds. Must be positive. + */ + public BundleConfig setDownloadTimeoutSeconds(int downloadTimeoutSeconds) { + this.downloadTimeoutSeconds = downloadTimeoutSeconds; + return this; + } + public long getMaxSizeBytes() { return maxSizeBytes; } diff --git a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java index 0c98bac5..b686f2f8 100644 --- a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java +++ b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java @@ -56,10 +56,7 @@ public abstract class BundleDownloader { protected String etag; protected long lastModifiedTime = 0; protected long maxSizeBytes = Config.BundleConfig.DEFAULT_MAX_SIZE_BYTES; - - // Fallback response timeout when no service config is available; matches the - // response_header_timeout_seconds default in Config.ServiceConfig. - private static final int DEFAULT_RESPONSE_TIMEOUT_SECONDS = 10; + protected int downloadTimeoutSeconds = Config.BundleConfig.DEFAULT_DOWNLOAD_TIMEOUT_SECONDS; private static final Set ALLOWED_CONTENT_TYPES = Set.of( @@ -169,6 +166,11 @@ public BundleDownloader setMaxSizeBytes(long maxSizeBytes) { return this; } + public BundleDownloader setDownloadTimeoutSeconds(int downloadTimeoutSeconds) { + this.downloadTimeoutSeconds = downloadTimeoutSeconds; + return this; + } + /** * @return a future that completes when the first bundle download succeeds, or completes * exceptionally with the underlying download/activation error @@ -230,13 +232,9 @@ private static long nextDelaySeconds(int minDelay, int maxDelay) { : ThreadLocalRandom.current().nextLong(minDelay, (long) maxDelay + 1); } - // Per-request response timeout + // Per-request download timeout. private Duration requestTimeout() { - int seconds = - authService != null - ? authService.getResponseHeaderTimeoutSeconds() - : DEFAULT_RESPONSE_TIMEOUT_SECONDS; - return Duration.ofSeconds(seconds > 0 ? seconds : DEFAULT_RESPONSE_TIMEOUT_SECONDS); + return Duration.ofSeconds(downloadTimeoutSeconds); } /** diff --git a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundlePlugin.java b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundlePlugin.java index 8f5d1904..d1989944 100644 --- a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundlePlugin.java +++ b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundlePlugin.java @@ -44,6 +44,11 @@ public Set validate(PluginManager manager) { if (bundle.getResource() == null || bundle.getResource().isEmpty()) { errors.add("Bundle '" + name + "' has missing or empty resource path"); } + if (bundle.getDownloadTimeoutSeconds() <= 0) { + errors.add( + "Bundle '" + name + "' download_timeout_seconds must be > 0 (got " + + bundle.getDownloadTimeoutSeconds() + ")"); + } errors.addAll(BundleDownloader.validatePolling(bundle.getPolling(), "Bundle '" + name + "'")); } } @@ -79,6 +84,7 @@ public Plugin initialize(PluginManager manager) { .setResource(bundleConfig.getResource()) .setPolling(bundleConfig.getPolling()); bundle.setMaxSizeBytes(bundleConfig.getMaxSizeBytes()); + bundle.setDownloadTimeoutSeconds(bundleConfig.getDownloadTimeoutSeconds()); plugin.bundles.put(name, bundle); } } diff --git a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java index 276acf11..bcdc4a47 100644 --- a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java +++ b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java @@ -343,12 +343,13 @@ void polling_serverAcceptsButNeverReplies_timesOutAndKeepsPolling() throws Excep config.setServices(Collections.singletonMap("test-service", new Config.ServiceConfig() .setName("test-service") - .setUrl("http://localhost:" + port) - .setResponseHeaderTimeoutSeconds(1))); + .setUrl("http://localhost:" + port))); config.setBundles(Collections.singletonMap("authz", new Config.BundleConfig() .setService("test-service") .setResource("/bundle.tar.gz") + // Short download timeout so the test does not wait on the 5-minute default. + .setDownloadTimeoutSeconds(1) .setPolling( new Config.PollingConfig().setMinDelaySeconds(0).setMaxDelaySeconds(0)))); From 96623dafdd8ff9c74f5fc09b71f854b6f22210d8 Mon Sep 17 00:00:00 2001 From: Chandrahas Reddy Pola Date: Tue, 11 Aug 2026 16:16:38 -0700 Subject: [PATCH 6/8] Removed the unnecessary configurations. Signed-off-by: Chandrahas Reddy Pola --- .../open_policy_agent/opa/config/Config.java | 27 ------------------- .../opa/plugins/BundlePlugin.java | 6 ----- 2 files changed, 33 deletions(-) diff --git a/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java b/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java index be03f160..d08306c1 100644 --- a/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java +++ b/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java @@ -330,26 +330,11 @@ public static class BundleConfig { /** Default maximum bundle size (compressed download and decompressed contents): 512 MB. */ public static final long DEFAULT_MAX_SIZE_BYTES = 512L * 1024 * 1024; - /** - * Default bundle download timeout: 5 minutes. This bounds the entire HTTP exchange, - * so it is sized to let a bundle at the default {@link #DEFAULT_MAX_SIZE_BYTES} (512 MB) limit - * download over a slow link, while still guaranteeing a stalled server cannot block polling - * indefinitely. - */ - public static final int DEFAULT_DOWNLOAD_TIMEOUT_SECONDS = 300; private PollingConfig polling; private String service; private String resource; - /** - * Maximum time allowed for a single bundle download, in seconds. Defaults to {@link - * #DEFAULT_DOWNLOAD_TIMEOUT_SECONDS} (5 minutes). - * - */ - @JsonProperty("download_timeout_seconds") - private int downloadTimeoutSeconds = DEFAULT_DOWNLOAD_TIMEOUT_SECONDS; - /** * Maximum bundle size in bytes, applied both to the compressed HTTP download and to the * decompressed tarball contents. Defaults to {@link #DEFAULT_MAX_SIZE_BYTES} (512 MB). @@ -397,18 +382,6 @@ public BundleConfig setResource(String resource) { return this; } - public int getDownloadTimeoutSeconds() { - return downloadTimeoutSeconds; - } - - /** - * Set the maximum time allowed for a single bundle download, in seconds. Must be positive. - */ - public BundleConfig setDownloadTimeoutSeconds(int downloadTimeoutSeconds) { - this.downloadTimeoutSeconds = downloadTimeoutSeconds; - return this; - } - public long getMaxSizeBytes() { return maxSizeBytes; } diff --git a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundlePlugin.java b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundlePlugin.java index d1989944..8f5d1904 100644 --- a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundlePlugin.java +++ b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundlePlugin.java @@ -44,11 +44,6 @@ public Set validate(PluginManager manager) { if (bundle.getResource() == null || bundle.getResource().isEmpty()) { errors.add("Bundle '" + name + "' has missing or empty resource path"); } - if (bundle.getDownloadTimeoutSeconds() <= 0) { - errors.add( - "Bundle '" + name + "' download_timeout_seconds must be > 0 (got " - + bundle.getDownloadTimeoutSeconds() + ")"); - } errors.addAll(BundleDownloader.validatePolling(bundle.getPolling(), "Bundle '" + name + "'")); } } @@ -84,7 +79,6 @@ public Plugin initialize(PluginManager manager) { .setResource(bundleConfig.getResource()) .setPolling(bundleConfig.getPolling()); bundle.setMaxSizeBytes(bundleConfig.getMaxSizeBytes()); - bundle.setDownloadTimeoutSeconds(bundleConfig.getDownloadTimeoutSeconds()); plugin.bundles.put(name, bundle); } } From ff6c4abc0544a5d6ee0d18c5c03df4ca56f83201 Mon Sep 17 00:00:00 2001 From: Chandrahas Reddy Pola Date: Tue, 11 Aug 2026 19:31:00 -0700 Subject: [PATCH 7/8] Switched to Streams Signed-off-by: Chandrahas Reddy Pola --- .../opa/plugins/BundleDownloader.java | 165 +++++++----------- .../plugins/BundleDownloaderSecurityTest.java | 5 +- 2 files changed, 68 insertions(+), 102 deletions(-) diff --git a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java index b686f2f8..7d529351 100644 --- a/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java +++ b/opa-services/src/main/java/io/github/open_policy_agent/opa/plugins/BundleDownloader.java @@ -2,24 +2,21 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.InputStream; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; -import java.nio.ByteBuffer; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.nio.file.attribute.FileTime; import java.time.Duration; import java.util.HashSet; -import java.util.List; import java.util.Locale; import java.util.OptionalLong; import java.util.Set; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionStage; -import java.util.concurrent.Flow; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledThreadPoolExecutor; @@ -56,7 +53,12 @@ public abstract class BundleDownloader { protected String etag; protected long lastModifiedTime = 0; protected long maxSizeBytes = Config.BundleConfig.DEFAULT_MAX_SIZE_BYTES; - protected int downloadTimeoutSeconds = Config.BundleConfig.DEFAULT_DOWNLOAD_TIMEOUT_SECONDS; + + // Fallback when no service config is available; matches the response_header_timeout_seconds + // default in Config.ServiceConfig. + private static final int DEFAULT_RESPONSE_HEADER_TIMEOUT_SECONDS = 10; + + private static final int READ_BUFFER_BYTES = 8192; private static final Set ALLOWED_CONTENT_TYPES = Set.of( @@ -166,11 +168,6 @@ public BundleDownloader setMaxSizeBytes(long maxSizeBytes) { return this; } - public BundleDownloader setDownloadTimeoutSeconds(int downloadTimeoutSeconds) { - this.downloadTimeoutSeconds = downloadTimeoutSeconds; - return this; - } - /** * @return a future that completes when the first bundle download succeeds, or completes * exceptionally with the underlying download/activation error @@ -234,7 +231,11 @@ private static long nextDelaySeconds(int minDelay, int maxDelay) { // Per-request download timeout. private Duration requestTimeout() { - return Duration.ofSeconds(downloadTimeoutSeconds); + int seconds = + authService != null + ? authService.getResponseHeaderTimeoutSeconds() + : DEFAULT_RESPONSE_HEADER_TIMEOUT_SECONDS; + return Duration.ofSeconds(seconds > 0 ? seconds : DEFAULT_RESPONSE_HEADER_TIMEOUT_SECONDS); } /** @@ -347,10 +348,15 @@ private CompletableFuture handleHttpDownload(URI uri) { } HttpRequest request = requestBuilder.build(); - return httpClient - .sendAsync(request, sizeLimitedBodyHandler(maxSizeBytes)) + + CompletableFuture> exchange = + httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofInputStream()); + return exchange .handle( (response, throwable) -> { + if (throwable != null) { + exchange.cancel(true); + } handleHttpResponse(response, throwable); return null; }); @@ -359,7 +365,7 @@ private CompletableFuture handleHttpDownload(URI uri) { // Handles a completed HTTP exchange: records success or failure on initialActivation. Runs on the // HTTP client's executor thread, but the poll chain never starts the next download until the // future returned by handleHttpDownload completes, so activateBundle is never called concurrently. - private void handleHttpResponse(HttpResponse response, Throwable throwable) { + private void handleHttpResponse(HttpResponse response, Throwable throwable) { if (throwable != null) { Throwable cause = throwable instanceof java.util.concurrent.CompletionException @@ -390,9 +396,20 @@ private void handleHttpResponse(HttpResponse response, Throwable throwab } return; } + OptionalLong contentLength = response.headers().firstValueAsLong("Content-Length"); + byte[] body; + try { + body = readBodyWithLimit(response.body(), contentLength); + } catch (Exception e) { + manager.getLogger().error("Bundle '%s': Download error: %s", name, e.getMessage()); + if (!initialActivation.isDone()) { + initialActivation.completeExceptionally(e); + } + return; + } response.headers().firstValue("ETag").ifPresent(newEtag -> this.etag = newEtag); try { - activateBundle(response.body()); + activateBundle(body); } catch (Exception e) { manager.getLogger().error("Bundle '%s': Activation failed: %s", name, e.getMessage()); if (!initialActivation.isDone()) { @@ -425,101 +442,51 @@ private static boolean isAcceptableContentType(String contentType) { } /** - * Returns a BodyHandler that rejects over-sized responses. - * - *

If Content-Length is present and already exceeds the limit, the body is rejected upfront - * without reading it. Otherwise, a {@link SizeLimitedByteArraySubscriber} streams the body and - * aborts the moment the cumulative byte count exceeds the limit — so a server that omits - * Content-Length (or uses chunked encoding) cannot force the client to buffer an unbounded body. + * Reads the response body into memory, enforcing {@code maxSizeBytes} both upfront (via + * Content-Length) and as the bytes arrive. * + *

The limit is capped at {@link Integer#MAX_VALUE} because the body is materialized into a + * {@code byte[]}, whose maximum length is ~2 GB. */ - private static HttpResponse.BodyHandler sizeLimitedBodyHandler(long maxBytes) { - long cappedMax = Math.min(maxBytes, Integer.MAX_VALUE); - return responseInfo -> { - OptionalLong contentLength = responseInfo.headers().firstValueAsLong("Content-Length"); - if (contentLength.isPresent() && contentLength.getAsLong() > cappedMax) { - return new SizeLimitedByteArraySubscriber( - cappedMax, - "Content-Length " - + contentLength.getAsLong() - + " exceeds limit of " - + cappedMax - + " bytes"); - } - return new SizeLimitedByteArraySubscriber(cappedMax); - }; + private byte[] readBodyWithLimit(InputStream body, OptionalLong contentLength) + throws IOException { + long limit = Math.min(maxSizeBytes, Integer.MAX_VALUE); + try (InputStream in = body) { + rejectIfContentLengthExceedsLimit(contentLength, limit); + return readUpTo(in, limit); + } } /** - * Accumulates the response body while enforcing a size limit as bytes arrive. The instant the cumulative byte - * count exceeds {@code limit}, it cancels the subscription. + * Rejects a response whose advertised Content-Length already exceeds the limit, so an oversized + * body is never read at all. Servers are turned away before a single byte is transferred. */ - private static final class SizeLimitedByteArraySubscriber - implements HttpResponse.BodySubscriber { - private final long limit; - private final String rejectImmediatelyMessage; - private final CompletableFuture result = new CompletableFuture<>(); - private final ByteArrayOutputStream buffer = new ByteArrayOutputStream(); - private long total = 0; - private Flow.Subscription subscription; - - SizeLimitedByteArraySubscriber(long limit) { - this(limit, null); - } - - SizeLimitedByteArraySubscriber(long limit, String rejectImmediatelyMessage) { - this.limit = limit; - this.rejectImmediatelyMessage = rejectImmediatelyMessage; - } - - @Override - public CompletionStage getBody() { - return result; - } - - @Override - public void onSubscribe(Flow.Subscription subscription) { - this.subscription = subscription; - if (rejectImmediatelyMessage != null) { - subscription.cancel(); - result.completeExceptionally(new BundleSizeLimitException(rejectImmediatelyMessage)); - return; - } - subscription.request(Long.MAX_VALUE); + private static void rejectIfContentLengthExceedsLimit(OptionalLong contentLength, long limit) { + if (contentLength.isPresent() && contentLength.getAsLong() > limit) { + throw new BundleSizeLimitException( + "Content-Length " + contentLength.getAsLong() + " exceeds limit of " + limit + " bytes"); } + } - @Override - public void onNext(List items) { - if (result.isDone()) { - return; - } - for (ByteBuffer item : items) { - total += item.remaining(); - } + /** + * Reads the stream into a byte array, aborting the moment the cumulative size exceeds the limit. + * A server that omits Content-Length (or uses chunked encoding) therefore cannot force the client + * to buffer an unbounded body: the read stops mid-stream and the caller closes the stream, which + * cancels the exchange so the remaining bytes are never transferred. + */ + private static byte[] readUpTo(InputStream in, long limit) throws IOException { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + byte[] chunk = new byte[READ_BUFFER_BYTES]; + long total = 0; + int read; + while ((read = in.read(chunk)) != -1) { + total += read; if (total > limit) { - subscription.cancel(); - result.completeExceptionally( - new BundleSizeLimitException("Response body exceeds limit of " + limit + " bytes")); - return; - } - for (ByteBuffer item : items) { - byte[] chunk = new byte[item.remaining()]; - item.get(chunk); - buffer.write(chunk, 0, chunk.length); - } - } - - @Override - public void onError(Throwable throwable) { - result.completeExceptionally(throwable); - } - - @Override - public void onComplete() { - if (!result.isDone()) { - result.complete(buffer.toByteArray()); + throw new BundleSizeLimitException("Response body exceeds limit of " + limit + " bytes"); } + buffer.write(chunk, 0, read); } + return buffer.toByteArray(); } static class BundleSizeLimitException extends RuntimeException { diff --git a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java index bcdc4a47..276acf11 100644 --- a/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java +++ b/opa-services/src/test/java/io/github/open_policy_agent/opa/plugins/BundleDownloaderSecurityTest.java @@ -343,13 +343,12 @@ void polling_serverAcceptsButNeverReplies_timesOutAndKeepsPolling() throws Excep config.setServices(Collections.singletonMap("test-service", new Config.ServiceConfig() .setName("test-service") - .setUrl("http://localhost:" + port))); + .setUrl("http://localhost:" + port) + .setResponseHeaderTimeoutSeconds(1))); config.setBundles(Collections.singletonMap("authz", new Config.BundleConfig() .setService("test-service") .setResource("/bundle.tar.gz") - // Short download timeout so the test does not wait on the 5-minute default. - .setDownloadTimeoutSeconds(1) .setPolling( new Config.PollingConfig().setMinDelaySeconds(0).setMaxDelaySeconds(0)))); From 7d36d56824f190106de5998accd5f0c4e303c0b8 Mon Sep 17 00:00:00 2001 From: Chandrahas Reddy Pola Date: Tue, 11 Aug 2026 19:41:42 -0700 Subject: [PATCH 8/8] Reverted the changes Signed-off-by: Chandrahas Reddy Pola --- .../main/java/io/github/open_policy_agent/opa/config/Config.java | 1 - 1 file changed, 1 deletion(-) diff --git a/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java b/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java index d08306c1..ed96d4ec 100644 --- a/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java +++ b/opa-services/src/main/java/io/github/open_policy_agent/opa/config/Config.java @@ -330,7 +330,6 @@ public static class BundleConfig { /** Default maximum bundle size (compressed download and decompressed contents): 512 MB. */ public static final long DEFAULT_MAX_SIZE_BYTES = 512L * 1024 * 1024; - private PollingConfig polling; private String service; private String resource;