Hello Pluggable Transports / Tor Project team, I am experimenting with pluggable transport designs and would greatly appreciate your review, security guidance, and advice on proper integration with Tor's pluggable-transport ecosystem. This email includes a short technical summary, exact prototype code (client + server), running instructions, suggested realistic hardening/padding/timing, and a minimal stdin/stdout PT wrapper skeleton so this can be adapted into a real ClientTransportPlugin. Please treat the attached code and descriptions as an educational research prototype — it is **not production-ready**. **Key goals** - Prototype a PT that mimics browser extension filter-list update checks (ad‑blocker‑like traffic) to carry an encrypted tunneled payload. - Provide a small test harness so Tor Project reviewers can run the demo locally. - Ask for guidance on whether this façade is appropriate, safe, and how best to integrate with Tor's PT interface (packaging, authentication, key exchange, deployment concerns). **Important legal & ethical note (please read)** - This code is a proof of concept for research only. Do not deploy it against third-party infrastructure, impersonate real vendors, or use domains/certificates you do not control. - I welcome explicit feedback if the Tor Project considers the approach harmful, unsafe, or unacceptable for inclusion as an official PT (I will stop development if advised). --- ## Technical summary (short) - Transport: TLS over TCP, shaped to look like browser extension update checks (e.g., `POST /updates/check`, JSON response keys `update_version`, `list_name`, `payload_b64`). - Encryption & framing: AES‑GCM (demo uses a pre-shared key). Frame format = 4‑byte big‑endian length prefix || (nonce || ciphertext+tag). Frames are base64-embedded in HTTP bodies/JSON to resemble benign payloads. - Prototype language: Python 3, asyncio, `cryptography` library for AES‑GCM. - Prototype features: TLS server + client example, framing/unframing, JSON wrapping, echo server to validate round-trip, and instructions to add jitter/padding/fragmentation. - Integration note: Prototype uses a static shared key for simplicity. Real PTs should use ephemeral ECDH or Tor-provided authentication tokens and follow Tor PT handshake requirements. --- ## Files included inline (copy/paste into files) ### 1) Server — `adblock_pt_server.py` ```python #!/usr/bin/env python3 """ Prototype "adblock-like" pluggable transport server (demo). - TLS server listening on host:port. - Accepts simple HTTP-like requests, expects base64-framed AES-GCM payloads. - Decrypts, prints, and echoes an encrypted framed reply inside a JSON update response. WARNING: demo only. Use certs/domains you control. Do not impersonate real update servers. Dependencies: cryptography """ import asyncio, ssl, argparse, json, os, struct, base64 from cryptography.hazmat.primitives.ciphers.aead import AESGCM # ----- CONFIG (demo) ----- SHARED_KEY = b'\x01' * 32 # demo key: replace with secure key exchange in production # -------------------------- def frame_message(data: bytes) -> bytes: return struct.pack('!I', len(data)) + data def unframe_stream(buf: bytearray): while True: if len(buf) < 4: break length = struct.unpack('!I', bytes(buf[:4]))[0] if len(buf) < 4 + length: break payload = bytes(buf[4:4+length]) del buf[:4+length] yield payload async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, aesgcm: AESGCM): peer = writer.get_extra_info('peername') print(f"[+] Connection from {peer}") try: data = await reader.readuntil(b"\r\n\r\n") except asyncio.IncompleteReadError: writer.close() await writer.wait_closed() return headers = data.decode(errors='ignore').split("\r\n") request_line = headers[0] if headers else "" method, path, _ = (request_line.split(" ", 2) + ["", ""])[:3] header_map = {} for h in headers[1:]: if ':' in h: k, v = h.split(':',1) header_map[k.strip().lower()] = v.strip() body = b"" if 'content-length' in header_map: try: length = int(header_map['content-length']) if length: body = await reader.readexactly(length) except Exception: pass try: raw = base64.b64decode(body) if body else b'' buf = bytearray(raw) outputs = [] for frame in unframe_stream(buf): nonce = frame[:12] ct = frame[12:] pt = aesgcm.decrypt(nonce, ct, None) print(f" -> Received framed payload (len {len(pt)}): {pt!r}") outputs.append(pt) reply_plain = b"".join(outputs) or b"PING" nonce = os.urandom(12) ct = aesgcm.encrypt(nonce, reply_plain, None) framed = nonce + ct payload = frame_message(framed) payload_b64 = base64.b64encode(payload).decode() resp_body = json.dumps({ "update_version": "2025-10-21", "list_name": "easy-filters-proto", "payload_b64": payload_b64 }).encode() resp = b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: " + str(len(resp_body)).encode() + b"\r\n\r\n" + resp_body writer.write(resp) await writer.drain() except Exception as e: print("Error handling payload:", e) writer.write(b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n") await writer.drain() writer.close() await writer.wait_closed() print(f"[-] Connection from {peer} closed") async def main(host, port, certfile, keyfile): aesgcm = AESGCM(SHARED_KEY) sslctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) sslctx.load_cert_chain(certfile=certfile, keyfile=keyfile) server = await asyncio.start_server(lambda r,w: handle_client(r,w,aesgcm), host, port, ssl=sslctx) addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets) print(f"Serving on {addrs}") async with server: await server.serve_forever() if __name__ == "__main__": p = argparse.ArgumentParser() p.add_argument('--host', default='0.0.0.0') p.add_argument('--port', type=int, default=8443) p.add_argument('--cert', default='server.crt') p.add_argument('--key', default='server.key') args = p.parse_args() try: asyncio.run(main(args.host, args.port, args.cert, args.key)) except KeyboardInterrupt: print("Server stopped") Client — adblock_pt_client.py #!/usr/bin/env python3 """ Prototype client: connects over TLS, sends one framed AES-GCM payload embedded as base64 in the HTTP body, and prints decrypted framed reply from server. Dependencies: cryptography """ import asyncio, ssl, argparse, json, os, struct, base64, time from cryptography.hazmat.primitives.ciphers.aead import AESGCM SHARED_KEY = b'\x01' * 32 # demo key; replace with secure key exchange in production def frame_message(data: bytes) -> bytes: return struct.pack('!I', len(data)) + data def unframe_stream(buf: bytearray): while True: if len(buf) < 4: break length = struct.unpack('!I', bytes(buf[:4]))[0] if len(buf) < 4 + length: break payload = bytes(buf[4:4+length]) del buf[:4+length] yield payload async def run_client(host, port, cafile=None): aesgcm = AESGCM(SHARED_KEY) sslctx = ssl.create_default_context() if cafile: sslctx.load_verify_locations(cafile) else: sslctx.check_hostname = False sslctx.verify_mode = ssl.CERT_NONE reader, writer = await asyncio.open_connection(host, port, ssl=sslctx) my_payload = b"HELLO-TOR-PT: test at %d" % int(time.time()) nonce = os.urandom(12) ct = aesgcm.encrypt(nonce, my_payload, None) framed = nonce + ct payload = frame_message(framed) payload_b64 = base64.b64encode(payload).decode() body = payload_b64.encode() req = (f"POST /updates/check HTTP/1.1\r\n" f"Host: updates.adblock-proto.example\r\n" f"User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) ExtensionUpdate/1.2\r\n" f"Accept: application/json\r\n" f"Content-Type: application/octet-stream\r\n" f"Content-Length: {len(body)}\r\n" f"Connection: close\r\n\r\n").encode() + body writer.write(req) await writer.drain() try: header_data = await reader.readuntil(b"\r\n\r\n") except asyncio.IncompleteReadError: print("No response headers") writer.close() await writer.wait_closed() return header_text = header_data.decode(errors='ignore') content_length = 0 for line in header_text.split("\r\n"): if line.lower().startswith("content-length:"): content_length = int(line.split(":",1)[1].strip()) resp_body = b"" if content_length: resp_body = await reader.readexactly(content_length) try: j = json.loads(resp_body.decode()) if 'payload_b64' in j: server_payload_b64 = j['payload_b64'].encode() raw = base64.b64decode(server_payload_b64) buf = bytearray(raw) for frame in unframe_stream(buf): nonce = frame[:12] ct = frame[12:] pt = aesgcm.decrypt(nonce, ct, None) print("[+] Received tunnel reply:", pt) except Exception as e: print("Failed to parse server response:", e, resp_body) writer.close() await writer.wait_closed() if __name__ == "__main__": p = argparse.ArgumentParser() p.add_argument('--host', default='127.0.0.1') p.add_argument('--port', type=int, default=8443) p.add_argument('--cafile', default=None) args = p.parse_args() try: asyncio.run(run_client(args.host, args.port, args.cafile)) except KeyboardInterrupt: pass Minimal stdin/stdout PT wrapper skeleton (ClientTransportPlugin style) This skeleton demonstrates the basic pattern where Tor launches a PT binary and communicates via stdin/stdout. Real PTs follow Tor’s PT handshake protocol — this is a minimal skeleton to adapt the above transport into a tokable PT process. DO NOT use as-is for production; consult Tor docs for required messages and handshake fields. #!/usr/bin/env python3 """ Very small PT stdin/stdout skeleton (demo). Tor launches PT and exchanges control messages on stdin/stdout. This skeleton demonstrates where to tie in the upstream SOCKS<->transport logic. """ import sys, json, subprocess, os def send_response(obj): s = json.dumps(obj) sys.stdout.write(s + "\n") sys.stdout.flush() def read_request(): line = sys.stdin.readline() if not line: return None return json.loads(line.strip()) def main(): # Example: Tor may send a 'version' / 'init' style message first (real protocol differs). # Read initial message req = read_request() if req is None: return # Inspect request and respond (example handshake) send_response({"status": "ok", "proto": "adblock-proto-demo", "version": "0.1"}) # In a real wrapper: # - parse Tor's assigned socks port / client address # - start a local listener that accepts a single connection, then forward data from socks -> encrypted HTTP pseudo-updates via the client transport above. # - implement multiplexing, proper framing, timeouts, logging, and robust error handling. while True: # keep alive or wait for Tor to kill process line = sys.stdin.readline() if not line: break if __name__ == "__main__": main() Suggested realism / hardening steps (non-exhaustive) - Replace static pre-shared key with ephemeral ECDH (X25519) handshake per-session or use Tor-provided authentication tokens. - Implement per-connection padding and body-size shaping to match observed filter-list update sizes (collect benign samples to model). - Add jittered periodic checks (e.g., pseudo-update every X hours, randomized within window). - Fragment larger Tor cells across multiple pseudo-update requests; reassemble server-side. - Use valid certs for domains you control; do not impersonate legitimate extension vendors. - Implement rate-limiting, connection reuse (keep-alive), and HTTP header variability to match multiple extension implementations. - Consider how middleboxes might fingerprint TLS features (ALPN, TLS versions, ciphers); try to match common browser/TLS stacks of real extension update servers you control. - Add server-side logging that preserves privacy and limits sensitive plaintext exposure (log only lengths/timestamps, not decrypted payloads). ------------------------------ Questions I have for you / requested guidance 1. Is this façade approach acceptable for a Tor pluggable transport from the Tor Project’s perspective (network safety, abuse potential, collateral damage)? 2. Are there existing PT design constraints / packaging rules I should follow (besides the obvious obfs4/meek references)? Any docs you recommend for integrating with the ClientTransportPlugin flow and signing/release packaging? 3. Any security pitfalls I missed (e.g., observable TLS fingerprint mismatches, directory authorities concerns, recommended key exchange patterns)? Kes Pembroke