✏️ Editing: dnskey.py Read Only

############################################################################
# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
#
# 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, "rU")

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)

🔐 Sid Gifari File Manager Pro