Add hook-based TCP proxy with gateway integration

- Extensions implement tcp_connection_matcher() to claim connections
- Monkeypatch HTTPChannel to inspect initial bytes before HTTP processing
- First matching extension wins and traffic routes to its backend
- TCP traffic multiplexed through main gateway port (4566) with HTTP
- Provide matcher helpers: create_prefix_matcher, create_signature_matcher, combine_matchers
- No central protocol registry - extensions define their own matchers

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: whummer

utils/localstack_extensions/utils/docker.py

@@ -19,6 +19,8 @@
 from rolo.proxy import Proxy
 from rolo.routing import RuleAdapter, WithHost
 from werkzeug.datastructures import Headers
+from twisted.internet import reactor
+from twisted.protocols.portforward import ProxyFactory
 
 LOG = logging.getLogger(__name__)
 
@@ -30,6 +32,9 @@ class ProxiedDockerContainerExtension(Extension):
 
     Requests may potentially use HTTP2 with binary content as the protocol (e.g., gRPC over HTTP2).
     To ensure proper routing of requests, subclasses can define the `http2_ports`.
+
+    For services requiring raw TCP proxying (e.g., native database protocols), use the `tcp_ports`
+    parameter to enable transparent TCP forwarding to the container.
     """
 
     name: str
@@ -41,7 +46,7 @@ class ProxiedDockerContainerExtension(Extension):
     host: str | None
     """
     Optional host on which to expose the container endpoints.
