Add logging, and tests

src/main/java/io/fusionauth/http/ParseException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,18 +21,23 @@
  * @author Brian Pontarelli
  */
 public class ParseException extends RuntimeException {
+ private final String state;
+
  public ParseException() {
+ this.state = null;
  }
 
  public ParseException(String message) {
  super(message);
+ this.state = null;
  }
 
- public ParseException(String message, Throwable cause) {
- super(message, cause);
+ public ParseException(String message, String state) {
+ super(message);
+ this.state = state;
  }
 
- public ParseException(Throwable cause) {
- super(cause);
+ public String getState() {
+ return state;
  }
 }

src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java

@@ -95,16 +95,33 @@ public void run() {
  var request = new HTTPRequest(configuration.getContextPath(), configuration.getMultipartBufferSize(),
  listener.getCertificate() != null ? "https" : "http", listener.getPort(), socket.getInetAddress().getHostAddress());
 
+ // Set up the output stream so that if we fail we have the opportunity to write a response that contains a status code.
+ var throughputOutputStream = new ThroughputOutputStream(socket.getOutputStream(), throughput);
+ response = new HTTPResponse();
+
+ HTTPOutputStream outputStream = new HTTPOutputStream(configuration, request.getAcceptEncodings(), response, throughputOutputStream, buffers, () -> state = State.Write);
+ response.setOutputStream(outputStream);
+
  var inputStream = new ThroughputInputStream(socket.getInputStream(), throughput);
  var bodyBytes = HTTPTools.parseRequestPreamble(inputStream, request, buffers.requestBuffer(), instrumenter, () -> state = State.Read);
  var httpInputStream = new HTTPInputStream(configuration, request, inputStream, bodyBytes);
  request.setInputStream(httpInputStream);
 
- var throughputOutputStream = new ThroughputOutputStream(socket.getOutputStream(), throughput);
- response = new HTTPResponse();
+ // Set the Connection response header as soon as possible
+ if (request.isKeepAlive()) {
+ response.setHeader(Headers.Connection, Connections.KeepAlive);
+ keepAlive = true;
+ } else {
+ response.setHeader(Headers.Connection, Connections.Close);
+ keepAlive = false;
+ }
 
- HTTPOutputStream outputStream = new HTTPOutputStream(configuration, request.getAcceptEncodings(), response, throughputOutputStream, buffers, () -> state = State.Write);
- response.setOutputStream(outputStream);
+ // Ensure the preamble is valid
+ Integer status = validatePreamble(request);
+ if (status != null) {
+ close(status, response);
+ return;
+ }
 
  // Handle the "expect" response
  String expect = request.getHeader(Headers.Expect);
@@ -113,22 +130,14 @@ public void run() {
 
  // If the "expect" wasn't accepted, close the socket and exit
  if (!expectContinue(request)) {
- close(Result.Success, response);
+ close(Status.OK, response);
  return;
  }
 
  // Otherwise, transition the state to Read
  state = State.Read;
  }
 
- if (request.isKeepAlive()) {
- response.setHeader(Headers.Connection, Connections.KeepAlive);
- keepAlive = true;
- } else {
- response.setHeader(Headers.Connection, Connections.Close);
- keepAlive = false;
- }
-
  var handler = configuration.getHandler();
  state = State.Process; // Transition to processing
  logger.debug("[{}] Set state [{}]. Call the request handler.", Thread.currentThread().threadId(), state);
@@ -139,7 +148,7 @@ public void run() {
  // Since the Handler can change the Keep-Alive state, we use the response here
  if (!keepAlive) {
  logger.debug("[{}] Close socket. No Keep-Alive", Thread.currentThread().threadId());
- close(Result.Success, response);
+ close(Status.OK, response);
  break;
  }
 
@@ -157,34 +166,39 @@ public void run() {
  }
  } catch (SocketTimeoutException | ConnectionClosedException e) {
  // This might be a read timeout or a Keep-Alive timeout. The failure state is based on that flag
- close(keepAlive ? Result.Success : Result.Failure, response);
+ var status = keepAlive ? Status.OK : Status.InternalServerError; // 200 or 500
+ close(status, response);
 
  if (keepAlive) {
- logger.debug(String.format("[%s] Closing socket. Keep-Alive expired.", Thread.currentThread().threadId()), e);
+ logger.debug(String.format("[%s] Closing socket with status [%d]. Keep-Alive expired.", Thread.currentThread().threadId(), status), e);
  } else {
- logger.debug("[{}] Closing socket. Socket timed out.", Thread.currentThread().threadId());
+ logger.debug("[{}] Closing socket with status [{}]. Socket timed out.", Thread.currentThread().threadId(), status);
  }
  } catch (ParseException pe) {
  if (instrumenter != null) {
  instrumenter.badRequest();
  }
 
- logger.debug("[{}] Closing socket. Bad request. Reason [{}]", Thread.currentThread().threadId(), pe.getMessage());
- close(Result.Failure, response);
+ int status = 400;
+ logger.debug("[{}] Closing socket with status [{}]. Bad request, failed to parse request. Reason [{}] Parser state [{}]", Thread.currentThread().threadId(), status, pe.getMessage(), pe.getState());
+ close(status, response);
  } catch (SocketException e) {
  // This should only happen when the server is shutdown and this thread is waiting to read or write. In that case, this will throw a
  // SocketException and the thread will be interrupted. Since the server is being shutdown, we should let the client know.
  if (Thread.currentThread().isInterrupted()) {
- logger.debug("[{}] Closing socket. Server is shutting down.", Thread.currentThread().threadId());
- close(Result.Success, response);
+ var status = Status.OK;
+ logger.debug("[{}] Closing socket with status [{}]. Server is shutting down.", Thread.currentThread().threadId(), status);
+ close(status, response);
  }
  } catch (IOException io) {
- logger.debug(String.format("[%s] Closing socket. An IO exception was thrown during processing. These are pretty common.", Thread.currentThread().threadId()), io);
- close(Result.Failure, response);
+ var status = Status.InternalServerError;
+ logger.debug(String.format("[%s] Closing socket with status [%d]. An IO exception was thrown during processing. These are pretty common.", Thread.currentThread().threadId(), status), io);
+ close(status, response);
  } catch (Throwable t) {
  // Log the error and signal a failure
- logger.error(String.format("[%s] Closing socket. An HTTP worker threw an exception while processing a request.", Thread.currentThread().threadId()), t);
- close(Result.Failure, response);
+ var status = Status.InternalServerError;
+ logger.error(String.format("[%s] Closing socket with status [%d]. An HTTP worker threw an exception while processing a request.", Thread.currentThread().threadId(), status), t);
+ close(status, response);
  } finally {
  if (instrumenter != null) {
  instrumenter.threadExited();
@@ -196,16 +210,21 @@ public State state() {
  return state;
  }
 
- private void close(Result result, HTTPResponse response) {
- if (result == Result.Failure && instrumenter != null) {
+ private void close(int status, HTTPResponse response) {
+ boolean error = status != 200;
+ if (error && instrumenter != null) {
  instrumenter.connectionClosed();
  }
 
  try {
  // If the conditions are perfect, we can still write back a 500
- if (result == Result.Failure && response != null && !response.isCommitted()) {
- response.reset();
- response.setStatus(500);
+ if (error && response != null && !response.isCommitted()) {
+ // Note that reset() clears the Connection response header.
+// response.reset();
+ response.setStatus(status);
+ // TODO: Daniel : Should we always be sending back a Connection: close on an error such as this?
+ // - I think this is correct since we are calling socket.close() below?
+ response.setHeader(Headers.Connection, Connections.Close);
  response.setContentLength(0L);
  response.close();
  }
@@ -231,15 +250,71 @@ private boolean expectContinue(HTTPRequest request) throws IOException {
  return expectResponse.getStatus() == 100;
  }
 
+ private Integer validatePreamble(HTTPRequest request) {
+ // Validate protocol. Only HTTP/1.1 is supported
+ String protocol = request.getProtocol();
+ if (protocol == null) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Invalid request. Missing HTTP Protocol");
+ }
+ return Status.BadRequest;
+ }
+
+ if (!protocol.startsWith("HTTP/")) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Invalid request. Invalid protocol [{}]. Supported versions [HTTP/1.1].", protocol);
+ }
+ return Status.BadRequest;
+ }
+
+ if (!"HTTP/1.1".equals(protocol)) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Invalid request. Unsupported HTTP version [{}]. Supported versions [HTTP/1.1].", protocol);
+ }
+ return Status.HTTPVersionNotSupported;
+ }
+
+ // Host header is required
+ var host = request.getRawHost();
+ if (host == null) {
+ logger.trace("Invalid request. Missing Host header.");
+ return Status.BadRequest;
+ }
+
+ var hostHeaders = request.getHeaders(Headers.Host);
+ if (hostHeaders.size() != 1) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Invalid request. Duplicate Host headers. [{}]", String.join(", ", hostHeaders));
+ }
+ return Status.BadRequest;
+ }
+
+ // Content-Length cannot be negative
+ var contentLength = request.getContentLength();
+ if (contentLength != null && contentLength < 0) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Invalid request. The Content-Length cannot be negative. [{}]", contentLength);
+ }
+ return Status.BadRequest;
+ }
+
+ return null;
+ }
+
  public enum State {
  Read,
  Process,
  Write,
  KeepAlive
  }
 
