Better logging for TLS message on non-secure transport channel (#45835)

This commit enhances logging for 2 cases:

1. If non-TLS enabled node receives transport message from TLS enabled
node on transport port.
2. If non-TLS enabled node receives HTTPs request on transport port.

(cherry picked from commit 4f52ebd32eb58526b4c8022f8863210bf88fc9be)

server/src/main/java/org/elasticsearch/transport/TcpTransport.java

@@ -607,6 +607,9 @@ public abstract class TcpTransport extends AbstractLifecycleComponent implements
                 BytesArray message = new BytesArray(e.getMessage().getBytes(StandardCharsets.UTF_8));
                 outboundHandler.sendBytes(channel, message, ActionListener.wrap(() -> CloseableChannel.closeChannel(channel)));
             }
+        } else if (e instanceof StreamCorruptedException) {
+            logger.warn(() -> new ParameterizedMessage("{}, [{}], closing connection", e.getMessage(), channel));
+            CloseableChannel.closeChannel(channel);
         } else {
             logger.warn(() -> new ParameterizedMessage("exception caught on transport layer [{}], closing connection", channel), e);
             // close the channel, which will cause a node to be disconnected if relevant
@@ -738,11 +741,17 @@ public abstract class TcpTransport extends AbstractLifecycleComponent implements
                 throw new TcpTransport.HttpOnTransportException("This is not an HTTP port");
             }
 
-            throw new StreamCorruptedException("invalid internal transport message format, got ("
+            String firstBytes = "("
                     + Integer.toHexString(headerBuffer.get(0) & 0xFF) + ","
                     + Integer.toHexString(headerBuffer.get(1) & 0xFF) + ","
                     + Integer.toHexString(headerBuffer.get(2) & 0xFF) + ","
-                + Integer.toHexString(headerBuffer.get(3) & 0xFF) + ")");
+                    + Integer.toHexString(headerBuffer.get(3) & 0xFF) + ")";
+
+            if (appearsToBeTLS(headerBuffer)) {
+                throw new StreamCorruptedException("SSL/TLS request received but SSL/TLS is not enabled on this node, got " + firstBytes);
+            }
+
+            throw new StreamCorruptedException("invalid internal transport message format, got " + firstBytes);
         }
         final int messageLength = headerBuffer.getInt(TcpHeader.MARKER_BYTES_SIZE);
 
@@ -775,6 +784,10 @@ public abstract class TcpTransport extends AbstractLifecycleComponent implements
             bufferStartsWith(headerBuffer, "TRACE");
     }
 
+    private static boolean appearsToBeTLS(BytesReference headerBuffer) {
+        return headerBuffer.get(0) == 0x16 && headerBuffer.get(1) == 0x03;
+    }
+
     private static boolean bufferStartsWith(BytesReference buffer, String method) {
         char[] chars = method.toCharArray();
         for (int i = 0; i < chars.length; i++) {

server/src/test/java/org/elasticsearch/transport/TcpTransportTests.java

@@ -288,6 +288,32 @@ public class TcpTransportTests extends ESTestCase {
         }
     }
 
+    public void testTLSHeader() throws IOException {
+        BytesStreamOutput streamOutput = new BytesStreamOutput(1 << 14);
+
+        streamOutput.write(0x16);
+        streamOutput.write(0x03);
+        byte byte1 = randomByte();
+        streamOutput.write(byte1);
+        byte byte2 = randomByte();
+        streamOutput.write(byte2);
+        streamOutput.write(randomByte());
+        streamOutput.write(randomByte());
+        streamOutput.write(randomByte());
+
+        try {
+            BytesReference bytes = streamOutput.bytes();
+            TcpTransport.decodeFrame(bytes);
+            fail("Expected exception");
+        } catch (Exception ex) {
+            assertThat(ex, instanceOf(StreamCorruptedException.class));
+            String expected = "SSL/TLS request received but SSL/TLS is not enabled on this node, got (16,3,"
+                    + Integer.toHexString(byte1 & 0xFF) + ","
+                    + Integer.toHexString(byte2 & 0xFF) + ")";
+            assertEquals(expected, ex.getMessage());
+        }
+    }
+
     public void testHTTPHeader() throws IOException {
         String[] httpHeaders = {"GET", "POST", "PUT", "HEAD", "DELETE", "OPTIONS", "PATCH", "TRACE"};