Python socket.gaierror方法代码示例

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

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

示例1: socket_connect

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def socket_connect(func):
    def wrapped(self, *args, **kwargs):
        if not self._connected:
            log.info("Connecting to DirectTcp socket")
            try:
                self._sock = socket.create_connection((self.server, self.port), timeout=self.timeout)
            except (OSError, socket.gaierror) as err:
                raise ValueError("Failed to connect to '%s:%s': %s" % (self.server, self.port, str(err)))
            self._sock.settimeout(None)  # Make sure the socket is in blocking mode.
            self._t_recv = threading.Thread(target=self.recv_thread, name="recv-%s:%s" % (self.server, self.port))
            self._t_recv.daemon = True
            self._t_recv.start()
            self._connected = True
        func(self, *args, **kwargs)
    return wrapped 
开发者ID:jborean93,项目名称:smbprotocol,代码行数:20,代码来源:transport.py


示例2: _do_dns_lookup

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [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


示例3: is_valid_ip

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [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


示例4: test_ipv6

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def test_ipv6(self):
        try:
            [sock] = bind_sockets(None, '::1', family=socket.AF_INET6)
            port = sock.getsockname()[1]
            self.http_server.add_socket(sock)
        except socket.gaierror as e:
            if e.args[0] == socket.EAI_ADDRFAMILY:
                # python supports ipv6, but it's not configured on the network
                # interface, so skip this test.
                return
            raise
        url = '%s://[::1]:%d/hello' % (self.get_protocol(), port)
        # ipv6 is currently enabled by default but can be disabled
        self.http_client.fetch(url, self.stop, allow_ipv6=False)
        response = self.wait()
        self.assertEqual(response.code, 599)
        self.http_client.fetch(url, self.stop)
        response = self.wait()
        self.assertEqual(response.body, b"Hello world!") 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:23,代码来源:simple_httpclient_test.py


示例5: get_host_ip

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def get_host_ip(hostIP=None):
    if hostIP is None or hostIP == 'auto':
        hostIP = 'ip'
    if hostIP == 'dns':
        hostIP = socket.getfqdn()
    elif hostIP == 'ip':
        from socket import gaierror
        try:
            hostIP = socket.gethostbyname(socket.getfqdn())
        except gaierror:
            logger.warn('gethostbyname(socket.getfqdn()) failed... trying on hostname()')
            hostIP = socket.gethostbyname(socket.gethostname())
        if hostIP.startswith("127."):
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            # doesn't have to be reachable
            s.connect(('10.255.255.255', 1))
            hostIP = s.getsockname()[0]
    return hostIP 
开发者ID:aws,项目名称:sagemaker-xgboost-container,代码行数:21,代码来源:tracker.py


示例6: tunnel_traffic

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def tunnel_traffic(self):
        "Tunnel traffic to remote host:port"
        logger.info("%03d " % self.reqNum + Fore.CYAN + '[D] SSL Pass-Thru: https://%s/' % self.path)
        server_conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            server_conn.connect((self.host, int(self.port)))
            self.wfile.write(("HTTP/1.1 200 Connection established\r\n" +
                              "Proxy-agent: %s\r\n" % self.version_string() +
                              "\r\n").encode('ascii'))
            read_write(self.connection, server_conn)
        except TimeoutError:
            self.wfile.write(b"HTTP/1.1 504 Gateway Timeout\r\n\r\n")
            logger.warning("%03d " % self.reqNum + Fore.YELLOW + 'Timed Out: https://%s:%s/' % (self.host, self.port))
        except socket.gaierror as e:
            self.wfile.write(b"HTTP/1.1 503 Service Unavailable\r\n\r\n")
            logger.warning("%03d " % self.reqNum + Fore.YELLOW + '%s: https://%s:%s/' % (e, self.host, self.port))
        finally:
            # We don't maintain a connection reuse pool, so close the connection anyway
            server_conn.close() 
开发者ID:wheever,项目名称:ProxHTTPSProxyMII,代码行数:21,代码来源:ProxyTool.py


示例7: select_address_family

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [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


示例8: select_ip_version

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [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


示例9: main

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def main():
    try:
        f = ftplib.FTP(HOST)
    except (socket.error, socket.gaierror) as e:
        print('ERROR: 无法连接 "{}"'.format(HOST))
        return
    print('*** 已连接到 "{}"'.format(HOST))
    try:
        f.login()
    except ftplib.error_perm:
        print('ERROR: 无法匿名登录')
        f.quit()
        return
    print('*** 已匿名身份登录')
    try:
        f.cwd(DIRN)
    except ftplib.error_perm:
        print('ERROR: 无法跳转到 "{}" 目录'.format(DIRN))
        f.quit()
        return
    print('*** 跳转到 "{}" 目录'.format(DIRN))
    try:
        f.retrbinary('RETR %s' % FILE, open(FILE, 'wb').write)
    except ftplib.error_perm:
        print('ERROR: 无法读取文件 "{}"'.format(FILE))
        os.unlink(FILE)
    else:
        print('*** 已下载 "{}" 到当前目录'.format(FILE))
    f.quit() 
开发者ID:wdxtub,项目名称:deep-learning-note,代码行数:34,代码来源:1_ftp_client.py


示例10: scan

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def scan(port):
		s = socket.socket()
		
		try:
			result = s.connect_ex((host,port))
		except socket.gaierror as e:
			if args.verbose == True:
				print "[" + t.red("!") + "]Critical. A GAIerror was raised with the following error message."
				print e
				sys.exit(0)
			else:
				print "[" + t.red("!") + "]Critical. An error was raised while attempting to connect."
				sys.exit(0)
			
		if args.verbose == True:
			print "\n[" + t.green("+") + "]working on port: " + str(port)
			time.sleep(0.250)
	
		if result == 0:
			counting_open.append(port)
			if args.verbose == True:
				print "\n[" + t.magenta("~") + "]" + str(port) + " -> open."
				time.sleep(0.250)
			s.close()
		else:
			counting_close.append(port)
			if args.verbose == True:
				print "\n[" + t.magenta("~") + "]" + str(port) + " -> closed."
				time.sleep(0.250)
			s.close() 
开发者ID:NullArray,项目名称:QuickScan,代码行数:32,代码来源:quickscan.py


示例11: _xmit_packet

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def _xmit_packet(self, retry=True, delay_xmit=None):
        if self.sequencenumber:
            self.sequencenumber += 1
        if delay_xmit is not None:
            # skip transmit, let retry timer do it's thing
            self.waiting_sessions[self] = {}
            self.waiting_sessions[self]['ipmisession'] = self
            self.waiting_sessions[self]['timeout'] = delay_xmit +  _monotonic_time()
            return
        if self.sockaddr:
            self.send_data(self.netpacket, self.sockaddr)
        else:
            self.allsockaddrs = []
            try:
                for res in socket.getaddrinfo(self.bmc, self.port, 0, socket.SOCK_DGRAM):
                    sockaddr = res[4]
                    if res[0] == socket.AF_INET:
                        # convert the sockaddr to AF_INET6
                        newhost = '::ffff:' + sockaddr[0]
                        sockaddr = (newhost, sockaddr[1], 0, 0)
                    self.allsockaddrs.append(sockaddr)
                    self.bmc_handlers[sockaddr] = self
                    self.send_data(self.netpacket, sockaddr)
            except socket.gaierror:
                raise exc.IpmiException("Unable to transmit to specified address")
        if retry:
            self.waiting_sessions[self] = {}
            self.waiting_sessions[self]['ipmisession'] = self
            self.waiting_sessions[self]['timeout'] = self.timeout + _monotonic_time() 
开发者ID:rhtyd,项目名称:ipmisim,代码行数:31,代码来源:fakesession.py


示例12: SyncClockToNtp

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def SyncClockToNtp(retries: int = 2, server: Text = 'time.google.com'):
  """Syncs the hardware clock to an NTP server."""
  logging.info('Reading time from NTP server %s.', server)
  attempts = 0
  client = ntplib.NTPClient()
  response = None
  while True:
    try:
      response = client.request(server, version=3)
    except (ntplib.NTPException, socket.gaierror) as e:
      logging.error('NTP client request error: %s', str(e))
    if response or attempts >= retries:
      break
    logging.info(
        'Unable to contact NTP server %s to sync machine clock.  This '
        'machine may not have an IP address yet; waiting %d seconds and '
        'trying again. Repeated failure may indicate network or driver '
        'problems.', server, RETRY_DELAY)
    time.sleep(RETRY_DELAY)
    attempts += 1
  if not response:
    raise NtpException('No response from NTP server.')
  local_time = time.localtime(response.ref_time)
  current_date = time.strftime('%m-%d-%Y', local_time)
  current_time = time.strftime('%H:%M:%S', local_time)
  logging.info('Current date/time is %s %s', current_date, current_time)
  date_set = r'%s\cmd.exe /c date %s' % (WINPE_SYSTEM32, current_date)
  result = subprocess.call(date_set, shell=True)
  logging.info('Setting date returned result %s', result)
  time_set = r'%s\cmd.exe /c time %s' % (WINPE_SYSTEM32, current_time)
  result = subprocess.call(time_set, shell=True)
  logging.info('Setting time returned result %s', result) 
开发者ID:google,项目名称:glazier,代码行数:39,代码来源:ntp.py


示例13: _is_ipv6_enabled

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def _is_ipv6_enabled():
    """Check whether IPv6 is enabled on this host."""
    if socket.has_ipv6:
        sock = None
        try:
            sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
            sock.bind(('::1', 0))
            return True
        except (socket.error, socket.gaierror):
            pass
        finally:
            if sock:
                sock.close()
    return False 
开发者ID:war-and-code,项目名称:jawfish,代码行数:16,代码来源:support.py


示例14: get_names

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def get_names(self):
        if FileHandler.names is None:
            try:
                FileHandler.names = tuple(
                    socket.gethostbyname_ex('localhost')[2] +
                    socket.gethostbyname_ex(socket.gethostname())[2])
            except socket.gaierror:
                FileHandler.names = (socket.gethostbyname('localhost'),)
        return FileHandler.names
    # not entirely sure what the rules are here 
开发者ID:Soft8Soft,项目名称:verge3d-blender-addon,代码行数:13,代码来源:request.py


示例15: _safe_gethostbyname

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def _safe_gethostbyname(host):
    try:
        return socket.gethostbyname(host)
    except socket.gaierror:
        return None 
开发者ID:Soft8Soft,项目名称:verge3d-blender-addon,代码行数:7,代码来源:request.py


示例16: thishost

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def thishost():
    """Return the IP addresses of the current host."""
    global _thishost
    if _thishost is None:
        try:
            _thishost = tuple(socket.gethostbyname_ex(socket.gethostname())[2])
        except socket.gaierror:
            _thishost = tuple(socket.gethostbyname_ex('localhost')[2])
    return _thishost 
开发者ID:Soft8Soft,项目名称:verge3d-blender-addon,代码行数:11,代码来源:request.py


示例17: resolve_hostname

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def resolve_hostname(addr):
    '''Try to resolve an IP to a host name and returns None
    on common failures.
    :param addr: IP address to resolve.
    :type addr: ``string``
    :returns: Host name if success else None.
    :rtype: ``string``
    :raises ValueError: If `addr` is not a valid address
    '''
    if ip_math.is_valid_ip(addr):
        try:
            name, _, _ = socket.gethostbyaddr(addr)
            return name
        except socket.gaierror:
            # [Errno 8] nodename nor servname provided, or not known
            pass
        except socket.herror:
            # [Errno 1] Unknown host
            pass
        except socket.timeout:
            # Timeout.
            pass
        return None
    else:
        raise ValueError('Invalid ip address.') 
开发者ID:remg427,项目名称:misp42splunk,代码行数:31,代码来源:net_utils.py


示例18: test_gaierror

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def test_gaierror(self):
        # Test that IOStream sets its exc_info on getaddrinfo error.
        # It's difficult to reliably trigger a getaddrinfo error;
        # some resolvers own't even return errors for malformed names,
        # so we mock it instead. If IOStream changes to call a Resolver
        # before sock.connect, the mock target will need to change too.
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        stream = IOStream(s, io_loop=self.io_loop)
        stream.set_close_callback(self.stop)
        with mock.patch('socket.socket.connect',
                        side_effect=socket.gaierror('boom')):
            with ExpectLog(gen_log, "Connect error"):
                stream.connect(('localhost', 80), callback=self.stop)
                self.wait()
                self.assertIsInstance(stream.error, socket.gaierror) 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:17,代码来源:iostream_test.py


示例19: _failing_getaddrinfo

# 需要导入模块: import socket [as 别名]
# 或者: from socket import gaierror [as 别名]
def _failing_getaddrinfo(*args):
    """Dummy implementation of getaddrinfo for use in mocks"""
    raise socket.gaierror("mock: lookup failed") 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:5,代码来源:netutil_test.py



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