Implement more robust socket hook protocol

scm-plugins/scm-hg-plugin/src/main/java/sonia/scm/repository/hooks/DefaultHookHandler.java

@@ -79,37 +79,47 @@ class DefaultHookHandler implements HookHandler {
       LOG.warn("failed to read hook request", e);
     } finally {
       LOG.trace("close client socket");
+      TransactionId.clear();
       close();
     }
   }
 
   private void handleHookRequest(InputStream input, OutputStream output) throws IOException {
-    Request request = Sockets.read(input, Request.class);
+    Request request = Sockets.receive(input, Request.class);
+    TransactionId.set(request.getTransactionId());
     Response response = handleHookRequest(request);
     Sockets.send(output, response);
   }
 
   private Response handleHookRequest(Request request) {
     LOG.trace("process {} hook for node {}", request.getType(), request.getNode());
-    TransactionId.set(request.getTransactionId());
 
+    if (!environment.isAcceptAble(request.getChallenge())) {
+      LOG.warn("received hook with invalid challenge: {}", request.getChallenge());
+      return error("invalid hook challenge");
+    }
+
+    try {
+      authenticate(request);
+
+      return fireHook(request);
+    } catch (AuthenticationException ex) {
+      LOG.warn("hook authentication failed", ex);
+      return error("hook authentication failed");
+    }
+  }
+
+  @Nonnull
+  private Response fireHook(Request request) {
     HgHookContextProvider context = hookContextProviderFactory.create(request.getRepositoryId(), request.getNode());
 
     try {
-      if (!environment.isAcceptAble(request.getChallenge())) {
-        LOG.warn("received hook with invalid challenge: {}", request.getChallenge());
-        return error("invalid hook challenge");
-      }
-
-      authenticate(request);
       environment.setPending(request.getType() == RepositoryHookType.PRE_RECEIVE);
 
       hookEventFacade.handle(request.getRepositoryId()).fireHookEvent(request.getType(), context);
 
       return new Response(context.getHgMessageProvider().getMessages(), false);
-    } catch (AuthenticationException ex) {
-      LOG.warn("hook authentication failed", ex);
-      return error("hook authentication failed");
+
     } catch (NotFoundException ex) {
       LOG.warn("could not find repository with id {}", request.getRepositoryId(), ex);
       return error("repository not found");
@@ -121,7 +131,6 @@ class DefaultHookHandler implements HookHandler {
       return error(context, "unknown error");
     } finally {
       environment.clearPendingState();
-      TransactionId.clear();
     }
   }
 

scm-plugins/scm-hg-plugin/src/main/java/sonia/scm/repository/hooks/Sockets.java

@@ -25,14 +25,20 @@
 package sonia.scm.repository.hooks;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import java.io.ByteArrayOutputStream;
+import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 
 class Sockets {
 
+  private static final Logger LOG = LoggerFactory.getLogger(Sockets.class);
+
+  private static final int READ_LIMIT = 8192;
+
   private static final ObjectMapper objectMapper = new ObjectMapper();
 
   private Sockets() {
@@ -40,18 +46,54 @@ class Sockets {
 
   static void send(OutputStream out, Object object) throws IOException {
     byte[] bytes = objectMapper.writeValueAsBytes(object);
+    LOG.trace("send message length of {} to socket", bytes.length);
+    writeInt(out, bytes.length);
+    LOG.trace("send message to socket");
     out.write(bytes);
-    out.write('\0');
+    LOG.trace("flush socket");
     out.flush();
   }
 
-  static <T> T read(InputStream in, Class<T> type) throws IOException {
-    ByteArrayOutputStream buffer = new ByteArrayOutputStream();
-    int c = in.read();
-    while (c != '\0') {
-      buffer.write(c);
-      c = in.read();
+  static <T> T receive(InputStream in, Class<T> type) throws IOException {
+    LOG.trace("read {} from socket", type);
+    int length = readInt(in);
+    LOG.trace("read message length of {} from socket", length);
+    if (length > READ_LIMIT) {
+      String message = String.format("received length of %d, which exceeds the limit of %d", length, READ_LIMIT);
+      throw new IOException(message);
     }
-    return objectMapper.readValue(buffer.toByteArray(), type);
+    byte[] data = read(in, length);
+    LOG.trace("convert message to {}", type);
+    return objectMapper.readValue(data, type);
   }
+
+  private static void writeInt(OutputStream out, int value) throws IOException {
+    out.write((value >>> 24) & 0xFF);
+    out.write((value >>> 16) & 0xFF);
+    out.write((value >>>  8) & 0xFF);
+    out.write(value & 0xFF);
+  }
+
+  private static int readInt(InputStream in) throws IOException {
+    int b1 = in.read();
+    int b2 = in.read();
+    int b3 = in.read();
+    int b4 = in.read();
+
+    if ((b1 | b2 | b3 | b4) < 0) {
+      throw new EOFException("failed to read int from socket");
+    }
+
+    return ((b1 << 24) + (b2 << 16) + (b3 << 8) + b4);
+  }
+
+  private static byte[] read(InputStream in, int length) throws IOException {
+    byte[] buffer = new byte[length];
+    int read = in.read(buffer);
+    if (read < length) {
+      throw new EOFException("failed to read bytes from socket");
+    }
+    return buffer;
+  }
+
 }