Python socket.AF_UNSPEC属性代码示例

本文整理汇总了Python中socket.AF_UNSPEC属性的典型用法代码示例。如果您正苦于以下问题:Python socket.AF_UNSPEC属性的具体用法?Python socket.AF_UNSPEC怎么用?Python socket.AF_UNSPEC使用的例子?那么恭喜您, 这里精选的属性代码示例或许可以为您提供帮助。您也可以进一步了解该属性所在模块socket的用法示例。

在下文中一共展示了socket.AF_UNSPEC属性的23个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: _do_dns_lookup

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def _do_dns_lookup(hostname: str, port: int) -> str:
    try:
        addr_infos = socket.getaddrinfo(hostname, port, socket.AF_UNSPEC, socket.IPPROTO_IP)
    except (socket.gaierror, IndexError, ConnectionError):
        raise ServerHostnameCouldNotBeResolved(f"Could not resolve {hostname}")
    family, socktype, proto, canonname, sockaddr = addr_infos[0]
    # By default use the first DNS entry, IPv4 or IPv6
    tentative_ip_addr = sockaddr[0]
    # But try to use IPv4 if we have both IPv4 and IPv6 addresses, to work around buggy networks
    for family, socktype, proto, canonname, sockaddr in addr_infos:
        if family == socket.AF_INET:
            tentative_ip_addr = sockaddr[0]
    return tentative_ip_addr 
开发者ID:nabla-c0d3,项目名称:sslyze,代码行数:19,代码来源:server_setting.py


示例2: is_valid_ip

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def is_valid_ip(ip):
    """Returns true if the given string is a well-formed IP address.
    Supports IPv4 and IPv6.
    """
    if not ip or '\x00' in ip:
        # getaddrinfo resolves empty strings to localhost, and truncates
        # on zero bytes.
        return False
    try:
        res = socket.getaddrinfo(ip, 0, socket.AF_UNSPEC,
                                 socket.SOCK_STREAM,
                                 0, socket.AI_NUMERICHOST)
        return bool(res)
    except socket.gaierror as e:
        if e.args[0] == socket.EAI_NONAME:
            return False
        raise
    return True 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:21,代码来源:netutil.py


示例3: connect

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def connect(self, host, port, af=socket.AF_UNSPEC, ssl_options=None,
                max_buffer_size=None):
        """Connect to the given host and port.
        Asynchronously returns an `.IOStream` (or `.SSLIOStream` if
        ``ssl_options`` is not None).
        """
        addrinfo = yield self.resolver.resolve(host, port, af)
        connector = _Connector(
            addrinfo, self.io_loop,
            functools.partial(self._create_stream, max_buffer_size))
        af, addr, stream = yield connector.start()
        # TODO: For better performance we could cache the (af, addr)
        # information here and re-use it on subsequent connections to
        # the same host. (http://tools.ietf.org/html/rfc6555#section-4.2)
        if ssl_options is not None:
            stream = yield stream.start_tls(False, ssl_options=ssl_options,
                                            server_hostname=host)
        raise gen.Return(stream) 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:21,代码来源:tcpclient.py


示例4: bind

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def bind(self, port, address=None, family=socket.AF_UNSPEC, backlog=128):
        u"""绑定该服务到指定的地址的指定端口上.
        要启动该服务, 调用 `start`. 如果你想要在一个单进程上运行该服务,
        你可以调用 `listen` 作为顺序调用 `bind` 和 `start` 的一个快捷方式.
        address 参数可以是 IP 地址或者主机名.  如果它是主机名,
        该服务将监听在和该名称有关的所有 IP 地址上.  地址也可以是空字符串或者
        None, 服务将监听所有可用的接口. family 可以被设置为 `socket.AF_INET` 或
        `socket.AF_INET6` 用来限定是 IPv4 或 IPv6 地址, 否则如果可用的话, 两者
        都将被使用.
        ``backlog`` 参数和 `socket.listen <socket.socket.listen>` 是相同含义.
        这个方法可能在 `start` 之前被调用多次来监听在多个端口或接口上.
        """
        sockets = bind_sockets(port, address=address, family=family,
                               backlog=backlog)
        if self._started:
            self.add_sockets(sockets)
        else:
            self._pending_sockets.extend(sockets) 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:24,代码来源:tcpserver.py


