"""Private, append-only corroboration of confirmed Telegram customer sends.""" from __future__ import annotations import base64 import fcntl import hashlib import json import os import re import stat from collections.abc import Iterator, Mapping from contextlib import contextmanager from dataclasses import asdict, dataclass from pathlib import Path from typing import Final _SCHEMA: Final = "telegram-customer-surface-receipt-v2" _DOMAIN: Final = "telegram-customer-surface-transport-receipt-v2" _REDACTION_ALGORITHM: Final = "unicode-non-whitespace-block" _REDACTION_VERSION: Final = 1 _RELATIVE_PATH: Final = Path("data/owner-actions/customer-surface-receipts-v2") _DIGEST: Final = re.compile(r"^[a-f0-9]{64}$") _PRIVATE_MODE: Final = 0o600 _DIRECTORY_MODE: Final = 0o700 _NOFOLLOW: Final = getattr(os, "O_NOFOLLOW", 0) def _canonical(value: Mapping[str, object]) -> bytes: return json.dumps( dict(value), ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode("utf-8") def redact_customer_surface_bytes(text: str) -> bytes: """Redact every non-whitespace scalar while preserving only layout shape.""" if not isinstance(text, str) or not text: raise ValueError("customer surface text is invalid") return "".join(character if character.isspace() else "█" for character in text).encode( "utf-8" ) def customer_route_digest(user_id: str, chat_id: str, topic_id: str) -> str: route = {"user_id": user_id, "chat_id": chat_id, "topic_id": topic_id} if any(not isinstance(value, str) or not value for value in route.values()): raise ValueError("customer surface route is invalid") return hashlib.sha256( _canonical({"domain": "telegram-customer-route-v1", **route}) ).hexdigest() @dataclass(frozen=True, slots=True) class CustomerSurfaceReceiptBinding: delivery_claim_id: str idempotency_key: str customer_route_digest: str provider_message_id: str approved_canonical_payload_digest: str candidate_digest: str customer_key: str draft_id: str approved_generation: int approved_generation_record_digest: str terminal_generation: int terminal_generation_record_digest: str approval_event_id: str approved_revision: str checkin_event_id: str checkin_revision: str @dataclass(frozen=True, slots=True) class CustomerSurfaceReceipt: schema: str authority: str delivery_claim_id: str idempotency_key: str customer_route_digest: str provider_message_id: str approved_canonical_payload_digest: str utf16_length: int redaction_algorithm: str redaction_version: int redacted_bytes_base64: str redacted_bytes_digest: str candidate_digest: str customer_key: str draft_id: str approved_generation: int approved_generation_record_digest: str terminal_generation: int terminal_generation_record_digest: str approval_event_id: str approved_revision: str checkin_event_id: str checkin_revision: str transport_receipt_digest: str class CustomerSurfaceReceiptStore: """Append one private corroborating receipt for each consumed delivery claim.""" def __init__(self, profile_root: Path | str) -> None: self.state_dir = Path(profile_root) / _RELATIVE_PATH self.ledger_path = self.state_dir / "receipts.jsonl" self.lock_path = self.state_dir / ".lock" self._prepare_paths() def append( self, binding: CustomerSurfaceReceiptBinding, customer_text: str, ) -> CustomerSurfaceReceipt: candidate = self._candidate(binding, customer_text) with self._locked(): records = self._read_unlocked() matches = [ record for record in records if record.delivery_claim_id == candidate.delivery_claim_id or record.idempotency_key == candidate.idempotency_key ] if matches: if len(matches) != 1 or matches[0] != candidate: raise ValueError("customer surface receipt conflicts with immutable evidence") return matches[0] descriptor = self._open_private( self.ledger_path, os.O_WRONLY | os.O_APPEND | os.O_CLOEXEC, "customer surface receipt ledger", ) try: payload = _canonical(asdict(candidate)) + b"\n" written = 0 while written < len(payload): count = os.write(descriptor, payload[written:]) if count <= 0: raise OSError("customer surface receipt append failed") written += count os.fsync(descriptor) finally: os.close(descriptor) return candidate def records(self) -> tuple[CustomerSurfaceReceipt, ...]: with self._locked(): return self._read_unlocked() @classmethod def _candidate( cls, binding: CustomerSurfaceReceiptBinding, customer_text: str, ) -> CustomerSurfaceReceipt: cls._validate_binding(binding) redacted = redact_customer_surface_bytes(customer_text) redacted_digest = hashlib.sha256(redacted).hexdigest() utf16_length = len(customer_text.encode("utf-16-le")) // 2 transport_values: dict[str, object] = { "domain": _DOMAIN, "delivery_claim_id": binding.delivery_claim_id, "idempotency_key": binding.idempotency_key, "customer_route_digest": binding.customer_route_digest, "provider_message_id": binding.provider_message_id, "approved_canonical_payload_digest": binding.approved_canonical_payload_digest, "utf16_length": utf16_length, "redaction_algorithm": _REDACTION_ALGORITHM, "redaction_version": _REDACTION_VERSION, "redacted_bytes_digest": redacted_digest, "candidate_digest": binding.candidate_digest, "customer_key": binding.customer_key, "draft_id": binding.draft_id, "approved_generation": binding.approved_generation, "approved_generation_record_digest": binding.approved_generation_record_digest, "terminal_generation": binding.terminal_generation, "terminal_generation_record_digest": binding.terminal_generation_record_digest, "approval_event_id": binding.approval_event_id, "approved_revision": binding.approved_revision, "checkin_event_id": binding.checkin_event_id, "checkin_revision": binding.checkin_revision, } return CustomerSurfaceReceipt( _SCHEMA, "corroboration_only", binding.delivery_claim_id, binding.idempotency_key, binding.customer_route_digest, binding.provider_message_id, binding.approved_canonical_payload_digest, utf16_length, _REDACTION_ALGORITHM, _REDACTION_VERSION, base64.b64encode(redacted).decode("ascii"), redacted_digest, binding.candidate_digest, binding.customer_key, binding.draft_id, binding.approved_generation, binding.approved_generation_record_digest, binding.terminal_generation, binding.terminal_generation_record_digest, binding.approval_event_id, binding.approved_revision, binding.checkin_event_id, binding.checkin_revision, hashlib.sha256(_canonical(transport_values)).hexdigest(), ) @staticmethod def _validate_binding(binding: CustomerSurfaceReceiptBinding) -> None: if not isinstance(binding, CustomerSurfaceReceiptBinding): raise ValueError("customer surface receipt binding is invalid") digest_fields = ( binding.delivery_claim_id, binding.idempotency_key, binding.customer_route_digest, binding.approved_canonical_payload_digest, binding.candidate_digest, binding.approved_generation_record_digest, binding.terminal_generation_record_digest, binding.approved_revision, binding.checkin_revision, ) identifiers = ( binding.provider_message_id, binding.customer_key, binding.draft_id, binding.approval_event_id, binding.checkin_event_id, ) if ( any(_DIGEST.fullmatch(value) is None for value in digest_fields) or any(not isinstance(value, str) or not value or len(value) > 256 for value in identifiers) or type(binding.approved_generation) is not int or type(binding.terminal_generation) is not int or binding.approved_generation < 1 or binding.terminal_generation <= binding.approved_generation or binding.approved_canonical_payload_digest != binding.approved_revision ): raise ValueError("customer surface receipt binding is invalid") def _prepare_paths(self) -> None: self._validate_ancestry() try: self.state_dir.mkdir(parents=True, mode=_DIRECTORY_MODE) except FileExistsError: pass self._validate_ancestry() info = os.lstat(self.state_dir) if not stat.S_ISDIR(info.st_mode) or stat.S_IMODE(info.st_mode) != _DIRECTORY_MODE: raise ValueError("customer surface receipt directory is invalid") self._ensure_private_file(self.lock_path) self._ensure_private_file(self.ledger_path) self._read_unlocked() def _validate_ancestry(self) -> None: for path in (*reversed(self.state_dir.parents), self.state_dir): try: info = os.lstat(path) except FileNotFoundError: continue if stat.S_ISLNK(info.st_mode): raise ValueError("customer surface receipt path must not contain symlinks") def _ensure_private_file(self, path: Path) -> None: try: descriptor = os.open( path, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | _NOFOLLOW, _PRIVATE_MODE, ) except FileExistsError: descriptor = self._open_private( path, os.O_RDWR | os.O_CLOEXEC, "customer surface receipt authority", ) os.close(descriptor) return except OSError as exc: raise ValueError("customer surface receipt authority is unavailable") from exc try: self._validate_descriptor(descriptor, "customer surface receipt authority") os.fsync(descriptor) finally: os.close(descriptor) @contextmanager def _locked(self) -> Iterator[None]: descriptor = self._open_private( self.lock_path, os.O_RDWR | os.O_CLOEXEC, "customer surface receipt lock", ) try: fcntl.flock(descriptor, fcntl.LOCK_EX) yield finally: os.close(descriptor) def _read_unlocked(self) -> tuple[CustomerSurfaceReceipt, ...]: descriptor = self._open_private( self.ledger_path, os.O_RDONLY | os.O_CLOEXEC, "customer surface receipt ledger", ) try: with os.fdopen(descriptor, "r", encoding="utf-8", closefd=False) as handle: lines = tuple(handle) finally: os.close(descriptor) records = tuple(self._parse(line) for line in lines) claims = {record.delivery_claim_id for record in records} keys = {record.idempotency_key for record in records} if len(claims) != len(records) or len(keys) != len(records): raise ValueError("customer surface receipt ledger is ambiguous") return records @classmethod def _parse(cls, line: str) -> CustomerSurfaceReceipt: try: value = json.loads(line) if not isinstance(value, dict) or set(value) != set(CustomerSurfaceReceipt.__slots__): raise ValueError receipt = CustomerSurfaceReceipt(**value) redacted = base64.b64decode(receipt.redacted_bytes_base64, validate=True) binding = CustomerSurfaceReceiptBinding( receipt.delivery_claim_id, receipt.idempotency_key, receipt.customer_route_digest, receipt.provider_message_id, receipt.approved_canonical_payload_digest, receipt.candidate_digest, receipt.customer_key, receipt.draft_id, receipt.approved_generation, receipt.approved_generation_record_digest, receipt.terminal_generation, receipt.terminal_generation_record_digest, receipt.approval_event_id, receipt.approved_revision, receipt.checkin_event_id, receipt.checkin_revision, ) cls._validate_binding(binding) if ( receipt.schema != _SCHEMA or receipt.authority != "corroboration_only" or receipt.redaction_algorithm != _REDACTION_ALGORITHM or receipt.redaction_version != _REDACTION_VERSION or hashlib.sha256(redacted).hexdigest() != receipt.redacted_bytes_digest ): raise ValueError transport_values: dict[str, object] = { "domain": _DOMAIN, "delivery_claim_id": receipt.delivery_claim_id, "idempotency_key": receipt.idempotency_key, "customer_route_digest": receipt.customer_route_digest, "provider_message_id": receipt.provider_message_id, "approved_canonical_payload_digest": receipt.approved_canonical_payload_digest, "utf16_length": receipt.utf16_length, "redaction_algorithm": receipt.redaction_algorithm, "redaction_version": receipt.redaction_version, "redacted_bytes_digest": receipt.redacted_bytes_digest, "candidate_digest": receipt.candidate_digest, "customer_key": receipt.customer_key, "draft_id": receipt.draft_id, "approved_generation": receipt.approved_generation, "approved_generation_record_digest": receipt.approved_generation_record_digest, "terminal_generation": receipt.terminal_generation, "terminal_generation_record_digest": receipt.terminal_generation_record_digest, "approval_event_id": receipt.approval_event_id, "approved_revision": receipt.approved_revision, "checkin_event_id": receipt.checkin_event_id, "checkin_revision": receipt.checkin_revision, } if hashlib.sha256(_canonical(transport_values)).hexdigest() != receipt.transport_receipt_digest: raise ValueError return receipt except (TypeError, ValueError, json.JSONDecodeError) as exc: raise ValueError("customer surface receipt ledger is invalid") from exc @classmethod def _open_private(cls, path: Path, flags: int, label: str) -> int: try: descriptor = os.open(path, flags | _NOFOLLOW) except OSError as exc: raise ValueError(f"{label} is unavailable") from exc try: cls._validate_descriptor(descriptor, label) except BaseException: os.close(descriptor) raise return descriptor @staticmethod def _validate_descriptor(descriptor: int, label: str) -> None: info = os.fstat(descriptor) if ( not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) != _PRIVATE_MODE ): raise ValueError(f"{label} is invalid")