Compare commits
3 Commits
6b4f5b228e
...
ae00a81f79
Author | SHA1 | Date |
---|---|---|
Stephane Bortzmeyer | ae00a81f79 | |
Stephane Bortzmeyer | 09f5ccc3e4 | |
Stephane Bortzmeyer | 65d8deb569 |
|
@ -11,6 +11,7 @@ check_expire follows the usual Nagios rules. The options are:
|
|||
* -w: warning threshold in days
|
||||
* -v: verbose details in output
|
||||
* -u: unixtime output
|
||||
* -t: timeout for RDAP requetss (in seconds)
|
||||
|
||||
## Installation
|
||||
|
||||
|
@ -30,6 +31,7 @@ object CheckCommand "expiration" {
|
|||
"-H" = "$address$",
|
||||
"-c" = "$expiration_critical$",
|
||||
"-w" = "$expiration_warning$",
|
||||
"-t" = "$expiration_timeout$",
|
||||
"-v" = { set_if = "$expiration_verbose$" }
|
||||
}
|
||||
|
||||
|
|
|
@ -54,8 +54,8 @@ def usage(msg=None):
|
|||
|
||||
def details():
|
||||
if verbose:
|
||||
print(" RDAP database \"%s\", version %s published on %s, RDAP server is %s" % \
|
||||
(database.description, database.version, database.publication, server))
|
||||
print(" RDAP database \"%s\", version %s published on %s, retrieved on %s, RDAP server is %s" % \
|
||||
(database.description, database.version, database.publication, database.retrieved, server))
|
||||
else:
|
||||
print("")
|
||||
|
||||
|
|
|
@ -12,12 +12,14 @@ import datetime
|
|||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import fcntl
|
||||
|
||||
IANABASE = "https://data.iana.org/rdap/dns.json"
|
||||
CACHE = os.environ["HOME"] + "/.ianardapcache.json"
|
||||
MAXAGE = 24 # Hours
|
||||
IANATIMEOUT = 10
|
||||
IANATIMEOUT = 10 # Seconds
|
||||
MAXTESTS = 3 # Maximum attempts to get the database
|
||||
|
||||
class IanaRDAPDatabase():
|
||||
|
||||
|
@ -26,22 +28,49 @@ class IanaRDAPDatabase():
|
|||
cache_valid = False
|
||||
self.cachefile = cachefile
|
||||
self.lockname = self.cachefile + ".lock"
|
||||
loaded = False
|
||||
tests = 0
|
||||
errmsg = "No error"
|
||||
while not loaded and tests < MAXTESTS:
|
||||
self.lock()
|
||||
if os.path.exists(cachefile) and \
|
||||
datetime.datetime.fromtimestamp(os.path.getmtime(cachefile)) >= \
|
||||
(datetime.datetime.utcnow() - datetime.timedelta(hours = maxage)):
|
||||
datetime.datetime.fromtimestamp(os.path.getmtime(cachefile)) > \
|
||||
(datetime.datetime.now() - datetime.timedelta(hours = maxage)):
|
||||
cache = open(cachefile, "rb")
|
||||
content = cache.read()
|
||||
cache.close()
|
||||
cache_valid = True
|
||||
self.unlock()
|
||||
try:
|
||||
database = json.loads(content)
|
||||
loaded = True
|
||||
self.retrieved = datetime.datetime.fromtimestamp(os.path.getmtime(cachefile))
|
||||
cache_valid = True
|
||||
except json.decoder.JSONDecodeError:
|
||||
tests += 1
|
||||
errmsg = "Invalid JSON content in %s" % cachefile
|
||||
# Delete it without mercy
|
||||
os.remove(cachefile)
|
||||
continue
|
||||
else:
|
||||
self.unlock()
|
||||
response = requests.get(IANABASE, timeout=IANATIMEOUT)
|
||||
if response.status_code != 200:
|
||||
raise Exception("Invalid HTTPS return code when trying to get %s: %s" % (IANABASE, response.status_code))
|
||||
time.sleep(2)
|
||||
tests += 1
|
||||
errmsg = "Invalid HTTPS return code when trying to get %s: %s" % (IANABASE, response.status_code)
|
||||
continue
|
||||
else:
|
||||
loaded = True
|
||||
self.retrieved = datetime.datetime.now()
|
||||
try:
|
||||
content = response.content
|
||||
database = json.loads(content)
|
||||
except json.decoder.JSONDecodeError:
|
||||
tests += 1
|
||||
errmsg = "Invalid JSON retrieved from %s" % IANABASE
|
||||
continue
|
||||
if not loaded:
|
||||
raise Exception("Cannot read IANA database: %s" % errmsg)
|
||||
self.description = database["description"]
|
||||
self.publication = database["publication"]
|
||||
self.version = database["version"]
|
||||
|
@ -67,6 +96,8 @@ class IanaRDAPDatabase():
|
|||
|
||||
def find(self, domain):
|
||||
""" Get the RDAP server for a given domain name. None if there is none."""
|
||||
if domain.endswith("."):
|
||||
domain = domain[:-1]
|
||||
labels = domain.split(".")
|
||||
tld = labels[len(labels)-1]
|
||||
if tld in self.services:
|
||||
|
@ -76,8 +107,8 @@ class IanaRDAPDatabase():
|
|||
|
||||
if __name__ == "__main__":
|
||||
rdap = IanaRDAPDatabase(maxage=1)
|
||||
print("Database \"%s\", version %s published on %s, %i services" % \
|
||||
(rdap.description, rdap.version, rdap.publication, len(rdap.services)))
|
||||
print("Database \"%s\", version %s published on %s, retrieved on %s, %i services" % \
|
||||
(rdap.description, rdap.version, rdap.publication, rdap.retrieved, len(rdap.services)))
|
||||
for domain in sys.argv[1:]:
|
||||
print("%s -> %s" % (domain, rdap.find(domain)))
|
||||
|
||||
|
|
|
@ -0,0 +1,43 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
"""Test by executing 'pytest' https://docs.pytest.org/
|
||||
|
||||
Don't forget to set PYTHONPATH if you want to test the development
|
||||
tree and not the installed package. """
|
||||
|
||||
import ianardap
|
||||
|
||||
import tempfile
|
||||
import os.path
|
||||
import datetime
|
||||
|
||||
def test_basic():
|
||||
database = ianardap.IanaRDAPDatabase()
|
||||
assert database.description.startswith("RDAP bootstrap file") and database.version == "1.0" and \
|
||||
len(database.services) > 1000
|
||||
|
||||
def test_alternative_cache():
|
||||
tmpfile = tempfile.NamedTemporaryFile(suffix=".testianacache", delete=False)
|
||||
database = ianardap.IanaRDAPDatabase(cachefile=tmpfile.name, maxage=0)
|
||||
assert os.path.exists(tmpfile.name) and \
|
||||
datetime.datetime.fromtimestamp(os.path.getmtime(tmpfile.name)) > \
|
||||
(datetime.datetime.now() - datetime.timedelta(minutes=1))
|
||||
os.remove(tmpfile.name)
|
||||
os.remove(tmpfile.name + ".lock")
|
||||
|
||||
def test_refresh():
|
||||
# Force a resfresh
|
||||
database = ianardap.IanaRDAPDatabase(maxage=0)
|
||||
assert (database.retrieved > (datetime.datetime.now() - datetime.timedelta(minutes=1))) and \
|
||||
(datetime.datetime.fromtimestamp(os.path.getmtime(database.cachefile)) > \
|
||||
(datetime.datetime.now() - datetime.timedelta(minutes=1)))
|
||||
|
||||
def test_find_exists():
|
||||
database = ianardap.IanaRDAPDatabase()
|
||||
server = database.find("www.foobar.ar")
|
||||
assert server == "https://rdap.nic.ar/"
|
||||
|
||||
def test_find_not_exists():
|
||||
database = ianardap.IanaRDAPDatabase()
|
||||
server = database.find("www.foobar.example")
|
||||
assert server is None
|
|
@ -72,6 +72,15 @@ tests:
|
|||
stderr: ''
|
||||
partstdout: 'RDAP server is'
|
||||
|
||||
- exe: './check_expire'
|
||||
args:
|
||||
- '-u'
|
||||
- '-H'
|
||||
- 'bortzmeyer.org'
|
||||
retcode: 0
|
||||
stderr: ''
|
||||
# TODO how to test we get an integer?
|
||||
|
||||
- exe: './check_expire'
|
||||
args:
|
||||
- '-c'
|
||||
|
|
Loading…
Reference in New Issue