From 0735d7522f01c4835c349d4746f1a566740d9ee5 Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Thu, 29 Oct 2026 21:59:49 +0100 Subject: [PATCH 1/4] Harden the receiver against a review of the Go and Python ports A code review of the sibling libraries found weaknesses that the Java receiver shares. The URI policy took any host name starting with "127." for loopback and so allowed plain http for it; it now accepts only localhost and names under it, 127.0.0.0/8 and ::1 (also IPv4-mapped), the same rule as the Go library. The poll endpoint is held to that policy before the access token is sent to it, configured or taken from the stream, with a new SsfPoller.setUriPolicy. Response bodies read from the transmitter are bounded, 4 MiB by default, through JdkSsfHttpClient.setMaxResponseSize and the Spring Boot property easyssf.receiver.http.max-response-size. A polled SET listed under a key other than its jti is reported as invalid_request under that key instead of handled. A polling thread ends as soon as it is no longer the poller's thread, so a stop() that timed out on a handler no longer leaves it running next to a restart. SsfStreamRegistrar.start() is idempotent, stop() joins the registration and a registration that completes after stop() publishes nothing. The stream client checks the issuer of every stream the transmitter returns, not only of a created one. Fixes #61 Signed-off-by: Thomas Darimont --- CHANGELOG.md | 35 ++++ README.md | 4 +- .../spring/boot/SsfReceiverProperties.java | 17 ++ .../spring/boot/SsfTransmitterFactory.java | 9 + .../SsfReceiverAutoConfiguration.java | 1 + ...sfReceiverRestClientAutoConfiguration.java | 4 +- .../boot/http/RestClientSsfHttpClient.java | 21 ++- .../SsfReceiverAutoConfigurationTests.java | 28 +++ .../http/RestClientSsfHttpClientTests.java | 13 ++ .../README.md | 8 +- .../receiver/http/JdkSsfHttpClient.java | 36 +++- .../receiver/http/SsfHttpResponse.java | 65 +++++++ .../org/easyssf/receiver/poll/SsfPoller.java | 86 ++++++++- .../easyssf/receiver/set/SsfSetClaims.java | 16 ++ .../receiver/stream/SsfStreamClient.java | 50 ++++-- .../SsfStreamIssuerMismatchException.java | 10 +- .../receiver/stream/SsfStreamRegistrar.java | 168 ++++++++++++------ .../transmitter/SsfTransmitterUriPolicy.java | 48 ++++- .../receiver/http/JdkSsfHttpClientTests.java | 99 +++++++++++ .../easyssf/receiver/poll/SsfPollerTests.java | 42 +++++ .../stream/SsfStreamManagementTests.java | 118 ++++++++++++ .../SsfTransmitterUriPolicyTests.java | 30 ++++ 22 files changed, 811 insertions(+), 97 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9d5759d..11439c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,8 +6,43 @@ The notable changes of every release. The section of a version is the text of it ## [Unreleased] +### Security + +- A host name that merely starts with `127.`, such as `127.attacker.example`, counted as a loopback + address and was allowed to use plain `http` under the default URI policy. `SsfTransmitterUriPolicy` + now takes only `localhost` and names under it, IPv4 addresses in `127.0.0.0/8` and `::1` (also as + the IPv4-mapped `::ffff:127.0.0.1`) for loopback, the same rule as the Go library. +- The poll endpoint is held to the URI policy of the transmitter's metadata endpoints before the + access token is sent to it, whether it was configured (`easyssf.receiver.poll.endpoint-url`, checked + at startup) or taken from the stream configuration the transmitter returned: `https`, or `http` on + loopback addresses only, unless `allow-insecure-http` is set. New API: `SsfPoller.setUriPolicy`, + which the Quarkus extension can wire to its insecure-http setting. +- Response bodies the receiver reads from the transmitter (metadata, JWK Set, stream configurations, + poll responses) are bounded, 4 MiB by default; a larger response fails the call as if the + transmitter were unavailable instead of being buffered. New API: `JdkSsfHttpClient.setMaxResponseSize` + and `SsfHttpResponse.readBody` for other `SsfHttpClient` implementations, and the property + `easyssf.receiver.http.max-response-size` of the Spring Boot starter, which the Quarkus extension + can mirror. + +### Changed + +- `SsfStreamClient` checks the issuer of every stream configuration the transmitter returns, not only + of a created one: `getStream`, `getStreams`, `updateStream` and `replaceStream` throw + `SsfStreamIssuerMismatchException` as `createStream` did, as the SETs of a stream with another + issuer would not verify. `SsfStreamRegistrar` deletes a stream it created with the wrong issuer, as + before, and leaves an existing one alone; it fails without retrying either way. + ### Fixed +- A SET the poll response lists under a key other than its `jti` (RFC 8936, section 2.4) is reported + to the transmitter as `invalid_request` under that key instead of being handled and acknowledged + under a key that is not its own. +- `SsfPoller.stop()` followed by `start()` while a handler was still running left the old polling + thread alive next to the new one once the handler returned; a polling thread now ends as soon as it + is no longer the poller's thread. `stop()` logs when the thread is still busy after five seconds. +- `SsfStreamRegistrar.start()` is idempotent and `stop()` waits for the registration thread to end + before deleting the stream. A registration that completes after `stop()` publishes nothing and + deletes a stream it created on the way if the registrar deletes on shutdown. - The `easyssf.receiver.poll.pending-acks` gauge no longer turns `NaN` after a garbage collection. Micrometer refers to what a gauge reads weakly, and the poller handed it a method reference nothing else held on to; the poller now keeps that reference for as long as it lives, so the diff --git a/README.md b/README.md index 21d4896..eb37c3d 100644 --- a/README.md +++ b/README.md @@ -310,7 +310,9 @@ verifier.setExpectedAudience("https://my-app.example"); // metadata.resolve() also tells specVersion(), deliveryMethodsSupported(), criticalSubjectMembers(), // authorizationSchemes() and defaultSubjects(). The resolver requires https for the issuer and the // endpoints of the metadata (http on loopback addresses only); pass SsfTransmitterUriPolicy.INSECURE -// as a fourth argument for a development setup. +// as a fourth argument for a development setup. SsfPoller.setUriPolicy holds the poll endpoint to the +// same rule, as the access token goes there. Response bodies are bounded, see +// JdkSsfHttpClient.setMaxResponseSize. SsfEventHandler handler = (eventContext) -> { if (eventContext.hasEvent("CaepSessionRevoked")) { diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java index 5620262..08838fc 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfReceiverProperties.java @@ -7,10 +7,12 @@ import java.util.Map; import org.easyssf.core.event.SsfSubject; +import org.easyssf.receiver.http.SsfHttpResponse; import org.easyssf.receiver.jdbc.JdbcSsfPollAckStore; import org.easyssf.receiver.session.SsfSubjectClaimsMatcher; import org.easyssf.receiver.transmitter.SsfTransmitter; import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.util.unit.DataSize; /** * Configuration properties for the OpenID Shared Signals Framework (SSF) receiver. @@ -180,6 +182,13 @@ public static class Http { */ private String userAgent; + /** + * The largest response body accepted from the transmitter; a larger response + * fails the call. Metadata, JWK Sets, stream configurations and poll responses + * stay far below the default of 4MB. + */ + private DataSize maxResponseSize = DataSize.ofBytes(SsfHttpResponse.DEFAULT_MAX_BODY_SIZE); + public boolean isUseRestClient() { return this.useRestClient; } @@ -212,6 +221,14 @@ public void setUserAgent(String userAgent) { this.userAgent = userAgent; } + public DataSize getMaxResponseSize() { + return this.maxResponseSize; + } + + public void setMaxResponseSize(DataSize maxResponseSize) { + this.maxResponseSize = maxResponseSize; + } + } public static class Metrics { diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java index 0066a8a..15005d2 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/SsfTransmitterFactory.java @@ -232,6 +232,7 @@ public static SsfHttpClient pollHttpClient(String name, SsfTransmitterProperties JdkSsfHttpClient pollClient = new JdkSsfHttpClient(connectTimeout, poll.getLongPollingHold().plusSeconds(15)); if (http != null) { pollClient.setUserAgent(http.getUserAgent()); + pollClient.setMaxResponseSize(http.getMaxResponseSize().toBytes()); } logger.info(prefix(name) + "poll.long-polling is on, the poller calls the transmitter with the JDK HTTP " + "client rather than the RestClient of the application"); @@ -260,6 +261,7 @@ public static SsfPoller poller(String name, SsfTransmitterProperties properties, if (poll.isLongPolling()) { poller.setLongPolling(poll.getLongPollingHold()); } + poller.setUriPolicy(uriPolicy(properties)); poller.setTransmitter(properties.getTransmitterIssuer()); poller.setMetrics((metrics != null) ? metrics : SsfReceiverMetrics.NOOP); poller.setInterval(poll.getInterval()); @@ -373,6 +375,13 @@ private static Supplier pollEndpoint(String name, SsfTransmitterProperties SsfReceiverStream receiverStream) { URI configured = properties.getPoll().getEndpointUrl(); if (configured != null) { + try { + uriPolicy(properties).checkEndpoint(configured, "The poll endpoint"); + } + catch (IllegalArgumentException ex) { + throw new InvalidConfigurationPropertyValueException(prefix(name) + "poll.endpoint-url", configured, + ex.getMessage()); + } return () -> configured; } if (!registersStream(properties)) { diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java index 22ff87d..3ab55bf 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfiguration.java @@ -100,6 +100,7 @@ JdkSsfHttpClient ssfHttpClient(SsfReceiverProperties properties) { : SsfReceiverProperties.Http.DEFAULT_TIMEOUT, (http.getReadTimeout() != null) ? http.getReadTimeout() : SsfReceiverProperties.Http.DEFAULT_TIMEOUT); httpClient.setUserAgent(http.getUserAgent()); + httpClient.setMaxResponseSize(http.getMaxResponseSize().toBytes()); return httpClient; } diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverRestClientAutoConfiguration.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverRestClientAutoConfiguration.java index 4f995d8..fe4e24f 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverRestClientAutoConfiguration.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverRestClientAutoConfiguration.java @@ -68,7 +68,9 @@ RestClientSsfHttpClient ssfRestClientHttpClient(RestClient.Builder restClientBui builder.defaultHeader(HttpHeaders.USER_AGENT, http.getUserAgent()); } logger.info("The SSF transmitter is called with the RestClient of the application"); - return new RestClientSsfHttpClient(builder.build()); + RestClientSsfHttpClient httpClient = new RestClientSsfHttpClient(builder.build()); + httpClient.setMaxResponseSize(http.getMaxResponseSize().toBytes()); + return httpClient; } /** diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/http/RestClientSsfHttpClient.java b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/http/RestClientSsfHttpClient.java index 3307009..7cc51a1 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/http/RestClientSsfHttpClient.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/main/java/org/easyssf/receiver/spring/boot/http/RestClientSsfHttpClient.java @@ -26,11 +26,27 @@ public class RestClientSsfHttpClient implements SsfHttpClient { private final RestClient restClient; + private long maxResponseSize = SsfHttpResponse.DEFAULT_MAX_BODY_SIZE; + public RestClientSsfHttpClient(RestClient restClient) { Assert.notNull(restClient, "restClient must not be null"); this.restClient = restClient; } + /** + * Bounds the response bodies, {@link SsfHttpResponse#DEFAULT_MAX_BODY_SIZE 4 MiB} by + * default; a larger response fails the call as if none was received. + * @param maxResponseSize the largest response body read, in bytes + */ + public void setMaxResponseSize(long maxResponseSize) { + Assert.isTrue(maxResponseSize > 0, "maxResponseSize must be positive"); + this.maxResponseSize = maxResponseSize; + } + + public long getMaxResponseSize() { + return this.maxResponseSize; + } + @Override public SsfHttpResponse execute(SsfHttpRequest request) throws IOException { RestClient.RequestBodySpec spec = this.restClient.method(HttpMethod.valueOf(request.method())) @@ -47,8 +63,9 @@ public SsfHttpResponse execute(SsfHttpRequest request) throws IOException { MediaType contentType = clientResponse.getHeaders().getContentType(); Charset charset = (contentType != null && contentType.getCharset() != null) ? contentType.getCharset() : StandardCharsets.UTF_8; - String body = new String(clientResponse.getBody().readAllBytes(), charset); - return new SsfHttpResponse(clientResponse.getStatusCode().value(), headers, body); + byte[] body = SsfHttpResponse.readBody(clientResponse.getBody(), + clientResponse.getHeaders().getContentLength(), this.maxResponseSize, request.uri()); + return new SsfHttpResponse(clientResponse.getStatusCode().value(), headers, new String(body, charset)); }); } catch (RestClientException ex) { diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java index 9ee56af..a9cfbae 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/autoconfigure/SsfReceiverAutoConfigurationTests.java @@ -386,6 +386,15 @@ void userAgentForCallsToTheTransmitterIsConfigurable() { .isEqualTo("my-receiver/1.0")); } + @Test + void responseSizeForCallsToTheTransmitterIsConfigurable() { + this.webContextRunner.run((context) -> assertThat(context.getBean(JdkSsfHttpClient.class).getMaxResponseSize()) + .isEqualTo(4L * 1024 * 1024)); + this.webContextRunner.withPropertyValues("easyssf.receiver.http.max-response-size=64KB") + .run((context) -> assertThat(context.getBean(JdkSsfHttpClient.class).getMaxResponseSize()) + .isEqualTo(64 * 1024)); + } + @Test void transmitterIsCalledWithRestClientOfTheApplication() { try (TestTransmitter transmitter = new TestTransmitter()) { @@ -536,6 +545,25 @@ void pollDeliveryReplacesPushEndpoint() { }); } + @Test + void pollEndpointWithoutTlsIsRefusedUnlessInsecureHttpIsAllowed() { + this.webContextRunner + .withPropertyValues("easyssf.receiver.delivery-method=poll", "easyssf.receiver.poll.auto-startup=false", + "easyssf.receiver.poll.endpoint-url=http://idp.example/realms/test/poll") + .run((context) -> { + assertThat(context).hasFailed(); + assertThat(context.getStartupFailure()).rootCause() + .isInstanceOf(InvalidConfigurationPropertyValueException.class) + .hasMessageContaining("easyssf.receiver.poll.endpoint-url") + .hasMessageContaining("must use https"); + }); + this.webContextRunner + .withPropertyValues("easyssf.receiver.delivery-method=poll", "easyssf.receiver.poll.auto-startup=false", + "easyssf.receiver.poll.endpoint-url=http://idp.example/realms/test/poll", + "easyssf.receiver.allow-insecure-http=true") + .run((context) -> assertThat(context).hasSingleBean(SsfPoller.class)); + } + @Test void pollDeliveryNeedsPollEndpoint() { this.webContextRunner.withPropertyValues("easyssf.receiver.delivery-method=poll").run((context) -> { diff --git a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/http/RestClientSsfHttpClientTests.java b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/http/RestClientSsfHttpClientTests.java index a03bcad..c83d2d0 100644 --- a/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/http/RestClientSsfHttpClientTests.java +++ b/easyssf-receiver-spring-boot-autoconfigure/src/test/java/org/easyssf/receiver/spring/boot/http/RestClientSsfHttpClientTests.java @@ -2,6 +2,7 @@ import java.io.IOException; import java.net.URI; +import java.nio.charset.StandardCharsets; import java.util.List; import org.easyssf.core.stream.SsfStreamConfiguration; @@ -49,6 +50,18 @@ void returnsResponseWhateverItsStatus() throws Exception { assertThat(unauthorized.isSuccessful()).isFalse(); } + @Test + void cutsOffResponseLargerThanTheMaximum() throws Exception { + RestClientSsfHttpClient bounded = new RestClientSsfHttpClient(RestClient.create()); + URI jwksUri = URI.create(transmitter.jwksUri()); + int size = bounded.execute(SsfHttpRequest.get(jwksUri)).body().getBytes(StandardCharsets.UTF_8).length; + bounded.setMaxResponseSize(size - 1); + assertThatIOException().isThrownBy(() -> bounded.execute(SsfHttpRequest.get(jwksUri))) + .withMessageContaining("larger than the " + (size - 1) + " bytes accepted"); + bounded.setMaxResponseSize(size); + assertThat(bounded.execute(SsfHttpRequest.get(jwksUri)).status()).isEqualTo(200); + } + @Test void reportsFailedConnectionAsIOException() { assertThatIOException().isThrownBy( diff --git a/easyssf-receiver-spring-boot-starter/README.md b/easyssf-receiver-spring-boot-starter/README.md index fac50d6..f7732a2 100644 --- a/easyssf-receiver-spring-boot-starter/README.md +++ b/easyssf-receiver-spring-boot-starter/README.md @@ -186,7 +186,8 @@ easyssf: ``` - The poll endpoint is taken from the stream (`stream.id` or a stream managed by the receiver), or set - it with `easyssf.receiver.poll.endpoint-url`. + it with `easyssf.receiver.poll.endpoint-url`. Either way it must use `https` (`http` on loopback + addresses only, or with `allow-insecure-http`), as the access token is sent to it. - A SET is acknowledged after it was handled. An invalid SET is reported to the transmitter (`setErrs`). A SET that could not be handled is neither acknowledged nor reported and is delivered again. - If more SETs are available than `poll.max-events`, they are fetched right away. @@ -434,7 +435,7 @@ name, with its own status. Switch it off with `management.health.easyssf.enabled | `easyssf.receiver.transmitters..*` | | Further transmitters, see [Several transmitters](#several-transmitters): every property of a transmitter, from `transmitter-issuer` to `set-validation.*`, under its name. | | `easyssf.receiver.transmitter-metadata-url` | derived | By default `/.well-known/ssf-configuration` (SSF 1.0, section 7.2), falling back to `/.well-known/ssf-configuration`. | | `easyssf.receiver.transmitter-jwks-url` | from metadata | Skips metadata discovery when set. | -| `easyssf.receiver.allow-insecure-http` | `false` | Accepts `http` for the transmitter issuer and the endpoints it publishes on any host, with a warning on startup. For development only. | +| `easyssf.receiver.allow-insecure-http` | `false` | Accepts `http` for the transmitter issuer, the endpoints it publishes and the poll endpoint on any host, with a warning on startup. For development only. | | `easyssf.receiver.expected-audience` | | Every SET must contain it in `aud`. Without it the audience of the stream the receiver registers or looks up (`stream.management=receiver`, `stream.id`) is required instead, once the stream is known. One of the two, or `set-validation.allow-any-audience`, is required: a receiver that cannot know the audience of its SETs does not start. | | `easyssf.receiver.delivery-method` | `push` | `push` or `poll`. | | `easyssf.receiver.understood-subject-members` | `user, session, device, application, tenant, org_unit, group` | The members of a complex subject the application interprets; a SET with a member the transmitter declared critical that is not listed is rejected. | @@ -442,6 +443,7 @@ name, with its own status. Switch it off with `management.health.easyssf.enabled | `easyssf.receiver.http.use-rest-client` | `true` | Call the transmitter with the `RestClient` of the application if it has a `RestClient.Builder`. | | `easyssf.receiver.http.connect-timeout` / `read-timeout` | `5s` | Calls to the transmitter. Unset, `spring.http.clients.*` applies to the `RestClient` before the 5 seconds do. | | `easyssf.receiver.http.user-agent` | default of the HTTP client | `User-Agent` header for calls to the transmitter. | +| `easyssf.receiver.http.max-response-size` | `4MB` | Largest response body accepted from the transmitter (metadata, JWK Set, stream configuration, poll response); a larger one fails the call. | | `easyssf.receiver.transmitter-access-token` | | Static access token for stream management and polling. | | `easyssf.receiver.oauth2.token-uri` | | Token endpoint for the client credentials grant. | | `easyssf.receiver.oauth2.client-id` / `client-secret` | | | @@ -454,7 +456,7 @@ name, with its own status. Switch it off with `management.health.easyssf.enabled | `easyssf.receiver.stream.events-requested` | `CaepSessionRevoked`, `CaepCredentialChange` | For a stream managed by the receiver. | | `easyssf.receiver.stream.description` | | For a stream managed by the receiver. | | `easyssf.receiver.stream.delete-on-shutdown` | `false` | For a stream managed by the receiver. | -| `easyssf.receiver.poll.endpoint-url` | from the stream | | +| `easyssf.receiver.poll.endpoint-url` | from the stream | Held to the same `https` rule as the transmitter issuer. | | `easyssf.receiver.poll.auto-startup` | `true` | | | `easyssf.receiver.poll.interval` | `30s` | | | `easyssf.receiver.poll.initial-delay` | `1s` | | diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java index 13f8e8d..6c81a28 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/JdkSsfHttpClient.java @@ -1,6 +1,7 @@ package org.easyssf.receiver.http; import java.io.IOException; +import java.io.InputStream; import java.io.InterruptedIOException; import java.net.http.HttpClient; import java.net.http.HttpRequest; @@ -10,7 +11,9 @@ import org.easyssf.core.support.SsfAssert; /** - * {@link SsfHttpClient} that uses the HTTP client of the JDK. + * {@link SsfHttpClient} that uses the HTTP client of the JDK. A response body larger than + * {@link #setMaxResponseSize the maximum} is not buffered: the call fails as if no + * response was received. */ public class JdkSsfHttpClient implements SsfHttpClient { @@ -22,6 +25,8 @@ public class JdkSsfHttpClient implements SsfHttpClient { private String userAgent; + private long maxResponseSize = SsfHttpResponse.DEFAULT_MAX_BODY_SIZE; + public JdkSsfHttpClient() { this(Duration.ofSeconds(5), Duration.ofSeconds(5)); } @@ -59,6 +64,21 @@ public String getUserAgent() { return this.userAgent; } + /** + * Bounds the response bodies, {@link SsfHttpResponse#DEFAULT_MAX_BODY_SIZE 4 MiB} by + * default. A transmitter that sends more is treated as unavailable: the receiver + * never buffers a response of a size the transmitter chooses. + * @param maxResponseSize the largest response body read, in bytes + */ + public void setMaxResponseSize(long maxResponseSize) { + SsfAssert.isTrue(maxResponseSize > 0, "maxResponseSize must be positive"); + this.maxResponseSize = maxResponseSize; + } + + public long getMaxResponseSize() { + return this.maxResponseSize; + } + @Override public SsfHttpResponse execute(SsfHttpRequest request) throws IOException { HttpRequest.BodyPublisher body = (request.body() != null) ? HttpRequest.BodyPublishers.ofString(request.body()) @@ -71,8 +91,18 @@ public SsfHttpResponse execute(SsfHttpRequest request) throws IOException { builder.header(USER_AGENT, this.userAgent); } try { - HttpResponse response = this.httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofString()); - return new SsfHttpResponse(response.statusCode(), response.headers().map(), response.body()); + HttpResponse response = this.httpClient.send(builder.build(), + HttpResponse.BodyHandlers.ofInputStream()); + byte[] bytes; + // closing the stream ends the response early if the body is cut off + try (InputStream input = response.body()) { + bytes = SsfHttpResponse.readBody(input, + response.headers().firstValueAsLong("Content-Length").orElse(-1), this.maxResponseSize, + request.uri()); + } + String contentType = response.headers().firstValue("Content-Type").orElse(null); + return new SsfHttpResponse(response.statusCode(), response.headers().map(), + new String(bytes, SsfHttpResponse.charsetOf(contentType))); } catch (InterruptedException ex) { Thread.currentThread().interrupt(); diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpResponse.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpResponse.java index d2e7893..f3f7336 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpResponse.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/http/SsfHttpResponse.java @@ -1,5 +1,10 @@ package org.easyssf.receiver.http; +import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; @@ -12,6 +17,66 @@ */ public record SsfHttpResponse(int status, Map> headers, String body) { + /** + * How large a response body may be unless an {@link SsfHttpClient} is configured + * otherwise, 4 MiB. The receiver reads transmitter metadata, JWK Sets, stream + * configurations and poll responses, none of which comes near that, so a larger + * response is a fault or an attack and is cut off rather than buffered. + */ + public static final long DEFAULT_MAX_BODY_SIZE = 4L * 1024 * 1024; + + /** + * Reads a response body of at most {@code maxSize} bytes, for implementations of + * {@link SsfHttpClient}. + * @param body the body as the HTTP client delivers it + * @param declaredLength the {@code Content-Length} the response declares, {@code -1} + * if it declares none + * @param maxSize the largest body accepted, in bytes + * @param uri the URI the response came from, for the message of the exception + * @return the body + * @throws IOException if the body is larger than {@code maxSize}, or cannot be read + */ + public static byte[] readBody(InputStream body, long declaredLength, long maxSize, URI uri) throws IOException { + if (declaredLength > maxSize) { + throw new IOException("The response of " + uri + " is " + declaredLength + " bytes, larger than the " + + maxSize + " bytes accepted"); + } + int limit = (int) Math.min(maxSize, Integer.MAX_VALUE - 16); + byte[] bytes = body.readNBytes(limit + 1); + if (bytes.length > limit) { + throw new IOException( + "The response of " + uri + " is larger than the " + maxSize + " bytes accepted and was cut off"); + } + return bytes; + } + + /** + * @param contentType the {@code Content-Type} header of a response, may be + * {@code null} + * @return the charset the header names, UTF-8 if it names none or an unknown one + */ + public static Charset charsetOf(String contentType) { + if (contentType == null) { + return StandardCharsets.UTF_8; + } + for (String parameter : contentType.split(";")) { + String trimmed = parameter.strip(); + if (trimmed.regionMatches(true, 0, "charset=", 0, 8)) { + String name = trimmed.substring(8).strip(); + if (name.length() > 1 && name.startsWith("\"") && name.endsWith("\"")) { + name = name.substring(1, name.length() - 1); + } + try { + return Charset.forName(name); + } + catch (IllegalArgumentException ex) { + return StandardCharsets.UTF_8; + } + } + } + return StandardCharsets.UTF_8; + } + public boolean isSuccessful() { return this.status >= 200 && this.status < 300; } diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java index b5e8f57..0f0043d 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java @@ -21,10 +21,12 @@ import org.easyssf.receiver.http.SsfHttpRequest; import org.easyssf.receiver.http.SsfHttpResponse; import org.easyssf.receiver.metrics.SsfReceiverMetrics; +import org.easyssf.receiver.set.SsfSetClaims; import org.easyssf.receiver.set.SsfSetProcessor; import org.easyssf.receiver.set.SsfSetVerificationException; import org.easyssf.receiver.transmitter.SsfTransmitterTokenProvider; import org.easyssf.receiver.transmitter.SsfTransmitterUnavailableException; +import org.easyssf.receiver.transmitter.SsfTransmitterUriPolicy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -57,6 +59,13 @@ * the startup, not a request of the transmitter to slow down. {@link #wakeUp()} ends the * pause between two polls altogether; {@code SsfTransmitter} calls it when the stream is * registered, so that the first poll follows the registration at once. + * + *

