bughz/3proxy
1
0
Fork 0
mirror of https://github.com/3proxy/3proxy.git synced 2026-08-26 01:45:49 +08:00
Code Issues Actions 10 Packages Projects Releases Wiki Activity

Add tests

Browse Source
This commit is contained in:
Vladimir Dubrovin 2026-08-25 18:03:50 +03:00
parent 7f430ccc79
commit 73fbf9d262
16 changed files with 1304 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.github/workflows/c-cpp-Linux.yml vendored
View File

@ -2,9 +2,9 @@ name: C/C++ CI Linux
on:
push:
paths: [ '**.c', '**.h', 'Makefile.Linux', '.github/configs', '.github/workflows/c-cpp-Linux.yml' ]
paths: [ '**.c', '**.h', 'Makefile.Linux', 'tests/**', '.github/configs', '.github/workflows/c-cpp-Linux.yml' ]
pull_request:
paths: [ "**.c", "**.h", "Makefile.Linux", ".github/configs", ".github/workflows/c-cpp-Linux.yml" ]
paths: [ "**.c", "**.h", "Makefile.Linux", "tests/**", ".github/configs", ".github/workflows/c-cpp-Linux.yml" ]
workflow_dispatch:
permissions:
@ -28,6 +28,8 @@ jobs:
run: sudo apt-get update && sudo apt-get install -y libssl-dev libpam-dev libpcre2-dev
- name: make
run: make -f Makefile.Linux
- name: regression tests
run: python3 tests/run.py
- name: mkdir
run: mkdir ~/3proxy
- name: make install

6
.github/workflows/c-cpp-MacOS.yml vendored
View File

@ -2,9 +2,9 @@ name: C/C++ CI MacOS
on:
push:
paths: [ '**.c', '**.h', 'Makefile.FreeBSD', '.github/configs', '.github/workflows/c-cpp-MacOS.yml' ]
paths: [ '**.c', '**.h', 'Makefile.FreeBSD', 'tests/**', '.github/configs', '.github/workflows/c-cpp-MacOS.yml' ]
pull_request:
paths: [ "**.c", "**.h", "Makefile.FreeBSD", ".github/configs", ".github/workflows/c-cpp-MacOS.yml" ]
paths: [ "**.c", "**.h", "Makefile.FreeBSD", "tests/**", ".github/configs", ".github/workflows/c-cpp-MacOS.yml" ]
workflow_dispatch:
permissions:
@ -29,5 +29,7 @@ jobs:
env:
LDFLAGS: "-L/usr/local/lib -L/opt/homebrew/lib -L/opt/homebrew/opt/openssl/lib"
CFLAGS: "-I/usr/local/include -I/opt/homebrew/include -I/usr/local/opt/openssl/include -I/opt/homebrew/opt/openssl/include"
- name: regression tests
run: python3 tests/run.py
- name: make clean MacOS
run: make -f Makefile.FreeBSD clean

7
.github/workflows/c-cpp-Windows.yml vendored
View File

@ -2,9 +2,9 @@ name: C/C++ CI Windows
on:
push:
paths: [ '**.c', '**.h', 'Makefile.msvc', '.github/configs', '.github/workflows/c-cpp-Windows.yml' ]
paths: [ '**.c', '**.h', 'Makefile.msvc', 'tests/**', '.github/configs', '.github/workflows/c-cpp-Windows.yml' ]
pull_request:
paths: [ "**.c", "**.h", "Makefile.msvc", ".github/configs", ".github/workflows/c-cpp-Windows.yml" ]
paths: [ "**.c", "**.h", "Makefile.msvc", "tests/**", ".github/configs", ".github/workflows/c-cpp-Windows.yml" ]
workflow_dispatch:
permissions:
@ -27,6 +27,8 @@ jobs:
env:
LDFLAGS: '-L "c:/msys64/mingw64/lib"'
CFLAGS: '-I "c:/msys64/mingw64/include"'
- name: regression tests (MinGW)
run: python tests\run.py --bin bin\3proxy.exe
- name: make clean Windows
run: make -f Makefile.win clean
- name: Add msbuild to PATH
@ -40,4 +42,5 @@ jobs:
set "LIB=%LIB%;c:/vcpkg/installed/x64-windows-static/lib;c:/vcpkg/installed/x64-windows/lib"
set "INCLUDE=%INCLUDE%;c:/vcpkg/installed/x64-windows-static/include;c:/vcpkg/installed/x64-windows/include"
nmake /F Makefile.msvc WOLFSSL=1 || exit /b 1
python tests\run.py --bin bin\3proxy.exe || exit /b 1
nmake /F Makefile.msvc clean

13
.github/workflows/c-cpp-cmake.yml vendored
View File

@ -2,9 +2,9 @@ name: C/C++ CI cmake
on:
push:
paths: [ '**.c', '**.h', '**.cmake', 'CMakeLists.txt', '.github/configs', '.github/workflows/c-cpp-cmake.yml' ]
paths: [ '**.c', '**.h', '**.cmake', 'CMakeLists.txt', 'tests/**', '.github/configs', '.github/workflows/c-cpp-cmake.yml' ]
pull_request:
paths: [ "**.c", "**.h", "**.cmake", "CMakeLists.txt", ".github/configs", ".github/workflows/c-cpp-cmake.yml" ]
paths: [ "**.c", "**.h", "**.cmake", "CMakeLists.txt", "tests/**", ".github/configs", ".github/workflows/c-cpp-cmake.yml" ]
workflow_dispatch:
permissions:
@ -43,7 +43,9 @@ jobs:
cmake --build .
mkdir ~/3proxy
DESTDIR=~/3proxy cmake --install .
cd .. && rm -rf build/
cd ..
python3 tests/run.py --bin build/bin/3proxy
rm -rf build/
- name: make with CMake Win
if: ${{ startsWith(matrix.target, 'windows') }}
shell: cmd
@ -56,6 +58,7 @@ jobs:
dir
cmake --build .
cd ..
python tests\run.py || exit /b 1
rmdir /s /q build
wolfssl:
@ -71,4 +74,6 @@ jobs:
cd build
cmake ..
cmake --build .
cd .. && rm -rf build/
cd ..
python3 tests/run.py --bin build/bin/3proxy
rm -rf build/

