How to use the pycares.QUERY_TYPE_A function in pycares

To help you get started, we’ve selected a few pycares examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github saghul / pycares / tests / tests.py View on Github external
def test_query_timeout(self):
        self.result, self.errorno = None, None
        def cb(result, errorno):
            self.result, self.errorno = result, errorno
        self.channel.servers = ['1.2.3.4']
        self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
        self.wait()
        self.assertEqual(self.result, None)
        self.assertEqual(self.errorno, pycares.errno.ARES_ETIMEOUT)
github saghul / pycares / tests / tests.py View on Github external
def test_search(self):
        self.result, self.errorno = None, None
        def cb(result, errorno):
            self.result, self.errorno = result, errorno
        self.channel = pycares.Channel(timeout=5.0, tries=1, domains=['google.com'])
        self.channel.search('cloud', pycares.QUERY_TYPE_A, cb)
        self.wait()
        self.assertNoError(self.errorno)
        for r in self.result:
            self.assertEqual(type(r), pycares.ares_query_a_result)
            self.assertNotEqual(r.host, None)
github saghul / pycares / tests / tests.py View on Github external
def test_custom_resolvconf(self):
        self.result, self.errorno = None, None
        def cb(result, errorno):
            self.result, self.errorno = result, errorno
        self.channel = pycares.Channel(tries=1, timeout=2.0, resolvconf_path=os.path.join(FIXTURES_PATH, 'badresolv.conf'))
        self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
        self.wait()
        self.assertEqual(self.result, None)
        self.assertEqual(self.errorno, pycares.errno.ARES_ETIMEOUT)
github saghul / pycares / tests / tests.py View on Github external
def test_query_a_rotate(self):
        self.result, self.errorno = None, None
        self.errorno_count, self.count = 0, 0
        def cb(result, errorno):
            self.result, self.errorno = result, errorno
            if errorno:
                self.errorno_count += 1
            self.count += 1
        self.channel = pycares.Channel(timeout=1.0, tries=1, rotate=True)
        self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
        self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
        self.channel.query('google.com', pycares.QUERY_TYPE_A, cb)
        self.wait()
        self.assertEqual(self.count, 3)
        self.assertEqual(self.errorno_count, 0)
github Q2h1Cg / dnsbrute / lib / dns.py View on Github external
thread = threading.Thread(target=_query_pan_dns, args=(parent_domain,))
        thread.start()
        thread.join()

    record = None
    for query_type in ("A", "CNAME"):
        try:
            with async_timeout.timeout(_Timeout):
                records = await _resolver.query(domain, query_type)
        except:
            pass
        else:
            if query_type == "CNAME":
                record = Record(domain, QUERY_TYPE_CNAME, records.ttl if records.ttl else -1, [records.cname])
            elif query_type == "A" and records:
                record = Record(domain, QUERY_TYPE_A, records[0].ttl, [record_.host for record_ in records])
            if record:
                break

    if record:
        return None if _is_pan_dns(record) else record
github jedisct1 / ipgrep / ipgrep.py View on Github external
def __init__(self, name, channel, res):
        def cb(results, err):
            if results is None:
                return
            for result in results:
                res.add((result.host, self.name))

        self.name = name
        channel.query(name, pycares.QUERY_TYPE_A, cb)
github saghul / aiodns / aiodns / __init__.py View on Github external
Optional,
    Set
)

from . import error


__version__ = '2.0.0'

__all__ = ('DNSResolver', 'error')


READ = 1
WRITE = 2

query_type_map = {'A'     : pycares.QUERY_TYPE_A,
                  'AAAA'  : pycares.QUERY_TYPE_AAAA,
                  'ANY'   : pycares.QUERY_TYPE_ANY,
                  'CNAME' : pycares.QUERY_TYPE_CNAME,
                  'MX'    : pycares.QUERY_TYPE_MX,
                  'NAPTR' : pycares.QUERY_TYPE_NAPTR,
                  'NS'    : pycares.QUERY_TYPE_NS,
                  'PTR'   : pycares.QUERY_TYPE_PTR,
                  'SOA'   : pycares.QUERY_TYPE_SOA,
                  'SRV'   : pycares.QUERY_TYPE_SRV,
                  'TXT'   : pycares.QUERY_TYPE_TXT
        }


class DNSResolver:
    def __init__(self, nameservers: Optional[List[str]] = None, loop: Optional[asyncio.AbstractEventLoop] = None,
                 **kwargs: Any) -> None:
github saghul / pycares / examples / cares-selectors.py View on Github external
def gethostbyname(self, name, cb):
        self._channel.gethostbyname(name, socket.AF_INET, cb)


if __name__ == '__main__':
    def query_cb(result, error):
        print(result)
        print(error)
    def gethostbyname_cb(result, error):
        print(result)
        print(error)
    resolver = DNSResolver()
    resolver.query('google.com', pycares.QUERY_TYPE_A, query_cb)
    resolver.query('google.com', pycares.QUERY_TYPE_AAAA, query_cb)
    resolver.query('facebook.com', pycares.QUERY_TYPE_A, query_cb)
    resolver.query('sip2sip.info', pycares.QUERY_TYPE_SOA, query_cb)
    resolver.gethostbyname('apple.com', gethostbyname_cb)
    resolver.wait_channel()
github saghul / pycares / examples / cares-select.py View on Github external
channel.process_fd(pycares.ARES_SOCKET_BAD, pycares.ARES_SOCKET_BAD)
            continue
        rlist, wlist, xlist = select.select(read_fds, write_fds, [], timeout)
        for fd in rlist:
            channel.process_fd(fd, pycares.ARES_SOCKET_BAD)
        for fd in wlist:
            channel.process_fd(pycares.ARES_SOCKET_BAD, fd)


if __name__ == '__main__':
    def cb(result, error):
        print(result)
        print(error)
    channel = pycares.Channel()
    channel.gethostbyname('google.com', socket.AF_INET, cb)
    channel.query('google.com', pycares.QUERY_TYPE_A, cb)
    channel.query('sip2sip.info', pycares.QUERY_TYPE_SOA, cb)
    wait_channel(channel)
    print('Done!')