S03-H3 CLOSED baseline
This commit is contained in:
@@ -0,0 +1,331 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fail-closed structural verifier for the S03-H3 MINIMAL profile."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
EXPECTED_TEST_SHA256 = (
|
||||
"d037561b32760fbd0e2aec9f1eefc5efda27b1918eca4742c89092a8a2f7e92e"
|
||||
)
|
||||
PLACEHOLDER = "__MARTE_RUN_DIR__"
|
||||
TEST_FUNCTIONS = (
|
||||
" Functions = "
|
||||
"{GAMTimer SentinelProducer GAMDisplay IOGAM_Writer}"
|
||||
)
|
||||
MINIMAL_FUNCTIONS = (
|
||||
" Functions = "
|
||||
"{GAMTimer SentinelProducer IOGAM_Writer}"
|
||||
)
|
||||
|
||||
|
||||
def sha256_bytes(data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def read_utf8(path: Path) -> tuple[bytes, str]:
|
||||
data = path.read_bytes()
|
||||
return data, data.decode("utf-8")
|
||||
|
||||
|
||||
def object_matches(text: str, name: str) -> list[re.Match[str]]:
|
||||
pattern = re.compile(
|
||||
rf"(?m)^[ \t]*\+{re.escape(name)}[ \t]*=[ \t]*\{{[ \t]*(?:\r?\n|$)"
|
||||
)
|
||||
return list(pattern.finditer(text))
|
||||
|
||||
|
||||
def object_count(text: str, name: str) -> int:
|
||||
return len(object_matches(text, name))
|
||||
|
||||
|
||||
def block_span(text: str, name: str) -> tuple[int, int]:
|
||||
matches = object_matches(text, name)
|
||||
if len(matches) != 1:
|
||||
raise ValueError(f"{name.upper()}_ANCHOR_COUNT_{len(matches)}")
|
||||
|
||||
match = matches[0]
|
||||
opening = text.find("{", match.start(), match.end())
|
||||
if opening < 0:
|
||||
raise ValueError(f"{name.upper()}_OPENING_BRACE_MISSING")
|
||||
|
||||
depth = 0
|
||||
quoted = False
|
||||
escaped = False
|
||||
closing = -1
|
||||
for index in range(opening, len(text)):
|
||||
char = text[index]
|
||||
if quoted:
|
||||
if escaped:
|
||||
escaped = False
|
||||
elif char == "\\":
|
||||
escaped = True
|
||||
elif char == '"':
|
||||
quoted = False
|
||||
continue
|
||||
if char == '"':
|
||||
quoted = True
|
||||
elif char == "{":
|
||||
depth += 1
|
||||
elif char == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
closing = index + 1
|
||||
break
|
||||
if depth < 0:
|
||||
raise ValueError(f"{name.upper()}_UNBALANCED_BRACE")
|
||||
|
||||
if closing < 0:
|
||||
raise ValueError(f"{name.upper()}_CLOSING_BRACE_MISSING")
|
||||
|
||||
if text.startswith("\r\n", closing):
|
||||
closing += 2
|
||||
elif text.startswith("\n", closing):
|
||||
closing += 1
|
||||
return match.start(), closing
|
||||
|
||||
|
||||
def derive_expected_minimal(canonical: str) -> str:
|
||||
if canonical.count(TEST_FUNCTIONS) != 1:
|
||||
raise ValueError(
|
||||
f"FUNCTIONS_ANCHOR_COUNT_{canonical.count(TEST_FUNCTIONS)}"
|
||||
)
|
||||
if canonical.count(MINIMAL_FUNCTIONS) != 0:
|
||||
raise ValueError("SOURCE_ALREADY_PARTIALLY_TRANSFORMED")
|
||||
|
||||
spans = [
|
||||
block_span(canonical, "GAMDisplay"),
|
||||
block_span(canonical, "Display"),
|
||||
]
|
||||
if spans[0][0] < spans[1][1] and spans[1][0] < spans[0][1]:
|
||||
raise ValueError("AUTHORIZED_BLOCK_SPANS_OVERLAP")
|
||||
|
||||
transformed = canonical
|
||||
for start, end in sorted(spans, reverse=True):
|
||||
transformed = transformed[:start] + transformed[end:]
|
||||
transformed = transformed.replace(TEST_FUNCTIONS, MINIMAL_FUNCTIONS, 1)
|
||||
return transformed
|
||||
|
||||
|
||||
class Report:
|
||||
def __init__(self) -> None:
|
||||
self.values: list[tuple[str, str]] = []
|
||||
self.failed = False
|
||||
self.first_error = "NONE"
|
||||
|
||||
def value(self, name: str, value: object) -> None:
|
||||
self.values.append((name, str(value)))
|
||||
|
||||
def status(self, name: str, passed: bool, error: str) -> None:
|
||||
self.value(name, "PASS" if passed else "FAIL")
|
||||
if not passed:
|
||||
self.failed = True
|
||||
if self.first_error == "NONE":
|
||||
self.first_error = error
|
||||
|
||||
def emit(self) -> int:
|
||||
for name, value in self.values:
|
||||
print(f"{name}={value}")
|
||||
print(f"MINIMAL_VERIFIER_ERROR={self.first_error}")
|
||||
return 1 if self.failed else 0
|
||||
|
||||
|
||||
def emit_canonical_failure(error: str) -> int:
|
||||
report = Report()
|
||||
report.value("VERIFICATION_MODE", "CANONICAL")
|
||||
report.value("GAMDisplay_OBJECT_COUNT", 0)
|
||||
report.value("Display_DATASOURCE_COUNT", 0)
|
||||
report.value("GAMDisplay_THREAD_TOKEN_COUNT", 0)
|
||||
report.value("GAMTimer_OBJECT_COUNT", 0)
|
||||
report.value("SentinelProducer_OBJECT_COUNT", 0)
|
||||
report.value("IOGAM_Writer_OBJECT_COUNT", 0)
|
||||
report.value("FileWriter_OBJECT_COUNT", 0)
|
||||
report.status("THREAD_FUNCTIONS_STATUS", False, error)
|
||||
report.status("BINARY_PATH_STRUCTURE_STATUS", False, error)
|
||||
report.status("CANONICAL_STRUCTURAL_COMPLETENESS_STATUS", False, error)
|
||||
report.status("PROFILE_CONFIG_STATUS", False, error)
|
||||
return report.emit()
|
||||
|
||||
|
||||
def verify_canonical(canonical_path: Path, minimal_path: Path) -> int:
|
||||
try:
|
||||
canonical_bytes, canonical = read_utf8(canonical_path)
|
||||
minimal_bytes, minimal = read_utf8(minimal_path)
|
||||
except (OSError, UnicodeError) as exc:
|
||||
return emit_canonical_failure(f"INPUT_READ_FAILED_{type(exc).__name__}")
|
||||
|
||||
canonical_hash = sha256_bytes(canonical_bytes)
|
||||
try:
|
||||
expected_minimal = derive_expected_minimal(canonical)
|
||||
except ValueError as exc:
|
||||
return emit_canonical_failure(str(exc))
|
||||
|
||||
expected_bytes = expected_minimal.encode("utf-8")
|
||||
complete = minimal_bytes == expected_bytes
|
||||
report = Report()
|
||||
report.value("VERIFICATION_MODE", "CANONICAL")
|
||||
report.value("CANONICAL_TEST_SHA256", canonical_hash)
|
||||
report.value("EXPECTED_CANONICAL_TEST_SHA256", EXPECTED_TEST_SHA256)
|
||||
report.value("MINIMAL_CONFIG_SHA256", sha256_bytes(minimal_bytes))
|
||||
report.value("EXPECTED_MINIMAL_CONFIG_SHA256", sha256_bytes(expected_bytes))
|
||||
|
||||
counts = {
|
||||
"GAMDisplay_OBJECT_COUNT": object_count(minimal, "GAMDisplay"),
|
||||
"Display_DATASOURCE_COUNT": object_count(minimal, "Display"),
|
||||
"GAMDisplay_THREAD_TOKEN_COUNT": len(
|
||||
re.findall(r"\bGAMDisplay\b", minimal)
|
||||
),
|
||||
"GAMTimer_OBJECT_COUNT": object_count(minimal, "GAMTimer"),
|
||||
"SentinelProducer_OBJECT_COUNT": object_count(
|
||||
minimal, "SentinelProducer"
|
||||
),
|
||||
"IOGAM_Writer_OBJECT_COUNT": object_count(minimal, "IOGAM_Writer"),
|
||||
"FileWriter_OBJECT_COUNT": object_count(minimal, "FileWriter"),
|
||||
}
|
||||
for name, value in counts.items():
|
||||
report.value(name, value)
|
||||
|
||||
expected_counts = {
|
||||
"GAMDisplay_OBJECT_COUNT": 0,
|
||||
"Display_DATASOURCE_COUNT": 0,
|
||||
"GAMDisplay_THREAD_TOKEN_COUNT": 0,
|
||||
"GAMTimer_OBJECT_COUNT": 1,
|
||||
"SentinelProducer_OBJECT_COUNT": 1,
|
||||
"IOGAM_Writer_OBJECT_COUNT": 1,
|
||||
"FileWriter_OBJECT_COUNT": 1,
|
||||
}
|
||||
counts_pass = counts == expected_counts
|
||||
thread_pass = minimal.count(MINIMAL_FUNCTIONS) == 1
|
||||
binary_pass = (
|
||||
counts["IOGAM_Writer_OBJECT_COUNT"] == 1
|
||||
and counts["FileWriter_OBJECT_COUNT"] == 1
|
||||
and minimal.count('Filename = "__MARTE_RUN_DIR__/s03_output.bin"') == 1
|
||||
and minimal.count('FileFormat = "binary"') == 1
|
||||
and minimal.count("DataSource = FileWriter") == 6
|
||||
)
|
||||
parent_pass = canonical_hash == EXPECTED_TEST_SHA256
|
||||
|
||||
report.value("UNAUTHORIZED_ADDITION_COUNT", 0 if complete else 1)
|
||||
report.status(
|
||||
"CANONICAL_TEST_IDENTITY_STATUS",
|
||||
parent_pass,
|
||||
"CANONICAL_TEST_SHA256_MISMATCH",
|
||||
)
|
||||
report.status(
|
||||
"AUTHORIZED_ABSENCE_AND_PRESENCE_STATUS",
|
||||
counts_pass,
|
||||
"OBJECT_OR_TOKEN_COUNT_MISMATCH",
|
||||
)
|
||||
report.status(
|
||||
"THREAD_FUNCTIONS_STATUS", thread_pass, "THREAD_FUNCTIONS_MISMATCH"
|
||||
)
|
||||
report.status(
|
||||
"BINARY_PATH_STRUCTURE_STATUS",
|
||||
binary_pass,
|
||||
"BINARY_PATH_STRUCTURE_MISMATCH",
|
||||
)
|
||||
report.status(
|
||||
"CANONICAL_STRUCTURAL_COMPLETENESS_STATUS",
|
||||
parent_pass and complete,
|
||||
"UNAUTHORIZED_STRUCTURAL_DIFFERENCE",
|
||||
)
|
||||
report.status(
|
||||
"PROFILE_CONFIG_STATUS",
|
||||
parent_pass and complete and counts_pass and thread_pass and binary_pass,
|
||||
"MINIMAL_PROFILE_CONFIG_NOT_PROVEN",
|
||||
)
|
||||
return report.emit()
|
||||
|
||||
|
||||
def extract_runtime_filename(text: str) -> list[str]:
|
||||
return re.findall(r'(?m)^[ \t]*Filename[ \t]*=[ \t]*"([^"]+)"[ \t]*$', text)
|
||||
|
||||
|
||||
def verify_runtime(packaged_path: Path, runtime_path: Path, run_dir: Path) -> int:
|
||||
report = Report()
|
||||
report.value("VERIFICATION_MODE", "RUNTIME")
|
||||
try:
|
||||
packaged_bytes, packaged = read_utf8(packaged_path)
|
||||
runtime_bytes, runtime = read_utf8(runtime_path)
|
||||
except (OSError, UnicodeError) as exc:
|
||||
report.status(
|
||||
"MINIMAL_RUNTIME_CONFIG_STATUS",
|
||||
False,
|
||||
f"INPUT_READ_FAILED_{type(exc).__name__}",
|
||||
)
|
||||
return report.emit()
|
||||
|
||||
supplied = str(run_dir)
|
||||
absolute = run_dir.is_absolute()
|
||||
exists = run_dir.is_dir()
|
||||
placeholder_count = packaged.count(PLACEHOLDER)
|
||||
unresolved_count = runtime.count(PLACEHOLDER)
|
||||
expected_runtime = packaged.replace(PLACEHOLDER, supplied)
|
||||
exact = runtime_bytes == expected_runtime.encode("utf-8")
|
||||
filenames = extract_runtime_filename(runtime)
|
||||
expected_filename = os.path.join(supplied, "s03_output.bin")
|
||||
filename_pass = filenames.count(expected_filename) == 1 and len(filenames) == 1
|
||||
|
||||
report.value("PACKAGED_MINIMAL_SHA256", sha256_bytes(packaged_bytes))
|
||||
report.value("RUNTIME_CONFIG_SHA256", sha256_bytes(runtime_bytes))
|
||||
report.value("PACKAGED_PLACEHOLDER_COUNT", placeholder_count)
|
||||
report.value("RUNTIME_UNRESOLVED_PLACEHOLDER_COUNT", unresolved_count)
|
||||
report.value("RUNTIME_FILENAME", filenames[0] if len(filenames) == 1 else "AMBIGUOUS")
|
||||
report.value("EXPECTED_RUNTIME_FILENAME", expected_filename)
|
||||
report.status("RUN_DIR_ABSOLUTE_STATUS", absolute, "RUN_DIR_NOT_ABSOLUTE")
|
||||
report.status("RUN_DIR_EXISTS_STATUS", exists, "RUN_DIR_NOT_FOUND")
|
||||
report.status(
|
||||
"PACKAGED_PLACEHOLDER_STATUS",
|
||||
placeholder_count == 1,
|
||||
"PACKAGED_PLACEHOLDER_COUNT_MISMATCH",
|
||||
)
|
||||
report.status(
|
||||
"RUNTIME_PLACEHOLDER_RESOLUTION_STATUS",
|
||||
unresolved_count == 0,
|
||||
"RUNTIME_PLACEHOLDER_UNRESOLVED",
|
||||
)
|
||||
report.status(
|
||||
"RUNTIME_FILENAME_STATUS", filename_pass, "RUNTIME_FILENAME_OUTSIDE_RUN_DIR"
|
||||
)
|
||||
report.status(
|
||||
"MINIMAL_RUNTIME_CONFIG_STATUS",
|
||||
absolute
|
||||
and exists
|
||||
and placeholder_count == 1
|
||||
and unresolved_count == 0
|
||||
and filename_pass
|
||||
and exact,
|
||||
"RUNTIME_DIFF_EXCEEDS_AUTHORIZED_SUBSTITUTION",
|
||||
)
|
||||
return report.emit()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers(dest="mode", required=True)
|
||||
|
||||
canonical = subparsers.add_parser("CANONICAL")
|
||||
canonical.add_argument("canonical_test", type=Path)
|
||||
canonical.add_argument("packaged_minimal", type=Path)
|
||||
|
||||
runtime = subparsers.add_parser("RUNTIME")
|
||||
runtime.add_argument("packaged_minimal", type=Path)
|
||||
runtime.add_argument("runtime_configuration", type=Path)
|
||||
runtime.add_argument("run_dir", type=Path)
|
||||
|
||||
args = parser.parse_args()
|
||||
if args.mode == "CANONICAL":
|
||||
return verify_canonical(args.canonical_test, args.packaged_minimal)
|
||||
return verify_runtime(
|
||||
args.packaged_minimal, args.runtime_configuration, args.run_dir
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user