通過已驗證的HTTP代理建立TCP連接？

Question

我在一個網絡上，需要我通過經過身份驗證的HTTP代理連接才能訪問網絡外的任何內容。我需要做的基本上是使socket（或同等）連接到Internet，但通過代理髮送所有數據，而不是直接發送它。任何想法如何做到這一點？

Answer 1

因爲我沒有找到任何可以使用的實際模塊或其他代碼，所以我最終編寫了自己的函數連接通過代理：

def http_proxy_connect(address, proxy = None, auth = None, headers = {}): 
    """ 
    Establish a socket connection through an HTTP proxy. 

    Arguments: 
    address (required)  = The address of the target 
    proxy (def: None)  = The address of the proxy server 
    auth (def: None)  = A tuple of the username and password used for authentication 
    headers (def: {})  = A set of headers that will be sent to the proxy 

    Returns: 
    A 3-tuple of the format: 
     (socket, status_code, headers) 
    Where `socket' is the socket object, `status_code` is the HTTP status code that the server 
    returned and `headers` is a dict of headers that the server returned. 
    """ 
    import socket 
    import base64 

    def valid_address(addr): 
    """ Verify that an IP/port tuple is valid """ 
    return isinstance(addr, (list, tuple)) and len(addr) == 2 and isinstance(addr[0], str) and isinstance(addr[1], (int, long)) 

    if not valid_address(address): 
    raise ValueError('Invalid target address') 

    if proxy == None: 
    s = socket.socket() 
    s.connect(address) 
    return s, 0, {} 

    if not valid_address(proxy): 
    raise ValueError('Invalid proxy address') 

    headers = { 
    'host': address[0] 
    } 

    if auth != None: 
    if isinstance(auth, str): 
     headers['proxy-authorization'] = auth 
    elif auth and isinstance(auth, (tuple, list)): 
     if len(auth) == 1: 
     raise ValueError('Invalid authentication specification') 

     t = auth[0] 
     args = auth[1:] 

     if t.lower() == 'basic' and len(args) == 2: 
     headers['proxy-authorization'] = 'Basic ' + base64.b64encode('%s:%s' % args) 
     else: 
     raise ValueError('Invalid authentication specification') 
    else: 
     raise ValueError('Invalid authentication specification') 

    s = socket.socket() 
    s.connect(proxy) 
    fp = s.makefile('r+') 

    fp.write('CONNECT %s:%d HTTP/1.0\r\n' % address) 
    fp.write('\r\n'.join('%s: %s' % (k, v) for (k, v) in headers.items()) + '\r\n\r\n') 
    fp.flush() 

    statusline = fp.readline().rstrip('\r\n') 

    if statusline.count(' ') < 2: 
    fp.close() 
    s.close() 
    raise IOError('Bad response') 
    version, status, statusmsg = statusline.split(' ', 2) 
    if not version in ('HTTP/1.0', 'HTTP/1.1'): 
    fp.close() 
    s.close() 
    raise IOError('Unsupported HTTP version') 
    try: 
    status = int(status) 
    except ValueError: 
    fp.close() 
    s.close() 
    raise IOError('Bad response') 

    response_headers = {} 

    while True: 
    tl = '' 
    l = fp.readline().rstrip('\r\n') 
    if l == '': 
     break 
    if not ':' in l: 
     continue 
    k, v = l.split(':', 1) 
    response_headers[k.strip().lower()] = v.strip() 

    fp.close() 
    return (s, status, response_headers) 

Answer 2

我不確定這是否會有很大的幫助，但你看看pycurl？這可能會幫助您連接到提供用戶名/密碼認證系統的代理（請參閱this和this）