fix: harden complex parameter model loading

[Rana Singh (ranadeepsingh)]

Related Issues/PRs

Follow-up to #2513 and the remaining MSRC recommendation in IcM 31000000568481.

What changes are proposed in this pull request?

This PR closes the remaining unrestricted Java deserialization path in
ComplexParams model persistence. A crafted artifact could previously execute
deserialization callbacks before the eventual type cast.

The patch:

• fails closed for unconstrained Java object graphs and permits only audited,
  type-specific class policies;
• rejects serialized lambdas and unsafe proxies and enforces graph, object,
  array, string, stream, metadata, path, and heap-aware limits;
• migrates data types and nested stages to bounded formats/readers where
  possible;
• validates model metadata, canonical artifact containment, aliases, links,
  and bounded remote enumeration before reader construction;
• scopes trusted legacy compatibility to the active Spark session and explicit
  artifact load;
• binds SynapseML Python readers to the active session and uses the bounded
  PipelineSerializer for generated nested-stage tests;
• gives language bindings a same-thread, closeable trusted scope so native
  MLflow Pipeline persistence remains compatible without trusting ambient
  configuration;
• routes generated R nested-stage loading through PipelineSerializer instead
  of native ml_load; and
• centralizes trust for harness-created fixtures instead of adding
  suite-specific opt-outs.

Compatibility contract

New data-only parameter artifacts load without unsafe Java deserialization.
Legacy artifacts containing arbitrary closures, BallTrees, DataFrames, native
readers, or native Spark Pipeline nesting must be explicitly treated as
trusted by setting
spark.synapseml.legacy.allowUnsafeJavaDeserialization=true on the reader's
Spark session. Native Pipeline compatibility must additionally run inside
Serializer.withTrustedArtifactLoad; language bindings that cannot pass a
Scala closure use Serializer.beginTrustedArtifactLoad and close the returned
scope on the same gateway thread. Untrusted nested models use SynapseML's
bounded PipelineSerializer.

Direct ConditionalBallTree.load fails closed. Trusted legacy callers can use
the explicit loadUnsafe API.

Scope

The effective diff contains 45 files: 20 connected production
security-boundary files and 25 focused test/harness files. It contains no review
artifacts, dependency changes, workflow changes, or generated target/ files.
The final cross-language compatibility fix stays within files already required by
the remediation and adds no files.

How is this patch tested?

• Master baseline (JDK 11 / Scala 2.12): affected module main/test
  compilation and Scalastyle.
• Core exploit, resource-budget, metadata/path, session-isolation,
  Pipeline/ComplexParams, generated-loader, UDF, recommendation, and
  persistence suites.
• Rebased head: 156 targeted security/compatibility tests on master and 158 on
  the Spark 4.1 replay, including R codegen and the LightGBM policy test.
• Final stream-cleanup review fix: compile plus the 29-test persistence suite and
  main/test Scalastyle on both baselines.
• Four exact generated Python failures from Azure reproduced locally and
  passed end-to-end through MLflow save, log, and load:
  RankingAdapterModelSpec, RankingTrainValidationSplitModelSpec,
  RecommendationIndexerModelSpec, and SARModelSpec.
• Generated persistence runtime tests: core (2), cognitive (1), and deep
  learning (3), including trust cleanup assertions.
• Core, cognitive, deep-learning, and OpenCV code/test generation plus
  generated Python syntax compilation.
• Generated core, deep-learning, and OpenCV R fixtures use the
  session-backed PipelineSerializer/ml_call_constructor path and contain no
  nested-model ml_load path.
• Exact-head Azure build 233083244: all 66 Azure checks passed; all 76 GitHub/Azure status checks are green on a17cabb005.
• Black 22.3.0 and git diff --check.
• Spark 4.1 replay (JDK 17 / Scala 2.13): affected module test compilation,
  focused security/compatibility tests, and main/test Scalastyle.

Does this PR change any dependencies?

• No. You can skip this section.
• Yes. Make sure the dependencies are resolved correctly, and list changes here.

Does this PR add a new feature? If so, have you added samples on website?

• No. You can skip this section.
• Yes. Make sure you have added samples following below steps.

