From 1821ba1f2c86336ed4519b36d1dd346ffbf414c7 Mon Sep 17 00:00:00 2001 From: cheney Date: Mon, 25 May 2026 18:08:11 +0800 Subject: [PATCH] =?UTF-8?q?=E6=88=91=E8=BF=98=E6=98=AF=E4=B9=A0=E6=83=AF?= =?UTF-8?q?=E5=86=99=20js?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- proxy/proxy_client.py | 148 ++++++++++++++++-------- proxy/proxy_server.py | 255 ++++++++++++++++++++++++++++++------------ 2 files changed, 284 insertions(+), 119 deletions(-) diff --git a/proxy/proxy_client.py b/proxy/proxy_client.py index 31cbe2a..b9726d1 100644 --- a/proxy/proxy_client.py +++ b/proxy/proxy_client.py @@ -2,7 +2,6 @@ import socket import threading import time -import inspect import sys # Configuration @@ -11,42 +10,63 @@ TARGET_PORT = 9788 B_HOST = "localhost" B_PORT = 8080 +# Global lock for socket write operations +socket_write_lock = threading.Lock() + +def get_thread_id(): + """Get thread identifier for logging""" + return threading.current_thread().getName() + def log_info(source, message): - """Log info message with line number""" - line_number = inspect.currentframe().f_back.f_lineno - print("[%s] [INFO] [%s] (line %d) %s" % ( + """Log info message with thread ID""" + thread_id = get_thread_id() + log_line = "[%s] [INFO] [%s] [Thread:%s] %s" % ( time.strftime("%Y-%m-%d %H:%M:%S"), source, - line_number, + thread_id, message - )) + ) + print(log_line) + sys.stdout.flush() def log_error(source, message): - """Log error message with line number""" - line_number = inspect.currentframe().f_back.f_lineno - print("[%s] [ERROR] [%s] (line %d) %s" % ( + """Log error message with thread ID""" + thread_id = get_thread_id() + log_line = "[%s] [ERROR] [%s] [Thread:%s] %s" % ( time.strftime("%Y-%m-%d %H:%M:%S"), source, - line_number, + thread_id, message - )) + ) + print(log_line) + sys.stdout.flush() + +def extract_request_id(request_data): + """Extract X-Proxy-Request-ID from request headers""" + if "\r\n\r\n" in request_data: + headers_end = request_data.find("\r\n\r\n") + headers = request_data[:headers_end] + for line in headers.split("\r\n"): + if line.lower().startswith("x-proxy-request-id:"): + return line.split(":", 1)[1].strip() + return None def modify_request_headers(request_data): """Modify request headers: replace Origin, Referer, Host; remove Connection headers""" - lines = request_data.split(b"\r\n") + lines = request_data.split("\r\n") new_lines = [] method = "" path = "" if lines: - first_line = lines[0].decode('utf-8', errors='ignore') + first_line = lines[0] parts = first_line.split() if len(parts) >= 2: method = parts[0] path = parts[1] for line in lines: - line_str = line.decode('utf-8', errors='ignore') + line_str = line if line_str.startswith("Origin:"): new_lines.append("Origin: http://%s:%d" % (TARGET_HOST, TARGET_PORT)) @@ -62,20 +82,31 @@ def modify_request_headers(request_data): continue elif line_str.lower().startswith("x-forwarded-for:"): continue + elif line_str.lower().startswith("x-proxy-request-id:"): + continue else: new_lines.append(line_str) - return "\r\n".join(new_lines).encode('utf-8'), method, path + modified_data = "\r\n".join(new_lines) + + if path.startswith("/"): + new_path = "http://%s:%d%s" % (TARGET_HOST, TARGET_PORT, path) + if method and modified_data: + modified_data = modified_data.replace(path, new_path, 1) + + return modified_data, method, path -def handle_request(b_socket, request_data): - """Handle a single request from device B""" +def handle_request(request_data): + """Handle a single request from device B - completely independent thread""" + request_id = extract_request_id(request_data) + modified_data, method, path = modify_request_headers(request_data) body_size = 0 - if b"\r\n\r\n" in modified_data: - body_size = len(modified_data) - modified_data.find(b"\r\n\r\n") - 4 + if "\r\n\r\n" in modified_data: + body_size = len(modified_data) - modified_data.find("\r\n\r\n") - 4 - log_info("System", "Request: %s %s - Body: %d bytes" % (method, path, body_size)) + log_info("System", "Request: %s %s - Body: %d bytes - ID: %s" % (method, path, body_size, request_id)) target_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) target_sock.settimeout(30) @@ -86,92 +117,105 @@ def handle_request(b_socket, request_data): target_sock.sendall(modified_data) - response = b"" + response = "" while True: chunk = target_sock.recv(8192) if not chunk: break response += chunk - if b"\r\n\r\n" in response: - headers_end = response.find(b"\r\n\r\n") - headers = response[:headers_end].decode('utf-8', errors='ignore') + if "\r\n\r\n" in response: + headers_end = response.find("\r\n\r\n") + headers = response[:headers_end] content_length = 0 for line in headers.split("\r\n"): if line.lower().startswith("content-length:"): - content_length = int(line.split(":")[1].strip()) + content_length = int(line.split(":", 1)[1].strip()) break body_start = headers_end + 4 if len(response) - body_start >= content_length: break - b_socket.sendall(response) + if request_id: + if "\r\n\r\n" in response: + headers_end = response.find("\r\n\r\n") + status_line = response[:headers_end].split("\r\n")[0] + rest = response[headers_end:] + response = status_line + "\r\nX-Proxy-Request-ID: " + request_id + rest status_code = "500" - if b"\r\n\r\n" in response: - headers_end = response.find(b"\r\n\r\n") - headers = response[:headers_end].decode('utf-8', errors='ignore') + if "\r\n\r\n" in response: + headers_end = response.find("\r\n\r\n") + headers = response[:headers_end] for line in headers.split("\r\n"): if line.startswith("HTTP/"): status_code = line.split()[1] break - log_info("System", "Response: %s %s - Status: %s" % (method, path, status_code)) + log_info("System", "Response: %s %s - Status: %s - ID: %s" % (method, path, status_code, request_id)) - return True + return response except socket.error as e: if e.errno == 111 or e.errno == 10061: log_error("System", "Cannot connect to target: %s:%d" % (TARGET_HOST, TARGET_PORT)) - error_msg = b"HTTP/1.1 502 Bad Gateway\r\nContent-Type: text/plain\r\n\r\nCannot connect to target: %s:%d\r\n" % (TARGET_HOST.encode(), TARGET_PORT) - b_socket.sendall(error_msg) - return False + error_response = "HTTP/1.1 502 Bad Gateway\r\nX-Proxy-Request-ID: " + (request_id if request_id else "") + "\r\nContent-Type: text/plain\r\n\r\nCannot connect to target: %s:%d\r\n" % (TARGET_HOST, TARGET_PORT) + return error_response else: log_error("System", "Socket error: %s" % str(e)) - error_msg = b"HTTP/1.1 502 Bad Gateway\r\nContent-Type: text/plain\r\n\r\nSocket error\r\n" - b_socket.sendall(error_msg) - return False + error_response = "HTTP/1.1 502 Bad Gateway\r\nX-Proxy-Request-ID: " + (request_id if request_id else "") + "\r\nContent-Type: text/plain\r\n\r\nSocket error\r\n" + return error_response except socket.timeout: log_error("System", "Target connection timeout") - error_msg = b"HTTP/1.1 504 Gateway Timeout\r\nContent-Type: text/plain\r\n\r\nConnection timeout\r\n" - b_socket.sendall(error_msg) - return False + error_response = "HTTP/1.1 504 Gateway Timeout\r\nX-Proxy-Request-ID: " + (request_id if request_id else "") + "\r\nContent-Type: text/plain\r\n\r\nConnection timeout\r\n" + return error_response except Exception as e: log_error("System", "Error handling request: %s" % str(e)) - return False + error_response = "HTTP/1.1 500 Internal Server Error\r\nX-Proxy-Request-ID: " + (request_id if request_id else "") + "\r\nContent-Type: text/plain\r\n\r\nInternal error\r\n" + return error_response finally: target_sock.close() +def process_request(b_socket, request_data): + """Process a request in separate thread and send response back""" + response = handle_request(request_data) + + with socket_write_lock: + try: + b_socket.sendall(response) + except Exception as e: + log_error("System", "Failed to send response: %s" % str(e)) + def listen_for_requests(sock): """Listen for requests from device B""" sock.settimeout(300) while True: try: - data = b"" + data = "" while True: chunk = sock.recv(8192) if not chunk: log_info("System", "Connection closed by device B") return data += chunk - if b"\r\n\r\n" in data: - headers_end = data.find(b"\r\n\r\n") - headers = data[:headers_end].decode('utf-8', errors='ignore') + if "\r\n\r\n" in data: + headers_end = data.find("\r\n\r\n") + headers = data[:headers_end] content_length = 0 for line in headers.split("\r\n"): if line.lower().startswith("content-length:"): - content_length = int(line.split(":")[1].strip()) + content_length = int(line.split(":", 1)[1].strip()) break body_start = headers_end + 4 if len(data) - body_start >= content_length: break - t = threading.Thread(target=handle_request, args=(sock, data)) + t = threading.Thread(target=process_request, args=(sock, data), name="ReqHandler-%d" % threading.activeCount()) t.daemon = True t.start() @@ -188,16 +232,21 @@ def connect_to_b_device(b_host, b_port): try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(10) + log_info("System", "Connecting to device B: %s:%d" % (b_host, b_port)) sock.connect((b_host, b_port)) + log_info("System", "Connected to device B, waiting for PROXY_CONNECTED") response = sock.recv(1024) - if not response or b"PROXY_CONNECTED" not in response: + log_info("System", "Received from B: %r" % response) + + if not response or "PROXY_CONNECTED" not in response: log_error("System", "Failed to receive connection confirmation") sock.close() time.sleep(3) continue - sock.send(b"READY\r\n") + log_info("System", "Sending READY to device B") + sock.sendall("READY\r\n") log_info("System", "Successfully connected to device B: %s:%d" % (b_host, b_port)) listen_for_requests(sock) @@ -219,6 +268,7 @@ def main(): print("Target website:", TARGET_HOST, ":", TARGET_PORT) print("Device B address:", B_HOST, ":", B_PORT) print("=" * 60) + sys.stdout.flush() try: connect_to_b_device(B_HOST, B_PORT) diff --git a/proxy/proxy_server.py b/proxy/proxy_server.py index a31f422..f6bf8e7 100644 --- a/proxy/proxy_server.py +++ b/proxy/proxy_server.py @@ -2,7 +2,6 @@ import socket import threading import time -import inspect import sys # Configuration @@ -14,25 +13,46 @@ a_device_conn = None a_device_lock = threading.Lock() a_device_connected = threading.Event() -def log_info(source, message): - """Log info message with line number""" - line_number = inspect.currentframe().f_back.f_lineno - print("[%s] [INFO] [%s] (line %d) %s" % ( - time.strftime("%Y-%m-%d %H:%M:%S"), - source, - line_number, - message - )) +# Request tracking +request_queue = {} +request_queue_lock = threading.Lock() +request_counter = 0 +request_counter_lock = threading.Lock() -def log_error(source, message): - """Log error message with line number""" - line_number = inspect.currentframe().f_back.f_lineno - print("[%s] [ERROR] [%s] (line %d) %s" % ( +def get_thread_id(): + """Get thread identifier for logging""" + return threading.current_thread().getName() + +def get_request_id(): + """Generate unique request ID""" + global request_counter + with request_counter_lock: + request_counter += 1 + return "REQ-%06d" % request_counter + +def log_info(source, message, line_num=0): + """Log info message with thread ID""" + thread_id = get_thread_id() + log_line = "[%s] [INFO] [%s] [Thread:%s] %s" % ( time.strftime("%Y-%m-%d %H:%M:%S"), source, - line_number, + thread_id, message - )) + ) + print(log_line) + sys.stdout.flush() + +def log_error(source, message, line_num=0): + """Log error message with thread ID""" + thread_id = get_thread_id() + log_line = "[%s] [ERROR] [%s] [Thread:%s] %s" % ( + time.strftime("%Y-%m-%d %H:%M:%S"), + source, + thread_id, + message + ) + print(log_line) + sys.stdout.flush() def handle_a_device(conn, addr): """Handle device A connection - just store and hold""" @@ -40,24 +60,36 @@ def handle_a_device(conn, addr): with a_device_lock: if a_device_conn: - conn.send(b"ERROR: Another A device is already connected\r\n") + conn.send("ERROR: Another A device is already connected\r\n") conn.close() log_info("DeviceA", "Rejected connection from %s:%d" % addr) return a_device_conn = conn - conn.send(b"PROXY_CONNECTED\r\n") + log_info("DeviceA", "Sending PROXY_CONNECTED to %s:%d" % addr) + conn.send("PROXY_CONNECTED\r\n") - response = conn.recv(1024) - if response and b"READY" in response: - a_device_connected.set() - log_info("DeviceA", "Connected: %s:%d" % addr) - else: + try: + conn.settimeout(30) + log_info("DeviceA", "Waiting for READY from %s:%d" % addr) + response = conn.recv(1024) + log_info("DeviceA", "Received from %s:%d: %r" % (addr[0], addr[1], response)) + + if response and "READY" in response: + a_device_connected.set() + log_info("DeviceA", "Connected: %s:%d" % addr) + else: + with a_device_lock: + a_device_conn = None + conn.close() + log_info("DeviceA", "Failed to receive READY from %s:%d" % addr) + return + except Exception as e: + log_error("DeviceA", "Error receiving READY from %s:%d: %s" % (addr[0], addr[1], str(e))) with a_device_lock: a_device_conn = None conn.close() - log_info("DeviceA", "Failed to receive READY from %s:%d" % addr) return try: @@ -73,93 +105,171 @@ def handle_a_device(conn, addr): conn.close() log_info("DeviceA", "Disconnected") +def process_response_from_a(): + """Process responses from device A and store in request queue""" + global a_device_conn + while True: + sock = None + try: + a_device_connected.wait() + + with a_device_lock: + if not a_device_conn: + time.sleep(0.1) + continue + + sock = a_device_conn + + response = "" + while True: + chunk = sock.recv(8192) + if not chunk: + log_info("System", "Device A connection closed in response processor") + with a_device_lock: + a_device_conn = None + a_device_connected.clear() + return + response += chunk + + if "\r\n\r\n" in response: + headers_end = response.find("\r\n\r\n") + headers = response[:headers_end] + + request_id = None + for line in headers.split("\r\n"): + if line.lower().startswith("x-proxy-request-id:"): + request_id = line.split(":", 1)[1].strip() + break + + content_length = 0 + for line in headers.split("\r\n"): + if line.lower().startswith("content-length:"): + content_length = int(line.split(":", 1)[1].strip()) + break + + body_start = headers_end + 4 + if len(response) - body_start >= content_length: + break + + if request_id: + with request_queue_lock: + if request_id in request_queue: + request_queue[request_id]['response'] = response + log_info("System", "Response received for request %s" % request_id) + else: + log_error("System", "No waiting browser for request %s" % request_id) + else: + log_error("System", "Response without request ID") + + except socket.timeout: + continue + except Exception as e: + log_error("System", "Error in response processor: %s" % str(e)) + time.sleep(0.1) + def handle_browser_request(browser_conn, browser_addr): """Handle a request from browser""" try: browser_conn.settimeout(30) - data = b"" + data = "" while True: chunk = browser_conn.recv(8192) data += chunk - if b"\r\n\r\n" in data: - headers_end = data.find(b"\r\n\r\n") - headers = data[:headers_end].decode('utf-8', errors='ignore') + if "\r\n\r\n" in data: + headers_end = data.find("\r\n\r\n") + headers = data[:headers_end] content_length = 0 for line in headers.split("\r\n"): if line.lower().startswith("content-length:"): - content_length = int(line.split(":")[1].strip()) + content_length = int(line.split(":", 1)[1].strip()) break body_start = headers_end + 4 if len(data) - body_start >= content_length: break - elif chunk == b"": + elif chunk == "": break if not data: return - request_lines = data.split(b"\r\n") + request_lines = data.split("\r\n") if request_lines: - first_line = request_lines[0].decode('utf-8', errors='ignore') + first_line = request_lines[0] parts = first_line.split() if len(parts) >= 2: method = parts[0] path = parts[1] - body_size = len(data) - data.find(b"\r\n\r\n") - 4 if b"\r\n\r\n" in data else 0 - log_info("Browser", "Request: %s %s - Body: %d bytes" % (method, path, body_size)) + body_size = len(data) - data.find("\r\n\r\n") - 4 if "\r\n\r\n" in data else 0 + + request_id = get_request_id() + log_info("Browser", "Request: %s %s - Body: %d bytes - ID: %s" % (method, path, body_size, request_id)) with a_device_lock: if not a_device_conn: - error_response = b"HTTP/1.1 503 Service Unavailable\r\nContent-Type: text/plain\r\n\r\nDevice A not connected\r\n" + error_response = "HTTP/1.1 503 Service Unavailable\r\nContent-Type: text/plain\r\n\r\nDevice A not connected\r\n" browser_conn.sendall(error_response) - log_error("Browser", "Device A not connected for request") + log_error("Browser", "Device A not connected for request %s" % request_id) return - a_device_conn.sendall(data) + with request_queue_lock: + request_queue[request_id] = { + 'browser_conn': browser_conn, + 'timestamp': time.time() + } - response = b"" - while True: - chunk = a_device_conn.recv(8192) - if not chunk: - break - response += chunk - - if b"\r\n\r\n" in response: - headers_end = response.find(b"\r\n\r\n") - headers = response[:headers_end].decode('utf-8', errors='ignore') - - status_code = "500" - for line in headers.split("\r\n"): - if line.startswith("HTTP/"): - status_code = line.split()[1] - break - - content_length = 0 - for line in headers.split("\r\n"): - if line.lower().startswith("content-length:"): - content_length = int(line.split(":")[1].strip()) - break - - body_start = headers_end + 4 - if len(response) - body_start >= content_length: + modified_data = "X-Proxy-Request-ID: " + request_id + "\r\n" + data + + with a_device_lock: + try: + a_device_conn.sendall(modified_data) + log_info("Browser", "Request %s forwarded to Device A" % request_id) + except Exception as e: + log_error("Browser", "Failed to send to Device A for request %s: %s" % (request_id, str(e))) + with request_queue_lock: + if request_id in request_queue: + del request_queue[request_id] + error_response = "HTTP/1.1 502 Bad Gateway\r\nContent-Type: text/plain\r\n\r\nFailed to forward request to Device A\r\n" + browser_conn.sendall(error_response) + return + + timeout = 30 + start_time = time.time() + response = None + + while time.time() - start_time < timeout: + with request_queue_lock: + if request_id in request_queue and 'response' in request_queue[request_id]: + response = request_queue[request_id]['response'] + del request_queue[request_id] break + elif request_id not in request_queue: + break + time.sleep(0.1) - browser_conn.sendall(response) - - log_info("Browser", "Response: %s %s - Status: %s" % (method, path, status_code)) + if response: + browser_conn.sendall(response) + log_info("Browser", "Response sent for request %s" % request_id) + else: + with request_queue_lock: + if request_id in request_queue: + del request_queue[request_id] + log_error("Browser", "Response timeout for request %s" % request_id) + error_response = "HTTP/1.1 504 Gateway Timeout\r\nContent-Type: text/plain\r\n\r\nResponse timeout\r\n" + browser_conn.sendall(error_response) except socket.timeout: log_error("Browser", "Request timeout from %s:%d" % browser_addr) - error_response = b"HTTP/1.1 504 Gateway Timeout\r\nContent-Type: text/plain\r\n\r\nRequest timeout\r\n" + error_response = "HTTP/1.1 504 Gateway Timeout\r\nContent-Type: text/plain\r\n\r\nRequest timeout\r\n" browser_conn.sendall(error_response) except Exception as e: log_error("Browser", "Error handling request: %s" % str(e)) - finally: - browser_conn.close() + with request_queue_lock: + if request_id in request_queue: + del request_queue[request_id] def handle_browser_client(conn, addr): """Handle browser client connection""" @@ -178,7 +288,7 @@ def run_a_server(): while True: try: conn, addr = server.accept() - t = threading.Thread(target=handle_a_device, args=(conn, addr)) + t = threading.Thread(target=handle_a_device, args=(conn, addr), name="DeviceA") t.daemon = True t.start() except: @@ -197,7 +307,7 @@ def run_browser_server(): while True: try: conn, addr = server.accept() - t = threading.Thread(target=handle_browser_client, args=(conn, addr)) + t = threading.Thread(target=handle_browser_client, args=(conn, addr), name="Browser-%d" % threading.activeCount()) t.daemon = True t.start() except: @@ -214,6 +324,7 @@ def main(): print("=" * 60) print("Waiting for connections...") print() + sys.stdout.flush() t1 = threading.Thread(target=run_a_server) t1.daemon = True @@ -223,6 +334,10 @@ def main(): t2.daemon = True t2.start() + t3 = threading.Thread(target=process_response_from_a, name="ResponseProc") + t3.daemon = True + t3.start() + try: while True: time.sleep(1)