Port #221 server hardening onto the #11 port

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/rules.py

@@ -131,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]

server/scene.py

@@ -162,9 +162,14 @@ class Scene:
     ground: list[GroundPatch]
     overheads: list[Measured]
     facing: list[Measured]
-    observed: dict[str, list[tuple[float, float, float | None]]]
+    # Coverage views as (span start, span end, out_ft, camera): the camera position is set only
+    # when the view declared one (a ground view's evidence that the camera stood there).
+    observed: dict[str, list[tuple[float, float, float | None, tuple[float, float] | None]]]
     end_kinds: dict[str, str]
     reach_ft: float  # how far out from the walls the outdoor area is modelled
+    # The plan positions of the packet's keyframe cameras: the walk's own record of where the
+    # cameras stood, what a coverage view's camera_pos_ft is checked against.
+    keyframe_positions: list[tuple[float, float]] = field(default_factory=list)
     # A wall view without out_ft reached headroom height (headroom.min_ft), per the contract.
     wall_default_ft: float = math.inf
     _cache: dict[str, Any] = field(default_factory=dict, repr=False)
@@ -262,19 +267,36 @@ def outdoor_band(self, out: float, span: tuple[float, float] | None = None) -> G
         lo, hi = span or (self.pieces[0].s0, self.pieces[-1].s1)
         return unary_union([self.band_polygon(lo, hi, out), *self.passages(out)])
 
-    def view_polygon(self, s_lo: float, s_hi: float, out: float) -> Geometry:
+    def view_polygon(
+        self, s_lo: float, s_hi: float, out: float, camera: tuple[float, float] | None = None
+    ) -> Geometry:
         """What a ground view over [s_lo, s_hi] out to `out` shows: the band in front of the
         chain, and where its span lies within a gap, the passage behind the gap's line as well:
         the camera pointed into the passage from the opening, as a view past a limit end shows
         both sides of the continued line. A view that runs along the walls across a gap shows
         only the front (final review: a side passage counted as seen from a view along the
-        walls)."""
+        walls). The passage is claimed only when the view was taken from there: its `camera`
+        position — checked against the walk's keyframes at parse time — must sit in front of
+        the gap's line within its stretch. A span alone does not say the camera was there."""
         parts = [self.band_polygon(s_lo, s_hi, out)]
         for g in self.gaps:
-            if s_lo >= g.s0 - GAP_VIEW_SLACK_FT and s_hi <= g.s1 + GAP_VIEW_SLACK_FT and out > EPS:
-                parts.append(g.rect(max(s_lo, g.s0), min(s_hi, g.s1), -out, 0.0))
+            if (
+                s_lo >= g.s0 - GAP_VIEW_SLACK_FT
+                and s_hi <= g.s1 + GAP_VIEW_SLACK_FT
+                and out > EPS
+                and camera is not None
+            ):
+                s_cam, out_cam = self._where(camera)
+                if g.s0 - GAP_VIEW_SLACK_FT <= s_cam <= g.s1 + GAP_VIEW_SLACK_FT and out_cam > EPS:
+                    parts.append(g.rect(max(s_lo, g.s0), min(s_hi, g.s1), -out, 0.0))
         return unary_union(parts)
 
+    def _where(self, point: tuple[float, float]) -> tuple[float, float]:
+        """A plan point's s along the chain and its signed out-distance: where it stands
+        relative to the walls' line."""
+        nearest = min(self.pieces, key=lambda p: Point(point).distance(LineString([p.a, p.b])))
+        return nearest.local(point)
+
     def s_of(self, p: Point2) -> float:
         """s of the chain point nearest to a plan point."""
         best = min(self.pieces, key=lambda q: Point(p).distance(LineString([q.a, q.b])))
