Prepare NTP wire contract and conformance vectors for review

.github/workflows/wire-vectors.yml

@@ -0,0 +1,10 @@
+name: Draft wire vectors
+on: [push, pull_request]
+jobs:
+  vectors:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
+      - run: python3 -m unittest discover -s tests -v

.gitignore

@@ -1 +1,3 @@
 .lake/
+
+__pycache__/

README.md

@@ -2,7 +2,20 @@
 
 A Lean parser for the NTP wire packet. This is a Proven Commons project: it rebuilds a widely used format in Lean and proves the new code follows its written specification.
 
-Status: not verified. No code has landed yet. Plan issue 1 holds the first slice, and it is unclaimed. The specification has not been approved by a reviewer, so do not call this verified.
+Status: not verified. The first parser slice is claimed in plan issue 1 and remains under its author's ownership. The specification has not been approved by a reviewer.
+
+## Wire contract under review
+
+[WIRE-CONTRACT.md](WIRE-CONTRACT.md) proposes a contract for the parser author to review. It comes with 79 fixed packet vectors, a Python reference oracle, and a JSON adapter runner. It adds no Lean implementation or proof. The protocol contract still needs independent review before integration.
+
+Run the fixtures and boundary checks:
+
+```
+python3 -m unittest discover -s tests -v
+python3 tests/check_vectors.py
+```
+
+A future parser adapter can consume the same fixtures with `python3 tests/check_vectors.py -- COMMAND ARGUMENTS`. The runner checks every response and fails on a missing answer, wrong answer, nonzero exit, or timeout.
 
 ## First slice
 

WIRE-CONTRACT.md

