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
