2

我所在的网络需要我通过经过身份验证的 HTTP 代理进行连接才能访问网络外的任何内容。我需要做的基本上是制作一个socket（或等效的）连接到互联网，但通过代理发送所有数据而不是尝试直接发送。关于如何做到这一点的任何想法？

4

2 回答 2

3

由于我没有找到任何可以使用的实际模块或其他代码，我最终编写了自己的通过代理连接的函数：

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)
于 2012-11-22T15:33:21.620 回答
1

我不确定这是否会有很大帮助，但你看过pycurl吗？这可以帮助您连接到提供用户名/密码身份验证系统的代理（请参阅此和此）

于 2012-11-20T19:06:57.907 回答