#163 - add accountId and userToken filter to SSE endpoint

osgi-bundles/bundles/logger/html-sse-demo.md

@@ -4,25 +4,28 @@
 2. Copy the JAR to `platform` plugin directory
 3. Copy HTML content below to `<any-directory>`
 4. Run this using `python3 -m http.server 3000 -d <any-directory>` or `npx serve -l 3000 <any-directory>`
-5. Open `http://localhost:3000`, fill in the Base URL and your Kill Bill username/password, then click **Connect**
+5. Open `http://localhost:3000`, fill in the Base URL and your Kill Bill username/password, optionally the tenant API key/secret and the filters, 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
-`<script>` tag below, so there's still nothing to install — just copy the HTML and go.
+### Filter combinations
 
-**CORS note:** this setup is cross-origin (demo on :3000, Kill Bill on :8080). Sending a custom
-`Authorization` header triggers a CORS preflight (`OPTIONS`); `BasicAuthFilter` explicitly lets
-`OPTIONS` through unauthenticated for this reason. If you serve the demo from the same origin as
-Kill Bill instead, the preflight doesn't happen at all.
+Filters combine with AND. Empty fields are ignored. Entries lacking the filtered attribute are excluded.
+`multitenant=false` servers skip the tenant checks (no `401`/`403`).
 
