123 lines
6.1 KiB
Python
123 lines
6.1 KiB
Python
"""Check dpi.py and evade.py on small records you can check by hand. Given; do not change it.
|
|
|
|
python3 selfcheck.py
|
|
|
|
Each check prints PASS, FAIL (with what it expected), or SKIP for a function
|
|
not written yet. Passing every check means your functions do what the
|
|
docstrings ask. It does not grade your report.
|
|
"""
|
|
|
|
import corpus
|
|
import dpi
|
|
import evade
|
|
|
|
results = []
|
|
|
|
CENSOR = dpi.Censor(hosts=["freeleaf-news.example"], ips={"192.0.2.50"}, keywords=["protest"])
|
|
|
|
|
|
def rec(**kw):
|
|
base = {"id": 0, "port": 443, "dns": "weather-now.example", "ip": "203.0.113.10",
|
|
"sni": "weather-now.example", "hello_bytes": 400, "payload": "a3f09c"}
|
|
base.update(kw)
|
|
return base
|
|
|
|
|
|
def http(host, text, **kw):
|
|
payload = f"GET / HTTP/1.1\r\nHost: {host}\r\n\r\nHTTP/1.1 200 OK\r\n\r\n<html><body>{text}</body></html>"
|
|
return rec(port=80, dns=host, sni=None, hello_bytes=0, payload=payload, **kw)
|
|
|
|
|
|
def check(name, fn):
|
|
try:
|
|
fn()
|
|
results.append("PASS")
|
|
print(f"PASS {name}")
|
|
except NotImplementedError as e:
|
|
results.append("SKIP")
|
|
print(f"SKIP {name} ({e} not written yet)")
|
|
except AssertionError as e:
|
|
results.append("FAIL")
|
|
print(f"FAIL {name}: {e}")
|
|
|
|
|
|
def t_name_matches():
|
|
hosts = ["freeleaf-news.example"]
|
|
for name, want in [("freeleaf-news.example", True), ("www.freeleaf-news.example", True),
|
|
("a.b.freeleaf-news.example", True), ("FreeLeaf-News.EXAMPLE", True),
|
|
("freeleaf-news.example.", True), ("notfreeleaf-news.example", False),
|
|
("freeleaf-news.example.org", False), ("weather-now.example", False), (None, False)]:
|
|
got = dpi.name_matches(name, hosts)
|
|
assert got is want or got == want, f"name_matches({name!r}, {hosts}) should be {want}, got {got}"
|
|
|
|
|
|
def t_dns():
|
|
assert dpi.dns_filter(rec(dns="www.freeleaf-news.example"), CENSOR), "a lookup of www.freeleaf-news.example should be blocked"
|
|
assert not dpi.dns_filter(rec(), CENSOR), "a lookup of weather-now.example should pass"
|
|
assert not dpi.dns_filter(rec(dns=None, sni="freeleaf-news.example"), CENSOR), (
|
|
"with no lookup seen, the DNS filter has nothing to block on, even if the SNI names the site")
|
|
|
|
|
|
def t_ip():
|
|
assert dpi.ip_filter(rec(ip="192.0.2.50", dns="weather-now.example"), CENSOR), (
|
|
"192.0.2.50 is on the censor's list, whatever name the record carries")
|
|
assert not dpi.ip_filter(rec(dns="freeleaf-news.example", sni="freeleaf-news.example"), CENSOR), (
|
|
"203.0.113.10 is not on this censor's list, so the IP filter lets it through even for a blocked name")
|
|
|
|
|
|
def t_sni():
|
|
assert dpi.sni_filter(rec(sni="freeleaf-news.example", dns=None), CENSOR), "SNI freeleaf-news.example should be blocked"
|
|
assert not dpi.sni_filter(rec(sni="ech.cdn-a.example", dns="freeleaf-news.example"), CENSOR), (
|
|
"the SNI filter reads only the SNI; ech.cdn-a.example is not a blocked name")
|
|
assert not dpi.sni_filter(http("freeleaf-news.example", "hello"), CENSOR), (
|
|
"plain HTTP has no ClientHello (sni is None), so there is nothing for the SNI filter to read")
|
|
|
|
|
|
def t_keyword():
|
|
assert dpi.keyword_filter(http("daily-herald.example", "Workers PROTEST wage cuts"), CENSOR), (
|
|
"'PROTEST' in a readable page should match the keyword 'protest'")
|
|
assert not dpi.keyword_filter(http("daily-herald.example", "Local team wins the cup"), CENSOR), (
|
|
"a page without the keyword should pass")
|
|
assert not dpi.keyword_filter(rec(payload="00ff" * 50, dns="protest.example"), CENSOR), (
|
|
"the keyword filter reads the payload only; the name in the dns field is not payload")
|
|
|
|
|
|
def t_hide_sni():
|
|
r = rec(dns="freeleaf-news.example", sni="www.freeleaf-news.example")
|
|
out = evade.hide_sni(r)
|
|
assert out["sni"] == corpus.ECH_PUBLIC_NAME["203.0.113.10"], f"on CDN A the SNI becomes ech.cdn-a.example, got {out['sni']!r}"
|
|
assert out["dns"] == r["dns"] and out["ip"] == r["ip"] and out["payload"] == r["payload"], "only the SNI changes"
|
|
assert r["sni"] == "www.freeleaf-news.example", "hide_sni must return a copy, not change the record it was given"
|
|
b = rec(ip="198.51.100.20", sni="rightswatch.example")
|
|
assert evade.hide_sni(b) == b, "CDN B does not offer ECH, so the record comes back unchanged"
|
|
h = http("freeleaf-news.example", "hello")
|
|
assert evade.hide_sni(h) == h, "plain HTTP has no ClientHello to hide, so it comes back unchanged"
|
|
|
|
|
|
def t_use_doh():
|
|
r = rec(dns="freeleaf-news.example", sni="freeleaf-news.example")
|
|
out = evade.use_doh(r)
|
|
assert out["dns"] is None, f"with DNS over HTTPS the censor sees no lookup: dns should be None, got {out['dns']!r}"
|
|
assert out["sni"] == r["sni"] and out["ip"] == r["ip"], "only the lookup changes"
|
|
assert r["dns"] == "freeleaf-news.example", "use_doh must return a copy"
|
|
|
|
|
|
def t_via_proxy():
|
|
h = http("freeleaf-news.example", "Thousands join the protest", id=7)
|
|
out = evade.via_proxy(h)
|
|
p = corpus.PROXY
|
|
assert (out["dns"], out["ip"], out["sni"]) == (p["host"], p["ip"], p["host"]), (
|
|
f"through the proxy the censor sees {p['host']} at {p['ip']}; got dns={out['dns']!r}, ip={out['ip']!r}, sni={out['sni']!r}")
|
|
assert out["port"] == 443 and out["id"] == 7, "the proxy connection is HTTPS (port 443) and keeps the record's id"
|
|
assert "protest" not in out["payload"].lower() and "freeleaf" not in out["payload"].lower(), (
|
|
"inside the proxy the page is encrypted; its words must not be readable in the payload")
|
|
assert set(corpus.records()[0]) == set(out), f"the new record should have exactly the corpus keys, got {sorted(out)}"
|
|
assert h["port"] == 80, "via_proxy must return a new record, not change the one it was given"
|
|
|
|
|
|
for name, fn in [("name_matches", t_name_matches), ("dns_filter", t_dns), ("ip_filter", t_ip),
|
|
("sni_filter", t_sni), ("keyword_filter", t_keyword), ("hide_sni", t_hide_sni),
|
|
("use_doh", t_use_doh), ("via_proxy", t_via_proxy)]:
|
|
check(name, fn)
|
|
print(f"\n{results.count('PASS')} passed, {results.count('FAIL')} failed, {results.count('SKIP')} not written yet")
|
|
raise SystemExit(1 if "FAIL" in results else 0)
|