示例5: select_address_family

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def select_address_family(host, port):
    """Return ``AF_INET4``, ``AF_INET6``, or ``AF_UNIX`` depending on
    the host and port."""
    # disabled due to problems with current ipv6 implementations
    # and various operating systems.  Probably this code also is
    # not supposed to work, but I can't come up with any other
    # ways to implement this.
    # try:
    #     info = socket.getaddrinfo(host, port, socket.AF_UNSPEC,
    #                               socket.SOCK_STREAM, 0,
    #                               socket.AI_PASSIVE)
    #     if info:
    #         return info[0][0]
    # except socket.gaierror:
    #     pass
    if host.startswith("unix://"):
        return socket.AF_UNIX
    elif ":" in host and hasattr(socket, "AF_INET6"):
        return socket.AF_INET6
    return socket.AF_INET 
开发者ID:Frank-qlu,项目名称:recruit,代码行数:22,代码来源:serving.py


示例6: test_basic_behavior

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def test_basic_behavior(self):
    inet4_server = self.mox.CreateMock(wsgi_server._SingleAddressWsgiServer)
    inet6_server = self.mox.CreateMock(wsgi_server._SingleAddressWsgiServer)
    self.mox.StubOutWithMock(wsgi_server, '_SingleAddressWsgiServer')
    self.mox.StubOutWithMock(socket, 'getaddrinfo')
    socket.getaddrinfo('localhost', 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0,
                       socket.AI_PASSIVE).AndReturn(
                           [(None, None, None, None, ('127.0.0.1', 0, 'baz')),
                            (None, None, None, None, ('::1', 0, 'baz'))])
    wsgi_server._SingleAddressWsgiServer(('127.0.0.1', 0), None).AndReturn(
        inet4_server)
    inet4_server.start()
    inet4_server.port = 123
    wsgi_server._SingleAddressWsgiServer(('::1', 123), None).AndReturn(
        inet6_server)
    inet6_server.start()
    self.mox.ReplayAll()
    self.server.start()
    self.mox.VerifyAll()
    self.assertItemsEqual([inet4_server, inet6_server],
                          self.server._servers) 
开发者ID:elsigh,项目名称:browserscope,代码行数:23,代码来源:wsgi_server_test.py


示例7: test_retry_limited

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def test_retry_limited(self):
    inet4_servers = [self.mox.CreateMock(wsgi_server._SingleAddressWsgiServer)
                     for _ in range(wsgi_server._PORT_0_RETRIES)]
    inet6_servers = [self.mox.CreateMock(wsgi_server._SingleAddressWsgiServer)
                     for _ in range(wsgi_server._PORT_0_RETRIES)]
    self.mox.StubOutWithMock(wsgi_server, '_SingleAddressWsgiServer')
    self.mox.StubOutWithMock(socket, 'getaddrinfo')
    socket.getaddrinfo('localhost', 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0,
                       socket.AI_PASSIVE).AndReturn(
                           [(None, None, None, None, ('127.0.0.1', 0, 'baz')),
                            (None, None, None, None, ('::1', 0, 'baz'))])
    for offset, (inet4_server, inet6_server) in enumerate(zip(
        inet4_servers, inet6_servers)):
      wsgi_server._SingleAddressWsgiServer(('127.0.0.1', 0), None).AndReturn(
          inet4_server)
      inet4_server.start()
      inet4_server.port = offset + 1
      wsgi_server._SingleAddressWsgiServer(('::1', offset + 1), None).AndReturn(
          inet6_server)
      inet6_server.start().AndRaise(
          wsgi_server.BindError('message', (errno.EADDRINUSE, 'in use')))
      inet4_server.quit()
    self.mox.ReplayAll()
    self.assertRaises(wsgi_server.BindError, self.server.start)
    self.mox.VerifyAll() 