@@ -0,0 +1,24 @@
+# Draft wire contract
+
+This is a review target for the existing parser author. Python fixtures and an executable oracle are included; no Lean parser or proof is added. Issue 1 remains open.
+
+The header follows [RFC 5905 sections 6 and 7.3](https://www.rfc-editor.org/rfc/rfc5905.html#section-7.3). Offsets are bytes, integers are big endian, and timestamp order is reference, origin, receive, transmit.
+
+| Offset | Bytes | Field |
+|---|---|---|
+| 0 | 1 | Leap: bits 7–6; version: 5–3; mode: 2–0 |
+| 1 | 1 | Stratum |
+| 2 | 1 | Signed poll exponent |
+| 3 | 1 | Signed precision exponent |
+| 4 | 4 | Root delay bits |
+| 8 | 4 | Root dispersion bits |
+| 12 | 4 | Reference identifier bytes |
+| 16, 24, 32, 40 | 8 each | Seconds (32 bits), fraction (32 bits) |
+
+Version 3 or 4 is this draft's supported set. All modes and strata are preserved; clock-selection validity is separate. Root fixed-point fields remain raw bits. A zero 64-bit timestamp means unknown. A nonzero fraction with zero seconds is known. Conversion requires an external integer era and returns exact rational Unix seconds. No era is guessed from the clock or packet.
+
+Extension framing follows [RFC 7822 section 3](https://www.rfc-editor.org/rfc/rfc7822.html#section-3). Each extension includes its four-byte header in its length, is at least 16 bytes, has four-byte alignment, and stays before the MAC. Without a MAC, the final extension is at least 28 bytes. Unknown types remain opaque, including padding.
+
+The caller supplies `mac_length`: 0, 4, 20, or 24. The oracle preserves those bytes, including a four-byte crypto-NAK, without authenticating them. It does not guess whether a 20-byte tail is an extension or MAC. Negotiated longer MACs and extension-specific requirements need separate context and are outside this contract.
+
+Run `python3 -m unittest discover -s tests -v`. Feed JSON lines to `python3 tests/wire_oracle.py`. Each fixture has `id`, `request`, and a complete `expect` response. A future parser adapter can use this schema unchanged. Independent review of this contract is required before integration.

tests/check_vectors.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""Run the fixed contract vectors against a JSON-line parser adapter."""
+
+import argparse
+import json
+from pathlib import Path
+import subprocess
+import sys
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument('command', nargs=argparse.REMAINDER,
+                    help='adapter command, optionally after --')
+    args = ap.parse_args()
+    command = args.command
+    if command[:1] == ['--']:
+        command = command[1:]
+    if not command:
+        command = [sys.executable, str(Path(__file__).with_name('wire_oracle.py'))]
+    cases = [json.loads(line) for line in Path(__file__).with_name('wire_vectors.jsonl').read_text().splitlines()]
+    try:
+        run = subprocess.run(command, input=''.join(json.dumps(c['request'])+'\n' for c in cases),
+                             capture_output=True, text=True, timeout=30)
+        if run.returncode:
+            raise ValueError(f'adapter exit {run.returncode}: {run.stderr[:300]}')
+        answers = [json.loads(line) for line in run.stdout.splitlines()]
+        if len(answers) != len(cases):
+            raise ValueError(f'{len(answers)} answers for {len(cases)} vectors')
+        failures = [c['id'] for c, got in zip(cases, answers) if got != c['expect']]
+    except (OSError, subprocess.TimeoutExpired, ValueError) as e:
+        print(f'FAIL: {e}', file=sys.stderr)
+        return 1
+    for name in failures:
+        print(f'FAIL: {name}', file=sys.stderr)
+    print(f'{len(cases)-len(failures)}/{len(cases)} wire vectors agree')
+    return bool(failures)
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tests/test_wire_oracle.py

@@ -0,0 +1,96 @@
+import json
+from pathlib import Path
+import random
+import struct
+import subprocess
+import sys
+import unittest
+from fractions import Fraction
+
+from wire_oracle import WireError, answer, decode, read_header, timestamp_unix
+
+HERE = Path(__file__).parent
+
+
+class WireContractTests(unittest.TestCase):
+    def test_fixed_vectors(self):
+        for line in (HERE / 'wire_vectors.jsonl').read_text().splitlines():
+            case = json.loads(line)
+            with self.subTest(case=case['id']):
+                self.assertEqual(answer(case['request']), case['expect'])
+
+    def test_header_against_struct_unpack(self):
+        rng = random.Random(20261028)
+        for _ in range(2000):
+            raw = bytearray(rng.randbytes(48))
+            raw[0] = (raw[0] & 0xc7) | (rng.choice([3, 4]) << 3)
+            first, stratum, poll, precision, delay, dispersion, ref, *stamps = struct.unpack('!BBbbII4s4Q', raw)
+            got = read_header(raw)
+            self.assertEqual([got[k] for k in ('leap','version','mode','stratum','poll','precision')],
+                             [first >> 6, (first >> 3) & 7, first & 7, stratum, poll, precision])
+            self.assertEqual((got['root_delay_bits'], got['root_dispersion_bits'], got['reference_id_hex']),
+                             (delay, dispersion, ref.hex()))
+            self.assertEqual([s['raw'] for s in got['timestamps']], stamps)
+
+    def test_explicit_era_and_unknown(self):
+        for era in (-1, 0, 1, 2):
+            self.assertIsNone(timestamp_unix(0, era))
+            self.assertEqual(timestamp_unix(1, era), Fraction(era * 2**32 - 2208988800) + Fraction(1, 2**32))
+        self.assertEqual(timestamp_unix(2208988800 << 32, 0), 0)
+        self.assertEqual(timestamp_unix(1, 1)-timestamp_unix(1, 0), 2**32)
+        for era in (None, True, 1.5):
+            with self.assertRaises(WireError):
+                timestamp_unix(1, era)
+        for raw in (-1, 2**64, True):
+            with self.assertRaises(WireError):
+                timestamp_unix(raw, 0)
+
+    def test_all_extension_lengths_at_boundary(self):
+        header = bytes.fromhex('23' + '00'*47)
+        for length in range(0, 81):
+            field = b'\xfe\xfe' + length.to_bytes(2, 'big') + bytes(max(0, length-4))
+            for mac in (0, 4, 20, 24):
+                packet = header + field + bytes(mac)
+                expected = length >= (16 if mac else 28) and length % 4 == 0
+                try:
+                    got = decode(packet, mac)
+                except WireError:
+                    self.assertFalse(expected, (length, mac))
+                else:
+                    self.assertTrue(expected, (length, mac))
+                    self.assertEqual(got['extensions'][0]['length'], length)
+                    self.assertEqual(got['mac_length'], mac)
+        # A maximum-size aligned extension; no heuristic 24-byte-tail cutoff.
+        field = b'\xfe\xfe\xff\xfc' + bytes(65528)
+        self.assertEqual(decode(header + field, 0)['extensions'][0]['length'], 65532)
+
+    def test_mac_ambiguity_is_context(self):
+        header = bytes.fromhex('23' + '00'*47)
+        tail = b'\xfe\xfe\x00\x14' + bytes(16)
+        self.assertEqual(decode(header+tail, 20)['mac_hex'], tail.hex())
+        with self.assertRaisesRegex(WireError, 'short_final_extension_without_mac'):
+            decode(header+tail, 0)
+        for n in (-1, 8, 16, 28, None, True):
+            self.assertEqual(answer({'packet_hex':header.hex(), 'mac_length':n}),
+                             {'error':'unsupported_mac_length'})
+
+    def test_cli_and_adapter_failure_detection(self):
+        good = subprocess.run([sys.executable, str(HERE/'check_vectors.py')], capture_output=True)
+        self.assertEqual(good.returncode, 0, good.stderr)
+        bad = subprocess.run([sys.executable, str(HERE/'check_vectors.py'), '--', sys.executable,
+                              '-c', 'print("{}")'], capture_output=True)
+        self.assertNotEqual(bad.returncode, 0)
+        self.assertIn(b'answers for', bad.stderr)
+        wrong = subprocess.run([sys.executable, str(HERE/'check_vectors.py'), '--', sys.executable,
+                                '-c', 'import sys; [print("{}") for line in sys.stdin]'],
+                               capture_output=True)
+        self.assertNotEqual(wrong.returncode, 0)
+        self.assertIn(b'header-distinct-fields', wrong.stderr)
+
+    def test_malformed_requests(self):
+        for request in (None, [], {}, {'packet_hex':'xx','mac_length':0}):
+            self.assertEqual(answer(request), {'error':'malformed_request'})
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/wire_oracle.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+"""Draft NTP wire-contract oracle. No authentication or era inference.
+
+JSON lines in: {"packet_hex": "...", "mac_length": 0}.
+JSON lines out: the fixture's expected object, or {"error": "..."}.
+RFC 5905 sections 6/7.3; RFC 7822 section 3. See WIRE-CONTRACT.md.
+"""
+
+import json
+import sys
+from fractions import Fraction
+
+
+class WireError(ValueError):
+    """An input outside this explicit draft contract."""
+
+
+def read_header(packet):
+    if len(packet) < 48:
+        raise WireError("short_header")
+    first = packet[0]
+    if (first >> 3) & 7 not in (3, 4):
+        raise WireError("unsupported_version")
+    signed = lambda n: n if n < 128 else n - 256
+    timestamps = []
+    for offset in (16, 24, 32, 40):
+        raw = int.from_bytes(packet[offset:offset + 8], "big")
+        timestamps.append({"raw": raw, "seconds": raw >> 32,
+                           "fraction": raw & 0xffffffff, "unknown": raw == 0})
+    return {"leap": first >> 6, "version": (first >> 3) & 7, "mode": first & 7,
+            "stratum": packet[1], "poll": signed(packet[2]),
+            "precision": signed(packet[3]),
+            "root_delay_bits": int.from_bytes(packet[4:8], "big"),
+            "root_dispersion_bits": int.from_bytes(packet[8:12], "big"),
+            "reference_id_hex": packet[12:16].hex(), "timestamps": timestamps}
+
+
+def timestamp_unix(raw, era):
+    """Exact Unix seconds, given an external era, or None for unknown."""
+    if type(raw) is not int or not 0 <= raw < 2 ** 64:
+        raise WireError("timestamp_range")
+    if type(era) is not int:
+        raise WireError("era_required")
+    if raw == 0:
+        return None
+    return Fraction(era * 2 ** 32 + (raw >> 32) - 2208988800) + Fraction(raw & 0xffffffff, 2 ** 32)
+
+
+def decode(packet, mac_length):
+    """Frame using caller-supplied MAC length; never guess from tail bytes."""
+    if type(mac_length) is not int or mac_length not in (0, 4, 20, 24):
+        raise WireError("unsupported_mac_length")
+    header = read_header(packet)
+    end = len(packet) - mac_length
+    if end < 48:
+        raise WireError("short_mac")
+    offset = 48
+    extensions = []
+    while offset < end:
+        if end - offset < 4:
+            raise WireError("short_extension_header")
+        field_type = int.from_bytes(packet[offset:offset + 2], "big")
+        length = int.from_bytes(packet[offset + 2:offset + 4], "big")
+        if length < 16:
+            raise WireError("short_extension")
+        if length % 4:
+            raise WireError("unaligned_extension")
+        if length > end - offset:
+            raise WireError("extension_overrun")
+        extensions.append({"offset": offset, "type": field_type, "length": length,
+                           "value_and_padding_hex": packet[offset + 4:offset + length].hex()})
+        offset += length
+    if not mac_length and extensions and extensions[-1]["length"] < 28:
+        raise WireError("short_final_extension_without_mac")
+    return {"header": header, "extensions": extensions,
+            "mac_hex": packet[end:].hex() if mac_length else "",
+            "mac_length": mac_length}
+
+
+def answer(request):
+    try:
+        packet = bytes.fromhex(request["packet_hex"])
+        return decode(packet, request["mac_length"])
+    except WireError as e:
+        return {"error": str(e)}
+    except (KeyError, TypeError, ValueError):
+        return {"error": "malformed_request"}
+
+
+def main():
+    for line in sys.stdin:
+        try:
+            request = json.loads(line)
+        except ValueError:
+            result = {"error": "malformed_request"}
+        else:
+            result = answer(request)
+        print(json.dumps(result, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()