-    Can be either a static hostname, or a pattern like `<regex("(.+\.)?"):subdomain>myext.<domain>`
+    Can be either a static hostname, or a pattern like `<regex("(.+\\.)?"):subdomain>myext.<domain>`
     """
     path: str | None
     """Optional path on which to expose the container endpoints."""
@@ -59,6 +64,22 @@ class ProxiedDockerContainerExtension(Extension):
     """Callable that returns the target port for a given request, for routing purposes"""
     http2_ports: list[int] | None
     """List of ports for which HTTP2 proxy forwarding into the container should be enabled."""
+    tcp_ports: list[int] | None
+    """
+    List of container ports for raw TCP proxying through the gateway.
+    Enables transparent TCP forwarding for protocols that don't use HTTP (e.g., native DB protocols).
+
+    When tcp_ports is set, the extension must implement tcp_connection_matcher() to identify
+    its traffic by inspecting initial connection bytes.
+    """
+
+    tcp_connection_matcher: Callable[[bytes], bool] | None
+    """
+    Optional function to identify TCP connections belonging to this extension.
+
+    Called with initial connection bytes (up to 512 bytes) to determine if this extension
+    should handle the connection. Return True to claim the connection, False otherwise.
+    """
 
     def __init__(
         self,
@@ -71,6 +92,7 @@ def __init__(
         health_check_fn: Callable[[], None] | None = None,
         request_to_port_router: Callable[[Request], int] | None = None,
         http2_ports: list[int] | None = None,
+        tcp_ports: list[int] | None = None,
     ):
         self.image_name = image_name
         if not container_ports:
@@ -84,6 +106,7 @@ def __init__(
         self.health_check_fn = health_check_fn
         self.request_to_port_router = request_to_port_router
         self.http2_ports = http2_ports
+        self.tcp_ports = tcp_ports
         self.main_port = self.container_ports[0]
         self.container_host = get_addressable_container_host()
 
@@ -106,6 +129,53 @@ def update_gateway_routes(self, router: http.Router[http.RouteHandler]):
                 self.container_host, port, self.should_proxy_request
             )
 
+        # set up raw TCP proxies with protocol detection
+        if self.tcp_ports:
+            self._setup_tcp_protocol_routing()
+
+    def _setup_tcp_protocol_routing(self):
+        """
+        Set up TCP routing on the LocalStack gateway for this extension.
+
+        This method patches the gateway's HTTP protocol handler to intercept TCP
+        connections and allow this extension to claim them via tcp_connection_matcher().
+        This enables multiple TCP protocols to share the main gateway port (4566).
+
+        Uses monkeypatching to intercept dataReceived() before HTTP processing.
+        """
+        from localstack_extensions.utils.tcp_protocol_router import (
+            patch_gateway_for_tcp_routing,
+            register_tcp_extension,
+        )
+
+        # Get the connection matcher from the extension
+        matcher = getattr(self, "tcp_connection_matcher", None)
+        if not matcher:
+            LOG.warning(
+                f"Extension {self.name} has tcp_ports but no tcp_connection_matcher(). "
+                "TCP routing will not work without a matcher."
+            )
+            return
+
+        # Apply gateway patches (only happens once globally)
+        patch_gateway_for_tcp_routing()
+
+        # Register this extension for TCP routing
+        # Use first port as the default target port
+        target_port = self.tcp_ports[0] if self.tcp_ports else self.main_port
+
+        register_tcp_extension(
+            extension_name=self.name,
+            matcher=matcher,
+            backend_host=self.container_host,
+            backend_port=target_port,
+        )
+
+        LOG.info(
+            f"Registered TCP extension {self.name} -> "
+            f"{self.container_host}:{target_port} on gateway"
+        )
+
     @abstractmethod
     def should_proxy_request(self, headers: Headers) -> bool:
         """Define whether a request should be proxied, based on request headers."""

utils/localstack_extensions/utils/tcp_protocol_detector.py

@@ -0,0 +1,115 @@
+"""
+Helper functions for creating TCP connection matchers.
+
+This module provides utilities for extensions to create custom matchers
+that identify their TCP connections from initial bytes.
+"""
+
+import logging
+from typing import Callable
+
+LOG = logging.getLogger(__name__)
+
+# Type alias for matcher functions
+ConnectionMatcher = Callable[[bytes], bool]
+
+
+def create_prefix_matcher(prefix: bytes) -> ConnectionMatcher:
+    """
+    Create a matcher that matches a specific byte prefix.
+
+    Args:
+        prefix: The byte prefix to match
+
+    Returns:
+        A matcher function
+
+    Example:
+        # Match Redis RESP protocol
+        matcher = create_prefix_matcher(b"*")
+    """
+
+    def matcher(data: bytes) -> bool:
+        return data.startswith(prefix)
+
+    return matcher
+
+
+def create_signature_matcher(
+    signature: bytes, offset: int = 0
+) -> ConnectionMatcher:
+    """
+    Create a matcher that matches bytes at a specific offset.
+
+    Args:
+        signature: The byte sequence to match
+        offset: The offset where the signature should appear
+
+    Returns:
+        A matcher function
+
+    Example:
+        # Match PostgreSQL protocol version at offset 4
+        matcher = create_signature_matcher(b"\\x00\\x03\\x00\\x00", offset=4)
+    """
+
+    def matcher(data: bytes) -> bool:
+        if len(data) < offset + len(signature):
+            return False
+        return data[offset : offset + len(signature)] == signature
+
+    return matcher
+
+
+def create_custom_matcher(check_func: Callable[[bytes], bool]) -> ConnectionMatcher:
+    """
+    Create a matcher from a custom checking function.
+
+    Args:
+        check_func: Function that takes bytes and returns bool
+
+    Returns:
+        A matcher function
+
+    Example:
+        def is_my_protocol(data):
+            return len(data) > 10 and data[5] == 0xFF
+
+        matcher = create_custom_matcher(is_my_protocol)
+    """
+    return check_func
+
+
+def combine_matchers(*matchers: ConnectionMatcher) -> ConnectionMatcher:
+    """
+    Combine multiple matchers with OR logic.
+
+    Returns True if any matcher returns True.
+
+    Args:
+        *matchers: Variable number of matcher functions
+
+    Returns:
+        A combined matcher function
+
+    Example:
+        # Match either of two custom protocols
+        matcher1 = create_prefix_matcher(b"PROTO1")
+        matcher2 = create_prefix_matcher(b"PROTO2")
+        combined = combine_matchers(matcher1, matcher2)
+    """
+
+    def combined(data: bytes) -> bool:
+        return any(matcher(data) for matcher in matchers)
+
+    return combined
+
+
+# Export all functions
+__all__ = [
+    "ConnectionMatcher",
+    "create_prefix_matcher",
+    "create_signature_matcher",
+    "create_custom_matcher",
+    "combine_matchers",
+]