-**Filter note:** the `accountId`/`userToken` query-param filtering (`LogEntriesFilter`) is a separate,
-not-yet-merged feature (`platform-163`) and isn't wired up on this branch. The corresponding UI fields
-are kept below, commented out, so they're easy to restore once that feature lands here.
+| API key + secret | accountId | userToken | Result |
+|---|---|---|---|
+| - | - | - | Every entry, all tenants |
+| valid | - | - | Entries of that tenant |
+| valid | valid UUID, known | - | Entries of that tenant and account |
+| valid | valid UUID, unknown or other tenant's | - | Connected, heartbeats only |
+| - | - | token | Entries with that token, all tenants |
+| valid | - | token | Entries of that tenant with that token |
+| valid | valid UUID, known | token | Entries of that tenant and account with that token |
+| wrong, or only key or only secret | any | any | `401` |
+| - | any `accountId` | any | `403` |
+| valid | not a UUID | any | `400` |
+| any | `accountId` or `userToken` given twice | any | `400` |
+
+### HTML
 
 ```html
 <!DOCTYPE html>
@@ -70,17 +73,22 @@ are kept below, commented out, so they're easy to restore once that feature land
             <label>Password</label>
             <input id="password" type="password" placeholder="password" />
         </div>
-        <!--
-        TODO: re-enable once LogEntriesFilter / RecordIdApi (platform-163) lands on this branch.
         <div class="field">
-            <label>accountId (UUID)</label>
+            <label>X-Killbill-ApiKey (tenant)</label>
+            <input id="apiKey" placeholder="optional" />
+        </div>
+        <div class="field">
+            <label>X-Killbill-ApiSecret (tenant)</label>
+            <input id="apiSecret" type="password" placeholder="optional" />
+        </div>
+        <div class="field">
+            <label>accountId (UUID, needs API key/secret)</label>
             <input id="accountId" placeholder="optional" />
         </div>
         <div class="field">
             <label>userToken</label>
             <input id="userToken" placeholder="optional" />
         </div>
-        -->
         <button id="connectBtn">Connect</button>
         <button id="clearBtn">Clear</button>
         <span id="status" class="status closed">● Disconnected</span>
@@ -115,14 +123,12 @@ ${n}`,h++;break;case"id":n.includes("\0")||(f=n);break;case"retry":/^\d+$/.test(
 
         function buildUrl() {
             const base = document.getElementById('baseUrl').value.replace(/\/$/, '');
-            // TODO: re-enable once LogEntriesFilter / RecordIdApi (platform-163) lands on this branch.
-            // const accountId = document.getElementById('accountId').value.trim();
-            // const userToken = document.getElementById('userToken').value.trim();
-            // const params = [];
-            // if (accountId) params.push('accountId=' + encodeURIComponent(accountId));
-            // if (userToken) params.push('userToken=' + encodeURIComponent(userToken));
-            // return params.length > 0 ? base + '/?' + params.join('&') : base + '/';
-            return base + '/';
+            const accountId = document.getElementById('accountId').value.trim();
+            const userToken = document.getElementById('userToken').value.trim();
+            const params = [];
+            if (accountId) params.push('accountId=' + encodeURIComponent(accountId));
+            if (userToken) params.push('userToken=' + encodeURIComponent(userToken));
+            return params.length > 0 ? base + '/?' + params.join('&') : base + '/';
         }
 
         function buildAuthHeader() {
@@ -135,6 +141,17 @@ ${n}`,h++;break;case"id":n.includes("\0")||(f=n);break;case"retry":/^\d+$/.test(
             return 'Basic ' + utf8ToBase64(username + ':' + password);
         }
 
+        // Tenant headers, only when both are filled in; a half-filled pair is sent as is so the
+        // server's 401 for an incomplete pair can be observed.
+        function buildTenantHeaders() {
+            const apiKey = document.getElementById('apiKey').value.trim();
+            const apiSecret = document.getElementById('apiSecret').value.trim();
+            const headers = {};
+            if (apiKey) headers['X-Killbill-ApiKey'] = apiKey;
+            if (apiSecret) headers['X-Killbill-ApiSecret'] = apiSecret;
+            return headers;
+        }
+
         function utf8ToBase64(str) {
             const bytes = new TextEncoder().encode(str);
             let binary = '';
@@ -177,13 +194,15 @@ ${n}`,h++;break;case"id":n.includes("\0")||(f=n);break;case"retry":/^\d+$/.test(
 
             const url = buildUrl();
             const authHeader = buildAuthHeader();
+            const tenantHeaders = buildTenantHeaders();
             appendMessage('Connecting to: ' + url, 'msg-system');
 
-            // Custom fetch: injects the Basic Auth header on every (re)connect attempt, since the
-            // native EventSource API has no way to set request headers at all.
+            // Custom fetch: injects the Basic Auth and tenant headers on every (re)connect attempt,
+            // since the native EventSource API has no way to set request headers at all.
             const authFetch = function(input, init) {
                 const headers = new Headers((init && init.headers) || {});
                 if (authHeader) headers.set('Authorization', authHeader);
+                Object.keys(tenantHeaders).forEach(function(name) { headers.set(name, tenantHeaders[name]); });
                 return fetch(input, Object.assign({}, init, { headers: headers }));
             };
 
@@ -224,7 +243,14 @@ ${n}`,h++;break;case"id":n.includes("\0")||(f=n);break;case"retry":/^\d+$/.test(
                 // leave a stale instance quietly reconnecting in the background after we've already
                 // torn down the UI/state for it (which otherwise looks like "stuck/always connected"
                 // and makes the Disconnect button appear to do nothing).
-                const detail = event.code ? (' (HTTP ' + event.code + (event.message ? ': ' + event.message : '') + ')') : '';
+                const hints = {
+                    400: 'accountId is not a UUID, or accountId/userToken is repeated',
+                    401: 'bad Basic Auth credentials, or a wrong/incomplete API key + secret',
+                    403: 'accountId requires the API key + secret',
+                    429: 'too many failed attempts, retry later'
+                };
+                const hint = hints[event.code] ? ' - ' + hints[event.code] : '';
+                const detail = event.code ? (' (HTTP ' + event.code + (event.message ? ': ' + event.message : '') + hint + ')') : '';
                 appendMessage('Connection error' + detail, 'msg-error');
                 if (eventSource) {
                     eventSource.close();

osgi-bundles/bundles/logger/pom.xml

@@ -99,6 +99,17 @@
             <groupId>org.kill-bill.commons</groupId>
             <artifactId>killbill-utils</artifactId>
         </dependency>
+        <!-- Used only by TestLogsSseEndpoint, but it must stay compile scope: jooby already brings it in at compile scope,
+             so dependency-scope-maven-plugin rejects test scope and maven-dependency-plugin (analyze) requires it declared. -->
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-server</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-servlet</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>

osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java

@@ -59,6 +59,9 @@ public void start(final BundleContext context) throws Exception {
         // 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()));
+        // TenantFilter is only installed when multi-tenancy is enabled (the default, when the property is missing)
+        final boolean multiTenant = !"false".equalsIgnoreCase(configProperties.getString("org.killbill.server.multitenant"));
+        pluginApp.use("*", new LogsSseFilter(killbillAPI.getRecordIdApi(), multiTenant));
         logsSseHandler = new LogsSseHandler(logEntriesManager);
         pluginApp.sse("/", logsSseHandler);
         final HttpServlet httpServlet = PluginApp.createServlet(pluginApp);

osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseFilter.java

@@ -0,0 +1,155 @@
+/*
+ * 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.List;
+import java.util.Optional;
+import java.util.UUID;
+
+import org.jooby.Mutant;
+import org.jooby.Request;
+import org.jooby.Response;
+import org.jooby.Route;
+import org.killbill.billing.ObjectType;
+import org.killbill.billing.plugin.api.PluginTenantContext;
+import org.killbill.billing.tenant.api.Tenant;
+import org.killbill.billing.util.api.RecordIdApi;
+import org.killbill.billing.util.callcontext.TenantContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+// Validates the filtering parameters (accountId, userToken) and tenant headers, and resolves them once per connection
+// into a SseParamsCriteria stored on the request. This must be a filter (not done in the SSE handler): by the time the
+// handler runs, the 200 status and text/event-stream content type have already been sent, so errors can't be returned.
+//
+// Evaluation order (first failure answers): 401 (tenant headers not validated), 403 (accountId without tenant), 400 (syntax).
+class LogsSseFilter implements Route.Filter {
+
+    private static final Logger logger = LoggerFactory.getLogger(LogsSseFilter.class);
+
+    static final String TENANT_ATTRIBUTE = "killbill_tenant"; // Set by Kill Bill's TenantFilter after validating the API key/secret
+    static final String API_KEY_HEADER = "X-Killbill-ApiKey";
+    static final String API_SECRET_HEADER = "X-Killbill-ApiSecret";
+    static final String ACCOUNT_ID_PARAM = "accountId";
+    static final String USER_TOKEN_PARAM = "userToken";
+
+    private final RecordIdApi recordIdApi;
+    private final boolean multiTenant;
+
+    LogsSseFilter(final RecordIdApi recordIdApi, final boolean multiTenant) {
+        this.recordIdApi = recordIdApi;
+        this.multiTenant = multiTenant;
+    }
+
+    @Override
+    public void handle(final Request req, final Response rsp, final Route.Chain chain) throws Throwable {
+        // CORS preflight is answered by BasicAuthFilter and never reaches here, but be safe if the order changes
+        if ("OPTIONS".equalsIgnoreCase(req.method())) {
+            chain.next(req, rsp);
+            return;
+        }
+
+        final Tenant tenant = req.<Tenant>ifGet(TENANT_ATTRIBUTE).orElse(null);
+
+        // On plugin paths TenantFilter lets a request with a wrong or half-supplied key/secret through, without tenant
+        if (multiTenant && tenant == null && (req.header(API_KEY_HEADER).isSet() || req.header(API_SECRET_HEADER).isSet())) {
+            reject(rsp, 401, "Invalid " + API_KEY_HEADER + " / " + API_SECRET_HEADER);
+            return;
+        }
+
+        final List<String> accountIds = values(req, ACCOUNT_ID_PARAM);
+        final List<String> userTokens = values(req, USER_TOKEN_PARAM);
+        final String accountId = nonBlank(accountIds);
+        final String userToken = nonBlank(userTokens);
+
+        // accountId lookups are tenant-scoped
+        if (multiTenant && tenant == null && accountId != null) {
+            reject(rsp, 403, ACCOUNT_ID_PARAM + " requires " + API_KEY_HEADER + " / " + API_SECRET_HEADER);
+            return;
+        }
+
+        if (accountIds.size() > 1 || userTokens.size() > 1) {
+            reject(rsp, 400, "Parameters " + ACCOUNT_ID_PARAM + " and " + USER_TOKEN_PARAM + " can be specified only once");
+            return;
+        }
+
+        final UUID accountUuid;
+        if (accountId == null) {
+            accountUuid = null;
+        } else {
+            try {
+                accountUuid = UUID.fromString(accountId);
+            } catch (final IllegalArgumentException e) {
+                reject(rsp, 400, "Invalid " + ACCOUNT_ID_PARAM + ": not a UUID");
+                return;
+            }
+        }
+
+        final UUID tenantId = tenant == null ? null : tenant.getId();
+        String tenantRecordId = null;
+        if (tenantId != null) {
+            tenantRecordId = recordId(tenantId, ObjectType.TENANT, new PluginTenantContext(null, tenantId));
+            if (tenantRecordId == null) {
+                rejectUnresolvableTenant(rsp, tenantId);
+                return;
+            }
+        }
+
+        String accountRecordId = null;
+        if (accountUuid != null) {
+            accountRecordId = recordId(accountUuid, ObjectType.ACCOUNT, new PluginTenantContext(accountUuid, tenantId));
+        }
+
+        req.set(SseParamsCriteria.REQUEST_KEY, new SseParamsCriteria(tenantRecordId, accountUuid != null, accountRecordId, userToken));
+        chain.next(req, rsp);
+    }
+
+    // A tenant validated by TenantFilter must have a record id. If not, this is a server-side problem (not the client's),
+    // and continuing without tenant scoping would silently expose the entries of all tenants. Fail closed: answer 500
+    // before the stream opens, which also closes the connection.
+    private void rejectUnresolvableTenant(final Response rsp, final UUID tenantId) throws Throwable {
+        logger.error("Unable to resolve the record id of validated tenant {}, rejecting the log stream request", tenantId);
+        reject(rsp, 500, "Unable to resolve the tenant");
+    }
+
+    // Returns null when the object is unknown (or belongs to another tenant)
+    private String recordId(final UUID objectId, final ObjectType objectType, final TenantContext context) {
+        final Long recordId = recordIdApi.getRecordId(objectId, objectType, context);
+        if (recordId == null || recordId <= 0L) {
+            logger.debug("No record id for {} {}", objectType, objectId);
+            return null;
+        }
+        return recordId.toString();
+    }
+
+    private static List<String> values(final Request req, final String name) {
+        final Mutant param = req.param(name);
+        return param.isSet() ? param.toList(String.class) : List.of();
+    }
+
+    // A bare (?p), empty (?p=) or whitespace-only value means "not provided"
+    private static String nonBlank(final List<String> values) {
+        final Optional<String> first = values.stream().filter(v -> v != null && !v.isBlank()).findFirst();
+        return first.orElse(null);
+    }
+
+    private static void reject(final Response rsp, final int status, final String message) throws Throwable {
+        rsp.status(status);
+        rsp.send(message);
+    }
+}