409 lines
20 KiB
Python
409 lines
20 KiB
Python
#!/usr/bin/python3 -I
|
|
"""Read-only admission of chrony 4.6 NTS observations for the C01 supervisor.
|
|
|
|
An assessment is not a recovery grant. The protected supervisor must additionally
|
|
bind its measured UTC/BOOTTIME mapping, boot/domain and platform qualification to
|
|
the authority's fresh challenge. No application-supplied report or saved JSON is
|
|
accepted by collect(). Neither this module nor the NTS observer sets host time.
|
|
"""
|
|
import csv
|
|
import hashlib
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal, InvalidOperation, ROUND_CEILING, ROUND_FLOOR
|
|
import io
|
|
import ipaddress
|
|
import os
|
|
from pathlib import Path
|
|
import pwd
|
|
import re
|
|
import selectors
|
|
import stat
|
|
import subprocess
|
|
import time
|
|
|
|
CHRONYC = Path('/usr/bin/chronyc')
|
|
CHRONYD = Path('/usr/sbin/chronyd')
|
|
OBSERVER_USER = 'archipelago-nts'
|
|
SOCKET = Path('/run/archipelago-nts/chronyd.sock')
|
|
PIDFILE = Path('/run/archipelago-nts/chronyd.pid')
|
|
CONFIG = Path('/etc/archipelago-nts/chrony.conf')
|
|
CA_BUNDLE = Path('/etc/ssl/certs/ca-certificates.crt')
|
|
MAX_REPORT = 65536
|
|
NANO = Decimal(1_000_000_000)
|
|
|
|
|
|
class Unavailable(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Policy:
|
|
allowed_names: frozenset[str]
|
|
# These are observation-quality admission limits, not expiry/drift grace.
|
|
max_age_seconds: int = 64
|
|
max_clock_error_ns: int = 250_000_000
|
|
max_skew_ppm: Decimal = Decimal(100)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Assessment:
|
|
reference_id: str
|
|
reference_address: str
|
|
reference_time: Decimal # Last measurement, never a substitute for current UTC.
|
|
system_correction: Decimal # chrony's virtual NTP clock minus system clock
|
|
clock_error_ns: int
|
|
selected: tuple[str, ...]
|
|
good_measurements: tuple[tuple[str, int], ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ClockReading:
|
|
before: int
|
|
wall: int
|
|
after: int
|
|
raw: dict | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Observation:
|
|
assessment: Assessment
|
|
boot_id: str
|
|
utc_ns: int
|
|
boottime_ns: int
|
|
sampling_bracket_ns: int
|
|
observation_error_ns: int
|
|
collected_at_ns: int
|
|
raw: dict | None = None
|
|
|
|
|
|
def configuration(policy):
|
|
require(policy.allowed_names and len(policy.allowed_names) <= 4, 'invalid NTS source policy')
|
|
for name in policy.allowed_names:
|
|
require(type(name) is str and len(name) <= 253 and name == name.lower() and
|
|
re.fullmatch(r'(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?', name),
|
|
'invalid NTS source name')
|
|
lines = [f'server {name} nts iburst minpoll 4 maxpoll 4' for name in sorted(policy.allowed_names)]
|
|
# No DHCP sources, local-clock fallback, arbitrary offsets, alternate trust
|
|
# roots, manual input or certificate-time-check exceptions are accepted.
|
|
lines += ['authselectmode require', 'port 0', 'cmdport 0', f'bindcmdaddress {SOCKET}',
|
|
f'pidfile {PIDFILE}', 'driftfile /run/archipelago-nts/drift', 'logdir /run/archipelago-nts']
|
|
return ('\n'.join(lines) + '\n').encode()
|
|
|
|
|
|
def require(condition, reason):
|
|
if not condition:
|
|
raise Unavailable(reason)
|
|
|
|
|
|
def rows(raw, columns):
|
|
require(type(raw) is str and len(raw.encode()) <= MAX_REPORT and '\x00' not in raw,
|
|
'invalid chrony report size/encoding')
|
|
try:
|
|
result = list(csv.reader(io.StringIO(raw), strict=True))
|
|
except csv.Error:
|
|
raise Unavailable('malformed chrony CSV report') from None
|
|
require(0 < len(result) <= 64 and all(len(row) == columns for row in result),
|
|
'unsupported chrony report shape')
|
|
return result
|
|
|
|
|
|
def decimal(value):
|
|
require(type(value) is str and len(value) <= 48 and
|
|
re.fullmatch(r'[+-]?[0-9]+(?:\.[0-9]+)?', value) is not None,
|
|
'invalid chrony numeric value')
|
|
try:
|
|
number = Decimal(value)
|
|
except InvalidOperation:
|
|
raise Unavailable('invalid chrony numeric value') from None
|
|
require(number.is_finite() and abs(number) <= Decimal(2 ** 53), 'chrony value out of range')
|
|
return number
|
|
|
|
|
|
def integer(value, maximum=2 ** 64 - 1):
|
|
require(type(value) is str and re.fullmatch(r'[0-9]{1,20}', value) is not None,
|
|
'invalid chrony integer')
|
|
result = int(value)
|
|
require(result <= maximum, 'chrony integer out of range')
|
|
return result
|
|
|
|
|
|
def address(value):
|
|
try:
|
|
require('%' not in value, 'scoped NTS address unsupported')
|
|
return str(ipaddress.ip_address(value))
|
|
except ValueError:
|
|
raise Unavailable('invalid NTS address') from None
|
|
|
|
|
|
def indexed(raw, columns, position):
|
|
result = {}
|
|
for row in rows(raw, columns):
|
|
key = address(row[position])
|
|
require(key not in result, 'duplicate chrony source')
|
|
result[key] = row
|
|
return result
|
|
|
|
|
|
def tracking(raw, policy):
|
|
values = rows(raw, 14)
|
|
require(len(values) == 1, 'ambiguous chrony tracking report')
|
|
row = values[0]
|
|
require(re.fullmatch(r'[0-9A-F]{8}', row[0]) is not None and
|
|
row[0] not in ('00000000', '7F7F0101'), 'local/unknown time reference refused')
|
|
reference = address(row[1])
|
|
require(1 <= integer(row[2], 16) <= 15 and row[13] == 'Normal', 'time is not qualified/synchronised')
|
|
epoch, offset = decimal(row[3]), decimal(row[4])
|
|
for value in row[5:13]:
|
|
decimal(value)
|
|
delay, dispersion, skew = decimal(row[10]), decimal(row[11]), decimal(row[9])
|
|
require(epoch > 0 and delay >= 0 and dispersion >= 0 and skew >= 0 and
|
|
decimal(row[6]) >= 0 and decimal(row[12]) > 0, 'invalid time quality')
|
|
require(skew <= policy.max_skew_ppm, 'time frequency is not yet qualified')
|
|
require(abs(decimal(row[7])) <= 100_000 and abs(decimal(row[8])) <= 100_000,
|
|
'clock frequency is outside the qualified range')
|
|
# chronyc(1): absolute clock error <= abs(system offset) + dispersion + delay/2.
|
|
error = int(((abs(offset) + dispersion + delay / 2) * NANO).to_integral_value(rounding=ROUND_CEILING))
|
|
require(error <= policy.max_clock_error_ns, 'time error exceeds admission limit')
|
|
return row[0], reference, epoch, offset, error
|
|
|
|
|
|
def assess(reports, names, policy):
|
|
"""Pure protocol checks for trusted reports; does not authenticate their origin."""
|
|
require(policy.allowed_names and all(type(name) is str and name for name in policy.allowed_names), 'missing source policy')
|
|
require(type(policy.max_age_seconds) is int and policy.max_age_seconds > 0 and
|
|
type(policy.max_clock_error_ns) is int and policy.max_clock_error_ns > 0 and
|
|
policy.max_skew_ppm.is_finite() and policy.max_skew_ppm > 0, 'invalid time policy')
|
|
require(set(reports) == {'tracking_before', 'sources', 'authdata', 'ntpdata', 'tracking_after'},
|
|
'incomplete time observations')
|
|
before = tracking(reports['tracking_before'], policy)
|
|
after = tracking(reports['tracking_after'], policy)
|
|
require(before[:3] == after[:3], 'time reference changed during observation')
|
|
sources = indexed(reports['sources'], 10, 2)
|
|
auth = indexed(reports['authdata'], 10, 0)
|
|
packets = indexed(reports['ntpdata'], 38, 0)
|
|
require(sources.keys() == auth.keys() == packets.keys() == names.keys(), 'source set changed during observation')
|
|
used = []
|
|
primary = []
|
|
measurements = []
|
|
for ip, source in sources.items():
|
|
require(source[0] == '^' and source[1] in ('*', '+', '-', '?', 'x', '~'), 'unsupported time source')
|
|
require(names[ip] in policy.allowed_names, 'unapproved time source')
|
|
# Even an inactive unauthenticated source could subsequently enter the
|
|
# combined estimate. The dedicated observer permits NTS sources only.
|
|
require(auth[ip][1] == 'NTS', 'non-NTS source refused')
|
|
if source[1] not in ('*', '+'):
|
|
continue
|
|
used.append(ip)
|
|
if source[1] == '*':
|
|
primary.append(ip)
|
|
require(re.fullmatch(r'[0-7]{1,3}', source[5]) is not None and int(source[5], 8) > 0,
|
|
'time source unreachable')
|
|
require(integer(source[6]) <= policy.max_age_seconds, 'stale authenticated time')
|
|
require(integer(auth[ip][2]) > 0 and
|
|
(integer(auth[ip][3]), integer(auth[ip][4])) in {(15, 256), (30, 128)} and
|
|
integer(auth[ip][7]) == 0 and 1 <= integer(auth[ip][8]) <= 8 and integer(auth[ip][9]) > 0,
|
|
'NTS key establishment/cookies unavailable')
|
|
packet = packets[ip]
|
|
require(1 <= integer(packet[2], 65535) <= 65535, 'invalid negotiated NTP port')
|
|
require(packet[5] == 'Normal' and packet[6] == '4' and packet[7] == 'Server' and
|
|
1 <= integer(packet[8], 16) <= 15, 'invalid NTP measurement')
|
|
require(packet[23:26] == ['111', '111', '1111'] and packet[27] == 'Yes',
|
|
'latest NTP measurement is not authenticated and accepted')
|
|
for value in packet[12:15] + packet[17:23]:
|
|
decimal(value)
|
|
require(all(decimal(packet[index]) >= 0 for index in (12, 13, 14, 19, 20, 21)) and
|
|
decimal(packet[17]) > 0, 'invalid NTP timing measurement')
|
|
good = integer(packet[33])
|
|
require(0 < good <= integer(packet[32]) <= integer(packet[31]) and integer(packet[30]) > 0,
|
|
'no good authenticated measurement')
|
|
measurements.append((ip, good))
|
|
require(primary == [after[1]], 'selected source differs from tracking reference')
|
|
require(packets[primary[0]][1] == after[0], 'reference identity mismatch')
|
|
return Assessment(after[0], after[1], after[2], after[3], max(before[4], after[4]),
|
|
tuple(sorted(used)), tuple(sorted(measurements)))
|
|
|
|
|
|
def protected_path(path, owners, kind):
|
|
"""Check the command path/producer namespace, including all ancestry."""
|
|
for parent in reversed(path.parents):
|
|
meta = parent.lstat()
|
|
require(stat.S_ISDIR(meta.st_mode) and meta.st_uid in owners and not meta.st_mode & 0o022,
|
|
'unprotected time-service ancestry')
|
|
meta = path.lstat()
|
|
require(meta.st_uid in owners and not stat.S_ISLNK(meta.st_mode), 'unprotected time-service object')
|
|
if kind == 'socket':
|
|
# chronyd uses a permissive socket mode inside its private command
|
|
# directory. A non-owner cannot reach that socket through a 0700 parent.
|
|
parent = path.parent.lstat()
|
|
private_parent = parent.st_uid in owners and parent.st_mode & 0o077 == 0
|
|
require(stat.S_ISSOCK(meta.st_mode) and meta.st_nlink == 1 and
|
|
(not meta.st_mode & 0o007 or private_parent), 'unprotected time-service socket')
|
|
else:
|
|
require(stat.S_ISREG(meta.st_mode) and meta.st_nlink == 1 and not meta.st_mode & 0o022,
|
|
f'unprotected time-service file: {path}')
|
|
return meta.st_dev, meta.st_ino, meta.st_uid, meta.st_gid
|
|
|
|
|
|
def query(command, collection_deadline):
|
|
require(command in (('tracking',), ('sources',), ('authdata',), ('ntpdata',)) or
|
|
len(command) == 2 and command[0] == 'sourcename' and address(command[1]) == command[1],
|
|
'unsupported time-service command')
|
|
# A single Unix path deliberately disables chronyc's default UDP fallback.
|
|
deadline = min(collection_deadline, time.clock_gettime_ns(time.CLOCK_BOOTTIME) + 1_000_000_000)
|
|
require(deadline > time.clock_gettime_ns(time.CLOCK_BOOTTIME), 'time collection timed out')
|
|
child = subprocess.Popen([str(CHRONYC), '-n', '-c', '-h', str(SOCKET), *command],
|
|
env={'PATH': '/usr/bin:/bin', 'LANG': 'C', 'LC_ALL': 'C'}, stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
|
|
output = {child.stdout: bytearray(), child.stderr: bytearray()}
|
|
try:
|
|
with selectors.DefaultSelector() as poller:
|
|
for pipe in output:
|
|
os.set_blocking(pipe.fileno(), False)
|
|
poller.register(pipe, selectors.EVENT_READ)
|
|
while poller.get_map():
|
|
remaining = deadline - time.clock_gettime_ns(time.CLOCK_BOOTTIME)
|
|
require(remaining > 0, 'time-service query timed out')
|
|
for key, _ in poller.select(min(remaining / 1_000_000_000, 0.05)):
|
|
part = os.read(key.fileobj.fileno(), 8192)
|
|
if not part:
|
|
poller.unregister(key.fileobj)
|
|
else:
|
|
output[key.fileobj].extend(part)
|
|
require(sum(map(len, output.values())) <= MAX_REPORT, 'oversized time-service response')
|
|
remaining = deadline - time.clock_gettime_ns(time.CLOCK_BOOTTIME)
|
|
require(remaining > 0 and child.wait(timeout=remaining / 1_000_000_000) == 0,
|
|
'time-service query refused')
|
|
require(not output[child.stderr], 'time-service query diagnostic')
|
|
return output[child.stdout].decode('utf-8', errors='strict')
|
|
finally:
|
|
if child.poll() is None:
|
|
child.kill(); child.wait()
|
|
child.stdout.close(); child.stderr.close()
|
|
|
|
|
|
def daemon_identity(owners, group):
|
|
protected_path(PIDFILE, owners, 'file')
|
|
raw = PIDFILE.read_bytes()
|
|
require(len(raw) <= 32, 'invalid time-service PID file')
|
|
pid = integer(raw.decode('ascii').strip(), 2 ** 31 - 1)
|
|
require(pid > 1, 'invalid time-service PID')
|
|
process = Path('/proc') / str(pid)
|
|
require((process / 'exe').resolve(strict=True) == CHRONYD, 'wrong time-service executable')
|
|
status = dict(line.split(':', 1) for line in (process / 'status').read_text().splitlines() if ':' in line)
|
|
require(set(map(int, status['Uid'].split())) == owners - {0}, 'wrong time-service credentials')
|
|
require(set(map(int, status['Gid'].split())) == {group} and
|
|
set(map(int, status['Groups'].split())) <= {group}, 'unexpected time-service group access')
|
|
require(status['NoNewPrivs'].strip() == '1' and
|
|
all(int(status[field].strip(), 16) & (1 << 25) == 0 for field in ('CapEff', 'CapPrm', 'CapAmb')),
|
|
'time observer retains host-clock control or elevation')
|
|
args = (process / 'cmdline').read_bytes().rstrip(b'\x00').split(b'\x00')
|
|
require(args == [bytes(CHRONYD), b'-x', b'-U', b'-n', b'-u', OBSERVER_USER.encode(), b'-f', bytes(CONFIG)],
|
|
'unexpected time-service invocation')
|
|
environment = (process / 'environ').read_bytes()
|
|
require(len(environment) <= MAX_REPORT, 'oversized time-service environment')
|
|
keys = [entry.split(b'=', 1)[0] for entry in environment.split(b'\x00') if entry]
|
|
require(not any(key.startswith((b'LD_', b'DYLD_', b'GNUTLS_')) or
|
|
key in (b'SSL_CERT_FILE', b'SSL_CERT_DIR', b'SSLKEYLOGFILE', b'NSS_SSLKEYLOGFILE')
|
|
for key in keys), 'unqualified time-service environment')
|
|
raw_stat = (process / 'stat').read_text()
|
|
return pid, int(raw_stat[raw_stat.rindex(')') + 2:].split()[19])
|
|
|
|
|
|
def platform_identity():
|
|
def text(path, limit):
|
|
with Path(path).open('rb') as stream:
|
|
data = stream.read(limit + 1)
|
|
require(0 < len(data) <= limit, 'clock identity input missing/oversized')
|
|
return data.decode('utf-8').strip()
|
|
release = text('/proc/sys/kernel/osrelease', 256)
|
|
version = text('/proc/sys/kernel/version', 1024)
|
|
source = text('/sys/devices/system/clocksource/clocksource0/current_clocksource', 128)
|
|
hardware = hashlib.sha256(b'C01 hardware v1\0')
|
|
for name in ('sys_vendor', 'product_name', 'board_name', 'bios_version', 'bios_date'):
|
|
value = text('/sys/class/dmi/id/' + name, 1024)
|
|
hardware.update(name.encode() + b'\0' + value.encode() + b'\0')
|
|
keys = ('vendor_id', 'cpu family', 'model', 'model name', 'stepping', 'microcode')
|
|
rows = set()
|
|
for line in text('/proc/cpuinfo', 262144).splitlines():
|
|
if ':' in line:
|
|
key, value = line.split(':', 1)
|
|
if key.strip() in keys:
|
|
rows.add(key.strip() + '=' + value.strip())
|
|
require(all(any(row.startswith(key + '=') for row in rows) for key in keys), 'clock CPU identity incomplete')
|
|
for row in sorted(rows):
|
|
hardware.update(row.encode() + b'\0')
|
|
return dict(kernel=list(hashlib.sha256((release + '\0' + version).encode()).digest()),
|
|
clocksource=list(hashlib.sha256(source.encode()).digest()), hardware=list(hardware.digest()))
|
|
|
|
|
|
def clock_reading(raw=False):
|
|
identity = platform_identity() if raw else None
|
|
raw_before = time.clock_gettime_ns(time.CLOCK_MONOTONIC_RAW) if raw else None
|
|
before = time.clock_gettime_ns(time.CLOCK_BOOTTIME)
|
|
wall = time.time_ns()
|
|
after = time.clock_gettime_ns(time.CLOCK_BOOTTIME)
|
|
raw_after = time.clock_gettime_ns(time.CLOCK_MONOTONIC_RAW) if raw else None
|
|
# Identity is checked on both sides of the actual clock witness, not inside
|
|
# its RAW/BOOTTIME bracket. Filesystem/CPU inventory latency is not clock
|
|
# adjustment and must not manufacture an unnecessarily wide RAW anchor.
|
|
if raw:
|
|
require(platform_identity() == identity, 'clock identity changed during collection')
|
|
def duration(n): return dict(secs=n // 1_000_000_000, nanos=n % 1_000_000_000)
|
|
witness = dict(before=duration(raw_before), after=duration(raw_after), identity=identity) if raw else None
|
|
return ClockReading(before, wall, after, witness)
|
|
|
|
|
|
def bind_observation(assessment, start, end, boot_before, boot_after, policy):
|
|
require(re.fullmatch(r'[0-9a-f]{32}', boot_before) is not None and
|
|
boot_before != '0' * 32 and boot_before == boot_after, 'time-observation boot changed')
|
|
require(0 <= start.before <= start.after <= end.before <= end.after,
|
|
'time-observation monotonic regression')
|
|
# CLOCK_REALTIME and CLOCK_BOOTTIME share the Linux adjusted timebase.
|
|
# Their boot-epoch intervals must overlap; a wall-clock step is not silently
|
|
# incorporated into an authenticated observation.
|
|
require(max(start.wall - start.after, end.wall - end.after) <=
|
|
min(start.wall - start.before, end.wall - end.before), 'wall clock stepped during time observation')
|
|
span = end.after - start.before
|
|
error = assessment.clock_error_ns + span
|
|
require(error <= policy.max_clock_error_ns, 'time collection exceeds the qualified error budget')
|
|
# Use the current kernel reading corrected against the authenticated virtual
|
|
# NTP clock. Ref time is deliberately NOT used as the current time.
|
|
correction = int((assessment.system_correction * NANO).to_integral_value(rounding=ROUND_CEILING))
|
|
utc = end.wall + correction
|
|
require(utc > 0, 'invalid authenticated UTC observation')
|
|
reference = int((assessment.reference_time * NANO).to_integral_value(rounding=ROUND_FLOOR))
|
|
require(0 <= utc - reference <= policy.max_age_seconds * 1_000_000_000,
|
|
'stale or future tracking reference')
|
|
return Observation(assessment, boot_before, utc, end.before, end.after - end.before, error, end.after, end.raw)
|
|
|
|
|
|
def collect(policy, raw=False):
|
|
"""Live root-only collection. No cached reports or caller-selected paths."""
|
|
require(os.geteuid() == 0, 'protected time collection requires the supervisor')
|
|
account = pwd.getpwnam(OBSERVER_USER)
|
|
chrony_uid = account.pw_uid
|
|
require(chrony_uid != 0 and account.pw_gid != 0, 'invalid dedicated chrony principal')
|
|
owners = {0, chrony_uid}
|
|
protected_path(CHRONYC, {0}, 'file')
|
|
protected_path(CHRONYD, {0}, 'file')
|
|
protected_path(CA_BUNDLE, {0}, 'file')
|
|
protected_path(CONFIG, {0}, 'file')
|
|
expected_config = configuration(policy)
|
|
require(CONFIG.read_bytes() == expected_config, 'time-service configuration differs from policy')
|
|
socket_identity = protected_path(SOCKET, owners, 'socket')
|
|
producer = daemon_identity(owners, account.pw_gid)
|
|
boot = Path('/proc/sys/kernel/random/boot_id').read_text().strip().replace('-', '')
|
|
start = clock_reading(raw=raw)
|
|
deadline = time.clock_gettime_ns(time.CLOCK_BOOTTIME) + 2_000_000_000
|
|
reports = {'tracking_before': query(('tracking',), deadline), 'sources': query(('sources',), deadline),
|
|
'authdata': query(('authdata',), deadline), 'ntpdata': query(('ntpdata',), deadline)}
|
|
names = {ip: query(('sourcename', ip), deadline).strip() for ip in indexed(reports['sources'], 10, 2)}
|
|
reports['tracking_after'] = query(('tracking',), deadline)
|
|
end = clock_reading(raw=raw)
|
|
require(socket_identity == protected_path(SOCKET, owners, 'socket') and producer == daemon_identity(owners, account.pw_gid),
|
|
'time service changed during observation')
|
|
require(CONFIG.read_bytes() == expected_config, 'time-service configuration changed during observation')
|
|
require(time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline, 'time collection timed out')
|
|
return bind_observation(assess(reports, names, policy), start, end, boot,
|
|
Path('/proc/sys/kernel/random/boot_id').read_text().strip().replace('-', ''), policy)
|