From 2b3b8240b7c83873f927915a67cccc7faec787ba Mon Sep 17 00:00:00 2001 From: xsalefter Date: Wed, 14 Oct 2026 07:42:35 +0700 Subject: [PATCH 1/2] implements basic authentication to SSE endpoint --- .../osgi/bundles/logger/Activator.java | 15 +- .../osgi/bundles/logger/BasicAuthFilter.java | 293 ++++++++++++++++++ .../bundles/logger/TestBasicAuthFilter.java | 230 ++++++++++++++ .../bundles/logger/TestBasicCredentials.java | 134 ++++++++ .../logger/TestFailureRateLimiter.java | 180 +++++++++++ 5 files changed, 845 insertions(+), 7 deletions(-) create mode 100644 osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/BasicAuthFilter.java create mode 100644 osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestBasicAuthFilter.java create mode 100644 osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestBasicCredentials.java create mode 100644 osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestFailureRateLimiter.java diff --git a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java index 26989abdd..b54dc3b49 100644 --- a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java +++ b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java @@ -37,16 +37,14 @@ public class Activator extends KillbillActivatorBase { public static final String PLUGIN_NAME = "killbill-osgi-logger"; - private LogEntriesManager logEntriesManager; private LogsSseHandler logsSseHandler; - private KillbillLogWriter killbillLogListener; - private KillbillLoggerFactory loggerFactory; @Override public void start(final BundleContext context) throws Exception { - loggerFactory = new KillbillLoggerFactory(context.getBundle()); - logEntriesManager = new LogEntriesManager(); - killbillLogListener = new KillbillLogWriter(logEntriesManager, loggerFactory); + super.start(context); + final KillbillLoggerFactory loggerFactory = new KillbillLoggerFactory(context.getBundle()); + final LogEntriesManager logEntriesManager = new LogEntriesManager(); + final KillbillLogWriter killbillLogListener = new KillbillLogWriter(logEntriesManager, loggerFactory); context.addBundleListener(killbillLogListener); context.addFrameworkListener(killbillLogListener); context.addServiceListener(killbillLogListener); @@ -58,6 +56,9 @@ public void start(final BundleContext context) throws Exception { registrar = new OSGIKillbillRegistrar(); final PluginApp pluginApp = new PluginAppBuilder(PLUGIN_NAME).build(); + // Require HTTP Basic credentials (validated against Kill Bill RBAC), since /plugins/** is not authenticated + // by the Kill Bill server by default. See BasicAuthFilter for details. + pluginApp.use("*", new BasicAuthFilter(killbillAPI.getSecurityApi())); logsSseHandler = new LogsSseHandler(logEntriesManager); pluginApp.sse("/", logsSseHandler); final HttpServlet httpServlet = PluginApp.createServlet(pluginApp); @@ -74,7 +75,7 @@ public void stop(final BundleContext context) throws Exception { } private void registerServlet(final BundleContext context, final HttpServlet servlet) { - final Hashtable props = new Hashtable(); + final Hashtable props = new Hashtable<>(); props.put(OSGIPluginProperties.PLUGIN_NAME_PROP, PLUGIN_NAME); registrar.registerService(context, Servlet.class, servlet, props); } diff --git a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/BasicAuthFilter.java b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/BasicAuthFilter.java new file mode 100644 index 000000000..5df87dbc7 --- /dev/null +++ b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/BasicAuthFilter.java @@ -0,0 +1,293 @@ +/* + * Copyright 2020-2026 Equinix, Inc + * Copyright 2014-2026 The Billing Project, LLC + * + * The Billing Project licenses this file to you under the Apache License, version 2.0 + * (the "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.killbill.billing.osgi.bundles.logger; + +import java.nio.charset.StandardCharsets; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.Base64; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.jooby.Request; +import org.jooby.Response; +import org.jooby.Route; +import org.killbill.billing.security.api.SecurityApi; +import org.killbill.commons.utils.annotation.VisibleForTesting; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +// By default, Kill Bill does NOT enforce authentication nor tenant information on /plugins/** requests +// (see KillBillShiroWebModule#configureShiroForRBAC and org.killbill.billing.osgi.http.OSGIServlet): +// it is up to each plugin to enforce its own access control. This filter requires HTTP Basic +// credentials to be validated against Kill Bill RBAC (the same realms used by /1.0/kb/**), without +// requiring tenant information (X-Killbill-ApiKey/ApiSecret), similar to NodesInfoResource. +class BasicAuthFilter implements Route.Filter { + + private static final Logger logger = LoggerFactory.getLogger(BasicAuthFilter.class); + + private static final String AUTHORIZATION_HEADER = "Authorization"; + private static final String BASIC_PREFIX = "Basic "; + private static final String WWW_AUTHENTICATE_HEADER = "WWW-Authenticate"; + private static final String WWW_AUTHENTICATE_VALUE = "Basic realm=\"".concat(Activator.PLUGIN_NAME).concat("\""); + + // Guards against a client (e.g. a buggy frontend that keeps reconnecting an SSE stream after + // a failed auth, instead of giving up) hammering this endpoint with repeated failed attempts. + private static final int MAX_FAILURES_PER_WINDOW = 10; + private static final Duration FAILURE_WINDOW = Duration.ofMinutes(1); + + private final SecurityApi securityApi; + private final FailureRateLimiter rateLimiter; + + BasicAuthFilter(final SecurityApi securityApi) { + this(securityApi, new FailureRateLimiter(MAX_FAILURES_PER_WINDOW, FAILURE_WINDOW)); + } + + @VisibleForTesting + BasicAuthFilter(final SecurityApi securityApi, final FailureRateLimiter rateLimiter) { + this.securityApi = securityApi; + this.rateLimiter = rateLimiter; + } + + @Override + public void handle(final Request req, final Response rsp, final Route.Chain chain) throws Throwable { + // CORS preflight requests never carry credentials (by the Fetch/CORS spec, browsers strip Authorization and + // other request headers from the OPTIONS preflight itself), so enforcing auth here would always 401 the + // preflight and block the real, cross-origin request before it's even sent. + // + // Respond 200 directly: there's no terminal route registered for OPTIONS (the SSE route only handles GET), + // so calling chain.next() would fall through to Jooby's own 405 Method Not Allowed, which also fails the + // preflight (needs a 2xx response). + // + // The actual CORS headers (Access-Control-Allow-*) are added upstream by Kill Bill's ResponseCorsFilter. + // We just need to not block the preflight with an auth check here. + if ("OPTIONS".equalsIgnoreCase(req.method())) { + rsp.status(200); + rsp.end(); + return; + } + + final String clientKey = req.ip(); + + if (rateLimiter.isRateLimited(clientKey)) { + this.handleTooManyRequests(rsp); + return; + } + + final Optional credentials = BasicCredentials.fromRequest(req); + + // A missing/malformed Authorization header is not itself a rejection reason. Subject may already + // be authenticated via an existing Shiro session (e.g. KAUI, which authenticates once via + // /1.0/kb/security/permissions and reuses the session cookie on subsequent requests, including this one). So, + // only attempt login when credentials are present, but always fall through to the isSubjectAuthenticated(). + credentials.ifPresent(this::login); + + if (!securityApi.isSubjectAuthenticated()) { + rateLimiter.recordFailure(clientKey); + this.handleUnAuthenticated(rsp); + return; + } + + rateLimiter.recordSuccess(clientKey); + chain.next(req, rsp); + } + + private void handleUnAuthenticated(final Response response) throws Throwable { + response.header(WWW_AUTHENTICATE_HEADER, WWW_AUTHENTICATE_VALUE); + response.status(401); + response.send("Unauthorized"); + } + + private void handleTooManyRequests(final Response response) throws Throwable { + response.header("Retry-After", String.valueOf(FAILURE_WINDOW.toSeconds())); + response.status(429); + response.send("Too Many Requests"); + } + + private void login(final BasicCredentials credentials) { + try { + securityApi.login(credentials.getUsername(), credentials.getPassword()); + } catch (final RuntimeException e) { + // Invalid credentials: isSubjectAuthenticated() below will correctly report false and trigger the + // 401 response. Also, we deliberately catch RuntimeException (not Shiro's AuthenticationException) since + // this bundle has no compile dependency on shiro-core. + logger.warn("SSE: invalid credentials"); + } + } + + /** + * Parses HTTP Basic credentials (username/password) out of a request's {@code Authorization} header. Returns + * empty if the header is missing, not a {@code Basic} scheme, not valid base64, doesn't contain a colon-separated + * {@code username:password} pair (the password itself may contain a colon), or if the username/password is empty. + */ + @VisibleForTesting + static class BasicCredentials { + + private final String username; + private final String password; + + private BasicCredentials(final String username, final String password) { + this.username = username; + this.password = password; + } + + String getUsername() { + return username; + } + + String getPassword() { + return password; + } + + static Optional fromRequest(final Request req) { + final Optional authHeader = req.header(AUTHORIZATION_HEADER).toOptional(); + if (authHeader.isEmpty()) { + return Optional.empty(); + } + return decode(authHeader.get()); + } + + @VisibleForTesting + static Optional decode(final String authorizationHeaderValue) { + // Auth-scheme tokens are matched case-insensitively (RFC 7235 section 2.1), so "basic ", "Basic " and + // "BASIC " are all equally valid. + if (!authorizationHeaderValue.regionMatches(true, 0, BASIC_PREFIX, 0, BASIC_PREFIX.length())) { + return Optional.empty(); + } + + final String base64Credentials = authorizationHeaderValue.substring(BASIC_PREFIX.length()).trim(); + final String decoded; + try { + decoded = new String(Base64.getDecoder().decode(base64Credentials), StandardCharsets.UTF_8); + } catch (final IllegalArgumentException e) { + return Optional.empty(); + } + + final int idx = decoded.indexOf(':'); + if (idx < 0) { + return Optional.empty(); + } + + final String username = decoded.substring(0, idx); + final String password = decoded.substring(idx + 1); + if (username.isEmpty() || password.isEmpty()) { + return Optional.empty(); + } + + return Optional.of(new BasicCredentials(username, password)); + } + } + + /** + * Tracks failed-authentication counts per client key (see {@link Request#ip()}) within a fixed time window, to + * blunt a client stuck in a reconnect loop against this endpoint (example: frontend bug that doesn't close an SSE + * connection after a failed auth). + * + * Not a distributed rate limiter: state is in-process/per-node only, which is acceptable here since the goal is + * containing one misbehaving client, not providing a hard security guarantee against a determined, IP-rotating + * attacker. + */ + @VisibleForTesting + static class FailureRateLimiter { + + // Opportunistic cleanup trigger to bound memory growth from many distinct/expired keys, since entries are + // otherwise only removed on access (lazy expiry). No scheduled background task is used here to avoid adding + // lifecycle (start/stop) management to the plugin. + private static final int CLEANUP_THRESHOLD = 10_000; + + // Minimum time between full-sweep cleanups once the threshold is hit. Without this, a burst of failures from + // many distinct, still-unexpired keys (e.g. a high-cardinality/distributed attempt) would keep the map above + // CLEANUP_THRESHOLD forever, turning every subsequent recordFailure() call into a wasted O(n) scan. + private static final Duration MIN_CLEANUP_INTERVAL = Duration.ofSeconds(10); + + private final int maxFailures; + private final Duration window; + private final Clock clock; // Prefer date.time.Clock instead of killbill Clock + private final ConcurrentHashMap failuresByKey = new ConcurrentHashMap<>(); + private final AtomicReference lastCleanupAttempt = new AtomicReference<>(Instant.MIN); + + FailureRateLimiter(final int maxFailures, final Duration window) { + this(maxFailures, window, Clock.systemUTC()); + } + + @VisibleForTesting + FailureRateLimiter(final int maxFailures, final Duration window, final Clock clock) { + this.maxFailures = maxFailures; + this.window = window; + this.clock = clock; + } + + boolean isRateLimited(final String key) { + final FailureWindow failureWindow = failuresByKey.get(key); + return failureWindow != null && !isExpired(failureWindow) && failureWindow.count.get() >= maxFailures; + } + + void recordFailure(final String key) { + failuresByKey.compute(key, (k, existing) -> { + if (existing == null || isExpired(existing)) { + return new FailureWindow(clock.instant()); + } + existing.count.incrementAndGet(); + return existing; + }); + + if (failuresByKey.size() > CLEANUP_THRESHOLD) { + final Instant now = clock.instant(); + final Instant previous = lastCleanupAttempt.get(); + // CAS, not just a volatile read-then-write: under a concurrent high-cardinality burst, multiple + // threads could otherwise all observe the throttle interval as elapsed before any of them updates + // the timestamp, and each would then redundantly run the full O(n) cleanup() scan -- defeating the + // very throttle meant to bound that cost. Only the thread that wins the CAS proceeds. + if (Duration.between(previous, now).compareTo(MIN_CLEANUP_INTERVAL) >= 0 + && lastCleanupAttempt.compareAndSet(previous, now)) { + cleanup(); + } + } + } + + void recordSuccess(final String key) { + failuresByKey.remove(key); + } + + @VisibleForTesting + int size() { + return failuresByKey.size(); + } + + private boolean isExpired(final FailureWindow failureWindow) { + return Duration.between(failureWindow.windowStart, clock.instant()).compareTo(window) >= 0; + } + + private void cleanup() { + failuresByKey.entrySet().removeIf(entry -> isExpired(entry.getValue())); + } + + private static final class FailureWindow { + + private final Instant windowStart; + private final AtomicInteger count = new AtomicInteger(1); + + private FailureWindow(final Instant windowStart) { + this.windowStart = windowStart; + } + } + } +} \ No newline at end of file diff --git a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestBasicAuthFilter.java b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestBasicAuthFilter.java new file mode 100644 index 000000000..20fd86dbc --- /dev/null +++ b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestBasicAuthFilter.java @@ -0,0 +1,230 @@ +/* + * Copyright 2020-2026 Equinix, Inc + * Copyright 2014-2026 The Billing Project, LLC + * + * The Billing Project licenses this file to you under the Apache License, version 2.0 + * (the "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.killbill.billing.osgi.bundles.logger; + +import java.util.Base64; +import java.util.Optional; + +import org.jooby.Mutant; +import org.jooby.Request; +import org.jooby.Response; +import org.jooby.Route; +import org.killbill.billing.security.api.SecurityApi; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class TestBasicAuthFilter { + + // -- handle() tests -- + + @Test(groups = "fast") + public void testHandleNoAuthorizationHeaderRejected() throws Throwable { + final SecurityApi securityApi = Mockito.mock(SecurityApi.class); + Mockito.when(securityApi.isSubjectAuthenticated()).thenReturn(false); + + final Request request = mockRequestWithAuthorizationHeader(null); + final Response response = Mockito.mock(Response.class); + Mockito.when(response.header(Mockito.anyString(), Mockito.any(Object.class))).thenReturn(response); + Mockito.when(response.status(Mockito.anyInt())).thenReturn(response); + final Route.Chain chain = Mockito.mock(Route.Chain.class); + + new BasicAuthFilter(securityApi).handle(request, response, chain); + + Mockito.verify(securityApi, Mockito.never()).login(Mockito.any(), Mockito.any()); + Mockito.verify(response).status(401); + Mockito.verify(response).header("WWW-Authenticate", "Basic realm=\"killbill-osgi-logger\""); + Mockito.verify(chain, Mockito.never()).next(Mockito.any(Request.class), Mockito.any(Response.class)); + } + + @Test(groups = "fast") + public void testHandleNoAuthorizationHeaderButAlreadyAuthenticatedAllowed() throws Throwable { + // No Authorization header on this request, but the Subject is already authenticated, + // e.g. via an existing Shiro session cookie (Kaui authenticates once via + // /1.0/kb/security/permissions and reuses the session on subsequent requests). This must + // not be rejected: login() should not even be attempted when there's no header, and the + // pre-existing authenticated state alone should be enough to let the request through. + final SecurityApi securityApi = Mockito.mock(SecurityApi.class); + Mockito.when(securityApi.isSubjectAuthenticated()).thenReturn(true); + + final Request request = mockRequestWithAuthorizationHeader(null); + final Response response = Mockito.mock(Response.class); + final Route.Chain chain = Mockito.mock(Route.Chain.class); + + new BasicAuthFilter(securityApi).handle(request, response, chain); + + Mockito.verify(securityApi, Mockito.never()).login(Mockito.any(), Mockito.any()); + Mockito.verify(response, Mockito.never()).status(Mockito.anyInt()); + Mockito.verify(chain).next(request, response); + } + + @Test(groups = "fast") + public void testHandleValidCredentialsAllowed() throws Throwable { + final SecurityApi securityApi = Mockito.mock(SecurityApi.class); + Mockito.when(securityApi.isSubjectAuthenticated()).thenReturn(true); + + final Request request = mockRequestWithAuthorizationHeader(basicHeader("admin", "password")); + final Response response = Mockito.mock(Response.class); + final Route.Chain chain = Mockito.mock(Route.Chain.class); + + new BasicAuthFilter(securityApi).handle(request, response, chain); + + Mockito.verify(securityApi).login("admin", "password"); + Mockito.verify(chain).next(request, response); + Mockito.verify(response, Mockito.never()).status(Mockito.anyInt()); + } + + @Test(groups = "fast") + public void testHandleInvalidCredentialsRejected() throws Throwable { + final SecurityApi securityApi = Mockito.mock(SecurityApi.class); + Mockito.doThrow(new RuntimeException("bad credentials")).when(securityApi).login(Mockito.any(), Mockito.any()); + Mockito.when(securityApi.isSubjectAuthenticated()).thenReturn(false); + + final Request request = mockRequestWithAuthorizationHeader(basicHeader("admin", "wrong")); + final Response response = Mockito.mock(Response.class); + Mockito.when(response.header(Mockito.anyString(), Mockito.any(Object.class))).thenReturn(response); + Mockito.when(response.status(Mockito.anyInt())).thenReturn(response); + final Route.Chain chain = Mockito.mock(Route.Chain.class); + + new BasicAuthFilter(securityApi).handle(request, response, chain); + + Mockito.verify(securityApi).login("admin", "wrong"); + Mockito.verify(response).status(401); + Mockito.verify(chain, Mockito.never()).next(Mockito.any(Request.class), Mockito.any(Response.class)); + } + + @Test(groups = "fast") + public void testHandleMalformedAuthorizationHeaderStillChecksAuthentication() throws Throwable { + final SecurityApi securityApi = Mockito.mock(SecurityApi.class); + Mockito.when(securityApi.isSubjectAuthenticated()).thenReturn(false); + + final Request request = mockRequestWithAuthorizationHeader("Bearer sometoken"); + final Response response = Mockito.mock(Response.class); + Mockito.when(response.header(Mockito.anyString(), Mockito.any(Object.class))).thenReturn(response); + Mockito.when(response.status(Mockito.anyInt())).thenReturn(response); + final Route.Chain chain = Mockito.mock(Route.Chain.class); + + new BasicAuthFilter(securityApi).handle(request, response, chain); + + Mockito.verify(securityApi, Mockito.never()).login(Mockito.any(), Mockito.any()); + Mockito.verify(response).status(401); + } + + // -- CORS preflight -- + + @Test(groups = "fast") + public void testHandleOptionsRequestBypassesAuth() throws Throwable { + // A CORS preflight (OPTIONS) never carries the Authorization header (stripped by the + // browser), so it must be let through unconditionally, or legitimate cross-origin clients + // sending Basic Auth would never get past the preflight in the first place. It must also + // get a 2xx response directly (not chain.next(), which would fall through to Jooby's own + // 405, since no terminal route is registered for OPTIONS). + final SecurityApi securityApi = Mockito.mock(SecurityApi.class); + + final Request request = mockRequestWithAuthorizationHeader(null); + Mockito.when(request.method()).thenReturn("OPTIONS"); + final Response response = Mockito.mock(Response.class); + final Route.Chain chain = Mockito.mock(Route.Chain.class); + + new BasicAuthFilter(securityApi).handle(request, response, chain); + + Mockito.verify(securityApi, Mockito.never()).login(Mockito.any(), Mockito.any()); + Mockito.verify(securityApi, Mockito.never()).isSubjectAuthenticated(); + Mockito.verify(response).status(200); + Mockito.verify(response).end(); + Mockito.verify(chain, Mockito.never()).next(Mockito.any(Request.class), Mockito.any(Response.class)); + } + + // -- rate limiting integration tests -- + + @Test(groups = "fast") + public void testHandleRateLimitedReturns429BeforeLogin() throws Throwable { + final SecurityApi securityApi = Mockito.mock(SecurityApi.class); + final BasicAuthFilter.FailureRateLimiter rateLimiter = Mockito.mock(BasicAuthFilter.FailureRateLimiter.class); + Mockito.when(rateLimiter.isRateLimited("1.2.3.4")).thenReturn(true); + + final Request request = mockRequestWithAuthorizationHeader(basicHeader("admin", "password"), "1.2.3.4"); + final Response response = Mockito.mock(Response.class); + Mockito.when(response.header(Mockito.anyString(), Mockito.any(Object.class))).thenReturn(response); + Mockito.when(response.status(Mockito.anyInt())).thenReturn(response); + final Route.Chain chain = Mockito.mock(Route.Chain.class); + + new BasicAuthFilter(securityApi, rateLimiter).handle(request, response, chain); + + Mockito.verify(securityApi, Mockito.never()).login(Mockito.any(), Mockito.any()); + Mockito.verify(response).status(429); + Mockito.verify(response).header("Retry-After", "60"); + Mockito.verify(chain, Mockito.never()).next(Mockito.any(Request.class), Mockito.any(Response.class)); + } + + @Test(groups = "fast") + public void testHandleFailureRecordedOnRejection() throws Throwable { + final SecurityApi securityApi = Mockito.mock(SecurityApi.class); + Mockito.when(securityApi.isSubjectAuthenticated()).thenReturn(false); + final BasicAuthFilter.FailureRateLimiter rateLimiter = Mockito.mock(BasicAuthFilter.FailureRateLimiter.class); + + final Request request = mockRequestWithAuthorizationHeader(basicHeader("admin", "wrong"), "1.2.3.4"); + final Response response = Mockito.mock(Response.class); + Mockito.when(response.header(Mockito.anyString(), Mockito.any(Object.class))).thenReturn(response); + Mockito.when(response.status(Mockito.anyInt())).thenReturn(response); + final Route.Chain chain = Mockito.mock(Route.Chain.class); + + new BasicAuthFilter(securityApi, rateLimiter).handle(request, response, chain); + + Mockito.verify(rateLimiter).recordFailure("1.2.3.4"); + Mockito.verify(rateLimiter, Mockito.never()).recordSuccess(Mockito.any()); + } + + @Test(groups = "fast") + public void testHandleSuccessRecordedAndChainContinues() throws Throwable { + final SecurityApi securityApi = Mockito.mock(SecurityApi.class); + Mockito.when(securityApi.isSubjectAuthenticated()).thenReturn(true); + final BasicAuthFilter.FailureRateLimiter rateLimiter = Mockito.mock(BasicAuthFilter.FailureRateLimiter.class); + + final Request request = mockRequestWithAuthorizationHeader(basicHeader("admin", "password"), "1.2.3.4"); + final Response response = Mockito.mock(Response.class); + final Route.Chain chain = Mockito.mock(Route.Chain.class); + + new BasicAuthFilter(securityApi, rateLimiter).handle(request, response, chain); + + Mockito.verify(rateLimiter).recordSuccess("1.2.3.4"); + Mockito.verify(rateLimiter, Mockito.never()).recordFailure(Mockito.any()); + Mockito.verify(chain).next(request, response); + } + + // -- helpers -- + + private static String basicHeader(final String username, final String password) { + final String encoded = Base64.getEncoder().encodeToString((username + ":" + password).getBytes()); + return "Basic " + encoded; + } + + private static Request mockRequestWithAuthorizationHeader(final String value) { + return mockRequestWithAuthorizationHeader(value, "127.0.0.1"); + } + + private static Request mockRequestWithAuthorizationHeader(final String value, final String ip) { + final Request request = Mockito.mock(Request.class); + final Mutant mutant = Mockito.mock(Mutant.class); + Mockito.when(mutant.toOptional()).thenReturn(value == null ? Optional.empty() : Optional.of(value)); + Mockito.when(request.header("Authorization")).thenReturn(mutant); + Mockito.when(request.ip()).thenReturn(ip); + return request; + } +} diff --git a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestBasicCredentials.java b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestBasicCredentials.java new file mode 100644 index 000000000..a1ddc9b91 --- /dev/null +++ b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestBasicCredentials.java @@ -0,0 +1,134 @@ +/* + * Copyright 2020-2026 Equinix, Inc + * Copyright 2014-2026 The Billing Project, LLC + * + * The Billing Project licenses this file to you under the Apache License, version 2.0 + * (the "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.killbill.billing.osgi.bundles.logger; + +import java.util.Base64; +import java.util.Optional; + +import org.jooby.Mutant; +import org.jooby.Request; +import org.killbill.billing.osgi.bundles.logger.BasicAuthFilter.BasicCredentials; +import org.mockito.Mockito; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class TestBasicCredentials { + + // -- decode() tests -- + + @Test(groups = "fast") + public void testDecodeValid() { + final Optional result = BasicCredentials.decode(basicHeader("admin", "password")); + Assert.assertTrue(result.isPresent()); + Assert.assertEquals(result.get().getUsername(), "admin"); + Assert.assertEquals(result.get().getPassword(), "password"); + } + + @Test(groups = "fast") + public void testDecodePasswordContainsColon() { + final Optional result = BasicCredentials.decode(basicHeader("admin", "pass:word")); + Assert.assertTrue(result.isPresent()); + Assert.assertEquals(result.get().getUsername(), "admin"); + Assert.assertEquals(result.get().getPassword(), "pass:word"); + } + + @Test(groups = "fast") + public void testDecodeNotBasicScheme() { + Assert.assertEquals(BasicCredentials.decode("******"), Optional.empty()); + } + + @Test(groups = "fast") + public void testDecodeSchemeIsCaseInsensitive() { + // Auth-scheme tokens are matched case-insensitively (RFC 7235 section 2.1). + final String encoded = Base64.getEncoder().encodeToString("admin:password".getBytes()); + + final Optional lowercase = BasicCredentials.decode("basic " + encoded); + Assert.assertTrue(lowercase.isPresent()); + Assert.assertEquals(lowercase.get().getUsername(), "admin"); + Assert.assertEquals(lowercase.get().getPassword(), "password"); + + final Optional mixedCase = BasicCredentials.decode("BaSiC " + encoded); + Assert.assertTrue(mixedCase.isPresent()); + Assert.assertEquals(mixedCase.get().getUsername(), "admin"); + Assert.assertEquals(mixedCase.get().getPassword(), "password"); + } + + @Test(groups = "fast") + public void testDecodeInvalidBase64() { + Assert.assertEquals(BasicCredentials.decode("Basic not-valid-base64!!"), Optional.empty()); + } + + @Test(groups = "fast") + public void testDecodeMissingColon() { + final String encoded = Base64.getEncoder().encodeToString("adminpassword".getBytes()); + Assert.assertEquals(BasicCredentials.decode("Basic " + encoded), Optional.empty()); + } + + @Test(groups = "fast") + public void testDecodeEmptyUsername() { + Assert.assertEquals(BasicCredentials.decode(basicHeader("", "password")), Optional.empty()); + } + + @Test(groups = "fast") + public void testDecodeEmptyPassword() { + Assert.assertEquals(BasicCredentials.decode(basicHeader("admin", "")), Optional.empty()); + } + + @Test(groups = "fast") + public void testDecodeEmptyUsernameAndPassword() { + Assert.assertEquals(BasicCredentials.decode(basicHeader("", "")), Optional.empty()); + } + + // -- fromRequest() tests -- + + @Test(groups = "fast") + public void testFromRequestNoAuthorizationHeader() { + final Request request = mockRequestWithAuthorizationHeader(null); + Assert.assertEquals(BasicCredentials.fromRequest(request), Optional.empty()); + } + + @Test(groups = "fast") + public void testFromRequestValid() { + final Request request = mockRequestWithAuthorizationHeader(basicHeader("admin", "password")); + final Optional result = BasicCredentials.fromRequest(request); + Assert.assertTrue(result.isPresent()); + Assert.assertEquals(result.get().getUsername(), "admin"); + Assert.assertEquals(result.get().getPassword(), "password"); + } + + @Test(groups = "fast") + public void testFromRequestMalformedHeader() { + final Request request = mockRequestWithAuthorizationHeader("******"); + Assert.assertEquals(BasicCredentials.fromRequest(request), Optional.empty()); + } + + // -- helpers -- + + private static String basicHeader(final String username, final String password) { + final String encoded = Base64.getEncoder().encodeToString((username + ":" + password).getBytes()); + return "Basic " + encoded; + } + + private static Request mockRequestWithAuthorizationHeader(final String value) { + final Request request = Mockito.mock(Request.class); + final Mutant mutant = Mockito.mock(Mutant.class); + Mockito.when(mutant.toOptional()).thenReturn(value == null ? Optional.empty() : Optional.of(value)); + Mockito.when(request.header("Authorization")).thenReturn(mutant); + return request; + } +} diff --git a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestFailureRateLimiter.java b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestFailureRateLimiter.java new file mode 100644 index 000000000..9d71764ec --- /dev/null +++ b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestFailureRateLimiter.java @@ -0,0 +1,180 @@ +/* + * Copyright 2020-2026 Equinix, Inc + * Copyright 2014-2026 The Billing Project, LLC + * + * The Billing Project licenses this file to you under the Apache License, version 2.0 + * (the "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.killbill.billing.osgi.bundles.logger; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.concurrent.atomic.AtomicReference; + +import org.killbill.billing.osgi.bundles.logger.BasicAuthFilter.FailureRateLimiter; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class TestFailureRateLimiter { + + private static final int MAX_FAILURES = 3; + private static final Duration WINDOW = Duration.ofMinutes(1); + + @Test(groups = "fast") + public void testNotRateLimitedInitially() { + final FailureRateLimiter limiter = newLimiter(Instant.EPOCH); + Assert.assertFalse(limiter.isRateLimited("1.2.3.4")); + } + + @Test(groups = "fast") + public void testNotRateLimitedUnderThreshold() { + final MutableClock clock = new MutableClock(Instant.EPOCH); + final FailureRateLimiter limiter = new FailureRateLimiter(MAX_FAILURES, WINDOW, clock); + + limiter.recordFailure("1.2.3.4"); + limiter.recordFailure("1.2.3.4"); + + Assert.assertFalse(limiter.isRateLimited("1.2.3.4")); + } + + @Test(groups = "fast") + public void testRateLimitedAtThreshold() { + final MutableClock clock = new MutableClock(Instant.EPOCH); + final FailureRateLimiter limiter = new FailureRateLimiter(MAX_FAILURES, WINDOW, clock); + + limiter.recordFailure("1.2.3.4"); + limiter.recordFailure("1.2.3.4"); + limiter.recordFailure("1.2.3.4"); + + Assert.assertTrue(limiter.isRateLimited("1.2.3.4")); + } + + @Test(groups = "fast") + public void testDifferentKeysAreIndependent() { + final MutableClock clock = new MutableClock(Instant.EPOCH); + final FailureRateLimiter limiter = new FailureRateLimiter(MAX_FAILURES, WINDOW, clock); + + limiter.recordFailure("1.2.3.4"); + limiter.recordFailure("1.2.3.4"); + limiter.recordFailure("1.2.3.4"); + + Assert.assertTrue(limiter.isRateLimited("1.2.3.4")); + Assert.assertFalse(limiter.isRateLimited("5.6.7.8")); + } + + @Test(groups = "fast") + public void testWindowExpiryResetsCount() { + final MutableClock clock = new MutableClock(Instant.EPOCH); + final FailureRateLimiter limiter = new FailureRateLimiter(MAX_FAILURES, WINDOW, clock); + + limiter.recordFailure("1.2.3.4"); + limiter.recordFailure("1.2.3.4"); + limiter.recordFailure("1.2.3.4"); + Assert.assertTrue(limiter.isRateLimited("1.2.3.4")); + + // Advance past the window: the old failures should no longer count. + clock.advance(WINDOW.plusSeconds(1)); + Assert.assertFalse(limiter.isRateLimited("1.2.3.4")); + + // A single new failure after expiry should not immediately re-trip the limiter. + limiter.recordFailure("1.2.3.4"); + Assert.assertFalse(limiter.isRateLimited("1.2.3.4")); + } + + @Test(groups = "fast") + public void testSuccessResetsFailureCount() { + final MutableClock clock = new MutableClock(Instant.EPOCH); + final FailureRateLimiter limiter = new FailureRateLimiter(MAX_FAILURES, WINDOW, clock); + + limiter.recordFailure("1.2.3.4"); + limiter.recordFailure("1.2.3.4"); + limiter.recordSuccess("1.2.3.4"); + + limiter.recordFailure("1.2.3.4"); + limiter.recordFailure("1.2.3.4"); + + // 2 failures after the reset, below MAX_FAILURES (3): should not be rate limited. + Assert.assertFalse(limiter.isRateLimited("1.2.3.4")); + } + + @Test(groups = "fast") + public void testRecordSuccessOnUnknownKeyIsNoOp() { + final FailureRateLimiter limiter = newLimiter(Instant.EPOCH); + limiter.recordSuccess("never-seen-before"); + Assert.assertFalse(limiter.isRateLimited("never-seen-before")); + } + + @Test(groups = "fast") + public void testCleanupIsThrottledUnderHighCardinalityBurst() { + // Simulate a burst of failures from > CLEANUP_THRESHOLD (10,000) distinct, still-unexpired keys. cleanup() + // can't shrink the map in that case (nothing is actually expired yet), so without throttling, every + // subsequent recordFailure() call would re-trigger a wasted full O(n) scan. + final MutableClock clock = new MutableClock(Instant.EPOCH); + final FailureRateLimiter limiter = new FailureRateLimiter(MAX_FAILURES, WINDOW, clock); + + for (int i = 0; i < 10_001; i++) { + limiter.recordFailure("key-" + i); + } + Assert.assertTrue(limiter.size() > 10_000); + + // Expire all of the above, then add one more failure right away: since the minimum cleanup interval hasn't + // elapsed since the burst above already attempted (and failed) a cleanup, this must not re-scan yet. + clock.advance(WINDOW.plusSeconds(1)); + limiter.recordFailure("key-after-expiry-1"); + Assert.assertTrue(limiter.size() > 10_000, "cleanup should not have run again before the throttle interval elapsed"); + + // Once the throttle interval has also elapsed, the next recordFailure() should finally be allowed to sweep + // out all of the now-expired entries. + clock.advance(Duration.ofSeconds(10)); + limiter.recordFailure("key-after-expiry-2"); + Assert.assertTrue(limiter.size() < 10_000, "cleanup should have run once the throttle interval elapsed"); + } + + private static FailureRateLimiter newLimiter(final Instant now) { + return new FailureRateLimiter(MAX_FAILURES, WINDOW, new MutableClock(now)); + } + + /** + * A {@link Clock} whose {@code instant()} can be advanced on demand, to deterministically test + * window-expiry logic without relying on wall-clock time or {@code Thread.sleep}. + */ + private static final class MutableClock extends Clock { + + private final AtomicReference now; + + MutableClock(final Instant now) { + this.now = new AtomicReference<>(now); + } + + void advance(final Duration duration) { + now.updateAndGet(instant -> instant.plus(duration)); + } + + @Override + public ZoneOffset getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(final java.time.ZoneId zone) { + throw new UnsupportedOperationException(); + } + + @Override + public Instant instant() { + return now.get(); + } + } +} From d8573ec6b7843a21d4891dca9cf5351165ed7622 Mon Sep 17 00:00:00 2001 From: xsalefter Date: Wed, 14 Oct 2026 09:56:59 +0700 Subject: [PATCH 2/2] add demo how to connect to SSE endpoint --- osgi-bundles/bundles/logger/html-sse-demo.md | 261 ++++++++++++++++++ .../logger/TestFailureRateLimiter.java | 16 +- 2 files changed, 269 insertions(+), 8 deletions(-) create mode 100644 osgi-bundles/bundles/logger/html-sse-demo.md diff --git a/osgi-bundles/bundles/logger/html-sse-demo.md b/osgi-bundles/bundles/logger/html-sse-demo.md new file mode 100644 index 000000000..a092ff575 --- /dev/null +++ b/osgi-bundles/bundles/logger/html-sse-demo.md @@ -0,0 +1,261 @@ +### HTML SSE Demo + +1. Build the project: `mvn clean install -pl osgi-bundles/bundles/logger` +2. Copy the JAR to `platform` plugin directory +3. Copy HTML content below to `` +4. Run this using `python3 -m http.server 3000 -d ` or `npx serve -l 3000 ` +5. Open `http://localhost:3000`, fill in the Base URL and your Kill Bill username/password, then click **Connect** + +**Auth note:** this plugin's `/plugins/killbill-osgi-logger/**` endpoint now requires HTTP Basic Auth +(see `BasicAuthFilter`, registered on this plugin's Jooby app only — other plugins are unaffected). The +native browser `EventSource` API cannot set custom request headers (no way to send `Authorization: Basic +...`), so this demo bundles [EventSource/eventsource](https://github.com/EventSource/eventsource) (MIT +license, v5.1.2), a browser-compatible `EventSource` polyfill/replacement built on `fetch`, which supports +a `fetch` override — used here to inject the Basic Auth header per request. It's bundled as a single +minified IIFE (via `esbuild`, no build step required on your end) and pasted directly into the +` + + + +``` diff --git a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestFailureRateLimiter.java b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestFailureRateLimiter.java index 9d71764ec..59d0442c1 100644 --- a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestFailureRateLimiter.java +++ b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestFailureRateLimiter.java @@ -129,16 +129,16 @@ public void testCleanupIsThrottledUnderHighCardinalityBurst() { } Assert.assertTrue(limiter.size() > 10_000); - // Expire all of the above, then add one more failure right away: since the minimum cleanup interval hasn't - // elapsed since the burst above already attempted (and failed) a cleanup, this must not re-scan yet. - clock.advance(WINDOW.plusSeconds(1)); - limiter.recordFailure("key-after-expiry-1"); + // A few seconds later (window not yet expired, throttle interval not yet elapsed): a cleanup attempt must + // not be allowed to re-scan the whole map again so soon after the burst's own (no-op) attempt. + clock.advance(Duration.ofSeconds(5)); + limiter.recordFailure("key-after-burst-1"); Assert.assertTrue(limiter.size() > 10_000, "cleanup should not have run again before the throttle interval elapsed"); - // Once the throttle interval has also elapsed, the next recordFailure() should finally be allowed to sweep - // out all of the now-expired entries. - clock.advance(Duration.ofSeconds(10)); - limiter.recordFailure("key-after-expiry-2"); + // Once both the failure window and the throttle interval have elapsed, the next recordFailure() should + // finally be allowed to sweep out all of the now-expired entries. + clock.advance(WINDOW); + limiter.recordFailure("key-after-burst-2"); Assert.assertTrue(limiter.size() < 10_000, "cleanup should have run once the throttle interval elapsed"); }