Introduce support for pcap traffic capturing of HTTP traffic

Author: geoand

src/main/generated/io/vertx/core/http/HttpClientOptionsConverter.java

@@ -140,6 +140,11 @@ static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, HttpCli
             obj.setName((String)member.getValue());
           }
           break;
+        case "pcapCaptureFile":
+          if (member.getValue() instanceof String) {
+            obj.setPcapCaptureFile((String)member.getValue());
+          }
+          break;
         case "pipelining":
           if (member.getValue() instanceof Boolean) {
             obj.setPipelining((Boolean)member.getValue());
@@ -264,6 +269,9 @@ static void toJson(HttpClientOptions obj, java.util.Map<String, Object> json) {
     if (obj.getName() != null) {
       json.put("name", obj.getName());
     }
+    if (obj.getPcapCaptureFile() != null) {
+      json.put("pcapCaptureFile", obj.getPcapCaptureFile());
+    }
     json.put("pipelining", obj.isPipelining());
     json.put("pipeliningLimit", obj.getPipeliningLimit());
     json.put("poolCleanerPeriod", obj.getPoolCleanerPeriod());

src/main/generated/io/vertx/core/http/HttpServerOptionsConverter.java

@@ -100,6 +100,11 @@ static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, HttpSer
             obj.setMaxWebSocketMessageSize(((Number)member.getValue()).intValue());
           }
           break;
+        case "pcapCaptureFile":
+          if (member.getValue() instanceof String) {
+            obj.setPcapCaptureFile((String)member.getValue());
+          }
+          break;
         case "perFrameWebSocketCompressionSupported":
           if (member.getValue() instanceof Boolean) {
             obj.setPerFrameWebSocketCompressionSupported((Boolean)member.getValue());
@@ -175,6 +180,9 @@ static void toJson(HttpServerOptions obj, java.util.Map<String, Object> json) {
     json.put("maxInitialLineLength", obj.getMaxInitialLineLength());
     json.put("maxWebSocketFrameSize", obj.getMaxWebSocketFrameSize());
     json.put("maxWebSocketMessageSize", obj.getMaxWebSocketMessageSize());
+    if (obj.getPcapCaptureFile() != null) {
+      json.put("pcapCaptureFile", obj.getPcapCaptureFile());
+    }
     json.put("perFrameWebSocketCompressionSupported", obj.getPerFrameWebSocketCompressionSupported());
     json.put("perMessageWebSocketCompressionSupported", obj.getPerMessageWebSocketCompressionSupported());
     if (obj.getTracingPolicy() != null) {

src/main/java/io/vertx/core/http/HttpClientOptions.java

@@ -224,6 +224,11 @@ public class HttpClientOptions extends ClientOptionsBase {
    */
   public static final String DEFAULT_NAME = "__vertx.DEFAULT";
 
+  /**
+   * Default pcap file name where capturing of HTTP traffic is recorded. When no file is set, capturing is disabled.
+   */
+  public static final String DEFAULT_PCAP_CAPTURE_FILE = "";
+
   private boolean verifyHost = true;
   private int maxPoolSize;
   private boolean keepAlive;
@@ -267,6 +272,7 @@ public class HttpClientOptions extends ClientOptionsBase {
 
   private boolean shared;
   private String name;
+  private String pcapCaptureFile;
 
   /**
    * Default constructor
@@ -322,6 +328,7 @@ public HttpClientOptions(HttpClientOptions other) {
     this.tracingPolicy = other.tracingPolicy;
     this.shared = other.shared;
     this.name = other.name;
+    this.pcapCaptureFile = other.pcapCaptureFile;
   }
 
   /**
@@ -386,6 +393,7 @@ private void init() {
     tracingPolicy = DEFAULT_TRACING_POLICY;
     shared = DEFAULT_SHARED;
     name = DEFAULT_NAME;
+    pcapCaptureFile = DEFAULT_PCAP_CAPTURE_FILE;
   }
 
   @Override
@@ -1434,4 +1442,16 @@ public HttpClientOptions setName(String name) {
     this.name = name;
     return this;
   }
+
+  /**
+   * @return the name of the PCAP file where the HTTP traffic will be written to.
+   */
+  public String getPcapCaptureFile() {
+    return pcapCaptureFile;
+  }
+
+  public HttpClientOptions setPcapCaptureFile(String pcapCaptureFile) {
+    this.pcapCaptureFile = pcapCaptureFile;
+    return this;
+  }
 }

src/main/java/io/vertx/core/http/HttpServerOptions.java

@@ -146,6 +146,11 @@ public class HttpServerOptions extends NetServerOptions {
    */
   public static final TracingPolicy DEFAULT_TRACING_POLICY = TracingPolicy.ALWAYS;
 
+  /**
+   * Default pcap file name where capturing of HTTP traffic is recorded. When no file is set, capturing is disabled.
+   */
+  public static final String DEFAULT_PCAP_CAPTURE_FILE = "";
+
   private boolean compressionSupported;
   private int compressionLevel;
   private int maxWebSocketFrameSize;
@@ -169,6 +174,7 @@ public class HttpServerOptions extends NetServerOptions {
   private boolean webSocketPreferredClientNoContext;
   private int webSocketClosingTimeout;
   private TracingPolicy tracingPolicy;
+  private String pcapCaptureFile;
 
   /**
    * Default constructor
@@ -209,6 +215,7 @@ public HttpServerOptions(HttpServerOptions other) {
     this.webSocketAllowServerNoContext = other.webSocketAllowServerNoContext;
     this.webSocketClosingTimeout = other.webSocketClosingTimeout;
     this.tracingPolicy = other.tracingPolicy;
+    this.pcapCaptureFile = other.pcapCaptureFile;
   }
 
   /**
@@ -257,6 +264,7 @@ private void init() {
     webSocketAllowServerNoContext = DEFAULT_WEBSOCKET_ALLOW_SERVER_NO_CONTEXT;
     webSocketClosingTimeout = DEFAULT_WEBSOCKET_CLOSING_TIMEOUT;
     tracingPolicy = DEFAULT_TRACING_POLICY;
+    pcapCaptureFile = DEFAULT_PCAP_CAPTURE_FILE;
   }
 
   @Override
@@ -977,4 +985,16 @@ public HttpServerOptions setTracingPolicy(TracingPolicy tracingPolicy) {
     this.tracingPolicy = tracingPolicy;
     return this;
   }
+
+  /**
+   * @return the name of the PCAP file where the HTTP traffic will be written to.
+   */
+  public String getPcapCaptureFile() {
+    return pcapCaptureFile;
+  }
+
+  public HttpServerOptions setPcapCaptureFile(String pcapCaptureFile) {
+    this.pcapCaptureFile = pcapCaptureFile;
+    return this;
+  }
 }

src/main/java/io/vertx/core/http/impl/HttpChannelConnector.java

@@ -32,7 +32,6 @@
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.net.impl.NetClientImpl;
 import io.vertx.core.net.impl.NetSocketImpl;
-import io.vertx.core.net.impl.NetSocketInternal;
 import io.vertx.core.net.impl.VertxHandler;
 import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.core.spi.metrics.HttpClientMetrics;
@@ -58,6 +57,7 @@ public class HttpChannelConnector {
   private final HttpVersion version;
   private final SocketAddress peerAddress;
   private final SocketAddress server;
+  private final boolean enablePcapCapture;
 
   public HttpChannelConnector(HttpClientImpl client,
                               NetClientImpl netClient,
@@ -78,6 +78,7 @@ public HttpChannelConnector(HttpClientImpl client,
     this.version = version;
     this.peerAddress = peerAddress;
     this.server = server;
+    this.enablePcapCapture = (options.getPcapCaptureFile() != null) && !options.getPcapCaptureFile().isEmpty();
   }
 
   public SocketAddress server() {
@@ -165,6 +166,9 @@ private void applyHttp1xConnectionOptions(ChannelPipeline pipeline) {
     if (options.getLogActivity()) {
       pipeline.addLast("logging", new LoggingHandler());
     }
+    if (enablePcapCapture) {
+      pipeline.addLast("pcapCapturing", new VertxPcapWriteHandler(options.getPcapCaptureFile()));
+    }
     pipeline.addLast("codec", new HttpClientCodec(
       options.getMaxInitialLineLength(),
       options.getMaxHeaderSize(),

src/main/java/io/vertx/core/http/impl/HttpServerWorker.java

@@ -35,6 +35,10 @@
 import io.vertx.core.net.impl.HAProxyMessageCompletionHandler;
 import io.vertx.core.spi.metrics.HttpServerMetrics;
 
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.UncheckedIOException;
 import java.nio.charset.StandardCharsets;
 import java.util.function.Supplier;
 
@@ -55,6 +59,7 @@ public class HttpServerWorker implements Handler<Channel> {
   private final String serverOrigin;
   private final boolean logEnabled;
   private final boolean disableH2C;
+  private final boolean enablePcapCapture;
   final Handler<HttpServerConnection> connectionHandler;
   private final Handler<Throwable> exceptionHandler;
 
@@ -77,6 +82,7 @@ public HttpServerWorker(EventLoopContext context,
     this.serverOrigin = serverOrigin;
     this.logEnabled = options.getLogActivity();
     this.disableH2C = disableH2C;
+    this.enablePcapCapture = (options.getPcapCaptureFile() != null) && !options.getPcapCaptureFile().isEmpty();
     this.connectionHandler = connectionHandler;
     this.exceptionHandler = exceptionHandler;
   }
@@ -252,6 +258,9 @@ private void configureHttp1OrH2C(ChannelPipeline pipeline) {
     if (logEnabled) {
       pipeline.addLast("logging", new LoggingHandler());
     }
+    if (enablePcapCapture) {
+      pipeline.addLast("pcapCapturing", new VertxPcapWriteHandler(options.getPcapCaptureFile()));
+    }
     if (HttpServerImpl.USE_FLASH_POLICY_HANDLER) {
       pipeline.addLast("flashpolicy", new FlashPolicyHandler());
     }

src/main/java/io/vertx/core/http/impl/VertxPcapWriteHandler.java

@@ -0,0 +1,119 @@
+package io.vertx.core.http.impl;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.pcap.PcapWriteHandler;
+import io.vertx.core.impl.logging.Logger;
+import io.vertx.core.impl.logging.LoggerFactory;
+import java.io.Closeable;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A handler that simply delegates to the built in {@link PcapWriteHandler}.
+ * Vert.x needs this because the handler might not have been added to the processing pipeline
+ * when the {@code channelRead} method is invoked, and thus the necessary setup is performed
+ * in the {@code channelRegistered} method.
+ * Furthermore, we want to support capturing the output of multiple Netty pipelines into a single file,
+ * so for example both the output of an HTTP server and an HTTP Client can be inspected via the same file.
+ */
+public class VertxPcapWriteHandler extends ChannelDuplexHandler implements Closeable {
+
+  private static final Logger log = LoggerFactory.getLogger(VertxPcapWriteHandler.class);
+
+  /**
+   * The idea of this map is to control the usage of each throughout the entire Vert.x application.
+   * When the same file is configured for multiple pipelines, we want each pipeline to write to the same
+   * OutputStream, but we only want to close it when the last pipeline has been closed.
+   */
+  private static final ConcurrentMap<String, Metadata> fileToMetadata = new ConcurrentHashMap<>();
+
+  private final PcapWriteHandler delegate;
+  private final String pcapCaptureFile;
+
+  public VertxPcapWriteHandler(String pcapCaptureFile) {
+    this.pcapCaptureFile = pcapCaptureFile;
+    Metadata metadata = fileToMetadata.computeIfAbsent(pcapCaptureFile, Metadata::new);
+    // pcap contains a global header section that should only be written by the first handler
+    int openedCount = metadata.openedCount.getAndIncrement();
+    this.delegate = new PcapWriteHandler(metadata.outputStream, false, openedCount == 0);
+  }
+
+  @Override
+  public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+    delegate.channelActive(ctx);
+  }
+
+  @Override
+  public void channelActive(ChannelHandlerContext ctx) throws Exception {
+    delegate.channelActive(ctx);
+  }
+
+  @Override
+  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+    delegate.channelRead(ctx, msg);
+  }
+
+  @Override
+  public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+    delegate.write(ctx, msg, promise);
+  }
+
+  @Override
+  public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+    int openedCount = fileToMetadata.get(pcapCaptureFile).openedCount.decrementAndGet();
+    if (openedCount == 0) {
+      delegate.handlerRemoved(ctx);
+    }
+  }
+
+  @Override
+  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+    delegate.exceptionCaught(ctx, cause);
+  }
+
+  @Override
+  public void close() throws IOException {
+    delegate.close();
+  }
+
+  private static class Metadata {
+
+    final AtomicInteger openedCount;
+    final OutputStream outputStream;
+
+    Metadata(String pcapFile) {
+      openedCount = new AtomicInteger(0);
+      outputStream = getOutputStream(pcapFile);
+    }
+
+    private OutputStream getOutputStream(String pcapFile) {
+      try {
+        return new FileOutputStream(pcapFile);
+      } catch (FileNotFoundException e) {
+        log.warn("Unable to open capture file for writing, so no capture information will be recorded.", e);
+        return NullOutputStream.INSTANCE;
+      }
+    }
+
+    private static class NullOutputStream extends OutputStream {
+
+      static final NullOutputStream INSTANCE = new NullOutputStream();
+
+      private NullOutputStream() {
+      }
+
+      @Override
+      public void write(int b) throws IOException {
+
+      }
+    }
+  }
+
+}