+ * The poll request carries the access token, so the poll endpoint is held to the + * {@link #setUriPolicy(SsfTransmitterUriPolicy) URI policy} of the transmitter's metadata + * endpoints before the first request goes out: {@code https}, or {@code http} on loopback + * addresses only, whether the endpoint was configured or taken from the stream + * configuration the transmitter returned. */ public class SsfPoller { @@ -121,9 +130,16 @@ public class SsfPoller { private volatile String lastPollError; + /** + * The thread of the current run; a thread that finds another one here, or none, has + * been stopped and ends, even if {@link #start()} was called again meanwhile + */ private volatile Thread thread; - private volatile boolean running; + private SsfTransmitterUriPolicy uriPolicy = SsfTransmitterUriPolicy.DEFAULT; + + /** the endpoint the policy last accepted, so that it is checked once, not per poll */ + private volatile URI checkedEndpoint; private final ReentrantLock polling = new ReentrantLock(); @@ -249,6 +265,19 @@ public void setThreadFactory(ThreadFactory threadFactory) { this.threadFactory = threadFactory; } + /** + * What the poll endpoint must look like before the access token is sent to it, the + * {@link SsfTransmitterUriPolicy#DEFAULT default} unless set: {@code https}, or + * {@code http} on loopback addresses only. Pass + * {@link SsfTransmitterUriPolicy#INSECURE} for a transmitter without TLS in a test + * setup. + */ + public void setUriPolicy(SsfTransmitterUriPolicy uriPolicy) { + SsfAssert.notNull(uriPolicy, "uriPolicy must not be null"); + this.uriPolicy = uriPolicy; + this.checkedEndpoint = null; + } + /** * @param transmitter the issuer of the transmitter that is polled, for the metrics * and as the key of its acknowledgements in the {@link SsfPollAckStore} @@ -297,7 +326,6 @@ public void start() { if (this.thread != null) { return; } - this.running = true; this.metrics.pollerStarted(this.transmitter, this.pendingAckCount); Thread thread = this.threadFactory.newThread(this::run); SsfAssert.notNull(thread, "the thread factory returned no thread"); @@ -319,11 +347,14 @@ public void stop() { try { Thread thread = this.thread; this.thread = null; - this.running = false; if (thread != null) { thread.interrupt(); try { - thread.join(Duration.ofSeconds(5)); + if (!thread.join(Duration.ofSeconds(5))) { + // a handler is still running; the thread ends once it returns, + // as it is no longer the thread of this poller + logger.warn("The polling thread is still handling a SET; it ends after the handlers return"); + } } catch (InterruptedException ex) { Thread.currentThread().interrupt(); @@ -398,10 +429,17 @@ public int getPendingAckCount() { return (key != null) ? this.ackStore.size(key) : 0; } + /** + * @return whether the calling thread is the thread of the current run + */ + private boolean isCurrentThread() { + return this.thread == Thread.currentThread(); + } + private void run() { try { Thread.sleep(this.initialDelay); - while (this.running) { + while (isCurrentThread()) { clearWakeUp(); Duration pause = this.interval; long started = System.nanoTime(); @@ -421,13 +459,13 @@ else if (isLongPolling()) { } } catch (RuntimeException ex) { - if (!this.running) { + if (!isCurrentThread()) { return; } logger.warn("Could not poll the SSF transmitter: " + ex.getMessage()); logger.debug("Cause of the failed poll", ex); } - if (this.running && pause.isPositive()) { + if (isCurrentThread() && pause.isPositive()) { pause(pause); } } @@ -444,7 +482,7 @@ private void pause(Duration pause) throws InterruptedException { long nanos = pause.toNanos(); this.pausing.lock(); try { - while (!this.wakeUpRequested && nanos > 0 && this.running) { + while (!this.wakeUpRequested && nanos > 0 && isCurrentThread()) { nanos = this.wokenUp.awaitNanos(nanos); } } @@ -517,6 +555,7 @@ private PollResult poll(boolean hold) { logger.debug("Not polling, the SSF transmitter asked to wait until " + this.pausedUntil); return PollResult.SKIPPED; } + checkEndpoint(endpoint); this.lastPollAt = Instant.now(); String key = ackKey(endpoint); try { @@ -548,9 +587,38 @@ private PollResult poll(boolean hold) { } } + /** + * The access token goes to the endpoint, so the endpoint is held to the URI policy + * first. Checked once per endpoint: the supplier returns the same one poll after + * poll. + * @throws IllegalArgumentException if the endpoint does not pass; the poll fails and + * is tried again at the interval, and the message of the exception names the reason + */ + private void checkEndpoint(URI endpoint) { + if (endpoint.equals(this.checkedEndpoint)) { + return; + } + this.uriPolicy.checkEndpoint(endpoint, "The poll endpoint"); + this.checkedEndpoint = endpoint; + } + private void process(String key, String jti, Object encodedSet) { + String set = String.valueOf(encodedSet); + // the key of a SET in the response is its jti (RFC 8936, section 2.4). A SET + // listed + // under another key would be acknowledged under a key that is not its own, so it + // is reported under that key instead of handled; one that does not parse is left + // to the verifier, which rejects it the same way + String setJti = SsfSetClaims.unverifiedJti(set); + if (setJti != null && !setJti.equals(jti)) { + String message = "The SET has the jti '" + setJti + "' but is listed under '" + jti + "'"; + logger.warn("Rejecting polled SET " + jti + ": " + message); + this.metrics.setReceived(this.transmitter, SsfDeliveryMethod.POLL, SsfReceiverMetrics.SetOutcome.INVALID); + this.ackStore.record(key, SsfPendingAck.error(jti, SsfSetVerificationException.INVALID_REQUEST, message)); + return; + } try { - this.processor.process(String.valueOf(encodedSet), SsfDeliveryMethod.POLL); + this.processor.process(set, SsfDeliveryMethod.POLL); this.ackStore.record(key, SsfPendingAck.ack(jti)); } catch (SsfSetVerificationException ex) { diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfSetClaims.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfSetClaims.java index 51f1e68..9e6f1bb 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfSetClaims.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/set/SsfSetClaims.java @@ -13,6 +13,22 @@ public final class SsfSetClaims { private SsfSetClaims() { } + /** + * @param encodedSet the SET as received + * @return the {@code jti} claim, {@code null} if the SET cannot be parsed or has none + */ + public static String unverifiedJti(String encodedSet) { + if (encodedSet == null || encodedSet.isBlank()) { + return null; + } + try { + return JWTParser.parse(encodedSet).getJWTClaimsSet().getJWTID(); + } + catch (ParseException | RuntimeException ex) { + return null; + } + } + /** * @param encodedSet the SET as received * @return the {@code iss} claim, {@code null} if the SET cannot be parsed or has none diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamClient.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamClient.java index 43d3eab..828b6ad 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamClient.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamClient.java @@ -26,7 +26,9 @@ /** * Client for the stream management API of the transmitter (SSF 1.0, section 8.1). The - * endpoints are discovered from the transmitter metadata. + * endpoints are discovered from the transmitter metadata. Every stream configuration the + * transmitter returns is checked to carry the issuer of the metadata before it is + * accepted (section 8.1.1): the SETs of a stream with another issuer would not verify. */ public class SsfStreamClient { @@ -53,6 +55,8 @@ public SsfStreamClient(SsfHttpClient httpClient, SsfTransmitterTokenProvider tok /** * The streams of this receiver. + * @throws SsfStreamIssuerMismatchException if a stream has another issuer than the + * transmitter */ @SuppressWarnings("unchecked") public List getStreams() { @@ -76,12 +80,17 @@ public List getStreams() { catch (java.text.ParseException ex) { throw new SsfStreamException("The transmitter returned an invalid list of streams", 0, ex); } + streams.forEach(this::checkIssuer); return streams; } + /** + * @throws SsfStreamIssuerMismatchException if the stream has another issuer than the + * transmitter + */ public SsfStreamConfiguration getStream(String streamId) { - return new SsfStreamConfiguration( - parse(exchange("GET", withStreamId(configurationEndpoint(), streamId), null))); + return checkIssuer(new SsfStreamConfiguration( + parse(exchange("GET", withStreamId(configurationEndpoint(), streamId), null)))); } /** @@ -92,30 +101,43 @@ public SsfStreamConfiguration getStream(String streamId) { * @see SsfStreamConfiguration#poll */ public SsfStreamConfiguration createStream(SsfStreamConfiguration stream) { - SsfStreamConfiguration created = new SsfStreamConfiguration( - parse(exchange("POST", configurationEndpoint(), stream.claims()))); - String issuer = this.metadataResolver.resolve().issuer(); - if (!issuer.equals(created.issuer())) { - throw new SsfStreamIssuerMismatchException(issuer, created); - } - return created; + return checkIssuer( + new SsfStreamConfiguration(parse(exchange("POST", configurationEndpoint(), stream.claims())))); } /** * Updates the given members of a stream, the others are left as they are. * @param changes the members to change + * @throws SsfStreamIssuerMismatchException if the stream has another issuer than the + * transmitter */ public SsfStreamConfiguration updateStream(String streamId, Map changes) { - return new SsfStreamConfiguration( - parse(exchange("PATCH", configurationEndpoint(), withStreamId(changes, streamId)))); + return checkIssuer(new SsfStreamConfiguration( + parse(exchange("PATCH", configurationEndpoint(), withStreamId(changes, streamId))))); } /** * Replaces the configuration of a stream. + * @throws SsfStreamIssuerMismatchException if the stream has another issuer than the + * transmitter */ public SsfStreamConfiguration replaceStream(String streamId, SsfStreamConfiguration stream) { - return new SsfStreamConfiguration( - parse(exchange("PUT", configurationEndpoint(), withStreamId(stream.claims(), streamId)))); + return checkIssuer(new SsfStreamConfiguration( + parse(exchange("PUT", configurationEndpoint(), withStreamId(stream.claims(), streamId))))); + } + + /** + * A stream belongs to the transmitter the metadata was obtained from (SSF 1.0, + * section 8.1.1, {@code iss}); SETs of a stream with another issuer would not verify + * against that transmitter's keys and issuer, so such a stream is refused wherever it + * comes from. + */ + private SsfStreamConfiguration checkIssuer(SsfStreamConfiguration stream) { + String issuer = this.metadataResolver.resolve().issuer(); + if (!issuer.equals(stream.issuer())) { + throw new SsfStreamIssuerMismatchException(issuer, stream); + } + return stream; } public void deleteStream(String streamId) { diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamIssuerMismatchException.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamIssuerMismatchException.java index 2f831ab..05d093b 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamIssuerMismatchException.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamIssuerMismatchException.java @@ -3,22 +3,22 @@ import org.easyssf.core.stream.SsfStreamConfiguration; /** - * Thrown when the transmitter created a stream whose issuer is not the issuer the - * transmitter metadata was obtained from (SSF 1.0, section 8.1.1.1). Such a stream must - * not be used. + * Thrown when the transmitter returned a stream whose issuer is not the issuer the + * transmitter metadata was obtained from (SSF 1.0, section 8.1.1). Such a stream must not + * be used: its SETs would not verify. */ public class SsfStreamIssuerMismatchException extends SsfStreamException { private final transient SsfStreamConfiguration stream; public SsfStreamIssuerMismatchException(String expectedIssuer, SsfStreamConfiguration stream) { - super("The transmitter created stream " + stream.streamId() + " with the issuer '" + stream.issuer() + super("The transmitter returned stream " + stream.streamId() + " with the issuer '" + stream.issuer() + "' instead of '" + expectedIssuer + "'", 0, null); this.stream = stream; } /** - * The stream the transmitter created, for example to delete it. + * The stream the transmitter returned, for example to delete it if it was created. */ public SsfStreamConfiguration getStream() { return this.stream; diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java index ad3fe46..eaa1c48 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java @@ -8,6 +8,7 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.locks.ReentrantLock; import org.easyssf.core.SsfDeliveryMethod; import org.easyssf.core.event.SsfEventTypes; @@ -49,8 +50,15 @@ public class SsfStreamRegistrar { private Duration initialRetryDelay = Duration.ofSeconds(1); + /** + * The thread of the current registration; a registration that finds another thread + * here, or none, was stopped and publishes nothing + */ private volatile Thread thread; + /** start() and stop() are idempotent and safe to call from any thread */ + private final ReentrantLock lifecycle = new ReentrantLock(); + private volatile State state = State.NOT_STARTED; private volatile String lastError; @@ -153,11 +161,23 @@ public void setInitialRetryDelay(Duration initialRetryDelay) { } /** - * Starts looking up or registering the stream in the background. + * Starts looking up or registering the stream in the background. Calling it again + * while a registration runs has no effect. */ public void start() { - this.state = State.REGISTERING; - this.thread = Thread.ofVirtual().name("ssf-stream-registrar").start(this::registerWithRetries); + this.lifecycle.lock(); + try { + if (this.thread != null) { + return; + } + this.state = State.REGISTERING; + Thread thread = Thread.ofVirtual().name("ssf-stream-registrar").unstarted(this::registerWithRetries); + this.thread = thread; + thread.start(); + } + finally { + this.lifecycle.unlock(); + } } public State getState() { @@ -177,21 +197,38 @@ public String getLastError() { * @see #setDeleteOnShutdown(boolean) */ public void stop() { - Thread thread = this.thread; - this.thread = null; - if (thread != null) { - thread.interrupt(); - } - String registeredStreamId = this.receiverStream.getStreamId(); - if (this.deleteOnShutdown && this.desiredStream != null && registeredStreamId != null) { - try { - this.streamClient.deleteStream(registeredStreamId); - logger.info("Deleted SSF stream " + registeredStreamId); + this.lifecycle.lock(); + try { + Thread thread = this.thread; + this.thread = null; + if (thread != null) { + thread.interrupt(); + try { + // a registration in flight ends with the interrupted call; one that + // completes regardless sees it is no longer current and publishes + // nothing + if (!thread.join(Duration.ofSeconds(5))) { + logger.warn("The registration of the SSF stream is still in progress"); + } + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } } - catch (RuntimeException ex) { - logger.warn("Could not delete SSF stream " + registeredStreamId + ": " + ex.getMessage()); + String registeredStreamId = this.receiverStream.getStreamId(); + if (this.deleteOnShutdown && this.desiredStream != null && registeredStreamId != null) { + try { + this.streamClient.deleteStream(registeredStreamId); + logger.info("Deleted SSF stream " + registeredStreamId); + } + catch (RuntimeException ex) { + logger.warn("Could not delete SSF stream " + registeredStreamId + ": " + ex.getMessage()); + } } } + finally { + this.lifecycle.unlock(); + } } public boolean isRunning() { @@ -199,23 +236,32 @@ public boolean isRunning() { } private void registerWithRetries() { + Thread thread = Thread.currentThread(); Duration delay = this.initialRetryDelay; - while (!Thread.currentThread().isInterrupted()) { + while (!thread.isInterrupted()) { try { - register(); + Registration registration = lookUp(); + if (this.thread != thread) { + // stopped while the transmitter was called: the result is not + // published, and a stream created on the way is not left behind + if (registration.created() && this.deleteOnShutdown) { + deleteQuietly(registration.stream().streamId()); + } + return; + } + publish(registration); return; } catch (SsfStreamIssuerMismatchException ex) { - // trying again would only create another stream that must not be used + // trying again would only find or create the same stream logger.error("Not using the SSF stream: " + ex.getMessage()); this.lastError = ex.getMessage(); this.state = State.FAILED; - deleteQuietly(ex.getStream().streamId()); return; } catch (RuntimeException ex) { this.lastError = ex.getMessage(); - if (this.thread == null) { + if (this.thread != thread) { return; } logger.warn("Could not set up the SSF stream, trying again in " + delay.toSeconds() + "s: " @@ -233,50 +279,61 @@ private void registerWithRetries() { } /** - * Looks up or registers the stream once. + * Looks up or registers the stream once, on the calling thread, and publishes it. * @return the stream * @throws SsfStreamException if the transmitter could not be called + * @throws SsfStreamIssuerMismatchException if the stream has another issuer than the + * transmitter; a stream created on the way is deleted again */ public SsfStreamConfiguration register() { - SsfStreamConfiguration stream; - String action; + Registration registration = lookUp(); + publish(registration); + return registration.stream(); + } + + /** + * The outcome of a look-up: the stream, and whether it was created on the way. + */ + private record Registration(SsfStreamConfiguration stream, String action, boolean created) { + } + + private Registration lookUp() { if (this.desiredStream == null) { - stream = this.streamClient.getStream(this.streamId); - action = "Using"; + return new Registration(this.streamClient.getStream(this.streamId), "Using", false); } - else { - List streams = this.streamClient.getStreams(); - SsfStreamConfiguration existing = streams.stream() - .filter(this::hasDesiredDelivery) - .findFirst() - .orElse(null); - if (existing != null && hasDesiredEvents(existing)) { - stream = existing; - action = "Using existing"; - } - else if (existing != null) { - stream = this.streamClient.updateStream(existing.streamId(), changes()); - action = "Updated"; - } - else { - try { - stream = this.streamClient.createStream(this.desiredStream); - action = "Created"; - } - catch (SsfStreamException ex) { - // the transmitter allows a single stream per receiver and it is - // configured differently - if (ex.getStatusCode() != 409 || streams.size() != 1) { - throw ex; - } - stream = this.streamClient.updateStream(streams.get(0).streamId(), changes()); - action = "Updated"; - } + List streams = this.streamClient.getStreams(); + SsfStreamConfiguration existing = streams.stream().filter(this::hasDesiredDelivery).findFirst().orElse(null); + if (existing != null && hasDesiredEvents(existing)) { + return new Registration(existing, "Using existing", false); + } + if (existing != null) { + return new Registration(this.streamClient.updateStream(existing.streamId(), changes()), "Updated", false); + } + try { + return new Registration(this.streamClient.createStream(this.desiredStream), "Created", true); + } + catch (SsfStreamIssuerMismatchException ex) { + // the stream was created but must not be used, and trying again would only + // create another one + deleteQuietly(ex.getStream().streamId()); + throw ex; + } + catch (SsfStreamException ex) { + // the transmitter allows a single stream per receiver and it is configured + // differently + if (ex.getStatusCode() != 409 || streams.size() != 1) { + throw ex; } + return new Registration(this.streamClient.updateStream(streams.get(0).streamId(), changes()), "Updated", + false); } + } + + private void publish(Registration registration) { + SsfStreamConfiguration stream = registration.stream(); this.receiverStream.setConfiguration(stream); - logger.info(action + " SSF stream " + stream.streamId() + " (delivery " + stream.deliveryMethod() + " " - + Objects.toString(stream.deliveryEndpointUrl(), "") + ", events " + logger.info(registration.action() + " SSF stream " + stream.streamId() + " (delivery " + stream.deliveryMethod() + + " " + Objects.toString(stream.deliveryEndpointUrl(), "") + ", events " + stream.eventsDelivered().stream().map(SsfEventTypes::aliasOf).toList() + ", audience " + stream.audience() + ")"); this.lastError = null; @@ -290,7 +347,6 @@ else if (existing != null) { logger.debug("Cause of the failed listener", ex); } } - return stream; } private void deleteQuietly(String streamId) { diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/transmitter/SsfTransmitterUriPolicy.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/transmitter/SsfTransmitterUriPolicy.java index 67d06bb..2f439b5 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/transmitter/SsfTransmitterUriPolicy.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/transmitter/SsfTransmitterUriPolicy.java @@ -1,6 +1,8 @@ package org.easyssf.receiver.transmitter; +import java.net.InetAddress; import java.net.URI; +import java.net.UnknownHostException; /** * What the issuer of a transmitter and the endpoints it publishes must look like. SSF @@ -89,7 +91,11 @@ private void checkScheme(URI uri, String description) { } /** - * @return whether the host of the URI is a loopback address + * @return whether the host of the URI is a loopback address: {@code localhost} and + * names under it (RFC 6761, section 6.3), an IPv4 address in {@code 127.0.0.0/8} (RFC + * 1122, section 3.2.1.3) or the IPv6 loopback address {@code ::1}, also as the + * IPv4-mapped {@code ::ffff:127.0.0.1}. A host name that merely starts with + * {@code 127.} is not: it resolves wherever its owner points it. */ public static boolean isLoopback(URI uri) { String host = uri.getHost(); @@ -97,8 +103,44 @@ public static boolean isLoopback(URI uri) { return false; } host = host.toLowerCase(); - return host.equals("localhost") || host.endsWith(".localhost") || host.startsWith("127.") - || host.equals("[::1]") || host.equals("::1") || host.equals("[0:0:0:0:0:0:0:1]"); + return host.equals("localhost") || host.endsWith(".localhost") || isIpv4Loopback(host) || isIpv6Loopback(host); + } + + /** + * A bracketed IPv6 literal (RFC 3986, section 3.2.2), parsed without a lookup: only a + * literal is handed to {@link InetAddress}, a name would be resolved through DNS. + */ + private static boolean isIpv6Loopback(String host) { + if (!host.startsWith("[") || !host.endsWith("]")) { + return false; + } + try { + return InetAddress.getByName(host).isLoopbackAddress(); + } + catch (UnknownHostException ex) { + return false; + } + } + + /** + * A dotted-quad IPv4 address with 127 as the first octet, nothing else: neither a + * name such as {@code 127.attacker.example} nor a shorthand such as {@code 127.1}. + */ + private static boolean isIpv4Loopback(String host) { + if (!host.startsWith("127.")) { + return false; + } + String[] octets = host.split("\\.", -1); + if (octets.length != 4) { + return false; + } + for (String octet : octets) { + if (octet.isEmpty() || octet.length() > 3 || !octet.chars().allMatch(Character::isDigit) + || Integer.parseInt(octet) > 255) { + return false; + } + } + return true; } } diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/http/JdkSsfHttpClientTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/http/JdkSsfHttpClientTests.java index 89035aa..c9523df 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/http/JdkSsfHttpClientTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/http/JdkSsfHttpClientTests.java @@ -1,12 +1,22 @@ package org.easyssf.receiver.http; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; import org.easyssf.test.TestTransmitter; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIOException; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; class JdkSsfHttpClientTests { @@ -32,6 +42,95 @@ void returnsResponseWhateverItsStatus() throws Exception { assertThat(missing.isSuccessful()).isFalse(); } + @Test + void cutsOffResponseLargerThanTheMaximum() throws Exception { + byte[] body = new byte[1000]; + Arrays.fill(body, (byte) 'x'); + // a response that declares its length is refused before it is read, one without + // (chunked) is cut off while it is read + try (OneResponseServer declared = new OneResponseServer("HTTP/1.1 200 OK\r\n" + + "Content-Type: text/plain; charset=UTF-8\r\nContent-Length: 1000\r\nConnection: close\r\n\r\n", body); + OneResponseServer chunked = new OneResponseServer( + "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n", chunk(body))) { + this.httpClient.setMaxResponseSize(999); + assertThatIOException().isThrownBy(() -> this.httpClient.execute(SsfHttpRequest.get(declared.uri()))) + .withMessageContaining("larger than the 999 bytes accepted"); + assertThatIOException().isThrownBy(() -> this.httpClient.execute(SsfHttpRequest.get(chunked.uri()))) + .withMessageContaining("larger than the 999 bytes accepted"); + + this.httpClient.setMaxResponseSize(1000); + assertThat(this.httpClient.execute(SsfHttpRequest.get(declared.uri())).body()) + .isEqualTo(new String(body, StandardCharsets.UTF_8)); + assertThat(this.httpClient.execute(SsfHttpRequest.get(chunked.uri())).body()).hasSize(1000); + } + } + + private static byte[] chunk(byte[] body) { + byte[] head = (Integer.toHexString(body.length) + "\r\n").getBytes(StandardCharsets.US_ASCII); + byte[] tail = "\r\n0\r\n\r\n".getBytes(StandardCharsets.US_ASCII); + byte[] chunked = new byte[head.length + body.length + tail.length]; + System.arraycopy(head, 0, chunked, 0, head.length); + System.arraycopy(body, 0, chunked, head.length, body.length); + System.arraycopy(tail, 0, chunked, head.length + body.length, tail.length); + return chunked; + } + + /** + * Answers every request with the same response; the module path of the tests has no + * {@code jdk.httpserver}. + */ + private static final class OneResponseServer implements AutoCloseable { + + private final ServerSocket socket; + + private final Thread thread; + + OneResponseServer(String head, byte[] body) throws IOException { + this.socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + this.thread = Thread.ofVirtual().start(() -> { + while (!this.socket.isClosed()) { + try (Socket connection = this.socket.accept()) { + InputStream in = connection.getInputStream(); + // the request ends with an empty line + int last = 0; + for (int c = in.read(); c != -1; c = in.read()) { + if (c == '\n' && last == '\n') { + break; + } + if (c != '\r') { + last = c; + } + } + OutputStream out = connection.getOutputStream(); + out.write(head.getBytes(StandardCharsets.US_ASCII)); + out.write(body); + out.flush(); + } + catch (IOException ex) { + // closed, or the client went away + } + } + }); + } + + URI uri() { + return URI.create("http://127.0.0.1:" + this.socket.getLocalPort() + "/"); + } + + @Override + public void close() throws IOException { + this.socket.close(); + this.thread.interrupt(); + } + + } + + @Test + void responseBodiesAreBoundedByDefault() { + assertThat(this.httpClient.getMaxResponseSize()).isEqualTo(4L * 1024 * 1024); + assertThatIllegalArgumentException().isThrownBy(() -> this.httpClient.setMaxResponseSize(0)); + } + @Test void sendsUserAgentOfTheJdkByDefault() throws Exception { this.httpClient.execute(SsfHttpRequest.get(URI.create(transmitter.jwksUri()))); diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java index 2b207d9..08d38ed 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java @@ -16,6 +16,7 @@ import org.easyssf.receiver.set.SsfSetProcessor; import org.easyssf.receiver.transmitter.ClientCredentialsSsfTransmitterTokenProvider; import org.easyssf.receiver.transmitter.SsfTransmitterTokenProvider; +import org.easyssf.receiver.transmitter.SsfTransmitterUriPolicy; import org.easyssf.test.TestTransmitter; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeEach; @@ -24,6 +25,8 @@ import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.assertj.core.api.Assertions.assertThatIllegalStateException; import static org.awaitility.Awaitility.await; import static org.easyssf.core.event.SsfSubjectIdentifiers.opaque; @@ -172,6 +175,45 @@ void reportsInvalidSetToTheTransmitter() { assertThat(counter("easyssf.receiver.sets", "delivery", "poll", "outcome", "invalid")).isEqualTo(1); } + @Test + void reportsSetWhoseJtiDiffersFromTheKeyItIsListedUnderWithoutHandlingIt() { + var claims = transmitter.setClaims("CaepSessionRevoked", opaque("session-1")).build(); + transmitter.queueSet("other-jti", transmitter.signSet(claims)); + String valid = queue("session-2"); + + assertThat(this.poller.pollNow()).isEqualTo(2); + + // the SET is not handled, and the error names the key so that the transmitter + // drops it + assertThat(this.handled).containsExactly(valid); + assertThat(transmitter.acknowledgedSets()).containsExactly(valid); + assertThat(transmitter.reportedErrors()).containsOnlyKeys("other-jti"); + assertThat(transmitter.reportedErrors().get("other-jti")) + .asInstanceOf(org.assertj.core.api.InstanceOfAssertFactories.map(String.class, Object.class)) + .containsEntry("err", "invalid_request") + .hasEntrySatisfying("description", + (description) -> assertThat(description.toString()).contains(claims.getJWTID()) + .contains("other-jti")); + assertThat(counter("easyssf.receiver.sets", "delivery", "poll", "outcome", "invalid")).isEqualTo(1); + } + + @Test + void doesNotSendTheAccessTokenToAPollEndpointWithoutTls() { + SsfPoller insecure = poller(() -> URI.create("http://transmitter.example/poll"), TestTransmitter.ACCESS_TOKEN); + assertThatIllegalArgumentException().isThrownBy(insecure::pollNow) + .withMessageContaining("poll endpoint") + .withMessageContaining("must use https"); + assertThat(transmitter.pollRequests()).isZero(); + + // the loopback transmitter of the tests passes, and so does any host with the + // insecure policy + assertThat(this.poller.pollNow()).isZero(); + assertThat(transmitter.pollRequests()).isEqualTo(1); + insecure.setUriPolicy(SsfTransmitterUriPolicy.INSECURE); + assertThatExceptionOfType(IllegalStateException.class).isThrownBy(insecure::pollNow) + .withMessageContaining("Poll request to http://transmitter.example/poll failed"); + } + @Test void doesNotAcknowledgeSetThatCouldNotBeHandledSoThatItIsDeliveredAgain() { String jti = queue("session-1"); diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java index 3fb5b05..ea368b3 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java @@ -5,6 +5,11 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.easyssf.core.SsfDeliveryMethod; import org.easyssf.core.event.SsfEventTypes; @@ -158,6 +163,119 @@ void registrarDeletesStreamCreatedWithAnotherIssuerAndGivesUp() { assertThat(this.receiverStream.getStreamId()).isNull(); } + @Test + void refusesExistingStreamsWithAnotherIssuer() { + transmitter.setStreamIssuer("https://other.example"); + Map existing = transmitter + .addStream(SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null).claims()); + String streamId = (String) existing.get("stream_id"); + assertThatExceptionOfType(SsfStreamIssuerMismatchException.class) + .isThrownBy(() -> this.streamClient.getStream(streamId)); + assertThatExceptionOfType(SsfStreamIssuerMismatchException.class) + .isThrownBy(() -> this.streamClient.getStreams()); + assertThatExceptionOfType(SsfStreamIssuerMismatchException.class) + .isThrownBy(() -> this.streamClient.updateStream(streamId, Map.of("description", "x"))); + assertThatExceptionOfType(SsfStreamIssuerMismatchException.class).isThrownBy(() -> this.streamClient + .replaceStream(streamId, SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null))); + } + + @Test + void registrarGivesUpOnExistingStreamWithAnotherIssuerWithoutDeletingIt() { + transmitter.setStreamIssuer("https://other.example"); + Map existing = transmitter + .addStream(SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null).claims()); + SsfStreamRegistrar registrar = SsfStreamRegistrar.forExistingStream(this.streamClient, this.receiverStream, + (String) existing.get("stream_id")); + registrar.setInitialRetryDelay(Duration.ofMillis(50)); + registrar.start(); + try { + await().atMost(Duration.ofSeconds(5)).until(() -> registrar.getState() == SsfStreamRegistrar.State.FAILED); + } + finally { + registrar.stop(); + } + assertThat(registrar.getLastError()).contains("https://other.example"); + assertThat(this.receiverStream.getStreamId()).isNull(); + // the stream is not this registrar's to delete + assertThat(transmitter.streams()).hasSize(1); + } + + @Test + void registrarStartIsIdempotentAndStopEndsTheRegistration() { + transmitter.setAvailable(false); + Set callers = ConcurrentHashMap.newKeySet(); + AtomicInteger calls = new AtomicInteger(); + SsfHttpClient recording = (request) -> { + callers.add(Thread.currentThread()); + calls.incrementAndGet(); + return http.execute(request); + }; + SsfStreamClient client = new SsfStreamClient(recording, () -> TestTransmitter.ACCESS_TOKEN, + new SsfTransmitterMetadataResolver(transmitter.issuer(), null, http)); + SsfStreamRegistrar registrar = SsfStreamRegistrar.forManagedStream(client, this.receiverStream, + SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); + registrar.setInitialRetryDelay(Duration.ofMillis(20)); + registrar.start(); + registrar.start(); + assertThat(registrar.isRunning()).isTrue(); + await().atMost(Duration.ofSeconds(5)).until(() -> calls.get() >= 3); + // the retries all come from the one registration thread + assertThat(callers).hasSize(1); + + registrar.stop(); + registrar.stop(); + assertThat(registrar.isRunning()).isFalse(); + int callsAtStop = calls.get(); + await().during(Duration.ofMillis(200)).until(() -> calls.get() == callsAtStop); + assertThat(this.receiverStream.getStreamId()).isNull(); + } + + @Test + void registrationThatCompletesAfterStopPublishesNothing() throws Exception { + CountDownLatch streamsRequested = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + // an HTTP client that outlives the interrupt of stop(), as one that swallows + // interrupts would, and completes the registration afterwards + SsfHttpClient slow = (request) -> { + if (request.uri().getPath().endsWith("/streams") && "GET".equals(request.method())) { + streamsRequested.countDown(); + boolean released = false; + while (!released) { + try { + release.await(); + released = true; + } + catch (InterruptedException ex) { + // keeps waiting, the interrupt is swallowed + } + } + } + return http.execute(request); + }; + SsfStreamClient client = new SsfStreamClient(slow, () -> TestTransmitter.ACCESS_TOKEN, + new SsfTransmitterMetadataResolver(transmitter.issuer(), null, http)); + List registered = new ArrayList<>(); + SsfStreamRegistrar registrar = SsfStreamRegistrar.forManagedStream(client, this.receiverStream, + SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); + registrar.setDeleteOnShutdown(true); + registrar.addListener(registered::add); + registrar.start(); + assertThat(streamsRequested.await(5, TimeUnit.SECONDS)).isTrue(); + + Thread stopping = Thread.ofVirtual().start(registrar::stop); + // stop() waits for the registration, which completes only once released + await().during(Duration.ofMillis(200)).until(stopping::isAlive); + release.countDown(); + stopping.join(Duration.ofSeconds(10)); + assertThat(stopping.isAlive()).isFalse(); + + // the stream created on the way is deleted, as it would have been on shutdown + await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.streams().isEmpty()); + assertThat(registered).isEmpty(); + assertThat(this.receiverStream.getStreamId()).isNull(); + assertThat(registrar.getState()).isNotEqualTo(SsfStreamRegistrar.State.REGISTERED); + } + @Test void obtainsNewAccessTokenWhenTheTransmitterRejectsTheCurrentOne() { SsfStreamClient client = new SsfStreamClient(http, this.tokenProvider, diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/transmitter/SsfTransmitterUriPolicyTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/transmitter/SsfTransmitterUriPolicyTests.java index 68ff428..0e8a03c 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/transmitter/SsfTransmitterUriPolicyTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/transmitter/SsfTransmitterUriPolicyTests.java @@ -1,10 +1,12 @@ package org.easyssf.receiver.transmitter; import java.net.URI; +import java.util.List; import org.easyssf.receiver.http.JdkSsfHttpClient; import org.junit.jupiter.api.Test; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.assertj.core.api.Assertions.assertThatNoException; @@ -28,6 +30,34 @@ void rejectsHttpIssuerExceptOnLoopback() { assertThatNoException().isThrownBy(() -> this.policy.checkIssuer("http://idp.localhost/realms/demo")); } + /** + * The same cases as {@code TestURIPolicyRequiresHTTPSExceptOnLoopback} of the Go + * library, so that the libraries agree on what a loopback address is. + */ + @Test + void loopbackIsLocalhostTheIpv4LoopbackNetworkAndTheIpv6LoopbackAddress() { + for (String issuer : List.of("http://localhost:8080/realms/demo", "http://LocalHost/x", + "http://127.0.0.1:8080/x", "http://127.255.255.254/x", "http://127.1.2.3/x", "http://[::1]:8080/x", + "http://[0:0:0:0:0:0:0:1]/x", "http://[::ffff:127.0.0.1]/x", "http://app.localhost/x")) { + assertThatNoException().as(issuer).isThrownBy(() -> this.policy.checkIssuer(issuer)); + assertThat(SsfTransmitterUriPolicy.isLoopback(URI.create(issuer))).as(issuer).isTrue(); + } + } + + @Test + void hostNameThatMerelyLooksLikeALoopbackAddressIsNotLoopback() { + // a host name that starts with 127. goes wherever its owner points it, and a + // shorthand such as 127.1 is not an address either (java.net.URI refuses it as a + // host, so the message differs) + for (String issuer : List.of("http://127.attacker.example/x", "http://127.0.0.1.attacker.example/x", + "http://127.1/x", "http://127/x", "http://127.0.0/x", "http://127.0.0.256/x", + "http://localhost.attacker.example/x", "http://[::2]/x", "http://[::ffff:10.0.0.1]/x", + "http://10.0.0.1/x")) { + assertThatIllegalArgumentException().as(issuer).isThrownBy(() -> this.policy.checkIssuer(issuer)); + assertThat(SsfTransmitterUriPolicy.isLoopback(URI.create(issuer))).as(issuer).isFalse(); + } + } + @Test void insecurePolicyAcceptsHttpAnywhere() { assertThatNoException() From 9657399e78a5ea8b198333c15a201a6c1602edc3 Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Thu, 29 Oct 2026 22:10:05 +0100 Subject: [PATCH 2/4] Check the poll endpoint on stop too, fence publication The acknowledgements the poller flushes on stop() went through the shared request method without the endpoint check, so an endpoint that had moved to plain http meanwhile would have received the access token. The check now sits in that method, where every poll request passes. The stream registrar checked that it was still the current registration and then published, with stop() free to run in between. Publication and the hand-over of the thread now exclude each other under a lock of their own, so a registration either publishes before stop() goes on, and the stream it published is the one stop() deletes, or not at all. Signed-off-by: Thomas Darimont --- .../org/easyssf/receiver/poll/SsfPoller.java | 11 ++-- .../receiver/stream/SsfStreamRegistrar.java | 50 +++++++++++++++---- .../easyssf/receiver/poll/SsfPollerTests.java | 33 +++++++++++- .../stream/SsfStreamManagementTests.java | 35 +++++++++++++ 4 files changed, 115 insertions(+), 14 deletions(-) diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java index 0f0043d..1d0fdd2 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java @@ -555,7 +555,6 @@ private PollResult poll(boolean hold) { logger.debug("Not polling, the SSF transmitter asked to wait until " + this.pausedUntil); return PollResult.SKIPPED; } - checkEndpoint(endpoint); this.lastPollAt = Instant.now(); String key = ackKey(endpoint); try { @@ -589,8 +588,8 @@ private PollResult poll(boolean hold) { /** * The access token goes to the endpoint, so the endpoint is held to the URI policy - * first. Checked once per endpoint: the supplier returns the same one poll after - * poll. + * first. Checked once per endpoint: the supplier returns the same one request after + * request, and a changed one is checked again. * @throws IllegalArgumentException if the endpoint does not pass; the poll fails and * is tried again at the interval, and the message of the exception names the reason */ @@ -635,7 +634,13 @@ private void process(String key, String jti, Object encodedSet) { } } + /** + * The one method that sends a poll request, for a poll as well as for the + * acknowledgements flushed on {@link #stop()}: the endpoint is checked here, so that + * no path sends the access token around the policy. + */ private Map poll(URI endpoint, String key, int maxEvents, boolean returnImmediately) { + checkEndpoint(endpoint); List pending = this.ackStore.pending(key, this.maxAckBatch); List acks = new ArrayList<>(); Map errors = new LinkedHashMap<>(); diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java index eaa1c48..b323ea3 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java @@ -59,6 +59,15 @@ public class SsfStreamRegistrar { /** start() and stop() are idempotent and safe to call from any thread */ private final ReentrantLock lifecycle = new ReentrantLock(); + /** + * Guards {@link #thread} and the publication of a registration: stop() takes the + * thread away under it, the registration checks that it is still the thread and + * publishes under it, so a registration either publishes before stop() proceeds or + * not at all. A lock of its own, as stop() must not hold it while it waits for the + * registration thread. + */ + private final ReentrantLock publication = new ReentrantLock(); + private volatile State state = State.NOT_STARTED; private volatile String lastError; @@ -172,7 +181,13 @@ public void start() { } this.state = State.REGISTERING; Thread thread = Thread.ofVirtual().name("ssf-stream-registrar").unstarted(this::registerWithRetries); - this.thread = thread; + this.publication.lock(); + try { + this.thread = thread; + } + finally { + this.publication.unlock(); + } thread.start(); } finally { @@ -199,8 +214,17 @@ public String getLastError() { public void stop() { this.lifecycle.lock(); try { - Thread thread = this.thread; - this.thread = null; + Thread thread; + // a publication in progress completes first, so that the stream it publishes + // is the one deleted below; none starts afterwards + this.publication.lock(); + try { + thread = this.thread; + this.thread = null; + } + finally { + this.publication.unlock(); + } if (thread != null) { thread.interrupt(); try { @@ -241,15 +265,21 @@ private void registerWithRetries() { while (!thread.isInterrupted()) { try { Registration registration = lookUp(); - if (this.thread != thread) { - // stopped while the transmitter was called: the result is not - // published, and a stream created on the way is not left behind - if (registration.created() && this.deleteOnShutdown) { - deleteQuietly(registration.stream().streamId()); + this.publication.lock(); + try { + if (this.thread != thread) { + // stopped while the transmitter was called: the result is not + // published, and a stream created on the way is not left behind + if (registration.created() && this.deleteOnShutdown) { + deleteQuietly(registration.stream().streamId()); + } + return; } - return; + publish(registration); + } + finally { + this.publication.unlock(); } - publish(registration); return; } catch (SsfStreamIssuerMismatchException ex) { diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java index 08d38ed..12d9f02 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java @@ -64,6 +64,11 @@ private SsfPoller poller(java.util.function.Supplier endpoint, String acces } private SsfPoller poller(java.util.function.Supplier endpoint, SsfTransmitterTokenProvider tokenProvider) { + return poller(endpoint, tokenProvider, new JdkSsfHttpClient()); + } + + private SsfPoller poller(java.util.function.Supplier endpoint, SsfTransmitterTokenProvider tokenProvider, + SsfHttpClient http) { SsfEventHandler handler = (eventContext) -> { if (!this.failures.isEmpty()) { throw this.failures.remove(0); @@ -71,7 +76,6 @@ private SsfPoller poller(java.util.function.Supplier endpoint, SsfTransmitt this.handled.add(eventContext.eventToken().jti()); this.afterHandling.run(); }; - SsfHttpClient http = new JdkSsfHttpClient(); NimbusSsfSetVerifier verifier = new NimbusSsfSetVerifier(transmitter.issuer(), transmitter::jwksUri, http); MicrometerSsfReceiverMetrics metrics = new MicrometerSsfReceiverMetrics(this.meterRegistry); SsfSetProcessor processor = new SsfSetProcessor(verifier, new InMemorySsfJtiDedupStore(100), List.of(handler)); @@ -214,6 +218,33 @@ void doesNotSendTheAccessTokenToAPollEndpointWithoutTls() { .withMessageContaining("Poll request to http://transmitter.example/poll failed"); } + @Test + void doesNotFlushAcknowledgementsOnStopToAPollEndpointWithoutTls() { + // the endpoint moves to a host without TLS while acknowledgements are pending, as + // a + // changed stream configuration could make it + AtomicReference endpoint = new AtomicReference<>(URI.create(transmitter.pollUri())); + List requested = new ArrayList<>(); + SsfHttpClient recording = (request) -> { + requested.add(request.uri()); + return new JdkSsfHttpClient().execute(request); + }; + SsfPoller poller = poller(endpoint::get, () -> TestTransmitter.ACCESS_TOKEN, recording); + InMemorySsfPollAckStore acks = new InMemorySsfPollAckStore(); + poller.setAckStore(acks); + poller.setInitialDelay(Duration.ofHours(1)); + poller.start(); + URI insecure = URI.create("http://transmitter.example/poll"); + endpoint.set(insecure); + acks.record(insecure.toString(), SsfPendingAck.ack("jti-1")); + assertThat(poller.getPendingAckCount()).isEqualTo(1); + + poller.stop(); + + assertThat(requested).noneMatch((uri) -> "transmitter.example".equals(uri.getHost())); + assertThat(acks.size(insecure.toString())).isEqualTo(1); + } + @Test void doesNotAcknowledgeSetThatCouldNotBeHandledSoThatItIsDeliveredAgain() { String jti = queue("session-1"); diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java index ea368b3..90025ae 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java @@ -276,6 +276,41 @@ void registrationThatCompletesAfterStopPublishesNothing() throws Exception { assertThat(registrar.getState()).isNotEqualTo(SsfStreamRegistrar.State.REGISTERED); } + @Test + void stopWaitsForAPublicationInProgressAndDeletesTheStreamItPublished() throws Exception { + CountDownLatch publishing = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + List registered = new ArrayList<>(); + SsfStreamRegistrar registrar = registrar(SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); + registrar.setDeleteOnShutdown(true); + registrar.addListener((stream) -> { + // the registration passed its check and is publishing when stop() is called + registered.add(stream); + publishing.countDown(); + try { + release.await(); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + }); + registrar.start(); + assertThat(publishing.await(5, TimeUnit.SECONDS)).isTrue(); + + Thread stopping = Thread.ofVirtual().start(registrar::stop); + await().during(Duration.ofMillis(200)).until(stopping::isAlive); + assertThat(transmitter.streams()).hasSize(1); + release.countDown(); + stopping.join(Duration.ofSeconds(10)); + assertThat(stopping.isAlive()).isFalse(); + + // the publication completed before stop() went on, so the stream it published is + // the one stop() deleted + assertThat(registered).hasSize(1); + assertThat(this.receiverStream.getStreamId()).isEqualTo(registered.get(0).streamId()); + assertThat(transmitter.streams()).isEmpty(); + } + @Test void obtainsNewAccessTokenWhenTheTransmitterRejectsTheCurrentOne() { SsfStreamClient client = new SsfStreamClient(http, this.tokenProvider, From 30fbeafa05cf4f704f143791d653969f663d460e Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Thu, 29 Oct 2026 22:17:30 +0100 Subject: [PATCH 3/4] Leave a stream to the registration that followed a timed-out stop Signed-off-by: Thomas Darimont --- CHANGELOG.md | 3 +- .../receiver/stream/SsfStreamRegistrar.java | 38 ++++++++++--- .../stream/SsfStreamManagementTests.java | 57 +++++++++++++++++++ 3 files changed, 89 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 11439c4..48784ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,7 +42,8 @@ The notable changes of every release. The section of a version is the text of it is no longer the poller's thread. `stop()` logs when the thread is still busy after five seconds. - `SsfStreamRegistrar.start()` is idempotent and `stop()` waits for the registration thread to end before deleting the stream. A registration that completes after `stop()` publishes nothing and - deletes a stream it created on the way if the registrar deletes on shutdown. + deletes a stream it created on the way if the registrar deletes on shutdown; if the registrar was + started again meanwhile, it leaves that stream to the new registration, which adopts it. - The `easyssf.receiver.poll.pending-acks` gauge no longer turns `NaN` after a garbage collection. Micrometer refers to what a gauge reads weakly, and the poller handed it a method reference nothing else held on to; the poller now keeps that reference for as long as it lives, so the diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java index b323ea3..2860a4b 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java @@ -38,6 +38,9 @@ public class SsfStreamRegistrar { private static final Duration MAX_RETRY_DELAY = Duration.ofSeconds(30); + /** how long stop() waits for a registration in flight */ + static final Duration STOP_TIMEOUT = Duration.ofSeconds(5); + private final SsfStreamClient streamClient; private final SsfReceiverStream receiverStream; @@ -231,7 +234,7 @@ public void stop() { // a registration in flight ends with the interrupted call; one that // completes regardless sees it is no longer current and publishes // nothing - if (!thread.join(Duration.ofSeconds(5))) { + if (!thread.join(STOP_TIMEOUT)) { logger.warn("The registration of the SSF stream is still in progress"); } } @@ -267,15 +270,24 @@ private void registerWithRetries() { Registration registration = lookUp(); this.publication.lock(); try { - if (this.thread != thread) { + if (this.thread == thread) { + publish(registration); + } + else if (this.thread == null) { // stopped while the transmitter was called: the result is not // published, and a stream created on the way is not left behind if (registration.created() && this.deleteOnShutdown) { deleteQuietly(registration.stream().streamId()); } - return; } - publish(registration); + else { + // stopped and started again while the transmitter was called, + // because stop() gave up waiting: the new registration has found + // the stream created on the way, or will, so it stays; the next + // stop() deletes the stream the receiver has then + logger.debug("A registration that outlived its stop() leaves SSF stream " + + registration.stream().streamId() + " to the one that followed"); + } } finally { this.publication.unlock(); @@ -284,16 +296,26 @@ private void registerWithRetries() { } catch (SsfStreamIssuerMismatchException ex) { // trying again would only find or create the same stream - logger.error("Not using the SSF stream: " + ex.getMessage()); - this.lastError = ex.getMessage(); - this.state = State.FAILED; + this.publication.lock(); + try { + if (this.thread == thread) { + logger.error("Not using the SSF stream: " + ex.getMessage()); + this.lastError = ex.getMessage(); + this.state = State.FAILED; + } + } + finally { + this.publication.unlock(); + } return; } catch (RuntimeException ex) { - this.lastError = ex.getMessage(); if (this.thread != thread) { + // stopped meanwhile, possibly started again: not this registration's + // error to report return; } + this.lastError = ex.getMessage(); logger.warn("Could not set up the SSF stream, trying again in " + delay.toSeconds() + "s: " + ex.getMessage()); logger.debug("Cause of the failed SSF stream setup", ex); diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java index 90025ae..3570d1e 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java @@ -311,6 +311,63 @@ void stopWaitsForAPublicationInProgressAndDeletesTheStreamItPublished() throws E assertThat(transmitter.streams()).isEmpty(); } + @Test + void registrationThatOutlivesItsStopLeavesTheStreamToTheRegistrationThatFollowed() throws Exception { + CountDownLatch created = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + // the transmitter created the stream, but the client does not return, and does + // not + // care for the interrupt of stop() either + SsfHttpClient stuck = (request) -> { + var response = http.execute(request); + if ("POST".equals(request.method()) && request.uri().getPath().endsWith("/streams")) { + created.countDown(); + boolean released = false; + while (!released) { + try { + release.await(); + released = true; + } + catch (InterruptedException ex) { + // keeps waiting + } + } + } + return response; + }; + SsfStreamClient client = new SsfStreamClient(stuck, () -> TestTransmitter.ACCESS_TOKEN, + new SsfTransmitterMetadataResolver(transmitter.issuer(), null, http)); + List registered = new ArrayList<>(); + SsfStreamRegistrar registrar = SsfStreamRegistrar.forManagedStream(client, this.receiverStream, + SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); + registrar.setDeleteOnShutdown(true); + registrar.addListener(registered::add); + registrar.start(); + assertThat(created.await(5, TimeUnit.SECONDS)).isTrue(); + String streamId = (String) transmitter.streams().get(0).get("stream_id"); + + // stop() gives the first registration up after its timeout, start() begins a + // second + // one, which finds the stream the first created and adopts it + long stopping = System.nanoTime(); + registrar.stop(); + assertThat(Duration.ofNanos(System.nanoTime() - stopping)) + .isGreaterThanOrEqualTo(SsfStreamRegistrar.STOP_TIMEOUT); + registrar.start(); + await().atMost(Duration.ofSeconds(5)).until(() -> this.receiverStream.getStreamId() != null); + assertThat(this.receiverStream.getStreamId()).isEqualTo(streamId); + + // the first registration returns at last: it neither publishes nor deletes the + // stream the second one uses + release.countDown(); + await().during(Duration.ofMillis(300)).until(() -> transmitter.streams().size() == 1); + assertThat(registered).hasSize(1); + assertThat(registrar.getState()).isEqualTo(SsfStreamRegistrar.State.REGISTERED); + + registrar.stop(); + assertThat(transmitter.streams()).isEmpty(); + } + @Test void obtainsNewAccessTokenWhenTheTransmitterRejectsTheCurrentOne() { SsfStreamClient client = new SsfStreamClient(http, this.tokenProvider, From 4afa66da40f77724b467960df395997e88dca08a Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Thu, 29 Oct 2026 22:21:07 +0100 Subject: [PATCH 4/4] Document stream management with several instances Signed-off-by: Thomas Darimont --- easyssf-receiver-spring-boot-starter/README.md | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/easyssf-receiver-spring-boot-starter/README.md b/easyssf-receiver-spring-boot-starter/README.md index f7732a2..25a20d4 100644 --- a/easyssf-receiver-spring-boot-starter/README.md +++ b/easyssf-receiver-spring-boot-starter/README.md @@ -253,6 +253,16 @@ pushed SET is answered with `503` and a polled one left unacknowledged, so the t it again. `easyssf.receiver.stream.delete-on-shutdown=true` deletes the stream when the application stops. +Things to know: + +- **Several instances** register under the same credentials and reuse the one stream, as the + registrar finds it on startup. Several instances that start at the same time against a transmitter + with no stream yet may each create one, unless the transmitter allows a single stream per receiver + (Keycloak does); every event is then delivered once per stream, with a `jti` of its own, which the + dedup store cannot tell apart. Start one instance first, or let the transmitter manage the stream. +- **`delete-on-shutdown` is for a single instance**: the stream is shared, so the first instance to + stop deletes it for the others, and the next one to start creates a new stream with a new audience. + The `SsfStreamClient` bean gives access to the whole stream management API (SSF 1.0, section 8.1): ```java @@ -455,7 +465,7 @@ name, with its own status. Switch it off with `management.health.easyssf.enabled | `easyssf.receiver.stream.id` | | Stream created at the transmitter, looked up on startup. | | `easyssf.receiver.stream.events-requested` | `CaepSessionRevoked`, `CaepCredentialChange` | For a stream managed by the receiver. | | `easyssf.receiver.stream.description` | | For a stream managed by the receiver. | -| `easyssf.receiver.stream.delete-on-shutdown` | `false` | For a stream managed by the receiver. | +| `easyssf.receiver.stream.delete-on-shutdown` | `false` | For a stream managed by the receiver. Single instance only: the stream is shared, see [Stream management](#stream-management). | | `easyssf.receiver.poll.endpoint-url` | from the stream | Held to the same `https` rule as the transmitter issuer. | | `easyssf.receiver.poll.auto-startup` | `true` | | | `easyssf.receiver.poll.interval` | `30s` | |