nvflare.fuel.utils.fobs.fobs module

add_type_name_whitelist(*type_names: str) → None

Add type names to the whitelist for deserialization.

Type names added here are allowed to be auto-loaded during deserialization when they have not been explicitly pre-registered. This prevents arbitrary class loading (RCE) while still supporting lazy registration use cases.

Parameters:

type_names – Fully qualified class names (e.g. “mypackage.MyClass”)

auto_register_enum_types(enabled=True) → None

Enable or disable the auto-registration of enum types.

Parameters:

enabled – Auto-registration of enum classes is enabled if True.

deserialize(data: bytes, manager: DatumManager | None = None, **kwargs) → Any

Deserialize bytes into an object.

Parameters:

• data – Serialized data

• manager – Datum manager to internalize datum

• kwargs – Arguments passed to msgpack.unpackb

Returns:

Deserialized object

deserialize_stream(stream: BinaryIO, manager: DatumManager | None = None, **kwargs) → Any

Deserialize bytes from stream into an object.

Parameters:

• stream – Stream to write serialized data to

• manager – Datum manager to internalize datum

• kwargs – Arguments passed to msgpack.unpackb

Returns:

Deserialized object

get_dot_handler(dot: int)

num_decomposers() → int

Returns the number of decomposers registered.

Returns:

The number of decomposers

register(decomposer: Decomposer | Type[Decomposer]) → None

Register a decomposer. It does nothing if decomposer is already registered for the type

Parameters:

decomposer – The decomposer type or instance

register_data_classes(*data_classes: Type[T]) → None

Register generic decomposers for data classes.

Also adds each class to the type-name whitelist so that it can be lazily re-loaded during deserialization. The whitelist entry is cleared when fobs.reset() is called.

Parameters:

data_classes – The classes to be registered

register_enum_types(*enum_types: Type[Enum]) → None

Register generic decomposers for enum classes.

Also adds each class to the type-name whitelist so that it can be lazily re-loaded during deserialization. The whitelist entry is cleared when fobs.reset() is called.

Parameters:

enum_types – The enum classes to be registered

register_folder(folder: str, package: str)

Scan the folder and register all decomposers found.

Parameters:

• folder – The folder to scan

• package – The package to import the decomposers from

reset()

Reset FOBS to initial state. Used for unit test

serialize(obj: Any, manager: DatumManager | None = None, **kwargs) → bytes

Serialize object into bytes.

Parameters:

• obj – Object to be serialized

• manager – Datum manager used to externalize datum

• kwargs – Arguments passed to msgpack.packb

Returns:

Serialized data

serialize_stream(obj: Any, stream: BinaryIO, manager: DatumManager | None = None, **kwargs)

Serialize object and write the data to a stream.

Parameters:

• obj – Object to be serialized

• stream – Stream to write the result to

• manager – Datum manager to externalize datum

• kwargs – Arguments passed to msgpack.packb