bughz/3proxy
1
0
Fork 0
mirror of https://github.com/3proxy/3proxy.git synced 2026-09-02 12:55:49 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Cover the port mappers, tlspr, dnspr, auto and the UDP data path

Browse Source 
tcppm and udppm forward a port each, so test both directions of each: a
request through the TCP mapper reaching its target, a datagram through the
UDP one coming back answered, and a mapper whose rules deny the client
answering nothing.

tlspr takes its destination from the name in the handshake, so point that
name at 127.0.0.1 with nsrecord and give the certificate the same name: the
name then both chooses where the request goes and is what the client checks.

dnspr answers from its cache, including the documented 0.0.0.0 record, which
is handed out as an address rather than withheld.

auto is asked to serve an HTTP proxy request, SOCKS4, SOCKS5 and a TLS
handshake on one port, and to make nothing of a request that is none of
them. Its protocols reach different places, so there are two origins.

The SOCKS UDP association was only checked for the port it binds. Send
datagrams through it as well, large and small, and check a second
association gets its own port - and that one bound inside an intport range
still relays.

A UDP service has no socket to connect to, so readiness is found by asking
until it answers rather than racing it.
This commit is contained in:
Vladimir Dubrovin 2026-08-26 10:54:04 +03:00
parent cfc3c2bd7d
commit daa0e36e41
8 changed files with 419 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
tests/README.md
View File

@ -51,6 +51,11 @@ Assertions are `eq`, `ne`, `contains`, `not_contains`, `in_range`,
`not_in_range`, plus `ok`, `fail` and `skip`. `harness.field()` and
`int_field()` pull a single line out of an `echo` reply.
For services with no TCP port to connect to, `t.udp_echo()` starts an echo
server, `t.udp_exchange()` sends a datagram, `t.wait_udp()` waits for a UDP
service to start answering, `t.socks_udp()` carries one through a SOCKS
association, and `t.dns_query()` asks a DNS server for an A record.
`t.certs()` generates a CA, a second unrelated CA, and a certificate for
127.0.0.1, once per run and inside the run's temporary directory, so no key
material lives in the tree. `t.https()`, `t.tls_proxy_http()` and

74
tests/cases/auto.py Normal file
View File

@ -0,0 +1,74 @@
"""auto: one port that works out which protocol the client is speaking.
Two origins, because the protocols reach different places: an HTTP or SOCKS
client names its own destination, while a TLS client names a host in the
handshake and the service supplies the port.
"""
def run(t):
certs = t.certs()
plain = t.free_port()
port = t.free_port()
secure = t.free_port() if certs else None
tls_origin = ""
if certs:
tls_origin = f"""
flush
ssl_server_cert {certs.server}
ssl_server_key {certs.server_key}
ssl_serv
auth iponly
allow *
http * /echo* echo
httpsrv -p{secure}
ssl_noserv"""
ports = [plain, port] + ([secure] if certs else [])
server = t.start("auto", f"""
log
auth iponly
allow *
http * /echo* echo
httpsrv -p{plain}
{tls_origin}
flush
nserver 127.0.0.1
nscache 1024
nsrecord sni.test 127.0.0.1
auth iponly
allow *
auto -p{port}{f' -P{secure}' if certs else ''}
""", ports=ports)
url = f"http://127.0.0.1:{plain}/echo"
at = f"127.0.0.1:{port}"
# --- as an HTTP proxy -------------------------------------------------
r = t.http(url, proxy=at)
t.eq(200, r.status, "the same port serves an HTTP proxy request")
t.contains(r, "path=/echo", "the origin sees it")
t.contains(t.http(url, proxy=at, method="POST", body="x=1"), "method=POST",
"a POST is recognised as HTTP too")
# --- as a SOCKS proxy --------------------------------------------------
r = t.socks_http(at, url)
t.eq(200, r.status, "the same port serves SOCKS5")
t.contains(r, "path=/echo", "the origin sees the SOCKS request")
t.eq(200, t.socks_http(at, url, socks4=True).status,
"and SOCKS4 on the same port")
# --- as a name-directed TLS proxy --------------------------------------
if certs and "Unknown command" not in server.output():
r = t.https(f"https://sni.test:{port}/echo", ca=certs.ca, strict=False,
connect_to=("127.0.0.1", port))
t.eq(200, r.status, "and a TLS handshake, routed by the name it carries")
t.contains(r, "path=/echo", "which reaches the TLS origin")
else:
t.skip("auto over TLS (no SSL support, or no openssl to make certificates)")
# --- what it is not ----------------------------------------------------
t.not_contains(t.raw(port, "GIBBERISH\r\n\r\n"), "200 OK",
"nonsense is not served as anything")

