mirror of
https://github.com/vee1e/mitmproxy.git
synced 2026-09-01 18:27:18 +00:00
strip HTTPS records ALPN if http3 is disabled (#7185)
* strip HTTPS records ALPN if http3 is disabled * Update strip_dns_https_records.py * Update strip_dns_https_records.py * [autofix.ci] apply automated fixes * Update strip_dns_https_records.py * [autofix.ci] apply automated fixes * Update strip_dns_https_records.py --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Maximilian Hils
GitHub
parent
8ff173a0aa
commit
8964deda8a
6 changed files with 123 additions and 25 deletions
|
|
@ -25,7 +25,7 @@ from mitmproxy.addons import script
|
|||
from mitmproxy.addons import serverplayback
|
||||
from mitmproxy.addons import stickyauth
|
||||
from mitmproxy.addons import stickycookie
|
||||
from mitmproxy.addons import strip_ech
|
||||
from mitmproxy.addons import strip_dns_https_records
|
||||
from mitmproxy.addons import tlsconfig
|
||||
from mitmproxy.addons import update_alt_svc
|
||||
from mitmproxy.addons import upstream_auth
|
||||
|
|
@ -36,7 +36,7 @@ def default_addons():
|
|||
core.Core(),
|
||||
browser.Browser(),
|
||||
block.Block(),
|
||||
strip_ech.StripECH(),
|
||||
strip_dns_https_records.StripDnsHttpsRecords(),
|
||||
blocklist.BlockList(),
|
||||
anticache.AntiCache(),
|
||||
anticomp.AntiComp(),
|
||||
|
|
|
|||
37
mitmproxy/addons/strip_dns_https_records.py
Normal file
37
mitmproxy/addons/strip_dns_https_records.py
Normal file
|
|
@ -0,0 +1,37 @@
|
|||
from mitmproxy import ctx
|
||||
from mitmproxy import dns
|
||||
from mitmproxy.net.dns import types
|
||||
|
||||
|
||||
class StripDnsHttpsRecords:
|
||||
def load(self, loader):
|
||||
loader.add_option(
|
||||
"strip_ech",
|
||||
bool,
|
||||
True,
|
||||
"Strip Encrypted ClientHello (ECH) data from DNS HTTPS records so that mitmproxy can generate matching certificates.",
|
||||
)
|
||||
|
||||
def dns_response(self, flow: dns.DNSFlow):
|
||||
assert flow.response
|
||||
if ctx.options.strip_ech:
|
||||
for answer in flow.response.answers:
|
||||
if answer.type == types.HTTPS:
|
||||
answer.https_ech = None
|
||||
if not ctx.options.http3:
|
||||
for answer in flow.response.answers:
|
||||
if (
|
||||
answer.type == types.HTTPS
|
||||
and answer.https_alpn is not None
|
||||
and any(
|
||||
# HTTP/3 or any of the spec drafts (h3-...)?
|
||||
a == b"h3" or a.startswith(b"h3-")
|
||||
for a in answer.https_alpn
|
||||
)
|
||||
):
|
||||
alpns = tuple(
|
||||
a
|
||||
for a in answer.https_alpn
|
||||
if a != b"h3" and not a.startswith(b"h3-")
|
||||
)
|
||||
answer.https_alpn = alpns or None
|
||||
|
|
@ -1,20 +0,0 @@
|
|||
from mitmproxy import ctx
|
||||
from mitmproxy import dns
|
||||
from mitmproxy.net.dns import types
|
||||
|
||||
|
||||
class StripECH:
|
||||
def load(self, loader):
|
||||
loader.add_option(
|
||||
"strip_ech",
|
||||
bool,
|
||||
True,
|
||||
"Strip DNS HTTPS records to prevent clients from sending Encrypted ClientHello (ECH) messages",
|
||||
)
|
||||
|
||||
def dns_response(self, flow: dns.DNSFlow):
|
||||
assert flow.response
|
||||
if ctx.options.strip_ech:
|
||||
for answer in flow.response.answers:
|
||||
if answer.type == types.HTTPS:
|
||||
answer.https_ech = None
|
||||
|
|
@ -5,6 +5,7 @@ import itertools
|
|||
import random
|
||||
import struct
|
||||
import time
|
||||
from collections.abc import Iterable
|
||||
from dataclasses import dataclass
|
||||
from ipaddress import IPv4Address
|
||||
from ipaddress import IPv6Address
|
||||
|
|
@ -106,6 +107,31 @@ class ResourceRecord(serializable.SerializableDataclass):
|
|||
def domain_name(self, name: str) -> None:
|
||||
self.data = domain_names.pack(name)
|
||||
|
||||
@property
|
||||
def https_alpn(self) -> tuple[bytes, ...] | None:
|
||||
record = https_records.unpack(self.data)
|
||||
alpn_bytes = record.params.get(SVCParamKeys.ALPN.value, None)
|
||||
if alpn_bytes is not None:
|
||||
i = 0
|
||||
ret = []
|
||||
while i < len(alpn_bytes):
|
||||
token_len = alpn_bytes[i]
|
||||
ret.append(alpn_bytes[i + 1 : i + 1 + token_len])
|
||||
i += token_len + 1
|
||||
return tuple(ret)
|
||||
else:
|
||||
return None
|
||||
|
||||
@https_alpn.setter
|
||||
def https_alpn(self, alpn: Iterable[bytes] | None) -> None:
|
||||
record = https_records.unpack(self.data)
|
||||
if alpn is None:
|
||||
record.params.pop(SVCParamKeys.ALPN.value, None)
|
||||
else:
|
||||
alpn_bytes = b"".join(bytes([len(a)]) + a for a in alpn)
|
||||
record.params[SVCParamKeys.ALPN.value] = alpn_bytes
|
||||
self.data = https_records.pack(record)
|
||||
|
||||
@property
|
||||
def https_ech(self) -> str | None:
|
||||
record = https_records.unpack(self.data)
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
from mitmproxy import dns
|
||||
from mitmproxy.addons import strip_ech
|
||||
from mitmproxy.addons import strip_dns_https_records
|
||||
from mitmproxy.net.dns import https_records
|
||||
from mitmproxy.net.dns import types
|
||||
from mitmproxy.net.dns.https_records import SVCParamKeys
|
||||
|
|
@ -9,8 +9,8 @@ from mitmproxy.test import tutils
|
|||
|
||||
|
||||
class TestStripECH:
|
||||
def test_simple(self):
|
||||
se = strip_ech.StripECH()
|
||||
def test_strip_ech(self):
|
||||
se = strip_dns_https_records.StripDnsHttpsRecords()
|
||||
with taddons.context(se) as tctx:
|
||||
params1 = {
|
||||
SVCParamKeys.PORT.value: b"\x01\xbb",
|
||||
|
|
@ -51,3 +51,35 @@ class TestStripECH:
|
|||
for answer in f.response.answers
|
||||
if answer.type == types.HTTPS
|
||||
)
|
||||
|
||||
def test_strip_alpn(self):
|
||||
se = strip_dns_https_records.StripDnsHttpsRecords()
|
||||
with taddons.context(se) as tctx:
|
||||
record2 = https_records.HTTPSRecord(
|
||||
1,
|
||||
"example.com",
|
||||
{
|
||||
SVCParamKeys.ALPN.value: b"\x02h2\x02h3",
|
||||
},
|
||||
)
|
||||
answers = [
|
||||
dns.ResourceRecord(
|
||||
"dns.google",
|
||||
dns.types.HTTPS,
|
||||
dns.classes.IN,
|
||||
32,
|
||||
https_records.pack(record2),
|
||||
)
|
||||
]
|
||||
f = tflow.tdnsflow(resp=tutils.tdnsresp(answers=answers))
|
||||
|
||||
se.dns_response(f)
|
||||
assert f.response.answers[0].https_alpn == (b"h2", b"h3")
|
||||
|
||||
tctx.configure(se, http3=False)
|
||||
se.dns_response(f)
|
||||
assert f.response.answers[0].https_alpn == (b"h2",)
|
||||
|
||||
f.response.answers[0].https_alpn = [b"h3"]
|
||||
se.dns_response(f)
|
||||
assert f.response.answers[0].https_alpn is None
|
||||
|
|
@ -79,6 +79,11 @@ class TestResourceRecord:
|
|||
assert rr.domain_name == "www.example.org"
|
||||
rr.text = "sample text"
|
||||
assert rr.text == "sample text"
|
||||
|
||||
def test_https_record_ech(self):
|
||||
rr = dns.ResourceRecord(
|
||||
"test", dns.types.ANY, dns.classes.IN, dns.ResourceRecord.DEFAULT_TTL, b""
|
||||
)
|
||||
params = {3: b"\x01\xbb"}
|
||||
record = dns.https_records.HTTPSRecord(1, "example.org", params)
|
||||
rr.data = dns.https_records.pack(record)
|
||||
|
|
@ -88,6 +93,24 @@ class TestResourceRecord:
|
|||
rr.https_ech = None
|
||||
assert rr.https_ech is None
|
||||
|
||||
def test_https_record_alpn(self):
|
||||
rr = dns.ResourceRecord(
|
||||
"test", dns.types.ANY, dns.classes.IN, dns.ResourceRecord.DEFAULT_TTL, b""
|
||||
)
|
||||
record = dns.https_records.HTTPSRecord(1, "example.org", {})
|
||||
rr.data = dns.https_records.pack(record)
|
||||
|
||||
assert rr.https_alpn is None
|
||||
assert rr.data == b"\x00\x01\x07example\x03org\x00"
|
||||
|
||||
rr.https_alpn = [b"h2", b"h3"]
|
||||
assert rr.https_alpn == (b"h2", b"h3")
|
||||
assert rr.data == b"\x00\x01\x07example\x03org\x00\x00\x01\x00\x06\x02h2\x02h3"
|
||||
|
||||
rr.https_alpn = None
|
||||
assert rr.https_alpn is None
|
||||
assert rr.data == b"\x00\x01\x07example\x03org\x00"
|
||||
|
||||
|
||||
class TestMessage:
|
||||
def test_json(self):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue