check_expire

@@ -128,33 +128,17 @@ try:
     database = ianardap.IanaRDAPDatabase()
 except Exception as e:
     unknown("Exception when retrieving the IANA database: \"%s\"" % e)
-servers = database.find(domain)
+server = database.find(domain)
-if servers is None:
+if server is None:
     unknown("No RDAP server found for %s" % domain)
-# find may return several RDAP servers
+if server.endswith("/"):
-found = False
+    server = server[:-1] # Donuts RDAP server balks when there are two slashes and reply 404
-unknowns= ""
+try:
-errors = ""
+    response = requests.get("%s/domain/%s" % (server, domain), timeout=timeout)
-for server in servers:
+except requests.exceptions.Timeout:
-    if server.endswith("/"):
+    unknown("Timeout when trying to reach %s" % server)    
-        server = server[:-1] # Donuts RDAP server balks when there are two slashes and reply 404
+if response.status_code != 200:
-        try:
+    error("Invalid RDAP return code: %s" % response.status_code)
-            response = requests.get("%s/domain/%s" % (server, domain), timeout=timeout)
-            if response.status_code != 200:
-                errors += "Invalid RDAP return code at %s: %s " % \
-                    (server, response.status_code)
-            else:
-                found = True
-                break
-        except requests.exceptions.Timeout:
-            unknowns += "Timeout when trying to reach %s " % server
-if not found:
-    if errors != "":
-        error(errors)
-    elif unknowns != "":
-        unknown(unknowns)
-    else:
-        unknown("No working server found and no error messages")
 rdap = json.loads(response.content)
 for event in rdap["events"]:
     if event["eventAction"] == "expiration":

ianardap.py

@@ -1,7 +1,8 @@
 #!/usr/bin/env python3
 
-"""A simple module to get the RDAP server for a given domain name, IP
+"""A simple module to get the RDAP server for a given domain name,
-prefix or object, from the IANA databases specified in RFC 9224/8521.
+from the IANA database specified in RFC 9224.
+
 """
 
 # http://python-requests.org/ for easier HTTPS retrieval
@@ -16,12 +17,8 @@ import fcntl
 import pickle
 import pathlib
 
-IANABASES = {"domains": "https://data.iana.org/rdap/dns.json",
+IANABASE = "https://data.iana.org/rdap/dns.json"
-             "v4prefixes": "https://data.iana.org/rdap/ipv4.json",
+CACHE = os.environ["HOME"] + "/.ianardapcache" 
-             "v6prefixes": "https://data.iana.org/rdap/ipv6.json",
-             "as": "https://data.iana.org/rdap/asn.json",
-             "objects": "https://data.iana.org/rdap/object-tags.json"}
-CACHE = os.environ["HOME"] + "/.ianardapcaches" 
 MAXAGE = 24 # Hours. Used only if the server no longer gives the information.
 IANATIMEOUT = 10 # Seconds
 MAXTESTS = 3 # Maximum attempts to get the database
@@ -49,33 +46,19 @@ def parse_expires(h):
 
 class IanaRDAPDatabase():
 
-    def __init__(self, category="domains", maxage=None, cachedir=CACHE,
+    def __init__(self, maxage=MAXAGE, cachefile=CACHE, pickleformat=False):
-                 pickleformat=False):
         """Retrieves the IANA database, if not already cached. maxage is in
-hours. The cachedir is a directory (it will be created if not already
+hours. The cache file argument should not have an extension (it will
-existant). pickleformat is not the default because it is not really
+be added automatically). pickleformat is not the default because it is
-faster *and* it introduces security risks if someone can write in the
+not really faster *and* it introduces security risks if someone can
-file (see the documentation of the module).
+write in the file (see the documentation of the module)."""
-
-        """
-        
         cache_valid = False
-        if not os.path.exists(cachedir):
-            os.mkdir(cachedir)
-        self.category = category
-        cachefile = os.path.join(cachedir, category)
         if pickleformat:
             self.cachefile = cachefile + ".pickle"
         else:
             self.cachefile = cachefile + ".json"
         self.lockname = self.cachefile + ".lock"
         self.expirationfile = self.cachefile + ".expires"
-        if maxage is not None:
-            with open(self.expirationfile, 'w'):
-                self.expirationtime = time.mktime((datetime.datetime.now() + \
-                                                  datetime.timedelta(hours=maxage)).timetuple())
-                os.utime(self.expirationfile,
-                         times = (self.expirationtime, self.expirationtime))
         loaded = False
         tests = 0
         errmsg = "No error"
