Introduce support for pcap traffic capturing of network traffic

Author: geoand

src/main/generated/io/vertx/core/eventbus/EventBusOptionsConverter.java

@@ -129,6 +129,11 @@ static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, EventBu
             obj.setOpenSslEngineOptions(new io.vertx.core.net.OpenSSLEngineOptions((io.vertx.core.json.JsonObject)member.getValue()));
           }
           break;
+        case "pcapCaptureFile":
+          if (member.getValue() instanceof String) {
+            obj.setPcapCaptureFile((String)member.getValue());
+          }
+          break;
         case "pemKeyCertOptions":
           if (member.getValue() instanceof JsonObject) {
             obj.setPemKeyCertOptions(new io.vertx.core.net.PemKeyCertOptions((io.vertx.core.json.JsonObject)member.getValue()));
@@ -319,6 +324,9 @@ static void toJson(EventBusOptions obj, java.util.Map<String, Object> json) {
     if (obj.getOpenSslEngineOptions() != null) {
       json.put("openSslEngineOptions", obj.getOpenSslEngineOptions().toJson());
     }
+    if (obj.getPcapCaptureFile() != null) {
+      json.put("pcapCaptureFile", obj.getPcapCaptureFile());
+    }
     if (obj.getPemKeyCertOptions() != null) {
       json.put("pemKeyCertOptions", obj.getPemKeyCertOptions().toJson());
     }

src/main/generated/io/vertx/core/net/NetworkOptionsConverter.java

@@ -25,6 +25,11 @@ static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, Network
             obj.setLogActivity((Boolean)member.getValue());
           }
           break;
+        case "pcapCaptureFile":
+          if (member.getValue() instanceof String) {
+            obj.setPcapCaptureFile((String)member.getValue());
+          }
+          break;
         case "receiveBufferSize":
           if (member.getValue() instanceof Number) {
             obj.setReceiveBufferSize(((Number)member.getValue()).intValue());
@@ -60,6 +65,9 @@ static void toJson(NetworkOptions obj, JsonObject json) {
 
    static void toJson(NetworkOptions obj, java.util.Map<String, Object> json) {
     json.put("logActivity", obj.getLogActivity());
+    if (obj.getPcapCaptureFile() != null) {
+      json.put("pcapCaptureFile", obj.getPcapCaptureFile());
+    }
     json.put("receiveBufferSize", obj.getReceiveBufferSize());
     json.put("reuseAddress", obj.isReuseAddress());
     json.put("reusePort", obj.isReusePort());

src/main/java/io/vertx/core/datagram/DatagramSocketOptions.java

@@ -135,6 +135,11 @@ public DatagramSocketOptions setReusePort(boolean reusePort) {
     return (DatagramSocketOptions) super.setReusePort(reusePort);
   }
 
+  @Override
+  public DatagramSocketOptions setPcapCaptureFile(String pcapCaptureFile) {
+    return (DatagramSocketOptions) super.setPcapCaptureFile(pcapCaptureFile);
+  }
+
   @Override
   public int getTrafficClass() {
     return super.getTrafficClass();

src/main/java/io/vertx/core/eventbus/EventBusOptions.java

@@ -440,6 +440,12 @@ public EventBusOptions setReusePort(boolean reusePort) {
     return this;
   }
 
+  @Override
+  public EventBusOptions setPcapCaptureFile(String pcapCaptureFile) {
+    super.setPcapCaptureFile(pcapCaptureFile);
+    return this;
+  }
+
   @Override
   public EventBusOptions setSendBufferSize(int sendBufferSize) {
     super.setSendBufferSize(sendBufferSize);

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
@@ -412,6 +420,12 @@ public HttpClientOptions setReusePort(boolean reusePort) {
     return this;
   }
 
+  @Override
+  public HttpClientOptions setPcapCaptureFile(String pcapCaptureFile) {
+    super.setPcapCaptureFile(pcapCaptureFile);
+    return this;
+  }
+
   @Override
   public HttpClientOptions setTrafficClass(int trafficClass) {
     super.setTrafficClass(trafficClass);

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
@@ -283,6 +291,12 @@ public HttpServerOptions setReusePort(boolean reusePort) {
     return this;
   }
 
+  @Override
+  public HttpServerOptions setPcapCaptureFile(String pcapCaptureFile) {
+    super.setPcapCaptureFile(pcapCaptureFile);
+    return this;
+  }
+
   @Override
   public HttpServerOptions setTrafficClass(int trafficClass) {
     super.setTrafficClass(trafficClass);

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 {
+
+      }
+    }
+  }
+
+}

src/main/java/io/vertx/core/net/ClientOptionsBase.java

@@ -361,6 +361,11 @@ public ClientOptionsBase setReusePort(boolean reusePort) {
     return (ClientOptionsBase) super.setReusePort(reusePort);
   }
 
+  @Override
+  public ClientOptionsBase setPcapCaptureFile(String pcapCaptureFile) {
+    return (ClientOptionsBase) super.setPcapCaptureFile(pcapCaptureFile);
+  }
+
   @Override
   public ClientOptionsBase setTrafficClass(int trafficClass) {
     return (ClientOptionsBase) super.setTrafficClass(trafficClass);