Read and apply IANA's caching instructions. Closes #3

This commit is contained in:
Stephane Bortzmeyer 2022-10-08 16:59:11 +02:00
parent 04fb72e408
commit 7ddffa7fa5

View File

@ -1,7 +1,7 @@
#!/usr/bin/env python3
"""A simple module to get the RDAP server for a given domain name,
from the IANA database specified in RFC 7484.
from the IANA database specified in RFC 9224.
"""
@ -15,13 +15,35 @@ import sys
import time
import fcntl
import pickle
import pathlib
IANABASE = "https://data.iana.org/rdap/dns.json"
CACHE = os.environ["HOME"] + "/.ianardapcache"
MAXAGE = 24 # Hours
MAXAGE = 24 # Hours. Used only if the server no longer gives the information.
IANATIMEOUT = 10 # Seconds
MAXTESTS = 3 # Maximum attempts to get the database
# Don't touch
HTTP_DATE_FORMAT = "%a, %d %b %Y %H:%M:%S %Z"
# RFC 9111, section 5.2
def parse_cachecontrol(h):
result = {}
directives = h.split(",")
for directive in directives:
directive = directive.strip()
if "=" in directive:
(key, value) = directive.split("=")
else:
key = directive
value = None
result[key.lower()] = value
return result
def parse_expires(h):
d = datetime.datetime.strptime(h, HTTP_DATE_FORMAT)
return d
class IanaRDAPDatabase():
def __init__(self, maxage=MAXAGE, cachefile=CACHE, pickleformat=False):
@ -36,14 +58,16 @@ write in the file (see the documentation of the module)."""
else:
self.cachefile = cachefile + ".json"
self.lockname = self.cachefile + ".lock"
self.expirationfile = self.cachefile + ".expires"
loaded = False
tests = 0
errmsg = "No error"
while not loaded and tests < MAXTESTS:
self.lock()
if os.path.exists(self.cachefile) and \
datetime.datetime.fromtimestamp(os.path.getmtime(self.cachefile)) > \
(datetime.datetime.now() - datetime.timedelta(hours = maxage)):
(pathlib.Path(self.expirationfile).exists() and \
datetime.datetime.fromtimestamp(os.path.getmtime(self.expirationfile)) > \
datetime.datetime.now()):
cache = open(self.cachefile, "rb")
content = cache.read()
cache.close()
@ -75,6 +99,18 @@ write in the file (see the documentation of the module)."""
else:
self.unlock()
response = requests.get(IANABASE, timeout=IANATIMEOUT)
expirationtime = None
if "cache-control" in response.headers:
directives = parse_cachecontrol(response.headers["cache-control"])
if "max-age" in directives:
maxage = int(directives["max-age"])
expirationtime = datetime.datetime.now() + datetime.timedelta(seconds=maxage)
if not expirationtime:
if "expires" in response.headers:
expirationtime = parse_expires(response.headers["expires"])
else:
expirationtime = datetime.datetime.now() + datetime.timedelta(hours=MAXAGE)
self.expirationtime = time.mktime(expirationtime.timetuple())
if response.status_code != 200:
time.sleep(2)
tests += 1
@ -86,6 +122,9 @@ write in the file (see the documentation of the module)."""
try:
content = response.content
database = json.loads(content)
with open(self.expirationfile, 'w'):
os.utime(self.expirationfile,
times = (self.expirationtime, self.expirationtime))
except json.decoder.JSONDecodeError:
tests += 1
errmsg = "Invalid JSON retrieved from %s" % IANABASE