diff --git a/docs/services/cloudfront.md b/docs/services/cloudfront.md index beda278..f3425da 100644 --- a/docs/services/cloudfront.md +++ b/docs/services/cloudfront.md @@ -156,9 +156,12 @@ PUT, PATCH and DELETE). method; any other method returns 403 `Invalid method.`. A behavior without `AllowedMethods` allows GET and HEAD. Origin forwarding preserves the raw path; custom-origin redirects are not followed. - POST, PUT, PATCH and DELETE are forwarded to custom origins with the viewer request body and its - `Content-Type`; `Content-Length` is set from the body. Their responses are never cached, as on AWS. - Signed URL and signed cookie enforcement applies to them like any other method. Request bodies are - bounded by the emulator-wide `floci.protocols.max-request-size` limit. + `Content-Type`. The body is streamed to the origin with the viewer's `Content-Length`, or chunked + when the viewer sent it chunked or, as HTTP/2 allows, with no length, so it is never held in + memory. Their responses are never cached, as on AWS. Signed URL and signed cookie enforcement + applies to them like any other method. Request bodies are bounded by the emulator-wide + `floci.protocols.max-request-size` limit (`FLOCI_PROTOCOLS_MAX_REQUEST_SIZE`, 2048 MB by default); + CloudFront accepts up to 64 GB, so raise it for larger uploads. - Every distribution is also served as `{id}.cloudfront.{host}` for each endpoint host Floci resolves: `localhost`, `localhost.floci.io`, `localhost.localstack.cloud`, `FLOCI_HOSTNAME` and every `FLOCI_DNS_EXTRA_SUFFIXES` entry. `{id}.cloudfront.localhost.floci.io` and @@ -223,6 +226,10 @@ PUT, PATCH and DELETE). `Elemental-MediaPackage` or the `UseOriginCacheControlHeaders` policies. - In-process S3 origins do not apply forwarding settings: viewer query strings, cookies and headers do not change the S3 read. +- Origin responses are streamed to the viewer, from in-process S3 origins and custom origins alike, + so a large object is never held in memory. A custom origin's `Content-Length` is passed on, and a + response it sends chunked stays chunked. Floci does not cache, so it serves objects larger than + 50 GB the way CloudFront does with caching disabled; with caching enabled, CloudFront refuses them. - Origin `Set-Cookie` headers always reach the viewer. With legacy `Forward=none`, AWS strips them from the response; Floci does not. - Custom origins that resolve to loopback, private, link-local, carrier-grade NAT, or other non-routable addresses are rejected by default. Development-only private origins must be explicitly allowlisted by exact hostname. diff --git a/src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginHttpClient.java b/src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginHttpClient.java index 4e7bffe..f457afd 100644 --- a/src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginHttpClient.java +++ b/src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginHttpClient.java @@ -3,6 +3,7 @@ package io.github.hectorvent.floci.services.cloudfront; import io.github.hectorvent.floci.core.common.SsrfProtection; import org.apache.hc.client5.http.DnsResolver; import org.apache.hc.client5.http.SystemDefaultDnsResolver; +import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase; import org.apache.hc.client5.http.config.ConnectionConfig; import org.apache.hc.client5.http.config.RequestConfig; import org.apache.hc.client5.http.config.TlsConfig; @@ -13,32 +14,32 @@ import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager; import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder; import org.apache.hc.client5.http.impl.routing.DefaultRoutePlanner; import org.apache.hc.client5.http.protocol.HttpClientContext; +import org.apache.hc.client5.http.routing.RoutingSupport; import org.apache.hc.client5.http.ssl.ClientTlsStrategyBuilder; +import org.apache.hc.core5.http.ClassicHttpResponse; import org.apache.hc.core5.http.Header; +import org.apache.hc.core5.http.HttpEntity; +import org.apache.hc.core5.http.HttpException; import org.apache.hc.core5.http.HttpVersion; -import org.apache.hc.core5.http.io.entity.ByteArrayEntity; -import org.apache.hc.core5.http.io.entity.EntityUtils; -import org.apache.hc.core5.http.io.support.ClassicRequestBuilder; +import org.apache.hc.core5.http.io.entity.InputStreamEntity; import org.apache.hc.core5.http2.HttpVersionPolicy; import org.apache.hc.core5.util.Timeout; +import org.jboss.logging.Logger; import javax.net.ssl.SSLContext; -import javax.net.ssl.SSLSession; +import java.io.FilterInputStream; import java.io.IOException; +import java.io.InputStream; import java.io.InterruptedIOException; import java.net.InetAddress; -import java.net.URI; import java.net.UnknownHostException; -import java.net.http.HttpClient; import java.net.http.HttpHeaders; import java.net.http.HttpRequest; -import java.net.http.HttpResponse; import java.time.Duration; import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Map; -import java.util.Optional; import java.util.Set; import java.util.TreeMap; import java.util.TreeSet; @@ -51,8 +52,11 @@ import java.util.stream.Collectors; */ final class CloudFrontOriginHttpClient implements AutoCloseable { + private static final Logger LOG = Logger.getLogger(CloudFrontOriginHttpClient.class); + private static final Duration CONNECT_TIMEOUT = Duration.ofSeconds(10); private static final Duration RESPONSE_TIMEOUT = Duration.ofSeconds(30); + static final int MAX_CONNECTIONS = 100; private final CloseableHttpClient client; @@ -86,8 +90,10 @@ final class CloudFrontOriginHttpClient implements AutoCloseable { .setDnsResolver(validatingResolver) .setDefaultConnectionConfig(connectionConfig) .setDefaultTlsConfig(tlsConfig) - .setMaxConnTotal(100) - .setMaxConnPerRoute(20); + // CloudFront has no per-origin connection limit, and a streamed response holds its + // connection for the whole transfer, so one origin may use the whole pool. + .setMaxConnTotal(MAX_CONNECTIONS) + .setMaxConnPerRoute(MAX_CONNECTIONS); if (sslContext != null) { connectionManagerBuilder.setTlsSocketStrategy(ClientTlsStrategyBuilder.create() .setSslContext(sslContext) @@ -113,39 +119,139 @@ final class CloudFrontOriginHttpClient implements AutoCloseable { .build(); } - HttpResponse send(HttpRequest request, HttpResponse.BodyHandler bodyHandler) - throws IOException, InterruptedException { - return send(request, Map.of(), bodyHandler); - } - - HttpResponse send(HttpRequest request, Map originHeaders, - HttpResponse.BodyHandler bodyHandler) - throws IOException, InterruptedException { - return send(request, List.of(), originHeaders, null, bodyHandler); + /** + * An origin response whose body has not been read. The caller reads {@code body} as it needs it + * and closes it, which releases the connection. {@code contentLength} is the length the body + * arrives with, {@code -1} when it comes chunked or with none. + */ + record OpenResponse(int statusCode, HttpHeaders headers, long contentLength, InputStream body) { } /** - * Sends {@code request} with {@code forwardedHeaders} added and {@code requestBody} as its - * entity, or with no entity when it is {@code null}. Each origin header replaces every - * same-named request or forwarded header, so a viewer cannot repeat a header to smuggle its own - * value past an origin custom header. The request's own body publisher must be empty: the body - * travels as bytes so its {@code Content-Length} is exact. + * Sends {@code request} with {@code forwardedHeaders} added and {@code requestBody}, when it is + * not {@code null}, streamed as the entity, with {@code requestBodyLength} as its + * {@code Content-Length}, or chunked when the length is negative. Each origin header replaces + * every same-named request or forwarded header, so a viewer cannot repeat a header to smuggle its + * own value past an origin custom header. The response comes back before its body is read, so + * neither body has to fit in memory. */ - HttpResponse send(HttpRequest request, - List forwardedHeaders, - Map originHeaders, byte[] requestBody, - HttpResponse.BodyHandler bodyHandler) + OpenResponse open(HttpRequest request, + List forwardedHeaders, + Map originHeaders, InputStream requestBody, long requestBodyLength) + throws IOException, InterruptedException { + // No entity content type: the caller's Content-Type request header is sent as is. + HttpUriRequestBase originRequest = originRequest(request, forwardedHeaders, originHeaders, + requestBody != null ? new InputStreamEntity(requestBody, requestBodyLength, null) : null); + ClassicHttpResponse response; + try { + response = client.executeOpen(RoutingSupport.determineHost(originRequest), originRequest, context(request)); + } catch (HttpException e) { + throw new IOException("CloudFront origin request has no target host", e); + } catch (InterruptedIOException e) { + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException("CloudFront origin request interrupted"); + } + throw e; + } + try { + HttpEntity entity = response.getEntity(); + InputStream body = entity != null + ? new ResponseBody(entity.getContent(), originRequest, response) + : new ResponseBody(InputStream.nullInputStream(), null, response); + // The entity's length follows the framing the body arrives with; with no entity, as for a + // HEAD, the header describes the body a GET would get. + long contentLength = entity != null ? entity.getContentLength() : headerContentLength(response); + return new OpenResponse(response.getCode(), responseHeaders(response), contentLength, body); + } catch (IOException | RuntimeException e) { + originRequest.cancel(); + closeAborted(response); + throw e; + } + } + + /** + * An origin response body that releases its connection when it is closed. Closed at its end, the + * connection goes back to the pool. Closed before its end, as when a viewer goes away or a custom + * error page replaces it, the request is cancelled and the connection closed, since closing the + * response normally reads the rest of the body to reuse the connection, however long it is. + */ + private static final class ResponseBody extends FilterInputStream { + + private final HttpUriRequestBase request; + private final ClassicHttpResponse response; + private boolean finished; + + ResponseBody(InputStream content, HttpUriRequestBase request, ClassicHttpResponse response) { + super(content); + this.request = request; + this.response = response; + } + + @Override + public int read() throws IOException { + int b = super.read(); + finished |= b < 0; + return b; + } + + @Override + public int read(byte[] buffer, int offset, int length) throws IOException { + int read = super.read(buffer, offset, length); + finished |= read < 0; + return read; + } + + @Override + public void close() throws IOException { + if (finished || request == null) { + try { + super.close(); + } finally { + response.close(); + } + return; + } + request.cancel(); + closeAborted(response); + } + } + + /** Closes a response whose request was cancelled; reading the rest of it fails, as it should. */ + private static void closeAborted(ClassicHttpResponse response) { + try { + response.close(); + } catch (IOException e) { + LOG.debugv("Closed an unfinished CloudFront origin response: {0}", e.getMessage()); + } + } + + private static long headerContentLength(ClassicHttpResponse response) { + Header header = response.getFirstHeader("Content-Length"); + if (header == null) { + return -1; + } + try { + return Long.parseLong(header.getValue().trim()); + } catch (NumberFormatException e) { + LOG.debugv("Ignoring an invalid custom-origin Content-Length: {0}", e.getMessage()); + return -1; + } + } + + private HttpUriRequestBase originRequest(HttpRequest request, + List forwardedHeaders, + Map originHeaders, HttpEntity entity) throws IOException, InterruptedException { if (Thread.currentThread().isInterrupted()) { throw new InterruptedException("CloudFront origin request interrupted"); } if (request.bodyPublisher().filter(publisher -> publisher.contentLength() != 0).isPresent()) { - throw new IOException("CloudFront origin request bodies must be passed as bytes"); + throw new IOException("CloudFront origin request bodies must be passed separately"); } - ClassicRequestBuilder builder = ClassicRequestBuilder.create(request.method()) - .setUri(request.uri()) - .setVersion(HttpVersion.HTTP_1_1); + // A request that can be cancelled, so a response closed before its end is not read to the end. + HttpUriRequestBase builder = new HttpUriRequestBase(request.method(), request.uri()); + builder.setVersion(HttpVersion.HTTP_1_1); Set replaced = new TreeSet<>(String.CASE_INSENSITIVE_ORDER); replaced.addAll(originHeaders.keySet()); request.headers().map().forEach((name, values) -> { @@ -159,11 +265,13 @@ final class CloudFrontOriginHttpClient implements AutoCloseable { } } originHeaders.forEach(builder::addHeader); - if (requestBody != null) { - // No entity content type: the caller's Content-Type request header is sent as is. - builder.setEntity(new ByteArrayEntity(requestBody, null)); + if (entity != null) { + builder.setEntity(entity); } + return builder; + } + private static HttpClientContext context(HttpRequest request) { HttpClientContext context = HttpClientContext.create(); Duration responseTimeout = request.timeout().orElse(RESPONSE_TIMEOUT); context.setRequestConfig(RequestConfig.custom() @@ -173,28 +281,16 @@ final class CloudFrontOriginHttpClient implements AutoCloseable { .setContentCompressionEnabled(false) .setProtocolUpgradeEnabled(false) .build()); + return context; + } - try { - return client.execute(builder.build(), context, response -> { - Map> responseHeaders = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); - for (Header header : response.getHeaders()) { - responseHeaders.computeIfAbsent(header.getName(), ignored -> new ArrayList<>()) - .add(header.getValue()); - } - byte[] body = response.getEntity() != null - ? EntityUtils.toByteArray(response.getEntity()) - : new byte[0]; - return new PinnedHttpResponse( - response.getCode(), request, - HttpHeaders.of(responseHeaders, (name, value) -> true), - body, request.uri()); - }); - } catch (InterruptedIOException e) { - if (Thread.currentThread().isInterrupted()) { - throw new InterruptedException("CloudFront origin request interrupted"); - } - throw e; + private static HttpHeaders responseHeaders(ClassicHttpResponse response) { + Map> responseHeaders = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + for (Header header : response.getHeaders()) { + responseHeaders.computeIfAbsent(header.getName(), ignored -> new ArrayList<>()) + .add(header.getValue()); } + return HttpHeaders.of(responseHeaders, (name, value) -> true); } @Override @@ -202,29 +298,6 @@ final class CloudFrontOriginHttpClient implements AutoCloseable { client.close(); } - private record PinnedHttpResponse( - int statusCode, - HttpRequest request, - HttpHeaders headers, - byte[] body, - URI uri) implements HttpResponse { - - @Override - public Optional> previousResponse() { - return Optional.empty(); - } - - @Override - public Optional sslSession() { - return Optional.empty(); - } - - @Override - public HttpClient.Version version() { - return HttpClient.Version.HTTP_1_1; - } - } - private record ValidatingDnsResolver(DnsResolver delegate, Set allowedHosts) implements DnsResolver { diff --git a/src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontServingController.java b/src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontServingController.java index 454b492..2a1b4e0 100644 --- a/src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontServingController.java +++ b/src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontServingController.java @@ -29,13 +29,16 @@ import jakarta.ws.rs.core.Context; import jakarta.ws.rs.core.HttpHeaders; import jakarta.ws.rs.core.MediaType; import jakarta.ws.rs.core.Response; +import jakarta.ws.rs.core.StreamingOutput; import jakarta.ws.rs.core.UriInfo; import org.jboss.logging.Logger; +import java.io.IOException; +import java.io.InputStream; +import java.io.PushbackInputStream; import java.net.IDN; import java.net.URI; import java.net.http.HttpRequest; -import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.time.Instant; @@ -169,7 +172,7 @@ public class CloudFrontServingController { @Path("/{proxy:.*}") @Consumes(MediaType.WILDCARD) public Response post(@PathParam("distId") String distId, @PathParam("proxy") String proxy, - @Context HttpHeaders headers, @Context UriInfo uriInfo, byte[] body) { + @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) { return serveWithBody(distId, "POST", headers, uriInfo, body); } @@ -177,7 +180,7 @@ public class CloudFrontServingController { @Path("/{proxy:.*}") @Consumes(MediaType.WILDCARD) public Response put(@PathParam("distId") String distId, @PathParam("proxy") String proxy, - @Context HttpHeaders headers, @Context UriInfo uriInfo, byte[] body) { + @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) { return serveWithBody(distId, "PUT", headers, uriInfo, body); } @@ -185,7 +188,7 @@ public class CloudFrontServingController { @Path("/{proxy:.*}") @Consumes(MediaType.WILDCARD) public Response patch(@PathParam("distId") String distId, @PathParam("proxy") String proxy, - @Context HttpHeaders headers, @Context UriInfo uriInfo, byte[] body) { + @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) { return serveWithBody(distId, "PATCH", headers, uriInfo, body); } @@ -193,12 +196,12 @@ public class CloudFrontServingController { @Path("/{proxy:.*}") @Consumes(MediaType.WILDCARD) public Response delete(@PathParam("distId") String distId, @PathParam("proxy") String proxy, - @Context HttpHeaders headers, @Context UriInfo uriInfo, byte[] body) { + @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) { return serveWithBody(distId, "DELETE", headers, uriInfo, body); } private Response serveWithBody(String distId, String method, HttpHeaders headers, - UriInfo uriInfo, byte[] body) { + UriInfo uriInfo, InputStream body) { HttpServerRequest request = currentVertxRequest.getCurrent().request(); String rawViewerPath = rawViewerPath(request.uri()); return serve(distId, rawViewerPath, decodedViewerPath(rawViewerPath), @@ -207,7 +210,41 @@ public class CloudFrontServingController { headers.getHeaderString(HttpHeaders.AUTHORIZATION), request.getHeader("Origin"), method, null, null, request.getHeader("Pragma"), - body != null ? body : new byte[0]); + viewerBody(body, headers.getHeaderString("Transfer-Encoding"), + headers.getHeaderString(HttpHeaders.CONTENT_LENGTH))); + } + + /** + * The viewer's request body and its length. It is {@code -1} when the body comes chunked, which + * takes precedence over any {@code Content-Length} as HTTP has it, and otherwise the + * {@code Content-Length}. With neither header an HTTP/1.1 request has no body, but an HTTP/2 one + * can still send one, so the body is looked at: one that is already at its end is empty, and one + * that is not is forwarded chunked. + */ + static ViewerBody viewerBody(InputStream body, String transferEncoding, String contentLength) { + InputStream stream = body != null ? body : InputStream.nullInputStream(); + if (transferEncoding != null) { + return new ViewerBody(stream, -1); + } + if (contentLength != null && !contentLength.isBlank()) { + try { + return new ViewerBody(stream, Long.parseLong(contentLength.trim())); + } catch (NumberFormatException e) { + LOG.debugv("Treating an invalid viewer Content-Length as chunked: {0}", e.getMessage()); + return new ViewerBody(stream, -1); + } + } + PushbackInputStream peekable = new PushbackInputStream(stream, 1); + try { + int first = peekable.read(); + if (first < 0) { + return new ViewerBody(InputStream.nullInputStream(), 0); + } + peekable.unread(first); + } catch (IOException e) { + LOG.debugv("Could not read the start of a viewer request body: {0}", e.getMessage()); + } + return new ViewerBody(peekable, -1); } private Response serve(String distId, String rawViewerPath, String decodedViewerPath, @@ -217,7 +254,7 @@ public class CloudFrontServingController { String accessControlRequestMethod, String accessControlRequestHeaders, String pragma, - byte[] viewerBody) { + ViewerBody viewerBody) { boolean includeBody = !"HEAD".equals(method); boolean preflightRequest = "OPTIONS".equals(method) && viewerOrigin != null && !viewerOrigin.isBlank() @@ -263,8 +300,14 @@ public class CloudFrontServingController { accessControlRequestMethod, accessControlRequestHeaders, viewerBody); if (origin.status() >= 400) { - Response fallback = applyCustomError( - dist, origin, viewerScheme, viewerAuthorization, includeBody, directives); + Response fallback; + try { + fallback = applyCustomError( + dist, origin, viewerScheme, viewerAuthorization, includeBody, directives); + } catch (RuntimeException e) { + origin.closeStream(); + throw e; + } if (fallback != null) { return fallback; } @@ -407,7 +450,7 @@ public class CloudFrontServingController { String method, String viewerOrigin, String accessControlRequestMethod, String accessControlRequestHeaders, - byte[] viewerBody) { + ViewerBody viewerBody) { DistributionConfig config = distribution.getConfig(); String originId = CloudFrontRequestRouter.matchTargetOriginId(config, normalized); Origin origin = CloudFrontRequestRouter.findOrigin(config, originId); @@ -531,10 +574,12 @@ public class CloudFrontServingController { authorizeS3OriginRead( distribution, origin, bucket, key, viewerAuthorization); if (includeBody) { - S3Object obj = s3Service.getObject(bucket, key); - byte[] data = obj.getData() != null ? obj.getData() : new byte[0]; + // The body is streamed to the viewer, so an object of any size is served without + // being read into the heap; the metadata and the stream come from one snapshot. + S3Service.ObjectRead read = s3Service.openObject(bucket, key, null); + S3Object obj = read.object(); response = new OriginResponse( - 200, contentType(obj), data, data.length, s3ObjectHeaders(obj)); + 200, contentType(obj), null, read.body(), obj.getSize(), s3ObjectHeaders(obj)); } else { S3Object meta = s3Service.headObject(bucket, key); response = new OriginResponse( @@ -543,8 +588,13 @@ public class CloudFrontServingController { } catch (AwsException e) { response = OriginResponse.error(e.getHttpStatus(), e.getMessage()); } - return withS3CorsHeaders( - bucket, origin, includeBody ? "GET" : "HEAD", response); + try { + return withS3CorsHeaders( + bucket, origin, includeBody ? "GET" : "HEAD", response); + } catch (RuntimeException e) { + response.closeStream(); + throw e; + } } private void authorizeS3OriginRead( @@ -641,7 +691,7 @@ public class CloudFrontServingController { */ private OriginResponse fetchFromCustomOrigin(Origin origin, String forwardUri, CloudFrontOriginRequestBuilder.OriginRequest originRequest, - String viewerScheme, String method, byte[] viewerBody) { + String viewerScheme, String method, ViewerBody viewerBody) { boolean includeBody = !"HEAD".equals(method); try { URI target = buildCustomOriginUri( @@ -664,15 +714,19 @@ public class CloudFrontServingController { } } rb.method(method, HttpRequest.BodyPublishers.noBody()); - // The transport derives Content-Length from the body itself. - HttpResponse resp = httpClient.send(rb.build(), originRequest.headers(), - originHeaders, originRequestBody(method, viewerBody), - HttpResponse.BodyHandlers.ofByteArray()); + // Both bodies are streamed, so neither has to fit in memory: the viewer's goes out with its + // own Content-Length, or chunked as it came, and the origin's comes back as it arrives. + ViewerBody requestBody = originRequestBody(method, viewerBody); + CloudFrontOriginHttpClient.OpenResponse resp = httpClient.open(rb.build(), originRequest.headers(), + originHeaders, requestBody != null ? requestBody.stream() : null, + requestBody != null ? requestBody.length() : -1); String ct = resp.headers().firstValue("content-type").orElse(DEFAULT_CONTENT_TYPE); - byte[] body = resp.body() != null ? resp.body() : new byte[0]; - long contentLength = includeBody ? body.length : responseContentLength(resp); - return new OriginResponse(resp.statusCode(), ct, includeBody ? body : null, - contentLength, resp.headers().map()); + long contentLength = resp.contentLength(); + if (!includeBody) { + resp.body().close(); + return new OriginResponse(resp.statusCode(), ct, null, contentLength, resp.headers().map()); + } + return new OriginResponse(resp.statusCode(), ct, null, resp.body(), contentLength, resp.headers().map()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); LOG.warnv("CloudFront custom origin fetch was interrupted for host {0}", @@ -690,16 +744,20 @@ public class CloudFrontServingController { * The body CloudFront sends to the origin: POST, PUT and PATCH always carry one (possibly empty, so * the origin still sees {@code Content-Length: 0}); DELETE carries one only when the viewer sent it. */ - private static byte[] originRequestBody(String method, byte[] viewerBody) { + private static ViewerBody originRequestBody(String method, ViewerBody viewerBody) { if (viewerBody == null || !BODY_METHODS.contains(method)) { return null; } - if ("DELETE".equals(method) && viewerBody.length == 0) { + if ("DELETE".equals(method) && viewerBody.length() == 0) { return null; } return viewerBody; } + /** A viewer request body, read as it is forwarded, and its length, {@code -1} when it comes chunked. */ + record ViewerBody(InputStream stream, long length) { + } + private static String originAccessIdentityId(Origin origin) { Map config = origin.getS3OriginConfig(); String value = config != null ? config.get("OriginAccessIdentity") : null; @@ -797,15 +855,6 @@ public class CloudFrontServingController { return normalized.toLowerCase(Locale.ROOT); } - private static long responseContentLength(HttpResponse response) { - try { - return response.headers().firstValueAsLong("content-length").orElse(-1L); - } catch (NumberFormatException e) { - LOG.debugv("Ignoring invalid custom-origin Content-Length: {0}", e.getMessage()); - return -1L; - } - } - private static String safeOriginName(Origin origin) { try { return normalizeOriginHost(origin.getDomainName()); @@ -846,8 +895,7 @@ public class CloudFrontServingController { String pagePath = str(cer.get("ResponsePagePath")); if (pagePath.isBlank()) { // ResponseCode override with no custom page: keep the origin body, change the status. - return toResponse(new OriginResponse(responseCode, origin.contentType(), origin.body(), - origin.contentLength(), origin.headers()), includeBody, directives); + return toResponse(origin.withStatus(responseCode), includeBody, directives); } String errNormalized = CloudFrontRequestRouter.normalizePath(pagePath); @@ -857,6 +905,8 @@ public class CloudFrontServingController { return null; } + // The custom error page replaces the origin's error response, which is not sent. + origin.closeStream(); OriginResponse page; if (CloudFrontRequestRouter.isS3Origin(errOrigin)) { String key = CloudFrontRequestRouter.resolveOriginKey(errOrigin.getOriginPath(), errNormalized, null); @@ -873,8 +923,7 @@ public class CloudFrontServingController { // (AWS behavior), without recursively applying custom error handling. return toResponse(page, includeBody, directives); } - return toResponse(new OriginResponse(responseCode, page.contentType(), page.body(), - page.contentLength(), page.headers()), includeBody, directives); + return toResponse(page.withStatus(responseCode), includeBody, directives); } private Map matchCustomError(DistributionConfig config, int status) { @@ -892,21 +941,40 @@ public class CloudFrontServingController { private Response toResponse(OriginResponse origin, boolean includeBody, ResponseHeadersPolicyConfigCodec.Directives directives) { - Response.ResponseBuilder rb = Response.status(origin.status()); - HeaderCollection collected = collectResponseHeaders(origin, includeBody); - applyResponseHeadersPolicy(collected.headers(), directives, collected.forbiddenPolicyHeaders()); - for (HeaderValues header : collected.headers().values()) { - if ("date".equalsIgnoreCase(header.name())) { - currentVertxRequest.getCurrent().response() - .putHeader(header.name(), header.values()); - } else { - header.values().forEach(value -> rb.header(header.name(), value)); + try { + Response.ResponseBuilder rb = Response.status(origin.status()); + HeaderCollection collected = collectResponseHeaders(origin, includeBody); + applyResponseHeadersPolicy(collected.headers(), directives, collected.forbiddenPolicyHeaders()); + for (HeaderValues header : collected.headers().values()) { + if ("date".equalsIgnoreCase(header.name())) { + currentVertxRequest.getCurrent().response() + .putHeader(header.name(), header.values()); + } else { + header.values().forEach(value -> rb.header(header.name(), value)); + } } + if (includeBody && origin.stream() != null) { + InputStream body = origin.stream(); + StreamingOutput entity = output -> { + try (InputStream in = body) { + in.transferTo(output); + } + }; + rb.entity(entity); + } else { + if (origin.stream() != null) { + // A body that will not be sent, as for a HEAD, still holds its source open. + origin.closeStream(); + } + if (includeBody && origin.body() != null) { + rb.entity(origin.body()); + } + } + return rb.build(); + } catch (RuntimeException e) { + origin.closeStream(); + throw e; } - if (includeBody && origin.body() != null) { - rb.entity(origin.body()); - } - return rb.build(); } /** @@ -942,7 +1010,8 @@ public class CloudFrontServingController { if (origin.contentType() != null) { setHeader(headers, "Content-Type", origin.contentType()); } - if (!includeBody && origin.contentLength() >= 0) { + // A byte body sets its own length; a HEAD response and a streamed body need it stated. + if ((!includeBody || origin.stream() != null) && origin.contentLength() >= 0) { setHeader(headers, "Content-Length", Long.toString(origin.contentLength())); } return new HeaderCollection(headers, forbiddenPolicyHeaders); @@ -1093,12 +1162,7 @@ public class CloudFrontServingController { "Access-Control-Request-Method")); } }); - return new OriginResponse( - response.status(), - response.contentType(), - response.body(), - response.contentLength(), - headers); + return response.withHeaders(headers); } private static String originCustomHeader(Origin origin, String requestedName) { @@ -1191,12 +1255,42 @@ public class CloudFrontServingController { Set forbiddenPolicyHeaders) { } - /** The result of fetching from an origin, including end-to-end response headers. */ - private record OriginResponse(int status, String contentType, byte[] body, long contentLength, - Map> headers) { + /** + * The result of fetching from an origin, including end-to-end response headers. An origin's + * content, from an S3 object or a custom origin, is an open {@code stream} instead of + * {@code body}, so it never passes through the heap; the viewer response writes and closes it. + * {@code body} holds the bytes of a response Floci makes itself, such as an error. + */ + private record OriginResponse(int status, String contentType, byte[] body, InputStream stream, + long contentLength, Map> headers) { + OriginResponse(int status, String contentType, byte[] body, long contentLength, + Map> headers) { + this(status, contentType, body, null, contentLength, headers); + } + static OriginResponse error(int status, String message) { byte[] b = message == null ? new byte[0] : message.getBytes(StandardCharsets.UTF_8); return new OriginResponse(status, MediaType.TEXT_PLAIN, b, b.length, Map.of()); } + + OriginResponse withStatus(int newStatus) { + return new OriginResponse(newStatus, contentType, body, stream, contentLength, headers); + } + + OriginResponse withHeaders(Map> newHeaders) { + return new OriginResponse(status, contentType, body, stream, contentLength, newHeaders); + } + + /** Closes the stream of a response that will not be written. */ + void closeStream() { + if (stream == null) { + return; + } + try { + stream.close(); + } catch (IOException e) { + LOG.debugv("Could not close an unsent origin body: {0}", e.getMessage()); + } + } } } diff --git a/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontCustomOriginServingTest.java b/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontCustomOriginServingTest.java index 681125a..93cc5bc 100644 --- a/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontCustomOriginServingTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontCustomOriginServingTest.java @@ -16,22 +16,29 @@ import io.github.hectorvent.floci.services.cloudfront.model.ResponseHeadersPolic import io.quarkus.test.junit.QuarkusTest; import io.quarkus.test.junit.QuarkusTestProfile; import io.quarkus.test.junit.TestProfile; +import io.restassured.RestAssured; import io.restassured.response.Response; import jakarta.inject.Inject; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import java.io.BufferedReader; import java.io.IOException; +import java.io.InputStreamReader; +import java.io.OutputStream; import java.net.InetSocketAddress; +import java.net.Socket; import java.nio.charset.StandardCharsets; import java.security.KeyPair; import java.security.KeyPairGenerator; import java.security.Signature; +import java.time.Duration; import java.time.Instant; import java.util.Base64; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Random; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -43,6 +50,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; import static org.junit.jupiter.api.Assertions.assertTrue; @QuarkusTest @@ -316,6 +324,106 @@ class CloudFrontCustomOriginServingTest { assertEquals(5, hits.get()); } + @Test + void streamsALargeOriginResponseWithItsLength() throws Exception { + byte[] video = randomBytes(3 * 1024 * 1024 + 1); + startServingOrigin(video, true); + Distribution created = cloudFrontService.createDistribution( + customOriginDistribution(allMethodsBehavior()), Map.of()); + + Response response = given().header("Host", created.getDomainName()).when().get("/video.bin"); + + assertEquals(200, response.statusCode()); + assertEquals(Integer.toString(video.length), response.header("Content-Length")); + assertArrayEquals(video, response.asByteArray()); + } + + @Test + void passesAChunkedOriginResponseThroughChunked() throws Exception { + byte[] video = randomBytes(3 * 1024 * 1024 + 1); + startServingOrigin(video, false); + Distribution created = cloudFrontService.createDistribution( + customOriginDistribution(allMethodsBehavior()), Map.of()); + + Response response = given().header("Host", created.getDomainName()).when().get("/video.bin"); + + assertEquals(200, response.statusCode()); + assertNull(response.header("Content-Length"), "an origin that sent no length was given one"); + assertArrayEquals(video, response.asByteArray()); + } + + @Test + void streamsALargeViewerBodyToTheOriginWithItsLength() throws Exception { + AtomicReference received = new AtomicReference<>(); + startRecordingOrigin(received, new AtomicInteger()); + Distribution created = cloudFrontService.createDistribution( + customOriginDistribution(allMethodsBehavior()), Map.of()); + byte[] upload = randomBytes(3 * 1024 * 1024 + 1); + + given() + .header("Host", created.getDomainName()) + .contentType("application/octet-stream") + .body(upload) + .when() + .put("/uploads/video.bin") + .then() + .statusCode(200); + + assertEquals(Integer.toString(upload.length), received.get().header("Content-Length")); + assertArrayEquals(upload, received.get().body()); + } + + @Test + void forwardsAChunkedViewerBodyChunked() throws Exception { + AtomicReference received = new AtomicReference<>(); + startRecordingOrigin(received, new AtomicInteger()); + Distribution created = cloudFrontService.createDistribution( + customOriginDistribution(allMethodsBehavior()), Map.of()); + byte[] upload = randomBytes(200_001); + + int status = assertTimeoutPreemptively(Duration.ofSeconds(60), + () -> postChunked(created.getDomainName(), "/uploads", upload)); + + assertEquals(201, status); + assertEquals("chunked", received.get().header("Transfer-Encoding")); + assertNull(received.get().header("Content-Length")); + assertArrayEquals(upload, received.get().body()); + } + + @Test + void releasesTheOriginConnectionOfEveryResponseItDoesNotSend() throws Exception { + // The origin client keeps a fixed number of connections. An error response replaced by a custom + // error page, or a HEAD's, that kept its connection would leave a request past that number + // waiting for one. + originServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + originServer.createContext("/", exchange -> { + boolean errorPage = "/errors/404.html".equals(exchange.getRequestURI().getRawPath()); + byte[] body = (errorPage ? "error-page" : "x".repeat(100_000)).getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "text/plain"); + boolean head = "HEAD".equals(exchange.getRequestMethod()); + exchange.sendResponseHeaders(errorPage ? 200 : 404, head ? -1 : body.length); + if (!head) { + exchange.getResponseBody().write(body); + } + exchange.close(); + }); + originServer.start(); + Map errorPage = new LinkedHashMap<>(Map.of( + "ErrorCode", "404", "ResponseCode", "200", "ResponsePagePath", "/errors/404.html")); + Distribution distribution = customOriginDistribution(allMethodsBehavior()); + distribution.getConfig().setCustomErrorResponses(List.of(errorPage)); + Distribution created = cloudFrontService.createDistribution(distribution, Map.of()); + + assertTimeoutPreemptively(Duration.ofSeconds(60), () -> { + for (int i = 0; i < CloudFrontOriginHttpClient.MAX_CONNECTIONS + 10; i++) { + given().header("Host", created.getDomainName()).when().get("/missing/" + i) + .then().statusCode(200).body(equalTo("error-page")); + given().header("Host", created.getDomainName()).when().head("/missing/" + i) + .then().statusCode(200); + } + }); + } + @Test void forwardsTheViewerAuthorizationOnWriteMethodsOnly() throws Exception { AtomicReference received = new AtomicReference<>(); @@ -705,6 +813,51 @@ class CloudFrontCustomOriginServingTest { originServer.start(); } + /** An origin that answers every GET with {@code body}, with its length or chunked. */ + private void startServingOrigin(byte[] body, boolean withLength) throws IOException { + originServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + originServer.createContext("/", exchange -> { + exchange.getRequestBody().readAllBytes(); + exchange.getResponseHeaders().add("Content-Type", "application/octet-stream"); + exchange.sendResponseHeaders(200, withLength ? body.length : 0); + exchange.getResponseBody().write(body); + exchange.close(); + }); + originServer.start(); + } + + /** + * POSTs {@code body} with {@code Transfer-Encoding: chunked} over a raw connection, since an HTTP + * client library buffers a stream it is given and sends its length instead. + */ + private static int postChunked(String host, String path, byte[] body) throws IOException { + try (Socket socket = new Socket("localhost", RestAssured.port)) { + // A response that never comes fails the read instead of leaving it waiting. + socket.setSoTimeout(30_000); + OutputStream out = socket.getOutputStream(); + out.write(("POST " + path + " HTTP/1.1\r\nHost: " + host + + "\r\nContent-Type: application/octet-stream\r\nTransfer-Encoding: chunked" + + "\r\nConnection: close\r\n\r\n").getBytes(StandardCharsets.US_ASCII)); + for (int offset = 0; offset < body.length; offset += 8192) { + int size = Math.min(8192, body.length - offset); + out.write((Integer.toHexString(size) + "\r\n").getBytes(StandardCharsets.US_ASCII)); + out.write(body, offset, size); + out.write("\r\n".getBytes(StandardCharsets.US_ASCII)); + } + out.write("0\r\n\r\n".getBytes(StandardCharsets.US_ASCII)); + out.flush(); + String statusLine = new BufferedReader( + new InputStreamReader(socket.getInputStream(), StandardCharsets.US_ASCII)).readLine(); + return Integer.parseInt(statusLine.split(" ")[1]); + } + } + + private static byte[] randomBytes(int size) { + byte[] bytes = new byte[size]; + new Random(size).nextBytes(bytes); + return bytes; + } + private Distribution customOriginDistribution(DefaultCacheBehavior behavior) { return customOriginDistribution(behavior, null); } diff --git a/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontDistributionServingTest.java b/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontDistributionServingTest.java index 43b0f94..feecc83 100644 --- a/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontDistributionServingTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontDistributionServingTest.java @@ -273,6 +273,7 @@ class CloudFrontDistributionServingTest { .header("x-amz-meta-source", equalTo("origin")) .header("x-amz-server-side-encryption", nullValue()) .header("x-amz-checksum-sha256", nullValue()) + .header("Content-Length", equalTo(Integer.toString(body.getBytes(StandardCharsets.UTF_8).length))) .body(equalTo(body)); given().header("Host", dist.getDomainName()).when().head("/asset.txt") diff --git a/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginHttpClientTest.java b/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginHttpClientTest.java index ec0c987..37f222b 100644 --- a/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginHttpClientTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginHttpClientTest.java @@ -16,12 +16,15 @@ import javax.net.ssl.SNIHostName; import javax.net.ssl.SNIServerName; import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManagerFactory; +import java.io.ByteArrayInputStream; import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.URI; import java.net.UnknownHostException; import java.net.http.HttpRequest; -import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; import java.security.KeyStore; import java.security.Security; @@ -29,12 +32,18 @@ import java.security.cert.X509Certificate; import java.time.Duration; import java.util.List; import java.util.Map; +import java.util.Random; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; class CloudFrontOriginHttpClientTest { @@ -46,8 +55,7 @@ class CloudFrontOriginHttpClientTest { resolver(InetAddress.getByName("127.0.0.1")), List.of())) { HttpRequest request = request("http://blocked.invalid:" + server.getAddress().getPort() + "/"); - assertThrows(UnknownHostException.class, - () -> client.send(request, HttpResponse.BodyHandlers.ofByteArray())); + assertThrows(UnknownHostException.class, () -> get(client, request)); assertEquals(0, hits.get()); } finally { server.stop(0); @@ -60,9 +68,7 @@ class CloudFrontOriginHttpClientTest { InetAddress.getByName("8.8.8.8"), InetAddress.getByName("127.0.0.1")); try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(resolver, List.of())) { - assertThrows(UnknownHostException.class, () -> client.send( - request("http://mixed.invalid:8080/"), - HttpResponse.BodyHandlers.ofByteArray())); + assertThrows(UnknownHostException.class, () -> get(client, request("http://mixed.invalid:8080/"))); } } @@ -99,12 +105,11 @@ class CloudFrontOriginHttpClientTest { try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient( resolver, List.of("rebind.invalid"))) { - HttpResponse response = client.send( - request("http://rebind.invalid:" + server.getAddress().getPort() + "/"), - HttpResponse.BodyHandlers.ofByteArray()); + CloudFrontOriginHttpClient.OpenResponse response = get(client, + request("http://rebind.invalid:" + server.getAddress().getPort() + "/")); assertEquals(200, response.statusCode()); - assertArrayEquals("origin-ok".getBytes(StandardCharsets.UTF_8), response.body()); + assertArrayEquals("origin-ok".getBytes(StandardCharsets.UTF_8), readAll(response)); assertEquals(1, resolutions.get()); assertEquals(1, hits.get()); assertEquals("rebind.invalid:" + server.getAddress().getPort(), hostHeader.get()); @@ -131,9 +136,9 @@ class CloudFrontOriginHttpClientTest { try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient( resolver(InetAddress.getByName("127.0.0.1")), List.of("redirect.invalid"))) { - HttpResponse response = client.send( - request("http://redirect.invalid:" + server.getAddress().getPort() + "/start"), - HttpResponse.BodyHandlers.ofByteArray()); + CloudFrontOriginHttpClient.OpenResponse response = get(client, + request("http://redirect.invalid:" + server.getAddress().getPort() + "/start")); + readAll(response); assertEquals(302, response.statusCode()); assertEquals(0, redirectedHits.get()); @@ -162,10 +167,7 @@ class CloudFrontOriginHttpClientTest { .GET() .build(); - client.send( - request, - Map.of("X-Origin-Verify", "configured-value"), - HttpResponse.BodyHandlers.ofByteArray()); + readAll(client.open(request, List.of(), Map.of("X-Origin-Verify", "configured-value"), null, -1)); assertEquals(List.of("configured-value"), receivedValues.get()); } finally { @@ -200,8 +202,9 @@ class CloudFrontOriginHttpClientTest { .method("PUT", HttpRequest.BodyPublishers.noBody()) .build(); - HttpResponse response = client.send( - request, List.of(), Map.of(), payload, HttpResponse.BodyHandlers.ofByteArray()); + CloudFrontOriginHttpClient.OpenResponse response = client.open( + request, List.of(), Map.of(), new ByteArrayInputStream(payload), payload.length); + readAll(response); assertEquals(204, response.statusCode()); assertEquals("PUT", method.get()); @@ -249,21 +252,125 @@ class CloudFrontOriginHttpClientTest { try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient( resolver, List.of("origin.invalid", "wrong.invalid"), clientContext)) { int port = server.getAddress().getPort(); - HttpResponse response = client.send( - request("https://origin.invalid:" + port + "/"), - HttpResponse.BodyHandlers.ofByteArray()); + CloudFrontOriginHttpClient.OpenResponse response = get(client, + request("https://origin.invalid:" + port + "/")); + readAll(response); assertEquals(200, response.statusCode()); assertEquals("origin.invalid:" + port, hostHeader.get()); assertEquals("origin.invalid", requestedSni.get()); - assertThrows(IOException.class, () -> client.send( - request("https://wrong.invalid:" + port + "/"), - HttpResponse.BodyHandlers.ofByteArray())); + assertThrows(IOException.class, () -> get(client, request("https://wrong.invalid:" + port + "/"))); } finally { server.stop(0); } } + @Test + void streamsARequestBodyChunkedWhenItsLengthIsUnknown() throws Exception { + AtomicReference transferEncoding = new AtomicReference<>(); + AtomicReference body = new AtomicReference<>(); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/", exchange -> { + transferEncoding.set(exchange.getRequestHeaders().getFirst("Transfer-Encoding")); + body.set(exchange.getRequestBody().readAllBytes()); + exchange.sendResponseHeaders(204, -1); + exchange.close(); + }); + server.start(); + + byte[] payload = new byte[100_000]; + new Random(1).nextBytes(payload); + try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient( + resolver(InetAddress.getByName("127.0.0.1")), List.of("origin.invalid"))) { + HttpRequest request = HttpRequest.newBuilder(URI.create( + "http://origin.invalid:" + server.getAddress().getPort() + "/items")) + .timeout(Duration.ofSeconds(5)) + .method("POST", HttpRequest.BodyPublishers.noBody()) + .build(); + + readAll(client.open(request, List.of(), Map.of(), new ByteArrayInputStream(payload), -1)); + + assertEquals("chunked", transferEncoding.get()); + assertArrayEquals(payload, body.get()); + } finally { + server.stop(0); + } + } + + @Test + void closingAResponseBeforeItsEndDoesNotReadTheRestOfIt() throws Exception { + long size = 512L * 1024 * 1024; + AtomicLong sent = new AtomicLong(); + CountDownLatch done = new CountDownLatch(1); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/", exchange -> { + byte[] chunk = new byte[64 * 1024]; + exchange.sendResponseHeaders(200, size); + try (OutputStream out = exchange.getResponseBody()) { + for (long written = 0; written < size; written += chunk.length) { + out.write(chunk); + sent.addAndGet(chunk.length); + } + } catch (IOException expected) { + // The client went away mid-body, which is what the test does. + } finally { + done.countDown(); + } + }); + server.start(); + + try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient( + resolver(InetAddress.getByName("127.0.0.1")), List.of("origin.invalid"))) { + CloudFrontOriginHttpClient.OpenResponse response = get(client, + request("http://origin.invalid:" + server.getAddress().getPort() + "/big")); + assertEquals(size, response.contentLength()); + response.body().readNBytes(1024); + response.body().close(); + + assertTrue(done.await(30, TimeUnit.SECONDS), "the origin is still sending"); + assertTrue(sent.get() < size / 2, "closing read " + (sent.get() >> 20) + " MiB of " + (size >> 20)); + } finally { + server.stop(0); + } + } + + @Test + void aResponseReadToItsEndLeavesItsConnectionForTheNextRequest() throws Exception { + List peers = new CopyOnWriteArrayList<>(); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/", exchange -> { + peers.add(exchange.getRemoteAddress()); + byte[] body = "origin-ok".getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + server.start(); + + try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient( + resolver(InetAddress.getByName("127.0.0.1")), List.of("origin.invalid"))) { + String uri = "http://origin.invalid:" + server.getAddress().getPort() + "/"; + readAll(get(client, request(uri))); + readAll(get(client, request(uri))); + + assertEquals(2, peers.size()); + assertEquals(peers.get(0), peers.get(1), "the second request opened a new connection"); + } finally { + server.stop(0); + } + } + + private static CloudFrontOriginHttpClient.OpenResponse get(CloudFrontOriginHttpClient client, + HttpRequest request) throws Exception { + return client.open(request, List.of(), Map.of(), null, -1); + } + + private static byte[] readAll(CloudFrontOriginHttpClient.OpenResponse response) throws IOException { + try (InputStream body = response.body()) { + return body.readAllBytes(); + } + } + private static HttpServer server(AtomicInteger hits) throws Exception { HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); server.createContext("/", exchange -> { diff --git a/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontS3DiskOriginServingTest.java b/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontS3DiskOriginServingTest.java new file mode 100644 index 0000000..f6b539d --- /dev/null +++ b/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontS3DiskOriginServingTest.java @@ -0,0 +1,108 @@ +package io.github.hectorvent.floci.services.cloudfront; + +import io.github.hectorvent.floci.services.cloudfront.model.DefaultCacheBehavior; +import io.github.hectorvent.floci.services.cloudfront.model.Distribution; +import io.github.hectorvent.floci.services.cloudfront.model.DistributionConfig; +import io.github.hectorvent.floci.services.cloudfront.model.Origin; +import io.github.hectorvent.floci.services.s3.S3Service; +import io.quarkus.test.junit.QuarkusTest; +import io.quarkus.test.junit.QuarkusTestProfile; +import io.quarkus.test.junit.TestProfile; +import io.restassured.response.Response; +import jakarta.inject.Inject; +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; + +import static io.restassured.RestAssured.given; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * An S3 origin whose objects live on disk, where CloudFront streams the object file to the viewer + * instead of reading it into memory first. + */ +@QuarkusTest +@TestProfile(CloudFrontS3DiskOriginServingTest.S3OnDiskProfile.class) +class CloudFrontS3DiskOriginServingTest { + + private static final String REGION = "us-east-1"; + + @Inject + S3Service s3Service; + + @Inject + CloudFrontService cloudFrontService; + + @Test + void streamsAnObjectFileWithItsLength() { + String suffix = Long.toString(System.nanoTime(), 36); + String bucket = "cf-disk-origin-" + suffix; + // Larger than any one buffer on the way, so the body arrives over many writes. + byte[] body = new byte[3 * 1024 * 1024 + 1]; + new Random(5080).nextBytes(body); + s3Service.createBucket(bucket, REGION); + try { + s3Service.putObject(bucket, "video.bin", body, "application/octet-stream", Map.of()); + s3Service.putObject(bucket, "index.html", ("INDEX-ON-DISK-" + suffix).getBytes(StandardCharsets.UTF_8), + "text/html", Map.of()); + Distribution dist = distributionFor(bucket); + + Response get = given().header("Host", dist.getDomainName()).when().get("/video.bin"); + + assertEquals(200, get.statusCode()); + assertEquals(Integer.toString(body.length), get.header("Content-Length")); + assertArrayEquals(body, get.asByteArray()); + given().header("Host", dist.getDomainName()).when().head("/video.bin") + .then().statusCode(200) + .header("Content-Length", equalTo(Integer.toString(body.length))); + // The single-page-app fallback serves its error page through the same stream. + given().header("Host", dist.getDomainName()).when().get("/missing/route") + .then().statusCode(200) + .header("Content-Length", equalTo(Integer.toString(("INDEX-ON-DISK-" + suffix).length()))) + .body(containsString("INDEX-ON-DISK-" + suffix)); + } finally { + // The bucket is on disk and outlives the test, so it is removed rather than left for the next run. + s3Service.deleteObject(bucket, "video.bin"); + s3Service.deleteObject(bucket, "index.html"); + s3Service.deleteBucket(bucket); + } + } + + private Distribution distributionFor(String bucket) { + Origin origin = new Origin(); + origin.setId("disk-origin"); + origin.setDomainName(bucket + ".s3." + REGION + ".amazonaws.com"); + origin.setS3OriginConfig(new LinkedHashMap<>(Map.of("OriginAccessIdentity", ""))); + DefaultCacheBehavior behavior = new DefaultCacheBehavior(); + behavior.setTargetOriginId(origin.getId()); + behavior.setViewerProtocolPolicy("allow-all"); + Map spaFallback = new LinkedHashMap<>(); + spaFallback.put("ErrorCode", "404"); + spaFallback.put("ResponseCode", "200"); + spaFallback.put("ResponsePagePath", "/index.html"); + DistributionConfig config = new DistributionConfig(); + config.setEnabled(true); + config.setOrigins(List.of(origin)); + config.setDefaultCacheBehavior(behavior); + config.setCustomErrorResponses(List.of(spaFallback)); + Distribution distribution = new Distribution(); + distribution.setConfig(config); + return cloudFrontService.createDistribution(distribution, Map.of()); + } + + public static final class S3OnDiskProfile implements QuarkusTestProfile { + @Override + public Map getConfigOverrides() { + return Map.of( + "floci.storage.services.s3.mode", "persistent", + "floci.storage.persistent-path", "target/cloudfront-s3-disk-origin-it"); + } + } +} diff --git a/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontViewerBodyTest.java b/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontViewerBodyTest.java new file mode 100644 index 0000000..f0043a7 --- /dev/null +++ b/src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontViewerBodyTest.java @@ -0,0 +1,62 @@ +package io.github.hectorvent.floci.services.cloudfront; + +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** How the length of a viewer's request body is taken before it is forwarded to a custom origin. */ +class CloudFrontViewerBodyTest { + + private static final byte[] DATA = "a viewer upload".getBytes(StandardCharsets.UTF_8); + + @Test + void aChunkedBodyIsForwardedChunkedWhateverItsContentLength() throws IOException { + CloudFrontServingController.ViewerBody body = + CloudFrontServingController.viewerBody(new ByteArrayInputStream(DATA), "chunked", "3"); + + assertEquals(-1, body.length()); + assertArrayEquals(DATA, readAll(body)); + } + + @Test + void aBodyWithAContentLengthKeepsIt() throws IOException { + CloudFrontServingController.ViewerBody body = CloudFrontServingController.viewerBody( + new ByteArrayInputStream(DATA), null, Integer.toString(DATA.length)); + + assertEquals(DATA.length, body.length()); + assertArrayEquals(DATA, readAll(body)); + } + + @Test + void anInvalidContentLengthIsForwardedChunked() { + assertEquals(-1, CloudFrontServingController.viewerBody( + new ByteArrayInputStream(DATA), null, "not-a-length").length()); + } + + @Test + void aRequestWithNeitherHeaderAndNoBodyIsEmpty() { + assertEquals(0, CloudFrontServingController.viewerBody(InputStream.nullInputStream(), null, null).length()); + assertEquals(0, CloudFrontServingController.viewerBody(null, null, null).length()); + } + + @Test + void aBodyWithNeitherHeaderAsHttp2AllowsIsForwardedChunkedAndWhole() throws IOException { + CloudFrontServingController.ViewerBody body = + CloudFrontServingController.viewerBody(new ByteArrayInputStream(DATA), null, null); + + assertEquals(-1, body.length()); + assertArrayEquals(DATA, readAll(body), "the byte read to look at the body was lost"); + } + + private static byte[] readAll(CloudFrontServingController.ViewerBody body) throws IOException { + try (InputStream stream = body.stream()) { + return stream.readAllBytes(); + } + } +}