Registry / http-networking / socketswap

socketswap

JSON →
library0.1.11pypypi✓ verified 82d ago

SocketSwap is a Python package designed to proxy traffic from any third-party libraries through a local TCP SOCKS5 Proxy. It is especially useful for applications and libraries that do not offer native proxying capabilities. The current version, 0.1.11, was released on September 29, 2023, and the project is generally considered stable.

pip install socketswap
INSTALL
IMPORT
SIG · SOCKETSWAP
S
socketswap
http-networkingpythonv0.1.11
Install
1.6s avg
Import
Disk
16MB
Pass rate
10/ 10
Env Coverage10 / 10
glibc
3.93.13
musl
3.93.13
Install & Compatibility
Where this runs
tested against v0.1.11 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
musl
py 3.103.920 runs
installs and imports cleanly · install 0.0s · import 0.000s · 17.8MB
glibc
py 3.103.920 runs
installs and imports cleanly · install 1.6s · import 0.000s · 18MB
16MB installed
● package 16MB
Code
Verified usage

Verified import paths — ran on the pinned version, not inferred.

SocketSwapContext
from SocketSwap import SocketSwapContext
from SocketSwap import ProxySwapContext
context_manager
from SocketSwap import context_manager
start_local_proxy
from SocketSwap import start_local_proxy

This quickstart demonstrates how to initialize and use `ProxySwapContext`. It sets up a local TCP proxy that redirects traffic to a 'dummy remote server'. A client then connects to the local proxy, and its traffic is routed to the remote server via SocketSwap. Logging is also configured for better visibility into the proxy's operations.

import socket import logging import threading import time from SocketSwap import ProxySwapContext # Configure SocketSwap logging for visibility socket_swap_logger = logging.getLogger("SocketSwap") socket_swap_logger.addHandler(logging.StreamHandler()) socket_swap_logger.setLevel(logging.DEBUG) # --- Dummy Remote Server (what ProxySwap will eventually connect to) --- def dummy_remote_server(host, port): server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind((host, port)) server_socket.listen(1) print(f"Dummy Remote Server listening on {host}:{port}") try: conn, addr = server_socket.accept() with conn: print(f"Dummy Remote Server: Connected by {addr}") data = conn.recv(1024) if data: print(f"Dummy Remote Server: Received {data.decode()}") conn.sendall(b"Hello from Remote Server") except Exception as e: print(f"Dummy Remote Server error: {e}") finally: server_socket.close() # --- Socket Factory for ProxySwapContext --- def create_remote_socket(target_host, target_port): remote_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) remote_socket.connect((target_host, target_port)) return remote_socket # --- Main application logic using ProxySwapContext --- def main(): REMOTE_HOST = "127.0.0.1" REMOTE_PORT = 54321 # Port for the dummy remote server LOCAL_PROXY_HOST = "127.0.0.1" LOCAL_PROXY_PORT = 22222 # Port for the local proxy # Start the dummy remote server in a separate thread server_thread = threading.Thread(target=dummy_remote_server, args=(REMOTE_HOST, REMOTE_PORT)) server_thread.daemon = True # Allow main program to exit even if server thread is running server_thread.start() time.sleep(0.5) # Give server a moment to start print(f"Starting ProxySwapContext on {LOCAL_PROXY_HOST}:{LOCAL_PROXY_PORT}") try: with ProxySwapContext( socket_factory=create_remote_socket, socket_factory_args=(REMOTE_HOST, REMOTE_PORT), local_proxy_host=LOCAL_PROXY_HOST, local_proxy_port=LOCAL_PROXY_PORT ): print("ProxySwapContext is active. Connecting client to local proxy...") # --- Dummy Client connecting to the local proxy --- client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.connect((LOCAL_PROXY_HOST, LOCAL_PROXY_PORT)) client_socket.sendall(b"Hello via ProxySwap") response = client_socket.recv(1024) print(f"Client received from local proxy (and ultimately remote server): {response.decode()}") client_socket.close() print("Client activity complete.") print("ProxySwapContext exited.") except Exception as e: print(f"An error occurred during ProxySwap operation: {e}") if __name__ == "__main__": main()
Debug
Known issues
gotchaThe `ProxySwapContext` constructor requires exactly four mandatory arguments: `socket_factory`, `socket_factory_args`, `local_proxy_host`, and `local_proxy_port`. Omitting any of these or providing incorrect types will lead to errors.
fix
Ensure all four arguments are provided to `ProxySwapContext`, with `socket_factory_args` being an iterable (e.g., a tuple or list) even if empty.
affects: 0.1.0+
gotchaLogging for SocketSwap is not enabled by default. To see debug information and operational details, you must manually configure the 'SocketSwap' logger with a handler and a logging level (e.g., `logging.DEBUG`).
fix
Add `import logging; socket_swap_logger = logging.getLogger("SocketSwap"); socket_swap_logger.addHandler(logging.StreamHandler()); socket_swap_logger.setLevel(logging.DEBUG)` at the start of your application.
affects: 0.1.0+
gotchaSocketSwap is designed for libraries that *do not natively support proxying*. Using it with libraries that already have built-in proxy configurations (e.g., `requests`) might lead to unexpected behavior or conflicts, or simply be redundant.
fix
Evaluate if the target library truly lacks proxy support before integrating SocketSwap. For libraries with native proxy settings, use their dedicated configurations first.
affects: 0.1.0+
gotchaAs SocketSwap operates on low-level sockets, misconfigurations or rapid restarts of proxied services can lead to 'Address already in use' errors. This typically occurs if a previous socket instance hasn't fully released the port.
fix
For server-side sockets in your `socket_factory` or `local_proxy_host`, consider setting `socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)` to allow the reuse of an address that is in a `TIME_WAIT` state. Ensure proper closing of sockets when they are no longer needed.
affects: 0.1.0+
Upgrade
Version history
0.1.11latest on PyPI · released Sep 29, 2023
Audit
Dependencies

No dependency data recorded yet.

Agent activity
12 hits · last 30 days
node
10
OpenAI (training)
1
Resources