MeshChatX/vendor/rns_over_http/HTTPInterface.py

import asyncio
import os
import socket
import ssl
import threading
from collections import deque
from http.server import BaseHTTPRequestHandler, HTTPServer
from queue import Empty, Queue
from socketserver import ThreadingMixIn
from urllib.parse import urlparse

import httpx
import RNS
from RNS.Interfaces.Interface import Interface


class HDLC:
    """Pipe-compatible simplified HDLC framing for HTTP bodies.

    Same FLAG/ESC scheme as RNS PipeInterface so multiple packets can
    share one HTTP request or response without losing boundaries.
    """

    FLAG = 0x7E
    ESC = 0x7D
    ESC_MASK = 0x20

    @staticmethod
    def escape(data):
        data = data.replace(
            bytes([HDLC.ESC]),
            bytes([HDLC.ESC, HDLC.ESC ^ HDLC.ESC_MASK]),
        )
        data = data.replace(
            bytes([HDLC.FLAG]),
            bytes([HDLC.ESC, HDLC.FLAG ^ HDLC.ESC_MASK]),
        )
        return data

    @staticmethod
    def frame(packet):
        return bytes([HDLC.FLAG]) + HDLC.escape(packet) + bytes([HDLC.FLAG])

    @staticmethod
    def deframe(buffer, max_frame_len):
        """Return (packets, remainder) from an HDLC byte buffer."""
        packets = []
        in_frame = False
        escape = False
        data_buffer = bytearray()
        i = 0
        last_complete = 0

        while i < len(buffer):
            byte = buffer[i]
            i += 1

            if in_frame and byte == HDLC.FLAG:
                packets.append(bytes(data_buffer))
                in_frame = False
                escape = False
                data_buffer = bytearray()
                last_complete = i
            elif byte == HDLC.FLAG:
                in_frame = True
                escape = False
                data_buffer = bytearray()
            elif in_frame and len(data_buffer) < max_frame_len:
                if byte == HDLC.ESC:
                    escape = True
                else:
                    if escape:
                        if byte == HDLC.FLAG ^ HDLC.ESC_MASK:
                            byte = HDLC.FLAG
                        elif byte == HDLC.ESC ^ HDLC.ESC_MASK:
                            byte = HDLC.ESC
                        escape = False
                    data_buffer.append(byte)

        remainder = buffer[last_complete:] if in_frame else b""
        return packets, remainder


