WIP

runtime/binding-http/src/main/java/io/aklivity/zilla/runtime/binding/http/internal/config/HttpBindingConfig.java

@@ -34,6 +34,8 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import org.agrona.DirectBuffer;
+
 import io.aklivity.zilla.runtime.binding.http.config.HttpAccessControlConfig;
 import io.aklivity.zilla.runtime.binding.http.config.HttpCredentialsConfig;
 import io.aklivity.zilla.runtime.binding.http.config.HttpOptionsConfig;
@@ -328,24 +330,22 @@ private Map<String, String8FW> parseQueryParams(
         return queryParams;
     }
 
-    public boolean validate(
+    public boolean validateHeader(
         HttpRequestType request,
         HttpBeginExFW beginEx)
     {
         boolean isValid = true;
         if (request != null)
         {
-            boolean isValidHeader = validateHeaders(request, beginEx);
+            boolean isValidHeader = validateHeaderValues(request, beginEx);
             boolean isValidPathParams = validatePathParams(request);
             boolean isValidQueryParams = validateQueryParams(request);
-            // TODO: Ati - validate content (probably somewhere else)
-            //boolean isValidContent = validateContent(requestType, beginEx);
-            isValid = isValidHeader && isValidPathParams && isValidQueryParams; // && isValidContent;
+            isValid = isValidHeader && isValidPathParams && isValidQueryParams;
         }
         return isValid;
     }
 
-    private boolean validateHeaders(
+    private boolean validateHeaderValues(
         HttpRequestType request,
         HttpBeginExFW beginEx)
     {
@@ -411,16 +411,19 @@ private boolean validateQueryParams(
         return isValid;
     }
 
-    private boolean validateContent(
-        HttpRequestType request)
+    public boolean validateContent(
+        HttpRequestType request,
+        DirectBuffer buffer,
+        int index,
+        int length)
     {
-        AtomicBoolean isValidContent = new AtomicBoolean(true);
+        boolean isValid = true;
         if (request != null && request.content != null)
         {
             Validator validator = request.content;
-            // TODO: Ati
+            isValid = validator.read(buffer, index, length);
         }
-        return isValidContent.get();
+        return isValid;
     }
 
     private static Function<Function<String, String>, String> orElseIfNull(

runtime/binding-http/src/main/java/io/aklivity/zilla/runtime/binding/http/internal/stream/HttpServerFactory.java

@@ -406,6 +406,7 @@ public final class HttpServerFactory implements HttpStreamFactory
     private final Array32FW<HttpHeaderFW> headers400;
     private final Array32FW<HttpHeaderFW> headers403;
     private final Array32FW<HttpHeaderFW> headers404;
+    private final DirectBuffer response400;
     private final DirectBuffer response403;
     private final DirectBuffer response404;
 
@@ -582,6 +583,7 @@ public HttpServerFactory(
         this.headers400 = initHeadersEmpty(config, STATUS_400);
         this.headers403 = initHeaders(config, STATUS_403);
         this.headers404 = initHeadersEmpty(config, STATUS_404);
+        this.response400 = initResponse(config, 400, "Bad Request");
         this.response403 = initResponse(config, 403, "Forbidden");
         this.response404 = initResponse(config, 404, "Not Found");
     }
@@ -1059,8 +1061,7 @@ else if (!isCorsRequestAllowed(server.binding, headers))
                         }
 
                         HttpRouteConfig route = binding.resolve(exchangeAuth, headers::get);
-                        server.request = binding.resolveRequest(headers::get);
-                        if (route != null && binding.validate(server.request, beginEx))
+                        if (route != null)
                         {
                             if (binding.options != null && binding.options.overrides != null)
                             {
@@ -1075,7 +1076,9 @@ else if (!isCorsRequestAllowed(server.binding, headers))
                             HttpPolicyConfig policy = binding.access().effectivePolicy(headers);
                             final String origin = policy == CROSS_ORIGIN ? headers.get(HEADER_NAME_ORIGIN) : null;
 
-                            server.onDecodeHeaders(server.routedId, route.id, traceId, exchangeAuth, policy, origin, beginEx);
+                            server.request = binding.resolveRequest(headers::get);
+                            error = server.onDecodeHeaders(server.routedId, route.id, traceId, exchangeAuth, policy, origin,
+                                beginEx);
                         }
                         else
                         {
@@ -2215,7 +2218,7 @@ private void onDecodeCorsPreflight(
             assert exchange == null;
         }
 
-        private void onDecodeHeaders(
+        private DirectBuffer onDecodeHeaders(
             long originId,
             long routedId,
             long traceId,
@@ -2224,14 +2227,24 @@ private void onDecodeHeaders(
             String origin,
             HttpBeginExFW beginEx)
         {
-            final HttpExchange exchange = new HttpExchange(originId, routedId, authorization, traceId, policy, origin);
-            exchange.doRequestBegin(traceId, beginEx);
-            exchange.doResponseWindow(traceId);
+            boolean isValid = binding.validateHeader(request, beginEx);
+            DirectBuffer error = null;
+            if (isValid)
+            {
+                final HttpExchange exchange = new HttpExchange(originId, routedId, authorization, traceId, policy, origin);
+                exchange.doRequestBegin(traceId, beginEx);
+                exchange.doResponseWindow(traceId);
 
-            final HttpHeaderFW connection = beginEx.headers().matchFirst(h -> HEADER_CONNECTION.equals(h.name()));
-            exchange.responseClosing = connection != null && connectionClose.reset(connection.value().asString()).matches();
+                final HttpHeaderFW connection = beginEx.headers().matchFirst(h -> HEADER_CONNECTION.equals(h.name()));
+                exchange.responseClosing = connection != null && connectionClose.reset(connection.value().asString()).matches();
 
-            this.exchange = exchange;
+                this.exchange = exchange;
+            }
+            else
+            {
+                error = response400;
+            }
+            return error;
         }
 
         private void onDecodeHeadersOnly(
@@ -2254,7 +2267,18 @@ private int onDecodeBody(
             int limit,
             Flyweight extension)
         {
-            return exchange.doRequestData(traceId, budgetId, buffer, offset, limit, extension);
+            boolean isValid = binding.validateContent(request, buffer, 0, limit - offset);
+            int result = 0;
+            if (isValid)
+            {
+                result = exchange.doRequestData(traceId, budgetId, buffer, offset, limit, extension);
+            }
+            else
+            {
+                doEncodeHeaders(exchange, traceId, authorization, budgetId, headers400);
+                exchange.doRequestAbort(traceId, EMPTY_OCTETS);
+            }
+            return result;
         }
 
         private void onDecodeTrailers(
@@ -3729,7 +3753,6 @@ private final class Http2Server
         private int decodedStreamId;
         private byte decodedFlags;
         private int decodableDataBytes;
-        private HttpRequestType request;
 
         private Http2Server(
             HttpBindingConfig binding,
@@ -4844,10 +4867,6 @@ else if (!isCorsRequestAllowed(binding, headers))
                             HttpPolicyConfig policy = binding.access().effectivePolicy(headers);
                             final String origin = policy == CROSS_ORIGIN ? headers.get(HEADER_NAME_ORIGIN) : null;
 
-                            final Http2Exchange exchange =
-                                    new Http2Exchange(originId, routedId, NO_REQUEST_ID, streamId, exchangeAuth,
-                                        traceId, policy, origin, contentLength);
-
                             if (binding.options != null && binding.options.overrides != null)
                             {
                                 binding.options.overrides.forEach((k, v) -> headers.put(k.asString(), v.asString()));
@@ -4858,8 +4877,12 @@ else if (!isCorsRequestAllowed(binding, headers))
                                     .headers(hs -> headers.forEach((n, v) -> hs.item(h -> h.name(n).value(v))))
                                     .build();
 
-                            this.request = binding.resolveRequest(headers::get);
-                            boolean isValid =  binding.validate(request, beginEx);
+                            HttpRequestType request = binding.resolveRequest(headers::get);
+
+                            final Http2Exchange exchange = new Http2Exchange(originId, routedId, NO_REQUEST_ID, streamId,
+                                exchangeAuth, traceId, policy, origin, contentLength, request);
+
+                            boolean isValid = binding.validateHeader(request, beginEx);
                             if (isValid)
                             {
                                 exchange.doRequestBegin(traceId, beginEx);
@@ -4870,7 +4893,7 @@ else if (!isCorsRequestAllowed(binding, headers))
                             }
                             else
                             {
-                                doEncodeHeaders(traceId, authorization, streamId, headers404, true);
+                                doEncodeHeaders(traceId, authorization, streamId, headers400, true);
                             }
                         }
                     }
@@ -5076,6 +5099,12 @@ private int onDecodeData(
 
                     if (payloadLength > 0)
                     {
+                        boolean isValid = binding.validateContent(exchange.request, payload, 0, payloadLength);
+                        if (!isValid)
+                        {
+                            doEncodeHeaders(traceId, authorization, streamId, headers400, true);
+                            exchange.doRequestAbort(traceId, EMPTY_OCTETS);
+                        }
                         payloadRemaining.set(payloadLength);
                         exchange.doRequestData(traceId, payload, payloadRemaining);
                         progress += payloadLength - payloadRemaining.value;
@@ -5225,7 +5254,7 @@ private void doEncodePromise(
                     doEncodePushPromise(traceId, authorization, pushId, promiseId, promise);
 
                     final Http2Exchange exchange = new Http2Exchange(originId, routedId, requestId, promiseId,
-                                exchangeAuth, traceId, policy, origin, contentLength);
+                                exchangeAuth, traceId, policy, origin, contentLength, null);
 
                     final HttpBeginExFW beginEx = beginExRW.wrap(extBuffer, 0, extBuffer.capacity())
                             .typeId(httpTypeId)
@@ -5541,6 +5570,8 @@ private final class Http2Exchange
             private long responseAck;
             private int responseMax;
 
+            private final HttpRequestType request;
+
             private Http2Exchange(
                 long originId,
                 long routedId,
@@ -5550,7 +5581,8 @@ private Http2Exchange(
                 long traceId,
                 HttpPolicyConfig policy,
                 String origin,
-                long requestContentLength)
+                long requestContentLength,
+                HttpRequestType request)
             {
                 this.originId = originId;
                 this.routedId = routedId;
@@ -5562,6 +5594,7 @@ private Http2Exchange(
                 this.requestId = requestId == NO_REQUEST_ID ? supplyInitialId.applyAsLong(routedId) : requestId;
                 this.responseId = supplyReplyId.applyAsLong(this.requestId);
                 this.expiringId = expireIfNecessary(guard, sessionId, originId, routedId, replyId, traceId, streamId);
+                this.request = request;
             }
 
             private int initialWindow()

specs/binding-http.spec/src/main/scripts/io/aklivity/zilla/specs/binding/http/streams/network/rfc7230/validation/invalid/client.rpt

@@ -24,7 +24,7 @@ write "POST /valid/1234567890123/123 HTTP/1.1" "\r\n"
 write "Host: localhost:8080" "\r\n"
 write "\r\n"
 
-read "HTTP/1.1 404 Not Found\r\n"
+read "HTTP/1.1 400 Bad Request\r\n"
 read "Connection: close\r\n"
 read "\r\n"
 read closed
@@ -39,7 +39,7 @@ write "POST /valid/1234567890123/1234567890123?page=123 HTTP/1.1" "\r\n"
 write "Host: localhost:8080" "\r\n"
 write "\r\n"
 
-read "HTTP/1.1 404 Not Found\r\n"
+read "HTTP/1.1 400 Bad Request\r\n"
 read "Connection: close\r\n"
 read "\r\n"
 read closed
@@ -55,7 +55,7 @@ write "Host: localhost:8080" "\r\n"
 write "Code: 123" "\r\n"
 write "\r\n"
 
-read "HTTP/1.1 404 Not Found\r\n"
+read "HTTP/1.1 400 Bad Request\r\n"
 read "Connection: close\r\n"
 read "\r\n"
 read closed
@@ -74,7 +74,7 @@ write "Content-Type: text/plain" "\r\n"
 write "\r\n"
 write "123"
 
-read "HTTP/1.1 404 Not Found\r\n"
+read "HTTP/1.1 400 Bad Request\r\n"
 read "Connection: close\r\n"
 read "\r\n"
 read closed

specs/binding-http.spec/src/main/scripts/io/aklivity/zilla/specs/binding/http/streams/network/rfc7230/validation/invalid/server.rpt

@@ -27,7 +27,7 @@ read "POST /valid/1234567890123/123 HTTP/1.1" "\r\n"
 read "Host: localhost:8080" "\r\n"
 read "\r\n"
 
-write "HTTP/1.1 404 Not Found\r\n"
+write "HTTP/1.1 400 Bad Request\r\n"
 write "Connection: close\r\n"
 write "\r\n"
 write close
@@ -40,7 +40,7 @@ read "POST /valid/1234567890123/1234567890123?page=123 HTTP/1.1" "\r\n"
 read "Host: localhost:8080" "\r\n"
 read "\r\n"
 
-write "HTTP/1.1 404 Not Found\r\n"
+write "HTTP/1.1 400 Bad Request\r\n"
 write "Connection: close\r\n"
 write "\r\n"
 write close
@@ -54,7 +54,7 @@ read "Host: localhost:8080" "\r\n"
 read "Code: 123" "\r\n"
 read "\r\n"
 
-write "HTTP/1.1 404 Not Found\r\n"
+write "HTTP/1.1 400 Bad Request\r\n"
 write "Connection: close\r\n"
 write "\r\n"
 write close
@@ -71,7 +71,7 @@ read "Content-Type: text/plain" "\r\n"
 read "\r\n"
 read "123"
 
-write "HTTP/1.1 404 Not Found\r\n"
+write "HTTP/1.1 400 Bad Request\r\n"
 write "Connection: close\r\n"
 write "\r\n"
 write close