#!/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 @staticmethod def _sanitize_payload(payload: Any) -> Optional[str]: if _input_error(payload): return None return normalize_payload(payload) or None @staticmethod 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())