Merge pull request #33 from riotkit-org/issue_15

#15: Fix support for .sk domains, tested on priamaakcia.sk

infracheck/checks/domain-expiration

@@ -2,25 +2,27 @@
 
 """
 <sphinx>
+
 domain-expiration
 -----------------
 
 Check if the domain is close to expiration date or if it is already expired.
 
 **Notice: Multiple usage of this check can cause a "request limit exceeded" error to happen**
 
-*Suggestion: If you check multiple domains, then separate domains checking from regular health checks and set REFRESH_TIME (docker) to once a day, and WAIT_TIME=300 for non-docker installations - in crontab set a check with --force once a day*
+**Warning:** *Due to limits per IP on whois usage we recommend to strongly cache the health check ex. 1-2 days cache,
+and in case of checking multiple domains to use feature called "wait time" to sleep between checks,
+to not send too many requests a once*
 
 Parameters:
 
 - domain (domain name)
 - alert_days_before (number of days before expiration date to start alerting)
+
 </sphinx>
 """
-
+from typing import Optional
 import whois
-from whois._3_adjust import str_to_date
-
 import datetime
 import pytz
 import os
@@ -29,16 +31,23 @@ import re
 import time
 import subprocess
 from collections import namedtuple
+from whois._3_adjust import str_to_date
 
 
-ManualCheck = namedtuple('ManualCheck', 'expiration_date')
+WhoisResult = namedtuple('WhoisResult', 'expiration_date')
 
 
-class DomainCheck:
+class DomainCheck(object):
     _domain: str
     _alert_days_before: int
     whois: whois
 
+    PATTERNS = [
+        r'Expiry Date:\s+([0-9-T:]+)Z',
+        r'Valid Until:\s+([0-9-T:]+)',  # Valid Until:                  2021-06-24
+        r'renewal date:\s+([0-9-T:]+)',
+    ]
+
     def __init__(self, domain: str, alert_days_before: int):
         self._domain = domain
         self._alert_days_before = int(alert_days_before)
@@ -47,43 +56,78 @@ class DomainCheck:
         if not self._domain or self._domain == '':
             raise Exception('Domain must be specified')
 
-    def _check_with_wait(self):
+    def _check_with_wait(self) -> WhoisResult:
         time.sleep(1)
         max_retries = 10
         retries = 0
 
-        while True:
+        while retries <= max_retries:
             try:
-                query = self.whois.query(self._domain)
+                query = self._ask_whois_library(self._domain)
 
                 if not query or not query.expiration_date:
                     return self._parse_shell_whois(self._domain)
 
                 return query
+
             except Exception as e:
                 retries += 1
 
                 if retries > max_retries:
                     raise e
 
+                if isinstance(e, whois.exceptions.UnknownTld):
+                    shell_check_result = self._parse_shell_whois(self._domain)
+
+                    if not shell_check_result or not shell_check_result.expiration_date:
+                        continue
+
+                    return shell_check_result
+
                 if "request limit exceeded" in str(e):
                     time.sleep(5)
 
-    def _parse_shell_whois(self, domain: str):
+    def _ask_whois_library(self, domain: str) -> WhoisResult:
+        _stdout_copy = sys.stdout
+
+        # the "whois" library is using print(), sys.stdout needs to be mocked for a moment to not print those
+        # messages
+        with open('/dev/null', 'w') as null_out:
+            try:
+                sys.stdout = null_out
+
+                return self.whois.query(domain)
+            finally:
+                sys.stdout = _stdout_copy
+
+    @staticmethod
+    def _ask_whois_in_shell(domain: str) -> str:
+        return subprocess.check_output(['whois', domain]).decode('utf-8')
+
+    def _parse_shell_whois(self, domain: str) -> WhoisResult:
+        """
+        Fallback to shell command in case, when a whois library does not support given domain TLD
+
+        :param domain:
+        :return:
+        """
+
         try:
