Specify nameservers directly

CHANGELOG

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- nameserver specification support for DNSBLIpChecker
+
+
 ## [1.1.0] - 2020-09-27
 ### Added
 - ipv6 support for DNSBLIpChecker

pydnsbl/checker.py

@@ -84,11 +84,13 @@ class BaseDNSBLChecker(abc.ABC):
             * providers(list) - list of providers (Provider instance or str)
             * timeout(int) - timeout of dns requests will be passed to resolver
             * tries(int) - retry times
+            * ns(list) - nameservers to use for lookup
     """
 
     def __init__(self, providers=BASE_PROVIDERS, timeout=5,
-                 tries=2, concurrency=200, loop=None):
+                 tries=2, concurrency=200, loop=None, ns=None):
         self.providers = []
+        self.ns = ns
         for provider in providers:
             if not isinstance(provider, Provider):
                 raise ValueError('providers should contain only Provider instances')
@@ -101,7 +103,7 @@ def __init__(self, providers=BASE_PROVIDERS, timeout=5,
                 asyncio.set_event_loop(self._loop)
         else:
             self._loop = loop
-        self._resolver = aiodns.DNSResolver(timeout=timeout, tries=tries, loop=self._loop)
+        self._resolver = aiodns.DNSResolver(timeout=timeout, tries=tries, loop=self._loop, nameservers=ns)
         self._semaphore = asyncio.Semaphore(concurrency)
 
 

pydnsbl/providers.py

@@ -2,6 +2,8 @@
 Place to define providers.
 Most part of _BASE_PROVIDERS was taken from https://github.com/vincecarney/dnsbl
 """
+import os
+
 ### DNSBL CATEGORIES ###
 # providers answers could be interpreted in one of the following categories
 DNSBL_CATEGORY_UNKNOWN = 'unknown'
@@ -12,11 +14,17 @@
 DNSBL_CATEGORY_CNC = 'cnc'
 DNSBL_CATEGORY_ABUSED = 'abused'
 DNSBL_CATEGORY_LEGIT = 'legit'
+DNSBL_CATEGORY_DYNAMIC = 'dynamic'
 
 class Provider(object):
 
-    def __init__(self, host):
+    def __init__(self, host, ns=None):
+        """
+        host: the hostname of provider
+        ns: the nameserver IP to use for direct lookup
+        """
         self.host = host
+        self.ns = ns
 
     def process_response(self, response):
         """
@@ -57,6 +65,8 @@ def process_response(self, response):
                 categories.add(DNSBL_CATEGORY_SPAM)
             elif result.host in ['127.0.0.4', '127.0.0.5', '127.0.0.6', '127.0.0.7']:
                 categories.add(DNSBL_CATEGORY_EXPLOITS)
+            elif result.host in ['127.0.0.10', '127.0.0.11']:
+                categories.add(DNSBL_CATEGORY_DYNAMIC)
             else:
                 categories.add(DNSBL_CATEGORY_UNKNOWN)
         return categories
@@ -138,8 +148,8 @@ class DblSpamhaus(Provider):
         '127.0.1.106': {DNSBL_CATEGORY_ABUSED,  DNSBL_CATEGORY_LEGIT, DNSBL_CATEGORY_CNC}
     }
 
-    def __init__(self, host='dbl.spamhaus.org'):
-        Provider.__init__(self, host=host)
+    def __init__(self, host='dbl.spamhaus.org', ns=None):
+        Provider.__init__(self, host=host, ns=ns)
 
     def process_response(self, response):
         categories = set()
@@ -156,5 +166,9 @@ def process_response(self, response):
     'rhsbl.sorbs.net '
 ]
 
-BASE_PROVIDERS = [Provider(host) for host in _BASE_PROVIDERS] + [ZenSpamhaus()]
-BASE_DOMAIN_PROVIDERS = [Provider(host) for host in _DOMAIN_PROVIDERS] + [DblSpamhaus()]
+BASE_PROVIDERS = [Provider(host) for host in _BASE_PROVIDERS]
+BASE_DOMAIN_PROVIDERS = [Provider(host) for host in _DOMAIN_PROVIDERS]
+
+if os.getenv('USE_SPAMHAUS', 'false') == 'true':
+    BASE_PROVIDERS.append(ZenSpamhaus())
+    BASE_DOMAIN_PROVIDERS.append(DblSpamhaus())

setup.py

@@ -18,7 +18,7 @@ def get_long_description():
 
 setup(
     name='pydnsbl',
-    version='1.1.2',
+    version='1.1.5',
     description='Async dnsbl lists checker based on asyncio/aiodns.',
     long_description=get_long_description(),
     long_description_content_type='text/markdown',