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.
'''
import imp
import os
import re
import subprocess
import sys

YELLOW = '\033[93m'
GREEN = '\033[92m'
RED = '\033[91m'
DEF = '\033[0m'

# 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(.*)$'
else:
    cmd = 'LANG=C timedatectl'
    expected = 'NTP synchronized'
    ntpconf = '/etc/systemd/timesyncd.conf'
    ntpconf_expected = r'^NTP=(.*)$'

print('Running %s' % cmd)
status = subprocess.getoutput(cmd)
if expected not in status:
    print('%sNTP not working: %s%s' % (RED, status, DEF))
    sys.exit(1)
print('%sSystem is NTP synchronized.%s' % (GREEN, DEF))

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

print('Checking NTP server conforms to conf...')
if not os.path.isfile('../utils.py'):
    print('%sCould not find envsetup conf file or not running from expected location.%s' % (RED, DEF))
    sys.exit(1)

es_utils = imp.load_source('es_utils', '../utils.py')
conf = es_utils.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 iburst']

with open(ntpconf, 'r') as f:
    content = f.read()
servers = list()
for l in content.split('\n'):
    m = re.match(ntpconf_expected, l)
    if m:
        servers.append(m.groups()[0].strip())
for expected_server in expected_servers:
    if expected_server not in servers:
        print('%sWarning: Expected NTP server %s not found in %s, found %s instead.%s' % (YELLOW, expected_server, ntpconf, ', '.join(servers), DEF))
        sys.exit(3)
    else:
        print('Expected NTP server %s found in configuration (total servers: %s).' % (expected_server, len(servers)))
print('%sNTP OK.%s' % (GREEN, DEF))