40
tests/cases/dnspr.py Normal file
View File

@ -0,0 +1,40 @@
"""dnspr: a caching DNS proxy, answering from what it has been told."""
def run(t):
port = t.free_port()
t.start("dnspr", f"""
log
flush
nserver 127.0.0.1
nscache 1024
nsrecord host.test 10.11.12.13
nsrecord other.test 10.11.12.14
nsrecord blocked.test 0.0.0.0
auth iponly
allow *
dnspr -p{port}
""")
# wait for the service: a datagram sent too early is simply lost
for _ in range(100):
if t.dns_query(port, "host.test"):
break
t.eq(["10.11.12.13"], t.dns_query(port, "host.test"),
"a static record is answered")
t.eq(["10.11.12.14"], t.dns_query(port, "other.test"),
"and so is another one")
# asking twice must give the same answer, which is what the cache is for
t.eq(["10.11.12.13"], t.dns_query(port, "host.test"),
"the same name answers the same way again")
# 0.0.0.0 is the documented way to make a name never resolve: the
# address is handed out, and it is the client that then gets nowhere
t.eq(["0.0.0.0"], t.dns_query(port, "blocked.test"),
"a name pointed at 0.0.0.0 answers with that address")
# a name it knows nothing about cannot be answered from here: the
# configured server does not exist, so there is nothing to forward to
t.ne(["10.11.12.13"], t.dns_query(port, "unknown.test") or [],
"an unknown name does not borrow another answer")

6
tests/cases/parent_ports.py
View File

