Release portable deterministic core 1.1.0: frozen formats, offline recovery and regression checks
ecad52f verified | #!/usr/bin/env python3 | |
| """BODULAI Matrix Core: portable v1 text checksums; Apache-2.0. | |
| Author: Damir Balenović (mirdax25-ux). | |
| The format and normalization table are frozen; see MATRIX_FORMAT.md. | |
| A checksum is not a signature, authorization, or evidence of truth. | |
| """ | |
| import hashlib | |
| import hmac | |
| import json | |
| import logging | |
| import re | |
| import sys | |
| from datetime import datetime, timezone | |
| from typing import Any, Dict, Optional | |
| ENGINE_VERSION = "3.6.0-OMEGA" | |
| CHECKSUM_FORMAT = "bodul-matrix-v1" | |
| NORMALIZATION_PROFILE = "bodul-whitespace-v1" | |
| HASH_ALGORITHM = "sha256" | |
| MAX_PAYLOAD_BYTES = 1_048_576 | |
| # Frozen compatibility table: Python 3.12 split() whitespace, 29 code points. | |
| # Do not replace with str.split(), isspace(), or regex \s: Unicode evolves. | |
| WHITESPACE_CODEPOINTS = ( | |
| 0x0009, 0x000A, 0x000B, 0x000C, 0x000D, | |
| 0x001C, 0x001D, 0x001E, 0x001F, 0x0020, 0x0085, 0x00A0, 0x1680, | |
| 0x2000, 0x2001, 0x2002, 0x2003, 0x2004, 0x2005, 0x2006, 0x2007, | |
| 0x2008, 0x2009, 0x200A, 0x2028, 0x2029, 0x202F, 0x205F, 0x3000, | |
| ) | |
| _SEPARATORS = re.compile("[" + "".join(chr(n) for n in WHITESPACE_CODEPOINTS) + "]+") | |
| _DATE = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}\Z") | |
| _HEX = re.compile(r"[0-9a-f]{64}\Z") | |
| _INSTANT = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]{1,6})?(?:Z|[+-][0-9]{2}:[0-9]{2})\Z") | |
| logger = logging.getLogger("BodulMatrixCore") | |
| logger.addHandler(logging.NullHandler()) | |
| def parse_instant(value: str) -> datetime: | |
| """Stable ASCII timestamp grammar with whole-minute offsets; return UTC.""" | |
| if not isinstance(value, str) or not _INSTANT.fullmatch(value): | |
| raise ValueError("Use YYYY-MM-DDTHH:MM:SS[.ffffff]Z or an explicit +/-HH:MM offset") | |
| # RFC 3339 -00:00 means unknown local offset, not confirmed UTC. | |
| if value.endswith("-00:00"): | |
| raise ValueError("Unknown UTC offset -00:00 is not accepted; use Z or +00:00") | |
| if not value.endswith("Z") and (int(value[-5:-3]) > 23 or int(value[-2:]) > 59): | |
| raise ValueError("UTC offset hours must be 00..23 and minutes 00..59") | |
| try: | |
| return datetime.fromisoformat(value[:-1] + "+00:00" if value.endswith("Z") else value).astimezone(timezone.utc) | |
| except (ValueError, OverflowError) as exc: | |
| raise ValueError("Timestamp is outside the supported Gregorian range 0001..9999") from exc | |
| def _date(value: Any) -> str: | |
| if not isinstance(value, str) or not _DATE.fullmatch(value): | |
| raise ValueError("Date must be YYYY-MM-DD") | |
| datetime(int(value[:4]), int(value[5:7]), int(value[8:10])) | |
| return value | |
| def _input_error(payload: Any) -> Optional[str]: | |
| if not isinstance(payload, str): | |
| return "EMPTY_OR_INVALID_PAYLOAD" | |
| if len(payload) > MAX_PAYLOAD_BYTES: | |
| return "PAYLOAD_TOO_LARGE" | |
| try: | |
| if len(payload.encode("utf-8", errors="strict")) > MAX_PAYLOAD_BYTES: | |
| return "PAYLOAD_TOO_LARGE" | |
| except UnicodeEncodeError: | |
| return "INVALID_UNICODE" | |
| return None | |
| def normalize_payload(payload: str) -> str: | |
| """Collapse the frozen whitespace table; no Unicode normalization/casefold.""" | |
| error = _input_error(payload) | |
| if error: | |
| raise ValueError(error) | |
| # trim only the explicit table; ordinary str.strip() is Unicode-dependent. | |
| return _SEPARATORS.sub(" ", payload).strip(" ") | |
| def checksum_for_date(payload: str, seal_date: str) -> str: | |
| """No clock or network: checksum of non-empty normalized text and UTC date.""" | |
| clean = normalize_payload(payload) | |
| if not clean: | |
| raise ValueError("EMPTY_OR_INVALID_PAYLOAD") | |
| raw = "BODUL_MATRIX::" + clean + "::" + _date(seal_date) + "::OMEGA_FLOW" | |
| return hashlib.sha256(raw.encode("utf-8", errors="strict")).hexdigest() | |
| class DeterministicMatrixCore: | |
| def __init__(self, engine_version: str = ENGINE_VERSION): | |
| if not isinstance(engine_version, str) or not engine_version.strip() or len(engine_version) > 128: | |
| raise ValueError("engine_version must be a non-empty string of at most 128 characters") | |
| self.engine_version = engine_version | |
| def _sanitize_payload(payload: Any) -> Optional[str]: | |
| if _input_error(payload): | |
| return None | |
| return normalize_payload(payload) or None | |
| def _utc_now(now: Optional[datetime]) -> datetime: | |
| if now is None: | |
| return datetime.now(timezone.utc) | |
| if not isinstance(now, datetime): | |
| raise TypeError("now must be a timezone-aware datetime or None") | |
| if now.tzinfo is None or now.utcoffset() is None: | |
| raise ValueError("now must include a UTC offset or timezone") | |
| try: | |
| return now.astimezone(timezone.utc) | |
| except (ValueError, OverflowError) as exc: | |
| raise ValueError("UTC instant is outside supported years 0001..9999") from exc | |
| def enforce_zero_tolerance_matrix(self, payload: Any, now: Optional[datetime] = None) -> Dict[str, Any]: | |
| instant = self._utc_now(now) | |
| metadata = { | |
| "timestamp": instant.isoformat(), "engine_version": self.engine_version, | |
| "checksum_format": CHECKSUM_FORMAT, "normalization_profile": NORMALIZATION_PROFILE, | |
| "hash_algorithm": HASH_ALGORITHM, | |
| } | |
| error = _input_error(payload) | |
| clean = None if error else normalize_payload(payload) | |
| if not clean: | |
| return {"valid": False, "error": error or "EMPTY_OR_INVALID_PAYLOAD", **metadata} | |
| seal_date = instant.date().isoformat() | |
| return { | |
| "valid": True, "input_payload": clean, "seal_date": seal_date, | |
| "matrix_verification_seal": checksum_for_date(clean, seal_date), **metadata, | |
| } | |
| def verify_result(result: Any) -> Dict[str, Any]: | |
| """Check checksum consistency of an exported result, never source identity. | |
| Reads original 3.5.1 records and explicit v1 records. Unknown formats and | |
| partially labeled records fail closed. Metadata is not authenticated. | |
| """ | |
| def reject(reason): | |
| return {"verified": False, "scope": "checksum_consistency", "error": reason} | |
| if not isinstance(result, dict) or result.get("valid") is not True: | |
| return reject("INVALID_RESULT") | |
| base = {"valid", "input_payload", "seal_date", "matrix_verification_seal", "timestamp", "engine_version"} | |
| identifiers = {"checksum_format", "normalization_profile", "hash_algorithm"} | |
| if set(result) == base: | |
| if result.get("engine_version") != "3.5.1-OMEGA": | |
| return reject("UNLABELED_FORMAT") | |
| elif set(result) == base | identifiers: | |
| if result.get("checksum_format") != CHECKSUM_FORMAT or result.get("normalization_profile") != NORMALIZATION_PROFILE or result.get("hash_algorithm") != HASH_ALGORITHM: | |
| return reject("UNSUPPORTED_FORMAT") | |
| else: | |
| return reject("UNEXPECTED_FIELDS") | |
| try: | |
| version = result["engine_version"] | |
| if not isinstance(version, str) or not version.strip() or len(version) > 128: | |
| return reject("INVALID_METADATA") | |
| payload = result["input_payload"] | |
| if not isinstance(payload, str) or not payload or normalize_payload(payload) != payload: | |
| return reject("NON_CANONICAL_PAYLOAD") | |
| instant = parse_instant(result["timestamp"]) | |
| if instant.date().isoformat() != _date(result["seal_date"]): | |
| return reject("DATE_MISMATCH") | |
| expected = checksum_for_date(payload, result["seal_date"]) | |
| provided = result["matrix_verification_seal"] | |
| if not isinstance(provided, str) or not _HEX.fullmatch(provided) or not hmac.compare_digest(expected, provided): | |
| return reject("CHECKSUM_MISMATCH") | |
| except (ValueError, TypeError, OverflowError): | |
| return reject("INVALID_RESULT") | |
| return {"verified": True, "scope": "checksum_consistency", "checksum_format": CHECKSUM_FORMAT, "metadata_authenticated": False} | |
| def main(argv=None) -> int: | |
| """JSON-only CLI. Exit 0 valid; 2 invalid input; 1 unexpected failure.""" | |
| import argparse | |
| for stream in (sys.stdout, sys.stderr): | |
| try: | |
| stream.reconfigure(encoding="utf-8") | |
| except (AttributeError, ValueError): | |
| pass | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| parser.add_argument("payload", nargs="*") | |
| parser.add_argument("--at", help="Explicit timezone required; see MATRIX_FORMAT.md") | |
| parser.add_argument("--verbose", action="store_true") | |
| args = parser.parse_args(argv) | |
| if args.verbose: | |
| logging.basicConfig(level=logging.INFO, stream=sys.stderr) | |
| try: | |
| instant = parse_instant(args.at) if args.at is not None else None | |
| result = DeterministicMatrixCore().enforce_zero_tolerance_matrix(" ".join(args.payload), instant) | |
| except (TypeError, ValueError) as exc: | |
| print(json.dumps({"valid": False, "error": "INVALID_TIMESTAMP", "detail": str(exc)}, ensure_ascii=True)) | |
| return 2 | |
| print(json.dumps(result, ensure_ascii=True, sort_keys=True, indent=2, allow_nan=False)) | |
| return 0 if result["valid"] else 2 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |