fix(s3): open an object's metadata and body stream together for CloudFront

The streaming change from upstream (#5278) serves S3 origins through
S3Service.openObject, which upstream added in a larger disk-streaming rework
(#4691) not taken here. This adds the same method on top of the existing
openObjectStream.

docs/services/cloudfront.md

@@ -3,7 +3,8 @@
 CloudFront management-plane and local content-delivery emulation. Supports distribution lifecycle,
 cache policies, origin request policies, response headers policies, origin access controls, origin
 access identities, public keys, trusted key groups, CloudFront Functions, invalidations, tagging, and
-GET/HEAD/OPTIONS delivery from S3 or custom origins.
+viewer delivery from S3 or custom origins for every CloudFront viewer method (GET, HEAD, OPTIONS, POST,
+PUT, PATCH and DELETE).
 
 **Protocol:** REST XML  
 **API version:** `2020-05-31`  
@@ -150,9 +151,17 @@ GET/HEAD/OPTIONS delivery from S3 or custom origins.
 - CNAME aliases are globally unique. `AssociateAlias` atomically transfers an alias from its current
   owner to the target distribution. Exact aliases take precedence over the most-specific matching
   wildcard alias.
-- Viewer GET/HEAD requests, and OPTIONS requests allowed by the matched cache behavior, addressed to
+- Viewer requests addressed to an enabled distribution's generated domain or alias are routed to the
-  an enabled distribution's generated domain or alias are routed to the matching S3 or custom
+  matching S3 or custom origin when the matched cache behavior's `AllowedMethods` includes the
-  origin. Origin forwarding preserves the raw path; custom-origin redirects are not followed.
+  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`. 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
@@ -160,7 +169,7 @@ GET/HEAD/OPTIONS delivery from S3 or custom origins.
   HTTPS certificate, so a signed URL for either can be downloaded over `https://`. See
   [Downloading over HTTPS](#downloading-over-https).
 - Origin custom headers are persisted through the CloudFront API and CloudFormation. They replace
-  same-named viewer headers on custom-origin GET/HEAD/OPTIONS requests. For in-process S3 origins, a
+  every same-named forwarded viewer header on custom-origin requests. For in-process S3 origins, a
   configured `Origin` header is used for S3 CORS evaluation. AWS-prohibited names, malformed
   values, inconsistent quantities, duplicates, and quota violations are rejected with modeled
   CloudFront errors when the distribution is created or updated.
@@ -181,9 +190,48 @@ GET/HEAD/OPTIONS delivery from S3 or custom origins.
   service-principal bucket-policy grants (including the distribution `AWS:SourceArn`) when strict S3
   authentication is enabled. OAC `always`, `never`, and unsigned `no-override` requests follow their
   documented signing behavior; signed `no-override` viewer requests retain their authorization.
-- Cache-policy, origin-request-policy, and legacy `ForwardedValues` data-plane evaluation is not
+- Requests to custom origins carry what the matched cache behavior forwards. With a cache policy,
-  implemented yet. Viewer query strings therefore follow CloudFront's default behavior and are not
+  the viewer headers, cookies and query strings selected by the cache policy's
-  forwarded to origins.
+  `ParametersInCacheKeyAndForwardedToOrigin` and by the origin request policy are forwarded (the
+  union of both). `HeaderBehavior` `none`, `whitelist`, `allViewer`,
+  `allViewerAndWhitelistCloudFront` and `allExcept`, `CookieBehavior` and `QueryStringBehavior`
+  `none`, `whitelist`, `allExcept` and `all` are evaluated. A cache policy that selects nothing and no
+  origin request policy forwards no viewer headers, cookies or query strings, as on AWS.
+- A cache behavior without a cache policy uses its legacy `ForwardedValues`: `QueryString=true`
+  forwards every query string (`QueryStringCacheKeys` only narrows the cache key), cookies follow
+  `Forward` `none`, `whitelist` (with `*` and `?` wildcards) or `all`, and viewer headers are
+  forwarded by default except `Accept`, `Accept-Charset`, `Accept-Language`, `Authorization`,
+  `Host`, `Referer`, `User-Agent` and `X-HTTP-Method-Override`, which `Headers` can add back (`*`
+  forwards all of them). An origin request policy needs a cache policy on AWS, so a behavior without
+  one ignores it.
+- Whatever the policies, CloudFront sets `Host` to the origin's domain name unless the viewer's
+  `Host` is forwarded, sends `User-Agent: Amazon CloudFront` unless the viewer's `User-Agent` is
+  forwarded, appends the viewer address to `X-Forwarded-For`, adds a
+  `Via: <viewer HTTP version> <distribution domain> (CloudFront)` hop and an `X-Amz-Cf-Id`.
+  `Authorization` is always forwarded on POST, PUT, PATCH and DELETE, and on OPTIONS unless OPTIONS
+  is a cached method; `Content-Type` travels with a request body. Cache policy compression settings
+  normalize `Accept-Encoding` to `br,gzip`, `gzip`, `br` or `identity`. Forwarded cookies are sorted
+  by name, and cookies that are not `name=value` or whose name starts with `$` are dropped.
+  Hop-by-hop headers, `Expect`, `X-Edge-*`, `X-Real-IP`, `X-Forwarded-Proto` and viewer-supplied
+  `CloudFront-*` headers are never forwarded.
+- Of the CloudFront request headers, `CloudFront-Forwarded-Proto`, `CloudFront-Viewer-Address` and
+  `CloudFront-Viewer-Http-Version` are generated when a policy names them, or when an `allExcept`
+  header behavior does not exclude them (AWS documents Managed-AllViewerExceptHostHeader as adding
+  them). Device-type, geolocation, TLS, JA3/JA4 and header-order headers are not generated.
+- The eight AWS managed origin request policies are available by their AWS IDs through
+  `GetOriginRequestPolicy` and `ListOriginRequestPolicies` (`Type` is `managed` or `custom`), and
+  cannot be updated or deleted. AWS managed cache policies are not modeled: a behavior that
+  references one forwards nothing on its account, which matches `CachingDisabled`,
+  `CachingOptimized` and `CachingOptimizedForUncompressedObjects` but not `Amplify`,
+  `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.
 - Response headers policies validate the AWS configuration shape and are applied after the origin
   response, including CORS preflight fields, origin override behavior, custom headers, security
@@ -339,3 +387,6 @@ as it is.
 - CloudFormation provisioning of custom `AWS::CloudFront::ResponseHeadersPolicy` resources
   (literal custom or managed policy IDs are supported on distributions)
 - Persistent edge caching and global CDN propagation
+- Forwarding POST, PUT, PATCH and DELETE to S3 origins. AWS passes them to S3, which evaluates them
+  against the bucket policy (OAC supports `PUT` and `DELETE`); Floci's in-process S3 origin answers
+  them with 403 `Access Denied`, as S3 does for a request that holds no write grant.

src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontController.java

@@ -468,7 +468,7 @@ public class CloudFrontController {
                                               @QueryParam("Type") String type) {
         try {
             Page<OriginRequestPolicy> page = page(
-                    service.listOriginRequestPolicies(marker, paginationFetchLimit(maxItems)),
+                    service.listOriginRequestPolicies(marker, paginationFetchLimit(maxItems), type),
                     maxItems, OriginRequestPolicy::getId);
 
             XmlBuilder xml = new XmlBuilder()
@@ -479,7 +479,8 @@ public class CloudFrontController {
                     .start("Items");
             for (OriginRequestPolicy p : page.items()) {
                 xml.start("OriginRequestPolicySummary")
-                        .elem("Type", "custom")
+                        .elem("Type", CloudFrontService.isManagedOriginRequestPolicy(p.getId())
+                                ? "managed" : "custom")
                         .raw(xmlOriginRequestPolicyResponse(p))
                         .end("OriginRequestPolicySummary");
             }

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,33 +14,35 @@ 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.EntityUtils;
+import org.apache.hc.core5.http.io.entity.InputStreamEntity;
-import org.apache.hc.core5.http.io.support.ClassicRequestBuilder;
 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;
 import java.util.stream.Collectors;
 
 /**
@@ -49,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;
 
@@ -84,8 +90,10 @@ final class CloudFrontOriginHttpClient implements AutoCloseable {
                         .setDnsResolver(validatingResolver)
                         .setDefaultConnectionConfig(connectionConfig)
                         .setDefaultTlsConfig(tlsConfig)
-                        .setMaxConnTotal(100)
+                        // CloudFront has no per-origin connection limit, and a streamed response holds its
-                        .setMaxConnPerRoute(20);
+                        // 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)
@@ -111,28 +119,159 @@ final class CloudFrontOriginHttpClient implements AutoCloseable {
                 .build();
     }
 
-    HttpResponse<byte[]> send(HttpRequest request, HttpResponse.BodyHandler<byte[]> bodyHandler)
+    /**
-            throws IOException, InterruptedException {
+     * An origin response whose body has not been read. The caller reads {@code body} as it needs it
-        return send(request, Map.of(), bodyHandler);
+     * 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) {
     }
 
-    HttpResponse<byte[]> send(HttpRequest request, Map<String, String> originHeaders,
+    /**
-                              HttpResponse.BodyHandler<byte[]> bodyHandler)
+     * 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.
+     */
+    OpenResponse open(HttpRequest request,
+                      List<CloudFrontOriginRequestBuilder.Header> forwardedHeaders,
+                      Map<String, String> 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<CloudFrontOriginRequestBuilder.Header> forwardedHeaders,
+                                             Map<String, String> 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 are not supported");
+            throw new IOException("CloudFront origin request bodies must be passed separately");
         }
 
-        ClassicRequestBuilder builder = ClassicRequestBuilder.create(request.method())
+        // A request that can be cancelled, so a response closed before its end is not read to the end.
-                .setUri(request.uri())
+        HttpUriRequestBase builder = new HttpUriRequestBase(request.method(), request.uri());
-                .setVersion(HttpVersion.HTTP_1_1);
+        builder.setVersion(HttpVersion.HTTP_1_1);
-        request.headers().map().forEach((name, values) ->
+        Set<String> replaced = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
-                values.forEach(value -> builder.addHeader(name, value)));
+        replaced.addAll(originHeaders.keySet());
-        originHeaders.forEach(builder::setHeader);
+        request.headers().map().forEach((name, values) -> {
+            if (!replaced.contains(name)) {
+                values.forEach(value -> builder.addHeader(name, value));
+            }
+        });
+        for (CloudFrontOriginRequestBuilder.Header header : forwardedHeaders) {
+            if (!replaced.contains(header.name())) {
+                builder.addHeader(header.name(), header.value());
+            }
+        }
+        originHeaders.forEach(builder::addHeader);
+        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()
@@ -142,28 +281,16 @@ final class CloudFrontOriginHttpClient implements AutoCloseable {
                 .setContentCompressionEnabled(false)
                 .setProtocolUpgradeEnabled(false)
                 .build());
+        return context;
+    }
 
-        try {
+    private static HttpHeaders responseHeaders(ClassicHttpResponse response) {
-            return client.execute(builder.build(), context, response -> {
         Map<String, List<String>> 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
+        return HttpHeaders.of(responseHeaders, (name, value) -> true);
-                        ? 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;
-        }
     }
 
     @Override
@@ -171,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<byte[]> {
-
-        @Override
-        public Optional<HttpResponse<byte[]>> previousResponse() {
-            return Optional.empty();
-        }
-
-        @Override
-        public Optional<SSLSession> sslSession() {
-            return Optional.empty();
-        }
-
-        @Override
-        public HttpClient.Version version() {
-            return HttpClient.Version.HTTP_1_1;
-        }
-    }
-
     private record ValidatingDnsResolver(DnsResolver delegate, Set<String> allowedHosts)
             implements DnsResolver {
 

src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginRequestBuilder.java

@@ -0,0 +1,497 @@
+package io.github.hectorvent.floci.services.cloudfront;
+
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Decides what CloudFront sends to a custom origin for one viewer request: the viewer headers,
+ * cookies and query strings the matched cache behavior forwards, and the headers CloudFront adds on
+ * its own. Implemented to the CloudFront Developer Guide:
+ *
+ * <ul>
+ *   <li>With a cache policy, only what the cache policy's
+ *       {@code ParametersInCacheKeyAndForwardedToOrigin} or the origin request policy selects is
+ *       forwarded; the two selections are unioned.</li>
+ *   <li>Without a cache policy (legacy cache settings), {@code ForwardedValues} applies, and viewer
+ *       headers are forwarded by default except the ones the custom-origin request table says
+ *       CloudFront removes.</li>
+ *   <li>Every origin request carries {@code User-Agent: Amazon CloudFront} unless the viewer's
+ *       {@code User-Agent} is forwarded, an {@code X-Forwarded-For} ending in the viewer address,
+ *       CloudFront's {@code Via} hop and an {@code X-Amz-Cf-Id}. {@code Host} is the origin's own
+ *       name unless the viewer's {@code Host} is forwarded.</li>
+ * </ul>
+ *
+ * <p>This class holds no state and does no I/O. Origin custom headers are applied afterwards by the
+ * transport, which lets them replace same-named viewer headers.
+ */
+final class CloudFrontOriginRequestBuilder {
+
+    static final String CLOUDFRONT_USER_AGENT = "Amazon CloudFront";
+
+    private static final Set<String> BODY_METHODS = Set.of("POST", "PUT", "PATCH", "DELETE");
+    /**
+     * Never copied from a viewer request: hop-by-hop and framing fields the transport owns, and
+     * fields CloudFront removes or sets itself whatever the policy says.
+     */
+    private static final Set<String> NEVER_FORWARDED = Set.of(
+            "connection", "keep-alive", "proxy-connection", "proxy-authenticate",
+            "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade",
+            "content-length", "expect", "x-amz-cf-id", "x-real-ip", "x-forwarded-proto");
+    /** Removed by legacy cache settings unless the behavior's {@code Headers} names them. */
+    private static final Set<String> LEGACY_REMOVED = Set.of(
+            "accept", "accept-charset", "accept-language", "authorization", "host", "referer",
+            "user-agent", "x-http-method-override");
+    private static final String FORWARDED_PROTO = "CloudFront-Forwarded-Proto";
+    private static final String VIEWER_ADDRESS = "CloudFront-Viewer-Address";
+    private static final String VIEWER_HTTP_VERSION = "CloudFront-Viewer-Http-Version";
+    /** The CloudFront request headers Floci can derive from a local viewer connection. */
+    private static final List<String> GENERATED_HEADERS =
+            List.of(FORWARDED_PROTO, VIEWER_ADDRESS, VIEWER_HTTP_VERSION);
+
+    private CloudFrontOriginRequestBuilder() {
+    }
+
+    /** One request header field, in the order it was received or will be sent. */
+    record Header(String name, String value) {
+    }
+
+    /**
+     * The viewer request as CloudFront received it. {@code rawQuery} has CloudFront's signing
+     * parameters removed already; {@code httpVersion} is the viewer's protocol version, such as
+     * {@code 1.1} or {@code 2.0}.
+     */
+    record ViewerRequest(String method, List<Header> headers, String rawQuery,
+                         String clientAddress, int clientPort, String scheme, String httpVersion) {
+    }
+
+    /**
+     * The forwarding configuration of the cache behavior that matched the request. A behavior
+     * without a cache policy uses legacy cache settings, where only {@code forwardedValues} applies.
+     * The policy maps use the shapes {@link CloudFrontPolicyConfigCodec} stores.
+     */
+    record Forwarding(boolean legacy, Map<?, ?> cachePolicyParameters,
+                      Map<?, ?> originRequestPolicy, Map<?, ?> forwardedValues,
+                      boolean optionsCached) {
+
+        static Forwarding legacy(Map<?, ?> forwardedValues, boolean optionsCached) {
+            return new Forwarding(true, null, null, forwardedValues, optionsCached);
+        }
+
+        static Forwarding policies(Map<?, ?> cachePolicyParameters, Map<?, ?> originRequestPolicy,
+                                   boolean optionsCached) {
+            return new Forwarding(false, cachePolicyParameters, originRequestPolicy, null,
+                    optionsCached);
+        }
+    }
+
+    /**
+     * What CloudFront sends to the origin before origin custom headers are applied. {@code rawQuery}
+     * is {@code null} when no query string is forwarded. A {@code Host} header is present only when
+     * the viewer's {@code Host} is forwarded.
+     */
+    record OriginRequest(List<Header> headers, String rawQuery) {
+
+        static OriginRequest empty() {
+            return new OriginRequest(List.of(), null);
+        }
+    }
+
+    static OriginRequest build(ViewerRequest viewer, Forwarding forwarding, String viaHost) {
+        Selections selections = forwarding.legacy()
+                ? legacySelections(forwarding.forwardedValues())
+                : policySelections(forwarding.cachePolicyParameters(),
+                        forwarding.originRequestPolicy());
+        Set<String> connectionTokens = connectionTokens(viewer.headers());
+        boolean forwardsAuthorization = BODY_METHODS.contains(viewer.method())
+                || ("OPTIONS".equals(viewer.method()) && !forwarding.optionsCached());
+
+        List<Header> headers = new ArrayList<>();
+        List<String> cookieHeaders = new ArrayList<>();
+        List<String> forwardedFor = new ArrayList<>();
+        List<String> via = new ArrayList<>();
+        List<String> acceptEncoding = new ArrayList<>();
+        boolean userAgentForwarded = false;
+        for (Header header : viewer.headers()) {
+            String name = normalize(header.name());
+            if (name.isEmpty() || NEVER_FORWARDED.contains(name) || connectionTokens.contains(name)
+                    || name.startsWith("x-edge-") || name.startsWith("cloudfront-")) {
+                continue;
+            }
+            switch (name) {
+                case "cookie" -> cookieHeaders.add(header.value());
+                case "x-forwarded-for" -> forwardedFor.add(header.value());
+                case "via" -> {
+                    if (selections.forwardsHeader(name)) {
+                        via.add(header.value());
+                    }
+                }
+                case "accept-encoding" -> {
+                    if (selections.normalizesAcceptEncoding()) {
+                        acceptEncoding.add(header.value());
+                    } else if (selections.forwardsHeader(name)) {
+                        headers.add(header);
+                    }
+                }
+                case "authorization" -> {
+                    if (forwardsAuthorization || selections.forwardsHeader(name)) {
+                        headers.add(header);
+                    }
+                }
+                case "content-type" -> {
+                    if (BODY_METHODS.contains(viewer.method()) || selections.forwardsHeader(name)) {
+                        headers.add(header);
+                    }
+                }
+                default -> {
+                    if (selections.forwardsHeader(name)) {
+                        headers.add(header);
+                        userAgentForwarded |= "user-agent".equals(name);
+                    }
+                }
+            }
+        }
+
+        if (!userAgentForwarded) {
+            headers.add(new Header("User-Agent", CLOUDFRONT_USER_AGENT));
+        }
+        if (!acceptEncoding.isEmpty()) {
+            String normalized = normalizedAcceptEncoding(acceptEncoding, selections);
+            if (normalized != null) {
+                headers.add(new Header("Accept-Encoding", normalized));
+            }
+        }
+        String cookies = forwardedCookies(cookieHeaders, selections.cookies());
+        if (cookies != null) {
+            headers.add(new Header("Cookie", cookies));
+        }
+        for (String generated : GENERATED_HEADERS) {
+            String value = selections.generatesHeader(normalize(generated))
+                    ? generatedValue(generated, viewer) : null;
+            if (value != null) {
+                headers.add(new Header(generated, value));
+            }
+        }
+        String clientForwardedFor = forwardedFor(forwardedFor, viewer.clientAddress());
+        if (clientForwardedFor != null) {
+            headers.add(new Header("X-Forwarded-For", clientForwardedFor));
+        }
+        via.add(viewerProtocolVersion(viewer.httpVersion()) + " " + viaHost + " (CloudFront)");
+        headers.add(new Header("Via", String.join(", ", via)));
+        headers.add(new Header("X-Amz-Cf-Id", requestId()));
+
+        return new OriginRequest(List.copyOf(headers),
+                forwardedQuery(viewer.rawQuery(), selections.queryStrings()));
+    }
+
+    private static Selections policySelections(Map<?, ?> cachePolicyParameters,
+                                               Map<?, ?> originRequestPolicy) {
+        List<Selection> headers = new ArrayList<>();
+        List<Selection> cookies = new ArrayList<>();
+        List<Selection> queryStrings = new ArrayList<>();
+        for (Map<?, ?> source : new Map<?, ?>[] {cachePolicyParameters, originRequestPolicy}) {
+            if (source == null) {
+                continue;
+            }
+            headers.add(Selection.of(source.get("HeadersConfig"), "HeaderBehavior", "Headers", true));
+            cookies.add(Selection.of(source.get("CookiesConfig"), "CookieBehavior", "Cookies", false));
+            queryStrings.add(Selection.of(
+                    source.get("QueryStringsConfig"), "QueryStringBehavior", "QueryStrings", false));
+        }
+        boolean gzip = cachePolicyParameters != null
+                && isTrue(cachePolicyParameters.get("EnableAcceptEncodingGzip"));
+        boolean brotli = cachePolicyParameters != null
+                && isTrue(cachePolicyParameters.get("EnableAcceptEncodingBrotli"));
+        return new Selections(false, headers, false, cookies, queryStrings, gzip, brotli,
+                gzip || brotli);
+    }
+
+    /**
+     * Legacy {@code ForwardedValues}: headers are forwarded by default apart from the ones CloudFront
+     * removes, {@code Headers} adds names back (or everything with {@code *}), cookie names may use
+     * {@code *} and {@code ?} wildcards, and {@code QueryString=true} forwards every query string
+     * ({@code QueryStringCacheKeys} only narrows the cache key).
+     */
+    private static Selections legacySelections(Map<?, ?> forwardedValues) {
+        List<String> headerNames = stringList(forwardedValues, "Headers");
+        boolean allHeaders = headerNames.contains("*");
+        Selection headers = allHeaders
+                ? new Selection("all", List.of(), true, false)
+                : new Selection("whitelist", headerNames, true, false);
+
+        String cookiesForward = forwardedValues != null && forwardedValues.get("CookiesForward") != null
+                ? forwardedValues.get("CookiesForward").toString() : "none";
+        Selection cookies = new Selection(cookiesForward, stringList(forwardedValues, "CookieNames"),
+                false, true);
+
+        boolean queryString = forwardedValues != null && isTrue(forwardedValues.get("QueryString"));
+        Selection queryStrings = new Selection(queryString ? "all" : "none", List.of(), false, false);
+
+        return new Selections(true, List.of(headers), !allHeaders, List.of(cookies),
+                List.of(queryStrings), true, true, !headers.includes("accept-encoding"));
+    }
+
+    private static Set<String> connectionTokens(List<Header> headers) {
+        Set<String> tokens = new HashSet<>();
+        for (Header header : headers) {
+            if ("connection".equals(normalize(header.name())) && header.value() != null) {
+                for (String token : header.value().split(",")) {
+                    String normalized = normalize(token);
+                    if (!normalized.isEmpty()) {
+                        tokens.add(normalized);
+                    }
+                }
+            }
+        }
+        return tokens;
+    }
+
+    /**
+     * Accept-Encoding normalization: the formats caching is enabled for that the viewer accepts, as
+     * {@code br,gzip}, {@code gzip} or {@code br}. A cache policy sends {@code identity} when the
+     * viewer accepts none of them; legacy settings drop the header instead.
+     */
+    private static String normalizedAcceptEncoding(List<String> values, Selections selections) {
+        Set<String> accepted = new HashSet<>();
+        for (String value : values) {
+            for (String coding : value.split(",")) {
+                int parameters = coding.indexOf(';');
+                String name = normalize(parameters >= 0 ? coding.substring(0, parameters) : coding);
+                if (!name.isEmpty()) {
+                    accepted.add(name);
+                }
+            }
+        }
+        boolean brotli = selections.brotli() && accepted.contains("br");
+        boolean gzip = selections.gzip() && accepted.contains("gzip");
+        if (brotli && gzip) {
+            return "br,gzip";
+        }
+        if (brotli) {
+            return "br";
+        }
+        if (gzip) {
+            return "gzip";
+        }
+        return selections.legacy() ? null : "identity";
+    }
+
+    /**
+     * The viewer cookies the behavior forwards, sorted by name as CloudFront sorts them. Pairs that are
+     * not {@code name=value}, and names starting with {@code $}, are dropped.
+     */
+    private static String forwardedCookies(List<String> cookieHeaders, List<Selection> selections) {
+        List<String[]> cookies = new ArrayList<>();
+        for (String cookieHeader : cookieHeaders) {
+            if (cookieHeader == null) {
+                continue;
+            }
+            for (String pair : cookieHeader.split(";")) {
+                int eq = pair.indexOf('=');
+                String name = eq > 0 ? pair.substring(0, eq).trim() : "";
+                if (name.isEmpty() || name.startsWith("$")) {
+                    continue;
+                }
+                if (selections.stream().anyMatch(selection -> selection.includes(name))) {
+                    cookies.add(new String[] {name, pair.substring(eq + 1).trim()});
+                }
+            }
+        }
+        if (cookies.isEmpty()) {
+            return null;
+        }
+        cookies.sort(Comparator.comparing(cookie -> cookie[0]));
+        List<String> pairs = new ArrayList<>(cookies.size());
+        for (String[] cookie : cookies) {
+            pairs.add(cookie[0] + "=" + cookie[1]);
+        }
+        return String.join("; ", pairs);
+    }
+
+    /** The raw query pairs the behavior forwards, in viewer order and with their original encoding. */
+    private static String forwardedQuery(String rawQuery, List<Selection> selections) {
+        if (rawQuery == null || rawQuery.isEmpty()) {
+            return null;
+        }
+        if (selections.stream().anyMatch(selection -> "all".equals(selection.behavior()))) {
+            return rawQuery;
+        }
+        List<String> kept = new ArrayList<>();
+        for (String pair : rawQuery.split("&")) {
+            if (pair.isEmpty()) {
+                continue;
+            }
+            int eq = pair.indexOf('=');
+            String name = decodeQueryName(eq >= 0 ? pair.substring(0, eq) : pair);
+            if (selections.stream().anyMatch(selection -> selection.includes(name))) {
+                kept.add(pair);
+            }
+        }
+        return kept.isEmpty() ? null : String.join("&", kept);
+    }
+
+    private static String decodeQueryName(String rawName) {
+        try {
+            return URLDecoder.decode(rawName, StandardCharsets.UTF_8);
+        } catch (IllegalArgumentException malformed) {
+            // A name that is not valid percent-encoding can only match a policy entry literally.
+            return rawName;
+        }
+    }
+
+    private static String generatedValue(String header, ViewerRequest viewer) {
+        return switch (header) {
+            case FORWARDED_PROTO -> viewer.scheme() != null
+                    ? viewer.scheme().toLowerCase(Locale.ROOT) : null;
+            case VIEWER_ADDRESS -> viewer.clientAddress() != null && viewer.clientPort() > 0
+                    ? viewer.clientAddress() + ":" + viewer.clientPort() : null;
+            case VIEWER_HTTP_VERSION -> viewer.httpVersion();
+            default -> null;
+        };
+    }
+
+    /** CloudFront appends the viewer's address to any {@code X-Forwarded-For} the viewer sent. */
+    private static String forwardedFor(List<String> viewerValues, String clientAddress) {
+        List<String> values = new ArrayList<>(viewerValues);
+        if (clientAddress != null && !clientAddress.isBlank()) {
+            values.add(clientAddress);
+        }
+        return values.isEmpty() ? null : String.join(",", values);
+    }
+
+    private static String viewerProtocolVersion(String httpVersion) {
+        return httpVersion == null || httpVersion.isBlank() ? "1.1" : httpVersion;
+    }
+
+    /** A request identifier shaped like CloudFront's: 40 random bytes, URL-safe Base64. */
+    private static String requestId() {
+        byte[] bytes = new byte[40];
+        ThreadLocalRandom.current().nextBytes(bytes);
+        return Base64.getUrlEncoder().encodeToString(bytes);
+    }
+
+    private static boolean isTrue(Object value) {
+        return value instanceof Boolean bool ? bool : value != null && "true".equalsIgnoreCase(value.toString());
+    }
+
+    private static List<String> stringList(Map<?, ?> map, String key) {
+        Object value = map != null ? map.get(key) : null;
+        if (!(value instanceof List<?> list)) {
+            return List.of();
+        }
+        List<String> strings = new ArrayList<>(list.size());
+        for (Object item : list) {
+            if (item != null) {
+                strings.add(item.toString());
+            }
+        }
+        return strings;
+    }
+
+    private static String normalize(String name) {
+        return name == null ? "" : name.trim().toLowerCase(Locale.ROOT);
+    }
+
+    /** The resolved selections of one request, unioned across the policies that apply. */
+    private record Selections(boolean legacy, List<Selection> headers, boolean legacyDefaults,
+                              List<Selection> cookies, List<Selection> queryStrings,
+                              boolean gzip, boolean brotli, boolean normalizesAcceptEncoding) {
+
+        /**
+         * Whether a viewer header is forwarded: a policy selects it, or legacy settings forward it
+         * by default.
+         */
+        boolean forwardsHeader(String normalizedName) {
+            if (headers.stream().anyMatch(selection -> selection.includes(normalizedName))) {
+                return true;
+            }
+            return legacyDefaults && !LEGACY_REMOVED.contains(normalizedName);
+        }
+
+        /**
+         * Whether CloudFront adds one of its own request headers: a whitelist or
+         * {@code allViewerAndWhitelistCloudFront} list names it, or {@code allExcept} does not
+         * exclude it (the managed AllViewerExceptHostHeader policy is documented to add them).
+         */
+        boolean generatesHeader(String normalizedName) {
+            for (Selection selection : headers) {
+                boolean generated = switch (selection.behavior()) {
+                    case "whitelist", "allViewerAndWhitelistCloudFront" -> selection.lists(normalizedName);
+                    case "allExcept" -> !selection.lists(normalizedName);
+                    default -> false;
+                };
+                if (generated) {
+                    return true;
+                }
+            }
+            return false;
+        }
+    }
+
+    /**
+     * One policy's choice of names of one kind: headers, cookies or query strings. Header names match
+     * case-insensitively, cookie and query string names case-sensitively. Legacy cookie allowlists may
+     * use {@code *} and {@code ?} wildcards.
+     */
+    private record Selection(String behavior, List<String> names, boolean ignoreCase,
+                             boolean wildcards) {
+
+        static Selection of(Object block, String behaviorKey, String listKey, boolean ignoreCase) {
+            if (!(block instanceof Map<?, ?> map) || map.get(behaviorKey) == null) {
+                return new Selection("none", List.of(), ignoreCase, false);
+            }
+            return new Selection(map.get(behaviorKey).toString(), stringList(map, listKey),
+                    ignoreCase, false);
+        }
+
+        boolean includes(String name) {
+            return switch (behavior) {
+                case "whitelist" -> lists(name);
+                case "all", "allViewer", "allViewerAndWhitelistCloudFront" -> true;
+                case "allExcept" -> !lists(name);
+                default -> false;
+            };
+        }
+
+        boolean lists(String name) {
+            for (String candidate : names) {
+                boolean matches = wildcards
+                        ? wildcardMatches(candidate, name)
+                        : ignoreCase ? candidate.equalsIgnoreCase(name) : candidate.equals(name);
+                if (matches) {
+                    return true;
+                }
+            }
+            return false;
+        }
+
+        private static boolean wildcardMatches(String pattern, String name) {
+            return wildcardMatches(pattern, 0, name, 0);
+        }
+
+        private static boolean wildcardMatches(String pattern, int p, String name, int n) {
+            if (p == pattern.length()) {
+                return n == name.length();
+            }
+            char c = pattern.charAt(p);
+            if (c == '*') {
+                for (int i = n; i <= name.length(); i++) {
+                    if (wildcardMatches(pattern, p + 1, name, i)) {
+                        return true;
+                    }
+                }
+                return false;
+            }
+            return n < name.length() && (c == '?' || c == name.charAt(n))
+                    && wildcardMatches(pattern, p + 1, name, n + 1);
+        }
+    }
+}

src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontRequestRouter.java

@@ -166,6 +166,35 @@ public final class CloudFrontRequestRouter {
         return allowedMethodsOrDefault(dflt != null ? dflt.getAllowedMethods() : null);
     }
 
+    /**
+     * The settings that decide what the cache behavior serving a normalized path forwards to its
+     * origin: cache policy, origin request policy, legacy {@code ForwardedValues} and cached methods.
+     */
+    public record BehaviorForwarding(String cachePolicyId, String originRequestPolicyId,
+                                     Map<String, Object> forwardedValues,
+                                     List<String> cachedMethods) {
+    }
+
+    /** Returns the forwarding settings of the cache behavior that serves a normalized path. */
+    public static BehaviorForwarding matchForwarding(DistributionConfig config, String normalizedPath) {
+        List<CacheBehavior> behaviors = config.getCacheBehaviors();
+        if (behaviors != null) {
+            for (CacheBehavior behavior : behaviors) {
+                if (pathPatternMatches(behavior.getPathPattern(), normalizedPath)) {
+                    return new BehaviorForwarding(behavior.getCachePolicyId(),
+                            behavior.getOriginRequestPolicyId(), behavior.getForwardedValues(),
+                            behavior.getCachedMethods());
+                }
+            }
+        }
+        DefaultCacheBehavior dflt = config.getDefaultCacheBehavior();
+        if (dflt == null) {
+            return new BehaviorForwarding(null, null, null, null);
+        }
+        return new BehaviorForwarding(dflt.getCachePolicyId(), dflt.getOriginRequestPolicyId(),
+                dflt.getForwardedValues(), dflt.getCachedMethods());
+    }
+
     private static List<String> allowedMethodsOrDefault(List<String> methods) {
         return methods == null || methods.isEmpty() ? List.of("GET", "HEAD") : methods;
     }

src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontService.java

@@ -59,6 +59,7 @@ public class CloudFrontService {
             Set.of("s3", "mediastore", "mediapackagev2", "lambda");
     private static final Set<String> OAC_SIGNING_BEHAVIORS =
             Set.of("always", "never", "no-override");
+    private static final Set<String> FUNCTION_RUNTIMES = Set.of("cloudfront-js-1.0", "cloudfront-js-2.0");
     private static final int MAX_ORIGIN_CUSTOM_HEADERS = 30;
     private static final int MAX_CUSTOM_HEADER_NAME_LENGTH = 256;
     private static final int MAX_CUSTOM_HEADER_VALUE_LENGTH = 1_783;
@@ -83,6 +84,24 @@ public class CloudFrontService {
             "60669652-455b-4ae9-85a4-c4c02393f86c";
     private static final Map<String, ResponseHeadersPolicy> MANAGED_RESPONSE_HEADERS_POLICIES =
             managedResponseHeadersPolicies();
+    static final String MANAGED_ALL_VIEWER_ORIGIN_REQUEST_POLICY_ID =
+            "216adef6-5c7f-47e4-b989-5492eafa07d3";
+    static final String MANAGED_ALL_VIEWER_AND_CLOUDFRONT_HEADERS_ORIGIN_REQUEST_POLICY_ID =
+            "33f36d7e-f396-46d9-90e0-52428a34d9dc";
+    static final String MANAGED_ALL_VIEWER_EXCEPT_HOST_HEADER_ORIGIN_REQUEST_POLICY_ID =
+            "b689b0a8-53d0-40ab-baf2-68738e2966ac";
+    static final String MANAGED_CORS_CUSTOM_ORIGIN_ORIGIN_REQUEST_POLICY_ID =
+            "59781a5b-3903-41f3-afcb-af62929ccde1";
+    static final String MANAGED_CORS_S3_ORIGIN_ORIGIN_REQUEST_POLICY_ID =
+            "88a5eaf4-2fd4-4709-b370-b4c650ea3fcf";
+    static final String MANAGED_MEDIATAILOR_ORIGIN_REQUEST_POLICY_ID =
+            "775133bc-15f2-49f9-abea-afb2e0bf67d2";
+    static final String MANAGED_HOST_HEADER_ONLY_ORIGIN_REQUEST_POLICY_ID =
+            "bf0718e1-ba1e-49d1-88b1-f726733018ae";
+    static final String MANAGED_USER_AGENT_REFERER_ORIGIN_REQUEST_POLICY_ID =
+            "acba4595-bd28-49b8-b9fe-13317c0390fa";
+    private static final Map<String, OriginRequestPolicy> MANAGED_ORIGIN_REQUEST_POLICIES =
+            managedOriginRequestPolicies();
 
     private final StorageBackend<String, Distribution> distStore;
     private final StorageBackend<String, List<Invalidation>> invalidationStore;
@@ -543,14 +562,19 @@ public class CloudFrontService {
     }
 
     public OriginRequestPolicy getOriginRequestPolicy(String id) {
-        return orpStore.get(id).orElseThrow(() ->
+        return orpStore.get(id)
-                new AwsException("NoSuchOriginRequestPolicy",
+                .or(() -> Optional.ofNullable(MANAGED_ORIGIN_REQUEST_POLICIES.get(id)))
+                .orElseThrow(() -> new AwsException("NoSuchOriginRequestPolicy",
                         "The specified origin request policy does not exist.", 404));
     }
 
     public synchronized OriginRequestPolicy updateOriginRequestPolicy(String id, String ifMatch,
                                                                        OriginRequestPolicy updated) {
         OriginRequestPolicy existing = getOriginRequestPolicy(id);
+        if (isManagedOriginRequestPolicy(id)) {
+            throw new AwsException("IllegalUpdate",
+                    "AWS managed origin request policies cannot be updated.", 400);
+        }
         if (!existing.getEtag().equals(ifMatch)) {
             throw new AwsException("InvalidIfMatchVersion",
                     "The If-Match version is missing or not valid for the resource.", 400);
@@ -564,6 +588,10 @@ public class CloudFrontService {
 
     public synchronized void deleteOriginRequestPolicy(String id, String ifMatch) {
         OriginRequestPolicy existing = getOriginRequestPolicy(id);
+        if (isManagedOriginRequestPolicy(id)) {
+            throw new AwsException("IllegalDelete",
+                    "AWS managed origin request policies cannot be deleted.", 400);
+        }
         if (!existing.getEtag().equals(ifMatch)) {
             throw new AwsException("InvalidIfMatchVersion",
                     "The If-Match version is missing or not valid for the resource.", 400);
@@ -577,12 +605,34 @@ public class CloudFrontService {
     }
 
     public List<OriginRequestPolicy> listOriginRequestPolicies(String marker, int maxItems) {
-        List<OriginRequestPolicy> all = new ArrayList<>(orpStore.scan(k -> true));
+        return listOriginRequestPolicies(marker, maxItems, null);
+    }
+
+    /**
+     * Lists custom and AWS managed origin request policies; {@code type} narrows the list to the
+     * AWS lowercase {@code managed} or {@code custom} value.
+     */
+    public List<OriginRequestPolicy> listOriginRequestPolicies(String marker, int maxItems, String type) {
+        if (type != null && !"custom".equals(type) && !"managed".equals(type)) {
+            throw new AwsException("InvalidArgument",
+                    "Origin request policy Type must be managed or custom.", 400);
+        }
+        List<OriginRequestPolicy> all = new ArrayList<>();
+        if (!"managed".equals(type)) {
+            all.addAll(orpStore.scan(k -> true));
+        }
+        if (!"custom".equals(type)) {
+            all.addAll(MANAGED_ORIGIN_REQUEST_POLICIES.values());
+        }
         all.sort((a, b) -> a.getName() != null && b.getName() != null
                 ? a.getName().compareTo(b.getName()) : a.getId().compareTo(b.getId()));
         return paginate(all, marker, maxItems, OriginRequestPolicy::getId);
     }
 
+    static boolean isManagedOriginRequestPolicy(String id) {
+        return MANAGED_ORIGIN_REQUEST_POLICIES.containsKey(id);
+    }
+
     // ── Response Headers Policies ─────────────────────────────────────────────
 
     public synchronized ResponseHeadersPolicy createResponseHeadersPolicy(ResponseHeadersPolicy policy) {
@@ -840,6 +890,112 @@ public class CloudFrontService {
         return Map.copyOf(policies);
     }
 
+    /**
+     * The AWS managed origin request policies, with the IDs and settings the CloudFront Developer
+     * Guide lists under "Use managed origin request policies".
+     */
+    private static Map<String, OriginRequestPolicy> managedOriginRequestPolicies() {
+        Map<String, OriginRequestPolicy> policies = new LinkedHashMap<>();
+        policies.put(MANAGED_ALL_VIEWER_ORIGIN_REQUEST_POLICY_ID, managedOriginRequestPolicy(
+                MANAGED_ALL_VIEWER_ORIGIN_REQUEST_POLICY_ID, "Managed-AllViewer",
+                "Includes all values (headers, cookies, and query strings) from the viewer request",
+                selection("HeaderBehavior", "allViewer", "Headers", List.of()),
+                selection("CookieBehavior", "all", "Cookies", List.of()),
+                selection("QueryStringBehavior", "all", "QueryStrings", List.of())));
+        policies.put(MANAGED_ALL_VIEWER_AND_CLOUDFRONT_HEADERS_ORIGIN_REQUEST_POLICY_ID,
+                managedOriginRequestPolicy(
+                        MANAGED_ALL_VIEWER_AND_CLOUDFRONT_HEADERS_ORIGIN_REQUEST_POLICY_ID,
+                        "Managed-AllViewerAndCloudFrontHeaders-2022-06",
+                        "Includes all values from the viewer request and all CloudFront headers "
+                                + "released through June 2022",
+                        selection("HeaderBehavior", "allViewerAndWhitelistCloudFront", "Headers",
+                                List.of("CloudFront-Forwarded-Proto", "CloudFront-Is-Android-Viewer",
+                                        "CloudFront-Is-Desktop-Viewer", "CloudFront-Is-IOS-Viewer",
+                                        "CloudFront-Is-Mobile-Viewer", "CloudFront-Is-SmartTV-Viewer",
+                                        "CloudFront-Is-Tablet-Viewer", "CloudFront-Viewer-Address",
+                                        "CloudFront-Viewer-ASN", "CloudFront-Viewer-City",
+                                        "CloudFront-Viewer-Country", "CloudFront-Viewer-Country-Name",
+                                        "CloudFront-Viewer-Country-Region",
+                                        "CloudFront-Viewer-Country-Region-Name",
+                                        "CloudFront-Viewer-Http-Version", "CloudFront-Viewer-Latitude",
+                                        "CloudFront-Viewer-Longitude", "CloudFront-Viewer-Metro-Code",
+                                        "CloudFront-Viewer-Postal-Code", "CloudFront-Viewer-Time-Zone",
+                                        "CloudFront-Viewer-TLS")),
+                        selection("CookieBehavior", "all", "Cookies", List.of()),
+                        selection("QueryStringBehavior", "all", "QueryStrings", List.of())));
+        policies.put(MANAGED_ALL_VIEWER_EXCEPT_HOST_HEADER_ORIGIN_REQUEST_POLICY_ID,
+                managedOriginRequestPolicy(
+                        MANAGED_ALL_VIEWER_EXCEPT_HOST_HEADER_ORIGIN_REQUEST_POLICY_ID,
+                        "Managed-AllViewerExceptHostHeader",
+                        "Includes all values from the viewer request except the Host header",
+                        selection("HeaderBehavior", "allExcept", "Headers", List.of("Host")),
+                        selection("CookieBehavior", "all", "Cookies", List.of()),
+                        selection("QueryStringBehavior", "all", "QueryStrings", List.of())));
+        policies.put(MANAGED_CORS_CUSTOM_ORIGIN_ORIGIN_REQUEST_POLICY_ID, managedOriginRequestPolicy(
+                MANAGED_CORS_CUSTOM_ORIGIN_ORIGIN_REQUEST_POLICY_ID, "Managed-CORS-CustomOrigin",
+                "Includes the header that enables CORS requests when the origin is a custom origin",
+                selection("HeaderBehavior", "whitelist", "Headers", List.of("Origin")),
+                selection("CookieBehavior", "none", "Cookies", List.of()),
+                selection("QueryStringBehavior", "none", "QueryStrings", List.of())));
+        policies.put(MANAGED_CORS_S3_ORIGIN_ORIGIN_REQUEST_POLICY_ID, managedOriginRequestPolicy(
+                MANAGED_CORS_S3_ORIGIN_ORIGIN_REQUEST_POLICY_ID, "Managed-CORS-S3Origin",
+                "Includes the headers that enable CORS requests when the origin is an Amazon S3 bucket",
+                selection("HeaderBehavior", "whitelist", "Headers", List.of(
+                        "Origin", "Access-Control-Request-Headers", "Access-Control-Request-Method")),
+                selection("CookieBehavior", "none", "Cookies", List.of()),
+                selection("QueryStringBehavior", "none", "QueryStrings", List.of())));
+        policies.put(MANAGED_MEDIATAILOR_ORIGIN_REQUEST_POLICY_ID, managedOriginRequestPolicy(
+                MANAGED_MEDIATAILOR_ORIGIN_REQUEST_POLICY_ID,
+                "Managed-Elemental-MediaTailor-PersonalizedManifests",
+                "For use with an origin that is an AWS Elemental MediaTailor endpoint",
+                selection("HeaderBehavior", "whitelist", "Headers", List.of(
+                        "Origin", "Access-Control-Request-Headers", "Access-Control-Request-Method",
+                        "User-Agent", "X-Forwarded-For")),
+                selection("CookieBehavior", "none", "Cookies", List.of()),
+                selection("QueryStringBehavior", "all", "QueryStrings", List.of())));
+        policies.put(MANAGED_HOST_HEADER_ONLY_ORIGIN_REQUEST_POLICY_ID, managedOriginRequestPolicy(
+                MANAGED_HOST_HEADER_ONLY_ORIGIN_REQUEST_POLICY_ID, "Managed-HostHeaderOnly",
+                "Includes only the Host header from the viewer request",
+                selection("HeaderBehavior", "whitelist", "Headers", List.of("Host")),
+                selection("CookieBehavior", "none", "Cookies", List.of()),
+                selection("QueryStringBehavior", "none", "QueryStrings", List.of())));
+        policies.put(MANAGED_USER_AGENT_REFERER_ORIGIN_REQUEST_POLICY_ID, managedOriginRequestPolicy(
+                MANAGED_USER_AGENT_REFERER_ORIGIN_REQUEST_POLICY_ID, "Managed-UserAgentRefererHeaders",
+                "Includes only the User-Agent and Referer headers",
+                selection("HeaderBehavior", "whitelist", "Headers", List.of("User-Agent", "Referer")),
+                selection("CookieBehavior", "none", "Cookies", List.of()),
+                selection("QueryStringBehavior", "none", "QueryStrings", List.of())));
+        return Map.copyOf(policies);
+    }
+
+    private static OriginRequestPolicy managedOriginRequestPolicy(
+            String id, String name, String comment, Map<String, Object> headers,
+            Map<String, Object> cookies, Map<String, Object> queryStrings) {
+        Map<String, Object> config = new LinkedHashMap<>();
+        config.put("HeadersConfig", headers);
+        config.put("CookiesConfig", cookies);
+        config.put("QueryStringsConfig", queryStrings);
+        OriginRequestPolicy policy = new OriginRequestPolicy();
+        policy.setId(id);
+        policy.setName(name);
+        policy.setComment(comment);
+        policy.setEtag("E23ZP02F085DFQ");
+        policy.setLastModifiedTime(Instant.EPOCH);
+        policy.setConfig(Map.copyOf(config));
+        return policy;
+    }
+
+    /** One policy selection block, in the shape {@link CloudFrontPolicyConfigCodec} stores. */
+    private static Map<String, Object> selection(String behaviorKey, String behavior,
+                                                 String listKey, List<String> names) {
+        Map<String, Object> block = new LinkedHashMap<>();
+        block.put(behaviorKey, behavior);
+        if (!names.isEmpty()) {
+            block.put(listKey, List.copyOf(names));
+        }
+        return Map.copyOf(block);
+    }
+
     private static ResponseHeadersPolicy managedResponseHeadersPolicy(
             String id, String name, String comment, Map<String, Object> config) {
         ResponseHeadersPolicy policy = new ResponseHeadersPolicy();
@@ -1072,6 +1228,7 @@ public class CloudFrontService {
     // ── CloudFront Functions ──────────────────────────────────────────────────
 
     public synchronized CloudFrontFunction createFunction(CloudFrontFunction fn) {
+        validateFunctionRuntime(fn.getRuntime());
         fn.setStage("DEVELOPMENT");
         fn.setStatus("UNPUBLISHED");
         fn.setEtag(UUID.randomUUID().toString());
@@ -1096,6 +1253,7 @@ public class CloudFrontService {
 
     public synchronized CloudFrontFunction updateFunction(String name, String ifMatch,
                                                           CloudFrontFunction updated) {
+        validateFunctionRuntime(updated.getRuntime());
         CloudFrontFunction existing = describeFunction(name, null);
         if (!existing.getEtag().equals(ifMatch)) {
             throw new AwsException("InvalidIfMatchVersion",
@@ -1168,6 +1326,13 @@ public class CloudFrontService {
         return effectiveStage;
     }
 
+    private static void validateFunctionRuntime(String runtime) {
+        if (runtime != null && !FUNCTION_RUNTIMES.contains(runtime)) {
+            throw new AwsException("InvalidArgument",
+                    "The parameter Runtime must be cloudfront-js-1.0 or cloudfront-js-2.0.", 400);
+        }
+    }
+
     private static AwsException invalidFunctionStage(String stage) {
         return new AwsException("InvalidArgument",
                 "The parameter Stage must be DEVELOPMENT or LIVE: " + stage, 400);

src/main/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontServingController.java

@@ -10,24 +10,36 @@ import io.github.hectorvent.floci.services.cloudfront.model.OriginAccessControl;
 import io.github.hectorvent.floci.services.s3.S3Service;
 import io.github.hectorvent.floci.services.s3.model.S3Object;
 import io.quarkus.vertx.http.runtime.CurrentVertxRequest;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.http.HttpVersion;
+import io.vertx.core.net.SocketAddress;
 import jakarta.annotation.PreDestroy;
 import jakarta.inject.Inject;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.DELETE;
 import jakarta.ws.rs.GET;
 import jakarta.ws.rs.HEAD;
 import jakarta.ws.rs.OPTIONS;
+import jakarta.ws.rs.PATCH;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.PUT;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.PathParam;
 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 org.jboss.resteasy.reactive.server.core.CurrentRequestManager;
 
+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;
@@ -53,14 +65,19 @@ import java.util.Set;
  * <ul>
  *   <li>S3 origins are read in-process through {@link S3Service} (the bucket is derived from the
  *       origin domain name).</li>
- *   <li>Custom origins are fetched over HTTP(S) with {@link java.net.http.HttpClient}.</li>
+ *   <li>Custom origins are fetched over HTTP(S) with {@link CloudFrontOriginHttpClient}, carrying the
+ *       viewer headers, cookies and query strings the matched behavior forwards, as decided by
+ *       {@link CloudFrontOriginRequestBuilder}.</li>
  *   <li>When an origin returns an error status, a matching {@code CustomErrorResponse} is applied —
  *       most importantly the single-page-app fallback that rewrites 403/404 to 200 {@code /index.html}.
  *       If the configured error page is itself missing, the received status is returned (no loop).</li>
  * </ul>
  *
  * <p>Viewer-protocol-policy enforcement is intentionally out of scope for this layer: the emulator
- * is HTTP-first. GET/HEAD are served, and OPTIONS is served only when the matched behavior allows it.
+ * is HTTP-first. Every CloudFront viewer method (GET, HEAD, OPTIONS, POST, PUT, PATCH, DELETE) is
+ * served when the matched behavior's {@code AllowedMethods} includes it. POST, PUT, PATCH and DELETE
+ * are forwarded to custom origins with the viewer request body and {@code Authorization} header; they
+ * are not forwarded to in-process S3 origins.
  */
 @Path("/_cloudfront/{distId}")
 public class CloudFrontServingController {
@@ -80,6 +97,8 @@ public class CloudFrontServingController {
 
     private static final java.util.Set<String> CLOUDFRONT_SIGNING_PARAMS =
             java.util.Set.of("Expires", "Signature", "Key-Pair-Id", "Policy", "Hash-Algorithm");
+    /** Viewer methods that can carry a request body and that CloudFront never caches. */
+    private static final Set<String> BODY_METHODS = Set.of("POST", "PUT", "PATCH", "DELETE");
 
     private final CloudFrontService service;
     private final S3Service s3Service;
@@ -117,7 +136,7 @@ public class CloudFrontServingController {
                 RequestHost.of(request),
                 headers.getHeaderString(HttpHeaders.AUTHORIZATION),
                 request.getHeader("Origin"), "GET", null, null,
-                request.getHeader("Pragma"));
+                request.getHeader("Pragma"), null);
     }
 
     @HEAD
@@ -131,7 +150,7 @@ public class CloudFrontServingController {
                 RequestHost.of(request),
                 headers.getHeaderString(HttpHeaders.AUTHORIZATION),
                 request.getHeader("Origin"), "HEAD", null, null,
-                request.getHeader("Pragma"));
+                request.getHeader("Pragma"), null);
     }
 
     @OPTIONS
@@ -147,7 +166,86 @@ public class CloudFrontServingController {
                 request.getHeader("Origin"), "OPTIONS",
                 request.getHeader("Access-Control-Request-Method"),
                 request.getHeader("Access-Control-Request-Headers"),
-                request.getHeader("Pragma"));
+                request.getHeader("Pragma"), null);
+    }
+
+    @POST
+    @Path("/{proxy:.*}")
+    @Consumes(MediaType.WILDCARD)
+    public Response post(@PathParam("distId") String distId, @PathParam("proxy") String proxy,
+                         @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) {
+        return serveWithBody(distId, "POST", headers, uriInfo, body);
+    }
+
+    @PUT
+    @Path("/{proxy:.*}")
+    @Consumes(MediaType.WILDCARD)
+    public Response put(@PathParam("distId") String distId, @PathParam("proxy") String proxy,
+                        @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) {
+        return serveWithBody(distId, "PUT", headers, uriInfo, body);
+    }
+
+    @PATCH
+    @Path("/{proxy:.*}")
+    @Consumes(MediaType.WILDCARD)
+    public Response patch(@PathParam("distId") String distId, @PathParam("proxy") String proxy,
+                          @Context HttpHeaders headers, @Context UriInfo uriInfo, InputStream body) {
+        return serveWithBody(distId, "PATCH", headers, uriInfo, body);
+    }
+
+    @DELETE
+    @Path("/{proxy:.*}")
+    @Consumes(MediaType.WILDCARD)
+    public Response delete(@PathParam("distId") String distId, @PathParam("proxy") String proxy,
+                           @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, InputStream body) {
+        HttpServerRequest request = currentVertxRequest.getCurrent().request();
+        String rawViewerPath = rawViewerPath(request.uri());
+        return serve(distId, rawViewerPath, decodedViewerPath(rawViewerPath),
+                uriInfo.getRequestUri().getScheme(),
+                RequestHost.of(request),
+                headers.getHeaderString(HttpHeaders.AUTHORIZATION),
+                request.getHeader("Origin"), method, null, null,
+                request.getHeader("Pragma"),
+                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,
@@ -156,7 +254,8 @@ public class CloudFrontServingController {
                            String viewerOrigin, String method,
                            String accessControlRequestMethod,
                            String accessControlRequestHeaders,
-                           String pragma) {
+                           String pragma,
+                           ViewerBody viewerBody) {
         boolean includeBody = !"HEAD".equals(method);
         boolean preflightRequest = "OPTIONS".equals(method)
                 && viewerOrigin != null && !viewerOrigin.isBlank()
@@ -199,11 +298,17 @@ public class CloudFrontServingController {
 
         OriginResponse origin = route(dist, normalized, rawViewerPath, decodedViewerPath,
                 viewerScheme, viewerAuthorization, method, viewerOrigin,
-                accessControlRequestMethod, accessControlRequestHeaders);
+                accessControlRequestMethod, accessControlRequestHeaders, viewerBody);
 
         if (origin.status() >= 400) {
-            Response fallback = applyCustomError(
+            Response fallback;
+            try {
+                fallback = applyCustomError(
                         dist, origin, viewerScheme, viewerAuthorization, includeBody, directives);
+            } catch (RuntimeException e) {
+                origin.closeStream();
+                throw e;
+            }
             if (fallback != null) {
                 return fallback;
             }
@@ -345,7 +450,8 @@ public class CloudFrontServingController {
                                  String viewerScheme, String viewerAuthorization,
                                  String method, String viewerOrigin,
                                  String accessControlRequestMethod,
-                                 String accessControlRequestHeaders) {
+                                 String accessControlRequestHeaders,
+                                 ViewerBody viewerBody) {
         DistributionConfig config = distribution.getConfig();
         String originId = CloudFrontRequestRouter.matchTargetOriginId(config, normalized);
         Origin origin = CloudFrontRequestRouter.findOrigin(config, originId);
@@ -357,6 +463,14 @@ public class CloudFrontServingController {
                 return fetchS3Preflight(origin, viewerOrigin, accessControlRequestMethod,
                         accessControlRequestHeaders);
             }
+            if (BODY_METHODS.contains(method)) {
+                // AWS forwards these methods and S3 evaluates them against the bucket policy (OAC
+                // supports PUT and DELETE). The in-process origin authorizes reads only, so a write is
+                // answered as S3 answers a request that holds no write grant.
+                LOG.debugv("CloudFront does not forward {0} requests to in-process S3 origin {1}",
+                        method, origin.getId());
+                return OriginResponse.error(403, "Access Denied");
+            }
             String key = CloudFrontRequestRouter.resolveOriginKey(
                     origin.getOriginPath(), decodedViewerPath, config.getDefaultRootObject());
             return fetchFromS3(
@@ -364,11 +478,86 @@ public class CloudFrontServingController {
         }
         String forwardUri = CloudFrontRequestRouter.resolveForwardUri(
                 origin.getOriginPath(), rawViewerPath, config.getDefaultRootObject());
-        // CloudFront forwards viewer query strings only when selected by a cache policy,
+        CloudFrontOriginRequestBuilder.OriginRequest originRequest =
-        // origin request policy, or legacy ForwardedValues configuration. Those policy
+                CloudFrontOriginRequestBuilder.build(viewerRequest(method, viewerScheme),
-        // semantics are not modeled in the data plane yet, so the AWS default is to omit them.
+                        forwarding(config, normalized), distribution.getDomainName());
-        return fetchFromCustomOrigin(origin, forwardUri, null, viewerScheme, method,
+        return fetchFromCustomOrigin(
-                viewerOrigin, accessControlRequestMethod, accessControlRequestHeaders);
+                origin, forwardUri, originRequest, viewerScheme, method, viewerBody);
+    }
+
+    /** The live viewer request, with CloudFront's signing parameters removed from its query. */
+    private CloudFrontOriginRequestBuilder.ViewerRequest viewerRequest(
+            String method, String viewerScheme) {
+        HttpServerRequest request = currentVertxRequest.getCurrent().request();
+        List<CloudFrontOriginRequestBuilder.Header> headers = new ArrayList<>();
+        for (Map.Entry<String, String> header : request.headers()) {
+            headers.add(new CloudFrontOriginRequestBuilder.Header(header.getKey(), header.getValue()));
+        }
+        String appQuery = stripCloudFrontSigningParams(request.query());
+        SocketAddress client = request.remoteAddress();
+        return new CloudFrontOriginRequestBuilder.ViewerRequest(method, headers,
+                appQuery.isEmpty() ? null : appQuery,
+                client != null ? client.hostAddress() : null,
+                client != null ? client.port() : -1,
+                viewerScheme, httpVersion(request.version()));
+    }
+
+    private static String httpVersion(HttpVersion version) {
+        if (version == null) {
+            return null;
+        }
+        return switch (version) {
+            case HTTP_1_0 -> "1.0";
+            case HTTP_1_1 -> "1.1";
+            case HTTP_2 -> "2.0";
+        };
+    }
+
+    /**
+     * Resolves what the matched behavior forwards: legacy {@code ForwardedValues} when it has no cache
+     * policy, otherwise its cache policy parameters and origin request policy.
+     */
+    private CloudFrontOriginRequestBuilder.Forwarding forwarding(
+            DistributionConfig config, String normalized) {
+        CloudFrontRequestRouter.BehaviorForwarding behavior =
+                CloudFrontRequestRouter.matchForwarding(config, normalized);
+        boolean optionsCached = behavior.cachedMethods() != null
+                && behavior.cachedMethods().contains("OPTIONS");
+        if (behavior.cachePolicyId() == null || behavior.cachePolicyId().isBlank()) {
+            return CloudFrontOriginRequestBuilder.Forwarding.legacy(
+                    behavior.forwardedValues(), optionsCached);
+        }
+        return CloudFrontOriginRequestBuilder.Forwarding.policies(
+                cachePolicyParameters(behavior.cachePolicyId()),
+                originRequestPolicyConfig(behavior.originRequestPolicyId()), optionsCached);
+    }
+
+    private Map<?, ?> cachePolicyParameters(String cachePolicyId) {
+        try {
+            Map<String, Object> config = service.getCachePolicy(cachePolicyId).getConfig();
+            return config != null
+                    && config.get("ParametersInCacheKeyAndForwardedToOrigin") instanceof Map<?, ?> parameters
+                    ? parameters : null;
+        } catch (AwsException e) {
+            // AWS managed cache policies are not modeled. The common ones (CachingDisabled,
+            // CachingOptimized) forward no viewer values, which is what an unknown id contributes.
+            LOG.debugv("Cache policy {0} is not defined; it forwards no viewer values", cachePolicyId);
+            return null;
+        }
+    }
+
+    private Map<?, ?> originRequestPolicyConfig(String originRequestPolicyId) {
+        if (originRequestPolicyId == null || originRequestPolicyId.isBlank()) {
+            return null;
+        }
+        try {
+            return service.getOriginRequestPolicy(originRequestPolicyId).getConfig();
+        } catch (AwsException e) {
+            // Distribution writes do not validate this reference, so keep serving without the policy
+            // but make the dangling id visible.
+            LOG.warnv("Distribution references missing origin request policy {0}", originRequestPolicyId);
+            return null;
+        }
     }
 
     private OriginResponse fetchFromS3(
@@ -386,10 +575,12 @@ public class CloudFrontServingController {
             authorizeS3OriginRead(
                     distribution, origin, bucket, key, viewerAuthorization);
             if (includeBody) {
-                S3Object obj = s3Service.getObject(bucket, key);
+                // The body is streamed to the viewer, so an object of any size is served without
-                byte[] data = obj.getData() != null ? obj.getData() : new byte[0];
+                // 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(
@@ -398,8 +589,13 @@ public class CloudFrontServingController {
         } catch (AwsException e) {
             response = OriginResponse.error(e.getHttpStatus(), e.getMessage());
         }
+        try {
             return withS3CorsHeaders(
                     bucket, origin, includeBody ? "GET" : "HEAD", response);
+        } catch (RuntimeException e) {
+            response.closeStream();
+            throw e;
+        }
     }
 
     private void authorizeS3OriginRead(
@@ -490,15 +686,17 @@ public class CloudFrontServingController {
                         "This CORS request is not allowed."));
     }
 
-    /** Fetches from a custom (non-S3) origin. {@code forwardUri} already includes the origin path. */
+    /**
-    private OriginResponse fetchFromCustomOrigin(Origin origin, String forwardUri, String rawQuery,
+     * Fetches from a custom (non-S3) origin. {@code forwardUri} already includes the origin path;
-                                                 String viewerScheme, String method, String viewerOrigin,
+     * {@code originRequest} carries the forwarded query and headers.
-                                                 String accessControlRequestMethod,
+     */
-                                                 String accessControlRequestHeaders) {
+    private OriginResponse fetchFromCustomOrigin(Origin origin, String forwardUri,
+                                                 CloudFrontOriginRequestBuilder.OriginRequest originRequest,
+                                                 String viewerScheme, String method, ViewerBody viewerBody) {
         boolean includeBody = !"HEAD".equals(method);
         try {
             URI target = buildCustomOriginUri(
-                    origin, viewerScheme, forwardUri, rawQuery);
+                    origin, viewerScheme, forwardUri, originRequest.rawQuery());
             HttpRequest.Builder rb = HttpRequest.newBuilder()
                     .uri(target)
                     .timeout(Duration.ofSeconds(30));
@@ -517,18 +715,19 @@ public class CloudFrontServingController {
                 }
             }
             rb.method(method, HttpRequest.BodyPublishers.noBody());
-            if ("OPTIONS".equals(method)) {
+            // Both bodies are streamed, so neither has to fit in memory: the viewer's goes out with its
-                addRequestHeader(rb, "Origin", viewerOrigin);
+            // own Content-Length, or chunked as it came, and the origin's comes back as it arrives.
-                addRequestHeader(rb, "Access-Control-Request-Method", accessControlRequestMethod);
+            ViewerBody requestBody = originRequestBody(method, viewerBody);
-                addRequestHeader(rb, "Access-Control-Request-Headers", accessControlRequestHeaders);
+            CloudFrontOriginHttpClient.OpenResponse resp = httpClient.open(rb.build(), originRequest.headers(),
-            }
+                    originHeaders, requestBody != null ? requestBody.stream() : null,
-            HttpResponse<byte[]> resp = httpClient.send(
+                    requestBody != null ? requestBody.length() : -1);
-                    rb.build(), originHeaders, HttpResponse.BodyHandlers.ofByteArray());
             String ct = resp.headers().firstValue("content-type").orElse(DEFAULT_CONTENT_TYPE);
-            byte[] body = resp.body() != null ? resp.body() : new byte[0];
+            long contentLength = resp.contentLength();
-            long contentLength = includeBody ? body.length : responseContentLength(resp);
+            if (!includeBody) {
-            return new OriginResponse(resp.statusCode(), ct, includeBody ? body : null,
+                resp.body().close();
-                    contentLength, resp.headers().map());
+                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}",
@@ -542,6 +741,24 @@ 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 ViewerBody originRequestBody(String method, ViewerBody viewerBody) {
+        if (viewerBody == null || !BODY_METHODS.contains(method)) {
+            return null;
+        }
+        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<String, String> config = origin.getS3OriginConfig();
         String value = config != null ? config.get("OriginAccessIdentity") : null;
@@ -639,15 +856,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());
@@ -688,8 +896,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(),
+            return toResponse(origin.withStatus(responseCode), includeBody, directives);
-                    origin.contentLength(), origin.headers()), includeBody, directives);
         }
 
         String errNormalized = CloudFrontRequestRouter.normalizePath(pagePath);
@@ -699,6 +906,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);
@@ -707,16 +916,15 @@ public class CloudFrontServingController {
         } else {
             String forwardUri = CloudFrontRequestRouter.resolveForwardUri(errOrigin.getOriginPath(), errNormalized, null);
             page = fetchFromCustomOrigin(
-                    errOrigin, forwardUri, null, viewerScheme,
+                    errOrigin, forwardUri, CloudFrontOriginRequestBuilder.OriginRequest.empty(),
-                    includeBody ? "GET" : "HEAD", null, null, null);
+                    viewerScheme, includeBody ? "GET" : "HEAD", null);
         }
         if (page.status() >= 400) {
             // Custom error page unavailable → return the status received from the error-page origin
             // (AWS behavior), without recursively applying custom error handling.
             return toResponse(page, includeBody, directives);
         }
-        return toResponse(new OriginResponse(responseCode, page.contentType(), page.body(),
+        return toResponse(page.withStatus(responseCode), includeBody, directives);
-                page.contentLength(), page.headers()), includeBody, directives);
     }
 
     private Map<String, Object> matchCustomError(DistributionConfig config, int status) {
@@ -734,6 +942,7 @@ public class CloudFrontServingController {
 
     private Response toResponse(OriginResponse origin, boolean includeBody,
                                 ResponseHeadersPolicyConfigCodec.Directives directives) {
+        try {
             Response.ResponseBuilder rb = Response.status(origin.status());
             HeaderCollection collected = collectResponseHeaders(origin, includeBody);
             applyResponseHeadersPolicy(collected.headers(), directives, collected.forbiddenPolicyHeaders());
@@ -745,10 +954,31 @@ public class CloudFrontServingController {
                     header.values().forEach(value -> rb.header(header.name(), value));
                 }
             }
+            if (includeBody && origin.stream() != null) {
+                // A disconnected viewer can skip StreamingOutput entirely, so its finally block
+                // alone cannot release the origin connection.
+                CurrentRequestManager.get().registerCompletionCallback(ignored -> origin.closeStream());
+                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;
+        }
     }
 
     /**
@@ -784,7 +1014,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);
@@ -935,12 +1166,7 @@ public class CloudFrontServingController {
                                 "Access-Control-Request-Method"));
             }
         });
-        return new OriginResponse(
+        return response.withHeaders(headers);
-                response.status(),
-                response.contentType(),
-                response.body(),
-                response.contentLength(),
-                headers);
     }
 
     private static String originCustomHeader(Origin origin, String requestedName) {
@@ -1006,12 +1232,6 @@ public class CloudFrontServingController {
         return values;
     }
 
-    private static void addRequestHeader(HttpRequest.Builder request, String name, String value) {
-        if (value != null && !value.isBlank()) {
-            request.header(name, value);
-        }
-    }
-
     private static String str(Object value) {
         return value != null ? value.toString() : "";
     }
@@ -1039,12 +1259,42 @@ public class CloudFrontServingController {
                                     Set<String> 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,
+     * 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<String, List<String>> headers) {
+        OriginResponse(int status, String contentType, byte[] body, long contentLength,
                        Map<String, List<String>> 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<String, List<String>> 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());
+            }
+        }
     }
 }

src/main/java/io/github/hectorvent/floci/services/s3/S3Service.java

@@ -1332,6 +1332,23 @@ public class S3Service implements Resettable, ResourceProvider {
         }
     }
 
+    /** An object's metadata together with an open stream of its body; the caller closes it. */
+    public record ObjectRead(S3Object object, InputStream body) implements java.io.Closeable {
+        @Override
+        public void close() throws IOException {
+            body.close();
+        }
+    }
+
+    /**
+     * Like {@link #getObject}, but the body is an open stream instead of a byte array, so a large
+     * object is served without being read into the heap.
+     */
+    public ObjectRead openObject(String bucketName, String key, String versionId) {
+        S3Object object = getObjectMetadata(bucketName, key, versionId);
+        return new ObjectRead(object, openObjectStream(bucketName, key, versionId));
+    }
+
     public InputStream openObjectStream(String bucketName, String key, String versionId) {
         getObjectMetadata(bucketName, key, versionId);
         String bucketOwnerAccount = resolveBucketEntry(bucketName)

src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontCustomOriginServingTest.java

@@ -1,39 +1,70 @@
 package io.github.hectorvent.floci.services.cloudfront;
 
+import com.sun.net.httpserver.Headers;
 import com.sun.net.httpserver.HttpExchange;
 import com.sun.net.httpserver.HttpServer;
+import io.github.hectorvent.floci.services.cloudfront.model.CacheBehavior;
+import io.github.hectorvent.floci.services.cloudfront.model.CachePolicy;
 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.KeyGroup;
 import io.github.hectorvent.floci.services.cloudfront.model.Origin;
+import io.github.hectorvent.floci.services.cloudfront.model.OriginRequestPolicy;
+import io.github.hectorvent.floci.services.cloudfront.model.PublicKey;
 import io.github.hectorvent.floci.services.cloudfront.model.ResponseHeadersPolicy;
 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.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static io.restassured.RestAssured.given;
 import static org.hamcrest.Matchers.equalTo;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 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.assertNotNull;
 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
 @TestProfile(CloudFrontCustomOriginServingTest.PrivateOriginProfile.class)
 class CloudFrontCustomOriginServingTest {
 
+    /** The AWS managed CachingDisabled cache policy, which forwards no viewer values. */
+    private static final String CACHING_DISABLED = "4135ea2d-6df8-44a3-9df3-4b5a84be39ad";
+
     @Inject
     CloudFrontService cloudFrontService;
 
@@ -220,6 +251,726 @@ class CloudFrontCustomOriginServingTest {
                 removeOnlyResponse.getHeader("Date"));
     }
 
+    @Test
+    void forwardsWriteMethodsWithTheirBodyToTheOrigin() throws Exception {
+        AtomicReference<ReceivedRequest> received = new AtomicReference<>();
+        AtomicInteger hits = new AtomicInteger();
+        startRecordingOrigin(received, hits);
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(allMethodsBehavior()), Map.of());
+
+        String json = "{\"name\":\"widget\",\"note\":\"café\"}";
+        given()
+            .header("Host", created.getDomainName())
+            .contentType("application/json; charset=UTF-8")
+            .body(json.getBytes(StandardCharsets.UTF_8))
+        .when()
+            .post("/api/things")
+        .then()
+            .statusCode(201)
+            .header("X-Origin-Method", equalTo("POST"))
+            .body(equalTo("created"));
+        ReceivedRequest post = received.get();
+        assertEquals("POST", post.method());
+        assertEquals("/api/things", post.path());
+        assertEquals("application/json; charset=UTF-8", post.header("Content-Type"));
+        assertEquals(Integer.toString(json.getBytes(StandardCharsets.UTF_8).length),
+                post.header("Content-Length"));
+        assertArrayEquals(json.getBytes(StandardCharsets.UTF_8), post.body());
+
+        byte[] binary = new byte[] {0, 1, 2, (byte) 0xff, (byte) 0xfe, 10, 13};
+        given()
+            .header("Host", created.getDomainName())
+            .contentType("application/octet-stream")
+            .body(binary)
+        .when()
+            .put("/api/things/1")
+        .then()
+            .statusCode(200)
+            .header("X-Origin-Method", equalTo("PUT"));
+        assertEquals("PUT", received.get().method());
+        assertTrue(received.get().header("Content-Type").startsWith("application/octet-stream"));
+        assertArrayEquals(binary, received.get().body());
+
+        given()
+            .header("Host", created.getDomainName())
+            .contentType("application/x-www-form-urlencoded")
+            .body("name=widget&size=2")
+        .when()
+            .patch("/api/things/1")
+        .then()
+            .statusCode(200)
+            .header("X-Origin-Method", equalTo("PATCH"));
+        assertEquals("PATCH", received.get().method());
+        assertTrue(received.get().header("Content-Type")
+                .startsWith("application/x-www-form-urlencoded"));
+        assertEquals("name=widget&size=2",
+                new String(received.get().body(), StandardCharsets.UTF_8));
+
+        given()
+            .header("Host", created.getDomainName())
+        .when()
+            .delete("/api/things/1")
+        .then()
+            .statusCode(200)
+            .header("X-Origin-Method", equalTo("DELETE"));
+        assertEquals("DELETE", received.get().method());
+        assertEquals("/api/things/1", received.get().path());
+        assertEquals(0, received.get().body().length);
+        assertNull(received.get().header("Content-Type"));
+
+        given()
+            .header("Host", created.getDomainName())
+        .when()
+            .post("/api/empty")
+        .then()
+            .statusCode(201);
+        assertEquals("0", received.get().header("Content-Length"));
+        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<ReceivedRequest> 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<ReceivedRequest> 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 releasesOriginConnectionsWhenViewersDisconnectBeforeOriginHeaders() throws Exception {
+        BlockingQueue<CountDownLatch> pendingResponses = new LinkedBlockingQueue<>();
+        originServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        originServer.createContext("/", exchange -> {
+            CountDownLatch sendResponse = new CountDownLatch(1);
+            pendingResponses.add(sendResponse);
+            try {
+                if (!sendResponse.await(10, TimeUnit.SECONDS)) {
+                    throw new IOException("Timed out waiting for viewer disconnect");
+                }
+                byte[] body = "origin-ok".getBytes(StandardCharsets.UTF_8);
+                exchange.sendResponseHeaders(200, body.length);
+                exchange.getResponseBody().write(body);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new IOException("Origin response interrupted", e);
+            } finally {
+                exchange.close();
+            }
+        });
+        originServer.createContext("/after-disconnects", exchange -> {
+            byte[] body = "origin-ok".getBytes(StandardCharsets.UTF_8);
+            exchange.sendResponseHeaders(200, body.length);
+            exchange.getResponseBody().write(body);
+            exchange.close();
+        });
+        originServer.start();
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(allMethodsBehavior()), Map.of());
+
+        for (int i = 0; i < CloudFrontOriginHttpClient.MAX_CONNECTIONS + 10; i++) {
+            disconnectViewerBeforeOriginHeaders(created.getDomainName(), i, pendingResponses);
+        }
+
+        given().header("Host", created.getDomainName()).when().get("/after-disconnects")
+                .then().statusCode(200).body(equalTo("origin-ok"));
+    }
+
+    @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<String, Object> 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<ReceivedRequest> received = new AtomicReference<>();
+        AtomicInteger hits = new AtomicInteger();
+        startRecordingOrigin(received, hits);
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(allMethodsBehavior()), Map.of());
+
+        for (String method : List.of("POST", "PUT", "PATCH", "DELETE")) {
+            given()
+                .header("Host", created.getDomainName())
+                .header("Authorization", "Bearer viewer-token")
+            .when()
+                .request(method, "/api/things/1")
+            .then()
+                .header("X-Origin-Method", equalTo(method));
+            assertEquals("Bearer viewer-token", received.get().header("Authorization"), method);
+        }
+
+        // CloudFront removes Authorization from GET and HEAD requests.
+        given()
+            .header("Host", created.getDomainName())
+            .header("Authorization", "Bearer viewer-token")
+        .when()
+            .get("/api/things/1")
+        .then()
+            .statusCode(200)
+            .header("X-Origin-Method", equalTo("GET"));
+        assertNull(received.get().header("Authorization"));
+        assertEquals(5, hits.get());
+    }
+
+    @Test
+    void customErrorPageForAWriteIsFetchedWithoutTheViewerAuthorization() throws Exception {
+        Map<String, String> authorizationByRequest = new ConcurrentHashMap<>();
+        originServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        originServer.createContext("/", exchange -> {
+            String request = exchange.getRequestMethod() + " "
+                    + exchange.getRequestURI().getRawPath();
+            authorizationByRequest.put(request,
+                    String.valueOf(exchange.getRequestHeaders().getFirst("Authorization")));
+            boolean errorPage = "GET /errors/500.html".equals(request);
+            byte[] body = (errorPage ? "error-page" : "failed").getBytes(StandardCharsets.UTF_8);
+            exchange.getResponseHeaders().add("Content-Type", "text/plain");
+            exchange.sendResponseHeaders(errorPage ? 200 : 500, body.length);
+            exchange.getResponseBody().write(body);
+            exchange.close();
+        });
+        originServer.start();
+
+        Map<String, Object> errorPage = new LinkedHashMap<>(Map.of(
+                "ErrorCode", "500", "ResponseCode", "503", "ResponsePagePath", "/errors/500.html"));
+        Distribution distribution = customOriginDistribution(allMethodsBehavior());
+        distribution.getConfig().setCustomErrorResponses(List.of(errorPage));
+        Distribution created = cloudFrontService.createDistribution(distribution, Map.of());
+
+        given()
+            .header("Host", created.getDomainName())
+            .header("Authorization", "Bearer viewer-token")
+            .contentType("application/json")
+            .body("{}")
+        .when()
+            .post("/api/things")
+        .then()
+            .statusCode(503)
+            .body(equalTo("error-page"));
+        assertEquals(Map.of(
+                "POST /api/things", "Bearer viewer-token",
+                "GET /errors/500.html", "null"), authorizationByRequest);
+    }
+
+    @Test
+    void rejectsMethodsTheMatchedBehaviorDoesNotAllowWithoutContactingTheOrigin() throws Exception {
+        AtomicReference<ReceivedRequest> received = new AtomicReference<>();
+        AtomicInteger hits = new AtomicInteger();
+        startRecordingOrigin(received, hits);
+
+        CacheBehavior readOnly = new CacheBehavior();
+        readOnly.setPathPattern("/static/*");
+        readOnly.setTargetOriginId("custom-origin");
+        readOnly.setViewerProtocolPolicy("allow-all");
+        readOnly.setAllowedMethods(List.of("GET", "HEAD"));
+        DistributionConfig config = customOriginDistribution(allMethodsBehavior()).getConfig();
+        config.setCacheBehaviors(List.of(readOnly));
+        Distribution distribution = new Distribution();
+        distribution.setConfig(config);
+        Distribution created = cloudFrontService.createDistribution(distribution, Map.of());
+
+        for (String method : List.of("POST", "PUT", "PATCH", "DELETE")) {
+            given()
+                .header("Host", created.getDomainName())
+                .body("ignored")
+            .when()
+                .request(method, "/static/app.js")
+            .then()
+                .statusCode(403)
+                .body(equalTo("Invalid method."));
+        }
+        assertEquals(0, hits.get());
+
+        // A behavior without AllowedMethods keeps CloudFront's GET/HEAD minimum.
+        Distribution defaultsOnly = cloudFrontService.createDistribution(
+                customOriginDistribution(defaultBehavior("custom-origin")), Map.of());
+        given()
+            .header("Host", defaultsOnly.getDomainName())
+            .body("ignored")
+        .when()
+            .post("/api/things")
+        .then()
+            .statusCode(403)
+            .body(equalTo("Invalid method."));
+        assertEquals(0, hits.get());
+    }
+
+    @Test
+    void signedBehaviorsRequireASignatureForWriteMethods() throws Exception {
+        AtomicReference<ReceivedRequest> received = new AtomicReference<>();
+        AtomicInteger hits = new AtomicInteger();
+        startRecordingOrigin(received, hits);
+
+        String suffix = Long.toString(System.nanoTime(), 36);
+        KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA");
+        generator.initialize(2048);
+        KeyPair keyPair = generator.generateKeyPair();
+        PublicKey publicKey = new PublicKey();
+        publicKey.setName("pk-" + suffix);
+        publicKey.setCallerReference("cr-" + suffix);
+        publicKey.setEncodedKey("-----BEGIN PUBLIC KEY-----\n"
+                + Base64.getMimeEncoder().encodeToString(keyPair.getPublic().getEncoded())
+                + "\n-----END PUBLIC KEY-----");
+        publicKey = cloudFrontService.createPublicKey(publicKey);
+        KeyGroup keyGroup = new KeyGroup();
+        keyGroup.setName("kg-" + suffix);
+        keyGroup.setItems(List.of(publicKey.getId()));
+        keyGroup = cloudFrontService.createKeyGroup(keyGroup);
+
+        DefaultCacheBehavior behavior = allMethodsBehavior();
+        behavior.setTrustedKeyGroups(List.of(keyGroup.getId()));
+        behavior.setCachePolicyId(CACHING_DISABLED);
+        behavior.setOriginRequestPolicyId(
+                CloudFrontService.MANAGED_ALL_VIEWER_ORIGIN_REQUEST_POLICY_ID);
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(behavior), Map.of());
+
+        given()
+            .header("Host", created.getDomainName())
+            .contentType("application/json")
+            .body("{\"unsigned\":true}")
+        .when()
+            .post("/api/things")
+        .then()
+            .statusCode(403);
+        assertEquals(0, hits.get());
+
+        long expires = Instant.now().getEpochSecond() + 3600;
+        String policy = "{\"Statement\":[{\"Resource\":\"*\",\"Condition\":{\"DateLessThan\":"
+                + "{\"AWS:EpochTime\":" + expires + "}}}]}";
+        Signature signer = Signature.getInstance("SHA1withRSA");
+        signer.initSign(keyPair.getPrivate());
+        signer.update(policy.getBytes(StandardCharsets.UTF_8));
+
+        given()
+            .header("Host", created.getDomainName())
+            .queryParam("color", "blue")
+            .queryParam("Policy", cfBase64(policy.getBytes(StandardCharsets.UTF_8)))
+            .queryParam("Signature", cfBase64(signer.sign()))
+            .queryParam("Key-Pair-Id", publicKey.getId())
+            .contentType("application/json")
+            .body("{\"signed\":true}")
+        .when()
+            .post("/api/things")
+        .then()
+            .statusCode(201);
+        assertEquals(1, hits.get());
+        assertEquals("{\"signed\":true}",
+                new String(received.get().body(), StandardCharsets.UTF_8));
+        // AllViewer forwards every query string, but never CloudFront's signing parameters.
+        assertEquals("color=blue", received.get().query());
+        assertEquals(created.getDomainName(), received.get().header("Host"));
+    }
+
+    @Test
+    void allViewerExceptHostHeaderForwardsViewerCookiesHeadersAndQuery() throws Exception {
+        AtomicReference<ReceivedRequest> received = new AtomicReference<>();
+        AtomicInteger hits = new AtomicInteger();
+        startRecordingOrigin(received, hits);
+        DefaultCacheBehavior behavior = policyBehavior(CACHING_DISABLED,
+                CloudFrontService.MANAGED_ALL_VIEWER_EXCEPT_HOST_HEADER_ORIGIN_REQUEST_POLICY_ID);
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(behavior, originVerifyHeader()), Map.of());
+
+        Response response = given()
+            .header("Host", created.getDomainName())
+            .header("Cookie", "session=abc123; csrf=xyz")
+            .header("Authorization", "Bearer t")
+            .header("X-Custom", "custom")
+            .header("User-Agent", "viewer-agent/1.0")
+            .header("X-Origin-Verify", "forged")
+            .header("X-Origin-Verify", "forged-again")
+        .when()
+            .get("/api/me?x=1")
+        .then()
+            .statusCode(200)
+            .extract().response();
+
+        ReceivedRequest request = received.get();
+        assertEquals("/api/me", request.path());
+        assertEquals("x=1", request.query());
+        assertEquals("csrf=xyz; session=abc123", request.header("Cookie"));
+        assertEquals("Bearer t", request.header("Authorization"));
+        assertEquals("custom", request.header("X-Custom"));
+        assertEquals("viewer-agent/1.0", request.header("User-Agent"));
+        assertEquals(List.of("s3cr3t"), request.headers().get("X-Origin-Verify"));
+        assertEquals("127.0.0.1:" + originServer.getAddress().getPort(), request.header("Host"));
+        assertEquals("1.1 " + created.getDomainName() + " (CloudFront)", request.header("Via"));
+        assertFalse(request.header("X-Forwarded-For").isBlank());
+        assertEquals(56, request.header("X-Amz-Cf-Id").length());
+        assertEquals("http", request.header("CloudFront-Forwarded-Proto"));
+        assertEquals(List.of("session=renewed; Path=/; HttpOnly"),
+                response.getHeaders().getValues("Set-Cookie"));
+    }
+
+    @Test
+    void cachePolicyWithoutAnOriginRequestPolicyForwardsNoViewerValues() throws Exception {
+        AtomicReference<ReceivedRequest> received = new AtomicReference<>();
+        AtomicInteger hits = new AtomicInteger();
+        startRecordingOrigin(received, hits);
+        DefaultCacheBehavior behavior = policyBehavior(CACHING_DISABLED, null);
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(behavior, originVerifyHeader()), Map.of());
+
+        given()
+            .header("Host", created.getDomainName())
+            .header("Cookie", "session=abc123")
+            .header("Authorization", "Bearer t")
+            .header("X-Custom", "custom")
+            .header("User-Agent", "viewer-agent/1.0")
+        .when()
+            .get("/api/me?x=1")
+        .then()
+            .statusCode(200);
+        ReceivedRequest get = received.get();
+        assertNull(get.query());
+        assertNull(get.header("Cookie"));
+        assertNull(get.header("Authorization"));
+        assertNull(get.header("X-Custom"));
+        assertEquals("Amazon CloudFront", get.header("User-Agent"));
+        assertEquals("s3cr3t", get.header("X-Origin-Verify"));
+        assertEquals("127.0.0.1:" + originServer.getAddress().getPort(), get.header("Host"));
+
+        // CloudFront never strips Authorization from methods it does not cache.
+        given()
+            .header("Host", created.getDomainName())
+            .header("Cookie", "session=abc123")
+            .header("Authorization", "Bearer t")
+            .contentType("application/json")
+            .body("{}")
+        .when()
+            .post("/api/things")
+        .then()
+            .statusCode(201);
+        ReceivedRequest post = received.get();
+        assertEquals("Bearer t", post.header("Authorization"));
+        assertTrue(post.header("Content-Type").startsWith("application/json"));
+        assertNull(post.header("Cookie"));
+    }
+
+    @Test
+    void customPoliciesFilterCookiesHeadersAndQueryStrings() throws Exception {
+        AtomicReference<ReceivedRequest> received = new AtomicReference<>();
+        AtomicInteger hits = new AtomicInteger();
+        startRecordingOrigin(received, hits);
+
+        CachePolicy cachePolicy = new CachePolicy();
+        cachePolicy.setName("forwarding-cache-policy-" + System.nanoTime());
+        cachePolicy.setConfig(CloudFrontPolicyConfigCodec.parseCachePolicy("""
+                <CachePolicyConfig>
+                  <MinTTL>0</MinTTL><DefaultTTL>0</DefaultTTL><MaxTTL>0</MaxTTL>
+                  <ParametersInCacheKeyAndForwardedToOrigin>
+                    <EnableAcceptEncodingGzip>false</EnableAcceptEncodingGzip>
+                    <HeadersConfig><HeaderBehavior>whitelist</HeaderBehavior>
+                      <Headers><Quantity>1</Quantity><Items><Name>Accept-Language</Name></Items></Headers>
+                    </HeadersConfig>
+                    <CookiesConfig><CookieBehavior>none</CookieBehavior></CookiesConfig>
+                    <QueryStringsConfig><QueryStringBehavior>none</QueryStringBehavior></QueryStringsConfig>
+                  </ParametersInCacheKeyAndForwardedToOrigin>
+                </CachePolicyConfig>
+                """));
+        cachePolicy = cloudFrontService.createCachePolicy(cachePolicy);
+        OriginRequestPolicy originRequestPolicy = new OriginRequestPolicy();
+        originRequestPolicy.setName("forwarding-origin-policy-" + System.nanoTime());
+        originRequestPolicy.setConfig(CloudFrontPolicyConfigCodec.parseOriginRequestPolicy("""
+                <OriginRequestPolicyConfig>
+                  <HeadersConfig><HeaderBehavior>whitelist</HeaderBehavior>
+                    <Headers><Quantity>1</Quantity><Items><Name>X-Custom</Name></Items></Headers>
+                  </HeadersConfig>
+                  <CookiesConfig><CookieBehavior>whitelist</CookieBehavior>
+                    <Cookies><Quantity>1</Quantity><Items><Name>session</Name></Items></Cookies>
+                  </CookiesConfig>
+                  <QueryStringsConfig><QueryStringBehavior>allExcept</QueryStringBehavior>
+                    <QueryStrings><Quantity>1</Quantity><Items><Name>utm_source</Name></Items></QueryStrings>
+                  </QueryStringsConfig>
+                </OriginRequestPolicyConfig>
+                """));
+        originRequestPolicy = cloudFrontService.createOriginRequestPolicy(originRequestPolicy);
+        Distribution created = cloudFrontService.createDistribution(customOriginDistribution(
+                policyBehavior(cachePolicy.getId(), originRequestPolicy.getId())), Map.of());
+
+        given()
+            .urlEncodingEnabled(false)
+            .header("Host", created.getDomainName())
+            .header("Cookie", "tracking=1; session=abc; theme=dark")
+            .header("X-Custom", "custom")
+            .header("X-Other", "other")
+            .header("Accept-Language", "en-GB")
+        .when()
+            .get("/search?q=red%20shoes&utm_source=ad&page=2")
+        .then()
+            .statusCode(200);
+
+        ReceivedRequest request = received.get();
+        assertEquals("q=red%20shoes&page=2", request.query());
+        assertEquals("session=abc", request.header("Cookie"));
+        assertEquals("custom", request.header("X-Custom"));
+        assertEquals("en-GB", request.header("Accept-Language"));
+        assertNull(request.header("X-Other"));
+    }
+
+    @Test
+    void legacyForwardedValuesSelectQueryCookiesAndHeaders() throws Exception {
+        AtomicReference<ReceivedRequest> received = new AtomicReference<>();
+        AtomicInteger hits = new AtomicInteger();
+        startRecordingOrigin(received, hits);
+        Map<String, Object> forwardedValues = new LinkedHashMap<>();
+        forwardedValues.put("QueryString", true);
+        forwardedValues.put("CookiesForward", "whitelist");
+        forwardedValues.put("CookieNames", List.of("session"));
+        forwardedValues.put("Headers", List.of("Authorization"));
+        DefaultCacheBehavior behavior = defaultBehavior("custom-origin");
+        behavior.setForwardedValues(forwardedValues);
+        Distribution created = cloudFrontService.createDistribution(
+                customOriginDistribution(behavior), Map.of());
+
+        given()
+            .header("Host", created.getDomainName())
+            .header("Cookie", "session=abc; other=1")
+            .header("Authorization", "Bearer t")
+            .header("X-Custom", "custom")
+            .header("Accept-Language", "en-GB")
+            .header("User-Agent", "viewer-agent/1.0")
+        .when()
+            .get("/legacy?x=1&y=2")
+        .then()
+            .statusCode(200);
+
+        ReceivedRequest request = received.get();
+        assertEquals("x=1&y=2", request.query());
+        assertEquals("session=abc", request.header("Cookie"));
+        assertEquals("Bearer t", request.header("Authorization"));
+        assertEquals("custom", request.header("X-Custom"));
+        assertNull(request.header("Accept-Language"));
+        assertEquals("Amazon CloudFront", request.header("User-Agent"));
+        assertEquals("127.0.0.1:" + originServer.getAddress().getPort(), request.header("Host"));
+    }
+
+    private static void disconnectViewerBeforeOriginHeaders(String host, int requestIndex,
+                                                            BlockingQueue<CountDownLatch> pendingResponses)
+            throws Exception {
+        CountDownLatch sendResponse = null;
+        try (Socket viewer = new Socket("127.0.0.1", RestAssured.port)) {
+            viewer.setSoLinger(true, 0);
+            viewer.getOutputStream().write(("GET /disconnect/" + requestIndex + " HTTP/1.1\r\nHost: "
+                    + host + "\r\n\r\n").getBytes(StandardCharsets.US_ASCII));
+            sendResponse = pendingResponses.poll(5, TimeUnit.SECONDS);
+            assertNotNull(sendResponse, "No origin connection for viewer " + requestIndex);
+        } finally {
+            if (sendResponse != null) {
+                sendResponse.countDown();
+            }
+        }
+    }
+
+    private void startRecordingOrigin(AtomicReference<ReceivedRequest> received, AtomicInteger hits)
+            throws IOException {
+        originServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        originServer.createContext("/", exchange -> {
+            hits.incrementAndGet();
+            received.set(new ReceivedRequest(
+                    exchange.getRequestMethod(),
+                    exchange.getRequestURI().getRawPath(),
+                    exchange.getRequestURI().getRawQuery(),
+                    exchange.getRequestHeaders(),
+                    exchange.getRequestBody().readAllBytes()));
+            byte[] body = ("POST".equals(exchange.getRequestMethod()) ? "created" : "ok")
+                    .getBytes(StandardCharsets.UTF_8);
+            exchange.getResponseHeaders().add("Content-Type", "text/plain");
+            exchange.getResponseHeaders().add("X-Origin-Method", exchange.getRequestMethod());
+            exchange.getResponseHeaders().add("Set-Cookie", "session=renewed; Path=/; HttpOnly");
+            exchange.sendResponseHeaders(
+                    "POST".equals(exchange.getRequestMethod()) ? 201 : 200, body.length);
+            exchange.getResponseBody().write(body);
+            exchange.close();
+        });
+        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);
+    }
+
+    private Distribution customOriginDistribution(DefaultCacheBehavior behavior,
+                                                  List<Map<String, String>> customHeaders) {
+        Origin customOrigin = new Origin();
+        customOrigin.setId("custom-origin");
+        customOrigin.setDomainName("127.0.0.1");
+        customOrigin.setCustomOriginConfig(customOriginConfig(originServer.getAddress().getPort()));
+        customOrigin.setCustomHeaders(customHeaders);
+        DistributionConfig config = new DistributionConfig();
+        config.setEnabled(true);
+        config.setOrigins(List.of(customOrigin));
+        config.setDefaultCacheBehavior(behavior);
+        Distribution distribution = new Distribution();
+        distribution.setConfig(config);
+        return distribution;
+    }
+
+    private static List<Map<String, String>> originVerifyHeader() {
+        return List.of(new LinkedHashMap<>(Map.of(
+                "HeaderName", "X-Origin-Verify", "HeaderValue", "s3cr3t")));
+    }
+
+    private static DefaultCacheBehavior policyBehavior(String cachePolicyId,
+                                                       String originRequestPolicyId) {
+        DefaultCacheBehavior behavior = allMethodsBehavior();
+        behavior.setCachePolicyId(cachePolicyId);
+        behavior.setOriginRequestPolicyId(originRequestPolicyId);
+        return behavior;
+    }
+
+    private static DefaultCacheBehavior allMethodsBehavior() {
+        DefaultCacheBehavior behavior = defaultBehavior("custom-origin");
+        behavior.setAllowedMethods(List.of("GET", "HEAD", "OPTIONS", "PUT", "POST", "PATCH", "DELETE"));
+        behavior.setCachedMethods(List.of("GET", "HEAD"));
+        return behavior;
+    }
+
+    private static String cfBase64(byte[] bytes) {
+        return Base64.getEncoder().encodeToString(bytes)
+                .replace('+', '-').replace('=', '_').replace('/', '~');
+    }
+
+    private record ReceivedRequest(String method, String path, String query, Headers headers,
+                                   byte[] body) {
+        String header(String name) {
+            return headers.getFirst(name);
+        }
+    }
+
     private static void respond(HttpExchange exchange, AtomicReference<String> receivedQuery,
                                 AtomicReference<String> receivedPath,
                                 AtomicReference<String> receivedCustomHeader,

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")
@@ -1424,6 +1425,34 @@ class CloudFrontDistributionServingTest {
                 .then().statusCode(200);
     }
 
+    @Test
+    void writeMethodsAreNotForwardedToInProcessS3Origins() {
+        String suffix = suffix();
+        String bucket = "cf-writes-" + suffix;
+        createBucket(bucket);
+        putObject(bucket, "doc.txt", "ORIGINAL-" + suffix, "text/plain");
+
+        DefaultCacheBehavior dflt = defaultBehavior("only-origin");
+        dflt.setAllowedMethods(List.of("GET", "HEAD", "OPTIONS", "PUT", "POST", "PATCH", "DELETE"));
+        DistributionConfig config = new DistributionConfig();
+        config.setEnabled(true);
+        config.setOrigins(List.of(s3Origin("only-origin", bucket)));
+        config.setDefaultCacheBehavior(dflt);
+        Distribution distribution = cloudFrontService.createDistribution(distribution(config), Map.of());
+
+        for (String method : List.of("POST", "PUT", "PATCH", "DELETE")) {
+            given()
+                    .header("Host", distribution.getDomainName())
+                    .body("REPLACED-" + suffix)
+                    .when().request(method, "/doc.txt")
+                    .then().statusCode(403)
+                    .body(equalTo("Access Denied"));
+        }
+
+        given().header("Host", distribution.getDomainName()).when().get("/doc.txt")
+                .then().statusCode(200).body(equalTo("ORIGINAL-" + suffix));
+    }
+
     @Test
     void managedPreflightPoliciesEmitOnlyAllowOriginForSimpleRequests() {
         String suffix = suffix();

src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontManagementProtocolTest.java

@@ -109,6 +109,38 @@ class CloudFrontManagementProtocolTest {
                     equalTo("InvalidArgument")));
     }
 
+    @Test
+    void createFunctionRejectsAnUnknownRuntime() {
+        String name = "runtime-test-" + UUID.randomUUID();
+        String createBody = """
+                <CreateFunctionRequest xmlns="http://cloudfront.amazonaws.com/doc/2020-05-31/">
+                  <Name>%s</Name>
+                  <FunctionConfig>
+                    <Comment>routing function</Comment>
+                    <Runtime>python3</Runtime>
+                  </FunctionConfig>
+                  <FunctionCode>function handler(event) { return event.request; }</FunctionCode>
+                </CreateFunctionRequest>
+                """.formatted(name);
+
+        given()
+            .contentType("application/xml")
+            .body(createBody)
+        .when()
+            .post(API + "function")
+        .then()
+            .statusCode(400)
+            .body(hasXPath(
+                    "//*[local-name()='Code']/text()",
+                    equalTo("InvalidArgument")));
+
+        given()
+        .when()
+            .get(API + "function/" + name + "/describe")
+        .then()
+            .statusCode(404);
+    }
+
     @Test
     void describeAndPublishFunctionPreserveBothStages() {
         String name = "describe-test-" + UUID.randomUUID();
@@ -409,6 +441,42 @@ class CloudFrontManagementProtocolTest {
         }
     }
 
+    @Test
+    void managedOriginRequestPoliciesAreReadableListableAndImmutable() {
+        String exceptHost = "b689b0a8-53d0-40ab-baf2-68738e2966ac";
+        given()
+            .when()
+                .get(API + "origin-request-policy/" + exceptHost)
+            .then()
+                .statusCode(200)
+                .body(hasXPath("//*[local-name()='Name']/text()",
+                        equalTo("Managed-AllViewerExceptHostHeader")))
+                .body(hasXPath("//*[local-name()='HeaderBehavior']/text()", equalTo("allExcept")))
+                .body(hasXPath("//*[local-name()='Headers']//*[local-name()='Name']/text()",
+                        equalTo("Host")))
+                .body(hasXPath("//*[local-name()='CookieBehavior']/text()", equalTo("all")))
+                .body(hasXPath("//*[local-name()='QueryStringBehavior']/text()", equalTo("all")));
+
+        given()
+                .queryParam("Type", "managed")
+            .when()
+                .get(API + "origin-request-policy")
+            .then()
+                .statusCode(200)
+                .body(hasXPath("/*/*[local-name()='Quantity']/text()", equalTo("8")))
+                .body(hasXPath("//*[local-name()='OriginRequestPolicySummary']"
+                        + "[.//*[local-name()='Id' and text()='" + exceptHost + "']]"
+                        + "/*[local-name()='Type']/text()", equalTo("managed")));
+
+        given()
+                .header("If-Match", "E23ZP02F085DFQ")
+            .when()
+                .delete(API + "origin-request-policy/" + exceptHost)
+            .then()
+                .statusCode(400)
+                .body(hasXPath("//*[local-name()='Code']/text()", equalTo("IllegalDelete")));
+    }
+
     private static String cachePolicyBody(String name, String comment, String defaultTtl, String cookie) {
         return """
                 <CachePolicyConfig xmlns="http://cloudfront.amazonaws.com/doc/2020-05-31/">

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,
+            assertThrows(UnknownHostException.class, () -> get(client, request));
-                    () -> client.send(request, HttpResponse.BodyHandlers.ofByteArray()));
             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(
+            assertThrows(UnknownHostException.class, () -> get(client, request("http://mixed.invalid:8080/")));
-                    request("http://mixed.invalid:8080/"),
-                    HttpResponse.BodyHandlers.ofByteArray()));
         }
     }
 
@@ -99,12 +105,11 @@ class CloudFrontOriginHttpClientTest {
 
         try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(
                 resolver, List.of("rebind.invalid"))) {
-            HttpResponse<byte[]> response = client.send(
+            CloudFrontOriginHttpClient.OpenResponse response = get(client,
-                    request("http://rebind.invalid:" + server.getAddress().getPort() + "/"),
+                    request("http://rebind.invalid:" + server.getAddress().getPort() + "/"));
-                    HttpResponse.BodyHandlers.ofByteArray());
 
             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<byte[]> response = client.send(
+            CloudFrontOriginHttpClient.OpenResponse response = get(client,
-                    request("http://redirect.invalid:" + server.getAddress().getPort() + "/start"),
+                    request("http://redirect.invalid:" + server.getAddress().getPort() + "/start"));
-                    HttpResponse.BodyHandlers.ofByteArray());
+            readAll(response);
 
             assertEquals(302, response.statusCode());
             assertEquals(0, redirectedHits.get());
@@ -162,10 +167,7 @@ class CloudFrontOriginHttpClientTest {
                     .GET()
                     .build();
 
-            client.send(
+            readAll(client.open(request, List.of(), Map.of("X-Origin-Verify", "configured-value"), null, -1));
-                    request,
-                    Map.of("X-Origin-Verify", "configured-value"),
-                    HttpResponse.BodyHandlers.ofByteArray());
 
             assertEquals(List.of("configured-value"), receivedValues.get());
         } finally {
@@ -173,6 +175,47 @@ class CloudFrontOriginHttpClientTest {
         }
     }
 
+    @Test
+    void sendsTheBodyWithItsExactLengthAndTheCallerContentType() throws Exception {
+        AtomicReference<String> method = new AtomicReference<>();
+        AtomicReference<String> contentType = new AtomicReference<>();
+        AtomicReference<String> contentLength = new AtomicReference<>();
+        AtomicReference<byte[]> body = new AtomicReference<>();
+        HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        server.createContext("/", exchange -> {
+            method.set(exchange.getRequestMethod());
+            contentType.set(exchange.getRequestHeaders().getFirst("Content-Type"));
+            contentLength.set(exchange.getRequestHeaders().getFirst("Content-Length"));
+            body.set(exchange.getRequestBody().readAllBytes());
+            exchange.sendResponseHeaders(204, -1);
+            exchange.close();
+        });
+        server.start();
+
+        byte[] payload = "{\"a\":1}".getBytes(StandardCharsets.UTF_8);
+        try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(
+                resolver(InetAddress.getByName("127.0.0.1")), List.of("origin.invalid"))) {
+            HttpRequest request = HttpRequest.newBuilder(java.net.URI.create(
+                            "http://origin.invalid:" + server.getAddress().getPort() + "/items"))
+                    .timeout(Duration.ofSeconds(5))
+                    .header("Content-Type", "application/json")
+                    .method("PUT", HttpRequest.BodyPublishers.noBody())
+                    .build();
+
+            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());
+            assertEquals("application/json", contentType.get());
+            assertEquals(Integer.toString(payload.length), contentLength.get());
+            assertArrayEquals(payload, body.get());
+        } finally {
+            server.stop(0);
+        }
+    }
+
     @Test
     void httpsPinningPreservesSniAndHostnameVerification() throws Exception {
         if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) {
@@ -209,21 +252,125 @@ class CloudFrontOriginHttpClientTest {
         try (CloudFrontOriginHttpClient client = new CloudFrontOriginHttpClient(
                 resolver, List.of("origin.invalid", "wrong.invalid"), clientContext)) {
             int port = server.getAddress().getPort();
-            HttpResponse<byte[]> response = client.send(
+            CloudFrontOriginHttpClient.OpenResponse response = get(client,
-                    request("https://origin.invalid:" + port + "/"),
+                    request("https://origin.invalid:" + port + "/"));
-                    HttpResponse.BodyHandlers.ofByteArray());
+            readAll(response);
 
             assertEquals(200, response.statusCode());
             assertEquals("origin.invalid:" + port, hostHeader.get());
             assertEquals("origin.invalid", requestedSni.get());
-            assertThrows(IOException.class, () -> client.send(
+            assertThrows(IOException.class, () -> get(client, request("https://wrong.invalid:" + port + "/")));
-                    request("https://wrong.invalid:" + port + "/"),
-                    HttpResponse.BodyHandlers.ofByteArray()));
         } finally {
             server.stop(0);
         }
     }
 
+    @Test
+    void streamsARequestBodyChunkedWhenItsLengthIsUnknown() throws Exception {
+        AtomicReference<String> transferEncoding = new AtomicReference<>();
+        AtomicReference<byte[]> 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<InetSocketAddress> 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 -> {

src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontOriginRequestBuilderTest.java

@@ -0,0 +1,413 @@
+package io.github.hectorvent.floci.services.cloudfront;
+
+import io.github.hectorvent.floci.services.cloudfront.CloudFrontOriginRequestBuilder.Forwarding;
+import io.github.hectorvent.floci.services.cloudfront.CloudFrontOriginRequestBuilder.Header;
+import io.github.hectorvent.floci.services.cloudfront.CloudFrontOriginRequestBuilder.OriginRequest;
+import io.github.hectorvent.floci.services.cloudfront.CloudFrontOriginRequestBuilder.ViewerRequest;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class CloudFrontOriginRequestBuilderTest {
+
+    private static final String VIA_HOST = "d111111abcdef8.cloudfront.net";
+
+    @Test
+    void cachePolicyWithoutForwardingSendsOnlyTheHeadersCloudFrontAdds() {
+        OriginRequest request = build(viewer("GET", "x=1&y=2",
+                "Host", "d111111abcdef8.cloudfront.net",
+                "User-Agent", "curl/8.0",
+                "Cookie", "session=abc123",
+                "Authorization", "Bearer t",
+                "X-Custom", "custom",
+                "Origin", "https://viewer.example",
+                "X-Forwarded-For", "203.0.113.9"),
+                Forwarding.policies(parameters(none("Header"), none("Cookie"), none("QueryString")),
+                        null, false));
+
+        assertNull(request.rawQuery());
+        assertEquals(List.of("User-Agent", "X-Forwarded-For", "Via", "X-Amz-Cf-Id"), names(request));
+        assertEquals("Amazon CloudFront", value(request, "User-Agent"));
+        assertEquals("203.0.113.9,127.0.0.1", value(request, "X-Forwarded-For"));
+        assertEquals("1.1 " + VIA_HOST + " (CloudFront)", value(request, "Via"));
+        assertEquals(56, value(request, "X-Amz-Cf-Id").length());
+    }
+
+    @Test
+    void allExceptForwardsEveryViewerValueButTheListedHeaders() {
+        Map<String, Object> allViewerExceptHost = orp(
+                selection("HeaderBehavior", "allExcept", "Headers", "host"),
+                selection("CookieBehavior", "all", "Cookies"),
+                selection("QueryStringBehavior", "all", "QueryStrings"));
+
+        OriginRequest request = build(viewer("GET", "x=1&encoded=a%2Fb",
+                "Host", "d111111abcdef8.cloudfront.net",
+                "User-Agent", "curl/8.0",
+                "Cookie", "session=abc123; csrf=xyz",
+                "Authorization", "Bearer t",
+                "X-Custom", "custom"),
+                Forwarding.policies(parameters(none("Header"), none("Cookie"), none("QueryString")),
+                        allViewerExceptHost, false));
+
+        assertEquals("x=1&encoded=a%2Fb", request.rawQuery());
+        assertNull(value(request, "Host"));
+        assertEquals("curl/8.0", value(request, "User-Agent"));
+        assertEquals("Bearer t", value(request, "Authorization"));
+        assertEquals("custom", value(request, "X-Custom"));
+        assertEquals("csrf=xyz; session=abc123", value(request, "Cookie"));
+        // AWS documents AllViewerExceptHostHeader as adding CloudFront's protocol, version and
+        // location headers; Floci generates the ones it can derive locally.
+        assertEquals("http", value(request, "CloudFront-Forwarded-Proto"));
+        assertEquals("1.1", value(request, "CloudFront-Viewer-Http-Version"));
+        assertEquals("127.0.0.1:54321", value(request, "CloudFront-Viewer-Address"));
+    }
+
+    @Test
+    void allViewerForwardsTheViewerHostAndAddsNoCloudFrontHeaders() {
+        OriginRequest request = build(viewer("GET", null,
+                "Host", "d111111abcdef8.cloudfront.net",
+                "Accept-Language", "en-GB"),
+                Forwarding.policies(null, orp(
+                        selection("HeaderBehavior", "allViewer", "Headers"),
+                        selection("CookieBehavior", "none", "Cookies"),
+                        selection("QueryStringBehavior", "none", "QueryStrings")), false));
+
+        assertEquals("d111111abcdef8.cloudfront.net", value(request, "Host"));
+        assertEquals("en-GB", value(request, "Accept-Language"));
+        assertTrue(names(request).stream().noneMatch(name -> name.startsWith("CloudFront-")),
+                names(request).toString());
+    }
+
+    @Test
+    void allViewerAndWhitelistCloudFrontGeneratesOnlyListedSupportedHeaders() {
+        OriginRequest request = build(viewerOverHttp2("https",
+                "X-Custom", "custom",
+                "CloudFront-Viewer-Country", "XX",
+                "CloudFront-Viewer-Address", "198.51.100.1:1"),
+                Forwarding.policies(null, orp(
+                        selection("HeaderBehavior", "allViewerAndWhitelistCloudFront", "Headers",
+                                "CloudFront-Viewer-Address", "CloudFront-Viewer-Country",
+                                "CloudFront-Forwarded-Proto"),
+                        selection("CookieBehavior", "none", "Cookies"),
+                        selection("QueryStringBehavior", "none", "QueryStrings")), false));
+
+        assertEquals("custom", value(request, "X-Custom"));
+        assertEquals("127.0.0.1:54321", value(request, "CloudFront-Viewer-Address"));
+        assertEquals("https", value(request, "CloudFront-Forwarded-Proto"));
+        assertNull(value(request, "CloudFront-Viewer-Country"), "viewer-supplied values are dropped");
+        assertNull(value(request, "CloudFront-Viewer-Http-Version"), "not listed");
+        assertEquals("2.0 " + VIA_HOST + " (CloudFront)", value(request, "Via"));
+    }
+
+    @Test
+    void whitelistedHeadersMatchCaseInsensitivelyAndUnionWithTheCachePolicy() {
+        OriginRequest request = build(viewer("GET", null,
+                "authorization", "Bearer t",
+                "x-custom", "custom",
+                "X-Other", "other",
+                "Referer", "https://viewer.example/"),
+                Forwarding.policies(
+                        parameters(selection("HeaderBehavior", "whitelist", "Headers", "Authorization"),
+                                none("Cookie"), none("QueryString")),
+                        orp(selection("HeaderBehavior", "whitelist", "Headers", "X-CUSTOM"),
+                                selection("CookieBehavior", "none", "Cookies"),
+                                selection("QueryStringBehavior", "none", "QueryStrings")),
+                        false));
+
+        assertEquals("Bearer t", value(request, "Authorization"));
+        assertEquals("custom", value(request, "X-Custom"));
+        assertNull(value(request, "X-Other"));
+        assertNull(value(request, "Referer"));
+    }
+
+    @Test
+    void cookieBehaviorsSelectSortAndDropMalformedCookies() {
+        ViewerRequest viewer = viewer("GET", null,
+                "Cookie", "session=abc; Session=upper; $Version=1; theme=dark; broken",
+                "Cookie", "csrf=xyz");
+
+        assertNull(value(build(viewer, cookies("none")), "Cookie"));
+        assertEquals("session=abc",
+                value(build(viewer, cookies("whitelist", "session", "missing")), "Cookie"));
+        assertEquals("Session=upper; csrf=xyz; theme=dark",
+                value(build(viewer, cookies("allExcept", "session")), "Cookie"));
+        assertEquals("Session=upper; csrf=xyz; session=abc; theme=dark",
+                value(build(viewer, cookies("all")), "Cookie"));
+    }
+
+    @Test
+    void queryStringBehaviorsKeepViewerOrderAndEncoding() {
+        ViewerRequest viewer = viewer("GET", "b=2&a=1&c=x%20y&flag&a=3", "Host", "d.example");
+
+        assertNull(build(viewer, queryStrings("none")).rawQuery());
+        assertEquals("b=2&a=1&c=x%20y&flag&a=3", build(viewer, queryStrings("all")).rawQuery());
+        assertEquals("a=1&flag&a=3", build(viewer, queryStrings("whitelist", "a", "flag")).rawQuery());
+        assertEquals("b=2&c=x%20y&flag", build(viewer, queryStrings("allExcept", "a")).rawQuery());
+        assertNull(build(viewer, queryStrings("whitelist", "A")).rawQuery(),
+                "query string names are case-sensitive");
+
+        OriginRequest union = build(viewer, Forwarding.policies(
+                parameters(none("Header"), none("Cookie"),
+                        selection("QueryStringBehavior", "whitelist", "QueryStrings", "b")),
+                orp(selection("HeaderBehavior", "none", "Headers"),
+                        selection("CookieBehavior", "none", "Cookies"),
+                        selection("QueryStringBehavior", "whitelist", "QueryStrings", "c")),
+                false));
+        assertEquals("b=2&c=x%20y", union.rawQuery());
+    }
+
+    @Test
+    void legacySettingsForwardDefaultHeadersAndRemoveTheDocumentedOnes() {
+        OriginRequest request = build(viewer("GET", "x=1",
+                "Host", "d111111abcdef8.cloudfront.net",
+                "User-Agent", "curl/8.0",
+                "Accept", "text/html",
+                "Accept-Language", "en-GB",
+                "Referer", "https://viewer.example/",
+                "Authorization", "Bearer t",
+                "Cookie", "session=abc",
+                "X-HTTP-Method-Override", "DELETE",
+                "X-Custom", "custom",
+                "Origin", "https://viewer.example",
+                "If-None-Match", "\"etag\"",
+                "Via", "1.1 proxy.example",
+                "Accept-Encoding", "gzip, deflate"),
+                Forwarding.legacy(null, false));
+
+        assertNull(request.rawQuery());
+        assertEquals("custom", value(request, "X-Custom"));
+        assertEquals("https://viewer.example", value(request, "Origin"));
+        assertEquals("\"etag\"", value(request, "If-None-Match"));
+        assertEquals("gzip", value(request, "Accept-Encoding"));
+        assertEquals("Amazon CloudFront", value(request, "User-Agent"));
+        assertEquals("1.1 proxy.example, 1.1 " + VIA_HOST + " (CloudFront)", value(request, "Via"));
+        for (String removed : List.of("Host", "Accept", "Accept-Language", "Referer", "Authorization",
+                "Cookie", "X-HTTP-Method-Override")) {
+            assertNull(value(request, removed), removed);
+        }
+    }
+
+    @Test
+    void legacyHeaderListsAddRemovedHeadersAndAWildcardForwardsAll() {
+        ViewerRequest viewer = viewer("GET", null,
+                "Host", "d111111abcdef8.cloudfront.net",
+                "User-Agent", "curl/8.0",
+                "Authorization", "Bearer t",
+                "Accept-Language", "en-GB",
+                "Accept-Encoding", "gzip, deflate",
+                "X-Custom", "custom");
+
+        OriginRequest listed = build(viewer, Forwarding.legacy(
+                forwardedValues(false, "none", List.of(), List.of("Authorization", "accept-language")),
+                false));
+        assertEquals("Bearer t", value(listed, "Authorization"));
+        assertEquals("en-GB", value(listed, "Accept-Language"));
+        assertEquals("custom", value(listed, "X-Custom"));
+        assertNull(value(listed, "Host"));
+
+        OriginRequest all = build(viewer, Forwarding.legacy(
+                forwardedValues(false, "none", List.of(), List.of("*")), false));
+        assertEquals("d111111abcdef8.cloudfront.net", value(all, "Host"));
+        assertEquals("curl/8.0", value(all, "User-Agent"));
+        assertEquals("Bearer t", value(all, "Authorization"));
+        assertEquals("gzip, deflate", value(all, "Accept-Encoding"));
+    }
+
+    @Test
+    void legacyCookiesAndQueryStringFollowForwardedValues() {
+        ViewerRequest viewer = viewer("GET", "a=1&b=2",
+                "Cookie", "userid_42=me; userid_=empty; uid9=x; theme=dark");
+
+        OriginRequest wildcard = build(viewer, Forwarding.legacy(
+                forwardedValues(true, "whitelist", List.of("userid_*", "uid?"), List.of()), false));
+        assertEquals("uid9=x; userid_=empty; userid_42=me", value(wildcard, "Cookie"));
+        // QueryString=true forwards every parameter; QueryStringCacheKeys only narrows the cache key.
+        Map<String, Object> keyed = forwardedValues(true, "all", List.of(), List.of());
+        keyed.put("QueryStringCacheKeys", List.of("a"));
+        OriginRequest all = build(viewer, Forwarding.legacy(keyed, false));
+        assertEquals("a=1&b=2", all.rawQuery());
+        assertEquals("theme=dark; uid9=x; userid_=empty; userid_42=me", value(all, "Cookie"));
+
+        OriginRequest none = build(viewer, Forwarding.legacy(
+                forwardedValues(false, "none", List.of(), List.of()), false));
+        assertNull(none.rawQuery());
+        assertNull(value(none, "Cookie"));
+    }
+
+    @Test
+    void authorizationIsAlwaysForwardedForMethodsCloudFrontDoesNotCache() {
+        Forwarding nothing = Forwarding.policies(null, null, false);
+        for (String method : List.of("POST", "PUT", "PATCH", "DELETE", "OPTIONS")) {
+            assertEquals("Bearer t", value(build(viewer(method, null, "Authorization", "Bearer t"),
+                    nothing), "Authorization"), method);
+        }
+        assertNull(value(build(viewer("GET", null, "Authorization", "Bearer t"), nothing),
+                "Authorization"));
+        assertNull(value(build(viewer("HEAD", null, "Authorization", "Bearer t"), nothing),
+                "Authorization"));
+        assertNull(value(build(viewer("OPTIONS", null, "Authorization", "Bearer t"),
+                Forwarding.policies(null, null, true)), "Authorization"),
+                "a cached OPTIONS response drops Authorization");
+    }
+
+    @Test
+    void contentTypeTravelsWithARequestBodyWhateverThePolicy() {
+        Forwarding nothing = Forwarding.policies(null, null, false);
+        assertEquals("application/json", value(build(
+                viewer("POST", null, "Content-Type", "application/json"), nothing), "Content-Type"));
+        assertNull(value(build(
+                viewer("GET", null, "Content-Type", "application/json"), nothing), "Content-Type"));
+    }
+
+    @Test
+    void hopByHopAndCloudFrontOwnedHeadersAreNeverForwarded() {
+        OriginRequest request = build(viewer("POST", null,
+                "Connection", "keep-alive, X-Hop",
+                "X-Hop", "hop",
+                "Keep-Alive", "timeout=5",
+                "Proxy-Authorization", "Basic x",
+                "TE", "trailers",
+                "Upgrade", "websocket",
+                "Transfer-Encoding", "chunked",
+                "Content-Length", "10",
+                "Expect", "100-continue",
+                "X-Amz-Cf-Id", "forged",
+                "X-Edge-Location", "forged",
+                "X-Real-IP", "10.0.0.1",
+                "X-Forwarded-Proto", "https",
+                "X-Kept", "kept"),
+                Forwarding.policies(null, orp(
+                        selection("HeaderBehavior", "allViewer", "Headers"),
+                        selection("CookieBehavior", "none", "Cookies"),
+                        selection("QueryStringBehavior", "none", "QueryStrings")), false));
+
+        assertEquals("kept", value(request, "X-Kept"));
+        for (String dropped : List.of("Connection", "X-Hop", "Keep-Alive", "Proxy-Authorization", "TE",
+                "Upgrade", "Transfer-Encoding", "Content-Length", "Expect", "X-Edge-Location",
+                "X-Real-IP", "X-Forwarded-Proto")) {
+            assertNull(value(request, dropped), dropped);
+        }
+        assertNotEquals("forged", value(request, "X-Amz-Cf-Id"));
+        assertNotEquals(value(request, "X-Amz-Cf-Id"),
+                value(build(viewer("GET", null), Forwarding.policies(null, null, false)), "X-Amz-Cf-Id"));
+    }
+
+    @Test
+    void cachePolicyCompressionSettingsNormalizeAcceptEncoding() {
+        Map<String, Object> both = parameters(none("Header"), none("Cookie"), none("QueryString"));
+        both.put("EnableAcceptEncodingGzip", "true");
+        both.put("EnableAcceptEncodingBrotli", "true");
+        Map<String, Object> gzipOnly = parameters(none("Header"), none("Cookie"), none("QueryString"));
+        gzipOnly.put("EnableAcceptEncodingGzip", "true");
+
+        assertEquals("br,gzip", value(build(viewer("GET", null, "Accept-Encoding", "gzip, deflate, br;q=0.9"),
+                Forwarding.policies(both, null, false)), "Accept-Encoding"));
+        assertEquals("gzip", value(build(viewer("GET", null, "Accept-Encoding", "gzip, br"),
+                Forwarding.policies(gzipOnly, null, false)), "Accept-Encoding"));
+        assertEquals("identity", value(build(viewer("GET", null, "Accept-Encoding", "deflate"),
+                Forwarding.policies(both, null, false)), "Accept-Encoding"));
+        assertNull(value(build(viewer("GET", null), Forwarding.policies(both, null, false)),
+                "Accept-Encoding"));
+        assertEquals("gzip, deflate", value(build(viewer("GET", null, "Accept-Encoding", "gzip, deflate"),
+                Forwarding.policies(null, orp(
+                        selection("HeaderBehavior", "allViewer", "Headers"),
+                        selection("CookieBehavior", "none", "Cookies"),
+                        selection("QueryStringBehavior", "none", "QueryStrings")), false)),
+                "Accept-Encoding"));
+    }
+
+    private static OriginRequest build(ViewerRequest viewer, Forwarding forwarding) {
+        return CloudFrontOriginRequestBuilder.build(viewer, forwarding, VIA_HOST);
+    }
+
+    private static ViewerRequest viewer(String method, String rawQuery, String... headerPairs) {
+        return new ViewerRequest(method, headers(headerPairs), rawQuery, "127.0.0.1", 54321,
+                "http", "1.1");
+    }
+
+    private static ViewerRequest viewerOverHttp2(String scheme, String... headerPairs) {
+        return new ViewerRequest("GET", headers(headerPairs), null, "127.0.0.1", 54321, scheme, "2.0");
+    }
+
+    private static List<Header> headers(String... headerPairs) {
+        List<Header> headers = new ArrayList<>();
+        for (int i = 0; i < headerPairs.length; i += 2) {
+            headers.add(new Header(headerPairs[i], headerPairs[i + 1]));
+        }
+        return headers;
+    }
+
+    private static Forwarding cookies(String behavior, String... names) {
+        return Forwarding.policies(null, orp(
+                selection("HeaderBehavior", "none", "Headers"),
+                selection("CookieBehavior", behavior, "Cookies", names),
+                selection("QueryStringBehavior", "none", "QueryStrings")), false);
+    }
+
+    private static Forwarding queryStrings(String behavior, String... names) {
+        return Forwarding.policies(null, orp(
+                selection("HeaderBehavior", "none", "Headers"),
+                selection("CookieBehavior", "none", "Cookies"),
+                selection("QueryStringBehavior", behavior, "QueryStrings", names)), false);
+    }
+
+    private static Map<String, Object> orp(Map<String, Object> headers, Map<String, Object> cookies,
+                                           Map<String, Object> queryStrings) {
+        Map<String, Object> config = new LinkedHashMap<>();
+        config.put("HeadersConfig", headers);
+        config.put("CookiesConfig", cookies);
+        config.put("QueryStringsConfig", queryStrings);
+        return config;
+    }
+
+    private static Map<String, Object> parameters(Map<String, Object> headers,
+                                                  Map<String, Object> cookies,
+                                                  Map<String, Object> queryStrings) {
+        return orp(headers, cookies, queryStrings);
+    }
+
+    private static Map<String, Object> none(String kind) {
+        return selection(kind + "Behavior", "none", kind + "s");
+    }
+
+    private static Map<String, Object> selection(String behaviorKey, String behavior, String listKey,
+                                                 String... names) {
+        Map<String, Object> block = new LinkedHashMap<>();
+        block.put(behaviorKey, behavior);
+        if (names.length > 0) {
+            block.put(listKey, List.of(names));
+        }
+        return block;
+    }
+
+    private static Map<String, Object> forwardedValues(boolean queryString, String cookiesForward,
+                                                       List<String> cookieNames, List<String> headers) {
+        Map<String, Object> values = new LinkedHashMap<>();
+        values.put("QueryString", queryString);
+        values.put("CookiesForward", cookiesForward);
+        values.put("CookieNames", cookieNames);
+        values.put("Headers", headers);
+        return values;
+    }
+
+    private static List<String> names(OriginRequest request) {
+        return request.headers().stream().map(Header::name).toList();
+    }
+
+    private static String value(OriginRequest request, String name) {
+        List<String> values = request.headers().stream()
+                .filter(header -> header.name().equalsIgnoreCase(name))
+                .map(Header::value)
+                .toList();
+        assertTrue(values.size() <= 1, name + " appears more than once: " + values);
+        return values.isEmpty() ? null : values.getFirst();
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontRequestRouterTest.java

@@ -7,6 +7,7 @@ import io.github.hectorvent.floci.services.cloudfront.model.Origin;
 import org.junit.jupiter.api.Test;
 
 import java.util.List;
+import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -121,6 +122,32 @@ class CloudFrontRequestRouterTest {
         assertNull(CloudFrontRequestRouter.matchResponseHeadersPolicyId(cfg, "/index.html"));
     }
 
+    @Test
+    void matchForwardingReturnsTheSelectedBehaviorsPoliciesAndForwardedValues() {
+        CacheBehavior api = behavior("api/*", "api-origin");
+        api.setCachePolicyId("api-cache-policy");
+        api.setOriginRequestPolicyId("api-origin-request-policy");
+        api.setCachedMethods(List.of("GET", "HEAD", "OPTIONS"));
+        DefaultCacheBehavior dflt = defaultBehavior("default-origin");
+        dflt.setForwardedValues(Map.of("QueryString", true));
+
+        DistributionConfig cfg = new DistributionConfig();
+        cfg.setCacheBehaviors(List.of(api));
+        cfg.setDefaultCacheBehavior(dflt);
+
+        CloudFrontRequestRouter.BehaviorForwarding matched =
+                CloudFrontRequestRouter.matchForwarding(cfg, "/api/data");
+        assertEquals("api-cache-policy", matched.cachePolicyId());
+        assertEquals("api-origin-request-policy", matched.originRequestPolicyId());
+        assertEquals(List.of("GET", "HEAD", "OPTIONS"), matched.cachedMethods());
+        assertNull(matched.forwardedValues());
+
+        CloudFrontRequestRouter.BehaviorForwarding fallback =
+                CloudFrontRequestRouter.matchForwarding(cfg, "/index.html");
+        assertNull(fallback.cachePolicyId());
+        assertEquals(Map.of("QueryString", true), fallback.forwardedValues());
+    }
+
     @Test
     void matchAllowedMethodsUsesTheSelectedBehaviorAndDefaultsToGetHead() {
         CacheBehavior api = behavior("api/*", "api-origin");
@@ -175,7 +202,7 @@ class CloudFrontRequestRouterTest {
         Origin s3 = new Origin();
         s3.setId("s3-origin");
         s3.setDomainName("my-bucket.s3.us-east-1.amazonaws.com");
-        s3.setS3OriginConfig(java.util.Map.of("originAccessIdentity", ""));
+        s3.setS3OriginConfig(Map.of("originAccessIdentity", ""));
         DistributionConfig cfg = new DistributionConfig();
         cfg.setOrigins(List.of(s3));
 

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<String, Object> 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<String, String> getConfigOverrides() {
+            return Map.of(
+                    "floci.storage.services.s3.mode", "persistent",
+                    "floci.storage.persistent-path", "target/cloudfront-s3-disk-origin-it");
+        }
+    }
+}

src/test/java/io/github/hectorvent/floci/services/cloudfront/CloudFrontServiceTest.java

@@ -164,6 +164,54 @@ class CloudFrontServiceTest {
         assertNotEquals(development.getEtag(), live.getEtag());
     }
 
+    @Test
+    void createFunction_unknownRuntime_isRejected() {
+        CloudFrontService service = serviceWithDomainSuffix("cloudfront.net");
+        CloudFrontFunction function = new CloudFrontFunction();
+        function.setName("routing");
+        function.setRuntime("python3");
+
+        AwsException error = assertThrows(AwsException.class, () -> service.createFunction(function));
+
+        assertEquals("InvalidArgument", error.getErrorCode());
+        assertEquals(400, error.getHttpStatus());
+        assertThrows(AwsException.class, () -> service.describeFunction("routing", null));
+    }
+
+    @Test
+    void createFunction_supportedRuntimes_areAccepted() {
+        CloudFrontService service = serviceWithDomainSuffix("cloudfront.net");
+        for (String runtime : List.of("cloudfront-js-1.0", "cloudfront-js-2.0")) {
+            CloudFrontFunction function = new CloudFrontFunction();
+            function.setName("routing-" + runtime);
+            function.setRuntime(runtime);
+
+            assertEquals(runtime, service.createFunction(function).getRuntime());
+        }
+    }
+
+    @Test
+    void updateFunction_unknownRuntime_isRejectedAndKeepsTheFunction() {
+        CloudFrontService service = serviceWithDomainSuffix("cloudfront.net");
+        CloudFrontFunction function = new CloudFrontFunction();
+        function.setName("routing");
+        function.setFunctionCode("first");
+        function.setRuntime("cloudfront-js-2.0");
+        String etag = service.createFunction(function).getEtag();
+        CloudFrontFunction update = new CloudFrontFunction();
+        update.setFunctionCode("second");
+        update.setRuntime("python3");
+
+        AwsException error = assertThrows(AwsException.class,
+                () -> service.updateFunction("routing", etag, update));
+
+        assertEquals("InvalidArgument", error.getErrorCode());
+        CloudFrontFunction kept = service.describeFunction("routing", null);
+        assertEquals("first", kept.getFunctionCode());
+        assertEquals("cloudfront-js-2.0", kept.getRuntime());
+        assertEquals(etag, kept.getEtag());
+    }
+
     @Test
     void listFunctionsWithoutStageReturnsDevelopmentAndLive() {
         CloudFrontService service = serviceWithDomainSuffix("cloudfront.net");
@@ -982,6 +1030,40 @@ class CloudFrontServiceTest {
         assertAws("NoSuchOriginRequestPolicy", () -> service.getOriginRequestPolicy(policy.getId()));
     }
 
+    @Test
+    void exposesTheAwsManagedOriginRequestPoliciesAsImmutable() {
+        CloudFrontService service = serviceWithDomainSuffix("cloudfront.net");
+
+        OriginRequestPolicy exceptHost = service.getOriginRequestPolicy(
+                CloudFrontService.MANAGED_ALL_VIEWER_EXCEPT_HOST_HEADER_ORIGIN_REQUEST_POLICY_ID);
+        assertEquals("Managed-AllViewerExceptHostHeader", exceptHost.getName());
+        assertEquals(Map.of("HeaderBehavior", "allExcept", "Headers", List.of("Host")),
+                exceptHost.getConfig().get("HeadersConfig"));
+        assertEquals(Map.of("CookieBehavior", "all"), exceptHost.getConfig().get("CookiesConfig"));
+        assertEquals(Map.of("QueryStringBehavior", "all"),
+                exceptHost.getConfig().get("QueryStringsConfig"));
+        assertEquals(Map.of("HeaderBehavior", "allViewer"), service.getOriginRequestPolicy(
+                CloudFrontService.MANAGED_ALL_VIEWER_ORIGIN_REQUEST_POLICY_ID)
+                .getConfig().get("HeadersConfig"));
+        assertEquals(Map.of("HeaderBehavior", "whitelist", "Headers", List.of("Host")),
+                service.getOriginRequestPolicy(
+                        CloudFrontService.MANAGED_HOST_HEADER_ONLY_ORIGIN_REQUEST_POLICY_ID)
+                        .getConfig().get("HeadersConfig"));
+
+        OriginRequestPolicy custom =
+                service.createOriginRequestPolicy(namedOriginRequestPolicy("custom-orp"));
+        assertEquals(8, service.listOriginRequestPolicies(null, 100, "managed").size());
+        assertEquals(List.of(custom.getId()), service.listOriginRequestPolicies(null, 100, "custom")
+                .stream().map(OriginRequestPolicy::getId).toList());
+        assertEquals(9, service.listOriginRequestPolicies(null, 100, null).size());
+        assertAws("InvalidArgument", () -> service.listOriginRequestPolicies(null, 100, "MANAGED"));
+
+        assertAws("IllegalUpdate", () -> service.updateOriginRequestPolicy(exceptHost.getId(),
+                exceptHost.getEtag(), namedOriginRequestPolicy("replacement")));
+        assertAws("IllegalDelete",
+                () -> service.deleteOriginRequestPolicy(exceptHost.getId(), exceptHost.getEtag()));
+    }
+
     @Test
     void deletesAnUnattachedCachePolicy() {
         CloudFrontService service = serviceWithDomainSuffix("cloudfront.net");

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();
+        }
+    }
+}