-            output = subprocess.check_output('whois %s' % domain, shell=True)
-            match = re.search(r'Expiry Date: ([0-9-T:]+)Z', output.decode('utf-8'), re.IGNORECASE)
+            output = self._ask_whois_in_shell(domain)
+
+            for pattern in self.PATTERNS:
+                match = re.search(pattern, output, re.IGNORECASE)
 
-            if match:
-                return ManualCheck(expiration_date=str_to_date(match.group(1)))
+                if match:
+                    return WhoisResult(expiration_date=str_to_date(match.group(1)))
 
         except subprocess.CalledProcessError:
             pass
 
-        return ManualCheck(expiration_date=None)
+        return WhoisResult(expiration_date=None)
 
     def perform_check(self) -> tuple:
-        domain_check = self._check_with_wait()
+        domain_check: Optional[WhoisResult] = self._check_with_wait()
 
         if domain_check is None or domain_check.expiration_date is None:
             return False, "Domain seems to be not registered"

requirements.txt

@@ -5,7 +5,7 @@ python-dateutil >= 2.8.1, < 3
 pytz >= 2019.3
 six >= 1.15.0, < 2
 tornado >= 5.1.1, < 6
-whois >= 0.9.4, < 1
+whois >= 0.9.13, < 1
 influxdb >= 5.3.1, < 6
 msgpack >= 1.0, < 2
 rkd>=2.3.3, <3

tests/unit_test_domain_expiration_check.py

@@ -3,6 +3,7 @@
 import inspect
 import unittest.mock
 import datetime
+import whois
 
 path = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe()))) + '/../infracheck/checks/'
 DomainCheck: any
@@ -47,3 +48,61 @@ def test_expired(self):
 
         self.assertIn("Domain google.com expired", result[1])
         self.assertFalse(result[0])
+
+    def test_sk_domain_manual_whois_parsing(self):
+        check = DomainCheck('priamaakcia.sk', 25)
+
+        def mock_library_raises_exception(*args, **kwargs):
+            raise whois.exceptions.UnknownTld()
+
+        check._ask_whois_library = mock_library_raises_exception
+        check._ask_whois_in_shell = lambda *args, **kwargs: """
+            Domain:                       priamaakcia.sk
+            Registrant:                   WEBHOUSE-AZPNWEY
+            Admin Contact:                WEBHOUSE
+            Tech Contact:                 WEBHOUSE
+            Registrar:                    IPEK-0001
+            Created:                      2005-06-24
+            Updated:                      2020-07-08
+            Valid Until:                  2999-06-24
+            Nameserver:                   ns1.afraid.org
+            Nameserver:                   ns2.afraid.org
+            EPP Status:                   ok
+            
+            Registrar:                    IPEK-0001
+            Name:                         WebHouse, s.r.o.
+            Organization:                 WebHouse, s.r.o.
+            Organization ID:              36743852
+            Phone:                        +421.332933500
+            Email:                        domeny@webhouse.sk
+            Street:                       Paulínska 20
+            City:                         Trnava
+            Postal Code:                  91701
+            Country Code:                 SK
+            Created:                      2017-09-01
+            Updated:                      2021-05-17
+            
+            Contact:                      WEBHOUSE-AZPNWEY
+            Registrar:                    IPEK-0001
+            Created:                      2017-09-02
+            Updated:                      2017-11-01
+            
+            Contact:                      WEBHOUSE
+            Name:                         WebHouse, s. r. o.
+            Organization:                 WebHouse, s. r. o.
+            Organization ID:              36743852
+            Phone:                        +421.910969601
+            Email:                        domeny@webhouse.sk
+            Street:                       Paulínska 20
+            City:                         Trnava
+            Postal Code:                  917 01
+            Country Code:                 SK
+            Registrar:                    IPEK-0001
+            Created:                      2017-09-02
+            Updated:                      2021-03-22
+        """
+
+        result = check.perform_check()
+
+        self.assertIn('Domain priamaakcia.sk is not expired.', result[1])
+        self.assertTrue(result[0])