2
tests/.gitignore vendored Normal file
View File

@ -0,0 +1,2 @@
__pycache__/
*.pyc

54
tests/README.md Normal file
View File

@ -0,0 +1,54 @@
# Regression tests
python3 tests/run.py # every case
python3 tests/run.py httpsrv # cases whose name matches
python3 tests/run.py --bin build/bin/3proxy
python3 tests/run.py -v # print every check
python3 tests/run.py --keep # keep the configurations and logs
Python 3.6 or later and a built 3proxy are the only requirements: the suite
is standard library throughout, so it runs wherever 3proxy builds. With no
`--bin` it looks in `bin/`, then `build/bin/`, then the per-configuration
directories a multi-configuration CMake generator uses.
The proxy under test is also the origin server the tests talk to: the `http`
command's `echo` operation reports back how a request arrived - method, path,
query, host, and the source port it came from - and `data` generates a body
of a requested size, framing, status and pace. So a case can state what a
proxy should do to a request and then read off what actually reached the
other side.
## Adding a case
A case is a module under `cases/` exporting `run(t)`. It writes the
configurations it needs, starts them, and says what it expects:
```python
def run(t):
srv = t.free_port()
t.start("my_case", f"""
log
auth iponly
allow *
http * /echo echo
httpsrv -p{srv}
""", ports=[srv])
r = t.http(f"http://127.0.0.1:{srv}/echo")
t.eq(200, r.status, "the server answers")
t.contains(r, "method=GET", "the method is reported")
```
Servers are stopped for you when the case ends, whether or not it passed.
`t` offers `http()` (direct, through an HTTP proxy, or over a CONNECT
tunnel), `socks_http()` and `socks_connect()` for SOCKS4 and SOCKS5,
`socks_udp_associate()`, `raw()` for bytes a real client would never send,
and `run_config()` for configurations that are meant to be rejected.
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.
Note that access rules accumulate until `flush`, so a service section that
means to stand on its own should start with one - otherwise an earlier
`allow *` matches first and the rule under test is never reached.

69
tests/cases/admin.py Normal file
View File

@ -0,0 +1,69 @@
"""The admin interface, now a set of handlers on the HTTP server."""
def run(t):
adm = t.free_port()
lim = t.free_port()
t.start("admin", f"""
log
auth iponly
allow *
countin 1 D 100 * * *
countin 2 D 200 * * *
admin -p{adm}
flush
auth iponly
allow *
admin -p{lim} -s1
""", ports=[adm, lim])
url = f"http://127.0.0.1:{adm}"
# --- the predefined pages -----------------------------------------
t.eq(200, t.http(url + "/").status, "the main page")
t.eq(200, t.http(url + "/C").status, "the counters page")
t.eq(200, t.http(url + "/R").status, "the reload page")
t.eq(200, t.http(url + "/S").status, "the services page")
counters = t.http(url + "/C")
t.contains(counters, "countin", "the counters page names the counter type")
t.contains(counters, "<tr>", "the counters page renders a table")
t.contains(t.http(url + "/R"), "Reload", "the reload page confirms the request")
t.contains(t.http(url + "/S"), "<", "the services page returns markup")
# --- the menu no longer offers the removed config editor -----------
main = t.http(url + "/")
t.contains(main, "HREF='/C'", "the menu links to the counters")
t.contains(main, "HREF='/R'", "the menu links to reload")
t.contains(main, "HREF='/S'", "the menu links to the services")
t.not_contains(main, "HREF='/F'",
"the menu no longer links to the config editor")
# /F and /U are gone, so they fall through to the catch-all rule
t.eq(200, t.http(url + "/F").status, "the removed /F falls through")
t.eq(200, t.http(url + "/U").status, "the removed /U falls through")
t.contains(t.http(url + "/F"), "configuration", "/F yields the main page")
# --- counter control through the glob ------------------------------
# /C<action><number> is routed by the /C* rule, the action arriving as
# the glob
t.http(url + "/CD0")
t.contains(t.http(url + "/C"), ">NO<", "a counter can be disabled")
t.http(url + "/CS0")
t.contains(t.http(url + "/C"), ">YES<", "a counter can be enabled again")
# --- limited mode ---------------------------------------------------
limited = f"http://127.0.0.1:{lim}"
t.eq(200, t.http(limited + "/").status, "limited mode serves the main page")
t.eq(200, t.http(limited + "/C").status, "limited mode serves the counters")
t.not_contains(t.http(limited + "/R"), "Reload scheduled",
"limited mode refuses a reload")
# --- the writable command is gone ------------------------------------
output = t.run_config("writable", f"""
log
writable
admin -p{t.free_port()}
""")
t.contains(output, "Unknown command", "the writable command is rejected")

49
tests/cases/httpsrv_auth.py Normal file
View File

@ -0,0 +1,49 @@
"""Authentication and access rules in front of the HTTP server."""
def run(t):
srv = t.free_port()
openport = t.free_port()
t.start("httpsrv_auth", f"""
log
http * /echo echo
auth strong
users alice:CL:secret bob:CL:hunter2
allow alice
httpsrv -p{srv}
flush
http * /echo echo
auth iponly
allow *
httpsrv -p{openport}
""", ports=[srv, openport])
url = f"http://127.0.0.1:{srv}"
r = t.http(url + "/echo")
t.eq(401, r.status, "no credentials gives 401")
t.ne(None, r.header("WWW-Authenticate"),
"the 401 carries a WWW-Authenticate header")
t.eq(200, t.http(url + "/echo", auth=("alice", "secret")).status,
"valid credentials pass")
t.eq(401, t.http(url + "/echo", auth=("alice", "wrong")).status,
"a wrong password gives 401")
t.eq(401, t.http(url + "/echo", auth=("nobody", "secret")).status,
"an unknown user gives 401")
# bob authenticates, but no rule admits him
t.eq(403, t.http(url + "/echo", auth=("bob", "hunter2")).status,
"authenticated but not allowed gives 403")
# authentication comes before dispatch, so an unmatched URL still needs it
t.eq(401, t.http(url + "/nosuchpath").status,
"authentication precedes the rule lookup")
# the second service kept its own iponly authentication
t.eq(200, t.http(f"http://127.0.0.1:{openport}/echo").status,
"the open service needs no credentials")
t.contains(t.http(url + "/echo", auth=("alice", "secret")), "path=/echo",
"an authenticated request is dispatched")