[azure-pipelines]

Azure Pipelines:
There may be pipelines that require an authorized user to comment /azp run to run.

[github-actions]

Hey Rana Singh (@ranadeepsingh) 👋!
Thank you so much for contributing to our repository 🙌.
Someone from SynapseML Team will be reviewing this pull request soon.

We use semantic commit messages to streamline the release process.
Before your pull request can be merged, you should make sure your first commit and PR title start with a semantic prefix.
This helps us to create release messages and credit you for your hard work!

Examples of commit messages with semantic prefixes:

• fix: Fix LightGBM crashes with empty partitions
• feat: Make HTTP on Spark back-offs configurable
• docs: Update Spark Serving usage
• build: Add codecov support
• perf: improve LightGBM memory usage
• refactor: make python code generation rely on classes
• style: Remove nulls from CNTKModel
• test: Add test coverage for CNTKModel

To test your commit locally, please follow our guild on building from source.
Check out the developer guide for additional guidance on testing your change.

[Rana Singh (ranadeepsingh)]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[Copilot]

Pull request overview

This PR hardens SynapseML/SparkML model persistence and loading paths to fail closed against unsafe Java deserialization, unbounded metadata/object graphs, and path-traversal/link/glob issues—while preserving an explicit trusted-legacy compatibility switch for artifacts that cannot be safely constrained.

Changes:

• Introduces a session-scoped “trusted legacy artifact” gate (spark.synapseml.legacy.allowUnsafeJavaDeserialization) and routes unsafe persistence (UDF closures, custom readers, BallTrees, DataFrames, etc.) behind explicit opt-in.
• Adds hardened, budgeted model metadata/path handling (canonical containment, link resolution, bounded metadata enumeration and decoding budgets, recursive model context) for Pipeline/ComplexParams loads.
• Migrates DataTypeParam persistence toward bounded JSON handling and adds/updates extensive regression coverage for security and compatibility.

Show a summary per file

