Add scene-witness reducer (checkpoint: contract module saved)

server/api.py

@@ -53,12 +53,31 @@ def _limit_bytes(env: str, default_mb: float) -> int:
     return int(mb * _MB)
 
 
+def _limit_int(env: str, default: int) -> int:
+    raw = os.environ.get(env)
+    if raw is None:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        value = -1
+    if value <= 0:
+        raise ValueError(f"{env}={raw!r} must be a positive whole number")
+    return value
+
+
 # Read once at import so a bad value stops the server at startup. Tests monkeypatch these names.
 MAX_UPLOAD_BYTES = _limit_bytes("HOUSESCAN_MAX_UPLOAD_MB", 256)
 MAX_UNZIPPED_BYTES = _limit_bytes("HOUSESCAN_MAX_UNZIPPED_MB", 512)
 # Parsing JSON takes many times its size in memory (10.5 MB took 255 MB), and a real scene.json
 # is well under 1 MB, so scene.json itself is capped far below the bundle.
 MAX_SCENE_BYTES = _limit_bytes("HOUSESCAN_MAX_SCENE_MB", 10)
+# A bundle's central directory is parsed in full the moment ZipFile is built, so the directory's
+# entry count and size are read from its end record BEFORE the bundle is opened. 20,000 entries
+# is far above any real walk's file count, and a directory holding them is far under this byte
+# cap; together they keep a crafted bundle from spending the request on metadata alone.
+MAX_ZIP_ENTRIES = _limit_int("HOUSESCAN_MAX_ZIP_ENTRIES", 20_000)
+MAX_ZIP_DIRECTORY_KB = _limit_int("HOUSESCAN_MAX_ZIP_DIRECTORY_KB", 8_192)
 
 # Loaded at import: invalid rules must stop the server before it answers anything.
 LOADED: LoadedRules = load_rules()
@@ -181,6 +200,8 @@ def _policy() -> dict[str, Any]:
         "version": p.version,
         # Same rule as solver.solve: no policy id means nothing is approved automatically.
         "auto_approve": p.auto_approve and p.id is not None,
+        # Same rule as solver.solve: reported so a client can ask why nothing was rejected.
+        "allow_reject": p.allow_reject,
         "sources": list(LOADED.sources),
         "rules_sha256": LOADED.sha256,
         "notice": p.notice,
@@ -298,8 +319,39 @@ def _check_scene_size(size: int) -> None:
         )
 
 
+def _zip_directory(data: bytes) -> tuple[int, int]:
+    """The central directory's entry count and size, read from the zip's end record before any
+    metadata is parsed. `zipfile.ZipFile` reads the whole directory and builds every entry while
+    it is constructed, so a later count comes too late. `zipfile._EndRecData` is the function
+    ZipFile itself uses (ZIP64 included), so the sizes checked here are the ones it will read."""
+    try:
+        with io.BytesIO(data) as f:
+            end = zipfile._EndRecData(f)
+    except OSError:
+        end = None
+    if not end:
+        raise ApiError(
+            400, "unreadable_zip", "The bundle has no zip end-of-central-directory record."
+        )
+    return end[zipfile._ECD_ENTRIES_TOTAL], end[zipfile._ECD_SIZE]
+
+
 def _open_bundle(data: bytes) -> tuple[bytes, set[str], str]:
     """scene.json's bytes, every file name in the bundle, and the folder scene.json sits in."""
+    entries, directory_bytes = _zip_directory(data)
+    if entries > MAX_ZIP_ENTRIES:
+        raise ApiError(
+            413,
+            "bundle_too_many_entries",
+            f"The bundle lists {entries} entries, over the {MAX_ZIP_ENTRIES} limit.",
+        )
+    if directory_bytes > MAX_ZIP_DIRECTORY_KB * 1024:
+        raise ApiError(
+            413,
+            "bundle_directory_too_large",
+            f"The bundle's zip directory is {directory_bytes} bytes, over the "
+            f"{MAX_ZIP_DIRECTORY_KB} KB limit.",
+        )
     try:
         zf = zipfile.ZipFile(io.BytesIO(data))
     except (zipfile.BadZipFile, zlib.error, ValueError, EOFError) as exc:

server/examples/smoke.py

@@ -45,7 +45,9 @@ def redirect_request(self, *args: object) -> None:
         return None
 
 
-_KEYED = urllib.request.build_opener(_NoRedirect)
+# An explicit empty proxy table: with http_proxy set, a keyed request would otherwise go through
+# the proxy, key included.
+_KEYED = urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect)
 
 
 def post(url: str, body: bytes, key: str | None) -> tuple[int, dict]:

server/rules.py

@@ -15,7 +15,7 @@
 from typing import Any, Literal
 
 import yaml
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 SERVER_DIR = Path(__file__).resolve().parent
 PUBLIC_RULES = SERVER_DIR / "rules.yaml"
@@ -90,6 +90,13 @@ class Sweep(_Strict):
     wall_join_ft: Value
     meter_to_wall_max_ft: Value
 
+    @model_validator(mode="after")
+    def _positive_step(self) -> "Sweep":
+        # Every solve steps the battery along the wall by this much; zero or less can't advance.
+        if self.step_ft.value <= 0:
+            raise ValueError(f"sweep.step_ft must be positive, got {self.step_ft.value}")
+        return self
+
 
 class Clearances(_Strict):
     gas_ft: Value