78
tests/cases/httpsrv_ops.py Normal file
View File

@ -0,0 +1,78 @@
"""The built-in HTTP server: the echo and data operations."""
import time
def run(t):
srv = t.free_port()
t.start("httpsrv_ops", f"""
log
auth iponly
allow *
http * /echo* echo
http * /data data
http * /small data size=64
httpsrv -p{srv}
""", ports=[srv])
url = f"http://127.0.0.1:{srv}"
# --- echo: request introspection ---------------------------------
r = t.http(url + "/echo?a=1")
t.eq(200, r.status, "echo answers 200")
t.contains(r, "method=GET", "echo reports the method")
t.contains(r, "path=/echo", "echo reports the path")
t.contains(r, "query=a=1", "echo reports the query")
t.contains(r, "peer.addr=127.0.0.1", "echo reports the peer address")
t.contains(r, f"host=127.0.0.1:{srv}", "echo reports the Host header")
# the glob is the wildcard-matched tail, which is how admin routes its
# sub-pages
r = t.http(url + "/echoXYZ")
t.contains(r, "glob=XYZ", "echo reports the glob text")
t.contains(r, "glob.len=3", "echo reports the glob length")
# --- data: generated payload -------------------------------------
t.eq(1000, t.http(url + "/data?size=1000").length, "data honours size")
t.eq(0, t.http(url + "/data?size=0").length, "data size=0 sends an empty body")
t.eq(64, t.http(url + "/small").length, "data takes its size from the rule")
t.eq(1000, t.http(url + "/small?size=1000").length,
"the query overrides the rule parameters")
# a size past one block exercises the send loop
t.eq(70000, t.http(url + "/data?size=70000").length,
"data spans several blocks")
t.eq(70000, t.http(url + "/data?size=70000&block=1024").length,
"data honours the block size")
# --- status and framing ------------------------------------------
t.eq(404, t.http(url + "/data?size=10&status=404").status,
"data honours the status")
t.eq(503, t.http(url + "/data?size=10&status=503").status,
"data returns 503 when asked")
t.eq(200, t.http(url + "/data?size=10&status=99").status,
"an out-of-range status falls back to 200")
r = t.http(url + "/data?size=100")
t.eq("100", r.header("Content-Length"), "an identity reply sets Content-Length")
r = t.http(url + "/data?size=100&chunked=1")
t.eq("chunked", r.header("Transfer-Encoding"),
"a chunked reply sets Transfer-Encoding")
t.eq(None, r.header("Content-Length"),
"a chunked reply omits Content-Length")
t.eq(100, r.length, "a chunked body decodes to the size asked for")
t.eq(70000, t.http(url + "/data?size=70000&chunked=1").length,
"a chunked body spans several blocks")
# --- delay --------------------------------------------------------
start = time.time()
t.http(url + "/data?size=4096&block=1024&delay=100")
elapsed = time.time() - start
if elapsed >= 0.3:
t.ok("delay slows the transfer")
else:
t.fail("delay slows the transfer", ">=0.3s", f"{elapsed:.2f}s")
# --- unmatched ----------------------------------------------------
t.eq(404, t.http(url + "/nosuchthing").status, "an unmatched URL gives 404")

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

@ -0,0 +1,64 @@
"""Request parsing: decoding, path safety, malformed and oversized input.
These go over a raw socket, because a well-behaved client would normalise
most of them away before they ever reached the server.
"""
def run(t):
srv = t.free_port()
t.start("httpsrv_parsing", f"""
log
auth iponly
allow *
http * /echo* echo
http * /safe/* echo
httpsrv -p{srv}
""", ports=[srv])
def request(path, host="t", extra=""):
return t.raw(srv, f"GET {path} HTTP/1.0\r\nHost: {host}\r\n{extra}\r\n")
# --- percent-decoding ---------------------------------------------
reply = request("/%65cho")
t.contains(reply, "200 OK", "a percent-encoded path is decoded before matching")
t.contains(reply, "path=/echo", "the decoded path is what gets reported")
t.contains(request("/echo%20space"), "glob= space",
"an encoded space decodes into the glob")
# --- traversal -----------------------------------------------------
for path in ("/safe/../etc/passwd", "/safe/%2e%2e/etc", "/safe/..%2fetc",
"/echo/../../x"):
t.not_contains(request(path), "200 OK", f"traversal is refused: {path}")
t.contains(request("/safe/./ok"), "200 OK",
"a harmless dot segment is still served")
# --- injection ------------------------------------------------------
t.not_contains(request("/echo%0d%0aInjected:%20yes"), "Injected: yes",
"an encoded CRLF cannot inject a header")
t.not_contains(request("/echo%00cut"), "200 OK", "an encoded NUL is refused")
# a header value cannot smuggle a newline into the echoed output
reply = request("/echo", host="evil", extra="X-Injected: yes\r\n")
t.not_contains(reply, "host=evil\nX-Injected",
"header values stay in their own fields")
# --- malformed ------------------------------------------------------
t.not_contains(t.raw(srv, "GARBAGE\r\n\r\n"), "200 OK",
"a malformed request line is not served")
t.not_contains(t.raw(srv, "GET\r\n\r\n"), "200 OK",
"a request line with no URL is not served")
# an over-long path has to be refused rather than quietly truncated to
# something shorter that might match another rule
t.not_contains(request("/echo" + "a" * 9000), "200 OK",
"an over-long path is refused, not truncated")
# --- methods --------------------------------------------------------
url = f"http://127.0.0.1:{srv}"
t.eq(200, t.http(url + "/echo", method="HEAD").status, "HEAD is accepted")
r = t.http(url + "/echo", method="POST", body="payload=1",
headers={"Content-Type": "application/x-www-form-urlencoded"})
t.contains(r, "method=POST", "POST reaches the handler")
t.contains(r, "content.length=9", "the POST content length is parsed")

