"""Structured file and JSON utilities shared across the TNFR engine.""" from __future__ import annotations import json import os import tempfile from dataclasses import dataclass from functools import partial from pathlib import Path from typing import Any, Callable from .._exact_time import finite_represented_real from ..errors import TNFRValueError from .init import LazyImportProxy, cached_import, get_logger logger = get_logger(__name__) @dataclass(frozen=True) class JsonDumpsParams: """Container describing the parameters used by :func:`json_dumps`.""" sort_keys: bool = False default: Callable[[Any], Any] | None = None ensure_ascii: bool = True separators: tuple[str, str] = (",", ":") cls: type[json.JSONEncoder] | None = None to_bytes: bool = False DEFAULT_PARAMS = JsonDumpsParams() def _reject_duplicate_json_keys(pairs: list[tuple[str, Any]]) -> None: """Reject colliding decoded names without retaining a second object tree.""" names: set[str] = set() for name, _ in pairs: if name in names: raise ValueError(f"JSON object keys collide after serialization: {name!r}") names.add(name) class _JSONValueError(ValueError): """Syntactically decodable JSON with ambiguous or unrepresentable values.""" def _decoded_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: _reject_duplicate_json_keys(pairs) return dict(pairs) def _reject_nonfinite_constant(token: str) -> None: raise ValueError(f"JSON numbers must be finite; found {token}") def _represented_json_float(token: str) -> float: value = float(token) finite_represented_real(value, "JSON number") # The decoder admits the grammar. Only the mantissa decides exact zero, # independently of the exponent's magnitude. mantissa = token.lower().split("e", 1)[0] if value == 0.0 and any(digit in "123456789" for digit in mantissa): raise ValueError("JSON number is nonzero but underflows to represented zero") return value def json_loads(text: str | bytes | bytearray) -> Any: """Decode JSON using the shared report and configuration admission policy. Reject duplicate decoded names, nonfinite numbers and nonzero fractional literals that underflow to binary64 zero. Integers retain Python's integer representation; finite fractions retain normal float rounding. Object, array and scalar roots are supported. This does not validate model fields or authenticate a report's provenance. """ try: return json.loads( text, object_pairs_hook=_decoded_object, parse_constant=_reject_nonfinite_constant, parse_float=_represented_json_float, ) except (json.JSONDecodeError, UnicodeDecodeError): raise except ValueError as exc: raise _JSONValueError(str(exc)) from exc def _json_dumps_std( obj: Any, params: JsonDumpsParams, **kwargs: Any, ) -> bytes | str: """Serialize using the standard library :func:`json.dumps`.""" result = json.dumps( obj, sort_keys=params.sort_keys, ensure_ascii=params.ensure_ascii, separators=params.separators, cls=params.cls, default=params.default, **kwargs, ) return result if not params.to_bytes else result.encode("utf-8") def json_dumps( obj: Any, *, sort_keys: bool = False, default: Callable[[Any], Any] | None = None, ensure_ascii: bool = True, separators: tuple[str, str] = (",", ":"), cls: type[json.JSONEncoder] | None = None, to_bytes: bool = False, **kwargs: Any, ) -> bytes | str: """Serialize with stable standard-library JSON semantics. Optional encoders must not change cache/signature bytes, turn nonfinite values into null, reject otherwise supported integers/keys, or ignore requested formatting. One encoder keeps those contracts independent of installed dependencies. Set allow_nan=False for strict finite JSON. """ if not isinstance(sort_keys, bool): raise TypeError("sort_keys must be a boolean") if default is not None and not callable(default): raise TypeError("default must be callable when provided") if not isinstance(ensure_ascii, bool): raise TypeError("ensure_ascii must be a boolean") if not isinstance(separators, tuple) or len(separators) != 2: raise TypeError("separators must be a tuple of two strings") if not all(isinstance(part, str) for part in separators): raise TypeError("separators must be a tuple of two strings") if cls is not None: if not isinstance(cls, type) or not issubclass(cls, json.JSONEncoder): raise TypeError("cls must be a subclass of json.JSONEncoder") if not isinstance(to_bytes, bool): raise TypeError("to_bytes must be a boolean") if ( sort_keys is False and default is None and ensure_ascii is True and separators == (",", ":") and cls is None and to_bytes is False ): params = DEFAULT_PARAMS else: params = JsonDumpsParams( sort_keys=sort_keys, default=default, ensure_ascii=ensure_ascii, separators=separators, cls=cls, to_bytes=to_bytes, ) return _json_dumps_std(obj, params, **kwargs) def _raise_import_error(name: str, *_: Any, **__: Any) -> Any: raise ImportError(f"{name} is not installed") _MISSING_TOML_ERROR = type( "MissingTOMLDependencyError", (Exception,), {"__doc__": "Fallback error used when tomllib/tomli is missing."}, ) _MISSING_YAML_ERROR = type( "MissingPyYAMLDependencyError", (Exception,), {"__doc__": "Fallback error used when pyyaml is missing."}, ) def _resolve_lazy(value: Any) -> Any: if isinstance(value, LazyImportProxy): return value.resolve() return value class _LazyBool: __slots__ = ("_value",) def __init__(self, value: Any) -> None: self._value = value def __bool__(self) -> bool: return _resolve_lazy(self._value) is not None _TOMLI_MODULE = cached_import("tomli", emit="log", lazy=True) tomllib = cached_import( "tomllib", emit="log", lazy=True, fallback=_TOMLI_MODULE, ) has_toml = _LazyBool(tomllib) _TOMLI_TOML_ERROR = cached_import( "tomli", "TOMLDecodeError", emit="log", lazy=True, fallback=_MISSING_TOML_ERROR, ) TOMLDecodeError = cached_import( "tomllib", "TOMLDecodeError", emit="log", lazy=True, fallback=_TOMLI_TOML_ERROR, ) _TOMLI_LOADS = cached_import( "tomli", "loads", emit="log", lazy=True, fallback=partial(_raise_import_error, "tomllib/tomli"), ) _TOML_LOADS: Callable[[str], Any] = cached_import( "tomllib", "loads", emit="log", lazy=True, fallback=_TOMLI_LOADS, ) yaml = cached_import("yaml", emit="log", lazy=True) YAMLError = cached_import( "yaml", "YAMLError", emit="log", lazy=True, fallback=_MISSING_YAML_ERROR, ) _YAML_SAFE_LOAD: Callable[[str], Any] = cached_import( "yaml", "safe_load", emit="log", lazy=True, fallback=partial(_raise_import_error, "pyyaml"), ) def _parse_yaml(text: str) -> Any: """Parse YAML ``text`` using ``safe_load`` if available.""" return _YAML_SAFE_LOAD(text) def _parse_toml(text: str) -> Any: """Parse TOML ``text`` using ``tomllib`` or ``tomli``.""" return _TOML_LOADS(text) PARSERS = { ".json": json_loads, ".yaml": _parse_yaml, ".yml": _parse_yaml, ".toml": _parse_toml, } def _get_parser(suffix: str) -> Callable[[str], Any]: try: return PARSERS[suffix] except KeyError as exc: raise TNFRValueError( f"Unsupported suffix: {suffix}", context={"suffix": suffix, "supported": list(PARSERS.keys())}, suggestion="Use a supported file extension.", ) from exc _BASE_ERROR_MESSAGES: dict[type[BaseException], str] = { OSError: "Could not read {path}: {e}", UnicodeDecodeError: "Encoding error while reading {path}: {e}", json.JSONDecodeError: "Error parsing JSON file at {path}: {e}", _JSONValueError: "Invalid JSON value at {path}: {e}", ImportError: "Missing dependency parsing {path}: {e}", } def _resolve_exception_type(candidate: Any) -> type[BaseException] | None: resolved = _resolve_lazy(candidate) if isinstance(resolved, type) and issubclass(resolved, BaseException): return resolved return None _OPTIONAL_ERROR_MESSAGE_FACTORIES: tuple[ tuple[Callable[[], type[BaseException] | None], str], ..., ] = ( ( lambda: _resolve_exception_type(YAMLError), "Error parsing YAML file at {path}: {e}", ), ( lambda: _resolve_exception_type(TOMLDecodeError), "Error parsing TOML file at {path}: {e}", ), ) _BASE_STRUCTURED_EXCEPTIONS = ( OSError, UnicodeDecodeError, json.JSONDecodeError, _JSONValueError, ImportError, ) def _iter_optional_exceptions() -> list[type[BaseException]]: errors: list[type[BaseException]] = [] for resolver, _ in _OPTIONAL_ERROR_MESSAGE_FACTORIES: exc_type = resolver() if exc_type is not None: errors.append(exc_type) return errors def _is_structured_error(exc: Exception) -> bool: if isinstance(exc, _BASE_STRUCTURED_EXCEPTIONS): return True for optional_exc in _iter_optional_exceptions(): if isinstance(exc, optional_exc): return True return False def _format_structured_file_error(path: Path, e: Exception) -> str: for exc, msg in _BASE_ERROR_MESSAGES.items(): if isinstance(e, exc): return msg.format(path=path, e=e) for resolver, msg in _OPTIONAL_ERROR_MESSAGE_FACTORIES: exc_type = resolver() if exc_type is not None and isinstance(e, exc_type): return msg.format(path=path, e=e) return f"Error parsing {path}: {e}" class StructuredFileError(Exception): """Error while reading or parsing a structured file.""" def __init__(self, path: Path, original: Exception) -> None: super().__init__(_format_structured_file_error(path, original)) self.path = path def read_structured_file( path: Path | str, *, base_dir: Path | str | None = None, allowed_extensions: tuple[str, ...] | None = (".json", ".yaml", ".yml", ".toml"), ) -> Any: """Read a JSON, YAML or TOML file and return parsed data. This function includes path traversal protection. When ``base_dir`` is provided, the resolved path must stay within that directory. JSON uses :func:`json_loads`'s strict value admission. YAML and TOML retain their respective parser semantics; consumers validate their model fields. Parameters ---------- path : Path | str Path to the structured file to read. base_dir : Path | str | None, optional Base directory to restrict file access. If provided, the resolved path must stay within this directory (prevents path traversal). allowed_extensions : tuple[str, ...] | None, optional tuple of allowed file extensions. Default is JSON, YAML, and TOML. Pass None to allow any extension (not recommended for user input). Returns ------- Any Parsed data from the file. Raises ------ StructuredFileError If the file cannot be read or parsed. ValueError If the path is invalid or contains unsafe patterns. PathTraversalError If path traversal is detected. Examples -------- >>> from pathlib import Path >>> # Read config file with path validation >>> data = read_structured_file("config.json") # doctest: +SKIP >>> # Read with base directory restriction >>> data = read_structured_file( ... "settings.yaml", ... base_dir="/home/user/configs" ... ) # doctest: +SKIP """ # Import here to avoid circular dependency from ..security import PathTraversalError, resolve_safe_path, validate_file_path # Validate and resolve the path try: if base_dir is not None: # Resolve path within base directory (prevents traversal) validated_path = resolve_safe_path( path, base_dir, must_exist=True, allowed_extensions=allowed_extensions, ) else: # Validate path without base directory restriction path_obj = Path(path) if not isinstance(path, Path) else path validated_path = validate_file_path( path_obj, allow_absolute=True, allowed_extensions=allowed_extensions, ).resolve() # Check existence if not validated_path.exists(): raise FileNotFoundError(f"File not found: {validated_path}") except (ValueError, PathTraversalError) as e: raise StructuredFileError(Path(path), e) from e except FileNotFoundError as e: raise StructuredFileError(Path(path), e) from e suffix = validated_path.suffix.lower() try: parser = _get_parser(suffix) except ValueError as e: raise StructuredFileError(validated_path, e) from e try: text = validated_path.read_text(encoding="utf-8") return parser(text) except Exception as e: if _is_structured_error(e): raise StructuredFileError(validated_path, e) from e raise def safe_write( path: str | Path, write: Callable[[Any], Any], *, mode: str = "w", encoding: str | None = "utf-8", atomic: bool = True, sync: bool | None = None, base_dir: str | Path | None = None, **open_kwargs: Any, ) -> None: """Write to ``path`` ensuring parent directory exists and handle errors. This function includes path traversal protection. When ``base_dir`` is provided, the resolved path must stay within that directory. Parameters ---------- path: Destination file path. write: Callback receiving the opened file object and performing the actual write. mode: File mode passed to :func:`open`. Text modes (default) use UTF-8 encoding unless ``encoding`` is ``None``. When a binary mode is used (``'b'`` in ``mode``) no encoding parameter is supplied so ``write`` may write bytes. Atomic writing supports replacement modes based on 'w' only. Use atomic=False for append, update or exclusive creation; a fresh temporary file cannot supply those semantics. encoding: Encoding for text modes. Ignored for binary modes. atomic: When ``True`` (default) writes to a temporary file and atomically replaces the destination after flushing to disk. When ``False`` writes directly to ``path`` without any atomicity guarantee. sync: When ``True`` flushes and fsyncs the file descriptor after writing. ``None`` uses ``atomic`` to determine syncing behaviour. base_dir: Optional base directory to restrict file writes. If provided, the resolved path must stay within this directory (prevents path traversal). Raises ------ ValueError If the path is invalid or contains unsafe patterns. PathTraversalError If path traversal is detected when base_dir is provided. """ # Import here to avoid circular dependency from ..security import PathTraversalError, resolve_safe_path, validate_file_path # Validate and resolve the path try: if base_dir is not None: # Resolve path within base directory (prevents traversal) validated_path = resolve_safe_path( path, base_dir, must_exist=False, ) else: # Validate path without base directory restriction path_obj = Path(path) if not isinstance(path, Path) else path validated_path = validate_file_path( path_obj, allow_absolute=True, ).resolve() except (ValueError, PathTraversalError) as e: raise type(e)(f"Invalid path {path!r}: {e}") from e if not isinstance(mode, str): raise TypeError("mode must be a string") if atomic and mode.replace("b", "").replace("t", "").replace("+", "") != "w": raise ValueError("Atomic writing requires a replacement mode based on 'w'") path = validated_path path.parent.mkdir(parents=True, exist_ok=True) open_params = dict(mode=mode, **open_kwargs) if "b" not in mode and encoding is not None: open_params["encoding"] = encoding if sync is None: sync = atomic tmp_path: Path | None = None try: if atomic: tmp_fd = tempfile.NamedTemporaryFile(dir=path.parent, delete=False) tmp_path = Path(tmp_fd.name) tmp_fd.close() with open(tmp_path, **open_params) as fd: write(fd) if sync: fd.flush() os.fsync(fd.fileno()) try: os.replace(tmp_path, path) except OSError as e: logger.error( "Atomic replace failed for %s -> %s: %s", tmp_path, path, e ) raise else: with open(path, **open_params) as fd: write(fd) if sync: fd.flush() os.fsync(fd.fileno()) except (OSError, ValueError, TypeError) as e: # Preserve structured exception arguments, including encoding details # and OSError.errno. add_note is available on Python 3.11+. if hasattr(e, "add_note"): e.add_note(f"Failed to write file {path}") raise finally: if tmp_path is not None: tmp_path.unlink(missing_ok=True) __all__ = ( "JsonDumpsParams", "DEFAULT_PARAMS", "json_dumps", "json_loads", "read_structured_file", "safe_write", "StructuredFileError", "TOMLDecodeError", "YAMLError", )