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