@ -134,6 +134,12 @@ def run(t):
t.in_range(t.socks_udp_associate(udps), ILOW, IHIGH,
"UDP ASSOCIATE binds inside the internal range")
# and the association still carries traffic while bound in the range
echo = t.udp_echo()
reply, bound = t.socks_udp(f"127.0.0.1:{udps}", "127.0.0.1", echo, b"data")
t.eq(b"echo:data", reply, "a range-bound association still relays")
t.in_range(bound, ILOW, IHIGH, "and the port it relays from is in the range")
# without a range the association still works, on an ephemeral port
udps2 = t.free_port()
t.start("parent_intport_none", f"""

64
tests/cases/portmap.py Normal file
View File

@ -0,0 +1,64 @@
"""The port mappers: tcppm forwards a TCP port, udppm a UDP one."""
def run(t):
# --- tcppm ---------------------------------------------------------
origin = t.free_port()
mapped = t.free_port()
refused = t.free_port()
t.start("portmap_tcp", f"""
log
auth iponly
allow *
http * /echo* echo
http * /data data
httpsrv -p{origin}
flush
auth iponly
allow *
tcppm {mapped} 127.0.0.1 {origin}
flush
auth iponly
deny *
tcppm {refused} 127.0.0.1 {origin}
""", ports=[origin, mapped, refused])
r = t.http(f"http://127.0.0.1:{mapped}/echo")
t.eq(200, r.status, "a mapped TCP port reaches the target")
t.contains(r, "path=/echo", "the target sees the request")
t.contains(r, "peer.addr=127.0.0.1", "the mapper makes the connection")
t.eq(20000, t.http(f"http://127.0.0.1:{mapped}/data?size=20000").length,
"a body passes through the mapper")
# the mapper is a service like any other, so its rules apply
r = t.http(f"http://127.0.0.1:{refused}/echo")
t.ne(200, r.status, "a mapper whose rules deny the client answers nothing")
t.stop_all()
# --- udppm ---------------------------------------------------------
# something has to be listening for the mapped datagrams to go anywhere
echo = t.udp_echo()
mapped = t.free_port()
t.start("portmap_udp", f"""
log
flush
auth iponly
allow *
udppm {mapped} 127.0.0.1 {echo}
""")
# a UDP service has no listening socket to wait for, so ask until it
# answers rather than racing it
t.wait_udp(mapped)
t.eq(b"echo:hello", t.udp_exchange(mapped, b"hello"),
"a datagram is relayed and the reply comes back")
t.eq(b"echo:second", t.udp_exchange(mapped, b"second"),
"a second datagram uses the mapping again")
big = b"x" * 2000
t.eq(b"echo:" + big, t.udp_exchange(mapped, big),
"a larger datagram survives the round trip")

16
tests/cases/socks.py
View File

@ -46,6 +46,22 @@ def run(t):
t.eq(200, t.socks_http(plain, origin + "/echo", socks4=True).status,
"a SOCKS4 connection")
# --- the UDP association, and what goes through it ---------------------
# Binding the association is one thing; carrying a datagram is what it
# is for.
echo = t.udp_echo()
reply, bound = t.socks_udp(plain, "127.0.0.1", echo, b"ping")
t.eq(b"echo:ping", reply, "a datagram is relayed and answered")
t.ne(None, bound, "the association reports the port to send to")
reply, _ = t.socks_udp(plain, "127.0.0.1", echo, b"x" * 2000)
t.eq(b"echo:" + b"x" * 2000, reply, "a larger datagram survives the relay")
# each association gets its own socket
_, first = t.socks_udp(plain, "127.0.0.1", echo, b"one")
_, second = t.socks_udp(plain, "127.0.0.1", echo, b"two")
t.ne(first, second, "a second association binds its own port")
# --- authentication ----------------------------------------------------
t.eq(200, t.socks_http(guarded, origin + "/echo",
auth=("alice", "secret")).status,

47
tests/cases/tlspr.py Normal file
View File

@ -0,0 +1,47 @@
"""tlspr: the destination comes from the name in the TLS handshake."""
def run(t):
certs = t.certs()
if not certs:
t.skip("tlspr (openssl is not available to generate certificates)")
return
origin = t.free_port()
sni = t.free_port()
server = t.start("tlspr", f"""
log
ssl_server_cert {certs.server}
ssl_server_key {certs.server_key}
ssl_serv
auth iponly
allow *
http * /echo* echo
httpsrv -p{origin}
flush
ssl_noserv
nserver 127.0.0.1
nscache 1024
nsrecord sni.test 127.0.0.1
auth iponly
allow *
tlspr -p{sni} -P{origin}
""", ports=[origin, sni])
if "Unknown command" in server.output():
t.skip("tlspr (this build has no SSL support)")
return
# The certificate names sni.test, so the name in the handshake is both
# what picks the destination and what the client checks.
r = t.https(f"https://sni.test:{sni}/echo", ca=certs.ca, strict=False,
connect_to=("127.0.0.1", sni))
t.eq(200, r.status, "the name in the handshake reaches its destination")
t.contains(r, "path=/echo", "the request arrives at the origin")
# a name the proxy cannot resolve has nowhere to go
r = t.https(f"https://nowhere.test:{sni}/echo", ca=certs.ca, strict=False,
verify_name=False, connect_to=("127.0.0.1", sni))
t.ne(200, r.status, "a name that does not resolve is refused")

170
tests/harness.py
View File

@ -29,6 +29,7 @@ import struct
import subprocess
import sys
import textwrap
import threading
import time
@ -125,6 +126,7 @@ class Tester:
self._skipped = 0
self._certs = None
self.logs = []
self.udp_servers = []
# ---- servers -----------------------------------------------------
@ -205,6 +207,9 @@ class Tester:
def stop_all(self):
"""Stop the servers, keeping what they printed for the report."""
for sock in self.udp_servers:
sock.close()
self.udp_servers = []
for server in self.servers:
server.stop()
self.logs.append((server.name, server.output()))
@ -294,6 +299,150 @@ class Tester:
except OSError as exc:
return f"<no reply: {exc}>"
# ---- UDP ---------------------------------------------------------
def udp_echo(self, prefix=b"echo:"):
"""Start a UDP server that echoes what it receives, and give its port.
Something has to be on the far side of a port mapper or a SOCKS
association for the data path to be visible at all.
"""
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(("127.0.0.1", 0))
port = sock.getsockname()[1]
def serve():
while True:
try:
data, peer = sock.recvfrom(65536)
except OSError:
return
try:
sock.sendto(prefix + data, peer)
except OSError:
return
thread = threading.Thread(target=serve, daemon=True)
thread.start()
self.udp_servers.append(sock)
return port
def udp_exchange(self, port, payload, host="127.0.0.1"):
"""Send one datagram and return the reply, or None."""
if not isinstance(payload, bytes):
payload = payload.encode()
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(self.timeout)
try:
sock.sendto(payload, (host, port))
return sock.recvfrom(65536)[0]
except OSError:
return None
finally:
sock.close()
def wait_udp(self, port, payload=b"ping", timeout=5.0):
"""Wait until a UDP service answers.
There is no socket to connect to, so readiness can only be found
out by asking; a datagram sent before the service is up is simply
lost.
"""
deadline = time.time() + timeout
while time.time() < deadline:
if self.udp_exchange(port, payload) is not None:
return True
time.sleep(0.05)
return False
def socks_udp(self, socks, host, port, payload, keep=None):
"""Relay a datagram through a SOCKS5 association.
Returns (reply payload, association port), or (None, port) if
nothing came back. The control connection has to stay open for the
association to live, so it is closed only on the way out.
"""
if not isinstance(payload, bytes):
payload = payload.encode()
shost, sport = self._hostport(socks)
ctrl = None
udp = None
try:
ctrl = socket.create_connection((shost, sport), self.timeout)
ctrl.settimeout(self.timeout)
ctrl.sendall(b"\x05\x01\x00")
if self._recvall(ctrl, 2) != b"\x05\x00":
return None, None
ctrl.sendall(b"\x05\x03\x00\x01\x00\x00\x00\x00" + struct.pack("!H", 0))
reply = self._recvall(ctrl, 4)
if len(reply) < 4 or reply[1] != 0:
return None, None
_, bound = self._read_socks_addr(ctrl, reply[3])
udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp.settimeout(self.timeout)
header = (b"\x00\x00\x00\x01" + socket.inet_aton(host) +
struct.pack("!H", port))
udp.sendto(header + payload, (shost, bound))
try:
data = udp.recvfrom(65536)[0]
except OSError:
return None, bound
# the reply carries the same kind of header, which is not payload
if len(data) < 10 or data[3] != 1:
return None, bound
return data[10:], bound
except OSError:
return None, None
finally:
if udp:
udp.close()
if ctrl:
ctrl.close()
# ---- DNS ---------------------------------------------------------
def dns_query(self, port, name, host="127.0.0.1"):
"""Ask for an A record and return the addresses in the answer."""
query = struct.pack("!HHHHHH", 0x2A2A, 0x0100, 1, 0, 0, 0)
for label in name.split("."):
query += bytes([len(label)]) + label.encode()
query += b"\x00" + struct.pack("!HH", 1, 1)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(self.timeout)
try:
sock.sendto(query, (host, port))
data = sock.recvfrom(65536)[0]
except OSError:
return None
finally:
sock.close()
if len(data) < 12 or data[:2] != query[:2]:
return None
answers = struct.unpack("!H", data[6:8])[0]
addresses = []
pos = 12
while pos < len(data) and data[pos]: # skip the question
pos += data[pos] + 1
pos += 5
for _ in range(answers):
if pos + 12 > len(data):
break
if data[pos] & 0xC0 == 0xC0:
pos += 2
else:
while pos < len(data) and data[pos]:
pos += data[pos] + 1
pos += 1
rtype, _, _, rdlen = struct.unpack("!HHIH", data[pos:pos + 10])
pos += 10
if rtype == 1 and rdlen == 4:
addresses.append(socket.inet_ntoa(data[pos:pos + 4]))
pos += rdlen
return addresses
# ---- SOCKS -------------------------------------------------------
def _socks_connect(self, shost, sport, host, port, socks4=False,
@ -442,7 +591,7 @@ class Tester:
# them for a signed certificate the way OpenSSL 3 does, and Python
# rejects a chain with no Authority Key Identifier from 3.13.
with open(ext, "w") as fp:
fp.write("subjectAltName=IP:127.0.0.1,DNS:localhost\n"
fp.write("subjectAltName=IP:127.0.0.1,DNS:localhost,DNS:sni.test\n"
"subjectKeyIdentifier=hash\n"
"authorityKeyIdentifier=keyid,issuer\n")
# A CA without these is not usable as one. They go in a file rather
@ -542,11 +691,26 @@ class Tester:
conn.close()
def https(self, url, proxy=None, ca=None, strict=True, verify_name=True,
method="GET", headers=None):
"""An https:// request, optionally tunnelled through a proxy."""
method="GET", headers=None, connect_to=None):
"""An https:// request, optionally tunnelled through a proxy.
connect_to sends the handshake somewhere other than the name in the
URL, which is how a name-directed proxy is reached: the name still
goes out in the handshake and is what the certificate is checked
against.
"""
host, port, path = self._split(url, default_port=443)
context = self._context(ca, strict, verify_name)
try:
if connect_to:
raw = socket.create_connection(connect_to, self.timeout)
conn = http.client.HTTPSConnection(host, port, context=context,
timeout=self.timeout)
conn.sock = context.wrap_socket(raw, server_hostname=host)
conn.request(method, path, headers=headers or {})
reply = conn.getresponse()
return Response(reply.status, reply.read(),
dict(reply.getheaders()))
if proxy:
phost, pport = self._hostport(proxy)
conn = http.client.HTTPSConnection(phost, pport, context=context,