security(request): snapshot untrusted extension mappings before timeout access

src/egressweave/request_safety.py

@@ -260,13 +260,20 @@ def _bind_bounded_request_timeouts(
     every phase after the destination has already passed policy validation.
     Missing or disabled values therefore receive the immutable policy maximum;
     stricter non-negative finite values are preserved and larger values are
-    capped. Malformed maps, unknown keys, booleans, negative numbers, and
+    capped. The outer mapping is detached exactly once before timeout lookup so
+    caller-controlled ``get`` methods cannot cross the generic denial boundary
+    and a stateful mapping cannot present different extension snapshots during
+    one decision. Malformed maps, unknown keys, booleans, negative numbers, and
     non-finite values fail through the generic policy boundary before HTTPCore
     can allocate a connection or wait on network I/O. Failures raised by
     attacker-controlled mapping, key-comparison, or numeric protocol methods
     are also masked.
     """
-    raw_timeout = extensions.get("timeout")
+    safe_extensions = _copy_request_extensions(extensions)
+    if safe_extensions is None:
+        raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
+
+    raw_timeout = safe_extensions.get("timeout")
     if raw_timeout is None:
         requested_timeouts: dict[object, object] | None = {}
     elif isinstance(raw_timeout, Mapping):
@@ -299,7 +306,6 @@ def _bind_bounded_request_timeouts(
             raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
         bounded_timeouts[key] = min(normalized_value, maximum)
 
-    safe_extensions = dict(extensions)
     safe_extensions["timeout"] = bounded_timeouts
     return safe_extensions
 

tests/test_request_timeout_untrusted_objects.py

@@ -33,6 +33,46 @@ def __len__(self) -> int:
         return 1
 
 
+class _ExplodingExtensionsCopyMapping(Mapping[str, object]):
+    """Raise while the untrusted outer request-extension mapping is detached."""
+
+    def __getitem__(self, key: str) -> object:
+        """Delegate indexed access to an unexpected secret-bearing failure."""
+        del key
+        return _raise_unexpected_protocol_failure("secret extensions copy failure")
+
+    def __iter__(self) -> Iterator[str]:
+        """Advertise one ordinary reviewed extension key."""
+        return iter(("timeout",))
+
+    def __len__(self) -> int:
+        """Report the one advertised extension key."""
+        return 1
+
+
+class _ExplodingExtensionsGetMapping(Mapping[str, object]):
+    """Expose safe items while making direct ``get`` dispatch attacker-controlled."""
+
+    def __getitem__(self, key: str) -> object:
+        """Return one ordinary timeout mapping through indexed access."""
+        if key == "timeout":
+            return {"connect": 1.0}
+        raise KeyError(key)
+
+    def __iter__(self) -> Iterator[str]:
+        """Advertise the single reviewed request-extension key."""
+        return iter(("timeout",))
+
+    def __len__(self) -> int:
+        """Report the one advertised extension key."""
+        return 1
+
+    def get(self, key: str, default: object = None) -> object:
+        """Raise if production code dynamically dispatches untrusted ``get``."""
+        del key, default
+        return _raise_unexpected_protocol_failure("secret extensions get failure")
+
+
 class _ExplodingReal:
     """Behave as a registered real number whose conversion raises arbitrarily."""
 
@@ -71,6 +111,36 @@ def _assert_generic_timeout_denial(timeout_value: object) -> None:
     assert error.value.__context__ is None
 
 
+def test_request_extension_copy_exceptions_are_masked() -> None:
+    """Mask arbitrary failures while detaching the outer extension mapping."""
+    with pytest.raises(
+        EgressNotAllowedError,
+        match=f"^{EGRESS_NOT_ALLOWED}$",
+    ) as error:
+        _bind_bounded_request_timeouts(
+            _ExplodingExtensionsCopyMapping(),
+            EgressTimeoutPolicy(),
+        )
+
+    assert error.value.__cause__ is None
+    assert error.value.__context__ is None
+
+
+def test_request_extensions_get_is_not_dynamically_dispatched() -> None:
+    """Snapshot a valid outer mapping without invoking its hostile ``get`` method."""
+    bounded = _bind_bounded_request_timeouts(
+        _ExplodingExtensionsGetMapping(),
+        EgressTimeoutPolicy(),
+    )
+
+    assert bounded["timeout"] == {
+        "connect": 1.0,
+        "read": 5.0,
+        "write": 5.0,
+        "pool": 5.0,
+    }
+
+
 def test_timeout_mapping_exceptions_are_masked() -> None:
     """Mask arbitrary failures raised while copying an untrusted timeout mapping."""
     _assert_generic_timeout_denial(_ExplodingTimeoutMapping())