开发者ID:elsigh,项目名称:browserscope,代码行数:27,代码来源:wsgi_server_test.py


示例8: test_ignore_other_errors

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def test_ignore_other_errors(self):
    inet4_server = self.mox.CreateMock(wsgi_server._SingleAddressWsgiServer)
    inet6_server = self.mox.CreateMock(wsgi_server._SingleAddressWsgiServer)
    self.mox.StubOutWithMock(wsgi_server, '_SingleAddressWsgiServer')
    self.mox.StubOutWithMock(socket, 'getaddrinfo')
    socket.getaddrinfo('localhost', 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0,
                       socket.AI_PASSIVE).AndReturn(
                           [(None, None, None, None, ('127.0.0.1', 0, 'baz')),
                            (None, None, None, None, ('::1', 0, 'baz'))])
    wsgi_server._SingleAddressWsgiServer(('127.0.0.1', 0), None).AndReturn(
        inet4_server)
    inet4_server.start()
    inet4_server.port = 123
    wsgi_server._SingleAddressWsgiServer(('::1', 123), None).AndReturn(
        inet6_server)
    inet6_server.start().AndRaise(
        wsgi_server.BindError('message', (errno.ENOPROTOOPT, 'no protocol')))
    self.mox.ReplayAll()
    self.server.start()
    self.mox.VerifyAll()
    self.assertItemsEqual([inet4_server],
                          self.server._servers) 
开发者ID:elsigh,项目名称:browserscope,代码行数:24,代码来源:wsgi_server_test.py


示例9: select_ip_version

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def select_ip_version(host, port):
    """Returns AF_INET4 or AF_INET6 depending on where to connect to."""
    # disabled due to problems with current ipv6 implementations
    # and various operating systems.  Probably this code also is
    # not supposed to work, but I can't come up with any other
    # ways to implement this.
    # try:
    #     info = socket.getaddrinfo(host, port, socket.AF_UNSPEC,
    #                               socket.SOCK_STREAM, 0,
    #                               socket.AI_PASSIVE)
    #     if info:
    #         return info[0][0]
    # except socket.gaierror:
    #     pass
    if ':' in host and hasattr(socket, 'AF_INET6'):
        return socket.AF_INET6
    return socket.AF_INET 
开发者ID:jpush,项目名称:jbox,代码行数:19,代码来源:serving.py


示例10: is_valid_ip

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def is_valid_ip(ip):
    """Returns true if the given string is a well-formed IP address.
    Supports IPv4 and IPv6.
    //取自 tornado
    """
    if not ip or '\x00' in ip:
        # getaddrinfo resolves empty strings to localhost, and truncates
        # on zero bytes.
        return False
    try:
        res = socket.getaddrinfo(ip, 0, socket.AF_UNSPEC,
                                 socket.SOCK_STREAM,
                                 0, socket.AI_NUMERICHOST)
        return bool(res)
    except socket.gaierror as e:
        if e.args[0] == socket.EAI_NONAME:
            return False
        raise
    return True 
开发者ID:alexsunday,项目名称:pyvpn,代码行数:22,代码来源:util.py


示例11: listen

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def listen(self):
    res0 = self._resolveAddr()
    socket_family = self._socket_family == socket.AF_UNSPEC and socket.AF_INET6 or self._socket_family
    for res in res0:
      if res[0] is socket_family or res is res0[-1]:
        break
    # We need remove the old unix socket if the file exists and
    # nobody is listening on it.
    if self._unix_socket:
      tmp = socket.socket(res[0], res[1])
      try:
        tmp.connect(res[4])
      except socket.error, err:
        eno, message = err.args
        if eno == errno.ECONNREFUSED:
          os.unlink(res[4]) 
