#!/usr/bin/env python3 """M10 toy receiver, offline Python standard library only. No aircraft/network. Authored teaching artifact, CC BY 4.0. See communications-support-readme.md. Accepted means this toy receiver's checks passed, never human receipt or permission. """ import argparse import csv import hashlib import io import json import os from pathlib import Path import re import stat import sys CASE_IDS = tuple(f"C{i:02}" for i in range(1, 13)) DISPOSITIONS = frozenset(("accepted", "duplicate_ignored", "id_conflict", "expired", "wrong_recipient", "queued_offline", "integrity_mismatch", "stale_plan", "not_mission_data", "future_timestamp")) MESSAGE_KEYS = {"message_id", "kind", "recipient", "plan_version", "created_s", "ttl_s", "payload", "payload_sha256"} def require(condition, explanation): if not condition: raise ValueError(explanation) def integer(value, low, high): return type(value) is int and low <= value <= high def exact_keys(value, keys): return type(value) is dict and set(value) == keys def digest(message): return hashlib.sha256(message["payload"].encode("utf-8")).hexdigest() def message_shape(message): require(exact_keys(message, MESSAGE_KEYS), "message fields must match the contract") for key in ("message_id", "recipient", "plan_version"): require(type(message[key]) is str and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,31}", message[key]) is not None, "invalid bounded ASCII identity") require(message["kind"] in ("mission_data", "heartbeat"), "invalid kind") require(type(message["created_s"]) is int and message["created_s"] >= 0, "invalid created_s") require(integer(message["ttl_s"], 1, 3600), "invalid ttl_s") require(type(message["payload"]) is str, "payload must be text") require(len(message["payload"].encode("utf-8")) <= 4096, "payload too large") require(type(message["payload_sha256"]) is str and re.fullmatch(r"[0-9a-f]{64}", message["payload_sha256"]) is not None, "invalid SHA256 declaration") def unique_object(pairs): result = {} for key, value in pairs: require(key not in result, "duplicate JSON key") result[key] = value return result def reject_constant(_value): raise ValueError("non-finite JSON constant") def bounded_text(path, maximum): with path.open("rb") as handle: require(stat.S_ISREG(os.fstat(handle.fileno()).st_mode), "input must be a regular file") raw = handle.read(maximum + 1) require(len(raw) <= maximum, "input file exceeds limit") return raw.decode("utf-8") def load_fixture(path): data = json.loads(bounded_text(path, 65536), object_pairs_hook=unique_object, parse_constant=reject_constant) require(exact_keys(data, {"schema_version", "evidence_basis", "cases"}), "invalid fixture fields") require(data["schema_version"] == "toy-m10-1", "unsupported schema") require(data["evidence_basis"] == "authored_offline_teaching_fixture", "invalid evidence basis") cases = data["cases"] require(type(cases) is list and 1 <= len(cases) <= 16, "invalid case count") seen = [] for case in cases: require(exact_keys(case, {"case_id", "now_s", "connected", "message", "prior_accepted"}), "invalid case fields") require(type(case["case_id"]) is str and case["case_id"] in CASE_IDS, "unknown case ID") seen.append(case["case_id"]) require(integer(case["now_s"], 0, 86400), "invalid now_s") require(type(case["connected"]) is bool, "connected must be boolean") message_shape(case["message"]) prior = case["prior_accepted"] if prior is not None: message_shape(prior) require(prior["kind"] == "mission_data" and prior["recipient"] == "TEAM-R" and prior["plan_version"] == "PLAN-1", "invalid authored prior identity") require(prior["created_s"] <= case["now_s"] and digest(prior) == prior["payload_sha256"], "invalid authored prior time or digest") require(tuple(seen) == CASE_IDS, "fixture must contain ordered C01 through C12 exactly once") return cases def load_expected(path): reader = csv.DictReader(io.StringIO(bounded_text(path, 4096), newline="")) require(reader.fieldnames == ["case_id", "expected_disposition"], "invalid expected CSV header") rows = list(reader) require(len(rows) == 12, "expected CSV must contain 12 rows") require(tuple(row.get("case_id") for row in rows) == CASE_IDS, "invalid expected IDs/order") for row in rows: require(set(row) == {"case_id", "expected_disposition"} and row["expected_disposition"] in DISPOSITIONS, "invalid expected disposition") return {row["case_id"]: row["expected_disposition"] for row in rows} def receive(case): """Each call resets receiver state; prior_accepted is an authored assumption.""" message = case["message"] if message["kind"] != "mission_data": return "not_mission_data" if message["recipient"] != "TEAM-R": return "wrong_recipient" if message["plan_version"] != "PLAN-1": return "stale_plan" age = case["now_s"] - message["created_s"] if age < 0: return "future_timestamp" if age >= message["ttl_s"]: return "expired" if digest(message) != message["payload_sha256"]: return "integrity_mismatch" prior = case["prior_accepted"] if prior is not None and prior["message_id"] == message["message_id"]: return "duplicate_ignored" if prior == message else "id_conflict" if not case["connected"]: return "queued_offline" return "accepted" def new_output_directory(path): # Reject links/junctions in existing ancestors before creating anything. # This is an ordinary local exercise, not protection against hostile filesystem races. absolute = Path(os.path.abspath(path)) for ancestor in (absolute, *absolute.parents): if os.path.lexists(ancestor): info = ancestor.lstat() require(not stat.S_ISLNK(info.st_mode) and not (getattr(info, "st_file_attributes", 0) & 0x400), "output ancestor cannot be a symlink or reparse point") require(not os.path.lexists(absolute), "output directory already exists") require(absolute.parent.is_dir(), "output parent must already exist") absolute.mkdir() return absolute def main(): folder = Path(__file__).absolute().parent parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--fixture", type=Path, default=folder / "communications-support-fixture.json") parser.add_argument("--expected", type=Path, default=folder / "communications-support-expected.csv") parser.add_argument("--output-dir", type=Path, required=True) args = parser.parse_args() try: cases = load_fixture(args.fixture) expected = load_expected(args.expected) rows = [] for case in cases: actual = receive(case) wanted = expected[case["case_id"]] rows.append((case["case_id"], wanted, actual, str(actual == wanted).lower(), "host_offline_simulation")) directory = new_output_directory(args.output_dir) with (directory / "receiver-results.csv").open("x", encoding="utf-8", newline="") as handle: writer = csv.writer(handle, lineterminator="\n") writer.writerow(("case_id", "expected_disposition", "actual_disposition", "matched", "evidence_level")) writer.writerows(rows) matched = sum(row[3] == "true" for row in rows) summary = {"schema_version": "toy-m10-1", "case_count": len(rows), "matched_count": matched, "evidence_level": "host_offline_simulation", "radio": "NOT_RUN", "flight": "NOT_RUN", "human_receipt": "NOT_RUN", "learner_results": "NOT_RUN"} with (directory / "summary.json").open("x", encoding="utf-8", newline="\n") as handle: handle.write(json.dumps(summary, ensure_ascii=False, indent=2) + "\n") print(json.dumps(summary, sort_keys=True)) return 0 if matched == len(rows) else 1 except (ValueError, TypeError, OSError, UnicodeError, RecursionError) as error: print("M10 input/output rejected: " + str(error), file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())