Add ignore list for DNSBL results

.github/workflows/python-package.yml

@@ -13,7 +13,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.6', '3.7', '3.8', '3.9', '3.10']
+        python-version: ['3.7', '3.8', '3.9', '3.10']
 
     steps:
     - uses: actions/checkout@v2

setup.py

@@ -28,7 +28,7 @@
     license = f.read()
 
 setup(name='python-slimta',
-      version='5.0.3',
+      version='5.0.4',
       author='Ian Good',
       author_email='ian@icgood.net',
       description='Lightweight, asynchronous SMTP libraries.',
@@ -38,7 +38,7 @@
       url='http://slimta.org/',
       include_package_data=True,
       packages=find_namespace_packages(include=['slimta.*']),
-      install_requires=['gevent >= 1.1rc',
+      install_requires=['gevent >= 1.1',
                         'pysasl >= 0.5.0',
                         'pycares >= 1'],
       extras_require={'spf': ['pyspf', 'py3dns'],
@@ -51,7 +51,6 @@
                    'Intended Audience :: Information Technology',
                    'License :: OSI Approved :: MIT License',
                    'Programming Language :: Python',
-                   'Programming Language :: Python :: 3.6',
                    'Programming Language :: Python :: 3.7',
                    'Programming Language :: Python :: 3.8',
                    'Programming Language :: Python :: 3.9',

slimta/util/dnsbl.py

@@ -28,6 +28,7 @@
 
 from __future__ import absolute_import
 
+from ipaddress import IPv4Address, IPv4Network
 from functools import wraps
 
 import gevent
@@ -54,11 +55,13 @@ class DnsBlocklist(object):
        :meth:`.get()` or ``__contains__``.
 
     :param address: The DNSBL domain name.
+    :param ignore: List of DNSBL result codes to ignore.
 
     """
 
-    def __init__(self, address):
+    def __init__(self, address, ignore=None):
         self.address = address
+        self.ignore = [IPv4Network(ip) for ip in (ignore or [])]
 
     def _build_query(self, ip):
         one, two, three, four = ip.split('.', 3)
@@ -86,12 +89,18 @@ def get(self, ip, timeout=None, strict=False):
         with gevent.Timeout(timeout, None):
             query = self._build_query(ip)
             try:
-                DNSResolver.query(query, 'A').get()
+                answers = DNSResolver.query(query, 'A').get()
             except DNSError as exc:
                 if exc.errno == ARES_ENOTFOUND:
                     return False
                 logging.log_exception(__name__, query=query)
             else:
+                if answers:
+                    for rdata in answers:
+                        ip = IPv4Address(rdata.host)
+                        for ignore in self.ignore:
+                            if ip in ignore:
+                                return False
                 return True
         return strict
 

test/test_slimta_util_dnsbl.py

@@ -10,8 +10,9 @@
 
 class FakeRdata(object):
 
-    def __init__(self, text):
-        self.text = text
+    def __init__(self, val):
+        self.host = val
+        self.text = val
 
 
 class FakeAsyncResult(object):
@@ -40,6 +41,14 @@ def test_dnsblocklist_get(self):
         self.assertTrue(self.dnsbl.get('1.2.3.4'))
         self.assertNotIn('5.6.7.8', self.dnsbl)
 
+    def test_dnsblocklist_get_ignore(self):
+        DNSResolver.query('4.3.2.1.test.example.com', 'A').AndReturn(FakeAsyncResult(['127.0.0.2']))
+        DNSResolver.query('8.7.6.5.test.example.com', 'A').AndReturn(FakeAsyncResult(['127.0.0.11']))
+        self.mox.ReplayAll()
+        dnsbl = DnsBlocklist('test.example.com', ['127.0.0.10/31'])
+        self.assertIn('1.2.3.4', dnsbl)
+        self.assertNotIn('5.6.7.8', dnsbl)
+
     def test_dnsblocklist_get_reason(self):
         DNSResolver.query('4.3.2.1.test.example.com', 'TXT').AndReturn(FakeAsyncResult())
         DNSResolver.query('4.3.2.1.test.example.com', 'TXT').AndReturn(FakeAsyncResult(['good reason']))