开发者ID:XiaoMi,项目名称:galaxy-sdk-python,代码行数:19,代码来源:TSocket.py


示例12: is_valid_ip

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def is_valid_ip(ip: str) -> bool:
    """Returns ``True`` if the given string is a well-formed IP address.
    Supports IPv4 and IPv6.
    """
    if not ip or "\x00" in ip:
        # getaddrinfo resolves empty strings to localhost, and truncates
        # on zero bytes.
        return False
    try:
        res = socket.getaddrinfo(
            ip, 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, socket.AI_NUMERICHOST
        )
        return bool(res)
    except socket.gaierror as e:
        if e.args[0] == socket.EAI_NONAME:
            return False
        raise
    return True 
开发者ID:opendevops-cn,项目名称:opendevops,代码行数:21,代码来源:netutil.py


示例13: resolve

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def resolve(
        self, host: str, port: int, family: socket.AddressFamily = socket.AF_UNSPEC
    ) -> Awaitable[List[Tuple[int, Any]]]:
        """Resolves an address.
        The ``host`` argument is a string which may be a hostname or a
        literal IP address.
        Returns a `.Future` whose result is a list of (family,
        address) pairs, where address is a tuple suitable to pass to
        `socket.connect <socket.socket.connect>` (i.e. a ``(host,
        port)`` pair for IPv4; additional fields may be present for
        IPv6). If a ``callback`` is passed, it will be run with the
        result as an argument when it is complete.
        :raises IOError: if the address cannot be resolved.
        .. versionchanged:: 4.4
           Standardized all implementations to raise `IOError`.
        .. versionchanged:: 6.0 The ``callback`` argument was removed.
           Use the returned awaitable object instead.
        """
        raise NotImplementedError() 
开发者ID:opendevops-cn,项目名称:opendevops,代码行数:27,代码来源:netutil.py


示例14: allowed_gai_family

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def allowed_gai_family():
    """This function is designed to work in the context of
    getaddrinfo, where family=socket.AF_UNSPEC is the default and
    will perform a DNS search for both IPv6 and IPv4 records."""
    family = socket.AF_INET
    if HAS_IPV6:
        family = socket.AF_UNSPEC
    return family 
开发者ID:danielecook,项目名称:gist-alfred,代码行数:11,代码来源:connection.py


示例15: find_free_address

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def find_free_address():
    """Bind to None, 0 to find an unused port on localhost (IPv4 or IPv6)
    :return:
    """
    err = None
    for info in socket.getaddrinfo(None, 0, socket.AF_UNSPEC,
                                   socket.SOCK_STREAM):
        family, stype, proto, _, addr = info
        sock = None
        try:
            sock = socket.socket(family, stype, proto)
            sock.bind(addr)
            if family == socket.AF_INET:
                return "{}:{}".format(*sock.getsockname())
            elif family == socket.AF_INET6:
                return "[{}]:{}".format(*sock.getsockname()[:2])
        except socket.error as e:
            err = e
        finally:
            if sock is not None:
                sock.close()
    if err is not None:
        raise err  # pylint: disable=raising-bad-type
    else:
        raise socket.error("getaddrinfo returns an empty list") 
开发者ID:latchset,项目名称:custodia,代码行数:28,代码来源:test_cli.py


示例16: resolve

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def resolve(self, host, port, family=socket.AF_UNSPEC, callback=None):
        """Resolves an address.
        The ``host`` argument is a string which may be a hostname or a
        literal IP address.
        Returns a `.Future` whose result is a list of (family,
        address) pairs, where address is a tuple suitable to pass to
        `socket.connect <socket.socket.connect>` (i.e. a ``(host,
        port)`` pair for IPv4; additional fields may be present for
        IPv6). If a ``callback`` is passed, it will be run with the
        result as an argument when it is complete.
        """
        raise NotImplementedError() 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:16,代码来源:netutil.py