File	Description
reviews/serializer-deserialization/task-2513-attempt-1-review-1-gpt-5.6-sol.md	Adds a long-form review artifact documenting the hardening work and verification steps.
lightgbm/src/test/scala/com/microsoft/azure/synapse/ml/lightgbm/params/VerifyLightGBMBoosterParam.scala	Adds a regression ensuring LightGBM booster param loads under constrained legacy handling.
lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/params/LightGBMBoosterParam.scala	Adds a narrow deserialization class policy for LightGBMBoosterParam.
core/src/test/scala/org/apache/spark/ml/VerifyArtifactPathResolver.scala	Adds tests for metadata listing guarantees and Java 8 linkage constraints.
core/src/test/scala/com/microsoft/azure/synapse/ml/stages/UDFTransformerSuite.scala	Adds trust-gating tests for persisted UDFs and session propagation through PipelineSerializer.
core/src/test/scala/com/microsoft/azure/synapse/ml/stages/LambdaSuite.scala	Enables trusted legacy load path for serialization fuzzing where needed.
core/src/test/scala/com/microsoft/azure/synapse/ml/recommendation/RankingTrainValidationSpec.scala	Enables trusted legacy load path for serialization fuzzing in ranking suites.
core/src/test/scala/com/microsoft/azure/synapse/ml/param/VerifyEvaluatorParam.scala	Adds explicit-trust tests for evaluator param persistence.
core/src/test/scala/com/microsoft/azure/synapse/ml/param/VerifyEstimatorArrayParam.scala	Adds coverage for stage-array persistence (safe writable vs trusted legacy cases).
core/src/test/scala/com/microsoft/azure/synapse/ml/param/VerifyDataTypeParam.scala	Adds coverage for JSON DataType persistence and legacy-stream gating/tripwires.
core/src/test/scala/com/microsoft/azure/synapse/ml/param/VerifyDataFrameParam.scala	Adds regression guarding against linked Parquet parts escaping artifact containment.
core/src/test/scala/com/microsoft/azure/synapse/ml/param/VerifyArrayParamMapParam.scala	Adds explicit-trust coverage for ArrayParamMapParam persistence.
core/src/test/scala/com/microsoft/azure/synapse/ml/nn/VerifySchemas.scala	Extends SafeObjectInputStream/BallTree coverage and validates fail-closed NN loading defaults.
core/src/test/scala/com/microsoft/azure/synapse/ml/io/split1/ParserSuite.scala	Enables trusted legacy load path for serialization fuzzing in parser suites where applicable.
core/src/test/scala/com/microsoft/azure/synapse/ml/core/utils/VerifySafeObjectInputStream.scala	Adds focused unit tests for SafeObjectInputStream resource and policy enforcement.
core/src/test/scala/com/microsoft/azure/synapse/ml/core/test/fuzzing/Fuzzing.scala	Adds a controlled trusted-legacy mode for serialization fuzzing and Pipeline/PipelineModel round trips.
core/src/test/scala/com/microsoft/azure/synapse/ml/core/serialize/VerifyModelLoadEnvironment.scala	Adds environment regressions around session scoping, trusted loading, and provider behaviors.
core/src/test/scala/com/microsoft/azure/synapse/ml/core/serialize/VerifyMetadataBudgets.scala	Adds regression ensuring aggregate decoded-metadata budgets are enforced.
core/src/test/scala/com/microsoft/azure/synapse/ml/core/serialize/ValidateComplexParamSerializer.scala	Significantly expands hardening regressions across metadata, paths, trust gates, and reader safety.
core/src/test/scala/com/microsoft/azure/synapse/ml/automl/VerifyFindBestModel.scala	Enables trusted legacy load path for serialization fuzzing in AutoML persistence tests.
core/src/main/scala/org/apache/spark/ml/StageReaderInspector.scala	Adds bytecode-based reader classification without class initialization.
core/src/main/scala/org/apache/spark/ml/Serializer.scala	Centralizes hardened serializer routing, trust scoping, Pipeline serializers, and safe read/write behavior.
core/src/main/scala/org/apache/spark/ml/ModelLoadContext.scala	Introduces shared load/write budgets (nodes, depth, metadata physical/decoded) across nested model loads.
core/src/main/scala/org/apache/spark/ml/DataTypeSerializer.scala	Implements bounded DataType JSON persistence with legacy-stream detection and UDT gating.
core/src/main/scala/org/apache/spark/ml/ComplexParamsSerializer.scala	Hardens ComplexParams read/write (session assignment, budgets, metadata accounting, and native pipeline boundaries).
core/src/main/scala/org/apache/spark/ml/ArtifactPathResolver.scala	Adds canonical containment, link resolution, bounded metadata enumeration/decoding, and safe directory validation helpers.
core/src/main/scala/com/microsoft/azure/synapse/ml/param/TransformerArrayParam.scala	Persists transformer arrays via hardened PipelineArraySerializer and validates load-time types.
core/src/main/scala/com/microsoft/azure/synapse/ml/param/EstimatorArrayParam.scala	Persists estimator arrays via hardened PipelineArraySerializer and validates load-time types.
core/src/main/scala/com/microsoft/azure/synapse/ml/param/DataTypeParam.scala	Adjusts DataTypeParam imports to align with new DataTypeSerializer path.
core/src/main/scala/com/microsoft/azure/synapse/ml/param/DataFrameParam.scala	Marks DataFrameParam as not supporting untrusted deserialization by default.
core/src/main/scala/com/microsoft/azure/synapse/ml/param/ByteArrayParam.scala	Adds an explicit deserialization class policy (primitive byte arrays).
core/src/main/scala/com/microsoft/azure/synapse/ml/param/BallTreeParam.scala	Marks BallTree params as requiring trusted loading (fail closed by default).
core/src/main/scala/com/microsoft/azure/synapse/ml/nn/BallTree.scala	Changes ConditionalBallTree default load to fail closed and adds explicit loadUnsafe.
core/src/main/scala/com/microsoft/azure/synapse/ml/core/utils/SafeObjectInputStream.scala	Expands SafeObjectInputStream with resource limits, filter composition, and class allowlisting model.
core/src/main/scala/com/microsoft/azure/synapse/ml/core/serialize/ComplexParam.scala	Adds per-param deserialization policy hooks and trust gating for unsafe complex params.

