Files
2026-08-27 11:12:22 +02:00

590 lines
21 KiB
Python
Executable File

#!/usr/bin/env python3
"""Fail-closed parser for the MARTe2 S03 FileWriter binary contract."""
import argparse
import struct
import sys
from pathlib import Path
UINT32_TYPE_HEX = "0408"
UINT32_BYTES = 4
SIGNAL_HEADER_BYTES = 2 + 32 + 4
SENTINEL_SCALAR_A = 324508639 # 0x13579BDF
SENTINEL_SCALAR_B = 610839776 # 0x2468ACE0
SENTINEL_VECTOR = (16909060, 286397204, 555885348)
EXPECTED_SIGNALS = (
("Counter", UINT32_TYPE_HEX, 1),
("Time", UINT32_TYPE_HEX, 1),
("State1_Thread1_CycleTime", UINT32_TYPE_HEX, 1),
("SentinelScalarA", UINT32_TYPE_HEX, 1),
("SentinelScalarB", UINT32_TYPE_HEX, 1),
("SentinelVector", UINT32_TYPE_HEX, 3),
)
EXPECTED_OFFSETS = (0, 4, 8, 12, 16, 20)
EXPECTED_RECORD_BYTES = 32
EXPECTED_HEADER_BYTES = 4 + len(EXPECTED_SIGNALS) * SIGNAL_HEADER_BYTES
STATUS_ORDER = (
"HEADER_STATUS",
"RECORD_LAYOUT_STATUS",
"PAYLOAD_ALIGNMENT_STATUS",
"TRUNCATION_STATUS",
"SIGNAL_COUNT_STATUS",
"SIGNAL_NAMES_STATUS",
"SIGNAL_TYPES_STATUS",
"SIGNAL_ELEMENTS_STATUS",
"SIGNAL_ORDER_STATUS",
"COUNTER_STATUS",
"TIME_STATUS",
"CYCLE_TIME_STATUS",
"SENTINEL_SCALAR_A_STATUS",
"SENTINEL_SCALAR_B_STATUS",
"SENTINEL_VECTOR_STATUS",
"SENTINEL_SEQUENCE_STATUS",
"FUNCTIONAL_BINARY_STATUS",
"SEMANTIC_BINARY_STATUS",
"BINARY_STATUS",
)
def safe_value(value):
return str(value).replace("\r", "\\r").replace("\n", "\\n")
class Report:
def __init__(self):
self.status = {name: "NOT_EVALUATED" for name in STATUS_ORDER}
self.details = {}
self.first_error = {
"FIRST_ERROR_CODE": "NONE",
"EXPECTED": "NOT_APPLICABLE",
"OBSERVED": "NOT_APPLICABLE",
"RECORD_INDEX": "NOT_APPLICABLE",
"SIGNAL_NAME": "NOT_APPLICABLE",
"ELEMENT_INDEX": "NOT_APPLICABLE",
"BYTE_OFFSET": "NOT_APPLICABLE",
}
def set_error(
self,
code,
expected="NOT_APPLICABLE",
observed="NOT_APPLICABLE",
record_index="NOT_APPLICABLE",
signal_name="NOT_APPLICABLE",
element_index="NOT_APPLICABLE",
byte_offset="NOT_APPLICABLE",
):
if self.first_error["FIRST_ERROR_CODE"] != "NONE":
return
self.first_error.update(
{
"FIRST_ERROR_CODE": safe_value(code),
"EXPECTED": safe_value(expected),
"OBSERVED": safe_value(observed),
"RECORD_INDEX": safe_value(record_index),
"SIGNAL_NAME": safe_value(signal_name),
"ELEMENT_INDEX": safe_value(element_index),
"BYTE_OFFSET": safe_value(byte_offset),
}
)
def emit(self):
for key in STATUS_ORDER:
print(f"{key}={self.status[key]}")
for key in sorted(self.details):
print(f"{key}={safe_value(self.details[key])}")
for key in (
"FIRST_ERROR_CODE",
"EXPECTED",
"OBSERVED",
"RECORD_INDEX",
"SIGNAL_NAME",
"ELEMENT_INDEX",
"BYTE_OFFSET",
):
print(f"{key}={self.first_error[key]}")
def padded_name(raw):
try:
return raw.split(b"\x00", 1)[0].decode("ascii")
except UnicodeDecodeError:
return raw.split(b"\x00", 1)[0].decode("ascii", errors="replace")
def finish_early(report, exit_code):
report.status["FUNCTIONAL_BINARY_STATUS"] = "FAIL"
report.status["SEMANTIC_BINARY_STATUS"] = "NOT_DEMONSTRATED"
report.status["BINARY_STATUS"] = "FAIL"
report.emit()
return exit_code
def parse_binary(path, min_samples):
report = Report()
try:
data = path.read_bytes()
except OSError as exc:
report.status["HEADER_STATUS"] = "FAIL"
report.status["TRUNCATION_STATUS"] = "FAIL"
report.details["FILE_SIZE_BYTES"] = "NOT_AVAILABLE"
report.set_error("FILE_READ_ERROR", observed=exc)
return report, finish_early(report, 31), True
report.details["FILE_SIZE_BYTES"] = len(data)
report.details["EXPECTED_HEADER_BYTES"] = EXPECTED_HEADER_BYTES
report.details["EXPECTED_RECORD_BYTES"] = EXPECTED_RECORD_BYTES
report.details["EXPECTED_SIGNAL_COUNT"] = len(EXPECTED_SIGNALS)
if len(data) < 4:
report.status["HEADER_STATUS"] = "FAIL"
report.status["TRUNCATION_STATUS"] = "FAIL"
report.status["SIGNAL_COUNT_STATUS"] = "NOT_EVALUATED"
report.set_error(
"HEADER_TOO_SMALL",
expected="AT_LEAST_4_BYTES",
observed=f"{len(data)}_BYTES",
byte_offset=len(data),
)
return report, 31, False
signal_count = struct.unpack_from("<I", data, 0)[0]
report.details["OBSERVED_SIGNAL_COUNT"] = signal_count
if signal_count > 4096:
report.status["HEADER_STATUS"] = "FAIL"
report.status["TRUNCATION_STATUS"] = "FAIL"
report.status["SIGNAL_COUNT_STATUS"] = "FAIL"
report.set_error(
"SIGNAL_COUNT_UNREASONABLE",
expected=len(EXPECTED_SIGNALS),
observed=signal_count,
byte_offset=0,
)
return report, 32, False
if signal_count == len(EXPECTED_SIGNALS):
report.status["SIGNAL_COUNT_STATUS"] = "PASS"
else:
report.status["SIGNAL_COUNT_STATUS"] = "FAIL"
report.set_error(
"SIGNAL_COUNT_MISMATCH",
expected=len(EXPECTED_SIGNALS),
observed=signal_count,
byte_offset=0,
)
offset = 4
observed = []
for index in range(signal_count):
if offset + SIGNAL_HEADER_BYTES > len(data):
report.status["HEADER_STATUS"] = "FAIL"
report.status["TRUNCATION_STATUS"] = "FAIL"
report.set_error(
"TRUNCATED_SIGNAL_HEADER",
expected=SIGNAL_HEADER_BYTES,
observed=max(0, len(data) - offset),
signal_name=f"SIGNAL_{index}",
byte_offset=offset,
)
report.details["OBSERVED_HEADER_BYTES"] = offset
return report, 33, False
type_offset = offset
type_hex = data[offset : offset + 2].hex()
offset += 2
name_offset = offset
name = padded_name(data[offset : offset + 32])
offset += 32
elements_offset = offset
elements = struct.unpack_from("<I", data, offset)[0]
offset += 4
observed.append(
{
"name": name,
"type_hex": type_hex,
"elements": elements,
"type_offset": type_offset,
"name_offset": name_offset,
"elements_offset": elements_offset,
}
)
report.details[f"SIGNAL_{index}_NAME"] = name
report.details[f"SIGNAL_{index}_TYPE_HEX"] = type_hex
report.details[f"SIGNAL_{index}_NUMBER_OF_ELEMENTS"] = elements
report.status["HEADER_STATUS"] = "PASS"
report.details["OBSERVED_HEADER_BYTES"] = offset
header_bytes = offset
observed_names = [item["name"] for item in observed]
expected_names = [item[0] for item in EXPECTED_SIGNALS]
report.details["OBSERVED_SIGNAL_NAMES"] = ",".join(observed_names)
report.details["EXPECTED_SIGNAL_NAMES"] = ",".join(expected_names)
duplicates = sorted({name for name in observed_names if observed_names.count(name) > 1})
if duplicates:
report.status["SIGNAL_NAMES_STATUS"] = "FAIL"
report.status["SIGNAL_ORDER_STATUS"] = "FAIL"
report.set_error(
"DUPLICATE_SIGNAL_NAME",
expected="UNIQUE_NAMES",
observed=",".join(duplicates),
)
elif sorted(observed_names) == sorted(expected_names):
report.status["SIGNAL_NAMES_STATUS"] = "PASS"
if observed_names == expected_names:
report.status["SIGNAL_ORDER_STATUS"] = "PASS"
else:
report.status["SIGNAL_ORDER_STATUS"] = "FAIL"
first_index = next(
index
for index, pair in enumerate(zip(observed_names, expected_names))
if pair[0] != pair[1]
)
report.set_error(
"SIGNAL_ORDER_MISMATCH",
expected=expected_names[first_index],
observed=observed_names[first_index],
signal_name=observed_names[first_index],
byte_offset=observed[first_index]["name_offset"],
)
else:
report.status["SIGNAL_NAMES_STATUS"] = "FAIL"
report.status["SIGNAL_ORDER_STATUS"] = "FAIL"
mismatch_index = 0
for mismatch_index in range(min(len(observed_names), len(expected_names))):
if observed_names[mismatch_index] != expected_names[mismatch_index]:
break
observed_name = (
observed_names[mismatch_index]
if mismatch_index < len(observed_names)
else "MISSING"
)
expected_name = (
expected_names[mismatch_index]
if mismatch_index < len(expected_names)
else "NO_ADDITIONAL_SIGNAL"
)
byte_offset = (
observed[mismatch_index]["name_offset"]
if mismatch_index < len(observed)
else header_bytes
)
report.set_error(
"SIGNAL_NAME_MISMATCH",
expected=expected_name,
observed=observed_name,
signal_name=observed_name,
byte_offset=byte_offset,
)
types_ok = len(observed) == len(EXPECTED_SIGNALS)
elements_ok = len(observed) == len(EXPECTED_SIGNALS)
observed_offsets = []
running_offset = 0
for index, item in enumerate(observed):
observed_offsets.append(running_offset)
width = UINT32_BYTES if item["type_hex"] == UINT32_TYPE_HEX else UINT32_BYTES
running_offset += width * item["elements"]
report.details[f"SIGNAL_{index}_OBSERVED_OFFSET"] = observed_offsets[-1]
if index < len(EXPECTED_OFFSETS):
report.details[f"SIGNAL_{index}_EXPECTED_OFFSET"] = EXPECTED_OFFSETS[index]
if index >= len(EXPECTED_SIGNALS):
types_ok = False
elements_ok = False
continue
expected_name, expected_type, expected_elements = EXPECTED_SIGNALS[index]
if item["type_hex"] != expected_type:
types_ok = False
report.set_error(
"SIGNAL_TYPE_MISMATCH",
expected=expected_type,
observed=item["type_hex"],
signal_name=expected_name,
byte_offset=item["type_offset"],
)
if item["elements"] != expected_elements:
elements_ok = False
report.set_error(
"SIGNAL_ELEMENTS_MISMATCH",
expected=expected_elements,
observed=item["elements"],
signal_name=expected_name,
byte_offset=item["elements_offset"],
)
report.status["SIGNAL_TYPES_STATUS"] = "PASS" if types_ok else "FAIL"
report.status["SIGNAL_ELEMENTS_STATUS"] = "PASS" if elements_ok else "FAIL"
report.details["OBSERVED_RECORD_BYTES_FROM_HEADER"] = running_offset
layout_ok = all(
report.status[name] == "PASS"
for name in (
"SIGNAL_COUNT_STATUS",
"SIGNAL_NAMES_STATUS",
"SIGNAL_TYPES_STATUS",
"SIGNAL_ELEMENTS_STATUS",
"SIGNAL_ORDER_STATUS",
)
) and observed_offsets == list(EXPECTED_OFFSETS)
report.status["RECORD_LAYOUT_STATUS"] = "PASS" if layout_ok else "FAIL"
if not layout_ok and report.first_error["FIRST_ERROR_CODE"] == "NONE":
report.set_error(
"RECORD_LAYOUT_MISMATCH",
expected=EXPECTED_RECORD_BYTES,
observed=running_offset,
)
payload_bytes = len(data) - header_bytes
remainder_bytes = payload_bytes % EXPECTED_RECORD_BYTES
sample_count = payload_bytes // EXPECTED_RECORD_BYTES
report.details["PAYLOAD_BYTES"] = payload_bytes
report.details["PAYLOAD_REMAINDER_BYTES"] = remainder_bytes
report.details["COMPLETE_SAMPLE_COUNT"] = sample_count
report.details["MINIMUM_SAMPLE_COUNT"] = min_samples
if remainder_bytes == 0:
report.status["PAYLOAD_ALIGNMENT_STATUS"] = "PASS"
report.status["TRUNCATION_STATUS"] = "PASS"
else:
report.status["PAYLOAD_ALIGNMENT_STATUS"] = "FAIL"
report.status["TRUNCATION_STATUS"] = "FAIL"
report.set_error(
"PAYLOAD_SIZE_NOT_MULTIPLE_OF_RECORD",
expected=f"MULTIPLE_OF_{EXPECTED_RECORD_BYTES}",
observed=f"REMAINDER_{remainder_bytes}",
byte_offset=header_bytes + sample_count * EXPECTED_RECORD_BYTES,
)
if sample_count < min_samples:
report.set_error(
"INSUFFICIENT_SAMPLES",
expected=f"AT_LEAST_{min_samples}",
observed=sample_count,
)
records = []
if remainder_bytes == 0:
for record_index in range(sample_count):
start = header_bytes + record_index * EXPECTED_RECORD_BYTES
records.append(struct.unpack_from("<8I", data, start))
counters = [record[0] for record in records]
times = [record[1] for record in records]
cycle_times = [record[2] for record in records]
counter_error = None
if counters:
if counters[0] != 1:
counter_error = (0, 1, counters[0])
else:
for index in range(1, len(counters)):
if counters[index] != counters[index - 1] + 1:
counter_error = (index, counters[index - 1] + 1, counters[index])
break
else:
counter_error = ("NOT_APPLICABLE", "AT_LEAST_ONE_RECORD", "NO_RECORDS")
if counter_error is None:
report.status["COUNTER_STATUS"] = "PASS"
else:
report.status["COUNTER_STATUS"] = "FAIL"
record_index, expected_counter, observed_counter = counter_error
byte_offset = (
header_bytes + int(record_index) * EXPECTED_RECORD_BYTES
if isinstance(record_index, int)
else header_bytes
)
report.set_error(
"COUNTER_SEQUENCE_MISMATCH",
expected=expected_counter,
observed=observed_counter,
record_index=record_index,
signal_name="Counter",
element_index=0,
byte_offset=byte_offset,
)
time_error = None
for index in range(1, len(times)):
if times[index] < times[index - 1]:
time_error = (index, f">={times[index - 1]}", times[index])
break
if records and time_error is None:
report.status["TIME_STATUS"] = "PASS"
else:
report.status["TIME_STATUS"] = "FAIL"
if time_error is not None:
record_index, expected_time, observed_time = time_error
report.set_error(
"TIME_REGRESSION",
expected=expected_time,
observed=observed_time,
record_index=record_index,
signal_name="Time",
element_index=0,
byte_offset=header_bytes + record_index * EXPECTED_RECORD_BYTES + 4,
)
nonzero_cycle_times = [value for value in cycle_times if value > 0]
if records and nonzero_cycle_times:
report.status["CYCLE_TIME_STATUS"] = "PASS"
else:
report.status["CYCLE_TIME_STATUS"] = "FAIL"
report.set_error(
"NO_NONZERO_CYCLE_TIME",
expected="AT_LEAST_ONE_NONZERO_SAMPLE",
observed=len(nonzero_cycle_times),
signal_name="State1_Thread1_CycleTime",
byte_offset=header_bytes + 8,
)
sentinel_failures = {
"A": None,
"B": None,
"VECTOR": None,
}
for record_index, record in enumerate(records):
if sentinel_failures["A"] is None and record[3] != SENTINEL_SCALAR_A:
sentinel_failures["A"] = (record_index, 0, SENTINEL_SCALAR_A, record[3], 12)
if sentinel_failures["B"] is None and record[4] != SENTINEL_SCALAR_B:
sentinel_failures["B"] = (record_index, 0, SENTINEL_SCALAR_B, record[4], 16)
for element_index, expected_value in enumerate(SENTINEL_VECTOR):
observed_value = record[5 + element_index]
if sentinel_failures["VECTOR"] is None and observed_value != expected_value:
sentinel_failures["VECTOR"] = (
record_index,
element_index,
expected_value,
observed_value,
20 + element_index * UINT32_BYTES,
)
for key, status_name, signal_name in (
("A", "SENTINEL_SCALAR_A_STATUS", "SentinelScalarA"),
("B", "SENTINEL_SCALAR_B_STATUS", "SentinelScalarB"),
("VECTOR", "SENTINEL_VECTOR_STATUS", "SentinelVector"),
):
failure = sentinel_failures[key]
if records and failure is None:
report.status[status_name] = "PASS"
else:
report.status[status_name] = "FAIL"
if failure is not None:
record_index, element_index, expected_value, observed_value, field_offset = failure
code = {
"A": "SENTINEL_SCALAR_A_MISMATCH",
"B": "SENTINEL_SCALAR_B_MISMATCH",
"VECTOR": "SENTINEL_VECTOR_MISMATCH",
}[key]
report.set_error(
code,
expected=expected_value,
observed=observed_value,
record_index=record_index,
signal_name=signal_name,
element_index=element_index,
byte_offset=header_bytes
+ record_index * EXPECTED_RECORD_BYTES
+ field_offset,
)
sentinel_values_ok = all(
report.status[name] == "PASS"
for name in (
"SENTINEL_SCALAR_A_STATUS",
"SENTINEL_SCALAR_B_STATUS",
"SENTINEL_VECTOR_STATUS",
)
)
report.status["SENTINEL_SEQUENCE_STATUS"] = (
"PASS" if sentinel_values_ok and records else "FAIL"
)
if counters:
report.details["COUNTER_FIRST"] = counters[0]
report.details["COUNTER_LAST"] = counters[-1]
if times:
report.details["TIME_FIRST"] = times[0]
report.details["TIME_LAST"] = times[-1]
report.details["NONZERO_CYCLE_TIME_SAMPLES"] = len(nonzero_cycle_times)
if nonzero_cycle_times:
report.details["CYCLE_TIME_MIN"] = min(nonzero_cycle_times)
report.details["CYCLE_TIME_MAX"] = max(nonzero_cycle_times)
base_header_ok = len(observed) >= 3 and all(
(
observed[index]["name"],
observed[index]["type_hex"],
observed[index]["elements"],
)
== EXPECTED_SIGNALS[index]
for index in range(3)
)
functional_ok = (
report.status["HEADER_STATUS"] == "PASS"
and base_header_ok
and report.status["PAYLOAD_ALIGNMENT_STATUS"] == "PASS"
and sample_count >= min_samples
and report.status["COUNTER_STATUS"] == "PASS"
and report.status["TIME_STATUS"] == "PASS"
and report.status["CYCLE_TIME_STATUS"] == "PASS"
)
semantic_ok = (
layout_ok
and report.status["PAYLOAD_ALIGNMENT_STATUS"] == "PASS"
and sample_count >= min_samples
and sentinel_values_ok
and report.status["SENTINEL_SEQUENCE_STATUS"] == "PASS"
)
report.status["FUNCTIONAL_BINARY_STATUS"] = "PASS" if functional_ok else "FAIL"
report.status["SEMANTIC_BINARY_STATUS"] = "PASS" if semantic_ok else "FAIL"
report.status["BINARY_STATUS"] = (
"PASS" if functional_ok and semantic_ok else "FAIL"
)
return report, 0 if report.status["BINARY_STATUS"] == "PASS" else 40, False
def main():
parser = argparse.ArgumentParser(
description="Validate the S03 MARTe2 FileWriter binary contract."
)
parser.add_argument("binary", type=Path)
parser.add_argument("--min-samples", type=int, default=50)
args = parser.parse_args()
if args.min_samples < 1:
parser.error("--min-samples must be at least 1")
report, exit_code, already_emitted = parse_binary(args.binary, args.min_samples)
if not already_emitted:
if report.status["FUNCTIONAL_BINARY_STATUS"] == "NOT_EVALUATED":
report.status["FUNCTIONAL_BINARY_STATUS"] = "FAIL"
if report.status["SEMANTIC_BINARY_STATUS"] == "NOT_EVALUATED":
report.status["SEMANTIC_BINARY_STATUS"] = "NOT_DEMONSTRATED"
if report.status["BINARY_STATUS"] == "NOT_EVALUATED":
report.status["BINARY_STATUS"] = "FAIL"
report.emit()
return exit_code
if __name__ == "__main__":
sys.exit(main())