# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/whenever-schema-validation-and-ingress-parsing-converge-how-must-an-instance-enforce-fail-closed-type-coercion-to-prevent-silent-default-corruption.md # Remnant Warrant: https://corpusfetch.com/c/whenever-schema-validation-and-ingress-parsing-converge-how-must-an-instance-enforce-fail-closed-type-coercion-to-prevent-silent-default-corruption.warrant.json # Markdown SHA-256: 62dc781ea4604242f11db965e022b6bbf645e1fdf695f9ed1354fe4afc0fdc77 # Code AST SHA-256: bcf30a87eba2900916a8b471d56bc8eb31353bbe0520bddad1087506da124a4d # Verified Classes: WorkerConfig # Defensive Guards: 7 fail-closed invariants (zero external deps) from dataclasses import dataclass from typing import Any, Mapping @dataclass(frozen=True) class WorkerConfig: host: str port: int timeout_ms: int read_only: bool def parse_strict_bool(val: Any, field_name: str) -> bool: ''' Parses a boolean strictly without falling into the bool("false") == True trap. ''' if isinstance(val, bool): return val if isinstance(val, str): normalized = val.strip().lower() if normalized in ("true", "1", "yes"): return True if normalized in ("false", "0", "no"): return False raise TypeError(f"Field '{field_name}' requires a strict boolean, got {val!r}") def parse_strict_int(val: Any, field_name: str, min_val: int | None = None) -> int: ''' Validates an exact integer while rejecting booleans (since isinstance(True, int) is True). ''' if isinstance(val, bool) or not isinstance(val, int): raise TypeError(f"Field '{field_name}' requires an integer, got {type(val).__name__} ({val!r})") if min_val is not None and val < min_val: raise ValueError(f"Field '{field_name}' must be >= {min_val}, got {val}") return val def validate_worker_config(raw: Mapping[str, Any]) -> WorkerConfig: ''' Enforces fail-closed ingress validation: forbids unknown keys, rejects None on defaulted fields, and validates strict types before returning a frozen config. ''' if not isinstance(raw, Mapping): raise TypeError(f"Expected a mapping payload, got {type(raw).__name__}") allowed_keys = {"host", "port", "timeout_ms", "read_only"} required_keys = {"host", "port"} # Gate 1: Reject unknown / misspelled keys (extra = forbid) unknown_keys = set(raw.keys()) - allowed_keys if unknown_keys: raise ValueError(f"Rejected unrecognized configuration keys: {sorted(unknown_keys)}") # Gate 2: Verify required keys exist missing_keys = required_keys - set(raw.keys()) if missing_keys: raise ValueError(f"Missing required configuration keys: {sorted(missing_keys)}") # Gate 3: Validate types and handle defaults without allowing explicit None to slip through host = raw["host"] if not isinstance(host, str) or not host.strip(): raise ValueError("Field 'host' must be a non-empty string") port = parse_strict_int(raw["port"], "port", min_val=1) raw_timeout = raw.get("timeout_ms") timeout_ms = 5000 if raw_timeout is None else parse_strict_int(raw_timeout, "timeout_ms", min_val=1) raw_read_only = raw.get("read_only") read_only = False if raw_read_only is None else parse_strict_bool(raw_read_only, "read_only") return WorkerConfig( host=host.strip(), port=port, timeout_ms=timeout_ms, read_only=read_only, )