强制请求使用 IPv4 / IPv6

问题描述 投票：0回答：8

如何强制 

requests
库对 get 请求使用特定的互联网协议版本？或者用 Python 中的另一种方法可以更好地实现这一点？我可以，但我不想使用
curl
…

阐明目的的示例：

import requests
r = requests.get('https://my-dyn-dns-service.domain/?hostname=my.domain',
                 auth = ('myUserName', 'my-password'))
python python-requests network-programming ipv6 ipv4
8个回答
31
投票

我找到了一个简单的解决方案来强制 urrlib3 使用 ipv4 或 ipv6。 urrlib3 使用此方法为 Http 和 Https 创建新连接。您可以在其中指定您想要使用的任何 AF_FAMILY。

import socket
import requests.packages.urllib3.util.connection as urllib3_cn
    
   
def allowed_gai_family():
    """
     https://github.com/shazow/urllib3/blob/master/urllib3/util/connection.py
    """
    family = socket.AF_INET
    if urllib3_cn.HAS_IPV6:
        family = socket.AF_INET6 # force ipv6 only if it is available
    return family

urllib3_cn.allowed_gai_family = allowed_gai_family
22
投票

您可以使用此 hack 强制 

requests
使用 IPv4：

requests.packages.urllib3.util.connection.HAS_IPV6 = False

（这是 Nulano 作为评论提交的，我认为它应该是一个正确的答案）。

20
投票

这是一个 hack，但您可以对 getaddrinfo 进行猴子修补以仅过滤 IPv4 地址：

# Monkey patch to force IPv4, since FB seems to hang on IPv6
import socket
old_getaddrinfo = socket.getaddrinfo
def new_getaddrinfo(*args, **kwargs):
    responses = old_getaddrinfo(*args, **kwargs)
    return [response
            for response in responses
            if response[0] == socket.AF_INET]
socket.getaddrinfo = new_getaddrinfo
6
投票

我为 

requests
+
urllib3
+
socket
编写了一个运行时补丁，允许根据每个请求选择性地传递所需的地址族。

与其他解决方案不同，不涉及猴子补丁，而是用修补的文件替换导入的

requests
，它提供了一个
request
兼容的接口，所有公开的类都被子类化和修补，并且所有“简单API”功能都被重新实现。唯一明显的区别应该是存在一个额外的 
family
参数，您可以使用该参数将名称解析期间使用的地址族限制为 
socket.AF_INET
或 
socket.AF_INET6
。然后，使用一系列有点复杂（但大多只是 LoC 密集型）的策略方法覆盖来将该值一直传递到 
urllib3
的底层，在那里它将用于 
socket.create_connection
函数调用的替代实现.

TL;DR 用法如下：

import socket

from . import requests_wrapper as requests  # Use this load the patch


# This will work (if IPv6 connectivity is available) …
requests.get("http://ip6only.me/", family=socket.AF_INET6)
# … but this won't
requests.get("http://ip6only.me/", family=socket.AF_INET)

# This one will fail as well
requests.get("http://127.0.0.1/", family=socket.AF_INET6)

# This one will work if you have IPv4 available
requests.get("http://ip6.me/", family=socket.AF_INET)

# This one will work on both IPv4 and IPv6 (the default)
requests.get("http://ip6.me/", family=socket.AF_UNSPEC)

补丁库的完整链接（~350 LoC）：https://gitlab.com/snippets/1900824

3
投票

我采用了与 https://stackoverflow.com/a/33046939/5059062 类似的方法，但是修补了 

socket
中发出 DNS 请求的部分，因此它仅执行 IPv6 或 IPv4，对于每个 请求，这意味着它可以在 urllib
 中使用，就像在
requests
 中一样有效。

如果你的程序也使用unix管道和其他类似的东西，这可能会很糟糕，所以我强烈建议谨慎使用monkeypatching。

import requests import socket from unittest.mock import patch import re orig_getaddrinfo = socket.getaddrinfo def getaddrinfoIPv6(host, port, family=0, type=0, proto=0, flags=0): return orig_getaddrinfo(host=host, port=port, family=socket.AF_INET6, type=type, proto=proto, flags=flags) def getaddrinfoIPv4(host, port, family=0, type=0, proto=0, flags=0): return orig_getaddrinfo(host=host, port=port, family=socket.AF_INET, type=type, proto=proto, flags=flags) with patch('socket.getaddrinfo', side_effect=getaddrinfoIPv6): r = requests.get('http://ip6.me') print('ipv6: '+re.search(r'\+3>(.*?)</',r.content.decode('utf-8')).group(1)) with patch('socket.getaddrinfo', side_effect=getaddrinfoIPv4): r = requests.get('http://ip6.me') print('ipv4: '+re.search(r'\+3>(.*?)</',r.content.decode('utf-8')).group(1))
并且没有

requests

:

