Method intercept_reduce_call

python/pyfory/policy.py:360–414

Intercept and validate __reduce__ protocol callable invocation. This hook is called when deserializing an object that was serialized using the __reduce__ or __reduce_ex__ protocol, right before the callable is invoked to reconstruct the object. When Called -

(self, callable_obj, args, **kwargs)

    # ============================================================================

    def intercept_reduce_call(self, callable_obj, args, **kwargs):
        """Intercept and validate __reduce__ protocol callable invocation.

        This hook is called when deserializing an object that was serialized using the
        __reduce__ or __reduce_ex__ protocol, right before the callable is invoked
        to reconstruct the object.

        When Called
        -----------
        - During deserialization of objects using __reduce__/__reduce_ex__
        - Before callable_obj(*args) is executed
        - After the callable and args have been deserialized

        Security Use Cases
        ------------------
        - Block dangerous callables (eval, exec, os.system, subprocess.Popen)
        - Validate that callables match expected signatures
        - Inspect arguments for malicious payloads
        - Return pre-constructed safe objects to skip callable invocation
        - Log reduce operations for auditing

        Args:
            callable_obj (callable): The callable that will be invoked to reconstruct
                                    the object (typically a class or factory function).
            args (tuple): The arguments that will be passed to the callable.
            **kwargs: Reserved for future extensions.

        Returns:
            None: Return None to proceed with normal callable invocation (callable_obj(*args)).
            object: Return an object to use directly, skipping the callable invocation.
                   This allows you to construct safe replacement objects.

        Raises:
            Exception: Raise any exception to reject the callable and abort deserialization.

        Example:
            >>> class ReduceChecker(DeserializationPolicy):
            ...     def intercept_reduce_call(self, callable_obj, args, **kwargs):
            ...         # Block subprocess.Popen
            ...         if callable_obj.__name__ == 'Popen':
            ...             raise ValueError("Popen not allowed")
            ...
            ...         # Audit all reduce operations
            ...         import logging
            ...         logging.info(f"Reducing with {callable_obj}({args})")
            ...
            ...         return None  # Proceed normally

        Note:
            This is one of the most critical security hooks, as __reduce__ is the primary
            vector for arbitrary code execution in pickle-based attacks.

            `check_reduce_callable` is an alias for this hook.
        """
        pass

    # Hook aliases
    def check_reduce_callable(self, callable_obj, args, **kwargs):