168 lines
9.2 KiB
Python
168 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Protocol/negative controls; synthetic reports never authorize live recovery."""
|
|
from decimal import Decimal
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
|
|
path = Path(__file__).resolve().parents[1] / 'c01_nts.py'
|
|
spec = importlib.util.spec_from_file_location('c01_nts', path)
|
|
nts = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = nts
|
|
spec.loader.exec_module(nts)
|
|
|
|
IP = '203.0.113.7'
|
|
NAME = 'clock.example'
|
|
|
|
|
|
def fixture():
|
|
# Column layout from chrony 4.6.1, with reserved documentation addresses.
|
|
tracking = 'CB007107,203.0.113.7,2,1700000000.000000000,-0.001000000,0.001000000,0.001000000,0.000,0.000,1.000,0.020000000,0.002000000,16.0,Normal\n'
|
|
packet = '203.0.113.7,CB007107,4123,192.0.2.1,C0000201,Normal,4,Server,1,4,16,-24,0.000000060,0.000000,0.000000,50505300,PPS,1700000000.000000000,-0.002000000,0.020000000,0.000100000,0.000004000,0.00,111,111,1111,No,Yes,Kernel,Kernel,9,9,9,9,9,9,0,0\n'
|
|
return {'tracking_before': tracking, 'tracking_after': tracking,
|
|
'sources': '^,*,203.0.113.7,1,4,377,1,0.001000000,0.002000000,0.012000000\n',
|
|
'authdata': '203.0.113.7,NTS,1,15,256,5,0,0,8,100\n', 'ntpdata': packet}
|
|
|
|
|
|
def replace_cell(reports, report, position, value):
|
|
row = reports[report].strip().split(',')
|
|
row[position] = value
|
|
reports[report] = ','.join(row) + '\n'
|
|
|
|
|
|
class NtsReports(unittest.TestCase):
|
|
def setUp(self):
|
|
self.policy = nts.Policy(frozenset({NAME}))
|
|
self.reports = fixture()
|
|
|
|
def assess(self):
|
|
return nts.assess(self.reports, {IP: NAME}, self.policy)
|
|
|
|
def test_authenticated_qualified_source_and_negotiated_nonstandard_port(self):
|
|
value = self.assess()
|
|
self.assertEqual(value.reference_address, IP)
|
|
self.assertEqual(value.system_correction, Decimal('-0.001000000'))
|
|
self.assertEqual(value.clock_error_ns, 13_000_000)
|
|
self.assertEqual(value.good_measurements, ((IP, 9),))
|
|
|
|
def test_configured_nts_without_establishment_does_not_qualify(self):
|
|
for position, value in [(2, '0'), (3, '0'), (4, '0'), (7, '1'), (8, '0'), (9, '0')]:
|
|
with self.subTest(position=position):
|
|
self.reports = fixture()
|
|
replace_cell(self.reports, 'authdata', position, value)
|
|
with self.assertRaises(nts.Unavailable): self.assess()
|
|
|
|
def test_keys_without_an_authenticated_accepted_measurement_do_not_qualify(self):
|
|
for position, value in [(27, 'No'), (23, '011'), (25, '1110'), (33, '0'), (33, '10')]:
|
|
with self.subTest(position=position):
|
|
self.reports = fixture()
|
|
replace_cell(self.reports, 'ntpdata', position, value)
|
|
with self.assertRaises(nts.Unavailable): self.assess()
|
|
|
|
def test_local_clock_unsynchronised_and_reference_changes_refuse(self):
|
|
for report, position, value in [('tracking_after', 0, '7F7F0101'), ('tracking_after', 0, '00000000'),
|
|
('tracking_after', 13, 'Not synchronised'), ('tracking_after', 13, 'Insert second'),
|
|
('tracking_after', 3, '1700000001.000000000')]:
|
|
with self.subTest(value=value):
|
|
self.reports = fixture()
|
|
replace_cell(self.reports, report, position, value)
|
|
with self.assertRaises(nts.Unavailable): self.assess()
|
|
|
|
def test_stale_unreachable_and_non_nts_sources_refuse(self):
|
|
for report, position, value in [('sources', 6, '65'), ('sources', 5, '0'), ('sources', 5, '999'),
|
|
('sources', 0, '#'), ('authdata', 1, 'SK'), ('authdata', 1, '-')]:
|
|
with self.subTest(value=value):
|
|
self.reports = fixture()
|
|
replace_cell(self.reports, report, position, value)
|
|
with self.assertRaises(nts.Unavailable): self.assess()
|
|
|
|
def test_error_bound_is_conservative_and_not_a_drift_allowance(self):
|
|
self.policy = nts.Policy(frozenset({NAME}), max_clock_error_ns=13_000_000)
|
|
self.assess()
|
|
replace_cell(self.reports, 'tracking_after', 10, '0.020000001')
|
|
with self.assertRaises(nts.Unavailable): self.assess()
|
|
|
|
def test_initial_unqualified_frequency_refuses_despite_nts_selection(self):
|
|
replace_cell(self.reports, 'tracking_after', 9, '1000000.000')
|
|
with self.assertRaises(nts.Unavailable): self.assess()
|
|
|
|
def test_malformed_ambiguous_and_nonfinite_reports_refuse(self):
|
|
for report, position, value in [('tracking_after', 4, 'NaN'), ('tracking_after', 10, '-1'),
|
|
('tracking_after', 11, 'Infinity'), ('ntpdata', 18, 'NaN'), ('sources', 2, '203.0.113.7;reboot')]:
|
|
with self.subTest(value=value):
|
|
self.reports = fixture()
|
|
replace_cell(self.reports, report, position, value)
|
|
with self.assertRaises(nts.Unavailable): self.assess()
|
|
for report in ('sources', 'authdata', 'ntpdata', 'tracking_after'):
|
|
self.reports = fixture(); self.reports[report] *= 2
|
|
with self.assertRaises(nts.Unavailable): self.assess()
|
|
self.reports = fixture(); self.reports['ntpdata'] = self.reports['ntpdata'].replace(',Kernel', '', 1)
|
|
with self.assertRaises(nts.Unavailable): self.assess()
|
|
|
|
def test_unapproved_name_and_unknown_reference_refuse(self):
|
|
with self.assertRaises(nts.Unavailable):
|
|
nts.assess(self.reports, {IP: 'foreign.example'}, self.policy)
|
|
replace_cell(self.reports, 'ntpdata', 1, 'CB007108')
|
|
with self.assertRaises(nts.Unavailable): self.assess()
|
|
|
|
def test_even_inactive_plain_ntp_is_not_an_allowed_fallback(self):
|
|
self.reports['sources'] += '^,?,203.0.113.8,1,4,0,4294967295,0,0,0\n'
|
|
self.reports['authdata'] += '203.0.113.8,-,0,0,0,4294967295,0,0,0,0\n'
|
|
self.reports['ntpdata'] += self.reports['ntpdata'].replace(IP, '203.0.113.8')
|
|
with self.assertRaises(nts.Unavailable):
|
|
nts.assess(self.reports, {IP: NAME, '203.0.113.8': NAME}, self.policy)
|
|
|
|
def test_combined_source_must_also_be_authenticated(self):
|
|
self.reports['sources'] += '^,+,203.0.113.8,1,4,377,1,0,0,0\n'
|
|
self.reports['authdata'] += '203.0.113.8,NTS,1,15,256,5,0,0,8,100\n'
|
|
self.reports['ntpdata'] += self.reports['ntpdata'].replace(IP, '203.0.113.8').replace(',No,Yes,', ',No,No,')
|
|
with self.assertRaises(nts.Unavailable):
|
|
nts.assess(self.reports, {IP: NAME, '203.0.113.8': NAME}, self.policy)
|
|
|
|
def test_current_utc_uses_kernel_sample_and_signed_offset_not_last_measurement(self):
|
|
assessment = self.assess()
|
|
start = nts.ClockReading(1_000_000, 1_700_000_001_001_000_005, 1_000_010)
|
|
end = nts.ClockReading(1_100_000, 1_700_000_001_001_100_005, 1_100_010)
|
|
observed = nts.bind_observation(assessment, start, end, '1' * 32, '1' * 32, self.policy)
|
|
# Confirmed against chrony 4.6.1 CSV and human reports: a negative
|
|
# correction means the system clock is fast, so current UTC is lower.
|
|
self.assertEqual(observed.utc_ns, end.wall - 1_000_000)
|
|
self.assertEqual(observed.boottime_ns, end.before)
|
|
self.assertEqual(observed.sampling_bracket_ns, 10)
|
|
self.assertEqual(observed.observation_error_ns, 13_100_010)
|
|
for report in ('tracking_before', 'tracking_after'):
|
|
replace_cell(self.reports, report, 4, '0.001000000')
|
|
slow = nts.bind_observation(self.assess(), start, end, '1' * 32, '1' * 32, self.policy)
|
|
self.assertEqual(slow.utc_ns, end.wall + 1_000_000)
|
|
|
|
def test_boot_change_clock_step_and_slow_collection_do_not_produce_a_mapping(self):
|
|
assessment = self.assess()
|
|
start = nts.ClockReading(1_000_000, 1_700_000_001_001_000_005, 1_000_010)
|
|
end = nts.ClockReading(1_100_000, 1_700_000_001_001_100_005, 1_100_010)
|
|
with self.assertRaises(nts.Unavailable):
|
|
nts.bind_observation(assessment, start, end, '1' * 32, '2' * 32, self.policy)
|
|
with self.assertRaises(nts.Unavailable):
|
|
nts.bind_observation(assessment, start, nts.ClockReading(end.before, end.wall + 100, end.after), '1' * 32, '1' * 32, self.policy)
|
|
with self.assertRaises(nts.Unavailable):
|
|
nts.bind_observation(assessment, start, nts.ClockReading(1_001_000_000, 1_700_000_002_001_000_005, 1_001_000_010), '1' * 32, '1' * 32, self.policy)
|
|
for delta in (-2_000_000_000, 100_000_000_000):
|
|
with self.assertRaises(nts.Unavailable):
|
|
nts.bind_observation(assessment, nts.ClockReading(start.before, start.wall + delta, start.after),
|
|
nts.ClockReading(end.before, end.wall + delta, end.after), '1' * 32, '1' * 32, self.policy)
|
|
|
|
def test_managed_configuration_has_no_plain_ntp_or_clock_setting_path(self):
|
|
value = nts.configuration(self.policy).decode()
|
|
self.assertIn('server clock.example nts iburst minpoll 4 maxpoll 4\n', value)
|
|
self.assertIn('authselectmode require\n', value)
|
|
self.assertIn('cmdport 0\n', value)
|
|
self.assertNotIn('nocerttimecheck', value)
|
|
self.assertNotIn('makestep', value)
|
|
self.assertNotIn('sourcedir', value)
|
|
for name in ('clock.example\nlocal stratum 1', 'clock.example extra', '-x', 'clock.example/other'):
|
|
with self.assertRaises(nts.Unavailable): nts.configuration(nts.Policy(frozenset({name})))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|