import urllib.request import socket from unittest.mock import patch import re orig_getaddrinfo = socket.getaddrinfo def getaddrinfoIPv6(host, port, family=0, type=0, proto=0, flags=0): return orig_getaddrinfo(host=host, port=port, family=socket.AF_INET6, type=type, proto=proto, flags=flags) def getaddrinfoIPv4(host, port, family=0, type=0, proto=0, flags=0): return orig_getaddrinfo(host=host, port=port, family=socket.AF_INET, type=type, proto=proto, flags=flags) with patch('socket.getaddrinfo', side_effect=getaddrinfoIPv6): r = urllib.request.urlopen('http://ip6.me') print('ipv6: '+re.search(r'\+3>(.*?)</',r.read().decode('utf-8')).group(1)) with patch('socket.getaddrinfo', side_effect=getaddrinfoIPv4): r = urllib.request.urlopen('http://ip6.me') print('ipv4: '+re.search(r'\+3>(.*?)</',r.read().decode('utf-8')).group(1))
在3.5.2中测试

1
投票
这完全未经测试，可能需要一些调整，但是结合

使用Python“请求”与现有套接字连接和如何强制python httplib库仅使用A请求的答案，看起来你应该能够创建一个仅限 IPv6 的套接字，然后让请求将其用于其连接池，例如：

try: from http.client import HTTPConnection except ImportError: from httplib import HTTPConnection class MyHTTPConnection(HTTPConnection): def connect(self): print("This actually called called") self.sock = socket.socket(socket.AF_INET6) self.sock.connect((self.host, self.port,0,0)) if self._tunnel_host: self._tunnel() requests.packages.urllib3.connectionpool.HTTPConnection = MyHTTPConnection
1
投票
读完前面的答案后，我不得不修改代码以强制使用 IPv4 而不是 IPv6。请注意，我使用了 socket.AF_INET 而不是 socket.AF_INET6，并且 self.sock.connect() 有 2 项元组参数。

我还需要重写 HTTP

SConnection，它与 HTTPConnection 有很大不同，因为 requests

 包装了 httplib.HTTPSConnection 以验证证书
ssl
 模块是否可用。

import socket import ssl try: from http.client import HTTPConnection except ImportError: from httplib import HTTPConnection from requests.packages.urllib3.connection import VerifiedHTTPSConnection # HTTP class MyHTTPConnection(HTTPConnection): def connect(self): self.sock = socket.socket(socket.AF_INET) self.sock.connect((self.host, self.port)) if self._tunnel_host: self._tunnel() requests.packages.urllib3.connectionpool.HTTPConnection = MyHTTPConnection requests.packages.urllib3.connectionpool.HTTPConnectionPool.ConnectionCls = MyHTTPConnection # HTTPS class MyHTTPSConnection(VerifiedHTTPSConnection): def connect(self): self.sock = socket.socket(socket.AF_INET) self.sock.connect((self.host, self.port)) if self._tunnel_host: self._tunnel() self.sock = ssl.wrap_socket(self.sock, self.key_file, self.cert_file) requests.packages.urllib3.connectionpool.HTTPSConnection = MyHTTPSConnection requests.packages.urllib3.connectionpool.VerifiedHTTPSConnection = MyHTTPSConnection requests.packages.urllib3.connectionpool.HTTPSConnectionPool.ConnectionCls = MyHTTPSConnection
0
投票
根据 Yuri 的回答，这里有一些简单的获取和设置 ip 版本函数，可以处理请求。

用途：

import requests set_ip_version(6) requests.get(...) set_ip_version(4) 
来源：


import sockets
import requests.packages.urllib3.util.connection


def get_ip_version():
    # type: () -> Optional[int]
    ip_version = requests.packages.urllib3.util.connection.allowed_gai_family()
    if ip_version == socket.AF_INET6:
        return 6
    elif ip_version == socket.AF_INET:
        return 4
    else:
        return None


def set_ip_version(ip_version: int = 6):
    # type: (str) -> bool
    """
    Sets preferred IP protocol version to use in requests / ofunctions
    Attention: this propagates
    By default, AF_INET6 is used which preferes IPv6 but fallbacks to IPv4
    """
    if ip_version == 4:

        def allowed_gai_family():
            return socket.AF_INET

    elif ip_version == 6:

        def allowed_gai_family():
            if requests.packages.urllib3.util.connection.HAS_IPV6:
                return socket.AF_INET6
            return socket.AF_INET

    else:
        return False

    requests.packages.urllib3.util.connection.allowed_gai_family = allowed_gai_family
    return True

这些函数存在于 ofunctions.network 库中，随 
pip install ofunctions.network
 安装。然后您可以通过
致电他们

from ofunctions.network import set_ip_version

set_ip_version(6)

    



      

      
    

  

  

    
最新问题


  





  

    © www.soinside.com 2019 - 2024. All rights reserved.