Review details

💡 Configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

• Files reviewed: 34/35 changed files
• Comments generated: 2
• Review effort level: Lite

[Rana Singh (ranadeepsingh)]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[Copilot]

Review details

• Files reviewed: 35/35 changed files
• Comments generated: 3
• Review effort level: Lite

[Rana Singh (ranadeepsingh)]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[Copilot]

Review details

Suppressed comments (2)

Previously missed (2) — in code that hasn't changed since the last review.

core/src/main/scala/org/apache/spark/ml/Serializer.scala:88

• typeToSerializer routes any Array[_] whose element type is a subtype of PipelineStage to PipelineArraySerializer and then casts it to Serializer[T]. Since arrays are reified on the JVM, PipelineArraySerializer will return an Array[PipelineStage], which cannot be safely treated as (for example) Array[Estimator[_]]/Array[Transformer] and can produce ClassCastException for any caller that goes through Serializer.typeToSerializer (outside the specialized Param overrides).

Consider returning a small adapter serializer for pipeline-stage arrays that converts to/from Array[PipelineStage] while preserving the requested runtime component type.

  def typeToSerializer[T](
      tpe: Type,
      sparkSession: SparkSession,
      classFilter: Option[DeserializationClassFilter]): Serializer[T] = {
    (if (tpe <:< typeOf[PipelineStage])              new PipelineSerializer(sparkSession)
     else if (isPipelineStageArray(tpe))             new PipelineArraySerializer(sparkSession)

core/src/main/scala/com/microsoft/azure/synapse/ml/core/utils/SafeObjectInputStream.scala:218

• SafeObjectInputStream.ResourceFilter is a shared singleton instance of DeserializationResourceFilter, but DeserializationResourceFilter is stateful (declaredArrayBytes accumulates across checkInput calls). This makes the exposed filter instance easy to misuse and can make future tests order-dependent if they call ResourceFilter.checkInput on multiple “allowed” inputs.

Prefer exposing a factory (e.g., def newResourceFilter(...)) that returns a fresh per-stream filter, and update VerifySafeObjectInputStream to construct a new instance for assertions.

  private[utils] val ResourceFilter: ObjectInputFilter =
    new DeserializationResourceFilter(defaultResourceLimits)

• Files reviewed: 35/35 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Rana Singh (ranadeepsingh)]

Resolved both suppressed findings from the 9aa866acbd automated review in 0e360a5827:

• Generic PipelineStage-array serializers now allocate and validate the requested reified component type, with direct Array[Estimator[_]] and Array[Transformer] round-trip coverage.
• DeserializationResourceFilter is now created fresh per stream/test instead of exposing a shared stateful singleton.
• The same commit fully qualifies Spark Transformer, and the complete patch now passes spark4.1 core/Test/compile locally under Java 17 / Scala 2.13.

Master core compilation, main/test Scalastyle, VerifyEstimatorArrayParam, VerifySafeObjectInputStream, and ValidateComplexParamSerializer all pass.

[Rana Singh (ranadeepsingh)]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[Copilot]

Review details

• Files reviewed: 36/36 changed files
• Comments generated: 1
• Review effort level: Lite

[Rana Singh (ranadeepsingh)]

/azp run

[Copilot]

Review details

• Files reviewed: 45/45 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Rana Singh (ranadeepsingh)]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[Copilot]

Review details

• Files reviewed: 45/45 changed files
• Comments generated: 1
• Review effort level: Lite

[Rana Singh (ranadeepsingh)]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[Rana Singh (ranadeepsingh)]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[Codecov Comments Bot (codecov-commenter)]

Codecov Report

❌ Patch coverage is 87.50000% with 133 lines in your changes missing coverage. Please review.
✅ Project coverage is 87.42%. Comparing base (7c72c49) to head (72a61f5).

