Blob Blame Raw
#!/usr/bin/python
#
# Convert ntp configuration to chrony
#
# Copyright (C) 2018  Miroslav Lichvar <mlichvar@redhat.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.


import argparse
import ipaddress
import os
import os.path
import re
import subprocess
import sys

# python2 compatibility hacks
if sys.version_info[0] < 3:
    from io import open
    reload(sys)
    sys.setdefaultencoding("utf-8")

class NtpConfiguration(object):
    def __init__(self, root_dir, ntp_conf, step_tickers, verbose):
        self.root_dir = root_dir if root_dir != "/" else ""
        self.ntp_conf_path = ntp_conf
        self.step_tickers_path = step_tickers
        self.verbose = verbose

        self.enabled_services = set()
        self.step_tickers = []
        self.time_sources = []
        self.fudges = {}
        self.restrictions = {
                # Built-in defaults
                ipaddress.ip_network(u"0.0.0.0/0"): set(),
                ipaddress.ip_network(u"::/0"): set(),
        }
        self.keyfile = ""
        self.keys = []
        self.trusted_keys = []
        self.driftfile = ""
        self.statistics = []
        self.leapfile = ""
        self.tos_options = []
        self.ignored_directives = set()
        self.ignored_lines = []

        #self.detect_enabled_services()
        self.parse_step_tickers()
        self.parse_ntp_conf()

    def detect_enabled_services(self):
        for service in ["ntpdate", "ntpd", "ntp-wait"]:
            if os.path.islink("{}/etc/systemd/system/multi-user.target.wants/{}.service"
                    .format(self.root_dir, service)):
                self.enabled_services.add(service)
        if self.verbose > 0:
            print("Enabled services found in /etc/systemd/system: " +
                    " ".join(self.enabled_services))

    def parse_step_tickers(self):
        if not self.step_tickers_path:
            return

        path = self.root_dir + self.step_tickers_path
        if not os.path.isfile(path):
            if self.verbose > 0:
                print("Missing " + path)
            return

        with open(path, encoding="latin-1") as f:
            for line in f:
                line = line[:line.find('#')]

                words = line.split()

                if not words:
                    continue

                self.step_tickers.extend(words)

    def parse_ntp_conf(self, path=None):
        if path is None:
            path = self.root_dir + self.ntp_conf_path

        with open(path, encoding="latin-1") as f:
            if self.verbose > 0:
                print("Reading " + path)

            for line in f:
                line = line[:line.find('#')]

                words = line.split()

                if not words:
                    continue

                if not self.parse_directive(words):
                    self.ignored_lines.append(line)

    def parse_directive(self, words):
        name = words.pop(0)
        if name.startswith("logconfig"):
            name = "logconfig"

        if words:
            if name in ["server", "peer", "pool"]:
                return self.parse_source(name, words)
            elif name == "fudge":
                return self.parse_fudge(words)
            elif name == "restrict":
                return self.parse_restrict(words)
            elif name == "tos":
                return self.parse_tos(words)
            elif name == "includefile":
                return self.parse_includefile(words)
            elif name == "keys":
                return self.parse_keys(words)
            elif name == "trustedkey":
                return self.parse_trustedkey(words)
            elif name == "driftfile":
                self.driftfile = words[0]
            elif name == "statistics":
                self.statistics = words
            elif name == "leapfile":
                self.leapfile = words[0]
            else:
                self.ignored_directives.add(name)
                return False
        else:
            self.ignored_directives.add(name)
            return False

        return True

    def parse_source(self, type, words):
        ipv4_only = False
        ipv6_only = False
        source = {
                "type": type,
                "options": []
        }

        if words[0] == "-4":
            ipv4_only = True
            words.pop(0)
        elif words[0] == "-6":
            ipv6_only = True
            words.pop(0)

        if not words:
            return False

        source["address"] = words.pop(0)

        # Check if -4/-6 corresponds to the address and ignore hostnames
        if ipv4_only or ipv6_only:
            try:
                version = ipaddress.ip_address(source["address"]).version
                if (ipv4_only and version != 4) or (ipv6_only and version != 6):
                    return False
            except ValueError:
                return False

        if source["address"].startswith("127.127."):
            if not source["address"].startswith("127.127.1."):
                # Ignore non-LOCAL refclocks
                return False

        while words:
            if len(words) >= 2 and words[0] in ["minpoll", "maxpoll", "version", "key"]:
                source["options"].append((words[0], words[1]))
                words = words[2:]
            elif words[0] in ["burst", "iburst", "noselect", "prefer", "true", "xleave"]:
                source["options"].append((words[0],))
                words.pop(0)
            else:
                return False

        self.time_sources.append(source)
        return True

    def parse_fudge(self, words):
        address = words.pop(0)
        options = {}

        while words:
            if len(words) >= 2:
                options[words[0]] = words[1]
                words = words[2:]
            else:
                return False

        self.fudges[address] = options
        return True

    def parse_restrict(self, words):
        ipv4_only = False
        ipv6_only = False
        flags = set()
        mask = ""

        if words[0] == "-4":
            ipv4_only = True
            words.pop(0)
        elif words[0] == "-6":
            ipv6_only = True
            words.pop(0)

        if not words:
            return False

        address = words.pop(0)

        while words:
            if len(words) >= 2 and words[0] == "mask":
                mask = words[1]
                words = words[2:]
            else:
                if words[0] not in ["kod", "nomodify", "notrap", "nopeer", "noquery",
                                    "limited", "ignore", "noserve"]:
                    return False
                flags.add(words[0])
                words.pop(0)

        # Convert to IP network(s), ignoring restrictions with hostnames
        networks = []
        if address == "default" and not mask:
            if not ipv6_only:
                networks.append(ipaddress.ip_network(u"0.0.0.0/0"))
            if not ipv4_only:
                networks.append(ipaddress.ip_network(u"::/0"))
        else:
            try:
                if mask:
                    networks.append(ipaddress.ip_network(u"{}/{}".format(address, mask)))
                else:
                    networks.append(ipaddress.ip_network(address))
            except ValueError:
                return False

            if (ipv4_only and networks[-1].version != 4) or \
                    (ipv6_only and networks[-1].version != 6):
                return False

        for network in networks:
            self.restrictions[network] = flags

        return True

    def parse_tos(self, words):
        options = []
        while words:
            if len(words) >= 2 and words[0] in ["minsane", "maxdist", "orphan"]:
                options.append((words[0], words[1]))
                words = words[2:]
            else:
                return False

        self.tos_options.extend(options)

        return True

    def parse_includefile(self, words):
        path = self.root_dir + words[0]
        if not os.path.isfile(path):
            return False

        self.parse_ntp_conf(path)
        return True

    def parse_keys(self, words):
        keyfile = words[0]
        path = self.root_dir + keyfile
        if not os.path.isfile(path):
            if self.verbose > 0:
                print("Missing file " + path)
            return False

        with open(path, encoding="latin-1") as f:
            if self.verbose > 0:
                print("Reading " + path)
            keys = []
            for line in f:
                words = line.split()
                if len(words) < 3 or not words[0].isdigit():
                    continue
                keys.append((int(words[0]), words[1], words[2]))

            self.keyfile = keyfile
            self.keys = keys

        return True

    def parse_trustedkey(self, words):
        key_ranges = []
        for word in words:
            if word.isdigit():
                key_ranges.append((int(word), int(word)))
            elif re.match("^[0-9]+-[0-9]+$", word):
                first, last = word.split("-")
                key_ranges.append((int(first), int(last)))
            else:
                return False

        self.trusted_keys = key_ranges
        return True

    def write_chrony_configuration(self, chrony_conf_path, chrony_keys_path,
                                   dry_run=False, backup=False):
        chrony_conf = self.get_chrony_conf(chrony_keys_path)
        if self.verbose > 1:
            print("Generated {}:\n{}".format(chrony_conf_path, chrony_conf))

        if not dry_run:
            self.write_file(chrony_conf_path, 0o644, chrony_conf, backup)

        chrony_keys = self.get_chrony_keys()
        if chrony_keys:
            if self.verbose > 1:
                print("Generated {}:\n{}".format(chrony_keys_path, chrony_keys))

        if not dry_run:
            self.write_file(chrony_keys_path, 0o640, chrony_keys, backup)

    def get_chrony_conf_sources(self):
        conf = ""

        if self.step_tickers:
            conf += "# Specify NTP servers used for initial correction.\n"
            conf += "initstepslew 0.1 {}\n".format(" ".join(self.step_tickers))
            conf += "\n"

        conf += "# Specify time sources.\n"

        for source in self.time_sources:
            address = source["address"]
            if address.startswith("127.127."):
                if address.startswith("127.127.1."):
                    continue
                assert False
            else:
                conf += "{} {}".format(source["type"], address)
                for option in source["options"]:
                    if option[0] in ["minpoll", "maxpoll", "version", "key",
                                     "iburst", "noselect", "prefer", "xleave"]:
                        conf += " {}".format(" ".join(option))
                    elif option[0] == "burst":
                        conf += " presend 6"
                    elif option[0] == "true":
                        conf += " trust"
                    else:
                        assert False
                conf += "\n"
        conf += "\n"

        return conf

    def get_chrony_conf_allows(self):
        allowed_networks = filter(lambda n: "ignore" not in self.restrictions[n] and
                                    "noserve" not in self.restrictions[n],
                                  self.restrictions.keys())

        conf = ""
        for network in sorted(allowed_networks, key=lambda n: (n.version, n)):
            if network.num_addresses > 1:
                conf += "allow {}\n".format(network)
            else:
                conf += "allow {}\n".format(network.network_address)

        if conf:
            conf = "# Allow NTP client access.\n" + conf
            conf += "\n"

        return conf

    def get_chrony_conf_cmdallows(self):
        allowed_networks = filter(lambda n: "ignore" not in self.restrictions[n] and
                                    "noquery" not in self.restrictions[n] and
                                    n != ipaddress.ip_network(u"127.0.0.1/32") and
                                    n != ipaddress.ip_network(u"::1/128"),
                                  self.restrictions.keys())

        ip_versions = set()
        conf = ""
        for network in sorted(allowed_networks, key=lambda n: (n.version, n)):
            ip_versions.add(network.version)
            if network.num_addresses > 1:
                conf += "cmdallow {}\n".format(network)
            else:
                conf += "cmdallow {}\n".format(network.network_address)

        if conf:
            conf = "# Allow remote monitoring.\n" + conf
            if 4 in ip_versions:
                conf += "bindcmdaddress 0.0.0.0\n"
            if 6 in ip_versions:
                conf += "bindcmdaddress ::\n"
            conf += "\n"

        return conf

    def get_chrony_conf(self, chrony_keys_path):
        local_stratum = 0
        maxdistance = 0.0
        minsources = 1
        orphan_stratum = 0
        logs = []

        for source in self.time_sources:
            address = source["address"]
            if address.startswith("127.127.1."):
                if address in self.fudges and "stratum" in self.fudges[address]:
                    local_stratum = int(self.fudges[address]["stratum"])
                else:
                    local_stratum = 5

        for tos in self.tos_options:
            if tos[0] == "maxdist":
                maxdistance = float(tos[1])
            elif tos[0] == "minsane":
                minsources = int(tos[1])
            elif tos[0] == "orphan":
                orphan_stratum = int(tos[1])
            else:
                assert False

        if "clockstats" in self.statistics:
            logs.append("refclocks");
        if "loopstats" in self.statistics:
            logs.append("tracking")
        if "peerstats" in self.statistics:
            logs.append("statistics");
        if "rawstats" in self.statistics:
            logs.append("measurements")

        conf = "# This file was converted from {}{}.\n".format(
                    self.ntp_conf_path,
                    " and " + self.step_tickers_path if self.step_tickers_path else "")
        conf += "\n"

        if self.ignored_lines:
            conf += "# The following directives were ignored in the conversion:\n"

            for line in self.ignored_lines:
                # Remove sensitive information
                line = re.sub(r"\s+pw\s+\S+", " pw XXX", line.rstrip())
                conf += "# " + line + "\n"
            conf += "\n"

        conf += self.get_chrony_conf_sources()

        conf += "# Record the rate at which the system clock gains/losses time.\n"
        if not self.driftfile:
            conf += "#"
        conf += "driftfile /var/lib/chrony/drift\n"
        conf += "\n"

        conf += "# Allow the system clock to be stepped in the first three updates\n"
        conf += "# if its offset is larger than 1 second.\n"
        conf += "makestep 1.0 3\n"
        conf += "\n"

        conf += "# Enable kernel synchronization of the real-time clock (RTC).\n"
        conf += "rtcsync\n"
        conf += "\n"

        conf += "# Enable hardware timestamping on all interfaces that support it.\n"
        conf += "#hwtimestamp *\n"
        conf += "\n"

        if maxdistance > 0.0:
            conf += "# Specify the maximum distance of sources to be selectable.\n"
            conf += "maxdistance {}\n".format(maxdistance)
            conf += "\n"

        conf += "# Increase the minimum number of selectable sources required to adjust\n"
        conf += "# the system clock.\n"
        if minsources > 1:
            conf += "minsources {}\n".format(minsources)
        else:
            conf += "#minsources 2\n"
        conf += "\n"

        conf += self.get_chrony_conf_allows()

        conf += self.get_chrony_conf_cmdallows()

        conf += "# Serve time even if not synchronized to a time source.\n"
        if orphan_stratum > 0 and orphan_stratum < 16:
            conf += "local stratum {} orphan\n".format(orphan_stratum)
        elif local_stratum > 0 and local_stratum < 16:
            conf += "local stratum {}\n".format(local_stratum)
        else:
            conf += "#local stratum 10\n"
        conf += "\n"

        conf += "# Specify file containing keys for NTP authentication.\n"
        conf += "keyfile {}\n".format(chrony_keys_path)
        conf += "\n"

        conf += "# Get TAI-UTC offset and leap seconds from the system tz database.\n"
        conf += "leapsectz right/UTC\n"
        conf += "\n"

        conf += "# Specify directory for log files.\n"
        conf += "logdir /var/log/chrony\n"
        conf += "\n"

        conf += "# Select which information is logged.\n"
        if logs:
            conf += "log {}\n".format(" ".join(logs))
        else:
            conf += "#log measurements statistics tracking\n"

        return conf

    def get_chrony_keys(self):
        if not self.keyfile:
            return ""

        keys = "# This file was converted from {}.\n".format(self.keyfile)
        keys += "\n"

        for key in self.keys:
            id = key[0]
            type = key[1]
            password = key[2]

            if type in ["m", "M"]:
                type = "MD5"
            elif type not in ["MD5", "SHA1", "SHA256", "SHA384", "SHA512"]:
                continue

            prefix = "ASCII" if len(password) <= 20 else "HEX"

            for first, last in self.trusted_keys:
                if first <= id <= last:
                    trusted = True
                    break
            else:
                trusted = False

            # Disable keys that were not marked as trusted
            if not trusted:
                keys += "#"

            keys += "{} {} {}:{}\n".format(id, type, prefix, password)

        return keys

    def write_file(self, path, mode, content, backup):
        path = self.root_dir + path
        if backup and os.path.isfile(path):
            os.rename(path, path + ".old")

        with open(os.open(path, os.O_CREAT | os.O_WRONLY | os.O_EXCL, mode), "w",
                  encoding="latin-1") as f:
            if self.verbose > 0:
                print("Writing " + path)
            f.write(u"" + content)

        # Fix SELinux context if restorecon is installed
        try:
            subprocess.call(["restorecon", path])
        except OSError:
            pass