class _H3Session:
    """Persistent HTTP/3 client over one QUIC connection."""

    def __init__(self, server_url, user_agent, verify=True, ca_certs=None):
        from aioquic.asyncio import connect
        from aioquic.asyncio.protocol import QuicConnectionProtocol
        from aioquic.h3.connection import H3_ALPN, H3Connection
        from aioquic.h3.events import DataReceived, HeadersReceived
        from aioquic.quic.configuration import QuicConfiguration
        from aioquic.quic.events import QuicEvent

        parsed = urlparse(server_url)
        if parsed.scheme != "https":
            raise ValueError("HTTP/3 requires an https:// server_url")

        self._host = parsed.hostname
        self._port = parsed.port or 443
        self._path = parsed.path or "/"
        if parsed.query:
            self._path += "?" + parsed.query
        self._authority = parsed.netloc
        self._user_agent = user_agent
        self._closed = False
        self._request_count = 0

        configuration = QuicConfiguration(is_client=True, alpn_protocols=H3_ALPN)
        if not verify:
            configuration.verify_mode = ssl.CERT_NONE
        elif ca_certs:
            configuration.load_verify_locations(ca_certs)

        class _Client(QuicConnectionProtocol):
            def __init__(self, *args, **kwargs):
                super().__init__(*args, **kwargs)
                self._http = H3Connection(self._quic)
                self._request_events = {}
                self._request_waiter = {}

            def http_event_received(self, event):
                if isinstance(event, (HeadersReceived, DataReceived)):
                    stream_id = event.stream_id
                    if stream_id in self._request_events:
                        self._request_events[stream_id].append(event)
                        if event.stream_ended:
                            waiter = self._request_waiter.pop(stream_id)
                            waiter.set_result(self._request_events.pop(stream_id))

            def quic_event_received(self, event: QuicEvent):
                for http_event in self._http.handle_event(event):
                    self.http_event_received(http_event)

            async def post_bytes(self, path, authority, user_agent, data):
                stream_id = self._quic.get_next_available_stream_id()
                headers = [
                    (b":method", b"POST"),
                    (b":scheme", b"https"),
                    (b":authority", authority.encode()),
                    (b":path", path.encode()),
                    (b"user-agent", user_agent.encode()),
                    (b"content-type", b"application/octet-stream"),
                    (b"content-length", str(len(data)).encode()),
                ]
                self._http.send_headers(
                    stream_id=stream_id,
                    headers=headers,
                    end_stream=not data,
                )
                if data:
                    self._http.send_data(stream_id=stream_id, data=data, end_stream=True)

                waiter = self._loop.create_future()
                self._request_events[stream_id] = deque()
                self._request_waiter[stream_id] = waiter
                self.transmit()
                events = await asyncio.shield(waiter)

                status = 0
                body = b""
                for event in events:
                    if isinstance(event, HeadersReceived):
                        for key, value in event.headers:
                            if key == b":status":
                                status = int(value.decode())
                    elif isinstance(event, DataReceived):
                        body += event.data
                return status, body

        self._Client = _Client
        self._connect = connect
        self._configuration = configuration
        self._loop = asyncio.new_event_loop()
        self._loop_thread = threading.Thread(
            target=self._run_loop,
            name="h3-client-loop",
            daemon=True,
        )
        self._ready = threading.Event()
        self._error = None
        self._client = None
        self._cm = None
        self._loop_thread.start()
        if not self._ready.wait(timeout=30):
            raise TimeoutError("HTTP/3 client failed to connect")
        if self._error is not None:
            raise self._error

    def _run_loop(self):
        asyncio.set_event_loop(self._loop)
        try:
            self._loop.run_until_complete(self._open())
            self._ready.set()
            self._loop.run_forever()
        except Exception as exc:
            self._error = exc
            self._ready.set()

    async def _open(self):
        self._cm = self._connect(
            self._host,
            self._port,
            configuration=self._configuration,
            create_protocol=self._Client,
        )
        self._client = await self._cm.__aenter__()

    def post(self, data, timeout=10.0):
        if self._closed:
            raise RuntimeError("HTTP/3 session is closed")

        async def _do_post():
            status, body = await self._client.post_bytes(
                self._path,
                self._authority,
                self._user_agent,
                data,
            )
            self._request_count += 1
            if status >= 400:
                raise RuntimeError(f"HTTP/3 status {status}")
            return body

        fut = asyncio.run_coroutine_threadsafe(_do_post(), self._loop)
        return fut.result(timeout=timeout)

    def close(self):
        if self._closed:
            return
        self._closed = True

        async def _close():
            if self._cm is not None:
                await self._cm.__aexit__(None, None, None)

        try:
            fut = asyncio.run_coroutine_threadsafe(_close(), self._loop)
            fut.result(timeout=5)
        except Exception:
            pass
        self._loop.call_soon_threadsafe(self._loop.stop)
        self._loop_thread.join(timeout=5)


