diff --git a/CHANGELOG.md b/CHANGELOG.md index 9d5759d..48784ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,8 +6,44 @@ 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; 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/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..25a20d4 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. @@ -252,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 @@ -434,7 +445,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 +453,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` | | | @@ -453,8 +465,8 @@ 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.poll.endpoint-url` | from the stream | | +| `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` | | | `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..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 @@ -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); } } @@ -548,9 +586,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 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 + */ + 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) { @@ -567,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/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..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 @@ -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; @@ -37,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; @@ -49,8 +53,24 @@ 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(); + + /** + * 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; @@ -153,11 +173,29 @@ 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.publication.lock(); + try { + this.thread = thread; + } + finally { + this.publication.unlock(); + } + thread.start(); + } + finally { + this.lifecycle.unlock(); + } } public State getState() { @@ -177,21 +215,47 @@ 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) { + this.lifecycle.lock(); + try { + 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 { - this.streamClient.deleteStream(registeredStreamId); - logger.info("Deleted SSF stream " + registeredStreamId); + thread = this.thread; + this.thread = null; } - catch (RuntimeException ex) { - logger.warn("Could not delete SSF stream " + registeredStreamId + ": " + ex.getMessage()); + finally { + this.publication.unlock(); + } + 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(STOP_TIMEOUT)) { + logger.warn("The registration of the SSF stream is still in progress"); + } + } + catch (InterruptedException ex) { + Thread.currentThread().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); + } + catch (RuntimeException ex) { + logger.warn("Could not delete SSF stream " + registeredStreamId + ": " + ex.getMessage()); + } } } + finally { + this.lifecycle.unlock(); + } } public boolean isRunning() { @@ -199,25 +263,59 @@ 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(); + this.publication.lock(); + try { + 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()); + } + } + 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(); + } return; } catch (SsfStreamIssuerMismatchException ex) { - // trying again would only create another stream that must not be used - logger.error("Not using the SSF stream: " + ex.getMessage()); - this.lastError = ex.getMessage(); - this.state = State.FAILED; - deleteQuietly(ex.getStream().streamId()); + // trying again would only find or create the same stream + 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 == null) { + 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); @@ -233,50 +331,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 +399,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..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 @@ -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; @@ -61,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); @@ -68,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)); @@ -172,6 +179,72 @@ 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 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 3fb5b05..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 @@ -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,211 @@ 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 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 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, 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()