mirror of
https://github.com/3proxy/3proxy.git
synced 2026-10-28 22:05:48 +08:00
8cc7d255af
A -r service waited up to 60 s for the go-ahead on the connection it made, without looking at a reload. One replaced by a reload went on waiting, holding a slot and a thread, and took a client if the far side's go-ahead came in that time; every reload left another such connection in the far side's queue. The wait is now a second at a time (60 s in all, as before): at a reload the service takes a go-ahead already there, its client going on in the service which replaced it, and otherwise closes the connection and stops. The far side (-R) sent its go-ahead to any connection it accepted, and a closed one cost it a client. A connection the -r side closed, at a reload or after its wait, is readable before any go-ahead, since that side sends nothing first: once past the ACL it is passed over, without counting as one of the tries for as many as the listen queue holds (1 + maxchild/16), so that connections made to be closed or to send something cannot keep a client's thread in the loop. Tests: reattach's far side passes over closed connections as -R does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
172 lines
7.0 KiB
Python
172 lines
7.0 KiB
Python
"""Connections of old services after a reload (unless noforce): each goes by
|
|
the rules of the newest service on its listener, taking one of its maxconn
|
|
slots, and is dropped where that cannot be."""
|
|
|
|
import base64
|
|
import select
|
|
import socket
|
|
import threading
|
|
import time
|
|
|
|
|
|
def run(t):
|
|
origin = t.free_port()
|
|
proxy = t.free_port()
|
|
adm = t.free_port()
|
|
|
|
def config(service_line, maxconn=50):
|
|
return f"""
|
|
log
|
|
auth iponly
|
|
allow *
|
|
http data * /data
|
|
httpsrv -p{origin}
|
|
|
|
flush
|
|
maxconn {maxconn}
|
|
auth iponly
|
|
allow *
|
|
{service_line}
|
|
|
|
flush
|
|
maxconn 50
|
|
auth iponly
|
|
allow *
|
|
admin -p{adm}
|
|
"""
|
|
|
|
server = t.start("reattach", config(f"proxy -p{proxy}"), ports=[origin, proxy, adm])
|
|
|
|
def reload(conf):
|
|
before = server.output().count("Accepting connections")
|
|
t.write_config("reattach", conf)
|
|
t.http(f"http://127.0.0.1:{adm}/R")
|
|
deadline = time.time() + 10
|
|
while time.time() < deadline and server.output().count("Accepting connections") < before + 3:
|
|
time.sleep(0.05)
|
|
|
|
def downloads(n):
|
|
"""n downloads of about two seconds each through the proxy, started
|
|
now; the results fill in as they finish."""
|
|
results = [{} for _ in range(n)]
|
|
|
|
def one(r):
|
|
got = t.http(f"http://127.0.0.1:{origin}/data?size=400000&block=8192&delay=40",
|
|
proxy=f"127.0.0.1:{proxy}")
|
|
r["length"] = got.length
|
|
|
|
threads = [threading.Thread(target=one, args=(r,)) for r in results]
|
|
for th in threads:
|
|
th.start()
|
|
time.sleep(0.4)
|
|
return threads, results
|
|
|
|
# --- the newest service has one slot free: one connection moves, one is
|
|
# cut. A service holds one of its maxconn slots for the next connection
|
|
# it accepts, so maxconn 2 leaves one.
|
|
threads, results = downloads(2)
|
|
reload(config(f"proxy -p{proxy}", maxconn=2))
|
|
for th in threads:
|
|
th.join(60)
|
|
whole = sum(1 for r in results if r.get("length") == 400000)
|
|
t.eq(1, whole, "with one slot free in the new service one download goes on")
|
|
t.contains(server.output(), " 00027 ",
|
|
"the other is cut for want of a slot (27)")
|
|
|
|
# --- the service on the port is of another type now: cut (26)
|
|
reload(config(f"proxy -p{proxy}"))
|
|
threads, results = downloads(1)
|
|
reload(config(f"socks -p{proxy}"))
|
|
for th in threads:
|
|
th.join(60)
|
|
t.ok("a proxy download is cut when a socks service takes the port") \
|
|
if results[0].get("length", 0) < 400000 else \
|
|
t.fail("a proxy download is cut when a socks service takes the port",
|
|
"less than 400000 bytes", results[0].get("length"))
|
|
t.contains(server.output(), " 00026 ", "as its service is another one now (26)")
|
|
|
|
# --- services of one configuration on one address share its socket, so
|
|
# they must be of one type with the same options: otherwise the file is
|
|
# refused and the configuration in force stays ---------------------------
|
|
for line, why in ((f"proxy -p{proxy}\n socks -p{proxy}",
|
|
"a service of another type listens there"),
|
|
(f"proxy -p{proxy}\n proxy -olSO_KEEPALIVE -p{proxy}",
|
|
"a service with other socket options listens there")):
|
|
mark = len(server.output())
|
|
t.write_config("reattach", config(line))
|
|
t.http(f"http://127.0.0.1:{adm}/R")
|
|
deadline = time.time() + 10
|
|
while time.time() < deadline and why not in server.output()[mark:]:
|
|
time.sleep(0.05)
|
|
t.contains(server.output()[mark:], why, f"refused: {why}")
|
|
# the socks service of the case before is still the one there
|
|
t.ne(200, t.http(f"http://127.0.0.1:{origin}/data?size=100", proxy=f"127.0.0.1:{proxy}").status,
|
|
"the configuration in force stays: the socks service, no HTTP proxy")
|
|
reload(config(f"proxy -p{proxy}\n proxy -p{proxy}"))
|
|
ok = sum(1 for _ in range(20)
|
|
if t.http(f"http://127.0.0.1:{origin}/data?size=100", proxy=f"127.0.0.1:{proxy}").status == 200)
|
|
t.eq(20, ok, "two proxy services on one address both serve, sharing its socket")
|
|
|
|
# --- a service connecting back (-r) has no socket, but a listener for the
|
|
# place it connects to, here shared by two services connecting there: its
|
|
# clients are re-authenticated all the same ---
|
|
peer = socket.socket()
|
|
peer.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
peer.bind(("127.0.0.1", 0))
|
|
peer.listen(16)
|
|
peer.settimeout(10)
|
|
cb = peer.getsockname()[1]
|
|
|
|
def cbconfig(users):
|
|
return config(f"proxy -p{proxy}").replace(
|
|
"flush\n maxconn 50\n auth iponly\n allow *\n admin",
|
|
f"flush\n users {users}\n auth strong\n allow *\n"
|
|
f" proxy -r127.0.0.1:{cb}\n proxy -r127.0.0.1:{cb}\n\n"
|
|
" flush\n maxconn 50\n"
|
|
" auth iponly\n allow *\n admin")
|
|
|
|
def cb_download(result):
|
|
"""A request over a connection the -r service made: the peer is its
|
|
client once it has sent C."""
|
|
try:
|
|
# as the far side of a connect-back service does: a connection
|
|
# closed by a service which stopped waiting is passed over
|
|
while True:
|
|
c, _ = peer.accept()
|
|
if not select.select([c], [], [], 0)[0]:
|
|
break
|
|
c.close()
|
|
c.settimeout(30)
|
|
c.sendall(b"C")
|
|
auth = base64.b64encode(b"alice:one").decode()
|
|
c.sendall((f"GET http://127.0.0.1:{origin}/data?size=400000&block=8192&delay=40 HTTP/1.1\r\n"
|
|
f"Host: 127.0.0.1:{origin}\r\nProxy-Authorization: Basic {auth}\r\n"
|
|
"Connection: close\r\n\r\n").encode())
|
|
got = b""
|
|
while True:
|
|
b = c.recv(65536)
|
|
if not b:
|
|
break
|
|
got += b
|
|
c.close()
|
|
result["body"] = len(got.split(b"\r\n\r\n", 1)[1]) if b"\r\n\r\n" in got else 0
|
|
except OSError as e:
|
|
result["error"] = str(e)
|
|
|
|
reload(cbconfig("alice:CL:one"))
|
|
for keep, users, want in ((True, "alice:CL:one bob:CL:two", "finishes whole"),
|
|
(False, "bob:CL:two", "is cut")):
|
|
result = {}
|
|
th = threading.Thread(target=cb_download, args=(result,))
|
|
th.start()
|
|
time.sleep(0.5)
|
|
reload(cbconfig(users))
|
|
th.join(60)
|
|
whole = result.get("body") == 400000
|
|
name = f"a download through a connect-back service {want} when alice is " + \
|
|
("kept" if keep else "removed")
|
|
t.ok(name) if whole == keep else t.fail(name, "whole" if keep else "cut", result)
|
|
if keep:
|
|
reload(cbconfig("alice:CL:one"))
|
|
peer.close()
|