"""Strict proof and revalidation for one torn-tail repair target.""" from __future__ import annotations import hashlib import json import os import stat from dataclasses import dataclass, replace from typing import Final from checkin_cli.weekly_operations import CustomerIdentityDigest, WeeklyOperationsCorruption, customer_storage_digest from checkin_cli.weekly_operations_history import HistoryValidation, validate_history_bytes from checkin_cli.weekly_operations_layout import customer_data_name, customer_lock_name REPAIR_SCHEMA: Final = "nutricoach-weekly-operations-repair-binding-v1" _FILE_FLAGS: Final = os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC | os.O_NONBLOCK @dataclass(frozen=True, slots=True) class WeeklyOperationsRepairBinding: schema_version: str customer_identity_digest: CustomerIdentityDigest customer_storage_digest: str data_device: int data_inode: int data_mode: int data_owner: int data_links: int lock_device: int lock_inode: int lock_mode: int lock_owner: int lock_links: int lock_size: int lock_digest: str valid_prefix_offset: int valid_prefix_digest: str torn_tail_length: int torn_tail_digest: str file_size: int file_digest: str binding_digest: str @property def data_identity(self) -> tuple[int, int]: return self.data_device, self.data_inode @property def lock_identity(self) -> tuple[int, int]: return self.lock_device, self.lock_inode def _binding_payload(binding: WeeklyOperationsRepairBinding) -> bytes: value = {field: getattr(binding, field) for field in binding.__dataclass_fields__ if field != "binding_digest"} return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() def _safe_file(info: os.stat_result, *, mode: int, owner: int, links: int) -> bool: return stat.S_ISREG(info.st_mode) and stat.S_IMODE(info.st_mode) == mode and info.st_uid == owner and info.st_nlink == links def validate_repair_binding(binding: WeeklyOperationsRepairBinding) -> None: expected = hashlib.sha256(_binding_payload(binding)).hexdigest() safe = binding.data_mode == binding.lock_mode == 0o600 and binding.data_owner == binding.lock_owner == os.geteuid() and binding.data_links == binding.lock_links == 1 if binding.schema_version != REPAIR_SCHEMA or binding.binding_digest != expected or binding.customer_storage_digest != customer_storage_digest(binding.customer_identity_digest) or binding.valid_prefix_offset <= 0 or binding.torn_tail_length <= 0 or binding.file_size != binding.valid_prefix_offset + binding.torn_tail_length or not safe: raise WeeklyOperationsCorruption("repair binding is invalid") def _read_existing(customers: int, name: str) -> tuple[os.stat_result, bytes]: try: descriptor = os.open(name, _FILE_FLAGS, dir_fd=customers) except OSError as error: raise WeeklyOperationsCorruption("repair data file is absent or unsafe") from error try: info = os.fstat(descriptor) named = os.stat(name, dir_fd=customers, follow_symlinks=False) if not _safe_file(info, mode=0o600, owner=os.geteuid(), links=1) or (info.st_dev, info.st_ino) != (named.st_dev, named.st_ino): raise WeeklyOperationsCorruption("repair data file identity is unsafe") chunks: list[bytes] = [] while chunk := os.read(descriptor, 65536): chunks.append(chunk) return info, b"".join(chunks) finally: os.close(descriptor) def _read_lock(customers: int, name: str) -> tuple[os.stat_result, bytes]: try: descriptor = os.open(name, _FILE_FLAGS, dir_fd=customers) except OSError as error: raise WeeklyOperationsCorruption("repair requires an existing safe lock") from error try: info = os.fstat(descriptor) named = os.stat(name, dir_fd=customers, follow_symlinks=False) if not _safe_file(info, mode=0o600, owner=os.geteuid(), links=1) or (info.st_dev, info.st_ino) != (named.st_dev, named.st_ino): raise WeeklyOperationsCorruption("repair lock identity is unsafe") chunks: list[bytes] = [] while chunk := os.read(descriptor, 65536): chunks.append(chunk) return info, b"".join(chunks) finally: os.close(descriptor) def inspect_repair_target(customers: int, customer_identity: CustomerIdentityDigest) -> WeeklyOperationsRepairBinding: """Issue a binding only for one positive, unambiguous torn tail.""" storage_digest = customer_storage_digest(customer_identity) data_info, payload = _read_existing(customers, customer_data_name(customer_identity)) validated = validate_history_bytes(payload, storage_digest, expected_customer=customer_identity, allow_torn_tail=True) if validated.complete_end <= 0 or validated.complete_end >= validated.size: raise WeeklyOperationsCorruption("repair target is not one positive torn tail") lock_info, lock_payload = _read_lock(customers, customer_lock_name(customer_identity)) prefix = payload[: validated.complete_end] tail = payload[validated.complete_end :] provisional = WeeklyOperationsRepairBinding( REPAIR_SCHEMA, customer_identity, storage_digest, data_info.st_dev, data_info.st_ino, stat.S_IMODE(data_info.st_mode), data_info.st_uid, data_info.st_nlink, lock_info.st_dev, lock_info.st_ino, stat.S_IMODE(lock_info.st_mode), lock_info.st_uid, lock_info.st_nlink, len(lock_payload), hashlib.sha256(lock_payload).hexdigest(), validated.complete_end, hashlib.sha256(prefix).hexdigest(), len(tail), hashlib.sha256(tail).hexdigest(), len(payload), hashlib.sha256(payload).hexdigest(), "", ) return replace(provisional, binding_digest=hashlib.sha256(_binding_payload(provisional)).hexdigest()) def validate_bound_repair_payload(binding: WeeklyOperationsRepairBinding, info: os.stat_result, payload: bytes) -> HistoryValidation: """Revalidate exact file identity, content, prefix, and tail under lock.""" validate_repair_binding(binding) if not _safe_file(info, mode=binding.data_mode, owner=binding.data_owner, links=binding.data_links) or (info.st_dev, info.st_ino) != binding.data_identity or len(payload) != binding.file_size or hashlib.sha256(payload).hexdigest() != binding.file_digest: raise WeeklyOperationsCorruption("repair target changed after issuance") validated = validate_history_bytes(payload, binding.customer_storage_digest, expected_customer=binding.customer_identity_digest, allow_torn_tail=True) prefix = payload[: validated.complete_end] tail = payload[validated.complete_end :] if validated.complete_end != binding.valid_prefix_offset or len(tail) != binding.torn_tail_length or hashlib.sha256(prefix).hexdigest() != binding.valid_prefix_digest or hashlib.sha256(tail).hexdigest() != binding.torn_tail_digest: raise WeeklyOperationsCorruption("repair offsets or torn tail changed after issuance") return validated def validate_bound_repair_lock(binding: WeeklyOperationsRepairBinding, info: os.stat_result, payload: bytes) -> None: validate_repair_binding(binding) if not _safe_file(info, mode=binding.lock_mode, owner=binding.lock_owner, links=binding.lock_links) or (info.st_dev, info.st_ino) != binding.lock_identity or len(payload) != binding.lock_size or hashlib.sha256(payload).hexdigest() != binding.lock_digest: raise WeeklyOperationsCorruption("repair lock changed after issuance")