示例17: resolve

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def resolve(self, host, port, family=0):
        if is_valid_ip(host):
            addresses = [host]
        else:
            # gethostbyname doesn't take callback as a kwarg
            self.channel.gethostbyname(host, family, (yield gen.Callback(1)))
            callback_args = yield gen.Wait(1)
            assert isinstance(callback_args, gen.Arguments)
            assert not callback_args.kwargs
            result, error = callback_args.args
            if error:
                raise Exception('C-Ares returned error %s: %s while resolving %s' %
                                (error, pycares.errno.strerror(error), host))
            addresses = result.addresses
        addrinfo = []
        for address in addresses:
            if '.' in address:
                address_family = socket.AF_INET
            elif ':' in address:
                address_family = socket.AF_INET6
            else:
                address_family = socket.AF_UNSPEC
            if family != socket.AF_UNSPEC and family != address_family:
                raise Exception('Requested socket family %d but got %d' %
                                (family, address_family))
            addrinfo.append((address_family, (address, port)))
        raise gen.Return(addrinfo) 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:29,代码来源:caresresolver.py


示例18: resolve

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def resolve(self, host, port, family=0):
        # getHostByName doesn't accept IP addresses, so if the input
        # looks like an IP address just return it immediately.
        if twisted.internet.abstract.isIPAddress(host):
            resolved = host
            resolved_family = socket.AF_INET
        elif twisted.internet.abstract.isIPv6Address(host):
            resolved = host
            resolved_family = socket.AF_INET6
        else:
            deferred = self.resolver.getHostByName(utf8(host))
            resolved = yield gen.Task(deferred.addBoth)
            if isinstance(resolved, failure.Failure):
                resolved.raiseException()
            elif twisted.internet.abstract.isIPAddress(resolved):
                resolved_family = socket.AF_INET
            elif twisted.internet.abstract.isIPv6Address(resolved):
                resolved_family = socket.AF_INET6
            else:
                resolved_family = socket.AF_UNSPEC
        if family != socket.AF_UNSPEC and family != resolved_family:
            raise Exception('Requested socket family %d but got %d' %
                            (family, resolved_family))
        result = [
            (resolved_family, (resolved, port)),
        ]
        raise gen.Return(result) 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:29,代码来源:twisted.py


示例19: test_future_interface

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def test_future_interface(self):
        addrinfo = yield self.resolver.resolve('localhost', 80,
                                               socket.AF_UNSPEC)
        self.assertIn((socket.AF_INET, ('127.0.0.1', 80)),
                      addrinfo)
# It is impossible to quickly and consistently generate an error in name
# resolution, so test this case separately, using mocks as needed. 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:11,代码来源:netutil_test.py


示例20: test_future_interface_bad_host

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def test_future_interface_bad_host(self):
        with self.assertRaises(Exception):
            yield self.resolver.resolve('an invalid domain', 80,
                                        socket.AF_UNSPEC) 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:6,代码来源:netutil_test.py


示例21: start_server

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def start_server(self, family):
        if family == socket.AF_UNSPEC and 'TRAVIS' in os.environ:
            self.skipTest("dual-stack servers often have port conflicts on travis")
        self.server = TestTCPServer(family)
        return self.server.port 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:7,代码来源:tcpclient_test.py


示例22: test_connect_unspec_ipv4

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def test_connect_unspec_ipv4(self):
        self.do_test_connect(socket.AF_UNSPEC, '127.0.0.1') 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:4,代码来源:tcpclient_test.py


示例23: test_connect_unspec_dual

# 需要导入模块: import socket [as 别名]
# 或者: from socket import AF_UNSPEC [as 别名]
def test_connect_unspec_dual(self):
        self.do_test_connect(socket.AF_UNSPEC, 'localhost') 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:4,代码来源:tcpclient_test.py



注:本文中的socket.AF_UNSPEC属性示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。