Reject serde-reserved keys in HITL params_input

[haseebmalik18]

A HITL response whose params_input contains a serde-reserved key (__classname__ or __id__) cannot be serialized when the task resumes. Today the response is silently discarded and the user has no way to resubmit a corrected one.

This validates params_input when the response is submitted and returns a 400 if it carries one of those keys, before anything is written. The task stays in AWAITING_INPUT, so the user can fix the input and resubmit.

Only the public HITL route needs this. The execution API route builds its params_input from trusted operator code rather than user input, so it is left as is.

closes: #71036

[amoghrajesh]

This doesn't have to be in 3.3.1, but if we are comfortable (cc @Lee-W) we can consider it.

[Lee-W]

A few nitpicks, but overall this looks good. I also don't think it's worth blocking the 3.3.1 release.

[vatsrahul1001]

Moved to 3.3.2

[haseebmalik18]

@Lee-W Applied changes left from review

[jason810496]

Thanks for fixing this.

[jason810496]

IMO, we should follow the convetion in XCom and validate at the data model layer:

airflow/airflow-core/src/airflow/api_fastapi/core_api/datamodels/xcom.py

Lines 87 to 119 in 9a85faa

	def _check_forbidden_xcom_keys(value: Any) -> Any:
	"""Recursively reject forbidden deserialization keys in user-provided XCom data."""
	from airflow._shared.serialization import FORBIDDEN_XCOM_KEYS
	def _walk(obj: Any, path: str = "value") -> None:
	if isinstance(obj, str):
	# A value submitted as a JSON string literal (e.g. ``json.dumps({...})``)
	# is stored verbatim and re-parsed into a dict/list on a
	# ``deserialize=true`` read, which would otherwise smuggle reserved keys
	# past the dict/list checks below. Re-parse and inspect the decoded
	# structure the same way the read path does.
	try:
	decoded = json.loads(obj)
	except (ValueError, TypeError):
	return
	if isinstance(decoded, (dict, list)):
	_walk(decoded, path)
	return
	if isinstance(obj, dict):
	found = FORBIDDEN_XCOM_KEYS & obj.keys()
	if found:
	raise ValueError(
	f"XCom {path} contains reserved serialization keys: {', '.join(sorted(found))}. "
	f"These keys are reserved for internal use."
	)
	for k, v in obj.items():
	_walk(v, f"{path}.{k}")
	elif isinstance(obj, (list, tuple)):
	for i, item in enumerate(obj):
	_walk(item, f"{path}[{i}]")
	_walk(value)
	return value

airflow/shared/serialization/src/airflow_shared/serialization/__init__.py

Lines 20 to 44 in 9a85faa

	CLASSNAME = "__classname__"
	VERSION = "__version__"
	DATA = "__data__"
	SCHEMA_ID = "__id__"
	CACHE = "__cache__"
	# Legacy serialization format keys (processed by serde for backwards compatibility)
	OLD_TYPE = "__type"
	OLD_SOURCE = "__source"
	OLD_DATA = "__var"
	OLD_DICT = "dict"
	FORBIDDEN_XCOM_KEYS = frozenset(
	{
	CLASSNAME,
	VERSION,
	DATA,
	SCHEMA_ID,
	CACHE,
	OLD_TYPE,
	OLD_SOURCE,
	OLD_DATA,
	}
	)

[haseebmalik18]

Good point, thanks! Only thing: XCom's validator raises ValueError so it comes back as 422, but the issue asks for 400 here? This route already does its option checks with HTTPException(400), so I matched that. Would you still want it at the datamodel layer even if it becomes 422?

I also kept the keys to just __classname__/__id__ (what serialize actually rejects) instead of the full FORBIDDEN_XCOM_KEYS, since the wider set would reject valid things like {"__version__": 1}. Sound ok?