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")

Kes Pembroke