69
tests/cases/httpsrv_rules.py Normal file
View File

@ -0,0 +1,69 @@
"""Rule dispatch: host and URL patterns, and per-service rule sets."""
def run(t):
srv = t.free_port()
srv2 = t.free_port()
t.start("httpsrv_rules", f"""
log
auth iponly
allow *
http * /exact echo
http * /pre* echo
http * *.suffix echo
http * *mid* echo
http host.example.com /byhost echo
http *.wild.example.com /bywild echo
http * /only-first echo
httpsrv -p{srv}
flush
auth iponly
allow *
http * /only-second echo
httpsrv -p{srv2}
""", ports=[srv, srv2])
url = f"http://127.0.0.1:{srv}"
# --- URL patterns -------------------------------------------------
t.eq(200, t.http(url + "/exact").status, "an exact URL matches")
t.eq(404, t.http(url + "/exactly").status,
"an exact URL does not match a longer path")
t.eq(200, t.http(url + "/pre").status, "a prefix matches the bare prefix")
t.eq(200, t.http(url + "/pretty/deep").status,
"a prefix matches a longer path")
t.eq(200, t.http(url + "/any.suffix").status, "a suffix matches")
t.eq(404, t.http(url + "/any.suffixx").status,
"a suffix is anchored at the end")
t.eq(200, t.http(url + "/xxmidxx").status, "a substring matches")
t.eq(404, t.http(url + "/nomatch").status, "an unmatched URL gives 404")
# --- host patterns ------------------------------------------------
def with_host(path, host):
return t.http(url + path, headers={"Host": host})
t.eq(200, with_host("/byhost", "host.example.com").status,
"an exact host matches")
t.eq(404, with_host("/byhost", "other.example.com").status,
"another host does not match")
t.eq(200, with_host("/bywild", "a.wild.example.com").status,
"a wildcard host matches")
t.eq(404, with_host("/bywild", "a.other.example.com").status,
"a wildcard host rejects another domain")
# the rules are ordered, and the first match wins
t.contains(t.http(url + "/exact"), "path=/exact",
"the first matching rule handles the request")
# --- per-service rule sets ----------------------------------------
# Rules accumulate until a service starts, which takes them; later rules
# belong to the next service only.
t.eq(200, t.http(f"http://127.0.0.1:{srv}/only-first").status,
"the first service has its own rules")
t.eq(404, t.http(f"http://127.0.0.1:{srv}/only-second").status,
"the first service does not have the later rules")
t.eq(200, t.http(f"http://127.0.0.1:{srv2}/only-second").status,
"the second service has its own rules")
t.eq(404, t.http(f"http://127.0.0.1:{srv2}/only-first").status,
"the second service does not have the earlier rules")

131
tests/cases/parent_ports.py Normal file
View File

@ -0,0 +1,131 @@
"""extport and intport: binding the local side of a connection to a range.
Access rules accumulate until "flush": without it an earlier "allow *"
matches first and the rule carrying the range is never reached.
"""
from harness import int_field
LOW = 21400
HIGH = 21449
ILOW = 21500
IHIGH = 21549
def run(t):
srv = t.free_port()
prx = t.free_port()
sks = t.free_port()
meth = t.free_port()
t.start("parent_ports", f"""
log
auth iponly
allow *
http * /echo* echo
httpsrv -p{srv}
# every outgoing connection binds inside the range
flush
auth iponly
allow *
parent 1000 extport 0.0.0.0 {LOW}-{HIGH}
proxy -p{prx}
# the range applies only to CONNECT: an HTTP proxy CONNECT is
# HTTP_CONNECT, the bare CONNECT operation being the SOCKS one
flush
auth iponly
allow * * * * HTTP_CONNECT
parent 1000 extport 0.0.0.0 {LOW}-{HIGH}
allow *
proxy -p{meth}
# socks, for the same setting on another service
flush
auth iponly
allow *
parent 1000 extport 0.0.0.0 {LOW}-{HIGH}
socks -p{sks}
""", ports=[srv, prx, sks, meth])
origin = f"http://127.0.0.1:{srv}"
proxy = f"127.0.0.1:{prx}"
# --- extport ---------------------------------------------------------
# the origin reports the source port it actually saw
port = int_field(t.http(origin + "/echo", proxy=proxy), "peer.port")
t.in_range(port, LOW, HIGH, "the outgoing connection binds inside the range")
seen = []
for _ in range(5):
seen.append(int_field(t.http(origin + "/echo", proxy=proxy), "peer.port"))
outside = [p for p in seen if p is None or not LOW <= p <= HIGH]
t.eq([], outside, "repeated connections all bind inside the range")
port = int_field(t.socks_http(f"127.0.0.1:{sks}", origin + "/echo"),
"peer.port")
t.in_range(port, LOW, HIGH,
"socks binds the outgoing connection inside the range")
# --- per-method scoping ------------------------------------------------
method_proxy = f"127.0.0.1:{meth}"
port = int_field(t.http(origin + "/echo", proxy=method_proxy, tunnel=True),
"peer.port")
t.in_range(port, LOW, HIGH, "CONNECT uses the range its rule sets")
# a plain GET matches the later rule, which sets no range
port = int_field(t.http(origin + "/echo", proxy=method_proxy), "peer.port")
t.not_in_range(port, LOW, HIGH,
"a method outside that rule keeps an ephemeral port")
# --- intport -----------------------------------------------------------
# A UDP association allocates its socket after the destination is known,
# so the range has to be applied when the rule matches rather than when
# the chain is walked.
udps = t.free_port()
t.start("parent_intport", f"""
log
flush
auth iponly
allow *
parent 1000 intport 0.0.0.0 {ILOW}-{IHIGH}
socks -p{udps}
""", ports=[udps])
t.in_range(t.socks_udp_associate(udps), ILOW, IHIGH,
"UDP ASSOCIATE binds inside the internal range")
# without a range the association still works, on an ephemeral port
udps2 = t.free_port()
t.start("parent_intport_none", f"""
log
flush
auth iponly
allow *
socks -p{udps2}
""", ports=[udps2])
t.ne(None, t.socks_udp_associate(udps2),
"UDP ASSOCIATE works without a range")
# --- configuration errors ------------------------------------------------
dead = t.free_port()
t.contains(t.run_config("badaddr", f"""
log
allow *
parent 1000 extport 127.0.0.1 {LOW}-{HIGH}
proxy -p{dead}
"""), "requires 0.0.0.0", "a non-zero address with extport is rejected")
t.contains(t.run_config("badrange", f"""
log
allow *
parent 1000 extport 0.0.0.0 notaport
proxy -p{dead}
"""), "bad port range", "a malformed range is rejected")
t.contains(t.run_config("badorder", f"""
log
allow *
parent 1000 extport 0.0.0.0 {HIGH}-{LOW}
proxy -p{dead}
"""), "bad port range", "a reversed range is rejected")

