mirror of
https://github.com/isc-projects/bind9.git
synced 2026-02-27 20:11:12 -05:00
0ceba5a75a
It is default when open is called and obsolete. Removed in python 3.11, where it triggers errors. Just use normal read-only text opening.
511 lines
16 KiB
Python
511 lines
16 KiB
Python
# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
|
|
#
|
|
# SPDX-License-Identifier: MPL-2.0
|
|
#
|
|
# This Source Code Form is subject to the terms of the Mozilla Public
|
|
# License, v. 2.0. If a copy of the MPL was not distributed with this
|
|
# file, you can obtain one at https://mozilla.org/MPL/2.0/.
|
|
#
|
|
# See the COPYRIGHT file distributed with this work for additional
|
|
# information regarding copyright ownership.
|
|
|
|
import os
|
|
import time
|
|
import calendar
|
|
from subprocess import Popen, PIPE
|
|
|
|
########################################################################
|
|
# Class dnskey
|
|
########################################################################
|
|
class TimePast(Exception):
|
|
def __init__(self, key, prop, value):
|
|
super(TimePast, self).__init__('%s time for key %s (%d) is already past'
|
|
% (prop, key, value))
|
|
|
|
class dnskey:
|
|
"""An individual DNSSEC key. Identified by path, name, algorithm, keyid.
|
|
Contains a dictionary of metadata events."""
|
|
|
|
_PROPS = ('Created', 'Publish', 'Activate', 'Inactive', 'Delete',
|
|
'Revoke', 'DSPublish', 'SyncPublish', 'SyncDelete')
|
|
_OPTS = (None, '-P', '-A', '-I', '-D', '-R', None, '-Psync', '-Dsync')
|
|
|
|
_ALGNAMES = (None, 'RSAMD5', 'DH', 'DSA', None, 'RSASHA1',
|
|
'NSEC3DSA', 'NSEC3RSASHA1', 'RSASHA256', None,
|
|
'RSASHA512', None, 'ECCGOST', 'ECDSAP256SHA256',
|
|
'ECDSAP384SHA384', 'ED25519', 'ED448')
|
|
|
|
def __init__(self, key, directory=None, keyttl=None):
|
|
# this makes it possible to use algname as a class or instance method
|
|
if isinstance(key, tuple) and len(key) == 3:
|
|
self._dir = directory or '.'
|
|
(name, alg, keyid) = key
|
|
self.fromtuple(name, alg, keyid, keyttl)
|
|
|
|
self._dir = directory or os.path.dirname(key) or '.'
|
|
key = os.path.basename(key)
|
|
(name, alg, keyid) = key.split('+')
|
|
name = name[1:-1]
|
|
alg = int(alg)
|
|
keyid = int(keyid.split('.')[0])
|
|
self.fromtuple(name, alg, keyid, keyttl)
|
|
|
|
def fromtuple(self, name, alg, keyid, keyttl):
|
|
if name.endswith('.'):
|
|
fullname = name
|
|
name = name.rstrip('.')
|
|
else:
|
|
fullname = name + '.'
|
|
|
|
keystr = "K%s+%03d+%05d" % (fullname, alg, keyid)
|
|
key_file = self._dir + (self._dir and os.sep or '') + keystr + ".key"
|
|
private_file = (self._dir + (self._dir and os.sep or '') +
|
|
keystr + ".private")
|
|
|
|
self.keystr = keystr
|
|
|
|
self.name = name
|
|
self.alg = int(alg)
|
|
self.keyid = int(keyid)
|
|
self.fullname = fullname
|
|
|
|
kfp = open(key_file, "r")
|
|
for line in kfp:
|
|
if line[0] == ';':
|
|
continue
|
|
tokens = line.split()
|
|
if not tokens:
|
|
continue
|
|
|
|
if tokens[1].lower() in ('in', 'ch', 'hs'):
|
|
septoken = 3
|
|
self.ttl = keyttl
|
|
else:
|
|
septoken = 4
|
|
self.ttl = int(tokens[1]) if not keyttl else keyttl
|
|
|
|
if (int(tokens[septoken]) & 0x1) == 1:
|
|
self.sep = True
|
|
else:
|
|
self.sep = False
|
|
kfp.close()
|
|
|
|
pfp = open(private_file, "r")
|
|
|
|
self.metadata = dict()
|
|
self._changed = dict()
|
|
self._delete = dict()
|
|
self._times = dict()
|
|
self._fmttime = dict()
|
|
self._timestamps = dict()
|
|
self._original = dict()
|
|
self._origttl = None
|
|
|
|
for line in pfp:
|
|
line = line.strip()
|
|
if not line or line[0] in ('!#'):
|
|
continue
|
|
punctuation = [line.find(c) for c in ':= '] + [len(line)]
|
|
found = min([pos for pos in punctuation if pos != -1])
|
|
name = line[:found].rstrip()
|
|
value = line[found:].lstrip(":= ").rstrip()
|
|
self.metadata[name] = value
|
|
|
|
for prop in dnskey._PROPS:
|
|
self._changed[prop] = False
|
|
if prop in self.metadata:
|
|
t = self.parsetime(self.metadata[prop])
|
|
self._times[prop] = t
|
|
self._fmttime[prop] = self.formattime(t)
|
|
self._timestamps[prop] = self.epochfromtime(t)
|
|
self._original[prop] = self._timestamps[prop]
|
|
else:
|
|
self._times[prop] = None
|
|
self._fmttime[prop] = None
|
|
self._timestamps[prop] = None
|
|
self._original[prop] = None
|
|
|
|
pfp.close()
|
|
|
|
def commit(self, settime_bin, **kwargs):
|
|
quiet = kwargs.get('quiet', False)
|
|
cmd = []
|
|
first = True
|
|
|
|
if self._origttl is not None:
|
|
cmd += ["-L", str(self.ttl)]
|
|
|
|
for prop, opt in zip(dnskey._PROPS, dnskey._OPTS):
|
|
if not opt or not self._changed[prop]:
|
|
continue
|
|
|
|
delete = False
|
|
if prop in self._delete and self._delete[prop]:
|
|
delete = True
|
|
|
|
when = 'none' if delete else self._fmttime[prop]
|
|
cmd += [opt, when]
|
|
first = False
|
|
|
|
if cmd:
|
|
fullcmd = [settime_bin, "-K", self._dir] + cmd + [self.keystr,]
|
|
if not quiet:
|
|
print('# ' + ' '.join(fullcmd))
|
|
try:
|
|
p = Popen(fullcmd, stdout=PIPE, stderr=PIPE)
|
|
stdout, stderr = p.communicate()
|
|
if stderr:
|
|
raise Exception(str(stderr))
|
|
except Exception as e:
|
|
raise Exception('unable to run %s: %s' %
|
|
(settime_bin, str(e)))
|
|
self._origttl = None
|
|
for prop in dnskey._PROPS:
|
|
self._original[prop] = self._timestamps[prop]
|
|
self._changed[prop] = False
|
|
|
|
@classmethod
|
|
def generate(cls, keygen_bin, randomdev, keys_dir, name, alg, keysize, sep,
|
|
ttl, publish=None, activate=None, **kwargs):
|
|
quiet = kwargs.get('quiet', False)
|
|
|
|
keygen_cmd = [keygen_bin, "-q", "-K", keys_dir, "-L", str(ttl)]
|
|
|
|
if randomdev:
|
|
keygen_cmd += ["-r", randomdev]
|
|
|
|
if sep:
|
|
keygen_cmd.append("-fk")
|
|
|
|
if alg:
|
|
keygen_cmd += ["-a", alg]
|
|
|
|
if keysize:
|
|
keygen_cmd += ["-b", str(keysize)]
|
|
|
|
if publish:
|
|
t = dnskey.timefromepoch(publish)
|
|
keygen_cmd += ["-P", dnskey.formattime(t)]
|
|
|
|
if activate:
|
|
t = dnskey.timefromepoch(activate)
|
|
keygen_cmd += ["-A", dnskey.formattime(activate)]
|
|
|
|
keygen_cmd.append(name)
|
|
|
|
if not quiet:
|
|
print('# ' + ' '.join(keygen_cmd))
|
|
|
|
p = Popen(keygen_cmd, stdout=PIPE, stderr=PIPE)
|
|
stdout, stderr = p.communicate()
|
|
if stderr:
|
|
raise Exception('unable to generate key: ' + str(stderr))
|
|
|
|
try:
|
|
keystr = stdout.splitlines()[0].decode('ascii')
|
|
newkey = dnskey(keystr, keys_dir, ttl)
|
|
return newkey
|
|
except Exception as e:
|
|
raise Exception('unable to parse generated key: %s' % str(e))
|
|
|
|
def generate_successor(self, keygen_bin, randomdev, prepublish, **kwargs):
|
|
quiet = kwargs.get('quiet', False)
|
|
|
|
if not self.inactive():
|
|
raise Exception("predecessor key %s has no inactive date" % self)
|
|
|
|
keygen_cmd = [keygen_bin, "-q", "-K", self._dir, "-S", self.keystr]
|
|
|
|
if self.ttl:
|
|
keygen_cmd += ["-L", str(self.ttl)]
|
|
|
|
if randomdev:
|
|
keygen_cmd += ["-r", randomdev]
|
|
|
|
if prepublish:
|
|
keygen_cmd += ["-i", str(prepublish)]
|
|
|
|
if not quiet:
|
|
print('# ' + ' '.join(keygen_cmd))
|
|
|
|
p = Popen(keygen_cmd, stdout=PIPE, stderr=PIPE)
|
|
stdout, stderr = p.communicate()
|
|
if stderr:
|
|
raise Exception('unable to generate key: ' + stderr)
|
|
|
|
try:
|
|
keystr = stdout.splitlines()[0].decode('ascii')
|
|
newkey = dnskey(keystr, self._dir, self.ttl)
|
|
return newkey
|
|
except:
|
|
raise Exception('unable to generate successor for key %s' % self)
|
|
|
|
@staticmethod
|
|
def algstr(alg):
|
|
name = None
|
|
if alg in range(len(dnskey._ALGNAMES)):
|
|
name = dnskey._ALGNAMES[alg]
|
|
return name if name else ("%03d" % alg)
|
|
|
|
@staticmethod
|
|
def algnum(alg):
|
|
if not alg:
|
|
return None
|
|
alg = alg.upper()
|
|
try:
|
|
return dnskey._ALGNAMES.index(alg)
|
|
except ValueError:
|
|
return None
|
|
|
|
def algname(self, alg=None):
|
|
return self.algstr(alg or self.alg)
|
|
|
|
@staticmethod
|
|
def timefromepoch(secs):
|
|
return time.gmtime(secs)
|
|
|
|
@staticmethod
|
|
def parsetime(string):
|
|
return time.strptime(string, "%Y%m%d%H%M%S")
|
|
|
|
@staticmethod
|
|
def epochfromtime(t):
|
|
return calendar.timegm(t)
|
|
|
|
@staticmethod
|
|
def formattime(t):
|
|
return time.strftime("%Y%m%d%H%M%S", t)
|
|
|
|
def setmeta(self, prop, secs, now, **kwargs):
|
|
force = kwargs.get('force', False)
|
|
|
|
if self._timestamps[prop] == secs:
|
|
return
|
|
|
|
if self._original[prop] is not None and \
|
|
self._original[prop] < now and not force:
|
|
raise TimePast(self, prop, self._original[prop])
|
|
|
|
if secs is None:
|
|
self._changed[prop] = False \
|
|
if self._original[prop] is None else True
|
|
|
|
self._delete[prop] = True
|
|
self._timestamps[prop] = None
|
|
self._times[prop] = None
|
|
self._fmttime[prop] = None
|
|
return
|
|
|
|
t = self.timefromepoch(secs)
|
|
self._timestamps[prop] = secs
|
|
self._times[prop] = t
|
|
self._fmttime[prop] = self.formattime(t)
|
|
self._changed[prop] = False if \
|
|
self._original[prop] == self._timestamps[prop] else True
|
|
|
|
def gettime(self, prop):
|
|
return self._times[prop]
|
|
|
|
def getfmttime(self, prop):
|
|
return self._fmttime[prop]
|
|
|
|
def gettimestamp(self, prop):
|
|
return self._timestamps[prop]
|
|
|
|
def created(self):
|
|
return self._timestamps["Created"]
|
|
|
|
def syncpublish(self):
|
|
return self._timestamps["SyncPublish"]
|
|
|
|
def setsyncpublish(self, secs, now=time.time(), **kwargs):
|
|
self.setmeta("SyncPublish", secs, now, **kwargs)
|
|
|
|
def publish(self):
|
|
return self._timestamps["Publish"]
|
|
|
|
def setpublish(self, secs, now=time.time(), **kwargs):
|
|
self.setmeta("Publish", secs, now, **kwargs)
|
|
|
|
def activate(self):
|
|
return self._timestamps["Activate"]
|
|
|
|
def setactivate(self, secs, now=time.time(), **kwargs):
|
|
self.setmeta("Activate", secs, now, **kwargs)
|
|
|
|
def revoke(self):
|
|
return self._timestamps["Revoke"]
|
|
|
|
def setrevoke(self, secs, now=time.time(), **kwargs):
|
|
self.setmeta("Revoke", secs, now, **kwargs)
|
|
|
|
def inactive(self):
|
|
return self._timestamps["Inactive"]
|
|
|
|
def setinactive(self, secs, now=time.time(), **kwargs):
|
|
self.setmeta("Inactive", secs, now, **kwargs)
|
|
|
|
def delete(self):
|
|
return self._timestamps["Delete"]
|
|
|
|
def setdelete(self, secs, now=time.time(), **kwargs):
|
|
self.setmeta("Delete", secs, now, **kwargs)
|
|
|
|
def syncdelete(self):
|
|
return self._timestamps["SyncDelete"]
|
|
|
|
def setsyncdelete(self, secs, now=time.time(), **kwargs):
|
|
self.setmeta("SyncDelete", secs, now, **kwargs)
|
|
|
|
def setttl(self, ttl):
|
|
if ttl is None or self.ttl == ttl:
|
|
return
|
|
elif self._origttl is None:
|
|
self._origttl = self.ttl
|
|
self.ttl = ttl
|
|
elif self._origttl == ttl:
|
|
self._origttl = None
|
|
self.ttl = ttl
|
|
else:
|
|
self.ttl = ttl
|
|
|
|
def keytype(self):
|
|
return ("KSK" if self.sep else "ZSK")
|
|
|
|
def __str__(self):
|
|
return ("%s/%s/%05d"
|
|
% (self.name, self.algname(), self.keyid))
|
|
|
|
def __repr__(self):
|
|
return ("%s/%s/%05d (%s)"
|
|
% (self.name, self.algname(), self.keyid,
|
|
("KSK" if self.sep else "ZSK")))
|
|
|
|
def date(self):
|
|
return (self.activate() or self.publish() or self.created())
|
|
|
|
# keys are sorted first by zone name, then by algorithm. within
|
|
# the same name/algorithm, they are sorted according to their
|
|
# 'date' value: the activation date if set, OR the publication
|
|
# if set, OR the creation date.
|
|
def __lt__(self, other):
|
|
if self.name != other.name:
|
|
return self.name < other.name
|
|
if self.alg != other.alg:
|
|
return self.alg < other.alg
|
|
return self.date() < other.date()
|
|
|
|
def check_prepub(self, output=None):
|
|
def noop(*args, **kwargs): pass
|
|
if not output:
|
|
output = noop
|
|
|
|
now = int(time.time())
|
|
a = self.activate()
|
|
p = self.publish()
|
|
|
|
if not a:
|
|
return False
|
|
|
|
if not p:
|
|
if a > now:
|
|
output("WARNING: Key %s is scheduled for\n"
|
|
"\t activation but not for publication."
|
|
% repr(self))
|
|
return False
|
|
|
|
if p <= now and a <= now:
|
|
return True
|
|
|
|
if p == a:
|
|
output("WARNING: %s is scheduled to be\n"
|
|
"\t published and activated at the same time. This\n"
|
|
"\t could result in a coverage gap if the zone was\n"
|
|
"\t previously signed. Activation should be at least\n"
|
|
"\t %s after publication."
|
|
% (repr(self),
|
|
dnskey.duration(self.ttl) or 'one DNSKEY TTL'))
|
|
return True
|
|
|
|
if a < p:
|
|
output("WARNING: Key %s is active before it is published"
|
|
% repr(self))
|
|
return False
|
|
|
|
if self.ttl is not None and a - p < self.ttl:
|
|
output("WARNING: Key %s is activated too soon\n"
|
|
"\t after publication; this could result in coverage \n"
|
|
"\t gaps due to resolver caches containing old data.\n"
|
|
"\t Activation should be at least %s after\n"
|
|
"\t publication."
|
|
% (repr(self),
|
|
dnskey.duration(self.ttl) or 'one DNSKEY TTL'))
|
|
return False
|
|
|
|
return True
|
|
|
|
def check_postpub(self, output = None, timespan = None):
|
|
def noop(*args, **kwargs): pass
|
|
if output is None:
|
|
output = noop
|
|
|
|
if timespan is None:
|
|
timespan = self.ttl
|
|
|
|
if timespan is None:
|
|
output("WARNING: Key %s using default TTL." % repr(self))
|
|
timespan = (60*60*24)
|
|
|
|
now = time.time()
|
|
d = self.delete()
|
|
i = self.inactive()
|
|
|
|
if not d:
|
|
return False
|
|
|
|
if not i:
|
|
if d > now:
|
|
output("WARNING: Key %s is scheduled for\n"
|
|
"\t deletion but not for inactivation." % repr(self))
|
|
return False
|
|
|
|
if d < now and i < now:
|
|
return True
|
|
|
|
if d < i:
|
|
output("WARNING: Key %s is scheduled for\n"
|
|
"\t deletion before inactivation."
|
|
% repr(self))
|
|
return False
|
|
|
|
if d - i < timespan:
|
|
output("WARNING: Key %s scheduled for\n"
|
|
"\t deletion too soon after deactivation; this may \n"
|
|
"\t result in coverage gaps due to resolver caches\n"
|
|
"\t containing old data. Deletion should be at least\n"
|
|
"\t %s after inactivation."
|
|
% (repr(self), dnskey.duration(timespan)))
|
|
return False
|
|
|
|
return True
|
|
|
|
@staticmethod
|
|
def duration(secs):
|
|
if not secs:
|
|
return None
|
|
|
|
units = [("year", 60*60*24*365),
|
|
("month", 60*60*24*30),
|
|
("day", 60*60*24),
|
|
("hour", 60*60),
|
|
("minute", 60),
|
|
("second", 1)]
|
|
|
|
output = []
|
|
for unit in units:
|
|
v, secs = secs // unit[1], secs % unit[1]
|
|
if v > 0:
|
|
output.append("%d %s%s" % (v, unit[0], "s" if v > 1 else ""))
|
|
|
|
return ", ".join(output)
|
|
|