@@ -124,6 +131,16 @@ class MeterWorkingSpace(_Strict):
     width_ft: Value
     depth_ft: Value
 
+    @model_validator(mode="after")
+    def _positive_workspace(self) -> "MeterWorkingSpace":
+        # A zero width or depth collapses the working space to a line or a point; the solve
+        # would compare against a degenerate polygon and let a battery stand in the meter.
+        for name in ("width_ft", "depth_ft"):
+            value = getattr(self, name).value
+            if value <= 0:
+                raise ValueError(f"meter_working_space.{name} must be positive, got {value}")
+        return self
+
 
 class Ground(_Strict):
     allowed: list[GroundType]
@@ -139,6 +156,17 @@ class Route(_Strict):
     corner_allowance_ft: Value
     crossing: dict[ObjectType, Effect]
 
+    @model_validator(mode="after")
+    def _confident_within_max(self) -> "Route":
+        # Past the confident reach a run goes to review, past the maximum it fails; a confident
+        # reach beyond the maximum would let a run over the maximum pass.
+        if self.confident_reach_ft.value > self.max_ft.value:
+            raise ValueError(
+                f"route.confident_reach_ft ({self.confident_reach_ft.value}) must not exceed "
+                f"route.max_ft ({self.max_ft.value})"
+            )
+        return self
+
 
 class Rules(_Strict):
     policy: Policy
@@ -154,6 +182,18 @@ class Rules(_Strict):
     ground: Ground
     route: Route
 
+    @model_validator(mode="after")
+    def _exemption_within_the_opening_checks_height(self) -> "Rules":
+        # The opening check needs the wall seen only up to headroom height; a window above that
+        # would still count under a higher exemption but could be missed, unseen, above the view.
+        exempt = self.openings.exempt_bottom_above_ft
+        if exempt is not None and exempt > self.headroom.min_ft.value:
+            raise ValueError(
+                f"openings.exempt_bottom_above_ft ({exempt}) must not exceed headroom.min_ft "
+                f"({self.headroom.min_ft.value}), the wall height the opening check requires seen"
+            )
+        return self
+
 
 @dataclass(frozen=True)
 class LoadedRules:
@@ -251,11 +291,67 @@ def load_rules(private_path: Path | None = None) -> LoadedRules:
     if private is None:
         return rules_from_dict(data)
     private_keys = frozenset(overridden_keys(data, private))
-    if "notice" not in private.get("policy", {}):
-        # The public notice says the answers are not under Base's rules; merged under private
-        # rules it would be false.
-        data["policy"]["notice"] = None
-    return rules_from_dict(deep_merge(data, private), ("public", "private"), private_keys)
+    merged = deep_merge(data, private)
+    # The public notice says the answers are not under Base's rules, which private rules make
+    # untrue; but any public placeholder they leave in place still decides answers, so the
+    # notice names those checks instead of going quiet.
+    own = private.get("policy", {}).get("notice")
+    merged["policy"]["notice"] = _mixed_notice(own, _placeholders_left(merged, private_keys))
+    return rules_from_dict(merged, ("public", "private"), private_keys)
+
+
+# The check each placeholder rule decides; placeholders not tied to one check are named by key.
+PLACEHOLDER_CHECKS = {
+    "clearances.drive_ft": "drive_clearance",
+    "clearances.pool_ft": "pool_clearance",
+    "clearances.wall_equipment_ft": "wall_equipment_above",
+    "headroom.min_ft": "headroom",
+    "ground": "ground_surface",
+    "route.confident_reach_ft": "route_length",
+    "route.corner_allowance_ft": "route_length",
+    "route.height_ft": "route_path",
+}
+
+
+def _placeholders_left(merged: dict[str, Any], private_keys: frozenset[str]) -> list[str]:
+    """Dotted keys still marked placeholder that the private file did not set."""
+    left: list[str] = []
+
+    def walk(node: dict[str, Any], path: str) -> None:
+        if node.get("placeholder") is True:
+            # A cited value is replaced when the private file sets it; a group (such as ground)
+            # only when it sets every child that decides something, not just one of them.
+            if "value" in node:
+                replaced = path in private_keys
+            else:
+                children = [k for k in node if k not in ("source", "placeholder")]
+                replaced = all(
+                    any(p == f"{path}.{k}" or p.startswith(f"{path}.{k}.") for p in private_keys)
+                    for k in children
+                )
+            if not replaced:
+                left.append(path)
+            return
+        for key, value in node.items():
+            if isinstance(value, dict):
+                walk(value, f"{path}.{key}" if path else key)
+
+    walk(merged, "")
+    return left
+
+
+def _mixed_notice(own: str | None, left: list[str]) -> str | None:
+    if not left:
+        return own
+    checks = sorted({PLACEHOLDER_CHECKS[k] for k in left if k in PLACEHOLDER_CHECKS})
+    other = sorted(k for k in left if k not in PLACEHOLDER_CHECKS)
+    parts = []
+    if checks:
+        parts.append("these checks still use public placeholder values: " + ", ".join(checks))
+    if other:
+        parts.append("placeholder settings still apply: " + ", ".join(other))
+    mixed = "Private rules, but " + "; ".join(parts) + "."
+    return f"{own} {mixed}" if own else mixed
 
 
 def _public_policy(data: dict[str, Any]) -> dict[str, Any]: