deep-packet-inspection/corpus.py
2026-10-19 08:42:33 -06:00

172 lines
7.9 KiB
Python

"""The traffic: a made-up city's connections for one afternoon. Given; do not change it.
Everything is synthetic. The 40 sites use the reserved .example domain and
the IP ranges set aside for documentation, so nothing here names a real site
or a real address.
records() the connections, one dict each (the format is below)
truth() the answer key: the site each record really visits, row for
row with records(). For scoring in experiments.py only; a
censor never has it, so dpi.py and evade.py must not use it.
resolve(name) public DNS: the address a name points to. The censor may
use this, as any real censor can.
The censor's targets are BLOCKLIST (sites) and KEYWORDS (words it looks for
in readable traffic).
Record format (every record has every key):
id 0, 1, 2, ...
port 80 (plain HTTP, readable) or 443 (HTTPS, encrypted)
dns the name the client looked up before connecting, or None when
no lookup was seen (the answer was cached)
ip the destination address, a string such as "203.0.113.10"
sni the server name from the TLS ClientHello, which is sent before
encryption starts; None for port 80
hello_bytes size of the ClientHello in bytes; 0 for port 80
payload for port 80, the request and the page text, readable;
for port 443, hex digits standing in for ciphertext
Things that happen in real traffic and happen here too: several sites share
one CDN address; clients visit "www." and other subdomains; some resolvers
randomize the upper and lower case of the names they look up; and some
clients on CDN A already use Encrypted Client Hello (ECH), so their SNI is the
CDN's public name instead of the site's.
This mirrors the setting of Zohaib et al. (2025), the week 10 reading.
"""
import random
BLOCKLIST = ["freeleaf-news.example", "rightswatch.example", "openvpn-mirror.example"]
KEYWORDS = ["freeleaf", "protest"]
CDN_A = "203.0.113.10"
CDN_B = "198.51.100.20"
ECH_PUBLIC_NAME = {CDN_A: "ech.cdn-a.example"} # CDN B does not offer ECH
PROXY = {"host": "relay-7.example", "ip": "192.0.2.200"}
# site: (address, share of its visits over plain HTTP, relative popularity)
SITES = {
# CDN A: twelve sites behind one address
"freeleaf-news.example": (CDN_A, 0.3, 5),
"weather-now.example": (CDN_A, 0.1, 6),
"recipe-box.example": (CDN_A, 0.1, 3),
"city-library.example": (CDN_A, 0.1, 2),
"school-portal.example": (CDN_A, 0.0, 5),
"clinic-booking.example": (CDN_A, 0.0, 2),
"bike-shop.example": (CDN_A, 0.1, 1),
"pet-rescue.example": (CDN_A, 0.1, 1),
"tax-help.example": (CDN_A, 0.0, 2),
"jobs-board.example": (CDN_A, 0.1, 3),
"photo-share.example": (CDN_A, 0.0, 4),
"music-stream.example": (CDN_A, 0.0, 5),
# CDN B: eight sites behind another
"rightswatch.example": (CDN_B, 0.2, 4),
"math-tutor.example": (CDN_B, 0.1, 2),
"farm-market.example": (CDN_B, 0.2, 1),
"bus-times.example": (CDN_B, 0.1, 4),
"chess-club.example": (CDN_B, 0.2, 1),
"open-courseware.example": (CDN_B, 0.1, 2),
"car-parts.example": (CDN_B, 0.1, 1),
"dentist-office.example": (CDN_B, 0.1, 1),
# twenty sites with an address each
"openvpn-mirror.example": ("192.0.2.50", 0.5, 3),
"ministry-info.example": ("192.0.2.11", 1.0, 2),
"daily-herald.example": ("192.0.2.12", 0.5, 5),
"city-history.example": ("192.0.2.13", 1.0, 1),
"wiki-encyclopedia.example": ("192.0.2.14", 0.3, 6),
"forum-hub.example": ("192.0.2.15", 0.5, 3),
"search-engine.example": ("192.0.2.16", 0.0, 12),
"video-site.example": ("192.0.2.17", 0.0, 10),
"social-feed.example": ("192.0.2.18", 0.0, 10),
"bank-online.example": ("192.0.2.19", 0.0, 4),
"mail-host.example": ("192.0.2.20", 0.0, 6),
"maps-app.example": ("192.0.2.21", 0.0, 4),
"shop-mall.example": ("192.0.2.22", 0.1, 5),
"sports-scores.example": ("192.0.2.23", 0.2, 4),
"game-store.example": ("192.0.2.24", 0.0, 3),
"cloud-drive.example": ("192.0.2.25", 0.0, 3),
"translate-tool.example": ("192.0.2.26", 0.1, 2),
"podcast-feed.example": ("192.0.2.27", 0.3, 2),
"dating-app.example": ("192.0.2.28", 0.0, 2),
"hospital-portal.example": ("192.0.2.29", 0.0, 2),
}
# Readable pages. A site not listed here serves only its generic pages.
PAGES = {
"freeleaf-news.example": [("/", "Freeleaf News: independent reporting"),
("/2026/rally", "Thousands join the protest downtown")],
"rightswatch.example": [("/", "RightsWatch: documenting detentions"),
("/reports/q3", "Report on arrests after the September protest")],
"openvpn-mirror.example": [("/", "Mirror of VPN client downloads"),
("/download", "client installer, version 2.6")],
"ministry-info.example": [("/", "Ministry of Information announcements"),
("/statement", "The ministry warns citizens that Freeleaf spreads rumors")],
"daily-herald.example": [("/", "Daily Herald front page"),
("/city/strike", "Transit workers protest wage cuts"),
("/sports", "Local team wins the cup")],
"city-history.example": [("/", "Our city, 1850 to today"),
("/1968", "The 1968 student protest and what followed")],
"wiki-encyclopedia.example": [("/wiki/Protest", "A protest is a public expression of objection"),
("/wiki/Tea", "Tea is an aromatic beverage")],
"forum-hub.example": [("/t/gardening", "Best time to plant tomatoes?"),
("/t/news", "Anyone going to the protest on Saturday?")],
}
GENERIC = [("/", "Welcome"), ("/about", "About us"), ("/contact", "Contact and opening hours")]
def _case_scramble(name, rng):
return "".join(c.upper() if rng.random() < 0.5 else c for c in name)
def _generate(seed=4930, n=600):
rng = random.Random(seed)
names = list(SITES)
weights = [SITES[s][2] for s in names]
recs, truth = [], []
for i in range(n):
site = rng.choices(names, weights)[0]
ip, http_share, _ = SITES[site]
name = rng.choice(["", "", "www.", "www.", "cdn.", "api."]) + site
looked_up = name if rng.random() < 0.85 else None
if looked_up and rng.random() < 0.1:
looked_up = _case_scramble(looked_up, rng)
if rng.random() < http_share:
path, text = rng.choice(PAGES.get(site, GENERIC))
payload = (f"GET {path} HTTP/1.1\r\nHost: {name}\r\nUser-Agent: Mozilla/5.0\r\n\r\n"
f"HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n<html><body>{text}</body></html>")
rec = {"port": 80, "dns": looked_up, "ip": ip, "sni": None, "hello_bytes": 0, "payload": payload}
else:
sni = name
if ip in ECH_PUBLIC_NAME and site not in BLOCKLIST and rng.random() < 0.25:
sni = ECH_PUBLIC_NAME[ip]
size = rng.randint(300, 1500)
rec = {"port": 443, "dns": looked_up, "ip": ip, "sni": sni, "hello_bytes": rng.randint(240, 620),
"payload": format(rng.getrandbits(8 * size), "x").zfill(2 * size)}
rec = {"id": i, **rec}
recs.append(rec)
truth.append(site)
return recs, truth
_RECORDS, _TRUTH = _generate()
def records():
"""The connections. Each call returns fresh copies, so changing one does not change the corpus."""
return [dict(r) for r in _RECORDS]
def truth():
"""The answer key, row for row with records(). For scoring only."""
return list(_TRUTH)
_DNS = {prefix + site: ip for site, (ip, _, _) in SITES.items() for prefix in ("", "www.", "cdn.", "api.")}
_DNS[PROXY["host"]] = PROXY["ip"]
def resolve(name):
"""Public DNS: the address a name such as "www.weather-now.example" points to, or None if unknown."""
return _DNS.get(name)