108
tests/cases/proxy_http.py Normal file
View File

@ -0,0 +1,108 @@
"""The HTTP proxy, with the built-in server as the origin.
Access rules accumulate until "flush", so each service section here starts
from a clean list.
"""
def run(t):
srv = t.free_port()
other = t.free_port()
prx = t.free_port()
deny = t.free_port()
auth = t.free_port()
t.start("proxy_http", f"""
log
auth iponly
allow *
http * /echo* echo
http * /data data
httpsrv -p{srv}
# a second origin, used as a destination the rules must keep out
flush
auth iponly
allow *
http * /echo* echo
httpsrv -p{other}
# an open proxy
flush
auth iponly
allow *
proxy -p{prx}
# only the first origin is reachable
flush
auth iponly
allow * * * {srv}
proxy -p{deny}
# credentials required
flush
auth strong
users alice:CL:secret
allow alice
proxy -p{auth}
""", ports=[srv, other, prx, deny, auth])
origin = f"http://127.0.0.1:{srv}"
second = f"http://127.0.0.1:{other}"
open_proxy = f"127.0.0.1:{prx}"
# --- plain proxying -------------------------------------------------
r = t.http(origin + "/echo", proxy=open_proxy)
t.eq(200, r.status, "a GET through the proxy")
t.contains(r, "path=/echo", "the origin sees the proxied path")
t.contains(r, "peer.addr=127.0.0.1", "the origin sees the proxy as the peer")
t.eq(10000, t.http(origin + "/data?size=10000", proxy=open_proxy).length,
"a sized body survives proxying")
t.eq(10000,
t.http(origin + "/data?size=10000&chunked=1", proxy=open_proxy).length,
"a chunked body survives proxying")
t.eq(503, t.http(origin + "/data?size=5&status=503", proxy=open_proxy).status,
"the origin status is relayed")
# --- POST and keep-alive ---------------------------------------------
r = t.http(origin + "/echo", proxy=open_proxy, method="POST", body="x=1")
t.contains(r, "method=POST", "POST is proxied")
# two requests on one connection, which may carry different methods
conn = t.connection("127.0.0.1", srv, proxy=open_proxy)
try:
first = t.http(origin + "/echo", proxy=open_proxy, method="POST",
body="x=1", conn=conn)
second_reply = t.http(origin + "/echo", proxy=open_proxy, conn=conn)
t.eq((200, 200), (first.status, second_reply.status),
"two requests on one proxied connection")
finally:
conn.close()
# --- CONNECT ----------------------------------------------------------
t.eq(200, t.http(origin + "/echo", proxy=open_proxy, tunnel=True).status,
"CONNECT tunnels to the origin")
# --- access control ----------------------------------------------------
denying = f"127.0.0.1:{deny}"
t.eq(200, t.http(origin + "/echo", proxy=denying).status,
"the permitted destination is reachable")
t.ne(200, t.http(second + "/echo", proxy=denying).status,
"a destination outside the rules is refused")
t.ne(200, t.http(second + "/echo", proxy=denying, tunnel=True).status,
"CONNECT to a destination outside the rules is refused")
# the open proxy still reaches it, so the refusal came from the rules
t.eq(200, t.http(second + "/echo", proxy=open_proxy).status,
"the same destination is reachable through the open proxy")
# --- proxy authentication -----------------------------------------------
needs_auth = f"127.0.0.1:{auth}"
t.eq(407, t.http(origin + "/echo", proxy=needs_auth).status,
"the proxy demands credentials")
t.eq(200, t.http(origin + "/echo", proxy=needs_auth,
proxy_auth=("alice", "secret")).status,
"valid proxy credentials pass")
t.eq(407, t.http(origin + "/echo", proxy=needs_auth,
proxy_auth=("alice", "wrong")).status,
"wrong proxy credentials are refused")

57
tests/cases/socks.py Normal file
View File

