From 7f115c01024b86e8ef64e161a5807e2492a92b83 Mon Sep 17 00:00:00 2001 From: Ivan Date: Sun, 5 Oct 2025 15:42:30 -0500 Subject: [PATCH] Implement Custom Interface --- HTTPInterface.py | 333 ++++++++++++++++++++++++++++++++++++++++++++++ README.md | 74 ++++++----- http_interface.py | 310 ------------------------------------------ 3 files changed, 376 insertions(+), 341 deletions(-) create mode 100644 HTTPInterface.py delete mode 100644 http_interface.py diff --git a/HTTPInterface.py b/HTTPInterface.py new file mode 100644 index 0000000..213e206 --- /dev/null +++ b/HTTPInterface.py @@ -0,0 +1,333 @@ +"""RNS-over-HTTP Interface + +Custom HTTP interface for Reticulum that implements bidirectional communication using HTTP POST requests. +""" + +import importlib +from http.server import BaseHTTPRequestHandler, HTTPServer +from queue import Empty, Queue +from socketserver import ThreadingMixIn +from threading import Event, Thread +from time import sleep + +import RNS +from RNS.Interfaces.Interface import Interface + +MTU = 4096 +DEFAULT_USER_AGENT = "RNS-HTTP-Tunnel/1.0" + + +class HTTPInterface(Interface): + """HTTPInterface provides a Reticulum interface over HTTP using bidirectional communication via HTTP POST requests.""" + + DEFAULT_IFAC_SIZE = 8 + + owner = None + mode = None + listen_host = None + listen_port = None + server_url = None + poll_interval = None + check_user_agent = None + _recv_queue = None + _send_queue = None + _stop_event = None + _worker_thread = None + _http_server = None + session = None + _consecutive_failures = None + _max_backoff = None + + def __init__(self, owner, configuration): + """Initialize the HTTPInterface with the given owner and configuration.""" + if importlib.util.find_spec("requests") is None: + RNS.log( + "Using this interface requires the requests module to be installed.", + RNS.LOG_CRITICAL, + ) + RNS.log( + "You can install it with the command: python3 -m pip install requests", + RNS.LOG_CRITICAL, + ) + RNS.panic() + + import requests + + super().__init__() + + ifconf = Interface.get_config_obj(configuration) + + name = ifconf["name"] + self.name = name + + mode = ifconf["mode"] 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 1.0 + ) + check_user_agent = ( + bool(ifconf["check_user_agent"]) if "check_user_agent" in ifconf else True + ) + user_agent = ( + ifconf["user_agent"] if "user_agent" in ifconf else DEFAULT_USER_AGENT + ) + mtu = int(ifconf["mtu"]) if "mtu" in ifconf else MTU + + if mode not in ["server", "client"]: + raise ValueError(f"Invalid mode '{mode}'. Must be 'server' or 'client'") + if mode == "client" and server_url is None: + raise ValueError("server_url is required for client mode") + + self.HW_MTU = mtu + + self.online = False + self.bitrate = 1000000 + + self.owner = owner + self.mode = mode + self.listen_host = listen_host + self.listen_port = listen_port + self.server_url = server_url + self.poll_interval = poll_interval + self.check_user_agent = check_user_agent + self.user_agent = user_agent + + self._recv_queue = Queue() + self._send_queue = Queue() + self._stop_event = Event() + + if self.mode == "server": + self._http_server = None + self._request_handler_class = self._create_request_handler() + else: + self.session = requests.Session() + self.session.headers.update({"User-Agent": self.user_agent}) + self._consecutive_failures = 0 + self._max_backoff = 30.0 + + try: + self._start_interface() + self._read_thread = Thread(target=self._read_loop, daemon=True) + self._read_thread.start() + except Exception as e: + RNS.log("Could not start HTTP interface " + str(self), RNS.LOG_ERROR) + raise e + + def _create_request_handler(self): + """Create a custom HTTP request handler class for the server mode.""" + interface_instance = self + + class TunnelRequestHandler(BaseHTTPRequestHandler): + """Handles HTTP POST requests for the HTTPInterface server.""" + + def __init__(self, request, client_address, server): + self.interface = interface_instance + super().__init__(request, client_address, server) + + def do_POST(self): + """Handle HTTP POST requests for bidirectional data exchange.""" + if self.path == "/": + if self.interface.check_user_agent: + user_agent = self.headers.get("User-Agent", "") + if user_agent != self.interface.user_agent: + RNS.log( + f"Rejected request with invalid User-Agent: {user_agent}", + RNS.LOG_WARNING, + ) + self.send_response(403) + self.send_header("Content-Type", "text/plain") + self.end_headers() + self.wfile.write(b"Forbidden") + return + + content_length = int(self.headers.get("Content-Length", 0)) + client_data = ( + self.rfile.read(content_length) if content_length > 0 else b"" + ) + + if client_data: + RNS.log( + f"Received {len(client_data)} bytes from client", + RNS.LOG_DEBUG, + ) + self.interface._recv_queue.put(client_data) + + server_data_parts = [] + while True: + try: + server_data_parts.append( + self.interface._send_queue.get_nowait(), + ) + except Empty: + break + + server_data = b"".join(server_data_parts) + if server_data: + RNS.log( + f"Sending {len(server_data)} bytes ({len(server_data_parts)} chunks) to client", + RNS.LOG_DEBUG, + ) + + self.send_response(200) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Content-Length", str(len(server_data))) + self.end_headers() + self.wfile.write(server_data) + else: + self.send_response(404) + self.end_headers() + + def log_message(self, fmt, *args): + """Override to suppress default logging.""" + + return TunnelRequestHandler + + def _start_interface(self): + """Start the interface in either server or client mode.""" + if self.mode == "server": + self._start_server() + else: + self._start_client() + + def _start_server(self): + """Start the HTTP server in a separate thread.""" + + class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): + """Threaded HTTP server for handling multiple connections.""" + + def run_server(): + """Run the HTTP server and handle requests.""" + try: + self._http_server = ThreadedHTTPServer( + (self.listen_host, self.listen_port), + self._request_handler_class, + ) + self.online = True + RNS.log( + f"HTTP server started on http://{self.listen_host}:{self.listen_port}", + RNS.LOG_INFO, + ) + self._http_server.serve_forever() + except Exception as e: + if not self._stop_event.is_set(): + RNS.log(f"Server error: {e}", RNS.LOG_ERROR) + self.online = False + + try: + self._http_server = ThreadedHTTPServer( + (self.listen_host, self.listen_port), + self._request_handler_class, + ) + self.online = True + RNS.log( + f"HTTP server started on http://{self.listen_host}:{self.listen_port}", + RNS.LOG_INFO, + ) + self._worker_thread = Thread( + target=self._http_server.serve_forever, daemon=True + ) + self._worker_thread.start() + except Exception as e: + RNS.log(f"Could not start HTTP server: {e}", RNS.LOG_ERROR) + self.online = False + raise e + + def _start_client(self): + """Start the HTTP client loop in a separate thread.""" + self._worker_thread = Thread(target=self._client_loop, daemon=True) + self._worker_thread.start() + self.online = True + RNS.log(f"HTTP client started, connecting to {self.server_url}", RNS.LOG_INFO) + + def _client_loop(self): + """Main loop for the HTTP client, handling polling and data transfer.""" + while not self._stop_event.is_set(): + data_to_send = b"" + if not self._send_queue.empty(): + from contextlib import suppress + + with suppress(Empty): + data_to_send = self._send_queue.get_nowait() + + try: + RNS.log(f"Sending {len(data_to_send)} bytes to server", RNS.LOG_DEBUG) + response = self.session.post( + self.server_url, + data=data_to_send, + timeout=5, + ) + response.raise_for_status() + + if response.content: + RNS.log( + f"Received {len(response.content)} bytes from server", + RNS.LOG_DEBUG, + ) + self._recv_queue.put(response.content) + + if self._consecutive_failures > 0: + RNS.log("Reconnected to server", RNS.LOG_INFO) + self._consecutive_failures = 0 + + except Exception as e: + self._consecutive_failures += 1 + if self._consecutive_failures % 10 == 1: + RNS.log( + f"Error communicating with server (attempt {self._consecutive_failures}): {e}", + RNS.LOG_ERROR, + ) + + 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 + + sleep(delay) + + def process_incoming(self, data): + """Process incoming data received from the underlying medium.""" + self.rxb += len(data) + self.owner.inbound(data, self) + + def process_outgoing(self, data): + """Process outgoing data to be transmitted by the interface.""" + if self.online: + if len(data) > self.HW_MTU: + RNS.log( + f"Packet too large ({len(data)} > {self.HW_MTU}), dropping", + RNS.LOG_WARNING, + ) + return + + self._send_queue.put(data) + self.txb += len(data) + + def should_ingress_limit(self): + """Indicate that this interface should not perform ingress limiting.""" + return False + + def __str__(self): + """Return a string representation of the HTTPInterface.""" + return f"HTTPInterface[{self.name}]" + + def _read_loop(self): + """Read loop that processes incoming data from the queue.""" + try: + while not self._stop_event.is_set(): + try: + data = self._recv_queue.get(timeout=0.1) + self.process_incoming(data) + except Empty: + continue + except Exception as e: + if not self._stop_event.is_set(): + RNS.log(f"Error in read loop: {e}", RNS.LOG_ERROR) + self.online = False + + +interface_class = HTTPInterface diff --git a/README.md b/README.md index 0173066..9f8b872 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ [Русский](README-RU.md) -A Reticulum interface that tunnels traffic over standard HTTP/S POST requests. This allows Reticulum to operate on networks where only web traffic is permitted, effectively bypassing firewalls, DPI, and other restrictions. +A custom Reticulum interface that tunnels traffic over standard HTTP/S POST requests. This allows Reticulum to operate on networks where only web traffic is permitted, effectively bypassing firewalls, DPI, and other restrictions. [Non-GitHub Mirror](https://lavaforge.org/Ivan/RNS-over-HTTP). Also available on the network `RNS-over-HTTP` node. @@ -40,35 +40,37 @@ This continuous cycle creates a reliable, albeit higher-latency, communication c ### Requirements - Python 3.9 or later -- `pip` for installing packages +- `rns` +- `requests` ### Installation -1. **Install the `requests` library if not already installed:** +1. **Install Reticulum and dependencies:** ```bash - pip install requests + pip install rns requests ``` -2. **Download the interface script:** - Place `http_interface.py` in a known location on both your client and server machines, for example, `~/.reticulum/interfaces/`. +2. **Install the custom interface:** + Place `http_interface.py` in your Reticulum interfaces directory: `~/.reticulum/interfaces/`. ## Configuration -Set up a `PipeInterface` in your `~/.reticulum/config` file on both the server and client machines. +Add an interface entry to your Reticulum configuration file (`~/.reticulum/config`) on both the server and client machines. ### Server Configuration The server listens for incoming connections from clients. ```ini -[[HTTP Interface]] - type = PipeInterface - enabled = True - # The command to run the server script. Listens on all interfaces by default. - command = python3 /path/to/your/http_interface.py server --host 0.0.0.0 --port 8080 - # Optional: delay before respawning the interface if it crashes. - respawn_delay = 5 - name = HTTP Interface Server +[[HTTP Server Interface]] + type = HTTPInterface + enabled = true + mode = server + listen_host = 0.0.0.0 + listen_port = 8080 + mtu = 4096 + check_user_agent = true + user_agent = RNS-HTTP-Tunnel/1.0 ``` ### Client Configuration @@ -76,26 +78,36 @@ The server listens for incoming connections from clients. The client connects to the server's public URL. ```ini -[[HTTP Interface]] - type = PipeInterface - enabled = True - # The command to run the client script. Point --url to your server. - command = python3 /path/to/your/http_interface.py client --url http:// - # Optional: delay before respawning the interface if it crashes. - respawn_delay = 5 - name = HTTP Interface Client +[[HTTP Client Interface]] + type = HTTPInterface + enabled = true + mode = client + server_url = http://your-server-ip-or-domain:8080 + poll_interval = 1.0 + mtu = 4096 + user_agent = RNS-HTTP-Tunnel/1.0 ``` -## Command-Line Options +## Configuration Options -You can customize the behavior of the script with these arguments: +### Common Options -- `--mtu`: Maximum Transmission Unit in bytes (default: `4096`). -- `--poll-interval`: Client polling interval in seconds (default: `0.1`). -- `--verbose` or `-v`: Enable verbose debug logging. -- `--host`: Server listen host (default: `0.0.0.0`). -- `--port`: Server listen port (default: `8080`). -- `--disable-user-agent-check`: Disable User-Agent validation on the server. +- `mtu`: Maximum Transmission Unit in bytes (default: `4096`). +- `name`: Interface name for logging and identification. +- `user_agent`: User-Agent string to use for HTTP requests (default: `"RNS-HTTP-Tunnel/1.0"`). + +### Server Mode Options + +- `mode`: Must be set to `server`. +- `listen_host`: Host to bind the HTTP server to (default: `0.0.0.0`). +- `listen_port`: Port to listen on (default: `8080`). +- `check_user_agent`: Whether to validate User-Agent headers (default: `true`). + +### Client Mode Options + +- `mode`: Must be set to `client`. +- `server_url`: Full URL of the server to connect to (required for client mode). +- `poll_interval`: Polling interval in seconds (default: `1.0`). ## Reverse Proxy Setup (Caddy Example) diff --git a/http_interface.py b/http_interface.py deleted file mode 100644 index 919dd6e..0000000 --- a/http_interface.py +++ /dev/null @@ -1,310 +0,0 @@ -""" -RNS-over-HTTP Interface - -HTTP interface for Reticulum that implements bidirectional communication using HTTP POST requests. -""" - -import logging -import logging.handlers -import sys -import threading -from abc import ABC, abstractmethod -from queue import Queue, Empty -from threading import Thread, Event -from time import sleep -from typing import Iterable -from http.server import HTTPServer, BaseHTTPRequestHandler -from socketserver import ThreadingMixIn -import socket - -import requests - -MTU = 4096 -TUNNEL_USER_AGENT = "RNS-HTTP-Tunnel/1.0" - -def setup_logging(): - logger = logging.getLogger() - logger.setLevel(logging.DEBUG) - - formatter = logging.Formatter( - "%(asctime)s - %(name)s - %(levelname)s - %(message)s" - ) - - console_handler = logging.StreamHandler(sys.stderr) - console_handler.setLevel(logging.INFO) - console_handler.setFormatter(formatter) - - logger.addHandler(console_handler) - -setup_logging() - -class AbstractTunnel(ABC): - def __init__(self, mtu: int): - self.mtu = mtu - self._recv_queue: Queue[bytes] = Queue() - self._send_queue: Queue[bytes] = Queue() - self._stop_event = Event() - self.logger = logging.getLogger(self.__class__.__name__) - - def send(self, pkt: bytes) -> None: - if len(pkt) > self.mtu: - raise ValueError(f"payload too large ({len(pkt)} > {self.mtu})") - self._send_queue.put(pkt) - - def recv(self) -> Iterable[bytes]: - while True: - yield self._recv_queue.get(block=True) - - @abstractmethod - def start(self) -> None: - pass - - @abstractmethod - def stop(self) -> None: - pass - -class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): - pass - -class Server(AbstractTunnel): - def __init__(self, listen_host: str, listen_port: int, mtu: int, check_user_agent: bool = True): - super().__init__(mtu) - self.listen_host = listen_host - self.listen_port = listen_port - self.check_user_agent = check_user_agent - self._server_thread: Thread | None = None - self._http_server: HTTPServer | None = None - - class TunnelRequestHandler(BaseHTTPRequestHandler): - def __init__(self, request, client_address, server, tunnel_instance=self): - self.tunnel = tunnel_instance - super().__init__(request, client_address, server) - - def do_POST(self): - if self.path == "/": - if self.tunnel.check_user_agent: - user_agent = self.headers.get('User-Agent', '') - if user_agent != TUNNEL_USER_AGENT: - self.tunnel.logger.warning(f"Rejected request with invalid User-Agent: {user_agent}") - self.send_response(403) - self.send_header('Content-Type', 'text/plain') - self.end_headers() - self.wfile.write(b'Forbidden') - return - - content_length = int(self.headers.get('Content-Length', 0)) - client_data = self.rfile.read(content_length) if content_length > 0 else b"" - - if client_data: - self.tunnel.logger.debug(f"Received {len(client_data)} bytes from client") - self.tunnel._recv_queue.put(client_data) - - server_data_parts = [] - while not self.tunnel._send_queue.empty(): - try: - server_data_parts.append(self.tunnel._send_queue.get_nowait()) - except Empty: - break - - server_data = b"".join(server_data_parts) - if server_data: - self.tunnel.logger.debug(f"Sending {len(server_data)} bytes ({len(server_data_parts)} chunks) to client") - - self.send_response(200) - self.send_header('Content-Type', 'application/octet-stream') - self.send_header('Content-Length', str(len(server_data))) - self.end_headers() - self.wfile.write(server_data) - else: - self.send_response(404) - self.end_headers() - - def log_message(self, format, *args): - pass - - self._request_handler_class = TunnelRequestHandler - - def start(self) -> None: - def run_server(): - try: - self._http_server = ThreadedHTTPServer((self.listen_host, self.listen_port), self._request_handler_class) - self._http_server.serve_forever() - except Exception as e: - if not self._stop_event.is_set(): - self.logger.error(f"Server error: {e}") - - self._server_thread = Thread(target=run_server, daemon=True) - self._server_thread.start() - self.logger.info(f"HTTP server started on http://{self.listen_host}:{self.listen_port}") - - def stop(self) -> None: - self.logger.info("Stopping HTTP server...") - self._stop_event.set() - if self._http_server: - self._http_server.shutdown() - self._http_server.server_close() - if self._server_thread: - self._server_thread.join(timeout=2) - -class Client(AbstractTunnel): - def __init__(self, server_url: str, mtu: int, poll_interval: float = 1.0): - super().__init__(mtu) - self.server_url = server_url - self.poll_interval = poll_interval - self._client_thread: Thread | None = None - self.session = requests.Session() - self.session.headers.update({'User-Agent': TUNNEL_USER_AGENT}) - self._consecutive_failures = 0 - self._max_backoff = 30.0 # Maximum backoff time in seconds - - def start(self) -> None: - self._client_thread = Thread(target=self._run, daemon=True) - self._client_thread.start() - self.logger.info(f"HTTP client started, connecting to {self.server_url}") - - def stop(self) -> None: - self.logger.info("Stopping HTTP client...") - self._stop_event.set() - if self._client_thread: - self._client_thread.join(timeout=2) - - def _run(self): - while not self._stop_event.is_set(): - data_to_send = b"" - if not self._send_queue.empty(): - try: - data_to_send = self._send_queue.get_nowait() - except Empty: - pass - - try: - self.logger.debug(f"Sending {len(data_to_send)} bytes to server") - response = self.session.post(self.server_url, data=data_to_send, timeout=5) - response.raise_for_status() - - if response.content: - self.logger.debug(f"Received {len(response.content)} bytes from server") - self._recv_queue.put(response.content) - - if self._consecutive_failures > 0: - self.logger.info("Reconnected to server") - self._consecutive_failures = 0 - - except requests.exceptions.RequestException as e: - self._consecutive_failures += 1 - if self._consecutive_failures % 10 == 1: - self.logger.error(f"Error communicating with server (attempt {self._consecutive_failures}): {e}") - - 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 - - sleep(delay) - - -if __name__ == "__main__": - import argparse - import os - import errno - - logger = logging.getLogger(__name__) - - parser = argparse.ArgumentParser(description="HTTP Tunnel - Server/Client") - parser.add_argument("mode", choices=["server", "client"], help="Run mode: server or client") - parser.add_argument("--mtu", type=int, default=MTU, help=f"MTU size (default: {MTU})") - parser.add_argument("--host", type=str, default="0.0.0.0", help="Listen host (for server mode)") - parser.add_argument("--port", type=int, default=8080, help="Listen port (for server mode)") - parser.add_argument("--url", type=str, help="Server URL (required for client mode)") - parser.add_argument("--poll-interval", type=float, default=0.1, help="Client poll interval in seconds") - parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose logging") - parser.add_argument("--disable-user-agent-check", action="store_true", help="Disable User-Agent validation (server mode only)") - - args = parser.parse_args() - - if args.verbose: - for handler in logging.getLogger().handlers: - if isinstance(handler, logging.StreamHandler): - handler.setLevel(logging.DEBUG) - - def receive_messages(tunnel, stop_event): - stdout_fd = sys.stdout.fileno() - try: - for received_data in tunnel.recv(): - if stop_event.is_set(): - break - if received_data: - try: - os.write(stdout_fd, received_data) - except OSError as e: - if e.errno == errno.EPIPE: - stop_event.set() - return - else: - logger.error("Error writing to stdout: %s", e) - stop_event.set() - return - except Exception: - if not stop_event.is_set(): - logger.error("Error in receive thread", exc_info=True) - return - - def read_stdin_bytes(): - try: - return os.read(0, 4096) - except (IOError, OSError): - return None - - try: - if args.mode == "server": - server = Server(listen_host=args.host, listen_port=args.port, mtu=args.mtu, check_user_agent=not args.disable_user_agent_check) - server.start() - - stop_event = threading.Event() - receive_thread = threading.Thread(target=receive_messages, args=(server, stop_event), daemon=True) - receive_thread.start() - - try: - while not stop_event.is_set(): - message = read_stdin_bytes() - if message: - server.send(message) - else: - sleep(0.01) - - except KeyboardInterrupt: - logger.info("Stopping server...") - finally: - stop_event.set() - server.stop() - receive_thread.join(timeout=1) - - elif args.mode == "client": - if not args.url: - parser.error("--url is required for client mode") - - client = Client(server_url=args.url, mtu=args.mtu, poll_interval=args.poll_interval) - client.start() - - stop_event = threading.Event() - receive_thread = threading.Thread(target=receive_messages, args=(client, stop_event), daemon=True) - receive_thread.start() - - try: - while not stop_event.is_set(): - message = read_stdin_bytes() - if message: - client.send(message) - else: - sleep(0.01) - - except KeyboardInterrupt: - logger.info("Stopping client...") - finally: - stop_event.set() - client.stop() - receive_thread.join(timeout=1) - - except Exception as e: - logger.error(f"A critical error occurred: {e}", exc_info=True) - sys.exit(1)