def main():
    parser = argparse.ArgumentParser(description="Convert ntp configuration to chrony.")
    parser.add_argument("-r", "--root", dest="roots", default=["/"], nargs="+",
                        metavar="DIR", help="specify root directory (default /)")
    parser.add_argument("--ntp-conf", action="store", default="/etc/ntp.conf",
                        metavar="FILE", help="specify ntp config (default /etc/ntp.conf)")
    parser.add_argument("--step-tickers", action="store", default="",
                        metavar="FILE", help="specify ntpdate step-tickers config (no default)")
    parser.add_argument("--chrony-conf", action="store", default="/etc/chrony.conf",
                        metavar="FILE", help="specify chrony config (default /etc/chrony.conf)")
    parser.add_argument("--chrony-keys", action="store", default="/etc/chrony.keys",
                        metavar="FILE", help="specify chrony keyfile (default /etc/chrony.keys)")
    parser.add_argument("-b", "--backup", action="store_true", help="backup existing configs before writing")
    parser.add_argument("-L", "--ignored-lines", action="store_true", help="print ignored lines")
    parser.add_argument("-D", "--ignored-directives", action="store_true",
                        help="print names of ignored directives")
    parser.add_argument("-n", "--dry-run", action="store_true", help="don't make any changes")
    parser.add_argument("-v", "--verbose", action="count", default=0, help="increase verbosity")

    args = parser.parse_args()

    for root in args.roots:
        conf = NtpConfiguration(root, args.ntp_conf, args.step_tickers, args.verbose)

        if args.ignored_lines:
            for line in conf.ignored_lines:
                print(line)

        if args.ignored_directives:
            for directive in conf.ignored_directives:
                print(directive)

        conf.write_chrony_configuration(args.chrony_conf, args.chrony_keys, args.dry_run, args.backup)

if __name__ == "__main__":
    main()