test_ntp.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright 2017, Florent Thiery
'''
Criticality: Low
Checks that the server is synchronized with the configured NTP server.
'''
from pathlib import Path
import os
import re
import subprocess
import sys

sys.path.append(str(Path(__file__).parents[1].resolve()))

# pylint: disable=wrong-import-position
from envsetup import utils as u  # noqa: E402


def main():
    # Check that ntpd is synced
    if os.path.isfile('/usr/bin/ntpq'):
        cmd = 'LANG=C ntpq -pd'
        expected = 'remote'
        ntpconf = '/etc/ntp.conf'
        ntpconf_expected = r'^(?:server|pool)\s([a-zA-Z0-9\._-]+)(?:\s*iburst)?$'
    else:
        cmd = 'LANG=C timedatectl'
        expected = 'NTP synchronized'
        ntpconf = '/etc/systemd/timesyncd.conf'
        ntpconf_expected = r'^NTP=(.*)$'

    u.log('Running %s' % cmd)
    status = subprocess.getoutput(cmd)
    if expected not in status:
        u.error('NTP not working: %s' % status)
        return 1
    u.success('System is NTP synchronized.')

    os.chdir(os.path.dirname(__file__))

    u.log('Checking NTP server conforms to conf...')
    if not os.path.isfile('../utils.py'):
        u.error('Could not find envsetup conf file or not running from expected location.')
        return 1

    conf = u.load_conf()

    expected_servers = None
    if conf.get('NTP_SERVER'):
        expected_servers = [s.strip() for s in conf['NTP_SERVER'].split(',')]
    if not expected_servers:
        if 'Ubuntu' in subprocess.getoutput('lsb_release -a'):
            expected_servers = ['ntp.ubuntu.com']
        else:
            expected_servers = ['0.debian.pool.ntp.org']

    with open(ntpconf, 'r') as fo:
        content = fo.read()
    servers = list()
    for line in content.split('\n'):
        m = re.match(ntpconf_expected, line)
        if m:
            servers.append(m.groups()[0].strip())
    for expected_server in expected_servers:
        if expected_server not in servers:
            u.warning('Warning: Expected NTP server %s not found in %s, found %s instead.' % (expected_server, ntpconf, ', '.join(servers)))
            return 3
        else:
            u.log('Expected NTP server %s found in configuration (total servers: %s).' % (expected_server, len(servers)))
    u.success('NTP OK.')
    return 0


if __name__ == '__main__':
    code = main()
    sys.exit(code)