add generic protocol message serialize/deserialize functionality

Signed-off-by: Lance-Drane <Lance-Drane@users.noreply.github.com>

Author: Lance-Drane

src/intersect_sdk_common/control_plane/custom_binary_protocol.py

@@ -0,0 +1,77 @@
+"""This module contains the binary specification for an INTERSECT message IF the protocol does not provide any built-in header support. If it does, use the built-in functionality of the protocol.
+
+Some protocols do not have any built-in header support, leaving it up to us to define the binary structure of the message. We want to avoid "chunking" messages into multiple parts, so a message should be guaranteed to include all metadata.
+"""
+
+from intersect_sdk_common import IntersectApplicationError
+
+_HEADER_KV_SEPARATOR = b'\x02'
+"""Indicates end of a header key and start of a header value.
+
+Used because applications should generally assign no special meaning to this byte, and this byte has no reason to appear in header keys or values.
+"""
+_HEADER_VALUE_SEPARATOR = b'\x03'
+"""Indicates end of a header value and start of the next header key.
+
+Used because applications should generally assign no special meaning to this byte, and this byte has no reason to appear in header keys or values.
+"""
+_PAYLOAD_SEPARATOR = b'\x01'
+"""Indicates end of headers and start of the payload.
+
+Used because applications should generally assign no special meaning to this byte, and this byte has no reason to appear in header keys or values.
+"""
+
+_TOTAL_REASONABLE_HEADER_BYTES = 131072
+"""The total number of bytes that should be reasonably expected to be used for header keys, values, and header separators combined. This provides applications with some level of DOS protection.
+
+It is rare in practice for ANY application to use this number of bytes for the total amount of headers, for example http2_max_header_size in NGINX is rarely set above 128KB
+"""
+
+CONTENT_TYPE_HEADER_KEY = 'content_type'
+CONTENT_TYPE_HEADER_KEY_BYTES = CONTENT_TYPE_HEADER_KEY.encode()
+
+
+def create_binary_message(body: bytes, content_type: str, headers: dict[str, str]) -> bytes:
+    """Create a binary message from the body, headers, and content type."""
+    return b''.join(
+        [
+            # content-type 'header' first (this is generally handled separately from other headers in many protocols)
+            CONTENT_TYPE_HEADER_KEY_BYTES,
+            _HEADER_KV_SEPARATOR,
+            content_type.encode(),
+            _HEADER_VALUE_SEPARATOR if len(headers) else b'',
+            # headers
+            _HEADER_VALUE_SEPARATOR.join(
+                _HEADER_KV_SEPARATOR.join([key.encode(), value.encode()])
+                for key, value in headers.items()
+            ),
+            # end of headers
+            _PAYLOAD_SEPARATOR,
+            # body
+            body,
+        ]
+    )
+
+
+def parse_binary_message(message: bytes) -> tuple[bytes, str, dict[str, str]]:
+    """Parse a binary message into its body, content type, and headers."""
+    # IMPORTANT!!! ----- Total length of header keys and values combined should be limited to first several bytes, terminate header search early if headers aren't a reasonable length.
+    payload_sep_location = message.find(_PAYLOAD_SEPARATOR, 0, _TOTAL_REASONABLE_HEADER_BYTES)
+    if payload_sep_location == -1:
+        msg = 'Probable malformed message: no payload separator found in first expected bytes, .'
+        raise IntersectApplicationError(msg)
+    header_string = message[:payload_sep_location]
+    headers = {
+        key.decode(): value.decode()
+        for key, value in (
+            header.split(_HEADER_KV_SEPARATOR)
+            for header in header_string.split(_HEADER_VALUE_SEPARATOR)
+        )
+    }
+    try:
+        content_type = headers.pop(CONTENT_TYPE_HEADER_KEY)
+    except KeyError as e:
+        msg = 'Probable malformed message: no content_type header found in message, discarding it.'
+        raise IntersectApplicationError(msg) from e
+
+    return message[payload_sep_location + 1 :], content_type, headers

tests/unit/test_custom_binary_protocol.py

@@ -0,0 +1,48 @@
+import pytest
+
+from intersect_sdk_common.control_plane.custom_binary_protocol import (
+    CONTENT_TYPE_HEADER_KEY,
+    create_binary_message,
+    parse_binary_message,
+)
+
+
+@pytest.mark.parametrize(
+    ('body', 'content_type', 'headers'),
+    [
+        (b'{"key": "value"}', 'application/json', {'header1': 'value1', 'header2': 'value2'}),
+        (b'<xml><key>value</key></xml>', 'application/xml', {'header1': 'value1'}),
+        (b'plain text body', 'text/plain', {}),
+        (
+            b'I\x01Have\x02Special\x03Characters',
+            'text/plain',
+            {'But headers': 'cannot have control characters'},
+        ),
+    ],
+)
+def test_custom_binary_protocol_idempotency(
+    body: bytes, content_type: str, headers: dict[str, str]
+):
+    message = create_binary_message(body, content_type, headers)
+    parsed_body, parsed_content_type, parsed_headers = parse_binary_message(message)
+
+    assert parsed_body == body
+    assert parsed_content_type == content_type
+    assert parsed_headers == headers
+
+    # message body should be at the end of the full message
+    body_len = len(body)
+    assert body == message[-body_len:]
+
+    # complete message should only add 2 bytes for each metadata field
+    assert (
+        len(body)
+        + len(CONTENT_TYPE_HEADER_KEY)
+        + len(content_type)
+        + 2  # header key separator and header value terminator
+        + sum(
+            len(key) + len(value) + 2  # header key separator and header value terminator
+            for key, value in headers.items()
+        )
+        == len(message)
+    )