#!/usr/bin/python3 # # Convert ntp configuration to chrony # # Copyright (C) 2018-2019 Miroslav Lichvar # # Permission is hereby granted, free of charge, to any person obtaining # a copy of this software and associated documentation files (the # "Software"), to deal in the Software without restriction, including # without limitation the rights to use, copy, modify, merge, publish, # distribute, sublicense, and/or sell copies of the Software, and to # permit persons to whom the Software is furnished to do so, subject to # the following conditions: # # The above copyright notice and this permission notice shall be included # in all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. # IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY # CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, # TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE # SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. import argparse import ipaddress import logging import os 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): self.root_dir = root_dir if root_dir != "/" else "" self.ntp_conf_path = ntp_conf self.step_tickers_path = step_tickers # Read and write files using an 8-bit transparent encoding self.file_encoding = "latin-1" 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"]: service_path = os.path.join(self.root_dir, "etc/systemd/system/multi-user.target.wants/{}.service".format(service)) if os.path.islink(service_path): self.enabled_services.add(service) logging.info("Enabled services found in /etc/systemd/system: %s", " ".join(self.enabled_services)) def parse_step_tickers(self): if not self.step_tickers_path: return path = os.path.join(self.root_dir, self.step_tickers_path) if not os.path.isfile(path): logging.info("Missing %s", path) return with open(path, encoding=self.file_encoding) 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 = os.path.join(self.root_dir, self.ntp_conf_path) with open(path, encoding=self.file_encoding) as f: logging.info("Reading %s", 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, source_type, words): ipv4_only = False ipv6_only = False source = { "type": source_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 and words[0] in ["stratum"]: if not words[1].isdigit(): return False options[words[0]] = int(words[1]) words = words[2:] elif len(words) >= 2: 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: # Count bits in the mask (ipaddress does not support # expanded IPv6 netmasks) mask_ip = ipaddress.ip_address(mask) mask_str = "{0:0{1}b}".format(int(mask_ip), mask_ip.max_prefixlen) networks.append(ipaddress.ip_network( u"{}/{}".format(address, len(mask_str.rstrip('0'))))) 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", "orphan"]: if not words[1].isdigit(): return False options[words[0]] = int(words[1]) words = words[2:] elif len(words) >= 2 and words[0] in ["maxdist"]: # Check if it is a float value if not words[1].replace('.', '', 1).isdigit(): return False options[words[0]] = float(words[1]) words = words[2:] else: return False self.tos_options.update(options) return True def parse_includefile(self, words): path = os.path.join(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 = os.path.join(self.root_dir, keyfile) if not os.path.isfile(path): logging.info("Missing %s", path) return False with open(path, encoding=self.file_encoding) as f: logging.info("Reading %s", 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) logging.debug("Generated %s:\n%s", 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: logging.debug("Generated %s:\n%s", chrony_keys_path, chrony_keys) if not dry_run: self.write_file(chrony_keys_path, 0o640, chrony_keys, backup) def get_processed_time_sources(self): # Convert {0,1,2,3}.*pool.ntp.org servers to 2.*pool.ntp.org pools # Make shallow copies of all sources (only type will be modified) time_sources = [s.copy() for s in self.time_sources] pools = {} for source in time_sources: if source["type"] != "server": continue m = re.match("^([0123])(\\.\\w+)?\\.pool\\.ntp\\.org$", source["address"]) if m is None: continue number = m.group(1) zone = m.group(2) if zone not in pools: pools[zone] = [] pools[zone].append((int(number), source)) remove_servers = set() for zone, pool in pools.items(): # sort and skip all pools not in [0, 3] range pool.sort() if [number for number, source in pool] != [0, 1, 2, 3]: # only exact group of 4 servers can be converted, nothing to do here continue # verify that parameters are the same for all servers in the pool if not all([p[1]["options"] == pool[0][1]["options"] for p in pool]): break remove_servers.update([pool[i][1]["address"] for i in [0, 1, 3]]) pool[2][1]["type"] = "pool" processed_sources = [] for source in time_sources: if source["type"] == "server" and source["address"] in remove_servers: continue processed_sources.append(source) return processed_sources 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.get_processed_time_sources(): address = source["address"] if address.startswith("127.127."): if address.startswith("127.127.1."): continue # No other refclocks are expected from the parser 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: # No other options are expected from the parser 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 = self.fudges[address]["stratum"] else: local_stratum = 5 if "maxdist" in self.tos_options: maxdistance = self.tos_options["maxdist"] if "minsane" in self.tos_options: minsources = self.tos_options["minsane"] if "orphan" in self.tos_options: orphan_stratum = self.tos_options["orphan"] 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: key_id = key[0] key_type = key[1] password = key[2] if key_type in ["m", "M"]: key_type = "MD5" elif key_type == "AES128CMAC": key_type = "AES128" elif key_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 <= key_id <= last: trusted = True break else: trusted = False # Disable keys that were not marked as trusted if not trusted: keys += "#" keys += "{} {} {}:{}\n".format(key_id, key_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=self.file_encoding) as f: logging.info("Writing %s", 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() logging.basicConfig(format="%(message)s", level=[logging.ERROR, logging.INFO, logging.DEBUG][min(args.verbose, 2)]) for root in args.roots: conf = NtpConfiguration(root, args.ntp_conf, args.step_tickers) 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()