@@ -307,7 +329,7 @@ def observed_intervals(
         return merge_intervals(
             [
                 (a, b)
-                for a, b, out in self.observed.get(band, [])
+                for a, b, out, _cam in self.observed.get(band, [])
                 if up_to is None
                 or (out is None and up_to <= self._default_reach(band) + EPS)
                 or (out is not None and out > up_to)
@@ -334,13 +356,15 @@ def unobserved_ground(self) -> Geometry:
             self._cache["ground"] = self.unobserved_ground_given(self.observed.get("ground", []))
         return self._cache["ground"]
 
-    def unobserved_ground_given(self, ground: list[tuple[float, float, float | None]]) -> Geometry:
+    def unobserved_ground_given(
+        self, ground: list[tuple[float, float, float | None, tuple[float, float] | None]]
+    ) -> Geometry:
         """The ground nobody saw, given these ground views: what unobserved_ground is for the
         scene's own views, and what a request is checked against with its capture added."""
         # Behind a scanned wall is the house: at an inside corner one wall's strip reaches back
         # across the other's line, and that ground no view in front of a wall can show.
         outdoor = _minus(self.outdoor_band(self.reach_ft), self.house())
-        seen = unary_union([self.view_polygon(a, b, out or 0.0) for a, b, out in ground])
+        seen = unary_union([self.view_polygon(a, b, out or 0.0, cam) for a, b, out, cam in ground])
         # Growing what was seen (SEEN_GROWTH_FT) closes gaps between observed spans narrower
         # than the tolerance, as missing() ignores them for the 1D bands.
         unseen = _minus(outdoor, seen.buffer(SEEN_GROWTH_FT))
@@ -384,8 +408,10 @@ def _unexplored_discs(
         lo, hi = self.coverable_span()
         seen_in_front = unary_union(
             [
-                self.view_polygon(max(a, lo), min(b, hi), out or 0.0)
-                for a, b, out in (self.observed.get("ground", []) if ground is None else ground)
+                self.view_polygon(max(a, lo), min(b, hi), out or 0.0, cam)
+                for a, b, out, cam in (
+                    self.observed.get("ground", []) if ground is None else ground
+                )
             ]
         ).buffer(SEEN_GROWTH_FT)
         known = unary_union([seen_in_front, self.house()])
@@ -396,7 +422,8 @@ def seen_to(self, band: str, s_lo: float, s_hi: float) -> float:
         deepest view covering it, and over the stretch the shallowest of those. A view without
         `out_ft` saw all the way (infinite); a stretch no view covers gives 0."""
         spans = [
-            (a, b, math.inf if out is None else out) for a, b, out in self.observed.get(band, [])
+            (a, b, math.inf if out is None else out)
+            for a, b, out, _cam in self.observed.get(band, [])
         ]
         cuts = sorted({s_lo, s_hi} | {x for a, b, _ in spans for x in (a, b) if s_lo < x < s_hi})
         least = math.inf
@@ -438,7 +465,7 @@ def covers(a: float, b: float, depth: float) -> bool:
             # ft a lens where a clearance circle meets the view's edge survived 1.7e-5 ft inside
             # the radius (test_final_review, the 108 degree corner); 1e-12 is still far above
             # the snapped overlays' noise (around 1e-18).
-            unseen = polygonal(self.unobserved_ground_given([*ground, (a, b, depth)]))
+            unseen = polygonal(self.unobserved_ground_given([*ground, (a, b, depth, None)]))
             return shapely.intersection(area, unseen, grid_size=1e-9).area <= 1e-12
 
         tol = COVERAGE_TOLERANCE_FT
@@ -727,6 +754,38 @@ def _minus(a: Geometry, b: Geometry) -> Geometry:
 # Coordinates beyond this are not a house scan; they would only exhaust memory in the sweep.
 MAX_COORDINATE_FT = 1e5
 
+# How close a coverage view's camera_pos_ft must sit to a keyframe's camera for the claim to be
+# the walk's own: about an arm's length of scene drift, far under the meter's worst default
+# error. Only a ground view taken from an opening may claim the passage behind a gap, so its
+# position is evidence and is checked, not taken on trust.
+CAMERA_TOLERANCE_FT = 1.0
+
+
+def _camera_of(
+    obs: dict[str, Any], keyframes: list[tuple[float, float]], path: str
+) -> tuple[float, float] | None:
+    """The camera position a coverage view claims, or None. A declared position must be one the
+    walk can vouch for: when the packet has keyframes, it is refused unless a keyframe's camera
+    sits within CAMERA_TOLERANCE_FT of it. A view's span alone does not say the camera was
+    there, so the position may not be invented either (review: a side passage counted as seen
+    from a view along the walls)."""
+    pos = obs.get("camera_pos_ft")
+    if pos is None:
+        return None
+    if not (isinstance(pos, list) and len(pos) == 2):
+        raise SceneError(path, "camera_pos_ft must be two plan coordinates [x, z]")
+    cam = (float(pos[0]), float(pos[1]))
+    vouched = not keyframes or any(
+        math.hypot(cam[0] - kx, cam[1] - kz) <= CAMERA_TOLERANCE_FT for kx, kz in keyframes
+    )
+    if not vouched:
+        raise SceneError(
+            path,
+            "camera_pos_ft names no keyframe's camera within "
+            f"{CAMERA_TOLERANCE_FT:g} ft; a coverage view's position must come from the walk",
+        )
+    return cam
+
 
 def _check_numbers(raw: Any) -> None:
     """JSON Schema accepts NaN and infinities as numbers; nothing in a scene may be either.
@@ -928,7 +987,11 @@ def parse_scene(raw: dict[str, Any], rules: Rules, input_bytes: bytes | None = N
         c.pool_ft.value,
     )
     wall_error = max(p.error_at(max(abs(p.s0), abs(p.s1))) for p in pieces if p.kind == "wall")
-    reach += rules.battery.depth_ft.value + wall_error + 1.0
+    # The meter's own position error shifts where every out-from-the-wall distance is measured
+    # from, so the outdoor area it can matter over widens by it too (review: an all-PASS
+    # candidate with meter error 2 ft and ground observed to 13 ft passed although its pool
+    # coverage needed 13.833333 ft).
+    reach += rules.battery.depth_ft.value + wall_error + meter_err + 1.0
     first, last = pieces[0], pieces[-1]
     ext_len = reach + rules.battery.width_ft.value
     left_ext = Piece(
@@ -959,6 +1022,10 @@ def parse_scene(raw: dict[str, Any], rules: Rules, input_bytes: bytes | None = N
     )
     pieces = [left_ext, *pieces, right_ext]
 
+    kf_positions = [
+        (float(kf["pose"][12]), float(kf["pose"][14])) for kf in raw.get("keyframes", [])
+    ]
+
     scene = Scene(
         input_sha256=hashlib.sha256(input_bytes).hexdigest(),
         pieces=pieces,
@@ -971,6 +1038,7 @@ def parse_scene(raw: dict[str, Any], rules: Rules, input_bytes: bytes | None = N
         ground=[],
         overheads=[],
         facing=[],
+        keyframe_positions=kf_positions,
         observed={},
         end_kinds={"left": "unexplored", "right": "unexplored"},
         reach_ft=reach,
@@ -1055,7 +1123,9 @@ def parse_scene(raw: dict[str, Any], rules: Rules, input_bytes: bytes | None = N
     coverage = raw.get("coverage", {})
     for i, obs in enumerate(coverage.get("observed", [])):
         span = _span(obs["span_ft"], f"/coverage/observed/{i}/span_ft")
-        scene.observed.setdefault(obs["band"], []).append((span[0], span[1], obs.get("out_ft")))
+        cam = _camera_of(obs, kf_positions, f"/coverage/observed/{i}/camera_pos_ft")
+        entry = (span[0], span[1], obs.get("out_ft"), cam)
+        scene.observed.setdefault(obs["band"], []).append(entry)
 
     _check_orientation(scene, raw.get("keyframes", []))
     return scene

server/schemas/result.schema.json

@@ -36,6 +36,7 @@
         "id": {"type": ["string", "null"]},
         "version": {"type": ["string", "null"]},
         "auto_approve": {"type": "boolean", "description": "Whether the rules allow automatic decisions: false when no policy is selected or the rules set auto_approve false (the public strict policy does, because some values are placeholders; the public demo policy decides anyway and says so in notice). When false, every would-be pass or reject becomes manual_review. Each check's rule.placeholder says which values decided it."},
+        "allow_reject": {"type": "boolean", "description": "Whether the rules in use permit automatic rejections. When false, every would-be reject becomes manual_review with a policy_review_before_reject reason saying so."},
         "sources": {"type": "array", "items": {"enum": ["public", "private"]}, "description": "Which rules files were loaded."},
         "rules_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$", "description": "Hash of the merged rules the decision used."},
         "notice": {"type": ["string", "null"], "description": "Whose rules these are, to show with the answer, for example that the public demo policy is not Base's. Null for rules that need no such label."}
@@ -218,6 +219,7 @@
             "properties": {"subject": {"type": "string"}, "extra_ft": {"type": "number"}}
           }
         },
+        "length_is_lower_bound": {"type": "boolean", "description": "True when a detour object on the run had no recorded height, so the detour's length is unknown and length_ft is a floor, not a measurement. The path check names the object."},
         "crossings": {
           "type": "array",
           "items": {

server/schemas/scene.schema.json

@@ -153,7 +153,8 @@
                 "description": "wall: the wall face from the ground up to `out_ft` (absent: up to headroom height), for the battery's back, the cable route, and boxes, vents, gas and openings above the battery. ground: the ground in front of the wall out to `out_ft`. overhead: the space above the ground strip up to headroom height. facing: the gap straight out from the wall to whatever faces it."
               },
               "span_ft": {"$ref": "#/$defs/span", "description": "Observed stretch in s. May extend past the chain's ends, continuing the end segments' lines. Past an end marked limit, observed ground covers both sides of that line."},
-              "out_ft": {"type": "number", "minimum": 0, "description": "How far the view reached, as a distance it is sure of. ground (required): how far out from the wall the ground was seen. facing: how far out from the wall the space was seen clear, for example the homeowner's walked path; absent means the view reached whatever faces the wall, so anything there is in `facing`. overhead: how high above the ground the space was seen clear; absent means it was seen all the way up, so anything above is in `overheads`. wall: how high above the ground the wall face was seen; absent means up to headroom height (headroom.min_ft), enough for every check. Each check that reads the wall band credits an entry only where its out_ft exceeds the height the check needs (server/README.md, \"What settles each check\")."}
+              "out_ft": {"type": "number", "minimum": 0, "description": "How far the view reached, as a distance it is sure of. ground (required): how far out from the wall the ground was seen. facing: how far out from the wall the space was seen clear, for example the homeowner's walked path; absent means the view reached whatever faces the wall, so anything there is in `facing`. overhead: how high above the ground the space was seen clear; absent means it was seen all the way up, so anything above is in `overheads`. wall: how high above the ground the wall face was seen; absent means up to headroom height (headroom.min_ft), enough for every check. Each check that reads the wall band credits an entry only where its out_ft exceeds the height the check needs (server/README.md, \"What settles each check\")."},
+              "camera_pos_ft": {"type": "array", "items": {"type": "number"}, "minItems": 2, "maxItems": 2, "description": "Plan position [x, z] of the camera that took this view, from the walk's keyframes. Optional. A ground view claims the passage behind a gap's line only with it, and it must sit within 1.0 ft of a keyframe's camera when the packet has keyframes; a span alone does not say the camera was there."}
             },
             "if": {"properties": {"band": {"const": "ground"}}},
             "then": {"required": ["out_ft"]}