Merge pull request #716 from tomato42/raw-writes

add support for writing to socket directly

docs/source/modifying-messages.rst

@@ -46,6 +46,11 @@ You can find a usage example in:
     If you want to send an otherwise valid message, only as plaintext, not
     encrypted, see the :ref:`clearing-encryption-settings` section.
 
+To write directly to the socket, without record layer encapsulation,
+use the :py:class:`~tlsfuzzer.messages.RawSocketWriteGenerator`.
+It accepts two parameters, one to specify the data to write and another,
+optional, used for debugging, the ``description``.
+
 Creating arbitrary messages
 ---------------------------
 

tests/test_tlsfuzzer_messages.py

@@ -30,7 +30,7 @@
         SetPaddingCallback, replace_plaintext, ch_cookie_handler, \
         ch_key_share_handler, SetRecordVersion, CopyVariables, \
         ResetWriteConnectionState, HeartbeatGenerator, Certificate, \
-        KeyUpdateGenerator, ClearContext
+        KeyUpdateGenerator, ClearContext, RawSocketWriteGenerator
 from tlsfuzzer.helpers import psk_ext_gen, psk_ext_updater, \
         psk_session_ext_gen, AutoEmptyExtension
 from tlsfuzzer.runner import ConnectionState
@@ -424,6 +424,43 @@ def test_process(self):
         self.assertEqual(state.msg_sock.version, (3, 2))
 
 
+class TestRawSocketWriteGenerator(unittest.TestCase):
+    def test___init__(self):
+        msg_gen = RawSocketWriteGenerator(bytearray(b'some data'))
+
+        self.assertIsNotNone(msg_gen)
+        self.assertTrue(msg_gen.is_command())
+        self.assertFalse(msg_gen.is_expect())
+        self.assertFalse(msg_gen.is_generator())
+
+    def test___repr__(self):
+        msg_gen = RawSocketWriteGenerator(bytearray(b'some data'))
+
+        self.assertEqual(repr(msg_gen),
+                         "RawSocketWriteGenerator(data=bytearray(b'some data'))")
+
+    def test___repr___with_description(self):
+        msg_gen = RawSocketWriteGenerator(bytearray(b'some data'),
+                                          description="STARTTLS")
+
+        self.assertEqual(repr(msg_gen),
+                         "RawSocketWriteGenerator("
+                         "data=bytearray(b'some data'), "
+                         "description='STARTTLS')")
+
+    def test_process(self):
+        state = ConnectionState()
+        state.msg_sock = mock.MagicMock()
+
+        msg_gen = RawSocketWriteGenerator(b'some data')
+
+        msg_gen.process(state)
+
+        self.assertTrue(
+            state.msg_sock._recordSocket.
+            sock.send.called_once_with(b'some data'))
+
+
 class TestPlaintextMessageGenerator(unittest.TestCase):
     def test___init__(self):
         msg_gen = PlaintextMessageGenerator(12, bytearray(b'\x00\x00'))

tlsfuzzer/messages.py

@@ -376,6 +376,33 @@ def process(self, state):
                 val.append(state.key[name])
 
 
+class RawSocketWriteGenerator(Command):
+    """
+    Send a plaintext data irrespective of encryption state.
+
+    Does not update handshake hashes, record layer state, does not fragment,
+    etc.
+
+    :ivar bytearray ~.data: data to send
+    :ivar str ~.description: identifier to print when processing of the node
+        fails
+    """
+
+    def __init__(self, data, description=None):
+        """Set the record layer type and payload to send."""
+        super(RawSocketWriteGenerator, self).__init__()
+        self.data = data
+        self.description = description
+
+    def __repr__(self):
+        """Return human readable representation of the object."""
+        return self._repr(["data", "description"])
+
+    def process(self, state):
+        """Send the message over the socket."""
+        state.msg_sock._recordSocket.sock.send(self.data)
+
+
 class PlaintextMessageGenerator(Command):
     """
     Send a plaintext data record irrespective of encryption state.