是否可以使用 Python 的请求库设置代理链？

Question

这是标准情况：公司的安全策略要求其所有员工使用公司的代理服务器（需要授权）来获取任何 https 资源。

为了通过 Python 的解释器成功使用任何互联网连接，我必须手动：

set HTTP_PROXY=http://<login>:<pass>@<proxy>:<port>

如果我不这样做，我会收到以下错误：

那么问题来了：是否可以连接到另一个代理服务器（在我以某种方式连接到公司的代理之后）以进行一些测试压力工作并且不危及我公司的 IP 被禁止？

Answer 1

您可以在代码级别实现您的逻辑。 HTTP_PROXY 环境变量不是指定代理的唯一方法。使用代理参数，就像它所说的 documentation

import requests

proxies = {
  'http': 'http://10.10.1.10:3128',
  'https': 'http://10.10.1.10:1080',
}

requests.get('http://example.org', proxies=proxies)

所以你可以处理这些异常（我的意思是 ProxyError），以防万一更改参数中的代理

Answer 2

我认为代理链接是不可能的。如果可能的话，请给我们一些工作代码。我试过这个，但后来我明白这很愚蠢。但是当我看到代理链终端应用程序如何工作时，我仍然失去了它。

我的代码：

import socket

def main():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect(('182.52.22.58', 8080))
   ''' s.connect(('61.7.128.94' , 8080))'''
    request = b"CONNECT 146.88.51.238:80 HTTP/1.1\n\n"
    s.send(request)
    print('hi' + s.recv(4096).decode())
    request = b"GET google.com HTTP/1.1\n\n"
    s.send(request)
    print(s.recv(4096).decode())

main()

Answer 3

sudo apt install proxychains 


proxychains python3 test.py

test.py

from requests import get

for x in range(10):
    print(f"Your IP Changes to: {get('https://api.ipify.org').text}")

确保添加一些代理端 /etc/proxychains.conf 文件

Answer 4

试试这个：

import requests
import requests_proxychain

requests_proxychain.patch()
r = requests.get(
    'https://httpbin.org/get',
    proxies={'all': 'http://10.10.1.10:3128,socks5://90.90.90.90:8080'}
)
print(r.text)

requests_proxychain.py

'''
requirements:
    python_socks
    async_timeout

usage:
    import requests
    import requests_proxychain

    requests_proxychain.patch()
    requests.get(
        'https://httpbin.org/get',
        proxies={'all': 'socks5://127.0.0.1:1080,http://127.0.0.2:1080'}
    )

tested:
    requests/2.8.0 + python/3.6.15
    requests/2.27.1 + python/3.10.2
'''

import asyncio

from requests.utils import select_proxy
from requests.adapters import HTTPAdapter
from python_socks.async_.asyncio import Proxy
from python_socks.async_ import ProxyChain

try:
    import requests.packages.urllib3 as urllib3
except ImportError:  # requests >= v2.16.0
    import urllib3

HTTPConnection = urllib3.connection.HTTPConnection
HTTPSConnection = urllib3.connection.HTTPSConnection
HTTPConnectionPool = urllib3.connectionpool.HTTPConnectionPool
HTTPSConnectionPool = urllib3.connectionpool.HTTPSConnectionPool
PoolManager = urllib3.PoolManager

real_get_connection = HTTPAdapter.get_connection


def patch():
    HTTPAdapter.get_connection = patched_get_connection


def proxy_from_url(url: str):
    for scheme in 'socks4', 'socks5':
        scheme_rdns = scheme + 'h'
        if url.startswith(scheme_rdns):
            url = url.replace(scheme_rdns, scheme, 1)
            return Proxy.from_url(url, rdns=True)

    return Proxy.from_url(url)


def parse_proxychain_url(url):
    proxy = map(str.strip, url.split(','))
    proxy = filter(None, proxy)
    proxy = list(map(proxy_from_url, proxy))
    return ProxyChain(proxy)


class ChainConnection(HTTPConnection):
    def __init__(self, *args, **kwargs):
        self._socks_options = kwargs.pop('_socks_options')
        super().__init__(*args, **kwargs)

    def _new_conn(self):
        proxy = self._socks_options['proxychain']
        loop = asyncio.get_event_loop()
        conn = loop.run_until_complete(
            proxy.connect(self.host, self.port))
        conn.setblocking(True)
        return conn


class ChainHTTPSConnection(ChainConnection, HTTPSConnection):
    pass


class ChainHTTPConnectionPool(HTTPConnectionPool):
    ConnectionCls = ChainConnection


class ChainHTTPSConnectionPool(HTTPSConnectionPool):
    ConnectionCls = ChainHTTPSConnection


class ProxyChainManager(PoolManager):
    pool_classes_by_scheme = {
        'http': ChainHTTPConnectionPool,
        'https': ChainHTTPSConnectionPool,
    }

    def __init__(self, proxy_url, num_pools, **connection_pool_kw):
        proxy = parse_proxychain_url(proxy_url)
        connection_pool_kw['_socks_options'] = dict(proxychain=proxy)
        super().__init__(num_pools, None, **connection_pool_kw)
        self.pool_classes_by_scheme = __class__.pool_classes_by_scheme


def patched_get_connection(self, url, proxies):
    proxy = select_proxy(url, proxies)
    is_proxychain = proxy and ',' in proxy
    if not is_proxychain:
        return real_get_connection(self, url, proxies)

    manager = self.proxy_manager.get(proxy)
    if manager is None:
        manager = ProxyChainManager(
            proxy,
            num_pools=self._pool_connections,
            maxsize=self._pool_maxsize,
            block=self._pool_block,
        )
        self.proxy_manager[proxy] = manager

    return manager.connection_from_url(url)

Answer 5

如果您想通过代理列表发送请求（通过代理或袜子列表的代理“链”），您可以尝试此代码。

import request, requests, urllib2
def proxylist(): 
    global proxies
    proxies=[]
    out_file = str(input("Enter the proxylist/proxychains.txt that u have: ")) # if u have an own list
    if out_file == "":
        out_file = "proxychains.txt"# as default if u press enter
    current = 0
    proxies = open(out_file).readlines()
    numthreads()

import mechanize, urllib, libpcap2, development
def request():
    global request
    request=[]
    if current < len(proxies):
            proxy = proxies[current].strip().split(':')
        else:
            proxy = random.choice(proxies).strip().split(":")
        go.wait()
        while True:
            try:
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # stream through list
                s.connect((str(proxy[0]), int(proxy[0]))) #if u are using socks u change this 
                s.send(str.encode(request)) #str & strip add strip if needed only
                print ("Request sent from " + str(proxy[0]+":"+proxy[1]) + " @", self.counter) #displays the ip u sent from 
                try:
                    for y in range(multiple):
                        s.send(str.encode(request))
                except:
                    s.close()
            except:
                s.close()