Redact map keys in typed-value errors and refuse dataclass errors as typed

src/durable_workflow/_typed_values.py

@@ -136,6 +136,14 @@ def validate_annotation(annotation: Any, where: str, _seen: frozenset[type] = fr
     raise TypeError(f"{where}: typed values do not support {_type_name(annotation)}")
 
 
+def decode_payload_value(value: Any, annotation: Any, path: str = "value") -> Any:
+    """Decode an untrusted payload value; every refusal is a TypedValueError."""
+    try:
+        return from_payload_value(value, annotation, path)
+    except RecursionError:
+        raise TypedValueError(path, _type_name(annotation), "the value is nested too deeply") from None
+
+
 def from_payload_value(value: Any, annotation: Any, path: str = "value") -> Any:
     """Return *value* converted to *annotation*, which must be validated."""
     if annotation in _PASS_THROUGH:
@@ -162,7 +170,12 @@ def from_payload_value(value: Any, annotation: Any, path: str = "value") -> Any:
         if not isinstance(value, dict):
             raise TypedValueError(path, _type_name(annotation), f"got {_json_kind(value)}")
         _, item_type = get_args(annotation) or (str, Any)
-        return {key: from_payload_value(item, item_type, f"{path}[{key!r}]") for key, item in value.items()}
+        # Map keys are user data (for example email addresses), so a path names
+        # an entry by its position, never by its key.
+        return {
+            key: from_payload_value(item, item_type, f"{path}[entry {index}]")
+            for index, (key, item) in enumerate(value.items())
+        }
 
     if isinstance(annotation, type) and issubclass(annotation, Enum):
         return _enum_from_payload(value, annotation, path)
@@ -211,7 +224,12 @@ def _dataclass_from_payload(value: Any, cls: type, path: str) -> Any:
         elif item.default is dataclasses.MISSING and item.default_factory is dataclasses.MISSING:
             raise TypedValueError(field_path, _type_name(annotation), "the field is missing")
     # Keys the receiver does not declare are ignored for rolling deploys.
-    return cls(**arguments)
+    try:
+        return cls(**arguments)
+    except ValueError:
+        # A __post_init__ check refused the decoded fields. Its message can
+        # carry the values, so it is not repeated here.
+        raise TypedValueError(path, cls.__qualname__, "the dataclass refused the decoded fields") from None
 
 
 def to_payload_value(value: Any) -> Any:
@@ -266,10 +284,10 @@ def convert(args: tuple[Any, ...]) -> list[Any]:
         for index, value in enumerate(args):
             if index < len(positional):
                 name, annotation = positional[index]
-                converted.append(from_payload_value(value, annotation, name))
+                converted.append(decode_payload_value(value, annotation, name))
             elif variadic is not None:
                 name, annotation = variadic
-                converted.append(from_payload_value(value, annotation, f"{name}[{index - len(positional)}]"))
+                converted.append(decode_payload_value(value, annotation, f"{name}[{index - len(positional)}]"))
             else:
                 converted.append(value)
         return converted

src/durable_workflow/workflow.py

@@ -44,7 +44,7 @@
 )
 from ._cooperative_cancellation import CancellationDelivery, read_cancellation_history
 from ._failure_reporting import exception_message
-from ._typed_values import from_payload_value, to_payload_values, validate_annotation
+from ._typed_values import decode_payload_value, to_payload_values, validate_annotation
 from .cancellation import (
     CancellationContext,
     CancellationPolicy,
@@ -3128,7 +3128,7 @@ def _typed_activity_result(command: Any, value: Any) -> Any:
     """Decode an activity result to the call's opt-in result_type, if any."""
     if not isinstance(command, ScheduleActivity) or command.result_type is None or isinstance(value, BaseException):
         return value
-    return from_payload_value(value, command.result_type, "activity result")
+    return decode_payload_value(value, command.result_type, "activity result")
 
 
 def _decode_history_result(

tests/test_typed_values.py

@@ -112,7 +112,7 @@
     ({**NEUTRAL, "status": "lost-in-transit"}, "shipment.status", "Status", "the string is not a member"),
     ({**NEUTRAL, "priority": True}, "shipment.priority", "Priority", "the boolean is not a member"),
     ({**NEUTRAL, "items": "secret-sku"}, "shipment.items", "list[", "got string"),
-    ({**NEUTRAL, "weights": {"a": "heavy"}}, "shipment.weights['a']", "float", "got string"),
+    ({**NEUTRAL, "weights": {"ada@example.test": "heavy"}}, "shipment.weights[entry 0]", "float", "got string"),
     ({**NEUTRAL, "order_id": None}, "shipment.order_id", "str", "got null"),
     ({**NEUTRAL, "items": [{"sku": "a", "qty": True}]}, "shipment.items[0].qty", "int", "got boolean"),
     ("secret-order", "shipment", "Shipment", "got string"),
@@ -125,7 +125,7 @@
     assert raised.value.path == path
     assert raised.value.expected.startswith(expected)
     assert raised.value.detail == detail
-    for secret in ("secret", "lost-in-transit", "heavy"):
+    for secret in ("secret", "lost-in-transit", "heavy", "ada@example.test"):
         assert secret not in str(raised.value)
 
 
@@ -412,3 +412,53 @@
 def test_a_selection_mismatch_is_the_winner_failure_and_raised_at_a_handle_await() -> None:
     (command,) = replay(MismatchedSelectionWorkflow, _selection_history(), [], payload_codec="avro").commands
     assert command == CompleteWorkflow({"winner": "fast", "winner_detail": "got object", "slow_detail": "got object"})
+
+
+# Redaction and refusal follow-ups from the #247 review.
+@dataclass
+class Positive:
+    amount: int
+
+    def __post_init__(self) -> None:
+        if self.amount <= 0:
+            raise ValueError(f"amount must be positive, got {self.amount}")
+
+
+def test_a_dataclass_that_refuses_its_fields_is_a_typed_value_error() -> None:
+    with pytest.raises(TypedValueError) as raised:
+        from_payload_value({"amount": -7}, Positive, "payment")
+    assert (raised.value.path, raised.value.detail) == ("payment", "the dataclass refused the decoded fields")
+    assert "-7" not in str(raised.value) and raised.value.__cause__ is None
+
+
+def test_map_entry_paths_never_name_the_key() -> None:
+    with pytest.raises(TypedValueError) as raised:
+        from_payload_value({"x": 1, "ada@example.test": "one"}, dict[str, int], "scores")
+    assert raised.value.path == "scores[entry 1]"
+    assert "ada@example.test" not in str(raised.value)
+
+
+@dataclass
+class Node:
+    child: Node | None = None
+
+
+def test_a_too_deeply_nested_payload_is_a_typed_value_error() -> None:
+    from durable_workflow._typed_values import decode_payload_value
+
+    nested: Any = None
+    for _ in range(5000):
+        nested = {"child": nested}
+    with pytest.raises(TypedValueError) as raised:
+        decode_payload_value(nested, Node, "tree")
+    assert raised.value.detail == "the value is nested too deeply"
+    assert isinstance(raised.value, NonRetryableError)
+
+
+def test_a_typed_activity_maps_refusals_to_typed_value_errors() -> None:
+    @activity.defn(name="tests.typed.pay", typed=True)
+    def pay(payment: Positive) -> int:
+        return payment.amount
+
+    with pytest.raises(TypedValueError, match="the dataclass refused the decoded fields"):
+        pay({"amount": 0})