txtorcon

import os from twisted.internet.task import react from twisted.internet.defer import inlineCallbacks, ensureDeferred from twisted.internet.endpoints import UNIXClientEndpoint, TCP4ClientEndpoint import treq import txtorcon @react @inlineCallbacks def main(reactor): # Connect to a running Tor instance, e.g., Tor Browser Bundle's control port (default 9151) # Or a system-wide Tor daemon (default 9051, or /var/run/tor/control) # You can also use txtorcon.launch(reactor) to start a new Tor process managed by txtorcon. # Example: Connect to a Unix socket (common for system Tor) control_endpoint = UNIXClientEndpoint(reactor, '/var/run/tor/control') # Example: Connect to a TCP port (common for TBB or custom Tor setup) # control_endpoint = TCP4ClientEndpoint(reactor, 'localhost', 9151) # Password for Tor control port, if required. Use environment variable for security. # For a newly launched Tor, this is usually not needed immediately. password = os.environ.get('TOR_CONTROL_PASSWORD', '') try: tor = yield txtorcon.connect( reactor, control_endpoint, password_function=lambda: password ) print(f"Connected to Tor version {tor.version}") url = u'https://www.torproject.org:443' print(f"Downloading {repr(url)} via Tor...") # Use tor.web_agent() to make requests over Tor's general circuit resp = yield treq.get(url, agent=tor.web_agent()) body_data = yield resp.text() print(f"Got {len(body_data)} bytes from {url}:") print(body_data[:200] + ('...' if len(body_data) > 200 else '')) # Print first 200 chars print("\nCreating a new Tor circuit...") state = yield tor.create_state() circ = yield state.build_circuit() yield circ.when_built() print(f"New circuit built with path: {' -> '.join([r.ip for r in circ.path])}") except Exception as e: print(f"An error occurred: {e}") # Handle specific connection errors or Tor failures finally: # It's good practice to disconnect or shut down Tor if launched by txtorcon if hasattr(tor, 'shutdown') and callable(tor.shutdown): print("Shutting down Tor (if launched by txtorcon)...") yield ensureDeferred(tor.shutdown()) else: print("Not shutting down Tor, it was an external instance.")