@ -0,0 +1,57 @@
"""The SOCKS proxy, reaching the built-in server."""
def run(t):
srv = t.free_port()
sks = t.free_port()
sauth = t.free_port()
t.start("socks", f"""
log
auth iponly
allow *
http * /echo* echo
http * /data data
httpsrv -p{srv}
flush
auth iponly
allow *
socks -p{sks}
flush
auth strong
users alice:CL:secret
allow alice
socks -p{sauth}
""", ports=[srv, sks, sauth])
origin = f"http://127.0.0.1:{srv}"
plain = f"127.0.0.1:{sks}"
guarded = f"127.0.0.1:{sauth}"
# --- SOCKS5 ---------------------------------------------------------
r = t.socks_http(plain, origin + "/echo")
t.eq(200, r.status, "a SOCKS5 connection")
t.contains(r, "path=/echo", "the origin sees the request made over SOCKS5")
t.eq(10000, t.socks_http(plain, origin + "/data?size=10000").length,
"a body survives SOCKS5")
# resolution delegated to the proxy
t.eq(200, t.socks_http(plain, f"http://localhost:{srv}/echo",
remote_dns=True).status,
"SOCKS5 resolves the hostname itself")
# --- SOCKS4 -----------------------------------------------------------
t.eq(200, t.socks_http(plain, origin + "/echo", socks4=True).status,
"a SOCKS4 connection")
# --- authentication ----------------------------------------------------
t.eq(200, t.socks_http(guarded, origin + "/echo",
auth=("alice", "secret")).status,
"valid SOCKS5 credentials pass")
t.ne(None, t.socks_connect(guarded, "127.0.0.1", srv,
auth=("alice", "wrong")),
"wrong SOCKS5 credentials are refused")
t.ne(None, t.socks_connect(guarded, "127.0.0.1", srv),
"SOCKS5 without credentials is refused")

460
tests/harness.py Normal file
View File