- private enum Result {
- Failure,
- Success
+ private static class Status {
+ public static final int BadRequest = 400;
+
+ public static final int HTTPVersionNotSupported = 505;
+
+ public static final int InternalServerError = 500;
+
+ public static final int OK = 200;
  }
 }

src/main/java/io/fusionauth/http/util/RequestPreambleState.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -98,7 +98,9 @@ public boolean store() {
  RequestProtocol {
  @Override
  public RequestPreambleState next(byte ch) {
- if (ch == 'H' || ch == 'T' || ch == 'P' || ch == '/' || ch == '1' || ch == '.') {
+ // While this server only supports HTTP/1.1, allow the request protocol to be parsed for any valid version.
+ // - The supported version will be validated elsewhere.
+ if (ch == 'H' || ch == 'T' || ch == 'P' || ch == '/' || ch == '.' || (ch >= '0' && ch <= '9')) {
  return RequestProtocol;
  } else if (ch == '\r') {
  return RequestCR;
@@ -268,6 +270,6 @@ public boolean store() {
  public abstract boolean store();
 
  ParseException makeParseException(char ch) {
- return new ParseException("Invalid character [" + ch + "] in state [" + this + "]");
+ return new ParseException("Invalid character [" + ch + "] in state [" + this + "]", this.name());
  }
 }

src/test/java/io/fusionauth/http/BaseTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -325,8 +325,19 @@ public void sendBadRequest(String message) {
 
  // Sockets are pretty resilient, so this will be closed by the server, but we'll just see that close are zero bytes read. If we were
  // to continue writing above, then that likely would throw an exception because the pipe would be broken
+
+ // TODO : Daniel :
+ // In many cases we still return a 400 to the client when a request is invalid. So there will be bytes to read.
+ // We could require the caller to provide and expect status code, and or response body.
+ // Currently, this is just used by one test, so I have hard coded the expected response.
+
  byte[] buffer = is.readAllBytes();
- assertEquals(buffer.length, 0);
+ assertEquals(new String(buffer), """
+ HTTP/1.1 400 \r
+ connection: close\r
+ content-length: 0\r
+ \r
+ """);
  } catch (Exception e) {
  fail(e.getMessage());
  }
@@ -392,12 +403,12 @@ public void onTestFailure(ITestResult result) {
  }
 
  System.out.println("""
-
+ 
  Test failure
  -----------------
  Exception: {{exception}}
  Message: {{message}}
-
+ 
  Stack traces (client side):
  {{threadDump}}
  -----------------

src/test/java/io/fusionauth/http/CoreTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -99,11 +99,14 @@ public void badPreambleButReset() throws Exception {
 
  var instrumenter = new CountingInstrumenter();
  try (var client = HttpClient.newHttpClient(); var ignore = makeServer("http", handler, instrumenter).start()) {
+ // Invalid request, missing Host header
+ // - This should cause the socket to be reset
  sendBadRequest("""
  GET / HTTP/1.1\r
  X-Bad-Header: Bad-Header\r\r
  """);
 
+ // TODO : Daniel : Review : What is this test doing? If I comment out reset() in HTTPWorker.close() the test still passes.
  URI uri = makeURI("http", "");
  HttpRequest request = HttpRequest.newBuilder()
  .uri(uri)