Files with missing lines	Patch %	Lines
...ynapse/ml/core/utils/Jep290ObjectInputFilter.scala	61.70%	36 Missing ⚠️
...ala/org/apache/spark/ml/ArtifactPathResolver.scala	87.31%	26 Missing ⚠️
...rc/main/scala/org/apache/spark/ml/Serializer.scala	90.74%	25 Missing ⚠️
.../synapse/ml/core/utils/SafeObjectInputStream.scala	84.48%	18 Missing ⚠️
...ala/org/apache/spark/ml/StageReaderInspector.scala	91.91%	8 Missing ⚠️
...n/scala/org/apache/spark/ml/ModelLoadContext.scala	94.50%	5 Missing ⚠️
...scala/org/apache/spark/ml/DataTypeSerializer.scala	91.66%	4 Missing ⚠️
...spark/ml/RuntimeTypedPipelineArraySerializer.scala	75.00%	4 Missing ⚠️
.../org/apache/spark/ml/ComplexParamsSerializer.scala	96.51%	3 Missing ⚠️
...a/com/microsoft/azure/synapse/ml/nn/BallTree.scala	71.42%	2 Missing ⚠️
... and 2 more

Additional details and impacted files

@@            Coverage Diff            @@
##           master    #2678     +/-   ##
=========================================
  Coverage   87.41%   87.42%             
=========================================
  Files         341      347      +6     
  Lines       20742    21765   +1023     
  Branches     2166     2327    +161     
=========================================
+ Hits        18131    19027    +896     
- Misses       2611     2738    +127     

Files with missing lines	Coverage Δ
...azure/synapse/ml/core/serialize/ComplexParam.scala	92.30% <100.00%> (+12.30%)	⬆️
...crosoft/azure/synapse/ml/param/BallTreeParam.scala	50.00% <100.00%> (+50.00%)	⬆️
...rosoft/azure/synapse/ml/param/ByteArrayParam.scala	100.00% <100.00%> (ø)
...rosoft/azure/synapse/ml/param/DataFrameParam.scala	100.00% <100.00%> (ø)
...crosoft/azure/synapse/ml/param/DataTypeParam.scala	100.00% <ø> (ø)
...rosoft/azure/synapse/ml/param/EstimatorParam.scala	94.44% <100.00%> (ø)
...apse/ml/lightgbm/params/LightGBMBoosterParam.scala	100.00% <100.00%> (ø)
...t/azure/synapse/ml/param/EstimatorArrayParam.scala	83.33% <75.00%> (-16.67%)	⬇️
...azure/synapse/ml/param/TransformerArrayParam.scala	66.66% <75.00%> (+16.66%)	⬆️
...a/com/microsoft/azure/synapse/ml/nn/BallTree.scala	85.84% <71.42%> (+3.43%)	⬆️
... and 9 more

... and 1 file with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Rana Singh (ranadeepsingh)]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[Copilot]

Review details

Suppressed comments (2)

Previously missed (2) — in code that hasn't changed since the last review.

core/src/main/python/synapse/ml/core/schema/Utils.py:141

• ComplexParamsMixin.read() / JavaMMLReader claim to bind readers to the active Spark session, but the constructor always uses SparkSession.builder.getOrCreate(). This can ignore a thread-local active session (e.g., spark.newSession() with different conf), undermining the session-scoped trust/config model described in the PR.

    def __init__(self, clazz):
        super(JavaMMLReader, self).__init__(clazz)
        self.session(SparkSession.builder.getOrCreate())

core/src/main/scala/org/apache/spark/ml/ArtifactPathResolver.scala:429

• resolveTreeInside recursively enumerates the entire directory tree via fs.listStatus(...) with no budgeting and without integrating with ModelLoadContext. When used for DataFrame params (Parquet), this adds a full extra traversal and can be unbounded in both time and memory for large artifacts, which is at odds with the PR’s stated “bounded enumeration” goal. Consider iterating with listStatusIterator and counting each resolved entry via ModelLoadContext.current.foreach(_.enterPath(...)) so model loads inherit the existing node limits and detect aliasing early.

    def validateDirectory(directory: Path): Unit = {
      val directoryKey = directory.toUri.normalize().toString
      require(
        visitedDirectories.add(directoryKey),
        s"$description contains a filesystem-link cycle at $directory"

• Files reviewed: 45/45 changed files
• Comments generated: 0 new
• Review effort level: Lite