From 25bd19be294cc846c3f6e9f2b75ffb35b2f85783 Mon Sep 17 00:00:00 2001
From: Benjamin Dauvergne <bdauvergne@entrouvert.com>
Date: Thu, 21 Oct 2021 17:53:12 +0200
Subject: [PATCH 2/3] utils/evaluate: add a dnsbl() dict like (#58055)

To check an IPv4 address is inside a DNSxL, use the following
expression in your condition:

    remote_addr in dnsbl('dnsbl.example.com')
---
 src/authentic2/utils/evaluate.py | 62 +++++++++++++++++++++++++++++++-
 tests/test_utils_evaluate.py     | 27 ++++++++++++++
 2 files changed, 88 insertions(+), 1 deletion(-)

diff --git a/src/authentic2/utils/evaluate.py b/src/authentic2/utils/evaluate.py
index 065667aa..3f448da0 100644
--- a/src/authentic2/utils/evaluate.py
+++ b/src/authentic2/utils/evaluate.py
@@ -15,18 +15,24 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import ast
+import logging
+import re
 import sys
 
+import dns.exception
+import dns.resolver
 from django.core.exceptions import ValidationError
+from django.core.validators import validate_ipv4_address
 
 try:
     from functools import lru_cache
 except ImportError:
     from django.utils.lru_cache import lru_cache
 
-
 from django.utils.translation import ugettext as _
 
+logger = logging.getLogger(__name__)
+
 
 class HTTPHeaders:
     def __init__(self, request):
@@ -67,6 +73,60 @@ class ExpressionError(ValidationError):
         self.text = Unparse().visit(node)
 
 
+def is_valid_hostname(hostname):
+    if hostname[-1] == ".":
+        # strip exactly one dot from the right, if present
+        hostname = hostname[:-1]
+    if len(hostname) > 253:
+        return False
+
+    labels = hostname.split(".")
+
+    # the TLD must be not all-numeric
+    if re.match(r"[0-9]+$", labels[-1]):
+        return False
+
+    allowed = re.compile(r"(?!-)[a-z0-9-]{1,63}(?<!-)$", re.IGNORECASE)
+    return all(allowed.match(label) for label in labels)
+
+
+def check_dnsbl(dnsbl, remote_addr):
+    domain = '.'.join(reversed(remote_addr.split('.'))) + '.' + dnsbl
+    exception = None
+    try:
+        answers = dns.resolver.resolve(domain, 'A', lifetime=1)
+        result = any(answer.address for answer in answers)
+    except dns.resolver.NXDOMAIN as e:
+        exception = e
+        result = False
+    except dns.resolver.NoAnswer as e:
+        exception = e
+        result = False
+    except dns.exception.DNSException as e:
+        exception = e
+        logger.warning(f'utils: could not check dnsbl {dnsbl} for domain "%s": %s', domain, e)
+        result = False
+    logger.debug('utils: dnsbl lookup of "%s", result=%s exception=%s', domain, result, exception)
+    return result
+
+
+class DNSBL:
+    def __init__(self, domain):
+        if not is_valid_hostname(domain):
+            raise ValueError('%s is not a valid domain name' % domain)
+        self.domain = domain
+
+    def __contains__(self, remote_addr):
+        if not remote_addr or not isinstance(remote_addr, str):
+            return False
+        validate_ipv4_address(remote_addr)
+        return check_dnsbl(self.domain, remote_addr)
+
+
+def dnsbl(domain):
+    return DNSBL(domain)
+
+
 class BaseExpressionValidator(ast.NodeVisitor):
     authorized_nodes = []
     forbidden_nodes = []
diff --git a/tests/test_utils_evaluate.py b/tests/test_utils_evaluate.py
index a79068df..14728432 100644
--- a/tests/test_utils_evaluate.py
+++ b/tests/test_utils_evaluate.py
@@ -16,6 +16,7 @@
 
 
 import ast
+from unittest import mock
 
 import pytest
 
@@ -95,3 +96,29 @@ def test_http_headers(rf):
     headers = HTTPHeaders(request)
     assert evaluate_condition('"X-Entrouvert" in headers', ctx={'headers': headers}) is True
     assert evaluate_condition('headers["X-Entrouvert"]', ctx={'headers': headers}) == '1'
+
+
+def test_dnsbl_ok():
+    from authentic2.utils.evaluate import dnsbl
+
+    with mock.patch('dns.resolver.resolve', return_value=[mock.Mock(address='127.0.0.2')]):
+        assert (
+            evaluate_condition(
+                "remote_addr in dnsbl('example.com')", ctx={'dnsbl': dnsbl, 'remote_addr': '1.2.3.4'}
+            )
+            is True
+        )
+
+
+def test_dnsbl_nok():
+    from dns.resolver import NXDOMAIN
+
+    from authentic2.utils.evaluate import dnsbl
+
+    with mock.patch('dns.resolver.resolve', side_effect=NXDOMAIN):
+        assert (
+            evaluate_condition(
+                "remote_addr in dnsbl('example.com')", ctx={'dnsbl': dnsbl, 'remote_addr': '1.2.3.4'}
+            )
+            is False
+        )
-- 
2.33.0