#!/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()