SocketSwap

Warnings

• gotcha The `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.
• gotcha Logging 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.
• gotcha SocketSwap 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.
• gotcha As 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.

Install

• pip install socketswap Install stable version

Imports

• ProxySwapContext

  from SocketSwap import ProxySwapContext

Quickstart

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()