@ -0,0 +1,460 @@
"""Support code for the 3proxy regression tests.
Everything here is standard library, so the suite runs wherever 3proxy
builds: no shell, no curl, no netcat.
A test case is a module under tests/cases/ exporting run(t). It writes the
configurations it needs, starts them, and states what it expects:
def run(t):
srv = t.free_port()
t.start("echo", f'''
log
auth iponly
allow *
http * /echo echo
httpsrv -p{srv}
''', ports=[srv])
r = t.http(f"http://127.0.0.1:{srv}/echo")
t.eq(200, r.status, "the server answers")
"""
import base64
import http.client
import os
import socket
import struct
import subprocess
import sys
import textwrap
import time
class Response:
"""A reply, or the reason there wasn't one."""
def __init__(self, status=None, body=b"", headers=None, error=None):
self.status = status
self.body = body
self.headers = headers or {}
self.error = error
@property
def text(self):
return self.body.decode("utf-8", "replace")
@property
def length(self):
return len(self.body)
def header(self, name):
for k, v in self.headers.items():
if k.lower() == name.lower():
return v
return None
def __repr__(self):
if self.error:
return f"<no reply: {self.error}>"
return f"<{self.status}, {len(self.body)} bytes>"
class Server:
"""A running 3proxy, with the configuration it was given."""
def __init__(self, name, path, proc, logfile):
self.name = name
self.path = path
self.proc = proc
self.logfile = logfile
def output(self):
try:
with open(self.logfile, "rb") as fp:
return fp.read().decode("utf-8", "replace")
except OSError:
return ""
def stop(self):
if self.proc.poll() is None:
self.proc.terminate()
try:
self.proc.wait(timeout=5)
except subprocess.TimeoutExpired:
self.proc.kill()
self.proc.wait(timeout=5)
class Failure(Exception):
"""Raised when a case cannot go on, e.g. a server refused to start."""
class Tester:
"""The API a case runs against: start servers, make requests, assert."""
def __init__(self, binary, tmpdir, case):
self.binary = binary
self.tmpdir = tmpdir
self.case = case
self.servers = []
self.checks = []
self.timeout = 10
self._skipped = 0
# ---- servers -----------------------------------------------------
def free_port(self):
"""A port nothing is listening on. Closed again before it is used,
which is racy in principle and reliable enough in practice."""
s = socket.socket()
try:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
finally:
s.close()
def write_config(self, name, config):
path = os.path.join(self.tmpdir, name + ".cfg")
text = textwrap.dedent(config).strip() + "\n"
# newline="" keeps the line endings as written, rather than letting
# Windows turn them into CRLF behind the parser's back
with open(path, "w", newline="") as fp:
fp.write(text)
return path
def start(self, name, config, ports=()):
"""Write a configuration, run it, and wait for its ports to open."""
path = self.write_config(name, config)
logfile = os.path.join(self.tmpdir, name + ".out")
with open(logfile, "wb") as out:
proc = subprocess.Popen([self.binary, path], stdout=out,
stderr=subprocess.STDOUT)
server = Server(name, path, proc, logfile)
self.servers.append(server)
for port in ports:
if not self.wait_port(port):
raise Failure(
f"{name} never listened on port {port}\n"
f"--- configuration ---\n{open(path).read()}"
f"--- output ---\n{server.output()}")
return server
def run_config(self, name, config):
"""Run a configuration expected to be rejected; return its output."""
path = self.write_config(name, config)
done = subprocess.run([self.binary, path], stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, timeout=15)
return done.stdout.decode("utf-8", "replace")
def wait_port(self, port, timeout=5.0):
deadline = time.time() + timeout
while time.time() < deadline:
try:
with socket.create_connection(("127.0.0.1", port), 0.25):
return True
except OSError:
time.sleep(0.02)
return False
def stop_all(self):
for server in self.servers:
server.stop()
self.servers = []
# ---- requests ----------------------------------------------------
def http(self, url, proxy=None, socks=None, socks4=False,
remote_dns=False, method="GET", body=None, headers=None,
auth=None, proxy_auth=None, tunnel=False, conn=None):
"""Make a request, directly or through a proxy, and read the reply.
proxy "host:port" of an HTTP proxy
socks "host:port" of a SOCKS proxy
tunnel reach the origin with CONNECT rather than an absolute URI
conn reuse a connection returned by connection()
"""
host, port, path = self._split(url)
headers = dict(headers or {})
if auth:
headers["Authorization"] = self._basic(auth)
if proxy_auth:
headers["Proxy-Authorization"] = self._basic(proxy_auth)
own = conn is None
try:
if own:
conn = self.connection(host, port, proxy=proxy, socks=socks,
socks4=socks4, remote_dns=remote_dns,
tunnel=tunnel)
target = path
if proxy and not tunnel:
target = f"http://{host}:{port}{path}"
if body is not None and not isinstance(body, bytes):
body = body.encode()
conn.request(method, target, body=body, headers=headers)
reply = conn.getresponse()
data = reply.read()
return Response(reply.status, data, dict(reply.getheaders()))
except (OSError, http.client.HTTPException) as exc:
return Response(error=f"{type(exc).__name__}: {exc}")
finally:
if own and conn is not None:
try:
conn.close()
except OSError:
pass
def connection(self, host, port, proxy=None, socks=None, socks4=False,
remote_dns=False, tunnel=False):
"""A connection to an origin, kept open for reuse."""
if socks:
shost, sport = self._hostport(socks)
sock = self._socks_connect(shost, sport, host, port,
socks4=socks4, remote_dns=remote_dns)
conn = http.client.HTTPConnection(host, port, timeout=self.timeout)
conn.sock = sock
return conn
if proxy:
phost, pport = self._hostport(proxy)
conn = http.client.HTTPConnection(phost, pport, timeout=self.timeout)
if tunnel:
conn.set_tunnel(host, port)
return conn
return http.client.HTTPConnection(host, port, timeout=self.timeout)
def raw(self, port, request, host="127.0.0.1"):
"""Send bytes as they are and return whatever comes back."""
if not isinstance(request, bytes):
request = request.encode("latin-1")
try:
with socket.create_connection((host, port), self.timeout) as sock:
sock.settimeout(self.timeout)
sock.sendall(request)
chunks = []
while True:
try:
piece = sock.recv(65536)
except OSError:
# a timeout, or a reset once the server is done:
# either way keep whatever already arrived
break
if not piece:
break
chunks.append(piece)
return b"".join(chunks).decode("utf-8", "replace")
except OSError as exc:
return f"<no reply: {exc}>"
# ---- SOCKS -------------------------------------------------------
def _socks_connect(self, shost, sport, host, port, socks4=False,
remote_dns=False, auth=None):
sock = socket.create_connection((shost, sport), self.timeout)
sock.settimeout(self.timeout)
try:
if socks4:
addr = socket.inet_aton(socket.gethostbyname(host))
sock.sendall(b"\x04\x01" + struct.pack("!H", port) + addr + b"\x00")
reply = self._recvall(sock, 8)
if len(reply) < 2 or reply[1] != 0x5a:
raise OSError("SOCKS4 request refused")
return sock
if auth:
sock.sendall(b"\x05\x02\x00\x02")
else:
sock.sendall(b"\x05\x01\x00")
reply = self._recvall(sock, 2)
if len(reply) < 2 or reply[0] != 5:
raise OSError("SOCKS5 handshake failed")
if reply[1] == 0x02:
if not auth:
raise OSError("SOCKS5 server demands credentials")
user, password = auth
sock.sendall(b"\x01" + bytes([len(user)]) + user.encode() +
bytes([len(password)]) + password.encode())
status = self._recvall(sock, 2)
if len(status) < 2 or status[1] != 0:
raise OSError("SOCKS5 credentials refused")
elif reply[1] != 0x00:
raise OSError("SOCKS5 offered no acceptable method")
if remote_dns:
target = b"\x03" + bytes([len(host)]) + host.encode()
else:
target = b"\x01" + socket.inet_aton(socket.gethostbyname(host))
sock.sendall(b"\x05\x01\x00" + target + struct.pack("!H", port))
reply = self._recvall(sock, 4)
if len(reply) < 4 or reply[1] != 0:
raise OSError("SOCKS5 request refused")
self._read_socks_addr(sock, reply[3])
return sock
except Exception:
sock.close()
raise
def socks_connect(self, socks, host, port, socks4=False, remote_dns=False,
auth=None):
"""Open a SOCKS connection, reporting failure rather than raising."""
shost, sport = self._hostport(socks)
try:
sock = self._socks_connect(shost, sport, host, port, socks4=socks4,
remote_dns=remote_dns, auth=auth)
sock.close()
return None
except OSError as exc:
return str(exc)
def socks_http(self, socks, url, auth=None, **kwargs):
"""A request through SOCKS, with optional SOCKS credentials."""
host, port, path = self._split(url)
shost, sport = self._hostport(socks)
try:
sock = self._socks_connect(shost, sport, host, port, auth=auth,
**kwargs)
except OSError as exc:
return Response(error=str(exc))
conn = http.client.HTTPConnection(host, port, timeout=self.timeout)
conn.sock = sock
try:
conn.request("GET", path)
reply = conn.getresponse()
return Response(reply.status, reply.read(), dict(reply.getheaders()))
except (OSError, http.client.HTTPException) as exc:
return Response(error=str(exc))
finally:
conn.close()
def socks_udp_associate(self, port, host="127.0.0.1"):
"""Ask for a UDP association and report the port handed back.
That socket is allocated per association, which is where an intport
range has to take effect.
"""
try:
with socket.create_connection((host, port), self.timeout) as sock:
sock.settimeout(self.timeout)
sock.sendall(b"\x05\x01\x00")
if self._recvall(sock, 2) != b"\x05\x00":
return None
sock.sendall(b"\x05\x03\x00\x01\x00\x00\x00\x00" +
struct.pack("!H", 0))
reply = self._recvall(sock, 4)
if len(reply) < 4 or reply[1] != 0:
return None
_, bound = self._read_socks_addr(sock, reply[3])
return bound
except OSError:
return None
def _read_socks_addr(self, sock, atyp):
if atyp == 1:
addr = socket.inet_ntoa(self._recvall(sock, 4))
elif atyp == 3:
length = self._recvall(sock, 1)[0]
addr = self._recvall(sock, length).decode()
elif atyp == 4:
addr = self._recvall(sock, 16).hex()
else:
raise OSError(f"unknown SOCKS address type {atyp}")
port = struct.unpack("!H", self._recvall(sock, 2))[0]
return addr, port
@staticmethod
def _recvall(sock, count):
data = b""
while len(data) < count:
piece = sock.recv(count - len(data))
if not piece:
break
data += piece
return data
# ---- helpers -----------------------------------------------------
@staticmethod
def _basic(credentials):
user, password = credentials
token = base64.b64encode(f"{user}:{password}".encode()).decode()
return "Basic " + token
@staticmethod
def _hostport(value):
host, _, port = value.rpartition(":")
return host or "127.0.0.1", int(port)
@staticmethod
def _split(url):
prefix = "http://"
if url.startswith(prefix):
url = url[len(prefix):]
authority, _, path = url.partition("/")
host, _, port = authority.rpartition(":")
return host or "127.0.0.1", int(port), "/" + path
# ---- assertions --------------------------------------------------
def _record(self, passed, label, expected=None, actual=None):
self.checks.append((passed, label, expected, actual))
return passed
def ok(self, label):
return self._record(True, label)
def fail(self, label, expected=None, actual=None):
return self._record(False, label, expected, actual)
def eq(self, expected, actual, label):
return self._record(expected == actual, label, expected, actual)
def ne(self, unexpected, actual, label):
return self._record(unexpected != actual, label,
f"anything but {unexpected!r}", actual)
def contains(self, haystack, needle, label):
if isinstance(haystack, Response):
haystack = haystack.text
return self._record(needle in haystack, label,
f"text containing {needle!r}", self._clip(haystack))
def not_contains(self, haystack, needle, label):
if isinstance(haystack, Response):
haystack = haystack.text
return self._record(needle not in haystack, label,
f"text without {needle!r}", self._clip(haystack))
def in_range(self, value, low, high, label):
good = isinstance(value, int) and low <= value <= high
return self._record(good, label, f"between {low} and {high}", value)
def not_in_range(self, value, low, high, label):
good = isinstance(value, int) and not (low <= value <= high)
return self._record(good, label, f"outside {low}-{high}", value)
def skip(self, label):
self._skipped += 1
self.checks.append((None, label, None, None))
@staticmethod
def _clip(text, limit=200):
text = str(text).replace("\r\n", " ").replace("\n", " ")
return text[:limit] + ("..." if len(text) > limit else "")
def field(response, name):
"""Pull one 'key=value' line out of an echo reply."""
text = response.text if isinstance(response, Response) else response
for line in text.splitlines():
key, _, value = line.partition("=")
if key == name:
return value
return None
def int_field(response, name):
value = field(response, name)
try:
return int(value)
except (TypeError, ValueError):
return None

