File size: 14,012 Bytes
ecad52f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 | """Regression checks for the portable source release; standard library only."""
import copy
import hashlib
import json
import os
import random
import socket
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
from demo import evaluate, evaluate_json, example
from matrix_core import (
CHECKSUM_FORMAT, ENGINE_VERSION, MAX_PAYLOAD_BYTES, WHITESPACE_CODEPOINTS,
DeterministicMatrixCore, checksum_for_date, normalize_payload, parse_instant, verify_result,
)
from safe_json import strict_loads
from verify_release import verify_release
ROOT = Path(__file__).resolve().parent.parent
FIXED = datetime(2026, 11, 3, 12, tzinfo=timezone.utc)
class LongevityTests(unittest.TestCase):
def setUp(self):
self.core = DeterministicMatrixCore()
def test_published_vectors_and_utc_dates(self):
vectors = json.loads((ROOT / "test_vectors.json").read_text(encoding="utf-8"))
for vector in vectors["vectors"]:
with self.subTest(id=vector["id"]):
result = self.core.enforce_zero_tolerance_matrix(vector["payload"], parse_instant(vector["at"]))
self.assertEqual(result["input_payload"], vector["normalized"])
self.assertEqual(result["seal_date"], vector["utc_date"])
self.assertEqual(result["matrix_verification_seal"], vector["sha256"])
def test_frozen_whitespace_table(self):
self.assertEqual(len(WHITESPACE_CODEPOINTS), 29)
for codepoint in WHITESPACE_CODEPOINTS:
self.assertEqual(normalize_payload(chr(codepoint) + "a" + chr(codepoint) * 3 + "b" + chr(codepoint)), "a b")
def test_non_whitespace_characters_are_preserved(self):
for codepoint in (0, 0x180E, 0x200B, 0x2060, 0xFEFF, 0x1F600):
payload = chr(codepoint) + "text" + chr(codepoint)
self.assertEqual(normalize_payload(payload), payload)
def test_no_implicit_unicode_normalization(self):
self.assertNotEqual(checksum_for_date("é", "2126-11-03"), checksum_for_date("e\u0301", "2126-11-03"))
def test_normalization_idempotence_for_seeded_unicode_inputs(self):
rng = random.Random(36500)
alphabet = ["a", "Č", "😀", "\u200b", "é"] + [chr(c) for c in WHITESPACE_CODEPOINTS]
for _ in range(100):
payload = "".join(rng.choice(alphabet) for _ in range(100))
clean = normalize_payload(payload)
self.assertEqual(normalize_payload(clean), clean)
if clean:
self.assertEqual(checksum_for_date(payload, "2126-11-03"), checksum_for_date(clean, "2126-11-03"))
def test_lone_surrogates_have_no_checksum(self):
for payload in ("\ud800", "x\udfff", "\ud800\udc00"):
result = self.core.enforce_zero_tolerance_matrix(payload, FIXED)
self.assertFalse(result["valid"])
self.assertEqual(result["error"], "INVALID_UNICODE")
self.assertNotIn("matrix_verification_seal", result)
def test_raw_utf8_size_limit_before_normalization(self):
for payload in (" " * (MAX_PAYLOAD_BYTES + 1), "😀" * (MAX_PAYLOAD_BYTES // 4 + 1)):
self.assertEqual(self.core.enforce_zero_tolerance_matrix(payload, FIXED)["error"], "PAYLOAD_TOO_LARGE")
self.assertTrue(self.core.enforce_zero_tolerance_matrix("a" * MAX_PAYLOAD_BYTES, FIXED)["valid"])
def test_future_gregorian_calendar_and_leap_rules(self):
for at in ("2126-11-03T12:00:00Z", "2400-02-29T12:00:00Z", "0001-01-01T00:00:00Z", "9999-12-31T23:59:59Z"):
self.assertTrue(self.core.enforce_zero_tolerance_matrix("x", parse_instant(at))["valid"])
for at in ("2100-02-29T12:00:00Z", "2200-02-29T12:00:00Z", "0000-01-01T00:00:00Z", "2026-02-30T12:00:00Z"):
with self.assertRaises(ValueError):
parse_instant(at)
def test_timestamp_grammar_is_explicit_and_ascii(self):
for at in ("2026-11-03", "2026-11-03 12:00:00Z", "20261103T120000Z", "2026-11-03T12:00:00-00:00", "2026-11-03T12:00:00+00:60", "2026-11-03T12:00:00+24:00", "2026-11-03T12:00:60Z", "2026-11-03T12:00:00.1234567Z"):
with self.subTest(at=at), self.assertRaises(ValueError):
parse_instant(at)
def test_utc_overflow_is_controlled(self):
for at in ("0001-01-01T00:00:00+01:00", "9999-12-31T23:59:59-01:00"):
with self.assertRaises(ValueError):
parse_instant(at)
def test_valid_export_roundtrips(self):
result = self.core.enforce_zero_tolerance_matrix("Č 😀", FIXED)
self.assertTrue(verify_result(strict_loads(json.dumps(result)))["verified"])
self.assertFalse(verify_result(result)["metadata_authenticated"])
def test_legacy_351_export_remains_readable(self):
legacy = self.core.enforce_zero_tolerance_matrix("Evidence before confidence.", FIXED)
for key in ("checksum_format", "normalization_profile", "hash_algorithm"):
legacy.pop(key)
legacy["engine_version"] = "3.5.1-OMEGA"
self.assertTrue(verify_result(legacy)["verified"])
legacy["engine_version"] = "unknown"
self.assertFalse(verify_result(legacy)["verified"])
def test_export_mutations_fail_closed(self):
result = self.core.enforce_zero_tolerance_matrix("Evidence before confidence.", FIXED)
mutations = {"valid": 1, "input_payload": "other", "seal_date": "2026-11-04", "timestamp": "2026-11-04T12:00:00Z", "matrix_verification_seal": "0" * 64, "checksum_format": "bodul-matrix-v999", "normalization_profile": "unicode-latest", "hash_algorithm": "md5", "engine_version": ""}
for key, value in mutations.items():
item = dict(result)
item[key] = value
with self.subTest(key=key):
self.assertFalse(verify_result(item)["verified"])
item = dict(result, extra="ignored?")
self.assertFalse(verify_result(item)["verified"])
item = dict(result)
del item["hash_algorithm"]
self.assertFalse(verify_result(item)["verified"])
def test_export_requires_canonical_payload(self):
result = self.core.enforce_zero_tolerance_matrix("x", FIXED)
result["input_payload"] = " x "
self.assertEqual(verify_result(result)["error"], "NON_CANONICAL_PAYLOAD")
def test_metadata_is_not_authenticated(self):
result = self.core.enforce_zero_tolerance_matrix("x", FIXED)
result["engine_version"] = "different-but-unsealed-metadata"
result["timestamp"] = "2026-11-03T23:59:59Z"
self.assertTrue(verify_result(result)["verified"])
def test_duplicate_keys_and_nonfinite_numbers_are_rejected(self):
for text in ('{"a":1,"a":2}', '{"a":{"x":1,"x":2}}', '{"x":NaN}', '{"x":Infinity}', '{"x":1e9999}'):
with self.subTest(text=text), self.assertRaises(ValueError):
strict_loads(text)
def test_json_limits_and_surrogates(self):
for text in ('[' * 33 + '0' + ']' * 33, '"\\ud800"', '1' * 129):
with self.assertRaises(ValueError):
strict_loads(text)
self.assertEqual(strict_loads('"\\ud83d\\ude00"'), "😀")
self.assertEqual(strict_loads('{"text":"[[[{{{"}'), {"text": "[[[{{{"})
def test_malformed_public_inputs_never_approve(self):
for payload in (None, True, [], {}, {"claim": {}}, {"evidence": "wrong"}):
self.assertEqual(evaluate(payload)["verdict"], "BLOCK")
self.assertEqual(evaluate_json('{"claim":1,"claim":2}')["reason"], "invalid_json")
def test_public_boolean_types_are_not_integers(self):
payload = example(at=FIXED)
payload["evidence"][0]["supports"] = 1
self.assertEqual(evaluate(payload)["verdict"], "BLOCK")
def test_public_timestamps_are_frozen_and_utc_normalized(self):
payload = example(at=FIXED)
other = copy.deepcopy(payload)
other["evaluation_time"] = "2026-11-03T14:00:00+02:00"
other["evidence"][0]["observed_at"] = "2026-11-03T14:00:00+02:00"
self.assertEqual(evaluate(payload), evaluate(other))
payload["evaluation_time"] = "2026-11-03 12:00:00Z"
self.assertEqual(evaluate(payload)["verdict"], "BLOCK")
def test_public_approval_flags_cannot_authorize_high_risk_claims(self):
payload = example(at=FIXED)
payload["claim"]["risk"] = "high"
payload["human_approved"] = True
self.assertEqual(evaluate(payload)["verdict"], "ESCALATE")
self.assertEqual(evaluate(payload)["reason"], "human_approval_required")
def test_missing_stale_future_and_conflicting_evidence_block(self):
payload = example(at=FIXED)
self.assertEqual(evaluate(payload)["verdict"], "APPROVE")
missing = example("unsupported", at=FIXED)
self.assertEqual(evaluate(missing)["classification"], "UNKNOWN")
for seconds in (-301, 1):
item = copy.deepcopy(payload)
item["evidence"][0]["observed_at"] = (FIXED + timedelta(seconds=seconds)).isoformat()
self.assertEqual(evaluate(item)["verdict"], "BLOCK")
contradictory = dict(payload["evidence"][0], source_id="second", supports=False)
payload["evidence"].append(contradictory)
self.assertEqual(evaluate(payload)["classification"], "CONFLICT")
self.assertEqual(evaluate(payload)["verdict"], "BLOCK")
def test_source_duplicates_do_not_satisfy_independence_quorum(self):
payload = example(at=FIXED)
payload["claim"]["risk"] = "medium"
payload["evidence"].append(dict(payload["evidence"][0], evidence_id="copy"))
self.assertEqual(evaluate(payload)["verdict"], "ESCALATE")
def test_offline_execution_does_not_use_network(self):
with patch.object(socket, "socket", side_effect=AssertionError("Network use forbidden")):
self.assertTrue(self.core.enforce_zero_tolerance_matrix("x", FIXED)["valid"])
self.assertEqual(evaluate(example(at=FIXED))["verdict"], "APPROVE")
def test_explicit_future_evaluation_repeats(self):
payload = example(at=datetime(2126, 11, 3, tzinfo=timezone.utc))
expected = evaluate(payload)
self.assertEqual(expected["verdict"], "APPROVE")
for _ in range(25):
self.assertEqual(evaluate(payload), expected)
def test_offline_cli_matrix_and_saved_result(self):
with tempfile.TemporaryDirectory() as directory:
folder = Path(directory)
text = folder / "input.txt"
result_file = folder / "result.json"
text.write_text("Evidence before confidence.", encoding="utf-8")
run = subprocess.run([sys.executable, "-B", str(ROOT / "offline_cli.py"), "matrix", str(text), "--at", FIXED.isoformat()], capture_output=True, text=True, encoding="utf-8")
self.assertEqual(run.returncode, 0, run.stderr)
result_file.write_text(run.stdout, encoding="utf-8")
replay = subprocess.run([sys.executable, "-B", str(ROOT / "offline_cli.py"), "verify-matrix", str(result_file)], capture_output=True, text=True, encoding="utf-8")
self.assertEqual(replay.returncode, 0, replay.stderr)
self.assertTrue(json.loads(replay.stdout)["verified"])
def test_offline_cli_block_has_nonzero_exit(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "claim.json"
path.write_text(json.dumps(example("unsupported", at=FIXED)), encoding="utf-8")
run = subprocess.run([sys.executable, "-B", str(ROOT / "offline_cli.py"), "claims", str(path)], capture_output=True, text=True, encoding="utf-8")
self.assertEqual(run.returncode, 3)
self.assertEqual(json.loads(run.stdout)["verdict"], "BLOCK")
class ArchiveIntegrityTests(unittest.TestCase):
def make_archive(self, folder):
data = b"portable source\n"
(folder / "source.txt").write_bytes(data)
manifest = {"manifest_format": "bodul-source-archive-v1", "files": {"source.txt": {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest(), "sha512": hashlib.sha512(data).hexdigest()}}}
(folder / "RELEASE_MANIFEST.json").write_text(json.dumps(manifest), encoding="utf-8")
return manifest
def test_clean_archive_and_tampering(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
self.make_archive(root)
self.assertTrue(verify_release(root)["verified"])
(root / "source.txt").write_bytes(b"modified source\n")
self.assertFalse(verify_release(root)["verified"])
def test_missing_and_unlisted_files(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
self.make_archive(root)
(root / "extra.txt").write_text("x")
self.assertFalse(verify_release(root)["verified"])
(root / "extra.txt").unlink()
(root / "source.txt").unlink()
self.assertFalse(verify_release(root)["verified"])
def test_unknown_format_and_unsafe_paths(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
manifest = self.make_archive(root)
for path in ("../escape.txt", "/absolute.txt", "drive:path", "foo\\bar", "."):
bad = copy.deepcopy(manifest)
bad["files"][path] = bad["files"].pop("source.txt")
(root / "RELEASE_MANIFEST.json").write_text(json.dumps(bad), encoding="utf-8")
self.assertFalse(verify_release(root)["verified"])
manifest["manifest_format"] = "future-unknown-format"
(root / "RELEASE_MANIFEST.json").write_text(json.dumps(manifest), encoding="utf-8")
self.assertFalse(verify_release(root)["verified"])
if __name__ == "__main__":
unittest.main()
|