class HTTPTunnelInterface(Interface):
    """HTTP Tunnel Interface for Reticulum.

    Bidirectional RNS transport over HTTP POST with pipe-compatible HDLC
    framing. Supports HTTP/1.1 (default), HTTP/2, and HTTP/3.

    Engineering defaults:
      - http_version=1 uses a stdlib HTTP/1.1 server and httpx client.
        Works over cleartext http:// and is ideal behind Caddy/nginx.
      - http_version=2 requires TLS and uses Hypercorn (h2) + httpx(http2).
      - http_version=3 requires TLS and uses Hypercorn QUIC + a persistent
        aioquic HTTP/3 client session.

    Configuration highlights:
        http_version: 1, 2, or 3 (default: 1)
        tls_certfile / tls_keyfile: required for versions 2 and 3 on server
        tls_verify: verify TLS certs on client (default: true)
        tls_ca_certs: optional CA bundle path for client verify
        server_url: use https:// for versions 2 and 3
    """

    DEFAULT_IFAC_SIZE = 16
    BITRATE_GUESS = 10_000_000
    AUTOCONFIGURE_MTU = True

    DEFAULT_MTU = 4096
    TUNNEL_USER_AGENT = "RNS-HTTP-Tunnel/1.0"
    DEFAULT_POLL_INTERVAL = 0.1
    DEFAULT_POOL_CONNECTIONS = 1
    DEFAULT_POOL_MAXSIZE = 1
    DEFAULT_KEEPALIVE_TIMEOUT = 60
    DEFAULT_HTTP_VERSION = 1

    def __init__(self, owner, configuration):
        super().__init__()

        ifconf = Interface.get_config_obj(configuration)

        self.name = ifconf["name"]

        mode = str(ifconf["mode"]).lower() if "mode" in ifconf else "client"
        listen_host = ifconf["listen_host"] if "listen_host" in ifconf else "0.0.0.0"
        listen_port = int(ifconf["listen_port"]) if "listen_port" in ifconf else 8080
        server_url = ifconf["server_url"] if "server_url" in ifconf else None
        poll_interval = (
            float(ifconf["poll_interval"])
            if "poll_interval" in ifconf
            else self.DEFAULT_POLL_INTERVAL
        )
        check_user_agent = (
            ifconf.as_bool("check_user_agent") if "check_user_agent" in ifconf else True
        )
        user_agent = (
            str(ifconf["user_agent"])
            if "user_agent" in ifconf
            else self.TUNNEL_USER_AGENT
        )
        mtu = int(ifconf["mtu"]) if "mtu" in ifconf else self.DEFAULT_MTU
        serve_html_page = (
            ifconf.as_bool("serve_html_page") if "serve_html_page" in ifconf else False
        )
        html_file_path = (
            ifconf["html_file_path"] if "html_file_path" in ifconf else None
        )
        pool_connections = (
            int(ifconf["pool_connections"])
            if "pool_connections" in ifconf
            else self.DEFAULT_POOL_CONNECTIONS
        )
        pool_maxsize = (
            int(ifconf["pool_maxsize"])
            if "pool_maxsize" in ifconf
            else self.DEFAULT_POOL_MAXSIZE
        )
        keepalive_timeout = (
            int(ifconf["keepalive_timeout"])
            if "keepalive_timeout" in ifconf
            else self.DEFAULT_KEEPALIVE_TIMEOUT
        )
        http_version = (
            int(ifconf["http_version"])
            if "http_version" in ifconf
            else self.DEFAULT_HTTP_VERSION
        )
        tls_certfile = ifconf["tls_certfile"] if "tls_certfile" in ifconf else None
        tls_keyfile = ifconf["tls_keyfile"] if "tls_keyfile" in ifconf else None
        tls_verify = (
            ifconf.as_bool("tls_verify") if "tls_verify" in ifconf else True
        )
        tls_ca_certs = ifconf["tls_ca_certs"] if "tls_ca_certs" in ifconf else None

        self.mode = mode
        self.http_version = http_version

        if mode not in ["client", "server"]:
            raise ValueError(
                f"Invalid mode '{mode}' for {self}. Must be 'client' or 'server'",
            )

        if http_version not in (1, 2, 3):
            raise ValueError(
                f"Invalid http_version '{http_version}' for {self}. Must be 1, 2, or 3",
            )

        if mode == "client" and server_url is None:
            raise ValueError(f"No server_url specified for client mode in {self}")

        if pool_connections < 1 or pool_maxsize < 1:
            raise ValueError(
                f"pool_connections and pool_maxsize must be >= 1 for {self}",
            )

        if http_version >= 2:
            if mode == "server" and (not tls_certfile or not tls_keyfile):
                raise ValueError(
                    f"tls_certfile and tls_keyfile are required for "
                    f"HTTP/{http_version} server mode in {self}",
                )
            if mode == "client":
                parsed = urlparse(server_url)
                if parsed.scheme != "https":
                    raise ValueError(
                        f"HTTP/{http_version} client requires https:// server_url in {self}",
                    )

        self.owner = owner
        self.IN = True
        self.mtu = mtu
        self.check_user_agent = check_user_agent
        self.user_agent = user_agent
        self.serve_html_page = serve_html_page
        self.html_file_path = html_file_path
        self.html_content = None
        self.pool_connections = pool_connections
        self.pool_maxsize = pool_maxsize
        self.keepalive_timeout = keepalive_timeout
        self.tls_certfile = tls_certfile
        self.tls_keyfile = tls_keyfile
        self.tls_verify = tls_verify
        self.tls_ca_certs = tls_ca_certs
        self._tcp_accepts = 0
        self._http_requests = 0
        self._stats_lock = threading.Lock()
        self._last_http_version = None
        self._client_requests = 0
        self.session = None
        self._h3_session = None
        self._asgi_shutdown = None
        self._asgi_loop = None

        if self.serve_html_page and self.html_file_path:
            self._load_html_content()

        self._recv_queue = Queue()
        self._send_queue = Queue()
        self._stop_event = threading.Event()
        self._frame_remainder = b""
        self._frame_lock = threading.Lock()

        self.HW_MTU = mtu
        self.online = False
        self.bitrate = HTTPTunnelInterface.BITRATE_GUESS

        if mode == "server":
            self.listen_host = listen_host
            self.listen_port = listen_port
        else:
            self.server_url = server_url
            self.poll_interval = poll_interval

        self.optimise_mtu()

        if mode == "server":
            if http_version == 1:
                self.setup_server_http1()
            else:
                self.setup_server_asgi()
        else:
            self.setup_client()

    def _load_html_content(self):
        try:
            if os.path.isfile(self.html_file_path):
                with open(self.html_file_path, encoding="utf-8") as f:
                    self.html_content = f.read()
                RNS.log(f"Loaded HTML content from {self.html_file_path}", RNS.LOG_INFO)
            else:
                RNS.log(f"HTML file not found: {self.html_file_path}", RNS.LOG_WARNING)
                self.html_content = None
        except Exception as e:
            RNS.log(
                f"Error loading HTML file {self.html_file_path}: {e}",
                RNS.LOG_ERROR,
            )
            self.html_content = None

    def _drain_send_frames(self):
        parts = []
        while not self._send_queue.empty():
            try:
                packet = self._send_queue.get_nowait()
            except Empty:
                break
            if packet:
                parts.append(HDLC.frame(packet))
        return b"".join(parts)

    def _ingest_wire_bytes(self, wire_data):
        if not wire_data:
            return

        with self._frame_lock:
            buffer = self._frame_remainder + wire_data
            packets, self._frame_remainder = HDLC.deframe(buffer, self.HW_MTU)

        for packet in packets:
            if packet:
                self._recv_queue.put(packet)

    def _record_http_request(self):
        with self._stats_lock:
            self._http_requests += 1

    def _handle_tunnel_exchange(self, method, path, headers, body):
        """Shared request handler for HTTP/1.1 and ASGI paths.

        Returns (status, content_type, response_body).
        """
        self._record_http_request()
        headers_l = {str(k).lower(): str(v) for k, v in headers.items()}

        if method == "GET" and path == "/":
            if self.serve_html_page and self.html_content:
                return (
                    200,
                    "text/html; charset=utf-8",
                    self.html_content.encode("utf-8"),
                )
            return 404, "text/plain", b""

        if method == "POST" and path == "/":
            if self.check_user_agent:
                user_agent = headers_l.get("user-agent", "")
                if user_agent != self.user_agent:
                    RNS.log(
                        f"Rejected request with invalid User-Agent: {user_agent}",
                        RNS.LOG_WARNING,
                    )
                    return 403, "text/plain", b"Forbidden"

            if body:
                RNS.log(f"Received {len(body)} bytes from client", RNS.LOG_EXTREME)
                self._ingest_wire_bytes(body)

            server_data = self._drain_send_frames()
            if server_data:
                RNS.log(
                    f"Sending {len(server_data)} framed bytes to client",
                    RNS.LOG_EXTREME,
                )
            return 200, "application/octet-stream", server_data

        return 404, "text/plain", b""

    def connection_stats(self):
        with self._stats_lock:
            stats = {
                "tcp_accepts": self._tcp_accepts,
                "http_requests": self._http_requests,
                "client_requests": self._client_requests,
                "last_http_version": self._last_http_version,
                "configured_http_version": self.http_version,
            }
        if self._h3_session is not None:
            stats["pool_num_connections"] = 0 if self._h3_session._closed else 1
            stats["pool_num_requests"] = self._h3_session._request_count
        return stats

    def _build_asgi_app(self):
        interface = self

        async def app(scope, receive, send):
            if scope["type"] != "http":
                return

            headers = {
                key.decode("latin1"): value.decode("latin1")
                for key, value in scope.get("headers", [])
            }
            body = b""
            while True:
                message = await receive()
                body += message.get("body", b"")
                if not message.get("more_body", False):
                    break

            status, content_type, response_body = interface._handle_tunnel_exchange(
                scope.get("method", "GET"),
                scope.get("path", "/"),
                headers,
                body,
            )
            await send(
                {
                    "type": "http.response.start",
                    "status": status,
                    "headers": [
                        (b"content-type", content_type.encode()),
                        (b"content-length", str(len(response_body)).encode()),
                    ],
                }
            )
            await send({"type": "http.response.body", "body": response_body})

        return app

    def setup_server_http1(self):
        interface_instance = self

        class TunnelRequestHandler(BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.1"

            def _send_common_headers(self, content_type, content_length):
                self.send_header("Content-Type", content_type)
                self.send_header("Content-Length", str(content_length))
                self.send_header("Connection", "keep-alive")
                self.send_header(
                    "Keep-Alive",
                    f"timeout={interface_instance.keepalive_timeout}, max=1000",
                )

            def do_GET(self):
                status, content_type, body = interface_instance._handle_tunnel_exchange(
                    "GET",
                    self.path,
                    self.headers,
                    b"",
                )
                self.send_response(status)
                self._send_common_headers(content_type, len(body))
                self.end_headers()
                if body:
                    self.wfile.write(body)

            def do_POST(self):
                content_length = int(self.headers.get("Content-Length", 0))
                body = self.rfile.read(content_length) if content_length > 0 else b""
                status, content_type, response = interface_instance._handle_tunnel_exchange(
                    "POST",
                    self.path,
                    self.headers,
                    body,
                )
                self.send_response(status)
                self._send_common_headers(content_type, len(response))
                self.end_headers()
                if response:
                    self.wfile.write(response)

            def log_message(self, fmt, *args):
                pass

        class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
            daemon_threads = True
            allow_reuse_address = True

            def get_request(self):
                request, client_address = super().get_request()
                with interface_instance._stats_lock:
                    interface_instance._tcp_accepts += 1
                try:
                    request.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
                except OSError:
                    pass
                return request, client_address

        def run_server():
            try:
                self._http_server = ThreadedHTTPServer(
                    (self.listen_host, self.listen_port),
                    TunnelRequestHandler,
                )
                self._http_server.daemon_threads = True
                self._http_server.serve_forever()
            except Exception as e:
                if not self._stop_event.is_set():
                    RNS.log(f"HTTP server error for {self}: {e}", RNS.LOG_ERROR)
                    if RNS.Reticulum.panic_on_interface_error:
                        RNS.panic()

        self._server_thread = threading.Thread(target=run_server, daemon=True)
        self._server_thread.start()
        self._start_receive_loop()
        self.online = True
        RNS.log(
            f"HTTP/1.1 server started on http://{self.listen_host}:{self.listen_port}",
            RNS.LOG_NOTICE,
        )

    def setup_server_asgi(self):
        from hypercorn.asyncio import serve
        from hypercorn.config import Config

        config = Config()
        bind = f"{self.listen_host}:{self.listen_port}"
        config.bind = [bind]
        config.certfile = self.tls_certfile
        config.keyfile = self.tls_keyfile
        config.alpn_protocols = ["h2", "http/1.1"]
        if self.http_version == 3:
            config.quic_bind = [bind]
            config.alpn_protocols = ["h3", "h2", "http/1.1"]

        app = self._build_asgi_app()
        self._asgi_shutdown = asyncio.Event()

        def run_server():
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            self._asgi_loop = loop
            try:
                loop.run_until_complete(
                    serve(app, config, shutdown_trigger=self._asgi_shutdown.wait),
                )
            except Exception as e:
                if not self._stop_event.is_set():
                    RNS.log(f"ASGI HTTP server error for {self}: {e}", RNS.LOG_ERROR)
                    if RNS.Reticulum.panic_on_interface_error:
                        RNS.panic()
            finally:
                loop.close()

        self._server_thread = threading.Thread(target=run_server, daemon=True)
        self._server_thread.start()
        self._start_receive_loop()
        self.online = True
        proto = f"HTTP/{self.http_version}"
        RNS.log(
            f"{proto} server started on https://{self.listen_host}:{self.listen_port}",
            RNS.LOG_NOTICE,
        )

    def _start_receive_loop(self):
        thread = threading.Thread(target=self.receive_loop, daemon=True)
        thread.start()

    def setup_client(self):
        self._consecutive_failures = 0
        self._max_backoff = 30.0

        if self.http_version == 3:
            self._h3_session = _H3Session(
                self.server_url,
                self.user_agent,
                verify=self.tls_verify,
                ca_certs=self.tls_ca_certs,
            )
            self._last_http_version = "HTTP/3"
        else:
            verify = self.tls_ca_certs if self.tls_ca_certs else self.tls_verify
            limits = httpx.Limits(
                max_connections=max(self.pool_maxsize, 1),
                max_keepalive_connections=max(self.pool_maxsize, 1),
                keepalive_expiry=float(self.keepalive_timeout),
            )
            self.session = httpx.Client(
                http2=(self.http_version == 2),
                verify=verify,
                headers={
                    "User-Agent": self.user_agent,
                    "Accept-Encoding": "identity",
                },
                limits=limits,
                timeout=httpx.Timeout(5.0),
            )

        thread = threading.Thread(target=self.client_loop, daemon=True)
        thread.start()
        self.online = True
        RNS.log(
            f"HTTP/{self.http_version} client started, connecting to {self.server_url}",
            RNS.LOG_NOTICE,
        )

    def receive_loop(self):
        while not self._stop_event.is_set():
            try:
                received_data = self._recv_queue.get(timeout=1)
                if received_data:
                    self.process_incoming(received_data)
            except Empty:
                continue
            except Exception as e:
                if not self._stop_event.is_set():
                    RNS.log(f"Error in receive loop for {self}: {e}", RNS.LOG_ERROR)

    def _client_exchange(self, data_to_send):
        if self.http_version == 3:
            body = self._h3_session.post(data_to_send, timeout=5.0)
            self._client_requests += 1
            self._last_http_version = "HTTP/3"
            return body

        response = self.session.post(self.server_url, content=data_to_send)
        response.raise_for_status()
        self._client_requests += 1
        self._last_http_version = response.http_version
        return response.content

    def client_loop(self):
        while not self._stop_event.is_set():
            data_to_send = self._drain_send_frames()

            try:
                RNS.log(
                    f"Sending {len(data_to_send)} framed bytes to server",
                    RNS.LOG_EXTREME,
                )
                content = self._client_exchange(data_to_send)

                if content:
                    RNS.log(
                        f"Received {len(content)} bytes from server",
                        RNS.LOG_EXTREME,
                    )
                    self._ingest_wire_bytes(content)

                    while not self._recv_queue.empty():
                        try:
                            packet = self._recv_queue.get_nowait()
                        except Empty:
                            break
                        if packet:
                            self.process_incoming(packet)

                if self._consecutive_failures > 0:
                    RNS.log(f"Reconnected to server for {self}", RNS.LOG_INFO)
                    self._consecutive_failures = 0

            except Exception as e:
                if self._stop_event.is_set():
                    break
                self._consecutive_failures += 1
                if self._consecutive_failures % 10 == 1:
                    RNS.log(
                        f"Error communicating with server for {self} "
                        f"(attempt {self._consecutive_failures}): {e}",
                        RNS.LOG_WARNING,
                    )

            if self._stop_event.is_set():
                break

            if self._consecutive_failures > 0:
                delay = min(
                    self.poll_interval * (2 ** min(self._consecutive_failures - 1, 5)),
                    self._max_backoff,
                )
            else:
                delay = self.poll_interval

            self._stop_event.wait(delay)

    def process_incoming(self, data):
        if len(data) > 0 and self.online:
            self.rxb += len(data)
            self.owner.inbound(data, self)

    def process_outgoing(self, data):
        if self.online:
            if len(data) > self.mtu:
                RNS.log(
                    f"Payload too large ({len(data)} > {self.mtu}) for {self}",
                    RNS.LOG_ERROR,
                )
                return

            self._send_queue.put(data)
            self.txb += len(data)

    def detach(self):
        RNS.log(f"Detaching {self}", RNS.LOG_DEBUG)
        self._stop_event.set()
        self.online = False

        if self.mode == "client":
            if self.session is not None:
                try:
                    self.session.close()
                except Exception as e:
                    RNS.log(
                        f"Error closing HTTP session for {self}: {e}",
                        RNS.LOG_DEBUG,
                    )
            if self._h3_session is not None:
                try:
                    self._h3_session.close()
                except Exception as e:
                    RNS.log(
                        f"Error closing HTTP/3 session for {self}: {e}",
                        RNS.LOG_DEBUG,
                    )

        if self.mode == "server":
            if self.http_version == 1:
                httpd = getattr(self, "_http_server", None)
                if httpd is not None:
                    def _shutdown():
                        try:
                            httpd.shutdown()
                        except Exception:
                            pass
                        try:
                            httpd.server_close()
                        except Exception:
                            pass

                    threading.Thread(target=_shutdown, daemon=True).start()

                if hasattr(self, "_server_thread") and self._server_thread:
                    self._server_thread.join(timeout=2)
                    if self._server_thread.is_alive() and httpd is not None:
                        try:
                            httpd.socket.close()
                        except Exception:
                            pass
                        self._server_thread.join(timeout=1)
            else:
                if self._asgi_loop is not None and self._asgi_shutdown is not None:
                    self._asgi_loop.call_soon_threadsafe(self._asgi_shutdown.set)
                if hasattr(self, "_server_thread") and self._server_thread:
                    self._server_thread.join(timeout=5)

    def should_ingress_limit(self):
        return False

    def __str__(self):
        name = getattr(self, "name", "?")
        ver = getattr(self, "http_version", "?")
        if self.mode == "server":
            lh = getattr(self, "listen_host", "?")
            lp = getattr(self, "listen_port", "?")
            return f"HTTPTunnelInterface[{name}/HTTP{ver}/server/{lh}:{lp}]"
        if self.mode == "client" and getattr(self, "server_url", None) is not None:
            return f"HTTPTunnelInterface[{name}/HTTP{ver}/client/{self.server_url}]"
        return f"HTTPTunnelInterface[{name}/HTTP{ver}/{self.mode}]"


interface_class = HTTPTunnelInterface