I don't know why I did this, but I think you might find it useful:
# tor.py
from tor_proxy import tor_proxy
import requests
from requests.exceptions import InvalidSchema
import atexit, subprocess
class TorProxy:
"""Utility for spinning up a Tor SOCKS proxy and using it with requests."""
def __init__(self):
# start Tor and remember the port that was allocated
self.port = tor_proxy()
self.proxies = {
"http": f"socks5h://127.0.0.1:{self.port}",
"https": f"socks5h://127.0.0.1:{self.port}",
}
# ensure we kill any lingering tor processes when the program exits
atexit.register(self._shutdown)
def get(self, target_url, timeout=30, **kwargs):
"""Perform a GET request using the Tor proxy."""
try:
return requests.get(target_url, proxies=self.proxies, timeout=timeout, **kwargs)
except InvalidSchema:
print("Error: SOCKS support is not available. Please install the required dependencies:")
print(" pip install requests[socks] or pip install PySocks")
raise
def session(self):
"""Return a requests.Session object pre-configured with the Tor proxies."""
s = requests.Session()
s.proxies.update(self.proxies)
return s
def _shutdown(self):
"""Called on program exit to make sure the tor proxy is torn down."""
# try running the CLI kill switch in case a detached process exists
try:
subprocess.run(["tor-proxy", "--kill"], check=False)
except OSError:
# command not found / cannot be executed
pass
# also perform cleanup on the active onion instance if available
from tor_proxy.tor_proxy import get_onion_instance
onion = get_onion_instance()
if onion:
try:
onion.cleanup()
except Exception:
pass
if __name__ == "__main__":
# simple standalone example, same as before
tor = TorProxy()
url = "https://api.ipify.org"
r = tor.get(url)
print(r.text)import requests # easy_install requests
import logging
# import the TorProxy class we just created
from tor import TorProxy
tor = TorProxy()
logging.basicConfig(level=logging.INFO)
# quick connectivity check: call ipify directly and via the Tor proxy
test_url = "https://api.ipify.org"
direct_ip = requests.get(test_url, timeout=30).text
proxied_ip = requests.get(test_url, proxies=tor.proxies, timeout=30).text
logging.info(f"direct request returned: {direct_ip}")
logging.info(f"proxied request returned: {proxied_ip}")
if direct_ip == proxied_ip:
logging.error("IP addresses match!")
raise Exception("IP addresses match! TOR not working!")
else:
logging.info("Proxy is working.")
try:
logging.info("Sending request...")
### INSERT CODE HERE ###
logging.info(f"Request result: {result}")
except Exception as e:
logging.error(f"Error occurred: {e}")
finally:
logging.info("Tor proxy shut down.")