141
tests/run.py Normal file
View File

@ -0,0 +1,141 @@
#!/usr/bin/env python3
"""Run the 3proxy regression tests.
python3 tests/run.py every case
python3 tests/run.py httpsrv cases whose name matches
python3 tests/run.py --bin build/bin/3proxy
python3 tests/run.py --keep leave the temporary files behind
Each case under tests/cases/ defines the configurations it needs and the
positive and negative scenarios expected from them.
"""
import argparse
import importlib.util
import os
import shutil
import sys
import tempfile
import traceback
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from harness import Failure, Tester # noqa: E402
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def default_binary():
"""Find a built 3proxy: the Makefiles put it in bin/, CMake in build/bin/,
and multi-configuration generators one level below that again."""
name = "3proxy.exe" if os.name == "nt" else "3proxy"
candidates = [os.path.join(ROOT, "bin", name),
os.path.join(ROOT, "build", "bin", name)]
for config in ("Release", "Debug", "RelWithDebInfo", "MinSizeRel"):
candidates.append(os.path.join(ROOT, "build", "bin", config, name))
for candidate in candidates:
if os.path.isfile(candidate):
return candidate
return candidates[0]
def load_case(path):
name = os.path.splitext(os.path.basename(path))[0]
spec = importlib.util.spec_from_file_location("case_" + name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return name, module
def main():
parser = argparse.ArgumentParser()
parser.add_argument("pattern", nargs="?", default="",
help="only run cases whose name contains this")
parser.add_argument("--bin", dest="binary", default=None,
help="the 3proxy binary to test")
parser.add_argument("--keep", action="store_true",
help="keep the temporary directory")
parser.add_argument("-v", "--verbose", action="store_true",
help="print every check, not just the failures")
args = parser.parse_args()
binary = args.binary or os.environ.get("BIN") or default_binary()
binary = os.path.abspath(binary)
if not os.path.isfile(binary):
print(f"no 3proxy binary at {binary} (build first, or pass --bin)",
file=sys.stderr)
return 2
case_dir = os.path.join(ROOT, "tests", "cases")
paths = sorted(os.path.join(case_dir, f) for f in os.listdir(case_dir)
if f.endswith(".py") and not f.startswith("_"))
paths = [p for p in paths if args.pattern in os.path.basename(p)]
if not paths:
print(f"no cases matched {args.pattern!r}", file=sys.stderr)
return 2
tmpdir = tempfile.mkdtemp(prefix="3proxy-tests.")
print(f"3proxy tests: {binary}")
print(f"working in: {tmpdir}\n")
passed = failed = skipped = 0
failures = []
try:
for path in paths:
name, module = load_case(path)
print(f" {name}")
tester = Tester(binary, tmpdir, name)
error = None
try:
module.run(tester)
except Failure as exc:
error = str(exc)
except Exception:
error = traceback.format_exc()
finally:
tester.stop_all()
for status, label, expected, actual in tester.checks:
if status is None:
skipped += 1
print(f" skip {label}")
elif status:
passed += 1
if args.verbose:
print(f" ok {label}")
else:
failed += 1
failures.append(f"{name}: {label}")
print(f" FAIL {label}")
if expected is not None:
print(f" expected: {expected}")
if actual is not None:
print(f" actual: {actual}")
if error:
failed += 1
failures.append(f"{name}: case aborted")
print(" ERROR the case could not finish:")
for line in error.rstrip().splitlines():
print(f" {line}")
print()
finally:
if args.keep:
print(f"temporary files left in {tmpdir}")
else:
shutil.rmtree(tmpdir, ignore_errors=True)
print("-" * 41)
total = passed + failed
summary = f"cases: {len(paths)} checks: {total} passed: {passed} failed: {failed}"
if skipped:
summary += f" skipped: {skipped}"
print(summary)
for item in failures:
print(f" FAIL {item}")
return 1 if failed else 0
if __name__ == "__main__":
sys.exit(main())