@@ -115,7 +98,7 @@ file (see the documentation of the module).
                         continue
             else:
                 self.unlock()
-                response = requests.get(IANABASES[category], timeout=IANATIMEOUT)
+                response = requests.get(IANABASE, timeout=IANATIMEOUT)
                 expirationtime = None
                 if "cache-control" in response.headers:
                     directives = parse_cachecontrol(response.headers["cache-control"])
@@ -152,27 +135,10 @@ file (see the documentation of the module).
         self.publication = database["publication"]
         self.version = database["version"]
         self.services = {}
-        if self.category == "domains":
+        for service in database["services"]:
-            for service in database["services"]:
+            for tld in service[0]:
-                for tld in service[0]:
+                for server in service[1]:
-                    if tld.lower() not in self.services:
+                    self.services[tld] = server
-                        self.services[tld.lower()] =[]
-                    for server in service[1]:
-                        # server is an URL so case-sensitive.
-                        self.services[tld.lower()].append(server) 
-        elif self.category == "objects":
-            for service in database["services"]:
-                maintainer = service[0]
-                for registry in service[1]:
-                    if registry.upper() not in self.services:
-                        self.services[registry.upper()] =[]
-                    for server in service[2]:
-                        self.services[registry.upper()]
-                    for server in service[2]:
-                        self.services[registry.upper()].append(server)
-        else: # IP addresses will be complicated, because of the
-              # longest prefix rule.
-            raise Exception("Unsupported category %s" % self.category)
         if not cache_valid:
             self.lock()
             cache = open(self.cachefile, "wb")
@@ -191,30 +157,16 @@ file (see the documentation of the module).
         fcntl.lockf(self.lockhandle, fcntl.LOCK_UN)
         self.lockhandle.close()
 
-    def find(self, id):
+    def find(self, domain):
-        """Get the RDAP server(s), as an array, for a given identifier. None
+        """ Get the RDAP server for a given domain name. None if there is none."""
-if there is none."""
+        if domain.endswith("."):
-        if self.category == "domains":
+            domain = domain[:-1]
-            domain = id
+        labels = domain.split(".")
-            if domain.endswith("."):
+        tld = labels[len(labels)-1]
-                domain = domain[:-1]
+        if tld in self.services:
-            labels = domain.lower().split(".")
+            return self.services[tld]
-            tld = labels[len(labels)-1]
-            if tld in self.services:
-                return self.services[tld]
-            else:
-                return None
-        elif self.category == "objects":
-            try:
-                (handle, registry) = id.rsplit("-", maxsplit=1) 
-            except ValueError:
-                raise Exception("Not a valid RFC 8521 identifier: \"%s\"" % id)
-            if registry.upper() in self.services:
-                return self.services[registry.upper()]
-            else:
-                return None
         else:
-            raise Exception("Unsupported category %s" % self.category)
+            return None
 
 if __name__ == "__main__":
     rdap = IanaRDAPDatabase(maxage=1)

test_ianardap.py

@@ -17,13 +17,13 @@ def test_basic():
         len(database.services) > 1000
     
 def test_alternative_cache():
-    tmpdir = tempfile.TemporaryDirectory(suffix=".testianacaches")
+    tmpfile = tempfile.NamedTemporaryFile(suffix=".testianacache", delete=False)
-    database = ianardap.IanaRDAPDatabase(cachedir=tmpdir.name, maxage=0)
+    database = ianardap.IanaRDAPDatabase(cachefile=tmpfile.name, maxage=0)
-    assert os.path.exists(database.cachefile) and \
+    assert os.path.exists(tmpfile.name) and \
-        datetime.datetime.fromtimestamp(os.path.getmtime(database.cachefile)) > \
+        datetime.datetime.fromtimestamp(os.path.getmtime(tmpfile.name)) > \
         (datetime.datetime.now() - datetime.timedelta(minutes=1))
-    os.remove(database.cachefile)
+    os.remove(tmpfile.name)
-    os.remove(database.cachefile + ".lock")
+    os.remove(tmpfile.name + ".json.lock")
 
 def test_refresh():
     # Force a resfresh
@@ -35,7 +35,7 @@ def test_refresh():
 def test_find_exists():
     database = ianardap.IanaRDAPDatabase()
     server = database.find("www.foobar.ar")
-    assert server == ["https://rdap.nic.ar/"]
+    assert server == "https://rdap.nic.ar/"
 
 def test_find_not_exists():
     database = ianardap.IanaRDAPDatabase()