时钟同步约束机器之间的时间偏差

from pathlib import Path
from tempfile import TemporaryDirectory
import json
import os
import socket
import struct
import subprocess
import time

assert os.geteuid() == 0 and Path('/.dockerenv').exists(), 'use a disposable root Linux container'
NTP_EPOCH = 2208988800
ERA = 2**32


def stamp(unix_time):
    value = unix_time + NTP_EPOCH
    whole = int(value)
    return struct.pack('!II', whole % ERA, int((value - whole) * ERA))


def seconds(data, reference):
    whole, fraction = struct.unpack('!II', data)
    era = round((reference + NTP_EPOCH - whole) / ERA)
    return whole + era * ERA - NTP_EPOCH + fraction / ERA


def estimate(t1, t2, t3, t4):
    return ((t2 - t1) + (t3 - t4)) / 2, (t4 - t1) - (t3 - t2)


def validate(data, sent):
    if len(data) < 48:
        raise ValueError('short packet')
    if data[0] >> 6 == 3:
        raise ValueError('unsynchronised source')
    if (data[0] & 7) != 4 or ((data[0] >> 3) & 7) != 4:
        raise ValueError('unexpected mode or version')
    if not 1 <= data[1] <= 15:
        raise ValueError('unusable stratum')
    if data[24:32] != sent:
        raise ValueError('origin timestamp mismatch')


with TemporaryDirectory(prefix='ntp-exchange-') as directory:
    root = Path(directory)
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as reservation:
        reservation.bind(('127.0.0.1', 0))
        port = reservation.getsockname()[1]
    config = root / 'chrony.conf'
    config.write_text(f'pidfile {root}/pid\ncmdport 0\nbindcmdaddress {root}/command.sock\n'
                      f'port {port}\nbindaddress 127.0.0.1\nlocal stratum 8\nallow 127.0.0.1\n')
    process = subprocess.Popen(['chronyd', '-x', '-n', '-u', 'root', '-f', str(config)],
                               stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    try:
        deadline = time.monotonic() + 5
        while not (root/'command.sock').exists():
            assert process.poll() is None
            if time.monotonic() > deadline:
                raise AssertionError('chronyd did not start')
            time.sleep(0.05)
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
            client.settimeout(5)
            client.connect(('127.0.0.1', port))
            request = bytearray(48)
            request[0] = 0x23
            t1, start = time.time(), time.monotonic()
            sent = stamp(t1)
            request[40:48] = sent
            client.send(request)
            data = client.recv(512)
            t4, finish = time.time(), time.monotonic()
        validate(data, sent)
        t2, t3 = seconds(data[32:40], t1), seconds(data[40:48], t1)
        offset, delay = estimate(t1, t2, t3, t4)
        assert data[1] == 8 and finish >= start
        bad_origin = bytearray(data); bad_origin[24:32] = bytes(8)
        alarm = bytearray(data); alarm[0] |= 0xC0
        denied = bytearray(data); denied[1] = 0
        rejected = []
        for candidate in (data[:40], bad_origin, alarm, denied):
            try:
                validate(candidate, sent)
            except ValueError as error:
                rejected.append(str(error))
            else:
                raise AssertionError('invalid packet accepted')
        symmetric = estimate(100, 105.020, 105.022, 100.042)
        asymmetric = estimate(100, 105.010, 105.012, 100.042)
        assert abs(symmetric[0] - 5) < 1e-9 and abs(symmetric[1] - 0.040) < 1e-9
        assert abs(asymmetric[0] - 4.990) < 1e-9 and abs(asymmetric[1] - 0.040) < 1e-9
        print(json.dumps({'actual_loopback_sample': {'stratum': data[1],
                          'offset_seconds': offset, 'delay_seconds': delay,
                          'monotonic_elapsed_seconds': finish-start},
                          'rejected_packets': rejected,
                          'synthetic_symmetric_estimate': list(symmetric),
                          'synthetic_asymmetric_estimate': list(asymmetric),
                          'known_synthetic_offset': 5,
                          'system_clock_adjustment_disabled': True,
                          'external_UTC_accuracy_tested': False}))
    finally:
        process.terminate()
        process.communicate(timeout=5)